Implement FIN-WP-0006-T01 settlement statement join; mark FIN-WP-0004/0005 blocked

Add SettlementStatement schema and a dedicated settlement_statements store
(ingest_settlement_statement, current_settlement_statements) so
resource-control internal transfer settlements join as a referenced
projection without ever entering the booked-cost ledger. exchange_health
now surfaces a current statement with payment_recognition=unknown as a
settlement_awaiting_payment_join residual instead of booked cost. Adds
CLI commands finhub ledger ingest-settlement / settlements.

FIN-WP-0006-T01 is done; the workplan is finished.

Reviewed all other open workplans for closeable work: FIN-WP-0004's only
remaining task (T05) waits on an external provider fact from
resource-control, and every FIN-WP-0005 task waits on operator/Steuerbüro
confirmation or a non-production DATEV tenant. Neither has actionable
in-repo work right now, so both move to status: blocked.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Assistant: claude-code
Assistant-Model: sonnet
Assistant-Process: 242919@bnt-lap001
Assistant-Session: 286d235a-1654-40ff-b8fd-eee4e2ff9ea6

src/fin_hub/cli.py

@@ -21,8 +21,10 @@ from fin_hub.services.billing import build_billing_basis
 from fin_hub.services.effectiveness import effectiveness_report
 from fin_hub.services.evaluate import evaluate_runway
 from fin_hub.services.exchange import (
+    current_settlement_statements,
     ingest_resource_forecast,
     ingest_resource_usage,
+    ingest_settlement_statement,
     reconcile_resource_period,
 )
 from fin_hub.services.reporting_allocation import (
@@ -304,6 +306,28 @@ def _cmd_ledger_reconcile_resource(args: argparse.Namespace) -> int:
     return 0
 
 
+def _cmd_ledger_ingest_settlement(args: argparse.Namespace) -> int:
+    payload = json.loads(Path(args.path).read_text(encoding="utf-8"))
+    statement = ingest_settlement_statement(
+        payload,
+        source_evidence_ref=args.source,
+        ledger_path=_ledger_path(args),
+    )
+    print(json.dumps(json.loads(statement.model_dump_json()), indent=2))
+    return 0
+
+
+def _cmd_ledger_settlements(args: argparse.Namespace) -> int:
+    statements = current_settlement_statements(ledger_path=_ledger_path(args))
+    print(
+        json.dumps(
+            [json.loads(statement.model_dump_json()) for statement in statements],
+            indent=2,
+        )
+    )
+    return 0
+
+
 def _cmd_ledger_allocations(args: argparse.Namespace) -> int:
     reports = shared_cost_allocations(ledger_path=_ledger_path(args))
     print(json.dumps([report.as_dict() for report in reports], indent=2, default=str))
@@ -442,6 +466,21 @@ def build_parser() -> argparse.ArgumentParser:
     ledger_reconcile.add_argument("--ledger", help="Ledger database path")
     ledger_reconcile.set_defaults(func=_cmd_ledger_reconcile_resource)
 
+    ledger_ingest_settlement = ledger_sub.add_parser(
+        "ingest-settlement",
+        help="Join a resource-control settlement statement (not booked spend)",
+    )
+    ledger_ingest_settlement.add_argument("path")
+    ledger_ingest_settlement.add_argument("--source", required=True)
+    ledger_ingest_settlement.add_argument("--ledger", help="Ledger database path")
+    ledger_ingest_settlement.set_defaults(func=_cmd_ledger_ingest_settlement)
+
+    ledger_settlements = ledger_sub.add_parser(
+        "settlements", help="List current settlement statement projections"
+    )
+    ledger_settlements.add_argument("--ledger", help="Ledger database path")
+    ledger_settlements.set_defaults(func=_cmd_ledger_settlements)
+
     ledger_summary = ledger_sub.add_parser("summary", help="Monthly rollup summary from ledger")
     ledger_summary.add_argument("--ledger", help="Ledger database path (default: .fin-hub/ledger.db)")
     ledger_summary.set_defaults(func=_cmd_ledger_summary)

src/fin_hub/schemas/exchange.py

@@ -294,6 +294,78 @@ class FinancialConstraintSignal(BaseModel):
         return self
 
 
+class SettlementStatement(BaseModel):
+    """Inbound resource-control internal transfer settlement (not booked spend)."""
+
+    model_config = ConfigDict(extra="forbid")
+
+    schema_version: Literal["0.1"] = "0.1"
+    record_type: Literal["settlement_statement"] = "settlement_statement"
+    terms_version: str = Field(min_length=1)
+    financial_entity_id: str = Field(pattern=r"^entity:[a-z0-9]+$")
+    procuring_entity_id: str = Field(pattern=r"^entity:[a-z0-9]+$")
+    period: str
+    statement_date: date
+    due_date: date
+    currency: Literal["EUR"] = "EUR"
+    line_items: list[dict]
+    known_delivered_cost_eur: Decimal
+    known_transfer_price_eur: Decimal
+    unknown_cost_remainder: bool
+    prior_outstanding_eur: Decimal
+    recognized_payments_eur: Decimal | None = None
+    payment_recognition: Literal["unknown", "recognized"]
+    interest_eur: Decimal
+    new_transfer_charges_eur: Decimal
+    outstanding_eur: Decimal
+    credit_limit_eur: Decimal | None = None
+    credit_headroom_eur: Decimal | None = None
+    consumption_mode: Literal["open", "restricted"]
+    next_month_allowance_eur: Decimal | None = None
+    payment_instruction: dict | None = None
+
+    @field_validator("period")
+    @classmethod
+    def validate_period(cls, value: str) -> str:
+        return reporting_month(value)
+
+    @field_validator(
+        "known_delivered_cost_eur",
+        "known_transfer_price_eur",
+        "prior_outstanding_eur",
+        "interest_eur",
+        "new_transfer_charges_eur",
+        "outstanding_eur",
+        mode="before",
+    )
+    @classmethod
+    def validate_nonnegative_money(cls, value):
+        return money(value, non_negative=True)
+
+    @field_validator(
+        "recognized_payments_eur",
+        "credit_limit_eur",
+        "credit_headroom_eur",
+        "next_month_allowance_eur",
+        mode="before",
+    )
+    @classmethod
+    def validate_optional_nonnegative_money(cls, value):
+        return None if value is None else money(value, non_negative=True)
+
+    @model_validator(mode="after")
+    def validate_dates(self):
+        if self.due_date < self.statement_date:
+            raise ValueError("due_date cannot precede statement_date")
+        return self
+
+    def statement_id(self) -> str:
+        return (
+            f"settlement:{self.financial_entity_id}:{self.procuring_entity_id}:"
+            f"{self.period}:{self.statement_date.isoformat()}"
+        )
+
+
 PlanningEvidence = Annotated[
     ForecastEvidence
     | UsageObservation

src/fin_hub/services/exchange.py

@@ -19,6 +19,7 @@ from fin_hub.schemas.exchange import (
     BookedCostEvidence,
     FinancialConstraintSignal,
     PlanningEvidence,
+    SettlementStatement,
 )
 from fin_hub.services.ledger import _connect, default_ledger_path
 
@@ -297,6 +298,100 @@ def ingest_planning_evidence(
     return record
 
 
+def _ensure_settlement_schema(conn: sqlite3.Connection) -> None:
+    conn.execute(
+        """
+        CREATE TABLE IF NOT EXISTS settlement_statements (
+            statement_id TEXT PRIMARY KEY,
+            financial_entity_id TEXT NOT NULL,
+            procuring_entity_id TEXT NOT NULL,
+            period TEXT NOT NULL,
+            payment_recognition TEXT NOT NULL,
+            payload_json TEXT NOT NULL,
+            source_evidence_ref 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_settlement_statements_entity_period "
+        "ON settlement_statements (financial_entity_id, procuring_entity_id, period, is_current)"
+    )
+    conn.commit()
+
+
+def ingest_settlement_statement(
+    payload: dict,
+    *,
+    source_evidence_ref: str,
+    ledger_path: Path | None = None,
+) -> SettlementStatement:
+    """Join a resource-control settlement statement as a referenced projection.
+
+    A settlement statement is never posted as booked spend: it is stored in
+    its own table, outside `ledger_entries`, and carries no path into the
+    booked-fact ledger.
+    """
+
+    if payload.get("record_type") != "settlement_statement":
+        raise ValueError("payload must have record_type=settlement_statement")
+    if not source_evidence_ref.strip():
+        raise ValueError("source_evidence_ref is required")
+    statement = SettlementStatement.model_validate(payload)
+    statement_id = statement.statement_id()
+    canonical = statement.model_dump_json()
+    ledger = ledger_path or default_ledger_path()
+    with _connect(ledger) as conn:
+        _ensure_settlement_schema(conn)
+        conn.execute("BEGIN IMMEDIATE")
+        existing = conn.execute(
+            "SELECT payload_json FROM settlement_statements WHERE statement_id = ?",
+            (statement_id,),
+        ).fetchone()
+        if existing is not None:
+            if json.loads(existing["payload_json"]) != json.loads(canonical):
+                raise ValueError("statement_id already exists with different content")
+            return statement
+        conn.execute(
+            "UPDATE settlement_statements SET is_current = 0 "
+            "WHERE financial_entity_id = ? AND procuring_entity_id = ? AND period = ?",
+            (statement.financial_entity_id, statement.procuring_entity_id, statement.period),
+        )
+        conn.execute(
+            "INSERT INTO settlement_statements "
+            "(statement_id, financial_entity_id, procuring_entity_id, period, "
+            "payment_recognition, payload_json, source_evidence_ref, is_current) "
+            "VALUES (?, ?, ?, ?, ?, ?, ?, 1)",
+            (
+                statement_id,
+                statement.financial_entity_id,
+                statement.procuring_entity_id,
+                statement.period,
+                statement.payment_recognition,
+                canonical,
+                source_evidence_ref,
+            ),
+        )
+        conn.commit()
+    return statement
+
+
+def current_settlement_statements(
+    *, ledger_path: Path | None = None
+) -> list[SettlementStatement]:
+    """Return current, referenced settlement statement projections."""
+
+    ledger = ledger_path or default_ledger_path()
+    with _connect(ledger) as conn:
+        _ensure_settlement_schema(conn)
+        rows = conn.execute(
+            "SELECT payload_json FROM settlement_statements "
+            "WHERE is_current = 1 ORDER BY statement_id"
+        ).fetchall()
+    return [SettlementStatement.model_validate_json(row["payload_json"]) for row in rows]
+
+
 def booked_cost_projection(
     *,
     ledger_path: Path | None = None,
@@ -473,6 +568,21 @@ def exchange_health(
             )
         )
 
+    for statement in current_settlement_statements(ledger_path=ledger):
+        if statement.payment_recognition == "unknown":
+            issues.append(
+                ExchangeQualityIssue(
+                    code="settlement_awaiting_payment_join",
+                    severity="warning",
+                    record_id=statement.statement_id(),
+                    detail=(
+                        f"{statement.outstanding_eur} EUR outstanding for "
+                        f"{statement.financial_entity_id} {statement.period} has no "
+                        "recognized payment join; this is a residual, not booked cost"
+                    ),
+                )
+            )
+
     return ExchangeHealthReport(
         schema_version="0.1",
         artifact_type="resource_cost_exchange_health",

tests/fixtures/settlement-statement-2026-08.json

@@ -0,0 +1,28 @@
+{
+  "schema_version": "0.1",
+  "record_type": "settlement_statement",
+  "terms_version": "V0.1",
+  "financial_entity_id": "entity:frontier",
+  "procuring_entity_id": "entity:railiance",
+  "period": "2026-08",
+  "statement_date": "2026-09-01",
+  "due_date": "2026-09-15",
+  "currency": "EUR",
+  "line_items": [
+    {"resource_id": "resource:platform:audit-storage", "delivered_cost_eur": "42.10", "markup_eur": "6.32"}
+  ],
+  "known_delivered_cost_eur": "42.10",
+  "known_transfer_price_eur": "48.42",
+  "unknown_cost_remainder": false,
+  "prior_outstanding_eur": "0.00",
+  "recognized_payments_eur": null,
+  "payment_recognition": "unknown",
+  "interest_eur": "0.00",
+  "new_transfer_charges_eur": "48.42",
+  "outstanding_eur": "48.42",
+  "credit_limit_eur": "500.00",
+  "credit_headroom_eur": "451.58",
+  "consumption_mode": "open",
+  "next_month_allowance_eur": null,
+  "payment_instruction": null
+}

tests/test_exchange.py

@@ -12,13 +12,16 @@ from fin_hub.schemas.exchange import (
     BookedCostEvidence,
     ForecastEvidence,
 )
+from fin_hub.schemas.exchange import SettlementStatement
 from fin_hub.services.exchange import (
     booked_cost_projection,
+    current_settlement_statements,
     exchange_health,
     financial_constraint_projection,
     ingest_planning_evidence,
     ingest_resource_forecast,
     ingest_resource_usage,
+    ingest_settlement_statement,
     reconcile_resource_period,
 )
 from fin_hub.services.ledger import import_csv
@@ -413,3 +416,86 @@ def test_audit_storage_usage_omits_nulls_and_does_not_invent_zero_spend(tmp_path
     assert august.invented_zero is False
     health = exchange_health(ledger_path=ledger)
     assert "usage_without_booked_fact" in {issue.code for issue in health.issues}
+
+
+def _settlement_payload() -> dict:
+    return json.loads((FIXTURES / "settlement-statement-2026-08.json").read_text())
+
+
+def test_settlement_statement_joins_as_referenced_projection_not_booked_cost(
+    tmp_path: Path,
+):
+    ledger = tmp_path / "ledger.db"
+    payload = _settlement_payload()
+
+    statement = ingest_settlement_statement(
+        payload,
+        source_evidence_ref="resource-control:data/settlements/2026-08.json",
+        ledger_path=ledger,
+    )
+
+    assert isinstance(statement, SettlementStatement)
+    assert statement.financial_entity_id == "entity:frontier"
+    assert statement.outstanding_eur == Decimal("48.42")
+    assert statement.payment_recognition == "unknown"
+
+    current = current_settlement_statements(ledger_path=ledger)
+    assert len(current) == 1
+    assert current[0].statement_id() == statement.statement_id()
+
+    # No booked cost is created by joining a settlement statement.
+    assert booked_cost_projection(ledger_path=ledger) == []
+
+
+def test_settlement_statement_is_idempotent_and_supersedes_prior_period_statement(
+    tmp_path: Path,
+):
+    ledger = tmp_path / "ledger.db"
+    payload = _settlement_payload()
+
+    ingest_settlement_statement(
+        payload, source_evidence_ref="resource-control:data/settlements/2026-08.json",
+        ledger_path=ledger,
+    )
+    ingest_settlement_statement(
+        payload, source_evidence_ref="resource-control:data/settlements/2026-08.json",
+        ledger_path=ledger,
+    )
+    assert len(current_settlement_statements(ledger_path=ledger)) == 1
+
+    corrected = {**payload, "statement_date": "2026-09-05", "outstanding_eur": "10.00",
+                 "new_transfer_charges_eur": "10.00", "known_transfer_price_eur": "10.00"}
+    ingest_settlement_statement(
+        corrected, source_evidence_ref="resource-control:data/settlements/2026-08-corrected.json",
+        ledger_path=ledger,
+    )
+
+    current = current_settlement_statements(ledger_path=ledger)
+    assert len(current) == 1
+    assert current[0].outstanding_eur == Decimal("10.00")
+
+
+def test_settlement_statement_rejects_provider_invoice_shaped_payload(tmp_path: Path):
+    ledger = tmp_path / "ledger.db"
+    with pytest.raises(ValueError, match="record_type=settlement_statement"):
+        ingest_settlement_statement(
+            {"record_type": "booked_cost"},
+            source_evidence_ref="resource-control:x",
+            ledger_path=ledger,
+        )
+
+
+def test_exchange_health_surfaces_settlement_without_payment_join_as_residual(
+    tmp_path: Path,
+):
+    ledger = tmp_path / "ledger.db"
+    ingest_settlement_statement(
+        _settlement_payload(),
+        source_evidence_ref="resource-control:data/settlements/2026-08.json",
+        ledger_path=ledger,
+    )
+
+    health = exchange_health(ledger_path=ledger)
+
+    assert "settlement_awaiting_payment_join" in {issue.code for issue in health.issues}
+    assert health.booked_fact_count == 0

workplans/FIN-WP-0004-resource-cost-evidence-contract.md

@@ -4,12 +4,12 @@ type: workplan
 title: "Establish the resource cost evidence contract"
 domain: infotech
 repo: fin-hub
-status: active
+status: blocked
 flavor: implementation
 owner: codex
 topic_slug: financials
 created: "2026-08-10"
-updated: "2026-08-15"
+updated: "2026-09-27"
 related:
   - FIN-WP-0001
   - RESOURCE-WP-0002
@@ -208,6 +208,13 @@ reconciles as usage present, `missing_booked_fact: true`,
 `docs/evidence/fin-wp-0004-t05-commissioning-2026-08.md`. T05 still
 waits on the first Scaleway charge or credit (expected 2026-09).
 
+Status 2026-09-27: reviewed the workplan for closeable work. T05 is the
+only open task and remains genuinely blocked on an external event (the
+first real `platform:audit-storage` provider charge or credit) that
+resource-control has not yet published; there is no remaining code or
+contract work in this repo to advance it. Set workplan `status: blocked`
+to reflect that nothing here is actionable until that fact arrives.
+
 ## T06 — Generalize and operate the contract
 
 ```task

workplans/FIN-WP-0005-datev-accounting-adapter-operations.md

@@ -4,12 +4,12 @@ type: workplan
 title: "DATEV accounting adapter operations"
 domain: financials
 repo: fin-hub
-status: active
+status: blocked
 flavor: implementation
 owner: codex
 topic_slug: financials
 created: "2026-08-11"
-updated: "2026-08-11"
+updated: "2026-09-27"
 related:
   - FIN-WP-0002
 state_hub_workstream_id: "58432770-da19-5b72-a946-cacbe1eb24ca"
@@ -31,7 +31,7 @@ invoice numbers, execute payments, or perform bookkeeping.
 
 ```task
 id: FIN-WP-0005-T01
-status: todo
+status: wait
 priority: high
 state_hub_task_id: "7cdda85d-cdd6-544c-b873-f27ec7ef448d"
 ```
@@ -42,6 +42,13 @@ retention policy, correction procedure, and required structured fields.
 Record which capabilities use Qonto's managed DATEV services and which require
 a direct DATEV integration. Route credentials only after this selection.
 
+Status 2026-09-27: reviewed for closeable work. T01 requires a live
+confirmation from the operator and Steuerbüro (plan/tenant selection,
+master-data ownership, retention and correction policy) that no in-repo
+change can substitute for; set to `wait` pending that conversation. T02-T04
+already wait on T01 and a non-production DATEV tenant, so no task in this
+workplan is actionable from the repo alone right now.
+
 ## Implement the selected transport
 
 ```task

workplans/FIN-WP-0006-internal-transfer-settlement.md

@@ -4,12 +4,12 @@ type: workplan
 title: "Consume resource-control internal transfer settlement"
 domain: financials
 repo: fin-hub
-status: proposed
+status: finished
 flavor: residual
 owner: codex
 topic_slug: financials
 created: "2026-08-14"
-updated: "2026-08-14"
+updated: "2026-09-27"
 related:
   - FIN-WP-0004
   - RESOURCE-WP-0005
@@ -43,7 +43,7 @@ Origin: `RESOURCE-WP-0005-T07`. Terms live in
 
 ```task
 id: FIN-WP-0006-T01
-status: todo
+status: done
 flavor: residual
 priority: high
 state_hub_task_id: "9ef3fab5-51a6-57ef-8121-b9714298c502"
@@ -57,3 +57,23 @@ a provider invoice.
 Done when an inbound fixture can be stored as a referenced projection
 and `exchange_health` shows statements lacking a payment join as a
 visible residual, not as booked cost.
+
+Completed 2026-09-27: added `SettlementStatement`
+(`src/fin_hub/schemas/exchange.py`) matching resource-control's v0.1
+`settlement-statement.schema.json`, with a dedicated
+`settlement_statements` table (`services/exchange.py`) that never touches
+`ledger_entries`. `ingest_settlement_statement` validates and stores each
+statement keyed by `financial_entity_id` + `procuring_entity_id` + `period`
++ `statement_date`, idempotent on identical resubmission, and superseding
+the prior current statement for the same entity/period pair on a later
+`statement_date`. `exchange_health` now emits
+`settlement_awaiting_payment_join` (severity `warning`) for any current
+statement with `payment_recognition: "unknown"`, listing its outstanding
+amount as a residual rather than booked cost.
+`current_settlement_statements` exposes the stored projection for
+reporting. CLI: `finhub ledger ingest-settlement` and
+`finhub ledger settlements`. Fixture:
+`tests/fixtures/settlement-statement-2026-08.json`. Tests:
+`tests/test_exchange.py::test_settlement_statement_*` and
+`::test_exchange_health_surfaces_settlement_without_payment_join_as_residual`.
+Full suite: 102 passed.