clustervision 0.16.0

//! `cv prune` / `cv rewind` / `cv splice` / `cv dataset` — reshaping sessions into new ones.

use crate::cmd::port::emit_session;
use crate::util::{dry_run_nulls, parse_harness, parse_range, report, resolve, short_id, usage};
use anyhow::{bail, Context, Result};
use cv_core::ir::{human_bytes, Block, Harness, Message, Role, Session};
use cv_core::EmitOptions;
use std::fmt::Write as _;
use std::fs;
use std::path::PathBuf;

// ---------- prune ----------

/// `cv prune <id>` — compact a Claude session into a new resumable one (see [`cv_core::prune`]).
/// Stashed originals come back with `cv cat <new-id> <tool_use_id>`.
#[allow(clippy::too_many_arguments)]
pub(crate) fn cmd_prune(
    id: &str,
    harness: Option<String>,
    min_size: usize,
    keep_last: usize,
    to: Option<String>,
    drop: bool,
    thinking: bool,
    copy_resources: bool,
    revive: bool,
    window: Option<u64>,
    keep_range: Option<(usize, Option<usize>)>,
    declassify: bool,
    declassify_tokens: Vec<String>,
    dry_run: bool,
    json: bool, // also emit the report as one JSON object on stdout (the human report stays on stderr)
) -> Result<()> {
    let want = parse_harness(&harness)?;
    let (r, _adapter) = resolve(id, want)?;

    if r.harness != Harness::Claude {
        bail!(
            "cv prune currently supports Claude Code sessions only (got {})",
            r.harness
        );
    }

    let pinned_id = to.is_some(); // --to: the new id is caller-chosen, so a dry run can still report it
    let opts = cv_core::prune::PruneOptions {
        min_size,
        keep_last,
        drop,
        thinking,
        new_id: to,
        copy_resources,
        revive,
        window,
        keep_range,
        declassify,
        declassify_tokens, // caller-supplied (external); cv ships no built-in list. min_hits from Default (2).
        dry_run,
        ..Default::default()
    };
    let res = cv_core::prune::prune_session(&r.path, &opts)?;

    let mut v = serde_json::to_value(&res)?;
    v["harness"] = r.harness.as_str().into();
    // revived: false when revive is off or the recorded usage was already honest; else the
    // stale → honest rewrite detail (what the human report prints as "revived: …k → …k").
    v["revived"] = match (res.revive_tokens, res.revive_old_tokens) {
        (Some(honest), Some(stale)) => serde_json::json!({
            "recorded_tokens_before": stale,
            "recorded_tokens_after": honest,
            "usage_records_rewritten": res.usage_rewritten,
        }),
        _ => serde_json::Value::Bool(false),
    };
    dry_run_nulls(
        &mut v,
        pinned_id,
        &["new_path", "sidecar_path", "copied_resources"],
        &["new_id"],
    );

    let mut h = String::new();
    let pct = if res.original_size > 0 {
        100.0 * res.bytes_saved() as f64 / res.original_size as f64
    } else {
        0.0
    };
    let tag = if dry_run { " (dry run — nothing written)" } else { "" };
    let _ = writeln!(
        h,
        "✦ pruned {} → {}{tag}",
        short_id(&res.source_id),
        short_id(&res.new_id)
    );
    let _ = writeln!(
        h,
        "  {} payload(s) snipped ({} image(s)) · {} → {} on disk (−{:.0}%) · ~{} context tokens freed",
        res.pruned_count,
        res.image_blocks,
        human_bytes(res.original_size),
        human_bytes(res.new_size),
        pct,
        res.est_context_tokens_saved,
    );
    if res.dropped_turns > 0 {
        let sized = match res.window_real_tokens {
            Some(real) => format!(
                " · kept ~{}k REAL tokens (Claude's recorded counts; resume loads ≈ this + ~30k overhead)",
                real / 1000
            ),
            None => " (sized by byte estimate — no usage records)".to_string(),
        };
        let _ = writeln!(
            h,
            "  sliding window: dropped {} older turn(s){} — the source keeps the full history",
            res.dropped_turns, sized
        );
    }
    if revive {
        match (res.revive_tokens, res.revive_old_tokens) {
            (Some(honest), Some(stale)) => {
                let _ = writeln!(
                    h,
                    "  revived: recorded context {}k → {}k across {} usage record(s) — resume gate will let you back in",
                    stale / 1000,
                    honest / 1000,
                    res.usage_rewritten,
                );
            }
            _ => {
                let _ = writeln!(
                    h,
                    "  revive: recorded usage already honest (≤ loaded content) — nothing to rewrite"
                );
            }
        }
    }
    for w in &res.warnings {
        let _ = writeln!(h, "  ⚠ {w}");
    }
    if !dry_run {
        let _ = writeln!(h, "  new session: {}", res.new_path.display());
        if let Some(sc) = &res.sidecar_path {
            let _ = writeln!(
                h,
                "  sidecar:     {} (fetch an original: cv cat {} <tool_use_id>)",
                sc.display(),
                short_id(&res.new_id)
            );
        }
        if let Some(rc) = &res.copied_resources {
            let _ = writeln!(h, "  resources:   copied subagents/workflows → {}", rc.display());
        }
        let _ = writeln!(h, "  resume with: claude --resume {}", res.new_id);
    }
    report(json, &h, &v)
}

