arborist/bench/batteries/b_5f.py

"""5F battery — Function / Finetuning / Falsification / Formulate /
Feedback Loop (ticket #000025).

Operational-quality axis from Dav1DPrometheus's framework. 5S
evaluates linguistic/sign-structure precision; 5T evaluates
temporal/cross-reasoning/logical coverage; 5F evaluates whether the
organism *acts, adapts, falsifies, formulates, and integrates
feedback* — the dimensions that turn a state-space into an adaptive
state-space.

All runners deterministic, no LLM-as-judge. Phase 1a fixtures use
text / claim_lattice / providence_record / selfmodel_snapshot /
memory_snapshot carriers; future carriers fail/skip explicitly.
"""

from __future__ import annotations

import time
from pathlib import Path

from bench.batteries.base import (
    BatteryResult,
    TaskResult,
    fixture_digest,
    fixture_meta,
    iter_tasks,
    validate_carrier,
)
from bench.batteries.b_5s import _content_tokens, _runtime_digest


def _carrier_check(task: dict) -> tuple[bool, str]:
    reason = validate_carrier(task)
    if reason is not None:
        return False, reason
    return True, ""


def _build_result(
    sub_battery: str,
    fixtures_path: Path,
    per_task: list[TaskResult],
    metrics: dict[str, float],
) -> BatteryResult:
    pass_count = sum(1 for t in per_task if t.passed)
    fail_count = len(per_task) - pass_count
    return BatteryResult(
        battery="5f",
        sub_battery=sub_battery,
        fixture_path=str(fixtures_path),
        fixture_digest=fixture_digest(fixtures_path),
        pass_count=pass_count,
        fail_count=fail_count,
        metrics=metrics,
        per_task=per_task,
        runtime_digest=_runtime_digest(),
    )


# ---------------------------------------------------------------------
# Function — does the system perform the designated operation?
# ---------------------------------------------------------------------


def _eval_shape_match(task: dict) -> tuple[bool, str]:
    """Check that the produced output shape matches the expected
    shape declaration.

    Phase 1a fixtures embed both the produced output and the
    expected_shape. Real production wiring (Phase 1b) will call
    arborist.qa.runner.ask and use the response shape.
    """
    produced = task.get("produced_output", {})
    shape = task["expected_shape"]
    claim_count_ok = len(produced.get("claims", [])) == shape.get("claim_count")
    pointers_ok = True
    if shape.get("pointers_required"):
        pointers_ok = all(
            isinstance(c.get("pointer_ids", []), list)
            and len(c.get("pointer_ids", [])) > 0
            for c in produced.get("claims", [])
        )
    return claim_count_ok and pointers_ok, "shape_match"


def _eval_pointer_set_match(task: dict) -> tuple[bool, str]:
    produced = set()
    for c in task.get("produced_output", {}).get("claims", []):
        for p in c.get("pointer_ids", []):
            produced.add(p)
    expected = set(task["expected_shape"]["pointer_set"])
    return produced == expected, f"pointer_set_match expected={sorted(expected)} got={sorted(produced)}"


def _eval_threshold_on_metric(task: dict) -> tuple[bool, str]:
    metric = task["expected_shape"]["metric"]
    threshold = task["expected_shape"]["threshold"]
    measured = task.get("produced_output", {}).get(metric)
    if measured is None:
        return False, f"metric {metric!r} not in produced_output"
    return measured >= threshold, f"{metric}={measured} threshold={threshold}"


_FUNCTION_EVALUATORS = {
    "shape_match": _eval_shape_match,
    "pointer_set_match": _eval_pointer_set_match,
    "threshold_on_metric": _eval_threshold_on_metric,
}


def _live_function_produced(task: dict) -> dict:
    """Derive ``produced_output`` by routing ``input_text`` through
    arborist.qa.parse_claims.parse_pointer_claims. Live mode
    primary surface: tests the actual claim-extraction pipeline
    against the same shape evaluators Phase 1a uses."""
    from arborist.qa.parse_claims import parse_pointer_claims

    parsed = parse_pointer_claims(task["input_text"])
    return {
        "claims": [
            {"claim_text": c.claim_text, "pointer_ids": list(c.pointer_ids)}
            for c in parsed
        ],
    }


