From 2ac29d234b2a7263a5585b3bb58cf66954e1ebf5 Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Fri, 16 Oct 2026 23:19:30 -0400 Subject: [PATCH] fix(reactor): open a message run when an approval resumes a parked call A run opens only when the loop dequeues message.received, and a correlated approval never becomes one. A parked call rehydrated by a new reactor has no run from the process that parked it, so its approval re-dispatched the call and re-inferred with no run open, and consumers that treat no open run as idle could rebuild the agent under the running continuation. The approval now opens a run keyed on its own message id when none is open. A resume inside the still-open parking run is unchanged. --- vendor/intx-inference/PATCHES.md | 34 ++++ vendor/intx-inference/src/reactor.test.ts | 193 ++++++++++++++++++++++ vendor/intx-inference/src/reactor.ts | 8 + 3 files changed, 235 insertions(+) diff --git a/vendor/intx-inference/PATCHES.md b/vendor/intx-inference/PATCHES.md index adbbfc664..17e3fae11 100644 --- a/vendor/intx-inference/PATCHES.md +++ b/vendor/intx-inference/PATCHES.md @@ -531,6 +531,39 @@ skip calls that already have a result. branch. Re-verify after upstream changes to rehydration or to the ask-rail timeout result. +## reactor-ts-resume-opens-message-run + +`reactor.ts` — A correlated approval that resumes a parked call opens a +message run when none is open. `tryCorrelate`, after the `redispatch` or +`error_result` dispatch has consumed the pending operation and enqueued the +continuation (and before `message.correlated`), calls `openMessageRun` keyed +on the approval message's `messageId`. The re-dispatched call and its +continuation then run inside a normal `message.run.started` / +`message.run.ended` bracket, closed by the usual reply/wait/done terminal or +a failure. When a run is already open (the live case, where the run that +parked the call is still open) nothing changes: the resume continues that +run and no second bracket opens. The `gate-cleared` mode (an async-marker +operation with no suspended call) is untouched. + +Upstream opens a run only when the loop dequeues `message.received`, and a +correlated approval never becomes one. A pending operation rehydrated by a +new reactor (restart, `corbits resume`, an agent rebuild) has no run from +the parking process, so its approval drove tool execution and inference +with no run open. Consumers that treat "no open run" as idle (the TUI defers +agent reloads to `message.run.ended`) then saw the session as idle mid-call +and could rebuild the agent under the running continuation. The +`message.run.started` doc comment in `@intx/types` still describes the +dequeue-only trigger; it is left verbatim. + +**Disposition:** Promotion candidate. Upstream's run bracket has no owner +for a resume that outlives the run that parked it. +**Removal path:** Upstream PR opening a message run for a correlated resume +when none is open. +**Re-carry:** new at this patch; sits between the `tryCorrelate` dispatch +switch and the `message.correlated` emit, inside +#reactor-ts-atomic-approval-acceptance's claim. Re-verify after upstream +changes to correlation dispatch or the run-bracket helpers. + ## reactor-ts-skip-unchanged-history `reactor.ts` — Skip re-serializing unchanged history on checkpoint. @@ -762,6 +795,7 @@ revisit point is the next vendored sync (see `docs/VENDORING.md`). | reactor-ts-correlating-ids-leak | Wrap `tryCorrelate` in try/finally so `correlatingIds` clears on success dispatch paths | Alexander Guy | This ledger (#reactor-ts-correlating-ids-leak) | Next vendored sync | | reactor-ts-tool-cycle-turns-only (supersedes reactor-ts-checkpoint-after-tool-cycle) | `executeTools` writes turns to store segments only and defers the cycle commit to terminals; metadata rides along only when a resolved approval dropped a pending operation | Alexander Guy | This ledger (#reactor-ts-tool-cycle-turns-only) | Next vendored sync | | reactor-ts-skip-answered-parked-call | Drop pending operations whose parked call already has a result at rehydration, and inject no approval-timeout result for such a call | Alexander Guy | This ledger (#reactor-ts-skip-answered-parked-call) | Next vendored sync | +| reactor-ts-resume-opens-message-run | Open a message run when a correlated approval resumes a parked call and no run is open | Alexander Guy | This ledger (#reactor-ts-resume-opens-message-run) | Next vendored sync | | reactor-ts-skip-unchanged-history (+ reactor-ts-last-written-turns-revision) | Skip `contextStore.writeTurns` when `getTurnsRevision()` is unchanged | Alexander Guy | This ledger (#reactor-ts-skip-unchanged-history) | Next vendored sync | | reactor-ts-compact-publish-then-memory | Persist compact blobs, stage `writeTurns`, commit, then `replaceTurns` so memory stays on the old generation until publication | Alexander Guy | This ledger (#reactor-ts-compact-publish-then-memory) | Next vendored sync | | reactor-ts-after-checkpoint-director-only | Gate `afterCheckpoint` on `hasOverride` so auto-commits do not emit it | Alexander Guy | This ledger (#reactor-ts-after-checkpoint-director-only) | Next vendored sync | diff --git a/vendor/intx-inference/src/reactor.test.ts b/vendor/intx-inference/src/reactor.test.ts index fc43f0e86..8eb08485e 100644 --- a/vendor/intx-inference/src/reactor.test.ts +++ b/vendor/intx-inference/src/reactor.test.ts @@ -6797,6 +6797,199 @@ describe("createReactor — message.run bracket emission", () => { }); }); +describe("createReactor — message.run bracket for a resumed parked call", () => { + function askExtension() { + return createAuthzExtension({ + authorize: async () => ({ + effect: "ask" as const, + matchingGrants: [], + resolvedBy: null, + }), + approvalTimeoutMs: 60_000, + }); + } + + // What a restarted reactor loads after the original run parked on the ask + // gate: the call is in history, its pending operation is on disk, and no + // run from the parking process survives. + function parkedStore(correlationId: string): PersistedCell { + return { + turns: [suspendToolCallTurn], + pendingOperations: [ + { + correlationId, + kind: "approval", + registeredAt: Date.now(), + gateId: `pending-${correlationId}`, + timeoutAt: Date.now() + 60_000, + suspendedCall: { id: "call-ask", name: "charge_card", arguments: {} }, + }, + ], + tokenUsage: emptyUsage(), + }; + } + + function resumeDirector(): ReactorDirector { + return directorFromTable( + { + "resume.execute_tools": (e, _s, caps) => + caps.executeTools(e.calls, false, true), + "resume.tool_result": (_e, _s, caps) => caps.infer(), + "tool.done": (_e, _s, caps) => caps.infer(), + "inference.done": (_e, _s, caps) => caps.reply("resumed"), + }, + "wait", + ); + } + + function runEvents(events: ReactorEmittedEvent[]) { + const started = events.filter((e) => e.type === "message.run.started"); + const ended = events.filter((e) => e.type === "message.run.ended"); + return { started, ended }; + } + + test("an approval for a rehydrated call opens a run around the re-dispatch and its continuation", async () => { + const correlationId = "rehydrated-run-approved"; + const { reactor, events, waitFor } = createTestReactor({ + contextStore: makePersistingContextStore(parkedStore(correlationId)), + director: resumeDirector(), + inferenceRunner: makeInferenceRunner({ + type: "done", + turn: makeAssistantTurn("resumed"), + usage: emptyUsage(), + }), + toolRunner: makeToolRunner(async (call) => ({ + callId: call.id, + content: "charged", + })), + beforeToolExtensions: [askExtension()], + }); + reactor.start(); + await waitFor("reactor.start"); + + const approval = makeApprovalMessage(correlationId); + reactor.deliver(approval); + await waitFor("connector.reply"); + + const { started, ended } = runEvents(events); + expect(started).toHaveLength(1); + expect(ended).toHaveLength(1); + const run = getEvent(events, "message.run.started"); + const end = getEvent(events, "message.run.ended"); + expect(run.data.messageId).toBe(approval.headers.messageId); + expect(end.data.messageRunId).toBe(run.data.messageRunId); + expect(end.data.status).toBe("completed"); + + const types = events.map((e) => e.type); + expect(types.indexOf("message.run.started")).toBeLessThan( + types.indexOf("tool.start"), + ); + expect(types.indexOf("message.run.ended")).toBeGreaterThan( + types.indexOf("connector.reply"), + ); + + reactor.abort("admin_kill"); + await waitFor("reactor.done"); + }); + + test("a rejection for a rehydrated call opens a run around its continuation", async () => { + const correlationId = "rehydrated-run-rejected"; + const { reactor, events, waitFor } = createTestReactor({ + contextStore: makePersistingContextStore(parkedStore(correlationId)), + director: resumeDirector(), + inferenceRunner: makeInferenceRunner({ + type: "done", + turn: makeAssistantTurn("resumed"), + usage: emptyUsage(), + }), + beforeToolExtensions: [askExtension()], + }); + reactor.start(); + await waitFor("reactor.start"); + + reactor.deliver(makeApprovalMessage(correlationId, "rejected")); + await waitFor("connector.reply"); + + const { started, ended } = runEvents(events); + expect(started).toHaveLength(1); + expect(ended).toHaveLength(1); + const types = events.map((e) => e.type); + expect(types.indexOf("message.run.started")).toBeLessThan( + types.indexOf("inference.done"), + ); + expect(types.indexOf("message.run.ended")).toBeGreaterThan( + types.indexOf("connector.reply"), + ); + + reactor.abort("admin_kill"); + await waitFor("reactor.done"); + }); + + test("an approval for a call parked by the open run opens no second run", async () => { + let inferences = 0; + const inferenceRunner = async function* ( + opts: InferenceHarnessOptions, + ): AsyncGenerator { + inferences += 1; + yield { + type: "inference.done", + seq: opts.nextSeq(), + data: { + turn: inferences === 1 ? suspendToolCallTurn : makeAssistantTurn("ok"), + usage: emptyUsage(), + source: TEST_SOURCE, + }, + }; + }; + const { reactor, events, waitFor } = createTestReactor({ + director: directorFromTable( + { + "message.received": (_e, _s, caps) => caps.infer(), + "inference.done": (_e, _s, caps) => + inferences === 1 + ? caps.executeTools([ + { id: "call-ask", name: "charge_card", arguments: {} }, + ]) + : caps.reply("done"), + "resume.execute_tools": (e, _s, caps) => + caps.executeTools(e.calls, false, true), + "tool.done": (_e, _s, caps) => caps.infer(), + }, + "wait", + ), + inferenceRunner, + toolRunner: makeToolRunner(async (call) => ({ + callId: call.id, + content: "charged", + })), + beforeToolExtensions: [askExtension()], + }); + reactor.start(); + const inbound = makeInboundMessage(); + reactor.deliver(inbound); + const blocked = await waitFor("reactor.gate.blocked"); + if (blocked.type !== "reactor.gate.blocked") throw new Error("unreachable"); + const correlationId = blocked.data.correlationId; + if (correlationId === undefined) throw new Error("expected correlationId"); + + reactor.deliver(makeApprovalMessage(correlationId)); + await waitFor("message.run.ended"); + + const { started, ended } = runEvents(events); + expect(started).toHaveLength(1); + expect(ended).toHaveLength(1); + const run = getEvent(events, "message.run.started"); + expect(run.data.messageId).toBe(inbound.headers.messageId); + const types = events.map((e) => e.type); + expect(types.indexOf("message.run.ended")).toBeGreaterThan( + types.indexOf("connector.reply"), + ); + + reactor.abort("admin_kill"); + await waitFor("reactor.done"); + }); +}); + describe("createReactor — source failover", () => { // Simulates a priority-ordered source list the reactor fails over through. // `resultFor(id, attemptOnThisSource)` decides what the inference runner diff --git a/vendor/intx-inference/src/reactor.ts b/vendor/intx-inference/src/reactor.ts index 13042206e..0445cc45a 100644 --- a/vendor/intx-inference/src/reactor.ts +++ b/vendor/intx-inference/src/reactor.ts @@ -845,6 +845,14 @@ export function createReactor(config: ReactorConfig): Reactor { break; } } + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-resume-opens-message-run + // A parked call normally resumes inside the run that parked it. One + // rehydrated from the store after a restart has no run open, so the + // approval opens one for the re-dispatch and its continuation; the + // usual reply/wait/done terminal or failure closes it. + if (dispatch.mode !== "gate-cleared" && currentMessageRunId === null) { + openMessageRun(message.headers.messageId); + } // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance // Observers may reenter: publish only after consumption and continuation // enqueue, while the claim still excludes a duplicate delivery.