omniterm 0.2.27

//! Backend-authoritative accumulation of an in-progress ACP assistant turn.
//!
//! The reliability problem this solves: assistant streaming content (thoughts,
//! tool cards, plans) used to live only in the browser's zustand store and was
//! persisted by the frontend after `prompt_done`. Refreshing mid-turn lost the
//! current turn entirely. The accumulator moves the source of truth to the backend:
//! it is fed from the `AcpClient` notification callback (which runs on the ACP
//! connection task, independent of any WebSocket), so a turn keeps being persisted
//! even with zero WS clients connected.
//!
//! Storage strategy — **raw single-turn frames, not re-classified blocks**. The
//! accumulator stores the raw `SessionUpdate` payloads of the active turn (bounded to
//! one turn) as a JSON wrapper `{"v":1,"frames":[...]}` in the streaming row's `blocks`
//! column. On refresh/reconnect the frontend re-runs its existing classifier over these
//! frames, guaranteeing identical rendering with zero duplication of the (large,
//! multi-implementation) classification logic in Rust.
//!
//! **Frame retention is bounded on two axes** — frame count ([`MAX_FRAMES`]) *and*
//! bytes ([`MAX_BLOCKS_BYTES`], plus a per-frame cap [`MAX_FRAME_BYTES`]). Long agent
//! turns emit a high rate of raw notifications (thought/tool chunks can exceed tens of
//! thousands per turn), and a frame count cap alone only bounds the column when frames
//! are small — which is not something we get to assume (see the byte-cap docs). The
//! accumulator keeps only the most recent window for mid-turn crash recovery —
//! `text` accumulates the agent message text under its own byte cap ([`MAX_TEXT_BYTES`],
//! head + tail with an explicit omission marker), and the frontend writes back
//! its complete structured blocks on `sync` — so memory, flush cost and the `blocks`
//! column stay bounded instead of growing quadratically over a long turn.
//!
//! Writes are debounced: the fold path only flags dirty and pings a dedicated writer
//! task, which coalesces bursts (trailing debounce + max-latency cap) so a high chunk
//! rate maps to a few SQLite writes per second. The one command that must never be
//! coalesced away — [`WriterCmd::EndTurn`], the non-idempotent finalize transition —
//! travels a separate unbounded channel ([`Sink::end_tx`]) so a saturated signal queue
//! can never silently drop it.
//!
//! The accumulator is also the **single accounting point for turn work time**: turn
//! duration (`work_ms`) and permission-wait time (`wait_ms`) are measured here, on the
//! same `active → false` transition that finalizes the row, so each turn is counted
//! exactly once. The measurement basis (wall time minus human-approval wait) and the
//! decision to keep it here rather than in a separate timer live in
//! `docs/dev/plans/archive/2026-08-30-acp-work-time.md` (D1/D2).

use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

use agent_client_protocol::schema::v1::{ContentBlock, SessionNotification, SessionUpdate};
use serde::Serialize;
use serde_json::value::RawValue;
use sqlx::SqlitePool;
use tokio::sync::mpsc;
use uuid::Uuid;

use crate::acp::chat_persistence;

/// Trailing debounce: flush after this idle gap since the last fold.
const DEBOUNCE: Duration = Duration::from_millis(250);
/// Max latency cap: never let a steady stream go unpersisted longer than this.
const MAX_LATENCY: Duration = Duration::from_millis(1000);
/// Writer command channel depth (folds only ever send the cheap `Flush` ping).
const WRITER_CHANNEL_CAPACITY: usize = 256;

/// blocks-column wrapper version for raw-frame turns (see module docs).
const RAW_FRAMES_VERSION: u64 = 1;

/// Bounded retention window for raw frames. Long turns can emit tens of thousands of
/// notifications; keeping all of them makes flush (full re-serialization) and the
/// `blocks` column grow quadratically. Only the most recent [`MAX_FRAMES`] frames are
/// kept for mid-turn recovery — the prose lives in `text` (bounded separately by
/// [`MAX_TEXT_BYTES`]), full structure comes back from the frontend's `sync` after the
/// turn ends.
///
/// This axis bounds *decode CPU* (the frontend re-classifies every retained frame on
/// hydrate), while [`MAX_BLOCKS_BYTES`] bounds I/O. Both are needed: 2000 tiny frames
/// and 28 fat ones cost the same bytes but very different classification work.
const MAX_FRAMES: usize = 2000;

/// Byte budget for the retained window — the axis that actually bounds the `blocks`
/// column. A frame-count cap only bounds bytes if frames are small, and frame size is
/// decided by the agent, not by us (AGENTS.md §8): codebuddy emits one
/// `tool_call_update` per streamed character yet re-sends the tool's **complete**
/// `rawInput` in every one of them — measured 4.5KB/frame median, so a 2000-frame
/// window became a 8.7MB single row (>97% of it duplicate copies of one `rawInput`),
/// which made `GET /messages` ship 15MB and stall ACP session switching for ~0.5s.
/// Budgeting bytes instead makes low-information frames earn fewer slots automatically,
/// so the column stays constant-order for any agent behaviour.
///
/// Trade-off: with fat frames the structural recovery window shrinks to a few dozen
/// frames. Acceptable because `text` keeps the prose (bounded, head + tail) — but only
/// because the frontend *uses* it: hydrate/`turn_snapshot` decode prepends the evicted
/// prose prefix recovered from the full `text` (exact-suffix guard), and a connection
/// that joined mid-turn skips its cooked write-back so these window remnants are never
/// persisted as cooked (see `useAcpChat` `prependEvictedProse` / `joinedMidTurn`, and
/// the 2026-08-10 plan errata 2026-08-19). Evicted *structure* (thought/tool cards) is
/// still unrecoverable mid-turn by design — the window is crash recovery, not archive.
const MAX_BLOCKS_BYTES: usize = 128 * 1024;

/// Per-frame cap: a frame larger than this never enters the window (its text, if any,
/// is still accumulated into `text`, subject to that column's own cap). Without it a
/// single oversized frame — an agent inlining a whole file into one notification —
/// would occupy the entire byte budget and make [`MAX_BLOCKS_BYTES`] no bound at all.
const MAX_FRAME_BYTES: usize = 64 * 1024;

// A frame that passes the per-frame cap must fit the window budget, otherwise the
// eviction loop below could drain the window down to nothing on every fold.
const _: () = assert!(MAX_FRAME_BYTES <= MAX_BLOCKS_BYTES);

/// Byte budget for the accumulated agent message text (the `text` column). The frame
/// window above is bounded on two axes, but `text` used to be the one structure that
/// grew without any cap at all — it was *deliberately* the unbounded fallback ("full
/// text lives in `text`"), which only holds as long as agents stay polite. They do not:
/// a measured dev-library row reached 9,150,950 characters in a 19-message session.
/// Every debounced flush re-writes the whole column, so an unbounded `text` turns one
/// long turn into O(n²) write amplification (performance-and-safety.md §P1).
///
/// Sized two orders of magnitude above the largest legitimate text observed so far
/// (36,834 characters), so folding stays a safety net rather than a routine event.
const MAX_TEXT_BYTES: usize = 1024 * 1024;

/// Head budget: the first bytes of the turn are frozen once the cap is hit, so the
/// reader keeps the *beginning* of the answer. A pure tail window (keep the last N
/// bytes) would make the prose start mid-sentence, which is the least readable option.
const TEXT_HEAD_BYTES: usize = 256 * 1024;

/// Reserved room for the omission marker, so head + marker + tail provably fits
/// [`MAX_TEXT_BYTES`] without the marker itself having to be budgeted at runtime.
const TEXT_MARKER_MAX_BYTES: usize = 96;

/// Tail budget: a sliding window over the most recent text, so the reader keeps the
/// *conclusion* of the answer.
const TEXT_TAIL_BYTES: usize = MAX_TEXT_BYTES - TEXT_HEAD_BYTES - TEXT_MARKER_MAX_BYTES;

