Test Antigravity preview as an optional Agent Factory backend

src/youtube_extension/services/agents/antigravity_backend.py

@@ -13,14 +13,23 @@
 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_-]+$")
+EVENTRELAY_MCP_SERVER = "eventrelay"
+PROVIDER_INDEPENDENT_CONTROL_PLANE = (
+    "approval_state",
+    "durable_receipts",
+    "portable_orchestration_contracts",
+    "provider_routing",
+    "provenance",
+    "replay",
+)
+_MCP_NAME = re.compile(r"^[a-z0-9]+$")
 
 
 class AntigravityConfigurationError(ValueError):
@@ -51,7 +60,7 @@ class AntigravityMCPServer:
     def validate(self, read_only_tools: frozenset[str]) -> None:
         if not _MCP_NAME.fullmatch(self.name):
             raise AntigravityConfigurationError(
-                "MCP server name must match ^[a-z0-9_-]+$"
+                "MCP server name must match ^[a-z0-9]+$"
             )
         parsed = urlparse(self.url)
         if parsed.scheme != "https" or not parsed.netloc:
@@ -91,9 +100,9 @@ def validate(self) -> None:
             raise AntigravityConfigurationError(
                 "max_total_tokens must be between 1 and 1,000,000"
             )
-        if not self.mcp_servers:
+        if len(self.mcp_servers) != 1:
             raise AntigravityConfigurationError(
-                "at least one read-only MCP server is required"
+                "exactly one read-only EventRelay MCP server is required"
             )
         names: set[str] = set()
         for server in self.mcp_servers:
@@ -103,6 +112,10 @@ def validate(self) -> None:
                     f"duplicate MCP server name: {server.name}"
                 )
             names.add(server.name)
+        if EVENTRELAY_MCP_SERVER not in names:
+            raise AntigravityConfigurationError(
+                "the configured MCP server must be named 'eventrelay'"
+            )
 
 
 @dataclass(frozen=True)
@@ -180,25 +193,34 @@ def build_payload(
                 + ", ".join(present)
             )
 
-        input_text = task
+        input_text = (
+            "Execute only this bounded build/test task for EventRelay. "
+            "Use the read-only EventRelay MCP server for receipts/provenance lookups. "
+            "Do not ingest direct media, start background work, or perform undeclared "
+            "code/filesystem side effects.\n\nTask:\n"
+            f"{task}"
+        )
         if context:
             input_text += "\n\nAgent Factory context:\n" + json.dumps(
                 context, sort_keys=True, separators=(",", ":"), default=str
             )
 
