Skip to content

Commit fb76aac

Browse files
authored
Carry typed conversation state through the journal, snapshots and the phone (CF-6, first half) (#285)
* Share the captured-stream test helpers between test files The fixture loader and the Codex and Claude stream players move from conversation-events.test.ts to test/conversation-fixtures.ts, so the journal tests can replay the same captures without re-running those tests. No behaviour changes. Refs #279 Coven-Harness: Claude Code / claude-opus-5-5 * Replay journal items as upserts and give the phone typed items replayItems appended an item event even when that item was already there, so a frame delivered twice doubled the item and left a stale copy; it now upserts by id as the renderer and phone do. A typed Codex capture converges across the runner, the saved snapshot, journal replay, a duplicated replay and the phone's full and delta snapshots. The phone gets phase, outcome and reasoning; a phone that reads only title and status sees how a finished tool that did not succeed ended in its title ("· Denied", "· Failed", "· Stopped", "· Status unavailable"). Refs #279 Coven-Harness: Claude Code / claude-opus-5-5 * Settle work a dead process left running in the snapshot and the journal The runner already read a running item from an earlier process as unknown (and a running agent as status unavailable), but only in memory: the saved transcript kept "running", and a paired desktop or phone resuming from the journal kept a spinner forever. Loading a thread now saves the settled items once and sends them as item updates, so the journal and every client agree. Nothing is run again. Old transcripts and journals load, replay and project unchanged. Refs #279 Coven-Harness: Claude Code / claude-opus-5-5 * Document typed conversation state in the event journal event-journal.md gains how typed items travel, merge and replay, how work left running at a restart is settled for every client, and what the phone receives; conversation-events.md records the CF-6 half that rests on #274. Status moves the merged CF-1 (#282) and CF-4 (#284) entries and this change into one "On main, unreleased" entry. Refs #279 Coven-Harness: Claude Code / claude-opus-5-5
1 parent 8e26534 commit fb76aac

10 files changed

Lines changed: 384 additions & 95 deletions

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

Lines changed: 11 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import { publishCompanion, type CompanionPublisher } from "./companion-discovery
1111
import type { EventJournal } from "./event-journal.js";
1212
import { discoverMobileCommands } from "./mobile-commands.js";
1313
import type { RoutineRun } from "../../shared/routines.js";
14+
import { toolStatusLabel } from "../../shared/tool-summary.js";
1415

1516
interface CompanionSource {
1617
state(): AppState;
@@ -375,11 +376,18 @@ function mobileRoutineRun(run: RoutineRun): Record<string, unknown> {
375376
return { id: run.id, routine: run.routineName, status: run.status, endedAt: run.endedAt ?? run.startedAt, ...(run.threadId ? { threadId: run.threadId } : {}) };
376377
}
377378

378-
function mobileItem(item: ThreadItem): Record<string, unknown> {
379+
/** A transcript item as the phone receives it. Exported for tests. */
380+
export function mobileItem(item: ThreadItem): Record<string, unknown> {
379381
switch (item.kind) {
380-
case "user": case "assistant": case "thinking": case "notice": return { id: item.id, kind: item.kind, text: item.text, at: item.at, ...(item.kind === "notice" ? { level: item.level } : {}) };
382+
// Typed fields (docs/conversation-events.md) ride along for clients that read them; older phones ignore them.
383+
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 } : {}) };
384+
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 } : {}) };
385+
case "user": case "notice": return { id: item.id, kind: item.kind, text: item.text, at: item.at, ...(item.kind === "notice" ? { level: item.level } : {}) };
381386
case "approval": return { id: item.id, kind: item.kind, question: item.question, detail: item.detail, answer: item.answer, ...(item.decidedBy?.source === "policy" ? { auto: item.decidedBy.ruleId.slice("policy:".length) } : item.decidedBy && item.decidedBy.decision !== "ask" ? { auto: "rule" } : {}), at: item.at };
382-
case "tool": return { id: item.id, kind: item.kind, title: item.title, status: item.status, ok: item.ok, at: item.at };
387+
// A phone that reads only title and status shows every finished tool as done, so a finished tool
388+
// that did not succeed says how it ended in its title.
389+
case "tool": return { id: item.id, kind: item.kind, title: item.status === "done" && (item.outcome ? item.outcome !== "succeeded" : item.ok === false) ? `${item.title} · ${toolStatusLabel(item)}` : item.title,
390+
status: item.status, ok: item.ok, ...(item.outcome ? { outcome: item.outcome } : {}), at: item.at };
383391
case "route": return { id: item.id, kind: item.kind, text: `${item.backend} · ${item.model || "default"}`, at: item.at };
384392
}
385393
}

