100 lines
4.1 KiB
Python
100 lines
4.1 KiB
Python
"""
|
|
core/team_runner.py
|
|
Top-level orchestration entry point — Phase 2 stub.
|
|
|
|
The TeamRunner is responsible for:
|
|
1. Loading config/team.yaml and config/role_registry.yaml.
|
|
2. Instantiating the correct adapter implementations (LLM, VCS, notify, runtime).
|
|
3. Creating a Blackboard for the run.
|
|
4. Constructing the root T1 TaskBrief and dispatching it to the T1 Visionary.
|
|
5. Recursively spawning T2→T5 briefs based on tier outputs.
|
|
6. Using EscalationHandler to manage retries, salvage, and escalation.
|
|
7. Writing final run status and summary to the Blackboard.
|
|
|
|
TODO (Phase 2):
|
|
- Load and validate team.yaml configuration.
|
|
- Build adapter registry (map adapter keys → concrete adapter classes).
|
|
- Implement tier dispatch loop: T1 → T2 (per workstream) → T3 → T4 → T5.
|
|
- Parse tier JSON outputs into child TaskBrief objects via make_child_brief().
|
|
- Integrate EscalationHandler into the dispatch loop.
|
|
- Support --dry-run flag (log actions without executing).
|
|
- Emit blackboard events at each stage (spawned, completed, failed, etc.).
|
|
- Expose a CLI entry point (argparse or click).
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
# TODO (Phase 2): Uncomment and implement imports as adapters are built.
|
|
# import argparse
|
|
# import yaml
|
|
# from core.task_brief import TaskBrief
|
|
# from core.blackboard import Blackboard
|
|
# from core.escalation import EscalationHandler
|
|
# from adapters.llm.anthropic import AnthropicAdapter
|
|
# from adapters.vcs.github import GitHubAdapter
|
|
# from adapters.notify.openclaw import OpenClawNotifyAdapter
|
|
# from adapters.runtime.openclaw import OpenClawRuntimeAdapter
|
|
# from adapters.runtime.claude_code import ClaudeCodeRuntimeAdapter
|
|
|
|
|
|
class TeamRunner:
|
|
"""
|
|
Orchestrates a full T1→T5 agent pipeline run.
|
|
|
|
Usage (Phase 2)::
|
|
|
|
runner = TeamRunner(config_path="config/team.yaml")
|
|
runner.run()
|
|
"""
|
|
|
|
def __init__(self, config_path: str = "config/team.yaml") -> None:
|
|
# TODO (Phase 2): Load YAML config.
|
|
# Instantiate adapters based on config.adapters keys.
|
|
# Create a Blackboard for this run.
|
|
raise NotImplementedError("TeamRunner.__init__ is not yet implemented.")
|
|
|
|
def run(self) -> None:
|
|
"""
|
|
Execute the full pipeline from T1 decomposition through T5 verification.
|
|
|
|
TODO (Phase 2):
|
|
- Build root T1 brief from config.run.goal.
|
|
- Dispatch to T1 Visionary via LLM adapter.
|
|
- Parse workstreams from T1 output.
|
|
- For each workstream: dispatch T2 Architect.
|
|
- For each T2 subtask: dispatch T3 Squad Lead.
|
|
- For each T3 task: dispatch T4 Implementer.
|
|
- For each T4 artifact set: dispatch T5 Verifier.
|
|
- Run escalation handler at each tier on failure.
|
|
- Commit passing artifacts via VCS adapter.
|
|
- Notify on completion or terminal failure via notify adapter.
|
|
"""
|
|
raise NotImplementedError("TeamRunner.run is not yet implemented.")
|
|
|
|
def _dispatch_brief(self, brief) -> dict:
|
|
"""
|
|
Send a single TaskBrief to the appropriate agent and return the result.
|
|
|
|
TODO (Phase 2):
|
|
- Select runtime based on brief.preferred_runtime.
|
|
- Load agent personality from brief.agent_personality (if set).
|
|
- Compose prompt from tier system prompt + brief payload.
|
|
- Spawn agent via runtime adapter.
|
|
- Await result via runtime.get_result().
|
|
- Log spawned/completed/failed events to Blackboard.
|
|
"""
|
|
raise NotImplementedError("TeamRunner._dispatch_brief is not yet implemented.")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI entry point (Phase 2)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# TODO (Phase 2): Implement argparse CLI.
|
|
# if __name__ == "__main__":
|
|
# parser = argparse.ArgumentParser(description="Run the-agency pipeline.")
|
|
# parser.add_argument("--config", default="config/team.yaml", help="Path to team.yaml")
|
|
# parser.add_argument("--dry-run", action="store_true", help="Log actions without executing")
|
|
# args = parser.parse_args()
|
|
# runner = TeamRunner(config_path=args.config)
|
|
# runner.run()
|