feat(fleet): accept a paired client's own earnings history (#256)

* feat(fleet): accept a paired client's own earnings history

A client that has been reading a provider account on its own -- CashPilot
Desktop before it was paired -- had no way to hand that history to the
server, so the fleet view began on the day of pairing and every earlier
day was simply lost from the total.

POST /api/workers/earnings-import takes those readings and stores them
under the importing client's own source, which the source-aware schema
added in the previous change makes possible. Separate series matter
here: earnings are clamped deltas between consecutive readings of the
same balance, so interleaving two samplers of one account makes every
apparent drop clamp to zero and understates the total. Each series is
differenced on its own and the results are summed.

Two properties carry the security of it:

* The source comes from the AUTHENTICATED worker, never the request
  body, so no client can write into another's history or into the
  server's own.
* Only a fully enrolled worker may import. A caller still presenting the
  shared enrollment key gets 403 with instructions to heartbeat first --
  every worker holds that key, and this writes durable money data.

Re-sending a day updates it rather than appending, so a retried or
repeated import is safe by construction.

Refs: CashPilot-Desktop-xjr

* fix(fleet): bound the import body and validate the reading date

Two ways the new endpoint could be handed input it would store without
complaint.

The date was free text. Both delta readers ORDER BY it, so a reading
dated 2026-1-2 or 01/02/2026 sorts into the wrong place in its own
series and the readings either side then difference against the wrong
neighbour. It fails silently, only for the client that sent it, and only
in the earned figure -- never in the balance the dashboard shows. It is
now required to be YYYY-MM-DD and a real calendar day, so 2026-02-30 is
refused rather than stored.

The readings list was unbounded. One authenticated client could hand the
server an arbitrarily large body to parse and then write row by row; a
single compromised worker is enough. Capped at 2000, comfortably above
an honest import (the server keeps 400 days and a client chunks at
1000).

Validation runs before authentication, so a malformed body cannot be
used to probe which client ids exist.

Refs: CashPilot-Desktop-xjr

* fix(compose): pin the examples to the 1.15 series

v1.15.0 released and the example compose files still pinned 1.14, so
anyone following the quickstart deployed a series behind. The pin test
caught it -- it exists because a stale pin is what gave issue #188 a
version with a first-run bug that had been fixed for months.

Unrelated to this branch's change, but it fails CI on every branch until
it is fixed.

* fix(fleet): a NaN balance was accepted, and rejecting one was a 500

Found by re-reading the endpoint rather than from a report.

JSON has no NaN or Infinity. Python's parser accepts them anyway, so
{"balance": NaN} was stored verbatim. One such reading poisons every
delta taken from that series -- NaN - x is NaN, and every comparison
against it is False, so the clamp silently misbehaves -- the account
total becomes NaN, and serialising that back out emits a bare NaN that
JSON.parse rejects. A single bad reading from one client breaks the
dashboard for everyone. Both float fields now refuse non-finite values.

That exposed a second problem underneath: FastAPI's 422 body echoes the
offending input, so the REJECTION could not be serialised either and the
client got a 500 for what is squarely a bad request. A
RequestValidationError handler now renders non-finite floats as their
names -- keeping the message diagnostic rather than dropping the field --
and fixes the whole class instead of the one endpoint that takes a float
today.

My first version of that handler broke every OTHER validation error: a
custom validator's error carries the raised ValueError OBJECT in ctx,
which is not serialisable, and skipping jsonable_encoder turned each one
into a 500. Caught by the date tests, and now pinned by its own
regression test.

Also: the skipped list is deduplicated. A client pushing 400 days of a
platform this server does not know got the same name back 400 times --
a response that grows with the request, echoing client-supplied strings,
and saying nothing the set does not.

A negative control showed the sanitiser's bool guard was dead (bool
subclasses int, not float), so it is gone along with the comment that
justified it incorrectly.

Refs: CashPilot-Desktop-xjr

* perf(fleet): import in one transaction, and roll back when it fails

A thousand-reading import committed a thousand times. Every commit is an
fsync that takes SQLite's write lock, so one import serialised a thousand
disk syncs against this server's own collector and request latency
tracked sync cost rather than row count. upsert_earnings_many does the
same upsert with executemany and one commit.

