Implement hosted Target Ledger append API (WP-0006-T04)

migrations/0002_ledger.sql adds ledger_entries with an identity-column
sequence for exact append order and no UPDATE/DELETE grant for trf_app.
src/target_revenue/ledger.py: append_entry() rejects caller-supplied
previous_entry_hash/signature, enforces per-Licensor phase ownership,
serializes concurrent appends via pg_advisory_xact_lock, computes the
chain tip and signs with the Trust Service instance's own Ed25519 key
(service/keys.py), reusing validation.py's checks unchanged. Adds
POST/GET /phases/{id}/ledger and an unauthenticated GET /public-key.

Also fixes a route-ordering bug found while wiring this in: phase IDs
never needed the {phase_id:path} converter (they contain colons, not
slashes), and its greedy matching was swallowing /ledger-suffixed
paths into the plain GET /phases/{id} route.

tests/test_ledger_hosting.py (8 Docker-gated tests) exercises hash-chain
linkage, forged-field rejection, cross-Licensor isolation, currency and
duplicate-id rejection, DB-privilege enforcement, signature
verification via the public-key endpoint, and the task's own
highest-priority property: append -> export -> offline fold reproduces
the exact expected Development/Remission Credit and Outstanding Target.

tests/test_ledger_hosting.py

@@ -0,0 +1,276 @@
+"""Integration tests for WP-0006-T04 (hosted Target Ledger append API).
+
+Same ephemeral, disposable Postgres-via-Docker pattern as
+test_registry_hosting.py (never the shared state-hub instance). The single
+property this file tests most aggressively, per the task's own framing:
+appending entries through the hosted API and then folding the exported
+result offline (`fold.fold_outstanding_target`) must produce exactly the
+same Outstanding Target a hosted metrics/attestation component would later
+report — a hosted service that silently diverges from the offline fold is
+the highest-risk defect this component could introduce.
+"""
+
+from __future__ import annotations
+
+import shutil
+import subprocess
+import time
+import uuid
+from pathlib import Path
+
+import pytest
+
+psycopg = pytest.importorskip("psycopg")
+pytest.importorskip("fastapi")
+from fastapi.testclient import TestClient  # noqa: E402
+
+from conftest import golden_manifest  # noqa: E402
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+MIGRATIONS = [
+    REPO_ROOT / "migrations" / "0001_registries.sql",
+    REPO_ROOT / "migrations" / "0002_ledger.sql",
+]
+
+pytestmark = pytest.mark.skipif(
+    shutil.which("docker") is None, reason="docker not available"
+)
+
+
+@pytest.fixture(scope="module")
+def pg_container():
+    name = f"trf-test-pg-ledger-{uuid.uuid4().hex[:8]}"
+    subprocess.run(
+        [
+            "docker", "run", "--rm", "-d",
+            "--name", name,
+            "-e", "POSTGRES_PASSWORD=postgres",
+            "-e", "POSTGRES_DB=target_revenue_test",
+            "-p", "127.0.0.1::5432",
+            "postgres:16-alpine",
+        ],
+        check=True, capture_output=True,
+    )
+    try:
+        port_out = subprocess.run(
+            ["docker", "port", name, "5432/tcp"], check=True, capture_output=True, text=True
+        ).stdout.strip()
+        host_port = port_out.split(":")[-1]
+        dsn = f"host=127.0.0.1 port={host_port} dbname=target_revenue_test user=postgres password=postgres"
+
+        for _ in range(60):
+            try:
+                with psycopg.connect(dsn, connect_timeout=1):
+                    break
+            except psycopg.OperationalError:
+                time.sleep(0.5)
+        else:
+            raise RuntimeError("postgres container did not become ready in time")
+
+        with psycopg.connect(dsn) as conn:
+            for migration in MIGRATIONS:
+                conn.execute(migration.read_text(encoding="utf-8"))
+                conn.commit()
+            token = "test-token-acme"
+            conn.execute(
+                "INSERT INTO licensors (token, licensor_id) VALUES (%s, %s)",
+                (token, "acme-corp"),
+            )
+            other_token = "test-token-other"
+            conn.execute(
+                "INSERT INTO licensors (token, licensor_id) VALUES (%s, %s)",
+                (other_token, "other-corp"),
+            )
+            conn.commit()
+
+        app_dsn = (
+            f"host=127.0.0.1 port={host_port} dbname=target_revenue_test "
+            f"user=trf_app password=changeme-in-deployment"
+        )
+        yield {
+            "admin_dsn": dsn,
+            "app_dsn": app_dsn,
+            "token": token,
+            "other_token": other_token,
+        }
+    finally:
+        subprocess.run(["docker", "stop", name], capture_output=True)
+
+
+@pytest.fixture()
+def client(pg_container, monkeypatch):
+    monkeypatch.setenv("TRF_DATABASE_URL", pg_container["app_dsn"])
+    monkeypatch.setenv("TRF_SIGNING_KEY_HEX", "11" * 32)
+    from target_revenue.service import app as app_module
+
+    if hasattr(app_module.app.state, "pool"):
+        app_module.app.state.pool.close()
+        del app_module.app.state.pool
+    if hasattr(app_module.app.state, "signing_key"):
+        del app_module.app.state.signing_key
+    with TestClient(app_module.app) as c:
+        yield c
+
+
+def auth_headers(token: str):
+    return {"Authorization": f"Bearer {token}"}
+
+
+@pytest.fixture()
+def registered_phase(client, pg_container):
+    manifest = golden_manifest()
+    manifest["phase"]["id"] = manifest["phase"]["id"] + "-ledger-test-" + uuid.uuid4().hex[:6]
+    resp = client.post("/phases", json=manifest, headers=auth_headers(pg_container["token"]))
+    assert resp.status_code == 201, resp.text
+    return manifest
+
+
+def _entry(phase_id: str, entry_id: str, entry_type: str, amount: float, recognized_at: str):
+    return {
+        "id": entry_id,
+        "phase": phase_id,
+        "type": entry_type,
+        "amount": amount,
+        "currency": "USD",
+        "recognized_at": recognized_at,
+        "evidence_reference": f"confidential:evidence:{entry_id}",
+        "extension": {"id": "trsl:extension:development-license", "version": "1.0"},
+    }
+
+
+def test_append_and_hash_chain_links(client, pg_container, registered_phase):
+    phase_id = registered_phase["phase"]["id"]
+    e1 = _entry(phase_id, "trsl:entry:ledgertest0001", "development-credit", 1000, "2026-08-01T00:00:00Z")
+    r1 = client.post(f"/phases/{phase_id}/ledger", json=e1, headers=auth_headers(pg_container["token"]))
+    assert r1.status_code == 201, r1.text
+    stored1 = r1.json()
+    assert stored1["previous_entry_hash"] == "GENESIS"
+    assert "signature" in stored1
+
+    e2 = _entry(phase_id, "trsl:entry:ledgertest0002", "development-credit", 500, "2026-08-02T00:00:00Z")
+    r2 = client.post(f"/phases/{phase_id}/ledger", json=e2, headers=auth_headers(pg_container["token"]))
+    assert r2.status_code == 201, r2.text
+    stored2 = r2.json()
+
+    from target_revenue import hashing
+
+    assert stored2["previous_entry_hash"] == hashing.entry_hash(stored1)
+    # Full chain, as exported, must independently verify.
+    hashing.verify_chain([stored1, stored2])
+
+
+def test_client_cannot_supply_previous_entry_hash_or_signature(client, pg_container, registered_phase):
+    phase_id = registered_phase["phase"]["id"]
+    forged = _entry(phase_id, "trsl:entry:ledgertest0003", "development-credit", 1, "2026-08-01T00:00:00Z")
+    forged["previous_entry_hash"] = "GENESIS"
+    resp = client.post(f"/phases/{phase_id}/ledger", json=forged, headers=auth_headers(pg_container["token"]))
+    assert resp.status_code == 422
+    assert "must not be supplied" in resp.json()["detail"]
+
+
+def test_other_licensor_cannot_append_to_this_phase(client, pg_container, registered_phase):
+    phase_id = registered_phase["phase"]["id"]
+    e = _entry(phase_id, "trsl:entry:ledgertest0004", "development-credit", 1, "2026-08-01T00:00:00Z")
+    resp = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["other_token"]))
+    assert resp.status_code == 422
+    assert "not authorized" in resp.json()["detail"]
+
+
+def test_currency_mismatch_rejected(client, pg_container, registered_phase):
+    phase_id = registered_phase["phase"]["id"]
+    e = _entry(phase_id, "trsl:entry:ledgertest0005", "development-credit", 1, "2026-08-01T00:00:00Z")
+    e["currency"] = "EUR"
+    resp = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
+    assert resp.status_code == 422
+    assert "currency" in resp.json()["detail"]
+
+
+def test_hosted_ledger_export_folds_identically_to_offline_computation(
+    client, pg_container, registered_phase
+):
+    """The property T04's task description calls out as highest-risk:
+    hosted append + export, folded offline, must match what was appended."""
+    from target_revenue import fold, hashing
+
+    phase_id = registered_phase["phase"]["id"]
+    plan = [
+        ("development-credit", 25000),
+        ("development-credit", 10000),
+        ("remission-credit", 5000),
+        ("credit-reversal", 2000),
+    ]
+    for i, (entry_type, amount) in enumerate(plan, start=1):
+        e = _entry(
+            phase_id,
+            f"trsl:entry:ledgertestfold{i:04d}",
+            entry_type,
+            amount,
+            f"2026-08-0{i}T00:00:00Z",
+        )
+        if entry_type == "credit-reversal":
+            e["reverses"] = "trsl:entry:ledgertestfold0001"
+        resp = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
+        assert resp.status_code == 201, resp.text
+
+    export = client.get(f"/phases/{phase_id}/ledger")
+    assert export.status_code == 200
+    entries = export.json()
+    assert len(entries) == len(plan)
+
+    # Chain integrity survives the round trip through Postgres/JSONB/API.
+    hashing.verify_chain(entries)
+
+    result = fold.fold_outstanding_target(
+        registered_phase["phase"]["initial_target"]["amount"], entries
+    )
+    expected_development_credit = 25000 + 10000 - 2000
+    expected_remission_credit = 5000
+    assert result.development_credit == expected_development_credit
+    assert result.remission_credit == expected_remission_credit
+    assert result.outstanding_target == max(
+        0.0,
+        registered_phase["phase"]["initial_target"]["amount"]
+        - expected_development_credit
+        - expected_remission_credit,
+    )
+
+
+def test_public_key_endpoint_unauthenticated_and_verifies_signature(client, pg_container, registered_phase):
+    phase_id = registered_phase["phase"]["id"]
+    e = _entry(phase_id, "trsl:entry:ledgertestpk0001", "development-credit", 1, "2026-08-01T00:00:00Z")
+    stored = client.post(
+        f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"])
+    ).json()
+
+    pk_resp = client.get("/public-key")
+    assert pk_resp.status_code == 200
+    assert pk_resp.json()["algorithm"] == "ed25519"
+
+    from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
+
+    from target_revenue import hashing
+
+    public_key = Ed25519PublicKey.from_public_bytes(
+        bytes.fromhex(pk_resp.json()["public_key_hex"])
+    )
+    assert hashing.verify_record_signature(stored, stored["signature"], public_key)
+
+
+def test_application_role_cannot_update_or_delete_ledger_entries(pg_container):
+    with psycopg.connect(pg_container["app_dsn"]) as conn:
+        with pytest.raises(psycopg.errors.InsufficientPrivilege):
+            conn.execute("UPDATE ledger_entries SET signature = 'x' WHERE entry_id = 'nonexistent'")
+        conn.rollback()
+        with pytest.raises(psycopg.errors.InsufficientPrivilege):
+            conn.execute("DELETE FROM ledger_entries WHERE entry_id = 'nonexistent'")
+        conn.rollback()
+
+
+def test_duplicate_entry_id_rejected(client, pg_container, registered_phase):
+    phase_id = registered_phase["phase"]["id"]
+    e = _entry(phase_id, "trsl:entry:ledgertestdup0001", "development-credit", 1, "2026-08-01T00:00:00Z")
+    first = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
+    assert first.status_code == 201
+    second = client.post(f"/phases/{phase_id}/ledger", json=e, headers=auth_headers(pg_container["token"]))
+    assert second.status_code == 422
+    assert "already exists" in second.json()["detail"]