// ---------- rewind ----------

/// The `--at` grammar: an all-digit value is a message index (the `cv show --range` counting), any
/// other 7–64 hex digits a commit sha, nothing at all the last message. (A sha whose digits happen to
/// be all decimal reads as an index — pass more of it; a full sha never is.)
pub(crate) fn parse_cut(at: Option<&str>) -> Result<cv_core::rewind::CutAt> {
    use cv_core::rewind::{normalize_sha, CutAt};
    let Some(at) = at.map(str::trim) else {
        return Ok(CutAt::End);
    };
    if !at.is_empty() && at.bytes().all(|b| b.is_ascii_digit()) {
        return match at.parse() {
            Ok(n) => Ok(CutAt::Message(n)),
            Err(_) => usage(format!("--at {at:?}: message index out of range")),
        };
    }
    match normalize_sha(at) {
        Some(sha) => Ok(CutAt::Commit(sha)),
        None => usage(format!(
            "--at {at:?}: expected a message index (as `cv show --range` counts) or a commit sha (7–64 hex digits)"
        )),
    }
}

/// `cv rewind <id>` — derive the agent as of a past message or commit (see [`cv_core::rewind`]).
/// `<id>` is any session id, an `agent-…` sub-agent id, or a transcript path; `--agent` picks a
/// sub-agent under `<id>` the way `cv show --agent` does.
#[allow(clippy::too_many_arguments)]
pub(crate) fn cmd_rewind(
    id: &str,
    harness: Option<String>,
    agent: Option<String>,
    at: Option<&str>,
    out: Option<PathBuf>,
    to: Option<String>,
    full: bool,
    dry_run: bool,
    json: bool,
    generator: &str,
) -> Result<()> {
    let cut = parse_cut(at)?;
    let path = std::path::Path::new(id);
    let r = if path.is_file() {
        // A transcript path (a sub-agent file straight from `subagents/`, or a copy elsewhere).
        cv_core::ir::SessionRef::from_path(path, Harness::Claude)
    } else {
        resolve(id, parse_harness(&harness)?)?.0
    };
    let r = match &agent {
        Some(a) => crate::cmd::view::resolve_subagent(&r, a)?.session,
        None => r,
    };
    if r.harness != Harness::Claude {
        return usage(format!(
            "cv rewind currently supports Claude Code sessions only (got {}) — `cv splice {}:..N` composes a \
             prefix of any harness",
            r.harness,
            short_id(&r.id)
        ));
    }

    let pinned = to.is_some();
    let res = cv_core::rewind::rewind_session(
        &r,
        &cv_core::rewind::RewindOptions {
            at: cut,
            full,
            out_dir: out,
            new_id: to,
            dry_run,
            generator: generator.to_string(),
        },
    )?;
    let cwd = res.cwd.as_deref().map(std::path::Path::new);
    let resume = crate::cmd::port::resume_lines(Harness::Claude, &res.new_id, cwd);
    // Claude Code finds a session by its project dir; a rewind written anywhere else must be moved
    // there first. (`--out` defaults to it, so this only fires for an explicit elsewhere.)
    let home = cv_core::rewind::project_dir(&r.path);
    let relocate = res.new_path.parent().is_some_and(|d| d != home);

    let mut v = serde_json::to_value(&res)?;
    v["harness"] = "claude".into();
    v["resume"] = serde_json::json!(resume);
    dry_run_nulls(&mut v, pinned, &["new_path", "provenance_path"], &["new_id", "resume"]);

    let mut h = String::new();
    let tag = if res.dry_run {
        " (dry run — nothing written)"
    } else {
        ""
    };
    let _ = writeln!(
        h,
        "✦ rewound {} → {}{tag}",
        short_id(&res.source_id),
        short_id(&res.new_id)
    );
    let of = match (&res.subagent, &res.source_session_id) {
        (true, Some(p)) => format!(" · sub-agent of {} → emitted as a top-level session", short_id(p)),
        _ => String::new(),
    };
    let _ = writeln!(
        h,
        "  source:  {} ({}, sha256 {}…){of}",
        res.source_path.display(),
        human_bytes(res.source_bytes),
        &res.source_sha256[..12],
    );
    let why = match &res.evidence {
        Some(e) => format!(" · {} — `{}`", e.kind.label(), cv_core::ir::truncate(&e.command, 80)),
        None => String::new(),
    };
    let _ = writeln!(h, "  cut:     msg {} · line {}{why}", res.cut_msg_idx, res.cut_line);
    let from = match (res.start_mode, res.boundary_msg_idx, res.preserved_head) {
        ("full", ..) => "the first record (--full)".to_string(),
        (_, None, _) => "the first record (no compaction before the cut)".to_string(),
        (_, Some(b), true) => format!("the compaction at msg {b}, from its preserved segment"),
        (_, Some(b), false) => format!("the compaction at msg {b}"),
    };
    let _ = writeln!(
        h,
        "  start:   msg {} · line {} · {from}",
        res.start_msg_idx, res.start_line
    );
    let closed = if res.closed_tool_calls > 0 {
        format!(
            " (+{} tool result(s) past the cut, closing its open calls)",
            res.closed_tool_calls
        )
    } else {
        String::new()
    };
    let _ = writeln!(
        h,
        "  copied:  lines {}..={} — {} line(s), {}{closed} · omitted the {} line(s) after it",
        res.start_line,
        res.end_line,
        res.lines_written,
        human_bytes(res.bytes_written),
        res.omitted_lines,
    );
    for w in &res.warnings {
        let _ = writeln!(h, "  ⚠ {w}");
    }
    if res.cut_cwd.is_some() && res.cut_cwd != res.cwd {
        let _ = writeln!(
            h,
            "  at the cut the agent was working in {}",
            res.cut_cwd.as_deref().unwrap_or_default()
        );
    }
    if !res.dry_run {
        let _ = writeln!(h, "  new session: {}", res.new_path.display());
        let _ = writeln!(h, "  provenance:  {}", res.provenance_path.display());
        if relocate {
            let _ = writeln!(h, "  ⚠ claude resumes from {} — copy it there first", home.display());
        }
        let _ = writeln!(h, "  resume with (add --fork-session to keep the rewind reusable):");
        for line in &resume {
            let _ = writeln!(h, "    {line}");
        }
    }
    report(json, &h, &v)
}

