"""Event envelope, emitter and session state machine (no I/O rendering).

This module is the single source of truth for the machine-readable control
plane: the ``Event`` envelope written to ``<session_dir>/events.jsonl`` and
the session state machine mirrored into ``<session_dir>/state.json``.  It
performs no rendering and no file I/O itself -- :mod:`spens.sinks` consumes
it.  ``runner.py``/``builder.py`` depend only on :func:`emit` (and the
validation helpers); they never import a sink directly.

Event stream schema (``schema_version`` 1):

    {"schema_version": 1, "event": "...", "session_id": "...",
     "timestamp": "<ISO-8601 UTC>", "data": {...}}

Event set (see specs/spens_non_interactive.md):

    started, warning, build_started, build_finished, interceptor_starting,
    interceptor_ready, agent_started, agent_output, agent_exited, finished,
    canceled, error, info

``interceptor_starting`` and ``info`` carry no state of their own beyond the
spec's table: ``interceptor_starting`` exists so the ``interceptor_starting``
state in ``state.json`` is backed by an event, and ``info`` carries today's
informational ``[spens] ...`` tty lines (e.g. "Interceptor reachable as
...") that are neither warnings nor any other listed event.  Consumers must
ignore event types they do not know.

Terminal events are ``finished``, ``canceled`` and ``error``; the first
writer of a terminal *state* wins (see :data:`TERMINAL_STATES` and
:func:`is_valid_transition`).
"""

from __future__ import annotations

import json
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import Any, Protocol

# Bump on breaking changes to the event set; consumers reject versions they
# do not understand.
SCHEMA_VERSION = 1

# Session states written to state.json, in lifecycle order.
STATES = (
    "building",
    "interceptor_starting",
    "running",
    "exited",
    "canceled",
    "error",
    "finished",
)

#: States that end a session; once written they are never overwritten
#: (this is what makes an external ``spens cancel`` race-free).
TERMINAL_STATES = frozenset({"finished", "canceled", "error"})

#: Allowed state transitions.  ``None`` is the pre-session state; ``error``
#: and ``canceled`` are reachable from every non-terminal state (setup can
#: fail at any point, and an external cancel can arrive at any point).
TRANSITIONS: dict[str | None, frozenset[str]] = {
    None: frozenset({"building"}),
    "building": frozenset({"interceptor_starting", "error", "canceled"}),
    "interceptor_starting": frozenset({"running", "error", "canceled"}),
    "running": frozenset({"exited", "error", "canceled"}),
    "exited": frozenset({"finished", "error", "canceled"}),
}
for _terminal in TERMINAL_STATES:
    TRANSITIONS[_terminal] = frozenset()

#: Events that imply a state transition (used by FileSink to update
#: state.json).  Events not listed here (warning, build_started,
#: build_finished, interceptor_ready, agent_output, info, ...) do not change
#: the state.
EVENT_STATES: dict[str, str] = {
    "started": "building",
    "interceptor_starting": "interceptor_starting",
    "agent_started": "running",
    "agent_exited": "exited",
    "finished": "finished",
    "canceled": "canceled",
    "error": "error",
}


def utc_now_iso() -> str:
    """Current time as an ISO-8601 UTC timestamp."""
    return datetime.now(UTC).isoformat()


def is_terminal(state: str | None) -> bool:
    return state in TERMINAL_STATES


def is_valid_transition(old: str | None, new: str) -> bool:
    """Whether ``old -> new`` is a legal state-machine transition."""
    return new in TRANSITIONS.get(old, frozenset())


@dataclass
class Event:
    """One machine-readable lifecycle event."""

    event: str
    session_id: str
    timestamp: str
    data: dict[str, Any] = field(default_factory=dict)

    def to_json(self) -> dict[str, Any]:
        return {
            "schema_version": SCHEMA_VERSION,
            "event": self.event,
            "session_id": self.session_id,
            "timestamp": self.timestamp,
            "data": self.data,
        }

    def to_line(self) -> str:
        """The canonical one-line JSON serialization (events.jsonl / stdout)."""
        return json.dumps(self.to_json())


class Sink(Protocol):
    """The single small interface every sink implements."""

    def on_event(self, event: Event) -> None: ...

    def close(self) -> None: ...


class Emitter:
    """Builds events for one session and fans them out to its sinks."""

    def __init__(self, session_id: str, sinks: list[Sink] | None = None) -> None:
        self.session_id = session_id
        self.sinks: list[Sink] = list(sinks or [])

    def emit(self, event: str, **data: Any) -> Event:
        built = Event(
            event=event,
            session_id=self.session_id,
            timestamp=utc_now_iso(),
            data=data,
        )
        for sink in self.sinks:
            sink.on_event(built)
        return built

    def close(self) -> None:
        for sink in self.sinks:
            sink.close()


class NullEmitter(Emitter):
    """The default emitter: builds events, delivers them to nobody.

    Keeps ``emit(...)`` callable from pure generator functions (builder
    entrypoint/profile generation) that also run in isolation (unit tests)
    where no session is configured.
    """


# ---------------------------------------------------------------------------
# Module-level plumbing
#
# builder.py / runner.py call ``events.emit(...)`` without holding a
# reference to the session's emitter; the active session's emitter is
# configured once per process (one session per process, including the
# background child).
# ---------------------------------------------------------------------------

_active: Emitter = NullEmitter("")


def configure(emitter: Emitter) -> Emitter:
    """Install ``emitter`` as the process-wide active emitter."""
    global _active
    previous = _active
    _active = emitter
    return previous


def reset() -> None:
    """Drop the active emitter (test isolation)."""
    global _active
    _active = NullEmitter("")


def active() -> Emitter:
    return _active


def emit(event: str, **data: Any) -> Event:
    """Emit ``event`` on the active session's emitter (no-op without one)."""
    return _active.emit(event, **data)
