capital: land ticket #000020 (8-capital-form cost ledger)

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.

arborist/capital/__init__.py

@@ -0,0 +1,53 @@
+"""Capital-cost ledger: tying arborist ops to the 8 forms of capital.
+
+Implements ticket #000020. Sibling table to ``audit_events`` —
+``capital_ledger`` rows do NOT enter ``audit_events.event_hash``
+preimage, so retroactive cost re-estimation cannot break the audit
+chain.
+
+The 8 forms (Roland & Landua via unturf.com/eight-forms-of-capital):
+
+1. **Living** — soil, water, human health, ecosystems
+2. **Material** — servers, tools, infrastructure
+3. **Financial** — money, currencies, securities
+4. **Intellectual** — ideas, knowledge, open source
+5. **Experiential** — embodied skill, mastery, know-how
+6. **Social** — trust, relationships, community networks
+7. **Cultural** — story, art, ceremony, shared ethics
+8. **Spiritual** — presence, connection to a greater whole
+
+Each :class:`CapitalProfile` records contributions / debits across all
+8 forms in their native units (no conversion). Profiles are advisory
+by default — passing ``capital_profile=None`` to ``record`` emits no
+ledger row, fully backward-compatible.
+
+Public surface:
+
+- :class:`CapitalProfile` — frozen dataclass; one float per form.
+- :func:`record` — write a ledger row tied to an audit_event_hash.
+- :func:`summary` / :func:`top_by_form` / :func:`op_cost` — read APIs.
+"""
+
+from __future__ import annotations
+
+from arborist.capital.profile import (
+    ESTIMATOR_VERSION,
+    CapitalProfile,
+    profile_for_op,
+)
+from arborist.capital.store import (
+    op_cost,
+    record,
+    summary,
+    top_by_form,
+)
+
+__all__ = [
+    "ESTIMATOR_VERSION",
+    "CapitalProfile",
+    "profile_for_op",
+    "record",
+    "summary",
+    "top_by_form",
+    "op_cost",
+]

arborist/capital/profile.py