def run_function(fixtures_path: Path) -> BatteryResult:
    """Function pass-rate over deterministic evaluator types.

    Two fixture modes:

    - **Embedded** (Phase 1a): ``produced_output`` in the fixture.
    - **Live** (Phase 1b.2): ``input_text`` in the fixture; runner
      derives ``produced_output`` via real
      :func:`arborist.qa.parse_claims.parse_pointer_claims` and
      then applies the same shape evaluator. Tests parser → shape
      pipeline against actual organism behavior.
    """
    per_task: list[TaskResult] = []
    for task in iter_tasks(fixtures_path):
        task_id = task["id"]
        ok, reason = _carrier_check(task)
        if not ok:
            per_task.append(
                TaskResult(task_id=task_id, passed=False, detail={"reason": reason})
            )
            continue
        evaluator = task.get("evaluator")
        fn = _FUNCTION_EVALUATORS.get(evaluator)
        if fn is None:
            per_task.append(
                TaskResult(task_id=task_id, passed=False,
                           detail={"reason": f"unknown_evaluator: {evaluator!r}"})
            )
            continue
        try:
            if "produced_output" in task:
                source = "embedded"
                effective_task = task
            elif "input_text" in task:
                source = "live"
                effective_task = {**task, "produced_output": _live_function_produced(task)}
            else:
                raise KeyError(
                    "task needs 'produced_output' (embedded) or "
                    "'input_text' (live)"
                )
            valid, why = fn(effective_task)
            expected = task.get("expected", "pass")
            observed = "pass" if valid else "fail"
            passed = observed == expected
        except Exception as exc:  # noqa: BLE001
            passed = False
            source = "error"
            why = f"{type(exc).__name__}: {exc}"
        per_task.append(TaskResult(
            task_id=task_id, passed=passed,
            detail={"source": source, "evaluator": evaluator, "reason": why},
        ))

    total = len(per_task)
    rate = sum(1 for t in per_task if t.passed) / total if total else 0.0
    meta = fixture_meta(fixtures_path)
    return _build_result(
        meta.get("sub_battery", "function"),
        fixtures_path,
        per_task,
        {"function_pass_rate": rate},
    )


# ---------------------------------------------------------------------
# Finetuning — does the organism improve parent→child?
# ---------------------------------------------------------------------


def _capital_cost_delta(task: dict) -> float:
    """Sum the resource_budget deltas as a single scalar capital cost.

    Phase 1a uses budget-as-cost: v1 fixtures don't carry actual
    measured costs, only budget caps. Phase 1b.2 will replace this
    with real values from the capital_ledger (#000020).

    Sums ``max_compute_ms_delta`` (treated 1 ms = 1e-3 cost units) +
    ``max_storage_delta_bytes`` (1 MB = 1 cost unit). Tunable later.
    """
    budget = task.get("resource_budget", {})
    compute_ms = float(budget.get("max_compute_ms_delta", 0))
    storage_bytes = float(budget.get("max_storage_delta_bytes", 0))
    return compute_ms * 1e-3 + storage_bytes / 1e6


# Sentinels for division-by-zero efficiency metrics. Per fox's
# 2026-05-08 review of #fbd99a8: explicit values, not Python NaN/inf
# accidents. We use Python ``float('inf')`` when cost=0 and gain>0 so
# downstream consumers can compare; JSON serialization uses
# ``default=str`` and renders inf as ``"inf"`` — callers checking
# numeric ordering must handle both representations.
EFFICIENCY_INFINITE = float("inf")
EFFICIENCY_UNDEFINED = 0.0   # zero gain at zero cost = no efficiency observed


def _efficiency(gain: float, cost: float) -> float:
    """Compute gain/cost with explicit zero-cost guards.

    - cost > 0      → gain / cost  (standard ratio)
    - cost == 0, gain > 0 → EFFICIENCY_INFINITE  (free improvement)
    - cost == 0, gain == 0 → EFFICIENCY_UNDEFINED  (no signal)
    - cost == 0, gain < 0 → -EFFICIENCY_INFINITE  (free regression — bad)
    """
    if cost > 0:
        return gain / cost
    if gain > 0:
        return EFFICIENCY_INFINITE
    if gain < 0:
        return -EFFICIENCY_INFINITE
    return EFFICIENCY_UNDEFINED


