//! Policy-checked, observable orchestrator → sub-agent steering.
//!
//! In the recursive architecture this is how a *parent* in the run tree exerts
//! typed control over a *child* it is currently running: an orchestrating agent
//! (or human, or graph supervisor) injects [`SteeringCommand`]s into a live
//! sub-agent loop without killing or restarting it. It is the mid-run
//! counterpart to durable child-session reuse in `tinyagents-orchestration` (which
//! resumes a *completed* child) — together they cover both ways an orchestrator
//! keeps a sub-agent "in play".
//!
//! Steering lets an orchestrator (a parent agent, a human UI, a graph
//! supervisor, or a test harness) guide an already-running agent loop without
//! breaking its run identity or observability. The flow is:
//!
//! 1. The orchestrator builds a [`SteeringPolicy`] allowlist and a
//!    [`SteeringHandle`], and attaches the handle to the run's
//!    [`RunContext`] via
//!    [`RunContext::with_steering`][crate::context::RunContext::with_steering].
//! 2. While the run executes, the orchestrator calls
//!    [`SteeringHandle::send`] to enqueue [`SteeringCommand`]s.
//! 3. The agent loop, at a safe checkpoint (before each model call), calls
//!    [`apply_pending_steering`] which drains the handle, checks each command
//!    against the policy, applies the permitted ones, and emits an
//!    [`AgentEvent::Steered`] for every command.
//!
//! Delivery is conservative and pull-based: commands become visible only at the
//! checkpoint, never in the middle of a provider stream or a side-effecting
//! tool call.
//!
//! # Example
//!
//! ```
//! use tinyagents_harness::context::{RunConfig, RunContext};
//! use tinyinference_llm::message::Message;
//! use tinyagents_harness::steering::{
//!     apply_pending_steering, SteeringCommand, SteeringCommandKind, SteeringHandle,
//!     SteeringOutcome, SteeringPolicy,
//! };
//!
//! let policy = SteeringPolicy::new().allow(SteeringCommandKind::InjectMessage);
//! let handle = SteeringHandle::new(policy);
//! handle.send(SteeringCommand::InjectMessage(Message::user("focus on billing")));
//!
//! let mut ctx: RunContext = RunContext::new(RunConfig::new("run-1"), ())
//!     .with_steering(handle.clone());
//!
//! let mut messages = vec![Message::user("start")];
//! let outcome = apply_pending_steering(&mut ctx, &mut messages).unwrap();
//! assert_eq!(outcome, SteeringOutcome::Continue);
//! // The injected instruction is now visible to the next model call.
//! assert_eq!(messages.len(), 2);
//! ```

mod types;

pub use types::*;

use std::collections::{HashSet, VecDeque};
use std::sync::{Arc, Mutex};

use crate::context::RunContext;
use crate::error::Result;
use crate::events::AgentEvent;
use crate::ids::RunId;
use tinyinference_llm::message::Message;

// ── SteeringPolicy ────────────────────────────────────────────────────────────

impl SteeringPolicy {
    /// Creates an empty policy that permits **no** steering commands.
    ///
    /// Grant kinds explicitly with [`SteeringPolicy::allow`] or start from
    /// [`SteeringPolicy::allow_all`].
    pub fn new() -> Self {
        Self {
            allowed: HashSet::new(),
        }
    }

    /// Creates a policy that permits every [`SteeringCommandKind`] **except**
    /// [`SteeringCommandKind::SwitchModel`].
    ///
    /// A model switch changes what the run costs and which provider receives
    /// the transcript, so it is never granted implicitly: use
    /// [`SteeringPolicy::allow_all_with_model_switch`] or
    /// `.allow(SteeringCommandKind::SwitchModel)` to opt in.
    pub fn allow_all() -> Self {
        Self {
            allowed: SteeringCommandKind::ALL
                .into_iter()
                .filter(|kind| *kind != SteeringCommandKind::SwitchModel)
                .collect(),
        }
    }

