Harden resource cost evidence contract

src/fin_hub/cli.py

@@ -139,13 +139,13 @@ def _cmd_ledger_set_price(args: argparse.Namespace) -> int:
         revision_of=args.revision_of,
         ledger_path=_ledger_path(args),
     )
-    print(json.dumps(price.__dict__, indent=2))
+    print(json.dumps(price.__dict__, indent=2, default=str))
     return 0
 
 
 def _cmd_ledger_margins(args: argparse.Namespace) -> int:
     rows = client_margin_report(ledger_path=_ledger_path(args))
-    print(json.dumps([row.as_dict() for row in rows], indent=2))
+    print(json.dumps([row.as_dict() for row in rows], indent=2, default=str))
     return 0
 
 

src/fin_hub/coupling/ops_hub.py

@@ -35,12 +35,14 @@ class ServiceCostLine:
 def build_service_cost_report(
     lines: Iterable[ServiceCostLine],
 ) -> dict:
-    by_service: dict[str, dict] = {}
-    totals_by_month: dict[str, float] = defaultdict(float)
-    by_attribution: dict[tuple[str, str], dict] = {}
+    by_service: dict[tuple[str, str, str], dict] = {}
+    totals_by_period_currency: dict[tuple[str, str], float] = defaultdict(float)
+    by_attribution: dict[tuple[str, str, str], dict] = {}
+    unattributed: dict[tuple[str, str], dict] = {}
     for line in lines:
