diff --git a/src/cli/commands/work.ts b/src/cli/commands/work.ts index 9dbf768..2a955d1 100644 --- a/src/cli/commands/work.ts +++ b/src/cli/commands/work.ts @@ -1,6 +1,7 @@ import { parseSSEBuffer } from './watch.js'; import { announceAgent, findAddressedOpenTask, claimAndPrintTask, type AgentContext } from './start.js'; import { discoverServer as discoverHubServer } from '../../discovery.js'; +import { remoteClient } from '../remoteClient.js'; /** * `agenthub work --agent --role ` — the auto-claim primitive @@ -20,6 +21,34 @@ interface WorkAgentContext extends AgentContext { reconnectBackoffMs?: number[]; } +/** + * Fetch + print + mark-read the agent's unread messages. Returns how many were + * surfaced. This is what lets the work loop wake on an architect follow-up / + * question (TSK-0119): after a `task review` submit the implementer re-arms + * `work` and stays reachable — a reopen or a new assignment wakes it via a task + * event, and a plain message wakes it here instead of leaving it dormant. + */ +async function drainAgentMessages(ctx: WorkAgentContext): Promise { + if (!ctx.serverUrl) return 0; + let msgs; + try { + msgs = await remoteClient.getInbox(ctx.serverUrl, ctx.agent, true); + } catch { + return 0; + } + if (!msgs.length) return 0; + console.log(`AgentHub: ${msgs.length} message${msgs.length === 1 ? '' : 's'} for ${ctx.agent}:`); + for (const m of msgs) { + console.log(` ${m.id} ${m.from} → ${m.to}${m.taskId ? ` [${m.taskId}]` : ''}: ${m.text}`); + try { + await remoteClient.markMessageRead(ctx.serverUrl, m.id); + } catch { + /* best-effort */ + } + } + return msgs.length; +} + export async function workAgent(ctx: WorkAgentContext): Promise { await announceAgent(ctx.serverUrl, ctx.agent, ctx.role); @@ -35,8 +64,12 @@ export async function workAgent(ctx: WorkAgentContext): Promise { return; } + // A message may already be waiting (architect followed up while we implemented + // + submitted). Surface it now instead of blocking past a pending question. + if (await drainAgentMessages(ctx)) return; + console.log( - `AgentHub: waiting for a task addressed to ${ctx.agent}…${ctx.timeoutSec ? ` (timeout ${ctx.timeoutSec}s)` : ''}`, + `AgentHub: waiting for a task or message addressed to ${ctx.agent}…${ctx.timeoutSec ? ` (timeout ${ctx.timeoutSec}s)` : ''}`, ); await waitAndClaim(ctx); } @@ -93,6 +126,17 @@ function waitAndClaim(ctx: WorkAgentContext): Promise { return true; }; + // Wake on an architect follow-up message (not just tasks): surface it and + // stop, so the implementer never sits dormant on a pending question. + const trySurfaceMessages = async (): Promise => { + ctx.serverUrl = serverUrl; + const n = await drainAgentMessages(ctx); + if (n === 0) return false; + if (timer) clearTimeout(timer); + finish(); + return true; + }; + const waitLoop = async () => { while (!settled && remainingMs(deadline) > 0) { controller = new AbortController(); @@ -101,9 +145,10 @@ function waitAndClaim(ctx: WorkAgentContext): Promise { 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'); - // Close the gap: a task may have appeared between the initial check and - // this subscription — check once more now that we're listening. + // Close the gap: a task or message may have appeared between the initial + // check and this subscription — check once more now that we're listening. if (await tryClaim()) return; + if (await trySurfaceMessages()) return; const reader = res.body.getReader(); const decoder = new TextDecoder(); @@ -126,6 +171,10 @@ function waitAndClaim(ctx: WorkAgentContext): Promise { if (events.some((e) => e.type === 'task')) { if (await tryClaim()) return; } + // A message event may be an architect follow-up/question for us. + if (events.some((e) => e.type === 'message')) { + if (await trySurfaceMessages()) return; + } } } catch (err: unknown) { if (settled || (err instanceof Error && err.name === 'AbortError')) return; diff --git a/src/mcp/server.ts b/src/mcp/server.ts index 5785bb7..312b415 100644 --- a/src/mcp/server.ts +++ b/src/mcp/server.ts @@ -215,6 +215,9 @@ export async function startMcpServer(cwd: string): Promise { : '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. ' + 'Report meaningful progress with agenthub_task_log while you work. ' + + 'ESPECIALLY after you submit with agenthub_task_review: do NOT end your turn — ' + + '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.'; const immediate = await finder(); if (immediate) return asText({ ...immediate, loop: LOOP }); diff --git a/tests/work.test.ts b/tests/work.test.ts index 003ba7e..75ab423 100644 --- a/tests/work.test.ts +++ b/tests/work.test.ts @@ -110,4 +110,74 @@ describe('agenthub work — wait then auto-claim (server SSE)', () => { expect(mine?.status).toBe('in_progress'); expect(mine?.assignedTo).toBe('kimi'); }, 8000); + + // ── TSK-0119: after a review submit the re-armed work loop must stay reachable ── + it('wakes on an architect follow-up message instead of going dormant', async () => { + // The implementer re-armed `work` after submitting; no task is addressed yet. + const workDone = workAgent({ + serverUrl: server.url, + projectCwd: cwd, + agent: 'kimi', + role: 'implementer', + timeoutSec: 4, + }); + + await new Promise((r) => setTimeout(r, 200)); + await fetch(`${server.url}/messages`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ from: 'claude', to: 'kimi', text: 'quick question about your submission' }), + }); + + await workDone; // resolves because the message woke the loop (not on timeout) + + // The loop drained + marked the message read — proving it woke on the message, + // not that it merely timed out (a timeout would leave it delivered/unread). + const inbox = (await fetch(`${server.url}/messages?agent=kimi`).then((r) => r.json())) as Array<{ text: string; status: string }>; + const m = inbox.find((x) => x.text.includes('quick question')); + expect(m).toBeDefined(); + expect(m?.status).toBe('read'); + }, 7000); + + it('wakes and re-claims when a submitted task is reopened', async () => { + // Seed a task addressed to kimi, claim it, submit for review. + await fetch(`${server.url}/tasks`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ title: 'kimi: build the widget', role: 'implementer' }), + }); + await fetch(`${server.url}/tasks/TSK-0001`, { + method: 'PATCH', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ status: 'in_progress', assignedTo: 'kimi' }), + }); + await fetch(`${server.url}/tasks/TSK-0001`, { + method: 'PATCH', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ status: 'review' }), + }); + + // Re-armed work loop: the task is in review, so nothing is addressed/open yet. + const workDone = workAgent({ + serverUrl: server.url, + projectCwd: cwd, + agent: 'kimi', + role: 'implementer', + timeoutSec: 4, + }); + + await new Promise((r) => setTimeout(r, 200)); + // Architect reopens (send back to implementer). + await fetch(`${server.url}/tasks/TSK-0001`, { + method: 'PATCH', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ status: 'open' }), + }); + + await workDone; + + const { task } = getTask(cwd, 'TSK-0001'); + expect(task.status).toBe('in_progress'); + expect(task.assignedTo).toBe('kimi'); + }, 7000); });