================================================================================
File: hikigai/appsdk/__init__.py
================================================================================

"""
hikigai-appsdk: Python SDK for invoking AI agents.

Invoke deployed agents in your applications.
"""

__version__ = "0.0.6"

from hikigai.appsdk.client import AppClient
from hikigai.appsdk.models import (
    RuntimeAgent,
    InvokeResponse,
    StreamChunk,
)
from hikigai.appsdk.streaming.session import StreamSession
from hikigai.appsdk.streaming.rooms import RoomError, RoomSession
from hikigai.appsdk.streaming.events import EventStream, EventStreamError
from hikigai.appsdk.context import SessionContext
from hikigai.appsdk.identity import IdentityClient, VerifiedUser
from hikigai.appsdk.cloud import (
    CloudCatalog,
    CloudClient,
    CloudCredentialSummary,
    CloudProvider,
    CloudRegion,
    CloudService,
    TargetValidation,
)
from hikigai.appsdk.events import (
    SIGNATURE_HEADER,
    EventsClient,
    SignatureVerificationError,
    WebhookDelivery,
    WebhookSubscription,
    parse_webhook_event,
    verify_webhook_signature,
)

# Re-export common exceptions from core
from hikigai.core.exceptions import (
    HikigaiError,
    AuthenticationError,
    PermissionDeniedError,
    RateLimitError,
    AgentNotFoundError,
    InvocationError,
    ConfigurationError,
    ConflictError,
    ValidationError,
)

__all__ = [
    "__version__",
    # Client
    "AppClient",
    # Models
    "RuntimeAgent",
    "InvokeResponse",
    "StreamChunk",
    "StreamSession",
    "RoomSession",
    "RoomError",
    # Multi-cloud deployment
    "CloudClient",
    "CloudCatalog",
    "CloudProvider",
    "CloudRegion",
    "CloudService",
    "CloudCredentialSummary",
    "TargetValidation",
    # Event bus — webhooks + live event stream
    "EventsClient",
    "EventStream",
    "EventStreamError",
    "WebhookSubscription",
    "WebhookDelivery",
    "SignatureVerificationError",
    "verify_webhook_signature",
    "parse_webhook_event",
    "SIGNATURE_HEADER",
    # Session context store
    "SessionContext",
    # End-user identity
    "IdentityClient",
    "VerifiedUser",
    # Exceptions
    "HikigaiError",
    "AuthenticationError",
    "PermissionDeniedError",
    "RateLimitError",
    "AgentNotFoundError",
    "InvocationError",
    "ConfigurationError",
    "ConflictError",
    "ValidationError",
]


================================================================================
File: hikigai/appsdk/client.py
================================================================================

"""
AppClient: Invoke AI agents in your applications.

This is the main client for application developers to call deployed agents.
"""

import os
import logging
import uuid
from typing import Optional, Dict, Any, List, Iterator, Union, TYPE_CHECKING
from datetime import datetime

if TYPE_CHECKING:  # imported lazily at runtime (see AppClient.events / .cloud)
    from hikigai.appsdk.cloud import CloudClient
    from hikigai.appsdk.events import EventsClient

from hikigai.core.api.client import APIClient
from hikigai.core.exceptions import (
    ConfigurationError,
    AgentNotFoundError,
    InvocationError,
)

from hikigai.appsdk.models.agent import RuntimeAgent
from hikigai.appsdk.models.response import InvokeResponse, StreamChunk, InvocationMetadata
from hikigai.appsdk.sona import SONAClient
from hikigai.appsdk.streaming.session import StreamSession
from hikigai.appsdk.context import SessionContext
from hikigai.appsdk.identity import IdentityClient

logger = logging.getLogger(__name__)


