Fix 2.3.0 recovery and conversation intake

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

@@ -216,6 +216,11 @@ def unavailable(handler):
                                 "Verified results retain their source. Unconfirmed returns remain open for recovery."
                                 for part in range(6)))
                     result = {"populated": True, "messages": len(server.chat_store.messages(sid))}
+                elif command == "history":
+                    for index in range(450):
+                        server.chat_store.append_message(sid, role="agent",
+                            text="Original full body. " * 1000 if index == 201 else f"Paged bound row {index}")
+                    result = {"history_rows": len(server.chat_store.messages(sid))}
                 elif command == "inspect":
                     saved = server.chat_store.load_session(sid)
                     result = {"sessions": len(server.chat_store.list_sessions()), "original_thread": saved["upstream_thread_id"],

apps/presentation/dashboard/smoke/bound-conversation-smoke.ts

@@ -80,12 +80,73 @@ try {
   assert.equal(calls[0].path, `/api/chat/sessions?conversation_binding_id=${bindingId}`);
   const routedRead = globalThis.fetch;
   if (!fallback) {
+    // Repeated transport loss at later pages must advance one original prefix
+    // through the existing retry caller, without ever publishing partial text.
+    child.stdin.write("history\n");
+    const expectedRows = (await next()).history_rows;
+    const complete = await fetchChatSessionHistory(original.session);
+    assert.equal(complete.messages.length, expectedRows);
+    const recoveredOffsets: number[] = [];
+    const failedOffsets = new Set<number>();
+    globalThis.fetch = async (input, init) => {
+      const response = await routedRead(input, init);
+      if (new URL(String(input), origin).pathname !== `/api/chat/sessions/${sessionId}`) return response;
+      const value = await response.clone().json();
+      const offset = value.message_page.offset;
+      recoveredOffsets.push(offset);
+      if (offset > 0 && !failedOffsets.has(offset)) {
+        failedOffsets.add(offset);
+        throw new Error("Synthetic interrupted transport after a verified prefix");
+      }
+      return response;
+    };
+    const interrupted = await fetchChatHistory(options);
+    assert.deepEqual(interrupted.unavailableSessionIds, [sessionId]);
+    assert.deepEqual(interrupted.messages, []);
+    assert.deepEqual(interrupted.snapshots, []);
+    const interruptedAgain = await fetchChatHistory(options, interrupted);
+    assert.deepEqual(interruptedAgain.messages, []);
+    const recovered = await fetchChatHistory(options, interruptedAgain);
+    assert.deepEqual(recovered.unavailableSessionIds, []);
+    assert.deepEqual(recovered.snapshots[0].messages, complete.messages);
+    assert.deepEqual(recoveredOffsets, [0, 200, 200, 400, 400], "Only failed pages are read again");
+    assert.equal(interrupted.pendingReads?.get(sessionId)?.messages.length, 200,
+      "Later retries cannot mutate the previously published read state");
+    assert.equal(recovered.pendingReads?.size, 0);
+    assert.ok(calls.every(row => row.method === "GET"), "Recovery never starts work");
+    globalThis.fetch = routedRead;
+    // Current source revocation blocks recovery before another history page.
+    child.stdin.write("deny\n"); assert.equal((await next()).source_verified, false);
+    calls.length = 0;
+    await assert.rejects(() => fetchChatHistory(options, interrupted));
+    assert.ok(calls.every(row => new URL(row.path, origin).pathname === "/api/chat/sessions"));
+    child.stdin.write("recover\n"); assert.equal((await next()).source_verified, true);
+    // Permanent denial, a changed source or malformed prefix cannot be retained
+    // as transport progress. The following read must validate page zero again.
+    for (const fault of ["denied", "source", "digest"]) {
+      globalThis.fetch = async (input, init) => {
+        const response = await routedRead(input, init);
+        if (!new URL(String(input), origin).searchParams.has("message_cursor")) return response;
+        if (fault === "denied") return new Response("{}", {status: 403});
+        const value = await response.json();
+        if (fault === "source") value.session.goal_id = "unrelated-goal";
+        else value.message_page.snapshot_sha256 = "0".repeat(64);
+        return new Response(JSON.stringify(value));
+      };
+      const rejected = await fetchChatHistory(options, interrupted);
+      assert.deepEqual(rejected.messages, []);
+      assert.equal(rejected.pendingReads?.size, 0);
+      globalThis.fetch = routedRead;
+      calls.length = 0;
+      assert.deepEqual((await fetchChatHistory(options, rejected)).snapshots[0].messages, complete.messages);
+      assert.equal(new URL(calls.find(row => row.path.includes(`${sessionId}?`))!.path, origin).searchParams.has("message_cursor"), false);
+    }
     // Real HTTP responses beyond the old interactive deadline must still hydrate
     // the complete original history. Source verification keeps its own deadline.
     child.stdin.write("slow-history\n");
     assert.equal((await next()).slow_history, true);
     const slow = await fetchChatHistory(options);
-    assert.deepEqual(slow.messages, history.messages);
+    assert.deepEqual(slow.messages, recovered.messages);
     assert.deepEqual(slow.unavailableSessionIds, []);
     // Delay only the post-history fresh source check, not initial discovery.
     const beforeSlowSource = calls.length;
@@ -98,7 +159,7 @@ try {
       return response;
     };
     const sourceSlow = await fetchChatHistory(options);
-    assert.deepEqual(sourceSlow.messages, history.messages);
+    assert.deepEqual(sourceSlow.messages, recovered.messages);
     assert.deepEqual(sourceSlow.unavailableSessionIds, []);
     assert.ok(calls.slice(beforeSlowSource).every(row => row.method === "GET"));
     globalThis.fetch = routedRead;

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

@@ -657,6 +657,7 @@ export async function readTodoRequest(goalId: string, todoId: string, signal: Ab
   const result = z.object({
     ok: z.literal(true), goal_id: z.string(), todo_id: z.string(), text: z.string(),
     status: z.string(), archive_state: z.string(), updated_at: z.string().nullable(),
+    evidence: z.string().nullable().optional(), note: z.string().nullable().optional(),
     review_basis: todoUpdateSourceBasisSchema.optional(),
     source_access: z.object({namespace: z.string()}).optional(),
   }).parse(await requestJson<unknown>(url, {signal}));
@@ -1055,15 +1056,40 @@ const chatMessagePageSchema = z.object({
 // its pages and fresh source checks; caller cancellation still takes priority.
 const CHAT_HISTORY_READ_TIMEOUT_MS = 30_000;
 
+type ChatHistoryReadProgress = {
+  first: ChatSessionSnapshot;
+  messages: ChatVisibleMessage[];
+  cursors: Set<string>;
+  cursor: string;
+  total: number;
+  digest: string;
+};
+
+function sameHistorySource(actual: ChatSessionSummary, expected: ChatSessionSummary) {
+  return actual.session_id === expected.session_id &&
+    (["agent_id", "goal_id", "channel_id", "managed_agent_id", "goal_instance_id", "project_ref"] as const)
+      .every(key => (actual[key] ?? null) === (expected[key] ?? null)) &&
+    (actual.session_mode ?? "managed_runtime") === (expected.session_mode ?? "managed_runtime");
+}
+
 /** Hydrate one complete original prefix; partial pages never become a snapshot. */
 export async function fetchChatSessionHistory(session: ChatSessionSummary | string, signal?: AbortSignal, turnId?: string): Promise<ChatSessionSnapshot> {
+  return readChatSessionHistory(session, signal, turnId);
+}
+
+async function readChatSessionHistory(session: ChatSessionSummary | string, signal?: AbortSignal, turnId?: string,
+  progress?: ChatHistoryReadProgress, checkpoint?: (progress: ChatHistoryReadProgress) => void): 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.
   let expectedSession = typeof session === "string" ? undefined : session;
-  let first: ChatSessionSnapshot | undefined, cursor: string | null = null;
-  let total: number | undefined, digest: string | undefined;
-  const messages: ChatVisibleMessage[] = [], cursors = new Set<string>(), ids = new Set<string>();
+  // Recovery is derived browser memory, scoped again by current discovery. It
+  // grants no source authority and never exposes an incomplete transcript.
+  if (progress && expectedSession && !sameHistorySource(progress.first.session, expectedSession)) progress = undefined;
+  let first = progress?.first, cursor: string | null = progress?.cursor ?? null;
+  let total = progress?.total, digest = progress?.digest;
+  const messages = [...(progress?.messages ?? [])], cursors = new Set(progress?.cursors),
+    ids = new Set(messages.map(message => message.message_id));
   do {
     signal?.throwIfAborted();
     const query = new URLSearchParams({message_limit: "200"});
@@ -1074,10 +1100,7 @@ export async function fetchChatSessionHistory(session: ChatSessionSummary | stri
       {signal: signal ? AbortSignal.any([signal, AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS)]) : AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS)});
     signal?.throwIfAborted();
     const expected = expectedSession;
-    const sameSource = snapshot.session?.session_id === sessionId && (!expected || (
-      (["agent_id", "goal_id", "channel_id", "managed_agent_id", "goal_instance_id", "project_ref"] as const)
-        .every(key => (snapshot.session[key] ?? null) === (expected[key] ?? null)) &&
-      (snapshot.session.session_mode ?? "managed_runtime") === (expected.session_mode ?? "managed_runtime")));
+    const sameSource = snapshot.session?.session_id === sessionId && (!expected || sameHistorySource(snapshot.session, expected));
     if (snapshot.ok !== true || snapshot.schema_version !== "loopx_chat_store_v1" || !sameSource || !Array.isArray(snapshot.messages)) {
       throw new Error("The original Session scope changed during history read.");
     }
@@ -1106,7 +1129,10 @@ export async function fetchChatSessionHistory(session: ChatSessionSummary | stri
     }
     first ??= snapshot;
     total = page.total; digest = page.snapshot_sha256; cursor = page.next_cursor;
-    if (cursor !== null) cursors.add(cursor);
+    if (cursor !== null) {
+      cursors.add(cursor);
+      checkpoint?.({first, messages: [...messages], cursors: new Set(cursors), cursor, total, digest});
+    }
   } while (cursor !== null);
   // Keep the first read's execution/revision hint. An append between pages is
   // intentionally outside this prefix and will trigger the normal next refresh.
@@ -1206,6 +1232,7 @@ export type ChatHistory = {
   sessions: ChatSessionSummary[];
   snapshots: ChatSessionSnapshot[];
   unavailableSessionIds: string[];
+  pendingReads?: Map<string, ChatHistoryReadProgress>;
 };
 
 export async function fetchChatHistory(options: {
@@ -1223,6 +1250,8 @@ export async function fetchChatHistory(options: {
   const listed = options.managed ? {sessions: [await resolveManagedChatSession(options.managed, AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS))]}
     : (!options.bound ? previous : undefined) ?? await fetchChatSessions({ ...options, signal: AbortSignal.timeout(CHAT_HISTORY_READ_TIMEOUT_MS) });
   const known = new Map(previous?.snapshots.map((snapshot) => [snapshot.session.session_id, snapshot]));
+  const pendingReads = new Map(previous?.pendingReads);
+  for (const id of pendingReads.keys()) if (!listed.sessions.some(session => session.session_id === id)) pendingReads.delete(id);
   // A failed historical read is not an empty transcript. Retrying only the
   // missing snapshots keeps this recovery read-only and bounds repeated work.
   const missing = listed.sessions.filter((session) => {
@@ -1234,7 +1263,20 @@ export async function fetchChatHistory(options: {
   });
   if (exact) for (const session of missing) known.delete(session.session_id);
   const results = await Promise.allSettled(missing.map(async (session) => {
-    const snapshot = await fetchChatSessionHistory(session);
+    const savedProgress = pendingReads.get(session.session_id);
+    if (savedProgress && !sameHistorySource(savedProgress.first.session, session)) pendingReads.delete(session.session_id);
+    let snapshot: ChatSessionSnapshot;
+    try {
+      snapshot = await readChatSessionHistory(session, undefined, undefined, pendingReads.get(session.session_id),
+        progress => pendingReads.set(session.session_id, progress));
+    } catch (error) {
+      // Reuse the hook's existing retry schedule only for a lost transport.
+      // Revoked sources, invalid pages and stale cursors must start a fresh read.
+      if (!(error instanceof ChatApiError && error.payload.error_code === "chat_api_unavailable"
+        && (error.payload.http_status === undefined || Number(error.payload.http_status) >= 500))) pendingReads.delete(session.session_id);
+      throw error;
+    }
+    pendingReads.delete(session.session_id);
     if (snapshot.session.session_id !== session.session_id || snapshot.session.agent_id !== session.agent_id) {
       throw new Error("The original Session identity changed during history read.");
     }
@@ -1258,6 +1300,7 @@ export async function fetchChatHistory(options: {
     sessions: listed.sessions,
     snapshots,
     unavailableSessionIds: listed.sessions.filter((session) => !known.has(session.session_id)).map((session) => session.session_id),
+    pendingReads,
   };
 }
 

apps/presentation/dashboard/src/features/personal-workspace/context-drawer.tsx

@@ -45,7 +45,7 @@ import type { LarkGoalConnection } from "../../data/chat";
 import { localizedGoalState, localizedSessionStatus, useWorkspaceI18n } from "./i18n";
 import { formatCostUsd, formatDurationMs, formatTokenCount, formatUsageValue, hasGoalUsage } from "./personal-workspace-model";
 import { TeamPlanResult } from "./team-plan-result";
-import { TaskRequest } from "./task-request";
+import { TaskRequest, taskRequestIdentity, type TaskRequestRead } from "./task-request";
 import { parseTodoResumeCondition } from "./todo-resume-condition";
 import { MarkdownText } from "./markdown";
 import { formatMonitorDate } from "./monitor-readback";
@@ -145,6 +145,7 @@ export function ContextDrawer({ agents, attentionHistory = [], onSelectAttention
   const { locale, t } = useWorkspaceI18n();
   const [correction, setCorrection] = useState("");
   const [diagnosticsOpen, setDiagnosticsOpen] = useState(false);
+  const [taskRequestRead, setTaskRequestRead] = useState<TaskRequestRead | null>(null);
   const [repositoryCopyState, setRepositoryCopyState] = useState<"idle" | "copied" | "error">("idle");
   const [runDrawerTab, setRunDrawerTab] = useState<"record" | "details">("record");
   const [runActions, setRunActions] = useState<Record<string, RunActionStates>>({});
@@ -355,6 +356,15 @@ export function ContextDrawer({ agents, attentionHistory = [], onSelectAttention
       : selection.item.goalId;
   const taskNamespace = goals.find(goal => goal.goalId === goalId)?.sourceAccess?.namespace;
   const taskReadable = !readOnly || Boolean(taskNamespace);
+  const taskDetail = selection.kind === "todo" && taskRequestRead?.phase === "ready"
+    && taskRequestRead.identity === taskRequestIdentity(selection.item, taskReadable, taskNamespace)
+    ? taskRequestRead.detail : null;
+  // Missing fields retain compatibility with older/local request-only readers;
+  // an explicit null from the current source clears an outdated display hint.
+  const taskEvidence = taskDetail && "evidence" in taskDetail ? taskDetail.evidence
+    : selection.kind === "todo" ? selection.item.evidence : undefined;
+  const taskNote = taskDetail && "note" in taskDetail ? taskDetail.note
+    : selection.kind === "todo" ? selection.item.note : undefined;
   const contextLabel = selection.kind === "attention" ? selection.item.goalTitle ?? t("drawer.currentGoal")
     : selection.kind === "todo" ? selection.item.goalTitle
       : selection.kind === "run" ? selection.item.goalTitle
@@ -684,7 +694,7 @@ export function ContextDrawer({ agents, attentionHistory = [], onSelectAttention
         ) : selection.item.detailMode ? (
           <section className="personal-task-inspector-summary">
             <p className="personal-proposal-explainer">{t(selection.item.detailMode === "missing_current" ? "drawer.taskReadbackUnavailable" : "drawer.taskRequestOnly")}</p>
-            <TaskRequest todo={selection.item} local={taskReadable} sourceNamespace={taskNamespace} />
+            <TaskRequest todo={selection.item} local={taskReadable} sourceNamespace={taskNamespace} onRead={setTaskRequestRead} />
           </section>
         ) : (
           <>
@@ -697,6 +707,7 @@ export function ContextDrawer({ agents, attentionHistory = [], onSelectAttention
                 <span>{selection.item.taskClass === "advancement_task" ? t("drawer.taskAdvancement") : selection.item.taskClass ?? t("drawer.taskOrdinary")}</span>
               </div>
               <TaskRequest key={`${taskNamespace}:${selection.item.goalId}:${selection.item.todoId}`} todo={selection.item} local={taskReadable} sourceNamespace={taskNamespace}
+                onRead={setTaskRequestRead}
                 onSave={!readOnly && !todoCompleted ? callbacks.onSaveTodoRequirements : undefined} />
               {selection.item.status === "blocked" ? <section aria-label={t("drawer.blockedReason")} className="personal-task-blocked-reason" data-recorded={selection.item.blockedReason ? "true" : "false"}>
                 <h4><AlertCircle size={14} />{t("drawer.blockedReason")}</h4>
@@ -788,13 +799,13 @@ export function ContextDrawer({ agents, attentionHistory = [], onSelectAttention
                   <small>{t("drawer.taskCompletionProofHint")}</small></label>
               </details>}
             </div> : null}
-            {selection.item.evidence ? <section className="personal-detail-card" aria-label={t("drawer.evidence")}>
+            {taskEvidence ? <section className="personal-detail-card" aria-label={t("drawer.evidence")}>
               <h4>{t("drawer.evidence")}</h4>
-              <MarkdownText text={selection.item.evidence} />
+              <MarkdownText text={taskEvidence} />
             </section> : null}
-            {selection.item.note ? <details className="personal-detail-card" aria-label={t("drawer.notes")}>
+            {taskNote ? <details className="personal-detail-card" aria-label={t("drawer.notes")}>
               <summary>{t("drawer.notes")}</summary>
-              <MarkdownText text={selection.item.note} />
+              <MarkdownText text={taskNote} />
             </details> : null}
             {todoCompleted ? <div className="personal-task-completed-note"><Check size={16} /><span><strong>{t("drawer.taskCompletedTitle")}</strong><small>{t("drawer.taskCompletedNote")}</small></span></div> : null}
           </>