solpbc.org / solstone
personal memory agent
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

fix(think): make daily-unit failure caps terminal for day completion

A daily unit at its deterministic-failure cap was permanently skipped for dispatch, but the day-completion predicate still demanded every applicable unit be complete. That meant health/daily.updated was never written, the catchup scheduler re-queued the day forever, and journal doctor reported it stuck indefinitely.

Install the invariant that a unit's failure cap is terminal: the same predicate that stops dispatch marks the unit terminal-degraded for day completion. Day completion means all applicable units terminal, not all units succeeded. Degradation is terminal but visible.

Recalibrate deterministic caps so schema_invalid moves to cap 3 while every other deterministic reason stays at 2. schema_invalid measured 24.3% per-call failure on the affected talent (87 complete / 28 schema_invalid since the local cutover), with same-day fail-then-pass observed on 20260723 for entity_observer:vconic, failed 00:24 and completed 00:36. The other reasons are unmeasured and deliberately kept tight.

Replace DETERMINISTIC_FAILURE_THRESHOLD outright with the cogitate_policy.py cap SOT, DETERMINISTIC_FAILURE_CAPS plus failure_capped. Extract thinking.evaluate_daily_completion as the pure predicate and thinking.finalize_day_completion as the marker-write, payload, and log seam from the inline main() block. Surface capped units on the daily_complete payload, the BacklogDay complete-day path, and the journal doctor caught-up check.

retry_on_deterministic_failure deliberately affects dispatch only; the completion predicate treats capped units as terminal regardless of the flag.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

