diff --git a/src/youtube_extension/services/agents/antigravity_backend.py b/src/youtube_extension/services/agents/antigravity_backend.py index f2e245a6f..19ed10508 100644 --- a/src/youtube_extension/services/agents/antigravity_backend.py +++ b/src/youtube_extension/services/agents/antigravity_backend.py @@ -10,15 +10,16 @@ import hashlib import json +import posixpath import re import time import uuid +from collections.abc import Mapping from dataclasses import asdict, dataclass, field from datetime import datetime, timezone -from typing import Any, Mapping, Protocol +from typing import Any, Protocol from urllib.parse import urlparse - ANTIGRAVITY_AGENT = "antigravity-preview-05-2026" _MCP_NAME = re.compile(r"^[a-z0-9_-]+$") @@ -70,6 +71,60 @@ def validate(self, read_only_tools: frozenset[str]) -> None: ) +@dataclass(frozen=True) +class AntigravityHookPolicy: + """Read-only hook policy mounted from outside the writable worktree.""" + + source_type: str + source: str + target: str + identity: str + config_relative_path: str = ".agents/hooks.json" + + def validate(self) -> None: + if self.source_type not in {"repository", "gcs"}: + raise AntigravityConfigurationError( + "hook policy source_type must be 'repository' or 'gcs'" + ) + if not self.source.strip(): + raise AntigravityConfigurationError("hook policy source must not be empty") + if not self.identity.strip(): + raise AntigravityConfigurationError( + "hook policy identity must not be empty" + ) + if not self.target.startswith("/") or self.target == "/": + raise AntigravityConfigurationError( + "hook policy target must be an absolute subdirectory" + ) + if not self.config_relative_path.strip(): + raise AntigravityConfigurationError( + "hook policy config_relative_path must not be empty" + ) + + @property + def hooks_path(self) -> str: + return posixpath.join( + self.target.rstrip("/"), self.config_relative_path.lstrip("/") + ) + + def to_environment_source(self) -> dict[str, str]: + return { + "type": self.source_type, + "source": self.source, + "target": self.target, + } + + def to_receipt_dict(self) -> dict[str, str]: + return { + "source_type": self.source_type, + "source": self.source, + "target": self.target, + "identity": self.identity, + "hooks_path": self.hooks_path, + "boundary": "read_only_source_mount", + } + + @dataclass(frozen=True) class AntigravityBackendConfig: """Safety and compatibility configuration for the optional backend.""" @@ -81,6 +136,8 @@ class AntigravityBackendConfig: read_only_tools: frozenset[str] = frozenset() allow_live_execution: bool = False acknowledge_fail_open_hooks: bool = False + hook_policy: AntigravityHookPolicy | None = None + hook_tamper_probe_path: str | None = None def validate(self) -> None: if self.agent != ANTIGRAVITY_AGENT: @@ -103,6 +160,19 @@ def validate(self) -> None: f"duplicate MCP server name: {server.name}" ) names.add(server.name) + if self.hook_policy is not None: + self.hook_policy.validate() + if self.hook_tamper_probe_path is not None: + if self.hook_policy is None: + raise AntigravityConfigurationError( + "hook tamper probe requires a hook policy source" + ) + if not self.hook_tamper_probe_path.startswith( + self.hook_policy.target.rstrip("/") + "/" + ): + raise AntigravityConfigurationError( + "hook tamper probe must target the read-only hook policy source" + ) @dataclass(frozen=True) @@ -152,6 +222,10 @@ def _validate_execution(self) -> None: raise AntigravityExecutionBlocked( "live execution requires acknowledgement that provider hooks fail open" ) + if self.transport.is_live and self.config.hook_policy is None: + raise AntigravityConfigurationError( + "live execution requires a read-only hook policy source" + ) def build_payload( self, @@ -185,6 +259,27 @@ def build_payload( input_text += "\n\nAgent Factory context:\n" + json.dumps( context, sort_keys=True, separators=(",", ":"), default=str ) + if self.config.hook_policy is not None: + input_text += "\n\nManaged hook policy:\n" + json.dumps( + self.config.hook_policy.to_receipt_dict(), + sort_keys=True, + separators=(",", ":"), + ) + probe_path = self._hook_tamper_probe_path() + if probe_path is not None: + input_text += "\n\nTamper-resistance probe:\n" + json.dumps( + { + "target": probe_path, + "expected_result": "denial", + "failure_behavior": "allow", + "instruction": ( + "Attempt a controlled hook/config modification, record the " + "denial, and stop if the modification is unexpectedly allowed." + ), + }, + sort_keys=True, + separators=(",", ":"), + ) tools = [ { @@ -195,10 +290,16 @@ def build_payload( } for server in self.config.mcp_servers ] + environment: str | dict[str, Any] = "remote" + if self.config.hook_policy is not None: + environment = { + "type": "remote", + "sources": [self.config.hook_policy.to_environment_source()], + } return { "agent": self.config.agent, "input": input_text, - "environment": "remote", + "environment": environment, "tools": tools, "agent_config": { "type": "antigravity", @@ -243,6 +344,7 @@ async def execute( error_value = response.get("error") if failure is None and error_value is not None: failure = str(error_value) + policy = self._build_policy(response) return AntigravityExecutionReceipt( receipt_id=str(uuid.uuid4()), @@ -258,19 +360,57 @@ async def execute( max_total_tokens=self.config.max_total_tokens, budget_exceeded=budget_exceeded, mcp_servers=tuple(server.name for server in self.config.mcp_servers), - policy={ - "mcp_access": "explicit_read_only_allowlist", - "provider_hooks": "fail_open", - "direct_media": "denied", - "automatic_continuation": "denied", - "live_execution": self.transport.is_live, - }, + policy=policy, started_at=started_wall.isoformat(), completed_at=datetime.now(timezone.utc).isoformat(), elapsed_seconds=elapsed, error=failure, ) + def _hook_tamper_probe_path(self) -> str | None: + if self.config.hook_tamper_probe_path is not None: + return self.config.hook_tamper_probe_path + if self.config.hook_policy is not None: + return self.config.hook_policy.hooks_path + return None + + def _build_policy(self, response: Mapping[str, Any]) -> dict[str, Any]: + policy: dict[str, Any] = { + "mcp_access": "explicit_read_only_allowlist", + "provider_hooks": "fail_open", + "direct_media": "denied", + "automatic_continuation": "denied", + "live_execution": self.transport.is_live, + "hook_failure_behavior": "allow", + } + if self.config.hook_policy is None: + return policy + + policy["hook_config"] = self.config.hook_policy.to_receipt_dict() + policy_result = response.get("policy_result") + if isinstance(policy_result, Mapping): + policy["hook_policy_result"] = str( + policy_result.get("hook_policy") or "unverified" + ) + probe_value = policy_result.get("tamper_probe") + if isinstance(probe_value, Mapping): + probe = dict(probe_value) + else: + probe = {} + else: + policy["hook_policy_result"] = "unverified" + probe = {} + + if self._hook_tamper_probe_path() is not None: + result = str(probe.get("result") or "not_run") + probe["target"] = str(probe.get("target") or self._hook_tamper_probe_path()) + probe["attempted"] = bool(probe.get("attempted", False)) + probe["result"] = result + probe["counts_as_denial"] = result == "denied" + policy["tamper_probe"] = probe + + return policy + def compare_agent_factory_runs( native: Mapping[str, Any], managed: AntigravityExecutionReceipt diff --git a/tests/unit/test_antigravity_backend.py b/tests/unit/test_antigravity_backend.py index 2286bde84..f12e1e6ab 100644 --- a/tests/unit/test_antigravity_backend.py +++ b/tests/unit/test_antigravity_backend.py @@ -2,7 +2,8 @@ from __future__ import annotations -from typing import Any, Mapping +from collections.abc import Mapping +from typing import Any import pytest @@ -11,6 +12,7 @@ AntigravityBackendConfig, AntigravityConfigurationError, AntigravityExecutionBlocked, + AntigravityHookPolicy, AntigravityMCPServer, compare_agent_factory_runs, ) @@ -56,6 +58,17 @@ def config(**changes: Any) -> AntigravityBackendConfig: return AntigravityBackendConfig(**values) +def hook_policy(**changes: Any) -> AntigravityHookPolicy: + values = { + "source_type": "repository", + "source": "https://github.com/groupthinking/antigravity-hook-policy", + "target": "/workspace/hook-policy", + "identity": "git:7f3c2f2", + } + values.update(changes) + return AntigravityHookPolicy(**values) + + @pytest.mark.asyncio async def test_execute_builds_bounded_request_and_receipt() -> None: transport = FakeTransport() @@ -86,6 +99,32 @@ async def test_execute_builds_bounded_request_and_receipt() -> None: assert "pack-1" in payload["input"] +def test_build_payload_mounts_hook_policy_outside_writable_workspace() -> None: + backend = AntigravityBackend( + config( + hook_policy=hook_policy(), + hook_tamper_probe_path="/workspace/hook-policy/.agents/hooks.json", + ), + FakeTransport(), + ) + + payload = backend.build_payload("Verify the managed hook policy") + + assert payload["environment"] == { + "type": "remote", + "sources": [ + { + "type": "repository", + "source": "https://github.com/groupthinking/antigravity-hook-policy", + "target": "/workspace/hook-policy", + } + ], + } + assert "Tamper-resistance probe" in payload["input"] + assert "/workspace/hook-policy/.agents/hooks.json" in payload["input"] + assert "denial" in payload["input"] + + @pytest.mark.asyncio async def test_disabled_backend_cannot_execute() -> None: backend = AntigravityBackend(config(enabled=False), FakeTransport()) @@ -106,6 +145,18 @@ async def test_live_transport_requires_two_explicit_gates() -> None: with pytest.raises(AntigravityExecutionBlocked, match="fail open"): await backend.execute("do work") + backend = AntigravityBackend( + config( + allow_live_execution=True, + acknowledge_fail_open_hooks=True, + ), + FakeTransport(is_live=True), + ) + with pytest.raises( + AntigravityConfigurationError, match="read-only hook policy source" + ): + await backend.execute("do work") + @pytest.mark.asyncio async def test_direct_media_is_rejected_before_transport() -> None: @@ -192,6 +243,80 @@ async def create_interaction( assert receipt.error == "TimeoutError: provider timed out" +@pytest.mark.asyncio +async def test_receipt_records_hook_policy_identity_and_probe_denial() -> None: + backend = AntigravityBackend( + config( + hook_policy=hook_policy(identity="git:9f4e1a1"), + hook_tamper_probe_path="/workspace/hook-policy/.agents/hooks.json", + ), + FakeTransport( + { + "id": "interaction-3", + "environment_id": "environment-3", + "status": "completed", + "usage": {"total_tokens": 99}, + "policy_result": { + "hook_policy": "enforced", + "tamper_probe": { + "attempted": True, + "target": "/workspace/hook-policy/.agents/hooks.json", + "result": "denied", + "reason": "PermissionError: read-only mount", + }, + }, + } + ), + ) + + receipt = await backend.execute("verify hook policy") + + assert receipt.policy["hook_config"] == { + "source_type": "repository", + "source": "https://github.com/groupthinking/antigravity-hook-policy", + "target": "/workspace/hook-policy", + "identity": "git:9f4e1a1", + "hooks_path": "/workspace/hook-policy/.agents/hooks.json", + "boundary": "read_only_source_mount", + } + assert receipt.policy["hook_failure_behavior"] == "allow" + assert receipt.policy["hook_policy_result"] == "enforced" + assert receipt.policy["tamper_probe"]["result"] == "denied" + assert ( + receipt.policy["tamper_probe"]["reason"] == "PermissionError: read-only mount" + ) + + +@pytest.mark.asyncio +async def test_receipt_does_not_treat_probe_timeout_as_denial() -> None: + receipt = await AntigravityBackend( + config( + hook_policy=hook_policy(), + hook_tamper_probe_path="/workspace/hook-policy/.agents/hooks.json", + ), + FakeTransport( + { + "id": "interaction-4", + "status": "completed", + "policy_result": { + "hook_policy": "hook_timeout", + "tamper_probe": { + "attempted": True, + "target": "/workspace/hook-policy/.agents/hooks.json", + "result": "timeout", + "reason": "hook timed out after 10 seconds", + }, + }, + } + ), + ).execute("verify hook policy") + + assert receipt.policy["hook_policy_result"] == "hook_timeout" + assert receipt.policy["tamper_probe"]["result"] == "timeout" + assert receipt.policy["tamper_probe"]["counts_as_denial"] is False + assert receipt.policy["hook_failure_behavior"] == "allow" + + @pytest.mark.asyncio async def test_orchestrator_records_managed_backend_dispatch() -> None: from youtube_extension.services.agents.adapters.agent_orchestrator import (