From f0c7be25f41f72047410a299ae2d959bbdac5571 Mon Sep 17 00:00:00 2001 From: chahinebrini Date: Sun, 12 Jul 2026 01:19:42 +0200 Subject: [PATCH] =?UTF-8?q?feat(agenthub):=20TSK-0118=20=E2=80=94=20Ask=20?= =?UTF-8?q?primitive=20for=20autonomous=20decision-routing?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - New Ask entity (ASK-####): schema + counter + paths(EntityType 'asks') + events(AgentHubEventType 'ask') + fsWatch(WATCHED + toEvent). Generic entities table, no migration. - askService: createAsk routes to config.roles.architect.preferredAgent, NEVER the CEO (a to='ceo' is rerouted); answerAsk / escalateAsk (escalatedTo='ceo', single channel) / getAsk / listAsks. Authority-policy JSDoc. - Asks kept OUT of FTS5: Index.upsert gains a { fts?: boolean } option; askService upserts with fts:false, so 'memory search' never returns asks. - routes: POST/GET /asks, GET /asks/:id, POST /asks/:id/{answer,escalate}, each emitChange type:'ask'. - CLI 'ask --from [--task][--wait][--timeout]' (SSE reconnect wait until status!=pending) + ask list/answer/escalate; remoteClient ask methods. - MCP agenthub_ask (wait via waitForTask, now woken by 'ask' events) + agenthub_ask_list/answer/escalate; agenthub_work architect branch surfaces pending asks ({reviews,asks,messages}). - Unattended mode (invocation flag): work.ts ctx + CLI 'work --unattended' + agenthub_work schema + LOOP reminder ('call agenthub_ask instead of pausing'). - tests: +askService.test.ts (routing/answer/escalate/list/FTS-exclusion), +ask-wait.test.ts (routes roundtrip + SSE wait: answer resolves, no-answer times out cleanly) Co-Authored-By: Claude Opus 4.8 --- src/cli/commands/ask.ts | 114 +++++++++++++++++++++++++ src/cli/commands/work.ts | 9 ++ src/cli/index.ts | 94 ++++++++++++++++++++- src/cli/remoteClient.ts | 24 +++++- src/core/counter.ts | 1 + src/core/index.ts | 19 +++-- src/core/paths.ts | 2 +- src/core/schema.ts | 24 ++++++ src/core/services/askService.ts | 142 ++++++++++++++++++++++++++++++++ src/mcp/server.ts | 59 +++++++++++-- src/server/events.ts | 2 +- src/server/fsWatch.ts | 3 + src/server/routes.ts | 62 +++++++++++++- tests/ask-wait.test.ts | 93 +++++++++++++++++++++ tests/askService.test.ts | 80 ++++++++++++++++++ 15 files changed, 709 insertions(+), 19 deletions(-) create mode 100644 src/cli/commands/ask.ts create mode 100644 src/core/services/askService.ts create mode 100644 tests/ask-wait.test.ts create mode 100644 tests/askService.test.ts diff --git a/src/cli/commands/ask.ts b/src/cli/commands/ask.ts new file mode 100644 index 0000000..4619d07 --- /dev/null +++ b/src/cli/commands/ask.ts @@ -0,0 +1,114 @@ +import { createAsk, listAsks, answerAsk, escalateAsk } from '../../core/services/askService.js'; +import { parseSSEBuffer } from './watch.js'; +import { remoteClient } from '../remoteClient.js'; +import type { Ask } from '../../core/schema.js'; + +export function askCreate(cwd: string, opts: { from: string; question: string; taskId?: string }): Ask { + const a = createAsk(cwd, { from: opts.from, question: opts.question, taskId: opts.taskId }); + console.log(`AgentHub: Ask sent ${a.id} (${a.from} → ${a.to})${a.taskId ? ` [${a.taskId}]` : ''}: ${a.question}`); + return a; +} + +export function askList(cwd: string, opts: { pending?: boolean } = {}): void { + const asks = listAsks(cwd, opts.pending ? { status: 'pending' } : {}); + if (asks.length === 0) { + console.log('No asks.'); + return; + } + for (const a of asks) { + const flag = a.status === 'pending' ? '●' : ' '; + let line = `${flag} ${a.id} ${a.from} → ${a.to} [${a.status}]${a.taskId ? ` (${a.taskId})` : ''}: ${a.question}`; + if (a.answer) line += `\n ↳ ${a.answeredBy ?? a.to}: ${a.answer}`; + if (a.status === 'escalated') line += `\n ↳ escalated to ${a.escalatedTo ?? 'ceo'}`; + console.log(line); + } +} + +export function askAnswer(cwd: string, id: string, text: string, by?: string): void { + const a = answerAsk(cwd, id, text, by); + console.log(`AgentHub: Ask answered ${a.id}${by ? ` by ${by}` : ''}`); +} + +export function askEscalate(cwd: string, id: string, note?: string): void { + const a = escalateAsk(cwd, id, note); + console.log(`AgentHub: Ask escalated ${a.id} → ${a.escalatedTo ?? 'ceo'}`); +} + +/** + * Block on the SSE stream until the Ask leaves `pending` (answered/escalated), + * or the timeout elapses. Mirrors the work-loop wait (reconnect-friendly) but + * wakes on `ask` events. Resolves with the settled Ask, or null on timeout. + */ +export function waitForAsk(serverUrl: string, askId: string, timeoutSec?: number): Promise { + return new Promise((resolve) => { + let settled = false; + let controller: AbortController | undefined; + const finish = (v: Ask | null) => { + if (settled) return; + settled = true; + try { + controller?.abort(); + } catch { + /* already aborted */ + } + resolve(v); + }; + const deadline = timeoutSec ? Date.now() + timeoutSec * 1000 : undefined; + const timer = timeoutSec ? setTimeout(() => finish(null), timeoutSec * 1000) : undefined; + + const check = async (): Promise => { + try { + const { ask } = await remoteClient.getAsk(serverUrl, askId); + if (ask.status !== 'pending') { + if (timer) clearTimeout(timer); + finish(ask); + return true; + } + } catch { + /* transient — keep listening */ + } + return false; + }; + + const loop = async () => { + while (!settled && (deadline === undefined || Date.now() < deadline)) { + controller = new AbortController(); + try { + const res = await fetch(`${serverUrl}/events`, { signal: controller.signal, headers: { Accept: 'text/event-stream' } }); + if (!res.body) throw new Error('SSE response has no body'); + if (await check()) return; // close the gap after subscribing + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + while (!settled) { + let done: boolean; + let value: Uint8Array | undefined; + try { + ({ done, value } = await reader.read()); + } catch { + break; + } + if (done) break; + if (value) buffer += decoder.decode(value, { stream: true }); + const { events, remaining } = parseSSEBuffer(buffer); + buffer = remaining; + if (events.some((e) => e.type === 'ask')) { + if (await check()) return; + } + } + } catch (err: unknown) { + if (settled || (err instanceof Error && err.name === 'AbortError')) return; + } + if (settled) break; + await new Promise((r) => setTimeout(r, 1000)); + } + if (timer) clearTimeout(timer); + finish(null); + }; + + loop().catch(() => { + if (timer) clearTimeout(timer); + finish(null); + }); + }); +} diff --git a/src/cli/commands/work.ts b/src/cli/commands/work.ts index 2a955d1..b1d3c21 100644 --- a/src/cli/commands/work.ts +++ b/src/cli/commands/work.ts @@ -19,6 +19,12 @@ interface WorkAgentContext extends AgentContext { timeoutSec?: number; discoverServer?: (timeoutMs?: number) => Promise; reconnectBackoffMs?: number[]; + /** + * Unattended mode (TSK-0118): the agent runs without a human at the keyboard. + * It must never pause for human input — when it needs a decision it routes an + * `agenthub ask` to the architect and awaits the answer, instead of stalling. + */ + unattended?: boolean; } /** @@ -71,6 +77,9 @@ export async function workAgent(ctx: WorkAgentContext): Promise { console.log( `AgentHub: waiting for a task or message addressed to ${ctx.agent}…${ctx.timeoutSec ? ` (timeout ${ctx.timeoutSec}s)` : ''}`, ); + if (ctx.unattended) { + console.log('AgentHub: unattended mode — never pause for human input; route decisions via `agenthub ask` to the architect.'); + } await waitAndClaim(ctx); } diff --git a/src/cli/index.ts b/src/cli/index.ts index 404db34..efcee16 100644 --- a/src/cli/index.ts +++ b/src/cli/index.ts @@ -6,6 +6,7 @@ import { taskCreate, taskList, taskShow, taskClaim, taskDone, taskReview, taskRe import { handoffCreate, handoffRead, handoffList } from './commands/handoff.js'; import { decisionCreate, decisionList } from './commands/decision.js'; import { messageSend, inboxList, messageRead, inboxMarkRead, messageAck, messageReply } from './commands/message.js'; +import { askCreate, askList, askAnswer, askEscalate, waitForAsk } from './commands/ask.js'; import { agentSetup, hookContext } from './commands/agentSetup.js'; import { syncOrgFromFile } from '../core/services/orgService.js'; import { delegate } from './commands/delegate.js'; @@ -604,6 +605,96 @@ export function createProgram(cwd: string): Command { } }); + // ─── asks (autonomous decision-routing) ──────────────────────────────────── + const askCmd = new Command('ask') + .description('Ask the architect a blocking question (routes to the architect, never the CEO)') + .argument('[question]', 'Question to route to the architect') + .option('--from ', 'Asking agent') + .option('--task ', 'Related task ID') + .option('--wait', 'Block until the architect answers or escalates') + .option('--timeout ', 'With --wait: stop waiting after N seconds') + .action(async (question: string | undefined, options: { from?: string; task?: string; wait?: boolean; timeout?: string }) => { + if (!question || !options.from) { + askCmd.help({ error: true }); + return; + } + const { serverUrl, projectCwd } = await resolveContext(program, cwd); + const payload = { from: options.from, question, taskId: options.task }; + if (serverUrl) { + await runRemote(serverUrl, async () => { + const a = await remoteClient.createAsk(serverUrl, payload); + console.log(`AgentHub: Ask sent ${a.id} (${a.from} → ${a.to})${a.taskId ? ` [${a.taskId}]` : ''}: ${a.question}`); + if (options.wait) { + const timeoutSec = options.timeout != null ? Number(options.timeout) : undefined; + console.log(`AgentHub: waiting for an answer to ${a.id}…${timeoutSec ? ` (timeout ${timeoutSec}s)` : ''}`); + const answered = await waitForAsk(serverUrl, a.id, timeoutSec); + if (!answered) { + console.log(`AgentHub: no answer to ${a.id}${timeoutSec ? ` within ${timeoutSec}s` : ''} — re-check with: agenthub ask list`); + return; + } + if (answered.status === 'answered') { + console.log(`AgentHub: ${a.id} answered by ${answered.answeredBy ?? answered.to}: ${answered.answer}`); + } else { + console.log(`AgentHub: ${a.id} escalated to ${answered.escalatedTo ?? 'ceo'} — await the CEO decision.`); + } + } + }); + } else { + askCreate(projectCwd, payload); + if (options.wait) console.log('AgentHub: --wait needs a running server; the ask was created without waiting.'); + } + }); + askCmd + .command('list') + .description('List asks (● = pending)') + .option('--pending', 'Only pending asks') + .action(async (options: { pending?: boolean }) => { + const { serverUrl, projectCwd } = await resolveContext(program, cwd); + if (serverUrl) { + await runRemote(serverUrl, async () => { + const asks = await remoteClient.listAsks(serverUrl, options.pending ? { status: 'pending' } : undefined); + if (asks.length === 0) { console.log('No asks.'); return; } + for (const a of asks) { + console.log(`${a.status === 'pending' ? '●' : ' '} ${a.id} ${a.from} → ${a.to} [${a.status}]${a.taskId ? ` (${a.taskId})` : ''}: ${a.question}`); + } + }); + } else { + askList(projectCwd, { pending: options.pending }); + } + }); + askCmd + .command('answer ') + .description('Answer an ask (architect)') + .requiredOption('--text ', 'Answer text') + .option('--by ', 'Answering agent') + .action(async (id: string, options: { text: string; by?: string }) => { + const { serverUrl, projectCwd } = await resolveContext(program, cwd); + if (serverUrl) { + await runRemote(serverUrl, async () => { + const a = await remoteClient.answerAsk(serverUrl, id, options.text, options.by); + console.log(`AgentHub: Ask answered ${a.id}${options.by ? ` by ${options.by}` : ''}`); + }); + } else { + askAnswer(projectCwd, id, options.text, options.by); + } + }); + askCmd + .command('escalate ') + .description('Escalate an ask to the CEO (architect: release/publish/push, OSS, architecture pivots)') + .option('--note ', 'Escalation note') + .action(async (id: string, options: { note?: string }) => { + const { serverUrl, projectCwd } = await resolveContext(program, cwd); + if (serverUrl) { + await runRemote(serverUrl, async () => { + const a = await remoteClient.escalateAsk(serverUrl, id, options.note); + console.log(`AgentHub: Ask escalated ${a.id} → ${a.escalatedTo ?? 'ceo'}`); + }); + } else { + askEscalate(projectCwd, id, options.note); + } + }); + program.addCommand(askCmd); + // ─── auto-start ────────────────────────────────────────────────────────── const agentCmd = new Command('agent').description('Per-agent machine setup'); agentCmd @@ -730,10 +821,11 @@ export function createProgram(cwd: string): Command { .requiredOption('--agent ', 'Agent name') .option('--role ', 'Role (default: implementer)', 'implementer') .option('--timeout ', 'Stop waiting after N seconds (default: wait indefinitely)') + .option('--unattended', 'Unattended mode: never pause for human input — route decisions via `agenthub ask`') .action(async (options) => { const { serverUrl, projectCwd } = await resolveContext(program, cwd); const timeoutSec = options.timeout != null ? Number(options.timeout) : undefined; - const ctx = { serverUrl, projectCwd, agent: options.agent, role: options.role, timeoutSec }; + const ctx = { serverUrl, projectCwd, agent: options.agent, role: options.role, timeoutSec, unattended: !!options.unattended }; if (serverUrl) { await runRemote(serverUrl, () => workAgent(ctx)); } else { diff --git a/src/cli/remoteClient.ts b/src/cli/remoteClient.ts index b5e2895..8063f36 100644 --- a/src/cli/remoteClient.ts +++ b/src/cli/remoteClient.ts @@ -1,4 +1,4 @@ -import type { Task, Handoff, Decision, Memory, Message, ActivityItem } from '../core/schema.js'; +import type { Task, Handoff, Decision, Memory, Message, Ask, ActivityItem } from '../core/schema.js'; import type { IndexEntry } from '../core/index.js'; import type { InboxMessage } from '../core/services/messageService.js'; import type { TaskLogEntry } from '../core/services/taskLogService.js'; @@ -154,6 +154,28 @@ export const remoteClient = { return request<{ message: Message; body: string }>(baseUrl, 'GET', `/messages/${id}`); }, + async createAsk(baseUrl: string, options: Partial): Promise { + return request(baseUrl, 'POST', '/asks', options); + }, + + async listAsks(baseUrl: string, filters?: { to?: string; status?: string }): Promise { + const params = new URLSearchParams((filters ?? {}) as Record); + const qs = params.toString(); + return request(baseUrl, 'GET', `/asks${qs ? '?' + qs : ''}`); + }, + + async getAsk(baseUrl: string, id: string): Promise<{ ask: Ask; body: string }> { + return request<{ ask: Ask; body: string }>(baseUrl, 'GET', `/asks/${id}`); + }, + + async answerAsk(baseUrl: string, id: string, text: string, by?: string): Promise { + return request(baseUrl, 'POST', `/asks/${id}/answer`, { text, by }); + }, + + async escalateAsk(baseUrl: string, id: string, note?: string): Promise { + return request(baseUrl, 'POST', `/asks/${id}/escalate`, note ? { note } : undefined); + }, + async delegate(baseUrl: string, auto: boolean): Promise<{ suggestion?: { task: IndexEntry; role: string; preferredAgent: string }; handoff?: Handoff }> { return request<{ suggestion?: { task: IndexEntry; role: string; preferredAgent: string }; handoff?: Handoff }>(baseUrl, 'POST', `/delegate?auto=${auto}`); }, diff --git a/src/core/counter.ts b/src/core/counter.ts index 94c2fed..eaabb37 100644 --- a/src/core/counter.ts +++ b/src/core/counter.ts @@ -7,6 +7,7 @@ const prefixes: Record = { decision: 'DEC', memory: 'MEM', message: 'MSG', + ask: 'ASK', }; export type CounterType = keyof typeof prefixes; diff --git a/src/core/index.ts b/src/core/index.ts index ab4c49a..3a88118 100644 --- a/src/core/index.ts +++ b/src/core/index.ts @@ -71,7 +71,14 @@ export class Index { } } - upsert(entry: IndexEntry): void { + /** + * Upsert an entity into the index. Pass `{ fts: false }` to keep it out of the + * full-text search table (e.g. Asks — transient decision-routing questions + * that shouldn't pollute `memory search`); it still lives in the `entities` + * table so it can be listed/filtered by type. + */ + upsert(entry: IndexEntry, opts: { fts?: boolean } = {}): void { + const { fts = true } = opts; const params = { status: null, role: null, @@ -100,10 +107,12 @@ export class Index { `); insert.run(params); - const search = this.db.prepare(` - INSERT OR REPLACE INTO search (id, title, content) VALUES (@id, @title, @content) - `); - search.run(params); + if (fts) { + const search = this.db.prepare(` + INSERT OR REPLACE INTO search (id, title, content) VALUES (@id, @title, @content) + `); + search.run(params); + } } /** Permanently drop an entity from both the entities table and the FTS index. */ diff --git a/src/core/paths.ts b/src/core/paths.ts index d2fa48d..0700b42 100644 --- a/src/core/paths.ts +++ b/src/core/paths.ts @@ -1,7 +1,7 @@ import { existsSync } from 'fs'; import { dirname, join, parse } from 'path'; -export type EntityType = 'tasks' | 'handoffs' | 'decisions' | 'memory' | 'messages' | 'status'; +export type EntityType = 'tasks' | 'handoffs' | 'decisions' | 'memory' | 'messages' | 'asks' | 'status'; export function getAgentHubDir(cwd: string = process.cwd()): string { return join(cwd, '.agenthub'); diff --git a/src/core/schema.ts b/src/core/schema.ts index 5e75e67..768810d 100644 --- a/src/core/schema.ts +++ b/src/core/schema.ts @@ -99,6 +99,29 @@ export const MessageSchema = z.object({ updatedAt: z.string().datetime(), }); +export const AskStatus = z.enum(['pending', 'answered', 'escalated']); + +/** + * A blocking question routed for a decision (TSK-0118). Unlike a message (a + * fire-and-forget ping), an Ask has a lifecycle: an implementer asks, the + * architect answers (closing it) or escalates it to the CEO. It is the single + * blocking channel — an implementer awaits exactly ONE Ask, never two. + */ +export const AskSchema = z.object({ + id: z.string().regex(/^ASK-\d{4}$/), + from: z.string().min(1), + to: z.string().min(1), + question: z.string().min(1), + taskId: z.string().optional(), + status: AskStatus.default('pending'), + answer: z.string().optional(), + answeredBy: z.string().optional(), + escalatedTo: z.string().optional(), + decisionId: z.string().optional(), + createdAt: z.string().datetime(), + updatedAt: z.string().datetime(), +}); + // Activity timeline item returned by GET /tasks/:id/activity export const ActivityItemSchema = z.object({ at: z.string().datetime(), @@ -181,5 +204,6 @@ export type Handoff = z.infer; export type Decision = z.infer; export type Memory = z.infer; export type Message = z.infer; +export type Ask = z.infer; export type Status = z.infer; export type Config = z.infer; diff --git a/src/core/services/askService.ts b/src/core/services/askService.ts new file mode 100644 index 0000000..9ad6e83 --- /dev/null +++ b/src/core/services/askService.ts @@ -0,0 +1,142 @@ +import { join } from 'path'; +import { getEntityDir } from '../paths.js'; +import { getNextId } from '../counter.js'; +import { readEntity, writeEntity } from '../files.js'; +import { AskSchema, type Ask } from '../schema.js'; +import { Index } from '../index.js'; +import { loadConfig } from '../config.js'; + +/** + * Ask primitive — autonomous decision-routing (TSK-0118). + * + * ## Authority policy + * An Ask is the single blocking channel for a decision an agent can't make + * alone. It flows strictly: + * + * implementer → ARCHITECT → (answer | escalate → CEO) + * + * - Implementer questions ALWAYS route to the architect (never the CEO). The + * implementer awaits exactly ONE Ask — never a second, parallel blocking + * channel. + * - The architect answers most Asks himself, within the approve/push gate. + * - The architect MUST escalate (not answer) for: release / publish / push, + * OSS decisions, and architecture pivots. Escalation flips the Ask to + * `escalated` (escalatedTo = 'ceo') and closes the loop through the SAME Ask — + * the implementer keeps waiting on that one Ask, it never opens a CEO channel. + */ + +function indexEntryFor(record: Ask, filePath: string) { + return { + id: record.id, + type: 'ask', + title: `${record.from} → ${record.to}`, + content: record.question, + filePath, + createdAt: record.createdAt, + updatedAt: record.updatedAt, + status: record.status, + fromAgent: record.from, + toAgent: record.to, + taskId: record.taskId, + }; +} + +/** + * Who an Ask routes to by default: the architect's preferred agent. NEVER the + * CEO — a raw question is the architect's to field; only an explicit escalation + * reaches the CEO. + */ +function defaultRoutedTo(cwd: string): string { + try { + return loadConfig(cwd).roles?.architect?.preferredAgent || 'architect'; + } catch { + return 'architect'; + } +} + +/** Ask a routed question. Defaults to the architect; a `to` of 'ceo' is refused + * (rerouted to the architect) — the CEO is reachable only via escalateAsk. */ +export function createAsk(cwd: string, options: Partial = {}): Ask { + if (!options.from) throw new Error('Ask requires a "from" agent'); + if (!options.question) throw new Error('Ask requires a question'); + + const now = new Date().toISOString(); + const to = options.to && options.to.toLowerCase() !== 'ceo' ? options.to : defaultRoutedTo(cwd); + const record: Ask = AskSchema.parse({ + id: getNextId(cwd, 'ask'), + from: options.from, + to, + question: options.question, + taskId: options.taskId, + status: 'pending', + createdAt: now, + updatedAt: now, + }); + + const filePath = join(getEntityDir(cwd, 'asks'), `${record.id}.md`); + writeEntity(filePath, record, `# ${record.from} → ${record.to}\n\n${record.question}`); + + const index = new Index(cwd); + index.upsert(indexEntryFor(record, filePath), { fts: false }); // keep Asks out of FTS5 + index.close(); + + return record; +} + +export function getAsk(cwd: string, id: string): { ask: Ask; body: string; filePath: string } { + if (!id) throw new Error('Ask ID is required'); + const filePath = join(getEntityDir(cwd, 'asks'), `${id}.md`); + const { frontmatter, body } = readEntity(filePath); + return { ask: AskSchema.parse(frontmatter), body, filePath }; +} + +/** Asks, newest first, optionally filtered by recipient and/or status. */ +export function listAsks(cwd: string, opts: { to?: string; status?: string } = {}): Ask[] { + const index = new Index(cwd); + const entries = index.list('ask', opts.status ? { status: opts.status } : undefined); + index.close(); + const ids = entries + .filter((e) => !opts.to || (e.toAgent != null && String(e.toAgent).toLowerCase() === opts.to.toLowerCase())) + .map((e) => e.id); + const out: Ask[] = []; + for (const id of ids) { + try { + out.push(getAsk(cwd, id).ask); + } catch { + /* torn file — skip */ + } + } + return out; +} + +/** Answer an Ask (architect resolves it in the approve/push gate). */ +export function answerAsk(cwd: string, id: string, text: string, by?: string): Ask { + if (!text || !text.trim()) throw new Error('Ask answer requires text'); + const { ask, body, filePath } = getAsk(cwd, id); + const updated: Ask = { ...ask, status: 'answered', answer: text, answeredBy: by, updatedAt: new Date().toISOString() }; + writeEntity(filePath, updated, body); + + const index = new Index(cwd); + index.upsert(indexEntryFor(updated, filePath), { fts: false }); + index.close(); + + return updated; +} + +/** + * Escalate an Ask to the CEO (required for release/publish/push, OSS, and + * architecture pivots). This does NOT open a second blocking channel — it flips + * the SAME Ask to `escalated`; the waiting implementer resolves on this one Ask. + */ +export function escalateAsk(cwd: string, id: string, note?: string): Ask { + const { ask, body, filePath } = getAsk(cwd, id); + const updated: Ask = { ...ask, status: 'escalated', escalatedTo: 'ceo', updatedAt: new Date().toISOString() }; + const newBody = note ? `${body}\n\n_escalated to ceo: ${note}_` : body; + writeEntity(filePath, updated, newBody); + + const index = new Index(cwd); + index.upsert(indexEntryFor(updated, filePath), { fts: false }); + index.close(); + + return updated; +} diff --git a/src/mcp/server.ts b/src/mcp/server.ts index 312b415..a4ff193 100644 --- a/src/mcp/server.ts +++ b/src/mcp/server.ts @@ -18,6 +18,8 @@ import { } from '../core/services/taskService.js'; import { createHandoff, getHandoff } from '../core/services/handoffService.js'; import { appendTaskLog } from '../core/services/taskLogService.js'; +import { createAsk, listAsks, answerAsk, escalateAsk } from '../core/services/askService.js'; +import type { Ask } from '../core/schema.js'; import { createMessage, listInbox, markMessageRead } from '../core/services/messageService.js'; import { addMemory, searchMemory } from '../core/services/memoryService.js'; import { createDecision } from '../core/services/decisionService.js'; @@ -111,8 +113,9 @@ function waitForTask( if (value) buffer += decoder.decode(value, { stream: true }); const { events, remaining } = parseSSEBuffer(buffer); buffer = remaining; - // Wake on a new task OR a new message addressed to the agent. - if (events.some((e) => e.type === 'task' || e.type === 'message')) { + // Wake on a new task, a message, or an ask (decision routed to the + // architect / answered back to the asker). + if (events.some((e) => e.type === 'task' || e.type === 'message' || e.type === 'ask')) { const claimed = await findClaim(currentUrl); if (claimed) { clearTimeout(timer); finish(claimed); return; } } @@ -151,8 +154,8 @@ export async function startMcpServer(cwd: string): Promise { server.tool('agenthub_work', 'Block until there is work for you, then return it. IMPLEMENTER: waits for a task addressed to you (newly delegated OR reopened), claims it, returns it with its handoff (loop: work -> implement -> agenthub_task_review -> work). ARCHITECT/REVIEWER: waits for any task submitted to review and returns the pending set to approve (agenthub_task_done) or reject (agenthub_task_reopen) — so review submissions reach you in realtime.', - { agent: z.string(), role: z.string().optional(), timeoutSec: z.number().optional() }, - async ({ agent, role, timeoutSec }) => { + { agent: z.string(), role: z.string().optional(), timeoutSec: z.number().optional(), unattended: z.boolean().optional() }, + async ({ agent, role, timeoutSec, unattended }) => { const ctx: AgentContext = { serverUrl, projectCwd: root, agent, role: role ?? 'implementer' }; const reviewer = isReviewerRole(ctx.role); const useServerUrl = (nextServerUrl?: string) => { @@ -189,18 +192,22 @@ export async function startMcpServer(cwd: string): Promise { }; // Architect/reviewer variant: wake on tasks submitted to review (not on // tasks addressed to you). Returns the pending review set — never claims. + const listPendingAsks = async (): Promise => + remote ? await remoteClient.listAsks(ctx.serverUrl!, { status: 'pending' }) : listAsks(root, { status: 'pending' }); const findReview = async (nextServerUrl?: string) => { if (nextServerUrl) useServerUrl(nextServerUrl); const reviews = await listReviewTasks(ctx); const messages = await drainInbox(ctx.serverUrl); - if (reviews.length) { + const asks = await listPendingAsks(); + if (reviews.length || asks.length) { return { reviews: reviews.map((r) => ({ id: r.id, title: r.title, assignedTo: r.assignedTo })), + asks: asks.map((a) => ({ id: a.id, from: a.from, question: a.question, taskId: a.taskId })), messages, - note: 'Tasks awaiting your review. Approve: agenthub_task_done. Reject: agenthub_task_reopen (send feedback first via agenthub_message or a handoff). Read the work with agenthub_task_show.', + note: 'Tasks awaiting your review (approve: agenthub_task_done; reject: agenthub_task_reopen). Pending asks: answer with agenthub_ask_answer, or escalate to the CEO with agenthub_ask_escalate (release/publish/push, OSS, architecture pivots).', }; } - if (messages.length) return { reviews: [], messages, note: 'Nothing in review, but you have messages — reply with agenthub_message.' }; + if (messages.length) return { reviews: [], asks: [], messages, note: 'Nothing in review, but you have messages — reply with agenthub_message.' }; return null; }; @@ -208,9 +215,9 @@ export async function startMcpServer(cwd: string): Promise { // Self-perpetuating loop: every response reminds the agent to relaunch // agenthub_work, so a finished task/message never leaves it dormant. - const LOOP = reviewer + let LOOP = reviewer ? 'STAY REACHABLE: the instant you finish reviewing, call agenthub_work again — ' + - 'run it in the background so the next review submission reaches you in realtime. ' + + 'run it in the background so the next review submission or ask reaches you in realtime. ' + 'Never end your turn without relaunching agenthub_work.' : 'STAY REACHABLE: the instant you finish handling this, call agenthub_work again — ' + 'run it in the background so the next task or message reaches you in realtime. ' + @@ -219,6 +226,10 @@ export async function startMcpServer(cwd: string): Promise { 'immediately relaunch agenthub_work and keep it running, so the architect\'s approval, ' + 'reopen (with feedback) or follow-up message wakes you here instead of leaving you dormant. ' + 'Never end your turn without relaunching agenthub_work.'; + if (unattended) { + LOOP += ' UNATTENDED MODE: never pause for human input — when you need a decision, ' + + 'call agenthub_ask (it routes to the architect) and await the answer instead of stalling.'; + } const immediate = await finder(); if (immediate) return asText({ ...immediate, loop: LOOP }); const emptyMsg = reviewer @@ -306,6 +317,36 @@ export async function startMcpServer(cwd: string): Promise { { agent: z.string(), unreadOnly: z.boolean().optional() }, async ({ agent, unreadOnly }) => asText(remote ? await remoteClient.getInbox(serverUrl!, agent, unreadOnly) : listInbox(root, agent, { unreadOnly }))); + server.tool('agenthub_ask', + 'Ask the architect a blocking question when you cannot decide alone (autonomous decision-routing). Routes to the architect — NEVER the CEO. With wait:true it blocks until the architect answers or escalates, then returns the resolution. Await exactly ONE ask at a time.', + { from: z.string(), question: z.string(), taskId: z.string().optional(), wait: z.boolean().optional(), timeoutSec: z.number().optional() }, + async ({ from, question, taskId, wait, timeoutSec }) => { + const ask = remote ? await remoteClient.createAsk(serverUrl!, { from, question, taskId }) : createAsk(root, { from, question, taskId }); + if (!wait) return asText(ask); + if (!remote) return asText({ ...ask, note: 'Created; --wait needs a running hub server.' }); + const settled = await waitForTask(serverUrl!, async (url) => { + const { ask: cur } = await remoteClient.getAsk(url, ask.id); + return cur.status !== 'pending' ? cur : null; + }, timeoutSec ?? 300); + if (settled) return asText(settled); + return asText({ ...ask, note: `No answer within ${timeoutSec ?? 300}s — re-check with agenthub_ask_list.` }); + }); + + server.tool('agenthub_ask_list', + 'List asks (routed decisions). Architect: your pending decision queue. Filter by status (e.g. pending) and/or recipient.', + { to: z.string().optional(), status: z.string().optional() }, + async ({ to, status }) => asText(remote ? await remoteClient.listAsks(serverUrl!, { to, status }) : listAsks(root, { to, status }))); + + server.tool('agenthub_ask_answer', + 'Answer an ask (architect resolves a routed decision within the approve/push gate).', + { id: z.string(), text: z.string(), by: z.string().optional() }, + async ({ id, text, by }) => asText(remote ? await remoteClient.answerAsk(serverUrl!, id, text, by) : answerAsk(root, id, text, by))); + + server.tool('agenthub_ask_escalate', + 'Escalate an ask to the CEO (architect: REQUIRED for release/publish/push, OSS decisions and architecture pivots). Closes the loop through the same ask — no second blocking channel.', + { id: z.string(), note: z.string().optional() }, + async ({ id, note }) => asText(remote ? await remoteClient.escalateAsk(serverUrl!, id, note) : escalateAsk(root, id, note))); + const transport = new StdioServerTransport(); await server.connect(transport); // stdio servers must not write to stdout (it's the protocol channel); log to stderr. diff --git a/src/server/events.ts b/src/server/events.ts index 30bb0b6..2a8b48c 100644 --- a/src/server/events.ts +++ b/src/server/events.ts @@ -1,6 +1,6 @@ import { EventEmitter } from 'node:events'; -export type AgentHubEventType = 'task' | 'handoff' | 'decision' | 'memory' | 'message' | 'agent'; +export type AgentHubEventType = 'task' | 'handoff' | 'decision' | 'memory' | 'message' | 'ask' | 'agent'; export type AgentHubEventAction = 'created' | 'updated' | 'deleted' | 'joined' | 'left'; export interface AgentHubEvent { diff --git a/src/server/fsWatch.ts b/src/server/fsWatch.ts index 2c67517..3b6ce7d 100644 --- a/src/server/fsWatch.ts +++ b/src/server/fsWatch.ts @@ -26,6 +26,7 @@ const WATCHED: { dir: EntityType; type: AgentHubEventType }[] = [ { dir: 'decisions', type: 'decision' }, { dir: 'memory', type: 'memory' }, { dir: 'messages', type: 'message' }, + { dir: 'asks', type: 'ask' }, ]; // fs.watch can fire several events (rename + change) for a single write, and a @@ -73,6 +74,8 @@ function toEvent( return { stamp, event: { type, action, id, title: str(fm.title) } }; case 'message': return { stamp, event: { type, action, id, title: `${str(fm.from)} → ${str(fm.to)}`, status: str(fm.status), assignedTo: str(fm.to) } }; + case 'ask': + return { stamp, event: { type, action, id, title: `${str(fm.from)} → ${str(fm.to)}`, status: str(fm.status), assignedTo: str(fm.to) } }; } } diff --git a/src/server/routes.ts b/src/server/routes.ts index 7bbc16c..6931ce0 100644 --- a/src/server/routes.ts +++ b/src/server/routes.ts @@ -5,6 +5,7 @@ import { appendTaskLog, readTaskLog } from '../core/services/taskLogService.js'; import { createHandoff, listHandoffs, getHandoff } from '../core/services/handoffService.js'; import { createDecision, listDecisions } from '../core/services/decisionService.js'; import { createMessage, listMessages, listInbox, markMessageRead, ackMessage, getMessage } from '../core/services/messageService.js'; +import { createAsk, listAsks, getAsk, answerAsk, escalateAsk } from '../core/services/askService.js'; import { addMemory, searchMemory, listMemory } from '../core/services/memoryService.js'; import { getStatus, updateStatus } from '../core/services/statusService.js'; import { autoDelegate, suggestDelegation } from '../core/services/delegateService.js'; @@ -20,7 +21,7 @@ import { renderMessagesHtml } from './messages.js'; import { renderTaskDetailHtml } from './taskDetail.js'; import { eventBus, emitChange } from './events.js'; import type { AgentHubEvent } from './events.js'; -import type { Task, Handoff, Decision, Memory, Message } from '../core/schema.js'; +import type { Task, Handoff, Decision, Memory, Message, Ask } from '../core/schema.js'; import { readFile } from 'node:fs/promises'; import { fileURLToPath } from 'node:url'; import { join, extname, normalize } from 'node:path'; @@ -523,6 +524,65 @@ export async function registerRoutes(app: FastifyInstance, cwd: string): Promise } }); + // ─── Asks ──────────────────────────────────────────────────────────────── + // Blocking decision-routing questions (TSK-0118). POST creates + routes to the + // architect; answer/escalate close them. Each mutation fires an 'ask' event so + // a waiting `ask --wait` / agenthub_ask wakes. + app.get('/asks', async (request) => { + const { to, status } = request.query as { to?: string; status?: string }; + return listAsks(cwd, { to, status }); + }); + app.post('/asks', async (request, reply) => { + let ask: Ask; + try { + ask = createAsk(cwd, request.body as Partial); + } catch (err) { + return badRequest(reply, err instanceof Error ? err.message : 'Invalid ask'); + } + emitChange( + { type: 'ask', action: 'created', id: ask.id, title: `${ask.from} → ${ask.to}`, status: ask.status, assignedTo: ask.to }, + ask.updatedAt, + ); + return ask; + }); + app.get('/asks/:id', async (request, reply) => { + const { id } = request.params as { id: string }; + try { + const { ask, body } = getAsk(cwd, id); + return { ask, body }; + } catch { + return notFound(reply, 'Ask'); + } + }); + app.post('/asks/:id/answer', async (request, reply) => { + const { id } = request.params as { id: string }; + const { text, by } = (request.body ?? {}) as { text?: string; by?: string }; + try { + const ask = answerAsk(cwd, id, text ?? '', by); + emitChange( + { type: 'ask', action: 'updated', id: ask.id, title: `${ask.from} → ${ask.to}`, status: ask.status, assignedTo: ask.to }, + ask.updatedAt, + ); + return ask; + } catch (err) { + return badRequest(reply, err instanceof Error ? err.message : 'Ask not found'); + } + }); + app.post('/asks/:id/escalate', async (request, reply) => { + const { id } = request.params as { id: string }; + const { note } = (request.body ?? {}) as { note?: string }; + try { + const ask = escalateAsk(cwd, id, note); + emitChange( + { type: 'ask', action: 'updated', id: ask.id, title: `${ask.from} → ${ask.to}`, status: ask.status, assignedTo: ask.escalatedTo }, + ask.updatedAt, + ); + return ask; + } catch (err) { + return badRequest(reply, err instanceof Error ? err.message : 'Ask not found'); + } + }); + // ─── Memory ────────────────────────────────────────────────────────────── app.get('/memory', async () => listMemory(cwd)); app.post('/memory', async (request, reply) => { diff --git a/tests/ask-wait.test.ts b/tests/ask-wait.test.ts new file mode 100644 index 0000000..32debaf --- /dev/null +++ b/tests/ask-wait.test.ts @@ -0,0 +1,93 @@ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { mkdtempSync, rmSync } from 'fs'; +import { tmpdir } from 'os'; +import { join } from 'path'; +import { init } from '../src/cli/commands/init.js'; +import { buildApp } from '../src/server/index.js'; +import { startServer } from '../src/server/index.js'; +import { waitForAsk } from '../src/cli/commands/ask.js'; +import type { Ask } from '../src/core/schema.js'; + +describe('ask routes (TSK-0118)', () => { + let cwd: string; + let app: ReturnType; + + beforeEach(() => { + cwd = mkdtempSync(join(tmpdir(), 'ah-ask-routes-')); + init(cwd, { projectName: 'ask-routes', yes: true }); + app = buildApp(cwd); + }); + + afterEach(() => { + rmSync(cwd, { recursive: true, force: true }); + }); + + it('POST/GET/answer/escalate roundtrip', async () => { + const created = JSON.parse((await app.inject({ method: 'POST', url: '/asks', payload: { from: 'kimi', question: 'q?' } })).payload) as Ask; + expect(created.id).toBe('ASK-0001'); + expect(created.to).toBe('claude'); + expect(created.status).toBe('pending'); + + // GET /asks — the data source agenthub_work's architect branch surfaces. + const pending = JSON.parse((await app.inject({ method: 'GET', url: '/asks?status=pending' })).payload) as Ask[]; + expect(pending.map((a) => a.id)).toContain('ASK-0001'); + + const answered = JSON.parse((await app.inject({ method: 'POST', url: '/asks/ASK-0001/answer', payload: { text: 'yes', by: 'claude' } })).payload) as Ask; + expect(answered.status).toBe('answered'); + expect(answered.answer).toBe('yes'); + + const created2 = JSON.parse((await app.inject({ method: 'POST', url: '/asks', payload: { from: 'kimi', question: 'ship?' } })).payload) as Ask; + const escalated = JSON.parse((await app.inject({ method: 'POST', url: `/asks/${created2.id}/escalate`, payload: { note: 'ceo call' } })).payload) as Ask; + expect(escalated.status).toBe('escalated'); + expect(escalated.escalatedTo).toBe('ceo'); + }); +}); + +describe('ask --wait (server SSE)', () => { + let cwd: string; + let server: Awaited>; + + beforeEach(async () => { + cwd = mkdtempSync(join(tmpdir(), 'ah-ask-wait-')); + init(cwd, { projectName: 'ask-wait', yes: true }); + server = await startServer(cwd, { host: '127.0.0.1', port: 0 }); + }); + + afterEach(async () => { + await server.app.close().catch(() => undefined); + rmSync(cwd, { recursive: true, force: true }); + }); + + it('resolves with the answer when the architect answers', async () => { + const created = (await fetch(`${server.url}/asks`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ from: 'kimi', question: 'which driver?' }), + }).then((r) => r.json())) as Ask; + + const waitP = waitForAsk(server.url, created.id, 4); + + await new Promise((r) => setTimeout(r, 200)); + await fetch(`${server.url}/asks/${created.id}/answer`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ text: 'node:sqlite', by: 'claude' }), + }); + + const settled = await waitP; + expect(settled).not.toBeNull(); + expect(settled?.status).toBe('answered'); + expect(settled?.answer).toBe('node:sqlite'); + }, 7000); + + it('times out cleanly when there is no answer', async () => { + const created = (await fetch(`${server.url}/asks`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ from: 'kimi', question: 'no answer coming' }), + }).then((r) => r.json())) as Ask; + + const settled = await waitForAsk(server.url, created.id, 1); + expect(settled).toBeNull(); + }, 4000); +}); diff --git a/tests/askService.test.ts b/tests/askService.test.ts new file mode 100644 index 0000000..2e2c9ec --- /dev/null +++ b/tests/askService.test.ts @@ -0,0 +1,80 @@ +import { describe, it, expect, beforeEach, afterEach } from 'vitest'; +import { mkdtempSync, rmSync } from 'fs'; +import { tmpdir } from 'os'; +import { join } from 'path'; +import { init } from '../src/cli/commands/init.js'; +import { createAsk, answerAsk, escalateAsk, listAsks, getAsk } from '../src/core/services/askService.js'; +import { askCreate, askAnswer, askEscalate } from '../src/cli/commands/ask.js'; +import { addMemory, searchMemory } from '../src/core/services/memoryService.js'; + +describe('askService (TSK-0118)', () => { + let cwd: string; + + beforeEach(async () => { + cwd = mkdtempSync(join(tmpdir(), 'ah-ask-')); + await init(cwd, { yes: true, projectName: 'test' }); + }); + + afterEach(() => { + rmSync(cwd, { recursive: true, force: true }); + }); + + it('routes to the architect (never the CEO) by default', () => { + const ask = createAsk(cwd, { from: 'kimi', question: 'Which DB driver?' }); + expect(ask.id).toMatch(/^ASK-\d{4}$/); + expect(ask.to).toBe('claude'); // config.roles.architect.preferredAgent + expect(ask.status).toBe('pending'); + }); + + it('refuses to route to the CEO — reroutes to the architect', () => { + const ask = createAsk(cwd, { from: 'kimi', to: 'ceo', question: 'ship it?' }); + expect(ask.to).toBe('claude'); + }); + + it('answerAsk closes it as answered with the answer + author', () => { + const ask = createAsk(cwd, { from: 'kimi', question: 'Which DB driver?' }); + const answered = answerAsk(cwd, ask.id, 'node:sqlite', 'claude'); + expect(answered.status).toBe('answered'); + expect(answered.answer).toBe('node:sqlite'); + expect(answered.answeredBy).toBe('claude'); + expect(getAsk(cwd, ask.id).ask.status).toBe('answered'); + }); + + it('escalateAsk flips to escalated → ceo (single channel, no second ask)', () => { + const ask = createAsk(cwd, { from: 'kimi', question: 'Publish to OSS?' }); + const escalated = escalateAsk(cwd, ask.id, 'OSS decision — CEO call'); + expect(escalated.status).toBe('escalated'); + expect(escalated.escalatedTo).toBe('ceo'); + }); + + it('listAsks filters by status and recipient', () => { + const a1 = createAsk(cwd, { from: 'kimi', question: 'q1' }); + createAsk(cwd, { from: 'codex', question: 'q2' }); + answerAsk(cwd, a1.id, 'yes'); + + expect(listAsks(cwd)).toHaveLength(2); + expect(listAsks(cwd, { status: 'pending' })).toHaveLength(1); + expect(listAsks(cwd, { to: 'claude' })).toHaveLength(2); + expect(listAsks(cwd, { to: 'nobody' })).toHaveLength(0); + }); + + it('asks are NOT indexed in FTS5 (memory search never returns them)', () => { + createAsk(cwd, { from: 'kimi', question: 'znamqvist widget architecture' }); + addMemory(cwd, { title: 'note', content: 'znamqvist widget architecture' }); + + const hits = searchMemory(cwd, 'znamqvist'); + expect(hits.some((r) => r.type === 'ask')).toBe(false); // ask excluded from FTS + expect(hits.some((r) => r.type === 'memory')).toBe(true); // search still works + }); + + it('ask-cmd helpers create/answer/escalate through the service', () => { + const ask = askCreate(cwd, { from: 'kimi', question: 'via cmd?' }); + expect(ask.status).toBe('pending'); + askAnswer(cwd, ask.id, 'ok', 'claude'); + expect(getAsk(cwd, ask.id).ask.status).toBe('answered'); + + const ask2 = askCreate(cwd, { from: 'kimi', question: 'escalate me' }); + askEscalate(cwd, ask2.id, 'needs ceo'); + expect(getAsk(cwd, ask2.id).ask.status).toBe('escalated'); + }); +});