test: make five CI-flaky unit and contract tests reliable

server/conversation.test.ts

@@ -286,6 +286,8 @@ describe("Devin pane identity", () => {
     const pane = { pane_id: "pane-a", cwd, agent: "devin", agent_session: { agent: "devin", kind: "id", value: "one" } } as HerdrPane;
     const db = new Database(dbPath);
     try {
+      // Cache identity needs committed rows, not a durable sync for each fixture node.
+      db.exec("PRAGMA synchronous = OFF");
       db.exec("CREATE TABLE sessions(id TEXT, working_directory TEXT, main_chain_id INTEGER, hidden INTEGER, model TEXT); CREATE TABLE message_nodes(session_id TEXT,node_id INTEGER,parent_node_id INTEGER,chat_message TEXT,created_at INTEGER); CREATE TABLE tool_call_state(session_id TEXT,tool_call_id TEXT,tool_call_json TEXT,tool_call_update_json TEXT)");
       // more turns than one page holds, so the newest page carries a cursor
       db.query("INSERT INTO sessions VALUES ('one', ?, 120, 0, NULL)").run(cwd);
@@ -1859,6 +1861,8 @@ describe("Agy native ledger through shared paging", () => {
     mkdirSync(join(path, ".."), { recursive: true });
     mkdirSync(join(root, "conversations"));
     db = new Database(join(root, "conversations", `${session}.db`));
+    // The ledger tests observe committed DB changes, never recovery after a crash.
+    db.exec("PRAGMA synchronous = OFF");
     db.exec("CREATE TABLE steps(idx INTEGER PRIMARY KEY)");
     writeFileSync(path, "");
     forgetTranscriptState();

server/core/omo-pane-send.test.ts

@@ -58,8 +58,11 @@ beforeEach(() => {
   });
 });
 
-afterEach(() => {
-  for (const child of children.splice(0)) child.kill();
+afterEach(async () => {
+  await Promise.all(children.splice(0).map(async (child) => {
+    child.kill();
+    await child.exited;
+  }));
   listener.stop(true);
   if (socketBefore === undefined) delete process.env["HERDR_SOCKET"];
   else process.env["HERDR_SOCKET"] = socketBefore;
@@ -72,10 +75,13 @@ afterEach(() => {
  */
 async function heldOmo(): Promise<{ pid: number; argv: string[]; cwd: string }> {
   const agentDir = join(root, "agent");
-  const child = Bun.spawn(["sh", "-c", "echo ready; exec sleep 60"], { cwd: root, env: { OMO_CODING_AGENT_DIR: agentDir, PATH: process.env["PATH"] ?? "" }, stdout: "pipe" });
+  // Keep the process that signals ready: another exec can make /proc/<pid>/environ
+  // briefly empty while the composer looks up the store holding this session.
+  const child = Bun.spawn(["sh", "-c", "echo ready; read -r stop"], { cwd: root, env: { OMO_CODING_AGENT_DIR: agentDir, PATH: process.env["PATH"] ?? "" }, stdin: "pipe", stdout: "pipe" });
   children.push(child);
   const reader = child.stdout.getReader();
-  await reader.read();
+  const ready = await reader.read();
+  expect(new TextDecoder().decode(ready.value)).toBe("ready\n");
   reader.releaseLock();
   const holders = join(agentDir, "sessions", omoSessionFolder(root), "session-holders", "sess-1");
   mkdirSync(holders, { recursive: true });

server/opencode.contract.test.ts

@@ -240,7 +240,8 @@ describe("OpenCode v2 native service", () => {
   /** runs once, inside the next session lookup, before the fake answers it */
   let duringLookup: (() => Promise<void>) | undefined;
   /** every chunk the stand-in reads from its terminal, reported by the stand-in itself */
-  let onTyped: ((chunk: string) => void) | undefined;
+  let onTyped: ((chunk: string, start: number) => void) | undefined;
+  const typedWaiters = new Set<() => void>();
   /** while set, session lookups get no answer (the bridge's own request timeout ends them) */
   let withholdLookups = false;
   /** while set, every request but /api/info gets no answer: a service that hangs */
@@ -285,12 +286,13 @@ describe("OpenCode v2 native service", () => {
     writeFileSync(join(nativeRoot, "screen.txt"), recorded("first-selected"));
     writeFileSync(join(nativeRoot, "accepted.txt"), recorded("accepted"));
     writeFileSync(standIn, `#!${process.execPath}
-const { appendFileSync, readFileSync } = require("node:fs");
+const { appendFileSync, readFileSync, statSync } = require("node:fs");
 let typed = "";
 process.stdin.setRawMode(true);
 process.stdin.on("data", (chunk) => {
   appendFileSync(${JSON.stringify(typedLog)}, chunk);
-  fetch(readFileSync(${JSON.stringify(join(nativeRoot, "fake.url"))}, "utf8") + "/test/typed", { method: "POST", body: chunk }).catch(() => {});
+  const start = statSync(${JSON.stringify(typedLog)}).size - chunk.length;
+  fetch(readFileSync(${JSON.stringify(join(nativeRoot, "fake.url"))}, "utf8") + "/test/typed", { method: "POST", headers: { "x-typed-start": String(start) }, body: chunk }).catch(() => {});
   const before = typed;
   typed += chunk.toString("latin1");
   if (!before.endsWith("@effect") && typed.endsWith("@effect")) process.stdout.write("\\u001b[2J\\u001b[H" + readFileSync(${JSON.stringify(join(nativeRoot, "screen.txt"))}, "utf8"));
@@ -303,7 +305,11 @@ process.stdin.on("data", (chunk) => {
       fetch(request) {
         const url = new URL(request.url);
         if (url.pathname === "/api/info") return Response.json(info);
-        if (url.pathname === "/test/typed") return request.text().then((chunk) => { onTyped?.(chunk); return new Response(null, { status: 204 }); });
+        if (url.pathname === "/test/typed") return request.text().then((chunk) => {
+          onTyped?.(chunk, Number(request.headers.get("x-typed-start")));
+          for (const wake of typedWaiters) wake();
+          return new Response(null, { status: 204 });
+        });
         lastRequestAt = performance.now();
         if (url.pathname === "/api/event" && eventMode === "slow") return new Promise<Response>((resolve) => withheld.add(() => resolve(new Response(null, { status: 503 }))));
         if (url.pathname === "/api/event" && eventMode !== "ok") return new Response(null, { status: eventMode === "absent" ? 404 : 503 });
@@ -456,13 +462,15 @@ process.stdin.on("data", (chunk) => {
         await Bun.sleep(50);
       }
       await browserCatalog();
+      const completed = typedSince(before);
       ws.send(JSON.stringify({ type: "submit", id: 2, pane_id: nativePane, text: "@effect apply it", payload: "@effect apply it", delivery: "queue" }));
       expect(await frame((message) => message.type === "submit-result" && message.id === 2)).toMatchObject({ ok: true, pending: { state: "queued" } });
       expect(typed().length).toBe(before);
       await herdrRpc("pane.report_agent", { pane_id: nativePane, source: "herdr:opencode", agent: "opencode", agent_session_id: NATIVE_SESSION, state: "idle" });
       const delivered = await frame((message) => message.type === "pending-messages" && Array.isArray(message.removed) && message.removed.length > 0);
       expect(JSON.stringify(delivered.removed)).not.toContain("error");
-      expect(typed().slice(before)).toBe("@effect\t\u001b[200~apply it\u001b[201~\r");
+      // The receipt acknowledges the PTY write, not the stand-in reading its Enter.
+      expect(await completed).toBe("@effect\t\u001b[200~apply it\u001b[201~\r");
     } finally { ws.close(); }
   });
 /** A socket recording every frame; `frame` resolves on the first one (already seen or next) matching. */
@@ -581,13 +589,21 @@ process.stdin.on("data", (chunk) => {
 
   const outcomesAt = async (port: number, pane: string): Promise<CommandOutcome[]> =>
     ((await (await fetch(`http://127.0.0.1:${port}/api/pane/prompt?pane_id=${encodeURIComponent(pane)}`)).json()) as { outcomes?: CommandOutcome[] }).outcomes ?? [];
-  const typedSince = async (before: number): Promise<string> => {
-    for (const deadline = Date.now() + 5_000; !typed().slice(before).endsWith("\r");) {
-      if (Date.now() > deadline) throw new Error(`typed: ${JSON.stringify(typed().slice(before))}`);
-      await Bun.sleep(50);
-    }
-    return typed().slice(before);
-  };
+  const typedSince = (before: number): Promise<string> => new Promise((resolve, reject) => {
+    const timeout = setTimeout(() => {
+      typedWaiters.delete(check);
+      reject(new Error(`typed: ${JSON.stringify(typed().slice(before))}`));
+    }, 5_000);
+    const check = () => {
+      const text = typed().slice(before);
+      if (!text.endsWith("\r")) return;
+      clearTimeout(timeout);
+      typedWaiters.delete(check);
+      resolve(text);
+    };
+    typedWaiters.add(check);
+    check();
+  });
 
   it("attaches @effect and records the send after the pane's OpenCode session changed, with no catalog request since", async () => {
     const { ws, frame } = await socket();
@@ -808,18 +824,25 @@ process.stdin.on("data", (chunk) => {
           forgetOpenCodeService();
           eventMode = mode;
           const typedFirst = Promise.withResolvers<number>();
-          onTyped = () => typedFirst.resolve(performance.now() - lastRequestAt);
           const text = path === "prompt" ? `/review ${mode}` : `@effect ${path} ${mode}`;
           const request = ++id;
           if (path === "queued") {
             await reportState("working");
             await untilStatus("working");
           }
+          const before = typed().length;
+          // Reports can arrive out of order; identify this send's first byte, regardless of chunking.
+          onTyped = (_chunk, start) => { if (start === before) typedFirst.resolve(performance.now() - lastRequestAt); };
+          const completed = typedSince(before);
           ws.send(JSON.stringify({ type: "submit", id: request, pane_id: nativePane, text, payload: text, ...(path === "queued" ? { delivery: "queue" } : {}) }));
           const answer = await frame((message) => message.type === "submit-result" && message.id === request);
           expect(answer).toMatchObject({ ok: true });
           if (path === "queued") await reportState("idle");
           gaps[`${path}/${mode}`] = Math.round(await typedFirst.promise);
+          // Keep this delivery's observer and service alive until its whole send has reached the TUI.
+          await completed;
+          if (path === "queued") await frame((message) => message.type === "pending-messages"
+            && Array.isArray(message.removed) && message.removed.some((item) => item.id === (answer.pending as { id: string }).id));
           onTyped = undefined;
           for (const release of withheld) release();
           withheld.clear();

server/opencode.test.ts

@@ -38,7 +38,8 @@ function store(session = "ses_test1") {
   const path = join(root, `opencode-${++stores}.db`);
   const db = new Database(path, { create: true });
   opened.push(db);
-  db.exec("PRAGMA journal_mode = WAL");
+  // These stores test committed visibility and locking, not crash durability.
+  db.exec("PRAGMA journal_mode = WAL; PRAGMA synchronous = OFF");
   db.exec(SCHEMA);
   db.query("INSERT INTO session_v2 (id, project_id, slug, directory, title, version, time_created, time_updated) VALUES (?, 'global', 'slug', ?, 'Test', '2.0.24', ?, ?)").run(session, root, T0, T0);
   let seq = 3;
@@ -82,7 +83,8 @@ function storeV1(session = "ses_old1") {
   const path = join(root, `opencode-v1-${++stores}.db`);
   const db = new Database(path, { create: true });
   opened.push(db);
-  db.exec("PRAGMA journal_mode = WAL");
+  // These stores test committed visibility and locking, not crash durability.
+  db.exec("PRAGMA journal_mode = WAL; PRAGMA synchronous = OFF");
   db.exec(SCHEMA_V1);
   db.query("INSERT INTO session (id, project_id, slug, directory, title, version, time_created, time_updated) VALUES (?, 'global', 'slug', ?, 'Old', '1.18.32', ?, ?)").run(session, root, T0, T0);
   let clock = T0;