Questions across replay and reconnect on the phone and paired desktops (CF-6)

apps/desktop/src/main/engine/companion.ts

@@ -433,7 +433,7 @@ const NO_KINDS: ReadonlySet<string> = new Set();
 export function phoneItem(item: ThreadItem, kinds: ReadonlySet<string>): Record<string, unknown> {
   switch (item.kind) {
     // Typed fields (docs/conversation-events.md) ride along for clients that read them; older phones ignore them.
-    case "assistant": return { id: item.id, kind: item.kind, text: item.text, at: item.at, ...(item.phase ? { phase: item.phase } : {}), ...(item.outcome ? { outcome: item.outcome } : {}) };
+    case "assistant": return { id: item.id, kind: item.kind, text: item.text, at: item.at, ...(item.phase ? { phase: item.phase } : {}), ...(item.nativePhase ? { nativePhase: item.nativePhase } : {}), ...(item.outcome ? { outcome: item.outcome } : {}) };
     case "thinking": return { id: item.id, kind: item.kind, text: item.text, at: item.at, ...(item.reasoning ? { reasoning: item.reasoning } : {}), ...(item.outcome ? { outcome: item.outcome } : {}) };
     case "user": case "notice": return { id: item.id, kind: item.kind, text: item.text, at: item.at, ...(item.kind === "notice" ? { level: item.level } : {}) };
     case "approval":
@@ -460,7 +460,7 @@ export function phoneItem(item: ThreadItem, kinds: ReadonlySet<string>): Record<
         ...(item.answeredBy ? { answeredBy: item.answeredBy } : {}), ...(item.delivery ? { delivery: item.delivery } : {}), ...(item.withdrawn ? { withdrawn: item.withdrawn } : {}),
       };
       return { id: item.id, kind: "notice", level: "info", at: item.at,
-      text: `The agent asked: ${item.questions.map((q) => q.text).join(" · ")}\n${item.delivery === "uncertain" ? "Delivery uncertain. Not resent; check the agent on your Mac." : item.delivery === "sending" ? "Sending answer on your Mac." : item.answers ? "Answered on your Mac." : item.withdrawn ? "Not answered." : "Answer it in Modex on your Mac."}` };
+      text: `The agent asked: ${item.questions.map((q) => q.text).join(" · ")}\n${item.delivery === "uncertain" ? "Delivery uncertain. Not resent; check the agent on your Mac." : item.delivery === "sending" ? "Sending answer on your Mac." : item.answers ? item.answeredBy === "phone" ? "Answered on your phone." : "Answered on your Mac." : item.withdrawn ? "Not answered." : "Answer it in Modex on your Mac."}` };
     // A kind this build does not know, such as one a newer Modex saved before a downgrade: an inert
     // notice, never `undefined`, which goes out as null and makes the phone reject the whole snapshot.
     // Its own fields are not forwarded, since they may hold anything.

apps/desktop/test/conversation-migration.test.ts

@@ -6,7 +6,7 @@ import { DatabaseSync } from "node:sqlite";
 import { EventJournal, replayItems } from "../src/main/engine/event-journal.js";
 import { Store } from "../src/main/engine/store.js";
 import { ThreadRunner } from "../src/main/engine/runner.js";
-import { mobileItem } from "../src/main/engine/companion.js";
+import { mobileItem, phoneItem } from "../src/main/engine/companion.js";
 import type { Backend } from "../src/main/engine/backends/types.js";
 import type { ThreadEvent, ThreadItem } from "../src/shared/types.js";
 import { itemFacts } from "../src/shared/conversation-events.js";
@@ -243,3 +243,49 @@ test("a journal this build wrote stays usable after a downgrade: v0.1.0's and v0
   assert.equal(back.since(0).length, 2);
   back.close();
 });