Only half the reported cause was real, and the distinction matters for
anyone reading this later: _get_db hands out a borrowed handle on a
SHARED per-loop connection whose close() is a documented no-op, so the
loop was never opening and closing a thousand connections. It was
committing a thousand times.

Batching then introduced a bug of its own, which the tests caught: the
connection is shared and outlives the request, so a failed batch left an
abandoned transaction holding the write lock and the NEXT write blocked
for twelve seconds before timing out. It now rolls back.

Two more from the same review:

* docs/fleet.md gains the Authorization header and every status the
  endpoint can answer with. A reader integrating against that page alone
  could not previously construct a valid request.
* Two assertions were tuple expressions -- `mock.assert_not_awaited(),
  "why"` builds a tuple and discards the message, so a red build showed
  the mock's generic text instead of the reason. Fixed here and in
  test_optional_runtime.py.

The review said ruff's B018 catches that pattern when bugbear is enabled.
Bugbear IS enabled here and B018 is not ignored, and ruff 0.15.14 passes
it clean -- checked against a minimal probe rather than assumed. So
nothing in CI would have caught a recurrence, and there is now an
AST-based test that does. Structural, not a string search, because a
string search would match the pattern inside its own docstring.

One negative control PASSED, which meant the test was wrong: the
transactionality test raised while BUILDING the rows, before any SQL ran,
so it proved only that the row build validates first and it passed
against a writer that committed after every row. It now fails inside the
statement, and both it and the wedged-connection test fail under their
controls.

Reported by CodeRabbit on PR #256.

Author: GeiserX

app/database.py

@@ -8,10 +8,12 @@
 from __future__ import annotations
 
 import asyncio
+import contextlib
 import json
 import logging
 import math
 import os
