Implement Phase 0b base project against the frozen test contract

All seven services (gateway, ledger, fraud, sanctions, insights, notifier,
audit), db/schema.sql plus optional_indexes.sql, scripts/seed.py and
smoke.py, the gateway-served frontend, and docker-compose.yml. Plain-text
stdlib logging and a weak manual X-Request-Id only -- no observability yet,
by design, per test_no_observability.py. Implements all 13 named chaos-flag
pathologies from docs/PATHOLOGIES.md behind Redis-backed runtime flags.

make smoke and make test-contract pass; make test-baseline and
make test-pathology pass except for a handful of load-sensitive tests
documented in tests/README.md rather than papered over.

services/audit/consumer.py

@@ -0,0 +1,68 @@
+import asyncio
+import json
+import logging
+
+import sqlalchemy as sa
+
+from common.db import Session
+from common.redis_client import redis_client
+from common.settings import settings
+
+logger = logging.getLogger("audit")
+
+GROUP = "audit"
+CONSUMER = "audit-1"
+
+
+async def ensure_group():
+    try:
+        await redis_client.xgroup_create(settings.stream_name, GROUP, id="$", mkstream=True)
+    except Exception as exc:
+        if "BUSYGROUP" not in str(exc):
+            raise
+
+
+async def _handle(fields: dict):
+    async with Session() as session:
+        await session.execute(
+            sa.text(
+                "INSERT INTO audit_events (transaction_id, event_type, payload, recorded_at) "
+                "VALUES (:tx_id, 'transaction_processed', CAST(:payload AS jsonb), now())"
+            ),
+            {"tx_id": fields.get("transaction_id"), "payload": json.dumps(fields)},
+        )
+        await session.commit()
+
+
+async def run():
+    """Second, independent consumer group on the same stream as `notifier` --
+    proves the async fork fans out to genuinely separate consumers, not one
+    consumer doing both jobs."""
+    await ensure_group()
+    logger.info("audit consumer started")
+    while True:
+        try:
+            resp = await redis_client.xreadgroup(
+                groupname=GROUP,
+                consumername=CONSUMER,
+                streams={settings.stream_name: ">"},
+                count=10,
+                block=2000,
+            )
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            logger.exception("audit consumer read failed")
+            await asyncio.sleep(1)
+            continue
+
+        if not resp:
+            continue
+
+        for _, messages in resp:
+            for msg_id, fields in messages:
+                try:
+                    await _handle(fields)
+                except Exception:
+                    logger.exception(f"audit failed to process message {msg_id}")
+                await redis_client.xack(settings.stream_name, GROUP, msg_id)

services/audit/worker.py

@@ -0,0 +1,10 @@
+import asyncio
+
+from common.logging_config import configure_logging
+
+from . import consumer
+
+configure_logging("audit")
+
+if __name__ == "__main__":
+    asyncio.run(consumer.run())

services/fraud/app.py

@@ -0,0 +1,77 @@
+import httpx
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+from common.http import downstream_client
+from common.logging_config import configure_logging
+from common.redis_client import redis_client
+from common.settings import settings
+
+from . import chaos
+from .scoring import score
+
+SERVICE_NAME = "fraud"
+logger = configure_logging(SERVICE_NAME)
+app = FastAPI(title=SERVICE_NAME)
+
+
+class ScoreRequest(BaseModel):
+    account_id: str
+    holder_name: str
+    category_id: int | None = None
+    amount_minor: int
+    request_id: str = "unknown"
+
+
+@app.get("/")
+async def root():
+    return {"service": SERVICE_NAME, "version": "0.1.0"}
+
+
+@app.get("/healthz")
+async def healthz(request: Request):
+    if request.headers.get("X-Ledgerline-Probe") == "chatty-health":
+        logger.info("received chatty-health healthz probe from gateway")
+    return {"status": "ok", "service": SERVICE_NAME}
+
+
+@app.get("/readyz")
+async def readyz():
+    try:
+        await redis_client.ping()
+    except Exception as exc:
+        logger.exception(f"readyz check failed: {exc}")
+        return JSONResponse(status_code=503, content={"status": "unavailable"})
+    return {"status": "ok"}
+
+
+@app.post("/internal/score")
+async def score_transaction(body: ScoreRequest):
+    logger.info(f"scoring account {body.account_id} request_id={body.request_id}")
+    await redis_client.incr("fraud:calls_total")
+
+    if await chaos.blocking_event_loop_enabled():
+        logger.info(f"blocking_event_loop chaos engaged for request_id={body.request_id}")
+        chaos.synchronous_cpu_work()
+
+    async with downstream_client(body.request_id) as client:
+        try:
+            resp = await client.post(
+                f"{settings.sanctions_url}/internal/lookup",
+                json={"holder_name": body.holder_name, "request_id": body.request_id},
+            )
+        except httpx.TimeoutException:
+            logger.exception(f"sanctions lookup timed out for request_id={body.request_id}")
+            return JSONResponse(status_code=504, content={"error": "sanctions lookup timed out"})
+        except httpx.RequestError as exc:
+            logger.exception(f"sanctions lookup failed for request_id={body.request_id}: {exc}")
+            return JSONResponse(status_code=502, content={"error": "sanctions lookup failed"})
+
+    if resp.status_code >= 300:
+        logger.info(f"sanctions returned {resp.status_code} for request_id={body.request_id}")
+        return JSONResponse(status_code=502, content={"error": "sanctions lookup failed"})
+
+    matched = resp.json().get("matched", False)
+    risk_score = score(body.category_id, matched)
+    return {"risk_score": risk_score, "matched": matched}

services/fraud/chaos.py

@@ -0,0 +1,22 @@
+"""blocking_event_loop -- phase-0-base-project-spec.md §7. Owned by `fraud`."""
+import hashlib
+import time
+
+from common.chaos_flags import is_enabled
+
+_CPU_WORK_SECONDS = 0.15
+
+
+async def blocking_event_loop_enabled() -> bool:
+    return await is_enabled("blocking_event_loop")
+
+
+def synchronous_cpu_work() -> None:
+    """~120ms of real, synchronous, CPU-bound work run directly inside an
+    `async def` handler -- no run_in_executor/to_thread. This is the point:
+    it blocks the whole event loop, serialising every other in-flight
+    request on this worker for the duration."""
+    deadline = time.perf_counter() + _CPU_WORK_SECONDS
+    data = b"ledgerline-fraud-scoring"
+    while time.perf_counter() < deadline:
+        data = hashlib.sha256(data).digest()

services/fraud/scoring.py

@@ -0,0 +1,18 @@
+import random
+
+
+def category_weight(category_id: int | None) -> float:
+    """A fake but deterministic-ish stand-in for merchant_categories.risk_weight
+    (fraud has no direct DB access -- Redis only, per phase-0-base-project-spec.md §4).
+    The business logic does not need to be correct, only to produce a plausible
+    spread of scores."""
+    if category_id is None:
+        return 0.3
+    return 0.01 + (category_id % 45) / 44 * 0.94
+
+
+def score(category_id: int | None, matched: bool) -> float:
+    base = category_weight(category_id) * random.uniform(0.05, 0.3)
+    if matched:
+        return min(1.0, base + random.uniform(0.55, 0.9))
+    return min(1.0, base)

services/gateway/app.py

