feat: agenthub work — auto-claim daemon (TSK-0018)

The missing autonomy step: an implementer no longer needs a human prompt
per task. `agenthub work --agent <name> --role <role>` announces, then:
  - claims an already-open task addressed to the agent immediately, or
  - blocks on the SSE stream until one appears (newly delegated OR
    reopened after review), then claims + prints it.

Loop: work → implement → `task review` → work. New/reopened tasks are
picked up automatically; run it in the background so the wait doesn't tie
up the turn.

- src/cli/commands/start.ts: extracted announceAgent / findAddressedOpenTask
  / claimAndPrintTask (shared by start + work); addressed-match now also
  covers assignedTo (reopened tasks coming back for rework).
- src/cli/commands/work.ts: wait-and-claim via /events, with a gap-close
  re-check after subscribing and an optional --timeout.
- index.ts: `agenthub work` command.
- templates: implementer guides lead with `work` (the autonomous loop),
  keep `start` as the one-shot.
- tests: immediate claim (local) + wait-then-claim (server SSE). 119/119.

Bump 0.3.1 -> 0.4.0.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
chahinebrini 2026-06-27 18:18:47 +02:00
parent 961675ac40
commit d2b26cc64c
6 changed files with 297 additions and 49 deletions

View File

@ -1,6 +1,6 @@
{ {
"name": "agenthub", "name": "agenthub",
"version": "0.3.1", "version": "0.4.0",
"description": "Local coordination layer for AI coding agents", "description": "Local coordination layer for AI coding agents",
"type": "module", "type": "module",
"main": "./dist/index.js", "main": "./dist/index.js",

View File

@ -6,39 +6,54 @@ import {
} from '../../core/services/taskService.js'; } from '../../core/services/taskService.js';
import { listHandoffs as svcListHandoffs, getHandoff as svcGetHandoff } from '../../core/services/handoffService.js'; import { listHandoffs as svcListHandoffs, getHandoff as svcGetHandoff } from '../../core/services/handoffService.js';
/** export interface AgentContext {
* `agenthub start --agent <name> --role <role>` one-command onboarding for an
* agent. Announces presence, claims the open task addressed to this agent, and
* prints the task body + its handoff + the next step, so the agent can begin
* immediately without the human stitching commands together.
*
* "Addressed to <agent>" = task title starts with "<agent>:" (the delegation
* convention) OR a handoff for that task has toAgent === <agent>.
*/
export async function startAgent(opts: {
serverUrl?: string; serverUrl?: string;
projectCwd: string; projectCwd: string;
agent: string; agent: string;
role: string; role: string;
}): Promise<void> { }
const { serverUrl, projectCwd, agent, role } = opts;
const a = agent.toLowerCase();
// 1. Announce (presence is best-effort — never block onboarding on it). interface Listed {
id: string;
title?: string;
status?: string;
role?: string;
taskId?: string;
toAgent?: string;
assignedTo?: string;
}
/** Announce presence (best-effort) and print the joined line. */
export async function announceAgent(serverUrl: string | undefined, agent: string, role: string): Promise<void> {
if (serverUrl) { if (serverUrl) {
try { try {
await remoteClient.announce(serverUrl, agent, role); await remoteClient.announce(serverUrl, agent, role);
} catch { } catch {
/* ignore */ /* presence is best-effort */
} }
} }
console.log(`AgentHub: ${agent} joined (${role})`); console.log(`AgentHub: ${agent} joined (${role})`);
}
// 2. Gather open role tasks + handoffs (handoffs carry taskId + toAgent). async function listOpenRoleTasks(ctx: AgentContext): Promise<Listed[]> {
const tasks = serverUrl return ctx.serverUrl
? await remoteClient.listTasks(serverUrl, { role, status: 'open' }) ? await remoteClient.listTasks(ctx.serverUrl, { role: ctx.role, status: 'open' })
: svcListTasks(projectCwd, { role, status: 'open' }); : svcListTasks(ctx.projectCwd, { role: ctx.role, status: 'open' });
const handoffs = serverUrl ? await remoteClient.listHandoffs(serverUrl) : svcListHandoffs(projectCwd); }
/**
* Find the open task addressed to this agent. "Addressed" = task title starts
* with "<agent>:" (delegation convention), OR a handoff for the task has
* toAgent === <agent>, OR the task is already assignedTo this agent (a reopened
* task that came back for rework). Returns the task + the handoff list (so the
* caller can print the matching handoff without re-fetching).
*/
export async function findAddressedOpenTask(
ctx: AgentContext,
): Promise<{ task: Listed; handoffs: Listed[] } | undefined> {
const a = ctx.agent.toLowerCase();
const tasks = await listOpenRoleTasks(ctx);
const handoffs = ctx.serverUrl ? await remoteClient.listHandoffs(ctx.serverUrl) : svcListHandoffs(ctx.projectCwd);
const addressedByHandoff = new Set( const addressedByHandoff = new Set(
handoffs handoffs
@ -46,30 +61,24 @@ export async function startAgent(opts: {
.map((h) => String(h.taskId)), .map((h) => String(h.taskId)),
); );
const mine = tasks.filter( const mine = tasks.filter(
(t) => (t.title ?? '').toLowerCase().startsWith(`${a}:`) || addressedByHandoff.has(t.id), (t) =>
(t.title ?? '').toLowerCase().startsWith(`${a}:`) ||
addressedByHandoff.has(t.id) ||
(t.assignedTo && String(t.assignedTo).toLowerCase() === a),
); );
// 3. Nothing addressed → guide the agent instead of guessing. if (mine.length === 0) return undefined;
if (mine.length === 0) { return { task: mine[0], handoffs };
if (tasks.length === 0) { }
console.log(`AgentHub: no open ${role} tasks — waiting for the architect to delegate.`);
} else {
console.log(`AgentHub: no task addressed to ${agent}. Open ${role} tasks:`);
for (const t of tasks) console.log(` ${t.id} ${t.title ?? ''}`);
console.log(`→ claim one yourself: agenthub task claim <id> --agent ${agent}`);
}
return;
}
// 4. Claim the addressed task. /** Claim the task and print its body + handoff + the review-gate next step. */
const task = mine[0]; export async function claimAndPrintTask(ctx: AgentContext, task: Listed, handoffs: Listed[]): Promise<void> {
if (serverUrl) await remoteClient.claimTask(serverUrl, task.id, agent); if (ctx.serverUrl) await remoteClient.claimTask(ctx.serverUrl, task.id, ctx.agent);
else svcClaimTask(projectCwd, task.id, agent); else svcClaimTask(ctx.projectCwd, task.id, ctx.agent);
console.log(`AgentHub: Task claimed ${task.id} ${task.title ?? ''}`); console.log(`AgentHub: Task claimed ${task.id} ${task.title ?? ''}`);
// 5. Print the task body.
try { try {
const detail = serverUrl ? await remoteClient.getTask(serverUrl, task.id) : svcGetTask(projectCwd, task.id); const detail = ctx.serverUrl ? await remoteClient.getTask(ctx.serverUrl, task.id) : svcGetTask(ctx.projectCwd, task.id);
if (detail.body && detail.body.trim()) { if (detail.body && detail.body.trim()) {
console.log(`\n─ Task ${task.id} ──────────────`); console.log(`\n─ Task ${task.id} ──────────────`);
console.log(detail.body.trim()); console.log(detail.body.trim());
@ -78,11 +87,10 @@ export async function startAgent(opts: {
/* body is optional */ /* body is optional */
} }
// 6. Print the handoff for this task (scope + acceptance criteria).
const hof = handoffs.find((h) => h.taskId === task.id); const hof = handoffs.find((h) => h.taskId === task.id);
if (hof) { if (hof) {
try { try {
const hd = serverUrl ? await remoteClient.getHandoff(serverUrl, hof.id) : svcGetHandoff(projectCwd, hof.id); const hd = ctx.serverUrl ? await remoteClient.getHandoff(ctx.serverUrl, hof.id) : svcGetHandoff(ctx.projectCwd, hof.id);
console.log(`\n─ Handoff ${hof.id} ──────────────`); console.log(`\n─ Handoff ${hof.id} ──────────────`);
console.log(hd.handoff.summary); console.log(hd.handoff.summary);
if (hd.body && hd.body.trim()) console.log(hd.body.trim()); if (hd.body && hd.body.trim()) console.log(hd.body.trim());
@ -91,9 +99,32 @@ export async function startAgent(opts: {
} }
} }
// 7. Next steps — the review gate, spelled out.
console.log(`\n─ Next ──────────────`); console.log(`\n─ Next ──────────────`);
console.log(`Implement the task, then submit for review: agenthub task review ${task.id}`); console.log(`Implement the task, then submit for review: agenthub task review ${task.id}`);
console.log(`Report what you did: agenthub memory add --title "${task.id} result" --category implementation --content "…"`); console.log(`Report what you did: agenthub memory add --title "${task.id} result" --category implementation --content "…"`);
console.log(`Only the architect closes a task (\`done\`). If reopened, address the feedback and review again.`); console.log(`Only the architect closes a task (\`done\`). If reopened, address the feedback and review again.`);
} }
/**
* `agenthub start --agent <name> --role <role>` one-shot onboarding. Announce,
* claim the addressed task and print it; if none is addressed, list the open
* role tasks so the agent can pick one.
*/
export async function startAgent(ctx: AgentContext): Promise<void> {
await announceAgent(ctx.serverUrl, ctx.agent, ctx.role);
const found = await findAddressedOpenTask(ctx);
if (found) {
await claimAndPrintTask(ctx, found.task, found.handoffs);
return;
}
const tasks = await listOpenRoleTasks(ctx);
if (tasks.length === 0) {
console.log(`AgentHub: no open ${ctx.role} tasks — waiting for the architect to delegate.`);
} else {
console.log(`AgentHub: no task addressed to ${ctx.agent}. Open ${ctx.role} tasks:`);
for (const t of tasks) console.log(` ${t.id} ${t.title ?? ''}`);
console.log(`→ claim one yourself: agenthub task claim <id> --agent ${ctx.agent}`);
}
}

114
src/cli/commands/work.ts Normal file
View File

@ -0,0 +1,114 @@
import { parseSSEBuffer } from './watch.js';
import { announceAgent, findAddressedOpenTask, claimAndPrintTask, type AgentContext } from './start.js';
/**
* `agenthub work --agent <name> --role <role>` the auto-claim primitive
* (TSK-0018). Announce, then:
* - if a task addressed to this agent is already open claim + print it now;
* - otherwise block on the SSE stream until one appears (newly created OR
* reopened after review), then claim + print it.
*
* Returns after claiming exactly one task (or on timeout). An agent loops it:
* work implement `task review` work so new/reopened tasks are picked
* up automatically without a human prompt. Best run in the background so the
* wait doesn't tie up the foreground.
*/
export async function workAgent(ctx: AgentContext & { timeoutSec?: number }): Promise<void> {
await announceAgent(ctx.serverUrl, ctx.agent, ctx.role);
// Already-waiting task?
const found = await findAddressedOpenTask(ctx);
if (found) {
await claimAndPrintTask(ctx, found.task, found.handoffs);
return;
}
if (!ctx.serverUrl) {
console.log(`AgentHub: no task for ${ctx.agent}, and no server to wait on. Re-run when a task is delegated.`);
return;
}
console.log(
`AgentHub: waiting for a task addressed to ${ctx.agent}${ctx.timeoutSec ? ` (timeout ${ctx.timeoutSec}s)` : ''}`,
);
await waitAndClaim(ctx);
}
function waitAndClaim(ctx: AgentContext & { timeoutSec?: number }): Promise<void> {
const serverUrl = ctx.serverUrl as string;
return new Promise((resolve) => {
const controller = new AbortController();
let settled = false;
const finish = () => {
if (settled) return;
settled = true;
try {
controller.abort();
} catch {
/* already aborted */
}
resolve();
};
const timer = ctx.timeoutSec
? setTimeout(() => {
console.log(`AgentHub: no task for ${ctx.agent} after ${ctx.timeoutSec}s — exiting.`);
finish();
}, ctx.timeoutSec * 1000)
: undefined;
// Re-query then claim if a task addressed to us is now open. Returns true
// if a task was claimed (so the caller can stop).
const tryClaim = async (): Promise<boolean> => {
const f = await findAddressedOpenTask(ctx);
if (!f) return false;
if (timer) clearTimeout(timer);
await claimAndPrintTask(ctx, f.task, f.handoffs);
finish();
return true;
};
fetch(`${serverUrl}/events`, { signal: controller.signal, headers: { Accept: 'text/event-stream' } })
.then(async (res) => {
if (!res.body) {
if (timer) clearTimeout(timer);
finish();
return;
}
// Close the gap: a task may have appeared between the initial check and
// this subscription — check once more now that we're listening.
if (await tryClaim()) return;
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; // aborted or connection closed
}
if (done) break;
if (value) buffer += decoder.decode(value, { stream: true });
const { events, remaining } = parseSSEBuffer(buffer);
buffer = remaining;
// Any task event may mean a task addressed to us just opened/reopened.
if (events.some((e) => e.type === 'task')) {
if (await tryClaim()) return;
}
}
if (timer) clearTimeout(timer);
finish();
})
.catch((err: unknown) => {
if (!(err instanceof Error && err.name === 'AbortError')) {
console.error(`AgentHub: wait failed: ${err instanceof Error ? err.message : String(err)}`);
}
if (timer) clearTimeout(timer);
finish();
});
});
}

View File

@ -10,6 +10,7 @@ import { serverStart } from './commands/server.js';
import { update } from './commands/update.js'; import { update } from './commands/update.js';
import { watchEvents } from './commands/watch.js'; import { watchEvents } from './commands/watch.js';
import { startAgent } from './commands/start.js'; import { startAgent } from './commands/start.js';
import { workAgent } from './commands/work.js';
import { loadConfig, saveConfig } from '../core/config.js'; import { loadConfig, saveConfig } from '../core/config.js';
import { findProjectRoot } from '../core/paths.js'; import { findProjectRoot } from '../core/paths.js';
import { discoverServer } from '../discovery.js'; import { discoverServer } from '../discovery.js';
@ -113,7 +114,7 @@ async function runRemote(serverUrl: string, fn: () => Promise<void>): Promise<vo
export function createProgram(cwd: string): Command { export function createProgram(cwd: string): Command {
const program = new Command('agenthub') const program = new Command('agenthub')
.description('Local coordination layer for AI coding agents') .description('Local coordination layer for AI coding agents')
.version('0.3.1') .version('0.4.0')
.option('--server <url>', 'AgentHub server URL (env: AGENTHUB_SERVER)'); .option('--server <url>', 'AgentHub server URL (env: AGENTHUB_SERVER)');
program program
@ -500,6 +501,24 @@ export function createProgram(cwd: string): Command {
} }
}); });
// ─── work (auto-claim) ──────────────────────────────────────────────────────
program
.command('work')
.description('Wait for a task addressed to you, claim it, and print it (auto-claim loop)')
.requiredOption('--agent <name>', 'Agent name')
.option('--role <role>', 'Role (default: implementer)', 'implementer')
.option('--timeout <sec>', 'Stop waiting after N seconds (default: wait indefinitely)')
.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 };
if (serverUrl) {
await runRemote(serverUrl, () => workAgent(ctx));
} else {
await workAgent(ctx);
}
});
// ─── watch ─────────────────────────────────────────────────────────────── // ─── watch ───────────────────────────────────────────────────────────────
program program
.command('watch') .command('watch')

