Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
33 changes: 33 additions & 0 deletions vendor/intx-inference/PATCHES.md
Original file line number Diff line number Diff line change
Expand Up @@ -498,6 +498,38 @@ batch committed); the turns still reach disk segments, so only the cycle's
manifest records and metadata staging stay uncommitted. Intentionally not
committed; re-verify after upstream doom-loop accounting changes.

## 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.
Expand Down Expand Up @@ -728,6 +760,7 @@ revisit point is the next vendored sync (see `docs/VENDORING.md`).
| harness-ts-inactivity-on-semantic-progress | Gate the inactivity watchdog's re-arm on parsed events, not raw SSE bytes | Alexander Guy <alexander.guy@pm.me> | This ledger (#harness-ts-inactivity-on-semantic-progress) | Next vendored sync |
| 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-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 |
Expand Down
193 changes: 193 additions & 0 deletions vendor/intx-inference/src/reactor.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -6678,6 +6678,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
Expand Down
8 changes: 8 additions & 0 deletions vendor/intx-inference/src/reactor.ts
Original file line number Diff line number Diff line change
Expand Up @@ -834,6 +834,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.
Expand Down
Loading