@@ -0,0 +1,151 @@
+import asyncio
+import uuid
+from pathlib import Path
+
+import httpx
+from fastapi import FastAPI, Request
+from fastapi.responses import FileResponse, JSONResponse
+from fastapi.staticfiles import StaticFiles
+from pydantic import BaseModel, field_validator
+
+from common.chaos_flags import ALL_FLAGS, get_flags, set_flag, set_profile
+from common.logging_config import configure_logging
+from common.settings import settings
+
+from . import orchestrator
+from .orchestrator import UpstreamFailure
+
+SERVICE_NAME = "gateway"
+logger = configure_logging(SERVICE_NAME)
+app = FastAPI(title=SERVICE_NAME)
+
+STATIC_DIR = Path(__file__).parent / "static"
+app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
+
+# A shared, connection-pooling client for the simple read-only proxy routes
+# below -- opening a fresh httpx.AsyncClient (and TCP connection) per request
+# is real, needless overhead under concurrent load. orchestrator.py's
+# downstream_client() stays per-request since it carries per-transaction
+# headers (X-Request-Id).
+_proxy_client = httpx.AsyncClient(
+    timeout=settings.downstream_timeout_seconds,
+    limits=httpx.Limits(max_connections=100, max_keepalive_connections=50),
+)
+
+DOWNSTREAM_URLS = {
+    "ledger": settings.ledger_url,
+    "fraud": settings.fraud_url,
+    "sanctions": settings.sanctions_url,
+    "insights": settings.insights_url,
+    "notifier": settings.notifier_url,
+}
+
+
+class TransactionRequest(BaseModel):
+    account_id: str
+    kind: str
+    amount_minor: int
+    currency: str
+    merchant_name: str
+    category_id: int | None = None
+
+    @field_validator("kind")
+    @classmethod
+    def kind_must_be_known(cls, v):
+        if v not in ("deposit", "withdrawal", "purchase"):
+            raise ValueError("kind must be one of deposit, withdrawal, purchase")
+        return v
+
+    @field_validator("amount_minor")
+    @classmethod
+    def deposit_amount_must_be_non_negative(cls, v, info):
+        if info.data.get("kind") == "deposit" and v < 0:
+            raise ValueError("amount_minor must be non-negative for a deposit")
+        return v
+
+
+class ChaosFlagRequest(BaseModel):
+    flag: str
+    enabled: bool
+
+
+class ChaosProfileRequest(BaseModel):
+    profile: str
+
+
+@app.get("/", include_in_schema=False)
+async def index():
+    """Serves the frontend (phase-0-base-project-spec.md §10). This overrides
+    §4's generic `GET / -> {"service", "version"}` info route, which the
+    other five HTTP services still expose as-is."""
+    return FileResponse(STATIC_DIR / "index.html")
+
+
+@app.get("/healthz")
+async def healthz():
+    return {"status": "ok", "service": SERVICE_NAME}
+
+
+@app.get("/readyz")
+async def readyz():
+    return {"status": "ok"}
+
+
+@app.get("/api/accounts")
+async def list_accounts():
+    resp = await _proxy_client.get(f"{settings.ledger_url}/internal/accounts")
+    return JSONResponse(status_code=resp.status_code, content=resp.json())
+
+
+@app.get("/api/accounts/{account_id}/insights")
+async def account_insights(account_id: str):
+    resp = await _proxy_client.get(f"{settings.insights_url}/internal/accounts/{account_id}/summary")
+    return JSONResponse(status_code=resp.status_code, content=resp.json())
+
+
+@app.post("/api/transactions", status_code=201)
+async def submit_transaction(body: TransactionRequest):
+    request_id = str(uuid.uuid4())
+    try:
+        result = await orchestrator.create_transaction(body.model_dump(), request_id)
+    except UpstreamFailure as exc:
+        return JSONResponse(
+            status_code=exc.status_code, content={"error": exc.message, "request_id": request_id}
+        )
+    return JSONResponse(status_code=201, content=result)
+
+
+@app.get("/api/health/all")
+async def health_all():
+    async def check(name, url):
+        try:
+            resp = await _proxy_client.get(f"{url}/healthz", timeout=2.0)
+            return name, "ok" if resp.status_code == 200 else "down"
+        except httpx.RequestError:
+            return name, "down"
+
+    results = await asyncio.gather(*[check(name, url) for name, url in DOWNSTREAM_URLS.items()])
+    services = {"gateway": "ok", **dict(results)}
+    return {"services": services}
+
+
+@app.get("/api/admin/chaos")
+async def get_chaos():
+    return {"flags": await get_flags()}
+
+
+@app.post("/api/admin/chaos")
+async def post_chaos(body: ChaosFlagRequest):
+    if body.flag not in ALL_FLAGS:
+        return JSONResponse(status_code=400, content={"error": f"unknown flag {body.flag!r}"})
+    await set_flag(body.flag, body.enabled)
+    return {"flags": await get_flags()}
+
+
+@app.post("/api/admin/chaos/profile")
+async def post_chaos_profile(body: ChaosProfileRequest):
+    try:
+        flags = await set_profile(body.profile)
+    except ValueError as exc:
+        return JSONResponse(status_code=400, content={"error": str(exc)})
+    return {"flags": flags}

services/gateway/chaos.py

@@ -0,0 +1,23 @@
+"""retry_storm, no_batching, chatty_health -- phase-0-base-project-spec.md §7.
+All three are owned by `gateway`."""
+import random
+
+from common.chaos_flags import is_enabled
+
+DOWNSTREAMS = ("ledger", "fraud", "sanctions", "insights", "notifier")
+
+
+async def retry_storm_enabled() -> bool:
+    return await is_enabled("retry_storm")
+
+
+async def no_batching_enabled() -> bool:
+    return await is_enabled("no_batching")
+
+
+async def chatty_health_enabled() -> bool:
+    return await is_enabled("chatty_health")
+
+
+def synthetic_line_item_count() -> int:
+    return random.randint(1, 5)

services/gateway/orchestrator.py

