from __future__ import annotations import argparse from pathlib import Path import sys from .activation import ( ActivationStep, activation_steps, blocked_steps, write_activation_artifacts, ) from .addresses import ( AddressCaptureResult, AddressRecord, address_registry_path, capture_addresses, effective_address_records, set_address, ) 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 ( InstallStateCopy, InstallStep, blocked_install_steps, install_state_copies, install_plan, write_install_artifacts, ) from .manifest import write_fixture_manifests from .plan import plan_gateway, write_plan_artifacts from .platforms import PLATFORM_NAMES, platform_by_name from .preview import PreviewServers, preview_endpoints, write_preview_fixture_manifests from .route_manifest import ( load_route_manifest, route_manifest_schema_json, validate_route_manifest, ) from .route_policy import RouteExposure, RouteIssue, route_exposures, route_issues from .selection import protocols_for_selection, select_profile, select_protocols from .state_contract import StatePathContract, state_contract from .status import DoctorIssue, TransportStatus, doctor_gateway, gateway_status def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(prog="ampg") parser.add_argument( "--config", type=Path, default=Path("gateway.toml"), help="Path to gateway TOML config.", ) subcommands = parser.add_subparsers(dest="command", required=True) plan_parser = subcommands.add_parser("plan", help="Print a dry-run plan.") _add_target_selection(plan_parser) plan_parser.add_argument( "--write-artifacts", action="store_true", help="Write generated config snippets to the configured plan root.", ) status_parser = subcommands.add_parser( "status", help="Show enabled transport daemon decisions.", ) _add_target_selection(status_parser) status_parser.add_argument( "--platform", choices=PLATFORM_NAMES, help="Override platform detection for dry-run checks.", ) doctor_parser = subcommands.add_parser( "doctor", help="Check source, renderer, route, and daemon readiness.", ) _add_target_selection(doctor_parser) doctor_parser.add_argument( "--platform", choices=PLATFORM_NAMES, help="Override platform detection for dry-run checks.", ) apply_parser = subcommands.add_parser( "apply", help="Show or run the transport activation sequence.", ) _add_target_selection(apply_parser) apply_parser.add_argument( "--dry-run", action="store_true", help="Print the activation sequence without changing services.", ) apply_parser.add_argument( "--write-artifacts", action="store_true", help="Write generated config snippets to the configured plan root.", ) apply_parser.add_argument( "--platform", choices=PLATFORM_NAMES, help="Override platform detection for dry-run checks.", ) install_plan_parser = subcommands.add_parser( "install-plan", help="Print managed daemon package, state, and supervisor steps.", ) _add_target_selection(install_plan_parser) install_plan_parser.add_argument( "--platform", choices=PLATFORM_NAMES, help="Override platform detection for install planning.", ) install_plan_parser.add_argument( "--write-artifacts", 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.") _add_target_selection(fixture_parser) addresses_parser = subcommands.add_parser( "addresses", help="Inspect or update captured transport addresses.", ) addresses_subcommands = addresses_parser.add_subparsers( dest="addresses_command", required=True, ) addresses_list = addresses_subcommands.add_parser( "list", help="Print effective configured, captured, derived, and placeholder addresses.", ) _add_target_selection(addresses_list) addresses_capture = addresses_subcommands.add_parser( "capture", help="Capture generated transport addresses from daemon state files.", ) _add_target_selection(addresses_capture) addresses_set = addresses_subcommands.add_parser( "set", help="Manually store a generated transport address in AMPG state.", ) addresses_set.add_argument("--site", required=True, help="Site id to update.") addresses_set.add_argument( "--protocol", dest="address_protocol", required=True, help="Protocol name to update.", ) addresses_set.add_argument("--url", required=True, help="Public transport URL or hostname.") addresses_set.add_argument( "--source", default="manual", help="Source label stored with the captured address.", ) state_contract_parser = subcommands.add_parser( "state-contract", help="Print AMPG-owned state paths and daemon-written files.", ) _add_target_selection(state_contract_parser) preview_parser = subcommands.add_parser("preview", help="Preview generated outputs locally.") preview_subcommands = preview_parser.add_subparsers(dest="preview_command", required=True) for preview_command in ("endpoints", "manifest", "serve"): command_parser = preview_subcommands.add_parser( preview_command, help=f"Preview {preview_command} for generated outputs.", ) _add_target_selection(command_parser) command_parser.add_argument( "--base-port", type=int, default=19080, help="First loopback port assigned to enabled protocol outputs.", ) command_parser.add_argument( "--host", default="127.0.0.1", help="Loopback host for preview URLs.", ) routes_parser = subcommands.add_parser("routes", help="Explain route exposure policy.") routes_subcommands = routes_parser.add_subparsers(dest="routes_command", required=True) routes_explain = routes_subcommands.add_parser( "explain", help="Print per-protocol route decisions.", ) _add_target_selection(routes_explain) routes_validate = routes_subcommands.add_parser( "validate", help="Fail when a public route has no compatible enabled protocol.", ) _add_target_selection(routes_validate) manifest_parser = subcommands.add_parser( "route-manifest", help="Validate or print the app route manifest contract.", ) manifest_subcommands = manifest_parser.add_subparsers( dest="route_manifest_command", required=True, ) manifest_validate = manifest_subcommands.add_parser( "validate", help="Validate an ampg.route-manifest.v1 JSON file.", ) manifest_validate.add_argument("path", type=Path, help="Route manifest JSON path.") manifest_schema = manifest_subcommands.add_parser( "schema", help="Print the JSON Schema for route manifests.", ) manifest_schema.add_argument( "--output", type=Path, help="Write the JSON Schema to this path instead of stdout.", ) audit_parser = subcommands.add_parser("audit", help="Audit source HTML quality.") audit_parser.add_argument( "--fail-on-warn", action="store_true", help="Exit non-zero when warnings are found.", ) docs_parser = subcommands.add_parser("docs", help="Generate or check generated docs.") docs_subcommands = docs_parser.add_subparsers(dest="docs_command", required=True) docs_generate = docs_subcommands.add_parser("generate", help="Generate docs from code.") docs_generate.add_argument( "--check", action="store_true", help="Fail if generated docs are stale instead of writing them.", ) args = parser.parse_args(argv) try: if args.command == "docs": return _cmd_docs(args) if args.command == "route-manifest": return _cmd_route_manifest(args) config = _selected_config(load_config(args.config), args) if args.command == "plan": return _cmd_plan(config, write_artifacts=_write_artifacts_enabled(config, args)) if args.command == "status": return _cmd_status(config, args) if args.command == "doctor": return _cmd_doctor(config, args) if args.command == "apply": 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": return _cmd_manifest(config) if args.command == "addresses": return _cmd_addresses(config, args) if args.command == "state-contract": return _cmd_state_contract(config) if args.command == "preview": return _cmd_preview(config, args) if args.command == "routes": return _cmd_routes(config, args) 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. print(f"AMPG status=error message={exc}", file=sys.stderr) return 1 return 1 def _cmd_apply(config, args) -> int: if not args.dry_run: print( 'AMPG_APPLY status=error message="live apply is not implemented; rerun with --dry-run"', file=sys.stderr, ) return 1 platform_provider = _platform_override(config, args) if _write_artifacts_enabled(config, args): for path in write_activation_artifacts(config, platform_provider=platform_provider): print(f"AMPG_APPLY_ARTIFACT path={path} status=written") for copy in install_state_copies(config, platform_provider=platform_provider): _print_install_state_copy(copy) steps = activation_steps(config, platform_provider=platform_provider) for step in steps: _print_activation_step(step) blocked = blocked_steps(steps) print( "AMPG_APPLY_SUMMARY " "mode=dry-run " f"sites={len(config.sites)} " f"steps={len(steps)} " f"ready={sum(1 for step in steps if step.status == 'ready')} " f"review={sum(1 for step in steps if step.status == 'review')} " f"planned={sum(1 for step in steps if step.status == 'planned')} " f"blocked={len(blocked)}" ) return 1 if blocked else 0 def _cmd_install_plan(config, args) -> int: platform_provider = _platform_override(config, args) if _write_artifacts_enabled(config, args): for artifact in write_install_artifacts( config, platform_provider=platform_provider, ): print( "AMPG_INSTALL_ARTIFACT " f"site={artifact.site_id} " f"protocol={artifact.protocol} " f"platform={artifact.platform} " f"kind={artifact.kind} " f"path={artifact.path}" ) steps = install_plan(config, platform_provider=platform_provider) for step in steps: _print_install_step(step) blocked = blocked_install_steps(steps) print( "AMPG_INSTALL_SUMMARY " f"sites={len(config.sites)} " f"steps={len(steps)} " f"planned={sum(1 for step in steps if step.status == 'planned')} " f"ready={sum(1 for step in steps if step.status == 'ready')} " f"skipped={sum(1 for step in steps if step.status == 'skipped')} " f"blocked={len(blocked)}" ) 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", help="Named deployment profile from gateway config.", ) parser.add_argument( "--protocol", action="append", default=[], help="Limit this command to one protocol. May be repeated or comma-separated.", ) def _selected_config(config, args): protocols = protocols_for_selection( config, raw_protocols=getattr(args, "protocol", None), profile_name=getattr(args, "profile", None), ) return select_protocols(config, protocols) def _write_artifacts_enabled(config, args) -> bool: if getattr(args, "write_artifacts", False): return True profile = select_profile(config, getattr(args, "profile", None)) return bool(profile and profile.write_artifacts) def _cmd_status(config, args) -> int: statuses = gateway_status(config, platform_provider=_platform_override(config, args)) for status in statuses: _print_transport_status(status) print( "AMPG_STATUS_SUMMARY " f"sites={len(config.sites)} " f"protocols={len(statuses)} " f"ok={sum(1 for status in statuses if status.status == 'ok')} " f"warnings={sum(1 for status in statuses if status.status == 'warn')} " f"errors={sum(1 for status in statuses if status.status == 'error')}" ) return 0 def _cmd_doctor(config, args) -> int: platform_provider = _platform_override(config, args) statuses = gateway_status(config, platform_provider=platform_provider) issues = doctor_gateway(config, platform_provider=platform_provider, statuses=statuses) for status in statuses: _print_transport_status(status) for issue in issues: _print_doctor_issue(issue) errors = sum(1 for issue in issues if issue.severity == "error") warnings = sum(1 for issue in issues if issue.severity == "warning") print( "AMPG_DOCTOR_SUMMARY " f"sites={len(config.sites)} " f"checks={len(statuses) + len(issues)} " f"warnings={warnings} " f"errors={errors}" ) return 1 if errors else 0 def _cmd_plan(config, *, write_artifacts: bool) -> int: for line in plan_gateway(config): artifact_text = ",".join(str(path) for path in line.artifacts) if line.artifacts else "-" print( "AMPG_PLAN " f"site={line.site_id} " f"protocol={line.protocol} " f"renderer={line.renderer} " f"output={line.output_root} " f"daemon={line.daemon} " f"policy={line.daemon_policy} " f"artifacts={artifact_text} " f"action=\"{line.action}\"" ) if write_artifacts: artifacts = write_plan_artifacts(config) for artifact in artifacts: print(f"AMPG_ARTIFACT path={artifact.path}") return 0 def _cmd_build(config) -> int: for result in build_gateway(config): extra = "" if result.privacy_stats is not None: stats = result.privacy_stats extra = ( f" removed_active_tags={stats.removed_active_tags}" f" removed_event_handlers={stats.removed_event_handlers}" f" removed_inline_styles={stats.removed_inline_styles}" f" removed_remote_assets={stats.removed_remote_assets}" ) print( "AMPG_BUILD " f"site={result.site_id} " f"protocol={result.protocol} " f"renderer={result.renderer} " f"output={result.output_root} " f"files={result.files_written}" 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 _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 " f"site={step.site_id} " f"protocol={step.protocol} " f"platform={step.platform} " f"stage={step.stage} " f"action={step.action} " f"target=\"{_quote(step.target)}\" " f"status={step.status} " f"command=\"{_quote(step.command)}\" " f"message=\"{_quote(step.message)}\"" ) def _print_install_state_copy(copy: InstallStateCopy) -> None: print( "AMPG_APPLY_STATE_COPY " f"site={copy.site_id} " f"protocol={copy.protocol} " f"platform={copy.platform} " f"kind={copy.kind} " f"source=\"{_quote(str(copy.source))}\" " f"target=\"{_quote(str(copy.target))}\" " f"status={copy.status} " f"command=\"{_quote(copy.command)}\" " f"message=\"{_quote(copy.message)}\"" ) def _print_activation_step(step: ActivationStep) -> None: print( "AMPG_APPLY_STEP " f"site={step.site_id} " f"protocol={step.protocol} " f"stage={step.stage} " f"action={step.action} " f"target={step.target} " f"status={step.status} " f"command=\"{_quote(step.command)}\" " f"message=\"{_quote(step.message)}\"" ) def _print_transport_status(status: TransportStatus) -> None: executable_path = status.executable_path or "-" print( "AMPG_STATUS " f"site={status.site_id} " f"protocol={status.protocol} " f"renderer={status.renderer} " f"daemon={status.daemon} " f"policy={status.daemon_policy} " f"platform={status.platform} " f"supervisor={status.supervisor} " f"adapter={status.adapter} " f"backend={status.backend} " f"installed={_bool(status.installed)} " f"running={_bool(status.running)} " f"adoptable={_bool(status.adoptable)} " f"manageable={_bool(status.manageable)} " f"executable={executable_path} " f"status={status.status} " f"action={status.action} " f"message=\"{_quote(status.message)}\"" ) def _print_doctor_issue(issue: DoctorIssue) -> None: print( "AMPG_DOCTOR " f"site={issue.site_id} " f"protocol={issue.protocol} " f"severity={issue.severity} " f"code={issue.code} " f"message=\"{_quote(issue.message)}\"" ) 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 def _cmd_addresses(config, args) -> int: if args.addresses_command == "list": records = effective_address_records(config) for record in records: _print_address(record) print( "AMPG_ADDRESS_SUMMARY " f"sites={len(config.sites)} " f"addresses={len(records)} " f"configured={sum(1 for record in records if record.address_status == 'configured')} " f"captured={sum(1 for record in records if record.address_status == 'captured')} " f"derived={sum(1 for record in records if record.address_status == 'derived')} " f"placeholder={sum(1 for record in records if record.address_status == 'placeholder')} " f"registry={address_registry_path(config)}" ) return 0 if args.addresses_command == "capture": results = capture_addresses(config) for result in results: _print_address_capture(result) print( "AMPG_ADDRESS_CAPTURE_SUMMARY " f"sites={len(config.sites)} " f"results={len(results)} " f"captured={sum(1 for result in results if result.status == 'captured')} " f"configured={sum(1 for result in results if result.status == 'configured')} " f"skipped={sum(1 for result in results if result.status == 'skipped')} " f"missing={sum(1 for result in results if result.status == 'missing')} " f"registry={address_registry_path(config)}" ) return 0 if args.addresses_command == "set": record = set_address( config, site_id=args.site, protocol=args.address_protocol, url=args.url, source=args.source, ) _print_address(record) print(f"AMPG_ADDRESS_SET status=written registry={address_registry_path(config)}") return 0 return 1 def _cmd_state_contract(config) -> int: contracts = state_contract(config) for contract in contracts: _print_state_contract(contract) print( "AMPG_STATE_SUMMARY " f"sites={len(config.sites)} " f"entries={len(contracts)} " f"required={sum(1 for contract in contracts if contract.required)} " f"sensitive={sum(1 for contract in contracts if contract.sensitive)}" ) return 0 def _print_state_contract(contract: StatePathContract) -> None: print( "AMPG_STATE " f"site={contract.site_id} " f"protocol={contract.protocol} " f"role={contract.role} " f"owner={contract.owner} " f"required={_bool(contract.required)} " f"sensitive={_bool(contract.sensitive)} " f"path=\"{_quote(str(contract.path))}\" " f"description=\"{_quote(contract.description)}\"" ) def _print_address(record: AddressRecord) -> None: print( "AMPG_ADDRESS " f"site={record.site_id} " f"protocol={record.protocol} " f"url=\"{_quote(record.url)}\" " f"address_status={record.address_status} " f"source=\"{_quote(record.source)}\"" ) def _print_address_capture(result: AddressCaptureResult) -> None: path = str(result.path) if result.path else "-" print( "AMPG_ADDRESS_CAPTURE " f"site={result.site_id} " f"protocol={result.protocol} " f"status={result.status} " f"url=\"{_quote(result.url)}\" " f"source=\"{_quote(result.source)}\" " f"path=\"{_quote(path)}\" " f"message=\"{_quote(result.message)}\"" ) def _cmd_preview(config, args) -> int: endpoints = preview_endpoints(config, base_port=args.base_port, host=args.host) if args.preview_command == "endpoints": for endpoint in endpoints: _print_preview_endpoint(endpoint) print( "AMPG_PREVIEW_SUMMARY " f"sites={len(config.sites)} endpoints={len(endpoints)}" ) return 0 if args.preview_command == "manifest": for endpoint in endpoints: _print_preview_endpoint(endpoint) for result in write_preview_fixture_manifests( config, base_port=args.base_port, host=args.host, ): print( "AMPG_PREVIEW_MANIFEST " f"site={result.site_id} " f"path={result.path} " f"fixtures={result.fixture_count}" ) return 0 if args.preview_command == "serve": for endpoint in endpoints: _print_preview_endpoint(endpoint) with PreviewServers(endpoints) as servers: print( "AMPG_PREVIEW_SERVE " f"status=running endpoints={len(servers.endpoints)}" ) servers.wait_forever() return 0 return 1 def _print_preview_endpoint(endpoint) -> None: print( "AMPG_PREVIEW_ENDPOINT " f"site={endpoint.site_id} " f"protocol={endpoint.protocol} " f"renderer={endpoint.renderer} " f"url={endpoint.url} " f"root={endpoint.root} " f"status={endpoint.status}" ) def _cmd_routes(config, args) -> int: exposures = route_exposures(config) issues = route_issues(config) for exposure in exposures: _print_route_exposure(exposure) for issue in issues: _print_route_issue(issue) print( "AMPG_ROUTE_SUMMARY " f"sites={len(config.sites)} " f"routes={sum(len(site.interactions.routes) for site in config.sites)} " f"decisions={len(exposures)} " f"exposed={sum(1 for exposure in exposures if exposure.status == 'exposed')} " f"skipped={sum(1 for exposure in exposures if exposure.status == 'skipped')} " f"issues={len(issues)}" ) if args.routes_command == "validate" and issues: return 1 return 0 def _cmd_route_manifest(args) -> int: if args.route_manifest_command == "schema": content = route_manifest_schema_json() if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(content, encoding="utf-8") print(f"AMPG_ROUTE_MANIFEST_SCHEMA path={args.output}") else: print(content, end="") return 0 if args.route_manifest_command == "validate": try: data = load_route_manifest(args.path) except ValueError: try: import json data = json.loads(args.path.read_text(encoding="utf-8")) issues = validate_route_manifest(data) except Exception as exc: # noqa: BLE001 - CLI should print concise failures. print( "AMPG_ROUTE_MANIFEST " f"path={args.path} status=error message=\"{exc}\"", file=sys.stderr, ) return 1 for issue in issues: print( "AMPG_ROUTE_MANIFEST_ISSUE " f"path={args.path} json_path={issue.path} " f"code={issue.code} message=\"{issue.message}\"" ) print( "AMPG_ROUTE_MANIFEST " f"path={args.path} status=fail issues={len(issues)}" ) return 1 print( "AMPG_ROUTE_MANIFEST " f"path={args.path} schema={data['schema']} routes={len(data['routes'])} status=ok" ) return 0 return 1 def _print_route_exposure(exposure: RouteExposure) -> None: print( "AMPG_ROUTE " f"site={exposure.site_id} " f"protocol={exposure.protocol} " f"route_index={exposure.route_index} " f"route=\"{exposure.match}\" " f"source={exposure.source} " f"tier={exposure.tier} " f"identity={exposure.identity} " f"payments={exposure.payments} " f"realtime={str(exposure.realtime).lower()} " f"public_allowed={str(exposure.public_allowed).lower()} " f"max_tier={exposure.max_tier} " f"status={exposure.status} " f"reason=\"{exposure.reason}\"" ) def _print_route_issue(issue: RouteIssue) -> None: print( "AMPG_ROUTE_ISSUE " f"site={issue.site_id} " f"route_index={issue.route_index} " f"route=\"{issue.match}\" " f"source={issue.source} " f"tier={issue.tier} " f"code={issue.code} " f"message=\"{issue.message}\"" ) def _cmd_audit(config, *, fail_on_warn: bool) -> int: issues = audit_gateway(config) for issue in issues: print( "AMPG_AUDIT " f"site={issue.site_id} " f"severity={issue.severity} " f"code={issue.code} " f"path={issue.path} " f"message=\"{issue.message}\"" ) print(f"AMPG_AUDIT_SUMMARY issues={len(issues)}") if issues and fail_on_warn: return 1 return 0 def _cmd_docs(args) -> int: if args.docs_command == "generate": changed = generate_docs(Path.cwd(), check=args.check) changed_text = ",".join(str(path) for path in changed) if changed else "-" mode = "check" if args.check else "write" print(f"AMPG_DOCS status=ok mode={mode} changed={changed_text}") return 0 return 1 def _platform_override(config, args): if getattr(args, "platform", None): return platform_by_name(args.platform) profile = select_profile(config, getattr(args, "profile", None)) if profile and profile.platform: return platform_by_name(profile.platform) return None def _bool(value: bool) -> str: return str(value).lower() def _quote(value: str) -> str: return value.replace("\\", "\\\\").replace('"', '\\"')