fix(chat): show original replies with collapsed quote and thread context

apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py

@@ -35,8 +35,13 @@ def __init__(self, root, *, steward=False, peer=False):
             self.transport_ref = "synthetic-peer"
 
     def observe(self):
-        return {"transport_ref": self.transport_ref, "provider_ref": ("e" if self.peer else "c") * 24,
-                "operator_ref": ("f" if self.peer else "d") * 24, "verified": self.verified}
+        provider = ("e" if self.peer else "c") * 24
+        operator = ("f" if self.peer else "d") * 24
+        if "--reply-context" in sys.argv and not self.peer:
+            from loopx.extensions.lark.conversation_identity import identity_ref
+            operator = identity_ref(provider, self.reply_event()["sender_id"])
+        return {"transport_ref": self.transport_ref, "provider_ref": provider,
+                "operator_ref": operator, "verified": self.verified}
 
     def health_snapshot(self):
         return {"status": "listening"}
@@ -54,6 +59,9 @@ def start(self, server):
             executor_endpoint_id="codex", context_kind="steward" if self.steward else "project", project_grant="workspace_read")
         source = {"source_ref": ("b" if self.peer else "a") * 24,
                   "sender_ref": ("f" if self.peer else "d") * 24, "private_human_message": True}
+        if "--reply-context" in sys.argv and not self.peer:
+            from loopx.extensions.lark.conversation_identity import lark_private_source
+            source = lark_private_source(provider_ref=self.observe()["provider_ref"], event=self.reply_event())
         original, _ = runtime.open_session(goal_id=None, agent_id="codex", objective="",
             work_dir=self.root / "project", mode="resume_latest",
             conversation_binding_id=binding["binding_id"], source_context=source)
@@ -63,6 +71,42 @@ def start(self, server):
             return
         server.chat_store.append_message(sid, role="user", text="Synthetic private conversation: review the research notes.")
         server.chat_store.append_message(sid, role="agent", text="The original conversation and its research notes are ready.")
+        reply_evidence = {}
+        if "--reply-context" in sys.argv:
+            from loopx.control_plane.collaboration import conversation_reply_context
+            from loopx.extensions.lark.conversation_identity import identity_ref
+            from loopx.chat_store import _atomic_write_json
+            event = self.reply_event()
+            parent = "原要求：保留已有结果和失败记录。Original requirement: retain results and failed attempts."
+            body = "请比较这次修改，并保留原话题中的限制。Compare this change against the original constraints."
+            raw = {**event, "source_conversation_id": event["chat_id"], "reply_context": {
+                "message_id": event["parent_id"], "conversation_id": event["chat_id"], "content": parent},
+                "thread_context": {"conversation_id": event["chat_id"], "root_message_id": event["root_id"],
+                    "thread_id": event["thread_id"], "messages": [
+                        {"message_id": event["root_id"], "conversation_id": event["chat_id"], "thread_id": event["thread_id"],
+                         "position": -1, "content": "### 话题背景 / Topic background\n\n" + "完整上下文保持可读。Keep the original context readable. " * 35,
+                         "sender": {"id": "ou_fixture_owner", "kind": "user"}},
+                        {"message_id": event["parent_id"], "conversation_id": event["chat_id"], "thread_id": event["thread_id"],
+                         "position": 0, "content": parent, "sender": {"id": "cli_fixture", "kind": "app"}},
+                        {"message_id": event["message_id"], "conversation_id": event["chat_id"], "thread_id": event["thread_id"],
+                         "position": 1, "content": body}]}}
+            frozen = conversation_reply_context(raw)["context_text"] + "\n\nCurrent user message:\n" + body
+            request = identity_ref(self.observe()["provider_ref"], event["message_id"])
+            turn, _ = server.chat_store.create_turn(sid, client_turn_id=f"external-{request}", message=frozen, origin="lark")
+            correlation = {"request_ref": request, "binding_id": binding["binding_id"], "source": source,
+                           "session_id": sid, "turn_id": turn["turn_id"]}
+            for directory, value in (("external-requests", {**correlation, "status": "accepted", "command": None, "message": frozen}),
+                    ("lark-private-deliveries", {**correlation, "status": "delivered", "event": event, "reply_context_message": frozen, "reply_context_source_text": body})):
+                _atomic_write_json(server.chat_store.root / directory / f"{request}.json", value)
+            server.chat_store.update_turn(sid, turn["turn_id"], status="completed")
+            server.chat_store.release_active_turn(sid, turn["turn_id"])
+            server.chat_store.append_message(sid, role="agent", text="已保留原要求，当前核验结果待进一步确认。", turn_id=turn["turn_id"])
+            paths = [server.chat_store.root / directory / f"{request}.json"
+                     for directory in ("external-requests", "lark-private-deliveries")]
+            paths.extend([server.chat_store.sessions_root / sid / "messages.jsonl",
+                          server.chat_store.sessions_root / sid / "turns" / f"{turn['turn_id']}.json"])
+            reply_evidence = {path: path.read_bytes() for path in paths}
+            bindings.select_app_steward(binding["binding_id"], expected_revision=bindings.read()["revision"])
         other, _ = runtime.open_session(goal_id=None, agent_id="codex", objective="", mode="new",
             work_dir=self.root / "project", project_ref=binding["project_ref"])
         server.chat_store.append_message(other["session_id"], role="agent", text="Unrelated desktop conversation — must remain separate.")
