mindfork 0.11.0

//! A slot registry for "silent" background tasks (self-model auto-reflection and
//! notes auto-consolidation). Both tasks are a mini agentic loop with no UI (the shared runner
//! [`tool_loop::spawn_silent_loop`](super::tool_loop)); their lifecycle (the "running"
//! flag, a failure streak, clearing the indicator, an error alert) used to be
//! duplicated as fields and handlers per task. Here it's one — the registry key
//! is the existing [`BackgroundKind`]. Adding task #3 (self-model auto-consolidation,
//! roadmap — architecture.md §9.9) doesn't touch the `run()`/`Quit` scaffolding.
//! See docs/history/refactoring-solid.md §4.

use std::sync::Arc;
use std::sync::atomic::{AtomicU8, Ordering};
use std::time::Duration;

use chrono::{DateTime, Utc};
use tokio::sync::mpsc::UnboundedReceiver;
use tokio_util::sync::CancellationToken;
use uuid::Uuid;

use crate::app::events::{AppEvent, BackgroundKind};
use crate::shared::i18n::Locale;

use super::Orchestrator;
use super::compaction::CompactResult;

/// What the quit's settle heard: a silent loop's outcome, or the roll's
/// result on its own channel.
enum Landing {
    Task(BgDone),
    Roll(CompactResult),
}

/// How a silent task ended (`bg_done_tx`): its work done, stopped by its
/// own token — the tasks screen's `F6`, or `Quit` — or failed with a
/// reason worded for the user. A stop is neither a success nor a failure
/// to the streak (docs/research/stop-silent-task.md §3.3); `consumed` says
/// whether a round of the task's tools had run by then, which decides
/// whether the window it advanced at spawn is given back
/// (docs/research/stop-refunds-window.md §3.2).
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) enum BgOutcome {
    Done,
    Cancelled { consumed: bool },
    Failed(String),
}

/// A silent task's landing: the outcome, and beside it the engine's prefill
/// figure — the largest sample of the task's streams, whatever the outcome,
/// since the figure is the engine's fact and not the task's verdict
/// (docs/research/loop-timings.md §3.2); `None` from a stream that ended
/// short (the usage chunk is the stream's last) or a server without
/// timings. Offered to the slow-prefill rule at the landing, once for every
/// kind (§3.3).
#[derive(Debug)]
pub(super) struct BgDone {
    pub(super) kind: BackgroundKind,
    pub(super) outcome: BgOutcome,
    pub(super) prefill: Option<crate::shared::api::contract::Prefill>,
}

/// What a spawn advanced, and how to put it back: reflection's watermark
/// and stamp before the spawn, or the reply count a consolidation's reset
/// took. Kept on the task's slot from the spawn to the landing; given back
/// only when the task was stopped before a round of its tools ran, so the
/// ordinary cadence makes it due again at the next landing — a window the
/// task acted on must not be read twice
/// (docs/research/stop-refunds-window.md §3).
#[derive(Debug, Clone, PartialEq, Eq)]
pub(super) enum Window {
    /// Reflection: the chat's `reflected_upto` and `reflected_at` before the spawn.
    Reflection {
        chat: Uuid,
        upto: Option<usize>,
        at: Option<DateTime<Utc>>,
    },
    /// A consolidation: the chat's reply count the spawn reset to zero.
    Counter { chat: Uuid, count: u32 },
}

/// Where a silent task stands with respect to the window its spawn
/// advanced (docs/research/acted-by-effect.md §3.2): nothing written and no
/// tools running — a quit gives the window back; a round's tools running —
/// a quit keeps it, since the write the round may make has not reported
/// yet; a call wrote — kept for good.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub(super) enum Acting {
    Idle = 0,
    InTools = 1,
    Wrote = 2,
}

/// The [`Acting`] state, shared between the loop that sets it and the slot
/// that reads it at a quit. `SeqCst` on both sides: the quit cancels the
/// token and then reads, the loop stores `InTools` and then checks the
/// token, and that order is what keeps a refunded window unwritten
/// (docs/research/quit-refunds-window.md §3.3).
#[derive(Debug, Default)]
pub(super) struct Acted(AtomicU8);

impl Acted {
    /// A state to start from (the tests' quits: a slot mid-tools, one that
    /// wrote).
    #[cfg(test)]
    pub(super) fn at(state: Acting) -> Self {
        Self(AtomicU8::new(state as u8))
    }