+
+test("questions and a plan v0.3.0 saved load closed and readable on every client, nothing is sent, and the draft is untouched", async () => {
+  const old = await legacyThread();
+  try {
+    const store = new Store(old.home);
+    const env = { id: "q-env", header: "Environment", text: "Which environment?", options: [{ label: "staging", description: "" }], allowOther: true, secret: false };
+    const token = { id: "q-token", header: "Token", text: "Paste the deploy token", options: null, allowOther: true, secret: true };
+    // As v0.3.0 saved them: one open when that process died, one answered before answers recorded who and how
+    // far they got, one answered on the phone; and Codex's plan as an assistant message.
+    const v030: ThreadItem[] = [
+      ...LEGACY,
+      { id: "q-open", kind: "question", questions: [env, token], blocking: true, at },
+      { id: "q-old", kind: "question", questions: [env], blocking: true, answers: { "q-env": ["staging"] }, at },
+      { id: "q-phone", kind: "question", questions: [env, token], blocking: true, answers: { "q-env": ["staging"], "q-token": ["(hidden)"] }, answeredBy: "phone", delivery: "sent", at },
+      { id: "plan", kind: "assistant", text: "### Plan\n\n- one step\n", phase: "final", nativePhase: "plan", outcome: "succeeded", at },
+    ];
+    store.saveItems(old.threadId, v030);
+    for (const item of v030.slice(LEGACY.length)) old.journal.record({ threadId: old.threadId, type: "item", item });
+    old.journal.flush();
+    fs.writeFileSync(path.join(store.dir, "update-recovery.json"), JSON.stringify({ drafts: [{ threadId: old.threadId, text: "half a reply" }] }));
+    const draft = fs.readFileSync(path.join(store.dir, "update-recovery.json"));
+    let ran = 0;
+    const backend: Backend = { id: "claude", listModels: async () => [], dispose: async () => {}, runTurn: async () => { ran++; return { status: "completed" }; } };
+    const { runner } = runnerOn(old.home, old.journal, backend);
+    const items = runner.items(old.threadId);
+    const card = (id: string) => items.find((item) => item.id === id) as ThreadItem & { answers?: unknown; answeredBy?: string; delivery?: string; withdrawn?: string };
+    assert.equal(card("q-open").withdrawn, "ended", "a question a dead process left open is closed, not answerable");
+    assert.deepEqual(card("q-old"), v030.find((item) => item.id === "q-old"), "an old answer gains no invented sender or delivery");
+    assert.deepEqual(card("q-phone"), v030.find((item) => item.id === "q-phone"));
+    assert.deepEqual(card("plan"), v030.at(-1), "the plan keeps its text and phases");
+    assert.deepEqual(items.slice(0, LEGACY.length), LEGACY, "every older word as saved");
+    old.journal.flush();
+    assert.deepEqual(replayItems(old.journal.since(0, { threadId: old.threadId }), old.threadId), items, "paired clients replay to the same");
+    assert.deepEqual(runner.answerQuestion(old.threadId, "q-open", { "q-env": ["staging"], "q-token": ["t"] }, "phone"), { ok: false, reason: "withdrawn" });
+
+    const drawn = new Set(["user", "assistant", "approval", "tool", "notice", "question"]);
+    assert.deepEqual(["q-open", "q-old", "q-phone"].map((id) => { const p = phoneItem(card(id), drawn); return [p.kind, p.withdrawn ?? null, p.answeredBy ?? null]; }),
+      [["question", "ended", null], ["question", null, null], ["question", null, "phone"]], "a phone that draws questions sees each closed card");
+    assert.deepEqual(["q-open", "q-old", "q-phone"].map((id) => String(mobileItem(card(id)).text).split("\n").at(-1)),
+      ["Not answered.", "Answered on your Mac.", "Answered on your phone."], "an older phone reads each as a notice");
+    assert.deepEqual(phoneItem(card("plan"), drawn), { id: "plan", kind: "assistant", text: "### Plan\n\n- one step\n", at, phase: "final", nativePhase: "plan", outcome: "succeeded" });
+    assert.ok(fs.readFileSync(path.join(store.dir, "update-recovery.json")).equals(draft), "the draft is byte-identical");
+    assert.equal(ran, 0, "nothing runs or is sent");
+    await runner.dispose();
+  } finally { old.journal.close(); }
+});

apps/desktop/test/journal-resume.test.ts

