diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md index 76dffcdf37..6d3aac567b 100644 --- a/loopx/capabilities/manager_context/README.md +++ b/loopx/capabilities/manager_context/README.md @@ -16,6 +16,22 @@ External channels need an owner-configured grant in Use the actual connection channel and provider sender identity. Keep this file private (0600); do not commit it. Missing grants disable external delivery. +For an existing channel with an authorized sender, use the local operator CLI +to preview, grant, or revoke one registered recipient without editing the +policy file by hand: + +```sh +loopx manager-inbox grant-delivery-target --channel-id manager.external.0123456789abcdef01234567 --goal-id research --agent-id worker +loopx manager-inbox grant-delivery-target --channel-id manager.external.0123456789abcdef01234567 --goal-id research --agent-id worker --execute +loopx manager-inbox revoke-delivery-target --channel-id manager.external.0123456789abcdef01234567 --goal-id research --agent-id worker --execute +``` + +Pass the same `--registry` and `--runtime-root` used by the manager connection. +Without `--execute`, these commands only preview the target and count change. +Grant requires an active registered Goal and Agent, an existing sender-bound +channel, and membership in any explicit audience Goal read scope. The command +does not create a sender grant, launch the Agent, or grant protected-operation +authority. Revocation also works when the former Agent is no longer registered. Remove a source/target grant to revoke future delivery, including replay attempts. Provider ingress receipts bind the current message digest, channel and sender; a model cannot create that provenance through its response. diff --git a/loopx/capabilities/manager_context/__init__.py b/loopx/capabilities/manager_context/__init__.py index 8a9fe8013c..de14582e89 100644 --- a/loopx/capabilities/manager_context/__init__.py +++ b/loopx/capabilities/manager_context/__init__.py @@ -291,3 +291,103 @@ def configure_evidence_scope(runtime_root: Path, registry_path: Path, *, channel return {"ok": True, "executed": execute, "channel_id": channel, "evidence_goal_ids": ids, "scope": "audience_goal_summaries", "delegation_authority_changed": False} + + +def configure_delivery_target( + runtime_root: Path, + registry_path: Path, + *, + channel: str, + goal_id: str, + agent_id: str, + grant: bool, + execute: bool = False, +) -> dict: + """Preview or change one sender-bound recipient on an existing external channel.""" + if not re.fullmatch(r"manager\.external\.[a-f0-9]{24}", channel): + raise ValueError("an exact external manager channel is required") + if not goal_id or not agent_id: + raise ValueError("an exact Goal and Agent are required") + target = {"goal_id": goal_id, "agent_id": agent_id} + + def is_target(item: dict) -> bool: + return item.get("goal_id") == goal_id and item.get("agent_id") == agent_id + + if grant: + registry = load_registry(registry_path) + goal = next( + (g for g in registry.get("goals", []) if isinstance(g, dict) and g.get("id") == goal_id), + None, + ) + if ( + goal is None + or goal_is_stopped(goal) + or agent_id not in registered_agent_ids_for_goal(goal) + ): + raise ValueError("delivery target must be a registered Agent in an active Goal") + + path = _root(runtime_root) / "policy.json" + + def update() -> dict: + policy = _read(path) + if policy.get("schema_version") != POLICY_SCHEMA or not isinstance( + policy.get("sources"), dict + ): + raise ValueError("invalid manager policy") + source = policy["sources"].get(channel) + if not isinstance(source, dict): + raise ValueError("external manager channel must already be configured") + senders = source.get("sender_ids") + if grant and ( + not isinstance(senders, list) + or not senders + or any(not isinstance(sender, str) or not sender for sender in senders) + ): + raise ValueError("external manager channel has no valid sender grant") + if ( + grant + and "evidence_goal_ids" in source + and goal_id not in (evidence_goal_scope(runtime_root, channel) or []) + ): + raise ValueError("target Goal is outside the channel read scope") + targets = source.get("targets", []) + if not isinstance(targets, list) or any( + not isinstance(item, dict) + or not isinstance(item.get("goal_id"), str) + or not isinstance(item.get("agent_id"), str) + for item in targets + ): + raise ValueError("invalid external manager delivery targets") + before = any(is_target(item) for item in targets) + if grant: + updated_targets = targets if before else [*targets, target] + else: + updated_targets = [item for item in targets if not is_target(item)] + changed = updated_targets != targets + if execute and changed: + source["targets"] = updated_targets + _write(path, policy) + return { + "ok": True, + "executed": execute, + "changed": changed if execute else False, + "would_change": changed, + "channel_id": channel, + "target": target, + "granted_before": before, + "granted_after": grant, + "existing_target_count": len(targets), + "resulting_target_count": len(updated_targets), + "scope": "sender_bound_context_delivery", + "execution_started": False, + } + + if not execute: + return update() + with exclusive_file_lock(path.with_suffix(".lock")): + result = update() + saved = _read(path) + saved_targets = saved.get("sources", {}).get(channel, {}).get("targets", []) + if any(is_target(item) for item in saved_targets) != grant: + raise ValueError("delivery target verification failed") + return {**result, "readback_verified": True} diff --git a/loopx/cli_commands/manager_inbox.py b/loopx/cli_commands/manager_inbox.py index d2a3918d0f..836e54ead7 100644 --- a/loopx/cli_commands/manager_inbox.py +++ b/loopx/cli_commands/manager_inbox.py @@ -6,6 +6,7 @@ from ..history import load_registry from ..capabilities.manager_context import ( acknowledge, + configure_delivery_target, configure_evidence_scope, ) @@ -28,6 +29,8 @@ def register_manager_inbox(subparsers, add_format): "status", "configure-read-scope", "configure-ssh-read-scope", + "grant-delivery-target", + "revoke-delivery-target", ), ) parser.add_argument("--peer-agent-id", help="For request: a registered peer of the same Goal.") @@ -75,6 +78,18 @@ def handle_manager_inbox(args, registry_path, runtime_root): ) print(json.dumps(result, ensure_ascii=False, indent=2)) return 0 + if args.manager_inbox_action in {"grant-delivery-target", "revoke-delivery-target"}: + result = configure_delivery_target( + runtime_root, + registry_path, + channel=args.channel_id or "", + goal_id=args.goal_id or "", + agent_id=args.agent_id or "", + grant=args.manager_inbox_action == "grant-delivery-target", + execute=args.execute, + ) + print(json.dumps(result, ensure_ascii=False, indent=2)) + return 0 registry = load_registry(registry_path) goal = next( (g for g in registry.get("goals", []) if g.get("id") == args.goal_id), None diff --git a/tests/test_manager_context_handoff.py b/tests/test_manager_context_handoff.py index 3ccdb4095a..d657730c19 100644 --- a/tests/test_manager_context_handoff.py +++ b/tests/test_manager_context_handoff.py @@ -1,4 +1,6 @@ import json +import subprocess +import sys from concurrent.futures import ThreadPoolExecutor import pytest @@ -9,6 +11,7 @@ _write, acknowledge, authority, + configure_delivery_target, deliver, pending, register_ingress, @@ -226,6 +229,130 @@ def test_external_authority_requires_exact_sender_source_and_recipient(fixture): assert receipt["status"] == "delivered" +def test_operator_delivery_target_preview_grant_revoke_and_live_authority(fixture): + root, registry, session, turn, request = fixture + channel = "manager.external." + "a" * 24 + session["channel_id"] = channel + turn["origin"] = "lark" + other = {"goal_id": "other", "agent_id": "peer"} + policy_path = _root(root) / "policy.json" + _write(policy_path, { + "schema_version": POLICY_SCHEMA, + "sources": {channel: { + "sender_ids": ["owner"], "targets": [other], + "evidence_goal_ids": ["research", "other"], + "evidence_ssh_hosts": {"example-host": ["research"]}, + }}, + }) + before = policy_path.read_bytes() + preview = configure_delivery_target( + root, registry, channel=channel, **request, grant=True + ) + assert preview["would_change"] and not preview["executed"] + assert preview["resulting_target_count"] == 2 + assert policy_path.read_bytes() == before + + register_ingress( + root, session_id=session["session_id"], client_turn_id=turn["client_turn_id"], + channel=channel, sender_id="owner", message=turn["message"], + source_id="lark:original", + ) + assert authority(root, registry, session, turn)["targets"] == [other] + applied = configure_delivery_target( + root, registry, channel=channel, **request, grant=True, execute=True + ) + assert applied["changed"] and applied["granted_after"] and applied["readback_verified"] + assert authority(root, registry, session, turn)["targets"] == [other, request] + assert not configure_delivery_target( + root, registry, channel=channel, **request, grant=True, execute=True + )["changed"] + saved = json.loads(policy_path.read_text()) + assert saved["sources"][channel]["sender_ids"] == ["owner"] + assert saved["sources"][channel]["evidence_ssh_hosts"] == {"example-host": ["research"]} + + revoked = configure_delivery_target( + root, registry, channel=channel, **request, grant=False, execute=True + ) + assert revoked["changed"] and not revoked["granted_after"] and revoked["readback_verified"] + assert authority(root, registry, session, turn)["targets"] == [other] + assert not configure_delivery_target( + root, registry, channel=channel, **request, grant=False, execute=True + )["changed"] + + # Older policy rows may carry metadata; recipient identity is still the pair. + saved = json.loads(policy_path.read_text()) + saved["sources"][channel]["targets"] = [other, {**request, "note": "legacy"}, request] + _write(policy_path, saved) + assert not configure_delivery_target( + root, registry, channel=channel, **request, grant=True, execute=True + )["changed"] + assert authority(root, registry, session, turn)["targets"] == [other, request] + assert configure_delivery_target( + root, registry, channel=channel, **request, grant=False, execute=True + )["readback_verified"] + assert json.loads(policy_path.read_text())["sources"][channel]["targets"] == [other] + + +def test_operator_target_grant_fails_closed_without_audited_source_or_agent(fixture): + root, registry, _, _, request = fixture + channel = "manager.external." + "b" * 24 + with pytest.raises((OSError, ValueError)): + configure_delivery_target(root, registry, channel=channel, **request, grant=True, execute=True) + + policy_path = _root(root) / "policy.json" + source = {"sender_ids": ["owner"], "evidence_goal_ids": ["other"], "targets": []} + _write(policy_path, {"schema_version": POLICY_SCHEMA, "sources": {channel: source}}) + with pytest.raises(ValueError, match="outside the channel read scope"): + configure_delivery_target(root, registry, channel=channel, **request, grant=True, execute=True) + source["evidence_goal_ids"] = ["research"] + _write(policy_path, {"schema_version": POLICY_SCHEMA, "sources": {channel: source}}) + with pytest.raises(ValueError, match="registered Agent"): + configure_delivery_target(root, registry, channel=channel, goal_id="research", + agent_id="unknown", grant=True, execute=True) + source["evidence_goal_ids"] = ["research", "*"] + _write(policy_path, {"schema_version": POLICY_SCHEMA, "sources": {channel: source}}) + with pytest.raises(ValueError, match="outside the channel read scope"): + configure_delivery_target(root, registry, channel=channel, **request, grant=True, execute=True) + source["evidence_goal_ids"] = ["research"] + _write(policy_path, {"schema_version": POLICY_SCHEMA, "sources": {channel: source}}) + data = json.loads(registry.read_text()) + data["goals"][0]["activation_state"] = "stopped" + registry.write_text(json.dumps(data)) + with pytest.raises(ValueError, match="active Goal"): + configure_delivery_target(root, registry, channel=channel, **request, grant=True, execute=True) + assert json.loads(policy_path.read_text())["sources"][channel]["targets"] == [] + + +def test_manager_inbox_cli_previews_and_applies_one_delivery_target(fixture): + root, registry, _, _, request = fixture + channel = "manager.external." + "c" * 24 + policy_path = _root(root) / "policy.json" + _write(policy_path, { + "schema_version": POLICY_SCHEMA, + "sources": {channel: {"sender_ids": ["owner"], "targets": []}}, + }) + base = [ + sys.executable, "-m", "loopx.cli", "--registry", str(registry), + "--runtime-root", str(root), "manager-inbox", + ] + options = ["--channel-id", channel, "--goal-id", request["goal_id"], + "--agent-id", request["agent_id"]] + + def call(action, execute=False): + completed = subprocess.run( + [*base, action, *options, *(["--execute"] if execute else [])], + capture_output=True, text=True, check=True, + ) + return json.loads(completed.stdout) + + original = policy_path.read_bytes() + assert call("grant-delivery-target")["would_change"] + assert policy_path.read_bytes() == original + assert call("grant-delivery-target", execute=True)["granted_after"] + assert call("revoke-delivery-target", execute=True)["granted_after"] is False + assert json.loads(policy_path.read_text())["sources"][channel]["targets"] == [] + + def test_same_goal_recipients_keep_inboxes_and_decisions_separate(fixture): root, registry, session, turn, request = fixture data = json.loads(registry.read_text())