    pub(super) fn set(&self, state: Acting) {
        self.0.store(state as u8, Ordering::SeqCst);
    }

    /// A round's tools are about to run: `InTools` — unless a call already
    /// wrote, since a write is kept for good.
    pub(super) fn enter_tools(&self, wrote_so_far: bool) {
        if !wrote_so_far {
            self.set(Acting::InTools);
        }
    }

    /// A round's tools have reported: a write is kept for good, a round of
    /// reads leaves the task where it was (docs/research/acted-by-effect.md
    /// §3.2). Returns the fact accumulated so far.
    pub(super) fn leave_tools(&self, wrote_so_far: bool, round_wrote: bool) -> bool {
        let wrote = wrote_so_far || round_wrote;
        self.set(if wrote { Acting::Wrote } else { Acting::Idle });
        wrote
    }

    pub(super) fn get(&self) -> Acting {
        match self.0.load(Ordering::SeqCst) {
            0 => Acting::Idle,
            1 => Acting::InTools,
            _ => Acting::Wrote,
        }
    }
}

/// What a stop or a quit can give back, and whether it still may: the
/// [`Window`] the spawn advanced, and the [`Acted`] state the loop keeps —
/// readable outside the loop at any moment, a quit's included, and not
/// only from the outcome the loop sends last
/// (docs/research/quit-refunds-window.md §3.1).
pub(super) struct Refund {
    pub window: Window,
    pub acted: Arc<Acted>,
}

/// A silent background task's slot: the active-run token + a failure streak. The streak
/// outlives a single run (it survives completions) — hence a slot, not a separate task.
#[derive(Default)]
pub(super) struct BgSlot {
    /// `Some` — the task is running (one at a time); the cancellation token (for the `Quit` branch).
    cancel: Option<CancellationToken>,
    /// The count of consecutive failures; at the [`BACKGROUND_FAILURE_ALERT`](super::BACKGROUND_FAILURE_ALERT)
    /// threshold we show a UI error once, then stay quiet until the first success.
    failures: u32,
    /// What the running task's spawn advanced, with the loop's flag
    /// ([`Refund`]); taken at the landing, given back on a stop — or a quit
    /// — before the first round of tools.
    refund: Option<Refund>,
}

impl Orchestrator {
    /// Is a background task of this kind running (the "one at a time" gate).
    pub(super) fn bg_running(&self, kind: BackgroundKind) -> bool {
        self.bg.get(&kind).is_some_and(|s| s.cancel.is_some())
    }

    /// Records a run: the slot is marked active (`cancel = Some`), what the
    /// spawn advanced is kept for a stop or a quit to give back (`refund`;
    /// `None` for the roll, which has nothing to refund), and a quiet
    /// "running …" indicator goes into the status bar. Called by the spawn
    /// tails of `maybe_auto_reflect`/`maybe_auto_consolidate`/`spawn_compact`.
    pub(super) fn begin_bg(
        &mut self,
        kind: BackgroundKind,
        cancel: CancellationToken,
        refund: Option<Refund>,
    ) {
        let slot = self.bg.entry(kind).or_default();
        slot.cancel = Some(cancel);
        slot.refund = refund;
        let _ = self
            .evt_tx
            .send(AppEvent::BackgroundTask { kind, active: true });
        // The tasks screen's "the app's own work" rows (spec §11.10).
        self.emit_task_list();
    }