// ---------- splice ----------

/// A parsed splice spec: which session, and a `[start, end)` window (end None → through the last).
struct SpliceSpec {
    id: String,
    start: usize,
    end: Option<usize>,
}

/// Parse `<id>:A..B` | `<id>:A..` | `<id>:..B` | `<id>` into a [`SpliceSpec`] — the same `A..B`
/// window grammar as `show --range` (0-based, end-exclusive). `<id>` may carry a `harness:` prefix
/// (`codex:019e…:0..12`), so the range is split off the END.
fn parse_splice_spec(spec: &str) -> Result<SpliceSpec> {
    match spec.rsplit_once(':') {
        Some((id, range)) if range.contains("..") => {
            let (start, end) = parse_range(range).with_context(|| format!("bad spec {spec:?}"))?;
            Ok(SpliceSpec {
                id: id.to_string(),
                start,
                end,
            })
        }
        Some((_, range))
            if !range.is_empty() && range.contains('-') && range.chars().all(|c| c.is_ascii_digit() || c == '-') =>
        {
            bail!(
                "bad spec {spec:?}: the `<id>:A-B` grammar is gone — use `<id>:A..B` (0-based, end-exclusive), \
                 `<id>:A..`, or `<id>:..B`"
            )
        }
        _ => Ok(SpliceSpec {
            id: spec.to_string(),
            start: 0,
            end: None,
        }),
    }
}

