#!/usr/bin/env python3 """ Regenerate existing customer apps with the current passwordless auth shell. Preview branches only. Production is never touched. """ from __future__ import annotations import argparse import asyncio import json import os import sys from pathlib import Path from uuid import UUID ROOT = Path(__file__).resolve().parents[1] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from sqlalchemy import text from src.core.orm_import import import_all_orm_models import_all_orm_models() from src.core.database import async_session_factory from src.domain.codegen.repository import CodegenRunRepository from src.domain.codegen.router import _ProjectInfoFacadeForCodegen from src.domain.codegen.service import CodegenService from src.domain.manifest.repository import ManifestRunRepository from src.domain.manifest.service import ManifestService from src.domain.projects.repository import ( ProjectEnvVarRepository, ProjectSettingsRepository, ) from src.workers.enqueue import enqueue_codegen_run GOLDEN_PROJECTS = { "1b96aa25-ab11-4d3f-bd2d-b39b2a645584", "2311fa60-ac9e-40f7-9b29-b005f671b61c", } AUTH_REGEN_PROMPT = """ Refresh the existing generated application's authentication shell using the resolved Core runtime contract for this project. Preserve the existing product UI, routes, data model, customer-authored content, and business behavior. Modify only the generated authentication shell and the minimum wiring required for session-aware staff routing. Requirements: - Call me() from the installed P2HS client when the staff/admin route loads. - Use the host-only HttpOnly session cookies through the same-origin BFF. - Never read or persist session tokens in browser storage. - Render only authentication methods and controls admitted by the resolved runtime contract. - Use the exact runtime_contract.routes.admin_panel_path and runtime_contract.routes.customer_login_path. Never invent, rename, or substitute route paths; do not create span.html, button.html, or alternate auth routes. - Do not expose or hardcode an administrator email before authentication. - Do not add local users, local authentication, or routes for capabilities not admitted by the runtime contract. - Preserve the project's resolved auth mode and do not infer a different one. - Preview branch only; never modify production. """ async def load_projects(selected: list[str] | None = None) -> list[str]: if selected: return [project_id for project_id in selected if project_id not in GOLDEN_PROJECTS] async with async_session_factory() as db: rows = await db.execute( text( """ SELECT id::text FROM projects WHERE deleted_at IS NULL AND status IN ('ready', 'creating') AND type IN ('html', 'react_app', 'nuxt_app') AND COALESCE(source_kind::text, 'generated') = 'generated' AND id::text <> ALL(:golden) ORDER BY id """ ), {"golden": list(GOLDEN_PROJECTS)}, ) return [row[0] for row in rows] async def project_owner(db, project_id: UUID) -> UUID | None: row = await db.execute( text(""" SELECT w.owner_id FROM projects p JOIN workspaces w ON w.id = p.workspace_id JOIN workspace_memberships wm ON wm.workspace_id = w.id AND wm.user_id = w.owner_id WHERE p.id = :project_id AND w.owner_id IS NOT NULL LIMIT 1 """), {"project_id": str(project_id)}, ) value = row.scalar_one_or_none() return UUID(str(value)) if value else None async def regenerate_project(project_id: str, retry_failed: bool = False) -> dict: pid = UUID(project_id) async with async_session_factory() as db: service = CodegenService( codegen_repo=CodegenRunRepository(db), project_facade=_ProjectInfoFacadeForCodegen(db), project_settings_repo=ProjectSettingsRepository(db), env_var_repo=ProjectEnvVarRepository(db), manifest_service=ManifestService( db=db, run_repo=ManifestRunRepository(db), ), ) owner_id = await project_owner(db, pid) if not owner_id: return { "project_id": project_id, "status": "error", "error": "workspace owner is not an active workspace member", } active = await CodegenRunRepository(db).get_active_reserving_run(pid) if active: return { "project_id": project_id, "status": "skipped_active_run", "run_id": str(active.id), "active_status": active.status, } if retry_failed: repo = CodegenRunRepository(db) failed = await repo.list_for_project(pid) candidates = [ run for run in failed if run.status in {"failed", "blocked", "canceled"} and ( (run.plan or {}).get("context", {}).get("prompt_source") == "phase5e5-passwordless-auth-shell-refresh" or "No module named 'src'" in (run.error_message or "") ) ] if candidates: source = sorted(candidates, key=lambda run: run.created_at)[-1] source.created_by = owner_id await db.flush() refreshed = await service.retry_run( source.id, pid, created_by=owner_id, ) await db.commit() queued = await enqueue_codegen_run(str(refreshed.id), project_id=project_id) return { "project_id": project_id, "run_id": str(refreshed.id), "status": "queued" if queued else "enqueue_failed", "mode": "retry_existing", "created_by": str(owner_id), "preview_only": True, } return { "project_id": project_id, "status": "no_retryable_run", "mode": "retry_existing", "created_by": str(owner_id), "preview_only": True, } run = await service.create_run( project_id=pid, user_prompt=AUTH_REGEN_PROMPT, created_by=owner_id, run_mode="agent", execution_target="cloud", supersede_active=False, prompt_source="phase5e5-passwordless-auth-shell-refresh", ) await db.commit() run_id = str(run.id) queued = await enqueue_codegen_run(run_id, project_id=project_id) return { "project_id": project_id, "run_id": run_id, "status": "queued" if queued else "enqueue_failed", "created_by": str(owner_id), "mode": "new_run", "preview_only": True, } async def main(apply: bool, selected: list[str], retry_failed: bool) -> None: projects = await load_projects(selected) print(json.dumps({ "candidate_count": len(projects), "golden_excluded": sorted(GOLDEN_PROJECTS), "preview_only": True, "apply": apply, "retry_failed": retry_failed, }, indent=2)) if not apply: print("\nDry run only. Re-run with --apply to queue regeneration.") for project_id in projects: print(project_id) return log_dir = Path("phase5e5-auth-regeneration-logs") log_dir.mkdir(exist_ok=True) results = [] for project_id in projects: try: result = await regenerate_project(project_id, retry_failed=retry_failed) except Exception as exc: result = { "project_id": project_id, "status": "error", "error": str(exc), } results.append(result) print(json.dumps(result, default=str), flush=True) (log_dir / f"{project_id}.json").write_text( json.dumps(result, indent=2, default=str) + "\n", encoding="utf-8", ) summary = { "candidate_count": len(projects), "queued": sum(r.get("status") == "queued" for r in results), "skipped_active_run": sum( r.get("status") == "skipped_active_run" for r in results ), "enqueue_failed": sum( r.get("status") == "enqueue_failed" for r in results ), "errors": sum(r.get("status") == "error" for r in results), } print("\nSUMMARY") print(json.dumps(summary, indent=2)) if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument( "--apply", action="store_true", help="Actually queue Core codegen runs", ) parser.add_argument( "--project-id", action="append", default=[], help="Limit the operation to this project; repeat for multiple projects", ) parser.add_argument( "--retry-failed", action="store_true", help="Resume the latest failed/blocked run from this phase instead of creating a new run", ) args = parser.parse_args() asyncio.run(main(args.apply, args.project_id, args.retry_failed))