@@ -0,0 +1,181 @@
+"""CapitalProfile dataclass + estimator dispatch.
+
+Estimator versioning rule (per ticket #000020 §6 — risks):
+
+- ``ESTIMATOR_VERSION`` is the global pin. Bump it when the
+  estimator function set changes. Old rows keep their version
+  recorded so retroactive re-estimation can be rebuilt cleanly.
+- Per-op-type estimators are registered in :data:`ESTIMATORS`;
+  unknown op types fall through to a zero-profile + the version
+  pin.
+
+Cost-unit conventions (no cross-form conversion at this layer):
+
+- ``living`` — person-minutes equivalent (operator + downstream user
+  attention).
+- ``material`` — kWh equivalent for compute + storage (rough).
+- ``financial`` — USD equivalent for paid resources (e.g., LLM API
+  spend).
+- ``intellectual`` — net contribution (positive when op produces
+  reusable knowledge artifacts; negative when op consumes them
+  without preservation).
+- ``experiential`` — operator wait-time minutes.
+- ``social`` — trust delta on a ±1 named scale.
+- ``cultural`` — analogous ±1.
+- ``spiritual`` — analogous ±1.
+"""
+
+from __future__ import annotations
+
+from dataclasses import asdict, dataclass, field, replace
+from typing import Callable, Optional
+
+
+ESTIMATOR_VERSION = "estimator-v1"
+
+
+@dataclass(frozen=True)
+class CapitalProfile:
+    living: float = 0.0
+    material: float = 0.0
+    financial: float = 0.0
+    intellectual: float = 0.0
+    experiential: float = 0.0
+    social: float = 0.0
+    cultural: float = 0.0
+    spiritual: float = 0.0
+
+    def as_dict(self) -> dict:
+        return asdict(self)
+
+    @classmethod
+    def zero(cls) -> "CapitalProfile":
+        return cls()
+
+
+# ---------------------------------------------------------------------
+# Estimator dispatch
+#
+# Each estimator returns a (CapitalProfile, inputs_dict) pair. The
+# inputs_dict goes to capital_ledger.estimator_inputs_blob so retroactive
+# re-estimation can be reproduced exactly.
+# ---------------------------------------------------------------------
+
+Estimator = Callable[[dict], tuple[CapitalProfile, dict]]
+
+
+def _estimate_ingest(args: dict) -> tuple[CapitalProfile, dict]:
+    """Cost of ingesting documents.
+
+    Inputs (all optional; missing keys default to 0):
+    - ``doc_count`` — number of documents ingested
+    - ``total_bytes`` — uncompressed corpus size
+
+    Heuristic constants (initial v1 — refine empirically):
+    - 1e-6 kWh per ingested byte (storage write + index)
+    - 1e-9 person-minutes per byte (sysadmin tax, amortized)
+    - +0.001 intellectual per doc (positive: knowledge preserved)
+    """
+    doc_count = float(args.get("doc_count", 0))
+    total_bytes = float(args.get("total_bytes", 0))
+    inputs = {"doc_count": doc_count, "total_bytes": total_bytes}
+    return (
+        CapitalProfile(
+            material=total_bytes * 1e-6,
+            living=total_bytes * 1e-9,
+            intellectual=doc_count * 0.001,
+        ),
+        inputs,
+    )
+
+
+def _estimate_qa(args: dict) -> tuple[CapitalProfile, dict]:
+    """Cost of answering one query.
+
+    Inputs:
+    - ``prompt_chars`` — total prompt size in chars
+    - ``answer_chars`` — answer size in chars
+    - ``llm_seconds`` — wall-clock seconds spent in LLM call
+    - ``cache_hit`` — bool; cache hits are nearly free
+
+    Heuristics (LLM dominates):
+    - Cache hit: nearly zero except 0.01 experiential (operator wait)
+    - LLM call: 1e-3 kWh/sec (small model on commodity GPU)
+    - LLM call: $1e-5 / 1k chars (rough self-hosted equivalent)
+    - Answer chars: -0.0001 intellectual per char (consuming context)
+    """
+    cache_hit = bool(args.get("cache_hit", False))
+    prompt_chars = float(args.get("prompt_chars", 0))
+    answer_chars = float(args.get("answer_chars", 0))
+    llm_seconds = float(args.get("llm_seconds", 0.0))
+    inputs = {
+        "cache_hit": cache_hit,
+        "prompt_chars": prompt_chars,
+        "answer_chars": answer_chars,
+        "llm_seconds": llm_seconds,
+    }
+    if cache_hit:
+        return CapitalProfile(experiential=0.01), inputs
+    return (
+        CapitalProfile(
+            material=llm_seconds * 1e-3,
+            financial=(prompt_chars + answer_chars) * 1e-5 / 1000.0,
+            experiential=llm_seconds / 60.0,
+            intellectual=-answer_chars * 0.0001,
+        ),
+        inputs,
+    )
+
+
+def _estimate_distill(args: dict) -> tuple[CapitalProfile, dict]:
+    """Cost of distilling surface documents into core documents.
+
+    Inputs:
+    - ``surface_count`` — number of surface docs distilled
+    - ``core_count`` — number of core docs produced
+    - ``llm_seconds`` — wall-clock seconds spent in LLM call (if any)
+
+    Distillation produces reusable knowledge artifacts (cores), so
+    intellectual capital is positive.
+    """
+    surface_count = float(args.get("surface_count", 0))
+    core_count = float(args.get("core_count", 0))
+    llm_seconds = float(args.get("llm_seconds", 0.0))
+    inputs = {
+        "surface_count": surface_count,
+        "core_count": core_count,
+        "llm_seconds": llm_seconds,
+    }
+    return (
+        CapitalProfile(
+            material=llm_seconds * 1e-3,
+            experiential=llm_seconds / 60.0,
+            intellectual=core_count * 0.005,
+        ),
+        inputs,
+    )
+
+
+def _estimate_default(args: dict) -> tuple[CapitalProfile, dict]:
+    """Fallback: zero profile + the inputs dict for record-keeping."""
+    return CapitalProfile.zero(), dict(args)
+
+
+ESTIMATORS: dict[str, Estimator] = {
+    "ingest": _estimate_ingest,
+    "qa": _estimate_qa,
+    "distill": _estimate_distill,
+}
+
+
+def profile_for_op(
+    op_type: str, inputs: Optional[dict] = None
+) -> tuple[CapitalProfile, dict]:
+    """Dispatch to the right estimator; fallback to zero profile.
+
+    Returns ``(profile, recorded_inputs)``. The recorded inputs are
+    written to ``capital_ledger.estimator_inputs_blob`` for replay.
+    """
+    inputs = dict(inputs or {})
+    estimator = ESTIMATORS.get(op_type, _estimate_default)
+    return estimator(inputs)

