# Copyright 2026 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
# pyformat: disable

"""Code generated by Speakeasy (https://speakeasy.com). DO NOT EDIT."""

from __future__ import annotations
from .. import BaseModel, UNSET_SENTINEL, UnrecognizedStr
from ...utils.unions import parse_open_union
from .agentoption import AgentOption
from .antigravityagentconfig import AntigravityAgentConfig, AntigravityAgentConfigParam
from .audiocontent import AudioContent, AudioContentParam
from .codemenderagentconfig import CodeMenderAgentConfig, CodeMenderAgentConfigParam
from .deepresearchagentconfig import (
    DeepResearchAgentConfig,
    DeepResearchAgentConfigParam,
)
from .dynamicagentconfig import DynamicAgentConfig, DynamicAgentConfigParam
from .environment import Environment, EnvironmentParam
from .generationconfig import GenerationConfig, GenerationConfigParam
from .imagecontent import ImageContent, ImageContentParam
from .interactionsinput import InteractionsInput, InteractionsInputParam
from .model import Model
from .responseformat import ResponseFormat, ResponseFormatParam
from .responsemodality import ResponseModality
from .safetysetting import SafetySetting, SafetySettingParam
from .servicetier import ServiceTier
from .step import Step, StepParam
from .tool import Tool, ToolParam
from .usage import Usage, UsageTypedDict
from .videocontent import VideoContent, VideoContentParam
from .webhookconfig import WebhookConfig, WebhookConfigParam
from functools import partial
import pydantic
from pydantic import ConfigDict, model_serializer, model_validator
from pydantic.functional_validators import BeforeValidator
from typing import Any, Dict, List, Literal, Optional, Union
from typing_extensions import Annotated, NotRequired, TypeAliasType, TypedDict


InteractionStatus = Union[
    Literal[
        "in_progress",
        "requires_action",
        "completed",
        "failed",
        "cancelled",
        "incomplete",
        "budget_exceeded",
    ],
    UnrecognizedStr,
]
r"""Required. Output only. The status of the interaction."""


InteractionResponseFormatTypedDict = TypeAliasType(
    "InteractionResponseFormatTypedDict",
    Union[List[ResponseFormatParam], ResponseFormatParam],
)
r"""Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field."""


InteractionResponseFormat = TypeAliasType(
    "InteractionResponseFormat", Union[List[ResponseFormat], ResponseFormat]
)
r"""Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field."""


InteractionEnvironmentTypedDict = TypeAliasType(
    "InteractionEnvironmentTypedDict", Union[EnvironmentParam, str]
)
r"""The environment configuration for the interaction. Can be an object specifying remote environment sources or a string referencing an existing environment ID."""


InteractionEnvironment = TypeAliasType(
    "InteractionEnvironment", Union[Environment, str]
)
r"""The environment configuration for the interaction. Can be an object specifying remote environment sources or a string referencing an existing environment ID."""


InteractionAgentConfigTypedDict = TypeAliasType(
    "InteractionAgentConfigTypedDict",
    Union[
        DynamicAgentConfigParam,
        AntigravityAgentConfigParam,
        DeepResearchAgentConfigParam,
        CodeMenderAgentConfigParam,
    ],
)
r"""Configuration parameters for the agent interaction."""


class UnknownInteractionAgentConfig(BaseModel):
    r"""A InteractionAgentConfig variant the SDK doesn't recognize. Preserves the raw payload."""

    type: Literal["UNKNOWN"] = "UNKNOWN"
    raw: Any
    is_unknown: Literal[True] = True

    model_config = ConfigDict(frozen=True)


_INTERACTION_AGENT_CONFIG_VARIANTS: dict[str, Any] = {
    "dynamic": DynamicAgentConfig,
    "deep-research": DeepResearchAgentConfig,
    "code-mender": CodeMenderAgentConfig,
    "antigravity": AntigravityAgentConfig,
}


InteractionAgentConfig = Annotated[
    Union[
        DynamicAgentConfig,
        DeepResearchAgentConfig,
        CodeMenderAgentConfig,
        AntigravityAgentConfig,
        UnknownInteractionAgentConfig,
    ],
    BeforeValidator(
        partial(
            parse_open_union,
            disc_key="type",
            variants=_INTERACTION_AGENT_CONFIG_VARIANTS,
            unknown_cls=UnknownInteractionAgentConfig,
            union_name="InteractionAgentConfig",
            lenient=True,
        )
    ),
]
r"""Configuration parameters for the agent interaction."""


