diff --git a/project/ticket-186/README.md b/project/ticket-186/README.md new file mode 100644 index 00000000..9590e998 --- /dev/null +++ b/project/ticket-186/README.md @@ -0,0 +1,53 @@ +# Ticket 186: Reduce cyclomatic complexity: MultiAgentOrchestrator.run (CC=31) + +- **ID**: ticket-186 +- **Owner**: unresolved:human +- **Status**: IN_PROGRESS +- **Workflow state**: EDIT +- **Created**: 2026-09-20 + +## Goal and scope + +SESSION_EXECUTION_AUTHORIZATION: the queue owner instructed re-running the +STARTER-619 planfile ticket unmodified after freeing one application-workstream +WIP slot (done: ticket-185 published via PR #370 protected merge). + +code2llm reports `src.koru.multi_agent.MultiAgentOrchestrator.run` at +`src/koru/multi_agent.py:304` with cyclomatic complexity 31 (limit 15). +Extract the run() stages (GitHub sync, queue collection, dry-run plan, +worker reaping, backlog promotion, worker spawn) into focused helper methods +with unchanged behavior, bringing the method under the limit 15. + +## Acceptance criteria + +- [x] AC-01: `MultiAgentOrchestrator.run` cyclomatic complexity is below 15 + (code2llm re-run on this checkout, output outside the repo tree). +- [x] AC-02: `tests/test_multi_agent.py` passes unmodified. +- [x] AC-03: `ruff check` and `ruff format --check` pass on + `src/koru/multi_agent.py`. +- [x] AC-04: `./project/governance-check.sh --base origin/main` passes. + +## Validation evidence + +Recorded 2026-09-20 in `.worktrees/ticket-186--multi-agent-run-complexity` +(branch `ticket/186-multi-agent-run-complexity`, base `origin/main` = `52c5a3a2`): + +- AC-01: `code2llm -f all -o /tmp/opencode/code2llm-ticket186 + --no-chunk --exclude *.md --exclude plugins` — the report no longer lists + any CC hotspot in `multi_agent` (module row `CC=9`; the only remaining CC + finding is `_parse_layer_hotspot_suggestions CC=15` in an unrelated + pre-existing module). Independent AST recount: `run` CC 32 → 9; every + extracted helper ≤ 8 (`_sync_github_issues` 7, `_try_spawn_next_worker` 8, + `_reap_finished_workers` 5, `_promote_backlog_task` 5, + `_print_dry_run_plan` 3, `_collect_pending_tasks` 2). +- AC-02: `PYTHONPATH=src python -m pytest tests/test_multi_agent.py -q` — + 8 passed; plus `tests/test_cli_auto.py` — 10 passed total, unmodified. +- AC-03: `ruff check src/koru/multi_agent.py` — All checks passed; + `ruff format --check` — clean (after one `ruff format` pass on the edit). +- AC-04: `bash project/governance-check.sh --base origin/main` — + `GOV-PASS: passed (0 errors, 0 warnings)`. + +## Tracking boundary + +This directory contains the minimal reviewed intent. Optional participant prose +and raw command logs are not required delivery output. diff --git a/project/ticket-186/intent.json b/project/ticket-186/intent.json new file mode 100644 index 00000000..0dbda9cd --- /dev/null +++ b/project/ticket-186/intent.json @@ -0,0 +1,96 @@ +{ + "schema": "new-project.intent/v3", + "ticket": "ticket-186", + "summary": "Reduce cyclomatic complexity: MultiAgentOrchestrator.run (CC=31)", + "workstream": "application", + "classification": { + "kind": "SERVICE", + "priority": "P2", + "origin": "health" + }, + "allowedPaths": [ + "project/ticket-186/**", + "TODO.md", + "project/TICKETS.md", + "src/koru/multi_agent.py", + "tests/test_multi_agent.py" + ], + "forbiddenPaths": [ + "project/ticket-*/user-*.md" + ], + "stacks": [], + "dependsOn": [], + "conflictsWith": [], + "integrationTicket": null, + "delivery": { + "acceptedBaseSha": "52c5a3a2f8158553770ea98c0b2e9b5fb7972cd7", + "targetBranch": "main", + "outcome": "MultiAgentOrchestrator.run drops from cyclomatic complexity 31 to below the code2llm limit 15 by extracting behavior-preserving private helpers (_sync_github_issues, _collect_pending_tasks, _print_dry_run_plan, _reap_finished_workers, _promote_backlog_task, _try_spawn_next_worker). No public API, CLI, output, or scheduling semantics change; tests/test_multi_agent.py passes unmodified.", + "nonGoals": [ + "Changing any observable output, exit code, worker scheduling, or timeout semantics", + "Regenerating project/analysis.toon.yaml or project/planfile-tickets.yaml (integration-owned artifacts)", + "Renaming or removing any public name" + ], + "complexity": "S", + "estimatedMinutes": 45, + "budgets": { + "maxImplementationFiles": 1, + "maxAffectedComponents": 1, + "maxPublicInterfaceChanges": 0, + "maxRuntimeDependencies": 0 + }, + "architecture": { + "status": "accepted", + "decision": "Decompose the monolithic run() loop into single-responsibility private methods on MultiAgentOrchestrator; the loop keeps only orchestration and accounting, each helper owns one stage and returns explicit deltas so behavior is byte-for-byte equivalent.", + "components": [ + { + "name": "multi-agent-orchestrator", + "paths": [ + "src/koru/multi_agent.py", + "tests/test_multi_agent.py" + ] + } + ], + "responsibilityChanges": false, + "interfaceChanges": [], + "dataChanges": [], + "ui": { + "impact": "none", + "states": [], + "evidence": [] + }, + "rollback": "Revert the commit; the original single run() method is restored verbatim." + }, + "runtimeDependencies": [], + "validation": [ + { + "criterion": "AC-01", + "commands": [ + "code2llm -f all -o /tmp/opencode/code2llm-ticket186 --no-chunk --exclude *.md --exclude plugins (complexity re-measured on MultiAgentOrchestrator.run)" + ], + "evidence": "Re-run code2llm reports CC < 15 for MultiAgentOrchestrator.run." + }, + { + "criterion": "AC-02", + "commands": [ + "PYTHONPATH=src .venv/bin/python -m pytest tests/test_multi_agent.py -q" + ], + "evidence": "Existing multi-agent suite passes unmodified." + }, + { + "criterion": "AC-03", + "commands": [ + "ruff check src/koru/multi_agent.py && ruff format --check src/koru/multi_agent.py" + ], + "evidence": "Lint and format pass on the refactored module." + }, + { + "criterion": "AC-04", + "commands": [ + "bash project/governance-check.sh --base origin/main" + ], + "evidence": "Governance gate passes with zero errors." + } + ] + } +} diff --git a/src/koru/multi_agent.py b/src/koru/multi_agent.py index 2e05b991..c395bba9 100644 --- a/src/koru/multi_agent.py +++ b/src/koru/multi_agent.py @@ -133,9 +133,7 @@ def is_koru_project(path: Path) -> bool: if not path.is_dir() or path.name.startswith("."): return False return ( - (path / ".planfile").exists() - or (path / "koru.yaml").exists() - or (path / "project" / "new-ticket.sh").exists() + (path / ".planfile").exists() or (path / "koru.yaml").exists() or (path / "project" / "new-ticket.sh").exists() ) @@ -301,6 +299,120 @@ def build_worker_env(self) -> dict[str, str]: env["KORU_STDIO_FORMAT"] = env.get("KORU_STDIO_FORMAT", "human") return env + def _sync_github_issues(self, projects: list[Path]) -> None: + """Synchronize GitHub issues with Planfile when ``--sync`` was requested.""" + if not self.config.sync_github: + return + if self.config.dry_run: + print("koru auto: [dry-run] would synchronize GitHub issues with Planfile for configured projects.") + return + from concurrent.futures import ThreadPoolExecutor + + sync_targets = [p for p in projects if has_github_sync_config(p)] + if sync_targets: + print(f"koru auto: synchronizing GitHub issues across {len(sync_targets)} project(s)...") + with ThreadPoolExecutor(max_workers=min(len(sync_targets), 8)) as pool: + list(pool.map(sync_github_planfile, sync_targets)) + + def _collect_pending_tasks(self, projects: list[Path]) -> list[TaskItem]: + """Gather actionable tickets from every discovered project.""" + pending_queue: list[TaskItem] = [] + for proj in projects: + pending_queue.extend(get_pending_tasks_for_project(proj)) + return pending_queue + + def _print_dry_run_plan(self, pending_queue: list[TaskItem]) -> None: + """Preview queued tasks and their worker assignments without executing.""" + print("\n[Dry Run Plan]") + for i, task in enumerate(pending_queue[: self.config.max_tickets], start=1): + client_desc = f" (client: {self.config.client})" if self.config.client else "" + print(f" {i}. [{task.project.name}] {task.ticket_id}: {task.title}{client_desc}") + + def _reap_finished_workers(self) -> tuple[int, int]: + """Poll active workers; enforce timeouts and count completions. + + Updates ``self.active_workers`` to the survivors and returns the + ``(completed, failed)`` deltas observed during this pass. + """ + completed = 0 + failed = 0 + still_active: list[ActiveWorker] = [] + for worker in self.active_workers: + code = worker.process.poll() + if code is None: + if time.time() - worker.start_time > self.config.timeout_per_ticket: + print( + f"koru auto: worker [{worker.task.project.name} / {worker.task.ticket_id}] " + "timed out, terminating...", + file=sys.stderr, + ) + worker.process.terminate() + failed += 1 + else: + still_active.append(worker) + else: + duration = time.time() - worker.start_time + if code == 0: + print( + f"✓ [{worker.task.project.name}] {worker.task.ticket_id} " + f"completed successfully in {duration:.1f}s" + ) + completed += 1 + else: + print( + f"✗ [{worker.task.project.name}] {worker.task.ticket_id} " + f"exited with error code {code} ({duration:.1f}s)", + file=sys.stderr, + ) + failed += 1 + self.active_workers = still_active + return completed, failed + + def _promote_backlog_task(self, task: TaskItem) -> None: + """Move a backlog ticket into the current sprint before dispatching it.""" + if task.sprint == "backlog" and not self.config.dry_run and not self.config.worker_dry_run: + py = os.environ.get("PY") or sys.executable + subprocess.run( + [py, "-m", "planfile.cli", "ticket", "move", task.ticket_id, "current"], + cwd=str(task.project), + capture_output=True, + ) + task.sprint = "current" + + def _try_spawn_next_worker(self, queue: list[TaskItem]) -> tuple[bool, int]: + """Spawn the next eligible queued task when capacity allows. + + Keeps the safety invariant of at most one worker per project at a + time. Returns ``(spawn_attempted, failed_delta)``; a spawn attempt + always ends the current scheduling pass, mirroring the original loop. + """ + if self._interrupted or len(self.active_workers) >= self.config.workers: + return False, 0 + + active_project_paths = {w.task.project for w in self.active_workers} + task = next((t for t in queue if t.project not in active_project_paths), None) + if task is None: + return False, 0 + + queue.remove(task) + self._promote_backlog_task(task) + cmd = self.build_worker_command(task) + env = self.build_worker_env() + print( + f"▶ Spawning agent [{len(self.active_workers) + 1}/{self.config.workers}] " + f"for [{task.project.name}] {task.ticket_id}: {' '.join(cmd)}" + ) + try: + proc = subprocess.Popen(cmd, cwd=str(task.project), env=env) + self.active_workers.append(ActiveWorker(task=task, process=proc, start_time=time.time())) + except Exception as exc: + print( + f"koru auto: failed to start worker for [{task.project.name}] {task.ticket_id}: {exc}", + file=sys.stderr, + ) + return True, 1 + return True, 0 + def run(self) -> int: self.setup_signals() projects = discover_workspace_projects(self.config.workspace) @@ -314,24 +426,10 @@ def run(self) -> int: ) # 1. Sync GitHub if requested - if self.config.sync_github: - if self.config.dry_run: - print("koru auto: [dry-run] would synchronize GitHub issues with Planfile for configured projects.") - else: - from concurrent.futures import ThreadPoolExecutor - - sync_targets = [p for p in projects if has_github_sync_config(p)] - if sync_targets: - print(f"koru auto: synchronizing GitHub issues across {len(sync_targets)} project(s)...") - with ThreadPoolExecutor(max_workers=min(len(sync_targets), 8)) as pool: - list(pool.map(sync_github_planfile, sync_targets)) + self._sync_github_issues(projects) # 2. Collect pending tasks - pending_queue: list[TaskItem] = [] - for proj in projects: - tasks = get_pending_tasks_for_project(proj) - pending_queue.extend(tasks) - + pending_queue = self._collect_pending_tasks(projects) if not pending_queue: print("koru auto: no open Planfile tickets found across discovered projects.") return 0 @@ -339,10 +437,7 @@ def run(self) -> int: print(f"koru auto: {len(pending_queue)} actionable ticket(s) queued.") if self.config.dry_run: - print("\n[Dry Run Plan]") - for i, task in enumerate(pending_queue[: self.config.max_tickets], start=1): - client_desc = f" (client: {self.config.client})" if self.config.client else "" - print(f" {i}. [{task.project.name}] {task.ticket_id}: {task.title}{client_desc}") + self._print_dry_run_plan(pending_queue) return 0 completed_count = 0 @@ -350,83 +445,20 @@ def run(self) -> int: queue = pending_queue[: self.config.max_tickets] while (queue or self.active_workers) and not self._interrupted: - # Poll existing workers - still_active: list[ActiveWorker] = [] - for worker in self.active_workers: - code = worker.process.poll() - if code is None: - # Check timeout - if time.time() - worker.start_time > self.config.timeout_per_ticket: - print( - f"koru auto: worker [{worker.task.project.name} / {worker.task.ticket_id}] " - "timed out, terminating...", - file=sys.stderr, - ) - worker.process.terminate() - failed_count += 1 - else: - still_active.append(worker) - else: - duration = time.time() - worker.start_time - if code == 0: - print( - f"✓ [{worker.task.project.name}] {worker.task.ticket_id} " - f"completed successfully in {duration:.1f}s" - ) - completed_count += 1 - else: - print( - f"✗ [{worker.task.project.name}] {worker.task.ticket_id} " - f"exited with error code {code} ({duration:.1f}s)", - file=sys.stderr, - ) - failed_count += 1 - self.active_workers = still_active - - # Spawn new workers up to capacity - # Keep safety invariant: at most one worker per project simultaneously - active_project_paths = {w.task.project for w in self.active_workers} - - eligible_idx = None - for idx, task in enumerate(queue): - if task.project not in active_project_paths: - eligible_idx = idx - break - - if eligible_idx is not None and len(self.active_workers) < self.config.workers and not self._interrupted: - task = queue.pop(eligible_idx) - if task.sprint == "backlog" and not self.config.dry_run and not self.config.worker_dry_run: - py = os.environ.get("PY") or sys.executable - subprocess.run( - [py, "-m", "planfile.cli", "ticket", "move", task.ticket_id, "current"], - cwd=str(task.project), - capture_output=True, - ) - task.sprint = "current" - cmd = self.build_worker_command(task) - env = self.build_worker_env() - print( - f"▶ Spawning agent [{len(self.active_workers) + 1}/{self.config.workers}] " - f"for [{task.project.name}] {task.ticket_id}: {' '.join(cmd)}" - ) - try: - proc = subprocess.Popen(cmd, cwd=str(task.project), env=env) - self.active_workers.append( - ActiveWorker(task=task, process=proc, start_time=time.time()) - ) - except Exception as exc: - print( - f"koru auto: failed to start worker for [{task.project.name}] {task.ticket_id}: {exc}", - file=sys.stderr, - ) - failed_count += 1 + completed, failed = self._reap_finished_workers() + completed_count += completed + failed_count += failed + + # 3. Spawn new workers up to capacity + spawn_attempted, spawn_failed = self._try_spawn_next_worker(queue) + failed_count += spawn_failed + if spawn_attempted: continue time.sleep(0.5) print( - f"\nkoru auto: run finished. Completed: {completed_count}, " - f"Failed: {failed_count}, Remaining: {len(queue)}" + f"\nkoru auto: run finished. Completed: {completed_count}, Failed: {failed_count}, Remaining: {len(queue)}" ) return 0 if failed_count == 0 else 1