Phase 2: answers to the Phase 1 judge feedback (11 workstreams)

.gitignore

@@ -18,6 +18,7 @@ out/
 .env
 .env.*
 !.env.example
+backend/secrets/
 
 # OS / tooling
 .DS_Store

Makefile

@@ -1,7 +1,7 @@
-.PHONY: help setup up redis down data features train policy insights intel pipeline test lint fmt \
+.PHONY: help setup up redis down data data-calibrated features train policy insights mitigation intel intel-compare adversary pipeline test lint fmt \
 	migrate seed load api replay replay-live verify platform \
-	review retrain shadow models promote console console-build smoke \
-	demo demo-reset
+	review retrain shadow models promote console console-build smoke label-realism e2e-phone \
+	demo demo-reset external
 
 API_PORT ?= 8010
 API_URL ?= http://127.0.0.1:$(API_PORT)
@@ -40,6 +40,9 @@ down: ## Stop everything (the demo too) and keep the data
 data: ## Generate the synthetic dataset
 	cd backend && uv run python -m fraudlens.simulator.generate
 
+data-calibrated: ## Generate the Bangladesh-calibrated dataset into data/full_calibrated
+	cd backend && uv run python -m fraudlens.simulator.generate --profile calibrated
+
 features: ## Replay the dataset through the feature engine
 	cd backend && uv run python -m fraudlens.features.build
 
@@ -52,10 +55,28 @@ policy: ## Evaluate the decision policy and build the similar-case index
 insights: ## Threshold sweep, drift and fairness tables for the dashboards
 	cd backend && uv run python -m fraudlens.decision.insights
 
+mitigation: ## Fit and measure young-wallet thresholds (policy v3) against v2 (docs/DECISION_POLICY.md §11)
+	cd backend && uv run python -m fraudlens.decision.mitigation
+
 intel: ## Train and evaluate the scam-message classifier (docs/FRAUD_TAXONOMY.md)
 	cd backend && uv run python -m fraudlens.intel.train
 
-pipeline: data features train policy insights intel ## Data, features, models and policy end to end
+intel-compare: ## Compare message models, MiniLM included with the transformer extra (compare.json)
+	cd backend && uv run --extra transformer python -m fraudlens.intel.compare
+
+adversary: ## Adaptive scammers vs frozen, retrained and drift-gated models (after `make pipeline`)
+	cd backend && uv run python -m fraudlens.models.adversary
+
+label-realism: ## Retrain on reported-only labels, and with recovery methods; compare on ground truth
+	cd backend && uv run python -m fraudlens.models.label_realism
+
+pipeline: data features train policy insights mitigation intel ## Data, features, models and policy end to end
+
+external: ## Public datasets (PaySim, BAF) and bootstrap intervals; downloads ~560 MB once (docs/MODEL_CARD.md)
+	cd backend && uv run python -m fraudlens.external.download
+	cd backend && uv run python -m fraudlens.external.paysim
+	cd backend && uv run python -m fraudlens.external.baf
+	cd backend && uv run python -m fraudlens.models.bootstrap
 
 migrate: ## Bring the database schema up to date
 	cd backend && uv run alembic upgrade head
@@ -69,6 +90,12 @@ load: ## Load the historical period into the database (replaces what is there)
 api: ## Run the API and its stream worker
 	cd backend && uv run uvicorn fraudlens.api:create_app --factory --host 127.0.0.1 --port $(API_PORT)
 
+worker: ## Run one more stream worker next to the API; start as many as needed (docs/SCALING.md)
+	cd backend && uv run python -m fraudlens.platform.worker --port $${PORT:-8081}
+
+loadtest: ## Throughput and latency for 1, 2 and 4 workers; RESETS the database (docs/SCALING.md)
+	cd backend && uv run python -m fraudlens.platform.loadtest --workers 1 2 4
+
 replay: ## Replay the test period, except its last day, through the event stream (needs `make api`)
 	cd backend && uv run python -m fraudlens.platform.replay --via stream --to-day 118
 
@@ -104,13 +131,34 @@ console-build: ## Production build of the console
 smoke: ## Open every console page in a headless browser as each role (needs the API and the console)
 	cd frontend && pnpm exec playwright install chromium-headless-shell && node scripts/smoke.cjs
 