/// Slack trimmed off in one go when the tail window overflows. Trimming exactly back to
/// the budget would memmove the whole tail on *every* subsequent chunk — the very O(n²)
/// this cap exists to prevent. Dropping a chunk of slack instead amortizes it to O(1)
/// per byte at the cost of the tail holding slightly less than its budget.
const TEXT_TAIL_TRIM_SLACK: usize = TEXT_TAIL_BYTES / 4;

/// Marker wrapped around the omitted-character count. Deliberately user-readable rather
/// than an internal sentinel: `text` doubles as the plain-text fallback when `blocks`
/// cannot be decoded (`ChatView.tsx` `toChatMessages`), so a reader must be able to tell
/// "content was dropped here" from the rendered bubble alone. Every reference
/// implementation states the omitted amount explicitly; none drop content silently.
const TEXT_OMISSION_PREFIX: &str = "\n…（已省略 ";
const TEXT_OMISSION_SUFFIX: &str = " 字符）…\n";

// The rendered text is head + marker + tail; the three budgets must fit the cap.
const _: () = assert!(TEXT_HEAD_BYTES + TEXT_MARKER_MAX_BYTES + TEXT_TAIL_BYTES <= MAX_TEXT_BYTES);
// A trim must leave a non-empty tail window, and must actually free slack — otherwise
// the tail either collapses to nothing or is memmoved on every chunk.
const _: () = assert!(TEXT_TAIL_TRIM_SLACK > 0 && TEXT_TAIL_TRIM_SLACK < TEXT_TAIL_BYTES);
// The marker's fixed part must leave room for the count (20 digits covers u64::MAX).
const _: () =
    assert!(TEXT_OMISSION_PREFIX.len() + TEXT_OMISSION_SUFFIX.len() + 20 <= TEXT_MARKER_MAX_BYTES);

#[derive(Default)]
struct TurnState {
    /// True between `begin_turn` and `finalize_turn`. Gates folding so replay /
    /// idle notifications (which never call `begin_turn`) are ignored.
    active: bool,
    /// 本 turn 的起点（单调时钟）。只在 `active` 期间被读，`begin_turn` 覆盖写。
    started: Option<Instant>,
    /// 本 turn 内**已结束**的审批等待段之和（ms）。`begin_turn` 归零。
    wait_ms: u64,
    /// 当前审批等待段的起算时刻；不变式：`is_some() == wait_depth > 0`。
    wait_since: Option<Instant>,
    /// 未决权限请求数。用计数而非布尔：agent 可并发发出多个 `session/request_permission`，
    /// 布尔会被第一个应答提前清零、把后续等待算成工作（计划 D4）。计数由 0 变正时起算
    /// 一段等待，回到 0 时累加该段。不用 id 集合是因为审批 id 取不出来 ——
    /// `on_receive_request` 回调的返回类型受 `IntoHandled` 约束，`handle_request` 无法把
    /// 它生成的 id 交给调用方，改成集合就得动 `PermissionManager` 的 API。
    wait_depth: u32,
    /// uuid of the in-progress `chat_messages` row. Created lazily on the first
    /// folded frame so a turn that emits nothing leaves no empty bubble.
    row_id: Option<String>,
    /// 上一次 `finalize_turn` 结算出的时长。与 `row_id` 同生命周期：定稿后仍可读，
    /// 下一次 `begin_turn` 才清 —— WS 层在 `mark_prompt_idle` 之后要拿它拼 `prompt_done`。
    /// `None` = 本 client 还没定稿过任何 turn。
    last_timing: Option<TurnTiming>,
    /// Raw `SessionUpdate` payloads for the active turn, in arrival order. Kept as
    /// pre-serialized [`RawValue`] so the window's byte size is known without
    /// re-formatting and flush only has to copy bytes. Bounded on both axes — frame
    /// count ([`MAX_FRAMES`]) and bytes ([`MAX_BLOCKS_BYTES`]) — front pops on overflow.
    frames: VecDeque<Box<RawValue>>,
    /// Running byte size of `frames` (sum of each frame's serialized length). Kept in
    /// sync on every push/pop so eviction never has to walk the window.
    frames_bytes: usize,
    /// Accumulated agent message text (for the NOT NULL `text` column and as a
    /// plain-text fallback). Only `AgentMessageChunk` text is collected here.
    /// Bounded by [`MAX_TEXT_BYTES`] — see [`BoundedText`].
    text: BoundedText,
    /// Per-client monotonic sequence, incremented on every folded frame. Never
    /// reset across turns; consumed by the reconnect reconciliation in a later phase.
    seq: u64,
    /// Set on fold, cleared by the writer after a flush.
    dirty: bool,
}

/// Byte-bounded accumulator for the turn's plain text.
///
/// Keeps a frozen **head** and a sliding **tail**, and counts what fell in between, so
/// the rendered value is `head + "…（已省略 N 字符）…" + tail` — the reader keeps both
/// the opening and the conclusion of the answer, and is told explicitly how much is
/// missing. Appending is amortized O(1) per byte: the head stops growing entirely and
/// the tail is trimmed in slack-sized chunks rather than on every push.
#[derive(Default)]
struct BoundedText {
    /// The turn's opening bytes; frozen once [`BoundedText::head_sealed`] is set.
    head: String,
    /// Set the moment any content lands in `tail`. The head must then stay frozen even
    /// though a UTF-8 boundary may have left it a few bytes below its budget — topping
    /// it up later would splice newer text *in front of* older text.
    head_sealed: bool,
    /// Sliding window over the most recent bytes; front-trimmed on overflow.
    tail: String,
    /// Characters (not bytes) dropped between `head` and `tail`, reported to the user.
    omitted_chars: usize,
}

impl BoundedText {
    fn clear(&mut self) {
        self.head.clear();
        self.head_sealed = false;
        self.tail.clear();
        self.omitted_chars = 0;
    }

    fn push_str(&mut self, chunk: &str) {
        let mut rest = chunk;
        if !self.head_sealed {
            let room = TEXT_HEAD_BYTES - self.head.len();
            if rest.len() <= room {
                self.head.push_str(rest);
                return;
            }
            // Cut on a char boundary (`floor_char_boundary`): slicing mid-character
            // would panic and take down the whole ACP connection task.
            let cut = rest.floor_char_boundary(room);
            self.head.push_str(&rest[..cut]);
            self.head_sealed = true;
            rest = &rest[cut..];
        }
        // A single chunk bigger than the tail budget (an agent inlining a whole file in
        // one `AgentMessageChunk`) must be trimmed *before* it lands: pushing it whole
        // and trimming afterwards would spike the buffer to the chunk's size, making the
        // cap no cap at all.
        if rest.len() > TEXT_TAIL_BYTES {
            let cut = rest.ceil_char_boundary(rest.len() - TEXT_TAIL_BYTES);
            self.omitted_chars += rest[..cut].chars().count();
            rest = &rest[cut..];
        }
        self.tail.push_str(rest);
        if self.tail.len() > TEXT_TAIL_BYTES {
            // Trim down to budget *minus slack* so the next chunks are free — trimming
            // back to exactly the budget would memmove the tail on every single push.
            let target = TEXT_TAIL_BYTES - TEXT_TAIL_TRIM_SLACK;
            let cut = self.tail.ceil_char_boundary(self.tail.len() - target);
            self.omitted_chars += self.tail[..cut].chars().count();
            self.tail.drain(..cut);
        }
    }

    fn render(&self) -> String {
        if !self.head_sealed {
            return self.head.clone();
        }
        let mut out =
            String::with_capacity(self.head.len() + TEXT_MARKER_MAX_BYTES + self.tail.len());
        out.push_str(&self.head);
        // `omitted_chars == 0` means the text merely outgrew the head budget without
        // anything being dropped yet — head + tail is still the complete text, so
        // claiming an omission would be a lie.
        if self.omitted_chars > 0 {
            out.push_str(TEXT_OMISSION_PREFIX);
            out.push_str(&self.omitted_chars.to_string());
            out.push_str(TEXT_OMISSION_SUFFIX);
        }
        out.push_str(&self.tail);
        out
    }
}

