From 0f31491c7c3228a5e94e95e2816f47a168c808d7 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 14 Sep 2026 21:39:47 -0700 Subject: [PATCH] fix(approval): discard expired decisions instead of ordinary input --- docs/ARCHITECTURE.md | 15 + src/session/approval-resume.test.ts | 54 ++ src/session/approval-resume.ts | 4 +- .../unit/reactor-approval-acceptance.test.ts | 521 ++++++++++++++++++ vendor/intx-inference/PATCHES.md | 26 + vendor/intx-inference/src/correlation.ts | 6 +- vendor/intx-inference/src/reactor.ts | 126 +++-- 7 files changed, 703 insertions(+), 49 deletions(-) create mode 100644 tests/unit/reactor-approval-acceptance.test.ts diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 8c4dbb52f..11006cdaf 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -14,6 +14,21 @@ The reactor (from `@intx/agent`) drives a single agent turn-by-turn. Each turn i This repeats until the director emits `capabilities.done()`. +Explicit approval messages use `headers.interchangeType` (`approval.granted` or +`approval.denied`) as intent, not authentication. The reactor accepts one only +while its exact pending operation and matching gate remain live, before both +epoch deadlines (equality expires), with a body agreeing with the header. Typed +messages without an eligible suspended call, expired/consumed decisions and +concurrent duplicates are discarded before body parsing, never ordinary input. +Validator false/throw also discards typed decisions but leaves a still-live +operation retryable; untyped and nonapproval response routing is unchanged. +After validation and synchronous grant callbacks, the reactor rechecks ownership, +deadlines and lifecycle, then consumes the operation and enqueues its continuation +before publishing `message.correlated`. The claim is released on every settled +exit; a never-settling validator retains it. Extension side effects are not rolled +back. Discards produce no acceptance receipt: the TUI's existing bounded 30-second +acceptance wait remains unchanged. + ### Reactor Events (Partial) This table is not the full reactor/stream event vocabulary. `tool.done` is diff --git a/src/session/approval-resume.test.ts b/src/session/approval-resume.test.ts index e86577bfb..403054dc3 100644 --- a/src/session/approval-resume.test.ts +++ b/src/session/approval-resume.test.ts @@ -15,6 +15,7 @@ import { createApprovalResume, resolveParkedCallIdFromStore, } from "./approval-resume.js"; +import { createSessionOperationQueue } from "../tui/delivery-queue.js"; function assistantTurn( calls: { id: string; name: string; command: string }[], @@ -106,6 +107,59 @@ function deliveredCorrelationId(message: InboundMessage): string { return correlationId; } +describe("approval decision intent headers", () => { + for (const allow of [true, false]) { + test(`preserves ${allow ? "granted" : "denied"} intent and correlation through the session queue`, async () => { + const delivered: InboundMessage[] = []; + const queue = createSessionOperationQueue(); + let release!: () => void; + const held = new Promise((resolve) => { + release = resolve; + }); + queue.enqueue(() => held); + const agent = { + history: async () => [], + deliver: (message: InboundMessage) => { + delivered.push(message); + }, + }; + const gate = { + resolveSuspended: async () => ({ allow, message: "decision reason" }), + } as unknown as PermissionGate; + const resume = createApprovalResume({ + getAgent: () => agent, + gate, + resolveParkedCallId: () => "typed-call", + deliver: (message) => + queue.enqueue(async () => { + agent.deliver(message); + }), + }); + const handling = resume.handle( + suspension("typed-correlation", "echo alpha"), + ); + expect(delivered).toHaveLength(0); + release(); + expect(await handling).toBe(true); + await queue.awaitTail(); + expect(delivered).toHaveLength(1); + const message = delivered[0]; + if (message === undefined) throw new Error("expected decision"); + expect(message.headers.interchangeType).toBe( + allow ? "approval.granted" : "approval.denied", + ); + expect(deliveredCorrelationId(message)).toBe("typed-correlation"); + expect(decisionBody(message).outcome).toBe( + allow ? "approved" : "rejected", + ); + if (!allow) + expect(JSON.parse(message.content ?? "").message).toBe( + "decision reason", + ); + }); + } +}); + describe("approval-resume parallel-parked approvals", () => { test("delivers A's decision when a different parked call's approval times out", async () => { const { resume, delivered } = setup({ diff --git a/src/session/approval-resume.ts b/src/session/approval-resume.ts index b412cf44a..bb6db8773 100644 --- a/src/session/approval-resume.ts +++ b/src/session/approval-resume.ts @@ -106,7 +106,7 @@ function decisionMessage( correlationId: string, outcome: "approved" | "rejected", message?: string, -) { +): InboundMessage { const body: { outcome: "approved" | "rejected"; message?: string } = { outcome, }; @@ -119,6 +119,8 @@ function decisionMessage( date: new Date().toISOString(), messageId: `approval-${correlationId}`, interchangeCorrelationId: correlationId, + interchangeType: + outcome === "approved" ? "approval.granted" : "approval.denied", }, flags: [], content: JSON.stringify(body), diff --git a/tests/unit/reactor-approval-acceptance.test.ts b/tests/unit/reactor-approval-acceptance.test.ts new file mode 100644 index 000000000..655eba097 --- /dev/null +++ b/tests/unit/reactor-approval-acceptance.test.ts @@ -0,0 +1,521 @@ +import { afterEach, describe, expect, spyOn, test } from "bun:test"; +import { createInboundMessage } from "@intx/mime"; +import type { + ContextStore, + InboundMessage, + PendingOperation, + ReactorInboundEvent, + ReactorState, +} from "@intx/types/runtime"; +import { + createReactor, + type ReactorConfig, + type ReactorEmittedEvent, +} from "../../vendor/intx-inference/src/reactor.js"; +import { createDefaultDependencies } from "../../vendor/intx-inference/src/providers/index.js"; +import { createSessionOperationQueue } from "../../src/tui/delivery-queue.js"; + +const call = { id: "parked-call", name: "test_tool", arguments: {} }; +const epoch = 100_000; +function pending( + timeoutAt: number | undefined = epoch + 1000, +): PendingOperation { + return { + correlationId: "approval", + kind: "approval", + gateId: "gate", + registeredAt: epoch, + suspendedCall: call, + ...(timeoutAt === undefined ? {} : { timeoutAt }), + }; +} +function decision( + outcome: "approved" | "rejected" = "approved", +): InboundMessage { + const message = createInboundMessage({ + from: "signal@local", + to: "agent@local", + correlationId: "approval", + content: JSON.stringify({ outcome }), + }); + message.headers.interchangeType = + outcome === "approved" ? "approval.granted" : "approval.denied"; + return message; +} +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise((yes) => { + resolve = yes; + }); + return { promise, resolve }; +} +async function drain() { + for (let i = 0; i < 100; i++) await Promise.resolve(undefined); +} + +const cleanups: (() => Promise)[] = []; +afterEach(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup(); +}); + +async function harness( + options: { + op?: PendingOperation; + validator?: ReactorConfig["correlationValidator"]; + grant?: () => void; + event?: (event: ReactorEmittedEvent) => void; + } = {}, +) { + let now = epoch; + const clock = spyOn(Date, "now").mockImplementation(() => now); + const timers: (() => void)[] = []; + const timer = spyOn(globalThis, "setTimeout").mockImplementation((( + callback: () => void, + ) => { + timers.push(callback); + return timers.length; + }) as typeof setTimeout); + const clear = spyOn(globalThis, "clearTimeout").mockImplementation( + (): void => undefined, + ); + const events: ReactorEmittedEvent[] = []; + const directed: ReactorInboundEvent[] = []; + const snapshots: ReactorState[] = []; + const grants: string[] = []; + const tools: string[] = []; + let hold: Promise | undefined; + const closing = deferred(); + const store: ContextStore = { + load: async () => ({ + turns: [ + { + role: "assistant", + content: [{ type: "tool_call", ...call }], + timestamp: epoch, + }, + ], + pendingOperations: [options.op ?? pending()], + tokenUsage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + thinking: 0, + }, + connectorState: null, + }), + setConnectorState: () => undefined, + commit: async ({ message }) => ({ hash: "test", message, timestamp: now }), + branch: async () => undefined, + log: async () => [], + readAt: async () => [], + writeBlob: async () => undefined, + readBlob: async () => { + throw new Error("unused"); + }, + writePrompt: async () => undefined, + writeResponse: async () => undefined, + writeManifest: async () => undefined, + writeTurns: async () => undefined, + writeMetadata: async () => undefined, + readManifestHistory: async () => [], + }; + const reactor = createReactor({ + sessionId: "atomic-approval", + source: { + id: "test", + provider: "anthropic", + model: "test", + baseURL: "https://example.com", + credentialId: "test", + }, + deps: createDefaultDependencies(), + contextStore: store, + gateTimeout: 1000, + ...(options.validator === undefined + ? {} + : { correlationValidator: options.validator }), + beforeToolExtensions: [ + { + beforeTool: async () => ({ type: "allow" }), + grantOneShot(id) { + grants.push(id); + options.grant?.(); + }, + }, + ], + toolRunner: { + async run(tool) { + tools.push(tool.id); + return { callId: tool.id, content: "executed" }; + }, + }, + onEvent(event) { + events.push(event); + options.event?.(event); + }, + director: { + async decide(event, state, caps) { + directed.push(event); + snapshots.push(state); + if (hold !== undefined) await Promise.race([hold, closing.promise]); + if (event.type === "resume.execute_tools") + return caps.executeTools(event.calls, false, true); + return caps.wait(); + }, + }, + }); + cleanups.push(async () => { + closing.resolve(undefined); + hold = undefined; + reactor.abort("admin_kill"); + await drain(); + timer.mockRestore(); + clear.mockRestore(); + clock.mockRestore(); + }); + reactor.start(); + await drain(); + expect(events.some((e) => e.type === "reactor.start")).toBe(true); + return { + reactor, + events, + directed, + snapshots, + grants, + tools, + advance(value: number) { + now = value; + }, + expire() { + now = epoch + 1000; + const fire = timers[0]; + if (!fire) throw new Error("gate timer missing"); + fire(); + }, + hold(value: Promise | undefined) { + hold = value; + }, + }; +} +function accepted(h: Awaited>) { + return h.events.filter((e) => e.type === "message.correlated"); +} +function received(h: Awaited>) { + return h.events.filter((e) => e.type === "message.received"); +} +function resumes(h: Awaited>) { + return h.directed.filter( + (e) => e.type === "resume.execute_tools" || e.type === "resume.tool_result", + ); +} +function expectDiscard(h: Awaited>) { + expect(accepted(h)).toHaveLength(0); + expect(received(h)).toHaveLength(0); + expect(h.grants).toHaveLength(0); + expect(h.tools).toHaveLength(0); +} + +describe("atomic typed approval acceptance", () => { + test("expiry before timeout result publication cannot become ordinary input", async () => { + const h = await harness(); + const hold = deferred(); + h.hold(hold.promise); + h.reactor.deliver( + createInboundMessage({ + from: "user@local", + to: "agent@local", + content: "hold director", + }), + ); + await drain(); + h.expire(); + const stale = decision(); + stale.content = "not JSON"; + h.reactor.deliver(stale); + await drain(); + expect(accepted(h)).toHaveLength(0); + expect(received(h)).toHaveLength(1); + expect(h.grants).toHaveLength(0); + expect(h.tools).toHaveLength(0); + expect(resumes(h)).toHaveLength(0); + hold.resolve(undefined); + h.hold(undefined); + await drain(); + expect(resumes(h)).toHaveLength(1); + expect(JSON.stringify(h.snapshots.at(-1)?.turns)).toContain( + "approval timed out", + ); + expect(JSON.stringify(h.snapshots.at(-1)?.turns)).not.toContain("not JSON"); + }); + test("actual session operation queue drains an expired approval to Reactor.deliver", async () => { + const h = await harness(); + const queue = createSessionOperationQueue(); + const held = deferred(); + queue.enqueue(() => held.promise); + queue.enqueue(async () => { + h.reactor.deliver(decision()); + }); + await drain(); + h.expire(); + held.resolve(undefined); + await queue.awaitTail(); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(1); + }); + test("expiry during validator await discards the claimed decision", async () => { + const valid = deferred(); + const h = await harness({ validator: { validate: () => valid.promise } }); + h.reactor.deliver(decision()); + await drain(); + h.expire(); + valid.resolve(true); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(1); + }); + for (const outcome of ["approved", "rejected"] as const) + test(`live ${outcome} and concurrent/postconsumption duplicates resume once`, async () => { + const valid = deferred(); + const h = await harness({ validator: { validate: () => valid.promise } }); + h.reactor.deliver(decision(outcome)); + h.reactor.deliver(decision(outcome)); + valid.resolve(true); + await drain(); + h.reactor.deliver(decision(outcome)); + await drain(); + expect(accepted(h)).toHaveLength(1); + expect(received(h)).toHaveLength(0); + expect(resumes(h)).toHaveLength(1); + expect(h.grants).toHaveLength(outcome === "approved" ? 1 : 0); + expect(h.tools).toHaveLength(outcome === "approved" ? 1 : 0); + expect(JSON.stringify(h.snapshots.at(-1)?.turns)).toContain( + outcome === "approved" ? "executed" : "denied by approver", + ); + }); + for (const deadline of [epoch - 1, epoch]) + test(`persisted deadline ${deadline} is not revived by the clamped timer`, async () => { + const h = await harness({ op: pending(deadline) }); + h.reactor.deliver(decision()); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(0); + }); + test("deadline equality wins before actual timer callback", async () => { + const h = await harness(); + h.advance(epoch + 1000); + h.reactor.deliver(decision()); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(0); + }); + test("termination during validation prevents successful acceptance", async () => { + const valid = deferred(); + const h = await harness({ validator: { validate: () => valid.promise } }); + h.reactor.deliver(decision()); + h.reactor.abort("admin_kill"); + await drain(); + valid.resolve(true); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(0); + }); + for (const failure of ["false", "throw"] as const) + test(`validator ${failure} releases claim for valid retry`, async () => { + let first = true; + const h = await harness({ + validator: { + async validate() { + if (!first) return true; + first = false; + if (failure === "throw") throw new Error("validator failed"); + return false; + }, + }, + }); + h.reactor.deliver(decision()); + await drain(); + expectDiscard(h); + h.reactor.deliver(decision()); + await drain(); + expect(accepted(h)).toHaveLength(1); + expect(resumes(h)).toHaveLength(1); + expect(h.tools).toHaveLength(1); + }); + for (const suspended of [true, false]) + test(`contradictory header/body with suspendedCall=${suspended} cannot resume`, async () => { + const op = pending(); + if (!suspended) delete op.suspendedCall; + const h = await harness({ op }); + const message = decision(); + message.content = JSON.stringify({ outcome: "rejected" }); + h.reactor.deliver(message); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(0); + if (suspended) + expect(h.events.some((e) => e.type === "reactor.error")).toBe(true); + }); + for (const callback of ["grant", "event"] as const) + test(`synchronous duplicate from ${callback} does not resume twice`, async () => { + let deliver = (): void => undefined; + const duplicate = () => deliver(); + const h = await harness( + callback === "grant" + ? { grant: duplicate } + : { + event: (event) => { + if (event.type === "message.correlated") duplicate(); + }, + }, + ); + deliver = () => h.reactor.deliver(decision()); + h.reactor.deliver(decision()); + await drain(); + expect(accepted(h)).toHaveLength(1); + expect(received(h)).toHaveLength(0); + expect(h.grants).toHaveLength(1); + expect(h.tools).toHaveLength(1); + expect(resumes(h)).toHaveLength(1); + }); + for (const action of ["throw", "abort"] as const) + test(`grant hook ${action} cannot publish successful acceptance`, async () => { + let abort = (): void => undefined; + const h = await harness({ + grant() { + if (action === "throw") throw new Error("grant failed"); + abort(); + }, + }); + abort = () => h.reactor.abort("admin_kill"); + h.reactor.deliver(decision()); + await drain(); + expect(accepted(h)).toHaveLength(0); + expect(received(h)).toHaveLength(0); + expect(h.grants).toHaveLength(1); + expect(h.tools).toHaveLength(0); + expect(resumes(h)).toHaveLength(0); + }); + for (const action of ["throw", "abort"] as const) + test(`acceptance observer ${action} happens after consumption, without a second resume`, async () => { + let reenter = (): void => undefined; + const h = await harness({ + event(event) { + if (event.type !== "message.correlated") return; + reenter(); + if (action === "throw") throw new Error("observer failed"); + }, + }); + reenter = () => { + h.reactor.deliver(decision()); + if (action === "abort") h.reactor.abort("admin_kill"); + }; + h.reactor.deliver(decision()); + await drain(); + expect(accepted(h)).toHaveLength(1); + expect(received(h)).toHaveLength(0); + expect(h.grants).toHaveLength(1); + expect(resumes(h).length).toBeLessThanOrEqual(1); + expect(h.tools.length).toBeLessThanOrEqual(1); + expect(h.events.some((e) => e.type === "reactor.done")).toBe(true); + }); + test("gate deadline applies when the operation has no persisted deadline", async () => { + const op = pending(); + delete op.timeoutAt; + const h = await harness({ op }); + h.advance(epoch + 1000); + h.reactor.deliver(decision()); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(0); + }); + test("grant callback advancing the clock cannot accept an expired operation", async () => { + let expire = (): void => undefined; + const h = await harness({ grant: () => expire() }); + expire = () => h.advance(epoch + 1000); + h.reactor.deliver(decision()); + await drain(); + expect(accepted(h)).toHaveLength(0); + expect(received(h)).toHaveLength(0); + expect(h.grants).toHaveLength(1); + expect(h.tools).toHaveLength(0); + expect(resumes(h)).toHaveLength(0); + }); + for (const correlation of [undefined, "unknown"]) + test(`typed unknown correlation ${correlation} discards before parsing`, async () => { + const h = await harness(); + const message = decision(); + message.content = "malformed"; + if (correlation === undefined) + delete message.headers.interchangeCorrelationId; + else message.headers.interchangeCorrelationId = correlation; + h.reactor.deliver(message); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(0); + expect(h.events.some((e) => e.type === "reactor.error")).toBe(false); + }); + test("live malformed typed decisions retain the fatal error policy", async () => { + const h = await harness(); + const message = decision(); + message.content = "malformed"; + h.reactor.deliver(message); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(0); + expect(h.events.some((e) => e.type === "reactor.error")).toBe(true); + expect(h.events.some((e) => e.type === "reactor.done")).toBe(true); + }); + for (const failure of ["false", "throw"] as const) + test(`nonapproval validator ${failure} retains ordinary fallback`, async () => { + const h = await harness({ + validator: { + async validate() { + if (failure === "throw") throw new Error("invalid sender"); + return false; + }, + }, + }); + const message = decision(); + message.headers.interchangeType = "offering.response"; + h.reactor.deliver(message); + await drain(); + expect(received(h)).toHaveLength(1); + expect(accepted(h)).toHaveLength(0); + expect(h.grants).toHaveLength(0); + expect(resumes(h)).toHaveLength(0); + }); + test("ordinary unknown correlations, approval-shaped text and nonapproval responses retain routing", async () => { + const op = pending(); + delete op.suspendedCall; + const h = await harness({ op }); + for (const correlationId of [undefined, "unknown"]) { + h.reactor.deliver( + createInboundMessage({ + from: "user@local", + to: "agent@local", + content: JSON.stringify({ outcome: "approved" }), + ...(correlationId === undefined ? {} : { correlationId }), + }), + ); + } + h.reactor.deliver( + createInboundMessage({ + from: "worker@local", + to: "agent@local", + content: "async response", + correlationId: "approval", + }), + ); + await drain(); + expect(received(h)).toHaveLength(2); + expect(accepted(h)).toHaveLength(1); + expect(h.grants).toHaveLength(0); + expect(h.tools).toHaveLength(0); + expect(JSON.stringify(h.snapshots.at(-1)?.turns)).toContain( + "async response", + ); + expect(JSON.stringify(h.snapshots.at(-1)?.turns)).toContain("approved"); + }); +}); diff --git a/vendor/intx-inference/PATCHES.md b/vendor/intx-inference/PATCHES.md index 0461fcd3f..7b100bc47 100644 --- a/vendor/intx-inference/PATCHES.md +++ b/vendor/intx-inference/PATCHES.md @@ -26,6 +26,32 @@ Re-carryable, "Kill candidate" = Droppable, and "Companion" entries are Re-carryable but ride their primary patch's disposition (they ship out or die with it). +## reactor-ts-atomic-approval-acceptance + +**Promotion candidate.** `reactor.ts` classifies native `approval.granted` / +`approval.denied` headers as conditional decision intent, not authentication. +Its private correlation disposition distinguishes ordinary input from discarded +approvals; stale, ineligible and already-claimed decisions never enter history +or emit `message.received`. `correlation.ts` documents the typed-only exception +to validator false/throw fallback, without changing the validator signature. + +The claim spans validation, parsing, dispatch, consumption and publication in +one `try/finally`. Before parsing and after synchronous grant hooks, acceptance +requires the same pending operation and live gate, an accepting lifecycle, and +strictly unexpired gate and persisted deadlines. A clamped rehydration timer +cannot revive an expired persisted operation. Typed targets without a suspended +call are ineligible; eligible bodies must agree with the header. Consumption +and continuation enqueue run without callbacks before `message.correlated`. +Grant hooks may reenter or abort; a failed transition publishes no acceptance. +Arbitrary extension side effects are not rolled back, and observers throwing +after the transition retain the existing fatal-error policy. + +**Re-carry:** preserve all acceptance checks and the ordinary-message path when +merging upstream correlation changes. Deterministic real-reactor coverage lives +in `tests/unit/reactor-approval-acceptance.test.ts`. A never-settling validator +retains its claim until settlement; there are no tombstones or discard receipts. +This is a local patch, not a change to the upstream pin or retrieval metadata. + ### 2026-09-07 re-sync (upstream `0205b07b`) Every entry below was re-carried against the new pin; none was dropped as diff --git a/vendor/intx-inference/src/correlation.ts b/vendor/intx-inference/src/correlation.ts index 974a03d03..9ab5d95a5 100644 --- a/vendor/intx-inference/src/correlation.ts +++ b/vendor/intx-inference/src/correlation.ts @@ -15,7 +15,11 @@ import type { InboundMessage, PendingOperation } from "@intx/types/runtime"; export interface CorrelationValidator { /** * Return true if `message` is a valid resolution for `pending`. - * False causes the message to be delivered as a regular uncorrelated event. + * False (or a thrown error) falls back to a regular uncorrelated event, + * except explicit approval.granted/approval.denied messages: those are + * discarded, leaving a still-live pending operation available for retry. + * These headers classify intent; they do not authenticate the sender. + * Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance */ validate( pending: PendingOperation, diff --git a/vendor/intx-inference/src/reactor.ts b/vendor/intx-inference/src/reactor.ts index 379adc5cc..08f18e732 100644 --- a/vendor/intx-inference/src/reactor.ts +++ b/vendor/intx-inference/src/reactor.ts @@ -495,9 +495,10 @@ export function createReactor(config: ReactorConfig): Reactor { | { mode: "gate-cleared" } | { mode: "error_result"; result: ToolResult }; - // Decide how a correlated approval-kind pending operation resumes, granting - // any one-shot bypass synchronously so no delivery can interleave between the - // grant and the re-dispatch enqueued by the caller. An operation that carries + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance + // Decide how a correlated approval-kind pending operation resumes. Grant + // callbacks can reenter; the caller owns the claim and final live checks. + // An operation that carries // a `suspendedCall` is an ask-flow suspension: the approver's decision routes // it down the re-dispatch rail. An operation without one is an async-tool // pending marker, which resumes on the normal gate-cleared rail. @@ -509,6 +510,8 @@ export function createReactor(config: ReactorConfig): Reactor { function resumePendingOperation( op: PendingOperation, message: InboundMessage, + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance + expectedOutcome: "approved" | "rejected" | undefined, ): ResumeDispatch { if (op.suspendedCall === undefined) { return { mode: "gate-cleared" }; @@ -536,6 +539,13 @@ export function createReactor(config: ReactorConfig): Reactor { ); } + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance + if (expectedOutcome !== undefined && decision.outcome !== expectedOutcome) { + throw new Error( + `Correlated approval decision for ${op.correlationId} contradicts its header`, + ); + } + switch (op.kind) { case "approval": switch (decision.outcome) { @@ -569,44 +579,63 @@ export function createReactor(config: ReactorConfig): Reactor { } } - async function tryCorrelate(message: InboundMessage): Promise { - const correlationId = message.headers.interchangeCorrelationId; - if (correlationId === undefined) return false; - - if (correlatingIds.has(correlationId)) return false; - const pending = correlations.lookup(correlationId); - if (pending === undefined) return false; - - correlatingIds.add(correlationId); + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance + type CorrelationDisposition = "correlated" | "ordinary" | "discarded"; - if (correlationValidator !== undefined) { - let valid: boolean; - try { - valid = await correlationValidator.validate(pending, message); - } catch (cause) { - logger.warn`Correlation validator threw for ${correlationId}: ${cause}`; - correlatingIds.delete(correlationId); - return false; - } - if (!valid) { - correlatingIds.delete(correlationId); - return false; - } + async function tryCorrelate( + message: InboundMessage, + ): Promise { + const expectedOutcome = + message.headers.interchangeType === "approval.granted" + ? "approved" + : message.headers.interchangeType === "approval.denied" + ? "rejected" + : undefined; + const typedApproval = expectedOutcome !== undefined; + const unmatched = typedApproval ? "discarded" : "ordinary"; + const correlationId = message.headers.interchangeCorrelationId; + if (correlationId === undefined || correlatingIds.has(correlationId)) { + return unmatched; } + const pending = correlations.lookup(correlationId); + if (pending === undefined) return unmatched; + const claimedGate = gates.findByCorrelationId(correlationId); + + const isLiveApproval = (): boolean => { + const now = Date.now(); + return ( + running && !done && !shutdownStarted && + !queue.some((event) => event.type === "abort") && + correlations.lookup(correlationId) === pending && + pending.kind === "approval" && + pending.suspendedCall !== undefined && + claimedGate !== undefined && + claimedGate.gateId === pending.gateId && + gates.findByCorrelationId(correlationId) === claimedGate && + now < claimedGate.timeoutAt && + (pending.timeoutAt === undefined || now < pending.timeoutAt) + ); + }; - // Capture the operation before removal so the resume dispatch can read its - // kind and suspended call. Removal happens only after the dispatch is - // decided, all inside this correlatingIds-guarded critical section so a - // double-deliver early-returns rather than double-dispatching. - const op = pending; - - // A finally clears the in-flight marker on every exit — success included. - // Without it the success path leaves the id in the set forever, leaking - // one entry per correlated message for the life of the session. - // // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-correlating-ids-leak + correlatingIds.add(correlationId); try { - const dispatch = resumePendingOperation(op, message); + if (typedApproval && !isLiveApproval()) return "discarded"; + if (correlationValidator !== undefined) { + let valid: boolean; + try { + valid = await correlationValidator.validate(pending, message); + } catch (cause) { + logger.warn`Correlation validator threw for ${correlationId}: ${cause}`; + return unmatched; + } + if (!valid) return unmatched; + } + if (typedApproval && !isLiveApproval()) return "discarded"; + const dispatch = resumePendingOperation(pending, message, expectedOutcome); + // Extension grants are synchronous but can reenter, abort or advance the + // clock. Only the internal consume-and-enqueue below is callback-free. + if (typedApproval && !isLiveApproval()) return "discarded"; const gate = gates.findByCorrelationId(correlationId); switch (dispatch.mode) { @@ -668,17 +697,18 @@ export function createReactor(config: ReactorConfig): Reactor { break; } } + // 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. + emit({ + type: "message.correlated", + seq: nextSeq(), + data: { message, correlationId }, + }); + return "correlated"; } finally { correlatingIds.delete(correlationId); } - - emit({ - type: "message.correlated", - seq: nextSeq(), - data: { message, correlationId }, - }); - - return true; } // ------------------------------------------------------------------------- @@ -1809,9 +1839,10 @@ export function createReactor(config: ReactorConfig): Reactor { function processDelivery(message: InboundMessage): void { void (async () => { - let correlated: boolean; + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance + let disposition: CorrelationDisposition; try { - correlated = await tryCorrelate(message); + disposition = await tryCorrelate(message); } catch (cause) { // A correlation-path invariant failed (e.g. a malformed approval // decision). Surface it as a fatal reactor error rather than a silent @@ -1830,7 +1861,8 @@ export function createReactor(config: ReactorConfig): Reactor { } return; } - if (!correlated) { + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance + if (disposition === "ordinary") { emit({ type: "message.received", seq: nextSeq(),