Harden the event journal against failed writes, crashes and oversized frames (CF-6, journal mechanics)

apps/desktop/e2e/event-journal.spec.ts

@@ -1,6 +1,7 @@
 import { test, expect } from "@playwright/test";
 import fs from "node:fs";
 import path from "node:path";
+import { DatabaseSync } from "node:sqlite";
 import { EventJournal, replayItems } from "../src/main/engine/event-journal";
 import { createThread, launch, seedHome } from "./support";
 
@@ -52,3 +53,47 @@ test("without the flag the app writes no journal", async () => {
     fs.rmSync(repo, { recursive: true, force: true });
   }
 });
+
+// CF-6 (#279): the app's journal keeps its identity across a clean relaunch, so paired clients still
+// resume, and takes a new one after the app is killed, so a cursor issued before the crash reloads.
+test("the app's journal keeps its identity across a clean relaunch and takes a new one after a kill", async () => {
+  const { home, repo } = seedHome();
+  const previous = process.env.MODEX_EVENT_JOURNAL;
+  process.env.MODEX_EVENT_JOURNAL = "1";
+  const file = path.join(home, "app", "events.sqlite");
+  // Read-only, so this test never marks the journal open or closed itself.
+  const meta = () => {
+    const db = new DatabaseSync(file, { readOnly: true });
+    try {
+      const value = (key: string) => (db.prepare("SELECT value FROM meta WHERE key = ?").get(key) as { value: string } | undefined)?.value;
+      return { id: value("id"), open: value("open") };
+    } finally { db.close(); }
+  };
+  try {
+    let run = await launch(home);
+    try { await createThread(run.page, "journal this session"); } finally { await run.app.close(); }
+    const first = meta();
+    expect(first.id).toBeTruthy();
+    expect(first.open).toBe("0");
+
+    run = await launch(home);
+    await run.app.close();
+    expect(meta()).toEqual(first);
+
+    run = await launch(home);
+    const killed = new Promise((resolve) => run.app.process().once("exit", resolve));
+    run.app.process().kill("SIGKILL");
+    await killed;
+    expect(meta()).toEqual({ id: first.id, open: "1" });
+
+    run = await launch(home);
+    try {
+      await expect.poll(() => meta().id).not.toBe(first.id);
+    } finally { await run.app.close(); }
+    expect(meta().open).toBe("0");
+  } finally {
+    if (previous === undefined) delete process.env.MODEX_EVENT_JOURNAL; else process.env.MODEX_EVENT_JOURNAL = previous;
+    fs.rmSync(home, { recursive: true, force: true });
+    fs.rmSync(repo, { recursive: true, force: true });
+  }
+});

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

