| from __future__ import annotations | |
| from hai_relay.authority import DenyAllAuthority | |
| def task(recipient="gemini", mode="advisory", key="task"): | |
| return { | |
| "recipient": recipient, | |
| "body": "inspect one artifact", | |
| "idempotency_key": key, | |
| "kind": "task", | |
| "mission_id": "mis_1", | |
| "lease_id": "lease_1", | |
| "working_directory": "/tmp", | |
| "operation_mode": mode, | |
| } | |
| def test_task_dispatches_after_authority(service): | |
| result = service.send("codex", task())["message"] | |
| assert result["status"] == "completed" | |
| assert result["result"]["text"] == "gemini:ok" | |
| def test_authority_unavailable_fails_closed(service): | |
| service.authority = DenyAllAuthority() | |
| result = service.send("codex", task(key="blocked"))["message"] | |
| assert result["status"] == "blocked" | |
| assert result["result"]["code"] == "authority_denied" | |
| def test_state_change_requires_guarded_claude(service): | |
| service.authority.assurance = "guarded" | |
| wrong_target = service.send( | |
| "codex", task(recipient="gemini", mode="state_changing", key="wrong") | |
| )["message"] | |
| assert wrong_target["status"] == "blocked" | |
| assert wrong_target["result"]["code"] == "write_denied" | |
| allowed = service.send( | |
| "codex", task(recipient="claude", mode="state_changing", key="allowed") | |
| )["message"] | |
| assert allowed["status"] == "completed" | |
| def test_denied_authority_does_not_dispatch(service): | |
| service.authority.allowed = False | |
| result = service.send("codex", task(key="denied"))["message"] | |
| assert result["status"] == "blocked" | |
| assert not service.dispatcher.calls | |
| def test_cancel_pending_task(service): | |
| service.authority.allowed = False | |
| message = service.send("codex", task(key="cancel"))["message"] | |
| cancelled = service.cancel_task("codex", message["message_id"])["message"] | |
| assert cancelled["status"] == "cancelled" | |