    /// Creates a policy that permits every [`SteeringCommandKind`], including
    /// [`SteeringCommandKind::SwitchModel`]. Only hand this to an orchestrator
    /// you trust with the run's spend and data residency.
    pub fn allow_all_with_model_switch() -> Self {
        Self::allow_all().allow(SteeringCommandKind::SwitchModel)
    }

    /// Adds `kind` to the allowlist (builder style) and returns the policy.
    pub fn allow(mut self, kind: SteeringCommandKind) -> Self {
        self.allowed.insert(kind);
        self
    }

    /// Returns `true` when `kind` is permitted by this policy.
    pub fn is_allowed(&self, kind: SteeringCommandKind) -> bool {
        self.allowed.contains(&kind)
    }
}

// ── SteeringHandle ──────────────────────────────────────────────────────────

impl SteeringHandle {
    /// Builds a handle backed by a fresh, empty queue gated by `policy`.
    ///
    /// The handle is unbound (empty `run_id`, `is_root = true`) until it is
    /// attached to a run via
    /// [`RunContext::with_steering`][crate::context::RunContext::with_steering],
    /// which binds it to that run's id as the root of its steering tree.
    pub fn new(policy: SteeringPolicy) -> Self {
        Self {
            inner: Arc::new(SteeringInner {
                queue: Mutex::new(VecDeque::new()),
                policy,
            }),
            run_id: RunId::new(""),
            is_root: true,
            local: Arc::new(SteeringLocal::default()),
        }
    }

    /// Convenience constructor for a handle whose policy permits every command
    /// kind. Equivalent to `SteeringHandle::new(SteeringPolicy::allow_all())`.
    pub fn allow_all() -> Self {
        Self::new(SteeringPolicy::allow_all())
    }

    /// Convenience constructor for a handle whose policy is
    /// [`SteeringPolicy::allow_all_with_model_switch`].
    pub fn allow_all_with_model_switch() -> Self {
        Self::new(SteeringPolicy::allow_all_with_model_switch())
    }

    /// Binds this handle to `run_id` as the **root** of its steering tree.
    ///
    /// Called by [`RunContext::with_steering`][crate::context::RunContext::with_steering]
    /// when an orchestrator attaches a handle to a run; every
    /// [`SteeringTarget::Root`]-addressed command drains here.
    ///
    /// A model switch is scoped to one run, but `local` is shared with the
    /// handle being bound, so a switch left over from an earlier run on the
    /// same handle is cleared here: the new run starts on its own model.
    pub(crate) fn bind_root(&self, run_id: RunId) -> Self {
        self.reject_model_override();
        Self {
            inner: Arc::clone(&self.inner),
            run_id,
            is_root: true,
            local: Arc::clone(&self.local),
        }
    }

    /// Derives a handle scoped to a child run.
    ///
    /// Shares the underlying queue and policy (so an orchestrator holding the
    /// root handle can still reach the child by [`SteeringTarget::Run`] or
    /// [`SteeringTarget::All`]), but gets its own identity and its own
    /// pause/checkpoint state: a command addressed to the parent (or to
    /// [`SteeringTarget::Root`]) is invisible to [`SteeringHandle::drain`] on
    /// the child, and a pause latched on the child does not latch the parent's.
    /// This is what keeps an `Inject`/`Pause` meant for the orchestrator from
    /// being consumed by whichever sub-agent happens to reach a checkpoint
    /// first (see I-5).
    pub(crate) fn for_child(&self, run_id: RunId) -> Self {
        Self {
            inner: Arc::clone(&self.inner),
            run_id,
            is_root: false,
            local: Arc::new(SteeringLocal::default()),
        }
    }

