From 0bbc94e4c7039391660c82daff32188618611d25 Mon Sep 17 00:00:00 2001 From: Yehonal <147092+Yehonal@users.noreply.github.com> Date: Mon, 28 Sep 2026 07:42:29 +0000 Subject: [PATCH] fix(gateway): recover long correlated Claude results from the transcript Long Claude replies were lost even when the agent answered correctly: the terminal collector drops replies that scroll out of the pane or outlast timeoutSeconds, and the transcript fallback went through the public search path, which clips each message to 4096 characters. - New `agent-session.sh --agent claude result` reads the whole correlated reply from the native transcript, optionally waiting for it. - The provider uses it after result_uncorrelated or result_timeout, waiting up to agentSessions.resultRecoverySeconds (0-14400, default 0). - agent-native-call.mjs and the new reader write stdout synchronously: process.exit() after an async write cut piped results at 8 KiB. - The bridge forces a UTF-8 locale so prompt and trust-dialog matching work when the host runs with LANG=C. - Fresh sessions: kill the new target when writer ownership can't be proven. - A forked resume no longer counts as a writer of its source session. --- packages/gateway/docs/mcp-ingress.md | 25 +++-- .../src/adapters/agent-session-provider.ts | 60 +++++------ packages/gateway/src/mcp/serve.ts | 8 +- .../test/agent-session-provider.test.js | 53 +++++++--- .../gateway/test/mesh-fresh-session.test.js | 53 ++++++++++ .../gateway/test/tmux-ingress-config.test.js | 9 ++ packages/tmux-bridge/bin/_mesh-tmux.sh | 7 ++ .../tmux-bridge/bin/agent-native-call.mjs | 17 +++- packages/tmux-bridge/bin/agent-session.sh | 8 ++ .../bin/claude-session-ownership.mjs | 13 ++- .../bin/claude-transcript-result.mjs | 99 +++++++++++++++++++ .../test/claude-session-ownership.test.mjs | 9 +- .../test/claude-transcript-result.test.mjs | 74 ++++++++++++++ .../tmux-bridge/test/native-call.test.mjs | 7 ++ scripts/mesh-readiness.mjs | 1 + skills/agent-tmux/bin/_mesh-tmux.sh | 7 ++ skills/agent-tmux/bin/agent-native-call.mjs | 17 +++- skills/agent-tmux/bin/agent-session.sh | 8 ++ .../bin/claude-session-ownership.mjs | 13 ++- .../bin/claude-transcript-result.mjs | 99 +++++++++++++++++++ 20 files changed, 517 insertions(+), 70 deletions(-) create mode 100755 packages/tmux-bridge/bin/claude-transcript-result.mjs create mode 100644 packages/tmux-bridge/test/claude-transcript-result.test.mjs create mode 100755 skills/agent-tmux/bin/claude-transcript-result.mjs diff --git a/packages/gateway/docs/mcp-ingress.md b/packages/gateway/docs/mcp-ingress.md index f58db5b..9c80f21 100644 --- a/packages/gateway/docs/mcp-ingress.md +++ b/packages/gateway/docs/mcp-ingress.md @@ -178,6 +178,7 @@ route: "meshSocket": "mesh-ingress", "timeoutSeconds": 180, "scanLimit": 500, + "resultRecoverySeconds": 3600, "providers": [ { "target_agent_id": "agent.ingress.codex", @@ -219,11 +220,17 @@ reuse rechecks the unique Claude writer PID, its pane ancestry, and the pane PID; the sender still refuses a busy TUI or occupied composer. This ownership proof is process-local and disappears on gateway restart. Otherwise Claude is writable only when the deployment explicitly owns its event-driven Monitor -transport. If Claude's TUI omits result text and the terminal collector reports -`result_uncorrelated`, the gateway checks the same authorized session's native -assistant transcript for one exact correlated marker pair. It returns that -reply only when the unique, nonempty match exists; otherwise the typed failure -is preserved. This fallback never sends another prompt. +transport. + +When the terminal collector reports `result_uncorrelated` (long replies scroll +out of the pane) or `result_timeout` (the turn outlasts `timeoutSeconds`), the +gateway reads the same session's native transcript through +`agent-session.sh --agent claude result`. It returns the whole reply only when +exactly one assistant text block holds exactly one correlated marker pair; +several or unterminated pairs give `result_parsing_failure`, and no match keeps +the original typed failure. `agentSessions.resultRecoverySeconds` (0 to 14400, +default 0 = check once) sets how long it keeps waiting for a late reply; the +task stays `working` meanwhile. This path never sends another prompt. Managed inbox files are named `.jsonl` under `agentManagedInboxRoot`; delivery additionally @@ -301,9 +308,11 @@ correlated result-marker path as other session turns. The session UUID is minted once, when the task is created. A replay with the same `idempotency_key` and input returns that task and session and starts nothing new; the launch itself is also idempotent because an existing target -with that name is reused rather than relaunched. A session created before a -delivery failure or cancellation stays running and is not reused by another -task. +with that name is reused rather than relaunched. If the new pane cannot be +proven to own the session, its target is killed before anything is sent. A +session that already received its prompt stays running after a delivery +failure or cancellation, since it may still be working; it is not reused by +another task. Session profiles (for example a read-only reviewer and a write-enabled implementer) are not implemented yet. When added, they must be server-side diff --git a/packages/gateway/src/adapters/agent-session-provider.ts b/packages/gateway/src/adapters/agent-session-provider.ts index 8f44cd9..d085571 100644 --- a/packages/gateway/src/adapters/agent-session-provider.ts +++ b/packages/gateway/src/adapters/agent-session-provider.ts @@ -1,5 +1,4 @@ import { execFile } from "node:child_process"; -import { createHash } from "node:crypto"; import { isAbsolute, relative, resolve } from "node:path"; import { env as processEnv } from "node:process"; @@ -171,6 +170,8 @@ export interface ShellAgentSessionProviderOptions { workspaceRoots: Readonly>; /** Trusted cwd per workspace for fresh sessions, must be inside that workspace's roots. */ freshSessionCwds?: Readonly>; + /** Seconds to keep polling the Claude transcript for a reply the collector missed. 0 = check once. */ + resultRecoverySeconds?: number; meshSocket?: string; timeoutSeconds?: number; scanLimit?: number; @@ -193,6 +194,7 @@ export class ShellAgentSessionProvider implements AgentSessionProvider { private readonly agentNativeCallPath?: string; private readonly agentManagedInboxRoot?: string; private readonly timeoutSeconds: number; + private readonly resultRecoverySeconds: number; private readonly run: ShellRun; private readonly sender: ShellTmuxSender; private readonly sendQueues = new Map>(); @@ -226,6 +228,7 @@ export class ShellAgentSessionProvider implements AgentSessionProvider { this.agentNativeCallPath = options.agentNativeCallPath; this.agentManagedInboxRoot = options.agentManagedInboxRoot; this.timeoutSeconds = options.timeoutSeconds ?? 120; + this.resultRecoverySeconds = options.resultRecoverySeconds ?? 0; this.run = options.run ?? defaultRun; this.sender = new ShellTmuxSender({ agentSendPath: options.agentSendPath, @@ -460,49 +463,48 @@ export class ShellAgentSessionProvider implements AgentSessionProvider { if (created.stdout.trim().split(/\r?\n/).filter(Boolean).at(-1) !== target) { return failed("Fresh session creation returned an unexpected target."); } + const ownershipError = await this.claimFreshTarget(input, target); + if (ownershipError !== undefined) { + // nothing sent yet, and the target name carries this task's session id + await this.command(["--agent", "claude", "kill", target], true); + return failed(ownershipError); + } + return this.sendToOwnedClaudeTarget(input, target); + } + + private async claimFreshTarget(input: AgentSessionSendInput, target: string): Promise { try { const writerStatus = await this.command(["--agent", "claude", "writer-status", input.sessionId, "--json"], true); - if (writerStatus.code !== 0) return failed(safeProcessError("Fresh session writer inspection failed", writerStatus)); + if (writerStatus.code !== 0) return safeProcessError("Fresh session writer inspection failed", writerStatus); const writers = parseWriterStatus(writerStatus.stdout, "claude", input.sessionId); const writer = writers.length === 1 && writers[0]?.kind === "claude-cli" ? writers[0] : undefined; - if (writer === undefined) return failed("Fresh session has no unique CLI writer."); + if (writer === undefined) return "Fresh session has no unique CLI writer."; const targetStatus = await this.command([ "--agent", "claude", "target-status", target, "--writer-pid", String(writer.pid), "--json" ], true); - if (targetStatus.code !== 0) return failed(safeProcessError("Fresh session target ownership failed", targetStatus)); + if (targetStatus.code !== 0) return safeProcessError("Fresh session target ownership failed", targetStatus); const owned = parseOwnedClaudeTarget(targetStatus.stdout, target, writer.pid); this.ownedClaudeTargets.set(`${input.workspaceId}:${input.sessionId}`, { target, writerPid: writer.pid, panePid: owned.panePid }); + return undefined; } catch (error) { - return failed(error instanceof Error ? error.message : String(error)); + return error instanceof Error ? error.message : String(error); } - return this.sendToOwnedClaudeTarget(input, target); } private async sendToOwnedClaudeTarget(input: AgentSessionSendInput, target: string): Promise { const result = await this.sender.send(this.tmuxSendInput(input, target)); - if (result.result_error_code !== "result_uncorrelated" || input.correlationId === undefined) return result; - const token = createHash("sha256").update(input.correlationId).digest("hex").slice(0, 16); - const begin = `[[R:${token}]]`; - const end = `[[/R:${token}]]`; - try { - const page = await this.search({ workspaceId: input.workspaceId, query: begin, limit: 20 }); - const matches = page.results - .filter((session) => session.session_id === input.sessionId) - .flatMap((session) => session.matches) - .filter((event) => event.role === "assistant"); - const replies = matches.map((event) => { - const start = event.text.indexOf(begin); - if (start < 0 || event.text.indexOf(begin, start + begin.length) >= 0) return undefined; - const finish = event.text.indexOf(end, start + begin.length); - if (finish < 0 || event.text.indexOf(end, finish + end.length) >= 0) return undefined; - return event.text.slice(start + begin.length, finish).trim(); - }).filter((reply): reply is string => typeof reply === "string" && reply.length > 0); - if (replies.length === 1) return { ok: true, reply: replies[0] }; - } catch { - // Native transcript recovery is best-effort; preserve the collector's typed failure. - } + const recoverable = result.result_error_code === "result_uncorrelated" || result.result_error_code === "result_timeout"; + if (!recoverable || input.correlationId === undefined) return result; + // long replies scroll out of the pane and long turns outlast the collector, the transcript has the full reply + const recovered = await this.command([ + "--agent", "claude", "result", input.sessionId, + "--correlation-id", input.correlationId, "--wait", String(this.resultRecoverySeconds) + ], true, (this.resultRecoverySeconds + 30) * 1000); + if (recovered.code === 0 && recovered.stdout.trim().length > 0) return { ok: true, reply: recovered.stdout.trim() }; + if (recovered.code === 65) return { ok: true, result_error_code: "result_no_output", error: "Agent produced no textual result." }; + if (recovered.code === 67) return { ok: true, result_error_code: "result_parsing_failure", error: "Correlated result markers could not be parsed." }; return result; } @@ -541,7 +543,7 @@ export class ShellAgentSessionProvider implements AgentSessionProvider { }; } - private async command(args: readonly string[], allowFailure = false) { + private async command(args: readonly string[], allowFailure = false, timeoutMs = 45_000) { const env = { ...processEnv }; if (this.meshSocket !== undefined) env.MESH_TMUX_SOCKET = this.meshSocket; if (args[2] === "new") env.MESH_STRICT_READY = "1"; @@ -554,7 +556,7 @@ export class ShellAgentSessionProvider implements AgentSessionProvider { env, // A cold resume waits up to 30 seconds for the provider TUI to become // ready; keep the host ceiling above that bridge-owned bound. - timeoutMs: 45_000 + timeoutMs }); if (!allowFailure && response.code !== 0) { throw new Error(safeProcessError("Agent session command failed", response)); diff --git a/packages/gateway/src/mcp/serve.ts b/packages/gateway/src/mcp/serve.ts index 789dd10..9ad988b 100644 --- a/packages/gateway/src/mcp/serve.ts +++ b/packages/gateway/src/mcp/serve.ts @@ -81,6 +81,7 @@ interface AgentSessionsConfig { meshSocket?: string; timeoutSeconds?: number; scanLimit?: number; + resultRecoverySeconds?: number; providers: Array<{ target_agent_id: string; agent_type: "codex" | "claude"; @@ -449,7 +450,9 @@ function validateAgentSessions( (value.meshSocket !== undefined && (typeof value.meshSocket !== "string" || !/^[A-Za-z0-9_.-]{1,64}$/.test(value.meshSocket))) || (value.timeoutSeconds !== undefined && (!Number.isInteger(value.timeoutSeconds) || value.timeoutSeconds < 1 || value.timeoutSeconds > 600)) || - (value.scanLimit !== undefined && (!Number.isInteger(value.scanLimit) || value.scanLimit < 1 || value.scanLimit > 1000)) + (value.scanLimit !== undefined && (!Number.isInteger(value.scanLimit) || value.scanLimit < 1 || value.scanLimit > 1000)) || + (value.resultRecoverySeconds !== undefined && + (!Number.isInteger(value.resultRecoverySeconds) || value.resultRecoverySeconds < 0 || value.resultRecoverySeconds > 14_400)) ) { throw new Error("Invalid agent sessions runtime limits configuration."); } @@ -522,7 +525,8 @@ function createAgentSessionRegistry(config: AgentSessionsConfig | undefined): Ag ...(provider.fresh_session === undefined ? {} : { freshSessionCwds: provider.fresh_session.workspace_cwd }), ...(config.meshSocket === undefined ? {} : { meshSocket: config.meshSocket }), ...(config.timeoutSeconds === undefined ? {} : { timeoutSeconds: config.timeoutSeconds }), - ...(config.scanLimit === undefined ? {} : { scanLimit: config.scanLimit }) + ...(config.scanLimit === undefined ? {} : { scanLimit: config.scanLimit }), + ...(config.resultRecoverySeconds === undefined ? {} : { resultRecoverySeconds: config.resultRecoverySeconds }) }))); } diff --git a/packages/gateway/test/agent-session-provider.test.js b/packages/gateway/test/agent-session-provider.test.js index fe57069..601f844 100644 --- a/packages/gateway/test/agent-session-provider.test.js +++ b/packages/gateway/test/agent-session-provider.test.js @@ -1,5 +1,4 @@ import assert from "node:assert/strict"; -import { createHash } from "node:crypto"; import test from "node:test"; import "./ts-extension-resolver.mjs"; @@ -321,11 +320,9 @@ test("two Claude calls reuse only the writer resumed by this provider", async () assert.equal(calls.some((call) => call.command === process.execPath), false); }); -test("Claude recovers an exact correlated reply from its native transcript when TUI capture loses it", async () => { +test("Claude recovers the whole correlated reply from its native transcript when TUI capture loses it", async () => { const correlationId = "task-transcript-fallback"; - const token = createHash("sha256").update(correlationId).digest("hex").slice(0, 16); - const begin = `[[R:${token}]]`; - const end = `[[/R:${token}]]`; + const longReply = `expected reply\n${"detail ".repeat(1500)}`.trim(); const calls = []; let resumed = false; const instance = new ShellAgentSessionProvider({ @@ -339,10 +336,7 @@ test("Claude recovers an exact correlated reply from its native transcript when if (args.includes("resume")) { resumed = true; return { code: 0, stdout: "mesh-claude-session-new\n", stderr: "" }; } if (args.includes("target-status")) return { code: 0, stdout: JSON.stringify({ target: "mesh-claude-session-new", pane_pid: 42, writer_pid: 43 }), stderr: "" }; if (command === "/bridge/agent-send.sh") return { code: 66, stdout: "", stderr: "UNCORRELATED-OUTPUT" }; - if (args.includes("search")) return { code: 0, stdout: JSON.stringify({ agent_type: "claude", results: [ - { agent_type: "claude", session_id: "other-session", cwd: "/workspace/project-a", updated_at: "2026-08-30T12:00:00Z", matches: [{ event_id: "wrong", role: "assistant", timestamp: null, text: `${begin}wrong${end}` }] }, - { agent_type: "claude", session_id: "session-new", cwd: "/workspace/project-a", updated_at: "2026-08-30T12:00:01Z", matches: [{ event_id: "right", role: "assistant", timestamp: null, text: `${begin}\nexpected reply\n${end}` }] } - ], next_cursor: null }), stderr: "" }; + if (args[2] === "result") return { code: 0, stdout: `${longReply}\n`, stderr: "" }; throw new Error(`Unexpected command: ${command} ${args.join(" ")}`); } }); @@ -350,9 +344,42 @@ test("Claude recovers an exact correlated reply from its native transcript when sessionId: "session-new", workspaceId: "workspace.allowed", message: "canary", messageId: "message", contextId: "context", correlationId, idempotencyKey: "idem" }); - assert.deepEqual(result, { ok: true, reply: "expected reply" }); + assert.deepEqual(result, { ok: true, reply: longReply }); assert.equal(calls.filter((call) => call.command === "/bridge/agent-send.sh").length, 1); - assert.equal(calls.filter((call) => call.args.includes("search")).length, 1); + const recovery = calls.find((call) => call.args[2] === "result"); + assert.deepEqual(recovery.args, [ + "--agent", "claude", "result", "session-new", "--correlation-id", correlationId, "--wait", "0" + ]); + assert.equal(calls.some((call) => call.args.includes("search")), false); +}); + +test("Claude waits for a late correlated reply after the terminal collector times out", async () => { + const calls = []; + let resumed = false; + const instance = new ShellAgentSessionProvider({ + agentId: "agent.ingress.claude", agentType: "claude", + agentSessionPath: "/bridge/agent-session.sh", agentSendPath: "/bridge/agent-send.sh", + workspaceRoots: { "workspace.allowed": ["/workspace"] }, + resultRecoverySeconds: 900, + run: async (command, args, options) => { + calls.push({ command, args: [...args], timeoutMs: options.timeoutMs }); + if (args.includes("inspect")) return { code: 0, stdout: JSON.stringify({ agent_type: "claude", sessions: [{ ...sessions[0], agent_type: "claude" }] }), stderr: "" }; + if (args.includes("writer-status")) return { code: 0, stdout: JSON.stringify({ agent: "claude", sessionId: "session-new", discovery: { complete: true }, writers: resumed ? [{ pid: 43, kind: "claude-cli" }] : [] }), stderr: "" }; + if (args.includes("resume")) { resumed = true; return { code: 0, stdout: "mesh-claude-session-new\n", stderr: "" }; } + if (args.includes("target-status")) return { code: 0, stdout: JSON.stringify({ target: "mesh-claude-session-new", pane_pid: 42, writer_pid: 43 }), stderr: "" }; + if (command === "/bridge/agent-send.sh") return { code: 124, stdout: "", stderr: "timed out" }; + if (args[2] === "result") return { code: 0, stdout: "late review\n", stderr: "" }; + throw new Error(`Unexpected command: ${command} ${args.join(" ")}`); + } + }); + const result = await instance.send({ + sessionId: "session-new", workspaceId: "workspace.allowed", message: "long review", + messageId: "message", contextId: "context", correlationId: "task-late", idempotencyKey: "idem" + }); + assert.deepEqual(result, { ok: true, reply: "late review" }); + const recovery = calls.find((call) => call.args[2] === "result"); + assert.deepEqual(recovery.args.slice(-2), ["--wait", "900"]); + assert.equal(recovery.timeoutMs, 930_000); }); test("Claude transcript recovery keeps an uncorrelated failure when no exact assistant reply exists", async () => { @@ -370,9 +397,7 @@ test("Claude transcript recovery keeps an uncorrelated failure when no exact ass if (args.includes("resume")) { resumed = true; return { code: 0, stdout: "mesh-claude-session-new\n", stderr: "" }; } if (args.includes("target-status")) return { code: 0, stdout: JSON.stringify({ target: "mesh-claude-session-new", pane_pid: 42, writer_pid: 43 }), stderr: "" }; if (command === "/bridge/agent-send.sh") return { code: 66, stdout: "", stderr: "UNCORRELATED-OUTPUT" }; - if (args.includes("search")) return { code: 0, stdout: JSON.stringify({ agent_type: "claude", results: [ - { agent_type: "claude", session_id: "session-new", cwd: "/workspace/project-a", updated_at: "2026-08-30T12:00:01Z", matches: [{ event_id: "user", role: "user", timestamp: null, text: args[3] }] } - ], next_cursor: null }), stderr: "" }; + if (args[2] === "result") return { code: 124, stdout: "", stderr: "ERROR: No correlated result in the session transcript yet." }; throw new Error(`Unexpected command: ${command} ${args.join(" ")}`); } }); diff --git a/packages/gateway/test/mesh-fresh-session.test.js b/packages/gateway/test/mesh-fresh-session.test.js index e593513..e1801e1 100644 --- a/packages/gateway/test/mesh-fresh-session.test.js +++ b/packages/gateway/test/mesh-fresh-session.test.js @@ -53,6 +53,15 @@ function fakeBridge(behavior = {}) { stderr: "" }; } + if (command === "/bridge/agent-session.sh" && sub === "result") { + const correlationId = args[args.indexOf("--correlation-id") + 1]; + return behavior.transcriptResult === undefined + ? { code: 124, stdout: "", stderr: "ERROR: No correlated result in the session transcript yet." } + : { code: 0, stdout: `${behavior.transcriptResult(correlationId)}\n`, stderr: "" }; + } + if (command === "/bridge/agent-session.sh" && sub === "kill") { + return { code: 0, stdout: `Killed ${args[3]}\n`, stderr: "" }; + } if (command === "/bridge/agent-session.sh" && sub === "target-status") { return { code: 0, stdout: JSON.stringify({ target: args[3], pane_pid: 4241, writer_pid: 4242 }), stderr: "" }; } @@ -348,10 +357,38 @@ test("a failed session creation fails the task without falling back to another s assert.equal(sessionCommands(h.bridge.calls, "new").length, 1); assert.equal(sessionCommands(h.bridge.calls, "resume").length, 0); assert.equal(h.bridge.calls.some((call) => call.command === "/bridge/agent-send.sh"), false); + // the bridge retires a failed launch itself, an unproven owner is retired by the provider + const killed = sessionCommands(h.bridge.calls, "kill").map((call) => call.args[3]); + assert.deepEqual(killed, behavior.noWriter ? [`mesh-claude-${task.session_id}`] : []); } finally { await h.cleanup(); } } }); +test("a fresh task whose provider lost fresh support fails with fresh_session_unsupported at execution", async () => { + const h = await harness(); + try { + const created = await h.store.create({ + contextId: "context-unsupported", + principalId: PRINCIPAL.id, + principalKind: PRINCIPAL.kind, + requesterId: PRINCIPAL.requesterId, + targetAgentId: CODEX, + sessionMode: "fresh", + sessionProvider: "codex", + workspaceId: WORKSPACE, + domainId: DOMAIN, + message: "Config changed after submit.", + labels: [], + idempotencyKey: `${PRINCIPAL.id}:unsupported` + }); + await h.taskCoordinator.resume(); + const task = await waitForTerminal(h.facade, created.task.task_id); + assert.equal(task.status, "failed"); + assert.equal(task.error.code, "fresh_session_unsupported"); + assert.equal(h.bridge.calls.length, 0); + } finally { await h.cleanup(); } +}); + test("a delivery failure after creation fails the task instead of completing it", async () => { const h = await harness({ send: () => ({ code: 1, stdout: "", stderr: "ERROR: paste failed" }) }); try { @@ -398,6 +435,22 @@ test("an uncorrelated or timed-out fresh result is never reported as completed", } }); +test("a long fresh reply lost by the terminal collector is recovered whole from the transcript", async () => { + const body = `OVERALL: PASS\n${"finding ".repeat(2000)}`.trim(); + const h = await harness({ + send: () => ({ code: 66, stdout: "", stderr: "markers scrolled away" }), + transcriptResult: () => body + }); + try { + const { task } = await h.facade.callTask(fresh(), 2_000); + assert.equal(task.status, "completed", JSON.stringify(task.error)); + assert.equal(task.result.text, body); + const recovery = sessionCommands(h.bridge.calls, "result")[0]; + assert.equal(recovery.args[recovery.args.indexOf("--correlation-id") + 1], task.task_id); + assert.equal(recovery.args[3], task.session_id); + } finally { await h.cleanup(); } +}); + test("mesh_call returns a task handle when the wait expires and mesh_task_get later reads the result", async () => { let release; const gate = new Promise((resolve) => { release = resolve; }); diff --git a/packages/gateway/test/tmux-ingress-config.test.js b/packages/gateway/test/tmux-ingress-config.test.js index dd4510b..748d6d4 100644 --- a/packages/gateway/test/tmux-ingress-config.test.js +++ b/packages/gateway/test/tmux-ingress-config.test.js @@ -112,6 +112,15 @@ test("runtime config accepts a workspace-scoped Codex session provider", async ( ); }); +test("result recovery wait is bounded", async () => { + const accepted = await readRuntimeConfig(await writeConfig({ agentSessions: { ...AGENT_SESSIONS, resultRecoverySeconds: 3600 } })); + assert.equal(accepted.agentSessions.resultRecoverySeconds, 3600); + for (const value of [-1, 14_401, 1.5, "60"]) { + const path = await writeConfig({ agentSessions: { ...AGENT_SESSIONS, resultRecoverySeconds: value } }); + await assert.rejects(readRuntimeConfig(path), /runtime limits/); + } +}); + test("session native call path must be absolute when configured", async () => { const path = await writeConfig({ agentSessions: { ...AGENT_SESSIONS, agentNativeCallPath: "agent-native-call.mjs" } diff --git a/packages/tmux-bridge/bin/_mesh-tmux.sh b/packages/tmux-bridge/bin/_mesh-tmux.sh index 70993b4..ddec90b 100644 --- a/packages/tmux-bridge/bin/_mesh-tmux.sh +++ b/packages/tmux-bridge/bin/_mesh-tmux.sh @@ -13,6 +13,13 @@ # # Override the socket name with MESH_TMUX_SOCKET (e.g. tests use a throwaway one). +# prompt and trust-dialog patterns use non-ASCII glyphs (❯, ·), which tmux and grep +# mangle under a C locale, so readiness and dialog checks would never match +case "${LC_ALL:-${LC_CTYPE:-${LANG:-}}}" in + *[Uu][Tt][Ff]-8*|*[Uu][Tt][Ff]8*) ;; + *) export LC_ALL=C.UTF-8 ;; +esac + : "${MESH_TMUX_SOCKET:=mesh}" # Wrapper: every bridge tmux call goes through the dedicated socket. diff --git a/packages/tmux-bridge/bin/agent-native-call.mjs b/packages/tmux-bridge/bin/agent-native-call.mjs index dff9f7d..2af38f1 100755 --- a/packages/tmux-bridge/bin/agent-native-call.mjs +++ b/packages/tmux-bridge/bin/agent-native-call.mjs @@ -3,6 +3,7 @@ import { createHash } from "node:crypto"; import { execFile } from "node:child_process"; +import { writeSync } from "node:fs"; import { access, open, readdir, stat } from "node:fs/promises"; import { homedir } from "node:os"; import { join, resolve } from "node:path"; @@ -140,7 +141,7 @@ function finish(bodies, begin, end) { if (beginIndex >= 0 && endIndex >= 0) { const result = text.slice(beginIndex + begin.length, endIndex).trim(); if (!result) fail("Agent produced no textual result.", 65); - process.stdout.write(`${result}\n`); + writeAll(`${result}\n`); process.exit(0); } if (bodies.length === 0) fail("Agent produced no textual result.", 65); @@ -148,7 +149,7 @@ function finish(bodies, begin, end) { // The unique user-turn anchor plus the following task_complete event already // establishes deterministic correlation. Markers remain an optional stronger // extraction protocol for agents that honor them. - process.stdout.write(`${text.trim()}\n`); + writeAll(`${text.trim()}\n`); process.exit(0); } @@ -305,6 +306,16 @@ function required(value, flag) { return text; } +// process.exit() right after an async stdout write cuts piped output at 8 KiB. +function writeAll(text) { + const buffer = Buffer.from(text); + let offset = 0; + while (offset < buffer.length) { + try { offset += writeSync(1, buffer, offset); } + catch (error) { if (error.code !== "EAGAIN") throw error; } + } +} + function sleep(ms) { return new Promise((resolveSleep) => setTimeout(resolveSleep, Math.max(10, ms))); } @@ -343,6 +354,6 @@ function blockClaudeVisibleTurn(id, correlation, reasonOverride) { ], }, }; - process.stdout.write(`${JSON.stringify(blocker)}\n`); + writeAll(`${JSON.stringify(blocker)}\n`); process.exit(78); } diff --git a/packages/tmux-bridge/bin/agent-session.sh b/packages/tmux-bridge/bin/agent-session.sh index 6d97349..007e40d 100755 --- a/packages/tmux-bridge/bin/agent-session.sh +++ b/packages/tmux-bridge/bin/agent-session.sh @@ -10,6 +10,7 @@ # agent-session.sh --agent transcript [--json] [--cursor ] [--limit ] # agent-session.sh --agent search [--json] [--limit ] # agent-session.sh --agent writer-status [--json] +# agent-session.sh --agent claude result --correlation-id [--wait ] # agent-session.sh --agent claude target-status --writer-pid --json # agent-session.sh --agent kill # @@ -850,6 +851,13 @@ case "$cmd" in python3 "$SCRIPT_DIR/agent-watch.py" "${WATCH_ARGS[@]}" ;; + result) + [[ "$AGENT_NAME" == "claude" ]] || { echo "ERROR: result is only implemented for claude" >&2; exit 2; } + SESSION_ID="${1:-}" + shift || true + node "$SCRIPT_DIR/claude-transcript-result.mjs" --session "$SESSION_ID" --root "$AGENT_SESSION_DIR" "$@" + ;; + writer-status) SESSION_ID="${1:-}" shift || true diff --git a/packages/tmux-bridge/bin/claude-session-ownership.mjs b/packages/tmux-bridge/bin/claude-session-ownership.mjs index 784f677..84e3631 100644 --- a/packages/tmux-bridge/bin/claude-session-ownership.mjs +++ b/packages/tmux-bridge/bin/claude-session-ownership.mjs @@ -88,11 +88,16 @@ function classifyClaudeProcess(argv) { return undefined; } -// --session-id owns the session the same way --resume does +// --session-id owns the session the same way --resume does, but a forked resume +// only reads its source and writes under a new id. function claimsSession(argv, sessionId) { + const forked = argv.includes("--fork-session"); return argv.some((value, index) => ( - value === `--resume=${sessionId}` - || value === `--session-id=${sessionId}` - || ((value === "--resume" || value === "-r" || value === "--session-id") && argv[index + 1] === sessionId) + value === `--session-id=${sessionId}` + || (value === "--session-id" && argv[index + 1] === sessionId) + || (!forked && ( + value === `--resume=${sessionId}` + || ((value === "--resume" || value === "-r") && argv[index + 1] === sessionId) + )) )); } diff --git a/packages/tmux-bridge/bin/claude-transcript-result.mjs b/packages/tmux-bridge/bin/claude-transcript-result.mjs new file mode 100755 index 0000000..bc06516 --- /dev/null +++ b/packages/tmux-bridge/bin/claude-transcript-result.mjs @@ -0,0 +1,99 @@ +#!/usr/bin/env node +/** Read one correlated Claude result, unclipped, from the session's native transcript. */ + +import { createHash } from "node:crypto"; +import { writeSync } from "node:fs"; +import { readdir, readFile } from "node:fs/promises"; +import { homedir } from "node:os"; +import { join } from "node:path"; +import { parseArgs } from "node:util"; + +const { values } = parseArgs({ + options: { + session: { type: "string" }, + "correlation-id": { type: "string" }, + wait: { type: "string", default: "0" }, + root: { type: "string" } + }, + strict: true +}); + +const sessionId = values.session ?? ""; +const correlationId = values["correlation-id"] ?? ""; +const waitSeconds = Number(values.wait); +if (!/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(sessionId)) fail("--session must be a complete UUID", 2); +if (!/^[A-Za-z0-9._:-]+$/.test(correlationId)) fail("--correlation-id contains unsafe characters", 2); +if (!Number.isInteger(waitSeconds) || waitSeconds < 0 || waitSeconds > 14_400) fail("--wait must be an integer from 0 to 14400", 2); + +const root = values.root || process.env.CLAUDE_SESSION_ROOT || join(process.env.CLAUDE_CONFIG_DIR || join(homedir(), ".claude"), "projects"); +const token = createHash("sha256").update(correlationId).digest("hex").slice(0, 16); +const begin = `[[R:${token}]]`; +const end = `[[/R:${token}]]`; +const deadline = Date.now() + waitSeconds * 1000; +const pollMs = Number(process.env.AGENT_TRANSCRIPT_RESULT_POLL_MS || 5000); + +while (true) { + const outcome = await collect(); + if (outcome.kind === "found") { + if (!outcome.text) fail("Agent produced no textual result.", 65); + writeAll(`${outcome.text}\n`); + process.exit(0); + } + if (Date.now() >= deadline) { + if (outcome.kind === "ambiguous") fail("Correlated result markers could not be parsed.", 67); + fail("No correlated result in the session transcript yet.", 124); + } + await new Promise((resolveSleep) => setTimeout(resolveSleep, Math.min(pollMs, Math.max(1, deadline - Date.now())))); +} + +async function collect() { + const path = await transcriptPath(); + if (path === undefined) return { kind: "missing" }; + let raw; + try { raw = await readFile(path, "utf8"); } + catch { return { kind: "missing" }; } + const texts = []; + for (const line of raw.split("\n")) { + if (!line.includes(token)) continue; + let record; + try { record = JSON.parse(line); } + catch { continue; } + if (record?.type !== "assistant" || !Array.isArray(record?.message?.content)) continue; + for (const part of record.message.content) { + if (part?.type === "text" && typeof part.text === "string" && part.text.includes(begin)) texts.push(part.text); + } + } + if (texts.length === 0) return { kind: "missing" }; + // exactly one text block with exactly one marker pair, anything else isn't trustworthy + if (texts.length > 1) return { kind: "ambiguous" }; + const text = texts[0]; + const start = text.indexOf(begin); + const finish = text.indexOf(end, start + begin.length); + if (finish < 0 || text.indexOf(begin, start + begin.length) >= 0 || text.indexOf(end, finish + end.length) >= 0) { + return { kind: "ambiguous" }; + } + return { kind: "found", text: text.slice(start + begin.length, finish).trim() }; +} + +async function transcriptPath() { + let entries; + try { entries = await readdir(root, { recursive: true, withFileTypes: true }); } + catch { return undefined; } + const match = entries.find((entry) => entry.isFile() && entry.name === `${sessionId}.jsonl`); + return match === undefined ? undefined : join(match.parentPath, match.name); +} + +// process.exit() right after an async stdout write cuts piped output at 8 KiB. +function writeAll(text) { + const buffer = Buffer.from(text); + let offset = 0; + while (offset < buffer.length) { + try { offset += writeSync(1, buffer, offset); } + catch (error) { if (error.code !== "EAGAIN") throw error; } + } +} + +function fail(message, code) { + process.stderr.write(`ERROR: ${message}\n`); + process.exit(code); +} diff --git a/packages/tmux-bridge/test/claude-session-ownership.test.mjs b/packages/tmux-bridge/test/claude-session-ownership.test.mjs index ea14220..bfb7883 100644 --- a/packages/tmux-bridge/test/claude-session-ownership.test.mjs +++ b/packages/tmux-bridge/test/claude-session-ownership.test.mjs @@ -8,7 +8,7 @@ import { inspectClaudeSessionOwnership } from "../bin/claude-session-ownership.m const SESSION = "22222222-2222-4222-8222-222222222222"; -test("a Claude CLI launched with --session-id owns that session", () => { +test("a Claude CLI launched with --session-id owns that session, a forked resume does not own its source", () => { const root = mkdtempSync(join(tmpdir(), "mesh-claude-ownership-")); try { const proc = join(root, "proc"); @@ -16,7 +16,9 @@ test("a Claude CLI launched with --session-id owns that session", () => { 200: ["claude", "--session-id", SESSION], 201: ["claude", `--session-id=${SESSION}`], 202: ["claude", "--session-id", "33333333-3333-4333-8333-333333333333"], - 203: ["node", "--session-id", SESSION] + 203: ["node", "--session-id", SESSION], + 204: ["claude", "--resume", SESSION, "--fork-session"], + 205: ["claude", "--resume", "44444444-4444-4444-8444-444444444444", "--fork-session", "--session-id", SESSION] }; for (const [pid, argv] of Object.entries(argvs)) { mkdirSync(join(proc, pid), { recursive: true }); @@ -31,7 +33,8 @@ test("a Claude CLI launched with --session-id owns that session", () => { assert.equal(ownership.discovery.complete, true); assert.deepEqual(ownership.writers.map((writer) => [writer.pid, writer.kind]), [ [200, "claude-cli"], - [201, "claude-cli"] + [201, "claude-cli"], + [205, "claude-cli"] ]); } finally { rmSync(root, { recursive: true, force: true }); diff --git a/packages/tmux-bridge/test/claude-transcript-result.test.mjs b/packages/tmux-bridge/test/claude-transcript-result.test.mjs new file mode 100644 index 0000000..5f73ae4 --- /dev/null +++ b/packages/tmux-bridge/test/claude-transcript-result.test.mjs @@ -0,0 +1,74 @@ +import assert from "node:assert/strict"; +import { execFile } from "node:child_process"; +import { createHash } from "node:crypto"; +import { mkdtemp, mkdir, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import test from "node:test"; + +const script = new URL("../bin/claude-transcript-result.mjs", import.meta.url).pathname; +const SESSION = "55555555-5555-4555-8555-555555555555"; +const TASK = "mesh_task_long_result"; +const token = createHash("sha256").update(TASK).digest("hex").slice(0, 16); + +function assistant(text, stop = "end_turn") { + return JSON.stringify({ type: "assistant", message: { stop_reason: stop, content: [{ type: "text", text }] } }); +} + +function run(root, wait = "0") { + return new Promise((resolveRun) => { + execFile(process.execPath, [script, "--session", SESSION, "--correlation-id", TASK, "--wait", wait, "--root", root], { + env: { ...process.env, AGENT_TRANSCRIPT_RESULT_POLL_MS: "20" }, + maxBuffer: 10 * 1024 * 1024 + }, (error, stdout, stderr) => resolveRun({ code: error ? error.code : 0, stdout, stderr })); + }); +} + +async function fixture(lines) { + const root = await mkdtemp(join(tmpdir(), "mesh-transcript-result-")); + const dir = join(root, "-srv-workspaces-example"); + await mkdir(dir, { recursive: true }); + if (lines !== undefined) await writeFile(join(dir, `${SESSION}.jsonl`), `${lines.join("\n")}\n`); + return { root, dir }; +} + +test("returns the whole correlated reply even past the pipe buffer", async () => { + const body = `OVERALL: PASS\n${"finding line\n".repeat(4000)}`; + const { root } = await fixture([ + assistant("Working on it.", "tool_use"), + assistant(`Summary first.\n[[R:${token}]]\n${body}\n[[/R:${token}]]`) + ]); + try { + const result = await run(root); + assert.equal(result.code, 0, result.stderr); + assert.equal(result.stdout, `${body.trim()}\n`); + } finally { await rm(root, { recursive: true, force: true }); } +}); + +test("reports a missing result as a timeout and ambiguous markers as a parse failure", async () => { + const empty = await fixture([assistant("no markers here")]); + const doubled = await fixture([ + assistant(`[[R:${token}]] a [[/R:${token}]]`), + assistant(`[[R:${token}]] b [[/R:${token}]]`) + ]); + const unterminated = await fixture([assistant(`[[R:${token}]] cut off`)]); + try { + assert.equal((await run(empty.root)).code, 124); + assert.equal((await run(doubled.root)).code, 67); + assert.equal((await run(unterminated.root)).code, 67); + } finally { + for (const item of [empty, doubled, unterminated]) await rm(item.root, { recursive: true, force: true }); + } +}); + +test("waits for a result that lands in the transcript later", async () => { + const { root, dir } = await fixture(undefined); + try { + const pending = run(root, "5"); + await new Promise((resolveDelay) => setTimeout(resolveDelay, 200)); + await writeFile(join(dir, `${SESSION}.jsonl`), `${assistant(`[[R:${token}]]late answer[[/R:${token}]]`)}\n`); + const result = await pending; + assert.equal(result.code, 0, result.stderr); + assert.equal(result.stdout, "late answer\n"); + } finally { await rm(root, { recursive: true, force: true }); } +}); diff --git a/packages/tmux-bridge/test/native-call.test.mjs b/packages/tmux-bridge/test/native-call.test.mjs index 30f7d26..9250849 100644 --- a/packages/tmux-bridge/test/native-call.test.mjs +++ b/packages/tmux-bridge/test/native-call.test.mjs @@ -37,6 +37,7 @@ if (${JSON.stringify(mode)} === "success" || ${JSON.stringify(mode)} === "rotati lines.push({ type: "event_msg", payload: { type: "agent_message", message: begin + " MULTI\\nLINE " + end } }); lines.push({ type: "response_item", payload: { type: "message", role: "assistant", phase: "final_answer", content: [{ type: "output_text", text: begin + " MULTI\\nLINE " + end }] } }); } +if (${JSON.stringify(mode)} === "large") lines.push({ type: "response_item", payload: { type: "message", role: "assistant", phase: "final_answer", content: [{ type: "output_text", text: begin + "\\n" + "0123456789".repeat(20000) + "\\n" + end }] } }); if (${JSON.stringify(mode)} === "fallback") lines.push({ type: "response_item", payload: { type: "message", role: "assistant", phase: "final_answer", content: [{ type: "output_text", text: "MESH_CORRELATION_OK" }] } }); if (${JSON.stringify(mode)} === "uncorrelated") lines.push({ type: "response_item", payload: { type: "message", role: "assistant", phase: "final_answer", content: [{ type: "output_text", text: "plain output" }] } }); if (${JSON.stringify(mode)} === "parsing") lines.push({ type: "response_item", payload: { type: "message", role: "assistant", phase: "final_answer", content: [{ type: "output_text", text: begin + " broken" }] } }); @@ -86,6 +87,12 @@ test("native Codex queue ignores an earlier completion and returns correlated mu assert.equal(result.stdout, "MULTI\nLINE\n"); }); +test("native Codex queue returns a result larger than the pipe buffer intact", async () => { + const result = await invoke("large"); + assert.equal(result.code, 0, result.stderr); + assert.equal(result.stdout, `${"0123456789".repeat(20000)}\n`); +}); + test("native Codex queue accepts the final answer when the anchored turn omits optional markers", async () => { const result = await invoke("fallback"); assert.equal(result.code, 0, result.stderr); diff --git a/scripts/mesh-readiness.mjs b/scripts/mesh-readiness.mjs index b3e74d1..5512dfe 100755 --- a/scripts/mesh-readiness.mjs +++ b/scripts/mesh-readiness.mjs @@ -17,6 +17,7 @@ const agentTmuxBundleFiles = [ "bin/agent-native-call.mjs", "bin/codex-resume-options.py", "bin/claude-session-ownership.mjs", + "bin/claude-transcript-result.mjs", "bin/agent-wait.sh", "bin/mesh-list-agents.sh", "bin/mesh-capacity-dispatch.mjs", diff --git a/skills/agent-tmux/bin/_mesh-tmux.sh b/skills/agent-tmux/bin/_mesh-tmux.sh index 70993b4..ddec90b 100644 --- a/skills/agent-tmux/bin/_mesh-tmux.sh +++ b/skills/agent-tmux/bin/_mesh-tmux.sh @@ -13,6 +13,13 @@ # # Override the socket name with MESH_TMUX_SOCKET (e.g. tests use a throwaway one). +# prompt and trust-dialog patterns use non-ASCII glyphs (❯, ·), which tmux and grep +# mangle under a C locale, so readiness and dialog checks would never match +case "${LC_ALL:-${LC_CTYPE:-${LANG:-}}}" in + *[Uu][Tt][Ff]-8*|*[Uu][Tt][Ff]8*) ;; + *) export LC_ALL=C.UTF-8 ;; +esac + : "${MESH_TMUX_SOCKET:=mesh}" # Wrapper: every bridge tmux call goes through the dedicated socket. diff --git a/skills/agent-tmux/bin/agent-native-call.mjs b/skills/agent-tmux/bin/agent-native-call.mjs index dff9f7d..2af38f1 100755 --- a/skills/agent-tmux/bin/agent-native-call.mjs +++ b/skills/agent-tmux/bin/agent-native-call.mjs @@ -3,6 +3,7 @@ import { createHash } from "node:crypto"; import { execFile } from "node:child_process"; +import { writeSync } from "node:fs"; import { access, open, readdir, stat } from "node:fs/promises"; import { homedir } from "node:os"; import { join, resolve } from "node:path"; @@ -140,7 +141,7 @@ function finish(bodies, begin, end) { if (beginIndex >= 0 && endIndex >= 0) { const result = text.slice(beginIndex + begin.length, endIndex).trim(); if (!result) fail("Agent produced no textual result.", 65); - process.stdout.write(`${result}\n`); + writeAll(`${result}\n`); process.exit(0); } if (bodies.length === 0) fail("Agent produced no textual result.", 65); @@ -148,7 +149,7 @@ function finish(bodies, begin, end) { // The unique user-turn anchor plus the following task_complete event already // establishes deterministic correlation. Markers remain an optional stronger // extraction protocol for agents that honor them. - process.stdout.write(`${text.trim()}\n`); + writeAll(`${text.trim()}\n`); process.exit(0); } @@ -305,6 +306,16 @@ function required(value, flag) { return text; } +// process.exit() right after an async stdout write cuts piped output at 8 KiB. +function writeAll(text) { + const buffer = Buffer.from(text); + let offset = 0; + while (offset < buffer.length) { + try { offset += writeSync(1, buffer, offset); } + catch (error) { if (error.code !== "EAGAIN") throw error; } + } +} + function sleep(ms) { return new Promise((resolveSleep) => setTimeout(resolveSleep, Math.max(10, ms))); } @@ -343,6 +354,6 @@ function blockClaudeVisibleTurn(id, correlation, reasonOverride) { ], }, }; - process.stdout.write(`${JSON.stringify(blocker)}\n`); + writeAll(`${JSON.stringify(blocker)}\n`); process.exit(78); } diff --git a/skills/agent-tmux/bin/agent-session.sh b/skills/agent-tmux/bin/agent-session.sh index 6d97349..007e40d 100755 --- a/skills/agent-tmux/bin/agent-session.sh +++ b/skills/agent-tmux/bin/agent-session.sh @@ -10,6 +10,7 @@ # agent-session.sh --agent transcript [--json] [--cursor ] [--limit ] # agent-session.sh --agent search [--json] [--limit ] # agent-session.sh --agent writer-status [--json] +# agent-session.sh --agent claude result --correlation-id [--wait ] # agent-session.sh --agent claude target-status --writer-pid --json # agent-session.sh --agent kill # @@ -850,6 +851,13 @@ case "$cmd" in python3 "$SCRIPT_DIR/agent-watch.py" "${WATCH_ARGS[@]}" ;; + result) + [[ "$AGENT_NAME" == "claude" ]] || { echo "ERROR: result is only implemented for claude" >&2; exit 2; } + SESSION_ID="${1:-}" + shift || true + node "$SCRIPT_DIR/claude-transcript-result.mjs" --session "$SESSION_ID" --root "$AGENT_SESSION_DIR" "$@" + ;; + writer-status) SESSION_ID="${1:-}" shift || true diff --git a/skills/agent-tmux/bin/claude-session-ownership.mjs b/skills/agent-tmux/bin/claude-session-ownership.mjs index 784f677..84e3631 100644 --- a/skills/agent-tmux/bin/claude-session-ownership.mjs +++ b/skills/agent-tmux/bin/claude-session-ownership.mjs @@ -88,11 +88,16 @@ function classifyClaudeProcess(argv) { return undefined; } -// --session-id owns the session the same way --resume does +// --session-id owns the session the same way --resume does, but a forked resume +// only reads its source and writes under a new id. function claimsSession(argv, sessionId) { + const forked = argv.includes("--fork-session"); return argv.some((value, index) => ( - value === `--resume=${sessionId}` - || value === `--session-id=${sessionId}` - || ((value === "--resume" || value === "-r" || value === "--session-id") && argv[index + 1] === sessionId) + value === `--session-id=${sessionId}` + || (value === "--session-id" && argv[index + 1] === sessionId) + || (!forked && ( + value === `--resume=${sessionId}` + || ((value === "--resume" || value === "-r") && argv[index + 1] === sessionId) + )) )); } diff --git a/skills/agent-tmux/bin/claude-transcript-result.mjs b/skills/agent-tmux/bin/claude-transcript-result.mjs new file mode 100755 index 0000000..bc06516 --- /dev/null +++ b/skills/agent-tmux/bin/claude-transcript-result.mjs @@ -0,0 +1,99 @@ +#!/usr/bin/env node +/** Read one correlated Claude result, unclipped, from the session's native transcript. */ + +import { createHash } from "node:crypto"; +import { writeSync } from "node:fs"; +import { readdir, readFile } from "node:fs/promises"; +import { homedir } from "node:os"; +import { join } from "node:path"; +import { parseArgs } from "node:util"; + +const { values } = parseArgs({ + options: { + session: { type: "string" }, + "correlation-id": { type: "string" }, + wait: { type: "string", default: "0" }, + root: { type: "string" } + }, + strict: true +}); + +const sessionId = values.session ?? ""; +const correlationId = values["correlation-id"] ?? ""; +const waitSeconds = Number(values.wait); +if (!/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(sessionId)) fail("--session must be a complete UUID", 2); +if (!/^[A-Za-z0-9._:-]+$/.test(correlationId)) fail("--correlation-id contains unsafe characters", 2); +if (!Number.isInteger(waitSeconds) || waitSeconds < 0 || waitSeconds > 14_400) fail("--wait must be an integer from 0 to 14400", 2); + +const root = values.root || process.env.CLAUDE_SESSION_ROOT || join(process.env.CLAUDE_CONFIG_DIR || join(homedir(), ".claude"), "projects"); +const token = createHash("sha256").update(correlationId).digest("hex").slice(0, 16); +const begin = `[[R:${token}]]`; +const end = `[[/R:${token}]]`; +const deadline = Date.now() + waitSeconds * 1000; +const pollMs = Number(process.env.AGENT_TRANSCRIPT_RESULT_POLL_MS || 5000); + +while (true) { + const outcome = await collect(); + if (outcome.kind === "found") { + if (!outcome.text) fail("Agent produced no textual result.", 65); + writeAll(`${outcome.text}\n`); + process.exit(0); + } + if (Date.now() >= deadline) { + if (outcome.kind === "ambiguous") fail("Correlated result markers could not be parsed.", 67); + fail("No correlated result in the session transcript yet.", 124); + } + await new Promise((resolveSleep) => setTimeout(resolveSleep, Math.min(pollMs, Math.max(1, deadline - Date.now())))); +} + +async function collect() { + const path = await transcriptPath(); + if (path === undefined) return { kind: "missing" }; + let raw; + try { raw = await readFile(path, "utf8"); } + catch { return { kind: "missing" }; } + const texts = []; + for (const line of raw.split("\n")) { + if (!line.includes(token)) continue; + let record; + try { record = JSON.parse(line); } + catch { continue; } + if (record?.type !== "assistant" || !Array.isArray(record?.message?.content)) continue; + for (const part of record.message.content) { + if (part?.type === "text" && typeof part.text === "string" && part.text.includes(begin)) texts.push(part.text); + } + } + if (texts.length === 0) return { kind: "missing" }; + // exactly one text block with exactly one marker pair, anything else isn't trustworthy + if (texts.length > 1) return { kind: "ambiguous" }; + const text = texts[0]; + const start = text.indexOf(begin); + const finish = text.indexOf(end, start + begin.length); + if (finish < 0 || text.indexOf(begin, start + begin.length) >= 0 || text.indexOf(end, finish + end.length) >= 0) { + return { kind: "ambiguous" }; + } + return { kind: "found", text: text.slice(start + begin.length, finish).trim() }; +} + +async function transcriptPath() { + let entries; + try { entries = await readdir(root, { recursive: true, withFileTypes: true }); } + catch { return undefined; } + const match = entries.find((entry) => entry.isFile() && entry.name === `${sessionId}.jsonl`); + return match === undefined ? undefined : join(match.parentPath, match.name); +} + +// process.exit() right after an async stdout write cuts piped output at 8 KiB. +function writeAll(text) { + const buffer = Buffer.from(text); + let offset = 0; + while (offset < buffer.length) { + try { offset += writeSync(1, buffer, offset); } + catch (error) { if (error.code !== "EAGAIN") throw error; } + } +} + +function fail(message, code) { + process.stderr.write(`ERROR: ${message}\n`); + process.exit(code); +}