# SPDX-License-Identifier: Apache-2.0
"""Sessions, voices and records (SPEC §4–6).

Every voice heard in a session gains an identity and its own record, and controls it:
its requested level, forgetting, retention. Anyone can lower; a voice raises only itself;
a raise for others applies at once but provisionally — for the objection window anyone can
undo it, retroactively.

    applied(v) = min( requested(v) if set else default,  ceiling,  cap(v) )
"""
from __future__ import annotations

import secrets
from dataclasses import dataclass, field

from .grammar import Intent
from .profile import Profile

TOP = 5
NEG, POS = float("-inf"), float("inf")


def _n(x):
    """Canonical JSON numbers: integral floats become ints (Python and JS logs then match)."""
    return int(x) if isinstance(x, float) and x.is_integer() else x


@dataclass
class Segment:
    """A stretch of one voice's speech. `held` is set while a provisional raise could still be undone."""
    voice: str
    t0: float
    t1: float
    level: int
    held: int | None = None


@dataclass
class Voice:
    id: str                          # pseudonymous, session-local (SPEC §4.2)
    label: str                       # the binding's label for it (speaker separator, roster, stream)
    slot: int                        # order of first speech
    requested: int | None = None     # what the voice asked for its own words (None = no preference)
    cap: int = TOP                   # a lowering by others that applies to this voice only (side conversation)
    retention: float | None = None
    record: list[Segment] = field(default_factory=list)


