Reserve Curated spend before provider attempts

README.md

@@ -152,9 +152,28 @@ Feature-complete across the planned phases:
 
   The scheduled job keeps its 30-minute overall lease and deadline, which can interrupt the
   four-call workflow before all per-call timeouts elapse. The selector defaults to `legacy` and
-  does not change deployed channels, schedules, or budgets. Existing budget checks compare
-  already-reported usage, so they are not hard spend reservations; [#117](https://github.com/R055LE/roger/issues/117)
-  tracks reservation enforcement before activating this profile in production.
+  does not change deployed channels, schedules, or budgets. Selecting the profile also requires
+  positive `DAILY_USD_CURATED`, `DAILY_ATTEMPTS_CURATED`, and `DAILY_WORKFLOWS_CURATED`; all three
+  default to zero. One workflow is a scheduled editorial evaluation or manual preview, with up to
+  four model attempts. These limits are independent of source observations and posted messages.
+
+  Each selected-profile attempt reserves its request-specific input bound and full output ceiling
+  before provider I/O. The input bound includes UTF-8 messages, the complete schema, and framing,
+  up to 32,768 tokens; no evidence is trimmed to fit the budget. The dollar reservation includes
+  the higher cache-write input price. Admission compares used plus outstanding amounts against
+  both daily caps, including legacy Curated usage. Trustworthy completed usage settles once;
+  malformed, interrupted, or ambiguous calls keep their reservation. Reported overruns retain the
+  larger amount and hold all further selected-profile admissions: the previous bound was wrong.
+  Higher caps, midnight, and restarts do not release that hold; explicit reconciliation is required.
+  Reservations never expire or automatically replay.
+
+  Budget days use the host-local day, as existing usage does; posting and source-observation limits
+  use `TZ`. Attempts count on admission day and actual settled charges on completion day. A workflow
+  continued after midnight counts on each day it makes a call. Unresolved reservations carry across
+  midnight and restarts until explicitly reconciled; there is no automatic release or reconciliation
+  command. `/status` shows used, reserved, and unknown amounts plus the last model or budget refusal.
+  The content-free accounting ledger follows existing usage retention and is not automatically
+  pruned. Legacy model profiles keep their existing budget behavior.
 
 Runs as a non-root, read-only-rootfs container. More than 300 tests cover the guard rules, the tool loop
 (including channel creation with access presets and the confirm-gated edit, post, and reorder

compose.yaml

@@ -48,6 +48,8 @@ services:
       DAILY_USD_AMBIENT: ${DAILY_USD_AMBIENT:-0}
       DAILY_USD_GIGABRAIN: ${DAILY_USD_GIGABRAIN:-0}
       DAILY_USD_CURATED: ${DAILY_USD_CURATED:-0}
+      DAILY_ATTEMPTS_CURATED: ${DAILY_ATTEMPTS_CURATED:-0}
+      DAILY_WORKFLOWS_CURATED: ${DAILY_WORKFLOWS_CURATED:-0}
       ADMIN_MAX_TOOL_CALLS: ${ADMIN_MAX_TOOL_CALLS:-10}
       ADMIN_MAX_TURNS: ${ADMIN_MAX_TURNS:-14}
       GIGABRAIN_MAX_TOOL_CALLS: ${GIGABRAIN_MAX_TOOL_CALLS:-10}

roger.env.example

@@ -42,6 +42,10 @@ DAILY_USD_ADMIN=0
 DAILY_USD_AMBIENT=0
 DAILY_USD_GIGABRAIN=0
 DAILY_USD_CURATED=0
+# Selected Curated profiles require positive USD, attempt, and workflow limits.
+# One workflow is a scheduled evaluation or manual preview, with at most four calls.
+DAILY_ATTEMPTS_CURATED=0
+DAILY_WORKFLOWS_CURATED=0
 
 # --- admin tool loop bounds (§2.9) ---
 # hard cap on tool calls / model round-trips within a single admin request. Raise these if

roger/bot.py

@@ -363,6 +363,10 @@ def _format_status(
     curated_last_check: dict[str, Any] | None = None,
     curated_last_decision: dict[str, Any] | None = None,
     curated_last_input: dict[str, Any] | None = None,
+    curated_budget: dict[str, Any] | None = None,
+    curated_attempt_cap: int = 0,
+    curated_workflow_cap: int = 0,
+    curated_profile_last_call: dict[str, Any] | None = None,
 ) -> str:
     """Render the /status readout body (pure). The caller wraps it in a code block."""
     usd_caps = usd_caps or {}
@@ -390,6 +394,21 @@ def _format_status(
     if curated_configured and curated_check_interval_minutes:
         curated = f"checks every {curated_check_interval_minutes} min ({tz} daily limits)"
     lines.append(f"curated: {curated}")
+    if curated_budget is not None:
+        lines.append("curated model budget (host day; unresolved calls carry forward):")
+        for category in ("used", "reserved", "unknown"):
+            lines.append(
+                f"  {category}: {curated_budget[f'{category}_tokens']:,} tokens; "
+                f"${curated_budget[f'{category}_usd']:.6f}"
+            )
+        lines.append(
+            f"curated model attempts: {curated_budget['attempts_today']}/{curated_attempt_cap}; "
+            f"workflows: {curated_budget['workflows_today']}/{curated_workflow_cap}"
+        )
+        if curated_budget["overrun_attempts"]:
+            lines.append("curated reservation hold: reservation_overrun; reconciliation required")
+        if curated_profile_last_call:
+            lines.append(f"curated last model result: {curated_profile_last_call['status']}")
     if curated_configured and curated_state is not None:
         lines.append(
             f"curated allowance: {curated_state['remaining_posts']}/{curated_max_posts_per_day} "
@@ -462,6 +481,14 @@ async def gather_status(*, store: Store, settings: Settings, guild: Any) -> str:
         curated_last_check=json.loads(await store.get_meta("curated_last_check") or "{}"),
         curated_last_decision=json.loads(await store.get_meta("curated_last_decision") or "{}"),
         curated_last_input=json.loads(await store.get_meta("curated_last_input") or "{}"),
+        curated_budget=(await store.curated_budget_state()
+                        if getattr(settings, "curated_request_profile", "legacy") != "legacy"
+                        else None),
+        curated_attempt_cap=getattr(settings, "daily_attempts_curated", 0),
+        curated_workflow_cap=getattr(settings, "daily_workflows_curated", 0),
+        curated_profile_last_call=json.loads(
+            await store.get_meta("curated_profile_last_call") or "{}",
+        ),
         tz=settings.tz,
     )
 

roger/brains/curated.py

@@ -545,13 +545,15 @@ async def finish(status: str, *, retry_at: float = 0) -> None:
                     post = None
                 else:
                     post = await draft([entry for _, entry in ready], llm)
-        except BudgetExceeded:
-            local = clock().astimezone(timezone)
+        except BudgetExceeded as exc:
+            local = clock().astimezone() if exc.reason else clock().astimezone(timezone)
             tomorrow = datetime.datetime.combine(
-                local.date() + datetime.timedelta(days=1), datetime.time(), tzinfo=timezone,
+                local.date() + datetime.timedelta(days=1), datetime.time(),
+                tzinfo=None if exc.reason else timezone,
             )
             await finish("budget", retry_at=tomorrow.timestamp())
-            return await result("budget exceeded; skipped")
+            reason = f" ({exc.reason})" if exc.reason else ""
+            return await result(f"budget exceeded{reason}; skipped")
         except LLMConfigError:
             await finish("configuration")
             return await result("curated brain not configured")
@@ -686,8 +688,9 @@ async def preview_curated_job(
                 batch.entries, history, llm, now=now,
                 max_age_hours=settings.scout_max_age_hours,
             )
-        except BudgetExceeded:
-            return {"status": "budget exceeded; skipped"}
+        except BudgetExceeded as exc:
+            reason = f" ({exc.reason})" if exc.reason else ""
+            return {"status": f"budget exceeded{reason}; skipped"}
         except LLMConfigError:
             return {"status": "curated brain not configured"}
         except CuratedModelError as exc:

roger/brains/stories.py

@@ -14,7 +14,7 @@
 
 from roger.brains.curated import DraftError, _json_object, _response_text, _safe_link
 from roger.identity import ROGER_IDENTITY
-from roger.llm import LLM
+from roger.llm import LLM, curated_workflow
 
 MAX_SOURCES = 8
 SOURCE_EXCERPT_CAP = 1_600
@@ -959,6 +959,7 @@ async def _review(
     return parse_review(_story_response_text(response), decision, sources)
 
 
+@curated_workflow()
 async def decide(
     entries: list[dict[str, Any]],
     history: dict[str, list[dict[str, Any]]],

roger/config.py

@@ -60,6 +60,9 @@ class Settings(BaseSettings):
     daily_usd_ambient: float = Field(default=0.0, ge=0, allow_inf_nan=False)
     daily_usd_gigabrain: float = Field(default=0.0, ge=0, allow_inf_nan=False)
     daily_usd_curated: float = Field(default=0.0, ge=0, allow_inf_nan=False)
+    # Selected profiles require explicit workflow and provider-attempt allowances.
+    daily_attempts_curated: int = Field(default=0, ge=0, le=2**31 - 1)
+    daily_workflows_curated: int = Field(default=0, ge=0, le=2**31 - 1)
 
     # --- admin tool loop bounds (§2.9) ---
     admin_max_tool_calls: int = Field(default=10, gt=0)
@@ -145,6 +148,14 @@ def _timezone_must_be_loadable(cls, value: str) -> str:
     def _curated_profile_requires_stories(self) -> Settings:
         if self.curated_request_profile != "legacy" and not self.curated_developing_stories:
             raise ValueError("CURATED_REQUEST_PROFILE requires CURATED_DEVELOPING_STORIES")
+        if self.curated_request_profile != "legacy" and (
+            self.daily_usd_curated <= 0 or self.daily_attempts_curated <= 0
+            or self.daily_workflows_curated <= 0
+        ):
+            raise ValueError(
+                "CURATED_REQUEST_PROFILE requires positive DAILY_USD_CURATED, "
+                "DAILY_ATTEMPTS_CURATED, and DAILY_WORKFLOWS_CURATED"
+            )
         return self
 
     @property

roger/llm.py

@@ -12,6 +12,10 @@
 import asyncio
 import json
 import logging
+import uuid
+from contextlib import asynccontextmanager
+from contextvars import ContextVar
+from decimal import Decimal
 from typing import Any
 
 from openai import (
@@ -40,6 +44,17 @@
 REQUEST_TIMEOUT_S = 60.0
 MAX_ATTEMPTS = 3
 _BASE_BACKOFF_S = 0.5
+_curated_workflow_id: ContextVar[str | None] = ContextVar("curated_workflow_id", default=None)
+
+
+@asynccontextmanager
+async def curated_workflow():
+    """One editorial evaluation, including its review and optional revision."""
+    token = _curated_workflow_id.set(uuid.uuid4().hex)
+    try:
+        yield _curated_workflow_id.get()
+    finally:
+        _curated_workflow_id.reset(token)
 
 
 def _retry_after_seconds(exc: Exception) -> float | None:
@@ -69,8 +84,13 @@ def __init__(self, status: str) -> None:
 
 
 class BudgetExceeded(RuntimeError):
-    def __init__(self, brain: str, used: float, cap: float, *, unit: str = "tokens") -> None:
-        if unit == "usd":
+    def __init__(
+        self, brain: str, used: float, cap: float, *, unit: str = "tokens",
+        reason: str | None = None,
+    ) -> None:
+        if reason is not None:
+            message = f"{brain} reservation denied: {reason}"
+        elif unit == "usd":
             message = f"{brain} daily $ budget exceeded (${used:.4f} >= ${cap:.4f})"
         else:
             message = f"{brain} daily token budget exceeded ({used} >= {cap})"
@@ -79,6 +99,7 @@ def __init__(self, brain: str, used: float, cap: float, *, unit: str = "tokens")
         self.used = used
         self.cap = cap
         self.unit = unit
+        self.reason = reason
 
 
 class LLM:
@@ -98,6 +119,8 @@ def __init__(self, settings: Settings, store: Store) -> None:
             "curated": settings.curated_models,
         }
         self._curated_profile = settings.curated_request_profile
+        self._curated_attempt_cap = settings.daily_attempts_curated
+        self._curated_workflow_cap = settings.daily_workflows_curated
         self._curated_review_models = settings.curated_review_models
         self._curated_public = settings.curated_channel_id is not None
         self._caps = {
@@ -204,6 +227,9 @@ async def _complete_curated_profile(
         self, messages: list[dict[str, Any]], response_format: dict[str, Any], *, review: bool,
     ) -> Any:
         profile = curated_protocol.CURATED_REVIEWER if review else curated_protocol.CURATED_WRITER
+        workflow_id = _curated_workflow_id.get()
+        if workflow_id is None:
+            raise LLMConfigError("selected Curated profile requires an editorial workflow")
         try:
             kwargs = curated_protocol.request_kwargs(
                 profile, messages, response_format, max_tokens=profile.max_tokens,
@@ -214,22 +240,45 @@ async def _complete_curated_profile(
         input_bound = curated_protocol.input_token_upper_bound(messages, response_format)
         if input_bound > curated_protocol.MAX_INPUT_TOKENS:
             raise LLMConfigError("Curated request exceeds the input reservation maximum")
-        used = await self._store.usage_today("curated")
-        if used >= self._caps["curated"]:
-            raise BudgetExceeded("curated", used, self._caps["curated"])
-        spent = await self._store.cost_today("curated")
-        if self._usd_caps["curated"] > 0 and spent >= self._usd_caps["curated"]:
-            raise BudgetExceeded("curated", spent, self._usd_caps["curated"], unit="usd")
+        reservation = {
+            "workflow_id": workflow_id, "role": "reviewer" if review else "writer",
+            "profile": profile.name, "input_tokens": input_bound,
+            "output_tokens": profile.max_tokens,
+            "cost_usd": (
+                input_bound * max(profile.prompt_usd_per_million,
+                                  profile.prompt_cache_write_usd_per_million)
+                + profile.max_tokens * profile.completion_usd_per_million
+            ) / Decimal(1_000_000),
+            "token_cap": self._caps["curated"], "usd_cap": self._usd_caps["curated"],
+            "max_attempts": self._curated_attempt_cap, "max_workflows": self._curated_workflow_cap,
+            "timeout_seconds": profile.timeout,
+        }
+        attempt_id = await self._store.reserve_curated_attempt(**reservation)
         kwargs["extra_headers"] = curated_protocol.PROFILE_HEADERS
         diagnostic = {
             "profile": self._curated_profile, "role": "reviewer" if review else "writer",
             "model": profile.model, "status": "pending", "usage": None,
         }
+        if attempt_id is None:
+            state = await self._store.curated_budget_state(request=reservation)
+            reason = state["denial_reason"] or "budget_changed"
+            diagnostic["status"] = f"budget_{reason}"
+            await self._store.set_meta("curated_profile_last_call", json.dumps(diagnostic))
+            metrics.LLM_BUDGET_EXCEEDED.labels("curated", reason).inc()
+            raise BudgetExceeded(
+                "curated", state["used_tokens"], self._caps["curated"], reason=reason,
+            )
         await self._store.set_meta("curated_profile_last_call", json.dumps(diagnostic))
         metrics.LLM_REQUESTS.labels("curated").inc()
         try:
             response = await self._client.chat.completions.create(**kwargs)
+        except asyncio.CancelledError:
+            await self._store.mark_curated_attempt_uncertain(attempt_id)
+            diagnostic["status"] = "request_interrupted_uncertain"
+            await self._store.set_meta("curated_profile_last_call", json.dumps(diagnostic))
+            raise
         except Exception:
+            await self._store.mark_curated_attempt_uncertain(attempt_id)
             diagnostic["status"] = "request_error_uncertain"
             await self._store.set_meta("curated_profile_last_call", json.dumps(diagnostic))
             metrics.LLM_ERRORS.labels("curated", "request_error_uncertain").inc()
@@ -243,11 +292,22 @@ async def _complete_curated_profile(
               or (usage["reasoning_tokens"] is not None
                   and usage["reasoning_tokens"] > usage["output_tokens"])):
             status = "invalid_usage_accounting"
-        else:
-            await self._store.add_usage(
-                "curated", usage["input_tokens"], usage["output_tokens"],
-                cost_usd=float(usage["cost_usd"]),
-            )
+        # Content rejection can still have trustworthy accounting. Verify the route
+        # independently because the response classifier checks content failures first.
+        accounting_route = curated_protocol._response_status(
+            {**visible, "content": "accounting", "finish_reason": "stop", "refusal": None},
+            profile,
+        )
+        raw_usage = getattr(response, "usage", None)
+        settled = await self._store.finish_curated_attempt(
+            attempt_id, input_tokens=getattr(raw_usage, "prompt_tokens", None),
+            output_tokens=getattr(raw_usage, "completion_tokens", None),
+            cost_usd=getattr(raw_usage, "cost", None),
+            uncertain=(usage is None or accounting_route != "completed"
+                       or status == "invalid_usage_accounting"),
+        )
+        if settled == "overrun":
+            status = "reservation_overrun"
         cache = visible["usage_details"]
         if status == "completed" and any(
             cache[key] is not None and cache[key] > 0