    /// Enqueues `command` addressed to [`SteeringTarget::Root`].
    ///
    /// The command becomes visible to the loop at its next steering checkpoint;
    /// this method never blocks and does not itself check the policy. Use
    /// [`SteeringHandle::send_to`] to address a specific descendant run, or
    /// [`SteeringHandle::send_all`] to reach every run sharing this handle.
    pub fn send(&self, command: SteeringCommand) {
        self.send_to(SteeringTarget::Root, command);
    }

    /// Enqueues `command` addressed to `target`.
    ///
    /// Queue accessors recover from a poisoned mutex (a panic in another
    /// holder) instead of panicking: the queue is a plain `VecDeque` with no
    /// invariants that a panicking holder could break mid-update.
    pub fn send_to(&self, target: SteeringTarget, command: SteeringCommand) {
        self.inner
            .queue
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .push_back((target, command));
    }

    /// Enqueues `command` addressed to every run sharing this handle
    /// ([`SteeringTarget::All`]).
    pub fn send_all(&self, command: SteeringCommand) {
        self.send_to(SteeringTarget::All, command);
    }

    /// Removes one queued command when its exact target and payload match.
    ///
    /// This is used by producers that must retract work which was queued for
    /// a run but could not be completed. Other commands, including commands
    /// addressed to different runs, remain in FIFO order.
    pub fn retract(&self, target: &SteeringTarget, command: &SteeringCommand) -> bool {
        let mut queue = self
            .inner
            .queue
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let Some(index) = queue.iter().position(|(queued_target, queued_command)| {
            queued_target == target && queued_command == command
        }) else {
            return false;
        };
        queue.remove(index);
        true
    }

    /// Removes and returns the commands addressed to *this* handle's run (its
    /// own [`SteeringTarget::Run`], [`SteeringTarget::Root`] if this handle is
    /// the root, or [`SteeringTarget::All`]), leaving commands addressed to
    /// other runs in the shared queue for them to drain later. Called by the
    /// agent loop at each checkpoint.
    pub fn drain(&self) -> Vec<SteeringCommand> {
        let mut queue = self
            .inner
            .queue
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let mut matched = Vec::new();
        let mut remaining = VecDeque::with_capacity(queue.len());
        for (target, command) in queue.drain(..) {
            if self.matches(&target) {
                matched.push(command);
            } else {
                remaining.push_back((target, command));
            }
        }
        *queue = remaining;
        matched
    }

    /// Returns `true` when `target` addresses this handle's run.
    fn matches(&self, target: &SteeringTarget) -> bool {
        match target {
            SteeringTarget::Root => self.is_root,
            SteeringTarget::Run(id) => *id == self.run_id,
            SteeringTarget::All => true,
        }
    }

    /// Returns `true` when no commands addressed to this handle's run are
    /// currently queued.
    pub fn is_empty(&self) -> bool {
        self.pending() == 0
    }

    /// Returns the number of commands currently queued that are addressed to
    /// this handle's run.
    pub fn pending(&self) -> usize {
        self.inner
            .queue
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .iter()
            .filter(|(target, _)| self.matches(target))
            .count()
    }

    /// Returns the policy gating this handle.
    pub fn policy(&self) -> &SteeringPolicy {
        &self.inner.policy
    }

    /// Returns the latched [`PauseState`] when the run is paused, `None`
    /// otherwise.
    ///
    /// This is how a caller distinguishes a run that stopped for a human from a
    /// run that finished with nothing to say: on a
    /// [`SteeringOutcome::Pause`] this is always `Some`.
    pub fn pause_state(&self) -> Option<PauseState> {
        self.lock_paused().clone()
    }

    /// Returns `true` when a pause is latched.
    pub fn is_paused(&self) -> bool {
        self.lock_paused().is_some()
    }