author
Jer Miller co-author
Claude Opus 4.8 (1M context) date (Jul 24, 2026, 12:30 PM -0600) commit 585ab459 parent ffb4b7cc
+727 -24
+26 -2
solstone/think/cogitate_policy.py
··· 33 33 # output rate ($2.50 / 1M tokens) for ALL fresh non-cache tokens so the estimate 34 34 # errs high and the ceiling trips early rather than late. 35 35 _FALLBACK_USD_PER_TOKEN = 0.0000025 36 - DETERMINISTIC_FAILURE_THRESHOLD = 2 37 36 DEFAULT_READ_CALL_BUDGET = 200 38 37 # Reason codes for content-deterministic crashes and high-recurrence stochastic 39 - # failures we decline to auto-retry past the threshold. 38 + # failures we decline to auto-retry past their caps. 40 39 DETERMINISTIC_FAILURE_REASON_CODES = frozenset( 41 40 { 42 41 "agent_stuck", ··· 48 47 "wall_clock_exceeded", 49 48 } 50 49 ) 50 + # Single source of truth for deterministic failure caps consumed by 51 + # thinking._check_daily_skip and thinking.evaluate_daily_completion. The 52 + # judgement shape is the latest deterministic reason plus the total in-set 53 + # deterministic count from pipeline_health.read_daily_deterministic_failures. 54 + # Scope-provided calibration: schema_invalid measured 24.3% per-call failure on 55 + # the affected talent (87 complete / 28 schema_invalid since the local cutover), 56 + # with same-day fail-then-pass observed on 20260723 for entity_observer:vconic 57 + # (failed 00:24, completed 00:36). The other reasons are unmeasured and kept 58 + # deliberately tight. 59 + DETERMINISTIC_FAILURE_CAPS: dict[str, int] = { 60 + "agent_stuck": 2, 61 + "context_window_exceeded": 2, 62 + "max_turns_exhausted": 2, 63 + "no_output": 2, 64 + "schema_invalid": 3, 65 + "token_budget_exceeded": 2, 66 + "wall_clock_exceeded": 2, 67 + } 68 + 69 + 70 + def failure_capped(reason_code: str | None, count: int) -> bool: 71 + """Return True when a deterministic failure count reaches its cap.""" 72 + cap = DETERMINISTIC_FAILURE_CAPS.get(reason_code or "") 73 + return cap is not None and count >= cap 74 + 51 75 52 76 _JOURNAL_COMMANDS = {"identity", "health", "talent"} 53 77 _SHELL_OPERATOR_CHARS = frozenset("();<>|&")
+13
solstone/think/doctor.py
··· 904 904 check = JOURNAL_CAUGHT_UP_CHECK 905 905 try: 906 906 from solstone.think.pipeline_health import ( 907 + BACKLOG_STATE_COMPLETE, 907 908 BACKLOG_STATE_UNKNOWN, 908 909 read_backlog_view, 909 910 ) ··· 924 925 "warn", 925 926 f"couldn't fully determine — {unknown_days} day(s) unknown", 926 927 _CAUGHT_UP_CANT_TELL_FIX, 928 + ) 929 + 930 + capped_complete_days = sum( 931 + 1 932 + for day in view.days 933 + if day.state == BACKLOG_STATE_COMPLETE and day.capped_daily_unit_count > 0 934 + ) 935 + if view.pending_days == 0 and view.stuck_days == 0 and capped_complete_days > 0: 936 + return make_result( 937 + check, 938 + "ok", 939 + f"caught up; {capped_complete_days} day(s) completed with capped daily unit(s)", 927 940 ) 928 941 929 942 if view.pending_days == 0 and view.stuck_days == 0:
+3
solstone/think/journal_stats.py
··· 100 100 data["segment_repair_cleared"] = day.segment_repair_cleared 101 101 if day.segment_repair_remaining is not None: 102 102 data["segment_repair_remaining"] = day.segment_repair_remaining 103 + if day.capped_daily_unit_count > 0: 104 + data["capped_daily_unit_count"] = day.capped_daily_unit_count 105 + data["capped_daily_unit"] = day.capped_daily_unit 103 106 return data 104 107 105 108
+30 -1
solstone/think/pipeline_health.py
··· 18 18 read_segment_repair_summary, 19 19 ) 20 20 from solstone.think.cluster import cluster_segments 21 - from solstone.think.cogitate_policy import DETERMINISTIC_FAILURE_REASON_CODES 21 + from solstone.think.cogitate_policy import ( 22 + DETERMINISTIC_FAILURE_REASON_CODES, 23 + failure_capped, 24 + ) 22 25 from solstone.think.data_state import DataState 23 26 from solstone.think.utils import ( 24 27 DEFAULT_STREAM, ··· 237 240 segment_repair_bounded: bool | None = None 238 241 segment_repair_cleared: int | None = None 239 242 segment_repair_remaining: int | None = None 243 + capped_daily_unit_count: int = 0 244 + capped_daily_unit: dict[str, object] | None = None 240 245 241 246 242 247 @dataclass(frozen=True) ··· 1401 1406 return tuple(why) 1402 1407 1403 1408 1409 + def _capped_daily_complete_fields(day: str) -> dict[str, object]: 1410 + capped = [ 1411 + { 1412 + "name": name, 1413 + "facet": facet, 1414 + "reason_code": failure.reason_code, 1415 + "count": failure.count, 1416 + } 1417 + for (name, facet), failure in sorted( 1418 + read_daily_deterministic_failures(day).items(), 1419 + key=lambda item: (item[0][0], item[0][1] or ""), 1420 + ) 1421 + if failure_capped(failure.reason_code, failure.count) 1422 + ] 1423 + if not capped: 1424 + return {} 1425 + return { 1426 + "capped_daily_unit_count": len(capped), 1427 + "capped_daily_unit": capped[0], 1428 + } 1429 + 1430 + 1404 1431 def _complete_backlog_day(day: str) -> BacklogDay: 1405 1432 return BacklogDay( 1406 1433 day=day, ··· 1414 1441 provider=None, 1415 1442 model=None, 1416 1443 error=None, 1444 + **_capped_daily_complete_fields(day), 1417 1445 ) 1418 1446 1419 1447 ··· 1486 1514 provider=None, 1487 1515 model=None, 1488 1516 error=error, 1517 + **_capped_daily_complete_fields(day), 1489 1518 **_segment_repair_fields(repair), 1490 1519 ) 1491 1520
+120 -21
solstone/think/thinking.py
··· 19 19 import threading 20 20 import time 21 21 from concurrent.futures import ThreadPoolExecutor, as_completed 22 + from dataclasses import dataclass 22 23 from datetime import date, datetime, timedelta, timezone 23 24 from pathlib import Path 24 25 from typing import Any ··· 39 40 ) 40 41 from solstone.think.change_detection import detect_segment_change, resolve_predecessor 41 42 from solstone.think.cluster import cluster_segments, read_segment_data_state 42 - from solstone.think.cogitate_policy import DETERMINISTIC_FAILURE_THRESHOLD 43 + from solstone.think.cogitate_policy import failure_capped 43 44 from solstone.think.cortex_client import ( 44 45 PATIENT_CLAIM_WINDOWS, 45 46 CortexNotClaimed, ··· 906 907 _SKIPPED: object = object() 907 908 908 909 910 + @dataclass(frozen=True) 911 + class CappedDailyUnit: 912 + name: str 913 + facet: str | None 914 + reason_code: str 915 + count: int 916 + 917 + 918 + @dataclass(frozen=True) 919 + class DailyCompletionVerdict: 920 + complete: bool 921 + daily_units_terminal: bool 922 + segment_blockers: tuple[dict[str, str], ...] 923 + capped_daily_units: tuple[CappedDailyUnit, ...] 924 + 925 + 909 926 class _NotClaimed: 910 927 __slots__ = ("use_id",) 911 928 ··· 1267 1284 return (True, "already_complete") 1268 1285 if not retry_on_deterministic_failure: 1269 1286 failure = deterministic_failures.get((name, facet)) 1270 - if failure is not None and failure.count >= DETERMINISTIC_FAILURE_THRESHOLD: 1287 + # Dispatch uses cogitate_policy.failure_capped, as does 1288 + # evaluate_daily_completion. retry_on_deterministic_failure affects 1289 + # dispatch only; completed day finalization ends retries, so forcing 1290 + # convergence-by-retry with this flag is unsupported. 1291 + if failure is not None and failure_capped(failure.reason_code, failure.count): 1271 1292 return (True, "deterministic_failure_no_retry") 1272 1293 return (False, None) 1294 + 1295 + 1296 + def evaluate_daily_completion( 1297 + applicable_units: set[tuple[str, str | None]], 1298 + completed_units: set[tuple[str, str, str | None]], 1299 + deterministic_failures: dict[tuple[str, str | None], DeterministicFailure], 1300 + segment_blockers: list[dict[str, str]] | tuple[dict[str, str], ...], 1301 + ) -> DailyCompletionVerdict: 1302 + """Return the terminal daily completion verdict without journal writes.""" 1303 + capped_daily_units: list[CappedDailyUnit] = [] 1304 + daily_units_terminal = True 1305 + 1306 + # Invariant shared with _check_daily_skip via cogitate_policy.failure_capped: 1307 + # a unit's failure cap is terminal; the same predicate that stops dispatch 1308 + # marks the unit terminal-degraded for day completion. Day completion means 1309 + # all applicable units terminal, not all units succeeded. Degradation is 1310 + # terminal but visible. This deliberately ignores the dispatch retry override. 1311 + for name, facet in sorted( 1312 + applicable_units, key=lambda unit: (unit[0], unit[1] or "") 1313 + ): 1314 + if ("daily", name, facet) in completed_units: 1315 + continue 1316 + failure = deterministic_failures.get((name, facet)) 1317 + if failure is not None and failure_capped(failure.reason_code, failure.count): 1318 + capped_daily_units.append( 1319 + CappedDailyUnit( 1320 + name=name, 1321 + facet=facet, 1322 + reason_code=failure.reason_code, 1323 + count=failure.count, 1324 + ) 1325 + ) 1326 + continue 1327 + daily_units_terminal = False 1328 + 1329 + blockers = tuple(segment_blockers) 1330 + return DailyCompletionVerdict( 1331 + complete=daily_units_terminal and not blockers, 1332 + daily_units_terminal=daily_units_terminal, 1333 + segment_blockers=blockers, 1334 + capped_daily_units=tuple(capped_daily_units), 1335 + ) 1336 + 1337 + 1338 + def _capped_daily_unit_payload(unit: CappedDailyUnit) -> dict[str, object]: 1339 + return { 1340 + "name": unit.name, 1341 + "facet": unit.facet, 1342 + "reason_code": unit.reason_code, 1343 + "count": unit.count, 1344 + } 1345 + 1346 + 1347 + def finalize_day_completion( 1348 + day: str, verdict: DailyCompletionVerdict 1349 + ) -> dict[str, object]: 1350 + """Write or withhold the daily marker and return daily_complete extras.""" 1351 + capped_payload = [ 1352 + _capped_daily_unit_payload(unit) for unit in verdict.capped_daily_units 1353 + ] 1354 + payload_fragment: dict[str, object] = {} 1355 + if capped_payload: 1356 + payload_fragment["capped_daily_units"] = capped_payload 1357 + 1358 + if verdict.complete: 1359 + health_dir = day_path(day) / "health" 1360 + health_dir.mkdir(parents=True, exist_ok=True) 1361 + (health_dir / "daily.updated").touch() 1362 + if capped_payload: 1363 + logging.info( 1364 + "Day %s complete with capped daily unit(s); wrote daily.updated: capped_daily_units=%s", 1365 + day, 1366 + capped_payload, 1367 + ) 1368 + else: 1369 + logging.info("Day %s fully complete; wrote daily.updated", day) 1370 + else: 1371 + logging.info( 1372 + "Day %s withholding daily.updated: daily_units_terminal=%s segment_blockers=%s capped_daily_units=%s", 1373 + day, 1374 + verdict.daily_units_terminal, 1375 + list(verdict.segment_blockers), 1376 + capped_payload, 1377 + ) 1378 + 1379 + return payload_fragment 1273 1380 1274 1381 1275 1382 def run_segment_sense( ··· 4704 4811 ) 4705 4812 4706 4813 # Touch daily.updated marker only after daily and segment work completes. 4814 + completion_payload_fragment: dict[str, object] = {} 4707 4815 try: 4708 4816 completed = read_completed_units(day) 4709 - daily_done = all( 4710 - ("daily", name, facet) in completed 4711 - for name, facet in applicable_units 4712 - ) 4817 + deterministic_failures = read_daily_deterministic_failures(day) 4713 4818 4714 4819 segments = cluster_segments(day) 4715 4820 progress = read_segment_progress(day) 4716 4821 completion = classify_segment_completion(segments, progress) 4717 4822 blocked_after_cycle = blocked_segment_keys(segments, progress) 4718 - blockers = completion.blockers 4719 - 4720 - if daily_done and not blockers: 4721 - health_dir = day_path(day) / "health" 4722 - health_dir.mkdir(parents=True, exist_ok=True) 4723 - (health_dir / "daily.updated").touch() 4724 - logging.info("Day %s fully complete; wrote daily.updated", day) 4725 - else: 4726 - logging.info( 4727 - "Day %s withholding daily.updated: " 4728 - "daily_units_complete=%s segment_blockers=%s", 4729 - day, 4730 - daily_done, 4731 - blockers, 4732 - ) 4823 + verdict = evaluate_daily_completion( 4824 + applicable_units, 4825 + completed, 4826 + deterministic_failures, 4827 + completion.blockers, 4828 + ) 4829 + completion_payload_fragment = finalize_day_completion(day, verdict) 4733 4830 except Exception: 4734 4831 logging.warning("Failed to update daily marker", exc_info=True) 4735 4832 ··· 4749 4846 ) 4750 4847 if remaining_cycle is not None: 4751 4848 daily_complete_payload["remaining"] = remaining_cycle 4849 + if completion_payload_fragment: 4850 + daily_complete_payload.update(completion_payload_fragment) 4752 4851 4753 4852 # Set first_daily_ready awareness flag after first daily analysis 4754 4853 try:
+17
tests/test_cogitate_policy.py
··· 40 40 ) == ["chronicle/20260425", "chronicle/20260426", "chronicle/20260427"] 41 41 42 42 43 + def test_failure_capped_schema_invalid_cap_is_three(): 44 + assert cogitate_policy.failure_capped("schema_invalid", 2) is False 45 + assert cogitate_policy.failure_capped("schema_invalid", 3) is True 46 + 47 + 48 + def test_failure_capped_default_deterministic_cap_is_two(): 49 + assert cogitate_policy.failure_capped("context_window_exceeded", 1) is False 50 + assert cogitate_policy.failure_capped("context_window_exceeded", 2) is True 51 + 52 + 53 + def test_deterministic_failure_caps_cover_reason_codes_exactly(): 54 + assert ( 55 + set(cogitate_policy.DETERMINISTIC_FAILURE_CAPS) 56 + == cogitate_policy.DETERMINISTIC_FAILURE_REASON_CODES 57 + ) 58 + 59 + 43 60 def test_policy_denies_write_tools(tmp_path): 44 61 policy = _policy(tmp_path) 45 62
+56
tests/test_journal_caught_up.py
··· 229 229 assert result.detail == "caught up" 230 230 231 231 232 + def test_journal_caught_up_ok_with_capped_complete_days_detail(doctor, monkeypatch): 233 + day = BacklogDay( 234 + day="20200229", 235 + state=BACKLOG_STATE_COMPLETE, 236 + segments=0, 237 + units=0, 238 + not_sensed=0, 239 + why=(), 240 + reason=None, 241 + reason_code=None, 242 + provider=None, 243 + model=None, 244 + error=None, 245 + capped_daily_unit_count=1, 246 + capped_daily_unit={ 247 + "name": "entities:entity_observer", 248 + "facet": "vconic", 249 + "reason_code": "context_window_exceeded", 250 + "count": 2, 251 + }, 252 + ) 253 + view = BacklogView( 254 + window=30, 255 + days=(day,), 256 + pending_days=0, 257 + stuck_days=0, 258 + oldest_pending_day=None, 259 + errors=(), 260 + ) 261 + monkeypatch.setattr(pipeline_health, "read_backlog_view", lambda: view) 262 + 263 + result = doctor.journal_caught_up_check(args(doctor)) 264 + 265 + assert result.status == "ok" 266 + assert result.detail == "caught up; 1 day(s) completed with capped daily unit(s)" 267 + 268 + 269 + def test_journal_caught_up_plain_caught_up_without_capped_complete_days( 270 + doctor, monkeypatch 271 + ): 272 + view = BacklogView( 273 + window=30, 274 + days=(backlog_day("20200229", BACKLOG_STATE_COMPLETE),), 275 + pending_days=0, 276 + stuck_days=0, 277 + oldest_pending_day=None, 278 + errors=(), 279 + ) 280 + monkeypatch.setattr(pipeline_health, "read_backlog_view", lambda: view) 281 + 282 + result = doctor.journal_caught_up_check(args(doctor)) 283 + 284 + assert result.status == "ok" 285 + assert result.detail == "caught up" 286 + 287 + 232 288 def test_journal_caught_up_reports_backoff_stuck_day(doctor, tmp_path, monkeypatch): 233 289 journal = tmp_path / "journal" 234 290 day = "20990401"
+52
tests/test_journal_stats.py
··· 214 214 assert progressing_data["segment_repair_remaining"] == 5 215 215 216 216 217 + def test_serialize_backlog_day_includes_capped_daily_fields_only_when_present(): 218 + stats_mod = importlib.import_module("solstone.think.journal_stats") 219 + health_mod = importlib.import_module("solstone.think.pipeline_health") 220 + 221 + clean = health_mod.BacklogDay( 222 + day="20240101", 223 + state=health_mod.BACKLOG_STATE_COMPLETE, 224 + segments=0, 225 + units=0, 226 + not_sensed=0, 227 + why=(), 228 + reason=None, 229 + reason_code=None, 230 + provider=None, 231 + model=None, 232 + error=None, 233 + ) 234 + capped = health_mod.BacklogDay( 235 + day="20240102", 236 + state=health_mod.BACKLOG_STATE_COMPLETE, 237 + segments=0, 238 + units=0, 239 + not_sensed=0, 240 + why=(), 241 + reason=None, 242 + reason_code=None, 243 + provider=None, 244 + model=None, 245 + error=None, 246 + capped_daily_unit_count=1, 247 + capped_daily_unit={ 248 + "name": "entities:entity_observer", 249 + "facet": "vconic", 250 + "reason_code": "context_window_exceeded", 251 + "count": 2, 252 + }, 253 + ) 254 + 255 + clean_data = stats_mod._serialize_backlog_day(clean) 256 + capped_data = stats_mod._serialize_backlog_day(capped) 257 + 258 + assert "capped_daily_unit_count" not in clean_data 259 + assert "capped_daily_unit" not in clean_data 260 + assert capped_data["capped_daily_unit_count"] == 1 261 + assert capped_data["capped_daily_unit"] == { 262 + "name": "entities:entity_observer", 263 + "facet": "vconic", 264 + "reason_code": "context_window_exceeded", 265 + "count": 2, 266 + } 267 + 268 + 217 269 def test_scan_day(tmp_path, monkeypatch): 218 270 stats_mod = importlib.import_module("solstone.think.journal_stats") 219 271 journal = tmp_path
+54
tests/test_pipeline_health.py
··· 395 395 assert view.stuck_days == 0 396 396 397 397 398 + def test_backlog_complete_day_reports_capped_daily_unit_without_pending( 399 + pipeline_journal, 400 + ): 401 + capped_day = "20990310" 402 + clean_day = "20990311" 403 + capped_base = pipeline_journal / "chronicle" / capped_day / "health" 404 + clean_base = pipeline_journal / "chronicle" / clean_day / "health" 405 + _write_jsonl( 406 + capped_base / "001_daily.jsonl", 407 + [ 408 + { 409 + "event": "talent.fail", 410 + "ts": 1, 411 + "mode": "daily", 412 + "name": "entities:entity_observer", 413 + "facet": "vconic", 414 + "reason_code": "context_window_exceeded", 415 + }, 416 + { 417 + "event": "talent.fail", 418 + "ts": 2, 419 + "mode": "daily", 420 + "name": "entities:entity_observer", 421 + "facet": "vconic", 422 + "reason_code": "context_window_exceeded", 423 + }, 424 + ], 425 + ) 426 + _write_jsonl( 427 + clean_base / "001_daily.jsonl", 428 + [{"event": "talent.complete", "ts": 1, "mode": "daily", "name": "alpha"}], 429 + ) 430 + for day in (capped_day, clean_day): 431 + _touch_marker(pipeline_journal, day, "stream.updated", mtime_ms=1000) 432 + _touch_marker(pipeline_journal, day, "daily.updated", mtime_ms=2000) 433 + 434 + view = read_backlog_view(window=2) 435 + by_day = {item.day: item for item in view.days} 436 + 437 + assert by_day[capped_day].state == BACKLOG_STATE_COMPLETE 438 + assert by_day[capped_day].capped_daily_unit_count == 1 439 + assert by_day[capped_day].capped_daily_unit == { 440 + "name": "entities:entity_observer", 441 + "facet": "vconic", 442 + "reason_code": "context_window_exceeded", 443 + "count": 2, 444 + } 445 + assert by_day[clean_day].state == BACKLOG_STATE_COMPLETE 446 + assert by_day[clean_day].capped_daily_unit_count == 0 447 + assert by_day[clean_day].capped_daily_unit is None 448 + assert view.pending_days == 0 449 + assert view.stuck_days == 0 450 + 451 + 398 452 def test_read_completed_units_terminal_presence(pipeline_journal): 399 453 day = "20990202" 400 454 base = pipeline_journal / "chronicle" / day / "health"
+356
tests/test_think_daily_idempotency.py
··· 7 7 8 8 import importlib 9 9 import json 10 + import os 10 11 from pathlib import Path 11 12 12 13 import pytest 14 + 15 + from solstone.think.cogitate_policy import DETERMINISTIC_FAILURE_CAPS 16 + from solstone.think.pipeline_health import ( 17 + DeterministicFailure, 18 + read_completed_units, 19 + read_daily_deterministic_failures, 20 + ) 21 + from solstone.think.utils import updated_days 13 22 14 23 DAY = "20990301" 15 24 ··· 172 181 assert "from_scratch" in names 173 182 174 183 184 + @pytest.mark.parametrize("retry_on_deterministic_failure", [False, True]) 185 + @pytest.mark.parametrize("at_cap", [False, True]) 186 + @pytest.mark.parametrize( 187 + ("reason_code", "cap"), sorted(DETERMINISTIC_FAILURE_CAPS.items()) 188 + ) 189 + def test_daily_skip_and_completion_cap_predicates_match( 190 + retry_on_deterministic_failure, 191 + at_cap, 192 + reason_code, 193 + cap, 194 + ): 195 + mod = importlib.import_module("solstone.think.thinking") 196 + count = cap if at_cap else cap - 1 197 + deterministic_failures = { 198 + ("beta", None): DeterministicFailure(count=count, reason_code=reason_code) 199 + } 200 + 201 + skip, _reason = mod._check_daily_skip( 202 + "beta", 203 + None, 204 + mode="daily", 205 + completed=set(), 206 + deterministic_failures=deterministic_failures, 207 + retry_on_deterministic_failure=retry_on_deterministic_failure, 208 + ) 209 + verdict = mod.evaluate_daily_completion( 210 + {("beta", None)}, 211 + set(), 212 + deterministic_failures, 213 + [], 214 + ) 215 + terminal_degraded = bool(verdict.capped_daily_units) 216 + 217 + if retry_on_deterministic_failure: 218 + assert skip is False 219 + assert terminal_degraded is at_cap 220 + else: 221 + assert skip is terminal_degraded 222 + assert not (skip and not terminal_degraded) 223 + 224 + 225 + def test_evaluate_daily_completion_terminal_cases(): 226 + mod = importlib.import_module("solstone.think.thinking") 227 + 228 + complete_and_capped = mod.evaluate_daily_completion( 229 + {("alpha", None), ("beta", None)}, 230 + {("daily", "alpha", None)}, 231 + { 232 + ("beta", None): DeterministicFailure( 233 + count=2, reason_code="context_window_exceeded" 234 + ) 235 + }, 236 + [], 237 + ) 238 + assert complete_and_capped.complete is True 239 + assert complete_and_capped.daily_units_terminal is True 240 + assert complete_and_capped.capped_daily_units == ( 241 + mod.CappedDailyUnit( 242 + name="beta", 243 + facet=None, 244 + reason_code="context_window_exceeded", 245 + count=2, 246 + ), 247 + ) 248 + 249 + below_cap = mod.evaluate_daily_completion( 250 + {("beta", None)}, 251 + set(), 252 + { 253 + ("beta", None): DeterministicFailure( 254 + count=1, reason_code="context_window_exceeded" 255 + ) 256 + }, 257 + [], 258 + ) 259 + assert below_cap.complete is False 260 + assert below_cap.capped_daily_units == () 261 + 262 + 263 + def test_evaluate_daily_completion_transient_latest_stays_incomplete(journal_copy): 264 + mod = importlib.import_module("solstone.think.thinking") 265 + day = "20990319" 266 + _prepare_main_day(journal_copy, day) 267 + _write_health( 268 + journal_copy, 269 + day, 270 + "001_daily.jsonl", 271 + [ 272 + _complete("alpha"), 273 + _fail("beta", ts=1, reason_code="context_window_exceeded"), 274 + _fail("beta", ts=2, reason_code="context_window_exceeded"), 275 + _fail("beta", ts=3, reason_code="schema_invalid"), 276 + _fail("beta", ts=4, reason_code="token_budget_exceeded"), 277 + _fail("beta", ts=5, reason_code="provider_transient"), 278 + ], 279 + ) 280 + 281 + completed = read_completed_units(day) 282 + deterministic_failures = read_daily_deterministic_failures(day) 283 + 284 + assert ("beta", None) not in deterministic_failures 285 + transient_latest = mod.evaluate_daily_completion( 286 + {("alpha", None), ("beta", None)}, 287 + completed, 288 + deterministic_failures, 289 + [], 290 + ) 291 + assert transient_latest.complete is False 292 + assert transient_latest.capped_daily_units == () 293 + 294 + 295 + def test_evaluate_daily_completion_dispatch_without_terminal_stays_incomplete( 296 + journal_copy, 297 + ): 298 + mod = importlib.import_module("solstone.think.thinking") 299 + day = "20990320" 300 + _prepare_main_day(journal_copy, day) 301 + _write_health( 302 + journal_copy, 303 + day, 304 + "001_daily.jsonl", 305 + [ 306 + _complete("alpha"), 307 + {"event": "talent.dispatch", "ts": 2, "mode": "daily", "name": "beta"}, 308 + ], 309 + ) 310 + 311 + completed = read_completed_units(day) 312 + deterministic_failures = read_daily_deterministic_failures(day) 313 + 314 + assert ("daily", "beta", None) not in completed 315 + assert ("beta", None) not in deterministic_failures 316 + dispatched_without_terminal = mod.evaluate_daily_completion( 317 + {("alpha", None), ("beta", None)}, 318 + completed, 319 + deterministic_failures, 320 + [], 321 + ) 322 + assert dispatched_without_terminal.complete is False 323 + assert dispatched_without_terminal.capped_daily_units == () 324 + 325 + 326 + def test_evaluate_daily_completion_withholds_with_segment_blockers(): 327 + mod = importlib.import_module("solstone.think.thinking") 328 + 329 + verdict = mod.evaluate_daily_completion( 330 + {("beta", None)}, 331 + set(), 332 + { 333 + ("beta", None): DeterministicFailure( 334 + count=2, reason_code="context_window_exceeded" 335 + ) 336 + }, 337 + [{"segment": "090000_300", "dimension": "not_thought", "detail": "floor"}], 338 + ) 339 + 340 + assert verdict.complete is False 341 + assert verdict.daily_units_terminal is True 342 + assert verdict.capped_daily_units 343 + assert verdict.segment_blockers == ( 344 + {"segment": "090000_300", "dimension": "not_thought", "detail": "floor"}, 345 + ) 346 + 347 + 175 348 def test_run_daily_prompts_skips_all_completed_units(daily_journal, monkeypatch): 176 349 mod = importlib.import_module("solstone.think.thinking") 177 350 _write_health( ··· 324 497 ) 325 498 326 499 500 + def test_run_daily_prompts_schema_invalid_retries_until_third_failure( 501 + daily_journal, monkeypatch 502 + ): 503 + mod = importlib.import_module("solstone.think.thinking") 504 + _write_health( 505 + daily_journal, 506 + DAY, 507 + "001_daily.jsonl", 508 + [ 509 + _fail("alpha", ts=1, reason_code="schema_invalid"), 510 + _fail("alpha", ts=2, reason_code="schema_invalid"), 511 + ], 512 + ) 513 + dispatched: list[tuple[str, dict]] = [] 514 + _install_daily_mocks(monkeypatch, mod, _single_configs("alpha"), dispatched) 515 + 516 + _run_daily_with_writer(mod, daily_journal, DAY, "002_daily.jsonl") 517 + 518 + assert [name for name, _config in dispatched] == ["alpha"] 519 + dispatched.clear() 520 + _write_health( 521 + daily_journal, 522 + DAY, 523 + "003_daily.jsonl", 524 + [_fail("alpha", ts=3, reason_code="schema_invalid")], 525 + ) 526 + 527 + _run_daily_with_writer(mod, daily_journal, DAY, "004_daily.jsonl") 528 + 529 + assert dispatched == [] 530 + skips = _skip_events(daily_journal, DAY, "004_daily.jsonl") 531 + assert skips[0]["reason"] == "deterministic_failure_no_retry" 532 + assert skips[0]["detail"] == ( 533 + "3 same-day deterministic failures (schema_invalid); not re-dispatching" 534 + ) 535 + 536 + 537 + def test_mixed_deterministic_reason_uses_latest_reason_cap(daily_journal, monkeypatch): 538 + mod = importlib.import_module("solstone.think.thinking") 539 + _write_health( 540 + daily_journal, 541 + DAY, 542 + "001_daily.jsonl", 543 + [ 544 + _fail("alpha", ts=1, reason_code="context_window_exceeded"), 545 + _fail("alpha", ts=2, reason_code="schema_invalid"), 546 + _fail("beta", ts=1, reason_code="schema_invalid"), 547 + _fail("beta", ts=2, reason_code="context_window_exceeded"), 548 + ], 549 + ) 550 + dispatched: list[tuple[str, dict]] = [] 551 + _install_daily_mocks(monkeypatch, mod, _single_configs("alpha", "beta"), dispatched) 552 + 553 + _run_daily_with_writer(mod, daily_journal, DAY, "002_daily.jsonl") 554 + 555 + assert [name for name, _config in dispatched] == ["alpha"] 556 + skips = _skip_events(daily_journal, DAY, "002_daily.jsonl") 557 + assert [(event["name"], event["reason"]) for event in skips] == [ 558 + ("beta", "deterministic_failure_no_retry") 559 + ] 560 + 561 + 327 562 def test_run_daily_prompts_reruns_one_deterministic_failure(daily_journal, monkeypatch): 328 563 mod = importlib.import_module("solstone.think.thinking") 329 564 _write_health( ··· 669 904 assert not (health / "daily.updated").exists() 670 905 671 906 907 + def test_main_writes_daily_marker_when_capped_unit_terminal_and_payload_includes_capped_units( 908 + journal_copy, monkeypatch 909 + ): 910 + mod = importlib.import_module("solstone.think.thinking") 911 + day = "20990316" 912 + health = _prepare_main_day(journal_copy, day) 913 + _write_health( 914 + journal_copy, 915 + day, 916 + "001_daily.jsonl", 917 + [ 918 + _complete("alpha"), 919 + _fail("beta", ts=1, reason_code="context_window_exceeded"), 920 + _fail("beta", ts=2, reason_code="context_window_exceeded"), 921 + ], 922 + ) 923 + _patch_main(monkeypatch, mod, {("alpha", None), ("beta", None)}) 924 + emitted: list[tuple[str, dict]] = [] 925 + monkeypatch.setattr( 926 + mod, 927 + "emit", 928 + lambda event, **fields: emitted.append((event, fields)), 929 + ) 930 + monkeypatch.setattr("sys.argv", ["sol think", "--day", day]) 931 + 932 + mod.main() 933 + 934 + assert (health / "daily.updated").exists() 935 + daily_complete = next( 936 + fields for event, fields in emitted if event == "daily_complete" 937 + ) 938 + assert daily_complete["capped_daily_units"] == [ 939 + { 940 + "name": "beta", 941 + "facet": None, 942 + "reason_code": "context_window_exceeded", 943 + "count": 2, 944 + } 945 + ] 946 + 947 + 948 + def test_degraded_completion_clears_after_later_complete(journal_copy): 949 + mod = importlib.import_module("solstone.think.thinking") 950 + day = "20990317" 951 + _prepare_main_day(journal_copy, day) 952 + _write_health( 953 + journal_copy, 954 + day, 955 + "001_daily.jsonl", 956 + [ 957 + _complete("alpha"), 958 + _fail("beta", ts=1, reason_code="context_window_exceeded"), 959 + _fail("beta", ts=2, reason_code="context_window_exceeded"), 960 + ], 961 + ) 962 + 963 + capped = mod.evaluate_daily_completion( 964 + {("alpha", None), ("beta", None)}, 965 + read_completed_units(day), 966 + read_daily_deterministic_failures(day), 967 + [], 968 + ) 969 + 970 + assert capped.complete is True 971 + assert capped.capped_daily_units 972 + 973 + _write_health(journal_copy, day, "002_daily.jsonl", [_complete("beta", ts=3)]) 974 + cleared = mod.evaluate_daily_completion( 975 + {("alpha", None), ("beta", None)}, 976 + read_completed_units(day), 977 + read_daily_deterministic_failures(day), 978 + [], 979 + ) 980 + 981 + assert cleared.complete is True 982 + assert cleared.capped_daily_units == () 983 + 984 + 672 985 def test_main_ignores_not_applicable_incomplete_units(journal_copy, monkeypatch): 673 986 mod = importlib.import_module("solstone.think.thinking") 674 987 day = "20990312" ··· 708 1021 "from_scratch": False, 709 1022 } 710 1023 ] 1024 + 1025 + 1026 + def test_capped_terminal_completion_clears_updated_days_until_stream_newer( 1027 + journal_copy, monkeypatch 1028 + ): 1029 + mod = importlib.import_module("solstone.think.thinking") 1030 + day = "20990318" 1031 + health = _prepare_main_day(journal_copy, day) 1032 + _write_health( 1033 + journal_copy, 1034 + day, 1035 + "001_daily.jsonl", 1036 + [ 1037 + _complete("alpha"), 1038 + _fail("beta", ts=1, reason_code="context_window_exceeded"), 1039 + _fail("beta", ts=2, reason_code="context_window_exceeded"), 1040 + ], 1041 + ) 1042 + (health / "stream.updated").touch() 1043 + dispatched: list[tuple[str, dict]] = [] 1044 + _install_daily_mocks(monkeypatch, mod, _single_configs("alpha", "beta"), dispatched) 1045 + monkeypatch.setattr( 1046 + mod, "run_bounded_phase", lambda *_args, **_kwargs: (True, False) 1047 + ) 1048 + monkeypatch.setattr(mod, "run_queued_command", lambda *_args, **_kwargs: True) 1049 + monkeypatch.setattr("sys.argv", ["sol think", "--day", day]) 1050 + 1051 + assert day in updated_days() 1052 + mod.main() 1053 + 1054 + assert dispatched == [] 1055 + assert (health / "daily.updated").exists() 1056 + assert day not in updated_days() 1057 + 1058 + os.utime(health / "daily.updated", (1000, 1000)) 1059 + os.utime(health / "stream.updated", (1010, 1010)) 1060 + assert day in updated_days() 1061 + monkeypatch.setattr("sys.argv", ["sol think", "--day", day]) 1062 + 1063 + mod.main() 1064 + 1065 + assert dispatched == [] 1066 + assert day not in updated_days() 711 1067 712 1068 713 1069 def test_main_passes_from_scratch_to_daily_prompts(journal_copy, monkeypatch):