@@ -9,7 +9,7 @@ import { once } from "node:events";
 import { EventJournal } from "../src/main/engine/event-journal.js";
 import { DesktopHost } from "../src/main/engine/desktop-host.js";
 import { DesktopClient } from "../src/main/engine/desktop-client.js";
-import { CompanionServer, mobileItem } from "../src/main/engine/companion.js";
+import { CompanionServer, mobileItem, phoneItem } from "../src/main/engine/companion.js";
 import { Store } from "../src/main/engine/store.js";
 import { ThreadRunner } from "../src/main/engine/runner.js";
 import { MockBackend } from "../src/main/engine/backends/mock.js";
@@ -34,14 +34,14 @@ function apply(items: ThreadItem[], event: ThreadEvent): ThreadItem[] {
 }
 
 /** Fetches a snapshot exactly as the iPhone app does: pinned certificate, bearer token, cursor query. */
-async function phoneSnapshot(pairingUri: string, home: string, threadId: string, after?: { id: string; cursor: number }): Promise<PhoneSnapshot> {
+async function phoneSnapshot(pairingUri: string, home: string, threadId: string, after?: { id: string; cursor: number }, kinds?: string): Promise<PhoneSnapshot> {
   const pairing = JSON.parse(Buffer.from(new URL(pairingUri).searchParams.get("data")!, "base64url").toString()) as { url: string; token: string; fingerprint: string };
   const url = new URL("/v1/snapshot", pairing.url);
   url.searchParams.set("threadId", threadId);
   if (after) { url.searchParams.set("journal", after.id); url.searchParams.set("after", String(after.cursor)); }
   return new Promise((resolve, reject) => {
     https.get(url, {
-      headers: { authorization: `Bearer ${pairing.token}` },
+      headers: { authorization: `Bearer ${pairing.token}`, ...(kinds ? { "x-modex-item-kinds": kinds } : {}) },
       ca: fs.readFileSync(path.join(home, "companion", "cert.pem")),
       checkServerIdentity: (_host, cert) => crypto.createHash("sha256").update(cert.raw).digest("hex") === pairing.fingerprint ? undefined : new Error("unpinned"),
     }, (res) => {
@@ -623,3 +623,134 @@ test("one mixed transcript converges on every path, survives hostile frames, and
     fs.rmSync(home, { recursive: true, force: true });
   }
 });
+
+test("a typed question and Codex's plan converge on every path, through a phone answer, a desktop's sleep and a retry", async () => {
+  const home = tmpdir("modex-journal-typed-");
+  const store = new Store(home);
+  const project = store.addProject(gitRepo());
+  const live = new EventJournal(path.join(home, "app", "events.sqlite"), { flushMs: 5 });
+  const coalesced = new EventJournal(path.join(home, "batched.sqlite"), { flushMs: 60_000 });
+  const emitted: ThreadEvent[] = [];
+  const emit = (event: ThreadEvent) => { emitted.push(event); live.record(event); coalesced.record(event); };
+  const secret = "typed-tok-2b7c-SECRET";
+  const plan = "### Add a new-user greeting\n\n- Create `greeting.txt`.\n- Verify the exact content.\n";
+  let runs = 0;
+  const heard: Record<string, string[]>[] = [];
+  const sinks: TurnSink[] = [];
+  const at = (item: string) => ({ runtime: "fixture", item });
+  const backend = { id: "codex" as const, listModels: async () => [], dispose: async () => {}, async runTurn(_text: string, _opts: unknown, sink: TurnSink) {
+    const n = ++runs;
+    sinks.push(sink);
+    if (n === 1) {
+      sink.observe({ type: "message.start", native: at("m1"), phase: "commentary", nativePhase: "commentary" });
+      sink.observe({ type: "message.delta", native: at("m1"), text: "I will settle the target first." });
+      sink.observe({ type: "message.done", native: at("m1"), phase: "commentary", nativePhase: "commentary" });
+      const native = { runtime: "fixture", request: "ask-1" };
+      const reply = await sink.question!({ native, blocking: true, questions: [
+        { id: "q-env", header: "Environment", text: "Which environment?", options: [{ label: "staging", description: "" }], allowOther: true, secret: false },
+        { id: "q-token", header: "Token", text: "Paste the deploy token", options: null, allowOther: true, secret: true },
+      ] });
+      if ("answers" in reply) await sink.deliverRequest?.(native, async () => { heard.push(reply.answers); });
+      // Codex's plan item starts streaming, then the attempt fails before it completes.
+      sink.observe({ type: "message.start", native: at("plan-1"), phase: "final", nativePhase: "plan" });
+      sink.observe({ type: "message.delta", native: at("plan-1"), text: "### Draft" });
+      return { status: "failed" as const, error: "try again" };
+    }
+    // The retry's plan reuses the native id, and the first attempt's sink speaks late.
+    sinks[0]!.observe({ type: "message.delta", native: at("plan-1"), text: " (late, from attempt 1)" });
+    sink.observe({ type: "message.start", native: at("plan-1"), phase: "final", nativePhase: "plan" });
+    sink.observe({ type: "message.delta", native: at("plan-1"), text: "### Add a new-user " });
+    sink.observe({ type: "message.delta", native: at("plan-1"), text: "greeting" });
+    // The completed item is authoritative, as Codex's item/completed is for a plan.
+    sink.observe({ type: "message.done", native: at("plan-1"), text: plan, phase: "final", nativePhase: "plan" });
+    return { status: "completed" as const };
+  } };
+  const runner = new ThreadRunner({ home, store, backends: { codex: backend }, emit });
+  const host = new DesktopHost(home, { version: "test", channels: () => [], async invoke() { return null; }, journal: () => live });
+  live.subscribe((entries) => host.broadcastJournal(entries));
+  const companion = new CompanionServer(home, { state: () => store.snapshot(), items: (id) => runner.items(id), status: (id) => runner.status(id),
+    send: async () => ({ ok: true }), answer: () => {}, journal: () => live }, () => ["127.0.0.1"], () => () => {});
+  const client = new DesktopClient(() => {});
+  const connections: boolean[] = [];
+  client.on("connected", ({ resumed }: { resumed: boolean }) => connections.push(resumed));
+  let desktop: ThreadItem[] = [];
+  let restarted: ThreadRunner | undefined;
+  const KINDS = "user,assistant,approval,tool,notice,question";
+  try {
+    const thread = await runner.createThread(project.id, { backend: "codex", auto: false });
+    store.updateThread(thread.id, { title: "Fixture", titleSource: "manual" });
+    client.on("thread", (event: ThreadEvent) => { if (event.threadId === thread.id) desktop = apply(desktop, event); });
+    await host.start();
+    const pairingUri = (await companion.start()).pairingUri!;
+    await client.pair(host.invite(), "a".repeat(64));
+    await once(client, "connected", { signal: AbortSignal.timeout(5000) });
+    desktop = structuredClone(runner.items(thread.id));
+
+    // The question waits: both phones load, the desktop sleeps, and the phone answers.
+    const turn = runner.send(thread.id, "add a greeting");
+    await until(() => runner.status(thread.id) === "waiting");
+    const mid = await phoneSnapshot(pairingUri, home, thread.id, undefined, KINDS);
+    const midLegacy = await phoneSnapshot(pairingUri, home, thread.id);
+    client.suspend();
+    const card = runner.items(thread.id).find((item) => item.kind === "question")!;
+    assert.deepEqual(runner.answerQuestion(thread.id, card.id, { "q-env": ["staging"], "q-token": [secret] }, "phone"), { ok: true });
+    assert.deepEqual(runner.answerQuestion(thread.id, card.id, { "q-env": ["production"], "q-token": ["late"] }), { ok: false, reason: "already-resolved" }, "the desktop's answer, arriving late");
+    await turn;
+    await runner.retry(thread.id);
+    client.resume();
+    await once(client, "connected", { signal: AbortSignal.timeout(5000) });
+    live.flush();
+    coalesced.flush();
+
+    const saved = new Store(home).items(thread.id);
+    const fromLive = live.since(0, { threadId: thread.id });
+    const fromBatched = coalesced.since(0, { threadId: thread.id });
+    const wire = (items: ThreadItem[]) => JSON.parse(JSON.stringify(items)) as ThreadItem[];
+    assert.deepEqual(wire(runner.items(thread.id)), saved, "the live projection is what a journal-off reload loads");
+    assert.deepEqual(wire(replayItems(emitted, thread.id)), saved, "the live event stream folds to it");
+    assert.deepEqual(replayItems(fromLive, thread.id), saved, "the journal replays to it");
+    assert.deepEqual(replayItems([...fromLive, ...fromLive], thread.id), saved, "and replaying it twice changes nothing");
+    assert.ok(fromBatched.length < fromLive.length, "the batched journal really coalesced");
+    assert.deepEqual(replayItems(fromBatched, thread.id), saved, "the batched journal replays to it too");
+    assert.deepEqual(connections, [false, true], "the desktop resumed after its sleep");
+    await until(() => JSON.stringify(desktop) === JSON.stringify(saved));
+    const delta = await phoneSnapshot(pairingUri, home, thread.id, mid.journal, KINDS);
+    assert.equal(delta.delta, true);
+    const drawn = new Set(KINDS.split(","));
+    assert.deepEqual(merge(mid.items, delta.items), JSON.parse(JSON.stringify(saved.map((item) => phoneItem(item, drawn)))), "a phone that draws questions: its mid-run snapshot plus its delta matches a full one");
+    const legacyDelta = await phoneSnapshot(pairingUri, home, thread.id, midLegacy.journal);
+    assert.deepEqual(merge(midLegacy.items, legacyDelta.items), JSON.parse(JSON.stringify(saved.map(mobileItem))), "and so does an older phone's");
+
+    // What the transcript says, on every path above.
+    const answers = saved.filter((item) => item.kind === "assistant") as (ThreadItem & { text: string; phase?: string; nativePhase?: string; outcome?: string })[];
+    assert.deepEqual(answers.map((item) => [item.text, item.phase, item.nativePhase, item.outcome]), [
+      ["I will settle the target first.", "commentary", "commentary", "succeeded"],
+      ["### Draft", "final", "plan", "unknown"],
+      [plan, "final", "plan", "succeeded"],
+    ], "the plan is the final answer; the failed attempt's draft stays its own, untouched by its late text");
+    const asked = saved.find((item) => item.kind === "question") as ThreadItem & { answers?: Record<string, string[]>; answeredBy?: string; delivery?: string };
+    assert.deepEqual([asked.answers, asked.answeredBy, asked.delivery], [{ "q-env": ["staging"], "q-token": ["(hidden)"] }, "phone", "sent"]);
+    const onPhone = merge(mid.items, delta.items) as (Phone & { phase?: string; nativePhase?: string })[];
+    assert.deepEqual(onPhone.filter((item) => item.kind === "assistant").map((item) => [item.phase, item.nativePhase]),
+      [["commentary", "commentary"], ["final", "plan"], ["final", "plan"]], "the phone can tell Codex's plan from an ordinary answer");
+
+    for (const [where, text] of [["the saved files", JSON.stringify(saved)], ["the journal", JSON.stringify(fromLive)], ["the batched journal", JSON.stringify(fromBatched)],
+      ["the paired desktop", JSON.stringify(desktop)], ["the phone", JSON.stringify([mid, delta, midLegacy, legacyDelta])]] as const) {
+      assert.ok(!text.includes(secret), `the secret is not in ${where}`);
+    }
+
+    restarted = new ThreadRunner({ home, store: new Store(home), backends: { codex: backend }, emit: () => {} });
+    assert.deepEqual(wire(restarted.items(thread.id)), saved, "a restarted Modex loads the same transcript");
+    assert.equal(runs, 2, "two attempts ran: the turn and its retry");
+    assert.deepEqual(heard, [{ "q-env": ["staging"], "q-token": [secret] }], "the CLI heard the phone's answer once");
+  } finally {
+    client.dispose();
+    await companion.dispose();
+    await host.dispose();
+    await restarted?.dispose();
+    await runner.dispose();
+    live.close();
+    coalesced.close();
+    fs.rmSync(home, { recursive: true, force: true });
+  }
+});