"""Apply a first-build brief without rewriting Git history or later user work.""" import json import shutil import tempfile from pathlib import Path from uuid import UUID import structlog from src.domain.app_blocks.errors import UnsatisfiableAppBlockPlanError from src.domain.codegen.app_block_plan import assert_discovery_installable from src.domain.projects.discovery_repository import DiscoveryRepository from src.domain.projects.discovery_service import available_templates, require_discovery_access from src.workers.repo_ops import run_git logger = structlog.get_logger() # A failed/canceled/blocked first build is retryable and must not permanently prevent a # corrected Discovery plan from being applied. Active runs and successful/approved runs still # protect the starter from replacement. _RETRYABLE_CODEGEN_STATUSES = frozenset({"failed", "canceled", "blocked"}) def codegen_has_started(runs) -> bool: return any(getattr(run, "status", None) not in _RETRYABLE_CODEGEN_STATUSES for run in runs) _APPLY_FAILURE_MESSAGES = { "Local checkout has changes; preserve them before discovery retry": ( "discovery_dirty_checkout", "Couldn't prepare the repository because the local checkout has uncommitted changes. " "Retry discovery after project setup finishes, or contact support.", ), "Remote changed since setup; refusing to replace existing work": ( "discovery_remote_diverged", "The repository changed after setup, so Discovery won't replace it. " "Existing work was preserved.", ), "Cannot replace a starter after codegen has begun": ( "discovery_codegen_started", "A build already started for this project, so the starter can't be replaced.", ), "Template no longer available for this plan": ( "discovery_template_unavailable", "The selected template is no longer available on this plan. Retry discovery.", ), "Imported projects must preserve their source": ( "discovery_imported_locked", "Imported projects must keep their original stack. Existing work was preserved.", ), "Required stack changed": ( "discovery_locked_type", "The required stack no longer matches Discovery. Existing work was preserved.", ), "model_authored_endpoint": ( "discovery_compile_invalid", "Discovery compilation produced an invalid plan. Please retry; no build has started.", ), "Discovery owner no longer exists": ( "discovery_owner_missing", "The approving user is no longer available. Ask an owner to retry discovery.", ), } def _iter_exception_chain(exc: BaseException): seen: set[int] = set() stack: list[BaseException] = [exc] while stack: current = stack.pop() if id(current) in seen: continue seen.add(id(current)) yield current for linked in (current.__cause__, current.__context__, getattr(current, "orig", None)): if isinstance(linked, BaseException): stack.append(linked) def _classify_apply_failure(exc: BaseException) -> tuple[str, str]: if isinstance(exc, UnsatisfiableAppBlockPlanError): return ( "unsatisfiable_app_block_plan", "This first-build plan has a blocking gap. Adjust discovery answers and retry.", ) if isinstance(exc, ValueError): mapped = _APPLY_FAILURE_MESSAGES.get(str(exc)) if mapped: return mapped for item in _iter_exception_chain(exc): name = type(item).__name__ if name == "DeadlockDetectedError" or "deadlock detected" in str(item).lower(): return ( "discovery_apply_deadlock", "Couldn't prepare the repository because another setup step was still writing. " "Retry discovery; existing work was preserved.", ) name = type(exc).__name__ if name in {"TimeoutError", "APITimeoutError", "ReadTimeout", "ConnectTimeout"} or ( "timeout" in name.lower() ): return ( "discovery_compile_timeout", "Preparing the build plan took too long. Please retry; existing work was preserved.", ) if isinstance(exc, RuntimeError) and str(exc).startswith("Discovery Git operation failed"): return ( "discovery_git_failed", "Couldn't prepare the repository (Git step failed). Existing work was preserved. " "Retry discovery or check project setup.", ) return ( "discovery_apply_failed", "Couldn't prepare the repository. Existing work was preserved. Retry " "discovery or check project setup.", ) async def _persist_apply_failure(factory, pid: UUID, version: int, exc: BaseException) -> None: error_code, message = _classify_apply_failure(exc) logger.warning( "discovery_apply_failed", project_id=str(pid), version=version, error_code=error_code, exception_type=type(exc).__name__, ) async with factory() as db: project = await DiscoveryRepository(db).get(pid, lock=True) if project and (project.discovery_json or {}).get("version") == version: state = dict(project.discovery_json) if not state.get("initial_run_id") and state.get("status") == "applying": project.discovery_json = { **state, "status": "failed", "error": message, "error_code": error_code, } await db.commit() def _read_json(path: Path) -> dict | None: if not path.exists(): return None try: loaded = json.loads(path.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError, UnicodeDecodeError): return None return loaded if isinstance(loaded, dict) else None async def git(path, *args): rc, out, _err = await run_git(str(path), *args) if rc: # Authenticated remotes may occur in stderr. Never persist them. raise RuntimeError(f"Discovery Git operation failed: {args[0]}") return out.strip() def write_brief_files( root: Path, result: dict, *, source_kind: str = "generated", source_prompt: str = "", composition_recipe=None, ): assert_discovery_installable( {"result": result, "initial_prompt": source_prompt}, composition_recipe=composition_recipe, ) folder = root / "p2hs" folder.mkdir(parents=True, exist_ok=True) (folder / "project-brief.md").write_text(result["specification"], encoding="utf-8") if result.get("cloud_contract") is not None: (folder / "discovery-cloud-contract.json").write_text( json.dumps(result["cloud_contract"], indent=2), encoding="utf-8" ) from src.domain.app_blocks.hosts import host_map_payload, merge_host_map from src.domain.starter_library.catalog import ( contract_map_payload, merge_contract_map, ) target = folder / "starter-contracts.json" existing = _read_json(target) target.write_text( json.dumps(merge_contract_map(existing, contract_map_payload()), indent=2), encoding="utf-8", ) hosts_target = folder / "app-block-hosts.json" hosts_target.write_text( json.dumps(merge_host_map(_read_json(hosts_target), host_map_payload()), indent=2), encoding="utf-8", ) spec = result.get("spec") if spec: from src.domain.app_blocks.adoption import adoption_for, apply_adoption from src.domain.app_blocks.compiler import compile_install_plan from src.domain.codegen.app_sdk import ensure_required_packs_from_plan plan = apply_adoption( compile_install_plan( spec, source_prompt=source_prompt, composition_recipe=composition_recipe ), adoption_for(source_kind=source_kind, completed_codegen=False), ) (folder / "app-blocks-plan.json").write_text( json.dumps(plan.model_dump(), indent=2), encoding="utf-8" ) ensure_required_packs_from_plan(str(root)) if source_kind == "generated" and plan.starter_id: from src.domain.starter_library.assembler import assemble_starter_files framework = "nuxt_app" if (root / "nuxt.config.ts").exists() or (root / "nuxt.config.js").exists(): framework = "nuxt_app" elif (root / "package.json").exists() and (root / "src").is_dir(): framework = "react_app" elif (root / "index.html").exists(): framework = "html" for relative, content in assemble_starter_files( plan.starter_id, framework=framework ).items(): target = root / relative target.parent.mkdir(parents=True, exist_ok=True) target.write_text(content, encoding="utf-8") state_file = folder / "project-state.json" if not state_file.exists(): state_file.write_text( json.dumps( { "version": 1, "built_features": [], "key_files": [], "remaining_work": ["Implement the agreed first-build brief"], "verification": [], "note": "Starter prepared; application features are not built yet.", }, indent=2, ), encoding="utf-8", ) directions = ( "\n\n\n" "## Project context\n" "Read and maintain p2hs/project-state.json as the cumulative implementation brief: " "built_features, key_files, remaining_work, and verification actually performed. " "Retain completed features across later changes; never mark untested work as verified.\n" "Read p2hs/project-brief.md for the agreed first-build requirements. It" " describes desired work, not completed features.\n" "If present, p2hs/discovery-cloud-contract.json records the compiler's Cloud contract. " "Use its operation mappings as guidance, but the current run manifest takes precedence.\n" "If present, p2hs/app-blocks-plan.json is the Core App Block install plan. " "The validated composition is architecture authority. Do not reinterpret the " "customer's request into architecture. Use customer/recipe terminology for UI; " "use canonical platform identities for runtime behavior. " "When adoption is authoritative, mount those blocks and fill slots; preserve " "protected bindings. When adoption is advisory, keep working product and use " "the plan only for new work. When omitted, do not treat the plan as an install mandate. " "Honor installed, configuration_required, and unavailable/gap states: implement " "installed integrations; surface configuration without simulating a provider; " "do not substitute a gap. " "Do not invent a private backend for a capability the plan already represents " "or reports as a build-blocking gap.\n" "If present, p2hs/starter-contracts.json is the Starter Library contract map. " "Reference logical IDs (form.standard, record.table, record.list), " "not framework filenames. " "Prefer an admitted contract whose use_when matches; do not rebuild it. " "proposed ids are discovery-only until admitted.\n" "If present, p2hs/app-block-hosts.json maps App Block UI slots to host implementations. " "Mount those hosts; do not copy a new form or table when a host path is set.\n" "Use the current run's Project Context and PlatformBackendCapabilities " "for implemented revisions, remaining work, " "Cloud schema and enabled integrations. Core owns policy; Cloud " "executes through the platform BFF. " "Never implement a shadow identity/record database or put provider " "credentials in browser code. " "Preserve existing behavior; inspect the files relevant to this change and verify changes. " "Report completed changes, checks actually run, and unfinished requirements separately.\n" ) agents = root / "AGENTS.md" original = agents.read_text(encoding="utf-8") if agents.exists() else "" if "" not in original: agents.write_text(original + directions, encoding="utf-8") def apply_discovery_runtime_invariants(project, specification: str) -> None: """Discovery used to be a parallel contract that never fed ProjectRuntimeInvariants -- a user-named admin path established during discovery only ever existed as prose in project-brief.md, none of the "sticky, never downgraded by a later unrelated prompt" protection that system exists for (runtime_invariants.py's own docstring). Run the agreed specification through the same deterministic extractor codegen uses per-prompt, marked source="discovery" so it out-sticks a later ordinary prompt (extract_runtime_invariants never downgrades an existing source="discovery" back to "prompt"). Mutates `project.runtime_invariants` in place (matches this function's caller, which sets several other `project.*` fields the same way in the same transaction, before a single later `db.commit()`) -- does nothing if nothing new was actually extracted. """ from src.domain.projects.runtime_invariants import ( ProjectRuntimeInvariants, extract_runtime_invariants, validate_runtime_routes, ) previous = ProjectRuntimeInvariants.model_validate(project.runtime_invariants or {}) updated = extract_runtime_invariants(specification, previous) route_errors = validate_runtime_routes(updated) if route_errors: raise ValueError("Discovery produced invalid runtime routes: " + ", ".join(route_errors)) if updated != previous: updated.source = "discovery" project.runtime_invariants = updated.model_dump() async def _load_apply_context(factory, pid: UUID, version: int, settings): """Short locked read: validate apply is allowed and snapshot fields for Git/finalize.""" from src.domain.codegen.repository import CodegenRunRepository from src.domain.projects.paths import get_project_repo_path from src.domain.projects.repository import ProjectRepository async with factory() as db: project = await DiscoveryRepository(db).get(pid, lock=True) state = dict(project.discovery_json or {}) if project else {} if state.get("version") != version or state.get("status") not in { "applying", "completed", }: await db.rollback() return {"status": "skipped"} await require_discovery_access(db, project, UUID(state["approved_by"])) from src.domain.app_blocks.planner import source_prompt_from_discovery from src.domain.composition_recipes.compile import pinned_recipe_for_state assert_discovery_installable( state, composition_recipe=await pinned_recipe_for_state( db, state, source_prompt=source_prompt_from_discovery(state), project_type=(state.get("result") or {}).get("project_type"), ), ) if state.get("status") == "completed": await db.rollback() return {"status": "completed"} if project.status != "ready": # Setup's ready transaction will enqueue this approved version again. await db.rollback() return {"status": "waiting_setup"} # Holding the same project lock as run creation prevents simultaneous first builds. runs = await CodegenRunRepository(db).list_for_project(pid) if codegen_has_started(runs): raise ValueError("Cannot replace a starter after codegen has begun") linked = await ProjectRepository(db).get_with_repo(pid) repo = linked.repo result = dict(state["result"]) target_type = result["project_type"] templates = await available_templates(db, project.workspace_id) selected = next( ( t for t in templates if t.ref == result.get("template_ref") and t.framework == target_type ), None, ) if result.get("template_ref") and selected is None: raise ValueError("Template no longer available for this plan") imported = project.source_kind != "generated" if imported and (target_type != project.type or result.get("template_ref")): raise ValueError("Imported projects must preserve their source") if state.get("locked_type") and target_type != state["locked_type"]: raise ValueError("Required stack changed") from src.domain.app_blocks.planner import source_prompt_from_discovery from src.domain.composition_recipes.compile import pinned_recipe_for_state composition_recipe = await pinned_recipe_for_state( db, state, source_prompt=source_prompt_from_discovery(state), project_type=target_type, ) replace = not imported and ( target_type != project.type or result.get("template_ref") != project.template_ref ) snapshot = { "status": "ready", "state": state, "result": result, "needs_compile": not bool(result.get("cloud_contract")), "target_type": target_type, "replace": replace, "imported": imported, "workspace_id": project.workspace_id, "source_kind": project.source_kind, "project_ref": project.project_ref, "approved_by": UUID(state["approved_by"]), "application_id": state["application_id"], "initial_prompt": state.get("initial_prompt", ""), "default_branch": repo.default_branch or "main", "preview_branch": repo.preview_branch or (repo.default_branch or "main"), "repo_full_name": repo.repo_full_name, "initial_commit_sha": repo.initial_commit_sha, "allow_starter_replace": bool(state.get("reset_for_retry")), "composition_recipe": composition_recipe, "canonical": Path( get_project_repo_path(settings.repo_workspace, project.workspace_id, pid) ), "selected": ( { "id": selected.id, "ref": selected.ref, "github_repo": selected.github_repo, "starter_branch": selected.starter_branch, "starter_sha": selected.starter_sha, } if selected else None ), } await db.commit() return snapshot async def _compile_apply_result(factory, pid: UUID, snapshot: dict, settings) -> dict: """Compile outside any project row lock; persist cloud_contract back under a short lock.""" from src.domain.projects.discovery_compilation import attach_build_plan, load_discovery_contract from src.domain.projects.discovery_schemas import DiscoveryResult from src.integrations.project_intake_client import compile_discovery from src.workers.tasks.project_discovery import reserve_discovery_call async with factory() as db: contract = await load_discovery_contract(db, pid) await db.rollback() await reserve_discovery_call(settings) plan = await compile_discovery( {"brief": snapshot["result"], "cloud_contract": contract}, settings ) result = attach_build_plan( DiscoveryResult.model_validate(snapshot["result"]), plan, contract ) async with factory() as db: project = await DiscoveryRepository(db).get(pid, lock=True) state = dict(project.discovery_json or {}) if project else {} if state.get("version") != snapshot["state"]["version"] or state.get("status") != "applying": await db.rollback() return {"status": "skipped"} state["result"] = result project.discovery_json = state await db.commit() snapshot["result"] = result snapshot["state"] = {**snapshot["state"], "result": result} snapshot["needs_compile"] = False return snapshot async def _git_apply_discovery(factory, pid: UUID, snapshot: dict, settings) -> dict: """Clone/push discovery starter with no open DB transaction (avoids deadlocks with setup).""" from src.integrations.github_client import GitHubAppClient from src.integrations.github_installation import resolve_installation_id_for_workspace from src.workers.project_setup_pipeline import _materialize_initial_repo_tree from src.workers.repo_ops import ( _is_missing_remote_branch_error, clone_repo, prepare_working_tree, ) result = snapshot["result"] default = snapshot["default_branch"] preview = snapshot["preview_branch"] canonical = snapshot["canonical"] replace = snapshot["replace"] boilerplate_version = None if (canonical / ".git").exists(): dirty = await git(canonical, "status", "--porcelain") if dirty: # No codegen yet (guarded in load). Local drift from setup/env sync must not # block discovery apply — temp clone + push is the source of truth. logger.warning( "discovery_apply_resetting_dirty_checkout", project_id=str(pid), dirty_paths=len(dirty.splitlines()), ) await git(canonical, "reset", "--hard", "HEAD") await git(canonical, "clean", "-fd") installation = await resolve_installation_id_for_workspace(factory, snapshot["workspace_id"]) auth_url = await GitHubAppClient( installation_id=installation ).get_authenticated_clone_url(snapshot["repo_full_name"]) with tempfile.TemporaryDirectory(prefix="p2hs-discovery-") as tmp: checkout = Path(tmp) / "checkout" await clone_repo(auth_url, str(checkout), branch=default) preview_missing = False preview_history_realign = False if preview != default: fetch_rc, _, fetch_err = await run_git( str(checkout), "fetch", "origin", f"{preview}:refs/remotes/origin/{preview}" ) if fetch_rc: # Fresh repositories normally have only the default branch. # Discovery owns the initial starter commit, so defer creating # the managed preview branch until after that commit is pushed. if not _is_missing_remote_branch_error(fetch_err): raise RuntimeError("Discovery Git operation failed: fetch") preview_missing = True if not preview_missing: merge_rc, _, _ = await run_git( str(checkout), "merge-base", "--is-ancestor", f"origin/{preview}", "HEAD" ) if merge_rc: # A managed preview branch can have workflow-only commits # that make history diverge while leaving the actual app # tree unchanged. Preserve real content divergence, but # safely realign an identical tree during first discovery. tree_rc, _, _ = await run_git( str(checkout), "diff", "--quiet", f"origin/{preview}", "HEAD" ) if tree_rc: raise RuntimeError("Discovery Git operation failed: merge-base") preview_history_realign = True marker = checkout / ".p2hs-discovery.json" applied = json.loads(marker.read_text()) if marker.exists() else {} if ( applied.get("project_id") != str(pid) or applied.get("application_id") != snapshot["application_id"] ): head = await git(checkout, "rev-parse", "HEAD") if replace and head != snapshot["initial_commit_sha"] and not snapshot["allow_starter_replace"]: # Setup records the initial commit before follow-up automation can # add history-only commits (for example workflow/recovery markers). # Those commits must not make a fresh discovery retry look like # customer work. Keep the safety guard for any actual tree change. initial_sha = snapshot["initial_commit_sha"] tree_rc, _, _ = await run_git( str(checkout), "diff", "--quiet", initial_sha, "HEAD" ) if tree_rc: raise ValueError("Remote changed since setup; refusing to replace existing work") if replace and head != snapshot["initial_commit_sha"] and snapshot["allow_starter_replace"]: # An explicit Discovery reset authorizes replacing the abandoned first-build # starter. Preserve the old remote head on a recoverable branch before doing so. backup_branch = ( f"p2hs/discovery-backup-{str(pid).replace('-', '')[:12]}-v" f"{snapshot['state']['version']}" ) await git(checkout, "push", "origin", f"HEAD:refs/heads/{backup_branch}") if replace: starter = Path(tmp) / "starter" source = ( { "ref": snapshot["selected"]["ref"], "id": str(snapshot["selected"]["id"]), "github_repo": snapshot["selected"]["github_repo"], "starter_branch": snapshot["selected"]["starter_branch"], "starter_sha": snapshot["selected"]["starter_sha"], # Template records may not carry an installation id. # Use the workspace installation that already proved it # can access the project's repository instead of falling # back to a separate/global template App. "github_installation_id": installation, } if snapshot["selected"] else None ) from src.domain.projects.repository import ProjectRepository async with factory() as db: project_row = await ProjectRepository(db).get_by_id(pid) materialized = await _materialize_initial_repo_tree( str(starter), snapshot["target_type"], snapshot["project_ref"], settings, template_source=source, project_created_at=getattr(project_row, "created_at", None), db_factory=factory, project_id=pid, ) # Only Git-tracked initial files in this disposable clone are removed. await git(checkout, "rm", "-r", "--ignore-unmatch", "--", ".") shutil.copytree( starter, checkout, dirs_exist_ok=True, ignore=shutil.ignore_patterns(".git", "node_modules", ".env", ".env.local"), ) boilerplate_version = materialized.get("version") write_brief_files( checkout, result, source_kind=snapshot["source_kind"], source_prompt=snapshot["initial_prompt"], composition_recipe=snapshot.get("composition_recipe"), ) marker.write_text( json.dumps( {"project_id": str(pid), "application_id": snapshot["application_id"]} ), encoding="utf-8", ) from src.workers.repo_ops import stage_publishable_and_assert await stage_publishable_and_assert(str(checkout)) await git( checkout, "-c", "user.name=P2HS", "-c", "user.email=codegen@p2hs.com", "commit", "-m", "Prepare project from discovery", ) await git(checkout, "push", "origin", default) sha = await git(checkout, "rev-parse", "HEAD") if preview != default: if preview_missing or preview_history_realign: # A new project has no preview branch yet. Seed it from the # exact prepared default-branch commit so preview and main # start from the same discovery result. A history-only # divergence is realigned only after the tree equality check. await git(checkout, "checkout", "-B", preview, sha) push_args = ["push", "origin", preview] if preview_history_realign: push_args = ["push", "--force-with-lease", "origin", preview] await git(checkout, *push_args) else: await git(checkout, "fetch", "origin", f"{preview}:refs/remotes/origin/{preview}") await git(checkout, "checkout", "-B", preview, f"origin/{preview}") await git(checkout, "merge", "--ff-only", sha) await git(checkout, "push", "origin", preview) await prepare_working_tree(str(canonical), auth_url, branch=preview) # Brief / plan / AGENTS context are control-plane: keep them local for codegen, # never rely on customer Git (stage_publishable rejects p2hs/*). write_brief_files( canonical, result, source_kind=snapshot["source_kind"], source_prompt=snapshot["initial_prompt"], composition_recipe=snapshot.get("composition_recipe"), ) return {"sha": sha, "boilerplate_version": boilerplate_version} async def _finalize_discovery_apply( factory, pid: UUID, version: int, snapshot: dict, git_result: dict, settings ) -> None: """Short locked write: instructions, revision, project fields after Git succeeded.""" from src.domain.codegen.repository import CodegenRunRepository from src.domain.instructions.disk_mirror import mirror_instruction_to_meta from src.domain.instructions.repository import InstructionRepository from src.domain.projects.paths import get_project_meta_path from src.domain.revisions.repository import RevisionRepository sha = git_result["sha"] async with factory() as db: project = await DiscoveryRepository(db).get(pid, lock=True) state = dict(project.discovery_json or {}) if project else {} if state.get("version") != version or state.get("status") != "applying": await db.rollback() return await require_discovery_access(db, project, snapshot["approved_by"]) runs = await CodegenRunRepository(db).list_for_project(pid) if codegen_has_started(runs): raise ValueError("Cannot replace a starter after codegen has begun") result = snapshot["result"] # Prefer freshly compiled/persisted result if phase-1 compile wrote it. if isinstance(state.get("result"), dict) and state["result"].get("cloud_contract"): result = state["result"] instructions = InstructionRepository(db) current = await instructions.get_current_content(pid) content = current + "\n\n# Agreed project brief\n" + result["specification"] apply_discovery_runtime_invariants(project, result["specification"]) iv = await instructions.create_version(pid, content, snapshot["approved_by"]) await instructions.set_current(pid, iv.id) revision = await RevisionRepository(db).get_by_git_sha(pid, sha) if not revision: revision = await RevisionRepository(db).create( project_id=pid, git_sha=sha, source="project_setup", commit_message="Project discovery prepared", ) project.active_revision_id = revision.id project.type = snapshot["target_type"] project.name = result["product_name"] if git_result.get("boilerplate_version") is not None: project.boilerplate_version = git_result["boilerplate_version"] if not snapshot["imported"]: selected = snapshot["selected"] project.template_id = selected["id"] if selected else None project.template_ref = selected["ref"] if selected else None project.template_github_repo = selected["github_repo"] if selected else None project.template_starter_branch = selected["starter_branch"] if selected else None project.template_starter_sha = selected["starter_sha"] if selected else None project.discovery_json = { **state, "result": result, "status": "completed", "prepared_commit_sha": sha, "error": None, } await db.commit() mirror_instruction_to_meta(snapshot["workspace_id"], pid, content) meta = Path( get_project_meta_path(settings.repo_workspace, snapshot["workspace_id"], pid) ) meta.mkdir(parents=True, exist_ok=True) (meta / "project-discovery.json").write_text( json.dumps(result, indent=2), encoding="utf-8" ) async def apply_discovered_starter(ctx, pid: UUID, version: int): from src.config import get_settings from src.core.database import async_session_factory # ARQ test/integration contexts may include the key with a null value. # Treat that the same as an omitted override so worker jobs never attempt # to call None as a session factory. factory = ctx.get("db_factory") or async_session_factory settings = get_settings() try: # Phase 1: short project lock — validate + snapshot (no Git, no long I/O). snapshot = await _load_apply_context(factory, pid, version, settings) if snapshot["status"] == "skipped": return {"status": "skipped"} if snapshot["status"] == "completed": return await start_first_build(factory, pid) if snapshot["status"] == "waiting_setup": return {"status": "waiting_setup"} # Compile outside the project row lock (LLM / network). if snapshot.get("needs_compile"): compiled = await _compile_apply_result(factory, pid, snapshot, settings) if compiled.get("status") == "skipped": return {"status": "skipped"} snapshot = compiled # Phase 2: Git with no open DB transaction (setup can finish seeding revisions). git_result = await _git_apply_discovery(factory, pid, snapshot, settings) # Phase 3: short project lock — instructions + revision + project update. await _finalize_discovery_apply(factory, pid, version, snapshot, git_result, settings) return await start_first_build(factory, pid) except Exception as exc: await _persist_apply_failure(factory, pid, version, exc) raise async def start_first_build(factory, pid): from datetime import UTC, datetime from src.config import get_settings from src.domain.app_blocks.planner import source_prompt_from_discovery from src.domain.composition_recipes.compile import ( compile_pinned_install_plan, pinned_recipe_for_state, ) from src.domain.auth.models import User from src.domain.codegen.agent_facade import AgentCodegenFacade from src.domain.codegen.generation_context import ( build_codegen_request, resolve_generation_context, ) from src.domain.codegen.generation_plan import compile_generation_plan from src.domain.projects.discovery_schemas import DiscoveryResult, DiscoverySpec from src.domain.projects.form_requirement_sync import extract_delta_for_prompt from src.domain.projects.form_requirements import apply_form_requirement_delta from src.workers.enqueue import enqueue_codegen_run # Phase 1: extract the onboarding conversation's form-field intent, if any, # before taking the project row lock below -- this is a network call and must # not hold the lock while it runs, same reasoning as "Compile outside the # project row lock" elsewhere in this module. The result is re-merged against # a freshly re-read DiscoverySpec once the lock is held, so a race here can # only make the extractor's `known_forms` context slightly stale, never cause # a stale write. extraction = None async with factory() as db: peek = await DiscoveryRepository(db).get(pid) if peek is None: raise ValueError("Project not found") peek_state = dict(peek.discovery_json or {}) if not peek_state.get("initial_run_id"): raw_peek_spec = (peek_state.get("result") or {}).get("spec") if isinstance(raw_peek_spec, dict): peek_requirements = tuple(DiscoverySpec.model_validate(raw_peek_spec).form_requirements) extraction = await extract_delta_for_prompt( source_prompt_from_discovery(peek_state), peek_requirements, get_settings() ) async with factory() as db: project = await DiscoveryRepository(db).get(pid, lock=True) state = dict(project.discovery_json) await require_discovery_access(db, project, UUID(state["approved_by"])) assert_discovery_installable( state, composition_recipe=await pinned_recipe_for_state( db, state, source_prompt=source_prompt_from_discovery(state), project_type=(state.get("result") or {}).get("project_type"), ), ) if state.get("initial_run_id"): rid = state["initial_run_id"] await db.rollback() await enqueue_codegen_run(rid) return {"status": "completed", "run_id": rid} user = await db.get(User, UUID(state["approved_by"])) if not user: raise ValueError("Discovery owner no longer exists") discovery_result = DiscoveryResult.model_validate(state["result"]) # Phase 2: merge the extracted delta against the now-locked, freshly-read # spec with the pure merge helper, and persist it in the same transaction # that compiles the GenerationPlan below -- so GenerationPlan.form_definitions # and the persisted DiscoverySpec can never disagree about this build. if extraction is not None and extraction.applies: if extraction.confident and extraction.delta is not None: existing_requirements = tuple(discovery_result.spec.form_requirements) merged_requirements = apply_form_requirement_delta( existing_requirements, extraction.delta ) discovery_result.spec.form_requirements = list(merged_requirements) state["result"]["spec"] = discovery_result.spec.model_dump(mode="json") state.pop("form_requirement_sync", None) else: # Ambiguous: never guess the schema for a first build. Proceed with # whatever is already resolved (here: nothing yet, so the Forms # runtime's canonical default) rather than risk a mismatched custom # schema, and record the same visibility marker the later-prompt # path uses so FormsProvisioner.status() flags it for a clarifying # follow-up instead of silently dropping the request. state["form_requirement_sync"] = { "ambiguous": True, "reason": "first_build_extraction_ambiguous", "at": datetime.now(UTC).isoformat(), } project.discovery_json = state source_prompt = source_prompt_from_discovery(state) app_block_plan = await compile_pinned_install_plan( db, state, discovery_result.spec, source_prompt=source_prompt, project_type=discovery_result.project_type, ) generation_plan = compile_generation_plan( discovery_result, app_block_plan, project_id=pid, source_discovery_revision=state.get("version"), catalog_digests={"discovery": "2"}, ) legacy_prompt = ( "Build the approved project from the completed Discovery result.\n\n" + str(state["result"].get("summary") or "") ).strip() resolved_context = resolve_generation_context(generation_plan) generation_request = build_codegen_request( project_id=pid, plan=generation_plan, resolved_context=resolved_context, legacy_prompt=legacy_prompt, ) # Discovery completion is an internal orchestration boundary, not a chat turn. # The plan and its composition instructions stay in CodegenRun.plan; they must # never be persisted as a customer-visible Prompt or RunChatMessage. facade = AgentCodegenFacade(db) run = await facade.service.create_run( project_id=pid, user_prompt=legacy_prompt, created_by=user.id, attachment_ids=[UUID(x) for x in state.get("attachment_ids", [])], supersede_active=False, generation_plan=generation_plan.model_dump(mode="json"), generation_request=generation_request.model_dump(mode="json"), ) from src.domain.composition_recipes.installed_store import write_initial_snapshot await write_initial_snapshot(db, pid, generation_plan) await db.commit() await enqueue_codegen_run(str(run.id), project_id=str(pid)) logger.info( "discovery_codegen_enqueued", project_id=str(pid), run_id=str(run.id), handoff="discovery_plan", ) return {"status": "completed", "run_id": str(run.id)}