+e2e-phone: ## The customer phone in Bangla and English, and an appeal round trip (needs the API and the console)
+	cd frontend && pnpm exec playwright install chromium-headless-shell && node scripts/phone-e2e.cjs
+
 test: ## Run backend tests (the platform tests need `make up`)
 	cd backend && uv run pytest -q
 
 lint: ## Lint and format check
 	cd backend && uv run ruff check src tests && uv run ruff format --check src tests
 	cd frontend && pnpm exec tsc --noEmit && pnpm exec eslint .
 
+.PHONY: tls rotate-jwt partner-key keys
+JWT_KEYRING ?= secrets/jwt.json
+INGEST_KEYRING ?= secrets/ingest.json
+
+tls: backend/.env ## The demo behind HTTPS with HSTS on https://localhost:8443 (deploy/Caddyfile)
+	docker compose --profile demo --profile tls up --build
+
+rotate-jwt: ## New token-signing key; sessions signed with the old one last until they expire
+	cd backend && mkdir -p -m 700 secrets && uv run python -m fraudlens.platform.keys rotate-jwt $(JWT_KEYRING)
+
+partner-key: ## New ingest HMAC key: make partner-key PARTNER=upay [CALLBACK=https://...]
+	cd backend && mkdir -p -m 700 secrets && uv run python -m fraudlens.platform.keys rotate-partner \
+		$(INGEST_KEYRING) --partner $(PARTNER) $(if $(CALLBACK),--callback-url $(CALLBACK))
+
+keys: ## List the key ids in both keyrings (never the secrets)
+	cd backend && for f in $(JWT_KEYRING) $(INGEST_KEYRING); do \
+		[ -f $$f ] && uv run python -m fraudlens.platform.keys list $$f; done; true
+
 fmt: ## Auto-format
 	cd backend && uv run ruff check --fix src tests && uv run ruff format src tests
 

README.md

@@ -426,12 +426,13 @@ To run a challenger in shadow mode, set `FRAUDLENS_SHADOW_MODEL_VERSION` to its
 <summary><b>Check it</b></summary>
 
 ```bash
-make test     # 227 backend tests; the platform tests need `make up`
+make test     # 230 backend tests; the platform tests need `make up`
 make lint     # ruff, tsc, eslint
 make smoke    # opens every console page as each role in a headless browser
+make e2e-phone  # the customer phone in Bangla and English, plus an appeal round trip
 ```
 
-`make smoke` needs the demo (or `make api` and `make console`) running and `make setup` done. CI (`.github/workflows/ci.yml`) runs the tests and the lint against real Postgres and Redis, builds the console and builds both images.
+`make smoke` and `make e2e-phone` need the demo (or `make api` and `make console`) running and `make setup` done. CI (`.github/workflows/ci.yml`) runs the tests and the lint against real Postgres and Redis, builds the console and builds both images.
 
 </details>
 
