diff --git a/src/agent/background-shell-tool.test.ts b/src/agent/background-shell-tool.test.ts index 025a52b46..b745c39b2 100644 --- a/src/agent/background-shell-tool.test.ts +++ b/src/agent/background-shell-tool.test.ts @@ -3,6 +3,7 @@ import { describe, expect, test } from "bun:test"; import { spawnSync } from "node:child_process"; import { randomUUID } from "node:crypto"; import { createPermissionGate } from "../permission/gate.js"; +import { shellCollectDefinition } from "./background-shell-tool.js"; import { createAgentToolset } from "./tools.js"; import type { BackgroundShellExit } from "../shell/background-shell.js"; import { buildShellBackgroundMessage } from "../session/runtime-assembly.js"; @@ -151,3 +152,11 @@ describe("background shell through the agent toolset", () => { expect(probe.status).not.toBe(0); }); }); + +describe("shell_collect tool copy", () => { + test("names the doom-loop exemption for still-running polls", () => { + expect(shellCollectDefinition.description).toContain("doom-loop guard"); + expect(shellCollectDefinition.description).toContain("liveness"); + expect(shellCollectDefinition.description).toContain("running"); + }); +}); diff --git a/src/agent/background-shell-tool.ts b/src/agent/background-shell-tool.ts index ddd38b759..8f79a4d15 100644 --- a/src/agent/background-shell-tool.ts +++ b/src/agent/background-shell-tool.ts @@ -21,7 +21,10 @@ export const shellCollectDefinition: ToolDefinition = { 'action="collect" returns the result once finished (or status running); ' + 'action="cancel" kills the process group. Completion also arrives as a ' + "system message on a later turn — collect is for polling or retrieving " + - "output again after eviction risk.", + 'output again after eviction risk. A "running" result is liveness, not a ' + + "stall or a crash: repeated identical collects of a still-running shell " + + "are exempt from the run's doom-loop guard, so keep polling rather than " + + "treating it as a failure.", inputSchema: { type: "object", properties: { diff --git a/src/provider/inference-dependencies.ts b/src/provider/inference-dependencies.ts index 0c9238941..c447f4ec5 100644 --- a/src/provider/inference-dependencies.ts +++ b/src/provider/inference-dependencies.ts @@ -17,6 +17,7 @@ import { } from "./codex-responses.js"; import { GROK_RESPONSES_PROVIDER } from "./grok-responses.js"; import { withReplaySanitizer } from "./replay-sanitizer.js"; +import { isPollOnlyPendingBatch } from "../subagent/poll-exempt.js"; import { OPENCODE_GO_PROVIDER_ID } from "../../packages/opencode-go/src/index.js"; import { BIFROST_PROVIDER } from "./bifrost-adapter.js"; import { OPENAI_RESPONSES_PROVIDER } from "./openai-responses.js"; @@ -89,6 +90,7 @@ export function createInferenceDependencies(): Promise { .then((deps) => ({ ...deps, fetch: withCodexContentTypeRepair(deps.fetch), + isPollOnlyPendingBatch, })); } return cached; diff --git a/src/subagent/agent-fleet.test.ts b/src/subagent/agent-fleet.test.ts index a5d96f0f0..286a21f73 100644 --- a/src/subagent/agent-fleet.test.ts +++ b/src/subagent/agent-fleet.test.ts @@ -8,6 +8,7 @@ import { createSpawnAgentTool, createWaitAgentsTool, createListAgentsTool, + waitAgentsToolDefinition, MAX_FLEET_RECORDS, type AgentFleetDeps, } from "./agent-fleet.js"; @@ -3176,3 +3177,11 @@ describe("wait_agents occupancy yield (CL-7518)", () => { gate.resolve({ report: "ok" }); }); }); + +describe("wait_agents tool copy", () => { + test("names the doom-loop exemption for still-pending polls", () => { + expect(waitAgentsToolDefinition.description).toContain("doom-loop guard"); + expect(waitAgentsToolDefinition.description).toContain("liveness"); + expect(waitAgentsToolDefinition.description).toContain("timeout"); + }); +}); diff --git a/src/subagent/agent-fleet.ts b/src/subagent/agent-fleet.ts index 936ecbe01..faee62119 100644 --- a/src/subagent/agent-fleet.ts +++ b/src/subagent/agent-fleet.ts @@ -560,7 +560,10 @@ export const waitAgentsToolDefinition: ToolDefinition = { `(interrupted, cancelled, incomplete-report, and similar). awaiting_director is not terminal: re-wait while still pending re-delivers the same question. ` + `Answer with send_input (soft). Do not call this in a tight zero-progress loop: a timeout means the targets are still ` + `queued, running, or awaiting a director answer, not "try again right away" — do other work, reply to the operator, or change the brief. Calling again with the ` + - `same targets is a real timed wait, not a spin, but wastes turns if nothing has changed.`, + `same targets is a real timed wait, not a spin, but wastes turns if nothing has changed. ` + + `Repeated identical waits that keep timing out are exempt from the run's doom-loop guard while ` + + `targets stay live — a timeout or still-running result is liveness, not a stall or a crash, so ` + + `keep waiting (or do other work) rather than treating it as a failure.`, inputSchema: { type: "object", properties: { diff --git a/src/subagent/poll-exempt.test.ts b/src/subagent/poll-exempt.test.ts new file mode 100644 index 000000000..11033cbf1 --- /dev/null +++ b/src/subagent/poll-exempt.test.ts @@ -0,0 +1,155 @@ +import { describe, expect, test } from "bun:test"; +import type { ToolCall, ToolResult } from "@intx/types/runtime"; +import { isPollOnlyPendingBatch } from "./poll-exempt.js"; + +function call(name: string, id = "c1"): ToolCall { + return { id, name, arguments: {} }; +} + +function result(content: string | Record): ToolResult { + return { callId: "c1", content }; +} + +function waitContent( + statuses: string[], + timedOut: boolean, +): Record { + return { + results: statuses.map((status, i) => ({ agent_id: `w${i}`, status })), + timed_out: timedOut, + }; +} + +describe("isPollOnlyPendingBatch", () => { + test("timed-out wait_agents batch is exempt", () => { + expect( + isPollOnlyPendingBatch( + [call("wait_agents")], + [result(waitContent(["running"], true))], + ), + ).toBe(true); + }); + + test("live wait statuses without timeout are exempt", () => { + for (const status of ["running", "queued", "awaiting_director"]) { + expect( + isPollOnlyPendingBatch( + [call("wait_agents")], + [result(waitContent([status], false))], + ), + ).toBe(true); + } + }); + + test("one live entry keeps a mixed-status wait exempt", () => { + expect( + isPollOnlyPendingBatch( + [call("wait_agents")], + [result(waitContent(["done", "running"], false))], + ), + ).toBe(true); + }); + + test("terminal wait_agents batch counts normally", () => { + for (const statuses of [ + ["done"], + ["failed"], + ["interrupted"], + ["done", "failed"], + ["unknown"], + [], + ]) { + expect( + isPollOnlyPendingBatch( + [call("wait_agents")], + [result(waitContent(statuses, false))], + ), + ).toBe(false); + } + }); + + test("running shell_collect is exempt; completed or cancelling counts", () => { + const collect = call("shell_collect"); + expect( + isPollOnlyPendingBatch( + [collect], + [result({ shell_id: "s1", status: "running" })], + ), + ).toBe(true); + for (const status of ["completed", "cancelling"]) { + expect( + isPollOnlyPendingBatch([collect], [result({ shell_id: "s1", status })]), + ).toBe(false); + } + }); + + test("unparseable or error poll output counts normally", () => { + expect( + isPollOnlyPendingBatch( + [call("wait_agents")], + [result("Error: timed out waiting")], + ), + ).toBe(false); + expect( + isPollOnlyPendingBatch( + [call("shell_collect")], + [result("No background shell with id s9.")], + ), + ).toBe(false); + }); + + test("non-poll calls are never exempt", () => { + expect( + isPollOnlyPendingBatch( + [call("spawn_agent")], + [result(waitContent(["running"], true))], + ), + ).toBe(false); + }); + + test("mixed poll and non-poll batches count normally", () => { + expect( + isPollOnlyPendingBatch( + [call("wait_agents"), call("shell_collect")], + [ + result(waitContent(["running"], true)), + result({ shell_id: "s1", status: "running" }), + ], + ), + ).toBe(true); + expect( + isPollOnlyPendingBatch( + [call("wait_agents"), call("read")], + [ + result(waitContent(["running"], true)), + result({ shell_id: "s1", status: "running" }), + ], + ), + ).toBe(false); + }); + + test("a settled poll beside a pending poll counts normally", () => { + expect( + isPollOnlyPendingBatch( + [call("wait_agents", "c1"), call("shell_collect", "c2")], + [ + { callId: "c1", content: waitContent(["done"], false) }, + { callId: "c2", content: { shell_id: "s1", status: "running" } }, + ], + ), + ).toBe(false); + }); + + test("empty batches and misaligned results are never exempt", () => { + expect(isPollOnlyPendingBatch([], [])).toBe(false); + expect( + isPollOnlyPendingBatch( + [call("wait_agents")], + [ + result(waitContent(["running"], true)), + result(waitContent(["running"], true)), + ], + ), + ).toBe(false); + }); +}); diff --git a/src/subagent/poll-exempt.ts b/src/subagent/poll-exempt.ts new file mode 100644 index 000000000..fa940fd03 --- /dev/null +++ b/src/subagent/poll-exempt.ts @@ -0,0 +1,62 @@ +import { type } from "arktype"; +import type { ToolCall, ToolResult } from "@intx/types/runtime"; +import { isLiveWaitStatus, type WaitJSONStatus } from "./lifecycle.js"; + +const WaitAgentsPayload = type({ + "timed_out?": "boolean", + "results?": type({ status: "string" }).array(), +}); + +const ShellCollectPayload = type({ + status: "string", +}); + +function resultPayload(result: ToolResult): unknown { + if (typeof result.content !== "string") return result.content; + try { + return JSON.parse(result.content) as unknown; + } catch { + return undefined; + } +} + +function isWaitAgentsPending(payload: unknown): boolean { + const parsed = WaitAgentsPayload(payload); + if (parsed instanceof type.errors) return false; + if (parsed.timed_out === true) return true; + return (parsed.results ?? []).some((entry) => + isLiveWaitStatus(entry.status as WaitJSONStatus), + ); +} + +function isShellCollectPending(payload: unknown): boolean { + const parsed = ShellCollectPayload(payload); + if (parsed instanceof type.errors) return false; + return parsed.status === "running"; +} + +/** + * Doom-loop liveness policy for poll tools. A batch is exempt only when every + * call is a known poll (`wait_agents`, `shell_collect`) and every result + * still shows pending — a timed-out or live-status wait, a `running` collect. + * Anything else (terminal polls, non-poll calls, mixed batches, unparseable + * output) returns false so the guard counts the batch normally. + */ +export function isPollOnlyPendingBatch( + calls: readonly ToolCall[], + results: readonly ToolResult[], +): boolean { + if (calls.length === 0 || results.length !== calls.length) return false; + return calls.every((call, index) => { + const result = results[index]; + if (result === undefined) return false; + if (call.name !== "wait_agents" && call.name !== "shell_collect") { + return false; + } + const payload = resultPayload(result); + if (payload === undefined) return false; + return call.name === "wait_agents" + ? isWaitAgentsPending(payload) + : isShellCollectPending(payload); + }); +} diff --git a/vendor/intx-inference/PATCHES.md b/vendor/intx-inference/PATCHES.md index 087156c13..3fed7ea14 100644 --- a/vendor/intx-inference/PATCHES.md +++ b/vendor/intx-inference/PATCHES.md @@ -249,6 +249,28 @@ path:** Upstream PR wrapping `tryCorrelate` in try/finally to clear in `reactor.ts`; re-verify exit paths after upstream changes to gate clearing. +## reactor-ts-doom-loop-poll-exemption + +`reactor.ts` — Doom-loop guard exempts still-pending poll batches. The batch +accounting asks an optional first-party liveness predicate +(`PollBatchLivenessPredicate`, typed in `harness.ts` on `Dependencies` and +resolved direct-wins-over-deps through `assembly.ts` into `ReactorConfig`, +mirroring `assembly-ts-deps-context-transforms`): when the batch is +poll-only and every result still shows pending, the stale signature and +repeat count reset instead of counting. Reset — not skip — so an earlier +streak cannot false-positive later; mixed batches and poll-only-terminal +batches count normally. Without the predicate every batch counts, same as +upstream. + +**Disposition:** Promotion candidate. Legitimate `wait_agents` / +`shell_collect` polling repeats the identical batch once per model turn and +trips the upstream guard at the default threshold while targets are still +running. **Removal path:** Upstream PR adding a liveness exemption to the +doom-loop accounting. +**Re-carry:** new at this patch; sits inside the upstream doom-loop +accounting block, re-verify the reset branch after upstream changes to batch +signatures or repeat counting. + ## reactor-ts-checkpoint-after-tool-cycle `reactor.ts` — Checkpoint after a tool cycle that appends to history. diff --git a/vendor/intx-inference/src/assembly.ts b/vendor/intx-inference/src/assembly.ts index f7d5a0ab0..8b4d257e5 100644 --- a/vendor/intx-inference/src/assembly.ts +++ b/vendor/intx-inference/src/assembly.ts @@ -30,7 +30,7 @@ import { type AuthzExtensionOptions, } from "./authz-extension"; import type { CorrelationValidator } from "./correlation"; -import type { Dependencies } from "./harness"; +import type { Dependencies, PollBatchLivenessPredicate } from "./harness"; import { createReactor, type Reactor, @@ -82,6 +82,13 @@ export type ReactorAssemblyConfig = { beforeToolExtensions?: BeforeToolExtension[]; toolResultTransforms?: ToolResultTransform[]; contextTransforms?: ContextTransform[]; + /** + * Liveness policy for the doom-loop guard's batch accounting. A direct + * value wins over the one riding `deps`. + * + * Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-doom-loop-poll-exemption + */ + isPollOnlyPendingBatch?: PollBatchLivenessPredicate; compactors?: Record; sizeCapMaxChars?: number; @@ -135,6 +142,7 @@ export function createReactorAssembly( beforeToolExtensions: callerBeforeToolExtensions, toolResultTransforms: callerToolResultTransforms, contextTransforms, + isPollOnlyPendingBatch, compactors, sizeCapMaxChars, afterCheckpoint: callerAfterCheckpoint, @@ -242,6 +250,15 @@ export function createReactorAssembly( // Locally patched — see vendor/intx-inference/PATCHES.md#assembly-ts-deps-context-transforms const resolvedContextTransforms = contextTransforms ?? deps.contextTransforms; + // The liveness policy resolves the same way: directly on the assembly + // config, or riding `deps` (the only channel the published `@intx/agent` + // forwards verbatim). A direct value wins so callers composing their own + // assembly are unaffected by whatever a shared deps object carries. + // + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-doom-loop-poll-exemption + const resolvedIsPollOnlyPendingBatch = + isPollOnlyPendingBatch ?? deps.isPollOnlyPendingBatch; + // exactOptionalPropertyTypes is on: only set optional keys when defined. const reactorConfig: ReactorConfig = { sessionId, @@ -260,6 +277,9 @@ export function createReactorAssembly( ...(resolvedContextTransforms !== undefined ? { contextTransforms: resolvedContextTransforms } : {}), + ...(resolvedIsPollOnlyPendingBatch !== undefined + ? { isPollOnlyPendingBatch: resolvedIsPollOnlyPendingBatch } + : {}), ...(compactors !== undefined ? { compactors } : {}), ...(composedAfterCheckpoint !== undefined ? { afterCheckpoint: composedAfterCheckpoint } diff --git a/vendor/intx-inference/src/harness.ts b/vendor/intx-inference/src/harness.ts index d3ff1985f..7d4c71495 100644 --- a/vendor/intx-inference/src/harness.ts +++ b/vendor/intx-inference/src/harness.ts @@ -31,6 +31,8 @@ import type { RetryDecision, SafetyRatingBlock, TokenUsage, + ToolCall, + ToolResult, AssistantTurn, ContentBlock, } from "@intx/types/runtime"; @@ -76,6 +78,19 @@ export const DEFAULT_TOTAL_TIMEOUT_MS = 600_000; export const HarnessId: unique symbol = Symbol("HarnessId"); +/** + * Liveness policy for the doom-loop guard's batch accounting. Receives the + * executed calls of one tool turn with their results aligned by index and + * returns true when the batch is legitimate liveness rather than a runaway + * loop. First-party runtimes recognize still-pending polls (`wait_agents` + * timeouts and live wait statuses, `running` shell collects); terminal polls, + * non-poll calls, and mixed batches return false so real loops still trip. + */ +export type PollBatchLivenessPredicate = ( + calls: readonly ToolCall[], + results: readonly ToolResult[], +) => boolean; + /** * Runtime dependencies injected into `runInference`. Code-only — not part of * any persisted schema. Test harnesses substitute `fetch` (and stamp the @@ -126,6 +141,16 @@ export type Dependencies = { * Locally patched — see vendor/intx-inference/PATCHES.md#harness-ts-context-transforms */ readonly contextTransforms?: ContextTransform[]; + /** + * Liveness policy for the doom-loop guard's batch accounting. When the + * batch about to be counted is a legitimate liveness signal (first-party + * runtimes recognize still-pending polls), the stale streak resets instead + * of counting. Optional — a custom `Dependencies` object without this + * field counts every batch, same as before. + * + * Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-doom-loop-poll-exemption + */ + readonly isPollOnlyPendingBatch?: PollBatchLivenessPredicate; readonly [HarnessId]?: symbol; }; diff --git a/vendor/intx-inference/src/index.ts b/vendor/intx-inference/src/index.ts index c139d39ff..2fba587bd 100644 --- a/vendor/intx-inference/src/index.ts +++ b/vendor/intx-inference/src/index.ts @@ -10,6 +10,7 @@ export { export type { Dependencies, InferenceHarnessOptions, + PollBatchLivenessPredicate, Scheduler, } from "./harness"; export type { diff --git a/vendor/intx-inference/src/reactor.test.ts b/vendor/intx-inference/src/reactor.test.ts index b8a56a0b4..23cdf194b 100644 --- a/vendor/intx-inference/src/reactor.test.ts +++ b/vendor/intx-inference/src/reactor.test.ts @@ -34,7 +34,11 @@ import type { } from "@intx/types/runtime"; import type { ReactorConfig, Reactor, ReactorEmittedEvent } from "./reactor"; -import type { Dependencies, InferenceHarnessOptions } from "./harness"; +import type { + Dependencies, + InferenceHarnessOptions, + PollBatchLivenessPredicate, +} from "./harness"; import type { CorrelationValidator } from "./correlation"; import type { AfterInferenceHook } from "./default-director"; @@ -1791,9 +1795,200 @@ describe("createReactor — doom-loop detection", () => { }); }); -// --------------------------------------------------------------------------- -// 8. Correlation matching -// --------------------------------------------------------------------------- +describe("createReactor — doom-loop poll exemption", () => { + // Production-faithful stand-in for the first-party liveness predicate (the + // predicate truth table itself is unit-tested beside the real + // implementation): exempt only when every call is a known poll and every + // result still shows pending. These tests lock the guard's reset-vs-count + // behavior around that verdict. + const pendingPollLiveness: PollBatchLivenessPredicate = (calls, results) => + calls.length > 0 && + calls.every((call, index) => { + const content = results[index]?.content; + if (typeof content !== "string") return false; + let payload: unknown; + try { + payload = JSON.parse(content) as unknown; + } catch { + return false; + } + if (typeof payload !== "object" || payload === null) return false; + if (call.name === "wait_agents") { + const { timed_out: timedOut, results: entries } = payload as { + timed_out?: unknown; + results?: { status?: unknown }[]; + }; + if (timedOut === true) return true; + return ( + Array.isArray(entries) && + entries.some( + (entry) => + entry.status === "running" || + entry.status === "queued" || + entry.status === "awaiting_director", + ) + ); + } + if (call.name === "shell_collect") { + return (payload as { status?: unknown }).status === "running"; + } + return false; + }); + + function depsWithLiveness(): Dependencies { + return { + ...createDefaultDependencies(), + isPollOnlyPendingBatch: pendingPollLiveness, + }; + } + + function pendingWaitResult(callId: string): { + callId: string; + content: string; + } { + return { + callId, + content: JSON.stringify({ + results: [{ agent_id: "w1", status: "running" }], + timed_out: true, + }), + }; + } + + function settledWaitResult(callId: string): { + callId: string; + content: string; + } { + return { + callId, + content: JSON.stringify({ + results: [{ agent_id: "w1", status: "done" }], + timed_out: false, + }), + }; + } + + test("does not trip on repeated still-pending poll batches", async () => { + const { reactor, events, waitFor } = createTestReactor({ + deps: depsWithLiveness(), + director: createBatchLoopDirector((turn) => + turn < 8 + ? [{ id: `c${turn}`, name: "wait_agents", arguments: { q: 1 } }] + : null, + ), + toolRunner: makeToolRunner(async (call) => pendingWaitResult(call.id)), + }); + + reactor.start(); + reactor.deliver(makeInboundMessage()); + await waitFor("reactor.done"); + + expect(events.some((e) => e.type === "reactor.error")).toBe(false); + expect(getEvent(events, "message.run.ended").data.status).toBe("completed"); + expect(events.filter((e) => e.type === "tool.start").length).toBe(8); + }); + + test("still trips on repeated non-poll batches when a policy is set", async () => { + const { reactor, events, waitFor } = createTestReactor({ + deps: depsWithLiveness(), + director: createBatchLoopDirector((turn) => + turn < 8 + ? [{ id: `c${turn}`, name: "spin", arguments: { q: 1 } }] + : null, + ), + }); + + reactor.start(); + reactor.deliver(makeInboundMessage()); + await waitFor("reactor.done"); + + expect(getEvent(events, "reactor.error").data.fatal).toBe(true); + expect(getEvent(events, "message.run.ended").data.error?.kind).toBe( + "doom_loop", + ); + }); + + test("mixed poll and non-poll batches count normally", async () => { + const { reactor, events, waitFor } = createTestReactor({ + deps: depsWithLiveness(), + director: createBatchLoopDirector((turn) => + turn < 8 + ? [ + { + id: `c${turn}-wait`, + name: "wait_agents", + arguments: { q: 1 }, + }, + { id: `c${turn}-spin`, name: "spin", arguments: {} }, + ] + : null, + ), + toolRunner: makeToolRunner(async (call) => + call.name === "wait_agents" + ? pendingWaitResult(call.id) + : { callId: call.id, content: "spun" }, + ), + }); + + reactor.start(); + reactor.deliver(makeInboundMessage()); + await waitFor("reactor.done"); + + expect(getEvent(events, "reactor.error").data.fatal).toBe(true); + expect(getEvent(events, "message.run.ended").data.error?.kind).toBe( + "doom_loop", + ); + }); + + test("a settled poll batch counts normally", async () => { + const { reactor, events, waitFor } = createTestReactor({ + deps: depsWithLiveness(), + director: createBatchLoopDirector((turn) => + turn < 8 + ? [{ id: `c${turn}`, name: "wait_agents", arguments: { q: 1 } }] + : null, + ), + toolRunner: makeToolRunner(async (call) => settledWaitResult(call.id)), + }); + + reactor.start(); + reactor.deliver(makeInboundMessage()); + await waitFor("reactor.done"); + + expect(getEvent(events, "reactor.error").data.fatal).toBe(true); + expect(getEvent(events, "message.run.ended").data.error?.kind).toBe( + "doom_loop", + ); + }); + + test("a pending poll batch clears a stale non-poll streak", async () => { + // spin x2 leaves a count of 2; a skip-only exemption would preserve it + // and the final spin would trip at 3. The reset clears it, so the run + // completes. + const batches: ToolCall[][] = [ + [{ id: "1", name: "spin", arguments: {} }], + [{ id: "2", name: "spin", arguments: {} }], + [{ id: "3", name: "wait_agents", arguments: { q: 1 } }], + [{ id: "4", name: "spin", arguments: {} }], + ]; + const { reactor, events, waitFor } = createTestReactor({ + deps: depsWithLiveness(), + director: createBatchLoopDirector((turn) => batches[turn] ?? null), + toolRunner: makeToolRunner(async (call) => + call.name === "wait_agents" + ? pendingWaitResult(call.id) + : { callId: call.id, content: "spun" }, + ), + }); + + reactor.start(); + reactor.deliver(makeInboundMessage()); + await waitFor("reactor.done"); + + expect(events.some((e) => e.type === "reactor.error")).toBe(false); + expect(getEvent(events, "message.run.ended").data.status).toBe("completed"); + }); +}); describe("createReactor — correlation", () => { test("message with matching correlationId triggers message.correlated", async () => { diff --git a/vendor/intx-inference/src/reactor.ts b/vendor/intx-inference/src/reactor.ts index b9cd1e332..61f3422a0 100644 --- a/vendor/intx-inference/src/reactor.ts +++ b/vendor/intx-inference/src/reactor.ts @@ -45,7 +45,11 @@ import { ApprovalDecision, signalKindToGateType } from "@intx/types"; import { canonicalJsonStringify } from "@intx/types/wire-definition-hash"; import { type } from "arktype"; import { runInference } from "./harness"; -import type { Dependencies, InferenceHarnessOptions } from "./harness"; +import type { + Dependencies, + InferenceHarnessOptions, + PollBatchLivenessPredicate, +} from "./harness"; import { createCapabilities } from "./director"; import { createGateManager } from "./gates"; import { createCorrelationRegistry } from "./correlation"; @@ -136,6 +140,14 @@ export type ReactorConfig = { beforeToolExtensions?: BeforeToolExtension[]; toolResultTransforms?: ToolResultTransform[]; contextTransforms?: ContextTransform[]; + /** + * Liveness policy for the doom-loop guard's batch accounting. A direct + * value wins over the one riding `deps`; when neither is set every batch + * counts, same as before. + * + * Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-doom-loop-poll-exemption + */ + isPollOnlyPendingBatch?: PollBatchLivenessPredicate; compactors?: Record; afterCheckpoint?: () => Promise; onShutdown?: () => Promise; @@ -242,6 +254,13 @@ export function createReactor(config: ReactorConfig): Reactor { // downstream comparison reads this binding, never the raw config value. const doomLoopThreshold = resolveDoomLoopThreshold(config.doomLoopThreshold); + // Liveness policy for the doom-loop guard's batch accounting, resolved + // direct-wins-over-deps at the construction edge: a value composed straight + // into the reactor config wins over one riding a shared `deps` object, and + // an absent policy counts every batch, same as before. + const isPollOnlyPendingBatch = + config.isPollOnlyPendingBatch ?? deps.isPollOnlyPendingBatch; + // Monotonic sequence counter, scoped to this session. let seq = 0; function nextSeq(): number { @@ -958,7 +977,18 @@ export function createReactor(config: ReactorConfig): Reactor { // when it reaches the threshold. A `null` threshold means detection is // disabled, so the accounting is skipped entirely. const ranCalls = calls.filter((_call, i) => outcomes[i] !== SUSPENDED); - if (doomLoopThreshold !== null && ranCalls.length > 0) { + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-doom-loop-poll-exemption + // A still-pending poll batch is liveness, not a loop: reset the streak + // (a skip would preserve a stale count and false-positive later) while + // mixed and terminal batches count normally. + const isLivePollBatch = + doomLoopThreshold !== null && + ranCalls.length > 0 && + isPollOnlyPendingBatch?.(ranCalls, results) === true; + if (isLivePollBatch) { + lastToolBatchSignature = null; + toolBatchRepeatCount = 0; + } else if (doomLoopThreshold !== null && ranCalls.length > 0) { const signature = toolBatchSignature(ranCalls); if (signature === lastToolBatchSignature) { toolBatchRepeatCount += 1;