From c52e700b6f77974185d24fc29f3e8823a4b4a0ad Mon Sep 17 00:00:00 2001 From: jwinterm Date: Mon, 6 Jul 2026 06:02:58 -0400 Subject: [PATCH] Add fixture-based health plans --- README.md | 3 + ampg/cli.py | 67 +++++++++++++ ampg/health.py | 122 +++++++++++++++++++++++ docs/architecture.md | 10 ++ docs/daemon-management.md | 5 + docs/fixture-manifests.md | 5 + tests/test_health_plan.py | 197 ++++++++++++++++++++++++++++++++++++++ 7 files changed, 409 insertions(+) create mode 100644 ampg/health.py create mode 100644 tests/test_health_plan.py diff --git a/README.md b/README.md index 3486277..c6a8123 100644 --- a/README.md +++ b/README.md @@ -106,6 +106,8 @@ python3 -m ampg --config examples/wownero.gateway.toml status python3 -m ampg --config examples/wownero.gateway.toml doctor python3 -m ampg --config examples/wownero.gateway.toml install-plan --profile mobile-i2p python3 -m ampg --config examples/wownero.gateway.toml install-plan --profile mobile-i2p --write-artifacts +python3 -m ampg --config examples/wownero.gateway.toml health-plan --profile mobile-i2p +python3 -m ampg --config examples/wownero.gateway.toml health-plan --profile mobile-i2p --mode preview python3 -m ampg --config examples/wownero.gateway.toml apply --dry-run python3 -m ampg --config examples/wownero.gateway.toml apply --dry-run --profile mobile-i2p python3 -m ampg --config examples/wownero.gateway.toml apply --dry-run --protocol i2p @@ -154,5 +156,6 @@ Explicit `--protocol` and `--platform` flags override the profile when present. - [ ] Run `ampg install-plan`; review managed daemon package, state, and supervisor steps. - [ ] Run `ampg build`; inspect generated variants. - [ ] Run AMPB fixture checks against generated manifests. +- [ ] Run `ampg health-plan`; review post-start transport checks. - [ ] Run `ampg apply --dry-run`; review the activation sequence. - [ ] Run live apply after the activation sequence is clean. diff --git a/ampg/cli.py b/ampg/cli.py index a93c768..70c4e56 100644 --- a/ampg/cli.py +++ b/ampg/cli.py @@ -14,6 +14,7 @@ from .audit import audit_gateway from .build import build_gateway from .config import load_config from .docsgen import generate_docs +from .health import HealthCheck, blocked_health_checks, health_plan from .install_plan import ( InstallStep, blocked_install_steps, @@ -105,6 +106,28 @@ def main(argv: list[str] | None = None) -> int: action="store_true", help="Write managed daemon config and supervisor artifacts to the plan root.", ) + health_parser = subcommands.add_parser( + "health-plan", + help="Print fixture-based transport health checks.", + ) + _add_target_selection(health_parser) + health_parser.add_argument( + "--mode", + choices=("published", "preview"), + default="published", + help="Plan checks against published transport URLs or local preview URLs.", + ) + health_parser.add_argument( + "--base-port", + type=int, + default=19080, + help="First loopback port assigned to preview health checks.", + ) + health_parser.add_argument( + "--host", + default="127.0.0.1", + help="Loopback host for preview health checks.", + ) build_parser = subcommands.add_parser("build", help="Build enabled protocol outputs.") _add_target_selection(build_parser) fixture_parser = subcommands.add_parser("manifest", help="Write AMPB fixture manifests.") @@ -195,6 +218,8 @@ def main(argv: list[str] | None = None) -> int: return _cmd_apply(config, args) if args.command == "install-plan": return _cmd_install_plan(config, args) + if args.command == "health-plan": + return _cmd_health_plan(config, args) if args.command == "build": return _cmd_build(config) if args.command == "manifest": @@ -272,6 +297,28 @@ def _cmd_install_plan(config, args) -> int: return 1 if blocked else 0 +def _cmd_health_plan(config, args) -> int: + checks = health_plan( + config, + mode=args.mode, + base_port=args.base_port, + host=args.host, + ) + for check in checks: + _print_health_check(check) + blocked = blocked_health_checks(checks) + print( + "AMPG_HEALTH_SUMMARY " + f"mode={args.mode} " + f"sites={len(config.sites)} " + f"checks={len(checks)} " + f"planned={sum(1 for check in checks if check.status == 'planned')} " + f"review={sum(1 for check in checks if check.status == 'review')} " + f"blocked={len(blocked)}" + ) + return 1 if blocked else 0 + + def _add_target_selection(parser) -> None: parser.add_argument( "--profile", @@ -388,6 +435,26 @@ def _cmd_build(config) -> int: return 0 +def _print_health_check(check: HealthCheck) -> None: + print( + "AMPG_HEALTH_CHECK " + f"site={check.site_id} " + f"protocol={check.protocol} " + f"mode={check.mode} " + f"route={check.route} " + f"url=\"{_quote(check.url)}\" " + f"output_root=\"{_quote(str(check.output_root))}\" " + f"output_path={check.output_path} " + f"transport={check.transport} " + f"profile={check.profile} " + f"address_status={check.address_status} " + f"status={check.status} " + f"action={check.action} " + f"command=\"{_quote(check.command)}\" " + f"message=\"{_quote(check.message)}\"" + ) + + def _print_install_step(step: InstallStep) -> None: print( "AMPG_INSTALL_STEP " diff --git a/ampg/health.py b/ampg/health.py new file mode 100644 index 0000000..ce4be0e --- /dev/null +++ b/ampg/health.py @@ -0,0 +1,122 @@ +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from .config import GatewayConfig, SiteConfig +from .manifest import fixture_manifest +from .preview import preview_endpoints, preview_fixture_manifest + + +@dataclass(frozen=True) +class HealthCheck: + site_id: str + protocol: str + mode: str + url: str + output_root: Path + output_path: str + transport: str + profile: str + address_status: str + route: str + status: str + action: str + command: str + message: str + + +def health_plan( + config: GatewayConfig, + *, + mode: str = "published", + base_port: int = 19080, + host: str = "127.0.0.1", +) -> list[HealthCheck]: + if mode not in {"published", "preview"}: + raise ValueError(f"unsupported health mode {mode!r}") + + endpoints = preview_endpoints(config, base_port=base_port, host=host) if mode == "preview" else [] + checks: list[HealthCheck] = [] + for site in config.sites: + manifest = _manifest_for_mode(site, mode=mode, endpoints=endpoints) + for fixture in manifest["fixtures"]: + checks.append(_health_check(site, fixture, mode=mode)) + return checks + + +def blocked_health_checks(checks: list[HealthCheck]) -> list[HealthCheck]: + return [check for check in checks if check.status == "blocked"] + + +def _manifest_for_mode( + site: SiteConfig, + *, + mode: str, + endpoints, +) -> dict[str, Any]: + if mode == "preview": + return preview_fixture_manifest(site, endpoints) + return fixture_manifest(site) + + +def _health_check(site: SiteConfig, fixture: dict[str, Any], *, mode: str) -> HealthCheck: + protocol = str(fixture["protocol"]) + output_root = site.outputs.root / protocol + route = str(fixture.get("route", {}).get("fixture_path", "/")) + transport = str(fixture["checks"]["transport"]) + profile = str(fixture["checks"]["profile"]) + address_status = str(fixture["address_status"]) + output_ready = (output_root / ".ampg-output").exists() + preview_status = fixture.get("preview", {}).get("status") + + if not output_ready or preview_status == "missing-output": + status = "blocked" + action = "build-output" + message = f"generated output is missing: {output_root}" + elif address_status == "placeholder": + status = "review" + action = "configure-address" + message = "fixture uses a placeholder address; update protocol URL after daemon identity exists" + else: + status = "planned" + action = "check-fixture" + message = ( + "verify fixture through the local preview server" + if mode == "preview" + else "verify fixture through the selected transport/profile after services start" + ) + + return HealthCheck( + site_id=site.id, + protocol=protocol, + mode=mode, + url=str(fixture["url"]), + output_root=output_root, + output_path=str(fixture["output_path"]), + transport=transport, + profile=profile, + address_status=address_status, + route=route, + status=status, + action=action, + command=_check_command(str(fixture["url"]), transport=transport, profile=profile), + message=message, + ) + + +def _check_command(url: str, *, transport: str, profile: str) -> str: + if transport == "clearnet": + return f"curl -fsSL {url}" + if transport == "tor": + return f"curl --proxy socks5h://127.0.0.1:9050 -fsSL {url}" + if transport == "i2p": + return f"curl --proxy http://127.0.0.1:4444 -fsSL {url}" + if transport == "gemini": + return f"gemini fetch {url}" + if transport == "ipfs": + return f"ipfs cat {url}" + if transport == "reticulum": + return f"rns-fetch {url}" + return f"fetch {url} with {profile}" diff --git a/docs/architecture.md b/docs/architecture.md index 5f812bb..fa71bc4 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -132,6 +132,16 @@ Tor and I2P HTTP publishing include the transport daemon and the loopback web-se layer in the install plan, because the hidden service or tunnel must have a local HTTP target. +## Health plan + +`ampg health-plan` turns fixture manifests into post-start verification checks. Published +mode targets real transport URLs. Preview mode rewrites checks to loopback HTTP preview +endpoints so local generated output can be verified before live daemons exist. + +Missing generated output blocks health checks. Placeholder transport addresses are +reported for review, because the operator must capture the real onion, I2P destination, +or other transport identity before published checks are exact. + ## Interaction boundary v1 is static-first, but the architecture reserves a path for interactive applications. diff --git a/docs/daemon-management.md b/docs/daemon-management.md index 6b78832..a0ed5e6 100644 --- a/docs/daemon-management.md +++ b/docs/daemon-management.md @@ -41,6 +41,7 @@ python3 -m ampg --config gateway.toml status python3 -m ampg --config gateway.toml doctor python3 -m ampg --config gateway.toml install-plan --profile mobile-i2p python3 -m ampg --config gateway.toml install-plan --profile mobile-i2p --write-artifacts +python3 -m ampg --config gateway.toml health-plan --profile mobile-i2p python3 -m ampg --config gateway.toml apply --dry-run python3 -m ampg --config gateway.toml apply --dry-run --profile mobile-i2p python3 -m ampg --config gateway.toml apply --dry-run --protocol i2p @@ -65,6 +66,10 @@ files under the configured `plan_root`, grouped by site, protocol, and platform. Tor/I2P HTTP publishing, AMPG writes both transport-daemon artifacts and loopback nginx artifacts. +`health-plan` prints fixture-based checks to run after services start. Published checks +target real transport URLs; preview checks target local loopback preview endpoints. +Missing output blocks the plan, while placeholder transport addresses are review items. + `apply --dry-run` prints the activation sequence for each enabled protocol: generated output readiness, config artifacts to review, daemon action, and post-apply health checks. It exits nonzero when any step is blocked. `--write-artifacts` writes reviewable diff --git a/docs/fixture-manifests.md b/docs/fixture-manifests.md index ac2b9d7..99eba91 100644 --- a/docs/fixture-manifests.md +++ b/docs/fixture-manifests.md @@ -33,6 +33,8 @@ or deployment notes. python3 -m ampg --config examples/wownero.gateway.toml build python3 -m ampg --config examples/wownero.gateway.toml manifest python3 -m ampg --config examples/wownero.gateway.toml preview manifest +python3 -m ampg --config examples/wownero.gateway.toml health-plan +python3 -m ampg --config examples/wownero.gateway.toml health-plan --mode preview ``` `build` emits the manifest after writing protocol outputs. `manifest` rewrites only the @@ -49,6 +51,9 @@ They preserve the intended published URL and transport checks under each fixture `published` metadata. Use preview manifests to verify generated output with AMPB before real Tor, I2P, Gemini, or other transport daemons are installed or adopted. +`health-plan` consumes the same fixture contract. Published mode prints checks for real +transport URLs. Preview mode prints loopback checks against generated preview URLs. + ## Address Status Some transports may not have final public addresses when a fixture is generated. In those diff --git a/tests/test_health_plan.py b/tests/test_health_plan.py new file mode 100644 index 0000000..2262bf6 --- /dev/null +++ b/tests/test_health_plan.py @@ -0,0 +1,197 @@ +import contextlib +import io +import tempfile +import unittest +from pathlib import Path + +from ampg.cli import main +from ampg.config import load_config +from ampg.health import blocked_health_checks, health_plan + + +def _write_config(root: Path, protocol_block: str) -> Path: + source = root / "site" + source.mkdir() + (source / "index.html").write_text("

Hello

", encoding="utf-8") + config_path = root / "gateway.toml" + config_path.write_text( + f""" +[profiles.mobile-i2p] +protocols = ["i2p"] + +[[site]] +id = "example" +domain = "example.test" + +[site.source] +kind = "static-html" +path = "./site" +canonical_url = "https://example.test" + +[site.outputs] +root = "./out" +plan_root = "./plan" + +{protocol_block} +""", + encoding="utf-8", + ) + return config_path + + +def _mark_output(root: Path, protocol: str) -> None: + output = root / f"out/{protocol}" + output.mkdir(parents=True) + (output / ".ampg-output").write_text("generated by AMPG\n", encoding="utf-8") + + +class HealthPlanTest(unittest.TestCase): + def test_published_health_plan_reviews_placeholder_i2p_address(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + config = load_config( + _write_config( + root, + """ +[site.protocols.i2p] +enabled = true +renderer = "privacy-html" +daemon = "i2pd" +daemon_policy = "auto" +""", + ) + ) + _mark_output(root, "i2p") + + checks = health_plan(config) + + self.assertEqual(1, len(checks)) + self.assertEqual([], blocked_health_checks(checks)) + self.assertEqual("review", checks[0].status) + self.assertEqual("configure-address", checks[0].action) + self.assertEqual("placeholder", checks[0].address_status) + self.assertIn("127.0.0.1:4444", checks[0].command) + + def test_configured_i2p_health_plan_is_planned(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + config = load_config( + _write_config( + root, + """ +[site.protocols.i2p] +enabled = true +renderer = "privacy-html" +i2p_url = "site.i2p" +""", + ) + ) + _mark_output(root, "i2p") + + checks = health_plan(config) + + self.assertEqual("planned", checks[0].status) + self.assertEqual("configured", checks[0].address_status) + self.assertEqual("http://site.i2p/", checks[0].url) + + def test_missing_output_blocks_health_plan(self): + with tempfile.TemporaryDirectory() as tmp: + config = load_config( + _write_config( + Path(tmp), + """ +[site.protocols.tor] +enabled = true +renderer = "privacy-html" +""", + ) + ) + + checks = health_plan(config) + + self.assertEqual(1, len(blocked_health_checks(checks))) + self.assertEqual("build-output", checks[0].action) + + def test_preview_health_plan_uses_loopback_urls(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + config = load_config( + _write_config( + root, + """ +[site.protocols.tor] +enabled = true +renderer = "privacy-html" +""", + ) + ) + _mark_output(root, "tor") + + checks = health_plan(config, mode="preview", base_port=19900) + + self.assertEqual(1, len(checks)) + self.assertEqual("planned", checks[0].status) + self.assertEqual("preview", checks[0].address_status) + self.assertEqual("http://127.0.0.1:19900/", checks[0].url) + self.assertEqual("clearnet", checks[0].transport) + + def test_health_plan_cli_uses_profile_scope(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + config_path = _write_config( + root, + """ +[site.protocols.clearnet] +enabled = true +renderer = "clearnet" + +[site.protocols.i2p] +enabled = true +renderer = "privacy-html" +i2p_url = "site.i2p" +""", + ) + _mark_output(root, "i2p") + + stdout = io.StringIO() + with contextlib.redirect_stdout(stdout): + status = main( + [ + "--config", + str(config_path), + "health-plan", + "--profile", + "mobile-i2p", + ] + ) + + output = stdout.getvalue() + + self.assertEqual(0, status) + self.assertIn("protocol=i2p", output) + self.assertIn("planned=1", output) + self.assertNotIn("protocol=clearnet", output) + + def test_health_plan_cli_blocks_missing_output(self): + with tempfile.TemporaryDirectory() as tmp: + config_path = _write_config( + Path(tmp), + """ +[site.protocols.i2p] +enabled = true +renderer = "privacy-html" +""", + ) + + stdout = io.StringIO() + with contextlib.redirect_stdout(stdout): + status = main(["--config", str(config_path), "health-plan"]) + + output = stdout.getvalue() + + self.assertEqual(1, status) + self.assertIn("blocked=1", output) + + +if __name__ == "__main__": + unittest.main()