"""Queue and inspect required resource provisioning; no Cloud work in request handlers.""" import asyncio import hashlib import json from time import monotonic from uuid import UUID import httpx from arq.jobs import Job from pydantic import ValidationError as SchemaValidationError from src.core.exceptions import NotFoundError, P2HSError, ValidationError from src.domain.app_blocks.errors import UnsatisfiableAppBlockPlanError from src.domain.codegen.app_block_plan import assert_discovery_installable from src.domain.codegen.provision_manifest import validate_manifest class PrimitiveGapError(Exception): """Raised by resolve_codegen_capabilities (codegen.py) when a required pack install is blocked or a composed requirement returns primitive_gap -- NOT a real error, but arq's result_info() only distinguishes job success/failure by whether the job raised, never by inspecting its return value (confirmed: request_provisioning below only ever checked result.success before this). Raising this specific, recognized type is what lets request_provisioning tell "a genuine gap" apart from an actual provisioning failure, both of which are arq job failures. ``correction`` is set when every blocking problem is something the coding agent itself can fix (an invalid capability manifest): structured, bounded field errors and renames, so a retry can send the agent back with exactly what to change (see work_branch.py). It rides on the exception because arq pickles the exception's state, and stays None for a real gap.""" def __init__(self, message: str = "", correction: dict | None = None): super().__init__(message) self.correction = correction _MESSAGE_LIMIT = 500 def classify_provisioning_failure(error: object) -> dict: """Say *why* provisioning failed, for the runner, the run's error_message and the agent -- never just "failed". `retryable` means "the same request can succeed without anyone changing it" (a transient outage, or a platform bug fixed by a deploy); False means the request itself has to change. Only our own validation messages are relayed. For anything unexpected the message carries the exception *type* only (the detail stays in Core's logs), because an arbitrary exception string can hold internals that must not reach a customer or an external agent. """ if isinstance(error, (SchemaValidationError, ValueError, P2HSError)): # ValueError is how entity/manifest validation reports deterministic problems # (assert_compatible_entity: incompatible access policy, "needs an explicit # migration"), and how the job rejects an inactive run. if isinstance(error, SchemaValidationError): detail = "; ".join( ".".join(str(part) for part in item["loc"]) + ": " + item["msg"] for item in error.errors()[:5] ) else: detail = str(error) return { "failure_class": "invalid_requirement", "retryable": False, "message": ( "The requested platform capability is invalid and needs to change: " + (detail or type(error).__name__) )[:_MESSAGE_LIMIT], } if isinstance(error, httpx.HTTPStatusError) and not ( error.response.status_code >= 500 or error.response.status_code == 429 ): return { "failure_class": "cloud_rejected", "retryable": False, "message": ( f"P2HS Cloud rejected a required resource (HTTP {error.response.status_code}). " "The project's Cloud state may be out of sync with its installed capabilities." ), } if isinstance(error, (httpx.HTTPError, ConnectionError, TimeoutError, OSError)): return { "failure_class": "cloud_unavailable", "retryable": True, "message": "P2HS Cloud was unavailable while provisioning. Retrying should work.", } kind = type(error).__name__ if isinstance(error, BaseException) else "unknown error" return { "failure_class": "internal_error", "retryable": True, "message": ( f"P2HS hit an internal error while provisioning capabilities ({kind}). " "This is not a problem with your request; the details are in Core's logs." ), } async def request_provisioning( repository, run_id: UUID, manifest: dict, pack_installs: list | None = None, pack_compositions: list | None = None, ) -> dict: run = await repository.get_by_id(run_id) if not run: raise NotFoundError("Run not found") if run.status in {"failed", "canceled", "completed", "completed_with_limitations", "completed_with_l3", "blocked"}: raise ValidationError("Run is not active") try: assert_discovery_installable(await repository.get_project_discovery(run.project_id)) except UnsatisfiableAppBlockPlanError as exc: return {"status": "blocked", "message": str(exc)} try: declaration = validate_manifest(manifest) except SchemaValidationError as exc: # Deterministic. Raised as an HTTP error this looked like "not resolved yet" to # the runner, which retried for five minutes and then blamed connectivity. return {"status": "failed", **classify_provisioning_failure(exc)} pack_installs = pack_installs or [] pack_compositions = pack_compositions or [] if pack_compositions or pack_installs: # Security-sensitive grants the agent declared need a workspace admin's decision. Decided # here, on every call, and not inside the job below: the job's result is cached by manifest, # and a cached "waiting for approval" would keep failing after the owner approved. from src.domain.codegen import capability_approval held = await capability_approval.verdict(repository, run, pack_installs, pack_compositions) if held: return held fingerprint = hashlib.sha256( json.dumps( {"provision": declaration, "pack_installs": pack_installs, "pack_compositions": pack_compositions}, sort_keys=True, ).encode() ).hexdigest() job_id = f"codegen-provision:{run_id}:{fingerprint}" from src.workers.enqueue import _get_redis_pool redis = await _get_redis_pool() try: job = Job(job_id, redis) result = await job.result_info() if result: if result.success: # Optional gaps don't block ("blocked" below is required-gap only), # but must still reach the run's final status/message -- carry the # job's own return value through rather than collapsing it to a # bare "ready". job_result = result.result or {} return { "status": "ready", "pack_install_outcomes": job_result.get("pack_install_outcomes") or [], "composition_outcomes": job_result.get("composition_outcomes") or [], } if isinstance(result.result, (PrimitiveGapError, UnsatisfiableAppBlockPlanError)): blocked = {"status": "blocked", "message": str(result.result)} correction = getattr(result.result, "correction", None) if correction: # The agent's own manifest was invalid: say so, with the errors, so a retry # can fix exactly that instead of treating it as a missing capability. blocked.update(failure_class="invalid_requirement", retryable=True, correction=correction) return blocked return {"status": "failed", **classify_provisioning_failure(result.result)} await redis.enqueue_job( "resolve_codegen_capabilities", str(run_id), declaration, pack_installs, pack_compositions, _job_id=job_id ) return {"status": "pending"} finally: await redis.aclose() async def wait_for_provisioning( db_factory, run_id: UUID, manifest: dict, pack_installs: list | None = None, pack_compositions: list | None = None, ) -> None: """Bundle uploads also wait on a worker before pushing their generated revision.""" from src.domain.codegen.repository import CodegenRunRepository deadline = monotonic() + 300 while monotonic() < deadline: async with db_factory() as db: result = await request_provisioning( CodegenRunRepository(db), run_id, manifest, pack_installs, pack_compositions ) if result["status"] == "ready": return if result["status"] == "blocked": raise ValidationError(result.get("message") or "A requested capability could not be applied.") if result["status"] == "failed": raise ValidationError( result.get("message") or "Required platform resource provisioning failed" ) await asyncio.sleep(2) raise ValidationError("Timed out waiting for required platform resources")