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

Author: TheGreatAxios

workflows/reviewer/src/workflow.ts

@@ -1,6 +1,11 @@
 import type { AgentDefinition, BaseEnv } from "@intx/agent/authoring";
 import { defineWorkflow, step, type Selector } from "@intx/workflow/definition";
 
+// No reviewer runs are measured yet, so this reuses the writer's bound: a slow model writing a full
+// 32,768-token reply. A stalled step fails instead of holding its run open. No retry: a retried
+// attempt would redo work the first may already have mailed.
+const STEP_TIMEOUT_MS = 1_800_000;
+
 // One bounded review. The document and instructions arrive in the first message.
 export function createReviewerWorkflow<Env extends BaseEnv>(options: {
   readonly id: string;
@@ -30,6 +35,7 @@ export function createReviewerWorkflow<Env extends BaseEnv>(options: {
         input: { from: "trigger.payload" },
         inference: options.inference,
         drainBehavior: "wait",
+        timeout: STEP_TIMEOUT_MS,
         triggers: 1,
       }),
     },

workflows/writer/src/workflow.test.ts

@@ -0,0 +1,93 @@
+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." },
+    }).complete.finally(() => {
+      settled = true;
+    });
+    // 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 complete;
+  } 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: {
+      message: expect.stringMatching(
+        /^step write exceeded its \d+ ms timeout$/,
+      ),
+    },
+  });
+  expect(events.at(-1)?.kind).toBe("RunFailed");
+});

workflows/writer/src/workflow.ts

@@ -1,6 +1,11 @@
 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, but a slow model writing a full 32,768-token
+// reply can take far longer. A stalled step fails instead of holding its run open. No retry: a
+// retried attempt would redo work the first may already have filed.
+const STEP_TIMEOUT_MS = 1_800_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;
@@ -30,6 +35,7 @@ export function createWriterWorkflow<Env extends BaseEnv>(options: {
         input: { from: "trigger.payload" },
         inference: options.inference,
         drainBehavior: "wait",
+        timeout: STEP_TIMEOUT_MS,
         triggers: 1,
       }),
     },