@@ -0,0 +1,167 @@
+import logging
+import time
+import uuid
+
+import httpx
+
+from common.http import downstream_client
+from common.redis_client import redis_client
+from common.settings import settings
+
+from . import chaos
+
+logger = logging.getLogger("gateway")
+
+REJECT_THRESHOLD = 0.8
+
+
+class UpstreamFailure(Exception):
+    def __init__(self, status_code: int, message: str):
+        self.status_code = status_code
+        self.message = message
+
+
+async def _post(client: httpx.AsyncClient, url: str, json: dict) -> httpx.Response:
+    try:
+        return await client.post(url, json=json)
+    except httpx.TimeoutException as exc:
+        raise UpstreamFailure(504, f"{url} timed out") from exc
+    except httpx.RequestError as exc:
+        raise UpstreamFailure(502, f"{url} unreachable") from exc
+
+
+async def _score_once(client: httpx.AsyncClient, payload: dict) -> dict:
+    resp = await _post(client, f"{settings.fraud_url}/internal/score", payload)
+    if resp.status_code >= 300:
+        status = resp.status_code if resp.status_code in (502, 504) else 502
+        raise UpstreamFailure(status, "fraud scoring failed")
+    return resp.json()
+
+
+async def _score_with_retry(client: httpx.AsyncClient, payload: dict, retry_storm: bool) -> dict:
+    """retry_storm: on any non-2xx from fraud, retry 3 times immediately --
+    no backoff, no jitter, no circuit breaker (phase-0-base-project-spec.md §7).
+    "No circuit breaker" is taken literally here: once triggered, it fires all
+    3 retries unconditionally rather than stopping early on the first success
+    -- it never checks whether it still needs to. The last attempt's outcome
+    wins."""
+    attempts = 4 if retry_storm else 1
+    result = None
+    last_exc = None
+    for _ in range(attempts):
+        try:
+            result = await _score_once(client, payload)
+            last_exc = None
+        except UpstreamFailure as exc:
+            last_exc = exc
+    if last_exc is not None:
+        raise last_exc
+    return result
+
+
+async def _run_chatty_health_probe(client: httpx.AsyncClient) -> None:
+    """chatty_health: GET /healthz on all five downstreams, sequentially,
+    before doing any real work."""
+    headers = {"X-Ledgerline-Probe": "chatty-health"}
+    for name in chaos.DOWNSTREAMS:
+        url = getattr(settings, f"{name}_url")
+        try:
+            await client.get(f"{url}/healthz", headers=headers, timeout=2.0)
+        except httpx.RequestError:
+            pass
+
+
+async def create_transaction(payload: dict, request_id: str) -> dict:
+    start = time.perf_counter()
+    logger.info(f"start transaction for account {payload['account_id']} request_id={request_id}")
+
+    retry_storm = await chaos.retry_storm_enabled()
+    no_batching = await chaos.no_batching_enabled()
+    chatty_health = await chaos.chatty_health_enabled()
+
+    async with downstream_client(request_id) as client:
+        if chatty_health:
+            await _run_chatty_health_probe(client)
+
+        create_resp = await _post(
+            client,
+            f"{settings.ledger_url}/internal/transactions",
+            {**payload, "request_id": request_id},
+        )
+        if create_resp.status_code == 404:
+            raise UpstreamFailure(404, "account not found")
+        if create_resp.status_code >= 500:
+            raise UpstreamFailure(create_resp.status_code, "ledger failed unexpectedly")
+        if create_resp.status_code >= 300:
+            raise UpstreamFailure(502, "ledger rejected the transaction")
+
+        created = create_resp.json()
+        tx_id = created["transaction_id"]
+
+        score_payload = {
+            "account_id": created["account_id"],
+            "holder_name": created["holder_name"],
+            "category_id": created["category_id"],
+            "request_id": request_id,
+        }
+
+        if no_batching:
+            item_count = chaos.synthetic_line_item_count()
+            scores = []
+            for _ in range(item_count):
+                item_payload = {**score_payload, "amount_minor": max(1, created["amount_minor"] // item_count)}
+                scores.append(await _score_with_retry(client, item_payload, retry_storm))
+            risk_score = max(s["risk_score"] for s in scores)
+        else:
+            score_payload["amount_minor"] = created["amount_minor"]
+            scored = await _score_with_retry(client, score_payload, retry_storm)
+            risk_score = scored["risk_score"]
+
+        decision = "rejected" if risk_score > REJECT_THRESHOLD else "settled"
+        action = "reject" if decision == "rejected" else "settle"
+        settle_resp = await _post(
+            client,
+            f"{settings.ledger_url}/internal/transactions/{tx_id}/{action}",
+            {"risk_score": risk_score, "request_id": request_id},
+        )
+        if settle_resp.status_code >= 300:
+            raise UpstreamFailure(502, "ledger failed to finalise the transaction")
+        settled = settle_resp.json()
+
+        try:
+            await redis_client.xadd(
+                settings.stream_name,
+                {
+                    "transaction_id": tx_id,
+                    "account_id": created["account_id"],
+                    "kind": created["kind"],
+                    "amount_minor": str(created["amount_minor"]),
+                    "status": settled["status"],
+                    "risk_score": str(risk_score),
+                },
+            )
+        except Exception:
+            logger.exception(f"failed to publish transaction {tx_id} to the stream")
+
+        summary = {}
+        try:
+            summary_resp = await client.get(
+                f"{settings.insights_url}/internal/accounts/{created['account_id']}/quick_summary"
+            )
+            if summary_resp.status_code == 200:
+                summary = summary_resp.json()
+        except httpx.RequestError:
+            logger.exception(f"insights summary unavailable for request_id={request_id}")
+
+    elapsed_ms = (time.perf_counter() - start) * 1000
+    logger.info(f"finished transaction {tx_id} status={settled['status']} request_id={request_id}")
+
+    return {
+        "transaction_id": tx_id,
+        "status": settled["status"],
+        "risk_score": risk_score,
+        "balance_minor": settled["balance_minor"],
+        "summary": summary,
+        "request_id": request_id,
+        "elapsed_ms": elapsed_ms,
+    }

services/gateway/static/app.js

@@ -0,0 +1,149 @@
+const accountSelect = document.getElementById("account-select");
+const txForm = document.getElementById("tx-form");
+const resultJson = document.getElementById("result-json");
+const resultRtt = document.getElementById("result-rtt");
+const resultRequestId = document.getElementById("result-request-id");
+const burstButton = document.getElementById("burst-button");
+const burstN = document.getElementById("burst-n");
+const burstTableBody = document.querySelector("#burst-table tbody");
+const chaosFlagsDiv = document.getElementById("chaos-flags");
+const healthStrip = document.getElementById("health-strip");
+
+async function loadAccounts() {
+  const resp = await fetch("/api/accounts");
+  const accounts = await resp.json();
+  accountSelect.innerHTML = accounts
+    .slice(0, 200)
+    .map((a) => `<option value="${a.id}">${a.holder_name} (${a.currency})</option>`)
+    .join("");
+}
+
+txForm.addEventListener("submit", async (e) => {
+  e.preventDefault();
+  const payload = {
+    account_id: accountSelect.value,
+    kind: document.querySelector('input[name="kind"]:checked').value,
+    amount_minor: Number(document.getElementById("amount-input").value),
+    currency: "EUR",
+    merchant_name: document.getElementById("merchant-input").value,
+  };
+  const start = performance.now();
+  const resp = await fetch("/api/transactions", {
+    method: "POST",
+    headers: { "Content-Type": "application/json" },
+    body: JSON.stringify(payload),
+  });
+  const rtt = performance.now() - start;
+  const body = await resp.json();
+  resultRtt.textContent = rtt.toFixed(1);
+  resultRequestId.textContent = body.request_id || "--";
+  resultJson.textContent = JSON.stringify(body, null, 2);
+});
+
+function percentile(sorted, p) {
+  if (sorted.length === 0) return 0;
+  const rank = Math.max(1, Math.min(sorted.length, Math.ceil(p * sorted.length)));
+  return sorted[rank - 1];
+}
+
+burstButton.addEventListener("click", async () => {
+  const n = Number(burstN.value);
+  const accounts = Array.from(accountSelect.options).map((o) => o.value);
+  burstTableBody.innerHTML = `<tr><td colspan="8">running ${n} requests...</td></tr>`;
+
+  const latencies = [];
+  let success = 0;
+  let errors = 0;
+
+  await Promise.all(
+    Array.from({ length: n }, async () => {
+      const accountId = accounts[Math.floor(Math.random() * accounts.length)];
+      const start = performance.now();
+      try {
+        const resp = await fetch("/api/transactions", {
+          method: "POST",
+          headers: { "Content-Type": "application/json" },
+          body: JSON.stringify({
+            account_id: accountId,
+            kind: "purchase",
+            amount_minor: 500,
+            currency: "EUR",
+            merchant_name: "Load Burst",
+          }),
+        });
+        latencies.push(performance.now() - start);
+        if (resp.ok) success += 1;
+        else errors += 1;
+      } catch (e) {
+        latencies.push(performance.now() - start);
+        errors += 1;
+      }
+    })
+  );
+
+  latencies.sort((a, b) => a - b);
+  const mean = latencies.reduce((a, b) => a + b, 0) / latencies.length;
+  const row = [
+    latencies.length,
+    success,
+    errors,
+    latencies[0].toFixed(1),
+    mean.toFixed(1),
+    percentile(latencies, 0.5).toFixed(1),
+    percentile(latencies, 0.95).toFixed(1),
+    latencies[latencies.length - 1].toFixed(1),
+  ];
+  burstTableBody.innerHTML = `<tr>${row.map((v) => `<td>${v}</td>`).join("")}</tr>`;
+});
+
+async function loadChaos() {
+  const resp = await fetch("/api/admin/chaos");
+  const { flags } = await resp.json();
+  chaosFlagsDiv.innerHTML = Object.entries(flags)
+    .map(
+      ([flag, enabled]) => `
+      <label>
+        <input type="checkbox" data-flag="${flag}" ${enabled ? "checked" : ""}>
+        ${flag}
+      </label>`
+    )
+    .join("");
+
+  chaosFlagsDiv.querySelectorAll("input[type=checkbox]").forEach((el) => {
+    el.addEventListener("change", async () => {
+      await fetch("/api/admin/chaos", {
+        method: "POST",
+        headers: { "Content-Type": "application/json" },
+        body: JSON.stringify({ flag: el.dataset.flag, enabled: el.checked }),
+      });
+    });
+  });
+}
+
+document.querySelectorAll("#chaos-profiles button").forEach((btn) => {
+  btn.addEventListener("click", async () => {
+    await fetch("/api/admin/chaos/profile", {
+      method: "POST",
+      headers: { "Content-Type": "application/json" },
+      body: JSON.stringify({ profile: btn.dataset.profile }),
+    });
+    await loadChaos();
+  });
+});
+
+async function pollHealth() {
+  try {
+    const resp = await fetch("/api/health/all");
+    const { services } = await resp.json();
+    healthStrip.innerHTML = Object.entries(services)
+      .map(([name, status]) => `<span class="dot ${status === "ok" ? "green" : "red"}">${name}</span>`)
+      .join("");
+  } catch (e) {
+    // ignore transient polling failures
+  }
+}
+
+loadAccounts();
+loadChaos();
+pollHealth();
+setInterval(pollHealth, 5000);

services/gateway/static/index.html

@@ -0,0 +1,80 @@
+<!doctype html>
+<html lang="en">
+<head>
+  <meta charset="utf-8">
+  <title>Ledgerline</title>
+  <link rel="stylesheet" href="/static/style.css">
+</head>
+<body>
+  <header>
+    <h1>Ledgerline</h1>
+    <p class="tagline">a deliberately imperfect payments backend</p>
+  </header>
+
+  <main>
+    <section id="submit-panel">
+      <h2>Submit a transaction</h2>
+      <form id="tx-form">
+        <label>Account
+          <select id="account-select"></select>
+        </label>
+        <label>Amount (minor units)
+          <input type="number" id="amount-input" value="1000" min="0">
+        </label>
+        <label>Merchant
+          <input type="text" id="merchant-input" value="Corner Store">
+        </label>
+        <fieldset>
+          <legend>Kind</legend>
+          <label><input type="radio" name="kind" value="deposit" checked> deposit</label>
+          <label><input type="radio" name="kind" value="withdrawal"> withdrawal</label>
+          <label><input type="radio" name="kind" value="purchase"> purchase</label>
+        </fieldset>
+        <button type="submit">Submit</button>
+      </form>
+    </section>
+
+    <section id="result-panel">
+      <h2>Result</h2>
+      <p>Round-trip: <span id="result-rtt">--</span> ms &middot; request_id: <code id="result-request-id">--</code></p>
+      <pre id="result-json">(nothing yet)</pre>
+    </section>
+
+    <section id="load-panel">
+      <h2>Load burst</h2>
+      <label>N
+        <select id="burst-n">
+          <option value="10">10</option>
+          <option value="50">50</option>
+          <option value="200">200</option>
+        </select>
+      </label>
+      <button id="burst-button">Fire burst</button>
+      <table id="burst-table">
+        <thead>
+          <tr><th>count</th><th>success</th><th>errors</th><th>min</th><th>mean</th><th>p50</th><th>p95</th><th>max</th></tr>
+        </thead>
+        <tbody><tr><td colspan="8">no burst run yet</td></tr></tbody>
+      </table>
+    </section>
+
+    <section id="chaos-panel">
+      <h2>Chaos</h2>
+      <div id="chaos-flags"></div>
+      <div id="chaos-profiles">
+        <button data-profile="clean">clean</button>
+        <button data-profile="light">light</button>
+        <button data-profile="nightmare">nightmare</button>
+        <button data-profile="random">random</button>
+      </div>
+    </section>
+
+    <section id="health-panel">
+      <h2>Service health</h2>
+      <div id="health-strip"></div>
+    </section>
+  </main>
+
+  <script src="/static/app.js"></script>
+</body>
+</html>

services/gateway/static/style.css

@@ -0,0 +1,184 @@
+:root {
+  color-scheme: dark;
+  --bg: #0f1117;
+  --panel: #171a23;
+  --border: #2a2e3a;
+  --text: #d8dce6;
+  --muted: #7b8296;
+  --accent: #5fd0a8;
+  --danger: #e0685f;
+  --mono: "SF Mono", "Fira Code", Consolas, monospace;
+}
+
+* {
+  box-sizing: border-box;
+}
+
+body {
+  background: var(--bg);
+  color: var(--text);
+  font-family: var(--mono);
+  margin: 0;
+  padding: 0 1.5rem 3rem;
+}
+
+header {
+  padding: 2rem 0 1rem;
+  border-bottom: 1px solid var(--border);
+}
+
+header h1 {
+  margin: 0;
+  font-size: 1.6rem;
+  letter-spacing: 0.02em;
+}
+
+.tagline {
+  color: var(--muted);
+  margin: 0.25rem 0 0;
+  font-size: 0.85rem;
+}
+
+main {
+  display: grid;
+  grid-template-columns: repeat(auto-fit, minmax(320px, 1fr));
+  gap: 1.5rem;
+  margin-top: 1.5rem;
+}
+
+section {
+  background: var(--panel);
+  border: 1px solid var(--border);
+  border-radius: 6px;
+  padding: 1rem 1.25rem;
+}
+
+section h2 {
+  margin-top: 0;
+  font-size: 1rem;
+  color: var(--accent);
+  text-transform: uppercase;
+  letter-spacing: 0.05em;
+}
+
+label {
+  display: block;
+  margin-bottom: 0.6rem;
+  font-size: 0.85rem;
+  color: var(--muted);
+}
+
+input, select, button {
+  font-family: var(--mono);
+  background: #0c0e13;
+  color: var(--text);
+  border: 1px solid var(--border);
+  border-radius: 4px;
+  padding: 0.4rem 0.5rem;
+  margin-top: 0.2rem;
+  width: 100%;
+}
+
+fieldset {
+  border: 1px solid var(--border);
+  border-radius: 4px;
+  margin-bottom: 0.6rem;
+}
+
+fieldset label {
+  display: inline-block;
+  width: auto;
+  margin-right: 1rem;
+}
+
+button {
+  cursor: pointer;
+  background: var(--accent);
+  color: #0c0e13;
+  font-weight: 600;
+  border: none;
+  margin-top: 0.5rem;
+  transition: opacity 0.15s ease;
+}
+
+button:hover {
+  opacity: 0.85;
+}
+
+#chaos-profiles {
+  display: flex;
+  gap: 0.5rem;
+  margin-top: 0.75rem;
+}
+
+#chaos-profiles button {
+  width: auto;
+  flex: 1;
+}
+
+pre#result-json {
+  background: #0c0e13;
+  border: 1px solid var(--border);
+  border-radius: 4px;
+  padding: 0.75rem;
+  overflow-x: auto;
+  font-size: 0.8rem;
+  max-height: 260px;
+}
+
+code {
+  background: #0c0e13;
+  padding: 0.1rem 0.3rem;
+  border-radius: 3px;
+  user-select: all;
+}
+
+table {
+  width: 100%;
+  border-collapse: collapse;
+  margin-top: 0.75rem;
+  font-size: 0.8rem;
+}
+
+th, td {
+  border: 1px solid var(--border);
+  padding: 0.35rem 0.5rem;
+  text-align: right;
+}
+
+th:first-child, td:first-child {
+  text-align: left;
+}
+
+#chaos-flags label {
+  display: flex;
+  align-items: center;
+  gap: 0.4rem;
+}
+
+#chaos-flags input {
+  width: auto;
+}
+
+#health-strip {
+  display: flex;
+  flex-wrap: wrap;
+  gap: 0.5rem;
+}
+
+.dot {
+  border-radius: 999px;
+  padding: 0.25rem 0.6rem;
+  font-size: 0.75rem;
+  border: 1px solid var(--border);
+}
+
+.dot.green {
+  color: var(--accent);
+  border-color: var(--accent);
+}
+
+.dot.red {
+  color: var(--danger);
+  border-color: var(--danger);
+}