def _live_finetuning_measure(task: dict) -> dict:
    """Round-trip parent/child SelfModel + capability_claim through a
    fresh temp shard; read back measured values from the persisted
    rows. Tests the SelfModel persistence + claim-attach surface,
    not just static fixture data.
    """
    import os
    import tempfile

    from arborist.selfmodel import (
        CapabilityClaim,
        snapshot,
        store_snapshot,
    )
    from arborist.selfmodel.canonical import with_claims
    from arborist.selfmodel.store import claims_for
    from arborist.store import connect, transaction

    metric = task["target_capability"]
    parent_val = float(task["parent_measured_value"])
    child_val = float(task["child_measured_value"])
    threshold = float(task.get("expected_improvement_min", 0.05))

    tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
    tmp.close()
    try:
        conn = connect(tmp.name)
        try:
            with transaction(conn):
                parent_claim = CapabilityClaim(
                    metric=metric,
                    threshold=threshold,
                    eval_digest="live-finetuning-eval",
                    measured_value=parent_val,
                    measured_at=1700000000,
                )
                sm_parent = with_claims(snapshot(conn), [parent_claim])
                parent_root = store_snapshot(
                    conn, sm_parent, claims=[parent_claim], ts=1700000000
                )
            with transaction(conn):
                child_claim = CapabilityClaim(
                    metric=metric,
                    threshold=threshold,
                    eval_digest="live-finetuning-eval",
                    measured_value=child_val,
                    measured_at=1700000100,
                )
                sm_child = with_claims(snapshot(conn), [child_claim])
                child_root = store_snapshot(
                    conn, sm_child, claims=[child_claim], ts=1700000100
                )
            parent_rows = claims_for(conn, parent_root)
            child_rows = claims_for(conn, child_root)
            return {
                "parent_root": parent_root,
                "child_root": child_root,
                "parent_measured": parent_rows[0]["measured_value"],
                "child_measured": child_rows[0]["measured_value"],
            }
        finally:
            conn.close()
    finally:
        try:
            os.unlink(tmp.name)
        except OSError:
            pass


def _chain_finetuning_measure(task: dict) -> dict:
    """Read the two most-recent SelfModel snapshots from a *persistent*
    chain shard (#000025 §10.11) and return the parent/child measured
    values for ``target_capability``.

    The shard is the one ``bench/scripts/selfmodel_chain_snapshot.py``
    appends to (default ``~/.arborist/shards/selfmodel-chain.db``,
    overridable per-task via ``selfmodel_shard`` and via the
    ``ARBORIST_SELFMODEL_CHAIN_DB`` env var). Unlike the Phase-1b.2
    "live" path — which fabricates a parent + child in a fresh temp
    shard every run — this reads a genuine lineage that outlives the
    process: ``latest()`` is the child, its ``parent_selfmodel_root``
    is the parent.

    Raises if the shard is absent, has < 2 snapshots, or neither
    snapshot carries a claim for ``target_capability`` — the caller
    turns that into an honest task failure (this pack is meant to run
    *after* the chain has been bootstrapped, not on a fresh checkout).
    """
    import os as _os

    from arborist.selfmodel.store import claims_for, latest, load
    from arborist.store import connect

    metric = task["target_capability"]
    # Resolution order: ARBORIST_SELFMODEL_CHAIN_DB env (the test /
    # operator override) → the fixture's ``selfmodel_shard`` value
    # (which is also the dispatch gate) → the conventional default.
    raw_path = (
        _os.environ.get("ARBORIST_SELFMODEL_CHAIN_DB")
        or task.get("selfmodel_shard")
        or "~/.arborist/shards/selfmodel-chain.db"
    )
    shard = Path(_os.path.expandvars(str(raw_path))).expanduser()
    if not shard.exists():
        raise FileNotFoundError(
            f"selfmodel chain shard not found: {shard} — run "
            "`make bench-5f-selfmodel-snapshot` (twice) to bootstrap it"
        )
    conn = connect(shard)
    try:
        child_row = latest(conn)
        if child_row is None:
            raise ValueError(f"no SelfModel snapshots in {shard}")
        parent_root = child_row.get("parent_selfmodel_root")
        if not parent_root:
            raise ValueError(
                f"chain shard {shard} has only one snapshot; need >= 2 "
                "(run the bootstrapper again to extend the lineage)"
            )
        parent_row = load(conn, parent_root)
        if parent_row is None:
            raise ValueError(f"parent snapshot {parent_root[:12]}… missing from {shard}")

        def _measured(root: str) -> float:
            for c in claims_for(conn, root):
                if c["metric"] == metric and c["measured_value"] is not None:
                    return float(c["measured_value"])
            raise ValueError(f"no claim for {metric!r} on snapshot {root[:12]}…")

        return {
            "parent_root": parent_root,
            "child_root": child_row["selfmodel_root"],
            "parent_measured": _measured(parent_root),
            "child_measured": _measured(child_row["selfmodel_root"]),
        }
    finally:
        conn.close()