    /// The shared background-task outcome handler (formerly `handle_reflect_done`/
    /// `handle_consolidate_done`): clears the "running" flag, clears the indicator, tracks the
    /// failure streak (at the threshold — one UI error, observability without spam). On success of
    /// **reflection** or **self-model consolidation**, additionally sends
    /// `SelfModelChanged` (an open `F3` screen re-requests a fresh snapshot);
    /// *notes* consolidation doesn't (it changes notes, not the "self-model"). The task's
    /// tools have already written the changes into `Storage`; this doesn't touch the chat/feed.
    /// A task **stopped** by its own token (docs/research/stop-silent-task.md
    /// §3.3) clears the slot like the others and touches the streak not at
    /// all — neither reset nor counted — and still announces
    /// `SelfModelChanged` for the two self-model kinds, since a partial run
    /// may have written before the stop. Stopped **before a round of its
    /// tools ran**, it also gets the window its spawn advanced back
    /// ([`Window`]; docs/research/stop-refunds-window.md §3.3); on every
    /// other outcome the window is dropped and the advance stands.
    pub(super) fn handle_bg_done(
        &mut self,
        kind: BackgroundKind,
        outcome: BgOutcome,
        prefill: Option<crate::shared::api::contract::Prefill>,
    ) {
        // Mutate the slot and compute whether an error alert is needed BEFORE sending events
        // (the borrow of `self.bg` doesn't overlap `self.evt_tx` in the send below).
        let (alert, window) = {
            let slot = self.bg.entry(kind).or_default();
            slot.cancel = None;
            let window = slot.refund.take().map(|r| r.window);
            let alert = match &outcome {
                BgOutcome::Done => {
                    slot.failures = 0;
                    None
                }
                BgOutcome::Cancelled { .. } => None,
                BgOutcome::Failed(reason) => {
                    slot.failures += 1;
                    (slot.failures == super::BACKGROUND_FAILURE_ALERT).then(|| reason.clone())
                }
            };
            (alert, window)
        };
        if let (Some(window), BgOutcome::Cancelled { consumed: false }) = (window, &outcome) {
            self.give_back(kind, window);
        }
        let _ = self.evt_tx.send(AppEvent::BackgroundTask {
            kind,
            active: false,
        });
        self.emit_task_list();
        if !matches!(outcome, BgOutcome::Failed(_))
            && matches!(
                kind,
                BackgroundKind::Reflection | BackgroundKind::SelfConsolidation
            )
        {
            let _ = self.evt_tx.send(AppEvent::SelfModelChanged);
        }
        if let Some(reason) = alert {
            let loc = self.ui_locale();
            let _ = self.evt_tx.send(AppEvent::Error(loc.tf(
                "ui.err.bg_failed",
                &[("label", kind_label(loc, kind)), ("reason", &reason)],
            )));
        }
        // Every silent task lands here, so the slow-prefill rule is asked here,
        // once — after the task's own landing, so the note reads as a footnote
        // to it (docs/research/loop-timings.md §3.3). The figure is the
        // engine's whatever the outcome; the rule's one claim per server
        // session decides whether anything is said.
        self.note_slow_prefill(prefill);
    }

    /// Puts back what a spawn advanced (docs/research/stop-refunds-window.md
    /// §3.3): the task was stopped before a round of its tools ran, so the
    /// window it was about to read is unread, and the ordinary cadence makes
    /// it due again at the next landing. Reflection's watermark and stamp
    /// are restored and the chat saved the way the advance was; a chat gone
    /// meanwhile is left alone. A counter is **added back**, since the
    /// landings during the run incremented it legitimately and the sum is
    /// what it would read had the spawn never happened.
    fn give_back(&mut self, kind: BackgroundKind, window: Window) {
        match window {
            Window::Reflection { chat, upto, at } => {
                let found = self.chats.iter_mut().find(|c| c.id == chat).map(|c| {
                    c.reflected_upto = upto;
                    c.reflected_at = at;
                });
                if found.is_some() {
                    self.mark_dirty(chat);
                }
            }
            Window::Counter { chat, count } => {
                let counts = match kind {
                    BackgroundKind::Consolidation => &mut self.consolidate_counts,
                    BackgroundKind::SelfConsolidation => &mut self.self_consolidate_counts,
                    // Neither keeps a counter; a spawn never records one for them.
                    BackgroundKind::Reflection | BackgroundKind::Compaction => return,
                };
                *counts.entry(chat).or_insert(0) += count;
            }
        }
    }

    /// Stops one running task of `kind` (`AppCommand::StopBackgroundTask`:
    /// the tasks screen's `F6` on its row, docs/research/stop-silent-task.md
    /// §3.2): its token is cancelled and it lands as `Cancelled` on its own
    /// path, at once — a wait returns, a stream ends on its next chunk. A
    /// kind with no task running is ignored: the screen may be a snapshot
    /// behind.
    pub(super) fn handle_stop_background_task(&self, kind: BackgroundKind) {
        if let Some(token) = self.bg.get(&kind).and_then(|s| s.cancel.as_ref()) {
            token.cancel();
        }
    }

    /// The `Quit` branch's first step: cancels every running task in the
    /// family and nothing else — the refunds stay on the slots for the
    /// landings to decide (docs/research/quit-waits-for-the-landing.md
    /// §3.1). The token is cancelled **before** any state is read: a loop
    /// that stores `InTools` after that read checks the token before its
    /// tools and starts none, so a refunded window is never written into
    /// (docs/research/quit-refunds-window.md §3.3).
    pub(super) fn cancel_bg_all(&self) {
        for slot in self.bg.values() {
            if let Some(token) = &slot.cancel {
                token.cancel();
            }
        }
    }