    /// Latches a pause with an optional reason, recorded as having taken effect
    /// at `checkpoint`. Idempotent: an existing pause keeps its original reason
    /// and checkpoint, so a repeated `Pause` does not rewrite why the run
    /// stopped.
    fn latch_pause(&self, checkpoint: usize, reason: Option<String>) -> PauseState {
        let mut paused = self.lock_paused();
        let state = paused.get_or_insert(PauseState {
            reason,
            paused_at_checkpoint: checkpoint,
        });
        tracing::debug!(
            target: "tinyagents::steering",
            checkpoint = state.paused_at_checkpoint,
            reason = state.reason.as_deref(),
            "[steering] pause latched"
        );
        state.clone()
    }

    /// Clears any latched pause, returning the state that was cleared.
    ///
    /// Equivalent to delivering a [`SteeringCommand::Resume`]; exposed directly
    /// so a host UI can resume without going through the queue. A no-op when no
    /// pause is in effect.
    pub fn resume(&self) -> Option<PauseState> {
        let cleared = self.lock_paused().take();
        if cleared.is_some() {
            tracing::debug!(target: "tinyagents::steering", "[steering] pause cleared by resume");
        }
        cleared
    }

    /// Returns this checkpoint's zero-based index and advances the counter, so
    /// the *current* checkpoint is what a pause records (not the next one).
    fn advance_checkpoint(&self) -> usize {
        let mut checkpoints = self
            .local
            .checkpoints
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        let current = *checkpoints;
        *checkpoints += 1;
        current
    }

    /// The model a [`SteeringCommand::SwitchModel`] selected for this run, if
    /// any. Read by the agent loop before it resolves a model binding; the
    /// loop validates the name against its registry and calls
    /// `SteeringHandle::reject_model_override` when it cannot honour it.
    pub fn model_override(&self) -> Option<String> {
        self.lock_model_override().clone()
    }

    /// Records the model a `SwitchModel` asked for, replacing any earlier one.
    fn set_model_override(&self, model: String) -> bool {
        let mut slot = self.lock_model_override();
        let superseded = slot.is_some()
            && !self
                .local
                .model_override_announced
                .load(std::sync::atomic::Ordering::SeqCst);
        *slot = Some(model);
        self.local
            .model_override_announced
            .store(false, std::sync::atomic::Ordering::SeqCst);
        superseded
    }

    /// Marks the current model override as reported, returning `true` only the
    /// first time since it was set -- the caller emits the one
    /// `Steered { accepted: true }` for the command exactly then.
    pub(crate) fn announce_model_override(&self, applied: &str) -> bool {
        // Held across the flag claim so it is atomic with `set_model_override`
        // replacing the override and resetting the flag.
        let slot = self.lock_model_override();
        slot.as_deref() == Some(applied)
            && !self
                .local
                .model_override_announced
                .swap(true, std::sync::atomic::Ordering::SeqCst)
    }

    /// Drops a model override the model call could not honour, so the run
    /// keeps its current model and the name is not re-checked on every call.
    ///
    /// Returns `true` when the override had already been reported as applied
    /// (`Steered { accepted: true }`), so the caller must not emit a second,
    /// contradictory outcome for the same command.
    pub(crate) fn reject_model_override(&self) -> bool {
        let mut slot = self.lock_model_override();
        slot.take();
        self.local
            .model_override_announced
            .swap(false, std::sync::atomic::Ordering::SeqCst)
    }

    /// Locks the model override, recovering from poisoning.
    fn lock_model_override(&self) -> std::sync::MutexGuard<'_, Option<String>> {
        self.local
            .model_override
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }

    /// Locks the pause latch, recovering from poisoning (see
    /// [`SteeringHandle::send`]).
    fn lock_paused(&self) -> std::sync::MutexGuard<'_, Option<PauseState>> {
        self.local
            .paused
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
    }
}

// ── Checkpoint application ────────────────────────────────────────────────────

