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/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/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

@@ -57,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>;
 };
@@ -222,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);
@@ -328,6 +335,24 @@ 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;
@@ -337,6 +362,7 @@ export function createWorkflowManager(
       dequeueNext(workflowName);
       const exitCode = extractExitCode(msg.payload);
       logWorkflowEvent(workflowName, msg.runId, msg.eventType, msg.payload, exitCode);
+      maybeDeferredHotReloadDrain(workflowName);
     }
   }
 
@@ -428,6 +454,7 @@ export function createWorkflowManager(
 
     state.active.clear();
     workers.delete(workflowName);
+    pendingDrains.delete(workflowName);
 
     if (stopped || workflowConfig(workflowName) === null) return;
 
@@ -484,6 +511,7 @@ export function createWorkflowManager(
         dequeueNext(workflowName);
       }
       logWorkflowEvent(workflowName, msg.runId, "failed", { error: msg.error }, msg.exitCode);
+      maybeDeferredHotReloadDrain(workflowName);
       return;
     }
 
@@ -681,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);
@@ -701,6 +758,7 @@ export function createWorkflowManager(
     totalActiveCount,
     updateConfig,
     drainAndRespawn,
+    drainWhenIdle,
     stop,
   };
 }

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/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);
+}