fix(manager): configure external delivery targets safely

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.

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}

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

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())