feat: reporting-allocate AI-plan cost across work
Split a booked AI-plan fact across workplans or repos using measured tokens, measured+estimated tokens, session counts, or an even split. Unmeasured remainder stays a residual. The record is not resource-control AllocationEvidence and does not create booked spend.
This commit is contained in:
parent
8916f3c4d0
commit
efd1b31001
10 changed files with 646 additions and 3 deletions
98
tests/test_reporting_allocation.py
Normal file
98
tests/test_reporting_allocation.py
Normal file
|
|
@ -0,0 +1,98 @@
|
|||
import sqlite3
|
||||
from decimal import Decimal
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from fin_hub.services.allocation import shared_cost_allocations
|
||||
from fin_hub.services.exchange import ingest_planning_evidence
|
||||
from fin_hub.services.ledger import import_csv, monthly_summary
|
||||
from fin_hub.services.reporting_allocation import (
|
||||
allocate_ai_plan_report,
|
||||
list_ai_plan_allocations,
|
||||
)
|
||||
from fin_hub.services.session_tokens import ingest_session_token_file
|
||||
|
||||
FIXTURES = Path(__file__).parent / "fixtures"
|
||||
|
||||
|
||||
def _claude_fact(tmp_path: Path) -> tuple[Path, str]:
|
||||
ledger = tmp_path / "ledger.db"
|
||||
import_csv(FIXTURES / "ai-plans.csv", "ai-plan", ledger_path=ledger)
|
||||
with sqlite3.connect(ledger) as conn:
|
||||
fact_id = conn.execute(
|
||||
"SELECT financial_fact_id FROM ledger_entries "
|
||||
"WHERE source_type = 'ai-plan' AND label = 'claude-max' AND is_current = 1"
|
||||
).fetchone()[0]
|
||||
return ledger, fact_id
|
||||
|
||||
|
||||
def test_measured_token_share_two_workplans_keeps_unmeasured_residual(tmp_path: Path):
|
||||
ledger, fact_id = _claude_fact(tmp_path)
|
||||
ingest_session_token_file(
|
||||
FIXTURES / "session-tokens-two-workplans.json", ledger_path=ledger
|
||||
)
|
||||
report = allocate_ai_plan_report(
|
||||
fact_id, method="measured_token_share", ledger_path=ledger
|
||||
)
|
||||
by_key = {target.target_key: target for target in report.targets}
|
||||
assert report.booked_amount == Decimal("200.00")
|
||||
assert report.method == "measured_token_share"
|
||||
assert report.residual_reason == "unmeasured"
|
||||
assert report.residual_share == Decimal("0.25")
|
||||
assert report.residual_amount == Decimal("50.00")
|
||||
assert by_key["workplan:workplan-a"].amount == Decimal("100.00")
|
||||
assert by_key["workplan:workplan-b"].amount == Decimal("50.00")
|
||||
assert sum((target.amount for target in report.targets), report.residual_amount) == Decimal(
|
||||
"200.00"
|
||||
)
|
||||
assert monthly_summary(ledger_path=ledger)[0].total == pytest.approx(240.0)
|
||||
assert shared_cost_allocations(ledger_path=ledger) == []
|
||||
assert list_ai_plan_allocations(ledger_path=ledger)[0].record_id == report.record_id
|
||||
|
||||
|
||||
def test_even_split_does_not_use_token_weights(tmp_path: Path):
|
||||
ledger, fact_id = _claude_fact(tmp_path)
|
||||
ingest_session_token_file(
|
||||
FIXTURES / "session-tokens-two-workplans.json", ledger_path=ledger
|
||||
)
|
||||
report = allocate_ai_plan_report(fact_id, method="even_split", ledger_path=ledger)
|
||||
amounts = sorted(target.amount for target in report.targets)
|
||||
assert amounts == [Decimal("100.00"), Decimal("100.00")]
|
||||
assert report.residual_share == Decimal("0.00")
|
||||
assert report.residual_reason == "none"
|
||||
|
||||
|
||||
def test_reporting_allocation_is_not_planning_allocation(tmp_path: Path):
|
||||
ledger, fact_id = _claude_fact(tmp_path)
|
||||
report = allocate_ai_plan_report(fact_id, method="even_split", ledger_path=ledger)
|
||||
with pytest.raises(ValueError, match="not resource-control"):
|
||||
ingest_planning_evidence(
|
||||
{
|
||||
"record_type": "reporting_allocation",
|
||||
"record_id": report.record_id,
|
||||
"financial_fact_id": fact_id,
|
||||
"method": "even_split",
|
||||
},
|
||||
ledger_path=ledger,
|
||||
)
|
||||
with sqlite3.connect(ledger) as conn:
|
||||
planning = conn.execute(
|
||||
"SELECT count(*) FROM sqlite_master WHERE name = 'planning_evidence'"
|
||||
).fetchone()[0]
|
||||
if planning:
|
||||
stored = conn.execute(
|
||||
"SELECT record_type FROM planning_evidence"
|
||||
).fetchall()
|
||||
assert stored == []
|
||||
|
||||
|
||||
def test_non_ai_plan_fact_is_rejected(tmp_path: Path):
|
||||
ledger = tmp_path / "ledger.db"
|
||||
import_csv(FIXTURES / "cloud-costs.csv", "cloud", ledger_path=ledger)
|
||||
with sqlite3.connect(ledger) as conn:
|
||||
fact_id = conn.execute(
|
||||
"SELECT financial_fact_id FROM ledger_entries WHERE is_current = 1"
|
||||
).fetchone()[0]
|
||||
with pytest.raises(ValueError, match="ai-plan"):
|
||||
allocate_ai_plan_report(fact_id, method="even_split", ledger_path=ledger)
|
||||
Loading…
Add table
Add a link
Reference in a new issue