fix(workflows): end stalled writer and reviewer steps after 10 minutes

vendor/interchange/packages/workflow/src/runtime/run.ts

@@ -1772,9 +1772,15 @@ async function runStep(
     };
     bridgeAbort(abort, onOuter);
     let timer: ReturnType<typeof setTimeout> | undefined;
+    let timedOut = false;
     if (step.timeout !== undefined) {
       timer = setTimeout(() => {
-        stepAbort.abort();
+        timedOut = true;
+        stepAbort.abort(
+          new Error(
+            `step ${step.id} exceeded its ${String(step.timeout)} ms timeout`,
+          ),
+        );
       }, step.timeout);
     }
 
@@ -1964,6 +1970,7 @@ async function runStep(
       return output;
     } catch (cause) {
       const message = cause instanceof Error ? cause.message : String(cause);
+      const error = timedOut ? { message, code: "step-timeout" } : { message };
       const exhausted = attempt >= maxAttempts;
       let after = await reloadState(env, runId);
       // Cancellation wins over step-level failure: if the run is
@@ -2043,7 +2050,7 @@ async function runStep(
           at: env.clock().toISOString(),
           stepId: step.id,
           attempt,
-          error: { message },
+          error,
           retriesExhausted: true,
           routedTo: step.onFailure,
         };
@@ -2057,7 +2064,7 @@ async function runStep(
         at: env.clock().toISOString(),
         stepId: step.id,
         attempt,
-        error: { message },
+        error,
         retriesExhausted: exhausted,
       };
       after = await commit(env, runId, failed);

workflows/reviewer/src/workflow.ts

@@ -1,12 +1,17 @@
 import type { AgentDefinition, BaseEnv } from "@intx/agent/authoring";
 import { defineWorkflow, step, type Selector } from "@intx/workflow/definition";
 
+// Matches the writer's bound until reviewer runs are measured.
+// A host that raises the output cap raises this with it.
+const STEP_TIMEOUT_MS = 600_000;
+
 // One bounded review. The document and instructions arrive in the first message.
 export function createReviewerWorkflow<Env extends BaseEnv>(options: {
   readonly id: string;
   readonly address: string;
   readonly agent: AgentDefinition<Env>;
   readonly inference: Selector;
+  readonly stepTimeoutMs?: number;
   readonly credentialBindings?: Parameters<
     typeof defineWorkflow
   >[0]["credentialBindings"];
@@ -30,6 +35,8 @@ export function createReviewerWorkflow<Env extends BaseEnv>(options: {
         input: { from: "trigger.payload" },
         inference: options.inference,
         drainBehavior: "wait",
+        // No retry: a second attempt would redo work the first may already have mailed.
+        timeout: options.stepTimeoutMs ?? STEP_TIMEOUT_MS,
         triggers: 1,
       }),
     },

workflows/writer/src/workflow.test.ts

@@ -0,0 +1,98 @@
+import { expect, jest, test } from "bun:test";
+import { createDefaultDirectorRegistry } from "@intx/agent";
+import {
+  createInMemoryBlobSubstrate,
+  createInMemoryRepoStore,
+  createInMemoryScheduler,
+  createNoopDrainController,
+  runtimeRun,
+  type StepInvoker,
+  type WorkflowRuntimeEnv,
+} from "@intx/workflow";
+import { createObservedSignalChannel } from "@intx/workflow/testing";
+import { createWriterAgent } from "./agent.ts";
+import { createWriterWorkflow } from "./workflow.ts";
+
+const workflow = createWriterWorkflow({
+  id: "writer",
+  address: "writer",
+  agent: createWriterAgent({
+    id: "writer",
+    inference: { provider: "provider", model: "model" },
+    tools: [],
+    capabilities: [],
+  }),
+  inference: { literal: {} },
+});
+
+// A model that never answers: the step ends only when the runtime aborts it, and rejects with the
+// abort reason as the real step invoker does.
+const stalled: StepInvoker = (request) =>
+  new Promise((_, reject) => {
+    request.signal.addEventListener("abort", () => {
+      reject(request.signal.reason);
+    });
+  });
+
+test("a stalled write fails once on its timeout and fails the run", async () => {
+  const runId = "run-writer";
+  const repoStore = createInMemoryRepoStore();
+  const clock = () => new Date();
+  const env: WorkflowRuntimeEnv = {
+    repoStore,
+    scheduler: createInMemoryScheduler({ repoStore, clock }),
+    signalChannel: createObservedSignalChannel(),
+    blobs: createInMemoryBlobSubstrate(),
+    directors: createDefaultDirectorRegistry(),
+    authorize: async () => ({
+      effect: "allow",
+      matchingGrants: [],
+      resolvedBy: null,
+    }),
+    invokeStep: stalled,
+    spawnChild: () => Promise.reject(new Error("no_children")),
+    clock,
+    newId: (prefix) => `${prefix}-${crypto.randomUUID()}`,
+    drain: createNoopDrainController(workflow),
+    hasUpstreamSignalResolver: true,
+  };
+  jest.useFakeTimers();
+  let settled = false;
+  try {
+    const { complete } = runtimeRun(workflow, env, {
+      runId,
+      triggerPayload: { body: "Write the constraints." },
+    });
+    async function settle() {
+      try {
+        await complete;
+      } finally {
+        settled = true;
+      }
+    }
+    const settling = settle();
+    // Let the run reach its stalled step between clock moves, until the step timeout has passed.
+    for (let minute = 0; minute < 120 && !settled; minute += 1) {
+      await new Promise<void>((resolve) => queueMicrotask(resolve));
+      jest.advanceTimersByTime(60_000);
+    }
+    expect(settled).toBe(true);
+    await settling;
+  } finally {
+    jest.useRealTimers();
+  }
+
+  const events = await repoStore.read(runId);
+  const failed = events.filter((event) => event.kind === "StepFailed");
+  expect(failed).toHaveLength(1);
+  expect(failed[0]).toMatchObject({
+    stepId: "write",
+    attempt: 1,
+    retriesExhausted: true,
+    error: {
+      code: "step-timeout",
+      message: expect.stringContaining("step write"),
+    },
+  });
+  expect(events.at(-1)?.kind).toBe("RunFailed");
+});

workflows/writer/src/workflow.ts

@@ -1,12 +1,17 @@
 import type { AgentDefinition, BaseEnv } from "@intx/agent/authoring";
 import { defineWorkflow, step, type Selector } from "@intx/workflow/definition";
 
+// The longest writer step in stored runs took 276 s; ten minutes leaves room for a slower model.
+// A host that raises the output cap raises this with it.
+const STEP_TIMEOUT_MS = 600_000;
+
 // One bounded write. The task, its documents, and the artifact to write arrive in the first message.
 export function createWriterWorkflow<Env extends BaseEnv>(options: {
   readonly id: string;
   readonly address: string;
   readonly agent: AgentDefinition<Env>;
   readonly inference: Selector;
+  readonly stepTimeoutMs?: number;
   readonly credentialBindings?: Parameters<
     typeof defineWorkflow
   >[0]["credentialBindings"];
@@ -30,6 +35,8 @@ export function createWriterWorkflow<Env extends BaseEnv>(options: {
         input: { from: "trigger.payload" },
         inference: options.inference,
         drainBehavior: "wait",
+        // No retry: a second attempt would redo work the first may already have filed.
+        timeout: options.stepTimeoutMs ?? STEP_TIMEOUT_MS,
         triggers: 1,
       }),
     },