"""Builder workspace - hydrate service composes project-rooted read model.""" from __future__ import annotations import re from typing import TYPE_CHECKING, Optional from uuid import UUID import structlog from src.config import get_settings from src.domain.hosting.hostnames import canonical_live_url, canonical_preview_url from src.domain.builder.schemas import ( ChangesSummary, ChangedFileSummary, PublishSummary, ProjectSummary, RouteEntry, RoutesSummary, RunSummary, SessionPromptSummary, TimelineEventSummary, TimelineSummary, PreviewSummary, PromptMessageSummary, WorkspaceResponse, ) logger = structlog.get_logger() # project_deploys.status (p2hs_deploy_status): pending | deploying | live | failed | rolled_back _IN_PROGRESS_DEPLOY_STATUSES = frozenset({"pending", "deploying"}) _TERMINAL_PREVIEW_PHASES = frozenset({"ready", "failed", "empty"}) _TERMINAL_RUN_STATUSES = frozenset( { "completed", "completed_with_limitations", "completed_with_l3", "blocked", "failed", "skipped", "canceled", "plan_ready", "awaiting_user_input", "user_input_required", "clarification_required", } ) def _sanitize_user_message(msg: Optional[str]) -> Optional[str]: """Strip server-internal details before exposing a message to end users.""" if not msg: return msg low = msg.lower() if "timed out" in low or "timeout" in low: return "Code generation took too long. Please try a more focused request." if "not found" in low and ("cli" in low or "binary" in low or "agent" in low): return "Code generation service is temporarily unavailable." if "build validation failed" in low: return "The generated code could not be built. Please try rephrasing your request." # Strip absolute server paths (e.g. /home2/p2hs/...) from the message cleaned = re.sub(r"/[/\w.\-]+(:\s*|\s+|$)", " ", msg).strip() # If the entire message was a path or now empty, give a generic fallback if not cleaned or len(cleaned) < 10: return "An unexpected error occurred. Please try again." return cleaned def _has_bootstrap_deploy_pending(deploys: list) -> bool: """True while the setup-pipeline production deploy is still in flight.""" return any( (getattr(d, "deployment_metadata", None) or {}).get("bootstrap_initial_hosting") and (getattr(d, "status", "") or "") in _IN_PROGRESS_DEPLOY_STATUSES for d in deploys ) def _live_bootstrap_deploy(deploys: list): return next( ( d for d in deploys if (getattr(d, "deployment_metadata", None) or {}).get("bootstrap_initial_hosting") and (getattr(d, "status", "") or "") == "live" ), None, ) def _preview_candidate_url(candidate: str | None, project_ref: str, apex: str) -> str | None: """Return a usable preview URL, correcting a stale W4P live-host value. Older W4P preview rows can contain the deterministic production hostname (``proj-*``) even though the preview route is ``prev-*``. Do not let that persisted value override the canonical preview identity. Other provider URLs (notably legacy Pages URLs) remain untouched. """ if not candidate: return None value = candidate.rstrip("/") if project_ref and value == canonical_live_url(project_ref, apex).rstrip("/"): return canonical_preview_url(project_ref, apex).rstrip("/") return value def _apply_bootstrap_preview( preview: PreviewSummary, deploys: list, *, has_bootstrap_deploy_pending: bool, latest_run_status: str | None, project_ref: str | None, ) -> PreviewSummary: """Map initial production hosting onto the builder preview pane. While the setup deploy is still running, keep the "building" overlay. Once it is live and no preview environment exists yet, show the production URL so the iframe is not stuck on an empty CTA. """ if has_bootstrap_deploy_pending and preview.preview_phase == "empty": return preview.model_copy(update={"preview_phase": "building"}) live = _live_bootstrap_deploy(deploys) if live is not None and not has_bootstrap_deploy_pending: # W4P activation can succeed before the preview row's asynchronous # finalizer runs (or for rows created by an older version of the # worker). A live bootstrap deployment is authoritative for hosting # readiness, so heal that stale ``provisioning`` row in the read model. # Do not replace a preview for a different revision: that can be a # newer codegen deployment still in progress. live_revision_id = getattr(live, "revision_id", None) if ( preview.preview_phase in {"empty", "building"} and preview.preview_source_revision_id is not None and live_revision_id is not None and preview.preview_source_revision_id != live_revision_id ): return preview ref = (project_ref or "").strip() live_url = None if ref: apex = get_settings().hosting_public_apex or "sites.p2hs.com" live_url = canonical_live_url(ref, apex).rstrip("/") elif getattr(live, "deploy_url", None): live_url = str(live.deploy_url).rstrip("/") if live_url: return preview.model_copy( update={ "preview_phase": "ready", "preview_url": live_url, "preview_source_revision_id": ( live_revision_id or preview.preview_source_revision_id ), } ) if ( preview.preview_phase != "empty" or latest_run_status is not None or has_bootstrap_deploy_pending ): return preview if live is None: failed = next( ( d for d in deploys if (getattr(d, "deployment_metadata", None) or {}).get("bootstrap_initial_hosting") and (getattr(d, "status", "") or "") in {"failed", "rolled_back"} ), None, ) if failed is not None: return preview.model_copy(update={"preview_phase": "failed"}) return preview ref = (project_ref or "").strip() live_url = None if ref: apex = get_settings().hosting_public_apex or "sites.p2hs.com" live_url = canonical_live_url(ref, apex).rstrip("/") elif getattr(live, "deploy_url", None): live_url = str(live.deploy_url).rstrip("/") if not live_url: return preview return preview.model_copy( update={ "preview_phase": "ready", "preview_url": live_url, "preview_source_revision_id": getattr(live, "revision_id", None), } ) def _resolve_polling_complete( *, project_status: str, latest_run_status: str | None, run_is_terminal: bool, preview_phase: str, has_bootstrap_deploy_pending: bool, ) -> bool: """Stop HTTP polling once setup, bootstrap hosting, and any active run have settled. A project with no codegen runs is complete as soon as the initial hosting deploy is no longer pending. ``live`` is the success status on ``project_deploys`` — treating it as in-flight leaves the builder overlay spinning. """ if project_status == "creating": return False if latest_run_status is None: return not has_bootstrap_deploy_pending return run_is_terminal and preview_phase in _TERMINAL_PREVIEW_PHASES if TYPE_CHECKING: from src.domain.builder.repository import ChangedFileRepository, TimelineEventRepository from src.domain.codegen.repository import CodegenRunRepository from src.domain.codegen.run_chat_repository import RunChatMessageRepository from src.domain.deployments.repository import DeployRepository, DomainRepository from src.domain.instructions.repository import InstructionRepository, PromptRepository, PromptSessionRepository from src.domain.previews.repository import PreviewRepository, RoutesManifestRepository from src.domain.projects.repository import ProjectRepository from src.domain.revisions.repository import RevisionRepository class WorkspaceHydrateService: """Composes workspace read model from multiple repositories.""" def __init__( self, project_repo: "ProjectRepository", revision_repo: "RevisionRepository", codegen_repo: "CodegenRunRepository", prompt_session_repo: "PromptSessionRepository", prompt_repo: "PromptRepository", run_chat_repo: "RunChatMessageRepository", preview_repo: "PreviewRepository", manifest_repo: "RoutesManifestRepository", deploy_repo: "DeployRepository", domain_repo: "DomainRepository", timeline_repo: "TimelineEventRepository", changed_file_repo: "ChangedFileRepository", ): self._project_repo = project_repo self._revision_repo = revision_repo self._codegen_repo = codegen_repo self._prompt_session_repo = prompt_session_repo self._prompt_repo = prompt_repo self._run_chat_repo = run_chat_repo self._preview_repo = preview_repo self._manifest_repo = manifest_repo self._deploy_repo = deploy_repo self._domain_repo = domain_repo self._timeline_repo = timeline_repo self._changed_file_repo = changed_file_repo async def hydrate( self, project_id: UUID, *, selected_run_id: Optional[UUID] = None, selected_session_id: Optional[UUID] = None, selected_route: Optional[str] = None, limit_timeline: int = 30, limit_changes: int = 20, ) -> WorkspaceResponse: project = await self._project_repo.get_with_repo(project_id) if not project: from src.core.exceptions import NotFoundError raise NotFoundError("Project not found") active_session_id = selected_session_id or project.active_session_id active_preview_id = project.active_preview_id active_revision_id = project.active_revision_id latest_revision = await self._revision_repo.get_latest(project_id) if not active_revision_id and latest_revision: active_revision_id = latest_revision.id latest_run = None runs = await self._codegen_repo.list_for_project(project_id) if runs: latest_run = runs[0] selected_run_id = selected_run_id or (latest_run.id if latest_run else None) session_prompt = await self._build_session_prompt( project_id, active_session_id ) run_summary = await self._build_run_summary( project_id, selected_run_id, latest_run ) timeline = await self._build_timeline_summary( project_id, selected_run_id, limit_timeline ) changes = await self._build_changes_summary( project_id, selected_run_id, active_revision_id, limit_changes ) latest_run_status = latest_run.status if latest_run else None preview = await self._build_preview_summary( project_id, active_preview_id, latest_run_status=latest_run_status, project_ref=getattr(project, "project_ref", None), ) routes = await self._build_routes_summary( project_id, active_revision_id, selected_route ) latest_deploy = None deploys = await self._deploy_repo.list_for_project(project_id) if deploys: latest_deploy = deploys[0] capabilities = await self._build_capabilities_summary(project_id) publish_summary = await self._build_publish_summary( project_id, active_revision_id, capabilities ) # polling_complete: # - status="creating" → keep polling until setup flips to ready # - ready, no runs → stop once the initial hosting deploy is no longer pending # - run submitted → stop when the run is terminal AND preview has settled project_status = getattr(project, "status", "ready") # Auto-heal: a project stuck in "error" that already has a usable revision # is effectively ready — the setup partially failed but the project is usable. # Persist the corrected status so future reads are consistent. if project_status == "error" and active_revision_id: project_status = "ready" await self._project_repo.update(project_id, status="ready") has_bootstrap_deploy_pending = _has_bootstrap_deploy_pending(deploys) preview = _apply_bootstrap_preview( preview, deploys, has_bootstrap_deploy_pending=has_bootstrap_deploy_pending, latest_run_status=run_summary.latest_run_status, project_ref=getattr(project, "project_ref", None), ) polling_complete = _resolve_polling_complete( project_status=project_status, latest_run_status=run_summary.latest_run_status, run_is_terminal=run_summary.is_terminal, preview_phase=preview.preview_phase, has_bootstrap_deploy_pending=has_bootstrap_deploy_pending, ) _repo = getattr(project, "repo", None) project_ref = getattr(project, "project_ref", None) live_url = None preview_url = None if project_ref: apex = get_settings().hosting_public_apex or "sites.p2hs.com" live_url = canonical_live_url(project_ref, apex).rstrip("/") preview_url = canonical_preview_url(project_ref, apex).rstrip("/") return WorkspaceResponse( project=ProjectSummary( id=project.id, workspace_id=project.workspace_id, name=project.name, slug=project.slug, type=project.type, status=project_status, created_at=project.created_at, project_ref=project_ref, live_url=live_url, preview_url=preview_url, active_revision_id=active_revision_id or (latest_revision.id if latest_revision else None), active_preview_id=active_preview_id, latest_deploy_id=latest_deploy.id if latest_deploy else None, publish_status_summary=latest_deploy.status if latest_deploy else None, setup_stage=getattr(_repo, "setup_stage", None), discovery_status=(getattr(project, "discovery_json", None) or {}).get("status"), ), session_prompt=session_prompt, run=run_summary, timeline=timeline, changes=changes, preview=preview, routes=routes, publish=publish_summary, capabilities=capabilities, selected_route=selected_route, polling_complete=polling_complete, ) async def _build_capabilities_summary(self, project_id: UUID) -> dict[str, "CapabilitySummary"]: """CapabilityProvisioner.status() for every registered capability. Read-only and DB-only per the base contract, so this is cheap on every workspace poll. A provisioner raising must not break the whole hydrate. Every entry, including "auth", now comes from the lifecycle.py/adapters.py lifecycle -- the same provisioners GET /api/projects/{id}/capabilities/status uses. This used to read "auth" from the older, separate base.py/ auth_provisioner.py lifecycle (a 5-state status with no issue codes and no shadow/enabled distinction); that duplication is what docs/CAPABILITY_RUNTIME_PROJECTION.md's "Deferred" section flagged as unreconciled. Reconciled by switching to AuthV2Provisioner, which already reports everything the old one did (config/key/cloud-sync health) plus the real projection_state (shadow/enabled), so the dashboard can show it instead of collapsing both into "ready".""" from src.domain.builder.schemas import CapabilitySummary from src.domain.capabilities.adapters import ( AuthV2Provisioner, FeaturePackProvisioner, FormsProvisioner, ) from src.domain.capabilities.native_adapters import ( FilesProvisioner, NotificationsProvisioner, RecordsProvisioner, SecretsProvisioner, WebhooksProvisioner, ) db = self._project_repo._db settings = get_settings() summaries: dict[str, CapabilitySummary] = {} for key, provisioner in ( ( "auth", AuthV2Provisioner( db, cloud_base_url=settings.cloud_control_api_url, cloud_token=settings.cloud_internal_token, ), ), ("forms", FormsProvisioner(db)), ("feature_packs", FeaturePackProvisioner(db)), ("files", FilesProvisioner(db)), ("records", RecordsProvisioner(db)), ("notifications", NotificationsProvisioner(db)), ("webhooks", WebhooksProvisioner(db)), ("secrets", SecretsProvisioner(db)), ): try: status = await provisioner.status(project_id) except Exception: logger.warning( "capability_status_failed", project_id=str(project_id), capability=key, exc_info=True, ) continue summaries[key] = CapabilitySummary( status=status.state.value, reason=status.blocked_reason, ) return summaries async def _build_session_prompt( self, project_id: UUID, active_session_id: Optional[UUID], ) -> SessionPromptSummary: """Align with GET …/prompt/messages: merge ``run_chat_messages`` + legacy prompts.""" current_slice: list[PromptMessageSummary] = [] linked_run_ids: list[UUID] = [] last_assistant_summary: Optional[str] = None if active_session_id: prompts = await self._prompt_repo.list_for_session(active_session_id) sender_labels = await self._prompt_repo.get_sender_labels(prompts) runs = await self._codegen_repo.list_for_project(project_id) prompt_to_run: dict[UUID, UUID] = { r.prompt_id: r.id for r in runs if r.prompt_id } unified = await self._run_chat_repo.list_for_session(active_session_id) if unified: covered = {m.prompt_id for m in unified if m.prompt_id} for m in unified: if m.run_id: linked_run_ids.append(m.run_id) current_slice.append( PromptMessageSummary( id=m.id, sender_label=( sender_labels.get(m.prompt_id) if m.role == "user" else None ), content=m.content, context=None, response=None, created_at=m.created_at, run_id=m.run_id, role=m.role, message_type=m.message_type, payload=m.payload, prompt_id=m.prompt_id, ) ) for p in prompts: if p.id in covered: continue run_id = prompt_to_run.get(p.id) if run_id: linked_run_ids.append(run_id) current_slice.append( PromptMessageSummary( id=p.id, sender_label=sender_labels.get(p.id), content=p.content, context=p.context, response=p.response, created_at=p.created_at, run_id=run_id, role="user", message_type="message", ) ) current_slice.sort(key=lambda x: x.created_at) else: for p in prompts: run_id = prompt_to_run.get(p.id) if run_id: linked_run_ids.append(run_id) current_slice.append( PromptMessageSummary( id=p.id, sender_label=sender_labels.get(p.id), content=p.content, context=p.context, response=p.response, created_at=p.created_at, run_id=run_id, ) ) if current_slice: tail = current_slice[-1] if (tail.response or "").strip(): s = tail.response or "" last_assistant_summary = s[:200] + "..." if len(s) > 200 else s elif tail.role == "assistant" and (tail.content or "").strip(): s = tail.content or "" last_assistant_summary = s[:200] + "..." if len(s) > 200 else s return SessionPromptSummary( active_session_id=active_session_id, current_prompt_slice=current_slice, last_assistant_response_summary=last_assistant_summary, linked_run_ids=list(dict.fromkeys(linked_run_ids)), ) async def _build_run_summary( self, project_id: UUID, selected_run_id: Optional[UUID], latest_run: Optional[any], ) -> RunSummary: run = None if selected_run_id: run = await self._codegen_repo.get_by_id(selected_run_id) status = run.status if run else (latest_run.status if latest_run else None) run_type = "code_applied" if (run and run.revision_id) else "plan_only" if run else None duration = None if run and run.started_at and run.completed_at: duration = (run.completed_at - run.started_at).total_seconds() _terminal = _TERMINAL_RUN_STATUSES _continuation = frozenset( { "plan_ready", "clarification_required", "user_input_required", "awaiting_user_input", } ) is_terminal = status in _terminal if status else False raw_error = run.error_message if run else None continuation_required = bool(status and status in _continuation) pu = run.pending_user_inputs if run else None intr = run.interruption_payload if run else None return RunSummary( selected_run_id=selected_run_id, latest_run_id=latest_run.id if latest_run else None, latest_run_status=status, is_terminal=is_terminal, run_type=run_type, duration_seconds=duration, error_message=_sanitize_user_message(raw_error), pending_user_inputs=pu, interruption_payload=intr, continuation_required=continuation_required, ) async def _build_timeline_summary( self, project_id: UUID, run_id: Optional[UUID], limit: int, ) -> TimelineSummary: events = await self._timeline_repo.list_for_project(project_id, run_id=run_id, limit=limit) latest = [ TimelineEventSummary( id=e.id, event_type=e.event_type, phase=e.phase, title=e.title, detail_text=e.detail_text, sequence_no=e.sequence_no, created_at=e.created_at, ) for e in events ] current_phase = latest[-1].phase if latest else None current_activity = latest[-1].title if latest else None return TimelineSummary( latest_events=latest, current_phase=current_phase, current_activity_label=current_activity, ) # Platform-internal paths that are never meaningful to show the user. _PLATFORM_PATH_PREFIXES = ( ".github/", ".p2hs", ".cursor/", "node_modules/", ".nuxt/", ".output/", "dist/", ) @classmethod def _is_platform_file(cls, file_path: str) -> bool: p = file_path.lstrip("/") return any(p.startswith(prefix) for prefix in cls._PLATFORM_PATH_PREFIXES) async def _build_changes_summary( self, project_id: UUID, run_id: Optional[UUID], revision_id: Optional[UUID], limit: int, ) -> ChangesSummary: files: list = [] if run_id: files = await self._changed_file_repo.list_for_run(project_id, run_id) elif revision_id: files = await self._changed_file_repo.list_for_revision(project_id, revision_id) # Strip platform-internal files; only count/show meaningful user files. user_files = [f for f in files if not self._is_platform_file(f.file_path)] top = [ ChangedFileSummary( id=f.id, file_path=f.file_path, change_type=f.change_type, summary=f.summary, lines_added=f.lines_added, lines_removed=f.lines_removed, ) for f in user_files[:limit] ] lines_added = sum(f.lines_added or 0 for f in user_files) lines_removed = sum(f.lines_removed or 0 for f in user_files) return ChangesSummary( changed_files_count=len(user_files), top_changed_files=top, lines_added=lines_added, lines_removed=lines_removed, ) async def _build_preview_summary( self, project_id: UUID, active_preview_id: Optional[UUID], *, latest_run_status: Optional[str] = None, project_ref: Optional[str] = None, ) -> PreviewSummary: _IN_PROGRESS = {"queued", "pending", "running"} pe = None if active_preview_id: pe = await self._preview_repo.get_by_id(active_preview_id) # Fall back to latest preview for the project (covers provisioning state # before active_preview_id is set, and after the boilerplate gate) if not pe: previews = await self._preview_repo.list_for_project(project_id) pe = previews[0] if previews else None settings = get_settings() apex = settings.hosting_public_apex or "sites.p2hs.com" ref = (project_ref or "").strip() def _resolved_preview_url() -> Optional[str]: """Best URL for the iframe: stored CF/W4P URL, or deterministic prev-.apex.""" if not pe: return None if pe.base_url: return _preview_candidate_url(pe.base_url, ref, apex) if pe.provider_preview_url: return _preview_candidate_url(pe.provider_preview_url, ref, apex) if ref and pe.status in ("running", "stopped"): return canonical_preview_url(ref, apex) return None resolved = _resolved_preview_url() # Determine the phase that drives the frontend UI state machine. if latest_run_status in _IN_PROGRESS: # AI is actively generating — hide any existing iframe, show animation. phase = "generating" elif pe is None: # No preview row yet — no prompt has been submitted. phase = "empty" elif pe.status in ("failed", "error"): phase = "failed" elif resolved and pe.status in ("running", "stopped"): # "stopped" often means TTL cleanup (`cleanup_expired_previews`) while the # CF Pages deployment is still reachable at base_url — do not treat as failed. phase = "ready" elif pe.status == "stopped": phase = "failed" else: # provisioning or unknown intermediate state — CF Pages still building. phase = "building" # Expose URL when we can target a live or deterministically routed deployment. preview_url = resolved if (pe and phase == "ready") else None if not pe: return PreviewSummary( active_preview_id=None, preview_status=None, preview_url=None, preview_source_revision_id=None, last_refreshed_at=None, preview_phase=phase, ) return PreviewSummary( active_preview_id=pe.id, preview_status=pe.status, preview_url=preview_url, preview_source_revision_id=pe.revision_id, last_refreshed_at=pe.stopped_at or getattr(pe, "updated_at", None), preview_phase=phase, ) async def _build_routes_summary( self, project_id: UUID, revision_id: Optional[UUID], selected_route: Optional[str], ) -> RoutesSummary: manifest = await self._manifest_repo.get_latest(project_id, revision_id) if not manifest: # No revision yet — return empty routes so the frontend doesn't show # boilerplate navigation before the user has submitted a prompt. if revision_id is None: return RoutesSummary(route_manifest_id=None, routes=[], default_route=None) # Revision exists but no manifest yet — show a safe default return RoutesSummary( route_manifest_id=None, routes=[RouteEntry(path="/", label="Home", type="page", is_default=True)], default_route="/", ) m = manifest.manifest if hasattr(manifest, "manifest") else manifest routes_data = m.get("routes", []) if isinstance(m, dict) else [] default = None entries = [] for r in routes_data: if isinstance(r, dict): entries.append( RouteEntry( path=r.get("path", "/"), label=r.get("label", r.get("path", "/")), type=r.get("type", "page"), is_default=r.get("is_default", False), ) ) if r.get("is_default"): default = r.get("path", "/") if not entries: entries = [RouteEntry(path="/", label="Home", type="page", is_default=True)] default = "/" return RoutesSummary( route_manifest_id=manifest.id, routes=entries, default_route=default or (entries[0].path if entries else "/"), ) async def _build_publish_summary( self, project_id: UUID, active_revision_id: Optional[UUID], capabilities: Optional[dict] = None, ) -> PublishSummary: project = await self._project_repo.get_by_id(project_id) deploys = await self._deploy_repo.list_for_project(project_id) latest = deploys[0] if deploys else None domains = await self._domain_repo.list_for_project(project_id) primary_domain = next((d for d in domains if d.is_primary), domains[0] if domains else None) custom_domain_status = primary_domain.verification_status if primary_domain else None generated_domain = None if project: pref = (getattr(project, "project_ref", None) or "").strip() if pref: apex = (get_settings().hosting_public_apex or "sites.p2hs.com").strip() generated_domain = canonical_live_url(pref, apex).rstrip("/") elif project.slug: generated_domain = f"https://{project.slug}-production.p2hs.app" # Find the most recent deploy for the *active* revision — ignore stale # initial-setup deploys that are for older revisions. active_deploy = next( (d for d in deploys if d.revision_id == active_revision_id), None, ) if active_revision_id else None _TERMINAL_FAIL = {"failed", "rolled_back"} blockers: list[str] = [] if not active_deploy: blockers.append("no_deploy") elif active_deploy.status in _IN_PROGRESS_DEPLOY_STATUSES: blockers.append("deploy_in_progress") elif active_deploy.status in _TERMINAL_FAIL: blockers.append("deploy_failed") # "live" → no deploy blocker if primary_domain and primary_domain.verification_status != "verified": blockers.append("domain_not_verified") # Artifact deployment success must not be reported as full application # readiness while a required capability is blocked (e.g. the static site is # live but Auth has no cloud_auth_config) -- see CLAUDE.md's Core<->Cloud # contract and docs/CAPABILITY_RUNTIME_PROJECTION.md. Prefixed so existing # consumers checking specific blocker strings are unaffected and a caller # that cares can still separate infra blockers from capability ones. for key, summary in (capabilities or {}).items(): if summary.status == "blocked": blockers.append(f"capability_blocked:{key}") return PublishSummary( latest_deploy_id=latest.id if latest else None, deploy_status=latest.status if latest else None, live_domain=primary_domain.domain if primary_domain else None, generated_domain=generated_domain, custom_domain_status=custom_domain_status, published_revision_id=latest.revision_id if latest else None, publish_blockers=blockers, )