class AppClient:
    """
    Client for invoking AI agents.
    
    Example:
        client = AppClient(
            api_key=os.environ["HIKIGAI_API_KEY"],
            project_id=os.environ["HIKIGAI_PROJECT_ID"]
        )
        
        # Get an agent
        agent = client.agent("medical-coder")
        
        # Invoke
        response = agent.invoke("Patient has fever...")
        print(response.content)
        
        # Stream
        for chunk in agent.stream("Tell me about diabetes"):
            print(chunk, end="")
    """
    
    def __init__(
        self,
        api_key: Optional[str] = None,
        project_id: Optional[str] = None,
        base_url: Optional[str] = None,
        timeout: float = 30.0,
        sona_url: Optional[str] = None,
        sona_api_key: Optional[str] = None,
    ):
        """
        Initialize AppClient.
        
        Args:
            api_key: Hikigai API key (defaults to HIKIGAI_API_KEY env var)
            project_id: Project ID (defaults to HIKIGAI_PROJECT_ID env var)
            base_url: API base URL (defaults to production)
            timeout: Default request timeout in seconds
            sona_url: SONA service URL (defaults to SONA_SERVICE_URL env var or http://localhost:8002)
            sona_api_key: SONA API key (defaults to SONA_API_KEY env var)
            
        Raises:
            ConfigurationError: If required credentials are missing
        """
        # Get credentials
        self.api_key = api_key or os.environ.get("HIKIGAI_API_KEY")
        self.project_id = project_id or os.environ.get("HIKIGAI_PROJECT_ID")
        self._sona_url = sona_url
        self._sona_api_key = sona_api_key
        
        if not self.api_key:
            raise ConfigurationError(
                "API key is required. Provide it via api_key parameter or HIKIGAI_API_KEY env var."
            )
        
        if not self.project_id:
            raise ConfigurationError(
                "Project ID is required. Provide it via project_id parameter or HIKIGAI_PROJECT_ID env var."
            )
        
        # Initialize HTTP client
        self.api = APIClient(
            api_key=self.api_key,
            project_id=self.project_id,
            base_url=base_url,
            timeout=timeout,
        )
        
        logger.info(f"AppClient initialized for project: {self.project_id}")

    @property
    def storage(self):
        """Access platform-managed object storage for this project."""
        if not hasattr(self, "_storage"):
            from hikigai.appsdk.storage import StorageClient
            self._storage = StorageClient(self.api, self.project_id)
        return self._storage
    
    def agent(self, agent_id: str) -> RuntimeAgent:
        """
        Get an agent for invocation.
        
        Args:
            agent_id: Agent ID or slug
            
        Returns:
            RuntimeAgent: Agent ready for invocation
            
        Raises:
            AgentNotFoundError: If agent doesn't exist
            
        Example:
            agent = client.agent("medical-coder")
            response = agent.invoke("Patient has fever...")
        """
        logger.debug(f"Getting agent: {agent_id}")
        
        try:
            response = self.api.get(f"/api/v1/agents/{agent_id}")
            
            agent = RuntimeAgent(
                id=response.get("id", "unknown-id"),
                name=response.get("name", "Unknown Agent"),
                slug=response.get("slug", "unknown-agent"),
                version=response.get("version", "1.0.0"),
                description=response.get("description"),
                status=response.get("deployment_status", "unknown"),
                endpoint=response.get("endpoint_url"),
                timeout=response.get("timeout"),
            )
            
            return agent.set_client(self)
            
        except Exception as e:
            logger.error(f"Failed to get agent: {e}")
            raise
    
    def list_agents(
        self,
        category: Optional[str] = None,
        tags: Optional[List[str]] = None,
        search: Optional[str] = None,
    ) -> List[RuntimeAgent]:
        """
        List available agents.
        
        Args:
            category: Filter by category
            tags: Filter by tags
            search: Search query
            
        Returns:
            List of available agents
        """
        logger.debug("Listing agents")
        
        params = {}
        if category:
            params["category"] = category
        if tags:
            params["tags"] = tags
        if search:
            params["search"] = search
        
        try:
            response = self.api.get("/api/v1/agents", params=params)
            
            agents = []
            for agent_data in response.get("agents", []):
                agent = RuntimeAgent(
                    id=agent_data["id"],
                    name=agent_data["name"],
                    slug=agent_data["slug"],
                    version=agent_data.get("version", "1.0.0"),
                    description=agent_data.get("description"),
                    status=agent_data.get("deployment_status", "unknown"),
                    endpoint=agent_data.get("endpoint_url"),
                    timeout=agent_data.get("timeout"),
                )
                agents.append(agent.set_client(self))
            
            return agents
            
        except Exception as e:
            logger.error(f"Failed to list agents: {e}")
            raise

    def invoke(
        self,
        agent_id: str,
        input: Union[str, Dict[str, Any]],
        session_id: Optional[str] = None,
        provider: Optional[str] = None,
        model: Optional[str] = None,
        timeout: Optional[float] = None,
        connectors: Optional[Dict[str, Dict[str, Any]]] = None,
        plugin_context: Optional[Dict[str, Dict[str, Any]]] = None,
    ) -> InvokeResponse:
        """
        Invoke an agent directly by ID or slug.
        
        Args:
            agent_id: Agent ID or slug
            input: Input message (string or structured dict)
            session_id: Optional session ID for conversation context
            provider: Optional provider override
            model: Optional model override
            timeout: Optional timeout override
            connectors: Optional MCP connector auth (query params + optional ``headers`` dict)
            plugin_context: Optional plugin data
            
        Returns:
            InvokeResponse: Agent response with metadata
            
        Example:
            client = AppClient(...)
            response = client.invoke("medical-coder", "Patient has fever...")
        """
        return self._invoke_agent(
            agent_id=agent_id,
            input=input,
            session_id=session_id,
            provider=provider,
            model=model,
            timeout=timeout,
            connectors=connectors,
            plugin_context=plugin_context,
        )
    
    def live(
        self,
        agent_id: str,
        session_id: Optional[str] = None,
        user_id: str = "sdk-user",
        context: Optional[Dict[str, Any]] = None,
    ) -> StreamSession:
        """
        Open a direct-connect bidirectional live session with a deployed agent.

        Uses POST /live/session for auth, then a single WebSocket to Cloud Run.
        """
        sid = session_id or str(uuid.uuid4())
        return StreamSession(
            client=self,
            agent_id=agent_id,
            session_id=sid,
            user_id=user_id,
            context=context,
        )

    def get_auth_token(self) -> Dict[str, Any]:
        """
        Exchange the API Key for a short-lived JWT Auth Token.

        Returns:
            Dict[str, Any]: Token details including access_token and expires_in.
        """
        logger.debug("Exchanging API Key for Auth Token")
        return self.api.post("/api/v1/auth/exchange")

    def create_session(
        self,
        *,
        external_user_ref: Optional[str] = None,
        ttl_seconds: Optional[int] = None,
        app_id: str = "_",
        metadata: Optional[Dict[str, Any]] = None,
    ) -> SessionContext:
        """Create a new session context store and return a bound handle.

        The platform mints an opaque ``session_id``. Store the returned
        ``session.session_id`` against your own end-user so you can rebind to
        it later with :meth:`session`.

        Args:
            external_user_ref: Your opaque reference for the end user (optional).
            ttl_seconds: Requested TTL; clamped to the platform min/max.
            app_id: Forward-compatible app identifier (defaults to "_").
            metadata: Arbitrary JSON metadata stored with the session.

        Returns:
            SessionContext: Handle bound to the new session.

        Example:
            sess = client.create_session(external_user_ref="dr-smith")
            sess.set("triage", result)
        """
        body: Dict[str, Any] = {"app_id": app_id}
        if external_user_ref is not None:
            body["external_user_ref"] = external_user_ref
        if ttl_seconds is not None:
            body["ttl_seconds"] = ttl_seconds
        if metadata is not None:
            body["metadata"] = metadata

        logger.debug("Creating session (user_ref=%s)", external_user_ref)
        result = self.api.post("/api/v1/sessions", json=body)
        return SessionContext(self, result["session_id"], app_id=result.get("app_id", app_id))

    def session(self, session_id: str, *, app_id: str = "_") -> SessionContext:
        """Bind to an existing session by id (no network call).

        Args:
            session_id: An opaque session id previously returned by
                :meth:`create_session`.
            app_id: App bucket the session belongs to (defaults to "_").

        Returns:
            SessionContext: Handle bound to the session.
        """
        return SessionContext(self, session_id, app_id=app_id)

    def list_sessions(
        self,
        *,
        user_ref: Optional[str] = None,
        limit: int = 50,
        cursor: Optional[str] = None,
    ) -> Dict[str, Any]:
        """List active sessions for the project.

        Args:
            user_ref: Filter by ``external_user_ref`` (optional).
            limit: Page size (1-200).
            cursor: Opaque pagination cursor from a previous response.

        Returns:
            ``{"sessions": [...], "next_cursor": str | None}``.
        """
        params: Dict[str, Any] = {"limit": limit}
        if user_ref is not None:
            params["user_ref"] = user_ref
        if cursor is not None:
            params["cursor"] = cursor
        return self.api.get("/api/v1/sessions", params=params)

    @property
    def identity(self) -> IdentityClient:
        """Access the End-User Identity Service (verify QR/PIN/SSO, manage end users).

        Example:
            user = client.identity.verify_qr(scanned_payload)
            print(user.first_name, user.last_name, user.role)
        """
        if not hasattr(self, "_identity"):
            self._identity = IdentityClient(self)
        return self._identity

    @property
    def cloud(self) -> "CloudClient":
        """Multi-cloud deployment catalog and cloud credentials.

        Providers, services, regions, zones, and each service's config
        schema — the same data the console renders its deployment form
        from — plus BYOC credential management.
        """
        if not hasattr(self, "_cloud"):
            from hikigai.appsdk.cloud import CloudClient
            self._cloud = CloudClient(self)
        return self._cloud

    @property
    def events(self) -> "EventsClient":
        """Access the platform event bus: webhooks and the live event stream.

        Webhooks push events to an endpoint you own (at-least-once, signed);
        ``events.stream()`` opens a live WebSocket subscription instead.
        """
        if not hasattr(self, "_events"):
            from hikigai.appsdk.events import EventsClient
            self._events = EventsClient(self)
        return self._events

    @property
    def sona(self) -> SONAClient:
        """Access SONA personalization features (edit tracking, preferences, suggestions).

        The client talks directly to the SONA service.  Configure via
        ``sona_url`` constructor arg or ``SONA_SERVICE_URL`` env var.
        """
        if not hasattr(self, "_sona"):
            self._sona = SONAClient(
                sona_url=getattr(self, "_sona_url", None),
                sona_api_key=getattr(self, "_sona_api_key", None),
            )
        return self._sona

    def _invoke_agent(
        self,
        agent_id: str,
        input: Union[str, Dict[str, Any]],
        session_id: Optional[str] = None,
        provider: Optional[str] = None,
        model: Optional[str] = None,
        timeout: Optional[float] = None,
        connectors: Optional[Dict[str, Dict[str, Any]]] = None,
        plugin_context: Optional[Dict[str, Dict[str, Any]]] = None,
    ) -> InvokeResponse:
        """
        Internal method to invoke an agent.
        
        Called by RuntimeAgent.invoke().
        """
        # Prepare payload
        payload: Dict[str, Any] = {"input": input}
        
        if session_id:
            payload["session_id"] = session_id
        if provider:
            payload["provider"] = provider
        if model:
            payload["model"] = model
        if connectors:
            payload["connectors"] = connectors
        if plugin_context:
            payload["plugin_context"] = plugin_context

        http_timeout = timeout if timeout is not None else self.api.timeout
        if timeout is not None:
            payload["timeout"] = int(timeout)
        
        logger.debug(f"Invoking agent {agent_id} (http_timeout={http_timeout}s)")
        
        try:
            response = self.api.post(
                f"/api/v1/agents/{agent_id}/invoke",
                json=payload,
                timeout=http_timeout,
            )
            
            # Parse response
            # Check for flat structure (current backend) vs nested structure (legacy/other)
            if "content" in response:
                content = response["content"]
                metadata_data = response # Use the whole response as metadata source
                
                # If content is a dict/json string, we might want to parse it, 
                # but for now we trust the backend returns a string as per schema
            else:
                # Legacy/Nested structure
                output_data = response.get("output", {})
                metadata_data = response.get("metadata", {})
                
                # Extract content
                if isinstance(output_data, dict):
                    content = output_data.get("content", output_data.get("response", str(output_data)))
                else:
                    content = str(output_data)
            
            # Build metadata
            metadata = InvocationMetadata(
                invocation_id=metadata_data.get("request_id", metadata_data.get("invocation_id", str(uuid.uuid4()))),
                latency_ms=metadata_data.get("latency_ms"),
                timestamp=datetime.fromisoformat(metadata_data["timestamp"]) 
                    if "timestamp" in metadata_data 
                    else datetime.utcnow(),
                status=metadata_data.get("status", "success"),
                tokens_used=metadata_data.get("tokens_used"),
                tools_called=metadata_data.get("tools_called", []),
                phi_redacted=metadata_data.get("phi_redacted", True),
                trace_id=metadata_data.get("trace_id"),
            )
            
            return InvokeResponse(
                content=content,
                agent_id=response.get("agent_id", agent_id),
                agent_version=response.get("version"),
                session_id=session_id,
                status=response.get("status", "success"),
                output=response.get("output"),
                confidence=response.get("confidence"),
                safety_flags=response.get("safety_flags", []),
                citations=response.get("citations", []),
                metadata=metadata,
                message=response.get("message"),
                plugins=response.get("plugins"),
                downstream_context=response.get("downstream_context"),
            )
            
        except Exception as e:
            logger.error(f"Invocation failed: {e}")
            raise InvocationError(
                f"Failed to invoke agent: {e}",
                agent_id=agent_id,
                session_id=session_id
            )
    
    def _stream_agent(
        self,
        agent_id: str,
        input: Union[str, Dict[str, Any]],
        session_id: Optional[str] = None,
        provider: Optional[str] = None,
        model: Optional[str] = None,
        connectors: Optional[Dict[str, Dict[str, Any]]] = None,
    ) -> Iterator[str]:
        """
        Internal method to stream agent responses.
        
        Called by RuntimeAgent.stream().
        """
        # Prepare payload
        payload: Dict[str, Any] = {
            "input": input,
            "stream": True,
        }
        
        if session_id:
            payload["session_id"] = session_id
        if provider:
            payload["provider"] = provider
        if model:
            payload["model"] = model
        if connectors:
            payload["connectors"] = connectors
        
        logger.debug(f"Streaming from agent {agent_id}")
        
        try:
            # Use streaming endpoint
            with self.api.stream(
                "POST",
                f"/api/v1/agents/{agent_id}/invoke",
                json=payload,
            ) as response:
                for line in response.iter_lines():
                    if line:
                        # Parse SSE format: "data: {content}"
                        try:
                            line_str = line.decode('utf-8') if isinstance(line, bytes) else line
                        except UnicodeDecodeError:
                            logger.warning(f"Failed to decode SSE line as UTF-8: {line!r}")
                            continue
                            
                        if line_str.startswith("data: "):
                            chunk_data = line_str[6:].strip()
                            if chunk_data and chunk_data != "[DONE]":
                                yield chunk_data
                                
        except Exception as e:
            logger.error(f"Streaming failed: {e}")
            raise InvocationError(
                f"Failed to stream from agent: {e}",
                agent_id=agent_id,
                session_id=session_id
            )
    
    def close(self):
        """Close HTTP client and cleanup resources."""
        if hasattr(self, "api"):
            self.api.close()
    
    def __enter__(self):
        """Context manager support."""
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        """Cleanup when exiting context."""
        self.close()
    
    def __del__(self):
        """Ensure cleanup on deletion."""
        try:
            self.close()
        except Exception:
            pass