arborist/capital/store.py

@@ -0,0 +1,142 @@
+"""CRUD over ``capital_ledger``.
+
+Sibling table semantics: ``capital_ledger`` rows reference an
+``audit_event_hash`` but do NOT enter the audit-event preimage. Rows
+can be re-estimated retroactively without invalidating any chain.
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+import time
+from typing import Optional
+
+from arborist.capital.profile import (
+    ESTIMATOR_VERSION,
+    CapitalProfile,
+)
+
+
+_FORMS = (
+    "living",
+    "material",
+    "financial",
+    "intellectual",
+    "experiential",
+    "social",
+    "cultural",
+    "spiritual",
+)
+
+
+def record(
+    conn: sqlite3.Connection,
+    *,
+    audit_event_hash: str,
+    op_type: str,
+    profile: CapitalProfile,
+    estimator_inputs: Optional[dict] = None,
+    estimator_version: str = ESTIMATOR_VERSION,
+    ts: Optional[int] = None,
+) -> int:
+    """Insert one ledger row; return ledger_id.
+
+    Idempotent on ``audit_event_hash`` only at the row level — i.e.,
+    we permit multiple ledger rows per audit event (e.g., one per
+    estimator version when re-estimating). Callers needing strict
+    idempotency should check first.
+    """
+    if ts is None:
+        ts = int(time.time())
+    inputs_blob = (
+        None
+        if estimator_inputs is None
+        else json.dumps(
+            estimator_inputs,
+            sort_keys=True,
+            separators=(",", ":"),
+            ensure_ascii=False,
+        )
+    )
+    cursor = conn.execute(
+        "INSERT INTO capital_ledger ("
+        " audit_event_hash, op_type, living, material, financial,"
+        " intellectual, experiential, social, cultural, spiritual,"
+        " estimator_version, estimator_inputs_blob, recorded_at"
+        ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+        (
+            audit_event_hash,
+            op_type,
+            profile.living,
+            profile.material,
+            profile.financial,
+            profile.intellectual,
+            profile.experiential,
+            profile.social,
+            profile.cultural,
+            profile.spiritual,
+            estimator_version,
+            inputs_blob,
+            ts,
+        ),
+    )
+    return int(cursor.lastrowid)
+
+
+def summary(
+    conn: sqlite3.Connection,
+    *,
+    since: Optional[int] = None,
+    op_type: Optional[str] = None,
+) -> dict:
+    """Aggregate totals across ledger; per-form sums + row count."""
+    where = []
+    params: list = []
+    if since is not None:
+        where.append("recorded_at >= ?")
+        params.append(int(since))
+    if op_type is not None:
+        where.append("op_type = ?")
+        params.append(op_type)
+    where_sql = ("WHERE " + " AND ".join(where)) if where else ""
+    sums_sql = ", ".join(f"SUM({f}) AS {f}" for f in _FORMS)
+    sql = (
+        f"SELECT COUNT(*) AS row_count, {sums_sql} "
+        f"FROM capital_ledger {where_sql}"
+    )
+    row = conn.execute(sql, params).fetchone()
+    if row is None:
+        return {"row_count": 0, **{f: 0.0 for f in _FORMS}}
+    out = {"row_count": int(row["row_count"])}
+    for f in _FORMS:
+        v = row[f]
+        out[f] = 0.0 if v is None else float(v)
+    return out
+
+
+def top_by_form(
+    conn: sqlite3.Connection,
+    form: str,
+    *,
+    limit: int = 10,
+) -> list[dict]:
+    """Return top-N op_types by total contribution to a single form."""
+    if form not in _FORMS:
+        raise ValueError(
+            f"unknown capital form: {form!r}; expected one of {_FORMS}"
+        )
+    rows = conn.execute(
+        f"SELECT op_type, SUM({form}) AS total_{form}, COUNT(*) AS row_count "
+        "FROM capital_ledger "
+        "GROUP BY op_type "
+        f"ORDER BY total_{form} DESC "
+        "LIMIT ?",
+        (int(limit),),
+    ).fetchall()
+    return [dict(r) for r in rows]
+
+
+def op_cost(conn: sqlite3.Connection, op_type: str) -> dict:
+    """Aggregate per-form totals for one op_type."""
+    return summary(conn, op_type=op_type)

arborist/cli.py

@@ -2721,6 +2721,45 @@ def _cmd_snapshot_diff(args: argparse.Namespace) -> int:
     return 0
 
 
+def _cmd_capital_summary(args: argparse.Namespace) -> int:
+    """Aggregate capital_ledger totals; per-form sums + row count."""
+    from arborist.capital import summary as capital_summary
+
+    conn = connect(args.db)
+    try:
+        out = capital_summary(conn, op_type=args.op_type, since=args.since)
+    finally:
+        conn.close()
+    print(json.dumps(out, indent=2, ensure_ascii=False))
+    return 0
+
+
+def _cmd_capital_op_cost(args: argparse.Namespace) -> int:
+    """Per-form totals for one op_type."""
+    from arborist.capital import op_cost
+
+    conn = connect(args.db)
+    try:
+        out = op_cost(conn, args.op_type)
+    finally:
+        conn.close()
+    print(json.dumps(out, indent=2, ensure_ascii=False))
+    return 0
+
+
+def _cmd_capital_top(args: argparse.Namespace) -> int:
+    """Top-N op_types by total contribution to a single capital form."""
+    from arborist.capital import top_by_form
+
+    conn = connect(args.db)
+    try:
+        out = top_by_form(conn, args.form, limit=args.limit)
+    finally:
+        conn.close()
+    print(json.dumps(out, indent=2, ensure_ascii=False))
+    return 0
+
+
 def _cmd_selfmodel_snapshot(args: argparse.Namespace) -> int:
     """Build a SelfModel from current store state and persist it.
 
