From f3257d188e42c24876fb30aa9844e6532b16da77 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:13:38 -0400 Subject: [PATCH 1/5] Refuse maintenance before connector store access Src-Budget: 8 --- src/agent_comms/comms.py | 4 ++++ src/agent_comms/connector.py | 5 +++-- src/agent_comms/maintenance_barrier.py | 4 +++- 3 files changed, 10 insertions(+), 3 deletions(-) diff --git a/src/agent_comms/comms.py b/src/agent_comms/comms.py index 3af24a50ba..2195dcb1c7 100644 --- a/src/agent_comms/comms.py +++ b/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 @@ -85,4 +86,7 @@ def client_wire(root: Path | None) -> Iterator[Comms]: """ route = resolve_comms_route(root) with route.hold_client_route(): + # Refuse before opening stores that maintenance may hold locked. This + # observation grants nothing: writes still check under their own locks. + MaintenanceBarrier(route.root / "registry.json").assert_open_unlocked() yield Comms(route, projection=CompletingRouteCounts) diff --git a/src/agent_comms/connector.py b/src/agent_comms/connector.py index 9ff7e81d2a..b765e7b0b1 100644 --- a/src/agent_comms/connector.py +++ b/src/agent_comms/connector.py @@ -48,6 +48,7 @@ CODE_SECONDS, AuthorizationRequest, ConnectorGrants, ConnectorStore, IssuedCode, RegisteredClient, digest, ) +from .maintenance_barrier import MaintenanceHeldError from .messages import Message from .todos import DirectiveText from .tool_caller import ThreadCaller @@ -291,7 +292,7 @@ async def result(self, connector: Connector) -> ToolsCallResult: ) try: reply = await asyncio.to_thread(connector.invoke, tool, self.params.arguments) - except ToolRefusedError as error: + except (ToolRefusedError, MaintenanceHeldError) as error: return ToolsCallResult.refused(error) return ToolsCallResult((TextContent(json.dumps(reply)),), False, reply) @@ -314,7 +315,7 @@ class ToolsCallResult: ) @classmethod - def refused(cls, error: ToolRefusedError) -> ToolsCallResult: + def refused(cls, error: ToolRefusedError | MaintenanceHeldError) -> ToolsCallResult: return cls((TextContent(CoreError.from_exception(error).model_dump_json()),), True) diff --git a/src/agent_comms/maintenance_barrier.py b/src/agent_comms/maintenance_barrier.py index 047deeb48c..36dc441b83 100644 --- a/src/agent_comms/maintenance_barrier.py +++ b/src/agent_comms/maintenance_barrier.py @@ -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. + """Capture admission while holding the registry lock OR the wire lock. + A client may observe this without a lock only for early refusal, never + to grant admission. An incomplete transition fails closed in 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. From 3e9dabd8296ee31631d8819f664d1e23f2864eab Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:14:39 -0400 Subject: [PATCH 2/5] Prove maintenance refusal through real connector on a copied bus --- tests/test_connector.py | 43 ++++++++++ tests/verify_connector_maintenance.py | 115 ++++++++++++++++++++++++++ 2 files changed, 158 insertions(+) create mode 100644 tests/verify_connector_maintenance.py diff --git a/tests/test_connector.py b/tests/test_connector.py index 15c389300a..cc572ab036 100644 --- a/tests/test_connector.py +++ b/tests/test_connector.py @@ -5,6 +5,7 @@ import secrets import shutil import json +import time from pathlib import Path from urllib.parse import parse_qs, urlsplit @@ -21,6 +22,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 +285,44 @@ 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): + 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 / "maintenance-copy" + shutil.copytree(comms.root, copied_root) + 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): + 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 diff --git a/tests/verify_connector_maintenance.py b/tests/verify_connector_maintenance.py new file mode 100644 index 0000000000..56c01e3e81 --- /dev/null +++ b/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 < 1 and float(curl_seconds) < 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() From c270107433004ce5b9937d44996ace90f020ad37 Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:18:12 -0400 Subject: [PATCH 3/5] Refuse held maintenance before route publication locks as well --- src/agent_comms/comms.py | 6 +++--- tests/test_connector.py | 16 +++++++++++++--- 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/src/agent_comms/comms.py b/src/agent_comms/comms.py index 2195dcb1c7..3e7f76d516 100644 --- a/src/agent_comms/comms.py +++ b/src/agent_comms/comms.py @@ -85,8 +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(): - # Refuse before opening stores that maintenance may hold locked. This - # observation grants nothing: writes still check under their own locks. - MaintenanceBarrier(route.root / "registry.json").assert_open_unlocked() yield Comms(route, projection=CompletingRouteCounts) diff --git a/tests/test_connector.py b/tests/test_connector.py index cc572ab036..5c472841f5 100644 --- a/tests/test_connector.py +++ b/tests/test_connector.py @@ -1,17 +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, @@ -286,12 +290,15 @@ async def test_a_call_returns_what_the_tool_returns_in_process(client, store, ro @pytest.mark.parametrize("phases", [(), (PausedPhase,), (PausedPhase, InstallingPhase)]) -async def test_maintenance_refuses_before_locked_stores_on_a_copied_bus(comms, tmp_path, phases): +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 / "maintenance-copy" + 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() @@ -301,7 +308,10 @@ async def test_maintenance_refuses_before_locked_stores_on_a_copied_bus(comms, t 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): + 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", From b9965646fa89f8a76944eddf4772a0c50b44bf5e Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:22:39 -0400 Subject: [PATCH 4/5] Describe maintenance refusal in plain language --- src/agent_comms/maintenance_barrier.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/agent_comms/maintenance_barrier.py b/src/agent_comms/maintenance_barrier.py index 36dc441b83..df2e2bb705 100644 --- a/src/agent_comms/maintenance_barrier.py +++ b/src/agent_comms/maintenance_barrier.py @@ -282,10 +282,10 @@ def _read(path: Path, declaration): return value def current_unlocked(self) -> MaintenanceStatus: - """Capture admission while holding the registry lock OR the wire lock. + """Check the phase under the registry lock OR the wire lock. - A client may observe this without a lock only for early refusal, never - to grant admission. An incomplete transition fails closed in that read. + 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. From 354b3aeec7e3375d27325a23de10b053aa19e80a Mon Sep 17 00:00:00 2001 From: Tristan Simas Date: Mon, 19 Oct 2026 17:30:01 -0400 Subject: [PATCH 5/5] Declare maintenance holds as tool refusals MaintenanceHeldError composes relation and tool refusal behavior; connector consumers catch the one owning family. Remove the explicit exception tuple and union. Tighten real copied-bus proof to 0.1 seconds. --- src/agent_comms/connector.py | 5 ++--- src/agent_comms/maintenance_barrier.py | 4 ++-- tests/verify_connector_maintenance.py | 2 +- 3 files changed, 5 insertions(+), 6 deletions(-) diff --git a/src/agent_comms/connector.py b/src/agent_comms/connector.py index b765e7b0b1..9ff7e81d2a 100644 --- a/src/agent_comms/connector.py +++ b/src/agent_comms/connector.py @@ -48,7 +48,6 @@ CODE_SECONDS, AuthorizationRequest, ConnectorGrants, ConnectorStore, IssuedCode, RegisteredClient, digest, ) -from .maintenance_barrier import MaintenanceHeldError from .messages import Message from .todos import DirectiveText from .tool_caller import ThreadCaller @@ -292,7 +291,7 @@ async def result(self, connector: Connector) -> ToolsCallResult: ) try: reply = await asyncio.to_thread(connector.invoke, tool, self.params.arguments) - except (ToolRefusedError, MaintenanceHeldError) as error: + except ToolRefusedError as error: return ToolsCallResult.refused(error) return ToolsCallResult((TextContent(json.dumps(reply)),), False, reply) @@ -315,7 +314,7 @@ class ToolsCallResult: ) @classmethod - def refused(cls, error: ToolRefusedError | MaintenanceHeldError) -> ToolsCallResult: + def refused(cls, error: ToolRefusedError) -> ToolsCallResult: return cls((TextContent(CoreError.from_exception(error).model_dump_json()),), True) diff --git a/src/agent_comms/maintenance_barrier.py b/src/agent_comms/maintenance_barrier.py index df2e2bb705..3901015c0b 100644 --- a/src/agent_comms/maintenance_barrier.py +++ b/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.""" diff --git a/tests/verify_connector_maintenance.py b/tests/verify_connector_maintenance.py index 56c01e3e81..71a36975a0 100644 --- a/tests/verify_connector_maintenance.py +++ b/tests/verify_connector_maintenance.py @@ -84,7 +84,7 @@ def prove(scratch: Path) -> None: "type": "held", "error": "Maintenance holds ingress: the bus is under maintenance", }, output - assert wall < 1 and float(curl_seconds) < 1, 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({