russell@unturf.com
69f91d39a6
CapitalProfile (8 forms: living, material, financial, intellectual,
experiential, social, cultural, spiritual) attached per state-changing
op as a sibling-table row in capital_ledger. Sibling semantics: ledger
rows reference an audit_event_hash but do NOT enter the audit-event
preimage, so retroactive cost re-estimation cannot break the chain.
Surface:
- arborist.capital.{profile,store}
- profile_for_op dispatch with per-op estimators (ingest/qa/distill)
- record/summary/op_cost/top_by_form
- CLI: arborist capital summary|op-cost|top
Wire-up at three op sites:
- ingest.py — one row per batch (doc_count + total_bytes)
- qa/runner.py — one row per cache-miss (answer_chars + llm_seconds)
- distill/runner.py — one row per derivation (positive intellectual)
Estimator constants are heuristic v1 (ESTIMATOR_VERSION pin in the
schema). Re-estimation is supported by re-running estimators against
the recorded inputs_blob and writing a new row with a bumped version
pin; old rows stay queryable.
Tests: tests/test_capital.py (13 cases). Sibling-table invariant
verified: audit chain stays intact across capital writes.
Full suite: 1025 passed, 36 skipped.
278 lines
8.1 KiB
Python
278 lines
8.1 KiB
Python
"""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()
|