def run_finetuning(fixtures_path: Path) -> BatteryResult:
    """Adaptation improvement check + adaptation_efficiency metric.

    Each task asserts that ``child_measured_value >=
    parent_measured_value + expected_improvement_min``.

    Three fixture modes (selected per-task):

    - **Embedded** (Phase 1a): parent/child measured values come
      directly from the fixture; runner does the math.
    - **Live** (Phase 1b.2 — gate via ``"live": true``): runner
      writes parent + child SelfModels to a fresh temp shard via
      real :func:`arborist.selfmodel.store_snapshot`, attaches a
      :class:`arborist.selfmodel.CapabilityClaim` to each, reads
      the persisted measured values back from
      :func:`arborist.selfmodel.store.claims_for`, then runs the
      same improvement check. Tests the SelfModel persistence
      surface, not just the fixture data.
    - **Shard-chain** (§10.11 — gate via ``"selfmodel_shard"``):
      runner reads the two most-recent snapshots from a *persistent*
      chain shard (the one ``bench/scripts/selfmodel_chain_snapshot.py``
      grows, one snapshot per run) and measures improvement on
      ``target_capability`` between them. This is the real lineage —
      not a pair fabricated for the test — so the chained Δ reflects
      genuine cross-run drift (0.0 today; the packs are at ceiling).
      A task whose shard is absent / too short fails honestly: the
      pack runs *after* the bootstrapper, not on a fresh checkout.

    Per #000025 §5.2 and the 2026-05-08 fbd99a8 review:
    ``adaptation_efficiency = Δimprovement / Δcapital_cost`` with
    explicit zero-cost guards (see :func:`_efficiency`).
    """
    per_task: list[TaskResult] = []
    efficiency_values: list[float] = []
    finite_efficiency_values: list[float] = []
    for task in iter_tasks(fixtures_path):
        task_id = task["id"]
        ok, reason = _carrier_check(task)
        if not ok:
            per_task.append(
                TaskResult(task_id=task_id, passed=False, detail={"reason": reason})
            )
            continue
        try:
            if task.get("selfmodel_shard"):
                source = "shard-chain"
                chain = _chain_finetuning_measure(task)
                parent = float(chain["parent_measured"])
                child = float(chain["child_measured"])
            elif task.get("live"):
                source = "live"
                live = _live_finetuning_measure(task)
                parent = float(live["parent_measured"])
                child = float(live["child_measured"])
            else:
                source = "embedded"
                parent = float(task["parent_measured_value"])
                child = float(task["child_measured_value"])
            min_improvement = float(task.get("expected_improvement_min", 0.05))
            improvement = child - parent
            improved = improvement >= min_improvement
            expected = task.get("expected", "pass")
            observed = "pass" if improved else "fail"
            passed = observed == expected
            cost = _capital_cost_delta(task)
            efficiency = _efficiency(improvement, cost)
            efficiency_values.append(efficiency)
            if efficiency not in (EFFICIENCY_INFINITE, -EFFICIENCY_INFINITE):
                finite_efficiency_values.append(efficiency)
            detail = {
                "source": source,
                "parent_measured_value": parent,
                "child_measured_value": child,
                "improvement": improvement,
                "min_improvement": min_improvement,
                "capital_cost_delta": cost,
                "adaptation_efficiency": efficiency,
                "expected": expected,
            }
            if source == "shard-chain":
                detail["chain_parent_root"] = chain["parent_root"]
                detail["chain_child_root"] = chain["child_root"]
        except Exception as exc:  # noqa: BLE001
            passed = False
            detail = {"reason": f"{type(exc).__name__}: {exc}"}
        per_task.append(TaskResult(task_id=task_id, passed=passed, detail=detail))

    total = len(per_task)
    rate = sum(1 for t in per_task if t.passed) / total if total else 0.0
    # Mean efficiency reported only over the finite tasks; +inf and
    # -inf participants are counted in their own buckets so the mean
    # stays a meaningful summary.
    mean_finite_efficiency = (
        sum(finite_efficiency_values) / len(finite_efficiency_values)
        if finite_efficiency_values else 0.0
    )
    inf_count = sum(1 for e in efficiency_values if e == EFFICIENCY_INFINITE)
    neg_inf_count = sum(1 for e in efficiency_values if e == -EFFICIENCY_INFINITE)
    meta = fixture_meta(fixtures_path)
    return _build_result(
        meta.get("sub_battery", "finetuning"),
        fixtures_path,
        per_task,
        {
            "adaptation_improvement_rate": rate,
            "adaptation_efficiency_mean_finite": mean_finite_efficiency,
            "adaptation_efficiency_infinite_count": float(inf_count),
            "adaptation_efficiency_neg_infinite_count": float(neg_inf_count),
        },
    )


# ---------------------------------------------------------------------
# Falsification — does the organism detect and correct errors?
# ---------------------------------------------------------------------


