code review

AGENTS.md

@@ -1,6 +1,6 @@
 # pitfalls
 
-prefer asserts dict['field'] == dict['field'] over assert dict == dict unless we know better (eg. no extra fields allowed)
+prefer asserts actual['field'] == expected['field'] over assert actual == expected unless we know better (eg. no extra fields allowed)
 
 more later
 

docs/superpowers/plans/2026-06-01-wf-api-extraction-roadmap.md

@@ -11,12 +11,13 @@
 
 **Architecture:** `wf_api` becomes the long-lived in-process application service layer. `wf_mcp`, `wf_cli`, and future HTTP/UI adapters call `wf_api`; `wf_api` must not import `wf_mcp`.
 
-**Current State:** Slice 1 introduced `wf_api.WorkflowApi`,
+**Current State:** Slice 1 originally introduced `wf_api.WorkflowApi`, a
-`wf_api.WorkflowApiBackend`, and `wf_mcp.broker.service.WfMcpWorkflowApiBackend`.
+`WorkflowApiBackend` protocol, and an MCP adapter backend. Later slices removed
-Both CLI and MCP workflow tools now call `WorkflowApi`; `wf_api` imports no
+that double-delegation seam: `WorkflowApi` now composes domain services directly
-`wf_mcp` modules. `WorkflowSurfaceHandlers` still contains the existing
+from `WorkflowOperationContext`. Both CLI and MCP workflow tools call
-operation implementation and still depends on `WfMcpService`, but it is now
+`WorkflowApi`; `wf_api` imports no `wf_mcp` modules. `WorkflowSurfaceHandlers`
-MCP-owned backend plumbing rather than the public application API.
+is now a thin MCP compatibility subclass rather than the operation
+implementation.
 
 Slice 3 moved the protocol-neutral workflow helpers into `wf_api`: constants,
 capability refs, wrapper hints, next actions, raw workflow plan model, runtime
@@ -24,9 +25,11 @@ dependency resolution, saved subgraph preparation, and durable run lifecycle
 helpers. The old `wf_mcp.workflow_surface.*` module paths remain compatibility
 shims for those helpers.
 
-**Current Constraint:** `WorkflowSurfaceHandlers` is large and still carries
+**Current Constraint:** `WfMcpService` still acts as a compatibility facade over
-most workflow-surface logic. Slice 1 fixed dependency direction only; later
+focused broker services. Recent slices have extracted source/catalog, runtime,
-slices can split and rename once the boundary is correct.
+upstream transport, events, connection sync, and content access; the remaining
+work is to keep shrinking facade responsibilities while preserving process-local
+behavior.
 
 ---
 

docs/superpowers/plans/2026-06-02-wfmcpservice-connection-service-extraction.md

@@ -34,6 +34,7 @@
 ### Task 1: Add Direct ConnectionService Tests
 
 **Files:**
+
 - Create: `tests/wf_mcp/service/test_connection_service.py`
 
 - [ ] **Step 1: Create direct tests for the new service boundary**
@@ -161,6 +162,7 @@ Expected: import failure for `wf_mcp.broker.service.connection_service`.
 ### Task 2: Implement ConnectionService
 
 **Files:**
+
 - Create: `src/wf_mcp/broker/service/connection_service.py`
 
 - [ ] **Step 1: Add the service implementation**
@@ -270,6 +272,7 @@ Expected: all checks pass.
 ### Task 3: Wire WfMcpService Through ConnectionService
 
 **Files:**
+
 - Modify: `src/wf_mcp/broker/service/core.py`
 
 - [ ] **Step 1: Update imports and dataclass fields**
@@ -379,6 +382,7 @@ Expected: all selected tests pass.
 ### Task 4: Add Facade Compatibility Tests
 
 **Files:**
+
 - Modify: `tests/wf_mcp/service/test_connection_service.py`
 
 - [ ] **Step 1: Add WfMcpService compatibility coverage**
@@ -446,6 +450,7 @@ Expected: all tests pass.
 ### Task 5: Clean Imports and Verify Call Sites
 
 **Files:**
+
 - Modify: `src/wf_mcp/broker/service/core.py`
 - Possibly modify files only if ruff reports stale imports.
 
@@ -458,6 +463,7 @@ rg -n 'parse_connection_id|RESERVED_CONNECTION_IDS|connection_service|Connection
 ```
 
 Expected:
+
 - `parse_connection_id` and `RESERVED_CONNECTION_IDS` appear in `connection_service.py`, not `core.py`.
 - `connections: ConnectionRegistry` appears in `connection_service.py`, not `core.py`.
 - `connection_service` appears in `core.py` and direct tests.
@@ -487,6 +493,7 @@ Expected: 0 errors.
 ### Task 6: Update Roadmap and Extraction Map
 
 **Files:**
+
 - Modify: `docs/current_roadmap.md`
 - Modify if present/relevant: `docs/superpowers/research/2026-06-01-wf-api-extraction-map.md`
 
@@ -538,6 +545,7 @@ rg -n 'ConnectionService|Connection ownership|Next planned service extraction' d
 ```
 
 Expected:
