diff --git a/agent-team/agent_team/coordinator.py b/agent-team/agent_team/coordinator.py index 623899d..a8336c6 100644 --- a/agent-team/agent_team/coordinator.py +++ b/agent-team/agent_team/coordinator.py @@ -152,6 +152,7 @@ def default_slack_listener_factory( transport: SlackTransport, db_path: Path, enqueue_resume: Callable[[Any], None], + new_task_callback: "Callable[[str, str], str] | None" = None, ) -> Any: """Build the live :class:`SlackListener` from the coordinator's seams (D-1). @@ -166,6 +167,11 @@ def default_slack_listener_factory( (AUTHZ-01), so this factory deliberately does not weaken that — it injects no ``owner_ids`` and lets ``serve`` read + enforce them. + ``new_task_callback`` (WS2) is forwarded to the listener so an allowlisted + ``/new-task`` command starts a task. Left ``None``, the listener ignores + ``/new-task`` (its built-in default) — the ``/new-task`` path still runs + AFTER the AUTHZ-01 owner check regardless. + Imported lazily for the same import-hygiene reason as the clarifier / planner factories (the listener pulls the transport + responder leaves). """ @@ -177,6 +183,7 @@ def default_slack_listener_factory( enqueue_resume, app_token=os.environ.get("SLACK_APP_TOKEN") or None, bot_token=os.environ.get("SLACK_BOT_TOKEN") or None, + new_task_callback=new_task_callback, ) @@ -434,6 +441,7 @@ class Coordinator: deadline_window: timedelta | None = None, alarm_hook: AlarmHook | None = None, build_listener: ListenerFactory | None = None, + new_task_callback: "Callable[[str, str], str] | None" = None, ) -> None: self._db_path = Path(db_path) self._transport = transport @@ -462,6 +470,10 @@ class Coordinator: # the start/no-start/shutdown wiring with no live socket. Only consulted # by serve() when Slack is the live transport AND the app token is set. self._build_listener = build_listener + # WS2: optional /new-task handler forwarded to the Slack listener. Left + # None, the listener ignores /new-task. The serve path wires the + # coordinator's own start_task adapter via set_new_task_callback(). + self._new_task_callback = new_task_callback # Built by setup(). self._graph: Any = None @@ -478,6 +490,19 @@ class Coordinator: # Accessors (the shared queue is the slack_listener handoff seam). # ------------------------------------------------------------------ # + def set_new_task_callback( + self, callback: "Callable[[str, str], str] | None" + ) -> None: + """Wire the WS2 ``/new-task`` handler (call before :meth:`serve`). + + Avoids the constructor chicken-and-egg of referencing the coordinator's + own ``start_task`` at build time: the serve path constructs the + coordinator, then sets ``lambda text, source: self.start_task(...)``. Must + be set before :meth:`_maybe_start_slack_listener` runs (i.e. before + :meth:`serve`); it is read when the listener is built. + """ + self._new_task_callback = callback + @property def resume_queue(self) -> "queue.Queue[Any]": """The shared resume-job queue (slack_listener enqueues, drainer drains).""" @@ -1028,6 +1053,7 @@ class Coordinator: transport=self._transport, db_path=self._db_path, enqueue_resume=self._resume_queue.put, + new_task_callback=self._new_task_callback, ) self._listener = listener thread = threading.Thread( diff --git a/agent-team/run-team.py b/agent-team/run-team.py index 7d07e3a..33d914f 100644 --- a/agent-team/run-team.py +++ b/agent-team/run-team.py @@ -66,7 +66,7 @@ import sqlite3 import sys from datetime import datetime, timezone from pathlib import Path -from typing import Any, Sequence +from typing import Any, Callable, Sequence # ``run-team.py`` lives in ``agent-team/`` next to the importable ``agent_team`` # package. The hyphenated filename cannot itself be imported, so when run as a @@ -492,6 +492,28 @@ def _cmd_supersede(args: argparse.Namespace, *, out: Any) -> int: return 0 +def _build_context_provider() -> "Callable[[], str]": + """Return the WS5 (D10) context provider: the Sea Haven handbook conventions. + + The ``context_provider`` seam is zero-arg (``Callable[[], str]``), so it + supplies STATIC context — the handbook conventions loaded from + ``SEA_HAVEN_HANDBOOK_DIR`` (or the default dir) via + :func:`agent_team.nodes.handbook.load_handbook_conventions`. That loader is + itself fail-safe (caps file count/size; returns ``""`` and never raises when + the dir is absent/empty/unreadable), so wiring it is safe even on a box where + the handbook has not been synced yet. + + (Task-keyed memory retrieval is NOT wired here: the seam takes no task text, + so it cannot form a retrieval query — that would need a task-aware seam.) + + Imported lazily for the same import-hygiene reason as the coordinator + factories (keeps ``--help`` / ledger commands import-clean). + """ + from agent_team.nodes.handbook import load_handbook_conventions + + return load_handbook_conventions + + def _build_coordinator(args: argparse.Namespace) -> Any: """Construct a :class:`Coordinator` for the ``start`` / ``serve`` commands. @@ -512,15 +534,34 @@ def _build_coordinator(args: argparse.Namespace) -> Any: ) transport = _build_transport(args) + # WS5 (D10): inject the Sea Haven handbook conventions into the planner prompt + # via the context_provider seam. The provider is the zero-arg handbook loader, + # which is itself fail-safe (returns "" when the handbook dir is absent), and + # planner.plan_node additionally swallows provider errors — so this never + # affects a run where the handbook is unavailable. + context_provider = _build_context_provider() + # Production runs the full P2 graph: the wrapped real planner + the bound # GPT-4.1 review loop (Plane-2 depth-first). These factories are lazy and # only build/bind the model seams when a task actually runs. - return Coordinator( + coordinator = Coordinator( db_path=args.db, transport=transport, - build_plan_node=default_plan_node_factory, + build_plan_node=lambda: default_plan_node_factory( + context_provider=context_provider + ), review_wiring=default_review_wiring, ) + # WS2: an allowlisted Slack /new-task starts a task on THIS coordinator. Set + # post-construction (the adapter closes over the just-built coordinator), and + # before serve() builds the listener. AUTHZ-01 (owner allowlist) gates this + # upstream in the listener; the source label is always "slack". + coordinator.set_new_task_callback( + lambda task_text, _source: coordinator.start_task( + task_text=task_text, transport_name="slack" + ) + ) + return coordinator def _build_transport(args: argparse.Namespace) -> Any: diff --git a/agent-team/tests/test_run_team.py b/agent-team/tests/test_run_team.py index 86b3574..7ca0ce7 100644 --- a/agent-team/tests/test_run_team.py +++ b/agent-team/tests/test_run_team.py @@ -655,11 +655,15 @@ class _FakeCoordinator: self.review_wiring = review_wiring self.setup_called = False self.start_kwargs: dict[str, Any] | None = None + self.new_task_callback: Any = None _FakeCoordinator.instances.append(self) def setup(self) -> None: self.setup_called = True + def set_new_task_callback(self, callback: Any) -> None: + self.new_task_callback = callback + def start_task(self, *, task_text: str, transport_name: str) -> str: self.start_kwargs = {"task_text": task_text, "transport_name": transport_name} return "thread-minted-42" diff --git a/agent-team/tests/test_ws_activation_wiring.py b/agent-team/tests/test_ws_activation_wiring.py new file mode 100644 index 0000000..b0d1bd7 --- /dev/null +++ b/agent-team/tests/test_ws_activation_wiring.py @@ -0,0 +1,160 @@ +"""Activation-wiring tests (integration branch): prove the WS seams that the +``serve`` path flips ON are actually wired, without a live Slack socket. + +Covers: +* run-team ``_build_context_provider`` returns the handbook loader (WS5 / D10). +* run-team ``_build_coordinator`` threads that context_provider into the planner + node factory. +* run-team ``_build_coordinator`` sets a ``/new-task`` callback that starts a + task on the SAME coordinator with transport_name="slack" (WS2). +* ``default_slack_listener_factory`` forwards ``new_task_callback`` to the + SlackListener (WS2). +* ``Coordinator`` stores ``new_task_callback`` / ``set_new_task_callback`` and + forwards it when it builds the default listener. +""" + +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path +from types import SimpleNamespace +from typing import Any +from unittest.mock import MagicMock + + +_AGENT_TEAM_DIR = Path(__file__).resolve().parents[1] +if str(_AGENT_TEAM_DIR) not in sys.path: + sys.path.insert(0, str(_AGENT_TEAM_DIR)) + + +def _load_run_team(): + """Import run-team.py (hyphenated, so loaded by path) as a module.""" + cli_path = _AGENT_TEAM_DIR / "run-team.py" + spec = importlib.util.spec_from_file_location("run_team_cli", cli_path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _dry_args(tmp_path: Path) -> SimpleNamespace: + return SimpleNamespace( + db=str(tmp_path / "agent_team.sqlite"), + transport="slack", + dry_run=True, + ) + + +# --------------------------------------------------------------------------- # +# WS5: context_provider = handbook loader, threaded into the plan node +# --------------------------------------------------------------------------- # + + +def test_build_context_provider_returns_handbook_loader() -> None: + cli = _load_run_team() + provider = cli._build_context_provider() + assert callable(provider) + # Zero-arg and returns a string (fail-safe: "" when no handbook dir). + result = provider() + assert isinstance(result, str) + + +def test_build_coordinator_threads_context_provider_into_plan_node( + tmp_path: Path, monkeypatch: Any +) -> None: + cli = _load_run_team() + from agent_team import coordinator as coord_mod + + captured: dict[str, Any] = {} + + def _spy_plan_factory(context_provider: Any = None): + captured["context_provider"] = context_provider + return lambda state: state + + monkeypatch.setattr(coord_mod, "default_plan_node_factory", _spy_plan_factory) + + coordinator = cli._build_coordinator(_dry_args(tmp_path)) + # The plan node factory the coordinator holds is the run-team lambda; calling + # it must invoke default_plan_node_factory WITH a non-None context_provider. + coordinator._build_plan_node() + assert "context_provider" in captured + assert callable(captured["context_provider"]) + + +# --------------------------------------------------------------------------- # +# WS2: /new-task callback wired to this coordinator's start_task +# --------------------------------------------------------------------------- # + + +def test_build_coordinator_wires_new_task_callback_to_start_task( + tmp_path: Path, monkeypatch: Any +) -> None: + cli = _load_run_team() + coordinator = cli._build_coordinator(_dry_args(tmp_path)) + + # Replace start_task so we can observe the callback routing without running + # the real graph. + calls: dict[str, Any] = {} + + def _fake_start_task(*, task_text: str, transport_name: str) -> str: + calls["task_text"] = task_text + calls["transport_name"] = transport_name + return "thread-xyz" + + monkeypatch.setattr(coordinator, "start_task", _fake_start_task) + + cb = coordinator._new_task_callback + assert cb is not None + thread_id = cb("fix the flaky test", "slack") + assert thread_id == "thread-xyz" + assert calls == {"task_text": "fix the flaky test", "transport_name": "slack"} + + +def test_set_new_task_callback_overrides() -> None: + from agent_team.coordinator import Coordinator + from agent_team.transport.base import Transport + + class _T(Transport): + def post_question(self, **kw: Any) -> str: # type: ignore[override] + return "q" + + def parse_answer(self, raw: Any): # type: ignore[override] + raise NotImplementedError + + coord = Coordinator(db_path=":memory:", transport=_T()) + assert coord._new_task_callback is None + sentinel = lambda t, s: "tid" # noqa: E731 + coord.set_new_task_callback(sentinel) + assert coord._new_task_callback is sentinel + + +# --------------------------------------------------------------------------- # +# WS2: factory + coordinator forward new_task_callback to the SlackListener +# --------------------------------------------------------------------------- # + + +def test_slack_listener_factory_forwards_new_task_callback( + tmp_path: Path, monkeypatch: Any +) -> None: + import agent_team.coordinator as coord_mod + + captured: dict[str, Any] = {} + + class _FakeListener: + def __init__(self, *args: Any, **kwargs: Any) -> None: + captured["new_task_callback"] = kwargs.get("new_task_callback") + + # Patch the lazily-imported SlackListener symbol. + import agent_team.transport.slack_listener as sl_mod + + monkeypatch.setattr(sl_mod, "SlackListener", _FakeListener) + + sentinel = lambda t, s: "tid" # noqa: E731 + coord_mod.default_slack_listener_factory( + transport=MagicMock(), + db_path=tmp_path / "x.sqlite", + enqueue_resume=lambda _x: None, + new_task_callback=sentinel, + ) + assert captured["new_task_callback"] is sentinel