Enforce fixture interaction policy

ampbrowser/cli.py

@@ -190,6 +190,11 @@ def _cmd_fixture(args) -> int:
                 f"actual_transport={check.actual_transport} "
                 f"expected_profile={check.expected_profile} "
                 f"actual_profile={check.actual_profile} "
+                f"tier={check.tier} "
+                f"identity={check.identity} "
+                f"payments={check.payments} "
+                f"realtime={str(check.realtime).lower()} "
+                f"public_allowed={str(check.public_allowed).lower()} "
                 f"status={check.status} "
                 f"message=\"{_safe(check.message)}\""
             )

ampbrowser/fixtures.py

@@ -5,6 +5,7 @@ import json
 from pathlib import Path
 from typing import Any
 
+from .interaction_policy import interaction_failures
 from .routing import route_url
 
 
@@ -20,6 +21,11 @@ class FixtureCheck:
     actual_transport: str
     expected_profile: str
     actual_profile: str
+    tier: str
+    identity: str
+    payments: str
+    realtime: bool
+    public_allowed: bool
     status: str
     message: str
 
@@ -70,6 +76,7 @@ def _check_fixture(site_id: str, fixture: Any, path: Path) -> FixtureCheck:
         raise ValueError(f"{path}: {protocol}: missing checks table")
     expected_transport = _string(checks, "transport")
     expected_profile = _string(checks, "profile")
+    interaction = _interaction_policy(fixture)
 
     route = route_url(url)
     failures: list[str] = []
@@ -77,6 +84,16 @@ def _check_fixture(site_id: str, fixture: Any, path: Path) -> FixtureCheck:
         failures.append(f"transport expected {expected_transport} got {route.transport}")
     if route.profile != expected_profile:
         failures.append(f"profile expected {expected_profile} got {route.profile}")
+    failures.extend(
+        interaction_failures(
+            expected_transport,
+            tier=interaction["tier"],
+            identity=interaction["identity"],
+            payments=interaction["payments"],
+            realtime=interaction["realtime"],
+            public_allowed=interaction["public_allowed"],
+        )
+    )
 
     return FixtureCheck(
         site_id=site_id,
@@ -86,6 +103,11 @@ def _check_fixture(site_id: str, fixture: Any, path: Path) -> FixtureCheck:
         actual_transport=route.transport,
         expected_profile=expected_profile,
         actual_profile=route.profile,
+        tier=interaction["tier"],
+        identity=interaction["identity"],
+        payments=interaction["payments"],
+        realtime=interaction["realtime"],
+        public_allowed=interaction["public_allowed"],
         status="fail" if failures else "ok",
         message="; ".join(failures) if failures else "route matches",
     )
@@ -103,3 +125,30 @@ def _string(data: dict[str, Any], key: str) -> str:
     if not isinstance(value, str) or not value:
         raise ValueError(f"missing required string field {key!r}")
     return value
+
+
+def _interaction_policy(fixture: dict[str, Any]) -> dict[str, Any]:
+    raw = fixture.get("interaction", {})
+    if not isinstance(raw, dict):
+        raise ValueError(f"{fixture.get('protocol', 'fixture')}: interaction must be a table")
+    return {
+        "tier": _optional_string(raw, "tier", "static"),
+        "identity": _optional_string(raw, "identity", "none"),
+        "payments": _optional_string(raw, "payments", "none"),
+        "realtime": _optional_bool(raw, "realtime", False),
+        "public_allowed": _optional_bool(raw, "public_allowed", True),
+    }
+
+
+def _optional_string(data: dict[str, Any], key: str, default: str) -> str:
+    value = data.get(key, default)
+    if not isinstance(value, str) or not value:
+        raise ValueError(f"interaction.{key} must be a non-empty string")
+    return value
+
+
+def _optional_bool(data: dict[str, Any], key: str, default: bool) -> bool:
+    value = data.get(key, default)
+    if not isinstance(value, bool):
+        raise ValueError(f"interaction.{key} must be true or false")
+    return value

ampbrowser/interaction_policy.py

@@ -0,0 +1,96 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+TIER_ORDER = {
+    "static": 0,
+    "interactive-lite": 1,
+    "identity": 2,
+    "transactional": 3,
+    "realtime": 4,
+    "internal": 5,
+}
+
+
+@dataclass(frozen=True)
+class TransportInteractionPolicy:
+    transport: str
+    max_tier: str
+    identities: tuple[str, ...]
+    payments: tuple[str, ...]
+    realtime: str
+
+
+TRANSPORT_POLICIES = {
+    "clearnet": TransportInteractionPolicy(
+        "clearnet",
+        "realtime",
+        ("none", "http-session", "siwe"),
+        ("none", "server-invoice", "wallet-signature"),
+        "yes",
+    ),
+    "tor": TransportInteractionPolicy(
+        "tor",
+        "transactional",
+        ("none", "http-session", "siwe"),
+        ("none", "server-invoice", "wallet-signature"),
+        "private-only",
+    ),
+    "i2p": TransportInteractionPolicy(
+        "i2p",
+        "transactional",
+        ("none", "http-session", "siwe"),
+        ("none", "server-invoice", "wallet-signature"),
+        "private-only",
+    ),
+    "gemini": TransportInteractionPolicy(
+        "gemini",
+        "interactive-lite",
+        ("none", "signed-link"),
+        ("none", "static-instructions"),
+        "no",
+    ),
+    "ipfs": TransportInteractionPolicy(
+        "ipfs",
+        "static",
+        ("none",),
+        ("none", "static-instructions"),
+        "no",
+    ),
+    "reticulum": TransportInteractionPolicy(
+        "reticulum",
+        "interactive-lite",
+        ("none", "signed-link"),
+        ("none", "static-instructions"),
+        "research",
+    ),
+}
+
+
+def interaction_failures(
+    transport: str,
+    *,
+    tier: str,
+    identity: str,
+    payments: str,
+    realtime: bool,
+    public_allowed: bool,
+) -> list[str]:
+    failures: list[str] = []
+    policy = TRANSPORT_POLICIES.get(transport)
+    if not policy:
+        return [f"no interaction policy for transport {transport}"]
+    if not public_allowed:
+        failures.append("fixture is not public_allowed")
+    if tier not in TIER_ORDER:
+        failures.append(f"unknown tier {tier}")
+    elif TIER_ORDER[tier] > TIER_ORDER[policy.max_tier]:
+        failures.append(f"tier {tier} exceeds {transport} max {policy.max_tier}")
+    if identity not in policy.identities:
+        failures.append(f"identity {identity} not allowed on {transport}")
+    if payments not in policy.payments:
+        failures.append(f"payments {payments} not allowed on {transport}")
+    if realtime and policy.realtime not in {"yes", "private-only", "research"}:
+        failures.append(f"realtime not allowed on {transport}")
+    return failures

