fix(agenthub): TSK-0119 — implementer stays reachable after a review submit (dormancy fix)

- work.ts: the CLI work loop now wakes on architect follow-up MESSAGES, not just
  tasks. drainAgentMessages() surfaces + marks-read the agent's inbox; wired into
  waitAndClaim (SSE message events) and as an upfront check in workAgent. reopen +
  next-assign already wake via task events; a plain message no longer leaves the
  re-armed loop dormant.
- mcp/server.ts: implementer agenthub_work LOOP reminder now explicitly instructs
  re-arming after agenthub_task_review so the architect's approval/reopen/message
  wakes it (waitForTask already wakes on task+message).
- tests: work.test.ts +message-wake +reopen-wake

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
chahinebrini 2026-07-12 01:08:59 +02:00
parent 2c2193ec55
commit 941c20d12a
3 changed files with 125 additions and 3 deletions

View File

@ -1,6 +1,7 @@
import { parseSSEBuffer } from './watch.js'; import { parseSSEBuffer } from './watch.js';
import { announceAgent, findAddressedOpenTask, claimAndPrintTask, type AgentContext } from './start.js'; import { announceAgent, findAddressedOpenTask, claimAndPrintTask, type AgentContext } from './start.js';
import { discoverServer as discoverHubServer } from '../../discovery.js'; import { discoverServer as discoverHubServer } from '../../discovery.js';
import { remoteClient } from '../remoteClient.js';
/** /**
* `agenthub work --agent <name> --role <role>` the auto-claim primitive * `agenthub work --agent <name> --role <role>` the auto-claim primitive
@ -20,6 +21,34 @@ interface WorkAgentContext extends AgentContext {
reconnectBackoffMs?: number[]; 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<number> {
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<void> { export async function workAgent(ctx: WorkAgentContext): Promise<void> {
await announceAgent(ctx.serverUrl, ctx.agent, ctx.role); await announceAgent(ctx.serverUrl, ctx.agent, ctx.role);
@ -35,8 +64,12 @@ export async function workAgent(ctx: WorkAgentContext): Promise<void> {
return; 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( 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); await waitAndClaim(ctx);
} }
@ -93,6 +126,17 @@ function waitAndClaim(ctx: WorkAgentContext): Promise<void> {
return true; 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<boolean> => {
ctx.serverUrl = serverUrl;
const n = await drainAgentMessages(ctx);
if (n === 0) return false;
if (timer) clearTimeout(timer);
finish();
return true;
};
const waitLoop = async () => { const waitLoop = async () => {
while (!settled && remainingMs(deadline) > 0) { while (!settled && remainingMs(deadline) > 0) {
controller = new AbortController(); controller = new AbortController();
@ -101,9 +145,10 @@ function waitAndClaim(ctx: WorkAgentContext): Promise<void> {
const res = await fetch(`${serverUrl}/events`, { signal: controller.signal, headers: { Accept: 'text/event-stream' } }); 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 (!res.body) throw new Error('SSE response has no body');
// Close the gap: a task may have appeared between the initial check and // Close the gap: a task or message may have appeared between the initial
// this subscription — check once more now that we're listening. // check and this subscription — check once more now that we're listening.
if (await tryClaim()) return; if (await tryClaim()) return;
if (await trySurfaceMessages()) return;
const reader = res.body.getReader(); const reader = res.body.getReader();
const decoder = new TextDecoder(); const decoder = new TextDecoder();
@ -126,6 +171,10 @@ function waitAndClaim(ctx: WorkAgentContext): Promise<void> {
if (events.some((e) => e.type === 'task')) { if (events.some((e) => e.type === 'task')) {
if (await tryClaim()) return; 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) { } catch (err: unknown) {
if (settled || (err instanceof Error && err.name === 'AbortError')) return; if (settled || (err instanceof Error && err.name === 'AbortError')) return;

View File

@ -215,6 +215,9 @@ export async function startMcpServer(cwd: string): Promise<void> {
: 'STAY REACHABLE: the instant you finish handling this, call agenthub_work again — ' + : '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. ' + '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. ' + '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.'; 'Never end your turn without relaunching agenthub_work.';
const immediate = await finder(); const immediate = await finder();
if (immediate) return asText({ ...immediate, loop: LOOP }); if (immediate) return asText({ ...immediate, loop: LOOP });

View File

@ -110,4 +110,74 @@ describe('agenthub work — wait then auto-claim (server SSE)', () => {
expect(mine?.status).toBe('in_progress'); expect(mine?.status).toBe('in_progress');
expect(mine?.assignedTo).toBe('kimi'); expect(mine?.assignedTo).toBe('kimi');
}, 8000); }, 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);
}); });