diff --git a/reflexio/models/api_schema/domain/entities.py b/reflexio/models/api_schema/domain/entities.py index 2907a75ef..4ad0cc2da 100644 --- a/reflexio/models/api_schema/domain/entities.py +++ b/reflexio/models/api_schema/domain/entities.py @@ -486,6 +486,23 @@ class AgentPlaybook(BaseModel): "stale_incumbent", "governance_invalidated", "infrastructure_failure", + # Written by reflexio_ext offline_tuner/open_world/runner.py:251 + # (_converge_terminal_failure, behind the regeneration fence) and assigned + # by the tenant stage-advance RPC's 'failed' arm + # (supabase/data/tenant/20260830020000:325-327). Admitted by + # playbook_optimization_jobs_terminal_outcome_check since 20260827040000. + # An attempt REFUSED before spending, not a fault -- see + # reflexio_ext open_world/models.py:210-227. + "regeneration_fenced", + # Written by reflexio_ext offline_tuner/open_world/runner.py + # (_converge_terminal_failure, from the slot-exhausted handler) when EVERY + # durable row identity the discovery question could occupy is already owned + # by a different optimization job -- so the attempt made NO provider call at + # all. Split out of 'infrastructure_failure', which it was previously + # indistinguishable from, hiding the fact that a failed attempt was burning + # its playbook's identity for the rest of the UTC day. See + # reflexio_ext open_world/models.py and identity.attempt_invocation_identity. + "invocation_slot_pinned", ] OptimizationArtifactKind = Literal[ diff --git a/reflexio/server/llm/_litellm_structured_output.py b/reflexio/server/llm/_litellm_structured_output.py index 0ec13d48c..b1fe8a779 100644 --- a/reflexio/server/llm/_litellm_structured_output.py +++ b/reflexio/server/llm/_litellm_structured_output.py @@ -36,6 +36,7 @@ from reflexio.server.llm.llm_utils import ( assert_provider_safe_schema, is_pydantic_model, + make_strict_json_schema, prompt_schema_instruction, strict_response_format_for_model, ) @@ -270,8 +271,27 @@ class StructuredOutputMixin: # for discriminated unions, which strict structured-output endpoints reject. # Listing the provider here forces our own # normalized strict schema (``oneOf`` folded into ``anyOf``) to be sent. - _JSON_SCHEMA_PROVIDER_ALLOWLIST: frozenset[str] = frozenset({"minimax"}) - _PROMPT_SCHEMA_PROVIDER_ALLOWLIST: frozenset[str] = frozenset({"zai"}) + # + # Empty by design, not by neglect: ``minimax`` was its only member and has + # moved to the prompt-schema allowlist below, because it turned out to + # ignore ``response_format`` rather than merely be under-reported. The + # mechanism stays for the next provider that genuinely fits the shape + # described above -- accepts json_schema, reported as unsupported. + _JSON_SCHEMA_PROVIDER_ALLOWLIST: frozenset[str] = frozenset() + + # Providers that ignore ``response_format`` outright, so the schema has to + # travel in the prompt or the model never sees it at all. + # + # ``minimax`` was previously in the json-schema allowlist above, on the + # belief that it accepts a ``json_schema`` response_format that LiteLLM + # merely under-reports. Measured against the live API, it does not: two + # identical calls differing only in ``drop_params`` both returned free + # prose rather than JSON, so the response_format is discarded whatever we + # send. The visible symptom was an analyst inventing a DIFFERENT set of + # field names on every run -- it was answering from the prompt alone, + # having never been given a schema. With the schema in the prompt the same + # model returns exact conforming JSON. + _PROMPT_SCHEMA_PROVIDER_ALLOWLIST: frozenset[str] = frozenset({"zai", "minimax"}) # Base-owned attribute read for the parse-failure error message (init'd in # the facade ``__init__``). Annotation-only; NEVER assign here. @@ -323,8 +343,16 @@ def _structured_output_strategy( def _prompt_schema_directive( self, *, response_format: type[BaseModel], tools_available: bool ) -> str: - """Build and guard the schema instruction used by prompt-only providers.""" - schema = response_format.model_json_schema() + """Build and guard the schema instruction used by prompt-only providers. + + The schema is normalized (``oneOf`` folded into ``anyOf``) BEFORE the + provider-safety assertion, exactly as the native json_schema path does. + Without that fold a discriminated-union output would trip + ``assert_provider_safe_schema`` and raise, so a model whose only + transport is the prompt could never carry such a schema at all -- it + would fail before the request was built rather than degrade. + """ + schema = make_strict_json_schema(response_format.model_json_schema()) assert_provider_safe_schema(schema, name=response_format.__name__) return prompt_schema_instruction(schema, tools_available=tools_available) diff --git a/reflexio/server/services/playbook_optimizer/optimizer.py b/reflexio/server/services/playbook_optimizer/optimizer.py index 29748adb1..d8c4464b4 100644 --- a/reflexio/server/services/playbook_optimizer/optimizer.py +++ b/reflexio/server/services/playbook_optimizer/optimizer.py @@ -249,8 +249,17 @@ def optimize( error_type=type(exc).__name__, ): logger.exception("Playbook optimization failed") + # The exception message is deliberately not persisted here. It can + # carry customer content -- a pydantic ValidationError raised on a + # provider response renders the model's own output into its message + # -- and ``decision_reason`` is a durable column read back into the + # domain model and shown to operators. The class name, the traceback + # and the tags are already captured by the ``error_tags`` block + # above, which is where an unbounded diagnostic belongs. self.storage.update_playbook_optimization_job( - job.job_id, status="failed", decision_reason=str(exc) + job.job_id, + status="failed", + decision_reason="optimization run raised an unexpected error", ) return "failed" diff --git a/tests/models/test_optimization_terminal_outcome.py b/tests/models/test_optimization_terminal_outcome.py new file mode 100644 index 000000000..6aa33a15f --- /dev/null +++ b/tests/models/test_optimization_terminal_outcome.py @@ -0,0 +1,53 @@ +"""``OptimizationTerminalOutcome`` must admit the outcomes the tuner writes. + +'regeneration_fenced' reached the tenant CHECK in 20260827040000 (re-declared by +20260830020000) and is written by reflexio_ext +offline_tuner/open_world/runner.py:251 via ``_converge_terminal_failure``, but +was never added to this union -- so reading such a row back through +``_row_to_playbook_optimization_job`` raised ``ValidationError``. + +That failure did not surface as a validation error either. ``handle_exceptions`` +converts it to ``StorageError``, and the open-world runner's ``except +StorageError`` arm reports ``infrastructure_failure`` -- so a refusal the fence +made on purpose was recorded, and reported to the enterprise error monitor, +under a reason that names +a fault that never happened. + +The union-versus-CHECK set equality is asserted in the enterprise tree, where +cross-repository assertions belong and ``supabase/`` actually exists: +``reflexio_ext/tests/server/services/offline_tuner/test_replay_removal_allowlists.py``. +This module stays inside the OSS package so it keeps passing in a standalone +checkout, where there is no ``supabase/`` directory to read. +""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from reflexio.models.api_schema.domain.entities import PlaybookOptimizationJob + + +def test_a_fenced_job_row_parses() -> None: + """The regression: this construction raised ``ValidationError`` before.""" + job = PlaybookOptimizationJob( + job_id=1, + target_kind="user_playbook", + target_id=7, + status="failed", + stage="failed", + terminal_outcome="regeneration_fenced", + ) + + assert job.terminal_outcome == "regeneration_fenced" + + +def test_an_invented_outcome_is_still_rejected() -> None: + """Widening the union must not degrade it into a free-form string.""" + with pytest.raises(ValidationError): + PlaybookOptimizationJob( + job_id=1, + target_kind="user_playbook", + target_id=7, + terminal_outcome="not_a_real_outcome", # type: ignore[arg-type] + ) diff --git a/tests/models/test_terminal_outcome_reachability.py b/tests/models/test_terminal_outcome_reachability.py index 1a15e55bd..673e10566 100644 --- a/tests/models/test_terminal_outcome_reachability.py +++ b/tests/models/test_terminal_outcome_reachability.py @@ -25,8 +25,8 @@ _TERMINAL_OUTCOMES_BY_OPTIMIZER, ) -# The ten outcomes that survive Phase 7 with a path that can reach them. Six are -# written by the stage-advance allowlist below; the other four are written +# The twelve outcomes that survive Phase 7 with a path that can reach them. Six +# are written by the stage-advance allowlist below; the other six are written # elsewhere and are named here with their writer so the split is auditable. _REACHABLE_TERMINAL_OUTCOMES = frozenset( { @@ -38,6 +38,24 @@ "generation_failed", # the governance erasure path "governance_erased", + # the regeneration fence: reflexio_ext open_world/runner.py:251 calls + # _converge_terminal_failure with it, and the TENANT stage-advance RPC's + # 'failed' arm assigns it (tenant 20260830020000:325-327). It is + # deliberately absent from the SQLite allowlist below -- SQLite carries + # no open-world fence -- which is why it is named here rather than left + # to the `writable <=` assertion to cover. + "regeneration_fenced", + # the invocation-slot pin: reflexio_ext open_world/runner.py:262-264 + # calls _converge_terminal_failure with it on + # OpenWorldInvocationSlotExhaustedError -- the attempt made NO provider + # call because every row identity its question could occupy is owned by + # another job. The TENANT stage-advance RPC's 'failed' arm assigns it + # (tenant 20260903010000:129-131); the 'abstained' arm deliberately does + # not, since nothing was judged. Absent from the SQLite allowlist below + # for the same reason as 'regeneration_fenced' -- SQLite carries no + # open-world invocation table, so no slot can be pinned -- which is why + # it is named here rather than left to the `writable <=` assertion. + "invocation_slot_pinned", # stage-advance: 'failed' "infrastructure_failure", "analyst_unqualified", @@ -88,8 +106,8 @@ def test_the_union_is_exactly_the_reachable_set_plus_the_retained_set() -> None: assert ( members - RETAINED_UNREACHABLE_TERMINAL_OUTCOMES == _REACHABLE_TERMINAL_OUTCOMES ) - assert len(members) == 17 - assert len(_REACHABLE_TERMINAL_OUTCOMES) == 10 + assert len(members) == 19 + assert len(_REACHABLE_TERMINAL_OUTCOMES) == 12 def test_no_retained_outcome_is_writable_through_the_stage_advance_allowlist() -> None: diff --git a/tests/server/llm/test_litellm_client_unit.py b/tests/server/llm/test_litellm_client_unit.py index 37dc68cee..cd67ded0e 100644 --- a/tests/server/llm/test_litellm_client_unit.py +++ b/tests/server/llm/test_litellm_client_unit.py @@ -1595,6 +1595,42 @@ def test_zai_uses_coding_endpoint_and_prompt_backed_json_mode(self): assert parser_schema is SampleResponse assert parse_structured is True + def test_minimax_carries_the_schema_in_the_prompt_not_response_format(self): + """MiniMax ignores ``response_format``, so the schema must be in the prompt. + + Measured against the live API: two identical calls differing only in + ``drop_params`` both returned free prose rather than JSON, so a + ``json_schema`` response_format is discarded however it is sent. The + symptom was an analyst inventing a different set of field names on + every run -- answering from the prompt alone, never having been given + a schema. Asserting the field NAMES appear in the instruction is the + point: a bare ``{"type": "json_object"}`` would leave the model to + guess them, which is the defect this guards. + """ + client = _build_client( + LiteLLMConfig( + model="minimax/MiniMax-M3", + api_key_config=APIKeyConfig(minimax=MiniMaxConfig(api_key="mm-key")), + ) + ) + messages = [{"role": "user", "content": "test"}] + + params, parser_schema, parse_structured, _, _ = client._build_completion_params( + messages, + response_format=SampleResponse, + ) + + assert params["response_format"] == {"type": "json_object"} + instruction = params["messages"][0]["content"] + assert params["messages"][0]["role"] == "system" + assert "Return ONLY a JSON object" in instruction + assert '"answer"' in instruction + assert '"score"' in instruction + # The caller's messages are not mutated, and local parsing stays typed. + assert messages == [{"role": "user", "content": "test"}] + assert parser_schema is SampleResponse + assert parse_structured is True + def test_zai_tool_turn_leaves_tools_free_and_constrains_only_terminus(self): client = _build_client(LiteLLMConfig(model="zai/glm-5.2")) messages = [ @@ -1676,16 +1712,34 @@ def test_zai_strict_response_format_false_preserves_passthrough(self): assert params["messages"] == messages def test_openai_compatible_underreported_provider_uses_strict_schema(self): - # Regression: minimax reports - # supports_response_schema=False, but it is an OpenAI-compatible endpoint - # LiteLLM would still hand a self-built json_schema. We must send our own - # normalized strict schema instead of the raw Pydantic model. + # Covers the _JSON_SCHEMA_PROVIDER_ALLOWLIST mechanism: a provider that + # genuinely accepts a json_schema response_format but that LiteLLM + # reports as unsupported must receive our own normalized strict schema, + # not the raw Pydantic model. + # + # The allowlist is EMPTY in production -- minimax was its only member + # and moved to the prompt path, having turned out to ignore + # response_format outright. So this patches a member in to exercise the + # mechanism itself. Without that, the code path would have no coverage + # at all and the next provider added to it would be unguarded. client = _build_client(LiteLLMConfig(model="minimax/MiniMax-M3")) - with patch.object( - LiteLLMClient, - "_supports_response_schema", - return_value=False, + with ( + patch.object( + LiteLLMClient, + "_supports_response_schema", + return_value=False, + ), + patch.object( + LiteLLMClient, + "_JSON_SCHEMA_PROVIDER_ALLOWLIST", + frozenset({"minimax"}), + ), + patch.object( + LiteLLMClient, + "_PROMPT_SCHEMA_PROVIDER_ALLOWLIST", + frozenset({"zai"}), + ), ): params, parser_schema, parse_structured, _, _ = ( client._build_completion_params( @@ -1718,10 +1772,20 @@ def test_discriminated_union_strips_oneof_for_underreported_provider(self): # to exercise make_strict's prod backstop. The by-construction boundary # guard would (correctly) raise on it under pytest, so patch it to a no-op # here — this test asserts the make_strict fallback, not the guard. + # The json-schema allowlist is EMPTY in production (minimax moved to the + # prompt path), so a member is patched in to exercise the mechanism. with ( patch.object( LiteLLMClient, "_supports_response_schema", return_value=False ), + patch.object( + LiteLLMClient, + "_JSON_SCHEMA_PROVIDER_ALLOWLIST", + frozenset({"minimax"}), + ), + patch.object( + LiteLLMClient, "_PROMPT_SCHEMA_PROVIDER_ALLOWLIST", frozenset({"zai"}) + ), patch( "reflexio.server.llm._litellm_structured_output.assert_provider_safe_schema" ), @@ -1756,12 +1820,20 @@ def test_real_minimax_gate_normalizes_without_mocking_predicate(self): [{"role": "user", "content": "test"}], response_format=_DiscriminatedOutput, ) - provider_format = params["response_format"] - assert isinstance(provider_format, dict), ( - "minimax must receive a normalized strict schema, not the raw Pydantic " - "model" - ) - schema = provider_format["json_schema"]["schema"] + # minimax now carries its schema in the PROMPT -- it ignores + # response_format outright -- so the normalization invariant moved + # transport with it. It did not stop mattering: an unfolded `oneOf` + # trips assert_provider_safe_schema and raises before the request is + # even built, so a prompt-only provider could otherwise never carry a + # discriminated union at all. + assert params["response_format"] == {"type": "json_object"} + instruction = params["messages"][0]["content"] + # `prompt_schema_instruction` renders "\n\n", + # and json.dumps(indent=2) emits no blank lines, so the first blank + # line is an unambiguous separator. + prose, _, schema_text = instruction.partition("\n\n") + assert "JSON Schema" in prose + schema = json.loads(schema_text) assert not find_schema_keyword(schema, "oneOf") assert not find_schema_keyword(schema, "discriminator") diff --git a/tests/server/services/playbook_optimizer/test_decision_reason_vocabulary_guard.py b/tests/server/services/playbook_optimizer/test_decision_reason_vocabulary_guard.py new file mode 100644 index 000000000..9ac137310 --- /dev/null +++ b/tests/server/services/playbook_optimizer/test_decision_reason_vocabulary_guard.py @@ -0,0 +1,123 @@ +"""Guard: ``decision_reason`` writers must pass a fixed string, never an expression. + +The invariant +------------- +``playbook_optimization_jobs.decision_reason`` is a durable ``TEXT NOT NULL`` +column that is read straight back into ``PlaybookOptimizationJob`` and shown to +operators, and a Postgres trigger compares it against literal values. It is a +controlled vocabulary: every writer names a decision the optimizer made. + +An expression in that keyword is how customer content gets in. ``str(exc)`` on +an arbitrary exception is the concrete case -- a pydantic ``ValidationError`` +raised on a provider response renders the model's own output into its message, +so persisting the message persists evidence text. The class name, the traceback +and the tags belong in the structured-logging / error-reporting path instead, +which every failure site already uses. + +Why a guard rather than review +------------------------------ +The leak is invisible at the call site: ``decision_reason=str(exc)`` reads like +helpful diagnostics, produces no test failure, and only shows its teeth on the +one exception whose message happens to quote a transcript. A behavioural test +pins the one site it exercises; this scan pins the shape of every site. + +Scan approach +------------- +Parse every module under ``reflexio/server/`` with :mod:`ast` and collect each +call to a job writer that passes ``decision_reason=``. The value must be a +string literal, or a conditional expression whose branches are all string +literals (the committed / winner-persisted-only site). Anything else fails. + +Known blind spots (documented, not silently tolerated) +------------------------------------------------------ +* Calls are matched by attribute/function NAME, so a writer reached through an + alias or ``getattr`` is invisible. The non-vacuity test below fails if the + scan ever stops finding the known writers, which is what would happen if the + call shape changed wholesale. +* ``services/storage/`` is skipped. That layer does not DECIDE a reason: it + forwards the one it was handed and hydrates the stored value back into the + entity (``PlaybookOptimizationJob(decision_reason=row["decision_reason"])``), + which is a read and would otherwise be a permanent false positive. The + vocabulary is set by the services that call into it, which is what is scanned. +* Only this package is scanned. ``reflexio_ext`` calls + ``update_playbook_optimization_job`` in exactly one place + (``offline_tuner/open_world/runner.py``) and passes no ``decision_reason``; + raw SQL in either package that writes the column directly is also out of + scope -- those sites all assign literals today. +""" + +from __future__ import annotations + +import ast +from pathlib import Path + +SERVER_ROOT = Path(__file__).resolve().parents[4] / "reflexio" / "server" + +WRITER_NAMES = { + "update_playbook_optimization_job", + "create_playbook_optimization_job", + "PlaybookOptimizationJob", +} + + +def _called_name(func: ast.expr) -> str | None: + if isinstance(func, ast.Attribute): + return func.attr + if isinstance(func, ast.Name): + return func.id + return None + + +def _is_fixed_vocabulary(value: ast.expr) -> bool: + """True when the expression can only ever produce a fixed string.""" + if isinstance(value, ast.Constant): + return isinstance(value.value, str) + if isinstance(value, ast.IfExp): + return _is_fixed_vocabulary(value.body) and _is_fixed_vocabulary(value.orelse) + return False + + +def _decision_reason_writes() -> list[tuple[str, int, bool]]: + """Collect ``(relative_path, lineno, is_fixed)`` for every writer call.""" + found: list[tuple[str, int, bool]] = [] + for path in sorted(SERVER_ROOT.rglob("*.py")): + if "storage" in path.relative_to(SERVER_ROOT).parts: + continue + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for node in ast.walk(tree): + if not isinstance(node, ast.Call): + continue + if _called_name(node.func) not in WRITER_NAMES: + continue + for keyword in node.keywords: + if keyword.arg != "decision_reason": + continue + found.append( + ( + str(path.relative_to(SERVER_ROOT)), + keyword.value.lineno, + _is_fixed_vocabulary(keyword.value), + ) + ) + return found + + +def test_scan_actually_finds_the_known_writers(): + """Non-vacuity: a scan that finds nothing would pass the guard below.""" + writes = _decision_reason_writes() + assert len(writes) >= 3, f"scan found {len(writes)} writers; it has gone blind" + optimizer_writes = [ + w for w in writes if w[0].endswith("playbook_optimizer/optimizer.py") + ] + assert optimizer_writes, f"optimizer.py contributed no writers; scanned {writes}" + + +def test_every_decision_reason_write_is_a_fixed_string(): + dynamic = [ + (path, lineno) for path, lineno, fixed in _decision_reason_writes() if not fixed + ] + assert not dynamic, ( + "decision_reason must be a fixed controlled-vocabulary string, never an " + "expression -- an exception message or other interpolated value can carry " + f"customer content into a durable column. Offending writes: {dynamic}" + ) diff --git a/tests/server/services/playbook_optimizer/test_playbook_optimizer.py b/tests/server/services/playbook_optimizer/test_playbook_optimizer.py index d43fc2a6e..035e9c53e 100644 --- a/tests/server/services/playbook_optimizer/test_playbook_optimizer.py +++ b/tests/server/services/playbook_optimizer/test_playbook_optimizer.py @@ -74,6 +74,7 @@ def _sqlite_storage(tmp_path): def _optimizer_for_test(storage, config) -> PlaybookOptimizer: context = SimpleNamespace( storage=storage, + org_id="opt-test", configurator=SimpleNamespace(get_config=lambda: config), ) llm_client = SimpleNamespace(config=SimpleNamespace(model="fake-model")) @@ -1481,3 +1482,65 @@ def callback() -> PlaybookOptimizationRunStatus: ) assert fired.wait(timeout=5), "scheduled callback never fired" + + +def test_optimizer_failure_does_not_persist_exception_message(tmp_path): + """A failed run records a controlled reason, never the exception message. + + ``decision_reason`` is a durable column read back into the domain model and + shown to operators. An arbitrary exception message can carry customer + content (a pydantic ``ValidationError`` on a provider response renders the + model's own output into its message), so the failure path must not write it. + """ + storage = _sqlite_storage(tmp_path) + config = Config( + storage_config=StorageConfigSQLite(db_path=str(tmp_path / "reflexio.db")), + playbook_optimizer_config=PlaybookOptimizerConfig( + enabled=True, + optimize_agent_playbooks=True, + webhook_url="https://assistant.example.test/rollout", + auto_update_pending_agent_playbooks=True, + min_commit_windows=1, + min_commit_score=0.1, + min_commit_likert=1, + ), + ) + optimizer = _optimizer_for_test(storage, config) + incumbent = AgentPlaybook( + agent_playbook_id=1, + playbook_name="support", + agent_version="v1", + content="incumbent", + playbook_status=PlaybookStatus.PENDING, + ) + optimizer._load_incumbent = Mock(return_value=incumbent) # type: ignore[method-assign] + optimizer._resolve_windows = Mock( # type: ignore[method-assign] + return_value=[_scenario_window(idx) for idx in range(1, 6)] + ) + + # Stands in for customer content that a real exception message would carry. + sentinel = "SENTINEL-c7e41f-my-bank-account-is-frozen" + message = f"1 validation error for Output\n input_value='{sentinel}'" + assert sentinel in message, "test is vacuous unless the message carries it" + + def raising_run_gepa(*args, **kwargs): # noqa: ARG001 + raise ValueError(message) + + optimizer._run_gepa = raising_run_gepa # type: ignore[method-assign] + + status = optimizer.optimize( + PlaybookOptimizationTarget(kind="agent_playbook", target_id=1) + ) + + assert status == "failed" + rows = storage.conn.execute( + "SELECT status, decision_reason, metadata_json FROM playbook_optimization_jobs" + ).fetchall() + assert len(rows) == 1 + assert rows[0]["status"] == "failed" + persisted_reason = rows[0]["decision_reason"] + assert sentinel not in persisted_reason + assert sentinel not in rows[0]["metadata_json"] + # Pinned literally, not against the source constant: an assertion that + # imported the constant would follow it if someone changed it back. + assert persisted_reason == "optimization run raised an unexpected error"