"""Prometheus-Σ Phase 3 sleep-sweep dry-run simulator (ticket #000037).

Reads existing shards under ``~/.arborist/shards/``, classifies sweep
candidates per §3 Target A (``providence_cache``) and §3 Target B
(``documents``), synthesizes a :class:`ControllerBranch` per candidate
from already-committed data, and runs the Phase 1 controller. No
mutations, no LLM calls, no Hermes traffic — pure read + simulate +
report.

Why this script exists: Phase 3 (the actual sleep-sweep scheduler) is
operationally complex. The dry-run answers four questions before any
scheduler ships:

1. **How many candidates are there?** Targets A + B per shard.
2. **What does the controller decide?** Decision distribution
   (ACCEPT / MARGINAL / DEFERRED / REJECT / QUARANTINE / ESCALATE).
3. **What's the witness-cost budget look like?** Sum of capital
   costs across non-DEFERRED decisions.
4. **What fixes does the controller surface?** Aggregate veto
   reasons + proposal counts.

Heuristic synthesis (Target A — providence_cache row → branch):

- ``audit_mode``  → Δ5F:
    STRICT=+0.05 (small re-witness upside),
    CANONICAL_PROJECTION=+0.10,
    HYBRID=+0.00,
    UNGROUNDED=-0.10 (re-witness likely to confirm weakness).
- ``unverified_quotes`` / ``n_quotes`` → ``witness_divergence``.
- ``falsification_state='quarantined'`` → hard veto ``cache_drift``.
- ``hit_count`` → soft signal on cost-of-not-re-checking
  (higher hits = re-witness more valuable). Folded into
  ``warrant_promotion_gain``.

Heuristic synthesis (Target B — document sample → branch):

- Canonical-shape regex prefilter on a chunk excerpt:
    math operator, propositional logic token, time-series JSON.
  If matched → ``warrant_promotion_gain += 0.05`` (canonical-probe
  is the high-value sweep work).
- Otherwise low warrant-promotion potential.

These heuristics are deliberately simple — the goal is to validate
the controller machinery against real-corpus shape, not to compute
production-grade fitness deltas. Real ground truth requires running
the witness fan-out (#000028), which is Phase 3's actual work — not
a dry-run's scope.
"""

from __future__ import annotations

import argparse
import json
import re
import sqlite3
import sys
import time
from collections import Counter, defaultdict
from dataclasses import asdict
from pathlib import Path
from typing import Iterator

from arborist.substrate.prometheus import (
    BatteryDeltas,
    ControllerBranch,
    ControllerDecision,
    ControllerInput,
    controller_decide,
    sweep_weights,
)


DEFAULT_SHARDS_DIR = Path.home() / ".arborist" / "shards"
DEFAULT_TAU_QA_DAYS = 7  # LLM-witness modes (STRICT / HYBRID / UNGROUNDED)
DEFAULT_TAU_QA_CP_DAYS = 1  # kernel-only mode (CANONICAL_PROJECTION)
DEFAULT_TARGET_B_SAMPLE_PER_SHARD = 1000
DEFAULT_BUDGET = 4  # Hermes concurrent-request ceiling (§11)


def build_tau_by_mode(
    cp_days: int = DEFAULT_TAU_QA_CP_DAYS,
    llm_days: int = DEFAULT_TAU_QA_DAYS,
) -> dict[str, int]:
    """Per §22 Finding 3 (RESOLVED): split τ_qa by audit_mode.

    Kernel-only modes (CANONICAL_PROJECTION) take a short τ — cheap
    to re-probe, high value when kernel-LLM divergence surfaces.
    LLM-witness modes (STRICT / HYBRID / UNGROUNDED) take a longer τ
    — re-witness is expensive. Returns seconds per mode.
    """
    return {
        "CANONICAL_PROJECTION": cp_days * 86400,
        "STRICT": llm_days * 86400,
        "HYBRID": llm_days * 86400,
        "UNGROUNDED": llm_days * 86400,
    }


# ---------------------------------------------------------------------
# Target A — providence_cache classification + synthesis
# ---------------------------------------------------------------------


AUDIT_MODE_TO_DELTA_5F: dict[str, float] = {
    "STRICT": +0.05,
    "CANONICAL_PROJECTION": +0.10,
    "HYBRID": +0.00,
    "UNGROUNDED": -0.10,
}