/// Drains any steering commands attached to `ctx`, applies the policy-permitted
/// ones to the working `messages` (and run metadata), and returns the resulting
/// control-flow [`SteeringOutcome`].
///
/// This is the single steering checkpoint used by the agent loop. It is also a
/// standalone, synchronous function so it can be unit-tested without a full
/// run. Behaviour:
///
/// - When `ctx` has no [`SteeringHandle`], returns
///   [`SteeringOutcome::Continue`] without emitting anything.
/// - Only commands addressed to this run are drained: see
///   [`SteeringHandle::drain`] and `SteeringHandle::for_child`. Commands
///   addressed to a different run stay queued for it.
/// - Each command is checked **individually** against the run's
///   [`SteeringPolicy`]. A disallowed command is rejected on its own — an
///   [`AgentEvent::Steered`] with `accepted = false` is emitted for it, and the
///   checkpoint moves on to the next command in the batch — rather than
///   aborting the whole batch or the run. (It used to reject the entire batch,
///   including commands the policy *did* permit, whenever one command in it
///   was disallowed.)
/// - [`SteeringCommand::Cancel`] takes precedence: it is applied (emitting an
///   accepted event) and the function returns [`SteeringOutcome::Cancel`]
///   immediately, ignoring the rest of the batch.
/// - [`SteeringCommand::Pause`] / [`SteeringCommand::PauseWith`] latch a pause
///   **on the handle**, so it survives past this batch. Any later
///   [`SteeringCommand::Resume`] — in this batch or a subsequent one — clears
///   it. While latched, every checkpoint returns [`SteeringOutcome::Pause`]
///   even with an empty queue.
/// - [`SteeringCommand::SwitchModel`] records the requested model on the
///   handle ([`SteeringHandle::model_override`]); the agent loop validates and
///   applies it before resolving the next model call. A blank name is
///   rejected here (`accepted: false`). A queued switch emits **no**
///   `Steered` event at the checkpoint: its single outcome is reported when
///   the model call applies it (`accepted: true`) or rejects it
///   (`accepted: false`).
/// - [`SteeringCommand::InjectMessage`] and [`SteeringCommand::Redirect`]
///   append to `messages`; [`SteeringCommand::SetMetadata`] replaces
///   `ctx.config.metadata`.
///
/// # Errors
///
/// This function no longer errors on a policy-disallowed command — see above.
/// It returns `Err` only if a future extension needs to signal a checkpoint
/// failure that is not representable as a rejected command.
pub fn apply_pending_steering<Ctx>(
    ctx: &mut RunContext<Ctx>,
    messages: &mut Vec<Message>,
) -> Result<SteeringOutcome> {
    // Clone the Arc-backed handle out so we do not hold a borrow of `ctx`
    // while we mutate its config/metadata below.
    let Some(handle) = ctx.steering.clone() else {
        return Ok(SteeringOutcome::Continue);
    };
    let checkpoint = handle.advance_checkpoint();
    let commands = handle.drain();

    tracing::debug!(
        target: "tinyagents::steering",
        checkpoint,
        batch_size = commands.len(),
        already_paused = handle.is_paused(),
        "[steering] applying checkpoint batch"
    );
    for command in commands {
        let kind = command.kind();

        // Each command is validated on its own: a disallowed command is
        // rejected individually (I-5/M-7) rather than voiding the whole
        // batch or killing the run.
        if !handle.policy().is_allowed(kind) {
            tracing::debug!(
                target: "tinyagents::steering",
                checkpoint,
                command_kind = kind.as_str(),
                "[steering] command rejected by policy; skipped"
            );
            ctx.emit(AgentEvent::Steered {
                command_kind: kind.as_str().to_string(),
                accepted: false,
            });
            continue;
        }

        match command {
            SteeringCommand::Pause => {
                handle.latch_pause(checkpoint, None);
            }
            SteeringCommand::PauseWith { reason } => {
                handle.latch_pause(checkpoint, Some(reason));
            }
            SteeringCommand::Resume => {
                handle.resume();
            }
            SteeringCommand::Cancel => {
                ctx.emit(AgentEvent::Steered {
                    command_kind: kind.as_str().to_string(),
                    accepted: true,
                });
                // Cancel wins over everything else in the batch.
                return Ok(SteeringOutcome::Cancel);
            }
            SteeringCommand::InjectMessage(message) => messages.push(message),
            SteeringCommand::Redirect { instruction } => {
                messages.push(Message::system(format!(
                    "[steering:redirect] {instruction}"
                )));
            }
            SteeringCommand::SetMetadata { metadata } => {
                ctx.config.metadata = metadata;
            }
            SteeringCommand::SwitchModel { model } => {
                let model = model.trim().to_owned();
                if model.is_empty() {
                    tracing::debug!(
                        target: "tinyagents::steering",
                        checkpoint,
                        "[steering] switch_model rejected: blank model name"
                    );
                    ctx.emit(AgentEvent::Steered {
                        command_kind: kind.as_str().to_string(),
                        accepted: false,
                    });
                    continue;
                }
                tracing::debug!(
                    target: "tinyagents::steering",
                    checkpoint,
                    model = %model,
                    "[steering] model switch queued for the next model call"
                );
                let superseded = handle.set_model_override(model);
                if superseded {
                    ctx.emit(AgentEvent::Steered {
                        command_kind: kind.as_str().to_string(),
                        accepted: false,
                    });
                }
                // Queuing is not an outcome: the agent loop reports the one
                // `Steered` event for this command when it applies or rejects
                // the switch at the model-call boundary.
                continue;
            }
        }

        ctx.emit(AgentEvent::Steered {
            command_kind: kind.as_str().to_string(),
            accepted: true,
        });
    }

    // The latch — not this batch — decides the outcome, so a pause applied at an
    // earlier checkpoint keeps holding the run.
    if handle.is_paused() {
        Ok(SteeringOutcome::Pause)
    } else {
        Ok(SteeringOutcome::Continue)
    }
}

