huangruiteng · Nov 3, 2026 · Nov 3, 2026 · Nov 3, 2026 · Nov 3, 2026
@@ -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);

@@ -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<ChatSessionSnapshot> {
+export async function fetchChatSessionHistory(session: ChatSessionSummary | string, signal?: AbortSignal, turnId?: string): Promise<ChatSessionSnapshot> {
   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<ChatSessionSnapshot>(
       `/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)))) {

@@ -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.");

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

@@ -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 = ["核对账单导出的字段映射", "检查重复导出是否会生成重复记录", "复核历史账单的金额差异",
             "整理迁移后的回滚步骤", "补齐取消导出时的错误提示", "核对欧洲地区的税额精度",

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

@@ -277,6 +277,26 @@ provider、授权、回执及结束时的来源检查仍实时执行。
 会话和 Goal 范围复用该次选择；检查点及请求之间不缓存权限。自定义范围解析器沿用
 原验证路径。
 
+Expanding **Original conversation** reads the complete original Turn through
+`GET /api/chat/sessions/<session>?message_limit=200&turn_id=<original-turn>`.
+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

@@ -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 {}),
     }}
@@ -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"]]

@@ -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 {},