/// DB destination for persistence. Absent until [`TurnAccumulator::attach_persistence`]
/// — capability-probe clients never attach, so their folds are in-memory no-ops.
struct Sink {
    /// `Flush` 信号通道（有界、可丢）：writer 只需知道"有活没干"，满队列本身就是
    /// 待办信号，丢一个 ping 无损——防抖语义本来就会合并。
    cmd_tx: mpsc::Sender<WriterCmd>,
    /// `EndTurn` 专用通道（无界、可靠）。定稿跃迁**不是**可合并信号：丢了就同时丢
    /// 会话级记账（`turn_count`/`work_ms`/`wait_ms`）与消息行定稿（行永久停在
    /// `streaming`），且无日志——UI 侧仍显示耗时（`turn_timing` 读发送前就写好的
    /// `last_timing`），DB 与界面静默分叉。故不共享 `cmd_tx` 的余量：高帧率折叠能在
    /// writer 一次 DB 写内塞满 `WRITER_CHANNEL_CAPACITY`，`try_send` 届时静默失败
    /// （2026-09-21 审查发现的缺口，回归测试钉住）。
    ///
    /// 无界在此是安全的（§P1）：条目速率 = turn 速率（人工发起），writer 存活时积压量
    /// ≈ 停滞时长 ÷ turn 间隔 × 定稿载荷（百字节级）；writer 死亡则通道关闭，`send`
    /// 返回 Err 并记 WARN——不会无限增长，也不会静默。
    end_tx: mpsc::UnboundedSender<WriterCmd>,
}

enum WriterCmd {
    /// 防抖信号：只表达"该写了"。满队列时可丢——队列里已有待办信号，防抖语义
    /// 本来就会合并（`send_cmd` 的 try_send 注释）。
    Flush,
    /// turn 已定稿：落该 turn 的时长。经 [`Sink::end_tx`] **可靠投递**（可丢的
    /// `Flush` 信号通道不承载它），不要求 `row_id` —— 一帧未发的空 turn 同样消耗了
    /// 时间，会话级累计不能因为它缺一格 `turn_count`；`row_id` 只决定消息行那半边
    /// 写不写。
    EndTurn(TurnTiming),
}

/// turn 定稿时刻算好的时长。两个消费者：writer 循环据此落库，WS 层把它随
/// `prompt_done` 下发（耗时在定稿那一刻就显示，不必等刷新读 DB）。
///
/// 跨通道只传**已结算**的 `work_ms`/`wait_ms`（而非墙钟原值）：`wait_ms ≤ wall_ms` 的
/// clamp 在 `finalize_turn` 锁内做完，接收方就无需再减一次、也就不存在下溢路径。
#[derive(Debug, Clone, Serialize)]
pub struct TurnTiming {
    /// agent 实际工作时长 = `wall_ms - wait_ms`。
    pub work_ms: u64,
    /// 其中等真人审批的时长。
    pub wait_ms: u64,
    /// 本 turn 的 `chat_messages` 行 id，`None` = 未折叠任何帧的空 turn。
    /// 不下发：`prompt_done` 帧另有顶层 `row_id` 承载同一个值。
    #[serde(skip)]
    pub row_id: Option<String>,
}

pub struct TurnAccumulator {
    inner: Mutex<TurnState>,
    sink: Mutex<Option<Sink>>,
}

/// Snapshot of the row to persist, taken under the lock and handed to the writer.
struct FlushSnapshot {
    row_id: String,
    text: String,
    blocks: String,
    last_seq: i64,
}

impl TurnAccumulator {
    pub fn new() -> Self {
        Self { inner: Mutex::new(TurnState::default()), sink: Mutex::new(None) }
    }

    /// Wire up persistence and spawn the debounce writer. Called once at real session
    /// registration (create-session / load_session restore). Idempotent-ish: a second
    /// call replaces the sink and spawns a new writer (not expected in practice).
    pub fn attach_persistence(self: &Arc<Self>, db: SqlitePool, db_session_id: String) {
        let (cmd_tx, cmd_rx) = mpsc::channel(WRITER_CHANNEL_CAPACITY);
        let (end_tx, end_rx) = mpsc::unbounded_channel();
        if let Ok(mut guard) = self.sink.lock() {
            *guard = Some(Sink { cmd_tx, end_tx });
        }
        let acc = self.clone();
        tokio::spawn(writer_loop(acc, db, db_session_id, cmd_rx, end_rx));
    }

    /// Open a new turn. Resets per-turn state; `seq` stays monotonic.
    pub fn begin_turn(&self) {
        if let Ok(mut st) = self.inner.lock() {
            st.active = true;
            st.started = Some(Instant::now());
            st.wait_ms = 0;
            st.wait_since = None;
            st.wait_depth = 0;
            st.row_id = None;
            st.last_timing = None;
            st.frames.clear();
            st.frames_bytes = 0;
            st.text.clear();
            st.dirty = false;
        }
    }

    /// Fold one agent notification into the active turn. No-op when no turn is active
    /// (e.g. `load_session` replay frames), keeping replay out of live persistence.
    ///
    /// Returns the monotonic `seq` assigned to this frame when a turn is active, or
    /// `None` otherwise. The caller stamps this seq onto the broadcast payload so
    /// reconnecting clients can reconcile live frames against a `turn_snapshot`.
    pub fn fold(&self, notification: &SessionNotification) -> Option<u64> {
        let seq = {
            let Ok(mut st) = self.inner.lock() else { return None };
            if !st.active {
                return None;
            }
            st.seq += 1;
            if st.row_id.is_none() {
                st.row_id = Some(Uuid::new_v4().to_string());
            }
            if let Some(text) = agent_message_text(&notification.update) {
                st.text.push_str(text);
            }
            match serde_json::value::to_raw_value(&notification.update) {
                Ok(frame) => {
                    let bytes = frame.get().len();
                    if bytes > MAX_FRAME_BYTES {
                        // Oversized single frame: never enters the window (see
                        // MAX_FRAME_BYTES). Its text is already in `st.text`.
                        tracing::debug!(
                            bytes,
                            "dropping oversized session update frame from turn window"
                        );
                    } else {
                        st.frames.push_back(frame);
                        st.frames_bytes += bytes;
                        // Bounded window on both axes: drop oldest frames until the
                        // frame-count and byte budgets both hold. `text` accumulates
                        // full agent text separately, so evicting a thought/tool frame
                        // only narrows the mid-turn recovery window.
                        while st.frames.len() > MAX_FRAMES || st.frames_bytes > MAX_BLOCKS_BYTES {
                            match st.frames.pop_front() {
                                Some(old) => st.frames_bytes -= old.get().len(),
                                None => break,
                            }
                        }
                    }
                }
                Err(e) => {
                    tracing::warn!("failed to serialize session update for persistence: {}", e)
                }
            }
            st.dirty = true;
            st.seq
        };
        self.send_cmd(WriterCmd::Flush);
        Some(seq)
    }

    /// 一段「等真人审批」开始：权限请求进入未决态。计数 +1，由 0 变正时起算计时。
    ///
    /// 无活跃 turn 时 no-op —— 审批可能落在 prompt 区间之外（上一 turn 已定稿），那时它
    /// 不属于任何 turn 的墙钟，计进去会把等待泄漏到下一个 turn。
    pub fn begin_wait(&self) {
        if let Ok(mut st) = self.inner.lock() {
            if !st.active {
                return;
            }
            st.wait_depth += 1;
            if st.wait_since.is_none() {
                st.wait_since = Some(Instant::now());
            }
        }
    }

