diff --git a/.changeset/plan-publication-lane.md b/.changeset/plan-publication-lane.md new file mode 100644 index 0000000..1f8c854 --- /dev/null +++ b/.changeset/plan-publication-lane.md @@ -0,0 +1,7 @@ +--- +"@colony/core": minor +"@colony/mcp-server": minor +"@colony/worker": minor +--- + +Add a plan publication lane on top of the existing task-thread + spec primitives. `task_plan_publish` writes a spec change document and opens one task thread per sub-task on `spec//sub-N` branches, linking them via `metadata.parent_plan_slug`. Independent sub-tasks must not share file scopes; sequence overlapping work via `depends_on` (zero-based, must point at earlier indices). `task_plan_list` returns plan-level rollups with sub-task counts (`available | claimed | completed | blocked`) and a `next_available` list of unblocked, unclaimed sub-tasks; filterable by `repo_root`, `only_with_available_subtasks`, and `capability_match`. `task_plan_claim_subtask` claims an available sub-task race-safely (scan-before-stamp inside a SQLite transaction so two concurrent claims serialize through the write lock — first wins, second sees the prior claim observation and rejects with `PLAN_SUBTASK_NOT_AVAILABLE`); on success it joins the caller to the sub-task thread and activates file claims. `task_plan_complete_subtask` releases file claims and stamps a completion observation; downstream sub-tasks become available automatically. New observation kinds: `plan-subtask` (initial advertisement) and `plan-subtask-claim` (lifecycle transitions). New worker route `GET /api/colony/plans` exposes the same rollup to the read-only viewer. No schema migration; the lane composes over existing `task_thread` and `@colony/spec` primitives. diff --git a/apps/mcp-server/src/server.ts b/apps/mcp-server/src/server.ts index 4619151..e5c5c58 100644 --- a/apps/mcp-server/src/server.ts +++ b/apps/mcp-server/src/server.ts @@ -12,6 +12,7 @@ import * as handoff from './tools/handoff.js'; import { installActiveSessionHeartbeat } from './tools/heartbeat.js'; import * as hivemind from './tools/hivemind.js'; import * as message from './tools/message.js'; +import * as plan from './tools/plan.js'; import * as profile from './tools/profile.js'; import * as proposal from './tools/proposal.js'; import * as recall from './tools/recall.js'; @@ -74,6 +75,7 @@ export function buildServer(store: MemoryStore, settings: Settings): McpServer { profile.register(server, ctx); wake.register(server, ctx); message.register(server, ctx); + plan.register(server, ctx); recall.register(server, ctx); // Spec-driven dev lane (@colony/spec). Adds spec_read, spec_change_open, diff --git a/apps/mcp-server/src/tools/plan.ts b/apps/mcp-server/src/tools/plan.ts new file mode 100644 index 0000000..b43daa1 --- /dev/null +++ b/apps/mcp-server/src/tools/plan.ts @@ -0,0 +1,381 @@ +import { + type SubtaskInfo, + TaskThread, + areDepsMet, + listPlans, + readSubtaskByBranch, +} from '@colony/core'; +import { SpecRepository } from '@colony/spec'; +import type { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js'; +import { z } from 'zod'; +import type { ToolContext } from './context.js'; +import { mcpErrorResponse } from './shared.js'; + +const SubtaskInputSchema = z.object({ + title: z.string().min(1), + description: z.string().min(1), + file_scope: z.array(z.string().min(1)).min(1), + depends_on: z.array(z.number().int().nonnegative()).optional(), + capability_hint: z + .enum(['ui_work', 'api_work', 'test_work', 'infra_work', 'doc_work']) + .optional(), +}); + +type SubtaskInput = z.infer; + +interface CodedError extends Error { + __code?: string; +} + +export function register(server: McpServer, ctx: ToolContext): void { + const { store } = ctx; + + server.tool( + 'task_plan_publish', + [ + 'Publish a multi-task plan. Creates a spec change document at', + 'openspec/changes//CHANGE.md, opens one task thread per sub-task on', + 'spec//sub-N branches, and links them via metadata.parent_plan_slug.', + 'Refuses to publish if independent sub-tasks have overlapping file scopes', + '(use depends_on to sequence overlapping work). Originating agent does NOT', + 'auto-join sub-tasks — claim via task_plan_claim_subtask.', + ].join(' '), + { + repo_root: z.string().min(1), + slug: z + .string() + .min(1) + .regex(/^[a-z0-9-]+$/, 'kebab-case only'), + session_id: z.string().min(1), + agent: z.string().min(1), + title: z.string().min(1), + problem: z.string().min(1).describe('Why this plan exists. Becomes spec change §problem.'), + acceptance_criteria: z + .array(z.string().min(1)) + .min(1) + .describe('What "done" looks like for the whole plan, not per sub-task.'), + subtasks: z + .array(SubtaskInputSchema) + .min(2) + .max(20) + .describe('At least 2 sub-tasks — if it is one task, use task_thread directly.'), + }, + async (args) => { + for (let i = 0; i < args.subtasks.length; i++) { + const subtask = args.subtasks[i]; + if (!subtask) continue; + for (const dep of subtask.depends_on ?? []) { + if (dep >= i) { + return mcpErrorResponse( + 'PLAN_INVALID_DEPENDENCY', + `sub-task ${i} depends on ${dep}; dependencies must point to earlier indices (no cycles)`, + ); + } + } + } + + const overlap = detectScopeOverlap(args.subtasks); + if (overlap) { + return mcpErrorResponse( + 'PLAN_SCOPE_OVERLAP', + `sub-tasks ${overlap.a} and ${overlap.b} share files [${overlap.shared.join(', ')}] without a depends_on edge between them`, + ); + } + + const repo = new SpecRepository({ repoRoot: args.repo_root, store }); + const proposal = renderProposal(args); + const opened = repo.openChange({ + slug: args.slug, + session_id: args.session_id, + agent: args.agent, + proposal, + }); + + const subtaskThreads = args.subtasks.map((subtask, index) => { + const branch = `spec/${args.slug}/sub-${index}`; + const thread = TaskThread.open(store, { + repo_root: args.repo_root, + branch, + session_id: args.session_id, + }); + store.addObservation({ + session_id: args.session_id, + task_id: thread.task_id, + kind: 'plan-subtask', + content: `${subtask.title}\n\n${subtask.description}`, + metadata: { + parent_plan_slug: args.slug, + parent_plan_title: args.title, + parent_spec_task_id: opened.task_id, + subtask_index: index, + file_scope: subtask.file_scope, + depends_on: subtask.depends_on ?? [], + capability_hint: subtask.capability_hint ?? null, + status: 'available', + }, + }); + return { + subtask_index: index, + branch, + task_id: thread.task_id, + title: subtask.title, + }; + }); + + return { + content: [ + { + type: 'text', + text: JSON.stringify({ + plan_slug: args.slug, + spec_task_id: opened.task_id, + spec_change_path: opened.path, + subtasks: subtaskThreads, + }), + }, + ], + }; + }, + ); + + server.tool( + 'task_plan_list', + 'List published plans. Returns plan-level rollup with sub-task counts by status (available/claimed/completed/blocked) and a next_available list of unblocked, unclaimed sub-tasks. Filter by repo_root, only_with_available_subtasks, or capability_match.', + { + repo_root: z.string().min(1).optional(), + only_with_available_subtasks: z.boolean().optional(), + capability_match: z + .enum(['ui_work', 'api_work', 'test_work', 'infra_work', 'doc_work']) + .optional(), + limit: z.number().int().positive().max(50).optional(), + }, + async (args) => { + const plans = listPlans(store, { + ...(args.repo_root !== undefined ? { repo_root: args.repo_root } : {}), + ...(args.only_with_available_subtasks !== undefined + ? { only_with_available_subtasks: args.only_with_available_subtasks } + : {}), + ...(args.capability_match !== undefined ? { capability_match: args.capability_match } : {}), + ...(args.limit !== undefined ? { limit: args.limit } : {}), + }); + return { content: [{ type: 'text', text: JSON.stringify(plans) }] }; + }, + ); + + server.tool( + 'task_plan_claim_subtask', + 'Claim an available sub-task on a published plan. Joins you to the sub-task thread and activates file claims for the sub-task file_scope. Refuses if dependencies are not met or another agent already holds the claim.', + { + plan_slug: z.string().min(1), + subtask_index: z.number().int().nonnegative(), + session_id: z.string().min(1), + agent: z.string().min(1), + }, + async (args) => { + const branch = `spec/${args.plan_slug}/sub-${args.subtask_index}`; + const located = readSubtaskByBranch(store, branch); + if (!located) { + return mcpErrorResponse('PLAN_SUBTASK_NOT_FOUND', `no sub-task at ${branch}`); + } + + const allTasks = store.storage.listTasks(2000); + const siblings = allTasks + .filter((t) => { + const m = t.branch.match(/^spec\/([a-z0-9-]+)\/sub-(\d+)$/); + return Boolean(m && m[1] === args.plan_slug); + }) + .map((t) => readSubtaskByBranch(store, t.branch)) + .filter((s): s is { task_id: number; info: SubtaskInfo } => s !== null) + .map((s) => s.info); + + if (!areDepsMet(located.info, siblings)) { + const unmet = located.info.depends_on.filter((idx) => { + const dep = siblings.find((s) => s.subtask_index === idx); + return dep?.status !== 'completed'; + }); + return mcpErrorResponse( + 'PLAN_SUBTASK_DEPS_UNMET', + `dependencies not met: sub-tasks [${unmet.join(', ')}] are not completed`, + ); + } + + // Race-safe claim. Re-scan the claim observations inside a transaction + // so two concurrent claims serialize through SQLite's write lock; the + // first commit wins, the second reads its claim row and rejects. + try { + store.storage.transaction(() => { + const claimRows = store.storage.taskObservationsByKind( + located.task_id, + 'plan-subtask-claim', + 500, + ); + for (const row of claimRows) { + const meta = parseRowMeta(row.metadata); + if (meta.status === 'claimed' || meta.status === 'completed') { + const err: CodedError = new Error( + `sub-task is ${meta.status}${meta.session_id ? ` by ${meta.session_id}` : ''}`, + ); + err.__code = 'PLAN_SUBTASK_NOT_AVAILABLE'; + throw err; + } + } + store.addObservation({ + session_id: args.session_id, + task_id: located.task_id, + kind: 'plan-subtask-claim', + content: `${args.agent} claimed sub-task ${args.subtask_index} of plan ${args.plan_slug}`, + metadata: { + status: 'claimed', + session_id: args.session_id, + agent: args.agent, + plan_slug: args.plan_slug, + subtask_index: args.subtask_index, + }, + }); + const thread = new TaskThread(store, located.task_id); + thread.join(args.session_id, args.agent); + for (const file of located.info.file_scope) { + store.storage.claimFile({ + task_id: located.task_id, + file_path: file, + session_id: args.session_id, + }); + } + }); + } catch (err) { + const code = (err as CodedError).__code; + if (code === 'PLAN_SUBTASK_NOT_AVAILABLE') { + return mcpErrorResponse(code, (err as Error).message); + } + throw err; + } + + return { + content: [ + { + type: 'text', + text: JSON.stringify({ + task_id: located.task_id, + branch, + file_scope: located.info.file_scope, + }), + }, + ], + }; + }, + ); + + server.tool( + 'task_plan_complete_subtask', + 'Mark your claimed sub-task complete. Releases file claims; downstream sub-tasks (if any) become available.', + { + plan_slug: z.string().min(1), + subtask_index: z.number().int().nonnegative(), + session_id: z.string().min(1), + summary: z.string().min(1).describe('What landed. Surfaces in the parent plan rollup.'), + }, + async (args) => { + const branch = `spec/${args.plan_slug}/sub-${args.subtask_index}`; + const located = readSubtaskByBranch(store, branch); + if (!located) { + return mcpErrorResponse('PLAN_SUBTASK_NOT_FOUND', `no sub-task at ${branch}`); + } + if (located.info.status !== 'claimed') { + return mcpErrorResponse( + 'PLAN_SUBTASK_NOT_CLAIMED', + `sub-task is ${located.info.status}, not claimed`, + ); + } + if (located.info.claimed_by_session_id !== args.session_id) { + return mcpErrorResponse( + 'PLAN_SUBTASK_NOT_YOURS', + `sub-task is claimed by ${located.info.claimed_by_session_id ?? '(nobody)'}, not ${args.session_id}`, + ); + } + + store.storage.transaction(() => { + for (const file of located.info.file_scope) { + store.storage.releaseClaim({ + task_id: located.task_id, + file_path: file, + session_id: args.session_id, + }); + } + store.addObservation({ + session_id: args.session_id, + task_id: located.task_id, + kind: 'plan-subtask-claim', + content: args.summary, + metadata: { + status: 'completed', + session_id: args.session_id, + agent: located.info.claimed_by_agent ?? 'unknown', + plan_slug: args.plan_slug, + subtask_index: args.subtask_index, + completed_at: Date.now(), + }, + }); + }); + + return { + content: [{ type: 'text', text: JSON.stringify({ status: 'completed' }) }], + }; + }, + ); +} + +function detectScopeOverlap( + subtasks: SubtaskInput[], +): { a: number; b: number; shared: string[] } | null { + for (let i = 0; i < subtasks.length; i++) { + for (let j = i + 1; j < subtasks.length; j++) { + const a = subtasks[i]; + const b = subtasks[j]; + if (!a || !b) continue; + if (isDependentChain(subtasks, i, j) || isDependentChain(subtasks, j, i)) continue; + const shared = a.file_scope.filter((f) => b.file_scope.includes(f)); + if (shared.length > 0) return { a: i, b: j, shared }; + } + } + return null; +} + +function isDependentChain(subtasks: SubtaskInput[], from: number, to: number): boolean { + const visited = new Set(); + const stack = [from]; + while (stack.length > 0) { + const cur = stack.pop(); + if (cur === undefined || visited.has(cur)) continue; + visited.add(cur); + const deps = subtasks[cur]?.depends_on ?? []; + if (deps.includes(to)) return true; + stack.push(...deps); + } + return false; +} + +function renderProposal(args: { + title: string; + problem: string; + acceptance_criteria: string[]; + subtasks: SubtaskInput[]; +}): string { + const subtaskBlocks = args.subtasks + .map((s, i) => { + const deps = s.depends_on?.length ? ` (depends on: ${s.depends_on.join(', ')})` : ''; + return `### Sub-task ${i}: ${s.title}${deps}\n\n${s.description}\n\nFile scope: ${s.file_scope.join(', ')}`; + }) + .join('\n\n'); + const criteria = args.acceptance_criteria.map((c) => `- ${c}`).join('\n'); + return `# ${args.title}\n\n## Problem\n\n${args.problem}\n\n## Acceptance criteria\n\n${criteria}\n\n## Sub-tasks\n\n${subtaskBlocks}\n`; +} + +function parseRowMeta(raw: string | null): Record { + if (!raw) return {}; + try { + const parsed = JSON.parse(raw) as unknown; + return typeof parsed === 'object' && parsed !== null ? (parsed as Record) : {}; + } catch { + return {}; + } +} diff --git a/apps/mcp-server/src/tools/shared.ts b/apps/mcp-server/src/tools/shared.ts index c6c63a3..688483d 100644 --- a/apps/mcp-server/src/tools/shared.ts +++ b/apps/mcp-server/src/tools/shared.ts @@ -69,7 +69,17 @@ export function mcpError(err: unknown): { } export function mcpErrorResponse( - code: TaskThreadErrorCode | 'SPEC_TASK_NOT_FOUND' | 'SPEC_CHANGE_NOT_FOUND', + code: + | TaskThreadErrorCode + | 'SPEC_TASK_NOT_FOUND' + | 'SPEC_CHANGE_NOT_FOUND' + | 'PLAN_INVALID_DEPENDENCY' + | 'PLAN_SCOPE_OVERLAP' + | 'PLAN_SUBTASK_NOT_FOUND' + | 'PLAN_SUBTASK_DEPS_UNMET' + | 'PLAN_SUBTASK_NOT_AVAILABLE' + | 'PLAN_SUBTASK_NOT_CLAIMED' + | 'PLAN_SUBTASK_NOT_YOURS', error: string, ): { content: Array<{ type: 'text'; text: string }>; diff --git a/apps/mcp-server/test/plan.test.ts b/apps/mcp-server/test/plan.test.ts new file mode 100644 index 0000000..5a08be0 --- /dev/null +++ b/apps/mcp-server/test/plan.test.ts @@ -0,0 +1,331 @@ +import { mkdtempSync, rmSync, writeFileSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { defaultSettings } from '@colony/config'; +import { MemoryStore } from '@colony/core'; +import { Client } from '@modelcontextprotocol/sdk/client/index.js'; +import { InMemoryTransport } from '@modelcontextprotocol/sdk/inMemory.js'; +import { afterEach, beforeEach, describe, expect, it } from 'vitest'; +import { buildServer } from '../src/server.js'; + +let dataDir: string; +let repoRoot: string; +let store: MemoryStore; +let client: Client; + +interface PublishResult { + plan_slug: string; + spec_task_id: number; + spec_change_path: string; + subtasks: Array<{ subtask_index: number; branch: string; task_id: number; title: string }>; +} + +interface PlanRollup { + plan_slug: string; + title: string; + subtask_counts: Record; + next_available: Array<{ subtask_index: number; capability_hint: string | null }>; + subtasks: Array<{ subtask_index: number; status: string; claimed_by_session_id: string | null }>; +} + +interface ClaimResult { + task_id: number; + branch: string; + file_scope: string[]; +} + +async function call(name: string, args: Record): Promise { + const res = await client.callTool({ name, arguments: args }); + const text = (res.content as Array<{ type: string; text: string }>)[0]?.text ?? '{}'; + return JSON.parse(text) as T; +} + +async function callError( + name: string, + args: Record, +): Promise<{ code: string; error: string }> { + const res = await client.callTool({ name, arguments: args }); + expect(res.isError).toBe(true); + const text = (res.content as Array<{ type: string; text: string }>)[0]?.text ?? '{}'; + return JSON.parse(text) as { code: string; error: string }; +} + +function basicPublishArgs(overrides: Record = {}): Record { + return { + repo_root: repoRoot, + slug: 'add-widget-page', + session_id: 'A', + agent: 'claude', + title: 'Add widget page', + problem: 'No widget page exists yet; users have no entry point.', + acceptance_criteria: ['Widget page renders', 'Widget API returns rows'], + subtasks: [ + { + title: 'Build widget API', + description: 'Add GET /api/widgets that returns rows.', + file_scope: ['apps/api/src/widgets.ts'], + capability_hint: 'api_work', + }, + { + title: 'Build widget page', + description: 'Render the widget list with a card per row.', + file_scope: ['apps/frontend/src/pages/widgets.tsx'], + depends_on: [0], + capability_hint: 'ui_work', + }, + ], + ...overrides, + }; +} + +const MINIMAL_SPEC = `# SPEC + +## §G goal +Test fixture spec for plan publication tests. + +## §C constraints +- markdown only. + +## §I interfaces +- none + +## §V invariants +id|rule|cites +-|-|- +V1|placeholder|- + +## §T tasks +id|task|cites +-|-|- +T1|placeholder|V1 + +## §B bugs +id|bug|cites +-|-|- +`; + +beforeEach(async () => { + dataDir = mkdtempSync(join(tmpdir(), 'colony-plan-data-')); + repoRoot = mkdtempSync(join(tmpdir(), 'colony-plan-repo-')); + writeFileSync(join(repoRoot, 'SPEC.md'), MINIMAL_SPEC, 'utf8'); + store = new MemoryStore({ dbPath: join(dataDir, 'data.db'), settings: defaultSettings }); + store.startSession({ id: 'A', ide: 'claude-code', cwd: repoRoot }); + store.startSession({ id: 'B', ide: 'codex', cwd: repoRoot }); + store.startSession({ id: 'C', ide: 'claude-code', cwd: repoRoot }); + const server = buildServer(store, defaultSettings); + const [clientT, serverT] = InMemoryTransport.createLinkedPair(); + client = new Client({ name: 'test', version: '0.0.0' }); + await Promise.all([server.connect(serverT), client.connect(clientT)]); +}); + +afterEach(async () => { + await client.close(); + store.close(); + rmSync(dataDir, { recursive: true, force: true }); + rmSync(repoRoot, { recursive: true, force: true }); +}); + +describe('task_plan_publish', () => { + it('publishes a plan: writes spec change, opens one thread per sub-task, stamps metadata', async () => { + const result = await call('task_plan_publish', basicPublishArgs()); + expect(result.plan_slug).toBe('add-widget-page'); + expect(result.subtasks).toHaveLength(2); + expect(result.subtasks[0]?.branch).toBe('spec/add-widget-page/sub-0'); + expect(result.subtasks[1]?.branch).toBe('spec/add-widget-page/sub-1'); + expect(result.spec_change_path).toContain('openspec/changes/add-widget-page/CHANGE.md'); + }); + + it('rejects overlapping file scopes between independent sub-tasks', async () => { + const err = await callError( + 'task_plan_publish', + basicPublishArgs({ + slug: 'overlap-bad', + subtasks: [ + { + title: 'A', + description: 'a', + file_scope: ['apps/foo.ts'], + }, + { + title: 'B', + description: 'b', + file_scope: ['apps/foo.ts'], + }, + ], + }), + ); + expect(err.code).toBe('PLAN_SCOPE_OVERLAP'); + }); + + it('allows overlapping file scopes when sub-tasks are sequenced via depends_on', async () => { + const result = await call( + 'task_plan_publish', + basicPublishArgs({ + slug: 'overlap-ok', + subtasks: [ + { + title: 'A', + description: 'first', + file_scope: ['apps/foo.ts'], + }, + { + title: 'B', + description: 'second', + file_scope: ['apps/foo.ts'], + depends_on: [0], + }, + ], + }), + ); + expect(result.subtasks).toHaveLength(2); + }); + + it('rejects forward / self dependencies (cycle prevention)', async () => { + const err = await callError( + 'task_plan_publish', + basicPublishArgs({ + slug: 'cycle-bad', + subtasks: [ + { + title: 'A', + description: 'a', + file_scope: ['apps/a.ts'], + depends_on: [1], + }, + { + title: 'B', + description: 'b', + file_scope: ['apps/b.ts'], + }, + ], + }), + ); + expect(err.code).toBe('PLAN_INVALID_DEPENDENCY'); + }); +}); + +describe('task_plan_list', () => { + it('rolls up sub-task statuses and surfaces next_available respecting depends_on', async () => { + await call('task_plan_publish', basicPublishArgs()); + const plans = await call('task_plan_list', {}); + expect(plans).toHaveLength(1); + expect(plans[0]?.plan_slug).toBe('add-widget-page'); + expect(plans[0]?.subtask_counts.available).toBe(2); + // sub-1 depends on sub-0, so only sub-0 is in next_available initially + expect(plans[0]?.next_available.map((s) => s.subtask_index)).toEqual([0]); + }); + + it('filters by capability_match against next_available sub-tasks', async () => { + await call('task_plan_publish', basicPublishArgs()); + const apiPlans = await call('task_plan_list', { capability_match: 'api_work' }); + const uiPlans = await call('task_plan_list', { capability_match: 'ui_work' }); + expect(apiPlans).toHaveLength(1); + // sub-1 (ui_work) is not in next_available because its dep is unmet + expect(uiPlans).toHaveLength(0); + }); +}); + +describe('task_plan_claim_subtask', () => { + it('claims an available sub-task and activates file claims', async () => { + await call('task_plan_publish', basicPublishArgs()); + const claim = await call('task_plan_claim_subtask', { + plan_slug: 'add-widget-page', + subtask_index: 0, + session_id: 'B', + agent: 'codex', + }); + expect(claim.branch).toBe('spec/add-widget-page/sub-0'); + expect(claim.file_scope).toEqual(['apps/api/src/widgets.ts']); + + // List should show sub-0 as claimed + const plans = await call('task_plan_list', {}); + const sub0 = plans[0]?.subtasks.find((s) => s.subtask_index === 0); + expect(sub0?.status).toBe('claimed'); + expect(sub0?.claimed_by_session_id).toBe('B'); + }); + + it('rejects claim when dependencies are not yet completed', async () => { + await call('task_plan_publish', basicPublishArgs()); + const err = await callError('task_plan_claim_subtask', { + plan_slug: 'add-widget-page', + subtask_index: 1, + session_id: 'B', + agent: 'codex', + }); + expect(err.code).toBe('PLAN_SUBTASK_DEPS_UNMET'); + }); + + it('rejects a second claim on an already-claimed sub-task (race)', async () => { + // The load-bearing test for the lane: two agents racing on the same + // available sub-task. The transaction-based scan-before-stamp inside the + // tool handler must serialize them; the second one sees the first claim + // observation and rejects. + await call('task_plan_publish', basicPublishArgs()); + const args = (sid: string, agent: string) => ({ + plan_slug: 'add-widget-page', + subtask_index: 0, + session_id: sid, + agent, + }); + + // First claim wins. + const first = await call('task_plan_claim_subtask', args('B', 'codex')); + expect(first.task_id).toBeGreaterThan(0); + + // Second claim must fail. + const err = await callError('task_plan_claim_subtask', args('C', 'claude')); + expect(err.code).toBe('PLAN_SUBTASK_NOT_AVAILABLE'); + }); +}); + +describe('task_plan_complete_subtask', () => { + it('marks claimed sub-task complete and unblocks downstream sub-tasks', async () => { + await call('task_plan_publish', basicPublishArgs()); + await call('task_plan_claim_subtask', { + plan_slug: 'add-widget-page', + subtask_index: 0, + session_id: 'B', + agent: 'codex', + }); + const done = await call<{ status: string }>('task_plan_complete_subtask', { + plan_slug: 'add-widget-page', + subtask_index: 0, + session_id: 'B', + summary: 'Widget API landed: GET /api/widgets serving rows.', + }); + expect(done.status).toBe('completed'); + + // sub-1 should now be unblocked. + const plans = await call('task_plan_list', {}); + expect(plans[0]?.next_available.map((s) => s.subtask_index)).toEqual([1]); + expect(plans[0]?.subtask_counts.completed).toBe(1); + }); + + it('rejects completion when called by a non-owning session', async () => { + await call('task_plan_publish', basicPublishArgs()); + await call('task_plan_claim_subtask', { + plan_slug: 'add-widget-page', + subtask_index: 0, + session_id: 'B', + agent: 'codex', + }); + const err = await callError('task_plan_complete_subtask', { + plan_slug: 'add-widget-page', + subtask_index: 0, + session_id: 'C', + summary: 'sneaky completion', + }); + expect(err.code).toBe('PLAN_SUBTASK_NOT_YOURS'); + }); + + it('rejects completion on an unclaimed sub-task', async () => { + await call('task_plan_publish', basicPublishArgs()); + const err = await callError('task_plan_complete_subtask', { + plan_slug: 'add-widget-page', + subtask_index: 0, + session_id: 'B', + summary: 'nothing claimed', + }); + expect(err.code).toBe('PLAN_SUBTASK_NOT_CLAIMED'); + }); +}); diff --git a/apps/mcp-server/test/server.test.ts b/apps/mcp-server/test/server.test.ts index 2d93494..bb130cb 100644 --- a/apps/mcp-server/test/server.test.ts +++ b/apps/mcp-server/test/server.test.ts @@ -75,6 +75,10 @@ describe('MCP server', () => { 'task_message', 'task_message_mark_read', 'task_messages', + 'task_plan_claim_subtask', + 'task_plan_complete_subtask', + 'task_plan_list', + 'task_plan_publish', 'task_post', 'task_propose', 'task_reinforce', diff --git a/apps/worker/src/server.ts b/apps/worker/src/server.ts index ac34087..e7eb0a4 100644 --- a/apps/worker/src/server.ts +++ b/apps/worker/src/server.ts @@ -3,7 +3,7 @@ import { writeFileSync } from 'node:fs'; import { join } from 'node:path'; import { expand } from '@colony/compress'; import { type Settings, loadSettings, resolveDataDir } from '@colony/config'; -import { type HivemindOptions, MemoryStore, readHivemind } from '@colony/core'; +import { type HivemindOptions, MemoryStore, listPlans, readHivemind } from '@colony/core'; import { createEmbedder } from '@colony/embedding'; import { isMainEntry, removePidFile, writePidFile } from '@colony/process'; import { serve } from '@hono/node-server'; @@ -115,6 +115,21 @@ export function buildApp( }); }); + app.get('/api/colony/plans', (c) => { + const repoRoot = c.req.query('repo_root'); + const onlyAvailable = c.req.query('only_with_available_subtasks') === 'true'; + const capability = c.req.query('capability_match'); + const limit = c.req.query('limit') ? Number(c.req.query('limit')) : undefined; + return c.json( + listPlans(store, { + ...(repoRoot ? { repo_root: repoRoot } : {}), + ...(onlyAvailable ? { only_with_available_subtasks: true } : {}), + ...(capability ? { capability_match: capability } : {}), + ...(limit !== undefined ? { limit } : {}), + }), + ); + }); + app.get('/api/sessions/:id/observations', (c) => { const id = c.req.param('id'); const limit = Number(c.req.query('limit') ?? 200); diff --git a/docs/mcp.md b/docs/mcp.md index 1d2e481..cac7840 100644 --- a/docs/mcp.md +++ b/docs/mcp.md @@ -687,6 +687,109 @@ Validate, three-way-merge, and archive an in-flight change. Atomic: either the a `strategy` ∈ `three_way | refuse_on_conflict | last_writer_wins`. Returns `{ status, archived_path, merged_root_hash, conflicts, applied }`. On `refuse_on_conflict` with real conflicts, the call returns `status: 'refused'` and `isError: true` so the caller sees the conflict set without committing. +## `task_plan_publish` + +Publish a multi-task plan as a spec change with one task thread per sub-task. Sub-tasks live on `spec//sub-N` branches and link back via `metadata.parent_plan_slug`. The originating agent does **not** auto-join sub-tasks — publishing is advertising, not claiming. + +```json +{ + "name": "task_plan_publish", + "input": { + "repo_root": "/abs/repo", + "slug": "add-widget-page", + "session_id": "sess_abc", + "agent": "claude", + "title": "Add widget page", + "problem": "No widget page exists yet; users have no entry point.", + "acceptance_criteria": ["Widget page renders", "Widget API returns rows"], + "subtasks": [ + { + "title": "Build widget API", + "description": "Add GET /api/widgets that returns rows.", + "file_scope": ["apps/api/src/widgets.ts"], + "capability_hint": "api_work" + }, + { + "title": "Build widget page", + "description": "Render the widget list with a card per row.", + "file_scope": ["apps/frontend/src/pages/widgets.tsx"], + "depends_on": [0], + "capability_hint": "ui_work" + } + ] + } +} +``` + +Validation: + +- `subtasks` must contain at least 2 entries; for a single task use `task_thread` directly. +- `depends_on` indices are zero-based and must point to **earlier** indices (cycle prevention). +- Independent sub-tasks (no `depends_on` chain between them) cannot share `file_scope` entries. To overlap files, sequence the work via `depends_on`. + +Returns `{ plan_slug, spec_task_id, spec_change_path, subtasks: [{ subtask_index, branch, task_id, title }] }`. Errors: `PLAN_INVALID_DEPENDENCY`, `PLAN_SCOPE_OVERLAP`. + +## `task_plan_list` + +List published plans with a sub-task rollup. + +```json +{ + "name": "task_plan_list", + "input": { + "repo_root": "/abs/repo", + "only_with_available_subtasks": true, + "capability_match": "ui_work", + "limit": 25 + } +} +``` + +Returns `[{ plan_slug, repo_root, spec_task_id, title, created_at, subtask_counts: { available, claimed, completed, blocked }, subtasks: [...], next_available: [...] }]`. `next_available` is the list of sub-tasks whose status is `available` **and** whose `depends_on` chain is fully `completed`. `capability_match` filters plans where at least one sub-task in `next_available` has the matching `capability_hint`. + +## `task_plan_claim_subtask` + +Claim an available sub-task. The handler runs scan-before-stamp inside a SQLite transaction so two concurrent claims serialize through the write lock; the first commit wins, the second sees the prior claim observation and rejects. + +```json +{ + "name": "task_plan_claim_subtask", + "input": { + "plan_slug": "add-widget-page", + "subtask_index": 0, + "session_id": "sess_def", + "agent": "codex" + } +} +``` + +On success: joins the caller to the sub-task thread and activates file claims for every entry in the sub-task `file_scope`. Returns `{ task_id, branch, file_scope }`. Errors: `PLAN_SUBTASK_NOT_FOUND`, `PLAN_SUBTASK_DEPS_UNMET`, `PLAN_SUBTASK_NOT_AVAILABLE`. + +## `task_plan_complete_subtask` + +Mark your claimed sub-task complete. Releases the sub-task file claims and stamps a `plan-subtask-claim` observation with `status: 'completed'`. Downstream sub-tasks (those whose `depends_on` includes this one) become available automatically — `task_plan_list` will surface them in `next_available` on the next read. + +```json +{ + "name": "task_plan_complete_subtask", + "input": { + "plan_slug": "add-widget-page", + "subtask_index": 0, + "session_id": "sess_def", + "summary": "Widget API landed: GET /api/widgets serving rows." + } +} +``` + +Returns `{ status: 'completed' }`. Errors: `PLAN_SUBTASK_NOT_FOUND`, `PLAN_SUBTASK_NOT_CLAIMED`, `PLAN_SUBTASK_NOT_YOURS`. + +## Plan observation kinds + +The lane introduces two observation kinds on the sub-task threads. They are written through `MemoryStore.addObservation`, so content is compressed and `metadata` carries the structured payload. + +- `plan-subtask` — the initial advertisement, one per sub-task at publish time. `metadata` carries `parent_plan_slug`, `parent_plan_title`, `parent_spec_task_id`, `subtask_index`, `file_scope`, `depends_on`, `capability_hint`, and an initial `status: 'available'`. +- `plan-subtask-claim` — every lifecycle transition (claim, complete). `metadata.status` is the new state; `metadata.session_id` and `metadata.agent` identify the actor. The latest `plan-subtask-claim` observation by timestamp is authoritative. + ## Contract stability Fields may be added. Existing fields will not be removed or renamed within a minor version. diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index 13205ed..54aa906 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -70,3 +70,12 @@ export { type AgentProfile, type CandidateScore, } from './response-thresholds.js'; +export { + areDepsMet, + listPlans, + readSubtaskByBranch, + type ListPlansOptions, + type PlanInfo, + type SubtaskInfo, + type SubtaskStatus, +} from './plan.js'; diff --git a/packages/core/src/plan.ts b/packages/core/src/plan.ts new file mode 100644 index 0000000..4f62ebe --- /dev/null +++ b/packages/core/src/plan.ts @@ -0,0 +1,175 @@ +import type { MemoryStore } from './memory-store.js'; + +export type SubtaskStatus = 'available' | 'claimed' | 'completed' | 'blocked'; + +export interface SubtaskInfo { + task_id: number; + subtask_index: number; + title: string; + description: string; + status: SubtaskStatus; + file_scope: string[]; + depends_on: number[]; + capability_hint: string | null; + claimed_by_session_id: string | null; + claimed_by_agent: string | null; + parent_plan_slug: string; + parent_plan_title: string | null; + parent_spec_task_id: number | null; +} + +export interface PlanInfo { + plan_slug: string; + repo_root: string; + spec_task_id: number; + title: string; + created_at: number; + subtask_counts: Record; + subtasks: SubtaskInfo[]; + next_available: SubtaskInfo[]; +} + +export interface ListPlansOptions { + repo_root?: string; + only_with_available_subtasks?: boolean; + capability_match?: string; + limit?: number; +} + +const SUBTASK_BRANCH_RE = /^spec\/([a-z0-9-]+)\/sub-(\d+)$/; +const PLAN_ROOT_BRANCH_RE = /^spec\/([a-z0-9-]+)$/; + +function parseMeta(raw: string | null): Record { + if (!raw) return {}; + try { + const parsed = JSON.parse(raw) as unknown; + return typeof parsed === 'object' && parsed !== null ? (parsed as Record) : {}; + } catch { + return {}; + } +} + +function readSubtask(store: MemoryStore, task_id: number, plan_slug: string): SubtaskInfo | null { + const rows = store.storage.taskTimeline(task_id, 500); + const initial = rows.find((r) => r.kind === 'plan-subtask'); + if (!initial) return null; + const meta = parseMeta(initial.metadata); + // taskTimeline returns DESC by ts, so the first claim row is the latest. + const latestClaim = rows.find((r) => r.kind === 'plan-subtask-claim'); + const claimMeta = latestClaim ? parseMeta(latestClaim.metadata) : {}; + + const status = + (claimMeta.status as SubtaskStatus | undefined) ?? + (meta.status as SubtaskStatus | undefined) ?? + 'available'; + + const [titleLine, ...rest] = initial.content.split('\n\n'); + return { + task_id, + subtask_index: typeof meta.subtask_index === 'number' ? meta.subtask_index : -1, + title: titleLine ?? '(untitled)', + description: rest.join('\n\n').trim(), + status, + file_scope: Array.isArray(meta.file_scope) ? (meta.file_scope as string[]) : [], + depends_on: Array.isArray(meta.depends_on) ? (meta.depends_on as number[]) : [], + capability_hint: + typeof meta.capability_hint === 'string' ? (meta.capability_hint as string) : null, + claimed_by_session_id: + typeof claimMeta.session_id === 'string' ? (claimMeta.session_id as string) : null, + claimed_by_agent: typeof claimMeta.agent === 'string' ? (claimMeta.agent as string) : null, + parent_plan_slug: plan_slug, + parent_plan_title: + typeof meta.parent_plan_title === 'string' ? (meta.parent_plan_title as string) : null, + parent_spec_task_id: + typeof meta.parent_spec_task_id === 'number' ? (meta.parent_spec_task_id as number) : null, + }; +} + +export function readSubtaskByBranch( + store: MemoryStore, + branch: string, +): { task_id: number; info: SubtaskInfo } | null { + const m = branch.match(SUBTASK_BRANCH_RE); + if (!m) return null; + const slug = m[1]; + if (!slug) return null; + const tasks = store.storage.listTasks(2000); + const t = tasks.find((x) => x.branch === branch); + if (!t) return null; + const info = readSubtask(store, t.id, slug); + if (!info) return null; + return { task_id: t.id, info }; +} + +export function areDepsMet(subtask: SubtaskInfo, all: SubtaskInfo[]): boolean { + return subtask.depends_on.every((idx) => { + const dep = all.find((s) => s.subtask_index === idx); + return dep?.status === 'completed'; + }); +} + +export function listPlans(store: MemoryStore, opts: ListPlansOptions = {}): PlanInfo[] { + const limit = opts.limit ?? 50; + // listTasks default is 50; the plan registry may grow past that, so reach + // for a generous bound. A schema-level branch-prefix index is the proper + // long-term fix once the lane proves out. + const allTasks = store.storage.listTasks(2000); + const planRoots = allTasks + .filter((t) => PLAN_ROOT_BRANCH_RE.test(t.branch)) + .filter((t) => !opts.repo_root || t.repo_root === opts.repo_root); + + const plans = planRoots + .map((root): PlanInfo | null => { + const slugMatch = root.branch.match(PLAN_ROOT_BRANCH_RE); + if (!slugMatch) return null; + const slug = slugMatch[1]; + if (!slug) return null; + + const subtaskTasks = allTasks.filter((t) => { + const m = t.branch.match(SUBTASK_BRANCH_RE); + return Boolean(m && m[1] === slug); + }); + + const subtasks = subtaskTasks + .map((t) => readSubtask(store, t.id, slug)) + .filter((s): s is SubtaskInfo => s !== null) + .sort((a, b) => a.subtask_index - b.subtask_index); + + // No sub-tasks found means this is a plain spec change, not a published + // plan. Keep the two lanes separate. + if (subtasks.length === 0) return null; + + const counts: Record = { + available: 0, + claimed: 0, + completed: 0, + blocked: 0, + }; + for (const s of subtasks) counts[s.status]++; + + const nextAvailable = subtasks.filter( + (s) => s.status === 'available' && areDepsMet(s, subtasks), + ); + + return { + plan_slug: slug, + repo_root: root.repo_root, + spec_task_id: root.id, + title: subtasks[0]?.parent_plan_title ?? slug, + created_at: root.created_at, + subtask_counts: counts, + subtasks, + next_available: nextAvailable, + }; + }) + .filter((p): p is PlanInfo => p !== null); + + return plans + .filter((p) => !opts.only_with_available_subtasks || p.next_available.length > 0) + .filter( + (p) => + !opts.capability_match || + p.next_available.some((s) => s.capability_hint === opts.capability_match), + ) + .slice(0, limit); +}