feat(integration): wire WS5 context_provider + WS2 /new-task into serve

Integration branch combining WS0-WS5 (PRs #43-#46) + the activation wiring that
flips the safe seams ON in the run-team serve path:

- WS5 (D10): inject the Sea Haven handbook conventions into the planner prompt
  via context_provider (zero-arg handbook loader; fail-safe to '' when absent).
- WS2: an allowlisted Slack /new-task starts a task on this coordinator
  (set_new_task_callback adapter -> start_task; AUTHZ-01 gates it upstream).
- WS1 bind_multi_invoker() is already wired in _cmd_serve.

Coordinator gains a new_task_callback param + set_new_task_callback() (resolves
the constructor chicken-and-egg of referencing the coordinator's own start_task);
default_slack_listener_factory forwards it to the SlackListener.

NOT wired (deliberately): the P3 dispatch_node / build_verify path. Activating
it correctly needs a per-task expected_run_id bound into gated_build_verify_wiring
(plumbing that does not exist yet) AND the CI trust-boundary security re-review.
It stays inert pending that work.

Tests: +5 activation-wiring tests; _FakeCoordinator stub gains
set_new_task_callback. Full agent-team suite: 1140 passed, ruff clean.

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(

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:

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"

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