Emit AMPB fixture manifests

ampg/cli.py

@@ -8,6 +8,7 @@ from .audit import audit_gateway
 from .build import build_gateway
 from .config import load_config
 from .docsgen import generate_docs
+from .manifest import write_fixture_manifests
 from .plan import plan_gateway, write_plan_artifacts
 
 
@@ -27,6 +28,7 @@ def main(argv: list[str] | None = None) -> int:
         help="Write generated config snippets to the configured plan root.",
     )
     subcommands.add_parser("build", help="Build enabled protocol outputs.")
+    subcommands.add_parser("manifest", help="Write AMPB fixture manifests.")
     audit_parser = subcommands.add_parser("audit", help="Audit source HTML quality.")
     audit_parser.add_argument(
         "--fail-on-warn",
@@ -52,6 +54,8 @@ def main(argv: list[str] | None = None) -> int:
             return _cmd_plan(config, write_artifacts=args.write_artifacts)
         if args.command == "build":
             return _cmd_build(config)
+        if args.command == "manifest":
+            return _cmd_manifest(config)
         if args.command == "audit":
             return _cmd_audit(config, fail_on_warn=args.fail_on_warn)
     except Exception as exc:  # noqa: BLE001 - CLI should print concise failures.
@@ -102,6 +106,24 @@ def _cmd_build(config) -> int:
             f" skipped={result.files_skipped}"
             f"{extra}"
         )
+    for manifest in write_fixture_manifests(config):
+        print(
+            "AMPG_MANIFEST "
+            f"site={manifest.site_id} "
+            f"path={manifest.path} "
+            f"fixtures={manifest.fixture_count}"
+        )
+    return 0
+
+
+def _cmd_manifest(config) -> int:
+    for manifest in write_fixture_manifests(config):
+        print(
+            "AMPG_MANIFEST "
+            f"site={manifest.site_id} "
+            f"path={manifest.path} "
+            f"fixtures={manifest.fixture_count}"
+        )
     return 0
 
 

ampg/config.py