‎apps/desktop/src/main/engine/event-journal.ts‎

Lines changed: 12 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -212,19 +212,24 @@ function merge(previous: ThreadEvent, next: ThreadEvent): ThreadEvent {
212212
}
213213

214214
/**
215-
* A thread's transcript rebuilt from journaled events, applied as the runner applies them: an
216-
* `item` appends, an `item_update` merges into the item with that id and is ignored when there is
217-
* none.
215+
* A thread's transcript rebuilt from journaled events, applied as clients apply them: an `item` is
216+
* an upsert by id (appended when new, replacing the item otherwise), and an `item_update` merges
217+
* into the item with that id and is ignored when there is none. So a frame delivered twice changes
218+
* nothing, and items keep their identity: two items never merge unless they share an id.
218219
*/
219220
export function replayItems(entries: Iterable<JournalEntry | ThreadEvent>, threadId: string): ThreadItem[] {
220221
const items: ThreadItem[] = [];
222+
const index = new Map<string, number>();
221223
for (const entry of entries) {
222224
const event = "event" in entry ? entry.event : entry;
223225
if (event.threadId !== threadId) continue;
224-
if (event.type === "item") items.push(event.item);
225-
else if (event.type === "item_update") {
226-
const index = items.findIndex((item) => item.id === event.id);
227-
if (index >= 0) items[index] = { ...items[index], ...event.patch } as ThreadItem;
226+
if (event.type === "item") {
227+
const at = index.get(event.item.id);
228+
if (at === undefined) { index.set(event.item.id, items.length); items.push(event.item); }
229+
else items[at] = event.item;
230+
} else if (event.type === "item_update") {
231+
const at = index.get(event.id);
232+
if (at !== undefined) items[at] = { ...items[at], ...event.patch } as ThreadItem;
228233
}
229234
}
230235
return items;

‎apps/desktop/src/main/engine/runner.ts‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -200,6 +200,20 @@ export class ThreadRunner {
200200
});
201201
l = { abort: null, run: null, backend: null, pending: new Map(), items, status: "idle", conversation: initialConversation(), attempt: null };
202202
this.live.set(threadId, l);
203+
// Settle it everywhere once, not only in this window: save it for the next process, and send the
204+
// change through the event stream so the journal and paired clients that resume from it see it too.
205+
const stored = this.o.store.items(threadId);
206+
const settled = items.flatMap((item, i) => {
207+
const before = stored[i];
208+
if (!before || item === before) return [];
209+
const patch = Object.fromEntries(Object.entries(item).filter(([key, value]) => (before as Record<string, unknown>)[key] !== value));
210+
return [{ id: item.id, patch: patch as Partial<ThreadItem> }];
211+
});
212+
if (settled.length) {
213+
try { this.persistItems(threadId, l); }
214+
catch (error) { console.error("[modex] could not save settled items:", error); }
215+
for (const { id, patch } of settled) this.o.emit({ threadId, type: "item_update", id, patch });
216+
}
203217
}
204218
return l;
205219
}

‎apps/desktop/test/conversation-events.test.ts‎

Lines changed: 1 addition & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -11,73 +11,14 @@ import type { Backend, TurnResult, TurnSink } from "../src/main/engine/backends/
1111
import type { ThreadEvent, ThreadItem } from "../src/shared/types.js";
1212
import { FakeProcess, fakeSpawn } from "./fakeproc.js";
1313
import { gitRepo, tmpdir } from "./helpers.js";
14+
import { CLAUDE_SESSION, CODEX_THREAD, FIXTURES, claudeStream, codexServer, fixture } from "./conversation-fixtures.js";
1415
import { CONVERSATION_SCHEMA, applyConversationEvent, finalAnswerIndex, itemFacts, parseConversationEvent, projectConversation, type AttemptMark, type ConversationEvent, type ConversationItem, type NativeRef, type Observation, type TurnRef } from "../src/shared/conversation-events.js";
1516

