feat(agentd): enforce the consent-plane terminal-surface envelope (enforce-sweep)

Extends the enforce-sweep (after goose voice + netwatch) to turtle-agentd, the
terminal-surface tool-execution host. An AUTONOMOUS AGENT acting on the terminal
surface may read/edit/test (discover/implement/verify) but may NOT egress
(push/publish/network) or operate (deploy/infra) — the consent-plane surface
envelope (deny_purposes=[egress,operate]) now holds at runtime.

- consent_plane_check(command, actor_id): denies egress/operate commands for
  agent actors; humans keep full authority (envelope governs agents only).
  _is_agent_actor is fail-closed — only an explicit `human:` id gets authority,
  every other non-empty id is governed.
- classification is local (regex over network/publish/deploy/infra commands) so
  containment holds even when the consent engine is unreachable.
- wired into policy_evaluate: a consent-plane deny short-circuits before Policy
  Fabric (defence in depth).
- 5 tests: agent egress denied, agent read/edit/test allowed, human unrestricted,
  policy_evaluate short-circuits, fail-closed actor detection. Daemon smoke green.

Author: Michael Heller

assets/sourceos/bin/turtle-agentd

@@ -700,6 +700,70 @@ def _local_policy_check(command: str) -> tuple[str, str, str | None]:
     return "allow", "No local policy rules matched.", None
 
 
+# ---------------------------------------------------------------------------
+# Consent-plane terminal-surface envelope (enforce-sweep)
+# ---------------------------------------------------------------------------
+# The consent-plane surface catalog gives the `terminal` surface
+# purposes=[discover, implement, verify] and deny_purposes=[egress, operate].
+# That envelope contains an AUTONOMOUS AGENT: a prompt-injected / runaway agent
+# acting on the terminal surface may read, edit, and test, but may NOT egress
+# (push/publish/network-send) or operate (deploy/infra). It does NOT restrict the
+# human operator, who holds full authority — so this check applies only to agent
+# actors. The classification is local so containment holds even when the consent
+# engine is unreachable (fail-closed for the agent); a reachable engine adds the
+# full role x space x consent check on top.
+
+_EGRESS_OPERATE_PATTERNS: list[re.Pattern[str]] = [
+    re.compile(p, re.IGNORECASE) for p in (
+        r"\bgit\s+push\b", r"\bgit\s+(fetch|pull|clone)\b",
+        r"\bgh\s+(pr\s+merge|release\s+create|repo\s+create|api\b.*-X\s*(POST|PUT|PATCH|DELETE))",
+        r"\b(curl|wget)\b", r"\bscp\b", r"\brsync\b.*::|\brsync\b.*@", r"\bssh\b\s+\S+@",
+        r"\bkubectl\s+(apply|create|delete|edit|patch|scale|rollout|drain|cordon)",
+        r"\bdocker\s+(push|run|-)", r"\bhelm\s+(install|upgrade|uninstall)",
+        r"\bterraform\s+(apply|destroy)", r"\bsystemctl\s+(start|stop|restart|enable|disable)",
+        r"\b(aws|gcloud|az)\s+\S+\s+(create|delete|update|deploy|put|set)",
+        r"\bnpm\s+publish\b", r"\bpip\s+(upload|install)\b", r"\bcargo\s+publish\b",
+        r"\bnc\b\s+\S+\s+\d+", r"\b(dd|mkfs|fdisk)\b",
+    )
+]
+
+
+def _is_agent_actor(actor_id: str | None) -> bool:
+    """The consent-plane envelope governs autonomous agents, not the human.
+    Only an explicit 'human:' actor keeps full authority; every other non-empty
+    id (agent:*, urn:srcos:agent:*, bot:*, svc:*, or an unrecognized id) is
+    governed — fail-closed, so an unknown actor cannot slip past as human.
+    (turtle-agentd defaults actor_id to 'human:local-user' at its call sites, so
+    an empty id here means the daemon's own default human path.)"""
+    a = (actor_id or "").lower()
+    if not a or a.startswith("human:"):
+        return False
+    return True
+
+
+def _terminal_purpose(command: str) -> str | None:
+    """Return 'egress-or-operate' when the command leaves the surface (network /
+    publish / deploy / infra mutation); None for read/edit/test (discover/
+    implement/verify), which the terminal surface allows."""
+    for pat in _EGRESS_OPERATE_PATTERNS:
+        if pat.search(command):
+            return "egress-or-operate"
+    return None
+
+
+def consent_plane_check(command: str, actor_id: str | None) -> tuple[str, str]:
+    """Apply the terminal-surface envelope to an agent actor's command.
+    Returns (decision, reason). Deny wins over Policy Fabric (defence in depth)."""
+    if not _is_agent_actor(actor_id):
+        return "allow", "human actor — consent-plane envelope governs agents only"
+    if _terminal_purpose(command):
+        return ("deny",
+                "[consent-plane] terminal surface denies egress/operate for an agent actor "
+                "(role x surface x space containment); a network/publish/deploy action is not "
+                "admissible on the terminal surface. Escalate to a human operator.")
+    return "allow", "[consent-plane] discover/implement/verify admissible on terminal surface"
+
+
 # ---------------------------------------------------------------------------
 # Policy Fabric wire (Track D)
 # ---------------------------------------------------------------------------
