mahbot 0.7.3

//! Turso-backed log storage.
//!
//! Each log entry is inserted asynchronously via a background channel task.
//! A broadcast channel feeds live log entries to the Iced native GUI dashboard.

use crate::db;
use crate::util::UnwrapPoison;
use crate::util::json;
use anyhow::Context;
use db::{Row, Value, params};
use futures_util::FutureExt;
use serde::{Deserialize, Serialize};
use std::io;
use std::panic::AssertUnwindSafe;
use std::path::Path;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use tokio::sync::OnceCell;
use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender};
use tracing::level_filters::LevelFilter;
use tracing_subscriber::fmt::MakeWriter;
use tracing_subscriber::{EnvFilter, Layer, fmt, layer::SubscriberExt};

/// Schema for a single log entry stored in Turso.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct LogEntry {
    pub timestamp: String,
    pub level: String,
    pub target: String,
    pub message: String,
    #[serde(default)]
    pub fields: serde_json::Value,
    #[serde(default)]
    pub agent_id: String,
    #[serde(default)]
    pub agent_role: String,
    #[serde(default)]
    pub workspace: String,
}

// Column definitions for `logs` SELECT queries.
crate::columns! {
    LOGS_COLUMNS [LOGS] {
        TIMESTAMP   => "timestamp",
        LEVEL       => "level",
        TARGET      => "target",
        MESSAGE     => "message",
        FIELDS      => "fields",
        AGENT_ID    => "agent_id",
        AGENT_ROLE  => "agent_role",
        WORKSPACE   => "workspace",
    }
}

/// Turso-backed log store.
///
/// NOTE: This store does NOT use `define_store!`, but does use `crate::columns!`.
/// The store is opened manually inside [`init_tracing()`] because bootstrapping order
/// requires logs to be available before other stores are initialized. See
/// [`LOG_STORE`] for details.
#[derive(Clone, Debug)]
pub struct LogStore {
    pub(crate) conn: crate::db::Connection,
}

/// Global log store, set during [`init_tracing()`].
///
/// # Access model
///
/// This store is initialized inside [`init_tracing()`] — it does NOT have an
/// `init_global()` like other stores. Do NOT add one. Calling `init_tracing()`
/// already opens `logs.db`. A second open via `init_global()` would create a
/// second connection to the same database, causing WAL coordination
/// conflicts between the two connections.
///
/// In addition to this global, [`crate::gui::BOOT_LOG_STORE`] holds another clone of
/// the same `LogStore`, and [`init_tracing()`] returns a third `Arc<LogStore>`
/// to its caller. All three point to the same underlying connection (which
/// is cheaply cloneable since `Connection` wraps an `Arc` internally).
pub static LOG_STORE: OnceCell<LogStore> = OnceCell::const_new();

/// One greppable shell call's grep-engine decision, persisted to the dedicated
/// `grep_telemetry` table (self-contained — no tool_calls.id correlation).
/// Bundled into a struct so the writer's signature stays shallow; the caller
/// constructs the row from the serve outcomes + the runtime facts (actual apply,
/// sentinel re-run, duration, exit). Every platform's grep engine reports
/// telemetry (the interception runs on both shells).
#[derive(Debug)]
pub(crate) struct GrepTelemetryRow<'a> {
    pub command: &'a str,
    pub served: bool,
    pub reason: &'a str,
    pub recursive: bool,
    pub piped: bool,
    pub operand_count: usize,
    pub flags: &'a str,
    pub mode: &'a str,
    pub workspace: &'a str,
    pub grep_count: usize,
    pub served_count: usize,
    pub skipped_count: usize,
    pub duration_ms: Option<i64>,
    pub exit_code: Option<i32>,
}

/// The facts of one event that outlived the call that produced it — the process
/// a command or a chrome-use call left running — recorded by
/// [`record_stray_event`] as ONE durable row in the log store.
///
/// A call that ran in a browser chrome-use started itself needs no record of its
/// own: that call FAILS, so its failure text — the browser note the agent is
/// shown — is already its durable record.
pub(crate) struct StrayEvent<'a> {
    /// The surface that reported it ([`STRAY_SOURCE_SHELL`],
    /// [`STRAY_SOURCE_CHROME_TOOL`]) — the log entry's `target`.
    pub source: &'a str,
    /// What outlived the call, stated plainly — the row's `message`, so the row
    /// says what happened rather than reading as a bare command. Metadata only as
    /// far as shells are concerned: no raw command lines, no page content, no
    /// credentials.
    pub message: &'a str,
    /// The stop recipe for what was left running, carried verbatim into the row's
    /// `fields` as `detail`. The record is a record: it carries the recipe even
    /// where the call's own note offers none — a ReadOnly shell call records it
    /// too, because it is the mode's own guard, not the recipe, that keeps it out
    /// of that mode's note.
    pub recipe: &'a str,
    /// What must be addressed to stop it (a shell leftover's process group /
    /// process tree root, or the chrome-use session whose helper was left
    /// behind); empty when unknown. For a chrome leftover it IS the session whose
    /// helper was left behind — the same name the row's `session` below carries —
    /// so that surface names both by definition.
    pub scope: &'a str,
    /// The physical chrome-use session driven; empty for the shell.
    pub session: &'a str,
    /// The workspace the call ran in — empty only where the surface has none.
    pub workspace: &'a str,
    /// How long the recorded call took, when the surface measured it.
    pub duration_ms: Option<i64>,
}

/// The durable record's `target` — the surface that reported the event.
pub(crate) const STRAY_SOURCE_SHELL: &str = "shell";
pub(crate) const STRAY_SOURCE_CHROME_TOOL: &str = "chrome-tool";

impl LogStore {
    /// Open (or create) the log database at `root/db/logs.db`.
    ///
    /// `pub(crate)` (matching every other store's generated `open`) so tests in
    /// other modules can create a real log store via [`crate::open_test_store!`].
    ///
    /// An existing logs store must be usable — [`crate::db::open_store`] refuses
    /// otherwise, exactly like the main store (the refusal covers both physical
    /// stores). A missing file is a first launch and is created here.
    pub(crate) async fn open(root: &Path) -> anyhow::Result<Self> {
        let conn = crate::db::open_store(root, "logs", "").await?;
        // The catalog owns the logs schema; a catalog failure is a hard boot failure.
        crate::db::migrations::run_migrations(&conn, crate::db::migrations::TargetDb::Logs).await?;
        Ok(Self { conn })
    }

    /// Insert a batch of log entries in a single transaction.
    ///
    /// Each entry would otherwise commit (and fsync the WAL) individually —
    /// in WAL mode the per-commit fsync dominates the insert cost. Batching
    /// the diagnostics logs into one transaction reduces N commits to one.
    /// On failure the whole batch is dropped (the caller's [`spawn_log_writer`]
    /// clears it regardless) — log entries are diagnostics, not durable state. A
    /// caller that cannot lose a row — the partly-finished-update record
    /// `self_update::record_update_unfinished` writes — inserts here directly instead.
    pub(crate) async fn insert_batch(&self, entries: &[LogEntry]) -> anyhow::Result<()> {
        if entries.is_empty() {
            return Ok(());
        }
        let tx = self
            .conn
            .begin_tx()
            .await
            .context("Failed to begin log insert transaction")?;
        for entry in entries {
            tx.execute(
                "INSERT INTO logs (timestamp, level, target, message, fields, agent_id, agent_role, workspace) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
                params![
                    entry.timestamp.clone(),
                    entry.level.clone(),
                    entry.target.clone(),
                    entry.message.clone(),
                    serde_json::to_string(&entry.fields)
                        .expect("log entry fields serialization failed; this should not happen"),
                    entry.agent_id.clone(),
                    entry.agent_role.clone(),
                    entry.workspace.clone(),
                ],
            )
            .await
            .context("Failed to insert log entry in batch")?;
        }
        tx.commit()
            .await
            .context("Failed to commit log insert transaction")?;
        Ok(())
    }

    /// Delete log entries matching a given `level` whose `timestamp` is older than the given
    /// RFC 3339 `cutoff`. Returns the number of deleted rows.
    pub async fn delete_older_than(&self, level: &str, cutoff: &str) -> anyhow::Result<u64> {
        let n = self
            .conn
            .execute(
                "DELETE FROM logs WHERE level = ?1 AND timestamp < ?2",
                params![level, cutoff],
            )
            .await
            .context("Failed to delete old log entries")?;
        Ok(n)
    }

    /// Delete stored log records in one Logs tab's scope, returning the number
    /// of rows deleted. `level_filter` uses [`LogQuery::level`] syntax; `None`
    /// means every level (the All Logs tab).
    ///
    /// The predicate comes from the same [`build_where_clause`] as the tab's
    /// listing query and is built from the tab's level filter alone — the tab's
    /// search text is deliberately never part of it, because the button empties
    /// the whole tab rather than the searched subset.
    pub(crate) async fn clear_logs(&self, level_filter: Option<&str>) -> anyhow::Result<u64> {
        let filters = LogQuery {
            level: level_filter.map(str::to_owned),
            ..LogQuery::default()
        };
        let (where_sql, values) = build_where_clause(&filters);
        self.conn
            .execute(&format!("DELETE FROM logs {where_sql}"), values)
            .await
            .context("Failed to clear log entries")
    }