class InteractionTypedDict(TypedDict):
    r"""The Interaction resource."""

    status: InteractionStatus
    r"""Required. Output only. The status of the interaction."""
    model: NotRequired[Model]
    r"""The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details."""
    agent: NotRequired[AgentOption]
    r"""The agent to interact with."""
    id: NotRequired[str]
    r"""Required. Output only. A unique identifier for the interaction completion."""
    created: NotRequired[str]
    r"""Output only. The time at which the response was created in ISO 8601 format
    (YYYY-MM-DDThh:mm:ssZ).
    """
    updated: NotRequired[str]
    r"""Output only. The time at which the response was last updated in ISO 8601 format
    (YYYY-MM-DDThh:mm:ssZ).
    """
    system_instruction: NotRequired[str]
    r"""System instruction for the interaction."""
    tools: NotRequired[List[ToolParam]]
    r"""A list of tool declarations the model may call during interaction."""
    usage: NotRequired[UsageTypedDict]
    r"""Statistics on the interaction request's token usage."""
    response_modalities: NotRequired[List[ResponseModality]]
    r"""The requested modalities of the response (TEXT, IMAGE, AUDIO)."""
    response_mime_type: NotRequired[str]
    r"""The mime type of the response. This is required if response_format is set."""
    previous_interaction_id: NotRequired[str]
    r"""The ID of the previous interaction, if any."""
    environment_id: NotRequired[str]
    r"""Output only. The environment ID for the interaction. Only populated if environment
    config is set in the request.
    """
    service_tier: NotRequired[ServiceTier]
    webhook_config: NotRequired[WebhookConfigParam]
    r"""Message for configuring webhook events for a request."""
    steps: NotRequired[List[StepParam]]
    r"""Output only. The steps that make up the interaction, when included in the response."""
    response_format: NotRequired[InteractionResponseFormatTypedDict]
    r"""Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field."""
    environment: NotRequired[InteractionEnvironmentTypedDict]
    r"""The environment configuration for the interaction. Can be an object specifying remote environment sources or a string referencing an existing environment ID."""
    generation_config: NotRequired[GenerationConfigParam]
    r"""Configuration parameters for model interactions."""
    agent_config: NotRequired[InteractionAgentConfigTypedDict]
    r"""Configuration parameters for the agent interaction."""
    safety_settings: NotRequired[List[SafetySettingParam]]
    r"""Safety settings for the interaction."""
    labels: NotRequired[Dict[str, str]]
    r"""The labels with user-defined metadata for the request."""
    input: NotRequired[InteractionsInputParam]
    r"""The input for the interaction."""
    output_text: NotRequired[str]
    r"""Concatenated text from the last model output in response to the current request.

    Note: this is added by the SDK.
    """
    output_image: NotRequired[ImageContentParam]
    r"""An image content block."""
    output_audio: NotRequired[AudioContentParam]
    r"""An audio content block."""
    output_video: NotRequired[VideoContentParam]
    r"""A video content block."""