/// Where a splice goes. `resumable` is `--session` / `--agent-of` (a Claude transcript the harness
/// resumes, shared with `distill`); otherwise `--harness`/`--out` emit to a harness and nothing at
/// all prints a summary (with `--export md|json`, the composed session).
#[allow(clippy::too_many_arguments)]
pub(crate) fn cmd_splice(
    specs: &[String],
    harness: Option<String>,
    out: Option<PathBuf>,
    export: Option<String>,
    cwd: Option<PathBuf>,
    generate: bool,
    gen_model: Option<String>,
    thinking: cv_core::emit::ThinkingMode,
    resumable: Option<crate::cmd::distill::Target>,
    json: bool,
) -> Result<()> {
    let parsed: Vec<SpliceSpec> = specs.iter().map(|s| parse_splice_spec(s)).collect::<Result<_>>()?;

    // Resolve + parse each spec's session FIRST, keeping the owners alive in a Vec so the spans can
    // borrow them. We pair each parsed session with the (start, end) it'll select. A spec may be a
    // transcript path (a sub-agent file straight from `subagents/`), as for `distill`.
    let mut owned: Vec<(Session, usize, Option<usize>)> = Vec::with_capacity(parsed.len());
    let mut first_path: Option<PathBuf> = None;
    for sp in &parsed {
        let (session, path) = crate::cmd::distill::load(&sp.id, None)?;
        first_path.get_or_insert(path);
        owned.push((session, sp.start, sp.end));
    }

    let spans: Vec<cv_core::loom::Span<'_>> = owned
        .iter()
        .map(|(s, start, end)| cv_core::loom::Span {
            source: s,
            start: *start,
            end: *end,
        })
        .collect();

    // Target harness: --harness, else the first spec's source harness.
    let to_h = match &harness {
        Some(s) => Harness::parse(s).with_context(|| format!("unknown target harness: {s}"))?,
        None => owned
            .first()
            .map(|(s, _, _)| s.harness)
            .context("splice needs at least one session")?,
    };

    let spliced = cv_core::loom::splice(&spans, None, to_h);
    if let Some(target) = resumable {
        // `--session` / `--agent-of`: a resumable Claude transcript, the way `distill` writes one.
        // Nothing is elided (every selected message is carried whole), so there is no sidecar.
        let label = format!("splice: {}", specs.join(" "));
        let src = first_path.context("splice needs at least one session")?;
        let e = crate::cmd::distill::emit_to(&spliced, &target, &[], &src, &label)?;
        let mut h = String::new();
        let _ = writeln!(
            h,
            "✦ spliced {} ({} messages from {} span(s))",
            short_id(&spliced.id),
            spliced.messages.len(),
            specs.len()
        );
        if let Some(p) = &e.path {
            let _ = writeln!(h, "  ↳ wrote {}", p.display());
        }
        if let Some(r) = &e.resume {
            let _ = writeln!(h, "  ↳ resume: {r}");
        }
        return report(json, &h, &serde_json::to_value(&e)?);
    }
    finish_composed(
        spliced,
        to_h,
        harness.is_some(),
        out,
        export,
        cwd,
        generate,
        gen_model,
        thinking,
    )
}