@@ -12,6 +12,7 @@ DEFAULT_DAEMONS = {
     "i2p": "i2pd",
     "gemini": "agate",
     "reticulum": "rnsd",
+    "ipfs": "ipfs",
 }
 
 DEFAULT_POLICIES = {
@@ -20,6 +21,7 @@ DEFAULT_POLICIES = {
     "i2p": "auto",
     "gemini": "auto",
     "reticulum": "auto",
+    "ipfs": "auto",
 }
 
 

ampg/manifest.py

@@ -0,0 +1,144 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+import json
+from pathlib import Path
+from typing import Any
+
+from .config import GatewayConfig, ProtocolConfig, SiteConfig
+
+
+SCHEMA = "ampg.fixture-manifest.v1"
+
+
+@dataclass(frozen=True)
+class ManifestWriteResult:
+    site_id: str
+    path: Path
+    fixture_count: int
+
+
+def fixture_manifest(site: SiteConfig) -> dict[str, Any]:
+    fixtures = [
+        _fixture_entry(site, protocol)
+        for protocol in site.protocols.values()
+        if protocol.enabled
+    ]
+    return {
+        "schema": SCHEMA,
+        "site": {
+            "id": site.id,
+            "domain": site.domain,
+            "canonical_url": site.source.canonical_url or "",
+        },
+        "fixtures": fixtures,
+    }
+
+
+def write_fixture_manifests(config: GatewayConfig) -> list[ManifestWriteResult]:
+    results: list[ManifestWriteResult] = []
+    for site in config.sites:
+        manifest = fixture_manifest(site)
+        path = manifest_path(site)
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(
+            json.dumps(manifest, indent=2, sort_keys=True) + "\n",
+            encoding="utf-8",
+        )
+        results.append(
+            ManifestWriteResult(
+                site_id=site.id,
+                path=path,
+                fixture_count=len(manifest["fixtures"]),
+            )
+        )
+    return results
+
+
+def manifest_path(site: SiteConfig) -> Path:
+    return site.outputs.root / "ampg-fixture-manifest.json"
+
+
+def _fixture_entry(site: SiteConfig, protocol: ProtocolConfig) -> dict[str, Any]:
+    url, address_status = _fixture_url(site, protocol)
+    transport = _transport_for(protocol.name)
+    return {
+        "protocol": protocol.name,
+        "renderer": protocol.renderer,
+        "url": url,
+        "address_status": address_status,
+        "output_path": protocol.name,
+        "checks": {
+            "transport": transport,
+            "profile": transport,
+        },
+    }
+
+
+def _transport_for(protocol_name: str) -> str:
+    if protocol_name in {"clearnet", "tor", "i2p", "gemini", "reticulum", "ipfs"}:
+        return protocol_name
+    return protocol_name
+
+
+def _fixture_url(site: SiteConfig, protocol: ProtocolConfig) -> tuple[str, str]:
+    configured_url = protocol.options.get("fixture_url") or protocol.options.get("browser_url")
+    if configured_url:
+        return _with_trailing_slash(str(configured_url)), "configured"
+
+    if protocol.name == "clearnet":
+        if site.source.canonical_url:
+            return _with_trailing_slash(site.source.canonical_url), "configured"
+        return f"https://{site.domain}/", "derived"
+
+    if protocol.name == "tor":
+        onion_url = protocol.options.get("onion_url")
+        if onion_url:
+            return _with_trailing_slash(str(onion_url)), "configured"
+        onion_location = str(protocol.options.get("onion_location", "auto"))
+        if onion_location and onion_location != "auto":
+            return _with_scheme(onion_location, "http"), "configured"
+        return f"http://{site.id}.onion/", "placeholder"
+
+    if protocol.name == "i2p":
+        i2p_url = protocol.options.get("i2p_url") or protocol.options.get("i2p_hostname")
+        if i2p_url:
+            return _with_scheme(str(i2p_url), "http"), "configured"
+        return f"http://{site.id}.i2p/", "placeholder"
+
+    if protocol.name == "gemini":
+        gemini_url = protocol.options.get("gemini_url")
+        if gemini_url:
+            return _with_scheme(str(gemini_url), "gemini"), "configured"
+        return f"gemini://{site.domain}/", "derived"
+
+    if protocol.name == "reticulum":
+        rns_url = protocol.options.get("rns_url")
+        if rns_url:
+            return _with_scheme(str(rns_url), "rns"), "configured"
+        return f"rns://{site.id}", "placeholder"
+
+    if protocol.name == "ipfs":
+        ipfs_url = protocol.options.get("ipfs_url")
+        if ipfs_url:
+            return _with_scheme(str(ipfs_url), "ipfs"), "configured"
+        cid = protocol.options.get("cid")
+        if cid:
+            return f"ipfs://{cid}", "configured"
+        return f"/ipfs/{site.id}", "placeholder"
+
+    return f"https://{site.domain}/", "derived"
+
+
+def _with_scheme(value: str, scheme: str) -> str:
+    if "://" in value:
+        return _with_trailing_slash(value)
+    return _with_trailing_slash(f"{scheme}://{value}")
+
+
+def _with_trailing_slash(value: str) -> str:
+    if value.startswith(("rns://", "lxmf://", "nomad://", "ipfs://", "ipns://")):
+        return value
+    if value.endswith("/"):
+        return value
+    return value + "/"

ampg/metadata.py

@@ -188,4 +188,14 @@ DAEMON_ADAPTERS = (
             "Certificate and key paths are plan values until install/apply support exists.",
         ),
     ),
+    DaemonAdapter(
+        daemon=DEFAULT_DAEMONS["ipfs"],
+        protocols=("ipfs",),
+        default_policy=DEFAULT_POLICIES["ipfs"],
+        generated_artifacts=("fixture manifest route expectations",),
+        notes=(
+            "IPFS output is static web content for a local gateway or later pinning.",
+            "IPFS is content-addressed distribution, not an anonymity layer.",
+        ),
+    ),
 )