diff --git a/apps/agent-orchestrator/src/agent/graph.test.ts b/apps/agent-orchestrator/src/agent/graph.test.ts index 9e68542..3e94aa6 100644 --- a/apps/agent-orchestrator/src/agent/graph.test.ts +++ b/apps/agent-orchestrator/src/agent/graph.test.ts @@ -14,7 +14,13 @@ import type { ToolDescriptor } from "../tool-descriptor.js"; import type { SkillFitChecker } from "./skill-fit-checker.js"; import type { AgentDescriptor, AgentStore } from "../agents/types.js"; import type { DelegateSelector } from "./delegate-selector.js"; -import { AgentTurnFailedError, type AgentOrchestratorChannel, type AgentTurnResult } from "../agents/nats-agent-channel.js"; +import { + AgentTurnFailedError, + AgentTurnTimeoutError, + AgentTurnTransportError, + type AgentOrchestratorChannel, + type AgentTurnResult, +} from "../agents/nats-agent-channel.js"; import type { AgentRunLauncherPort } from "../k8s/agentrun-launcher.js"; import type { ToolFitChecker } from "./tool-fit-checker.js"; import type { BestEffortResponder } from "./best-effort-responder.js"; @@ -3434,3 +3440,264 @@ describe("credentials never reach the model (docs/adr/0030 §3)", () => { expect(serializedState).not.toContain(SECRET_CREDS); }); }); + +/** + * Resumability of an agent turn whose wait is interrupted by something that has + * nothing to do with the agent — an orchestrator rollout, a dropped NATS + * connection. The run keeps working in its own Job pod and holds its concluding + * message for us, so the correct behaviour is to pause, keep an anchor, and + * re-attach on the next turn — never to report the run as failed. + */ +describe("buildAgentGraph agent-turn resumability", () => { + const swe: AgentDescriptor = { + id: "claude-code-swe", + name: "claude-code-swe", + description: "Does software engineering tasks", + allowedRoles: ["reader"], + agentRunTemplate: { namespace: "default", agentRef: "claude-code-swe" }, + }; + + function resumeDeps(overrides: Partial = {}) { + const agentStore: AgentStore = { + upsert: vi.fn(), + query: vi.fn().mockResolvedValue([{ agent: swe, score: 0.9 }]), + getByIds: vi.fn().mockResolvedValue([{ agent: swe, score: 1 }]), + }; + const delegateSelector: DelegateSelector = { + select: vi.fn().mockResolvedValue({ type: "agent", agent: swe }), + }; + const agentChannel: AgentOrchestratorChannel = { + awaitReply: vi.fn().mockResolvedValue({ message: "done", final: true, narration: [] } satisfies AgentTurnResult), + sendPrompt: vi.fn(), + close: vi.fn(), + }; + const agentRunLauncher: AgentRunLauncherPort = { + launch: vi.fn().mockResolvedValue({ name: "run-1", namespace: "default" }), + }; + return baseDeps({ + agentStore, + delegateSelector, + agentChannel, + agentRunLauncher, + callbackBaseUrl: "http://orchestrator", + callbackSecretRef: { name: "secret", key: "token" }, + markAgentRunAwaitingReply: vi.fn().mockResolvedValue(undefined), + clearAgentRunAwaitingReply: vi.fn().mockResolvedValue(undefined), + ...overrides, + }); + } + + it("anchors the run to the conversation BEFORE waiting, so a killed pod still leaves something to resume", async () => { + const deps = resumeDeps(); + const graph = buildAgentGraph(deps); + + await graph.invoke({ request: "fix the bug", authToken: "tok", sessionId: "session-1" }); + + // Written during the turn, not with its outcome: the failure mode this + // covers is the process not surviving to produce an outcome at all. + expect(deps.markAgentRunAwaitingReply).toHaveBeenCalledWith("session-1", { + subject: "alice", + agentId: "claude-code-swe", + agentRunId: expect.any(String), + }); + }); + + it("reports a lost channel as a resumable pause, not a failure", async () => { + const deps = resumeDeps(); + (deps.agentChannel!.awaitReply as ReturnType).mockRejectedValue( + new AgentTurnTransportError("lost the NATS subscription for agent run run-1 before it replied"), + ); + const graph = buildAgentGraph(deps); + + const final = await graph.invoke({ request: "fix the bug", authToken: "tok", sessionId: "session-1" }); + + expect(final.error).toBeUndefined(); + expect(final.agentResumePending).toBe(true); + expect(final.result).toMatch(/Still working/); + // The anchor's survival is what the next turn re-attaches on, so the + // selected agent has to come back with it (the server persists both). + expect(final.selectedAgent?.id).toBe("claude-code-swe"); + }); + + it("re-attaches WITHOUT sending a prompt when the conversation is owed a reply", async () => { + const deps = resumeDeps(); + const graph = buildAgentGraph(deps); + + const final = await graph.invoke({ + request: "any update?", + authToken: "tok", + sessionId: "session-1", + activeAgentId: "claude-code-swe", + activeAgentRunId: "run-1", + activeAgentRunAwaitingReply: true, + sessionSubject: "alice", + }); + + // The agent is working, not waiting for input: publishing "any update?" as a + // prompt would inject it into the run's conversation. + expect(deps.agentChannel!.sendPrompt).not.toHaveBeenCalled(); + expect(deps.agentChannel!.awaitReply).toHaveBeenCalledWith("run-1", expect.anything()); + expect(final.result).toBe("done"); + expect(final.error).toBeUndefined(); + }); + + it("still sends a prompt when the run is parked on a question (ordinary HITL)", async () => { + const deps = resumeDeps(); + const graph = buildAgentGraph(deps); + + await graph.invoke({ + request: "use the main branch", + authToken: "tok", + sessionId: "session-1", + activeAgentId: "claude-code-swe", + activeAgentRunId: "run-1", + sessionSubject: "alice", + }); + + expect(deps.agentChannel!.sendPrompt).toHaveBeenCalledWith("run-1", "use the main branch"); + }); + + it("bounds a re-attached wait far more tightly than a live one", async () => { + const deps = resumeDeps({ agentIdleTimeoutSeconds: 600 }); + const graph = buildAgentGraph(deps); + + await graph.invoke({ + request: "any update?", + authToken: "tok", + sessionId: "session-1", + activeAgentId: "claude-code-swe", + activeAgentRunId: "run-1", + activeAgentRunAwaitingReply: true, + sessionSubject: "alice", + }); + + // Silence is diagnostic on a re-attach (anything alive announces itself + // within seconds), so the user is not made to wait out the full window to + // learn the run is gone. + const opts = (deps.agentChannel!.awaitReply as ReturnType).mock.calls[0]?.[1] as { + idleTimeoutMs?: number; + }; + expect(opts.idleTimeoutMs).toBe(45_000); + }); + + it("drops the anchor when a re-attached wait finds the run gone", async () => { + const deps = resumeDeps(); + (deps.agentChannel!.awaitReply as ReturnType).mockRejectedValue( + new AgentTurnTimeoutError("agent run run-1 went silent for 45000ms after 0 progress message(s)"), + ); + const graph = buildAgentGraph(deps); + + const final = await graph.invoke({ + request: "any update?", + authToken: "tok", + sessionId: "session-1", + activeAgentId: "claude-code-swe", + activeAgentRunId: "run-1", + activeAgentRunAwaitingReply: true, + sessionSubject: "alice", + }); + + // Otherwise every later turn in this conversation would re-attach to a dead + // run and spend its whole window before falling through. + expect(deps.clearAgentRunAwaitingReply).toHaveBeenCalledWith("session-1"); + expect(final.agentResumePending).toBeFalsy(); + expect(final.result).toMatch(/no longer reachable/); + }); +}); + +/** + * The label-triggered path's half of resumability. A matched IntegrationRoute + * dispatches deterministically, skipping retrieval — which also skipped the + * re-attach check, so a re-applied trigger label launched a SECOND run while the + * first was still holding the answer (docs/adr/0033). + */ +describe("buildAgentGraph route-driven turns re-attach before dispatching again", () => { + const routedAgent: AgentDescriptor = { + id: "claude-code-swe", + name: "claude-code-swe", + description: "Does software engineering tasks", + allowedRoles: ["reader"], + agentRunTemplate: { namespace: "default", agentRef: "claude-code-swe" }, + }; + + function routeDeps(overrides: Partial = {}) { + const agentStore: AgentStore = { + upsert: vi.fn(), + query: vi.fn().mockResolvedValue([]), + getByIds: vi.fn().mockResolvedValue([{ agent: routedAgent }]), + }; + const agentChannel: AgentOrchestratorChannel = { + awaitReply: vi.fn().mockResolvedValue({ message: "the held answer", final: true, narration: [] } satisfies AgentTurnResult), + sendPrompt: vi.fn(), + close: vi.fn(), + }; + const agentRunLauncher: AgentRunLauncherPort = { + launch: vi.fn().mockResolvedValue({ name: "run-2", namespace: "default" }), + }; + return baseDeps({ + agentStore, + agentChannel, + agentRunLauncher, + callbackBaseUrl: "http://orchestrator", + callbackSecretRef: { name: "secret", key: "token" }, + markAgentRunAwaitingReply: vi.fn().mockResolvedValue(undefined), + clearAgentRunAwaitingReply: vi.fn().mockResolvedValue(undefined), + ...overrides, + }); + } + + it("collects the held reply instead of launching a second run", async () => { + const deps = routeDeps(); + const graph = buildAgentGraph(deps); + + const final = await graph.invoke({ + request: "triage and resolve this issue", + authToken: "tok", + forcedAgentId: "claude-code-swe", + sessionId: "github:acme/widgets#42", + activeAgentId: "claude-code-swe", + activeAgentRunId: "run-1", + activeAgentRunAwaitingReply: true, + sessionSubject: "alice", + }); + + expect(deps.agentChannel!.awaitReply).toHaveBeenCalledWith("run-1", expect.anything()); + // The point: no second AgentRun. On a real coding agent that would be a + // second branch and a second PR for one request. + expect(deps.agentRunLauncher!.launch).not.toHaveBeenCalled(); + expect(deps.agentChannel!.sendPrompt).not.toHaveBeenCalled(); + expect(final.result).toBe("the held answer"); + }); + + it("still dispatches the route's own target when the anchor is stale", async () => { + const deps = routeDeps({ + // Anchor points at a run whose Agent CR is gone/revoked -> the re-attach + // misses, and the turn must fall back to the route's target rather than + // silently becoming an ordinary retrieval turn. + agentStore: { + upsert: vi.fn(), + query: vi.fn().mockResolvedValue([]), + getByIds: vi + .fn() + .mockResolvedValueOnce([{ agent: routedAgent }]) // checkIntegrationRoute's own lookup + .mockResolvedValueOnce([]), // checkActiveAgentRun's re-verify: gone + } as AgentStore, + }); + const graph = buildAgentGraph(deps); + + const final = await graph.invoke({ + request: "triage and resolve this issue", + authToken: "tok", + forcedAgentId: "claude-code-swe", + sessionId: "github:acme/widgets#42", + activeAgentId: "claude-code-swe", + activeAgentRunId: "run-1", + activeAgentRunAwaitingReply: true, + sessionSubject: "alice", + }); + + expect(deps.agentRunLauncher!.launch).toHaveBeenCalled(); + expect(deps.skillStore.query).not.toHaveBeenCalled(); + expect(final.selectedAgent?.id).toBe("claude-code-swe"); + }); +}); diff --git a/apps/agent-orchestrator/src/agent/graph.ts b/apps/agent-orchestrator/src/agent/graph.ts index 2517398..a110710 100644 --- a/apps/agent-orchestrator/src/agent/graph.ts +++ b/apps/agent-orchestrator/src/agent/graph.ts @@ -102,6 +102,28 @@ export const AgentStateAnnotation = Annotation.Root({ reducer: (_current, update) => update, default: () => undefined, }), + /** + * True when `activeAgentRunId` still owes this conversation a reply, rather + * than being parked on a question (see + * `SessionRecord.activeAgentRunAwaitingReply`). Decides whether + * `checkActiveAgentRun` publishes this turn's text as a `prompt` or simply + * re-attaches and waits. + */ + activeAgentRunAwaitingReply: Annotation({ + reducer: (_current, update) => update, + default: () => undefined, + }), + /** + * Set when a turn ended without the agent's reply through no fault of the + * agent's — the wait lost its channel while the run was still working. Tells + * the server to LEAVE the awaiting-reply anchor in place (the default for a + * turn that produced no reply is to clear it) so the next turn re-attaches + * instead of starting over. + */ + agentResumePending: Annotation({ + reducer: (_current, update) => update, + default: () => false, + }), /** * Per-tool continuation tokens from the caller's session, keyed by tool id * (docs/adr/0017). Set by the server from the session store, consumed by @@ -517,6 +539,19 @@ export interface AgentGraphDeps { * see `agentAwaitReplyIdleTimeoutMs`. */ agentIdleTimeoutSeconds?: number; + /** + * Records, before the wait begins, that this conversation is owed a reply by + * a specific AgentRun — the resume anchor a later turn re-attaches to + * (`session/inflight-agent-run.ts`). Optional: without a session store there + * is nowhere to persist it and nothing to resume from, so agent turns behave + * exactly as they did before resumability existed. + */ + markAgentRunAwaitingReply?: ( + sessionId: string, + run: { subject: string; agentId: string; agentRunId: string }, + ) => Promise; + /** Drops the resume anchor once there is nothing left to wait for (same module). */ + clearAgentRunAwaitingReply?: (sessionId: string) => Promise; /** * k8s Secret name/key the AgentRun CR's (currently vestigial) callback * field references — reuses the same secretRef as ToolRun. Required @@ -875,6 +910,33 @@ async function handleAgentTurnFailure( return { error: agentTurnErrorMessage(err) }; } +/** + * Outcome for a turn whose wait lost its channel while the agent was still + * working: a RESUMABLE pause, not a failure. + * + * This is the honest reading of an `AgentTurnTransportError`. Nothing about the + * run went wrong — the orchestrator was rolled, killed, or lost NATS — and the + * agent holds its concluding message for exactly this case (the protocol's + * `reply_ack`), re-offering it until someone collects it. So the turn reports a + * pause and leaves the resume anchor in place, and the next turn re-attaches + * and returns the real answer. + * + * Reported as `result` rather than `error` deliberately: an error reads as "your + * request failed, try again", which would be the third variation on telling a + * user their successful run failed. + */ +function resumableAgentTurnOutcome(state: Pick, agentRunId: string): Partial { + const nudge = state.progressListener + ? "Send any message and I'll pick up its reply where this left off." + : "Re-apply the label and I'll pick up its reply where this left off."; + return { + agentResumePending: true, + result: + `⏳ Still working — I lost my connection to agent run \`${agentRunId}\`, not the run itself. ` + + `It's still going (or already finished) and is holding its answer for me. ${nudge}`, + }; +} + /** * When a live progress listener is attached (the SSE streaming path), the * delegated agent's narrative was already streamed to the user as it was @@ -937,6 +999,23 @@ function agentAwaitReplyIdleTimeoutMs(deps: Pick { @@ -1213,7 +1294,15 @@ export function buildAgentGraph(deps: AgentGraphDeps) { sessionId: state.sessionId, }), }); - await deps.agentChannel.sendPrompt(state.activeAgentRunId, state.request); + // RE-ATTACH vs CONTINUE. Parked on a question -> this turn's text is the + // answer, publish it. Owed a reply (a previous turn's wait lost its + // channel) -> the agent is not waiting for input and this text is not an + // instruction, so say nothing and just collect the reply it is holding + // for us. Prompting here would inject "any update?" into a working + // agent's conversation. + if (!state.activeAgentRunAwaitingReply) { + await deps.agentChannel.sendPrompt(state.activeAgentRunId, state.request); + } const reply = await awaitReply; const message = composeAgentTurnMessage(state, reply); return { @@ -1229,6 +1318,31 @@ export function buildAgentGraph(deps: AgentGraphDeps) { : {}), }; } catch (err) { + if (err instanceof AgentTurnTransportError) { + return { + agentRunId: state.activeAgentRunId, + selectedAgent: found.agent, + ...resumableAgentTurnOutcome(state, state.activeAgentRunId), + }; + } + // Silence on a RE-ATTACH means something different from silence on a + // live turn, and is bounded far more tightly (see + // `agentAwaitReplyIdleTimeoutMs`): a run still working would have + // narrated, and one still holding an answer would have re-offered it + // within seconds. So this is the unrecoverable case -- the agent exited + // during the gap and its concluding message went with it. Drop the + // anchor so later turns don't each re-attach to a dead run, and say what + // is actually knowable. + if (err instanceof AgentTurnTimeoutError && state.activeAgentRunAwaitingReply) { + if (state.sessionId) await deps.clearAgentRunAwaitingReply?.(state.sessionId); + return { + agentRunId: state.activeAgentRunId, + selectedAgent: found.agent, + result: + `⚠️ Agent run \`${state.activeAgentRunId}\` is no longer reachable, so I couldn't recover the reply it was ` + + "working on. The run itself may well have completed — check it directly before re-running the request.", + }; + } return { agentRunId: state.activeAgentRunId, ...(await handleAgentTurnFailure(err, deps, state, found.agent)), @@ -1393,6 +1507,26 @@ export function buildAgentGraph(deps: AgentGraphDeps) { ...(identitySecretEnv ? { secretEnv: identitySecretEnv } : {}), ...(state.sessionId ? { sessionId: state.sessionId } : {}), }); + // The run now exists and we are about to wait on it. Anchor it to the + // conversation HERE, not with the rest of the turn's outcome: from this + // line until the reply arrives is exactly the window in which the + // orchestrator can be rolled out from under the wait, and an outcome + // persisted after the graph returns is precisely what a killed pod never + // gets to write. Best-effort -- a session store hiccup should cost + // resumability, not the turn. + if (state.sessionId && deps.markAgentRunAwaitingReply) { + await deps + .markAgentRunAwaitingReply(state.sessionId, { + subject: identity.subject, + agentId: agent.id, + agentRunId: runId, + }) + .catch((err: unknown) => { + console.error( + `[agent] failed to record agent run ${runId} as awaiting reply; a rollout mid-turn will not be resumable: ${err instanceof Error ? err.message : String(err)}`, + ); + }); + } const reply = await awaitReply; const message = composeAgentTurnMessage(state, reply); return { @@ -1409,6 +1543,9 @@ export function buildAgentGraph(deps: AgentGraphDeps) { : {}), }; } catch (err) { + if (err instanceof AgentTurnTransportError) { + return { agentRunId: runId, identity, selectedAgent: agent, ...resumableAgentTurnOutcome(state, runId) }; + } return { agentRunId: runId, identity, ...(await handleAgentTurnFailure(err, deps, { ...state, identity }, agent)) }; } }) @@ -1729,11 +1866,21 @@ export function buildAgentGraph(deps: AgentGraphDeps) { .addConditionalEdges("checkIntegrationRoute", (state) => state.error ? END - : state.selectedAgent - ? "delegateToAgent" - : state.selectedSkill - ? "loadSkillTools" - : "checkPendingIdentityLink", + : // A conversation still owed a reply by a run re-attaches FIRST, even on + // a route-driven turn. The route's whole purpose is to skip retrieval + // and dispatch deterministically, but dispatching again while an + // earlier run of this same conversation is holding an answer would do + // the work twice -- a second branch and a second PR on a real coding + // agent. This is the path a re-applied trigger label takes + // (docs/adr/0033), and `checkActiveAgentRun` falls back to the route's + // own target below if the anchor turns out to be stale. + state.activeAgentRunAwaitingReply && state.activeAgentRunId + ? "checkActiveAgentRun" + : state.selectedAgent + ? "delegateToAgent" + : state.selectedSkill + ? "loadSkillTools" + : "checkPendingIdentityLink", ) // A pending device-flow link either just completed (an agent was // re-selected -> resume straight into delegateToAgent), is still being @@ -1749,11 +1896,21 @@ export function buildAgentGraph(deps: AgentGraphDeps) { .addConditionalEdges("checkActiveSkill", (state) => state.error ? END : state.selectedSkill ? "loadSkillTools" : "checkActiveAgentRun", ) - // A continuing agent run either produced a terminal turn result - // (question or final reply, agentRunId set) or errored -> END either - // way; a miss (agentRunId still unset, e.g. no active run or agent - // delegation not configured) falls through to full retrieval. - .addConditionalEdges("checkActiveAgentRun", (state) => (state.error || state.agentRunId ? END : "checkNeedsCapability")) + // A continuing agent run either produced a terminal turn result (question, + // final reply, or resumable pause -- all set `agentRunId`) or errored -> END + // either way. On a miss, an already-selected target is honored before + // falling through to retrieval: this node is now also reachable from + // `checkIntegrationRoute`, whose route already chose one, and dropping that + // choice would turn a stale anchor into a silently different turn. + .addConditionalEdges("checkActiveAgentRun", (state) => + state.error || state.agentRunId + ? END + : state.selectedAgent + ? "delegateToAgent" + : state.selectedSkill + ? "loadSkillTools" + : "checkNeedsCapability", + ) // A "no" (docs/adr/0019) skips catalog retrieval entirely and answers // directly; a "yes" (or classifier error) proceeds exactly as before. .addConditionalEdges("checkNeedsCapability", (state) => diff --git a/apps/agent-orchestrator/src/agents/nats-agent-channel.test.ts b/apps/agent-orchestrator/src/agents/nats-agent-channel.test.ts index 78084e9..146ae3d 100644 --- a/apps/agent-orchestrator/src/agents/nats-agent-channel.test.ts +++ b/apps/agent-orchestrator/src/agents/nats-agent-channel.test.ts @@ -180,6 +180,89 @@ describe("NatsAgentChannel", () => { expect(down).toEqual([expect.objectContaining({ type: "tool_result", callId: "c1", ok: false, error: "tool exploded" })]); sub.unsubscribe(); }); + + /** + * The orchestrator's half of surviving its own disappearance: the agent holds + * its concluding message until this ack lands, so failing to send one would + * leave every finished run re-offering an answer nobody ever confirmed. + */ + it("acks a reply on the down subject, quoting that message's seq", async () => { + const { channel, nc } = makeChannel(); + const codec = JSONCodec(); + const down: unknown[] = []; + const sub = nc.subscribe("agent.run-1.down"); + void (async () => { + for await (const m of sub) down.push(codec.decode(m.data)); + })(); + + const pending = channel.awaitReply("run-1"); + await new Promise((r) => setImmediate(r)); + const codecUp = JSONCodec(); + nc.publish( + "agent.run-1.up", + codecUp.encode({ agent_run_id: "run-1", seq: 7, ts: "2026-07-13T00:00:00.000Z", type: "reply", message: "done", final: true }), + ); + + await expect(pending).resolves.toMatchObject({ message: "done" }); + expect(down).toEqual([expect.objectContaining({ type: "reply_ack", ackSeq: 7 })]); + sub.unsubscribe(); + }); + + it("acks a failed message too, so a failure is not re-offered forever", async () => { + const { channel, nc } = makeChannel(); + const codec = JSONCodec(); + const down: unknown[] = []; + const sub = nc.subscribe("agent.run-1.down"); + void (async () => { + for await (const m of sub) down.push(codec.decode(m.data)); + })(); + + const pending = channel.awaitReply("run-1"); + await new Promise((r) => setImmediate(r)); + nc.publish( + "agent.run-1.up", + JSONCodec().encode({ + agent_run_id: "run-1", + seq: 3, + ts: "2026-07-13T00:00:00.000Z", + type: "failed", + code: "agent_error", + message: "boom", + }), + ); + + await expect(pending).rejects.toBeInstanceOf(AgentTurnFailedError); + expect(down).toEqual([expect.objectContaining({ type: "reply_ack", ackSeq: 3 })]); + sub.unsubscribe(); + }); + + it("acks a non-final reply (a question) as well", async () => { + const { channel, nc } = makeChannel(); + const codec = JSONCodec(); + const down: unknown[] = []; + const sub = nc.subscribe("agent.run-1.down"); + void (async () => { + for await (const m of sub) down.push(codec.decode(m.data)); + })(); + + const pending = channel.awaitReply("run-1"); + await new Promise((r) => setImmediate(r)); + nc.publish( + "agent.run-1.up", + JSONCodec().encode({ + agent_run_id: "run-1", + seq: 2, + ts: "2026-07-13T00:00:00.000Z", + type: "reply", + message: "Which branch?", + final: false, + }), + ); + + await expect(pending).resolves.toMatchObject({ final: false }); + expect(down).toEqual([expect.objectContaining({ type: "reply_ack", ackSeq: 2 })]); + sub.unsubscribe(); + }); }); /** diff --git a/apps/agent-orchestrator/src/agents/nats-agent-channel.ts b/apps/agent-orchestrator/src/agents/nats-agent-channel.ts index 653eb99..8f6e053 100644 --- a/apps/agent-orchestrator/src/agents/nats-agent-channel.ts +++ b/apps/agent-orchestrator/src/agents/nats-agent-channel.ts @@ -33,6 +33,11 @@ export class AgentTurnTimeoutError extends Error {} * because these were previously conflated, producing the actively misleading * "produced no reply within 3660000ms" on a run that was healthy and went on * to succeed — the number was the *configured* bound, never the elapsed time. + * + * Callers treat this as a RESUMABLE pause rather than a failure (docs/adr/0033): + * the agent holds its concluding message until acked, so the next turn can + * re-attach and collect it. Reporting it honestly was the previous fix; not + * losing the answer is this one. */ export class AgentTurnTransportError extends Error {} export class AgentTurnFailedError extends Error { @@ -269,9 +274,17 @@ export class NatsAgentChannel implements AgentOrchestratorChannel { opts.onProgress?.("warning", msg.message); break; case "reply": + // Ack BEFORE unsubscribing/returning: the agent holds its + // concluding message until this lands (see the protocol's + // `reply_ack`), re-offering it meanwhile, and a re-offer arriving + // after we unsubscribe would be dropped. A non-final reply (a HITL + // question) is acked too -- losing a question strands the + // conversation exactly the way losing an answer does. + this.ackConcluding(agentRunId, msg.seq); sub.unsubscribe(); return { message: msg.message, final: msg.final, result: msg.result, narration }; case "failed": + this.ackConcluding(agentRunId, msg.seq); sub.unsubscribe(); throw new AgentTurnFailedError(msg.code, msg.message); case "tool_call": @@ -321,6 +334,23 @@ export class NatsAgentChannel implements AgentOrchestratorChannel { } } + /** + * Confirms receipt of a concluding up-message so the agent can stop holding + * it (see the protocol's `reply_ack`). Fire-and-forget by design: if this ack + * is itself lost the agent simply re-offers, and the next receipt acks again. + */ + private ackConcluding(agentRunId: string, ackSeq: number): void { + const { down } = agentSubjects(agentRunId, this.subjectPrefix); + const msg: AgentDownMessage = { + type: "reply_ack", + ackSeq, + agent_run_id: agentRunId, + seq: this.seq++, + ts: new Date().toISOString(), + }; + this.nc.publish(down, this.codec.encode(msg)); + } + async sendPrompt(agentRunId: string, message: string): Promise { const { down } = agentSubjects(agentRunId, this.subjectPrefix); const msg: AgentDownMessage = { diff --git a/apps/agent-orchestrator/src/callback/receiver.test.ts b/apps/agent-orchestrator/src/callback/receiver.test.ts index 30c592f..f0d5a0a 100644 --- a/apps/agent-orchestrator/src/callback/receiver.test.ts +++ b/apps/agent-orchestrator/src/callback/receiver.test.ts @@ -2,6 +2,21 @@ import { createHmac } from "node:crypto"; import { describe, expect, it } from "vitest"; import { CallbackAuthError, CallbackReceiver, verifyAndParseCallback } from "./receiver.js"; +/** + * Every request in this file goes over a fresh TCP connection — see the long + * explanation on the identical shim in agent-orchestrator's `server.test.ts`. + * Short version: these tests listen on ephemeral ports and close their servers, + * and Node's `fetch` caches keep-alive sockets per origin, so a recycled port + * can hand a later test a dead socket ("other side closed", zero bytes read). + * `connection: close` keeps anything from being pooled in the first place. + */ +const nativeFetch = globalThis.fetch; +const fetch: typeof globalThis.fetch = (input, init = {}) => { + const headers = new Headers(init.headers); + headers.set("connection", "close"); + return nativeFetch(input, { ...init, headers }); +}; + const SECRET = "shh"; function sign(body: string, secret = SECRET): string { diff --git a/apps/agent-orchestrator/src/index.ts b/apps/agent-orchestrator/src/index.ts index df30387..640cde0 100644 --- a/apps/agent-orchestrator/src/index.ts +++ b/apps/agent-orchestrator/src/index.ts @@ -39,6 +39,7 @@ import { OpenAiTaskCompleter } from "./openai/task-completer.js"; import { InMemorySessionStore } from "./session/in-memory-session-store.js"; import { RedisSessionStore } from "./session/redis-session-store.js"; import type { SessionStore } from "./session/types.js"; +import { clearAgentRunAwaitingReply, markAgentRunAwaitingReply } from "./session/inflight-agent-run.js"; import { InvokeServer } from "./server.js"; import { retryWithBackoff } from "./retry.js"; import type { ToolDescriptor } from "./tool-descriptor.js"; @@ -472,6 +473,35 @@ async function main(): Promise { secretReader: K8sSecretReader.fromKubeConfig(config.namespace, kubeConfig), }); + // Conversation-session store (docs/adr/0012): remembers each chat's active + // skill so follow-up turns skip RAG re-selection when the fit-check + // passes. Redis-backed (docs/adr/0016) when AGENT_REDIS_URL is set, so + // sessions survive restarts and are shared across replicas; otherwise + // falls back to the single-replica in-memory adapter. + // + // Built before the graph because the graph writes to it mid-turn now: the + // agent-run resume anchor (`session/inflight-agent-run.ts`) has to be + // persisted before a wait begins, not with the turn's outcome. + let redisSessionStore: RedisSessionStore | undefined; + let sessionStore: SessionStore; + if (config.redisUrl) { + redisSessionStore = new RedisSessionStore(config.redisUrl, { + ttlSeconds: config.sessionTtlSeconds, + }); + await retryWithBackoff("redis startup check", () => redisSessionStore!.connect(), { + attempts: 12, + initialDelayMs: 1_000, + maxDelayMs: 15_000, + }); + console.error(`Using Redis session store: ${config.redisUrl}`); + sessionStore = redisSessionStore; + } else { + sessionStore = new InMemorySessionStore({ + ttlMs: config.sessionTtlSeconds * 1000, + maxEntries: config.sessionMaxEntries, + }); + } + const graph = buildAgentGraph({ identityResolver, forwardedUserIdentityResolver, @@ -514,35 +544,17 @@ async function main(): Promise { agentTopK: config.agentTopK, agentRunTimeoutSeconds: config.agentRunTimeoutSeconds, agentIdleTimeoutSeconds: config.agentIdleTimeoutSeconds, + // Resume anchor for an agent turn whose wait is interrupted (a + // rollout, a lost NATS channel): written before the wait, read by the + // next turn's `checkActiveAgentRun` to re-attach instead of failing a + // run that is still working. + markAgentRunAwaitingReply: (sessionId, run) => markAgentRunAwaitingReply(sessionStore, sessionId, run), + clearAgentRunAwaitingReply: (sessionId) => clearAgentRunAwaitingReply(sessionStore, sessionId), callbackSecretRef: { name: config.callbackSecretRefName ?? "", key: config.callbackSecretRefKey }, } : {}), }); - // Conversation-session store (docs/adr/0012): remembers each chat's active - // skill so follow-up turns skip RAG re-selection when the fit-check - // passes. Redis-backed (docs/adr/0016) when AGENT_REDIS_URL is set, so - // sessions survive restarts and are shared across replicas; otherwise - // falls back to the single-replica in-memory adapter. - let redisSessionStore: RedisSessionStore | undefined; - let sessionStore: SessionStore; - if (config.redisUrl) { - redisSessionStore = new RedisSessionStore(config.redisUrl, { - ttlSeconds: config.sessionTtlSeconds, - }); - await retryWithBackoff("redis startup check", () => redisSessionStore!.connect(), { - attempts: 12, - initialDelayMs: 1_000, - maxDelayMs: 15_000, - }); - console.error(`Using Redis session store: ${config.redisUrl}`); - sessionStore = redisSessionStore; - } else { - sessionStore = new InMemorySessionStore({ - ttlMs: config.sessionTtlSeconds * 1000, - maxEntries: config.sessionMaxEntries, - }); - } // Answers Open WebUI's internal housekeeping completions (title/tags/query // generation) directly, bypassing the agent graph -- see server.ts's // handleInternalUiTask and isInternalUiTaskRequest. @@ -601,6 +613,13 @@ async function main(): Promise { // regardless: past the deadline we stop waiting and tear down anyway, so // a turn that outlives the grace period at least fails the same way it // would have, rather than blocking the other closers entirely. + // + // A drain window is therefore never enough on its own -- an agent turn takes + // minutes and this gives it seconds. What keeps the ANSWER is that the run's + // agent holds its concluding message until acked and the conversation was + // anchored to the run before the wait began, so the next turn re-attaches + // and collects it (docs/adr/0033). This drain just lets the turns that can + // finish, finish. const httpDrained = Promise.all([ invokeServer.close(), ...(callbackReceiver ? [callbackReceiver.close()] : []), diff --git a/apps/agent-orchestrator/src/server.test.ts b/apps/agent-orchestrator/src/server.test.ts index 73ce03b..1f764b1 100644 --- a/apps/agent-orchestrator/src/server.test.ts +++ b/apps/agent-orchestrator/src/server.test.ts @@ -5,6 +5,36 @@ import type { AgentState } from "./agent/graph.js"; import type { AgentOrchestratorChannel } from "./agents/nats-agent-channel.js"; import { InMemorySessionStore } from "./session/in-memory-session-store.js"; +/** + * Every request in this file goes over a fresh TCP connection. + * + * Without this the suite failed roughly one run in ten, always as + * `TypeError: fetch failed` / `SocketError: other side closed` on a request to a + * server that had just started -- with `bytesWritten: 339, bytesRead: 0`, i.e. + * the request was written to a socket whose peer had already gone. Nothing was + * wrong with the server: it is connection REUSE across tests. + * + * Node's `fetch` (undici) keeps a keep-alive pool keyed by origin + * (`127.0.0.1:`) for several seconds. Each test here listens on port 0, + * gets an ephemeral port, and closes the server when it finishes -- so under + * load, when the OS recycles an ephemeral port quickly enough, a later test's + * server can land on a port whose pooled (and now dead) socket is still cached. + * The next request to that origin is handed the corpse. + * + * `connection: close` opts every request out of pooling, so there is never a + * cached socket to inherit. Verified: three requests to one origin open three + * sockets with this header and reuse one without it. Declared at module scope so + * it shadows the global for the whole file -- no call site has to remember. + */ +const nativeFetch = globalThis.fetch; +const fetch: typeof globalThis.fetch = (input, init = {}) => { + // Via the Headers API rather than an object spread, so a caller passing + // Headers or an entry array keeps its headers instead of losing them. + const headers = new Headers(init.headers); + headers.set("connection", "close"); + return nativeFetch(input, { ...init, headers }); +}; + function listenOn(server: InvokeServer): Promise { return server.listen(0).then(() => { const address = server["server"]?.address(); @@ -307,6 +337,89 @@ describe("InvokeServer session-scoped pending identity link (GitHub OAuth Device return new InMemorySessionStore({ ttlMs: 60_000, maxEntries: 10 }); } + /** + * A turn that lost its channel to a still-working run must leave the + * awaiting-reply anchor behind. The default for a turn that produced no reply + * is to CLEAR the active-run state (the run concluded), which for this case + * would throw away the only pointer back to a run that is still holding the + * answer. + */ + it("keeps the awaiting-reply anchor when the turn ended with a lost channel", async () => { + const identity = { subject: "alice", roles: ["reader"] }; + const graph: AgentGraphLike = { + invoke: vi.fn().mockResolvedValue({ + request: "x", + authToken: "tok-1", + skillCandidates: [], + identity, + selectedAgent: { id: "claude-code-swe" }, + agentRunId: "run-1", + agentAwaitingReply: false, + agentResumePending: true, + result: "Still working -- I lost my connection to agent run `run-1`.", + } as unknown as AgentState), + stream: vi.fn(), + }; + const store = sessionStore(); + const server = new InvokeServer(graph, store); + const port = await listenOn(server); + + await fetch(`http://127.0.0.1:${port}/invoke`, { + method: "POST", + headers: { "content-type": "application/json", authorization: "Bearer tok-1" }, + body: JSON.stringify({ request: "fix the bug", session_id: "session-1" }), + }); + await new Promise((r) => setTimeout(r, 10)); + + expect(await store.get("session-1")).toMatchObject({ + subject: "alice", + activeAgentId: "claude-code-swe", + activeAgentRunId: "run-1", + activeAgentRunAwaitingReply: true, + }); + + await server.close(); + }); + + it("clears the anchor once the agent actually replies", async () => { + const identity = { subject: "alice", roles: ["reader"] }; + const graph: AgentGraphLike = { + invoke: vi.fn().mockResolvedValue({ + request: "x", + authToken: "tok-1", + skillCandidates: [], + identity, + selectedAgent: { id: "claude-code-swe" }, + agentRunId: "run-1", + agentAwaitingReply: false, + result: "opened a PR", + } as unknown as AgentState), + stream: vi.fn(), + }; + const store = sessionStore(); + await store.set("session-1", { + subject: "alice", + activeAgentId: "claude-code-swe", + activeAgentRunId: "run-1", + activeAgentRunAwaitingReply: true, + }); + const server = new InvokeServer(graph, store); + const port = await listenOn(server); + + await fetch(`http://127.0.0.1:${port}/invoke`, { + method: "POST", + headers: { "content-type": "application/json", authorization: "Bearer tok-1" }, + body: JSON.stringify({ request: "any update?", session_id: "session-1" }), + }); + await new Promise((r) => setTimeout(r, 10)); + + const record = await store.get("session-1"); + expect(record?.activeAgentRunId).toBeUndefined(); + expect(record?.activeAgentRunAwaitingReply).toBeUndefined(); + + await server.close(); + }); + it("persists pendingIdentityLink from a turn that paused on device-flow authorization, and offers it to the graph on the next turn", async () => { const identity = { subject: "alice", roles: ["reader"] }; const pendingIdentityLink = { diff --git a/apps/agent-orchestrator/src/server.ts b/apps/agent-orchestrator/src/server.ts index b0b47a6..6a8bc17 100644 --- a/apps/agent-orchestrator/src/server.ts +++ b/apps/agent-orchestrator/src/server.ts @@ -102,6 +102,8 @@ export interface AgentGraphInput { activeAgentId?: string; /** Name of the specific AgentRun CR the conversation is continuing, if any. */ activeAgentRunId?: string; + /** True when `activeAgentRunId` still owes this conversation a reply (see `SessionRecord.activeAgentRunAwaitingReply`). */ + activeAgentRunAwaitingReply?: boolean; /** Identity subject the session record was created under (docs/adr/0012). */ sessionSubject?: string; /** Per-tool continuation tokens from the caller's session, keyed by tool id (docs/adr/0017). */ @@ -293,6 +295,7 @@ export class InvokeServer { input.activeSkillId = record.activeSkillId; input.activeAgentId = record.activeAgentId; input.activeAgentRunId = record.activeAgentRunId; + input.activeAgentRunAwaitingReply = record.activeAgentRunAwaitingReply; input.sessionSubject = record.subject; input.toolContinuations = record.toolContinuations; input.agentContinuations = record.agentContinuations; @@ -323,6 +326,7 @@ export class InvokeServer { selectedAgent?: { id: string }; agentRunId?: string; agentAwaitingReply?: boolean; + agentResumePending?: boolean; extractedContinuation?: { toolId: string; token: string }; extractedAgentContinuation?: { agentId: string; token: string }; pendingIdentityLink?: { agentId: string; provider: string; flow: "device" | "authcode" | "page"; deviceCode?: string; expiresAt: number; subject?: string; request?: string }; @@ -370,7 +374,21 @@ export class InvokeServer { return; } if (outcome.agentRunId) { - if (outcome.agentAwaitingReply && outcome.selectedAgent) { + if (outcome.agentResumePending && outcome.selectedAgent) { + // The turn ended without the reply because the channel to the run was + // lost, not because the run concluded. Keep the anchor (already written + // eagerly before the wait) with the awaiting-reply flag intact, so the + // next turn re-attaches and waits rather than publishing its text as a + // prompt to an agent that never asked anything. + await this.sessionStore.set(sessionId, { + ...base, + activeAgentId: outcome.selectedAgent.id, + activeAgentRunId: outcome.agentRunId, + activeAgentRunAwaitingReply: true, + }); + } else if (outcome.agentAwaitingReply && outcome.selectedAgent) { + // Parked on a question: the next turn's text IS the answer, so no + // awaiting-reply flag -- it gets published as a prompt. await this.sessionStore.set(sessionId, { ...base, activeAgentId: outcome.selectedAgent.id, @@ -735,6 +753,7 @@ export class InvokeServer { selectedAgent: state.selectedAgent, agentRunId: state.agentRunId, agentAwaitingReply: state.agentAwaitingReply, + agentResumePending: state.agentResumePending, extractedContinuation: state.extractedContinuation, extractedAgentContinuation: state.extractedAgentContinuation, pendingIdentityLink: state.pendingIdentityLink, @@ -897,6 +916,7 @@ export class InvokeServer { selectedAgent: state.selectedAgent, agentRunId: state.agentRunId, agentAwaitingReply: state.agentAwaitingReply, + agentResumePending: state.agentResumePending, extractedContinuation: state.extractedContinuation, extractedAgentContinuation: state.extractedAgentContinuation, pendingIdentityLink: state.pendingIdentityLink, diff --git a/apps/agent-orchestrator/src/session/inflight-agent-run.test.ts b/apps/agent-orchestrator/src/session/inflight-agent-run.test.ts new file mode 100644 index 0000000..0010a6d --- /dev/null +++ b/apps/agent-orchestrator/src/session/inflight-agent-run.test.ts @@ -0,0 +1,96 @@ +import { describe, expect, it } from "vitest"; +import { InMemorySessionStore } from "./in-memory-session-store.js"; +import { clearAgentRunAwaitingReply, markAgentRunAwaitingReply } from "./inflight-agent-run.js"; + +function store(): InMemorySessionStore { + return new InMemorySessionStore({ ttlMs: 60_000, maxEntries: 10 }); +} + +describe("markAgentRunAwaitingReply", () => { + it("writes the anchor for a conversation with no record yet", async () => { + const s = store(); + + await markAgentRunAwaitingReply(s, "session-1", { subject: "alice", agentId: "swe", agentRunId: "run-1" }); + + expect(await s.get("session-1")).toMatchObject({ + subject: "alice", + activeAgentId: "swe", + activeAgentRunId: "run-1", + activeAgentRunAwaitingReply: true, + lastAgentRunId: "run-1", + }); + }); + + /** + * `SessionStore.set` replaces the whole record, so the read-modify-write is + * load-bearing: buying resumability by dropping a conversation's continuation + * tokens would trade one broken follow-up turn for another. + */ + it("preserves everything else already stored for the conversation", async () => { + const s = store(); + await s.set("session-1", { + subject: "alice", + toolContinuations: { "recipe-publisher": "slug-42" }, + agentContinuations: { swe: "branch-token" }, + }); + + await markAgentRunAwaitingReply(s, "session-1", { subject: "alice", agentId: "swe", agentRunId: "run-1" }); + + expect(await s.get("session-1")).toMatchObject({ + toolContinuations: { "recipe-publisher": "slug-42" }, + agentContinuations: { swe: "branch-token" }, + activeAgentRunId: "run-1", + }); + }); + + it("keeps the stored subject rather than overwriting it", async () => { + const s = store(); + await s.set("session-1", { subject: "alice" }); + + await markAgentRunAwaitingReply(s, "session-1", { subject: "bob", agentId: "swe", agentRunId: "run-1" }); + + // A conversation belongs to the subject it was created under (ADR 0012); + // this helper records a run, it does not re-own the session. + expect((await s.get("session-1"))?.subject).toBe("alice"); + }); + + it("clears any active skill, since a turn continues a skill or a run, never both", async () => { + const s = store(); + await s.set("session-1", { subject: "alice", activeSkillId: "publish-recipe" }); + + await markAgentRunAwaitingReply(s, "session-1", { subject: "alice", agentId: "swe", agentRunId: "run-1" }); + + expect((await s.get("session-1"))?.activeSkillId).toBeUndefined(); + }); +}); + +describe("clearAgentRunAwaitingReply", () => { + it("drops the anchor but keeps continuations", async () => { + const s = store(); + await s.set("session-1", { + subject: "alice", + activeAgentId: "swe", + activeAgentRunId: "run-1", + activeAgentRunAwaitingReply: true, + agentContinuations: { swe: "branch-token" }, + lastAgentRunId: "run-1", + }); + + await clearAgentRunAwaitingReply(s, "session-1"); + + const record = await s.get("session-1"); + expect(record?.activeAgentRunId).toBeUndefined(); + expect(record?.activeAgentId).toBeUndefined(); + expect(record?.activeAgentRunAwaitingReply).toBeUndefined(); + expect(record?.agentContinuations).toEqual({ swe: "branch-token" }); + // Kept deliberately (ADR 0026): a live-session viewer still needs a run id + // to probe, even one that has concluded. + expect(record?.lastAgentRunId).toBe("run-1"); + }); + + it("is a no-op for a conversation with no record", async () => { + const s = store(); + await clearAgentRunAwaitingReply(s, "session-nope"); + expect(await s.get("session-nope")).toBeUndefined(); + }); +}); diff --git a/apps/agent-orchestrator/src/session/inflight-agent-run.ts b/apps/agent-orchestrator/src/session/inflight-agent-run.ts new file mode 100644 index 0000000..5046597 --- /dev/null +++ b/apps/agent-orchestrator/src/session/inflight-agent-run.ts @@ -0,0 +1,63 @@ +import type { SessionStore } from "./types.js"; + +/** + * Records, mid-turn, that this conversation has an AgentRun in flight that it + * is owed a reply from — the anchor a later turn re-attaches to + * (`SessionRecord.activeAgentRunAwaitingReply`). + * + * Deliberately written BEFORE the wait rather than with the rest of the turn's + * outcome. `InvokeServer.persistSession` runs after the graph returns, which is + * fine for every outcome that has one; the failure this anchor exists for is the + * orchestrator pod going away mid-wait, where the graph never returns at all and + * a post-hoc write is exactly the write that doesn't happen. A rollout with a + * bounded drain is the survivable version of that, a SIGKILL past the grace + * period the unsurvivable one — both leave the anchor behind if it was written + * up front, and neither does if it wasn't. + * + * Read-modify-write because `SessionStore.set` replaces the whole record: + * everything already stored for the conversation (continuation tokens above + * all) has to be carried over, or resumability would be bought by throwing away + * cross-episode state. + */ +export async function markAgentRunAwaitingReply( + store: SessionStore, + sessionId: string, + run: { subject: string; agentId: string; agentRunId: string }, +): Promise { + const existing = await store.get(sessionId); + await store.set(sessionId, { + // `existing` carries an `updatedAt` the port's input type doesn't declare; + // spreading it is harmless (every adapter stamps its own) and keeps this a + // patch of whatever is stored rather than a rewrite of a subset of fields. + ...existing, + subject: existing?.subject ?? run.subject, + activeAgentId: run.agentId, + activeAgentRunId: run.agentRunId, + activeAgentRunAwaitingReply: true, + lastAgentRunId: run.agentRunId, + // Mutually exclusive with an active skill by the same rule the + // post-turn path follows (see `SessionRecord`): a turn continues a skill + // or an agent run, never both. + activeSkillId: undefined, + }); +} + +/** + * Drops the awaiting-reply anchor while leaving the rest of the record intact. + * + * Used when a re-attached wait establishes that there is nothing more to wait + * for — the reply arrived, or the run is terminal and its answer is + * unrecoverable. Without this the conversation would re-attach to a dead run on + * every subsequent turn, silently spending each one's whole idle window before + * falling through to ordinary handling. + */ +export async function clearAgentRunAwaitingReply(store: SessionStore, sessionId: string): Promise { + const existing = await store.get(sessionId); + if (!existing) return; + await store.set(sessionId, { + ...existing, + activeAgentId: undefined, + activeAgentRunId: undefined, + activeAgentRunAwaitingReply: undefined, + }); +} diff --git a/apps/agent-orchestrator/src/session/types.ts b/apps/agent-orchestrator/src/session/types.ts index f10965f..16d9fea 100644 --- a/apps/agent-orchestrator/src/session/types.ts +++ b/apps/agent-orchestrator/src/session/types.ts @@ -38,6 +38,27 @@ export interface SessionRecord { * Agent id, since one Agent can have many concurrent runs. */ activeAgentRunId?: string; + /** + * True when `activeAgentRunId` is a run this conversation is still owed a + * REPLY from, rather than one parked on a question awaiting the user's next + * prompt. + * + * The distinction decides what the next turn does. Parked-on-a-question is + * the ordinary HITL case: the turn's message is the answer, so it gets + * published as a `prompt`. Owed-a-reply means the previous turn's wait was + * cut short while the agent was still working -- the orchestrator pod was + * rolled or lost its NATS channel -- and the run is very likely still going + * or already finished. Sending it a `prompt` in that state would be wrong + * twice over: the agent isn't waiting for input, and the text (e.g. "any + * update?") isn't an instruction. So the next turn re-attaches to the run's + * subject and waits for the reply the agent is still holding for us (see the + * protocol's `reply_ack`). + * + * Written EAGERLY, before the wait starts, precisely because the failure it + * covers can be a hard pod kill: an outcome persisted only after the graph + * returns is exactly what does not survive a SIGKILL. + */ + activeAgentRunAwaitingReply?: boolean; /** * The most recent AgentRun id this conversation delegated to, kept even * after the run concludes and `activeAgentRunId` is cleared (ADR 0026) -- diff --git a/apps/integration-gateway/src/identity-link/api.test.ts b/apps/integration-gateway/src/identity-link/api.test.ts index 2af2bd3..66368b8 100644 --- a/apps/integration-gateway/src/identity-link/api.test.ts +++ b/apps/integration-gateway/src/identity-link/api.test.ts @@ -6,6 +6,21 @@ import type { GithubReplyClient } from "../github-client.js"; import { GatewayServer } from "../server.js"; import type { GithubDeviceFlowLinker } from "./device-flow-linker.js"; +/** + * Every request in this file goes over a fresh TCP connection — see the long + * explanation on the identical shim in agent-orchestrator's `server.test.ts`. + * Short version: these tests listen on ephemeral ports and close their servers, + * and Node's `fetch` caches keep-alive sockets per origin, so a recycled port + * can hand a later test a dead socket ("other side closed", zero bytes read). + * `connection: close` keeps anything from being pooled in the first place. + */ +const nativeFetch = globalThis.fetch; +const fetch: typeof globalThis.fetch = (input, init = {}) => { + const headers = new Headers(init.headers); + headers.set("connection", "close"); + return nativeFetch(input, { ...init, headers }); +}; + const TOKEN = "test-identity-link-token"; describe("GatewayServer identity-link routes", () => { diff --git a/apps/integration-gateway/src/server.test.ts b/apps/integration-gateway/src/server.test.ts index 7fea611..0eb2e11 100644 --- a/apps/integration-gateway/src/server.test.ts +++ b/apps/integration-gateway/src/server.test.ts @@ -8,6 +8,21 @@ import type { OrchestratorClient } from "./orchestrator-client.js"; import { GatewayServer, sessionIdFor } from "./server.js"; import { InMemorySessionPageStore } from "./session-page-store.js"; +/** + * Every request in this file goes over a fresh TCP connection — see the long + * explanation on the identical shim in agent-orchestrator's `server.test.ts`. + * Short version: these tests listen on ephemeral ports and close their servers, + * and Node's `fetch` caches keep-alive sockets per origin, so a recycled port + * can hand a later test a dead socket ("other side closed", zero bytes read). + * `connection: close` keeps anything from being pooled in the first place. + */ +const nativeFetch = globalThis.fetch; +const fetch: typeof globalThis.fetch = (input, init = {}) => { + const headers = new Headers(init.headers); + headers.set("connection", "close"); + return nativeFetch(input, { ...init, headers }); +}; + const SECRET = "test-secret"; function sign(body: string): string { diff --git a/charts/community-components/templates/agent-stub.yaml b/charts/community-components/templates/agent-stub.yaml index bb567b4..3ee6b51 100644 --- a/charts/community-components/templates/agent-stub.yaml +++ b/charts/community-components/templates/agent-stub.yaml @@ -58,6 +58,10 @@ spec: value: {{ .Values.stubAgent.pacing.narrateEveryMs | quote }} - name: STUB_SILENT_FOR_MS value: {{ .Values.stubAgent.pacing.silentForMs | quote }} + {{- if .Values.stubAgent.replyAckTimeoutMs }} + - name: AGENT_REPLY_ACK_TIMEOUT_MS + value: {{ .Values.stubAgent.replyAckTimeoutMs | quote }} + {{- end }} resources: requests: cpu: "50m" diff --git a/charts/community-components/values-e2e.yaml b/charts/community-components/values-e2e.yaml index 8b32edd..3d16ed3 100644 --- a/charts/community-components/values-e2e.yaml +++ b/charts/community-components/values-e2e.yaml @@ -39,6 +39,14 @@ stubAgent: providers: - claude - claude-remote + # 90s, well under the runtime's 10-minute default. resilience.e2e.ts's rollout + # spec deliberately strands a reply with no orchestrator to collect it; the + # stub holds it (that hold is what makes the turn resumable), which keeps the + # Job's pod alive and the AgentRun non-terminal until the hold ends. Long + # enough for the spec's follow-up trigger to re-attach and collect, short + # enough that a spec which fails to collect still reaches a terminal phase + # inside its own budget instead of timing out on a working hold. + replyAckTimeoutMs: 90000 # Removed from the e2e catalog entirely, not merely bypassed by the triage route. # diff --git a/charts/community-components/values.yaml b/charts/community-components/values.yaml index 47d3267..517717d 100644 --- a/charts/community-components/values.yaml +++ b/charts/community-components/values.yaml @@ -373,6 +373,14 @@ stubAgent: narrateForMs: 0 narrateEveryMs: 2000 silentForMs: 0 + # How long the stub holds an unacked concluding message before giving up + # (AGENT_REPLY_ACK_TIMEOUT_MS -- packages/agent-runtime's `publishHeld`). + # Empty = the runtime's own 10-minute default. Worth shortening in a test + # environment: an uncollected reply keeps the Job's pod alive for this long, + # so a spec that deliberately strands one (resilience.e2e.ts's rollout) would + # otherwise wait out the full default before the AgentRun reaches a terminal + # phase. + replyAckTimeoutMs: "" integrationRoutes: # github-issue-labeled-triage: an IntegrationRoute CR (docs/adr/0024) diff --git a/controllers/core-controller/internal/controller/agentrun_controller.go b/controllers/core-controller/internal/controller/agentrun_controller.go index 8bba200..feab35d 100644 --- a/controllers/core-controller/internal/controller/agentrun_controller.go +++ b/controllers/core-controller/internal/controller/agentrun_controller.go @@ -23,6 +23,7 @@ import ( batchv1 "k8s.io/api/batch/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" @@ -229,14 +230,28 @@ func (r *AgentRunReconciler) syncJobStatus(ctx context.Context, run *toolv1alpha func (r *AgentRunReconciler) markFailed(ctx context.Context, run *toolv1alpha1.AgentRun, reason, message string) (ctrl.Result, error) { run.Status.Phase = toolv1alpha1.ToolRunPhaseFailed run.Status.Message = message - cond := metav1.Condition{ + // meta.SetStatusCondition, not append: it stamps LastTransitionTime (which + // the CRD schema REQUIRES, so a hand-built condition without it is rejected + // outright -- "status.conditions[0].lastTransitionTime: Required value") and + // replaces any existing condition of the same type instead of adding a + // duplicate, which the schema also rejects since conditions is a map-list + // keyed by type. + // + // Appending raw wedged runs permanently. This function is the ONLY thing that + // moves a run whose Job has vanished to a terminal phase, so a rejected + // update meant the run stayed Running forever, was re-reconciled every few + // minutes, failed the same way, and was never eligible for the retention + // sweep below -- which only reclaims TERMINAL runs. Observed in production: + // three AgentRuns and a ToolRun stuck Running for two days, each logging this + // error on every reconcile. Every other controller here already used this + // helper; these two were the outliers. + meta.SetStatusCondition(&run.Status.Conditions, metav1.Condition{ Type: "Ready", Status: metav1.ConditionFalse, Reason: reason, Message: message, ObservedGeneration: run.Generation, - } - run.Status.Conditions = append(run.Status.Conditions, cond) + }) if err := r.Status().Update(ctx, run); err != nil { return ctrl.Result{}, err } diff --git a/controllers/core-controller/internal/controller/agentrun_controller_test.go b/controllers/core-controller/internal/controller/agentrun_controller_test.go index 06894b4..ca3c279 100644 --- a/controllers/core-controller/internal/controller/agentrun_controller_test.go +++ b/controllers/core-controller/internal/controller/agentrun_controller_test.go @@ -253,3 +253,99 @@ var _ = Describe("AgentRun Controller", func() { }) }) }) + +// Regression: a run whose Job has vanished must actually REACH a terminal +// phase. +// +// markFailed hand-built its condition and appended it, omitting +// LastTransitionTime -- which the CRD schema requires. Every status update it +// attempted was rejected ("status.conditions[0].lastTransitionTime: Required +// value"), so the run stayed Running, was re-reconciled every few minutes, +// failed identically, and never became eligible for the retention sweep (which +// only reclaims terminal runs). Production held three AgentRuns and a ToolRun +// wedged this way for two days. +// +// It needs a real API server to catch: only schema validation rejects the +// update, so this could not have been caught by a fake client. +var _ = Describe("AgentRun Controller reaching a terminal phase when its Job is gone", func() { + const runName = "orphaned-run" + const agentName = "orphan-agent" + ctx := context.Background() + key := types.NamespacedName{Name: runName, Namespace: "default"} + + BeforeEach(func() { + agent := &toolv1alpha1.Agent{ + ObjectMeta: metav1.ObjectMeta{Name: agentName, Namespace: "default"}, + Spec: toolv1alpha1.AgentSpec{ + Description: "orphan-run test agent", + Input: "a natural-language goal", + Output: "a final response payload", + AllowedRoles: []string{"reader"}, + Image: "ghcr.io/example/agent:latest", + ServiceAccountName: "default", + }, + } + Expect(k8sClient.Create(ctx, agent)).To(Succeed()) + + run := &toolv1alpha1.AgentRun{ + ObjectMeta: metav1.ObjectMeta{Name: runName, Namespace: "default"}, + Spec: toolv1alpha1.AgentRunSpec{AgentRef: agentName, Goal: "do the thing"}, + } + Expect(k8sClient.Create(ctx, run)).To(Succeed()) + // The wedged state: mid-flight, pointing at a Job that does not exist. + run.Status.Phase = toolv1alpha1.ToolRunPhaseRunning + run.Status.JobName = "agentrun-" + runName + Expect(k8sClient.Status().Update(ctx, run)).To(Succeed()) + }) + + AfterEach(func() { + run := &toolv1alpha1.AgentRun{} + if err := k8sClient.Get(ctx, key, run); err == nil { + Expect(k8sClient.Delete(ctx, run)).To(Succeed()) + } + agent := &toolv1alpha1.Agent{} + if err := k8sClient.Get(ctx, types.NamespacedName{Name: agentName, Namespace: "default"}, agent); err == nil { + Expect(k8sClient.Delete(ctx, agent)).To(Succeed()) + } + }) + + It("marks the run Failed and persists the condition, instead of erroring forever", func() { + reconciler := &AgentRunReconciler{Client: k8sClient, Scheme: k8sClient.Scheme()} + + _, err := reconciler.Reconcile(ctx, reconcile.Request{NamespacedName: key}) + Expect(err).NotTo(HaveOccurred()) + + var updated toolv1alpha1.AgentRun + Expect(k8sClient.Get(ctx, key, &updated)).To(Succeed()) + Expect(updated.Status.Phase).To(Equal(toolv1alpha1.ToolRunPhaseFailed)) + Expect(updated.Status.Message).To(ContainSubstring("no longer exists")) + Expect(updated.Status.Conditions).To(HaveLen(1)) + Expect(updated.Status.Conditions[0].Type).To(Equal("Ready")) + Expect(updated.Status.Conditions[0].Reason).To(Equal("JobMissing")) + // The field whose absence wedged this in production. + Expect(updated.Status.Conditions[0].LastTransitionTime.IsZero()).To(BeFalse()) + }) + + It("does not accumulate duplicate conditions when marked failed more than once", func() { + reconciler := &AgentRunReconciler{Client: k8sClient, Scheme: k8sClient.Scheme()} + + _, err := reconciler.Reconcile(ctx, reconcile.Request{NamespacedName: key}) + Expect(err).NotTo(HaveOccurred()) + + // Put it back mid-flight and fail it again. `conditions` is a map-list + // keyed by type, so a second "Ready" entry would be rejected outright -- + // the second way appending raw could wedge a run. + var again toolv1alpha1.AgentRun + Expect(k8sClient.Get(ctx, key, &again)).To(Succeed()) + again.Status.Phase = toolv1alpha1.ToolRunPhaseRunning + Expect(k8sClient.Status().Update(ctx, &again)).To(Succeed()) + + _, err = reconciler.Reconcile(ctx, reconcile.Request{NamespacedName: key}) + Expect(err).NotTo(HaveOccurred()) + + var final toolv1alpha1.AgentRun + Expect(k8sClient.Get(ctx, key, &final)).To(Succeed()) + Expect(final.Status.Phase).To(Equal(toolv1alpha1.ToolRunPhaseFailed)) + Expect(final.Status.Conditions).To(HaveLen(1)) + }) +}) diff --git a/controllers/core-controller/internal/controller/toolrun_controller.go b/controllers/core-controller/internal/controller/toolrun_controller.go index ac6b8c3..f5f9833 100644 --- a/controllers/core-controller/internal/controller/toolrun_controller.go +++ b/controllers/core-controller/internal/controller/toolrun_controller.go @@ -22,6 +22,7 @@ import ( batchv1 "k8s.io/api/batch/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" @@ -163,14 +164,28 @@ func (r *ToolRunReconciler) syncJobStatus(ctx context.Context, run *toolv1alpha1 func (r *ToolRunReconciler) markFailed(ctx context.Context, run *toolv1alpha1.ToolRun, reason, message string) (ctrl.Result, error) { run.Status.Phase = toolv1alpha1.ToolRunPhaseFailed run.Status.Message = message - cond := metav1.Condition{ + // meta.SetStatusCondition, not append: it stamps LastTransitionTime (which + // the CRD schema REQUIRES, so a hand-built condition without it is rejected + // outright -- "status.conditions[0].lastTransitionTime: Required value") and + // replaces any existing condition of the same type instead of adding a + // duplicate, which the schema also rejects since conditions is a map-list + // keyed by type. + // + // Appending raw wedged runs permanently. This function is the ONLY thing that + // moves a run whose Job has vanished to a terminal phase, so a rejected + // update meant the run stayed Running forever, was re-reconciled every few + // minutes, failed the same way, and was never eligible for the retention + // sweep below -- which only reclaims TERMINAL runs. Observed in production: + // three AgentRuns and a ToolRun stuck Running for two days, each logging this + // error on every reconcile. Every other controller here already used this + // helper; these two were the outliers. + meta.SetStatusCondition(&run.Status.Conditions, metav1.Condition{ Type: "Ready", Status: metav1.ConditionFalse, Reason: reason, Message: message, ObservedGeneration: run.Generation, - } - run.Status.Conditions = append(run.Status.Conditions, cond) + }) if err := r.Status().Update(ctx, run); err != nil { return ctrl.Result{}, err } diff --git a/controllers/core-controller/internal/controller/toolrun_controller_test.go b/controllers/core-controller/internal/controller/toolrun_controller_test.go index 74a8733..8168ef0 100644 --- a/controllers/core-controller/internal/controller/toolrun_controller_test.go +++ b/controllers/core-controller/internal/controller/toolrun_controller_test.go @@ -228,3 +228,74 @@ var _ = Describe("ToolRun Controller", func() { }) }) }) + +// Same regression as the AgentRun spec of this name, in the controller it was +// copied from: markFailed appended a hand-built condition with no +// LastTransitionTime, the schema rejected every update, and a ToolRun whose Job +// had vanished stayed Running forever. Production held one wedged this way +// alongside the AgentRuns. +var _ = Describe("ToolRun Controller reaching a terminal phase when its Job is gone", func() { + const runName = "orphaned-toolrun" + const toolName = "orphan-tool" + ctx := context.Background() + key := types.NamespacedName{Name: runName, Namespace: "default"} + + BeforeEach(func() { + tool := &toolv1alpha1.Tool{ + ObjectMeta: metav1.ObjectMeta{Name: toolName, Namespace: "default"}, + Spec: toolv1alpha1.ToolSpec{ + Description: "orphan-run test tool", + Input: "a URL", + Output: "a recipe JSON envelope", + AllowedRoles: []string{"reader"}, + Image: "example.com/recipe-scraper:latest", + ServiceAccountName: "default", + }, + } + Expect(k8sClient.Create(ctx, tool)).To(Succeed()) + + run := &toolv1alpha1.ToolRun{ + ObjectMeta: metav1.ObjectMeta{Name: runName, Namespace: "default"}, + Spec: toolv1alpha1.ToolRunSpec{ + ToolRef: toolName, + Args: []string{"https://example.com/recipe"}, + Callback: toolv1alpha1.ToolRunCallback{ + URL: "http://agent-orchestrator-callback.default.svc.cluster.local:8080", + SecretRef: toolv1alpha1.SecretKeySelector{ + Name: "agent-orchestrator-secrets", + Key: "AGENT_CALLBACK_SECRET", + }, + }, + }, + } + Expect(k8sClient.Create(ctx, run)).To(Succeed()) + run.Status.Phase = toolv1alpha1.ToolRunPhaseRunning + run.Status.JobName = "toolrun-" + runName + Expect(k8sClient.Status().Update(ctx, run)).To(Succeed()) + }) + + AfterEach(func() { + run := &toolv1alpha1.ToolRun{} + if err := k8sClient.Get(ctx, key, run); err == nil { + Expect(k8sClient.Delete(ctx, run)).To(Succeed()) + } + tool := &toolv1alpha1.Tool{} + if err := k8sClient.Get(ctx, types.NamespacedName{Name: toolName, Namespace: "default"}, tool); err == nil { + Expect(k8sClient.Delete(ctx, tool)).To(Succeed()) + } + }) + + It("marks the run Failed and persists the condition", func() { + reconciler := &ToolRunReconciler{Client: k8sClient, Scheme: k8sClient.Scheme()} + + _, err := reconciler.Reconcile(ctx, reconcile.Request{NamespacedName: key}) + Expect(err).NotTo(HaveOccurred()) + + var updated toolv1alpha1.ToolRun + Expect(k8sClient.Get(ctx, key, &updated)).To(Succeed()) + Expect(updated.Status.Phase).To(Equal(toolv1alpha1.ToolRunPhaseFailed)) + Expect(updated.Status.Conditions).To(HaveLen(1)) + Expect(updated.Status.Conditions[0].Reason).To(Equal("JobMissing")) + Expect(updated.Status.Conditions[0].LastTransitionTime.IsZero()).To(BeFalse()) + }) +}) diff --git a/docs/adr/0033-resumable-agent-turns.md b/docs/adr/0033-resumable-agent-turns.md new file mode 100644 index 0000000..2363381 --- /dev/null +++ b/docs/adr/0033-resumable-agent-turns.md @@ -0,0 +1,129 @@ +# 0033. An agent turn survives losing the orchestrator that was waiting on it + +Date: 2026-07-26 + +## Status + +Accepted + +Completes the fix begun in "Fix agent turns failing on orchestrator rollout, not +on any timeout" (`55b0e4b`), which made the failure honest but did not remove it. + +## Context + +An agent turn's work lives in its own Job pod. The turn's *wait* lives in an +orchestrator pod: one HTTP request parked on `awaitReply`, holding a core NATS +subscription to the run's up subject. Those two lifetimes are unrelated, and the +second is far shorter than the first. + +`release.yml` deploys on every push to `main`, and each deploy restarts +agent-orchestrator. The incident that prompted `55b0e4b` recorded eleven +rollouts in fourteen hours. Any agent turn in flight during any of them is a +turn whose waiter disappears. + +`55b0e4b` fixed three real defects — a bound that was never armed, NATS drained +in the same `Promise.all` as the HTTP close, and every failure relabelled as a +timeout — and added a bounded drain (`shutdownDrainMs`, 25s, sized under the +pod's 30s termination grace period). What it could not fix is the shape of the +problem: + +- an agent turn takes minutes; the drain gives it 25 seconds; +- past that, `awaitReply` throws `AgentTurnTransportError` and the turn fails + while the run goes on to succeed; +- core NATS has no durability, so the `reply` the agent publishes into that gap + is **discarded** — there is no subscriber. Even a replacement orchestrator + re-subscribing to the (deterministic) subject one second later gets nothing. + +So the honest message was still reporting a real loss. The answer existed, was +correct, and was unrecoverable. Meanwhile the chat path had a session record +that could have anchored a resume, and the GitHub path already sends a stable +per-issue `session_id` — the pieces were there; nothing joined them. + +The obvious fix is durability at the transport: JetStream over `agent.*.up`, +consumed by a per-run durable consumer. The deployed NATS has no `jetstream` +block in its config, so this means enabling JetStream on the shared server, +provisioning storage, and moving both sides onto a stream — infrastructure work +whose failure modes (disk, retention, consumer leakage) are new and permanent. + +## Decision + +Make the turn resumable, using the pod that already outlives the orchestrator — +the agent's own — as the buffer. Three parts. + +### 1. The agent holds its concluding message until it is acked + +New down-message `reply_ack { ackSeq }`. The agent publishes `reply` (either +finality) and `failed` through `publishHeld`, which re-offers the *identical +envelope* — same `seq`, so a duplicate is recognizable as one — every +`AGENT_REPLY_ACK_RETRY_MS` (10s) until the orchestrator acks that `seq`, giving +up after `AGENT_REPLY_ACK_TIMEOUT_MS` (10min). + +Narration (`progress`/`warning`) is deliberately not held: it is commentary, +worthless once the turn it narrated is over. A question (non-final `reply`) *is* +held — losing a question strands the conversation exactly as badly as losing an +answer — and its hold is released the moment the answer arrives, since an answer +proves the question landed. + +### 2. The conversation is anchored to the run before the wait, not after + +`SessionRecord.activeAgentRunAwaitingReply` distinguishes "this run owes us a +reply" from the pre-existing "this run is parked on a question". It is written +by `markAgentRunAwaitingReply` immediately after the AgentRun is created and +*before* `awaitReply` is entered. + +The timing is the whole point. `InvokeServer.persistSession` runs after the +graph returns, which is fine for every outcome that has one — and useless for +this one, where the process may be SIGKILLed mid-wait. An anchor written after +the fact is an anchor that is never written. + +### 3. The next turn re-attaches instead of re-delegating + +`checkActiveAgentRun` already ran before retrieval to continue a parked run. It +now branches: parked on a question → publish this turn's text as a `prompt` +(unchanged); owed a reply → publish **nothing** and simply collect. Prompting +here would inject "any update?" into a working agent's conversation, and +re-delegating would do the work twice — a second branch and a second PR on a +real coding agent. + +A re-attached wait is bounded at 45s rather than the full idle window, because +silence *is* diagnostic there: a working run heartbeats every 20s, a finished +one re-offers every 10s. Hearing nothing means the pod is gone, at which point +the anchor is dropped and the user is told the answer is unrecoverable — rather +than being made to wait out ten minutes to learn it. + +An interrupted turn now reports a resumable pause (`result`, not `error`): +nothing failed, and "your request failed, try again" would be the third +variation on telling someone their successful run failed. + +## Consequences + +**An uncollected answer keeps a Job pod alive** for up to the hold timeout, +which delays that AgentRun reaching a terminal phase. That is the mechanism +working, not a leak — the hold *is* the buffer — but it makes the timeout an +operational setting (`AGENT_REPLY_ACK_TIMEOUT_MS`, `0` disables holding +entirely) rather than an internal detail. values-e2e sets 90s so a spec that +deliberately strands a reply still terminates inside its budget. + +**A mixed deployment holds pointlessly.** An orchestrator too old to send +`reply_ack` never acks, so agents from this commit forward hold every concluding +message for the full timeout before exiting. Bounded and harmless to +correctness (the reply is published immediately either way), but it is why the +timeout is env-tunable and why `0` exists. + +**The interrupted turn itself is still lost.** The invocation record +integration-gateway polls lives in an in-process `Map`, so the in-flight *turn* +cannot be recovered — only the answer, on the next turn. Making the turn itself +survive means durable invocation records, which this does not attempt. + +**Agent-backed Tools (`runTool`) remain unresumable.** They have no session slot +to anchor a tool-launched AgentRun, the same v1 scope cut already recorded there +for non-final replies. A rollout mid-dispatch still fails that turn. + +**Duplicate concluding messages are now possible on the wire.** Re-offers reuse +their original `seq` precisely so a consumer can tell a re-offer from a second +reply; anything added later that consumes the up subject must not assume +at-most-once delivery. + +**JetStream is not ruled out.** If durable orchestration arrives for the +invocation records above, moving the up subject onto a stream would subsume the +hold entirely, and `AGENT_REPLY_ACK_TIMEOUT_MS=0` is the switch that retires it. diff --git a/docs/adr/README.md b/docs/adr/README.md index f2c452c..22aefeb 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -38,5 +38,6 @@ See [../orchestrator.md](../orchestrator.md) for how these fit together. | [0030](0030-authorization-preflight-outside-the-llm.md) | Authorization becomes a single deterministic pre-flight owned by agent-orchestrator — batch-resolving every provider an Agent CRD declares before launch, handing the run a sealed actor context so agents stop re-deriving identity, with the LLM able only to ask a human to link an account and never to decide authorization or see a credential | | [0031](0031-principal-establishing-account-link.md) | Establishing the caller's principal is its own authorization pre-flight step — a link-only `github` link, gated on a resolver-asserted per-user subject and independent of `Agent.identityProviders` — so Open WebUI chat can learn the login that keys its Claude credentials and actually converge with GitHub-webhook triage, instead of the two flows sharing in one direction only; a pre-principal credential is MOVED onto the principal (new gateway `rekey`) so converging costs no re-authorization | | [0032](0032-tool-level-identity-delegation-and-github-cli-tool.md) | Per-user GitHub identity delegation (0022) extends from `Agent`/`AgentRun` to container `Tool`/`ToolRun` (new `ToolRunSpec.SecretEnv`, `ToolSpec.IdentityProviders`), and a new `github` Tool (gh CLI preinstalled) is added as its reference implementation | +| [0033](0033-resumable-agent-turns.md) | An agent turn survives losing the orchestrator waiting on it: the agent holds its concluding message until acked (new `reply_ack`), the conversation is anchored to the run BEFORE the wait, and the next turn re-attaches to collect the held reply instead of re-delegating — so an orchestrator rollout mid-turn costs the turn, not the answer | Status values: `proposed` | `accepted` | `superseded by NNNN`. diff --git a/e2e/specs/resilience.e2e.ts b/e2e/specs/resilience.e2e.ts index 7d3e8fe..33ab78b 100644 --- a/e2e/specs/resilience.e2e.ts +++ b/e2e/specs/resilience.e2e.ts @@ -122,9 +122,17 @@ describe("resilience: infrastructure moving under an in-flight agent turn", () = await cleanupAgentRunsSince(suiteStartedAt); }); - /** Fires a triage webhook and returns the issue number it used. */ - async function trigger(): Promise<{ issueNumber: number; startedAt: Date }> { - const issueNumber = Date.now() % 100000; + /** + * Fires a triage webhook and returns the issue number it used. + * + * `onIssue` re-triggers against an issue already used. That matters for + * resumability: integration-gateway derives its `session_id` from the issue + * (`github:owner/repo#N`), so a second trigger on the SAME issue is the same + * conversation to the orchestrator, and therefore the thing that re-attaches + * to a run a previous turn was cut off from. + */ + async function trigger(onIssue?: number): Promise<{ issueNumber: number; startedAt: Date }> { + const issueNumber = onIssue ?? Date.now() % 100000; const startedAt = new Date(); const status = await withPortForward( "agent-controller-integration-gateway", @@ -296,13 +304,67 @@ describe("resilience: infrastructure moving under an in-flight agent turn", () = // Nor may transport loss be relabelled as the agent going quiet. expect(comment?.body).not.toMatch(/went silent for/); - // Deliberately NOT asserted: that the comment carries STUB_REPLY_MARKER. - // The invocation record the gateway polls lives in an in-process Map - // (server.ts), so a rollout can lose it however cleanly NATS was handled, - // and the gateway then relays a failure. That gap is real, called out in - // the PR, and closed by durable orchestration -- not by this change. - // Asserting the marker would be asserting a fix that does not exist. It is - // logged above so a future reader can see which way it actually went. + // Deliberately NOT asserted here: that THIS comment carries + // STUB_REPLY_MARKER. The invocation record the gateway polls lives in an + // in-process Map (server.ts), so the interrupted turn itself still cannot + // produce the reply however cleanly NATS was handled. What recovers it is + // the NEXT turn re-attaching -- see the resumability spec below, which + // asserts exactly that. + expect(comment).toBeDefined(); + }); + + /** + * The other half: a rollout costs the interrupted turn, but not the ANSWER. + * + * The agent holds its concluding message until someone acks it (the protocol's + * `reply_ack`), and the orchestrator wrote a resume anchor onto the + * conversation before it started waiting, so a second trigger on the same + * issue re-attaches to the SAME run and collects the reply it was holding. + * + * This is the assertion the spec above deliberately withholds. It only means + * anything end-to-end: the hold lives in the agent's pod, the anchor in Redis, + * the re-attach in the orchestrator, and the session identity in the gateway's + * issue-derived `session_id`. + */ + it("recovers the reply on a follow-up turn after a rollout, without launching a second run", async () => { + await paceStubAgent({ narrateForMs: 20_000, narrateEveryMs: 2000 }); + + const { issueNumber, startedAt } = await trigger(); + const created = await runCreated(startedAt); + console.log(` [resilience] rolling the orchestrator while ${created.name} is in flight`); + await rollOrchestrator(); + + // Second trigger on the SAME issue -> same session_id -> the orchestrator + // finds the anchor and re-attaches instead of starting over. + await trigger(issueNumber); + + const comment = await waitFor( + `the recovered reply on issue #${issueNumber}`, + async () => { + const posted = (await fakeGithubRequests()).filter( + (r) => + r.method === "POST" && + r.path === `/repos/${OWNER}/${REPO}/issues/${issueNumber}/comments` && + typeof r.body === "string" && + r.body.includes(STUB_REPLY_MARKER), + ); + return posted.length > 0 ? posted[posted.length - 1] : undefined; + }, + { timeoutMs: GATEWAY_POLL_BUDGET_MS }, + ); expect(comment).toBeDefined(); + + // The point of re-attaching rather than re-delegating: the original run + // produced the answer, so a second AgentRun would mean the work was done + // twice (and on a real coding agent, a second branch and a second PR). + const runs = await agentRunsSince(startedAt); + expect(runs).toHaveLength(1); + expect(runs[0]?.name).toBe(created.name); + + // Collecting the reply acks it, which releases the agent's hold and lets the + // Job finish -- so the run reaching terminal is itself evidence the ack + // arrived rather than the hold having simply timed out. + const run = await runTerminal(startedAt); + expect(run.phase).toBe("Succeeded"); }); }); diff --git a/packages/agent-runtime/src/config.test.ts b/packages/agent-runtime/src/config.test.ts new file mode 100644 index 0000000..0f884ba --- /dev/null +++ b/packages/agent-runtime/src/config.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from "vitest"; +import { AgentConfigError, loadConfig } from "./config.js"; + +describe("loadConfig reply-ack overrides", () => { + const base = { AGENT_NATS_URL: "nats://x", AGENT_RUN_ID: "run-1", AGENT_GOAL: "do it" }; + + it("omits the overrides entirely when unset, so the runtime default applies", () => { + const cfg = loadConfig(base as NodeJS.ProcessEnv, []); + expect(cfg.replyAckTimeoutMs).toBeUndefined(); + expect(cfg.replyAckRetryMs).toBeUndefined(); + }); + + it("reads both overrides", () => { + const cfg = loadConfig({ ...base, AGENT_REPLY_ACK_TIMEOUT_MS: "5000", AGENT_REPLY_ACK_RETRY_MS: "250" } as NodeJS.ProcessEnv, []); + expect(cfg.replyAckTimeoutMs).toBe(5000); + expect(cfg.replyAckRetryMs).toBe(250); + }); + + it("accepts 0 as an explicit 'do not hold'", () => { + expect(loadConfig({ ...base, AGENT_REPLY_ACK_TIMEOUT_MS: "0" } as NodeJS.ProcessEnv, []).replyAckTimeoutMs).toBe(0); + }); + + /** + * A typo'd timeout that silently reads as "use the default" would only be + * discovered during the incident the setting was meant to prevent. + */ + it("fails fast on a non-numeric or negative value", () => { + expect(() => loadConfig({ ...base, AGENT_REPLY_ACK_TIMEOUT_MS: "10s" } as NodeJS.ProcessEnv, [])).toThrow(AgentConfigError); + expect(() => loadConfig({ ...base, AGENT_REPLY_ACK_RETRY_MS: "-1" } as NodeJS.ProcessEnv, [])).toThrow(AgentConfigError); + }); +}); diff --git a/packages/agent-runtime/src/config.ts b/packages/agent-runtime/src/config.ts index f271971..299ee9e 100644 --- a/packages/agent-runtime/src/config.ts +++ b/packages/agent-runtime/src/config.ts @@ -11,6 +11,20 @@ export interface AgentRuntimeConfig { subjectPrefix: string; /** The initial goal for this run (AGENT_GOAL, or argv[2] as a fallback). */ goal: string; + /** + * How long to keep re-offering a concluding message that has not been acked + * (AGENT_REPLY_ACK_TIMEOUT_MS), and how often (AGENT_REPLY_ACK_RETRY_MS) — + * see `runtime.ts`'s `publishHeld`. + * + * Operationally relevant because holding is what keeps the answer collectable + * across an orchestrator rollout, and it does so by keeping the Job's pod + * alive that much longer when nobody collects. `0` disables holding entirely, + * restoring the pre-`reply_ack` publish-and-exit behaviour — the escape hatch + * if an orchestrator that never acks is ever deployed against a newer agent + * image. + */ + replyAckTimeoutMs?: number; + replyAckRetryMs?: number; } export class AgentConfigError extends Error {} @@ -39,5 +53,23 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env, argv: string[] runId: runId!, subjectPrefix: env.AGENT_NATS_SUBJECT_PREFIX ?? "agent", goal: goal!, + ...numeric("replyAckTimeoutMs", env.AGENT_REPLY_ACK_TIMEOUT_MS), + ...numeric("replyAckRetryMs", env.AGENT_REPLY_ACK_RETRY_MS), }; } + +/** + * Parses an optional numeric env override, omitting the key entirely when unset + * so the runtime's own default applies. A non-numeric value is a + * misconfiguration worth failing on rather than silently ignoring: a typo'd + * timeout that reads as "use the default" is the kind of thing that is only + * discovered during the incident it was meant to prevent. + */ +function numeric(key: string, raw: string | undefined): Record { + if (raw === undefined || raw === "") return {}; + const value = Number(raw); + if (!Number.isFinite(value) || value < 0) { + throw new AgentConfigError(`invalid ${key} (${raw}): expected a non-negative number of milliseconds`); + } + return { [key]: value }; +} diff --git a/packages/agent-runtime/src/runtime.test.ts b/packages/agent-runtime/src/runtime.test.ts index 438da15..1876d63 100644 --- a/packages/agent-runtime/src/runtime.test.ts +++ b/packages/agent-runtime/src/runtime.test.ts @@ -10,14 +10,28 @@ const config: AgentRuntimeConfig = { goal: "do the thing", }; -/** In-memory channel: records up-messages, lets the test push down-messages. */ +/** + * In-memory channel: records up-messages, lets the test push down-messages. + * + * Acks `reply`/`failed` by default, because a live orchestrator does (see the + * protocol's `reply_ack`) and the runtime holds those messages until it hears + * one. Tests that exercise the holding itself construct this with + * `{ autoAck: false }` to stand in for an orchestrator that has gone away. + */ class FakeChannel implements AgentChannel { readonly up: AgentUpMessage[] = []; private handler: ((msg: AgentDownMessage) => void) | undefined; closed = false; + constructor(private readonly opts: { autoAck?: boolean } = {}) {} + publishUp(msg: AgentUpMessage): Promise { this.up.push(msg); + if ((this.opts.autoAck ?? true) && (msg.type === "reply" || msg.type === "failed")) { + // Asynchronously, like a real round trip -- an ack delivered + // synchronously inside publishUp would hide ordering bugs. + queueMicrotask(() => this.send({ type: "reply_ack", ackSeq: msg.seq })); + } return Promise.resolve(); } onDown(handler: (msg: AgentDownMessage) => void): void { @@ -242,4 +256,135 @@ describe("runAgent", () => { expect(ch.types()).toEqual(["ready", "reply"]); expect(ch.closed).toBe(true); }); + + /** + * The agent's half of surviving an orchestrator that vanishes mid-turn. Core + * NATS drops a message published with no live subscriber, so the concluding + * message is held and re-offered until acked -- that hold is what a + * replacement orchestrator re-attaches to and collects. + */ + describe("holding a concluding message until it is acked", () => { + it("re-offers the same reply, seq unchanged, until an ack arrives", async () => { + const ch = new FakeChannel({ autoAck: false }); + const done = runAgent(async () => "the answer", { + channel: ch, + config, + replyAckRetryMs: 5, + replyAckTimeoutMs: 5_000, + }); + + // Enough retry intervals to be sure it is re-offering rather than having + // published once and moved on. + await new Promise((r) => setTimeout(r, 40)); + const offers = ch.up.filter((m) => m.type === "reply"); + expect(offers.length).toBeGreaterThan(1); + // A re-offer must be the SAME message, not a second reply: the seq is how + // the orchestrator recognizes a duplicate. + expect(new Set(offers.map((m) => m.seq)).size).toBe(1); + expect(offers.every((m) => (m as Extract).message === "the answer")).toBe(true); + + ch.send({ type: "reply_ack", ackSeq: offers[0]!.seq }); + await done; + + const after = ch.up.filter((m) => m.type === "reply").length; + await new Promise((r) => setTimeout(r, 20)); + // Acked, so it stopped: no further offers after the run resolved. + expect(ch.up.filter((m) => m.type === "reply").length).toBe(after); + expect(ch.closed).toBe(true); + }); + + it("gives up after the ack timeout instead of holding the pod open forever", async () => { + const ch = new FakeChannel({ autoAck: false }); + await runAgent(async () => "the answer", { + channel: ch, + config, + replyAckRetryMs: 5, + replyAckTimeoutMs: 20, + }); + + // Resolved without an ack -- the whole point is that an uncollected answer + // costs a logged warning, not a Job that never finishes. + expect(ch.up.filter((m) => m.type === "reply").length).toBeGreaterThan(0); + expect(ch.closed).toBe(true); + }); + + it("holds a failed message too", async () => { + const ch = new FakeChannel({ autoAck: false }); + const done = runAgent( + async () => { + throw new AgentFailure("claude_auth_expired", "credential expired"); + }, + { channel: ch, config, replyAckRetryMs: 5, replyAckTimeoutMs: 5_000 }, + ); + + await new Promise((r) => setTimeout(r, 30)); + const offers = ch.up.filter((m) => m.type === "failed"); + expect(offers.length).toBeGreaterThan(1); + expect(new Set(offers.map((m) => m.seq)).size).toBe(1); + + ch.send({ type: "reply_ack", ackSeq: offers[0]!.seq }); + await done; + expect(ch.closed).toBe(true); + }); + + it("stops holding a question once its answer arrives, even unacked", async () => { + const ch = new FakeChannel({ autoAck: false }); + const done = runAgent( + async (s) => { + const answer = await s.ask("Which branch?"); + return `on ${answer}`; + }, + { channel: ch, config, replyAckRetryMs: 5, replyAckTimeoutMs: 5_000 }, + ); + + await new Promise((r) => setTimeout(r, 20)); + // The answer proves the question landed; re-offering it after that would + // surface a stale question to the next turn. + ch.send({ type: "prompt", message: "main" }); + await new Promise((r) => setTimeout(r, 30)); + const questionOffers = ch.up.filter( + (m) => m.type === "reply" && !(m as Extract).final, + ).length; + await new Promise((r) => setTimeout(r, 20)); + expect( + ch.up.filter((m) => m.type === "reply" && !(m as Extract).final).length, + ).toBe(questionOffers); + + // And the final reply is a separate hold with its own seq. + const final = ch.up.find((m) => m.type === "reply" && (m as Extract).final); + expect(final).toBeDefined(); + ch.send({ type: "reply_ack", ackSeq: final!.seq }); + await done; + }); + }); +}); + +/** + * The hold keeps a finished agent's pod alive until its answer is collected, + * which makes it an operational knob, not just an internal detail: an + * orchestrator that never acks (an older image deployed against a newer agent) + * would otherwise keep every Job Running for the full timeout. + */ +describe("reply-ack configuration", () => { + it("takes the hold window from config when the caller passes none", async () => { + const ch = new FakeChannel({ autoAck: false }); + await runAgent(async () => "the answer", { + channel: ch, + config: { ...config, replyAckRetryMs: 5, replyAckTimeoutMs: 20 }, + }); + + expect(ch.up.filter((m) => m.type === "reply").length).toBeGreaterThan(1); + expect(ch.closed).toBe(true); + }); + + it("publishes once and exits when holding is disabled (timeout 0)", async () => { + const ch = new FakeChannel({ autoAck: false }); + await runAgent(async () => "the answer", { + channel: ch, + config: { ...config, replyAckTimeoutMs: 0 }, + }); + + expect(ch.up.filter((m) => m.type === "reply").length).toBe(1); + expect(ch.closed).toBe(true); + }); }); diff --git a/packages/agent-runtime/src/runtime.ts b/packages/agent-runtime/src/runtime.ts index dfcb1c2..3147c26 100644 --- a/packages/agent-runtime/src/runtime.ts +++ b/packages/agent-runtime/src/runtime.ts @@ -85,6 +85,10 @@ export interface RunAgentOptions { channel?: AgentChannel; /** Inject config (tests); default reads the environment. */ config?: AgentRuntimeConfig; + /** Override {@link REPLY_ACK_RETRY_MS} (tests). */ + replyAckRetryMs?: number; + /** Override {@link REPLY_ACK_TIMEOUT_MS} (tests). */ + replyAckTimeoutMs?: number; } class CancelledError extends Error { @@ -96,6 +100,25 @@ class CancelledError extends Error { /** Distributive Omit so each union variant keeps its own fields (a plain Omit over a union keeps only common keys). */ type WithoutEnvelope = T extends unknown ? Omit : never; +/** + * How often to re-offer an unacked concluding message, and how long to keep + * offering it before giving up and exiting. + * + * This is the agent's half of `reply_ack` (see the protocol's doc comment): + * core NATS drops anything published while the orchestrator has no live + * subscription, and the concluding message is the only one whose loss turns a + * successful run into a failed-looking turn. The Job pod outlives an + * orchestrator rollout, so simply holding the answer and re-offering it lets a + * replacement orchestrator reattach and collect it. + * + * 10 minutes because the gap being covered is an orchestrator pod replacement + * (seconds) plus however long it takes something to re-drive the turn -- for a + * chat conversation, the user's next message. Past that the answer is stale + * enough that holding a pod open for it is the worse trade. + */ +const REPLY_ACK_RETRY_MS = 10_000; +const REPLY_ACK_TIMEOUT_MS = 10 * 60 * 1000; + /** * Boots a sub-agent: connects the channel, announces `ready`, runs `handler` * against a {@link AgentSession}, and publishes the terminal `reply`/`failed`. @@ -118,7 +141,109 @@ export async function runAgent(handler: AgentHandler, opts: RunAgentOptions = {} } as AgentUpMessage); const abort = new AbortController(); - let pendingAsk: { resolve: (answer: string) => void; reject: (err: Error) => void } | undefined; + let pendingAsk: + | { resolve: (answer: string) => void; reject: (err: Error) => void; seq: number } + | undefined; + + /** + * Concluding messages currently being held for an ack, keyed by `seq`. A map + * rather than a single slot because two can legitimately overlap: a question + * (`ask()`, a non-final `reply`) is still being held when its answer arrives + * and the agent goes on to publish its final reply. With one slot the older + * hold would never be woken and would keep re-offering a question the + * conversation had already moved past. + */ + const holds = new Map void }>(); + + /** Stops holding `seq` without an ack — the message became moot on its own terms. */ + const releaseHold = (seq: number): void => { + const hold = holds.get(seq); + if (!hold) return; + hold.done = true; + hold.wake?.(); + }; + + /** + * Publishes a message whose loss would break the turn (`reply` of either + * finality, `failed`) and holds it until the orchestrator acks that exact + * `seq`, re-offering it every {@link REPLY_ACK_RETRY_MS} until then. Every + * re-offer reuses the ORIGINAL envelope, `seq` included, so the orchestrator + * sees a duplicate as one rather than as a second reply. + * + * Resolves either way — an unacked message is logged and abandoned rather + * than holding the Job's pod open indefinitely. Cancellation and + * {@link releaseHold} both end the wait early. + */ + const publishHeld = async ( + msg: WithoutEnvelope, + // Callers that must know the seq BEFORE the hold resolves (ask(), which + // has to record it on `pendingAsk` so the answer can release the hold) + // allocate it themselves and pass it in. + allocatedSeq: number = seq++, + ): Promise => { + const envelope = { + ...msg, + agent_run_id: config.runId, + seq: allocatedSeq, + ts: new Date().toISOString(), + } as AgentUpMessage; + const retryMs = opts.replyAckRetryMs ?? config.replyAckRetryMs ?? REPLY_ACK_RETRY_MS; + const timeoutMs = opts.replyAckTimeoutMs ?? config.replyAckTimeoutMs ?? REPLY_ACK_TIMEOUT_MS; + // Holding disabled (AGENT_REPLY_ACK_TIMEOUT_MS=0): publish once and move on, + // exactly as this did before `reply_ack` existed. + if (timeoutMs === 0) { + await channel.publishUp(envelope); + return envelope.seq; + } + const deadline = Date.now() + timeoutMs; + const hold = { acked: false, done: false } as { acked: boolean; done: boolean; wake?: () => void }; + // Registered BEFORE the first publish so an ack can never arrive with + // nowhere to land (a fast orchestrator can ack before we would otherwise + // have parked). + holds.set(envelope.seq, hold); + + // A publish failure is precisely the case this loop exists for -- the + // connection is down, so nothing could have received it. Log and let the + // next attempt try again rather than abandoning the message. + const offer = async (): Promise => { + try { + await channel.publishUp(envelope); + } catch (err) { + console.error( + `[agent-runtime] failed to publish ${envelope.type} (seq ${envelope.seq}), will retry: ${err instanceof Error ? err.message : String(err)}`, + ); + } + }; + + try { + await offer(); + while (!hold.acked && !hold.done && !abort.signal.aborted && Date.now() < deadline) { + await new Promise((resolve) => { + const timer = setTimeout(() => { + hold.wake = undefined; + resolve(); + }, retryMs); + hold.wake = () => { + clearTimeout(timer); + hold.wake = undefined; + resolve(); + }; + }); + if (hold.acked || hold.done || abort.signal.aborted || Date.now() >= deadline) break; + await offer(); + } + if (!hold.acked && !hold.done && !abort.signal.aborted) { + console.error( + `[agent-runtime] ${envelope.type} (seq ${envelope.seq}) was never acknowledged within ${timeoutMs}ms; ` + + "giving up on it -- the turn that launched this run may report it as unfinished even though the work is done", + ); + } + return envelope.seq; + } finally { + holds.delete(envelope.seq); + } + }; + const pendingToolCalls = new Map void; reject: (err: Error) => void }>(); channel.onDown((msg: AgentDownMessage) => { @@ -128,8 +253,12 @@ export async function runAgent(handler: AgentHandler, opts: RunAgentOptions = {} // ask(). Anything else (agent isn't waiting) is dropped — a fresh user // turn is a new AgentRun, not this one. if (pendingAsk) { - const { resolve } = pendingAsk; + const { resolve, seq: askSeq } = pendingAsk; pendingAsk = undefined; + // The answer proves the question was received even if its ack never + // was, and re-offering an already-answered question would surface a + // stale message to the next turn. + releaseHold(askSeq); resolve(msg.message); } break; @@ -152,7 +281,20 @@ export async function runAgent(handler: AgentHandler, opts: RunAgentOptions = {} pendingToolCalls.delete(callId); pending.reject(new CancelledError(msg.reason)); } + // Nobody is waiting for these any more; wake the holds now rather than + // letting each sit out its retry interval before noticing the abort. + for (const seqHeld of [...holds.keys()]) releaseHold(seqHeld); + break; + case "reply_ack": { + // An ack for a seq we are not holding is normal and ignorable: a + // re-offer that crossed with the ack for the original produces a second + // ack after the hold is gone. + const hold = holds.get(msg.ackSeq); + if (!hold) break; + hold.acked = true; + hold.wake?.(); break; + } case "signal": // Extension point; no built-in signals yet. break; @@ -171,8 +313,13 @@ export async function runAgent(handler: AgentHandler, opts: RunAgentOptions = {} reject(new CancelledError()); return; } - pendingAsk = { resolve, reject }; - void publishUp({ type: "reply", message: question, final: false }); + // Held like a final reply: a dropped question strands the conversation + // just as badly as a dropped answer -- the user sees a failed turn while + // this agent sits waiting for an answer to something nobody ever saw. + // The seq is allocated up front so the answer can release the hold. + const askSeq = seq++; + pendingAsk = { resolve, reject, seq: askSeq }; + void publishHeld({ type: "reply", message: question, final: false }, askSeq); }), callTool: (tool, input) => new Promise((resolve, reject) => { @@ -191,7 +338,7 @@ export async function runAgent(handler: AgentHandler, opts: RunAgentOptions = {} try { const res = await handler(session); const reply: AgentReply = typeof res === "string" ? { message: res } : res; - await publishUp({ type: "reply", message: reply.message, final: true, result: reply.result }); + await publishHeld({ type: "reply", message: reply.message, final: true, result: reply.result }); } catch (err) { if (!(err instanceof CancelledError) && !abort.signal.aborted) { // A handler-supplied `code` wins (see AgentFailure). Matched by `name` @@ -208,7 +355,7 @@ export async function runAgent(handler: AgentHandler, opts: RunAgentOptions = {} : err instanceof AgentConfigError ? "config_error" : "agent_error"; - await publishUp({ type: "failed", code, message: err instanceof Error ? err.message : String(err) }); + await publishHeld({ type: "failed", code, message: err instanceof Error ? err.message : String(err) }); } // On cancellation the orchestrator already knows; exit quietly. } finally { diff --git a/packages/messaging/src/agent-protocol.ts b/packages/messaging/src/agent-protocol.ts index 7dc531e..aa8cbfe 100644 --- a/packages/messaging/src/agent-protocol.ts +++ b/packages/messaging/src/agent-protocol.ts @@ -12,8 +12,8 @@ import { z } from "zod"; * `opencode_event`, `opencode_response`, `session_idle`, `session_ended`, * `tool_call`. * - **down** (orchestrator -> agent): `prompt` (a user turn — the initial - * goal or any follow-up), `cancel`, `signal`, `opencode_request`, - * `tool_result`. + * goal or any follow-up), `cancel`, `signal`, `reply_ack`, + * `opencode_request`, `tool_result`. * * The protocol is transport-agnostic (this package deliberately has no NATS * dependency): messages are plain JSON validated by the schemas below. The @@ -162,6 +162,22 @@ export const AgentDownMessageSchema = z.discriminatedUnion("type", [ path: z.string(), body: z.unknown().optional(), }), + // Confirms a specific `reply`/`failed` up-message was actually RECEIVED, + // identified by that message's `seq`. Core NATS is fire-and-forget with no + // durability: a message published while the orchestrator has no live + // subscription (it was rolled, its pod died, the connection dropped) is + // gone forever. Since only the concluding message carries the turn's whole + // outcome, losing it turns a run that succeeded into a turn that visibly + // failed. So the agent holds its concluding message until this ack arrives, + // re-publishing periodically -- the Job's pod outlives any orchestrator + // rollout, so it can simply keep offering the answer until a replacement + // orchestrator reattaches and takes it. Narration (`progress`/`warning`) is + // deliberately NOT acked: it is best-effort commentary, worthless once the + // turn it narrated is over. + AgentMessageBaseSchema.extend({ + type: z.literal("reply_ack"), + ackSeq: z.number().int().nonnegative(), + }), // Reply to a `tool_call` up-message, correlated by `callId` (docs/adr/0028). // Exactly one of `result`/`error` is meaningful depending on `ok`, mirroring // the up-message contract's own `reply`/`failed` split but folded into one @@ -196,6 +212,7 @@ export type AgentDownMessage = | (AgentMessageBase & { type: "prompt"; message: string }) | (AgentMessageBase & { type: "cancel"; reason?: string }) | (AgentMessageBase & { type: "signal"; name: string; data?: unknown }) + | (AgentMessageBase & { type: "reply_ack"; ackSeq: number }) | (AgentMessageBase & { type: "opencode_request"; requestId: string; method: string; path: string; body?: unknown }) | (AgentMessageBase & { type: "tool_result"; callId: string; ok: boolean; result?: unknown; error?: string }); diff --git a/tools/github/Dockerfile b/tools/github/Dockerfile index 1cebaef..054a340 100644 --- a/tools/github/Dockerfile +++ b/tools/github/Dockerfile @@ -49,15 +49,26 @@ RUN apt-get update \ arm64) GH_ARCH=arm64 ;; \ *) echo "unsupported TARGETARCH: ${TARGETARCH}" >&2; exit 1 ;; \ esac \ - && curl -fsSL "https://github.com/cli/cli/releases/download/v${GH_CLI_VERSION}/gh_${GH_CLI_VERSION}_linux_${GH_ARCH}.tar.gz" -o /tmp/gh.tar.gz \ + # Saved under the asset's PUBLISHED name, not a convenience name: the names in + # the checksums file are what `sha256sum -c` opens, so downloading this to + # /tmp/gh.tar.gz made verification fail on a missing file every time. + && GH_TARBALL="gh_${GH_CLI_VERSION}_linux_${GH_ARCH}.tar.gz" \ + && curl -fsSL "https://github.com/cli/cli/releases/download/v${GH_CLI_VERSION}/${GH_TARBALL}" -o "/tmp/${GH_TARBALL}" \ && curl -fsSL "https://github.com/cli/cli/releases/download/v${GH_CLI_VERSION}/gh_${GH_CLI_VERSION}_checksums.txt" -o /tmp/gh_checksums.txt \ - && (cd /tmp && sha256sum -c <(grep "gh_${GH_CLI_VERSION}_linux_${GH_ARCH}.tar.gz" gh_checksums.txt)) \ - && tar -xzf /tmp/gh.tar.gz -C /tmp \ + # Filtered through a temp FILE rather than process substitution: `<(...)` is a + # bashism and Debian's /bin/sh is dash, which rejects it outright with + # `Syntax error: "(" unexpected` -- so this layer could never build at all. + && grep " ${GH_TARBALL}$" /tmp/gh_checksums.txt > /tmp/gh_checksum.txt \ + # A checksums file that stops listing this asset must fail the build rather + # than silently verify nothing. + && test -s /tmp/gh_checksum.txt \ + && (cd /tmp && sha256sum -c gh_checksum.txt) \ + && tar -xzf "/tmp/${GH_TARBALL}" -C /tmp \ && cp "/tmp/gh_${GH_CLI_VERSION}_linux_${GH_ARCH}/bin/gh" /usr/local/bin/gh \ && chmod a+rx /usr/local/bin/gh \ && apt-get purge -y curl \ && apt-get autoremove -y \ - && rm -rf /var/lib/apt/lists/* /tmp/gh.tar.gz /tmp/gh_checksums.txt "/tmp/gh_${GH_CLI_VERSION}_linux_${GH_ARCH}" + && rm -rf /var/lib/apt/lists/* "/tmp/${GH_TARBALL}" /tmp/gh_checksums.txt /tmp/gh_checksum.txt "/tmp/gh_${GH_CLI_VERSION}_linux_${GH_ARCH}" WORKDIR /app ENV NODE_ENV=production