    /// Whether any row carrying `message` holds `reason` in its `reason` field.
    ///
    /// The reader a caller dedupes against: the rows carrying one message are one
    /// per distinct reason, so they are all read and parsed here rather than
    /// matched in SQL against the JSON they store.
    pub(crate) async fn has_reason(&self, message: &str, reason: &str) -> anyhow::Result<bool> {
        /// The `reason` field of a row's stored `fields`, or `None` when it holds none.
        fn reason_of(fields: &str) -> Option<String> {
            serde_json::from_str::<serde_json::Value>(fields)
                .ok()?
                .get("reason")
                .and_then(serde_json::Value::as_str)
                .map(str::to_owned)
        }
        let rows = self
            .conn
            .query(
                "SELECT fields FROM logs WHERE message = ?1",
                params![message],
            )
            .await
            .context("Failed to read the log rows carrying a message")?;
        for row in &rows {
            // The one selected column, so its position is its own.
            if reason_of(&row.get::<String>(0)?).as_deref() == Some(reason) {
                return Ok(true);
            }
        }
        Ok(false)
    }

    /// Query log entries with optional filters.
    ///
    /// Uses LIKE-based search on target and message columns.
    ///
    /// Returns `(entries, total_count)` where `entries` respects pagination
    /// and `total_count` is the total number of entries matching the same filters.
    pub async fn query(&self, filters: &LogQuery) -> anyhow::Result<(Vec<LogEntry>, usize)> {
        let (where_sql, values) = build_where_clause(filters);

        let count_sql = format!("SELECT COUNT(*) FROM logs {where_sql}");
        let total = self
            .conn
            .query_row(&count_sql, values.clone(), |row| row.get::<i64>(0))
            .await
            .map(|n| usize::try_from(n).unwrap_or(0))?;
        if total == 0 {
            return Ok((vec![], 0));
        }

        let limit: i64 = i64::try_from(filters.limit.unwrap_or(100).min(1000))
            .expect("log query limit overflowed i64; limit must be <= i64::MAX");
        let offset: i64 = i64::try_from(filters.offset.unwrap_or(0))
            .expect("log query offset overflowed i64; offset must be <= i64::MAX");
        let mut data_values = values;
        data_values.push(Value::Integer(limit));
        data_values.push(Value::Integer(offset));

        let data_sql = format!(
            "SELECT {LOGS_COLUMNS} FROM logs {where_sql} ORDER BY id DESC LIMIT ? OFFSET ?",
        );
        let rows = self
            .conn
            .query(&data_sql, data_values)
            .await
            .context("Data query failed")?;

        let mut entries = Vec::new();
        for row in rows {
            entries.push(log_entry_from_row(&row)?);
        }

        Ok((entries, total))
    }

    /// Persist one greppable shell call's grep-engine decision to the dedicated
    /// `grep_telemetry` table. Metadata only; the caller treats failures as
    /// fail-open. Self-contained for grep-specific analysis.
    pub(crate) async fn record_grep_telemetry(
        &self,
        row: GrepTelemetryRow<'_>,
    ) -> anyhow::Result<()> {
        self.conn
            .execute(
                "INSERT INTO grep_telemetry \
                 (recorded_at, command, served, reason, recursive, piped, operand_count, flags, \
                  mode, workspace, grep_count, served_count, skipped_count, duration_ms, exit_code) \
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15)",
                params![
                    crate::db::now(),
                    row.command,
                    i64::from(row.served),
                    row.reason,
                    i64::from(row.recursive),
                    i64::from(row.piped),
                    i64::try_from(row.operand_count)?,
                    row.flags,
                    row.mode,
                    row.workspace,
                    i64::try_from(row.grep_count)?,
                    i64::try_from(row.served_count)?,
                    i64::try_from(row.skipped_count)?,
                    row.duration_ms,
                    row.exit_code.map(i64::from),
                ],
            )
            .await?;
        Ok(())
    }
}

/// Record one event that outlived its call as ONE durable row in the log store:
/// written straight to the store (never through the tracing writer, which batches
/// lines and may drop them), at `WARN`, which the Issues view shows and the
/// retention pass keeps. The row says what happened and carries the recipe that
/// stops it, so the occurrence stays readable after the fact; the agent's own note
/// on the call is the caller-visible text, never a second record.
///
/// The row is attributed to the calling agent from
/// [`crate::agent::tool_record_attribution`] here, rather than from the caller:
/// both call sites are tool calls, and that helper documents the attribution.
///
/// Fail-open by design: a failed write never affects the call's result, and a
/// process with no logs store at all (the `mahbot chrome` CLI dispatches before
/// [`init_tracing`] ever runs, so [`LOG_STORE`] is unset there) writes nothing.
/// A write that was attempted and failed is a `debug!` — the fact is lost either
/// way, and the agent's own note on the call, which carries no less, is what the
/// caller is shown.
pub(crate) async fn record_stray_event(event: StrayEvent<'_>) {
    let Some(store) = LOG_STORE.get() else {
        return;
    };
    let (agent_id, agent_role) = crate::agent::tool_record_attribution();
    let written = write_warn_row(
        store,
        LogEntry {
            target: event.source.to_string(),
            message: event.message.to_string(),
            fields: serde_json::json!({
                "detail": event.recipe,
                "scope": event.scope,
                "session": event.session,
                "duration_ms": event.duration_ms,
            }),
            workspace: event.workspace.to_string(),
            agent_id,
            agent_role,
            ..LogEntry::default()
        },
    )
    .await;
    if let Err(error) = written {
        tracing::debug!(
            %error,
            target = %event.source,
            "could not persist the leftover-process record"
        );
    }
}

/// What became of one fact the Issues view is owed ([`record_issue_once`]): a row's own
/// outcome, the fact already being in the view, and the one way no row carries it.
pub(crate) enum IssueWrite {
    /// This call is what put the row in the store.
    Written,
    /// The store already held the very fact: nothing was written now, and a caller that tells
    /// the owner about every row it writes has already had its turn.
    AlreadyRecorded,
    /// No row carries the fact: there is no logs store to write to (the `mahbot chrome` CLI
    /// dispatches before [`init_tracing`] ever runs, so [`LOG_STORE`] is unset there), or the
    /// write was attempted and failed ([`record_issue_once`] says what each caller does then).
    NotRecorded,
}

/// Write one durable `WARN` row straight to the logs store — never through the tracing
/// writer, which batches lines and may drop them. `WARN` is what the Issues view shows and
/// what the retention pass keeps, so this is the one place a fact the product must not lose
/// is made to outlive the process: [`record_stray_event`] and [`record_issue_once`] differ in
/// what they carry, not in how.
///
/// Nothing is logged here: what a failed write means differs by caller — [`record_stray_event`]
/// notes it at `debug!` because the agent's own answer already carries the fact, and
/// [`record_issue_once`] warns, because its fact is what the Issues view is owed. Returns
/// `Ok(())` when this call put the row in the store, `Err` with the store's own error text when
/// the write was attempted and failed.
async fn write_warn_row(store: &LogStore, entry: LogEntry) -> Result<(), String> {
    let entry = LogEntry {
        timestamp: crate::db::now(),
        level: "WARN".to_string(),
        ..entry
    };
    store
        .insert_batch(&[entry])
        .await
        .map_err(|e| e.to_string())
}

/// Record one durable row for the product's own Issues view, at most once for a
/// `(message, reason)` pair: the row is written straight to the store (never
/// through the tracing writer, which batches and can drop lines) at `WARN`, which
/// the Issues view shows and the retention pass never reaps — so a repeat is a
/// duplicate, not news. A store read that fails records anyway, and says so at `WARN`:
/// the dedupe was skipped, so the row is written without it. Fail-open: no
/// store (the `mahbot chrome` CLI dispatches before [`init_tracing`]) writes
/// nothing.
///
/// `fields` is the row's `fields` JSON object; `reason` is written into it here, so
/// the dedupe key and the stored row can never disagree. The row has no tool
/// attribution (empty agent and workspace): it is written from a background task,
/// not a tool call.
///
/// Reports what became of the fact ([`IssueWrite`]): [`IssueWrite::Written`] for the row this
/// call made, [`IssueWrite::AlreadyRecorded`] for a reason the Issues view already held, and
/// [`IssueWrite::NotRecorded`] for the one way no row carries it — no store at all, or a write
/// that was attempted and failed. A caller holding its OWN de-dupe may mark the reason as told on
/// [`IssueWrite::Written`] or [`IssueWrite::AlreadyRecorded`]; on a failed write this function
/// warns with the store's own error text — a warning about the store, not about what was lost —
/// and a caller whose process exits immediately after ([`crate::self_update`]) adds its own line
/// naming the fact, so the two are deliberately complementary rather than a repeat.
pub(crate) async fn record_issue_once(
    message: &str,
    reason: &str,
    target: &str,
    fields: serde_json::Value,
) -> IssueWrite {
    let Some(store) = LOG_STORE.get() else {
        return IssueWrite::NotRecorded;
    };
    match store.has_reason(message, reason).await {
        // This reason is already in the Issues view: one fact, told once.
        Ok(true) => return IssueWrite::AlreadyRecorded,
        Ok(false) => {}
        Err(e) => tracing::warn!(
            error = %e,
            "could not read whether this issue was already recorded — recording it again"
        ),
    }
    // `reason` goes into the row's own `fields` here, so the dedupe key ([`LogStore::has_reason`])
    // and the stored row can never disagree.
    let mut fields = fields;
    if let Some(object) = fields.as_object_mut() {
        object.insert(
            "reason".to_string(),
            serde_json::Value::String(reason.to_string()),
        );
    }
    match write_warn_row(
        store,
        LogEntry {
            target: target.to_string(),
            message: message.to_string(),
            fields,
            ..LogEntry::default()
        },
    )
    .await
    {
        Ok(()) => IssueWrite::Written,
        Err(error) => {
            tracing::warn!(%error, "could not record an issue for the Issues view");
            IssueWrite::NotRecorded
        }
    }
}