+from collections.abc import Mapping, Sequence
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 from typing import Any
@@ -921,6 +923,78 @@ async def upsert_earnings(
         await db.close()
 
 
+async def upsert_earnings_many(readings: Sequence[Mapping[str, Any]]) -> int:
+    """Upsert many readings in ONE transaction. Returns how many were written.
+
+    Same statement and same conflict rules as :func:`upsert_earnings` — this is
+    purely about how often the work is committed.
+
+    WHY IT EXISTS
+    -------------
+    A client importing its pre-pairing history sends up to a thousand readings
+    per request, and calling ``upsert_earnings`` in a loop commits once per row.
+    Every commit is an fsync, and every one of them takes SQLite's write lock —
+    so a single import serialised a thousand disk syncs against the server's own
+    collector, and request latency tracked disk sync cost rather than row count.
+
+    (The connection is NOT the problem, despite appearances: ``_get_db`` hands
+    out a borrowed handle on a shared per-loop connection whose ``close()`` is a
+    documented no-op, so the loop was never opening and closing a thousand
+    connections. It was committing a thousand times.)
+
+    ONE TRANSACTION ALSO MEANS ALL-OR-NOTHING, which is the behaviour to want
+    here: a failure part-way through leaves the caller's history exactly as it
+    was rather than half-applied, and the import is idempotent so retrying costs
+    a round trip.
+    """
+    rows = [
+        (
+            r["platform"],
+            float(r["balance"]),
+            (r.get("currency") or "USD"),
+            r.get("date") or datetime.now(UTC).strftime("%Y-%m-%d"),
+            r.get("fx_rate_usd"),
+            (r.get("source") or "server"),
+        )
+        for r in readings
+    ]
+    if not rows:
+        # No statement, no commit. An empty import is a normal case, and taking
+        # the write lock to do nothing would still block the collector.
+        return 0
+
+    db = await _get_db()
+    try:
+        try:
+            await db.executemany(
+                """
+                INSERT INTO earnings (platform, balance, currency, date, fx_rate_usd, source)
+                VALUES (?, ?, ?, ?, ?, ?)
+                ON CONFLICT(platform, source, date) DO UPDATE SET
+                    balance = excluded.balance,
+                    currency = excluded.currency,
+                    fx_rate_usd = COALESCE(excluded.fx_rate_usd, earnings.fx_rate_usd),
+                    created_at = datetime('now')
+                WHERE earnings.balance != excluded.balance
+                   OR earnings.fx_rate_usd IS NULL
+                """,
+                rows,
+            )
+            await db.commit()
+        except Exception:
+            # ROLL BACK, do not merely propagate. The connection is SHARED and
+            # outlives the request, so an abandoned transaction keeps SQLite's
+            # write lock and every later write on this loop -- including this
+            # server's own collector -- blocks until it times out. Measured: the
+            # next write took twelve seconds before this was added.
+            with contextlib.suppress(Exception):
+                await db.rollback()
+            raise
+    finally:
+        await db.close()
+    return len(rows)
+
+
 async def get_earnings_summary() -> list[dict[str, Any]]:
     """Return the latest balance for each platform."""
     db = await _get_db()

app/main.py

@@ -11,6 +11,7 @@
 import hmac
 import json
 import logging
+import math
 import os
 import re
 import secrets
@@ -23,9 +24,11 @@
 from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_MISSED
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from fastapi import FastAPI, HTTPException, Request
+from fastapi.encoders import jsonable_encoder
+from fastapi.exceptions import RequestValidationError
 from fastapi.responses import JSONResponse, PlainTextResponse
 from fastapi.staticfiles import StaticFiles
-from pydantic import BaseModel
+from pydantic import BaseModel, Field, field_validator
 from starlette.middleware.base import BaseHTTPMiddleware
 
 from app import (
@@ -984,6 +987,51 @@ async def dispatch(self, request, call_next):
 
 app.add_middleware(_SecurityHeadersMiddleware)
 
+
+@app.exception_handler(RequestValidationError)
+async def _validation_error_handler(request: Request, exc: RequestValidationError) -> JSONResponse:
+    """FastAPI's 422 body echoes the offending input, and some inputs cannot be
+    encoded — so the rejection itself became a 500.
+
+    JSON has no ``NaN`` or ``Infinity``. Python's parser accepts them anyway, so
+    a client can put one in a float field; pydantic correctly rejects it, and
+    then the default handler tries to serialise ``{"input": nan}`` and Starlette's
+    encoder raises "Out of range float values are not JSON compliant". The client
+    gets an opaque 500 for what is squarely a bad request, and the log fills with
+    a traceback that names the encoder rather than the cause.
+
+    Non-finite floats are therefore rendered as their names. That keeps the
+    message diagnostic ("we saw NaN") instead of dropping the field, and it fixes
+    the whole class rather than the one endpoint that happens to take a float
+    today. Every other error is passed through byte-for-byte, so the response
+    shape callers already parse is unchanged.
+    """
+    # jsonable_encoder FIRST, exactly as FastAPI's own handler does: a custom
+    # validator's error carries the raised ValueError OBJECT in ctx, which is not
+    # serialisable either. Encoding then sanitising fixes both without changing
+    # the body for any error that was already fine.
+    return JSONResponse(status_code=422, content={"detail": _json_safe(jsonable_encoder(exc.errors()))})
+
+
+def _json_safe(value: Any) -> Any:
+    """Recursively replace values the JSON encoder refuses — currently only
+    non-finite floats.
+
+    Deliberately narrow. ``bool`` needs no special case: it subclasses ``int``,
+    not ``float``, so the check below never sees it. An earlier version guarded
+    for it anyway, with a comment that was simply wrong; a negative control
+    showed removing the guard changed nothing, so it went.
+    (``test_booleans_survive_as_booleans`` still pins the guarantee.)
+    """
+    if isinstance(value, dict):
+        return {k: _json_safe(v) for k, v in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_json_safe(v) for v in value]
+    if isinstance(value, float) and not math.isfinite(value):
+        return repr(value)  # "nan", "inf", "-inf"
+    return value
+
+
 # Static files
 app.mount("/static", StaticFiles(directory="app/static"), name="static")
 
@@ -3713,6 +3761,139 @@ async def _earnings_for_worker(body: WorkerHeartbeat, days: int = 30) -> dict[st
     }
 
 
+#: The exact shape ``upsert_earnings`` writes and both delta readers ORDER BY.
+#: A date in any other shape would sort into the wrong place in its own series,
+#: so the readings either side of it difference against the wrong neighbour --
+#: silently, and only for the client that sent it.
+_ISO_DATE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
+
+
+class EarningsReading(BaseModel):
+    """One historical balance reading from a paired client."""
+
+    slug: str
+    #: allow_inf_nan=False because JSON's `NaN` and `Infinity` -- which Python's
+    #: parser accepts even though the spec does not -- are otherwise stored
+    #: verbatim. One NaN balance poisons every delta taken from that series
+    #: (NaN - x is NaN, and every comparison against it is False, so the clamp
+    #: silently misbehaves), the account total becomes NaN, and serialising that
+    #: back out emits a bare `NaN` that JSON.parse rejects. So a single bad
+    #: reading from one client breaks the dashboard for everyone.
+    balance: float = Field(allow_inf_nan=False)
+    date: str
+    currency: str = "USD"
+    fx_rate_usd: float | None = Field(default=None, allow_inf_nan=False)
+
+    @field_validator("date")
+    @classmethod
+    def _iso_date(cls, value: str) -> str:
+        value = (value or "").strip()
+        if not _ISO_DATE.match(value):
+            raise ValueError("date must be YYYY-MM-DD")
+        # Reject 2026-02-30 and friends: the pattern above only proves shape.
+        try:
+            datetime.strptime(value, "%Y-%m-%d")  # noqa: DTZ007 -- a calendar date, not an instant
+        except ValueError as exc:
+            raise ValueError("date must be a real calendar date") from exc
+        return value
+
+
+class EarningsImport(BaseModel):
+    """A client pushing the history it collected before it was paired.
+
+    Deliberately carries NO source field. The source is taken from the
+    AUTHENTICATED worker, never from the body -- otherwise any enrolled client
+    could write into the 'server' series, or into another machine's, and
+    overwrite readings it never took.
+    """
+
+    client_id: str
+    #: Bounded so one authenticated client cannot hand the server an
+    #: arbitrarily large body to parse and then write row by row. 2000 is
+    #: comfortably above a real import -- the server keeps 400 days, and a
+    #: client chunks anything larger -- while staying a body the process can
+    #: hold. An unbounded list here is a denial of service that needs only one
+    #: compromised worker.
+    readings: list[EarningsReading] = Field(default_factory=list, max_length=2000)
+
+
+@app.post("/api/workers/earnings-import")
+async def api_worker_earnings_import(request: Request, body: EarningsImport) -> dict[str, Any]:
+    """Accept a paired client's historical earnings under its own source.
+
+    This exists because the server and a Desktop can both have been reading the
+    SAME provider account. Their readings must not be merged into one series:
+    earnings are clamped deltas between consecutive readings, so interleaving two
+    samplers makes every drop clamp to zero and the total comes out
+    systematically understated. Each client's rows therefore land under its own
+    ``source`` and are differenced separately.
+
+    Idempotent by construction: the (platform, source, date) unique index means a
+    re-pair or a retried import UPDATES a day rather than adding a second reading
+    for it, which would difference against itself and read as zero.
+    """
+    cid = (body.client_id or "").strip()
+    if not cid:
+        raise HTTPException(status_code=400, detail="client_id required")
+
+    state = await _authenticate_worker_heartbeat(request, cid)
+    # Only a CONFIRMED worker may import. "enroll" and "reissue" both mean the
+    # caller presented the SHARED key, which every worker holds -- accepting it
+    # here would let anyone with that token write a history for any client_id
+    # they cared to name. A heartbeat is idempotent status; this is durable
+    # money data, so it gets the stricter bar.
+    if state != "ok":
+        raise HTTPException(
+            status_code=403,
+            detail=(
+                "Importing earnings requires this worker's own key. Send a heartbeat first "
+                "to complete enrollment, then retry."
+            ),
+        )
+
+    known = {s["slug"] for s in catalog.get_services()}
+    # DISTINCT slugs, not one entry per skipped reading. A client pushing 400
+    # days of a platform this server does not know would otherwise get the same
+    # name back 400 times -- a response that grows with the request, echoing
+    # client-supplied strings, and tells the reader nothing the set does not.
+    skipped: set[str] = set()
+    rows: list[dict[str, Any]] = []
+    for reading in body.readings:
+        slug = (reading.slug or "").strip()
+        # An unknown slug is dropped rather than stored: it would create a
+        # platform the catalog cannot render, name or ever collect for again.
+        if not slug or slug not in known:
+            skipped.add(slug or "(blank)")
+            continue
+        rows.append(
+            {
+                "platform": slug,
+                "balance": float(reading.balance),
+                "currency": (reading.currency or "USD").upper(),
+                "date": reading.date,
+                "fx_rate_usd": reading.fx_rate_usd,
+                "source": cid,
+            }
+        )
+
+    # ONE transaction for the whole batch. Writing row by row committed once per
+    # reading, and every commit is an fsync that takes SQLite's write lock -- so
+    # a thousand-reading import serialised a thousand disk syncs against this
+    # server's own collector, and latency tracked sync cost rather than row
+    # count. It also makes the import all-or-nothing: a failure part-way leaves
+    # the client's history exactly as it was rather than half-applied, and the
+    # import is idempotent so retrying costs a round trip. (CodeRabbit, PR #256.)
+    written = await database.upsert_earnings_many(rows)
+
+    logger.info(
+        "Imported %d earnings reading(s) from worker '%s' (%d unknown platform(s) skipped)",
+        written,
+        cid,
+        len(skipped),
+    )
+    return {"status": "ok", "imported": written, "skipped": sorted(skipped), "source": cid}
+
+
 @app.post("/api/workers/heartbeat")
 async def api_worker_heartbeat(request: Request, body: WorkerHeartbeat) -> dict[str, Any]:
     """Receive a heartbeat from a worker. Registers or updates the worker."""

docs/fleet.md

@@ -34,6 +34,69 @@ Workers use **REST HTTP** to communicate with the UI:
 
 Workers must be reachable from the UI for commands. The UI must be reachable from workers for heartbeats.
 
+### Importing earnings a client collected on its own
+
+`POST /api/workers/earnings-import` lets a client that has been reading a
+provider account by itself — CashPilot Desktop, typically, before it was paired —
+hand that history to the UI, so the fleet view shows the complete picture rather
+than starting from the day of pairing.
+
+```http
+POST /api/workers/earnings-import
+Authorization: Bearer <this worker's own key, not the shared enrollment key>
+Content-Type: application/json
+
+{
+  "client_id": "desktop-macbook",
+  "readings": [
+    {"slug": "honeygain", "balance": 12.4, "date": "2026-07-01", "currency": "USD"},
+    {"slug": "mysterium", "balance": 88.0, "date": "2026-07-01", "currency": "MYST", "fx_rate_usd": 0.41}
+  ]
+}
+```
+
+It answers:
+
+| Status | Meaning |
+|---|---|
+| `200` | `{"status": "ok", "imported": N, "skipped": [...], "source": "<client_id>"}` |
+| `400` | `client_id` was missing or blank |
+| `401` | the key is wrong or revoked |
+| `403` | this worker is not fully enrolled yet — send a heartbeat first, then retry |
+| `422` | the body was rejected: a date that is not a real `YYYY-MM-DD` day, a non-finite `balance` or `fx_rate_usd`, or more than 2000 readings |
+
+Three things about it are deliberate:
+
+- **Each client's readings are stored under their own source, not merged with the
+  server's.** Earnings are clamped deltas between consecutive readings of the same
+  balance, so interleaving two samplers of one account makes every apparent drop
+  clamp to zero and understates the total. Separate series are differenced
+  separately and then summed.
+- **The source is taken from the authenticated worker, never from the request
+  body.** Otherwise any enrolled client could write into another's history, or
+  into the server's own.
+- **Only a fully enrolled worker may import.** A caller still presenting the
+  shared enrollment key gets `403` with instructions to heartbeat first: every
+  worker holds that key, and this writes durable money data.
+
+Re-sending the same day updates it rather than appending, so a retried or
+repeated import is safe.
+
+`date` must be `YYYY-MM-DD` and a real calendar day. It is not free text: both
+delta readers order by it, so a differently shaped date sorts into the wrong
+place in its own series and the readings either side then difference against the
+wrong neighbour — silently, and only in the earned figure. A request may carry at
+most **2000 readings**; send larger histories in chunks.
+
+`balance` and `fx_rate_usd` must be finite. JSON has no `NaN` or `Infinity`, but
+Python's parser accepts them, and one `NaN` balance poisons every earned figure
+taken from that series — every comparison against it is false, so the clamp
+misbehaves silently and the account total becomes `NaN`. They are rejected with
+`422`.
+
+Unknown slugs come back in `skipped` as a **distinct, sorted** list, so a year of
+one unrecognised platform is reported once rather than 400 times.
+
 ## Setting Up the Fleet
 
 ### Main server (UI + local worker)