+
 - `docs/current_roadmap.md` mentions completed `ConnectionService` ownership.
 - No stale "Next planned service extraction" wording for this same slice remains.
 
@@ -546,6 +554,7 @@ Expected:
 ### Task 7: Final Verification
 
 **Files:**
+
 - No new files.
 
 - [ ] **Step 1: Run focused tests**
@@ -579,6 +588,7 @@ uv run basedpyright --level error
 ```
 
 Expected:
+
 - ruff check passes.
 - ruff format check passes for Python files.
 - basedpyright reports 0 errors.

src/wf_mcp/broker/service/connection_service.py

@@ -51,17 +51,46 @@ class ConnectionService:
         next_ids = {connection.id for connection in config.connections}
         previous_ids = set(self.connections.connections)
         for connection_id in previous_ids - next_ids:
+            previous = self.connections.connections[connection_id]
+            # This is the low-level config reconciliation path. ConnectionRegistry
+            # and SourceCatalogService do not yet expose paired unregister methods,
+            # so this method owns direct mutation plus the observable events.
             del self.connections.connections[connection_id]
             source_catalog.capability_sources.pop(connection_id, None)
+            self.events.record_kind(
+                "connection_removed",
+                connection_id=connection_id,
+                payload={"server": previous.server, "account": previous.account},
+            )
 
         for connection in config.connections:
             self._validate_connection_id(connection.id)
+            previous = self.connections.connections.get(connection.id)
             self.connections.register(connection)
             source = source_catalog.capability_sources.get(connection.id)
             if source is None:
                 source_catalog.hydrate_connection_source_from_snapshot(connection)
             else:
                 source.enabled = connection.enabled
+            if previous is None:
+                self.events.record_kind(
+                    "connection_registered",
+                    connection_id=connection.id,
+                    payload={
+                        "server": connection.server,
+                        "account": connection.account,
+                    },
+                )
+            elif previous != connection:
+                self.events.record_kind(
+                    "connection_updated",
+                    connection_id=connection.id,
+                    payload={
+                        "server": connection.server,
+                        "account": connection.account,
+                        "enabled": connection.enabled,
+                    },
+                )
 
     def _source_catalog(self) -> SourceCatalogService:
         if self.source_catalog is None:

src/wf_mcp/broker/service/workflow_operation_context.py

@@ -15,6 +15,7 @@ from wf_api.operation_context import (
 )
 from .core import WfMcpService
 from .events import BrokerEventRecorder
+from .source_catalog import SourceCatalogService
 from .workflow_runtime import WorkflowRuntimeService
 
 
@@ -43,16 +44,16 @@ class WfMcpWorkflowEventRecorder(WorkflowEventRecorder):
 
 @dataclass(frozen=True, slots=True)
 class WfMcpWorkflowSpecProvider(WorkflowSpecProvider):
-    """Adapter-owned spec provider backed by WfMcpService."""
+    """Adapter-owned spec provider backed by SourceCatalogService."""
 
-    service: WfMcpService
+    source_catalog: SourceCatalogService
 
     @property
     def capability_sources(self):
-        return self.service.source_catalog.capability_sources
+        return self.source_catalog.capability_sources
 
     def get_qualified_spec(self, qualified_name: str) -> NodeSpec[Any, Any]:
-        return self.service.source_catalog.get_qualified_spec(qualified_name)
+        return self.source_catalog.get_qualified_spec(qualified_name)
 
 
 @dataclass(frozen=True, slots=True)
@@ -120,7 +121,7 @@ class WfMcpWorkflowLiveSourceChecker(WorkflowLiveSourceChecker):
 
 def context_from_service(service: WfMcpService) -> WorkflowOperationContext:
     """Adapt the current MCP service stack into a protocol-neutral context."""
-    specs = WfMcpWorkflowSpecProvider(service)
+    specs = WfMcpWorkflowSpecProvider(service.source_catalog)
     return WorkflowOperationContext(
         artifact_store=service.artifact_store,
         draft_workspace_store=service.draft_workspace_store,

src/wf_mcp/broker/service/workflow_runtime.py

@@ -58,10 +58,10 @@ class WorkflowRuntimeService:
 
         nodes = []
         for node in plan.nodes:
-            payload = node.model_dump(by_alias=True)
+            node_payload = node.model_dump(by_alias=True)
             if isinstance(node, NodeUse):
-                payload["node"] = bindings.get(node.node, node.node)
+                node_payload["node"] = bindings.get(node.node, node.node)
-            nodes.append(payload)
+            nodes.append(node_payload)
 
         payload = {
             "name": plan.name,

tests/wf_api/test_operation_context.py

@@ -2,7 +2,6 @@ from __future__ import annotations
 
 import ast
 import json
-import tempfile
 from pathlib import Path
 
 from wf_api.operation_context import WorkflowOperationContext
@@ -12,12 +11,8 @@ from wf_mcp.broker.service.workflow_operation_context import context_from_servic
 from wf_mcp.storage import FileStore
 
 
-def _local_temp_root() -> Path:
+def test_context_uses_source_catalog_mapping(tmp_path: Path) -> None:
-    return Path(tempfile.mkdtemp())
+    service = WfMcpService(store=FileStore(tmp_path / "context_sources"))
-
-
-def test_context_uses_source_catalog_mapping() -> None:
-    service = WfMcpService(store=FileStore(_local_temp_root() / "context_sources"))
     context = context_from_service(service)
 
     assert context.specs.capability_sources is service.source_catalog.capability_sources
@@ -113,8 +108,8 @@ def test_context_from_service_record_workflow_event(tmp_path: Path) -> None:
     assert recorded.payload["version"] == 1
 
 
-def test_context_runtime_runner_uses_workflow_runtime_service() -> None:
+def test_context_runtime_runner_uses_workflow_runtime_service(tmp_path: Path) -> None:
-    service = WfMcpService(store=FileStore(_local_temp_root() / "context_runtime"))
+    service = WfMcpService(store=FileStore(tmp_path / "context_runtime"))
     context = context_from_service(service)
 
     assert getattr(context.runtime, "runtime") is service.workflow_runtime

tests/wf_mcp/service/test_connection_service.py

@@ -77,6 +77,11 @@ def test_connection_service_sync_removes_retired_connections_and_sources() -> No
 
     assert service.list_all() == []
     assert "demo.personal" not in catalog.capability_sources
+    removed = service.events.list_events()[-1]
+    assert removed.kind == "connection_removed"
+    assert removed.connection_id == "demo.personal"
+    assert removed.payload["server"] == "demo"
+    assert removed.payload["account"] == "personal"
 
 
 def test_connection_service_sync_updates_existing_source_enabled_flag() -> None:
@@ -102,6 +107,34 @@ def test_connection_service_sync_updates_existing_source_enabled_flag() -> None:
 
     assert service.get("demo.personal").enabled is False
     assert catalog.capability_sources["demo.personal"].enabled is False
+    updated = service.events.list_events()[-1]
+    assert updated.kind == "connection_updated"
+    assert updated.connection_id == "demo.personal"
+    assert updated.payload["enabled"] is False
+
+
+def test_connection_service_sync_registers_new_connections_with_event() -> None:
+    service = ConnectionService(events=BrokerEventRecorder(EventBus()))
+    catalog = _source_catalog(service)
+
+    service.sync_connections_from_config(
+        BrokerConfig(
+            store_root=local_temp_root(),
+            connections=[
+                ConnectionConfig(
+                    id="demo.personal",
+                    server="demo",
+                    account="personal",
+                )
+            ],
+        )
+    )
+
+    assert service.get("demo.personal").account == "personal"
+    assert catalog.capability_sources["demo.personal"].enabled is True
+    registered = service.events.list_events()[-1]
+    assert registered.kind == "connection_registered"
+    assert registered.connection_id == "demo.personal"
 
 
 def test_wfmcpservice_exposes_connection_registry_from_connection_service() -> None:

tests/wf_mcp/service/test_content_access.py

@@ -2,6 +2,8 @@ from __future__ import annotations
 
 import asyncio
 
+import pytest
+
 from wf_mcp.broker.service.content_access import ContentAccessService
 from wf_mcp.broker.service.connection_service import ConnectionService
 from wf_mcp.broker.service.events import BrokerEventRecorder
@@ -152,8 +154,6 @@ def test_content_access_renders_local_documentation_prompt() -> None:
 
 
 def test_content_access_raises_on_unknown_resource() -> None:
-    import pytest
-
     content_access, _ = _make_content_access(
         store_root="content_access_missing_resource"
     )

tests/wf_mcp/service/test_upstream_transport.py

@@ -127,11 +127,13 @@ def test_upstream_transport_live_diagnostics_report_missing_connection() -> None
         store=FileStore(local_temp_root() / "upstream_live_missing"),
         event_sink=lambda event: None,
     )
+
+    def _raise_missing_connection(connection_id: str) -> ConnectionConfig:
+        raise KeyError(connection_id)
+
     source_catalog = SourceCatalogService(
         store=transport.store,
-        connection_lookup=lambda connection_id: (_ for _ in ()).throw(
+        connection_lookup=_raise_missing_connection,
-            KeyError(connection_id)
-        ),
         connection_list_enabled=lambda: [],
         connection_list_all=lambda: [],
         tool_executor_for=transport.tool_executor_for,