def _live_falsification_violations(task: dict) -> list[str]:
    """Run real :func:`arborist.qa.verify.verify_quotes` against the
    fixture's (answer_text, context) pair; surface the violation
    signals as a flat list the runner can match against
    ``expected_reason``.

    Live mode signals that map to fixture ``expected_reason`` tags:

    - ``UNGROUNDED`` audit_mode → emit ``"UNGROUNDED"``.
    - Any unverified_quotes → emit ``"UNVERIFIED_QUOTE"`` for each.
    - HYBRID with quote-strategy resolution → emit ``"HYBRID_QUOTE"``.

    The full violation taxonomy (TITLE_MISMATCH, WARRANT_MISSING,
    LAZY_ANCHOR_DEMOTED, etc.) lives in the claim-lattice verifier;
    Phase 1b.2 wires only the quote-path verifier — the broader
    surface is reserved for Phase 1b.3.
    """
    from arborist.qa.verify import verify_quotes

    verdict = verify_quotes(task["answer_text"], task["context"])
    out: list[str] = []
    if verdict.get("audit_mode") == "UNGROUNDED":
        out.append("UNGROUNDED")
    if verdict.get("audit_mode") == "HYBRID":
        out.append(f"HYBRID_{verdict.get('verifier_method', 'unknown').upper()}")
    if verdict.get("audit_mode") == "STRICT":
        out.append(f"STRICT_{verdict.get('verifier_method', 'unknown').upper()}")
    if verdict.get("unverified_quotes"):
        out.append("UNVERIFIED_QUOTE")
    return out


def run_falsification(fixtures_path: Path) -> BatteryResult:
    """Planted-error detection rate.

    Two fixture modes:

    - **Embedded** (Phase 1a): fixture lists ``observed_violations``
      directly; runner checks ``expected_reason`` is present.
    - **Live** (Phase 1b.2): fixture provides
      ``answer_text`` + ``context``; runner calls real
      :func:`arborist.qa.verify.verify_quotes` and translates the
      verdict into violation signals (``UNGROUNDED``,
      ``UNVERIFIED_QUOTE``, ``HYBRID_<method>``, ``STRICT_<method>``).
      Tests the live verifier surface against planted (claim,
      evidence) pairs.

    Each fixture pinned to ``verifier_method_root`` so verifier-
    shape changes warn rather than false-fail (per ticket #000025
    risk #3).
    """
    per_task: list[TaskResult] = []
    for task in iter_tasks(fixtures_path):
        task_id = task["id"]
        ok, reason = _carrier_check(task)
        if not ok:
            per_task.append(
                TaskResult(task_id=task_id, passed=False, detail={"reason": reason})
            )
            continue
        try:
            if "answer_text" in task and "context" in task:
                source = "live"
                observed = set(_live_falsification_violations(task))
            else:
                source = "embedded"
                observed = set(task.get("observed_violations", []))
            expected_reason = task["expected_reason"]
            expected_outcome = task.get("expected", "pass")
            detected = expected_reason in observed
            observed_outcome = "pass" if detected else "fail"
            passed = observed_outcome == expected_outcome
            detail = {
                "source": source,
                "expected_reason": expected_reason,
                "observed_violations": sorted(observed),
                "detected": detected,
                "verifier_method_root": task.get("verifier_method_root"),
            }
        except Exception as exc:  # noqa: BLE001
            passed = False
            detail = {"reason": f"{type(exc).__name__}: {exc}"}
        per_task.append(TaskResult(task_id=task_id, passed=passed, detail=detail))

    total = len(per_task)
    rate = sum(1 for t in per_task if t.passed) / total if total else 0.0
    meta = fixture_meta(fixtures_path)
    return _build_result(
        meta.get("sub_battery", "falsification"),
        fixtures_path,
        per_task,
        {"error_detection_rate": rate},
    )


# ---------------------------------------------------------------------
# Formulate — turn unstructured input into structured output.
# ---------------------------------------------------------------------


def _live_produced_lattice(input_text: str) -> dict:
    """Derive a ``produced_lattice`` by routing ``input_text`` through
    arborist's actual claim-lattice parser. Phase 1b.2 entry point —
    moves the runner from synthetic gold to live organism output.

    Returns the same shape Phase 1a fixtures embed:
    ``{"claims": [{"claim_text": ..., "pointer_ids": [...]}, ...]}``.
    """
    from arborist.qa.parse_claims import parse_pointer_claims

    parsed = parse_pointer_claims(input_text)
    return {
        "claims": [
            {"claim_text": c.claim_text, "pointer_ids": list(c.pointer_ids)}
            for c in parsed
        ],
    }