    /// 一段等待里**有一个**审批得到应答：计数 -1，回到 0 才累加该段等待。
    ///
    /// 调用方须只在审批确实被消费时才调用（`PermissionManager::resolve` 返回 `false`
    /// 表示该 id 早已不在未决表中），否则计数会向下漂移。这里 saturating 而非断言，是
    /// 宁可少扣一段也不让计数变负 —— 负数会让之后整段等待都不被扣除。
    pub fn end_wait(&self) {
        let Ok(mut st) = self.inner.lock() else { return };
        if !st.active {
            return;
        }
        st.wait_depth = st.wait_depth.saturating_sub(1);
        if st.wait_depth == 0
            && let Some(since) = st.wait_since.take()
        {
            st.wait_ms += elapsed_ms(since);
        }
    }

    /// 未决审批被一次性清空：`session/cancel` 规范要求以 `Cancelled` 应答所有 pending 的
    /// `session/request_permission`。此刻等待结束，之后 agent 若在收尾则是工作不是等待。
    pub fn end_all_waits(&self) {
        let Ok(mut st) = self.inner.lock() else { return };
        if !st.active {
            return;
        }
        st.wait_depth = 0;
        if let Some(since) = st.wait_since.take() {
            st.wait_ms += elapsed_ms(since);
        }
    }

    /// Close the active turn and finalize its row. Idempotent: a no-op if no turn is
    /// active, so racing callers (send_prompt completion / cancel / crash watcher) are safe.
    ///
    /// 时长记账挂在同一个 `active → false` 跃迁上：赢得 CAS 的调用者才发 `EndTurn`，
    /// 于是每个 turn 恰好被累计一次（多调用方并发定稿不会重复计数）。
    pub fn finalize_turn(&self) {
        let timing = {
            let Ok(mut st) = self.inner.lock() else { return };
            if !st.active {
                return;
            }
            st.active = false;
            let wall_ms = st.started.map_or(0, elapsed_ms);
            // 定稿时审批可能仍挂着（用户点取消 / reaper 卡死兜底），那一段若不结算就会被
            // 算成工作 —— 正是扣除等待想消除的失真。截到定稿时刻（计划 D5）。
            let mut wait_ms = st.wait_ms;
            if let Some(since) = st.wait_since.take() {
                wait_ms += elapsed_ms(since);
            }
            // 各段等待都落在 turn 内，理论上不会超过墙钟；clamp 兜住舍入误差，让唯一
            // 一处减法（下一行）恒不下溢。
            let wait_ms = wait_ms.min(wall_ms);
            st.wait_depth = 0;
            st.wait_ms = 0;
            let timing =
                TurnTiming { work_ms: wall_ms - wait_ms, wait_ms, row_id: st.row_id.clone() };
            st.last_timing = Some(timing.clone());
            timing
        };
        self.send_end_turn(timing);
    }

    /// `Flush` 信号：best-effort。writer 只需知道有活没干，满队列本身就是待办信号
    /// （防抖语义合并），通道关闭（sink 被替换 / client 已释放）则无需再写。
    fn send_cmd(&self, cmd: WriterCmd) {
        if let Ok(guard) = self.sink.lock()
            && let Some(sink) = guard.as_ref()
        {
            // try_send: the writer only needs to know work is pending; a full queue
            // already has a pending signal, and a closed queue means no sink.
            let _ = sink.cmd_tx.try_send(cmd);
        }
    }

    /// `EndTurn` 定稿跃迁：可靠投递。无界通道的 `send` 只在 writer 已消失时失败，
    /// 此时必须留痕——定稿命令被静默丢弃会同时丢掉会话级记账与消息行定稿，而 UI 仍
    /// 显示耗时（`turn_timing` 读发送前就写好的 `last_timing`），形成 DB 与界面静默
    /// 分叉。曾经它与 `Flush` 共用一条 `try_send` 通道，信号通道被高帧率折叠塞满时
    /// 被静默丢弃（2026-09-21 审查发现，回归测试钉住）。
    fn send_end_turn(&self, timing: TurnTiming) {
        if let Ok(guard) = self.sink.lock()
            && let Some(sink) = guard.as_ref()
            && let Err(e) = sink.end_tx.send(WriterCmd::EndTurn(timing))
        {
            tracing::warn!("turn EndTurn 无法送达（writer 已退出）: {}", e);
        }
    }

    /// Take a snapshot of the current row for persistence, or None if nothing was folded.
    fn snapshot(&self) -> Option<FlushSnapshot> {
        let st = self.inner.lock().ok()?;
        let row_id = st.row_id.clone()?;
        Some(FlushSnapshot {
            row_id,
            text: st.text.render(),
            blocks: wrap_frames(&st.frames),
            last_seq: st.seq as i64,
        })
    }

    /// The DB row id of the current (or just-finished) turn; `None` until the first frame
    /// is folded. Deliberately separate from [`Self::turn_snapshot`], which clones the
    /// whole `text` and re-serializes the frame window (up to `MAX_BLOCKS_BYTES`) — far
    /// too much work when a caller only needs the id. Survives `finalize_turn` (only
    /// `begin_turn` clears it), so a turn-end event can still report it.
    pub fn turn_row_id(&self) -> Option<String> {
        self.inner.lock().ok()?.row_id.clone()
    }

    /// 上一次定稿结算出的时长；`None` = 当前 turn 尚未定稿过（`begin_turn` 已清）或本
    /// client 从未跑完过一个 turn。与 [`Self::turn_row_id`] 同样活到下一次 `begin_turn`，
    /// 故 `mark_prompt_idle` 之后立刻读仍能拿到本 turn 的值。
    pub fn turn_timing(&self) -> Option<TurnTiming> {
        self.inner.lock().ok()?.last_timing.clone()
    }

    /// Snapshot the live turn state for a freshly-connected WS client (reconnect
    /// reconciliation). Always reports `active` and the current `seq` high-water mark;
    /// `row_id` is `None` until the first frame is folded. See the WS `turn_snapshot`
    /// frame and the frontend reconciliation in `useAcpChat`.
    pub fn turn_snapshot(&self) -> TurnSnapshot {
        let Ok(st) = self.inner.lock() else {
            return TurnSnapshot::default();
        };
        TurnSnapshot {
            active: st.active,
            row_id: st.row_id.clone(),
            text: st.text.render(),
            blocks: wrap_frames(&st.frames),
            seq: st.seq,
        }
    }
}

/// Wrap the retained frames into the `blocks` column payload
/// (`{"v":1,"frames":[...]}`, see module docs). Frames are already serialized
/// ([`RawValue`]) so this only concatenates bytes — no re-formatting per flush.
fn wrap_frames(frames: &VecDeque<Box<RawValue>>) -> String {
    #[derive(Serialize)]
    struct Wrapper<'a> {
        v: u64,
        frames: &'a VecDeque<Box<RawValue>>,
    }
    match serde_json::to_string(&Wrapper { v: RAW_FRAMES_VERSION, frames }) {
        Ok(s) => s,
        Err(e) => {
            // Concatenating valid RawValues cannot realistically fail; if it ever does,
            // persist an empty (but well-formed) window rather than a corrupt column.
            tracing::warn!("failed to serialize turn frames wrapper: {}", e);
            format!("{{\"v\":{RAW_FRAMES_VERSION},\"frames\":[]}}")
        }
    }
}

/// Live turn state handed to a connecting WS client so it can resume an in-progress
/// assistant turn without gaps or duplicates.
#[derive(Default)]
pub struct TurnSnapshot {
    pub active: bool,
    pub row_id: Option<String>,
    pub text: String,
    pub blocks: String,
    pub seq: u64,
}

impl Default for TurnAccumulator {
    fn default() -> Self {
        Self::new()
    }
}

/// 距 `since` 已过的毫秒数（单调时钟）。落库的恒为**差值**，`Instant` 本身绝不入库
/// —— 它跨进程无意义（计划「风险与缓解」）。
fn elapsed_ms(since: Instant) -> u64 {
    since.elapsed().as_millis() as u64
}

