//// Actor-owned session data and its bounded event stream.

import albedo/daemon/bus
import albedo/daemon/context_snapshot
import albedo/daemon/conversation
import albedo/daemon/event_buffer
import albedo/daemon/transcript
import albedo/daemon/turn.{type Submission}
import albedo/daemon/usage
import albedo/harness/loop
import albedo/harness/runtime
import gleam/erlang/process.{type Subject}
import gleam/list
import gleam/option.{type Option}

pub type Watcher {
  Watcher(owner: process.Pid, notify: fn() -> Nil, notified: Bool)
}

pub type State(message) {
  State(
    info: conversation.Info,
    host: runtime.Runtime,
    kernel: Option(runtime.Session),
    home: String,
    self: Subject(message),
    history: Option(List(transcript.Entry)),
    latest_usage: Option(usage.Metadata),
    activity: turn.Activity,
    steering: List(Submission),
    active_submissions: List(Submission),
    sequence: Int,
    events: event_buffer.Buffer,
    watchers: List(Watcher),
    notice: Option(String),
    context: context_snapshot.Snapshot,
    pin: loop.Pin,
    /// Inputs evicted by the last prepared projection; the next live reload
    /// starts its pin at this baseline, even if its first turn compacts.
    prepared_head: Option(Int),
    last_touch: Int,
    /// While the kernel boots: boot attempts so far, and the work waiting for
    /// it, re-sent to the actor once the kernel is ready.
    booting: Option(#(Int, List(message))),
    /// Where an ancestor's move sent this session while it was busy; taken
    /// when the run ends. Not persisted: a restart forgets it.
    following: Option(String),
    blocked_until: Int,
    generation: String,
  )
}

/// Streaming clients are woken as each event is published, so a model delta
/// reaches a terminal without waiting for a polling interval.
pub fn emit(state: State(message), event: String) -> State(message) {
  let seq = state.sequence + 1
  let events = event_buffer.push(state.events, seq, event)
  let watchers =
    list.map(state.watchers, fn(watcher) {
      case watcher.notified {
        True -> watcher
        False -> {
          watcher.notify()
          Watcher(..watcher, notified: True)
        }
      }
    })
  bus.activity(state.info.id, event)
  State(..state, sequence: seq, events: events, watchers: watchers)
}

pub fn unprepared() -> context_snapshot.Snapshot {
  context_snapshot.pending(
    "runtime session has not prepared a provider request",
  )
}