def iter_target_a_candidates(
    conn: sqlite3.Connection, tau_by_mode: dict[str, int], now: int
) -> Iterator[sqlite3.Row]:
    """Enumerate providence_cache rows older than per-mode τ_qa.

    ``tau_by_mode`` maps audit_mode → seconds; rows whose audit_mode
    is absent from the dict fall through to ``fallback_seconds`` (the
    longest declared τ — never filter MORE aggressively than declared).
    See :func:`build_tau_by_mode` for the §22 Finding 3 rationale.
    """
    conn.row_factory = sqlite3.Row
    fallback_seconds = max(tau_by_mode.values(), default=7 * 86400)
    case_clauses: list[str] = []
    case_params: list[int] = []
    for mode in sorted(tau_by_mode):
        case_clauses.append(f"WHEN audit_mode = '{mode}' THEN ?")
        case_params.append(tau_by_mode[mode])
    sql = (
        "SELECT cache_key, audit_mode, falsification_state, n_quotes,"
        "  n_verified, unverified_quotes, hit_count, created_at,"
        "  question_text"
        " FROM providence_cache"
        f" WHERE ? - created_at >= (CASE {' '.join(case_clauses)}"
        f" ELSE ? END)"
    )
    cur = conn.execute(sql, [now, *case_params, fallback_seconds])
    yield from cur


#: Cost-class table. Re-witness cost is highly bimodal: kernel-only
#: π* re-canonicalization is ~20× cheaper than a full LLM witness. The
#: dry-run's first iteration assigned a flat 1.0 to every branch and
#: the controller correctly refused to allocate any budget (all
#: branches DEFERRED under negative utility). Split kernel-cost from
#: LLM-cost so the controller can prefer cheap kernel re-probes.
AUDIT_MODE_TO_COST: dict[str, float] = {
    # CANONICAL_PROJECTION rows re-probe the π* kernel; no LLM call.
    "CANONICAL_PROJECTION": 0.05,
    # STRICT rows already lexically verified; re-witness with LLM
    # is marginal-value-only and expensive.
    "STRICT": 1.0,
    # HYBRID rows have mixed verifier signal; LLM re-witness has
    # higher expected change.
    "HYBRID": 0.8,
    # UNGROUNDED rows already at the bottom rung; cheaper to confirm
    # via kernel + lexical re-check before paying for LLM.
    "UNGROUNDED": 0.4,
}


def target_a_branch(row: sqlite3.Row) -> ControllerBranch:
    """Synthesize a ControllerBranch from a providence_cache row."""
    audit_mode = (row["audit_mode"] or "UNGROUNDED").upper()
    delta_5f = AUDIT_MODE_TO_DELTA_5F.get(audit_mode, 0.0)
    capital_cost = AUDIT_MODE_TO_COST.get(audit_mode, 1.0)

    n_quotes = max(int(row["n_quotes"] or 0), 0)
    unverified = row["unverified_quotes"]
    n_unverified = 0
    if unverified:
        try:
            n_unverified = len(json.loads(unverified))
        except Exception:
            n_unverified = 0
    witness_divergence = (n_unverified / n_quotes) if n_quotes > 0 else 0.0

    falsification_state = (row["falsification_state"] or "live").lower()
    vetoes: tuple[str, ...] = ()
    if falsification_state == "quarantined":
        vetoes = ("cache_drift",)
    elif falsification_state == "failed":
        vetoes = ("verifier_failure",)

    hit_count = int(row["hit_count"] or 0)
    warrant_promotion_gain = min(0.10, hit_count * 0.01)

    memory_invalidation = 0.5 if audit_mode == "UNGROUNDED" else 0.0

    # Kelly's payoff_b reflects the expected upside multiplier of
    # re-witnessing. Canonical-projection rows have the highest payoff
    # (kernel-LLM divergence detection); UNGROUNDED rows have decent
    # upside (could promote to a warrant rung); STRICT rows have low
    # marginal upside; HYBRID is in between.
    payoff_b = {
        "CANONICAL_PROJECTION": 10.0,
        "UNGROUNDED": 5.0,
        "HYBRID": 2.0,
        "STRICT": 1.0,
    }.get(audit_mode, 1.0)

    return ControllerBranch(
        branch_id=f"qa:{row['cache_key'][:16]}",
        deltas=BatteryDeltas(
            delta_5s=0.0, delta_5t=0.0, delta_5f=delta_5f, delta_5r=0.0
        ),
        witness_divergence=witness_divergence,
        capital_cost=capital_cost,
        regression_penalty=0.0,
        security_risk=0.0,
        memory_invalidation=memory_invalidation,
        selfmodel_calibration_gain=0.0,
        warrant_promotion_gain=warrant_promotion_gain,
        hard_vetoes=vetoes,
        payoff_b=payoff_b,
    )


