From fdbd9c2d478218444c3ee824d0969d4384f34e9b Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Thu, 9 Jul 2026 12:24:19 -0600 Subject: [PATCH] fix(think): make schema-invalid generate output a terminal error _execute_generate previously wrote the model output to output_path whenever it was truthy and then emitted finish. get_use_end_state() classifies finish as a clean talent.complete, so a local model that emitted schema-violating JSON could both corrupt the output artifact and count as a successful daily completion. Cloud providers enforce schemas server-side, so this primarily affects the local provider path, which is the default engine direction. The new pre-write gate validates the candidate that would actually be written, using the load-bearing predicate output_path and result and not _output_valid_for_schema(result, ...). Using _schema_validation_clean here would have short-circuited on the raw-text advisory and fired for pulse and steward, whose post-hooks repair unparseable raw output into a valid default; that would turn graceful degradation into a terminal error. _schema_validation_clean is unchanged and keeps its one caller in the clean-provenance predicate, where the conservative raw short-circuit is correct. The output_path and result prefix is intentional. It excludes the story post-hook talents conversation, work, and event, which return an empty string on every path because their real output is a merge into the activity record. The error event includes schema_validation verbatim when present, but that field describes the raw provider text rather than the rejected post-hook candidate. It can therefore read valid: true when a hook produced invalid output. The human error message derives from errors[0] only when the raw text itself was invalid; otherwise it uses a generic fallback. schema_invalid now joins DETERMINISTIC_FAILURE_REASON_CODES so the existing threshold-2 backoff applies. The first schema-invalid run still gets a normal re-dispatch, so a stochastic local model that recovers next cadence is not penalized; only a talent that fails twice consecutively backs off. The set comment now reflects that these include high-recurrence stochastic failures, since "will crash identically" was already false for no_output. Fenced JSON is now an honest failure. tests/test_markdown_fence_strip.py::test_generate_json_output_not_stripped used to assert silent success for a fenced JSON payload; it now asserts terminal schema_invalid while preserving the original invariant that output: json is never fence-stripped. JSON fence-stripping remains a deliberate follow-up, not part of this change. The _output_valid_for_schema warning no longer says cached, since that was already wrong at two of its three call sites. _execute_with_tools also documents that no cogitate talent declares output: json or a schema today, and any future gate there should mirror the generate path. Co-Authored-By: Claude Opus 4.8 (1M context) --- solstone/think/cogitate_policy.py | 5 +- solstone/think/talents.py | 46 +++- tests/test_generate_full.py | 333 +++++++++++++++++++++++++++++ tests/test_markdown_fence_strip.py | 7 +- tests/test_pipeline_health.py | 53 +++++ 5 files changed, 437 insertions(+), 7 deletions(-) diff --git a/solstone/think/cogitate_policy.py b/solstone/think/cogitate_policy.py index 79ce9350a..f90c11ea8 100644 --- a/solstone/think/cogitate_policy.py +++ b/solstone/think/cogitate_policy.py @@ -35,14 +35,15 @@ MAX_TURNS_HEADROOM = 2 _FALLBACK_USD_PER_TOKEN = 0.0000025 DETERMINISTIC_FAILURE_THRESHOLD = 2 DEFAULT_READ_CALL_BUDGET = 200 -# Reason codes for content-deterministic crashes: re-dispatching a daily -# unit that hit one of these will crash identically. +# Reason codes for content-deterministic crashes and high-recurrence stochastic +# failures we decline to auto-retry past the threshold. DETERMINISTIC_FAILURE_REASON_CODES = frozenset( { "agent_stuck", "context_window_exceeded", "max_turns_exhausted", "no_output", + "schema_invalid", "token_budget_exceeded", "wall_clock_exceeded", } diff --git a/solstone/think/talents.py b/solstone/think/talents.py index 280b766ed..56d54bdb8 100644 --- a/solstone/think/talents.py +++ b/solstone/think/talents.py @@ -1077,9 +1077,7 @@ def _output_valid_for_schema( if runtime_json_schema is not None: Draft202012Validator(runtime_json_schema).validate(parsed) except Exception: - LOG.warning( - "cached JSON output failed current schema validation", exc_info=True - ) + LOG.warning("JSON output failed current schema validation", exc_info=True) return False return True @@ -1092,6 +1090,20 @@ def _schema_validation_clean(gen_result: dict, result: str, config: dict) -> boo return _output_valid_for_schema(result, config.get("output"), runtime_json_schema) +def _schema_invalid_message(gen_result: dict) -> str: + validation = gen_result.get("schema_validation") + if isinstance(validation, dict) and validation.get("valid") is False: + errors = validation.get("errors") + if errors: + first = errors[0] + text = ( + f"{first['path'] or ''}: " + f"{first['constraint']}: {first['message']}" + ) + return text if len(text) <= 200 else text[:197] + "..." + return "talent output failed JSON schema validation" + + def _terminal_unit(config: dict) -> TerminalUnit | None: day = config.get("day") mode = config.get("schedule") @@ -1401,6 +1413,7 @@ async def _execute_with_tools( completed_at_ms = now_ms() runtime_json_schema = hydrate_runtime_enums(config.get("json_schema")) + # No cogitate talent declares JSON output/schema today; if one does, mirror the generate-path terminal schema gate. schema_clean = _output_valid_for_schema( result, config.get("output"), @@ -1688,6 +1701,33 @@ async def _execute_generate( ) return + if ( + output_path + and result + and not _output_valid_for_schema( + result, + config.get("output"), + runtime_json_schema, + ) + ): + _mark_terminal_error_evented(config) + error_event: dict[str, Any] = { + "event": "error", + "error": _schema_invalid_message(gen_result), + "reason_code": "schema_invalid", + "provider": config.get("provider"), + "terminal": True, + "ts": now_ms(), + } + # Describes the raw provider text, not the post-hook candidate the gate + # rejected: it can read valid=True when a hook produced invalid output. + if "schema_validation" in gen_result: + error_event["schema_validation"] = gen_result["schema_validation"] + if retries: + error_event["retries"] = retries + emit_event(error_event) + return + # Write output output_changed = False if output_path and result: diff --git a/tests/test_generate_full.py b/tests/test_generate_full.py index d6145781c..9635c8e95 100644 --- a/tests/test_generate_full.py +++ b/tests/test_generate_full.py @@ -162,6 +162,339 @@ def test_execute_generate_blank_expected_output_emits_terminal_no_output( assert output_path.read_text(encoding="utf-8") == "old output" +def test_execute_generate_schema_invalid_emits_terminal_error(tmp_path, monkeypatch): + from solstone.think import models + from solstone.think.talents import _execute_generate + + output_path = tmp_path / "out.json" + validation = { + "valid": False, + "errors": [ + { + "path": "/summary", + "constraint": "type", + "message": "42 is not of type 'string'", + } + ], + } + monkeypatch.setattr( + models, + "generate_with_result", + lambda **kwargs: { + "text": '{"summary": 42}', + "usage": {"input_tokens": 1, "output_tokens": 3}, + "schema_validation": validation, + }, + ) + events: list[dict] = [] + + asyncio.run( + _execute_generate( + { + "provider": "google", + "model": "gemini-2.0-flash", + "name": "schema_bad", + "prompt": "x", + "output": "json", + "output_path": str(output_path), + "json_schema": { + "type": "object", + "required": ["summary"], + "properties": {"summary": {"type": "string"}}, + }, + }, + events.append, + ) + ) + + assert [event["event"] for event in events] == ["error"] + assert events[0]["reason_code"] == "schema_invalid" + assert events[0]["terminal"] is True + assert events[0]["schema_validation"] == validation + assert events[0]["schema_validation"]["valid"] is False + assert not output_path.exists() + + +def test_execute_generate_schema_invalid_preserves_existing_output( + tmp_path, monkeypatch +): + from solstone.think import models + from solstone.think.talents import _execute_generate + + output_path = tmp_path / "out.json" + original = b'{"summary": "old"}' + output_path.write_bytes(original) + validation = { + "valid": False, + "errors": [ + { + "path": "/summary", + "constraint": "type", + "message": "42 is not of type 'string'", + } + ], + } + monkeypatch.setattr( + models, + "generate_with_result", + lambda **kwargs: { + "text": '{"summary": 42}', + "usage": {"input_tokens": 1, "output_tokens": 3}, + "schema_validation": validation, + }, + ) + events: list[dict] = [] + + asyncio.run( + _execute_generate( + { + "provider": "google", + "model": "gemini-2.0-flash", + "name": "schema_bad_existing", + "prompt": "x", + "output": "json", + "output_path": str(output_path), + "json_schema": { + "type": "object", + "required": ["summary"], + "properties": {"summary": {"type": "string"}}, + }, + }, + events.append, + ) + ) + + assert output_path.read_bytes() == original + assert [event["event"] for event in events] == ["error"] + assert events[0]["reason_code"] == "schema_invalid" + assert events[0]["terminal"] is True + + +def test_execute_generate_schema_clean_writes_file_and_clean_provenance( + tmp_path, monkeypatch +): + from solstone.think import models + from solstone.think.talent_provenance import read_provenance + from solstone.think.talents import _execute_generate + + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + output_path = tmp_path / "chronicle" / "20240101" / "talents" / "schema_clean.json" + validation = {"valid": True, "errors": []} + monkeypatch.setattr( + models, + "generate_with_result", + lambda **kwargs: { + "text": '{"summary": "ok"}', + "usage": {"input_tokens": 1, "output_tokens": 3}, + "schema_validation": validation, + }, + ) + events: list[dict] = [] + + asyncio.run( + _execute_generate( + { + "provider": "google", + "model": "gemini-2.0-flash", + "name": "schema_clean", + "prompt": "x", + "day": "20240101", + "schedule": "daily", + "output": "json", + "output_path": str(output_path), + "json_schema": { + "type": "object", + "required": ["summary"], + "properties": {"summary": {"type": "string"}}, + }, + }, + events.append, + ) + ) + + assert [event["event"] for event in events] == ["finish"] + assert not [event for event in events if event["event"] == "error"] + assert output_path.read_text(encoding="utf-8") == '{"summary": "ok"}' + provenance = read_provenance(output_path) + assert provenance is not None + assert provenance["output_path"] == "chronicle/20240101/talents/schema_clean.json" + + +def test_execute_generate_md_output_without_schema_still_writes(tmp_path, monkeypatch): + from solstone.think import models + from solstone.think.talents import _execute_generate + + output_path = tmp_path / "out.md" + monkeypatch.setattr( + models, + "generate_with_result", + lambda **kwargs: { + "text": "plain markdown", + "usage": {"input_tokens": 1, "output_tokens": 3}, + }, + ) + events: list[dict] = [] + + asyncio.run( + _execute_generate( + { + "provider": "google", + "model": "gemini-2.0-flash", + "name": "md_gen", + "prompt": "x", + "output": "md", + "output_path": str(output_path), + }, + events.append, + ) + ) + + assert [event["event"] for event in events] == ["finish"] + assert output_path.read_text(encoding="utf-8") == "plain markdown" + + +def test_execute_generate_json_empty_post_hook_result_finishes_without_write( + tmp_path, monkeypatch +): + from solstone.think import models, talents + from solstone.think.talents import _execute_generate + + output_path = tmp_path / "out.json" + monkeypatch.setattr( + models, + "generate_with_result", + lambda **kwargs: { + "text": '{"summary": "ok"}', + "usage": {"input_tokens": 1, "output_tokens": 3}, + "schema_validation": {"valid": True, "errors": []}, + }, + ) + monkeypatch.setattr( + talents, "load_post_hook", lambda config: lambda result, ctx: "" + ) + events: list[dict] = [] + + asyncio.run( + _execute_generate( + { + "provider": "google", + "model": "gemini-2.0-flash", + "name": "empty_hook", + "prompt": "x", + "output": "json", + "output_path": str(output_path), + "json_schema": { + "type": "object", + "required": ["summary"], + "properties": {"summary": {"type": "string"}}, + }, + }, + events.append, + ) + ) + + assert [event["event"] for event in events] == ["finish"] + assert not [event for event in events if event["event"] == "error"] + assert not output_path.exists() + + +def test_execute_generate_json_without_schema_nonparseable_emits_schema_invalid( + tmp_path, monkeypatch +): + from solstone.think import models + from solstone.think.talents import _execute_generate + + output_path = tmp_path / "out.json" + monkeypatch.setattr( + models, + "generate_with_result", + lambda **kwargs: { + "text": "not json", + "usage": {"input_tokens": 1, "output_tokens": 3}, + }, + ) + events: list[dict] = [] + + asyncio.run( + _execute_generate( + { + "provider": "google", + "model": "gemini-2.0-flash", + "name": "json_no_schema_bad", + "prompt": "x", + "output": "json", + "output_path": str(output_path), + }, + events.append, + ) + ) + + assert [event["event"] for event in events] == ["error"] + assert events[0]["reason_code"] == "schema_invalid" + assert events[0]["terminal"] is True + assert events[0]["error"] == "talent output failed JSON schema validation" + assert not output_path.exists() + + +def test_execute_generate_invalid_raw_repaired_by_hook_still_finishes( + tmp_path, monkeypatch +): + from solstone.think import models, talents + from solstone.think.talents import _execute_generate + + output_path = tmp_path / "out.json" + repaired = '{"summary": "repaired"}' + validation = { + "valid": False, + "errors": [ + { + "path": "", + "constraint": "json_parse", + "message": "Expecting value", + } + ], + } + monkeypatch.setattr( + models, + "generate_with_result", + lambda **kwargs: { + "text": "not json", + "usage": {"input_tokens": 1, "output_tokens": 3}, + "schema_validation": validation, + }, + ) + # Pulse/steward repair hooks depend on this raw-invalid/result-valid quadrant. + monkeypatch.setattr( + talents, + "load_post_hook", + lambda config: lambda result, ctx: repaired, + ) + events: list[dict] = [] + + asyncio.run( + _execute_generate( + { + "provider": "google", + "model": "gemini-2.0-flash", + "name": "repaired_hook", + "prompt": "x", + "output": "json", + "output_path": str(output_path), + "json_schema": { + "type": "object", + "required": ["summary"], + "properties": {"summary": {"type": "string"}}, + }, + }, + events.append, + ) + ) + + assert [event["event"] for event in events] == ["finish"] + assert not [event for event in events if event["event"] == "error"] + assert output_path.read_text(encoding="utf-8") == repaired + + def test_no_output_does_not_log_day_ok(tmp_path, monkeypatch): mod = importlib.import_module("solstone.think.talents") copy_day(tmp_path, monkeypatch) diff --git a/tests/test_markdown_fence_strip.py b/tests/test_markdown_fence_strip.py index 38bb52bae..907f99da6 100644 --- a/tests/test_markdown_fence_strip.py +++ b/tests/test_markdown_fence_strip.py @@ -183,6 +183,9 @@ def test_generate_json_output_not_stripped(tmp_path, monkeypatch, caplog): ) finish_events = [event for event in events if event["event"] == "finish"] - assert len(finish_events) == 1 - assert finish_events[0]["result"] == wrapped_text + assert finish_events == [] + error_events = [event for event in events if event["event"] == "error"] + assert len(error_events) == 1 + assert error_events[0]["reason_code"] == "schema_invalid" + assert error_events[0]["error"] == "talent output failed JSON schema validation" assert all(FENCE_STRIP_LOG not in record.getMessage() for record in caplog.records) diff --git a/tests/test_pipeline_health.py b/tests/test_pipeline_health.py index 117b57e5f..07383f419 100644 --- a/tests/test_pipeline_health.py +++ b/tests/test_pipeline_health.py @@ -15,6 +15,7 @@ from pathlib import Path import pytest from solstone.think import catchup_state +from solstone.think.cogitate_policy import DETERMINISTIC_FAILURE_REASON_CODES from solstone.think.pipeline_health import ( BACKLOG_STATE_COMPLETE, BACKLOG_STATE_PENDING, @@ -579,6 +580,58 @@ def test_read_daily_deterministic_failures(pipeline_journal): assert result[("facet_newsletter", "personal")].count == 1 +def test_schema_invalid_is_daily_deterministic_failure(pipeline_journal): + assert "schema_invalid" in DETERMINISTIC_FAILURE_REASON_CODES + + day = "20990217" + base = pipeline_journal / "chronicle" / day / "health" + _write_jsonl( + base / "001_daily.jsonl", + [ + { + "event": "talent.fail", + "ts": 1, + "mode": "daily", + "name": "schema_bad", + "reason_code": "schema_invalid", + }, + { + "event": "talent.fail", + "ts": 2, + "mode": "daily", + "name": "schema_bad", + "reason_code": "schema_invalid", + }, + { + "event": "talent.fail", + "ts": 1, + "mode": "daily", + "name": "schema_recovered", + "reason_code": "schema_invalid", + }, + { + "event": "talent.fail", + "ts": 2, + "mode": "daily", + "name": "schema_recovered", + "reason_code": "schema_invalid", + }, + { + "event": "talent.complete", + "ts": 3, + "mode": "daily", + "name": "schema_recovered", + }, + ], + ) + + result = read_daily_deterministic_failures(day) + + assert result[("schema_bad", None)].count == 2 + assert result[("schema_bad", None)].reason_code == "schema_invalid" + assert ("schema_recovered", None) not in result + + def test_read_completed_units_skips_malformed_records(pipeline_journal): day = "20990207" path = pipeline_journal / "chronicle" / day / "health" / "001_daily.jsonl" -- 2.51.2