From 5254be28be3fbb91c3c3301399b2da42838c560c Mon Sep 17 00:00:00 2001 From: Ross Tomsic Date: Sat, 7 Nov 2026 09:46:23 -0500 Subject: [PATCH] Use the selected profile input budget for curated stories --- ARCHITECTURE.md | 8 +- roger/brains/curated.py | 7 +- roger/brains/stories.py | 39 +++++-- roger/llm.py | 13 ++- tests/test_llm.py | 60 ++++++++++- tests/test_stories.py | 222 +++++++++++++++++++++++++++++++++++++++- 6 files changed, 331 insertions(+), 18 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index d71b492..07644ef 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -301,9 +301,11 @@ recent output at `CURATED_HOUR=7` in `TZ` by default, after Scout's 05:30 local extends it only when `CURATED_DEVELOPING_STORIES=true`: validated story decisions may follow a changed source version through the same delivery ledger. `/status` and sanitized job outcomes report allowances, spacing, pending deliveries, last input age, and recent check/editorial outcomes. -- Developing-story mode keeps at most eight Scout sources in an 18,000-character request including - the serialized response schema, four confirmed prior deliveries, 100 current story pointers, and - 500 immutable decisions for 90 days. Drafts, the one allowed revision, and independent reviews +- Developing-story mode keeps at most eight Scout sources, four confirmed prior deliveries, + 100 current story pointers, and 500 immutable decisions for 90 days. The selected profile admits + complete requests using its existing 32,768-token conservative UTF-8/schema bound; legacy mode + retains its 18,000-character limit. Oversized requests fail before budget reservation or dispatch. + Drafts, the one allowed revision, and independent reviews use provider-enforced strict JSON schemas plus local parsing, exact-quote validation, and semantic review. A valid draft uses two or four model calls; an initial local-validation failure reuses the one revision slot and uses three, so every path remains at four or fewer. Actual coverage is derived by diff --git a/roger/brains/curated.py b/roger/brains/curated.py index 683949f..8351591 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, CuratedModelError, LLMConfigError +from roger.llm import LLM, BudgetExceeded, CuratedModelError, CuratedRequestError, LLMConfigError from roger.scout_source import collect_from_scout from roger.store import CURATED_CHECK_LEASE_SECONDS, Store @@ -557,6 +557,9 @@ async def finish(status: str, *, retry_at: float = 0) -> None: except LLMConfigError: await finish("configuration") return await result("curated brain not configured") + except CuratedRequestError as exc: + await finish("rejected") + return await result(f"model request rejected: {exc}; skipped") except CuratedModelError as exc: await finish("rejected") return await result(f"model response rejected: {exc.status}; skipped") @@ -693,6 +696,8 @@ async def preview_curated_job( return {"status": f"budget exceeded{reason}; skipped"} except LLMConfigError: return {"status": "curated brain not configured"} + except CuratedRequestError as exc: + return {"status": f"model request rejected: {exc}; skipped"} except CuratedModelError as exc: return {"status": f"model response rejected: {exc.status}; skipped"} except OpenAIError: diff --git a/roger/brains/stories.py b/roger/brains/stories.py index d56921c..98be7c0 100644 --- a/roger/brains/stories.py +++ b/roger/brains/stories.py @@ -12,9 +12,10 @@ import discord +from roger import curated_protocol from roger.brains.curated import DraftError, _json_object, _response_text, _safe_link from roger.identity import ROGER_IDENTITY -from roger.llm import LLM, curated_workflow +from roger.llm import LLM, CuratedRequestError, curated_workflow MAX_SOURCES = 8 SOURCE_EXCERPT_CAP = 1_600 @@ -588,6 +589,7 @@ def relevant(item: dict[str, Any]) -> bool: def _model_input( sources: list[dict[str, Any]], history: dict[str, list[dict[str, Any]]], + *, input_token_limit: int | None = None, ) -> dict[str, Any]: story_ids = {source["story_id"] for source in sources} stories = [] @@ -606,8 +608,7 @@ def _model_input( "stories": stories, "confirmed_sent": delivered, } - encoded = json.dumps(data, ensure_ascii=False) - if len(encoded) > MAX_CONTEXT_CHARS: + if input_token_limit is None and len(json.dumps(data, ensure_ascii=False)) > MAX_CONTEXT_CHARS: raise DraftError("developing story context exceeds limit") return data @@ -684,12 +685,17 @@ def _messages( system: str, payload: dict[str, Any], response_format: dict[str, Any], + *, input_token_limit: int | None = None, ) -> list[dict[str, str]]: content = _compact_json(payload) - request_chars = len(system) + len(content) + len(_compact_json(response_format)) - if request_chars > MAX_CONTEXT_CHARS: - raise DraftError("developing story model context exceeds limit") - return [{"role": "system", "content": system}, {"role": "user", "content": content}] + messages = [{"role": "system", "content": system}, {"role": "user", "content": content}] + if input_token_limit is None: + request_chars = len(system) + len(content) + len(_compact_json(response_format)) + if request_chars > MAX_CONTEXT_CHARS: + raise DraftError("developing story model context exceeds limit") + elif curated_protocol.input_token_upper_bound(messages, response_format) > input_token_limit: + raise CuratedRequestError("input context exceeds limit") + return messages def _story_response_text(response: Any) -> str: @@ -952,6 +958,7 @@ async def _review( REVIEW, review_payload(decision, sources, history), REVIEW_RESPONSE_FORMAT, + input_token_limit=getattr(llm, "curated_input_token_limit", None), ), curated_review=True, response_format=REVIEW_RESPONSE_FORMAT, @@ -977,10 +984,13 @@ async def decide( ) if not sources: raise DraftError("no usable developing story evidence") - model_input = _model_input(sources, history) + input_token_limit = getattr(llm, "curated_input_token_limit", None) + model_input = _model_input(sources, history, input_token_limit=input_token_limit) response = await llm.complete( "curated", - _messages(SYSTEM, model_input, DECISION_RESPONSE_FORMAT), + _messages( + SYSTEM, model_input, DECISION_RESPONSE_FORMAT, input_token_limit=input_token_limit, + ), curated_story=True, response_format=DECISION_RESPONSE_FORMAT, ) @@ -1005,7 +1015,10 @@ async def decide( } response = await llm.complete( "curated", - _messages(revision_system, revision_payload, DECISION_RESPONSE_FORMAT), + _messages( + revision_system, revision_payload, DECISION_RESPONSE_FORMAT, + input_token_limit=input_token_limit, + ), curated_story=True, response_format=DECISION_RESPONSE_FORMAT, ) @@ -1053,7 +1066,11 @@ def input_fingerprint( ] profile = getattr(settings, "curated_request_profile", "legacy") if profile != "legacy": - payload.append({"request_profile": profile}) + payload.append({ + "request_profile": profile, + "input_budget_policy": "utf8-json-plus-2048-v1", + "input_token_limit": curated_protocol.MAX_INPUT_TOKENS, + }) return hashlib.sha256( json.dumps(payload, sort_keys=True, ensure_ascii=False).encode() ).hexdigest() diff --git a/roger/llm.py b/roger/llm.py index fa32ba4..594a359 100644 --- a/roger/llm.py +++ b/roger/llm.py @@ -75,6 +75,10 @@ class LLMConfigError(RuntimeError): """A brain was invoked with no models configured (MODEL_ is empty).""" +class CuratedRequestError(RuntimeError): + """A local input-admission failure, before reservation or provider dispatch.""" + + class CuratedModelError(RuntimeError): """A classified non-publishing result; never includes vendor exception text.""" @@ -138,6 +142,13 @@ def __init__(self, settings: Settings, store: Store) -> None: # Opt-in OpenRouter `reasoning.effort` passthrough — only gigabrain ever sets this today. self._reasoning_effort = {"gigabrain": settings.gigabrain_reasoning_effort or None} + @property + def curated_input_token_limit(self) -> int | None: + """Selected profiles admit complete requests by their conservative token bound.""" + if self._curated_profile == "legacy": + return None + return curated_protocol.MAX_INPUT_TOKENS + async def complete( self, brain: str, @@ -239,7 +250,7 @@ async def _complete_curated_profile( 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") + raise CuratedRequestError("input context exceeds limit") reservation = { "workflow_id": workflow_id, "role": "reviewer" if review else "writer", "profile": profile.name, "input_tokens": input_bound, diff --git a/tests/test_llm.py b/tests/test_llm.py index 4465658..dee8f57 100644 --- a/tests/test_llm.py +++ b/tests/test_llm.py @@ -18,6 +18,7 @@ MAX_ATTEMPTS, BudgetExceeded, CuratedModelError, + CuratedRequestError, LLMConfigError, _retry_after_seconds, curated_workflow, @@ -856,7 +857,7 @@ async def test_profile_counts_utf8_and_schema_before_reserving(monkeypatch, tmp_ store = await Store(str(tmp_path / "input.db")).open() llm = LLM(Settings(), store) try: - with pytest.raises(LLMConfigError, match="input reservation maximum"): + with pytest.raises(CuratedRequestError, match="input context exceeds limit"): await _profile_call(llm, content="é" * 15_000) assert (await store.curated_budget_state())["attempts_today"] == 0 finally: @@ -864,6 +865,63 @@ async def test_profile_counts_utf8_and_schema_before_reserving(monkeypatch, tmp_ await store.close() +@pytest.mark.parametrize("review", [False, True]) +@pytest.mark.parametrize("excess", [0, 1]) +@curated_workflow() +async def test_profile_complete_request_boundary_before_reservation( + monkeypatch, tmp_path, review, excess, +): + _env(monkeypatch, CURATED_REQUEST_PROFILE="sol-astra-v1", CURATED_DEVELOPING_STORIES="true", + DAILY_TOKENS_CURATED="200000", DAILY_USD_CURATED="2.14") + store = await Store(str(tmp_path / "boundary.db")).open() + requests = [] + schema = stories.REVIEW_RESPONSE_FORMAT if review else stories.DECISION_RESPONSE_FORMAT + messages = [{"role": "system", "content": "Evaluate source evidence."}, + {"role": "user", "content": 'é"\\' * 1000}] + remaining = 32768 - curated_protocol.input_token_upper_bound(messages, schema) + messages[1]["content"] += "x" * (remaining + excess) + assert curated_protocol.input_token_upper_bound(messages, schema) == 32768 + excess + + async def handle(request): + requests.append(request) + state = await store.curated_budget_state() + assert state["attempts_today"] == 1 + assert state["reserved_tokens"] == 32768 + (8192 if review else 16384) + return httpx.Response(200, json=_profile_response(review=review)) + + llm = LLM(Settings(), store) + assert llm.curated_input_token_limit == 32768 + 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: + async def complete(): + return await llm.complete( + "curated", messages, curated_story=not review, curated_review=review, + response_format=schema, + ) + + if excess: + with pytest.raises(CuratedRequestError, match="input context exceeds limit"): + await complete() + assert requests == [] + state = await store.curated_budget_state() + assert state["attempts_today"] == state["workflows_today"] == state["used_tokens"] == 0 + assert await store.get_meta("curated_profile_last_call") is None + else: + await complete() + assert len(requests) == 1 + body = json.loads(requests[0].content) + assert body["messages"] == messages + assert body["response_format"] == schema + assert (await store.curated_budget_state())["attempts_today"] == 1 + finally: + await llm._client.close() + await store.close() + + @curated_workflow() async def test_profile_fifth_call_in_one_workflow_is_denied(monkeypatch, tmp_path): _env(monkeypatch, CURATED_REQUEST_PROFILE="sol-astra-v1", CURATED_DEVELOPING_STORIES="true") diff --git a/tests/test_stories.py b/tests/test_stories.py index b3b04c7..8cf58aa 100644 --- a/tests/test_stories.py +++ b/tests/test_stories.py @@ -2,6 +2,7 @@ import copy import datetime +import hashlib import json import time from types import SimpleNamespace @@ -14,7 +15,7 @@ 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.llm import CuratedModelError, CuratedRequestError from roger.scout_source import ScoutBatch from roger.store import CURATED_STORY_DECISION_LIMIT, CURATED_STORY_LIMIT, Store @@ -755,6 +756,54 @@ async def send(**kwargs): await store.close() +@pytest.mark.parametrize("preview", [False, True]) +async def test_selected_oversized_request_is_local_before_ledger_or_provider( + tmp_path, monkeypatch, preview, +): + now = datetime.datetime.now(datetime.UTC) + entries = [_entry(str(index), "é" * 1600, observed_at=now.isoformat()) + for index in range(8)] + + async def collect(*args, **kwargs): + return ScoutBatch(entries, "oversized-fixture", 1.0, "") + + async def forbidden(**kwargs): + raise AssertionError("oversized inputs must not call a provider or send a post") + + monkeypatch.setattr("roger.brains.curated.collect_from_scout", collect) + 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, daily_usd_curated=2, + daily_attempts_curated=8, daily_workflows_curated=2, scout_digest_dir=str(tmp_path), + ) + store = await Store(str(tmp_path / "oversized.db")).open() + llm = ProductionLLM(settings, store) + monkeypatch.setattr(llm._client.chat.completions, "create", forbidden) + try: + if preview: + result = await preview_curated_job( + settings=settings, llm=llm, store=store, developing_stories=True, + ) + else: + client = SimpleNamespace(get_channel=lambda _: SimpleNamespace(send=forbidden)) + result = await run_curated_job( + client=client, settings=settings, llm=llm, store=store, now=now, + ) + assert result["status"] == "model request rejected: input context exceeds limit; skipped" + state = await store.curated_budget_state() + assert state["attempts_today"] == state["workflows_today"] == state["used_tokens"] == 0 + assert await store.get_meta("curated_profile_last_call") is None + cursor = await store._conn.execute("SELECT status FROM curated_observation") + assert [row[0] for row in await cursor.fetchall()] == ([] if preview else ["rejected"]) + for table in ("curated_delivery", "curated_story_decision", "seen"): + 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() + + async def test_long_source_url_is_rejected_before_model_or_delivery(tmp_path, monkeypatch): entry = _entry( "long-url", SYNTH_FIRST, production=True, @@ -911,6 +960,70 @@ def test_story_fingerprint_includes_profile_without_changing_legacy_fingerprint( assert stories.input_fingerprint(sources, history, settings) != original +@pytest.mark.parametrize("profile", ["legacy", "sol-astra-v1"]) +async def test_input_policy_reconsiders_old_rejections_only_for_selected_profile( + tmp_path, monkeypatch, profile, +): + entries = [_entry("synthetic", SYNTH_FIRST, production=True)] + sources = stories.prepare_sources(entries, now=NOW, max_age_hours=36) + history = {"stories": [], "delivered": []} + settings = Settings( + discord_token="test", openrouter_api_key="test", owner_id=1, guild_id=2, # noqa: S106 + curated_channel_id=42, curated_request_profile=profile, + curated_developing_stories=True, daily_usd_curated=2, + daily_attempts_curated=8, daily_workflows_curated=2, scout_digest_dir=str(tmp_path), + ) + old_payload = [ + [{key: value for key, value in source.items() if key != "observed_at"} + for source in sources], + [], settings.curated_models, settings.curated_review_models, + stories.SYSTEM, stories.REVIEW, stories.REVISION, stories.VALIDATION_REVISION, + stories.DECISION_RESPONSE_FORMAT, stories.REVIEW_RESPONSE_FORMAT, + ] + if profile != "legacy": + old_payload.append({"request_profile": profile}) + old_key = hashlib.sha256( + json.dumps(old_payload, sort_keys=True, ensure_ascii=False).encode(), + ).hexdigest() + new_key = stories.input_fingerprint(sources, history, settings) + assert (new_key == old_key) is (profile == "legacy") + + async def collect(*args, **kwargs): + return ScoutBatch(entries, "reconsider-fixture", 1.0, "") + + async def send(**kwargs): + raise AssertionError("quiet decisions must not send") + + monkeypatch.setattr("roger.brains.curated.collect_from_scout", collect) + llm = FakeLLM(_decision("skip", [], [], reason="No material update.")) + llm.curated_input_token_limit = 32768 if profile != "legacy" else None + client = SimpleNamespace(get_channel=lambda _: SimpleNamespace(send=send)) + store = await Store(str(tmp_path / "reconsider.db")).open() + try: + await store._conn.execute( + "INSERT INTO curated_observation (fingerprint, status, failures, retry_at, ts) " + "VALUES (?, 'rejected', 1, 0, ?)", (old_key, NOW.timestamp() - 3600), + ) + await store._conn.commit() + result = await run_curated_job( + client=client, settings=settings, llm=llm, store=store, now=NOW, + ) + assert result["status"] == ( + "unchanged input; skipped" if profile == "legacy" else "no post-worthy items" + ) + repeated = await run_curated_job( + client=client, settings=settings, llm=llm, store=store, + now=NOW + datetime.timedelta(hours=1), + ) + assert repeated["status"] == "unchanged input; skipped" + assert len(llm.calls) == (0 if profile == "legacy" else 2) + records = await store.curated_observations([old_key, new_key]) + assert records[old_key]["status"] == "rejected" + assert records[new_key]["status"] == ("rejected" if profile == "legacy" else "quiet") + finally: + await store.close() + + async def test_story_records_restart_and_enforce_count_caps(tmp_path): path = str(tmp_path / "roger.db") store = await Store(path).open() @@ -1866,6 +1979,25 @@ def test_model_request_limit_counts_full_response_format(response_format): stories._messages(system, {"value": "x" * (filler_size + 1)}, response_format) +@pytest.mark.parametrize("system,schema", [ + (stories.SYSTEM, stories.DECISION_RESPONSE_FORMAT), + (stories.VALIDATION_REVISION, stories.DECISION_RESPONSE_FORMAT), + (stories.REVISION, stories.DECISION_RESPONSE_FORMAT), + (stories.REVIEW, stories.REVIEW_RESPONSE_FORMAT), +]) +def test_selected_request_boundary_counts_utf8_and_escaped_json(system, schema): + payload = {"evidence": 'é"\\' * 1000} + initial = stories._messages(system, payload, schema, input_token_limit=32768) + remaining = 32768 - curated_protocol.input_token_upper_bound(initial, schema) + payload["evidence"] += "x" * remaining + messages = stories._messages(system, payload, schema, input_token_limit=32768) + assert curated_protocol.input_token_upper_bound(messages, schema) == 32768 + assert json.loads(messages[1]["content"]) == payload + payload["evidence"] += "x" + with pytest.raises(CuratedRequestError, match="input context exceeds limit"): + stories._messages(system, payload, schema, input_token_limit=32768) + + def test_max_length_retained_sources_fit_projected_model_history(): entries = [ _entry(str(index), f"Evidence {index} " + "x" * 1_700, @@ -2063,6 +2195,94 @@ def test_eight_max_length_current_sources_fit_draft_request_limit(): ) <= stories.MAX_CONTEXT_CHARS +def _full_evidence_and_history(): + entries = [] + for index in range(stories.MAX_SOURCES): + quote = f"Synthetic source {index} confirms a distinct demonstrated result." + excerpt = (quote + ' Quoted "résumé" at C:\\examples. ' * 4).ljust(1600, "x") + url = f"https://example.test/source-{index}/".ljust(120, "p") + entries.append(_entry(str(index), excerpt, url=url, published="2026-10-01")) + history = {"stories": [], "delivered": [{ + "id": index + 1, "action": "publish", "story_ids": [f"prior-{index}"], + "reason": _grounded("The earlier source supported a bounded report."), + "change": _grounded("The prior report described the observed rollout."), + "claims": [{"field": field, "text": text} for field, text in ( + ("headline", "Earlier synthetic release expands access"), + ("stage", "rollout"), + ("fact", "The earlier announcement described staged access for test accounts. " * 4), + ("why", "The earlier evidence left access and deployment timing unresolved. " * 3), + )], + "correction_of": None, "ts": NOW.timestamp() - (index + 1) * 86400, + "message_id": str(100 + index), + "message_url": f"https://discord.com/channels/1/2/{100 + index}", + } for index in range(stories.MAX_PRIOR_DELIVERIES)]} + return entries, history + + +@pytest.mark.parametrize("revision", [None, "validation", "semantic"]) +async def test_selected_profile_preserves_full_evidence_through_every_stage(revision): + entries, history = _full_evidence_and_history() + sources = stories.prepare_sources(entries, now=NOW, max_age_hours=36) + draft = _decision( + "combine", [source["story_id"] for source in sources], + [(index, source["excerpt"].split(" Quoted")[0]) + for index, source in enumerate(sources, 1)], + ) + corrected = copy.deepcopy(draft) + if revision == "validation": + draft["claims"][1]["text"] = SYNTH_STAGE_SENTENCE + values = [draft, corrected, _review(corrected)] + elif revision == "semantic": + rejected = _review(draft) + rejected["supported"][0] = False + corrected["reason"]["text"] = "The distinct results warrant a combined report." + values = [draft, rejected, corrected, _review(corrected)] + else: + values = [draft, _review(draft)] + llm = RecordingQueueLLM(*values) + llm.curated_input_token_limit = curated_protocol.MAX_INPUT_TOKENS + + decision, actual_sources = await stories.decide( + entries, history, llm, now=NOW, max_age_hours=36, + ) + + assert actual_sources == sources + assert decision.action == "combine" + assert len(llm.calls) == {None: 2, "validation": 3, "semantic": 4}[revision] + payload = json.loads(llm.calls[0][3][1]["content"]) + assert len(json.dumps(payload, ensure_ascii=False)) > stories.MAX_CONTEXT_CHARS + assert payload["sources"] == stories._numbered_sources(sources) + assert payload["confirmed_sent"] == stories._delivered_view(history) + for call, schema in zip(llm.calls, llm.response_formats, strict=True): + assert _request_chars(call[3], schema) > stories.MAX_CONTEXT_CHARS + assert curated_protocol.input_token_upper_bound(call[3], schema) <= 32768 + stage = json.loads(call[3][1]["content"]) + if call[1]: + assert stage["confirmed_sent"] == payload["confirmed_sent"] + assert stage["sources"] == stories._source_metadata(sources) + assert stage["rendered_post"]["title"] == corrected["claims"][0]["text"] + for evidence, source in zip(stage["completeness_evidence"], sources, strict=True): + excerpt = evidence["excerpt"] + for index, citation in enumerate(stage["decision"]["citation_pool"], 1): + if citation["source"] == evidence["number"]: + excerpt = excerpt.replace(f"[cited span {index}]", citation["quote"]) + assert excerpt == source["excerpt"] + assert evidence["published_at"] == source["published_at"] + assert evidence["observed_at"] == source["observed_at"] + elif "input" in stage: + assert stage["input"]["confirmed_sent"] == payload["confirmed_sent"] + assert (stage["input"]["sources"] if revision == "validation" + else _revision_sources(stage["input"])) == payload["sources"] + + +async def test_legacy_still_rejects_full_evidence_before_dispatch(): + entries, history = _full_evidence_and_history() + llm = RecordingQueueLLM() + with pytest.raises(DraftError, match="context exceeds limit"): + await stories.decide(entries, history, llm, now=NOW, max_age_hours=36) + assert llm.calls == [] + + def test_related_history_does_not_match_only_on_url_protocol(): exact = {"stories": [], "delivered": []} recent = {"stories": [{