================================================================================
File: hikigai/appsdk/models/agent.py
================================================================================

"""
Runtime agent model for invocation.
"""

from typing import Optional, Dict, Any, Iterator, TYPE_CHECKING
from pydantic import BaseModel, Field
import uuid
from datetime import datetime

if TYPE_CHECKING:
    from hikigai.appsdk.client import AppClient

from hikigai.appsdk.models.response import InvokeResponse, StreamChunk, InvocationMetadata


class RuntimeAgent(BaseModel):
    """
    Runtime representation of an agent for invocation.
    
    This model focuses on runtime capabilities (invoke, stream, sessions).
    """
    
    model_config = {"arbitrary_types_allowed": True}
    
    id: str = Field(..., description="Agent ID")
    name: str = Field(..., description="Agent  name/slug")
    slug: str = Field(..., description="URL-friendly slug")
    version: Optional[str] = Field(None, description="Agent version")
    description: Optional[str] = Field(None, description="Short description")
    status: str = Field(..., description="Deployment status")
    endpoint: Optional[str] = Field(None, description="Invocation endpoint")
    timeout: Optional[float] = Field(
        None,
        description="Deploy-time invoke timeout in seconds (from runtime_config)",
    )
    
    # Internal reference to client (not serialized)
    _client: Optional["AppClient"] = None
    _session_id: Optional[str] = None
    
    def set_client(self, client: "AppClient") -> "RuntimeAgent":
        """Set the client reference (internal use)."""
        self._client = client
        return self
    
    def invoke(
        self,
        input: str | Dict[str, Any],
        session_id: Optional[str] = None,
        provider: Optional[str] = None,
        model: Optional[str] = None,
        timeout: Optional[float] = None,
        connectors: Optional[Dict[str, Dict[str, Any]]] = None,
        plugin_context: Optional[Dict[str, Dict[str, Any]]] = None,
    ) -> InvokeResponse:
        """
        Invoke the agent with input.
        
        Args:
            input: Input message (string or structured dict)
            session_id: Optional session ID for conversation context
            provider: Optional provider override ('gemini', 'openai', 'anthropic')
            model: Optional model override
            timeout: Optional timeout override
            connectors: Optional MCP connector auth.
                Format: {"connector-slug": {"param": "value", "headers": {"Authorization": "Bearer ..."}}}
                Flat keys become URL query params; ``headers`` is sent as HTTP headers.
            plugin_context: Optional plugin data passed to active plugins.
                Format: {"sona": {"user_id": "dr-smith-uuid", "context_type": "follow_up"}}
            
        Returns:
            InvokeResponse: Agent response with metadata.
            When SONA is active, ``response.plugins["sona"]`` contains
            the ``output_id``, ``pending_suggestions``, and ``status``.
            
        Example:
            response = agent.invoke("What is diabetes?")
            print(response.content)
            
            # With SONA personalization:
            response = agent.invoke(
                "Transcript: patient presents with...",
                plugin_context={"sona": {"user_id": "dr-smith-uuid"}}
            )
            output_id = response.plugins["sona"]["output_id"]
        """
        if not self._client:
            raise ValueError("Agent not properly initialized (missing client connection)")
        
        effective_session = session_id or self._session_id
        
        effective_timeout = timeout if timeout is not None else self.timeout

        return self._client._invoke_agent(
            agent_id=self.id,
            input=input,
            session_id=effective_session,
            provider=provider,
            model=model,
            timeout=effective_timeout,
            connectors=connectors,
            plugin_context=plugin_context,
        )
    
    def stream(
        self,
        input: str | Dict[str, Any],
        session_id: Optional[str] = None,
        provider: Optional[str] = None,
        model: Optional[str] = None,
        connectors: Optional[Dict[str, Dict[str, Any]]] = None,
    ) -> Iterator[str]:
        """
        Stream agent response in real-time.
        
        Args:
            input: Input message (string or structured dict)
            session_id: Optional session ID
            provider: Optional provider override
            model: Optional model override
            connectors: Optional MCP connector auth (query params + optional ``headers`` dict)
            
        Yields:
            str: Response chunks
            
        Example:
            for chunk in agent.stream("Tell me a story"):
                print(chunk, end="", flush=True)
        """
        if not self._client:
            raise ValueError("Agent not properly initialized (missing client connection)")
        
        effective_session = session_id or self._session_id
        
        yield from self._client._stream_agent(
            agent_id=self.id,
            input=input,
            session_id=effective_session,
            provider=provider,
            model=model,
            connectors=connectors,
        )
    
    def with_session(self, session_id: str) -> "RuntimeAgent":
        """
        Create a session-bound copy of this agent.
        
        All invocations will use the same session ID for context.
        
        Args:
            session_id: Session identifier
            
        Returns:
            RuntimeAgent: New agent instance with session bound
            
        Example:
            session_agent = agent.with_session("user-123")
            session_agent.invoke("What is diabetes?")
            session_agent.invoke("How is it treated?")  # Remembers previous context
        """
        # Create a copy with session ID
        agent_copy = self.model_copy()
        agent_copy._session_id = session_id
        agent_copy._client = self._client
        return agent_copy

    def live_session(self, session_id: Optional[str] = None) -> "StreamSession":
        """
        Create a high-performance bidirectional live session.
        Useful for real-time audio streaming and transcription.
        
        Args:
            session_id: Optional session ID override.
            
        Returns:
            StreamSession: An async context manager for the live connection.
            
        Example:
            async with agent.live_session() as session:
                await session.send_audio(microphone_data)
                async for event in session:
                    if event["type"] == "transcript":
                        print(event["text"])
        """
        from hikigai.appsdk.streaming.session import StreamSession
        return StreamSession(
            client=self._client,
            agent_id=self.id,
            session_id=session_id or self._session_id or str(uuid.uuid4())
        )