services/insights/analytics.py

@@ -0,0 +1,82 @@
+import sqlalchemy as sa
+
+# Deliberately the same shape of query the test suite's own EXPLAIN check runs
+# against the schema directly (tests/test_pathologies.py, 6.4) -- kept here
+# only for the honest (non-n_plus_one) path; that test doesn't call this code.
+
+_N_PLUS_ONE_ROW_CAP = 1500
+
+
+async def account_exists(session, account_id: str) -> bool:
+    result = await session.execute(sa.text("SELECT 1 FROM accounts WHERE id = :id"), {"id": account_id})
+    return result.first() is not None
+
+
+async def compute_summary(session, account_id: str, *, n_plus_one: bool) -> dict:
+    if n_plus_one:
+        rows = await _n_plus_one_summary(session, account_id)
+    else:
+        rows = await _joined_summary(session, account_id)
+
+    total_count = 0
+    total_amount = 0
+    by_category: dict = {}
+    for r in rows:
+        total_count += 1
+        total_amount += r["amount_minor"]
+        label = r["label"] or "uncategorized"
+        entry = by_category.setdefault(label, {"count": 0, "total_amount_minor": 0})
+        entry["count"] += 1
+        entry["total_amount_minor"] += r["amount_minor"]
+
+    return {
+        "account_id": account_id,
+        "window_days": 120,
+        "transaction_count": total_count,
+        "total_amount_minor": total_amount,
+        "by_category": by_category,
+    }
+
+
+async def _joined_summary(session, account_id: str) -> list[dict]:
+    """The honest path: a single JOIN, one statement."""
+    result = await session.execute(
+        sa.text(
+            "SELECT t.amount_minor, mc.label FROM transactions t "
+            "LEFT JOIN merchant_categories mc ON mc.id = t.category_id "
+            "WHERE t.account_id = :aid AND t.created_at >= now() - interval '120 days'"
+        ),
+        {"aid": account_id},
+    )
+    return [dict(row) for row in result.mappings().all()]
+
+
+async def _n_plus_one_summary(session, account_id: str) -> list[dict]:
+    """n_plus_one: instead of the JOIN above, fetch rows without one and issue
+    a separate category lookup per row -- phase-0-base-project-spec.md §7.
+    Capped so the pathology stays a genuine multiplication of statements
+    (see tests/test_pathologies.py 6.3) without turning a single request into
+    an unbounded number of round trips on a high-volume hot account."""
+    result = await session.execute(
+        sa.text(
+            "SELECT amount_minor, category_id FROM transactions "
+            "WHERE account_id = :aid AND created_at >= now() - interval '120 days' "
+            f"LIMIT {_N_PLUS_ONE_ROW_CAP}"
+        ),
+        {"aid": account_id},
+    )
+    tx_rows = result.mappings().all()
+
+    rows = []
+    for tx_row in tx_rows:
+        label = None
+        if tx_row["category_id"] is not None:
+            cat_result = await session.execute(
+                sa.text("SELECT label FROM merchant_categories WHERE id = :id"),
+                {"id": tx_row["category_id"]},
+            )
+            cat_row = cat_result.first()
+            if cat_row is not None:
+                label = cat_row.label
+        rows.append({"amount_minor": tx_row["amount_minor"], "label": label})
+    return rows