/// Extract plain text from an `AgentMessageChunk`'s text content block; None otherwise.
fn agent_message_text(update: &SessionUpdate) -> Option<&str> {
    if let SessionUpdate::AgentMessageChunk(chunk) = update
        && let ContentBlock::Text(t) = &chunk.content
    {
        return Some(&t.text);
    }
    None
}

/// 取下一条 writer 命令。`biased` 且 `end_rx` 优先：两条通道的 sender 同生命周期
/// （同一个 `Sink`），但无界通道里可能还有缓冲的 `EndTurn` 未消费——若让先关闭的
/// `cmd_rx` 返回 `None` 抢先结束循环，那条定稿命令就被丢在死通道里了。
async fn next_cmd(
    cmd_rx: &mut mpsc::Receiver<WriterCmd>,
    end_rx: &mut mpsc::UnboundedReceiver<WriterCmd>,
) -> Option<WriterCmd> {
    tokio::select! {
        biased;
        end = end_rx.recv() => end,
        flush = cmd_rx.recv() => flush,
    }
}

/// Debounce writer: coalesces `Flush` pings into a bounded number of SQLite writes,
/// and performs the final flush + row finalize + session work-time accumulation on
/// [`WriterCmd::EndTurn`].
async fn writer_loop(
    acc: Arc<TurnAccumulator>,
    db: SqlitePool,
    session_id: String,
    mut cmd_rx: mpsc::Receiver<WriterCmd>,
    mut end_rx: mpsc::UnboundedReceiver<WriterCmd>,
) {
    // Instant of the first un-flushed fold in the current pending window.
    let mut pending_since: Option<Instant> = None;

    loop {
        let cmd = if let Some(since) = pending_since {
            // A flush is pending: wait up to the debounce gap, but never exceed the
            // max-latency cap measured from the first pending fold.
            let elapsed = since.elapsed();
            let wait = if elapsed >= MAX_LATENCY {
                Duration::ZERO
            } else {
                DEBOUNCE.min(MAX_LATENCY - elapsed)
            };
            match tokio::time::timeout(wait, next_cmd(&mut cmd_rx, &mut end_rx)).await {
                Err(_elapsed) => {
                    // Debounce window elapsed with pending work → write once.
                    flush_once(&acc, &db, &session_id).await;
                    pending_since = None;
                    continue;
                }
                Ok(None) => break, // all senders dropped
                Ok(Some(c)) => c,
            }
        } else {
            match next_cmd(&mut cmd_rx, &mut end_rx).await {
                Some(c) => c,
                None => break,
            }
        };

        match cmd {
            WriterCmd::Flush => {
                if pending_since.is_none() {
                    pending_since = Some(Instant::now());
                }
            }
            WriterCmd::EndTurn(t) => {
                // 定稿前刷完最后的帧，使行内容与时长对应同一个 turn。
                // 顺手消费已排队的 Flush 信号：它们要的"写一次"由下面这次 flush_once
                // 完成（快照取的是活状态，是其超集），留着只会让定稿后再触发一次内容
                // 相同的冗余写。
                while cmd_rx.try_recv().is_ok() {}
                flush_once(&acc, &db, &session_id).await;
                pending_since = None;
                let (work, wait) = (t.work_ms as i64, t.wait_ms as i64);
                if let Some(row_id) = &t.row_id
                    && let Err(e) =
                        chat_persistence::finalize_message(&db, row_id, work, wait).await
                {
                    tracing::warn!("failed to finalize streaming chat message: {}", e);
                }
                // 会话累计无条件写：空 turn 没有消息行，但那段时间确实发生了。
                if let Err(e) =
                    chat_persistence::accumulate_turn(&db, &session_id, work, wait).await
                {
                    tracing::warn!("failed to accumulate session work time: {}", e);
                }
            }
        }
    }
}

