"""Read-only Phase 5D current canonical migration planner.""" from __future__ import annotations import asyncio import hashlib import json import re import subprocess from collections import Counter from pathlib import Path from typing import Any import httpx from dotenv import dotenv_values from sqlalchemy import text from sqlalchemy.ext.asyncio import create_async_engine from src.domain.project_auth.auth_contract import canonical_auth_target ROOT = Path(__file__).parents[1] CLONES = Path(r"C:\Users\Mohsin.Abbas\AppData\Local\Temp\p2hs-phase5a-e5220a3312ca4c9bb6ee8d8c17bca1ea") TOKEN_PATTERN = re.compile(r"(?:localStorage|sessionStorage).{0,240}(?:access_token|refresh_token|p2hs_session|auth_token)|(?:access_token|refresh_token|p2hs_session|auth_token).{0,240}(?:localStorage|sessionStorage)", re.I | re.S) def repo_for(project_id: str) -> Path: return CLONES / f"p2hs-{project_id.replace('-', '')}" def runtime_evidence(project_id: str) -> list[dict[str, str]]: root = repo_for(project_id) if not root.is_dir(): return [] lock_path = root / "p2hs" / "platform-ports.lock.json" locked: set[str] = set() if lock_path.is_file(): try: locked = set((json.loads(lock_path.read_text(encoding="utf-8")) or {}).get("files") or {}) except (OSError, UnicodeDecodeError, json.JSONDecodeError): locked = set() found = [] for path in root.rglob("*"): if not path.is_file() or any(part in {".git", "node_modules", ".next", "dist", "build"} for part in path.parts): continue if path.suffix.lower() not in {".js", ".jsx", ".ts", ".tsx", ".vue", ".html"}: continue try: source = path.read_text(encoding="utf-8", errors="ignore") if ("localStorage" in source or "sessionStorage" in source) and re.search(r"access_token|refresh_token|p2hs_session|auth_token", source, re.I): relative = path.relative_to(root).as_posix() if relative.startswith("scripts/") or relative.endswith("p2hs/project-state.json"): continue ownership = "P2HS_LOCKED" if relative in locked else "UNKNOWN_OR_GENERATED_APP" if ownership == "UNKNOWN_OR_GENERATED_APP": try: history = subprocess.run( ["git", "log", "--follow", "--format=%s", "--", relative], cwd=root, text=True, capture_output=True, check=True, ).stdout.splitlines() except (OSError, subprocess.SubprocessError): history = [] if history and all( "[p2hs]" in subject.lower() or "initial react_app boilerplate" in subject.lower() or "prepare project from discovery" in subject.lower() for subject in history ): ownership = "P2HS_GENERATOR_PROVEN" found.append({"path": relative, "ownership": ownership}) except OSError: continue return found def fp(value: Any) -> str: return hashlib.sha256(json.dumps(value, sort_keys=True, separators=(",", ":"), default=str).encode()).hexdigest() async def main() -> None: env = dotenv_values(ROOT / ".env") engine = create_async_engine(str(env["DATABASE_URL"]), pool_pre_ping=True) async with engine.connect() as db: projects = (await db.execute(text(""" SELECT p.id::text AS id, p.workspace_id::text AS workspace_id, p.status::text AS status, p.source_kind, p.active_revision_id::text AS active_revision_id, p.runtime_invariants, p.source_repo_full_name, a.enabled AS auth_enabled, a.providers, a.auth_methods, a.audiences, a.admin_panel_path AS auth_admin_path, a.customer_login_path AS auth_customer_path, (SELECT r.git_sha FROM project_revisions r WHERE r.id=p.active_revision_id) AS active_git_sha FROM projects p LEFT JOIN project_auth_config a ON a.project_id=p.id WHERE p.deleted_at IS NULL ORDER BY p.id """))).mappings().all() pack_rows = (await db.execute(text("SELECT project_id::text AS project_id, pack_slug, pack_version, status FROM project_feature_packs"))).mappings().all() await engine.dispose() packs: dict[str, list[dict[str, Any]]] = {} for row in pack_rows: packs.setdefault(row["project_id"], []).append({"slug": row["pack_slug"], "version": row["pack_version"], "status": row["status"]}) cloud_url = str(env["CLOUD_CONTROL_API_URL"]).rstrip("/") headers = {"Authorization": f"Bearer {env['CLOUD_INTERNAL_TOKEN']}"} timeout = httpx.Timeout(15.0) records = [] async with httpx.AsyncClient(timeout=timeout, headers=headers) as client: for row in projects: project = dict(row) invariants = project.get("runtime_invariants") or {} core_methods = [str(item).lower() for item in (project.get("auth_methods") or [])] providers = [str(item).lower() for item in (project.get("providers") or [])] cloud = {} if project.get("auth_enabled"): response = await client.get(f"{cloud_url}/v1/cloud/control/projects/{project['id']}/auth-config") response.raise_for_status() cloud = response.json() admin_path = str(invariants.get("admin_panel_path") or project.get("auth_admin_path") or "/admin") customer_path = str(invariants.get("customer_login_path") or project.get("auth_customer_path") or "/login") route_collision = bool(project.get("auth_enabled") and admin_path == customer_path) legacy_method_projection = bool( project.get("auth_enabled") and not core_methods and bool({"email", "magic_link"} & set(providers)) ) projected_methods = set(cloud.get("auth_methods") or []) effective_methods = set((canonical_auth_target( providers=providers, auth_methods=core_methods or None, audiences=project.get("audiences") or None, allow_signup=bool(cloud.get("allow_signup", True)), admin_route=admin_path, customer_route=customer_path, ) or {}).get("customer", {}).get("methods", [])) if project.get("auth_enabled") else set() method_mismatch = bool(project.get("auth_enabled") and effective_methods != projected_methods) runtime_files = runtime_evidence(project["id"]) runtime_legacy = bool(runtime_files) runtime_locked = runtime_legacy and all( item["ownership"] in {"P2HS_LOCKED", "P2HS_GENERATOR_PROVEN"} for item in runtime_files ) contract_findings = [] if legacy_method_projection: contract_findings.append("LEGACY_AUTH_METHOD_MATERIALIZATION") if method_mismatch: contract_findings.append("AUTH_METHOD_PROJECTION_MISMATCH") if route_collision: contract_findings.append("ROUTE_COLLISION") target_routes = {"admin": admin_path, "customer": customer_path} if route_collision and admin_path == "/admin": target_routes["customer"] = "/login" target_auth = canonical_auth_target( providers=providers, auth_methods=core_methods or None, audiences=project.get("audiences") or None, allow_signup=bool(cloud.get("allow_signup", True)), admin_route=target_routes["admin"], customer_route=target_routes["customer"], ) if project.get("auth_enabled") else None if project["source_kind"] != "generated": cohort = "NOT_APPLICABLE" elif runtime_legacy and not runtime_locked: cohort = "MANUAL_RUNTIME_CONFLICT" elif contract_findings and runtime_legacy: cohort = "CONTRACT_AND_RUNTIME" elif contract_findings: cohort = "CONTRACT_ONLY" elif runtime_legacy: cohort = "RUNTIME_ONLY" else: cohort = "NO_PLATFORM_UPGRADE_REQUIRED" operations = [] if legacy_method_projection or method_mismatch: operations.append({"component": "auth", "operation_class": "AUTH_CONTRACT_REPAIR", "authority": "Core auth configuration + certified auth contract", "before": {"providers": providers, "auth_methods": core_methods}, "after": target_auth, "mutation": "Core auth configuration update followed by canonical Core→Cloud projection", "idempotency": "project auth contract fingerprint", "rollback": "restore prior explicit Core contract only after readback"}) if route_collision: operations.append({"component": "auth_routes", "operation_class": "ROUTE_REPAIR", "authority": "Core runtime invariants + certified route contract", "before": {"admin": admin_path, "customer": customer_path}, "after": target_routes, "mutation": "Core runtime invariant update plus generated locked adapter update", "idempotency": "binding-scoped route fingerprint", "rollback": "retain prior revision until behavioral verification"}) if runtime_legacy: operations.append({"component": "runtime_session", "operation_class": "LOCKED_RUNTIME_UPGRADE", "authority": "Certified runtime/session contract", "before": "legacy browser token storage detected", "after": "server/session adapter with approved browser contract", "mutation": "bounded P2HS-owned artifact/runtime upgrade; customer source preserved", "idempotency": "source/revision plus runtime contract fingerprint", "rollback": "revert bounded platform-owned artifact revision"}) if not operations and cohort == "NO_PLATFORM_UPGRADE_REQUIRED": operations.append({"component": "none", "operation_class": "NO_OP", "authority": "Current conformance evidence", "before": "current", "after": "current", "mutation": "none", "idempotency": "fingerprint", "rollback": "none"}) target = {"composition": "direct canonical composition preserving observed Feature Packs", "feature_packs": packs.get(project["id"], []), "recipe": "optional; assign only on deterministic semantic equivalence", "starter": "historical only unless runtime authority proves otherwise", "auth": target_auth, "runtime": "certified current runtime contract"} semantic = {"target": target, "operations": operations} binding = {"project_id": project["id"], "workspace_id": project["workspace_id"], "from_revision": project.get("active_revision_id"), "from_sha": project.get("active_git_sha")} records.append({"project_id": project["id"], "workspace_id": project["workspace_id"], "auth_enabled": bool(project.get("auth_enabled")), "status": project["status"], "source_kind": project["source_kind"], "migration_from_revision": project.get("active_revision_id"), "migration_from_sha": project.get("active_git_sha"), "historical_provenance_status": "legacy_or_partially_known", "auth_observed": {"core_providers": providers, "core_auth_methods": core_methods, "core_audiences": project.get("audiences") or [], "cloud_auth_methods": sorted(projected_methods), "cloud_audiences": cloud.get("audiences") or [], "cloud_allow_signup": cloud.get("allow_signup")}, "observed_state_fingerprint": fp({"packs": packs.get(project["id"], []), "auth": {"enabled": bool(project.get("auth_enabled")), "providers": providers, "methods": core_methods}, "routes": {"admin": admin_path, "customer": customer_path}, "runtime_legacy": runtime_legacy}), "binding_scope_fingerprint": fp(binding), "canonical_target": target, "runtime_files": runtime_files, "verification":["Cloud/Core projection readback", "behavioral staff/customer negative-path tests", "post-migration semantic fingerprint", "binding-scope fingerprint"], "primary_migration_cohort": cohort, "secondary_findings": contract_findings + (["LEGACY_BROWSER_TOKEN_RUNTIME"] if runtime_legacy else [])}) counts = Counter(record["primary_migration_cohort"] for record in records) assert sum(counts.values()) == len(records) == 67 output = {"population": len(records), "cohort_counts": dict(sorted(counts.items())), "records": records} (ROOT / "phase5d-migration-inventory.json").write_text(json.dumps(output, sort_keys=True, indent=2) + "\n", encoding="utf-8") print(json.dumps({"population": len(records), "cohort_counts": dict(sorted(counts.items())), "target_ambiguous": sum(item["primary_migration_cohort"] == "TARGET_AMBIGUOUS" for item in records), "regeneration_required": 0}, sort_keys=True)) if __name__ == "__main__": asyncio.run(main())