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

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

Invoke deployed agents in your applications.
"""

__version__ = "0.2.3"

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.streaming.speech import (
    SpeechClient,
    SpeechError,
    SpeechEvent,
    SpeechSession,
)
from hikigai.appsdk.audio import (
    MODEL_INPUT_SAMPLE_RATE,
    MODEL_OUTPUT_SAMPLE_RATE,
    RECOMMENDED_CHUNK_MS,
    chunk_audio,
    decode_alaw,
    decode_mulaw,
    from_model_audio,
    pcm_duration_seconds,
    pcm_to_wav,
    to_model_audio,
)
from hikigai.appsdk.context import SessionContext
from hikigai.appsdk.identity import IdentityClient, VerifiedUser
from hikigai.appsdk.identity_verification import (
    EndUserVerificationStatus,
    IdentityVerificationClient,
    PatientVerificationStatus,
    VerificationSession,
)
from hikigai.appsdk.cloud import (
    CloudCatalog,
    CloudClient,
    CloudCredentialSummary,
    CloudProvider,
    CloudRegion,
    CloudService,
    TargetValidation,
)
from hikigai.appsdk.even_g2 import (
    EVEN_G2_DISPLAY,
    EVEN_G2_MAX_TEXT_CHARS,
    EvenG2ManifestValidation,
    even_g2_dev_mode_url,
    normalize_for_glasses,
    paginate_for_glasses,
    validate_even_g2_manifest,
)
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,
    PlanLimitError,
)

__all__ = [
    "__version__",
    # Client
    "AppClient",
    # Even Realities G2 glasses helpers
    "EVEN_G2_DISPLAY",
    "EVEN_G2_MAX_TEXT_CHARS",
    "EvenG2ManifestValidation",
    "even_g2_dev_mode_url",
    "normalize_for_glasses",
    "paginate_for_glasses",
    "validate_even_g2_manifest",
    # 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",
    # Speech — real-time translate & transcribe
    "SpeechClient",
    "SpeechSession",
    "SpeechEvent",
    "SpeechError",
    "to_model_audio",
    "from_model_audio",
    "chunk_audio",
    "decode_mulaw",
    "decode_alaw",
    "pcm_to_wav",
    "pcm_duration_seconds",
    "MODEL_INPUT_SAMPLE_RATE",
    "MODEL_OUTPUT_SAMPLE_RATE",
    "RECOMMENDED_CHUNK_MS",
    # Session context store
    "SessionContext",
    # End-user identity
    "IdentityClient",
    "VerifiedUser",
    # Patient identity verification (CLEAR / ID.me)
    "IdentityVerificationClient",
    "VerificationSession",
    "PatientVerificationStatus",
    "EndUserVerificationStatus",
    # Exceptions
    "HikigaiError",
    "AuthenticationError",
    "PermissionDeniedError",
    "RateLimitError",
    "AgentNotFoundError",
    "InvocationError",
    "ConfigurationError",
    "ConflictError",
    "ValidationError",
    "PlanLimitError",
]


================================================================================
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.appsdk.identity_verification import IdentityVerificationClient

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

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

    @property
    def speech(self):
        """Real-time speech translation and transcription.

        Example::

            async with await client.speech.translate(target_language="es") as s:
                await s.send_audio(pcm)
                async for event in s:
                    if event.type == "audio":
                        speaker.write(event.audio)
        """
        if not hasattr(self, "_speech"):
            from hikigai.appsdk.streaming.speech import SpeechClient
            self._speech = SpeechClient(self)
        return self._speech
    
    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,
        attachments: Optional[List[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
            attachments: Optional multimodal files (image, PDF, or audio such as
                mp3/wav) when the agent's ``input_modality`` includes that kind
                (``data_base64`` + ``mime_type``, ``url``, or ``storage_object_id``)
            
        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,
            attachments=attachments,
        )
    
    def live(
        self,
        agent_id: str,
        session_id: Optional[str] = None,
        user_id: str = "sdk-user",
        context: Optional[Dict[str, Any]] = None,
        connectors: Optional[Dict[str, Dict[str, Any]]] = None,
    ) -> StreamSession:
        """
        Open a bidirectional live session with a deployed agent.

        Uses POST /live/session for auth, then a WebSocket to the minted
        ``wss_url``. Pass ``connectors`` (same shape as ``invoke``) to attach
        MCP tools on the live path.
        """
        sid = session_id or str(uuid.uuid4())
        return StreamSession(
            client=self,
            agent_id=agent_id,
            session_id=sid,
            user_id=user_id,
            context=context,
            connectors=connectors,
        )

    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 identity_verification(self) -> "IdentityVerificationClient":
        """Patient identity verification (identity proofing) via CLEAR / ID.me.

        Example:
            session = client.identity_verification.start("idme", patient_reference="p-42",
                                                         return_url="https://app/verify/done")
            redirect(session.verification_url)
        """
        if not hasattr(self, "_identity_verification"):
            from hikigai.appsdk.identity_verification import IdentityVerificationClient
            self._identity_verification = IdentityVerificationClient(self)
        return self._identity_verification

    @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,
        attachments: Optional[List[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
        if attachments:
            payload["attachments"] = attachments

        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"),
                artifacts=response.get("artifacts") or [],
            )
            
        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:
                self.api.raise_for_error_response(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 (PlanLimitError, RateLimitError):
            raise
        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/context.py
================================================================================

"""
SessionContext — client handle for the platform session context store (PRD 9.2).

A ``SessionContext`` is a thin, ergonomic wrapper over the session REST API. It
is bound to one ``session_id`` and lets an application store and read shared
workflow state across multiple agent invocations.

Obtain one from :class:`~hikigai.appsdk.client.AppClient`::

    sess = client.create_session(external_user_ref="dr-smith")
    sess.set("triage", triage_output)
    value = sess.get("triage")
    everything = sess.get_all()
    sess.update("counter", lambda v: (v or 0) + 1)   # OCC-safe
    sess.clear()

Concurrency: :meth:`set` supports optimistic concurrency via an ``If-Match``
etag. :meth:`update` performs a get → apply → set retry loop, transparently
recovering from :class:`~hikigai.core.exceptions.ConflictError`.
"""

from __future__ import annotations

import logging
from typing import Any, Callable, Dict, Optional, Tuple

from hikigai.core.exceptions import ConflictError

logger = logging.getLogger(__name__)


class SessionContext:
    """A handle to a single session's context store."""

    def __init__(self, client: Any, session_id: str, app_id: str = "_"):
        self._client = client
        self._api = client.api
        self._session_id = session_id
        self._app_id = app_id or "_"

    # -- properties ---------------------------------------------------------

    @property
    def session_id(self) -> str:
        """The opaque, server-minted session identifier."""
        return self._session_id

    @property
    def app_id(self) -> str:
        return self._app_id

    def _params(self) -> Dict[str, str]:
        return {"app_id": self._app_id}

    def _base(self) -> str:
        return f"/api/v1/sessions/{self._session_id}"

    # -- writes -------------------------------------------------------------

    def set(self, key: str, value: Any, *, if_match: Optional[str] = None) -> Dict[str, Any]:
        """Store ``value`` under ``key``.

        Args:
            key: Context key (1-128 chars of [a-zA-Z0-9._-]).
            value: Any JSON-serialisable value.
            if_match: Optional etag for optimistic concurrency. When provided
                and stale, a :class:`ConflictError` is raised.

        Returns:
            ``{"key", "etag", "expires_at"}``.

        Raises:
            ConflictError: stale ``if_match`` (409).
            HikigaiError subclasses for other HTTP errors.
        """
        headers = {"If-Match": str(if_match)} if if_match is not None else None
        result = self._api.put(
            f"{self._base()}/context/{key}",
            json={"value": value},
            params=self._params(),
            headers=headers,
        )
        logger.debug("[session %s] set %s -> etag %s", self._session_id, key, result.get("etag"))
        return result

    def delete(self, key: str) -> None:
        """Delete a single key (idempotent)."""
        self._api.delete(f"{self._base()}/context/{key}", params=self._params())

    def clear(self) -> None:
        """Delete the entire session (idempotent)."""
        self._api.delete(self._base(), params=self._params())

    def touch(self, ttl_seconds: int) -> Dict[str, Any]:
        """Extend (or shorten) the session TTL."""
        return self._api.request(
            "PATCH",
            self._base(),
            json={"ttl_seconds": ttl_seconds},
            params=self._params(),
        )

    # -- reads --------------------------------------------------------------

    def get(self, key: str, default: Any = None) -> Any:
        """Return the value for ``key``, or ``default`` if it is absent."""
        try:
            result = self._api.get(f"{self._base()}/context/{key}", params=self._params())
        except Exception as exc:  # 404 → default; re-raise everything else
            from hikigai.core.exceptions import HikigaiError

            if isinstance(exc, HikigaiError) and exc.status_code == 404:
                return default
            raise
        return result.get("value", default)

    def get_with_etag(self, key: str) -> Optional[Tuple[Any, str]]:
        """Return ``(value, etag)`` for ``key``, or ``None`` if absent."""
        try:
            result = self._api.get(f"{self._base()}/context/{key}", params=self._params())
        except Exception as exc:
            from hikigai.core.exceptions import HikigaiError

            if isinstance(exc, HikigaiError) and exc.status_code == 404:
                return None
            raise
        return result.get("value"), result.get("etag")

    def get_all(self) -> Dict[str, Any]:
        """Return ``{key: value}`` for the whole session context."""
        result = self._api.get(f"{self._base()}/context", params=self._params())
        context = result.get("context", {})
        return {k: entry.get("value") for k, entry in context.items()}

    def get_all_with_etags(self) -> Dict[str, Dict[str, Any]]:
        """Return the raw ``{key: {"value", "etag"}}`` context view."""
        result = self._api.get(f"{self._base()}/context", params=self._params())
        return result.get("context", {})

    def metadata(self) -> Dict[str, Any]:
        """Return session metadata (TTL, key count, user ref)."""
        return self._api.get(self._base(), params=self._params())

    # -- atomic update ------------------------------------------------------

    def update(
        self,
        key: str,
        fn: Callable[[Any], Any],
        *,
        default: Any = None,
        retries: int = 3,
    ) -> Dict[str, Any]:
        """Atomically read-modify-write a key with optimistic concurrency.

        Fetches the current value (and etag), applies ``fn``, and writes back
        with ``If-Match``. On a :class:`ConflictError` it refetches and retries
        up to ``retries`` times.

        Example::

            sess.update("count", lambda v: (v or 0) + 1)
        """
        last_error: Optional[ConflictError] = None
        for attempt in range(retries + 1):
            current = self.get_with_etag(key)
            if current is None:
                cur_value, etag = default, None
            else:
                cur_value, etag = current
            new_value = fn(cur_value)
            try:
                return self.set(key, new_value, if_match=etag)
            except ConflictError as exc:
                last_error = exc
                logger.debug(
                    "[session %s] update conflict on %s (attempt %d/%d)",
                    self._session_id, key, attempt + 1, retries + 1,
                )
                continue
        raise last_error  # exhausted retries

    def __repr__(self) -> str:  # pragma: no cover
        return f"SessionContext(session_id={self._session_id!r}, app_id={self._app_id!r})"


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

