diff --git a/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs b/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs index 6f7636292a..8b8ecf52ae 100644 --- a/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs +++ b/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs @@ -109,9 +109,20 @@ try { let historyMode = 'hold'; const seenHistory = new Promise(resolve => sawHistory = resolve); await page.route(`**/api/chat/sessions/${ready.session_id}?*`, async route => { - if (!new URL(route.request().url()).searchParams.has('message_cursor')) return route.continue(); + const url = new URL(route.request().url()); + assert.ok(url.searchParams.get('turn_id'), 'Original reading must select its exact Turn'); + // The real owner serves one-message pages to exercise full-Turn staging + // without fetching hundreds of unrelated messages to force a boundary. + url.searchParams.set('message_limit', '1'); + if (!url.searchParams.has('message_cursor')) return route.fulfill({response: await route.fetch({url: url.href})}); if (historyMode === 'fault') return route.fulfill({status: 503, json: {ok: false}}); - const response = await route.fetch(); + const response = await route.fetch({url: url.href}); + if (historyMode === 'wrong-turn' || historyMode === 'foreign-message') { + const body = await response.json(); + if (historyMode === 'wrong-turn') body.message_page.turn_id = 'other-original'; + else body.messages[0].turn_id = 'other-original'; + return route.fulfill({json: body}); + } if (historyMode === 'hold') {sawHistory(); await new Promise(resolve => releaseHistory = resolve);} return route.fulfill({response}).catch(() => {}); }); @@ -130,6 +141,10 @@ try { await page.setViewportSize({width: 1440, height: 1000}); await original.click(); historyMode = 'fault'; await original.click(); await dialog.getByRole('alert').waitFor(); assert.equal(await articles.count(), 0); + for (const invalid of ['wrong-turn', 'foreign-message']) { + await original.click(); historyMode = invalid; await original.click(); + await dialog.getByRole('alert').waitFor(); assert.equal(await articles.count(), 0); + } await original.click(); historyMode = 'normal'; await original.click(); await articles.nth(1).waitFor(); await original.click(); historyMode = 'hold'; const lateHistory = new Promise(resolve => sawHistory = resolve); diff --git a/apps/presentation/dashboard/src/data/chat.ts b/apps/presentation/dashboard/src/data/chat.ts index 5b46e1cc0a..556fcf34dd 100644 --- a/apps/presentation/dashboard/src/data/chat.ts +++ b/apps/presentation/dashboard/src/data/chat.ts @@ -969,10 +969,11 @@ const chatMessagePageSchema = z.object({ schema_version: z.literal("loopx_chat_message_page_v1"), offset: z.number().int().nonnegative(), total: z.number().int().nonnegative(), snapshot_sha256: z.string().regex(/^[a-f0-9]{64}$/), next_cursor: z.string().min(1).max(1024).nullable(), + turn_id: z.string().min(1).optional(), }); /** Hydrate one complete original prefix; partial pages never become a snapshot. */ -export async function fetchChatSessionHistory(session: ChatSessionSummary | string, signal?: AbortSignal): Promise { +export async function fetchChatSessionHistory(session: ChatSessionSummary | string, signal?: AbortSignal, turnId?: string): Promise { const sessionId = typeof session === "string" ? session : session.session_id; // A delegation has the original Session ID; pin its first verified page's // scope. History discovery callers keep validating against their own source. @@ -983,6 +984,7 @@ export async function fetchChatSessionHistory(session: ChatSessionSummary | stri do { signal?.throwIfAborted(); const query = new URLSearchParams({message_limit: "200"}); + if (turnId !== undefined) query.set("turn_id", turnId); if (cursor !== null) query.set("message_cursor", cursor); const snapshot = await requestJson( `/api/chat/sessions/${encodeURIComponent(sessionId)}?${query}`, @@ -1001,12 +1003,13 @@ export async function fetchChatSessionHistory(session: ChatSessionSummary | stri // page cannot silently lose the cursor contract and masquerade as complete. if (!Object.hasOwn(snapshot, "message_page")) { if (cursor !== null) throw new Error("The history page contract changed during read."); - return snapshot; + return turnId === undefined ? snapshot : {...snapshot, messages: snapshot.messages.filter(row => row.turn_id === turnId)}; } const parsed = chatMessagePageSchema.safeParse(snapshot.message_page); if (!parsed.success) throw new Error("The original history page could not be verified."); const page = parsed.data, end = messages.length + snapshot.messages.length; - if (page.offset !== messages.length || snapshot.messages.length > 200 || end > page.total + if (page.turn_id !== turnId || (turnId !== undefined && snapshot.messages.some(row => row.turn_id !== turnId)) + || page.offset !== messages.length || snapshot.messages.length > 200 || end > page.total || (total !== undefined && total !== page.total) || (digest !== undefined && digest !== page.snapshot_sha256) || (page.next_cursor !== null) !== (end < page.total) || (page.next_cursor !== null && (!snapshot.messages.length || cursors.has(page.next_cursor)))) { diff --git a/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx b/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx index 7bc360d8d1..36ca5afb60 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/conversation-requests.tsx @@ -174,7 +174,7 @@ function OriginalRequest({row, zh, revision, active}: {row: ConversationRequest; setError(false); if (!open || !active) {setMessages(null); return;} const abort = new AbortController(); - void fetchChatSessionHistory(row.source_session_id, abort.signal).then(value => { + void fetchChatSessionHistory(row.source_session_id, abort.signal, row.source_turn_id).then(value => { if (abort.signal.aborted) return; const source = value.messages.filter(message => message.turn_id === row.source_turn_id); if (!source.length) throw new Error("Original turn unavailable."); diff --git a/examples/issue-fix-pr-lifecycle-smoke.py b/examples/issue-fix-pr-lifecycle-smoke.py index e47bf3feae..9bd67f55bf 100644 --- a/examples/issue-fix-pr-lifecycle-smoke.py +++ b/examples/issue-fix-pr-lifecycle-smoke.py @@ -568,7 +568,6 @@ def check_lifecycle(provider: str) -> None: if item["todo_id"] == "todo_merge_gate_1716" ) assert gate_item["status"] == "done", gate_item - assert "state=MERGED" in gate_item["evidence"], gate_item assert_public_safe(reconciled_gate) repeated_gate = json.loads( @@ -696,7 +695,6 @@ def check_lifecycle(provider: str) -> None: ) assert review_item["status"] == "done", review_item assert review_item["no_followup"] is True, review_item - assert "state=MERGED" in review_item["evidence"], review_item assert_public_safe(reconciled_review) repeated_review = json.loads( @@ -796,6 +794,7 @@ def check_lifecycle(provider: str) -> None: for todo_id in ("todo_merge_gate_1716", "todo_review_action_1716"): assert rows[todo_id]["status"] == "done", rows[todo_id] assert rows[todo_id]["last_actor_agent_id"] == "codex-test", rows[todo_id] + assert "state=MERGED" in rows[todo_id]["evidence"], rows[todo_id] def main() -> int: diff --git a/examples/personal-workspace-browser/steward_requests_fixture.py b/examples/personal-workspace-browser/steward_requests_fixture.py index bb5eb215b5..179107d860 100644 --- a/examples/personal-workspace-browser/steward_requests_fixture.py +++ b/examples/personal-workspace-browser/steward_requests_fixture.py @@ -31,12 +31,16 @@ def main(): args.add_argument("--status-file", type=Path, required=True) args.add_argument("--assets-dir", type=Path) args.add_argument("--long-history", action="store_true") + args.add_argument("--earlier-messages", type=int, default=199, + help="Unrelated message prefix when --long-history is selected") args.add_argument("--multi-recipient", action="store_true") args.add_argument("--canonical-storage", choices=("sqlite", "file"), default="sqlite") args.add_argument("--requests", type=int, default=13) selected = args.parse_args() if not 1 <= selected.requests <= 120: args.error("--requests must be between 1 and 120") + if not 0 <= selected.earlier_messages <= 5000: + args.error("--earlier-messages must be between 0 and 5000") status = json.loads(selected.status_file.read_text()) with TemporaryDirectory(prefix="loopx-steward-requests-") as directory, ExitStack() as cleanup: root = Path(directory).resolve() @@ -77,7 +81,7 @@ def main(): agent_model_id="gpt-6.1-sol", agent_reasoning_effort="xhigh") if selected.long_history: # The first delegated question and its answer straddle a real page. - for index in range(199): + for index in range(selected.earlier_messages): store.append_message(session["session_id"], role="user", text=f"Earlier original message {index}") titles = ["核对账单导出的字段映射", "检查重复导出是否会生成重复记录", "复核历史账单的金额差异", "整理迁移后的回滚步骤", "补齐取消导出时的错误提示", "核对欧洲地区的税额精度", diff --git a/loopx/canary/module_metric_baseline.json b/loopx/canary/module_metric_baseline.json index b97807447d..7f94359675 100644 --- a/loopx/canary/module_metric_baseline.json +++ b/loopx/canary/module_metric_baseline.json @@ -41,7 +41,7 @@ "loopx/chat_server.py": { "any_count": 25, "dict_any_count": 0, - "lines": 2035 + "lines": 2039 }, "loopx/chat_store.py": { "lines": 2111 diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md index 0b713106b2..31b3871e5f 100644 --- a/loopx/capabilities/manager_context/README.md +++ b/loopx/capabilities/manager_context/README.md @@ -277,6 +277,26 @@ provider、授权、回执及结束时的来源检查仍实时执行。 会话和 Goal 范围复用该次选择；检查点及请求之间不缓存权限。自定义范围解析器沿用 原验证路径。 +Expanding **Original conversation** reads the complete original Turn through +`GET /api/chat/sessions/?message_limit=200&turn_id=`. +The optional selector excludes unrelated Turns before history projection and +paging; it does not shorten any selected message or relax strict journal reads. +`message_page.turn_id` echoes the selection, and opaque cursors bind both the +Session and Turn. Append boundaries, prefix integrity, collaboration carriers, +return delivery and provider-owned reply context retain their existing owners. +A missing Turn, changed cursor scope or damaged journal fails visibly. This +historical read does not recover an unrelated active execution. Unscoped Chat +history keeps its existing response and recovery behavior. The scoped client +needs the matching server; a paged response without its exact scope is rejected, +while an older complete snapshot can still be filtered locally. + +展开**原对话**时，通过同一历史接口按原 Turn 完整读取，再沿原分页协议补齐这一轮的 +全部消息；不再为了筛选一轮而读完会话的其它轮次。完整正文、附件、原引用与返回状态 +仍由原 owner 投影，严格日志检查、快照前缀和跨 Session/Turn 游标隔离保留。读取历史 +不恢复其它正在执行的轮次；不带选择器的普通历史及其恢复行为不变。缺失、损坏或来源 +不匹配时显示失败，可重新展开重试。新分页客户端需要配套服务端；旧版完整快照仍可在 +本地筛选。此路径不新增权限、缓存、执行入口或独立能力。 + The automatic agenda is a partial read (`read_complete=false`): it cannot fulfill the full Inbox read or replace its requirements. Before acting, read the current brief and independently assess its authority, evidence and constraints. The diff --git a/loopx/capabilities/manager_context/chat_history.py b/loopx/capabilities/manager_context/chat_history.py index 3b54355130..ba068362ad 100644 --- a/loopx/capabilities/manager_context/chat_history.py +++ b/loopx/capabilities/manager_context/chat_history.py @@ -10,11 +10,12 @@ def page_chat_messages(session_id: str, messages: list[dict[str, Any]], *, - limit: int, cursor: str | None = None) -> dict[str, Any]: + limit: int, cursor: str | None = None, turn_id: str | None = None) -> dict[str, Any]: """Page an append-ordered frozen prefix; a cursor is a read hint, not authority.""" if type(limit) is not int or not 1 <= limit <= 200: raise ValueError("message_limit must be between 1 and 200") offset, total = 0, len(messages) + source_identity = session_id if turn_id is None else [session_id, turn_id] expected_digest = None if cursor is not None: try: @@ -23,7 +24,7 @@ def page_chat_messages(session_id: str, messages: list[dict[str, Any]], *, decoded = json.loads(base64.b64decode(cursor + "=" * (-len(cursor) % 4), altchars=b"-_", validate=True)) source, offset, total, expected_digest = decoded - if (source != session_id or type(offset) is not int or type(total) is not int + if (source != source_identity or type(offset) is not int or type(total) is not int or not 0 < offset < total <= len(messages) or not isinstance(expected_digest, str) or not BARE_SHA256_PATTERN.fullmatch(expected_digest)): @@ -35,9 +36,10 @@ def page_chat_messages(session_id: str, messages: list[dict[str, Any]], *, if expected_digest is not None and expected_digest != digest: raise ValueError("history cursor source changed; restart the read") end = min(offset + limit, total) - next_cursor = base64.urlsafe_b64encode(json.dumps([session_id, end, total, digest], + next_cursor = base64.urlsafe_b64encode(json.dumps([source_identity, end, total, digest], separators=(",", ":")).encode("utf-8")).decode("ascii").rstrip("=") if end < total else None return {"messages": messages[offset:end], "message_page": { "schema_version": "loopx_chat_message_page_v1", "offset": offset, "total": total, "snapshot_sha256": digest, "next_cursor": next_cursor, + **({"turn_id": turn_id} if turn_id is not None else {}), }} diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py index 489ad6d7a9..4d76d93003 100644 --- a/loopx/capabilities/manager_context/roundtrip.py +++ b/loopx/capabilities/manager_context/roundtrip.py @@ -334,14 +334,23 @@ def candidate_paths(): def project_chat_session_snapshot(root, store, session_id, *, registry, - message_limit=None, message_cursor=None): + message_limit=None, message_cursor=None, turn_id=None): """Project return delivery state into one existing Chat snapshot.""" snapshot = (store.session_snapshot(session_id, strict_messages=True) if message_limit is not None else store.session_snapshot(session_id)) + if turn_id is not None: + if message_limit is None: + raise ValueError("a selected original Turn requires paged history") + turn = store.load_turn(session_id, turn_id) + if not turn or turn.get("session_id") != session_id or turn.get("turn_id") != turn_id: + raise ValueError("original Turn is unavailable from this Session") + # OriginalRequest already displays only this Turn. Select it before + # paging/projection, retaining every message and the strict journal read. + snapshot["messages"] = [row for row in snapshot["messages"] if row.get("turn_id") == turn_id] from .chat_history import page_chat_messages page = (page_chat_messages(session_id, snapshot["messages"], limit=message_limit, - cursor=message_cursor) if message_limit is not None else None) + cursor=message_cursor, turn_id=turn_id) if message_limit is not None else None) context_messages = snapshot["messages"] if page is not None: context_messages = context_messages[:page["message_page"]["total"]] diff --git a/loopx/chat_server.py b/loopx/chat_server.py index b2e6b54346..1166fe1a33 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -922,20 +922,24 @@ def _session_turn(self, session_id: str) -> None: def _session_snapshot(self, session_id: str) -> None: try: query = parse_qs(urlparse(self.path).query, keep_blank_values=True) - if (set(query) - {"message_limit", "message_cursor"} + if (set(query) - {"message_limit", "message_cursor", "turn_id"} or any(len(values) != 1 or not values[0] for values in query.values()) - or ("message_cursor" in query and "message_limit" not in query)): + or (any(key in query for key in ("message_cursor", "turn_id")) and "message_limit" not in query)): raise ValueError("invalid history page arguments") message_limit = int(query["message_limit"][0]) if "message_limit" in query else None if message_limit is not None and not 1 <= message_limit <= 200: raise ValueError("message_limit must be between 1 and 200") - self.server.runtime_controller.recover_exited_turn(session_id) + # Reading one historical Turn must not recover an unrelated active + # execution. Ordinary Chat refresh keeps its existing recovery path. + if "turn_id" not in query: + self.server.runtime_controller.recover_exited_turn(session_id) snapshot = cast(Callable[..., dict[str, object]], project_chat_session_snapshot) payload = snapshot( self.server.runtime_controller.coordination_runtime_root, self.server.chat_store, session_id, registry=self.server.registry_path, message_limit=message_limit, - message_cursor=query.get("message_cursor", [None])[0]) + message_cursor=query.get("message_cursor", [None])[0], + turn_id=query.get("turn_id", [None])[0]) # The HTTP composition root joins provider history onto the selected page. payload["messages"] = project_private_reply_history( self.server.chat_store, self.server.chat_store.load_session(session_id) or {}, diff --git a/tests/test_chat_conversation_bindings.py b/tests/test_chat_conversation_bindings.py index 318d7f51b4..f6a848d037 100644 --- a/tests/test_chat_conversation_bindings.py +++ b/tests/test_chat_conversation_bindings.py @@ -345,6 +345,33 @@ def test_return_delivery_readback_survives_a_separate_history_page(flow): # noq assert sum(bool(row.get("collaboration")) for row in rows) == 1 +def test_turn_history_matches_full_history_collaboration_and_returns(flow): # noqa: F811 + from loopx.capabilities.manager_context import acknowledge + from loopx.capabilities.manager_context.roundtrip import drain, report, project_chat_session_snapshot + + root, registry, store, create = flow + session, turn, receipt = create(brief=_history_page_brief()) + sid, tid = session["session_id"], turn["turn_id"] + store.append_message(sid, role="agent", text="Original commitment", turn_id=tid) + acknowledge(root, "research", "worker", receipt["request_id"], "adopt", "Keep original audience") + report(root, "research", "worker", receipt["request_id"], "conclusion", "Original conclusion") + drain(root, registry, store, lambda *_: pytest.fail("A private App return cannot send to a provider")) + store.append_message(sid, role="user", text="Unrelated history") + whole = project_chat_session_snapshot(root, store, sid, registry=registry) + expected = [row for row in whole["messages"] if row.get("turn_id") == tid] + cursor, rows = None, [] + while True: + page = project_chat_session_snapshot(root, store, sid, registry=registry, + message_limit=1, message_cursor=cursor, turn_id=tid) + rows.extend(page["messages"]) + cursor = page["message_page"]["next_cursor"] + if cursor is None: + break + assert rows == expected + assert sum(bool(row.get("collaboration")) for row in rows) == 1 + assert rows[-1].get("return_delivery"), "Scoped history retains actual return delivery readback" + + def test_original_history_pages_freeze_append_boundary_and_keep_full_rows(ordinary, monkeypatch): # noqa: F811 from urllib.parse import urlencode from loopx.chat_server import ChatHTTPServer @@ -388,6 +415,63 @@ def unreadable(*_, **__): assert code == 503 and "private journal path" not in json.dumps(unavailable) +def test_original_turn_history_keeps_complete_rows_and_scope_fences(ordinary, monkeypatch): # noqa: F811 + from urllib.parse import urlencode + from loopx.chat_server import ChatHTTPServer + + store, runtime, contexts, request, capture, _, _ = ordinary + monkeypatch.setattr(ChatHTTPServer, "runtime_root", store.root.parent, raising=False) + monkeypatch.setattr(ChatHTTPServer, "registry_path", runtime.registry_path, raising=False) + _, opened = request("/api/chat/sessions", { + "context_kind": "project", "project_ref": contexts.available()[0]["project_ref"], + }) + sid = opened["session_id"] + for i in range(430): + store.append_message(sid, role="user", text=f"Unrelated history {i}") + turn, _ = store.create_turn(sid, client_turn_id="selected-original", message="完整原始要求" * 3000, origin="web") + tid = turn["turn_id"] + for i in range(202): + store.append_message(sid, role="agent", text=f"Original response {i}", turn_id=tid, + attachments=[{"name": "original.png", "data_url": "data:image/png;base64,AA=="}] if i == 201 else None) + store.update_turn(sid, tid, status="completed", response={"message": "Original response 201"}) + path = f"/api/chat/sessions/{sid}" + before = capture.read_bytes() + whole = request(path)[1] + expected = [row for row in whole["messages"] if row.get("turn_id") == tid] + recovered, recover = [], runtime.recover_exited_turn + def observe_recovery(source): + recovered.append(source) + return recover(source) + monkeypatch.setattr(runtime, "recover_exited_turn", observe_recovery) + query = {"message_limit": 200, "turn_id": tid} + code, first = request(path + "?" + urlencode(query)) + assert code == 200 and first["message_page"]["turn_id"] == tid + assert len(first["messages"]) == 200 and first["message_page"]["total"] == len(expected) == 203 + cursor = first["message_page"]["next_cursor"] + store.append_message(sid, role="agent", text="Outside the frozen original prefix", turn_id=tid) + code, rest = request(path + "?" + urlencode({**query, "message_cursor": cursor})) + assert code == 200 and rest["message_page"]["next_cursor"] is None + assert first["messages"] + rest["messages"] == expected + assert recovered == [], "A historical Turn read must not recover the Session's active execution" + assert capture.read_bytes() == before, "A scoped read cannot execute or recover model work" + assert "turn_id" not in request(path + "?message_limit=200")[1]["message_page"] + assert recovered == [sid], "An unscoped Chat refresh keeps its ordinary recovery behavior" + other, _ = store.create_turn(sid, client_turn_id="other-original", message="Other question", origin="web") + store.update_turn(sid, other["turn_id"], status="completed", response={"message": "Other answer"}) + for selected in [None, other["turn_id"]]: + wrong = {"message_limit": 200, "message_cursor": cursor, **({"turn_id": selected} if selected else {})} + assert request(path + "?" + urlencode(wrong))[0] == 400 + for invalid in ["turn_id=missing", "message_limit=200&turn_id=missing", "message_limit=200&turn_id=..", + "message_limit=200&turn_id=", "message_limit=200&turn_id=a&turn_id=b"]: + assert request(path + "?" + invalid)[0] == 400 + journal = store._session_dir(sid) / "messages.jsonl" + original = journal.read_bytes() + journal.write_bytes(original + b'{invalid record}\n') + assert request(path + "?" + urlencode(query))[0] == 400 + journal.write_bytes(original) + assert request(path + "?" + urlencode(query))[1]["message_page"]["total"] == 204 + + def test_app_steward_corrupt_legacy_configuration_never_falls_back(tmp_path): bindings = ChatConversationBindings(root=tmp_path, project_contexts=None, observe=lambda _: pytest.fail("an unselected entry must not probe a provider")) diff --git a/tests/test_manager_context_roundtrip.py b/tests/test_manager_context_roundtrip.py index 23409095db..616a6b32d8 100644 --- a/tests/test_manager_context_roundtrip.py +++ b/tests/test_manager_context_roundtrip.py @@ -113,14 +113,15 @@ def read(**query): thread.join(timeout=2) -def test_frozen_private_reply_projects_original_text_without_mutating_model_input(frozen_private_reply, tmp_path, private_snapshot): +@pytest.mark.parametrize("scoped", [False, True]) +def test_frozen_private_reply_projects_original_text_without_mutating_model_input(frozen_private_reply, tmp_path, private_snapshot, scoped): store, session, turn, _, _, _, source_text, parent = frozen_private_reply before = {str(p): p.read_bytes() for p in store.root.rglob("*") if p.is_file()} # The core projection carries no provider history; the composition root joins it. generic = project_chat_session_snapshot(tmp_path, store, session["session_id"], registry=tmp_path / "absent.json", message_limit=1) assert "reply_context" not in generic["messages"][0] - snapshot = private_snapshot(message_limit=1) + snapshot = private_snapshot(message_limit=1, **({"turn_id": turn["turn_id"]} if scoped else {})) message = snapshot["messages"][0] assert message["text"] == turn["message"] assert message["reply_context"]["source_text"] == source_text