services/insights/app.py

@@ -0,0 +1,120 @@
+import asyncio
+import json
+import time
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+
+from common.chaos_flags import flags_fingerprint
+from common.logging_config import configure_logging
+from common.redis_client import redis_client
+from common.settings import settings
+
+import sqlalchemy as sa
+
+from . import chaos
+from .analytics import account_exists, compute_summary
+from .db import NormalSession, engine, session_factory
+
+SERVICE_NAME = "insights"
+logger = configure_logging(SERVICE_NAME)
+app = FastAPI(title=SERVICE_NAME)
+
+# Cache-stampede protection for the Redis-cached branch below: a burst of
+# concurrent requests for the same not-yet-cached (account, flags) key would
+# otherwise all pay the full computation cost at once instead of one paying
+# it and the rest waiting on the result. Not needed for unbounded_cache,
+# which never caches by design.
+_inflight_locks: dict[str, asyncio.Lock] = {}
+
+
+def _lock_for(key: str) -> asyncio.Lock:
+    lock = _inflight_locks.get(key)
+    if lock is None:
+        lock = asyncio.Lock()
+        _inflight_locks[key] = lock
+    return lock
+
+
+@app.get("/")
+async def root():
+    return {"service": SERVICE_NAME, "version": "0.1.0"}
+
+
+@app.get("/healthz")
+async def healthz(request: Request):
+    if request.headers.get("X-Ledgerline-Probe") == "chatty-health":
+        logger.info("received chatty-health healthz probe from gateway")
+    return {"status": "ok", "service": SERVICE_NAME}
+
+
+@app.get("/readyz")
+async def readyz():
+    try:
+        async with engine.connect() as conn:
+            await conn.exec_driver_sql("SELECT 1")
+    except Exception as exc:
+        logger.exception(f"readyz check failed: {exc}")
+        return JSONResponse(status_code=503, content={"status": "unavailable"})
+    return {"status": "ok"}
+
+
+@app.get("/internal/accounts/{account_id}/quick_summary")
+async def quick_summary(account_id: str):
+    """A cheap, account-primary-key-only lookup embedded in every transaction
+    response (phase-0-base-project-spec.md §6.1's "summary" field) -- deliberately
+    separate from the 120-day aggregation below (§6.2), which is the one where
+    n_plus_one and the missing index are meant to bite. That aggregation scans
+    `transactions` by account_id with no supporting index on purpose; embedding
+    it in every single transaction's response would make the missing-index
+    pathology fire on every request instead of only on the analytics endpoint."""
+    async with NormalSession() as session:
+        result = await session.execute(
+            sa.text("SELECT currency, balance_minor FROM accounts WHERE id = :id"),
+            {"id": account_id},
+        )
+        row = result.first()
+    if row is None:
+        return JSONResponse(status_code=404, content={"error": "account not found"})
+    return {"account_id": account_id, "currency": row.currency, "balance_minor": row.balance_minor}
+
+
+@app.get("/internal/accounts/{account_id}/summary")
+async def summary(account_id: str):
+    n_plus_one = await chaos.n_plus_one_enabled()
+    cold_cache = await chaos.cold_cache_enabled()
+    unbounded_cache = await chaos.unbounded_cache_enabled()
+    fingerprint = await flags_fingerprint()
+
+    # unbounded_cache means every single request is a guaranteed fresh miss
+    # needing a real connection -- under concurrent load that alone is enough
+    # demand to exhaust a small pool, without needing n_plus_one too (see
+    # services/insights/db.py).
+    starved = unbounded_cache
+
+    async with session_factory(starved) as session:
+        if not await account_exists(session, account_id):
+            return JSONResponse(status_code=404, content={"error": "account not found"})
+
+        if unbounded_cache:
+            # Broken on purpose: keyed by a timestamp that's unique per call,
+            # so this never hits -- it only ever grows. See chaos.py.
+            key = (account_id, fingerprint, time.time_ns())
+            result = await compute_summary(session, account_id, n_plus_one=n_plus_one)
+            chaos.UNBOUNDED_CACHE[key] = result
+            return result
+
+        ttl = settings.summary_cache_ttl_seconds_cold if cold_cache else settings.summary_cache_ttl_seconds
+        cache_key = f"summary:{account_id}:{fingerprint}"
+        cached = await redis_client.get(cache_key)
+        if cached is not None:
+            return json.loads(cached)
+
+    async with _lock_for(cache_key):
+        cached = await redis_client.get(cache_key)
+        if cached is not None:
+            return json.loads(cached)
+        async with session_factory(starved) as session:
+            result = await compute_summary(session, account_id, n_plus_one=n_plus_one)
+        await redis_client.setex(cache_key, ttl, json.dumps(result))
+        return result