================================================================================
File: hikigai/appsdk/models/response.py
================================================================================

"""
Invocation response models.
"""

from datetime import datetime
from typing import Optional, Dict, Any, List
from pydantic import BaseModel, Field, model_validator


class InvocationMetadata(BaseModel):
    """Metadata about an agent invocation."""
    
    invocation_id: str = Field(..., description="Unique invocation ID")
    latency_ms: Optional[int] = Field(None, description="Latency in milliseconds")
    timestamp: datetime = Field(..., description="Invocation timestamp")
    status: str = Field(..., description="'success' or 'error'")
    tokens_used: Optional[int] = Field(None, description="Total tokens consumed")
    tools_called: List[str] = Field(default_factory=list, description="Tools invoked")
    phi_redacted: bool = Field(True, description="Whether PHI was redacted")
    trace_id: Optional[str] = Field(None, description="Trace ID for troubleshooting")
    
    class Config:
        json_encoders = {
            datetime: lambda v: v.isoformat()
        }


class ClinicalConfidence(BaseModel):
    """Healthcare confidence scoring."""
    score: float = Field(..., ge=0, le=1)
    quality_metrics: Optional[Dict[str, Any]] = None
    reasoning: Optional[str] = None

    @model_validator(mode="before")
    @classmethod
    def _normalize_score_field(cls, values: Any) -> Any:
        """Accept 'overall_score' as an alias for 'score' (agents may use either key)."""
        if isinstance(values, dict) and "score" not in values and "overall_score" in values:
            values = dict(values)  # avoid mutating the original
            values["score"] = values.pop("overall_score")
        return values

    @property
    def overall_score(self) -> float:
        """Alias for score for backward compatibility with some scripts."""
        return self.score


class SafetyFlag(BaseModel):
    """Healthcare safety flag."""
    category: str
    severity: str = Field(..., pattern="^(low|medium|high|critical)$")
    message: str
    mitigation: Optional[str] = None


class ClinicalCitation(BaseModel):
    """Evidence-based clinical citations."""
    id: str
    text: str
    source: str
    url: Optional[str] = None


class InvokeResponse(BaseModel):
    """
    Response from an agent invocation.
    
    Contains the agent's output and metadata about the invocation.
    When SONA is active, ``plugins["sona"]`` contains the note_id,
    preferences applied, and any pending style suggestions.
    """
    
    content: str = Field(..., description="Agent response content")
    agent_id: str = Field(..., description="Agent ID")
    agent_version: Optional[str] = Field(None, description="Agent version")
    session_id: Optional[str] = Field(None, description="Session ID")
    status: str = Field("success", description="Status (success, requires_human_review, etc.)")
    output: Optional[Dict[str, Any]] = Field(None, description="Structured agent output")
    confidence: Optional[ClinicalConfidence] = Field(None, description="Healthcare clinical confidence")
    safety_flags: List[SafetyFlag] = Field(default_factory=list, description="Healthcare clinical safety flags")
    citations: List[ClinicalCitation] = Field(default_factory=list, description="Healthcare clinical citations")
    metadata: InvocationMetadata = Field(..., description="Invocation metadata")
    message: Optional[str] = Field(None, description="Optional warning or info message from the platform")
    plugins: Optional[Dict[str, Any]] = Field(None, description="Plugin response metadata (e.g. SONA note_id, suggestions)")
    downstream_context: Optional[Dict[str, Any]] = None
    
    # Convenience properties
    @property
    def raw_output(self) -> str:
        """Alias for content."""
        return self.content
    
    @property
    def latency_ms(self) -> Optional[int]:
        """Convenience accessor for latency."""
        return self.metadata.latency_ms
    
    @property
    def tokens_used(self) -> Optional[int]:
        """Convenience accessor for tokens."""
        return self.metadata.tokens_used