def run_formulate(fixtures_path: Path) -> BatteryResult:
    """Structural match against expected claim lattice.

    Match policy: claim count + sorted-approx claim text + exact
    pointer-ID-set. NOT exact-string.

    Two fixture modes (selected per-task):

    - **Embedded** (Phase 1a): fixture provides ``produced_lattice``
      directly. Used for the 10 seed fixtures landed under #000025.
    - **Live** (Phase 1b.2): fixture provides ``input_text`` only;
      runner calls
      :func:`arborist.qa.parse_claims.parse_pointer_claims` and
      tests the actual organism's parser against
      ``expected_lattice``.

    Embedded takes precedence if both fields are present.
    Per-task ``detail.source`` reports which path ran, so the bench
    output distinguishes synthetic from live signal.
    """
    per_task: list[TaskResult] = []
    for task in iter_tasks(fixtures_path):
        task_id = task["id"]
        ok, reason = _carrier_check(task)
        if not ok:
            per_task.append(
                TaskResult(task_id=task_id, passed=False, detail={"reason": reason})
            )
            continue
        try:
            if "produced_lattice" in task:
                produced = task["produced_lattice"]
                source = "embedded"
            elif "input_text" in task:
                produced = _live_produced_lattice(task["input_text"])
                source = "live"
            else:
                raise KeyError(
                    "task needs 'produced_lattice' (embedded) or "
                    "'input_text' (live derivation)"
                )
            expected = task["expected_lattice"]
            count_ok = len(produced.get("claims", [])) == expected.get("claim_count")
            order_ok = True
            pointer_ok = True
            text_ok = True
            for i, exp in enumerate(expected.get("claims", [])):
                if i >= len(produced.get("claims", [])):
                    count_ok = False
                    break
                got = produced["claims"][i]
                # Pointer-set exact match.
                if set(got.get("pointer_ids", [])) != set(exp.get("pointer_ids", [])):
                    pointer_ok = False
                # Sorted-approx claim text — content-token overlap ≥ 0.7.
                got_t = _content_tokens(got.get("claim_text", ""))
                exp_t = _content_tokens(exp.get("claim_text", ""))
                if not exp_t or len(got_t & exp_t) / len(exp_t) < 0.7:
                    text_ok = False
            passed_all = count_ok and order_ok and pointer_ok and text_ok
            expected_outcome = task.get("expected", "pass")
            observed = "pass" if passed_all else "fail"
            passed = observed == expected_outcome
            detail = {
                "source": source,
                "count_ok": count_ok,
                "pointer_ok": pointer_ok,
                "text_ok": text_ok,
            }
        except Exception as exc:  # noqa: BLE001
            passed = False
            detail = {"reason": f"{type(exc).__name__}: {exc}"}
        per_task.append(TaskResult(task_id=task_id, passed=passed, detail=detail))

    total = len(per_task)
    rate = sum(1 for t in per_task if t.passed) / total if total else 0.0
    meta = fixture_meta(fixtures_path)
    return _build_result(
        meta.get("sub_battery", "formulate"),
        fixtures_path,
        per_task,
        {"structural_match_rate": rate},
    )


# ---------------------------------------------------------------------
# Feedback Loop — does observation at t affect state at t+1?
# ---------------------------------------------------------------------


def _live_feedback_chain(live_chain: list[dict]) -> dict:
    """Apply ``live_chain`` ops against a fresh temp arborist shard;
    return the final state for the runner to assert against.

    Supported ops:

    - ``append_audit`` — calls :func:`arborist.store.append_audit`
      with the fixture-supplied ``event_type``, ``subject_root``,
      and ``body``.
    - ``memory_snapshot`` — calls
      :func:`arborist.memory.snapshot` +
      :func:`arborist.memory.store_snapshot` to persist a memory
      snapshot.
    - ``selfmodel_snapshot`` — same for SelfModel.

    Returns ``{"audit_event_types": [...], "memory_branch_ids":
    [...], "memory_event_bodies": [...]}`` — enough for the runner
    to check whether `expected_delta` propagated into downstream
    state.
    """
    import os
    import tempfile
    import json as _json

    from arborist.memory import (
        snapshot as memory_snapshot,
        store_snapshot as memory_store,
    )
    from arborist.selfmodel import (
        snapshot as selfmodel_snapshot,
        store_snapshot as selfmodel_store,
    )
    from arborist.store import append_audit, connect, transaction

    tmp = tempfile.NamedTemporaryFile(suffix=".db", delete=False)
    tmp.close()
    try:
        conn = connect(tmp.name)
        try:
            with transaction(conn):
                for step in live_chain:
                    op = step["op"]
                    if op == "append_audit":
                        append_audit(
                            conn,
                            event_type=step["event_type"],
                            subject_root=step.get("subject_root"),
                            body=step.get("body", {}),
                        )
                    elif op == "memory_snapshot":
                        ms = memory_snapshot(conn)
                        memory_store(conn, ms)
                    elif op == "selfmodel_snapshot":
                        sm = selfmodel_snapshot(conn)
                        selfmodel_store(conn, sm)
                    else:
                        raise ValueError(f"unknown live op: {op!r}")
            # Read final state.
            audit_types = [
                row["event_type"] for row in conn.execute(
                    "SELECT event_type FROM audit_events ORDER BY seq"
                )
            ]
            memory_branches = [
                row["branch_id"] for row in conn.execute(
                    "SELECT DISTINCT branch_id FROM memory_branch_summaries"
                )
            ]
            event_bodies = [
                row["body"] for row in conn.execute(
                    "SELECT body FROM audit_events ORDER BY seq"
                )
            ]
            return {
                "audit_event_types": audit_types,
                "memory_branch_ids": memory_branches,
                "audit_event_bodies": event_bodies,
            }
        finally:
            conn.close()
    finally:
        try:
            os.unlink(tmp.name)
        except OSError:
            pass