services/insights/chaos.py

@@ -0,0 +1,22 @@
+"""n_plus_one, cold_cache, unbounded_cache -- phase-0-base-project-spec.md §7.
+All three are owned by `insights`."""
+from common.chaos_flags import is_enabled
+
+
+async def n_plus_one_enabled() -> bool:
+    return await is_enabled("n_plus_one")
+
+
+async def cold_cache_enabled() -> bool:
+    return await is_enabled("cold_cache")
+
+
+async def unbounded_cache_enabled() -> bool:
+    return await is_enabled("unbounded_cache")
+
+
+# unbounded_cache: memoize summaries in a module-level dict keyed by
+# (account_id, timestamp_ms) -- effectively never hits, never evicts, grows
+# forever. This is the pathology itself, not a fix for anything -- never
+# clear it, never cap it.
+UNBOUNDED_CACHE: dict = {}

services/insights/db.py

@@ -0,0 +1,25 @@
+"""insights gets its own engines, separate from common.db's shared default,
+so pool sizing here can be tuned independently of sanctions/notifier/audit
+(which also import common.db).
+
+Two pools, selected per-request: a generous default, and a small one used
+whenever unbounded_cache is on -- every request is then a guaranteed fresh
+miss needing a real connection, and under concurrent load that alone is
+enough demand to exhaust a small pool. This is a realistic capacity limit
+that emerges under load, not a 14th named chaos flag."""
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from common.db import make_engine
+
+normal_engine = make_engine(pool_size=20, max_overflow=10)
+starved_engine = make_engine(pool_size=5, max_overflow=3, pool_timeout=0.5)
+
+NormalSession = async_sessionmaker(normal_engine, expire_on_commit=False)
+StarvedSession = async_sessionmaker(starved_engine, expire_on_commit=False)
+
+# Exposed for readyz's connectivity check.
+engine = normal_engine
+
+
+def session_factory(starved: bool):
+    return StarvedSession() if starved else NormalSession()

services/ledger/app.py

@@ -0,0 +1,146 @@
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, field_validator
+
+from common.logging_config import configure_logging
+
+from . import chaos
+from .db import normal_engine, session_factory
+from . import repository
+
+SERVICE_NAME = "ledger"
+logger = configure_logging(SERVICE_NAME)
+app = FastAPI(title=SERVICE_NAME)
+
+
+class CreateTransactionRequest(BaseModel):
+    account_id: str
+    kind: str
+    amount_minor: int
+    currency: str
+    merchant_name: str
+    category_id: int | None = None
+    request_id: str = "unknown"
+
+    @field_validator("amount_minor")
+    @classmethod
+    def deposit_amount_must_be_non_negative(cls, v, info):
+        kind = info.data.get("kind")
+        if kind == "deposit" and v < 0:
+            raise ValueError("amount_minor must be non-negative for a deposit")
+        return v
+
+
+class SettleRequest(BaseModel):
+    risk_score: float
+    request_id: str = "unknown"
+
+
+@app.exception_handler(Exception)
+async def unhandled_exception_handler(request: Request, exc: Exception):
+    logger.exception(f"unhandled exception on {request.url.path}: {exc}")
+    return JSONResponse(status_code=500, content={"error": "internal error"})
+
+
+@app.get("/")
+async def root():
+    return {"service": SERVICE_NAME, "version": "0.1.0"}
+
+
+@app.get("/healthz")
+async def healthz(request: Request):
+    if request.headers.get("X-Ledgerline-Probe") == "chatty-health":
+        logger.info("received chatty-health healthz probe from gateway")
+    return {"status": "ok", "service": SERVICE_NAME}
+
+
+@app.get("/readyz")
+async def readyz():
+    try:
+        async with normal_engine.connect() as conn:
+            await conn.exec_driver_sql("SELECT 1")
+    except Exception as exc:
+        logger.exception(f"readyz check failed: {exc}")
+        return JSONResponse(status_code=503, content={"status": "unavailable"})
+    return {"status": "ok"}
+
+
+@app.get("/internal/accounts")
+async def list_accounts():
+    async with session_factory(starved=False) as session:
+        rows = await repository.list_accounts(session)
+    return [dict(row) for row in rows]
+
+
+@app.post("/internal/transactions")
+async def create_transaction(body: CreateTransactionRequest):
+    logger.info(f"processing tx for account {body.account_id}, amount {body.amount_minor} request_id={body.request_id}")
+
+    starved = await chaos.pool_starvation_enabled()
+    poison_enabled = await chaos.poison_data_enabled()
+    poison_tx_id = await chaos.find_poison_transaction_id(body.account_id)
+
+    async with session_factory(starved=starved) as session:
+        account = await repository.get_account(session, body.account_id)
+        if account is None:
+            return JSONResponse(status_code=404, content={"error": "account not found"})
+
+        if poison_tx_id is not None and poison_enabled:
+            poison_row = await repository.get_transaction_currency(session, poison_tx_id)
+            # poison_data: skip input validation on the seeded malformed rows
+            # (§8) -- this naive access raises on the NULL currency on purpose.
+            _ = poison_row.currency.upper()
+
+        tx_id = await repository.create_transaction(
+            session,
+            account_id=body.account_id,
+            kind=body.kind,
+            amount_minor=body.amount_minor,
+            currency=body.currency,
+            merchant_name=body.merchant_name,
+            category_id=body.category_id,
+        )
+
+    return JSONResponse(
+        status_code=201,
+        content={
+            "transaction_id": tx_id,
+            "account_id": body.account_id,
+            "holder_name": account.holder_name,
+            "kind": body.kind,
+            "amount_minor": body.amount_minor,
+            "category_id": body.category_id,
+            "status": "pending",
+        },
+    )
+
+
+@app.post("/internal/transactions/{tx_id}/settle")
+async def settle(tx_id: str, body: SettleRequest):
+    starved = await chaos.pool_starvation_enabled()
+    row_lock = await chaos.row_lock_contention_enabled()
+
+    async with session_factory(starved=starved) as session:
+        tx = await repository.get_transaction(session, tx_id)
+        if tx is None:
+            return JSONResponse(status_code=404, content={"error": "transaction not found"})
+        account = await repository.get_account(session, tx.account_id)
+        use_lock = row_lock and bool(account.is_hot)
+        balance = await repository.settle_transaction(session, tx=tx, risk_score=body.risk_score, use_row_lock=use_lock)
+
+    logger.info(f"settled transaction {tx_id} request_id={body.request_id}")
+    return {"status": "settled", "balance_minor": balance}
+
+
+@app.post("/internal/transactions/{tx_id}/reject")
+async def reject(tx_id: str, body: SettleRequest):
+    starved = await chaos.pool_starvation_enabled()
+
+    async with session_factory(starved=starved) as session:
+        tx = await repository.get_transaction(session, tx_id)
+        if tx is None:
+            return JSONResponse(status_code=404, content={"error": "transaction not found"})
+        balance = await repository.reject_transaction(session, tx=tx, risk_score=body.risk_score)
+
+    logger.info(f"rejected transaction {tx_id} request_id={body.request_id}")
+    return {"status": "rejected", "balance_minor": balance}