@@ -400,6 +400,13 @@ export function mobileItem(item: ThreadItem): Record<string, unknown> {
     // Phones cannot answer questions yet: a notice every client can show, with the questions and never the answers.
     case "question": return { id: item.id, kind: "notice", level: "info", at: item.at,
       text: `The agent asked: ${item.questions.map((q) => q.text).join(" · ")}\n${item.answers ? (item.delivery === "uncertain" ? "Answered on your Mac; Modex can't tell whether the agent received it." : "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.
+    default: {
+      const unknown = item as { id: string; at: string };
+      return { id: unknown.id, kind: "notice", level: "info", text: "An item from a newer version of Modex. Update Modex on your Mac to see it.", at: unknown.at };
+    }
   }
 }
 

apps/desktop/src/main/engine/desktop-client.ts

@@ -273,7 +273,9 @@ export class DesktopClient extends EventEmitter {
         res.on("data", (chunk: string) => {
           if (generation !== this.generation || ended) return;
           buffer += chunk;
-          if (buffer.length > 1024 * 1024) { res.destroy(); return; }
+          // A frame this large is dropped. Resuming would replay it into the same limit forever, so
+          // forget the cursor: the reconnect reloads state, which carries the item another way.
+          if (buffer.length > 1024 * 1024) { this.journal = undefined; res.destroy(); return; }
           let end: number;
           while ((end = buffer.indexOf("\n\n")) >= 0) {
             const frame = buffer.slice(0, end);

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

@@ -23,7 +23,10 @@ export const RESUME_LIMIT = 10_000;
 export interface EventJournalOptions {
   flushMs?: number;
   retention?: JournalRetention;
-  /** A timed flush failed. The journal stays usable for the caller to close. */
+  /**
+   * A write failed, on a timer or while a client read. Rows were lost, so the journal records
+   * nothing more and resumes no cursor; it stays usable for the caller to close. Called once.
+   */
   onError?: (error: unknown) => void;
 }
 
@@ -52,7 +55,11 @@ export class EventJournal {
   private listeners = new Set<(entries: JournalEntry[]) => void>();
   /** The highest cursor retention has deleted; 0 before any pruning. */
   private prunedThrough: number;
+  /** The highest cursor written, so a reload can still name one after a failed write. */
+  private written: number;
   private closed = false;
+  /** A write failed and its rows are gone: nothing more is recorded and no cursor resumes. */
+  private failed = false;
 
   constructor(file: string, options: EventJournalOptions = {}) {
     this.flushMs = options.flushMs ?? 250;
@@ -71,18 +78,37 @@ export class EventJournal {
       );
       CREATE INDEX IF NOT EXISTS events_thread ON events (thread_id, seq);
       CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);`);
-    this.db.prepare("INSERT OR IGNORE INTO meta (key, value) VALUES ('id', ?)").run(crypto.randomUUID());
+    // `open` stays "1" until a clean close. A journal still marked open was left by a process that
+    // died, or one whose write failed, so its last batch may be missing while the state files have
+    // it: a new identity makes every cursor issued before then reload instead of resuming past it.
+    if (this.meta("open") === "1") this.setMeta("id", crypto.randomUUID());
+    else this.db.prepare("INSERT OR IGNORE INTO meta (key, value) VALUES ('id', ?)").run(crypto.randomUUID());
+    this.setMeta("open", "1");
     this.id = this.meta("id")!;
     this.prunedThrough = Number(this.meta("pruned_through") ?? 0);
+    this.written = Number((this.db.prepare("SELECT COALESCE(MAX(seq), 0) AS seq FROM events").get() as { seq: number }).seq);
     this.insert = this.db.prepare("INSERT INTO events (at, thread_id, type, event) VALUES (?, ?, ?, ?)");
   }
 
   private meta(key: string): string | undefined {
     return (this.db.prepare("SELECT value FROM meta WHERE key = ?").get(key) as { value: string } | undefined)?.value;
   }
 
+  private setMeta(key: string, value: string): void {
+    this.db.prepare("INSERT INTO meta (key, value) VALUES (?, ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value").run(key, value);
+  }
+
+  /** Rows were lost: record nothing more, resume no cursor, and tell the caller once. */
+  private fail(error: unknown): void {
+    if (this.failed) return;
+    this.failed = true;
+    this.pending = [];
+    this.slots = new Map();
+    this.onError?.(error);
+  }
+
   record(event: ThreadEvent): void {
-    if (this.closed) return;
+    if (this.closed || this.failed) return;
     const key = mergeKey(event);
     const slot = key === undefined ? undefined : this.slots.get(key);
     if (slot !== undefined) this.pending[slot] = merge(this.pending[slot]!, event);
@@ -94,7 +120,7 @@ export class EventJournal {
     }
     this.timer ??= setTimeout(() => {
       this.timer = undefined;
-      try { this.flush(); } catch (error) { this.onError?.(error); }
+      try { this.flush(); } catch { /* flush reported it */ }
     }, this.flushMs);
     this.timer.unref?.();
   }
