"""Client-binding HTTP admission, actor lifecycle, and real approval effects.""" import json import time import unittest import urllib.error import urllib.parse from harness import Albedo, Provider, exclusive, python, text def model(request): if request["input"][-1].get("type") == "function_call_output": return text("finished") return python( "from pathlib import Path\nPath('authorized-effect').write_text('allowed')" ) class ClientBindingHTTPTest(unittest.TestCase): def setUp(self): provider = Provider(model) self.addCleanup(provider.close) self.app = Albedo(provider, protocol="responses") self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.sid = self.app.session() self.route = f"/sessions/{self.sid}/client-binding" def value(self, path=None, body=None, **kwargs): with self.app.api(path or self.route, body, **kwargs) as response: return json.load(response) def install(self, expected=None): return self.value( body={ "binding_id": "test-editor", "expected_generation": expected, "capabilities": { "require_tool_approval": True, "read_text": True, "write_text": True, }, "mcp_servers": [], }, method="PUT", ) def reject_http(self, status, path=None, body=None, **kwargs): with self.assertRaises(urllib.error.HTTPError) as raised: self.value(path, body, **kwargs) self.assertEqual(raised.exception.code, status, raised.exception.read()) def revoke(self, generation, native=False): query = urllib.parse.urlencode( {"generation": generation, "adopt_native": str(native).lower()} ) with self.app.api(self.route + "?" + query, method="DELETE") as response: self.assertEqual(response.status, 204) self.assertEqual(response.read(), b"") def pending(self): deadline = time.monotonic() + 10 while time.monotonic() < deadline: snapshot = self.value() if snapshot["pending"]: return snapshot["pending"][0] time.sleep(0.02) self.fail("approval was not published") def answer(self, pending, reply, generation=None): with self.app.api( self.route + "/requests/" + pending["request_id"], {"generation": generation or pending["generation"], "reply": reply}, method="PUT", ) as response: self.assertEqual(response.status, 204) self.assertEqual(response.read(), b"") def test_lifecycle_generation_reset_and_native_adoption(self): self.reject_http(404) installed = self.install() generation = installed["generation"] self.assertEqual(installed["state"], "active") self.assertTrue(installed["require_approval"]) self.assertEqual(installed["mcp_server_names"], []) self.reject_http(409, body={"generation": "stale"}, method="PATCH") renewed = self.value(body={"generation": generation}, method="PATCH") self.assertGreaterEqual( renewed["lease_expires_at_ms"], installed["lease_expires_at_ms"] ) self.reject_http(409, self.route + "?generation=stale") with self.app.api( self.route, headers={"Accept": "text/event-stream"} ) as response: reset = next( json.loads(line[6:]) for line in response if line.startswith(b"data: ") ) self.assertEqual(reset["generation"], generation) self.assertEqual(reset["binding_id"], "test-editor") self.reject_http( 409, body={ "binding_id": "second", "expected_generation": None, "capabilities": installed["capabilities"], "mcp_servers": [], }, method="PUT", ) replaced = self.install(generation) self.assertNotEqual(replaced["generation"], generation) self.revoke(replaced["generation"]) self.assertEqual(self.value()["state"], "disconnected") self.assertTrue(self.value()["require_approval"]) self.revoke(replaced["generation"], native=True) self.reject_http(404) def test_approval_kind_generation_and_idle_actor_before_effect(self): installed = self.install() with self.app.prompt(self.sid, "run approved marker") as response: response.read() pending = self.pending() self.assertEqual(pending["request"]["kind"], "authorize_tool") marker = self.app.workspace / "authorized-effect" self.assertFalse(marker.exists()) path = self.route + "/requests/" + pending["request_id"] self.reject_http( 409, path, { "generation": "stale", "reply": {"kind": "decision", "decision": "allow_once"}, }, method="PUT", ) self.reject_http( 409, path, {"generation": installed["generation"], "reply": {"kind": "written"}}, method="PUT", ) self.reject_http( 400, path, { "generation": installed["generation"], "reply": {"kind": "decision", "decision": "allow_once", "extra": True}, }, method="PUT", ) self.reject_http( 409, self.route + "?generation=" + installed["generation"] + "&adopt_native=true", method="DELETE", ) self.assertFalse(marker.exists()) self.answer(pending, {"kind": "decision", "decision": "allow_once"}) self.app.idle(self.sid) self.assertEqual(marker.read_text(), "allowed") self.revoke(installed["generation"], native=True) def test_interrupt_cancels_pending_permission_before_lease_expiry(self): installed = self.install() with self.app.prompt(self.sid, "interrupt while permission waits") as response: response.read() pending = self.pending() snapshot = self.value(f"/sessions/{self.sid}?tail=0") started = time.monotonic() self.value( f"/sessions/{self.sid}/interrupt", { "run_id": snapshot["status"]["run_id"], "through_input_order": snapshot["input_order"], }, ) self.app.idle(self.sid, timeout=8) self.assertLess(time.monotonic() - started, 8) self.assertEqual(self.value()["pending"], []) self.reject_http( 409, self.route + "/requests/" + pending["request_id"], { "generation": pending["generation"], "reply": {"kind": "decision", "decision": "allow_once"}, }, method="PUT", ) self.assertFalse((self.app.workspace / "authorized-effect").exists()) self.revoke(installed["generation"], native=True) def test_failed_reload_retains_prior_generation_and_redacts_descriptor(self): installed = self.install() with self.assertRaises(urllib.error.HTTPError) as raised: self.value( body={ "binding_id": "broken-editor", "expected_generation": installed["generation"], "capabilities": installed["capabilities"], "mcp_servers": [ { "name": "broken", "type": "stdio", "command": "/no/such/acp-binding-command", "args": [], "env": {"EDITOR_SECRET": "must-not-echo"}, } ], }, method="PUT", ) self.assertEqual(raised.exception.code, 503) self.assertNotIn("must-not-echo", raised.exception.read().decode()) surviving = self.value() self.assertEqual(surviving["generation"], installed["generation"]) self.assertEqual(surviving["state"], "active") self.assertTrue(surviving["require_approval"]) self.revoke(installed["generation"], native=True) @exclusive def test_restart_retains_failclosed_policy_without_descriptors(self): installed = self.install() with self.app.prompt( self.sid, "hold outstanding permission across crash" ) as response: response.read() pending = self.pending() self.assertFalse((self.app.workspace / "authorized-effect").exists()) self.app.restart(crash=True) self.reject_http(404) with self.app.prompt(self.sid, "approval required after restart") as response: response.read() self.app.idle(self.sid) self.assertFalse((self.app.workspace / "authorized-effect").exists()) self.reject_http( 404, self.route + "/requests/" + pending["request_id"], { "generation": installed["generation"], "reply": {"kind": "decision", "decision": "allow_once"}, }, method="PUT", ) self.revoke(installed["generation"], native=True) with self.app.prompt( self.sid, "native execution explicitly adopted" ) as response: response.read() self.app.idle(self.sid) self.assertEqual( (self.app.workspace / "authorized-effect").read_text(), "allowed" )