/// Parameters for filtering log queries.
#[derive(Debug, Clone, Default)]
pub struct LogQuery {
    pub level: Option<String>,
    pub target: Option<String>,
    pub search: Option<String>,
    pub since: Option<String>,
    pub limit: Option<usize>,
    pub offset: Option<usize>,
}

// ── Shared helpers ───────────────────────────────────────────────────────────

/// Build WHERE clause and bind values from `LogQuery` filters.
/// Returns `(WHERE ...`, `[values]`) — an empty string when no filters are set.
fn build_where_clause(filters: &LogQuery) -> (String, Vec<Value>) {
    let mut conditions: Vec<String> = Vec::new();
    let mut values: Vec<Value> = Vec::new();

    if let Some(ref levels_str) = filters.level
        && !levels_str.is_empty()
    {
        let levels: Vec<Value> = levels_str
            .split(',')
            .map(str::trim)
            .filter(|s| !s.is_empty())
            .map(|s| Value::Text(s.to_string()))
            .collect();
        if !levels.is_empty() {
            conditions.push(format!(
                "level IN ({})",
                db::sql_in_placeholders(levels.len()),
            ));
            values.extend(levels);
        }
    }

    if let Some(ref target) = filters.target {
        conditions.push("target LIKE ?".into());
        values.push(Value::Text(format!("{target}%")));
    }

    if let Some(ref search) = filters.search
        && !search.is_empty()
    {
        let val = Value::Text(format!("%{search}%"));
        conditions.push("(target LIKE ? OR message LIKE ?)".into());
        values.push(val.clone());
        values.push(val);
    }

    if let Some(ref since) = filters.since {
        conditions.push("timestamp >= ?".into());
        values.push(Value::Text(since.clone()));
    }

    if conditions.is_empty() {
        (String::new(), values)
    } else {
        (format!("WHERE {}", conditions.join(" AND ")), values)
    }
}

fn log_entry_from_row(row: &Row) -> anyhow::Result<LogEntry> {
    let timestamp = row.get::<String>(COL_LOGS_TIMESTAMP)?;
    let level = row.get::<String>(COL_LOGS_LEVEL)?;
    let target = row.get::<String>(COL_LOGS_TARGET)?;
    let message = row.get::<String>(COL_LOGS_MESSAGE)?;
    let fields_str = row.get::<String>(COL_LOGS_FIELDS)?;
    let fields: serde_json::Value =
        serde_json::from_str(&fields_str).unwrap_or(serde_json::Value::Null);

    let agent_id = row.get::<String>(COL_LOGS_AGENT_ID)?;
    let agent_role = row.get::<String>(COL_LOGS_AGENT_ROLE)?;
    let workspace = row.get::<String>(COL_LOGS_WORKSPACE)?;

    Ok(LogEntry {
        timestamp,
        level,
        target,
        message,
        fields,
        agent_id,
        agent_role,
        workspace,
    })
}

// ── Tracing initialization ──────────────────────────────────────────

/// The default filter, applied as a LAYER filter on the JSON log layer — see
/// [`log_layers`] for why it must not be a global subscriber filter. `tantivy`
/// reports through the standard `log` interface, so its directive only ever
/// matches records that arrive over the bridge [`install_log_bridge`] installs.
///
/// `fff` mutes the file-search library the tools and the editor search with —
/// `fff_search`, `fff_grep`, `fff_query_parser` and `fff_notify_debouncer_full`,
/// every module of each: EnvFilter matches a target directive as a raw prefix, so
/// this one directive covers the whole family. It must stay `off`, with no
/// narrower directive added for one of its modules — the most specific directive
/// wins, so `fff_search::grep=error` would re-open that module. A level cut is not
/// enough: it is the pre-existing `fff_search=error` that let the library's
/// in-memory file table, once full after a burst of newly created files, write
/// 24 473 identical ERROR rows in under two seconds — rows nothing reaps, which
/// buried every real problem in the owner's view of the logs. The product's own
/// record for that condition is the edit tool's notice that a written file could
/// not be added to the search index (target `mahbot::tools::edit`), and it is
/// unaffected.
///
/// `pdf_extract` is the PDF text layer that [`crate::document`] converts with: on a
/// font table whose glyphs and Unicode disagree it warns once per page, so a single
/// document produces hundreds of near-identical records. The crate is held to ERROR
/// for that reason — a genuine parse failure is still recorded, everything quieter is
/// not.
pub(crate) const DEFAULT_LOG_FILTER: &str =
    "info,turso_core=warn,tantivy=warn,fff=off,pdf_extract=error";

/// The production layer stack: the JSON log layer, with the log filter applied
/// PER LAYER, plus the engine-cause capture layer (see
/// [`crate::db::checkpoint_cause`]). The filter must be a layer filter and never
/// the subscriber's: installed globally it would drop the engine's DEBUG
/// checkpoint-failure event and silently degrade the failure record to the
/// product's constant text. Extracted from [`init_tracing`] so a test can
/// install the real stack instead of a copy of it.
pub(crate) fn log_layers<W>(
    writer: W,
    env_filter: EnvFilter,
) -> impl tracing::Subscriber + Send + Sync + 'static
where
    W: for<'a> fmt::MakeWriter<'a> + Send + Sync + 'static,
{
    tracing_subscriber::registry()
        .with(
            fmt::Layer::new()
                .json()
                .with_writer(writer)
                .with_ansi(false)
                .with_filter(env_filter),
        )
        .with(
            crate::db::checkpoint_cause::CauseCaptureLayer
                .with_filter(crate::db::checkpoint_cause::filter()),
        )
}

/// Install the `log` → `tracing` bridge: the libraries this product depends on
/// report through the standard `log` interface, and without this bridge their
/// records never reach the log store. MahBot requests the forwarding itself, and
/// a failed install is a boot failure — a `log` logger some dependency installed
/// before us included — because silently losing every dependency's records is
/// the regression this bridge exists to prevent. Upstream's `init` provided it
/// only through the `tracing-log` feature this manifest declares (see
/// `Cargo.toml`).
///
/// `level` is what the `log` crate gates on, and it must be the log layer's own
/// filter level ([`log_bridge_level`]) — never [`LevelFilter::current`], which
/// the engine-cause capture layer raises to DEBUG process-wide; a higher level
/// here would let every dependency's `log::debug!` through only to be dropped
/// again one layer later.
fn install_log_bridge(level: LevelFilter) -> anyhow::Result<()> {
    use tracing_log::AsLog;

    tracing_log::LogTracer::builder()
        .with_max_level(level.as_log())
        .init()
        .map_err(|e| anyhow::anyhow!("failed to install the log→tracing bridge: {e}"))
}

/// The level the `log` bridge gates on: the log layer's own filter's level, read
/// before the filter moves into [`log_layers`] (see [`install_log_bridge`] for
/// why the layer's level and not the subscriber's). A filter with no level hint
/// leaves the bridge at `TRACE`: the layers still decide what is written, so the
/// conservative side is to hand the `log` crate nothing it has to drop.
fn log_bridge_level(env_filter: &EnvFilter) -> LevelFilter {
    env_filter.max_level_hint().unwrap_or(LevelFilter::TRACE)
}

/// Initialize tracing: JSON to Turso store only (no terminal output).
/// Returns the [`LogStore`] for querying and a broadcast sender
/// for live streaming to the Iced native GUI dashboard.
///
/// Installs [`log_layers`] as the global subscriber and, for the same filter's
/// level, the `log` bridge ([`install_log_bridge`]).
pub async fn init_tracing(
    storage_root: &Path,
) -> anyhow::Result<(Arc<LogStore>, tokio::sync::broadcast::Sender<String>)> {
    let store = match LogStore::open(storage_root).await {
        Ok(store) => store,
        Err(e) => {
            // The pre-tracing diagnostics were already handed to stderr (a
            // best-effort channel, see `boot`); drop the buffer, since the
            // logs-store replay can never run.
            crate::boot::clear_boot_diagnostics();
            return Err(e);
        }
    };
    LOG_STORE
        .set(store.clone())
        .map_err(|_| anyhow::anyhow!("LOG_STORE already initialized"))?;
    let log_store = Arc::new(store);
    let (log_tx, log_rx) = tokio::sync::mpsc::unbounded_channel();
    let (broadcast_tx, _) = tokio::sync::broadcast::channel(256);

    spawn_log_writer(Arc::clone(&log_store), log_rx, broadcast_tx.clone());

    let env_filter =
        EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(DEFAULT_LOG_FILTER));
    let bridge_level = log_bridge_level(&env_filter);
    tracing::subscriber::set_global_default(log_layers(make_log_writer(log_tx), env_filter))
        .expect("Unable to install the global tracing subscriber");
    install_log_bridge(bridge_level)?;
    crate::boot::mark_tracing_initialized();

    // Surface pre-tracing boot diagnostics (the pre-flight shape check) in the
    // logs store so the GUI boot log shows them.
    crate::boot::replay_boot_diagnostics();

    Ok((log_store, broadcast_tx))
}

// ── Tracing integration ──────────────────────────────────────────