def _live_delta_satisfied(state: dict, expected: dict) -> tuple[bool, str]:
    """Check expected_delta predicates against live final state.

    Supported predicates:

    - ``audit_event_type_present`` — the named event_type appears at
      least once in the audit chain.
    - ``memory_branch_present`` — the named branch_id appears in
      memory_branch_summaries.
    - ``body_substring_present`` — the substring appears in at least
      one audit event body (loose check).
    """
    for et in expected.get("audit_event_type_present", []):
        if et not in state["audit_event_types"]:
            return False, f"expected audit event_type {et!r} absent"
    for bid in expected.get("memory_branch_present", []):
        if bid not in state["memory_branch_ids"]:
            return False, f"expected memory branch {bid!r} absent"
    for needle in expected.get("body_substring_present", []):
        if not any(needle in body for body in state["audit_event_bodies"]):
            return False, f"substring {needle!r} absent from audit bodies"
    return True, "all live deltas satisfied"


def _persisted_cost(state: dict) -> float:
    """Real-workload cost of a live feedback chain: the persisted
    footprint it actually produced, not the count of *requested* ops.

    ``audit_event`` rows the chain wrote (count) + their body bytes /
    1e6 (matching :func:`_capital_cost_delta`'s 1 MB = 1 cost-unit
    convention). Every non-empty live chain writes ≥ 1 audit event
    (the ``append_audit`` op, the ``memory_snapshot_landed`` event,
    etc.), so this is > 0 for any chain that did work — :func:`_efficiency`
    still handles the cost-0 edge as documented.
    """
    n_audit = len(state.get("audit_event_types", []))
    body_bytes = sum(len(b) for b in state.get("audit_event_bodies", []) if b)
    return float(n_audit) + body_bytes / 1e6