docs/fixture-manifests.md

@@ -3,7 +3,8 @@
 > Status: draft | Updated 2026-07-06 | Applies to: AMPB and AMPG compatibility checks
 
 AMPB can check AMPG fixture manifests without contacting any network. The check confirms
-that each generated fixture URL routes to the expected AMPB transport and profile.
+that each generated fixture URL routes to the expected AMPB transport and profile, and
+that its declared interaction policy fits that transport.
 
 ## Command
 
@@ -11,12 +12,13 @@ that each generated fixture URL routes to the expected AMPB transport and profil
 python3 -m ampbrowser fixture check ../adaptive-multi-protocol-gateway/dist/wownero/ampg-fixture-manifest.json
 ```
 
-The command exits non-zero when any fixture route does not match the manifest.
+The command exits non-zero when any fixture route or interaction policy does not match the
+manifest.
 
 ## Scope
 
-Fixture checks validate route selection only. They do not verify daemon health, hidden
+Fixture checks validate route selection and declared interaction policy only. They do not
-service reachability, IPFS pinning, TLS, or content freshness.
+verify daemon health, hidden service reachability, IPFS pinning, TLS, or content freshness.
 
 The manifest must not contain private deployment notes, service keys, credentials, or
 host inventory.

tests/test_fixtures.py

@@ -32,6 +32,7 @@ class FixtureCheckTest(unittest.TestCase):
                                 "protocol": "ipfs",
                                 "url": "ipfs://bafyexample",
                                 "checks": {"transport": "ipfs", "profile": "ipfs"},
+                                "interaction": {"tier": "static"},
                             },
                         ],
                     }
@@ -70,6 +71,58 @@ class FixtureCheckTest(unittest.TestCase):
         self.assertEqual("fail", result.checks[0].status)
         self.assertIn("transport expected tor got clearnet", result.checks[0].message)
 
+    def test_rejects_transactional_ipfs_fixture(self) -> None:
+        result = _check_one(
+            "ipfs",
+            "ipfs://bafyexample",
+            {"tier": "transactional", "payments": "server-invoice"},
+        )
+
+        self.assertFalse(result.ok)
+        self.assertIn("tier transactional exceeds ipfs max static", result.checks[0].message)
+        self.assertIn("payments server-invoice not allowed on ipfs", result.checks[0].message)
+
+    def test_allows_transactional_tor_fixture(self) -> None:
+        result = _check_one(
+            "tor",
+            "http://example.onion/",
+            {"tier": "transactional", "identity": "http-session", "payments": "server-invoice"},
+        )
+
+        self.assertTrue(result.ok)
+
+    def test_allows_interactive_lite_gemini_but_rejects_realtime(self) -> None:
+        ok = _check_one("gemini", "gemini://example.test/", {"tier": "interactive-lite"})
+        bad = _check_one("gemini", "gemini://example.test/", {"tier": "realtime", "realtime": True})
+
+        self.assertTrue(ok.ok)
+        self.assertFalse(bad.ok)
+        self.assertIn("tier realtime exceeds gemini max interactive-lite", bad.checks[0].message)
+        self.assertIn("realtime not allowed on gemini", bad.checks[0].message)
+
+
+def _check_one(transport: str, url: str, interaction: dict) -> object:
+    with tempfile.TemporaryDirectory() as tmp:
+        path = Path(tmp) / "manifest.json"
+        path.write_text(
+            json.dumps(
+                {
+                    "schema": "ampg.fixture-manifest.v1",
+                    "site": {"id": "example", "domain": "example.test"},
+                    "fixtures": [
+                        {
+                            "protocol": transport,
+                            "url": url,
+                            "checks": {"transport": transport, "profile": transport},
+                            "interaction": interaction,
+                        }
+                    ],
+                }
+            ),
+            encoding="utf-8",
+        )
+        return check_fixture_manifest(path)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_interaction_policy.py

@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import unittest
+
+from ampbrowser.interaction_policy import interaction_failures
+
+
+class InteractionPolicyTest(unittest.TestCase):
+    def test_static_ipfs_is_allowed(self) -> None:
+        self.assertEqual(
+            [],
+            interaction_failures(
+                "ipfs",
+                tier="static",
+                identity="none",
+                payments="none",
+                realtime=False,
+                public_allowed=True,
+            ),
+        )
+
+    def test_public_allowed_false_fails(self) -> None:
+        failures = interaction_failures(
+            "tor",
+            tier="static",
+            identity="none",
+            payments="none",
+            realtime=False,
+            public_allowed=False,
+        )
+
+        self.assertEqual(["fixture is not public_allowed"], failures)
+
+
+if __name__ == "__main__":
+    unittest.main()