/// A [`MakeWriter`] that sends JSON log lines over an unbounded channel.
const fn make_log_writer(tx: UnboundedSender<String>) -> LogWriter {
    LogWriter { tx }
}

#[derive(Clone)]
struct LogWriter {
    tx: UnboundedSender<String>,
}

impl io::Write for LogWriter {
    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        let line = String::from_utf8_lossy(buf).to_string();
        let _ = self.tx.send(line);
        Ok(buf.len())
    }

    fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }
}

impl MakeWriter<'_> for LogWriter {
    type Writer = Self;

    fn make_writer(&self) -> Self::Writer {
        self.clone()
    }
}

/// Maximum number of log entries accumulated before a forced DB flush.
///
/// Bounds the write-lock hold: one flush inserts at most this many rows in a
/// single transaction.
const LOG_BATCH_MAX: usize = 50;

/// Maximum age of an accumulated batch before a timer flush.
///
/// Keeps the DB insert path fresh under low log volume while the GUI live
/// broadcast (which stays per-message, unbuffered) is unaffected.
const LOG_FLUSH_INTERVAL: std::time::Duration = std::time::Duration::from_millis(500);

/// Spawn a background task that receives JSON log lines and writes them to Turso
/// and broadcasts them over the channel to the Iced GUI dashboard.
///
/// The GUI live broadcast stays per-message (no GUI lag). Only the DB-insert
/// path batches: entries accumulate in [`LOG_BATCH_MAX`]-sized batches flushed
/// by [`LOG_FLUSH_INTERVAL`], on batch-cap, or on channel close (flush-before-
/// shutdown). A crash mid-batch loses at most [`LOG_BATCH_MAX`] entries —
/// acceptable, since DB log inserts are diagnostics, not durable state.
///
/// A storage-layer panic (e.g. the shared-WAL frame-index invariant violation)
/// must not silently freeze persistence: the flush is panic-absorbed with a
/// bounded number of restarts and backoff, recorded on the visible failure
/// surface, and past the bound the writer stops flushing with a terminal banner
/// instead of spinning on a broken connection. During backoff sleeps the writer
/// is not polling the channel, so lines accumulate in the unbounded channel —
/// bounded by the backoff schedule, drained once flushing resumes. In the
/// terminal stopped state the writer keeps draining the channel (broadcast +
/// drop), so the channel does not grow indefinitely.
fn spawn_log_writer(
    store: Arc<LogStore>,
    rx: UnboundedReceiver<String>,
    broadcast: tokio::sync::broadcast::Sender<String>,
) {
    spawn_log_writer_with_interval(store, rx, broadcast, LOG_FLUSH_INTERVAL);
}

/// [`spawn_log_writer`] with an explicit flush interval — tests inject a
/// non-production interval (very long, or very short) to exercise the
/// batch-cap and timer flush paths without racing the production timer.
fn spawn_log_writer_with_interval(
    store: Arc<LogStore>,
    mut rx: UnboundedReceiver<String>,
    broadcast: tokio::sync::broadcast::Sender<String>,
    flush_interval: std::time::Duration,
) {
    tokio::spawn(async move {
        let mut batch: Vec<LogEntry> = Vec::new();
        let mut flush_timer = tokio::time::interval(flush_interval);
        flush_timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        // The first interval tick fires immediately — consume it so the timer
        // only fires after the first full interval.
        flush_timer.tick().await;

        loop {
            tokio::select! {
                maybe_line = rx.recv() => {
                    let Some(line) = maybe_line else {
                        // Channel closed (all senders dropped, e.g. tracing
                        // teardown on shutdown) — flush remaining and exit.
                        if !log_writer_stopped() {
                            absorb_flush(&store, &mut batch).await;
                        }
                        break;
                    };

                    let trimmed = line.trim();
                    if trimmed.is_empty() {
                        continue;
                    }

                    let Some(entry) = parse_tracing_json(trimmed) else {
                        continue;
                    };

                    // Broadcast to dashboard subscribers before inserting (fast path)
                    let _ = broadcast.send(serde_json::to_string(&entry).expect(
                        "log entry broadcast serialization failed; this should not happen",
                    ));

                    if log_writer_stopped() {
                        // Terminal state: keep draining + broadcasting, but drop
                        // entries instead of pushing them onto a batch that will
                        // never be flushed (prevents unbounded memory growth).
                        continue;
                    }

                    batch.push(entry);
                    if batch.len() >= LOG_BATCH_MAX {
                        absorb_flush(&store, &mut batch).await;
                    }
                }
                _ = flush_timer.tick() => {
                    if !batch.is_empty() && !log_writer_stopped() {
                        absorb_flush(&store, &mut batch).await;
                    }
                }
            }
        }
    });
}

/// Flush the accumulated batch, absorbing storage-layer panics.
///
/// A panic here indicates a broken connection (e.g. the frame-index invariant
/// violation); it is recorded on the failure surface, the batch is dropped, and
/// the writer backs off before the next attempt. After
/// [`LOG_WRITER_MAX_CONSECUTIVE_PANICS`] consecutive panics the writer enters
/// the terminal stopped state (visible banner) rather than spinning forever.
/// A successful flush resets the consecutive-panic counter.
async fn absorb_flush(store: &LogStore, batch: &mut Vec<LogEntry>) {
    let result = AssertUnwindSafe(flush_log_batch(store, batch))
        .catch_unwind()
        .await;
    match result {
        Ok(()) => reset_log_writer_panic_state(),
        Err(payload) => {
            batch.clear();
            let message = format!(
                "log writer storage panic: {}",
                crate::util::panic_message(&*payload)
            );
            let consecutive = record_log_writer_panic(&message);
            if log_writer_stopped() {
                crate::boot::timestamped_stderr(&format!(
                    "log store writer stopped after {consecutive} consecutive storage \
                     panics: {message}"
                ));
            } else {
                tokio::time::sleep(log_writer_panic_backoff(consecutive)).await;
            }
        }
    }
}

/// Insert all accumulated entries in one transaction and clear the batch.
///
/// On persistent failure the batch is still cleared (entries are dropped) —
/// log entries are diagnostics, not durable state. Failures are **not**
/// swallowed: they are recorded on the [`log_write_error_info`] surface
/// (rendered on the GUI Logs page, which is the surface that always reaches the
/// user) and reported to stderr at a bounded rate — a channel a launch with no
/// console discards (see the `boot` module).
/// No `tracing!` call is made from here — the writer task consumes the tracing
/// channel, so tracing from inside it would recurse into itself.
async fn flush_log_batch(store: &LogStore, batch: &mut Vec<LogEntry>) {
    if batch.is_empty() {
        return;
    }

    let mut last_error: Option<anyhow::Error> = None;
    for attempt in 0..LOG_INSERT_MAX_ATTEMPTS {
        match store.insert_batch(batch).await {
            Ok(()) => {
                batch.clear();
                return;
            }
            Err(e) => {
                last_error = Some(e);
                if attempt + 1 < LOG_INSERT_MAX_ATTEMPTS {
                    tokio::time::sleep(LOG_INSERT_RETRY_BACKOFF).await;
                }
            }
        }
    }

    record_log_write_failure(last_error);
    batch.clear();
}

// ── Log-writer error observability ──────────────────────────────────

/// Maximum number of insert attempts (including the first) for one log batch.
///
/// Retrying inside the writer is safe: the batch is only dropped after all
/// attempts fail. The total added latency is bounded by
/// `(LOG_INSERT_MAX_ATTEMPTS - 1) × LOG_INSERT_RETRY_BACKOFF`.
const LOG_INSERT_MAX_ATTEMPTS: usize = 3;

/// Backoff between log-batch insert retry attempts.
const LOG_INSERT_RETRY_BACKOFF: std::time::Duration = std::time::Duration::from_millis(250);

/// Minimum interval between stderr warnings about log-write failures.
const LOG_WRITE_STDERR_WARN_INTERVAL_MS: u64 = 60_000;

/// Consecutive storage-panic restarts before the writer stops flushing
/// permanently. A storage-layer panic indicates a broken connection (e.g. the
/// shared-WAL frame-index invariant violation); retrying past this bound would
/// spin forever on a connection that keeps panicking.
const LOG_WRITER_MAX_CONSECUTIVE_PANICS: u32 = 5;

/// Base backoff after a storage panic (doubles per consecutive panic, capped).
const LOG_WRITER_PANIC_BACKOFF_MS: u64 = 500;

/// Consecutive-panic restart state machine for the log writer.
///
/// Pure and unit-testable; the global writer state ([`LOG_WRITE_LAST_ERROR`])
/// mirrors this struct.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct LogWriterPanicState {
    /// Consecutive storage-panic restarts in progress (0 when the last flush
    /// succeeded).
    consecutive_panics: u32,
    /// True once the writer stopped flushing permanently after exceeding
    /// [`LOG_WRITER_MAX_CONSECUTIVE_PANICS`] — the terminal banner state.
    pub writer_stopped: bool,
}

impl LogWriterPanicState {
    /// Record a storage panic; returns the updated consecutive-panic count.
    #[must_use]
    fn record_panic(&mut self) -> u32 {
        self.consecutive_panics += 1;
        if self.consecutive_panics >= LOG_WRITER_MAX_CONSECUTIVE_PANICS {
            self.writer_stopped = true;
        }
        self.consecutive_panics
    }