@@ -730,6 +794,16 @@ def policy_evaluate(
             "source": "local-policy",
         }
 
+    # Consent-plane terminal-surface envelope: an agent actor may not egress/operate
+    # on the terminal surface. Deny short-circuits (defence in depth, holds offline).
+    cp_decision, cp_reason = consent_plane_check(command, actor_id)
+    if cp_decision == "deny":
+        return {
+            "decision": {"outcome": "deny", "reason": cp_reason, "matched_rule": "consent-plane:terminal"},
+            "decision_id": None,
+            "source": "consent-plane",
+        }
+
     if risk_level is None:
         # host is high-risk, isolated domains are lower
         risk_level = "high" if execution_domain == "host" else "medium"

assets/sourceos/tests/test_turtle_agentd.py

@@ -80,3 +80,55 @@ def main() -> int:
 
 if __name__ == "__main__":
     raise SystemExit(main())
+
+
+# --------------------------------------------------------------------- consent-plane
+def _agentd_module():
+    """Import turtle-agentd (extensionless) to unit-test the consent-plane envelope."""
+    from importlib.machinery import SourceFileLoader
+    import importlib.util
+    loader = SourceFileLoader("turtle_agentd_mod", str(AGENTD))
+    spec = importlib.util.spec_from_loader("turtle_agentd_mod", loader)
+    mod = importlib.util.module_from_spec(spec)
+    loader.exec_module(mod)
+    return mod
+
+
+_ad = _agentd_module()
+
+
+def test_agent_actor_detection():
+    assert _ad._is_agent_actor("urn:srcos:agent:turtle-copilot")
+    assert _ad._is_agent_actor("agent:autonomous")
+    assert _ad._is_agent_actor("mystery:actor")  # unknown -> governed (fail-closed)
+    assert not _ad._is_agent_actor("human:local-user")  # only explicit human has authority
+    assert not _ad._is_agent_actor("")  # empty == daemon's default human path
+
+
+def test_agent_egress_denied_on_terminal_surface():
+    for cmd in ("git push origin main", "kubectl apply -f x.yaml", "curl https://evil.example",
+                "gh pr merge 12", "docker push repo/img", "scp f user@host:/tmp"):
+        d, reason = _ad.consent_plane_check(cmd, "agent:autonomous")
+        assert d == "deny", cmd
+        assert "consent-plane" in reason
+
+
+def test_agent_read_edit_test_allowed_on_terminal():
+    for cmd in ("ls -la", "cat README.md", "grep -r foo .", "git status", "git diff",
+                "pytest -q", "cargo test"):
+        d, _ = _ad.consent_plane_check(cmd, "agent:autonomous")
+        assert d == "allow", cmd
+
+
+def test_human_keeps_full_authority():
+    # the envelope governs agents only — a human may push/deploy
+    d, _ = _ad.consent_plane_check("git push origin main", "human:local-user")
+    assert d == "allow"
+
+
+def test_policy_evaluate_denies_agent_egress():
+    # deny short-circuits before Policy Fabric, offline
+    r = _ad.policy_evaluate("terminal.execute_command", "git push origin main",
+                            execution_domain="host", actor_id="agent:autonomous")
+    assert r["decision"]["outcome"] == "deny"
+    assert r["source"] == "consent-plane"