"""Task: Run AI code generation.

Uses **short DB sessions** around DB work only. Clone / Cursor / git push run **without** holding a
pool connection so the API stays responsive on a shared Postgres host.

After a successful ``git push``, a durable checkpoint is written so a retry can finish DB updates
and preview enqueue without re-running the provider.

run_mode values:
  "agent"       — standard code generation (default)
  "plan"        — read-only: generate markdown plan, set status=plan_ready, no git push
  "visual_edit" — agent mode enriched with selected_elements context from the frontend

When ``p2hs/required-user-inputs.json`` lists required values, the run ends as
``user_input_required`` (not ``completed``) until the owner continues via ``POST …/chat``
with the same ``run_id`` (or legacy supply-user-inputs on that run).
"""
from __future__ import annotations

import asyncio
import json
import os
import re
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from math import ceil
from time import monotonic
from typing import Any, Optional
from uuid import UUID

import structlog

from src.config import AIProvider, get_settings
from src.domain.codegen.app_block_plan import assert_discovery_installable
from src.domain.projects.paths import get_project_meta_path

logger = structlog.get_logger()

REPO_BASED_PROVIDERS = (AIProvider.CURSOR, AIProvider.CLAUDE, AIProvider.CODEX)

# Cap stored transcript size (Postgres text + API payloads); tail preserved for debugging.
_AGENT_TRANSCRIPT_MAX_CHARS = 500_000
_INTERNAL_CODEGEN_PATH_PREFIXES = (".p2hs/ai-context/", ".p2hs-control/")
_INTERNAL_CODEGEN_PATHS = frozenset(
    {
        ".cursorrules",
        ".p2hs.json",
        ".p2hs/local-result.json",
        ".p2hs/ai-context/system-prompt.md",
        ".p2hs/ai-context/last-run.json",
        "p2hs/project-brief.md",
        "p2hs/app-blocks-plan.json",
        "p2hs/discovery-cloud-contract.json",
        "p2hs/project-state.json",
        "CLAUDE.md",
    }
)


def _cap_agent_transcript(lines: list[str]) -> Optional[str]:
    if not lines:
        return None
    raw = "\n".join(lines)
    if len(raw) <= _AGENT_TRANSCRIPT_MAX_CHARS:
        return raw
    return "...[truncated beginning]\n" + raw[-_AGENT_TRANSCRIPT_MAX_CHARS:]


def _normalize_repo_relative_path(path: Optional[str]) -> str:
    normalized = (path or "").replace("\\", "/").strip()
    if normalized.startswith("./"):
        normalized = normalized[2:]
    return normalized


def _is_internal_codegen_path(path: Optional[str]) -> bool:
    normalized = _normalize_repo_relative_path(path)
    if not normalized:
        return False
    if normalized in _INTERNAL_CODEGEN_PATHS:
        return True
    return any(normalized.startswith(prefix) for prefix in _INTERNAL_CODEGEN_PATH_PREFIXES)


def _filter_user_visible_changes(
    files_changed: Optional[list[str]],
    files_changed_detail: Optional[list[dict]],
) -> tuple[Optional[list[str]], Optional[list[dict]], bool]:
    from src.integrations.platform_ports import is_platform_projection_path

    def hidden(path: Optional[str]) -> bool:
        return _is_internal_codegen_path(path) or is_platform_projection_path(path)

    filtered_files = [
        _normalize_repo_relative_path(path)
        for path in (files_changed or [])
        if not hidden(path)
    ]
    filtered_detail = [
        {**entry, "path": _normalize_repo_relative_path(entry.get("path"))}
        for entry in (files_changed_detail or [])
        if not hidden(entry.get("path"))
    ]
    had_reported_changes = bool((files_changed or []) or (files_changed_detail or []))
    return (
        filtered_files or None,
        filtered_detail or None,
        had_reported_changes,
    )


def _manifest_requires_user_input(pu: Any) -> bool:
    """True when ``pending_user_inputs`` has at least one manifest item (blocking state)."""
    return bool(
        pu
        and isinstance(pu, dict)
        and isinstance(pu.get("items"), list)
        and len(pu["items"]) > 0
    )


def _placeholder_for_user_input_item(item: dict[str, Any]) -> str:
    """Return a safe non-secret placeholder for an owner-supplied config value."""
    haystack = " ".join(
        str(item.get(k, ""))
        for k in ("id", "key", "label", "title", "description", "placeholder", "type", "kind")
    ).lower()
    if "phone" in haystack or "tel" in haystack:
        return "+1 555 123 4567"
    if "email" in haystack:
        return "hello@example.com"
    if "reservation" in haystack or "booking" in haystack:
        return "https://example.com/reservations"
    if "url" in haystack or "link" in haystack or "website" in haystack:
        return "https://example.com"
    if "address" in haystack or "location" in haystack:
        return "123 Main Street"
    if "hour" in haystack or "schedule" in haystack:
        return "Mon-Fri 9:00 AM-5:00 PM"
    if "key" in haystack or "token" in haystack or "secret" in haystack:
        return "replace-me"
    return "Update me"


def _placeholder_values_for_manifest(manifest: Any) -> dict[str, str]:
    if not _manifest_requires_user_input(manifest):
        return {}
    out: dict[str, str] = {}
    for item in manifest.get("items") or []:
        if not isinstance(item, dict):
            continue
        key = str(item.get("id") or item.get("key") or "").strip()
        if key:
            out[key] = _placeholder_for_user_input_item(item)
    return out


def _remaining_pending_user_input_items(
    manifest: Any,
    *,
    builder_user_values: Optional[dict[str, Any]] = None,
    env_var_keys: Optional[set[str]] = None,
) -> list[dict[str, Any]]:
    if not manifest or not isinstance(manifest, dict):
        return []
    items = manifest.get("items")
    if not isinstance(items, list) or not items:
        return []
    buv = builder_user_values or {}
    known_env_keys = {str(k).strip() for k in (env_var_keys or set()) if str(k).strip()}
    remaining: list[dict[str, Any]] = []
    for item in items:
        if not isinstance(item, dict):
            continue
        key = str(item.get("id") or item.get("key") or "").strip()
        if not key:
            continue
        if str(buv.get(key, "")).strip():
            continue
        if key in known_env_keys:
            continue
        remaining.append(item)
    return remaining


async def _persist_placeholder_user_inputs(
    *,
    db_factory,
    project_id: UUID,
    manifest: Any,
) -> dict[str, str]:
    """Store placeholder config in project settings/env vars so codegen never blocks."""
    from src.domain.codegen.service import _fan_out_inputs_to_env_vars
    from src.domain.projects.repository import ProjectEnvVarRepository, ProjectSettingsRepository

    async with db_factory() as db:
        settings_repo = ProjectSettingsRepository(db)
        env_var_repo = ProjectEnvVarRepository(db)
        settings = await settings_repo.get_or_create(project_id)
        builder_user_values = (settings.settings or {}).get("builder_user_values") or {}
        env_var_keys = {
            str(row.key).strip()
            for row in await env_var_repo.list_for_project(project_id)
            if getattr(row, "key", None)
        }
        remaining = _remaining_pending_user_input_items(
            manifest,
            builder_user_values=builder_user_values,
            env_var_keys=env_var_keys,
        )
        if not remaining:
            return {}
        values = _placeholder_values_for_manifest({**dict(manifest or {}), "items": remaining})
        if not values:
            return {}
        await settings_repo.merge_builder_user_values(project_id, values, overwrite=False)
        await _fan_out_inputs_to_env_vars(
            env_var_repo=env_var_repo,
            project_id=project_id,
            values=values,
            manifest_items=remaining,
            overwrite=False,
        )
        await db.commit()

    try:
        from src.workers.tasks.env_sync import enqueue_env_sync

        await enqueue_env_sync(project_id)
    except Exception as exc:
        logger.warning(
            "codegen_placeholder_env_sync_enqueue_failed",
            project_id=str(project_id),
            error=str(exc),
        )

    return values


def _placeholder_user_input_message(values: dict[str, str]) -> str:
    if not values:
        return ""
    keys = ", ".join(sorted(values))
    return (
        "I used temporary placeholder values for "
        f"{keys} so the build could keep moving. "
        "Update them in Project Variables when you have the real details."
    )


def _capability_gap_message(
    blocked_packs: list[dict], gaps: list[dict], pending_configs: list[dict] | None = None
) -> str:
    """A primitive_gap or blocked pack install must be visible to the owner, not just
    buried in a timeline log entry -- a run that silently completes despite one is the
    "fake success" this whole composition system exists to avoid. Pure and simple on
    purpose (same convention as _placeholder_user_input_message): the status change
    alone (_select_final_status) isn't enough, the owner needs to see *what* wasn't
    fulfilled and why.

    `pending_configs` (status="configuration_required") is reported separately and
    deliberately not framed as "skipped" -- the feature WAS built; only its connector
    connection or webhook secret is outstanding. Conflating the two would recreate the
    exact "missing credentials looks like a build failure" problem this status exists
    to fix."""
    pending_configs = pending_configs or []
    if not blocked_packs and not gaps and not pending_configs:
        return ""
    sections: list[str] = []
    if blocked_packs or gaps:
        lines: list[str] = []
        for o in blocked_packs:
            errors = "; ".join(o.get("errors") or ["blocked"])
            lines.append(f"- Feature pack '{o.get('pack_slug')}' could not be installed: {errors}")
        for g in gaps:
            criticality = g.get("criticality", "required")
            if g.get("status") == "invalid_manifest":
                errors = "; ".join(g.get("errors") or ["invalid composition declaration"])
                lines.append(
                    f"- '{g.get('name')}' has an invalid platform declaration: {errors} "
                    f"-- marked {criticality}."
                )
            elif g.get("status") != "primitive_gap" and not g.get("missing"):
                errors = "; ".join(g.get("errors") or [g.get("status", "composition failed")])
                lines.append(
                    f"- '{g.get('name')}' could not be composed: {errors} "
                    f"-- marked {criticality}."
                )
            else:
                missing = ", ".join(g.get("missing") or [])
                lines.append(
                    f"- '{g.get('name')}' needs a platform capability that doesn't exist yet "
                    f"({missing}) -- marked {criticality}, so it was skipped rather than faked."
                )
        header = (
            "Note: some requested capabilities couldn't be fully applied and were "
            "skipped rather than faked:"
        )
        sections.append(header + "\n" + "\n".join(lines))
    if pending_configs:
        lines = []
        for o in pending_configs:
            blocker = o.get("blocker") or {}
            connections = list(blocker.get("pending_connections") or [])
            webhooks = [w.get("provider") for w in (blocker.get("webhooks_pending_setup") or []) if w.get("provider")]
            needs = ", ".join(connections + webhooks) or "a connector"
            lines.append(
                f"- Action required: '{o.get('pack_slug')}' was built but can't run yet -- "
                f"connect {needs} in Project Settings -> Integrations to activate it."
            )
        header = "Some features were built but are waiting on configuration:"
        sections.append(header + "\n" + "\n".join(lines))
    return "\n\n".join(sections)


def _select_final_status(
    blocking: bool,
    blocked_packs: list[dict],
    gaps: list[dict],
    pending_configs: list[dict] | None = None,
) -> str:
    """Chooses the run's terminal status. `blocking` (pending_user_inputs) always
    takes priority -- the run is waiting on the owner, not yet evaluable as
    complete/incomplete. Otherwise: any blocked pack install, or any *required*
    primitive_gap, means a requirement genuinely wasn't fulfilled -- "blocked", never
    plain "completed". Only *optional* gaps, or packs sitting at
    "configuration_required" (built, just not connected to a provider yet), remain ->
    "completed_with_limitations": everything required was delivered, but say so
    honestly rather than silently. A configuration_required pack never contributes to
    `blocked_packs` in the first place (see service.py's install_pack) -- it's ok=True.

    Note on naming: by the time this runs, the code has already been committed and
    pushed (_run_repo_based_codegen attaches/pushes before _process_pack_installs/
    _process_pack_compositions ever run -- see the Level 2 follow-up plan's Part A
    for the traced call order). "blocked" describes the capability's status, not that
    the run was prevented from shipping.
    """
    if blocking:
        return "user_input_required"
    required_gap = bool(blocked_packs) or any(
        g.get("criticality", "required") == "required" for g in gaps
    )
    if required_gap:
        return "blocked"
    if gaps or pending_configs:
        return "completed_with_limitations"
    return "completed"


async def _maybe_persist_cursor_agent_session(
    db_factory,
    project_id: UUID,
    settings: Any,
    result: Any,
) -> None:
    """Store Cursor CLI chat id when the agent emits one (enables ``--resume`` on the next run)."""
    if settings.ai_provider != AIProvider.CURSOR:
        return
    if getattr(result, "provider_invocation_failed", False):
        return
    sid = getattr(result, "cursor_session_id", None)
    if not sid or not str(sid).strip():
        return
    from src.domain.projects.repository import ProjectSettingsRepository

    async with db_factory() as db:
        await ProjectSettingsRepository(db).set_cursor_agent_session_id(
            project_id, str(sid).strip()
        )
        await db.commit()


async def _process_fn_deploys(
    *,
    db_factory,
    project_id: UUID,
    entries: list[dict],
) -> list[dict]:
    """Deploy library functions declared in p2hs/functions.json.

    Non-blocking: each entry is attempted independently; failures are captured
    and returned so the caller can write a timeline event, but they never fail
    the codegen run.
    """
    if not entries:
        return []

    from src.domain.connectors.repository import ConnectorRepository
    from src.domain.edge_functions.repository import (
        FunctionLibraryRepository,
        ProjectFunctionRepository,
        ProjectFunctionTokenRepository,
        ProjectFunctionVersionRepository,
        ProjectWebhookEndpointRepository,
        WorkerRuntimeVersionRepository,
    )
    from src.domain.edge_functions.service import EdgeFunctionsService
    from src.domain.projects.repository import ProjectRepository
    from src.integrations.cloud_control_client import (
        CloudControlClient,
        get_cloud_client_for_workspace,
    )

    outcomes: list[dict] = []
    async with db_factory() as db:
        project = await ProjectRepository(db).get_by_id(project_id)
        if project:
            cloud_client = await get_cloud_client_for_workspace(project.workspace_id, db)
        else:
            cloud_client = CloudControlClient()
        svc = EdgeFunctionsService(
            library_repo=FunctionLibraryRepository(db),
            fn_repo=ProjectFunctionRepository(db),
            version_repo=ProjectFunctionVersionRepository(db),
            webhook_repo=ProjectWebhookEndpointRepository(db),
            runtime_repo=WorkerRuntimeVersionRepository(db),
            token_repo=ProjectFunctionTokenRepository(db),
            project_repo=ProjectRepository(db),
            connector_repo=ConnectorRepository(db),
            cloud_client=cloud_client,
        )
        for entry in entries:
            proj_slug = entry["project_slug"]
            try:
                await svc.deploy_from_library_by_slug(
                    project_id=project_id,
                    library_slug=entry["library_slug"],
                    project_slug=proj_slug,
                    trigger_config=entry.get("trigger_config") or {},
                    access_policy=entry.get("access_policy", "authenticated"),
                    secret_refs=entry.get("secret_refs") or [],
                    created_by="codegen",
                )
                outcomes.append({"slug": proj_slug, "ok": True})
                logger.info(
                    "codegen_fn_deployed",
                    project_id=str(project_id),
                    slug=proj_slug,
                    library_slug=entry["library_slug"],
                )
            except Exception as exc:
                outcomes.append({"slug": proj_slug, "ok": False, "error": str(exc)[:200]})
                logger.warning(
                    "codegen_fn_deploy_failed",
                    project_id=str(project_id),
                    slug=proj_slug,
                    library_slug=entry["library_slug"],
                    error=str(exc),
                )
        await db.commit()
    return outcomes


async def _attach_fn_deploys_if_present(result: Any, repo_path: str) -> Any:
    """If ``p2hs/functions.json`` exists in the checkout, copy the validated deploy list onto the result."""
    from dataclasses import replace as dc_replace
    from src.domain.codegen.functions_manifest import load_manifest_from_repo_path

    entries = await asyncio.to_thread(load_manifest_from_repo_path, repo_path)
    if not entries:
        return result
    return dc_replace(result, pending_fn_deploys=entries)