/// Shared tail for splice: emit to a harness when `--harness`/`--out` is in play, otherwise
/// print a summary and (with `--export md|json`) the composed session to stdout.
#[allow(clippy::too_many_arguments)]
fn finish_composed(
    mut session: Session,
    to_h: Harness,
    explicit_to: bool,
    out: Option<PathBuf>,
    export: Option<String>,
    cwd: Option<PathBuf>,
    generate: bool,
    gen_model: Option<String>,
    thinking: cv_core::emit::ThinkingMode,
) -> Result<()> {
    // The generative half of looming: grow the composed branch with an LLM-generated continuation.
    if generate {
        match crate::llm::available_provider() {
            Some(prov) => eprintln!("✦ looming a continuation via {prov}…"),
            None => bail!(
                "--generate needs an LLM provider: set OPENROUTER_API_KEY, ANTHROPIC_API_KEY, or \
                 LMSTUDIO_API_BASE=local (free local LM Studio)"
            ),
        }
        let text = crate::llm::generate(
            &session,
            &crate::llm::GenerateOptions {
                model: gen_model,
                ..Default::default()
            },
        )?;
        let mut m = Message::new(Role::Assistant);
        m.content.push(Block::Text { text: text.into() });
        session.messages.push(m);
        eprintln!(
            "  ↳ grew the branch by 1 generated turn ({} msg total)",
            session.messages.len()
        );
    }

    // Emit path: an explicit --harness or an --out directory means "materialize this for a harness".
    if explicit_to || out.is_some() {
        if let Some(dir) = &cwd {
            session.cwd = Some(dir.clone());
        }
        return emit_session(
            &session,
            to_h,
            out,
            EmitOptions {
                new_cwd: cwd,
                thinking,
                ..Default::default()
            },
        );
    }

    // Otherwise: print a summary, and with --export, dump the composed session to stdout.
    println!(
        "✦ composed {} ({}) · {} msg",
        short_id(&session.id),
        session.harness.as_str(),
        session.messages.len()
    );
    if let Some(prov) = session.cv_extra().and_then(|b| b.get("loom")) {
        if let Some(arr) = prov.as_array() {
            for p in arr {
                println!(
                    "  ↳ {} {}[{}..{}]",
                    p.get("harness").and_then(|v| v.as_str()).unwrap_or("?"),
                    p.get("source_id")
                        .and_then(|v| v.as_str())
                        .map(short_id)
                        .unwrap_or_default(),
                    p.get("start").and_then(|v| v.as_u64()).unwrap_or(0),
                    p.get("end").and_then(|v| v.as_u64()).unwrap_or(0),
                );
            }
        }
    }
    match export.as_deref() {
        None => {
            println!("\n(use --export md|json to print it, or --harness <harness> [--out <dir>] to emit it)");
        }
        Some("json") => println!("{}", serde_json::to_string_pretty(&session)?),
        Some("md") | Some("markdown") => print!("{}", cv_core::render::to_markdown(&session)),
        Some(other) => bail!("unknown export format {other:?} (use md or json)"),
    }
    Ok(())
}

// ---------- dataset ----------

#[allow(clippy::too_many_arguments)]
pub(crate) fn cmd_dataset(
    format: &str,
    harness: Option<String>,
    query: Option<String>,
    subagents: bool,
    limit: Option<usize>,
    min_messages: usize,
    redact: bool,
    redact_only: Option<String>,
    out: Option<PathBuf>,
) -> Result<()> {
    use std::io::Write;
    if !matches!(format, "chatml" | "sharegpt") {
        bail!("unknown format {format:?} (use chatml or sharegpt)");
    }
    let want = parse_harness(&harness)?;
    let query = match query {
        Some(q) => Some(cv_core::SessionQuery::parse(&q).map_err(|e| anyhow::anyhow!(e))?),
        None => None,
    };
    let fmt = match format {
        "chatml" => cv_core::dataset::Format::Chatml,
        _ => cv_core::dataset::Format::ShareGpt,
    };
    // `--redact-only <classes>` scopes redaction to specific secret classes (and implies it);
    // plain `--redact` scrubs every class.
    let redact_opts = match &redact_only {
        Some(spec) => Some(cv_core::redact::RedactOptions {
            classes: cv_core::redact::RedactClasses::parse_only(spec).map_err(|e| anyhow::anyhow!(e))?,
            ..Default::default()
        }),
        None => redact.then(cv_core::redact::RedactOptions::default),
    };
    // Pre-resolve any `text:` predicates against the full-text index (one search per needle).
    let text_sets = query
        .as_ref()
        .map(crate::cmd::query::text_sets)
        .unwrap_or_else(crate::cmd::query::TextSets::empty);
    let mut writer: Box<dyn Write> = match &out {
        Some(p) => Box::new(std::io::BufWriter::new(fs::File::create(p)?)),
        None => Box::new(std::io::BufWriter::new(std::io::stdout())),
    };

    // Stream the corpus one session at a time (parse -> emit -> drop) so memory stays bounded.
    let mut emitted = 0usize;
    let mut skipped = 0usize;
    let mut ctx = EmitCtx {
        writer: writer.as_mut(),
        query: query.as_ref(),
        text_sets: &text_sets,
        fmt,
        redact_opts: redact_opts.as_ref(),
        min_messages,
        limit,
        emitted: &mut emitted,
        skipped: &mut skipped,
    };

    'outer: for r in cv_core::discover_all() {
        if let Some(w) = want {
            if r.harness != w {
                continue;
            }
        }
        if ctx.emit_ref(&r)? == Flow::LimitHit {
            break;
        }
        // `--subagents`: a Claude session fans out into a forest of sub-agent transcripts (both
        // directly-spawned and `Workflow`-spawned) that live in sidecar files `discover_all` never
        // lists. For a distillation corpus those agent transcripts are first-class training data —
        // and a parent on one model often spawns sub-agents on another, so a `model:` query that
        // misses them misses most of the matching turns. Walk the forest and emit each match too.
        if subagents && r.harness == cv_core::ir::Harness::Claude {
            for info in cv_core::harness::claude::subagent_tree(&r.path) {
                if ctx.emit_ref(&info.session)? == Flow::LimitHit {
                    break 'outer;
                }
            }
        }
    }
    writer.flush()?;
    let dest = out.as_ref().map(|p| format!(" to {}", p.display())).unwrap_or_default();
    let forest = if subagents { " (incl. sub-agent forest)" } else { "" };
    eprintln!("✦ {format}: wrote {emitted} record(s){dest}{forest} ({skipped} skipped)");
    Ok(())
}

