Spine Milestone 3: The CityOps API

Nine posts ago, the intake pipeline started filling a database nobody could query but you. Today that changes: the 311 data gets served — filtered, paginated, exported, enriched, and guarded. Every lesson in this stage lands somewhere in the machine you're about to build, and the milestone ends the way every milestone ends: not with more code, but with the memo that explains what the code chose.

Assumes: Stage 3 posts 1–6, and the Milestone 2 database the API reads from. Installs: pip install fastapi httpx. Everything below runs locally against a seeded SQLite database — 24 deterministic complaint rows, so every number shown is computed, not asserted.

Before you code: clarify the ask

Input: the Milestone 2 SQLite database, refilling every morning
Output: the CityOps API — filtered reads, a decision queue, CSV exports, weather context, spike alerts, all behind role-based access
Deadline: demo to Lisa and Tom on Friday

Four requirements, and notice what nobody asked for: real-time ingestion, a public API, or a web frontend. The ask is a serving layer for three named humans. That scoping is the whole milestone — everything you refuse to build is a decision the memo will defend.

The minimal concept

This milestone has no new topic. It's the stage, assembled. Every lesson lands somewhere:

Stage lesson	Becomes in the API
REST, Deeply	Nouns and verbs: /complaints with filters as query params, 404s with useful bodies
Auth in the Wild	API keys mapped to roles server-side; HMAC-signed alert webhooks
Pagination & Limits	limit/offset with validation; a streaming CSV export; timeouts on the webhook POST
Webhooks	The spike rule POSTs a signed alert to the consumer's receiver
FastAPI	The framework itself — and the 422s its validation gives you for free
Operator Console	/ops/queue and escalate, now behind RBAC instead of raw DB access

The SDK-design lesson's influence is quieter: the API is shaped so a client library over it would be thin — clean nouns, consistent envelopes. The systems-design lesson's influence is the memo's last section: what breaks as each load dimension grows. The legacy lesson's influence is a field you'll notice missing: crew assignments from the translation layer join this API later, through the same adapter boundary.

Three ideas carry the build:

The API is a serving layer, not a new database. It reads the database Milestone 2 fills. The pipeline owns the data; the API owns access to it. That separation is why the console can move off direct DB reads without anything breaking. The database stores facts. The serving layer decides who may see them, in what shape, and what actions they're allowed to take — that's the architectural payoff of the entire stage.

Authorization is a server-side fact. The client presents a key; the server decides what that key may do. Roles are never asserted by the caller — Lisa's rule, enforced in one dependency every endpoint shares.

The memo is the deliverable. Code ships Friday; the memo explains what the code chose, what it refused, and what breaks first. Every milestone ends with one because the decisions outlive the code.

Build it, part 1: the serving layer

The seed: the Milestone 2 schema plus two small tables — a weather table (seeded sample data, labeled as such throughout) and the API-key registry Lisa asked for:

CREATE TABLE service_requests(
  id INTEGER PRIMARY KEY, unique_key TEXT NOT NULL, created_at TEXT NOT NULL,
  agency TEXT NOT NULL, complaint_type TEXT NOT NULL, descriptor TEXT,
  borough TEXT NOT NULL, status TEXT NOT NULL DEFAULT 'open',
  crew TEXT, exact_location TEXT
);
CREATE TABLE weather(
  day TEXT NOT NULL, borough TEXT NOT NULL,
  temp_f REAL NOT NULL, precip_in REAL NOT NULL,
  PRIMARY KEY(day, borough)
);
-- the bearer secret is never stored: only its hash. id is the stable
-- principal; name is display text, not identity.
CREATE TABLE api_keys(
  id TEXT PRIMARY KEY, key_hash TEXT NOT NULL,
  role TEXT NOT NULL, name TEXT NOT NULL
);
CREATE TABLE audit_log(
  id INTEGER PRIMARY KEY AUTOINCREMENT, action TEXT NOT NULL,
  complaint_id INTEGER, actor TEXT NOT NULL, at TEXT NOT NULL,
  detail TEXT
);

