feat: enforce worker cost budgets

plan/designs/orchestration_design.md

@@ -113,10 +113,12 @@ document" — requirements). Rules:
 
 - **Declaration**: budgets are per `(deployment, stage, lane, window)` — e.g. "extract_claims
   / steady / $N per day" — in deployment config.
-- **Enforcement is a deterministic pre-flight check in the worker**: read the deduplicated
-  `cost_ledger` total for the row's authoritative lane and window (cached, refreshed on the
-  order of minutes — enforcement is a dam, not a scalpel; minutes of lag are priced in). If the
-  budget is exhausted, the worker
+- **Enforcement is a deterministic pre-flight check in the worker**: after locking one due row,
+  read the deduplicated `cost_ledger` total for its authoritative lane and aligned window in the
+  same claim transaction. The single-deployment library uses the existing range index directly;
+  it adds no cache or second budget-state service. Concurrent calls may still finish after another
+  worker passed pre-flight — enforcement is a dam, not a reservation system. If the budget is
+  exhausted, the worker
   **parks** by setting `status='pending'`, `defer_reason='budget'`, and `not_before` to the
   window roll, then re-announces the row and exits *without* starting the handler
   (`processing_state.attempts` and `last_error` are untouched).
@@ -142,6 +144,15 @@ splitting. `cost_ledger.lane` is copied from
 `lane IS NULL`, but they do not silently join either plane-E lane. A delivery envelope or Cloud
 Tasks header cannot choose the attribution.
 
+The model-provider port returns every successful generation or embedding together with required
+provider accounting (resolved model name, input/output tokens, USD cost, and latency). A worker
+binds a small cost-meter port to the running `processing_id`; each model-using handler writes that
+usage under its deterministic call key immediately after the provider returns, before it consumes
+the output. Missing or malformed provider accounting is an adapter error rather than an implicit
+zero, so configured ceilings cannot be silently bypassed. Deterministic and non-worker consumers
+may use the provider port without inventing a processing row; enforced worker-route budgets remain
+anchored solely to `processing_state`.
+
 ## 5. Provider-neutral I/O discipline (portable core of design-review F9)
 
 The spine or another provider may be remote from a worker. The library therefore avoids chatty

plan/plans/phase-7-scale-ops.md

@@ -30,7 +30,7 @@ their round trips.
 |---|---|---|---|---|---|---|
 | WP-7.1 | Backfill lanes + seeding + reprocessing orchestration (version bumps) | orchestration §3–4 | Phase 6 | lane machinery | steady-state unaffected during backfill test | done |
 | WP-7.2 | Reproducible scale battery: D23 partitions/indexes, hub entities/lineages, recount cost, and provider-neutral read/write batching | schema §12; D23; lifecycle §11.5; orchestration §5; retrieval §13.7 | WP-7.1 | fixed synthetic profiles + report | shapes and batching invariants recorded; timings remain measurements, not hosted SLAs | done |
