fix(agent): preserve choice receipt timestamps

Refs #7314
This commit is contained in:
jxxghp
2026-11-06 22:23:37 +08:00
parent 0e93eda0e8
commit db07e6ff0f
8 changed files with 237 additions and 33 deletions

View File

@@ -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

View File

@@ -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

View File

@@ -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) |

View File

@@ -74,6 +74,8 @@
后端按用户、会话和选择 ID 领取一次执行;入队前明确拒绝可重试,执行结果未知时不重放。
`echo_user=false` 仅隐藏用户回显,选择结果仍保存;完整快照替换保留隐藏选择收据,
即使旧快照不含卡片,也不能让后续快照把已选卡片改回 `pending`。普通消息仍按完整快照替换。
选择回执合并时保留原始 `createdAt` 毫秒时间戳;旧历史缺少该字段时,读取和保存边界
使用相邻消息时间补齐,不丢弃选择内容。整份快照都没有时间时以 `0` 表示未知。
会话 `is_processing` 只表示实际执行或排队,空闲等待下一条消息的 worker 不计入。
工具事件沿用 `tool_id` 与 `assistant_message_id`,跨 steering 后仍更新原助手段。

View File

@@ -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` |

View File

@@ -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",

View File

@@ -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())

View File

@@ -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"