cw-hans/the-agency
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

refactor(team_runner): make runtimes config-driven — replace hardcoded slots with dict

Browse Source 
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Hans Heinemann
2026-03-16 00:50:17 -04:00
parent 71316b3090
commit 7b1cf7315c
2 changed files with 53 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
config/team.yaml
View File

@@ -7,7 +7,6 @@ adapters:
llm: anthropic
vcs: github
notify: openclaw
runtime: openclaw
models:
default_max_tokens: 4096

69
core/team_runner.py
View File

@@ -5,6 +5,12 @@ Top-level orchestration entry point for the-agency pipeline.
The TeamRunner loads team.yaml, builds the adapter registry, and drives the
full T1 → T2 → T3 → T4 → T5 dispatch loop with escalation handling.
Runtime adapters are config-driven: every string-valued key in the top-level
``runtime:`` section of team.yaml is instantiated as a RuntimeAdapter and
stored in ``self._runtimes[name]``. Non-string values (e.g. ``native_teams:
false``) are silently skipped. Dispatch routing uses
``brief.preferred_runtime`` to look up the right adapter at call time.
CLI usage::
python -m core.team_runner --config config/team.yaml [--dry-run] [--verbose]
@@ -151,6 +157,12 @@ class TeamRunner:
config_path : Path to team.yaml.
dry_run : When True, skip LLM calls, VCS commits, and notifications.
All planned actions are logged at INFO level.
Runtime adapters are built from the top-level ``runtime:`` section of
team.yaml. Each string-valued key becomes an entry in
``self._runtimes``; non-string values (e.g. ``native_teams: false``)
are ignored. Adding a new runtime type requires only a new key in
team.yaml — no changes to TeamRunner are needed.
"""
self._dry_run = dry_run
@@ -172,12 +184,9 @@ class TeamRunner:
self._notify: Optional[NotifyAdapter] = self._build_optional( # type: ignore[assignment]
_NOTIFY_ADAPTERS, adapter_cfg.get("notify"), "notify"
)
self._default_runtime: RuntimeAdapter = self._build_runtime(
runtime_cfg.get("default", "openclaw")
)
self._coding_runtime: RuntimeAdapter = self._build_runtime(
runtime_cfg.get("coding_agent", "claude_code")
)
# Runtime adapters are fully config-driven — one entry per string-valued
# key in the top-level ``runtime:`` section of team.yaml.
self._runtimes: dict[str, RuntimeAdapter] = self._build_runtimes(runtime_cfg)
logger.info(
"TeamRunner initialised: run_id=%s dry_run=%s", run_id, dry_run
@@ -229,6 +238,22 @@ class TeamRunner:
cls = _load_adapter_class(key, _RUNTIME_ADAPTERS, "runtime")
return cls(self._config)
def _build_runtimes(self, runtime_cfg: dict) -> dict[str, RuntimeAdapter]:
"""
Build a name → RuntimeAdapter mapping from the ``runtime:`` config block.
Every key whose value is a string is treated as a runtime adapter name
and instantiated via ``_build_runtime``. Non-string values (e.g.
``native_teams: false``) are skipped so that boolean/numeric control
flags can coexist in the same config section.
"""
runtimes: dict[str, RuntimeAdapter] = {}
for name, value in runtime_cfg.items():
if not isinstance(value, str):
continue
runtimes[name] = self._build_runtime(value)
return runtimes
# ------------------------------------------------------------------
# Role registry
# ------------------------------------------------------------------
@@ -298,8 +323,9 @@ class TeamRunner:
Routing
-------
preferred_runtime == "coding_agent" → coding runtime adapter
preferred_runtime == "standard" → LLM adapter directly
preferred_runtime == "standard" or empty/None → LLM adapter directly
Otherwise → look up self._runtimes[preferred_runtime]; falls back to
self._runtimes["default"] and then to LLM if no runtime is found.
Blackboard events emitted: spawned → completed | failed.
"""
@@ -320,10 +346,21 @@ class TeamRunner:
)
try:
if brief.preferred_runtime == "coding_agent":
result = self._dispatch_via_runtime(brief)
else:
pref = brief.preferred_runtime
if not pref or pref == "standard":
result = self._dispatch_via_llm(brief)
else:
runtime = self._runtimes.get(pref) or self._runtimes.get("default")
if runtime is None:
logger.warning(
"No runtime adapter found for %r (and no 'default') — "
"falling back to LLM for brief %s",
pref,
brief.brief_id,
)
result = self._dispatch_via_llm(brief)
else:
result = self._dispatch_via_runtime(brief, runtime)
self._bb.update_brief_result(brief.brief_id, result)
self._bb.log_event(
@@ -359,22 +396,22 @@ class TeamRunner:
)
return self._extract_json(raw)
def _dispatch_via_runtime(self, brief: TaskBrief) -> dict:
"""Spawn a coding agent via the runtime adapter and collect its result."""
def _dispatch_via_runtime(self, brief: TaskBrief, runtime: RuntimeAdapter) -> dict:
"""Spawn an agent via *runtime* and collect its result."""
task_str = json.dumps(brief.to_dict(), indent=2)
capability = _TIER_CAPABILITIES.get(brief.tier, "capable")
timeout_s: int = brief.context.get("timeout_s", 300)
agent_id = self._coding_runtime.spawn(
agent_id = runtime.spawn(
task=task_str,
capability=capability,
context=brief.context,
)
logger.info(
"Spawned coding agent %s for brief %s", agent_id, brief.brief_id
"Spawned runtime agent %s for brief %s", agent_id, brief.brief_id
)
result = self._coding_runtime.get_result(agent_id, timeout_s=timeout_s)
result = runtime.get_result(agent_id, timeout_s=timeout_s)
# Attempt to parse JSON from the agent's text output.
if isinstance(result.get("output"), str) and result["output"].strip():