refactor(errors): represent domain failures as Schema.TaggedError

sample/workflows/ready-sweep.test.ts

@@ -112,7 +112,7 @@ function harness(
           dispatch: async (_child, rawInput, opts) => {
             const input_ = rawInput as { issueNumber: number };
             const key = opts?.dedupeKey ?? "";
-            if (held.has(key)) throw new DedupeKeyError(key, `run-holder-${key}`);
+            if (held.has(key)) throw DedupeKeyError.of({ key, holderRunId: `run-holder-${key}` });
             dispatches.push({ issueNumber: input_.issueNumber, dedupeKey: key });
             return `run-child-issue-${input_.issueNumber}`;
           },

src/index.test.ts

@@ -32,8 +32,9 @@ test("defineWorkflow round-trips through the package specifier", () => {
 });
 
 test("DedupeKeyError is catchable by type through the barrel (issue #18)", () => {
-  const err = new DedupeKeyError("issue:1", "run-holder");
+  const err = DedupeKeyError.of({ key: "issue:1", holderRunId: "run-holder" });
   expect(err instanceof Error).toBe(true);
+  expect(err._tag).toBe("DedupeKeyError");
   expect(err.key).toBe("issue:1");
   expect(err.holderRunId).toBe("run-holder");
 });

src/lib/dedupe.test.ts

@@ -14,6 +14,22 @@
 import { describe, expect, test } from "bun:test";
 import { createDedupeRegistry, DedupeKeyError } from "./dedupe";
 
+describe("DedupeKeyError as TaggedError (#34)", () => {
+  test("carries the _tag, key, and holderRunId fields", () => {
+    const err = DedupeKeyError.of({ key: "issue:41", holderRunId: "run-a" });
+    expect(err._tag).toBe("DedupeKeyError");
+    expect(err.key).toBe("issue:41");
+    expect(err.holderRunId).toBe("run-a");
+    expect(err instanceof Error).toBe(true);
+  });
+
+  test("constructs its message, so the stack header names the error and the key", () => {
+    const err = DedupeKeyError.of({ key: "issue:41", holderRunId: "run-a" });
+    expect(err.message).toBe('dedupe key held: "issue:41" is currently held by run run-a');
+    expect(err.stack?.split("\n")[0]).toBe(`DedupeKeyError: ${err.message}`);
+  });
+});
+
 describe("the holder registry (issue #15)", () => {
   test("an unclaimed key claims silently and reports its holder", () => {
     const registry = createDedupeRegistry();
@@ -40,10 +56,9 @@ describe("the holder registry (issue #15)", () => {
       }
       throw new Error("unreachable");
     })();
+    expect(err._tag).toBe("DedupeKeyError");
     expect(err.key).toBe("issue:41");
     expect(err.holderRunId).toBe("run-a");
-    expect(err.message).toContain("issue:41");
-    expect(err.message).toContain("run-a");
   });
 
   test("release frees the key when the holder releases it", () => {

src/lib/dedupe.ts

@@ -18,15 +18,29 @@
  * see `startTrackedRun` in server/runs.ts.
  */
 
-export class DedupeKeyError extends Error {
-  readonly key: string;
-  readonly holderRunId: string;
+import { Schema } from "effect";
 
-  constructor(key: string, holderRunId: string) {
-    super(`dedupe key held: "${key}" is currently held by run ${holderRunId}`);
-    this.name = "DedupeKeyError";
-    this.key = key;
-    this.holderRunId = holderRunId;
+export class DedupeKeyError extends Schema.TaggedError<DedupeKeyError>()("DedupeKeyError", {
+  key: Schema.String,
+  holderRunId: Schema.String,
+  message: Schema.String,
+}) {
+  // Bun takes the `.stack` header from the *immediate* prototype's `name`;
+  // the schema class sets it one level up, which would leave `Error: …`.
+  static {
+    this.prototype.name = "DedupeKeyError";
+  }
+
+  /**
+   * The single source of truth for the collision message (HTTP, runtime,
+   * scheduler alike). Built at construction — not a getter — so `.stack`
+   * opens with it too.
+   */
+  static of(props: { readonly key: string; readonly holderRunId: string }): DedupeKeyError {
+    return new DedupeKeyError({
+      ...props,
+      message: `dedupe key held: "${props.key}" is currently held by run ${props.holderRunId}`,
+    });
   }
 }
 
@@ -49,7 +63,8 @@ export function createDedupeRegistry(): DedupeRegistry {
     holderOf: (key) => held.get(key),
     claim(key, runId) {
       const holder = held.get(key);
-      if (holder !== undefined && holder !== runId) throw new DedupeKeyError(key, holder);
+      if (holder !== undefined && holder !== runId)
+        throw DedupeKeyError.of({ key, holderRunId: holder });
       held.set(key, runId);
     },
     release(key, runId) {

src/runtime/run.test.ts

@@ -10,7 +10,16 @@ import { describe, expect, test } from "bun:test";
 import type { RunEvent } from "../events";
 import { defineWorkflow, Schema } from "../workflow";
 import { createSlowFakeAdapter } from "../replay/adapter";
-import { startRun } from "./run";
+import { startRun, RunCancelledSignal } from "./run";
+
+describe("RunCancelledSignal as TaggedError (#34)", () => {
+  test("carries the _tag and the cancellation message", () => {
+    const signal = RunCancelledSignal.of();
+    expect(signal._tag).toBe("RunCancelledSignal");
+    expect(signal.message).toBe("run cancelled");
+    expect(signal instanceof Error).toBe(true);
+  });
+});
 
 const SLOW_CHUNKS = [
   { type: "TEXT_MESSAGE_START" },

src/runtime/run.ts

@@ -32,10 +32,12 @@ import { DedupeKeyError } from "../lib/dedupe";
 import type { AgentAdapter } from "./agent-adapter";
 import { buildAgentStepEffect } from "./agent-step";
 
-export class RunCancelledSignal extends Error {
-  constructor() {
-    super("run cancelled");
-    this.name = "RunCancelledSignal";
+export class RunCancelledSignal extends Schema.TaggedError<RunCancelledSignal>()(
+  "RunCancelledSignal",
+  { message: Schema.String },
+) {
+  static of(): RunCancelledSignal {
+    return new RunCancelledSignal({ message: "run cancelled" });
   }
 }
 
@@ -161,7 +163,7 @@ export function startRun<I, O>(
   const workspaceKind = options.workspaceKind ?? "clone";
 
   const execImpl = async (argv: ReadonlyArray<string>): Promise<ExecResult> => {
-    if (cancelled) throw new RunCancelledSignal();
+    if (cancelled) throw RunCancelledSignal.of();
 
     const execId = nextExecId();
     emit({ _tag: "ExecStarted", execId, command: [...argv], cwd: options.dir });
@@ -180,7 +182,7 @@ export function startRun<I, O>(
       durationMs,
     });
 
-    if (cancelled) throw new RunCancelledSignal();
+    if (cancelled) throw RunCancelledSignal.of();
     return result;
   };
 
@@ -189,7 +191,7 @@ export function startRun<I, O>(
     prompt: string,
     opts?: AgentCallOptions,
   ): Promise<AgentResult<O2>> => {
-    if (cancelled) throw new RunCancelledSignal();
+    if (cancelled) throw RunCancelledSignal.of();
 
     const stepId = nextStepId();
     // Precedence (issue #16): per-call option > the starting schedule's
@@ -250,7 +252,7 @@ export function startRun<I, O>(
             : {}),
           ...(handle.partial.usage !== undefined ? { usage: handle.partial.usage } : {}),
         });
-        throw new RunCancelledSignal();
+        throw RunCancelledSignal.of();
       }
 
       const message = Cause.pretty(cause);

src/server/dedupe.test.ts

@@ -268,8 +268,7 @@ describe("dedupe keys through startTrackedRun (issue #15)", () => {
     if (err instanceof DedupeKeyError) {
       expect(err.key).toBe("item:41");
       expect(err.holderRunId).toBe("run-holder");
-      expect(err.message).toContain("item:41");
-      expect(err.message).toContain("run-holder");
+      expect(err._tag).toBe("DedupeKeyError");
     }
 
     gate.release();

src/server/http.test.ts

@@ -609,6 +609,9 @@ describe("POST /api/runs {workflowId, input} (D31)", () => {
         body: JSON.stringify({ workflowId: "registry-test", input: { issueNumber: 2 } }),
       });
       expect(second.status).toBe(409);
+      expect(((await second.json()) as { error: string }).error).toBe(
+        "concurrency limit reached (max 1 concurrent runs)",
+      );
 
       await waitForTerminal(db, runId, 10_000);
 

src/server/http.ts

@@ -348,7 +348,7 @@ export function createHandler(options: ServerOptions): (req: Request) => Promise
       const maxConcurrentRuns = options.config?.maxConcurrentRuns;
       if (maxConcurrentRuns !== undefined && !admitRun(maxConcurrentRuns, activeRunIds().length)) {
         return json(
-          { error: `concurrency limit reached (max ${maxConcurrentRuns} concurrent runs)` },
+          { error: ConcurrencyLimitError.of({ maxConcurrentRuns }).message },
           { status: 409 },
         );
       }
@@ -544,7 +544,7 @@ export function createHandler(options: ServerOptions): (req: Request) => Promise
       const maxConcurrentRuns = options.config.maxConcurrentRuns;
       if (!admitRun(maxConcurrentRuns, activeRunIds().length)) {
         return json(
-          { error: `concurrency limit reached (max ${maxConcurrentRuns} concurrent runs)` },
+          { error: ConcurrencyLimitError.of({ maxConcurrentRuns }).message },
           { status: 409 },
         );
       }

src/server/runs.test.ts

@@ -20,6 +20,7 @@ import { createSlowFakeAdapter } from "../replay/adapter";
 import { defineWorkflow, Schema } from "../workflow";
 import {
   ConcurrencyLimitError,
+  DispatchCapError,
   activeRunIds,
   cancelRegisteredRun,
   getActiveHandle,
@@ -71,6 +72,28 @@ async function waitFor(predicate: () => boolean, timeoutMs = 5_000): Promise<voi
   throw new Error("condition not met within timeout");
 }
 
+describe("domain errors as TaggedError (#34)", () => {
+  test("ConcurrencyLimitError carries _tag and maxConcurrentRuns", () => {
+    const err = ConcurrencyLimitError.of({ maxConcurrentRuns: 5 });
+    expect(err._tag).toBe("ConcurrencyLimitError");
+    expect(err.maxConcurrentRuns).toBe(5);
+    expect(err instanceof Error).toBe(true);
+  });
+
+  test("DispatchCapError carries _tag and message", () => {
+    const err = new DispatchCapError({ message: "depth exceeded" });
+    expect(err._tag).toBe("DispatchCapError");
+    expect(err.message).toContain("depth exceeded");
+    expect(err instanceof Error).toBe(true);
+  });
+
+  test("ConcurrencyLimitError constructs its message, so the stack header names it", () => {
+    const err = ConcurrencyLimitError.of({ maxConcurrentRuns: 5 });
+    expect(err.message).toBe("concurrency limit reached (max 5 concurrent runs)");
+    expect(err.stack?.split("\n")[0]).toBe(`ConcurrencyLimitError: ${err.message}`);
+  });
+});
+
 describe("startTrackedRun admission (M1: the slot is reserved before any await)", () => {
   test("a second start while the first is still reserving is refused atomically, and the slot frees after", async () => {
     const { root, finish } = tmpRoot();

src/server/runs.ts

@@ -13,6 +13,7 @@
  * start rather than dropped.
  */
 
+import { Schema } from "effect";
 import type { Database } from "bun:sqlite";
 import { rm } from "node:fs/promises";
 import { admitRun } from "./admission";
@@ -41,21 +42,33 @@ export const DEFAULT_MAX_DISPATCH_DEPTH = 5;
 export const DEFAULT_MAX_CHILDREN_PER_RUN = 20;
 
 /** Issue #14: a `ctx.dispatch` rejected by a cap, for the parent to surface. */
-export class DispatchCapError extends Error {
-  constructor(message: string) {
-    super(message);
-    this.name = "DispatchCapError";
-  }
-}
+export class DispatchCapError extends Schema.TaggedError<DispatchCapError>()("DispatchCapError", {
+  message: Schema.String,
+}) {}
 
 function isReserved(entry: RunHandle<unknown> | ReservedSlot | undefined): boolean {
   return entry !== undefined && !("result" in entry) && "cancelled" in entry;
 }
 
-export class ConcurrencyLimitError extends Error {
-  constructor(maxConcurrentRuns: number) {
-    super(`concurrency limit reached (max ${maxConcurrentRuns} concurrent runs)`);
-    this.name = "ConcurrencyLimitError";
+export class ConcurrencyLimitError extends Schema.TaggedError<ConcurrencyLimitError>()(
+  "ConcurrencyLimitError",
+  { maxConcurrentRuns: Schema.Number, message: Schema.String },
+) {
+  // Bun takes the `.stack` header from the *immediate* prototype's `name`;
+  // the schema class sets it one level up, which would leave `Error: …`.
+  static {
+    this.prototype.name = "ConcurrencyLimitError";
+  }
+
+  /**
+   * The single source of truth for the message (HTTP, runtime, scheduler
+   * alike). Built at construction — not a getter — so `.stack` opens with it too.
+   */
+  static of(props: { readonly maxConcurrentRuns: number }): ConcurrencyLimitError {
+    return new ConcurrencyLimitError({
+      ...props,
+      message: `concurrency limit reached (max ${props.maxConcurrentRuns} concurrent runs)`,
+    });
   }
 }
 
@@ -231,21 +244,22 @@ async function dispatchChildRun(
     env.maxConcurrentRuns !== undefined &&
     !admitRun(env.maxConcurrentRuns, activeRunIds().length)
   ) {
-    throw new ConcurrencyLimitError(env.maxConcurrentRuns);
+    throw ConcurrencyLimitError.of({ maxConcurrentRuns: env.maxConcurrentRuns });
   }
 
   const depth = dispatchDepth(db, parentRunId);
   if (depth + 1 > maxDepth) {
-    throw new DispatchCapError(
-      `dispatch depth exceeded: run ${parentRunId} is nested ${depth} levels deep; ` +
+    throw new DispatchCapError({
+      message:
+        `dispatch depth exceeded: run ${parentRunId} is nested ${depth} levels deep; ` +
         `max ${maxDepth} (a workflow that dispatches itself must not fill the daemon)`,
-    );
+    });
   }
   const childCount = countDispatchedChildren(db, parentRunId);
   if (childCount >= maxChildren) {
-    throw new DispatchCapError(
-      `dispatch child cap exceeded: run ${parentRunId} already dispatched ${childCount} children; max ${maxChildren}`,
-    );
+    throw new DispatchCapError({
+      message: `dispatch child cap exceeded: run ${parentRunId} already dispatched ${childCount} children; max ${maxChildren}`,
+    });
   }
 
   // Issue #15: the collision check is synchronous with the child id in hand
@@ -313,7 +327,7 @@ export async function startTrackedRun(
   }
   if (existing === undefined && options.maxConcurrentRuns !== undefined) {
     if (!admitRun(options.maxConcurrentRuns, active.size)) {
-      throw new ConcurrencyLimitError(options.maxConcurrentRuns);
+      throw ConcurrencyLimitError.of({ maxConcurrentRuns: options.maxConcurrentRuns });
     }
     active.set(runId, { cancelled: false });
   }