"""Discovery inference and starter application run off the request path.""" import time from uuid import UUID import structlog from src.config import get_settings from src.domain.projects.discovery_repository import DiscoveryRepository from src.domain.projects.discovery_service import available_templates, require_discovery_access logger = structlog.get_logger() _SAFE_FAILURES = { "missing_provider": ( "Discovery is not configured on this environment. " "Please retry later; no build has started." ), "budget_exhausted": ( "Discovery is busy right now. Please retry shortly; no build has started." ), "discovery_response_unavailable": ( "Discovery couldn't finish. Please retry; no build has started." ), "invalid_starter_selection": ( "Discovery couldn't finish. Please retry; no build has started." ), "invalid_template_selection": ( "Discovery couldn't finish. Please retry; no build has started." ), "invalid_capability_selection": ( "Discovery couldn't finish. Please retry; no build has started." ), "provider_timeout": ( "Discovery took too long. Please retry; no build has started." ), "provider_rate_limited": ( "AI provider is rate-limited. Please retry shortly; no build has started." ), } def _classify_discovery_failure(exc: BaseException) -> tuple[str, str]: """Map exceptions to a safe user message + machine error_code. Never echo exc text.""" from src.core.provider_capacity import ProviderCapacityUnavailableError if isinstance(exc, ProviderCapacityUnavailableError): return ( "provider_capacity_unavailable", "AI provider capacity is busy. Please retry shortly; no build has started.", ) if isinstance(exc, ValueError) and str(exc) in _SAFE_FAILURES: return str(exc), _SAFE_FAILURES[str(exc)] name = type(exc).__name__ module = type(exc).__module__ or "" if name in {"TimeoutError", "APITimeoutError", "ReadTimeout", "ConnectTimeout"} or ( "timeout" in name.lower() ): return "provider_timeout", _SAFE_FAILURES["provider_timeout"] if name in {"RateLimitError"} or "rate" in name.lower() and "limit" in name.lower(): return "provider_rate_limited", _SAFE_FAILURES["provider_rate_limited"] if "openai" in module.lower() or name.startswith("API"): return "provider_error", "Discovery couldn't finish. Please retry; no build has started." return "discovery_failed", "Discovery couldn't finish. Please retry; no build has started." async def run_project_discovery(ctx, project_id: str, version: int): from src.core.database import async_session_factory from src.domain.projects.discovery_catalog import ( CLOUD_CAPABILITIES, build_discovery_payload, model_facing_discovery_payload, ) from src.integrations.project_intake_client import discover_project factory = ctx.get("db_factory") or async_session_factory pid = UUID(project_id) settings = get_settings() async with factory() as db: project = await DiscoveryRepository(db).get(pid) state = dict(project.discovery_json or {}) if project else {} if state.get("status") != "processing" or state.get("version") != version: return {"status": "skipped"} try: await require_discovery_access(db, project, UUID(state["actor_id"])) templates = await available_templates(db, project.workspace_id) payload = build_discovery_payload( project.type, project.source_kind, state, templates, environment_status="ready" if project.status == "ready" else "preparing", ) supported = list(payload["boilerplates"]) locked = payload["locked_type"] setup_ready = project.status == "ready" await db.rollback() from src.domain.projects.discovery_fidelity import local_choose_for_me_turn result = local_choose_for_me_turn(state) if result is None: if not settings.openai_api_key: raise ValueError("missing_provider") await reserve_discovery_call(settings) # B1 harness measured against the slim model-facing payload; keep worker # aliases only for local starter/template checks below. result = await discover_project( model_facing_discovery_payload(payload), settings ) from src.domain.projects.discovery_readiness import ( discovery_assistant_message, finalize_discovery_result, ) result = finalize_discovery_result(result, payload) if result.project_type not in supported or (locked and result.project_type != locked): raise ValueError("invalid_starter_selection") allowed_refs = { t["ref"] for t in payload["templates"] if t["framework"] == result.project_type } if result.template_ref and result.template_ref not in allowed_refs: raise ValueError("invalid_template_selection") if set(result.requested_capabilities) - CLOUD_CAPABILITIES.keys(): raise ValueError("invalid_capability_selection") compiled = result.model_dump() pin = {} if result.decision == "ready": from src.domain.composition_recipes.service import CompositionRecipeService pin = await CompositionRecipeService(db).pin_ready_composition( state, spec=compiled.get("spec") or {}, project_type=result.project_type, ) updated = { **state, "status": result.decision, "result": compiled, "core_blockers": list(result.readiness.blocking_decision_ids), "messages": [ *state["messages"], { "role": "assistant", "content": discovery_assistant_message(result), }, ], "error": None, "error_code": None, **pin, } if result.decision == "ready" and setup_ready: from src.domain.projects.discovery_compilation import ( attach_build_plan, load_discovery_contract, ) from src.integrations.project_intake_client import compile_discovery contract = await load_discovery_contract(db, pid) await db.rollback() await reserve_discovery_call(settings) plan = await compile_discovery( {"brief": compiled, "cloud_contract": contract}, settings ) compiled = attach_build_plan(result, plan, contract) updated["result"] = compiled except Exception as exc: # Provider errors can contain the brief: persist a safe message only. error_code, message = _classify_discovery_failure(exc) logger.warning( "project_discovery_failed", project_id=project_id, version=version, error_code=error_code, exception_type=type(exc).__name__, ) updated = { **state, "status": "failed", "error": message, "error_code": error_code, } await db.rollback() project = await DiscoveryRepository(db).get(pid, lock=True) if ( project and (project.discovery_json or {}).get("version") == version and project.discovery_json.get("status") == "processing" ): project.discovery_json = updated await db.commit() return {"status": updated["status"]} async def apply_project_discovery(ctx, project_id: str, version: int): from src.workers.discovery_apply import apply_discovered_starter return await apply_discovered_starter(ctx, UUID(project_id), version) async def reserve_discovery_call(settings): """Per-minute intake/discovery budget. Redis infra failures fail open.""" from redis.asyncio import Redis try: async with Redis.from_url( settings.redis_url, socket_timeout=2, socket_connect_timeout=2 ) as redis: used = await redis.eval( "local n=redis.call('INCR',KEYS[1]); if n==1 then " "redis.call('EXPIRE',KEYS[1],120) end; return n", 1, f"project-intake:budget:{int(time.time()) // 60}", ) if used > settings.project_intake_calls_per_minute: raise ValueError("budget_exhausted") except ValueError: raise except Exception as exc: # Do not block discovery when Redis admission is unreachable (timeouts, # auth, protected-mode). Policy denials still raise budget_exhausted. logger.warning( "discovery_budget_redis_unavailable", exception_type=type(exc).__name__, )