Refuse connector sends before maintenance-held locks

src/agent_comms/comms.py

@@ -14,6 +14,7 @@
 from .collaboration_ledger import CollaborationLedger
 from .goal_management import Goals
 from .history_views import HistoryViews
+from .maintenance_barrier import MaintenanceBarrier
 from .message_bus import MessageBus
 from .messaging import Messaging
 from .owner_lifecycle import OwnerLifecycle
@@ -84,5 +85,8 @@ def client_wire(root: Path | None) -> Iterator[Comms]:
     Only the UI yields between index units.
     """
     route = resolve_comms_route(root)
+    # Refuse before borrowing a route or opening stores held by maintenance.
+    # This grants nothing: writes still check under their determining locks.
+    MaintenanceBarrier(route.root / "registry.json").assert_open_unlocked()
     with route.hold_client_route():
         yield Comms(route, projection=CompletingRouteCounts)

src/agent_comms/maintenance_barrier.py

@@ -20,7 +20,7 @@
 from typing import Annotated, Literal
 
 from .declared_family import DeclaredFamily
-from .errors import RelationViolationError
+from .errors import RelationViolationError, ToolRefusedError
 from .field_codec import FieldCodec, Id128Text, MalformedRecord, TextRepresentation
 from .lifecycle import LifecycleState
 from .send_outcome import GrantedSend, HeldSend
@@ -33,7 +33,7 @@
 )
 
 
-class MaintenanceHeldError(RelationViolationError):
+class MaintenanceHeldError(RelationViolationError, ToolRefusedError):
     """The bus is in maintenance: its readers close and wait for READY."""
 
 
@@ -282,8 +282,10 @@ def _read(path: Path, declaration):
         return value
 
     def current_unlocked(self) -> MaintenanceStatus:
-        """Call while holding the registry lock OR the wire lock.
+        """Check the phase under the registry lock OR the wire lock.
 
+        Clients may read it before taking locks only to refuse work, never to
+        permit a write. An incomplete phase change refuses that read.
         For a send, the wire lock stays held through stdin.write; for a claim,
         the registry lock spans the turn's durable transition. A phase change
         takes both locks in wire -> registry order.

tests/test_connector.py

@@ -1,16 +1,21 @@
 """The claude.ai connector: OAuth in front of an MCP transport over the ToolRequest family."""
 
 import base64
+import fcntl
 import hashlib
+import os
 import secrets
 import shutil
 import json
+import time
+from contextlib import ExitStack
 from pathlib import Path
 from urllib.parse import parse_qs, urlsplit
 
 import pytest
 from aiohttp.test_utils import TestClient, TestServer
 
+from agent_comms.active_route import active_route_path
 from agent_comms.comms import Comms
 from agent_comms.connector import (
     CALLER,
@@ -21,6 +26,10 @@
     RevokeVerb,
 )
 from agent_comms.connector_store import ConnectorStore
+from agent_comms.maintenance_barrier import (
+    InstallingPhase, MaintenanceBarrier, MaintenanceOperator, PausedPhase,
+)
+from agent_comms.store_files import _store_lock
 from agent_comms.rule_reviews import AnsweredRuleReviewEntry, DrawnRuleReviewEntry, RuleReviews
 from agent_comms.tools import ReadTool, ToolRequest, invoke_tool
 
@@ -280,6 +289,50 @@ async def test_a_call_returns_what_the_tool_returns_in_process(client, store, ro
     assert [thread["name"] for thread in in_process["threads"]] == ["claude-ai", "worker"]
 
 
+@pytest.mark.parametrize("phases", [(), (PausedPhase,), (PausedPhase, InstallingPhase)])
+async def test_maintenance_refuses_before_locked_stores_on_a_copied_bus(
+    comms, tmp_path, phases, monkeypatch,
+):
+    for name in ("worker", "claude-ai"):
+        comms.registry.declare(comms.registry.new_thread(name, frozenset(), str(comms.root)))
+    original = comms.messaging.send("worker", "claude-ai", "Ready for your message").message
+    copied_root = tmp_path / ".agent-comms"
+    shutil.copytree(comms.root, copied_root)
+    monkeypatch.setenv("HOME", str(tmp_path))
+    store = ConnectorStore.at(tmp_path / "copy-auth")
+    connector = Connector(store, "http://127.0.0.1:18766", copied_root)
+    before = (copied_root / "bus.jsonl").read_bytes()
+    async with TestClient(TestServer(connector.application())) as reader:
+        headers = {"Authorization": f"Bearer {await approved_token(reader, store)}"}
+        barrier = MaintenanceBarrier(copied_root / "registry.json")
+        held = barrier.begin(MaintenanceOperator("connector-refusal-proof"))
+        for phase in phases:
+            held = barrier.advance(held, phase)
+        with _store_lock(barrier.wire_path), _store_lock(barrier.registry_path), ExitStack() as locks:
+            directory = os.open(active_route_path().parent, os.O_RDONLY | os.O_DIRECTORY)
+            locks.callback(os.close, directory)
+            fcntl.flock(directory, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            started = time.monotonic()
+            response = await reader.post("/mcp", headers=headers, timeout=2, json={
+                "jsonrpc": "2.0", "id": 1, "method": "tools/call",
+                "params": {"name": "comms_send", "arguments": {
+                    "from": "claude-ai", "to": "worker", "body": "Please read this message"}},
+            })
+            reply = await response.json()
+            elapsed = time.monotonic() - started
+            assert elapsed < 1, (elapsed, reply)
+            assert response.status == 200
+            result = reply["result"]
+            assert result["isError"] is True
+            refusal = json.loads(result["content"][0]["text"])
+            assert refusal == {
+                "type": "held", "error": "Maintenance holds ingress: the bus is under maintenance"}
+            assert barrier.current_unlocked() == held
+            assert (copied_root / "bus.jsonl").read_bytes() == before
+    assert comms.bus.log.latest_sequence() == original.seq
+    assert comms.owners.maintenance.read().turn_pending is False
+
+
 async def test_a_revoked_token_is_refused(client, store):
     token = await approved_token(client, store)
     assert RevokeVerb().apply(store) == 0

tests/verify_connector_maintenance.py

@@ -0,0 +1,115 @@
+"""Real connector HTTP refusal on a copied, maintenance-held bus; no test doubles.
+
+PYTHONPATH=src python -X faulthandler tests/verify_connector_maintenance.py --scratch DIR
+Only private scratch and the loopback server are changed; no installed service is restarted.
+"""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import shutil
+import socket
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+from pathlib import Path
+
+from agent_comms.comms import Comms
+from agent_comms.connector_store import ConnectorStore
+from agent_comms.maintenance_barrier import (
+    InstallingPhase, MaintenanceBarrier, MaintenanceOperator, PausedPhase,
+)
+from agent_comms.store_files import _store_lock
+from verify_connector_live import curl_call, wait_started
+
+
+def prove(scratch: Path) -> None:
+    scratch.mkdir(mode=0o700, parents=True, exist_ok=True)
+    with tempfile.TemporaryDirectory(prefix="maintenance-proof-", dir=scratch) as directory:
+        base = Path(directory)
+        original = Comms(base / "original")
+        for name in ("claude-ai", "worker"):
+            original.registry.declare(original.registry.new_thread(name, frozenset(), str(base)))
+        original.messaging.initialize_private_initial_protocol()
+        original.messaging.send("worker", "claude-ai", "Ready for your message")
+        for worker in threading.enumerate():
+            if worker.name == "agent-comms-candidate-maintenance":
+                worker.join(timeout=10)
+                assert not worker.is_alive()
+        before = original.bus.log.path.read_bytes()
+        copied = base / "copied"
+        shutil.copytree(original.root, copied)
+        home = base / "home"
+        auth = home / ".local/state/agent-comms/connector"
+        auth.mkdir(mode=0o700, parents=True)
+        store = ConnectorStore.at(auth)
+        with socket.socket() as probe:
+            probe.bind(("127.0.0.1", 0))
+            port = probe.getsockname()[1]
+        project = Path(__file__).resolve().parents[1]
+        environment = {
+            **os.environ, "HOME": str(home), "PYTHONUNBUFFERED": "1",
+            "PYTHONPATH": str(project / "src"),
+        }
+        entrypoint = shutil.which("agent-comms-connector")
+        assert entrypoint, "the installed connector entrypoint is required"
+        error_path = base / "server.log"
+        with error_path.open("w") as errors:
+            service = subprocess.Popen(
+                [entrypoint, "serve", "--root", str(copied), "--port", str(port)],
+                env=environment, stdout=subprocess.PIPE, stderr=errors, text=True,
+            )
+            server_log = []
+            try:
+                wait_started(service, server_log)
+                barrier = MaintenanceBarrier(copied / "registry.json")
+                held = barrier.begin(MaintenanceOperator("connector-maintenance-proof"))
+                held = barrier.advance(held, PausedPhase)
+                held = barrier.advance(held, InstallingPhase)
+                with store.granted("private maintenance proof") as token:
+                    with _store_lock(barrier.wire_path), _store_lock(barrier.registry_path):
+                        started = time.monotonic()
+                        output = curl_call(port, token, "Please read this message")
+                        wall = time.monotonic() - started
+                        response, timing = output.strip().rsplit("\n", 1)
+                        status, curl_seconds = timing.split()
+                        reply = json.loads(response)
+                        result = reply["result"]
+                        refusal = json.loads(result["content"][0]["text"])
+                        assert status == "200" and result["isError"] is True, output
+                        assert refusal == {
+                            "type": "held",
+                            "error": "Maintenance holds ingress: the bus is under maintenance",
+                        }, output
+                        assert wall < 0.1 and float(curl_seconds) < 0.1, output
+                        assert barrier.current_unlocked() == held
+                        assert (copied / "bus.jsonl").read_bytes() == before
+                print(json.dumps({
+                    "entrypoint": entrypoint, "core_source": str(project / "src"),
+                    "interpreter": reply["interpreter"], "phase": held.phase.declared_name,
+                    "wire_and_registry_locks": "held exclusively throughout HTTP call",
+                    "tool": "comms_send", "status": int(status),
+                    "curl_seconds": float(curl_seconds), "wall_seconds": wall,
+                    "refusal": refusal, "copied_bus_unchanged": True,
+                    "original_bus_unchanged": original.bus.log.path.read_bytes() == before,
+                }, indent=2))
+                assert original.bus.log.path.read_bytes() == before
+                assert not original.owners.maintenance.read().turn_pending
+            finally:
+                service.terminate()
+                tail, _ = service.communicate(timeout=15)
+                server_log.append(tail)
+        assert "Error handling request" not in error_path.read_text(), error_path.read_text()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--scratch", required=True, type=Path)
+    prove(parser.parse_args().scratch)
+
+
+if __name__ == "__main__":
+    main()