Tripwire — streaming fraud detection pipeline

E-commerce transactions flow through Kafka, a deterministic rules service flags suspicious ones, and (eventually) an LLM agent investigates flagged cases and writes an auditable verdict.

POST /transactions ─▶ [transactions] ─▶ rules consumer ─┬─ no flags ─▶ Postgres (verdict: pass)
                                                        └─ flags ────▶ [flagged-transactions] ─▶ agent worker ─▶ Postgres (verdict: escalate)

Invariant: nothing talks to anything directly — services communicate via topics or the DB. The LLM gate is pipeline topology, not an agent decision: only flagged traffic reaches the agent.

Progress

#	Step	Status
1	Skeleton day — compose (Kafka + Postgres), gateway produces, amount > 5000 rule, stub agent, cases table	done
2	Real schema + seeded customers; idempotent upsert on txn_id	next
3	Full rules set (velocity, geo, device) emitting flags + score	todo
4	Agent v1: LLM call, no tools — reads flags, writes a report	todo
5	Agent v2: tools (history, profile lookup) + audit logging	todo
6	UI: cases list, case detail, fire-test-transaction button	todo
7	Polish: architecture diagram, demo script, ADR notes	todo

What step 1 actually built

• Gateway (gateway.py) — FastAPI, validates the transaction schema, produces to transactions keyed by user_id so one user's transactions land on one partition (this is what makes the in-memory velocity counter in step 3 correct without Redis). Invalid payloads 422 and produce nothing.
• Rules (rules.py) — v1 is a single check, amount > 5000 → ["high_amount"]. Returns evidence, never a verdict.
• Rules consumer (rules_consumer.py) — clean transactions auto-pass straight to Postgres; flagged ones are forwarded to flagged-transactions with their flags attached.
• Agent worker (agent.py) — day-1 stub, writes verdict: escalate for everything it sees.
• Persistence — one cases table (schema.sql), loaded by Postgres on first boot.
• Delivery — consumers commit offsets only after the handler returns (at-least-once).

Verified end to end: three transactions fired at the running stack produced t-small → pass, t-big → escalate, and a 422 for the malformed one, with consumer lag back to 0.

Deviation from the MVP doc: bitnami/kafka images were pulled from Docker Hub, so compose uses apache/kafka:3.9.0 (KRaft mode, 3 partitions).

Setup

python3 -m venv .venv && .venv/bin/pip install -r requirements.txt
direnv allow          # activates .venv and exports TRIPWIRE_KAFKA / TRIPWIRE_DSN on cd

.envrc puts .venv/bin on PATH, so python, pytest, and uvicorn below need no prefix. Without direnv, prefix them with .venv/bin/ and the defaults in kafka_io.py / store.py still point at localhost. Secrets (e.g. ANTHROPIC_API_KEY in step 4) go in .env — gitignored, and .envrc loads it if present.

Run it

docker compose up -d                       # Kafka + Postgres (schema.sql runs on first boot)

uvicorn tripwire.run_gateway:app --port 8000 &
python -m tripwire.run_rules &
python -m tripwire.run_agent &

Fire a transaction and read the case:

curl -X POST localhost:8000/transactions -H 'content-type: application/json' \
  -d '{"txn_id":"t1","user_id":"u1","amount":9000,"merchant":"acme","card_last4":"1234",
       "ip":"1.2.3.4","device_id":"d1","ts":"2026-08-15T00:00:00Z"}'

docker compose exec postgres psql -U tripwire -c 'SELECT * FROM cases;'

Amounts over 5000 come back escalate; anything else pass.

Tests

pytest

Built test-first. Handlers take their producer and store as arguments, so the unit tests drive real handler code against recording fakes — no Kafka or Postgres needed. The adapters in kafka_io.py and store.py are the only pieces that require the stack running; they're verified by running it.

Layout

File	Role
tripwire/gateway.py	FastAPI app; validates and produces keyed by user_id
tripwire/rules.py	Deterministic checks → flags (evidence, never a verdict)
tripwire/rules_consumer.py	Routes: clean → Postgres, flagged → flagged-transactions
tripwire/agent.py	Day-1 stub verdict writer
tripwire/kafka_io.py, tripwire/store.py	Kafka / Postgres adapters
tripwire/run_*.py	Entrypoints wiring adapters into handlers
schema.sql, docker-compose.yml	cases table; Kafka + Postgres for local dev