/// Longest steer text (in bytes) a parent agent may send a child, checked
/// where agent-origin steers are produced. Matches the session ledger's
/// `MAX_TOOL_OUTPUT_BYTES`: a steer is an instruction, not a payload channel.
pub const MAX_STEER_TEXT_BYTES: usize = 32 * 1024;

/// Validation failure from [`RecentRequestIds::claim`].
#[non_exhaustive]
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum RequestIdError {
    /// The identifier exceeds [`RecentRequestIds::MAX_REQUEST_ID_BYTES`].
    TooLong,
}

impl RecentRequestIds {
    /// Number of ids remembered per task by [`Self::default`].
    pub const DEFAULT_CAPACITY: usize = 64;

    /// Longest `request_id` (in bytes) callers should accept; bounds the
    /// memory a task's id window can hold.
    pub const MAX_REQUEST_ID_BYTES: usize = 128;

    /// Creates a memory holding at most `capacity` ids (minimum one).
    pub fn with_capacity(capacity: usize) -> Self {
        Self {
            capacity: capacity.max(1),
            order: VecDeque::new(),
        }
    }

    /// Records `request_id`, returning `true` when it is new and `false` when
    /// it was already remembered (a duplicate). The oldest id is forgotten once
    /// the memory is full.
    pub fn claim(&mut self, request_id: &str) -> std::result::Result<bool, RequestIdError> {
        if request_id.len() > Self::MAX_REQUEST_ID_BYTES {
            return Err(RequestIdError::TooLong);
        }
        if self.order.iter().any(|seen| seen == request_id) {
            return Ok(false);
        }
        if self.order.len() >= self.capacity {
            self.order.pop_front();
        }
        self.order.push_back(request_id.to_owned());
        Ok(true)
    }
}

impl Default for RecentRequestIds {
    fn default() -> Self {
        Self::with_capacity(Self::DEFAULT_CAPACITY)
    }
}

#[cfg(test)]
#[path = "mod_tests.rs"]
mod test;