from __future__ import annotations import json import subprocess from pathlib import Path from typing import Iterable from app.core.config import settings DOMAIN_API_ROOT = Path(__file__).resolve().parents[2] WORKSPACE_ROOT = DOMAIN_API_ROOT.parent _REGISTERED_ROUTE_PATHS: set[str] = set() def remember_registered_route_paths(route_paths: Iterable[str] | None) -> None: global _REGISTERED_ROUTE_PATHS normalized = { str(item).strip() for item in list(route_paths or []) if str(item).strip() } _REGISTERED_ROUTE_PATHS = normalized def _read_json(path: Path) -> dict: try: return json.loads(path.read_text(encoding="utf-8")) except Exception: return {} def _discover_manifest_candidate() -> tuple[Path | None, str]: configured = str(settings.build_manifest_path or "").strip() if configured: path = Path(configured) if path.exists(): return path, "configured_manifest" root_manifest = WORKSPACE_ROOT / "release_manifest.json" if root_manifest.exists(): return root_manifest, "release_manifest" latest_release = WORKSPACE_ROOT / "release" / "latest_release.json" if latest_release.exists(): return latest_release, "latest_release" return None, "" def _build_info_from_env() -> dict | None: if not any( [ str(settings.build_commit_sha or "").strip(), str(settings.build_commit_ref or "").strip(), str(settings.build_generated_at or "").strip(), str(settings.build_package_name or "").strip(), str(settings.build_checksum or "").strip(), ] ): return None return { "source": str(settings.build_source_label or "").strip() or "env", "package_name": str(settings.build_package_name or "").strip(), "generated_at": str(settings.build_generated_at or "").strip(), "commit_sha": str(settings.build_commit_sha or "").strip(), "commit_ref": str(settings.build_commit_ref or "").strip(), "checksum": str(settings.build_checksum or "").strip(), "manifest_path": str(settings.build_manifest_path or "").strip(), } def _build_info_from_manifest() -> dict | None: manifest_path, source = _discover_manifest_candidate() if manifest_path is None: return None payload = _read_json(manifest_path) if not payload: return { "source": source, "package_name": "", "generated_at": "", "commit_sha": "", "commit_ref": "", "checksum": "", "manifest_path": str(manifest_path), } checksum = str(payload.get("checksum") or payload.get("sha256") or "").strip() return { "source": source, "package_name": str(payload.get("package_name") or "").strip(), "generated_at": str(payload.get("generated_at") or "").strip(), "commit_sha": str(payload.get("commit_sha") or "").strip(), "commit_ref": str(payload.get("commit_ref") or "").strip(), "checksum": checksum, "manifest_path": str(manifest_path), } def _build_info_from_git() -> dict | None: try: inside_worktree = subprocess.run( ["git", "-C", str(WORKSPACE_ROOT), "rev-parse", "--is-inside-work-tree"], check=False, capture_output=True, text=True, timeout=2, ) except Exception: return None if inside_worktree.returncode != 0 or str(inside_worktree.stdout or "").strip() != "true": return None def _read_git(*args: str) -> str: try: result = subprocess.run( ["git", "-C", str(WORKSPACE_ROOT), *args], check=False, capture_output=True, text=True, timeout=2, ) except Exception: return "" if result.returncode != 0: return "" return str(result.stdout or "").strip() return { "source": "git", "package_name": "", "generated_at": "", "commit_sha": _read_git("rev-parse", "--short=12", "HEAD"), "commit_ref": _read_git("rev-parse", "--abbrev-ref", "HEAD"), "checksum": "", "manifest_path": "", } def _resolve_build_identity() -> dict: return ( _build_info_from_env() or _build_info_from_manifest() or _build_info_from_git() or { "source": "unknown", "package_name": "", "generated_at": "", "commit_sha": "", "commit_ref": "", "checksum": "", "manifest_path": "", } ) def _expected_route_paths() -> dict[str, str]: prefix = str(settings.api_prefix or "/api/v1").rstrip("/") return { "ops_contracts": f"{prefix}/ops/contracts", "ops_contract_detail": f"{prefix}/ops/contracts/{{contract_key}}", "ops_stack_diagnosis": f"{prefix}/ops/stack-diagnosis", "ops_migration_source_profile": f"{prefix}/ops/migration/source-profile", "ops_migration_preview": f"{prefix}/ops/migration/preview", "ops_migration_execute": f"{prefix}/ops/migration/execute", "ops_node_handover": f"{prefix}/ops/nodes/{{node_code}}/handover", "ops_node_onboarding": f"{prefix}/ops/nodes/{{node_code}}/onboarding", "ops_node_onboarding_bootstrap_preview": f"{prefix}/ops/nodes/{{node_code}}/onboarding/bootstrap/preview", "ops_node_onboarding_bootstrap_execute": f"{prefix}/ops/nodes/{{node_code}}/onboarding/bootstrap/execute", "ops_node_onboarding_acceptance_preview": f"{prefix}/ops/nodes/{{node_code}}/onboarding/acceptance/preview", "ops_node_onboarding_acceptance_execute": f"{prefix}/ops/nodes/{{node_code}}/onboarding/acceptance/execute", "ops_node_onboarding_recovery_preview": f"{prefix}/ops/nodes/{{node_code}}/onboarding/recovery/preview", "ops_node_onboarding_recovery_execute": f"{prefix}/ops/nodes/{{node_code}}/onboarding/recovery/execute", "ops_node_scene_log": f"{prefix}/ops/nodes/{{node_code}}/scene-log", "ops_node_handover_bootstrap_plan": f"{prefix}/ops/nodes/{{node_code}}/handover/bootstrap-plan", "ops_driver_feed": f"{prefix}/ops/driver-feed", "ops_codex_brief": f"{prefix}/ops/codex-brief", "ops_activity_stream": f"{prefix}/ops/activity-stream", "ops_runbook_resolve": f"{prefix}/ops/runbook/sequences/{{sequence_key}}/resolve", "ops_runbook_execute": f"{prefix}/ops/runbook/sequences/{{sequence_key}}/execute", "runtime_build_info": f"{prefix}/runtime/build-info", } def _read_text(path: Path) -> str: try: return path.read_text(encoding="utf-8", errors="replace") except Exception: return "" def _repository_capabilities() -> dict: ops_agent_service_text = _read_text(DOMAIN_API_ROOT / "app" / "services" / "ops_agent_service.py") bootstrap_node_agent_text = _read_text(DOMAIN_API_ROOT / "deploy" / "multi-region" / "bootstrap_node_agent.sh") return { "supports_install_command_block": ( "install_command_block" in ops_agent_service_text and "_candidate_node_agent_install_paths" in ops_agent_service_text ), "supports_multi_layout_bootstrap": "resolve_project_root" in bootstrap_node_agent_text, } def _build_route_surface(route_paths: Iterable[str] | None = None) -> dict: expected = _expected_route_paths() normalized_paths = { str(item).strip() for item in (list(route_paths) if route_paths is not None else list(_REGISTERED_ROUTE_PATHS)) if str(item).strip() } mode = "registered" if normalized_paths else "declared_contract" flags = {key: (path in normalized_paths if normalized_paths else True) for key, path in expected.items()} missing_keys = [key for key, available in flags.items() if not available] return { "mode": mode, "registered_paths_total": len(normalized_paths), "surface_flags": flags, "expected_paths": expected, "missing_keys": missing_keys, "missing_paths": [expected[key] for key in missing_keys], "surface_complete": len(missing_keys) == 0, } def get_runtime_build_info(route_paths: Iterable[str] | None = None) -> dict: build = _resolve_build_identity() route_surface = _build_route_surface(route_paths=route_paths) return { "source": str(build.get("source") or "").strip() or "unknown", "package_name": str(build.get("package_name") or "").strip(), "generated_at": str(build.get("generated_at") or "").strip(), "commit_sha": str(build.get("commit_sha") or "").strip(), "commit_ref": str(build.get("commit_ref") or "").strip(), "checksum": str(build.get("checksum") or "").strip(), "manifest_path": str(build.get("manifest_path") or "").strip(), "workspace_root": str(WORKSPACE_ROOT), "domain_api_root": str(DOMAIN_API_ROOT), "repository_capabilities": _repository_capabilities(), "route_surface": route_surface, }