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.

Author: bwachman

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 <alexander.guy@pm.me> | 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 <alexander.guy@pm.me> | 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 <alexander.guy@pm.me> | 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 <alexander.guy@pm.me> | 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 <alexander.guy@pm.me> | 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 <alexander.guy@pm.me> | 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 <alexander.guy@pm.me> | This ledger (#reactor-ts-after-checkpoint-director-only) | Next vendored sync |

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<InferenceEvent> {
+      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

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.