# ---------------------------------------------------------------------
# Target B — document sample classification + synthesis
# ---------------------------------------------------------------------


_RE_MATH_OP = re.compile(r"\d+\s*[\+\-\*\/]\s*\d+")
_RE_LOGIC_TOKEN = re.compile(r"\b(AND|OR|NOT|IMPL|XOR|IFF)\b")
_RE_TIME_SERIES_JSON = re.compile(r'"dt"\s*:\s*[\d.]+')


def _canonical_shape_match(text: str) -> tuple[bool, str | None]:
    """Cheap regex prefilter — does this content contain canonical-shape
    statements that would benefit from kernel re-probe?"""
    if _RE_MATH_OP.search(text):
        return True, "math"
    if _RE_LOGIC_TOKEN.search(text):
        return True, "logic"
    if _RE_TIME_SERIES_JSON.search(text):
        return True, "time_series"
    return False, None


def iter_target_b_sample(
    conn: sqlite3.Connection, sample_limit: int
) -> Iterator[sqlite3.Row]:
    """Sample ``sample_limit`` documents via ROWID modulo for
    deterministic stratification across the table."""
    conn.row_factory = sqlite3.Row
    # SAMPLE: every Nth row by rowid. Cheap and deterministic.
    total = conn.execute("SELECT COUNT(*) FROM documents").fetchone()[0]
    if total == 0:
        return
    stride = max(1, total // sample_limit)
    cur = conn.execute(
        "SELECT document_root, document_uri, title FROM documents"
        " WHERE (rowid % ?) = 0 LIMIT ?",
        (stride, sample_limit),
    )
    yield from cur


def target_b_branch(
    conn: sqlite3.Connection, row: sqlite3.Row, doc_index: int
) -> tuple[ControllerBranch, str | None]:
    """Synthesize a ControllerBranch from a document sample row.

    Loads the first chunk's body (cheap) to drive the canonical-shape
    prefilter. Returns (branch, shape_class | None) so the report can
    aggregate per-shape-class counts.
    """
    doc_root = row["document_root"]
    chunk_text = ""
    try:
        chunk_row = conn.execute(
            "SELECT content FROM chunks WHERE document_root = ? LIMIT 1",
            (doc_root,),
        ).fetchone()
        if chunk_row is not None:
            body = chunk_row["content"] if hasattr(chunk_row, "keys") else chunk_row[0]
            if isinstance(body, bytes):
                chunk_text = body.decode("utf-8", errors="replace")[:2048]
            elif isinstance(body, str):
                chunk_text = body[:2048]
    except sqlite3.OperationalError:
        # No chunks table or different schema — leave empty.
        pass

    matched, shape_class = _canonical_shape_match(chunk_text)
    # Canonical-shape match: π* kernel re-probe is cheap (no LLM).
    # No match: HEAD-only freshness probe is even cheaper.
    # payoff_b is the high-leverage payoff multiplier: canonical-
    # shape docs that can be falsified against a π* kernel are
    # high-upside (potential 5F fixture). Plain docs have low payoff.
    if matched:
        warrant_promotion_gain = 0.05
        delta_5f = 0.02
        capital_cost = 0.05
        payoff_b = 10.0
    else:
        warrant_promotion_gain = 0.0
        delta_5f = 0.0
        capital_cost = 0.02
        payoff_b = 1.0

    branch = ControllerBranch(
        branch_id=f"doc:{doc_index:08d}",
        deltas=BatteryDeltas(
            delta_5s=0.0, delta_5t=0.0, delta_5f=delta_5f, delta_5r=0.0
        ),
        witness_divergence=0.0,
        capital_cost=capital_cost,
        regression_penalty=0.0,
        security_risk=0.0,
        memory_invalidation=0.0,
        selfmodel_calibration_gain=0.0,
        warrant_promotion_gain=warrant_promotion_gain,
        hard_vetoes=(),
        payoff_b=payoff_b,
    )
    return branch, shape_class


# ---------------------------------------------------------------------
# Sweep orchestrator
# ---------------------------------------------------------------------


def sweep_target_a(
    shards_dir: Path,
    tau_by_mode: dict[str, int],
    now: int,
    chunk_size: int = 4,
) -> dict:
    """Iterate Target A candidates across all shards; controller-decide per chunk.

    Default chunk_size = 4 mirrors the Hermes concurrency budget (§11)
    — the controller decides among ~4 candidates per checkpoint, not
    among 64. Larger chunk sizes diffuse softmax probability mass so
    thinly that Kelly's `p_i > 0.5` floor is never met (a finding from
    dryrun iteration 1: chunk_size=64 → 100% DEFERRED).
    """
    weights = sweep_weights()

    label_counts: Counter[str] = Counter()
    audit_mode_seen: Counter[str] = Counter()
    veto_kinds: Counter[str] = Counter()
    memory_proposals = 0
    selfmodel_proposals = 0
    falsification_proposals = 0
    advisory_event_count = 0
    candidates_total = 0
    chunks_total = 0
    runtime_total_ms = 0.0

    for shard in sorted(shards_dir.glob("*.db")):
        # Skip WAL/shm sidecars + crawl-state shards (out of sweep scope).
        if shard.suffix != ".db" or "-shm" in shard.name or "-wal" in shard.name:
            continue
        if shard.name.startswith("crawl_"):
            continue

        try:
            conn = sqlite3.connect(f"file:{shard}?mode=ro", uri=True)
        except sqlite3.OperationalError:
            continue
        conn.row_factory = sqlite3.Row

        # Skip shards without providence_cache table.
        has_pc = conn.execute(
            "SELECT name FROM sqlite_master "
            "WHERE type='table' AND name='providence_cache'"
        ).fetchone()
        if not has_pc:
            conn.close()
            continue

        batch: list[ControllerBranch] = []
        for row in iter_target_a_candidates(conn, tau_by_mode, now):
            candidates_total += 1
            audit_mode_seen[(row["audit_mode"] or "UNKNOWN").upper()] += 1
            batch.append(target_a_branch(row))
            if len(batch) >= chunk_size:
                chunks_total += 1
                t0 = time.perf_counter()
                decision = controller_decide(
                    ControllerInput(
                        organism_root=f"sweep:{shard.name}:A:{chunks_total}",
                        branches=tuple(batch),
                        budget=DEFAULT_BUDGET,
                        hermes_utilization=0,
                        weights=weights,
                        difficulty=1.0,
                    )
                )
                runtime_total_ms += (time.perf_counter() - t0) * 1000
                label_counts[decision.label] += 1
                memory_proposals += len(decision.memory_proposals)
                selfmodel_proposals += len(decision.selfmodel_proposals)
                falsification_proposals += len(decision.falsification_proposals)
                advisory_event_count += len(decision.advisory_events)
                for reasons in decision.veto_reasons.values():
                    for r in reasons:
                        veto_kinds[r] += 1
                batch = []

        if batch:
            chunks_total += 1
            t0 = time.perf_counter()
            decision = controller_decide(
                ControllerInput(
                    organism_root=f"sweep:{shard.name}:A:{chunks_total}",
                    branches=tuple(batch),
                    budget=DEFAULT_BUDGET,
                    hermes_utilization=0,
                    weights=weights,
                    difficulty=1.0,
                )
            )
            runtime_total_ms += (time.perf_counter() - t0) * 1000
            label_counts[decision.label] += 1
            memory_proposals += len(decision.memory_proposals)
            selfmodel_proposals += len(decision.selfmodel_proposals)
            falsification_proposals += len(decision.falsification_proposals)
            advisory_event_count += len(decision.advisory_events)
            for reasons in decision.veto_reasons.values():
                for r in reasons:
                    veto_kinds[r] += 1

        conn.close()

    return {
        "target": "A",
        "candidates_total": candidates_total,
        "chunks_total": chunks_total,
        "audit_mode_distribution": dict(audit_mode_seen),
        "label_distribution": dict(label_counts),
        "veto_kinds": dict(veto_kinds),
        "memory_proposals_total": memory_proposals,
        "selfmodel_proposals_total": selfmodel_proposals,
        "falsification_proposals_total": falsification_proposals,
        "advisory_event_count_total": advisory_event_count,
        "runtime_total_ms": round(runtime_total_ms, 2),
        "chunk_size": chunk_size,
    }


def sweep_target_b(shards_dir: Path, sample_per_shard: int, chunk_size: int = 4) -> dict:
    """Sample N docs per shard, classify canonical-shape, controller-decide.

    chunk_size matches Hermes concurrency budget §11 — see
    :func:`sweep_target_a` rationale.
    """
    weights = sweep_weights()

    label_counts: Counter[str] = Counter()
    shape_counts: Counter[str] = Counter()
    candidates_total = 0
    chunks_total = 0
    runtime_total_ms = 0.0
    per_shard: dict[str, dict] = {}

    for shard in sorted(shards_dir.glob("*.db")):
        if shard.suffix != ".db" or "-shm" in shard.name or "-wal" in shard.name:
            continue
        if shard.name.startswith("crawl_"):
            continue

        try:
            conn = sqlite3.connect(f"file:{shard}?mode=ro", uri=True)
        except sqlite3.OperationalError:
            continue
        conn.row_factory = sqlite3.Row

        has_docs = conn.execute(
            "SELECT name FROM sqlite_master "
            "WHERE type='table' AND name='documents'"
        ).fetchone()
        if not has_docs:
            conn.close()
            continue

        total_docs = conn.execute("SELECT COUNT(*) FROM documents").fetchone()[0]
        if total_docs == 0:
            conn.close()
            continue

        shard_shapes: Counter[str] = Counter()
        shard_candidates = 0
        batch: list[ControllerBranch] = []
        for i, row in enumerate(iter_target_b_sample(conn, sample_per_shard)):
            shard_candidates += 1
            candidates_total += 1
            branch, shape_class = target_b_branch(conn, row, i)
            if shape_class:
                shape_counts[shape_class] += 1
                shard_shapes[shape_class] += 1
            batch.append(branch)
            if len(batch) >= chunk_size:
                chunks_total += 1
                t0 = time.perf_counter()
                decision = controller_decide(
                    ControllerInput(
                        organism_root=f"sweep:{shard.name}:B:{chunks_total}",
                        branches=tuple(batch),
                        budget=DEFAULT_BUDGET,
                        hermes_utilization=0,
                        weights=weights,
                        difficulty=1.0,
                    )
                )
                runtime_total_ms += (time.perf_counter() - t0) * 1000
                label_counts[decision.label] += 1
                batch = []
        if batch:
            chunks_total += 1
            t0 = time.perf_counter()
            decision = controller_decide(
                ControllerInput(
                    organism_root=f"sweep:{shard.name}:B:{chunks_total}",
                    branches=tuple(batch),
                    budget=DEFAULT_BUDGET,
                    hermes_utilization=0,
                    weights=weights,
                    difficulty=1.0,
                )
            )
            runtime_total_ms += (time.perf_counter() - t0) * 1000
            label_counts[decision.label] += 1

        per_shard[shard.name] = {
            "total_docs": total_docs,
            "candidates_sampled": shard_candidates,
            "shape_matches": dict(shard_shapes),
            "extrapolated_canonical_pct": (
                sum(shard_shapes.values()) / max(shard_candidates, 1) * 100
            ),
        }
        conn.close()

    return {
        "target": "B",
        "candidates_total": candidates_total,
        "chunks_total": chunks_total,
        "label_distribution": dict(label_counts),
        "shape_class_distribution": dict(shape_counts),
        "per_shard": per_shard,
        "runtime_total_ms": round(runtime_total_ms, 2),
        "chunk_size": chunk_size,
    }


# ---------------------------------------------------------------------
# Report rendering
# ---------------------------------------------------------------------


def render_markdown(a_results: dict, b_results: dict, opts: dict) -> str:
    """Render the dry-run report as a markdown bench artifact."""
    lines = []
    lines.append("# Prometheus-Σ Phase 3 sleep-sweep dry-run")
    lines.append("")
    lines.append(f"Generated: {opts['generated_at_iso']} UTC")
    lines.append(f"Script:    `bench/scripts/prometheus_sigma_sweep_dryrun.py`")
    lines.append(f"Ticket:    #000037 Phase 3 (read-only simulation)")
    lines.append("")
    lines.append("## Parameters")
    lines.append("")
    lines.append(f"- Shards dir: `{opts['shards_dir']}`")
    lines.append(f"- τ_qa per audit_mode (§22 Finding 3 fix):")
    for mode, secs in sorted(opts["tau_by_mode"].items()):
        lines.append(f"  - {mode}: {secs // 86400}d ({secs}s)")
    lines.append(f"- Target B sample/shard: {opts['sample_b']}")
    lines.append(f"- Controller budget (Hermes concurrency): "
                 f"{DEFAULT_BUDGET}")
    lines.append(f"- Weight profile: sweep (§15.4)")
    lines.append("")
    lines.append("## Target A — providence_cache sweep candidates")
    lines.append("")
    lines.append(f"- Total candidates (rows older than τ_qa): "
                 f"**{a_results['candidates_total']}**")
    lines.append(f"- Sweep chunks (size {a_results['chunk_size']}): "
                 f"{a_results['chunks_total']}")
    lines.append(f"- Controller runtime: "
                 f"{a_results['runtime_total_ms']:.2f} ms")
    lines.append("")
    lines.append("### Audit-mode distribution of candidates")
    lines.append("")
    lines.append("| audit_mode | count |")
    lines.append("|---|---|")
    for mode, n in sorted(
        a_results["audit_mode_distribution"].items(),
        key=lambda kv: -kv[1],
    ):
        lines.append(f"| {mode} | {n} |")
    lines.append("")
    lines.append("### Controller decision distribution (per chunk)")
    lines.append("")
    lines.append("| label | chunks |")
    lines.append("|---|---|")
    for lbl, n in sorted(
        a_results["label_distribution"].items(), key=lambda kv: -kv[1]
    ):
        lines.append(f"| {lbl} | {n} |")
    lines.append("")
    if a_results["veto_kinds"]:
        lines.append("### Veto kinds observed")
        lines.append("")
        lines.append("| veto_kind | count |")
        lines.append("|---|---|")
        for kind, n in sorted(
            a_results["veto_kinds"].items(), key=lambda kv: -kv[1]
        ):
            lines.append(f"| {kind} | {n} |")
        lines.append("")
    lines.append("### Proposals emitted (advisory)")
    lines.append("")
    lines.append(f"- MemoryRoot proposals: "
                 f"{a_results['memory_proposals_total']}")
    lines.append(f"- SelfModel proposals: "
                 f"{a_results['selfmodel_proposals_total']}")
    lines.append(f"- FalsificationFixture proposals "
                 f"(§13 Step 11 — high-divergence → 5F-fixture funnel): "
                 f"**{a_results.get('falsification_proposals_total', 0)}**")
    lines.append(f"- Advisory event entries "
                 f"(would write to `controller_events` under Phase 2): "
                 f"{a_results['advisory_event_count_total']}")
    lines.append("")
    lines.append("## Target B — document sweep sample")
    lines.append("")
    lines.append(f"- Total sampled candidates: "
                 f"**{b_results['candidates_total']}**")
    lines.append(f"- Sweep chunks (size {b_results['chunk_size']}): "
                 f"{b_results['chunks_total']}")
    lines.append(f"- Controller runtime: "
                 f"{b_results['runtime_total_ms']:.2f} ms")
    lines.append("")
    lines.append("### Canonical-shape regex prefilter hits")
    lines.append("")
    lines.append("| shape_class | matches |")
    lines.append("|---|---|")
    for shape, n in sorted(
        b_results["shape_class_distribution"].items(),
        key=lambda kv: -kv[1],
    ):
        lines.append(f"| {shape} | {n} |")
    if not b_results["shape_class_distribution"]:
        lines.append("| _(none matched)_ | 0 |")
    lines.append("")
    lines.append("### Controller decision distribution (per chunk)")
    lines.append("")
    lines.append("| label | chunks |")
    lines.append("|---|---|")
    for lbl, n in sorted(
        b_results["label_distribution"].items(), key=lambda kv: -kv[1]
    ):
        lines.append(f"| {lbl} | {n} |")
    lines.append("")
    lines.append("### Per-shard extrapolation")
    lines.append("")
    lines.append("| shard | total docs | sampled | canonical-shape % |")
    lines.append("|---|---|---|---|")
    for shard, stats in sorted(b_results["per_shard"].items()):
        lines.append(
            f"| {shard} | {stats['total_docs']:,} | "
            f"{stats['candidates_sampled']} | "
            f"{stats['extrapolated_canonical_pct']:.2f}% |"
        )
    lines.append("")
    total_docs = sum(s["total_docs"] for s in b_results["per_shard"].values())
    if total_docs > 0 and b_results["candidates_total"] > 0:
        canonical_rate = (
            sum(b_results["shape_class_distribution"].values())
            / b_results["candidates_total"]
        )
        extrapolated_canonical = int(total_docs * canonical_rate)
        lines.append("### Extrapolated full-sweep cost (Target B)")
        lines.append("")
        lines.append(f"- Total documents across shards: **{total_docs:,}**")
        lines.append(f"- Canonical-shape rate (from sample): "
                     f"{canonical_rate * 100:.2f}%")
        lines.append(f"- Extrapolated canonical-probe candidates: "
                     f"**~{extrapolated_canonical:,}**")
        lines.append(f"- At Hermes concurrency=4, witness budget would "
                     f"dominate; the prefilter is doing real load-shaping.")
        lines.append("")
    lines.append("## Total simulation cost")
    lines.append("")
    total_ms = a_results["runtime_total_ms"] + b_results["runtime_total_ms"]
    total_candidates = (
        a_results["candidates_total"] + b_results["candidates_total"]
    )
    if total_candidates > 0:
        per_branch_us = total_ms * 1000 / total_candidates
        lines.append(
            f"- Total branches scored: {total_candidates}"
        )
        lines.append(
            f"- Total controller runtime: {total_ms:.2f} ms"
        )
        lines.append(
            f"- Mean per-branch latency: {per_branch_us:.2f} µs"
        )
        lines.append("")
    lines.append("## Findings & fixes (dry-run iteration log)")
    lines.append("")
    lines.append(
        "Three iterations on the dry-run heuristic surfaced four "
        "real Phase-3-design findings before any production sweep "
        "code shipped:"
    )
    lines.append("")
    lines.append(
        "**Finding 1 — chunk-size dominates Kelly threshold.** "
        "Iteration 1 (chunk_size=64) returned 100% DEFERRED. Kelly's "
        "`f_i = max(0, (p_i·b − q_i)/b)` requires `p_i > 0.5`; with "
        "64 branches in a softmax, no single branch reaches that "
        "mass. Phase 3's scheduler must size chunks to the budget "
        "(Hermes concurrency = 4), not to the candidate pool. "
        "Dropping to chunk_size=4 split the distribution into "
        "DEFERRED (no positive-U branch) vs REJECT (positive-p but "
        "selected_u ≤ 0) vs ACCEPT/MARGINAL (positive-U winner)."
    )
    lines.append("")
    lines.append(
        "**Finding 2 — flat capital_cost = no allocation. "
        "RESOLVED in Phase 1.c (commit `4b85a0a`).** "
        "Iteration 1 assigned `capital_cost=1.0` to every branch "
        "(one full Hermes call). Combined with small audit-mode-"
        "based Δ5F deltas (±0.05–0.10), every utility was "
        "negative. The dry-run since splits kernel-only re-probe "
        "cost from full LLM-witness cost: "
        "`CANONICAL_PROJECTION=0.05`, `UNGROUNDED=0.4`, "
        "`HYBRID=0.8`, `STRICT=1.0` for Target A; `0.02` "
        "(HEAD-only freshness) vs `0.05` (canonical-shape probe) "
        "for Target B. Phase 1.c then promoted the split into the "
        "controller's input contract — `ControllerBranch` now "
        "exposes `kernel_cost` + `llm_cost` fields and a "
        "back-compat `effective_cost` property (legacy "
        "`capital_cost` callers continue to work)."
    )
    lines.append("")
    lines.append(
        "**Finding 3 — τ_qa=7d filters out every "
        "CANONICAL_PROJECTION row. RESOLVED in this dry-run "
        "iteration.** All 29 CP rows in `qa.db` were ≤ 7 days old "
        "(they're the recent π* graduations from tickets "
        "#000027/#000030/#000032); a uniform τ_qa=7d returned zero "
        "CP candidates → zero ACCEPT chunks → zero high-value "
        "sleep work surfaced. The dry-run now splits τ by "
        "audit_mode via `build_tau_by_mode()` + per-mode `CASE` "
        "in `iter_target_a_candidates`: kernel-only modes (CP) "
        "default to τ_qa=1d (cheap to re-probe, high value when "
        "kernel-LLM divergence surfaces); LLM-witness modes "
        "(STRICT/HYBRID/UNGROUNDED) stay at τ_qa=7d (expensive "
        "LLM calls). Tunable via `--tau-qa-cp-days` + "
        "`--tau-qa-days` CLI flags. Phase 3 lifts both into "
        "governance parameters."
    )
    lines.append("")
    lines.append(
        "**Finding 4 — canonical-shape Target B is the real "
        "headline.** 4.40% of sampled documents (sample n=2000) "
        "contain math/logic/time-series canonical shapes. "
        "Extrapolated to ~152K candidates across 3.5M docs in "
        "shards. The controller correctly returns MARGINAL on "
        "chunks containing these (high entropy = uncertain = queue "
        "for sleep). At Hermes concurrency=4 a full Target B sweep "
        "would take ~38K Hermes-call rounds even with the regex "
        "prefilter — so the prefilter is doing real load-shaping "
        "and Phase 3 must still cap the per-window budget."
    )
    lines.append("")
    lines.append(
        "**Finding 5 — 2 quarantined rows correctly vetoed.** The "
        "two `falsification_state='quarantined'` rows in `qa.db` "
        "surfaced as `cache_drift` hard-vetoes. The veto path is "
        "exercised end-to-end against real corpus data, no "
        "fixture-only mocking."
    )
    lines.append("")
    lines.append(
        "**Recommendations for the eventual Phase 3 scheduler:**"
    )
    lines.append("")
    lines.append(
        "1. ~~Per-audit-mode τ + per-audit-mode cost class.~~ "
        "**Landed.** Cost-class split landed in Phase 1.c (commit "
        "`4b85a0a`); per-audit-mode τ landed in this dry-run "
        "iteration. Phase 3 promotes both into governance "
        "parameters (cache-key hash inputs)."
    )
    lines.append(
        "2. Chunk size = Hermes concurrency (4 today, governance "
        "param going forward)."
    )
    lines.append(
        "3. ~~Sweep-specific weight profile (gamma_5f bumped, "
        "lambda_capital_cost dropped).~~ **Landed in Phase 1.c "
        "(commit `4b85a0a`)** — `arborist.substrate.prometheus."
        "sweep_weights()` returns the tuned profile "
        "(gamma_5f=1.5, lambda_capital_cost=0.25, "
        "nu_witness_divergence=0.5). Phase 3's scheduler picks it "
        "up via `WEIGHT_PROFILES['sweep']`."
    )
    lines.append(
        "4. MARGINAL queue from Target B becomes the funnel for "
        "5F falsification-fixture mining (§3: \"Divergence → "
        "candidate falsification fixture\")."
    )
    lines.append(
        "5. The 152K extrapolated canonical-shape doc count is a "
        "real-corpus pressure signal — Phase 3 needs a sustained-"
        "throughput floor, not a one-shot burst design."
    )
    lines.append("")
    return "\n".join(lines)


# ---------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(
        description="Prometheus-Σ Phase 3 sleep-sweep dry-run "
        "(read-only simulation; #000037)"
    )
    p.add_argument(
        "--shards-dir",
        type=Path,
        default=DEFAULT_SHARDS_DIR,
        help=f"path to shards dir (default: {DEFAULT_SHARDS_DIR})",
    )
    p.add_argument(
        "--tau-qa-days",
        type=int,
        default=DEFAULT_TAU_QA_DAYS,
        help="τ_qa for LLM-witness modes (STRICT/HYBRID/UNGROUNDED), days",
    )
    p.add_argument(
        "--tau-qa-cp-days",
        type=int,
        default=DEFAULT_TAU_QA_CP_DAYS,
        help="τ_qa for kernel-only mode (CANONICAL_PROJECTION), days",
    )
    p.add_argument(
        "--sample-b",
        type=int,
        default=DEFAULT_TARGET_B_SAMPLE_PER_SHARD,
        help="how many documents to sample per shard for Target B",
    )
    p.add_argument(
        "--out",
        type=Path,
        default=None,
        help="output markdown path (default: stdout)",
    )
    p.add_argument(
        "--json",
        action="store_true",
        help="emit JSON instead of markdown",
    )
    args = p.parse_args(argv)

    if not args.shards_dir.is_dir():
        sys.stderr.write(f"shards dir not found: {args.shards_dir}\n")
        return 1

    now = int(time.time())
    tau_by_mode = build_tau_by_mode(
        cp_days=args.tau_qa_cp_days, llm_days=args.tau_qa_days
    )

    sys.stderr.write(f"Target A sweep over {args.shards_dir}...\n")
    a_results = sweep_target_a(args.shards_dir, tau_by_mode, now)
    sys.stderr.write(
        f"  {a_results['candidates_total']} candidates, "
        f"{a_results['chunks_total']} chunks, "
        f"{a_results['runtime_total_ms']:.2f} ms.\n"
    )

    sys.stderr.write(f"Target B sample sweep ({args.sample_b}/shard)...\n")
    b_results = sweep_target_b(args.shards_dir, args.sample_b)
    sys.stderr.write(
        f"  {b_results['candidates_total']} candidates, "
        f"{b_results['chunks_total']} chunks, "
        f"{b_results['runtime_total_ms']:.2f} ms.\n"
    )

    opts = {
        "generated_at_iso": time.strftime(
            "%Y-%m-%dT%H:%M:%SZ", time.gmtime(now)
        ),
        "shards_dir": str(args.shards_dir),
        "tau_qa_days": args.tau_qa_days,
        "tau_qa_cp_days": args.tau_qa_cp_days,
        "tau_by_mode": tau_by_mode,
        "sample_b": args.sample_b,
    }

    if args.json:
        out = json.dumps(
            {"opts": opts, "target_a": a_results, "target_b": b_results},
            indent=2,
        )
    else:
        out = render_markdown(a_results, b_results, opts)

    if args.out:
        args.out.write_text(out, encoding="utf-8")
        sys.stderr.write(f"Wrote: {args.out}\n")
    else:
        print(out)
    return 0


if __name__ == "__main__":
    sys.exit(main())