    /// Reset the consecutive-panic counter after a successful flush. The
    /// terminal stopped state is sticky — a stopped writer never flushes again
    /// (the broken connection cannot be healed without reopening the store).
    fn reset(&mut self) {
        self.consecutive_panics = 0;
    }
}

/// Snapshot of the log-writer failure surface, for display and tests.
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub(crate) struct LogWriteErrorInfo {
    /// Total failures recorded since startup — batch insert failures and writer panics.
    pub count: u64,
    /// RFC 3339 timestamp of the most recent failure.
    pub last_timestamp: Option<String>,
    /// Message of the most recent failure.
    pub last_message: Option<String>,
    /// Writer panic-restart state.
    pub panic_state: LogWriterPanicState,
}

/// Most recent log-batch insert failure observed by the writer task.
///
/// Count, timestamp, and message live behind a single mutex so readers always
/// observe a consistent triple — a torn pair (timestamp from one failure with
/// the message from the next) would be misleading on the observability
/// surface.
static LOG_WRITE_LAST_ERROR: std::sync::Mutex<LogWriteErrorInfo> =
    std::sync::Mutex::new(LogWriteErrorInfo {
        count: 0,
        last_timestamp: None,
        last_message: None,
        panic_state: LogWriterPanicState {
            consecutive_panics: 0,
            writer_stopped: false,
        },
    });

/// Unix millis of the last stderr warning (rate limiter).
static LOG_WRITE_LAST_STDERR_WARN_MS: AtomicU64 = AtomicU64::new(0);

/// Read the log-writer failure surface.
///
/// This is the sanctioned surface for log-persistence outages: the GUI Logs
/// page renders a warning banner from it. It is safe to call from anywhere
/// (no tracing involved), including from inside the writer task itself.
#[must_use]
pub(crate) fn log_write_error_info() -> LogWriteErrorInfo {
    LOG_WRITE_LAST_ERROR.lock().unwrap_poison().clone()
}

/// Record a failed log-batch insert on the observability surface.
fn record_log_write_failure(error: Option<anyhow::Error>) {
    let message = error.map_or_else(
        || "unknown log insert failure".to_string(),
        |e| format!("{e:#}"),
    );
    record_log_write_failure_impl(&message, LogFailureKind::Insert);
}

/// Record a storage-layer panic absorbed by the writer. Returns the updated
/// consecutive-panic count. The terminal stopped state additionally gets an
/// unconditional banner from the caller.
fn record_log_writer_panic(message: &str) -> u32 {
    record_log_write_failure_impl(message, LogFailureKind::WriterPanic)
}

/// Which failure kind is being recorded — drives the stderr label and whether
/// the storage-panic restart counter is bumped.
#[derive(Clone, Copy)]
enum LogFailureKind {
    Insert,
    WriterPanic,
}

impl LogFailureKind {
    fn label(self) -> &'static str {
        match self {
            Self::Insert => "insert failure",
            Self::WriterPanic => "writer panic",
        }
    }

    fn records_panic(self) -> bool {
        matches!(self, Self::WriterPanic)
    }
}

/// Shared failure-recording core: bumps the counter, stamps timestamp/message,
/// optionally records a storage-panic restart, and emits a rate-limited stderr
/// warning (`kind` labels the failure on the stderr line). stderr is not
/// routed through tracing, so this cannot recurse into the writer task.
fn record_log_write_failure_impl(message: &str, kind: LogFailureKind) -> u32 {
    let (count, consecutive) = {
        let mut guard = LOG_WRITE_LAST_ERROR.lock().unwrap_poison();
        guard.count += 1;
        guard.last_timestamp = Some(db::now());
        guard.last_message = Some(message.to_string());
        let consecutive = if kind.records_panic() {
            guard.panic_state.record_panic()
        } else {
            guard.panic_state.consecutive_panics
        };
        (guard.count, consecutive)
    };
    emit_stderr_warning(count, message, kind.label());
    consecutive
}

/// Reset the consecutive-panic counter after a successful flush. The terminal
/// stopped state is sticky — only [`LogWriterPanicState::reset`]'s counter is
/// cleared here, a stopped writer never flushes again.
fn reset_log_writer_panic_state() {
    let mut guard = LOG_WRITE_LAST_ERROR.lock().unwrap_poison();
    guard.panic_state.reset();
}

/// True once the writer has permanently stopped flushing (terminal banner).
fn log_writer_stopped() -> bool {
    LOG_WRITE_LAST_ERROR
        .lock()
        .unwrap_poison()
        .panic_state
        .writer_stopped
}

/// Backoff after the `n`-th consecutive storage panic: 500ms, 1s, 2s, … capped
/// at 30s. The terminal bound ([`LOG_WRITER_MAX_CONSECUTIVE_PANICS`]) ends the
/// sequence before the cap engages today; the cap guards a future bound
/// increase.
fn log_writer_panic_backoff(consecutive: u32) -> std::time::Duration {
    let shift = consecutive.saturating_sub(1).min(6);
    let ms = LOG_WRITER_PANIC_BACKOFF_MS.saturating_mul(1 << shift);
    std::time::Duration::from_millis(ms.min(30_000))
}

/// Rate-limited stderr warning. stderr is not routed through tracing, so this
/// cannot recurse into the writer task — and it is best-effort delivery (see the
/// `boot` module), which is why [`log_write_error_info`] is the surface a user can
/// always read.
fn emit_stderr_warning(count: u64, message: &str, kind: &str) {
    let now_ms = crate::util::unix_millis();
    let last_warn_ms = LOG_WRITE_LAST_STDERR_WARN_MS.load(Ordering::SeqCst);
    if now_ms.saturating_sub(last_warn_ms) >= LOG_WRITE_STDERR_WARN_INTERVAL_MS {
        LOG_WRITE_LAST_STDERR_WARN_MS.store(now_ms, Ordering::SeqCst);
        crate::boot::timestamped_stderr(&format!("log store {kind} #{count}: {message}"));
    }
}

/// Extract a string field from a JSON value, defaulting to `""`.
fn get_str_or_empty(val: &serde_json::Value, key: &str) -> String {
    json::get_opt_str(val, key).unwrap_or("").to_string()
}

/// Parse a tracing-subscriber JSON line into a `LogEntry`.
fn parse_tracing_json(line: &str) -> Option<LogEntry> {
    let val: serde_json::Value = serde_json::from_str(line).ok()?;

    let timestamp = get_str_or_empty(&val, "timestamp");
    let level = get_str_or_empty(&val, "level");
    let target = get_str_or_empty(&val, "target");

    let mut fields = val
        .get("fields")
        .cloned()
        .unwrap_or(serde_json::Value::Null);

    let message = get_str_or_empty(&fields, "message");

    if let Some(obj) = fields.as_object_mut() {
        obj.remove("message");
    }
    let fields = if fields.as_object().is_some_and(serde_json::Map::is_empty) {
        serde_json::Value::Null
    } else {
        fields
    };

    // Extract agent_id, agent_role and workspace from the innermost span
    let (agent_id, agent_role, workspace) = extract_agent_from_span(&val);

    Some(LogEntry {
        timestamp,
        level,
        target,
        message,
        fields,
        agent_id,
        agent_role,
        workspace,
    })
}

/// Extract the three agent-related fields from a span JSON object.
fn extract_agent_fields(span: &serde_json::Value) -> (String, String, String) {
    (
        get_str_or_empty(span, "agent_id"),
        get_str_or_empty(span, "role"),
        get_str_or_empty(span, "workspace"),
    )
}