@@ -462,14 +463,18 @@ Keep two browser windows open: one as `analyst1`, one private window as `supervi
 | [DATA_ASSUMPTIONS.md](docs/DATA_ASSUMPTIONS.md) | What the synthetic world contains and what it leaves out |
 | [MODEL_CARD.md](docs/MODEL_CARD.md) | Models, results, ablations, fairness, limits |
 | [DECISION_POLICY.md](docs/DECISION_POLICY.md) | Tiers, rules, thresholds, explanations, the language model's role |
+| [BUSINESS_CASE.md](docs/BUSINESS_CASE.md) | The threshold sweep as a monthly P&L in taka, assumptions and sensitivity |
 | [PLATFORM.md](docs/PLATFORM.md) | API, workflow, security, stream, measured latency |
+| [SCALING.md](docs/SCALING.md) | Several stream workers in one consumer group, measured; Kubernetes manifests in `deploy/` |
 | [FRAUD_TAXONOMY.md](docs/FRAUD_TAXONOMY.md) | Eight kinds of fraud in the Bangladesh context, what detects each, the scam-message classifier and its limits |
 | [DEMO_SCRIPT.md](docs/DEMO_SCRIPT.md) | The walk-through |
+| [INGEST.md](docs/INGEST.md) | The signed webhook for core banking: ISO 20022 pacs.008 mapping, HMAC signing, replay protection, decision callbacks, a partner SDK |
+| [SECURITY.md](docs/SECURITY.md) | STRIDE threat model with the status of each mitigation, personal data and retention |
 
 ## ⚠️ What this is not
 
 - **Everything runs on synthetic data,** so absolute numbers will not transfer to real traffic. Fraud is more common and more scripted here than in real life, and thresholds must be re-fitted on real data.
-- **It is a prototype, not a deployment.** One scorer process holds the feature state, and there is no TLS, secrets manager or horizontal scaling.
+- **It is a prototype, not a deployment.** Stream workers scale out, but every one holds the whole feature state and events enter it one at a time, so throughput stops growing at a few workers ([SCALING.md](docs/SCALING.md)). TLS ends at a Caddy proxy, secrets come from files with no vault client, and keys rotate by command, not on a schedule ([SECURITY.md](docs/SECURITY.md)).
 - **The demo scaffolding is not the product.** The demo accounts, the phone demo endpoints and the review simulator exist only outside production mode.
 - **No model output moves or blocks money on its own.** A hold waits for an analyst and a freeze needs two people.
 - **The Bangla texts were written by the developers** and have not been reviewed by a professional translator.

backend/alembic/versions/0002_shadow_scores.py

@@ -28,7 +28,10 @@ def upgrade() -> None:
         sa.Column("tier", sa.String(length=8), nullable=False),
         sa.Column("latency_ms", sa.Float(), nullable=True),
         sa.Column(
-            "created_at", sa.DateTime(timezone=True), server_default=sa.text("now()"), nullable=False
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
         ),
         sa.CheckConstraint(
             "tier IN ('allow', 'warn', 'step_up', 'hold')", name=op.f("ck_shadow_scores_tier")

backend/alembic/versions/0004_customer_appeals.py

@@ -0,0 +1,78 @@
+"""customer appeals
+
+Revision ID: 0004_appeals
+Revises: 0003
+Create Date: 2026-10-07 12:00:00.000000
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "0004_appeals"
+down_revision: str | Sequence[str] | None = "0003"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    op.create_table(
+        "appeals",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("txn_id", sa.BigInteger(), nullable=False),
+        sa.Column("wallet_id", sa.String(length=32), nullable=False),
+        sa.Column("case_id", sa.Integer(), nullable=True),
+        sa.Column("tier", sa.String(length=8), nullable=False),
+        sa.Column("relation", sa.String(length=16), nullable=False),
+        sa.Column("reason", sa.Text(), nullable=False),
+        sa.Column("status", sa.String(length=12), server_default="pending", nullable=False),
+        sa.Column("filed_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("sla_due_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("decided_by", sa.Integer(), nullable=True),
+        sa.Column("decision_note", sa.Text(), nullable=True),
+        sa.Column("decided_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.CheckConstraint(
+            "(status = 'pending') = (decided_by IS NULL)",
+            name=op.f("ck_appeals_decided_has_decider"),
+        ),
+        sa.CheckConstraint(
+            "relation IN ('family', 'friend', 'business', 'seller', 'landlord', 'employer', "
+            "'other', 'none')",
+            name=op.f("ck_appeals_relation"),
+        ),
+        sa.CheckConstraint(
+            "status IN ('pending', 'approved', 'rejected')", name=op.f("ck_appeals_status")
+        ),
+        sa.CheckConstraint("tier IN ('warn', 'step_up', 'hold')", name=op.f("ck_appeals_tier")),
+        sa.ForeignKeyConstraint(["case_id"], ["cases.id"], name=op.f("fk_appeals_case_id_cases")),
+        sa.ForeignKeyConstraint(
+            ["decided_by"], ["users.id"], name=op.f("fk_appeals_decided_by_users")
+        ),
+        sa.ForeignKeyConstraint(
+            ["txn_id"],
+            ["transactions.txn_id"],
+            name=op.f("fk_appeals_txn_id_transactions"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_appeals")),
+        sa.UniqueConstraint("txn_id", name=op.f("uq_appeals_txn_id")),
+    )
+    op.create_index("ix_appeals_status_sla", "appeals", ["status", "sla_due_at"], unique=False)
+    op.create_index(op.f("ix_appeals_wallet_id"), "appeals", ["wallet_id"], unique=False)
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    op.drop_index(op.f("ix_appeals_wallet_id"), table_name="appeals")
+    op.drop_index("ix_appeals_status_sla", table_name="appeals")
+    op.drop_table("appeals")

backend/alembic/versions/0005_pending_decisions.py

@@ -0,0 +1,44 @@
+"""pending decisions
+
+Revision ID: 0005
+Revises: 0004_appeals
+Create Date: 2026-10-07 09:30:00.000000
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = "0005"
+down_revision: str | Sequence[str] | None = "0004_appeals"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    op.create_table(
+        "pending_decisions",
+        sa.Column("txn_id", sa.BigInteger(), nullable=False),
+        sa.Column("features", postgresql.ARRAY(sa.Float(precision=53)), nullable=False),
+        sa.Column("context", postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False
+        ),
+        sa.ForeignKeyConstraint(
+            ["txn_id"],
+            ["transactions.txn_id"],
+            name=op.f("fk_pending_decisions_txn_id_transactions"),
+            ondelete="CASCADE",
+        ),
+        sa.PrimaryKeyConstraint("txn_id", name=op.f("pk_pending_decisions")),
+    )
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    op.drop_table("pending_decisions")

backend/pyproject.toml

@@ -27,6 +27,11 @@ dependencies = [
     "uvicorn[standard]>=0.54.0",
 ]
 
+[project.optional-dependencies]
+transformer = [
+    "sentence-transformers>=3.0",
+]
+
 [build-system]
 requires = ["uv_build>=0.12.0,<0.13.0"]
 build-backend = "uv_build"

backend/src/fraudlens/api/app.py

@@ -4,6 +4,7 @@
 
 import logging
 import os
+import threading
 from contextlib import asynccontextmanager
 
 import redis.asyncio as aioredis
@@ -19,14 +20,29 @@
 from ..config import Settings
 from ..intel.text import load_text_model
 from ..platform.audit import WorkflowError
+from ..platform.callbacks import Dispatcher
 from ..platform.db import make_engine, make_sessions
 from ..platform.graph import Graph
 from ..platform.scoring import Scorer
 from ..platform.security import RateLimiter, check_production
 from ..platform.stream import Worker
 from .deps import Platform
 from .middleware import RequestContext, error_body
-from .routes import alerts, auth, cases, customer, demo, intel, mlops, network, ops, scoring
+from .routes import (
+    alerts,
+    appeals,
+    auth,
+    cases,
+    consortium,
+    customer,
+    demo,
+    ingest,
+    intel,
+    mlops,
+    network,
+    ops,
+    scoring,
+)
 
 log = logging.getLogger(__name__)
 
@@ -79,6 +95,7 @@ def build_platform(settings: Settings) -> Platform:
         message_limit=RateLimiter(redis, "message-check", 20, 60),
         # Tighter: each call is a question about the ledger.
         proof_limit=RateLimiter(redis, "payment-verify", 10, 60),
+        callbacks=Dispatcher(settings, sessions, redis) if settings.ingest_keyring else None,
     )
 
 
@@ -96,10 +113,24 @@ async def lifespan(app: FastAPI):
         app.state.aredis = aioredis.from_url(settings.redis_url, decode_responses=True)
         if settings.run_worker:
             platform.worker.start()
+            if platform.callbacks is not None:
+                platform.callbacks.start()
+        # Stream workers in other processes change the state too: follow the log.
+        following = threading.Event()
+        if settings.follow_interval_s > 0:
+            threading.Thread(
+                target=platform.scorer.follow,
+                args=(following, settings.follow_interval_s),
+                name="fraudlens-follower",
+                daemon=True,
+            ).start()
         log.info("fraudlens api ready: scoring in %s mode", platform.scorer.mode)
         try:
             yield
         finally:
+            if platform.callbacks is not None:
+                platform.callbacks.stop()
+            following.set()
             platform.worker.stop()
             if platform.worker.is_alive():
                 platform.worker.join(timeout=5.0)
@@ -165,7 +196,20 @@ async def unavailable(request: Request, exc: Exception) -> JSONResponse:
     for kind in (OperationalError, InterfaceError, RedisError):
         app.add_exception_handler(kind, unavailable)
 
-    for module in (auth, scoring, alerts, cases, network, customer, intel, ops, mlops):
+    for module in (
+        auth,
+        scoring,
+        ingest,
+        alerts,
+        cases,
+        appeals,
+        network,
+        customer,
+        intel,
+        ops,
+        mlops,
+        consortium,
+    ):
         app.include_router(module.router, prefix="/v1")
     if not settings.production:  # stand-ins for the customer app; see routes/demo.py
         app.include_router(demo.router, prefix="/v1")