Skip to content

fix(gateway): recover long correlated Claude results from the transcript #59

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Yehonal wants to merge 1 commit into
base: main
Choose a base branch
from
Open

fix(gateway): recover long correlated Claude results from the transcript #59

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 17 additions & 8 deletions packages/gateway/docs/mcp-ingress.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -178,6 +178,7 @@ route:
"meshSocket": "mesh-ingress",
"timeoutSeconds": 180,
"scanLimit": 500,
"resultRecoverySeconds": 3600,
"providers": [
{
"target_agent_id": "agent.ingress.codex",
Expand Down Expand Up @@ -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
`<session-id>.jsonl` under `agentManagedInboxRoot`; delivery additionally
Expand Down Expand Up @@ -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
Expand Down
60 changes: 31 additions & 29 deletions packages/gateway/src/adapters/agent-session-provider.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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";

Expand Down Expand Up @@ -171,6 +170,8 @@ export interface ShellAgentSessionProviderOptions {
workspaceRoots: Readonly<Record<string, readonly string[]>>;
/** Trusted cwd per workspace for fresh sessions, must be inside that workspace's roots. */
freshSessionCwds?: Readonly<Record<string, string>>;
/** Seconds to keep polling the Claude transcript for a reply the collector missed. 0 = check once. */
resultRecoverySeconds?: number;
meshSocket?: string;
timeoutSeconds?: number;
scanLimit?: number;
Expand All @@ -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<string, Promise<void>>();
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -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<string | undefined> {
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<TmuxSendResult> {
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;
}

Expand Down Expand Up @@ -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";
Expand All @@ -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));
Expand Down
8 changes: 6 additions & 2 deletions packages/gateway/src/mcp/serve.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -81,6 +81,7 @@ interface AgentSessionsConfig {
meshSocket?: string;
timeoutSeconds?: number;
scanLimit?: number;
resultRecoverySeconds?: number;
providers: Array<{
target_agent_id: string;
agent_type: "codex" | "claude";
Expand Down Expand Up @@ -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.");
}
Expand Down Expand Up @@ -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 })
})));
}

Expand Down
53 changes: 39 additions & 14 deletions packages/gateway/test/agent-session-provider.test.js
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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";
Expand Down Expand Up @@ -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({
Expand All @@ -339,20 +336,50 @@ 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(" ")}`);
}
});
const result = await instance.send({
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 () => {
Expand All @@ -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(" ")}`);
}
});
Expand Down
Loading
Loading