diff --git a/changelog.d/gjc-native-turn-status.md b/changelog.d/gjc-native-turn-status.md new file mode 100644 index 0000000000..1e5db15a7e --- /dev/null +++ b/changelog.d/gjc-native-turn-status.md @@ -0,0 +1,7 @@ +### Fixed +- A Gajae Code pane's status now comes from gjc's own session transcript instead of what herdr + reports for the pane, which could keep a finished turn at RUN for work a provider child was + still doing — and a turn gjc ran while the child was quiet never read RUN or sent a done + alert. A turn that ended in an error or was aborted reads READY rather than DONE, a + background job's result that wakes the agent reads RUN, and a pane whose transcript cannot + be told keeps herdr's status. diff --git a/docs/development.md b/docs/development.md index 5fb3c040ae..57b639cd44 100644 --- a/docs/development.md +++ b/docs/development.md @@ -157,6 +157,12 @@ retried automatically. The restart gate waits for the interrupted turn's readine by the new turn's working/ready cycle. If Codex never exposes that cycle, remaining messages are held for explicit review. +## GJC parent-turn status + +GJC status follows the native transcript bound to that pane, rather than a provider child's stale status or session ID. A user message, tool result or tool call keeps the parent turn working; an explicit final answer ends it. Native background-result, monitor and scheduled-task notifications start another turn, and an open `ask` waits for input. + +This is parent-turn state, not a count of live background jobs. GJC's job manager keeps that live state in memory, so historical job starts are not treated as proof that a job still runs. Missing or unreadable native evidence leaves herdr's status in charge; interrupted old turns are not successful completions. The isolated `scripts/gjc-status-browser-regression.ts` fixture exercises the real server and browser without invoking a provider or touching existing sessions. + ## OmO runtime recognition A pane runs OmO when its foreground processes prove it. The first proof is the program diff --git a/docs/screenshots/gjc-status-completion.png b/docs/screenshots/gjc-status-completion.png new file mode 100644 index 0000000000..78e02d7c0c Binary files /dev/null and b/docs/screenshots/gjc-status-completion.png differ diff --git a/package.json b/package.json index 53edff0dfd..d3fac8c23a 100644 --- a/package.json +++ b/package.json @@ -57,5 +57,6 @@ "patchedDependencies": { "@legendapp/list@3.3.5": "patches/@legendapp%2Flist@3.3.5.patch", "@xterm/xterm@5.5.0": "patches/@xterm%2Fxterm@5.5.0.patch" - } + }, + "packageManager": "bun@1.4.2" } diff --git a/scripts/ci-browser.sh b/scripts/ci-browser.sh index 9f13d5f5f1..6f98a23eef 100644 --- a/scripts/ci-browser.sh +++ b/scripts/ci-browser.sh @@ -72,6 +72,7 @@ run_script scripts/chat-fallback-browser-qa.ts run_script scripts/grok-chat-browser-qa.ts run_script scripts/omo-chat-browser-qa.ts run_script scripts/agy-chat-browser-qa.ts +run_script scripts/gjc-status-browser-regression.ts run_script scripts/math-browser-qa.ts run_script scripts/file-viewer-regression.ts run_script scripts/pwa-update-browser-qa.ts diff --git a/scripts/ci-browser.test.ts b/scripts/ci-browser.test.ts index b95f77232e..c26d2fafce 100644 --- a/scripts/ci-browser.test.ts +++ b/scripts/ci-browser.test.ts @@ -69,7 +69,8 @@ describe("browser lane", () => { it("summarizes every script and exits zero on success without enabling local evidence", async () => { const { text, code, evidence } = await lane(); expect(code).toBe(0); - expect(rows(text)).toHaveLength(53); + expect(rows(text)).toHaveLength(54); + expect(rows(text).map((row) => row.script)).toContain("scripts/gjc-status-browser-regression.ts"); expect(rows(text).map((row) => row.script)).toContain("scripts/chat-virtual-browser-qa.ts"); expect(rows(text).map((row) => row.script)).toContain("scripts/pwa-update-browser-qa.ts"); expect(rows(text).map((row) => row.script)).toContain("scripts/omo-chat-browser-qa.ts"); diff --git a/scripts/gjc-status-browser-regression.ts b/scripts/gjc-status-browser-regression.ts new file mode 100644 index 0000000000..2354c8032f --- /dev/null +++ b/scripts/gjc-status-browser-regression.ts @@ -0,0 +1,268 @@ +import "./test-herdr.ts"; +import assert from "node:assert/strict"; +import { mkdtempSync, mkdirSync, writeFileSync, appendFileSync, rmSync, readFileSync, chmodSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { chromium, type Browser, type Page } from "playwright-core"; +import { createServer } from "../server/index.ts"; +import { herdrRpc, herdrSocketPath, sessionSnapshot, workspaceCreate, workspaceClose } from "../server/herdr/client.ts"; +import { gjcTranscriptForPane, isGjcProcess } from "../server/gjc-runtime.ts"; +import { UsageService } from "../server/usage.ts"; +import type { ServerMessage, SessionSnapshot } from "../shared/protocol.ts"; + +type Status = "idle" | "working" | "done" | "blocked"; +type NativeRecord = Readonly>; +const root = mkdtempSync(join(tmpdir(), "herdr-gjc-status-browser-")); +const store = join(root, ".gjc", "agent", "sessions", "v2-fixture"); +const script = join(root, "gjc"); +const oldPath = process.env.PATH; +const sdkRows: Record[] = []; +const sdkRegistry = join(root, "sdk-fixture.json"); +const oldHome = process.env.HOME, oldSocket = process.env.HERDR_SOCKET; +const evidenceDirectory = process.env.UI_EVIDENCE_DIR || process.env.CHECK_BROWSER_EVIDENCE_DIR; +const workspaces: string[] = [], frames: ServerMessage[] = []; +const outcomes: { scenario: string; outcome: "passed" }[] = []; +let server: ReturnType | undefined, browser: Browser | undefined; +let statusSocket: WebSocket | undefined, page: Page | undefined; +let socketError: Error | undefined, origin = "", scenario = "fixture setup"; +let deferredCleanupError: AggregateError | undefined; + +function quote(value: string): string { return `'${value.replaceAll("'", "'\\''")}'`; } +async function until(check: () => boolean | Promise, label: string): Promise { + const deadline = Date.now() + 20_000; + while (!(await check())) { + if (socketError) throw socketError; + assert.ok(Date.now() < deadline, `Timed out: ${label}`); + await Bun.sleep(50); + } +} +function history(path: string) { + let previousId: string | null = null, sequence = 0; + const record = (fields: NativeRecord): NativeRecord => { + const id = `fixture-${++sequence}`; + const result = { id, parentId: previousId, timestamp: new Date().toISOString(), ...fields }; + previousId = id; + return result; + }; + return { + message(role: "user" | "assistant" | "toolResult", text: string, ending?: "stop" | "toolUse") { + return record({ type: "message", message: { role, content: [{ type: "text", text }], + ...(role === "assistant" ? { stopReason: ending ?? "stop" } : {}), + ...(role === "toolResult" ? { toolCallId: "q1" } : {}) } }); + }, + ask: () => record({ type: "message", message: { role: "assistant", stopReason: "toolUse", content: [ + { type: "toolCall", id: "q1", name: "ask", arguments: { questions: [ + { question: "Synthetic choice?", options: [{ label: "Continue" }] }, + ] } }, + ] } }), + wake: () => record({ type: "custom_message", customType: "async-result", display: false, content: "Synthetic background result" }), + append: (...records: NativeRecord[]) => appendFileSync(path, records.map((item) => JSON.stringify(item) + "\n").join("")), + }; +} +async function fixture(transcript: string): Promise { + const result = await workspaceCreate({ cwd: root, label: "Synthetic GJC status" }); + workspaces.push(result.workspace.workspace_id); + const paneId = result.root_pane.pane_id; + await herdrRpc("pane.send_text", { pane_id: paneId, text: `cd ${quote(root)} && ${quote(process.execPath)} ${quote(script)} ${quote(transcript)}\n` }); + let foreground: { pid: number; argv?: string[] } | undefined; + await until(async () => { + const info = await herdrRpc<{ process_info?: { foreground_processes?: { pid: number; argv?: string[] }[] } }>("pane.process_info", { pane_id: paneId }); + foreground = info.process_info?.foreground_processes?.find((item) => item.argv?.includes(script)); + return foreground !== undefined; + }, "owned GJC process"); + assert.ok(foreground?.argv); + assert.ok(isGjcProcess(foreground.argv), "fixture must be recognized as GJC"); + const header: { id: string } = JSON.parse(readFileSync(transcript, "utf8").split("\n")[0] ?? ""); + sdkRows.push({ sessionId: header.id, pid: foreground.pid, hostIncarnation: "owned-fixture", + endpointGeneration: 1, identityProvenance: "composite", live: true, terminal: false, + terminalUncertain: false, ambiguous: false, + locator: { cwd: root, worktreeRoot: null, stateRoot: join(root, ".gjc", "agent") }, path: transcript }); + writeFileSync(sdkRegistry, JSON.stringify(sdkRows)); + await until(async () => await gjcTranscriptForPane(paneId, root, root) === transcript, "exact native transcript binding"); + await herdrRpc("pane.report_agent", { pane_id: paneId, source: "manual", agent: "gjc", state: "working" }); + await herdrRpc("pane.report_agent_session", { pane_id: paneId, source: "herdr:codex", agent: "codex", seq: Date.now(), agent_session_id: "01a00000-0000-7000-8000-000000000001" }); + await herdrRpc("pane.report_agent", { pane_id: paneId, source: "manual", agent: "gjc", state: "working" }); + return paneId; +} +async function rawWorking(paneId: string): Promise { + const pane = (await sessionSnapshot()).panes.find((item) => item.pane_id === paneId); + assert.ok(pane); + assert.equal(pane.agent, "gjc"); + assert.equal(pane.agent_status, "working", "raw provider signal must remain stale working"); +} +async function httpSnapshot(): Promise { + const response = await fetch(`${origin}/api/session`, { signal: AbortSignal.timeout(5_000) }); + assert.equal(response.status, 200); + const body: { snapshot: SessionSnapshot } = await response.json(); + return body.snapshot; +} +async function conversation(paneId: string): Promise { + const response = await fetch(`${origin}/api/pane/conversation?pane_id=${encodeURIComponent(paneId)}`, { signal: AbortSignal.timeout(5_000) }); + assert.equal(response.status, 200); + const body: { source: string } = await response.json(); + assert.equal(body.source, "gjc-transcript"); + return JSON.stringify(body); +} +function agentRow(paneId: string) { + assert.ok(page); + return page.locator(`.agents-sidebar .agent-item[data-machine="local"][data-pane="${paneId}"]`); +} +async function settled(paneId: string, status: Status, start?: number): Promise { + if (start !== undefined) await until(() => frames.slice(start).some((frame) => + frame.type === "pane-status" && frame.pane_id === paneId && frame.agent_status === status), `${scenario}: pushed ${status}`); + await until(async () => (await httpSnapshot()).panes.find((item) => item.pane_id === paneId)?.agent_status === status, `${scenario}: HTTP ${status}`); + await until(() => agentRow(paneId).locator(`.badge[data-status="${status}"]`).isVisible(), `${scenario}: browser ${status}`); + await rawWorking(paneId); +} +async function checkpoint(name: string): Promise { + if (evidenceDirectory && page) { + await page.screenshot({ path: join(evidenceDirectory, `gjc-status-${name}.png`), fullPage: true }); + await page.locator(".agents-sidebar").screenshot({ path: join(evidenceDirectory, `gjc-status-${name}-agents.png`) }); + } +} +async function staysDone(paneId: string, start: number): Promise { + const observedUntil = Date.now() + 2_000; + await until(async () => { + assert.equal((await httpSnapshot()).panes.find((item) => item.pane_id === paneId)?.agent_status, "done"); + assert.ok(await agentRow(paneId).locator('.badge[data-status="done"]').isVisible()); + assert.ok(!frames.slice(start).some((frame) => frame.type === "pane-status" && frame.pane_id === paneId && frame.agent_status !== "done"), "stale raw events must not push a native status regression"); + return Date.now() >= observedUntil; + }, "DONE remains stable while stale raw events are processed"); +} +function passed(): void { + outcomes.push({ scenario, outcome: "passed" }); + console.log(`PASS ${scenario}`); +} + +process.env.HERDR_SOCKET = herdrSocketPath(); +process.env.HOME = root; +try { + const sdkBin = join(root, "bin"); + mkdirSync(sdkBin); + const sdkCommand = join(sdkBin, "gjc"); + writeFileSync(sdkCommand, `#!${process.execPath} +import { readFileSync } from "node:fs"; +const rows = JSON.parse(readFileSync(${JSON.stringify(sdkRegistry)}, "utf8")); +const at = process.argv.indexOf("--json-input"); +const input = at < 0 ? {} : JSON.parse(process.argv[at + 1]); +const row = rows.find((candidate) => candidate.sessionId === input.resolveSessionId); +console.log(JSON.stringify({ type: "broker_response", ok: true, result: { + indexSeq: 1, sessions: rows, continuationCursor: null, + savedSession: row ? { id: row.sessionId, path: row.path } : null, +} })); +`); + chmodSync(sdkCommand, 0o755); + process.env.PATH = `${sdkBin}:${oldPath ?? ""}`; + mkdirSync(store, { recursive: true }); + if (evidenceDirectory) mkdirSync(evidenceDirectory, { recursive: true }); + writeFileSync(script, "import { openSync } from 'node:fs'; globalThis.held = openSync(process.argv[2], 'r'); console.log('Synthetic GJC fixture'); setInterval(() => {}, 1000);\n"); + const transcript = join(store, "main.jsonl"), delayedTranscript = join(store, "delayed.jsonl"); + const header = JSON.stringify({ type: "session", id: "fixture-native-session", cwd: root, timestamp: new Date().toISOString() }) + "\n"; + writeFileSync(transcript, header); + writeFileSync(delayedTranscript, header.replace("fixture-native-session", "fixture-delayed-session")); + const native = history(transcript), delayed = history(delayedTranscript); + native.append(native.message("user", "Synthetic initial request"), native.message("assistant", "Synthetic GJC completed answer")); + const paneId = await fixture(transcript), delayedPaneId = await fixture(delayedTranscript); + server = createServer({ port: 0, hostname: "127.0.0.1", token: "", stateDir: join(root, "state"), tailscaleOwner: null, usage: new UsageService(undefined, []) }); + origin = `http://127.0.0.1:${server.port}`; + statusSocket = new WebSocket(`ws://127.0.0.1:${server.port}/ws`); + statusSocket.addEventListener("message", (event) => { + try { + const frame: ServerMessage = JSON.parse(String(event.data)); + frames.push(frame); + } catch (error) { socketError = error instanceof Error ? error : new Error(String(error)); } + }); + statusSocket.addEventListener("error", () => { socketError = new Error("Status WebSocket failed"); }); + await until(() => frames.some((frame) => frame.type === "snapshot"), "initial WebSocket snapshot"); + browser = await chromium.launch({ executablePath: process.env.CHROME_PATH ?? "/opt/google/chrome/chrome", headless: true, args: ["--no-sandbox", "--accept-lang=en-US"] }); + const context = await browser.newContext({ viewport: { width: 1280, height: 800 } }); + await context.addInitScript((ids) => { + localStorage.setItem("herdr-web-ui:settings", JSON.stringify({ language: "en", quietOpenedDone: false })); + for (const id of ids) localStorage.setItem(`herdr-web-ui:view:${id}`, "chat"); + }, [paneId, delayedPaneId]); + page = await context.newPage(); + await page.goto(origin); + + scenario = "Initial completed history is READY without fabricated completion"; + await settled(paneId, "idle"); + assert.ok((await conversation(paneId)).includes("Synthetic GJC completed answer")); + await agentRow(paneId).locator(".agent-select").click(); + await page.getByText("Synthetic GJC completed answer", { exact: true }).waitFor({ state: "visible", timeout: 20_000 }); + assert.ok(!frames.some((frame) => frame.type === "pane-status" && frame.pane_id === paneId && frame.agent_status === "done")); + await checkpoint("initial-rest"); passed(); + + scenario = "Native user work is RUN despite stale provider status"; + let start = frames.length; + native.append(native.message("user", "Synthetic new work")); + await settled(paneId, "working", start); + await checkpoint("working"); passed(); + + scenario = "Native stop is DONE and stale raw idle/working cannot undo it"; + start = frames.length; + native.append(native.message("assistant", "Synthetic foreground answer")); + await settled(paneId, "done", start); + start = frames.length; + await herdrRpc("pane.report_agent", { pane_id: paneId, source: "manual", agent: "gjc", state: "idle" }); + // A provider report may keep herdr at RUN even after this raw idle report; native DONE must stand either way. + await staysDone(paneId, start); + start = frames.length; + await herdrRpc("pane.report_agent", { pane_id: paneId, source: "manual", agent: "gjc", state: "working" }); + await settled(paneId, "done"); + await staysDone(paneId, start); passed(); + + scenario = "Native async-result wakes RUN and a final stop returns DONE"; + start = frames.length; native.append(native.wake()); + await settled(paneId, "working", start); + start = frames.length; native.append(native.message("assistant", "Synthetic background completed answer")); + await settled(paneId, "done", start); passed(); + + scenario = "Native ask is INPUT through narration, then tool result resumes RUN"; + start = frames.length; + native.append(native.message("user", "Synthetic question request"), native.ask()); + await settled(paneId, "blocked", start); + await checkpoint("input"); + native.append(native.message("assistant", "Synthetic unrelated narration", "toolUse")); + await until(async () => (await conversation(paneId)).includes("Synthetic unrelated narration"), "native narration read"); + await settled(paneId, "blocked"); + start = frames.length; native.append(native.message("toolResult", "Synthetic Continue result")); + await settled(paneId, "working", start); + start = frames.length; native.append(native.message("assistant", "Synthetic choice completed answer")); + await settled(paneId, "done", start); passed(); + + scenario = "Delayed native readiness pushes DONE from raw RUN"; + await settled(delayedPaneId, "working"); + await agentRow(delayedPaneId).locator(".agent-select").click(); + start = frames.length; + delayed.append(delayed.message("user", "Synthetic delayed request"), delayed.message("assistant", "Synthetic delayed completed answer")); + await settled(delayedPaneId, "done", start); + assert.ok((await conversation(delayedPaneId)).includes("Synthetic delayed completed answer")); + await page.getByText("Synthetic delayed completed answer", { exact: true }).waitFor({ state: "visible", timeout: 20_000 }); + await checkpoint("final-done"); passed(); +} catch (error) { + try { await checkpoint("failure"); } catch (captureError) { console.error("GJC screenshot failed", captureError); } + throw error; +} finally { + try { + if (evidenceDirectory) writeFileSync(join(evidenceDirectory, "gjc-status-summary.json"), JSON.stringify({ scenarios: outcomes, failedScenario: outcomes.length === 6 ? null : scenario }, null, 2) + "\n"); + } finally { + statusSocket?.close(); + try { await browser?.close(); } finally { + try { await server?.stop(); } finally { + try { + const cleanup = await Promise.allSettled(workspaces.map((id) => workspaceClose(id))); + const failures = cleanup.flatMap((result) => result.status === "rejected" ? [result.reason] : []); + if (failures.length) { + deferredCleanupError = new AggregateError(failures, "Owned GJC workspace cleanup failed"); + console.error(deferredCleanupError); + } + } finally { + if (oldHome === undefined) delete process.env.HOME; else process.env.HOME = oldHome; + if (oldSocket === undefined) delete process.env.HERDR_SOCKET; else process.env.HERDR_SOCKET = oldSocket; + if (oldPath === undefined) delete process.env.PATH; else process.env.PATH = oldPath; + rmSync(root, { recursive: true, force: true }); + } + } + } + } +} +if (deferredCleanupError) throw deferredCleanupError; diff --git a/server/gjc-status.test.ts b/server/gjc-status.test.ts new file mode 100644 index 0000000000..bd20c241c0 --- /dev/null +++ b/server/gjc-status.test.ts @@ -0,0 +1,707 @@ +import { afterAll, describe, expect, it } from "bun:test"; +import { appendFileSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import type { AgentStatus, HerdrPane, SessionSnapshot } from "../shared/protocol.ts"; +import { CompletionTracker, type OmoRest } from "./completion.ts"; +import { createGjcStatus, gjcPanes, gjcTurnAfter, readOmoSnapshot, reconcileProviderStatus } from "./gjc-status.ts"; +import { noTurn, OMO_ALIASES, OmoStatus, type OmoLine, type OmoPane, type OmoTurn } from "./omo-status.ts"; + +// records as gjc's native transcript writes them: a message envelope with the pi family's roles +const message = (role: string, stopReason?: string, at = "2026-10-02T00:00:10.000Z", text = "") => JSON.stringify({ type: "message", id: "x", parentId: null, timestamp: at, message: { role, content: [{ type: "text", text }], ...(stopReason ? { stopReason } : {}) } }); +// a finished background job's wake, which gjc does not always display +const asyncResult = (display = false) => JSON.stringify({ type: "custom_message", customType: "async-result", display, content: "…", timestamp: "2026-10-02T00:00:20.000Z" }); +const bookkeeping = (customType: string) => JSON.stringify({ type: "custom", customType }); +const lines = (...entries: string[]) => entries.join("\n") + "\n"; +const status = (turn: OmoTurn, entry: string) => gjcTurnAfter(turn, { text: entry }); + +const root = mkdtempSync(join(tmpdir(), "herdr-gjc-status-")); +afterAll(() => rmSync(root, { recursive: true, force: true })); + +describe("a GJC turn, read from its transcript", () => { + it("runs from a prompt through its tool calls, and is over at an answer that stopped for good", () => { + const reduce = (...entries: string[]) => entries.reduce(status, noTurn()); + expect(reduce(message("user")).status).toBe("working"); + expect(reduce(message("user"), message("assistant", "toolUse")).status).toBe("working"); + expect(reduce(message("user"), message("assistant", "toolUse"), message("toolResult")).status).toBe("working"); + expect(reduce(message("user"), message("assistant", "toolUse"), message("toolResult"), message("assistant", "stop")).status).toBe("idle"); + expect(reduce(message("user"), message("assistant", "length")).status).toBe("idle"); + // the record's timestamp says when the turn's last word was written + expect(reduce(message("user", undefined, "2026-10-02T00:01:00.000Z")).at).toBe(Date.parse("2026-10-02T00:01:00.000Z")); + }); + + it("is at rest after an error or an abort: no finish (#687)", () => { + const reduce = (...entries: string[]) => entries.reduce(status, noTurn()); + expect(reduce(message("user"), message("assistant", "error"))).toMatchObject({ status: "idle", failed: true }); + expect(reduce(message("user"), message("assistant", "aborted"))).toMatchObject({ status: "idle", failed: true }); + // a retry that got through finishes the turn after all + const answered = reduce(message("user"), message("assistant", "error"), message("assistant", "stop")); + expect([answered.status, answered.failed ?? null]).toEqual(["idle", null]); + expect(reduce(message("user"), message("assistant", "error"), message("assistant", "toolUse")).status).toBe("working"); + }); + + it("keeps the turn an assistant message with a missing or unknown stopReason cannot decide", () => { + const reduce = (turn: OmoTurn, ...entries: string[]) => entries.reduce(status, turn); + const unknown = JSON.stringify({ type: "message", timestamp: "2026-10-02T00:00:30.000Z", message: { role: "assistant", content: [], stopReason: "suspended" } }); + const bare = JSON.stringify({ type: "message", timestamp: "2026-10-02T00:00:30.000Z", message: { role: "assistant", content: [] } }); + expect(reduce(noTurn(), unknown, bare).status).toBeNull(); + expect(reduce({ status: "working", at: null, asks: [] }, unknown, bare).status).toBe("working"); + expect(reduce({ status: "idle", at: null, asks: [] }, unknown, bare).status).toBe("idle"); + }); + + it("starts a turn on a background job's result, displayed or not, and on no other custom record", () => { + const reduce = (...entries: string[]) => entries.reduce(status, noTurn()); + const rested = [message("user"), message("assistant", "stop")]; + expect(reduce(...rested, asyncResult()).status).toBe("working"); + expect(reduce(...rested, asyncResult(true)).status).toBe("working"); + // the runtime's own bookkeeping and other agents' wake names are no gjc turn + for (const note of [bookkeeping("gjc.model_change"), bookkeeping("senpi.hooks.stop-state"), + JSON.stringify({ type: "custom_message", customType: "omo-senpi:wake", display: false }), + JSON.stringify({ type: "custom_message", customType: "async-task", display: true }), + JSON.stringify({ type: "custom_message", content: "no type named" }), + JSON.stringify({ type: "session", cwd: "/work" }), "not json"]) { + expect(reduce(...rested, note).status).toBe("idle"); + } + expect(reduce(bookkeeping("x"), "not json").status).toBeNull(); + }); + + it("starts a turn on the monitor's and the timer's hidden wakes too, and on no text that names them", () => { + const reduce = (...entries: string[]) => entries.reduce(status, noTurn()); + const rested = [message("user"), message("assistant", "stop")]; + const wake = (customType: string) => JSON.stringify({ type: "custom_message", customType, display: false, triggerTurn: true, deliverAs: "followUp", content: "…", timestamp: "2026-10-02T00:00:20.000Z" }); + for (const customType of ["async-result", "task-notification", "cron-fire"]) { + expect(reduce(...rested, wake(customType)).status).toBe("working"); + } + // a wake's name inside another record's text is no wake + expect(reduce(...rested, message("assistant", "stop", "2026-10-02T00:00:30.000Z", "the async-result arrived")).status).toBe("idle"); + expect(reduce(...rested, bookkeeping("task-notification")).status).toBe("idle"); + expect(reduce(...rested, JSON.stringify({ type: "custom_message", customType: "task-notificationed" })).status).toBe("idle"); + }); + + it("rejects a record that is not an object, and never throws on one", () => { + for (const entry of ["null", "[null]", `[${message("user")}]`, "42", "\"async-result\"", "not json", "{"]) { + expect(() => gjcTurnAfter(noTurn(), { text: entry })).not.toThrow(); + expect(gjcTurnAfter(noTurn(), { text: entry }).status).toBeNull(); + expect(gjcTurnAfter({ status: "working", at: 1, asks: [] }, { text: entry })).toMatchObject({ status: "working", at: 1 }); + } + }); + + it("holds a question the agent asked open until its result, an error result too", () => { + const asking = (id: string, extra: Record = {}) => JSON.stringify({ type: "message", timestamp: "2026-10-02T00:00:30.000Z", message: { role: "assistant", stopReason: "toolUse", content: [{ type: "toolCall", id, name: "ask", arguments: { question: "Which way?" }, ...extra }] } }); + const result = (id: string, isError = false) => JSON.stringify({ type: "message", timestamp: "2026-10-02T00:00:40.000Z", message: { role: "toolResult", toolCallId: id, toolName: "ask", content: [], isError } }); + const reduce = (...entries: string[]) => entries.reduce(status, noTurn()); + const asked = reduce(message("user"), asking("q1")); + expect(asked).toMatchObject({ status: "working" }); + expect(asked.asks.map((call) => call.id)).toEqual(["q1"]); + // unrelated assistant narration leaves the question open + expect(reduce(message("user"), asking("q1"), message("assistant", "toolUse", "2026-10-02T00:00:35.000Z", "thinking")).asks).toHaveLength(1); + // its result resumes the turn; another call's result does not + expect(reduce(message("user"), asking("q1"), result("other")).asks).toHaveLength(1); + expect(reduce(message("user"), asking("q1"), result("q1")).asks).toEqual([]); + // an error result ends the asking the same, and a call with no usable id opens none + expect(reduce(message("user"), asking("q2"), result("q2", true)).asks).toEqual([]); + expect(reduce(message("user"), asking(""), asking("q3", { incomplete: true })).asks).toEqual([]); + for (const stopped of ["stop", "error", "aborted"]) { + expect(reduce(message("user"), asking("unexecuted").replace('"stopReason":"toolUse"', `"stopReason":"${stopped}"`)).asks).toEqual([]); + } + }); + + it("reads a record of any size by its projected JSON, and only whole lines", () => { + const big = "x".repeat(200_000); + const long = message("assistant", "toolUse").replace('"content":[', `"content":[{"type":"thinking","thinking":"${big}"},`); + const ends: OmoLine = { head: long.slice(0, 4096), tail: long.slice(-4096), record: long }; + expect(gjcTurnAfter(noTurn(), ends).status).toBe("working"); + // the projection truncates long strings but stays complete JSON + const projected = JSON.parse(long.replace(big, big.slice(0, 64 * 1024))); + expect(gjcTurnAfter(noTurn(), { head: "", tail: "", record: JSON.stringify(projected) }).status).toBe("working"); + // ends without a projected record say nothing (no regex on text) + expect(gjcTurnAfter(noTurn(), { head: long.slice(0, 4096), tail: long.slice(-4096) }).status).toBeNull(); + expect(gjcTurnAfter({ status: "working", at: null, asks: [] }, { head: message("assistant", "stop").slice(0, 40), tail: message("assistant", "stop").slice(-40) }).status).toBe("working"); + }); +}); + +describe("a GJC pane's status in place of herdr's provider-child status", () => { + const pane = (id: string, agent: string | null, status: AgentStatus): HerdrPane => ({ pane_id: id, agent, agent_status: status, cwd: "/work" } as unknown as HerdrPane); + const snapshotOf = (...panes: HerdrPane[]): SessionSnapshot => ({ panes, agents: panes.map((p) => ({ pane_id: p.pane_id, agent: p.agent, agent_status: p.agent_status })) } as unknown as SessionSnapshot); + /** herdr follows the provider child: gjc's parent turn can be over while the pane reads working */ + const herdr = () => snapshotOf(pane("gjc", "gjc", "working"), pane("lost", "gjc", "working"), pane("shell", null, "unknown")); + const FILE = "/s/gjc-session.jsonl"; + + function setup(initial = lines(message("user"), message("assistant", "stop"))) { + const files: Record = { [FILE]: initial }; + const ids: Record = {}; + const told: ([string, AgentStatus, number, boolean] | [string, AgentStatus, number, boolean, OmoRest])[] = []; + const found: string[] = []; + const state = { discovered: new Map([["gjc", { path: FILE, startedAt: null }], ["lost", { path: null, startedAt: null }]]), lookups: 0, clock: 0, failing: false, during: () => {} }; + const gjc = createGjcStatus({ + discover: async () => { state.lookups += 1; state.during(); return new Map(state.discovered); }, + snapshot: async () => herdr(), + onChange: (id, status, count, turn, rest) => told.push(rest ? [id, status, count, turn, rest] : [id, status, count, turn]), + onFound: (id) => found.push(id), + file: { + stat: (path) => path in files ? { size: Buffer.byteLength(files[path]!), id: ids[path] ?? "1" } : null, + lines: (path, from, size, each) => { if (state.failing) return from; const text = Buffer.from(files[path]!).subarray(from, size).toString("utf8"); const whole = text.slice(0, text.lastIndexOf("\n") + 1); for (const line of whole.split("\n").slice(0, -1)) each({ text: line }); return from + Buffer.byteLength(whole); }, + }, + now: () => state.clock, + }); + const append = (...entries: string[]) => { files[FILE] += lines(...entries); }; + /** the file put anew in place of the old one, as a rewrite does */ + const replace = (...entries: string[]) => { files[FILE] = lines(...entries); ids[FILE] = `${Number(ids[FILE] ?? 1) + 1}`; }; + return { gjc, told, found, append, replace, state, files }; + } + const statuses = (snapshot: SessionSnapshot) => Object.fromEntries(snapshot.panes.map((p) => [p.pane_id, `${p.agent}/${p.agent_status}`])); + + it("reads READY where herdr's provider child still reads working, and no finish from a first idle baseline", async () => { + const { gjc, told, found, state } = setup(); + const completions = new CompletionTracker(null); + const raw = herdr(); + await gjc.refresh(raw.panes); + // the transcript's final answer is the truth: gjc's turn is over whatever the provider child does + expect(statuses(gjc.apply(raw))).toEqual({ gjc: "gjc/idle", lost: "gjc/working", shell: "null/unknown" }); + expect([gjc.tracks("gjc"), gjc.tracks("lost"), gjc.runs("lost"), gjc.runs("shell")]).toEqual([true, false, true, false]); + expect(found.sort()).toEqual(["gjc", "lost"]); + for (const paneId of found) completions.adopt(paneId, "gjc", []); + // a pane first known at rest never works, so its idle is no finish and no alert + expect(told).toEqual([]); + expect(statuses(completions.present(gjc.apply(raw)))["gjc"]).toBe("gjc/idle"); + // the session that cannot be told keeps herdr's word under the same identity + expect(statuses(completions.present(gjc.apply(raw)))["lost"]).toBe("gjc/working"); + state.discovered.set("gjc", { path: null, startedAt: null }); + state.clock += 10_000; + await gjc.refresh(raw.panes); + state.discovered.set("gjc", { path: FILE, startedAt: null }); + state.clock += 10_000; + await gjc.refresh(raw.panes); + expect(told).toEqual([["gjc", "idle", 0, true]]); + }); + + it("reads RUN while a turn runs and DONE when it ends, in events and in every snapshot", async () => { + const { gjc, told, append, state } = setup(); + const completions = new CompletionTracker(null); + const served = () => completions.readSnapshot(async () => { const raw = herdr(); await gjc.refresh(raw.panes); return gjc.apply(raw); }); + await served(); + append(message("user")); + gjc.poll(); + expect(told).toEqual([["gjc", "working", 0, true]]); + expect(completions.observe("gjc", "working", "gjc")).toBe("working"); + for (let i = 0; i < 3; i++) { state.clock += 10_000; expect(statuses(await served())["gjc"]).toBe("gjc/working"); } + append(message("assistant", "toolUse"), message("toolResult")); + gjc.poll(); + expect(told).toHaveLength(1); + append(message("assistant", "stop")); + gjc.poll(); + expect(told.at(-1)).toEqual(["gjc", "idle", 0, true]); + expect(completions.observe("gjc", "idle", "gjc")).toBe("done"); + for (let i = 0; i < 3; i++) { state.clock += 10_000; expect(statuses(await served())["gjc"]).toBe("gjc/done"); } + // a finished background job wakes the session: a turn of its own, though gjc displays nothing + append(asyncResult()); + gjc.poll(); + expect(completions.observe("gjc", told.at(-1)![1], "gjc")).toBe("working"); + expect(statuses(await served())["gjc"]).toBe("gjc/working"); + }); + + it("reads READY not DONE for a turn that ended in an error or an abort, and DONE once a retry answers", async () => { + const { gjc, told, append, state } = setup(lines(message("user"), message("assistant", "toolUse"), message("toolResult"))); + const completions = new CompletionTracker(null); + const served = () => completions.readSnapshot(async () => { const raw = herdr(); await gjc.refresh(raw.panes); return gjc.apply(raw); }); + await served(); + const shown: AgentStatus[] = []; + const settle = () => { for (const [id, status, , turn, rest] of told.splice(0)) if (id === "gjc" && turn) shown.push(completions.observe(id, status, "gjc", rest)); }; + append(message("assistant", "error", "2026-10-02T00:01:00.000Z")); + gjc.poll(); settle(); + expect(shown).toEqual(["working", "idle"]); + for (let i = 0; i < 3; i++) { state.clock += 10_000; expect(statuses(await served())["gjc"]).toBe("gjc/idle"); } + // a retry that gets an answer ends the turn after all: that is a finish, alerted as one + append(message("assistant", "stop", "2026-10-02T00:02:00.000Z")); + gjc.poll(); settle(); + expect(shown.slice(2)).toEqual(["done"]); + // an aborted prompt rests the same way; asked and aborted between two polls, only its rest is told + append(message("user"), message("assistant", "aborted", "2026-10-02T00:03:00.000Z")); + gjc.poll(); settle(); + expect(shown.slice(3)).toEqual(["idle"]); + expect(statuses(await served())["gjc"]).toBe("gjc/idle"); + }); + + it("leaves herdr's word alone while the transcript is missing, unreadable or holds no turn record", async () => { + const { gjc, files } = setup(); + const raw = herdr(); + await gjc.refresh(raw.panes); + expect(gjc.tracks("gjc")).toBe(true); + // the file vanished: the rest it last read is nobody's, and herdr's word stands for the pane at once + delete files[FILE]; + gjc.poll(); + expect(gjc.tracks("gjc")).toBe(false); + expect(statuses(gjc.apply(raw))["gjc"]).toBe("gjc/working"); + // a file that was never readable + const { gjc: blind, told: blindTold, state: blindState } = setup(lines("not json", bookkeeping("gjc.model_change"))); + blindState.discovered.set("gjc", { path: "/s/elsewhere.jsonl", startedAt: null }); + await blind.refresh(raw.panes); + expect([blind.tracks("gjc"), blind.runs("gjc")]).toEqual([false, true]); + expect(statuses(blind.apply(raw))["gjc"]).toBe("gjc/working"); + expect(blindTold).toEqual([]); + // a read that fails leaves herdr's `working` standing, not an idle nobody saw + blindState.discovered.set("gjc", { path: FILE, startedAt: null }); + blindState.failing = true; + blindState.clock += 10_000; + await blind.refresh(raw.panes); + expect(blind.tracks("gjc")).toBe(false); + expect(statuses(blind.apply(raw))["gjc"]).toBe("gjc/working"); + expect(blindTold).toEqual([]); + // an empty file has no turn record either: herdr's word stands + const { gjc: empty, state: emptyState } = setup(""); + emptyState.discovered.set("gjc", { path: FILE, startedAt: null }); + await empty.refresh(raw.panes); + expect(empty.tracks("gjc")).toBe(false); + expect(statuses(empty.apply(raw))["gjc"]).toBe("gjc/working"); + }); + + it("does not take an unfinished turn of a process that is gone for a running one", async () => { + // killed after a tool call, then resumed by a new process without a prompt + const { gjc, told, state, append } = setup(lines(message("user", undefined, "2026-10-02T00:00:10.000Z"), message("assistant", "toolUse", "2026-10-02T00:00:12.000Z"))); + state.discovered.set("gjc", { path: FILE, startedAt: Date.parse("2026-10-02T00:05:00.000Z") }); + const raw = herdr(); + await gjc.refresh(raw.panes); + expect(statuses(gjc.apply(raw))["gjc"]).toBe("gjc/idle"); + // a later lookup that cannot read the start keeps the one read before + state.discovered.set("gjc", { path: FILE, startedAt: null }); + state.clock += 10_000; + await gjc.refresh(raw.panes); + expect(statuses(gjc.apply(raw))["gjc"]).toBe("gjc/idle"); + // the interrupted turn is read as a failed rest of the history, never a successful finish + expect(told).toEqual([["gjc", "idle", 0, true, "failed-before"]]); + // a prompt to the new process is a turn + append(message("user", undefined, "2026-10-02T00:06:00.000Z")); + gjc.poll(); + expect(told).toEqual([["gjc", "idle", 0, true, "failed-before"], ["gjc", "working", 0, true]]); + }); + + it("reads a session the pane switched to from its own start, and lets go when another agent is named", async () => { + const { gjc, told, files, state } = setup(lines(message("user"))); + const raw = herdr(); + await gjc.refresh(raw.panes); + expect(told).toEqual([["gjc", "working", 0, true]]); + // /resume puts a finished session in the pane: its last answer is no new work + const other = "/s/gjc-other.jsonl"; + files[other] = lines(message("user"), message("assistant", "error", "2026-10-02T00:02:00.000Z")); + state.discovered.set("gjc", { path: other, startedAt: null }); + state.clock += 10_000; + await gjc.refresh(raw.panes); + // an error read from a file found anew is history: it says nothing of a DONE kept since + expect(told.at(-1)).toEqual(["gjc", "idle", 0, true, "failed-before"]); + expect(statuses(gjc.apply(raw))["gjc"]).toBe("gjc/idle"); + // Codex started in the pane: its events are no longer swallowed, and gjc lets go at once + expect(gjc.named("gjc", "codex")).toBe(true); + expect(gjc.named("gjc", "codex")).toBe(false); + expect([gjc.tracks("gjc"), gjc.runs("gjc")]).toEqual([false, false]); + expect(statuses(gjc.apply(raw))["gjc"]).toBe("gjc/working"); + // herdr's own name for gjc, and no name, change nothing + for (const name of [null, "gjc"]) expect(gjc.named("lost", name)).toBe(false); + }); + + it("reads again what a failed read left, and only whole lines of a record still being written", async () => { + const { gjc, told, append, files, state } = setup(lines(message("user"))); + await gjc.refresh(herdr().panes); + expect(told).toEqual([["gjc", "working", 0, true]]); + state.failing = true; + append(message("assistant", "toolUse")); + gjc.poll(); + expect(told).toHaveLength(1); + state.failing = false; + gjc.poll(); + // the turn was the same, but herdr's word stood for the pane meanwhile: its own is told again + expect(told).toHaveLength(2); + // a torn tail waits for its newline: the stop record half-written is not the turn's end yet + const stop = message("assistant", "stop", "2026-10-02T00:02:00.000Z"); + files[FILE] += stop.slice(0, 60); + gjc.poll(); + expect(told).toHaveLength(2); + files[FILE] += stop.slice(60) + "\n"; + gjc.poll(); + expect(told.at(-1)).toEqual(["gjc", "idle", 0, true]); + }); + + it("reads INPUT while the agent's question waits on the user, and RUN again once its result arrives", async () => { + const ask = JSON.stringify({ type: "message", timestamp: "2026-10-02T00:00:30.000Z", message: { role: "assistant", stopReason: "toolUse", content: [{ type: "toolCall", id: "q1", name: "ask", arguments: { question: "Which way?" } }] } }); + const answer = JSON.stringify({ type: "message", timestamp: "2026-10-02T00:00:40.000Z", message: { role: "toolResult", toolCallId: "q1", toolName: "ask", content: [] } }); + const { gjc, told, append } = setup(lines(message("user"))); + const raw = herdr(); + await gjc.refresh(raw.panes); + append(ask); + gjc.poll(); + expect(told.at(-1)).toEqual(["gjc", "blocked", 0, true]); + expect(statuses(gjc.apply(raw))["gjc"]).toBe("gjc/blocked"); + // unrelated narration leaves the question waiting + append(message("assistant", "toolUse", "2026-10-02T00:00:35.000Z", "thinking")); + gjc.poll(); + expect(told).toHaveLength(2); + expect(statuses(gjc.apply(raw))["gjc"]).toBe("gjc/blocked"); + append(answer); + gjc.poll(); + expect(told.at(-1)).toEqual(["gjc", "working", 0, true]); + append(message("assistant", "stop")); + gjc.poll(); + expect(told.at(-1)).toEqual(["gjc", "idle", 0, true]); + const resumed = setup(lines(ask)); + await resumed.gjc.refresh(raw.panes); + resumed.state.discovered.set("gjc", { path: FILE, startedAt: Date.parse("2026-10-02T00:05:00Z") }); + resumed.state.clock += 10_000; + await resumed.gjc.refresh(raw.panes); + expect(resumed.told.at(-1)).toEqual(["gjc", "idle", 0, true, "failed"]); + resumed.append(message("user", undefined, "2026-10-02T00:06:00Z")); + resumed.gjc.poll(); + expect(statuses(resumed.gjc.apply(raw))["gjc"]).toBe("gjc/working"); + }); + + it("tells the pane's own rest once its transcript first answers, so herdr's word does not stand", async () => { + const { gjc, told, append } = setup(""); + const completions = new CompletionTracker(null); + const raw = herdr(); + // herdr's provider-child word stands for the pane meanwhile + expect(completions.observe("gjc", "working", "gjc")).toBe("working"); + await gjc.refresh(raw.panes); + expect(gjc.tracks("gjc")).toBe(false); + expect(statuses(gjc.apply(raw))["gjc"]).toBe("gjc/working"); + expect(told).toEqual([]); + // a whole turn landed between two polls: the pane's own rest is told though nothing moved since + append(message("user"), message("assistant", "stop")); + gjc.poll(); + expect(told).toEqual([["gjc", "idle", 0, true]]); + expect(completions.observe("gjc", "idle", "gjc")).toBe("done"); + }); + + it("hands the pane back to herdr's word when the transcript cannot be read whole, and takes it back whole", async () => { + const { gjc, told, append, files, state } = setup(lines(message("user"), message("assistant", "stop"))); + const raw = herdr(); + await gjc.refresh(raw.panes); + expect(gjc.tracks("gjc")).toBe(true); + expect(statuses(gjc.apply(raw))["gjc"]).toBe("gjc/idle"); + // gone: herdr's working stands for the pane at once + delete files[FILE]; + gjc.poll(); + expect(gjc.tracks("gjc")).toBe(false); + expect(statuses(gjc.apply(raw))["gjc"]).toBe("gjc/working"); + // back, but unreadable: no better + const prompt = message("user", undefined, "2026-10-02T00:03:00.000Z") + "\n"; + files[FILE] = lines(message("user"), message("assistant", "stop")) + prompt.slice(0, 50); + state.failing = true; + gjc.poll(); + expect(gjc.tracks("gjc")).toBe(false); + expect(statuses(gjc.apply(raw))["gjc"]).toBe("gjc/working"); + // and a record still being written ends nothing either + state.failing = false; + gjc.poll(); + expect(gjc.tracks("gjc")).toBe(false); + // whole again, what it gained meanwhile is told as the turn it is + files[FILE] += prompt.slice(50); + gjc.poll(); + expect(told).toEqual([["gjc", "working", 0, true]]); + expect(gjc.tracks("gjc")).toBe(true); + append(message("assistant", "stop", "2026-10-02T00:04:00.000Z")); + gjc.poll(); + expect(told.at(-1)).toEqual(["gjc", "idle", 0, true]); + expect(statuses(gjc.apply(raw))["gjc"]).toBe("gjc/idle"); + }); + + it("tells a turn the process now holding the session could not have run as an interruption, never a finish", async () => { + const { gjc, told, state } = setup(lines(message("user", undefined, "2026-10-02T00:00:10.000Z"), message("assistant", "toolUse", "2026-10-02T00:00:12.000Z"))); + // process A ran the turn: it is live work while it holds the session + state.discovered.set("gjc", { path: FILE, startedAt: Date.parse("2026-10-02T00:00:00.000Z") }); + const completions = new CompletionTracker(null); + const raw = herdr(); + await gjc.refresh(raw.panes); + const settle = () => told.splice(0).map(([id, status, , , rest]) => completions.observe(id, status, "gjc", rest)); + expect(settle()).toEqual(["working"]); + expect(statuses(completions.present(gjc.apply(raw)))["gjc"]).toBe("gjc/working"); + // the same file, now held by a process started after the turn's last word: interrupted, not finished + state.discovered.set("gjc", { path: FILE, startedAt: Date.parse("2026-10-02T00:05:00.000Z") }); + state.clock += 10_000; + await gjc.refresh(raw.panes); + expect(settle()).toEqual(["idle"]); + expect(statuses(completions.present(gjc.apply(raw)))["gjc"]).toBe("gjc/idle"); + expect(completions.current("gjc")).toBe("idle"); + }); + + it("reads a real transcript file, whole lines at a time and again from where it stopped", async () => { + const path = join(root, "gjc-session.jsonl"); + writeFileSync(path, lines(message("user"), message("assistant", "toolUse", "2026-10-02T00:00:12.000Z", "x".repeat(200_000)))); + const told: [string, AgentStatus][] = []; + // no file injection: the shared bounded reader feeds gjc's reducer + const gjc = createGjcStatus({ + discover: async () => new Map([["gjc", { path, startedAt: null }]]), + snapshot: async () => herdr(), + onChange: (id, status) => told.push([id, status]), + }); + const raw = herdr(); + await gjc.refresh(raw.panes); + expect(statuses(gjc.apply(raw))["gjc"]).toBe("gjc/working"); + appendFileSync(path, lines(message("toolResult"), asyncResult(true), message("assistant", "stop", "2026-10-02T00:02:00.000Z"))); + gjc.poll(); + expect(told).toEqual([["gjc", "working"], ["gjc", "idle"]]); + expect(statuses(gjc.apply(raw))["gjc"]).toBe("gjc/idle"); + }); + + it("keeps the shared tracker's OmO behavior when no GJC options are given", async () => { + // a default OmoStatus: `omo` identity and aliases, the OmO reducer, an empty file still authoritative + const omoFiles: Record = { [FILE]: "" }; + const told: [string, AgentStatus][] = []; + const omo = new OmoStatus({ + discover: async () => new Map([["gjc", { path: FILE, startedAt: null }]]), + snapshot: async () => herdr(), + onChange: (id, status) => told.push([id, status]), + file: { + stat: (path) => path in omoFiles ? { size: Buffer.byteLength(omoFiles[path]!), id: "1" } : null, + lines: (_path, from) => from, + }, + }); + const raw = snapshotOf(pane("gjc", "claude", "idle")); + await omo.refresh(raw.panes); + expect(omo.tracks("gjc")).toBe(true); + expect(statuses(omo.apply(raw))["gjc"]).toBe("omo/idle"); + expect(omo.named("gjc", "claude")).toBe(false); + expect(omo.named("gjc", "codex")).toBe(true); + expect(told).toEqual([]); + }); +}); + +describe("native OmO ownership alongside GJC", () => { + function setup(native = true) { + const cwd = mkdtempSync(join(root, "ownership-")); + const omoPath = join(cwd, "2026-10-02_native-omo-session.jsonl"); + const gjcPath = join(cwd, "gjc.jsonl"); + writeFileSync(omoPath, lines(message("user"))); + writeFileSync(gjcPath, lines(message("user"), message("assistant", "stop"))); + const state = { native, known: true, clock: 0, duringOmo: async () => {}, duringGjc: async () => {} }; + const pane = (paneId: string): HerdrPane => ({ pane_id: paneId, agent: "gjc", agent_status: "working", cwd, focused: false, revision: 0, tab_id: "t", terminal_id: paneId, workspace_id: "w" }); + const raw: SessionSnapshot = { panes: [pane("wrapped"), pane("standalone")], agents: [pane("wrapped"), pane("standalone")], layouts: [], tabs: [], workspaces: [], protocol: 22, version: "test" }; + const completions = new CompletionTracker(null); + const adopted: string[] = []; + const turns: [string, AgentStatus][] = []; + const candidates: string[][] = []; + const omo = new OmoStatus({ + discover: async () => { await state.duringOmo(); return state.native ? new Map([["wrapped", { path: state.known ? omoPath : null, startedAt: null }]]) : new Map(); }, + snapshot: async () => raw, + onFound: (id) => completions.adopt(id, "omo", OMO_ALIASES), + onChange: (id, status, _background, turn, rest) => { if (turn) completions.observe(id, status, "omo", rest); }, + background: () => new Map([["native-omo-session", 3]]), + now: () => state.clock, + }); + const snapshot = () => readOmoSnapshot(async () => raw, omo); + const gjc = createGjcStatus({ + snapshot, + discover: async (panes) => { + candidates.push(panes.map((pane) => pane.pane_id)); + const found = await gjcPanes(panes, { processInfo: async () => null, transcript: async () => gjcPath, underShell: async () => null }); + await state.duringGjc(); + return found; + }, + onFound: (id) => { adopted.push(id); completions.adopt(id, "gjc", []); }, + onChange: (id, status, _background, turn, rest) => { turns.push([id, status]); if (turn) completions.observe(id, status, "gjc", rest); }, + now: () => state.clock, + }, omo); + const served = () => completions.readSnapshot(async () => { + const named = await snapshot(); + await gjc.refresh(named.panes); + return gjc.apply(omo.apply(named)); + }); + return { omo, gjc, state, raw, snapshot, served, adopted, turns, candidates, completions, omoPath, gjcPath }; + } + + it("preserves native status and background work when the raw pane is gjc", async () => { + const f = setup(); + const shown = await f.served(); + expect(shown.panes.find((pane) => pane.pane_id === "wrapped")).toMatchObject({ agent: "omo", agent_status: "working" }); + expect(shown.agents.find((pane) => pane.pane_id === "wrapped")).toMatchObject({ agent: "omo", agent_status: "working" }); + expect(f.omo.backgroundOf("wrapped")).toBe(3); + expect(f.gjc.runs("wrapped")).toBe(false); + expect(f.adopted).toEqual(["standalone"]); + expect(f.turns).toEqual([]); + expect(shown.panes.find((pane) => pane.pane_id === "standalone")).toMatchObject({ agent: "gjc", agent_status: "idle" }); + expect(f.gjc.tracks("standalone")).toBe(true); + }); + + it("uses native ownership when periodic discovery invokes the shared snapshot dependency", async () => { + const f = setup(); + let release = () => {}; + let entered = () => {}; + const discovered = new Promise((resolve) => { release = resolve; }); + const discovering = new Promise((resolve) => { entered = resolve; }); + f.state.duringOmo = async () => { entered(); await discovered; }; + const periodic = f.snapshot().then((snapshot) => f.gjc.refresh(snapshot.panes)); + await discovering; + await Promise.resolve(); + expect(f.candidates).toEqual([]); + release(); + await periodic; + expect(f.candidates).toEqual([["standalone"]]); + expect(f.adopted).toEqual(["standalone"]); + expect(f.turns).toEqual([]); + expect(f.omo.runs("wrapped")).toBe(true); + expect(f.omo.backgroundOf("wrapped")).toBe(3); + }); + + it("hands ownership back to GJC when native discovery loses OmO", async () => { + const f = setup(); + await f.served(); + f.state.native = false; + f.state.clock += 10_000; + const shown = await f.served(); + expect(f.omo.runs("wrapped")).toBe(false); + expect(f.gjc.tracks("wrapped")).toBe(true); + expect(f.adopted).toEqual(["standalone", "wrapped"]); + expect(shown.panes.find((pane) => pane.pane_id === "wrapped")).toMatchObject({ agent: "gjc", agent_status: "idle" }); + appendFileSync(f.gjcPath, lines(message("user"))); + f.gjc.poll(); + expect(f.turns).toContainEqual(["wrapped", "working"]); + }); + + it("keeps the new GJC turn when a raw event clears inactive OmO history", async () => { + const f = setup(); + await f.served(); + f.state.native = false; + f.state.clock += 10_000; + await f.served(); + appendFileSync(f.gjcPath, lines(message("user"))); + f.gjc.poll(); + if (reconcileProviderStatus(f.omo, f.gjc, { pane_id: "wrapped", agent: "gjc" })) f.completions.forget("wrapped"); + expect(f.gjc.tracks("wrapped")).toBe(true); + expect(f.completions.current("wrapped")).toBe("working"); + }); + + it("rejects a stale discovery result when native OmO is found during the GJC lookup", async () => { + const f = setup(false); + f.state.duringGjc = async () => { f.state.native = true; await f.omo.refresh(f.raw.panes); }; + await f.gjc.refresh(f.raw.panes); + expect(f.adopted).toEqual(["standalone"]); + expect(f.turns).toEqual([]); + expect(f.completions.current("wrapped")).toBe("working"); + expect(f.gjc.runs("wrapped")).toBe(false); + expect(f.omo.backgroundOf("wrapped")).toBe(3); + }); + + it("suppresses a stale GJC poll after native OmO takes its pane", async () => { + const f = setup(false); + await f.served(); + f.state.native = true; + f.state.clock += 10_000; + await f.omo.refresh(f.raw.panes); + appendFileSync(f.omoPath, lines(message("assistant", "stop"))); + f.omo.poll(); + appendFileSync(f.gjcPath, lines(message("user"))); + f.gjc.poll(); + expect(f.turns).toEqual([["standalone", "working"]]); + expect(f.completions.current("wrapped")).toBe("done"); + expect(f.omo.backgroundOf("wrapped")).toBe(3); + }); + + it.each(["gjc", "codex", "devin", "unknown", null])("preserves native ownership when a stale %s provider event arrives", async (agent) => { + const f = setup(false); + await f.served(); + f.state.native = true; + f.state.clock += 10_000; + await f.omo.refresh(f.raw.panes); + appendFileSync(f.omoPath, lines(message("assistant", "stop"))); + f.omo.poll(); + if (reconcileProviderStatus(f.omo, f.gjc, { pane_id: "wrapped", agent })) f.completions.forget("wrapped"); + expect(f.omo.runs("wrapped")).toBe(true); + expect(f.gjc.runs("wrapped")).toBe(false); + expect(f.completions.current("wrapped")).toBe("done"); + expect(f.omo.backgroundOf("wrapped")).toBe(3); + }); + + it("keeps native ownership with an unknown session while raw status remains authoritative", async () => { + const f = setup(); + f.state.known = false; + await f.served(); + if (reconcileProviderStatus(f.omo, f.gjc, { pane_id: "wrapped", agent: "codex" })) f.completions.forget("wrapped"); + const shown = await f.served(); + expect(f.omo.runs("wrapped")).toBe(true); + expect(f.omo.tracks("wrapped")).toBe(false); + expect(f.adopted).toEqual(["standalone"]); + expect(shown.panes.find((pane) => pane.pane_id === "wrapped")).toMatchObject({ agent: "omo", agent_status: "working" }); + }); + + it("releases standalone GJC when another provider is named", async () => { + const f = setup(); + await f.served(); + expect(reconcileProviderStatus(f.omo, f.gjc, { pane_id: "standalone", agent: "codex" })).toBe(true); + expect(f.gjc.runs("standalone")).toBe(false); + }); + + it("returns to raw unknown status when native OmO disappears without another agent", async () => { + const f = setup(); + await f.served(); + f.state.native = false; + f.state.clock += 10_000; + for (const pane of [...f.raw.panes, ...f.raw.agents]) { + if (pane.pane_id === "wrapped") { pane.agent = null; pane.agent_status = "unknown"; } + } + const shown = await f.served(); + expect(f.omo.runs("wrapped")).toBe(false); + expect(f.gjc.runs("wrapped")).toBe(false); + expect(shown.panes.find((pane) => pane.pane_id === "wrapped")).toMatchObject({ agent: null, agent_status: "unknown" }); + }); +}); + +describe("the gjc process a pane's transcript and start are read from", () => { + const gjcPane = (id: string, extra: Record = {}): HerdrPane => + ({ pane_id: id, agent: "gjc", agent_status: "working", cwd: "/work", ...extra } as unknown as HerdrPane); + const transcript = async (id: string, _cwd: string) => `/t/${id}.jsonl`; + + it("binds the pane's current runtime cwd, not where it started", async () => { + const asked: string[] = []; + const panes = await gjcPanes([gjcPane("p", { foreground_cwd: "/work/now" }), gjcPane("q")], { + processInfo: async () => null, + transcript: async (_id, cwd) => { asked.push(cwd); return null; }, + underShell: async () => null, + }); + expect(asked.sort()).toEqual(["/work", "/work/now"]); + // a transcript nobody could tell is still the pane's own record: herdr's word stands for it + expect([...panes.values()]).toEqual([{ path: null, startedAt: null }, { path: null, startedAt: null }]); + }); + + it("takes the native start from the gjc in the foreground, and only there", async () => { + const seen: number[] = []; + const panes = await gjcPanes([gjcPane("p")], { + processInfo: async () => ({ process_info: { foreground_processes: [ + { pid: 11, argv: ["node", "/home/u/.gjc/gjc"] }, + { pid: 12, argv: ["/usr/bin/claude", "--session", "abc"] }, + ] } }), + transcript, + startedAt: (pid) => { seen.push(pid); return pid === 11 ? 1_000 : 2_000; }, + underShell: async () => null, + }); + expect(seen).toEqual([11]); + expect(panes.get("p")).toEqual({ path: "/t/p.jsonl", startedAt: 1_000 }); + }); + + it("takes the gjc under the pane's shell on Windows, with the start the table read for it", async () => { + const panes = await gjcPanes([gjcPane("p")], { + platform: "win32", + processInfo: async () => ({ process_info: { shell_pid: 50, foreground_processes: [{ pid: 50, argv: ["pwsh.exe"] }] } }), + transcript, + underShell: async (shellPid) => shellPid === 50 ? { pid: 77, started: 4_242 } : null, + }); + expect(panes.get("p")).toEqual({ path: "/t/p.jsonl", startedAt: 4_242 }); + }); + + it("says nothing of a start when the process table could not be read at all", async () => { + const panes = await gjcPanes([gjcPane("p")], { + processInfo: async () => null, + transcript, + underShell: async () => undefined, + }); + expect(panes.get("p")).toEqual({ path: "/t/p.jsonl", startedAt: null }); + }); + + it("takes no start from the provider child herdr follows", async () => { + // a foreground that holds only the provider child: its pid is never the session's start + const panes = await gjcPanes([gjcPane("p")], { + processInfo: async () => ({ process_info: { foreground_processes: [{ pid: 88, argv: ["/usr/bin/claude"] }] } }), + transcript, + startedAt: () => 9_999, + underShell: async () => null, + }); + expect(panes.get("p")).toEqual({ path: "/t/p.jsonl", startedAt: null }); + }); +}); diff --git a/server/gjc-status.ts b/server/gjc-status.ts new file mode 100644 index 0000000000..96cbe48437 --- /dev/null +++ b/server/gjc-status.ts @@ -0,0 +1,179 @@ +/** + * The status of a GJC pane, read from gjc's own transcript. + * + * herdr follows the pane's provider child, not the gjc parent: a turn gjc finished keeps + * reading `working` for what the child does, and the sidebar holds RUN with no done alert. + * The transcript the pane's gjc process holds (gjc-runtime.ts finds exactly that file, never a + * newest file by cwd) tells the parent turn instead: a user prompt, a tool result or an + * assistant message that stopped for a tool means work; one that stopped for good (`stop`, + * `length`) is rest, and one that errored or was aborted is rest after a failure — READY, not + * DONE (#687's rule holds here too). gjc wakes the agent with a custom message when work it + * asked for lands elsewhere — a background job's `async-result`, a monitor's + * `task-notification`, a timer's `cron-fire` — displayed or not: that begins a new parent + * turn. Everything else gjc writes (custom bookkeeping, model notices) is no turn's business. + * A tool call named `ask` holds a question for the user open until its result arrives, an + * error result included: while one is open the pane reads INPUT, whatever the turn does. + * + * The status replaces herdr's for the pane only once a turn record was actually read and the + * read is current: an empty, missing, unreadable or half-written transcript proves nothing, + * so herdr's `working` stands rather than a rest no one saw. The machinery is omo-status.ts's, + * composed with the `gjc` identity and this file's reducer; gjc keeps no background-task + * records to count. A turn the process holding the session now could not have run (it started + * after the record was written) is another process's unfinished work: an interruption, told as + * a failed rest, never a finish. + */ +import type { HerdrPane, SessionSnapshot } from "../shared/protocol.ts"; +import { herdrRpc } from "./herdr/client.ts"; +import { gjcPidUnderShell, gjcTranscriptForPane, isGjcProcess, type GjcProcess } from "./gjc-runtime.ts"; +import type { OmoAsks } from "./omo-ask.ts"; +import { processStartedAt } from "./process-start.ts"; +import { OmoStatus, type OmoLine, type OmoPane, type OmoStatusDeps, type OmoTurn } from "./omo-status.ts"; + +/** the custom messages gjc wakes the parent turn with, displayed or not */ +const WAKE_TYPES = new Set(["async-result", "task-notification", "cron-fire"]); + +interface GjcEntry { + type?: unknown; + customType?: unknown; + timestamp?: unknown; + message?: { role?: unknown; stopReason?: unknown; content?: unknown; toolCallId?: unknown }; +} + +/** + * The questions a gjc turn holds open for the user: an `ask` call opens one until the result + * for its id arrives — an error result ends the asking the same, since nothing waits on the + * user then. Assistant narration that carries no call leaves every question as it was. + */ +function gjcAsksAfter(open: OmoAsks, entry: GjcEntry): OmoAsks { + if (entry.type !== "message") return open; + const message = entry.message; + if (message?.role === "assistant" && message.stopReason === "toolUse") { + const calls = (Array.isArray(message.content) ? message.content as Record[] : []) + .filter((part) => part?.["type"] === "toolCall" && part["name"] === "ask" && typeof part["id"] === "string" && part["id"] !== "" && part["incomplete"] !== true) + .map((part) => ({ id: part["id"] as string, wait: true, args: part["arguments"] })); + const ids = new Set(calls.map((call) => call.id)); + return calls.length > 0 ? [...open.filter((call) => !ids.has(call.id)), ...calls] : open; + } + if (message?.role === "toolResult" && typeof message.toolCallId === "string") { + return open.filter((call) => call.id !== message.toolCallId); + } + return open; +} + +/** + * One transcript record applied to the parent turn so far, from a whole line or a projected + * record's complete JSON (never its text: a record too long to hold says nothing). An + * assistant message without a readable stopReason keeps the turn as it was; a record that is + * not an object (a bare value, a list) is no record at all. + */ +export function gjcTurnAfter(turn: OmoTurn, line: OmoLine): OmoTurn { + const text = "text" in line ? line.text : line.record; + if (text === undefined) return turn; + let entry: GjcEntry; + try { entry = JSON.parse(text); } catch { return turn; } + if (typeof entry !== "object" || entry === null || Array.isArray(entry)) return turn; + const parsed = typeof entry.timestamp === "string" ? Date.parse(entry.timestamp) : NaN; + const at = Number.isFinite(parsed) ? parsed : turn.at; + const asks = gjcAsksAfter(turn.asks, entry); + if (entry.type === "custom_message") { + return typeof entry.customType === "string" && WAKE_TYPES.has(entry.customType) ? { status: "working", at, asks } : { ...turn, asks }; + } + if (entry.type !== "message") return { ...turn, asks }; + const role = entry.message?.role; + const stopReason = entry.message?.stopReason; + if (role === "user" || role === "toolResult") return { status: "working", at, asks }; + if (role !== "assistant") return { ...turn, asks }; + if (stopReason === "toolUse") return { status: "working", at, asks }; + if (stopReason === "stop" || stopReason === "length") return { status: "idle", at, asks }; + if (stopReason === "error" || stopReason === "aborted") return { status: "idle", failed: true, at, asks }; + return { ...turn, asks }; +} + +type ProcessInfo = { process_info?: { shell_pid?: unknown; foreground_processes?: { pid?: unknown; argv?: unknown }[] } } | null; + +/** + * How `gjcPanes` finds out what it needs, each part unset meaning the real lookup: the pane's + * process info and its transcript (herdr's RPCs), when a pid started (process-start.ts), the + * gjc under a Windows pane's shell (gjc-runtime.ts), and the platform that decides between the + * two process identities. Tests give their own, so no live RPC is needed to tell the cases. + */ +export interface GjcDiscovery { + processInfo?: (paneId: string) => Promise; + transcript?: (paneId: string, cwd: string) => Promise; + startedAt?: (pid: number) => number | null; + underShell?: (shellPid: unknown) => Promise; + platform?: string; +} + +/** + * The panes of a snapshot herdr labels `gjc`, with the exact transcript each holds (null when + * it cannot be told: herdr's status stands for the pane) and when its native foreground gjc + * process started, so a turn a resumed session left unfinished is not taken for running work. + * On Windows herdr names only the pane's shell: the gjc under it, and its start, is the PC's + * process table's answer. `agent_session` is never consulted: it may name the provider child, + * not the gjc parent. The transcript is bound at the pane's current runtime cwd — what the + * foreground process runs in now, not where the pane started — the same cwd the chat binds by. + */ +export async function gjcPanes(panes: HerdrPane[], deps: GjcDiscovery = {}): Promise> { + const processInfo = deps.processInfo ?? ((paneId: string) => herdrRpc>("pane.process_info", { pane_id: paneId }).catch(() => null)); + const transcript = deps.transcript ?? ((paneId: string, cwd: string) => gjcTranscriptForPane(paneId, cwd).catch(() => null)); + const startedAt = deps.startedAt ?? processStartedAt; + const underShell = deps.underShell ?? ((shellPid: unknown) => gjcPidUnderShell(shellPid, deps.platform)); + const found = new Map(); + await Promise.all(panes.filter((pane) => pane.agent === "gjc").map(async (pane) => { + const info = await processInfo(pane.pane_id); + const starts = (info?.process_info?.foreground_processes ?? []) + .filter((process) => typeof process.pid === "number" && Array.isArray(process.argv) && isGjcProcess(process.argv.map(String))) + .map((process) => startedAt(process.pid as number)) + .filter((start): start is number => start !== null); + let started = starts.length > 0 ? Math.min(...starts) : null; + if (started === null) { + const native = await underShell(info?.process_info?.shell_pid).catch(() => undefined); + if (native) started = native.started; + } + const path = await transcript(pane.pane_id, pane.foreground_cwd || pane.cwd || ""); + found.set(pane.pane_id, { path, startedAt: started }); + })); + return found; +} + +/** + * The tracker's GJC composition: the `gjc` identity (herdr gives it no other name, so no + * aliases), this file's reducer, no background counts, an unfinished turn of a process that is + * gone told as the failed rest it is, and the readiness gate that releases a transcript + * nothing — or not everything — was read from back to herdr's status. + */ +export function createGjcStatus(deps: Omit, nativeOmo?: OmoStatus): OmoStatus { + return new OmoStatus({ + ...deps, + discover: async (panes) => { + const found = await deps.discover(panes.filter((pane) => !nativeOmo?.runs(pane.pane_id))); + // Native discovery may have finished while gjc's process lookup was in flight. + for (const paneId of found.keys()) if (nativeOmo?.runs(paneId)) found.delete(paneId); + return found; + }, + onFound: (paneId) => { if (!nativeOmo?.runs(paneId)) deps.onFound?.(paneId); }, + onChange: (...change) => { if (!nativeOmo?.runs(change[0])) deps.onChange(...change); }, + agent: "gjc", + aliases: [], + turnAfter: gjcTurnAfter, + ready: ({ turn, available, complete }) => turn.status !== null && available && complete, + background: () => new Map(), + staleIsFailure: true, + }); +} + +/** Shared by GJC's periodic discovery and the combined snapshot: native OmO is found first. */ +export async function readOmoSnapshot(snapshot: OmoStatusDeps["snapshot"], omo: OmoStatus): Promise { + const raw = await snapshot(); + await omo.refresh(raw.panes); + return omo.apply(raw); +} + +/** Provider labels cannot revoke native OmO; only discovery can. Discard any stale GJC hold. */ +export function reconcileProviderStatus(omo: OmoStatus, gjc: OmoStatus, pane: Pick): boolean { + const native = omo.runs(pane.pane_id); + const leftOmo = !native && omo.named(pane.pane_id, pane.agent ?? null); + const leftGjc = gjc.named(pane.pane_id, native ? "omo" : pane.agent ?? null); + return !native && (leftGjc || (leftOmo && !gjc.runs(pane.pane_id))); +} diff --git a/server/index.ts b/server/index.ts index aed07d8dc7..ab06a5f30d 100644 --- a/server/index.ts +++ b/server/index.ts @@ -25,6 +25,7 @@ import { OPENCODE_TOOL_REF } from "./opencode.ts"; import { DevinHistoryChanged } from "./devin.ts"; import { ConversationMonitor } from "./conversation-monitor.ts"; import { omoPanes } from "./omo.ts"; +import { createGjcStatus, gjcPanes, readOmoSnapshot, reconcileProviderStatus } from "./gjc-status.ts"; import { OMO_ALIASES, OmoStatus, processAlive } from "./omo-status.ts"; import { omoRuns, omoTasks } from "./omo-tasks.ts"; import { ClaudeSubagentStatus, claudeSubagents, within } from "./claude-subagents.ts"; @@ -1093,6 +1094,14 @@ export function createServer( // herdr called it `claude` or `pi` until now: what it finished under that name is its own onFound: (paneId) => completions.adopt(paneId, "omo", OMO_ALIASES), }); + const omoSnapshot = () => readOmoSnapshot(sessionSnapshot, omo); + /** GJC panes get their status from gjc's own transcript: herdr's `working` can be a provider child's (server/gjc-status.ts) */ + const gjc = createGjcStatus({ + discover: (panes) => gjcPanes(panes), + snapshot: omoSnapshot, + onChange: (paneId, derived, _background, turn, rest) => gjcChanged(paneId, derived, turn, rest), + onFound: (paneId) => completions.adopt(paneId, "gjc", []), + }, omo); /** Claude panes whose turn ended while work it started runs in the background (server/background-wait.ts) */ const waits = new BackgroundWait(options.statusNow ?? Date.now, options.backgroundWait); /** the pane whose status event is reading its background work now: that event sends the one frame and alert of it */ @@ -1115,11 +1124,11 @@ export function createServer( codexPath: async (pane) => codexOpenRollout(pane.pane_id, pane.cwd, await paneCodexHome(pane.pane_id, options.codexHome)), onChange: (paneId, summary) => planChanged(paneId, summary), }); - /** herdr's snapshot with OmO's own status in it: what the completion tracker and web push are given */ + /** herdr's snapshot with OmO's and GJC's own status in it: what the completion tracker and web push are given */ const rawSnapshot = async (): Promise => { - const snapshot = await sessionSnapshot(); - await omo.refresh(snapshot.panes); - const named = omo.apply(snapshot); + const snapshot = await omoSnapshot(); + await gjc.refresh(snapshot.panes); + const named = gjc.apply(omo.apply(snapshot)); // not waited for: finding a transcript costs process calls, and a snapshot must not wait on them. // The plans come after: a Claude pane's transcript is the one the subagent tracker found void claudeAgents.refresh(named.panes).catch(() => undefined).then(() => plans.refresh(named.panes)).catch(() => undefined); @@ -1606,6 +1615,19 @@ export function createServer( if (turn) push.onStatus(paneId, status).catch(logPushError); } + /** + * A GJC pane's own turn, which herdr's provider-child status never shows: back at work, its + * form has had its answer. gjc keeps no background records to count, so a change is always a + * turn, and an idle read after work is the finish — one ended in an error is not (#687). + */ + function gjcChanged(paneId: string, derived: AgentStatus, turn: boolean, rest: OmoRest | undefined): void { + if (turn && derived === "working") promptWaitEnded(paneId); + const status = turn ? completions.observe(paneId, derived, "gjc", rest) : completions.current(paneId) ?? completions.observe(paneId, derived, "gjc"); + if (turn) { pending.status(paneId, status); drainPending(paneId); } + broadcastAll(paneStatus(paneId, status)); + if (turn) push.onStatus(paneId, status).catch(logPushError); + } + /** * A Claude pane's subagents or background commands started or ended: no turn, so the status * stands. Nothing is alerted unless the pane began or stopped waiting on its turn's work at rest: @@ -1644,17 +1666,20 @@ export function createServer( const collector = startStatusCollector({ onStatus: (paneId, read, agent, replay) => { const raw = launchedAgents.settle(paneId, read); + gjc.poll(); // read back from a snapshot around a gap between subscriptions. An OmO pane's status there - // is OmO's own or herdr's by turns (server/omo-status.ts), and a difference is no change - if (replay && (omo.runs(paneId) || !completions.replayed(paneId, raw, replay))) return; - // another agent took an OmO pane: what OmO worked on there is not that agent's to finish - if (omo.named(paneId, agent)) completions.forget(paneId); + // is OmO's own or herdr's by turns (server/omo-status.ts), and a difference is no change; + // a GJC pane's may be a provider child's while gjc's own transcript is tracked + if (replay && (omo.runs(paneId) || gjc.tracks(paneId) || !completions.replayed(paneId, raw, replay))) return; + // Native OmO discovery owns the pane even when herdr names its provider child. + if (reconcileProviderStatus(omo, gjc, { pane_id: paneId, agent })) completions.forget(paneId); // the frame below names no agent: one herdr names anew is read into the roster now. An OmO // pane is `omo` in every snapshot, whatever herdr calls it in an event - const named = [{ pane_id: paneId, agent: omo.runs(paneId) ? "omo" : agent }]; + const named = [{ pane_id: paneId, agent: omo.runs(paneId) ? "omo" : gjc.runs(paneId) ? "gjc" : agent }]; if (machines) machines.localAgents(named); else bridgeAgents!.status(named); - // herdr says `claude/idle` for an OmO pane whatever it does: its own status stands, and no Claude turn waits there - if (omo.tracks(paneId)) { + // herdr says `claude/idle` for an OmO pane whatever it does, and a provider child's work for + // a GJC one while its transcript is tracked: their own status stands, and no Claude turn waits there + if (omo.tracks(paneId) || gjc.tracks(paneId)) { waits.forget(paneId); return; } @@ -1665,7 +1690,7 @@ export function createServer( if (raw === "working") promptWaitEnded(paneId); // an agent herdr lost on the way still works and finishes as such (server/completion.ts); // an OmO pane whose session is not known keeps herdr's status, under its own name - const status = completions.observe(paneId, raw, omo.runs(paneId) ? "omo" : agent); + const status = completions.observe(paneId, raw, omo.runs(paneId) ? "omo" : gjc.runs(paneId) ? "gjc" : agent); pending.status(paneId, status); drainPending(paneId); // a Claude turn that ended holds on its background work: what it started is read now, while // the pane is still taken for the status before (work that ended before the turn did holds @@ -1726,6 +1751,7 @@ export function createServer( onStructureChange: () => broadcastAll({ type: "session-changed" }), }, { snapshot: rawSnapshot }); omo.start(); + gjc.start(); claudeAgents.start(); const waitTimer = setInterval(waitsRanOut, 1000); waitTimer.unref?.(); @@ -3518,6 +3544,7 @@ export function createServer( clearInterval(outputTimer); collector.stop(); omo.stop(); + gjc.stop(); claudeAgents.stop(); clearInterval(waitTimer); plans.stop(); diff --git a/server/omo-status.ts b/server/omo-status.ts index 3aef577e3c..85495f2f32 100644 --- a/server/omo-status.ts +++ b/server/omo-status.ts @@ -158,6 +158,21 @@ export interface OmoStatusDeps { onChange: (paneId: string, status: AgentStatus, background: number, turn: boolean, rest?: OmoRest) => void; /** a pane was found to run OmO: what herdr called it until now is the same agent */ onFound?: (paneId: string) => void; + /** the identity the panes' status is reported under (`omo`), and the names herdr gave the agent before */ + agent?: string; + aliases?: readonly string[]; + /** a session record applied to the turn so far (`omoTurnAfter`): another agent's records have their own */ + turnAfter?: (turn: OmoTurn, line: OmoLine) => OmoTurn; + /** + * Whether the pane's own status may replace herdr's (default: always, once the session file + * is known). GJC withholds it until a turn record was read and the read is current: + * `available` says the file could be opened at the last poll, `complete` that everything it + * held was read to a record's end. An empty, gone or half-written transcript is nobody's + * rest, and an idle that cannot be shown is not the pane's. + */ + ready?: (tracked: { status: OmoPaneStatus; turn: OmoTurn; available: boolean; complete: boolean }) => boolean; + /** a turn the process holding the file now could not have run is an interruption told as a failed rest (GJC); default: a plain rest (OmO) */ + staleIsFailure?: boolean; file?: OmoFile; background?: (cwd: string) => Map; /** how often the session files are looked at, and the panes found anew */ @@ -166,7 +181,7 @@ export interface OmoStatusDeps { now?: () => number; } -interface Tracked extends OmoPane { cwd: string; offset: number; size: number; id: string; /** found again after a while herdr's status stood for it: a turn still running is told anew */ retell: boolean; turn: OmoTurn; status: OmoPaneStatus; /** at rest after an error, and when that error was written */ failed: boolean; failedAt: number | null; /** what is read now is the file's history, found anew: not written since it was being read */ history: boolean; background: number } +interface Tracked extends OmoPane { cwd: string; offset: number; size: number; id: string; /** found again after a while herdr's status stood for it: a turn still running is told anew */ retell: boolean; turn: OmoTurn; status: OmoPaneStatus; /** at rest after an error, and when that error was written */ failed: boolean; failedAt: number | null; /** what is read now is the file's history, found anew: not written since it was being read */ history: boolean; background: number; /** the file could be opened at the last poll, and all it held was read to a record's end */ available: boolean; complete: boolean; /** a poll has considered this pane: a baseline it then read is nobody's news */ polled: boolean; /** the pane's own status stood for it at the last poll */ authoritative: boolean } const FILES: OmoFile = { stat: (path) => { try { const fd = openSync(path, "r"); try { const stat = fstatSync(fd); return { size: stat.size, id: `${stat.dev}:${stat.ino}` }; } finally { closeSync(fd); } } catch { return null; } }, @@ -196,12 +211,22 @@ export class OmoStatus { private readonly background: NonNullable; private readonly now: () => number; private readonly refreshMs: number; + private readonly agent: string; + private readonly aliases: readonly string[]; + private readonly turnAfter: NonNullable; + private readonly ready: NonNullable; + private readonly staleIsFailure: boolean; constructor(private readonly deps: OmoStatusDeps) { this.file = deps.file ?? FILES; this.background = deps.background ?? omoBackgroundTasks; this.now = deps.now ?? Date.now; this.refreshMs = deps.refreshMs ?? 5000; + this.agent = deps.agent ?? "omo"; + this.aliases = deps.aliases ?? OMO_ALIASES; + this.turnAfter = deps.turnAfter ?? omoTurnAfter; + this.ready = deps.ready ?? (() => true); + this.staleIsFailure = deps.staleIsFailure ?? false; } /** Watches with no browser connected too: web push depends on it. */ @@ -221,9 +246,15 @@ export class OmoStatus { this.timer = null; } + /** the pane's own status may replace herdr's only while it is allowed to (omo: once the session is known) */ + private authoritative(tracked: Tracked): boolean { + return tracked.path !== null && this.ready(tracked); + } + /** herdr's status for this pane is replaced: its events say nothing true. */ tracks(paneId: string): boolean { - return this.panes.get(paneId)?.path != null; + const tracked = this.panes.get(paneId); + return tracked !== undefined && this.authoritative(tracked); } /** The pane runs OmO, its session known or not: its identity is `omo` whatever herdr calls it. */ @@ -276,7 +307,7 @@ export class OmoStatus { const startedAt = pane.startedAt ?? before?.startedAt ?? (prior?.path === pane.path ? prior.startedAt : null); // an error the pane rested on is the same session's alone: another one's answers finish nothing of it const same = prior !== undefined && prior.path === pane.path; - this.panes.set(paneId, { ...pane, startedAt, cwd: cwds.get(paneId) ?? "", offset: -1, size: -1, id: "", retell: prior !== undefined, turn: noTurn(), status: prior?.status ?? "idle", failed: same && prior.failed, failedAt: same ? prior.failedAt : null, history: false, background: prior?.background ?? 0 }); + this.panes.set(paneId, { ...pane, startedAt, cwd: cwds.get(paneId) ?? "", offset: -1, size: -1, id: "", retell: prior !== undefined, turn: noTurn(), status: prior?.status ?? "idle", failed: same && prior.failed, failedAt: same ? prior.failedAt : null, history: false, background: prior?.background ?? 0, available: false, complete: false, polled: this.deps.ready !== undefined && (before?.polled === true || prior !== undefined), authoritative: false }); } if (pane.path !== null) this.last.delete(paneId); } @@ -298,7 +329,7 @@ export class OmoStatus { * OmO's until now: what OmO did there is then no part of the other agent's status. */ named(paneId: string, agent: string | null): boolean { - if (agent === null || agent === "omo" || OMO_ALIASES.includes(agent)) return false; + if (agent === null || agent === this.agent || this.aliases.includes(agent)) return false; const held = this.panes.delete(paneId); const remembered = this.last.delete(paneId); if (!held && !remembered) return false; @@ -314,23 +345,26 @@ export class OmoStatus { poll(): void { const counts = new Map>(); for (const [paneId, tracked] of this.panes) { - if (tracked.path === null) continue; + if (tracked.path === null) { this.noteAuthority(tracked); continue; } const stat = this.file.stat(tracked.path); - if (stat === null) continue; + tracked.available = stat !== null; + if (stat === null) { tracked.complete = false; this.noteAuthority(tracked); continue; } const { size } = stat; if (tracked.offset === -1 || size < tracked.offset || stat.id !== tracked.id) this.readFrom(tracked, tracked.path, size); // also what a read that failed left unread - else if (size !== tracked.size || tracked.offset < size) tracked.offset = this.file.lines(tracked.path, tracked.offset, size, (line) => { tracked.turn = omoTurnAfter(tracked.turn, line); }); + else if (size !== tracked.size || tracked.offset < size) tracked.offset = this.file.lines(tracked.path, tracked.offset, size, (line) => { tracked.turn = this.turnAfter(tracked.turn, line); }); tracked.size = size; tracked.id = stat.id; + tracked.complete = tracked.offset >= size; // nothing could be read of it (a read that failed): what the pane read before stands until one succeeds - if (tracked.turn.status === null && tracked.offset < size) continue; + if (tracked.turn.status === null && tracked.offset < size) { this.noteAuthority(tracked); continue; } const stale = tracked.turn.status === "working" && tracked.turn.at !== null && tracked.startedAt !== null && tracked.turn.at < tracked.startedAt - STALE_TURN_MS; + if (this.staleIsFailure && stale && tracked.turn.asks.length > 0) tracked.turn = { ...tracked.turn, asks: [] }; const status: OmoPaneStatus = tracked.turn.asks.length > 0 ? "blocked" : tracked.turn.status === "working" && !stale ? "working" : "idle"; const sessionId = omoSessionId(tracked.path); if (!counts.has(tracked.cwd)) counts.set(tracked.cwd, tracked.cwd ? this.background(tracked.cwd) : new Map()); const background = sessionId ? counts.get(tracked.cwd)!.get(sessionId) ?? 0 : 0; - const failed = status === "idle" && tracked.turn.failed === true; + const failed = status === "idle" && (tracked.turn.failed === true || (this.staleIsFailure && stale)); // an answer written after the error that ended the turn (a retry's, or one to a new prompt): // the pane worked meanwhile, though it read READY, and this rest is the finish of that work. // Without the times to tell it newer, it is not taken for one: a finish missed, never made up @@ -346,16 +380,32 @@ export class OmoStatus { tracked.failedAt = failed ? tracked.turn.at : null; tracked.history = false; tracked.background = background; - if (changed) this.deps.onChange(paneId, status, background, turn, rest); + // a pane whose file said nothing while it was polled had herdr's word stand for it: + // its own status, once it holds again, is told so that last word does not keep standing + const regained = this.noteAuthority(tracked); + if ((changed || regained) && this.ready(tracked)) this.deps.onChange(paneId, status, background, turn || regained, rest); } } + /** + * What this poll found the pane's authority to be. True when it returns to a pane that was + * polled without it: its own status must then be told, though it did not change — the raw + * word that stood for the pane in between is the one a listener holds. + */ + private noteAuthority(tracked: Tracked): boolean { + const authoritative = this.authoritative(tracked); + const regained = tracked.polled && authoritative && !tracked.authoritative; + tracked.authoritative = authoritative; + tracked.polled = true; + return regained; + } + /** Replay from the start once, then consume only appended complete records. */ private readFrom(tracked: Tracked, path: string, size: number): void { tracked.turn = noTurn(); tracked.history = true; tracked.offset = this.file.lines(path, 0, size, (line) => { - tracked.turn = omoTurnAfter(tracked.turn, line); + tracked.turn = this.turnAfter(tracked.turn, line); }); } @@ -368,7 +418,7 @@ export class OmoStatus { const own =

(entry: P): P => { const tracked = this.panes.get(entry.pane_id); if (!tracked) return entry; - return { ...entry, agent: "omo", ...(tracked.path !== null ? { agent_status: tracked.status } : {}) }; + return { ...entry, agent: this.agent, ...(this.authoritative(tracked) ? { agent_status: tracked.status } : {}) }; }; return { ...snapshot, panes: snapshot.panes.map(own), ...(snapshot.agents ? { agents: snapshot.agents.map(own) } : {}) }; } diff --git a/tsconfig.tools.json b/tsconfig.tools.json index 2b2be4504b..a9fa1d5956 100644 --- a/tsconfig.tools.json +++ b/tsconfig.tools.json @@ -21,6 +21,7 @@ "scripts/chat-virtual-fixture.tsx", "scripts/chat-virtual-browser-qa.ts", "scripts/grok-chat-browser-qa.ts", + "scripts/gjc-status-browser-regression.ts", "scripts/file-preview-text-browser-qa.ts", "scripts/file-preview-text-fixture.tsx", "scripts/virtual-source-browser-qa.ts",