From 005d2ec4c5374c0eee5ea2b2cc123a3b75cfed72 Mon Sep 17 00:00:00 2001 From: Hako Date: Thu, 22 Oct 2026 22:24:36 +0900 Subject: [PATCH 1/5] test(omo): keep the pane fixture stable after readiness The ready shell exec'd sleep while the composer read its environment. Linux can expose an empty /proc/pid/environ during exec, hiding the fixture's store and dropping bracketed paste markers. Keep the ready shell blocked on stdin instead, validate its signal, and await fixture exit during cleanup. --- server/core/omo-pane-send.test.ts | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/server/core/omo-pane-send.test.ts b/server/core/omo-pane-send.test.ts index 71e3591e2c..2a68696d18 100644 --- a/server/core/omo-pane-send.test.ts +++ b/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//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 }); From aa9b3741a38a401ddf7e64229bc2b4d01d608158 Mon Sep 17 00:00:00 2001 From: Hako Date: Thu, 22 Oct 2026 22:25:55 +0900 Subject: [PATCH 2/5] test(opencode): avoid durability syncs in paging fixtures --- server/opencode.test.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/server/opencode.test.ts b/server/opencode.test.ts index a7cce5456b..e5dc795d08 100644 --- a/server/opencode.test.ts +++ b/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; From 1174614342b9c04953800bd6d0c70035c6df86d5 Mon Sep 17 00:00:00 2001 From: Hako Date: Thu, 22 Oct 2026 22:26:09 +0900 Subject: [PATCH 3/5] test(conversation): avoid durability syncs in cache and ledger fixtures --- server/conversation.test.ts | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/server/conversation.test.ts b/server/conversation.test.ts index 2baec39855..d788635ab7 100644 --- a/server/conversation.test.ts +++ b/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(); From 45ad81cbc4e1385f054995efd529b373ae234080 Mon Sep 17 00:00:00 2001 From: Hako Date: Thu, 22 Oct 2026 22:27:09 +0900 Subject: [PATCH 4/5] test(opencode): await terminal consumption between contract sends --- server/opencode.contract.test.ts | 46 +++++++++++++++++++++++--------- 1 file changed, 34 insertions(+), 12 deletions(-) diff --git a/server/opencode.contract.test.ts b/server/opencode.contract.test.ts index dc2bcd6f2f..4fc080c1a6 100644 --- a/server/opencode.contract.test.ts +++ b/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) | 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 */ @@ -290,7 +291,7 @@ 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(() => {}); + fetch(readFileSync(${JSON.stringify(join(nativeRoot, "fake.url"))}, "utf8") + "/test/typed", { method: "POST", headers: { "x-typed-start": String(typed.length) }, 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 +304,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((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 +461,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 +588,21 @@ process.stdin.on("data", (chunk) => { const outcomesAt = async (port: number, pane: string): Promise => ((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 => { - 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 => 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 +823,25 @@ process.stdin.on("data", (chunk) => { forgetOpenCodeService(); eventMode = mode; const typedFirst = Promise.withResolvers(); - 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(); From 20cb46b5170b96606edc01e82786467bdaae9b40 Mon Sep 17 00:00:00 2001 From: Hako Date: Thu, 22 Oct 2026 22:30:33 +0900 Subject: [PATCH 5/5] test(opencode): correlate reports with shared terminal log offsets --- server/opencode.contract.test.ts | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/server/opencode.contract.test.ts b/server/opencode.contract.test.ts index 4fc080c1a6..32a9bbfaf4 100644 --- a/server/opencode.contract.test.ts +++ b/server/opencode.contract.test.ts @@ -286,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", headers: { "x-typed-start": String(typed.length) }, 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"));