"""Codegen domain - business logic and orchestration.""" from __future__ import annotations import json from typing import Optional from uuid import UUID import structlog from src.core.exceptions import NotFoundError, ValidationError from src.domain.codegen.interfaces import ProjectInfoFacade from src.domain.codegen.prompt_wrapper import ( build_agent_user_prompt, build_plan_system_prompt, build_plan_user_prompt, build_system_prompt, build_wrapped_prompt, merge_plan_user_prompt, ) from src.domain.codegen.platform_backend import build_platform_backend_capabilities from src.domain.codegen.repository import CodegenRunRepository from src.domain.codegen.run_lifecycle import USER_INPUT_BLOCKING_STATUSES from src.domain.codegen.schemas import CodegenActionsCompleteBody, CodegenActionsProgressBody from src.domain.codegen.user_inputs_manifest import validate_submitted_values from src.domain.manifest.service import ManifestService from src.domain.projects.repository import ( ProjectEnvVarRepository, ProjectRepository, ProjectSettingsRepository, ) from src.domain.projects.runtime_invariants import ( ProjectRuntimeInvariants, extract_runtime_invariants, ) logger = structlog.get_logger() class CodegenService: def __init__( self, codegen_repo: CodegenRunRepository, project_facade: ProjectInfoFacade, project_settings_repo: Optional[ProjectSettingsRepository] = None, env_var_repo: Optional[ProjectEnvVarRepository] = None, manifest_service: Optional[ManifestService] = None, ): self._codegen_repo = codegen_repo self._project_facade = project_facade self._project_settings_repo = project_settings_repo self._env_var_repo = env_var_repo self._manifest_service = manifest_service async def _platform_backend_for_prompt(self, project_id: UUID) -> Optional[dict]: try: from src.domain.codegen.legacy_cloud_cutover import ensure_cloud_auth_for_project await ensure_cloud_auth_for_project(self._codegen_repo._db, project_id) except Exception: pass if self._manifest_service is None: return None manifest = await self._manifest_service.get_manifest(project_id, compact=False) runtime_contract = None try: from src.domain.codegen.runtime_contract import resolve_runtime_contract project = await ProjectRepository(self._codegen_repo._db).get_by_id(project_id) if project: runtime_contract = ( await resolve_runtime_contract(self._codegen_repo._db, project) ).model_dump() except Exception: logger.warning("runtime_contract.resolve_failed", project_id=str(project_id), exc_info=True) return build_platform_backend_capabilities(manifest, runtime_contract=runtime_contract) async def _update_runtime_invariants(self, project_id: UUID, user_prompt: str) -> None: """Merge new signal from `user_prompt` onto the project's persisted invariants. Deterministic, cheap, and previous-preserving -- see runtime_invariants.py. Never raises; a failure here must not block the codegen run itself. """ try: project_repo = ProjectRepository(self._codegen_repo._db) project = await project_repo.get_by_id(project_id) if not project: return previous = ProjectRuntimeInvariants.model_validate(project.runtime_invariants or {}) updated = extract_runtime_invariants(user_prompt, previous) if updated == previous: return await project_repo.update_runtime_invariants(project_id, updated.model_dump()) logger.info( "runtime_invariants.updated", project_id=str(project_id), admin_panel_path=updated.admin_panel_path, customer_login_path=updated.customer_login_path, auth_modes=updated.auth_modes, persistence_required=updated.persistence_required, admin_must_read_persisted=updated.admin_must_read_persisted, email_preference=updated.email_preference, source=updated.source, ) # These two are the only invariants project_auth's Cloud payload actually # carries (see ProjectAuthService._push_auth_config) -- a locked admin # path used to sit in Postgres and never reach Cloud until some unrelated # auth-config edit happened to trigger a push. Don't push on every prompt, # only when what Cloud actually needs changed. if ( updated.admin_panel_path != previous.admin_panel_path or updated.customer_login_path != previous.customer_login_path ): await self._push_runtime_invariants_to_cloud(project_id) except Exception: logger.warning("runtime_invariants.update_failed", project_id=str(project_id), exc_info=True) async def _push_runtime_invariants_to_cloud(self, project_id: UUID) -> None: """Best-effort resync -- a project with no auth configured yet has nothing to push (push_config_with_roles no-ops), and a push failure here must not block the codegen run itself.""" try: from src.domain.project_auth.repository import ( ProjectAuthConfigRepository, ProjectAuthKeyRepository, ProjectAuthRoleRepository, ) from src.domain.project_auth.service import ProjectAuthService from src.integrations.cloud_control_client import CloudControlClient db = self._codegen_repo._db auth_service = ProjectAuthService( config_repo=ProjectAuthConfigRepository(db), key_repo=ProjectAuthKeyRepository(db), role_repo=ProjectAuthRoleRepository(db), cloud_client=CloudControlClient(), ) await auth_service.push_config_with_roles(project_id) except Exception: logger.warning("runtime_invariants.cloud_push_failed", project_id=str(project_id), exc_info=True) async def _form_policy_for_prompt(self, project_id: UUID) -> dict: if self._project_settings_repo is None: return {} return await self._project_settings_repo.get_form_policy(project_id) async def supersede_active_run_for_new_chat(self, project_id: UUID) -> Optional[UUID]: """Cancel any run that still reserves the project slot and clear Cursor resume id. A request **without** ``run_id`` is treated as a new chat: the previous run is canceled and the next codegen job will not pass ``--resume`` to Cursor (fresh agent session). Continuation requests **with** ``run_id`` must not call this — they keep the stored Cursor session id so the agent can finish pending work. """ active = await self._codegen_repo.get_active_reserving_run(project_id) if not active: return None await self._codegen_repo.cancel_run( active.id, error_message="Superseded by new chat request", ) if self._project_settings_repo is not None: await self._project_settings_repo.clear_cursor_agent_session_id(project_id) return active.id async def create_run( self, project_id: UUID, user_prompt: str, prompt_id: Optional[UUID] = None, created_by: Optional[UUID] = None, selected_route: Optional[str] = None, viewport: Optional[str] = None, run_mode: str = "agent", selected_elements: Optional[list] = None, parent_plan_id: Optional[UUID] = None, attachment_ids: Optional[list[UUID]] = None, voice_transcript: Optional[str] = None, credit_estimate: Optional[dict[str, int]] = None, execution_target: str = "cloud", device_id: Optional[UUID] = None, supersede_active: bool = True, generation_plan: Optional[dict] = None, generation_request: Optional[dict] = None, provider_override: Optional[str] = None, prompt_source: Optional[str] = None, initiating_grant_id: Optional[UUID] = None, ): from src.domain.projects.discovery_repository import DiscoveryRepository from src.domain.projects.discovery_service import require_discovery_complete await self._codegen_repo.lock_project_for_run_creation(project_id) discovery_project = await DiscoveryRepository(self._codegen_repo._db).get(project_id) require_discovery_complete(discovery_project) project_type = await self._project_facade.get_project_type(project_id) if not project_type: raise NotFoundError("Project not found") understanding = None active = await self._codegen_repo.get_active_reserving_run(project_id) if active: if not supersede_active: from src.core.exceptions import ConflictError raise ConflictError("Project already has an active build") await self.supersede_active_run_for_new_chat(project_id) instruction_content = await self._project_facade.get_project_instruction_content(project_id) await self._update_runtime_invariants(project_id, user_prompt or "") platform_backend = await self._platform_backend_for_prompt(project_id) form_policy = await self._form_policy_for_prompt(project_id) # Plan mode gets a read-only system prompt; all other modes get the full agent prompt. if run_mode == "plan": system_prompt = build_plan_system_prompt( project_type=project_type, instruction_content=instruction_content, selected_route=selected_route, viewport=viewport, platform_backend=platform_backend, form_policy=form_policy, ) else: system_prompt = build_system_prompt( project_type=project_type, instruction_content=instruction_content, selected_route=selected_route, viewport=viewport, selected_elements=selected_elements if run_mode == "visual_edit" else None, platform_backend=platform_backend, form_policy=form_policy, ) #wrapped = build_wrapped_prompt(user_prompt or "") if run_mode == "plan": wrapped = build_plan_user_prompt(user_prompt or "") else: wrapped = build_agent_user_prompt(user_prompt or "") run = await self._codegen_repo.create( project_id=project_id, prompt_id=prompt_id, created_by=created_by, run_mode=run_mode, selected_elements=selected_elements, parent_plan_id=parent_plan_id, execution_target=execution_target, device_id=device_id, ) ctx: dict = { "selected_route": selected_route, "viewport": viewport, "project_type": project_type, "run_mode": run_mode, } from src.domain.codegen.provider_registry import CodegenConfigResolver resolved_codegen = await CodegenConfigResolver(self._codegen_repo._db).resolve_optional(project_id) if resolved_codegen: ctx.update({ "codegen_provider_override": provider_override or resolved_codegen.provider.slug, "codegen_model_key": resolved_codegen.model.model_key, "codegen_cli_model_value": resolved_codegen.model.cli_model_value, "codegen_funding_source": resolved_codegen.funding_source, "codegen_credential_id": str(resolved_codegen.credential.id) if resolved_codegen.credential else None, "codegen_configuration_source": "database", }) if provider_override: ctx["codegen_provider_override"] = provider_override if understanding: ctx["project_understanding"] = understanding.model_dump() if attachment_ids: ctx["attachment_ids"] = [str(x) for x in attachment_ids] vt = (voice_transcript or "").strip() if vt: ctx["voice_transcript"] = vt if credit_estimate: ctx["credit_estimate"] = credit_estimate if prompt_source: ctx["prompt_source"] = prompt_source if initiating_grant_id: ctx["initiating_grant_id"] = str(initiating_grant_id) if generation_request: ctx["source_adoption"] = generation_request.get("source_adoption") ctx["deployment_intent"] = generation_request.get("deployment_intent", "modify_only") ctx["source_analysis"] = generation_request.get("source_analysis", []) ctx["source_adoption_plan"] = generation_request.get("source_adoption_plan") if generation_request.get("import_snapshot"): ctx["import_snapshot"] = generation_request["import_snapshot"] plan = { "system_prompt": system_prompt, "user_prompt": wrapped, "original_user_prompt": user_prompt or "", "context": ctx, } if generation_plan is not None: plan = { **plan, "schema": "generation_plan.v1", "generation_plan": generation_plan, "compatibility": plan, } if generation_request is not None: request = dict(generation_request) request["generation_id"] = str(run.id) if request.get("schema") == "source_adoption_request.v1": plan["source_adoption_request"] = request else: plan["codegen_request"] = request await self._codegen_repo.update_status(run.id, "queued", plan=plan) from src.domain.codegen.execution import resolve_codegen_execution from src.config import get_settings execution = resolve_codegen_execution(get_settings()) if resolved_codegen: registry = CodegenConfigResolver(self._codegen_repo._db) await registry.create_snapshot( run.id, resolved_codegen, adapter_version="1", execution=execution, pricing_version=get_settings().codegen_execution_pricing_version, ) await registry.record_usage(run.id, resolved_codegen.funding_source, usage_source="unavailable") else: # Legacy runs without DB-selected AI configuration still receive an # immutable execution snapshot. ``legacy`` describes the AI side only; # placement is authoritative and remains provider-neutral. from src.domain.codegen.provider_models import CodegenExecutionSnapshot fields = execution.snapshot() self._codegen_repo._db.add(CodegenExecutionSnapshot( run_id=run.id, provider_slug="legacy", model_key="legacy", funding_source="p2hs", execution_provider=fields["provider"], execution_target=fields["target"], runner_class=fields["runner_class"], execution_config_version=fields["execution_config_version"], pricing_version=get_settings().codegen_execution_pricing_version, )) await self._codegen_repo._db.flush() discovery = getattr(discovery_project, "discovery_json", None) if discovery and not discovery.get("initial_run_id"): discovery_project.discovery_json = {**discovery, "initial_run_id": str(run.id)} return run async def execute_plan( self, plan_run_id: UUID, project_id: UUID, created_by: Optional[UUID] = None, ): """Approve a plan_ready run: same row transitions to agent mode and re-queues.""" plan_run = await self._codegen_repo.get_by_id(plan_run_id) if not plan_run or plan_run.project_id != project_id: raise NotFoundError("Plan run not found") if plan_run.run_mode != "plan": raise ValidationError("Run is not a plan mode run") if plan_run.status != "plan_ready": raise ValidationError(f"Plan is not ready for execution (status: {plan_run.status})") original_prompt = plan_run.plan.get("original_user_prompt", "") if plan_run.plan else "" selected_route = (plan_run.plan.get("context") or {}).get("selected_route") viewport = (plan_run.plan.get("context") or {}).get("viewport") project_type = await self._project_facade.get_project_type(project_id) instruction_content = await self._project_facade.get_project_instruction_content(project_id) await self._update_runtime_invariants(project_id, original_prompt) platform_backend = await self._platform_backend_for_prompt(project_id) form_policy = await self._form_policy_for_prompt(project_id) base_system_prompt = build_system_prompt( project_type=project_type or "html", instruction_content=instruction_content, selected_route=selected_route, viewport=viewport, platform_backend=platform_backend, form_policy=form_policy, ) if plan_run.plan_document: system_prompt = ( base_system_prompt + "\n\n## Approved Implementation Plan\n" + plan_run.plan_document + "\n\nImplement exactly this plan. Follow each step in order." ) else: system_prompt = base_system_prompt prior_ctx = dict((plan_run.plan or {}).get("context") or {}) merged_ctx = { **prior_ctx, "selected_route": selected_route, "viewport": viewport, "project_type": project_type, "run_mode": "agent", "executed_from_plan": str(plan_run_id), } wrapped = merge_plan_user_prompt(original_prompt, merged_ctx) plan_data = { "system_prompt": system_prompt, "user_prompt": wrapped, "original_user_prompt": original_prompt, "context": merged_ctx, } updated = await self._codegen_repo.reset_run_for_agent_execution(plan_run_id, plan_data) if not updated: raise NotFoundError("Plan run not found") return updated async def cancel_run(self, run_id: UUID, project_id: UUID): """Cancel a plan_ready, queued, pending, or running run.""" run = await self._codegen_repo.get_by_id(run_id) if not run or run.project_id != project_id: raise NotFoundError("Run not found") result = await self._codegen_repo.cancel_run(run_id) if result is None: raise ValidationError(f"Run cannot be canceled (status: {run.status})") return result async def get_run(self, run_id: UUID): run = await self._codegen_repo.get_by_id(run_id) if not run: raise NotFoundError("Codegen run not found") return run async def list_runs(self, project_id: UUID): return await self._codegen_repo.list_for_project(project_id) async def retry_run(self, run_id: UUID, project_id: UUID, created_by: Optional[UUID] = None): """Re-queue the same run row after failure/cancel. Rebuilds `platform_backend` (and therefore `runtime_contract`) fresh instead of replaying the plan stored at the original queue time -- capabilities/routes can change between a failed attempt and its retry (e.g. an email connector added, or the admin path locked by a different run), and a stale contract is exactly the class of bug this whole effort is against. `blocked` (a required capability gap reported by resolveCapabilitiesBeforePush, e.g. contact-capture waiting on a notify-email connector) is retryable for the same reason: the fresh platform_backend rebuild is exactly what re-resolves the gap once the owner configures the missing capability, and unlike `failed` the executor never pushed a commit for that attempt, so there is nothing else to resume from. """ # Locked: two simultaneous retries of one run must not both pass the status check and # dispatch onto the same work branch. The second waits, then sees the run queued. src = await self._codegen_repo.get_by_id(run_id, for_update=True) if not src or src.project_id != project_id: raise NotFoundError("Run not found") if src.status not in ("failed", "canceled", "blocked"): raise ValidationError(f"Only failed, canceled, or blocked runs can be retried (status: {src.status})") if not src.plan: raise ValidationError("Run has no stored plan to retry") plan = dict(src.plan) ctx = dict(plan.get("context") or {}) run_mode = ctx.get("run_mode", "agent") project_type = ctx.get("project_type") or await self._project_facade.get_project_type(project_id) instruction_content = await self._project_facade.get_project_instruction_content(project_id) platform_backend = await self._platform_backend_for_prompt(project_id) form_policy = await self._form_policy_for_prompt(project_id) if run_mode == "plan": system_prompt = build_plan_system_prompt( project_type=project_type, instruction_content=instruction_content, selected_route=ctx.get("selected_route"), viewport=ctx.get("viewport"), platform_backend=platform_backend, form_policy=form_policy, ) else: system_prompt = build_system_prompt( project_type=project_type, instruction_content=instruction_content, selected_route=ctx.get("selected_route"), viewport=ctx.get("viewport"), selected_elements=getattr(src, "selected_elements", None) if run_mode == "visual_edit" else None, platform_backend=platform_backend, form_policy=form_policy, ) plan["system_prompt"] = system_prompt # Work-branch mode: the branch already holds the agent's output, so a retry resumes from it # (skipping the agent after a platform/Cloud failure, or sending it back with the exact # errors after an invalid manifest) instead of regenerating the frontend. from src.config import get_settings as _get_settings from src.domain.codegen import work_branch as _wb _settings = _get_settings() _state = ctx.get("work_branch") ctx.pop("resume", None) ctx.pop("correction", None) if _settings.codegen_work_branch_enabled and _wb.valid_state(_state, run_id): _resume = _wb.resume_plan( _state, max_corrections=int(_settings.codegen_manifest_correction_attempts) ) _new_state, _updates = _wb.begin_retry(_state, _resume) ctx["work_branch"] = _new_state ctx.update(_updates) plan["context"] = ctx if created_by is not None: src.created_by = created_by await self._codegen_repo._db.flush() await self._codegen_repo.reset_run_from_failure(run_id) refreshed = await self._codegen_repo.requeue_with_plan(run_id, plan) if not refreshed: raise NotFoundError("Run not found") return refreshed @staticmethod def _apply_work_report(run, body: CodegenActionsCompleteBody) -> None: """Record where the run branch checkpoint is (work_branch.py). Trusts only the state Core stored itself: a report that names another branch is ignored, and a checkpoint can only move forward. Never raises: losing this pointer must not fail a run (the branch has the commits).""" from src.domain.codegen import work_branch as wb try: plan = dict(run.plan or {}) ctx = dict(plan.get("context") or {}) state = ctx.get("work_branch") if not wb.valid_state(state, run.id): return if body.work_branch and body.work_branch != state["branch"]: return status = { "blocked": "blocked", "failed": "failed", "canceled": "failed", "needs_continuation": "active", }.get(body.status, "promoted") failed = status in {"failed", "blocked"} ctx["work_branch"] = wb.apply_report( state, status=status, checkpoint=body.checkpoint, checkpoint_sha=body.checkpoint_sha, failure_class=body.failure_class if failed else None, retryable=body.retryable if failed else None, message=(body.error_message or body.summary) if failed else None, correction=body.correction if failed else None, ) plan["context"] = ctx run.plan = plan except Exception: return async def _settle_external_execution_failure(self, run, body: CodegenActionsCompleteBody) -> None: """Record provider cost and release customer credits for a non-success callback. Provider cost can exist even when P2HS elects not to charge a customer for an infrastructure failure. The existing reservation operation remains the idempotent financial boundary. """ from decimal import Decimal from sqlalchemy import select from src.config import get_settings from src.domain.codegen.execution import CodegenExecutionProvider, CodegenExecutionTarget from src.domain.codegen.execution_pricing import ExecutionUsage, calculate_execution_price from src.domain.codegen.provider_models import CodegenExecutionSnapshot, CodegenUsage 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 db = self._codegen_repo._db settings = get_settings() snapshot = await db.get(CodegenExecutionSnapshot, run.id) seconds = max(0, int(body.execution_seconds or 0)) 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: return 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, ) 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="external_callback") 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"} project = await ProjectRepository(db).get_by_id(run.project_id) if not project: return ledger = WorkspaceCreditLedgerRepository(db) reservation = 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: # Failed/canceled/blocked external executions are not customer-billable. await reservation.settle( workspace_id=project.workspace_id, reference=f"codegen:{run.id}", charged_units=0, description="Released reservation after external Codegen execution failure", ) except ValueError: pass async def report_run_complete(self, run_id: UUID, body: CodegenActionsCompleteBody) -> dict: """Accept a completion callback from a repo-local executor (GitHub Actions or a paired desktop device) and resume Core finalization. The executor only executes code. Core owns the run row, final revision creation, usage accounting, and preview enqueue. Shared by both executors via ``require_run_grant_or_actions_token`` — this method itself is executor-agnostic. """ from src.core.public_output import redact_public_output # Executor transcripts and summaries are replayed to project clients. # Keep provider/repository URLs and runner paths out of persisted output. body.transcript = redact_public_output(body.transcript) body.summary = redact_public_output(body.summary) body.error_message = redact_public_output(body.error_message) current = await self._codegen_repo.get_by_id(run_id) if not current: raise NotFoundError("Run not found") if current.status in ("completed", "completed_with_limitations", "completed_with_l3", "blocked", "failed", "canceled"): return {"status": "skipped", "reason": f"already {current.status}"} previous = dict(current.checkpoint_payload or {}) # Continuations are separate GitHub jobs under one logical run. Carry the # measured runner interval forward so the final settlement charges the # attempts exactly once under the existing run reservation. execution_total = int(previous.get("execution_seconds") or 0) + int(body.execution_seconds or 0) body.execution_seconds = execution_total or None if current.worker_checkpoint == "pushed" and body.status not in ("completed", "completed_with_l3"): return {"status": "skipped", "reason": "already finalizing"} if (body.continuation_attempt is not None and body.continuation_attempt < int(previous.get("continuation_attempt") or 0)): return {"status": "skipped", "reason": "stale executor attempt"} self._apply_work_report(current, body) if body.status == "deploying": if not body.commit_sha: raise ValidationError("Deployment requires a commit SHA") if current.worker_checkpoint == "pushed": return {"status": "skipped", "reason": "already finalizing"} await self._codegen_repo.update_checkpoint( run_id, "github_actions_deploying", {**previous, **body.model_dump(mode="json")} ) await self._codegen_repo.update_status( run_id, "preview_refreshing", result_commit_sha=body.commit_sha, agent_transcript=body.transcript, ) await self._codegen_repo.commit() from src.core.pubsub import get_pubsub_redis from src.core.run_events import publish_project_event_dual redis = await get_pubsub_redis() try: await publish_project_event_dual(redis, str(current.project_id), "codegen.status_changed", { "run_id": str(run_id), "status": "preview_refreshing", "stage": "preview", "message": "Code saved. Building and deploying your preview.", "commit_sha": body.commit_sha, }) finally: await redis.aclose() return {"status": "deploying", "run_id": str(run_id)} if current.worker_checkpoint == "github_actions_deploying": if body.commit_sha != previous.get("commit_sha"): raise ValidationError("Deployment callback does not match the pending commit") if body.status == "needs_continuation": return await self._continue_github_actions_run(current, body) if body.status == "blocked" and body.failure_class == "invalid_requirement" and body.correction: # The agent's own capability manifest was invalid. Its work is saved on the run branch, # so send it back with the exact errors instead of ending the run. corrected = await self._correct_github_actions_run(current) if corrected: return corrected if body.status == "blocked": # A required primitive_gap/blocked pack install -- not an error (unlike # "failed"), a required requirement genuinely wasn't fulfilled. The # runner never pushed for this attempt (see resolveCapabilitiesBeforePush # in actions_pipeline.py), so there's no commit_sha to record here. message = body.error_message or body.summary or "A requested capability could not be applied." await self._settle_external_execution_failure(current, body) await self._codegen_repo.update_status( run_id, "blocked", error_message=message, agent_transcript=body.transcript, ) await self._codegen_repo.commit() from src.core.pubsub import get_pubsub_redis from src.core.run_events import publish_project_event_dual redis = await get_pubsub_redis() try: await publish_project_event_dual(redis, str(current.project_id), "codegen.failed", { "run_id": str(run_id), "status": "blocked", "message": message, }) finally: await redis.aclose() return {"status": "blocked"} if body.status not in ("completed", "completed_with_l3"): message = body.error_message or body.summary or f"Executor reported {body.status}" await self._settle_external_execution_failure(current, body) await self._codegen_repo.update_status( run_id, "failed" if body.status == "failed" else "canceled", error_message=message, agent_transcript=body.transcript, ) await self._codegen_repo.commit() from src.core.pubsub import get_pubsub_redis from src.core.run_events import publish_project_event_dual redis = await get_pubsub_redis() try: await publish_project_event_dual(redis, str(current.project_id), "codegen.failed", { "run_id": str(run_id), "status": body.status, "message": message, "commit_sha": body.commit_sha, }) finally: await redis.aclose() return {"status": body.status} if not body.commit_sha: await self._settle_external_execution_failure(current, body) await self._codegen_repo.update_status( run_id, "failed", error_message="Executor completed without a commit SHA", agent_transcript=body.transcript, ) return {"status": "failed", "reason": "missing_commit_sha"} from src.domain.l3_registry.service import record_codegen_candidates recorded_l3 = await record_codegen_candidates( self._codegen_repo._db, project_id=current.project_id, generation_run_id=run_id, commit_sha=body.commit_sha, candidates=body.l3_candidates, ) try: from src.domain.codegen.legacy_cloud_cutover import ( discovered_users_from_payload, ensure_cloud_auth_for_project, import_legacy_users_for_project, ) await ensure_cloud_auth_for_project(self._codegen_repo._db, current.project_id) if body.legacy_users: await import_legacy_users_for_project( self._codegen_repo._db, current.project_id, discovered_users_from_payload(body.legacy_users), ) except Exception: pass if current.status == "preview_refreshing": await self._codegen_repo.update_status(run_id, "running") await self._codegen_repo.update_checkpoint( run_id, "pushed", { "commit_sha": body.commit_sha, "summary": body.summary or "Codegen completed.", "plan": None, "files_changed": body.files_changed, "files_changed_detail": body.files_changed_detail, "tokens_used": body.tokens_used or 0, "branch": body.branch, "provider": body.provider or "github_actions", "pending_user_inputs": body.pending_user_inputs, "pending_provision": body.pending_provision, "pending_email_templates": body.pending_email_templates, "pending_pack_install_outcomes": body.pending_pack_install_outcomes, "pending_pack_composition_outcomes": body.pending_pack_composition_outcomes, "l3_candidate_ids": sorted(set(body.l3_candidate_ids) | set(recorded_l3)), "execution_seconds": body.execution_seconds, }, ) if body.transcript: current.agent_transcript = body.transcript[:80000] from src.workers.enqueue import enqueue_codegen_run await self._codegen_repo.commit() await enqueue_codegen_run(str(run_id), project_id=str(current.project_id)) return {"status": "accepted", "run_id": str(run_id)} async def _correct_github_actions_run(self, run) -> Optional[dict]: """Re-dispatch the SAME run so the agent fixes its own invalid manifest, on top of its work. The run branch already holds the agent's output, so nothing is regenerated: the new job checks out that branch and the prompt carries only the errors (work_branch.build_correction_appendix). Bounded by ``codegen_manifest_correction_attempts``. Returns None (caller then ends the run as blocked, as before) when this is not possible or the dispatch fails.""" import structlog from src.config import get_settings from src.core.database import async_session_factory from src.domain.codegen import work_branch as wb from src.domain.codegen.actions_pipeline import dispatch_codegen_actions_run settings = get_settings() if not settings.codegen_work_branch_enabled or (getattr(run, "execution_target", None) or "cloud") == "local": return None plan = dict(run.plan or {}) ctx = dict(plan.get("context") or {}) state = ctx.get("work_branch") resume = wb.resume_plan( state, max_corrections=int(settings.codegen_manifest_correction_attempts) ) if not resume or resume.get("skip_agent") or not resume.get("correction"): return None new_state, updates = wb.begin_retry(state, resume) ctx["work_branch"] = new_state ctx.update(updates) plan["context"] = ctx previous_plan = run.plan run.plan = plan try: dispatch = await dispatch_codegen_actions_run( db_factory=async_session_factory, run_id=run.id, project_id=run.project_id, system_prompt=str(plan.get("system_prompt") or ""), user_prompt=str(plan.get("user_prompt") or plan.get("original_user_prompt") or ""), plan_context=ctx, ) except Exception as exc: structlog.get_logger().warning("manifest_correction_dispatch_failed", run_id=str(run.id), error_type=type(exc).__name__) run.plan = previous_plan return None await self._codegen_repo.update_checkpoint( run.id, "github_actions_dispatched", {**dict(run.checkpoint_payload or {}), **dispatch} ) await self._codegen_repo.refresh_started_at(run.id) await self._codegen_repo.commit() return {"status": "correcting", "run_id": str(run.id), "correction_attempt": new_state["correction_attempts"]} async def _continue_github_actions_run( self, run, body: CodegenActionsCompleteBody, ) -> dict: """Save executor progress and dispatch a new GitHub Actions job for the same run.""" from src.config import get_settings from src.core.database import async_session_factory from src.domain.codegen.actions_pipeline import dispatch_codegen_actions_run if (getattr(run, "execution_target", None) or "cloud") == "local": await self._codegen_repo.update_status( run.id, "failed", error_message="Local executor timed out before finishing the run.", agent_transcript=body.transcript, ) return {"status": "failed", "reason": "local_timeout"} settings = get_settings() max_attempts = int(settings.codegen_actions_max_continuations) prev = dict(run.checkpoint_payload or {}) attempt = int(body.continuation_attempt or 0) if attempt <= 0: attempt = int(prev.get("continuation_attempt") or 0) + 1 elif attempt <= int(prev.get("continuation_attempt") or 0): return {"status": "continuing", "run_id": str(run.id), "continuation_attempt": attempt} if not body.commit_sha: await self._codegen_repo.update_status( run.id, "failed", error_message="Executor timed out without a checkpoint commit", agent_transcript=body.transcript, ) return {"status": "failed", "reason": "timeout_without_progress"} if attempt > max_attempts: await self._codegen_repo.update_status( run.id, "failed", error_message=( f"Codegen hit the {settings.codegen_actions_timeout_minutes}-minute " f"GitHub Actions limit {max_attempts} times without finishing." ), agent_transcript=body.transcript, ) return {"status": "failed", "reason": "max_continuations_exceeded"} plan = dict(run.plan or {}) project_id = run.project_id run_id = run.id if body.transcript: run.agent_transcript = (body.transcript or "")[:80000] if body.cursor_session_id and self._project_settings_repo is not None: await self._project_settings_repo.set_cursor_agent_session_id( project_id, body.cursor_session_id ) continuing_payload = { **prev, "commit_sha": body.commit_sha, "summary": body.summary, "files_changed": body.files_changed, "cursor_session_id": body.cursor_session_id, "continuation_attempt": attempt, "dispatch_commit_sha": body.commit_sha, "execution_seconds": body.execution_seconds, } await self._codegen_repo.update_checkpoint( run_id, "github_actions_continuing", continuing_payload ) await self._codegen_repo.refresh_started_at(run_id) await self._codegen_repo.commit() dispatch = await dispatch_codegen_actions_run( db_factory=async_session_factory, run_id=run_id, project_id=project_id, system_prompt=str(plan.get("system_prompt") or ""), user_prompt=str(plan.get("user_prompt") or plan.get("original_user_prompt") or ""), plan_context=dict(plan.get("context") or {}), continuation={ "attempt": attempt, "cursor_session_id": body.cursor_session_id, "summary": body.summary, }, ) await self._codegen_repo.update_checkpoint( run_id, "github_actions_dispatched", { **continuing_payload, **dispatch, "dispatch_commit_sha": dispatch.get("dispatch_commit_sha"), }, ) await self._codegen_repo.refresh_started_at(run_id) await self._codegen_repo.commit() return { "status": "continuing", "run_id": str(run_id), "continuation_attempt": attempt, "commit_sha": body.commit_sha, } async def report_run_progress(self, run_id: UUID, body: CodegenActionsProgressBody) -> dict: """Append repo-local executor output and publish it to project WebSocket clients. Shared by GitHub Actions and a paired desktop device (see ``report_run_complete``). """ from src.core.pubsub import get_pubsub_redis from src.core.run_events import publish_project_event_dual current = await self._codegen_repo.get_by_id(run_id) if not current: raise NotFoundError("Run not found") if current.status in ("completed", "completed_with_limitations", "completed_with_l3", "blocked", "failed", "canceled"): return {"status": "skipped", "reason": f"already {current.status}"} from src.core.public_output import redact_public_output line = redact_public_output((body.text or "").strip("\r\n")) or "" if not line: return {"status": "ignored"} prefix = "[stderr] " if body.stream == "stderr" else "" previous = current.agent_transcript or "" current.agent_transcript = (previous + ("\n" if previous else "") + prefix + line)[-80000:] # Do not hold a pooled Postgres connection while publishing to Redis/WebSockets. await self._codegen_repo.commit() redis = await get_pubsub_redis() try: payload = { "run_id": str(run_id), "text": line, "type": "output", "stream": body.stream, "stage": "generating", } if body.detail_format: payload["detail_format"] = body.detail_format await publish_project_event_dual( redis, str(current.project_id), "codegen.ai_output", payload, ) finally: await redis.aclose() return {"status": "accepted"} async def supply_user_inputs( self, run_id: UUID, project_id: UUID, values: dict[str, str], created_by: Optional[UUID], ): """Persist owner values and re-queue the same run to apply them (no fork).""" if self._project_settings_repo is None: raise ValidationError("Project settings are not configured for this endpoint") run = await self._codegen_repo.get_by_id(run_id) if not run or run.project_id != project_id: raise NotFoundError("Run not found") if run.status not in USER_INPUT_BLOCKING_STATUSES: raise ValidationError( f"Run is not waiting for user input (status: {run.status})" ) manifest = run.pending_user_inputs if not manifest or not manifest.get("items"): raise ValidationError("This run has no pending user inputs to supply") validate_submitted_values(manifest, values) await self._project_settings_repo.merge_builder_user_values(project_id, values) # Fan out into project_env_vars (source of truth for runtime env) if self._env_var_repo is not None: await _fan_out_inputs_to_env_vars( env_var_repo=self._env_var_repo, project_id=project_id, values=values, manifest_items=manifest.get("items") or [], ) project_type = await self._project_facade.get_project_type(project_id) if not project_type: raise NotFoundError("Project not found") instruction_content = await self._project_facade.get_project_instruction_content(project_id) platform_backend = await self._platform_backend_for_prompt(project_id) form_policy = await self._form_policy_for_prompt(project_id) system_prompt = build_system_prompt( project_type=project_type, instruction_content=instruction_content, selected_route=None, viewport=None, platform_backend=platform_backend, form_policy=form_policy, ) blob = json.dumps(values, indent=2, ensure_ascii=False) original = f"[supply_user_inputs] Apply {len(values)} configuration value(s)" wrapped = build_wrapped_prompt( "## Apply owner-supplied configuration\n\n" f"Submitted values (manifest keys):\n\n```json\n{blob}\n```\n\n" "Wire these into the correct files (see `p2hs/required-user-inputs.json` if it still exists). " "Commit all changes. Clear the `items` array in `p2hs/required-user-inputs.json` or delete " "that file when everything is applied." ) plan = { "system_prompt": system_prompt, "user_prompt": wrapped, "original_user_prompt": original, "context": { "project_type": project_type, "run_mode": "agent", "trigger": "supply_user_inputs", "source_run_id": str(run_id), }, } updated = await self._codegen_repo.requeue_with_plan(run_id, plan) if not updated: raise NotFoundError("Run not found") return updated async def _fan_out_inputs_to_env_vars( *, env_var_repo: "ProjectEnvVarRepository", project_id: UUID, values: dict[str, str], manifest_items: list[dict], overwrite: bool = True, ) -> None: """Write submitted builder-input values into project_env_vars (source of truth). manifest_items come from ``required-user-inputs.json``; each item may carry ``"type": "secret" | "public"`` — absent defaults to "public" since that file is historically for non-secret config (Calendly URLs, public IDs, etc.). """ from src.core.security.encryption import encrypt_value type_map = { str(it.get("id", "")): it.get("type", "public") for it in manifest_items if isinstance(it, dict) } for key, value in values.items(): if not key or not str(value).strip(): continue visibility = type_map.get(key, "public") if not overwrite: existing = await env_var_repo.get_by_key(project_id, key, "all") if existing is not None: continue await env_var_repo.upsert_by_key( project_id=project_id, key=key, encrypted_value=encrypt_value(value), environment="all", visibility=visibility, )