"""GenericWorkflowBridge — shallow and deep modes.

Supports user-defined orchestration code that does not use
``pydantic_ai.Agent`` or ``pydantic_graph.Graph``.

Built on :class:`~easycat.integrations.agents.template.BridgeTemplate`,
which owns the ``invoke()`` cursor lifecycle, the four-step atomic
interruption journal protocol, and the safe no-op mutation defaults —
this module only implements event streaming, interruption planning,
and state snapshots.
"""

from __future__ import annotations

import inspect
import logging
from collections.abc import AsyncIterator
from typing import Any

from easycat.cancel import CancelToken
from easycat.integrations.agents._helpers import aclose_quietly
from easycat.integrations.agents._state_serialization import serialize_framework_state
from easycat.integrations.agents.base import (
    AgentBridgeEvent,
    AgentRecorder,
    AgentTurnInput,
    BridgeInputError,
    CancellationMode,
    FrameworkStateSnapshot,
    InterruptionPlan,
    ShallowModeInterruptionError,
)
from easycat.integrations.agents.template import BridgeTemplate
from easycat.runtime.records import ErrorInfo

logger = logging.getLogger(__name__)


class GenericWorkflowBridge(BridgeTemplate):
    """Bridge for user-defined orchestration code.

    Two modes:

    - **Shallow** (default): ``on_user_turn(text) -> str|AsyncIterator[str]``.
      One opaque cursor per turn.
    - **Deep** (opt-in): ``on_user_turn(text, *, recorder, cancel_token)``
      with ``recorder: AgentRecorder`` parameter.  User calls
      ``recorder.record_*`` methods from inside their orchestration.

    Mode is detected at construction via signature inspection.

    Interruption / barge-in behaviour
    ----------------------------------
    **Deep mode** exposes the session's cancellation token so the workflow
    can stop upstream work cooperatively.  The session's streaming path then
    calls :meth:`apply_interruption`, and the bridge runs the four-step atomic
    write ordering (plan -> commit -> mutate -> paired record) used by
    journaled interruption handling.

    In **shallow mode**, the bridge still observes cancellation and stops
    forwarding streamed chunks, while the session cancels queued output.
    The workflow cannot see that token or safely reconcile opaque internal
    state by default, though.  ``apply_interruption`` therefore raises
    :class:`ShallowModeInterruptionError` unless the workflow explicitly
    implements its own reconciliation hook.

    To opt in to mid-turn interruption in shallow mode, implement
    ``apply_interruption(delivered_text, mode)`` directly on the workflow
    object.  The bridge delegates to it via the same four-step atomic
    write ordering used by deep mode.
    """

    TURN_UNIT_ID_PREFIX = "workflow"

    def __init__(
        self,
        workflow: Any,
        *,
        display_name: str | None = None,
    ) -> None:
        super().__init__(display_name=display_name or type(workflow).__name__)
        self._workflow = workflow

        fn = getattr(workflow, "on_user_turn", None)
        if not callable(fn):
            raise BridgeInputError(
                "Workflow must implement on_user_turn(). "
                "See GenericWorkflowBridge docs for the supported signatures."
            )

        sig = inspect.signature(fn)
        self._deep_mode = "recorder" in sig.parameters
        self._accepts_cancel_token = "cancel_token" in sig.parameters
        self._mcp_warning_emitted = False

    @property
    def deep_mode(self) -> bool:
        return self._deep_mode

    # ── BridgeTemplate hooks ──────────────────────────────────────

    def on_turn_start(self, turn_input: AgentTurnInput, recorder: AgentRecorder) -> None:
        # Emit one-time warning if MCP servers configured on shallow workflow.
        if not self._mcp_warning_emitted and not self._deep_mode and recorder.context.mcp_servers:
            self._mcp_warning_emitted = True
            recorder.record_framework_error(
                ErrorInfo(
                    type="MCPShallowModeWarning",
                    message=(
                        f"MCP servers {list(recorder.context.mcp_servers)!r} "
                        "configured but GenericWorkflowBridge is in shallow mode. "
                        "MCP wiring is not supported in shallow mode — convert "
                        "to deep mode to use MCP servers."
                    ),
                )
            )

    def stream_events(
        self,
        turn_input: AgentTurnInput,
        recorder: AgentRecorder,
        cancel_token: CancelToken | None,
    ) -> AsyncIterator[AgentBridgeEvent]:
        if self._deep_mode:
            return self._invoke_deep(turn_input, recorder, cancel_token)
        return self._invoke_shallow(turn_input, cancel_token)

    def snapshot_state(self) -> FrameworkStateSnapshot:
        fields: dict[str, Any] = {
            "display_name": self._display_name,
            "mode": "deep" if self._deep_mode else "shallow",
        }
        if hasattr(self._workflow, "snapshot_state"):
            try:
                ws = self._workflow.snapshot_state()
                if isinstance(ws, dict):
                    fields["workflow_state"] = ws
            except Exception:  # noqa: BLE001, S110 intentional boundary or best-effort cleanup
                pass
        return FrameworkStateSnapshot(
            fields=fields,
            kind="generic_workflow",
        )

    def check_interruption_supported(self) -> None:
        # Shallow mode without explicit override → raise immediately.
        if not self._deep_mode and not hasattr(self._workflow, "apply_interruption"):
            raise ShallowModeInterruptionError(
                "Interruption is not supported in GenericWorkflowBridge "
                "shallow mode. Convert the workflow to deep mode by adding "
                "a `recorder: AgentRecorder` parameter to `on_user_turn`, "
                "or implement `workflow.apply_interruption(delivered_text, "
                "mode)` on the workflow object itself."
            )

    def _serialize_framework_state(self) -> bytes:
        """Serialize workflow state for artifact storage.

        Prefer the workflow's own ``snapshot_state()`` when available so
        user code decides what is safe to persist.  The fallback path
        walks ``__dict__`` but scrubs values whose keys look like
        credentials — artifacts end up in debug bundles that can be
        shared, so a blanket ``__dict__`` dump would otherwise leak API
        keys and auth headers stored on the workflow object.
        """
        from easycat.runtime.safe_defaults import _is_secret_name

        try:
            state: dict[str, Any] | None = None
            if hasattr(self._workflow, "snapshot_state"):
                try:
                    ws = self._workflow.snapshot_state()
                    if isinstance(ws, dict):
                        state = ws
                except Exception:  # noqa: BLE001 intentional boundary or best-effort cleanup
                    state = None
            if state is None:
                raw = getattr(self._workflow, "__dict__", {})
                if isinstance(raw, dict):
                    # Substring-match attribute names (``authtoken``,
                    # ``apisecret``) that the serializer's word-boundary
                    # classifier would keep; a blanket ``__dict__`` dump
                    # warrants the over-exclusion.
                    state = {k: v for k, v in raw.items() if not _is_secret_name(str(k))}
                else:
                    state = {}
            return serialize_framework_state(state)
        except Exception:  # noqa: BLE001 intentional boundary or best-effort cleanup
            return b"{}"

    def _plan_interruption(self, delivered_text: str, mode: CancellationMode) -> InterruptionPlan:
        has_override = hasattr(self._workflow, "apply_interruption")
        mutation_kind = "interrupt_workflow_override" if has_override else "interrupt_cancel_token"
        pre_ref = f"workflow-pre-{id(self._workflow):x}"
        post_ref = f"workflow-post-{id(self._workflow):x}"
        return InterruptionPlan(
            mutation_kind=mutation_kind,
            pre_state_ref=pre_ref,
            post_state_ref=post_ref,
            framework_instructions={
                "has_override": has_override,
                "delivered_text": delivered_text,
                "mode": mode.value,
            },
        )

    def _apply_planned_mutation(self, plan: InterruptionPlan) -> None:
        instructions = plan.framework_instructions
        if instructions.get("has_override"):
            mode = CancellationMode(instructions["mode"])
            self._workflow.apply_interruption(instructions["delivered_text"], mode)
        else:
            # Deep mode without explicit apply_interruption — best-effort.
            logger.debug(
                "Deep-mode workflow %s has no apply_interruption; relying on cancel_token",
                self._display_name,
            )

    def reset(self) -> None:
        if hasattr(self._workflow, "reset"):
            self._workflow.reset()
        elif hasattr(self._workflow, "clear_history"):
            self._workflow.clear_history()
        super().reset()

    # ── History post-processing ──────────────────────────────────

    def replace_last_assistant_text(self, text: str) -> None:
        """Delegate to the workflow if it supports history rewrites."""
        fn = getattr(self._workflow, "replace_last_assistant_text", None)
        if callable(fn):
            try:
                fn(text)
            except Exception:
                logger.debug("Workflow replace_last_assistant_text failed", exc_info=True)

    def append_interruption_note(self, note: str) -> None:
        """Delegate to the workflow if it supports interruption notes."""
        fn = getattr(self._workflow, "append_interruption_note", None)
        if callable(fn):
            try:
                fn(note)
            except Exception:
                logger.debug("Workflow append_interruption_note failed", exc_info=True)

    # ── Shallow mode ─────────────────────────────────────────────

    async def _invoke_shallow(  # noqa: PLR0912 - explicit close at each stream boundary
        self,
        turn_input: AgentTurnInput,
        cancel_token: CancelToken | None,
    ) -> AsyncIterator[AgentBridgeEvent]:
        # Check for streaming variant first.
        if hasattr(self._workflow, "on_user_turn_streaming"):
            # Streaming chunks are inherently unstructured text, so we leave
            # ``_last_output`` at its ``None`` default rather than emitting the
            # concatenated text as ``structured_output``.  That would merely
            # duplicate the ``done`` event's ``text`` field and could surface a
            # partial-but-presented-as-complete value on barge-in cancel.  This
            # matches the deep-mode streaming branch, which likewise leaves
            # ``structured_output`` unset for streamed chunks; the awaitable /
            # plain-value branches still expose a real structured object.
            stream = self._workflow.on_user_turn_streaming(turn_input.text)
            try:
                async for chunk in stream:
                    if cancel_token and cancel_token.is_cancelled:
                        break
                    if chunk:
                        yield AgentBridgeEvent(kind="text_delta", text=str(chunk))
            finally:
                await aclose_quietly(stream)
            return

        result = self._workflow.on_user_turn(turn_input.text)
        if isinstance(result, AsyncIterator):
            # See streaming note above: leave ``_last_output`` unset for
            # streamed chunks (parity with deep mode, no partial-on-cancel).
            try:
                async for chunk in result:
                    if cancel_token and cancel_token.is_cancelled:
                        break
                    if chunk:
                        yield AgentBridgeEvent(kind="text_delta", text=str(chunk))
            finally:
                await aclose_quietly(result)
        elif inspect.isawaitable(result):
            output = await result
            self._last_output = output
            text = self._extract_text(output)
            if text:
                yield AgentBridgeEvent(kind="text_delta", text=text)
        else:
            self._last_output = result
            text = self._extract_text(result)
            if text:
                yield AgentBridgeEvent(kind="text_delta", text=text)

    # ── Deep mode ────────────────────────────────────────────────

    async def _invoke_deep(
        self,
        turn_input: AgentTurnInput,
        recorder: AgentRecorder,
        cancel_token: CancelToken | None,
    ) -> AsyncIterator[AgentBridgeEvent]:
        kwargs: dict[str, Any] = {"recorder": recorder}
        if self._accepts_cancel_token:
            kwargs["cancel_token"] = cancel_token
        result = self._workflow.on_user_turn(turn_input.text, **kwargs)
        # Deep mode may return str, AsyncIterator[str], or a coroutine.
        if isinstance(result, AsyncIterator):
            try:
                async for chunk in result:
                    if cancel_token and cancel_token.is_cancelled:
                        break
                    if chunk:
                        yield AgentBridgeEvent(kind="text_delta", text=str(chunk))
            finally:
                await aclose_quietly(result)
        elif inspect.isawaitable(result):
            output = await result
            self._last_output = output
            text = self._extract_text(output)
            if text:
                yield AgentBridgeEvent(kind="text_delta", text=text)
        else:
            self._last_output = result
            text = self._extract_text(result)
            if text:
                yield AgentBridgeEvent(kind="text_delta", text=text)

    # ── Helpers ──────────────────────────────────────────────────

    def _extract_text(self, result: Any) -> str:
        if isinstance(result, str):
            return result
        if hasattr(result, "text"):
            return str(result.text)
        return str(result) if result is not None else ""
