diff --git a/README.md b/README.md index c5b8a1f..20d5da6 100644 --- a/README.md +++ b/README.md @@ -131,7 +131,8 @@ Feature-complete across the planned phases: change Curated state. Ask for `preview_curated` with `developing_stories=true` to inspect the bounded story-memory path. It follows confirmed sent reports across source updates, records explicit publish/update/combine/hold/skip decisions, and stays disabled until - `CURATED_DEVELOPING_STORIES=true`. Story drafts and the one allowed revision have a fixed 4,096 + `CURATED_DEVELOPING_STORIES=true`. With the default `CURATED_REQUEST_PROFILE=legacy`, story + drafts and the one allowed revision have a fixed 4,096 token output ceiling; legacy Curated drafts and reviews keep the 900-token ceiling, and every call spends from the same Curated daily token and dollar budgets. Set `CURATED_CHANNEL_ID` only after reviewing preview quality. See @@ -139,6 +140,22 @@ Feature-complete across the planned phases: fetches only newly reported items; [Scout #21](https://github.com/R055LE/scout/issues/21) tracks same-page refresh evidence. + `CURATED_REQUEST_PROFILE=sol-astra-v1` requires developing stories and selects the evaluated + Sol 20260929 writer (16,384 output tokens, 1,200-second timeout) and Astra 20260903 reviewer + (8,192 output tokens, 900-second timeout). Both use OpenAI, default service tier, medium + reasoning, no temperature, and one transport attempt with no provider fallback. Response + caching is disabled; prompt caching uses explicit mode with no cache markers. Unknown cache + usage stays unknown. Reported cache activity, a mismatched route, or an unusable response + prevents publication; trustworthy usage is recorded even on rejected responses. Missing usage + and transport failures are recorded as uncertain, without retaining response text. Selected + profiles reject legacy Curated calls and do not automatically replay uncertain calls. + + 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. + 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 tools), the rate limiter, and the Scout ingestion path. diff --git a/compose.yaml b/compose.yaml index 041bd65..dc0d4de 100644 --- a/compose.yaml +++ b/compose.yaml @@ -39,6 +39,7 @@ services: MODEL_GIGABRAIN: ${MODEL_GIGABRAIN:-} MODEL_CURATED: ${MODEL_CURATED:-} MODEL_CURATED_REVIEW: ${MODEL_CURATED_REVIEW:-} + CURATED_REQUEST_PROFILE: ${CURATED_REQUEST_PROFILE:-legacy} DAILY_TOKENS_ADMIN: ${DAILY_TOKENS_ADMIN:-150000} DAILY_TOKENS_AMBIENT: ${DAILY_TOKENS_AMBIENT:-40000} DAILY_TOKENS_GIGABRAIN: ${DAILY_TOKENS_GIGABRAIN:-100000} diff --git a/roger.env.example b/roger.env.example index 38aaf90..de4615f 100644 --- a/roger.env.example +++ b/roger.env.example @@ -26,6 +26,8 @@ MODEL_GIGABRAIN=anthropic/claude-opus-4.5,openai/gpt-5,google/gemini-2.5-pro MODEL_CURATED=google/gemini-2.5-flash-lite,anthropic/claude-haiku-4.5 # independent fact/excerpt review; no Curated draft can post when this is blank MODEL_CURATED_REVIEW=google/gemini-2.5-pro +# Opt-in story profiles; requires CURATED_DEVELOPING_STORIES=true. +CURATED_REQUEST_PROFILE=legacy # --- budgets (daily in+out tokens per brain) --- DAILY_TOKENS_ADMIN=150000 diff --git a/roger/brains/curated.py b/roger/brains/curated.py index d6d8804..92dbd52 100644 --- a/roger/brains/curated.py +++ b/roger/brains/curated.py @@ -19,7 +19,7 @@ from openai import OpenAIError from roger.identity import ROGER_IDENTITY -from roger.llm import LLM, BudgetExceeded, LLMConfigError +from roger.llm import LLM, BudgetExceeded, CuratedModelError, LLMConfigError from roger.scout_source import collect_from_scout from roger.store import CURATED_CHECK_LEASE_SECONDS, Store @@ -555,7 +555,13 @@ async def finish(status: str, *, retry_at: float = 0) -> None: except LLMConfigError: await finish("configuration") return await result("curated brain not configured") + except CuratedModelError as exc: + await finish("rejected") + return await result(f"model response rejected: {exc.status}; skipped") except (OpenAIError, TimeoutError): + if getattr(settings, "curated_request_profile", "legacy") != "legacy": + await finish("rejected") + return await result("model request interrupted; usage uncertain; skipped") log.exception("curated model request failed") await finish("transient", retry_at=clock().timestamp() + 3600) return await result("model request failed; skipped") @@ -684,6 +690,8 @@ async def preview_curated_job( return {"status": "budget exceeded; skipped"} except LLMConfigError: return {"status": "curated brain not configured"} + except CuratedModelError as exc: + return {"status": f"model response rejected: {exc.status}; skipped"} except OpenAIError: log.exception("developing story preview model request failed") return {"status": "model request failed; skipped"} diff --git a/roger/brains/stories.py b/roger/brains/stories.py index 465da2e..1687c01 100644 --- a/roger/brains/stories.py +++ b/roger/brains/stories.py @@ -1050,6 +1050,9 @@ def input_fingerprint( DECISION_RESPONSE_FORMAT, REVIEW_RESPONSE_FORMAT, ] + profile = getattr(settings, "curated_request_profile", "legacy") + if profile != "legacy": + payload.append({"request_profile": profile}) return hashlib.sha256( json.dumps(payload, sort_keys=True, ensure_ascii=False).encode() ).hexdigest() diff --git a/roger/config.py b/roger/config.py index 0d1d6f5..0ed45eb 100644 --- a/roger/config.py +++ b/roger/config.py @@ -7,11 +7,14 @@ from __future__ import annotations import pathlib +from typing import Literal from zoneinfo import ZoneInfo, ZoneInfoNotFoundError -from pydantic import Field, field_validator +from pydantic import Field, field_validator, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict +from roger.curated_protocol import CURATED_REVIEWER, CURATED_WRITER + def _split_csv(value: str) -> list[str]: return [item.strip() for item in value.split(",") if item.strip()] @@ -40,6 +43,7 @@ class Settings(BaseSettings): model_gigabrain: str = "" model_curated: str = "" model_curated_review: str = "" + curated_request_profile: Literal["legacy", "sol-astra-v1"] = "legacy" # --- budgets (daily in+out tokens per brain) --- daily_tokens_admin: int = Field(default=150_000, gt=0) @@ -137,6 +141,12 @@ def _timezone_must_be_loadable(cls, value: str) -> str: raise ValueError("must name a loadable IANA timezone") from exc return value + @model_validator(mode="after") + 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") + return self + @property def admin_models(self) -> list[str]: return _split_csv(self.model_admin) @@ -151,10 +161,14 @@ def gigabrain_models(self) -> list[str]: @property def curated_models(self) -> list[str]: + if self.curated_request_profile != "legacy": + return [CURATED_WRITER.model] return _split_csv(self.model_curated) @property def curated_review_models(self) -> list[str]: + if self.curated_request_profile != "legacy": + return [CURATED_REVIEWER.model] return _split_csv(self.model_curated_review) @property diff --git a/roger/curated_protocol.py b/roger/curated_protocol.py new file mode 100644 index 0000000..6f18c74 --- /dev/null +++ b/roger/curated_protocol.py @@ -0,0 +1,303 @@ +"""Fixed Curated profiles and the provider protocol shared with offline evaluations.""" + +from __future__ import annotations + +import dataclasses +import json +import re +from decimal import Decimal, InvalidOperation +from typing import Any + +PROFILE_HEADERS = {"X-OpenRouter-Metadata": "enabled", "X-OpenRouter-Cache": "false"} +MAX_INPUT_TOKENS = 32_768 + + +@dataclasses.dataclass(frozen=True) +class CuratedProfile: + name: str + model: str + returned_models: tuple[str, ...] + max_tokens: int + timeout: float + prompt_usd_per_million: Decimal + completion_usd_per_million: Decimal + prompt_cache_write_usd_per_million: Decimal + provider: str = "openai" + provider_display_name: str = "OpenAI" + service_tier: str = "default" + reasoning_effort: str = "medium" + prompt_cache: str = "disabled" + + +CURATED_WRITER = CuratedProfile( + "sol", "openai/gpt-6.1-sol-20260929", + ("openai/gpt-6.1-sol-20260929", "openai/gpt-6.1-sol"), + 16_384, 1200.0, Decimal("2"), Decimal("10"), Decimal("2.5"), +) +CURATED_REVIEWER = CuratedProfile( + "astra", "openai/gpt-6-astra-20260903", + ("openai/gpt-6-astra-20260903", "openai/gpt-6-astra"), + 8_192, 900.0, Decimal("10"), Decimal("50"), Decimal("12.5"), +) + + +def input_token_upper_bound(messages: list[dict[str, Any]], response_format: dict[str, Any]) -> int: + content = json.dumps( + {"messages": messages, "response_format": response_format}, + ensure_ascii=False, separators=(",", ":"), + ).encode("utf-8") + return len(content) + 2_048 + + +def request_kwargs( + profile: Any, messages: list[dict[str, Any]], response_format: dict[str, Any], + *, max_tokens: int, timeout: float, +) -> dict[str, Any]: + if any( + not isinstance(message, dict) or set(message) != {"role", "content"} + or not isinstance(message["content"], str) + for message in messages + ): + raise ValueError("profile messages must be plain text without cache controls") + kwargs = { + "model": profile.model, + "messages": messages, + "max_tokens": max_tokens, + "response_format": response_format, + "service_tier": profile.service_tier, + "timeout": timeout, + "extra_body": { + "provider": { + "only": [profile.provider], + "allow_fallbacks": False, + "require_parameters": True, + "max_price": { + "prompt": float(profile.prompt_usd_per_million), + "completion": float(profile.completion_usd_per_million), + }, + }, + "reasoning": {"effort": profile.reasoning_effort, "exclude": True}, + }, + } + if profile.prompt_cache == "disabled": + kwargs["extra_body"]["prompt_cache_options"] = {"mode": "explicit"} + return kwargs + + +def _is_passed_moderation(stage: Any) -> bool: + if not isinstance(stage, dict) or set(stage) != {"type", "name", "summary", "data"}: + return False + data = stage["data"] + return ( + stage["type"] == "guardrail" + and stage["name"] == "moderation" + and stage["summary"] == "Content passed OpenAI moderation" + and isinstance(data, dict) + and set(data) == {"engine", "flagged", "moderator_ids"} + and data["engine"] == "openai-moderations" + and data["flagged"] is False + and data["moderator_ids"] == ["OpenAIOmniModerator"] + ) + + +def _pipeline_annotation(stage: Any) -> dict[str, Any]: + if not isinstance(stage, dict): + return {"type": str(stage), "recognized_pass_only": False} + annotation = { + key: stage[key] + for key in ("type", "name", "summary") + if isinstance(stage.get(key), str) + } + data = stage.get("data") + if isinstance(data, dict): + clean_data = {} + if isinstance(data.get("engine"), str): + clean_data["engine"] = data["engine"] + if isinstance(data.get("flagged"), bool): + clean_data["flagged"] = data["flagged"] + moderator_ids = data.get("moderator_ids") + if isinstance(moderator_ids, list) and all( + isinstance(value, str) for value in moderator_ids + ): + clean_data["moderator_ids"] = moderator_ids + if clean_data: + annotation["data"] = clean_data + annotation["recognized_pass_only"] = _is_passed_moderation(stage) + return annotation + + +def _route_metadata(response: Any) -> dict[str, Any] | None: + raw = getattr(response, "openrouter_metadata", None) + if raw is None: + extra = getattr(response, "model_extra", None) + raw = extra.get("openrouter_metadata") if isinstance(extra, dict) else None + if not isinstance(raw, dict): + return None + endpoints = raw.get("endpoints") + available = endpoints.get("available") if isinstance(endpoints, dict) else None + clean_endpoints = [] + if isinstance(available, list): + for endpoint in available: + if isinstance(endpoint, dict): + clean_endpoints.append({ + key: endpoint.get(key) for key in ("provider", "model", "selected") + }) + pipeline = raw.get("pipeline") + if isinstance(pipeline, list): + pipeline_stages = [_pipeline_annotation(stage) for stage in pipeline] + pipeline_pass_only = all(_is_passed_moderation(stage) for stage in pipeline) + elif pipeline is None: + pipeline_stages = [] + pipeline_pass_only = True + else: + pipeline_stages = [{"type": "invalid", "recognized_pass_only": False}] + pipeline_pass_only = False + attempts = raw.get("attempts") + return { + "requested": raw.get("requested"), + "strategy": raw.get("strategy"), + "attempt": raw.get("attempt"), + "attempt_count": len(attempts) if isinstance(attempts, list) else None, + "available": clean_endpoints, + "pipeline_stages": pipeline_stages, + "pipeline_pass_only": pipeline_pass_only, + } + + +def _visible_response(response: Any) -> dict[str, Any]: + choices = getattr(response, "choices", None) + shape_error = None + if choices is None: + choice = None + shape_error = "null_choices" + elif not isinstance(choices, list) or not choices: + choice = None + shape_error = "empty_choices" if choices == [] else "invalid_choices" + else: + choice = choices[0] + if choice is None: + shape_error = "null_choice" + elif len(choices) != 1: + shape_error = "multiple_choices" + message = getattr(choice, "message", None) if choice is not None else None + if choice is not None and message is None: + shape_error = "missing_message" + visible = { + "generation_id": _generation_id(response), + "usage_details": _usage_details(response), + "model": getattr(response, "model", None), + "finish_reason": getattr(choice, "finish_reason", None), + "content": getattr(message, "content", None), + "refusal": getattr(message, "refusal", None), + "service_tier": getattr(response, "service_tier", None), + "route_metadata": _route_metadata(response), + } + if shape_error is not None: + visible["shape_error"] = shape_error + return visible + + +def _generation_id(response: Any) -> str | None: + value = getattr(response, "id", None) + if isinstance(value, str) and re.fullmatch(r"gen-[A-Za-z0-9_-]{1,252}", value): + return value + return None + + +def _usage_details(response: Any) -> dict[str, Any]: + usage = getattr(response, "usage", None) + prompt_details = getattr(usage, "prompt_tokens_details", None) + cost_details = getattr(usage, "cost_details", None) + + def field(container: Any, key: str) -> Any: + return container.get(key) if isinstance(container, dict) else getattr(container, key, None) + + counts = {} + prompt_tokens = getattr(usage, "prompt_tokens", None) + for key in ("cached_tokens", "cache_write_tokens"): + value = field(prompt_details, key) + counts[key] = value if ( + type(value) is int and value >= 0 + and (type(prompt_tokens) is not int or prompt_tokens < 0 or value <= prompt_tokens) + ) else None + costs = {} + for key in ( + "upstream_inference_cost", "upstream_inference_prompt_cost", + "upstream_inference_completions_cost", + ): + value = field(cost_details, key) + try: + amount = Decimal(str(value)) + except (InvalidOperation, TypeError, ValueError): + amount = None + costs[key] = ( + str(amount) if amount is not None and amount.is_finite() and amount >= 0 else None + ) + return {**counts, "cost_details": costs} + + +def _usage(response: Any) -> dict[str, int | Decimal | None] | None: + usage = getattr(response, "usage", None) + if usage is None: + return None + cost = getattr(usage, "cost", None) + input_tokens = getattr(usage, "prompt_tokens", None) + output_tokens = getattr(usage, "completion_tokens", None) + if cost is None or input_tokens is None or output_tokens is None: + return None + details = getattr(usage, "completion_tokens_details", None) + reasoning_tokens = getattr(details, "reasoning_tokens", None) if details else None + try: + parsed_cost = Decimal(str(cost)) + except (InvalidOperation, TypeError, ValueError): + return None + counts = (input_tokens, output_tokens) + if ( + not parsed_cost.is_finite() or parsed_cost < 0 + or any(type(value) is not int or value < 0 for value in counts) + or (reasoning_tokens is not None + and (type(reasoning_tokens) is not int or reasoning_tokens < 0)) + ): + return None + return { + "cost_usd": parsed_cost, + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "reasoning_tokens": reasoning_tokens, + } + + +def _response_status(visible: dict[str, Any], profile: Any) -> str: + if visible.get("shape_error") is not None: + return "malformed_response" + if visible["model"] not in profile.returned_models: + return "model_mismatch" + if visible["service_tier"] is not None and visible["service_tier"] != profile.service_tier: + return "service_tier_mismatch" + if visible["finish_reason"] == "length": + return "finish_truncation" + if visible["finish_reason"] == "content_filter": + return "content_filter" + if visible["refusal"]: + return "refusal" + if not isinstance(visible["content"], str) or not visible["content"].strip(): + return "blank_response" + metadata = visible["route_metadata"] + if metadata is None: + return "missing_route_metadata" + if metadata["requested"] != profile.model or metadata["attempt"] != 1: + return "route_metadata_mismatch" + if not metadata["pipeline_pass_only"]: + return "pipeline_interference" + if metadata["attempt_count"] not in (None, 1): + return "routing_multiple_attempts" + selected = [row for row in metadata["available"] if row["selected"] is True] + if len(selected) != 1: + return "missing_selected_endpoint" + expected_provider = "".join(profile.provider_display_name.lower().split()) + actual_provider = "".join(str(selected[0]["provider"]).lower().split()) + if actual_provider != expected_provider or selected[0]["model"] not in profile.returned_models: + return "route_mismatch" + return "completed" + + diff --git a/roger/llm.py b/roger/llm.py index 4d58179..7da6b38 100644 --- a/roger/llm.py +++ b/roger/llm.py @@ -10,6 +10,7 @@ from __future__ import annotations import asyncio +import json import logging from typing import Any @@ -22,7 +23,7 @@ RateLimitError, ) -from roger import metrics +from roger import curated_protocol, metrics from roger.config import Settings from roger.store import Store @@ -59,6 +60,14 @@ class LLMConfigError(RuntimeError): """A brain was invoked with no models configured (MODEL_ is empty).""" +class CuratedModelError(RuntimeError): + """A classified non-publishing result; never includes vendor exception text.""" + + def __init__(self, status: str) -> None: + self.status = status + super().__init__(status) + + class BudgetExceeded(RuntimeError): def __init__(self, brain: str, used: float, cap: float, *, unit: str = "tokens") -> None: if unit == "usd": @@ -88,6 +97,7 @@ def __init__(self, settings: Settings, store: Store) -> None: "gigabrain": settings.gigabrain_models, "curated": settings.curated_models, } + self._curated_profile = settings.curated_request_profile self._curated_review_models = settings.curated_review_models self._curated_public = settings.curated_channel_id is not None self._caps = { @@ -127,6 +137,12 @@ async def complete( raise ValueError("response format is only available to curated story or review calls") if response_format is not None and tools: raise ValueError("response format is not available with tools") + if brain == "curated" and self._curated_profile != "legacy": + if response_format is None or not (curated_story or curated_review) or tools: + raise LLMConfigError("selected Curated profile requires a tool-free story request") + return await self._complete_curated_profile( + messages, response_format, review=curated_review, + ) chain = self._curated_review_models if curated_review else self._chains[brain] if not chain: setting = "MODEL_CURATED_REVIEW" if curated_review else f"MODEL_{brain.upper()}" @@ -184,6 +200,72 @@ async def complete( ) return response + 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 + try: + kwargs = curated_protocol.request_kwargs( + profile, messages, response_format, max_tokens=profile.max_tokens, + timeout=profile.timeout, + ) + except ValueError: + raise LLMConfigError("unsupported Curated profile request") from None + 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") + 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, + } + 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 Exception: + 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() + raise CuratedModelError("request_error_uncertain") from None + visible = curated_protocol._visible_response(response) + usage = curated_protocol._usage(response) + status = curated_protocol._response_status(visible, profile) + if usage is None: + status = "missing_usage_or_cost" if status != "malformed_response" else status + elif (usage["input_tokens"] > input_bound or usage["output_tokens"] > profile.max_tokens + 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"]), + ) + cache = visible["usage_details"] + if status == "completed" and any( + cache[key] is not None and cache[key] > 0 + for key in ("cached_tokens", "cache_write_tokens") + ): + status = "prompt_cache_policy_mismatch" + diagnostic.update( + status=status, generation_id=visible["generation_id"], + usage_details=visible["usage_details"], + usage={key: str(value) if value is not None else None + for key, value in usage.items()} if usage else None, + ) + await self._store.set_meta("curated_profile_last_call", json.dumps(diagnostic)) + if status != "completed": + metrics.LLM_ERRORS.labels("curated", status).inc() + raise CuratedModelError(status) + return response + async def preflight(self) -> list[str]: """OpenRouter key + configured model chains, checked once at boot (ยง11, BACKLOG 4.2). diff --git a/roger/story_eval.py b/roger/story_eval.py index 135d828..82a3278 100644 --- a/roger/story_eval.py +++ b/roger/story_eval.py @@ -11,7 +11,6 @@ import json import os import pathlib -import re import sys import time from decimal import Decimal, InvalidOperation @@ -19,6 +18,7 @@ from openai import APIStatusError, AsyncOpenAI +from roger import curated_protocol from roger.brains import stories from roger.brains.curated import DraftError @@ -27,6 +27,9 @@ DEFAULT_MODELS = _ROOT / "evals/stories/models.json" _REQUEST_FRAMING_BYTES = 2_048 _INPUT_TOKEN_RESERVE = 32_768 +_visible_response = curated_protocol._visible_response +_response_status = curated_protocol._response_status +_usage = curated_protocol._usage class EvaluationError(RuntimeError): @@ -445,35 +448,13 @@ def total(rows: list[dict[str, Any]]) -> str: def _request_kwargs(spec: CallSpec) -> dict[str, Any]: - if any( - not isinstance(message, dict) or set(message) != {"role", "content"} - or not isinstance(message["content"], str) - for message in spec.messages - ): - raise EvaluationError("evaluation messages must be plain text without cache controls") - kwargs = { - "model": spec.profile.model, - "messages": spec.messages, - "max_tokens": spec.max_tokens, - "response_format": spec.response_format, - "service_tier": spec.profile.service_tier, - "timeout": 900.0 if "reviewer" in spec.stage else 1200.0, - "extra_body": { - "provider": { - "only": [spec.profile.provider], - "allow_fallbacks": False, - "require_parameters": True, - "max_price": { - "prompt": float(spec.profile.prompt_usd_per_million), - "completion": float(spec.profile.completion_usd_per_million), - }, - }, - "reasoning": {"effort": spec.profile.reasoning_effort, "exclude": True}, - }, - } - if spec.profile.prompt_cache == "disabled": - kwargs["extra_body"]["prompt_cache_options"] = {"mode": "explicit"} - return kwargs + try: + return curated_protocol.request_kwargs( + spec.profile, spec.messages, spec.response_format, max_tokens=spec.max_tokens, + timeout=900.0 if "reviewer" in spec.stage else 1200.0, + ) + except ValueError as exc: + raise EvaluationError(str(exc)) from exc def _request_record(spec: CallSpec) -> dict[str, Any]: @@ -489,223 +470,6 @@ def _request_record(spec: CallSpec) -> dict[str, Any]: } -def _is_passed_moderation(stage: Any) -> bool: - if not isinstance(stage, dict) or set(stage) != {"type", "name", "summary", "data"}: - return False - data = stage["data"] - return ( - stage["type"] == "guardrail" - and stage["name"] == "moderation" - and stage["summary"] == "Content passed OpenAI moderation" - and isinstance(data, dict) - and set(data) == {"engine", "flagged", "moderator_ids"} - and data["engine"] == "openai-moderations" - and data["flagged"] is False - and data["moderator_ids"] == ["OpenAIOmniModerator"] - ) - - -def _pipeline_annotation(stage: Any) -> dict[str, Any]: - if not isinstance(stage, dict): - return {"type": str(stage), "recognized_pass_only": False} - annotation = { - key: stage[key] - for key in ("type", "name", "summary") - if isinstance(stage.get(key), str) - } - data = stage.get("data") - if isinstance(data, dict): - clean_data = {} - if isinstance(data.get("engine"), str): - clean_data["engine"] = data["engine"] - if isinstance(data.get("flagged"), bool): - clean_data["flagged"] = data["flagged"] - moderator_ids = data.get("moderator_ids") - if isinstance(moderator_ids, list) and all( - isinstance(value, str) for value in moderator_ids - ): - clean_data["moderator_ids"] = moderator_ids - if clean_data: - annotation["data"] = clean_data - annotation["recognized_pass_only"] = _is_passed_moderation(stage) - return annotation - - -def _route_metadata(response: Any) -> dict[str, Any] | None: - raw = getattr(response, "openrouter_metadata", None) - if raw is None: - extra = getattr(response, "model_extra", None) - raw = extra.get("openrouter_metadata") if isinstance(extra, dict) else None - if not isinstance(raw, dict): - return None - endpoints = raw.get("endpoints") - available = endpoints.get("available") if isinstance(endpoints, dict) else None - clean_endpoints = [] - if isinstance(available, list): - for endpoint in available: - if isinstance(endpoint, dict): - clean_endpoints.append({ - key: endpoint.get(key) for key in ("provider", "model", "selected") - }) - pipeline = raw.get("pipeline") - if isinstance(pipeline, list): - pipeline_stages = [_pipeline_annotation(stage) for stage in pipeline] - pipeline_pass_only = all(_is_passed_moderation(stage) for stage in pipeline) - elif pipeline is None: - pipeline_stages = [] - pipeline_pass_only = True - else: - pipeline_stages = [{"type": "invalid", "recognized_pass_only": False}] - pipeline_pass_only = False - attempts = raw.get("attempts") - return { - "requested": raw.get("requested"), - "strategy": raw.get("strategy"), - "attempt": raw.get("attempt"), - "attempt_count": len(attempts) if isinstance(attempts, list) else None, - "available": clean_endpoints, - "pipeline_stages": pipeline_stages, - "pipeline_pass_only": pipeline_pass_only, - } - - -def _visible_response(response: Any) -> dict[str, Any]: - choices = getattr(response, "choices", None) - shape_error = None - if choices is None: - choice = None - shape_error = "null_choices" - elif not isinstance(choices, list) or not choices: - choice = None - shape_error = "empty_choices" if choices == [] else "invalid_choices" - else: - choice = choices[0] - if choice is None: - shape_error = "null_choice" - elif len(choices) != 1: - shape_error = "multiple_choices" - message = getattr(choice, "message", None) if choice is not None else None - if choice is not None and message is None: - shape_error = "missing_message" - visible = { - "generation_id": _generation_id(response), - "usage_details": _usage_details(response), - "model": getattr(response, "model", None), - "finish_reason": getattr(choice, "finish_reason", None), - "content": getattr(message, "content", None), - "refusal": getattr(message, "refusal", None), - "service_tier": getattr(response, "service_tier", None), - "route_metadata": _route_metadata(response), - } - if shape_error is not None: - visible["shape_error"] = shape_error - return visible - - -def _generation_id(response: Any) -> str | None: - value = getattr(response, "id", None) - if isinstance(value, str) and re.fullmatch(r"gen-[A-Za-z0-9_-]{1,252}", value): - return value - return None - - -def _usage_details(response: Any) -> dict[str, Any]: - usage = getattr(response, "usage", None) - prompt_details = getattr(usage, "prompt_tokens_details", None) - cost_details = getattr(usage, "cost_details", None) - - def field(container: Any, key: str) -> Any: - return container.get(key) if isinstance(container, dict) else getattr(container, key, None) - - counts = {} - prompt_tokens = getattr(usage, "prompt_tokens", None) - for key in ("cached_tokens", "cache_write_tokens"): - value = field(prompt_details, key) - counts[key] = value if ( - type(value) is int and value >= 0 - and (type(prompt_tokens) is not int or prompt_tokens < 0 or value <= prompt_tokens) - ) else None - costs = {} - for key in ( - "upstream_inference_cost", "upstream_inference_prompt_cost", - "upstream_inference_completions_cost", - ): - value = field(cost_details, key) - try: - amount = Decimal(str(value)) - except (InvalidOperation, TypeError, ValueError): - amount = None - costs[key] = ( - str(amount) if amount is not None and amount.is_finite() and amount >= 0 else None - ) - return {**counts, "cost_details": costs} - - -def _usage(response: Any) -> dict[str, int | Decimal | None] | None: - usage = getattr(response, "usage", None) - if usage is None: - return None - cost = getattr(usage, "cost", None) - input_tokens = getattr(usage, "prompt_tokens", None) - output_tokens = getattr(usage, "completion_tokens", None) - if cost is None or input_tokens is None or output_tokens is None: - return None - details = getattr(usage, "completion_tokens_details", None) - reasoning_tokens = getattr(details, "reasoning_tokens", None) if details else None - try: - parsed_cost = Decimal(str(cost)) - except (InvalidOperation, TypeError, ValueError): - return None - counts = (input_tokens, output_tokens) - if ( - not parsed_cost.is_finite() or parsed_cost < 0 - or any(type(value) is not int or value < 0 for value in counts) - or (reasoning_tokens is not None - and (type(reasoning_tokens) is not int or reasoning_tokens < 0)) - ): - return None - return { - "cost_usd": parsed_cost, - "input_tokens": input_tokens, - "output_tokens": output_tokens, - "reasoning_tokens": reasoning_tokens, - } - - -def _response_status(visible: dict[str, Any], profile: ModelProfile) -> str: - if visible.get("shape_error") is not None: - return "malformed_response" - if visible["model"] not in profile.returned_models: - return "model_mismatch" - if visible["service_tier"] is not None and visible["service_tier"] != profile.service_tier: - return "service_tier_mismatch" - if visible["finish_reason"] == "length": - return "finish_truncation" - if visible["finish_reason"] == "content_filter": - return "content_filter" - if visible["refusal"]: - return "refusal" - if not isinstance(visible["content"], str) or not visible["content"].strip(): - return "blank_response" - metadata = visible["route_metadata"] - if metadata is None: - return "missing_route_metadata" - if metadata["requested"] != profile.model or metadata["attempt"] != 1: - return "route_metadata_mismatch" - if not metadata["pipeline_pass_only"]: - return "pipeline_interference" - if metadata["attempt_count"] not in (None, 1): - return "routing_multiple_attempts" - selected = [row for row in metadata["available"] if row["selected"] is True] - if len(selected) != 1: - return "missing_selected_endpoint" - expected_provider = "".join(profile.provider_display_name.lower().split()) - actual_provider = "".join(str(selected[0]["provider"]).lower().split()) - if actual_provider != expected_provider or selected[0]["model"] not in profile.returned_models: - return "route_mismatch" - return "completed" - - def _safe_error(exc: Exception) -> dict[str, Any]: result: dict[str, Any] = {"class": type(exc).__name__} if isinstance(exc, APIStatusError): @@ -969,10 +733,7 @@ async def run_matrix( base_url="https://openrouter.ai/api/v1", max_retries=0, timeout=1200.0, - default_headers={ - "X-OpenRouter-Metadata": "enabled", - "X-OpenRouter-Cache": "false", - }, + default_headers=curated_protocol.PROFILE_HEADERS, ) runner = Runner(client, ledger, responses) results: list[dict[str, Any]] = [] diff --git a/tests/test_config.py b/tests/test_config.py index 09bd61d..91a56ac 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -159,6 +159,34 @@ def test_curated_defaults(monkeypatch): assert settings.curated_min_spacing_minutes == 60 assert settings.curated_max_observations_per_day == 8 assert settings.curated_developing_stories is False + assert settings.curated_request_profile == "legacy" + + +def test_curated_profile_requires_explicit_story_opt_in(monkeypatch): + _set_required(monkeypatch) + monkeypatch.setenv("CURATED_REQUEST_PROFILE", "sol-astra-v1") + with pytest.raises(ValidationError, match="requires CURATED_DEVELOPING_STORIES"): + Settings() + + +def test_curated_profile_reports_effective_models_without_changing_legacy_config(monkeypatch): + _set_required(monkeypatch) + monkeypatch.setenv("MODEL_CURATED", "legacy/writer") + monkeypatch.setenv("MODEL_CURATED_REVIEW", "legacy/reviewer") + monkeypatch.setenv("CURATED_REQUEST_PROFILE", "sol-astra-v1") + monkeypatch.setenv("CURATED_DEVELOPING_STORIES", "true") + settings = Settings() + assert settings.curated_models == ["openai/gpt-6.1-sol-20260929"] + assert settings.curated_review_models == ["openai/gpt-6-astra-20260903"] + assert settings.model_curated == "legacy/writer" + assert settings.model_curated_review == "legacy/reviewer" + + +def test_unknown_curated_profile_is_rejected(monkeypatch): + _set_required(monkeypatch) + monkeypatch.setenv("CURATED_REQUEST_PROFILE", "unknown-profile") + with pytest.raises(ValidationError): + Settings() @pytest.mark.parametrize("setting", ["CURATED_CHECK_INTERVAL_MINUTES", diff --git a/tests/test_llm.py b/tests/test_llm.py index bd8ccd6..92c87ca 100644 --- a/tests/test_llm.py +++ b/tests/test_llm.py @@ -7,10 +7,19 @@ import httpx import pytest from openai import APIConnectionError, APIStatusError, AsyncOpenAI +from openai.types.chat import ChatCompletion +from roger import curated_protocol, story_eval from roger.brains import stories from roger.config import Settings -from roger.llm import LLM, MAX_ATTEMPTS, BudgetExceeded, LLMConfigError, _retry_after_seconds +from roger.llm import ( + LLM, + MAX_ATTEMPTS, + BudgetExceeded, + CuratedModelError, + LLMConfigError, + _retry_after_seconds, +) from roger.store import Store @@ -559,3 +568,219 @@ async def test_curated_budget_exceeded_before_network(monkeypatch, tmp_path): await llm.complete("curated", [{"role": "user", "content": "hi"}]) finally: await store.close() + + +@pytest.mark.parametrize("review", [False, True]) +async def test_selected_curated_profile_uses_evaluated_output_limits(monkeypatch, tmp_path, review): + _env(monkeypatch, MODEL_CURATED="legacy/writer", MODEL_CURATED_REVIEW="legacy/reviewer", + CURATED_REQUEST_PROFILE="sol-astra-v1", CURATED_DEVELOPING_STORIES="true") + store = await Store(str(tmp_path / "profile.db")).open() + try: + llm = LLM(Settings(), store) + calls = [] + + async def create(**kwargs): + calls.append(kwargs) + return ChatCompletion.model_validate(_profile_response(review=review)) + + monkeypatch.setattr(llm._client.chat.completions, "create", create) + await llm.complete( + "curated", [{"role": "user", "content": "one measured development"}], + curated_story=not review, curated_review=review, + response_format=(stories.REVIEW_RESPONSE_FORMAT if review + else stories.DECISION_RESPONSE_FORMAT), + ) + assert calls[0]["max_tokens"] == (8192 if review else 16384) + finally: + await store.close() + + +def _profile_response(*, review=False, content="{}"): + profile = curated_protocol.CURATED_REVIEWER if review else curated_protocol.CURATED_WRITER + return { + "id": "gen-profile-fixture", "object": "chat.completion", "created": 1, + "model": profile.returned_models[-1], "service_tier": "default", + "choices": [{"index": 0, "finish_reason": "stop", + "message": {"role": "assistant", "content": content}}], + "usage": {"prompt_tokens": 100, "completion_tokens": 50, "total_tokens": 150, + "cost": 0.01, "completion_tokens_details": {"reasoning_tokens": 10}}, + "openrouter_metadata": { + "requested": profile.model, "strategy": "direct", "attempt": 1, + "attempts": [{"provider": "OpenAI"}], + "endpoints": {"available": [{"provider": "OpenAI", "model": profile.model, + "selected": True}]}, "pipeline": [], + }, + } + + +def test_production_curated_profiles_match_evaluated_contract(): + profiles = story_eval.load_profiles() + for production in (curated_protocol.CURATED_WRITER, curated_protocol.CURATED_REVIEWER): + evaluated = profiles[production.name] + for field in ("model", "returned_models", "provider", "provider_display_name", + "service_tier", "reasoning_effort", "prompt_cache", + "prompt_usd_per_million", "completion_usd_per_million", + "prompt_cache_write_usd_per_million"): + assert getattr(production, field) == getattr(evaluated, field) + assert curated_protocol.CURATED_WRITER.max_tokens == profiles["sol"].writer_max_tokens + assert curated_protocol.CURATED_REVIEWER.max_tokens == profiles["astra"].reviewer_max_tokens + + +@pytest.mark.parametrize("review", [False, True]) +async def test_production_profile_serializes_exact_request(monkeypatch, tmp_path, review): + _env(monkeypatch, CURATED_REQUEST_PROFILE="sol-astra-v1", CURATED_DEVELOPING_STORIES="true") + store = await Store(str(tmp_path / "wire.db")).open() + requests = [] + + async def handle(request): + requests.append(request) + return httpx.Response(200, json=_profile_response(review=review)) + + llm = LLM(Settings(), store) + await llm._client.close() + llm._client = AsyncOpenAI( + api_key="test", base_url="https://openrouter.ai/api/v1", max_retries=0, + http_client=httpx.AsyncClient(transport=httpx.MockTransport(handle)), + ) + try: + await llm.complete( + "curated", [{"role": "user", "content": "a supported finding"}], + curated_story=not review, curated_review=review, + response_format=(stories.REVIEW_RESPONSE_FORMAT if review + else stories.DECISION_RESPONSE_FORMAT), + ) + profile = curated_protocol.CURATED_REVIEWER if review else curated_protocol.CURATED_WRITER + body = json.loads(requests[0].content) + assert len(requests) == 1 + assert body["model"] == profile.model + assert body["max_tokens"] == (8192 if review else 16384) + assert body["service_tier"] == "default" + assert body["reasoning"] == {"effort": "medium", "exclude": True} + assert body["provider"] == { + "only": ["openai"], "allow_fallbacks": False, "require_parameters": True, + "max_price": {"prompt": float(profile.prompt_usd_per_million), + "completion": float(profile.completion_usd_per_million)}, + } + assert body["prompt_cache_options"] == {"mode": "explicit"} + assert body["response_format"]["json_schema"]["strict"] is True + assert not {"temperature", "top_p", "models", "tools"}.intersection(body) + assert requests[0].headers["x-openrouter-cache"] == "false" + assert requests[0].headers["x-openrouter-metadata"] == "enabled" + assert requests[0].extensions["timeout"]["read"] == profile.timeout + assert await store.usage_today("curated") == 150 + assert await store.cost_today("curated") == 0.01 + diagnostic = json.loads(await store.get_meta("curated_profile_last_call")) + assert diagnostic["status"] == "completed" + assert diagnostic["usage_details"]["cache_write_tokens"] is None + assert "content" not in diagnostic + finally: + await llm._client.close() + await store.close() + + +@pytest.mark.parametrize("kind,status", [ + ("null_choices", "malformed_response"), ("missing_usage", "missing_usage_or_cost"), + ("empty_choices", "malformed_response"), + ("cache_write", "prompt_cache_policy_mismatch"), ("wrong_model", "model_mismatch"), + ("blank", "blank_response"), +]) +async def test_profile_response_failures_preserve_safe_usage( + monkeypatch, tmp_path, kind, status, +): + _env(monkeypatch, CURATED_REQUEST_PROFILE="sol-astra-v1", CURATED_DEVELOPING_STORIES="true") + store = await Store(str(tmp_path / "failure.db")).open() + llm = LLM(Settings(), store) + payload = _profile_response(content="private source text") + if kind == "null_choices": + payload["choices"] = None + elif kind == "empty_choices": + payload["choices"] = [] + elif kind == "missing_usage": + del payload["usage"] + elif kind == "cache_write": + payload["usage"]["prompt_tokens_details"] = {"cached_tokens": 0, "cache_write_tokens": 10} + elif kind == "wrong_model": + payload["model"] = "unexpected/private-model" + elif kind == "blank": + payload["choices"][0]["message"]["content"] = "" + response = ChatCompletion.model_validate({**payload, "choices": payload["choices"] or []}) + if kind == "null_choices": + response.choices = None + + async def create(**kwargs): + return response + + monkeypatch.setattr(llm._client.chat.completions, "create", create) + try: + with pytest.raises(CuratedModelError, match=status): + await llm.complete("curated", [{"role": "user", "content": "private source text"}], + curated_story=True, response_format=stories.DECISION_RESPONSE_FORMAT) + diagnostic = await store.get_meta("curated_profile_last_call") + assert json.loads(diagnostic)["status"] == status + assert "private source text" not in diagnostic + assert "unexpected/private-model" not in diagnostic + assert await store.cost_today("curated") == (0 if kind == "missing_usage" else 0.01) + finally: + await llm._client.close() + await store.close() + + +async def test_profile_transport_has_one_attempt_and_sanitized_uncertainty(monkeypatch, tmp_path): + _env(monkeypatch, CURATED_REQUEST_PROFILE="sol-astra-v1", CURATED_DEVELOPING_STORIES="true") + store = await Store(str(tmp_path / "transport.db")).open() + llm = LLM(Settings(), store) + calls = 0 + + async def create(**kwargs): + nonlocal calls + calls += 1 + raise APIConnectionError(message="private vendor response", request=_REQUEST) + + monkeypatch.setattr(llm._client.chat.completions, "create", create) + try: + with pytest.raises(CuratedModelError, match="request_error_uncertain") as caught: + await llm.complete("curated", [{"role": "user", "content": "finding"}], + curated_story=True, response_format=stories.DECISION_RESPONSE_FORMAT) + assert calls == 1 + assert caught.value.__suppress_context__ is True + diagnostic = await store.get_meta("curated_profile_last_call") + assert "private vendor response" not in diagnostic + assert json.loads(diagnostic)["usage"] is None + assert await store.cost_today("curated") == 0 + finally: + await llm._client.close() + await store.close() + + +async def test_selected_profile_rejects_legacy_curated_before_io(monkeypatch, tmp_path): + _env(monkeypatch, CURATED_REQUEST_PROFILE="sol-astra-v1", CURATED_DEVELOPING_STORIES="true") + store = await Store(str(tmp_path / "legacy.db")).open() + llm = LLM(Settings(), store) + try: + with pytest.raises(LLMConfigError, match="tool-free story request"): + await llm.complete("curated", [{"role": "user", "content": "legacy draft"}]) + assert await store.get_meta("curated_profile_last_call") is None + finally: + await llm._client.close() + await store.close() + + +async def test_profile_preflight_checks_effective_pair_not_legacy_chains(monkeypatch, tmp_path): + _env(monkeypatch, CURATED_REQUEST_PROFILE="sol-astra-v1", CURATED_DEVELOPING_STORIES="true", + CURATED_CHANNEL_ID="42", MODEL_CURATED="legacy/writer", MODEL_CURATED_REVIEW="") + store = await Store(str(tmp_path / "preflight.db")).open() + llm = LLM(Settings(), store) + + async def get(path, **kwargs): + if path == "/key": + return {"data": {}} + return {"data": [{"id": profile.model} for profile in ( + curated_protocol.CURATED_WRITER, curated_protocol.CURATED_REVIEWER, + )]} + + monkeypatch.setattr(llm._client, "get", get) + try: + assert await llm.preflight() == [] + finally: + await llm._client.close() + await store.close() diff --git a/tests/test_stories.py b/tests/test_stories.py index b79dea6..7e76ab6 100644 --- a/tests/test_stories.py +++ b/tests/test_stories.py @@ -7,9 +7,14 @@ from types import SimpleNamespace import pytest +from openai.types.chat import ChatCompletion +from roger import curated_protocol from roger.brains import stories from roger.brains.curated import DraftError, preview_curated_job, run_curated_job +from roger.config import Settings +from roger.llm import LLM as ProductionLLM +from roger.llm import CuratedModelError from roger.scout_source import ScoutBatch from roger.store import CURATED_STORY_DECISION_LIMIT, CURATED_STORY_LIMIT, Store @@ -612,6 +617,121 @@ async def collect(*args, **kwargs): await store.close() +@pytest.mark.parametrize("review_failure", [False, True]) +async def test_selected_profile_real_preview_path_is_no_send( + tmp_path, monkeypatch, review_failure, +): + now = datetime.datetime.now(datetime.UTC) + entry = _entry("synthetic", SYNTH_FIRST, production=True, observed_at=now.isoformat()) + + async def collect(*args, **kwargs): + return ScoutBatch([entry], "fixture-run", 1.0, "") + + monkeypatch.setattr("roger.brains.curated.collect_from_scout", collect) + decision = _decision("publish", [_sid("synthetic")], [(1, SYNTH_FIRST)]) + settings = Settings( + discord_token="test", openrouter_api_key="test", owner_id=1, guild_id=2, # noqa: S106 + curated_request_profile="sol-astra-v1", curated_developing_stories=True, + scout_digest_dir=str(tmp_path), + ) + store = await Store(str(tmp_path / "profile-preview.db")).open() + llm = ProductionLLM(settings, store) + calls = [] + + async def create(**kwargs): + calls.append(kwargs) + review = len(calls) == 2 + profile = curated_protocol.CURATED_REVIEWER if review else curated_protocol.CURATED_WRITER + return ChatCompletion.model_validate({ + "id": "gen-offline-preview", "object": "chat.completion", "created": 1, + "model": profile.model, "service_tier": "default", + "choices": [] if review and review_failure else [{ + "index": 0, "finish_reason": "stop", "message": { + "role": "assistant", "content": json.dumps(_review(decision) if review + else decision), + }, + }], + "usage": {"prompt_tokens": 100, "completion_tokens": 50, "total_tokens": 150, + "cost": 0.01}, + "openrouter_metadata": { + "requested": profile.model, "strategy": "direct", "attempt": 1, + "attempts": [{"provider": "OpenAI"}], "pipeline": [], + "endpoints": {"available": [{"provider": "OpenAI", "model": profile.model, + "selected": True}]}, + }, + }) + + monkeypatch.setattr(llm._client.chat.completions, "create", create) + try: + result = await preview_curated_job( + settings=settings, llm=llm, store=store, developing_stories=True, + ) + assert result["status"] == ( + "model response rejected: malformed_response; skipped" if review_failure else "draft" + ) + assert [call["model"] for call in calls] == [ + curated_protocol.CURATED_WRITER.model, curated_protocol.CURATED_REVIEWER.model, + ] + assert await store.usage_today("curated") == 300 + assert await store.cost_today("curated") == 0.02 + for table in ("curated_story", "curated_story_decision", "curated_delivery", "seen", + "curated_observation"): + cursor = await store._conn.execute(f"SELECT COUNT(*) FROM {table}") # noqa: S608 + assert (await cursor.fetchone())[0] == 0 + finally: + await llm._client.close() + await store.close() + + +@pytest.mark.parametrize("error", [CuratedModelError("request_error_uncertain"), TimeoutError()]) +async def test_selected_profile_job_does_not_replay_uncertain_calls( + tmp_path, monkeypatch, caplog, error, +): + entry = _entry("synthetic", SYNTH_FIRST, production=True) + + async def collect(*args, **kwargs): + return ScoutBatch([entry], "fixture-run", 1.0, "") + + calls = 0 + + async def decide(*args, **kwargs): + nonlocal calls + calls += 1 + raise error + + monkeypatch.setattr("roger.brains.curated.collect_from_scout", collect) + monkeypatch.setattr(stories, "decide", decide) + settings = Settings( + discord_token="test", openrouter_api_key="test", owner_id=1, guild_id=2, # noqa: S106 + curated_channel_id=42, curated_request_profile="sol-astra-v1", + curated_developing_stories=True, curated_min_spacing_minutes=0, + scout_digest_dir=str(tmp_path), + ) + store = await Store(str(tmp_path / "profile-job.db")).open() + try: + async def send(**kwargs): + raise AssertionError("uncertain model calls must not send") + + client = SimpleNamespace(get_channel=lambda _: SimpleNamespace(send=send)) + result = await run_curated_job( + client=client, settings=settings, llm=None, store=store, now=NOW, + ) + assert "skipped" in result["status"] + await run_curated_job( + client=client, settings=settings, llm=None, store=store, + now=NOW + datetime.timedelta(hours=2), + ) + assert calls == 1 + cursor = await store._conn.execute("SELECT status, retry_at FROM curated_observation") + rows = await cursor.fetchall() + assert len(rows) == 1 and tuple(rows[0]) == ("rejected", 0) + assert "Traceback" not in caplog.text + cursor = await store._conn.execute("SELECT COUNT(*) FROM curated_delivery") + assert (await cursor.fetchone())[0] == 0 + finally: + await store.close() + + async def test_long_source_url_is_rejected_before_model_or_delivery(tmp_path, monkeypatch): entry = _entry( "long-url", SYNTH_FIRST, production=True, @@ -755,6 +875,19 @@ def test_story_fingerprint_includes_validation_revision_prompt(monkeypatch): assert stories.input_fingerprint(sources, history, settings) != original +def test_story_fingerprint_includes_profile_without_changing_legacy_fingerprint(): + settings = SimpleNamespace(curated_models=["draft"], curated_review_models=["review"]) + sources = stories.prepare_captured_sources( + [_entry("release", SYNTH_FIRST)], now=NOW, max_age_hours=36, + ) + history = {"stories": [], "delivered": []} + original = stories.input_fingerprint(sources, history, settings) + settings.curated_request_profile = "legacy" + assert stories.input_fingerprint(sources, history, settings) == original + settings.curated_request_profile = "sol-astra-v1" + assert stories.input_fingerprint(sources, history, settings) != original + + async def test_story_records_restart_and_enforce_count_caps(tmp_path): path = str(tmp_path / "roger.db") store = await Store(path).open()