async def _process_pack_installs(
    *,
    db_factory,
    project_id: UUID,
    entries: list[dict],
    user_prompt: str = "",
    run_id: UUID | None = None,
    sync_requirements: bool = True,
) -> list[dict]:
    """Install feature packs from the agent declaration union the App Block plan.

    Non-blocking: each pack is attempted independently; failures are captured and
    returned so the caller can write a timeline event, but they never fail the codegen run.
    """
    from src.domain.app_blocks.pack_install import merge_pack_install_entries
    from src.domain.codegen.app_block_plan import resolve_pack_policy
    from src.domain.connectors.repository import ConnectorRepository
    from src.domain.edge_functions.repository import (
        FunctionLibraryRepository,
        ProjectFunctionRepository,
        ProjectFunctionTokenRepository,
        ProjectFunctionVersionRepository,
        ProjectWebhookEndpointRepository,
        WorkerRuntimeVersionRepository,
    )
    from src.domain.edge_functions.service import EdgeFunctionsService
    from src.domain.feature_packs.service import PackInstallService
    from src.domain.projects.repository import ProjectRepository
    from src.integrations.cloud_control_client import (
        CloudControlClient,
        get_cloud_client_for_workspace,
    )

    outcomes: list[dict] = []
    agent_entries = list(entries or [])
    skip_uninstall = not agent_entries
    async with db_factory() as db:
        # Composition reconcile already synced requirements and stored this
        # run's plan. Repeat the sync only when that pass did not run.
        if sync_requirements:
            try:
                from src.domain.projects.form_requirement_sync import (
                    sync_form_requirements_from_prompt,
                )

                await sync_form_requirements_from_prompt(
                    db, project_id, user_prompt, get_settings()
                )
            except Exception:
                logger.warning(
                    "forms.requirement_sync_failed", project_id=str(project_id), exc_info=True
                )
            try:
                from src.domain.projects.entity_requirement_sync import (
                    sync_entity_field_requirements_from_prompt,
                )

                await sync_entity_field_requirements_from_prompt(
                    db, project_id, user_prompt, get_settings()
                )
            except Exception:
                logger.warning(
                    "entities.requirement_sync_failed", project_id=str(project_id), exc_info=True
                )
        _, install_slugs, protect_slugs = await resolve_pack_policy(
            db, project_id, run_id=run_id
        )
        entries = merge_pack_install_entries(agent_entries, install_slugs=install_slugs)
        protect = set(protect_slugs)
        project = await ProjectRepository(db).get_by_id(project_id)
        if project:
            cloud_client = await get_cloud_client_for_workspace(project.workspace_id, db)
        else:
            cloud_client = CloudControlClient()
        edge_svc = EdgeFunctionsService(
            library_repo=FunctionLibraryRepository(db),
            fn_repo=ProjectFunctionRepository(db),
            version_repo=ProjectFunctionVersionRepository(db),
            webhook_repo=ProjectWebhookEndpointRepository(db),
            runtime_repo=WorkerRuntimeVersionRepository(db),
            token_repo=ProjectFunctionTokenRepository(db),
            project_repo=ProjectRepository(db),
            connector_repo=ConnectorRepository(db),
            cloud_client=cloud_client,
        )
        pack_svc = PackInstallService(db, edge_svc, cloud_client)
        declared_slugs = {entry["pack_slug"] for entry in entries}
        for entry in entries:
            outcome = await pack_svc.install_pack(
                project_id=project_id,
                pack_slug=entry["pack_slug"],
                config=entry.get("config") or {},
                user_prompt=user_prompt,
            )
            outcomes.append(outcome)
            logger.info(
                "codegen_pack_install_attempted",
                project_id=str(project_id),
                pack_slug=entry["pack_slug"],
                ok=outcome["ok"],
                errors=outcome.get("errors") or None,
            )
            if not outcome["ok"]:
                logger.error(
                    "codegen_pack_install_blocked",
                    project_id=str(project_id),
                    pack_slug=entry["pack_slug"],
                    errors=outcome.get("errors") or ["blocked"],
                )

        # Agent-declared install is a full set, but App Block plan packs are protected.
        from sqlalchemy import select as _select

        from src.domain.feature_packs.models import ProjectFeaturePack

        if not skip_uninstall:
            installed = await db.execute(
                _select(ProjectFeaturePack.pack_slug).where(
                    ProjectFeaturePack.project_id == project_id,
                    ProjectFeaturePack.status == "installed",
                )
            )
            for (installed_slug,) in installed.all():
                if installed_slug in declared_slugs or installed_slug in protect:
                    continue
                outcome = await pack_svc.uninstall_pack(
                    project_id=project_id, pack_slug=installed_slug
                )
                outcomes.append(outcome)
                logger.info(
                    "codegen_pack_uninstall_attempted",
                    project_id=str(project_id),
                    pack_slug=installed_slug,
                    ok=outcome["ok"],
                )
        await db.commit()
    return outcomes


async def _process_pack_compositions(
    *,
    db_factory,
    project_id: UUID,
    entries: list,
    user_prompt: str = "",
    run_id: UUID | None = None,
) -> list[dict]:
    """Level 1 capability composition: for each CompositionRequirement no existing
    catalog pack already covers, attempt to compose+install a project_local pack.

    Same non-blocking posture as _process_pack_installs: one entry's failure or
    primitive_gap never blocks another, and never fails the run.

    Raw manifest entries (the GitHub Actions path) are AI-written intent, not a trusted
    contract: each is normalized and safely repaired (manifest_repair.py), the original,
    the repaired entry and every repair are kept on the run when ``run_id`` is given, and
    only an entry that is still invalid or ambiguous after repair is reported invalid.
    """
    if not entries:
        return []

    from src.domain.feature_packs.composer import compose_and_register_pack
    from src.domain.feature_packs.manifest_repair import repair_composition
    from src.domain.feature_packs.requirements_spec import CompositionRequirement
    from src.domain.projects.repository import ProjectRepository
    from src.integrations.cloud_control_client import (
        CloudControlClient,
        get_cloud_client_for_workspace,
    )

    outcomes: list[dict] = []
    async with db_factory() as db:
        project = await ProjectRepository(db).get_by_id(project_id)
        if not project or not project.project_ref:
            return []
        assert_discovery_installable(project.discovery_json)
        cloud_client = await get_cloud_client_for_workspace(project.workspace_id, db)
        for entry in entries:
            # The local codegen path already hands over validated CompositionRequirement
            # objects, but the GitHub Actions path forwards the raw `compose` JSON from
            # p2hs/feature-packs.json (request_provisioning -> arq job args). That file is
            # generated by the coding agent, so it is treated as intent: repaired where the
            # repair is deterministic and meaning-preserving, otherwise reported field by
            # field. Each entry is handled independently: one bad entry never drops the rest.
            repaired = None
            if isinstance(entry, CompositionRequirement):
                req = entry
            else:
                repaired = repair_composition(entry)
                req = repaired.requirement
                if req is None:
                    name = entry.get("name") if isinstance(entry, dict) else None
                    shown = repaired.repaired if isinstance(repaired.repaired, dict) else {}
                    outcome = {
                        "name": name,
                        "ok": False,
                        "status": "invalid_manifest",
                        "criticality": (
                            shown.get("criticality")
                            if shown.get("criticality") in {"required", "optional"}
                            else "required"
                        ),
                        "errors": [
                            f"invalid_composition_requirement:{e['path']}: {e['message']}"
                            for e in repaired.errors
                        ],
                        "error_details": repaired.errors,
                        "repairs": repaired.repairs,
                        "renamed": repaired.renames,
                        "_manifest": {"name": name, **repaired.record()},
                    }
                    outcomes.append(outcome)
                    logger.warning(
                        "codegen_pack_composition_failed",
                        project_id=str(project_id),
                        name=name,
                        errors=[e["path"] + ":" + e["code"] for e in repaired.errors],
                    )
                    continue
            # Core-resolved entity fields win over whatever this run's `compose` entry
            # declares for a field Core already has an opinion on; a never-before-seen
            # field is adopted from the declaration. See
            # src.domain.projects.entity_requirement_sync.resolve_entity_fields.
            try:
                from src.domain.codegen.provision_manifest import ProvisionField
                from src.domain.projects.entity_requirement_sync import resolve_entity_fields

                resolved_fields = await resolve_entity_fields(
                    db, project_id, req.name, [f.model_dump(mode="json") for f in req.fields]
                )
                req = req.model_copy(
                    update={"fields": [ProvisionField.model_validate(f) for f in resolved_fields]}
                )
            except Exception:
                logger.warning(
                    "entities.field_resolution_failed",
                    project_id=str(project_id),
                    name=req.name,
                    exc_info=True,
                )
            outcome = await compose_and_register_pack(
                db, cloud_client, project_id, project.project_ref, req, user_prompt
            )
            if repaired is not None:
                # Light copy travels with the outcome; the full record is retained below.
                outcome["repairs"] = repaired.repairs
                if repaired.renames:
                    outcome["renamed"] = repaired.renames
                # Core's canonical form of what the agent asked for: Core, not the repo
                # file, is authoritative for pack identity, fields, actors and criticality.
                canonical = req.model_dump(mode="json", exclude_none=True)
                outcome["canonical"] = canonical
                outcome["_manifest"] = {
                    "name": req.name,
                    **repaired.record(),
                    "canonical_manifest": canonical,
                }
            outcomes.append(outcome)
            if outcome.get("status") == "primitive_gap":
                logger.info(
                    "codegen_pack_composition_primitive_gap",
                    project_id=str(project_id),
                    name=outcome.get("name"),
                    missing=outcome.get("missing"),
                )
            elif not outcome.get("ok"):
                logger.warning(
                    "codegen_pack_composition_failed",
                    project_id=str(project_id),
                    name=outcome.get("name"),
                    errors=outcome.get("errors"),
                )
            else:
                logger.info(
                    "codegen_pack_composition_installed",
                    project_id=str(project_id),
                    name=outcome.get("name"),
                    pack_slug=outcome.get("pack_slug"),
                )
        await db.commit()
    records = [o.pop("_manifest") for o in outcomes if "_manifest" in o]
    if records and run_id is not None:
        await _retain_composition_manifests(db_factory, run_id, records)
    return outcomes


_MANIFEST_HISTORY_LIMIT = 5


def _correction_from_outcomes(blocked_packs: list, gaps: list) -> dict | None:
    """Structured errors for the coding agent, but only when EVERY blocking problem is an invalid
    manifest entry it can fix. A real primitive gap or a blocked pack cannot be corrected by
    rewriting a file, so mixing them in would burn a correction attempt for nothing."""
    if blocked_packs:
        return None
    blocking = [g for g in gaps if g.get("criticality", "required") == "required"]
    if not blocking or any(g.get("status") != "invalid_manifest" for g in blocking):
        return None
    from src.domain.codegen.work_branch import bounded_correction

    return bounded_correction(
        {
            "entries": [
                {
                    "name": g.get("name"),
                    "errors": g.get("error_details") or [],
                    "renamed": g.get("renamed") or [],
                }
                for g in blocking
            ]
        }
    )


def _composition_repair_message(outcomes: list[dict]) -> str | None:
    """Tell the coding agent/owner when Core normalized a capability declaration.

    A rename matters: the generated frontend may already call the old entity or field
    name, so the canonical name has to be surfaced, not silently substituted."""
    renames = [
        (o.get("name"), r)
        for o in outcomes or []
        for r in o.get("renamed") or []
    ]
    repaired = sum(1 for o in outcomes or [] if o.get("repairs"))
    if not renames and not repaired:
        return None
    lines = []
    if renames:
        lines.append(
            "Note: the platform normalized these capability names. Use the canonical name "
            "everywhere the app refers to it (forms, API calls, records):"
        )
        for name, rename in renames[:8]:
            lines.append(f"- {rename['path']}: '{rename['from']}' -> '{rename['to']}'")
        if len(renames) > 8:
            lines.append(f"- ... and {len(renames) - 8} more")
    elif repaired:
        lines.append(
            f"Note: {repaired} capability declaration(s) were normalized by the platform "
            "(defaults made explicit or cosmetic fields removed); no names changed."
        )
    return "\n".join(lines)


async def _retain_composition_manifests(db_factory, run_id: UUID, records: list[dict]) -> None:
    """Keep what the coding agent wrote (original), what Core made of it (repaired) and
    every repair on the run itself: an audit trail, and the exact input for telling the
    agent what changed or what to fix. Best-effort: never fails the provisioning gate.
    Bounded history, one entry per gate attempt."""
    from src.domain.codegen.repository import CodegenRunRepository

    try:
        async with db_factory() as db:
            run = await CodegenRunRepository(db).get_by_id(run_id)
            if not run:
                return
            plan = dict(run.plan or {})
            history = list(plan.get("composition_manifests") or [])
            history.append({"at": datetime.now(timezone.utc).isoformat(), "entries": records})
            plan["composition_manifests"] = history[-_MANIFEST_HISTORY_LIMIT:]
            run.plan = plan
            await db.commit()
    except Exception as exc:
        logger.warning(
            "composition_manifest_retention_failed",
            run_id=str(run_id),
            error_type=type(exc).__name__,
        )


async def _attach_pack_installs_if_present(result: Any, repo_path: str) -> Any:
    """If ``p2hs/feature-packs.json`` exists in the checkout, copy the validated install list onto the result."""
    from dataclasses import replace as dc_replace
    from src.domain.codegen.feature_packs_manifest import load_manifest_from_repo_path

    entries = await asyncio.to_thread(load_manifest_from_repo_path, repo_path)
    if not entries:
        return result
    return dc_replace(result, pending_pack_installs=entries)


async def _attach_pack_compositions_if_present(result: Any, repo_path: str) -> Any:
    """If p2hs/feature-packs.json has a `compose` array, copy its raw entries onto the result.
    They are repaired and validated in _process_pack_compositions (manifest_repair.py), the
    same as on the GitHub Actions path, so an invalid entry is reported, never dropped."""
    from dataclasses import replace as dc_replace
    from src.domain.codegen.feature_packs_manifest import load_compositions_from_repo_path

    entries = await asyncio.to_thread(load_compositions_from_repo_path, repo_path)
    if not entries:
        return result
    return dc_replace(result, pending_pack_compositions=entries)


async def _attach_provision_manifest_if_present(result: Any, repo_path: str) -> Any:
    """If ``p2hs/provision.json`` exists in the checkout, copy the validated manifest onto the result."""
    from dataclasses import replace as dc_replace
    from src.domain.codegen.provision_manifest import load_manifest_from_repo_path
    from src.domain.codegen.backend_guard import validate_backend_repo

    await asyncio.to_thread(validate_backend_repo, repo_path)
    manifest = await asyncio.to_thread(load_manifest_from_repo_path, repo_path)
    if not manifest:
        return result
    return dc_replace(result, pending_provision=manifest)


async def _attach_email_templates_manifest_if_present(result: Any, repo_path: str) -> Any:
    """If ``p2hs/email-templates.json`` exists in the checkout, copy the validated manifest
    onto the result. Unlike provision.json, an invalid file here is never a hard failure for
    the whole run — see email_templates_manifest.load_manifest_from_repo_path, which raises
    on malformed JSON/schema; that exception is allowed to propagate here deliberately (a
    template edit the agent got wrong should be visible, not silently dropped), but nothing
    downstream blocks push/deploy on it the way provisioning does."""
    from dataclasses import replace as dc_replace
    from src.domain.codegen.email_templates_manifest import load_manifest_from_repo_path

    manifest = await asyncio.to_thread(load_manifest_from_repo_path, repo_path)
    if not manifest:
        return result
    return dc_replace(result, pending_email_templates=manifest)


async def _apply_cloud_cutover_checkout(*, db_factory, project_id: UUID, repo_path: str) -> None:
    """Import leftover local users and force Cloud record/auth BFF adapters."""
    from src.domain.codegen.legacy_cloud_cutover import cut_over_generated_repo
    from src.domain.projects.repository import ProjectRepository

    async with db_factory() as db:
        project = await ProjectRepository(db).get_by_id(project_id)
        if not project:
            return
        await cut_over_generated_repo(
            db=db,
            project=project,
            repo_path=repo_path,
            import_dry_run=False,
            adapter_dry_run=False,
            ensure_auth=True,
        )
        await db.commit()


async def _process_provision_manifest(
    *,
    db_factory,
    project_id: UUID,
    manifest: dict,
) -> list[dict]:
    """Required resources must succeed before push, preview, or deployment."""
    from src.domain.codegen.provision_manifest import (
        installed_pack_entity_keys,
        provision_entities,
        validate_manifest,
    )
    from src.domain.project_storage.repository import ProjectStorageBucketRepository, ProjectStorageUsageRepository
    from src.domain.project_storage.schemas import CreateBucketRequest
    from src.domain.project_storage.service import ProjectStorageService
    from src.domain.projects.repository import ProjectRepository
    from src.integrations.cloud_control_client import get_cloud_client_for_workspace

    declaration = validate_manifest(manifest)
    if not declaration["entities"] and not declaration["storage_buckets"]:
        return []
    async with db_factory() as db:
        project = await ProjectRepository(db).get_by_id(project_id)
        if not project or project.status == "archived":
            raise ValueError("Project is unavailable for provisioning")
        # Project ID comes from the authorized run, never from the declaration.
        from src.domain.billing.access_policy import BillingAccessPolicy
        from src.domain.billing.repository import SubscriptionRepository
        from src.domain.workspaces.repository import WorkspaceRepository
        await BillingAccessPolicy(SubscriptionRepository(db), WorkspaceRepository(db)).require_active_billing(project.workspace_id)
        client = await get_cloud_client_for_workspace(project.workspace_id, db)
        if not client._is_configured():
            raise ValueError("Cloud is not configured for resource provisioning")
        from src.domain.projects.cloud_schema_catalog import (
            ProjectCloudSchemaSnapshotRepository,
        )
        snapshot_repo = ProjectCloudSchemaSnapshotRepository(db)
        known = await snapshot_repo.get_entities(project_id)
        # Core-resolved entity fields win over whatever this run's p2hs/provision.json
        # declares for a field Core already has an opinion on; a never-before-seen
        # field is adopted from the declaration. See
        # src.domain.projects.entity_requirement_sync.resolve_entity_fields.
        for entity_declaration in declaration["entities"]:
            try:
                from src.domain.projects.entity_requirement_sync import resolve_entity_fields

                entity_declaration["fields"] = await resolve_entity_fields(
                    db, project_id, entity_declaration["key"], entity_declaration["fields"]
                )
            except Exception:
                logger.warning(
                    "entities.field_resolution_failed",
                    project_id=str(project_id),
                    entity_key=entity_declaration.get("key"),
                    exc_info=True,
                )
        outcomes = await provision_entities(
            client,
            str(project_id),
            declaration["entities"],
            known_entities=known,
            protected_entity_keys=await installed_pack_entity_keys(db, project_id),
        )
        repo = ProjectStorageBucketRepository(db)
        service = ProjectStorageService(
            bucket_repo=repo, usage_repo=ProjectStorageUsageRepository(db), cloud_client=client
        )
        for entry in declaration["storage_buckets"]:
            existing = await repo.get_by_name(project_id, entry["name"])
            if existing:
                if (not existing.is_active or existing.access_policy != entry["access_policy"]
                        or existing.rls_enabled != entry["rls_enabled"]):
                    raise ValueError(f"Incompatible bucket {entry['name']}")
                # A Core row alone is not proof that a prior Cloud sync succeeded.
                await service.sync_buckets_to_cloud(project_id)
            else:
                await service.create_bucket(project_id, CreateBucketRequest(**entry))
            outcomes.append({"bucket": entry["name"], "ok": True, "action": "existing" if existing else "created"})
        try:
            live = (await client.get_project_schema(str(project_id))).get("entities") or []
            if live:
                from src.domain.codegen.entity_catalog import compact_entities
                await snapshot_repo.upsert(project_id, compact_entities(live))
        except Exception:
            pass
        await db.commit()
    return outcomes


