diff --git a/apps/sidecar/src/workflow-substrate-factory/child-runtime.ts b/apps/sidecar/src/workflow-substrate-factory/child-runtime.ts index 556ddeaea..02d1d1f11 100644 --- a/apps/sidecar/src/workflow-substrate-factory/child-runtime.ts +++ b/apps/sidecar/src/workflow-substrate-factory/child-runtime.ts @@ -43,12 +43,27 @@ import { type SourcesSnapshotRef, } from "@intx/workflow-host"; import type { MailPartReader } from "@intx/types/runtime"; +import { reportError } from "@corbits/error-sink"; import { parseStepInferenceSources, type StepInferenceSourceTable, } from "./config"; +function fireAndForget( + work: Promise, + operation: string, + childRunId: string, + report: typeof reportError, +): void { + void work.catch((error: unknown) => { + report(error, { + operation, + extra: { childRunId }, + }); + }); +} + /** * Real per-step invoker for an onTrigger BODY child, distinct from the * `StepInvoker` stub the shared `SidecarRunChildDeps.invokeStep` carries for @@ -154,6 +169,27 @@ export interface SidecarRunChildDeps { * a monotonic counter combined with a random suffix. */ newId?: (prefix: string) => string; + /** + * Test seam only — no production caller ever sets this. Defaults to + * `@intx/workflow`'s `runtimeRun`. Exists because bun's `mock.module` + * replaces a module process-wide and cannot be undone for later files + * in the same `bun test` process. + */ + runtimeRun?: typeof runtimeRun; + /** + * Test seam only — no production caller ever sets this. Defaults to + * `@corbits/error-sink`'s `reportError`. Paired with `runtimeRun` so a + * cancel-rejection test can observe reports without a process-wide + * module swap. + */ + reportError?: typeof reportError; + /** + * Test seam only — no production caller ever sets this. Defaults to + * `@intx/workflow-host`'s `createWorkflowHostSignalChannel`. Paired with + * `reportError` so a stop-rejection test can observe reports without a + * process-wide module swap. + */ + createSignalChannel?: typeof createWorkflowHostSignalChannel; } /** @@ -184,6 +220,8 @@ export function createSidecarRunChild( const directors = deps.directors ?? createDefaultDirectorRegistry(); const clock = deps.clock ?? defaultClock; const newId = deps.newId ?? defaultNewId; + const startRun = deps.runtimeRun ?? runtimeRun; + const report = deps.reportError ?? reportError; // The in-process child runs under real supervisor authority; its // control-plane cancel (`CancelRequested`) must be signed by a supervisor // principal, which the kind handler requires and the substrate authorizes @@ -234,7 +272,7 @@ export function createSidecarRunChild( childRunId, }); try { - const handle = runtimeRun(rewrittenDefinition, env, { + const handle = startRun(rewrittenDefinition, env, { runId: childRunId, triggerPayload: input, }); @@ -245,7 +283,12 @@ export function createSidecarRunChild( // workflow-process-signed cancel would be refused and a parent abort // would surface as a failed rather than cancelled child. const cancelOnAbort = (): void => { - void handle.cancel("supervisor-operator", "parent cancelled"); + fireAndForget( + handle.cancel("supervisor-operator", "parent cancelled"), + "sidecar.child-runtime.cancel", + childRunId, + report, + ); }; if (signal.aborted) { cancelOnAbort(); @@ -303,6 +346,8 @@ export function createSidecarSpawnSuspendableChild( const directors = deps.directors ?? createDefaultDirectorRegistry(); const clock = deps.clock ?? defaultClock; const newId = deps.newId ?? defaultNewId; + const startRun = deps.runtimeRun ?? runtimeRun; + const report = deps.reportError ?? reportError; // The in-process body child runs under real supervisor authority; its // control-plane cancel (`CancelRequested`) must be signed by a supervisor // principal, which the kind handler requires and the substrate authorizes @@ -455,7 +500,7 @@ export function createSidecarSpawnSuspendableChild( // silently (a re-park does not re-fire onPark), and the caller relays // the grant via resume on the correlation it recovered from its own // log. On a fresh spawn, seed the run with the event's trigger payload. - const handle = runtimeRun( + const handle = startRun( rewrittenDefinition, env, resumeFromEvents !== undefined @@ -470,7 +515,12 @@ export function createSidecarSpawnSuspendableChild( // cancel would be refused and a parent abort would surface as a failed // rather than cancelled child. const cancelOnAbort = (): void => { - void handle.cancel("supervisor-operator", "parent cancelled"); + fireAndForget( + handle.cancel("supervisor-operator", "parent cancelled"), + "sidecar.child-runtime.cancel", + childRunId, + report, + ); }; if (signal.aborted) { cancelOnAbort(); @@ -491,7 +541,12 @@ export function createSidecarSpawnSuspendableChild( }) .finally(() => { signal.removeEventListener("abort", cancelOnAbort); - void signalChannel.stop(); + fireAndForget( + signalChannel.stop(), + "sidecar.child-runtime.signal-channel-stop", + childRunId, + report, + ); notify(); }); @@ -505,9 +560,14 @@ export function createSidecarSpawnSuspendableChild( // Cancel the child so its terminal (and the channel teardown // tied to it) fires, then surface the error: the throw lands // the section run's terminal via `runOnTrigger`. - void handle.cancel( - "supervisor-operator", - "onTrigger body re-armed an unsupported input park", + fireAndForget( + handle.cancel( + "supervisor-operator", + "onTrigger body re-armed an unsupported input park", + ), + "sidecar.child-runtime.cancel", + childRunId, + report, ); throw event.error; } @@ -582,7 +642,9 @@ function buildChildRunEnv(args: { runId: childRunId, ref: deps.workflowRunRef, }); - const signalChannel = createWorkflowHostSignalChannel({ + const startChannel = + deps.createSignalChannel ?? createWorkflowHostSignalChannel; + const signalChannel = startChannel({ repoStore: deps.substrate, principal: deps.principal, repoId: deps.workflowRunRepoId, diff --git a/apps/sidecar/test/workflow-substrate-factory-child-cancel-report.test.ts b/apps/sidecar/test/workflow-substrate-factory-child-cancel-report.test.ts new file mode 100644 index 000000000..b367ff7b6 --- /dev/null +++ b/apps/sidecar/test/workflow-substrate-factory-child-cancel-report.test.ts @@ -0,0 +1,331 @@ +// Force handle.cancel to reject during parent abort and prove the +// fire-and-forget call reports through reportError instead of becoming +// an unhandled rejection. Injected via SidecarRunChildDeps test seams +// rather than bun's `mock.module`, whose process-wide registry swap +// cannot be undone for later files in the same `bun test` process. + +import { afterAll, beforeAll, expect, mock, test } from "bun:test"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { defineAgent } from "@intx/agent"; +import { generateKeyPair } from "@intx/crypto"; +import { + createRepoStore, + workflowRunKindHandler, + WORKFLOW_RUN_GITIGNORE_PATH, +} from "@intx/hub-sessions"; +import type { AuthorizeFn } from "@intx/hub-sessions"; +import type { + RepoId, + WorkflowRunWorkflowProcessPrincipal, +} from "@intx/hub-sessions/substrate"; +import type { KeyPair } from "@intx/types/runtime"; +import { + createInMemoryRepoStore, + createInMemoryScheduler, + defineWorkflow, + step, +} from "@intx/workflow"; +import { + createWorkflowHostSignalChannel, + type RunChildWorkflow, +} from "@intx/workflow-host"; + +import { + createSidecarRunChild, + createSidecarSpawnSuspendableChild, +} from "../src/workflow-substrate-factory/child-runtime"; + +const REF = "refs/heads/main"; +const DEPLOYMENT_ID = "deployment-cancel-report"; +const WORKFLOW_RUN_REPO_ID: RepoId = { + kind: "workflow-run", + id: DEPLOYMENT_ID, +}; +const allowAll: AuthorizeFn = () => ({ allowed: true }); +const PRINCIPAL: WorkflowRunWorkflowProcessPrincipal = { + kind: "workflow-process", + anchorRunId: DEPLOYMENT_ID, +}; + +const tempDirs: string[] = []; +let signingKey: KeyPair; + +beforeAll(async () => { + signingKey = await generateKeyPair(); +}); + +afterAll(async () => { + for (const d of tempDirs.splice(0)) { + await fs.promises.rm(d, { recursive: true, force: true }).catch(() => { + /* best effort */ + }); + } +}); + +test("a cancel rejection during abort is reported and is not unhandled", async () => { + const forced = new Error("forced cancel reject"); + const cancelMock = mock(() => Promise.reject(forced)); + const reportErrorMock = mock( + (_error: unknown, _context: unknown) => "ref-test", + ); + + const dataDir = await fs.promises.mkdtemp( + path.join(os.tmpdir(), "run-child-cancel-report-"), + ); + tempDirs.push(dataDir); + const substrate = createRepoStore({ + dataDir, + signingKey, + handlers: { "workflow-run": workflowRunKindHandler }, + authorize: allowAll, + }); + await substrate.writeTree({ kind: "hub" }, WORKFLOW_RUN_REPO_ID, REF, { + files: { [WORKFLOW_RUN_GITIGNORE_PATH]: "" }, + message: "genesis", + }); + + const runChild: RunChildWorkflow = createSidecarRunChild({ + substrate, + workflowRunRepoId: WORKFLOW_RUN_REPO_ID, + workflowRunRef: REF, + principal: PRINCIPAL, + scheduler: createInMemoryScheduler({ + repoStore: createInMemoryRepoStore(), + clock: () => new Date(), + }), + invokeStep: async () => ({ output: { done: true } }), + runtimeRun: () => ({ + runId: "run-child-forced-cancel", + complete: new Promise(() => undefined), + cancel: cancelMock, + signal: async () => undefined, + }), + reportError: reportErrorMock, + }); + + const agent = defineAgent({ + id: "child-step", + systemPrompt: "s", + tools: [], + capabilities: [], + inference: { sources: [{ provider: "anthropic", model: "m" }] }, + }); + const definition = defineWorkflow({ + id: "child-wf-cancel-report", + trigger: { type: "manual" }, + steps: { s: step({ agent }) }, + }); + + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => { + unhandled.push(reason); + }; + process.on("unhandledRejection", onUnhandled); + try { + const abort = new AbortController(); + abort.abort(); + void runChild({ + definition, + definitionRef: REF, + childRunId: "run-child-cancel-report", + input: { text: "event" }, + parentRunId: "run-parent", + parentStepId: "section", + signal: abort.signal, + }); + await Bun.sleep(50); + expect(cancelMock).toHaveBeenCalled(); + expect(reportErrorMock).toHaveBeenCalledTimes(1); + expect(reportErrorMock.mock.calls[0]?.[0]).toBe(forced); + expect(reportErrorMock.mock.calls[0]?.[1]).toMatchObject({ + operation: "sidecar.child-runtime.cancel", + }); + expect(unhandled).toEqual([]); + } finally { + process.off("unhandledRejection", onUnhandled); + } +}); + +test("a signalChannel.stop rejection is reported and is not unhandled", async () => { + const forced = new Error("forced stop reject"); + const stopMock = mock(() => Promise.reject(forced)); + const reportErrorMock = mock( + (_error: unknown, _context: unknown) => "ref-test", + ); + + const dataDir = await fs.promises.mkdtemp( + path.join(os.tmpdir(), "run-child-stop-report-"), + ); + tempDirs.push(dataDir); + const substrate = createRepoStore({ + dataDir, + signingKey, + handlers: { "workflow-run": workflowRunKindHandler }, + authorize: allowAll, + }); + await substrate.writeTree({ kind: "hub" }, WORKFLOW_RUN_REPO_ID, REF, { + files: { [WORKFLOW_RUN_GITIGNORE_PATH]: "" }, + message: "genesis", + }); + + const spawn = createSidecarSpawnSuspendableChild({ + substrate, + workflowRunRepoId: WORKFLOW_RUN_REPO_ID, + workflowRunRef: REF, + principal: PRINCIPAL, + scheduler: createInMemoryScheduler({ + repoStore: createInMemoryRepoStore(), + clock: () => new Date(), + }), + invokeStep: async () => ({ output: { done: true } }), + runtimeRun: () => ({ + runId: "run-child-forced-stop", + complete: Promise.resolve({ + runId: "run-child-forced-stop", + terminalStatus: "completed" as const, + outputs: {}, + events: [], + }), + cancel: async () => undefined, + signal: async () => undefined, + }), + reportError: reportErrorMock, + createSignalChannel: (opts) => { + const channel = createWorkflowHostSignalChannel(opts); + return { ...channel, stop: stopMock }; + }, + }); + + const agent = defineAgent({ + id: "child-step", + systemPrompt: "s", + tools: [], + capabilities: [], + inference: { sources: [{ provider: "anthropic", model: "m" }] }, + }); + const definition = defineWorkflow({ + id: "child-wf-stop-report", + trigger: { type: "manual" }, + steps: { s: step({ agent }) }, + }); + + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => { + unhandled.push(reason); + }; + process.on("unhandledRejection", onUnhandled); + try { + await spawn( + { + definition, + definitionRef: REF, + childRunId: "run-child-stop-report", + input: { text: "event" }, + parentRunId: "run-parent", + parentStepId: "section", + signal: new AbortController().signal, + }, + () => undefined, + ); + await Bun.sleep(50); + expect(stopMock).toHaveBeenCalled(); + expect(reportErrorMock).toHaveBeenCalledTimes(1); + expect(reportErrorMock.mock.calls[0]?.[0]).toBe(forced); + expect(reportErrorMock.mock.calls[0]?.[1]).toMatchObject({ + operation: "sidecar.child-runtime.signal-channel-stop", + }); + expect(unhandled).toEqual([]); + } finally { + process.off("unhandledRejection", onUnhandled); + } +}); + +test("a spawn-suspendable abort cancel rejection is reported and is not unhandled", async () => { + const forced = new Error("forced suspendable cancel reject"); + const cancelMock = mock(() => Promise.reject(forced)); + const reportErrorMock = mock( + (_error: unknown, _context: unknown) => "ref-test", + ); + + const dataDir = await fs.promises.mkdtemp( + path.join(os.tmpdir(), "run-suspendable-cancel-report-"), + ); + tempDirs.push(dataDir); + const substrate = createRepoStore({ + dataDir, + signingKey, + handlers: { "workflow-run": workflowRunKindHandler }, + authorize: allowAll, + }); + await substrate.writeTree({ kind: "hub" }, WORKFLOW_RUN_REPO_ID, REF, { + files: { [WORKFLOW_RUN_GITIGNORE_PATH]: "" }, + message: "genesis", + }); + + const spawn = createSidecarSpawnSuspendableChild({ + substrate, + workflowRunRepoId: WORKFLOW_RUN_REPO_ID, + workflowRunRef: REF, + principal: PRINCIPAL, + scheduler: createInMemoryScheduler({ + repoStore: createInMemoryRepoStore(), + clock: () => new Date(), + }), + invokeStep: async () => ({ output: { done: true } }), + runtimeRun: () => ({ + runId: "run-suspendable-forced-cancel", + complete: new Promise(() => undefined), + cancel: cancelMock, + signal: async () => undefined, + }), + reportError: reportErrorMock, + }); + + const agent = defineAgent({ + id: "child-step", + systemPrompt: "s", + tools: [], + capabilities: [], + inference: { sources: [{ provider: "anthropic", model: "m" }] }, + }); + const definition = defineWorkflow({ + id: "child-wf-suspendable-cancel-report", + trigger: { type: "manual" }, + steps: { s: step({ agent }) }, + }); + + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => { + unhandled.push(reason); + }; + process.on("unhandledRejection", onUnhandled); + try { + const abort = new AbortController(); + abort.abort(); + await spawn( + { + definition, + definitionRef: REF, + childRunId: "run-suspendable-cancel-report", + input: { text: "event" }, + parentRunId: "run-parent", + parentStepId: "section", + signal: abort.signal, + }, + () => undefined, + ); + await Bun.sleep(50); + expect(cancelMock).toHaveBeenCalled(); + expect(reportErrorMock).toHaveBeenCalledTimes(1); + expect(reportErrorMock.mock.calls[0]?.[0]).toBe(forced); + expect(reportErrorMock.mock.calls[0]?.[1]).toMatchObject({ + operation: "sidecar.child-runtime.cancel", + }); + expect(unhandled).toEqual([]); + } finally { + process.off("unhandledRejection", onUnhandled); + } +});