@@ -108,14 +134,20 @@ export class EventJournal {
     this.slots = new Map();
     const at = new Date().toISOString();
     const entries: JournalEntry[] = [];
-    this.db.exec("BEGIN");
     try {
-      for (const event of batch) entries.push({ seq: Number(this.insert.run(at, event.threadId, event.type, JSON.stringify(event)).lastInsertRowid), at, event });
-      this.db.exec("COMMIT");
+      this.db.exec("BEGIN");
+      try {
+        for (const event of batch) entries.push({ seq: Number(this.insert.run(at, event.threadId, event.type, JSON.stringify(event)).lastInsertRowid), at, event });
+        this.db.exec("COMMIT");
+      } catch (error) {
+        try { this.db.exec("ROLLBACK"); } catch { /* the write already failed */ }
+        throw error;
+      }
     } catch (error) {
-      this.db.exec("ROLLBACK");
+      this.fail(error);
       throw error;
     }
+    this.written = entries.at(-1)!.seq;
     for (const listener of this.listeners) {
       try { listener(entries); }
       catch (error) { console.error("[modex] event journal subscriber failed:", error); }
@@ -152,9 +184,11 @@ export class EventJournal {
    * usable cursor, the answer is "reload state" with the cursor to continue from after that reload.
    */
   resume(id: string | undefined, cursor: number | undefined, opts: { threadId?: string } = {}): JournalResume {
-    const latest = this.latest();
+    let latest: number;
+    // Reading flushes. A write that fails here is reported, and the client reloads.
+    try { latest = this.latest(); } catch { return { id: this.id, latest: Math.max(this.prunedThrough, this.written) }; }
     const reload = { id: this.id, latest };
-    if (id !== this.id || cursor === undefined || !Number.isSafeInteger(cursor) || cursor < this.prunedThrough || cursor > latest) return reload;
+    if (this.failed || id !== this.id || cursor === undefined || !Number.isSafeInteger(cursor) || cursor < this.prunedThrough || cursor > latest) return reload;
     const count = Number(((opts.threadId
       ? this.db.prepare("SELECT COUNT(*) AS n FROM events WHERE seq > ? AND thread_id = ?").get(cursor, opts.threadId)
       : this.db.prepare("SELECT COUNT(*) AS n FROM events WHERE seq > ?").get(cursor)) as { n: number }).n);
@@ -191,6 +225,10 @@ export class EventJournal {
     try { this.flush(); } finally {
       this.closed = true;
       this.listeners.clear();
+      // Only a journal that lost nothing closes clean; otherwise the next open takes a new identity.
+      if (!this.failed) {
+        try { this.setMeta("open", "0"); } catch { /* left open: the next open is treated as a crash */ }
+      }
       this.db.close();
     }
   }

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

@@ -2,6 +2,7 @@ import { test } from "node:test";
 import assert from "node:assert/strict";
 import fs from "node:fs";
 import path from "node:path";
+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";
@@ -136,3 +137,109 @@ test("loading a thread with nothing to settle sends nothing and leaves its file
     await runner.dispose();
   } finally { old.journal.close(); }
 });
+
+test("an item kind this Mac does not know reaches the phone as a readable, inert notice, never as null", () => {
+  // After a downgrade, the state files can hold items a newer Modex wrote. One null item makes the
+  // phone reject its whole snapshot, so every item must map to something it can show.
+  const future = { id: "w1", kind: "widget", at: "2026-10-18T00:00:00.000Z", body: "whatever a newer Modex put here" } as unknown as ThreadItem;
+  const phone = mobileItem(future);
+  assert.ok(phone, "never null or undefined");
+  assert.equal(phone.id, "w1");
+  assert.equal(phone.kind, "notice", "a kind every phone build renders, with nothing to act on");
+  assert.equal(typeof phone.text, "string");
+  assert.ok(!JSON.stringify(phone).includes("whatever a newer Modex"), "an unknown item's fields are not forwarded: they may hold anything");
+  assert.ok(!JSON.stringify([future, future].map(mobileItem)).includes("null"), "a snapshot with unknown items has no null entries");
+});
+
+/** Every file under `home` except the journal's own, by its bytes. */
+function besideJournal(home: string): Map<string, string> {
+  const files = new Map<string, string>();
+  const walk = (dir: string) => {
+    for (const entry of fs.readdirSync(dir, { withFileTypes: true })) {
+      const file = path.join(dir, entry.name);
+      if (entry.isDirectory()) walk(file);
+      else if (!entry.name.startsWith("events.sqlite")) files.set(path.relative(home, file), fs.readFileSync(file).toString("base64"));
+    }
+  };
+  walk(home);
+  return files;
+}
+
+// v0.1.0's journal schema, verbatim (`git show v0.1.0:apps/desktop/src/main/engine/event-journal.ts`):
+// an events table, no meta table, so no identity and no cursors.
+const V010_SCHEMA = `PRAGMA journal_mode = WAL;
+  PRAGMA synchronous = NORMAL;
+  CREATE TABLE IF NOT EXISTS events (
+    seq INTEGER PRIMARY KEY AUTOINCREMENT,
+    at TEXT NOT NULL,
+    thread_id TEXT NOT NULL,
+    type TEXT NOT NULL,
+    event TEXT NOT NULL
+  );
+  CREATE INDEX IF NOT EXISTS events_thread ON events (thread_id, seq);`;
+
+test("a journal v0.1.0 wrote opens with its rows, takes an identity, replays unchanged, and leaves drafts and state files alone", async () => {
+  const home = tmpdir("modex-migration-v010-");
+  const store = new Store(home);
+  const project = store.addProject(gitRepo());
+  const setup = new ThreadRunner({ home, store, emit: () => {} });
+  const thread = await setup.createThread(project.id, { backend: "claude", auto: false });
+  store.updateThread(thread.id, { title: "Old thread", titleSource: "manual" });
+  store.saveItems(thread.id, LEGACY);
+  await setup.dispose();
+  fs.writeFileSync(path.join(store.dir, "update-recovery.json"), JSON.stringify({ drafts: [{ threadId: thread.id, text: "half a reply" }] }));
+  const file = path.join(home, "app", "events.sqlite");
+  const old = new DatabaseSync(file);
+  old.exec(V010_SCHEMA);
+  const insert = old.prepare("INSERT INTO events (at, thread_id, type, event) VALUES (?, ?, ?, ?)");
+  for (const item of LEGACY) insert.run(at, thread.id, "item", JSON.stringify({ threadId: thread.id, type: "item", item }));
+  old.close();
+  const files = besideJournal(home);
+
+  const journal = new EventJournal(file, { flushMs: 5 });
+  try {
+    assert.match(journal.id, /^[\w-]{1,64}$/, "it gains an identity clients can resume from");
+    assert.equal(journal.latest(), LEGACY.length, "every row v0.1.0 wrote is kept, and cursors continue after them");
+    assert.deepEqual(replayItems(journal.since(0, { threadId: thread.id }), thread.id), LEGACY, "and replays to the saved transcript");
+    assert.equal(journal.resume(undefined, undefined).entries, undefined, "a client from before cursors reloads");
+    const { runner } = runnerOn(home, journal);
+    assert.deepEqual(runner.items(thread.id), LEGACY, "the transcript loads with every word");
+    await runner.dispose();
+  } finally { journal.close(); }
+  const relaunched = new EventJournal(file);
+  const id = relaunched.id;
+  relaunched.close();
+  assert.equal(new EventJournal(file).id, id, "the identity is kept across clean relaunches");
+  assert.deepEqual(besideJournal(home), files, "the draft and the state files are byte-identical");
+});
+
+test("a journal this build wrote stays usable after a downgrade: v0.1.0's and v0.2.0's own statements still read and write it", () => {
+  const file = path.join(tmpdir("modex-migration-rollback-"), "events.sqlite");
+  const journal = new EventJournal(file);
+  journal.record({ threadId: "t1", type: "item", item: LEGACY[0]! });
+  const id = journal.id;
+  const cursor = journal.latest();
+  journal.close();
+
+  // v0.1.0 opening it: its schema is a subset, so it opens, reads and appends.
+  const v010 = new DatabaseSync(file);
+  v010.exec(V010_SCHEMA);
+  v010.prepare("INSERT INTO events (at, thread_id, type, event) VALUES (?, ?, ?, ?)").run(at, "t1", "item", JSON.stringify({ threadId: "t1", type: "item", item: LEGACY[1] }));
+  assert.equal((v010.prepare("SELECT seq, at, event FROM events WHERE seq > ? ORDER BY seq LIMIT ?").all(0, 10) as unknown[]).length, 2);
+  v010.close();
+
+  // v0.2.0 opening it: the same meta table; it keeps the identity and ignores the `open` flag.
+  const v020 = new DatabaseSync(file);
+  v020.exec(`${V010_SCHEMA}
+  CREATE TABLE IF NOT EXISTS meta (key TEXT PRIMARY KEY, value TEXT NOT NULL);`);
+  v020.prepare("INSERT OR IGNORE INTO meta (key, value) VALUES ('id', ?)").run("would-be-new-id");
+  assert.equal((v020.prepare("SELECT value FROM meta WHERE key = ?").get("id") as { value: string }).value, id, "v0.2.0 keeps the identity this build gave it");
+  assert.equal(Number((v020.prepare("SELECT COALESCE(MAX(seq), 0) AS seq FROM events").get() as { seq: number }).seq), cursor + 1);
+  v020.close();
+
+  // Back on this build: nothing the older builds did confuses it.
+  const back = new EventJournal(file);
+  assert.equal(back.id, id, "an older build's clean use does not look like a crash");
+  assert.equal(back.since(0).length, 2);
+  back.close();
+});