async def _process_email_templates_manifest(
    *,
    db_factory,
    project_id: UUID,
    declaration: dict,
) -> list[dict]:
    """Never blocks or fails the run — an unapplied/invalid override just means the
    platform default copy keeps being used. Unlike provisioning, nothing else depends
    on this existing before the app is usable."""
    from src.domain.codegen.email_templates_manifest import apply_email_templates
    from src.domain.projects.repository import ProjectRepository
    from src.integrations.cloud_control_client import get_cloud_client_for_workspace

    if not declaration:
        return []
    try:
        async with db_factory() as db:
            project = await ProjectRepository(db).get_by_id(project_id)
            if not project or project.status == "archived":
                return [{"ok": False, "error": "project_unavailable"}]
            client = await get_cloud_client_for_workspace(project.workspace_id, db)
            if not client._is_configured():
                return [{"ok": False, "error": "cloud_not_configured"}]
            return await apply_email_templates(client, str(project_id), declaration)
    except Exception as exc:
        logger.warning("email_templates_apply_failed", project_id=str(project_id), error=str(exc))
        return [{"ok": False, "error": str(exc)[:200]}]


async def resolve_codegen_capabilities(
    ctx: dict,
    run_id: str,
    manifest: dict,
    pack_installs: list | None = None,
    pack_compositions: list | None = None,
) -> dict:
    """Actions pre-push gate. All external work stays on an arq worker.

    Originally provision.json-only (named provision_codegen_resources); now also
    resolves p2hs/feature-packs.json's install/compose entries here, before the
    GitHub Actions runner's own push -- see provision_gate.py's PrimitiveGapError for
    why a required gap is raised rather than returned (arq's result_info() only
    distinguishes success/failure by whether the job raised)."""
    from src.core.database import async_session_factory
    from src.domain.codegen.provision_gate import PrimitiveGapError
    from src.domain.codegen.repository import CodegenRunRepository
    from src.domain.projects.repository import ProjectRepository
    from src.domain.workspaces.repository import MembershipRepository
    from src.domain.codegen.provision_manifest import validate_manifest

    declaration = validate_manifest(manifest)
    pack_installs = pack_installs or []
    pack_compositions = pack_compositions or []

    db_factory = ctx.get("db_factory", async_session_factory)
    async with db_factory() as db:
        run = await CodegenRunRepository(db).get_by_id(UUID(run_id))
        if not run or run.status in {"failed", "canceled", "completed", "completed_with_limitations", "completed_with_l3", "blocked"}:
            raise ValueError("Run is not active")
        project = await ProjectRepository(db).get_by_id(run.project_id)
        if not project or not run.created_by or not await MembershipRepository(db).get(project.workspace_id, run.created_by):
            raise ValueError("Run creator is no longer a member of the project's workspace")
        project_id = run.project_id
        assert_discovery_installable(getattr(project, "discovery_json", None))

    if declaration["entities"] or declaration["storage_buckets"]:
        await _process_provision_manifest(db_factory=db_factory, project_id=project_id, manifest=manifest)

    pack_install_outcomes = await _process_pack_installs(
        db_factory=db_factory,
        project_id=project_id,
        entries=pack_installs,
        user_prompt="",
        run_id=UUID(run_id),
        sync_requirements=False,
    )
    composition_outcomes = await _process_pack_compositions(
        db_factory=db_factory, project_id=project_id, entries=pack_compositions, user_prompt="",
        run_id=UUID(run_id),
    )
    blocked_packs = [o for o in pack_install_outcomes if not o.get("ok")]
    gaps = [o for o in composition_outcomes if not o.get("ok")]
    if blocked_packs or any(g.get("criticality", "required") == "required" for g in gaps):
        raise PrimitiveGapError(
            _capability_gap_message(blocked_packs, gaps) or "A requested capability could not be applied.",
            correction=_correction_from_outcomes(blocked_packs, gaps),
        )
    # Only optional gaps can reach here (a required one already raised above), but
    # they must still surface in the run's final status/message -- carried back
    # through request_provisioning -> the runner -> resultBody -> report_run_complete
    # -> run_codegen's resume_finalize reconstruction, instead of being silently
    # discarded once the job merely returns without raising.
    return {
        "status": "ready",
        "pack_install_outcomes": pack_install_outcomes,
        "composition_outcomes": composition_outcomes,
    }


async def _attach_pending_user_inputs_if_present(result: Any, repo_path: str) -> Any:
    """If ``p2hs/required-user-inputs.json`` exists in the checkout, copy it onto the result."""
    from dataclasses import replace as dc_replace

    from src.domain.codegen.user_inputs_manifest import load_manifest_from_repo_path

    manifest = await asyncio.to_thread(load_manifest_from_repo_path, repo_path)
    if not manifest:
        return result
    return dc_replace(result, pending_user_inputs=manifest)


