diff --git a/VENDORED.md b/VENDORED.md index 71ff4536e..fd0e30abe 100644 --- a/VENDORED.md +++ b/VENDORED.md @@ -95,7 +95,17 @@ are NOT from upstream faremeter/interchange at all — upstream's own commit. They are copied from gtm-workbench's own `packages/workflow-host` workspace fork (see `docs/revendor-inventory.md` for the full provenance note and why no ordinary upstream-publish kill date applies to this -sub-delta). `vendor/intx/inference-catalog` (CL-6280) is +sub-delta). `vendor/intx/workflow` (CL-6326) adds an +`onBodyFailure` policy field to the `onTrigger` primitive: absent (or +`"end"`) preserves terminal-is-final exactly as before, `"continue"` lets a +body run that ends `failed` (never `cancelled`) leave the section +subscribed instead of ending the whole run, so one bad turn does not kill a +long-lived section. The gate is read live off `primitive.onBodyFailure` at +both the steady-state drive loop and the crash-recovery resume plan in +`runtime/run.ts`, mirroring how `awaitSignal.onTimeout` is read live rather +than defaulted at construction. This delta targets the current pin +(`59f5e7b9`) and re-applies against the re-pinned tree once PR #59 lands — +see `docs/revendor-inventory.md`. `vendor/intx/inference-catalog` (CL-6280) is pinned separately at `5d2aa94a`, a later `main` tip than the other twenty rows' `59f5e7b9`, since that commit is where the package's folded provider/model catalog first landed upstream; its own local modification diff --git a/docs/revendor-inventory.md b/docs/revendor-inventory.md index 3bef04700..9b3cbdfbe 100644 --- a/docs/revendor-inventory.md +++ b/docs/revendor-inventory.md @@ -294,6 +294,40 @@ no code-side mitigation to retire. Recommend re-running this check against the actual Linear ticket bodies before concluding they're stale — this pass searched by inferred keyword/CL-number, not by reading each ticket. +### A failed body occurrence permanently ended a long-lived onTrigger section (CL-6326) + +`vendor/intx/workflow/src/runtime/run.ts` — `runOnTrigger`'s steady-state +drive loop and its crash-recovery counterpart, `planOnTriggerResume` + +Both treat a body run that ends `failed` identically to one that ends +`cancelled`: either throws terminal-is-final, ending the whole section run. +For a long-lived, event-driven section (chat-style: one occurrence per +inbound message), that means a single bad turn — one agent step throwing — +permanently ends the entire conversation's run, with no way to keep servicing +later occurrences. + +**Local fix:** a `BodyFailurePolicy` (`"end" | "continue"`) field, +`onTrigger.onBodyFailure`, defaulting to `"end"` (current behavior, +byte-compatible). `"continue"` lets the section re-arm on its input park +after a `failed` occurrence instead of throwing; `cancelled` is untouched and +always still ends the section — cancellation reflects a drain/operator +decision, not a turn-level error. The failed occurrence stays durably +recorded and loudly observable: the existing `ChildCompleted{terminalStatus: +"failed"}` commit lands on the run's own audit log before the policy check +runs either way, so a `"continue"`-policy failure is never silent, only +non-fatal to the section. Covered by +`vendor/intx/workflow/src/runtime/run.test.ts` — the intended home +`runlocal/run-local.ts:8` already names for this layer's coverage. + +**This delta targets the current pin** (`59f5e7b9`). Upstream's `486a6b1b` +changed `run.ts` after this pin; the delta above re-diffs against the +re-pinned tree once PR #59 lands. + +**Retire when** upstream ships a non-fatal body-failure edge on `onTrigger` +(or the whole source-format workflow line reaches a published `@intx/*` npm +version, per `vendor/intx/workflow`'s broader retirement condition in +`VENDORED.md`). + ## Estimated fix effort | Item | Shape | Rough size | diff --git a/scripts/checks/kill-dates.txt b/scripts/checks/kill-dates.txt index 4092d6ffb..d47883bc0 100644 --- a/scripts/checks/kill-dates.txt +++ b/scripts/checks/kill-dates.txt @@ -32,9 +32,9 @@ vendor/intx/pack-transport | sawyer | 2026-09-14 | 81f230269ae916111bd7242698f4e vendor/intx/storage-isogit | sawyer | 2026-09-14 | 26f30a4fd27645a620bded1e9359fbb8d87d20f759b9e02c52e213efb459ca73 vendor/intx/tool-packaging | sawyer | 2026-09-14 | 47f29256729105eebab38b23b7326ec9fadc9ebcb9ab460eeba44f94a86d7e45 vendor/intx/types | sawyer | 2026-09-14 | 29d8a7b2589979a04a38706e40e14491a4c57856d6368f74e944c386afed2ef7 -vendor/intx/workflow | sawyer | 2026-09-14 | 81af2a4cb3669540b98d31a9a530908fb701a2a19da94ae69dcbe55a58f8f73b +vendor/intx/workflow | sawyer | 2026-09-14 | 81d7ff7b8cfde64ab8b7422f393b5f46c9e98a8af341685f099b01c71ea17197 vendor/intx/workflow-deploy | sawyer | 2026-09-14 | de72de087e7b499d42b69b7eae6185a4c5448c34a02d3c21badf7c6d0ad3b66d -vendor/intx/workflow-host | sawyer | 2026-09-14 | 6bbc9fa69c28c81d34285016c7cf88ed470401b2bff8c57d49a9bfd4b22cf426 +vendor/intx/workflow-host | sawyer | 2026-09-14 | 2cc2fc754bf195ee0eab4d36e7e2ff0d7600f5111dcb5ddc7ae3e4412412d0bb packages/folded-runs | sawyer | 2026-11-01 diff --git a/vendor/intx/workflow/VENDORED-FROM b/vendor/intx/workflow/VENDORED-FROM index c09c99df4..838cb3b84 100644 --- a/vendor/intx/workflow/VENDORED-FROM +++ b/vendor/intx/workflow/VENDORED-FROM @@ -1,4 +1,4 @@ Source: https://github.com/faremeter/interchange (packages/workflow) Commit: 59f5e7b9d94e7bcccfc180e7d9d11434e2e18eec License: LGPL-2.1-only (see vendor/intx/LICENSE) -Local modifications: exports map repointed from the upstream intx-src condition to direct TypeScript source resolution (types/default -> ./src/...); dist references removed. +Local modifications: exports map repointed from the upstream intx-src condition to direct TypeScript source resolution (types/default -> ./src/...); dist references removed. CL-6326: `onTrigger` gains an `onBodyFailure?: "end" | "continue"` policy field (definition/primitives.ts); `runtime/run.ts`'s steady-state drive loop and `planOnTriggerResume` read it live to let a `"continue"`-policy section re-arm past a `failed` body occurrence instead of ending the whole run (`cancelled` is unaffected, always terminal-is-final). See VENDORED.md and docs/revendor-inventory.md. diff --git a/vendor/intx/workflow/src/definition/primitives.ts b/vendor/intx/workflow/src/definition/primitives.ts index d325d0229..948e697bc 100644 --- a/vendor/intx/workflow/src/definition/primitives.ts +++ b/vendor/intx/workflow/src/definition/primitives.ts @@ -30,6 +30,8 @@ import type { Trigger } from "./triggers"; export type DrainBehavior = "cancel" | "wait"; +export type BodyFailurePolicy = "end" | "continue"; + export interface RetryPolicy { /** Maximum number of attempts including the first. */ maxAttempts: number; @@ -217,13 +219,15 @@ export interface LoopPrimitive extends PrimitiveBase { * `trigger.payload`. * * The section never self-completes: the workflow stays running while - * subscribed and terminates only on a body error or an explicit - * end-of-workflow, and a terminated run is final -- never relaunched. The - * first occurrence is the run's own firing trigger (its - * `RunStarted.trigger.payload`); each later occurrence arrives as an input - * signal carrying the next payload. `defineWorkflow` collects every `on` - * into the workflow's `triggers`, so `on` is the first-class binding - * between a trigger and the section it drives. + * subscribed and terminates only on a body run ending `cancelled`, a body + * run ending `failed` under the default `onBodyFailure: "end"` policy, or + * an explicit end-of-workflow -- `onBodyFailure: "continue"` keeps the + * section alive through a failed occurrence -- and a terminated run is + * final -- never relaunched. The first occurrence is the run's own firing + * trigger (its `RunStarted.trigger.payload`); each later occurrence + * arrives as an input signal carrying the next payload. `defineWorkflow` + * collects every `on` into the workflow's `triggers`, so `on` is the + * first-class binding between a trigger and the section it drives. * * `drainBehavior` defaults to `"wait"`: a live interactive section is not * abandoned mid-conversation at redeploy unless the author opts into @@ -234,6 +238,17 @@ export interface OnTriggerPrimitive extends PrimitiveBase { on: Trigger; body: OnTriggerBody; drainBehavior?: DrainBehavior; + /** + * How a body run that ends `failed` affects the section. Absent (or + * `"end"`) preserves terminal-is-final: a failed body run ends the + * whole section run, exactly as before this field existed. `"continue"` + * records the failed occurrence and keeps the section subscribed -- + * the next occurrence spawns and runs normally. A body run that ends + * `cancelled` is unaffected by this field and always ends the section: + * cancellation reflects a drain/operator decision, not a turn-level + * error. + */ + onBodyFailure?: BodyFailurePolicy; } /** @@ -553,6 +568,7 @@ export interface OnTriggerOpts { on: Trigger; body: WorkflowDefinition; drainBehavior?: DrainBehavior; + onBodyFailure?: BodyFailurePolicy; after?: readonly string[]; } @@ -565,6 +581,9 @@ export function onTrigger(opts: OnTriggerOpts): OnTriggerPrimitive { // Authored inline; the deploy step rewrites this to `{ ref }`. body: { inline: opts.body }, drainBehavior, + ...(opts.onBodyFailure !== undefined + ? { onBodyFailure: opts.onBodyFailure } + : {}), ...(opts.after !== undefined ? { after: opts.after } : {}), }; } diff --git a/vendor/intx/workflow/src/runtime/run.test.ts b/vendor/intx/workflow/src/runtime/run.test.ts new file mode 100644 index 000000000..4a5607bce --- /dev/null +++ b/vendor/intx/workflow/src/runtime/run.test.ts @@ -0,0 +1,334 @@ +// `onTrigger`'s non-fatal body-failure edge (`onBodyFailure`). +// +// These are unit tests of the runtime state machine, not the sidecar +// wiring: `runtimeRun` is exercised directly against a hand-built +// `WorkflowRuntimeEnv` (the same in-memory pieces `runLocal` wires, +// plus a fake `spawnSuspendableChild` this file controls per +// `childRunId`) rather than through `runLocal`, which does not expose a +// `spawnSuspendableChild` override. No real agent or substrate is +// involved -- this is the "runtime/run.test.ts" the discipline comment +// in `runlocal/run-local.ts` names as the intended home for this +// coverage. + +import { describe, test, expect } from "bun:test"; + +import { createDefaultDirectorRegistry } from "@intx/agent"; + +import type { OnTriggerPrimitive } from "../definition/primitives"; +import type { WorkflowDefinition } from "../definition/workflow"; +import { createInMemoryBlobSubstrate } from "../runlocal/blob-substrate"; +import { createInMemoryRepoStore } from "../runlocal/repo-store"; +import { createInMemoryScheduler } from "../runlocal/scheduler"; +import { createInMemorySignalChannel } from "../runlocal/signal-channel"; +import { createNoopDrainController } from "./drain"; +import { runtimeRun } from "./run"; +import type { + SpawnSuspendableChild, + SuspendableChildHandle, + WorkflowRuntimeEnv, +} from "./env"; +import { + controlParkKindOf, + resumeFromLog, + type RunState, + type WorkflowEvent, +} from "../state-machine/index"; + +const SECTION_ID = "section"; + +function definitionWith( + onTriggerOverrides: Partial = {}, +): WorkflowDefinition { + const primitive: OnTriggerPrimitive = { + kind: "onTrigger", + id: SECTION_ID, + on: { type: "manual" }, + body: { ref: "test-body" }, + ...onTriggerOverrides, + }; + return { + id: "wf-onbodyfailure", + triggers: [{ type: "manual" }], + steps: { [SECTION_ID]: primitive }, + stepOrder: [SECTION_ID], + }; +} + +type TerminalStatus = "completed" | "failed" | "cancelled"; + +/** + * A fake `spawnSuspendableChild` keyed by `childRunId`, each occurrence + * settling immediately on the terminal status the test scripted for it. + * `resume`/`deliverSignal` are unused by every scenario here (no body + * ever parks) so they throw if called, matching the pattern + * `apps/sidecar/test/workflow-substrate-factory-suspendable-child.test.ts` + * uses for handle members a scenario does not exercise. + */ +function fakeSpawn( + responses: Record, +): { spawn: SpawnSuspendableChild; spawnedChildRunIds: string[] } { + const spawnedChildRunIds: string[] = []; + const spawn: SpawnSuspendableChild = async ({ childRunId }) => { + spawnedChildRunIds.push(childRunId); + const terminalStatus = responses[childRunId]; + if (terminalStatus === undefined) { + throw new Error(`fakeSpawn: no scripted response for ${childRunId}`); + } + let delivered = false; + const handle: SuspendableChildHandle = { + async next() { + if (delivered) { + throw new Error( + `fakeSpawn: ${childRunId} next() called more than once`, + ); + } + delivered = true; + return { kind: "terminal", terminalStatus }; + }, + async resume() { + throw new Error(`fakeSpawn: ${childRunId} unexpected resume()`); + }, + async deliverSignal() { + throw new Error(`fakeSpawn: ${childRunId} unexpected deliverSignal()`); + }, + }; + return handle; + }; + return { spawn, spawnedChildRunIds }; +} + +function buildEnv(spawn: SpawnSuspendableChild): WorkflowRuntimeEnv { + const repoStore = createInMemoryRepoStore(); + const clock = () => new Date(); + let idCounter = 0; + const newId = (prefix: string): string => { + idCounter += 1; + return `${prefix}-${String(idCounter)}`; + }; + const definitionForDrain = definitionWith(); + return { + repoStore, + scheduler: createInMemoryScheduler({ repoStore, clock }), + signalChannel: createInMemorySignalChannel({ newId: () => newId("sig") }), + blobs: createInMemoryBlobSubstrate(), + directors: createDefaultDirectorRegistry(), + authorize: async () => ({ + effect: "allow", + matchingGrants: [], + resolvedBy: null, + }), + invokeStep: async () => { + throw new Error("no step primitive is exercised by these tests"); + }, + spawnChild: async () => { + throw new Error("no childWorkflow primitive is exercised by these tests"); + }, + spawnSuspendableChild: spawn, + clock, + newId, + drain: createNoopDrainController(definitionForDrain), + }; +} + +async function readState( + env: WorkflowRuntimeEnv, + runId: string, +): Promise { + const events = await env.repoStore.read(runId); + return resumeFromLog(runId, events); +} + +/** Poll the durable log until `predicate` holds. In-memory, so this settles fast. */ +async function waitFor( + env: WorkflowRuntimeEnv, + runId: string, + predicate: (state: RunState) => boolean, +): Promise { + for (let attempt = 0; attempt < 1000; attempt += 1) { + const state = await readState(env, runId); + if (predicate(state)) return state; + await new Promise((resolve) => setTimeout(resolve, 0)); + } + throw new Error("waitFor: predicate never became true"); +} + +function inputParkName(state: RunState): string | undefined { + const container = state.steps.get(SECTION_ID); + if (container === undefined || container.phase !== "awaiting-signal") { + return undefined; + } + if (container.awaitingSignal === undefined) return undefined; + if (controlParkKindOf(container.awaitingSignal) !== "input") return undefined; + return container.awaitingSignal.name; +} + +describe("onTrigger onBodyFailure", () => { + test("default policy: a failed body run ends the whole section run", async () => { + const { spawn, spawnedChildRunIds } = fakeSpawn({ + "section__0": "failed", + }); + const env = buildEnv(spawn); + const definition = definitionWith(); // no onBodyFailure -- default "end" + + const run = runtimeRun(definition, env, { triggerPayload: {} }); + const result = await run.complete; + + expect(result.terminalStatus).toBe("failed"); + // The failed occurrence is still durably recorded before the throw. + const events = await env.repoStore.read(run.runId); + const childCompleted = events.find( + (e): e is WorkflowEvent & { kind: "ChildCompleted" } => + e.kind === "ChildCompleted", + ); + expect(childCompleted?.terminalStatus).toBe("failed"); + // The section never re-arms for a second occurrence under the default. + expect(spawnedChildRunIds).toEqual(["section__0"]); + }); + + test('onBodyFailure: "continue" keeps the section alive through a failed occurrence', async () => { + const { spawn } = fakeSpawn({ + "section__0": "failed", + "section__1": "completed", + }); + const env = buildEnv(spawn); + const definition = definitionWith({ onBodyFailure: "continue" }); + + const run = runtimeRun(definition, env, { triggerPayload: { n: 0 } }); + + // The run does not settle terminal after occurrence 0 fails -- it + // re-arms on the input park instead. + const afterFirstFailure = await waitFor(env, run.runId, (state) => + inputParkName(state) !== undefined, + ); + expect(afterFirstFailure.phase).not.toBe("failed"); + expect(afterFirstFailure.children.get("section__0")?.terminalStatus).toBe( + "failed", + ); + + // The failed occurrence's ChildCompleted is a durable, loud audit event + // on the run's own log -- the section did not silently swallow it. + const eventsAfterFirstFailure = await env.repoStore.read(run.runId); + const childCompleted = eventsAfterFirstFailure.find( + (e): e is WorkflowEvent & { kind: "ChildCompleted" } => + e.kind === "ChildCompleted" && e.childRunId === "section__0", + ); + expect(childCompleted).toBeDefined(); + expect(childCompleted?.terminalStatus).toBe("failed"); + + const parkName = inputParkName(afterFirstFailure); + if (parkName === undefined) throw new Error("expected an input park"); + await run.signal(parkName, { n: 1 }); + + // Occurrence 1 spawns and succeeds normally -- the run proceeds, it + // does not throw. + await waitFor( + env, + run.runId, + (state) => state.children.get("section__1")?.terminalStatus === "completed", + ); + + await run.cancel("self", "test cleanup"); + const result = await run.complete; + expect(result.terminalStatus).toBe("cancelled"); + }); + + test('onBodyFailure: "continue" never swallows a cancelled body run', async () => { + const { spawn } = fakeSpawn({ + "section__0": "cancelled", + }); + const env = buildEnv(spawn); + const definition = definitionWith({ onBodyFailure: "continue" }); + + const run = runtimeRun(definition, env, { triggerPayload: {} }); + const result = await run.complete; + + // A cancelled body run still throws terminal-is-final -- it lands the + // section's own step as StepFailed, so the whole run's terminalStatus + // is "failed" (a thrown primitive error, not a run-level cancel); + // what matters here is that `onBodyFailure` did NOT swallow it into a + // re-arm the way it does for "failed". + expect(result.terminalStatus).toBe("failed"); + const events = await env.repoStore.read(run.runId); + const message = events.find( + (e): e is WorkflowEvent & { kind: "StepFailed" } => e.kind === "StepFailed", + )?.error.message; + expect(message).toContain("cancelled"); + }); + + test("crash-recovery honors onBodyFailure: a failed-but-continuing section resumes on the input re-arm", async () => { + // Hand-built seed log: the container's mid-flight state right after + // occurrence 0's ChildCompleted{failed} commits, but BEFORE the + // re-arm park lands -- the exact crash window `planOnTriggerResume`'s + // ordering comment describes. Built by hand (rather than captured off + // a live run) so the test is deterministic about which side of that + // race it exercises. + const definition = definitionWith({ onBodyFailure: "continue" }); + const seedLog: WorkflowEvent[] = [ + { + kind: "RunStarted", + seq: 1, + at: "2026-01-01T00:00:00.000Z", + runId: "resume-test", + definitionHash: "seed-hash", + trigger: { type: "manual", payload: {} }, + }, + { + kind: "StepStarted", + seq: 2, + at: "2026-01-01T00:00:00.000Z", + stepId: SECTION_ID, + attempt: 1, + input: { ref: "unused-input-ref" }, + }, + { + kind: "ChildSpawned", + seq: 3, + at: "2026-01-01T00:00:00.000Z", + stepId: SECTION_ID, + childRunId: "section__0", + childDefinitionRef: "test-body", + }, + { + kind: "ChildCompleted", + seq: 4, + at: "2026-01-01T00:00:00.000Z", + childRunId: "section__0", + terminalStatus: "failed", + }, + ]; + + // Resume a fresh env from that seed log with the same policy. The + // resume path (`planOnTriggerResume`) must take the reawait-input + // arm, not terminal-is-final, so the section keeps going. + const resumeSpawn = fakeSpawn({ + "section__0": "failed", + "section__1": "completed", + }); + const resumeEnv = buildEnv(resumeSpawn.spawn); + const resumeRun = runtimeRun(definition, resumeEnv, { + runId: "resume-test", + resumeFromEvents: seedLog, + }); + + const afterResume = await waitFor(resumeEnv, resumeRun.runId, (state) => + inputParkName(state) !== undefined, + ); + // Resume did not re-spawn occurrence 0's body -- it recovered position + // from the log rather than throwing terminal-is-final. + expect(resumeSpawn.spawnedChildRunIds).toEqual([]); + const parkName = inputParkName(afterResume); + if (parkName === undefined) throw new Error("expected an input park"); + await resumeRun.signal(parkName, { n: 1 }); + + await waitFor( + resumeEnv, + resumeRun.runId, + (state) => + state.children.get("section__1")?.terminalStatus === "completed", + ); + + await resumeRun.cancel("self", "test cleanup"); + const result = await resumeRun.complete; + expect(result.terminalStatus).toBe("cancelled"); + }); +}); diff --git a/vendor/intx/workflow/src/runtime/run.ts b/vendor/intx/workflow/src/runtime/run.ts index 8a765a553..7a247ea9b 100644 --- a/vendor/intx/workflow/src/runtime/run.ts +++ b/vendor/intx/workflow/src/runtime/run.ts @@ -2080,15 +2080,27 @@ async function runOnTrigger( } await flush(env, runId); - if (terminalStatus !== "completed") { - // Terminal-is-final: a body run that failed or was cancelled ends the - // whole section run. Throwing lands the parent terminal via + if (terminalStatus === "cancelled") { + // Terminal-is-final, unconditionally: a cancelled body run always ends + // the section. Cancellation reflects a drain/operator decision, not a + // turn-level error, so `onBodyFailure` never swallows it. + throw new Error( + `onTrigger ${primitive.id} body run ${childRunId} ended cancelled`, + ); + } + if (terminalStatus === "failed" && primitive.onBodyFailure !== "continue") { + // Terminal-is-final (default): a failed body run ends the whole + // section run. Throwing lands the parent terminal via // `runPrimitiveSafe`; the run does not relaunch. throw new Error( - `onTrigger ${primitive.id} body run ${childRunId} ended ` + - `${terminalStatus}`, + `onTrigger ${primitive.id} body run ${childRunId} ended failed`, ); } + // terminalStatus is "completed", or "failed" with onBodyFailure: + // "continue" -- the failed occurrence is already recorded (the + // ChildCompleted commit above this block carries + // `terminalStatus: "failed"`, the run's durable audit event for it); fall + // through to the same re-arm every completed occurrence takes. // Re-arm: park on a fresh input channel for the next event. The park is // snapshot-less (`kind: "input"`); the run's owner delivers the next @@ -2395,20 +2407,29 @@ function planOnTriggerResume( // is already owned -- a body that then completed is caught HERE (reawait- // input), not by the in-flight throw. Inverting the order would wrongly fail a // post-abandon-completed body. + if (child.terminalStatus === "cancelled") { + return { + kind: "terminal-is-final", + eventIndex, + terminalStatus: "cancelled", + }; + } if ( - child.terminalStatus === "failed" || - child.terminalStatus === "cancelled" + child.terminalStatus === "failed" && + primitive.onBodyFailure !== "continue" ) { return { kind: "terminal-is-final", eventIndex, - terminalStatus: child.terminalStatus, + terminalStatus: "failed", }; } const container = state.steps.get(primitive.id); - if (child.terminalStatus === "completed") { - // The event's body finished; the section is idle on its input re-arm. Re- - // adopt the durable input park if it was committed, else re-arm fresh. + if (child.terminalStatus === "completed" || child.terminalStatus === "failed") { + // The event's body finished -- completed, or failed with + // `onBodyFailure: "continue"`, both of which are "the event is over" -- + // the section is idle on its input re-arm. Re-adopt the durable input + // park if it was committed, else re-arm fresh. if ( container !== undefined && container.phase === "awaiting-signal" &&