///|
/// Agent-facing projection of committed Puppet envelopes onto observers,
/// plus the per-turn stream-chunk callback wiring.

///|
/// Agent-facing projection of committed Puppet envelopes. It never scans a
/// final transcript. The pending call cache is correlation state populated by
/// `ToolBatchStarted` and consumed by completion events. A committed
/// Pending events retain the model's original call; completion events carry
/// the canonical call that reached the host.
priv struct AgentEventSubscriber {
  observers : Array[&@port.Observer]
  pending_calls : Map[String, @kernel.ToolCall]
  chunk_dispatcher : Ref[ChunkDispatcher?]
}

///|
fn AgentEventSubscriber::AgentEventSubscriber(
  observers : Array[&@port.Observer],
  chunk_dispatcher : Ref[ChunkDispatcher?],
) -> AgentEventSubscriber {
  { observers, pending_calls: Map::from_array([]), chunk_dispatcher, }
}

///|
impl @puppetry.EventSubscriber for AgentEventSubscriber with fn subscriber_id(
  _self : AgentEventSubscriber,
) -> String {
  "agent_committed_event_projection"
}

///|
impl @puppetry.EventSubscriber for AgentEventSubscriber with fn provenance(
  _self : AgentEventSubscriber,
) -> String {
  "posoco.agent"
}

///|
fn AgentEventSubscriber::emit(
  self : AgentEventSubscriber,
  scope : @types.EventScope?,
  event : @types.TurnEvent,
) -> Unit {
  // Piggyback flush: catch up buffered chunk telemetry before every committed
  // event so observers see chunks before ModelResponseReceived, ToolCallPending,
  // etc., and so they catch up at boundaries during an unbroken SSE burst.
  match self.chunk_dispatcher.val {
    Some(dispatcher) => dispatcher.flush()
    None => ()
  }
  for observer in self.observers {
    observer.on_event_at(scope, snapshot_turn_event(event))
  }
}

///|
fn AgentEventSubscriber::emit_tool_result(
  self : AgentEventSubscriber,
  scope : @types.EventScope,
  call : @kernel.ToolCall,
  outcome : @kernel.ToolOutcome,
) -> Unit raise @puppetry.SubscriberError {
  let key = call.call_id.to_string()
  if !self.pending_calls.contains(key) {
    raise ContractViolation(
      subscriber_id="agent_committed_event_projection",
      detail="ToolCompleted without committed ToolBatchStarted for call " + key,
    )
  }
  self.pending_calls.remove(key)
  self.emit(Some(scope), @types.TurnEvent::tool_call_result(call, outcome))
}

///|
impl @puppetry.EventSubscriber for AgentEventSubscriber with fn on_event(
  self : AgentEventSubscriber,
  envelope : @puppetry.EventEnvelope,
) -> Unit raise @puppetry.SubscriberError {
  let event = envelope.event()
  // Envelope-projected events always carry the envelope's run identity as
  // their attribution scope.
  let scope : @types.EventScope = {
    session_id: envelope.session_id(),
    run_id: envelope.run_id(),
    turn_id: envelope.turn_id(),
  }
  match event {
    RunStarted(..) => self.pending_calls.clear()
    ModelStepCompleted(completion~, ..) => {
      let payload = completion.message
      let message : @kernel.Message = AssistantMessage(
        content=payload.content,
        tool_calls=payload.tool_calls,
        reasoning=payload.reasoning,
        finish_reason=payload.finish_reason,
      )
      self.emit(
        Some(scope),
        ModelResponseReceived(message~, usage=completion.usage),
      )
    }
    ToolBatchStarted(calls~, ..) =>
      for call in calls {
        let snapshot = snapshot_tool_call(call)
        self.pending_calls[snapshot.call_id.to_string()] = snapshot
        self.emit(Some(scope), ToolCallPending(snapshot))
      }
    ToolCompleted(call~, outcome~) =>
      self.emit_tool_result(scope, call, outcome)
    _ => ()
  }
}

///|
/// Wrap the puppet's `HostChunkCallback` so each `StreamChunk` is re-emitted
/// to observers as `StreamChunkReceived`. The chunk is already the canonical
/// typed shape; no JSON decoding is needed here.
fn create_agent_stream_callback(
  observers : Array[&@port.Observer],
  dispatcher_ref : Ref[ChunkDispatcher?],
) -> @kernel_exec.HostChunkCallback? {
  if observers.is_empty() {
    return None
  }
  let cb = fn(chunk : @types.StreamChunk) {
    // Route to the active turn's dispatcher. If no turn is active (should not
    // happen for chunks produced by the Puppet), drop silently.
    match dispatcher_ref.val {
      Some(dispatcher) => dispatcher.enqueue(chunk)
      None => ()
    }
  }
  Some(cb)
}