feat(khala): complete MVP — CF Worker with D1, DO, auth, task queue, moderator
Implements Khala cloud workflow orchestrator (Phase 0-4): - Project scaffolding: Hono + Wrangler + D1 + DO - D1 schema: agents, threads, messages, tasks tables - Data access layer with atomic claim/release - Agent auth (SHA-256 Bearer token) + admin API - ThreadDO workflow engine with JSONata moderator - Task queue API: poll/claim/release - Cron-based timeout sweep - Ping-pong demo workflow Closes #124, closes #125, closes #127, closes #128, closes #129
This commit is contained in:
@@ -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);
|
||||
@@ -0,0 +1,24 @@
|
||||
{
|
||||
"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": {
|
||||
"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"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
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;
|
||||
}
|
||||
@@ -0,0 +1,389 @@
|
||||
import { monotonicFactory, type ULID } from "ulidx";
|
||||
import type { Agent, GetThreadMessagesOpts, Message, Task, Thread } from "./types.js";
|
||||
import type { Result } from "./result.js";
|
||||
import { err, ok } from "./result.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> {
|
||||
await db
|
||||
.prepare(
|
||||
`INSERT INTO messages (thread_id, role, content, meta, step, agent_id, created_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, datetime('now'))`,
|
||||
)
|
||||
.bind(threadId, role, content, meta, step, agentId)
|
||||
.run();
|
||||
const row = await db
|
||||
.prepare(
|
||||
`SELECT * FROM messages WHERE thread_id = ? AND step = ? ORDER BY id DESC LIMIT 1`,
|
||||
)
|
||||
.bind(threadId, step)
|
||||
.first<MessageRow>();
|
||||
if (!row) {
|
||||
throw new Error("appendMessage: row not found after insert");
|
||||
}
|
||||
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);
|
||||
}
|
||||
@@ -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;
|
||||
};
|
||||
@@ -0,0 +1,149 @@
|
||||
import { Hono } from "hono";
|
||||
import type { KhalaBindings } from "./env.js";
|
||||
import { admin } from "./routes/admin.js";
|
||||
import { agentAuth } from "./auth.js";
|
||||
import { ThreadDO } from "./thread-do.js";
|
||||
import {
|
||||
createThread,
|
||||
getThread,
|
||||
getThreadMessages,
|
||||
getOpenTasks,
|
||||
claimTask,
|
||||
releaseTask,
|
||||
getTaskById,
|
||||
expireTimedOutTasks,
|
||||
} from "./db.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
|
||||
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 };
|
||||
@@ -0,0 +1,37 @@
|
||||
import jsonata from "jsonata";
|
||||
import type { Result } from "./result.js";
|
||||
import { err, ok } from "./result.js";
|
||||
|
||||
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");
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
export type Result<T, E = string> = { ok: true; value: T } | { ok: false; error: E };
|
||||
|
||||
export function ok<T>(value: T): Result<T, never> {
|
||||
return { ok: true, value };
|
||||
}
|
||||
|
||||
export function err<E = string>(error: E): Result<never, E> {
|
||||
return { ok: false, error };
|
||||
}
|
||||
@@ -0,0 +1,61 @@
|
||||
import { Hono } from "hono";
|
||||
import { requireAdmin, sha256Hex } from "../auth.js";
|
||||
import { registerAgent, deleteAgent, listAgents } from "../db.js";
|
||||
import type { Result } from "../result.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] });
|
||||
});
|
||||
|
||||
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 };
|
||||
@@ -0,0 +1,206 @@
|
||||
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, 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 });
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
};
|
||||
@@ -0,0 +1,20 @@
|
||||
export type CloudRole = {
|
||||
prompt: string;
|
||||
};
|
||||
|
||||
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;
|
||||
}
|
||||
@@ -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" },
|
||||
ponger: { prompt: "Send the literal word pong" },
|
||||
},
|
||||
moderator: `$count(steps) >= 6 ? { "role": "__end__" } : $count(steps) % 2 = 0 ? { "role": "pinger" } : { "role": "ponger" }`,
|
||||
};
|
||||
|
||||
registerWorkflow(pingPong);
|
||||
|
||||
export { pingPong };
|
||||
@@ -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"]
|
||||
}
|
||||
+17
@@ -0,0 +1,17 @@
|
||||
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 {}
|
||||
}
|
||||
|
||||
export {};
|
||||
@@ -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 = [ "* * * * *" ]
|
||||
Generated
+1019
-4
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user