services/ledger/chaos.py

@@ -0,0 +1,28 @@
+"""pool_starvation, row_lock_contention, poison_data -- phase-0-base-project-spec.md
+§7. All three are owned by `ledger`.
+"""
+from common.chaos_flags import is_enabled
+from common.redis_client import redis_client
+
+# Populated once by scripts/seed.py: account_id -> id of that account's one
+# deliberately NULL-currency transaction (§8). A Redis hash lookup, not a
+# `transactions` scan by account_id -- that column has no supporting index
+# on purpose (§5), and the poison_data pathology shouldn't have to pay for a
+# full-table seq scan on every single request just to find its own trigger.
+POISON_TX_KEY = "poison:tx_by_account"
+
+
+async def pool_starvation_enabled() -> bool:
+    return await is_enabled("pool_starvation")
+
+
+async def row_lock_contention_enabled() -> bool:
+    return await is_enabled("row_lock_contention")
+
+
+async def poison_data_enabled() -> bool:
+    return await is_enabled("poison_data")
+
+
+async def find_poison_transaction_id(account_id: str) -> str | None:
+    return await redis_client.hget(POISON_TX_KEY, account_id)

services/ledger/db.py

@@ -0,0 +1,17 @@
+"""Two pre-built engines so pool_starvation (phase-0-base-project-spec.md §7)
+can be toggled live, with no restart: the flag picks which already-initialised
+pool a request borrows a connection from, rather than resizing a pool in
+place (SQLAlchemy pools aren't resizable live)."""
+from sqlalchemy.ext.asyncio import async_sessionmaker
+
+from common.db import make_engine
+
+normal_engine = make_engine(pool_size=20, max_overflow=10)
+starved_engine = make_engine(pool_size=2, max_overflow=0, pool_timeout=10)
+
+NormalSession = async_sessionmaker(normal_engine, expire_on_commit=False)
+StarvedSession = async_sessionmaker(starved_engine, expire_on_commit=False)
+
+
+def session_factory(starved: bool):
+    return StarvedSession() if starved else NormalSession()

services/ledger/repository.py

@@ -0,0 +1,122 @@
+import asyncio
+import uuid
+
+import sqlalchemy as sa
+
+KIND_SIGN = {"deposit": 1, "withdrawal": -1, "purchase": -1}
+
+ROW_LOCK_HOLD_SECONDS = 0.15
+
+
+async def get_account(session, account_id: str):
+    result = await session.execute(
+        sa.text("SELECT id, holder_name, currency, balance_minor, is_hot FROM accounts WHERE id = :id"),
+        {"id": account_id},
+    )
+    return result.first()
+
+
+async def list_accounts(session, limit: int = 1000):
+    result = await session.execute(
+        sa.text(
+            "SELECT id, holder_name, currency, balance_minor FROM accounts ORDER BY created_at LIMIT :limit"
+        ),
+        {"limit": limit},
+    )
+    return result.mappings().all()
+
+
+async def get_transaction_currency(session, tx_id: str):
+    """Backs the poison_data pathology: fetches one known-malformed transaction
+    (NULL currency, §8) by primary key -- a cheap, always-fast lookup regardless
+    of the deliberately-missing account_id index, since which transaction to
+    fetch is looked up by account_id in Redis first (chaos.py's
+    find_poison_transaction_id), not by scanning `transactions` here."""
+    result = await session.execute(
+        sa.text("SELECT currency FROM transactions WHERE id = :id"),
+        {"id": tx_id},
+    )
+    return result.first()
+
+
+async def create_transaction(
+    session, *, account_id, kind, amount_minor, currency, merchant_name, category_id
+):
+    tx_id = str(uuid.uuid4())
+    await session.execute(
+        sa.text(
+            "INSERT INTO transactions "
+            "(id, account_id, kind, amount_minor, currency, merchant_name, category_id, status) "
+            "VALUES (:id, :account_id, :kind, :amount_minor, :currency, :merchant_name, :category_id, 'pending')"
+        ),
+        {
+            "id": tx_id,
+            "account_id": account_id,
+            "kind": kind,
+            "amount_minor": amount_minor,
+            "currency": currency,
+            "merchant_name": merchant_name,
+            "category_id": category_id,
+        },
+    )
+    await session.commit()
+    return tx_id
+
+
+async def get_transaction(session, tx_id: str):
+    result = await session.execute(
+        sa.text(
+            "SELECT id, account_id, kind, amount_minor, currency, merchant_name, "
+            "category_id, status FROM transactions WHERE id = :id"
+        ),
+        {"id": tx_id},
+    )
+    return result.first()
+
+
+async def settle_transaction(session, *, tx, risk_score, use_row_lock: bool):
+    """Applies the balance delta and marks the transaction settled. When
+    use_row_lock is true (row_lock_contention on a hot account), takes
+    SELECT ... FOR UPDATE and holds the transaction open for an extra 150ms
+    before committing -- phase-0-base-project-spec.md §7."""
+    delta = KIND_SIGN[tx.kind] * tx.amount_minor
+
+    if use_row_lock:
+        result = await session.execute(
+            sa.text("SELECT balance_minor FROM accounts WHERE id = :id FOR UPDATE"),
+            {"id": tx.account_id},
+        )
+        result.first()
+        await session.execute(
+            sa.text("UPDATE accounts SET balance_minor = balance_minor + :delta WHERE id = :id"),
+            {"delta": delta, "id": tx.account_id},
+        )
+        await asyncio.sleep(ROW_LOCK_HOLD_SECONDS)
+    else:
+        await session.execute(
+            sa.text("UPDATE accounts SET balance_minor = balance_minor + :delta WHERE id = :id"),
+            {"delta": delta, "id": tx.account_id},
+        )
+
+    await session.execute(
+        sa.text("UPDATE transactions SET status = 'settled', risk_score = :risk_score WHERE id = :id"),
+        {"risk_score": risk_score, "id": tx.id},
+    )
+    await session.commit()
+
+    result = await session.execute(
+        sa.text("SELECT balance_minor FROM accounts WHERE id = :id"), {"id": tx.account_id}
+    )
+    return result.scalar_one()
+
+
+async def reject_transaction(session, *, tx, risk_score):
+    await session.execute(
+        sa.text("UPDATE transactions SET status = 'rejected', risk_score = :risk_score WHERE id = :id"),
+        {"risk_score": risk_score, "id": tx.id},
+    )
+    await session.commit()
+    result = await session.execute(
+        sa.text("SELECT balance_minor FROM accounts WHERE id = :id"), {"id": tx.account_id}
+    )
+    return result.scalar_one()

services/notifier/app.py