class StreamChunk(BaseModel):
    """A chunk from a streaming response."""
    
    content: str = Field(..., description="Chunk content")
    is_final: bool = Field(False, description="Whether this is the final chunk")
    metadata: Optional[Dict[str, Any]] = Field(None, description="Chunk metadata")


================================================================================
File: hikigai/appsdk/sona.py
================================================================================

"""
SONAClient — Convenience wrapper for SONA personalization features.

Accessible via ``app_client.sona``. Provides methods for edit tracking,
preference management, suggestion handling, and analytics.

Calls go directly to the SONA service (default http://localhost:8002),
NOT through the Hikigai backend.
"""

import logging
import os
from typing import Any, Dict, List, Optional

import httpx

logger = logging.getLogger(__name__)


class SONAClient:
    """Client for SONA edit tracking, preferences, and suggestions.

    Connects directly to the SONA service via HTTP.
    """

    def __init__(
        self,
        sona_url: Optional[str] = None,
        sona_api_key: Optional[str] = None,
        timeout: float = 15.0,
    ):
        self._base_url = (
            sona_url
            or os.environ.get("SONA_SERVICE_URL")
            or "http://localhost:8002"
        ).rstrip("/")
        self._api_key = (
            sona_api_key
            or os.environ.get("SONA_API_KEY")
            or "sona-change-this-in-production"
        )
        self._client = httpx.Client(
            base_url=self._base_url,
            headers={"X-SONA-API-Key": self._api_key},
            timeout=timeout,
        )

    def _get(self, path: str, params: Optional[Dict] = None) -> Dict[str, Any]:
        resp = self._client.get(path, params=params)
        resp.raise_for_status()
        return resp.json()

    def _post(self, path: str, json: Optional[Dict] = None) -> Dict[str, Any]:
        resp = self._client.post(path, json=json or {})
        resp.raise_for_status()
        return resp.json()

    # ── Edit tracking ───────────────────────────────────────────────

    def submit_edit(
        self,
        output_id: str,
        final_text: str,
        edit_duration_seconds: Optional[int] = None,
        user_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        """
        Submit a user's edited version of an agent output.

        Args:
            output_id: The ``output_id`` returned in ``response.plugins["sona"]``.
            final_text: The text after the user edited it.
            edit_duration_seconds: Optional time the user spent editing.
            user_id: End-user identifier. Required if not inferrable.

        Returns:
            Dict with ``output_id``, ``status``, ``edit_count``, ``was_edited``,
            and ``diff_summary``.
        """
        payload: Dict[str, Any] = {"final_text": final_text}
        if user_id:
            payload["user_id"] = user_id
        if edit_duration_seconds is not None:
            payload["edit_duration_seconds"] = edit_duration_seconds

        return self._post(f"/api/v1/edits/{output_id}/submit", json=payload)

    def approve_output(self, output_id: str, user_id: Optional[str] = None) -> Dict[str, Any]:
        """
        Mark an output as approved without edits.

        Args:
            output_id: The ``output_id`` from the invoke response.
            user_id: End-user identifier.
        """
        payload: Dict[str, Any] = {}
        if user_id:
            payload["user_id"] = user_id
        return self._post(f"/api/v1/edits/{output_id}/approve", json=payload)

    # ── Preferences ─────────────────────────────────────────────────

    def get_preferences(self, user_id: str, agent_id: Optional[str] = None) -> Dict[str, Any]:
        """Get a user's personalization preferences."""
        params: Dict[str, str] = {}
        if agent_id:
            params["agent_id"] = agent_id
        return self._get(f"/api/v1/preferences/{user_id}", params=params)

    def update_preferences(self, user_id: str, agent_id: str, **kwargs) -> Dict[str, Any]:
        """
        Update preferences. Keyword args can include ``output_length``,
        ``abbreviations``, ``custom_phrases``, ``section_order``, etc.
        """
        payload = {"user_id": user_id, "agent_id": agent_id, **kwargs}
        return self._post("/api/v1/preferences", json=payload)

    # ── Suggestions ─────────────────────────────────────────────────

    def get_suggestions(self, user_id: str, agent_id: Optional[str] = None) -> Dict[str, Any]:
        """Get pending pattern suggestions for a user."""
        params: Dict[str, str] = {}
        if agent_id:
            params["agent_id"] = agent_id
        return self._get(f"/api/v1/suggestions/{user_id}", params=params)

    def respond_to_suggestion(self, pattern_id: str, response: str, user_id: Optional[str] = None) -> Dict[str, Any]:
        """
        Accept or reject a pattern suggestion.

        Args:
            pattern_id: The ``pattern_id`` from suggestions.
            response: ``"accepted"`` or ``"rejected"``.
            user_id: End-user identifier.
        """
        payload: Dict[str, Any] = {"response": response}
        if user_id:
            payload["user_id"] = user_id
        return self._post(f"/api/v1/suggestions/{pattern_id}/respond", json=payload)

    # ── Analytics ───────────────────────────────────────────────────

    def get_analytics(self, user_id: str, agent_id: Optional[str] = None) -> Dict[str, Any]:
        """Get edit analytics for a user."""
        params: Dict[str, str] = {}
        if agent_id:
            params["agent_id"] = agent_id
        return self._get(f"/api/v1/analytics/{user_id}", params=params)


================================================================================
File: hikigai/appsdk/events.py
================================================================================

"""
EventsClient — platform event bus surface for applications.

Accessible via ``app_client.events``. Two ways to receive platform events
(``job.*``, ``agent.*``, ``invocation.*``, ``storage.*``, ``stream.*``):

**Webhooks** — the platform POSTs each matching event to an HTTPS endpoint
you own, signed with HMAC-SHA256::

    sub = client.events.create_webhook(
        url="https://app.example.com/hooks/hikigai",
        event_types=["job.*", "agent.deployed"],
    )
    print(sub.secret)   # shown once — store it in your secret manager

Then, in your endpoint, verify before trusting the body::

    from hikigai.appsdk import parse_webhook_event

    event = parse_webhook_event(
        secret=os.environ["HIKIGAI_WEBHOOK_SECRET"],
        signature_header=request.headers["X-Hikigai-Signature"],
        body=request.get_data(as_text=True),   # RAW body, before JSON parsing
    )
    if event["type"] == "job.completed":
        ...

**WebSocket** — a live subscription for processes that are already running
(dashboards, workers)::

    async with client.events.stream(patterns=["job.*"]) as stream:
        async for event in stream:
            print(event["type"], event["data"])

Delivery semantics differ and that is deliberate: webhooks are
at-least-once (retried with backoff, so dedupe on ``event["id"]``), the
WebSocket is at-most-once (dropped under backpressure, so never use it as
the only path for state you cannot lose).
"""

from __future__ import annotations

import hashlib
import hmac
import json
import logging
import time
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

from hikigai.appsdk.streaming.events import EventStream

logger = logging.getLogger(__name__)

#: Header carrying the delivery signature (``t=<unix-ts>,v1=<hex digest>``).
SIGNATURE_HEADER = "X-Hikigai-Signature"

#: Default replay window for signature timestamps, in seconds.
DEFAULT_TOLERANCE_SECONDS = 300


class SignatureVerificationError(Exception):
    """Raised when a webhook delivery fails signature verification.

    Treat this as a hard reject: return 400 and do not process the body.
    """


@dataclass
class WebhookSubscription:
    """A registration to receive events at an HTTPS endpoint.

    ``secret`` is populated only by :meth:`EventsClient.create_webhook` and
    :meth:`EventsClient.rotate_secret` — the platform stores it encrypted
    and never returns it again.
    """

    id: str
    project_id: str
    url: str
    event_types: List[str] = field(default_factory=list)
    description: Optional[str] = None
    is_active: bool = True
    created_at: Optional[str] = None
    updated_at: Optional[str] = None
    secret: Optional[str] = None

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "WebhookSubscription":
        return cls(
            id=d["id"],
            project_id=d.get("project_id", ""),
            url=d.get("url", ""),
            event_types=list(d.get("event_types") or []),
            description=d.get("description"),
            is_active=bool(d.get("is_active", True)),
            created_at=d.get("created_at"),
            updated_at=d.get("updated_at"),
            secret=d.get("secret"),
        )


@dataclass
class WebhookDelivery:
    """One attempt-tracked delivery of one event to one subscription.

    ``status`` is ``pending`` → ``success`` | ``failed`` (retry scheduled)
    | ``dead`` (retries exhausted; redeliver manually).
    """

    id: str
    subscription_id: str
    event_id: str
    event_type: str
    status: str
    attempts: int = 0
    response_status: Optional[int] = None
    last_error: Optional[str] = None
    next_attempt_at: Optional[str] = None
    created_at: Optional[str] = None
    delivered_at: Optional[str] = None

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "WebhookDelivery":
        return cls(
            id=d["id"],
            subscription_id=d.get("subscription_id", ""),
            event_id=d.get("event_id", ""),
            event_type=d.get("event_type", ""),
            status=d.get("status", "pending"),
            attempts=int(d.get("attempts") or 0),
            response_status=d.get("response_status"),
            last_error=d.get("last_error"),
            next_attempt_at=d.get("next_attempt_at"),
            created_at=d.get("created_at"),
            delivered_at=d.get("delivered_at"),
        )


# ---------------------------------------------------------------------------
# Signature verification (no client or network needed — safe to import
# standalone inside a request handler)
# ---------------------------------------------------------------------------

def verify_webhook_signature(
    secret: str,
    signature_header: str,
    body: str,
    tolerance_seconds: int = DEFAULT_TOLERANCE_SECONDS,
) -> bool:
    """Return True when ``signature_header`` authenticates ``body``.

    Args:
        secret: The subscription's signing secret.
        signature_header: The raw ``X-Hikigai-Signature`` header value.
        body: The RAW request body, byte-for-byte as received. Re-serializing
            parsed JSON will not reproduce the signed bytes.
        tolerance_seconds: Reject signatures whose timestamp is further from
            now than this, which bounds replay of a captured delivery.

    Never short-circuits on a malformed header, and compares digests in
    constant time.
    """
    if not secret or not signature_header:
        return False
    try:
        parts = dict(item.split("=", 1) for item in signature_header.split(","))
        timestamp = int(parts["t"])
        expected = parts["v1"]
    except (ValueError, KeyError):
        return False
    if abs(time.time() - timestamp) > tolerance_seconds:
        return False
    digest = hmac.new(
        secret.encode(), f"{timestamp}.{body}".encode(), hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(digest, expected)


def parse_webhook_event(
    secret: str,
    signature_header: str,
    body: str,
    tolerance_seconds: int = DEFAULT_TOLERANCE_SECONDS,
) -> Dict[str, Any]:
    """Verify a delivery and return the parsed CloudEvents envelope.

    Raises:
        SignatureVerificationError: The signature is missing, stale, or
            does not match — or the verified body is not a JSON object.

    The returned envelope has ``id``, ``type``, ``source``, ``time``,
    ``projectid`` and ``data``. Dedupe on ``id``: webhook delivery is
    at-least-once, so the same event id can arrive more than once.
    """
    if not verify_webhook_signature(secret, signature_header, body, tolerance_seconds):
        raise SignatureVerificationError(
            "Webhook signature verification failed (bad, missing, or stale signature)"
        )
    try:
        envelope = json.loads(body)
    except ValueError as exc:
        raise SignatureVerificationError(f"Webhook body is not valid JSON: {exc}") from exc
    if not isinstance(envelope, dict):
        raise SignatureVerificationError("Webhook body is not a JSON object")
    return envelope


class EventsClient:
    """Client for tenant webhooks and the project events WebSocket."""

    def __init__(self, client: Any):
        self._client = client
        self._api = client.api

    # -- webhook subscriptions ---------------------------------------------

    def create_webhook(
        self,
        *,
        url: str,
        event_types: List[str],
        description: Optional[str] = None,
    ) -> WebhookSubscription:
        """Register an endpoint. The returned ``secret`` is shown only here.

        ``event_types`` are glob patterns over event types — ``["job.*"]``,
        ``["agent.deployed", "invocation.completed"]``, or ``["*"]`` for
        everything visible to your project.
        """
        result = self._api.post(
            "/api/v1/webhooks",
            json={
                "url": url,
                "event_types": list(event_types),
                **({"description": description} if description is not None else {}),
            },
        )
        return WebhookSubscription.from_dict(result)

    def list_webhooks(self) -> List[WebhookSubscription]:
        """All webhook subscriptions for the current project."""
        result = self._api.get("/api/v1/webhooks")
        return [WebhookSubscription.from_dict(w) for w in result.get("webhooks", [])]

    def get_webhook(self, webhook_id: str) -> WebhookSubscription:
        return WebhookSubscription.from_dict(self._api.get(f"/api/v1/webhooks/{webhook_id}"))

    def update_webhook(
        self,
        webhook_id: str,
        *,
        url: Optional[str] = None,
        event_types: Optional[List[str]] = None,
        description: Optional[str] = None,
        is_active: Optional[bool] = None,
    ) -> WebhookSubscription:
        """Patch a subscription. Omitted fields are left unchanged.

        Pause deliveries without losing the subscription (and its secret)
        with ``is_active=False``.
        """
        body: Dict[str, Any] = {}
        if url is not None:
            body["url"] = url
        if event_types is not None:
            body["event_types"] = list(event_types)
        if description is not None:
            body["description"] = description
        if is_active is not None:
            body["is_active"] = is_active
        result = self._api.request("PATCH", f"/api/v1/webhooks/{webhook_id}", json=body)
        return WebhookSubscription.from_dict(result)

    def delete_webhook(self, webhook_id: str) -> Dict[str, Any]:
        return self._api.delete(f"/api/v1/webhooks/{webhook_id}")

    def rotate_secret(self, webhook_id: str) -> str:
        """Issue a new signing secret and return it (shown only here).

        Deliveries signed with the old secret stop verifying immediately,
        so roll the new value into your endpoint promptly.
        """
        result = self._api.post(f"/api/v1/webhooks/{webhook_id}/rotate-secret")
        return result["secret"]

    def test_webhook(self, webhook_id: str) -> Dict[str, Any]:
        """Send a synthetic ``webhook.test`` event to the endpoint now.

        Returns ``{"delivered": bool, "delivery": {...}}`` — the fastest way
        to confirm reachability and signature verification end to end.
        """
        return self._api.post(f"/api/v1/webhooks/{webhook_id}/test")

    # -- delivery status ----------------------------------------------------

    def list_deliveries(
        self,
        webhook_id: str,
        *,
        status: Optional[str] = None,
        limit: int = 50,
    ) -> List[WebhookDelivery]:
        """Recent deliveries, newest first. Filter with ``status="dead"``."""
        params: Dict[str, Any] = {"limit": limit}
        if status is not None:
            params["status"] = status
        result = self._api.get(f"/api/v1/webhooks/{webhook_id}/deliveries", params=params)
        return [WebhookDelivery.from_dict(d) for d in result.get("deliveries", [])]

    def redeliver(self, webhook_id: str, delivery_id: str) -> Dict[str, Any]:
        """Retry a failed or dead delivery immediately."""
        return self._api.post(
            f"/api/v1/webhooks/{webhook_id}/deliveries/{delivery_id}/redeliver"
        )

    # -- signature verification (also importable standalone) ----------------

    @staticmethod
    def verify_signature(
        secret: str,
        signature_header: str,
        body: str,
        tolerance_seconds: int = DEFAULT_TOLERANCE_SECONDS,
    ) -> bool:
        """See :func:`verify_webhook_signature`."""
        return verify_webhook_signature(secret, signature_header, body, tolerance_seconds)

    @staticmethod
    def parse_event(
        secret: str,
        signature_header: str,
        body: str,
        tolerance_seconds: int = DEFAULT_TOLERANCE_SECONDS,
    ) -> Dict[str, Any]:
        """See :func:`parse_webhook_event`."""
        return parse_webhook_event(secret, signature_header, body, tolerance_seconds)

    # -- live subscription --------------------------------------------------

    def stream(self, patterns: Optional[List[str]] = None) -> EventStream:
        """Open a live event subscription over WebSocket.

        The returned :class:`~hikigai.appsdk.streaming.events.EventStream`
        is an async context manager; ``patterns`` defaults to ``["*"]``.
        Requires the ``websockets`` extra (``pip install hikigai-appsdk[live]``).
        """
        return EventStream(
            self._api.base_url,
            api_key=getattr(self._client, "api_key", None),
            project_id=getattr(self._client, "project_id", None),
            patterns=patterns,
        )


================================================================================
File: hikigai/appsdk/cloud.py
================================================================================

"""
CloudClient — multi-cloud deployment catalog and credentials.

Accessible via ``app_client.cloud``. Everything the console shows in its
deployment UI is available here, from the same endpoints, so an app can
pick a provider, region, and service configuration programmatically::

    catalog = client.cloud.catalog(workload="app")
    for service in catalog.services:
        print(service.id, service.display_name, service.workloads)

Nothing about providers, regions, or services is hardcoded in this
module. The platform serves them from ``specs/cloud/`` plus any
operator-registered services, so a target added on the backend is
selectable here immediately — no SDK release required.

**Validate before deploying.** ``validate_target`` is a dry run: it
checks the workload, region, zones, and configuration against the
service's schema and returns the resolved configuration with defaults
applied. Far better than discovering a bad region halfway through a
deployment with resources already created::

    result = client.cloud.validate_target(
        workload="app",
        service_id="gcp-cloud-run",
        region="europe-west1",
        config={"cpu": "2", "memory": "1Gi"},
    )
    if not result.valid:
        raise SystemExit(result.error)

**Credentials are write-only.** ``create_credential`` sends material to
the platform, which stores it in a secrets manager; no method returns a
stored secret, because no endpoint does. Reads give a masked hint.
"""

from __future__ import annotations

import logging
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

logger = logging.getLogger(__name__)


@dataclass
class CloudRegion:
    id: str
    display_name: str
    zones: List[str] = field(default_factory=list)

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "CloudRegion":
        return cls(
            id=d["id"],
            display_name=d.get("display_name", d["id"]),
            zones=list(d.get("zones") or []),
        )


@dataclass
class CloudProvider:
    id: str
    display_name: str
    default_region: str
    credential_kinds: List[str] = field(default_factory=list)
    regions: List[CloudRegion] = field(default_factory=list)

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "CloudProvider":
        return cls(
            id=d["id"],
            display_name=d.get("display_name", d["id"]),
            default_region=d.get("default_region", ""),
            credential_kinds=list(d.get("credential_kinds") or []),
            regions=[CloudRegion.from_dict(r) for r in d.get("regions") or []],
        )


@dataclass
class CloudService:
    """A deployment target and everything configurable about it."""

    id: str
    provider: str
    display_name: str
    description: str
    workloads: List[str]
    supports_availability_zones: bool
    #: JSON Schema for this service's settings — the same schema the API
    #: validates against, so it is the authoritative field list.
    config_schema: Dict[str, Any]
    regions: List[CloudRegion] = field(default_factory=list)
    deprecated: bool = False
    user_defined: bool = False
    #: False when a spec exists but no deployer implements it yet.
    deployer_available: bool = True

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "CloudService":
        return cls(
            id=d["id"],
            provider=d.get("provider", ""),
            display_name=d.get("display_name", d["id"]),
            description=d.get("description", ""),
            workloads=list(d.get("workloads") or []),
            supports_availability_zones=bool(d.get("supports_availability_zones", False)),
            config_schema=d.get("config_schema") or {},
            regions=[
                CloudRegion.from_dict(r)
                for r in (d.get("resolved_regions") or [])
            ],
            deprecated=bool(d.get("deprecated", False)),
            user_defined=bool(d.get("user_defined", False)),
            deployer_available=bool(d.get("deployer_available", True)),
        )

    def defaults(self) -> Dict[str, Any]:
        """The configuration this service deploys with when unspecified.

        Read from the schema rather than duplicated here, so it stays
        correct as specs change.
        """
        properties = (self.config_schema or {}).get("properties") or {}
        return {
            name: prop["default"]
            for name, prop in properties.items()
            if "default" in prop
        }

    def required_fields(self) -> List[str]:
        """Fields with no default that a caller must supply."""
        schema = self.config_schema or {}
        properties = schema.get("properties") or {}
        return [
            name
            for name in (schema.get("required") or [])
            if "default" not in (properties.get(name) or {})
        ]


@dataclass
class CloudCatalog:
    providers: List[CloudProvider]
    services: List[CloudService]
    workloads: List[str]
    byoc_enabled: bool = True
    credential_kinds: List[str] = field(default_factory=list)

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "CloudCatalog":
        return cls(
            providers=[CloudProvider.from_dict(p) for p in d.get("providers") or []],
            services=[CloudService.from_dict(s) for s in d.get("services") or []],
            workloads=list(d.get("workloads") or []),
            byoc_enabled=bool(d.get("byoc_enabled", True)),
            credential_kinds=list(d.get("credential_kinds") or []),
        )

    def service(self, service_id: str) -> Optional[CloudService]:
        return next((s for s in self.services if s.id == service_id), None)

    def services_for(self, workload: str) -> List[CloudService]:
        return [s for s in self.services if workload in s.workloads]


@dataclass
class TargetValidation:
    """Result of a dry-run target check."""

    valid: bool
    error: Optional[str] = None
    #: Present when valid — the selection with schema defaults applied.
    target: Optional[Dict[str, Any]] = None

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "TargetValidation":
        return cls(valid=bool(d.get("valid")), error=d.get("error"), target=d.get("target"))


@dataclass
class CloudCredentialSummary:
    """A stored credential. Never carries the material itself."""

    id: str
    name: str
    provider: str
    kind: str
    hint: Optional[str]
    is_default: bool
    is_active: bool
    last_validated_at: Optional[str] = None
    last_validation_error: Optional[str] = None

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "CloudCredentialSummary":
        return cls(
            id=d["id"],
            name=d.get("name", ""),
            provider=d.get("provider", ""),
            kind=d.get("kind", ""),
            hint=d.get("hint"),
            is_default=bool(d.get("is_default", False)),
            is_active=bool(d.get("is_active", True)),
            last_validated_at=d.get("last_validated_at"),
            last_validation_error=d.get("last_validation_error"),
        )


class CloudClient:
    """Client for the deployment catalog and cloud credentials."""

    def __init__(self, client: Any):
        self._api = client.api

    # -- catalog -------------------------------------------------------

    def catalog(self, workload: Optional[str] = None) -> CloudCatalog:
        """Providers, services, regions, zones, and config schemas.

        Pass ``workload`` (``app``, ``mcp``, or ``agent``) to see only
        the services that accept it.
        """
        params = {"workload": workload} if workload else None
        return CloudCatalog.from_dict(self._api.get("/api/v1/cloud/catalog", params=params))

    def providers(self) -> List[CloudProvider]:
        result = self._api.get("/api/v1/cloud/providers")
        return [CloudProvider.from_dict(p) for p in result.get("providers", [])]

    def services(
        self, *, provider: Optional[str] = None, workload: Optional[str] = None
    ) -> List[CloudService]:
        params: Dict[str, Any] = {}
        if provider:
            params["provider"] = provider
        if workload:
            params["workload"] = workload
        result = self._api.get("/api/v1/cloud/services", params=params or None)
        return [CloudService.from_dict(s) for s in result.get("services", [])]

    def service(self, service_id: str) -> CloudService:
        return CloudService.from_dict(self._api.get(f"/api/v1/cloud/services/{service_id}"))

    def validate_target(
        self,
        *,
        workload: str,
        service_id: str,
        region: Optional[str] = None,
        availability_zones: Optional[List[str]] = None,
        config: Optional[Dict[str, Any]] = None,
    ) -> TargetValidation:
        """Dry-run a deployment selection.

        Returns ``valid=False`` with a message rather than raising — an
        invalid selection is something to show a user, not an exception.
        """
        body: Dict[str, Any] = {"workload": workload, "service_id": service_id}
        if region is not None:
            body["region"] = region
        if availability_zones is not None:
            body["availability_zones"] = list(availability_zones)
        if config is not None:
            body["config"] = config
        return TargetValidation.from_dict(
            self._api.post("/api/v1/cloud/targets/validate", json=body)
        )

    # -- credentials ---------------------------------------------------

    def list_credentials(self) -> List[CloudCredentialSummary]:
        result = self._api.get("/api/v1/cloud/credentials")
        return [CloudCredentialSummary.from_dict(c) for c in result.get("credentials", [])]

    def create_credential(
        self,
        *,
        name: str,
        kind: str,
        payload: Dict[str, str],
        is_default: bool = False,
        validate_now: bool = True,
    ) -> Dict[str, Any]:
        """Store a cloud credential for this project.

        ``kind`` is one of ``aws_access_key``, ``aws_assume_role``, or
        ``gcp_service_account``; ``payload`` carries that kind's fields.
        The material goes to a secrets manager and is never returned by
        any read — only a masked hint is.

        With ``validate_now``, the platform proves the credential
        authenticates and returns the outcome under ``validation``. A
        failure still stores the credential so you can fix IAM and
        re-validate rather than re-sending everything.
        """
        return self._api.post(
            "/api/v1/cloud/credentials",
            json={
                "name": name,
                "kind": kind,
                "payload": payload,
                "is_default": is_default,
                "validate_now": validate_now,
            },
        )

    def rotate_credential(
        self, credential_id: str, payload: Dict[str, str], *, validate_now: bool = True
    ) -> Dict[str, Any]:
        """Replace the material, keeping the credential id.

        Deployments reference the credential, so rotation does not
        require touching any of them.
        """
        return self._api.post(
            f"/api/v1/cloud/credentials/{credential_id}/rotate",
            json={"payload": payload, "validate_now": validate_now},
        )

    def validate_credential(self, credential_id: str) -> Dict[str, Any]:
        """Re-check a stored credential against its provider."""
        return self._api.post(f"/api/v1/cloud/credentials/{credential_id}/validate")

    def update_credential(
        self,
        credential_id: str,
        *,
        name: Optional[str] = None,
        is_default: Optional[bool] = None,
        is_active: Optional[bool] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {}
        if name is not None:
            body["name"] = name
        if is_default is not None:
            body["is_default"] = is_default
        if is_active is not None:
            body["is_active"] = is_active
        return self._api.request(
            "PATCH", f"/api/v1/cloud/credentials/{credential_id}", json=body
        )

    def delete_credential(self, credential_id: str) -> Dict[str, Any]:
        return self._api.delete(f"/api/v1/cloud/credentials/{credential_id}")

    # -- operator-registered services ----------------------------------

    def register_service(
        self,
        *,
        service_id: str,
        provider: str,
        display_name: str,
        workloads: List[str],
        config_schema: Optional[Dict[str, Any]] = None,
        regions: Optional[List[str]] = None,
        supports_availability_zones: bool = False,
        description: Optional[str] = None,
        docs_url: Optional[str] = None,
    ) -> Dict[str, Any]:
        """Register a deployment target without a platform release.

        The definition is merged over the built-in specs by ``service_id``
        — registering an existing id overrides it, which is the supported
        way to pin a narrower region list or a stricter schema.
        """
        return self._api.request(
            "PUT",
            f"/api/v1/cloud/service-definitions/{service_id}",
            json={
                "service_id": service_id,
                "provider": provider,
                "display_name": display_name,
                "workloads": list(workloads),
                "config_schema": config_schema or {},
                "regions": regions or [],
                "supports_availability_zones": supports_availability_zones,
                "description": description,
                "docs_url": docs_url,
            },
        )

    def delete_service(self, service_id: str) -> Dict[str, Any]:
        return self._api.delete(f"/api/v1/cloud/service-definitions/{service_id}")
