diff --git a/daemon/README.md b/daemon/README.md index e2066d8..89d9a68 100644 --- a/daemon/README.md +++ b/daemon/README.md @@ -81,6 +81,30 @@ app prefix are test-only static overrides and are ignored unless `DAEMON_TEST_MO production uses client credentials. Optional settings are `PORT`, `BIND_ADDR`, `REPLAY_WINDOW_MS`, `LINEAR_GRAPHQL_URL`, and `LINEAR_TOKEN_URL`. +### Doozy work provider + +`WORK_PROVIDER` selects the work source and defaults to `linear`. Set it to `doozy` to +poll Doozy todos instead of accepting Linear webhooks. A Doozy run requires +`DOOZY_API_URL`, `DOOZY_API_KEY`, and at least one selector: `DOOZY_AGENT_ID` for todos +assigned to a specific agent, or `DOOZY_TAG` for an API tag or a bracketed title tag such +as `[daemon-agent]`. +`DOOZY_POLL_INTERVAL_MS` defaults to 5000 and `DOOZY_POLL_LIMIT` defaults to one new +claim per poll. + +The poller claims a matching `ready` todo by changing it to `in_progress`, then feeds the +todo through the same durable turn queue, worktree manager, Claude session persistence, +progress queue, terminal delivery, and pull request detection used by Linear. A title or +tag containing `implementer`, `implementation`, or `/do` selects the implementer role; +other todos select the planner role. Progress and pull request links are appended to an +`Orchestra daemon activity` section in the todo body. The terminal result is posted once +to an existing todo-linked chat, or to the activity section when no chat exists, and a +successful todo is marked `done`. Later human messages in an existing chat resume the +stored Claude session after the todo is moved back to `in_progress`. + +Doozy does not currently expose a webhook or streaming ingress shape equivalent to +Linear `AgentSessionEvent`. Polling is the v1 adapter boundary. A future provider can +replace the poller without changing the durable turn worker. + Set `ARTIFACT_TOKEN` to enable artifact hosting. An authenticated `POST /a` creates a bundle with a server-generated id; authenticated `PUT /a/` atomically replaces an existing bundle. Both accept a JSON manifest whose file contents are base64 encoded: diff --git a/daemon/src/cleanup.ts b/daemon/src/cleanup.ts index d2f69ef..76baef5 100644 --- a/daemon/src/cleanup.ts +++ b/daemon/src/cleanup.ts @@ -3,7 +3,7 @@ import type { CleanupJobRow, CleanupNotificationRow, } from "./eventlog.js"; -import type { LinearGateway } from "./linear.js"; +import type { WorkProvider } from "./provider.js"; import { WorktreeManager } from "./worktrees.js"; import { buildInvocationSpan, @@ -38,7 +38,7 @@ export class CleanupWorker { private readonly logger: Logger; constructor( private readonly log: EventLog, - private readonly gateway: LinearGateway, + private readonly gateway: WorkProvider, worktreesRoot: string, targetRepoPath: string, private readonly options: CleanupWorkerOptions = {}, diff --git a/daemon/src/config.ts b/daemon/src/config.ts index 2b09e89..4f5228a 100644 --- a/daemon/src/config.ts +++ b/daemon/src/config.ts @@ -2,6 +2,7 @@ import { dirname } from "node:path"; export type AppName = "planner" | "implementer"; export type HarnessPreference = "claude" | "claudex"; +export type WorkProviderName = "linear" | "doozy"; export interface AppConfig { name: AppName; @@ -22,6 +23,7 @@ function harnessPreference(env: NodeJS.ProcessEnv, name: string): HarnessPrefere } export interface Config { + workProvider?: WorkProviderName; port: number; bindAddr: string; dbPath: string; @@ -71,6 +73,12 @@ export interface Config { attachmentsEnabled: boolean; attachmentHosts: string[]; ntfyUrl?: string; + doozyApiUrl?: string; + doozyApiKey?: string; + doozyAgentId?: string; + doozyTag?: string; + doozyPollIntervalMs?: number; + doozyPollLimit?: number; } function required(env: NodeJS.ProcessEnv, name: string): string { @@ -116,12 +124,14 @@ function stringMap(env: NodeJS.ProcessEnv, name: string): Record return value as Record; } -function appConfig(env: NodeJS.ProcessEnv, name: AppName, testMode: boolean): AppConfig { +function appConfig(env: NodeJS.ProcessEnv, name: AppName, testMode: boolean, provider: WorkProviderName): AppConfig { const prefix = name.toUpperCase(); const staticToken = env[`${prefix}_LINEAR_TOKEN`]?.trim(); const appActorId = env[`${prefix}_APP_ACTOR_ID`]?.trim(); const base = { name, harness: harnessPreference(env, `${prefix}_HARNESS`), - webhookSecret: required(env, `${prefix}_WEBHOOK_SECRET`), ...(appActorId ? { appActorId } : {}) }; + webhookSecret: provider === "linear" ? required(env, `${prefix}_WEBHOOK_SECRET`) : "", + ...(appActorId ? { appActorId } : {}) }; + if (provider === "doozy") return base; if (testMode && staticToken) return { ...base, staticToken }; return { ...base, @@ -132,14 +142,28 @@ function appConfig(env: NodeJS.ProcessEnv, name: AppName, testMode: boolean): Ap export function loadConfig(env: NodeJS.ProcessEnv = process.env): Config { const testMode = env.DAEMON_TEST_MODE === "1"; + const workProviderRaw = env.WORK_PROVIDER?.trim() || "linear"; + if (workProviderRaw !== "linear" && workProviderRaw !== "doozy") + throw new Error("WORK_PROVIDER must be linear or doozy"); + const workProvider: WorkProviderName = workProviderRaw; const dbPath = env.DB_PATH?.trim() || "/var/lib/linear-agent-daemon/events.db"; const sessionsEnabled = enabled(env, "SESSIONS_ENABLED"); const targetRepoPath = env.TARGET_REPO_PATH?.trim(); const linearApiKey = env.LINEAR_API_KEY?.trim(); const artifactToken = env.ARTIFACT_TOKEN?.trim(); - const webhookBaseUrl = env.WEBHOOK_BASE_URL?.trim() || (testMode ? "http://127.0.0.1:8787" : required(env, "WEBHOOK_BASE_URL")); + const webhookBaseUrl = env.WEBHOOK_BASE_URL?.trim() || (testMode || workProvider === "doozy" ? "http://127.0.0.1:8787" : required(env, "WEBHOOK_BASE_URL")); if (sessionsEnabled && !targetRepoPath) required(env, "TARGET_REPO_PATH"); - if (sessionsEnabled && !linearApiKey) required(env, "LINEAR_API_KEY"); + if (sessionsEnabled && workProvider === "linear" && !linearApiKey) required(env, "LINEAR_API_KEY"); + const doozyApiUrl = env.DOOZY_API_URL?.trim()?.replace(/\/+$/, ""); + const doozyApiKey = env.DOOZY_API_KEY?.trim(); + const doozyAgentId = env.DOOZY_AGENT_ID?.trim(); + const doozyTag = env.DOOZY_TAG?.trim(); + if (workProvider === "doozy") { + if (!doozyApiUrl) required(env, "DOOZY_API_URL"); + if (!doozyApiKey) required(env, "DOOZY_API_KEY"); + if (!doozyAgentId && !doozyTag) + throw new Error("DOOZY_AGENT_ID or DOOZY_TAG is required"); + } const claudeArgv = (env.CLAUDE_BIN?.trim() || "claude").split(/\s+/); const claudexArgv = optionalArgv(env, "CLAUDEX_BIN"); const claudexEnv = stringMap(env, "CLAUDEX_ENV"); @@ -161,6 +185,7 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): Config { throw new Error("DO_MAX_BUDGET_USD must be a positive number"); } return { + workProvider, port: positiveInteger(env, "PORT", 8787), bindAddr: env.BIND_ADDR?.trim() || "127.0.0.1", dbPath, @@ -194,7 +219,7 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): Config { reconcileIntervalMs: positiveInteger(env, "RECONCILE_INTERVAL_MS", 60_000), reconcileRequestTimeoutMs: positiveInteger(env, "RECONCILE_REQUEST_TIMEOUT_MS", 10_000), reconcileSessionMaxAgeMs: positiveInteger(env, "RECONCILE_SESSION_MAX_AGE_MS", 6 * 60 * 60_000), - apps: { planner: appConfig(env, "planner", testMode), implementer: appConfig(env, "implementer", testMode) }, + apps: { planner: appConfig(env, "planner", testMode, workProvider), implementer: appConfig(env, "implementer", testMode, workProvider) }, sessionsEnabled, worktreesRoot: env.WORKTREES_ROOT?.trim() || `${dirname(dbPath)}/worktrees`, ...(targetRepoPath ? { targetRepoPath } : {}), @@ -220,5 +245,11 @@ export function loadConfig(env: NodeJS.ProcessEnv = process.env): Config { ...(env.NTFY_URL?.trim() ? { ntfyUrl: env.NTFY_URL.trim() } : {}), attachmentsEnabled: enabled(env, "ATTACHMENTS_ENABLED"), attachmentHosts: (env.ATTACHMENT_HOSTS?.trim() || "uploads.linear.app").split(",").map(host => host.trim()).filter(Boolean), + ...(doozyApiUrl ? { doozyApiUrl } : {}), + ...(doozyApiKey ? { doozyApiKey } : {}), + ...(doozyAgentId ? { doozyAgentId } : {}), + ...(doozyTag ? { doozyTag } : {}), + doozyPollIntervalMs: positiveInteger(env, "DOOZY_POLL_INTERVAL_MS", 5_000), + doozyPollLimit: positiveInteger(env, "DOOZY_POLL_LIMIT", 1), }; } diff --git a/daemon/src/doozy.ts b/daemon/src/doozy.ts new file mode 100644 index 0000000..3bc5677 --- /dev/null +++ b/daemon/src/doozy.ts @@ -0,0 +1,424 @@ +import type { AppName, Config } from "./config.js"; +import type { EventLog, TurnRow } from "./eventlog.js"; +import type { + PostResult, + ProgressContent, + TerminalContent, +} from "./linear.js"; +import type { WorkProvider } from "./provider.js"; + +interface Logger { + log(...args: unknown[]): void; + error(...args: unknown[]): void; +} + +type JsonObject = Record; + +export interface DoozyProviderOptions { + now?: () => number; + logger?: Logger; + onInserted?: () => void; +} + +export class DoozyProvider implements WorkProvider { + private timer?: NodeJS.Timeout; + private polling: Promise | undefined; + private stopped = false; + private readonly now: () => number; + private readonly logger: Logger; + private readonly chatIds = new Map(); + + constructor( + private readonly log: EventLog, + private readonly config: Config, + private readonly options: DoozyProviderOptions = {}, + ) { + this.now = options.now ?? Date.now; + this.logger = options.logger ?? console; + } + + start(): void { + this.stopped = false; + this.timer = setInterval( + () => void this.trigger(), + this.config.doozyPollIntervalMs ?? 5_000, + ); + this.timer.unref(); + void this.trigger(); + } + + trigger(): Promise { + if (this.stopped) return Promise.resolve(); + this.polling ??= this.poll().finally(() => { + this.polling = undefined; + }); + return this.polling; + } + + async stop(): Promise { + this.stopped = true; + if (this.timer) clearInterval(this.timer); + await this.polling; + } + + private async poll(): Promise { + try { + const todos = await this.listTodos("active"); + let claimed = 0; + for (const todo of todos) { + if (!this.eligible(todo)) continue; + const id = string(todo.id); + if (!id) continue; + const status = string(todo.status); + const existing = this.log.getSession(id); + if (!existing && status === "ready") { + if (claimed >= (this.config.doozyPollLimit ?? 1)) continue; + if (!(await this.claim(todo))) continue; + claimed++; + this.appendCreated(todo); + continue; + } + if (existing && status === "in_progress") + await this.appendFollowUps(todo, existing.lastSeenActivityAt ?? 0); + } + } catch (error) { + this.logger.error(JSON.stringify({ + event: "doozy_poll_failed", + error: message(error), + })); + } + } + + private async listTodos(status: string): Promise { + const todos: JsonObject[] = []; + let cursor: string | undefined; + do { + const query = new URLSearchParams({ status, limit: "100" }); + if (this.config.doozyAgentId) query.set("assigneeId", this.config.doozyAgentId); + if (cursor) query.set("cursor", cursor); + const response = await this.request("GET", `/todos?${query}`); + todos.push(...objects(response, "todos")); + cursor = response.hasMore === true ? string(response.nextCursor) : undefined; + } while (cursor); + return todos; + } + + private eligible(todo: JsonObject): boolean { + const agentId = this.config.doozyAgentId; + const tag = this.config.doozyTag?.toLowerCase(); + const assigned = new Set([ + string(todo.assigneeId), + string(todo.agentId), + string(todo.assignedToId), + string(object(todo.assignee)?.id), + string(object(todo.agent)?.id), + ...array(todo.assignees).map((value) => string(object(value)?.id) ?? string(value)), + ].filter((value): value is string => Boolean(value))); + const tags = array(todo.tags).map((value) => + (string(value) ?? string(object(value)?.name) ?? string(object(value)?.slug) ?? "").toLowerCase(), + ); + const title = (string(todo.title) ?? "").toLowerCase(); + return Boolean((agentId && assigned.has(agentId)) + || (tag && (tags.includes(tag) || title.includes(`[${tag}]`)))); + } + + private async claim(todo: JsonObject): Promise { + const id = string(todo.id)!; + try { + const response = await this.request("PATCH", `/todos/${encodeURIComponent(id)}`, { + status: "in_progress", + }); + const updated = object(response.todo) ?? object(response.data) ?? response; + if (string(updated.status) && string(updated.status) !== "in_progress") return false; + this.logger.log(JSON.stringify({ event: "doozy_todo_claimed", todoId: id })); + return true; + } catch (error) { + if (error instanceof DoozyHttpError && (error.status === 409 || error.status === 412)) return false; + throw error; + } + } + + private appendCreated(todo: JsonObject): void { + const id = string(todo.id)!; + const app = this.appFor(todo); + const title = string(todo.title) ?? `Doozy todo ${id}`; + const content = string(todo.content) ?? string(todo.description) ?? ""; + const raw = { + action: "created", + provider: "doozy", + promptContext: [title, content].filter(Boolean).join("\n\n"), + todo, + agentSession: { id, issue: { id, identifier: id } }, + }; + const result = this.log.append({ + deliveryId: `doozy:created:${id}`, + app, + action: "created", + agentSessionId: id, + issueId: id, + issueIdentifier: id, + receivedAt: this.now(), + rawBody: Buffer.from(JSON.stringify(raw)), + }); + this.log.updateLastSeenActivity(id, this.now(), this.now()); + if (result.inserted) this.options.onInserted?.(); + } + + private appFor(todo: JsonObject): AppName { + const labels = [ + ...array(todo.tags).map((value) => string(value) ?? string(object(value)?.name) ?? ""), + string(todo.mode) ?? "", + string(todo.title) ?? "", + ].join(" "); + return /\bimplement(?:er|ation)?\b|\/do\b/i.test(labels) + ? "implementer" + : "planner"; + } + + private async appendFollowUps(todo: JsonObject, since: number): Promise { + const id = string(todo.id)!; + const chatId = this.chatId(id, todo); + if (!chatId) return; + const response = await this.request("GET", `/chats/${encodeURIComponent(chatId)}/messages`); + const messages = objects(response, "messages") + .filter((entry) => this.isHumanMessage(entry)) + .map((entry) => ({ + entry, + id: string(entry.id), + body: string(entry.text) ?? string(entry.content) ?? string(entry.body) ?? string(entry.message), + createdAt: timestamp(entry.createdAt) ?? timestamp(entry.created_at), + })) + .filter((entry): entry is { entry: JsonObject; id: string; body: string; createdAt: number } => + Boolean(entry.id && entry.body && entry.createdAt && entry.createdAt > since)) + .sort((a, b) => a.createdAt - b.createdAt); + for (const item of messages) { + const raw = { + action: "prompted", + provider: "doozy", + agentActivity: { id: item.id, body: item.body, createdAt: new Date(item.createdAt).toISOString() }, + agentSession: { id, issue: { id, identifier: id } }, + }; + const result = this.log.append({ + deliveryId: `doozy:prompt:${id}:${item.id}`, + app: this.appFor(todo), + action: "prompted", + agentSessionId: id, + sourceActivityId: item.id, + issueId: id, + issueIdentifier: id, + receivedAt: this.now(), + rawBody: Buffer.from(JSON.stringify(raw)), + }); + this.log.updateLastSeenActivity(id, item.createdAt, this.now()); + if (result.inserted) this.options.onInserted?.(); + } + } + + private isHumanMessage(entry: JsonObject): boolean { + const role = (string(entry.role) ?? string(entry.senderType) ?? string(object(entry.sender)?.type) ?? "").toLowerCase(); + const body = string(entry.text) ?? string(entry.content) ?? string(entry.body) ?? string(entry.message) ?? ""; + return (role === "user" || role === "human" || role === "member") + && !body.startsWith("[orchestra daemon]") + && !body.startsWith("Help me accomplish this to-do."); + } + + async postAckActivity( + app: AppName, + sessionId: string, + activityId: string, + deadlineAt: number, + ): Promise { + return this.postActivity(app, sessionId, activityId, + { type: "thought", body: "picked up, starting work" }, true, deadlineAt); + } + + async postActivity( + _app: AppName, + sessionId: string, + activityId: string, + content: ProgressContent | TerminalContent, + ephemeral: boolean, + deadlineAt: number, + ): Promise { + try { + if (deadlineAt <= this.now()) return failure("Doozy activity request deadline exceeded", true); + const body = "body" in content ? content.body : `${content.action}: ${content.parameter}`; + const todo = await this.getTodo(sessionId); + if (ephemeral) { + await this.appendTodoNote(sessionId, todo, `Progress: ${body}`, deadlineAt); + } else { + await this.postResult(sessionId, todo, activityId, body, deadlineAt); + } + if (!ephemeral && content.type === "response") + await this.request("PATCH", `/todos/${encodeURIComponent(sessionId)}`, { status: "done" }, deadlineAt); + return { ok: true }; + } catch (error) { + return this.postFailure(error); + } + } + + async setSessionExternalUrl( + _app: AppName, + sessionId: string, + label: string, + url: string, + deadlineAt: number, + ): Promise { + try { + const todo = await this.getTodo(sessionId); + await this.appendTodoNote(sessionId, todo, `${label}: ${url}`, deadlineAt); + return { ok: true }; + } catch (error) { + return this.postFailure(error); + } + } + + turnPrompt(turn: TurnRow, identifier: string, implementer: boolean, resuming: boolean): string { + const payload = parse(turn.rawBody); + if (resuming) { + const activity = object(payload.agentActivity); + return string(activity?.body) ?? "Continue using the latest Doozy todo conversation."; + } + const todo = object(payload.todo); + const title = string(todo?.title) ?? `Doozy todo ${identifier}`; + const content = string(todo?.content) ?? string(todo?.description) ?? string(payload.promptContext) ?? ""; + const role = implementer ? "bloom-implementer" : "bloom-planner"; + const instruction = implementer + ? "Implement the requested change end to end in this repository, verify it, and open a pull request when the work is complete." + : "Discuss, research, and converge on the requested outcome. Use the repository's planning skills when a plan or specification is needed."; + return `You are ${role}, working on Doozy todo ${identifier}. ${instruction}\n\nTitle: ${title}\n\n${content}`; + } + + mcpConfigJson(): string { + return JSON.stringify({ mcpServers: {} }); + } + + private async getTodo(id: string): Promise { + const response = await this.request("GET", `/todos/${encodeURIComponent(id)}`); + return object(response.todo) ?? object(response.data) ?? response; + } + + private chatId(todoId: string, todo: JsonObject): string | undefined { + const cached = this.chatIds.get(todoId); + if (cached) return cached; + const found = string(todo.chatId) ?? string(todo.conversationId) + ?? string(object(todo.chat)?.id) ?? string(object(todo.conversation)?.id); + const latestRunChatId = string(object(todo.latestRun)?.chatId); + const resolved = found ?? latestRunChatId; + if (resolved) { + this.chatIds.set(todoId, resolved); + return resolved; + } + return undefined; + } + + private async postResult( + todoId: string, + todo: JsonObject, + activityId: string, + body: string, + deadlineAt: number, + ): Promise { + const messageBody = `[orchestra daemon]\n${body}`; + const chatId = this.chatId(todoId, todo); + if (chatId) { + await this.request("POST", `/chats/${encodeURIComponent(chatId)}/messages`, { + message: messageBody, + }, deadlineAt, activityId); + return; + } + await this.appendTodoNote(todoId, todo, `Result: ${body}`, deadlineAt); + } + + private async appendTodoNote( + todoId: string, + todo: JsonObject, + note: string, + deadlineAt: number, + ): Promise { + const marker = "\n\n### Orchestra daemon activity\n"; + const current = string(todo.content) ?? ""; + const before = current.includes(marker) ? current.slice(0, current.indexOf(marker)) : current; + const existing = current.includes(marker) ? current.slice(current.indexOf(marker) + marker.length) : ""; + const lines = [...existing.split("\n").filter(Boolean), `- ${note}`].slice(-20); + const activity = lines.join("\n").slice(-2_900); + const content = `${before.slice(0, 12_000)}${marker}${activity}`; + await this.request("PATCH", `/todos/${encodeURIComponent(todoId)}`, { content }, deadlineAt); + } + + private postFailure(error: unknown): PostResult { + if (error instanceof DoozyHttpError) + return failure(error.message, error.status === 408 || error.status === 409 || error.status === 429 || error.status >= 500, + error.retryAfterMs); + return failure(message(error), true); + } + + private async request(method: string, path: string, body?: unknown, deadlineAt = this.now() + 10_000, + idempotencyKey?: string): Promise { + const remaining = deadlineAt - this.now(); + if (remaining <= 0) throw new DoozyHttpError("Doozy request deadline exceeded", 408); + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), remaining); + timer.unref(); + try { + const response = await fetch(`${this.config.doozyApiUrl}${path}`, { + method, + headers: { + Authorization: `Bearer ${this.config.doozyApiKey}`, + Accept: "application/json", + ...(body === undefined ? {} : { "Content-Type": "application/json" }), + ...(idempotencyKey ? { "Idempotency-Key": idempotencyKey } : {}), + }, + ...(body === undefined ? {} : { body: JSON.stringify(body) }), + signal: controller.signal, + }); + const payload = await response.json().catch(() => ({})) as JsonObject; + if (!response.ok) { + const retry = response.headers.get("retry-after"); + const retryAfterMs = retry && Number.isFinite(Number(retry)) ? Number(retry) * 1_000 : undefined; + throw new DoozyHttpError(`Doozy API ${method} ${path.split("?")[0]} returned ${response.status}`, + response.status, retryAfterMs); + } + return payload; + } finally { + clearTimeout(timer); + } + } +} + +class DoozyHttpError extends Error { + constructor(message: string, readonly status: number, readonly retryAfterMs?: number) { + super(message); + } +} + +function failure(error: string, retriable: boolean, retryAfterMs?: number): PostResult { + return retryAfterMs === undefined + ? { ok: false, retriable, error } + : { ok: false, retriable, error, retryAfterMs }; +} +function object(value: unknown): JsonObject | undefined { + return value !== null && typeof value === "object" && !Array.isArray(value) ? value as JsonObject : undefined; +} +function array(value: unknown): unknown[] { return Array.isArray(value) ? value : []; } +function string(value: unknown): string | undefined { + return typeof value === "string" && value.trim() ? value.trim() : undefined; +} +function timestamp(value: unknown): number | undefined { + if (typeof value === "number" && Number.isFinite(value)) return value; + if (typeof value !== "string") return undefined; + const parsed = Date.parse(value); + return Number.isNaN(parsed) ? undefined : parsed; +} +function parse(raw: Buffer): JsonObject { + try { return object(JSON.parse(raw.toString("utf8"))) ?? {}; } catch { return {}; } +} +function objects(value: JsonObject, key: string): JsonObject[] { + const source = Array.isArray(value[key]) ? value[key] + : Array.isArray(object(value.data)?.[key]) ? object(value.data)![key] + : Array.isArray(value.data) ? value.data : []; + return (source as unknown[]).map(object).filter((entry): entry is JsonObject => Boolean(entry)); +} +function message(error: unknown): string { + return error instanceof Error ? error.message : String(error); +} diff --git a/daemon/src/index.ts b/daemon/src/index.ts index 06b28fc..a19cbbc 100644 --- a/daemon/src/index.ts +++ b/daemon/src/index.ts @@ -15,22 +15,34 @@ import { ArtifactStore } from "./artifacts.js"; import { OtlpRelay } from "./otel-relay.js"; import { resolveOtlpTraces } from "./otel.js"; import { LinearMcpMonitor } from "./linear-mcp-monitor.js"; +import { DoozyProvider } from "./doozy.js"; +import type { WorkProvider } from "./provider.js"; const config = loadConfig(); let log: EventLog; log = new EventLog(config.dbPath, (app) => selectSessionProfile(log, config, app), ); -const gateway = new LinearGateway( +const linearGateway = new LinearGateway( log, config.apps, config.linearGraphqlUrl, config.linearTokenUrl, ); -const worker = new AckWorker(log, gateway); let cleanupWorker: CleanupWorker | undefined; let sessionWorker: SessionWorker | undefined; -const linearMcpMonitor = config.sessionsEnabled +let worker: AckWorker; +const doozyProvider = config.workProvider === "doozy" + ? new DoozyProvider(log, config, { + onInserted: () => { + worker.trigger(); + sessionWorker?.trigger(); + }, + }) + : undefined; +const gateway: WorkProvider = doozyProvider ?? linearGateway; +worker = new AckWorker(log, gateway); +const linearMcpMonitor = config.sessionsEnabled && config.workProvider !== "doozy" ? new LinearMcpMonitor({ url: config.linearMcpUrl, token: config.linearApiKey!, @@ -96,8 +108,8 @@ const triggerWorkers = () => { void cleanupWorker?.trigger(); }; const onStop = (id: string) => sessionWorker?.stopSession(id); -const reconcileWorker = hasLinearApiCreds() - ? new ReconcileWorker(log, gateway, config, { +const reconcileWorker = config.workProvider !== "doozy" && hasLinearApiCreds() + ? new ReconcileWorker(log, linearGateway, config, { onInserted: triggerWorkers, onStop, }) @@ -134,6 +146,7 @@ if (providerPoller) { worker.start(); linearMcpMonitor?.start(); await sessionWorker?.start(); +doozyProvider?.start(); cleanupWorker?.start(); reconcileWorker?.start(); const address = await server.listen(); @@ -161,6 +174,7 @@ async function shutdown(signal: string): Promise { }), ); await reconcileWorker?.stop(); + await doozyProvider?.stop(); await linearMcpMonitor?.stop(); await server.close(); await worker.stop(); diff --git a/daemon/src/provider.ts b/daemon/src/provider.ts new file mode 100644 index 0000000..cfd515c --- /dev/null +++ b/daemon/src/provider.ts @@ -0,0 +1,38 @@ +import type { AppName, Config } from "./config.js"; +import type { TurnRow } from "./eventlog.js"; +import type { + PostResult, + ProgressContent, + TerminalContent, +} from "./linear.js"; + +export interface WorkProvider { + postAckActivity( + app: AppName, + sessionId: string, + activityId: string, + deadlineAt: number, + ): Promise; + postActivity( + app: AppName, + sessionId: string, + activityId: string, + content: ProgressContent | TerminalContent, + ephemeral: boolean, + deadlineAt: number, + ): Promise; + setSessionExternalUrl( + app: AppName, + sessionId: string, + label: string, + url: string, + deadlineAt: number, + ): Promise; + turnPrompt?( + turn: TurnRow, + identifier: string, + implementer: boolean, + resuming: boolean, + ): string; + mcpConfigJson?(config: Config): string; +} diff --git a/daemon/src/server.ts b/daemon/src/server.ts index 904ccf2..f88fb05 100644 --- a/daemon/src/server.ts +++ b/daemon/src/server.ts @@ -63,6 +63,9 @@ export class WebhookServer { return; } const routeMatch = /^\/webhook\/(planner|implementer)$/.exec(pathname); + if (routeMatch && this.options.config.workProvider === "doozy") { + this.earlyJson(request, response, 404, { error: "not_found" }); return; + } if (routeMatch && request.method !== "POST") { this.earlyJson(request, response, 405, { error: "method_not_allowed" }, { Allow: "POST" }); return; diff --git a/daemon/src/sessions.ts b/daemon/src/sessions.ts index 007bce0..38af5e1 100644 --- a/daemon/src/sessions.ts +++ b/daemon/src/sessions.ts @@ -38,7 +38,8 @@ import type { TurnActivityRow, TurnRow, } from "./eventlog.js"; -import type { LinearGateway, PostResult, ProgressContent } from "./linear.js"; +import type { PostResult, ProgressContent } from "./linear.js"; +import type { WorkProvider } from "./provider.js"; import { buildTurnSpan, mintSpanId, postSpans, traceContext } from "./otel.js"; import type { OtlpRelay, RelayCapability } from "./otel-relay.js"; import { @@ -344,7 +345,7 @@ export class SessionWorker { constructor( private readonly log: EventLog, - private readonly gateway: LinearGateway, + private readonly gateway: WorkProvider, private readonly config: Config, private readonly options: SessionWorkerOptions = {}, ) { @@ -529,8 +530,9 @@ export class SessionWorker { const cliproxyApiKey = await readCliproxyApiKey( this.config.cliproxyEnvFile, ); - let prompt = - implementer && !resuming + let prompt = this.gateway.turnPrompt + ? this.gateway.turnPrompt(turn, identifier, implementer, resuming) + : implementer && !resuming ? `/do ${identifier}` : this.composePrompt(turn, identifier); if ((!implementer || resuming) && this.config.attachmentsEnabled) @@ -567,7 +569,7 @@ export class SessionWorker { Math.max(10, Math.min(this.config.keepaliveMs, 60_000)), ); keepalive.unref(); - const linearMcpConfigJson = JSON.stringify({ + const linearMcpConfigJson = this.gateway.mcpConfigJson?.(this.config) ?? JSON.stringify({ mcpServers: { linear: { type: "http", @@ -675,7 +677,7 @@ export class SessionWorker { CLIPROXY_API_KEY: cliproxyApiKey, BASH_DEFAULT_TIMEOUT_MS: String(this.config.bashDefaultTimeoutMs), BASH_MAX_TIMEOUT_MS: String(this.config.bashMaxTimeoutMs), - LINEAR_API_KEY: this.config.linearApiKey!, + ...(this.config.linearApiKey ? { LINEAR_API_KEY: this.config.linearApiKey } : {}), GH_TOKEN: process.env.GH_TOKEN, GITHUB_TOKEN: process.env.GITHUB_TOKEN, ...(this.config.artifactToken diff --git a/daemon/test/config.test.ts b/daemon/test/config.test.ts index 61d5ddd..507fdda 100644 --- a/daemon/test/config.test.ts +++ b/daemon/test/config.test.ts @@ -12,6 +12,7 @@ describe("loadConfig", () => { it("loads test static tokens and defaults", () => { const config = loadConfig(base); expect(config.bindAddr).toBe("127.0.0.1"); + expect(config.workProvider).toBe("linear"); expect(config.replayWindowMs).toBe(60_000); expect(config.webhookBaseUrl).toBe("http://127.0.0.1:8787"); expect(config.artifactToken).toBeUndefined(); @@ -122,6 +123,31 @@ describe("loadConfig", () => { it("names missing variables", () => { expect(() => loadConfig({ ...base, PLANNER_WEBHOOK_SECRET: "" })).toThrow("PLANNER_WEBHOOK_SECRET"); }); + it("loads Doozy without Linear credentials and validates provider settings", () => { + const doozy = loadConfig({ + DAEMON_TEST_MODE: "1", + WORK_PROVIDER: "doozy", + SESSIONS_ENABLED: "0", + DOOZY_API_URL: "https://staging.example.test/api/v1/", + DOOZY_API_KEY: "key", + DOOZY_TAG: "daemon-agent", + DOOZY_POLL_INTERVAL_MS: "2500", + DOOZY_POLL_LIMIT: "2", + }); + expect(doozy).toMatchObject({ + workProvider: "doozy", + doozyApiUrl: "https://staging.example.test/api/v1", + doozyTag: "daemon-agent", + doozyPollIntervalMs: 2500, + doozyPollLimit: 2, + }); + expect(doozy.apps.planner.webhookSecret).toBe(""); + expect(() => loadConfig({ ...base, WORK_PROVIDER: "other" })).toThrow("WORK_PROVIDER"); + expect(() => loadConfig({ + DAEMON_TEST_MODE: "1", WORK_PROVIDER: "doozy", SESSIONS_ENABLED: "0", + DOOZY_API_URL: "https://staging.example.test", DOOZY_API_KEY: "key", + })).toThrow("DOOZY_AGENT_ID or DOOZY_TAG"); + }); it("parses and validates the optional Claudex runtime", () => { expect(loadConfig({ ...base, CLAUDEX_BIN: "claude --model gpt-5.6-sol", CLAUDEX_ENV: '{"ANTHROPIC_BASE_URL":"http://proxy","ENABLE_TOOL_SEARCH":"true"}' })) diff --git a/daemon/test/doozy.test.ts b/daemon/test/doozy.test.ts new file mode 100644 index 0000000..819d4ea --- /dev/null +++ b/daemon/test/doozy.test.ts @@ -0,0 +1,145 @@ +import { execFileSync } from "node:child_process"; +import { mkdtempSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { createServer } from "node:http"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { afterEach, describe, expect, it } from "vitest"; +import { loadConfig } from "../src/config.js"; +import { DoozyProvider } from "../src/doozy.js"; +import { EventLog } from "../src/eventlog.js"; +import { SessionWorker } from "../src/sessions.js"; + +const dirs: string[] = []; +const oldMode = process.env.CLAUDE_FAKE_MODE; +const oldEnvFile = process.env.CLAUDE_FAKE_ENV_FILE; +afterEach(() => { + for (const dir of dirs.splice(0)) rmSync(dir, { recursive: true, force: true }); + if (oldMode === undefined) delete process.env.CLAUDE_FAKE_MODE; + else process.env.CLAUDE_FAKE_MODE = oldMode; + if (oldEnvFile === undefined) delete process.env.CLAUDE_FAKE_ENV_FILE; + else process.env.CLAUDE_FAKE_ENV_FILE = oldEnvFile; +}); + +describe("DoozyProvider", () => { + it("claims a tagged todo, runs it, posts progress and PR, then resumes from human chat", async () => { + const dir = mkdtempSync(join(tmpdir(), "doozy-provider-")); + dirs.push(dir); + const repo = setupRepo(dir); + const todo = { + id: "a0000000-0000-0000-0000-000000000164", + title: "[daemon-agent] Implementer: prove the adapter", + content: "Open a pull request and report it here.", + status: "ready", + latestRun: { chatId: "chat-164" }, + }; + const messages: Array> = []; + const requests: Array<{ method: string; path: string; body?: Record }> = []; + const api = createServer(async (request, response) => { + const body = await readJson(request); + requests.push({ method: request.method!, path: request.url!, ...(body ? { body } : {}) }); + expect(request.headers.authorization).toBe("Bearer test-doozy-key"); + response.setHeader("Content-Type", "application/json"); + if (request.method === "GET" && request.url?.startsWith("/todos?")) + return response.end(JSON.stringify({ todos: [todo] })); + if (request.method === "GET" && request.url === `/todos/${todo.id}`) + return response.end(JSON.stringify({ todo })); + if (request.method === "PATCH" && request.url === `/todos/${todo.id}`) { + Object.assign(todo, body); + return response.end(JSON.stringify({ todo })); + } + if (request.method === "GET" && request.url === "/chats/chat-164/messages") + return response.end(JSON.stringify({ messages })); + if (request.method === "POST" && request.url === "/chats/chat-164/messages") { + messages.push({ id: `daemon-${messages.length}`, role: "user", text: body?.message, createdAt: new Date().toISOString() }); + response.statusCode = 201; + return response.end(JSON.stringify({ message: messages.at(-1) })); + } + response.statusCode = 404; + response.end(JSON.stringify({ error: "not found" })); + }); + await new Promise((resolveListen) => api.listen(0, "127.0.0.1", resolveListen)); + const port = (api.address() as { port: number }).port; + const proxyFile = join(dir, "proxy.env"); + writeFileSync(proxyFile, "CLIPROXY_API_KEY=fake-provider-key\n"); + const config = loadConfig({ + DAEMON_TEST_MODE: "1", + WORK_PROVIDER: "doozy", + SESSIONS_ENABLED: "1", + TARGET_REPO_PATH: repo, + DB_PATH: join(dir, "events.db"), + WORKTREES_ROOT: join(dir, "worktrees"), + DOOZY_API_URL: `http://127.0.0.1:${port}`, + DOOZY_API_KEY: "test-doozy-key", + DOOZY_TAG: "daemon-agent", + CLAUDE_BIN: `${process.execPath} ${resolve("test/fixtures/fake-claude.mjs")}`, + FABLE_BIN: `${process.execPath} ${resolve("test/fixtures/fake-claude.mjs")}`, + CLIPROXY_ENV_FILE: proxyFile, + BROWSER_ENABLED: "0", + KEEPALIVE_MS: "1000", + }); + const log = new EventLog(config.dbPath); + let worker: SessionWorker; + const provider = new DoozyProvider(log, config, { onInserted: () => worker.trigger() }); + worker = new SessionWorker(log, provider, config, { pollMs: 10, reconcileMs: 20 }); + process.env.CLAUDE_FAKE_MODE = "do-pr"; + process.env.CLAUDE_FAKE_ENV_FILE = join(dir, "claude-env.jsonl"); + await worker.start(); + await provider.trigger(); + await waitFor(() => todo.status === "done" && String(todo.content).includes("pull/42")); + + expect(requests).toContainEqual(expect.objectContaining({ + method: "PATCH", + path: `/todos/${todo.id}`, + body: { status: "in_progress" }, + })); + expect(String(todo.content)).toContain("implementation started"); + expect(String(todo.content)).toContain("Pull Request: https://github.com/"); + expect(messages.some((entry) => String(entry.text).includes("Opened https://github.com/"))).toBe(true); + const childEnv = JSON.parse(readFileSync(process.env.CLAUDE_FAKE_ENV_FILE, "utf8").split("\n")[0]).env; + expect(childEnv.DOOZY_API_KEY).toBeUndefined(); + + todo.status = "in_progress"; + messages.push({ + id: "human-follow-up", + role: "user", + text: "Please double-check the result.", + createdAt: new Date(Date.now() + 1_000).toISOString(), + }); + process.env.CLAUDE_FAKE_MODE = "happy"; + await provider.trigger(); + await waitFor(() => messages.some((entry) => String(entry.text).includes("resumed claude-do-session"))); + + await provider.stop(); + await worker.stop(); + log.close(); + await new Promise((resolveClose) => api.close(() => resolveClose())); + }); +}); + +function setupRepo(dir: string): string { + const seed = join(dir, "seed"), origin = join(dir, "origin.git"), repo = join(dir, "repo"); + mkdirSync(seed); + git(["init", "-b", "main"], seed); + git(["config", "user.email", "test@example.com"], seed); + git(["config", "user.name", "Test"], seed); + git(["commit", "--allow-empty", "-m", "initial"], seed); + git(["clone", "--bare", seed, origin]); + git(["clone", origin, repo]); + return repo; +} +function git(args: string[], cwd?: string): void { + execFileSync("git", args, { cwd, stdio: "ignore" }); +} +async function readJson(request: import("node:http").IncomingMessage): Promise | undefined> { + const chunks: Buffer[] = []; + for await (const chunk of request) chunks.push(Buffer.from(chunk)); + if (!chunks.length) return undefined; + return JSON.parse(Buffer.concat(chunks).toString("utf8")) as Record; +} +async function waitFor(predicate: () => boolean, timeoutMs = 5_000): Promise { + const end = Date.now() + timeoutMs; + while (!predicate()) { + if (Date.now() > end) throw new Error("timed out"); + await new Promise((resolveWait) => setTimeout(resolveWait, 20)); + } +}