Something went wrong. Try again.
personal memory agent
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732173317341735173617371738173917401741174217431744174517461747174817491750175117521753175417551756175717581759176017611762176317641765176617671768176917701771177217731774177517761777177817791780178117821783178417851786178717881789179017911792179317941795179617971798179918001801180218031804180518061807180818091810181118121813181418151816181718181819182018211822182318241825182618271828182918301831183218331834183518361837183818391840184118421843184418451846184718481849185018511852185318541855185618571858185918601861186218631864186518661867186818691870187118721873187418751876187718781879188018811882188318841885188618871888188918901891189218931894189518961897189818991900190119021903190419051906190719081909191019111912191319141915191619171918191919201921192219231924192519261927192819291930193119321933193419351936193719381939194019411942194319441945194619471948194919501951195219531954195519561957195819591960196119621963196419651966196719681969197019711972197319741975197619771978197919801981198219831984198519861987198819891990199119921993199419951996199719981999200020012002200320042005200620072008200920102011201220132014201520162017201820192020202120222023202420252026202720282029203020312032203320342035203620372038203920402041204220432044204520462047204820492050205120522053205420552056205720582059206020612062206320642065206620672068206920702071207220732074207520762077207820792080208120822083208420852086208720882089209020912092209320942095209620972098209921002101210221032104210521062107210821092110211121122113211421152116211721182119212021212122212321242125212621272128212921302131213221332134213521362137213821392140214121422143214421452146214721482149215021512152215321542155215621572158215921602161216221632164216521662167216821692170217121722173217421752176217721782179218021812182218321842185218621872188218921902191219221932194219521962197219821992200220122022203220422052206220722082209221022112212221322142215221622172218221922202221222222232224222522262227222822292230223122322233223422352236223722382239224022412242224322442245224622472248224922502251225222532254225522562257225822592260226122622263226422652266226722682269227022712272227322742275227622772278227922802281228222832284228522862287228822892290229122922293229422952296229722982299230023012302230323042305230623072308230923102311231223132314231523162317231823192320232123222323232423252326232723282329233023312332233323342335233623372338233923402341234223432344234523462347234823492350235123522353235423552356235723582359236023612362236323642365236623672368236923702371237223732374237523762377237823792380238123822383238423852386238723882389239023912392239323942395239623972398239924002401240224032404240524062407240824092410241124122413241424152416241724182419242024212422242324242425242624272428242924302431243224332434243524362437243824392440244124422443244424452446244724482449245024512452245324542455245624572458245924602461246224632464246524662467246824692470247124722473247424752476247724782479248024812482248324842485248624872488248924902491249224932494249524962497249824992500250125022503250425052506250725082509251025112512251325142515251625172518251925202521252225232524252525262527252825292530253125322533253425352536253725382539254025412542254325442545254625472548254925502551255225532554255525562557255825592560256125622563256425652566256725682569257025712572257325742575257625772578257925802581258225832584258525862587258825892590259125922593259425952596259725982599260026012602260326042605260626072608260926102611261226132614261526162617261826192620262126222623262426252626262726282629263026312632263326342635263626372638263926402641264226432644264526462647264826492650265126522653265426552656265726582659266026612662266326642665266626672668266926702671267226732674267526762677267826792680268126822683268426852686268726882689269026912692269326942695269626972698269927002701270227032704270527062707270827092710271127122713271427152716271727182719272027212722272327242725272627272728272927302731273227332734273527362737273827392740274127422743274427452746274727482749275027512752275327542755275627572758275927602761276227632764276527662767276827692770277127722773277427752776277727782779278027812782278327842785278627872788278927902791279227932794279527962797279827992800280128022803280428052806280728082809281028112812281328142815281628172818281928202821282228232824282528262827282828292830283128322833283428352836283728382839284028412842284328442845284628472848284928502851285228532854285528562857285828592860286128622863286428652866286728682869287028712872287328742875# SPDX-License-Identifier: AGPL-3.0-only# Copyright (c) 2026 sol pbc
"""Tests for think.models module."""
import asyncioimport jsonimport loggingimport timefrom pathlib import Pathfrom types import SimpleNamespacefrom unittest.mock import AsyncMock, MagicMock, patch
import pytest
import solstone.think.models as models_modulefrom solstone.think.models import ( CLAUDE_OPUS_4, CLAUDE_SONNET_4, DEFAULT_MODEL_BY_PROVIDER, DEFAULT_PROVIDER_TIMEOUT_S, GEMINI_FLASH, GEMMA4_26B_A4B_4BIT, GPT_5_MINI, LOCAL_MODEL, NO_BRAIN_PROVIDER, PROMPT_PATHS, QWEN_35_9B, IncompleteJSONError, IncompleteTextError, NoBrainConfiguredError, ProviderResponseInvalidError, SchemaValidationError, _Family, _find_pricing_fallback, _parse_family_anthropic, _parse_family_gemini, _parse_family_openai, _validate_schema, agenerate, agenerate_with_result, calc_agent_cost, calc_token_cost, default_model_for_provider, finish_reason_error, generate, generate_with_result, get_context_registry, get_model_provider, get_usage_cost, is_local_provider_needed, iter_token_log, model_supports, resolve_provider,)from solstone.think.responsiveness import ( NON_RESPONSIVE_RAW_OUTPUT_CAP_CHARS, NON_RESPONSIVE_REASON_CODE, NonResponsiveOutputError,)from solstone.think.schema_prep import SCHEMA_TRUNCATE_KEY
_FINISH_METADATA_MARKER = "zq7marker9x"_NON_RESPONSIVE_REFUSAL = "I cannot describe this screen."_RESPONSIVE_CHAT_FIXTURE = { "message": "The release check passed and Noah owns the follow-up.", "notes": "The owner asked about the release, so I summarized the check result.", "talent_request": None,}_REFUSAL_CHAT_FIXTURE = { "message": _NON_RESPONSIVE_REFUSAL, "notes": "The owner asked about the release, so I summarized the check result.", "talent_request": None,}_GENERATE_USAGE = { "input_tokens": 11, "output_tokens": 3, "reasoning_tokens": 2, "total_tokens": 16,}
def _fresh_generate_result(text: str): def make_result(*_args, **_kwargs): return {"text": text, "finish_reason": "stop"}
return make_result
def _generate_result( text: str, *, usage: dict | None = None, finish_reason: str = "stop", model: str = "provider-model",) -> dict: result = {"text": text, "model": model, "finish_reason": finish_reason} if usage is not None: result["usage"] = usage return result
def _provider_module_for_result(result: dict) -> SimpleNamespace: return SimpleNamespace( run_generate=MagicMock(return_value=result), run_agenerate=AsyncMock(return_value=result), )
def _call_generate_entrypoint(entrypoint: str, **kwargs): if entrypoint == "generate": return generate("hello", "test.context", **kwargs) if entrypoint == "generate_with_result": return generate_with_result("hello", "test.context", **kwargs) if entrypoint == "agenerate_with_result": return asyncio.run(agenerate_with_result("hello", "test.context", **kwargs)) if entrypoint == "agenerate": return asyncio.run(agenerate("hello", "test.context", **kwargs)) raise AssertionError(entrypoint)
def test_calc_token_cost_basic(): """Test basic cost calculation with a known model.""" token_data = { "model": "gpt-4o", "usage": { "input_tokens": 1000, "output_tokens": 100, "total_tokens": 1100, }, }
result = calc_token_cost(token_data)
assert result is not None assert "total_cost" in result assert "input_cost" in result assert "output_cost" in result assert "currency" in result assert result["currency"] == "USD" assert result["total_cost"] > 0 assert result["input_cost"] > 0 assert result["output_cost"] > 0
def test_calc_token_cost_with_cache(): """Test cost calculation with cached tokens.""" token_data = { "model": "claude-sonnet-4-20250514", "usage": { "input_tokens": 1000, "output_tokens": 100, "cached_tokens": 500, "total_tokens": 1600, }, }
result = calc_token_cost(token_data)
assert result is not None assert result["total_cost"] > 0 # Cached tokens should reduce the cost compared to all uncached assert result["input_cost"] >= 0
def test_calc_agent_cost_uses_resolved_model_version_from_usage(): usage = { "input_tokens": 1000, "output_tokens": 500, "model_version": "gemini-2.5-flash", }
cost = calc_agent_cost("gemini-3.5-flash", usage)
assert cost is not None and cost > 0
def test_calc_token_cost_unknown_model(): """Test that unknown models return None.""" token_data = { "model": "random-model-xyz", "usage": { "input_tokens": 1000, "output_tokens": 100, }, }
result = calc_token_cost(token_data) assert result is None
def test_get_model_provider_mlx_backend_models_are_local(): assert get_model_provider(GEMMA4_26B_A4B_4BIT) == "local" assert get_model_provider(QWEN_35_9B) == "local"
@pytest.mark.parametrize("reason", ["length", "max_tokens", "MAX_TOKENS", " Length "])def test_incomplete_json_error_sets_length_reason_code(reason): exc = IncompleteJSONError(reason, "")
assert exc.reason_code == "incomplete_json_length"
@pytest.mark.parametrize("reason", ["safety", "content_filter", "recitation", "error"])def test_incomplete_json_error_non_length_reasons_have_no_reason_code(reason): exc = IncompleteJSONError(reason, "")
assert not hasattr(exc, "reason_code")
def test_incomplete_json_error_preserves_positional_and_keyword_construction(): positional = IncompleteJSONError("length", "partial") keyword = IncompleteJSONError(reason="max_tokens", partial_text="body")
assert positional.reason == "length" assert positional.partial_text == "partial" assert keyword.reason == "max_tokens" assert keyword.partial_text == "body" assert positional.reason_code == "incomplete_json_length" assert keyword.reason_code == "incomplete_json_length"
def test_classify_provider_error_uses_incomplete_json_reason_code(): from solstone.think.providers.shared import classify_provider_error
assert ( classify_provider_error(IncompleteJSONError("length", ""), "local") == "incomplete_json_length" ) assert ( classify_provider_error(IncompleteJSONError("safety", ""), "local") != "incomplete_json_length" )
def test_calc_token_cost_gemma4_zero_cost(): token_data = { "model": GEMMA4_26B_A4B_4BIT, "usage": { "input_tokens": 1000, "output_tokens": 100, }, }
assert calc_token_cost(token_data) == { "total_cost": 0.0, "input_cost": 0.0, "output_cost": 0.0, "currency": "USD", }
def test_calc_token_cost_missing_data(): """Test that missing data returns None.""" # Missing model assert calc_token_cost({"usage": {"input_tokens": 1000}}) is None
# Missing usage assert calc_token_cost({"model": "gpt-4o"}) is None
# Empty dict assert calc_token_cost({}) is None
def test_calc_token_cost_with_reasoning_tokens(): """Test cost calculation includes reasoning tokens in output.""" token_data = { "model": "gpt-4o", "usage": { "input_tokens": 1000, "output_tokens": 100, "reasoning_tokens": 50, "total_tokens": 1150, }, }
result = calc_token_cost(token_data)
# Should succeed - reasoning tokens are implicitly part of output pricing assert result is not None assert result["total_cost"] > 0
# ---------------------------------------------------------------------------# resolve_provider tests# ---------------------------------------------------------------------------
@pytest.fixturedef use_fixtures_journal(monkeypatch): """Use the fixtures journal for provider config tests.""" monkeypatch.setenv("SOLSTONE_JOURNAL", "tests/fixtures/journal")
def _write_tmp_journal_config( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, config: dict) -> None: config_dir = tmp_path / "config" config_dir.mkdir() (config_dir / "journal.json").write_text(json.dumps(config)) monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
def test_resolve_provider_default_generate(use_fixtures_journal): """Generate resolves the configured active provider/model.""" provider, model = resolve_provider("generate") assert provider == "google" assert model == "gemini-custom-flash-test"
def test_resolve_provider_default_cogitate(use_fixtures_journal): """Cogitate resolves the configured active provider/model.""" provider, model = resolve_provider("cogitate") assert provider == "google" assert model == "gemini-custom-flash-test"
def test_resolve_provider_contexts_are_inert(monkeypatch, tmp_path): """Legacy exact/glob contexts cannot influence active route resolution.""" _write_tmp_journal_config( tmp_path, monkeypatch, { "providers": { "active": { "provider": "google", "model": "gemini-3.5-flash", }, "contexts": { "test.openai": { "provider": "openai", "model": "gpt-5-mini", }, "observe.*": { "provider": "anthropic", "model": "claude-haiku-4-5", }, }, } }, )
provider, model = resolve_provider("generate") assert provider == "google" assert model == GEMINI_FLASH
def test_resolve_provider_ordering_witness(monkeypatch, tmp_path): """Explicit provider/model wins over key-presence fallback.""" _write_tmp_journal_config( tmp_path, monkeypatch, { "providers": { "active": { "provider": "anthropic", "model": "claude-haiku-4-5", } } }, ) monkeypatch.setenv("OPENAI_API_KEY", "test-openai-key")
assert resolve_provider("generate") == ("anthropic", "claude-haiku-4-5")
def test_resolve_provider_empty_context(use_fixtures_journal): """The resolver takes only an interface, not a context.""" assert resolve_provider("generate") == ("google", "gemini-custom-flash-test")
def test_resolve_provider_no_config(monkeypatch, tmp_path): """Test no-brain resolution when no provider config exists.""" # Use a journal path with no config empty_journal = tmp_path / "empty_journal" empty_journal.mkdir() monkeypatch.setenv("SOLSTONE_JOURNAL", str(empty_journal)) monkeypatch.delenv("GOOGLE_API_KEY", raising=False) monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("OPENAI_API_KEY", raising=False) provider, model = resolve_provider("generate") assert provider == NO_BRAIN_PROVIDER assert provider != "google" assert model == ""
provider, model = resolve_provider("cogitate") assert provider == NO_BRAIN_PROVIDER assert provider != "google" assert model == ""
def test_prompt_paths_exist(): """Test all PROMPT_PATHS files exist and have valid frontmatter.""" from pathlib import Path
import frontmatter
base_dir = Path(__file__).parent.parent / "solstone" # Package root required_keys = {"context", "label", "group"}
for rel_path in PROMPT_PATHS: path = base_dir / rel_path assert path.exists(), f"Prompt file not found: {rel_path}"
post = frontmatter.load(path) meta = post.metadata or {}
assert required_keys <= set(meta.keys()), ( f"{rel_path} missing keys: {required_keys - set(meta.keys())}" ) assert isinstance(meta["label"], str) and meta["label"], ( f"{rel_path} has invalid label: {meta['label']}" ) assert isinstance(meta["group"], str) and meta["group"], ( f"{rel_path} has invalid group: {meta['group']}" )
def test_prompt_contexts_in_registry(): """Test prompt contexts are discovered and in registry.""" registry = get_context_registry()
# Verify known prompt contexts exist with correct values assert "observe.describe.frame" in registry assert registry["observe.describe.frame"]["group"] == "Observe"
assert "detect.created" in registry
def test_default_model_by_provider(): assert DEFAULT_MODEL_BY_PROVIDER == { "google": GEMINI_FLASH, "openai": GPT_5_MINI, "anthropic": CLAUDE_SONNET_4, "local": LOCAL_MODEL, } for provider, model in DEFAULT_MODEL_BY_PROVIDER.items(): assert default_model_for_provider(provider) == model
@pytest.mark.parametrize( "config", [ {"providers": {"active": {"provider": "local"}}}, ],)def test_is_local_provider_needed_true_for_selected_surfaces(config): assert is_local_provider_needed(config) is True
@pytest.mark.parametrize( "config", [ {}, {"providers": {"active": {"provider": "google"}}}, {"providers": {"contexts": {"talent.*": {"provider": "anthropic"}}}}, {"providers": {"contexts": {"talent.*": {"provider": "local"}}}}, {"providers": []}, ],)def test_is_local_provider_needed_false_when_not_selected(config, monkeypatch): assert is_local_provider_needed(config) is False
def test_is_local_provider_needed_does_not_infer_from_runtime(monkeypatch): monkeypatch.delenv("GOOGLE_API_KEY", raising=False) monkeypatch.delenv("ANTHROPIC_API_KEY", raising=False) monkeypatch.delenv("OPENAI_API_KEY", raising=False)
assert is_local_provider_needed({}) is False
def test_resolve_provider_legacy_keys_are_inert( use_fixtures_journal, monkeypatch, tmp_path): """Tier/backup/contexts/models legacy keys do not affect active routing.""" _write_tmp_journal_config( tmp_path, monkeypatch, { "providers": { "active": { "provider": "google", "model": "gemini-3.5-flash", "tier": 1, "backup": "anthropic", }, "contexts": { "talent.timeline.segment_summary": {"provider": "local"}, "observe.*": {"provider": "anthropic", "tier": 3}, }, "models": { "google": {"1": "gemini-3.5-flash"}, "anthropic": {"3": "claude-haiku-4-5"}, }, } }, )
assert resolve_provider("generate") == ("google", GEMINI_FLASH)
def test_resolve_provider_model_key_wins_even_when_tier_present( use_fixtures_journal, monkeypatch, tmp_path): """The retired AC3a quirk is gone: model is honored even with tier present.""" _write_tmp_journal_config( tmp_path, monkeypatch, { "providers": { "active": { "provider": "google", "tier": 1, "model": "gemini-custom-flash-test", } } }, )
assert resolve_provider("generate") == ("google", "gemini-custom-flash-test")
def test_resolve_provider_local_type_default_ignores_context_pins( use_fixtures_journal, monkeypatch, tmp_path): """Context pins cannot push a local active interface onto cloud.""" _write_tmp_journal_config( tmp_path, monkeypatch, { "providers": { "active": {"provider": "local"}, "contexts": { "talent.timeline.segment_summary": { "provider": "google", "model": "gemini-3.1-flash-lite", }, }, } }, )
assert resolve_provider("generate") == ("local", LOCAL_MODEL)
def test_legacy_context_toggles_remain_on_disk_but_not_routing( use_fixtures_journal, monkeypatch, tmp_path): """Grandfathered context keys stay inert next to disabled/extract toggles.""" _write_tmp_journal_config( tmp_path, monkeypatch, { "providers": { "active": {"provider": "google", "model": GEMINI_FLASH}, }, "talent_overrides": {"talent.x": {"disabled": True, "extract": "foo"}}, }, )
assert resolve_provider("generate") == ("google", GEMINI_FLASH)
stored = json.loads((tmp_path / "config" / "journal.json").read_text()) context = stored["talent_overrides"]["talent.x"] assert context["disabled"] is True assert context["extract"] == "foo"
def test_prepare_config_legacy_context_routing_keys_are_inert( journal_copy: Path,) -> None: """Legacy contexts survive on disk but cannot change prepared identity.""" from solstone.think.talents import prepare_config
config_path = journal_copy / "config" / "journal.json" config = json.loads(config_path.read_text(encoding="utf-8")) config["providers"] = { "active": {"provider": "anthropic", "model": CLAUDE_SONNET_4}, } config["talent_overrides"] = { "talent.timeline.segment_summary": { "disabled": False, } } config_path.write_text(json.dumps(config), encoding="utf-8")
prepared = prepare_config({"name": "timeline:segment_summary"})
assert prepared["provider"] == "anthropic" assert prepared["model"] == CLAUDE_SONNET_4
def test_prepare_config_frontmatter_provider_pin_is_dead_through_dispatch_identity( journal_copy: Path,) -> None: """Removed google frontmatter pins do not override the active brain.""" from solstone.think.talents import prepare_config
config_path = journal_copy / "config" / "journal.json" config = json.loads(config_path.read_text(encoding="utf-8")) config["env"] = {"GOOGLE_API_KEY": "test-google-key"} config["providers"] = { "active": {"provider": "anthropic", "model": CLAUDE_SONNET_4} } config_path.write_text(json.dumps(config), encoding="utf-8")
segment = prepare_config({"name": "timeline:segment_summary"}) detection = prepare_config({"name": "entities:detection"})
assert segment["provider"] == "anthropic" assert segment["model"] == CLAUDE_SONNET_4 assert detection["provider"] == "anthropic" assert detection["model"] == CLAUDE_SONNET_4
def test_resolve_provider_cogitate_system_talents_stay_local( use_fixtures_journal, monkeypatch, tmp_path): """Cogitate system talents stay local when the cogitate lane is local.""" _write_tmp_journal_config( tmp_path, monkeypatch, {"providers": {"active": {"provider": "local"}}}, )
assert resolve_provider("cogitate") == ("local", LOCAL_MODEL)
def test_generate_and_cogitate_share_one_active_profile( use_fixtures_journal, monkeypatch, tmp_path): """Both runtime interfaces resolve the same active profile.""" _write_tmp_journal_config( tmp_path, monkeypatch, { "providers": { "active": {"provider": "openai"}, } }, )
assert resolve_provider("generate") == ("openai", GPT_5_MINI) assert resolve_provider("cogitate") == ("openai", GPT_5_MINI)
@pytest.mark.parametrize( "call", [ lambda: generate("hello", "test.context", model="gpt-5.5"), lambda: generate_with_result("hello", "test.context", model="gpt-5.5"), lambda: asyncio.run(agenerate("hello", "test.context", model="gpt-5.5")), ],)def test_generate_entry_points_reject_request_model_overrides(call): """Provider/model selection belongs exclusively to the active profile.""" with pytest.raises(TypeError, match="unexpected keyword argument 'model'"): call()
def test_local_generate_entry_points_do_not_pass_cloud_transport_options(): provider_module = SimpleNamespace( run_generate=MagicMock(return_value={"text": "ok", "finish_reason": "stop"}), run_agenerate=AsyncMock(return_value={"text": "ok", "finish_reason": "stop"}), ) with ( patch( "solstone.think.models.resolve_provider", return_value=("local", LOCAL_MODEL), ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), ): assert generate("hello", "test.context") == "ok" assert asyncio.run(agenerate("hello", "test.context")) == "ok"
assert "provider" not in provider_module.run_generate.call_args.kwargs assert "provider" not in provider_module.run_agenerate.call_args.kwargs
def test_generate_stops_before_provider_module_when_no_brain(): with ( patch( "solstone.think.models.resolve_provider", return_value=(NO_BRAIN_PROVIDER, ""), ), patch( "solstone.think.providers.get_provider_module", side_effect=AssertionError("provider module should not be invoked"), ), ): with pytest.raises(NoBrainConfiguredError): generate("hello", "test.context")
def test_generate_with_result_stops_before_provider_module_when_no_brain(): with ( patch( "solstone.think.models.resolve_provider", return_value=(NO_BRAIN_PROVIDER, ""), ), patch( "solstone.think.providers.get_provider_module", side_effect=AssertionError("provider module should not be invoked"), ), ): with pytest.raises(NoBrainConfiguredError): generate_with_result("hello", "test.context")
def test_agenerate_stops_before_provider_module_when_no_brain(): with ( patch( "solstone.think.models.resolve_provider", return_value=(NO_BRAIN_PROVIDER, ""), ), patch( "solstone.think.providers.get_provider_module", side_effect=AssertionError("provider module should not be invoked"), ), ): with pytest.raises(NoBrainConfiguredError): asyncio.run(agenerate("hello", "test.context"))
# ---------------------------------------------------------------------------# Dynamic context registry tests# ---------------------------------------------------------------------------
def test_context_registry_includes_prompt_contexts(): """Test that registry includes all contexts from PROMPT_PATHS.""" from pathlib import Path
import frontmatter
registry = get_context_registry() base_dir = Path(__file__).parent.parent / "solstone"
# All prompt contexts should be in registry with matching metadata for rel_path in PROMPT_PATHS: path = base_dir / rel_path post = frontmatter.load(path) meta = post.metadata or {} context = meta.get("context")
assert context in registry, f"Prompt context {context} not in registry" assert registry[context]["label"] == meta["label"] assert registry[context]["group"] == meta["group"]
def test_context_registry_includes_categories(): """Test that registry includes discovered category contexts.""" registry = get_context_registry()
# Should have category entries (from observe/categories/*.md) category_contexts = [k for k in registry if k.startswith("observe.describe.")]
# Should have frame + all categories (browsing, code, gaming, etc.) assert len(category_contexts) > 5, "Should discover category contexts"
# Each category context should have required fields for context in category_contexts: assert "label" in registry[context] assert "group" in registry[context]
def test_context_registry_includes_talent_configs(): """Test that registry includes discovered talent contexts (agents + generators).""" registry = get_context_registry()
# Should have talent entries (from talent/*.md and apps/*/talent/*.md) talent_contexts = [k for k in registry if k.startswith("talent.")]
# Should have multiple talent contexts (agents + generators) assert len(talent_contexts) > 1, "Should discover talent contexts"
# Should have system talent configs system_talent = [k for k in talent_contexts if k.startswith("talent.system.")] assert len(system_talent) > 0, "Should discover system talent configs"
# Should have app talent configs app_talent = [ k for k in talent_contexts if k.startswith("talent.") and not k.startswith("talent.system.") ] assert len(app_talent) > 0, "Should discover app talent configs"
# Should include type field for talent contexts for context in talent_contexts: assert "type" in registry[context], f"{context} missing type field"
def test_context_registry_structure(): """Test that all registry entries have required fields.""" registry = get_context_registry() required_keys = {"label", "group"}
for context, config in registry.items(): assert isinstance(config, dict), f"{context} should be a dict" assert required_keys <= set(config.keys()), ( f"{context} missing keys: {required_keys - set(config.keys())}" )
def test_context_registry_is_cached(): """Test that registry is built once and cached.""" registry1 = get_context_registry() registry2 = get_context_registry()
# Should return the same object (cached) assert registry1 is registry2
# ---------------------------------------------------------------------------# Model pricing support tests# ---------------------------------------------------------------------------
def test_all_default_models_have_pricing(): """Verify all default provider models have genai-prices support.
This test ensures that when default models are updated, we catch any missing pricing data early. If this test fails:
1. Run: make update-prices 2. Re-run this test 3. If still failing, the model may be too new for genai-prices
See think/models.py model constants section for more details. """ all_models = set(DEFAULT_MODEL_BY_PROVIDER.values()) all_models.add(CLAUDE_OPUS_4)
missing_pricing = [] for model in sorted(all_models): token_data = { "model": model, "usage": { "input_tokens": 1000, "output_tokens": 100, "total_tokens": 1100, }, } result = calc_token_cost(token_data) if result is None: missing_pricing.append(model)
if missing_pricing: pytest.fail( f"Models missing genai-prices support: {missing_pricing}\n" "Run 'make update-prices' and re-test. " "If still failing, model may be too new for genai-prices." )
@pytest.mark.parametrize( ("model", "expected"), [ ("gpt-5.5", _Family(("openai", None), (5, 5))), ("gpt-5", _Family(("openai", None), (5, 0))), ("gpt-5.5-mini", _Family(("openai", "mini"), (5, 5))), ("gpt-5.4-nano", _Family(("openai", "nano"), (5, 4))), ("gpt-5.2-pro", _Family(("openai", "pro"), (5, 2))), ],)def test_parse_family_openai(model, expected): assert _parse_family_openai(model) == expected
@pytest.mark.parametrize( "model", [ "ft:gpt-5", "gpt-5-image", "gpt-5-image-mini", "gpt-5.1-codex-mini", "gpt-4o", "o3", "text-embedding-3-small", ],)def test_parse_family_openai_rejects_unsupported_models(model): assert _parse_family_openai(model) is None
@pytest.mark.parametrize( ("model", "expected"), [ ("claude-opus-4", _Family(("anthropic", "opus"), (4, 0))), ("claude-opus-4-7", _Family(("anthropic", "opus"), (4, 7))), ("claude-sonnet-4-6", _Family(("anthropic", "sonnet"), (4, 6))), ("claude-haiku-5", _Family(("anthropic", "haiku"), (5, 0))), ],)def test_parse_family_anthropic(model, expected): assert _parse_family_anthropic(model) == expected
@pytest.mark.parametrize( "model", [ "claude-3-opus-latest", "claude-3-5-haiku-latest", "claude-v2", "claude-2", "claude-sonnet-4-6-latest", ],)def test_parse_family_anthropic_rejects_unsupported_models(model): assert _parse_family_anthropic(model) is None
@pytest.mark.parametrize( ("model", "expected"), [ ("gemini-3.5-flash", _Family(("gemini", "flash"), (3, 5))), ("gemini-3-flash", _Family(("gemini", "flash"), (3, 0))), ("gemini-3.1-flash-lite", _Family(("gemini", "flash-lite"), (3, 1))), ("gemini-2.5-pro-preview", _Family(("gemini", "pro"), (2, 5))), ("gemini-flash-latest", _Family(("gemini", "flash"), (0, 0))), ("gemini-pro-latest", _Family(("gemini", "pro"), (0, 0))), ("gemini-flash-lite-latest", _Family(("gemini", "flash-lite"), (0, 0))), ],)def test_parse_family_gemini(model, expected): assert _parse_family_gemini(model) == expected
@pytest.mark.parametrize( "model", [ "gemini-3-pro-image-preview", "gemini-2.5-flash-image", "gemini-pro", "gemini-embedding-001", "gemini-live-2.5-flash-preview", "gemini-flash-1.5", "gemma-3", ],)def test_parse_family_gemini_rejects_unsupported_models(model): assert _parse_family_gemini(model) is None
@pytest.mark.parametrize( ("model", "provider_id", "expected"), [ ("gemini-3.5-flash", "google", "gemini-3-flash-preview"), ("gemini-3.1-flash-lite", "google", "gemini-2.5-flash-lite"), ("gemini-flash-latest", "google", "gemini-3-flash-preview"), ("gemini-pro-latest", "google", "gemini-3.1-pro-preview"), ("gemini-flash-lite-latest", "google", "gemini-2.5-flash-lite"), ("claude-opus-4-7", "anthropic", "claude-opus-4-6"), ("claude-sonnet-4-6", "anthropic", "claude-sonnet-4-6"), ("claude-haiku-5", "anthropic", "claude-haiku-4-5"), ("gpt-5.5", "openai", "gpt-5.2"), ("gpt-5.5-mini", "openai", "gpt-5-mini"), ("totally-fake-model", "openai", None), ("text-embedding-3-small", "openai", None), ],)def test_find_pricing_fallback(model, provider_id, expected): _find_pricing_fallback.cache_clear()
assert _find_pricing_fallback(model, provider_id) == expected
@pytest.mark.parametrize( "model", [ "gemini-3.5-flash", "gemini-3.1-flash-lite", "claude-opus-4-7", "gpt-5.5", ],)def test_calc_token_cost_fallback(model): result = calc_token_cost( {"model": model, "usage": {"input_tokens": 1000, "output_tokens": 100}} )
assert result is not None assert result["total_cost"] > 0
def test_calc_token_cost_fallback_returns_none_for_unknown_model(): assert ( calc_token_cost( { "model": "totally-fake-model", "usage": {"input_tokens": 1000, "output_tokens": 100}, } ) is None )
def test_calc_token_cost_fallback_keeps_local_free(): assert ( calc_token_cost( { "model": LOCAL_MODEL, "usage": {"input_tokens": 1000, "output_tokens": 100}, } )["total_cost"] == 0.0 )
def test_fallback_logging(caplog): models_module._LOGGED_FALLBACKS.clear()
with caplog.at_level(logging.INFO, logger="solstone.think.models"): calc_token_cost( { "model": "gemini-3.5-flash", "usage": {"input_tokens": 1000, "output_tokens": 100}, } ) calc_token_cost( { "model": "gemini-3.5-flash", "usage": {"input_tokens": 1000, "output_tokens": 100}, } )
fallback_logs = [ record.getMessage() for record in caplog.records if record.getMessage().startswith("pricing: family-fallback") ] assert fallback_logs == [ "pricing: family-fallback gemini-3.5-flash -> gemini-3-flash-preview" ]
with caplog.at_level(logging.INFO, logger="solstone.think.models"): calc_token_cost( { "model": "claude-opus-4-7", "usage": {"input_tokens": 1000, "output_tokens": 100}, } )
fallback_logs = [ record.getMessage() for record in caplog.records if record.getMessage().startswith("pricing: family-fallback") ] assert fallback_logs == [ "pricing: family-fallback gemini-3.5-flash -> gemini-3-flash-preview", "pricing: family-fallback claude-opus-4-7 -> claude-opus-4-6", ]
# ---------------------------------------------------------------------------# get_usage_cost tests# ---------------------------------------------------------------------------
def test_get_usage_cost_nonexistent_day(use_fixtures_journal): """Test that nonexistent day returns zeros.""" result = get_usage_cost("19000101") assert result == {"requests": 0, "tokens": 0, "cost": 0.0}
def test_get_usage_cost_day_total(use_fixtures_journal): """Test aggregating all entries for a day.""" # 20250823 has test entries with gemini models result = get_usage_cost("20250823") assert result["requests"] > 0 assert isinstance(result["tokens"], int) assert isinstance(result["cost"], float)
def test_iter_token_log_preserves_type_field(use_fixtures_journal): """Token log iterator should preserve top-level type field.""" entries = list(iter_token_log("20250823")) generate_entries = [entry for entry in entries if entry.get("type") == "generate"]
assert generate_entries assert any( entry.get("context") == "think.detect_created.classify_new_file" for entry in generate_entries )
def test_get_usage_cost_context_filter(use_fixtures_journal): """Test filtering by context prefix.""" # Filter to test contexts result = get_usage_cost("20250823", context="tests.test_gemini") assert result["requests"] > 0
# Filter to non-matching context should return zeros result_empty = get_usage_cost("20250823", context="nonexistent.context") assert result_empty["requests"] == 0
def test_get_usage_cost_segment_filter(use_fixtures_journal): """Test filtering by segment key.""" # Fixture data includes one entry tagged with segment 143022_300 result = get_usage_cost("20250823", segment="143022_300") assert result["requests"] == 1 assert result["tokens"] == 7000 assert result["cost"] > 0.0
def test_get_usage_cost_combined_filters(use_fixtures_journal): """Test combined segment and context filters.""" # With both filters, entries must match both result = get_usage_cost( "20250823", segment="nonexistent", context="tests.test_gemini", ) # Segment doesn't exist, so no matches assert result["requests"] == 0
# ---------------------------------------------------------------------------# log_token_usage normalization tests# ---------------------------------------------------------------------------
def test_log_token_usage_computes_total_tokens(tmp_path, monkeypatch): """total_tokens is computed from input+output when missing (Codex CLI format).""" import json
from solstone.think.models import log_token_usage
monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
# Codex CLI format: no total_tokens log_token_usage( model="gpt-5.2", usage={"input_tokens": 1000, "output_tokens": 200}, context="test", )
log_file = tmp_path / "tokens" / (__import__("time").strftime("%Y%m%d") + ".jsonl") entry = json.loads(log_file.read_text().strip()) assert entry["usage"]["total_tokens"] == 1200 assert entry["usage"]["input_tokens"] == 1000 assert entry["usage"]["output_tokens"] == 200
def test_log_token_usage_preserves_existing_total_tokens(tmp_path, monkeypatch): """total_tokens is preserved when already present and non-zero.""" import json
from solstone.think.models import log_token_usage
monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
log_token_usage( model="gpt-5.2", usage={"input_tokens": 1000, "output_tokens": 200, "total_tokens": 1500}, context="test", )
log_file = tmp_path / "tokens" / (__import__("time").strftime("%Y%m%d") + ".jsonl") entry = json.loads(log_file.read_text().strip()) assert entry["usage"]["total_tokens"] == 1500
def test_log_token_usage_maps_cached_input_tokens(tmp_path, monkeypatch): """cached_input_tokens (Codex CLI format) maps to cached_tokens.""" import json
from solstone.think.models import log_token_usage
monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
log_token_usage( model="gpt-5.2", usage={ "input_tokens": 1000, "cached_input_tokens": 800, "output_tokens": 200, }, context="test", )
log_file = tmp_path / "tokens" / (__import__("time").strftime("%Y%m%d") + ".jsonl") entry = json.loads(log_file.read_text().strip()) assert entry["usage"]["cached_tokens"] == 800 assert entry["usage"]["total_tokens"] == 1200
def test_log_token_usage_passes_through_reasoning_tokens(tmp_path, monkeypatch): """reasoning_tokens from provider-normalized usage are preserved in log.""" import json
from solstone.think.models import log_token_usage
monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
# Normalized usage from Google provider (the bug: reasoning_tokens were dropped) log_token_usage( model="gemini-3-flash-preview", usage={ "input_tokens": 13319, "output_tokens": 969, "total_tokens": 37878, "reasoning_tokens": 23590, }, context="test", )
log_file = tmp_path / "tokens" / (__import__("time").strftime("%Y%m%d") + ".jsonl") entry = json.loads(log_file.read_text().strip()) assert entry["usage"]["reasoning_tokens"] == 23590 assert entry["usage"]["total_tokens"] == 37878 assert entry["usage"]["input_tokens"] == 13319 assert entry["usage"]["output_tokens"] == 969
def test_log_token_usage_passes_through_cache_creation_tokens(tmp_path, monkeypatch): """cache_creation_tokens from Anthropic provider are preserved in log.""" import json
from solstone.think.models import log_token_usage
monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path))
log_token_usage( model="claude-sonnet-4-5", usage={ "input_tokens": 5000, "output_tokens": 1000, "total_tokens": 6000, "cached_tokens": 3000, "cache_creation_tokens": 2000, }, context="test", )
log_file = tmp_path / "tokens" / (__import__("time").strftime("%Y%m%d") + ".jsonl") entry = json.loads(log_file.read_text().strip()) assert entry["usage"]["cache_creation_tokens"] == 2000 assert entry["usage"]["cached_tokens"] == 3000
def test_log_token_usage_logs_append_failure(tmp_path, monkeypatch, caplog): import builtins
from solstone.think.models import log_token_usage
monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) real_open = builtins.open
def fail_token_open(path, *args, **kwargs): if str(path).endswith(".jsonl"): raise OSError("disk full") return real_open(path, *args, **kwargs)
monkeypatch.setattr(builtins, "open", fail_token_open)
with caplog.at_level(logging.WARNING, logger="solstone.think.models"): log_token_usage( model="gpt-5.2", usage={"input_tokens": 1000, "output_tokens": 200}, context="test", )
assert "failed to log token usage" in caplog.text
class TestModelSupports: def test_opus_4_7_temperature_not_supported(self): assert model_supports(CLAUDE_OPUS_4, "temperature") is False
def test_sonnet_4_6_temperature_supported(self): assert model_supports(CLAUDE_SONNET_4, "temperature") is True
def test_unlisted_param_defaults_supported(self): assert model_supports(CLAUDE_OPUS_4, "max_tokens") is True
def test_unlisted_model_defaults_supported(self): assert model_supports("gpt-5.5", "temperature") is True
class TestValidateSchema: def test_valid_instance(self): schema = { "type": "object", "properties": {"field": {"type": "string"}}, "required": ["field"], }
result = _validate_schema('{"field": "ok"}', schema)
assert result == {"valid": True, "errors": []}
def test_schema_violation_type(self): schema = { "type": "object", "properties": {"field": {"type": "integer"}}, "required": ["field"], }
result = _validate_schema('{"field": "ok"}', schema)
assert result["valid"] is False assert len(result["errors"]) == 1 assert result["errors"][0]["path"] == "/field" assert result["errors"][0]["constraint"] == "type" assert result["errors"][0]["message"]
def test_schema_violation_required(self): schema = {"type": "object", "required": ["field"]}
result = _validate_schema("{}", schema)
assert result["valid"] is False assert len(result["errors"]) == 1 assert result["errors"][0]["path"] == "" assert result["errors"][0]["constraint"] == "required"
def test_multiple_violations(self): schema = { "type": "object", "properties": { "field": {"type": "integer"}, "other": {"type": "string"}, }, "required": ["field", "other"], }
result = _validate_schema('{"field": "bad"}', schema)
assert result["valid"] is False assert len(result["errors"]) == 2 assert {error["constraint"] for error in result["errors"]} == { "required", "type", }
def test_parse_failure(self): schema = {"type": "object"}
result = _validate_schema("{", schema)
assert result["valid"] is False assert result["errors"] == [ { "path": "", "constraint": "json_parse", "message": result["errors"][0]["message"], } ]
def test_json_pointer_escape(self): schema = { "type": "object", "properties": { "a/b": { "type": "object", "properties": {"c~d": {"type": "integer"}}, } }, }
result = _validate_schema('{"a/b": {"c~d": "bad"}}', schema)
assert result["valid"] is False assert result["errors"][0]["path"] == "/a~1b/c~0d"
def test_warning_logged_for_violations(self, caplog): schema = { "type": "object", "properties": {"field": {"type": "integer"}}, }
with caplog.at_level(logging.WARNING): _validate_schema('{"field": "bad"}', schema)
assert any( record.levelno == logging.WARNING and "schema_validation:" in record.getMessage() for record in caplog.records )
def test_invalid_schema_does_not_raise(self): result = _validate_schema('{"field": "ok"}', {"type": "not-a-real-type"})
assert result["valid"] is False assert len(result["errors"]) == 1 assert result["errors"][0]["constraint"] == "schema_validation"
def _assert_exception_metadata_omits_marker(exc: BaseException, marker: str) -> None: assert marker not in str(exc) for value in vars(exc).values(): assert marker not in str(value) for value in exc.args: assert marker not in str(value)
class TestGenerateJsonSchemaPlumbing: def test_finish_reason_predicate_json_rejects_any_non_stop(self): error = finish_reason_error( {"text": "{}", "finish_reason": "content_filter"}, json_output=True, )
assert isinstance(error, IncompleteJSONError)
def test_finish_reason_predicate_plain_text_rejects_length(self): error = finish_reason_error( {"text": "partial", "finish_reason": "max_tokens"}, json_output=False, )
assert isinstance(error, IncompleteTextError) assert error.reason_code == "incomplete_text_length"
def test_finish_reason_predicate_plain_text_rejects_non_length(self): error = finish_reason_error( { "text": "", "model": "provider-model", "usage": { "input_tokens": 1, "output_tokens": 2, "raw": object(), }, "finish_reason": "content_filter", }, json_output=False, )
assert isinstance(error, ProviderResponseInvalidError) assert error.reason_code == "provider_response_invalid" assert error.finish_reason == "content_filter" assert error.model == "provider-model" assert error.token_counts == {"input_tokens": 1, "output_tokens": 2}
def test_strict_validation_plain_text_finish_leniency_is_unchanged(self): models_module.validate_generate_result_strict( {"text": "partial", "finish_reason": "max_tokens"}, json_output=False, )
@pytest.mark.parametrize("json_output", [True, False]) @pytest.mark.parametrize( ("text_case", "expect_blank"), [ ("nonblank", False), ("empty", True), ("spaces", True), ("missing", True), ("non_string", True), ], ) @pytest.mark.parametrize("finish_reason", ["stop", "STOP", " stop ", " STOP "]) def test_strict_validation_stop_finish_variants_blank_output_matrix( self, finish_reason, text_case, expect_blank, json_output, ): result = {"finish_reason": finish_reason} if text_case == "nonblank": result["text"] = "complete text" elif text_case == "empty": result["text"] = "" elif text_case == "spaces": result["text"] = " " elif text_case == "non_string": result["text"] = 123
if not expect_blank: models_module.validate_generate_result_strict( result, json_output=json_output, ) return
with pytest.raises(ProviderResponseInvalidError) as exc_info: models_module.validate_generate_result_strict( result, json_output=json_output, )
exc = exc_info.value assert exc.reason == "blank_visible_output" assert exc.finish_reason == "stop"
@pytest.mark.parametrize( ("finish_reason", "expected_reason"), [ ("length", "length"), ("LENGTH", "length"), (" length ", "length"), ("max_tokens", "max_tokens"), ("MAX_TOKENS", "max_tokens"), (" max_tokens ", "max_tokens"), ], ) def test_strict_validation_json_length_finish_variants_use_safe_reason( self, finish_reason, expected_reason, ): with pytest.raises(IncompleteJSONError) as exc_info: models_module.validate_generate_result_strict( {"text": "partial", "finish_reason": finish_reason}, json_output=True, )
exc = exc_info.value assert exc.reason == expected_reason assert exc.reason_code == "incomplete_json_length"
def test_strict_validation_json_empty_finish_reason_is_unknown_non_stop(self): # Empty finish strings are reduced to the fixed safe token before JSON # or text classification; this intentionally replaces the old falsy # short-circuit. with pytest.raises(IncompleteJSONError) as exc_info: models_module.validate_generate_result_strict( {"text": "partial", "finish_reason": ""}, json_output=True, )
assert exc_info.value.reason == "unknown"
@pytest.mark.parametrize( "finish_reason", [ str({"secret": _FINISH_METADATA_MARKER}).strip().lower(), "a" * 65 + _FINISH_METADATA_MARKER, ], ) def test_strict_validation_json_unsafe_finish_reason_uses_unknown_without_raw_metadata( self, finish_reason, ): marker = _FINISH_METADATA_MARKER
with pytest.raises(IncompleteJSONError) as exc_info: models_module.validate_generate_result_strict( {"text": "partial", "finish_reason": finish_reason}, json_output=True, )
exc = exc_info.value assert exc.reason == "unknown" _assert_exception_metadata_omits_marker(exc, marker)
@pytest.mark.parametrize( "entrypoint", ["generate", "generate_with_result", "agenerate_with_result", "agenerate"], ) def test_all_generate_entrypoints_reject_bare_non_responsive_output( self, entrypoint, ): result = _generate_result(_NON_RESPONSIVE_REFUSAL, usage=_GENERATE_USAGE) provider_module = _provider_module_for_result(result)
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "requested-model"), ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), patch("solstone.think.models.log_token_usage") as mock_log_usage, ): with pytest.raises(NonResponsiveOutputError) as exc_info: _call_generate_entrypoint(entrypoint)
exc = exc_info.value assert exc.reason_code == NON_RESPONSIVE_REASON_CODE assert exc.non_responsive_output == _NON_RESPONSIVE_REFUSAL assert exc.non_responsive_matched_signal == "i cannot" mock_log_usage.assert_called_once_with( model="provider-model", usage=_GENERATE_USAGE, context="test.context", type="generate", non_responsive_output=_NON_RESPONSIVE_REFUSAL, non_responsive_matched_signal="i cannot", )
@pytest.mark.parametrize( "entrypoint", ["generate", "generate_with_result", "agenerate_with_result", "agenerate"], ) def test_all_generate_entrypoints_reject_chat_schema_message_refusal( self, entrypoint, ): provider_module = _provider_module_for_result( _generate_result(json.dumps(_REFUSAL_CHAT_FIXTURE), usage=_GENERATE_USAGE) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "requested-model"), ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), patch("solstone.think.models.log_token_usage"), ): with pytest.raises(NonResponsiveOutputError) as exc_info: _call_generate_entrypoint(entrypoint)
assert exc_info.value.non_responsive_output == json.dumps(_REFUSAL_CHAT_FIXTURE)
@pytest.mark.parametrize( "entrypoint", ["generate", "generate_with_result", "agenerate_with_result", "agenerate"], ) def test_all_generate_entrypoints_accept_responsive_chat_schema_fixture( self, entrypoint, ): provider_module = _provider_module_for_result( _generate_result( json.dumps(_RESPONSIVE_CHAT_FIXTURE), usage=_GENERATE_USAGE ) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "requested-model"), ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), patch("solstone.think.models.log_token_usage") as mock_log_usage, ): result = _call_generate_entrypoint(entrypoint)
text = result["text"] if isinstance(result, dict) else result assert json.loads(text) == _RESPONSIVE_CHAT_FIXTURE mock_log_usage.assert_called_once_with( model="provider-model", usage=_GENERATE_USAGE, context="test.context", type="generate", )
def test_non_responsive_usage_log_carries_capped_raw_output( self, tmp_path, monkeypatch, ): monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) raw_output = _NON_RESPONSIVE_REFUSAL + " " + ("overflow " * 200) provider_module = _provider_module_for_result( _generate_result(raw_output, usage=_GENERATE_USAGE) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "requested-model"), ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), pytest.raises(NonResponsiveOutputError), ): generate_with_result("hello", "test.context")
[entry] = list(iter_token_log(time.strftime("%Y%m%d"))) assert ( entry["non_responsive_output"] == raw_output[:NON_RESPONSIVE_RAW_OUTPUT_CAP_CHARS] ) assert ( len(entry["non_responsive_output"]) == NON_RESPONSIVE_RAW_OUTPUT_CAP_CHARS ) assert raw_output not in json.dumps(entry)
def test_non_responsive_token_log_append_preserves_existing_bytes( self, tmp_path, monkeypatch, ): monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) token_dir = tmp_path / "tokens" token_dir.mkdir() log_file = token_dir / f"{time.strftime('%Y%m%d')}.jsonl" prior_lines = [ b'{"old":1,"payload":"kept exactly"}\n', b'{"old":2,"payload":"also kept"}\n', ] log_file.write_bytes(b"".join(prior_lines)) provider_module = _provider_module_for_result( _generate_result(_NON_RESPONSIVE_REFUSAL, usage=_GENERATE_USAGE) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "requested-model"), ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), pytest.raises(NonResponsiveOutputError), ): generate_with_result("hello", "test.context")
lines = log_file.read_bytes().splitlines(keepends=True) assert lines[: len(prior_lines)] == prior_lines assert len(lines) == len(prior_lines) + 1 appended = json.loads(lines[-1]) assert appended["non_responsive_output"] == _NON_RESPONSIVE_REFUSAL
def test_non_responsive_without_usage_logs_empty_usage_and_raw_output( self, tmp_path, monkeypatch, ): monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) provider_module = _provider_module_for_result( _generate_result(_NON_RESPONSIVE_REFUSAL) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "requested-model"), ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), pytest.raises(NonResponsiveOutputError), ): generate_with_result("hello", "test.context")
[entry] = list(iter_token_log(time.strftime("%Y%m%d"))) assert entry["usage"] == {} assert entry["non_responsive_output"] == _NON_RESPONSIVE_REFUSAL
def test_non_responsive_token_log_survives_lone_surrogate_roundtrip( self, tmp_path, monkeypatch, ): monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) raw_output = f"{_NON_RESPONSIVE_REFUSAL}\ud800"
models_module.log_token_usage( model="provider-model", usage={}, context="test.context", type="generate", non_responsive_output=raw_output, )
entries = list(iter_token_log(time.strftime("%Y%m%d"))) assert len(entries) == 1 assert entries[0]["non_responsive_output"] == raw_output
def test_truncation_wins_over_non_responsive_output(self): provider_module = _provider_module_for_result( _generate_result( _NON_RESPONSIVE_REFUSAL, usage=_GENERATE_USAGE, finish_reason="length", ) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "requested-model"), ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), patch("solstone.think.models.log_token_usage"), ): with pytest.raises(IncompleteJSONError): generate_with_result("hello", "test.context", json_output=True)
def test_responsiveness_classifier_exception_surfaces_outside_token_logging( self, caplog, ): class ClassifierFailure(RuntimeError): pass
provider_module = _provider_module_for_result( _generate_result("ordinary visible text", usage=_GENERATE_USAGE) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "requested-model"), ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), patch( "solstone.think.models.classify_output_responsiveness", side_effect=ClassifierFailure("classifier down"), ), caplog.at_level(logging.WARNING, logger="solstone.think.models"), ): with pytest.raises(ClassifierFailure): generate_with_result("hello", "test.context")
assert "failed to log token usage" not in caplog.text
@pytest.mark.parametrize( "entrypoint", ["generate", "generate_with_result", "agenerate_with_result", "agenerate"], ) def test_generate_entrypoints_keep_existing_log_call_for_responsive_outputs( self, entrypoint, ): provider_module = _provider_module_for_result( _generate_result("The release check passed.", usage=_GENERATE_USAGE) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "requested-model"), ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), patch("solstone.think.models.log_token_usage") as mock_log_usage, ): _call_generate_entrypoint(entrypoint)
mock_log_usage.assert_called_once_with( model="provider-model", usage=_GENERATE_USAGE, context="test.context", type="generate", )
def test_non_responsive_usage_log_omits_additive_keys_when_responsive(self): result = _generate_result("The release check passed.", usage=_GENERATE_USAGE) result["notes"] = _NON_RESPONSIVE_REFUSAL provider_module = _provider_module_for_result(result)
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "requested-model"), ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), patch("solstone.think.models.log_token_usage") as mock_log_usage, ): assert generate_with_result("hello", "test.context") is result
kwargs = mock_log_usage.call_args.kwargs assert "non_responsive_output" not in kwargs assert "non_responsive_matched_signal" not in kwargs
def test_generate_with_result_can_disable_responsiveness_gate_for_brain_probe(self): provider_module = _provider_module_for_result( _generate_result(_NON_RESPONSIVE_REFUSAL, usage=_GENERATE_USAGE) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "requested-model"), ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), patch("solstone.think.models.log_token_usage") as mock_log_usage, ): result = generate_with_result( "hello", "test.context", enforce_responsiveness=False, )
assert result["text"] == _NON_RESPONSIVE_REFUSAL mock_log_usage.assert_called_once_with( model="provider-model", usage=_GENERATE_USAGE, context="test.context", type="generate", )
@pytest.mark.parametrize("text", ["", " ", None, 123]) @pytest.mark.parametrize( "entrypoint", ["generate", "generate_with_result", "agenerate_with_result", "agenerate"], ) def test_all_generate_entrypoints_reject_blank_clean_visible_output( self, entrypoint, text, ): result = { "text": text, "model": "provider-model", "finish_reason": "stop", "usage": { "input_tokens": 11, "output_tokens": 0, "reasoning_tokens": 3, "total_tokens": 14, "raw_usage": object(), }, } provider_module = SimpleNamespace( run_generate=MagicMock(return_value=result), run_agenerate=AsyncMock(return_value=result), )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "requested-model"), ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), patch("solstone.think.models.log_token_usage") as mock_log_usage, patch("solstone.think.models._validate_schema") as mock_validate_schema, ): with pytest.raises(ProviderResponseInvalidError) as exc_info: if entrypoint == "generate": generate("hello", "test.context", json_schema={"type": "object"}) elif entrypoint == "generate_with_result": generate_with_result( "hello", "test.context", json_schema={"type": "object"} ) elif entrypoint == "agenerate_with_result": asyncio.run( agenerate_with_result( "hello", "test.context", json_schema={"type": "object"}, ) ) else: asyncio.run( agenerate( "hello", "test.context", json_schema={"type": "object"} ) )
exc = exc_info.value assert exc.reason == "blank_visible_output" assert exc.finish_reason == "stop" assert exc.model == "provider-model" assert exc.token_counts == { "input_tokens": 11, "output_tokens": 0, "reasoning_tokens": 3, "total_tokens": 14, } assert not hasattr(exc, "usage") assert not hasattr(exc, "text") assert not hasattr(exc, "prompt") assert not hasattr(exc, "response") assert not hasattr(exc, "body") mock_log_usage.assert_called_once_with( model="provider-model", usage=result["usage"], context="test.context", type="generate", ) mock_validate_schema.assert_not_called()
def test_strict_validation_omits_non_finite_token_counts(self): result = { "text": " ", "model": "provider-model", "finish_reason": "stop", "usage": { "input_tokens": float("nan"), "output_tokens": float("inf"), "reasoning_tokens": 3, "total_tokens": 14, }, }
with pytest.raises(ProviderResponseInvalidError) as exc_info: models_module.validate_generate_result_strict(result, json_output=False)
exc = exc_info.value assert exc.reason == "blank_visible_output" assert exc.model == "provider-model" assert exc.token_counts == { "reasoning_tokens": 3, "total_tokens": 14, }
def test_missing_text_and_missing_finish_are_typed_provider_invalid(self): provider_module = SimpleNamespace( run_generate=MagicMock(return_value={"finish_reason": None}) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), ): with pytest.raises(ProviderResponseInvalidError) as exc_info: generate("hello", "test.context")
assert exc_info.value.reason == "blank_visible_output"
def test_non_string_finish_metadata_is_typed_provider_invalid(self): provider_module = SimpleNamespace( run_generate=MagicMock( return_value={"text": "ok", "finish_reason": {"bad": "shape"}} ) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), ): with pytest.raises(ProviderResponseInvalidError) as exc_info: generate("hello", "test.context")
assert exc_info.value.reason == "malformed_finish_reason"
def test_strict_validation_bounds_non_string_finish_metadata(self): marker = _FINISH_METADATA_MARKER
with pytest.raises(ProviderResponseInvalidError) as exc_info: models_module.validate_generate_result_strict( { "text": "ok", "finish_reason": {"secret": {"nested": marker}}, }, json_output=False, )
exc = exc_info.value assert exc.reason == "malformed_finish_reason" assert exc.finish_reason is None _assert_exception_metadata_omits_marker(exc, marker)
@pytest.mark.parametrize( "finish_reason", [ str({"secret": _FINISH_METADATA_MARKER}).strip().lower(), "a" * 65 + _FINISH_METADATA_MARKER, ], ) def test_finish_reason_error_bounds_unsafe_finish_tokens(self, finish_reason): marker = _FINISH_METADATA_MARKER
error = finish_reason_error( {"text": "partial", "finish_reason": finish_reason}, json_output=False, )
assert isinstance(error, ProviderResponseInvalidError) assert error.reason == "unknown" assert error.finish_reason == "unknown" _assert_exception_metadata_omits_marker(error, marker)
@pytest.mark.parametrize( "finish_reason", ["content_filter", "CONTENT_FILTER", " content_filter "], ) def test_finish_reason_error_preserves_safe_finish_tokens(self, finish_reason): result = { "text": "partial", "model": "provider-model", "finish_reason": finish_reason, "usage": { "input_tokens": 11, "output_tokens": 2, "reasoning_tokens": 3, "total_tokens": 16, }, }
error = finish_reason_error(result, json_output=False)
assert isinstance(error, ProviderResponseInvalidError) assert error.reason == "content_filter" assert error.finish_reason == "content_filter" assert error.model == "provider-model" assert error.token_counts == { "input_tokens": 11, "output_tokens": 2, "reasoning_tokens": 3, "total_tokens": 16, }
def test_length_finish_classification_keeps_incomplete_text(self): error = finish_reason_error( {"text": "partial", "finish_reason": "max_tokens"}, json_output=False, )
assert isinstance(error, IncompleteTextError) assert error.reason == "max_tokens" assert error.reason_code == "incomplete_text_length" assert str(error) == "Text response incomplete (reason: max_tokens)"
def test_json_length_finish_stays_incomplete_json_not_provider_invalid(self): provider_module = SimpleNamespace( run_generate=MagicMock( return_value={"text": "", "finish_reason": "max_tokens"} ) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), ): with pytest.raises(IncompleteJSONError): generate("hello", "test.context", json_output=True)
def test_generate_forces_json_output_with_schema(self): schema = {"type": "object"} provider_module = SimpleNamespace( run_generate=MagicMock(return_value={"text": "{}", "finish_reason": "stop"}) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), ): result = generate("hello", "test.context", json_schema=schema)
assert result == "{}" call_kwargs = provider_module.run_generate.call_args.kwargs assert call_kwargs["json_output"] is True assert call_kwargs["json_schema"] == schema
def test_agenerate_forces_json_output_with_schema(self): schema = {"type": "object"} provider_module = SimpleNamespace( run_agenerate=AsyncMock( return_value={"text": "{}", "finish_reason": "stop"} ) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), ): result = asyncio.run(agenerate("hello", "test.context", json_schema=schema))
assert result == "{}" call_kwargs = provider_module.run_agenerate.call_args.kwargs assert call_kwargs["json_output"] is True assert call_kwargs["json_schema"] == schema
def test_generate_with_result_truncates_annotated_overrun(self): schema = { "type": "object", "properties": { "field": { "type": "string", "maxLength": 3, SCHEMA_TRUNCATE_KEY: True, } }, "required": ["field"], "additionalProperties": False, } response_text = json.dumps({"field": "abcdef"}) expected_text = json.dumps({"field": "abc"}, ensure_ascii=False) provider_module = SimpleNamespace( run_generate=MagicMock(side_effect=_fresh_generate_result(response_text)) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), ): result = generate_with_result( "hello", "test.context", json_schema=schema, )
assert result["text"] == expected_text assert result["schema_validation"] == { "valid": True, "errors": [], "truncated": ["/field"], }
def test_generate_returns_truncated_annotated_overrun(self): schema = { "type": "object", "properties": { "field": { "type": "string", "maxLength": 3, SCHEMA_TRUNCATE_KEY: True, } }, "required": ["field"], "additionalProperties": False, } response_text = json.dumps({"field": "abcdef"}) provider_module = SimpleNamespace( run_generate=MagicMock(side_effect=_fresh_generate_result(response_text)) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), ): result = generate("hello", "test.context", json_schema=schema)
assert result == json.dumps({"field": "abc"}, ensure_ascii=False)
def test_non_annotated_overrun_still_fails_advisory_and_raising_paths(self): schema = { "type": "object", "properties": { "content": { "type": ["string", "null"], "maxLength": 600, } }, "required": ["content"], "additionalProperties": False, } response_text = json.dumps({"content": "c" * 700}) advisory_provider = SimpleNamespace( run_generate=MagicMock(side_effect=_fresh_generate_result(response_text)) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=advisory_provider, ), ): result = generate_with_result( "hello", "test.context", json_schema=schema, )
assert result["text"] == response_text assert "truncated" not in result["schema_validation"] assert result["schema_validation"]["valid"] is False assert result["schema_validation"]["errors"][0]["path"] == "/content" assert result["schema_validation"]["errors"][0]["constraint"] == "maxLength"
raising_provider = SimpleNamespace( run_generate=MagicMock(side_effect=_fresh_generate_result(response_text)) ) with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=raising_provider, ), ): with pytest.raises(SchemaValidationError) as exc_info: generate("hello", "test.context", json_schema=schema)
assert exc_info.value.text == response_text assert exc_info.value.errors[0]["path"] == "/content" assert exc_info.value.errors[0]["constraint"] == "maxLength"
def test_multiple_array_item_truncations_record_instance_pointers(self): schema = { "type": "object", "properties": { "entities": { "type": "array", "items": { "type": "object", "properties": { "operations": { "type": "array", "items": { "type": "object", "properties": { "reasoning": { "type": ["string", "null"], "maxLength": 300, SCHEMA_TRUNCATE_KEY: True, } }, "required": ["reasoning"], "additionalProperties": False, }, } }, "required": ["operations"], "additionalProperties": False, }, } }, "required": ["entities"], "additionalProperties": False, } response = { "entities": [ { "operations": [ {"reasoning": "r" * 350}, {"reasoning": "s" * 350}, {"reasoning": "t" * 350}, ] } ] } expected = { "entities": [ { "operations": [ {"reasoning": "r" * 300}, {"reasoning": "s" * 300}, {"reasoning": "t" * 300}, ] } ] } provider_module = SimpleNamespace( run_generate=MagicMock( side_effect=_fresh_generate_result(json.dumps(response)) ) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), ): result = generate_with_result( "hello", "test.context", json_schema=schema, )
assert result["text"] == json.dumps(expected, ensure_ascii=False) assert result["schema_validation"] == { "valid": True, "errors": [], "truncated": [ "/entities/0/operations/0/reasoning", "/entities/0/operations/1/reasoning", "/entities/0/operations/2/reasoning", ], }
def test_non_string_at_annotated_path_is_not_coerced(self): schema = { "type": "object", "properties": { "reasoning": { "type": ["string", "null"], "maxLength": 300, SCHEMA_TRUNCATE_KEY: True, } }, "required": ["reasoning"], "additionalProperties": False, } response_text = json.dumps({"reasoning": 42}) provider_module = SimpleNamespace( run_generate=MagicMock(side_effect=_fresh_generate_result(response_text)) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), ): result = generate_with_result( "hello", "test.context", json_schema=schema, )
assert result["text"] == response_text assert "truncated" not in result["schema_validation"] assert result["schema_validation"]["valid"] is False assert result["schema_validation"]["errors"][0]["path"] == "/reasoning" assert result["schema_validation"]["errors"][0]["constraint"] == "type"
def test_clean_annotated_response_preserves_byte_identity(self): schema = { "type": "object", "properties": { "z": { "type": "string", "maxLength": 10, SCHEMA_TRUNCATE_KEY: True, }, "a": {"type": "string", "maxLength": 10}, }, "required": ["z", "a"], "additionalProperties": False, } response_text = '{\n "z" : "é",\n "a" : "ok"\n}' provider_module = SimpleNamespace( run_generate=MagicMock(side_effect=_fresh_generate_result(response_text)) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), ): result = generate_with_result( "hello", "test.context", json_schema=schema, )
assert result["text"] == response_text assert result["schema_validation"] == {"valid": True, "errors": []}
def test_async_generate_entrypoints_match_sync_truncation(self): schema = { "type": "object", "properties": { "field": { "type": "string", "maxLength": 3, SCHEMA_TRUNCATE_KEY: True, } }, "required": ["field"], "additionalProperties": False, } response_text = json.dumps({"field": "abcdef"}) provider_module = SimpleNamespace( run_generate=MagicMock(side_effect=_fresh_generate_result(response_text)), run_agenerate=AsyncMock(side_effect=_fresh_generate_result(response_text)), )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), ): sync_text = generate("hello", "test.context", json_schema=schema) sync_result = generate_with_result( "hello", "test.context", json_schema=schema, ) async_text = asyncio.run( agenerate("hello", "test.context", json_schema=schema) ) async_result = asyncio.run( agenerate_with_result( "hello", "test.context", json_schema=schema, ) )
expected_text = json.dumps({"field": "abc"}, ensure_ascii=False) expected_validation = { "valid": True, "errors": [], "truncated": ["/field"], } assert sync_text == expected_text assert async_text == expected_text assert sync_result["text"] == expected_text assert async_result["text"] == expected_text assert sync_result["schema_validation"] == expected_validation assert async_result["schema_validation"] == expected_validation
def test_generate_with_result_adds_schema_validation(self): provider_module = SimpleNamespace( run_generate=MagicMock(return_value={"text": "{}", "finish_reason": "stop"}) ) validation = {"valid": True, "errors": []}
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), patch("solstone.think.models._validate_schema", return_value=validation), ): result = generate_with_result( "hello", "test.context", json_schema={"type": "object"}, )
assert result["schema_validation"] == validation
def test_agenerate_with_result_adds_schema_validation(self): provider_module = SimpleNamespace( run_agenerate=AsyncMock( return_value={"text": "{}", "finish_reason": "stop"} ) ) validation = {"valid": True, "errors": []}
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), patch("solstone.think.models._validate_schema", return_value=validation), ): result = asyncio.run( agenerate_with_result( "hello", "test.context", json_schema={"type": "object"}, ) )
assert result["schema_validation"] == validation
def test_generate_with_result_returns_failed_schema_validation_without_raising( self, ): provider_module = SimpleNamespace( run_generate=MagicMock( return_value={"text": '{"field": "bad"}', "finish_reason": "stop"} ) ) validation = { "valid": False, "errors": [{"path": "/field", "constraint": "type", "message": "bad"}], }
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), patch("solstone.think.models._validate_schema", return_value=validation), ): result = generate_with_result( "hello", "test.context", json_schema={"type": "object"}, )
assert result["schema_validation"] == validation
def test_generate_with_result_omits_schema_validation_without_schema(self): provider_module = SimpleNamespace( run_generate=MagicMock(return_value={"text": "{}", "finish_reason": "stop"}) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), patch("solstone.think.models._validate_schema") as mock_validate_schema, ): result = generate_with_result("hello", "test.context")
assert "schema_validation" not in result mock_validate_schema.assert_not_called()
def test_generate_and_agenerate_do_not_surface_schema_validation(self): sync_provider = SimpleNamespace( run_generate=MagicMock(return_value={"text": "{}", "finish_reason": "stop"}) ) async_provider = SimpleNamespace( run_agenerate=AsyncMock( return_value={"text": "{}", "finish_reason": "stop"} ) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=sync_provider, ), patch( "solstone.think.models._validate_schema", return_value={"valid": True, "errors": []}, ) as mock_validate_schema, ): sync_result = generate( "hello", "test.context", json_schema={"type": "object"} )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=async_provider, ), patch( "solstone.think.models._validate_schema", return_value={"valid": True, "errors": []}, ) as mock_async_validate, ): async_result = asyncio.run( agenerate("hello", "test.context", json_schema={"type": "object"}) )
assert sync_result == "{}" assert async_result == "{}" mock_validate_schema.assert_called_once() mock_async_validate.assert_called_once()
def test_generate_raises_schema_validation_error_with_structured_fields(self): long_text = '{"field": "' + ("x" * 300) + '"}' provider_module = SimpleNamespace( run_generate=MagicMock( return_value={"text": long_text, "finish_reason": "stop"} ) ) schema = { "type": "object", "properties": {"field": {"type": "integer"}}, }
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), ): with pytest.raises(SchemaValidationError) as exc_info: generate("hello", "test.context", json_schema=schema)
exc = exc_info.value assert exc.text == long_text assert exc.errors[0]["constraint"] == "type" assert len(exc.preview) == 200 assert exc.preview.endswith("...") assert "x" * 250 not in str(exc)
def test_agenerate_raises_schema_validation_error(self): provider_module = SimpleNamespace( run_agenerate=AsyncMock( return_value={"text": '{"field": "bad"}', "finish_reason": "stop"} ) ) schema = { "type": "object", "properties": {"field": {"type": "integer"}}, }
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), ): with pytest.raises(SchemaValidationError) as exc_info: asyncio.run(agenerate("hello", "test.context", json_schema=schema))
assert exc_info.value.text == '{"field": "bad"}' assert exc_info.value.errors[0]["constraint"] == "type"
def test_generate_empty_schema_response_raises_provider_invalid_before_schema( self, ): provider_module = SimpleNamespace( run_generate=MagicMock(return_value={"text": "", "finish_reason": "stop"}) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), patch("solstone.think.models._validate_schema") as mock_validate_schema, ): with pytest.raises(ProviderResponseInvalidError) as exc_info: generate("hello", "test.context", json_schema={"type": "object"})
assert exc_info.value.reason == "blank_visible_output" mock_validate_schema.assert_not_called()
def test_truncation_raises_before_schema_validation(self): provider_module = SimpleNamespace( run_generate=MagicMock( return_value={"text": "{}", "finish_reason": "max_tokens"} ) )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), patch("solstone.think.models._validate_schema") as mock_validate_schema, ): with pytest.raises(IncompleteJSONError): generate_with_result( "hello", "test.context", json_schema={"type": "object"} )
mock_validate_schema.assert_not_called()
class TestDefaultProviderTimeout: def test_default_timeout_forwards_to_all_generate_entrypoints(self): provider_module = SimpleNamespace( run_generate=MagicMock( return_value={"text": "ok", "finish_reason": "stop"} ), run_agenerate=AsyncMock( return_value={"text": "ok", "finish_reason": "stop"} ), )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), ): generate("hello", "test.context") generate_with_result("hello", "test.context") asyncio.run(agenerate("hello", "test.context"))
assert ( provider_module.run_generate.call_args_list[0].kwargs["timeout_s"] == DEFAULT_PROVIDER_TIMEOUT_S ) assert ( provider_module.run_generate.call_args_list[1].kwargs["timeout_s"] == DEFAULT_PROVIDER_TIMEOUT_S ) assert ( provider_module.run_agenerate.call_args.kwargs["timeout_s"] == DEFAULT_PROVIDER_TIMEOUT_S )
def test_explicit_timeout_wins_for_all_generate_entrypoints(self): provider_module = SimpleNamespace( run_generate=MagicMock( return_value={"text": "ok", "finish_reason": "stop"} ), run_agenerate=AsyncMock( return_value={"text": "ok", "finish_reason": "stop"} ), )
with ( patch( "solstone.think.models.resolve_provider", return_value=("fake", "model") ), patch( "solstone.think.providers.get_provider_module", return_value=provider_module, ), ): generate("hello", "test.context", timeout_s=5) generate_with_result("hello", "test.context", timeout_s=5) asyncio.run(agenerate("hello", "test.context", timeout_s=5))
assert provider_module.run_generate.call_args_list[0].kwargs["timeout_s"] == 5 assert provider_module.run_generate.call_args_list[1].kwargs["timeout_s"] == 5 assert provider_module.run_agenerate.call_args.kwargs["timeout_s"] == 5