/// Extract `agent_id`, `role`, and `workspace` from the current span data
/// in tracing JSON.
///
/// `tracing-subscriber` JSON format puts the current span's fields under
/// `span.agent_id`, `span.role`, and `span.workspace` (or `spans[last].*`).
/// The three sources are merged per-component — the event's own fields win
/// where present, the span fills the gaps — so every corner attributes
/// correctly:
///
/// - full event fields (e.g. `run_agent`'s "Agent failed" entry, analyze-tool
///   sub-agent failures) name the failing agent directly, beating the
///   inherited caller span;
/// - workspace-only events (e.g. "Search index capacity exhausted") keep the
///   span's agent attribution;
/// - agent_id-only events (e.g. session-persistence warnings) keep the span's
///   role/workspace attribution.
fn extract_agent_from_span(val: &serde_json::Value) -> (String, String, String) {
    let mut agent_id = String::new();
    let mut role = String::new();
    let mut workspace = String::new();
    for candidate in std::iter::once(val.get("fields"))
        .chain(std::iter::once(val.get("span")))
        .chain(std::iter::once(
            val.get("spans")
                .and_then(|v| v.as_array())
                .and_then(|a| a.last()),
        ))
        .flatten()
    {
        let (id, r, ws) = extract_agent_fields(candidate);
        if agent_id.is_empty() {
            agent_id = id;
        }
        if role.is_empty() {
            role = r;
        }
        if workspace.is_empty() {
            workspace = ws;
        }
        if !agent_id.is_empty() && !role.is_empty() && !workspace.is_empty() {
            break;
        }
    }
    (agent_id, role, workspace)
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_parse_tracing_json_full() {
        let line = r#"{"timestamp":"2025-05-06T12:34:56.000000Z","level":"INFO","target":"mahbot::orchestrator","span":{"name":"agent","agent_id":"00000000-0000-0000-0000-000000000000","role":"lead","workspace":"/some/workspace"},"fields":{"message":"Hello world","key":"value"}}"#;
        let entry = parse_tracing_json(line).unwrap();
        assert_eq!(entry.timestamp, "2025-05-06T12:34:56.000000Z");
        assert_eq!(entry.level, "INFO");
        assert_eq!(entry.target, "mahbot::orchestrator");
        assert_eq!(entry.message, "Hello world");
        assert_eq!(entry.fields, serde_json::json!({"key": "value"}));
        assert_eq!(entry.agent_id, "00000000-0000-0000-0000-000000000000");
        assert_eq!(entry.agent_role, "lead");
        assert_eq!(entry.workspace, "/some/workspace");
    }

    #[test]
    fn test_parse_tracing_json_no_fields() {
        let line = r#"{"timestamp":"2025-05-06T12:34:56.000000Z","level":"WARN","target":"test","fields":{"message":"warning"}}"#;
        let entry = parse_tracing_json(line).unwrap();
        assert_eq!(entry.message, "warning");
        assert_eq!(entry.fields, serde_json::Value::Null);
        assert_eq!(entry.agent_id, "");
        assert_eq!(entry.agent_role, "");
        assert_eq!(entry.workspace, "");
    }

    #[test]
    fn test_parse_tracing_json_lenient() {
        let entry = parse_tracing_json(r#"{"incomplete": true}"#).unwrap();
        assert_eq!(entry.timestamp, "");
        assert_eq!(entry.level, "");
        assert_eq!(entry.target, "");
        assert_eq!(entry.message, "");
        assert_eq!(entry.fields, serde_json::Value::Null);
        assert_eq!(entry.agent_id, "");
        assert_eq!(entry.agent_role, "");
        assert_eq!(entry.workspace, "");
    }

    /// Agent-attribution corners for `parse_tracing_json`: the event's own
    /// fields win where present, the span fills the gaps, and the `spans`
    /// array is the last resort. Each case: (name, line, agent_id, role,
    /// workspace).
    #[test]
    fn test_parse_tracing_json_agent_attribution() {
        let cases = [
            (
                "span only",
                r#"{"timestamp":"...","level":"INFO","target":"test","span":{"name":"agent","agent_id":"abc-123","role":"analyst"},"fields":{"message":"researching"}}"#,
                "abc-123",
                "analyst",
                "",
            ),
            (
                "spans array",
                r#"{"timestamp":"...","level":"INFO","target":"test","spans":[{"name":"parent"},{"name":"agent","agent_id":"xyz-456","role":"coder","workspace":"/ws"}],"fields":{"message":"writing code"}}"#,
                "xyz-456",
                "coder",
                "/ws",
            ),
            (
                "event fields without span",
                r#"{"timestamp":"...","level":"ERROR","target":"mahbot::agent","fields":{"message":"Agent failed","agent_id":"ticket_123_engineer","role":"engineer","workspace":"my-ws","classification":"transport"}}"#,
                "ticket_123_engineer",
                "engineer",
                "my-ws",
            ),
            (
                "event beats inherited span",
                r#"{"timestamp":"...","level":"ERROR","target":"mahbot::agent","span":{"name":"agent","agent_id":"caller_42","role":"engineer","workspace":"parent-ws"},"fields":{"message":"Agent failed","agent_id":"analyze_ws_1_2_analyst","role":"analyst","workspace":"my-ws","classification":"runtime"}}"#,
                "analyze_ws_1_2_analyst",
                "analyst",
                "my-ws",
            ),
            (
                "workspace-only event keeps span agent",
                r#"{"timestamp":"...","level":"WARN","target":"mahbot::tools::edit","span":{"name":"agent","agent_id":"ticket_7_engineer","role":"engineer","workspace":"my-ws"},"fields":{"message":"Search index capacity exhausted","workspace":"my-ws","path":"src/a.rs"}}"#,
                "ticket_7_engineer",
                "engineer",
                "my-ws",
            ),
            (
                "agent_id-only event merges span role/workspace",
                r#"{"timestamp":"...","level":"WARN","target":"mahbot::agent","span":{"name":"agent","agent_id":"ticket_7_engineer","role":"engineer","workspace":"my-ws"},"fields":{"message":"Failed to persist incoming messages to session DB","agent_id":"ticket_7_engineer","error":"io"}}"#,
                "ticket_7_engineer",
                "engineer",
                "my-ws",
            ),
        ];
        for (name, line, id, role, ws) in cases {
            let entry = parse_tracing_json(line).unwrap();
            assert_eq!(entry.agent_id, id, "{name}: agent_id");
            assert_eq!(entry.agent_role, role, "{name}: agent_role");
            assert_eq!(entry.workspace, ws, "{name}: workspace");
        }
    }

    /// Create a temporary LogStore for tests.
    /// Returns the store and a TempDir that must be held to prevent premature cleanup.
    async fn test_store() -> (Arc<LogStore>, tempfile::TempDir) {
        let (store, dir) = crate::open_test_store!(LogStore, "log");
        (Arc::new(store), dir)
    }

    // Helper to seed log entries in tests
    async fn seed_entries(store: &LogStore, entries: &[LogEntry]) {
        store.insert_batch(entries).await.unwrap();
    }

    /// The per-tab clear scopes: the Issues scope deletes warnings and errors
    /// only, the All Logs scope deletes every stored record (including the ones
    /// the Issues view shows).
    #[tokio::test]
    async fn clear_logs_scopes_match_the_tabs() {
        let (store, _dir) = test_store().await;
        seed_entries(
            &store,
            &[
                LogEntry {
                    level: "INFO".into(),
                    ..Default::default()
                },
                LogEntry {
                    level: "WARN".into(),
                    ..Default::default()
                },
                LogEntry {
                    level: "ERROR".into(),
                    ..Default::default()
                },
            ],
        )
        .await;

        // Issues scope: warnings and errors only — the informational row survives.
        assert_eq!(store.clear_logs(Some("ERROR,WARN")).await.unwrap(), 2);
        let (entries, total) = store.query(&LogQuery::default()).await.unwrap();
        assert_eq!(total, 1);
        assert_eq!(entries[0].level, "INFO");

        // All Logs scope: every stored record.
        assert_eq!(store.clear_logs(None).await.unwrap(), 1);
        assert_eq!(store.query(&LogQuery::default()).await.unwrap().1, 0);
    }

    /// [`LogStore::has_reason`] finds a reason any row carrying a message holds, so
    /// a reason already recorded is never recorded again — and a reason the store
    /// has not been told is not found.
    #[tokio::test]
    async fn a_reason_already_recorded_is_found_and_a_new_one_is_not() {
        let (store, _dir) = test_store().await;

        store
            .insert_batch(&[
                LogEntry {
                    message: "the update stopped".into(),
                    fields: serde_json::json!({ "reason": "first" }),
                    ..Default::default()
                },
                LogEntry {
                    message: "the update stopped".into(),
                    fields: serde_json::json!({ "reason": "second" }),
                    ..Default::default()
                },
                // A row with no reason at all is not evidence for one.
                LogEntry {
                    message: "the update stopped".into(),
                    ..Default::default()
                },
                // A different message is its own fact, whatever its reason.
                LogEntry {
                    message: "another fact".into(),
                    fields: serde_json::json!({ "reason": "first" }),
                    ..Default::default()
                },
            ])
            .await
            .unwrap();

        for (message, reason, found) in [
            ("the update stopped", "first", true),
            ("the update stopped", "second", true),
            ("the update stopped", "third", false),
            ("another fact", "first", true),
            ("another fact", "second", false),
            ("never recorded", "first", false),
        ] {
            assert_eq!(
                store.has_reason(message, reason).await.unwrap(),
                found,
                "{message:?} + {reason:?}"
            );
        }
    }

    /// Open a healthy store in a fresh temp dir, seed one entry, and checkpoint
    /// so all pages land in the main DB file, then close it. Returns the dir —
    /// it must be held for the lifetime of the reopened store.
    async fn seeded_closed_store() -> tempfile::TempDir {
        let tmp = tempfile::TempDir::new().expect("temp dir for test");
        {
            let store = LogStore::open(tmp.path())
                .await
                .expect("open healthy store");
            seed_entries(
                &store,
                &[LogEntry {
                    timestamp: "2025-01-01T00:00:00Z".into(),
                    level: "INFO".into(),
                    target: "test".into(),
                    message: "pre-corruption".into(),
                    ..Default::default()
                }],
            )
            .await;
            store
                .conn
                .checkpoint_ungated()
                .await
                .expect("checkpoint so pages land in the main DB file");
        }
        tmp
    }

    #[tokio::test]
    async fn test_spawn_log_writer_writes_to_store() {
        let (store, _dir) = test_store().await;
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        let (broadcast_tx, _) = tokio::sync::broadcast::channel(256);

        spawn_log_writer(store.clone(), rx, broadcast_tx);

        tx.send(
            r#"{"timestamp":"2025-01-01T00:00:00Z","level":"INFO","target":"test","fields":{"message":"hi"}}"#
                .to_string(),
        )
        .unwrap();
        tx.send(
            r#"{"timestamp":"2025-01-01T00:00:01Z","level":"ERROR","target":"test","fields":{"message":"oh no","err":"boom"}}"#
                .to_string(),
        )
        .unwrap();

        // Poll instead of a fixed sleep: the writer shares this runtime and
        // its DB work can lag under load (see `wait_for_total`).
        drop(tx);
        wait_for_total(&store, 2, std::time::Duration::from_secs(2)).await;

        let (entries, total) = store.query(&LogQuery::default()).await.unwrap();
        assert_eq!(total, 2);
        assert_eq!(entries.len(), 2);
        assert_eq!(entries[0].message, "oh no");
        assert_eq!(entries[1].message, "hi");
    }

    /// Poll `store.query` until the total entry count reaches `expected` or the
    /// timeout elapses. Avoids the fixed-sleep races of the earlier version
    /// (the writer and the test share a runtime, so wall-clock sleeps can be
    /// skewed by writer-side DB work on loaded machines).
    async fn wait_for_total(store: &LogStore, expected: usize, timeout: std::time::Duration) {
        let deadline = std::time::Instant::now() + timeout;
        loop {
            let (_, total) = store.query(&LogQuery::default()).await.unwrap();
            if total == expected {
                return;
            }
            assert!(
                std::time::Instant::now() < deadline,
                "timed out waiting for total == {expected}, got {total}"
            );
            tokio::time::sleep(std::time::Duration::from_millis(20)).await;
        }
    }

    #[tokio::test]
    async fn test_flush_log_batch_records_failure_on_surface() {
        let (store, _dir) = test_store().await;
        let baseline = log_write_error_info().count;

        let entry = LogEntry {
            timestamp: "2025-01-01T00:00:00Z".to_string(),
            level: "INFO".to_string(),
            target: "test".to_string(),
            message: "should not persist".to_string(),
            ..Default::default()
        };

        // Deterministic insert failure: drop the logs table through the store's
        // own connection, so the batch INSERT fails at prepare time.
        store
            .conn
            .execute("DROP TABLE logs", ())
            .await
            .expect("drop logs table for failure test");

        let mut batch = vec![entry];
        flush_log_batch(&store, &mut batch).await;

        // The batch must be dropped (entries are diagnostics) and the failure
        // must be visible on the sanctioned surface.
        assert!(batch.is_empty(), "failed batch must still be cleared");
        let info = log_write_error_info();
        assert!(
            info.count > baseline,
            "failure count must advance: baseline {baseline}, now {}",
            info.count
        );
        assert!(
            info.last_message.is_some(),
            "last-error message must be recorded"
        );
    }

    #[tokio::test]
    async fn test_flush_log_batch_persists_without_failure() {
        // A healthy flush clears the batch, persists the entry, and records no
        // write failure. The bounded-retry retry-then-succeed branch
        // (`LOG_INSERT_MAX_ATTEMPTS` loop) is deliberately left untested since
        // failure injection isn't worth the machinery.
        let (store, _dir) = test_store().await;
        let baseline = log_write_error_info().count;

        let mut batch = vec![LogEntry {
            timestamp: "2025-01-01T00:00:01Z".to_string(),
            level: "INFO".to_string(),
            target: "test".to_string(),
            message: "persisted".to_string(),
            ..Default::default()
        }];
        flush_log_batch(&store, &mut batch).await;

        assert!(batch.is_empty());
        assert_eq!(
            log_write_error_info().count,
            baseline,
            "no failure recorded"
        );
        let (entries, total) = store.query(&LogQuery::default()).await.unwrap();
        assert_eq!(total, 1);
        assert_eq!(entries[0].message, "persisted");
    }

    #[tokio::test]
    async fn test_log_writer_batches_and_timer_flushes() {
        // Writer 1: a flush interval long enough that the timer can never fire
        // during the test — only the batch-cap path can insert rows.
        let (store, _dir) = test_store().await;
        let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
        let (broadcast_tx, _) = tokio::sync::broadcast::channel(256);
        spawn_log_writer_with_interval(
            store.clone(),
            rx,
            broadcast_tx,
            std::time::Duration::from_mins(1),
        );

        // One entry first — below the batch cap, so it stays buffered.
        tx.send(
            r#"{"timestamp":"2025-01-01T00:00:02Z","level":"INFO","target":"test","fields":{"message":"timer flush"}}"#
                .to_string(),
        )
        .unwrap();

        // Then LOG_BATCH_MAX more entries — the batch-cap flush fires as soon
        // as the writer accumulates a full batch (the timer cannot fire here).
        for i in 0..LOG_BATCH_MAX {
            tx.send(
                format!(
                    r#"{{"timestamp":"2025-01-01T00:00:03Z","level":"INFO","target":"test","fields":{{"message":"batch {i}"}}}}"#
                ),
            )
            .unwrap();
        }

        // The cap flush must insert exactly LOG_BATCH_MAX entries; the lone
        // earlier entry is still buffered (60s timer never fired).
        wait_for_total(&store, LOG_BATCH_MAX, std::time::Duration::from_secs(10)).await;

        // Writer 2: a very short flush interval — the timer path must flush a
        // lone buffered entry without needing the cap or channel close.
        let (store2, _dir2) = test_store().await;
        let (tx2, rx2) = tokio::sync::mpsc::unbounded_channel();
        let (broadcast_tx2, _) = tokio::sync::broadcast::channel(256);
        spawn_log_writer_with_interval(
            store2.clone(),
            rx2,
            broadcast_tx2,
            std::time::Duration::from_millis(50),
        );
        tx2.send(
            r#"{"timestamp":"2025-01-01T00:00:04Z","level":"INFO","target":"test","fields":{"message":"timer fired"}}"#
                .to_string(),
        )
        .unwrap();
        wait_for_total(&store2, 1, std::time::Duration::from_secs(10)).await;

        drop(tx);
        drop(tx2);
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }

    #[tokio::test]
    async fn test_like_search_substring() {
        let (store, _dir) = test_store().await;

        let entries = vec![
            LogEntry {
                timestamp: "2025-01-01T00:00:00Z".into(),
                level: "INFO".into(),
                target: "module_a".into(),
                message: "processing request".into(),
                ..Default::default()
            },
            LogEntry {
                timestamp: "2025-01-01T00:00:01Z".into(),
                level: "ERROR".into(),
                target: "module_b".into(),
                message: "failed to process".into(),
                ..Default::default()
            },
            LogEntry {
                timestamp: "2025-01-01T00:00:02Z".into(),
                level: "INFO".into(),
                target: "module_c".into(),
                message: "started".into(),
                ..Default::default()
            },
        ];

        seed_entries(&store, &entries).await;

        // LIKE %...% matches substrings: "proc" matches "processing" and "process"
        let (results, total) = store
            .query(&LogQuery {
                search: Some("proc".into()),
                ..Default::default()
            })
            .await
            .unwrap();
        assert_eq!(total, 2, "substring 'proc' should match both entries");
        assert_eq!(results.len(), 2);
        let (results, total) = store
            .query(&LogQuery {
                search: Some("request".into()),
                ..Default::default()
            })
            .await
            .unwrap();
        assert_eq!(total, 1);
        assert_eq!(results[0].message, "processing request");

        // LIKE matches the target column too
        let (_results, total) = store
            .query(&LogQuery {
                search: Some("module".into()),
                ..Default::default()
            })
            .await
            .unwrap();
        assert_eq!(total, 3, "all targets contain 'module'");
    }

    #[tokio::test]
    async fn test_like_search_combined_filters() {
        let (store, _dir) = test_store().await;

        let entries = vec![
            LogEntry {
                timestamp: "2025-01-01T00:00:00Z".into(),
                level: "INFO".into(),
                target: "mahbot::orchestrator".into(),
                message: "processing request".into(),
                ..Default::default()
            },
            LogEntry {
                timestamp: "2025-01-01T00:00:01Z".into(),
                level: "ERROR".into(),
                target: "mahbot::tools".into(),
                message: "failed to process".into(),
                fields: serde_json::json!({"code": 1}),
                ..Default::default()
            },
            LogEntry {
                timestamp: "2025-01-01T00:00:02Z".into(),
                level: "INFO".into(),
                target: "mahbot::api".into(),
                message: "started".into(),
                ..Default::default()
            },
        ];

        seed_entries(&store, &entries).await;

        // LIKE + level filter
        let (results, total) = store
            .query(&LogQuery {
                level: Some("ERROR".into()),
                search: Some("process".into()),
                ..Default::default()
            })
            .await
            .unwrap();
        assert_eq!(total, 1, "only ERROR log matching 'process'");
        assert_eq!(results[0].message, "failed to process");
        let (_results, total) = store
            .query(&LogQuery {
                target: Some("mahbot::tools".into()),
                search: Some("process".into()),
                ..Default::default()
            })
            .await
            .unwrap();
        assert_eq!(total, 1, "only tools target entry matching 'process'");

        // LIKE + since
        let (_results, total) = store
            .query(&LogQuery {
                since: Some("2025-01-01T00:00:01Z".into()),
                search: Some("process".into()),
                ..Default::default()
            })
            .await
            .unwrap();
        assert_eq!(total, 1, "only entry after timestamp matching 'process'");
    }

    #[tokio::test]
    async fn test_like_search_with_special_chars() {
        let (store, _dir) = test_store().await;

        let entries = vec![
            LogEntry {
                timestamp: "2025-01-01T00:00:00Z".into(),
                level: "INFO".into(),
                target: "module_a".into(),
                message: "processing `Hello ${name}` template".into(),
                ..Default::default()
            },
            LogEntry {
                timestamp: "2025-01-01T00:00:01Z".into(),
                level: "ERROR".into(),
                target: "module_b".into(),
                message: "normal log entry".into(),
                ..Default::default()
            },
        ];

        seed_entries(&store, &entries).await;

        // LIKE is literal substring — backtick and ${} match as-is
        let (results, total) = store
            .query(&LogQuery {
                search: Some("template".into()),
                ..Default::default()
            })
            .await
            .unwrap();
        assert_eq!(total, 1, "LIKE should match partial word in message");
        assert!(
            results[0].message.contains("template"),
            "should match the correct entry"
        );

        // Empty search returns all entries
        let (_results, total) = store
            .query(&LogQuery {
                search: None,
                ..Default::default()
            })
            .await
            .unwrap();
        assert_eq!(total, 2, "no search filter should return all entries");
    }

    /// The writer's panic-restart bound: after enough consecutive panics the
    /// writer enters the terminal stopped state (banner), and a successful
    /// flush resets the counter (the stopped state is sticky).
    #[test]
    fn test_log_writer_panic_state_machine() {
        let mut state = LogWriterPanicState::default();
        assert!(!state.writer_stopped);
        for i in 1..=LOG_WRITER_MAX_CONSECUTIVE_PANICS {
            let _ = state.record_panic();
            assert_eq!(state.consecutive_panics, i);
        }
        assert!(
            state.writer_stopped,
            "writer must stop after the consecutive-panic bound"
        );
        state.reset();
        assert_eq!(state.consecutive_panics, 0);
        assert!(
            state.writer_stopped,
            "terminal stopped state is sticky across reset"
        );
    }

    /// Boot-time refusal: a logs store whose main DB file is corrupt is never
    /// quarantined or recreated — `LogStore::open` errors with a `StoreRefusal`
    /// and leaves the file byte-for-byte as it was.
    #[tokio::test]
    async fn test_log_store_open_refuses_corrupt_store() {
        let tmp = seeded_closed_store().await;
        let root = tmp.path();

        // Corrupt a b-tree page in the main DB file (zero page 2; the header
        // page 1 stays intact so the file still passes the shape check).
        let db_path = db::store_db_path(root, "logs");
        let bytes = std::fs::read(&db_path).expect("read db file");
        assert!(bytes.len() > 8192, "test needs a multi-page db file");
        let mut corrupted = bytes.clone();
        corrupted[4096..8192].fill(0);
        std::fs::write(&db_path, &corrupted).expect("corrupt db file");

        let err = LogStore::open(root)
            .await
            .expect_err("a corrupt store must be refused, not recreated");
        assert!(
            err.downcast_ref::<crate::db::StoreRefusal>().is_some(),
            "expected a StoreRefusal, got: {err:#}"
        );
        assert_eq!(
            std::fs::read(&db_path).expect("read db file"),
            corrupted,
            "the refused store's main file must be unchanged"
        );
        crate::db::test_support::assert_not_quarantined(&root.join("db"), "a refused logs store");
    }

    /// A record written through the standard third-party `log` interface — how
    /// every dependency of this product reports — reaches the log layer under its
    /// real target, through the production stack and the production bridge, and
    /// the line the layer wrote is a row the store's own parser accepts.
    ///
    /// The bridge installs a process-global `log` logger, which is why the
    /// production `install_log_bridge` treats a logger that is already there as a
    /// failure; the stack itself goes on this thread's dispatcher, so the rest of
    /// the suite sees no global subscriber change. The real `EnvFilter` is in the
    /// stack, so the assertion on the record's `target` also pins the
    /// `tracing-log` feature this manifest declares: without it the layer stores
    /// the literal target `log` instead of the callsite's own.
    #[test]
    fn dependency_log_records_reach_the_log_store() {
        const MESSAGE: &str = "a dependency record through the standard logging interface";

        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let env_filter = EnvFilter::new(DEFAULT_LOG_FILTER);
        let bridge_level = log_bridge_level(&env_filter);
        let _guard = tracing::subscriber::set_default(log_layers(make_log_writer(tx), env_filter));
        install_log_bridge(bridge_level).expect("install the log→tracing bridge");

        tracing_log::log::warn!("{MESSAGE}");

        let mut written = String::new();
        while let Ok(line) = rx.try_recv() {
            written.push_str(&line);
        }
        let record = written
            .lines()
            .find(|line| line.contains(MESSAGE))
            .unwrap_or_else(|| {
                panic!(
                    "a record written through the standard `log` interface must reach the log \
                     layer: {written}"
                )
            });
        let entry = parse_tracing_json(record).expect("the log store must parse the written line");
        assert_eq!(
            entry.message, MESSAGE,
            "the store row must carry the record's message",
        );
        assert_eq!(
            entry.target,
            module_path!(),
            "the record's real target must survive the bridge, not become the literal `log`",
        );
    }

    /// The level cut [`DEFAULT_LOG_FILTER`] applies to the PDF text layer the read
    /// tool's document conversion runs: that dependency's warnings are dropped
    /// before the store, its errors still reach it. The cut is the target's level,
    /// not the target itself, and a document conversion is exactly what cannot be
    /// driven hermetically, so the layer's view of the dependency is what is
    /// asserted.
    #[test]
    fn the_default_filter_holds_the_pdf_dependency_to_error() {
        const TARGET: &str = "pdf_extract";
        const DROPPED: &str = "a warning the dependency emits once per page";
        const KEPT: &str = "the dependency's own error";
        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let _guard = tracing::subscriber::set_default(log_layers(
            make_log_writer(tx),
            EnvFilter::new(DEFAULT_LOG_FILTER),
        ));
        tracing::warn!(target: TARGET, "{DROPPED}");
        tracing::error!(target: TARGET, "{KEPT}");
        let mut written = String::new();
        while let Ok(line) = rx.try_recv() {
            written.push_str(&line);
        }
        assert!(
            !written.contains(DROPPED),
            "the dependency's warnings must not reach the log layer: {written}"
        );
        assert!(
            written.contains(KEPT),
            "the dependency's errors must still reach the log layer: {written}"
        );
    }

    /// The whole `fff_*` family — the file-search library the tools, the editor
    /// and the read tool's path recovery search with — is muted outright: not one
    /// of its targets reaches the log layer at any level, while records from
    /// everywhere else do, including the product's own notice that a written file
    /// could not be added to the search index.
    ///
    /// The records are written with explicit targets on the tracing API rather
    /// than through the `log` interface: the bridge is a process-global `log`
    /// logger that only one test in this process can install, and it preserves the
    /// record's real target — which
    /// [`dependency_log_records_reach_the_log_store`] pins. What is asserted here
    /// is the layer's own verdict, per target and per level.
    #[test]
    fn the_default_filter_silences_the_search_library() {
        // The burst that motivated the mute: the library's file table refusing to
        // grow, one ERROR per file that does not fit. The other targets are the
        // family's other crates and modules — the pre-existing filter held
        // `fff_search` and `fff_search::grep` at ERROR, so a broader mute that
        // leaves a narrower directive behind re-opens them.
        const BURST_TARGET: &str = "fff_search::stable_vec";
        const BURST: &str = "StableVec: capacity exhausted — dropping item to prevent reallocation";
        const GREP_TARGET: &str = "fff_search::grep";
        const GREP: &str = "a grep diagnostic from the file-search library";
        const SHARED_TARGET: &str = "fff_search::shared";
        const SHARED: &str = "a watcher diagnostic from the file-search library";
        const FFF_GREP_TARGET: &str = "fff_grep::matcher";
        const FFF_GREP: &str = "a record from the family's grep crate";
        const PARSER_TARGET: &str = "fff_query_parser::parse";
        const PARSER: &str = "a record from the family's query parser";
        const DEBOUNCER_TARGET: &str = "fff_notify_debouncer_full";
        const DEBOUNCER: &str = "a record the family logs over the `log` interface";
        // The product's own record for the same condition, and the workspace
        // engine's own INFO — both emit under `mahbot::*` and must be unaffected.
        const NOTICE_TARGET: &str = "mahbot::tools::edit";
        const NOTICE: &str =
            "Search index capacity exhausted after file write — background rescan needed";
        const ENGINE_TARGET: &str = "mahbot::search_engine";
        const ENGINE: &str = "Search engine created — background scan started";

        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
        let _guard = tracing::subscriber::set_default(log_layers(
            make_log_writer(tx),
            EnvFilter::new(DEFAULT_LOG_FILTER),
        ));

        tracing::error!(target: BURST_TARGET, len = 1025, capacity = 1025, "{BURST}");
        tracing::error!(target: GREP_TARGET, "{GREP}");
        tracing::warn!(target: SHARED_TARGET, "{SHARED}");
        tracing::info!(target: FFF_GREP_TARGET, "{FFF_GREP}");
        tracing::error!(target: PARSER_TARGET, "{PARSER}");
        tracing::info!(target: DEBOUNCER_TARGET, "{DEBOUNCER}");
        tracing::warn!(target: NOTICE_TARGET, "{NOTICE}");
        tracing::info!(target: ENGINE_TARGET, "{ENGINE}");

        let mut written = String::new();
        while let Ok(line) = rx.try_recv() {
            written.push_str(&line);
        }
        for (target, message) in [
            (BURST_TARGET, BURST),
            (GREP_TARGET, GREP),
            (SHARED_TARGET, SHARED),
            (FFF_GREP_TARGET, FFF_GREP),
            (PARSER_TARGET, PARSER),
            (DEBOUNCER_TARGET, DEBOUNCER),
        ] {
            assert!(
                !written.contains(message),
                "a record from {target} reached the log layer: {written}"
            );
        }
        assert!(
            written.contains(NOTICE),
            "the product's own notice about the search index must still reach the log layer: \
             {written}"
        );
        assert!(
            written.contains(ENGINE),
            "the product's own search-engine record must still reach the log layer: {written}"
        );
    }
}