class Interaction(BaseModel):
    r"""The Interaction resource."""

    status: InteractionStatus
    r"""Required. Output only. The status of the interaction."""

    model: Optional[Model] = None
    r"""The model that will complete your prompt.\n\nSee [models](https://ai.google.dev/gemini-api/docs/models) for additional details."""

    agent: Optional[AgentOption] = None
    r"""The agent to interact with."""

    id: Optional[str] = ""
    r"""Required. Output only. A unique identifier for the interaction completion."""

    created: Optional[str] = None
    r"""Output only. The time at which the response was created in ISO 8601 format
    (YYYY-MM-DDThh:mm:ssZ).
    """

    updated: Optional[str] = None
    r"""Output only. The time at which the response was last updated in ISO 8601 format
    (YYYY-MM-DDThh:mm:ssZ).
    """

    system_instruction: Optional[str] = None
    r"""System instruction for the interaction."""

    tools: Optional[List[Tool]] = None
    r"""A list of tool declarations the model may call during interaction."""

    usage: Optional[Usage] = None
    r"""Statistics on the interaction request's token usage."""

    response_modalities: Optional[List[ResponseModality]] = None
    r"""The requested modalities of the response (TEXT, IMAGE, AUDIO)."""

    response_mime_type: Annotated[
        Optional[str],
        pydantic.Field(
            deprecated="warning: ** DEPRECATED ** - This will be removed in a future release, please migrate away from it as soon as possible."
        ),
    ] = None
    r"""The mime type of the response. This is required if response_format is set."""

    previous_interaction_id: Optional[str] = None
    r"""The ID of the previous interaction, if any."""

    environment_id: Optional[str] = None
    r"""Output only. The environment ID for the interaction. Only populated if environment
    config is set in the request.
    """

    service_tier: Optional[ServiceTier] = None

    webhook_config: Optional[WebhookConfig] = None
    r"""Message for configuring webhook events for a request."""

    steps: Optional[List[Step]] = None
    r"""Output only. The steps that make up the interaction, when included in the response."""

    response_format: Optional[InteractionResponseFormat] = None
    r"""Enforces that the generated response is a JSON object that complies with the JSON schema specified in this field."""

    environment: Optional[InteractionEnvironment] = None
    r"""The environment configuration for the interaction. Can be an object specifying remote environment sources or a string referencing an existing environment ID."""

    generation_config: Optional[GenerationConfig] = None
    r"""Configuration parameters for model interactions."""

    agent_config: Optional[InteractionAgentConfig] = None
    r"""Configuration parameters for the agent interaction."""

    safety_settings: Optional[List[SafetySetting]] = None
    r"""Safety settings for the interaction."""

    labels: Optional[Dict[str, str]] = None
    r"""The labels with user-defined metadata for the request."""

    input: Optional[InteractionsInput] = None
    r"""The input for the interaction."""

    output_text: Optional[str] = None
    r"""Concatenated text from the last model output in response to the current request.

    Note: this is added by the SDK.
    """

    output_image: Optional[ImageContent] = None
    r"""An image content block."""

    output_audio: Optional[AudioContent] = None
    r"""An audio content block."""

    output_video: Optional[VideoContent] = None
    r"""A video content block."""

    @model_serializer(mode="wrap")
    def _serialize_model(self, handler):
        optional_fields = set(
            [
                "model",
                "agent",
                "id",
                "created",
                "updated",
                "system_instruction",
                "tools",
                "usage",
                "response_modalities",
                "response_mime_type",
                "previous_interaction_id",
                "environment_id",
                "service_tier",
                "webhook_config",
                "steps",
                "response_format",
                "environment",
                "generation_config",
                "agent_config",
                "safety_settings",
                "labels",
                "input",
                "output_text",
                "output_image",
                "output_audio",
                "output_video",
            ]
        )
        serialized = handler(self)
        m = {}

        for n, f in type(self).model_fields.items():
            k = f.alias or n
            val = serialized.get(k, serialized.get(n))

            if val != UNSET_SENTINEL:
                if val is not None or k not in optional_fields:
                    m[k] = val

        return m

    @classmethod
    def _maybe_coerce_outputs(cls, data: Any) -> Any:
        legacy_lyria_models = ("lyria-3-pro-preview", "lyria-3-clip-preview")
        if not isinstance(data, dict):
            return data
        if (
            not isinstance(data.get("model"), str)
            or data["model"] not in legacy_lyria_models
        ):
            return data
        if "steps" in data:
            return data
        if "outputs" not in data:
            return {**data, "steps": []}
        outputs = data["outputs"]
        if not isinstance(outputs, list):
            return {**data, "steps": []}

        coerced = {**data}
        coerced.pop("outputs")
        coerced["steps"] = [{"type": "model_output", "content": outputs}]
        return coerced

    @model_validator(mode="before")
    @classmethod
    def _coerce_outputs_to_steps(cls, data: Any) -> Any:
        return cls._maybe_coerce_outputs(data)

    @classmethod
    def model_construct(cls, _fields_set=None, **values):
        # Coerce legacy lyria ``outputs`` -> ``steps`` here as well: validators
        # do not run on model_construct (used by deferred SSE parsing).
        return super().model_construct(_fields_set, **cls._maybe_coerce_outputs(values))

    @model_validator(mode="after")
    def _populate_output_helpers(self):
        steps = self.steps if isinstance(self.steps, list) else []

        text_parts: List[str] = []
        collecting = False
        for step in reversed(steps):
            step_type = getattr(step, "type", None)
            if step_type == "user_input":
                break
            if step_type != "model_output":
                if collecting:
                    break
                continue
            content = getattr(step, "content", None)
            if not isinstance(content, list):
                if collecting:
                    break
                continue
            should_stop = False
            for item in reversed(content):
                if getattr(item, "type", None) == "text":
                    collecting = True
                    text = getattr(item, "text", None)
                    text_parts.append(text if isinstance(text, str) else "")
                elif collecting:
                    should_stop = True
                    break
            if should_stop:
                break
        # Always derived from the trailing model-output text; a str, "" when none.
        self.output_text = "".join(reversed(text_parts))

        output_image = None
        output_audio = None
        output_video = None
        for step in reversed(steps):
            step_type = getattr(step, "type", None)
            if step_type == "user_input":
                break
            if step_type != "model_output":
                continue
            content = getattr(step, "content", None)
            if not isinstance(content, list):
                continue
            for item in reversed(content):
                content_type = getattr(item, "type", None)
                if content_type == "image" and output_image is None:
                    output_image = item
                if content_type == "audio" and output_audio is None:
                    output_audio = item
                if content_type == "video" and output_video is None:
                    output_video = item
        if output_image is not None and self.output_image is None:
            self.output_image = output_image
        if output_audio is not None and self.output_audio is None:
            self.output_audio = output_audio
        if output_video is not None and self.output_video is None:
            self.output_video = output_video
        return self
