fix(daemon): guard senseReflex.on null check in reflex-scheduler

Fixes crash when reflex config has no 'on' field (null).

biome.json

@@ -19,7 +19,7 @@
   },
   "overrides": [
     {
-      "include": ["tsup.config.ts", "*/rslib.config.ts"],
+      "include": ["tsup.config.ts", "*/rslib.config.ts", "packages/khala/src/index.ts"],
       "linter": {
         "rules": {
           "style": {

docs/plans/2026-04-25-khala-mvp.md

@@ -0,0 +1,558 @@
+# Khala MVP Implementation Plan
+
+> **For Hermes:** Use subagent-driven-development skill to implement this plan task-by-task.
+
+**Goal:** Build Khala — a Cloudflare Workers + D1 + Durable Objects cloud workflow orchestrator that lets agents coordinate multi-agent workflows as a stateless worker pool.
+
+**Architecture:** Khala is a CF Worker that receives events from agents via REST API. Each workflow thread runs in a Durable Object with a JSONata moderator. Agents poll a task queue for unclaimed turns, execute locally, and POST results back. Thread messages are stored in D1.
+
+**Tech Stack:** Cloudflare Workers, D1 (SQLite), Durable Objects, Hono (routing), JSONata (moderator engine), TypeScript
+
+---
+
+## Phase 0: Project Scaffolding
+
+### Task 0.1: Create khala package
+
+**Objective:** Set up the `packages/khala` CF Worker project with wrangler, Hono, and D1 binding.
+
+**Files:**
+- Create: `packages/khala/package.json`
+- Create: `packages/khala/wrangler.toml`
+- Create: `packages/khala/tsconfig.json`
+- Create: `packages/khala/src/index.ts`
+
+**Step 1: Create package.json**
+
+```json
+{
+  "name": "@uncaged/khala",
+  "version": "0.0.1",
+  "private": true,
+  "scripts": {
+    "dev": "wrangler dev",
+    "deploy": "wrangler deploy",
+    "test": "vitest run"
+  },
+  "dependencies": {
+    "hono": "^4.7.0",
+    "jsonata": "^2.0.5"
+  },
+  "devDependencies": {
+    "@cloudflare/workers-types": "^4.20250410.0",
+    "vitest": "^4.1.5",
+    "wrangler": "^4.14.0"
+  }
+}
+```
+
+**Step 2: Create wrangler.toml**
+
+```toml
+name = "khala"
+main = "src/index.ts"
+compatibility_date = "2025-04-01"
+
+[[d1_databases]]
+binding = "DB"
+database_name = "khala"
+database_id = "placeholder"
+
+[durable_objects]
+bindings = [
+  { name = "THREAD", class_name = "ThreadDO" }
+]
+
+[[migrations]]
+tag = "v1"
+new_classes = ["ThreadDO"]
+```
+
+**Step 3: Create minimal Hono entrypoint**
+
+```typescript
+// src/index.ts
+import { Hono } from "hono";
+
+export type Env = {
+  DB: D1Database;
+  THREAD: DurableObjectNamespace;
+};
+
+const app = new Hono<{ Bindings: Env }>();
+
+app.get("/health", (c) => c.json({ ok: true }));
+
+export default app;
+```
+
+**Step 4: Create tsconfig.json**
+
+```json
+{
+  "compilerOptions": {
+    "target": "ES2022",
+    "module": "ES2022",
+    "moduleResolution": "bundler",
+    "strict": true,
+    "noUncheckedIndexedAccess": true,
+    "types": ["@cloudflare/workers-types"]
+  },
+  "include": ["src"]
+}
+```
+
+**Step 5: Install dependencies and verify**
+
+```bash
+cd packages/khala && pnpm install
+pnpm exec wrangler types  # generates worker-configuration.d.ts
+```
+
+**Step 6: Commit**
+
+```bash
+git add packages/khala/
+git commit -m "chore(khala): scaffold CF Worker package"
+```
+
+---
+
+## Phase 1: D1 Schema & Data Layer
+
+### Task 1.1: Create D1 migration — core tables
+
+**Objective:** Define D1 schema for agents, threads, messages, and task queue.
+
+**Files:**
+- Create: `packages/khala/migrations/0001_initial.sql`
+
+**SQL:**
+
+```sql
+-- Agent registry
+CREATE TABLE agents (
+  id TEXT PRIMARY KEY,           -- agent name (e.g. "tuanzi")
+  token_hash TEXT NOT NULL,      -- bcrypt/sha256 hash of agent token
+  created_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+-- Workflow threads
+CREATE TABLE threads (
+  id TEXT PRIMARY KEY,           -- ulid
+  workflow TEXT NOT NULL,        -- workflow name (e.g. "code-review")
+  status TEXT NOT NULL DEFAULT 'active',  -- active | completed | failed
+  initiator TEXT NOT NULL,       -- agent id or external caller
+  result TEXT,                   -- final result JSON (set on completion)
+  created_at TEXT NOT NULL DEFAULT (datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+-- Thread messages (append-only)
+CREATE TABLE messages (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  thread_id TEXT NOT NULL REFERENCES threads(id),
+  role TEXT NOT NULL,            -- role name or "__moderator__"
+  content TEXT NOT NULL,
+  meta TEXT,                     -- JSON
+  step INTEGER NOT NULL,         -- 0-indexed step number
+  agent_id TEXT,                 -- which agent executed this turn
+  created_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+CREATE INDEX idx_messages_thread ON messages(thread_id, step);
+
+-- Task queue
+CREATE TABLE tasks (
+  id TEXT PRIMARY KEY,           -- ulid
+  thread_id TEXT NOT NULL REFERENCES threads(id),
+  role TEXT NOT NULL,
+  instruction TEXT NOT NULL,     -- turn instruction from moderator
+  status TEXT NOT NULL DEFAULT 'open',  -- open | claimed | completed | expired
+  claim_id TEXT,                 -- set when claimed
+  claimed_by TEXT,               -- agent id
+  claimed_at TEXT,
+  timeout_seconds INTEGER NOT NULL DEFAULT 300,
+  created_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+CREATE INDEX idx_tasks_status ON tasks(status, created_at);
+```
+
+**Step 1: Write the migration file**
+
+**Step 2: Apply locally**
+
+```bash
+cd packages/khala
+pnpm exec wrangler d1 migrations apply khala --local
+```
+
+**Step 3: Commit**
+
+```bash
+git add packages/khala/migrations/
+git commit -m "feat(khala): D1 schema — agents, threads, messages, tasks"
+```
+
+### Task 1.2: Create data access functions
+
+**Objective:** Type-safe D1 query functions for all tables.
+
+**Files:**
+- Create: `packages/khala/src/db.ts`
+- Create: `packages/khala/src/types.ts`
+
+**types.ts:**
+
+```typescript
+export type Agent = {
+  id: string;
+  token_hash: string;
+  created_at: string;
+};
+
+export type Thread = {
+  id: string;
+  workflow: string;
+  status: "active" | "completed" | "failed";
+  initiator: string;
+  result: string | null;
+  created_at: string;
+  updated_at: string;
+};
+
+export type Message = {
+  id: number;
+  thread_id: string;
+  role: string;
+  content: string;
+  meta: string | null;
+  step: number;
+  agent_id: string | null;
+  created_at: string;
+};
+
+export type Task = {
+  id: string;
+  thread_id: string;
+  role: string;
+  instruction: string;
+  status: "open" | "claimed" | "completed" | "expired";
+  claim_id: string | null;
+  claimed_by: string | null;
+  claimed_at: string | null;
+  timeout_seconds: number;
+  created_at: string;
+};
+```
+
+**db.ts:** Query functions — `createThread`, `appendMessage`, `createTask`, `claimTask`, `completeTask`, `getOpenTasks`, `getThreadMessages`, etc. Each is a plain function taking `D1Database` as first arg.
+
+**Step 1: Write types.ts**
+
+**Step 2: Write db.ts with all query functions**
+
+Key functions:
+- `createThread(db, workflow, initiator) → Thread`
+- `appendMessage(db, threadId, role, content, meta, step, agentId) → Message`
+- `createTask(db, threadId, role, instruction, timeoutSeconds) → Task`
+- `claimTask(db, taskId, agentId) → { ok: true, claimId } | { ok: false }`
+- `completeTask(db, taskId, claimId) → boolean`
+- `expireTimedOutTasks(db) → number` (count expired)
+- `getOpenTasks(db, limit) → Task[]`
+- `getThreadMessages(db, threadId, opts?) → Message[]` (opts: role, since, step, last)
+
+Use `ulid()` for IDs (add `ulidx` dependency).
+
+**Step 3: Commit**
+
+```bash
+git add packages/khala/src/
+git commit -m "feat(khala): data access layer — types and D1 queries"
+```
+
+---
+
+## Phase 2: Auth & Agent Registry
+
+### Task 2.1: Agent auth middleware
+
+**Objective:** Bearer token auth for agents. Hash-based token verification.
+
+**Files:**
+- Create: `packages/khala/src/auth.ts`
+- Modify: `packages/khala/src/index.ts`
+
+**auth.ts:**
+
+```typescript
+import { createMiddleware } from "hono/factory";
+import type { Env } from "./index.ts";
+
+export const agentAuth = createMiddleware<{ Bindings: Env }>(async (c, next) => {
+  const header = c.req.header("Authorization");
+  if (!header?.startsWith("Bearer ")) {
+    return c.json({ error: "missing token" }, 401);
+  }
+  const token = header.slice(7);
+  const hash = await sha256(token);
+  const agent = await c.env.DB.prepare(
+    "SELECT id FROM agents WHERE token_hash = ?"
+  ).bind(hash).first<{ id: string }>();
+  if (!agent) {
+    return c.json({ error: "invalid token" }, 401);
+  }
+  c.set("agentId", agent.id);
+  await next();
+});
+
+async function sha256(input: string): Promise<string> {
+  const data = new TextEncoder().encode(input);
+  const buf = await crypto.subtle.digest("SHA-256", data);
+  return [...new Uint8Array(buf)].map(b => b.toString(16).padStart(2, "0")).join("");
+}
+```
+
+### Task 2.2: Admin routes for agent management
+
+**Objective:** Admin API to register/remove agents (protected by admin secret in env).
+
+**Files:**
+- Create: `packages/khala/src/routes/admin.ts`
+
+**Routes:**
+- `POST /admin/agents` — body `{ id, token }` → hash token, insert agent
+- `DELETE /admin/agents/:id` — remove agent
+- `GET /admin/agents` — list agents (no tokens)
+
+Protected by `ADMIN_SECRET` env var check.
+
+**Commit:**
+
+```bash
+git commit -m "feat(khala): agent auth middleware and admin API"
+```
+
+---
+
+## Phase 3: Workflow Engine (Durable Object)
+
+### Task 3.1: Workflow registry
+
+**Objective:** Load workflow definitions from a simple in-memory registry (hardcoded for MVP, later from D1 or KV).
+
+**Files:**
+- Create: `packages/khala/src/workflows.ts`
+
+**Structure:**
+
+```typescript
+export type CloudRole = {
+  prompt: string;
+};
+
+export type CloudWorkflowDef = {
+  name: string;
+  roles: Record<string, CloudRole>;
+  moderator: string;  // JSONata expression
+};
+
+// For MVP: hardcoded registry
+const registry = new Map<string, CloudWorkflowDef>();
+
+export function registerWorkflow(def: CloudWorkflowDef): void {
+  registry.set(def.name, def);
+}
+
+export function getWorkflow(name: string): CloudWorkflowDef | null {
+  return registry.get(name) ?? null;
+}
+```
+
+### Task 3.2: ThreadDO — Durable Object
+
+**Objective:** Each workflow thread runs as a Durable Object. The DO manages the moderator state machine, creates tasks, and processes responses.
+
+**Files:**
+- Create: `packages/khala/src/thread-do.ts`
+
+**Key behavior:**
+
+1. `POST /start` — Initialize thread: save workflow def, run moderator to get first turn, create task
+2. `POST /response` — Agent posts turn result: validate claim_id, append message, run moderator for next turn or END
+3. `GET /messages` — Query thread messages with filters
+
+**Moderator execution:**
+- Build context from messages (start frame + role steps)
+- Evaluate JSONata expression → returns `{ role: "reviewer" }` or `{ role: "__end__" }`
+- If not END, create new task in queue
+- If END, mark thread completed, set result
+
+**Important:** The DO holds workflow state in memory during the request but persists everything to D1. The DO itself uses `ctx.storage` only for the thread ID mapping.
+
+### Task 3.3: Wire ThreadDO into worker
+
+**Objective:** Export the DO class, add routes that proxy to the DO.
+
+**Files:**
+- Modify: `packages/khala/src/index.ts`
+
+**Routes:**
+- `POST /workflows/:name/threads` — Create thread → instantiate DO → start
+- `POST /threads/:id/response` — Forward to DO
+- `GET /threads/:id/messages` — Forward to DO (or query D1 directly)
+- `GET /threads/:id` — Thread status
+
+**Commit:**
+
+```bash
+git commit -m "feat(khala): ThreadDO workflow engine with JSONata moderator"
+```
+
+---
+
+## Phase 4: Task Queue API
+
+### Task 4.1: Task queue endpoints
+
+**Objective:** Agents poll for work and claim tasks.
+
+**Files:**
+- Create: `packages/khala/src/routes/tasks.ts`
+
+**Routes (all require agentAuth):**
+- `GET /tasks` — List open tasks (optionally filter by workflow)
+- `POST /tasks/:id/claim` — Claim a task → returns `{ claimId, role, instruction, threadId }`
+- `POST /tasks/:id/release` — Release a claimed task back to queue
+
+**Claim logic:**
+- Atomic: UPDATE ... WHERE status = 'open' → if rowsWritten = 0, already claimed
+- Returns claim_id (ulid) for optimistic lock on response
+
+### Task 4.2: Task timeout sweep
+
+**Objective:** Periodically expire timed-out tasks back to open.
+
+**Implementation:** CF Worker Cron Trigger (every 1 minute) that calls `expireTimedOutTasks(db)`.
+
+**Files:**
+- Modify: `packages/khala/src/index.ts` (add scheduled handler)
+- Modify: `packages/khala/wrangler.toml` (add cron trigger)
+
+```toml
+[triggers]
+crons = ["* * * * *"]
+```
+
+**Commit:**
+
+```bash
+git commit -m "feat(khala): task queue API with claim/release and timeout sweep"
+```
+
+---
+
+## Phase 5: Agent-Side Integration (KhalaSense)
+
+### Task 5.1: Khala client library
+
+**Objective:** A small client that agents use to interact with Khala.
+
+**Files:**
+- Create: `packages/core/src/khala-client.ts`
+
+**API:**
+
+```typescript
+export type KhalaClientConfig = {
+  baseUrl: string;
+  token: string;
+};
+
+export function createKhalaClient(config: KhalaClientConfig) {
+  return {
+    pollTasks: () => GET /tasks,
+    claimTask: (taskId) => POST /tasks/:id/claim,
+    submitResponse: (threadId, content, meta, claimId) => POST /threads/:id/response,
+    getMessages: (threadId, opts) => GET /threads/:id/messages,
+  };
+}
+```
+
+### Task 5.2: KhalaSense
+
+**Objective:** A Sense that polls Khala for open tasks and emits signals.
+
+**Files:**
+- Create: `packages/daemon/src/senses/khala-sense.ts`
+
+**Behavior:**
+- `compute()`: poll `/tasks`, if tasks available → return task info as signal value
+- Reflex picks up signal → triggers a workflow that executes the turn locally
+- After local execution → POST response back to Khala
+
+**This task depends on understanding the existing Sense pattern in daemon. Check `packages/daemon/src/` for examples.**
+
+**Commit:**
+
+```bash
+git commit -m "feat(khala): KhalaSense — agent-side polling and integration"
+```
+
+---
+
+## Phase 6: End-to-End Demo
+
+### Task 6.1: Register a demo workflow
+
+**Objective:** Register a simple 2-role "ping-pong" workflow for testing.
+
+**Files:**
+- Create: `packages/khala/src/workflows/ping-pong.ts`
+
+**Workflow:**
+- Roles: `pinger` (says ping), `ponger` (says pong)
+- Moderator: alternate pinger/ponger for 3 rounds then END
+- JSONata: `steps.length >= 6 ? { "role": "__end__" } : steps.length % 2 = 0 ? { "role": "pinger" } : { "role": "ponger" }`
+
+### Task 6.2: Integration test
+
+**Objective:** Test the full flow with miniflare.
+
+**Files:**
+- Create: `packages/khala/src/__tests__/e2e.test.ts`
+
+**Test:**
+1. Create thread via API
+2. Poll tasks → get first task
+3. Claim task
+4. POST response
+5. Poll again → get next task
+6. Repeat until workflow completes
+7. Verify thread status = completed and all messages present
+
+**Commit:**
+
+```bash
+git commit -m "feat(khala): ping-pong demo workflow and e2e test"
+```
+
+---
+
+## Summary
+
+| Phase | Tasks | Description |
+|-------|-------|-------------|
+| 0 | 0.1 | Project scaffolding |
+| 1 | 1.1-1.2 | D1 schema & data layer |
+| 2 | 2.1-2.2 | Auth & agent registry |
+| 3 | 3.1-3.3 | Workflow engine (DO + moderator) |
+| 4 | 4.1-4.2 | Task queue API |
+| 5 | 5.1-5.2 | Agent-side integration |
+| 6 | 6.1-6.2 | End-to-end demo |
+
+**Deployment:** `khala.shazhou.workers.dev`
+
+**First milestone:** Phase 0-4 (cloud side complete), testable with curl.
+**Second milestone:** Phase 5-6 (agent integration + demo).

packages/cli/src/__tests__/workflow.test.ts

@@ -45,7 +45,7 @@ function upsertRun(
 ): void {
   store.upsertWorkflowRun(
     { source: "workflow", type: status, refId: runId, payload: null, timestamp: timestampMs },
-    { runId, workflow, status, timestamp: timestampMs },
+    { runId, workflow, status, timestamp: timestampMs, exitCode: null },
   );
 }
 
@@ -83,6 +83,7 @@ describe("statusIcon", () => {
     ["crashed", "💥"],
     ["dropped", "🗑"],
     ["interrupted", "⚠️"],
+    ["killed", "🛑"],
   ] as const)("maps status=%s to icon=%s", (status, icon) => {
     expect(statusIcon(status)).toBe(icon);
   });

packages/cli/src/commands/workflow.ts

@@ -7,7 +7,7 @@ import { defineCommand } from "citty";
 import { stringify } from "yaml";
 
 import type { LogStore, ThreadRoundRow, WorkflowRun } from "@uncaged/nerve-store";
-import { triggerWorkflowViaDaemon } from "../daemon-client.js";
+import { killWorkflowViaDaemon, triggerWorkflowViaDaemon } from "../daemon-client.js";
 import { loadDaemonModule } from "../workspace-daemon.js";
 import { getNerveRoot, getSocketPath, isRunning } from "../workspace.js";
 
@@ -59,6 +59,8 @@ export function statusIcon(status: WorkflowRun["status"]): string {
       return "🗑";
     case "interrupted":
       return "⚠️";
+    case "killed":
+      return "🛑";
     default: {
       const _exhaustive: never = status;
       return `?(${_exhaustive})`;
@@ -79,7 +81,8 @@ export function getAllWorkflowRuns(store: LogStore, filterWorkflow: string | nul
  */
 export function formatRunLine(run: WorkflowRun): string {
   const icon = statusIcon(run.status);
-  return `  ${icon} ${run.runId}  workflow=${run.workflow}  status=${run.status}  timestamp=${formatTs(run.timestamp)}\n`;
+  const exitCodeStr = run.exitCode !== null ? `  exit_code=${run.exitCode}` : "";
+  return `  ${icon} ${run.runId}  workflow=${run.workflow}  status=${run.status}${exitCodeStr}  timestamp=${formatTs(run.timestamp)}\n`;
 }
 
 /**
@@ -563,6 +566,46 @@ const workflowTriggerCommand = defineCommand({
   },
 });
 
+// ---------------------------------------------------------------------------
+// nerve workflow kill <runId>
+// ---------------------------------------------------------------------------
+
+const workflowKillCommand = defineCommand({
+  meta: {
+    name: "kill",
+    description: "Kill a running or queued workflow thread by runId",
+  },
+  args: {
+    runId: {
+      type: "positional",
+      description: "The run ID to kill",
+    },
+  },
+  async run({ args }) {
+    if (!isRunning()) {
+      process.stderr.write("❌ Nerve daemon is not running. Start it with `nerve start`.\n");
+      process.exit(1);
+    }
+
+    const socketPath = getSocketPath();
+    let response: DaemonIpcTriggerResponse;
+    try {
+      response = await killWorkflowViaDaemon(socketPath, args.runId);
+    } catch (e) {
+      const msg = e instanceof Error ? e.message : String(e);
+      process.stderr.write(`❌ Failed to contact daemon: ${msg}\n`);
+      process.exit(1);
+    }
+
+    if (!response.ok) {
+      process.stderr.write(`❌ Kill failed: ${response.error}\n`);
+      process.exit(1);
+    }
+
+    process.stdout.write(`✅ Kill signal sent for run "${args.runId}".\n`);
+  },
+});
+
 // ---------------------------------------------------------------------------
 // nerve workflow (parent command)
 // ---------------------------------------------------------------------------
@@ -577,5 +620,6 @@ export const workflowCommand = defineCommand({
     inspect: workflowInspectCommand,
     thread: workflowThreadCommand,
     trigger: workflowTriggerCommand,
+    kill: workflowKillCommand,
   },
 });

packages/cli/src/daemon-client.ts

@@ -167,3 +167,15 @@ export function listSensesViaDaemon(socketPath: string): Promise<DaemonIpcListSe
   const message: DaemonIpcRequest = { type: "list-senses" };
   return sendAndReceive(socketPath, message, parseListSensesResponse);
 }
+
+/**
+ * Send a kill-workflow message to the running daemon via its Unix socket.
+ * Resolves with the daemon's response or rejects on connection/timeout errors.
+ */
+export function killWorkflowViaDaemon(
+  socketPath: string,
+  runId: string,
+): Promise<DaemonIpcTriggerResponse> {
+  const message: DaemonIpcRequest = { type: "kill-workflow", runId };
+  return sendAndReceive(socketPath, message, parseDaemonResponse);
+}

packages/core/src/daemon-ipc-protocol.ts

@@ -27,11 +27,18 @@ export type DaemonIpcListSensesRequest = {
   type: "list-senses";
 };
 
+/** Client → daemon: kill a running or queued workflow thread by runId. */
+export type DaemonIpcKillWorkflowRequest = {
+  type: "kill-workflow";
+  runId: string;
+};
+
 /** Union of all JSON requests the daemon IPC server accepts. */
 export type DaemonIpcRequest =
   | DaemonIpcTriggerWorkflowRequest
   | DaemonIpcTriggerSenseRequest
-  | DaemonIpcListSensesRequest;
+  | DaemonIpcListSensesRequest
+  | DaemonIpcKillWorkflowRequest;
 
 /** Successful trigger / trigger-sense reply (no body). */
 export type DaemonIpcTriggerOkResponse = { ok: true };
@@ -87,6 +94,10 @@ export function parseDaemonIpcRequest(line: string): DaemonIpcRequest | null {
     if (req.type === "list-senses") {
       return { type: "list-senses" };
     }
+    if (req.type === "kill-workflow") {
+      if (typeof req.runId !== "string" || req.runId.length === 0) return null;
+      return { type: "kill-workflow", runId: req.runId };
+    }
     return null;
   } catch {
     return null;

packages/core/src/index.ts

@@ -38,6 +38,7 @@ export type {
   DaemonIpcTriggerWorkflowRequest,
   DaemonIpcTriggerSenseRequest,
   DaemonIpcListSensesRequest,
+  DaemonIpcKillWorkflowRequest,
   DaemonIpcRequest,
   DaemonIpcTriggerOkResponse,
   DaemonIpcErrorResponse,

packages/daemon/src/__tests__/crash-recovery.test.ts

@@ -76,6 +76,7 @@ function makeLogStore(
     timestamp: number;
   }> = [],
 ) {
+  const runsWithExitCode = activeRuns.map((r) => ({ ...r, exitCode: null }));
   const store = {
     append: vi.fn(),
     query: vi.fn(() => []),
@@ -86,9 +87,9 @@ function makeLogStore(
     getWorkflowRun: vi.fn(() => null),
     getActiveWorkflowRuns: vi.fn((_workflowName?: string) => {
       if (_workflowName !== undefined) {
-        return activeRuns.filter((r) => r.workflow === _workflowName);
+        return runsWithExitCode.filter((r) => r.workflow === _workflowName);
       }
-      return activeRuns;
+      return runsWithExitCode;
     }),
     getTriggerPayload: vi.fn((): unknown => ({ value: 42 })),
     getThreadEvents: vi.fn(

packages/daemon/src/__tests__/daemon-ipc.test.ts

@@ -31,6 +31,7 @@ function makeMockWorkflowManager() {
     stop: vi.fn(async () => {}),
     totalActiveCount: vi.fn(() => 0),
     drainAndRespawn: vi.fn(async () => {}),
+    drainWhenIdle: vi.fn(),
     updateConfig: vi.fn(),
     getActiveWorkflowRuns: vi.fn(() => []),
   };

packages/daemon/src/__tests__/hot-reload.test.ts

@@ -3,6 +3,7 @@
  *
  * Verifies that:
  * - drainAndRespawn() sends shutdown, waits for exit, then respawns the worker
+ * - drainWhenIdle() defers drain+respawn until in-flight threads finish
  * - Kernel dispatches handleWorkflowFileChange when file-watcher emits a workflow change
  * - Kernel logs a workflow_reload system event on hot reload
  * - drainAndRespawn on a non-existent worker is a no-op
@@ -238,6 +239,199 @@ describe("WorkflowManager — drainAndRespawn (Phase 3 hot reload)", () => {
   });
 });
 
+describe("WorkflowManager — drainWhenIdle (hot reload without interrupting in-flight)", () => {
+  beforeEach(() => {
+    mockChildren.length = 0;
+    vi.useFakeTimers({ shouldAdvanceTime: true });
+  });
+
+  afterEach(() => {
+    vi.useRealTimers();
+    vi.clearAllMocks();
+  });
+
+  it("does not send shutdown while a thread is still active", () => {
+    const logStore = makeLogStore();
+    const config = makeWfConfig({ "my-wf": { concurrency: 1, overflow: "drop" } });
+    const mgr = createWorkflowManager("/nerve-root", config, logStore);
+
+    mgr.startWorkflow("my-wf", { prompt: "test", maxRounds: 10, dryRun: false });
+    const child = mockChildren[0];
+
+    mgr.drainWhenIdle("my-wf");
+
+    const shutdownCalls = (child.send as ReturnType<typeof vi.fn>).mock.calls.filter(
+      (args: unknown[]) =>
+        args[0] !== null &&
+        typeof args[0] === "object" &&
+        (args[0] as Record<string, unknown>).type === "shutdown",
+    );
+    expect(shutdownCalls).toHaveLength(0);
+  });
+
+  it("sends shutdown after the last active thread completes (deferred drain)", async () => {
+    const logStore = makeLogStore();
+    const config = makeWfConfig({ "my-wf": { concurrency: 1, overflow: "drop" } });
+    const mgr = createWorkflowManager("/nerve-root", config, logStore);
+
+    mgr.startWorkflow("my-wf", { prompt: "test", maxRounds: 10, dryRun: false });
+    const child = mockChildren[0];
+    const runId = (child.send as ReturnType<typeof vi.fn>).mock.calls[0][0] as { runId: string };
+
+    mgr.drainWhenIdle("my-wf");
+
+    child.emit("message", {
+      type: "thread-event",
+      runId: runId.runId,
+      eventType: "completed",
+      payload: null,
+    });
+
+    await vi.runAllTimersAsync();
+
+    expect(child.send).toHaveBeenCalledWith(expect.objectContaining({ type: "shutdown" }));
+    expect(mockChildren).toHaveLength(2);
+
+    const stopPromise = mgr.stop();
+    await vi.runAllTimersAsync();
+    await stopPromise;
+  });
+
+  it("waits for all concurrent threads before draining", async () => {
+    const logStore = makeLogStore();
+    const config = makeWfConfig({ "my-wf": { concurrency: 2, overflow: "drop" } });
+    const mgr = createWorkflowManager("/nerve-root", config, logStore);
+
+    mgr.startWorkflow("my-wf", { prompt: "a", maxRounds: 10, dryRun: false });
+    mgr.startWorkflow("my-wf", { prompt: "b", maxRounds: 10, dryRun: false });
+    const child = mockChildren[0];
+    const sendMock = child.send as ReturnType<typeof vi.fn>;
+    const runIdA = (sendMock.mock.calls[0][0] as { runId: string }).runId;
+    const runIdB = (sendMock.mock.calls[1][0] as { runId: string }).runId;
+
+    mgr.drainWhenIdle("my-wf");
+
+    child.emit("message", {
+      type: "thread-event",
+      runId: runIdA,
+      eventType: "completed",
+      payload: null,
+    });
+    await vi.runAllTimersAsync();
+
+    const shutdownBefore = sendMock.mock.calls.filter(
+      (args: unknown[]) =>
+        args[0] !== null &&
+        typeof args[0] === "object" &&
+        (args[0] as Record<string, unknown>).type === "shutdown",
+    );
+    expect(shutdownBefore).toHaveLength(0);
+
+    child.emit("message", {
+      type: "thread-event",
+      runId: runIdB,
+      eventType: "completed",
+      payload: null,
+    });
+    await vi.runAllTimersAsync();
+
+    expect(child.send).toHaveBeenCalledWith(expect.objectContaining({ type: "shutdown" }));
+
+    const stopPromise = mgr.stop();
+    await vi.runAllTimersAsync();
+    await stopPromise;
+  });
+
+  it("duplicate drainWhenIdle while busy only schedules one deferred drain", async () => {
+    const logStore = makeLogStore();
+    const config = makeWfConfig({ "my-wf": { concurrency: 1, overflow: "drop" } });
+    const mgr = createWorkflowManager("/nerve-root", config, logStore);
+
+    mgr.startWorkflow("my-wf", { prompt: "test", maxRounds: 10, dryRun: false });
+    const child = mockChildren[0];
+    const runId = (child.send as ReturnType<typeof vi.fn>).mock.calls[0][0] as { runId: string };
+
+    mgr.drainWhenIdle("my-wf");
+    mgr.drainWhenIdle("my-wf");
+
+    child.emit("message", {
+      type: "thread-event",
+      runId: runId.runId,
+      eventType: "completed",
+      payload: null,
+    });
+    await vi.runAllTimersAsync();
+
+    const shutdownCalls = (child.send as ReturnType<typeof vi.fn>).mock.calls.filter(
+      (args: unknown[]) =>
+        args[0] !== null &&
+        typeof args[0] === "object" &&
+        (args[0] as Record<string, unknown>).type === "shutdown",
+    );
+    expect(shutdownCalls).toHaveLength(1);
+
+    const stopPromise = mgr.stop();
+    await vi.runAllTimersAsync();
+    await stopPromise;
+  });
+
+  it("deferred drain runs after workflow-error clears the active thread", async () => {
+    const logStore = makeLogStore();
+    const config = makeWfConfig({ "my-wf": { concurrency: 1, overflow: "drop" } });
+    const mgr = createWorkflowManager("/nerve-root", config, logStore);
+
+    mgr.startWorkflow("my-wf", { prompt: "test", maxRounds: 10, dryRun: false });
+    const child = mockChildren[0];
+    const runId = (child.send as ReturnType<typeof vi.fn>).mock.calls[0][0] as { runId: string };
+
+    mgr.drainWhenIdle("my-wf");
+
+    child.emit("message", {
+      type: "workflow-error",
+      runId: runId.runId,
+      error: "role failed",
+      exitCode: 1,
+    });
+    await vi.runAllTimersAsync();
+
+    expect(child.send).toHaveBeenCalledWith(expect.objectContaining({ type: "shutdown" }));
+
+    const stopPromise = mgr.stop();
+    await vi.runAllTimersAsync();
+    await stopPromise;
+  });
+
+  it("drains immediately when there is no in-flight thread", async () => {
+    const logStore = makeLogStore();
+    const config = makeWfConfig({ "my-wf": { concurrency: 1, overflow: "drop" } });
+    const mgr = createWorkflowManager("/nerve-root", config, logStore);
+
+    mgr.startWorkflow("my-wf", { prompt: "once", maxRounds: 10, dryRun: false });
+    const firstChild = mockChildren[0];
+    const runId = (firstChild.send as ReturnType<typeof vi.fn>).mock.calls[0][0] as {
+      runId: string;
+    };
+
+    firstChild.emit("message", {
+      type: "thread-event",
+      runId: runId.runId,
+      eventType: "completed",
+      payload: null,
+    });
+    await vi.runAllTimersAsync();
+
+    mgr.drainWhenIdle("my-wf");
+    await vi.runAllTimersAsync();
+
+    expect(firstChild.send).toHaveBeenCalledWith(expect.objectContaining({ type: "shutdown" }));
+    expect(mockChildren).toHaveLength(2);
+
+    const stopPromise = mgr.stop();
+    await vi.runAllTimersAsync();
+    await stopPromise;
+  });
+});
+
 describe("Kernel — workflow hot reload via file-watcher (Phase 3)", () => {
   let nerveRoot: string;
 

packages/daemon/src/daemon-ipc.ts

@@ -67,6 +67,12 @@ export function createDaemonIpcServer(
         const senses = opts.listSenses();
         const resp: DaemonIpcResponse = { ok: true, senses };
         socket.write(`${JSON.stringify(resp)}\n`);
+      } else if (req.type === "kill-workflow") {
+        const found = workflowManager.killThread(req.runId);
+        const resp: DaemonIpcResponse = found
+          ? { ok: true }
+          : { ok: false, error: `Run not found or already finished: ${req.runId}` };
+        socket.write(`${JSON.stringify(resp)}\n`);
       } else {
         const _exhaustive: never = req;
         void _exhaustive;

packages/daemon/src/ipc.ts

@@ -50,13 +50,20 @@ export type ResumeThreadMessage = {
   dryRun: boolean;
 };
 
+/** Parent → Workflow Worker: kill a specific running thread */
+export type KillThreadMessage = {
+  type: "kill-thread";
+  runId: string;
+};
+
 /** Union of all messages the parent sends to a worker */
 export type ParentToWorkerMessage =
   | ComputeMessage
   | ShutdownMessage
   | HealthRequestMessage
   | StartThreadMessage
-  | ResumeThreadMessage;
+  | ResumeThreadMessage
+  | KillThreadMessage;
 
 /** Worker → Parent: compute produced a signal */
 export type SignalMessage = {
@@ -89,7 +96,13 @@ export type HealthResponseMessage = {
 // ---------------------------------------------------------------------------
 
 /** Valid lifecycle event types for a workflow thread. */
-export type ThreadEventType = "queued" | "started" | "step_complete" | "completed" | "failed";
+export type ThreadEventType =
+  | "queued"
+  | "started"
+  | "step_complete"
+  | "completed"
+  | "failed"
+  | "killed";
 
 /**
  * Workflow Worker → Parent: a thread lifecycle event.
@@ -106,6 +119,8 @@ export type WorkflowErrorMessage = {
   type: "workflow-error";
   runId: string;
   error: string;
+  /** Exit code conveying the failure reason (1=role error, 2=maxRounds exhausted). */
+  exitCode: number;
 };
 
 /** Workflow Worker → Parent: a WorkflowMessage produced by a role (for crash recovery). */
@@ -132,6 +147,7 @@ const PARENT_MSG_TYPES = new Set([
   "health-request",
   "start-thread",
   "resume-thread",
+  "kill-thread",
 ]);
 
 function validateStartThreadMsg(obj: Record<string, unknown>): string | null {
@@ -201,6 +217,12 @@ export function parseParentMessage(raw: unknown): Result<ParentToWorkerMessage>
       dryRun: obj.dryRun,
     } as ResumeThreadMessage);
   }
+  if (obj.type === "kill-thread") {
+    if (typeof obj.runId !== "string") {
+      return err(new Error("'kill-thread' message missing string 'runId'"));
+    }
+    return ok({ type: "kill-thread", runId: obj.runId } as KillThreadMessage);
+  }
   return err(new Error(`Unhandled IPC message type: "${obj.type}"`));
 }
 
@@ -254,6 +276,7 @@ function isThreadEventType(value: string): value is ThreadEventType {
     case "step_complete":
     case "completed":
     case "failed":
+    case "killed":
       return true;
     default:
       return false;
@@ -287,10 +310,12 @@ function parseWorkflowErrorMsg(obj: Record<string, unknown>): Result<WorkerToPar
   if (typeof obj.error !== "string") {
     return err(new Error("Worker 'workflow-error' message missing string 'error' field"));
   }
+  const exitCode = typeof obj.exitCode === "number" ? obj.exitCode : 1;
   return ok({
     type: "workflow-error",
     runId: obj.runId,
     error: obj.error,
+    exitCode,
   });
 }
 

packages/daemon/src/kernel-file-watch.ts

@@ -57,10 +57,7 @@ export function createKernelFileWatchHandlers(deps: KernelFileWatchDeps): Kernel
       payload: null,
       timestamp: Date.now(),
     });
-    deps.workflowManager.drainAndRespawn(workflowName).catch((e) => {
-      const msg = e instanceof Error ? e.message : String(e);
-      process.stderr.write(`[kernel] drainAndRespawn error for "${workflowName}": ${msg}\n`);
-    });
+    deps.workflowManager.drainWhenIdle(workflowName);
   }
 
   function onConfigFileChange(): void {

packages/daemon/src/reflex-scheduler.ts

@@ -164,7 +164,7 @@ export function createReflexScheduler(
       intervals.push(id);
     }
 
-    if (senseReflex.on.length > 0) {
+    if (senseReflex.on !== null && senseReflex.on.length > 0) {
       const watchedSenses = new Set(senseReflex.on);
       const unsub = bus.subscribe((signal) => {
         if (watchedSenses.has(signal.senseId)) {

packages/daemon/src/workflow-manager.ts

@@ -16,6 +16,7 @@ import { START, isPlainRecord } from "@uncaged/nerve-core";
 
 import type { LogStore, WorkflowRunStatus } from "@uncaged/nerve-store";
 import type {
+  KillThreadMessage,
   ResumeThreadMessage,
   ShutdownMessage,
   StartThreadMessage,
@@ -37,6 +38,11 @@ export type WorkflowLaunchParams = {
 export type WorkflowManager = {
   /** Trigger a new workflow thread (Sense-driven launch or CLI / IPC). */
   startWorkflow: (workflowName: string, launch: WorkflowLaunchParams) => void;
+  /**
+   * Kill a running or queued workflow thread by runId.
+   * Returns true if the thread was found, false if not found.
+   */
+  killThread: (runId: string) => boolean;
   /** Number of currently active (running) threads for a workflow. */
   activeCount: (workflowName: string) => number;
   /** Number of pending queued threads waiting to run for a workflow. */
@@ -51,6 +57,12 @@ export type WorkflowManager = {
    * Waits up to `drainTimeoutMs` for threads to complete before force-killing.
    */
   drainAndRespawn: (workflowName: string, drainTimeoutMs?: number) => Promise<void>;
+  /**
+   * Schedule a drain+respawn that waits for in-flight runs to finish first.
+   * If no runs are active, drains immediately. Otherwise marks a pending reload
+   * and drains automatically when the last active run completes.
+   */
+  drainWhenIdle: (workflowName: string) => void;
   /** Gracefully shut down all workflow workers. */
   stop: () => Promise<void>;
 };
@@ -181,6 +193,16 @@ function sendResumeThread(worker: ChildProcess, msg: ResumeThreadMessage): void
   }
 }
 
+function sendKillThread(worker: ChildProcess, runId: string): void {
+  if (worker.connected === false) return;
+  const msg: KillThreadMessage = { type: "kill-thread", runId };
+  try {
+    worker.send(msg);
+  } catch {
+    // IPC channel closed between connected check and send
+  }
+}
+
 function waitForExit(child: ChildProcess, timeoutMs: number): Promise<void> {
   return new Promise((resolve) => {
     const timer = setTimeout(() => {
@@ -206,6 +228,7 @@ export function createWorkflowManager(
   const crashTimestamps = new Map<string, number[]>();
   let stopped = false;
   let config = initialConfig;
+  const pendingDrains = new Set<string>();
 
   function getOrCreateState(workflowName: string): WorkflowState {
     let state = states.get(workflowName);
@@ -229,15 +252,24 @@ export function createWorkflowManager(
       crashed: "crashed",
       dropped: "dropped",
       interrupted: "interrupted",
+      killed: "killed",
     };
     return map[eventType] ?? null;
   }
 
+  function extractExitCode(payload: unknown): number | null {
+    if (isPlainRecord(payload) && typeof payload.exitCode === "number") {
+      return payload.exitCode;
+    }
+    return null;
+  }
+
   function logWorkflowEvent(
     workflowName: string,
     runId: string,
     eventType: string,
     payload?: unknown,
+    exitCode: number | null = null,
   ): void {
     const timestamp = Date.now();
     const serialised = payload !== undefined ? JSON.stringify(payload) : null;
@@ -252,7 +284,7 @@ export function createWorkflowManager(
           payload: serialised,
           timestamp,
         },
-        { runId, workflow: workflowName, status, timestamp },
+        { runId, workflow: workflowName, status, timestamp, exitCode },
       );
     } else {
       logStore.append({
@@ -303,17 +335,34 @@ export function createWorkflowManager(
     }
   }
 
+  /** If a hot-reload was deferred, run drain+respawn once there are no active threads. */
+  function maybeDeferredHotReloadDrain(workflowName: string): void {
+    if (!pendingDrains.has(workflowName)) return;
+    const state = states.get(workflowName);
+    if (state === undefined || state.active.size !== 0) return;
+
+    pendingDrains.delete(workflowName);
+    process.stderr.write(
+      `[workflow-manager] all runs complete for "${workflowName}", executing deferred hot-reload drain\n`,
+    );
+    drainAndRespawn(workflowName).catch((e) => {
+      const errMsg = e instanceof Error ? e.message : String(e);
+      process.stderr.write(
+        `[workflow-manager] deferred drainAndRespawn error for "${workflowName}": ${errMsg}\n`,
+      );
+    });
+  }
+
   function handleThreadEvent(workflowName: string, msg: ThreadEventMessage): void {
     const state = states.get(workflowName);
     if (state === undefined) return;
 
-    if (msg.eventType === "completed" || msg.eventType === "failed") {
+    if (msg.eventType === "completed" || msg.eventType === "failed" || msg.eventType === "killed") {
       state.active.delete(msg.runId);
       dequeueNext(workflowName);
-    }
-
-    if (msg.eventType === "completed" || msg.eventType === "failed") {
-      logWorkflowEvent(workflowName, msg.runId, msg.eventType, msg.payload);
+      const exitCode = extractExitCode(msg.payload);
+      logWorkflowEvent(workflowName, msg.runId, msg.eventType, msg.payload, exitCode);
+      maybeDeferredHotReloadDrain(workflowName);
     }
   }
 
@@ -399,12 +448,13 @@ export function createWorkflowManager(
         `[workflow-manager] worker for "${workflowName}" crashed with ${crashedCount} active thread(s)\n`,
       );
       for (const runId of state.active) {
-        logWorkflowEvent(workflowName, runId, "crashed");
+        logWorkflowEvent(workflowName, runId, "crashed", undefined, 255);
       }
     }
 
     state.active.clear();
     workers.delete(workflowName);
+    pendingDrains.delete(workflowName);
 
     if (stopped || workflowConfig(workflowName) === null) return;
 
@@ -460,7 +510,8 @@ export function createWorkflowManager(
         state.active.delete(msg.runId);
         dequeueNext(workflowName);
       }
-      logWorkflowEvent(workflowName, msg.runId, "failed", { error: msg.error });
+      logWorkflowEvent(workflowName, msg.runId, "failed", { error: msg.error }, msg.exitCode);
+      maybeDeferredHotReloadDrain(workflowName);
       return;
     }
 
@@ -541,6 +592,26 @@ export function createWorkflowManager(
     return entry;
   }
 
+  function killThread(runId: string): boolean {
+    for (const [workflowName, state] of states) {
+      const queueIdx = state.queue.findIndex((q) => q.runId === runId);
+      if (queueIdx !== -1) {
+        state.queue.splice(queueIdx, 1);
+        logWorkflowEvent(workflowName, runId, "killed", { exitCode: 137 }, 137);
+        return true;
+      }
+
+      if (state.active.has(runId)) {
+        const workerEntry = workers.get(workflowName);
+        if (workerEntry !== undefined) {
+          sendKillThread(workerEntry.process, runId);
+        }
+        return true;
+      }
+    }
+    return false;
+  }
+
   function startWorkflow(workflowName: string, launch: WorkflowLaunchParams): void {
     if (stopped) return;
 
@@ -638,9 +709,38 @@ export function createWorkflowManager(
     await waitForExit(entry.process, drainTimeoutMs);
     // The exit handler (draining branch) will respawn the worker automatically
   }
+  function drainWhenIdle(workflowName: string): void {
+    const state = states.get(workflowName);
+    const hasActiveRuns = state !== undefined && state.active.size > 0;
+
+    if (!hasActiveRuns) {
+      pendingDrains.delete(workflowName);
+      drainAndRespawn(workflowName).catch((e) => {
+        const msg = e instanceof Error ? e.message : String(e);
+        process.stderr.write(
+          `[workflow-manager] drainAndRespawn error for "${workflowName}": ${msg}\n`,
+        );
+      });
+      return;
+    }
+
+    // Defer until all active runs finish
+    if (pendingDrains.has(workflowName)) {
+      process.stderr.write(
+        `[workflow-manager] hot-reload already pending for "${workflowName}", skipping duplicate\n`,
+      );
+      return;
+    }
+
+    pendingDrains.add(workflowName);
+    process.stderr.write(
+      `[workflow-manager] deferring hot-reload for "${workflowName}" until ${state.active.size} active run(s) complete\n`,
+    );
+  }
 
   async function stop(): Promise<void> {
     stopped = true;
+    pendingDrains.clear();
     const exitPromises: Promise<void>[] = [];
     for (const entry of workers.values()) {
       sendShutdown(entry.process, entry);
@@ -652,11 +752,13 @@ export function createWorkflowManager(
 
   return {
     startWorkflow,
+    killThread,
     activeCount,
     queueLength,
     totalActiveCount,
     updateConfig,
     drainAndRespawn,
+    drainWhenIdle,
     stop,
   };
 }

packages/daemon/src/workflow-worker.ts

@@ -41,8 +41,8 @@ function sendThreadEvent(runId: string, eventType: ThreadEventType, payload: unk
   send({ type: "thread-event", runId, eventType, payload });
 }
 
-function sendWorkflowError(runId: string, error: string): void {
-  send({ type: "workflow-error", runId, error });
+function sendWorkflowError(runId: string, error: string, exitCode = 1): void {
+  send({ type: "workflow-error", runId, error, exitCode });
 }
 
 function sendWorkflowMessage(runId: string, message: WorkflowMessage): void {
@@ -178,7 +178,7 @@ async function executeRole(
     result = await role(start, messages);
   } catch (e: unknown) {
     const errMsg = e instanceof Error ? e.message : String(e);
-    sendThreadEvent(runId, "failed", { error: errMsg });
+    sendThreadEvent(runId, "failed", { error: errMsg, exitCode: 1 });
     return null;
   }
 
@@ -186,10 +186,13 @@ async function executeRole(
   return result;
 }
 
+type KillFlag = { value: boolean };
+
 async function runThread(
   def: WorkflowDefinition<RoleMeta>,
   runId: string,
   maxRounds: number,
+  killFlag: KillFlag,
   resumeMessages: WorkflowMessage[] = [],
   freshPrompt: string | null = null,
   dryRun = false,
@@ -219,15 +222,26 @@ async function runThread(
     });
   }
 
+  if (killFlag.value) {
+    sendThreadEvent(runId, "killed", { exitCode: 137 });
+    return;
+  }
+
   let nextRole = def.moderator({ start, steps });
 
   if (nextRole === END) {
-    sendThreadEvent(runId, "completed", null);
+    sendThreadEvent(runId, "completed", { exitCode: 0 });
     return;
   }
 
   while (steps.length < maxRounds) {
     const result = await executeRole(def, nextRole, start, roleMessages, runId);
+
+    if (killFlag.value) {
+      sendThreadEvent(runId, "killed", { exitCode: 137 });
+      return;
+    }
+
     if (result === null) return;
 
     const message: WorkflowMessage = {
@@ -249,12 +263,12 @@ async function runThread(
     nextRole = def.moderator({ start, steps });
 
     if (nextRole === END) {
-      sendThreadEvent(runId, "completed", null);
+      sendThreadEvent(runId, "completed", { exitCode: 0 });
       return;
     }
   }
 
-  sendWorkflowError(runId, `Thread exceeded maximum rounds (${maxRounds})`);
+  sendWorkflowError(runId, `Thread exceeded maximum rounds (${maxRounds})`, 2);
 }
 
 // ---------------------------------------------------------------------------
@@ -309,6 +323,7 @@ function handleMessage(
   raw: unknown,
   def: WorkflowDefinition<RoleMeta>,
   inFlight: Map<string, Promise<void>>,
+  killFlags: Map<string, KillFlag>,
   shuttingDown: { value: boolean },
 ): void {
   const parseResult = parseParentMessage(raw);
@@ -332,15 +347,19 @@ function handleMessage(
     if (shuttingDown.value) return;
     const { runId, prompt, maxRounds, dryRun } = msg;
 
+    const killFlag: KillFlag = { value: false };
+    killFlags.set(runId, killFlag);
+
     const previous = inFlight.get(runId) ?? Promise.resolve();
     const next = previous
-      .then(() => runThread(def, runId, maxRounds, [], prompt, dryRun))
+      .then(() => runThread(def, runId, maxRounds, killFlag, [], prompt, dryRun))
       .catch((e: unknown) => {
         const errMsg = e instanceof Error ? e.message : String(e);
         sendWorkflowError(runId, errMsg);
       })
       .finally(() => {
         inFlight.delete(runId);
+        killFlags.delete(runId);
       });
 
     inFlight.set(runId, next);
@@ -351,20 +370,32 @@ function handleMessage(
     if (shuttingDown.value) return;
     const { runId, messages, maxRounds, dryRun } = msg;
 
+    const killFlag: KillFlag = { value: false };
+    killFlags.set(runId, killFlag);
+
     const previous = inFlight.get(runId) ?? Promise.resolve();
     const next = previous
-      .then(() => runThread(def, runId, maxRounds, messages, null, dryRun))
+      .then(() => runThread(def, runId, maxRounds, killFlag, messages, null, dryRun))
       .catch((e: unknown) => {
         const errMsg = e instanceof Error ? e.message : String(e);
         sendWorkflowError(runId, errMsg);
       })
       .finally(() => {
         inFlight.delete(runId);
+        killFlags.delete(runId);
       });
 
     inFlight.set(runId, next);
     return;
   }
+
+  if (msg.type === "kill-thread") {
+    const flag = killFlags.get(msg.runId);
+    if (flag !== undefined) {
+      flag.value = true;
+    }
+    return;
+  }
 }
 
 // ---------------------------------------------------------------------------
@@ -382,12 +413,13 @@ async function bootstrap(nerveRoot: string, workflowName: string): Promise<void>
   }
 
   const inFlight = new Map<string, Promise<void>>();
+  const killFlags = new Map<string, KillFlag>();
   const shuttingDown = { value: false };
 
   sendReady();
 
   process.on("message", (raw: unknown) => {
-    handleMessage(raw, def, inFlight, shuttingDown);
+    handleMessage(raw, def, inFlight, killFlags, shuttingDown);
   });
 }
 

packages/khala/migrations/0001_initial.sql

@@ -0,0 +1,47 @@
+-- Agent registry
+CREATE TABLE agents (
+  id TEXT PRIMARY KEY,
+  token_hash TEXT NOT NULL,
+  created_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+-- Workflow threads
+CREATE TABLE threads (
+  id TEXT PRIMARY KEY,
+  workflow TEXT NOT NULL,
+  status TEXT NOT NULL DEFAULT 'active',
+  initiator TEXT NOT NULL,
+  result TEXT,
+  created_at TEXT NOT NULL DEFAULT (datetime('now')),
+  updated_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+-- Thread messages (append-only)
+CREATE TABLE messages (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  thread_id TEXT NOT NULL REFERENCES threads (id),
+  role TEXT NOT NULL,
+  content TEXT NOT NULL,
+  meta TEXT,
+  step INTEGER NOT NULL,
+  agent_id TEXT,
+  created_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+CREATE INDEX idx_messages_thread ON messages (thread_id, step);
+
+-- Task queue
+CREATE TABLE tasks (
+  id TEXT PRIMARY KEY,
+  thread_id TEXT NOT NULL REFERENCES threads (id),
+  role TEXT NOT NULL,
+  instruction TEXT NOT NULL,
+  status TEXT NOT NULL DEFAULT 'open',
+  claim_id TEXT,
+  claimed_by TEXT,
+  claimed_at TEXT,
+  timeout_seconds INTEGER NOT NULL DEFAULT 300,
+  created_at TEXT NOT NULL DEFAULT (datetime('now'))
+);
+
+CREATE INDEX idx_tasks_status ON tasks (status, created_at);

packages/khala/package.json

@@ -0,0 +1,25 @@
+{
+  "name": "@uncaged/khala",
+  "version": "0.0.1",
+  "private": true,
+  "type": "module",
+  "scripts": {
+    "dev": "wrangler dev",
+    "deploy": "wrangler deploy",
+    "test": "vitest run",
+    "check": "biome check ."
+  },
+  "dependencies": {
+    "@uncaged/nerve-core": "workspace:*",
+    "hono": "^4.7.0",
+    "jsonata": "^2.0.5",
+    "ulidx": "^2.4.1"
+  },
+  "devDependencies": {
+    "@cloudflare/vitest-pool-workers": "^0.14.0",
+    "@cloudflare/workers-types": "^4.20250410.0",
+    "typescript": "^5.5.0",
+    "vitest": "^4.1.5",
+    "wrangler": "^4.14.0"
+  }
+}

packages/khala/src/__tests__/result.test.ts

@@ -0,0 +1,16 @@
+import { err, ok } from "@uncaged/nerve-core";
+import { describe, expect, it } from "vitest";
+
+describe("Result", () => {
+  it("ok wraps a value", () => {
+    const r = ok(42);
+    expect(r.ok).toBe(true);
+    if (r.ok) expect(r.value).toBe(42);
+  });
+
+  it("err wraps an error", () => {
+    const r = err("boom");
+    expect(r.ok).toBe(false);
+    if (!r.ok) expect(r.error).toBe("boom");
+  });
+});

packages/khala/src/auth.ts

@@ -0,0 +1,49 @@
+import { createMiddleware } from "hono/factory";
+import type { KhalaBindings } from "./env.js";
+
+const encoder = new TextEncoder();
+
+function hexDigest(buf: ArrayBuffer): string {
+  return [...new Uint8Array(buf)].map((b) => b.toString(16).padStart(2, "0")).join("");
+}
+
+export async function sha256Hex(input: string): Promise<string> {
+  const data = encoder.encode(input);
+  const dig = await crypto.subtle.digest("SHA-256", data);
+  return hexDigest(dig);
+}
+
+export const agentAuth = createMiddleware<{
+  Bindings: KhalaBindings;
+  Variables: { agentId: string };
+}>(async (c, next) => {
+  const header = c.req.header("Authorization");
+  if (header === undefined || !header.startsWith("Bearer ")) {
+    return c.json({ error: "missing token" }, 401);
+  }
+  const token = header.slice(7);
+  const hash = await sha256Hex(token);
+  const agent = await c.env.DB.prepare("SELECT id FROM agents WHERE token_hash = ?")
+    .bind(hash)
+    .first<{ id: string }>();
+  if (agent === null) {
+    return c.json({ error: "invalid token" }, 401);
+  }
+  c.set("agentId", agent.id);
+  return next();
+});
+
+export function requireAdmin(c: {
+  req: { header: (k: string) => string | undefined };
+  env: KhalaBindings;
+}): boolean {
+  const header = c.req.header("Authorization");
+  const expected = c.env.ADMIN_SECRET;
+  if (expected.length === 0) {
+    return false;
+  }
+  if (header === undefined || !header.startsWith("Bearer ")) {
+    return false;
+  }
+  return header.slice(7) === expected;
+}

packages/khala/src/db.ts

@@ -0,0 +1,388 @@
+import type { Result } from "@uncaged/nerve-core";
+import { err, ok } from "@uncaged/nerve-core";
+import { type ULID, monotonicFactory } from "ulidx";
+import type { Agent, GetThreadMessagesOpts, Message, Task, Thread } from "./types.js";
+
+const generateUlid = monotonicFactory((): number => Date.now());
+
+export function newId(): ULID {
+  return generateUlid();
+}
+
+export async function createThread(
+  db: D1Database,
+  workflow: string,
+  initiator: string,
+): Promise<Thread> {
+  const id = newId();
+  const now = new Date().toISOString();
+  await db
+    .prepare(
+      `INSERT INTO threads (id, workflow, status, initiator, result, created_at, updated_at)
+       VALUES (?, ?, 'active', ?, NULL, datetime('now'), datetime('now'))`,
+    )
+    .bind(id, workflow, initiator)
+    .run();
+  return {
+    id,
+    workflow,
+    status: "active",
+    initiator,
+    result: null,
+    created_at: now,
+    updated_at: now,
+  };
+}
+
+export async function getThread(db: D1Database, id: string): Promise<Thread | null> {
+  const row = await db.prepare("SELECT * FROM threads WHERE id = ?").bind(id).first<ThreadRow>();
+  if (!row) return null;
+  return rowToThread(row);
+}
+
+type ThreadRow = {
+  id: string;
+  workflow: string;
+  status: string;
+  initiator: string;
+  result: string | null;
+  created_at: string;
+  updated_at: string;
+};
+
+function rowToThread(row: ThreadRow): Thread {
+  const status = row.status === "completed" || row.status === "failed" ? row.status : "active";
+  return {
+    id: row.id,
+    workflow: row.workflow,
+    status,
+    initiator: row.initiator,
+    result: row.result,
+    created_at: row.created_at,
+    updated_at: row.updated_at,
+  };
+}
+
+export async function setThreadResult(
+  db: D1Database,
+  id: string,
+  result: string,
+  status: "completed" | "failed",
+): Promise<void> {
+  await db
+    .prepare(`UPDATE threads SET result = ?, status = ?, updated_at = datetime('now') WHERE id = ?`)
+    .bind(result, status, id)
+    .run();
+}
+
+export async function failThread(db: D1Database, id: string, result: string): Promise<void> {
+  await setThreadResult(db, id, result, "failed");
+}
+
+export async function appendMessage(
+  db: D1Database,
+  threadId: string,
+  role: string,
+  content: string,
+  meta: string | null,
+  step: number,
+  agentId: string | null,
+): Promise<Message> {
+  const row = await db
+    .prepare(
+      `INSERT INTO messages (thread_id, role, content, meta, step, agent_id, created_at)
+       VALUES (?, ?, ?, ?, ?, ?, datetime('now'))
+       RETURNING *`,
+    )
+    .bind(threadId, role, content, meta, step, agentId)
+    .first<MessageRow>();
+  if (!row) {
+    throw new Error("appendMessage: INSERT RETURNING returned no row");
+  }
+  return rowToMessage(row);
+}
+
+type MessageRow = {
+  id: number;
+  thread_id: string;
+  role: string;
+  content: string;
+  meta: string | null;
+  step: number;
+  agent_id: string | null;
+  created_at: string;
+};
+
+function rowToMessage(row: MessageRow): Message {
+  return {
+    id: row.id,
+    thread_id: row.thread_id,
+    role: row.role,
+    content: row.content,
+    meta: row.meta,
+    step: row.step,
+    agent_id: row.agent_id,
+    created_at: row.created_at,
+  };
+}
+
+export async function getThreadMessages(
+  db: D1Database,
+  threadId: string,
+  opts: GetThreadMessagesOpts | null,
+): Promise<Message[]> {
+  const o = opts ?? { role: null, minStep: null, maxStep: null, limit: null };
+  const maxRows = o.limit === null ? 10_000 : Math.min(10_000, Math.max(1, o.limit));
+  let sql = "SELECT * FROM messages WHERE thread_id = ?";
+  const binds: (string | number)[] = [threadId];
+  if (o.role !== null) {
+    sql += " AND role = ?";
+    binds.push(o.role);
+  }
+  if (o.minStep !== null) {
+    sql += " AND step >= ?";
+    binds.push(o.minStep);
+  }
+  if (o.maxStep !== null) {
+    sql += " AND step <= ?";
+    binds.push(o.maxStep);
+  }
+  sql += " ORDER BY step ASC, id ASC LIMIT ?";
+  binds.push(maxRows);
+  const st = await db
+    .prepare(sql)
+    .bind(...binds)
+    .all<MessageRow>();
+  const list = (st.results as MessageRow[] | null) ?? [];
+  return list.map(rowToMessage);
+}
+
+export async function getMaxMessageStep(db: D1Database, threadId: string): Promise<number> {
+  const row = await db
+    .prepare("SELECT MAX(step) as m FROM messages WHERE thread_id = ?")
+    .bind(threadId)
+    .first<{ m: number | null }>();
+  if (!row || row.m === null) return -1;
+  return row.m;
+}
+
+export async function createTask(
+  db: D1Database,
+  threadId: string,
+  role: string,
+  instruction: string,
+  timeoutSeconds: number,
+): Promise<Task> {
+  const id = newId();
+  const now = new Date().toISOString();
+  const to = Math.max(1, Math.floor(timeoutSeconds));
+  await db
+    .prepare(
+      `INSERT INTO tasks (id, thread_id, role, instruction, status, claim_id, claimed_by, claimed_at, timeout_seconds, created_at)
+       VALUES (?, ?, ?, ?, 'open', NULL, NULL, NULL, ?, datetime('now'))`,
+    )
+    .bind(id, threadId, role, instruction, to)
+    .run();
+  return {
+    id,
+    thread_id: threadId,
+    role,
+    instruction,
+    status: "open",
+    claim_id: null,
+    claimed_by: null,
+    claimed_at: null,
+    timeout_seconds: to,
+    created_at: now,
+  };
+}
+
+type TaskRow = {
+  id: string;
+  thread_id: string;
+  role: string;
+  instruction: string;
+  status: string;
+  claim_id: string | null;
+  claimed_by: string | null;
+  claimed_at: string | null;
+  timeout_seconds: number;
+  created_at: string;
+};
+
+function rowToTask(row: TaskRow): Task {
+  const status =
+    row.status === "open" ||
+    row.status === "claimed" ||
+    row.status === "completed" ||
+    row.status === "expired"
+      ? row.status
+      : "open";
+  return {
+    id: row.id,
+    thread_id: row.thread_id,
+    role: row.role,
+    instruction: row.instruction,
+    status,
+    claim_id: row.claim_id,
+    claimed_by: row.claimed_by,
+    claimed_at: row.claimed_at,
+    timeout_seconds: row.timeout_seconds,
+    created_at: row.created_at,
+  };
+}
+
+export type ClaimResult = { ok: true; claimId: string } | { ok: false };
+
+export async function claimTask(
+  db: D1Database,
+  taskId: string,
+  agentId: string,
+): Promise<ClaimResult> {
+  const claimId = newId();
+  const res = await db
+    .prepare(
+      `UPDATE tasks
+       SET status = 'claimed', claim_id = ?, claimed_by = ?, claimed_at = datetime('now')
+       WHERE id = ? AND status = 'open'`,
+    )
+    .bind(claimId, agentId, taskId)
+    .run();
+  const changes = res.meta.changes;
+  if (typeof changes === "number" && changes > 0) {
+    return { ok: true, claimId };
+  }
+  return { ok: false };
+}
+
+export async function getTaskById(db: D1Database, id: string): Promise<Task | null> {
+  const row = await db.prepare("SELECT * FROM tasks WHERE id = ?").bind(id).first<TaskRow>();
+  if (!row) return null;
+  return rowToTask(row);
+}
+
+export async function completeTask(
+  db: D1Database,
+  taskId: string,
+  claimId: string,
+): Promise<boolean> {
+  const res = await db
+    .prepare(
+      `UPDATE tasks SET status = 'completed' WHERE id = ? AND claim_id = ? AND status = 'claimed'`,
+    )
+    .bind(taskId, claimId)
+    .run();
+  const changes = res.meta.changes;
+  if (typeof changes === "number") return changes > 0;
+  return res.success === true;
+}
+
+export async function releaseTask(
+  db: D1Database,
+  taskId: string,
+  claimId: string,
+): Promise<boolean> {
+  const res = await db
+    .prepare(
+      `UPDATE tasks
+       SET status = 'open', claim_id = NULL, claimed_by = NULL, claimed_at = NULL
+       WHERE id = ? AND claim_id = ? AND status = 'claimed'`,
+    )
+    .bind(taskId, claimId)
+    .run();
+  const changes = res.meta.changes;
+  if (typeof changes === "number") return changes > 0;
+  return res.success === true;
+}
+
+export async function expireTimedOutTasks(db: D1Database): Promise<number> {
+  const res = await db
+    .prepare(
+      `UPDATE tasks
+       SET status = 'open', claim_id = NULL, claimed_by = NULL, claimed_at = NULL
+       WHERE status = 'claimed'
+         AND claimed_at IS NOT NULL
+         AND (strftime('%s', 'now') - strftime('%s', claimed_at)) > timeout_seconds`,
+    )
+    .run();
+  const changes = res.meta.changes;
+  if (typeof changes === "number") return changes;
+  return 0;
+}
+
+export async function getOpenTasks(
+  db: D1Database,
+  limit: number,
+  workflow: string | null,
+): Promise<Task[]> {
+  const cap = Math.min(500, Math.max(1, limit));
+  if (workflow === null) {
+    const st = await db
+      .prepare(`SELECT t.* FROM tasks t WHERE t.status = 'open' ORDER BY t.created_at ASC LIMIT ?`)
+      .bind(cap)
+      .all<TaskRow>();
+    const results = (st.results as TaskRow[] | null) ?? [];
+    return results.map(rowToTask);
+  }
+  const st = await db
+    .prepare(
+      `SELECT t.* FROM tasks t
+       INNER JOIN threads th ON t.thread_id = th.id
+       WHERE t.status = 'open' AND th.workflow = ?
+       ORDER BY t.created_at ASC LIMIT ?`,
+    )
+    .bind(workflow, cap)
+    .all<TaskRow>();
+  const results = (st.results as TaskRow[] | null) ?? [];
+  return results.map(rowToTask);
+}
+
+export async function registerAgent(
+  db: D1Database,
+  id: string,
+  tokenHash: string,
+): Promise<Result<Agent, string>> {
+  const existing = await db.prepare("SELECT id FROM agents WHERE id = ?").bind(id).first<{
+    id: string;
+  }>();
+  if (existing) {
+    return err(`agent already exists: ${id}`);
+  }
+  await db
+    .prepare("INSERT INTO agents (id, token_hash, created_at) VALUES (?, ?, datetime('now'))")
+    .bind(id, tokenHash)
+    .run();
+  const row = await db.prepare("SELECT * FROM agents WHERE id = ?").bind(id).first<Agent>();
+  if (!row) return err("registerAgent: insert failed");
+  return ok(row);
+}
+
+export async function listAgents(
+  db: D1Database,
+): Promise<ReadonlyArray<{ id: string; created_at: string }>> {
+  const st = await db
+    .prepare("SELECT id, created_at FROM agents ORDER BY id ASC")
+    .all<{ id: string; created_at: string }>();
+  return (st.results as { id: string; created_at: string }[] | null) ?? [];
+}
+
+export async function deleteAgent(db: D1Database, id: string): Promise<boolean> {
+  const res = await db.prepare("DELETE FROM agents WHERE id = ?").bind(id).run();
+  const changes = res.meta.changes;
+  if (typeof changes === "number") return changes > 0;
+  return false;
+}
+
+export async function getTaskForThreadAndClaim(
+  db: D1Database,
+  threadId: string,
+  taskId: string,
+  claimId: string,
+): Promise<Task | null> {
+  const row = await db
+    .prepare("SELECT * FROM tasks WHERE id = ? AND thread_id = ? AND claim_id = ?")
+    .bind(taskId, threadId, claimId)
+    .first<TaskRow>();
+  if (!row) return null;
+  return rowToTask(row);
+}

packages/khala/src/env.ts

@@ -0,0 +1,9 @@
+/**
+ * Cloudflare bindings for the Khala worker. Mirrored in worker-configuration.d.ts (ProvidedEnv).
+ */
+export type KhalaBindings = {
+  DB: D1Database;
+  THREAD: DurableObjectNamespace;
+  ADMIN_SECRET: string;
+  TEST_MIGRATIONS: { id: string; name: string; queries: { sql: string }[] }[] | undefined;
+};

packages/khala/src/index.ts

@@ -0,0 +1,149 @@
+import { Hono } from "hono";
+import { agentAuth } from "./auth.js";
+import {
+  claimTask,
+  createThread,
+  expireTimedOutTasks,
+  getOpenTasks,
+  getTaskById,
+  getThread,
+  releaseTask,
+} from "./db.js";
+import type { KhalaBindings } from "./env.js";
+import { admin } from "./routes/admin.js";
+import { ThreadDO } from "./thread-do.js";
+import { getWorkflow } from "./workflows.js";
+
+// Register built-in workflows
+import "./workflows/ping-pong.js";
+
+type HonoEnv = { Bindings: KhalaBindings; Variables: { agentId: string } };
+
+const app = new Hono<HonoEnv>();
+
+// Health
+app.get("/health", (c) => c.json({ ok: true }));
+
+// Admin routes
+app.route("/admin", admin);
+
+// --- Agent-authenticated routes ---
+
+// Create workflow thread
+app.post("/workflows/:name/threads", agentAuth, async (c) => {
+  const name = c.req.param("name");
+  const def = getWorkflow(name);
+  if (def === null) {
+    return c.json({ error: `workflow not found: ${name}` }, 404);
+  }
+  const thread = await createThread(c.env.DB, name, c.get("agentId"));
+  const doId = c.env.THREAD.idFromName(thread.id);
+  const stub = c.env.THREAD.get(doId);
+  const doReq = new Request("https://do/start", {
+    method: "POST",
+    headers: { "X-Thread-Id": thread.id },
+  });
+  const doRes = await stub.fetch(doReq);
+  if (!doRes.ok) {
+    const body = await doRes.text();
+    return c.json({ error: `failed to start thread: ${body}` }, 500);
+  }
+  return c.json({ threadId: thread.id, workflow: name, status: "active" }, 201);
+});
+
+// Get thread status
+app.get("/threads/:id", agentAuth, async (c) => {
+  const thread = await getThread(c.env.DB, c.req.param("id"));
+  if (thread === null) {
+    return c.json({ error: "not found" }, 404);
+  }
+  return c.json(thread);
+});
+
+// Get thread messages
+app.get("/threads/:id/messages", agentAuth, async (c) => {
+  const doId = c.env.THREAD.idFromName(c.req.param("id"));
+  const stub = c.env.THREAD.get(doId);
+  const url = new URL(c.req.url);
+  const doReq = new Request(`https://do/messages${url.search}`, {
+    method: "GET",
+    headers: { "X-Thread-Id": c.req.param("id") },
+  });
+  return stub.fetch(doReq);
+});
+
+// Post response to thread
+app.post("/threads/:id/response", agentAuth, async (c) => {
+  const threadId = c.req.param("id");
+  const body = await c.req.text();
+  const doId = c.env.THREAD.idFromName(threadId);
+  const stub = c.env.THREAD.get(doId);
+  const doReq = new Request("https://do/response", {
+    method: "POST",
+    headers: { "X-Thread-Id": threadId, "Content-Type": "application/json" },
+    body,
+  });
+  return stub.fetch(doReq);
+});
+
+// --- Task Queue ---
+
+// List open tasks
+app.get("/tasks", agentAuth, async (c) => {
+  const workflow = c.req.query("workflow") ?? null;
+  const limit = Number.parseInt(c.req.query("limit") ?? "50", 10);
+  const tasks = await getOpenTasks(c.env.DB, Number.isFinite(limit) ? limit : 50, workflow);
+  return c.json({ tasks });
+});
+
+// Claim a task
+// TODO(#132): Add max_concurrent_claims per agent to prevent greedy claiming
+app.post("/tasks/:id/claim", agentAuth, async (c) => {
+  const taskId = c.req.param("id");
+  const agentId = c.get("agentId");
+  const result = await claimTask(c.env.DB, taskId, agentId);
+  if (!result.ok) {
+    return c.json({ error: "task not available" }, 409);
+  }
+  const task = await getTaskById(c.env.DB, taskId);
+  if (task === null) {
+    return c.json({ error: "task not found" }, 404);
+  }
+  return c.json({
+    claimId: result.claimId,
+    taskId: task.id,
+    threadId: task.thread_id,
+    role: task.role,
+    instruction: task.instruction,
+  });
+});
+
+// Release a task
+app.post("/tasks/:id/release", agentAuth, async (c) => {
+  const taskId = c.req.param("id");
+  const raw: unknown = await c.req.json();
+  const body = raw as Readonly<Record<string, unknown>>;
+  const claimId = body.claimId;
+  if (typeof claimId !== "string") {
+    return c.json({ error: "claimId required" }, 400);
+  }
+  const ok = await releaseTask(c.env.DB, taskId, claimId);
+  if (!ok) {
+    return c.json({ error: "release failed" }, 409);
+  }
+  return c.json({ ok: true });
+});
+
+// --- Cron: expire timed-out tasks ---
+
+export default {
+  fetch: app.fetch,
+  async scheduled(_event: ScheduledEvent, env: KhalaBindings, _ctx: ExecutionContext) {
+    const expired = await expireTimedOutTasks(env.DB);
+    if (expired > 0) {
+      console.log(`khala: expired ${expired} timed-out task(s)`);
+    }
+  },
+};
+
+export { ThreadDO };

packages/khala/src/moderator.ts

@@ -0,0 +1,37 @@
+import type { Result } from "@uncaged/nerve-core";
+import { err, ok } from "@uncaged/nerve-core";
+import jsonata from "jsonata";
+
+export type ModeratorDecision = { role: string };
+
+export async function evaluateModerator(
+  expression: string,
+  context: Readonly<Record<string, unknown>>,
+): Promise<Result<ModeratorDecision, string>> {
+  let expr: ReturnType<typeof jsonata>;
+  try {
+    expr = jsonata(expression);
+  } catch (e) {
+    const msg = e instanceof Error ? e.message : String(e);
+    return err(`jsonata compile: ${msg}`);
+  }
+  let out: unknown;
+  try {
+    out = await expr.evaluate(context);
+  } catch (e) {
+    const msg = e instanceof Error ? e.message : String(e);
+    return err(`jsonata eval: ${msg}`);
+  }
+  if (out === null || out === undefined) {
+    return err("moderator returned empty");
+  }
+  if (typeof out !== "object" || out === null || Array.isArray(out)) {
+    return err("moderator must return an object");
+  }
+  const rec = out as Readonly<Record<string, unknown>>;
+  const role = rec.role;
+  if (typeof role === "string") {
+    return ok({ role });
+  }
+  return err("moderator result missing string role");
+}

packages/khala/src/routes/admin.ts

@@ -0,0 +1,67 @@
+import type { Result } from "@uncaged/nerve-core";
+import { Hono } from "hono";
+import { requireAdmin, sha256Hex } from "../auth.js";
+import { deleteAgent, listAgents, registerAgent } from "../db.js";
+import type { KhalaBindings } from "../env.js";
+
+const admin = new Hono<{ Bindings: KhalaBindings }>();
+
+function jsonErr(res: Result<unknown, string>, status: number) {
+  if (res.ok) {
+    return new Response("unexpected", { status: 500 });
+  }
+  return Response.json({ error: String(res.error) }, { status });
+}
+
+admin.get("/agents", async (c) => {
+  if (!requireAdmin(c)) {
+    return c.json({ error: "unauthorized" }, 401);
+  }
+  const agents = await listAgents(c.env.DB);
+  return c.json({ agents: [...agents] });
+});
+
+// TODO(#132): Add rate limiting for agent registration endpoint
+admin.post("/agents", async (c) => {
+  if (!requireAdmin(c)) {
+    return c.json({ error: "unauthorized" }, 401);
+  }
+  const raw: unknown = await c.req.json();
+  if (raw === null || typeof raw !== "object" || Array.isArray(raw)) {
+    return c.json({ error: "invalid body" }, 400);
+  }
+  const body = raw as Readonly<Record<string, unknown>>;
+  const id = body.id;
+  const token = body.token;
+  if (
+    typeof id !== "string" ||
+    id.length === 0 ||
+    typeof token !== "string" ||
+    token.length === 0
+  ) {
+    return c.json({ error: "id and token required" }, 400);
+  }
+  const hash = await sha256Hex(token);
+  const res = await registerAgent(c.env.DB, id, hash);
+  if (!res.ok) {
+    if (res.error.includes("already exists")) {
+      return jsonErr(res, 409);
+    }
+    return jsonErr(res, 400);
+  }
+  return c.json({ id: res.value.id, created_at: res.value.created_at });
+});
+
+admin.delete("/agents/:id", async (c) => {
+  if (!requireAdmin(c)) {
+    return c.json({ error: "unauthorized" }, 401);
+  }
+  const id = c.req.param("id");
+  const ok = await deleteAgent(c.env.DB, id);
+  if (!ok) {
+    return c.json({ error: "not found" }, 404);
+  }
+  return c.json({ ok: true });
+});
+
+export { admin };

packages/khala/src/thread-do.ts

@@ -0,0 +1,213 @@
+import {
+  appendMessage,
+  completeTask,
+  createTask,
+  getMaxMessageStep,
+  getTaskForThreadAndClaim,
+  getThread,
+  getThreadMessages,
+  setThreadResult,
+} from "./db.js";
+import type { KhalaBindings } from "./env.js";
+import { evaluateModerator } from "./moderator.js";
+import type { CloudWorkflowDef } from "./workflows.js";
+import { getWorkflow } from "./workflows.js";
+
+type StepRow = { role: string; content: string; meta: unknown };
+
+function asRecord(value: unknown): Readonly<Record<string, unknown>> {
+  if (value !== null && typeof value === "object" && !Array.isArray(value)) {
+    return value as Readonly<Record<string, unknown>>;
+  }
+  return {};
+}
+
+function normalizeMetaString(value: unknown): string | null {
+  if (value === null || value === undefined) return null;
+  if (typeof value === "string") return value;
+  return JSON.stringify(value);
+}
+
+function buildSteps(msgs: Awaited<ReturnType<typeof getThreadMessages>>): StepRow[] {
+  return msgs
+    .filter((m) => m.role !== "__moderator__")
+    .map((m) => ({
+      role: m.role,
+      content: m.content,
+      meta: m.meta
+        ? (() => {
+            try {
+              return JSON.parse(m.meta) as unknown;
+            } catch {
+              return m.meta;
+            }
+          })()
+        : null,
+    }));
+}
+
+/**
+ * Durable object for one workflow thread. Relies on D1 (messages, tasks, threads) as source of truth.
+ */
+export class ThreadDO {
+  private readonly env: KhalaBindings;
+
+  constructor(_state: DurableObjectState, env: KhalaBindings) {
+    this.env = env;
+  }
+
+  private async runModeratorAndSchedule(
+    db: D1Database,
+    def: CloudWorkflowDef,
+    threadId: string,
+  ): Promise<{ ok: true } | { ok: false; error: string }> {
+    const messages = await getThreadMessages(db, threadId, null);
+    const steps = buildSteps(messages);
+    const mod = await evaluateModerator(def.moderator, { steps: [...steps] });
+    if (!mod.ok) {
+      return { ok: false, error: mod.error };
+    }
+    if (mod.value.role === "__end__") {
+      await setThreadResult(
+        db,
+        threadId,
+        JSON.stringify({ completed: true, stepCount: steps.length }),
+        "completed",
+      );
+      return { ok: true };
+    }
+    const role = mod.value.role;
+    const r = def.roles[role] ?? null;
+    if (r === null) {
+      await setThreadResult(
+        db,
+        threadId,
+        JSON.stringify({ error: `unknown role: ${role}` }),
+        "failed",
+      );
+      return { ok: true };
+    }
+    await createTask(db, threadId, role, r.prompt, r.timeoutSeconds ?? 300);
+    return { ok: true };
+  }
+
+  private async startThread(threadId: string): Promise<Response> {
+    const db = this.env.DB;
+    const thread = await getThread(db, threadId);
+    if (thread === null) {
+      return Response.json({ error: "thread not found" }, { status: 404 });
+    }
+    if (thread.status !== "active") {
+      return Response.json({ error: "thread not active" }, { status: 400 });
+    }
+    const def = getWorkflow(thread.workflow);
+    if (def === null) {
+      return Response.json({ error: "workflow not registered" }, { status: 500 });
+    }
+    const r = await this.runModeratorAndSchedule(db, def, threadId);
+    if (!r.ok) {
+      await setThreadResult(db, threadId, JSON.stringify({ error: r.error }), "failed");
+      return Response.json({ error: r.error }, { status: 500 });
+    }
+    return Response.json({ ok: true });
+  }
+
+  private async postResponse(
+    threadId: string,
+    body: Readonly<Record<string, unknown>>,
+  ): Promise<Response> {
+    const taskId = body.taskId;
+    const content = body.content;
+    const claimId = body.claimId;
+    const agentId = body.agentId;
+    if (
+      typeof taskId !== "string" ||
+      typeof content !== "string" ||
+      typeof claimId !== "string" ||
+      typeof agentId !== "string"
+    ) {
+      return Response.json(
+        { error: "taskId, content, claimId, agentId required" },
+        { status: 400 },
+      );
+    }
+    const db = this.env.DB;
+    const task = await getTaskForThreadAndClaim(db, threadId, taskId, claimId);
+    if (task === null) {
+      return Response.json({ error: "task or claim not found" }, { status: 404 });
+    }
+    if (task.claimed_by !== agentId) {
+      return Response.json({ error: "agent mismatch" }, { status: 403 });
+    }
+    if (task.status !== "claimed") {
+      return Response.json({ error: "task not claimable" }, { status: 400 });
+    }
+    const thread = await getThread(db, threadId);
+    if (thread === null) {
+      return Response.json({ error: "thread not found" }, { status: 404 });
+    }
+    const def = getWorkflow(thread.workflow);
+    if (def === null) {
+      return Response.json({ error: "workflow not registered" }, { status: 500 });
+    }
+    const nextStep = (await getMaxMessageStep(db, threadId)) + 1;
+    const meta = Object.hasOwn(body, "meta")
+      ? normalizeMetaString((body as Readonly<Record<string, unknown>>).meta)
+      : null;
+    await appendMessage(db, threadId, task.role, content, meta, nextStep, agentId);
+    const done = await completeTask(db, taskId, claimId);
+    if (!done) {
+      return Response.json({ error: "could not complete task" }, { status: 409 });
+    }
+    const r = await this.runModeratorAndSchedule(db, def, threadId);
+    if (!r.ok) {
+      await setThreadResult(db, threadId, JSON.stringify({ error: r.error }), "failed");
+      return Response.json({ error: r.error }, { status: 500 });
+    }
+    return Response.json({ ok: true });
+  }
+
+  private async listMessages(url: URL, threadId: string): Promise<Response> {
+    const roleP = url.searchParams.get("role");
+    const minStepRaw = url.searchParams.get("minStep");
+    const maxStepRaw = url.searchParams.get("maxStep");
+    const limitRaw = url.searchParams.get("limit");
+    const toInt = (s: string | null): number | null => {
+      if (s === null) return null;
+      const n = Number.parseInt(s, 10);
+      return Number.isFinite(n) ? n : null;
+    };
+    const list = await getThreadMessages(this.env.DB, threadId, {
+      role: roleP,
+      minStep: toInt(minStepRaw),
+      maxStep: toInt(maxStepRaw),
+      limit: toInt(limitRaw),
+    });
+    return Response.json({ messages: list });
+  }
+
+  async fetch(request: Request): Promise<Response> {
+    const url = new URL(request.url);
+    const path = url.pathname;
+    const threadId = request.headers.get("X-Thread-Id");
+    if (threadId === null || threadId.length === 0) {
+      return Response.json({ error: "X-Thread-Id required" }, { status: 400 });
+    }
+    if (path === "/start" && request.method === "POST") {
+      return this.startThread(threadId);
+    }
+    if (path === "/response" && request.method === "POST") {
+      const text = await request.text();
+      let body: Readonly<Record<string, unknown>> = {};
+      if (text.length > 0) {
+        const parsed: unknown = JSON.parse(text) as unknown;
+        body = asRecord(parsed);
+      }
+      return this.postResponse(threadId, body);
+    }
+    if (path === "/messages" && request.method === "GET") {
+      return this.listMessages(url, threadId);
+    }
+    return Response.json({ error: "not found" }, { status: 404 });
+  }
+}

packages/khala/src/types.ts

@@ -0,0 +1,46 @@
+export type Agent = {
+  id: string;
+  token_hash: string;
+  created_at: string;
+};
+
+export type Thread = {
+  id: string;
+  workflow: string;
+  status: "active" | "completed" | "failed";
+  initiator: string;
+  result: string | null;
+  created_at: string;
+  updated_at: string;
+};
+
+export type Message = {
+  id: number;
+  thread_id: string;
+  role: string;
+  content: string;
+  meta: string | null;
+  step: number;
+  agent_id: string | null;
+  created_at: string;
+};
+
+export type Task = {
+  id: string;
+  thread_id: string;
+  role: string;
+  instruction: string;
+  status: "open" | "claimed" | "completed" | "expired";
+  claim_id: string | null;
+  claimed_by: string | null;
+  claimed_at: string | null;
+  timeout_seconds: number;
+  created_at: string;
+};
+
+export type GetThreadMessagesOpts = {
+  role: string | null;
+  minStep: number | null;
+  maxStep: number | null;
+  limit: number | null;
+};

packages/khala/src/workflows.ts

@@ -0,0 +1,22 @@
+export type CloudRole = {
+  prompt: string;
+  /** Turn timeout in seconds. Defaults to 300. */
+  timeoutSeconds: number | null;
+};
+
+export type CloudWorkflowDef = {
+  name: string;
+  roles: Readonly<Record<string, CloudRole>>;
+  /** JSONata expression: context includes `steps` (role turn messages). */
+  moderator: string;
+};
+
+const registry = new Map<string, CloudWorkflowDef>();
+
+export function registerWorkflow(def: CloudWorkflowDef): void {
+  registry.set(def.name, def);
+}
+
+export function getWorkflow(name: string): CloudWorkflowDef | null {
+  return registry.get(name) ?? null;
+}

packages/khala/src/workflows/ping-pong.ts

@@ -0,0 +1,14 @@
+import { type CloudWorkflowDef, registerWorkflow } from "../workflows.js";
+
+const pingPong: CloudWorkflowDef = {
+  name: "ping-pong",
+  roles: {
+    pinger: { prompt: "Send the literal word ping", timeoutSeconds: null },
+    ponger: { prompt: "Send the literal word pong", timeoutSeconds: null },
+  },
+  moderator: `$count(steps) >= 6 ? { "role": "__end__" } : $count(steps) % 2 = 0 ? { "role": "pinger" } : { "role": "ponger" }`,
+};
+
+registerWorkflow(pingPong);
+
+export { pingPong };

packages/khala/tsconfig.json

@@ -0,0 +1,13 @@
+{
+  "compilerOptions": {
+    "target": "ES2022",
+    "lib": ["ES2022", "WebWorker"],
+    "module": "ES2022",
+    "moduleResolution": "bundler",
+    "strict": true,
+    "noUncheckedIndexedAccess": true,
+    "skipLibCheck": true,
+    "types": ["@cloudflare/workers-types", "./worker-configuration.d.ts"]
+  },
+  "include": ["src", "worker-configuration.d.ts", "vitest.config.ts"]
+}

packages/khala/worker-configuration.d.ts

@@ -0,0 +1,15 @@
+import type { D1Migration } from "@cloudflare/workers-types";
+
+declare module "cloudflare:workers" {
+  type KhalaEnv = {
+    DB: D1Database;
+    THREAD: DurableObjectNamespace;
+    /** Used by Vitest: migrations to apply in setup (see d1-apply.ts). */
+    TEST_MIGRATIONS: D1Migration[] | undefined;
+  };
+  interface Env extends KhalaEnv {
+    /** Plain-text admin bearer value (set via [vars] or wrangler secret). */
+    ADMIN_SECRET: string;
+  }
+  interface ProvidedEnv extends Env {}
+}

packages/khala/wrangler.toml

@@ -0,0 +1,24 @@
+name = "khala"
+main = "src/index.ts"
+compatibility_date = "2025-04-01"
+
+[vars]
+ADMIN_SECRET = "dev-admin-replace-in-production"
+
+[[d1_databases]]
+binding = "DB"
+database_name = "khala"
+database_id = "00000000-0000-0000-0000-000000000000"
+migrations_dir = "migrations"
+
+[[migrations]]
+tag = "v1"
+new_classes = [ "ThreadDO" ]
+
+[durable_objects]
+bindings = [
+  { name = "THREAD", class_name = "ThreadDO" }
+]
+
+[triggers]
+crons = [ "* * * * *" ]

packages/store/src/__tests__/log-store-crash-recovery.test.ts

@@ -41,7 +41,7 @@ describe("LogStore — crash recovery helpers (Phase 3)", () => {
           payload: JSON.stringify({ triggerPayload: payload }),
           timestamp: 1000,
         },
-        { runId: "run-1", workflow: "my-wf", status: "started", timestamp: 1000 },
+        { runId: "run-1", workflow: "my-wf", status: "started", timestamp: 1000, exitCode: null },
       );
 
       const result = store.getTriggerPayload("run-1");
@@ -57,7 +57,7 @@ describe("LogStore — crash recovery helpers (Phase 3)", () => {
           payload: null,
           timestamp: 1000,
         },
-        { runId: "run-2", workflow: "my-wf", status: "started", timestamp: 1000 },
+        { runId: "run-2", workflow: "my-wf", status: "started", timestamp: 1000, exitCode: null },
       );
 
       expect(store.getTriggerPayload("run-2")).toBeNull();
@@ -148,7 +148,7 @@ describe("LogStore — crash recovery helpers (Phase 3)", () => {
           payload: JSON.stringify({ triggerPayload: {} }),
           timestamp: 1000,
         },
-        { runId: "run-6", workflow: "my-wf", status: "started", timestamp: 1000 },
+        { runId: "run-6", workflow: "my-wf", status: "started", timestamp: 1000, exitCode: null },
       );
       store.append({
         source: "workflow",

packages/store/src/__tests__/log-store-workflow.test.ts

@@ -31,6 +31,7 @@ describe("LogStore — workflow_runs", () => {
         workflow: "cleanup",
         status: "started",
         timestamp: 1000,
+        exitCode: null,
       };
 
       const entry = store.upsertWorkflowRun(
@@ -53,12 +54,18 @@ describe("LogStore — workflow_runs", () => {
     it("updates existing workflow_runs row on upsert (status transition)", () => {
       store.upsertWorkflowRun(
         { source: "workflow", type: "started", refId: "run-2", payload: null, timestamp: 1000 },
-        { runId: "run-2", workflow: "cleanup", status: "started", timestamp: 1000 },
+        { runId: "run-2", workflow: "cleanup", status: "started", timestamp: 1000, exitCode: null },
       );
 
       store.upsertWorkflowRun(
         { source: "workflow", type: "completed", refId: "run-2", payload: null, timestamp: 2000 },
-        { runId: "run-2", workflow: "cleanup", status: "completed", timestamp: 2000 },
+        {
+          runId: "run-2",
+          workflow: "cleanup",
+          status: "completed",
+          timestamp: 2000,
+          exitCode: null,
+        },
       );
 
       const stored = store.getWorkflowRun("run-2");
@@ -79,7 +86,7 @@ describe("LogStore — workflow_runs", () => {
       ] as const) {
         store.upsertWorkflowRun(
           { source: "workflow", type, refId: "run-3", payload: null, timestamp },
-          { runId: "run-3", workflow: "cleanup", status, timestamp },
+          { runId: "run-3", workflow: "cleanup", status, timestamp, exitCode: null },
         );
       }
 
@@ -98,11 +105,23 @@ describe("LogStore — workflow_runs", () => {
     it("returns the latest state after multiple upserts", () => {
       store.upsertWorkflowRun(
         { source: "workflow", type: "queued", refId: "run-4", payload: null, timestamp: 100 },
-        { runId: "run-4", workflow: "code-review", status: "queued", timestamp: 100 },
+        {
+          runId: "run-4",
+          workflow: "code-review",
+          status: "queued",
+          timestamp: 100,
+          exitCode: null,
+        },
       );
       store.upsertWorkflowRun(
         { source: "workflow", type: "started", refId: "run-4", payload: null, timestamp: 200 },
-        { runId: "run-4", workflow: "code-review", status: "started", timestamp: 200 },
+        {
+          runId: "run-4",
+          workflow: "code-review",
+          status: "started",
+          timestamp: 200,
+          exitCode: null,
+        },
       );
 
       const run = store.getWorkflowRun("run-4");
@@ -115,19 +134,19 @@ describe("LogStore — workflow_runs", () => {
     beforeEach(() => {
       store.upsertWorkflowRun(
         { source: "workflow", type: "queued", refId: "r1", payload: null, timestamp: 100 },
-        { runId: "r1", workflow: "cleanup", status: "queued", timestamp: 100 },
+        { runId: "r1", workflow: "cleanup", status: "queued", timestamp: 100, exitCode: null },
       );
       store.upsertWorkflowRun(
         { source: "workflow", type: "started", refId: "r2", payload: null, timestamp: 200 },
-        { runId: "r2", workflow: "cleanup", status: "started", timestamp: 200 },
+        { runId: "r2", workflow: "cleanup", status: "started", timestamp: 200, exitCode: null },
       );
       store.upsertWorkflowRun(
         { source: "workflow", type: "completed", refId: "r3", payload: null, timestamp: 300 },
-        { runId: "r3", workflow: "cleanup", status: "completed", timestamp: 300 },
+        { runId: "r3", workflow: "cleanup", status: "completed", timestamp: 300, exitCode: null },
       );
       store.upsertWorkflowRun(
         { source: "workflow", type: "failed", refId: "r4", payload: null, timestamp: 400 },
-        { runId: "r4", workflow: "deploy", status: "queued", timestamp: 400 },
+        { runId: "r4", workflow: "deploy", status: "queued", timestamp: 400, exitCode: null },
       );
     });
 
@@ -171,13 +190,13 @@ describe("LogStore — workflow_runs", () => {
   });
 
   describe("all statuses are storable", () => {
-    it.each(["queued", "started", "completed", "failed", "crashed", "dropped"] as const)(
+    it.each(["queued", "started", "completed", "failed", "crashed", "dropped", "killed"] as const)(
       "stores status=%s",
       (status) => {
         const runId = `run-${status}`;
         store.upsertWorkflowRun(
           { source: "workflow", type: status, refId: runId, payload: null, timestamp: 1 },
-          { runId, workflow: "test", status, timestamp: 1 },
+          { runId, workflow: "test", status, timestamp: 1, exitCode: null },
         );
         expect(store.getWorkflowRun(runId)?.status).toBe(status);
       },

packages/store/src/log-store.ts

@@ -58,7 +58,8 @@ export type WorkflowRunStatus =
   | "failed"
   | "crashed"
   | "dropped"
-  | "interrupted";
+  | "interrupted"
+  | "killed";
 
 const VALID_WORKFLOW_STATUSES = new Set<string>([
   "queued",
@@ -68,6 +69,7 @@ const VALID_WORKFLOW_STATUSES = new Set<string>([
   "crashed",
   "dropped",
   "interrupted",
+  "killed",
 ]);
 
 function isWorkflowRunStatus(value: string): value is WorkflowRunStatus {
@@ -87,6 +89,7 @@ export type WorkflowRun = {
   workflow: string;
   status: WorkflowRunStatus;
   timestamp: number;
+  exitCode: number | null;
 };
 
 /** One role-produced workflow-message row with 1-based round index (ROW_NUMBER over role messages only). */
@@ -192,10 +195,11 @@ CREATE TABLE IF NOT EXISTS meta (
 );
 
 CREATE TABLE IF NOT EXISTS workflow_runs (
-  run_id   TEXT PRIMARY KEY,
-  workflow TEXT NOT NULL,
-  status   TEXT NOT NULL,
-  timestamp INTEGER NOT NULL
+  run_id    TEXT PRIMARY KEY,
+  workflow  TEXT NOT NULL,
+  status    TEXT NOT NULL,
+  timestamp INTEGER NOT NULL,
+  exit_code INTEGER
 );
 
 CREATE INDEX IF NOT EXISTS idx_workflow_runs_status ON workflow_runs(status);
@@ -332,6 +336,13 @@ export function createLogStore(dbPath: string): LogStore {
   sqlite.exec("PRAGMA journal_mode=WAL");
   sqlite.exec(SCHEMA_SQL);
 
+  // Migration: add exit_code column for existing databases
+  try {
+    sqlite.exec("ALTER TABLE workflow_runs ADD COLUMN exit_code INTEGER");
+  } catch {
+    // Column already exists — safe to ignore
+  }
+
   const insertStmt = sqlite.prepare(
     "INSERT INTO logs (source, type, ref_id, payload, timestamp) VALUES (@source, @type, @refId, @payload, @timestamp)",
   );
@@ -342,11 +353,11 @@ export function createLogStore(dbPath: string): LogStore {
   );
 
   const upsertWorkflowRunStmt = sqlite.prepare(
-    "INSERT OR REPLACE INTO workflow_runs (run_id, workflow, status, timestamp) VALUES (@runId, @workflow, @status, @timestamp)",
+    "INSERT OR REPLACE INTO workflow_runs (run_id, workflow, status, timestamp, exit_code) VALUES (@runId, @workflow, @status, @timestamp, @exitCode)",
   );
 
   const getWorkflowRunStmt = sqlite.prepare(
-    "SELECT run_id, workflow, status, timestamp FROM workflow_runs WHERE run_id = ?",
+    "SELECT run_id, workflow, status, timestamp, exit_code FROM workflow_runs WHERE run_id = ?",
   );
 
   const getTriggerPayloadStmt = sqlite.prepare(
@@ -386,19 +397,19 @@ export function createLogStore(dbPath: string): LogStore {
   );
 
   const getActiveWorkflowRunsStmt = sqlite.prepare(
-    "SELECT run_id, workflow, status, timestamp FROM workflow_runs WHERE status IN ('queued', 'started') ORDER BY timestamp ASC",
+    "SELECT run_id, workflow, status, timestamp, exit_code FROM workflow_runs WHERE status IN ('queued', 'started') ORDER BY timestamp ASC",
   );
 
   const getActiveWorkflowRunsByNameStmt = sqlite.prepare(
-    "SELECT run_id, workflow, status, timestamp FROM workflow_runs WHERE status IN ('queued', 'started') AND workflow = ? ORDER BY timestamp ASC",
+    "SELECT run_id, workflow, status, timestamp, exit_code FROM workflow_runs WHERE status IN ('queued', 'started') AND workflow = ? ORDER BY timestamp ASC",
   );
 
   const getAllWorkflowRunsStmt = sqlite.prepare(
-    "SELECT run_id, workflow, status, timestamp FROM workflow_runs ORDER BY timestamp DESC",
+    "SELECT run_id, workflow, status, timestamp, exit_code FROM workflow_runs ORDER BY timestamp DESC",
   );
 
   const getAllWorkflowRunsByNameStmt = sqlite.prepare(
-    "SELECT run_id, workflow, status, timestamp FROM workflow_runs WHERE workflow = ? ORDER BY timestamp DESC",
+    "SELECT run_id, workflow, status, timestamp, exit_code FROM workflow_runs WHERE workflow = ? ORDER BY timestamp DESC",
   );
 
   const minLogTsStmt = sqlite.prepare("SELECT MIN(timestamp) AS m FROM logs");
@@ -423,6 +434,7 @@ export function createLogStore(dbPath: string): LogStore {
         workflow: run.workflow,
         status: run.status,
         timestamp: run.timestamp,
+        exitCode: run.exitCode,
       });
       return { ...entry, id: Number(info.lastInsertRowid) };
     });
@@ -504,31 +516,37 @@ export function createLogStore(dbPath: string): LogStore {
     return upsertWorkflowRunTx(entry, run);
   }
 
-  function getWorkflowRun(runId: string): WorkflowRun | null {
-    const row = getWorkflowRunStmt.get(runId) as
-      | { run_id: string; workflow: string; status: string; timestamp: number }
-      | undefined;
-    if (row === undefined) return null;
+  type SqlWorkflowRunRow = {
+    run_id: string;
+    workflow: string;
+    status: string;
+    timestamp: number;
+    exit_code: number | null;
+  };
+
+  function mapWorkflowRunRow(r: SqlWorkflowRunRow): WorkflowRun {
     return {
-      runId: row.run_id,
-      workflow: row.workflow,
-      status: validateWorkflowRunStatus(row.status),
-      timestamp: row.timestamp,
+      runId: r.run_id,
+      workflow: r.workflow,
+      status: validateWorkflowRunStatus(r.status),
+      timestamp: r.timestamp,
+      exitCode: r.exit_code ?? null,
     };
   }
 
+  function getWorkflowRun(runId: string): WorkflowRun | null {
+    const row = getWorkflowRunStmt.get(runId) as SqlWorkflowRunRow | undefined;
+    if (row === undefined) return null;
+    return mapWorkflowRunRow(row);
+  }
+
   function getActiveWorkflowRuns(workflowName?: string): WorkflowRun[] {
     const rows = (
       workflowName !== undefined
         ? getActiveWorkflowRunsByNameStmt.all(workflowName)
         : getActiveWorkflowRunsStmt.all()
-    ) as Array<{ run_id: string; workflow: string; status: string; timestamp: number }>;
-    return rows.map((r) => ({
-      runId: r.run_id,
-      workflow: r.workflow,
-      status: validateWorkflowRunStatus(r.status),
-      timestamp: r.timestamp,
-    }));
+    ) as SqlWorkflowRunRow[];
+    return rows.map(mapWorkflowRunRow);
   }
 
   function getAllWorkflowRuns(workflowName: string | null): WorkflowRun[] {
@@ -536,13 +554,8 @@ export function createLogStore(dbPath: string): LogStore {
       workflowName !== null
         ? getAllWorkflowRunsByNameStmt.all(workflowName)
         : getAllWorkflowRunsStmt.all()
-    ) as Array<{ run_id: string; workflow: string; status: string; timestamp: number }>;
-    return rows.map((r) => ({
-      runId: r.run_id,
-      workflow: r.workflow,
-      status: validateWorkflowRunStatus(r.status),
-      timestamp: r.timestamp,
-    }));
+    ) as SqlWorkflowRunRow[];
+    return rows.map(mapWorkflowRunRow);
   }
 
   function getTriggerPayload(runId: string): unknown {

packages/workflow-utils/src/__tests__/llm-extract.test.ts

@@ -108,7 +108,7 @@ describe("llmExtract", () => {
     expect(result.error.kind).toBe("schema_validation_failed");
   });
 
-  it("dryRun skips fetch and returns an empty stub value", async () => {
+  it("dryRun skips fetch and returns schema-shaped stub values", async () => {
     const fetchMock = vi.fn();
     vi.stubGlobal("fetch", fetchMock);
 
@@ -125,6 +125,6 @@ describe("llmExtract", () => {
     if (!result.ok) {
       return;
     }
-    expect(result.value).toEqual({});
+    expect(result.value).toEqual({ n: 0 });
   });
 });

packages/workflow-utils/src/__tests__/schema-defaults.test.ts

@@ -0,0 +1,59 @@
+import { describe, expect, it } from "vitest";
+import { z } from "zod";
+
+import { schemaDefaults } from "../schema-defaults.js";
+
+describe("schemaDefaults", () => {
+  it("fills nested objects with primitive placeholders", () => {
+    const schema = z.object({
+      meta: z.object({
+        id: z.string(),
+        count: z.number(),
+        flag: z.boolean(),
+      }),
+    });
+    expect(schemaDefaults(schema)).toEqual({
+      meta: { id: "", count: 0, flag: false },
+    });
+  });
+
+  it("uses empty arrays for array fields", () => {
+    const schema = z.object({
+      roles: z.array(z.object({ name: z.string(), level: z.number() })),
+    });
+    const out = schemaDefaults(schema) as { roles: { name: string; level: number }[] };
+    expect(out.roles).toEqual([]);
+    expect(out.roles.map((r) => r.name)).toEqual([]);
+  });
+
+  it("uses the first enum value", () => {
+    const schema = z.object({
+      status: z.enum(["pending", "done", "failed"]),
+      code: z.nativeEnum({ A: 1, B: 2 }),
+    });
+    expect(schemaDefaults(schema)).toEqual({
+      status: "pending",
+      code: 1,
+    });
+  });
+
+  it("sets optional fields to undefined and omits exactOptional keys", () => {
+    const schema = z.object({
+      req: z.string(),
+      maybe: z.string().optional(),
+      exact: z.string().exactOptional(),
+    });
+    expect(schemaDefaults(schema)).toEqual({
+      req: "",
+      maybe: undefined,
+    });
+    expect(Object.keys(schemaDefaults(schema) as object).includes("exact")).toBe(false);
+  });
+
+  it("respects .default()", () => {
+    const schema = z.object({
+      n: z.number().default(42),
+    });
+    expect(schemaDefaults(schema)).toEqual({ n: 42 });
+  });
+});

packages/workflow-utils/src/index.ts

@@ -11,6 +11,7 @@ export {
   type LlmExtractOptions,
   type LlmProvider,
 } from "./llm-extract.js";
+export { schemaDefaults } from "./schema-defaults.js";
 export {
   nerveCommandEnv,
   spawnSafe,

packages/workflow-utils/src/llm-extract.ts

@@ -1,6 +1,8 @@
 import { type Result, err, ok } from "@uncaged/nerve-core";
 import { toJSONSchema, type z } from "zod";
 
+import { schemaDefaults } from "./schema-defaults.js";
+
 export type LlmProvider = {
   baseUrl: string;
   apiKey: string;
@@ -102,7 +104,7 @@ export async function llmExtract<T>(
 ): Promise<Result<T, LlmError>> {
   const dryRun = resolveLlmExtractDryRun(options);
   if (dryRun) {
-    return ok({} as T);
+    return ok(schemaDefaults(options.schema) as T);
   }
 
   const rawJsonSchema = toJSONSchema(options.schema) as Record<string, unknown>;

packages/workflow-utils/src/schema-defaults.ts

@@ -0,0 +1,190 @@
+import type { z } from "zod";
+
+type ZodTypeAny = z.ZodType;
+
+type Def = Record<string, unknown> & { type: string };
+type TypeHandler = (schema: ZodTypeAny, def: Def) => unknown;
+
+function isPlainObject(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null && !Array.isArray(value);
+}
+
+function isZodExactOptional(s: ZodTypeAny): boolean {
+  return s.constructor.name === "ZodExactOptional";
+}
+
+function resolveDefaultValue(defaultValue: unknown | (() => unknown)): unknown {
+  if (typeof defaultValue === "function") {
+    return (defaultValue as () => unknown)();
+  }
+  return defaultValue;
+}
+
+function mergeIntersection(left: unknown, right: unknown): unknown {
+  if (isPlainObject(left) && isPlainObject(right)) {
+    return { ...left, ...right };
+  }
+  return right;
+}
+
+function defaultsForObject(_schema: ZodTypeAny, def: Def): unknown {
+  const shape = def.shape as Record<string, ZodTypeAny> | undefined;
+  if (shape === undefined) {
+    return {};
+  }
+  const out: Record<string, unknown> = {};
+  for (const key of Object.keys(shape)) {
+    const child = shape[key];
+    const cdef = child.def as { type: string };
+    if (cdef.type === "optional") {
+      if (isZodExactOptional(child)) {
+        continue;
+      }
+      out[key] = undefined;
+    } else {
+      out[key] = schemaDefaultsInner(child);
+    }
+  }
+  return out;
+}
+
+function firstUnionOption(_schema: ZodTypeAny, def: Def): unknown {
+  const options = def.options as readonly ZodTypeAny[] | undefined;
+  if (options === undefined || options.length === 0) {
+    return null;
+  }
+  return schemaDefaultsInner(options[0]);
+}
+
+function defaultsFromNullable(_schema: ZodTypeAny, _def: Def): unknown {
+  return null;
+}
+
+function defaultsFromInner(_schema: ZodTypeAny, def: Def): unknown {
+  const inner = def.innerType as ZodTypeAny | undefined;
+  if (inner === undefined) {
+    return null;
+  }
+  return schemaDefaultsInner(inner);
+}
+
+function defaultsForPipe(_schema: ZodTypeAny, def: Def): unknown {
+  const out = def.out as ZodTypeAny | undefined;
+  if (out === undefined) {
+    return null;
+  }
+  return schemaDefaultsInner(out);
+}
+
+function defaultsForIntersection(_schema: ZodTypeAny, def: Def): unknown {
+  const left = def.left as ZodTypeAny | undefined;
+  const right = def.right as ZodTypeAny | undefined;
+  if (left === undefined || right === undefined) {
+    return null;
+  }
+  return mergeIntersection(schemaDefaultsInner(left), schemaDefaultsInner(right));
+}
+
+function defaultsForTuple(_schema: ZodTypeAny, def: Def): unknown {
+  const items = def.items as readonly ZodTypeAny[] | undefined;
+  if (items === undefined) {
+    return [];
+  }
+  return items.map((item) => schemaDefaultsInner(item));
+}
+
+function defaultsForLazy(schema: ZodTypeAny, def: Def): unknown {
+  const inner =
+    (schema as { _zod?: { innerType?: ZodTypeAny } })._zod?.innerType ??
+    (def.getter as (() => ZodTypeAny) | undefined)?.();
+  if (inner === undefined) {
+    return null;
+  }
+  return schemaDefaultsInner(inner);
+}
+
+function defaultsForPromise(_schema: ZodTypeAny, def: Def): unknown {
+  const inner = def.innerType as ZodTypeAny | undefined;
+  if (inner === undefined) {
+    return Promise.resolve(null);
+  }
+  return Promise.resolve(schemaDefaultsInner(inner));
+}
+
+function firstEnumValue(_schema: ZodTypeAny, def: Def): unknown {
+  const entries = def.entries as Record<string, string | number> | undefined;
+  if (entries === undefined) {
+    return null;
+  }
+  const values = Object.values(entries);
+  return values[0] ?? null;
+}
+
+function firstLiteralValue(_schema: ZodTypeAny, def: Def): unknown {
+  const values = def.values as unknown[] | undefined;
+  if (values === undefined || values.length === 0) {
+    return null;
+  }
+  return values[0];
+}
+
+const TYPE_HANDLERS: Record<string, TypeHandler> = {
+  string: () => "",
+  number: () => 0,
+  boolean: () => false,
+  bigint: () => 0n,
+  date: () => new Date(0),
+  symbol: () => Symbol(),
+  undefined: () => undefined,
+  null: () => null,
+  void: () => undefined,
+  any: () => null,
+  unknown: () => null,
+  never: () => undefined,
+  nan: () => Number.NaN,
+  array: () => [],
+  object: defaultsForObject,
+  record: () => ({}),
+  map: () => new Map(),
+  set: () => new Set(),
+  enum: firstEnumValue,
+  literal: firstLiteralValue,
+  optional: () => undefined,
+  nullable: defaultsFromNullable,
+  default: (_s, def) => resolveDefaultValue(def.defaultValue as unknown | (() => unknown)),
+  prefault: (_s, def) => resolveDefaultValue(def.defaultValue as unknown | (() => unknown)),
+  nonoptional: defaultsFromInner,
+  catch: defaultsFromInner,
+  success: () => false,
+  readonly: defaultsFromInner,
+  union: firstUnionOption,
+  xor: firstUnionOption,
+  intersection: defaultsForIntersection,
+  pipe: defaultsForPipe,
+  transform: () => null,
+  tuple: defaultsForTuple,
+  lazy: defaultsForLazy,
+  promise: defaultsForPromise,
+  file: () => new File([], ""),
+  function: () => null,
+  custom: () => null,
+  template_literal: () => "",
+};
+
+/**
+ * Produces a structurally valid placeholder that mirrors primitive/array/object
+ * shape for a Zod schema. Used for `llmExtract` dry runs so downstream code
+ * (e.g. `.roles.map`) does not throw on `undefined` fields.
+ */
+export function schemaDefaults(schema: z.ZodType): unknown {
+  return schemaDefaultsInner(schema as ZodTypeAny);
+}
+
+function schemaDefaultsInner(schema: ZodTypeAny): unknown {
+  const def = schema.def as Def;
+  const run = TYPE_HANDLERS[def.type];
+  if (run === undefined) {
+    return null;
+  }
+  return run(schema, def);
+}