-        tools = [
-            {
-                "type": "mcp_server",
-                "name": server.name,
-                "url": server.url,
-                "allowed_tools": list(server.allowed_tools),
-            }
-            for server in self.config.mcp_servers
-        ]
+        tools = [{"type": "code_execution"}]
+        tools.extend(
+            [
+                {
+                    "type": "mcp_server",
+                    "name": server.name,
+                    "url": server.url,
+                    "allowed_tools": list(server.allowed_tools),
+                }
+                for server in self.config.mcp_servers
+            ]
+        )
         return {
             "agent": self.config.agent,
             "input": input_text,
-            "environment": "remote",
+            "environment": self._build_environment(),
             "tools": tools,
             "agent_config": {
                 "type": "antigravity",
@@ -208,6 +230,91 @@ def build_payload(
             "store": True,
         }
 
+    def _build_environment(self) -> dict[str, Any]:
+        hooks = {
+            "side-effect-gate": {
+                "enabled": True,
+                "pre_tool_execution": [
+                    {
+                        "matcher": "code_execution",
+                        "hooks": [
+                            {
+                                "type": "command",
+                                "command": "python3 /.agents/hooks-scripts/policy_gate.py",
+                                "timeout": 10,
+                            }
+                        ],
+                    },
+                    {
+                        "matcher": "write_file|delete_file",
+                        "hooks": [
+                            {
+                                "type": "command",
+                                "command": "python3 /.agents/hooks-scripts/policy_gate.py",
+                                "timeout": 10,
+                            }
+                        ],
+                    },
+                ],
+            }
+        }
+        gate_script = """#!/usr/bin/env python3
+import json
+import sys
+
+payload = json.load(sys.stdin)
+tool_call = payload.get("tool_call", {})
+name = str(tool_call.get("name", ""))
+args = tool_call.get("args") or {}
+command = str(args.get("code", ""))
+allowed_commands = {
+    "python -m pytest",
+    "pytest",
+    "npm test",
+    "npm run build",
+    "turbo run test",
+    "turbo run build",
+}
+# Any shell metacharacter can chain undeclared side effects onto an
+# otherwise-allowed command, so reject them outright and require the
+# command to exactly match a vetted entry (a prefix check is bypassable
+# via `pytest && curl ... | sh`, `pytest; rm -rf /`, etc.).
+forbidden_tokens = (";", "&", "|", "`", "$(", "${", ">", "<", "\\n", "\\r", "(", ")", "\\\\")
+
+def _is_allowed(cmd):
+    if any(token in cmd for token in forbidden_tokens):
+        return False
+    return cmd.strip() in allowed_commands
+
+if name in {"write_file", "delete_file"}:
+    print(json.dumps({
+        "decision": "deny",
+        "reason": "Filesystem side effects must be declared in EventRelay receipts first."
+    }))
+elif name == "code_execution" and not _is_allowed(command):
+    print(json.dumps({
+        "decision": "deny",
+        "reason": "Only bounded build/test commands are allowed for the Antigravity spike."
+    }))
+else:
+    print(json.dumps({"decision": "allow"}))
+"""
+        return {
+            "type": "remote",
+            "sources": [
+                {
+                    "type": "inline",
+                    "target": ".agents/hooks.json",
+                    "content": json.dumps(hooks, sort_keys=True, separators=(",", ":")),
+                },
+                {
+                    "type": "inline",
+                    "target": ".agents/hooks-scripts/policy_gate.py",
+                    "content": gate_script,
+                },
+            ],
+        }
+
     async def execute(
         self,
         task: str,
@@ -260,7 +367,8 @@ async def execute(
             mcp_servers=tuple(server.name for server in self.config.mcp_servers),
             policy={
                 "mcp_access": "explicit_read_only_allowlist",
-                "provider_hooks": "fail_open",
+                "provider_hooks": "fail_open_acknowledged",
+                "synchronous_hooks": "inline_pre_tool_execution_guard",
                 "direct_media": "denied",
                 "automatic_continuation": "denied",
                 "live_execution": self.transport.is_live,
@@ -277,20 +385,59 @@ def compare_agent_factory_runs(
 ) -> dict[str, Any]:
     """Return a small provider-neutral comparison artifact for evaluation."""
     native_success = bool(native.get("success", native.get("status") == "ok"))
+    managed_success = managed.status == "completed" and managed.error is None
+    artifact_determinism = {
+        "native_sha256": native.get("artifact_sha256"),
+        "antigravity_request_sha256": managed.request_sha256,
+        "native_output_present": bool(native.get("output") or native.get("results")),
+        "antigravity_output_present": bool(managed.output_text),
+    }
+    provenance = {
+        "native": {
+            "receipt_id": native.get("receipt_id"),
+        },
+        "antigravity": {
+            "receipt_id": managed.receipt_id,
+            "interaction_id": managed.interaction_id,
+            "environment_id": managed.environment_id,
+        },
+    }
+    recovery = {
+        "native_retryable": native.get("retryable"),
+        "antigravity_can_resume": bool(
+            managed.interaction_id and managed.environment_id
+        ),
+    }
     return {
         "native": {
             "success": native_success,
             "elapsed_seconds": native.get("total_processing_time"),
-            "output_present": bool(native.get("output") or native.get("results")),
+            "output_present": artifact_determinism["native_output_present"],
         },
         "antigravity": {
-            "success": managed.status == "completed" and managed.error is None,
+            "success": managed_success,
             "elapsed_seconds": managed.elapsed_seconds,
             "total_tokens": managed.usage.get("total_tokens"),
             "budget_exceeded": managed.budget_exceeded,
-            "output_present": bool(managed.output_text),
+            "output_present": artifact_determinism["antigravity_output_present"],
             "receipt_id": managed.receipt_id,
         },
+        "completion": {"native": native_success, "antigravity": managed_success},
+        "latency_seconds": {
+            "native": native.get("total_processing_time"),
+            "antigravity": managed.elapsed_seconds,
+        },
+        "cost": {
+            "native_usd": native.get("cost_usd"),
+            "antigravity_usd": managed.usage.get("cost_usd")
+            or managed.usage.get("estimated_cost_usd"),
+        },
+        "artifact_determinism": artifact_determinism,
+        "provenance": provenance,
+        "recovery": recovery,
+        "control_plane": {
+            "provider_independent": list(PROVIDER_INDEPENDENT_CONTROL_PLANE)
+        },
     }
 
 

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
 
@@ -68,22 +69,35 @@ async def test_execute_builds_bounded_request_and_receipt() -> None:
     assert receipt.environment_id == "environment-1"
     assert receipt.usage == {"total_tokens": 321}
     assert receipt.policy["mcp_access"] == "explicit_read_only_allowlist"
-    assert receipt.policy["provider_hooks"] == "fail_open"
+    assert receipt.policy["provider_hooks"] == "fail_open_acknowledged"
+    assert receipt.policy["synchronous_hooks"] == "inline_pre_tool_execution_guard"
     assert len(receipt.request_sha256) == 64
 
     payload = transport.payloads[0]
     assert payload["agent"] == "antigravity-preview-05-2026"
     assert payload["agent_config"]["max_total_tokens"] == 1_000
     assert payload["tools"] == [
+        {
+            "type": "code_execution",
+        },
         {
             "type": "mcp_server",
             "name": "eventrelay",
             "url": "https://mcp.example.test/mcp",
             "allowed_tools": ["evidence_get"],
-        }
+        },
     ]
     assert "headers" not in payload["tools"][0]
+    assert payload["environment"]["type"] == "remote"
+    assert payload["environment"]["sources"][0]["target"] == ".agents/hooks.json"
+    assert "pre_tool_execution" in payload["environment"]["sources"][0]["content"]
+    assert "write_file|delete_file" in payload["environment"]["sources"][0]["content"]
+    assert (
+        payload["environment"]["sources"][1]["target"]
+        == ".agents/hooks-scripts/policy_gate.py"
+    )
     assert "pack-1" in payload["input"]
+    assert "bounded build/test task" in payload["input"]
 
 
 @pytest.mark.asyncio
@@ -117,7 +131,9 @@ async def test_direct_media_is_rejected_before_transport() -> None:
     assert transport.payloads == []
 
 
-@pytest.mark.parametrize("name", ["EventRelay", "event relay", "eventrelay!"])
+@pytest.mark.parametrize(
+    "name", ["EventRelay", "event relay", "eventrelay!", "event-relay", "event_relay"]
+)
 def test_mcp_name_must_match_provider_contract(name: str) -> None:
     invalid = config(
         mcp_servers=(
@@ -158,6 +174,26 @@ def test_mcp_transport_and_read_only_allowlist_are_enforced() -> None:
         undeclared.validate()
 
 
+def test_requires_exactly_one_eventrelay_mcp_server() -> None:
+    invalid = config(
+        mcp_servers=(
+            AntigravityMCPServer(
+                name="eventrelay",
+                url="https://mcp.example.test/mcp",
+                allowed_tools=("evidence_get",),
+            ),
+            AntigravityMCPServer(
+                name="shadow",
+                url="https://mcp-2.example.test/mcp",
+                allowed_tools=("evidence_get",),
+            ),
+        )
+    )
+
+    with pytest.raises(AntigravityConfigurationError, match="exactly one"):
+        invalid.validate()
+
+
 @pytest.mark.asyncio
 async def test_receipt_records_transport_failure_and_budget_overrun() -> None:
     backend = AntigravityBackend(
@@ -219,8 +255,35 @@ async def test_comparison_artifact_is_provider_neutral() -> None:
     receipt = await AntigravityBackend(config(), transport).execute("compare")
 
     comparison = compare_agent_factory_runs(
-        {"success": True, "total_processing_time": 0.5, "results": {"a": "b"}},
+        {
+            "success": True,
+            "total_processing_time": 0.5,
+            "results": {"a": "b"},
+            "cost_usd": 0.02,
+            "artifact_sha256": "native-hash",
+            "receipt_id": "native-receipt",
+            "retryable": False,
+        },
         receipt,
     )
-    assert comparison["native"]["success"] is True
-    assert comparison["antigravity"]["total_tokens"] == 321
+    assert comparison["completion"] == {"native": True, "antigravity": True}
+    assert comparison["latency_seconds"]["native"] == 0.5
+    assert comparison["latency_seconds"]["antigravity"] == pytest.approx(
+        receipt.elapsed_seconds
+    )
+    assert comparison["cost"] == {"native_usd": 0.02, "antigravity_usd": None}
+    assert comparison["artifact_determinism"]["native_sha256"] == "native-hash"
+    assert len(comparison["artifact_determinism"]["antigravity_request_sha256"]) == 64
+    assert comparison["provenance"]["antigravity"]["interaction_id"] == "interaction-1"
+    assert comparison["recovery"] == {
+        "native_retryable": False,
+        "antigravity_can_resume": True,
+    }
+    assert comparison["control_plane"]["provider_independent"] == [
+        "approval_state",
+        "durable_receipts",
+        "portable_orchestration_contracts",
+        "provider_routing",
+        "provenance",
+        "replay",
+    ]