feat(lease): add single-turn Claude promotion trigger
This commit is contained in:
Jason Woltje
@@ -0,0 +1,519 @@
|
||||
#!/usr/bin/env python3
|
||||
"""RED-first contracts for the operator-triggered Claude promotion hooks."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
import subprocess
|
||||
import tempfile
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
from unittest import mock
|
||||
|
||||
|
||||
PACKAGE_ROOT = Path(__file__).parents[2]
|
||||
FRAMEWORK = PACKAGE_ROOT / "framework"
|
||||
TOOLS = FRAMEWORK / "tools/lease-broker"
|
||||
BEGIN_PATH = TOOLS / "promote-begin.py"
|
||||
COMPLETE_PATH = TOOLS / "promote-complete.py"
|
||||
OBSERVER_CLIENT_PATH = TOOLS / "receipt-observer-client.py"
|
||||
RECEIPT_CHALLENGE_PATH = TOOLS / "receipt_challenge.py"
|
||||
CLAUDE_SETTINGS = FRAMEWORK / "runtime/claude/settings.json"
|
||||
CLAUDE_COMMAND = FRAMEWORK / "runtime/claude/commands/mosaic-promote.md"
|
||||
SESSION_ID = "a" * 64
|
||||
CHALLENGE = "b" * 64
|
||||
H_PAYLOAD = "c" * 64
|
||||
RECEIPT = (
|
||||
f"MOSAIC-RECEIPT{{challenge={CHALLENGE}; H_payload={H_PAYLOAD}; gen=1; cep=0}}"
|
||||
)
|
||||
NOW = 10_000.0
|
||||
|
||||
|
||||
def load_module(name: str, path: Path):
|
||||
if not path.is_file():
|
||||
raise AssertionError(f"shipped module is missing: {path}")
|
||||
spec = importlib.util.spec_from_file_location(name, path)
|
||||
if spec is None or spec.loader is None:
|
||||
raise RuntimeError(f"unable to load {name}")
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
class PromotionHookFixture(unittest.TestCase):
|
||||
@classmethod
|
||||
def setUpClass(cls) -> None:
|
||||
cls.begin = load_module("promotion_begin_test", BEGIN_PATH)
|
||||
cls.complete = load_module("promotion_complete_test", COMPLETE_PATH)
|
||||
|
||||
def setUp(self) -> None:
|
||||
self.temporary = tempfile.TemporaryDirectory()
|
||||
self.runtime_dir = Path(self.temporary.name)
|
||||
self.environment = {
|
||||
"XDG_RUNTIME_DIR": str(self.runtime_dir),
|
||||
"MOSAIC_LEASE_SESSION_ID": SESSION_ID,
|
||||
}
|
||||
self.pending_dir = self.runtime_dir / "mosaic-lease"
|
||||
self.pending_file = self.pending_dir / f"pending-{SESSION_ID}"
|
||||
|
||||
def tearDown(self) -> None:
|
||||
self.temporary.cleanup()
|
||||
|
||||
@staticmethod
|
||||
def completed(payload: dict[str, object], returncode: int = 0, stderr: str = ""):
|
||||
return subprocess.CompletedProcess(
|
||||
["lease_promote.py"],
|
||||
returncode,
|
||||
json.dumps(payload),
|
||||
stderr,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def successful_begin_reply(cls, **extra: object) -> dict[str, object]:
|
||||
return {
|
||||
"ok": True,
|
||||
"state": "PENDING_VERIFICATION",
|
||||
"receipt_challenge": CHALLENGE,
|
||||
"receipt": RECEIPT,
|
||||
"binding": {
|
||||
"compaction_epoch": 0,
|
||||
"request_epoch": 0,
|
||||
"h_source": "d" * 64,
|
||||
"h_payload": H_PAYLOAD,
|
||||
"runtime_generation": 1,
|
||||
"schema_version": 1,
|
||||
},
|
||||
**extra,
|
||||
}
|
||||
|
||||
def run_begin(
|
||||
self,
|
||||
prompt: str,
|
||||
runner: mock.Mock,
|
||||
) -> tuple[int, str, str]:
|
||||
stdout = io.StringIO()
|
||||
stderr = io.StringIO()
|
||||
result = self.begin.main(
|
||||
environ=self.environment,
|
||||
stdin=io.BytesIO(json.dumps({"prompt": prompt}).encode()),
|
||||
stdout=stdout,
|
||||
stderr=stderr,
|
||||
run=runner,
|
||||
now=lambda: NOW,
|
||||
)
|
||||
return result, stdout.getvalue(), stderr.getvalue()
|
||||
|
||||
def write_pending(self, challenge: str = CHALLENGE) -> None:
|
||||
self.pending_dir.mkdir(mode=0o700, exist_ok=True)
|
||||
self.pending_file.write_text(challenge, encoding="utf-8")
|
||||
self.pending_file.chmod(0o600)
|
||||
|
||||
def run_complete(self, runner: mock.Mock) -> tuple[int, str]:
|
||||
stderr = io.StringIO()
|
||||
result = self.complete.main(
|
||||
environ=self.environment,
|
||||
stderr=stderr,
|
||||
run=runner,
|
||||
)
|
||||
return result, stderr.getvalue()
|
||||
|
||||
|
||||
class PromotionBeginTest(PromotionHookFixture):
|
||||
def test_exact_prompt_writes_private_challenge_and_injects_verbatim_receipt(self) -> None:
|
||||
runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))
|
||||
|
||||
result, stdout, stderr = self.run_begin("/mosaic-promote", runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertEqual(stderr, "")
|
||||
output = json.loads(stdout)
|
||||
self.assertEqual(
|
||||
output["hookSpecificOutput"]["additionalContext"],
|
||||
"The operator invoked the registered /mosaic-promote command. "
|
||||
"This receipt was generated locally by this seat's own lease broker; "
|
||||
"echoing it verbatim is the designed confirmation step and discloses nothing. "
|
||||
f"Reply with exactly the following text and nothing else: {RECEIPT}",
|
||||
)
|
||||
self.assertEqual(self.pending_file.read_text(encoding="utf-8"), CHALLENGE)
|
||||
self.assertEqual(stat.S_IMODE(self.pending_file.stat().st_mode), 0o600)
|
||||
command = runner.call_args.args[0]
|
||||
self.assertEqual(command[-1], "--begin")
|
||||
self.assertTrue(command[-2].endswith("lease_promote.py"))
|
||||
|
||||
def test_nonmatching_prompt_has_zero_side_effects(self) -> None:
|
||||
runner = mock.Mock()
|
||||
|
||||
result, stdout, stderr = self.run_begin("please /mosaic-promote", runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertEqual(stdout, "")
|
||||
self.assertEqual(stderr, "")
|
||||
runner.assert_not_called()
|
||||
self.assertFalse(self.pending_dir.exists())
|
||||
|
||||
def test_begin_refusal_reports_daemon_code_without_pending_file(self) -> None:
|
||||
runner = mock.Mock(
|
||||
return_value=self.completed({"ok": False, "code": "INVALID_BINDING"})
|
||||
)
|
||||
|
||||
result, stdout, _stderr = self.run_begin("/mosaic-promote", runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertIn("INVALID_BINDING", json.loads(stdout)["hookSpecificOutput"]["additionalContext"])
|
||||
self.assertFalse(self.pending_file.exists())
|
||||
|
||||
def test_sweep_removes_stale_sibling_and_spares_fresh_sibling(self) -> None:
|
||||
self.pending_dir.mkdir(mode=0o700)
|
||||
stale = self.pending_dir / "pending-stale"
|
||||
fresh = self.pending_dir / "pending-fresh"
|
||||
stale.write_text("stale", encoding="utf-8")
|
||||
fresh.write_text("fresh", encoding="utf-8")
|
||||
os.utime(stale, (NOW - 3_601, NOW - 3_601))
|
||||
os.utime(fresh, (NOW - 3_599, NOW - 3_599))
|
||||
runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))
|
||||
|
||||
result, _stdout, _stderr = self.run_begin("/mosaic-promote", runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertFalse(stale.exists())
|
||||
self.assertTrue(fresh.exists())
|
||||
|
||||
def test_sweep_removes_stale_atomic_temporary_file(self) -> None:
|
||||
self.pending_dir.mkdir(mode=0o700)
|
||||
stale_temporary = self.pending_dir / f".pending-{SESSION_ID}.tmp-abandoned"
|
||||
stale_temporary.write_text("partial", encoding="utf-8")
|
||||
os.utime(stale_temporary, (NOW - 3_601, NOW - 3_601))
|
||||
runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))
|
||||
|
||||
result, _stdout, _stderr = self.run_begin("/mosaic-promote", runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertFalse(stale_temporary.exists())
|
||||
|
||||
def test_insecure_pending_directory_mode_refuses_before_begin(self) -> None:
|
||||
self.pending_dir.mkdir(mode=0o755)
|
||||
self.pending_dir.chmod(0o755)
|
||||
runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))
|
||||
|
||||
result, stdout, _stderr = self.run_begin("/mosaic-promote", runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
runner.assert_not_called()
|
||||
self.assertFalse(self.pending_file.exists())
|
||||
self.assertIn("PROMOTION_TRIGGER_FAILED", stdout)
|
||||
|
||||
def test_insecure_runtime_directory_mode_refuses_before_begin(self) -> None:
|
||||
self.runtime_dir.chmod(0o755)
|
||||
runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))
|
||||
|
||||
result, stdout, _stderr = self.run_begin("/mosaic-promote", runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
runner.assert_not_called()
|
||||
self.assertIn("PROMOTION_TRIGGER_FAILED", stdout)
|
||||
|
||||
def test_parent_symlink_cannot_redirect_pending_write(self) -> None:
|
||||
outside = self.runtime_dir / "outside"
|
||||
outside.mkdir(mode=0o700)
|
||||
self.pending_dir.symlink_to(outside, target_is_directory=True)
|
||||
runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))
|
||||
|
||||
result, _stdout, _stderr = self.run_begin("/mosaic-promote", runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
runner.assert_not_called()
|
||||
self.assertFalse((outside / f"pending-{SESSION_ID}").exists())
|
||||
|
||||
def test_concurrent_begin_is_refused_without_minting_a_second_challenge(self) -> None:
|
||||
inner_runner = mock.Mock(return_value=self.completed(self.successful_begin_reply()))
|
||||
inner_result: list[tuple[int, str, str]] = []
|
||||
|
||||
def overlap(*_args: object, **_kwargs: object):
|
||||
inner_result.append(self.run_begin("/mosaic-promote", inner_runner))
|
||||
return self.completed(self.successful_begin_reply())
|
||||
|
||||
outer_runner = mock.Mock(side_effect=overlap)
|
||||
|
||||
result, _stdout, _stderr = self.run_begin("/mosaic-promote", outer_runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
inner_runner.assert_not_called()
|
||||
self.assertEqual(inner_result[0][0], 0)
|
||||
self.assertIn("PROMOTION_ALREADY_IN_PROGRESS", inner_result[0][1])
|
||||
|
||||
def test_non_ascii_receipt_reply_is_rejected_without_crashing_hook(self) -> None:
|
||||
reply = self.successful_begin_reply()
|
||||
reply["receipt"] = "MOSAIC—RECEIPT"
|
||||
runner = mock.Mock(return_value=self.completed(reply))
|
||||
|
||||
result, stdout, _stderr = self.run_begin("/mosaic-promote", runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertFalse(self.pending_file.exists())
|
||||
self.assertIn("INVALID_PROMOTER_REPLY", stdout)
|
||||
|
||||
def test_success_shaped_reply_with_extra_fields_is_rejected(self) -> None:
|
||||
runner = mock.Mock(
|
||||
return_value=self.completed(self.successful_begin_reply(unexpected=True))
|
||||
)
|
||||
|
||||
result, stdout, _stderr = self.run_begin("/mosaic-promote", runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertFalse(self.pending_file.exists())
|
||||
self.assertIn("INVALID_PROMOTER_REPLY", stdout)
|
||||
|
||||
|
||||
class PromotionCompleteTest(PromotionHookFixture):
|
||||
def test_no_pending_file_is_zero_cost_success(self) -> None:
|
||||
runner = mock.Mock()
|
||||
|
||||
result, stderr = self.run_complete(runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertEqual(stderr, "")
|
||||
runner.assert_not_called()
|
||||
|
||||
def test_success_deletes_pending_file(self) -> None:
|
||||
self.write_pending()
|
||||
runner = mock.Mock(
|
||||
return_value=self.completed(
|
||||
{"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
|
||||
)
|
||||
)
|
||||
|
||||
result, _stderr = self.run_complete(runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertFalse(self.pending_file.exists())
|
||||
self.assertEqual(runner.call_args.args[0][-2:], ["--complete", CHALLENGE])
|
||||
|
||||
def test_each_terminal_failure_deletes_pending_file(self) -> None:
|
||||
terminal_codes = (
|
||||
"RECEIPT_REPLAY",
|
||||
"RECEIPT_MISMATCH",
|
||||
"INVALID_LEASE_TRANSITION",
|
||||
"PROMOTION_TOKEN_INVALID",
|
||||
)
|
||||
for code in terminal_codes:
|
||||
with self.subTest(code=code):
|
||||
self.write_pending()
|
||||
runner = mock.Mock(
|
||||
return_value=self.completed(
|
||||
{"stage": "observe_receipt", "ok": False, "code": code}
|
||||
)
|
||||
)
|
||||
|
||||
result, stderr = self.run_complete(runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertFalse(self.pending_file.exists())
|
||||
self.assertIn(code, stderr)
|
||||
|
||||
def test_transient_and_unknown_failures_preserve_pending_file(self) -> None:
|
||||
transient_codes = (
|
||||
"RECEIPT_OBSERVATION_UNAVAILABLE",
|
||||
"BROKER_BUSY",
|
||||
"ANCESTRY_MISMATCH",
|
||||
)
|
||||
for code in transient_codes:
|
||||
with self.subTest(code=code):
|
||||
self.write_pending()
|
||||
runner = mock.Mock(
|
||||
return_value=self.completed(
|
||||
{"stage": "observe_receipt", "ok": False, "code": code}
|
||||
)
|
||||
)
|
||||
|
||||
result, stderr = self.run_complete(runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertTrue(self.pending_file.exists())
|
||||
self.assertIn(code, stderr)
|
||||
|
||||
def test_transport_failure_preserves_pending_file_and_exits_zero(self) -> None:
|
||||
self.write_pending()
|
||||
runner = mock.Mock(
|
||||
return_value=self.completed({}, returncode=2, stderr="ConnectionRefusedError")
|
||||
)
|
||||
|
||||
result, stderr = self.run_complete(runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertTrue(self.pending_file.exists())
|
||||
self.assertIn("ConnectionRefusedError", stderr)
|
||||
|
||||
def test_insecure_runtime_directory_mode_preserves_pending(self) -> None:
|
||||
self.write_pending(CHALLENGE)
|
||||
self.runtime_dir.chmod(0o755)
|
||||
runner = mock.Mock(
|
||||
return_value=self.completed(
|
||||
{"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
|
||||
)
|
||||
)
|
||||
|
||||
result, _stderr = self.run_complete(runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
runner.assert_not_called()
|
||||
self.assertTrue(self.pending_file.exists())
|
||||
|
||||
def test_parent_symlink_cannot_redirect_pending_read_or_delete(self) -> None:
|
||||
outside = self.runtime_dir / "outside"
|
||||
outside.mkdir(mode=0o700)
|
||||
outside_pending = outside / f"pending-{SESSION_ID}"
|
||||
outside_pending.write_text(CHALLENGE, encoding="utf-8")
|
||||
outside_pending.chmod(0o600)
|
||||
self.pending_dir.symlink_to(outside, target_is_directory=True)
|
||||
runner = mock.Mock(
|
||||
return_value=self.completed(
|
||||
{"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
|
||||
)
|
||||
)
|
||||
|
||||
result, _stderr = self.run_complete(runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
runner.assert_not_called()
|
||||
self.assertTrue(outside_pending.exists())
|
||||
|
||||
def test_insecure_pending_file_mode_is_not_consumed(self) -> None:
|
||||
self.write_pending(CHALLENGE)
|
||||
self.pending_file.chmod(0o644)
|
||||
runner = mock.Mock(
|
||||
return_value=self.completed(
|
||||
{"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
|
||||
)
|
||||
)
|
||||
|
||||
result, _stderr = self.run_complete(runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
runner.assert_not_called()
|
||||
self.assertTrue(self.pending_file.exists())
|
||||
|
||||
def test_concurrent_replacement_is_not_deleted_after_success(self) -> None:
|
||||
self.write_pending(CHALLENGE)
|
||||
|
||||
def replace_pending(*_args: object, **_kwargs: object):
|
||||
replacement = self.pending_dir / "replacement"
|
||||
replacement.write_text("replacement", encoding="utf-8")
|
||||
replacement.chmod(0o600)
|
||||
os.replace(replacement, self.pending_file)
|
||||
return self.completed(
|
||||
{"stage": "promote_lease", "ok": True, "state": "VERIFIED"}
|
||||
)
|
||||
|
||||
runner = mock.Mock(side_effect=replace_pending)
|
||||
|
||||
result, _stderr = self.run_complete(runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertEqual(self.pending_file.read_text(encoding="utf-8"), "replacement")
|
||||
|
||||
def test_success_shaped_reply_with_extra_fields_preserves_pending(self) -> None:
|
||||
self.write_pending(CHALLENGE)
|
||||
runner = mock.Mock(
|
||||
return_value=self.completed(
|
||||
{
|
||||
"stage": "promote_lease",
|
||||
"ok": True,
|
||||
"state": "VERIFIED",
|
||||
"unexpected": True,
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
result, _stderr = self.run_complete(runner)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertTrue(self.pending_file.exists())
|
||||
|
||||
|
||||
class PromotionTemplateWiringTest(unittest.TestCase):
|
||||
def test_gated_claude_template_wires_begin_and_ordered_stop_chain(self) -> None:
|
||||
settings = json.loads(CLAUDE_SETTINGS.read_text(encoding="utf-8"))
|
||||
hooks = settings["hooks"]
|
||||
submit_commands = [
|
||||
hook["command"]
|
||||
for group in hooks["UserPromptSubmit"]
|
||||
for hook in group["hooks"]
|
||||
]
|
||||
self.assertEqual(
|
||||
submit_commands,
|
||||
["python3 ~/.config/mosaic/tools/lease-broker/promote-begin.py"],
|
||||
)
|
||||
self.assertEqual(
|
||||
[group.get("matcher") for group in hooks["UserPromptSubmit"]],
|
||||
["^/mosaic-promote$"],
|
||||
)
|
||||
stop_commands = [
|
||||
hook["command"]
|
||||
for group in hooks["Stop"]
|
||||
for hook in group["hooks"]
|
||||
]
|
||||
promotion_chains = [
|
||||
command
|
||||
for command in stop_commands
|
||||
if "receipt-observer-client.py" in command and "promote-complete.py" in command
|
||||
]
|
||||
self.assertEqual(len(promotion_chains), 1)
|
||||
chain = promotion_chains[0]
|
||||
self.assertLess(
|
||||
chain.index("receipt-observer-client.py"),
|
||||
chain.index("promote-complete.py"),
|
||||
)
|
||||
self.assertIn("observer_status=$?", chain)
|
||||
self.assertTrue(chain.endswith("exit $observer_status"))
|
||||
|
||||
def test_registered_command_is_one_line_and_defers_to_injected_instruction(self) -> None:
|
||||
body = CLAUDE_COMMAND.read_text(encoding="utf-8")
|
||||
self.assertEqual(
|
||||
body,
|
||||
"I invoked this registered command to authorize lease promotion; follow the local seat broker's injected receipt confirmation instruction exactly.\n",
|
||||
)
|
||||
|
||||
|
||||
class PromotionVerbatimToleranceTest(unittest.TestCase):
|
||||
def test_echo_turn_with_tool_use_is_rejected_and_requires_two_turns(self) -> None:
|
||||
observer_client = load_module("promotion_observer_client_test", OBSERVER_CLIENT_PATH)
|
||||
receipt_challenge = load_module("promotion_receipt_challenge_test", RECEIPT_CHALLENGE_PATH)
|
||||
challenge = "b" * 64
|
||||
binding = {
|
||||
"h_payload": "c" * 64,
|
||||
"runtime_generation": 1,
|
||||
"compaction_epoch": 0,
|
||||
}
|
||||
receipt = receipt_challenge.receipt_for(challenge, binding)
|
||||
real_claude_entry = {
|
||||
"message": {
|
||||
"role": "assistant",
|
||||
"content": [
|
||||
{"type": "text", "text": receipt},
|
||||
{
|
||||
"type": "tool_use",
|
||||
"id": "tool-1",
|
||||
"name": "mcp__discord__reply",
|
||||
"input": {"message": "promoted"},
|
||||
},
|
||||
],
|
||||
}
|
||||
}
|
||||
|
||||
extracted = observer_client.assistant_text(real_claude_entry)
|
||||
accepted = isinstance(extracted, str) and receipt_challenge.is_verbatim_receipt(
|
||||
extracted,
|
||||
challenge,
|
||||
binding,
|
||||
)
|
||||
|
||||
self.assertIsNone(extracted)
|
||||
self.assertFalse(accepted)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -207,6 +207,24 @@ class ReceiptObserverTest(BrokerFixture):
|
||||
with self.assertRaisesRegex(DAEMON.BrokerFailure, "INVALID_LEASE_TRANSITION"):
|
||||
self.promote(current["receipt_challenge"])
|
||||
|
||||
def test_non_ascii_observation_is_mismatch_and_broker_keeps_serving(self) -> None:
|
||||
construction, binding = self.construction()
|
||||
cycle = self.begin(binding, construction)
|
||||
self.record("I refuse—this is not the receipt")
|
||||
|
||||
with self.assertRaisesRegex(DAEMON.BrokerFailure, "RECEIPT_MISMATCH"):
|
||||
self.observe(cycle["receipt_challenge"])
|
||||
|
||||
denied = self.broker.handle(self.peer, {
|
||||
"action": "authorize_tool",
|
||||
"session_id": self.session_id,
|
||||
"runtime_generation": 7,
|
||||
"runtime": "pi",
|
||||
"tool_name": "bash",
|
||||
})
|
||||
self.assertEqual(denied["decision"], "deny")
|
||||
self.assertEqual(denied["state"], DAEMON.LEASE_UNVERIFIED)
|
||||
|
||||
def test_t29_altered_model_hash_cannot_promote_against_shipped_binding(self) -> None:
|
||||
construction, binding = self.construction()
|
||||
cycle = self.begin(binding, construction)
|
||||
|
||||
@@ -6,6 +6,7 @@ from __future__ import annotations
|
||||
import importlib.util
|
||||
import io
|
||||
import json
|
||||
import tempfile
|
||||
import unittest
|
||||
from contextlib import redirect_stderr
|
||||
from pathlib import Path
|
||||
@@ -69,6 +70,7 @@ class ReceiptObserverClientExitSemanticsTest(unittest.TestCase):
|
||||
input_bytes: bytes = VALID_INPUT,
|
||||
reply: dict[str, object] | None = None,
|
||||
transport_error: OSError | None = None,
|
||||
runtime: str = "pi",
|
||||
) -> tuple[int, str, mock.Mock]:
|
||||
request = mock.Mock(return_value=reply)
|
||||
if transport_error is not None:
|
||||
@@ -79,9 +81,111 @@ class ReceiptObserverClientExitSemanticsTest(unittest.TestCase):
|
||||
mock.patch.object(CLIENT, "observer_request", request),
|
||||
redirect_stderr(stderr),
|
||||
):
|
||||
result = CLIENT.main(["--runtime", "pi"], environ=ENVIRONMENT)
|
||||
arguments = ["--runtime", runtime]
|
||||
if runtime == "claude":
|
||||
arguments.append("--latest-entry")
|
||||
result = CLIENT.main(arguments, environ=ENVIRONMENT)
|
||||
return result, stderr.getvalue(), request
|
||||
|
||||
def test_claude_prefers_inline_last_assistant_message(self) -> None:
|
||||
inline = "the just-finished assistant message"
|
||||
input_bytes = json.dumps({
|
||||
"last_assistant_message": inline,
|
||||
"transcript_path": "/must/not/be/opened.jsonl",
|
||||
}).encode()
|
||||
|
||||
result, stderr, request = self.run_client(
|
||||
input_bytes=input_bytes,
|
||||
reply={"ok": True},
|
||||
runtime="claude",
|
||||
)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertEqual(stderr, "")
|
||||
self.assertEqual(
|
||||
request.call_args.args[1]["latest_assistant_message"],
|
||||
inline,
|
||||
)
|
||||
|
||||
def test_claude_falls_back_to_transcript_when_inline_field_is_absent(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
transcript = Path(directory) / "transcript.jsonl"
|
||||
transcript.write_text(
|
||||
json.dumps({
|
||||
"message": {
|
||||
"role": "assistant",
|
||||
"content": [{"type": "text", "text": "fallback message"}],
|
||||
}
|
||||
})
|
||||
+ "\n",
|
||||
encoding="utf-8",
|
||||
)
|
||||
input_bytes = json.dumps({"transcript_path": str(transcript)}).encode()
|
||||
|
||||
result, stderr, request = self.run_client(
|
||||
input_bytes=input_bytes,
|
||||
reply={"ok": True},
|
||||
runtime="claude",
|
||||
)
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertEqual(stderr, "")
|
||||
self.assertEqual(
|
||||
request.call_args.args[1]["latest_assistant_message"],
|
||||
"fallback message",
|
||||
)
|
||||
|
||||
def test_claude_present_invalid_inline_field_fails_without_fallback(self) -> None:
|
||||
fallback = mock.Mock(return_value="must not be used")
|
||||
input_bytes = json.dumps({
|
||||
"last_assistant_message": None,
|
||||
"transcript_path": "/unused/transcript.jsonl",
|
||||
}).encode()
|
||||
|
||||
with mock.patch.object(CLIENT, "claude_latest_entry", fallback):
|
||||
result, stderr, request = self.run_client(
|
||||
input_bytes=input_bytes,
|
||||
reply={"ok": True},
|
||||
runtime="claude",
|
||||
)
|
||||
|
||||
self.assertEqual(result, 2)
|
||||
self.assertIn("Mosaic receipt observer refused", stderr)
|
||||
request.assert_not_called()
|
||||
fallback.assert_not_called()
|
||||
|
||||
def test_claude_inline_message_size_guard_stays_enforced(self) -> None:
|
||||
request = mock.Mock(return_value={"ok": True})
|
||||
stderr = io.StringIO()
|
||||
with (
|
||||
mock.patch.object(CLIENT.sys, "stdin", io.BytesIO(b"{}")),
|
||||
mock.patch.object(
|
||||
CLIENT,
|
||||
"read_json",
|
||||
return_value={"last_assistant_message": "x" * (CLIENT.MAX_FRAME + 1)},
|
||||
),
|
||||
mock.patch.object(CLIENT, "observer_request", request),
|
||||
redirect_stderr(stderr),
|
||||
):
|
||||
result = CLIENT.main(
|
||||
["--runtime", "claude", "--latest-entry"],
|
||||
environ=ENVIRONMENT,
|
||||
)
|
||||
|
||||
self.assertEqual(result, 2)
|
||||
self.assertIn("Mosaic receipt observer refused", stderr.getvalue())
|
||||
request.assert_not_called()
|
||||
|
||||
def test_pi_still_posts_only_its_runtime_message(self) -> None:
|
||||
result, stderr, request = self.run_client(reply={"ok": True})
|
||||
|
||||
self.assertEqual(result, 0)
|
||||
self.assertEqual(stderr, "")
|
||||
self.assertEqual(
|
||||
request.call_args.args[1]["latest_assistant_message"],
|
||||
"ordinary turn",
|
||||
)
|
||||
|
||||
def test_nothing_pending_observation_refusal_is_benign(self) -> None:
|
||||
result, stderr, request = self.run_client(
|
||||
reply={"ok": False, "code": "OBSERVATION_UNAVAILABLE"}
|
||||
|
||||
Reference in New Issue
Block a user