24 deterministic complaint rows across five boroughs — 12 of them stamped 2026-09-30, which gives the spike rule something real to find. Three API keys: Priya the analyst (viewer), Dev (operator), Lisa (admin). The demo key secrets are placeholders with names that say so — and the database holds only their SHA-256 hashes, so a leaked database doesn't hand over live credentials.

The app's foundation — config from the environment, a per-request DB connection, and the auth dependency everything hangs off. Read get_auth carefully: the key goes in, the role comes out of the database — and the database holds only the key's hash, never the secret itself. The client never tells the server who it is beyond presenting the key:

import csv, hashlib, hmac, io, json, os, sqlite3
from datetime import datetime, timedelta, timezone
import httpx
from fastapi import Depends, FastAPI, Header, HTTPException, Query
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

DB = os.environ.get("CITYOPS_DB", "/tmp/m3/cityops.db")
ALERT_URL = os.environ.get("ALERT_URL", "http://127.0.0.1:8931/hooks/cityops-alerts")
ALERT_SECRET = os.environ.get("ALERT_SECRET", "whsec-demo-secret").encode()

app = FastAPI(title="CityOps API", version="0.1.0")

def db():
    con = sqlite3.connect(DB)
    con.row_factory = sqlite3.Row
    try:
        yield con
    finally:
        con.close()

def get_auth(x_api_key: str | None = Header(default=None)):
    if x_api_key is None:
        raise HTTPException(401, "missing X-API-Key")
    # the key is a bearer secret: hash it, then look up the hash.
    # a leaked database must not hand over live credentials.
    digest = hashlib.sha256(x_api_key.encode()).hexdigest()
    con = sqlite3.connect(DB)
    con.row_factory = sqlite3.Row
    try:
        row = con.execute(
            "SELECT id, role, name FROM api_keys WHERE key_hash=?",
            (digest,)).fetchone()
    finally:
        con.close()
    if row is None:
        raise HTTPException(401, "unknown api key")
    return {"actor_id": row["id"], "role": row["role"], "name": row["name"]}

def require_role(*allowed):
    def check(auth=Depends(get_auth)):
        if auth["role"] not in allowed:
            raise HTTPException(
                403, f"role '{auth['role']}' may not use this endpoint")
        return auth
    return check

And the field-level redaction Lisa demanded — exact_location simply doesn't exist in a viewer's response. Not masked, not null: absent. The frontend can't leak what the API never sent:

PUBLIC_FIELDS = ["id", "unique_key", "created_at", "agency", "complaint_type",
                 "descriptor", "borough", "status", "crew"]

def shape(row, role):
    d = {name: row[name] for name in PUBLIC_FIELDS}
    if role in ("operator", "admin"):
        d["exact_location"] = row["exact_location"]
    return d

The list endpoint — Post 1's nouns with Post 3's pagination, filters as query params, total in the envelope so clients can page honestly:

@app.get("/complaints")
def list_complaints(
    borough: str | None = None,
    status: str | None = None,
    complaint_type: str | None = None,
    limit: int = Query(20, ge=1, le=100),
    offset: int = Query(0, ge=0),
    auth=Depends(require_role("viewer", "operator", "admin")),
    con=Depends(db),
):
    where, params = [], []
    for col, val in (("borough", borough), ("status", status),
                     ("complaint_type", complaint_type)):
        if val is not None:
            where.append(f"{col}=?")
            params.append(val)
    clause = ("WHERE " + " AND ".join(where)) if where else ""
    total = con.execute(
        f"SELECT COUNT(*) FROM service_requests {clause}", params).fetchone()[0]
    rows = con.execute(
        f"SELECT * FROM service_requests {clause} ORDER BY id LIMIT ? OFFSET ?",
        (*params, limit, offset)).fetchall()
    return {"total": total, "limit": limit, "offset": offset,
            "items": [shape(r, auth["role"]) for r in rows]}