def run_feedback_loop(fixtures_path: Path) -> BatteryResult:
    """Integration coverage + feedback_efficiency + feedback_latency.

    Two fixture modes (selected per-task):

    - **Embedded** (Phase 1a): fixture provides a ``chain`` of
      ``(operation, observation)`` pairs; runner aggregates the
      observation text and checks ``expected_delta`` (a string
      marker) appears in it. No real work happens, so latency is
      ``None`` and the cost-proxy stays ``len(chain)``.
    - **Live** (Phase 1b.2): fixture provides ``live_chain`` —
      a sequence of ops applied to a fresh temp arborist shard via
      :func:`_live_feedback_chain` — plus an ``expected_delta``
      dict with predicates against the resulting audit chain /
      memory branches. Tests whether observations actually
      propagate into downstream arborist state.

    Per #000025 §5.5 and the 2026-05-08 fbd99a8 review:
    ``feedback_efficiency = downstream_effect_count / Δcapital_cost``
    with explicit zero-cost guards (see :func:`_efficiency`).

    **§10.13 (2026-05-11) — calibrated against real workload.** For a
    live chain the cost is the *persisted footprint* :func:`_persisted_cost`
    — the audit-event rows the chain actually wrote plus their body
    bytes — instead of the Phase-1a ``len(chain)`` proxy (which
    counted *requested* ops, not work done). §10.13 also computes
    ``feedback_latency`` — listed in §5.5 since Phase 1a but never
    implemented — as the wall-clock seconds to apply a live chain
    against its temp shard. Latency is a wall-clock field
    (non-deterministic run-to-run, like ``BatteryResult.timestamp``)
    and is **not** consumed by ``fork_score`` — it's surfaced for
    operational observability + the capital ledger (#000020), not for
    fork selection. Embedded tasks keep ``len(chain)`` as cost and
    report ``feedback_latency_seconds = None``.
    """
    per_task: list[TaskResult] = []
    integrated = 0
    total_obs = 0
    efficiency_values: list[float] = []
    finite_efficiency_values: list[float] = []
    live_latencies: list[float] = []
    for task in iter_tasks(fixtures_path):
        task_id = task["id"]
        ok, reason = _carrier_check(task)
        if not ok:
            per_task.append(
                TaskResult(task_id=task_id, passed=False, detail={"reason": reason})
            )
            continue
        try:
            if "live_chain" in task:
                source = "live"
                live_chain = task["live_chain"]
                _t0 = time.perf_counter()
                state = _live_feedback_chain(live_chain)
                latency_s = time.perf_counter() - _t0
                live_latencies.append(latency_s)
                ok_live, why = _live_delta_satisfied(
                    state, task.get("expected_delta", {}) or {}
                )
                integrated_ok = ok_live
                chain_length = len(live_chain)
                total_obs += chain_length
                if integrated_ok:
                    integrated += 1
                expected_outcome = task.get("expected", "pass")
                observed = "pass" if integrated_ok else "fail"
                passed = observed == expected_outcome
                downstream_effect = 1.0 if integrated_ok else 0.0
                cost = _persisted_cost(state)  # §10.13: real footprint
                efficiency = _efficiency(downstream_effect, cost)
                efficiency_values.append(efficiency)
                if efficiency not in (
                    EFFICIENCY_INFINITE, -EFFICIENCY_INFINITE
                ):
                    finite_efficiency_values.append(efficiency)
                detail = {
                    "source": source,
                    "integrated": integrated_ok,
                    "reason": why,
                    "chain_length": chain_length,
                    "feedback_efficiency": efficiency,
                    "feedback_latency_seconds": latency_s,
                    "persisted_cost": cost,
                    "persisted_audit_rows": len(state.get("audit_event_types", [])),
                    "live_state": {
                        "audit_event_count": len(state["audit_event_types"]),
                        "memory_branch_count": len(state["memory_branch_ids"]),
                    },
                }
            else:
                source = "embedded"
                chain = task["chain"]
                total_obs += len(chain)
                observed_text = " ".join(
                    step.get("observation", "") for step in chain
                ).lower()
                final = chain[-1]
                delta_marker = (final.get("expected_delta") or "").lower()
                integrated_ok = bool(delta_marker) and delta_marker in observed_text
                if integrated_ok:
                    integrated += 1
                expected_outcome = task.get("expected", "pass")
                observed = "pass" if integrated_ok else "fail"
                passed = observed == expected_outcome
                downstream_effect = 1.0 if integrated_ok else 0.0
                cost = float(len(chain))
                efficiency = _efficiency(downstream_effect, cost)
                efficiency_values.append(efficiency)
                if efficiency not in (
                    EFFICIENCY_INFINITE, -EFFICIENCY_INFINITE
                ):
                    finite_efficiency_values.append(efficiency)
                detail = {
                    "source": source,
                    "integrated": integrated_ok,
                    "delta_marker": delta_marker,
                    "chain_length": len(chain),
                    "feedback_efficiency": efficiency,
                    "feedback_latency_seconds": None,  # no real work
                }
        except Exception as exc:  # noqa: BLE001
            passed = False
            detail = {"reason": f"{type(exc).__name__}: {exc}"}
        per_task.append(TaskResult(task_id=task_id, passed=passed, detail=detail))

    total = len(per_task)
    rate = sum(1 for t in per_task if t.passed) / total if total else 0.0
    integration_rate = integrated / total_obs if total_obs else 0.0
    mean_finite_efficiency = (
        sum(finite_efficiency_values) / len(finite_efficiency_values)
        if finite_efficiency_values else 0.0
    )
    inf_count = sum(1 for e in efficiency_values if e == EFFICIENCY_INFINITE)
    # §10.13: mean wall-clock latency over the live tasks (0.0 when a
    # pack has no live_chain tasks). Wall-clock, run-to-run variable —
    # not a fork_score input.
    feedback_latency_mean = (
        sum(live_latencies) / len(live_latencies) if live_latencies else 0.0
    )
    meta = fixture_meta(fixtures_path)
    return _build_result(
        meta.get("sub_battery", "feedback-loop"),
        fixtures_path,
        per_task,
        {
            "integration_coverage_rate": rate,
            "downstream_effect_rate": integration_rate,
            "feedback_efficiency_mean_finite": mean_finite_efficiency,
            "feedback_efficiency_infinite_count": float(inf_count),
            "feedback_latency_mean_seconds": feedback_latency_mean,
            "feedback_live_task_count": float(len(live_latencies)),
        },
    )


SUB_BATTERIES = {
    "function": run_function,
    "finetuning": run_finetuning,
    "falsification": run_falsification,
    "formulate": run_formulate,
    "feedback-loop": run_feedback_loop,
}