"""document.* Core resolver + internal execution plans. Validate capability input, resolve a customer-bound google_docs connection, and map to google_docs.* plans. Catalog ``invocable`` is runtime truth (2F). Does not persist created document IDs. Does not call authorize_execution. """ from __future__ import annotations import re from dataclasses import dataclass, field from typing import Any from uuid import UUID from src.domain.connectors.invoke_capabilities import ( DOCUMENT_FORBIDDEN_METADATA_KEYS, DOCUMENT_FROZEN_VERBS, DOCUMENT_PROVIDER_INPUT_KEYS, GOOGLE_DOCS_APPEND_TEXT_MAPPING, GOOGLE_DOCS_CREATE_MAPPING, GOOGLE_DOCS_GET_MAPPING, GOOGLE_DOCS_INSERT_TEXT_MAPPING, GOOGLE_DOCS_REPLACE_TEXT_MAPPING, INVOCABLE_FALSE_MEANS, INVOKE_CAPABILITIES, capability_surface, ) EXISTING_DOCUMENT_VERBS = frozenset({ "document.get", "document.insert_text", "document.append_text", "document.replace_text", }) CREATE_CAPABILITY = "document.create" _LOGICAL_DOCUMENT = re.compile(r"^[a-z][a-z0-9-]{0,62}$") _PUBLIC_RESULT_KEYS = frozenset({ "capability", "status", "document", "reason_code", "message", }) _PROVIDER_SHAPED_KEYS = frozenset({ "document_id", "revision_id", "google_document_id", "docs_document_id", "index", "match_case", "action", "connection_id", "request_type", "payload", }) ACTION_BY_CAPABILITY = { "document.create": GOOGLE_DOCS_CREATE_MAPPING["provider_action"], "document.get": GOOGLE_DOCS_GET_MAPPING["provider_action"], "document.insert_text": GOOGLE_DOCS_INSERT_TEXT_MAPPING["provider_action"], "document.append_text": GOOGLE_DOCS_APPEND_TEXT_MAPPING["provider_action"], "document.replace_text": GOOGLE_DOCS_REPLACE_TEXT_MAPPING["provider_action"], } GOOGLE_DOCS_CONNECTOR = "google_docs" def _catalog_invocable(capability: str) -> bool: return bool(capability_surface(capability, []).get("invocable")) def normalize_logical_document(value: str) -> str | None: raw = value.strip() if not _LOGICAL_DOCUMENT.match(raw): return None return raw def connection_document_alias(config: dict[str, Any]) -> str | None: raw = config.get("document") if isinstance(raw, str) and raw.strip(): return normalize_logical_document(raw) return None def provider_document_id_for_alias( config: dict[str, Any], alias: str, ) -> str | None: mapped = config.get("document_map") if isinstance(mapped, dict): value = mapped.get(alias) if isinstance(value, str) and value.strip(): return value.strip() if connection_document_alias(config) == alias: document_id = config.get("document_id") if isinstance(document_id, str) and document_id.strip(): return document_id.strip() return None @dataclass(frozen=True) class DocumentConnectionSnapshot: connection_id: UUID connector_key: str status: str has_secrets: bool permission_mode: str config: dict[str, Any] = field(default_factory=dict) requires_secrets: bool = True client_side_only: bool = False @dataclass(frozen=True) class DocumentPlan: capability: str connection_id: UUID action: str provider: str document: str | None resolved_input: dict[str, Any] def integration_execute_payload(self) -> dict[str, Any]: return { "request_type": "integration.execute", "connection_id": str(self.connection_id), "action": self.action, "input": dict(self.resolved_input), } @dataclass(frozen=True) class DocumentResolution: ok: bool reason_code: str | None = None message: str | None = None plan: DocumentPlan | None = None diagnostics: dict[str, Any] = field(default_factory=dict) @dataclass(frozen=True) class DispatchAuthorization: """integration.execute authorization — independent of resolver eligibility.""" allowed: bool reason_code: str | None = None message: str | None = None @dataclass(frozen=True) class DocumentPublicResult: capability: str status: str = "rejected" document: dict[str, Any] | None = None reason_code: str | None = None message: str | None = None def as_dict(self) -> dict[str, Any]: body: dict[str, Any] = { "capability": self.capability, "status": self.status, } if self.document is not None: body["document"] = dict(self.document) if self.reason_code is not None: body["reason_code"] = self.reason_code if self.message is not None: body["message"] = self.message return body def _fail( code: str, message: str, diagnostics: dict[str, Any] | None = None, ) -> DocumentResolution: return DocumentResolution( ok=False, reason_code=code, message=message, diagnostics=diagnostics or {}, ) def _forbidden_provider_fields(payload: dict[str, Any]) -> list[str]: return sorted(key for key in payload if key in DOCUMENT_PROVIDER_INPUT_KEYS) def _planned_connector(connector_key: str) -> bool: return connector_key == GOOGLE_DOCS_CONNECTOR def map_document_capability_input( capability: str, *, document_id: str | None, payload: dict[str, Any], ) -> dict[str, Any] | str: """Generic fields → google_docs action input. Provider extras stay omitted.""" if capability == CREATE_CAPABILITY: mapped: dict[str, Any] = {} title = payload.get("title") if isinstance(title, str) and title.strip(): mapped["title"] = title.strip() return mapped if document_id is None: return "document_unbound" if capability == "document.get": return {"document_id": document_id} if capability == "document.append_text": text = payload.get("text") if not isinstance(text, str) or not text.strip(): return "text_required" return {"document_id": document_id, "text": text} if capability == "document.insert_text": text = payload.get("text") if not isinstance(text, str) or not text.strip(): return "text_required" position = payload.get("position", "start") if position is None or position == "": position = "start" if position != "start": return "position_invalid" index = GOOGLE_DOCS_INSERT_TEXT_MAPPING["position_map"]["start"] return {"document_id": document_id, "text": text, "index": index} if capability == "document.replace_text": find = payload.get("find") if not isinstance(find, str) or not find.strip(): return "find_required" replace = payload.get("replace") mapped = {"document_id": document_id, "find": find} if isinstance(replace, str): mapped["replace"] = replace return mapped return "document_invalid" def _paragraph_text(element: dict[str, Any]) -> str: paragraph = element.get("paragraph") if not isinstance(paragraph, dict): return "" chunks: list[str] = [] for piece in paragraph.get("elements") or []: if not isinstance(piece, dict): continue run = piece.get("textRun") if isinstance(run, dict) and isinstance(run.get("content"), str): chunks.append(run["content"]) return "".join(chunks) def normalize_google_document(raw: dict[str, Any]) -> dict[str, str]: """Provider Docs JSON → capability document. No Google IDs or revision.""" title = raw.get("title") body = raw.get("body") if isinstance(raw.get("body"), dict) else {} parts: list[str] = [] for element in body.get("content") or []: if isinstance(element, dict): parts.append(_paragraph_text(element)) return { "title": title.strip() if isinstance(title, str) else "", "body_text": "".join(parts), } def _enqueue_provider_output(enqueue: dict[str, Any]) -> dict[str, Any] | None: for key in ("document", "result", "output"): value = enqueue.get(key) if isinstance(value, dict) and value: return value return None def resolve_document( *, capability: str, payload: dict[str, Any], connections: list[DocumentConnectionSnapshot], enabled_capabilities: set[str] | None = None, provider_hint: str | None = None, binding_connection_id: UUID | None = None, document_map: dict[str, UUID] | None = None, approval_present: bool = False, enforce_invocable: bool = True, ) -> DocumentResolution: """Deterministic plan. Does not call Google or persist bindings.""" diagnostics: dict[str, Any] = {"rejected": []} if capability not in DOCUMENT_FROZEN_VERBS or capability not in INVOKE_CAPABILITIES: return _fail("capability_not_bound", f"Unknown capability {capability}") if enforce_invocable and not _catalog_invocable(capability): return _fail("capability_not_invocable", INVOCABLE_FALSE_MEANS) leaked = _forbidden_provider_fields(payload) if leaked: return _fail( "provider_fields_forbidden", "document.* does not accept Google Docs API fields; " "use google_docs.* provider actions", {"forbidden_fields": leaked}, ) metadata = payload.get("metadata") if isinstance(payload.get("metadata"), dict) else {} stripped = sorted(set(metadata) & set(DOCUMENT_FORBIDDEN_METADATA_KEYS)) if stripped: diagnostics["stripped_metadata_keys"] = stripped alias: str | None = None if capability in EXISTING_DOCUMENT_VERBS: document_raw = payload.get("document") if not isinstance(document_raw, str) or not document_raw.strip(): return _fail("document_invalid", f"{capability} requires a logical document alias") alias = normalize_logical_document(document_raw) if alias is None: return _fail( "document_invalid", "document must be a logical alias, not a Google document ID", ) needs_text = capability in {"document.insert_text", "document.append_text"} if needs_text: text = payload.get("text") if not isinstance(text, str) or not text.strip(): return _fail("text_required", f"{capability} requires text") if capability == "document.replace_text": find = payload.get("find") if not isinstance(find, str) or not find.strip(): return _fail("find_required", "document.replace_text requires find") if capability == "document.insert_text": position = payload.get("position", "start") if position is None or position == "": position = "start" if position != "start": return _fail( "position_invalid", "document.insert_text position is abstract start only; " "end-of-body insert is document.append_text", ) candidates = [ row for row in connections if _planned_connector(row.connector_key) and not row.client_side_only ] hint = (provider_hint or "").strip().lower() if hint: candidates = [row for row in candidates if row.connector_key == hint] eligible: list[DocumentConnectionSnapshot] = [] for row in candidates: reason = _ineligibility_reason(row, enabled_capabilities, approval_present) if reason: diagnostics["rejected"].append({ "connection_id": str(row.connection_id), "connector_key": row.connector_key, "reason": reason, }) continue eligible.append(row) if binding_connection_id is not None: intended = next( (row for row in connections if row.connection_id == binding_connection_id), None, ) if intended is None or not _planned_connector(intended.connector_key): return _fail( "capability_not_bound", "Explicit document binding does not match a project google_docs connection", diagnostics, ) unique_fail = _unique_connection_gate( intended, enabled_capabilities, approval_present ) if unique_fail: return unique_fail return _plan_for(intended, capability, payload, alias, diagnostics) if alias and document_map and alias in document_map: mapped_id = document_map[alias] mapped = next((row for row in eligible if row.connection_id == mapped_id), None) if mapped is None: return _fail( "document_unbound", "document_map does not match an eligible google_docs connection", diagnostics, ) unique_fail = _unique_connection_gate( mapped, enabled_capabilities, approval_present ) if unique_fail: return unique_fail return _plan_for(mapped, capability, payload, alias, diagnostics) if alias: aliased = [ row for row in eligible if provider_document_id_for_alias(row.config, alias) is not None or connection_document_alias(row.config) == alias ] if len(aliased) == 1: return _plan_for(aliased[0], capability, payload, alias, diagnostics) if len(aliased) > 1: return _fail( "binding_ambiguous", "Multiple eligible google_docs connections share this document alias", diagnostics, ) if not eligible: return _fail( "capability_not_bound", "No eligible customer-bound google_docs connection. " "Not P2HS-owned Google and not spreadsheet.* .", diagnostics, ) if len(eligible) > 1: return _fail( "binding_ambiguous", "Multiple eligible google_docs connections; set a document binding. " "provider_hint=google_docs does not select among them.", { **diagnostics, "eligible": [ { "connection_id": str(row.connection_id), "connector_key": row.connector_key, } for row in eligible ], }, ) if capability == CREATE_CAPABILITY: return _plan_for(eligible[0], capability, payload, alias, diagnostics) return _fail( "document_unbound", "Logical document is not bound on the eligible connection; " "set connection.config.document / document_map. Not first-wins.", diagnostics, ) def _unique_connection_gate( row: DocumentConnectionSnapshot, enabled_capabilities: set[str] | None, approval_present: bool, ) -> DocumentResolution | None: if row.status != "active": return _fail("connection_inactive", "Bound google_docs connection is not active") if row.requires_secrets and not row.has_secrets: return _fail( "credentials_missing", "Bound google_docs connection is missing credentials", ) reason = _ineligibility_reason(row, enabled_capabilities, approval_present) if reason == "connection_inactive": return _fail("connection_inactive", "Bound google_docs connection is not active") if reason == "credentials_missing": return _fail( "credentials_missing", "Bound google_docs connection is missing credentials", ) if reason: return _fail( "capability_not_bound", "Bound google_docs connection is not eligible", {"rejected": [{"connection_id": str(row.connection_id), "reason": reason}]}, ) return None def _ineligibility_reason( row: DocumentConnectionSnapshot, enabled_capabilities: set[str] | None, approval_present: bool, ) -> str | None: if not _planned_connector(row.connector_key): return "adapter_not_planned" if row.client_side_only: return "client_side_only" if row.status != "active": return "connection_inactive" if row.requires_secrets and not row.has_secrets: return "credentials_missing" if {"integrations.execute"} - (enabled_capabilities or set()): return "entitlement_missing" mode = (row.permission_mode or "deny").lower() if mode == "deny": return "permission_denied" if mode == "ask" and not approval_present: return "approval_required" return None def _plan_for( row: DocumentConnectionSnapshot, capability: str, payload: dict[str, Any], alias: str | None, diagnostics: dict[str, Any], ) -> DocumentResolution: document_id: str | None = None if capability in EXISTING_DOCUMENT_VERBS: if alias is None: return _fail("document_invalid", f"{capability} requires a logical document alias") document_id = provider_document_id_for_alias(row.config, alias) if document_id is None: return _fail( "document_unbound", "Logical document is not bound to a provider document_id " "on the selected connection", diagnostics, ) mapped = map_document_capability_input( capability, document_id=document_id, payload=payload ) if isinstance(mapped, str): return _fail(mapped, f"{capability} could not map to a provider plan", diagnostics) notes = { **diagnostics, "execution_domain": "C", "create_does_not_persist_binding": capability == CREATE_CAPABILITY, } return DocumentResolution( ok=True, plan=DocumentPlan( capability=capability, connection_id=row.connection_id, action=str(ACTION_BY_CAPABILITY[capability]), provider=row.connector_key, document=alias, resolved_input=mapped, ), diagnostics=notes, ) def dispatch_document_plan( plan: DocumentPlan, authorization: DispatchAuthorization, *, diagnostics: dict[str, Any] | None = None, ) -> DocumentResolution: """Re-check execution authorization. Does not enqueue Cloud work.""" notes = dict(diagnostics or {}) notes["payload"] = dict(plan.resolved_input) if not authorization.allowed: notes["authorization"] = "rejected" return _fail( authorization.reason_code or "provider_rejected", authorization.message or "integration.execute rejected the document plan", notes, ) notes["authorization"] = "allowed" return DocumentResolution(ok=True, plan=plan, diagnostics=notes) def invoke_document( *, capability: str, payload: dict[str, Any], connections: list[DocumentConnectionSnapshot], authorize: DispatchAuthorization | None = None, enabled_capabilities: set[str] | None = None, provider_hint: str | None = None, binding_connection_id: UUID | None = None, document_map: dict[str, UUID] | None = None, approval_present: bool = False, enforce_invocable: bool = True, ) -> DocumentResolution: resolution = resolve_document( capability=capability, payload=payload, connections=connections, enabled_capabilities=enabled_capabilities, provider_hint=provider_hint, binding_connection_id=binding_connection_id, document_map=document_map, approval_present=approval_present, enforce_invocable=enforce_invocable, ) if not resolution.ok or resolution.plan is None or authorize is None: return resolution return dispatch_document_plan( resolution.plan, authorize, diagnostics=resolution.diagnostics ) def to_public_document_result( resolution: DocumentResolution, *, capability: str, enqueue: dict[str, Any] | None = None, ) -> DocumentPublicResult: if not resolution.ok: return DocumentPublicResult( capability=capability, status="rejected", reason_code=resolution.reason_code, message=resolution.message, ) if enqueue is None: return DocumentPublicResult( capability=capability, status="rejected", reason_code="provider_unavailable", message="document.* has no enqueue path", ) if not enqueue.get("ok"): return DocumentPublicResult( capability=capability, status="rejected", reason_code=str(enqueue.get("reason_code") or "provider_unavailable"), message=str(enqueue.get("message") or "integration.execute enqueue failed"), ) output = _enqueue_provider_output(enqueue) if output is not None: if capability == "document.get": return DocumentPublicResult( capability=capability, status="succeeded", document=normalize_google_document(output), ) return DocumentPublicResult(capability=capability, status="succeeded") return DocumentPublicResult(capability=capability, status="queued") def public_result_is_capability_shaped(body: dict[str, Any]) -> bool: keys = set(body) return keys <= _PUBLIC_RESULT_KEYS and not (keys & _PROVIDER_SHAPED_KEYS)