@@ -112,6 +156,10 @@ def unavailable(handler):
                     result = {"sessions": len(server.chat_store.list_sessions()), "original_thread": saved["upstream_thread_id"],
                               "messages": server.chat_store.messages(sid),
                               "native_requests": len((self.root / "synthetic-requests.jsonl").read_text().splitlines())}
+                    if reply_evidence:
+                        result.update(model_turns=sum(json.loads(line).get("method") == "turn/start"
+                            for line in (self.root / "synthetic-requests.jsonl").read_text().splitlines()),
+                            reply_evidence_unchanged=all(path.read_bytes() == content for path, content in reply_evidence.items()))
                 elif command == "stop":
                     server.shutdown()
                     return
@@ -120,6 +168,12 @@ def unavailable(handler):
                 print(json.dumps(result), flush=True)
         threading.Thread(target=control, daemon=True).start()
 
+    @staticmethod
+    def reply_event():
+        return {"message_id": "om_fixture_reply", "parent_id": "om_fixture_parent", "root_id": "om_fixture_root",
+                "thread_id": "omt_fixture_topic", "chat_id": "oc_fixture_private", "chat_type": "p2p",
+                "sender_type": "user", "sender_id": "ou_fixture_owner"}
+
 
 def main():
     with tempfile.TemporaryDirectory(prefix="loopx-bound-ui-") as directory, pytest.MonkeyPatch.context() as patch:
@@ -134,9 +188,9 @@ def main():
             '        output.write(json.dumps(request) + "\\n")\n    method = request.get("method")')
         fake.write_text(model)
         fake.chmod(0o700)
-        steward = "--steward" in sys.argv
+        steward = "--steward" in sys.argv or "--reply-context" in sys.argv
         source = SyntheticPrivateSource(root, steward=steward)
-        peer = SyntheticPrivateSource(root, steward=True, peer=True) if steward else None
+        peer = SyntheticPrivateSource(root, steward=True, peer=True) if steward and "--reply-context" not in sys.argv else None
         source.peer_provider = peer
         with socket.socket() as sock:
             sock.bind(("127.0.0.1", 0))

apps/presentation/dashboard/smoke/conversation-reply-browser-smoke.mjs