"""
IdentityClient — End-User Identity Service client (PRD 9.6).

Accessible via ``app_client.identity``. Lets an application verify clinicians
(QR badge / PIN / SSO) and manage end users + credentials through the platform.

Verification (the runtime hot path)::

    user = client.identity.verify_qr(scanned_payload)
    print(user.first_name, user.last_name, user.role)

    user = client.identity.verify_pin(user_id, "123456")

Management::

    eu = client.identity.create_user(first_name="Jane", last_name="Smith", role="doctor")
    issued = client.identity.issue_qr(eu["id"])
    print(issued["qr_payload"])          # encode this onto the badge (shown once)
    client.identity.issue_pin(eu["id"], "246810")

On verification failure the underlying client raises
:class:`~hikigai.core.exceptions.AuthenticationError` (HTTP 401); a PIN lockout
surfaces as :class:`~hikigai.core.exceptions.HikigaiError` with
``status_code == 423``.
"""

from __future__ import annotations

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

logger = logging.getLogger(__name__)


@dataclass
class VerifiedUser:
    """The result of a successful end-user verification."""

    end_user_id: str
    project_id: str
    first_name: str
    last_name: str
    role: Optional[str]
    method: str
    verified_at: str
    metadata: Optional[Dict[str, Any]] = None

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "VerifiedUser":
        return cls(
            end_user_id=d["end_user_id"],
            project_id=d["project_id"],
            first_name=d.get("first_name") or "",
            last_name=d.get("last_name") or "",
            role=d.get("role"),
            method=d.get("method", "unknown"),
            verified_at=d.get("verified_at", ""),
            metadata=d.get("metadata"),
        )