First contact — real runs against the app (via FastAPI's TestClient, same code paths as HTTP):

GET /health -> 200 {'status': 'ok', 'version': '0.1.0'}
GET /complaints -> 200 total=24 items=20
  viewer sees keys: ['agency', 'borough', 'complaint_type', 'created_at',
                     'crew', 'descriptor', 'id', 'status', 'unique_key']
GET /complaints?borough=QUEENS -> 200 total = 8
GET /complaints?limit=5&offset=5 -> 200 ids = [6, 7, 8, 9, 10]

Liveness versus readiness — two different questions, so two endpoints. /health says the process is alive; /health/ready says the process can actually serve, because it opened the database it depends on:

@app.get("/health")
def health():
    return {"status": "ok", "version": "0.1.0"}   # liveness: alive

@app.get("/health/ready")
def ready():
    con = sqlite3.connect(DB)                     # readiness: can serve
    try:
        con.execute("SELECT 1").fetchone()
    finally:
        con.close()
    return {"status": "ready", "version": "0.1.0"}

GET /health/ready -> 200 {'status': 'ready', 'version': '0.1.0'}

24 total, 20 per default page, filters narrow honestly, and the viewer's item has no exact_location key at all. The limit=500 and limit=abc cases? Post 5's free 422s:

GET /complaints?limit=500 -> 422
GET /complaints?limit=abc -> 422
GET /complaints/9999 -> 404 {'detail': 'complaint 9999 not found'}
GET /complaints (no key) -> 401 {'detail': 'missing X-API-Key'}
GET /complaints (bad key) -> 401 {'detail': 'unknown api key'}

Build it, part 2: the detail view with weather

Maria's team kept asking "was it raining when this was reported?" — so the detail endpoint joins the seeded weather table on the complaint's day and borough. Labeled for what it is: sample data standing in for the weather feed the enrichment lesson would wire up:

@app.get("/complaints/{cid}")
def get_complaint(cid: int,
                  auth=Depends(require_role("viewer", "operator", "admin")),
                  con=Depends(db)):
    row = con.execute(
        "SELECT * FROM service_requests WHERE id=?", (cid,)).fetchone()
    if row is None:
        raise HTTPException(404, f"complaint {cid} not found")
    d = shape(row, auth["role"])
    # weather enrichment: seeded sample table, joined on day + borough
    w = con.execute(
        "SELECT temp_f, precip_in FROM weather WHERE day=date(?) AND borough=?",
        (row["created_at"], row["borough"])).fetchone()
    d["weather"] = dict(w) if w else None
    return d

GET /complaints/11 -> 200 Water Leak | weather: {'temp_f': 67.0, 'precip_in': 0.0}
                                    | exact_location: Atlantic Ave & Nostrand
  viewer exact_location present: False

The operator sees the Brooklyn water leak with its weather context and the exact location; the viewer sees the same complaint, same weather — and no location key. One endpoint, two shapes, decided server-side.

Build it, part 3: Tom's export

Tom's requirement sounds trivial until the row count grows: "a CSV that opens in Excel." The first version of this endpoint built the whole CSV as one string and returned it. It worked on 24 rows. It would not have survived Tom's actual weekly pull — that's Break #1 below. The shipped version streams:

EXPORT_FIELDS = ["id", "created_at", "agency", "complaint_type",
                 "borough", "status", "crew"]

def sanitize_cell(value):
    """CSV escaping protects the file's structure. It does not stop a
    spreadsheet from executing a cell that looks like a formula."""
    s = "" if value is None else str(value)
    if s[:1] in ("=", "+", "-", "@"):
        return "'" + s
    return s

@app.get("/export.csv")
def export_csv(borough: str | None = None, status: str | None = None,
               auth=Depends(require_role("operator", "admin"))):
    where, params = [], []
    for col, val in (("borough", borough), ("status", status)):
        if val is not None:
            where.append(f"{col}=?")
            params.append(val)
    clause = ("WHERE " + " AND ".join(where)) if where else ""

    # Lisa's audit requirement, implemented not deferred: every export
    # is recorded the same way escalations are — actor, filters, timestamp
    audit = sqlite3.connect(DB)
    try:
        audit.execute(
            "INSERT INTO audit_log(action, actor, at, detail)"
            " VALUES (?,?,?,?)",
            ("export", auth["actor_id"],
             datetime.now(timezone.utc).isoformat(),
             json.dumps({"borough": borough, "status": status})))
        audit.commit()
    finally:
        audit.close()

    def gen():
        # the generator owns this connection: its lifetime matches the
        # stream, not the request handler's dependency scope
        con = sqlite3.connect(DB)
        con.row_factory = sqlite3.Row
        try:
            rows = con.execute(
                f"SELECT * FROM service_requests {clause} ORDER BY id", params)
            buf = io.StringIO()
            w = csv.writer(buf)
            w.writerow(EXPORT_FIELDS)
            yield buf.getvalue()
            for r in rows:
                buf.seek(0)
                buf.truncate(0)
                w.writerow([sanitize_cell(r[c]) for c in EXPORT_FIELDS])
                yield buf.getvalue()
        finally:
            con.close()

    return StreamingResponse(
        gen(), media_type="text/csv",
        headers={"Content-Disposition":
                 'attachment; filename="cityops-complaints.csv"'})

Three things to notice. First, the generator opens its own connection in a try/finally — the old version iterated a cursor borrowed from the request's dependency, which couples the stream's lifetime to framework dependency-cleanup behavior you shouldn't have to think about. A streaming generator should own resources whose lifetime must match the stream. Second, the CSV is written by the csv module, not f-strings — don't hand-build a serialization format when the standard library already knows its escaping rules. Watch what it does with a comma inside a value, for real: the seed has a genuine Noise, Residential complaint type, and the export quotes exactly that cell. Third, sanitize_cell: Tom opens this in Excel, and a source field containing =HYPERLINK(...) would execute as a formula. CSV escaping protects the file's structure; it does not make untrusted cell values safe for spreadsheet execution — so formula-like cells get a leading ', which Excel treats as text:

hostile cell, through the endpoint's own writer:
  in:  Noise, Residential
  csv: "Noise, Residential"
  in:  =HYPERLINK("http://evil.example","claim refund")
  csv: "'=HYPERLINK(""http://evil.example"",""claim refund"")"

Note the role list: viewer is absent. Tom is ops; analysts don't get bulk exports. And note what the CSV deliberately excludes: exact_location — Lisa's clarified rule from Monday's conversation (need-to-know doesn't mean bulk-download) survives the export path too, because the export shares the same require_role gate and the same shaping instinct:

viewer export -> 403
operator export?borough=BRONX -> 200 | attachment; filename="cityops-complaints.csv"
                                   | text/csv; charset=utf-8
  first 3 lines:
   id,created_at,agency,complaint_type,borough,status,crew
   6,2026-09-28T13:44:00,DOT,Streetlight,BRONX,open,CRW-04
   8,2026-09-28T15:00:00,DEP,"Noise, Residential",BRONX,open,CRW-04
  audit: export by u-dev recorded with filters {"borough": "BRONX", "status": null}

Build it, part 4: the spike alert

Maria's 9 AM surprise, converted into a rule: the server defines the window — the 24 hours of pipeline data ending at the freshest row — counts complaints inside it, and if the count crosses the threshold, POSTs a signed webhook to the consumer's receiver. Callers can't relabel an arbitrary since as "spike-24h"; Post 4's pattern, HMAC signature in the header, timeout on the POST:

class SpikeCheck(BaseModel):
    threshold: int = Field(10, ge=1)

def sign(body: bytes) -> str:
    return hmac.new(ALERT_SECRET, body, hashlib.sha256).hexdigest()

@app.post("/alerts/check")
def check_spike(check: SpikeCheck,
                auth=Depends(require_role("operator", "admin")),
                con=Depends(db)):
    # the window is server-defined: the 24h of pipeline data ending at the
    # freshest row. "banana" is not a valid since, and neither is 2026-01-01
    # relabeled as 24 hours.
    latest = con.execute(
        "SELECT MAX(created_at) FROM service_requests").fetchone()[0]
    since = (datetime.fromisoformat(latest) - timedelta(hours=24)).isoformat()
    count = con.execute(
        "SELECT COUNT(*) FROM service_requests WHERE created_at >= ?",
        (since,)).fetchone()[0]
    if count < check.threshold:
        return {"rule": "spike-24h", "count": count,
                "threshold": check.threshold, "alert_sent": False}
    # event_id lets the receiver recognize the same event twice —
    # signing proves who sent it, the id proves it isn't new
    event_id = f"spike-24h:{since}"
    payload = {"event_id": event_id, "rule": "spike-24h", "count": count,
               "threshold": check.threshold, "since": since,
               "at": datetime.now(timezone.utc).isoformat()}
    body = json.dumps(payload).encode()
    headers = {"X-CityOps-Signature": sign(body),
               "Content-Type": "application/json"}
    try:
        with httpx.Client(timeout=5) as client:
            r = client.post(ALERT_URL, content=body, headers=headers)
    except httpx.HTTPError as exc:
        # delivery is ambiguous now — the receiver may have stored the alert
        # and lost the response, or never seen it. Report that honestly
        # instead of claiming success.
        return {"rule": "spike-24h", "event_id": event_id, "count": count,
                "threshold": check.threshold,
                "delivery_attempted": True, "delivery_succeeded": False,
                "webhook_status": None,
                "note": f"delivery failed: {type(exc).__name__}"}
    ok = 200 <= r.status_code < 300
    return {"rule": "spike-24h", "event_id": event_id, "count": count,
            "threshold": check.threshold,
            "delivery_attempted": True, "delivery_succeeded": ok,
            "webhook_status": r.status_code}

