Backend Overview

kxinspect_backend_python is a contract-frozen FastAPI mock that implements every rule in docs/contract-v1.md without pretending to be a production service. It exists so the Flutter app can be developed and reviewed against a real HTTP + SSE surface, while also producing the canonical fixture bundle the frontend ships.

Design tenets

Tenet	How it shows up
Contract is the API	openapi-v1.json is frozen; scripts/export_openapi.py --check fails on drift
Pure domain	app/domain/* imports no FastAPI/Pydantic/IO/clock — replayable from Dart
One atomic commit	JsonStore.commit() + os.replace is the only durability primitive
Test clock, not wall clock	Clock abstraction + ManualClock makes every test deterministic
Fixture parity	scripts/export_fixtures.py produces the bundle Flutter commits verbatim

Capabilities

• Reads: health, bookings, GET /bookings/{id}/hub snapshot, inventory reports, inspections, charges, notifications, GET /sync-snapshot cursor
• Commands: POST /charges/{id}/accept, .../contest (multipart), .../pay, POST /tasks, POST /notifications/{id}/read
• Attachments: staged pre-commit, SHA-256 verified, per-file 10 MiB / total 25 MiB / 5 files max, GET /attachments/{id}
• SSE: GET /api/v1/events with Last-Event-ID replay, ring buffer 100, heartbeat 15 s
• Deadline reconciliation: runs before every charge/Hub read, before every mutation, and on a scheduler
• Idempotency: (method, route template, resource id, Idempotency-Key) scoped, RFC 8785 digest, ledger survives restart
• Chaos: latency + probabilistic 429 injection behind config for frontend resilience testing
• Dev routes (/_dev/*): reset/reseed/operator actions, token-gated even on loopback

Non-goals & hard limits

Explicitly in contract-v1.md §15:

• No authentication, no payment provider, no cloud storage, no malware scanning
• Single Uvicorn worker only (KX_WORKERS=1 else startup raises)
• Loopback by default, CORS * rejected, body ceiling MAX_REQUEST_BYTES = 25 MiB + 1 MiB
• fcntl advisory lock — POSIX host required; multi-process would silently diverge so it fails fast
• Every error is envelope-bounded; attachment bytes + SSE stream are the only non-JSON responses

Versions & anchors

schemaVersion   1
contractVersion 1  (API prefix /api/v1)
seedVersion     1
referenceNow    2026-08-01T12:00:00Z  (en-GB / Europe/London)
python          >=3.13,<3.14  (.python-version)
fastapi         0.141.1  | pydantic 2.13.4 | uvicorn 0.52.1
storage         runtime/state.json (atomically replaced) + runtime/uploads/

Layout at a glance

flowchart TD
  main[app/main.py\nbuild_app factory] --> api[app/api/v1]
  api --> svc[app/services]
  svc --> domain[app/domain\npure]
  svc --> db[app/db\nJsonStore]
  svc --> core[app/core\nclock, ids, graphemes, canonical_json]
  db --> seed[app/seed/*.json]
  db --> runtime[(runtime/state.json)]
  svc --> bus[EventBroker\nSSE]
  api --> schemas[app/schemas\nStrictModel]

:::tip One file to read first If you only read one file, read app/main.py:build_app() — every collaborator is injected explicitly, so tests build a parallel world with ManualClock, sequential IdGenerator and a temp runtime_root. :::