async fn flush_once(acc: &Arc<TurnAccumulator>, db: &SqlitePool, session_id: &str) {
    let Some(snap) = acc.snapshot() else { return };
    if let Err(e) = chat_persistence::upsert_streaming_message(
        db,
        &snap.row_id,
        session_id,
        &snap.text,
        Some(&snap.blocks),
        snap.last_seq,
    )
    .await
    {
        tracing::warn!("failed to upsert streaming chat message: {}", e);
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use agent_client_protocol::schema::v1::{ContentChunk, SessionId, TextContent};
    use serde_json::Value;

    /// Wrapper overhead of `{"v":1,"frames":[]}` plus the `,` separators between frames.
    /// The byte budget applies to the frames themselves, so assertions allow this slack.
    const WRAPPER_SLACK: usize = 4096;

    fn text_chunk(sid: &SessionId, text: impl Into<String>) -> SessionNotification {
        SessionNotification::new(
            sid.clone(),
            SessionUpdate::AgentMessageChunk(ContentChunk::new(ContentBlock::Text(
                TextContent::new(text.into()),
            ))),
        )
    }

    fn retained_frames(acc: &TurnAccumulator) -> (usize, usize) {
        let st = acc.inner.lock().expect("lock");
        (st.frames.len(), st.frames_bytes)
    }

    /// 装内存命令通道代替真实 sink，捕获发给 writer 的命令而不触库。返回
    /// `(Flush 信号通道, EndTurn 通道)`；测试深度取 64，远超本模块测试产生的命令数。
    fn capture_cmds(
        acc: &TurnAccumulator,
    ) -> (mpsc::Receiver<WriterCmd>, mpsc::UnboundedReceiver<WriterCmd>) {
        let (tx, rx) = mpsc::channel(64);
        let (end_tx, end_rx) = mpsc::unbounded_channel();
        *acc.sink.lock().expect("sink lock") = Some(Sink { cmd_tx: tx, end_tx });
        (rx, end_rx)
    }

    /// 取走 EndTurn 通道里已排队的命令，只留下载荷（每 turn 至多一条）。
    fn end_turns(rx: &mut mpsc::UnboundedReceiver<WriterCmd>) -> Vec<(u64, u64, Option<String>)> {
        let mut out = Vec::new();
        while let Ok(cmd) = rx.try_recv() {
            if let WriterCmd::EndTurn(t) = cmd {
                out.push((t.work_ms, t.wait_ms, t.row_id));
            }
        }
        out
    }

    /// 直读等待计时的中间态：`(未决审批数, 等待段是否在走, 已累加等待 ms)`。
    fn wait_state(acc: &TurnAccumulator) -> (u32, bool, u64) {
        let st = acc.inner.lock().expect("lock");
        (st.wait_depth, st.wait_since.is_some(), st.wait_ms)
    }

    /// 等待一段可被毫秒时钟分辨的时间。阈值一律取下限的 6 成，避免调度抖动致脆。
    fn nap(ms: u64) {
        std::thread::sleep(Duration::from_millis(ms));
    }

    /// Internal text buffers: `(head bytes, tail bytes, omitted chars)`. Asserted
    /// directly because bounding only the *rendered* value would still leave the
    /// in-memory accumulation — and therefore every debounced re-write — unbounded.
    fn text_buffers(acc: &TurnAccumulator) -> (usize, usize, usize) {
        let st = acc.inner.lock().expect("lock");
        (st.text.head.len(), st.text.tail.len(), st.text.omitted_chars)
    }

    /// Split a rendered text into `(head, omitted_chars, tail)`; `None` when unfolded.
    fn split_folded(text: &str) -> Option<(&str, usize, &str)> {
        let (head, rest) = text.split_once(TEXT_OMISSION_PREFIX)?;
        let (count, tail) = rest.split_once(TEXT_OMISSION_SUFFIX)?;
        Some((head, count.parse().expect("折叠标记应包含可解析的省略字符数"), tail))
    }

    /// Fold `unit` repeatedly until at least `total_bytes` of text has been fed in.
    /// Returns the full text that was fed, for conservation assertions.
    fn fold_until(
        acc: &TurnAccumulator,
        sid: &SessionId,
        unit: &str,
        total_bytes: usize,
    ) -> String {
        let mut fed = String::new();
        while fed.len() < total_bytes {
            acc.fold(&text_chunk(sid, unit));
            fed.push_str(unit);
        }
        fed
    }

    fn parse_frames(blocks: &str) -> Vec<Value> {
        let wrapper: Value = serde_json::from_str(blocks).expect("blocks 应为合法 JSON");
        wrapper["frames"].as_array().expect("frames 应为数组").clone()
    }

    /// `turn_row_id` must survive `finalize_turn`: the WS layer reads it *after* the turn
    /// is finalized in order to hand it to the frontend in `prompt_done`, which is what
    /// lets the frontend write cooked blocks back to that exact row. Only `begin_turn`
    /// may clear it — otherwise every turn would end with "no row to write to".
    #[test]
    fn row_id_is_readable_after_finalize_and_reset_by_next_turn() {
        let acc = TurnAccumulator::new();
        let sid = SessionId::new("s1");

        assert_eq!(acc.turn_row_id(), None, "未开始 turn 时无行 id");
        acc.begin_turn();
        assert_eq!(acc.turn_row_id(), None, "首帧折叠前仍无行 id（行是惰创建的）");

        acc.fold(&text_chunk(&sid, "hello"));
        let row_id = acc.turn_row_id().expect("首帧折叠后应有行 id");
        assert_eq!(acc.turn_snapshot().row_id.as_deref(), Some(row_id.as_str()), "与快照同源");

        acc.finalize_turn();
        assert_eq!(acc.turn_row_id().as_deref(), Some(row_id.as_str()), "定稿后仍可读到");

        acc.begin_turn();
        assert_eq!(acc.turn_row_id(), None, "下一 turn 开始才清除");
    }

    /// 时长载荷的存活契约与 `turn_row_id` 同源：WS 层在 `mark_prompt_idle` **之后**读它
    /// 拼 `prompt_done` 的 `duration`，前端据此当场显示耗时。
    #[test]
    fn timing_is_readable_after_finalize_and_reset_by_next_turn() {
        let acc = TurnAccumulator::new();
        assert!(acc.turn_timing().is_none(), "从未定稿过 → None");

        acc.begin_turn();
        assert!(acc.turn_timing().is_none(), "turn 进行中不得读到上一 turn 的结算值");
        nap(20);
        acc.finalize_turn();
        let t = acc.turn_timing().expect("定稿后 WS 层仍要能读到");
        assert!(t.work_ms >= 12, "墙钟应记进 work_ms，实得 {}", t.work_ms);
        assert_eq!(t.wait_ms, 0, "本 turn 没有审批等待");

        acc.begin_turn();
        assert!(acc.turn_timing().is_none(), "下一 turn 开始才清除");
    }

    /// 长 turn 高频帧（thought/tool chunk）不能把 frames 撑成无界：窗口在帧数与字节
    /// 两个维度上都有上限，text 仍全量累积。回归防护 2026-08-04 线上问题（单条 blocks
    /// 累积到 100MB、tokio worker 99% CPU、进程 RES 4.5GB）。
    #[test]
    fn window_is_bounded_on_both_axes_over_long_turn() {
        const FOLD_COUNT: u32 = 5000;
        assert!(FOLD_COUNT as usize > MAX_FRAMES, "测试需超过窗口以触发淘汰");

        let acc = TurnAccumulator::new();
        acc.begin_turn();
        let sid = SessionId::new("s1");
        for i in 0..FOLD_COUNT {
            assert!(
                acc.fold(&text_chunk(&sid, format!("chunk-{i} "))).is_some(),
                "turn active 时应折叠出 seq"
            );
        }

        let (len, bytes) = retained_frames(&acc);
        assert!(len <= MAX_FRAMES, "frames 不得超帧数上限，得到 {len}");
        assert!(bytes <= MAX_BLOCKS_BYTES, "frames 不得超字节上限，得到 {bytes}");

        let snap = acc.turn_snapshot();
        assert!(
            snap.blocks.len() <= MAX_BLOCKS_BYTES + WRAPPER_SLACK,
            "blocks 列不随 turn 长度增长，得到 {} 字节",
            snap.blocks.len()
        );
        assert_eq!(parse_frames(&snap.blocks).len(), len, "包裹后的帧数应等于窗口帧数");

        // text 在未触及 [`MAX_TEXT_BYTES`] 前全量累积：丢弃的只是 UI 恢复窗口，正文不受影响。
        assert!(
            snap.text.len() < MAX_TEXT_BYTES,
            "本例输入量应远低于正文上限，否则下方全量断言不成立"
        );
        assert!(snap.text.starts_with("chunk-0 "), "text 保留最早 chunk");
        assert!(snap.text.ends_with("chunk-4999 "), "text 保留最新 chunk");
        assert_eq!(snap.text.matches("chunk-").count(), FOLD_COUNT as usize);
    }

    /// 正文超过 [`MAX_TEXT_BYTES`] 时必须收敛，且**头尾都保留**、省略量显式可读。
    /// 每次 debounce flush 都会重写整个 `text` 列，无界正文即 O(n²) 写放大
    /// （performance-and-safety.md §P1）。
    #[test]
    fn long_text_folds_to_head_and_tail_within_cap() {
        let acc = TurnAccumulator::new();
        acc.begin_turn();
        let sid = SessionId::new("s1");

        // 每条 chunk 带序号，以便区分“保留了开头”与“保留了结尾”。
        let mut fed = String::new();
        let mut i = 0usize;
        while fed.len() < MAX_TEXT_BYTES * 2 {
            let chunk = format!("line-{i:08}-{}\n", "f".repeat(48));
            acc.fold(&text_chunk(&sid, chunk.clone()));
            fed.push_str(&chunk);
            i += 1;
        }

        let text = acc.turn_snapshot().text;
        assert!(text.len() <= MAX_TEXT_BYTES, "正文不得超字节上限，得到 {} 字节", text.len());

        let (head, omitted, tail) = split_folded(&text).expect("超限后应出现可读的折叠标记");
        assert!(omitted > 0, "折叠标记必须报出实际省略量");
        assert!(head.starts_with("line-00000000-"), "头部必须是正文开头（约 {head:.20}）");
        assert!(
            tail.ends_with(&format!("line-{:08}-{}\n", i - 1, "f".repeat(48))),
            "尾部必须是正文结尾"
        );

        // 守恒：保留的 + 声称省略的 == 全部输入。静默丢弃无人采用（D3）。
        assert_eq!(
            head.chars().count() + omitted + tail.chars().count(),
            fed.chars().count(),
            "保留量与省略量之和必须等于输入量（不得静默丢弃）"
        );

        // 内存侧也必须有界：仅限界渲染值会让累积缓冲与每次重写仍然无界。
        let (head_bytes, tail_bytes, _) = text_buffers(&acc);
        assert!(head_bytes <= TEXT_HEAD_BYTES, "头部缓冲超预算：{head_bytes}");
        assert!(tail_bytes <= TEXT_TAIL_BYTES, "尾窗缓冲超预算：{tail_bytes}");
    }

    /// 未达上限时正文必须逐字节原样 —— 折叠是安全网，不得污染常规路径。
    #[test]
    fn text_under_cap_is_verbatim() {
        let acc = TurnAccumulator::new();
        acc.begin_turn();
        let sid = SessionId::new("s1");
        acc.fold(&text_chunk(&sid, "你好，"));
        acc.fold(&text_chunk(&sid, "world \u{1F642}"));

        let text = acc.turn_snapshot().text;
        assert_eq!(text, "你好，world \u{1F642}");
        assert!(!text.contains(TEXT_OMISSION_PREFIX), "未超限不得出现折叠标记");
    }

    /// 头部封口与尾窗修剪都会落在任意位置，必须按 UTF-8 字符边界切 —— 切在多字节
    /// 字符中间在 Rust 里直接 panic（`String` 无法持有非法 UTF-8），在生产里就是折断
    /// 整个 ACP 连接任务。单元长 16 字节且与各预算不整除，确保边界落在字符内部。
    #[test]
    fn folding_never_splits_multibyte_chars() {
        const UNIT: &str = "中文测试\u{1F642}"; // 3*4 + 4 = 16 字节
        let acc = TurnAccumulator::new();
        acc.begin_turn();
        let sid = SessionId::new("s1");
        let fed = fold_until(&acc, &sid, UNIT, MAX_TEXT_BYTES * 2);

        let text = acc.turn_snapshot().text;
        assert!(text.len() <= MAX_TEXT_BYTES, "多字节正文仍须守住上限：{}", text.len());

        let (head, omitted, tail) = split_folded(&text).expect("超限后应折叠");
        let unit_chars: Vec<char> = UNIT.chars().collect();
        for (label, part) in [("头部", head), ("尾部", tail)] {
            assert!(
                part.chars().all(|c| unit_chars.contains(&c)),
                "{label}出现不属于输入字符集的字符（说明字符被切坏）"
            );
            assert!(!part.contains('\u{FFFD}'), "{label}不得出现替换字符");
        }
        assert_eq!(
            head.chars().count() + omitted + tail.chars().count(),
            fed.chars().count(),
            "字符级守恒在多字节下同样成立"
        );
    }

    /// 单个 chunk 本身就超过上限（agent 把整份文件塞进一个 AgentMessageChunk）时，不得先把
    /// 它整段推进缓冲再修剪 —— 那会让内存峰值等于 chunk 大小，上限形同虚设。
    #[test]
    fn single_oversized_chunk_is_bounded_on_arrival() {
        let acc = TurnAccumulator::new();
        acc.begin_turn();
        let sid = SessionId::new("s1");
        let huge = "x".repeat(MAX_TEXT_BYTES * 3);
        acc.fold(&text_chunk(&sid, huge.clone()));

        let (head_bytes, tail_bytes, _) = text_buffers(&acc);
        assert!(head_bytes <= TEXT_HEAD_BYTES, "头部缓冲超预算：{head_bytes}");
        assert!(tail_bytes <= TEXT_TAIL_BYTES, "尾窗缓冲超预算：{tail_bytes}");

        let text = acc.turn_snapshot().text;
        assert!(text.len() <= MAX_TEXT_BYTES, "超大单帧正文仍须守住上限：{}", text.len());
        let (head, omitted, tail) = split_folded(&text).expect("超限后应折叠");
        assert_eq!(head.chars().count() + omitted + tail.chars().count(), huge.chars().count());
    }

    /// 新 turn 必须重置折叠状态：否则上一个长 turn 的封口标志与省略计数会泄漏到下一
    /// 个 turn，让一句短回答迷不丁地带上“已省略 N 字符”。
    #[test]
    fn begin_turn_resets_text_folding_state() {
        const UNIT: &str = "旧 turn 的冗长正文 ";
        let acc = TurnAccumulator::new();
        let sid = SessionId::new("s1");
        acc.begin_turn();
        fold_until(&acc, &sid, UNIT, MAX_TEXT_BYTES + TEXT_TAIL_BYTES);
        assert!(text_buffers(&acc).2 > 0, "前一个 turn 应已发生折叠");

        acc.begin_turn();
        acc.fold(&text_chunk(&sid, "新 turn"));
        assert_eq!(text_buffers(&acc), ("新 turn".len(), 0, 0), "新 turn 应从空缓冲开始");
        assert_eq!(acc.turn_snapshot().text, "新 turn");
    }

    /// 折叠标记必须装得进为它预留的预算，否则 head + marker + tail 会溢出总上限；
    /// 预算里的 20 位数字余量由 const 断言保证，这里取最坏情况实测。
    #[test]
    fn omission_marker_fits_reserved_budget() {
        let widest = format!("{TEXT_OMISSION_PREFIX}{}{TEXT_OMISSION_SUFFIX}", usize::MAX);
        assert!(
            widest.len() <= TEXT_MARKER_MAX_BYTES,
            "标记最坏情况 {} 字节超出预算 {TEXT_MARKER_MAX_BYTES}",
            widest.len()
        );
    }

    /// 回归防护 2026-08-10：codebuddy 每个 tool_call_update 只带 1 字符增量，却重复携带
    /// 完整 rawInput（4.5KB/帧）——帧数上限在这种帧下完全拦不住体积（2000 帧 = 8.7MB
    /// 单行，使 GET /messages 下发 15MB、切会话卡顿 0.5s）。字节预算必须先生效。
    #[test]
    fn fat_repeated_frames_stay_within_byte_budget() {
        const FRAME_TEXT_BYTES: usize = 4500;
        const FOLD_COUNT: usize = 500;

        let acc = TurnAccumulator::new();
        acc.begin_turn();
        let sid = SessionId::new("s1");
        let fat = "x".repeat(FRAME_TEXT_BYTES);
        for _ in 0..FOLD_COUNT {
            acc.fold(&text_chunk(&sid, fat.clone()));
        }

        let (len, bytes) = retained_frames(&acc);
        assert!(bytes <= MAX_BLOCKS_BYTES, "肥帧下字节预算仍须守住，得到 {bytes}");
        assert!(
            len < FOLD_COUNT,
            "字节维度应已触发淘汰（帧数上限 {MAX_FRAMES} 在 {FOLD_COUNT} 肥帧下根本拦不住），得到 {len} 帧"
        );
        assert!(len > 0, "单帧未超限时窗口不得被清空");

        let snap = acc.turn_snapshot();
        assert!(
            snap.blocks.len() <= MAX_BLOCKS_BYTES + WRAPPER_SLACK,
            "blocks 列应被字节预算钉住，得到 {} 字节",
            snap.blocks.len()
        );
    }

    /// 单帧自身超过 [`MAX_FRAME_BYTES`]（agent 把整份文件塞进一帧）时不得入窗，否则它
    /// 会独占整个字节预算使上限形同虚设；text 仍兼底正文。
    #[test]
    fn oversized_frame_never_enters_window() {
        let acc = TurnAccumulator::new();
        acc.begin_turn();
        let sid = SessionId::new("s1");
        let huge = "y".repeat(MAX_FRAME_BYTES + 1);
        assert!(acc.fold(&text_chunk(&sid, huge.clone())).is_some(), "帧被丢弃也应分配 seq");

        let (len, bytes) = retained_frames(&acc);
        assert_eq!(len, 0, "超大单帧不得入窗");
        assert_eq!(bytes, 0, "丢弃的帧不得计入字节账");

        let snap = acc.turn_snapshot();
        assert_eq!(snap.text, huge, "正文文本仍全量累积");
        assert!(parse_frames(&snap.blocks).is_empty(), "blocks 应为合法的空窗口");
    }

    /// 新 turn 必须重置字节账，否则跨 turn 累积会让窗口逐渐营养不良（预算被已丢弃
    /// 的旧帧占满，新 turn 一 fold 就被淘汰到只剩一帧）。
    #[test]
    fn begin_turn_resets_byte_accounting() {
        let acc = TurnAccumulator::new();
        let sid = SessionId::new("s1");
        acc.begin_turn();
        for _ in 0..50 {
            acc.fold(&text_chunk(&sid, "z".repeat(4096)));
        }
        assert!(retained_frames(&acc).1 > 0, "前一个 turn 应有累积");

        acc.begin_turn();
        assert_eq!(retained_frames(&acc), (0, 0), "新 turn 应清空帧与字节账");
    }

    // ---- 工作时长记账（docs/dev/plans/archive/2026-08-30-acp-work-time.md）----

    /// 并发定稿只能记一次账：多个收尾路径（prompt 完成 / cancel 兜底 / reaper /
    /// 崩溃看护）都会调 `finalize_turn`，若不去重就会把同一 turn 累计多遍。
    #[test]
    fn repeated_finalize_turn_emits_exactly_one_end_turn() {
        let acc = TurnAccumulator::new();
        let (_flush_rx, mut end_rx) = capture_cmds(&acc);
        let sid = SessionId::new("s1");

        acc.begin_turn();
        acc.fold(&text_chunk(&sid, "hello"));
        acc.finalize_turn();
        acc.finalize_turn();
        acc.finalize_turn();

        let turns = end_turns(&mut end_rx);
        assert_eq!(turns.len(), 1, "重复定稿不得重复发记账命令，得到 {turns:?}");
        assert!(turns[0].2.is_some(), "折叠过帧的 turn 应携带消息行 id");
    }

    /// 深度模型：并发未决审批下，只应答其中一个时等待**不能**提前结束（布尔标志会错），
    /// 全部应答完才累加那一段。
    #[test]
    fn concurrent_permits_hold_one_wait_segment_until_all_resolve() {
        let acc = TurnAccumulator::new();
        acc.begin_turn();

        acc.begin_wait();
        acc.begin_wait();
        nap(50);
        acc.end_wait();
        let (depth, running, waited) = wait_state(&acc);
        assert_eq!(depth, 1, "还剩一个未决审批");
        assert!(running, "仍有未决审批时等待段不得结束");
        assert_eq!(waited, 0, "集合未清空前不得累加，得到 {waited}");

        acc.end_wait();
        let (depth, running, waited) = wait_state(&acc);
        assert_eq!((depth, running), (0, false), "全部应答后等待段应关闭");
        assert!(waited >= 30, "清空时应累加整段等待，得到 {waited}ms");

        // 多余的 end_wait（例如另一条连接重复应答）不得让计数变负、不得再开新段。
        acc.end_wait();
        assert_eq!(wait_state(&acc).0, 0, "计数下溢须被 saturating 挡住");
    }

    /// 定稿时审批仍挂着（用户点取消 / reaper 卡死兜底的真实时序）：那一段等待必须
    /// 截到定稿时刻结算，否则会被算成工作时间（计划 D5）。
    #[test]
    fn finalize_clips_an_open_wait_segment() {
        let acc = TurnAccumulator::new();
        let (_flush_rx, mut end_rx) = capture_cmds(&acc);

        acc.begin_turn();
        acc.begin_wait();
        nap(50);
        acc.finalize_turn();

        let turns = end_turns(&mut end_rx);
        assert_eq!(turns.len(), 1, "一个 turn 只应有一条记账：{turns:?}");
        let (work, wait, _) = &turns[0];
        assert!(*wait >= 30, "未决审批那段应截到定稿时刻计入 wait，得到 {wait}ms");
        // 唯一一处减法在 finalize 锁内且已 clamp，故 work 恒非负、两者相加即墙钟。
        assert!(*work <= 50, "等待不得反噬工作时长，work={work} wait={wait}");
        assert!(*work + *wait >= 50, "work + wait 应覆盖整段墙钟（{work} + {wait}）");
    }

    /// 一帧未发的空 turn（agent 立即报错返回）没有消息行，但那段时间确实发生了 ——
    /// 记账命令必须照样发出，`row_id` 为 None 让 writer 跳过消息行那半边。
    #[test]
    fn empty_turn_is_still_accounted() {
        let acc = TurnAccumulator::new();
        let (_flush_rx, mut end_rx) = capture_cmds(&acc);

        acc.begin_turn();
        nap(20);
        acc.finalize_turn();

        let turns = end_turns(&mut end_rx);
        assert_eq!(turns.len(), 1, "空 turn 也要记账，得到 {turns:?}");
        assert_eq!(turns[0].2, None, "未折叠任何帧 → 无消息行 id");
    }

    /// turn 之外到达的审批（上一 turn 已定稿、agent 自行发起）不得泄漏到下一个 turn
    /// 的等待账上，否则下一个正常 turn 会被整段算成等待。
    #[test]
    fn wait_outside_an_active_turn_is_dropped() {
        let acc = TurnAccumulator::new();
        let (_flush_rx, mut end_rx) = capture_cmds(&acc);

        acc.begin_wait();
        acc.end_all_waits();
        assert_eq!(wait_state(&acc), (0, false, 0), "无活跃 turn 时两个动作都应为 no-op");

        acc.begin_turn();
        nap(20);
        acc.finalize_turn();

        let turns = end_turns(&mut end_rx);
        assert_eq!(turns.len(), 1);
        assert_eq!(turns[0].1, 0, "turn 外的审批不得算进本 turn 的等待：{:?}", turns[0]);
    }

    /// `end_all_waits`（session/cancel 会以 Cancelled 应答全部未决审批）就地结束等待段，
    /// 之后 agent 的收尾时间算工作而非等待。
    #[test]
    fn end_all_waits_closes_an_open_segment_immediately() {
        let acc = TurnAccumulator::new();
        let (_flush_rx, mut end_rx) = capture_cmds(&acc);
        acc.begin_turn();
        acc.begin_wait();
        acc.begin_wait();
        nap(30);
        acc.end_all_waits();
        let (depth, running, waited) = wait_state(&acc);
        assert_eq!((depth, running), (0, false), "全部取消后不得继续走等待表");
        assert!(waited >= 18, "取消时刻即该段终点，得到 {waited}ms");

        // 取消后 agent 继续干活的 30ms 必须归工作：若等待表没在取消时结掉，会被算进 wait。
        nap(30);
        acc.finalize_turn();
        let turns = end_turns(&mut end_rx);
        assert_eq!(turns.len(), 1, "应恰好一条 EndTurn：{turns:?}");
        let (work, wait, _) = &turns[0];
        assert!(*work >= 18, "取消后的收尾应算工作，得到 {work}ms");
        assert!(*wait < 48, "等待不得把取消后的收尾也算进去，得到 {wait}ms");
    }

    /// 回归（2026-09-21 审查发现）：`EndTurn` 曾与 `Flush` 信号共用一条 `try_send`
    /// 通道。信号通道被高帧率折叠塞满时（writer 卡在一次 DB 写内，万级帧率下 25ms
    /// 即可填满 256 深度），定稿命令被**静默丢弃**——会话级记账（`turn_count` /
    /// `work_ms` / `wait_ms`）与消息行定稿（`finalize_message`，行永久停在
    /// `streaming`）同时丢失，且无日志；UI 却照显示耗时（`turn_timing` 读发送前就
    /// 写好的 `last_timing`），DB 与界面静默分叉。定稿跃迁必须走独立可靠通道，
    /// 不依赖信号通道余量。
    #[test]
    fn end_turn_is_delivered_when_the_flush_signal_channel_is_saturated() {
        let acc = TurnAccumulator::new();
        let sid = SessionId::new("s1");
        // 容量 2 的信号通道：折叠几帧即满，复现"writer 卡在 DB 写内、ping 积压"现场。
        let (flush_tx, _flush_rx) = mpsc::channel(2);
        let probe = flush_tx.clone();
        let (end_tx, mut end_rx) = mpsc::unbounded_channel();
        *acc.sink.lock().expect("sink lock") = Some(Sink { cmd_tx: flush_tx, end_tx });

        acc.begin_turn();
        for i in 0..8 {
            acc.fold(&text_chunk(&sid, format!("chunk {i}")));
        }
        assert!(
            probe.try_send(WriterCmd::Flush).is_err(),
            "前置条件：Flush 信号通道应已被折叠 ping 塞满"
        );

        acc.finalize_turn();

        let turns = end_turns(&mut end_rx);
        assert_eq!(turns.len(), 1, "信号通道饱和时 EndTurn 仍须恰好送达一次：{turns:?}");
        assert!(turns[0].2.is_some(), "折叠过帧的 turn 应携带消息行 id");
    }
}