+        service_key = (line.service_id, line.environment, line.currency)
         bucket = by_service.setdefault(
-            line.service_id,
+            service_key,
             {
                 "service_id": line.service_id,
                 "environment": line.environment,
@@ -52,8 +54,12 @@ def build_service_cost_report(
         month_total = bucket["months"].get(line.period_month, 0.0) + line.amount
         bucket["months"][line.period_month] = month_total
         bucket["total"] += line.amount
-        totals_by_month[line.period_month] += line.amount
-        key = (line.cost_attribution_key or "unattributed", line.currency)
+        totals_by_period_currency[(line.period_month, line.currency)] += line.amount
+        key = (
+            line.cost_attribution_key or "unattributed",
+            line.environment,
+            line.currency,
+        )
         attribution = by_attribution.setdefault(
             key,
             {
@@ -61,6 +67,7 @@ def build_service_cost_report(
                 "client_id": line.client_id,
                 "application_id": line.application_id,
                 "app_instance_id": line.app_instance_id,
+                "environment": line.environment,
                 "currency": line.currency,
                 "months": {},
                 "total": 0.0,
@@ -70,6 +77,21 @@ def build_service_cost_report(
             attribution["months"].get(line.period_month, 0.0) + line.amount
         )
         attribution["total"] += line.amount
+        if line.cost_attribution_key is None:
+            residual_key = (line.environment, line.currency)
+            residual = unattributed.setdefault(
+                residual_key,
+                {
+                    "environment": line.environment,
+                    "currency": line.currency,
+                    "months": {},
+                    "total": 0.0,
+                },
+            )
+            residual["months"][line.period_month] = (
+                residual["months"].get(line.period_month, 0.0) + line.amount
+            )
+            residual["total"] += line.amount
     services = sorted(by_service.values(), key=lambda item: item["total"], reverse=True)
     return {
         "source_hub": "fin-hub",
@@ -79,5 +101,11 @@ def build_service_cost_report(
         "attributions": sorted(
             by_attribution.values(), key=lambda item: item["total"], reverse=True
         ),
-        "totals_by_month": dict(sorted(totals_by_month.items())),
+        "unattributed": sorted(
+            unattributed.values(), key=lambda item: (item["currency"], item["environment"])
+        ),
+        "totals_by_period_currency": [
+            {"period_month": period, "currency": currency, "total": total}
+            for (period, currency), total in sorted(totals_by_period_currency.items())
+        ],
     }

src/fin_hub/models/engagement_price.py

@@ -5,29 +5,77 @@ from __future__ import annotations
 import uuid
 from datetime import date
 
-from sqlalchemy import Date, Float, ForeignKey, String
+from decimal import Decimal
+
+from sqlalchemy import Boolean, CheckConstraint, Date, ForeignKey, Index, Numeric, String, event, text
 from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.orm import Mapped, mapped_column
 
 from hub_core.models.base import Base, TimestampMixin
+from fin_hub.money import EngagementPriceTerms
 
 
 class EngagementPrice(Base, TimestampMixin):
     """A reporting entitlement price, not an invoice or payment record."""
 
     __tablename__ = "fin_engagement_prices"
+    __table_args__ = (
+        CheckConstraint("amount >= 0", name="ck_fin_engagement_prices_nonnegative"),
+        CheckConstraint("length(currency) = 3", name="ck_fin_engagement_prices_currency"),
+        CheckConstraint(
+            "effective_to IS NULL OR effective_to >= effective_from",
+            name="ck_fin_engagement_prices_effective_dates",
+        ),
+        Index(
+            "uq_fin_engagement_prices_current_basis",
+            "cost_attribution_key",
+            "period_month",
+            "currency",
+            unique=True,
+            postgresql_where=text("is_current"),
+            sqlite_where=text("is_current = 1"),
+        ),
+    )
 
     id: Mapped[uuid.UUID] = mapped_column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
     client_id: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
     application_id: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
     app_instance_id: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
     cost_attribution_key: Mapped[str] = mapped_column(String(423), nullable=False, index=True)
+    period_month: Mapped[str] = mapped_column(String(7), nullable=False, index=True)
     effective_from: Mapped[date] = mapped_column(Date, nullable=False, index=True)
     effective_to: Mapped[date | None] = mapped_column(Date, nullable=True)
-    amount: Mapped[float] = mapped_column(Float, nullable=False)
+    amount: Mapped[Decimal] = mapped_column(Numeric(18, 2), nullable=False)
     currency: Mapped[str] = mapped_column(String(3), nullable=False, default="EUR")
     source: Mapped[str] = mapped_column(String(128), nullable=False)
     revision_of: Mapped[uuid.UUID | None] = mapped_column(
         UUID(as_uuid=True), ForeignKey("fin_engagement_prices.id"), nullable=True
     )
+    is_current: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True)
     notes: Mapped[str | None] = mapped_column(String(512), nullable=True)
+
+
+@event.listens_for(EngagementPrice, "before_insert")
+@event.listens_for(EngagementPrice, "before_update")
+def _validate_engagement_price(_mapper, _connection, target: EngagementPrice) -> None:
+    if target.effective_to is not None and target.effective_to < target.effective_from:
+        raise ValueError("effective_to cannot precede effective_from")
+    terms = EngagementPriceTerms.validate(
+        client_id=target.client_id,
+        application_id=target.application_id,
+        app_instance_id=target.app_instance_id,
+        period_month=target.effective_from.strftime("%Y-%m"),
+        amount=target.amount,
+        currency=target.currency,
+        source=target.source,
+    )
+    target.client_id = terms.attribution.client_id
+    target.application_id = terms.attribution.application_id
+    target.app_instance_id = terms.attribution.app_instance_id
+    if target.cost_attribution_key not in (None, terms.attribution.key):
+        raise ValueError("cost_attribution_key does not match its attribution dimensions")
+    target.cost_attribution_key = terms.attribution.key
+    target.period_month = terms.period_month
+    target.amount = terms.amount
+    target.currency = terms.currency
+    target.source = terms.source

src/fin_hub/models/service_cost.py

@@ -4,13 +4,15 @@ from __future__ import annotations
 
 import uuid
 from datetime import date
+from decimal import Decimal
 
-from sqlalchemy import CheckConstraint, Date, Float, String, event
+from sqlalchemy import CheckConstraint, Date, Numeric, String, event
 from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.orm import Mapped, mapped_column
 
 from hub_core.models.base import Base, TimestampMixin
 from fin_hub.attribution import optional_attribution
+from fin_hub.money import currency_code, money, reporting_month
 
 
 class ServiceCost(Base, TimestampMixin):
@@ -29,7 +31,7 @@ class ServiceCost(Base, TimestampMixin):
     service_id: Mapped[str] = mapped_column(String(128), nullable=False, index=True)
     environment: Mapped[str] = mapped_column(String(64), nullable=False, default="production")
     period_month: Mapped[str] = mapped_column(String(7), nullable=False, index=True)
-    amount: Mapped[float] = mapped_column(Float, nullable=False, default=0.0)
+    amount: Mapped[Decimal] = mapped_column(Numeric(18, 2), nullable=False, default=Decimal("0"))
     currency: Mapped[str] = mapped_column(String(3), nullable=False, default="EUR")
     source: Mapped[str] = mapped_column(String(32), nullable=False)
     incurred_on: Mapped[date | None] = mapped_column(Date, nullable=True)
@@ -50,3 +52,9 @@ def _normalize_service_cost_attribution(_mapper, _connection, target: ServiceCos
     if target.cost_attribution_key not in (None, expected_key):
         raise ValueError("cost_attribution_key does not match its attribution dimensions")
     target.cost_attribution_key = expected_key
+    target.period_month = reporting_month(target.period_month)
+    target.amount = money(target.amount)
+    target.currency = currency_code(target.currency)
+    target.source = target.source.strip()
+    if not target.source:
+        raise ValueError("source is required")

src/fin_hub/money.py

@@ -0,0 +1,81 @@
+"""Shared financial-domain validation and decimal arithmetic."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import date
+from decimal import Decimal, InvalidOperation, ROUND_HALF_EVEN
+
+from fin_hub.attribution import ClientAttribution
+
+MONEY_QUANTUM = Decimal("0.01")
+MONEY_ROUNDING = ROUND_HALF_EVEN
+
+
+def money(value: Decimal | str | int | float, *, non_negative: bool = False) -> Decimal:
+    try:
+        normalized = Decimal(str(value)).quantize(MONEY_QUANTUM, rounding=MONEY_ROUNDING)
+    except (InvalidOperation, ValueError) as exc:
+        raise ValueError("money must be a finite decimal value") from exc
+    if not normalized.is_finite():
+        raise ValueError("money must be a finite decimal value")
+    if non_negative and normalized < 0:
+        raise ValueError("money cannot be negative")
+    return normalized
+
+
+def money_minor(value: Decimal | str | int | float, *, non_negative: bool = False) -> int:
+    return int(money(value, non_negative=non_negative) / MONEY_QUANTUM)
+
+
+def minor_money(value: int) -> Decimal:
+    return (Decimal(value) * MONEY_QUANTUM).quantize(MONEY_QUANTUM)
+
+
+def currency_code(value: str) -> str:
+    normalized = value.strip().upper()
+    if len(normalized) != 3 or not normalized.isascii() or not normalized.isalpha():
+        raise ValueError("currency must be a three-letter ASCII code")
+    return normalized
+
+
+def reporting_month(value: str) -> str:
+    if len(value) != 7 or value[4] != "-":
+        raise ValueError("period_month must use YYYY-MM")
+    try:
+        date.fromisoformat(f"{value}-01")
+    except ValueError as exc:
+        raise ValueError("period_month must use YYYY-MM") from exc
+    return value
+
+
+@dataclass(frozen=True)
+class EngagementPriceTerms:
+    attribution: ClientAttribution
+    period_month: str
+    amount: Decimal
+    currency: str
+    source: str
+
+    @classmethod
+    def validate(
+        cls,
+        *,
+        client_id: str,
+        application_id: str,
+        app_instance_id: str,
+        period_month: str,
+        amount: Decimal | str | int | float,
+        currency: str,
+        source: str,
+    ) -> "EngagementPriceTerms":
+        normalized_source = source.strip()
+        if not normalized_source:
+            raise ValueError("source is required")
+        return cls(
+            attribution=ClientAttribution(client_id, application_id, app_instance_id),
+            period_month=reporting_month(period_month),
+            amount=money(amount, non_negative=True),
+            currency=currency_code(currency),
+            source=normalized_source,
+        )

src/fin_hub/schemas/__init__.py

@@ -1,5 +1,23 @@
 """Pydantic schemas for fin-hub HTTP surfaces."""
 
 from fin_hub.schemas.runway import MonthlyBurnRead, RunwayAlertRead, RunwaySummaryRead
+from fin_hub.schemas.exchange import (
+    AllocationEvidence,
+    BookedCostEvidence,
+    CommitmentCandidate,
+    ForecastEvidence,
+    OptimizationEvidence,
+    UsageObservation,
+)
 
-__all__ = ["MonthlyBurnRead", "RunwayAlertRead", "RunwaySummaryRead"]
+__all__ = [
+    "AllocationEvidence",
+    "BookedCostEvidence",
+    "CommitmentCandidate",
+    "ForecastEvidence",
+    "MonthlyBurnRead",
+    "OptimizationEvidence",
+    "RunwayAlertRead",
+    "RunwaySummaryRead",
+    "UsageObservation",
+]

src/fin_hub/schemas/exchange.py

@@ -0,0 +1,237 @@
+"""Versioned fin-hub/resource-control exchange records."""
+
+from __future__ import annotations
+
+from datetime import date, datetime
+from decimal import Decimal
+from typing import Annotated, Literal
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+from fin_hub.money import currency_code, money, reporting_month
+
+
+class ExchangeRecord(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    schema_version: Literal["0.1"] = "0.1"
+    record_id: str = Field(min_length=1, max_length=256)
+    revision_of: str | None = None
+    resource_id: str | None = None
+    service_id: str | None = None
+    workload_id: str | None = None
+    tenant_id: str | None = None
+    environment: str | None = None
+    cost_attribution_key: str | None = None
+    period_start: date
+    period_end: date
+    source_evidence: list[str]
+    created_at: datetime
+
+    @model_validator(mode="after")
+    def validate_period(self):
+        if self.period_end < self.period_start:
+            raise ValueError("period_end cannot precede period_start")
+        return self
+
+
+class MonetaryRecord(ExchangeRecord):
+    currency: str
+
+    @field_validator("currency")
+    @classmethod
+    def validate_currency(cls, value: str) -> str:
+        return currency_code(value)
+
+class BookedCostEvidence(BaseModel):
+    """Authoritative fin-hub fact; amounts serialize as decimal strings."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    schema_version: Literal["0.1"] = "0.1"
+    record_type: Literal["booked_cost"] = "booked_cost"
+    financial_fact_id: str = Field(min_length=1, max_length=256)
+    correction_of: str | None = None
+    adjustment_kind: Literal["charge", "credit", "refund", "reversal", "correction"]
+    source_type: str = Field(min_length=1)
+    source_document_id: str = Field(min_length=1)
+    source_line_id: str = Field(min_length=1)
+    content_fingerprint: str = Field(min_length=1)
+    provider: str = Field(min_length=1)
+    provider_account_ref: str | None = None
+    accounting_period: str
+    service_period_start: date | None = None
+    service_period_end: date | None = None
+    currency: str
+    net_amount: Decimal = Decimal("0.00")
+    discount_amount: Decimal = Decimal("0.00")
+    tax_status: Literal["known", "unknown", "not_applicable"]
+    tax_amount: Decimal | None = None
+    gross_amount: Decimal
+    adjustment_amount: Decimal = Decimal("0.00")
+    effective_amount: Decimal
+    resource_id: str | None = None
+    service_id: str | None = None
+    workload_id: str | None = None
+    tenant_id: str | None = None
+    environment: str | None = None
+    cost_attribution_key: str | None = None
+    source_evidence_ref: str = Field(min_length=1)
+    recorded_at: datetime
+
+    @field_validator(
+        "net_amount", "discount_amount", "gross_amount", mode="before"
+    )
+    @classmethod
+    def validate_nonnegative_money(cls, value):
+        return money(value, non_negative=True)
+
+    @field_validator("tax_amount", "adjustment_amount", "effective_amount", mode="before")
+    @classmethod
+    def validate_signed_money(cls, value):
+        return None if value is None else money(value)
+
+    @field_validator("currency")
+    @classmethod
+    def validate_currency(cls, value: str) -> str:
+        return currency_code(value)
+
+    @field_validator("accounting_period")
+    @classmethod
+    def validate_accounting_period(cls, value: str) -> str:
+        return reporting_month(value)
+
+    @model_validator(mode="after")
+    def validate_accounting_relationships(self):
+        if self.tax_status == "unknown" and self.tax_amount is not None:
+            raise ValueError("tax_amount must be null when tax_status is unknown")
+        if self.tax_status != "unknown" and self.tax_amount is None:
+            raise ValueError("tax_amount is required when tax status is known")
+        tax = self.tax_amount or Decimal("0.00")
+        if self.tax_status != "unknown":
+            expected_gross = money(self.net_amount - self.discount_amount + tax)
+            if self.gross_amount != expected_gross:
+                raise ValueError("gross_amount must equal net - discount + tax")
+        if self.effective_amount != money(self.gross_amount + self.adjustment_amount):
+            raise ValueError("effective_amount must equal gross + adjustment")
+        if self.adjustment_kind != "charge" and self.correction_of is None:
+            raise ValueError("credit/refund/reversal/correction requires correction_of")
+        if self.adjustment_kind in {"credit", "refund", "reversal"} and self.adjustment_amount >= 0:
+            raise ValueError("credit/refund/reversal adjustment_amount must be negative")
+        if (
+            self.service_period_start is not None
+            and self.service_period_end is not None
+            and self.service_period_end < self.service_period_start
+        ):
+            raise ValueError("service period end cannot precede start")
+        return self
+
+
+class CostBreakdown(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    infrastructure: Decimal = Decimal("0.00")
+    internal_labor: Decimal = Decimal("0.00")
+    external_services: Decimal = Decimal("0.00")
+    setup: Decimal = Decimal("0.00")
+    other: Decimal = Decimal("0.00")
+
+    @field_validator("*", mode="before")
+    @classmethod
+    def validate_money(cls, value):
+        return money(value, non_negative=True)
+
+    @property
+    def total(self) -> Decimal:
+        return money(
+            sum(
+                (getattr(self, name) for name in self.__class__.model_fields),
+                Decimal("0"),
+            )
+        )
+
+
+class ForecastEvidence(MonetaryRecord):
+    record_type: Literal["forecast"] = "forecast"
+    scenario: Literal["low", "base", "high"]
+    forecast_version: str = Field(min_length=1)
+    costs: CostBreakdown
+    uncertainty: str | None = None
+    assumptions: list[str]
+
+
+class UsageMeasure(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    name: str = Field(min_length=1)
+    value: Decimal
+    unit: str = Field(min_length=1)
+
+
+class UsageObservation(ExchangeRecord):
+    record_type: Literal["usage_observation"] = "usage_observation"
+    measures: list[UsageMeasure]
+
+
+class AllocationShare(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+    target_key: str = Field(min_length=1)
+    share: Decimal = Field(ge=0, le=1)
+
+
+class AllocationEvidence(MonetaryRecord):
+    record_type: Literal["allocation"] = "allocation"
+    financial_fact_ids: list[str] = Field(min_length=1)
+    method: str = Field(min_length=1)
+    allocated_amount: Decimal
+    shares: list[AllocationShare]
+    residual_share: Decimal = Field(ge=0, le=1)
+
+    @field_validator("allocated_amount", mode="before")
+    @classmethod
+    def validate_amount(cls, value):
+        return money(value, non_negative=True)
+
+    @model_validator(mode="after")
+    def validate_shares(self):
+        total = sum((share.share for share in self.shares), Decimal("0"))
+        if total + self.residual_share != Decimal("1"):
+            raise ValueError("allocation shares plus residual_share must equal 1")
+        return self
+
+
+class OptimizationEvidence(MonetaryRecord):
+    record_type: Literal["optimization"] = "optimization"
+    baseline: CostBreakdown
+    alternative: CostBreakdown
+    one_time_cost: Decimal
+    expected_period_savings: Decimal
+    assumptions: list[str]
+
+    @field_validator("one_time_cost", "expected_period_savings", mode="before")
+    @classmethod
+    def validate_amount(cls, value):
+        return money(value, non_negative=True)
+
+
+class CommitmentCandidate(MonetaryRecord):
+    record_type: Literal["commitment_candidate"] = "commitment_candidate"
+    setup_cost: Decimal
+    recurring_cost: Decimal
+    cadence: Literal["monthly", "quarterly", "annual", "one_time"]
+    term_start: date
+    term_end: date | None = None
+    approval_status: Literal["candidate"] = "candidate"
+
+    @field_validator("setup_cost", "recurring_cost", mode="before")
+    @classmethod
+    def validate_amount(cls, value):
+        return money(value, non_negative=True)
+
+
+PlanningEvidence = Annotated[
+    ForecastEvidence
+    | UsageObservation
+    | AllocationEvidence
+    | OptimizationEvidence
+    | CommitmentCandidate,
+    Field(discriminator="record_type"),
+]

src/fin_hub/services/exchange.py

@@ -0,0 +1,203 @@
+"""Persistence boundary for non-booked resource-control evidence."""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+from calendar import monthrange
+from datetime import date
+from pathlib import Path
+
+from pydantic import TypeAdapter
+
+from fin_hub.money import minor_money, money
+from fin_hub.schemas.exchange import BookedCostEvidence, PlanningEvidence
+from fin_hub.services.ledger import _connect, default_ledger_path
+
+_PLANNING_ADAPTER = TypeAdapter(PlanningEvidence)
+
+
+def _ensure_planning_schema(conn: sqlite3.Connection) -> None:
+    conn.execute(
+        """
+        CREATE TABLE IF NOT EXISTS planning_evidence (
+            record_id TEXT PRIMARY KEY,
+            record_type TEXT NOT NULL,
+            revision_of TEXT REFERENCES planning_evidence(record_id),
+            payload_json TEXT NOT NULL,
+            is_current INTEGER NOT NULL DEFAULT 1,
+            received_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
+        )
+        """
+    )
+    conn.execute(
+        "CREATE INDEX IF NOT EXISTS ix_planning_evidence_type_current "
+        "ON planning_evidence (record_type, is_current)"
+    )
+    conn.commit()
+
+
+def ingest_planning_evidence(
+    payload: dict,
+    *,
+    ledger_path: Path | None = None,
+) -> PlanningEvidence:
+    """Validate and idempotently retain planning evidence outside booked spend."""
+
+    record = _PLANNING_ADAPTER.validate_python(payload)
+    canonical = record.model_dump_json()
+    ledger = ledger_path or default_ledger_path()
+    with _connect(ledger) as conn:
+        _ensure_planning_schema(conn)
+        conn.execute("BEGIN IMMEDIATE")
+        existing = conn.execute(
+            "SELECT payload_json FROM planning_evidence WHERE record_id = ?",
+            (record.record_id,),
+        ).fetchone()
+        if existing is not None:
+            if json.loads(existing["payload_json"]) != json.loads(canonical):
+                raise ValueError("record_id already exists with different content")
+            return record
+        if record.revision_of is not None:
+            predecessor = conn.execute(
+                "SELECT record_type, is_current FROM planning_evidence WHERE record_id = ?",
+                (record.revision_of,),
+            ).fetchone()
+            if predecessor is None or predecessor["record_type"] != record.record_type:
+                raise ValueError("revision_of must reference an existing record of the same type")
+            if predecessor["is_current"] != 1:
+                raise ValueError("revision_of must reference the current record")
+            conn.execute(
+                "UPDATE planning_evidence SET is_current = 0 WHERE record_id = ?",
+                (record.revision_of,),
+            )
+        conn.execute(
+            "INSERT INTO planning_evidence "
+            "(record_id, record_type, revision_of, payload_json, is_current) "
+            "VALUES (?, ?, ?, ?, 1)",
+            (record.record_id, record.record_type, record.revision_of, canonical),
+        )
+        conn.commit()
+    return record
+
+
+def booked_cost_projection(*, ledger_path: Path | None = None) -> list[BookedCostEvidence]:
+    """Project current authoritative facts without exposing raw invoice content."""
+
+    ledger = ledger_path or default_ledger_path()
+    with _connect(ledger) as conn:
+        rows = conn.execute(
+            "SELECT * FROM ledger_entries WHERE is_current = 1 ORDER BY id"
+        ).fetchall()
+    projected: list[BookedCostEvidence] = []
+    for row in rows:
+        amount = minor_money(int(row["amount_minor"]))
+        adjustment_kind = row["adjustment_kind"]
+        gross_amount = amount
+        adjustment_amount = minor_money(0)
+        if adjustment_kind == "reversal":
+            with _connect(ledger) as conn:
+                predecessor = conn.execute(
+                    "SELECT amount_minor FROM ledger_entries WHERE financial_fact_id = ?",
+                    (row["correction_of"],),
+                ).fetchone()
+            if predecessor is None:
+                raise ValueError("reversal predecessor is missing")
+            gross_amount = minor_money(int(predecessor["amount_minor"]))
+            adjustment_amount = -gross_amount
+        projected.append(
+            BookedCostEvidence(
+                financial_fact_id=row["financial_fact_id"],
+                correction_of=row["correction_of"],
+                adjustment_kind=adjustment_kind,
+                source_type=row["source_type"],
+                source_document_id=row["source_document_id"],
+                source_line_id=row["source_line_id"],
+                content_fingerprint=row["content_fingerprint"],
+                provider=row["source_type"],
+                accounting_period=row["period_month"],
+                service_period_start=row["incurred_on"],
+                service_period_end=row["incurred_on"],
+                currency=row["currency"],
+                net_amount="0.00",
+                discount_amount="0.00",
+                tax_status="unknown",
+                tax_amount=None,
+                gross_amount=gross_amount,
+                adjustment_amount=adjustment_amount,
+                effective_amount=amount,
+                service_id=row["category"],
+                environment=row["environment"],
+                cost_attribution_key=row["cost_attribution_key"],
+                source_evidence_ref=row["correction_source"] or row["source_document_id"],
+                recorded_at=row["imported_at"],
+            )
+        )
+    return projected
+
+
+def ingest_resource_forecast(
+    payload: dict,
+    *,
+    resource_id: str,
+    source_evidence_ref: str,
+    ledger_path: Path | None = None,
+) -> list[PlanningEvidence]:
+    """Adapt resource-control's v0.1 monthly forecast into typed evidence."""
+
+    if payload.get("record_type") != "forecast":
+        raise ValueError("resource forecast payload must have record_type=forecast")
+    created_at = payload["created_at"]
+    version = f"{payload['provider_id']}:{created_at}"
+    records: list[PlanningEvidence] = []
+    for row in payload["rows"]:
+        expected_total = money(row["infrastructure_eur"]) + money(row["internal_labor_eur"])
+        if money(row["total_eur"]) != expected_total:
+            raise ValueError("resource forecast total_eur does not match its cost breakdown")
+        year, month = (int(part) for part in row["period"].split("-"))
+        period_start = date(year, month, 1)
+        period_end = date(year, month, monthrange(year, month)[1])
+        record = ingest_planning_evidence(
+            {
+                "schema_version": "0.1",
+                "record_type": "forecast",
+                "record_id": (
+                    f"forecast:{payload['provider_id']}:"
+                    f"{payload['cost_attribution_key']}:{row['period']}:{created_at}"
+                ),
+                "revision_of": payload.get("forecast_ref"),
+                "resource_id": resource_id,
+                "service_id": payload["provider_id"],
+                "workload_id": payload["workload"],
+                "tenant_id": None,
+                "environment": "production",
+                "cost_attribution_key": payload["cost_attribution_key"],
+                "period_start": period_start.isoformat(),
+                "period_end": period_end.isoformat(),
+                "currency": "EUR",
+                "source_evidence": [source_evidence_ref, *row.get("evidence", [])],
+                "created_at": created_at,
+                "scenario": payload.get("scenario") or "base",
+                "forecast_version": version,
+                "costs": {
+                    "infrastructure": row["infrastructure_eur"],
+                    "internal_labor": row["internal_labor_eur"],
+                    "external_services": 0,
+                    "setup": 0,
+                    "other": 0,
+                },
+                "uncertainty": None,
+                "assumptions": [
+                    f"database_gb={row['database_gb']}",
+                    f"stored_gb={row['stored_gb']}",
+                    f"wal_gb={row['wal_gb']}",
+                    f"restore_egress_gb={row['restore_egress_gb']}",
+                    f"write_requests={row['write_requests']}",
+                    f"read_requests={row['read_requests']}",
+                    f"internal_labor_hours={row['internal_labor_hours']}",
+                ],
+            },
+            ledger_path=ledger_path,
+        )
+        records.append(record)
+    return records

src/fin_hub/services/ledger.py

@@ -3,15 +3,25 @@
 from __future__ import annotations
 
 import json
+import hashlib
 import sqlite3
 import uuid
 from dataclasses import asdict, dataclass
 from datetime import date, datetime, timezone
+from decimal import Decimal
 from pathlib import Path
 
 from fin_hub.ingest.anthropic import parse_anthropic_billing_csv
 from fin_hub.ingest.cloud import parse_cloud_cost_csv
 from fin_hub.ingest.hosteurope import parse_hosteurope_csv
+from fin_hub.money import (
+    EngagementPriceTerms,
+    currency_code,
+    minor_money,
+    money,
+    money_minor,
+    reporting_month,
+)
 
 DEFAULT_LEDGER_PATH = Path(".fin-hub/ledger.db")
 
@@ -21,16 +31,26 @@ class LedgerEntry:
     source_type: str
     category: str
     label: str
-    amount: float
+    amount: Decimal
     currency: str
     period_month: str
     incurred_on: date | None
     source_path: str
+    environment: str = "production"
     client_id: str | None = None
     application_id: str | None = None
     app_instance_id: str | None = None
     cost_attribution_key: str | None = None
 
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "amount", money(self.amount))
+        object.__setattr__(self, "currency", currency_code(self.currency))
+        object.__setattr__(self, "period_month", reporting_month(self.period_month))
+        normalized_environment = self.environment.strip()
+        if not normalized_environment:
+            raise ValueError("environment is required")
+        object.__setattr__(self, "environment", normalized_environment)
+
 
 @dataclass(frozen=True)
 class MonthlyRollup:
@@ -62,7 +82,7 @@ class EngagementPriceRecord:
     app_instance_id: str
     cost_attribution_key: str
     period_month: str
-    amount: float
+    amount: Decimal
     currency: str
     source: str
     revision_of: str | None
@@ -77,9 +97,9 @@ class ClientMargin:
     app_instance_id: str
     period_month: str
     currency: str
-    revenue: float
-    attributed_cost: float
-    margin: float
+    revenue: Decimal
+    attributed_cost: Decimal
+    margin: Decimal
     price_id: str
 
     def as_dict(self) -> dict:
@@ -136,14 +156,14 @@ def _ensure_schema(conn: sqlite3.Connection) -> None:
             cost_attribution_key TEXT NOT NULL,
             period_month TEXT NOT NULL,
             amount REAL NOT NULL,
+            amount_minor INTEGER,
             currency TEXT NOT NULL,
             source TEXT NOT NULL,
             revision_of TEXT REFERENCES engagement_prices(id),
+            is_current INTEGER NOT NULL DEFAULT 1,
             recorded_at TEXT NOT NULL
         );
 
-        CREATE INDEX IF NOT EXISTS ix_engagement_prices_basis
-        ON engagement_prices (cost_attribution_key, period_month, currency);
         """
     )
     columns = {row["name"] for row in conn.execute("PRAGMA table_info(ledger_entries)")}
@@ -152,15 +172,99 @@ def _ensure_schema(conn: sqlite3.Connection) -> None:
         "application_id",
         "app_instance_id",
         "cost_attribution_key",
+        "environment",
     ):
         if name not in columns:
             conn.execute(f"ALTER TABLE ledger_entries ADD COLUMN {name} TEXT")
+    entry_columns = {row["name"] for row in conn.execute("PRAGMA table_info(ledger_entries)")}
+    if "amount_minor" not in entry_columns:
+        conn.execute("ALTER TABLE ledger_entries ADD COLUMN amount_minor INTEGER")
+    for name, definition in (
+        ("financial_fact_id", "TEXT"),
+        ("source_document_id", "TEXT"),
+        ("source_line_id", "TEXT"),
+        ("content_fingerprint", "TEXT"),
+        ("correction_of", "TEXT"),
+        ("adjustment_kind", "TEXT NOT NULL DEFAULT 'charge'"),
+        ("correction_source", "TEXT"),
+        ("is_current", "INTEGER NOT NULL DEFAULT 1"),
+    ):
+        if name not in entry_columns:
+            conn.execute(f"ALTER TABLE ledger_entries ADD COLUMN {name} {definition}")
+    conn.execute(
+        "UPDATE ledger_entries SET amount_minor = ROUND(amount * 100) "
+        "WHERE amount_minor IS NULL"
+    )
+    conn.execute(
+        "UPDATE ledger_entries SET financial_fact_id = 'legacy:' || id, "
+        "source_document_id = 'legacy:' || source_path, "
+        "source_line_id = 'legacy:' || id, "
+        "content_fingerprint = 'legacy:' || id "
+        "WHERE financial_fact_id IS NULL"
+    )
+    conn.execute(
+        "CREATE UNIQUE INDEX IF NOT EXISTS uq_ledger_financial_fact_id "
+        "ON ledger_entries (financial_fact_id)"
+    )
+    conn.execute(
+        "CREATE UNIQUE INDEX IF NOT EXISTS uq_ledger_current_source_line "
+        "ON ledger_entries (source_line_id) WHERE is_current = 1"
+    )
+    price_columns = {
+        row["name"] for row in conn.execute("PRAGMA table_info(engagement_prices)")
+    }
+    if "amount_minor" not in price_columns:
+        conn.execute("ALTER TABLE engagement_prices ADD COLUMN amount_minor INTEGER")
+    if "is_current" not in price_columns:
+        conn.execute(
+            "ALTER TABLE engagement_prices ADD COLUMN is_current INTEGER NOT NULL DEFAULT 1"
+        )
+    conn.execute(
+        "UPDATE engagement_prices SET amount_minor = ROUND(amount * 100) "
+        "WHERE amount_minor IS NULL"
+    )
+    conn.execute(
+        "UPDATE engagement_prices SET is_current = 0 WHERE id IN ("
+        "SELECT revision_of FROM engagement_prices WHERE revision_of IS NOT NULL)"
+    )
+    conn.execute(
+        "CREATE INDEX IF NOT EXISTS ix_engagement_prices_basis "
+        "ON engagement_prices (cost_attribution_key, period_month, currency)"
+    )
+    conn.execute(
+        "CREATE UNIQUE INDEX IF NOT EXISTS uq_engagement_prices_current_basis "
+        "ON engagement_prices (cost_attribution_key, period_month, currency) "
+        "WHERE is_current = 1"
+    )
+    conn.commit()
 
 
 def _utc_now() -> str:
     return datetime.now(timezone.utc).isoformat()
 
 
+def _digest(*parts: object) -> str:
+    payload = "\x1f".join("" if part is None else str(part) for part in parts)
+    return hashlib.sha256(payload.encode("utf-8")).hexdigest()
+
+
+def _entry_business_key(entry: LedgerEntry) -> str:
+    return _digest(
+        entry.source_type,
+        entry.category,
+        entry.label,
+        entry.currency,
+        entry.period_month,
+        entry.incurred_on.isoformat() if entry.incurred_on else None,
+        entry.environment,
+        entry.cost_attribution_key,
+    )
+
+
+def _entry_fingerprint(entry: LedgerEntry) -> str:
+    return _digest(_entry_business_key(entry), money_minor(entry.amount))
+
+
 def set_opening_balance(path: Path, balance: float, *, currency: str = "EUR") -> None:
     with _connect(path) as conn:
         conn.execute(
@@ -234,6 +338,7 @@ def _entries_from_hosteurope(path: Path) -> list[LedgerEntry]:
             period_month=row.period_month,
             incurred_on=row.incurred_on,
             source_path=resolved,
+            environment=row.environment,
             client_id=row.client_id,
             application_id=row.application_id,
             app_instance_id=row.app_instance_id,
@@ -279,20 +384,50 @@ def import_csv(
                     skipped=True,
                 )
 
-        for entry in entries:
+        business_keys = [_entry_business_key(entry) for entry in entries]
+        source_document_id = f"{source_type}:{_digest(*sorted(business_keys))}"
+        occurrences: dict[str, int] = {}
+        rows_imported = 0
+        for entry, business_key in zip(entries, business_keys, strict=True):
+            occurrence = occurrences.get(business_key, 0) + 1
+            occurrences[business_key] = occurrence
+            source_line_id = f"{source_type}:{business_key}:{occurrence}"
+            fingerprint = _entry_fingerprint(entry)
+            current = conn.execute(
+                "SELECT financial_fact_id, content_fingerprint FROM ledger_entries "
+                "WHERE source_line_id = ? AND is_current = 1",
+                (source_line_id,),
+            ).fetchone()
+            if current is not None and current["content_fingerprint"] == fingerprint:
+                continue
+            if current is not None and not force:
+                raise ValueError(
+                    "changed financial fact requires force=True to append an explicit correction: "
+                    f"{current['financial_fact_id']}"
+                )
+            correction_of = current["financial_fact_id"] if current is not None else None
+            financial_fact_id = f"fact:{_digest(source_line_id, fingerprint)}"
+            if current is not None:
+                conn.execute(
+                    "UPDATE ledger_entries SET is_current = 0 WHERE financial_fact_id = ?",
+                    (correction_of,),
+                )
             conn.execute(
                 """
                 INSERT INTO ledger_entries (
                     source_type, category, label, amount, currency,
                     period_month, incurred_on, source_path, imported_at,
-                    client_id, application_id, app_instance_id, cost_attribution_key
-                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    client_id, application_id, app_instance_id, cost_attribution_key,
+                    environment, amount_minor, financial_fact_id, source_document_id,
+                    source_line_id, content_fingerprint, correction_of,
+                    adjustment_kind, correction_source, is_current
+                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1)
                 """,
                 (
                     entry.source_type,
                     entry.category,
                     entry.label,
-                    entry.amount,
+                    float(entry.amount),
                     entry.currency,
                     entry.period_month,
                     entry.incurred_on.isoformat() if entry.incurred_on else None,
@@ -302,33 +437,108 @@ def import_csv(
                     entry.application_id,
                     entry.app_instance_id,
                     entry.cost_attribution_key,
+                    entry.environment,
+                    money_minor(entry.amount),
+                    financial_fact_id,
+                    source_document_id,
+                    source_line_id,
+                    fingerprint,
+                    correction_of,
+                    "correction" if correction_of else "charge",
+                    entry.source_path if correction_of else None,
                 ),
             )
+            rows_imported += 1
         conn.execute(
             """
             INSERT OR REPLACE INTO ledger_imports (
                 source_path, source_mtime, source_type, row_count, imported_at
             ) VALUES (?, ?, ?, ?, ?)
             """,
-            (resolved, mtime, source_type, len(entries), imported_at),
+            (resolved, mtime, source_type, rows_imported, imported_at),
         )
         conn.commit()
 
     return ImportResult(
         source_type=source_type,
         source_path=resolved,
-        rows_imported=len(entries),
-        skipped=False,
+        rows_imported=rows_imported,
+        skipped=rows_imported == 0,
     )
 
 
+def reverse_financial_fact(
+    financial_fact_id: str,
+    *,
+    source: str,
+    ledger_path: Path | None = None,
+) -> str:
+    """Append a zero-effective reversal and retain the reversed predecessor."""
+
+    normalized_source = source.strip()
+    if not normalized_source:
+        raise ValueError("source is required")
+    ledger = ledger_path or default_ledger_path()
+    with _connect(ledger) as conn:
+        conn.execute("BEGIN IMMEDIATE")
+        current = conn.execute(
+            "SELECT * FROM ledger_entries WHERE financial_fact_id = ? AND is_current = 1",
+            (financial_fact_id,),
+        ).fetchone()
+        if current is None:
+            raise ValueError("financial_fact_id must reference a current fact")
+        fingerprint = _digest(current["content_fingerprint"], "reversal", normalized_source)
+        reversal_id = f"fact:{_digest(current['source_line_id'], fingerprint)}"
+        conn.execute(
+            "UPDATE ledger_entries SET is_current = 0 WHERE financial_fact_id = ?",
+            (financial_fact_id,),
+        )
+        conn.execute(
+            """
+            INSERT INTO ledger_entries (
+                source_type, category, label, amount, currency, period_month,
+                incurred_on, source_path, imported_at, client_id, application_id,
+                app_instance_id, cost_attribution_key, environment, amount_minor,
+                financial_fact_id, source_document_id, source_line_id,
+                content_fingerprint, correction_of, adjustment_kind,
+                correction_source, is_current
+            ) VALUES (?, ?, ?, 0, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 0, ?, ?, ?, ?, ?, 'reversal', ?, 1)
+            """,
+            (
+                current["source_type"],
+                current["category"],
+                current["label"],
+                current["currency"],
+                current["period_month"],
+                current["incurred_on"],
+                current["source_path"],
+                _utc_now(),
+                current["client_id"],
+                current["application_id"],
+                current["app_instance_id"],
+                current["cost_attribution_key"],
+                current["environment"],
+                reversal_id,
+                current["source_document_id"],
+                current["source_line_id"],
+                fingerprint,
+                financial_fact_id,
+                normalized_source,
+            ),
+        )
+        conn.commit()
+    return reversal_id
+
+
 def monthly_summary(*, ledger_path: Path | None = None) -> list[MonthlyRollup]:
     ledger = ledger_path or default_ledger_path()
     with _connect(ledger) as conn:
         rows = conn.execute(
             """
-            SELECT period_month, currency, SUM(amount) AS total, COUNT(*) AS entry_count
+            SELECT period_month, currency, SUM(amount_minor) AS total_minor,
+                   COUNT(*) AS entry_count
             FROM ledger_entries
+            WHERE is_current = 1
             GROUP BY period_month, currency
             ORDER BY period_month, currency
             """
@@ -337,7 +547,7 @@ def monthly_summary(*, ledger_path: Path | None = None) -> list[MonthlyRollup]:
         MonthlyRollup(
             period_month=row["period_month"],
             currency=row["currency"],
-            total=float(row["total"]),
+            total=float(minor_money(int(row["total_minor"]))),
             entry_count=int(row["entry_count"]),
         )
         for row in rows
@@ -367,31 +577,25 @@ def record_engagement_price(
     price_id: str | None = None,
     revision_of: str | None = None,
 ) -> EngagementPriceRecord:
-    from fin_hub.attribution import ClientAttribution
-
-    attribution = ClientAttribution(client_id, application_id, app_instance_id)
-    if len(period_month) != 7 or period_month[4] != "-":
-        raise ValueError("period_month must use YYYY-MM")
-    try:
-        date.fromisoformat(f"{period_month}-01")
-    except ValueError as exc:
-        raise ValueError("period_month must use YYYY-MM") from exc
-    normalized_currency = currency.strip().upper()
-    if len(normalized_currency) != 3 or not normalized_currency.isalpha():
-        raise ValueError("currency must be a three-letter code")
-    if amount < 0:
-        raise ValueError("engagement price amount cannot be negative")
-    if not source.strip():
-        raise ValueError("source is required")
+    terms = EngagementPriceTerms.validate(
+        client_id=client_id,
+        application_id=application_id,
+        app_instance_id=app_instance_id,
+        period_month=period_month,
+        amount=amount,
+        currency=currency,
+        source=source,
+    )
 
     ledger = ledger_path or default_ledger_path()
     identifier = price_id or str(uuid.uuid4())
     recorded_at = _utc_now()
     with _connect(ledger) as conn:
-        basis = (attribution.key, period_month, normalized_currency)
+        conn.execute("BEGIN IMMEDIATE")
+        basis = (terms.attribution.key, terms.period_month, terms.currency)
         current = conn.execute(
             "SELECT id FROM engagement_prices WHERE cost_attribution_key = ? "
-            "AND period_month = ? AND currency = ? "
+            "AND period_month = ? AND currency = ? AND is_current = 1 "
             "ORDER BY recorded_at DESC, rowid DESC LIMIT 1",
             basis,
         ).fetchone()
@@ -399,21 +603,28 @@ def record_engagement_price(
             raise ValueError(f"revision_of must reference current price {current['id']}")
         if current is None and revision_of is not None:
             raise ValueError("revision_of cannot be used without an existing price")
+        if current is not None:
+            conn.execute(
+                "UPDATE engagement_prices SET is_current = 0 WHERE id = ?",
+                (current["id"],),
+            )
         conn.execute(
             "INSERT INTO engagement_prices ("
             "id, client_id, application_id, app_instance_id, cost_attribution_key, "
-            "period_month, amount, currency, source, revision_of, recorded_at"
-            ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            "period_month, amount, amount_minor, currency, source, revision_of, "
+            "is_current, recorded_at"
+            ") VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 1, ?)",
             (
                 identifier,
-                attribution.client_id,
-                attribution.application_id,
-                attribution.app_instance_id,
-                attribution.key,
-                period_month,
-                amount,
-                normalized_currency,
-                source.strip(),
+                terms.attribution.client_id,
+                terms.attribution.application_id,
+                terms.attribution.app_instance_id,
+                terms.attribution.key,
+                terms.period_month,
+                float(terms.amount),
+                money_minor(terms.amount),
+                terms.currency,
+                terms.source,
                 revision_of,
                 recorded_at,
             ),
@@ -421,14 +632,14 @@ def record_engagement_price(
         conn.commit()
     return EngagementPriceRecord(
         id=identifier,
-        client_id=attribution.client_id,
-        application_id=attribution.application_id,
-        app_instance_id=attribution.app_instance_id,
-        cost_attribution_key=attribution.key,
-        period_month=period_month,
-        amount=amount,
-        currency=normalized_currency,
-        source=source.strip(),
+        client_id=terms.attribution.client_id,
+        application_id=terms.attribution.application_id,
+        app_instance_id=terms.attribution.app_instance_id,
+        cost_attribution_key=terms.attribution.key,
+        period_month=terms.period_month,
+        amount=terms.amount,
+        currency=terms.currency,
+        source=terms.source,
         revision_of=revision_of,
         recorded_at=recorded_at,
     )
@@ -440,23 +651,18 @@ def client_margin_report(*, ledger_path: Path | None = None) -> list[ClientMargi
         rows = conn.execute(
             """
             WITH latest_prices AS (
-                SELECT p.*
-                FROM engagement_prices p
-                WHERE NOT EXISTS (
-                    SELECT 1 FROM engagement_prices revision
-                    WHERE revision.revision_of = p.id
-                )
+                SELECT p.* FROM engagement_prices p WHERE p.is_current = 1
             ), attributed_costs AS (
                 SELECT cost_attribution_key, period_month, currency,
-                       SUM(amount) AS attributed_cost
+                       SUM(amount_minor) AS attributed_cost_minor
                 FROM ledger_entries
-                WHERE cost_attribution_key IS NOT NULL
+                WHERE cost_attribution_key IS NOT NULL AND is_current = 1
                 GROUP BY cost_attribution_key, period_month, currency
             )
             SELECT p.id AS price_id, p.client_id, p.application_id,
                    p.app_instance_id, p.cost_attribution_key, p.period_month,
-                   p.currency, p.amount AS revenue,
-                   COALESCE(c.attributed_cost, 0) AS attributed_cost
+                   p.currency, p.amount_minor AS revenue_minor,
+                   COALESCE(c.attributed_cost_minor, 0) AS attributed_cost_minor
             FROM latest_prices p
             LEFT JOIN attributed_costs c
               ON c.cost_attribution_key = p.cost_attribution_key
@@ -473,9 +679,11 @@ def client_margin_report(*, ledger_path: Path | None = None) -> list[ClientMargi
             app_instance_id=row["app_instance_id"],
             period_month=row["period_month"],
             currency=row["currency"],
-            revenue=float(row["revenue"]),
-            attributed_cost=float(row["attributed_cost"]),
-            margin=float(row["revenue"]) - float(row["attributed_cost"]),
+            revenue=minor_money(int(row["revenue_minor"])),
+            attributed_cost=minor_money(int(row["attributed_cost_minor"])),
+            margin=minor_money(
+                int(row["revenue_minor"]) - int(row["attributed_cost_minor"])
+            ),
             price_id=row["price_id"],
         )
         for row in rows