@@ -0,0 +1,94 @@
+// Packaged UI and real SQLite/Chat HTTP; only provider/model identity is synthetic.
+import assert from "node:assert/strict";
+import {spawn} from "node:child_process";
+import {once} from "node:events";
+import {createInterface} from "node:readline";
+import {mkdir} from "node:fs/promises";
+import {resolve} from "node:path";
+import {resolveTestPython} from "../../../../scripts/test-python.mjs";
+import {launchBrowser, loadPlaywright} from "../../../../examples/dashboard-browser-smoke-support.mjs";
+
+const repo = resolve(import.meta.dirname, "../../../..");
+const output = resolve(process.env.LOOPX_REPLY_OUTPUT ?? resolve(repo, "output/playwright/conversation-reply"));
+await mkdir(output, {recursive: true});
+const child = spawn(resolveTestPython({repoRoot: repo}), ["-u",
+  "apps/presentation/dashboard/smoke/bound-conversation-http-fixture.py", "--reply-context"],
+  {cwd: repo, env: {...process.env, LOOPX_USAGE_PING: "0"}, stdio: ["pipe", "pipe", "pipe"]});
+const exited = once(child, "exit"), queue = [], waiting = [];
+let stderr = "", browser;
+child.stderr.on("data", chunk => stderr += chunk);
+createInterface({input: child.stdout}).on("line", line => {
+  try {const value = JSON.parse(line); if (waiting.length) waiting.shift()(value); else queue.push(value);} catch { /* startup */ }
+});
+function next() {
+  return new Promise((resolve, reject) => {
+    const timer = setTimeout(() => reject(Error(`Fixture timeout: ${stderr}`)), 60000);
+    const receive = value => {clearTimeout(timer); resolve(value);};
+    if (queue.length) receive(queue.shift()); else waiting.push(receive);
+  });
+}
+try {
+  const ready = await next();
+  browser = await launchBrowser(loadPlaywright().chromium);
+  const page = await browser.newPage({viewport: {width: 1440, height: 1000}});
+  await page.addInitScript(() => {if (!localStorage.getItem("loopx-pw-locale")) localStorage.setItem("loopx-pw-locale", "zh-CN");});
+  const errors = [], writes = [];
+  page.on("pageerror", error => errors.push(error.message));
+  page.on("request", request => {if (request.method() !== "GET") writes.push(new URL(request.url()).pathname);});
+  const snapshot = await (await page.request.get(`${ready.origin}/api/chat/sessions/${ready.sessionId}`)).json();
+  const source = snapshot.messages.find(row => row.role === "user" && row.turn_id);
+  assert.ok(source.text.startsWith("Referenced message ("), "Canonical model input stays unchanged");
+  await page.goto(ready.origin + "/chat/");
+  await page.getByRole("navigation", {name: "管家视图"}).getByRole("button", {name: /^(Chat|对话)$/}).click();
+  const body = "请比较这次修改，并保留原话题中的限制。Compare this change against the original constraints.";
+  const parent = "原要求：保留已有结果和失败记录。Original requirement: retain results and failed attempts.";
+  if (process.env.LOOPX_REPLY_BEFORE === "1") {
+    await page.getByText(source.text, {exact: true}).waitFor();
+    await page.screenshot({path: resolve(output, "before-desktop.png"), animations: "disabled"});
+    assert.equal(await page.locator(".conversation-reply-context").count(), 0);
+    console.log("BASELINE: original request renders the model envelope without a reply reader.");
+  } else {
+    await page.locator(".personal-channel-timeline").getByText("已保留原要求，当前核验结果待进一步确认。", {exact: true}).waitFor();
+    assert.equal(await page.locator(".personal-channel-timeline").getByText(body, {exact: true}).count(), 1,
+      "The primary user message shows the original body without the model envelope");
+    assert.equal(source.reply_context.source_text, body);
+    assert.equal(await page.getByText(source.text, {exact: true}).count(), 0);
+    const context = page.locator(".personal-channel-timeline .conversation-reply-context");
+    assert.equal(await context.count(), 1);
+    assert.equal(await context.getAttribute("open"), null);
+    await page.screenshot({path: resolve(output, "default-desktop.png"), animations: "disabled"});
+    await context.locator("summary").focus();
+    await page.keyboard.press("Enter");
+    await context.getByText(parent, {exact: true}).waitFor();
+    assert.equal(await context.locator("blockquote").count(), 1);
+    assert.equal(await context.locator("section").count(), 1, "The exact parent is not repeated in thread history");
+    await context.getByText("话题上下文为节选", {exact: true}).waitFor();
+    await page.waitForFunction(() => {
+      const rect = document.querySelector(".conversation-reply-context summary").getBoundingClientRect();
+      return rect.top >= 74 && rect.bottom < innerHeight - 100;
+    });
+    await page.screenshot({path: resolve(output, "expanded-desktop.png"), animations: "disabled"});
+    await page.setViewportSize({width: 390, height: 844});
+    await context.locator("summary").scrollIntoViewIfNeeded();
+    assert.ok(await page.locator("body").evaluate(node => node.scrollWidth <= innerWidth + 1));
+    await page.screenshot({path: resolve(output, "expanded-narrow.png"), animations: "disabled"});
+    await context.locator("summary").click();
+    await page.screenshot({path: resolve(output, "default-narrow.png"), animations: "disabled"});
+    await page.evaluate(() => localStorage.setItem("loopx-pw-locale", "en"));
+    await page.reload();
+    await page.getByRole("navigation", {name: "Manager view"}).getByRole("button", {name: /^(Chat|Conversation)$/}).click();
+    await page.getByText("Reply and thread context", {exact: true}).waitFor();
+    await page.screenshot({path: resolve(output, "english-narrow.png"), animations: "disabled"});
+    assert.deepEqual(errors, []);
+    assert.ok(!writes.some(path => /\/turns(?:\/|$)/u.test(path)), "Browsing cannot submit a Turn");
+    child.stdin.write("inspect\n");
+    const inspection = await next();
+    assert.equal(inspection.model_turns, 0);
+    assert.equal(inspection.reply_evidence_unchanged, true);
+    console.log("conversation-reply-browser: passed (real bound source, original body, complete quote, bounded thread, desktop/narrow, keyboard, English and zero model Turns)");
+  }
+} finally {
+  await browser?.close();
+  child.stdin.end("stop\n");
+  if (child.exitCode === null) await exited;
+}