16-
// Sanitized captures from real CLI runs (see test/fixtures/conversation/README.md). Replaying them
17-
// tests the adapters against bytes the CLIs really sent; it is not live-provider acceptance.
18-
const FIXTURES = fileURLToPath(new URL("../../test/fixtures/conversation/", import.meta.url));
19-
const fixture = (name: string) => fs.readFileSync(path.join(FIXTURES, name), "utf8").split("\n").filter((l) => l.trim());
20-
const CODEX_THREAD = "01a14da2-063a-77a3-9092-0e3543b84bd0";
21-
const CLAUDE_SESSION = "852a2cd1-6453-4922-802d-193d60be6980";
22-
2317
/** What a test reads off a transcript item, typed loosely so it also reads items written before CF-1. */
2418
type Loose = ThreadItem & Record<string, unknown>;
2519
const conversation = (items: ThreadItem[]) => items.filter((i) => i.kind === "assistant" || i.kind === "thinking" || i.kind === "tool") as Loose[];
2620
const origin = (item: Loose) => item.origin as { thread: string; turn: string; attempt: string; backend: string; native: { runtime: string; thread?: string; turn?: string; item: string; part?: string } } | undefined;
2721

28-
/** Writes `text` to the fake CLI's stdout in chunks of 1, 2, 3, 5 and 7 bytes, splitting UTF-8 sequences and JSON lines. */
29-
async function trickle(proc: FakeProcess, text: string): Promise<void> {
30-
const bytes = Buffer.from(text, "utf8");
31-
const sizes = [1, 2, 3, 5, 7];
32-
for (let at = 0, n = 0; at < bytes.length; n++) {
33-
const size = sizes[n % sizes.length]!;
34-
proc.stdout.write(bytes.subarray(at, at + size));
35-
at += size;
36-
if (n % 16 === 0) await new Promise((r) => setImmediate(r));
37-
}
38-
}
39-
40-
/**
41-
* Plays a captured app-server stream to CodexBackend. Setup requests are answered with the capture's
42-
* own thread and turn ids; each server request (an approval) waits for Modex's reply before the next line.
43-
*/
44-
function codexServer(proc: FakeProcess, lines: string[], opts: { split?: boolean; close?: number } = {}) {
45-
const turnId = (lines.map((l) => JSON.parse(l) as { method?: string; params?: { turn?: { id?: string } } }).find((m) => m.method === "turn/started")?.params?.turn?.id) ?? "turn-1";
46-
const threadId = (JSON.parse(lines[0]!) as { params?: { threadId?: string; thread?: { id?: string } } }).params;
47-
const tid = threadId?.threadId ?? threadId?.thread?.id ?? CODEX_THREAD;
48-
let played: Promise<void> | null = null;
49-
proc.stdin.on("data", (d: Buffer) => {
50-
for (const line of d.toString().split("\n")) {
51-
if (!line.trim()) continue;
52-
const msg = JSON.parse(line) as { id?: number; method?: string };
53-
if (msg.method === "initialize") proc.emitLine({ id: msg.id, result: { userAgent: "fixture/0.163.0" } });
54-
else if (msg.method === "thread/start" || msg.method === "thread/resume") proc.emitLine({ id: msg.id, result: { thread: { id: tid } } });
55-
else if (msg.method === "turn/start") {
56-
proc.emitLine({ id: msg.id, result: { turn: { id: turnId } } });
57-
played ??= (async () => {
58-
for (const raw of lines) {
59-
const m = JSON.parse(raw) as { id?: number; method?: string };
60-
if (opts.split) await trickle(proc, raw + "\n"); else proc.stdout.write(raw + "\n");
61-
if (m.id !== undefined && m.method) await proc.waitFor((l) => { const r = JSON.parse(l) as { id?: number; result?: unknown }; return r.id === m.id && r.result !== undefined; }, 5000);
62-
}
63-
if (opts.close !== undefined) proc.close(opts.close);
64-
})();
65-
}
66-
}
67-
});
68-
}
69-
70-
/** Plays a captured `claude -p` stream. Each `control_request` waits for Modex's `control_response`. */
71-
async function claudeStream(proc: FakeProcess, lines: string[], opts: { split?: boolean } = {}): Promise<void> {
72-
await proc.waitFor((l) => l.includes('"type":"user"'));
73-
for (const raw of lines) {
74-
const m = JSON.parse(raw) as { type: string; request_id?: string };
75-
if (opts.split) await trickle(proc, raw + "\n"); else proc.stdout.write(raw + "\n");
76-
if (m.type === "control_request") await proc.waitFor((l) => l.includes('"control_response"') && l.includes(m.request_id!), 5000);
77-
}
78-
proc.close(0);
79-
}
80-
8122
/** A runner on a fresh store with the given backends, and a thread on `backend` whose approvals the test answers. */
8223
async function harness(backends: Partial<Record<"claude" | "codex" | "mock", Backend>>, backend: "claude" | "codex" | "mock", answer: (question: string) => "yes" | "no" = () => "yes") {
8324
const home = tmpdir("modex-conversation-");
Lines changed: 65 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,65 @@
1+
import fs from "node:fs";
2+
import path from "node:path";
3+
import { fileURLToPath } from "node:url";
4+
import type { FakeProcess } from "./fakeproc.js";
5+
6+
// Sanitized captures from real CLI runs (see test/fixtures/conversation/README.md). Replaying them
7+
// tests the adapters against bytes the CLIs really sent; it is not live-provider acceptance.
8+
export const FIXTURES = fileURLToPath(new URL("../../test/fixtures/conversation/", import.meta.url));
9+
export const fixture = (name: string) => fs.readFileSync(path.join(FIXTURES, name), "utf8").split("\n").filter((l) => l.trim());
10+
export const CODEX_THREAD = "01a14da2-063a-77a3-9092-0e3543b84bd0";
11+
export const CLAUDE_SESSION = "852a2cd1-6453-4922-802d-193d60be6980";
12+
13+
/** Writes `text` to the fake CLI's stdout in chunks of 1, 2, 3, 5 and 7 bytes, splitting UTF-8 sequences and JSON lines. */
14+
export async function trickle(proc: FakeProcess, text: string): Promise<void> {
15+
const bytes = Buffer.from(text, "utf8");
16+
const sizes = [1, 2, 3, 5, 7];
17+
for (let at = 0, n = 0; at < bytes.length; n++) {
18+
const size = sizes[n % sizes.length]!;
19+
proc.stdout.write(bytes.subarray(at, at + size));
20+
at += size;
21+
if (n % 16 === 0) await new Promise((r) => setImmediate(r));
22+
}
23+
}
24+
25+
/**
26+
* Plays a captured app-server stream to CodexBackend. Setup requests are answered with the capture's
27+
* own thread and turn ids; each server request (an approval) waits for Modex's reply before the next line.
28+
*/
29+
export function codexServer(proc: FakeProcess, lines: string[], opts: { split?: boolean; close?: number } = {}) {
30+
const turnId = (lines.map((l) => JSON.parse(l) as { method?: string; params?: { turn?: { id?: string } } }).find((m) => m.method === "turn/started")?.params?.turn?.id) ?? "turn-1";
31+
const threadId = (JSON.parse(lines[0]!) as { params?: { threadId?: string; thread?: { id?: string } } }).params;
32+
const tid = threadId?.threadId ?? threadId?.thread?.id ?? CODEX_THREAD;
33+
let played: Promise<void> | null = null;
34+
proc.stdin.on("data", (d: Buffer) => {
35+
for (const line of d.toString().split("\n")) {
36+
if (!line.trim()) continue;
37+
const msg = JSON.parse(line) as { id?: number; method?: string };
38+
if (msg.method === "initialize") proc.emitLine({ id: msg.id, result: { userAgent: "fixture/0.163.0" } });
39+
else if (msg.method === "thread/start" || msg.method === "thread/resume") proc.emitLine({ id: msg.id, result: { thread: { id: tid } } });
40+
else if (msg.method === "turn/start") {
41+
proc.emitLine({ id: msg.id, result: { turn: { id: turnId } } });
42+
played ??= (async () => {
43+
for (const raw of lines) {
44+
const m = JSON.parse(raw) as { id?: number; method?: string };
45+
if (opts.split) await trickle(proc, raw + "\n"); else proc.stdout.write(raw + "\n");
46+
if (m.id !== undefined && m.method) await proc.waitFor((l) => { const r = JSON.parse(l) as { id?: number; result?: unknown }; return r.id === m.id && r.result !== undefined; }, 5000);
47+
}
48+
if (opts.close !== undefined) proc.close(opts.close);
49+
})();
50+
}
51+
}
52+
});
53+
}
54+
55+
/** Plays a captured `claude -p` stream. Each `control_request` waits for Modex's `control_response`. */
56+
export async function claudeStream(proc: FakeProcess, lines: string[], opts: { split?: boolean } = {}): Promise<void> {
57+
await proc.waitFor((l) => l.includes('"type":"user"'));
58+
for (const raw of lines) {
59+
const m = JSON.parse(raw) as { type: string; request_id?: string };
60+
if (opts.split) await trickle(proc, raw + "\n"); else proc.stdout.write(raw + "\n");
61+
if (m.type === "control_request") await proc.waitFor((l) => l.includes('"control_response"') && l.includes(m.request_id!), 5000);
62+
}
63+
proc.close(0);
64+
}
65+

0 commit comments

Comments
 (0)