@@ -4142,6 +4181,59 @@ def build_parser() -> argparse.ArgumentParser:
     snap_diff.add_argument("snapshot_root", help="hex snapshot_root to diff against current")
     snap_diff.set_defaults(func=_cmd_snapshot_diff)
 
+    # ----- capital subcommands (ticket #000020) -------------------------------
+    capital_cmd = sub.add_parser(
+        "capital",
+        help="8-capital-form cost ledger (ticket #000020)",
+    )
+    capital_sub = capital_cmd.add_subparsers(
+        dest="capital_op", required=True
+    )
+
+    cap_summary = capital_sub.add_parser(
+        "summary",
+        help="aggregate per-form sums across the ledger",
+    )
+    cap_summary.add_argument(
+        "--op-type",
+        dest="op_type",
+        default=None,
+        help="filter to one op_type",
+    )
+    cap_summary.add_argument(
+        "--since",
+        type=int,
+        default=None,
+        help="only rows with recorded_at >= this Unix timestamp",
+    )
+    cap_summary.set_defaults(func=_cmd_capital_summary)
+
+    cap_op = capital_sub.add_parser(
+        "op-cost", help="per-form totals for one op_type"
+    )
+    cap_op.add_argument("op_type", help="e.g. ingest|qa|distill")
+    cap_op.set_defaults(func=_cmd_capital_op_cost)
+
+    cap_top = capital_sub.add_parser(
+        "top", help="top-N op_types by total in one capital form"
+    )
+    cap_top.add_argument(
+        "--form",
+        required=True,
+        choices=[
+            "living",
+            "material",
+            "financial",
+            "intellectual",
+            "experiential",
+            "social",
+            "cultural",
+            "spiritual",
+        ],
+    )
+    cap_top.add_argument("--limit", type=int, default=10)
+    cap_top.set_defaults(func=_cmd_capital_top)
+
     # ----- selfmodel subcommands (ticket #000014) -----------------------------
     selfmodel_cmd = sub.add_parser(
         "selfmodel",

arborist/distill/runner.py

@@ -278,7 +278,7 @@ def _persist_no_tx(
         "VALUES (?, ?, ?, 'derived_from', '')",
         (p.core_root, p.src_root, p.src_uri),
     )
-    append_audit(
+    derive_event_hash = append_audit(
         conn,
         event_type="derive",
         subject_root=p.core_root,
@@ -297,4 +297,25 @@ def _persist_no_tx(
         },
         ts=now,
     )
+    # Capital ledger (ticket #000020). Distillation produces reusable
+    # core artifacts → positive intellectual capital.
+    from arborist.capital import profile_for_op, record as capital_record
+
+    capital_profile, capital_inputs = profile_for_op(
+        "distill",
+        {
+            "surface_count": 1,
+            "core_count": 1,
+            # llm_seconds is unknown at this site; estimator falls
+            # through to 0.0 cleanly when missing.
+        },
+    )
+    capital_record(
+        conn,
+        audit_event_hash=derive_event_hash,
+        op_type="distill",
+        profile=capital_profile,
+        estimator_inputs=capital_inputs,
+        ts=now,
+    )
     return "distilled"

arborist/ingest.py

@@ -308,6 +308,33 @@ def _flush_batch(
                 audit_rows,
             )
 
+            # Capital ledger (ticket #000020). One row per batch attached
+            # to the last event_hash. Sibling table — does NOT enter
+            # audit_events.event_hash preimage.
+            from arborist.capital import profile_for_op, record as capital_record
+
+            total_chunks = sum(
+                ev["body"].get("chunks", 0) for ev in audit_events
+            )
+            # Heuristic: ~512 tokens × ~4 chars/token per chunk.
+            approx_bytes = total_chunks * 512 * 4
+            profile, inputs = profile_for_op(
+                "ingest",
+                {
+                    "doc_count": len(audit_events),
+                    "total_bytes": approx_bytes,
+                },
+            )
+            last_event_hash = audit_rows[-1][0]
+            capital_record(
+                conn,
+                audit_event_hash=last_event_hash,
+                op_type="ingest",
+                profile=profile,
+                estimator_inputs=inputs,
+                ts=ingest_ts,
+            )
+
     return inserted, skipped
 
 

arborist/qa/runner.py

@@ -999,6 +999,25 @@ def ask(
             },
             ts=now,
         )
+        # Capital ledger (ticket #000020). Sibling table; advisory.
+        from arborist.capital import profile_for_op, record as capital_record
+
+        capital_profile, capital_inputs = profile_for_op(
+            "qa",
+            {
+                "cache_hit": False,
+                "answer_chars": len(answer_text),
+                "llm_seconds": llm_ms / 1000.0,
+            },
+        )
+        capital_record(
+            conn,
+            audit_event_hash=event_hash,
+            op_type="qa",
+            profile=capital_profile,
+            estimator_inputs=capital_inputs,
+            ts=now,
+        )
         conn.execute(
             "INSERT INTO providence_cache "
             "(cache_key, source_root, document_uri, question_hash, question_text, "

docs/TICKETS.md

@@ -59,7 +59,7 @@ Newest first. Update on every open/close.
 |----------|------------------------------------------------|-----------------------|------------|-----------|
 | #000022  | Adapter LossReport (PRD I9 analogue)           | open · awaiting go/no-go | 2026-05-07 | —         |
 | #000021  | 5S/5T/5R benchmark fixtures + harness          | open · awaiting go/no-go | 2026-05-07 | —         |
-| #000020  | Capital-cost ledger (8-capital queues)         | open · awaiting go/no-go | 2026-05-07 | —         |
+| #000020  | Capital-cost ledger (8-capital queues)         | closed · landed 2026-05-07 | 2026-05-07 | —         |
 | #000019  | Specification methodology for π* and V         | open · awaiting go/no-go | 2026-05-07 | —         |
 | #000018  | Adversarial soft-hash covert-channel analysis  | open · awaiting go/no-go | 2026-05-07 | —         |
 | #000017  | Memory-root: lifelong learning audit chain     | open · awaiting go/no-go | 2026-05-07 | —         |

docs/tickets/ticket-000020-capital-cost-ledger.md

@@ -1,7 +1,8 @@
 # Ticket #000020 — Capital-cost ledger: tying verifiable AGI to 8-capital queues
 
-**Status:** open · awaiting go/no-go
+**Status:** closed · landed 2026-05-07
 **Opened:** 2026-05-07
+**Closed:** 2026-05-07
 **Scope:** Schema + audit-event tag for a capital-cost ledger that
 attributes every state-changing arborist op to one or more of the 8
 forms of capital (Living, Material, Financial, Intellectual,
@@ -276,11 +277,40 @@ arborist capital top --form living --limit 10
 
 ## 7. Status
 
-**Open · awaiting go/no-go.** Small implementation surface
-(~300 lines code + ~150 lines doc + schema). One of the more
-shippable tickets in this batch.
+**Closed 2026-05-07.** Scope delivered:
 
-Closure criterion: `capital_ledger` schema lands, at least three op
-types (`ingest`, `qa.runner.ask`, `distill.runner.distill`) emit
-ledger rows on execution, `arborist capital summary` returns
-non-empty totals after a workload, audit chain stays clean.
+- Schema migration `_migrate_capital_ledger` adds `capital_ledger`
+  table (8 capital-form columns, op_type, estimator_version,
+  estimator_inputs_blob). Sibling table — does NOT enter
+  `audit_events.event_hash` preimage.
+- Module `arborist.capital`:
+  - `profile.py` — `CapitalProfile` dataclass + `profile_for_op`
+    estimator dispatch + per-op estimators (ingest, qa, distill).
+    `ESTIMATOR_VERSION = "estimator-v1"` global pin.
+  - `store.py` — `record(...)` writes one row keyed to an
+    audit_event_hash; `summary(...)`, `top_by_form(...)`,
+    `op_cost(...)` aggregate reads.
+- Wire-up at three op sites:
+  - `arborist/ingest.py` — one row per batch, attached to last
+    audit event_hash; estimator inputs `doc_count` + `total_bytes`.
+  - `arborist/qa/runner.py` — one row per cache-miss QA, attached
+    to `providence_write` event; estimator inputs `answer_chars` +
+    `llm_seconds`.
+  - `arborist/distill/runner.py` — one row per distillation,
+    attached to `derive` event; positive intellectual capital.
+- CLI: `arborist capital summary|op-cost|top`.
+- Tests: `tests/test_capital.py` — 13 cases covering profile
+  defaults, estimator dispatch, ledger row write, summary
+  aggregation, op-type filter, top-by-form ranking, sibling-table
+  invariant (audit chain unaffected by capital writes). All pass;
+  full suite 1025 passed, 36 skipped.
+
+Out-of-scope items (deferred to follow-ups):
+
+- Stewardship-halt runtime enforcement (`halt_on` policy
+  evaluation). Schema supports it; runtime check is its own ticket
+  once op profiles are tuned against real workloads.
+- Conversion tables between capital forms.
+- Real-time instrumentation (true measurement vs hardcoded
+  constants). v1 ships heuristic estimators; v2 will calibrate.
+- Cross-shard / cross-node ledger aggregation.

tests/test_capital.py

@@ -0,0 +1,278 @@
+"""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()