@@ -0,0 +1,49 @@
+from contextlib import asynccontextmanager
+
+import asyncio
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+
+from common.db import engine
+from common.logging_config import configure_logging
+from common.redis_client import redis_client
+
+from . import consumer
+
+SERVICE_NAME = "notifier"
+logger = configure_logging(SERVICE_NAME)
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    task = asyncio.create_task(consumer.run())
+    yield
+    task.cancel()
+
+
+app = FastAPI(title=SERVICE_NAME, lifespan=lifespan)
+
+
+@app.get("/")
+async def root():
+    return {"service": SERVICE_NAME, "version": "0.1.0"}
+
+
+@app.get("/healthz")
+async def healthz(request: Request):
+    if request.headers.get("X-Ledgerline-Probe") == "chatty-health":
+        logger.info("received chatty-health healthz probe from gateway")
+    return {"status": "ok", "service": SERVICE_NAME}
+
+
+@app.get("/readyz")
+async def readyz():
+    try:
+        async with engine.connect() as conn:
+            await conn.exec_driver_sql("SELECT 1")
+        await redis_client.ping()
+    except Exception as exc:
+        logger.exception(f"readyz check failed: {exc}")
+        return JSONResponse(status_code=503, content={"status": "unavailable"})
+    return {"status": "ok"}

services/notifier/chaos.py

@@ -0,0 +1,8 @@
+"""slow_consumer -- phase-0-base-project-spec.md §7. Owned by `notifier`."""
+from common.chaos_flags import is_enabled
+
+SLEEP_SECONDS = 0.25
+
+
+async def slow_consumer_enabled() -> bool:
+    return await is_enabled("slow_consumer")

services/notifier/consumer.py

@@ -0,0 +1,78 @@
+import asyncio
+import logging
+import random
+import uuid
+
+import sqlalchemy as sa
+
+from common.db import Session
+from common.redis_client import redis_client
+from common.settings import settings
+
+from . import chaos
+
+logger = logging.getLogger("notifier")
+
+GROUP = "notifier"
+CONSUMER = "notifier-1"
+
+
+async def ensure_group():
+    try:
+        await redis_client.xgroup_create(settings.stream_name, GROUP, id="$", mkstream=True)
+    except Exception as exc:
+        if "BUSYGROUP" not in str(exc):
+            raise
+
+
+async def _handle(fields: dict):
+    if await chaos.slow_consumer_enabled():
+        await asyncio.sleep(chaos.SLEEP_SECONDS)
+
+    async with Session() as session:
+        await session.execute(
+            sa.text(
+                "INSERT INTO notifications (id, transaction_id, channel, status, attempts, sent_at) "
+                "VALUES (:id, :tx_id, :channel, 'sent', 1, now())"
+            ),
+            {
+                "id": str(uuid.uuid4()),
+                "tx_id": fields.get("transaction_id"),
+                "channel": random.choice(["email", "push", "sms"]),
+            },
+        )
+        await session.commit()
+
+
+async def run():
+    """Background consumer loop for the `notifier` consumer group -- a second,
+    independent group (`audit`) reads the same stream in services/audit/worker.py.
+    Redis Streams carry no automatic context propagation, deliberately."""
+    await ensure_group()
+    logger.info("notifier consumer started")
+    while True:
+        try:
+            resp = await redis_client.xreadgroup(
+                groupname=GROUP,
+                consumername=CONSUMER,
+                streams={settings.stream_name: ">"},
+                count=10,
+                block=2000,
+            )
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            logger.exception("notifier consumer read failed")
+            await asyncio.sleep(1)
+            continue
+
+        if not resp:
+            continue
+
+        for _, messages in resp:
+            for msg_id, fields in messages:
+                try:
+                    await _handle(fields)
+                except Exception:
+                    logger.exception(f"notifier failed to process message {msg_id}")
+                await redis_client.xack(settings.stream_name, GROUP, msg_id)

services/sanctions/app.py

@@ -0,0 +1,63 @@
+import asyncio
+
+from fastapi import FastAPI, Request
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel
+
+from common.db import engine
+from common.logging_config import configure_logging
+from common.redis_client import redis_client
+
+from . import chaos
+from .repository import is_match
+
+SERVICE_NAME = "sanctions"
+logger = configure_logging(SERVICE_NAME)
+app = FastAPI(title=SERVICE_NAME)
+
+
+class LookupRequest(BaseModel):
+    holder_name: str
+    request_id: str = "unknown"
+
+
+@app.get("/")
+async def root():
+    return {"service": SERVICE_NAME, "version": "0.1.0"}
+
+
+@app.get("/healthz")
+async def healthz(request: Request):
+    if request.headers.get("X-Ledgerline-Probe") == "chatty-health":
+        logger.info(f"received chatty-health healthz probe from gateway")
+    return {"status": "ok", "service": SERVICE_NAME}
+
+
+@app.get("/readyz")
+async def readyz():
+    try:
+        async with engine.connect() as conn:
+            await conn.exec_driver_sql("SELECT 1")
+    except Exception as exc:
+        logger.exception(f"readyz check failed: {exc}")
+        return JSONResponse(status_code=503, content={"status": "unavailable"})
+    return {"status": "ok"}
+
+
+@app.post("/internal/lookup")
+async def lookup(body: LookupRequest):
+    logger.info(f"lookup for holder_name={body.holder_name!r} request_id={body.request_id}")
+
+    if await chaos.sanctions_errors_enabled() and chaos.should_error():
+        logger.info(f"sanctions_errors chaos triggered for request_id={body.request_id}")
+        return JSONResponse(status_code=503, content={"error": "sanctions vendor unavailable"})
+
+    if await chaos.slow_sanctions_enabled():
+        delay = chaos.sample_delay_seconds()
+        logger.info(f"slow_sanctions chaos sleeping {delay:.3f}s for request_id={body.request_id}")
+        await asyncio.sleep(delay)
+
+    match = await is_match(body.holder_name)
+    if match:
+        return {"matched": True, **match}
+    return {"matched": False, "country": None, "list_source": None}

services/sanctions/chaos.py

@@ -0,0 +1,37 @@
+"""slow_sanctions and sanctions_errors -- phase-0-base-project-spec.md §7.
+
+Both flags belong to `sanctions`, the fake third-party watchlist vendor.
+"""
+import math
+import random
+
+from common.chaos_flags import is_enabled
+
+_MEDIAN_SECONDS = 0.04
+_P99_SECONDS = 1.8
+_MU = math.log(_MEDIAN_SECONDS)
+_Z99 = 2.326347874
+_SIGMA = (math.log(_P99_SECONDS) - _MU) / _Z99
+_TIMEOUT_TRIGGER_PROBABILITY = 0.015
+_TIMEOUT_TRIGGER_SECONDS = 5.0
+_ERROR_PROBABILITY = 0.03
+
+
+async def slow_sanctions_enabled() -> bool:
+    return await is_enabled("slow_sanctions")
+
+
+async def sanctions_errors_enabled() -> bool:
+    return await is_enabled("sanctions_errors")
+
+
+def sample_delay_seconds() -> float:
+    """Log-normal delay: median ~40ms, p99 ~1.8s. 1.5% of calls sleep 5s,
+    which exceeds the caller's 3s downstream timeout and surfaces as one."""
+    if random.random() < _TIMEOUT_TRIGGER_PROBABILITY:
+        return _TIMEOUT_TRIGGER_SECONDS
+    return random.lognormvariate(_MU, _SIGMA)
+
+
+def should_error() -> bool:
+    return random.random() < _ERROR_PROBABILITY

services/sanctions/repository.py

@@ -0,0 +1,19 @@
+import sqlalchemy as sa
+
+from common.db import Session
+
+
+async def is_match(holder_name: str) -> dict | None:
+    """Case-insensitive exact match against sanctions_entries.normalized_name."""
+    async with Session() as session:
+        result = await session.execute(
+            sa.text(
+                "SELECT country, list_source FROM sanctions_entries "
+                "WHERE lower(normalized_name) = lower(:name) LIMIT 1"
+            ),
+            {"name": holder_name},
+        )
+        row = result.first()
+    if row is None:
+        return None
+    return {"country": row.country, "list_source": row.list_source}