-| WP-7.3 | Cost metering + configurable budget enforcement | orchestration §4; schema §2 `cost_ledger` | WP-7.1 | enforcement + admin inspection | explicit fixture ceiling parks and resumes an over-budget lane; attribution is visible | planned |
+| WP-7.3 | Cost metering + configurable budget enforcement | orchestration §4; schema §2 `cost_ledger` | WP-7.1 | enforcement + admin inspection | explicit fixture ceiling parks and resumes an over-budget lane; attribution is visible | done |
 | WP-7.4 | Operational correctness surfaces + drills: typed telemetry, pipeline/DLQ inspection and replay, P2/P3 rebuild, currency-ledger audit | orchestration §6–7; D7, D60–D61 | WP-7.1 | telemetry/admin surfaces + deterministic drills | failures remain visible and drills pass without a dashboard or hosted control plane | planned |
 | WP-7.5 | **Hard-delete end-to-end**: design first, then purge active P1/P2/P3/K surfaces and prevent restore resurrection through the portable purge record/adapter contract | new design (gate #24); lifecycle §8; k_layers §10; S55 | gate #24 | forget pipeline | **S55 CI gate ON and green** across library-controlled surfaces + restore canary | blocked(#24) |
 | WP-7.6 | **Release engineering**: semver across PyPI + GHCR images + pinned compose; migrations-before-workers upgrade drill; quickstart cold-start release gate | packaging §1, §5–6; D62 | WP-7.1, rename/CLA gate | release pipeline | tagged release produces all artifacts; upgrade drill green; quickstart under target | blocked(rename-gate) |
@@ -73,3 +73,27 @@ The provider-neutral measurement uses the real SQLAlchemy engine with explicit i
 statements, while currency writes and hub recount retain their constant statement counts. Only
 shape, correctness, query-count, and transaction-count invariants gate acceptance; every elapsed
 time is recorded as a machine-specific measurement, never an OSS SLA or topology commitment.
+
+## WP-7.3 implementation
+
+Operators declare no implicit monetary policy. An optional typed `UGM_WORK_BUDGETS` list supplies
+explicit ceilings keyed by deployment, stage, lane, and aligned fixed window; an omitted route is
+unlimited. After locking one due row, the existing claim transaction sums the deduplicated
+`cost_ledger` range for that route. Exhaustion moves the row to durable `pending` / `budget` state
+until the window boundary without starting a handler, consuming an attempt, changing the last
+error, or creating a second scheduling ledger. The worker re-announces that existing row through
+the delivery port with the stored resume time.
+
+`WorkLedger.budget_status` and `ugm budget inspect` read the same two authoritative Postgres
+tables. They expose configured ceiling, current-window spend, remaining amount, tier attribution,
+aligned bounds, and parked-work count; they do not add a dashboard, hosted billing policy, cache,
+or control plane. The PostgreSQL acceptance fixture records two attributed calls, proves an
+over-budget handler never starts, then crosses the fixture window and proves the exact row resumes
+and completes normally.
+
+Successful generation and embedding responses carry mandatory provider-reported usage through the
+existing model port. The worker binds that usage to its running processing row, and every
+model-using stage records a deterministic logical call key and cascade tier before consuming the
+response. OpenRouter responses without cost/token accounting fail visibly instead of degrading to
+zero spend; the deterministic test provider emits zero-cost usage so end-to-end worker tests prove
+the same production attribution path without network calls.

plan/plans/roadmap.md

@@ -102,7 +102,7 @@ needs to restate it:
 | 4 | Projections | P2 (spikes → views → rebuild → snapshots), P3 (tree + mounts incl. raw), communities | p2_graph, e0 §6, `p3_agent_navigation.md` | done (exit criteria met 2026-07-19; PRs #100-#104 — see the phase file) |
 | 5 | Retrieval complete | full primitives + recipe registry, envelope contract CI, MCP/CLI, batch scan, **consumption skill + S58** | retrieval | done (exit criteria met 2026-07-20; PRs #105–#111 — see the phase file) |
 | 6 | Plane K | planner/writer/driver, fact-sheet → prose bands, citations/staleness, authored + sidecars, triggers + subscriptions, K1 + K2 purpose scopes | k_layers | done (exit criteria met 2026-07-21; PRs #112–#117; former WP-6.7 removed by D73) |
-| 7 | Operational correctness + portability | backfill/reprocessing, fixed scale batteries, configurable budgets, failure inspection/drills, hard-delete, release, export/import | orchestration, packaging, schema §12–13 | planned |
+| 7 | Operational correctness + portability | backfill/reprocessing, fixed scale batteries, configurable budgets, failure inspection/drills, hard-delete, release, export/import | orchestration, packaging, schema §12–13 | in progress (WP-7.1–7.3 done; see the phase file) |
 | 8 | Competitive benchmarks | external benchmark harness, adapters, baselines (Mem0/Zep-class), capability benchmark, published methodology + results | D22 (internal) + `phase-8` survey | planned |
 
 Sequencing calls already argued (see the phase files for the rest): **K after retrieval**

src/tests/adapters/test_openrouter.py

@@ -0,0 +1,42 @@
+"""Provider-accounting proofs for the shipped OpenRouter adapter."""
+
+from decimal import Decimal
+
+import pytest
+
+from ultimate_memory.adapters.openrouter import _usage
+from ultimate_memory.model import ProviderAccountingError
+
+
+def test_usage_keeps_exact_cost_and_defaults_embedding_output_tokens() -> None:
+    """Parse required accounting without introducing float rounding."""
+    usage = _usage(
+        body={
+            "model": "resolved/provider-model",
+            "usage": {"prompt_tokens": 17, "cost": "0.000123"},
+        },
+        requested_model="requested/model",
+        latency_ms=9,
+    )
+
+    assert usage.model_name == "resolved/provider-model"
+    assert usage.tokens_in == 17
+    assert usage.tokens_out == 0
+    assert usage.cost_usd == Decimal("0.000123")
+    assert usage.latency_ms == 9
+
+
+@pytest.mark.parametrize(
+    "body",
+    (
+        {},
+        {"usage": {"prompt_tokens": 1}},
+        {"usage": {"prompt_tokens": 1, "cost": "not-a-number"}},
+    ),
+)
+def test_usage_fails_closed_when_required_accounting_is_unusable(
+    body: dict[str, object],
+) -> None:
+    """Never let a worker interpret absent or malformed provider cost as zero."""
+    with pytest.raises(ProviderAccountingError):
+        _usage(body=body, requested_model="requested/model", latency_ms=1)

src/tests/adapters/test_selfhost_queue.py

@@ -27,6 +27,7 @@
 from ultimate_memory.model import ProcessingTarget
 from ultimate_memory.model import QueueRoute
 from ultimate_memory.model import RunResultOutcome
+from ultimate_memory.ports.cost_meter import CostMeterPort
 from ultimate_memory.ports.queue import TaskQueuePort
 from ultimate_memory.spine import DeploymentBootstrapper
 from ultimate_memory.spine import WorkLedger
@@ -161,7 +162,7 @@ def test_announce_reannounces_an_existing_row(
             stage=PipelineStage.EXTRACT_CLAIMS,
             lane=ProcessingLane.STEADY,
         )
-        assert claimed is not None
+        assert isinstance(claimed, ClaimedWork)
         ledger.fail(
             processing_id=claimed.processing_id,
             error="Traceback: transient",
@@ -178,9 +179,9 @@ def test_announce_reannounces_an_existing_row(
 class _NoOpHandler:
     """Succeed without work — the demo chain's terminal stage handler."""
 
-    def handle(self, *, work: ClaimedWork) -> HandlerOutcome:
+    def handle(self, *, work: ClaimedWork, meter: CostMeterPort) -> HandlerOutcome:
         """Do nothing and chain nothing."""
-        del work
+        del work, meter
         return HandlerOutcome()
 
 

src/tests/spine/test_backfill.py

@@ -14,6 +14,7 @@
 
 from ultimate_memory.model import BackfillNotDrainedError
 from ultimate_memory.model import BackfillSeedRequest
+from ultimate_memory.model import ClaimedWork
 from ultimate_memory.model import DeploymentBootstrapInput
 from ultimate_memory.model import EnqueueWork
 from ultimate_memory.model import LaneRouteError
@@ -101,7 +102,7 @@ def _complete_prior_work(*, ledger: WorkLedger, target_ids: tuple[UUID, ...]) ->
             stage=PipelineStage.EXTRACT_CLAIMS,
             lane=ProcessingLane.STEADY,
         )
-        assert claimed is not None
+        assert isinstance(claimed, ClaimedWork)
         ledger.complete(processing_id=claimed.processing_id)
 
 
@@ -187,13 +188,14 @@ def test_version_bump_is_bounded_resumable_and_cannot_starve_steady_work(
         stage=PipelineStage.EXTRACT_CLAIMS,
         lane=ProcessingLane.STEADY,
     )
-    assert claimed_live is not None and claimed_live.target_id == target_ids[3]
+    assert isinstance(claimed_live, ClaimedWork)
+    assert claimed_live.target_id == target_ids[3]
     claimed_backfill = ledger.claim_one(
         deployment_id=_DEPLOYMENT_ID,
         stage=PipelineStage.EXTRACT_CLAIMS,
         lane=ProcessingLane.BACKFILL,
     )
-    assert claimed_backfill is not None
+    assert isinstance(claimed_backfill, ClaimedWork)
     assert claimed_backfill.target_id in target_ids[:3]
 
 
@@ -227,7 +229,7 @@ def test_search_indexes_build_only_after_backfill_has_drained(
         stage=PipelineStage.EXTRACT_CLAIMS,
         lane=ProcessingLane.BACKFILL,
     )
-    assert claimed is not None
+    assert isinstance(claimed, ClaimedWork)
     ledger.complete(processing_id=claimed.processing_id)
 
     finalizer.build_search_indexes(deployment_id=_DEPLOYMENT_ID)