diff --git a/.github/scripts/headless_agent_qualification.py b/.github/scripts/headless_agent_qualification.py new file mode 100644 index 00000000..3a1caf71 --- /dev/null +++ b/.github/scripts/headless_agent_qualification.py @@ -0,0 +1,442 @@ +#!/usr/bin/env python3 +"""Private, bounded MCP grant client for a future independent-agent journey. + +This module is a fixture seam, NOT a two-agent qualification result. A separate +orchestrator must create two clients on distinct machines, perform the pinned +host exchange and refuse a pass unless its live cases and cleanup complete. +Owner-admin calls here issue/revoke grants only; agent data calls use MCP. +Never upload the grant file, process input/output, home or message content. +""" + +import json +import os +from pathlib import Path +import re +import stat +import time + +import headless_mcp_qualification as mcp +import headless_qualification as public + + +# A finite ceiling, not an allowance to exhaust during an unbounded poll. +BUDGET = {"calls": 512, "send_attempts": 8, "body_bytes": 4096, + "read_records": 2048, "read_bytes": 8 * 1024 * 1024} +PERMISSIONS = {name: True for name in ("status", "messages", "send", "outbox_status")} +MAX_GRANT_BYTES = 16384 +MAX_FRAME_BYTES = 65536 +MAX_MESSAGE_PAGES = 16 +MAX_OUTBOX_PAGES = 16 +MAX_U64 = (1 << 64) - 1 + + +def require(condition, reason): + if not condition: + raise ValueError(reason) + + +def no_duplicate_fields(pairs): + """Never let a repeated field change how a scoped MCP reply is interpreted.""" + value = {} + for key, item in pairs: + require(key not in value, "duplicate MCP JSON field") + value[key] = item + return value + + +def fixed_public_scope(status, pin): + """Bind an agent to the selected signed room, never a caller-supplied room.""" + require(isinstance(status, dict) and status.get("kind") == "public" + and status.get("pin") == pin and isinstance(pin, str) + and re.fullmatch(r"[a-f0-9]{64}", pin), "agent room is not the pinned public room") + policy = status.get("policy") + require(isinstance(policy, dict) and isinstance(policy.get("id"), str) + and re.fullmatch(r"[a-f0-9]{64}", policy["id"]) + and type(policy.get("revision")) is int and policy["revision"] >= 0 + and isinstance(status.get("author"), str) + and re.fullmatch(r"[a-f0-9]{64}", status["author"]), + "agent room has no verified author and policy") + return {"kind": "public", "pin": pin, "author": status["author"], + "policy": policy["id"], "revision": policy["revision"]} + + +def check_grant_path(path): + parent = path.parent.lstat() + require(stat.S_ISDIR(parent.st_mode) and not path.parent.is_symlink() + and parent.st_uid == os.geteuid() and parent.st_mode & 0o077 == 0 + and not path.exists() and not path.is_symlink(), "grant path is not new and private") + + +def private_grant(path, grant): + """Create a non-public one-use launch file without following any links.""" + check_grant_path(path) + require(isinstance(grant, dict) and set(grant) == {"generation", "token"} + and all(isinstance(grant[key], str) and re.fullmatch(r"[a-f0-9]{64}", grant[key]) + for key in ("generation", "token")), "invalid grant credentials") + raw = (json.dumps(grant, sort_keys=True, separators=(",", ":")) + "\n").encode() + require(len(raw) <= MAX_GRANT_BYTES, "grant file exceeds bound") + fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) + try: + with os.fdopen(fd, "wb") as stream: + stream.write(raw) + except BaseException: + path.unlink(missing_ok=True) + raise + + +class AgentRefusal(Exception): + """A typed MCP refusal, with no untrusted text or private values.""" + def __init__(self, code): + super().__init__("scoped MCP operation refused") + self.code = code + + +class ScopedAgent: + """One independently issued grant consumed only by MCP on its home VM. + + The daemon object is the existing owned headless public fixture daemon. + `exchange` uses its bounded, no-shell pipe transport and process group. + No owner-admin send/read shortcut is available through this object. + """ + + def __init__(self, daemon, room, pin, grant_path, *, transport=public.exchange, + clock=time.time): + require(isinstance(room, str) and re.fullmatch(r"[a-f0-9]{32}", room) + and room != "0" * 32, "invalid local room") + require(isinstance(grant_path, Path) and grant_path.is_absolute(), + "grant file must be absolute") + require(grant_path.parent == Path(daemon.config["work"]) + and grant_path.parent != daemon.home, "grant must stay in the owned private work") + self.daemon, self.room, self.pin = daemon, room, pin + self.grant_path = grant_path + self.transport, self.clock = transport, clock + self.scope = None + self.grant = None + self.owns_file = False + # One owner grant.issue per role. A lost reply may already have minted a + # grant, and a lost revoke reply may already have removed it. Neither + # ambiguity permits another owner write from this fixture instance. + self.issue_attempted = False + self.revoke_uncertain = False + + def revoke(self): + require(self.grant is not None and not self.revoke_uncertain, + "grant revocation unconfirmed; do not retry") + self.revoke_uncertain = True + revoked = self.daemon.call("grant.revoke", **self.grant) + require(isinstance(revoked, dict) and revoked.get("revoked") is True, + "grant revocation unconfirmed") + self.grant = None + self.revoke_uncertain = False + + def issue(self, operation): + require(not self.issue_attempted and self.grant is None and not self.owns_file, + "agent grant already attempted or issued") + require(isinstance(operation, str) and re.fullmatch(r"[a-f0-9]{32}", operation) + and operation != "0" * 32, "invalid grant operation") + check_grant_path(self.grant_path) # Refuse an existing file before minting anything. + status = self.daemon.call("room.status", room=self.room) + require(isinstance(status, dict) and status.get("room") == self.room, + "room status differs from selected room") + self.scope = fixed_public_scope(status, self.pin) + now = int(self.clock()) + self.issue_attempted = True + result = self.daemon.call("grant.issue", operation=operation, room=self.room, + grant={"scope": self.scope, "permissions": PERMISSIONS, "budget": BUDGET, + "not_before": now - 1, "expires_at": now + 600}) + # A successful but unexpected owner reply may already have minted a live + # grant. Capture its credentials before checking display metadata so it + # can be revoked; without credentials the outcome cannot be reconciled. + credentials = {name: result.get(name) for name in ("generation", "token")} if isinstance(result, dict) else {} + require(all(isinstance(credentials.get(name), str) + and re.fullmatch(r"[a-f0-9]{64}", credentials[name]) + for name in ("generation", "token")), + "issued grant credentials unavailable; revocation unconfirmed") + self.grant = credentials + try: + require(result.get("room") == self.room and result.get("scope") == self.scope + and result.get("permissions") == PERMISSIONS + and result.get("expires_at") == now + 600, + "issued grant differs from selected scope") + # Strip owner-only display metadata before writing the private launcher. + private_grant(self.grant_path, credentials) + self.owns_file = True + except BaseException: + # No retry of issuance. If revocation is uncertain, surface it. + self.revoke() + raise + return {"scope": self.scope.copy(), "permissions": PERMISSIONS.copy(), + "budget": BUDGET.copy(), "expires_in_seconds": 600} + + def call(self, name, **arguments): + require(self.grant is not None and self.owns_file, "MCP grant has not been issued") + require(name in mcp.TOOLS, "unknown scoped MCP operation") + initialize = {"jsonrpc": "2.0", "id": "initialize", "method": "initialize", "params": { + "protocolVersion": mcp.INITIALIZE_PROTOCOL, "capabilities": {}, + "clientInfo": {"name": "valhalla-agent-qualification", "version": "1"}}} + notification = {"jsonrpc": "2.0", "method": "notifications/initialized"} + request = mcp.tool(1, name, **arguments) + payload = b"".join(json.dumps(item, separators=(",", ":")).encode() + b"\n" + for item in (initialize, notification, request)) + require(len(payload) <= MAX_FRAME_BYTES, "MCP input exceeds bound") + argv = self.daemon.argv + ["mcp", "--home", str(self.daemon.home), + "--grant", str(self.grant_path)] + code, raw = self.transport(argv, payload, self.daemon.env, timeout=35) + require(code == 0 and len(raw) <= public.MAX_REPLY and raw.endswith(b"\n"), + "MCP process failed or output is incomplete") + require(all(self.grant[key].encode() not in raw for key in ("generation", "token")), + "MCP output exposed grant credentials") + require(len(raw.splitlines()) == 2, "MCP protocol output differs") + first, second = [json.loads(line, object_pairs_hook=no_duplicate_fields) + for line in raw.splitlines()] + require(first.get("jsonrpc") == "2.0" and first.get("id") == "initialize" + and first.get("result", {}).get("protocolVersion") == mcp.INITIALIZE_PROTOCOL + and first["result"].get("serverInfo", {}).get("name") == "valhalla", + "MCP handshake differs") + require(second.get("jsonrpc") == "2.0" and second.get("id") == 1, + "MCP reply ID differs") + if "error" in second: + error = second["error"] + require(isinstance(error, dict) and type(error.get("code")) is int, + "untyped MCP protocol refusal") + raise AgentRefusal(error["code"]) + result = second.get("result") + require(isinstance(result, dict) and type(result.get("isError")) is bool + and isinstance(result.get("structuredContent"), dict), + "MCP result is not structured") + data = result["structuredContent"] + if result["isError"]: + require(data.get("status") == "refused" and isinstance(data.get("code"), str), + "untyped daemon refusal") + raise AgentRefusal(data["code"]) + require(data.get("room") == self.room, "agent response escaped selected room") + return data + + def status(self): + data = self.call("agent.status") + require(data.get("kind") == "public" and data.get("author") == self.scope["author"] + and data.get("policy") == self.scope["policy"] + and type(data.get("revision")) is int + and data["revision"] == self.scope["revision"], + "agent status changed pinned public scope") + remaining = data.get("remaining") + require(data.get("freshness") == "local authenticated state" + and isinstance(remaining, dict) and set(remaining) == set(BUDGET) + and all(type(remaining[name]) is int and 0 <= remaining[name] <= maximum + for name, maximum in BUDGET.items()), + "agent status lacks finite authenticated allowance") + return data + + def send(self, operation, text): + result = self.call("agent.send", operation=operation, text=text) + require(result.get("operation") == operation and type(result.get("exact_retry")) is bool + and result.get("delivery") == "not asserted" + and isinstance(result.get("frame_hash"), str) + and re.fullmatch(r"[a-f0-9]{64}", result["frame_hash"]), + "agent send has unsupported delivery or retry claim") + return result + + def messages(self, after=0, limit=4): + require(type(after) is int and after >= 0 and type(limit) is int and 1 <= limit <= 16, + "invalid bounded page") + result = self.call("agent.messages", after=after, limit=limit) + records = result.get("records") + require(isinstance(records, list) and len(records) <= limit + and all(isinstance(row, dict) and row.get("trust") == "inert untrusted room content" + for row in records), "message page is not inert bounded data") + return records + + def peer_event(self, peer, text): + """Find one locally admitted signed peer event, not delivery or execution. + + The caller must independently pin the peer author and use a unique + synthetic text. Scan a complete stable local MCP inbox, not an owner + read or a single page; never return the message body in the summary. + """ + require(self.scope is not None and isinstance(peer, str) + and re.fullmatch(r"[a-f0-9]{64}", peer) + and peer != self.scope["author"] and isinstance(text, str) + and 0 < len(text.encode("utf-8")) <= BUDGET["body_bytes"], + "invalid pinned peer event intent") + after, head, match = 0, None, None + for _ in range(MAX_MESSAGE_PAGES): + page = self.call("agent.messages", after=after, limit=16) + rows = page.get("records") + require(page.get("kind") == "public" + and type(page.get("head")) is int and after <= page["head"] <= MAX_U64 + and "next" in page + and (page["next"] is None or (type(page["next"]) is int + and after < page["next"] < page["head"])) + and isinstance(rows, list) and len(rows) <= 16, + "peer message page is incomplete or unsupported") + if head is None: + head = page["head"] + require(page["head"] == head, "local inbox changed during peer check") + previous = after + for row in rows: + require(isinstance(row, dict) and type(row.get("cursor")) is int + and previous < row["cursor"] <= head + and type(row.get("sequence")) is int and 0 <= row["sequence"] <= MAX_U64 + and all(isinstance(row.get(key), str) + and re.fullmatch(r"[a-f0-9]{64}", row[key]) + for key in ("event", "author", "policy")) + and row.get("visibility") in ("Provisional", "OwnerSealed", + "ContinuityOnly", "Incomplete") + and row.get("trust") == "inert untrusted room content" + and isinstance(row.get("text"), str) + and len(row["text"].encode("utf-8")) <= BUDGET["body_bytes"], + "peer message row lacks bounded signed metadata") + previous = row["cursor"] + if row["text"] == text: + require(row["author"] == peer and row["policy"] == self.scope["policy"] + and row["visibility"] in ("Provisional", "OwnerSealed"), + "peer message is not admitted under the pinned author and policy") + require(match is None, "peer message was duplicated") + match = {"event": row["event"], "author": peer, + "sequence": row["sequence"], "visibility": row["visibility"], + "local_signed_event_once": True, "delivery": "not asserted"} + require(page["next"] is None or previous <= page["next"], + "peer message cursor regressed") + if page["next"] is None: + require(match is not None, "pinned peer event not found in local inbox") + return match + after = page["next"] + raise ValueError("local inbox exceeded peer scan bound") + + def outbox_status(self, after=0, limit=16): + """Local completion metadata only; neither remote delivery nor execution.""" + require(type(after) is int and 0 <= after <= MAX_U64 + and type(limit) is int and 1 <= limit <= 16, + "invalid bounded outbox page") + page = self.call("agent.outbox_status", after=after, limit=limit) + rows = page.get("operations") + require(page.get("kind") == "public" and page.get("delivery") == "not asserted" + and type(page.get("head")) is int and after <= page["head"] <= MAX_U64 + and "next" in page + and (page["next"] is None or (type(page["next"]) is int + and after < page["next"] < page["head"])) + and isinstance(rows, list) and len(rows) <= limit, + "local outbox page is incomplete or unsupported") + previous = after + for row in rows: + require(isinstance(row, dict) and type(row.get("cursor")) is int + and previous < row["cursor"] <= page["head"] + and isinstance(row.get("operation"), str) + and re.fullmatch(r"[a-f0-9]{32}", row["operation"]) + and row.get("kind") in ("Event", "Policy") + and isinstance(row.get("frame_hash"), str) + and re.fullmatch(r"[a-f0-9]{64}", row["frame_hash"]), + "local outbox row is not bounded completion metadata") + previous = row["cursor"] + require(page["next"] is None or previous <= page["next"], + "local outbox cursor regressed") + return page + + def exact_local_retry(self, operation, text): + """Check two successful MCP sends against one complete local outbox scan. + + Never retry after an ambiguous send. This proves neither the remote + peer's delivery nor a human read; the two-machine runner must check both. + """ + require(isinstance(operation, str) and re.fullmatch(r"[a-f0-9]{32}", operation) + and operation != "0" * 32 and isinstance(text, str) + and 0 < len(text.encode("utf-8")) * 2 <= BUDGET["body_bytes"], + "invalid bounded retry intent") + first = self.send(operation, text) + require(first.get("state") in ("Provisional", "OwnerSealed"), + "first send is not locally admitted") + require(first["exact_retry"] is False, "first send was already an exact retry") + second = self.send(operation, text) + require(second.get("state") in ("Provisional", "OwnerSealed"), + "exact retry is not locally admitted") + require(second["exact_retry"] is True and second["frame_hash"] == first["frame_hash"], + "exact retry changed signed frame") + after, head, matches = 0, None, [] + for _ in range(MAX_OUTBOX_PAGES): + page = self.outbox_status(after) + if head is None: + head = page["head"] + require(page["head"] == head, "local outbox changed during retry check") + matches.extend(row for row in page["operations"] if row["operation"] == operation) + if page["next"] is None: + require(len(matches) == 1 and matches[0]["kind"] == "Event" + and matches[0]["frame_hash"] == first["frame_hash"], + "retry has no unique local signed completion") + return {"frame_hash": first["frame_hash"], "local_completion_once": True, + "delivery": "not asserted"} + after = page["next"] + raise ValueError("local outbox exceeded retry scan bound") + + def close(self): + """Revoke and remove our private file; do not hide an uncertain revoke.""" + try: + if self.grant is not None: + self.revoke() + else: + require(not self.revoke_uncertain, + "grant revocation unconfirmed; do not retry") + finally: + if self.owns_file: + self.grant_path.unlink(missing_ok=True) + self.owns_file = False + + +def issue_agent_pair(first, second, operations, machine_ids): + """Mint two disjoint scoped grants after checking one pinned public room. + + Machine IDs are caller declarations, NOT evidence of independent hardware. + This is only an issuance fixture: no MCP call, network peer, delivery or + physical placement is established. The runner must independently verify + placement, exchange, restart, refusals, and cleanup before publishing a pass. + """ + require(isinstance(first, ScopedAgent) and isinstance(second, ScopedAgent) + and first is not second and first.daemon is not second.daemon + and first.daemon.home.resolve() != second.daemon.home.resolve() + and first.grant_path.parent.resolve() != second.grant_path.parent.resolve() + and all(not agent.issue_attempted and agent.grant is None and not agent.owns_file + for agent in (first, second)), + "paired agents must have separate homes and private work roots") + require(isinstance(machine_ids, (tuple, list)) and len(machine_ids) == 2 + and all(isinstance(value, str) and re.fullmatch(r"[a-f0-9]{64}", value) + and value != "0" * 64 for value in machine_ids) + and machine_ids[0] != machine_ids[1], + "paired machine identifiers must be distinct declarations") + require(isinstance(operations, (tuple, list)) and len(operations) == 2 + and all(isinstance(value, str) and re.fullmatch(r"[a-f0-9]{32}", value) + and value != "0" * 32 for value in operations) + and operations[0] != operations[1], "paired grant operations must be distinct") + scopes = [] + for agent in (first, second): + check_grant_path(agent.grant_path) + status = agent.daemon.call("room.status", room=agent.room) + require(isinstance(status, dict) and status.get("room") == agent.room, + "paired room status differs from selected room") + scopes.append(fixed_public_scope(status, agent.pin)) + require(all(scopes[0][key] == scopes[1][key] for key in ("pin", "policy", "revision")) + and scopes[0]["author"] != scopes[1]["author"], + "paired rooms lack a shared pin and policy with distinct authors") + issued_agents, issued = [], [] + try: + for agent, operation, scope in zip((first, second), operations, scopes): + summary = agent.issue(operation) # Re-check each room at issuance. + issued_agents.append(agent) + require(summary["scope"] == scope, "paired room changed during issuance") + issued.append(summary) + require(first.grant != second.grant, "paired grants reused credentials") + except BaseException: + # A failing issue already tried its own revocation. Never repeat an + # uncertain revoke; clean up only earlier, successfully issued grants. + failures = [] + for agent in reversed(issued_agents): + if agent.grant is not None: + try: + agent.close() + except BaseException as failure: + failures.append(failure) + if failures: + raise ValueError("paired grant cleanup unconfirmed") from failures[0] + raise + return {"agents": [{"declared_machine_id": machine, **summary} + for machine, summary in zip(machine_ids, issued)], + "machine_identity": "caller declared; placement unverified", + "delivery": "not asserted"} diff --git a/.github/scripts/test_headless_agent_qualification.py b/.github/scripts/test_headless_agent_qualification.py new file mode 100644 index 00000000..95789241 --- /dev/null +++ b/.github/scripts/test_headless_agent_qualification.py @@ -0,0 +1,743 @@ +"""Deterministic contract tests for the scoped MCP fixture seam; no network pass.""" + +import json +import os +from pathlib import Path +import shutil +import stat +import tempfile +from types import SimpleNamespace +import unittest + +import headless_qualification as public +from headless_agent_qualification import (AgentRefusal, BUDGET, MAX_OUTBOX_PAGES, PERMISSIONS, + ScopedAgent, fixed_public_scope, issue_agent_pair, private_grant) +from headless_managed_qualification import digest +from headless_mcp_qualification import INITIALIZE_PROTOCOL, McpRun + +ROOM, PIN, AUTHOR, POLICY = "1" * 32, "2" * 64, "3" * 64, "4" * 64 +GENERATION, TOKEN, HASH = "5" * 64, "6" * 64, "7" * 64 +PEER, EVENT = "a" * 64, "b" * 64 + + +def frame(number, result=None, error=None): + value = {"jsonrpc": "2.0", "id": number} + value["error" if error is not None else "result"] = error if error is not None else result + return value + + +def reply(data, *, refused=False, error=None): + initialized = frame("initialize", {"protocolVersion": INITIALIZE_PROTOCOL, + "serverInfo": {"name": "valhalla"}}) + called = frame(1, {"isError": refused, "structuredContent": data}, error=error) + return b"".join(json.dumps(value).encode() + b"\n" for value in (initialized, called)) + + +def completion(operation, cursor, frame_hash=HASH): + return {"cursor": cursor, "operation": operation, "kind": "Event", "frame_hash": frame_hash} + + +def outbox(rows, head=40, next_cursor=None): + return reply({"room": ROOM, "kind": "public", "head": head, "next": next_cursor, + "operations": rows, "delivery": "not asserted"}) + + +def message(cursor=31, text="synthetic", author=PEER, visibility="Provisional"): + return {"cursor": cursor, "event": EVENT, "author": author, "sequence": 1, + "policy": POLICY, "text": text, "visibility": visibility, + "trust": "inert untrusted room content"} + + +def inbox(rows, head=40, next_cursor=None): + return reply({"room": ROOM, "kind": "public", "head": head, "next": next_cursor, + "records": rows}) + + +class FakeDaemon: + def __init__(self, work, *, room=ROOM, author=AUTHOR, + generation=GENERATION, token=TOKEN): + self.config = {"work": str(work)} + self.home = work / "home" + self.argv = ["/immutable/fixture", "--no-update", "daemon"] + self.env = {"PATH": "/usr/bin:/bin"} + self.status = {"room": room, "kind": "public", "pin": PIN, "author": author, + "policy": {"id": POLICY, "revision": 1}} + self.generation, self.token = generation, token + self.calls = [] + self.revoke = True + self.issued_override = None + + def call(self, op, **fields): + self.calls.append((op, fields)) + if op == "room.status": + return self.status + if op == "grant.issue": + if self.issued_override is not None: + return self.issued_override + return {"generation": self.generation, "token": self.token, + "room": self.status["room"], + "scope": fields["grant"]["scope"], "permissions": PERMISSIONS, + "expires_at": fields["grant"]["expires_at"], "allowance": "retained"} + if op == "grant.revoke": + return {"revoked": self.revoke} + raise AssertionError("agent data was routed to owner-admin: " + op) + + +class ScopedAgentTests(unittest.TestCase): + def setUp(self): + temp = tempfile.TemporaryDirectory() + self.addCleanup(temp.cleanup) + self.work = Path(temp.name) / "role" + self.work.mkdir(mode=0o700) + self.daemon = FakeDaemon(self.work) + self.responses = [] + self.launches = [] + self.agent = ScopedAgent(self.daemon, ROOM, PIN, self.work / "grant.json", + transport=self.transport, clock=lambda: 2_000_000_000) + + def transport(self, argv, payload, env, timeout): + self.launches.append((argv, payload, env, timeout)) + return 0, self.responses.pop(0) + + def issue(self): + return self.agent.issue("8" * 32) + + def test_issue_is_fixed_room_finite_and_file_only_contains_credentials(self): + summary = self.issue() + self.assertEqual(summary["scope"], {"kind": "public", "pin": PIN, "author": AUTHOR, + "policy": POLICY, "revision": 1}) + self.assertEqual(summary["budget"], BUDGET) + self.assertNotIn("token", json.dumps(summary)) + self.assertEqual(json.loads(self.agent.grant_path.read_text()), + {"generation": GENERATION, "token": TOKEN}) + self.assertEqual(stat.S_IMODE(self.agent.grant_path.stat().st_mode), 0o600) + self.assertEqual(self.daemon.calls[1][1]["grant"]["expires_at"], 2_000_000_600) + with self.assertRaisesRegex(ValueError, "already attempted or issued"): + self.issue() + self.agent.close() + self.assertFalse(self.agent.grant_path.exists()) + self.assertEqual([name for name, _ in self.daemon.calls], + ["room.status", "grant.issue", "grant.revoke"]) + with self.assertRaisesRegex(ValueError, "already attempted or issued"): + self.agent.issue("9" * 32) + self.assertEqual(len(self.daemon.calls), 3) + + def test_confirmed_direct_revoke_still_removes_private_file_on_close(self): + self.issue() + self.agent.revoke() + self.assertIsNone(self.agent.grant) + self.assertTrue(self.agent.grant_path.is_file()) + self.agent.close() + self.assertFalse(self.agent.grant_path.exists()) + self.assertFalse(self.agent.owns_file) + self.agent.close() # Idempotent local cleanup, not another owner write. + self.assertEqual([name for name, _ in self.daemon.calls], + ["room.status", "grant.issue", "grant.revoke"]) + + def test_foreign_pin_policy_and_nonpublic_room_never_mint_grant(self): + for field, value in (("pin", "a" * 64), ("kind", "private"), + ("policy", {"id": POLICY, "revision": True})): + status = self.daemon.status.copy() + status[field] = value + with self.assertRaises(ValueError): + fixed_public_scope(status, PIN) + self.daemon.status["pin"] = "a" * 64 + with self.assertRaises(ValueError): + self.issue() + self.assertEqual([name for name, _ in self.daemon.calls], ["room.status"]) + + def test_wrong_room_status_refuses_before_grant_mint(self): + self.daemon.status["room"] = "a" * 32 + with self.assertRaisesRegex(ValueError, "differs from selected room"): + self.issue() + self.assertEqual([name for name, _ in self.daemon.calls], ["room.status"]) + self.assertFalse(self.agent.grant_path.exists()) + self.assertEqual(self.launches, []) + + def test_mismatched_issued_grant_is_revoked_before_launch(self): + self.daemon.issued_override = {"generation": GENERATION, "token": TOKEN, + "room": ROOM, "scope": {"kind": "private"}, + "permissions": PERMISSIONS, "expires_at": 2_000_000_600} + with self.assertRaisesRegex(ValueError, "differs from selected scope"): + self.issue() + self.assertEqual([name for name, _ in self.daemon.calls], + ["room.status", "grant.issue", "grant.revoke"]) + self.assertEqual(self.daemon.calls[-1][1], + {"generation": GENERATION, "token": TOKEN}) + self.assertIsNone(self.agent.grant) + self.assertFalse(self.agent.grant_path.exists()) + self.assertEqual(self.launches, []) + with self.assertRaisesRegex(ValueError, "already attempted or issued"): + self.agent.issue("9" * 32) + self.assertEqual(len(self.daemon.calls), 3) + + def test_missing_issued_credentials_leave_explicit_uncertainty(self): + self.daemon.issued_override = {"room": ROOM, "scope": {"kind": "private"}} + with self.assertRaisesRegex(ValueError, "revocation unconfirmed"): + self.issue() + self.assertEqual([name for name, _ in self.daemon.calls], ["room.status", "grant.issue"]) + self.assertFalse(self.agent.grant_path.exists()) + self.assertEqual(self.launches, []) + with self.assertRaisesRegex(ValueError, "already attempted or issued"): + self.agent.issue("9" * 32) + self.assertEqual([name for name, _ in self.daemon.calls], ["room.status", "grant.issue"]) + + def test_owner_issue_lost_reply_is_never_retried(self): + def lost(op, **fields): + if op == "grant.issue": + self.daemon.calls.append((op, fields)) + raise TimeoutError("owner reply lost") + return FakeDaemon.call(self.daemon, op, **fields) + self.daemon.call = lost + with self.assertRaises(TimeoutError): + self.issue() + with self.assertRaisesRegex(ValueError, "already attempted or issued"): + self.agent.issue("9" * 32) + self.assertEqual([name for name, _ in self.daemon.calls], ["room.status", "grant.issue"]) + self.assertFalse(self.agent.grant_path.exists()) + + def test_existing_grant_file_and_link_are_refused(self): + path = self.agent.grant_path + path.write_text("existing") + with self.assertRaises(ValueError): + private_grant(path, {"generation": GENERATION, "token": TOKEN}) + path.unlink() + path.symlink_to(self.work / "other") + with self.assertRaises(ValueError): + private_grant(path, {"generation": GENERATION, "token": TOKEN}) + self.assertTrue(path.is_symlink()) + + def test_mcp_send_retry_and_read_are_scoped_and_not_delivery_claims(self): + self.issue() + sent = {"room": ROOM, "operation": "9" * 32, "frame_hash": HASH, + "exact_retry": False, "delivery": "not asserted"} + self.responses = [reply(sent), reply({**sent, "exact_retry": True}), + reply({"room": ROOM, "records": [{"text": "inert", + "trust": "inert untrusted room content"}]})] + self.assertFalse(self.agent.send("9" * 32, "inert")["exact_retry"]) + self.assertTrue(self.agent.send("9" * 32, "inert")["exact_retry"]) + self.assertEqual(self.agent.messages()[0]["text"], "inert") + self.assertEqual(len(self.launches), 3) + for argv, payload, _env, timeout in self.launches: + self.assertEqual(argv[-2:], ["--grant", str(self.agent.grant_path)]) + self.assertEqual(timeout, 35) + frames = [json.loads(line) for line in payload.splitlines()] + self.assertEqual([row["method"] for row in frames], + ["initialize", "notifications/initialized", "tools/call"]) + self.assertNotIn(TOKEN.encode(), payload) + self.assertEqual([name for name, _ in self.daemon.calls], ["room.status", "grant.issue"]) + + def test_status_requires_pinned_policy_and_finite_authenticated_allowance(self): + self.issue() + status = {"room": ROOM, "kind": "public", "author": AUTHOR, "policy": POLICY, + "revision": 1, "remaining": {**BUDGET, "calls": BUDGET["calls"] - 1}, + "freshness": "local authenticated state"} + self.responses = [reply(status)] + self.assertEqual(self.agent.status()["remaining"]["calls"], BUDGET["calls"] - 1) + cases = [({"policy": "a" * 64}, "pinned public scope"), + ({"revision": 2}, "pinned public scope"), + ({"revision": True}, "pinned public scope"), + ({"freshness": "unverified"}, "finite authenticated allowance"), + ({"remaining": {**BUDGET, "calls": BUDGET["calls"] + 1}}, + "finite authenticated allowance"), + ({"remaining": {**BUDGET, "send_attempts": True}}, + "finite authenticated allowance"), + ({"remaining": {**BUDGET, "read_bytes": -1}}, + "finite authenticated allowance"), + ({"remaining": {"calls": 1}}, "finite authenticated allowance")] + for changed, reason in cases: + with self.subTest(changed=changed): + self.responses = [reply({**status, **changed})] + with self.assertRaisesRegex(ValueError, reason): + self.agent.status() + self.assertEqual(len(self.launches), len(cases) + 1) + self.assertEqual([name for name, _ in self.daemon.calls], ["room.status", "grant.issue"]) + + def test_duplicate_scoped_mcp_json_field_refuses_before_status_admission(self): + self.issue() + status = {"room": ROOM, "kind": "public", "author": AUTHOR, "policy": POLICY, + "revision": 1, "remaining": BUDGET, + "freshness": "local authenticated state"} + good = reply(status) + key = f'"room": "{ROOM}"'.encode() + self.assertEqual(good.count(key), 1) + self.responses = [good.replace(key, key + b', "room": "' + b'a' * 32 + b'"')] + with self.assertRaisesRegex(ValueError, "duplicate MCP JSON field"): + self.agent.status() + self.assertEqual(len(self.launches), 1) + self.assertEqual([name for name, _ in self.daemon.calls], ["room.status", "grant.issue"]) + + def test_peer_event_scans_stable_pages_and_returns_body_free_local_evidence(self): + self.issue() + self.responses = [inbox([message(12, "unrelated", AUTHOR)], next_cursor=20), + inbox([message()])] + summary = self.agent.peer_event(PEER, "synthetic") + self.assertEqual(summary, {"event": EVENT, "author": PEER, "sequence": 1, + "visibility": "Provisional", "local_signed_event_once": True, + "delivery": "not asserted"}) + self.assertNotIn("synthetic", json.dumps(summary)) + self.assertNotIn(TOKEN, json.dumps(summary)) + self.assertEqual([json.loads(launch[1].splitlines()[-1])["params"]["arguments"] + for launch in self.launches], + [{"after": 0, "limit": 16}, {"after": 20, "limit": 16}]) + self.assertEqual([name for name, _ in self.daemon.calls], ["room.status", "grant.issue"]) + + def test_peer_event_refuses_foreign_author_policy_unadmitted_and_duplicate(self): + self.issue() + cases = [inbox([message(author=AUTHOR)]), + inbox([message() | {"policy": "c" * 64}]), + inbox([message(visibility="Incomplete")]), + inbox([message(visibility="ContinuityOnly")]), + inbox([message(12), message(31)])] + for page in cases: + with self.subTest(page=page): + self.responses = [page] + with self.assertRaises(ValueError): + self.agent.peer_event(PEER, "synthetic") + self.assertEqual(len(self.launches), len(cases)) + + def test_peer_event_refuses_moving_or_malformed_page_and_absent_peer(self): + self.issue() + cases = ([inbox([], next_cursor=20), inbox([message()], head=41)], + [reply({"room": ROOM, "kind": "public", "head": 40, "records": [message()]})], + [inbox([message() | {"cursor": True}])], + [inbox([message() | {"trust": "verified by prompt"}])], + [inbox([message() | {"event": "not-an-event"}])], + [inbox([message()], next_cursor=30)], + [inbox([])]) + for pages in cases: + with self.subTest(pages=pages): + self.responses = list(pages) + with self.assertRaises(ValueError): + self.agent.peer_event(PEER, "synthetic") + + def test_peer_event_refuses_non_u64_metadata(self): + self.issue() + # The daemon's head, cursor and author sequence are unsigned 64-bit values. + # Python's JSON decoder accepts wider integers; they must not become evidence. + cases = ([inbox([message()], head=1 << 64)], + [inbox([message() | {"sequence": 1 << 64}])]) + for pages in cases: + with self.subTest(pages=pages): + self.responses = list(pages) + with self.assertRaises(ValueError): + self.agent.peer_event(PEER, "synthetic") + + def test_peer_event_refuses_invalid_intent_and_bounded_incomplete_scan(self): + self.issue() + for peer, text in ((AUTHOR, "synthetic"), ("not-a-peer", "synthetic"), + (PEER, ""), (PEER, "a" * 4097)): + with self.subTest(peer=peer, size=len(text)): + with self.assertRaisesRegex(ValueError, "invalid pinned peer event intent"): + self.agent.peer_event(peer, text) + self.assertEqual(self.launches, []) + self.responses = [*(inbox([], next_cursor=after + 1) + for after in range(16))] + with self.assertRaisesRegex(ValueError, "exceeded peer scan bound"): + self.agent.peer_event(PEER, "synthetic") + self.assertEqual(len(self.launches), 16) + + def test_exact_local_retry_scans_completed_outbox_without_delivery_claim(self): + self.issue() + operation, text = "9" * 32, "private message body" + sent = {"room": ROOM, "operation": operation, "frame_hash": HASH, + "state": "Provisional", "exact_retry": False, "delivery": "not asserted"} + self.responses = [reply(sent), reply({**sent, "exact_retry": True}), + outbox([completion("a" * 32, 12)], next_cursor=20), + outbox([completion(operation, 31)])] + summary = self.agent.exact_local_retry(operation, text) + self.assertEqual(summary, {"frame_hash": HASH, "local_completion_once": True, + "delivery": "not asserted"}) + self.assertNotIn(text, json.dumps(summary)) + self.assertNotIn(TOKEN, json.dumps(summary)) + self.assertEqual(len(self.launches), 4) + self.assertEqual([json.loads(launch[1].splitlines()[-1])["params"]["name"] + for launch in self.launches], + ["agent.send", "agent.send", "agent.outbox_status", "agent.outbox_status"]) + self.assertEqual(json.loads(self.launches[-1][1].splitlines()[-1])["params"]["arguments"], + {"after": 20, "limit": 16}) + self.assertEqual([name for name, _ in self.daemon.calls], ["room.status", "grant.issue"]) + + def test_outbox_refuses_metadata_beyond_daemon_u64_range(self): + self.issue() + with self.assertRaisesRegex(ValueError, "invalid bounded outbox page"): + self.agent.outbox_status(after=1 << 64) + self.assertEqual(self.launches, []) + self.responses = [outbox([completion("9" * 32, 31)], head=1 << 64)] + with self.assertRaisesRegex(ValueError, "local outbox page is incomplete"): + self.agent.outbox_status() + self.assertEqual(len(self.launches), 1) + + def test_retry_refuses_bad_intent_before_any_mcp_launch(self): + self.issue() + for operation, text in (("0" * 32, "message"), ("9" * 32, "a" * 2049), + ("9" * 32, ""), ("not-an-operation", "message")): + with self.subTest(operation=operation, size=len(text)): + with self.assertRaisesRegex(ValueError, "bounded retry intent"): + self.agent.exact_local_retry(operation, text) + self.assertEqual(self.launches, []) + + def test_retry_refuses_replayed_first_send_or_changed_signed_frame(self): + self.issue() + operation = "9" * 32 + sent = {"room": ROOM, "operation": operation, "frame_hash": HASH, + "state": "Provisional", "exact_retry": False, "delivery": "not asserted"} + self.responses = [reply({**sent, "exact_retry": True})] + with self.assertRaisesRegex(ValueError, "already an exact retry"): + self.agent.exact_local_retry(operation, "message") + self.responses = [reply(sent), reply({**sent, "exact_retry": True, "frame_hash": "b" * 64})] + with self.assertRaisesRegex(ValueError, "changed signed frame"): + self.agent.exact_local_retry(operation, "message") + self.assertEqual(len(self.launches), 3) + + def test_retry_refuses_unadmitted_signed_frame_without_overclaiming_completion(self): + self.issue() + operation = "9" * 32 + sent = {"room": ROOM, "operation": operation, "frame_hash": HASH, + "state": "Provisional", "exact_retry": False, "delivery": "not asserted"} + self.responses = [reply({**sent, "state": "PendingHistory"})] + with self.assertRaisesRegex(ValueError, "first send is not locally admitted"): + self.agent.exact_local_retry(operation, "message") + self.assertEqual(len(self.launches), 1) # Never retry after an unadmitted result. + self.responses = [reply(sent), reply({**sent, "state": "NeedsRepost", + "exact_retry": True})] + with self.assertRaisesRegex(ValueError, "exact retry is not locally admitted"): + self.agent.exact_local_retry(operation, "message") + self.assertEqual(len(self.launches), 3) + + def test_retry_refuses_duplicate_completion_and_moving_outbox(self): + self.issue() + operation = "9" * 32 + sent = {"room": ROOM, "operation": operation, "frame_hash": HASH, + "state": "Provisional", "exact_retry": False, "delivery": "not asserted"} + self.responses = [reply(sent), reply({**sent, "exact_retry": True}), + outbox([completion(operation, 12)], next_cursor=20), + outbox([completion(operation, 31)])] + with self.assertRaisesRegex(ValueError, "no unique local signed completion"): + self.agent.exact_local_retry(operation, "message") + self.responses = [reply(sent), reply({**sent, "exact_retry": True}), + outbox([], next_cursor=20), outbox([completion(operation, 31)], head=41)] + with self.assertRaisesRegex(ValueError, "changed during retry check"): + self.agent.exact_local_retry(operation, "message") + + def test_retry_refuses_incomplete_or_forged_outbox(self): + self.issue() + operation = "9" * 32 + sent = {"room": ROOM, "operation": operation, "frame_hash": HASH, + "state": "Provisional", "exact_retry": False, "delivery": "not asserted"} + cases = [outbox([completion(operation, 31, "b" * 64)]), + outbox([completion(operation, 31)], next_cursor=0), + outbox([completion(operation, 31)], next_cursor=30), + outbox([completion(operation, 31, HASH) | {"kind": "Policy"}]), + reply({"room": ROOM, "kind": "public", "head": 40, + "operations": [completion(operation, 31)], "delivery": "not asserted"})] + for page in cases: + with self.subTest(page=page): + self.responses = [reply(sent), reply({**sent, "exact_retry": True}), page] + with self.assertRaises(ValueError): + self.agent.exact_local_retry(operation, "message") + + def test_retry_refuses_scan_beyond_finite_page_budget(self): + self.issue() + operation = "9" * 32 + sent = {"room": ROOM, "operation": operation, "frame_hash": HASH, + "state": "Provisional", "exact_retry": False, "delivery": "not asserted"} + self.responses = [reply(sent), reply({**sent, "exact_retry": True}), + *(outbox([], head=40, next_cursor=after + 1) for after in range(16))] + with self.assertRaisesRegex(ValueError, "exceeded retry scan bound"): + self.agent.exact_local_retry(operation, "message") + self.assertEqual(len(self.launches), 18) + + def test_cross_room_and_exhausted_grant_refuse_by_typed_codes(self): + self.issue() + self.responses = [reply(None, error={"code": -32602, "message": "Invalid tool arguments"}), + reply({"status": "refused", "code": "permission-denied"}, refused=True)] + with self.assertRaises(AgentRefusal) as cross: + self.agent.call("agent.status", room="another") + self.assertEqual(cross.exception.code, -32602) + with self.assertRaises(AgentRefusal) as exhausted: + self.agent.status() + self.assertEqual(exhausted.exception.code, "permission-denied") + + def test_wrong_room_and_untrusted_page_fail_closed(self): + self.issue() + self.responses = [reply({"room": "a" * 32, "kind": "public", "author": AUTHOR}), + reply({"room": ROOM, "records": [{"text": "secret"}]})] + with self.assertRaisesRegex(ValueError, "escaped"): + self.agent.status() + with self.assertRaisesRegex(ValueError, "inert"): + self.agent.messages() + + def test_credential_echo_and_unsupported_delivery_are_refused(self): + self.issue() + self.responses = [reply({"room": ROOM, "secret": TOKEN}), + reply({"room": ROOM, "operation": "9" * 32, "frame_hash": HASH, + "exact_retry": False, "delivery": "delivered"})] + with self.assertRaisesRegex(ValueError, "exposed"): + self.agent.status() + with self.assertRaisesRegex(ValueError, "unsupported delivery"): + self.agent.send("9" * 32, "inert") + + def test_failed_revoke_is_not_reported_as_cleanup(self): + self.issue() + self.daemon.revoke = False + with self.assertRaisesRegex(ValueError, "unconfirmed"): + self.agent.close() + self.assertFalse(self.agent.grant_path.exists()) + self.assertIsNotNone(self.agent.grant) + with self.assertRaisesRegex(ValueError, "do not retry"): + self.agent.close() + self.assertEqual([name for name, _ in self.daemon.calls].count("grant.revoke"), 1) + + def test_lost_revoke_reply_does_not_trigger_a_second_write(self): + self.issue() + def lost(op, **fields): + if op == "grant.revoke": + self.daemon.calls.append((op, fields)) + raise TimeoutError("revoke reply lost") + return FakeDaemon.call(self.daemon, op, **fields) + self.daemon.call = lost + with self.assertRaises(TimeoutError): + self.agent.close() + with self.assertRaisesRegex(ValueError, "do not retry"): + self.agent.close() + self.assertEqual([name for name, _ in self.daemon.calls], + ["room.status", "grant.issue", "grant.revoke"]) + self.assertFalse(self.agent.grant_path.exists()) + self.assertIsNotNone(self.agent.grant) + + +class PairedAgentTests(unittest.TestCase): + """Two fake local issuers; never evidence of separate physical machines.""" + + def setUp(self): + ScopedAgentTests.setUp(self) + other = self.work.parent / "peer-role" + other.mkdir(mode=0o700) + self.second_daemon = FakeDaemon(other, room="a" * 32, author=PEER, + generation="c" * 64, token="d" * 64) + self.second = ScopedAgent(self.second_daemon, "a" * 32, PIN, + other / "grant.json", clock=lambda: 2_000_000_000) + self.machines = ("e" * 64, "f" * 64) + self.operations = ("8" * 32, "9" * 32) + + def transport(self, argv, payload, env, timeout): + return ScopedAgentTests.transport(self, argv, payload, env, timeout) + + def paired_issue(self): + return issue_agent_pair(self.agent, self.second, self.operations, self.machines) + + def test_two_scopes_grants_and_summary_remain_separate_and_body_free(self): + result = self.paired_issue() + self.assertEqual([entry["scope"]["author"] for entry in result["agents"]], + [AUTHOR, PEER]) + self.assertEqual([entry["declared_machine_id"] for entry in result["agents"]], + list(self.machines)) + self.assertEqual([entry["budget"] for entry in result["agents"]], [BUDGET, BUDGET]) + self.assertEqual(result["delivery"], "not asserted") + self.assertIn("unverified", result["machine_identity"]) + self.assertNotIn(TOKEN, json.dumps(result)) + self.assertNotIn(self.second_daemon.token, json.dumps(result)) + self.assertNotIn("grant.json", json.dumps(result)) + self.assertEqual([name for name, _ in self.daemon.calls], + ["room.status", "room.status", "grant.issue"]) + self.assertEqual([name for name, _ in self.second_daemon.calls], + ["room.status", "room.status", "grant.issue"]) + self.assertEqual(json.loads(self.agent.grant_path.read_text()), + {"generation": GENERATION, "token": TOKEN}) + self.assertEqual(json.loads(self.second.grant_path.read_text()), + {"generation": "c" * 64, "token": "d" * 64}) + self.assertEqual(self.launches, []) + self.second.close() + self.agent.close() + self.assertFalse(self.agent.grant_path.exists()) + self.assertFalse(self.second.grant_path.exists()) + with self.assertRaisesRegex(ValueError, "separate homes and private work"): + self.paired_issue() + self.assertEqual(len(self.daemon.calls), 4) + self.assertEqual(len(self.second_daemon.calls), 4) + + def test_separate_objects_with_shared_home_or_work_refuse_before_any_mint(self): + # Distinct daemon wrappers and declared machine IDs do not isolate custody. + original_home = self.second_daemon.home + self.second_daemon.home = self.daemon.home + with self.assertRaisesRegex(ValueError, "separate homes and private work"): + self.paired_issue() + self.second_daemon.home = original_home + shared = FakeDaemon(self.work, room=self.second.room, author=PEER, + generation="c" * 64, token="d" * 64) + shared.home = original_home + collocated = ScopedAgent(shared, self.second.room, PIN, + self.work / "other-grant.json", clock=lambda: 2_000_000_000) + with self.assertRaisesRegex(ValueError, "separate homes and private work"): + issue_agent_pair(self.agent, collocated, self.operations, self.machines) + self.assertEqual(self.daemon.calls, []) + self.assertEqual(self.second_daemon.calls, []) + self.assertEqual(shared.calls, []) + self.assertFalse(self.agent.grant_path.exists()) + self.assertFalse(collocated.grant_path.exists()) + + def test_duplicate_machine_role_and_operation_refuse_before_any_mint(self): + with self.assertRaisesRegex(ValueError, "machine identifiers"): + issue_agent_pair(self.agent, self.second, self.operations, + (self.machines[0], self.machines[0])) + with self.assertRaisesRegex(ValueError, "grant operations"): + issue_agent_pair(self.agent, self.second, (self.operations[0],) * 2, self.machines) + with self.assertRaisesRegex(ValueError, "separate homes and private work"): + issue_agent_pair(self.agent, self.agent, self.operations, self.machines) + self.assertEqual(self.daemon.calls, []) + self.assertEqual(self.second_daemon.calls, []) + + def test_wrong_pin_policy_or_same_author_refuses_without_mint(self): + for changed, value in (("pin", "b" * 64), + ("policy", {"id": "b" * 64, "revision": 1}), + ("author", AUTHOR)): + with self.subTest(changed=changed): + old = self.second_daemon.status[changed] + self.second_daemon.status[changed] = value + with self.assertRaises(ValueError): + self.paired_issue() + self.second_daemon.status[changed] = old + self.assertFalse(self.agent.grant_path.exists()) + self.assertFalse(self.second.grant_path.exists()) + self.assertNotIn("grant.issue", [name for name, _ in self.daemon.calls]) + self.assertNotIn("grant.issue", [name for name, _ in self.second_daemon.calls]) + + def test_second_issuance_mismatch_rolls_back_first_and_never_launches(self): + self.second_daemon.issued_override = {"generation": "c" * 64, "token": "d" * 64, + "room": self.second.room, "scope": {"kind": "private"}, + "permissions": PERMISSIONS, "expires_at": 2_000_000_600} + with self.assertRaisesRegex(ValueError, "differs from selected scope"): + self.paired_issue() + self.assertEqual([name for name, _ in self.daemon.calls][-2:], + ["grant.issue", "grant.revoke"]) + self.assertEqual([name for name, _ in self.second_daemon.calls][-2:], + ["grant.issue", "grant.revoke"]) + self.assertFalse(self.agent.grant_path.exists()) + self.assertFalse(self.second.grant_path.exists()) + self.assertIsNone(self.agent.grant) + self.assertIsNone(self.second.grant) + self.assertEqual(self.launches, []) + + def test_credential_reuse_refuses_and_revokes_both(self): + self.second_daemon.generation, self.second_daemon.token = GENERATION, TOKEN + with self.assertRaisesRegex(ValueError, "reused credentials"): + self.paired_issue() + self.assertFalse(self.agent.grant_path.exists()) + self.assertFalse(self.second.grant_path.exists()) + self.assertEqual([name for name, _ in self.daemon.calls].count("grant.revoke"), 1) + self.assertEqual([name for name, _ in self.second_daemon.calls].count("grant.revoke"), 1) + + def test_second_issuance_cleanup_uncertainty_is_explicit(self): + self.second_daemon.issued_override = {"generation": "c" * 64, "token": "d" * 64, + "room": self.second.room, "scope": {"kind": "private"}, + "permissions": PERMISSIONS, "expires_at": 2_000_000_600} + self.daemon.revoke = False + with self.assertRaisesRegex(ValueError, "cleanup unconfirmed"): + self.paired_issue() + self.assertFalse(self.agent.grant_path.exists()) + self.assertFalse(self.second.grant_path.exists()) + self.assertIsNotNone(self.agent.grant) + self.assertEqual(self.launches, []) + with self.assertRaisesRegex(ValueError, "do not retry"): + self.agent.close() + self.assertEqual([name for name, _ in self.daemon.calls].count("grant.revoke"), 1) + + +class RealProcessGrantTests(unittest.TestCase): + """Opt-in single-machine daemon/MCP regression, never two-agent evidence.""" + + @unittest.skipUnless(os.environ.get("HEADLESS_AGENT_TEST_BINARY"), + "set HEADLESS_AGENT_TEST_BINARY to an owned built vhalla CLI") + def test_fresh_mcp_launches_share_one_finite_grant_and_paginated_outbox(self): + source = Path(os.environ["HEADLESS_AGENT_TEST_BINARY"]) + self.assertTrue(source.is_absolute() and source.is_file() and not source.is_symlink()) + # Use a short, owned private directory for macOS's Unix socket bound. + # The dummy SHA is never emitted as a receipt or interpreted as a source attestation. + scratch = Path(os.environ.get("HEADLESS_AGENT_TEST_WORKDIR", ".")).absolute() + self.assertTrue(scratch.is_dir() and scratch.resolve() == scratch + and scratch.stat().st_uid == os.geteuid(), + "choose an owned HEADLESS_AGENT_TEST_WORKDIR inside the workspace") + self.assertLess(len(os.fsencode(scratch / "aq-XXXXXXXX" / "r" / "home" + / "control" / "admin.sock")), 104, + "choose a shorter HEADLESS_AGENT_TEST_WORKDIR inside the workspace") + with tempfile.TemporaryDirectory(prefix="aq-", dir=scratch) as temp: + root = Path(temp).resolve() + candidate = root / "candidate" + shutil.copyfile(source, candidate) + candidate.chmod(0o500) + process = McpRun(candidate, digest(candidate), root / "r", "0" * 40) + admin_calls = [] + launches = [] + + def admin_call(op, **fields): + admin_calls.append(op) + return process.call(op, **fields) + + def scoped_launch(argv, payload, env, timeout): + launches.append(argv[-4]) # action only; never record the private payload/grant + return public.exchange(argv, payload, env, timeout) + + daemon = SimpleNamespace(config={"work": str(process.work)}, home=process.home, + argv=[str(candidate), "--no-update", "daemon"], + env=process.env, call=admin_call) + agent = None + try: + process.start() + created = process.call("room.create", operation=f"{1:032x}", kind="public", + limits={"max_records": 128, "max_record_bytes": 1048576}) + agent = ScopedAgent(daemon, created["room"], created["pin"], + process.work / "agent-grant.json", transport=scoped_launch) + agent.issue(f"{2:032x}") + before = agent.status() + self.assertEqual(before["remaining"]["send_attempts"], BUDGET["send_attempts"]) + operation, text = f"{100:032x}", "synthetic finite-grant retry" + first = agent.send(operation, text) + self.assertFalse(first["exact_retry"]) + retried = agent.send(operation, text) + self.assertTrue(retried["exact_retry"]) + self.assertEqual(retried["frame_hash"], first["frame_hash"]) + for number in range(1, BUDGET["send_attempts"] - 1): + sent = agent.send(f"{100 + number:032x}", f"synthetic finite-grant {number}") + self.assertFalse(sent["exact_retry"]) + after = agent.status() + self.assertEqual(after["remaining"]["send_attempts"], 0) + self.assertLess(after["remaining"]["calls"], before["remaining"]["calls"]) + + cursor, head, rows = 0, None, [] + for _ in range(MAX_OUTBOX_PAGES): + page = agent.outbox_status(after=cursor, limit=1) + self.assertEqual(page["head"], head if head is not None else page["head"]) + head = page["head"] + rows.extend(page["operations"]) + if page["next"] is None: + break + cursor = page["next"] + else: + self.fail("local outbox exceeded finite page bound") + self.assertGreater(len(rows), 1) # Actual pagination, not a single-page stub. + matched = [row for row in rows if row["operation"] == operation] + self.assertEqual(len(matched), 1) + self.assertEqual(matched[0]["kind"], "Event") + self.assertEqual(matched[0]["frame_hash"], first["frame_hash"]) + messages = agent.messages(after=0, limit=16) + self.assertEqual(sum(row["text"] == text for row in messages), 1) + with self.assertRaises(AgentRefusal) as refused: + agent.send(f"{999:032x}", "synthetic exhausted-grant refusal") + self.assertEqual(refused.exception.code, "permission-denied") + self.assertEqual(agent.status()["remaining"]["send_attempts"], 0) + self.assertEqual(admin_calls, ["room.status", "grant.issue"]) + self.assertGreaterEqual(len(launches), BUDGET["send_attempts"] + 5) + finally: + try: + if agent is not None: + agent.close() + finally: + self.assertTrue(process.cleanup()) + self.assertFalse(process.result["cleanup_fallback_used"]) + self.assertFalse(agent.grant_path.exists()) + self.assertEqual(admin_calls, ["room.status", "grant.issue", "grant.revoke"]) + + +if __name__ == "__main__": + unittest.main()