@dataclass
class Session:
    profile: Profile
    id: str = field(default_factory=lambda: secrets.token_hex(8))
    default: int = -1
    ceiling: int = TOP
    retention: float | None = None
    provisional: dict | None = None  # {"since", "opened", "default", "ceiling", "caps"} before an open raise
    voices: dict[str, Voice] = field(default_factory=dict)
    over: set[str] = field(default_factory=set)              # labels heard beyond max_participants
    events: list[dict] = field(default_factory=list)
    last_heard: float | None = None
    started: float | None = None
    closed_at: float | None = None       # when the previous session ended (late speech from it is refused)
    _forgotten: list[tuple[float, float, str | None]] = field(default_factory=list)
    _timeline: list[tuple[float, dict, int]] = field(default_factory=list)
    _links: dict[frozenset, int] = field(default_factory=dict)
    _last_turn: tuple[str, float] | None = None
    _ids: int = 0
    _seq: int = 0

    def __post_init__(self) -> None:
        if self.default < 0:
            self.default = self.profile.default_level

    # --- time and voices ----------------------------------------------------------
    def advance(self, now: float) -> None:
        """The one clock entry point: close an expired objection window, then end the session after
        `session_gap` of silence (SPEC §6.3, §6.7). An open window at the gap is undone, not confirmed."""
        gap_end = None if self.last_heard is None else self.last_heard + self.profile.session_gap
        if self.provisional and (gap_end is None or self.provisional["opened"] + self.profile.objection_window < gap_end):
            self._confirm_if_due(now)
        if gap_end is not None and now >= gap_end:
            self.end(gap_end)

    def heard(self, label: str, now: float) -> Voice | None:
        """The binding heard `label` speak at `now`. A new voice gains an identity and a record."""
        self.advance(now)
        self.last_heard = now if self.last_heard is None else max(self.last_heard, now)
        if self.started is None:
            self.started = now
        if self._last_turn and self._last_turn[0] != label and now - self._last_turn[1] <= self.profile.reply_window:
            k = frozenset((self._last_turn[0], label))
            self._links[k] = self._links.get(k, 0) + 1
        self._last_turn = (label, now)
        return self._voice(label, now)

    def _voice(self, label: str, now: float) -> Voice | None:
        """Give `label` an identity and a record if it has none — without touching the clock or reply links."""
        if label not in self.voices and label not in self.over:
            if len(self.voices) >= self.profile.max_participants:
                self.over.add(label)
                return None
            self._ids += 1
            v = self.voices[label] = Voice(id=f"v{self._ids}", label=label, slot=len(self.voices))
            self._log(now, "voice", voice=v.id, slot=v.slot, level=self.applied(label))
            self._snapshot(now)
        return self.voices.get(label)

    def side_conversation(self, label: str | None) -> set[str]:
        """The voices `label` has been answering, when the session has split (extension: side_conversations)."""
        if label is None or "side_conversations" not in self.profile.extensions:
            return set()
        adj: dict[str, set[str]] = {}
        for pair, n in self._links.items():
            if n >= self.profile.replies_for_group:
                a, b = tuple(pair)
                adj.setdefault(a, set()).add(b)
                adj.setdefault(b, set()).add(a)
        comps, seen = [], set()
        for s0 in adj:
            if s0 in seen:
                continue
            comp, stack = set(), [s0]
            while stack:
                x = stack.pop()
                if x not in comp:
                    comp.add(x)
                    stack.extend(adj[x] - comp)
            seen |= comp
            comps.append(comp)
        if sum(len(c) >= 2 for c in comps) < 2:
            return set()                               # not split: the whole session is one conversation
        return next((c for c in comps if label in c), set())

    # --- levels ------------------------------------------------------------------
    def applied(self, label: str | None = None) -> int:
        """The level the recorder applies to `label`'s words now; with no label, to a voice that set nothing."""
        if label in self.over:
            return 0                                   # never recorded (SPEC §4.1)
        v = self.voices.get(label) if label is not None else None
        base = v.requested if (v and v.requested is not None) else self.default
        return min(base, self.ceiling, v.cap if v else TOP)

    def retention_for(self, label: str | None) -> float | None:
        v = self.voices.get(label) if label is not None else None
        caps = [x for x in (self.retention, v.retention if v else None) if x is not None]
        return min(caps) if caps else None

    def _applied_at(self, label: str, t: float) -> int | None:
        lvl = None
        for (ts, per, dflt) in self._timeline:
            if ts > t:
                break
            lvl = per.get(label, dflt)
        return lvl

    # --- records -----------------------------------------------------------------
    def stamp(self, label: str, t0: float, t1: float) -> Segment | None:
        """File [t0, t1] of `label`'s speech in its record, at the lowest level that applied over the span."""
        self.advance(t1)
        floor_t = self.started if self.started is not None else self.closed_at
        if label in self.over or (floor_t is not None and t0 < floor_t):
            return None                                # never recorded: over the cap, or from before this session (SPEC §6.7)
        if self.started is None:
            self.started = t0
        v = self._voice(label, t0)
        if v is None:
            return None
        levels = [self.applied(label)] + [per.get(label, d) for (ts, per, d) in self._timeline if t0 < ts <= t1]   # in force at t0, plus changes during
        at0 = self._applied_at(label, t0)
        if at0 is not None:
            levels.append(at0)
        lvl = min(levels)
        for (a, b, who) in self._forgotten:
            if (who is None or who == v.id) and t0 <= b and a <= t1:
                lvl = 0
        seg = Segment(v.id, t0, t1, lvl)
        if self.provisional and t1 >= self.provisional["since"]:
            seg.held = min(lvl, self._before(label))
        v.record.append(seg)
        self.last_heard = t1 if self.last_heard is None else max(self.last_heard, t1)
        return seg

    def readable(self, seg: Segment, now: float) -> int:
        """The level a segment may be read at now: stamped, held, and retention (SPEC §5)."""
        from .retention import readable_level
        lvl = seg.level if seg.held is None else min(seg.level, seg.held)
        label = next((lab for lab, v in self.voices.items() if v.id == seg.voice), None)
        return min(lvl, readable_level(now - seg.t1, self.profile.retention, self.retention_for(label)))

    def usable(self, level: int, t0: float = NEG, t1: float = POS, now: float | None = None) -> list[Segment]:
        """Segments a derived artefact at `level` may draw on (SPEC §5.4): provisional content at its held level,
        and, given `now`, after retention."""
        def lvl(s):
            return self.readable(s, now) if now is not None else (s.level if s.held is None else min(s.level, s.held))
        return [s for v in self.voices.values() for s in v.record if lvl(s) >= level and s.t0 <= t1 and t0 <= s.t1]

    # --- intents -----------------------------------------------------------------
    def apply(self, i: Intent, label: str | None, confidence: float, now: float) -> str:
        """Apply one intent attributed to `label` with `confidence`. Returns applied | provisional | ignored | reverted."""
        self.advance(now)
        p = self.profile
        sure = label is not None and label not in self.over and confidence >= p.attribution_threshold
        if i.kind == "claim" and "claiming" not in p.extensions:
            return self._log(now, "ignored", intent=i.kind, reason="unsupported")
        scope = i.scope
        if scope == "self" and not sure:
            if i.kind in ("lower", "forget", "shorten"):
                scope = "room"                           # unsure lowering → everyone (SPEC §6.5)
            else:
                return self._log(now, "ignored", intent=i.kind, reason="unsure-speaker")
        before = self._levels()
        if scope == "self":
            out = self._self(i, self.voices.get(label) or self._voice(label, now), now)
        else:
            out = self._room(i, label if sure else None, now)
        self._report(now, before)
        return out

    def _self(self, i: Intent, v: Voice | None, now: float) -> str:
        if v is None:
            return self._log(now, "ignored", intent=i.kind, reason="no-slot")
        cur = v.requested if v.requested is not None else self.default
        if i.kind == "lower" and i.level is not None:
            v.requested = min(cur, i.level)
        elif i.kind == "raise" and i.level is not None:
            v.requested = max(cur, min(i.level, self.profile.max_voice_level))
        elif i.kind == "resume":
            v.requested = None
        elif i.kind == "forget":
            if i.seconds is None:                        # forgetting everything also thins the voice's events (SPEC §9)
                self.events = [{k: e[k] for k in ("session", "seq", "t", "type")} if e.get("voice") == v.id else
                               ({**e, "levels": {k: x for k, x in e["levels"].items() if k != v.id}} if "levels" in e else e) for e in self.events]
            return self._forget(now - i.seconds if i.seconds is not None else NEG, now, v, now, anonymous=i.seconds is None)
        elif i.kind == "shorten" and i.seconds is not None:
            v.retention = i.seconds if v.retention is None else min(v.retention, i.seconds)
            return self._log(now, "retention", voice=v.id, seconds=v.retention)
        elif i.kind == "claim":
            return self._log(now, "claim", voice=v.id)
        else:
            return self._log(now, "ignored", intent=i.kind, reason="unsupported")
        return "applied"

    def _room(self, i: Intent, label: str | None, now: float) -> str:
        p, k = self.profile, i.kind
        if k == "lower" and i.level is not None:
            if self.provisional:
                self._revert(now, report=False)          # a room lowering inside the window undoes the raise
            group = self.side_conversation(label)
            if group:
                for lab in group:
                    v = self.voices.get(lab)
                    if v and i.level < v.cap:
                        v.cap = i.level
            else:
                self.ceiling = min(self.ceiling, i.level)
            return "applied"
        if k == "forget":
            return self._forget(now - i.seconds if i.seconds is not None else NEG, now, None, now)
        if k == "shorten" and i.seconds is not None:
            self.retention = i.seconds if self.retention is None else min(self.retention, i.seconds)
            return self._log(now, "retention", seconds=self.retention)
        if k == "object":
            if self.provisional is None:
                return self._log(now, "ignored", intent=k, reason="nothing-to-undo")
            self._revert(now, report=False)
            return "reverted"
        if k in ("raise", "resume") and label is None and i.source == "speech":
            return self._log(now, "ignored", intent=k, reason="unsure-speaker")   # a raise for others needs a known voice (SPEC §6.5)
        if k == "resume":
            group = self.side_conversation(label)
            capped = [v for lab, v in self.voices.items() if (not group or lab in group) and v.cap < TOP]
            if not capped and self.ceiling == TOP:
                return self._log(now, "ignored", intent=k, reason="not-lowered")
            self._open_window(now)
            for v in capped:
                v.cap = TOP
            if not group:
                self.ceiling = TOP
            return self._log(now, "provisional", intent=k, until=_n(now + p.objection_window))
        if k == "raise" and i.level is not None:
            if p.room_raise == "disabled":
                return self._log(now, "ignored", intent=k, reason="room-raise-disabled")
            if not i.addressed:
                return self._log(now, "ignored", intent=k, reason="not-addressed")
            lvl = min(i.level, p.max_default_level)
            if lvl <= self.default:
                return self._log(now, "ignored", intent=k, reason="not-a-raise")
            self._open_window(now)
            self.default = lvl                             # ceilings stay: lifting them is resume's job (SPEC §6.4)
            return self._log(now, "provisional", intent=k, until=_n(now + p.objection_window))
        return self._log(now, "ignored", intent=k, reason="unsupported")

    # --- provisional raises (SPEC §6.3) ---------------------------------------------
    def _open_window(self, now: float) -> None:
        """Snapshot the state before the first open raise; a later raise restarts the window."""
        if self.provisional is None:
            self.provisional = {"since": now, "default": self.default, "ceiling": self.ceiling,
                                "caps": {lab: v.cap for lab, v in self.voices.items()}}
        self.provisional["opened"] = now

    def _before(self, label: str) -> int:
        pv = self.provisional
        v = self.voices.get(label)
        base = v.requested if (v and v.requested is not None) else pv["default"]
        return min(base, pv["ceiling"], pv["caps"].get(label, TOP))

    def _confirm_if_due(self, now: float) -> None:
        pv = self.provisional
        if pv and now - pv["opened"] > self.profile.objection_window:   # an objection at the deadline still counts
            self.provisional = None
            for v in self.voices.values():
                for s in v.record:
                    s.held = None
            self._log(pv["opened"] + self.profile.objection_window, "confirmed")

    def _revert(self, now: float, report: bool = True) -> None:
        pv, before = self.provisional, self._levels()
        floors = {lab: self._before(lab) for lab in self.voices}
        floor_new = min(pv["default"], pv["ceiling"])     # for voices first heard inside the window
        self.provisional = None
        self.default, self.ceiling = pv["default"], pv["ceiling"]
        for lab, v in self.voices.items():
            v.cap = pv["caps"].get(lab, TOP)
            for s in v.record:
                if s.t1 >= pv["since"]:
                    s.level = min(s.level, floors[lab])
                s.held = None
        # rewrite history too, so speech from the window filed later gets the earlier level (SPEC §6.3)
        self._timeline = [(ts, {lab: min(x, floors.get(lab, floor_new)) for lab, x in per.items()}, min(d, floor_new)) if ts >= pv["since"] else (ts, per, d)
                          for (ts, per, d) in self._timeline]
        self._log(now, "reverted")
        if report:
            self._report(now, before)

    def tick(self, now: float) -> None:
        self.advance(now)

    def end(self, now: float) -> dict[str, list[Segment]]:
        """Close the session and hand its records to the store (SPEC §6.7). An open raise is undone."""
        if self.provisional:
            self._revert(now)
        records = {v.id: v.record for v in self.voices.values()}
        was = self.applied(None)
        self.default, self.ceiling, self.retention = self.profile.default_level, TOP, None
        if self.applied(None) != was:
            self._log(now, "level", levels={"*": self.applied(None)})
        self.voices = {}
        self.over = set()
        self._forgotten = []
        self._timeline = []
        self._links = {}
        self._last_turn = None
        self.last_heard = None
        self.started = None
        self.closed_at = now
        self._log(now, "end")
        self.id = secrets.token_hex(8)
        return records

    # --- internals ------------------------------------------------------------------
    def _forget(self, t0: float, t1: float, v: Voice | None, now: float, anonymous: bool = False) -> str:
        self._forgotten.append((t0, t1, v.id if v else None))
        for x in ([v] if v else self.voices.values()):
            for s in x.record:
                if s.t0 <= t1 and t0 <= s.t1:
                    s.level = 0
                    s.held = None
        if anonymous:                                    # forgetting everything leaves no trace of who (SPEC §9)
            return self._log(now, "forget", scope="self", t0=None, t1=_n(t1))
        return self._log(now, "forget", voice=v.id if v else None, t0=None if t0 == NEG else _n(t0), t1=_n(t1))

    def _levels(self) -> dict:
        return {"*": self.applied(None), **{v.id: self.applied(lab) for lab, v in self.voices.items()}}

    def _snapshot(self, now: float) -> None:
        self._timeline.append((now, {lab: self.applied(lab) for lab in self.voices}, self.applied(None)))

    def _report(self, now: float, before: dict) -> None:
        """Report every changed applied level (SPEC §6.8, §11)."""
        after = self._levels()
        changed = {k: v for k, v in after.items() if before.get(k) != v}
        if changed:
            self._log(now, "level", levels=changed)
        self._snapshot(now)

    def _log(self, now: float, type_: str, **kw) -> str:
        self._seq += 1
        self.events.append({"session": self.id, "seq": self._seq, "t": _n(now), "type": type_, **{k: _n(v) for k, v in kw.items()}})
        return {"provisional": "provisional", "ignored": "ignored", "reverted": "reverted"}.get(type_, "applied")