/// Whether the emit loop should keep going or stop because `--limit` was reached.
#[derive(PartialEq, Eq)]
enum Flow {
    Continue,
    LimitHit,
}

/// The shared state for emitting one session ref as a dataset record — so the top-level corpus walk
/// and the per-session sub-agent forest walk run identical parse → filter → write logic.
struct EmitCtx<'a> {
    writer: &'a mut dyn std::io::Write,
    query: Option<&'a cv_core::SessionQuery>,
    text_sets: &'a crate::cmd::query::TextSets,
    fmt: cv_core::dataset::Format,
    redact_opts: Option<&'a cv_core::redact::RedactOptions>,
    min_messages: usize,
    limit: Option<usize>,
    emitted: &'a mut usize,
    skipped: &'a mut usize,
}

impl EmitCtx<'_> {
    /// Parse one ref and stream it as a record if it passes the filters. Returns [`Flow::LimitHit`]
    /// once `--limit` records have been emitted.
    fn emit_ref(&mut self, r: &cv_core::ir::SessionRef) -> Result<Flow> {
        // Catalog-cheap prefilter (everything but `model:`) prunes before we pay to parse.
        if let Some(q) = self.query {
            if !q.prefilter(r) {
                return Ok(Flow::Continue);
            }
        }
        let Some(adapter) = cv_core::harness::for_harness(r.harness) else {
            return Ok(Flow::Continue);
        };
        // Lazy opts: large content stays a span (held as a 16-byte ref), and `write_record` streams
        // it straight to the writer in chunks — so even a 700 MB record never materializes. `extra`
        // isn't read by dataset rendering, so it's skipped too.
        let session = match cv_core::stream::collect_with(adapter.as_ref(), r, &cv_core::ParseOptions::lazy()) {
            Ok(s) => s,
            Err(e) => {
                eprintln!("cv dataset: skipping {} ({}): {e:#}", r.id, r.harness);
                return Ok(Flow::Continue);
            }
        };
        if session.messages.len() < self.min_messages {
            *self.skipped += 1;
            return Ok(Flow::Continue);
        }
        // Full evaluation now that we have the parsed session (resolves model/git/text/tool/… that
        // the catalog-cheap prefilter left as Unknown).
        if let Some(q) = self.query {
            if !crate::cmd::query::matches_full(q, r, &session, self.text_sets) {
                *self.skipped += 1;
                return Ok(Flow::Continue);
            }
        }
        // Stream the record directly to the writer (chunked, JSON-escaped per chunk — never holds the
        // whole record). `--redact` scrubs one materialized message at a time inside the writer.
        let wrote = cv_core::dataset::write_record(&session, &mut self.writer, self.fmt, self.redact_opts)?;
        if wrote {
            writeln!(self.writer)?;
            *self.emitted += 1;
            if self.limit.is_some_and(|l| *self.emitted >= l) {
                return Ok(Flow::LimitHit);
            }
        } else {
            *self.skipped += 1;
        }
        Ok(Flow::Continue)
    }
}