    /// Whether any silent task's slot is still taken — a landing has not
    /// cleared it.
    fn any_bg_active(&self) -> bool {
        self.bg.values().any(|s| s.cancel.is_some())
    }

    /// The `Quit` branch's second step, after `run`'s loop has broken:
    /// listens on the tasks' outcome channels a little longer, so that every
    /// cancelled task's own landing — a wait returned, a stream ended, a
    /// round of tools finished — decides its window through the very path a
    /// stop takes (`handle_bg_done`, `consumed` from the loop). The roll
    /// lands on its own channel, through `handle_compact_result`: a cancelled
    /// roll clears its slot at once, a roll that finished just before the
    /// quit is applied for the flush (docs/research/quit-settle-roll-and-cap.md
    /// §3.1). Over as soon as no slot is active, or — with a `cap` — when it
    /// runs out, leaving the rest to [`Self::refund_unlanded`]; `None` waits
    /// for every landing, each task bounded by its own run time limit
    /// (docs/research/quit-waits-for-the-landing.md §3.1).
    pub(super) async fn settle_silent_tasks(
        &mut self,
        done_rx: &mut UnboundedReceiver<BgDone>,
        compact_rx: &mut UnboundedReceiver<CompactResult>,
        cap: Option<Duration>,
    ) {
        let deadline = cap.map(|cap| tokio::time::Instant::now() + cap);
        while self.any_bg_active() {
            let next = async {
                tokio::select! {
                    landed = done_rx.recv() => landed.map(Landing::Task),
                    result = compact_rx.recv() => result.map(Landing::Roll),
                }
            };
            let landed = match deadline {
                Some(deadline) => tokio::time::timeout_at(deadline, next).await.ok(),
                None => Some(next.await),
            };
            match landed {
                Some(Some(Landing::Task(d))) => self.handle_bg_done(d.kind, d.outcome, d.prefill),
                Some(Some(Landing::Roll(result))) => self.handle_compact_result(result),
                // A channel closed, or the cap ran out.
                Some(None) | None => break,
            }
        }
    }

    /// The `Quit` branch's last step: the window of every task that did not
    /// land within the cap is decided by its state — `Idle` (nothing
    /// written, no round of tools running) given back, `InTools` and
    /// `Wrote` kept, since the round may be writing or has
    /// (docs/research/acted-by-effect.md §3.2). The exit flush after this
    /// writes what `give_back` marked dirty.
    pub(super) fn refund_unlanded(&mut self) {
        let refunds: Vec<(BackgroundKind, Window)> = self
            .bg
            .iter_mut()
            .filter_map(|(kind, slot)| {
                let refund = slot.refund.take()?;
                (refund.acted.get() == Acting::Idle).then_some((*kind, refund.window))
            })
            .collect();
        for (kind, window) in refunds {
            self.give_back(kind, window);
        }
    }

    /// The task's consecutive-failure count (for error-alert tests).
    #[cfg(test)]
    pub(super) fn bg_failures(&self, kind: BackgroundKind) -> u32 {
        self.bg.get(&kind).map_or(0, |s| s.failures)
    }
}

/// The silent lane's label for a task kind (`SessionBudget::acquire_silent`,
/// docs/research/silent-tasks-budget.md §4.6): what the budget reports as
/// streaming, and what the tasks screen's snapshot compares against to say
/// which running task is *waiting*. Stable identifiers, never shown.
pub(super) fn lane_label(kind: BackgroundKind) -> &'static str {
    match kind {
        BackgroundKind::Reflection => "reflection",
        BackgroundKind::Consolidation => "consolidation",
        BackgroundKind::SelfConsolidation => "self_consolidation",
        BackgroundKind::Compaction => "compaction",
    }
}

/// A human-readable label for the task kind (the interface language, axis B) — error
/// texts are assembled from it (**byte-for-byte** with the previous Russian wording).
fn kind_label(loc: &'static Locale, kind: BackgroundKind) -> &'static str {
    loc.t(match kind {
        BackgroundKind::Reflection => "ui.err.bg_reflection",
        BackgroundKind::Consolidation => "ui.err.bg_consolidation",
        BackgroundKind::SelfConsolidation => "ui.err.bg_self_consolidation",
        BackgroundKind::Compaction => "ui.err.bg_compaction",
    })
}