"""Internal Cloud API — authorize jobs and reconcile usage (P2HS Cloud → Core).""" from __future__ import annotations from datetime import UTC, datetime from typing import TYPE_CHECKING from uuid import UUID from sqlalchemy.ext.asyncio import AsyncSession from src.config import get_settings from src.core.canonical_command import ( PUBLIC_COMMANDS, REQUIRES_USER_PRINCIPAL, SENSITIVE_COMMANDS, STUB_REQUEST_TYPES, command_hash, ) from src.core.security.cloud_tokens import ( create_cloud_authorization_token, create_cloud_grant_jwt, decode_cloud_authorization_token, decode_cloud_grant_jwt, ) from src.domain.billing.normalization import ( normalized_billing_status_from_row, requires_payment_flag, ) from src.domain.billing.repository import SubscriptionRepository, UsageLedgerRepository from src.domain.cloud.email_delivery import ( EMAIL_JOB_TYPES, platform_default_email_delivery, resolve_email_delivery, ) from src.domain.cloud.sms_delivery import platform_default_sms_delivery from src.domain.connectors.payment_resolve import ( PAYMENT_JOB_TYPES, PAYMENT_PROVIDERS, canonical_payment_capability, resolve_payment_capability, snapshots_from_rows, ) from src.domain.cloud.schemas import ( AuthorizeRequest, AuthorizeResponse, AuthorizeResponseAllowed, AuthorizeResponseDenied, CRMInboundEventsRequest, GrantBundle, PolicySnapshot, Reservation, StorageUsageAckResponse, StorageUsageReportRequest, SystemGrantRequest, UsageAckResponse, UsageFinalizeRequest, UsageReleaseRequest, ) from src.domain.cloud_credits.service import UNAVAILABLE_MESSAGE, CloudCreditService from src.domain.connectors.repository import ConnectorRepository from src.domain.connectors.service import ConnectorService from src.domain.integration_events.schemas import CommandAuthorizeRequest from src.domain.livechat.repository import LivechatRepository from src.domain.project_storage.repository import ProjectStorageUsageRepository from src.domain.projects.models import Project from src.domain.projects.repository import ProjectRepository, ProjectSettingsRepository from src.domain.runtime_ai.catalog import ( estimate_livechat_millicredits, estimate_voice_millicredits, millicredits_for_usage, ) from src.domain.runtime_ai.wallet import RuntimeAiWallet from src.domain.voice.repository import VoiceRepository from src.domain.voice.service import VoiceService from src.domain.workspaces.repository import WorkspaceRepository from src.integrations.cloud_control_client import CloudControlClient from src.integrations.email_provider import apply_platform_email_branding from src.integrations.platform_ports import ( REQUEST_TYPES, ) from src.integrations.platform_ports import ( allowed_request_types as _allowed_request_types, ) from src.integrations.platform_ports import ( enabled_capabilities as _enabled_project_capabilities, ) if TYPE_CHECKING: from src.domain.email_domains.repository import EmailDomainRepository from src.domain.email_domains.service import EmailQuotaService DEFAULT_CLOUD_REQUEST_TYPES = REQUEST_TYPES _DEFAULT_EXECUTION_POLICY: dict = { "max_attempts": 8, "max_runtime_seconds": 120, "requires_webhook": False, "artifact_retention_days": 30, } # Default execution policy per request type (metadata-driven; later from plan DB) _REQUEST_TYPE_POLICY: dict[str, dict] = { "email.send": { "max_attempts": 8, "max_runtime_seconds": 120, "requires_webhook": False, "artifact_retention_days": 30, }, "form.forward": { "max_attempts": 8, "max_runtime_seconds": 60, "requires_webhook": False, "artifact_retention_days": 14, }, "webhook.dispatch": { "max_attempts": 8, "max_runtime_seconds": 60, "requires_webhook": False, "artifact_retention_days": 7, }, "integration.execute": { "max_attempts": 8, "max_runtime_seconds": 120, "requires_webhook": False, "artifact_retention_days": 30, }, # Platform-native record engine operations (query_dsl_v1) "record.query": { "max_attempts": 3, "max_runtime_seconds": 30, "requires_webhook": False, "artifact_retention_days": 1, }, "record.aggregate": { "max_attempts": 3, "max_runtime_seconds": 60, "requires_webhook": False, "artifact_retention_days": 1, }, "record.bulk.update": { "max_attempts": 5, "max_runtime_seconds": 120, "requires_webhook": False, "artifact_retention_days": 7, }, "record.bulk.delete": { "max_attempts": 5, "max_runtime_seconds": 120, "requires_webhook": False, "artifact_retention_days": 7, }, "data.import": { "max_attempts": 3, "max_runtime_seconds": 600, # large CSV files can take several minutes "requires_webhook": False, # progress via realtime jobs channel "artifact_retention_days": 7, # import report / error log retained }, "dropbox.import": { "max_attempts": 5, "max_runtime_seconds": 300, "requires_webhook": False, "artifact_retention_days": 30, }, "storage.extract": { "max_attempts": 50, "max_runtime_seconds": 900, "requires_webhook": False, "artifact_retention_days": 30, }, "voice.call.handle": { "max_attempts": 1, "max_runtime_seconds": 3600, "requires_webhook": True, "artifact_retention_days": 30, }, "voice.call.outbound": { "max_attempts": 3, "max_runtime_seconds": 3600, "requires_webhook": True, "artifact_retention_days": 30, }, "voice.tool.execute": { "max_attempts": 5, "max_runtime_seconds": 60, "requires_webhook": False, "artifact_retention_days": 7, }, "livechat.message": { "max_attempts": 1, "max_runtime_seconds": 120, "requires_webhook": False, "artifact_retention_days": 30, }, "booking.reserve": { "max_attempts": 3, "max_runtime_seconds": 60, "requires_webhook": False, "artifact_retention_days": 30, }, "booking.cancel": { "max_attempts": 3, "max_runtime_seconds": 30, "requires_webhook": False, "artifact_retention_days": 14, }, "inventory.reserve": { "max_attempts": 3, "max_runtime_seconds": 30, "requires_webhook": False, "artifact_retention_days": 30, }, "inventory.release": { "max_attempts": 3, "max_runtime_seconds": 30, "requires_webhook": False, "artifact_retention_days": 14, }, "workforce.shift_assign": { "max_attempts": 3, "max_runtime_seconds": 30, "requires_webhook": False, "artifact_retention_days": 30, }, "workforce.shift_release": { "max_attempts": 3, "max_runtime_seconds": 30, "requires_webhook": False, "artifact_retention_days": 14, }, "reminder.send": { "max_attempts": 8, "max_runtime_seconds": 60, "requires_webhook": False, "artifact_retention_days": 14, }, "reminder.schedule": { "max_attempts": 3, "max_runtime_seconds": 30, "requires_webhook": False, "artifact_retention_days": 30, }, "payment.checkout.create": { "max_attempts": 3, "max_runtime_seconds": 60, "requires_webhook": False, "artifact_retention_days": 30, }, "payment.capture": { "max_attempts": 3, "max_runtime_seconds": 60, "requires_webhook": False, "artifact_retention_days": 30, }, "payment.refund": { "max_attempts": 3, "max_runtime_seconds": 60, "requires_webhook": False, "artifact_retention_days": 30, }, "payment.catalog.upsert": { "max_attempts": 3, "max_runtime_seconds": 60, "requires_webhook": False, "artifact_retention_days": 30, }, } _LONG_RUNNING_CONNECTOR_ACTIONS = frozenset({ "cloudflare_r2.extract_zip", "cloudflare_r2.import_from_url", "cloudflare_r2.bulk_export_bucket", }) def _expected_client_id(project: Project) -> str: return f"cli_{project.project_ref or str(project.id)}" async def _payment_resolution_for_authorize( *, connector_repo: ConnectorRepository, connector_svc: ConnectorService, workspace_id: UUID, project_id: UUID, request_type: str, enabled_capabilities: set[str], connection_id: str | None, action: str | None, approval_token: str | None, provider_hint: str | None, ): capability = canonical_payment_capability(request_type) if capability is None: return None rows = await connector_repo.list_active_connections_for_connector_keys( project_id, connector_keys=PAYMENT_PROVIDERS ) snapshots = snapshots_from_rows( [row for row, _catalog in rows], {catalog.id: catalog for _row, catalog in rows}, ) hinted: UUID | None = None if connection_id: try: hinted = UUID(connection_id) except ValueError: from src.domain.cloud.schemas import AuthorizeResponseDenied return AuthorizeResponseDenied( reason_code="invalid_connection_id", message="Invalid connector connection_id", ) resolution = resolve_payment_capability( capability=capability, snapshots=snapshots, enabled_capabilities=enabled_capabilities, connection_id=hinted, provider_hint=provider_hint, ) if not resolution.ok: from src.domain.cloud.schemas import AuthorizeResponseDenied return AuthorizeResponseDenied( reason_code=resolution.reason_code or "capability_not_bound", message=resolution.message or "Payment capability is not bound", ) if resolution.connection_id is None: from src.domain.cloud.schemas import AuthorizeResponseDenied return AuthorizeResponseDenied( reason_code="capability_not_bound", message="Connect Stripe or PayPal before invoking payments capabilities", ) if not resolution.action: return resolution connector_decision = await connector_svc.authorize_execution( workspace_id=workspace_id, project_id=project_id, connection_id=resolution.connection_id, action_slug=resolution.action, enabled_capabilities=enabled_capabilities, approval_token=approval_token, ) if not connector_decision.allowed: from src.domain.cloud.schemas import AuthorizeResponseDenied return AuthorizeResponseDenied( reason_code=connector_decision.reason_code or "connector_denied", message=connector_decision.message or "Connector action is not allowed", ) return (resolution, connector_decision) class CloudInternalService: def __init__( self, *, db: AsyncSession | None = None, project_repo: ProjectRepository, workspace_repo: WorkspaceRepository, settings_repo: ProjectSettingsRepository, subscription_repo: SubscriptionRepository, usage_ledger: UsageLedgerRepository, storage_usage_repo: ProjectStorageUsageRepository, connector_repo: ConnectorRepository, voice_repo: VoiceRepository | None = None, livechat_repo: LivechatRepository | None = None, runtime_wallet: RuntimeAiWallet | None = None, cloud_credits: CloudCreditService | None = None, email_domain_repo: EmailDomainRepository | None = None, email_quota_service: EmailQuotaService | None = None, ) -> None: self._db = db self._project_repo = project_repo self._workspace_repo = workspace_repo self._settings_repo = settings_repo self._subscription_repo = subscription_repo self._usage_ledger = usage_ledger self._storage_usage = storage_usage_repo self._connector_repo = connector_repo self._voice_repo = voice_repo self._livechat_repo = livechat_repo self._runtime_wallet = runtime_wallet self._cloud_credits = cloud_credits self._email_domain_repo = email_domain_repo self._email_quota_service = email_quota_service async def authorize(self, body: AuthorizeRequest) -> AuthorizeResponse: try: ws_id = UUID(body.workspace_id) proj_id = UUID(body.project_id) except ValueError: return AuthorizeResponseDenied( reason_code="invalid_ids", message="Invalid workspace_id or project_id" ) project = await self._project_repo.get_by_id(proj_id) if not project or project.deleted_at is not None: return AuthorizeResponseDenied(reason_code="project_not_found", message="Project not found") if project.workspace_id != ws_id: return AuthorizeResponseDenied( reason_code="workspace_mismatch", message="Project does not belong to workspace" ) if project.status != "ready": return AuthorizeResponseDenied( reason_code="project_not_ready", message="Project is not ready for cloud jobs" ) if body.client_id != _expected_client_id(project): return AuthorizeResponseDenied( reason_code="invalid_client", message="Trusted client id does not match project" ) workspace = await self._workspace_repo.get_by_id(ws_id) if not workspace or workspace.deleted_at is not None: return AuthorizeResponseDenied(reason_code="workspace_inactive", message="Workspace not available") settings_row = await self._settings_repo.get_or_create(proj_id) settings_json = settings_row.settings or {} allowed = _allowed_request_types(settings_json) # livechat.enabled is a non-billing probe. It is not a queue request type, # so it is not in the platform-port allowlist. Capability and operator # checks still apply; credit reservation is skipped below. capability_probe = body.request_type == "livechat.enabled" if not capability_probe and body.request_type not in allowed: return AuthorizeResponseDenied( reason_code="capability_not_enabled", message=f"Request type not allowed for this project: {body.request_type}", ) enabled_capabilities = _enabled_project_capabilities(settings_json) if body.request_type.startswith("voice.") and "voice.enabled" not in enabled_capabilities: return AuthorizeResponseDenied( reason_code="capability_not_enabled", message="Voice is not enabled for this project", ) if body.request_type.startswith("livechat.") and "livechat.enabled" not in enabled_capabilities: return AuthorizeResponseDenied( reason_code="capability_not_enabled", message="Live chat is not enabled for this project", ) settings = get_settings() if str(ws_id) not in settings.quota_exempt_workspace_id_set: sub = await self._subscription_repo.get_latest_for_workspace(ws_id) billing_status = normalized_billing_status_from_row( sub, getattr(workspace, "selected_plan_slug", None), ) if ( not workspace.skip_plan_quota_limits and requires_payment_flag(billing_status, getattr(workspace, "selected_plan_slug", None)) ): return AuthorizeResponseDenied( reason_code="billing_required", message="Workspace requires an active subscription or valid free plan", ) if self._cloud_credits and not await self._cloud_credits.jobs_allowed(ws_id): return AuthorizeResponseDenied( reason_code="cloud_credits_exhausted", message=UNAVAILABLE_MESSAGE, ) connector_svc = ConnectorService( repo=self._connector_repo, cloud_client=CloudControlClient(), ) connector_decision = None email_delivery = None payment_resolution = None if body.request_type in EMAIL_JOB_TYPES: email_delivery = await resolve_email_delivery( connector_service=connector_svc, workspace_id=ws_id, project_id=proj_id, enabled_capabilities=enabled_capabilities, connection_id=body.connection_id, action=body.action, approval_token=body.approval_token, provider_hint=body.request_meta.provider_hint, settings=settings, email_domain_repo=self._email_domain_repo, email_quota_service=self._email_quota_service, ) if not email_delivery.allowed: return AuthorizeResponseDenied( reason_code=email_delivery.reason_code or "email_provider_not_configured", message=email_delivery.message or "Email delivery is not configured", ) connector_decision = email_delivery.connector_decision elif body.request_type in PAYMENT_JOB_TYPES: payment_result = await _payment_resolution_for_authorize( connector_repo=self._connector_repo, connector_svc=connector_svc, workspace_id=ws_id, project_id=proj_id, request_type=body.request_type, enabled_capabilities=enabled_capabilities, connection_id=body.connection_id, action=body.action, approval_token=body.approval_token, provider_hint=body.request_meta.provider_hint, ) if isinstance(payment_result, AuthorizeResponseDenied): return payment_result if isinstance(payment_result, tuple): payment_resolution, connector_decision = payment_result else: payment_resolution = payment_result elif body.request_type == "reminder.send" and not (body.connection_id or body.action): email_delivery = platform_default_email_delivery(settings) elif body.request_type == "integration.execute" and body.action == "p2hs.sms.send": sms_delivery = platform_default_sms_delivery(settings) if not sms_delivery.allowed: return AuthorizeResponseDenied( reason_code=sms_delivery.reason_code or "sms_provider_not_configured", message=sms_delivery.message or "SMS delivery is not configured", ) elif body.connection_id or body.action or body.request_type == "integration.execute": if not body.connection_id or not body.action: return AuthorizeResponseDenied( reason_code="connector_request_incomplete", message="connection_id and action are required for integration execution", ) try: connection_id = UUID(body.connection_id) except ValueError: return AuthorizeResponseDenied( reason_code="invalid_connection_id", message="Invalid connector connection_id", ) connector_decision = await connector_svc.authorize_execution( workspace_id=ws_id, project_id=proj_id, connection_id=connection_id, action_slug=body.action, enabled_capabilities=enabled_capabilities, approval_token=body.approval_token, ) if not connector_decision.allowed: return AuthorizeResponseDenied( reason_code=connector_decision.reason_code or "connector_denied", message=connector_decision.message or "Connector action is not allowed", ) pol = _REQUEST_TYPE_POLICY.get(body.request_type, _DEFAULT_EXECUTION_POLICY) provider = body.request_meta.provider_hint if email_delivery and email_delivery.provider: provider = email_delivery.provider connector_key = None connection_id_snapshot = None action_snapshot = None permission_mode = None required_scopes: list[str] = [] required_capabilities: list[str] = [] if payment_resolution: provider = payment_resolution.provider or provider connector_key = payment_resolution.provider connection_id_snapshot = ( str(payment_resolution.connection_id) if payment_resolution.connection_id else None ) action_snapshot = payment_resolution.action if connector_decision: provider = connector_decision.connector.key if connector_decision.connector else provider connector_key = connector_decision.connector.key if connector_decision.connector else None connection_id_snapshot = str(connector_decision.connection.id) if connector_decision.connection else None action_snapshot = connector_decision.action.slug if connector_decision.action else body.action permission_mode = connector_decision.permission.mode if connector_decision.permission else "deny" required_scopes = ( list(connector_decision.action.required_scopes) if connector_decision.action else [] ) required_capabilities = ( list(connector_decision.action.required_capabilities) if connector_decision.action else [] ) action_for_policy = action_snapshot or body.action if action_for_policy in _LONG_RUNNING_CONNECTOR_ACTIONS: pol = { **pol, "max_attempts": max(pol.get("max_attempts", 8), 50), "max_runtime_seconds": max(pol.get("max_runtime_seconds", 120), 900), } policy_snapshot = PolicySnapshot( capability=body.request_type, provider=provider, connector_key=connector_key, connection_id=connection_id_snapshot, action=action_snapshot, permission_mode=permission_mode, required_scopes=required_scopes, required_capabilities=required_capabilities, max_attempts=pol["max_attempts"], max_runtime_seconds=pol["max_runtime_seconds"], requires_webhook=pol["requires_webhook"], artifact_retention_days=pol["artifact_retention_days"], ) if email_delivery: policy_snapshot.delivery_scope = email_delivery.delivery_scope policy_snapshot.email_delivery_scope = email_delivery.delivery_scope if not policy_snapshot.provider: policy_snapshot.provider = email_delivery.provider if email_delivery.delivery_scope == "internal": apply_platform_email_branding( policy_snapshot, settings, project_name=getattr(project, "name", None) ) elif email_delivery.delivery_scope == "platform_domain": policy_snapshot.email_from = email_delivery.email_from policy_snapshot.email_configuration_set = email_delivery.email_configuration_set # No platform footer here — this is the workspace's own branded domain, # not the shared P2HS default sender. if body.request_type.startswith("voice.") and self._voice_repo: resolution = await VoiceService( repo=self._voice_repo, cloud_client=CloudControlClient(), ).resolve_inbound(proj_id, body.request_meta.to_number) policy_snapshot.agent_id = resolution.agent_id policy_snapshot.agent_config_version = resolution.agent_config_version policy_snapshot.phone_number_id = resolution.phone_number_id policy_snapshot.voice_provider = resolution.voice_provider policy_snapshot.model = resolution.model policy_snapshot.voice = resolution.voice policy_snapshot.instructions = resolution.instructions policy_snapshot.max_call_seconds = resolution.max_call_seconds policy_snapshot.max_concurrent_calls = resolution.max_concurrent_calls policy_snapshot.tools = resolution.tools policy_snapshot.transfer_targets = resolution.transfer_targets policy_snapshot.secret_ref = resolution.secret_ref policy_snapshot.billing_mode = resolution.billing_mode policy_snapshot.credential_ref = resolution.credential_ref policy_snapshot.usage_product = resolution.usage_product if not policy_snapshot.provider: policy_snapshot.provider = resolution.voice_provider if resolution.billing_mode != "managed": return AuthorizeResponseDenied( reason_code="voice_byok_unavailable", message="Voice BYOK is not available until the credential broker is ready. Use managed billing.", ) reserved_voice = await self._reserve_runtime( workspace_id=ws_id, project_id=proj_id, reference=body.idempotency_key, usage_product="voice", millicredits=estimate_voice_millicredits( resolution.model, max(1, body.request_meta.estimated_units or 1) ), ) if reserved_voice is None: return AuthorizeResponseDenied( reason_code="runtime_ai_credits_exhausted", message="Purchase Runtime AI credits to run managed voice calls.", ) if ( body.request_type.startswith("livechat.") and body.request_type != "livechat.enabled" and self._livechat_repo ): live_cfg = await self._livechat_repo.get_config(proj_id) billing_mode = (live_cfg.billing_mode if live_cfg else "managed") or "managed" policy_snapshot.billing_mode = billing_mode policy_snapshot.usage_product = "livechat" policy_snapshot.model = live_cfg.model if live_cfg else None policy_snapshot.agent_id = live_cfg.agent_id if live_cfg else None policy_snapshot.agent_config_version = live_cfg.agent_version_id if live_cfg else None policy_snapshot.instructions = live_cfg.instructions if live_cfg else None policy_snapshot.tools = list(live_cfg.tools_json or []) if live_cfg else [] if live_cfg is None or not live_cfg.enabled: return AuthorizeResponseDenied( reason_code="livechat_disabled", message="Live chat is not enabled for this project", ) if billing_mode == "managed": reserved_chat = await self._reserve_runtime( workspace_id=ws_id, project_id=proj_id, reference=body.idempotency_key, usage_product="livechat", millicredits=estimate_livechat_millicredits( live_cfg.provider, live_cfg.model, live_cfg.max_tokens ), ) if reserved_chat is None: return AuthorizeResponseDenied( reason_code="runtime_ai_credits_exhausted", message="Purchase Runtime AI credits to run managed live chat.", ) auth_token, auth_jti = create_cloud_authorization_token( workspace_id=ws_id, project_id=proj_id, request_type=body.request_type, idempotency_key=body.idempotency_key, ) grant_jwt, grant_exp = create_cloud_grant_jwt( workspace_id=ws_id, project_id=proj_id, authorization_jti=auth_jti, request_type=body.request_type, policy_capability=body.request_type, policy_provider=provider, max_attempts=policy_snapshot.max_attempts, max_runtime_seconds=policy_snapshot.max_runtime_seconds, requires_webhook=policy_snapshot.requires_webhook, artifact_retention_days=policy_snapshot.artifact_retention_days, ) reserved = max(1, body.request_meta.estimated_units or 1) return AuthorizeResponseAllowed( authorization_id=auth_token, grant=GrantBundle( jwt=grant_jwt, expires_at=_iso_utc(grant_exp), ), reservation=Reservation(reserved_units=reserved), policy_snapshot=policy_snapshot, ) async def authorize_system_job(self, body: SystemGrantRequest) -> AuthorizeResponse: """Grant for Cloud-internal callers with a project function token (generated-app staff BFF, edge functions, verified webhooks). No browser handshake or origin allowlist. Core still enforces project readiness, the request-type allow-list, billing, credits, email delivery, and connector execution when connection_id and action are present. """ try: proj_id = UUID(body.project_id) except ValueError: return AuthorizeResponseDenied(reason_code="invalid_ids", message="Invalid project_id") project = await self._project_repo.get_by_id(proj_id) if not project or project.deleted_at is not None: return AuthorizeResponseDenied(reason_code="project_not_found", message="Project not found") if project.status != "ready": return AuthorizeResponseDenied( reason_code="project_not_ready", message="Project is not ready for cloud jobs" ) ws_id = project.workspace_id workspace = await self._workspace_repo.get_by_id(ws_id) if not workspace or workspace.deleted_at is not None: return AuthorizeResponseDenied(reason_code="workspace_inactive", message="Workspace not available") settings_row = await self._settings_repo.get_or_create(proj_id) settings_json = settings_row.settings or {} allowed = _allowed_request_types(settings_json) if body.request_type not in allowed: return AuthorizeResponseDenied( reason_code="capability_not_enabled", message=f"Request type not allowed for this project: {body.request_type}", ) settings = get_settings() if str(ws_id) not in settings.quota_exempt_workspace_id_set: sub = await self._subscription_repo.get_latest_for_workspace(ws_id) billing_status = normalized_billing_status_from_row( sub, getattr(workspace, "selected_plan_slug", None), ) if ( not workspace.skip_plan_quota_limits and requires_payment_flag(billing_status, getattr(workspace, "selected_plan_slug", None)) ): return AuthorizeResponseDenied( reason_code="billing_required", message="Workspace requires an active subscription or valid free plan", ) if self._cloud_credits and not await self._cloud_credits.jobs_allowed(ws_id): return AuthorizeResponseDenied( reason_code="cloud_credits_exhausted", message=UNAVAILABLE_MESSAGE, ) pol = _REQUEST_TYPE_POLICY.get(body.request_type, _DEFAULT_EXECUTION_POLICY) policy_snapshot = PolicySnapshot( capability=body.request_type, provider=None, max_attempts=pol["max_attempts"], max_runtime_seconds=pol["max_runtime_seconds"], requires_webhook=pol["requires_webhook"], artifact_retention_days=pol["artifact_retention_days"], ) enabled_capabilities = _enabled_project_capabilities(settings_json) connector_svc = ConnectorService( repo=self._connector_repo, cloud_client=CloudControlClient(), ) if body.request_type in EMAIL_JOB_TYPES: if body.delivery_scope == "internal": email_delivery = platform_default_email_delivery(settings) else: email_delivery = await resolve_email_delivery( connector_service=connector_svc, workspace_id=ws_id, project_id=proj_id, enabled_capabilities=enabled_capabilities, settings=settings, email_domain_repo=self._email_domain_repo, email_quota_service=self._email_quota_service, ) if not email_delivery.allowed: return AuthorizeResponseDenied( reason_code=email_delivery.reason_code or "email_provider_not_configured", message=email_delivery.message or "Email delivery is not configured", ) policy_snapshot.delivery_scope = email_delivery.delivery_scope policy_snapshot.email_delivery_scope = email_delivery.delivery_scope policy_snapshot.provider = email_delivery.provider if email_delivery.delivery_scope == "internal": apply_platform_email_branding( policy_snapshot, settings, project_name=getattr(project, "name", None) ) elif email_delivery.delivery_scope == "platform_domain": policy_snapshot.email_from = email_delivery.email_from policy_snapshot.email_configuration_set = email_delivery.email_configuration_set elif email_delivery.connector_decision: decision = email_delivery.connector_decision policy_snapshot.connector_key = ( decision.connector.key if decision.connector else None ) policy_snapshot.connection_id = ( str(decision.connection.id) if decision.connection else None ) policy_snapshot.action = decision.action.slug if decision.action else None policy_snapshot.permission_mode = ( decision.permission.mode if decision.permission else "deny" ) elif body.request_type == "integration.execute" and body.action == "p2hs.sms.send": sms_delivery = platform_default_sms_delivery(settings) if not sms_delivery.allowed: return AuthorizeResponseDenied( reason_code=sms_delivery.reason_code or "sms_provider_not_configured", message=sms_delivery.message or "SMS delivery is not configured", ) policy_snapshot.capability = "messaging.sms.send" policy_snapshot.delivery_scope = "internal" policy_snapshot.sms_delivery_scope = "internal" policy_snapshot.sms_from = sms_delivery.from_number policy_snapshot.action = "p2hs.sms.send" elif body.request_type in PAYMENT_JOB_TYPES: payment_result = await _payment_resolution_for_authorize( connector_repo=self._connector_repo, connector_svc=connector_svc, workspace_id=ws_id, project_id=proj_id, request_type=body.request_type, enabled_capabilities=enabled_capabilities, connection_id=body.connection_id, action=body.action, approval_token=body.approval_token, provider_hint=None, ) if isinstance(payment_result, AuthorizeResponseDenied): return payment_result if isinstance(payment_result, tuple): payment_resolution, connector_decision = payment_result else: payment_resolution, connector_decision = payment_result, None policy_snapshot.capability = payment_resolution.capability policy_snapshot.provider = payment_resolution.provider policy_snapshot.connector_key = payment_resolution.provider policy_snapshot.connection_id = ( str(payment_resolution.connection_id) if payment_resolution.connection_id else None ) policy_snapshot.action = payment_resolution.action if connector_decision: policy_snapshot.permission_mode = ( connector_decision.permission.mode if connector_decision.permission else "deny" ) elif body.request_type == "reminder.send" and not (body.connection_id or body.action): email_delivery = platform_default_email_delivery(settings) policy_snapshot.delivery_scope = email_delivery.delivery_scope policy_snapshot.email_delivery_scope = email_delivery.delivery_scope policy_snapshot.provider = email_delivery.provider if email_delivery.delivery_scope == "internal": apply_platform_email_branding( policy_snapshot, settings, project_name=getattr(project, "name", None) ) elif body.connection_id or body.action or body.request_type == "integration.execute": if not body.connection_id or not body.action: return AuthorizeResponseDenied( reason_code="connector_request_incomplete", message="connection_id and action are required for integration execution", ) try: connection_id = UUID(body.connection_id) except ValueError: return AuthorizeResponseDenied( reason_code="invalid_connection_id", message="Invalid connector connection_id", ) connector_decision = await connector_svc.authorize_execution( workspace_id=ws_id, project_id=proj_id, connection_id=connection_id, action_slug=body.action, enabled_capabilities=enabled_capabilities, approval_token=body.approval_token, ) if not connector_decision.allowed: return AuthorizeResponseDenied( reason_code=connector_decision.reason_code or "connector_denied", message=connector_decision.message or "Connector action is not allowed", ) policy_snapshot.provider = ( connector_decision.connector.key if connector_decision.connector else None ) policy_snapshot.connector_key = ( connector_decision.connector.key if connector_decision.connector else None ) policy_snapshot.connection_id = ( str(connector_decision.connection.id) if connector_decision.connection else None ) policy_snapshot.action = ( connector_decision.action.slug if connector_decision.action else None ) policy_snapshot.permission_mode = ( connector_decision.permission.mode if connector_decision.permission else "deny" ) auth_token, auth_jti = create_cloud_authorization_token( workspace_id=ws_id, project_id=proj_id, request_type=body.request_type, idempotency_key=body.idempotency_key, ) grant_jwt, grant_exp = create_cloud_grant_jwt( workspace_id=ws_id, project_id=proj_id, authorization_jti=auth_jti, request_type=body.request_type, policy_capability=body.request_type, policy_provider=policy_snapshot.provider, max_attempts=policy_snapshot.max_attempts, max_runtime_seconds=policy_snapshot.max_runtime_seconds, requires_webhook=policy_snapshot.requires_webhook, artifact_retention_days=policy_snapshot.artifact_retention_days, ) return AuthorizeResponseAllowed( authorization_id=auth_token, grant=GrantBundle(jwt=grant_jwt, expires_at=_iso_utc(grant_exp)), reservation=Reservation(reserved_units=1), policy_snapshot=policy_snapshot, ) async def authorize_command(self, body: CommandAuthorizeRequest) -> AuthorizeResponse: if body.command in STUB_REQUEST_TYPES: return AuthorizeResponseDenied( reason_code="stub_request_type", message="Stub request types are not public commands", ) if body.command not in PUBLIC_COMMANDS: return AuthorizeResponseDenied( reason_code="unknown_command", message=f"Command is not on the public surface: {body.command}", ) if body.command in REQUIRES_USER_PRINCIPAL and ( body.principal_type != "user" or not (body.principal_id or "").strip() ): return AuthorizeResponseDenied( reason_code="user_principal_required", message="This command requires a mapped P2HS user principal", ) if body.command in SENSITIVE_COMMANDS: expected = command_hash( {"command": body.command, "resource": body.resource, "input": body.input} ) if not body.command_hash or body.command_hash.lower() != expected: return AuthorizeResponseDenied( reason_code="command_hash_mismatch", message="command_hash does not match the canonical payload", ) try: proj_id = UUID(body.project_id) ws_id = UUID(body.workspace_id) connection_id = UUID(body.connection_id) except ValueError: return AuthorizeResponseDenied(reason_code="invalid_ids", message="Invalid ids") project = await self._project_repo.get_by_id(proj_id) if not project or project.deleted_at is not None or project.workspace_id != ws_id: return AuthorizeResponseDenied(reason_code="project_not_found", message="Project not found") if project.status != "ready": return AuthorizeResponseDenied( reason_code="project_not_ready", message="Project is not ready" ) connection = await self._connector_repo.get_connection_for_project(proj_id, connection_id) if not connection or connection.status != "active": return AuthorizeResponseDenied( reason_code="installation_not_found", message="Connector connection is not an active installation for this project", ) settings = get_settings() workspace = await self._workspace_repo.get_by_id(ws_id) if not workspace or workspace.deleted_at is not None: return AuthorizeResponseDenied(reason_code="workspace_inactive", message="Workspace not available") if str(ws_id) not in settings.quota_exempt_workspace_id_set: sub = await self._subscription_repo.get_latest_for_workspace(ws_id) billing_status = normalized_billing_status_from_row( sub, getattr(workspace, "selected_plan_slug", None), ) if ( not workspace.skip_plan_quota_limits and requires_payment_flag(billing_status, getattr(workspace, "selected_plan_slug", None)) ): return AuthorizeResponseDenied( reason_code="billing_required", message="Workspace requires an active subscription or valid free plan", ) policy_snapshot = PolicySnapshot( capability=body.command, provider=body.source_provider, connection_id=str(connection_id), action=body.command, permission_mode="allow", max_attempts=8, max_runtime_seconds=120, requires_webhook=False, artifact_retention_days=14, command_id=body.command_id, command=body.command, command_hash=body.command_hash, principal_type=body.principal_type, principal_id=body.principal_id, environment=body.environment, grant_id=body.command_id, ) auth_token, auth_jti = create_cloud_authorization_token( workspace_id=ws_id, project_id=proj_id, request_type=body.command, idempotency_key=body.idempotency_key, ) extra = { "cmd_id": body.command_id, "cmd": body.command, "cid": str(connection_id), "prin_type": body.principal_type, "prin_id": body.principal_id, "environment": body.environment, "req_id": body.request_id, "cor_id": body.correlation_id, "idem": body.idempotency_key, } if body.command_hash: extra["cmd_hash"] = body.command_hash grant_jwt, grant_exp = create_cloud_grant_jwt( workspace_id=ws_id, project_id=proj_id, authorization_jti=auth_jti, request_type=body.command, policy_capability=body.command, policy_provider=body.source_provider, max_attempts=policy_snapshot.max_attempts, max_runtime_seconds=policy_snapshot.max_runtime_seconds, requires_webhook=False, artifact_retention_days=policy_snapshot.artifact_retention_days, extra_claims=extra, ) return AuthorizeResponseAllowed( authorization_id=auth_token, grant=GrantBundle(jwt=grant_jwt, expires_at=_iso_utc(grant_exp)), reservation=Reservation(reserved_units=1), policy_snapshot=policy_snapshot, ) async def execute_granted_command( self, *, grant_jwt: str, command: str, project_id: str, input_data: dict, command_id: str, ) -> dict: claims = decode_cloud_grant_jwt(grant_jwt) if not claims: return {"ok": False, "error": "invalid_grant"} if str(claims.get("pid")) != project_id or str(claims.get("cmd") or claims.get("sub")) != command: return {"ok": False, "error": "grant_mismatch"} if command == "codegen.request_change": prompt = str(input_data.get("prompt") or input_data.get("user_prompt") or "").strip() if not prompt: return {"ok": False, "error": "prompt_required"} from src.domain.codegen.repository import CodegenRunRepository from src.domain.codegen.router import _ProjectInfoFacadeForCodegen from src.domain.codegen.service import CodegenService from src.domain.manifest.repository import ManifestRunRepository from src.domain.manifest.service import ManifestService from src.domain.projects.repository import ProjectEnvVarRepository svc = CodegenService( codegen_repo=CodegenRunRepository(self._db), project_facade=_ProjectInfoFacadeForCodegen(self._db), project_settings_repo=ProjectSettingsRepository(self._db), env_var_repo=ProjectEnvVarRepository(self._db), manifest_service=ManifestService(db=self._db, run_repo=ManifestRunRepository(self._db)), ) principal = claims.get("prin_id") created_by = None if claims.get("prin_type") == "user" and principal: try: created_by = UUID(str(principal)) except ValueError: created_by = None run = await svc.create_run( project_id=UUID(project_id), user_prompt=prompt, created_by=created_by, supersede_active=False, ) return {"ok": True, "run_id": str(run.id), "command_id": command_id} if command == "codegen.cancel": run_id = str(input_data.get("run_id") or "") if not run_id: return {"ok": False, "error": "run_id_required"} from src.domain.codegen.repository import CodegenRunRepository from src.domain.codegen.router import _ProjectInfoFacadeForCodegen from src.domain.codegen.service import CodegenService svc = CodegenService( codegen_repo=CodegenRunRepository(self._db), project_facade=_ProjectInfoFacadeForCodegen(self._db), ) await svc.cancel_run(UUID(run_id), UUID(project_id)) return {"ok": True, "command_id": command_id} return {"ok": False, "error": "command_not_executed_in_core", "command": command} async def usage_finalize(self, body: UsageFinalizeRequest) -> UsageAckResponse | None: """Return None if authorization_id is missing or not a valid Core token.""" if not body.authorization_id: return None claims = decode_cloud_authorization_token(body.authorization_id) if not claims: return None ws_id = UUID(claims["ws"]) proj_id = UUID(claims["pid"]) existing = await self._usage_ledger.find_by_cloud_job_id(ws_id, body.job_id) if existing: await self._settle_runtime(claims, body.job_id, body.metadata or {}, release=False) return UsageAckResponse(ok=True, duplicate=True) qty = max(0, body.actual_units) await self._usage_ledger.record( workspace_id=ws_id, event_type="cloud_job", quantity=qty or 1, unit_label="cloud_units", project_id=proj_id, metadata={ "job_id": body.job_id, "kind": "finalize", "authorization_jti": claims.get("jti"), "extra": body.metadata or {}, }, reported_to_stripe=True, ) await self._settle_runtime(claims, body.job_id, body.metadata or {}, release=False) return UsageAckResponse(ok=True, duplicate=False) async def usage_release(self, body: UsageReleaseRequest) -> UsageAckResponse | None: """Return None if authorization_id is missing or not a valid Core token.""" if not body.authorization_id: return None claims = decode_cloud_authorization_token(body.authorization_id) if not claims: return None ws_id = UUID(claims["ws"]) proj_id = UUID(claims["pid"]) existing = await self._usage_ledger.find_by_cloud_job_id(ws_id, body.job_id) if existing: await self._settle_runtime(claims, body.job_id, body.metadata or {}, release=True) return UsageAckResponse(ok=True, duplicate=True) await self._usage_ledger.record( workspace_id=ws_id, event_type="cloud_job", quantity=0, unit_label=None, project_id=proj_id, metadata={ "job_id": body.job_id, "kind": "release", "failure_stage": body.failure_stage, "reason_code": body.reason_code, "authorization_jti": claims.get("jti"), "extra": body.metadata or {}, }, reported_to_stripe=True, ) await self._settle_runtime(claims, body.job_id, body.metadata or {}, release=True) return UsageAckResponse(ok=True, duplicate=False) async def usage_storage( self, body: StorageUsageReportRequest ) -> StorageUsageAckResponse: """Apply a storage delta from Cloud and update the running counter in Core.""" try: project_id = UUID(body.project_id) ws_id = UUID(body.workspace_id) except ValueError: return StorageUsageAckResponse(ok=False, total_bytes=0, total_files=0) row = await self._storage_usage.apply_delta( project_id=project_id, delta_bytes=body.delta_bytes, delta_files=body.delta_files, ) # Record audit event to usage ledger await self._usage_ledger.record( workspace_id=ws_id, event_type=f"storage_{body.event}", quantity=abs(body.delta_bytes), unit_label="bytes", project_id=project_id, metadata={ "event": body.event, "delta_bytes": body.delta_bytes, "delta_files": body.delta_files, "file_id": body.file_id, "bucket_name": body.bucket_name, }, reported_to_stripe=True, ) return StorageUsageAckResponse( ok=True, total_bytes=row.total_bytes, total_files=row.total_files, ) async def _reserve_runtime( self, *, workspace_id: UUID, project_id: UUID, reference: str, usage_product: str, millicredits: int, ): if not self._runtime_wallet: return "unconfigured" return await self._runtime_wallet.reserve( workspace_id=workspace_id, millicredits=max(1, millicredits), reference=reference, usage_product=usage_product, project_id=project_id, ) async def _settle_runtime( self, claims: dict, job_id: str, metadata: dict, *, release: bool, ) -> None: if not self._runtime_wallet: return request_type = str(claims.get("rt") or "") if not (request_type.startswith("livechat.") or request_type.startswith("voice.")): return if metadata.get("billing_mode") == "byok": return usage_product = str(metadata.get("usage_product") or ("livechat" if request_type.startswith("livechat.") else "voice")) reference = str(claims.get("idem") or job_id) workspace_id = UUID(claims["ws"]) try: if release: await self._runtime_wallet.release( workspace_id=workspace_id, reference=reference, usage_product=usage_product ) return charged = millicredits_for_usage( provider=str(metadata.get("provider") or "") or None, model=str(metadata.get("model") or "") or None, input_tokens=int(metadata.get("input_tokens") or 0), output_tokens=int(metadata.get("output_tokens") or 0), audio_input_seconds=int( metadata.get("audio_input_seconds") or (0 if metadata.get("input_tokens") else metadata.get("voice_seconds") or 0) ), audio_output_seconds=int( metadata.get("audio_output_seconds") or (0 if metadata.get("output_tokens") else metadata.get("voice_seconds") or 0) ), stt_seconds=int(metadata.get("stt_seconds") or 0), ) await self._runtime_wallet.settle( workspace_id=workspace_id, reference=reference, charged_millicredits=max(0, charged), usage_product=usage_product, ) except ValueError: return async def ingest_crm_inbound(self, body: CRMInboundEventsRequest) -> dict: """Normalize verified CRM provider events. Does not call authorize_execution.""" from src.domain.connectors.crm_events import CRMEventError from src.domain.connectors.service import ConnectorService from src.integrations.cloud_control_client import CloudControlClient claimed = None if body.claimed_project_id: claimed = UUID(body.claimed_project_id) connector_svc = ConnectorService( repo=self._connector_repo, cloud_client=CloudControlClient(), ) try: result = await connector_svc.accept_crm_inbound_events( provider=body.provider, verified=body.verified, payload=body.payload, provider_account_id=body.provider_account_id, claimed_project_id=claimed, ) except CRMEventError as exc: raise exc return { "ok": True, "accepted": [row.model_dump() for row in result.accepted], "duplicates": result.duplicates, "suppressed": result.suppressed, "ignored": result.ignored, "replays": [row.model_dump() for row in result.replays], } def _iso_utc(dt: datetime) -> str: if dt.tzinfo is None: dt = dt.replace(tzinfo=UTC) return dt.astimezone(UTC).strftime("%Y-%m-%dT%H:%M:%SZ")