async def _attach_l3_candidates_if_present(result: Any, repo_path: str) -> Any:
    """Carry the Core-owned pre-run L3 declaration into finalization."""
    from dataclasses import replace as dc_replace

    path = Path(repo_path) / "p2hs" / "l3-plan.json"
    try:
        payload = json.loads(await asyncio.to_thread(path.read_text, encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError):
        return result
    candidates = payload.get("candidates") if isinstance(payload, dict) else None
    if not isinstance(candidates, list):
        return result
    valid = [item for item in candidates if isinstance(item, dict) and item.get("candidate_id")]
    return dc_replace(result, pending_l3_candidates=valid)


async def _filter_pending_user_inputs_by_stored_builder_values(
    db_factory,
    project_id: UUID,
    result: Any,
) -> Any:
    """Drop manifest items already satisfied in DB (``settings.builder_user_values``).

    The repo file may stay stale if it was gitignored or the agent could not commit a clear;
    submitted public URLs/IDs live in the platform DB and must not keep re-blocking runs.
    """
    from dataclasses import replace as dc_replace

    pu = getattr(result, "pending_user_inputs", None)
    if not pu or not isinstance(pu, dict):
        return result
    items = pu.get("items")
    if not isinstance(items, list) or not items:
        return result
    async with db_factory() as db:
        from src.domain.projects.repository import ProjectEnvVarRepository, ProjectSettingsRepository

        settings = await ProjectSettingsRepository(db).get_or_create(project_id)
        env_var_keys = {
            str(row.key).strip()
            for row in await ProjectEnvVarRepository(db).list_for_project(project_id)
            if getattr(row, "key", None)
        }
    buv = (settings.settings or {}).get("builder_user_values") or {}
    remaining = _remaining_pending_user_input_items(
        pu,
        builder_user_values=buv,
        env_var_keys=env_var_keys,
    )
    if not remaining:
        return dc_replace(result, pending_user_inputs=None)
    return dc_replace(result, pending_user_inputs={**pu, "items": remaining})


async def _finalize_codegen_as_failed(
    *,
    db_factory,
    project_id: UUID,
    run_id: UUID,
    transcript_lines: list[str],
    ws_publish,
    user_message: str,
    log_key: str = "codegen_failed_finalized",
) -> None:
    """Persist ``failed`` + timeline + WS so the run never stays ``running`` (idempotent UX)."""
    from src.domain.builder.repository import TimelineEventRepository
    from src.domain.codegen.repository import CodegenRunRepository

    msg = (user_message or "Code generation failed.").strip()[:8000]
    err_db = msg[:4000] if len(msg) > 4000 else msg
    timeline_detail = err_db[:500] if len(err_db) > 500 else err_db

    async with db_factory() as db:
        await TimelineEventRepository(db).create(
            project_id,
            "run_failed",
            run_id=run_id,
            phase="error",
            title="Build failed",
            detail_text=timeline_detail,
        )
        cg = CodegenRunRepository(db)
        await cg.update_status(
            run_id,
            "failed",
            error_message=err_db,
            agent_transcript=_cap_agent_transcript(transcript_lines),
            result_summary=err_db[:500] if err_db else None,
        )
        cr = await cg.get_by_id(run_id)
        if cr:
            from src.workers.run_chat_persist import persist_assistant_for_run_session

            await persist_assistant_for_run_session(
                db, cr, message_type="system", content=msg
            )
        await db.commit()
    await _settle_codegen_credit_reservation(
        db_factory=db_factory,
        project_id=project_id,
        run_id=run_id,
        charged_units=0,
        description="Released reservation after failed code generation",
    )
    try:
        await ws_publish(
            "codegen.failed",
            {"run_id": str(run_id), "error": timeline_detail},
        )
    except Exception:
        pass
    logger.info(log_key, run_id=str(run_id))


async def _settle_codegen_credit_reservation(
    *,
    db_factory,
    project_id: UUID,
    run_id: UUID,
    charged_units: int,
    description: str,
) -> None:
    """Settle a codegen reservation if this run created one.

    A missing reservation is normal for legacy and quota-exempt runs, so it is a
    no-op instead of an error path.
    """
    from src.domain.admin.repository import SubscriptionPlanRepository
    from src.domain.billing.balance_service import WorkspaceUsageBalanceService
    from src.domain.billing.credit_reservation_service import CreditReservationService
    from src.domain.billing.repository import (
        SubscriptionRepository,
        WorkspaceCreditLedgerRepository,
        WorkspaceUsageAllowanceRepository,
    )
    from src.domain.projects.repository import ProjectRepository

    async with db_factory() as db:
        project = await ProjectRepository(db).get_by_id(project_id)
        if not project:
            return
        ledger = WorkspaceCreditLedgerRepository(db)
        service = CreditReservationService(
            balance_service=WorkspaceUsageBalanceService(
                subscription_repo=SubscriptionRepository(db),
                plan_repo=SubscriptionPlanRepository(db),
                allowance_repo=WorkspaceUsageAllowanceRepository(db),
                ledger_repo=ledger,
            ),
            ledger_repo=ledger,
            allowance_repo=WorkspaceUsageAllowanceRepository(db),
        )
        try:
            await service.settle(
                workspace_id=project.workspace_id,
                reference=f"codegen:{run_id}",
                charged_units=charged_units,
                description=description,
            )
        except ValueError:
            await db.rollback()
            return
        await db.commit()


async def _record_codegen_execution_usage(*, db_factory, run_id: UUID, seconds: int, settings) -> int:
    """Persist provider-neutral runner usage once and return customer credits.

    The usage row is an operational snapshot; the existing credit reservation
    service remains the financial/idempotency boundary.
    """
    from decimal import Decimal
    from src.domain.codegen.execution import CodegenExecutionTarget, CodegenExecutionProvider, resolve_codegen_execution
    from src.domain.codegen.execution_pricing import ExecutionUsage, calculate_execution_price
    from sqlalchemy import select
    from src.domain.codegen.provider_models import CodegenExecutionSnapshot, CodegenUsage

    async with db_factory() as db:
        snapshot = await db.get(CodegenExecutionSnapshot, run_id)
    if snapshot:
        target = CodegenExecutionTarget(
            provider=CodegenExecutionProvider(snapshot.execution_provider),
            target=snapshot.execution_target,
            runner_class=snapshot.runner_class,
            config_version=snapshot.execution_config_version,
        )
        pricing_version = snapshot.pricing_version
    else:
        target = resolve_codegen_execution(settings)
        pricing_version = settings.codegen_execution_pricing_version
    price = calculate_execution_price(
        ExecutionUsage(target.provider.value, target.target, target.runner_class, seconds),
        pricing_version=pricing_version,
        provider_rate_usd_per_minute=Decimal(str(settings.codegen_execution_provider_rate_usd_per_minute)),
        credits_per_minute=settings.codegen_execution_credits_per_minute,
    )
    async with db_factory() as db:
        usage = (await db.execute(select(CodegenUsage).where(CodegenUsage.run_id == run_id))).scalar_one_or_none()
        if usage is None:
            usage = CodegenUsage(run_id=run_id, funding_source="p2hs", usage_source="github_actions")
            db.add(usage)
        usage.execution_seconds = seconds
        usage.execution_provider = target.provider.value
        usage.execution_target = target.target
        usage.runner_class = target.runner_class
        usage.provider_cost_microusd = price.provider_cost_microusd
        usage.execution_credits = price.credit_charge
        usage.execution_pricing_version = price.pricing_version
        usage.metadata_json = {
            **(usage.metadata_json or {}),
            "execution_usage_operation_id": f"codegen:{run_id}:execution",
        }
        await db.commit()
    return price.credit_charge


async def _maybe_dispatch_codegen_local(
    *,
    db_factory,
    snap: "_RunSnapshot",
    transcript_lines: list[str],
    ws_publish,
    ws_line_callback,
    system_prompt: str = "",
    user_prompt: str = "",
    plan_context: Optional[dict] = None,
    provider: str = "",
) -> Optional[dict]:
    """If ``snap.execution_target == 'local'``, hand the run to a paired desktop device instead
    of GitHub Actions / the in-process AI provider, and return the arq task's result dict.
    Returns ``None`` when execution_target is 'cloud' (default) — the caller falls through to
    the existing GitHub Actions / server dispatch unchanged.

    Fail-closed: workspace opt-in, ``codegen.run.desktop``, and a user-owned active device
    that belongs to a workspace member. Device routing prefers ``snap.device_id``, else the
    run creator's most recently active device.
    """
    if snap.execution_target != "local":
        return None

    from src.domain.codegen.actions_pipeline import dispatch_codegen_local_run
    from src.domain.devices.local_execution import LocalExecutionPolicy
    from src.domain.devices.repository import DeviceRepository
    from src.domain.projects.repository import ProjectRepository
    from src.domain.workspaces.repository import MembershipRepository, WorkspaceSettingsRepository

    device_id = snap.device_id
    try:
        async with db_factory() as db:
            project = await ProjectRepository(db).get_by_id(snap.project_id)
            if not project:
                raise RuntimeError("Project not found")
            user_id = snap.created_by
            if not user_id:
                raise RuntimeError("Local run has no creating user")
            policy = LocalExecutionPolicy(
                devices=DeviceRepository(db),
                memberships=MembershipRepository(db),
                settings=WorkspaceSettingsRepository(db),
            )
            device = await policy.resolve_device(
                workspace_id=project.workspace_id,
                user_id=user_id,
                preferred_device_id=device_id,
            )
            device_id = device.id
    except Exception as exc:
        await _finalize_codegen_as_failed(
            db_factory=db_factory,
            project_id=snap.project_id,
            run_id=snap.id,
            transcript_lines=transcript_lines,
            ws_publish=ws_publish,
            user_message=(
                "Local execution was requested but is not allowed: "
                f"{exc}. Enable local execution for this workspace, pair a device, "
                "and ensure you have the codegen.run.desktop permission."
            ),
            log_key="codegen_local_gated",
        )
        return {"status": "failed", "reason": "local_execution_not_allowed"}

    dispatch = await dispatch_codegen_local_run(
        db_factory=db_factory,
        run_id=snap.id,
        project_id=snap.project_id,
        device_id=device_id,
        system_prompt=system_prompt,
        user_prompt=user_prompt,
        plan_context=plan_context,
        provider=provider,
    )
    await ws_line_callback("Codegen has been queued for local execution on your device.", "stdout")
    await ws_publish(
        "codegen.status_changed",
        {
            "run_id": str(snap.id),
            "status": "queued",
            "stage": "local_device",
            "message": "Waiting for your device to pick up the run.",
            **dispatch,
        },
    )
    logger.info("codegen_local_dispatched", run_id=str(snap.id), device_id=str(device_id))
    return {"status": "queued", "executor": "local"}


_USER_RESPONSE_MARKER_RE = re.compile(
    r"P2HS_USER_RESPONSE:\s*(.+?)(?:\n\s*(?:P2HS_[A-Z0-9_]+):|\Z)",
    re.IGNORECASE | re.DOTALL,
)
_APPLIED_LINE_RE = re.compile(
    r"(?m)^\s*\*{0,2}Applied:\*{0,2}\s+(.+)$",
    re.IGNORECASE,
)
_GIT_NOISE_RE = re.compile(
    r"^(?:"
    r"create mode\s+\d+"
    r"|delete mode\s+\d+"
    r"|new file mode\s+\d+"
    r"|deleted file mode\s+\d+"
    r"|old mode\s+\d+"
    r"|new mode\s+\d+"
    r"|rewrite\s+"
    r"|rename (?:from|to)\s+"
    r"|index\s+[0-9a-f]"
    r"|diff --git\s"
    r"|@@\s"
    r"|[+-]{3}\s+[ab]/"
    r"|similarity index\s"
    r"|dissimilarity index\s"
    r"|Binary files\s"
    r"|commit\s+[0-9a-f]{7,}"
    r"|Author:"
    r"|Date:"
    r"|Co-authored-by"
    r"|Made-with:"
    r"|Signed-off-by:"
    r"|Authored by P2HS only"
    r"|\[(?:Shell|Read|Write|Edit|Glob|Grep|Search|Bash|Tool)[^\]]*\]"
    r")",
    re.IGNORECASE,
)
_VAGUE_RESPONSE_RE = re.compile(
    r"^(changes applied|done|ok|okay|success|completed|no changes were made)\.?$",
    re.IGNORECASE,
)
# Provider transcripts occasionally contain a unified diff even when the provider
# did not emit a P2HS_USER_RESPONSE marker.  Filtering only git metadata is not
# sufficient: CSS/JS lines in the hunk are otherwise plausible-looking prose and
# leak internal implementation details into the builder chat.
_INTERNAL_PATCH_MARKER_RE = re.compile(
    r"(?:^|\n)\s*(?:diff --git\s|index\s+[0-9a-f]+|@@\s|[+-]{3}\s+[ab]/)"
    r"|\[(?:internal path|internal url) redacted\]",
    re.IGNORECASE,
)
_CODE_FRAGMENT_RE = re.compile(
    r"^(?:[+-]\s*)?(?:@(?:import|media|supports|keyframes)\b|:root\s*\{|"
    r"(?:const|let|var|function|async function|return)\b|"
    r"(?:if|else|for|while)\s*\(|[}\]);,]$|[{};]\s*$)",
    re.IGNORECASE,
)
_USER_RESPONSE_MAX_CHARS = 800


def _friendly_file_label(path: str) -> str:
    name = (path or "").replace("\\", "/").rstrip("/").split("/")[-1]
    return name or path


def _fallback_from_files(
    files_changed: Optional[list[str]], commit_sha: Optional[str]
) -> str:
    from src.integrations.platform_ports import is_platform_projection_path

    visible = [
        p
        for p in (files_changed or [])
        if p and not str(p).startswith(".") and not is_platform_projection_path(p)
    ]
    if not visible:
        return (
            "I updated your project with the requested changes."
            if commit_sha
            else "No changes were made."
        )
    labels = [_friendly_file_label(p) for p in visible[:3]]
    if len(visible) == 1:
        return f"Updated {labels[0]} to apply your request."
    if len(visible) <= 3:
        return f"Updated {', '.join(labels[:-1])} and {labels[-1]} to apply your request."
    extra = len(visible) - 3
    noun = "file" if extra == 1 else "files"
    return f"Updated {', '.join(labels)} and {extra} other {noun} to apply your request."


def _is_noise_response_line(line: str) -> bool:
    stripped = line.strip().lstrip("#- *`")
    if not stripped:
        return True
    if stripped.startswith("{") or stripped.startswith("["):
        return True
    if _GIT_NOISE_RE.match(stripped):
        return True
    if _VAGUE_RESPONSE_RE.match(stripped):
        return True
    return False


def _looks_like_internal_patch(text: str) -> bool:
    """Reject provider patch/code output as a user-facing summary.

    This is deliberately conservative.  A genuine prose response may mention a
    file or a CSS selector, but it should not contain unified-diff markers,
    redacted internal paths, or a run of source-code fragments.
    """
    if _INTERNAL_PATCH_MARKER_RE.search(text):
        return True
    lines = [line.strip() for line in text.splitlines() if line.strip()]
    if len(lines) < 2:
        return False
    code_lines = sum(bool(_CODE_FRAGMENT_RE.match(line)) for line in lines)
    return code_lines >= 2 and code_lines / len(lines) >= 0.4


def _clean_user_facing_text(raw: str) -> str:
    if _looks_like_internal_patch(raw):
        return ""
    cleaned = re.sub(r"```[\w]*\n.*?```", "", raw, flags=re.DOTALL)
    lines = [ln.strip() for ln in cleaned.splitlines()]
    kept: list[str] = []
    for line in lines:
        if _is_noise_response_line(line):
            continue
        kept.append(line.strip().lstrip("#- *`"))
    text = re.sub(r"\n{3,}", "\n\n", "\n".join(kept)).strip()
    text = re.sub(r"[ \t]{2,}", " ", text)
    if len(text) <= _USER_RESPONSE_MAX_CHARS:
        return text
    clipped = text[:_USER_RESPONSE_MAX_CHARS].rsplit(".", 1)[0].strip()
    if len(clipped) >= 40:
        return clipped + "."
    return text[:_USER_RESPONSE_MAX_CHARS].rstrip() + "…"


def _build_user_response(
    raw_summary: Optional[str],
    commit_sha: Optional[str],
    files_changed: Optional[list[str]] = None,
) -> str:
    """Return a clear, user-facing paragraph for the builder chat."""
    fallback = _fallback_from_files(files_changed, commit_sha)
    if not raw_summary or not raw_summary.strip():
        return fallback

    marker = _USER_RESPONSE_MARKER_RE.search(raw_summary)
    if marker:
        marked = _clean_user_facing_text(marker.group(1))
        if marked and not _VAGUE_RESPONSE_RE.match(marked):
            return marked

    applied = _APPLIED_LINE_RE.search(raw_summary)
    if applied:
        applied_text = _clean_user_facing_text(applied.group(1))
        if applied_text and not _VAGUE_RESPONSE_RE.match(applied_text):
            return applied_text.rstrip(".") + "."

    cleaned = _clean_user_facing_text(raw_summary)
    if cleaned and not _VAGUE_RESPONSE_RE.match(cleaned) and len(cleaned) >= 12:
        return cleaned
    return fallback


@dataclass(frozen=True)
class _RunSnapshot:
    id: UUID
    project_id: UUID
    created_by: Optional[UUID]
    plan: dict[str, Any]
    run_mode: str
    prompt_id: Optional[UUID] = None
    parent_plan_id: Optional[UUID] = None
    selected_elements: Optional[list] = None
    # Desktop execution agent (Phase 0): "cloud" (default) | "local"
    execution_target: str = "cloud"
    device_id: Optional[UUID] = None
    l3_candidate_ids: Optional[list[str]] = None
    execution_seconds: Optional[int] = None


async def _maybe_update_persona(db_factory, project_id: UUID, plan: dict) -> None:
    from src.domain.codegen.prompt_wrapper import extract_persona_from_prompt
    from src.domain.projects.repository import ProjectRepository, ProjectSettingsRepository

    original_prompt = plan.get("original_user_prompt", "")
    if not original_prompt:
        return

    async with db_factory() as db:
        settings_repo = ProjectSettingsRepository(db)
        existing = await settings_repo.get_persona(project_id)
        if existing:
            return
        project = await ProjectRepository(db).get_by_id(project_id)
        project_name = project.name if project else ""
        persona = extract_persona_from_prompt(original_prompt, project_name)
        if persona:
            await settings_repo.set_persona(project_id, persona)
            await db.commit()
            logger.info(
                "project_persona_established",
                project_id=str(project_id),
                industry=persona.get("industry"),
                business_name=persona.get("business_name"),
            )


async def _timeline_commit(db_factory, project_id: UUID, event_type: str, **fields: Any) -> None:
    from src.domain.builder.repository import TimelineEventRepository

    async with db_factory() as db:
        await TimelineEventRepository(db).create(project_id, event_type, **fields)
        await db.commit()


async def _get_parent_plan_document(db_factory, parent_plan_id: UUID) -> Optional[str]:
    """Fetch the plan_document from the parent plan run."""
    async with db_factory() as db:
        from src.domain.codegen.repository import CodegenRunRepository
        plan_run = await CodegenRunRepository(db).get_by_id(parent_plan_id)
        return plan_run.plan_document if plan_run else None


async def _ensure_generation_plan_context(
    db_factory,
    project_id: UUID,
    plan_context: dict[str, Any],
    *,
    user_prompt: str = "",
    run_id: UUID | None = None,
    settings=None,
) -> dict[str, Any]:
    """Ensure every execution path receives Core-owned composition artifacts.

    A GenerationPlan already sitting on the run is the previous composition,
    not this prompt. Reconcile first so the current sentence selects blocks,
    declares form fields, and is what pack install and codegen both read.
    """
    if plan_context.get("composition_reconciled") and isinstance(
        plan_context.get("generation_plan"), dict
    ):
        return plan_context

    from src.config import get_settings
    from src.domain.codegen.composition_manifest import render_core_composition_files
    from src.domain.codegen.composition_reconcile import reconcile_composition

    async with db_factory() as db:
        reconciled = await reconcile_composition(
            db,
            project_id,
            user_prompt,
            settings or get_settings(),
            run_id=run_id,
        )
    if isinstance(reconciled, dict):
        plan_context["generation_plan"] = reconciled
        plan_context["composition_reconciled"] = True
        plan_context["core_composition_files"] = render_core_composition_files(reconciled)
        logger.info(
            "codegen_composition_reconciled",
            project_id=str(project_id),
            app_blocks=(reconciled.get("composition") or {}).get("app_block_ids") or [],
            feature_packs=(reconciled.get("composition") or {}).get("feature_pack_ids") or [],
        )
        return plan_context

    from src.domain.app_blocks.planner import source_prompt_from_discovery
    from src.domain.codegen.generation_plan import compile_generation_plan
    from src.domain.composition_recipes.compile import compile_pinned_install_plan
    from src.domain.codegen.app_block_plan import _latest_generation_plan
    from src.domain.projects.discovery_schemas import DiscoveryResult
    from src.domain.projects.repository import ProjectRepository

    generation_plan: dict[str, Any] | None = None
    async with db_factory() as db:
        project = await ProjectRepository(db).get_by_id(project_id)
        if project:
            discovery_json = project.discovery_json or {}
            raw_result = (discovery_json.get("result") or {})
            if isinstance(raw_result, dict) and raw_result.get("spec"):
                result = DiscoveryResult.model_validate(raw_result)
                source_prompt = source_prompt_from_discovery(discovery_json)
                app_block_plan = await compile_pinned_install_plan(
                    db,
                    discovery_json,
                    result.spec,
                    source_prompt=source_prompt,
                    project_type=result.project_type,
                )
                generation_plan = compile_generation_plan(
                    result,
                    app_block_plan,
                    project_id=project_id,
                    source_discovery_revision=discovery_json.get("version"),
                    catalog_digests={"discovery": "2"},
                ).model_dump(mode="json")
        if generation_plan is None:
            generation_plan = await _latest_generation_plan(db, project_id)

    if not isinstance(generation_plan, dict):
        logger.warning(
            "codegen_composition_context_unavailable",
            project_id=str(project_id),
        )
        return plan_context

    plan_context["generation_plan"] = generation_plan
    plan_context["core_composition_files"] = render_core_composition_files(generation_plan)
    logger.info(
        "codegen_composition_context_ready",
        project_id=str(project_id),
        app_blocks=(generation_plan.get("composition") or {}).get("app_block_ids") or [],
        feature_packs=(generation_plan.get("composition") or {}).get("feature_pack_ids") or [],
    )
    return plan_context


async def run_codegen(ctx: dict, run_id: str) -> dict:
    from src.core.database import async_session_factory
    from src.core.pubsub import get_pubsub_redis, publish_event
    from src.core.run_events import publish_project_event_dual
    from src.domain.builder.repository import ChangedFileRepository, TimelineEventRepository
    from src.domain.codegen.repository import CodegenRunRepository
    from src.domain.instructions.repository import PromptRepository
    from src.domain.previews.repository import PreviewRepository
    from src.domain.projects.repository import ProjectRepository
    from src.domain.revisions.repository import RevisionRepository
    from src.integrations.ai_provider import CodegenResult, get_ai_provider

    db_factory = ctx.get("db_factory", async_session_factory)
    rid = UUID(run_id)
    settings = get_settings()

    logger.info(
        "codegen_task_started",
        run_id=run_id,
        configured_ai_provider=str(settings.ai_provider),
    )

    resume_finalize = False
    result: Optional[CodegenResult] = None
    snap: Optional[_RunSnapshot] = None

    async with db_factory() as db:
        codegen_repo = CodegenRunRepository(db)
        run = await codegen_repo.get_by_id(rid)
        if not run:
            logger.warning("codegen_run_not_found", run_id=run_id)
            return {"status": "skipped", "reason": "not found"}
        logger.info("codegen_run_loaded", run_id=run_id, status=run.status, project_id=str(run.project_id))
        if run.status in (
            "completed",
            "completed_with_limitations",
            "completed_with_l3",
            "blocked",
            "plan_ready",
            "awaiting_user_input",
            "user_input_required",
            "clarification_required",
        ):
            return {"status": "skipped", "reason": f"already {run.status}"}
        if run.status == "failed":
            return {"status": "skipped", "reason": "failed"}
        if run.status == "canceled":
            return {"status": "skipped", "reason": "canceled"}

        if run.run_mode != "plan":
            project = await ProjectRepository(db).get_by_id(run.project_id)
            if project:
                assert_discovery_installable(project.discovery_json)

        if (
            run.status == "running"
            and run.worker_checkpoint == "pushed"
            and run.checkpoint_payload
        ):
            resume_finalize = True
            pl = run.checkpoint_payload
            result = CodegenResult(
                commit_sha=pl.get("commit_sha"),
                summary=pl.get("summary") or "",
                plan=pl.get("plan"),
                files_changed=pl.get("files_changed"),
                files_changed_detail=pl.get("files_changed_detail"),
                tokens_used=int(pl.get("tokens_used") or 0),
                provider=str(pl.get("provider") or ""),
                pending_user_inputs=pl.get("pending_user_inputs"),
                pending_provision=pl.get("pending_provision"),
                pending_email_templates=pl.get("pending_email_templates"),
                # Resolved pre-push by resolve_codegen_capabilities (only ever
                # optional gaps -- a required one means this run never got here at
                # all, see report_run_complete's "blocked" branch). Reconstructed
                # here so the timeline-commit/status logic below reads real
                # outcomes instead of silently defaulting to "nothing to report".
                pack_install_outcomes=pl.get("pending_pack_install_outcomes"),
                composition_outcomes=pl.get("pending_pack_composition_outcomes"),
            )
            snap = _RunSnapshot(
                id=run.id,
                project_id=run.project_id,
                created_by=run.created_by,
                plan=dict(run.plan or {}),
                run_mode=run.run_mode or "agent",
                prompt_id=run.prompt_id,
                parent_plan_id=run.parent_plan_id,
                selected_elements=run.selected_elements,
                execution_target=run.execution_target or "cloud",
                device_id=run.device_id,
                l3_candidate_ids=list(pl.get("l3_candidate_ids") or []),
                execution_seconds=int(pl.get("execution_seconds") or 0) or None,
            )
            await db.commit()
        elif run.status == "running" and not (
            run.worker_checkpoint == "pushed" and run.checkpoint_payload
        ):
            if run.worker_checkpoint in ("github_actions_dispatched", "github_actions_continuing"):
                # GitHub Actions owns execution. Its completion webhook will checkpoint the
                # generated commit; redispatching here could apply the same prompt twice.
                logger.warning(
                    "codegen_waiting_for_github_actions_completion",
                    run_id=run_id,
                )
                return {"status": "skipped", "reason": "github actions completion pending"}
            # Orphaned ``running`` (worker crash / ARQ timeout): was previously skipped forever.
            started = run.started_at
            age = (
                (datetime.now(timezone.utc) - started)
                if started
                else timedelta(days=999)
            )
            reclaim_after = timedelta(seconds=settings.codegen_reclaim_running_seconds)
            if age < reclaim_after:
                logger.info(
                    "codegen_skip_already_running",
                    run_id=run_id,
                    age_seconds=int(age.total_seconds()),
                )
                return {"status": "skipped", "reason": "already running"}
            logger.warning(
                "codegen_reclaim_stale_running",
                run_id=run_id,
                age_seconds=int(age.total_seconds()),
            )
            await TimelineEventRepository(db).create(
                run.project_id,
                "run_reclaimed",
                run_id=rid,
                phase="init",
                title="Resuming interrupted build",
            )
            await codegen_repo.update_checkpoint(rid, None, None)
            await codegen_repo.refresh_started_at(rid)
            await db.commit()
            snap = _RunSnapshot(
                id=run.id,
                project_id=run.project_id,
                created_by=run.created_by,
                plan=dict(run.plan or {}),
                run_mode=run.run_mode or "agent",
                prompt_id=run.prompt_id,
                parent_plan_id=run.parent_plan_id,
                selected_elements=run.selected_elements,
                execution_target=run.execution_target or "cloud",
                device_id=run.device_id,
            )
        elif run.status in ("queued", "pending"):
            await TimelineEventRepository(db).create(
                run.project_id,
                "run_created",
                run_id=rid,
                phase="init",
                title="Starting",
            )
            await codegen_repo.update_checkpoint(rid, None, None)
            await codegen_repo.update_status(rid, "running")
            await db.commit()
            snap = _RunSnapshot(
                id=run.id,
                project_id=run.project_id,
                created_by=run.created_by,
                plan=dict(run.plan or {}),
                run_mode=run.run_mode or "agent",
                prompt_id=run.prompt_id,
                parent_plan_id=run.parent_plan_id,
                selected_elements=run.selected_elements,
                execution_target=run.execution_target or "cloud",
                device_id=run.device_id,
            )
        else:
            return {"status": "skipped", "reason": f"unexpected status {run.status}"}

    assert snap is not None
    system_prompt = snap.plan.get("system_prompt", "")
    from src.domain.codegen.project_context import build_project_context, context_prompt, write_project_context
    async with db_factory() as context_db:
        project_context = await build_project_context(context_db, snap.project_id, snap.id)
        context_project = await ProjectRepository(context_db).get_by_id(snap.project_id)
        if context_project:
            write_project_context(get_project_meta_path(settings.repo_workspace, context_project.workspace_id, snap.project_id), project_context)
    system_prompt += context_prompt(project_context)
    user_prompt = snap.plan.get("user_prompt", "")
    original_user_prompt = snap.plan.get("original_user_prompt") or user_prompt
    plan_ctx: dict[str, Any] = dict(snap.plan.get("context") or {})
    provider_name = plan_ctx.get("codegen_provider_override") or settings.ai_provider.value
    api_key_override = None
    model_override = plan_ctx.get("codegen_cli_model_value")
    if plan_ctx.get("codegen_configuration_source") == "database":
        from src.domain.codegen.provider_registry import CodegenConfigResolver
        async with db_factory() as config_db:
            resolved_codegen = await CodegenConfigResolver(config_db).resolve_for_project(snap.project_id)
            provider_name = resolved_codegen.provider.slug
            model_override = resolved_codegen.model.cli_model_value
            api_key_override = resolved_codegen.secret()
    provider = AIProvider(provider_name)
    ai = get_ai_provider(provider, api_key_override=api_key_override, model_override=model_override)
    logger.info(
        "codegen_effective_provider",
        run_id=run_id,
        provider=provider.value,
        overridden=bool(plan_ctx.get("codegen_provider_override")),
    )

    # For agent runs executed from an approved plan, inject the plan document into context
    if snap.parent_plan_id and not resume_finalize:
        parent_plan_doc = await _get_parent_plan_document(db_factory, snap.parent_plan_id)
        if parent_plan_doc:
            system_prompt = (
                system_prompt
                + "\n\n## Approved Implementation Plan\n"
                + parent_plan_doc
                + "\n\nImplement exactly this plan. Follow each step in order."
            )
            logger.info(
                "codegen_plan_injected",
                run_id=run_id,
                parent_plan_id=str(snap.parent_plan_id),
                plan_length=len(parent_plan_doc),
            )

    logger.info(
        "codegen_prompts_loaded",
        run_id=run_id,
        run_mode=snap.run_mode,
        system_prompt_len=len(system_prompt),
        user_prompt_preview=user_prompt[:200] if user_prompt else "",
        resume_finalize=resume_finalize,
    )

    transcript_lines: list[str] = []
    reserved_credits = 0
    credit_started_at: Optional[float] = None

    # Create a dedicated Redis connection for real-time WS event publishing
    pubsub_redis = await get_pubsub_redis()

    async def _ws_publish(event: str, data: dict) -> None:
        await publish_project_event_dual(pubsub_redis, str(snap.project_id), event, data)

    # Plan-first is an explicit cutover mode. Compatibility prompts remain the
    # input for legacy and shadow runs; they are never merged into plan-first
    # architecture context.
    codegen_request = None
    raw_codegen_request = snap.plan.get("codegen_request")
    if raw_codegen_request:
        requested_execution_mode = (
            raw_codegen_request.get("execution_mode")
            if isinstance(raw_codegen_request, dict)
            else None
        )
        from src.domain.codegen.generation_context import (
            CodegenRequestV1,
            build_codegen_input,
        )

        try:
            codegen_request = CodegenRequestV1.model_validate(raw_codegen_request)
            system_prompt, user_prompt, input_telemetry = build_codegen_input(
                codegen_request,
                legacy_system_prompt=system_prompt,
                legacy_user_prompt=original_user_prompt,
            )
            plan_ctx.update(input_telemetry)
            # Keep the typed Core plan available to every execution backend.
            # The provider receives it as context, while Core materializes the
            # protected composition files before provider execution.
            plan_ctx["generation_plan"] = codegen_request.generation_plan
            from src.domain.codegen.composition_manifest import render_core_composition_files

            plan_ctx["core_composition_files"] = render_core_composition_files(
                codegen_request.generation_plan
            )
            logger.info(
                "codegen_input_resolved",
                run_id=run_id,
                execution_mode=codegen_request.execution_mode,
                generation_plan_hash=input_telemetry.get("generation_plan_hash"),
                resolved_context_hash=input_telemetry.get("resolved_context_hash"),
            )
        except Exception as exc:
            if requested_execution_mode == "plan_first":
                await _finalize_codegen_as_failed(
                    db_factory=db_factory,
                    project_id=snap.project_id,
                    run_id=snap.id,
                    transcript_lines=[],
                    ws_publish=_ws_publish,
                    user_message=f"PLAN_FIRST_INPUT_FAILED: {exc}",
                    log_key="codegen_plan_first_input_failed",
                )
                return {"status": "failed", "reason": "plan_first_input_failed"}
            logger.warning(
                "codegen_shadow_input_invalid",
                run_id=run_id,
                error=str(exc),
            )

    # Plan-first runs must prove that the resolved architecture is executable
    # before a provider is allowed to edit the repository. This is deliberately
    # after request/hash validation and before credits/provider dispatch.
    if codegen_request and codegen_request.execution_mode == "plan_first":
        try:
            from src.domain.cloud.facades import project_port_context
            from src.domain.codegen.preflight import validate_codegen_preflight
            from src.domain.projects.repository import ProjectRepository

            async with db_factory() as preflight_db:
                preflight_project = await ProjectRepository(preflight_db).get_by_id(
                    snap.project_id
                )
                if not preflight_project:
                    raise ValueError("project_not_found")
                port_context = await project_port_context(preflight_db, preflight_project)
                preflight = validate_codegen_preflight(
                    generation_plan=codegen_request.generation_plan,
                    resolved_context=codegen_request.resolved_context.model_dump(mode="json"),
                    port_context=port_context,
                    runtime_invariants=preflight_project.runtime_invariants or {},
                )
            plan_ctx["preflight"] = preflight
            logger.info(
                "codegen_preflight_passed",
                run_id=run_id,
                required_client_operations=preflight["required_client_operations"],
            )
        except Exception as exc:
            await _finalize_codegen_as_failed(
                db_factory=db_factory,
                project_id=snap.project_id,
                run_id=snap.id,
                transcript_lines=transcript_lines,
                ws_publish=_ws_publish,
                user_message=str(exc),
                log_key="codegen_preflight_failed",
            )
            return {"status": "failed", "reason": "codegen_preflight_failed"}

    # ── CREDIT GATE ───────────────────────────────────────────────────────────
    # Skip when resuming a finalize-only checkpoint (AI already ran; no new spend).
    if not resume_finalize and snap.run_mode != "plan":
        from src.domain.admin.repository import SubscriptionPlanRepository as _PlanRepo
        from src.domain.billing.balance_service import WorkspaceUsageBalanceService
        from src.domain.billing.credit_reservation_service import CreditReservationService
        from src.domain.billing.repository import (
            SubscriptionRepository as _SubRepo,
            UsageCreditPackageRepository as _PkgRepo,
            WorkspaceCreditLedgerRepository as _LedgerRepo,
            WorkspaceUsageAllowanceRepository as _AllowanceRepo,
        )
        from src.domain.projects.repository import ProjectRepository as _ProjRepo

        _credits_exhausted = False
        _workspace_id_for_gate: Optional[UUID] = None
        _pkg_list: list[dict] = []

        async with db_factory() as _db:
            _proj = await _ProjRepo(_db).get_by_id(snap.project_id)
            if _proj:
                _workspace_id_for_gate = _proj.workspace_id

            if _workspace_id_for_gate and plan_ctx.get("codegen_funding_source") != "byok":
                from src.domain.billing.credit_policy import workspace_skips_ai_credits

                _skip_quota = await workspace_skips_ai_credits(_db, _workspace_id_for_gate)
                if not _skip_quota:
                    _balance_svc = WorkspaceUsageBalanceService(
                        subscription_repo=_SubRepo(_db),
                        plan_repo=_PlanRepo(_db),
                        allowance_repo=_AllowanceRepo(_db),
                        ledger_repo=_LedgerRepo(_db),
                    )
                    _ledger = _LedgerRepo(_db)
                    _reservation = await CreditReservationService(
                        balance_service=_balance_svc,
                        ledger_repo=_ledger,
                        allowance_repo=_AllowanceRepo(_db),
                    ).reserve(
                        workspace_id=_workspace_id_for_gate,
                        units=int(
                            (plan_ctx.get("credit_estimate") or {}).get(
                                "maximum_credits", 5
                            )
                        ),
                        reference=f"codegen:{snap.id}",
                        project_id=snap.project_id,
                        codegen_job_id=snap.id,
                        description="Reserved credits for code generation",
                    )
                    if _reservation is None:
                        _credits_exhausted = True
                    else:
                        reserved_credits = _reservation.reserved_units
                    _pkgs = await _PkgRepo(_db).list_all(active_only=True)
                    _pkg_list = [
                        {"slug": p.slug, "name": p.name, "units": p.units, "price_minor": p.price_minor, "currency": p.currency}
                        for p in _pkgs
                    ]
                    await _db.commit()

        if _credits_exhausted:
            await _finalize_codegen_as_failed(
                db_factory=db_factory,
                project_id=snap.project_id,
                run_id=snap.id,
                transcript_lines=transcript_lines,
                ws_publish=_ws_publish,
                user_message="You've used all your available AI credits. Purchase a top-up to continue.",
                log_key="codegen_credit_exhausted",
            )
            try:
                await _ws_publish(
                    "billing.credit_exhausted",
                    {
                        "run_id": str(snap.id),
                        "workspace_id": str(_workspace_id_for_gate),
                        "packages": _pkg_list,
                    },
                )
            except Exception:
                pass
            logger.warning(
                "codegen_blocked_no_credits",
                run_id=run_id,
                workspace_id=str(_workspace_id_for_gate),
            )
            return {"status": "failed", "reason": "credit_exhausted"}

        if reserved_credits:
            credit_started_at = monotonic()

    async def _ws_line_callback(
        text: str,
        stream: str = "stdout",
        *,
        detail_format: Optional[str] = None,
    ) -> None:
        """Stream AI agent lines to WebSocket clients and buffer for persisted replay."""
        transcript_lines.append(text if stream == "stdout" else f"[stderr] {text}")
        payload: dict = {
            "run_id": str(snap.id),
            "text": text,
            "type": "output",
            "stream": stream,
        }
        if detail_format:
            payload["detail_format"] = detail_format
        await _ws_publish("codegen.ai_output", payload)

    try:
        if resume_finalize:
            await _timeline_commit(
                db_factory,
                snap.project_id,
                "codegen_db_resume",
                run_id=snap.id,
                phase="finalize",
                title="Resuming",
            )
        else:
            await _ws_publish("codegen.started", {"run_id": str(snap.id), "run_mode": snap.run_mode})

            # ── PLAN MODE ─────────────────────────────────────────────────────────
            if snap.run_mode == "plan":
                await _timeline_commit(
                    db_factory,
                    snap.project_id,
                    "plan_started",
                    run_id=snap.id,
                    phase="planning",
                    title="Analyzing your request",
                )
                await _ws_publish("codegen.status_changed", {"run_id": str(snap.id), "status": "running"})

                plan_result = await _run_plan_mode(
                    db_factory=db_factory,
                    run_id=snap.id,
                    project_id=snap.project_id,
                    system_prompt=system_prompt,
                    user_prompt=original_user_prompt,
                    ai=ai,
                    settings=settings,
                    plan_context=plan_ctx,
                    line_callback=_ws_line_callback,
                )

                plan_document = (
                    (plan_result.plan or {}).get("document", "")
                    or plan_result.summary
                    or ""
                )

                if getattr(plan_result, "provider_invocation_failed", False):
                    await _finalize_codegen_as_failed(
                        db_factory=db_factory,
                        project_id=snap.project_id,
                        run_id=snap.id,
                        transcript_lines=transcript_lines,
                        ws_publish=_ws_publish,
                        user_message=plan_document.strip()
                        or plan_result.summary
                        or "Plan generation failed.",
                        log_key="codegen_plan_provider_failed",
                    )
                    return {"status": "failed"}

                from src.domain.codegen.clarification_parse import extract_clarification_from_transcript
                from src.domain.codegen.interruption_schemas import clarification_to_interruption

                clar_scan = f"{plan_document}\n" + "\n".join(transcript_lines)
                clar_req = extract_clarification_from_transcript(clar_scan)
                if clar_req:
                    env_wrap = clarification_to_interruption(clar_req)
                    canonical = clar_req.model_dump(mode="json")
                    async with db_factory() as db:
                        cr_repo = CodegenRunRepository(db)
                        await cr_repo.update_status(
                            snap.id,
                            "clarification_required",
                            plan_document=plan_document or None,
                            result_summary=(clar_req.thinking_summary or "")[:500]
                            if clar_req.thinking_summary
                            else None,
                            tokens_used=plan_result.tokens_used or None,
                            agent_transcript=_cap_agent_transcript(transcript_lines),
                            interruption_payload=env_wrap.to_storage_dict(),
                        )
                        pr_run = await cr_repo.get_by_id(snap.id)
                        if pr_run and snap.prompt_id:
                            from src.domain.instructions.repository import PromptRepository
                            from src.workers.run_chat_persist import persist_assistant_for_run_session

                            await PromptRepository(db).update_response(
                                snap.prompt_id,
                                clar_req.thinking_summary
                                or "A few details are needed before we can continue.",
                            )
                            await persist_assistant_for_run_session(
                                db,
                                pr_run,
                                message_type="clarification_required",
                                content=clar_req.thinking_summary
                                or "A few details are needed before we can continue.",
                                payload=canonical,
                            )
                        await db.commit()
                    await _timeline_commit(
                        db_factory,
                        snap.project_id,
                        "plan_needs_clarification",
                        run_id=snap.id,
                        phase="planning",
                        title="Clarification needed",
                    )
                    await _ws_publish(
                        "codegen.clarification_required",
                        {"run_id": str(snap.id), **canonical},
                    )
                    logger.info("codegen_plan_clarification_required", run_id=run_id)
                    return {"status": "clarification_required"}

                async with db_factory() as db:
                    await CodegenRunRepository(db).update_status(
                        snap.id,
                        "plan_ready",
                        plan_document=plan_document,
                        result_summary=plan_document[:500] if plan_document else "",
                        tokens_used=plan_result.tokens_used or None,
                        agent_transcript=_cap_agent_transcript(transcript_lines),
                    )
                    if snap.prompt_id:
                        from src.domain.instructions.repository import PromptRepository
                        from src.workers.run_chat_persist import persist_assistant_for_run_session

                        await PromptRepository(db).update_response(
                            snap.prompt_id, "Plan ready. Review and approve to implement."
                        )
                        pr_run = await CodegenRunRepository(db).get_by_id(snap.id)
                        if pr_run:
                            await persist_assistant_for_run_session(
                                db,
                                pr_run,
                                message_type="plan",
                                content=(plan_document or "")[:80000],
                                payload={"chars": len(plan_document or "")},
                            )
                    await db.commit()

                await _timeline_commit(
                    db_factory,
                    snap.project_id,
                    "plan_generated",
                    run_id=snap.id,
                    phase="planning",
                    title="Plan ready",
                )
                await _ws_publish("codegen.plan_ready", {
                    "run_id": str(snap.id),
                    "plan_document": plan_document,
                })

                logger.info("codegen_plan_mode_completed", run_id=run_id)
                return {"status": "plan_ready"}

            # ── AGENT / VISUAL_EDIT MODE ──────────────────────────────────────────
            await _timeline_commit(
                db_factory,
                snap.project_id,
                "plan_started",
                run_id=snap.id,
                phase="planning",
                title="Analyzing your request",
            )

            if plan_ctx.get("mode") == "visual_edit" and plan_ctx.get("visual_edit_elements"):
                from src.integrations.ai_provider import CodegenResult

                if provider not in REPO_BASED_PROVIDERS:
                    result = CodegenResult(
                        summary="Visual edit requires a repo-based AI provider (Cursor or Claude).",
                        provider_invocation_failed=True,
                        provider="deterministic_visual",
                    )
                else:
                    await _timeline_commit(
                        db_factory,
                        snap.project_id,
                        "apply_started",
                        run_id=snap.id,
                        phase="apply",
                        title="Applying visual edits",
                    )
                    result = await _run_visual_deterministic_edit(
                        db_factory=db_factory,
                        run_id=snap.id,
                        project_id=snap.project_id,
                        settings=settings,
                        plan_context=plan_ctx,
                        ws_publish=_ws_publish,
                        transcript_lines=transcript_lines,
                    )
            elif provider in REPO_BASED_PROVIDERS:
                logger.info("codegen_using_repo_based_provider", run_id=run_id, provider=provider.value)
                await _timeline_commit(
                    db_factory,
                    snap.project_id,
                    "apply_started",
                    run_id=snap.id,
                    phase="apply",
                    title="Writing code",
                )
                _local_result = await _maybe_dispatch_codegen_local(
                    db_factory=db_factory,
                    snap=snap,
                    transcript_lines=transcript_lines,
                    ws_publish=_ws_publish,
                    ws_line_callback=_ws_line_callback,
                    system_prompt=system_prompt,
                    user_prompt=original_user_prompt,
                    plan_context=plan_ctx,
                    provider=provider.value,
                )
                if _local_result is not None:
                    return _local_result
                from src.domain.codegen.execution import CodegenExecutionProvider, resolve_codegen_execution
                execution = resolve_codegen_execution(settings)
                if execution.provider is CodegenExecutionProvider.GITHUB:
                    from src.domain.codegen.actions_pipeline import dispatch_codegen_actions_run

                    await _ensure_generation_plan_context(
                        db_factory,
                        snap.project_id,
                        plan_ctx,
                        user_prompt=original_user_prompt,
                        run_id=snap.id,
                        settings=settings,
                    )
                    dispatch = await dispatch_codegen_actions_run(
                        db_factory=db_factory,
                        run_id=snap.id,
                        project_id=snap.project_id,
                        system_prompt=system_prompt,
                        user_prompt=original_user_prompt,
                        plan_context=plan_ctx,
                    )
                    async with db_factory() as db:
                        await CodegenRunRepository(db).update_checkpoint(
                            snap.id,
                            "github_actions_dispatched",
                            dispatch,
                        )
                        await db.commit()
                    await _ws_line_callback(
                        "Codegen has been dispatched to GitHub Actions.",
                        "stdout",
                    )
                    await _ws_publish(
                        "codegen.status_changed",
                        {
                            "run_id": str(snap.id),
                            "status": "running",
                            "stage": "p2hs_ai",
                            "message": "Deploy Agent is running.",
                            **dispatch,
                        },
                    )
                    logger.info(
                        "codegen_github_actions_dispatched",
                        run_id=run_id,
                        repo=dispatch.get("repo_full_name"),
                        branch=dispatch.get("branch"),
                    )
                    return {"status": "running", "executor": "github_actions"}
                else:
                    if execution.provider is not CodegenExecutionProvider.LOCAL_DEV:
                        raise RuntimeError(
                            f"Codegen execution provider {execution.provider.value!r} is not active yet; "
                            "the run was not executed on the Core host"
                        )
                    result = await _run_repo_based_codegen(
                        db_factory=db_factory,
                        run_id=snap.id,
                        project_id=snap.project_id,
                        system_prompt=system_prompt,
                        user_prompt=original_user_prompt,
                        ai=ai,
                        settings=settings,
                        plan_context=plan_ctx,
                        line_callback=_ws_line_callback,
                        ws_publish=_ws_publish,
                    )
                logger.info(
                    "codegen_repo_based_result",
                    run_id=run_id,
                    commit_sha=result.commit_sha,
                    files_changed=len(result.files_changed or []),
                    summary_preview=(result.summary or "")[:200],
                    provider=result.provider,
                    provider_invocation_failed=getattr(
                        result, "provider_invocation_failed", False
                    ),
                )
            else:
                _local_result = await _maybe_dispatch_codegen_local(
                    db_factory=db_factory,
                    snap=snap,
                    transcript_lines=transcript_lines,
                    ws_publish=_ws_publish,
                    ws_line_callback=_ws_line_callback,
                    system_prompt=system_prompt,
                    user_prompt=original_user_prompt,
                    plan_context=plan_ctx,
                    provider=provider.value,
                )
                if _local_result is not None:
                    return _local_result
                from src.domain.codegen.execution import CodegenExecutionProvider, resolve_codegen_execution
                execution = resolve_codegen_execution(settings)
                if execution.provider is CodegenExecutionProvider.GITHUB:
                    from src.domain.codegen.actions_pipeline import dispatch_codegen_actions_run

                    await _ensure_generation_plan_context(
                        db_factory,
                        snap.project_id,
                        plan_ctx,
                        user_prompt=original_user_prompt,
                        run_id=snap.id,
                        settings=settings,
                    )
                    dispatch = await dispatch_codegen_actions_run(
                        db_factory=db_factory,
                        run_id=snap.id,
                        project_id=snap.project_id,
                        system_prompt=system_prompt,
                        user_prompt=original_user_prompt,
                        plan_context=plan_ctx,
                    )
                    async with db_factory() as db:
                        await CodegenRunRepository(db).update_checkpoint(
                            snap.id,
                            "github_actions_dispatched",
                            dispatch,
                        )
                        await db.commit()
                    await _ws_line_callback(
                        "Codegen has been dispatched to GitHub Actions.",
                        "stdout",
                    )
                    await _ws_publish(
                        "codegen.status_changed",
                        {
                            "run_id": str(snap.id),
                            "status": "running",
                            "stage": "p2hs_ai",
                            "message": "Deploy Agent is running.",
                            **dispatch,
                        },
                    )
                    logger.info(
                        "codegen_github_actions_dispatched",
                        run_id=run_id,
                        repo=dispatch.get("repo_full_name"),
                        branch=dispatch.get("branch"),
                    )
                    return {"status": "running", "executor": "github_actions"}
                else:
                    if execution.provider is not CodegenExecutionProvider.LOCAL_DEV:
                        raise RuntimeError(
                            f"Codegen execution provider {execution.provider.value!r} is not active yet; "
                            "the run was not executed on the Core host"
                        )
                    await _timeline_commit(
                        db_factory,
                        snap.project_id,
                        "plan_generated",
                        run_id=snap.id,
                        phase="planning",
                        title="Plan ready",
                    )
                    from src.workers.prompt_media_sync import augment_prompt_with_media

                    augmented_user = await augment_prompt_with_media(
                        db_factory=db_factory,
                        project_id=snap.project_id,
                        run_id=snap.id,
                        repo_path=None,
                        base_user_prompt=original_user_prompt,
                        plan_context=plan_ctx,
                    )
                    result = await ai.generate_plan(
                        system_prompt,
                        augmented_user,
                        line_callback=_ws_line_callback,
                    )

        assert result is not None

        if not resume_finalize and getattr(result, "provider_invocation_failed", False):
            await _finalize_codegen_as_failed(
                db_factory=db_factory,
                project_id=snap.project_id,
                run_id=snap.id,
                transcript_lines=transcript_lines,
                ws_publish=_ws_publish,
                user_message=result.summary
                or "The coding agent failed. Use Retry or send a new prompt.",
                log_key="codegen_agent_invocation_failed",
            )
            return {"status": "failed"}

        # Check if run was canceled while the worker was executing
        async with db_factory() as db:
            current_run = await CodegenRunRepository(db).get_by_id(snap.id)
            if current_run and current_run.status == "canceled":
                logger.info("codegen_run_was_canceled", run_id=run_id)
                return {"status": "canceled"}

        if not resume_finalize:
            from src.domain.codegen.clarification_parse import extract_clarification_from_transcript
            from src.domain.codegen.interruption_schemas import clarification_to_interruption

            cl_text = "\n".join(transcript_lines) + "\n" + (result.summary or "")
            clar_req = extract_clarification_from_transcript(cl_text)
            if clar_req:
                env_wrap = clarification_to_interruption(clar_req)
                canonical = clar_req.model_dump(mode="json")
                async with db_factory() as db:
                    from src.domain.instructions.repository import PromptRepository
                    from src.workers.run_chat_persist import persist_assistant_for_run_session

                    cg = CodegenRunRepository(db)
                    await cg.update_status(
                        snap.id,
                        "clarification_required",
                        plan=result.plan,
                        result_summary=(clar_req.thinking_summary or "")[:500]
                        if clar_req.thinking_summary
                        else None,
                        tokens_used=result.tokens_used,
                        agent_transcript=_cap_agent_transcript(transcript_lines),
                        interruption_payload=env_wrap.to_storage_dict(),
                    )
                    cr = await cg.get_by_id(snap.id)
                    if cr and cr.prompt_id:
                        await PromptRepository(db).update_response(
                            cr.prompt_id,
                            clar_req.thinking_summary
                            or "A few details are needed before we can continue.",
                        )
                        await persist_assistant_for_run_session(
                            db,
                            cr,
                            message_type="clarification_required",
                            content=clar_req.thinking_summary
                            or "A few details are needed before we can continue.",
                            payload=canonical,
                        )
                    await db.commit()
                await _ws_publish(
                    "codegen.clarification_required",
                    {"run_id": str(snap.id), **canonical},
                )
                logger.info("codegen_agent_clarification_required", run_id=run_id)
                return {"status": "clarification_required"}

        fn_deploy_outcomes = await _process_fn_deploys(
            db_factory=db_factory,
            project_id=snap.project_id,
            entries=getattr(result, "pending_fn_deploys", None) or [],
        )
        if fn_deploy_outcomes:
            await _timeline_commit(
                db_factory,
                snap.project_id,
                "functions_deployed",
                run_id=snap.id,
                phase="apply",
                title=f"Deployed {len([o for o in fn_deploy_outcomes if o['ok']])} function(s)",
                detail_text="; ".join(
                    f"{o['slug']}: {'ok' if o['ok'] else o.get('error', 'error')}"
                    for o in fn_deploy_outcomes
                )[:500],
            )

        blocked_packs: list[dict] = []
        pending_configs: list[dict] = []
        gaps: list[dict] = []

        # Already resolved inside _run_repo_based_codegen, BEFORE the push (a required
        # primitive_gap vetoes the push there -- see that function's pre-push gate).
        # Read the outcomes instead of re-invoking these -- they're not re-run here.
        pack_install_outcomes = getattr(result, "pack_install_outcomes", None) or []
        if pack_install_outcomes:
            ok_packs = [o for o in pack_install_outcomes if o.get("ok")]
            blocked_packs = [o for o in pack_install_outcomes if not o.get("ok")]
            pending_configs = [
                o for o in pack_install_outcomes if o.get("status") == "configuration_required"
            ]
            await _timeline_commit(
                db_factory,
                snap.project_id,
                "feature_packs_installed",
                run_id=snap.id,
                phase="apply",
                title=f"Installed {len(ok_packs)} feature pack(s)",
                detail_text="; ".join(
                    f"{o['pack_slug']}: {'ok' if o['ok'] else '; '.join(o.get('errors') or ['error'])}"
                    for o in pack_install_outcomes
                )[:500],
            )
            if blocked_packs:
                await _timeline_commit(
                    db_factory,
                    snap.project_id,
                    "feature_packs_blocked",
                    run_id=snap.id,
                    phase="apply",
                    title=f"{len(blocked_packs)} feature pack(s) blocked",
                    detail_text="; ".join(
                        f"{o['pack_slug']}: {'; '.join(o.get('errors') or ['blocked'])}"
                        for o in blocked_packs
                    )[:500],
                )
            if pending_configs:
                # Distinct from "blocked" on purpose -- these packs were built, they just
                # need a provider connected before they're live. See _capability_gap_message.
                await _timeline_commit(
                    db_factory,
                    snap.project_id,
                    "feature_packs_pending_configuration",
                    run_id=snap.id,
                    phase="apply",
                    title=f"{len(pending_configs)} feature pack(s) built, awaiting configuration",
                    detail_text="; ".join(
                        f"{o['pack_slug']}: {', '.join((o.get('blocker') or {}).get('pending_connections') or [])}"
                        for o in pending_configs
                    )[:500],
                )

        composition_outcomes = getattr(result, "composition_outcomes", None) or []
        if composition_outcomes:
            composed_ok = [o for o in composition_outcomes if o.get("ok")]
            gaps = [o for o in composition_outcomes if not o.get("ok")]
            await _timeline_commit(
                db_factory,
                snap.project_id,
                "capability_composed",
                run_id=snap.id,
                phase="apply",
                title=f"Composed {len(composed_ok)} capability(ies)",
                detail_text="; ".join(
                    f"{o.get('name')}: {'ok' if o.get('ok') else o.get('status', 'blocked')}"
                    for o in composition_outcomes
                )[:500],
            )
            if gaps:
                await _timeline_commit(
                    db_factory,
                    snap.project_id,
                    "capability_primitive_gap",
                    run_id=snap.id,
                    phase="apply",
                    title=f"{len(gaps)} requirement(s) need a platform primitive that doesn't exist yet",
                    detail_text="; ".join(
                        f"{o.get('name')}: missing {', '.join(o.get('missing') or [])}"
                        for o in gaps
                    )[:500],
                )

        provision_outcomes = await _process_provision_manifest(
            db_factory=db_factory,
            project_id=snap.project_id,
            manifest=getattr(result, "pending_provision", None) or {},
        )
        if provision_outcomes:
            created = [o for o in provision_outcomes if o.get("ok") and o.get("action") == "created"]
            failed = [o for o in provision_outcomes if not o.get("ok")]
            if created or failed:
                detail_parts = (
                    [f"{o.get('bucket') or o.get('entity')} created" for o in created]
                    + [f"{o.get('bucket') or o.get('entity')} failed ({o.get('error', '')})" for o in failed]
                )
                await _timeline_commit(
                    db_factory,
                    snap.project_id,
                    "resources_provisioned",
                    run_id=snap.id,
                    phase="apply",
                    title=f"Provisioned {len(created)} resource(s)"
                    + (f" ({len(failed)} failed)" if failed else ""),
                    detail_text="; ".join(detail_parts)[:500],
                )

        email_template_outcomes = await _process_email_templates_manifest(
            db_factory=db_factory,
            project_id=snap.project_id,
            declaration=getattr(result, "pending_email_templates", None) or {},
        )
        if email_template_outcomes:
            applied = [o for o in email_template_outcomes if o.get("ok")]
            failed_templates = [o for o in email_template_outcomes if not o.get("ok")]
            await _timeline_commit(
                db_factory,
                snap.project_id,
                "email_templates_updated",
                run_id=snap.id,
                phase="apply",
                title=f"Updated {len(applied)} email template(s)"
                + (f" ({len(failed_templates)} failed)" if failed_templates else ""),
                detail_text="; ".join(
                    o.get("template_key") or o.get("error", "error") for o in email_template_outcomes
                )[:500],
            )

        filtered_files_changed, filtered_files_changed_detail, had_reported_changes = (
            _filter_user_visible_changes(result.files_changed, result.files_changed_detail)
        )
        result.files_changed = filtered_files_changed
        result.files_changed_detail = filtered_files_changed_detail
        if result.commit_sha and had_reported_changes and not (
            result.files_changed or result.files_changed_detail
        ):
            await _finalize_codegen_as_failed(
                db_factory=db_factory,
                project_id=snap.project_id,
                run_id=snap.id,
                transcript_lines=transcript_lines,
                ws_publish=_ws_publish,
                user_message=(
                    "Code generation finished without any customer-visible file changes. "
                    "The executor only reported internal bookkeeping updates, so the run was not surfaced as applied."
                ),
                log_key="codegen_internal_only_changes",
            )
            return {"status": "failed", "reason": "internal_only_changes"}

        user_facing = _build_user_response(
            result.summary,
            result.commit_sha,
            result.files_changed,
        )
        update_kwargs: dict[str, Any] = {
            "plan": result.plan,
            "result_summary": user_facing,
            "tokens_used": result.tokens_used,
            "agent_transcript": _cap_agent_transcript(transcript_lines),
            "pending_user_inputs": getattr(result, "pending_user_inputs", None),
        }
        placeholder_user_values = await _persist_placeholder_user_inputs(
            db_factory=db_factory,
            project_id=snap.project_id,
            manifest=update_kwargs.get("pending_user_inputs"),
        )
        placeholder_message = _placeholder_user_input_message(placeholder_user_values)
        if placeholder_message:
            user_facing = f"{user_facing}\n\n{placeholder_message}"
            update_kwargs["result_summary"] = user_facing
        if placeholder_user_values:
            update_kwargs["pending_user_inputs"] = None
        capability_gap_message = _capability_gap_message(blocked_packs, gaps, pending_configs)
        if capability_gap_message:
            user_facing = f"{user_facing}\n\n{capability_gap_message}"
            update_kwargs["result_summary"] = user_facing
        repair_message = _composition_repair_message(composition_outcomes)
        if repair_message:
            user_facing = f"{user_facing}\n\n{repair_message}"
            update_kwargs["result_summary"] = user_facing
        rev = None
        pe = None
        _pe_id: Optional[UUID] = None
        _rev_id: Optional[UUID] = None

        async with db_factory() as db:
            revision_repo = RevisionRepository(db)
            timeline_repo = TimelineEventRepository(db)
            changed_file_repo = ChangedFileRepository(db)
            preview_repo = PreviewRepository(db)
            project_repo = ProjectRepository(db)
            codegen_repo = CodegenRunRepository(db)
            prompt_repo = PromptRepository(db)
            from src.domain.l3_registry.service import record_codegen_candidates

            recorded_l3 = await record_codegen_candidates(
                db,
                project_id=snap.project_id,
                generation_run_id=snap.id,
                commit_sha=result.commit_sha,
                candidates=getattr(result, "pending_l3_candidates", None),
            )
            if recorded_l3:
                snap.l3_candidate_ids = recorded_l3
                await timeline_repo.create(
                    snap.project_id,
                    "l3_candidates_registered",
                    run_id=snap.id,
                    phase="apply",
                    title=f"Registered {len(recorded_l3)} project extension(s)",
                    detail_text="; ".join(recorded_l3)[:500],
                )

            if result.commit_sha:
                rev = await revision_repo.get_by_source_and_source_id(
                    snap.project_id, "codegen", snap.id
                )
                if not rev:
                    rev = await revision_repo.create(
                        project_id=snap.project_id,
                        git_sha=result.commit_sha,
                        source="codegen",
                        source_id=snap.id,
                        commit_message=result.summary,
                        created_by=snap.created_by,
                    )
                    update_kwargs["result_commit_sha"] = result.commit_sha
                    update_kwargs["revision_id"] = rev.id

                    await timeline_repo.create(
                        snap.project_id,
                        "file_modified",
                        run_id=snap.id,
                        revision_id=rev.id,
                        phase="apply",
                        title="Code changes applied",
                        detail_text=user_facing,
                    )

                    if result.files_changed_detail:
                        for f in result.files_changed_detail:
                            await changed_file_repo.create(
                                snap.project_id,
                                f["path"],
                                f.get("change_type", "modified"),
                                run_id=snap.id,
                                revision_id=rev.id if rev else None,
                                lines_added=f.get("lines_added"),
                                lines_removed=f.get("lines_removed"),
                            )
                    elif result.files_changed:
                        for fp in result.files_changed:
                            await changed_file_repo.create(
                                snap.project_id,
                                fp,
                                "modified",
                                run_id=snap.id,
                                revision_id=rev.id if rev else None,
                            )
                else:
                    update_kwargs["result_commit_sha"] = rev.git_sha
                    update_kwargs["revision_id"] = rev.id
            elif result.files_changed_detail or result.files_changed:
                if result.files_changed_detail:
                    for f in result.files_changed_detail:
                        await changed_file_repo.create(
                            snap.project_id,
                            f["path"],
                            f.get("change_type", "modified"),
                            run_id=snap.id,
                            lines_added=f.get("lines_added"),
                            lines_removed=f.get("lines_removed"),
                        )
                elif result.files_changed:
                    for fp in result.files_changed:
                        await changed_file_repo.create(
                            snap.project_id,
                            fp,
                            "modified",
                            run_id=snap.id,
                        )

            if rev:
                await project_repo.update(snap.project_id, active_revision_id=rev.id)
                pe = await preview_repo.get_latest_for_revision(snap.project_id, rev.id)
                if not pe:
                    pe = await preview_repo.create(snap.project_id, rev.id, status="provisioning")
                    await timeline_repo.create(
                        snap.project_id,
                        "preview_refresh_started",
                        run_id=snap.id,
                        revision_id=rev.id,
                        preview_id=pe.id,
                        phase="preview",
                        title="Building preview",
                    )

            blocking = _manifest_requires_user_input(update_kwargs.get("pending_user_inputs"))
            final_status = _select_final_status(blocking, blocked_packs, gaps, pending_configs)
            if final_status == "completed" and getattr(snap, "l3_candidate_ids", None):
                final_status = "completed_with_l3"
            if blocking:
                from src.domain.codegen.interruption_schemas import manifest_to_interruption

                pu = update_kwargs.get("pending_user_inputs") or {}
                update_kwargs["interruption_payload"] = manifest_to_interruption(pu).to_storage_dict()
                await timeline_repo.create(
                    snap.project_id,
                    "awaiting_user_input",
                    run_id=snap.id,
                    phase="input",
                    title="Waiting for your input",
                    detail_text="This step needs configuration values before it can finish. "
                    "Submit them in the dashboard, then we will continue automatically.",
                )
            elif placeholder_message:
                await timeline_repo.create(
                    snap.project_id,
                    "configuration_placeholders_added",
                    run_id=snap.id,
                    phase="apply",
                    title="Added temporary configuration",
                    detail_text=placeholder_message,
                )

            await codegen_repo.update_status(snap.id, final_status, **update_kwargs)

            if snap.prompt_id:
                if blocking:
                    await prompt_repo.update_response(
                        snap.prompt_id,
                        "This request needs a bit more from you: please fill in the required "
                        "configuration in the dashboard. After you submit, we will apply it and continue.",
                    )
                else:
                    await prompt_repo.update_response(snap.prompt_id, user_facing)

            from src.workers.run_chat_persist import persist_assistant_for_run_session

            cr = await codegen_repo.get_by_id(snap.id)
            if cr:
                if blocking:
                    await persist_assistant_for_run_session(
                        db,
                        cr,
                        message_type="user_input_request",
                        content=(
                            "This request needs configuration values before it can finish. "
                            "Submit them via chat with the same run_id, then we will continue."
                        ),
                        payload=update_kwargs.get("pending_user_inputs"),
                    )
                else:
                    await persist_assistant_for_run_session(
                        db,
                        cr,
                        message_type="message",
                        content=user_facing,
                    )

            if pe is not None:
                _pe_id = pe.id
            if rev is not None:
                _rev_id = rev.id

            await db.commit()

        # Publish completion / blocking events
        blocking = _manifest_requires_user_input(update_kwargs.get("pending_user_inputs"))
        summary_ws = (
            "Waiting for your configuration in the dashboard."
            if blocking
            else user_facing
        )
        final_status_ws = _select_final_status(blocking, blocked_packs, gaps, pending_configs)
        if blocking:
            # One WebSocket event only: do not emit codegen.completed (would also dual-publish
            # run.completed) and never emit legacy codegen.awaiting_user_input.
            await publish_event(
                pubsub_redis,
                str(snap.project_id),
                "run.user_input_required",
                {
                    "run_id": str(snap.id),
                    "project_id": str(snap.project_id),
                    "manifest": getattr(result, "pending_user_inputs", None),
                    "summary": summary_ws,
                    "commit_sha": result.commit_sha,
                    "run_mode": snap.run_mode,
                    "status": final_status_ws,
                },
            )
        else:
            await _ws_publish(
                "codegen.completed",
                {
                    "run_id": str(snap.id),
                    "commit_sha": result.commit_sha,
                    "summary": summary_ws,
                    "run_mode": snap.run_mode,
                    "status": final_status_ws,
                    "pending_user_inputs": False,
                },
            )

        # The preview push starts deploy.yml. workflow_run activates routing.
        # Do not enqueue a second provisioner that would dispatch the same workflow.

        deployment_intent = (
            (snap.plan or {}).get("context", {}).get("deployment_intent")
            if isinstance((snap.plan or {}).get("context"), dict)
            else None
        )
        if (
            deployment_intent == "modify_and_deploy"
            and result.commit_sha
            and _rev_id is not None
            and final_status_ws in {"completed", "completed_with_l3"}
            and not blocking
            and not pending_configs
        ):
            from src.workers.enqueue import enqueue_merge_and_deploy

            await enqueue_merge_and_deploy(
                str(snap.project_id),
                triggered_by=str(snap.created_by) if snap.created_by else None,
            )
            await _ws_publish(
                "deployment.requested",
                {
                    "run_id": str(snap.id),
                    "project_id": str(snap.project_id),
                    "revision_id": str(_rev_id) if _rev_id else None,
                    "intent": deployment_intent,
                },
            )

        if result.commit_sha and snap.prompt_id:
            await _maybe_update_persona(db_factory, snap.project_id, snap.plan)

        if reserved_credits and credit_started_at is not None:
            elapsed_minutes = max(1, ceil((monotonic() - credit_started_at) / 60))
            charged_credits = min(
                reserved_credits,
                elapsed_minutes * settings.codegen_credit_rate_per_minute,
            )
            await _settle_codegen_credit_reservation(
                db_factory=db_factory,
                project_id=snap.project_id,
                run_id=snap.id,
                charged_units=charged_credits,
                description="Settled credits from measured codegen runtime",
            )
        elif resume_finalize and snap.execution_seconds is not None:
            # GitHub callback measured the actual runner interval. Queue time and
            # Core finalization time are not billable execution time.
            execution_credits = await _record_codegen_execution_usage(
                db_factory=db_factory,
                run_id=snap.id,
                seconds=snap.execution_seconds,
                settings=settings,
            )
            await _settle_codegen_credit_reservation(
                db_factory=db_factory,
                project_id=snap.project_id,
                run_id=snap.id,
                charged_units=execution_credits,
                description="Settled provider-neutral Codegen execution usage",
            )

        logger.info(
            "codegen_run_completed",
            run_id=run_id,
            provider=result.provider,
            tokens=result.tokens_used,
            sha=result.commit_sha,
            terminal_status=final_status_ws,
        )
        return {"status": final_status_ws, "provider": result.provider}

    except Exception as e:
        error_text = str(e)
        config_error = (
            "CODEGEN_ACTIONS_" in error_text
            or "GitHub Actions codegen" in error_text
            or "workflow_dispatch was not accepted" in error_text
        )
        await _finalize_codegen_as_failed(
            db_factory=db_factory,
            project_id=snap.project_id,
            run_id=snap.id,
            transcript_lines=transcript_lines,
            ws_publish=_ws_publish,
            user_message=error_text if config_error else "Code generation failed. Please try again.",
            log_key="codegen_run_failed",
        )
        logger.error("codegen_run_failed", run_id=run_id, error=error_text)
        if config_error:
            return {"status": "failed", "reason": error_text}
        raise
    finally:
        try:
            await pubsub_redis.aclose()
        except Exception:
            pass


async def _run_plan_mode(
    *,
    db_factory,
    run_id: UUID,
    project_id: UUID,
    system_prompt: str,
    user_prompt: str,
    ai,
    settings,
    plan_context: Optional[dict] = None,
    line_callback=None,
):
    """Run the AI in planning mode: analyze repo, output markdown plan, no code changes."""
    from src.integrations.ai_provider import CodegenResult

    if settings.ai_provider in REPO_BASED_PROVIDERS:
        from src.domain.projects.repository import ProjectRepository
        from src.integrations.github_client import GitHubAppClient
        from src.integrations.github_installation import resolve_installation_id_for_workspace
        from src.workers.repo_ops import get_project_repo_path, prepare_working_tree

        async with db_factory() as db:
            project = await ProjectRepository(db).get_with_repo(project_id)
            if not project or not hasattr(project, "repo") or not project.repo:
                raise RuntimeError("Project repo not found")
            repo_full_name = project.repo.repo_full_name
            default_branch = project.repo.default_branch or "main"
            preview_branch = (
                getattr(project.repo, "preview_branch", None)
                or settings.hosting_preview_branch
                or "preview"
            )
            workspace_id = project.workspace_id

        installation_id = await resolve_installation_id_for_workspace(db_factory, workspace_id)
        github = GitHubAppClient(installation_id=installation_id)
        clone_url = await github.get_authenticated_clone_url(repo_full_name)

        repo_path = get_project_repo_path(settings.repo_workspace, workspace_id, project_id)
        meta_path = get_project_meta_path(settings.repo_workspace, workspace_id, project_id)
        os.makedirs(meta_path, exist_ok=True)
        await prepare_working_tree(
            repo_path,
            clone_url,
            branch=preview_branch,
            heal_missing_remote_from=default_branch,
        )

        from src.workers.prompt_media_sync import augment_prompt_with_media

        user_prompt = await augment_prompt_with_media(
            db_factory=db_factory,
            project_id=project_id,
            run_id=run_id,
            repo_path=repo_path,
            base_user_prompt=user_prompt,
            plan_context=dict(plan_context or {}),
        )

        result = await ai.generate_plan(
            system_prompt,
            user_prompt,
            repo_path=repo_path,
            line_callback=line_callback,
            project_meta_path=meta_path,
        )
    else:
        from src.workers.prompt_media_sync import augment_prompt_with_media

        augmented = await augment_prompt_with_media(
            db_factory=db_factory,
            project_id=project_id,
            run_id=run_id,
            repo_path=None,
            base_user_prompt=user_prompt,
            plan_context=dict(plan_context or {}),
        )
        result = await ai.generate_plan(
            system_prompt,
            augmented,
            line_callback=line_callback,
        )

    return result


async def _run_visual_deterministic_edit(
    *,
    db_factory,
    run_id: UUID,
    project_id: UUID,
    settings,
    plan_context: dict,
    ws_publish,
    transcript_lines: list[str],
):
    """Clone preview branch, apply ``data-p2hs-id`` patches, commit, push (no AI)."""
    from src.domain.codegen.repository import CodegenRunRepository
    from src.domain.projects.repository import ProjectRepository
    from src.domain.visual_edit.deterministic import apply_visual_edits_to_repo
    from src.integrations.ai_provider import CodegenResult
    from src.integrations.github_client import GitHubAppClient
    from src.integrations.github_installation import resolve_installation_id_for_workspace
    from src.workers.repo_ops import get_project_repo_path, prepare_working_tree, push_repo, run_git

    elements = plan_context.get("visual_edit_elements") or []
    if not elements:
        return CodegenResult(
            summary="No visual_edit_elements in plan context.",
            provider_invocation_failed=True,
            provider="deterministic_visual",
        )

    async with db_factory() as db:
        project = await ProjectRepository(db).get_with_repo(project_id)
        if not project or not getattr(project, "repo", None):
            return CodegenResult(
                summary="Project repo not found.",
                provider_invocation_failed=True,
                provider="deterministic_visual",
            )
        repo_full_name = project.repo.repo_full_name
        default_branch = project.repo.default_branch or "main"
        preview_branch = (
            getattr(project.repo, "preview_branch", None)
            or settings.hosting_preview_branch
            or "preview"
        )
        workspace_id = project.workspace_id

    installation_id = await resolve_installation_id_for_workspace(db_factory, workspace_id)
    github = GitHubAppClient(installation_id=installation_id)
    clone_url = await github.get_authenticated_clone_url(repo_full_name)

    repo_path = get_project_repo_path(settings.repo_workspace, workspace_id, project_id)
    await prepare_working_tree(
        repo_path,
        clone_url,
        branch=preview_branch,
        heal_missing_remote_from=default_branch,
    )

    transcript_lines.append("[visual_edit] Applying deterministic patches")
    ok, msg = await asyncio.to_thread(apply_visual_edits_to_repo, repo_path, list(elements))
    if not ok:
        transcript_lines.append(f"[visual_edit] {msg}")
        return CodegenResult(
            summary=msg or "mapping_failed",
            provider_invocation_failed=True,
            provider="deterministic_visual",
        )

    rc, out, _ = await run_git(repo_path, "status", "--porcelain")
    if rc != 0:
        return CodegenResult(
            summary=f"git status failed (code {rc})",
            provider_invocation_failed=True,
            provider="deterministic_visual",
        )
    if not (out or "").strip():
        return CodegenResult(
            summary="No file changes from visual edit.",
            provider="deterministic_visual",
            tokens_used=0,
        )

    from src.workers.repo_ops import stage_publishable_and_assert

    await stage_publishable_and_assert(repo_path)
    rc_c, _, err_c = await run_git(
        repo_path, "commit", "-m", "chore: apply visual editor changes"
    )
    if rc_c != 0:
        return CodegenResult(
            summary=f"git commit failed: {err_c}",
            provider_invocation_failed=True,
            provider="deterministic_visual",
        )

    rc_h, sha_out, _ = await run_git(repo_path, "rev-parse", "HEAD")
    sha = (sha_out or "").strip() if rc_h == 0 else ""

    rc_n, names_out, _ = await run_git(
        repo_path, "diff-tree", "--no-commit-id", "--name-only", "-r", "HEAD"
    )
    files_changed = [x.strip() for x in (names_out or "").splitlines() if x.strip()]

    await _timeline_commit(
        db_factory,
        project_id,
        "push_started",
        run_id=run_id,
        phase="apply",
        title="Saving visual edits",
    )
    if ws_publish:
        await ws_publish(
            "git.push_started",
            {"run_id": str(run_id), "branch": preview_branch, "commit_sha": sha},
        )
    from src.domain.codegen.backend_guard import validate_backend_repo

    await asyncio.to_thread(validate_backend_repo, repo_path)
    await push_repo(repo_path, branch=preview_branch, force_with_lease=True)
    await _timeline_commit(
        db_factory,
        project_id,
        "push_completed",
        run_id=run_id,
        phase="apply",
        title="Changes saved",
    )
    if ws_publish:
        await ws_publish(
            "git.push_completed",
            {
                "run_id": str(run_id),
                "commit_sha": sha,
                "branch": preview_branch,
            },
        )

    async with db_factory() as db:
        await CodegenRunRepository(db).update_checkpoint(
            run_id,
            "pushed",
            {
                "commit_sha": sha,
                "summary": "Applied visual editor changes.",
                "plan": None,
                "files_changed": files_changed,
                "files_changed_detail": None,
                "tokens_used": 0,
                "branch": preview_branch,
                "provider": "deterministic_visual",
                "pending_user_inputs": None,
            },
        )
        await db.commit()

    return CodegenResult(
        commit_sha=sha or None,
        summary="Applied visual editor changes.",
        files_changed=files_changed or None,
        provider="deterministic_visual",
        tokens_used=0,
    )


async def _run_repo_based_codegen(
    *,
    db_factory,
    run_id: UUID,
    project_id: UUID,
    system_prompt: str,
    user_prompt: str,
    ai,
    settings,
    plan_context: Optional[dict] = None,
    line_callback=None,
    ws_publish=None,
):
    """Sync working tree, run a repo-based provider, and push changes."""
    plan_context = await _ensure_generation_plan_context(
        db_factory,
        project_id,
        dict(plan_context or {}),
        user_prompt=user_prompt,
        run_id=run_id,
        settings=settings,
    )
    from src.domain.codegen.repository import CodegenRunRepository
    from src.domain.previews.adapters import get_adapter
    from src.domain.projects.repository import ProjectRepository
    from src.integrations.github_client import GitHubAppClient
    from src.integrations.github_installation import resolve_installation_id_for_workspace
    from src.workers.repo_ops import ensure_deploy_workflow, get_project_repo_path, prepare_working_tree, push_repo, run_git

    async with db_factory() as db:
        project = await ProjectRepository(db).get_with_repo(project_id)
        if not project or not hasattr(project, "repo") or not project.repo:
            raise RuntimeError("Project repo not found")
        repo_full_name = project.repo.repo_full_name
        default_branch = project.repo.default_branch or "main"
        preview_branch = getattr(project.repo, "preview_branch", None) or settings.hosting_preview_branch or "preview"
        project_type = getattr(project, "type", "html") or "html"
        project_created_at = getattr(project, "created_at", None)
        project_ref = (getattr(project, "project_ref", None) or "").strip()
        workspace_id = project.workspace_id
        from src.domain.cloud.facades import project_port_context
        port_context = await project_port_context(db, project)

    logger.info(
        "codegen_repo_info",
        run_id=str(run_id),
        repo_full_name=repo_full_name,
        preview_branch=preview_branch,
        project_type=project_type,
        project_ref=project_ref,
        workspace_id=str(workspace_id),
    )

    installation_id = await resolve_installation_id_for_workspace(db_factory, workspace_id)
    logger.info("codegen_github_installation", run_id=str(run_id), installation_id=installation_id)
    github = GitHubAppClient(installation_id=installation_id)
    clone_url = await github.get_authenticated_clone_url(repo_full_name)
    logger.info("codegen_clone_url_obtained", run_id=str(run_id), repo=repo_full_name)

    repo_path = get_project_repo_path(
        settings.repo_workspace, workspace_id, project_id
    )
    meta_path = get_project_meta_path(settings.repo_workspace, workspace_id, project_id)
    os.makedirs(meta_path, exist_ok=True)
    logger.info(
        "codegen_repo_path",
        run_id=str(run_id),
        repo_path=repo_path,
        project_meta_path=meta_path,
    )

    await _timeline_commit(
        db_factory,
        project_id,
        "cloning_started",
        run_id=run_id,
        phase="apply",
        title="Preparing workspace",
    )

    logger.info("codegen_prepare_working_tree_start", run_id=str(run_id), branch=preview_branch)
    await prepare_working_tree(
        repo_path,
        clone_url,
        branch=preview_branch,
        heal_missing_remote_from=default_branch,
    )
    logger.info("codegen_prepare_working_tree_done", run_id=str(run_id))
    generation_plan = (plan_context or {}).get("generation_plan")
    if isinstance(generation_plan, dict):
        from src.domain.codegen.composition_manifest import materialize_core_composition

        await asyncio.to_thread(materialize_core_composition, repo_path, generation_plan)
        logger.info("codegen_core_composition_materialized", run_id=str(run_id))
    from src.domain.codegen.port_adapter import install_catalog_adapters, catalog_adapter_files, LOCK_PATH
    from src.domain.codegen.backend_guard import validate_port_artifacts
    await asyncio.to_thread(install_catalog_adapters, repo_path, project_type, port_context)
    expected_port_lock = json.loads(
        catalog_adapter_files(project_type, port_context, repo_path=repo_path).get(
            LOCK_PATH, "null"
        )
    )

    # Keep platform/resolver writes out of the provider's app delta. The tree
    # was reset by prepare_working_tree, so everything written by the install
    # phase is a platform baseline and can be checkpointed separately.
    rc_platform, platform_status, platform_err = await run_git(
        repo_path, "status", "--porcelain"
    )
    if rc_platform != 0:
        raise RuntimeError(f"P2HS_CODEGEN_PREFLIGHT: base checkpoint status failed: {platform_err}")
    if platform_status.strip():
        rc_add, _, add_err = await run_git(repo_path, "add", "--all")
        if rc_add != 0:
            raise RuntimeError(f"P2HS_CODEGEN_PREFLIGHT: platform baseline staging failed: {add_err}")
        rc_commit, _, commit_err = await run_git(
            repo_path, "commit", "-m", "chore: sync P2HS platform contract"
        )
        if rc_commit != 0:
            raise RuntimeError(f"P2HS_CODEGEN_PREFLIGHT: platform baseline commit failed: {commit_err}")
    rc_clean, clean_status, clean_err = await run_git(repo_path, "status", "--porcelain")
    if rc_clean != 0 or clean_status.strip():
        raise RuntimeError(
            "P2HS_CODEGEN_PREFLIGHT: codegen base checkpoint is dirty"
            + (f": {clean_err}" if clean_err else "")
        )

    from src.workers.prompt_media_sync import augment_prompt_with_media

    user_prompt = await augment_prompt_with_media(
        db_factory=db_factory,
        project_id=project_id,
        run_id=run_id,
        repo_path=repo_path,
        base_user_prompt=user_prompt,
        plan_context=dict(plan_context or {}),
    )

    if project_ref:
        workflow_committed = await ensure_deploy_workflow(
            repo_path, project_type, project_ref,
            git_name=settings.git_committer_name,
            git_email=settings.git_committer_email,
            project_created_at=project_created_at,
            settings=settings,
        )
        logger.info("codegen_ensure_deploy_workflow", run_id=str(run_id), committed=workflow_committed)
    else:
        logger.warning("codegen_no_project_ref_skip_workflow", run_id=str(run_id))

    rc, _origin_sha, _ = await run_git(repo_path, "rev-parse", f"origin/{preview_branch}")
    origin_sha = _origin_sha.strip() if rc == 0 else ""
    logger.info("codegen_origin_sha", run_id=str(run_id), origin_sha=origin_sha[:12] if origin_sha else "unknown")

    cursor_resume_session_id: Optional[str] = None
    cursor_agent_home: Optional[str] = None
    if settings.ai_provider == AIProvider.CURSOR:
        cursor_agent_home = os.path.normpath(os.path.join(meta_path, "cursor-agent-home"))
        os.makedirs(cursor_agent_home, mode=0o700, exist_ok=True)
        async with db_factory() as db:
            from src.domain.projects.repository import ProjectSettingsRepository

            cursor_resume_session_id = await ProjectSettingsRepository(
                db
            ).get_cursor_agent_session_id(project_id)
            await db.commit()

    logger.info("codegen_ai_generate_start", run_id=str(run_id), provider=str(type(ai).__name__))
    result = await ai.generate_and_apply(
        repo_path,
        system_prompt,
        user_prompt,
        line_callback=line_callback,
        cursor_resume_session_id=cursor_resume_session_id,
        cursor_agent_home=cursor_agent_home,
        project_meta_path=meta_path,
    )
    logger.info(
        "codegen_ai_generate_done",
        run_id=str(run_id),
        commit_sha=result.commit_sha,
        summary_preview=(result.summary or "")[:200],
    )
    from dataclasses import replace as dc_replace

    if getattr(result, "provider_invocation_failed", False):
        rc2, _head_sha, _ = await run_git(repo_path, "rev-parse", "HEAD")
        head_sha = _head_sha.strip() if rc2 == 0 else ""
        if origin_sha and head_sha and head_sha != origin_sha:
            logger.warning(
                "codegen_provider_failed_local_commits_not_pushed",
                run_id=str(run_id),
                head=head_sha[:12] if head_sha else "",
            )
        return dc_replace(
            result,
            commit_sha=None,
            files_changed=None,
            files_changed_detail=None,
            pending_user_inputs=None,
        )

    # Provider edits are intentionally applied after the platform baseline
    # checkpoint, but an older checkout may still contain homemade routes or
    # stale platform adapters that the provider can touch despite the prompt.
    # Re-project the current catalog after the user-requested change so every
    # codegen run also upgrades an existing app to the current architecture.
    # This keeps the user's feature work while restoring Core-owned BFF routes
    # and the pinned Cloud contract before anything is pushed.
    await asyncio.to_thread(install_catalog_adapters, repo_path, project_type, port_context)
    expected_port_lock = json.loads(
        catalog_adapter_files(project_type, port_context, repo_path=repo_path).get(
            LOCK_PATH, "null"
        )
    )
    if expected_port_lock is not None:
        await asyncio.to_thread(validate_port_artifacts, repo_path, expected_port_lock)
    source_adoption_plan = (plan_context or {}).get("source_adoption_plan")
    source_analysis = (plan_context or {}).get("source_analysis") or []
    if isinstance(source_adoption_plan, dict) and source_analysis:
        from src.domain.codegen.source_adoption import verify_adapted_repository

        await asyncio.to_thread(
            verify_adapted_repository,
            repo_path,
            adoption_plan=source_adoption_plan,
            analyses=source_analysis,
        )

    try:
        await _apply_cloud_cutover_checkout(
            db_factory=db_factory, project_id=project_id, repo_path=repo_path,
        )
        rc_s, porcelain, _ = await run_git(repo_path, "status", "--porcelain")
        if rc_s == 0 and (porcelain or "").strip():
            from src.workers.repo_ops import stage_publishable_and_assert

            await stage_publishable_and_assert(repo_path)
            await run_git(
                repo_path, "commit", "-m",
                "chore: wire generated app to P2HS Cloud",
            )
    except Exception:
        logger.warning("codegen_cloud_cutover_failed", run_id=str(run_id), exc_info=True)

    rc2, _head_sha, _ = await run_git(repo_path, "rev-parse", "HEAD")
    head_sha = _head_sha.strip() if rc2 == 0 else ""
    logger.info(
        "codegen_sha_comparison",
        run_id=str(run_id),
        origin_sha=origin_sha[:12] if origin_sha else "unknown",
        head_sha=head_sha[:12] if head_sha else "unknown",
        changed=head_sha != origin_sha,
    )

    await _maybe_persist_cursor_agent_session(db_factory, project_id, settings, result)

    result = await _attach_pending_user_inputs_if_present(result, repo_path)
    result = await _filter_pending_user_inputs_by_stored_builder_values(
        db_factory, project_id, result
    )
    result = await _attach_fn_deploys_if_present(result, repo_path)
    result = await _attach_pack_installs_if_present(result, repo_path)
    result = await _attach_pack_compositions_if_present(result, repo_path)
    result = await _attach_l3_candidates_if_present(result, repo_path)

    # Pre-push gate, same pattern as resolve_codegen_capabilities below (which the
    # GitHub Actions runner path calls to resolve pack + provision data instead --
    # this "server" path resolves packs inline here since it isn't polling an arq
    # job): resolve pack installs/compositions HERE, before the commit is pushed, so a required
    # primitive_gap can veto the push instead of only being discovered afterward (the
    # old behavior -- these used to run in run_codegen, after this function had
    # already returned and pushed, which meant "blocked" status was a true label
    # applied to an already-shipped revision). run_codegen now reads the outcomes
    # below off `result` instead of calling these itself.
    pack_install_outcomes = await _process_pack_installs(
        db_factory=db_factory,
        project_id=project_id,
        entries=getattr(result, "pending_pack_installs", None) or [],
        user_prompt=user_prompt,
        run_id=run_id,
        sync_requirements=not bool((plan_context or {}).get("composition_reconciled")),
    )
    composition_outcomes = await _process_pack_compositions(
        db_factory=db_factory,
        project_id=project_id,
        entries=getattr(result, "pending_pack_compositions", None) or [],
        user_prompt=user_prompt,
        run_id=run_id,
    )
    result = dc_replace(
        result,
        pack_install_outcomes=pack_install_outcomes,
        composition_outcomes=composition_outcomes,
    )
    blocked_packs = [o for o in pack_install_outcomes if not o.get("ok")]
    gaps = [o for o in composition_outcomes if not o.get("ok")]
    if blocked_packs or any(g.get("criticality", "required") == "required" for g in gaps):
        gap_message = _capability_gap_message(blocked_packs, gaps) or (
            "A requested capability could not be applied."
        )
        logger.warning(
            "codegen_required_capability_gap_blocks_push",
            run_id=str(run_id),
            blocked_packs=[o.get("pack_slug") for o in blocked_packs],
            gaps=[g.get("name") for g in gaps],
        )
        return dc_replace(
            result,
            commit_sha=None,
            files_changed=None,
            files_changed_detail=None,
            summary=gap_message,
        )

    result = await _attach_provision_manifest_if_present(result, repo_path)
    result = await _attach_email_templates_manifest_if_present(result, repo_path)

    if head_sha and (not origin_sha or head_sha != origin_sha):
        result = dc_replace(result, commit_sha=head_sha)

    if origin_sha and head_sha and head_sha == origin_sha:
        logger.warning(
            "codegen_no_new_commits",
            run_id=str(run_id),
            sha=head_sha[:12],
            note="cursor agent made no new commits — skipping push and preview enqueue",
        )
        result = dc_replace(
            result,
            commit_sha=None,
            summary="The AI reviewed your request but found nothing to change.",
        )
        return result

    if result.commit_sha:
        # Packs already resolved above (this path resolves them inline, not via the
        # arq-queued gate) -- no pack_installs/pack_compositions passed here.
        await resolve_codegen_capabilities(
            {"db_factory": db_factory}, str(run_id), getattr(result, "pending_provision", None) or {}
        )
        await _timeline_commit(
            db_factory,
            project_id,
            "push_started",
            run_id=run_id,
            phase="apply",
            title="Saving changes",
        )
        if ws_publish:
            await ws_publish("git.push_started", {
                "run_id": str(run_id),
                "branch": preview_branch,
                "commit_sha": result.commit_sha,
            })
        await push_repo(repo_path, branch=preview_branch, force_with_lease=True)
        await _timeline_commit(
            db_factory,
            project_id,
            "push_completed",
            run_id=run_id,
            phase="apply",
            title="Changes saved",
        )
        if ws_publish:
            await ws_publish("git.push_completed", {
                "run_id": str(run_id),
                "commit_sha": result.commit_sha,
                "branch": preview_branch,
            })

        async with db_factory() as db:
            await CodegenRunRepository(db).update_checkpoint(
                run_id,
                "pushed",
                {
                    "commit_sha": result.commit_sha,
                    "summary": result.summary,
                    "plan": result.plan,
                    "files_changed": result.files_changed,
                    "files_changed_detail": result.files_changed_detail,
                    "tokens_used": result.tokens_used,
                    "branch": preview_branch,
                    "provider": result.provider
                    or (
                        settings.ai_provider.value
                        if hasattr(settings.ai_provider, "value")
                        else str(settings.ai_provider)
                    ),
                    "pending_user_inputs": getattr(result, "pending_user_inputs", None),
                    "pending_provision": getattr(result, "pending_provision", None),
                    "pending_email_templates": getattr(result, "pending_email_templates", None),
                },
            )
            await db.commit()

        build_validation = getattr(settings, "codegen_build_validation_enabled", False) or False
        if build_validation and project_type in ("react_app", "nuxt_app", "html"):
            await _timeline_commit(
                db_factory,
                project_id,
                "build_started",
                run_id=run_id,
                phase="validation",
                title="Running build validation",
            )
            try:
                adapter = get_adapter(project_type)
                artifact_path = await adapter.install_deps(repo_path)
                await adapter.build(artifact_path)
                await _timeline_commit(
                    db_factory,
                    project_id,
                    "build_succeeded",
                    run_id=run_id,
                    phase="validation",
                    title="Build succeeded",
                )
            except Exception as build_err:
                await _timeline_commit(
                    db_factory,
                    project_id,
                    "build_failed",
                    run_id=run_id,
                    phase="validation",
                    title="Build failed",
                    detail_text=str(build_err),
                )
                raise RuntimeError(f"Build validation failed: {build_err}") from build_err

    return result