apps/presentation/dashboard/smoke/delegation-media-browser-smoke.mjs

@@ -62,6 +62,8 @@ try {
     const response = await route.fetch(), value = await response.json();
     const source = value.messages.find(message => message.role === 'user');
     source.text = body;
+    source.reply_context = {source_text: body, parent: {message_id: 'synthetic-original-parent',
+      content: 'Original quoted requirement remains available.', complete: true}};
     const image = (id, name, data_url = raster) => ({id, name, data_url, mime_type: 'image/png', size: 2048});
     source.attachments = mode === 'text' ? [] : [image('one', '原始图一.png'), image('two', 'Original image two.png')];
     if (mode === 'broken') source.attachments = [image('broken', '不可读图片.png', 'data:image/png;base64,AA==')];
@@ -94,7 +96,7 @@ try {
   await original.click();
   const articles = row.locator('article');
   await articles.first().waitFor();
-  assert.equal(await articles.first().locator('.personal-md').innerText(), body);
+  assert.equal(await articles.first().locator(':scope > .personal-md').innerText(), body);
   assert.equal(await row.getByText('Other turn must stay hidden.', {exact: true}).count(), 0);
   const images = row.locator('img');
   if (process.env.LOOPX_MEDIA_BEFORE === '1') {
@@ -104,6 +106,11 @@ try {
     console.log('BASELINE: the original full body is present, but both supplied images are absent.');
   } else {
     assert.equal(await images.count(), 2);
+    const quote = articles.first().locator('.conversation-reply-context');
+    assert.equal(await quote.getAttribute('open'), null);
+    await quote.locator('summary').click();
+    await quote.getByText('Original quoted requirement remains available.', {exact: true}).waitFor();
+    await quote.locator('summary').click();
     assert.deepEqual(await images.evaluateAll(nodes => nodes.map(node => node.alt)), ['原始图一.png', 'Original image two.png']);
     await images.first().waitFor({state: 'visible'});
     await page.waitForFunction(() => [...document.querySelectorAll('.conversation-request-images img')].every(img => img.complete && img.naturalWidth === 600));
@@ -122,7 +129,7 @@ try {
     await page.waitForFunction(() => document.querySelector('.conversation-requests-body')?.getAttribute('aria-busy') === 'false');
     assert.equal(await original.evaluate(node => node.parentElement.open), true,
       'A same-source refresh must keep the original conversation expanded');
-    assert.equal(await articles.first().locator('.personal-md').innerText(), body);
+    assert.equal(await articles.first().locator(':scope > .personal-md').innerText(), body);
     assert.equal(await images.count(), 2);
     const refreshOriginal = async nextMode => {
       mode = nextMode;

apps/presentation/dashboard/src/data/chat.ts

@@ -901,6 +901,8 @@ export type ChatVisibleMessage = {
   collaboration?: CollaborationReadback;
   origin?: string;
   attachments?: ChatImageAttachment[];
+  /** Read-only original-source view; text remains the canonical model input. */
+  reply_context?: ChatReplyContext;
   message_id: string;
   turn_id: string | null;
   role: string;
@@ -918,6 +920,17 @@ export type ChatVisibleMessage = {
   };
 };
 
+export type ChatReplyContext = {
+  source_text: string;
+  parent: {message_id: string; content: string; complete: boolean};
+  thread?: {
+    root_message_id: string; thread_id: string; before_message_id: string;
+    omitted_message_count: number; complete: false;
+    messages: Array<{message_id: string; content: string; content_truncated: boolean;
+      position: number; sender?: {id: string; kind: string}; created_at?: string}>;
+  };
+};
+
 export type ChatImageAttachment = {
   data_url: string;
   id: string;

apps/presentation/dashboard/src/data/conversation-returns.test.mjs

@@ -86,7 +86,21 @@ assert.equal(recoveredInstructions.find(row => row.text === "My request").source
 assert.equal(recoveredInstructions.find(row => row.text === "Chinese first").sourceMessageId, "instruction-1");
 assert.equal(recoveredInstructions.find(row => row.text === "Do not publish").sourceMessageId, "instruction-2");
 assert.equal(reconcileConversationHistory(recoveredInstructions, withInstructions, createHistory), recoveredInstructions);
-console.log("conversation-returns: passed (session isolation, late return, deduplication, transport uncertainty, stream preservation and continued revision observation)");
+// A source-backed view updates only the matching user's canonical input.
+const replyContext = {source_text: "Compare this change", parent: {message_id: "original-parent", content: "Earlier constraint", complete: true}};
+const rawReply = {message_id: "reply-user", turn_id: "reply-turn", role: "user", text: "Frozen model input", reply_context: replyContext};
+const replyRows = [
+  {sourceSessionId: "original", sourceMessageId: "reply-user", role: "user", text: "Frozen model input"},
+  {sourceSessionId: "foreign", sourceMessageId: "reply-user", role: "user", text: "Keep foreign text"},
+  {sourceSessionId: "original", sourceMessageId: "answer", role: "assistant", text: "Live streamed answer"},
+];
+const projectedReply = reconcileConversationReturns(replyRows, "original", [rawReply, {message_id: "answer", role: "agent", text: "Stored answer"}], createReply);
+assert.equal(projectedReply[0].text, "Compare this change");
+assert.deepEqual(projectedReply[0].replyContext, replyContext);
+assert.equal(projectedReply[1], replyRows[1]);
+assert.equal(projectedReply[2].text, "Live streamed answer");
+assert.equal(reconcileConversationReturns(projectedReply, "original", [rawReply], createReply), projectedReply);
+console.log("conversation-returns: passed (session isolation, late return, deduplication, transport uncertainty, stream preservation, reply context and continued revision observation)");
 // A fast history read may add the stored queue row before its HTTP receipt
 // hydrates the original bubble. Both now carry the same canonical identity.
 const queueRace = [