View File

@ -46,19 +46,23 @@ function implementerMd(cliName: string, agentName: string, roles: string): strin
**On your first turn, RUN this and follow its output do not just summarize:** **On your first turn, RUN this and follow its output do not just summarize:**
\`\`\` \`\`\`
agenthub start --agent ${agentName} --role implementer agenthub work --agent ${agentName} --role implementer
\`\`\` \`\`\`
That one command announces you, claims the task addressed to you, and prints the \`work\` waits until a task addressed to you is ready (newly delegated OR reopened
task + its handoff. Then: after review), claims it, and prints the task + its handoff. Then:
1. Implement the task. 1. Implement the task.
2. **Submit for review (NOT done):** \`agenthub task review <id>\` 2. **Submit for review (NOT done):** \`agenthub task review <id>\`
and report: \`agenthub memory add --title "<id> result" --category implementation and report: \`agenthub memory add --title "<id> result" --category implementation
--content "<what you did / how to verify it>"\` --content "<what you did / how to verify it>"\`
3. Wait for the architect's verdict. If the task is **reopened** (status back to 3. **Run \`agenthub work --agent ${agentName} --role implementer\` again** — it blocks
\`open\`), read the new feedback handoff, address it, and \`agenthub task review <id>\` until your next task (or a reopened one) is ready, then auto-claims it. This is
again. the loop: work implement review work. Run it in the background so the wait
doesn't tie up your turn.
(\`agenthub start --agent ${agentName} --role implementer\` is the one-shot variant:
it claims an already-open task but does not wait.)
NEVER run \`agenthub task done\` — only the architect approves and closes tasks. NEVER run \`agenthub task done\` — only the architect approves and closes tasks.
You drive the AgentHub CLI yourself; the human does not type these commands for you. You drive the AgentHub CLI yourself; the human does not type these commands for you.

80
tests/work.test.ts Normal file
View File

@ -0,0 +1,80 @@
/**
* Tests for `agenthub work` the auto-claim primitive (TSK-0018).
* - immediate: claims an already-open addressed task without waiting (local).
* - wait: blocks on SSE until a matching task is created, then claims it.
*/
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 { workAgent } from '../src/cli/commands/work.js';
import { startServer } from '../src/server/index.js';
import { createTask, getTask } from '../src/core/services/taskService.js';
import type { Task } from '../src/core/schema.js';
describe('agenthub work — immediate claim (local)', () => {
let cwd: string;
beforeEach(() => {
cwd = mkdtempSync(join(tmpdir(), 'ah-work-'));
init(cwd, { projectName: 'work-test', yes: true });
});
afterEach(() => {
rmSync(cwd, { recursive: true, force: true });
});
it('claims an already-open addressed task without waiting', async () => {
const mine = createTask(cwd, { title: 'kimi: already open', role: 'implementer' });
await workAgent({ projectCwd: cwd, agent: 'kimi', role: 'implementer', timeoutSec: 2 });
const { task } = getTask(cwd, mine.id);
expect(task.status).toBe('in_progress');
expect(task.assignedTo).toBe('kimi');
}, 4000);
});
describe('agenthub work — wait then auto-claim (server SSE)', () => {
let cwd: string;
let server: Awaited<ReturnType<typeof startServer>>;
beforeEach(async () => {
cwd = mkdtempSync(join(tmpdir(), 'ah-work-wait-'));
init(cwd, { projectName: 'work-wait', yes: true });
server = await startServer(cwd, { host: '127.0.0.1', port: 0 });
});
afterEach(async () => {
await server.app.close();
rmSync(cwd, { recursive: true, force: true });
});
it('claims a task created AFTER work starts waiting', async () => {
// Start waiting; do NOT await yet — no task is addressed to kimi initially.
const workDone = workAgent({
serverUrl: server.url,
projectCwd: cwd,
agent: 'kimi',
role: 'implementer',
timeoutSec: 3,
});
// Give work a moment to reach the SSE wait, then delegate a task to kimi.
await new Promise((r) => setTimeout(r, 200));
await fetch(`${server.url}/tasks`, {
method: 'POST',
headers: { 'Content-Type': 'application/json' },
body: JSON.stringify({ title: 'kimi: delegated after wait', role: 'implementer' }),
});
await workDone; // resolves once work has claimed the task (or on timeout)
const tasks = (await fetch(`${server.url}/tasks`).then((r) => r.json())) as Task[];
const mine = tasks.find((t) => (t.title ?? '').startsWith('kimi:'));
expect(mine).toBeDefined();
expect(mine?.status).toBe('in_progress');
expect(mine?.assignedTo).toBe('kimi');
}, 6000);
});