class IdentityClient:
    """Client for end-user identity verification and management."""

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

    # -- verification (runtime) --------------------------------------------

    def verify_qr(self, qr_payload: str) -> VerifiedUser:
        """Verify a scanned QR badge. Raises AuthenticationError on failure."""
        result = self._api.post("/api/v1/identity/verify-qr", json={"qr_payload": qr_payload})
        return VerifiedUser.from_dict(result)

    def verify_pin(self, user_id: str, pin: str) -> VerifiedUser:
        """Verify an end user's PIN. Raises AuthenticationError (401) on a wrong
        PIN, or HikigaiError(status_code=423) when locked out."""
        result = self._api.post(
            "/api/v1/identity/verify-pin", json={"user_id": user_id, "pin": pin}
        )
        return VerifiedUser.from_dict(result)

    def verify_sso(self, token: str) -> VerifiedUser:
        """Verify an SSO id token (when SSO is configured for the project)."""
        result = self._api.post("/api/v1/identity/verify-sso", json={"token": token})
        return VerifiedUser.from_dict(result)

    # -- management ---------------------------------------------------------

    def create_user(
        self,
        *,
        first_name: str,
        last_name: str,
        email: Optional[str] = None,
        role: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
        external_subject: Optional[str] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"first_name": first_name, "last_name": last_name}
        if email is not None:
            body["email"] = email
        if role is not None:
            body["role"] = role
        if metadata is not None:
            body["metadata"] = metadata
        if external_subject is not None:
            body["external_subject"] = external_subject
        return self._api.post("/api/v1/end-users", json=body)

    def get_user(self, user_id: str) -> Dict[str, Any]:
        return self._api.get(f"/api/v1/end-users/{user_id}")

    def list_users(self, *, limit: int = 50, offset: int = 0) -> List[Dict[str, Any]]:
        result = self._api.get("/api/v1/end-users", params={"limit": limit, "offset": offset})
        return result.get("end_users", [])

    def update_user(self, user_id: str, **fields: Any) -> Dict[str, Any]:
        """Update end-user profile fields (role, name, metadata, is_active).

        Email and other identity-bound fields are rejected by the API.
        """
        forbidden = {"email", "external_subject", "password", "user_sub", "app_id", "project_id", "id"}
        bad = forbidden.intersection(fields)
        if bad:
            raise ValueError(
                f"{', '.join(sorted(bad))} cannot be changed via update_user"
            )
        return self._api.request("PATCH", f"/api/v1/end-users/{user_id}", json=fields)

    def list_credentials(self, user_id: str) -> List[Dict[str, Any]]:
        result = self._api.get(f"/api/v1/end-users/{user_id}/credentials")
        return result.get("credentials", [])

    def issue_qr(self, user_id: str, *, expires_at: Optional[str] = None) -> Dict[str, Any]:
        """Issue a QR badge. The returned ``qr_payload`` is shown ONCE — render
        it onto the badge now."""
        body: Dict[str, Any] = {}
        if expires_at is not None:
            body["expires_at"] = expires_at
        return self._api.post(f"/api/v1/end-users/{user_id}/credentials/qr", json=body)

    def issue_pin(self, user_id: str, pin: str) -> Dict[str, Any]:
        """Set or replace an end user's PIN."""
        return self._api.post(f"/api/v1/end-users/{user_id}/credentials/pin", json={"pin": pin})

    def rotate_qr(self, user_id: str, credential_id: str) -> Dict[str, Any]:
        """Rotate a QR badge — issues a new one and revokes the old immediately."""
        return self._api.post(
            f"/api/v1/end-users/{user_id}/credentials/{credential_id}/rotate"
        )

    def revoke_credential(self, user_id: str, credential_id: str) -> None:
        self._api.delete(f"/api/v1/end-users/{user_id}/credentials/{credential_id}")

    # -- Cognito Auth-as-a-Service (per-app signup/login/tokens) ------------

    def enable_identity(
        self,
        app_id: str,
        *,
        mode: str = "managed",
        byo_role_arn: Optional[str] = None,
        enabled_methods: Optional[List[str]] = None,
    ) -> Dict[str, Any]:
        """Provision a Cognito user pool for an app (turns on User Management)."""
        body: Dict[str, Any] = {"mode": mode}
        if byo_role_arn is not None:
            body["byo_role_arn"] = byo_role_arn
        if enabled_methods is not None:
            body["enabled_methods"] = enabled_methods
        return self._api.post(f"/api/v1/identity/apps/{app_id}/enable", json=body)

    def get_identity_config(self, app_id: str) -> Dict[str, Any]:
        return self._api.get(f"/api/v1/identity/apps/{app_id}")

    def signup(
        self,
        app_id: str,
        email: str,
        password: str,
        *,
        first_name: Optional[str] = None,
        last_name: Optional[str] = None,
        role: Optional[str] = None,
        attributes: Optional[Dict[str, str]] = None,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        """Register a new end user in the app's pool."""
        body: Dict[str, Any] = {"app_id": app_id, "email": email, "password": password}
        if first_name is not None:
            body["first_name"] = first_name
        if last_name is not None:
            body["last_name"] = last_name
        if role is not None:
            body["role"] = role
        if attributes is not None:
            body["attributes"] = attributes
        if metadata is not None:
            body["metadata"] = metadata
        return self._api.post("/api/v1/identity/signup", json=body)

    def confirm(self, app_id: str, email: str, code: str) -> Dict[str, Any]:
        return self._api.post(
            "/api/v1/identity/confirm", json={"app_id": app_id, "email": email, "code": code}
        )

    def login(self, app_id: str, email: str, password: str) -> Dict[str, Any]:
        """Authenticate an end user. Returns OIDC tokens (or an MFA challenge).

        Raises AuthenticationError (401) on bad credentials.
        """
        return self._api.post(
            "/api/v1/identity/login",
            json={"app_id": app_id, "email": email, "password": password},
        )

    def refresh(self, app_id: str, refresh_token: str) -> Dict[str, Any]:
        """Exchange a refresh token for fresh access/id tokens."""
        return self._api.post(
            "/api/v1/identity/refresh",
            json={"app_id": app_id, "refresh_token": refresh_token},
        )

    def logout(self, app_id: str, email: str) -> Dict[str, Any]:
        """Global sign-out — invalidates the user's refresh tokens."""
        return self._api.post(
            "/api/v1/identity/logout", json={"app_id": app_id, "email": email}
        )

    def forgot_password(self, app_id: str, email: str) -> Dict[str, Any]:
        """Request a password-reset verification code by email."""
        return self._api.post(
            "/api/v1/identity/forgot-password", json={"app_id": app_id, "email": email}
        )

    def reset_password(
        self, app_id: str, email: str, code: str, new_password: str,
    ) -> Dict[str, Any]:
        """Set a new password using the emailed reset code."""
        return self._api.post(
            "/api/v1/identity/reset-password",
            json={
                "app_id": app_id, "email": email, "code": code, "new_password": new_password,
            },
        )

    def change_password(
        self, app_id: str, access_token: str, current_password: str, new_password: str,
    ) -> Dict[str, Any]:
        """Change password for a logged-in user."""
        return self._api.post(
            "/api/v1/identity/change-password",
            json={
                "app_id": app_id,
                "access_token": access_token,
                "current_password": current_password,
                "new_password": new_password,
            },
        )

    def list_app_users(self, app_id: str, *, limit: int = 50, offset: int = 0) -> List[Dict[str, Any]]:
        result = self._api.get(
            f"/api/v1/identity/apps/{app_id}/users", params={"limit": limit, "offset": offset}
        )
        return result.get("users", [])

    def delete_app_user(self, app_id: str, email: str) -> None:
        self._api.delete(f"/api/v1/identity/apps/{app_id}/users/{email}")

    # -- MFA (TOTP) ---------------------------------------------------------

    def mfa_associate(self, app_id: str, access_token: str) -> Dict[str, Any]:
        """Begin TOTP enrollment. Returns ``secret_code`` (render as an
        otpauth:// QR for the authenticator app)."""
        return self._api.post(
            "/api/v1/identity/mfa/associate", json={"app_id": app_id, "access_token": access_token}
        )

    def mfa_verify(self, app_id: str, access_token: str, code: str, *, device_name: Optional[str] = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {"app_id": app_id, "access_token": access_token, "code": code}
        if device_name is not None:
            body["device_name"] = device_name
        return self._api.post("/api/v1/identity/mfa/verify", json=body)

    def mfa_set_preference(self, app_id: str, access_token: str, *, enabled: bool = True) -> Dict[str, Any]:
        return self._api.post(
            "/api/v1/identity/mfa/set-preference",
            json={"app_id": app_id, "access_token": access_token, "enabled": enabled},
        )

    def mfa_respond(self, app_id: str, email: str, session: str, code: str) -> Dict[str, Any]:
        """Answer an MFA challenge raised during login → returns tokens."""
        return self._api.post(
            "/api/v1/identity/mfa/respond",
            json={"app_id": app_id, "email": email, "session": session, "code": code},
        )

    # -- QR login (badge -> tokens) ----------------------------------------

    def issue_qr_login(self, app_id: str, email: str) -> Dict[str, Any]:
        """Issue a QR-login badge for a user. The ``qr_payload`` is shown ONCE."""
        return self._api.post(f"/api/v1/identity/apps/{app_id}/users/{email}/qr-login")

    def qr_login(self, app_id: str, qr_payload: str) -> Dict[str, Any]:
        """Log in by scanning a QR badge → returns Cognito tokens."""
        return self._api.post(
            "/api/v1/identity/qr-login", json={"app_id": app_id, "qr_payload": qr_payload}
        )

    # -- BYO (bring-your-own Cognito) --------------------------------------

    def byo_setup(self) -> Dict[str, Any]:
        """Get the CloudFormation template + params for bring-your-own-Cognito."""
        return self._api.get("/api/v1/identity/byo/setup")

    def byo_test(self, role_arn: str, *, region: Optional[str] = None) -> Dict[str, Any]:
        """Verify a customer's cross-account IAM role works for Cognito."""
        body: Dict[str, Any] = {"role_arn": role_arn}
        if region is not None:
            body["region"] = region
        return self._api.post("/api/v1/identity/byo/test", json=body)

    # -- SSO / social federation -------------------------------------------

    def configure_sso(
        self,
        app_id: str,
        *,
        provider_name: str,
        provider_type: str,
        provider_details: Dict[str, str],
        callback_urls: List[str],
        attribute_mapping: Optional[Dict[str, str]] = None,
        scopes: Optional[List[str]] = None,
    ) -> Dict[str, Any]:
        """Configure a SAML/OIDC/social IdP on the app's pool (turns on SSO)."""
        body: Dict[str, Any] = {
            "provider_name": provider_name, "provider_type": provider_type,
            "provider_details": provider_details, "callback_urls": callback_urls,
        }
        if attribute_mapping is not None:
            body["attribute_mapping"] = attribute_mapping
        if scopes is not None:
            body["scopes"] = scopes
        return self._api.post(f"/api/v1/identity/apps/{app_id}/sso", json=body)

    def sso_authorize_url(
        self, app_id: str, redirect_uri: str, *, provider: Optional[str] = None,
        scopes: Optional[List[str]] = None,
    ) -> Dict[str, Any]:
        """Get the hosted-UI URL to redirect the end user to for SSO login."""
        body: Dict[str, Any] = {"app_id": app_id, "redirect_uri": redirect_uri}
        if provider is not None:
            body["provider"] = provider
        if scopes is not None:
            body["scopes"] = scopes
        return self._api.post("/api/v1/identity/sso/authorize-url", json=body)

    def sso_exchange(self, app_id: str, code: str, redirect_uri: str) -> Dict[str, Any]:
        """Exchange the SSO authorization code for tokens."""
        return self._api.post(
            "/api/v1/identity/sso/token",
            json={"app_id": app_id, "code": code, "redirect_uri": redirect_uri},
        )


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

"""
IdentityVerificationClient — patient identity proofing via CLEAR / ID.me.

Accessible via ``client.identity_verification``. Lets an application prove
WHO a patient is through a certified identity-proofing vendor without ever
talking to the vendor itself: the platform brokers the OpenID Connect flow,
keeps the verification record, and returns a vendor-neutral result.

Typical flow::

    iv = client.identity_verification

    # 1. Start: get a URL to send the patient to (their own browser).
    session = iv.start("idme", patient_reference="fhir-patient-42",
                       return_url="https://app.example.com/verify/done")
    redirect(session.verification_url)

    # 2. Later (webhook `identity.verification_completed`, or poll):
    session = iv.get(session.session_id)
    if session.status == "verified":
        print(session.assurance_level)      # e.g. "ial2/aal2"

    # 3. Confirm the verified identity matches the chart WITHOUT receiving
    #    the demographics.
    ok = iv.match(session.session_id, first_name="Jane", last_name="Doe",
                  date_of_birth="1990-01-02")["match"]

    # 4. Link it to the patient's app account once known, and re-check later.
    iv.link(session.session_id, end_user_id)
    iv.end_user_status(end_user_id).verified
    iv.patient_status("fhir-patient-42").verified

Provider configuration (usually done once from the console) is available via
``configure_provider`` / ``list_providers`` for automation.

Errors surface as the usual core exceptions: 400 -> ``ValidationError``,
401 -> ``AuthenticationError`` (vendor rejected the code exchange),
404 -> ``HikigaiError`` (provider not configured / session not found),
409 -> ``ConflictError`` (session no longer pending), 503 -> vendor down.
"""

from __future__ import annotations

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

logger = logging.getLogger(__name__)

_BASE = "/api/v1/identity-verification"
TERMINAL_STATUSES = ("verified", "failed", "expired", "cancelled")


@dataclass
class VerificationSession:
    """A patient identity verification session (vendor-neutral)."""

    session_id: str
    project_id: str
    provider: str
    status: str
    environment: str = "sandbox"
    app_id: Optional[str] = None
    end_user_id: Optional[str] = None
    patient_reference: Optional[str] = None
    requested_assurance: Optional[str] = None
    assurance_level: Optional[str] = None
    provider_subject: Optional[str] = None
    id_token_verified: Optional[bool] = None
    failure_reason: Optional[str] = None
    has_attributes: bool = False
    can_match_demographics: bool = False
    initiated_via: str = "api"
    return_url: Optional[str] = None
    metadata: Optional[Dict[str, Any]] = None
    linked_at: Optional[str] = None
    end_user: Optional[Dict[str, Any]] = None
    created_at: Optional[str] = None
    expires_at: Optional[str] = None
    completed_at: Optional[str] = None
    verification_url: Optional[str] = None
    attributes: Optional[Dict[str, Any]] = None
    raw: Dict[str, Any] = field(default_factory=dict, repr=False)

    @property
    def is_terminal(self) -> bool:
        return self.status in TERMINAL_STATUSES

    @property
    def verified(self) -> bool:
        return self.status == "verified"

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "VerificationSession":
        return cls(
            session_id=d["session_id"],
            project_id=d.get("project_id", ""),
            provider=d.get("provider", ""),
            status=d.get("status", "pending"),
            environment=d.get("environment", "sandbox"),
            app_id=d.get("app_id"),
            end_user_id=d.get("end_user_id"),
            patient_reference=d.get("patient_reference"),
            requested_assurance=d.get("requested_assurance"),
            assurance_level=d.get("assurance_level"),
            provider_subject=d.get("provider_subject"),
            id_token_verified=d.get("id_token_verified"),
            failure_reason=d.get("failure_reason"),
            has_attributes=bool(d.get("has_attributes", False)),
            can_match_demographics=bool(d.get("can_match_demographics", False)),
            initiated_via=d.get("initiated_via", "api"),
            return_url=d.get("return_url"),
            metadata=d.get("metadata"),
            linked_at=d.get("linked_at"),
            end_user=d.get("end_user"),
            created_at=d.get("created_at"),
            expires_at=d.get("expires_at"),
            completed_at=d.get("completed_at"),
            verification_url=d.get("verification_url"),
            attributes=d.get("attributes"),
            raw=d,
        )


@dataclass
class EndUserVerificationStatus:
    end_user_id: str
    verified: bool
    session: Optional["VerificationSession"] = None

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "EndUserVerificationStatus":
        session = d.get("session")
        return cls(
            end_user_id=d.get("end_user_id", ""),
            verified=bool(d.get("verified", False)),
            session=VerificationSession.from_dict(session) if session else None,
        )


@dataclass
class PatientVerificationStatus:
    patient_reference: str
    verified: bool
    session: Optional[VerificationSession] = None

    @classmethod
    def from_dict(cls, d: Dict[str, Any]) -> "PatientVerificationStatus":
        session = d.get("session")
        return cls(
            patient_reference=d.get("patient_reference", ""),
            verified=bool(d.get("verified", False)),
            session=VerificationSession.from_dict(session) if session else None,
        )


class IdentityVerificationClient:
    """Client for patient identity verification (CLEAR / ID.me)."""

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

    # -- providers -----------------------------------------------------------

    def list_providers(self) -> Dict[str, Any]:
        """Configured providers (secrets masked) plus the supported-vendor catalog."""
        return self._api.get(f"{_BASE}/providers")

    def configure_provider(
        self,
        provider: str,
        *,
        environment: Optional[str] = None,
        display_name: Optional[str] = None,
        client_id: Optional[str] = None,
        client_secret: Optional[str] = None,
        api_key: Optional[str] = None,
        scope: Optional[str] = None,
        allowed_return_urls: Optional[List[str]] = None,
        endpoint_overrides: Optional[Dict[str, str]] = None,
        attribute_policy: Optional[Dict[str, Any]] = None,
        enabled: Optional[bool] = None,
    ) -> Dict[str, Any]:
        """Create or update the vendor configuration for this project.

        ``client_secret`` / ``api_key`` are write-only: reads return a masked hint.
        """
        body: Dict[str, Any] = {}
        for key, value in (
            ("environment", environment), ("display_name", display_name),
            ("client_id", client_id), ("client_secret", client_secret), ("api_key", api_key),
            ("scope", scope), ("allowed_return_urls", allowed_return_urls),
            ("endpoint_overrides", endpoint_overrides), ("attribute_policy", attribute_policy),
            ("enabled", enabled),
        ):
            if value is not None:
                body[key] = value
        return self._api.put(f"{_BASE}/providers/{provider}", json=body)

    def delete_provider(self, provider: str) -> Dict[str, Any]:
        return self._api.delete(f"{_BASE}/providers/{provider}")

    def test_provider(self, provider: str) -> Dict[str, Any]:
        """Check the vendor's discovery document is reachable with the stored config."""
        return self._api.post(f"{_BASE}/providers/{provider}/test", json={})

    # -- sessions --------------------------------------------------------------

    def start(
        self,
        provider: str,
        *,
        patient_reference: Optional[str] = None,
        app_id: Optional[str] = None,
        end_user_id: Optional[str] = None,
        return_url: Optional[str] = None,
        redirect_uri: Optional[str] = None,
        requested_assurance: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> VerificationSession:
        """Start a verification. Send the patient to ``session.verification_url``."""
        body: Dict[str, Any] = {"provider": provider}
        for key, value in (
            ("patient_reference", patient_reference), ("app_id", app_id),
            ("end_user_id", end_user_id), ("return_url", return_url),
            ("redirect_uri", redirect_uri), ("requested_assurance", requested_assurance),
            ("metadata", metadata),
        ):
            if value is not None:
                body[key] = value
        return VerificationSession.from_dict(self._api.post(f"{_BASE}/sessions", json=body))

    def get(self, session_id: str, *, include_attributes: bool = False) -> VerificationSession:
        params = {"include_attributes": "true"} if include_attributes else None
        return VerificationSession.from_dict(self._api.get(f"{_BASE}/sessions/{session_id}", params=params))

    def list(
        self,
        *,
        app_id: Optional[str] = None,
        patient_reference: Optional[str] = None,
        end_user_id: Optional[str] = None,
        status: Optional[str] = None,
        provider: Optional[str] = None,
        limit: int = 50,
        offset: int = 0,
    ) -> List[VerificationSession]:
        params: Dict[str, Any] = {"limit": limit, "offset": offset}
        for key, value in (
            ("app_id", app_id), ("patient_reference", patient_reference),
            ("end_user_id", end_user_id), ("status", status), ("provider", provider),
        ):
            if value is not None:
                params[key] = value
        result = self._api.get(f"{_BASE}/sessions", params=params)
        return [VerificationSession.from_dict(s) for s in result.get("sessions", [])]

    def complete(self, session_id: str, *, code: str, state: str) -> VerificationSession:
        """Finish a session when you registered your own redirect URI with the vendor."""
        return VerificationSession.from_dict(
            self._api.post(f"{_BASE}/sessions/{session_id}/complete", json={"code": code, "state": state})
        )

    def cancel(self, session_id: str) -> VerificationSession:
        return VerificationSession.from_dict(self._api.post(f"{_BASE}/sessions/{session_id}/cancel", json={}))

    def match(
        self, session_id: str, *, first_name: str, last_name: str, date_of_birth: str,
    ) -> Dict[str, Any]:
        """``{"session_id", "match": bool, "reason"}`` without disclosing the demographics on record."""
        return self._api.post(
            f"{_BASE}/sessions/{session_id}/match",
            json={"first_name": first_name, "last_name": last_name, "date_of_birth": date_of_birth},
        )

    def link(self, session_id: str, end_user_id: Optional[str]) -> VerificationSession:
        """Link a pending or verified session to an app end user (the patient's
        account). Pass ``None`` to unlink."""
        return VerificationSession.from_dict(
            self._api.post(f"{_BASE}/sessions/{session_id}/link", json={"end_user_id": end_user_id})
        )

    def end_user_status(self, end_user_id: str) -> EndUserVerificationStatus:
        """Latest successful verification linked to an app end user."""
        return EndUserVerificationStatus.from_dict(self._api.get(f"{_BASE}/end-users/{end_user_id}"))

    def patient_status(self, patient_reference: str, *, provider: Optional[str] = None) -> PatientVerificationStatus:
        params = {"provider": provider} if provider else None
        return PatientVerificationStatus.from_dict(
            self._api.get(f"{_BASE}/patients/{patient_reference}", params=params)
        )

    def wait(
        self, session_id: str, *, timeout_seconds: float = 900.0, poll_interval_seconds: float = 3.0,
    ) -> VerificationSession:
        """Block until the session reaches a terminal status (or the timeout).

        Prefer the ``identity.verification_completed`` webhook in production;
        polling is for scripts and tests.
        """
        deadline = time.monotonic() + timeout_seconds
        while True:
            session = self.get(session_id)
            if session.is_terminal or time.monotonic() >= deadline:
                return session
            time.sleep(max(0.5, poll_interval_seconds))


================================================================================
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/storage.py
================================================================================

"""Object storage client for deployed apps."""

from __future__ import annotations

import json
from typing import Any, BinaryIO, Dict, Optional
from urllib.parse import urljoin

from hikigai.appsdk.models.storage import (
    SignedUrlResult,
    StorageListResult,
    StorageObjectInfo,
    StorageUploadResult,
)


class StorageClient:
    """Platform-managed object storage for app outputs and documents."""

    def __init__(self, api, project_id: str):
        self._api = api
        self._project_id = project_id

    def _base_path(self, app_id: str) -> str:
        return f"/api/v1/projects/{self._project_id}/apps/{app_id}/storage"

    def upload(
        self,
        app_id: str,
        file: BinaryIO,
        filename: str,
        content_type: str = "application/octet-stream",
        metadata: Optional[Dict[str, Any]] = None,
        ttl: Optional[int] = None,
    ) -> StorageUploadResult:
        path = f"{self._base_path(app_id)}/upload"
        data: Dict[str, Any] = {}
        if metadata is not None:
            data["metadata"] = json.dumps(metadata)
        if ttl is not None:
            data["ttl"] = str(ttl)
        if filename:
            data["filename"] = filename

        headers = self._api._get_headers()
        headers.pop("Content-Type", None)
        url = urljoin(self._api.base_url, path.lstrip("/"))
        response = self._api._client.post(
            url,
            data=data,
            files={"file": (filename, file, content_type)},
            headers=headers,
            timeout=max(self._api.timeout, 120.0),
        )
        payload = self._api._handle_response(response, headers.get("X-Request-ID", "storage-upload"))
        return StorageUploadResult(
            object_id=payload["object_id"],
            object_key=payload["object_key"],
            original_filename=payload["original_filename"],
            content_type=payload["content_type"],
            size_bytes=payload["size_bytes"],
            signed_url=payload["signed_url"],
            expires_at=payload.get("expires_at"),
            created_at=payload["created_at"],
        )

    def signed_url(
        self,
        app_id: str,
        object_id: str,
        ttl: int = 3600,
    ) -> SignedUrlResult:
        payload = self._api.get(
            f"{self._base_path(app_id)}/{object_id}/url",
            params={"ttl": ttl},
        )
        return SignedUrlResult(
            object_id=payload["object_id"],
            signed_url=payload["signed_url"],
            expires_in=payload["expires_in"],
            expires_at=payload["expires_at"],
        )

    def list(
        self,
        app_id: str,
        content_type: Optional[str] = None,
        status: str = "active",
        limit: int = 20,
        page: int = 1,
    ) -> StorageListResult:
        params: Dict[str, Any] = {"page": page, "limit": limit, "status": status}
        if content_type:
            params["content_type"] = content_type
        payload = self._api.get(self._base_path(app_id), params=params)
        objects = [StorageObjectInfo.from_api(item) for item in payload.get("objects", [])]
        return StorageListResult(
            objects=objects,
            total=payload.get("total", len(objects)),
            page=payload.get("page", page),
            limit=payload.get("limit", limit),
            total_pages=payload.get("total_pages", 1),
        )

    def get(self, app_id: str, object_id: str) -> StorageObjectInfo:
        payload = self._api.get(f"{self._base_path(app_id)}/{object_id}")
        return StorageObjectInfo.from_api(payload)

    def delete(self, app_id: str, object_id: str) -> None:
        self._api.delete(f"{self._base_path(app_id)}/{object_id}")


================================================================================
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}")


================================================================================
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/audio.py
================================================================================

"""Audio helpers for the speech endpoints.

Gemini Live wants exactly one input format — 16 kHz, mono, signed 16-bit
little-endian PCM — and almost no real audio source produces that. A phone
bridge hands you 8 kHz mu-law, a browser hands you 48 kHz float, a WAV file
hands you whatever it was recorded at. Getting this wrong does not raise;
it produces confident, garbled transcripts, which is miserable to debug.

So the conversion lives here rather than in every caller's app. Pure
stdlib: ``audioop`` was removed in Python 3.13, so the G.711 tables and the
resampler are implemented directly.

Example — telephony audio to model input::

    from hikigai.appsdk.audio import to_model_audio

    pcm = to_model_audio(mulaw_chunk, encoding="mulaw", sample_rate=8000)
    await session.send_audio(pcm)
"""

from __future__ import annotations

import struct
from array import array
from typing import Iterator, Literal, Union

__all__ = [
    "MODEL_INPUT_SAMPLE_RATE",
    "MODEL_OUTPUT_SAMPLE_RATE",
    "RECOMMENDED_CHUNK_MS",
    "AudioEncoding",
    "chunk_audio",
    "decode_alaw",
    "decode_mulaw",
    "downmix_to_mono",
    "from_model_audio",
    "pcm_duration_seconds",
    "pcm_to_wav",
    "resample_pcm16",
    "to_model_audio",
]

#: The input format the speech models require.
MODEL_INPUT_SAMPLE_RATE = 16000

#: The sample rate of translated audio coming back from the model.
MODEL_OUTPUT_SAMPLE_RATE = 24000

#: Google's recommended chunk duration — the latency/throughput sweet spot.
RECOMMENDED_CHUNK_MS = 100

AudioEncoding = Literal["pcm16", "f32", "mulaw", "alaw"]

Bytes = Union[bytes, bytearray, memoryview]


# ---------------------------------------------------------------------------
# G.711 decoding
# ---------------------------------------------------------------------------

def _build_mulaw_table() -> array:
    table = array("h", [0]) * 256
    for i in range(256):
        value = ~i & 0xFF
        sign = value & 0x80
        exponent = (value >> 4) & 0x07
        mantissa = value & 0x0F
        sample = (((mantissa << 3) + 0x84) << exponent) - 0x84
        table[i] = -sample if sign else sample
    return table


def _build_alaw_table() -> array:
    table = array("h", [0]) * 256
    for i in range(256):
        value = i ^ 0x55
        sign = value & 0x80
        exponent = (value & 0x70) >> 4
        mantissa = value & 0x0F
        if exponent == 0:
            sample = (mantissa << 4) + 8
        else:
            sample = ((mantissa << 4) + 0x108) << (exponent - 1)
        table[i] = -sample if sign else sample
    return table


_MULAW_TABLE = _build_mulaw_table()
_ALAW_TABLE = _build_alaw_table()


def decode_mulaw(data: Bytes) -> array:
    """Decode 8-bit G.711 mu-law bytes into 16-bit PCM samples."""
    return array("h", [_MULAW_TABLE[byte] for byte in bytes(data)])


def decode_alaw(data: Bytes) -> array:
    """Decode 8-bit G.711 A-law bytes into 16-bit PCM samples."""
    return array("h", [_ALAW_TABLE[byte] for byte in bytes(data)])


# ---------------------------------------------------------------------------
# Channel and rate conversion
# ---------------------------------------------------------------------------

def downmix_to_mono(samples: array, channels: int) -> array:
    """Average interleaved channels down to mono."""
    if channels <= 1:
        return samples
    frames = len(samples) // channels
    out = array("h", [0]) * frames
    for frame in range(frames):
        base = frame * channels
        total = sum(samples[base + channel] for channel in range(channels))
        out[frame] = _clamp16(round(total / channels))
    return out


def resample_pcm16(samples: array, from_rate: int, to_rate: int) -> array:
    """Resample by linear interpolation.

    Not a polyphase filter — speech at these rates does not need one, and a
    dependency-free helper that ships with the SDK beats a correct one the
    caller has to go find.
    """
    if from_rate == to_rate or not samples:
        return samples
    ratio = to_rate / from_rate
    out_length = max(1, round(len(samples) * ratio))
    out = array("h", [0]) * out_length
    last = len(samples) - 1
    for i in range(out_length):
        position = i / ratio
        left = int(position)
        right = min(left + 1, last)
        weight = position - left
        out[i] = _clamp16(round(samples[left] * (1 - weight) + samples[right] * weight))
    return out


# ---------------------------------------------------------------------------
# The one function most callers need
# ---------------------------------------------------------------------------

def to_model_audio(
    data: Union[Bytes, array],
    *,
    encoding: AudioEncoding = "pcm16",
    sample_rate: int = MODEL_INPUT_SAMPLE_RATE,
    channels: int = 1,
) -> bytes:
    """Convert audio from any supported format into model input.

    Returns 16 kHz, mono, signed 16-bit little-endian PCM, ready to pass to
    :meth:`SpeechSession.send_audio`.
    """
    samples = _decode_samples(data, encoding)
    samples = downmix_to_mono(samples, channels)
    samples = resample_pcm16(samples, sample_rate, MODEL_INPUT_SAMPLE_RATE)
    return _to_le_bytes(samples)


def _decode_samples(data: Union[Bytes, array], encoding: AudioEncoding) -> array:
    if encoding == "mulaw":
        return decode_mulaw(_as_bytes(data))
    if encoding == "alaw":
        return decode_alaw(_as_bytes(data))

    if encoding == "f32":
        floats = data if isinstance(data, array) and data.typecode == "f" else None
        if floats is None:
            floats = array("f")
            floats.frombytes(_as_bytes(data))
        return array(
            "h", [_clamp16(round(max(-1.0, min(1.0, value)) * 32767)) for value in floats]
        )

    if isinstance(data, array) and data.typecode == "h":
        return data
    samples = array("h")
    raw = _as_bytes(data)
    samples.frombytes(raw[: len(raw) - (len(raw) % 2)])
    if _BIG_ENDIAN:
        samples.byteswap()
    return samples


def from_model_audio(data: Bytes) -> array:
    """Interpret 24 kHz model output as samples, e.g. to feed a player."""
    samples = array("h")
    raw = _as_bytes(data)
    samples.frombytes(raw[: len(raw) - (len(raw) % 2)])
    if _BIG_ENDIAN:
        samples.byteswap()
    return samples


def chunk_audio(
    pcm: Bytes,
    *,
    ms: int = RECOMMENDED_CHUNK_MS,
    sample_rate: int = MODEL_INPUT_SAMPLE_RATE,
) -> Iterator[bytes]:
    """Split PCM into fixed-duration chunks.

    Sending one enormous buffer works but defeats streaming — the model
    cannot start translating until it arrives.
    """
    per_chunk = max(2, (sample_rate * 2 * ms // 1000) & ~1)
    raw = bytes(pcm)
    for offset in range(0, len(raw), per_chunk):
        yield raw[offset : offset + per_chunk]


def pcm_duration_seconds(pcm: Bytes, sample_rate: int = MODEL_INPUT_SAMPLE_RATE) -> float:
    """Duration of a 16-bit mono PCM buffer."""
    return len(pcm) / 2 / sample_rate


def pcm_to_wav(
    pcm: Bytes,
    *,
    sample_rate: int = MODEL_OUTPUT_SAMPLE_RATE,
    channels: int = 1,
) -> bytes:
    """Wrap raw PCM in a WAV header so it can be written straight to a file."""
    data = bytes(pcm)
    byte_rate = sample_rate * channels * 2
    header = b"".join(
        [
            b"RIFF",
            struct.pack("<I", 36 + len(data)),
            b"WAVEfmt ",
            struct.pack("<IHHIIHH", 16, 1, channels, sample_rate, byte_rate, channels * 2, 16),
            b"data",
            struct.pack("<I", len(data)),
        ]
    )
    return header + data


# ---------------------------------------------------------------------------
# Internals
# ---------------------------------------------------------------------------

_BIG_ENDIAN = array("h", [1]).tobytes()[0] == 0


def _clamp16(value: int) -> int:
    return 32767 if value > 32767 else -32768 if value < -32768 else value


def _as_bytes(data: Union[Bytes, array]) -> bytes:
    if isinstance(data, array):
        return data.tobytes()
    return bytes(data)


def _to_le_bytes(samples: array) -> bytes:
    if _BIG_ENDIAN:
        copy = array("h", samples)
        copy.byteswap()
        return copy.tobytes()
    return samples.tobytes()


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

"""Model package for AppSDK."""

from hikigai.appsdk.models.agent import RuntimeAgent
from hikigai.appsdk.models.response import InvokeResponse, StreamChunk

__all__ = [
    "RuntimeAgent",
    "InvokeResponse",
    "StreamChunk",
]


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

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

from typing import Optional, Dict, Any, Iterator, List, 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,
        attachments: Optional[List[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"}}
            attachments: Optional multimodal files (image, PDF, or audio such as
                mp3/wav) when the agent's ``input_modality`` includes that kind.
                Each item is one of ``{data_base64, mime_type}``, ``{url}``, or
                ``{storage_object_id}``. Spoken conversation stays on ``live()``.
            
        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,
            attachments=attachments,
        )
    
    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,
        *,
        connectors: Optional[Dict[str, Dict[str, Any]]] = None,
        context: Optional[Dict[str, Any]] = None,
        user_id: str = "sdk-user",
    ) -> "StreamSession":
        """
        Create a high-performance bidirectional live session.
        Useful for real-time audio streaming and transcription.

        Args:
            session_id: Optional session ID override.
            connectors: Optional MCP connector auth (same shape as ``invoke``).
            context: Optional hydration context embedded in the live token.
            user_id: Logical end-user id for ADK session isolation.

        Returns:
            StreamSession: An async context manager for the live connection.

        Example:
            async with agent.live_session(
                connectors={"epic-ehr": {}},
            ) as session:
                await session.send_audio(microphone_data)
                async for event in session:
                    if event.get("type") == "tool_call":
                        print(event)
        """
        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()),
            user_id=user_id,
            context=context,
            connectors=connectors,
        )


================================================================================
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
    artifacts: List[Dict[str, Any]] = Field(
        default_factory=list,
        description="Non-text outputs: [{kind, mime_type, data_base64|url|storage_object_id}]",
    )
    
    # 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/models/storage.py
================================================================================

"""Storage response models for AppSDK."""

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


@dataclass
class StorageUploadResult:
    object_id: str
    object_key: str
    original_filename: str
    content_type: str
    size_bytes: int
    signed_url: str
    expires_at: Optional[str]
    created_at: str


@dataclass
class SignedUrlResult:
    object_id: str
    signed_url: str
    expires_in: int
    expires_at: str


@dataclass
class StorageObjectInfo:
    object_id: str
    original_filename: str
    content_type: str
    size_bytes: int
    metadata: Dict[str, Any] = field(default_factory=dict)
    status: str = "active"
    upload_source: str = "api"
    expires_at: Optional[str] = None
    created_at: str = ""

    @classmethod
    def from_api(cls, data: Dict[str, Any]) -> "StorageObjectInfo":
        return cls(
            object_id=data["id"],
            original_filename=data["original_filename"],
            content_type=data["content_type"],
            size_bytes=data["size_bytes"],
            metadata=data.get("metadata") or {},
            status=data.get("status", "active"),
            upload_source=data.get("upload_source", "api"),
            expires_at=data.get("expires_at"),
            created_at=data.get("created_at", ""),
        )


@dataclass
class StorageListResult:
    objects: List[StorageObjectInfo]
    total: int
    page: int
    limit: int
    total_pages: int


================================================================================
File: hikigai/appsdk/streaming/session.py
================================================================================

import json
import base64
import logging
from typing import Optional, Any, Dict, AsyncIterator

_logger = logging.getLogger(__name__)

try:
    import websockets
except ImportError:
    websockets = None

# Protocol v2: JSON control frames + binary WebSocket frames for raw PCM audio.
LIVE_PROTOCOL_VERSION = 2


class StreamSession:
    """
    Bidirectional live session with a deployed agent.

    Connect flow:
      1. POST /api/v1/agents/{id}/live/session (API key auth at gateway)
      2. WebSocket to ``wss_url`` (direct Cloud Run or gateway /live) with live_token
      3. JSON control + binary PCM audio frames (protocol v2)

    Pass ``connectors`` (same shape as ``invoke``) to attach MCP tools on the
    live path. Mint then returns a gateway ``wss_url`` so connectors resolve
    before true bidirectional ``run_live``.
    """

    def __init__(
        self,
        client: Any,
        agent_id: str,
        session_id: str,
        *,
        user_id: str = "sdk-user",
        context: Optional[Dict[str, Any]] = None,
        connectors: Optional[Dict[str, Dict[str, Any]]] = None,
    ):
        self.client = client
        self.agent_id = agent_id
        self.session_id = session_id
        self.user_id = user_id
        self.context = context or {}
        self.connectors = connectors
        self._ws: Optional[Any] = None
        self._connected = False
        self._audio_mode = False

    async def __aenter__(self) -> "StreamSession":
        await self.connect()
        return self

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        await self.close()

    async def connect(self) -> None:
        """Mint live_token via HTTP, then open direct WebSocket to Cloud Run."""
        if websockets is None:
            raise ImportError(
                "The 'websockets' library is required for live streaming. "
                "Install it with: pip install websockets"
            )

        if self._connected:
            return

        _logger.info(
            "[live] requesting direct session agent_id=%s session_id=%s",
            self.agent_id,
            self.session_id[:8],
        )

        mint_body: Dict[str, Any] = {
            "session_id": self.session_id,
            "user_id": self.user_id,
            "context": self.context or None,
        }
        if self.connectors:
            mint_body["connectors"] = self.connectors

        session_payload = self.client.api.post(
            f"/api/v1/agents/{self.agent_id}/live/session",
            json=mint_body,
        )

        wss_url = session_payload["wss_url"]
        live_token = session_payload["live_token"]
        self.session_id = session_payload.get("session_id", self.session_id)

        _logger.debug(
            "[live] connecting wss_url=%s protocol=%s mcp=%s direct=%s",
            wss_url.split("?")[0],
            session_payload.get("protocol_version", LIVE_PROTOCOL_VERSION),
            bool(self.connectors),
            session_payload.get("direct_connect"),
        )

        self._ws = await websockets.connect(
            wss_url,
            open_timeout=45.0,
            ping_interval=20.0,
            ping_timeout=60.0,
        )

        auth_payload: Dict[str, Any] = {
            "type": "auth",
            "live_token": live_token,
            "session_id": self.session_id,
            "protocol_version": LIVE_PROTOCOL_VERSION,
        }
        # Gateway /live resolves these into mcp_credentials for the container.
        if self.connectors:
            auth_payload["connectors"] = self.connectors
        await self._ws.send(json.dumps(auth_payload))

        self._connected = True
        _logger.info("[live] connect established session_id=%s", self.session_id[:8])

    async def start(self, modality: str = "text", mime_type: str = "audio/pcm;rate=16000") -> None:
        """Signal start of a new input turn."""
        if not self._connected:
            await self.connect()
        self._audio_mode = modality == "audio"
        payload: Dict[str, Any] = {
            "type": "start",
            "modality": modality,
            "protocol_version": LIVE_PROTOCOL_VERSION,
        }
        if self._audio_mode:
            payload["mime_type"] = mime_type
        await self._ws.send(json.dumps(payload))
        _logger.debug("[live] start modality=%s session_id=%s", modality, self.session_id[:8])

    async def send_data(self, content: Any, mime_type: str = None) -> None:
        """
        Send a data chunk. Text uses JSON; audio uses binary frames (protocol v2).
        """
        if not self._connected:
            await self.connect()

        if isinstance(content, (bytes, bytearray)):
            await self._ws.send(bytes(content))
            _logger.debug(
                "[live] sent binary audio chunk bytes=%d session_id=%s",
                len(content),
                self.session_id[:8],
            )
        else:
            payload = {
                "type": "data",
                "modality": "text",
                "content": str(content),
            }
            await self._ws.send(json.dumps(payload))
            _logger.debug("[live] sent text data session_id=%s", self.session_id[:8])

    async def end(self) -> None:
        """Signal end of the current input turn."""
        if not self._connected:
            await self.connect()
        await self._ws.send(json.dumps({"type": "end"}))
        self._audio_mode = False
        _logger.debug("[live] end turn session_id=%s", self.session_id[:8])

    async def send_text(self, text: str) -> None:
        await self.start(modality="text")
        await self.send_data(text)
        await self.end()

    async def send_audio(self, audio_bytes: bytes, mime_type: str = "audio/pcm;rate=16000") -> None:
        await self.start(modality="audio", mime_type=mime_type)
        await self.send_data(audio_bytes)
        await self.end()

    async def __aiter__(self) -> AsyncIterator[Dict[str, Any]]:
        if not self._connected:
            await self.connect()

        try:
            async for message in self._ws:
                if isinstance(message, bytes):
                    yield {
                        "type": "data",
                        "modality": "audio",
                        "content": base64.b64encode(message).decode("utf-8"),
                        "binary": True,
                    }
                    continue
                try:
                    event = json.loads(message)
                except Exception:
                    event = {"type": "raw", "content": str(message)}

                if event.get("type") == "error":
                    _logger.error(
                        "[live] server error session_id=%s message=%s",
                        self.session_id[:8],
                        event.get("message"),
                    )
                else:
                    _logger.debug(
                        "[live] event type=%s session_id=%s",
                        event.get("type"),
                        self.session_id[:8],
                    )
                yield event
        except Exception as e:
            _logger.warning(
                "[live] iterator closed session_id=%s reason=%s",
                self.session_id[:8],
                e,
            )

    async def close(self) -> None:
        if self._ws:
            try:
                await self._ws.send(json.dumps({"type": "disconnect"}))
            except Exception:
                pass
            await self._ws.close()
            self._ws = None
            self._connected = False
        _logger.debug("[live] session closed session_id=%s", self.session_id[:8])


================================================================================
File: hikigai/appsdk/streaming/rooms.py
================================================================================

"""RoomSession — WebSocket rooms & broadcast client (mirrors StreamSession ergonomics).

Usage:
    room = RoomSession(base_url, api_key=..., project_id=...)
    async with room:
        await room.join("game-42")
        await room.publish("game-42", "move", {"x": 1})
        async for frame in room:
            if frame["type"] == "message":
                ...

Reconnect: the session tracks the last seen ``seq`` per room; call
``reconnect()`` to reopen the socket and rejoin every room with
``last_seq`` so at-least-once rooms replay the gap. A ``cursor_expired``
error means the gap outgrew the server's replay buffer — resync app
state from scratch.

WebRTC: ``rtc_join`` announces this connection as a peer in a room it has
already joined, and the ``rtc_*`` helpers carry SDP and ICE between
peers. This client is signalling only — it holds no media stack, so plug
the frames into whatever peer connection you use (``aiortc`` on Python,
``RTCPeerConnection`` in a browser)::

    await room.join("call-1")
    ack = await room.rtc_join("call-1")     # -> ice_servers, peers, polite
    async for frame in room:
        if frame["type"] == "rtc_peer_joined" and frame["initiator"]:
            await room.rtc_offer("call-1", frame["id"], await make_offer())
        elif frame["type"] == "rtc_offer":
            await room.rtc_answer("call-1", frame["sender"], await answer(frame["sdp"]))
"""

from __future__ import annotations

import json
import logging
import uuid
from typing import Any, AsyncIterator, Dict, List, Optional

_logger = logging.getLogger(__name__)

try:
    import websockets
except ImportError:  # pragma: no cover - optional dependency
    websockets = None


class RoomError(Exception):
    """Server-reported room error (carries the normative code)."""

    def __init__(self, code: str, message: str, room: Optional[str] = None):
        super().__init__(f"{code}: {message}")
        self.code = code
        self.room = room


class RoomSession:
    """Client for the rooms WebSocket endpoint (``/api/v1/rooms/ws``)."""

    def __init__(
        self,
        base_url: str,
        *,
        api_key: Optional[str] = None,
        access_token: Optional[str] = None,
        room_token: Optional[str] = None,
        project_id: Optional[str] = None,
    ):
        if not any([api_key, access_token, room_token]):
            raise ValueError("One of api_key, access_token, or room_token is required")
        self.base_url = base_url.rstrip("/")
        self.api_key = api_key
        self.access_token = access_token
        self.room_token = room_token
        self.project_id = project_id
        self.connection_id: Optional[str] = None
        self.capabilities: List[str] = []
        self.last_seq: Dict[str, int] = {}  # room -> last delivered seq
        self._joined: set = set()
        self._rtc_rooms: set = set()
        self._ws: Optional[Any] = None
        self._connected = False

    # ------------------------------------------------------------------
    async def __aenter__(self) -> "RoomSession":
        await self.connect()
        return self

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        await self.close()

    @property
    def _ws_url(self) -> str:
        url = self.base_url.replace("http://", "ws://").replace("https://", "wss://")
        return f"{url}/api/v1/rooms/ws"

    async def connect(self) -> None:
        if websockets is None:
            raise ImportError(
                "The 'websockets' library is required for rooms. "
                "Install it with: pip install websockets"
            )
        if self._connected:
            return
        self._ws = await websockets.connect(self._ws_url, max_size=2 * 1024 * 1024)
        auth: Dict[str, Any] = {"type": "auth"}
        if self.room_token:
            auth["room_token"] = self.room_token
        elif self.api_key:
            auth["api_key"] = self.api_key
        else:
            auth["access_token"] = self.access_token
        if self.project_id:
            auth["project_id"] = self.project_id
        await self._ws.send(json.dumps(auth))
        ready = json.loads(await self._ws.recv())
        if ready.get("type") == "error":
            raise RoomError(ready.get("code", "unauthorized"), ready.get("message", "auth failed"))
        self.connection_id = ready.get("connection_id")
        self.capabilities = ready.get("capabilities", [])
        self._connected = True
        _logger.info("[rooms] connected connection_id=%s", self.connection_id)

    async def close(self) -> None:
        if self._ws is not None:
            try:
                await self._ws.send(json.dumps({"type": "disconnect"}))
            except Exception:
                pass
            try:
                await self._ws.close()
            except Exception:
                pass
        self._ws = None
        self._connected = False

    async def reconnect(self) -> None:
        """Reopen the socket and rejoin all rooms with their last seen seq.

        RTC peers are re-announced too, but the peer id changes with the
        connection: treat every peer in the new ``rtc_joined`` ack as a
        fresh peer and tear down the old peer connections.
        """
        rooms = list(self._joined)
        rtc_rooms = list(self._rtc_rooms)
        self._connected = False
        self._ws = None
        self._rtc_rooms = set()
        await self.connect()
        for room in rooms:
            await self.join(room, last_seq=self.last_seq.get(room))
        for room in rtc_rooms:
            await self.rtc_join(room)

    # ------------------------------------------------------------------
    async def _send(self, frame: Dict[str, Any]) -> None:
        if not self._connected or self._ws is None:
            raise RuntimeError("RoomSession is not connected; call connect() first")
        await self._ws.send(json.dumps(frame))

    async def join(self, room: str, last_seq: Optional[int] = None) -> Dict[str, Any]:
        frame: Dict[str, Any] = {"type": "join", "room": room}
        if last_seq is not None:
            frame["last_seq"] = last_seq
        await self._send(frame)
        self._joined.add(room)
        return frame

    async def leave(self, room: str) -> None:
        await self._send({"type": "leave", "room": room})
        self._joined.discard(room)
        # Room membership is a precondition for being an RTC peer, so the
        # server drops the peer too — mirror that locally.
        self._rtc_rooms.discard(room)

    async def publish(
        self, room: str, event: str, data: Any = None, *, echo: bool = False
    ) -> None:
        await self._send(
            {"type": "publish", "room": room, "event": event, "data": data, "echo": echo}
        )

    async def who(self, room: str) -> None:
        await self._send({"type": "who", "room": room})

    async def ping(self) -> None:
        await self._send({"type": "ping"})

    async def refresh(self, room_token: str) -> None:
        self.room_token = room_token
        await self._send({"type": "refresh", "room_token": room_token})


    # ------------------------------------------------------------------
    # WebRTC signalling
    #
    # The server relays SDP/ICE between peers and never touches media, so
    # these helpers are deliberately transport-shaped: they send a frame
    # and return, and the responses arrive on the frame iterator.
    # ------------------------------------------------------------------
    async def rtc_join(
        self,
        room: str,
        *,
        mode: Optional[str] = None,
        agent_id: Optional[str] = None,
        metadata: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        """Announce this connection as an RTC peer in an already-joined room.

        Returns the frame that was sent; the server's ``rtc_joined`` ack
        (with ``ice_servers``, existing ``peers`` and this peer's id)
        arrives on the iterator. Requires the ``rtc`` capability.
        """
        frame: Dict[str, Any] = {"type": "rtc_join", "room": room}
        if mode is not None:
            frame["mode"] = mode
        if agent_id is not None:
            frame["agent_id"] = agent_id
        if metadata is not None:
            frame["metadata"] = metadata
        await self._send(frame)
        self._rtc_rooms.add(room)
        return frame

    async def rtc_leave(self, room: str) -> None:
        """Leave the RTC session but stay a member of the room."""
        await self._send({"type": "rtc_leave", "room": room})
        self._rtc_rooms.discard(room)

    async def rtc_offer(self, room: str, target: str, sdp: str) -> None:
        """Send an SDP offer to one peer."""
        await self._send(
            {"type": "rtc_offer", "room": room, "target": target, "sdp": sdp}
        )

    async def rtc_answer(self, room: str, target: str, sdp: str) -> None:
        """Answer one peer's offer."""
        await self._send(
            {"type": "rtc_answer", "room": room, "target": target, "sdp": sdp}
        )

    async def rtc_candidate(
        self, room: str, target: str, candidate: Dict[str, Any]
    ) -> None:
        """Trickle one ICE candidate to a peer.

        Pass the ``RTCIceCandidateInit`` shape (``candidate``, ``sdpMid``,
        ``sdpMLineIndex``). An empty ``candidate`` string is the standard
        end-of-candidates signal and is forwarded as-is.
        """
        await self._send(
            {
                "type": "rtc_candidate",
                "room": room,
                "target": target,
                "candidate": candidate,
            }
        )

    async def rtc_who(self, room: str) -> None:
        """Request the current peer list; the reply arrives as ``rtc_who``."""
        await self._send({"type": "rtc_who", "room": room})

    # ------------------------------------------------------------------
    def __aiter__(self) -> AsyncIterator[Dict[str, Any]]:
        return self._iter_frames()

    async def _iter_frames(self) -> AsyncIterator[Dict[str, Any]]:
        """Yield server frames; tracks per-room last_seq and dedupes replays."""
        if not self._connected or self._ws is None:
            raise RuntimeError("RoomSession is not connected; call connect() first")
        async for raw in self._ws:
            try:
                frame = json.loads(raw)
            except (TypeError, ValueError):
                continue
            if frame.get("type") == "message":
                room = frame.get("room")
                seq = frame.get("seq")
                if isinstance(seq, int) and room:
                    # Client contract (PRD §7): dedupe by seq.
                    if seq <= self.last_seq.get(room, 0):
                        continue
                    self.last_seq[room] = seq
            yield frame


def make_client_id() -> str:
    return f"room-client-{uuid.uuid4().hex[:8]}"


================================================================================
File: hikigai/appsdk/streaming/speech.py
================================================================================

"""Real-time speech translation and transcription.

Two WebSocket endpoints, one session object each. Both take 16 kHz mono PCM
in; translate gives you 24 kHz translated speech plus transcripts,
transcribe gives you text.

``connect()`` returns only after the model has acknowledged its
configuration, so audio sent immediately afterwards is never dropped.

Example — translate a microphone stream into Spanish::

    async with await client.speech.translate(target_language="es") as session:
        async def pump():
            async for chunk in microphone:
                await session.send_audio(chunk, encoding="f32", sample_rate=48000)

        asyncio.create_task(pump())

        async for event in session:
            if event.type == "audio":
                speaker.write(event.audio)
            elif event.type == "transcript":
                print(event.source, event.text, event.language_code)

Example — switch language mid-conversation::

    await session.set_language("ja")   # returns once the switch is live
"""

from __future__ import annotations

import asyncio
import json
import logging
from dataclasses import dataclass, field
from typing import Any, AsyncIterator, Dict, Optional, Sequence, Union

from hikigai.appsdk.audio import to_model_audio

try:
    import websockets
except ImportError:  # pragma: no cover - surfaced at connect()
    websockets = None

_logger = logging.getLogger(__name__)

__all__ = [
    "SpeechClient",
    "SpeechError",
    "SpeechEvent",
    "SpeechSession",
]

_DEFAULT_QUEUE_DEPTH = 500
_MAX_RECONNECT_ATTEMPTS = 3

# Errors that answer an in-flight `configure` rather than the stream at large.
_CONFIG_ERROR_CODES = frozenset(
    {"config_invalid", "unsupported_language", "forbidden", "upstream_unavailable"}
)


class SpeechError(Exception):
    """The server rejected the session or a configuration change."""

    def __init__(self, code: str, message: str):
        super().__init__(f"[{code}] {message}")
        self.code = code
        self.message = message


@dataclass
class SpeechEvent:
    """An event from the model.

    ``type`` is one of ``transcript``, ``audio``, ``text``, ``turn_complete``,
    ``interrupted``, ``reconfigured``, ``reconnected``, or ``error``.
    """

    type: str
    #: For transcripts: "input" (what was said) or "output" (the translation).
    source: str = ""
    text: str = ""
    #: True once the model considers this transcript segment settled.
    final: bool = False
    #: Detected BCP-47 code for this transcript (``es``, ``en-US``, …).
    language_code: str = ""
    #: Translated speech, 24 kHz mono 16-bit PCM. Translate sessions only.
    audio: bytes = b""
    config: Dict[str, Any] = field(default_factory=dict)
    changed: bool = False
    code: str = ""
    message: str = ""


class SpeechSession:
    """A live speech session, stable across language switches and reconnects."""

    def __init__(
        self,
        *,
        base_url: str,
        endpoint: str,
        api_key: Optional[str] = None,
        project_id: Optional[str] = None,
        speech_token: Optional[str] = None,
        config: Optional[Dict[str, Any]] = None,
        auto_reconnect: bool = True,
        max_queue_size: int = _DEFAULT_QUEUE_DEPTH,
    ):
        self.base_url = base_url.rstrip("/")
        self.endpoint = endpoint
        self._api_key = api_key
        self._project_id = project_id
        self._speech_token = speech_token
        self._initial_config = config or {}
        self._auto_reconnect = auto_reconnect

        #: Server-assigned id, stable across reconnects and language switches.
        self.session_id = ""
        #: The configuration currently in force, as the server reports it.
        self.config: Dict[str, Any] = {}
        #: True when the session was minted locked and cannot be reconfigured.
        self.locked = False
        #: The model backing this session.
        self.model = ""

        self._ws: Optional[Any] = None
        self._resume_token: Optional[str] = None
        self._queue: asyncio.Queue = asyncio.Queue(maxsize=max_queue_size)
        self._reader: Optional[asyncio.Task] = None
        self._pending_configure: Optional[asyncio.Future] = None
        self._closed_by_user = False
        self._done = False
        self._failure: Optional[BaseException] = None
        self._reconnects = 0

    @property
    def url(self) -> str:
        scheme = "wss" if self.base_url.startswith("https") else "ws"
        host = self.base_url.split("://", 1)[-1]
        return f"{scheme}://{host}/api/v1/speech/{self.endpoint}"

    # -- lifecycle ---------------------------------------------------------

    async def __aenter__(self) -> "SpeechSession":
        await self.connect()
        return self

    async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
        await self.close()

    async def connect(self) -> "SpeechSession":
        """Open the socket and wait for the model to accept the config.

        Returns only after the server's ``ready`` frame, which it withholds
        until the upstream model has acknowledged setup — audio sent before
        that point is discarded by the model, so this wait is load-bearing.
        """
        if websockets is None:
            raise ImportError(
                "The 'websockets' library is required for speech sessions. "
                "Install it with: pip install websockets"
            )
        if self._ws is not None:
            return self
        await self._open()
        return self

    async def _open(self) -> None:
        auth: Dict[str, Any] = {"type": "auth"}
        token = self._resume_token or self._speech_token
        if token:
            auth["speech_token"] = token
        else:
            auth["api_key"] = self._api_key
            auth["project_id"] = self._project_id
        # On reconnect the server already holds the current config; resending
        # it would be rejected outright if the token was minted locked.
        if self._initial_config and not self._resume_token:
            auth["config"] = self._initial_config

        self._ws = await websockets.connect(self.url, max_size=8 * 1024 * 1024)
        await self._ws.send(json.dumps(auth))

        raw = await self._ws.recv()
        frame = json.loads(raw if isinstance(raw, str) else raw.decode("utf-8"))
        if frame.get("type") == "error":
            await self._ws.close()
            self._ws = None
            raise SpeechError(
                str(frame.get("code", "internal")), str(frame.get("message", ""))
            )
        if frame.get("type") != "ready":
            await self._ws.close()
            self._ws = None
            raise SpeechError("internal", f"Unexpected first frame: {frame.get('type')!r}")

        self._adopt_ready(frame)
        self._reader = asyncio.create_task(self._read_loop(self._ws))

    def _adopt_ready(self, frame: Dict[str, Any]) -> None:
        self.session_id = str(frame.get("session_id") or self.session_id)
        self.model = str(frame.get("model") or self.model)
        self.locked = bool(frame.get("locked"))
        self.config = frame.get("config") or {}
        resume = frame.get("resume_token")
        if isinstance(resume, str):
            self._resume_token = resume
        self._reconnects = 0
        _logger.debug(
            "[speech] session ready session_id=%s model=%s", self.session_id, self.model
        )

    async def close(self) -> None:
        """Close the session. Safe to call more than once."""
        self._closed_by_user = True
        socket, self._ws = self._ws, None
        if socket is not None:
            try:
                await socket.send(json.dumps({"type": "disconnect"}))
            except Exception:
                pass  # already gone; closing below is enough
            try:
                await socket.close()
            except Exception:
                pass
        if self._reader is not None:
            self._reader.cancel()
            self._reader = None
        self._finish()

    # -- sending -----------------------------------------------------------

    async def send_audio(
        self,
        audio: Union[bytes, bytearray, memoryview],
        *,
        encoding: Optional[str] = None,
        sample_rate: Optional[int] = None,
        channels: int = 1,
    ) -> None:
        """Stream a chunk of audio.

        Pass ``encoding``/``sample_rate``/``channels`` when the source is not
        already 16 kHz mono PCM16 and the SDK will convert it.
        """
        if encoding is not None or sample_rate is not None or channels != 1:
            audio = to_model_audio(
                audio,
                encoding=encoding or "pcm16",  # type: ignore[arg-type]
                sample_rate=sample_rate or 16000,
                channels=channels,
            )
        await self._send(bytes(audio))

    async def end_input(self) -> None:
        """Tell the model the input ended so it flushes its final output."""
        await self._send(json.dumps({"type": "end"}))

    async def set_language(self, target_language: str) -> None:
        """Change the translation target mid-conversation.

        Returns once the new language is live. The gateway brings up a
        replacement model session before retiring the old one, so audio keeps
        flowing throughout and the caller hears no gap.
        """
        await self.configure(target_language_code=target_language)

    async def configure(self, **fields: Any) -> None:
        """Change configuration mid-session. Returns once applied."""
        if self.locked:
            raise SpeechError(
                "forbidden", "This session is locked to its minted configuration"
            )
        if self._pending_configure is not None and not self._pending_configure.done():
            raise SpeechError("conflict", "A configuration change is already in flight")

        loop = asyncio.get_running_loop()
        self._pending_configure = loop.create_future()
        try:
            await self._send(json.dumps({"type": "configure", "config": fields}))
        except Exception:
            self._pending_configure = None
            raise
        await self._pending_configure

    async def _send(self, payload: Union[str, bytes]) -> None:
        if self._ws is None:
            raise SpeechError("not_ready", "Speech session is not connected")
        await self._ws.send(payload)

    # -- receiving ---------------------------------------------------------

    async def _read_loop(self, socket: Any) -> None:
        try:
            async for raw in socket:
                if isinstance(raw, (bytes, bytearray)):
                    await self._emit(SpeechEvent(type="audio", audio=bytes(raw)))
                    continue
                try:
                    frame = json.loads(raw)
                except (TypeError, ValueError):
                    continue
                await self._handle_frame(frame)
        except asyncio.CancelledError:
            raise
        except Exception as exc:
            _logger.debug("[speech] read loop ended: %s", exc)
        await self._handle_close()

    async def _handle_frame(self, frame: Dict[str, Any]) -> None:
        kind = frame.get("type")

        if kind == "transcript":
            await self._emit(
                SpeechEvent(
                    type="transcript",
                    source="output" if frame.get("source") == "output" else "input",
                    text=str(frame.get("text") or ""),
                    final=bool(frame.get("final")),
                    language_code=str(frame.get("language_code") or ""),
                )
            )
        elif kind == "text":
            await self._emit(SpeechEvent(type="text", text=str(frame.get("text") or "")))
        elif kind in ("turn_complete", "interrupted"):
            await self._emit(SpeechEvent(type=kind))
        elif kind == "reconfigured":
            self.config = frame.get("config") or self.config
            self._settle_configure(None)
            await self._emit(
                SpeechEvent(
                    type="reconfigured",
                    config=self.config,
                    changed=bool(frame.get("changed")),
                )
            )
        elif kind == "error":
            code = str(frame.get("code") or "internal")
            message = str(frame.get("message") or "unknown error")
            # A rejected `configure` must fail that call rather than look like
            # an unrelated stream error to the consumer.
            if self._pending_awaiting_config() and code in _CONFIG_ERROR_CODES:
                self._settle_configure(SpeechError(code, message))
                return
            await self._emit(SpeechEvent(type="error", code=code, message=message))

    def _pending_awaiting_config(self) -> bool:
        return self._pending_configure is not None and not self._pending_configure.done()

    def _settle_configure(self, error: Optional[BaseException]) -> None:
        pending, self._pending_configure = self._pending_configure, None
        if pending is None or pending.done():
            return
        if error is None:
            pending.set_result(None)
        else:
            pending.set_exception(error)

    async def _emit(self, event: SpeechEvent) -> None:
        if self._done:
            return
        try:
            self._queue.put_nowait(event)
        except asyncio.QueueFull:
            # Drop the oldest so live audio keeps moving for a slow consumer.
            _logger.warning(
                "[speech] consumer is not keeping up; dropping events session_id=%s",
                self.session_id,
            )
            try:
                self._queue.get_nowait()
                self._queue.put_nowait(event)
            except (asyncio.QueueEmpty, asyncio.QueueFull):
                pass

    async def _handle_close(self) -> None:
        self._ws = None
        if self._closed_by_user or self._done:
            self._finish()
            return

        can_resume = (
            self._auto_reconnect
            and self._resume_token is not None
            and self._reconnects < _MAX_RECONNECT_ATTEMPTS
        )
        if not can_resume:
            self._finish()
            return

        self._reconnects += 1
        try:
            await self._open()
            await self._emit(SpeechEvent(type="reconnected"))
        except Exception as exc:
            self._failure = exc
            self._finish()

    def _finish(self) -> None:
        self._done = True
        self._settle_configure(
            self._failure or SpeechError("closed", "Speech session closed")
        )
        try:
            self._queue.put_nowait(None)
        except asyncio.QueueFull:
            pass

    def __aiter__(self) -> AsyncIterator[SpeechEvent]:
        return self

    async def __anext__(self) -> SpeechEvent:
        if self._done and self._queue.empty():
            if self._failure:
                raise self._failure
            raise StopAsyncIteration
        event = await self._queue.get()
        if event is None:
            if self._failure:
                raise self._failure
            raise StopAsyncIteration
        return event


# ---------------------------------------------------------------------------
# SpeechClient
# ---------------------------------------------------------------------------

class SpeechClient:
    """Entry point for the speech endpoints, reached via ``client.speech``."""

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

    # -- sessions ----------------------------------------------------------

    async def translate(
        self,
        *,
        target_language: Optional[str] = None,
        echo_target_language: Optional[bool] = None,
        input_transcription: Optional[bool] = None,
        output_transcription: Optional[bool] = None,
    ) -> SpeechSession:
        """Open a translation session, connected and ready for audio.

        ``target_language`` takes bare BCP-47 tags (``es``, ``ja``, ``pt-BR``),
        not locales. Echo and both transcriptions default on.
        """
        config = _translate_config(
            target_language, echo_target_language, input_transcription, output_transcription
        )
        return await self._start("translate", config)

    async def transcribe(
        self,
        *,
        languages: Optional[Sequence[str]] = None,
        custom_vocabulary: Optional[Sequence[str]] = None,
    ) -> SpeechSession:
        """Open a transcription session, connected and ready for audio.

        ``languages`` takes full locales (``en-US``, ``es-ES``); omit it to
        let the model detect the language.
        """
        return await self._start("transcribe", _transcribe_config(languages, custom_vocabulary))

    async def attach(
        self, endpoint: str, speech_token: str, **config: Any
    ) -> SpeechSession:
        """Attach to a session using a token minted elsewhere.

        This is the untrusted-client path: your server calls
        :meth:`mint_token` and sends the result on, so the API key never
        leaves your server.
        """
        session = SpeechSession(
            base_url=self._base_url,
            endpoint=endpoint,
            speech_token=speech_token,
            config=config or None,
        )
        return await session.connect()

    async def _start(self, endpoint: str, config: Dict[str, Any]) -> SpeechSession:
        session = SpeechSession(
            base_url=self._base_url,
            endpoint=endpoint,
            api_key=self._client.api_key,
            project_id=self._client.project_id,
            config=config or None,
        )
        return await session.connect()

    # -- REST --------------------------------------------------------------

    def mint_token(
        self,
        endpoint: str,
        *,
        principal: Optional[str] = None,
        locked: bool = False,
        ttl_seconds: Optional[int] = None,
        **config: Any,
    ) -> Dict[str, Any]:
        """Mint a short-lived pass for an untrusted client.

        Set ``locked=True`` to pin the configuration — useful when a
        deployment should always translate into one language and end users
        must not be able to change it.
        """
        return self._client.api.post(
            "/api/v1/speech/tokens",
            json={
                "endpoint": endpoint,
                "config": config,
                "principal": principal,
                "locked": locked,
                "ttl_seconds": ttl_seconds,
            },
        )

    def languages(self) -> Dict[str, Any]:
        """The language codes each endpoint accepts."""
        return self._client.api.get("/api/v1/speech/languages")

    @property
    def _base_url(self) -> str:
        return self._client.api.base_url


def _translate_config(
    target_language: Optional[str],
    echo_target_language: Optional[bool],
    input_transcription: Optional[bool],
    output_transcription: Optional[bool],
) -> Dict[str, Any]:
    config: Dict[str, Any] = {}
    if target_language is not None:
        config["target_language_code"] = target_language
    if echo_target_language is not None:
        config["echo_target_language"] = echo_target_language
    if input_transcription is not None:
        config["input_transcription"] = input_transcription
    if output_transcription is not None:
        config["output_transcription"] = output_transcription
    return config


def _transcribe_config(
    languages: Optional[Sequence[str]], custom_vocabulary: Optional[Sequence[str]]
) -> Dict[str, Any]:
    config: Dict[str, Any] = {}
    if languages is not None:
        config["language_codes"] = list(languages)
    if custom_vocabulary is not None:
        config["custom_vocabulary"] = list(custom_vocabulary)
    return config


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

"""EventStream — live project event subscription over WebSocket.

Connects to ``/api/v1/events/ws`` and yields CloudEvents envelopes for
every platform event matching the subscription patterns. Reach for this
from a process that is already running (a dashboard backend, a worker);
reach for webhooks when the platform should call you instead.

Usage::

    from hikigai.appsdk import AppClient

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

Delivery is at-most-once: the server keeps a bounded per-connection queue
and drops the oldest envelope under backpressure rather than blocking the
publisher. Anything you cannot afford to miss belongs on a webhook (which
retries) or a reconciling read, not on this socket alone.

The socket survives only as long as the process. On reconnect you resume
from *now* — there is no replay cursor — so re-read current state from the
REST API after :meth:`EventStream.reconnect`.
"""

from __future__ import annotations

import json
import logging
from typing import Any, AsyncIterator, Dict, List, Optional

_logger = logging.getLogger(__name__)

try:
    import websockets
except ImportError:  # pragma: no cover - optional dependency
    websockets = None


class EventStreamError(Exception):
    """Server-reported events-stream error (carries the normative code)."""

    def __init__(self, code: str, message: str):
        super().__init__(f"{code}: {message}")
        self.code = code
        self.message = message


class EventStream:
    """Client for the project events WebSocket (``/api/v1/events/ws``)."""

    def __init__(
        self,
        base_url: str,
        *,
        api_key: Optional[str] = None,
        access_token: Optional[str] = None,
        project_id: Optional[str] = None,
        patterns: Optional[List[str]] = None,
    ):
        if not any([api_key, access_token]):
            raise ValueError("One of api_key or access_token is required")
        self.base_url = (base_url or "").rstrip("/")
        self.api_key = api_key
        self.access_token = access_token
        self.project_id = project_id
        self.patterns: List[str] = list(patterns) if patterns else ["*"]
        self._ws: Optional[Any] = None
        self._connected = False

    # ------------------------------------------------------------------
    async def __aenter__(self) -> "EventStream":
        await self.connect()
        return self

    async def __aexit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        await self.close()

    @property
    def _ws_url(self) -> str:
        url = self.base_url.replace("http://", "ws://").replace("https://", "wss://")
        return f"{url}/api/v1/events/ws"

    async def connect(self) -> None:
        """Open the socket and complete the auth handshake."""
        if websockets is None:
            raise ImportError(
                "The 'websockets' library is required for event streaming. "
                "Install it with: pip install hikigai-appsdk[live]"
            )
        if self._connected:
            return
        self._ws = await websockets.connect(self._ws_url, max_size=2 * 1024 * 1024)
        auth: Dict[str, Any] = {"type": "auth", "patterns": self.patterns}
        if self.api_key:
            auth["api_key"] = self.api_key
        else:
            auth["access_token"] = self.access_token
        if self.project_id:
            auth["project_id"] = self.project_id
        await self._ws.send(json.dumps(auth))

        ready = json.loads(await self._ws.recv())
        if ready.get("type") == "error":
            raise EventStreamError(
                ready.get("code", "unauthorized"), ready.get("message", "auth failed")
            )
        # The server normalizes patterns (caps the count, defaults to "*");
        # trust its echo over what we asked for.
        self.patterns = ready.get("patterns", self.patterns)
        self.project_id = ready.get("project_id", self.project_id)
        self._connected = True
        _logger.info(
            "[events] connected project=%s patterns=%s", self.project_id, self.patterns
        )

    async def close(self) -> None:
        if self._ws is not None:
            try:
                await self._ws.send(json.dumps({"type": "disconnect"}))
            except Exception:
                pass
            try:
                await self._ws.close()
            except Exception:
                pass
        self._ws = None
        self._connected = False

    async def reconnect(self) -> None:
        """Reopen the socket with the current patterns.

        Events published while disconnected are gone — this stream has no
        replay buffer. Re-read state you care about from the REST API.
        """
        self._connected = False
        self._ws = None
        await self.connect()

    # ------------------------------------------------------------------
    async def _send(self, frame: Dict[str, Any]) -> None:
        if not self._connected or self._ws is None:
            raise RuntimeError("EventStream is not connected; call connect() first")
        await self._ws.send(json.dumps(frame))

    async def subscribe(self, patterns: List[str]) -> None:
        """Replace the active pattern set (not additive)."""
        self.patterns = list(patterns) if patterns else ["*"]
        await self._send({"type": "subscribe", "patterns": self.patterns})

    async def ping(self) -> None:
        await self._send({"type": "ping"})

    # ------------------------------------------------------------------
    def __aiter__(self) -> AsyncIterator[Dict[str, Any]]:
        return self.events()

    async def events(self) -> AsyncIterator[Dict[str, Any]]:
        """Yield CloudEvents envelopes (control frames are handled inline)."""
        if not self._connected or self._ws is None:
            raise RuntimeError("EventStream is not connected; call connect() first")
        async for raw in self._ws:
            try:
                frame = json.loads(raw)
            except (TypeError, ValueError):
                continue
            if not isinstance(frame, dict):
                continue
            frame_type = frame.get("type")
            if frame_type == "event":
                envelope = frame.get("event")
                if isinstance(envelope, dict):
                    yield envelope
            elif frame_type == "error":
                raise EventStreamError(
                    frame.get("code", "error"), frame.get("message", "stream error")
                )
            elif frame_type == "ready":
                self.patterns = frame.get("patterns", self.patterns)
            # pong and unknown frames are ignored (forward compatibility).

    async def frames(self) -> AsyncIterator[Dict[str, Any]]:
        """Yield raw server frames, including ``ready``/``pong`` control frames.

        Use this when you need the protocol layer; :meth:`events` is the
        ergonomic path.
        """
        if not self._connected or self._ws is None:
            raise RuntimeError("EventStream is not connected; call connect() first")
        async for raw in self._ws:
            try:
                frame = json.loads(raw)
            except (TypeError, ValueError):
                continue
            if isinstance(frame, dict):
                yield frame
