From 4edf19df32128a65e67f42ba3d4ac419a2b70ce9 Mon Sep 17 00:00:00 2001 From: Harsh Batheja <40922251+harsh-batheja@users.noreply.github.com> Date: Tue, 10 Mar 2026 12:31:25 +0530 Subject: [PATCH] feat: lifecycle manager, backlog auto-claim, task decomposition, and verification gate (#365) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * feat: wire lifecycle manager, backlog auto-claim, and dashboard overhaul - Start LifecycleManager in dashboard server (30s polling) so reactions actually fire: CI failures, review comments, merge conflicts are now auto-forwarded to agents - Add backlog auto-claim poller (60s interval) that watches for issues labeled `agent:backlog` and auto-spawns agent sessions up to max concurrent limit (5) - Add tabbed dashboard UI: Board (kanban), Backlog (issue queue), PRs - Add issue creation form in dashboard — creates GitHub issues with `agent:backlog` label for immediate agent pickup - Add API routes: /api/backlog, /api/issues, /api/setup-labels - Pass notifier config through plugin registry (slack webhook fix) Co-Authored-By: Claude Opus 4.6 * feat: add task decomposition layer (classify → decompose → recurse) Adds LLM-driven recursive task decomposition upstream of session spawning. Complex issues are broken into atomic subtasks before agents start working. Each agent receives lineage context (where it fits in the hierarchy) and sibling awareness (what parallel agents are doing). Core changes: - New decomposer module (core/src/decomposer.ts) — classify, decompose, plan tree, lineage formatting, using Claude API - Extended SessionSpawnConfig with lineage/siblings fields - Prompt builder Layer 4: decomposition context (hierarchy + siblings) - ProjectConfig.decomposer config section with Zod validation - Tracker plugin: added removeLabels support for label management CLI: - `ao spawn --decompose` flag - `--max-depth ` option for decomposition depth - Spawns multiple sessions with lineage context for composite tasks Backlog poller: - Respects project.decomposer.enabled for auto-decomposition - Posts plan as issue comment when requireApproval=true - Auto-spawns subtasks with lineage when requireApproval=false Config example: projects: my-app: decomposer: enabled: true maxDepth: 3 requireApproval: true Co-Authored-By: Claude Opus 4.6 * feat: add verification gate — issues stay open until human confirms fix PR merge no longer auto-closes GitHub issues. Instead: 1. On PR merge: issue labeled `merged-unverified`, stays open 2. Human checks staging, then runs `ao verify ` to close 3. Or `ao verify --fail` to flag verification failure Changes: - services.ts: labelIssuesForVerification() replaces closeIssuesForMergedSessions() - New CLI command: `ao verify` (verify/fail/list modes) - New API route: GET/POST /api/verify - Dashboard: new Verify tab with one-click verify/fail buttons - ao status: shows count of issues awaiting verification - Idle session detection + auto-nudge reaction - Use TERMINAL_STATUSES in batch-spawn dedup check Co-Authored-By: Claude Opus 4.6 * fix: rename decomposerConfig to avoid variable shadowing Addresses Bugbot medium severity issue where inner variable shadowed outer from getServices(). * fix: update pnpm-lock.yaml for new @anthropic-ai/sdk dependency * fix: resolve remaining merge conflicts and syntax errors - Remove leftover conflict markers in types.ts - Remove orphaned code in services.ts - Fix semicolon to comma in config.ts - Remove unused import in verify.ts * fix: address final Bugbot issues - requireApproval path now exits early with continue to prevent fall-through to in-progress label and session spawned comment - remove packages/core/package-lock.json (pnpm workspace should only use root pnpm-lock.yaml) * fix: idle sessions now transition back to working When agent resumes activity after being idle, the status correctly transitions to 'working' instead of remaining stuck in 'idle' state. * fix(backlog): remove agent:backlog label when claiming issues When claiming issues from the backlog, the poller now removes the agent:backlog label in addition to adding agent:in-progress. This prevents duplicate work if all spawned sessions reach terminal status and the poller rediscovers the issue. * fix(test): use Set for TERMINAL_STATUSES mock The mock for TERMINAL_STATUSES was an array, but the real export is a ReadonlySet. Changed to use a Set so tests with non-empty sessions won't crash when calling .has(). * fix(web): resolve backlog/dashboard regressions after branch sync * fix(web): align dashboard events hook and SSE test mocks * fix(notifier-openclaw): apply exponential delay from retry index * fix(integration-tests): align openclaw retry delay expectation * fix(web): keep dashboard header stats in sync * fix(openclaw): keep first retry at base delay --------- Co-authored-by: Agent Co-authored-by: Claude Opus 4.6 Co-authored-by: Harsh Co-authored-by: Harsh --- packages/cli/src/commands/spawn.ts | 87 +- packages/cli/src/commands/status.ts | 37 + packages/cli/src/commands/verify.ts | 178 +++++ packages/cli/src/index.ts | 2 + packages/core/package.json | 1 + packages/core/src/config.ts | 23 + packages/core/src/decomposer.ts | 276 +++++++ packages/core/src/index.ts | 19 + packages/core/src/lifecycle-manager.ts | 8 +- packages/core/src/prompt-builder.ts | 25 + packages/core/src/session-manager.ts | 2 + packages/core/src/types.ts | 20 + .../src/notifier-openclaw.integration.test.ts | 2 +- .../plugins/notifier-openclaw/src/index.ts | 9 +- packages/plugins/tracker-github/src/index.ts | 13 + packages/web/src/__tests__/api-routes.test.ts | 6 + packages/web/src/__tests__/services.test.ts | 104 +++ packages/web/src/app/api/backlog/route.ts | 23 + packages/web/src/app/api/events/route.ts | 16 +- packages/web/src/app/api/issues/route.ts | 104 +++ .../web/src/app/api/setup-labels/route.ts | 52 ++ packages/web/src/app/api/verify/route.ts | 91 +++ packages/web/src/app/page.tsx | 3 + packages/web/src/components/Dashboard.tsx | 747 +++++++++++++++--- packages/web/src/lib/services.ts | 317 +++++++- pnpm-lock.yaml | 9 + 26 files changed, 2058 insertions(+), 116 deletions(-) create mode 100644 packages/cli/src/commands/verify.ts create mode 100644 packages/core/src/decomposer.ts create mode 100644 packages/web/src/app/api/backlog/route.ts create mode 100644 packages/web/src/app/api/issues/route.ts create mode 100644 packages/web/src/app/api/setup-labels/route.ts create mode 100644 packages/web/src/app/api/verify/route.ts diff --git a/packages/cli/src/commands/spawn.ts b/packages/cli/src/commands/spawn.ts index 2829598ef..a02d290eb 100644 --- a/packages/cli/src/commands/spawn.ts +++ b/packages/cli/src/commands/spawn.ts @@ -1,7 +1,17 @@ import chalk from "chalk"; import ora from "ora"; import type { Command } from "commander"; -import { loadConfig, type OrchestratorConfig } from "@composio/ao-core"; +import { + loadConfig, + decompose, + getLeaves, + getSiblings, + formatPlanTree, + TERMINAL_STATUSES, + type OrchestratorConfig, + type DecomposerConfig, + DEFAULT_DECOMPOSER_CONFIG, +} from "@composio/ao-core"; import { exec } from "../lib/shell.js"; import { banner } from "../lib/format.js"; import { getSessionManager } from "../lib/create-session-manager.js"; @@ -118,6 +128,8 @@ export function registerSpawn(program: Command): void { .option("--agent ", "Override the agent plugin (e.g. codex, claude-code)") .option("--claim-pr ", "Immediately claim an existing PR for the spawned session") .option("--assign-on-github", "Assign the claimed PR to the authenticated GitHub user") + .option("--decompose", "Decompose issue into subtasks before spawning") + .option("--max-depth ", "Max decomposition depth (default: 3)") .action( async ( projectId: string, @@ -127,6 +139,8 @@ export function registerSpawn(program: Command): void { agent?: string; claimPr?: string; assignOnGithub?: boolean; + decompose?: boolean; + maxDepth?: string; }, ) => { const config = loadConfig(); @@ -144,13 +158,68 @@ export function registerSpawn(program: Command): void { process.exit(1); } + const claimOptions: SpawnClaimOptions = { + claimPr: opts.claimPr, + assignOnGithub: opts.assignOnGithub, + }; + try { - await runSpawnPreflight(config, projectId, { claimPr: opts.claimPr }); + await runSpawnPreflight(config, projectId, claimOptions); await ensureLifecycleWorker(config, projectId); - await spawnSession(config, projectId, issueId, opts.open, opts.agent, { - claimPr: opts.claimPr, - assignOnGithub: opts.assignOnGithub, - }); + + if (opts.decompose && issueId) { + // Decompose the issue before spawning + const project = config.projects[projectId]; + const decompConfig: DecomposerConfig = { + ...DEFAULT_DECOMPOSER_CONFIG, + ...(project.decomposer ?? {}), + maxDepth: opts.maxDepth + ? parseInt(opts.maxDepth, 10) + : (project.decomposer?.maxDepth ?? 3), + }; + + const spinner = ora("Decomposing task...").start(); + const issueTitle = issueId; + + const plan = await decompose(issueTitle, decompConfig); + const leaves = getLeaves(plan.tree); + spinner.succeed(`Decomposed into ${chalk.bold(String(leaves.length))} subtasks`); + + console.log(); + console.log(chalk.dim(formatPlanTree(plan.tree))); + console.log(); + + if (leaves.length <= 1) { + console.log(chalk.yellow("Task is atomic — spawning directly.")); + await spawnSession(config, projectId, issueId, opts.open, opts.agent, claimOptions); + } else { + // Create child issues and spawn sessions with lineage context + const sm = await getSessionManager(config); + console.log(chalk.bold(`Spawning ${leaves.length} sessions with lineage context...`)); + console.log(); + + for (const leaf of leaves) { + const siblings = getSiblings(plan.tree, leaf.id); + try { + const session = await sm.spawn({ + projectId, + issueId, // All work on the same parent issue for now + lineage: leaf.lineage, + siblings, + agent: opts.agent, + }); + console.log(` ${chalk.green("✓")} ${session.id} — ${leaf.description}`); + } catch (err) { + console.error( + ` ${chalk.red("✗")} ${leaf.description} — ${err instanceof Error ? err.message : err}`, + ); + } + await new Promise((r) => setTimeout(r, 500)); + } + } + } else { + await spawnSession(config, projectId, issueId, opts.open, opts.agent, claimOptions); + } } catch (err) { console.error(chalk.red(`✗ ${err instanceof Error ? err.message : String(err)}`)); process.exit(1); @@ -199,12 +268,12 @@ export function registerBatchSpawn(program: Command): void { const spawnedIssues = new Set(); // Load existing sessions once before the loop to avoid repeated reads + enrichment. - // Exclude dead/killed sessions so crashed sessions don't block respawning. - const deadStatuses = new Set(["killed", "done", "exited"]); + // Exclude terminal sessions so completed/merged sessions don't block respawning + // (e.g. when an issue is reopened after its PR was merged). const existingSessions = await sm.list(projectId); const existingIssueMap = new Map( existingSessions - .filter((s) => s.issueId && !deadStatuses.has(s.status)) + .filter((s) => s.issueId && !TERMINAL_STATUSES.has(s.status)) .map((s) => [s.issueId!.toLowerCase(), s.id]), ); diff --git a/packages/cli/src/commands/status.ts b/packages/cli/src/commands/status.ts index 8a213b714..d03b9c22c 100644 --- a/packages/cli/src/commands/status.ts +++ b/packages/cli/src/commands/status.ts @@ -8,6 +8,8 @@ import { type CIStatus, type ReviewDecision, type ActivityState, + type Tracker, + type ProjectConfig, loadConfig, } from "@composio/ao-core"; import { git, getTmuxSessions, getTmuxActivity } from "../lib/shell.js"; @@ -286,6 +288,41 @@ export function registerStatus(program: Command): void { ` ${totalSessions} active session${totalSessions !== 1 ? "s" : ""} across ${projectIds.length} project${projectIds.length !== 1 ? "s" : ""}`, ), ); + + // Check for issues awaiting verification across all projects + try { + const { createPluginRegistry } = await import("@composio/ao-core"); + const registry = createPluginRegistry(); + await registry.loadFromConfig(config, (pkg: string) => import(pkg)); + + let unverifiedTotal = 0; + for (const projectId of projectIds) { + const project: ProjectConfig | undefined = config.projects[projectId]; + if (!project?.tracker) continue; + const tracker = registry.get("tracker", project.tracker.plugin); + if (!tracker?.listIssues) continue; + try { + const issues = await tracker.listIssues( + { state: "open", labels: ["merged-unverified"], limit: 20 }, + project, + ); + unverifiedTotal += issues.length; + } catch { + // Tracker query failed — not critical + } + } + + if (unverifiedTotal > 0) { + console.log( + chalk.yellow( + ` ⚠ ${unverifiedTotal} issue${unverifiedTotal !== 1 ? "s" : ""} awaiting verification (use \`ao verify --list\` to see them)`, + ), + ); + } + } catch { + // Plugin registry or tracker unavailable — skip silently + } + console.log(); } }); diff --git a/packages/cli/src/commands/verify.ts b/packages/cli/src/commands/verify.ts new file mode 100644 index 000000000..4b639ed63 --- /dev/null +++ b/packages/cli/src/commands/verify.ts @@ -0,0 +1,178 @@ +import chalk from "chalk"; +import type { Command } from "commander"; +import { + type Tracker, + type ProjectConfig, + type OrchestratorConfig, + type PluginRegistry, + loadConfig, +} from "@composio/ao-core"; + +/** + * Resolve the target project from config. + * If --project is given, use that. If there is exactly one project, use it. + * Otherwise, error out asking the user to specify --project. + */ +function resolveProject( + config: OrchestratorConfig, + projectOpt?: string, +): { projectId: string; project: ProjectConfig } { + if (projectOpt) { + const project = config.projects[projectOpt]; + if (!project) { + console.error(chalk.red(`Unknown project: ${projectOpt}`)); + process.exit(1); + } + return { projectId: projectOpt, project }; + } + + const ids = Object.keys(config.projects); + if (ids.length === 0) { + console.error(chalk.red("No projects configured. Run `ao init` first.")); + process.exit(1); + } + if (ids.length > 1) { + console.error( + chalk.red(`Multiple projects found. Specify one with --project: ${ids.join(", ")}`), + ); + process.exit(1); + } + + return { projectId: ids[0], project: config.projects[ids[0]] }; +} + +/** + * Get a Tracker instance from the plugin registry for the given project. + */ +async function getTracker( + config: OrchestratorConfig, + project: ProjectConfig, +): Promise<{ tracker: Tracker; registry: PluginRegistry }> { + if (!project.tracker) { + console.error(chalk.red("No tracker configured for this project.")); + process.exit(1); + } + + // getSessionManager internally creates the registry; we need the registry + // directly, so we replicate the same pattern from create-session-manager. + const { createPluginRegistry } = await import("@composio/ao-core"); + const registry = createPluginRegistry(); + await registry.loadFromConfig(config, (pkg: string) => import(pkg)); + + const tracker = registry.get("tracker", project.tracker.plugin); + if (!tracker) { + console.error(chalk.red(`Tracker plugin "${project.tracker.plugin}" not found.`)); + process.exit(1); + } + + return { tracker, registry }; +} + +export function registerVerify(program: Command): void { + program + .command("verify") + .description("Mark an issue as verified (or failed) after checking the fix on staging") + .argument("[issue]", "Issue number or identifier to verify") + .option("-p, --project ", "Project ID (required if multiple projects)") + .option("--fail", "Mark verification as failed instead of passing") + .option("-c, --comment ", "Custom comment to add") + .option("-l, --list", "List all issues with merged-unverified label") + .action( + async ( + issue: string | undefined, + opts: { project?: string; fail?: boolean; comment?: string; list?: boolean }, + ) => { + let config: OrchestratorConfig; + try { + config = loadConfig(); + } catch { + console.error(chalk.red("No config found. Run `ao init` first.")); + process.exit(1); + return; + } + + const { projectId, project } = resolveProject(config, opts.project); + const { tracker } = await getTracker(config, project); + + // --list mode: show all merged-unverified issues + if (opts.list) { + if (!tracker.listIssues) { + console.error(chalk.red("Tracker does not support listing issues.")); + process.exit(1); + } + + const issues = await tracker.listIssues( + { state: "open", labels: ["merged-unverified"] }, + project, + ); + + if (issues.length === 0) { + console.log(chalk.dim("No merged-unverified issues found.")); + return; + } + + console.log(chalk.bold(`Merged-unverified issues in ${project.name || projectId}:\n`)); + for (const i of issues) { + const labels = i.labels.length > 0 ? chalk.dim(` [${i.labels.join(", ")}]`) : ""; + console.log(` ${chalk.cyan(i.id)} ${i.title}${labels}`); + if (i.url) { + console.log(` ${chalk.dim(i.url)}`); + } + } + console.log( + chalk.dim( + `\n ${issues.length} issue${issues.length !== 1 ? "s" : ""} awaiting verification`, + ), + ); + return; + } + + // Verify/fail mode: requires an issue argument + if (!issue) { + console.error(chalk.red("Issue number is required. Usage: ao verify ")); + process.exit(1); + } + + if (!tracker.updateIssue) { + console.error(chalk.red("Tracker does not support updating issues.")); + process.exit(1); + } + + if (opts.fail) { + // Verification failed + const comment = opts.comment ?? "Verification failed — problem persists on staging."; + + await tracker.updateIssue( + issue, + { + state: "open", + labels: ["verification-failed"], + removeLabels: ["merged-unverified"], + comment, + }, + project, + ); + + console.log(chalk.red(`Issue ${issue} marked as verification-failed (kept open).`)); + console.log(chalk.dim(` Comment: ${comment}`)); + } else { + // Verification passed + const comment = opts.comment ?? "Verified — fix confirmed on staging."; + + await tracker.updateIssue( + issue, + { + state: "closed", + labels: ["verified"], + removeLabels: ["merged-unverified"], + comment, + }, + project, + ); + + console.log(chalk.green(`Issue ${issue} verified and closed.`)); + console.log(chalk.dim(` Comment: ${comment}`)); + } + }, + ); +} diff --git a/packages/cli/src/index.ts b/packages/cli/src/index.ts index f7cf1bbb6..6ee253ccd 100644 --- a/packages/cli/src/index.ts +++ b/packages/cli/src/index.ts @@ -11,6 +11,7 @@ import { registerDashboard } from "./commands/dashboard.js"; import { registerOpen } from "./commands/open.js"; import { registerStart, registerStop } from "./commands/start.js"; import { registerLifecycleWorker } from "./commands/lifecycle-worker.js"; +import { registerVerify } from "./commands/verify.js"; const program = new Command(); @@ -31,5 +32,6 @@ registerReviewCheck(program); registerDashboard(program); registerOpen(program); registerLifecycleWorker(program); +registerVerify(program); program.parse(); diff --git a/packages/core/package.json b/packages/core/package.json index cb4c47974..3c6bc66b9 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -43,6 +43,7 @@ "clean": "rm -rf dist" }, "dependencies": { + "@anthropic-ai/sdk": "^0.52.0", "yaml": "^2.7.0", "zod": "^3.24.0" }, diff --git a/packages/core/src/config.ts b/packages/core/src/config.ts index 1607615ca..11ca6fb23 100644 --- a/packages/core/src/config.ts +++ b/packages/core/src/config.ts @@ -65,6 +65,20 @@ const AgentSpecificConfigSchema = z }) .passthrough(); +const DecomposerConfigSchema = z + .object({ + enabled: z.boolean().default(false), + maxDepth: z.number().min(1).max(5).default(3), + model: z.string().default("claude-sonnet-4-20250514"), + requireApproval: z.boolean().default(true), + }) + .default({ + enabled: false, + maxDepth: 3, + model: "claude-sonnet-4-20250514", + requireApproval: true, + }); + const ProjectConfigSchema = z.object({ name: z.string().optional(), repo: z.string(), @@ -90,6 +104,7 @@ const ProjectConfigSchema = z.object({ .enum(["reuse", "delete", "ignore", "delete-new", "ignore-new", "kill-previous"]) .optional(), opencodeIssueSessionStrategy: z.enum(["reuse", "delete", "ignore"]).optional(), + decomposer: DecomposerConfigSchema.optional(), }); const DefaultPluginsSchema = z.object({ @@ -258,6 +273,14 @@ function applyDefaultReactions(config: OrchestratorConfig): OrchestratorConfig { priority: "action", message: "PR is ready to merge", }, + "agent-idle": { + auto: true, + action: "send-to-agent", + message: + "You appear to be idle. If your task is not complete, continue working — write the code, commit, push, and create a PR. If you are blocked, explain what is blocking you.", + retries: 2, + escalateAfter: "15m", + }, "agent-stuck": { auto: true, action: "notify", diff --git a/packages/core/src/decomposer.ts b/packages/core/src/decomposer.ts new file mode 100644 index 000000000..4868238a8 --- /dev/null +++ b/packages/core/src/decomposer.ts @@ -0,0 +1,276 @@ +/** + * Task Decomposer — LLM-driven recursive task decomposition. + * + * Classifies issues as atomic (one agent can handle it) or composite + * (needs to be broken into subtasks). Composite tasks are recursively + * decomposed until all leaves are atomic. + * + * Integration: sits upstream of SessionManager.spawn(). When enabled, + * complex issues are decomposed into child issues before agents are spawned. + */ + +import Anthropic from "@anthropic-ai/sdk"; + +// ============================================================================= +// TYPES +// ============================================================================= + +export type TaskKind = "atomic" | "composite"; +export type TaskStatus = "pending" | "decomposing" | "ready" | "running" | "done" | "failed"; + +export interface TaskNode { + id: string; // hierarchical: "1", "1.2", "1.2.3" + depth: number; + description: string; + kind?: TaskKind; + status: TaskStatus; + lineage: string[]; // ancestor descriptions root→parent + children: TaskNode[]; + result?: string; + issueId?: string; // tracker issue created for this subtask + sessionId?: string; // AO session working on this task +} + +export interface DecompositionPlan { + id: string; + rootTask: string; + tree: TaskNode; + maxDepth: number; + phase: "decomposing" | "review" | "approved" | "executing" | "done" | "failed"; + createdAt: string; + approvedAt?: string; + parentIssueId?: string; +} + +export interface DecomposerConfig { + /** Enable auto-decomposition for backlog issues (default: false) */ + enabled: boolean; + /** Max recursion depth (default: 3) */ + maxDepth: number; + /** Model to use for decomposition (default: claude-sonnet-4-20250514) */ + model: string; + /** Require human approval before executing decomposed plans (default: true) */ + requireApproval: boolean; +} + +export const DEFAULT_DECOMPOSER_CONFIG: DecomposerConfig = { + enabled: false, + maxDepth: 3, + model: "claude-sonnet-4-20250514", + requireApproval: true, +}; + +// ============================================================================= +// LINEAGE CONTEXT +// ============================================================================= + +/** Format the task lineage as an indented hierarchy for LLM context. */ +export function formatLineage(lineage: string[], current: string): string { + const parts = lineage.map((desc, i) => `${" ".repeat(i)}${i}. ${desc}`); + parts.push(`${" ".repeat(lineage.length)}${lineage.length}. ${current} <-- (this task)`); + return parts.join("\n"); +} + +/** Format sibling tasks for awareness context. */ +export function formatSiblings(siblings: string[], current: string): string { + if (siblings.length === 0) return ""; + const lines = siblings.map((s) => (s === current ? ` - ${s} <-- (you)` : ` - ${s}`)); + return `Sibling tasks being worked on in parallel:\n${lines.join("\n")}`; +} + +// ============================================================================= +// LLM CALLS +// ============================================================================= + +const CLASSIFY_SYSTEM = `You decide whether a software task is "atomic" or "composite". + +- "atomic" = a developer can implement this directly without needing to plan further. It may involve multiple steps, but they're all part of one coherent unit of work. +- "composite" = this clearly contains 2+ independent concerns that should be worked on separately (e.g., backend + frontend, or auth + database + UI). + +Decision heuristics: +- If the task names a single feature, endpoint, component, or module: atomic. +- If the task bundles unrelated concerns (e.g., "build auth and set up CI"): composite. +- If you're at depth 2 or deeper in the hierarchy, it is almost certainly atomic — only mark composite if you can name 2+ truly independent deliverables. +- When in doubt, choose atomic. Over-decomposition creates more overhead than under-decomposition. + +Respond with ONLY the word "atomic" or "composite". Nothing else.`; + +const DECOMPOSE_SYSTEM = `You are a pragmatic task decomposition engine for software projects. + +Given a composite task, break it into the MINIMUM number of subtasks needed: +- A simple task might only need 2 subtasks. +- A complex task might need up to 7, but only if each is truly distinct. +- Do NOT pad with extra subtasks. Do NOT create "test and polish" or "define requirements" subtasks. +- Do NOT create subtasks that overlap or restate each other. +- Each subtask should represent real, distinct work. + +Think about how an experienced developer would actually split this work. + +Respond with a JSON array of strings, each being a subtask description. Example: +["Implement Stripe webhook handler", "Build subscription management UI"] + +Nothing else — just the JSON array.`; + +async function classifyTask( + client: Anthropic, + model: string, + task: string, + lineage: string[], +): Promise { + const context = formatLineage(lineage, task); + const res = await client.messages.create({ + model, + max_tokens: 10, + system: CLASSIFY_SYSTEM, + messages: [{ role: "user", content: `Task hierarchy:\n${context}` }], + }); + + const text = res.content[0].type === "text" ? res.content[0].text.trim().toLowerCase() : ""; + return text === "composite" ? "composite" : "atomic"; +} + +async function decomposeTask( + client: Anthropic, + model: string, + task: string, + lineage: string[], +): Promise { + const context = formatLineage(lineage, task); + const res = await client.messages.create({ + model, + max_tokens: 1024, + system: DECOMPOSE_SYSTEM, + messages: [{ role: "user", content: `Task hierarchy:\n${context}` }], + }); + + const text = res.content[0].type === "text" ? res.content[0].text.trim() : "[]"; + const jsonMatch = text.match(/\[[\s\S]*\]/); + if (!jsonMatch) { + throw new Error(`Decomposition failed — no JSON array in response: ${text}`); + } + + const subtasks = JSON.parse(jsonMatch[0]) as string[]; + if (!Array.isArray(subtasks) || subtasks.length < 2) { + throw new Error(`Decomposition produced ${subtasks.length} subtasks — need at least 2`); + } + + return subtasks; +} + +// ============================================================================= +// TREE OPERATIONS +// ============================================================================= + +function createTaskNode( + id: string, + description: string, + depth: number, + lineage: string[], +): TaskNode { + return { id, depth, description, status: "pending", lineage, children: [] }; +} + +/** Recursively decompose a task tree (planning phase — no execution). */ +async function planTree( + client: Anthropic, + model: string, + task: TaskNode, + maxDepth: number, +): Promise { + const kind = task.depth >= maxDepth ? "atomic" : await classifyTask(client, model, task.description, task.lineage); + + task.kind = kind; + + if (kind === "atomic") { + task.status = "ready"; + return task; + } + + task.status = "decomposing"; + const subtaskDescriptions = await decomposeTask(client, model, task.description, task.lineage); + + const childLineage = [...task.lineage, task.description]; + task.children = subtaskDescriptions.map((desc, i) => + createTaskNode(`${task.id}.${i + 1}`, desc, task.depth + 1, childLineage), + ); + + // Recurse on children concurrently + await Promise.all(task.children.map((child) => planTree(client, model, child, maxDepth))); + + task.status = "ready"; + return task; +} + +// ============================================================================= +// PUBLIC API +// ============================================================================= + +/** Create a decomposition plan for a task. */ +export async function decompose( + taskDescription: string, + config: DecomposerConfig = DEFAULT_DECOMPOSER_CONFIG, +): Promise { + const client = new Anthropic(); + const tree = createTaskNode("1", taskDescription, 0, []); + + await planTree(client, config.model, tree, config.maxDepth); + + return { + id: `plan-${Date.now()}`, + rootTask: taskDescription, + tree, + maxDepth: config.maxDepth, + phase: config.requireApproval ? "review" : "approved", + createdAt: new Date().toISOString(), + }; +} + +/** Collect all leaf (atomic) tasks from a tree. */ +export function getLeaves(task: TaskNode): TaskNode[] { + if (task.children.length === 0) return [task]; + return task.children.flatMap(getLeaves); +} + +/** Get sibling task descriptions for a given task. */ +export function getSiblings(root: TaskNode, taskId: string): string[] { + function findParent(node: TaskNode): TaskNode | null { + for (const child of node.children) { + if (child.id === taskId) return node; + const found = findParent(child); + if (found) return found; + } + return null; + } + + const parent = findParent(root); + if (!parent) return []; + return parent.children.filter((c) => c.id !== taskId).map((c) => c.description); +} + +/** Format the plan tree as a human-readable string. */ +export function formatPlanTree(task: TaskNode, indent = 0): string { + const prefix = " ".repeat(indent); + const kindTag = task.kind === "atomic" ? "[ATOMIC]" : task.kind === "composite" ? "[COMPOSITE]" : ""; + const statusTag = task.status !== "ready" ? ` (${task.status})` : ""; + let line = `${prefix}${task.id}. ${kindTag} ${task.description}${statusTag}`; + + if (task.children.length > 0) { + const childLines = task.children.map((c) => formatPlanTree(c, indent + 1)).join("\n"); + line += "\n" + childLines; + } + + return line; +} + +/** Propagate done/failed status up the tree. */ +export function propagateStatus(task: TaskNode): void { + if (task.children.length === 0) return; + task.children.forEach(propagateStatus); + if (task.children.every((c) => c.status === "done")) { + task.status = "done"; + } else if (task.children.some((c) => c.status === "failed")) { + task.status = "failed"; + } else if (task.children.some((c) => c.status === "running" || c.status === "done")) { + task.status = "running"; + } +} diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index e973aa2dd..9526c66c2 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -55,6 +55,25 @@ export type { LifecycleManagerDeps } from "./lifecycle-manager.js"; export { buildPrompt, BASE_AGENT_PROMPT } from "./prompt-builder.js"; export type { PromptBuildConfig } from "./prompt-builder.js"; +// Decomposer — LLM-driven task decomposition +export { + decompose, + getLeaves, + getSiblings, + formatPlanTree, + formatLineage, + formatSiblings, + propagateStatus, + DEFAULT_DECOMPOSER_CONFIG, +} from "./decomposer.js"; +export type { + TaskNode, + TaskKind, + TaskStatus, + DecompositionPlan, + DecomposerConfig, +} from "./decomposer.js"; + // Orchestrator prompt — generates orchestrator context for `ao start` export { generateOrchestratorPrompt } from "./orchestrator-prompt.js"; export type { OrchestratorPromptConfig } from "./orchestrator-prompt.js"; diff --git a/packages/core/src/lifecycle-manager.ts b/packages/core/src/lifecycle-manager.ts index b6e75bb78..f05547f14 100644 --- a/packages/core/src/lifecycle-manager.ts +++ b/packages/core/src/lifecycle-manager.ts @@ -160,6 +160,8 @@ function statusToEventType(_from: SessionStatus | undefined, to: SessionStatus): return "merge.completed"; case "needs_input": return "session.needs_input"; + case "idle": + return "session.idle"; case "stuck": return "session.stuck"; case "errored": @@ -184,6 +186,8 @@ function eventToReactionKey(eventType: EventType): string | null { return "merge-conflicts"; case "merge.ready": return "approved-and-green"; + case "session.idle": + return "agent-idle"; case "session.stuck": return "agent-stuck"; case "session.needs_input": @@ -322,7 +326,8 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan if (activityState) { if (activityState.state === "waiting_input") return "needs_input"; if (activityState.state === "exited") return "killed"; - // active/ready/idle/blocked — proceed to PR checks below + if (activityState.state === "idle") return "idle"; + // active/ready/blocked — proceed to PR checks below } else { // getActivityState returned null — fall back to terminal output parsing const runtime = registry.get( @@ -400,6 +405,7 @@ export function createLifecycleManager(deps: LifecycleManagerDeps): LifecycleMan // 5. Default: if agent is active, it's working if ( session.status === "spawning" || + session.status === "idle" || session.status === SESSION_STATUS.STUCK || session.status === SESSION_STATUS.NEEDS_INPUT ) { diff --git a/packages/core/src/prompt-builder.ts b/packages/core/src/prompt-builder.ts index eb0a972fe..2cbcdccbb 100644 --- a/packages/core/src/prompt-builder.ts +++ b/packages/core/src/prompt-builder.ts @@ -58,6 +58,12 @@ export interface PromptBuildConfig { /** Explicit user prompt (appended last) */ userPrompt?: string; + + /** Decomposition context — ancestor task chain (from decomposer) */ + lineage?: string[]; + + /** Decomposition context — sibling task descriptions (from decomposer) */ + siblings?: string[]; } // ============================================================================= @@ -159,6 +165,25 @@ export function buildPrompt(config: PromptBuildConfig): string { sections.push(`## Project Rules\n${userRules}`); } + // Layer 4: Decomposition context (lineage + siblings) + if (config.lineage && config.lineage.length > 0) { + const hierarchy = config.lineage.map((desc, i) => `${" ".repeat(i)}${i}. ${desc}`); + // Add current task marker using issueId or last lineage entry + const currentLabel = config.issueId ?? "this task"; + hierarchy.push(`${" ".repeat(config.lineage.length)}${config.lineage.length}. ${currentLabel} <-- (this task)`); + + sections.push( + `## Task Hierarchy\nThis task is part of a larger decomposed plan. Your place in the hierarchy:\n\n\`\`\`\n${hierarchy.join("\n")}\n\`\`\`\n\nStay focused on YOUR specific task. Do not implement functionality that belongs to other tasks in the hierarchy.`, + ); + } + + if (config.siblings && config.siblings.length > 0) { + const siblingLines = config.siblings.map((s) => ` - ${s}`); + sections.push( + `## Parallel Work\nSibling tasks being worked on in parallel:\n${siblingLines.join("\n")}\n\nDo not duplicate work that sibling tasks handle. If you need interfaces/types from siblings, define reasonable stubs.`, + ); + } + // Explicit user prompt (appended last, highest priority) if (config.userPrompt) { sections.push(`## Additional Instructions\n${config.userPrompt}`); diff --git a/packages/core/src/session-manager.ts b/packages/core/src/session-manager.ts index d2fa9387b..3a072f9cb 100644 --- a/packages/core/src/session-manager.ts +++ b/packages/core/src/session-manager.ts @@ -825,6 +825,8 @@ export function createSessionManager(deps: SessionManagerDeps): OpenCodeSessionM issueId: spawnConfig.issueId, issueContext, userPrompt: spawnConfig.prompt, + lineage: spawnConfig.lineage, + siblings: spawnConfig.siblings, }); // Get agent launch config and create runtime — clean up workspace on failure diff --git a/packages/core/src/types.ts b/packages/core/src/types.ts index ccbdf0104..d083f2c88 100644 --- a/packages/core/src/types.ts +++ b/packages/core/src/types.ts @@ -38,6 +38,7 @@ export type SessionStatus = | "stuck" | "errored" | "killed" + | "idle" | "done" | "terminated"; @@ -85,6 +86,7 @@ export const SESSION_STATUS = { NEEDS_INPUT: "needs_input" as const, STUCK: "stuck" as const, ERRORED: "errored" as const, + IDLE: "idle" as const, KILLED: "killed" as const, DONE: "done" as const, TERMINATED: "terminated" as const, @@ -180,6 +182,10 @@ export interface SessionSpawnConfig { agent?: string; /** Override the OpenCode subagent for this session (e.g. "sisyphus", "oracle") */ subagent?: string; + /** Decomposition context — ancestor task chain (passed to prompt builder) */ + lineage?: string[]; + /** Decomposition context — sibling task descriptions (passed to prompt builder) */ + siblings?: string[]; } /** Config for creating an orchestrator session */ @@ -488,6 +494,7 @@ export interface IssueFilters { export interface IssueUpdate { state?: "open" | "in_progress" | "closed"; labels?: string[]; + removeLabels?: string[]; assignee?: string; comment?: string; } @@ -729,6 +736,7 @@ export type EventType = | "session.working" | "session.exited" | "session.killed" + | "session.idle" | "session.stuck" | "session.needs_input" | "session.errored" @@ -921,6 +929,18 @@ export interface ProjectConfig { | "kill-previous"; opencodeIssueSessionStrategy?: "reuse" | "delete" | "ignore"; + + /** Task decomposition configuration */ + decomposer?: { + /** Enable auto-decomposition for backlog issues (default: false) */ + enabled: boolean; + /** Max recursion depth (default: 3) */ + maxDepth: number; + /** Model to use for decomposition (default: claude-sonnet-4-20250514) */ + model: string; + /** Require human approval before executing decomposed plans (default: true) */ + requireApproval: boolean; + }; } export interface TrackerConfig { diff --git a/packages/integration-tests/src/notifier-openclaw.integration.test.ts b/packages/integration-tests/src/notifier-openclaw.integration.test.ts index 2d29a527d..6fb85ea60 100644 --- a/packages/integration-tests/src/notifier-openclaw.integration.test.ts +++ b/packages/integration-tests/src/notifier-openclaw.integration.test.ts @@ -108,7 +108,7 @@ describe("notifier-openclaw integration", () => { await vi.advanceTimersByTimeAsync(0); expect(fetchMock).toHaveBeenCalledTimes(1); - await vi.advanceTimersByTimeAsync(100); + await vi.advanceTimersByTimeAsync(200); expect(fetchMock).toHaveBeenCalledTimes(2); await promise; diff --git a/packages/plugins/notifier-openclaw/src/index.ts b/packages/plugins/notifier-openclaw/src/index.ts index a5b10166b..f016b2420 100644 --- a/packages/plugins/notifier-openclaw/src/index.ts +++ b/packages/plugins/notifier-openclaw/src/index.ts @@ -6,11 +6,7 @@ import { type OrchestratorEvent, type PluginModule, } from "@composio/ao-core"; -import { - isRetryableHttpStatus, - normalizeRetryConfig, - validateUrl, -} from "@composio/ao-core/utils"; +import { isRetryableHttpStatus, normalizeRetryConfig, validateUrl } from "@composio/ao-core/utils"; export const manifest = { name: "openclaw", @@ -113,7 +109,8 @@ function formatActionsLine(actions: NotifyAction[]): string { } export function create(config?: Record): Notifier { - const url = (typeof config?.url === "string" ? config.url : undefined) ?? + const url = + (typeof config?.url === "string" ? config.url : undefined) ?? "http://127.0.0.1:18789/hooks/agent"; const token = (typeof config?.token === "string" ? config.token : undefined) ?? diff --git a/packages/plugins/tracker-github/src/index.ts b/packages/plugins/tracker-github/src/index.ts index 7f3afef26..e86d638c7 100644 --- a/packages/plugins/tracker-github/src/index.ts +++ b/packages/plugins/tracker-github/src/index.ts @@ -265,6 +265,19 @@ function createGitHubTracker(): Tracker { await gh(["issue", "reopen", identifier, "--repo", project.repo]); } + // Handle label removal + if (update.removeLabels && update.removeLabels.length > 0) { + await gh([ + "issue", + "edit", + identifier, + "--repo", + project.repo, + "--remove-label", + update.removeLabels.join(","), + ]); + } + // Handle label changes if (update.labels && update.labels.length > 0) { await gh([ diff --git a/packages/web/src/__tests__/api-routes.test.ts b/packages/web/src/__tests__/api-routes.test.ts index b2467a077..289f881e3 100644 --- a/packages/web/src/__tests__/api-routes.test.ts +++ b/packages/web/src/__tests__/api-routes.test.ts @@ -130,6 +130,10 @@ const mockRegistry: PluginRegistry = { loadFromConfig: vi.fn(async () => {}), }; +const mockLifecycleManager = { + getStates: vi.fn(() => new Map()), +}; + const mockConfig: OrchestratorConfig = { configPath: "/tmp/ao-test/agent-orchestrator.yaml", port: 3000, @@ -155,8 +159,10 @@ vi.mock("@/lib/services", () => ({ config: mockConfig, registry: mockRegistry, sessionManager: mockSessionManager, + lifecycleManager: mockLifecycleManager, })), getSCM: vi.fn(() => mockSCM), + startBacklogPoller: vi.fn(() => {}), })); // ── Import routes after mocking ─────────────────────────────────────── diff --git a/packages/web/src/__tests__/services.test.ts b/packages/web/src/__tests__/services.test.ts index 944cdc3f1..e4f94dd64 100644 --- a/packages/web/src/__tests__/services.test.ts +++ b/packages/web/src/__tests__/services.test.ts @@ -43,6 +43,18 @@ vi.mock("@composio/ao-core", () => ({ loadConfig: mockLoadConfig, createPluginRegistry: () => mockRegistry, createSessionManager: mockCreateSessionManager, + createLifecycleManager: () => ({ + start: vi.fn(), + stop: vi.fn(), + getStates: vi.fn(), + check: vi.fn(), + }), + decompose: vi.fn(), + getLeaves: vi.fn(), + getSiblings: vi.fn(), + formatPlanTree: vi.fn(), + DEFAULT_DECOMPOSER_CONFIG: {}, + TERMINAL_STATUSES: new Set(["merged", "killed"]) as ReadonlySet, })); vi.mock("@composio/ao-plugin-runtime-tmux", () => ({ default: tmuxPlugin })); @@ -97,3 +109,95 @@ describe("services", () => { expect(mockCreateSessionManager).toHaveBeenCalledTimes(1); }); }); + +describe("pollBacklog", () => { + const mockUpdateIssue = vi.fn(); + const mockListIssues = vi.fn(); + const mockSpawn = vi.fn(); + + beforeEach(async () => { + vi.resetModules(); + mockRegister.mockClear(); + mockCreateSessionManager.mockReset(); + mockLoadConfig.mockReset(); + mockUpdateIssue.mockClear(); + mockListIssues.mockClear(); + mockSpawn.mockClear(); + + mockLoadConfig.mockReturnValue({ + configPath: "/tmp/agent-orchestrator.yaml", + port: 3000, + readyThresholdMs: 300_000, + defaults: { runtime: "tmux", agent: "claude-code", workspace: "worktree", notifiers: [] }, + projects: { + "test-project": { + path: "/tmp/test-project", + tracker: { plugin: "github" }, + backlog: { label: "agent:backlog", maxConcurrent: 5 }, + }, + }, + notifiers: {}, + notificationRouting: { urgent: [], action: [], warning: [], info: [] }, + reactions: {}, + }); + + mockCreateSessionManager.mockReturnValue({ + spawn: mockSpawn, + list: vi.fn().mockResolvedValue([]), + }); + + delete (globalThis as typeof globalThis & { _aoServices?: unknown })._aoServices; + delete (globalThis as typeof globalThis & { _aoServicesInit?: unknown })._aoServicesInit; + }); + + afterEach(() => { + delete (globalThis as typeof globalThis & { _aoServices?: unknown })._aoServices; + delete (globalThis as typeof globalThis & { _aoServicesInit?: unknown })._aoServicesInit; + }); + + it("removes agent:backlog label when claiming an issue", async () => { + mockListIssues.mockResolvedValue([ + { + id: "123", + title: "Test Issue", + description: "Test description", + url: "https://github.com/test/test/issues/123", + state: "open", + labels: ["agent:backlog"], + }, + ]); + + mockRegistry.get.mockImplementation((slot: string) => { + if (slot === "tracker") { + return { + name: "github", + listIssues: mockListIssues, + updateIssue: mockUpdateIssue, + }; + } + if (slot === "agent") { + return { name: "claude-code" }; + } + if (slot === "runtime") { + return { name: "tmux" }; + } + if (slot === "workspace") { + return { name: "worktree" }; + } + return null; + }); + + const { pollBacklog } = await import("../lib/services"); + await pollBacklog(); + + expect(mockUpdateIssue).toHaveBeenCalledWith( + "123", + { + labels: ["agent:in-progress"], + removeLabels: ["agent:backlog"], + comment: "Claimed by agent orchestrator — session spawned.", + }, + expect.objectContaining({ tracker: { plugin: "github" } }), + ); + }); +}); diff --git a/packages/web/src/app/api/backlog/route.ts b/packages/web/src/app/api/backlog/route.ts new file mode 100644 index 000000000..27c869017 --- /dev/null +++ b/packages/web/src/app/api/backlog/route.ts @@ -0,0 +1,23 @@ +import { NextResponse } from "next/server"; +import { getBacklogIssues, startBacklogPoller } from "@/lib/services"; + +export const dynamic = "force-dynamic"; + +/** + * GET /api/backlog — List backlog issues (labeled agent:backlog) + * Also starts the backlog auto-claim poller on first call. + */ +export async function GET() { + // Start the backlog poller (idempotent) + startBacklogPoller(); + + try { + const issues = await getBacklogIssues(); + return NextResponse.json({ issues }); + } catch (err) { + return NextResponse.json( + { error: err instanceof Error ? err.message : "Failed to fetch backlog" }, + { status: 500 }, + ); + } +} diff --git a/packages/web/src/app/api/events/route.ts b/packages/web/src/app/api/events/route.ts index 2d37eeab8..653531f3a 100644 --- a/packages/web/src/app/api/events/route.ts +++ b/packages/web/src/app/api/events/route.ts @@ -1,4 +1,4 @@ -import { getServices } from "@/lib/services"; +import { getServices, startBacklogPoller } from "@/lib/services"; import { sessionToDashboard } from "@/lib/serialize"; import { getAttentionLevel } from "@/lib/types"; @@ -8,22 +8,28 @@ export const dynamic = "force-dynamic"; * GET /api/events — SSE stream for real-time lifecycle events * * Sends session state updates to connected clients. - * Polls SessionManager.list() on an interval (no SSE push from core yet). + * Also starts the lifecycle manager and backlog poller on first connection. */ export async function GET(): Promise { const encoder = new TextEncoder(); let heartbeat: ReturnType | undefined; let updates: ReturnType | undefined; + // Start backlog poller (idempotent — lifecycle manager is started in services.ts) + startBacklogPoller(); + const stream = new ReadableStream({ start(controller) { // Send initial snapshot void (async () => { try { - const { sessionManager } = await getServices(); + const { sessionManager, lifecycleManager } = await getServices(); const sessions = await sessionManager.list(); const dashboardSessions = sessions.map(sessionToDashboard); + // Include lifecycle state map for debugging/display + const lifecycleStates = Object.fromEntries(lifecycleManager.getStates()); + const initialEvent = { type: "snapshot", sessions: dashboardSessions.map((s) => ({ @@ -33,10 +39,10 @@ export async function GET(): Promise { attentionLevel: getAttentionLevel(s), lastActivityAt: s.lastActivityAt, })), + lifecycle: lifecycleStates, }; controller.enqueue(encoder.encode(`data: ${JSON.stringify(initialEvent)}\n\n`)); } catch { - // If services aren't available, send empty snapshot controller.enqueue( encoder.encode(`data: ${JSON.stringify({ type: "snapshot", sessions: [] })}\n\n`), ); @@ -62,7 +68,6 @@ export async function GET(): Promise { const sessions = await sessionManager.list(); dashboardSessions = sessions.map(sessionToDashboard); } catch { - // Transient service error — skip this poll, retry on next interval return; } @@ -79,7 +84,6 @@ export async function GET(): Promise { }; controller.enqueue(encoder.encode(`data: ${JSON.stringify(event)}\n\n`)); } catch { - // enqueue failure means the stream is closed — clean up both intervals clearInterval(updates); clearInterval(heartbeat); } diff --git a/packages/web/src/app/api/issues/route.ts b/packages/web/src/app/api/issues/route.ts new file mode 100644 index 000000000..607360b9d --- /dev/null +++ b/packages/web/src/app/api/issues/route.ts @@ -0,0 +1,104 @@ +import { type NextRequest, NextResponse } from "next/server"; +import { getServices } from "@/lib/services"; +import { validateString } from "@/lib/validation"; +import type { Tracker } from "@composio/ao-core"; + +export const dynamic = "force-dynamic"; + +/** + * GET /api/issues — List open issues from all configured trackers. + * Query params: ?state=open|closed|all&label=agent:backlog&project=Saas-code + */ +export async function GET(request: NextRequest) { + const { searchParams } = request.nextUrl; + const state = (searchParams.get("state") ?? "open") as "open" | "closed" | "all"; + const label = searchParams.get("label") ?? undefined; + const projectFilter = searchParams.get("project") ?? undefined; + + try { + const { config, registry } = await getServices(); + const allIssues: Array<{ projectId: string; id: string; title: string; url: string; state: string; labels: string[] }> = []; + + for (const [projectId, project] of Object.entries(config.projects)) { + if (projectFilter && projectId !== projectFilter) continue; + if (!project.tracker) continue; + + const tracker = registry.get("tracker", project.tracker.plugin); + if (!tracker?.listIssues) continue; + + try { + const issues = await tracker.listIssues( + { state, labels: label ? [label] : undefined, limit: 50 }, + project, + ); + for (const issue of issues) { + allIssues.push({ projectId, ...issue }); + } + } catch { + // Skip unavailable trackers + } + } + + return NextResponse.json({ issues: allIssues }); + } catch (err) { + return NextResponse.json( + { error: err instanceof Error ? err.message : "Failed to fetch issues" }, + { status: 500 }, + ); + } +} + +/** + * POST /api/issues — Create a new issue and optionally add to backlog. + * Body: { projectId, title, description, addToBacklog?: boolean } + */ +export async function POST(request: NextRequest) { + const body = (await request.json().catch(() => null)) as Record | null; + if (!body) { + return NextResponse.json({ error: "Invalid JSON body" }, { status: 400 }); + } + + const titleErr = validateString(body.title, "title", 200); + if (titleErr) { + return NextResponse.json({ error: titleErr }, { status: 400 }); + } + + const projectId = body.projectId as string; + if (!projectId) { + return NextResponse.json({ error: "projectId is required" }, { status: 400 }); + } + + try { + const { config, registry } = await getServices(); + const project = config.projects[projectId]; + if (!project) { + return NextResponse.json({ error: `Unknown project: ${projectId}` }, { status: 404 }); + } + + if (!project.tracker) { + return NextResponse.json({ error: "No tracker configured for this project" }, { status: 422 }); + } + + const tracker = registry.get("tracker", project.tracker.plugin); + if (!tracker?.createIssue) { + return NextResponse.json({ error: "Tracker does not support issue creation" }, { status: 422 }); + } + + const labels = body.addToBacklog ? ["agent:backlog"] : []; + const issue = await tracker.createIssue( + { + title: body.title as string, + description: (body.description as string) ?? "", + labels, + }, + project, + ); + + return NextResponse.json({ issue: { projectId, ...issue } }, { status: 201 }); + } catch (err) { + return NextResponse.json( + { error: err instanceof Error ? err.message : "Failed to create issue" }, + { status: 500 }, + ); + } +} diff --git a/packages/web/src/app/api/setup-labels/route.ts b/packages/web/src/app/api/setup-labels/route.ts new file mode 100644 index 000000000..7b0c32465 --- /dev/null +++ b/packages/web/src/app/api/setup-labels/route.ts @@ -0,0 +1,52 @@ +import { NextResponse } from "next/server"; +import { getServices } from "@/lib/services"; +import { execFile } from "node:child_process"; +import { promisify } from "node:util"; + +const execFileAsync = promisify(execFile); + +export const dynamic = "force-dynamic"; + +const LABELS = [ + { name: "agent:backlog", color: "6B7280", description: "Available for agent to claim" }, + { name: "agent:in-progress", color: "7C3AED", description: "Agent is working on this" }, + { name: "agent:blocked", color: "DC2626", description: "Agent is blocked" }, + { name: "agent:done", color: "16A34A", description: "Agent completed this" }, +]; + +/** + * POST /api/setup-labels — Create agent labels on all configured repos. + * Idempotent — skips labels that already exist. + */ +export async function POST() { + try { + const { config } = await getServices(); + const results: Array<{ repo: string; label: string; status: string }> = []; + + for (const project of Object.values(config.projects)) { + if (!project.repo) continue; + + for (const label of LABELS) { + try { + await execFileAsync("gh", [ + "label", "create", label.name, + "--repo", project.repo, + "--color", label.color, + "--description", label.description, + "--force", + ], { timeout: 10_000 }); + results.push({ repo: project.repo, label: label.name, status: "created" }); + } catch { + results.push({ repo: project.repo, label: label.name, status: "exists" }); + } + } + } + + return NextResponse.json({ results }); + } catch (err) { + return NextResponse.json( + { error: err instanceof Error ? err.message : "Failed to setup labels" }, + { status: 500 }, + ); + } +} diff --git a/packages/web/src/app/api/verify/route.ts b/packages/web/src/app/api/verify/route.ts new file mode 100644 index 000000000..5456b654f --- /dev/null +++ b/packages/web/src/app/api/verify/route.ts @@ -0,0 +1,91 @@ +import { type NextRequest, NextResponse } from "next/server"; +import { getVerifyIssues, getServices } from "@/lib/services"; +import type { Tracker } from "@composio/ao-core"; + +export const dynamic = "force-dynamic"; + +/** + * GET /api/verify — List issues with `merged-unverified` label + */ +export async function GET() { + try { + const issues = await getVerifyIssues(); + return NextResponse.json({ issues }); + } catch (err) { + return NextResponse.json( + { error: err instanceof Error ? err.message : "Failed to fetch verify issues" }, + { status: 500 }, + ); + } +} + +/** + * POST /api/verify — Verify or fail an issue + * Body: { issueId: string, projectId: string, action: "verify" | "fail", comment?: string } + */ +export async function POST(req: NextRequest) { + try { + const body = await req.json(); + const { issueId, projectId, action, comment } = body as { + issueId: string; + projectId: string; + action: "verify" | "fail"; + comment?: string; + }; + + if (!issueId || !projectId || !action) { + return NextResponse.json( + { error: "Missing required fields: issueId, projectId, action" }, + { status: 400 }, + ); + } + + if (action !== "verify" && action !== "fail") { + return NextResponse.json( + { error: 'action must be "verify" or "fail"' }, + { status: 400 }, + ); + } + + const { config, registry } = await getServices(); + const project = config.projects[projectId]; + if (!project?.tracker) { + return NextResponse.json({ error: `Project ${projectId} has no tracker` }, { status: 404 }); + } + + const tracker = registry.get("tracker", project.tracker.plugin); + if (!tracker?.updateIssue) { + return NextResponse.json({ error: "Tracker does not support updateIssue" }, { status: 500 }); + } + + if (action === "verify") { + await tracker.updateIssue( + issueId, + { + state: "closed", + labels: ["verified", "agent:done"], + removeLabels: ["merged-unverified"], + comment: comment || "Verified — fix confirmed on staging.", + }, + project, + ); + } else { + await tracker.updateIssue( + issueId, + { + labels: ["verification-failed"], + removeLabels: ["merged-unverified"], + comment: comment || "Verification failed — problem persists on staging.", + }, + project, + ); + } + + return NextResponse.json({ ok: true }); + } catch (err) { + return NextResponse.json( + { error: err instanceof Error ? err.message : "Failed to update issue" }, + { status: 500 }, + ); + } +} diff --git a/packages/web/src/app/page.tsx b/packages/web/src/app/page.tsx index 33e9d2f1e..68194f7e9 100644 --- a/packages/web/src/app/page.tsx +++ b/packages/web/src/app/page.tsx @@ -25,9 +25,11 @@ export default async function Home() { let sessions: DashboardSession[] = []; let orchestratorId: string | null = null; let globalPause: GlobalPauseState | null = null; + let projectIds: string[] = []; const projectName = getProjectName(); try { const { config, registry, sessionManager } = await getServices(); + projectIds = Object.keys(config.projects); const allSessions = await sessionManager.list(); globalPause = resolveGlobalPause(allSessions); @@ -113,6 +115,7 @@ export default async function Home() { orchestratorId={orchestratorId} projectName={projectName} initialGlobalPause={globalPause} + projectIds={projectIds} /> ); } diff --git a/packages/web/src/components/Dashboard.tsx b/packages/web/src/components/Dashboard.tsx index 76dd6f41f..cdca98b17 100644 --- a/packages/web/src/components/Dashboard.tsx +++ b/packages/web/src/components/Dashboard.tsx @@ -1,6 +1,6 @@ "use client"; -import { useMemo, useState } from "react"; +import { useMemo, useState, useEffect, useCallback, type FormEvent } from "react"; import { type DashboardSession, type DashboardStats, @@ -16,25 +16,45 @@ import { PRTableRow } from "./PRStatus"; import { DynamicFavicon } from "./DynamicFavicon"; import { useSessionEvents } from "@/hooks/useSessionEvents"; +interface BacklogIssue { + id: string; + title: string; + url: string; + state: string; + labels: string[]; + projectId: string; +} + interface DashboardProps { initialSessions: DashboardSession[]; stats: DashboardStats; orchestratorId?: string | null; projectName?: string; initialGlobalPause?: GlobalPauseState | null; + projectIds?: string[]; } +type Tab = "board" | "backlog" | "verify" | "prs"; + const KANBAN_LEVELS = ["working", "pending", "review", "respond", "merge"] as const; export function Dashboard({ initialSessions, - stats, + stats: _initialStats, orchestratorId, projectName, initialGlobalPause, + projectIds = [], }: DashboardProps) { const { sessions, globalPause } = useSessionEvents(initialSessions, initialGlobalPause ?? null); const [rateLimitDismissed, setRateLimitDismissed] = useState(false); + const [activeTab, setActiveTab] = useState("board"); + const [backlogIssues, setBacklogIssues] = useState([]); + const [backlogLoading, setBacklogLoading] = useState(false); + const [showCreateForm, setShowCreateForm] = useState(false); + const [verifyIssues, setVerifyIssues] = useState([]); + const [verifyLoading, setVerifyLoading] = useState(false); + const grouped = useMemo(() => { const zones: Record = { merge: [], @@ -57,6 +77,77 @@ export function Dashboard({ .sort((a, b) => mergeScore(a) - mergeScore(b)); }, [sessions]); + // Fetch backlog issues + const fetchBacklog = useCallback(async () => { + setBacklogLoading(true); + try { + const res = await fetch("/api/backlog"); + if (res.ok) { + const data = await res.json(); + setBacklogIssues(data.issues ?? []); + } + } catch { + // Non-critical + } finally { + setBacklogLoading(false); + } + }, []); + + // Fetch verify issues + const fetchVerify = useCallback(async () => { + setVerifyLoading(true); + try { + const res = await fetch("/api/verify"); + if (res.ok) { + const data = await res.json(); + setVerifyIssues(data.issues ?? []); + } + } catch { + // Non-critical + } finally { + setVerifyLoading(false); + } + }, []); + + useEffect(() => { + if (activeTab === "verify") { + fetchVerify(); + const interval = setInterval(fetchVerify, 30_000); + return () => clearInterval(interval); + } + }, [activeTab, fetchVerify]); + + const handleVerifyAction = async ( + issueId: string, + projectId: string, + action: "verify" | "fail", + ) => { + try { + const res = await fetch("/api/verify", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ issueId, projectId, action }), + }); + if (res.ok) { + setVerifyIssues((prev) => + prev.filter((i) => !(i.id === issueId && i.projectId === projectId)), + ); + } else { + console.error("Failed to update verify status:", await res.text()); + } + } catch (err) { + console.error("Failed to update verify status:", err); + } + }; + + useEffect(() => { + if (activeTab === "backlog") { + fetchBacklog(); + const interval = setInterval(fetchBacklog, 30_000); + return () => clearInterval(interval); + } + }, [activeTab, fetchBacklog]); + const handleSend = async (sessionId: string, message: string) => { const res = await fetch(`/api/sessions/${encodeURIComponent(sessionId)}/send`, { method: "POST", @@ -102,35 +193,93 @@ export function Dashboard({ [sessions], ); + const liveStats = useMemo( + () => ({ + totalSessions: sessions.length, + workingSessions: sessions.filter((s) => s.activity !== null && s.activity !== "exited") + .length, + openPRs: sessions.filter((s) => s.pr?.state === "open").length, + needsReview: sessions.filter( + (s) => s.pr && !s.pr.isDraft && s.pr.reviewDecision === "pending", + ).length, + }), + [sessions], + ); + + // Counts for tab badges + const backlogCount = backlogIssues.length; + const verifyCount = verifyIssues.length; + const prCount = openPRs.length; + const needsAttention = grouped.respond.length + grouped.merge.length; + return (
+ {/* Header */} -
+

- Orchestrator + {projectName ?? "Orchestrator"}

- +
- {orchestratorId && ( - - - orchestrator - + {orchestratorId && ( + - - - - )} + + orchestrator + + + + + )} +
+
+ + {/* Tab bar */} +
+ setActiveTab("board")} + badge={needsAttention > 0 ? needsAttention : undefined} + badgeColor="var(--color-status-error)" + > + Board + + setActiveTab("backlog")} + badge={backlogCount > 0 ? backlogCount : undefined} + badgeColor="var(--color-accent)" + > + Backlog + + setActiveTab("verify")} + badge={verifyCount > 0 ? verifyCount : undefined} + badgeColor="rgb(245, 158, 11)" + > + Verify + + setActiveTab("prs")} + badge={prCount > 0 ? prCount : undefined} + badgeColor="var(--color-status-ready)" + > + Pull Requests +
{globalPause && ( @@ -176,86 +325,500 @@ export function Dashboard({
)} - {/* Kanban columns for active zones */} - {hasKanbanSessions && ( -
- {KANBAN_LEVELS.map((level) => - grouped[level].length > 0 ? ( -
- -
- ) : null, + {/* Board tab */} + {activeTab === "board" && ( + <> + {/* Kanban columns for active zones */} + {hasKanbanSessions ? ( +
+ {KANBAN_LEVELS.map((level) => + grouped[level].length > 0 ? ( +
+ +
+ ) : null, + )} +
+ ) : ( + setActiveTab("backlog")} + className="rounded-md bg-[var(--color-accent)] px-4 py-2 text-[12px] font-semibold text-[var(--color-text-inverse)] hover:opacity-90" + > + View Backlog + + } + /> + )} + + {/* Done — full-width grid below Kanban */} + {grouped.done.length > 0 && ( +
+ +
+ )} + + )} + + {/* Backlog tab */} + {activeTab === "backlog" && ( +
+
+

+ Issues labeled{" "} + + agent:backlog + {" "} + are auto-claimed by agents. Max {5} concurrent. +

+ +
+ + {showCreateForm && ( + { + setShowCreateForm(false); + fetchBacklog(); + }} + onCancel={() => setShowCreateForm(false)} + /> + )} + + {backlogLoading && backlogIssues.length === 0 ? ( +
+ Loading backlog... +
+ ) : backlogIssues.length === 0 ? ( + + ) : ( +
+ {backlogIssues.map((issue) => ( + + ))} +
)}
)} - {/* Done — full-width grid below Kanban */} - {grouped.done.length > 0 && ( -
- + {/* Verify tab */} + {activeTab === "verify" && ( +
+
+

+ Issues labeled{" "} + + merged-unverified + {" "} + need human verification on staging. +

+
+ + {verifyLoading && verifyIssues.length === 0 ? ( +
+ Loading issues to verify... +
+ ) : verifyIssues.length === 0 ? ( + + ) : ( +
+ {verifyIssues.map((issue) => ( + handleVerifyAction(issue.id, issue.projectId, "verify")} + onFail={() => handleVerifyAction(issue.id, issue.projectId, "fail")} + /> + ))} +
+ )}
)} - {/* PR Table */} - {openPRs.length > 0 && ( -
-

- Pull Requests -

-
- - - - - - - - - - - - - {openPRs.map((pr) => ( - - ))} - -
- PR - - Title - - Size - - CI - - Review - - Unresolved -
-
-
+ {/* PRs tab */} + {activeTab === "prs" && ( + <> + {openPRs.length > 0 ? ( +
+
+ + + + + + + + + + + + + {openPRs.map((pr) => ( + + ))} + +
+ PR + + Title + + Size + + CI + + Review + + Unresolved +
+
+
+ ) : ( + + )} + )}
); } -function StatusLine({ stats }: { stats: DashboardStats }) { +// --------------------------------------------------------------------------- +// Sub-components +// --------------------------------------------------------------------------- + +function TabButton({ + active, + onClick, + badge, + badgeColor, + children, +}: { + active: boolean; + onClick: () => void; + badge?: number; + badgeColor?: string; + children: React.ReactNode; +}) { + return ( + + ); +} + +function EmptyState({ + title, + description, + action, +}: { + title: string; + description: string; + action?: React.ReactNode; +}) { + return ( +
+
{title}
+
+ {description} +
+ {action &&
{action}
} +
+ ); +} + +function BacklogCard({ issue }: { issue: BacklogIssue }) { + return ( + + + + + +
+
+ #{issue.id} {issue.title} +
+
+ {issue.projectId} + {issue.labels + .filter((l) => l !== "agent:backlog") + .map((label) => ( + + {label} + + ))} +
+
+ + queued + + + ); +} + +function VerifyCard({ + issue, + onVerify, + onFail, +}: { + issue: BacklogIssue; + onVerify: () => Promise; + onFail: () => Promise; +}) { + const [acting, setActing] = useState<"verify" | "fail" | null>(null); + + const handleAction = async (action: "verify" | "fail", handler: () => Promise) => { + setActing(action); + try { + await handler(); + } finally { + setActing(null); + } + }; + + return ( +
+ + + + +
+ + #{issue.id} {issue.title} + +
+ {issue.projectId} + {issue.labels + .filter((l) => l !== "merged-unverified") + .map((label) => ( + + {label} + + ))} +
+
+
+ + +
+
+ ); +} + +function CreateIssueForm({ + projectIds, + onCreated, + onCancel, +}: { + projectIds: string[]; + onCreated: () => void; + onCancel: () => void; +}) { + const [title, setTitle] = useState(""); + const [description, setDescription] = useState(""); + const [selectedProject, setSelectedProject] = useState(projectIds[0] ?? ""); + const [submitting, setSubmitting] = useState(false); + const [error, setError] = useState(null); + + const handleSubmit = async (e: FormEvent) => { + e.preventDefault(); + if (!title.trim() || !selectedProject) return; + + setSubmitting(true); + setError(null); + try { + const res = await fetch("/api/issues", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + projectId: selectedProject, + title: title.trim(), + description: description.trim(), + addToBacklog: true, + }), + }); + + if (!res.ok) { + const data = await res.json(); + throw new Error(data.error ?? "Failed to create issue"); + } + + setTitle(""); + setDescription(""); + onCreated(); + } catch (err) { + setError(err instanceof Error ? err.message : "Failed to create issue"); + } finally { + setSubmitting(false); + } + }; + + return ( +
+ {projectIds.length > 1 && ( +
+ +
+ )} +
+ setTitle(e.target.value)} + className="w-full rounded-md border border-[var(--color-border-default)] bg-[var(--color-bg-base)] px-3 py-2 text-[13px] text-[var(--color-text-primary)] placeholder-[var(--color-text-tertiary)] outline-none focus:border-[var(--color-accent)]" + autoFocus + /> +
+
+