"""Capital-cost ledger tests (ticket #000020). Covers: - CapitalProfile defaults to zero - profile_for_op dispatch hits per-op estimators and falls through to zero-profile for unknown op types - record() writes one ledger row keyed to an audit_event_hash - summary() aggregates per-form sums correctly - top_by_form() ranks op_types - Sibling-table semantics: capital_ledger writes do NOT alter audit_events.event_hash chain """ from __future__ import annotations import time import pytest from arborist.capital import ( CapitalProfile, op_cost, profile_for_op, record, summary, top_by_form, ) from arborist.capital.profile import ( ESTIMATOR_VERSION, ESTIMATORS, _estimate_default, ) from arborist.store import append_audit, connect, transaction def test_capital_profile_zero_defaults(): p = CapitalProfile.zero() for f in ( "living", "material", "financial", "intellectual", "experiential", "social", "cultural", "spiritual", ): assert getattr(p, f) == 0.0 def test_profile_for_op_ingest_basic(): p, inputs = profile_for_op( "ingest", {"doc_count": 5, "total_bytes": 1_000_000} ) assert p.material > 0 # storage cost positive assert p.intellectual > 0 # docs preserved → positive assert inputs["doc_count"] == 5 def test_profile_for_op_qa_cache_hit_path(): p, inputs = profile_for_op("qa", {"cache_hit": True}) assert p.financial == 0 # cache hits don't charge LLM assert p.material == 0 assert p.experiential > 0 # operator wait, even if small assert inputs["cache_hit"] is True def test_profile_for_op_qa_miss_path(): p, inputs = profile_for_op( "qa", { "cache_hit": False, "answer_chars": 1000, "llm_seconds": 5.0, }, ) assert p.material > 0 assert p.financial > 0 assert p.experiential > 0 assert p.intellectual < 0 # consumes context def test_profile_for_op_distill_positive_intellectual(): p, _ = profile_for_op( "distill", {"surface_count": 1, "core_count": 1, "llm_seconds": 2.0}, ) assert p.intellectual > 0 # cores preserved def test_profile_for_op_unknown_falls_through(): p, inputs = profile_for_op("unknown_op", {"x": 1}) assert p == CapitalProfile.zero() assert inputs == {"x": 1} def test_record_writes_row(tmp_path): db = tmp_path / "shard.db" conn = connect(db) try: with transaction(conn): event_hash = append_audit( conn, event_type="ingest", subject_root="00" * 32, body={"x": 1}, ) ledger_id = record( conn, audit_event_hash=event_hash, op_type="ingest", profile=CapitalProfile(material=0.5, intellectual=0.01), estimator_inputs={"doc_count": 1}, ) assert ledger_id > 0 row = conn.execute( "SELECT op_type, material, intellectual, estimator_version," " estimator_inputs_blob" " FROM capital_ledger WHERE ledger_id = ?", (ledger_id,), ).fetchone() assert row["op_type"] == "ingest" assert row["material"] == pytest.approx(0.5) assert row["intellectual"] == pytest.approx(0.01) assert row["estimator_version"] == ESTIMATOR_VERSION assert "doc_count" in row["estimator_inputs_blob"] finally: conn.close() def test_summary_aggregates(tmp_path): db = tmp_path / "shard.db" conn = connect(db) try: with transaction(conn): for i in range(3): ev = append_audit( conn, event_type="ingest", subject_root=f"{i:064x}", body={"i": i}, ) record( conn, audit_event_hash=ev, op_type="ingest", profile=CapitalProfile(material=0.1, financial=0.01), ) out = summary(conn) assert out["row_count"] == 3 assert out["material"] == pytest.approx(0.3) assert out["financial"] == pytest.approx(0.03) finally: conn.close() def test_summary_filtered_by_op_type(tmp_path): db = tmp_path / "shard.db" conn = connect(db) try: with transaction(conn): ev1 = append_audit( conn, event_type="ingest", subject_root=None, body={"a": 1} ) record( conn, audit_event_hash=ev1, op_type="ingest", profile=CapitalProfile(material=1.0), ) ev2 = append_audit( conn, event_type="qa", subject_root=None, body={"b": 1} ) record( conn, audit_event_hash=ev2, op_type="qa", profile=CapitalProfile(material=2.0), ) ingest_only = summary(conn, op_type="ingest") qa_only = summary(conn, op_type="qa") all_in = summary(conn) assert ingest_only["material"] == pytest.approx(1.0) assert qa_only["material"] == pytest.approx(2.0) assert all_in["material"] == pytest.approx(3.0) finally: conn.close() def test_top_by_form_ranks(tmp_path): db = tmp_path / "shard.db" conn = connect(db) try: with transaction(conn): for op, val in [("ingest", 1.0), ("qa", 5.0), ("distill", 2.0)]: ev = append_audit( conn, event_type=op, subject_root=None, body={"k": op} ) record( conn, audit_event_hash=ev, op_type=op, profile=CapitalProfile(material=val), ) ranked = top_by_form(conn, "material", limit=10) op_types = [r["op_type"] for r in ranked] assert op_types[0] == "qa" assert op_types[-1] == "ingest" finally: conn.close() def test_op_cost_returns_op_summary(tmp_path): db = tmp_path / "shard.db" conn = connect(db) try: with transaction(conn): ev = append_audit( conn, event_type="qa", subject_root=None, body={"k": "v"} ) record( conn, audit_event_hash=ev, op_type="qa", profile=CapitalProfile(financial=0.05, experiential=0.5), ) out = op_cost(conn, "qa") assert out["financial"] == pytest.approx(0.05) assert out["experiential"] == pytest.approx(0.5) assert out["row_count"] == 1 finally: conn.close() def test_top_by_form_rejects_unknown_form(tmp_path): db = tmp_path / "shard.db" conn = connect(db) try: with pytest.raises(ValueError): top_by_form(conn, "imaginary_form") finally: conn.close() def test_audit_chain_unaffected_by_capital_writes(tmp_path): """capital_ledger writes are sibling-table; chain stays intact.""" import hashlib db = tmp_path / "shard.db" conn = connect(db) try: with transaction(conn): ev1 = append_audit( conn, event_type="x", subject_root=None, body={"a": 1} ) record( conn, audit_event_hash=ev1, op_type="x", profile=CapitalProfile(material=0.1), ) ev2 = append_audit( conn, event_type="y", subject_root=None, body={"b": 2} ) rows = conn.execute( "SELECT event_hash, prev_event_hash, body " "FROM audit_events ORDER BY seq" ).fetchall() prev = None for r in rows: h = hashlib.sha256() if prev is not None: h.update(bytes.fromhex(prev)) h.update(r["body"].encode("utf-8", errors="surrogatepass")) assert h.hexdigest() == r["event_hash"] prev = r["event_hash"] finally: conn.close()