From c9516199c1681c77cccca7e766502563e40db0e9 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 10 Sep 2026 18:16:07 -0700 Subject: [PATCH] Exempt pending poll batches from doom-loop guard Legitimate waits repeat identical polls while work is still in flight. Reset the streak only when every polled target is still pending. Mixed and settled batches still count toward the guard. --- src/agent/background-shell-tool.test.ts | 9 + src/agent/background-shell-tool.ts | 5 +- src/provider/inference-dependencies.ts | 2 + src/subagent/agent-fleet.test.ts | 9 + src/subagent/agent-fleet.ts | 5 +- src/subagent/poll-exempt.test.ts | 155 +++++++++++++++++ src/subagent/poll-exempt.ts | 62 +++++++ vendor/intx-inference/PATCHES.md | 22 +++ vendor/intx-inference/src/assembly.ts | 22 ++- vendor/intx-inference/src/harness.ts | 25 +++ vendor/intx-inference/src/index.ts | 1 + vendor/intx-inference/src/reactor.test.ts | 203 +++++++++++++++++++++- vendor/intx-inference/src/reactor.ts | 34 +++- 13 files changed, 545 insertions(+), 9 deletions(-) create mode 100644 src/subagent/poll-exempt.test.ts create mode 100644 src/subagent/poll-exempt.ts 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;