The receiver is a stand-in for the consumer's server — in production it lives on Maria's team's infrastructure; here it's a local HTTP server so you can watch both sides. It verifies the HMAC before it believes a word (the verifying half of Break #3) — then checks the event is fresh and not a duplicate. Signing proves who sent the event; the timestamp and event ID prove it isn't a replay or a retry:

seen_events = set()

class HookHandler(BaseHTTPRequestHandler):
    def do_POST(self):
        body = self.rfile.read(int(self.headers["Content-Length"]))
        sig = self.headers.get("X-CityOps-Signature", "")
        good = hmac.compare_digest(
            sig, hmac.new(SECRET, body, hashlib.sha256).hexdigest())
        if not good:
            self.send_response(401); self.end_headers()
            self.wfile.write(b'{"error":"bad signature"}')
            return
        event = json.loads(body)
        # replay protection: a captured valid payload must also be fresh
        at = datetime.fromisoformat(event["at"])
        if datetime.now(timezone.utc) - at > timedelta(minutes=10):
            self.send_response(401); self.end_headers()
            self.wfile.write(b'{"error":"stale event"}')
            return
        # idempotency: the same event_id twice is a retry, not a new alert
        if event["event_id"] in seen_events:
            self.send_response(200); self.end_headers()
            self.wfile.write(b'{"ok":true,"duplicate":true}')
            return
        seen_events.add(event["event_id"])
        received.append(event)
        self.send_response(200); self.end_headers()
        self.wfile.write(b'{"ok":true}')

The real firing — 12 complaints in the server-defined 24h window against a threshold of 10, a real HTTP round trip on localhost, signature verified:

POST /alerts/check {"threshold": 10} -> 200 {'rule': 'spike-24h', 'count': 12,
    'threshold': 10, 'event_id': 'spike-24h:2026-09-29T11:00:00',
    'delivery_attempted': True, 'delivery_succeeded': True, 'webhook_status': 200}
  receiver stored: {'event_id': 'spike-24h:2026-09-29T11:00:00', 'rule': 'spike-24h',
    'count': 12, 'threshold': 10, 'since': '2026-09-29T11:00:00',
    'at': '2026-10-03T18:44:20.958404+00:00'}
POST /alerts/check {"threshold": 50} -> 200 {'rule': 'spike-24h', 'count': 12,
    'threshold': 50, 'alert_sent': False}

And the failure case, for real — the receiver down, the alert POST going nowhere. The old code returned "alert_sent": True here, which was a lie: the POST was attempted, nothing more:

POST /alerts/check {"threshold": 10} -> 200 {'rule': 'spike-24h',
    'event_id': 'spike-24h:2026-09-29T11:00:00', 'count': 12, 'threshold': 10,
    'delivery_attempted': True, 'delivery_succeeded': False,
    'webhook_status': None, 'note': 'delivery failed: ConnectError'}

HTTP 200 from our API — the check itself ran fine — but delivery_succeeded: False. Note the ambiguity the note is honest about: a timeout (rather than a refused connection) means the receiver may have stored the alert and lost the response — which is exactly why the payload carries an event_id the receiver deduplicates on. Same event delivered twice, for real:

delivery 1 -> 200 {'ok': True}
delivery 2 -> 200 {'ok': True, 'duplicate': True}

Build it, part 5: the operator console, wired in

Dev's console from Post 6 read the database directly. Now it reads the API — one serving layer, not two. The queue endpoint is the decision queue: open complaints, oldest first, the ones whose SLAs are burning:

@app.get("/ops/queue")
def ops_queue(auth=Depends(require_role("operator", "admin")), con=Depends(db)):
    # the operator console's decision queue: open complaints, oldest first
    rows = con.execute(
        "SELECT * FROM service_requests WHERE status='open'"
        " ORDER BY created_at ASC LIMIT 20").fetchall()
    return {"decisions_pending": len(rows),
            "queue": [shape(r, auth["role"]) for r in rows]}

And the console's action — escalate — becomes an API operation with an audit trail, because Lisa audits what operators do:

@app.post("/complaints/{cid}/escalate")
def escalate(cid: int, auth=Depends(require_role("operator", "admin")),
             con=Depends(db)):
    row = con.execute(
        "SELECT * FROM service_requests WHERE id=?", (cid,)).fetchone()
    if row is None:
        raise HTTPException(404, f"complaint {cid} not found")
    # state transition, not blind update: an already-escalated complaint
    # is a 409, not a second audit row for the same decision
    if row["status"] == "escalated":
        raise HTTPException(409, f"complaint {cid} is already escalated")
    con.execute("UPDATE service_requests SET status='escalated' WHERE id=?",
                (cid,))
    con.execute(
        "INSERT INTO audit_log(action, complaint_id, actor, at)"
        " VALUES (?,?,?,?)",
        ("escalate", cid, auth["actor_id"],
         datetime.now(timezone.utc).isoformat()))
    con.commit()
    return {"id": cid, "status": "escalated", "by": auth["name"]}

GET /ops/queue -> 200 decisions_pending = 18
  oldest: 1 2026-09-28T08:12:00 Missed Collection QUEENS
viewer escalate -> 403 {'detail': "role 'viewer' may not use this endpoint"}
operator escalate -> 200 {'id': 2, 'status': 'escalated', 'by': 'Dev (field ops)'}
operator escalate again -> 409 {'detail': 'complaint 2 is already escalated'}
audit rows: [('escalate', 2, 'u-dev')]

18 open complaints, the oldest a Missed Collection from September 28th — that's the top of Dev's Monday queue. Priya the viewer can't escalate; Dev can, and the audit log records his stable actor id (u-dev), not his display name — names change, audit trails shouldn't. And Dev double-clicking doesn't create a second escalation: the 409 says the transition already happened. Post 6's principle holds at the API layer: the queue is still decisions, not dashboards.

Break it, three ways

A milestone that only works on demo day is a demo. Here's what the bad days look like — all three reproduced for real.

1. Tom's export, at Tom's actual scale. The first export implementation built the entire CSV as one string and returned it. On 24 rows, instant. So I generated 200,000 and 2,000,000 synthetic rows and timed both approaches — the naive full-string build versus the streaming generator that yields one row at a time (both through the csv module):

naive,     200k rows:  0.24s, 14.3 MB string held in memory
streaming, 200k rows:  0.29s, one row in memory at a time
naive,       2M rows:  2.52s, 144.9 MB string held in memory
streaming,   2M rows:  3.06s, one row in memory at a time

Same speed — wildly different memory. The naive version's cost isn't time, it's the string sitting in RAM per concurrent export — 145 MB at 2M rows, before the application or runtime adds its own overhead — plus the latency truth the timer hides: with the naive build, the first byte leaves the server only after the last row is formatted; with the generator, Tom's download starts immediately. This is Post 3's lesson wearing an export costume: bound the work per unit, never hold the whole world to serve part of it.

2. The role the client chose for itself. The first RBAC draft took the role from a query parameter — POST /complaints/2/escalate?role=operator — because "the frontend knows who's logged in." The frontend knows; the attacker also knows. Real run against that draft:

v0: viewer + ?role=operator escalate -> 200 {'escalated': True, 'by_role': 'operator'}

Priya's viewer key, self-promoted to operator, escalating complaints. The fix is the get_auth dependency you already read: the key goes in, the role comes out of the server's own table. Same caller, fixed code:

viewer escalate -> 403 {'detail': "role 'viewer' may not use this endpoint"}

Post 2's rule, restated for APIs: identity is part of architecture — and architecture never trusts the client to describe itself. Any authorization decision that reads a client-asserted value is decoration, not security.

3. The alert nobody signed. The first webhook draft POSTed the alert with no signature — "it's localhost, it's fine." Then I pointed a forged alert at two receivers: one naive, one verifying. Real runs:

forged alert (no signature) -> naive receiver:    200 {'stored': True}
  naive receiver stored: [{'rule': 'spike-24h', 'count': 9999}]
forged alert (no signature) -> verifying receiver: 401 {'error': 'bad signature'}
signed alert                -> verifying receiver: 200 {'ok': True}
stale replay (valid signature, 2-hour-old timestamp) -> 401 {'error': 'stale event'}

The naive receiver happily stored a spike of 9,999 complaints that never happened — and Maria would have walked into the 9 AM standup with it. Post 4's rule survives the milestone: never trust a knock you didn't verify, even when the knock comes from your own server. The signature isn't paranoia; it's what makes the alert evidence instead of rumor. And the last line is the point the lesson now makes explicitly: a valid signature is necessary but not sufficient — a captured payload is a replay waiting to happen, so the receiver also demands freshness. Valid HMAC does not equal complete webhook security.

Productionize: Friday and after

The demo is Friday. The Monday after is what productionize is for:

Run it like a service. uvicorn cityops_api:app --host 0.0.0.0 --port 8000 behind the team's reverse proxy; CITYOPS_DB, ALERT_URL, and ALERT_SECRET come from the environment, never from the code. The demo secret in the listing is a placeholder with a name that says so — Lisa's first review comment, preempted.

Health and observability. /health is the load balancer's liveness check — the process is alive. /health/ready is the readiness check — the process can actually serve, because it opened the database it depends on. The audit log records every escalation and every export with actor id and timestamp — Lisa's audit requirement, implemented, and the answer to "who downloaded the whole borough?"

Rate limits, per Post 3. The export and spike-check endpoints are the expensive ones; they get the polite-client treatment — per-key rate limits, and a 429 with Retry-After instead of a silent slowdown. The 24-row demo doesn't need it; Tom's weekly pull concurrent with Maria's team does.

What the API doesn't do. No writes except escalate — the pipeline owns ingestion, and the API doesn't get a second door into the data. No real-time: the database refills each morning, and every response is as fresh as the last pipeline run. Saying so in the memo is what keeps Maria from building a dashboard that assumes otherwise.

The tradeoff memo

The centerpiece. This is the document that goes to Lisa, Tom, and Maria on Friday — the decisions, written down:

Read that memo again and notice what it is: every "not built" has a reason, every tradeoff names its trigger, and "what breaks as each load dimension grows" is a list of future decisions already scheduled. That's the milestone's real deliverable. The code is v0.1.0; the memo is what lets v0.2.0 happen without archaeology.

Explain it to the customer

Where this lands in CityOps

The CityOps API is the third spine. Milestone 1 gave the project its skeleton; Milestone 2 gave it a heartbeat — a database that refills itself every morning, with proof. Milestone 3 gives it a face: everything downstream now goes through one serving layer — the console, the analysts, Tom's spreadsheets, the spike alerts. The crew assignments from the legacy translation layer will join as just another field behind this same API, because the boundary work is already done.

And the stage exit check, stated plainly: you can now take "serve this data to three kinds of humans with different permissions" from a vague request to a tested, guarded, documented API with a memo defending its choices. That's the whole of Stage 3 in one sentence — and it's the job.

Post 5's principle was an API is a contract you sign with strangers. Post 6's was operators don't need dashboards, they need a queue with decisions. Milestone 3's: a milestone isn't done when the code works — it's done when the decisions are written down.