diff --git a/app/application/messaging/chat.py b/app/application/messaging/chat.py index 2e44b0f59..d85690fa8 100644 --- a/app/application/messaging/chat.py +++ b/app/application/messaging/chat.py @@ -14,6 +14,7 @@ from app.application.database import ( AsyncDatabaseExecutor, ) from app.application.messaging.recall import RecallService +from app.domain.chat import normalize_display_timestamps from app.runtime.observability import record_metric from app.runtime.priority import INTERACTIVE_DATABASE_CAPACITY, is_interactive from app.schemas.agent import AgentChatSessionDetail, AgentChatSessionSummary @@ -465,11 +466,16 @@ class AgentChatService: @classmethod def to_detail(cls, record: AgentChatRecord) -> AgentChatSessionDetail: """ - 把持久化投影转换为会话详情 DTO。 + 把持久化投影转换为会话详情 DTO,兼容旧消息缺失时间戳。 + + 只修正响应副本,避免读取接口改写数据库或运行中的内存快照。 + + :param record: 已完成授权检查的会话持久化投影 + :return: 所有展示消息都包含时间戳的会话详情 """ return AgentChatSessionDetail.model_validate({ **cls.to_summary(record).model_dump(), - "messages": record.messages, + "messages": normalize_display_timestamps(record.messages), }) @staticmethod diff --git a/app/domain/chat.py b/app/domain/chat.py index 30cd2dd48..cb9005ff7 100644 --- a/app/domain/chat.py +++ b/app/domain/chat.py @@ -1,48 +1,79 @@ """ -会话展示快照中选择结果的合并规则,不负责交互执行或持久化。 +会话展示快照的时间戳兼容与选择结果合并规则,不负责交互执行或持久化。 """ from copy import deepcopy from typing import Any -def _choice_selections(messages: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: +def normalize_display_timestamps(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: + """ + 为历史展示消息补齐缺失时间戳,不修改原快照或已有时间。 + + 缺失值沿用前方最近的有效时间;开头缺失时使用后方首个有效时间。 + 整个快照没有时间时使用 0 表示未知,避免每次读取都改变历史时间。 + + :param messages: 可能含有缺失 createdAt 字段的历史展示快照 + :return: 时间戳字段完整的快照;已有非法值仍由响应模型报错 + """ + timestamp = next((message["createdAt"] for message in messages + if isinstance(message.get("createdAt"), (int, float))), 0) + result = [] + for message in messages: + if isinstance(message.get("createdAt"), (int, float)): + timestamp = message["createdAt"] + result.append({"createdAt": timestamp, **message}) + return result + + +def _choice_receipts(messages: list[dict[str, Any]]) -> dict[str, dict[str, Any]]: """ 优先采用完整选择回执,旧记录只有已选卡片时从卡片恢复。 + + :param messages: 当前会话快照,旧回执可能缺少时间戳 + :return: 以选择 ID 索引的回执,同时保留首次选择与对应时间 """ - selections: dict[str, dict[str, Any]] = {} - for message in messages: + receipts: dict[str, dict[str, Any]] = {} + normalized = normalize_display_timestamps(messages) + for message in normalized: selected = message.get("choice_selection") if isinstance(selected, dict) and selected.get("choice_id"): - selections.setdefault(str(selected["choice_id"]), selected) - for message in messages: + receipts.setdefault(str(selected["choice_id"]), { + "id": f"choice-selection-{selected['choice_id']}", "role": "user", "content": "", + "createdAt": message["createdAt"], "choice_selection": selected, + }) + for message in normalized: for choice in message.get("choices") or []: if isinstance(choice, dict) and choice.get("status") == "selected" and choice.get("id"): - selections.setdefault(str(choice["id"]), { - "choice_id": str(choice["id"]), - "prompt": choice.get("prompt") or "", - "buttons": choice.get("buttons") or [], - "button_rows": choice.get("button_rows") or [], - **{key: choice.get(key) for key in ( - "title", "selected_label", "selected_value", "selected_description", - )}, + receipts.setdefault(str(choice["id"]), { + "id": f"choice-selection-{choice['id']}", "role": "user", "content": "", + "createdAt": message["createdAt"], + "choice_selection": { + "choice_id": str(choice["id"]), + "prompt": choice.get("prompt") or "", + "buttons": choice.get("buttons") or [], + "button_rows": choice.get("button_rows") or [], + **{key: choice.get(key) for key in ( + "title", "selected_label", "selected_value", "selected_description", + )}, + }, }) - return selections + return receipts def merge_choice_state( previous: list[dict[str, Any]], messages: list[dict[str, Any]], ) -> list[dict[str, Any]]: """ - 保留首次已选结果,旧客户端的 pending 快照不能降级已选卡片。 + 保留首次已选结果及回执时间,旧客户端的 pending 快照不能降级已选卡片。 :param previous: 当前持久化的权威快照 :param messages: 本次准备保存的完整快照,不原地修改 :return: 应用已选结果并按稳定 ID 合并选择回显后的快照 """ - selections = _choice_selections(previous) - for choice_id, selection in _choice_selections(messages).items(): - selections.setdefault(choice_id, selection) + selection_receipts = _choice_receipts(previous) + for choice_id, receipt in _choice_receipts(messages).items(): + selection_receipts.setdefault(choice_id, receipt) result: list[dict[str, Any]] = [] receipts: dict[str, int] = {} received_choices: set[str] = set() @@ -52,11 +83,13 @@ def merge_choice_state( if isinstance(selected, dict) and selected.get("choice_id"): choice_id = str(selected["choice_id"]) received_choices.add(choice_id) - message["choice_selection"] = deepcopy(selections[choice_id]) + receipt = selection_receipts[choice_id] + message["choice_selection"] = deepcopy(receipt["choice_selection"]) + message["createdAt"] = receipt["createdAt"] for choice in message.get("choices") or []: - if not isinstance(choice, dict) or str(choice.get("id") or "") not in selections: + if not isinstance(choice, dict) or str(choice.get("id") or "") not in selection_receipts: continue - selection = selections[str(choice["id"])] + selection = selection_receipts[str(choice["id"])]["choice_selection"] choice.update({"status": "selected", **{ key: selection.get(key) for key in ("selected_label", "selected_value", "selected_description") }}) @@ -67,10 +100,7 @@ def merge_choice_state( if key: receipts[key] = len(result) result.append(message) - for choice_id, selection in selections.items(): + for choice_id, receipt in selection_receipts.items(): if choice_id not in received_choices: - result.append({ - "id": f"choice-selection-{choice_id}", "role": "user", "content": "", - "choice_selection": deepcopy(selection), - }) + result.append(deepcopy(receipt)) return result diff --git a/docs/architecture-overview.md b/docs/architecture-overview.md index fd4881f78..789488a25 100644 --- a/docs/architecture-overview.md +++ b/docs/architecture-overview.md @@ -825,7 +825,7 @@ flowchart LR | 指标 | 当前值 | |---|---:| | Python 模块 | 1150 | -| 内部导入边 | 9,963 | +| 内部导入边 | 9,965 | | 非平凡 SCC | 1(29 个模块的 TMDB vendor 精确 containment) | | Application / Chain 具体 Adapter 直连 | 0 / 0 | | Direct egress | 51(精确 policy 准入,Application / Chain 均为 0) | diff --git a/docs/mcp-api.md b/docs/mcp-api.md index a996bb01d..d17e4a608 100644 --- a/docs/mcp-api.md +++ b/docs/mcp-api.md @@ -74,6 +74,8 @@ 后端按用户、会话和选择 ID 领取一次执行;入队前明确拒绝可重试,执行结果未知时不重放。 `echo_user=false` 仅隐藏用户回显,选择结果仍保存;完整快照替换保留隐藏选择收据, 即使旧快照不含卡片,也不能让后续快照把已选卡片改回 `pending`。普通消息仍按完整快照替换。 +选择回执合并时保留原始 `createdAt` 毫秒时间戳;旧历史缺少该字段时,读取和保存边界 +使用相邻消息时间补齐,不丢弃选择内容。整份快照都没有时间时以 `0` 表示未知。 会话 `is_processing` 只表示实际执行或排队,空闲等待下一条消息的 worker 不计入。 工具事件沿用 `tool_id` 与 `assistant_message_id`,跨 steering 后仍更新原助手段。 diff --git a/docs/rules/05-architecture.md b/docs/rules/05-architecture.md index 38f8b2a91..2f7308c37 100644 --- a/docs/rules/05-architecture.md +++ b/docs/rules/05-architecture.md @@ -1342,7 +1342,7 @@ driven workflow registration. | `app/application/server/report.py` | Server reporting use cases over injected local readers and transport callbacks | | `app/application/server/share.py` | Server sharing use cases over injected repositories and transport callbacks | | `app/domain/plugin.py` | Pure local-source, generation-compatibility and Release-install-plan rules | -| `app/domain/chat.py` | 会话展示快照的纯选择状态合并规则;保留首次选择,不读取交互管理器或持久化服务 | +| `app/domain/chat.py` | 会话展示快照的纯时间戳兼容与选择状态合并规则;保留首次选择及回执时间,不读取交互管理器或持久化服务 | | `app/adapters/external/plugin/client.py` | `PluginMarketTransport`, the unique market transport, index/Release and local-candidate owner | | `app/adapters/system/plugin/package.py` | `PluginPackageManager`, the unique install, checkpoint, backup, restore and physical-delete owner | | `app/adapters/system/plugin/dependency.py` | `PluginDependencyInstaller`, enabled-source dependency inspection and constrained per-plugin recovery owner; `async_install` has two completion-owned worker boundaries for request preparation and fallback planning, while package subprocess cancellation remains with `PluginRuntimeHealth` | diff --git a/tests/fixtures/architecture/dependency-baseline.json b/tests/fixtures/architecture/dependency-baseline.json index 073c7a5cc..33fe40746 100644 --- a/tests/fixtures/architecture/dependency-baseline.json +++ b/tests/fixtures/architecture/dependency-baseline.json @@ -959,8 +959,8 @@ "runtime_only": true } }, - "edge_count": 9963, - "edge_sha256": "44cf52b7ca35534e04cf6fa3de8aadcaa7f54ccdeec543b79156f6d7aed3b8a0", + "edge_count": 9965, + "edge_sha256": "95b80752f6b6f2ab82c79f2c5464b52b74a57bc13aecadb452dcb51e587b400b", "edges": [ "app -> app.foundation", "app -> app.foundation.environment", @@ -3853,6 +3853,8 @@ "app.application.messaging.chat -> app.application.database", "app.application.messaging.chat -> app.application.messaging", "app.application.messaging.chat -> app.application.messaging.recall", + "app.application.messaging.chat -> app.domain", + "app.application.messaging.chat -> app.domain.chat", "app.application.messaging.chat -> app.runtime", "app.application.messaging.chat -> app.runtime.observability", "app.application.messaging.chat -> app.runtime.priority", diff --git a/tests/test_agent_chat_timestamps.py b/tests/test_agent_chat_timestamps.py new file mode 100644 index 000000000..99e5d4b98 --- /dev/null +++ b/tests/test_agent_chat_timestamps.py @@ -0,0 +1,161 @@ +""" +选择回执的时间戳保留及历史会话读取兼容回归。 +""" + +from copy import deepcopy +from types import SimpleNamespace +from uuid import uuid4 + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from pydantic import ValidationError + +from app.api.endpoints import agent as agent_endpoint +from app.application.messaging.chat import get_configured_agent_chat_service +from app.db.models.agentchat import AgentChat +from app.db.oper.agentchat import AgentChatOper +from app.db.session import SessionFactory +from app.domain.chat import merge_choice_state, normalize_display_timestamps +from app.schemas.agent import AgentChatMessage + + +def _receipt(**fields: object) -> dict[str, object]: + """ + 构造带稳定标识的选择回执,允许省略时间戳以重放旧数据。 + + :param fields: 回执需要补充或覆盖的展示字段 + :return: 用户选择电影的展示回执 + """ + return { + "id": "choice-selection-test", "role": "user", "content": "", + "choice_selection": {"choice_id": "test", "selected_label": "电影", "selected_value": "选择电影"}, + **fields, + } + + +def test_replacement_keeps_original_receipt_timestamp_across_stale_snapshots() -> None: + """ + 旧快照省略回执或重复提交回执时,首次选择和原始时间都保持稳定。 + """ + card = {"id": "card", "role": "assistant", "content": "", "createdAt": 1000, + "choices": [{"id": "test", "status": "pending"}]} + original = [card, _receipt(createdAt=1010)] + before = deepcopy(original) + saved = original + for incoming in ([card], [], [card, _receipt(createdAt=9000)], [card, _receipt()], [card]): + saved = merge_choice_state(saved, incoming) + receipt = next(message for message in saved if message.get("choice_selection")) + assert AgentChatMessage.model_validate(receipt).createdAt == 1010 + assert receipt["choice_selection"] == original[1]["choice_selection"] + assert original == before + + +def test_selected_card_recovery_supplies_receipt_timestamp() -> None: + """ + 旧记录只有已选卡片时,重建回执使用卡片时间并保留选择内容。 + """ + original = [{"id": "card", "role": "assistant", "content": "", "createdAt": 1234.5, + "choices": [{"id": "test", "status": "selected", "selected_value": "选择电影"}]}] + result = merge_choice_state(original, []) + assert len(result) == 1 + receipt = AgentChatMessage.model_validate(result[0]) + assert receipt.id == "choice-selection-test" + assert receipt.createdAt == 1234.5 + assert receipt.choice_selection.selected_value == "选择电影" + + +@pytest.mark.parametrize("position", [0, 1, 2]) +def test_legacy_receipt_is_repaired_on_save_without_mutating_input(position: int) -> None: + """ + 历史回执缺少时间时借用相邻消息时间,保存后可读取且不修改旧内存快照。 + + :param position: 缺字段回执位于完整快照中的位置 + """ + original = [ + {"id": "first", "role": "user", "createdAt": 1000}, + {"id": "last", "role": "assistant", "createdAt": 2000}, + ] + original.insert(position, _receipt()) + before = deepcopy(original) + result = merge_choice_state([], original) + assert result[position]["createdAt"] == (2000 if position == 2 else 1000) + assert all(AgentChatMessage.model_validate(message) for message in result) + assert original == before + + +def test_legacy_receipt_without_any_known_time_uses_stable_unknown_timestamp() -> None: + """ + 完全缺少时间的旧快照返回固定未知值,重复保存不会伪造不断变化的创建时间。 + """ + result = merge_choice_state([], [_receipt()]) + assert AgentChatMessage.model_validate(result[0]).createdAt == 0 + assert merge_choice_state(result, []) == result + + +@pytest.mark.parametrize("timestamp", [None, "invalid"]) +def test_timestamp_compatibility_preserves_explicit_invalid_values(timestamp: object) -> None: + """ + 仅兼容缺失字段,显式非法值仍由原有模型验证报告。 + + :param timestamp: 已提供但不符合毫秒数值约定的字段值 + """ + result = normalize_display_timestamps([_receipt(createdAt=timestamp)]) + with pytest.raises(ValidationError, match="createdAt"): + AgentChatMessage.model_validate(result[0]) + + +@pytest.mark.parametrize("replace_snapshot", [False, True]) +def test_history_endpoint_reads_legacy_receipt_and_survives_old_snapshot_save( + monkeypatch: pytest.MonkeyPatch, replace_snapshot: bool, +) -> None: + """ + 实际详情路由能读取脏历史,且旧内存快照或缺少回执的快照再次写回后仍返回 200。 + + :param monkeypatch: 隔离运行中 Agent 管理器的 pytest 替换工具 + :param replace_snapshot: 后续保存完整旧快照或省略回执的快照 + """ + session_id = f"web-agent:timestamps-{uuid4().hex}" + legacy = [ + {"id": "first", "role": "user", "content": "选择电影", "createdAt": 1000}, + _receipt(), + {"id": "last", "role": "assistant", "content": "继续", "createdAt": 2000}, + ] + with SessionFactory() as db: + db.add(AgentChat(session_id=session_id, user_id="501", username="alice", + display_messages=legacy, message_count=len(legacy))) + db.commit() + monkeypatch.setattr(agent_endpoint.agent_application, "get_running_agent_manager", lambda: None) + app = FastAPI() + app.include_router(agent_endpoint.router) + app.dependency_overrides[agent_endpoint.get_agent_chat_service] = get_configured_agent_chat_service + app.dependency_overrides[agent_endpoint.get_current_active_user] = lambda: SimpleNamespace( + id=501, name="alice", is_superuser=False, + ) + with TestClient(app) as client: + for attempt in range(3): + response = client.get(f"/sessions/{session_id}") + assert response.status_code == 200, response.text + messages = response.json()["data"]["messages"] + assert len(messages) == 3 + assert next(message for message in messages if message.get("choice_selection"))["createdAt"] == 1000 + with SessionFactory() as db: + row = AgentChatOper(db).get(session_id) + if attempt == 0: + assert row.display_messages == legacy + AgentChatOper(db).save_display_messages( + session_id=session_id, user_id="501", + messages=legacy if not replace_snapshot else [legacy[0], legacy[2]], + ) + db.commit() + with SessionFactory() as db: + row = AgentChatOper(db).get(session_id) + assert all(AgentChatMessage.model_validate(message) for message in row.display_messages) + + +def test_history_timestamp_compatibility_does_not_weaken_save_request_validation() -> None: + """ + 历史兼容只发生在读取和合并边界,客户端展示消息仍必须提供时间戳。 + """ + with pytest.raises(ValidationError, match="createdAt"): + AgentChatMessage.model_validate(_receipt()) diff --git a/tests/test_agent_session_consistency.py b/tests/test_agent_session_consistency.py index a1563e440..8be7206c9 100644 --- a/tests/test_agent_session_consistency.py +++ b/tests/test_agent_session_consistency.py @@ -20,6 +20,7 @@ from app.application.messaging.interaction.agent import ( from app.db.oper.agentchat import AgentChatOper from app.db.session import SessionFactory from app.domain.chat import merge_choice_state +from app.schemas.agent import AgentChatMessage from app.schemas.types import NotificationChannel @@ -75,6 +76,7 @@ async def test_choice_callback_persists_first_selection_and_replays_without_chan assert row.display_messages[0]["choices"][0]["selected_value"] == "选择电影" receipts = [message for message in row.display_messages if message.get("choice_selection")] assert len(receipts) == 1 and receipts[0]["content"] == "" + assert AgentChatMessage.model_validate(receipts[0]).createdAt > 0 @pytest.mark.asyncio @@ -148,6 +150,7 @@ def test_selected_card_survives_old_pending_snapshot_without_mutating_inputs(cho before_card = merge_choice_state(previous, []) assert before_card == [{ "id": f"choice-selection-{request.request_id}", "role": "user", "content": "", "choice_selection": selected, + "createdAt": 0, }] result = merge_choice_state(before_card, original) assert result[0]["choices"][0]["status"] == "selected"