solpbc.org / solstone
personal memory agent
0
Fork 0

Configure Feed

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

feat(catchup): durable segment-repair degradation lane for daily repair fairness

The daily segment_think repair pre-phase already bounds itself and emits a
machine-readable timeout in its phase.complete sidecar, but nothing downstream
reasoned about that degradation: a degraded/timed-out pre-phase let daily
synthesis advance daily.updated, so day_is_complete() returned True and every
caught-up/health surface reported the day as fully caught up — and a poison day
could consume catchup drain slots forever without backing off.

Add a durable, (day, raw-input-fingerprint)-keyed segment-repair lane to the
existing source-layer catchup-state model, reusing its backoff/threshold/
retention discipline:

- catchup_state.py: new KIND_SEGMENT_REPAIR lane with record_segment_repair_
attempt/outcome writers and a fail-closed read_segment_repair_summary reader;
success clears the lane, repeated degradation advances backoff -> quarantine
at STUCK_THRESHOLD, a fingerprint change reopens a fresh attempt. All catchup-
state read-modify-writes are now guarded by hold_lock for cross-process safety
(thinking subprocess + supervisor both write the file).
- thinking.py: records the lane at the daily pre-phase boundary (degraded on
failure carrying wall_clock_exceeded/timeout_seconds/bounded, cleared on
success) for both supervisor-spawned and manual runs.
- supervisor.py: run_catchup_drain skips active/backed-off/quarantined repair
days (unless the fingerprint changed) while later eligible days still fill
slots.
- pipeline_health.py / journal_stats.py: fold the lane into read_backlog_view
with distinct reason codes (segment_repair_degraded/stuck/unknown) mapped onto
PENDING/STUCK/UNKNOWN, fix the day_is_complete COMPLETE short-circuit to
consult repair state first, and set BacklogView.degraded so doctor caught-up,
stats totals, convey verdict, and the health UI all go honest. Unwritable/
unreadable/malformed state surfaces as UNKNOWN, never caught up. Additive
fields only — no stats SCHEMA_VERSION bump.
- check_journal_io_access.py: register catchup_state.py as the owner of its
ops/runtime state file (it now uses hold_lock).

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

author
Jer Miller co-author
Claude Opus 4.8 (1M context) date (Jun 30, 2026, 12:20 AM -0600) commit f434d282 parent e84450db
+1069 -20
+2
scripts/check_journal_io_access.py
··· 98 98 "solstone/apps/observer/utils.py", 99 99 "solstone/think/activities.py", 100 100 "solstone/think/awareness.py", 101 + # Sole ops/runtime writer of health/catchup-state.json; uses hold_lock for cross-process RMW. 102 + "solstone/think/catchup_state.py", 101 103 "solstone/think/day_accumulator.py", 102 104 "solstone/think/entities/journal.py", 103 105 "solstone/think/entities/merge.py",
+34
solstone/apps/health/tests/test_workspace_template.py
··· 370 370 assert 'id="backlogNeedsHand"' not in rendered 371 371 372 372 373 + def test_backlog_segment_repair_stuck_day_renders_needs_hand(health_env): 374 + rendered = _render_health_workspace_with_stats( 375 + health_env, 376 + { 377 + "backlog": _backlog( 378 + stuck_days=1, 379 + days=[ 380 + { 381 + "day": "20260323", 382 + "state": "stuck", 383 + "segments": 0, 384 + "units": 0, 385 + "reason": "segment_repair_stuck", 386 + "reason_code": "segment_repair_stuck", 387 + "segment_repair_status": "stuck", 388 + "segment_repair_attempts": 3, 389 + "segment_repair_consecutive_non_completion": 3, 390 + "segment_repair_last_outcome": "timeout", 391 + "segment_repair_next_retry_at": 1600.0, 392 + "segment_repair_reason_code": "wall_clock_exceeded", 393 + "segment_repair_timeout_seconds": 300, 394 + "segment_repair_bounded": True, 395 + } 396 + ], 397 + ) 398 + }, 399 + ) 400 + 401 + assert _verdict_text(rendered) != backlog_copy.BACKLOG_VERDICT_CAUGHT_UP 402 + section = _section_by_id(rendered, "backlogNeedsHand") 403 + assert "20260323" in section 404 + assert "reason_code=segment_repair_stuck" in section 405 + 406 + 373 407 def test_backlog_needs_hand_bucket_rows(health_env): 374 408 rendered = _render_health_workspace_with_stats( 375 409 health_env,
+148 -16
solstone/think/catchup_state.py
··· 14 14 from datetime import datetime, timedelta 15 15 from pathlib import Path 16 16 17 + from solstone.think.journal_io.locking import hold_lock 17 18 from solstone.think.utils import ( 18 19 day_dirs, 19 20 day_is_complete, ··· 29 30 KIND_DAILY_CATCHUP = "daily-catchup" 30 31 KIND_DAILY_FROM_SCRATCH = "daily-from-scratch" 31 32 KIND_SEGMENT = "segment" 33 + KIND_SEGMENT_REPAIR = "segment-repair" 32 34 RECORDED_KINDS = frozenset({KIND_DAILY_CATCHUP, KIND_DAILY_FROM_SCRATCH, KIND_SEGMENT}) 35 + RECONCILABLE_KINDS = RECORDED_KINDS | frozenset({KIND_SEGMENT_REPAIR}) 36 + 37 + SEGMENT_REPAIR_TIMEOUT_REASON = "wall_clock_exceeded" 38 + SEGMENT_REPAIR_FAILED_REASON = "repair_failed" 33 39 34 40 BACKOFF_BASE_SECONDS = 600 35 41 BACKOFF_MAX_SECONDS = 86400 ··· 178 184 "notified_at": None, 179 185 "fingerprint": None, 180 186 "active": None, 187 + "reason_code": None, 188 + "timeout_seconds": None, 189 + "bounded": None, 181 190 } 182 191 183 192 ··· 311 320 } 312 321 313 322 323 + def read_segment_repair_summary(day: str) -> dict | None: 324 + path = Path(get_journal()) / "health" / "catchup-state.json" 325 + try: 326 + with open(path, "r", encoding="utf-8") as file: 327 + raw = json.load(file) 328 + except FileNotFoundError: 329 + return None 330 + except (json.JSONDecodeError, OSError) as exc: 331 + logger.warning( 332 + "Failed to read segment-repair catchup state for %s: %s", day, exc 333 + ) 334 + return {"status": "unknown"} 335 + 336 + state = _normalize_state(raw) 337 + record = state["entries"].get(_key(day, KIND_SEGMENT_REPAIR)) 338 + if not isinstance(record, dict): 339 + return None 340 + 341 + consecutive = int(record.get("consecutive_non_completion") or 0) 342 + if consecutive == 0: 343 + return None 344 + if read_raw_input_fingerprint(day) != record.get("fingerprint"): 345 + return None 346 + 347 + status = "stuck" if record.get("entered_backoff_at") is not None else "degraded" 348 + return { 349 + "status": status, 350 + "attempts": int(record.get("attempts") or 0), 351 + "consecutive_non_completion": consecutive, 352 + "last_outcome": record.get("last_outcome") or "", 353 + "next_retry_at": float(record.get("next_retry_at") or 0), 354 + "repair_reason_code": record.get("reason_code"), 355 + "timeout_seconds": record.get("timeout_seconds"), 356 + "bounded": record.get("bounded"), 357 + } 358 + 359 + 314 360 def _prune(state: dict) -> None: 315 361 days = day_dirs() 316 362 if not days: ··· 360 406 read_raw_input_fingerprint(cmd_day) if kind == KIND_DAILY_CATCHUP else None 361 407 ) 362 408 363 - with _CATCHUP_STATE_LOCK: 409 + state_path = _state_path() 410 + with hold_lock(state_path), _CATCHUP_STATE_LOCK: 364 411 state = _read_state_from_disk() 365 412 entries = state["entries"] 366 413 key = _key(cmd_day, kind) ··· 412 459 cmd_day, 413 460 ) 414 461 415 - with _CATCHUP_STATE_LOCK: 462 + state_path = _state_path() 463 + with hold_lock(state_path), _CATCHUP_STATE_LOCK: 416 464 state = _read_state_from_disk() 417 465 entries = state["entries"] 418 466 key = _key(cmd_day, kind) ··· 503 551 504 552 def clear_day_backoff(day: str) -> None: 505 553 try: 506 - with _CATCHUP_STATE_LOCK: 554 + state_path = _state_path() 555 + with hold_lock(state_path), _CATCHUP_STATE_LOCK: 507 556 state = _read_state_from_disk() 508 557 entries = state["entries"] 509 558 entries.pop(_key(day, KIND_DAILY_CATCHUP), None) ··· 514 563 logger.warning("Failed to clear catchup backoff for %s", day, exc_info=True) 515 564 516 565 566 + def record_segment_repair_attempt(day: str, *, started_at: float) -> None: 567 + try: 568 + fingerprint = read_raw_input_fingerprint(day) 569 + state_path = _state_path() 570 + with hold_lock(state_path), _CATCHUP_STATE_LOCK: 571 + state = _read_state_from_disk() 572 + entries = state["entries"] 573 + key = _key(day, KIND_SEGMENT_REPAIR) 574 + record = _record_from_entry(entries.get(key), day, KIND_SEGMENT_REPAIR) 575 + if record.get("fingerprint") != fingerprint: 576 + record["consecutive_non_completion"] = 0 577 + record["entered_backoff_at"] = None 578 + record["notified_at"] = None 579 + record["next_retry_at"] = 0 580 + record["reason_code"] = None 581 + record["timeout_seconds"] = None 582 + record["bounded"] = None 583 + record["fingerprint"] = fingerprint 584 + record["attempts"] = int(record.get("attempts") or 0) + 1 585 + record["last_attempt_at"] = started_at 586 + record["active"] = { 587 + "ref": "segment-repair", 588 + "started_at": started_at, 589 + } 590 + entries[key] = record 591 + _prune(state) 592 + _write_state(state) 593 + except Exception: 594 + logger.warning( 595 + "Failed to record segment-repair attempt for %s", day, exc_info=True 596 + ) 597 + 598 + 599 + def record_segment_repair_outcome( 600 + day: str, 601 + *, 602 + success: bool, 603 + timed_out: bool, 604 + timeout_seconds: float | None, 605 + ended_at: float, 606 + ) -> None: 607 + try: 608 + state_path = _state_path() 609 + with hold_lock(state_path), _CATCHUP_STATE_LOCK: 610 + state = _read_state_from_disk() 611 + entries = state["entries"] 612 + key = _key(day, KIND_SEGMENT_REPAIR) 613 + if success: 614 + entries.pop(key, None) 615 + _prune(state) 616 + _write_state(state) 617 + return 618 + 619 + record = _record_from_entry(entries.get(key), day, KIND_SEGMENT_REPAIR) 620 + record["active"] = None 621 + record["last_outcome"] = "timeout" if timed_out else "error" 622 + record["reason_code"] = ( 623 + SEGMENT_REPAIR_TIMEOUT_REASON 624 + if timed_out 625 + else SEGMENT_REPAIR_FAILED_REASON 626 + ) 627 + record["timeout_seconds"] = timeout_seconds if timed_out else None 628 + record["bounded"] = bool(timed_out) 629 + consecutive = int(record.get("consecutive_non_completion") or 0) + 1 630 + record["consecutive_non_completion"] = consecutive 631 + record["next_retry_at"] = ended_at + _backoff_delay(consecutive) 632 + if ( 633 + consecutive >= STUCK_THRESHOLD 634 + and record.get("entered_backoff_at") is None 635 + ): 636 + record["entered_backoff_at"] = ended_at 637 + record["notified_at"] = ended_at 638 + entries[key] = record 639 + _prune(state) 640 + _write_state(state) 641 + except Exception: 642 + logger.warning( 643 + "Failed to record segment-repair outcome for %s", day, exc_info=True 644 + ) 645 + 646 + 517 647 def reconcile_interrupted_attempts() -> list[BackoffTransition]: 518 648 try: 519 - with _CATCHUP_STATE_LOCK: 649 + state_path = _state_path() 650 + with hold_lock(state_path), _CATCHUP_STATE_LOCK: 520 651 state = _read_state_from_disk() 521 652 entries = state["entries"] 522 653 transitions: list[BackoffTransition] = [] ··· 529 660 continue 530 661 kind = record.get("command_kind") 531 662 day = record.get("day") 532 - if not isinstance(day, str) or kind not in RECORDED_KINDS: 663 + if not isinstance(day, str) or kind not in RECONCILABLE_KINDS: 533 664 record["active"] = None 534 665 changed = True 535 666 continue ··· 546 677 record["active"] = None 547 678 record["last_outcome"] = "interrupted" 548 679 changed = True 549 - if kind == KIND_DAILY_CATCHUP: 680 + if kind in (KIND_DAILY_CATCHUP, KIND_SEGMENT_REPAIR): 550 681 consecutive = int(record.get("consecutive_non_completion") or 0) + 1 551 682 next_retry_at = now + _backoff_delay(consecutive) 552 683 record["consecutive_non_completion"] = consecutive ··· 557 688 ): 558 689 record["entered_backoff_at"] = now 559 690 record["notified_at"] = now 560 - transitions.append( 561 - BackoffTransition( 562 - day=day, 563 - command_kind=kind, 564 - attempts=int(record.get("attempts") or 0), 565 - consecutive_non_completion=consecutive, 566 - last_outcome="interrupted", 567 - next_retry_at=next_retry_at, 568 - entered_backoff_at=now, 691 + if kind == KIND_DAILY_CATCHUP: 692 + transitions.append( 693 + BackoffTransition( 694 + day=day, 695 + command_kind=kind, 696 + attempts=int(record.get("attempts") or 0), 697 + consecutive_non_completion=consecutive, 698 + last_outcome="interrupted", 699 + next_retry_at=next_retry_at, 700 + entered_backoff_at=now, 701 + ) 569 702 ) 570 - ) 571 703 entries[key] = record 572 704 573 705 if changed:
+11
solstone/think/journal_stats.py
··· 85 85 ) 86 86 data["backoff_last_outcome"] = day.backoff_last_outcome 87 87 data["backoff_next_retry_at"] = day.backoff_next_retry_at 88 + if day.segment_repair_status is not None: 89 + data["segment_repair_status"] = day.segment_repair_status 90 + data["segment_repair_attempts"] = day.segment_repair_attempts 91 + data["segment_repair_consecutive_non_completion"] = ( 92 + day.segment_repair_consecutive_non_completion 93 + ) 94 + data["segment_repair_last_outcome"] = day.segment_repair_last_outcome 95 + data["segment_repair_next_retry_at"] = day.segment_repair_next_retry_at 96 + data["segment_repair_reason_code"] = day.segment_repair_reason_code 97 + data["segment_repair_timeout_seconds"] = day.segment_repair_timeout_seconds 98 + data["segment_repair_bounded"] = day.segment_repair_bounded 88 99 return data 89 100 90 101
+94 -3
solstone/think/pipeline_health.py
··· 12 12 from datetime import datetime, timedelta 13 13 from pathlib import Path 14 14 15 - from solstone.think.catchup_state import read_backoff_summary 15 + from solstone.think.catchup_state import ( 16 + read_backoff_summary, 17 + read_segment_repair_summary, 18 + ) 16 19 from solstone.think.cluster import cluster_segments 17 20 from solstone.think.cogitate_policy import DETERMINISTIC_FAILURE_REASON_CODES 18 21 from solstone.think.utils import ( ··· 48 51 REASON_CORRUPT_RAW = "corrupt_raw" 49 52 REASON_FAILING_STEP = "failing_step" 50 53 REASON_CATCHUP_BACKOFF = "catchup_backoff" 54 + REASON_SEGMENT_REPAIR_DEGRADED = "segment_repair_degraded" 55 + REASON_SEGMENT_REPAIR_STUCK = "segment_repair_stuck" 56 + REASON_SEGMENT_REPAIR_UNKNOWN = "segment_repair_unknown" 51 57 52 58 BACKLOG_STATE_COMPLETE = "complete" 53 59 BACKLOG_STATE_PENDING = "pending" ··· 186 192 backoff_consecutive_non_completion: int = 0 187 193 backoff_last_outcome: str | None = None 188 194 backoff_next_retry_at: float | None = None 195 + segment_repair_status: str | None = None 196 + segment_repair_attempts: int = 0 197 + segment_repair_consecutive_non_completion: int = 0 198 + segment_repair_last_outcome: str | None = None 199 + segment_repair_next_retry_at: float | None = None 200 + segment_repair_reason_code: str | None = None 201 + segment_repair_timeout_seconds: int | None = None 202 + segment_repair_bounded: bool | None = None 189 203 190 204 191 205 @dataclass(frozen=True) ··· 1146 1160 ) 1147 1161 1148 1162 1163 + _SEGMENT_REPAIR_STATE = { 1164 + "degraded": (BACKLOG_STATE_PENDING, REASON_SEGMENT_REPAIR_DEGRADED), 1165 + "stuck": (BACKLOG_STATE_STUCK, REASON_SEGMENT_REPAIR_STUCK), 1166 + "unknown": (BACKLOG_STATE_UNKNOWN, REASON_SEGMENT_REPAIR_UNKNOWN), 1167 + } 1168 + _STATE_SEVERITY = { 1169 + BACKLOG_STATE_COMPLETE: 0, 1170 + BACKLOG_STATE_PENDING: 1, 1171 + BACKLOG_STATE_STUCK: 2, 1172 + BACKLOG_STATE_UNKNOWN: 3, 1173 + } 1174 + 1175 + 1176 + def _segment_repair_fields(repair: dict | None) -> dict: 1177 + if not repair: 1178 + return {} 1179 + return { 1180 + "segment_repair_status": repair["status"], 1181 + "segment_repair_attempts": int(repair.get("attempts") or 0), 1182 + "segment_repair_consecutive_non_completion": int( 1183 + repair.get("consecutive_non_completion") or 0 1184 + ), 1185 + "segment_repair_last_outcome": repair.get("last_outcome") or None, 1186 + "segment_repair_next_retry_at": repair.get("next_retry_at"), 1187 + "segment_repair_reason_code": repair.get("repair_reason_code"), 1188 + "segment_repair_timeout_seconds": repair.get("timeout_seconds"), 1189 + "segment_repair_bounded": repair.get("bounded"), 1190 + } 1191 + 1192 + 1193 + def _escalate_for_repair(state, reason, reason_code, error, day, repair): 1194 + if not repair: 1195 + return state, reason, reason_code, error 1196 + sr_state, sr_reason = _SEGMENT_REPAIR_STATE[repair["status"]] 1197 + if _STATE_SEVERITY[sr_state] > _STATE_SEVERITY[state]: 1198 + state = sr_state 1199 + if reason_code is None: 1200 + reason = sr_reason 1201 + reason_code = sr_reason 1202 + if repair["status"] == "unknown" and error is None: 1203 + error = BacklogError( 1204 + day=day, 1205 + stage="segment_repair", 1206 + message="segment-repair state unreadable", 1207 + ) 1208 + return state, reason, reason_code, error 1209 + 1210 + 1211 + def _backlog_day_for_complete(day: str, repair: dict | None) -> BacklogDay: 1212 + if not repair: 1213 + return _complete_backlog_day(day) 1214 + state, reason, reason_code, error = _escalate_for_repair( 1215 + BACKLOG_STATE_COMPLETE, None, None, None, day, repair 1216 + ) 1217 + return BacklogDay( 1218 + day=day, 1219 + state=state, 1220 + segments=0, 1221 + units=0, 1222 + not_sensed=0, 1223 + why=(), 1224 + reason=reason, 1225 + reason_code=reason_code, 1226 + provider=None, 1227 + model=None, 1228 + error=error, 1229 + **_segment_repair_fields(repair), 1230 + ) 1231 + 1232 + 1149 1233 def read_backlog_view(window: int = BACKLOG_DEFAULT_WINDOW) -> BacklogView: 1150 1234 """Return a bounded cross-day backlog view.""" 1151 1235 backlog_days: list[BacklogDay] = [] 1152 1236 errors: list[BacklogError] = [] 1153 1237 1154 1238 for day in sorted(day_dirs().keys(), reverse=True)[:window]: 1239 + repair = read_segment_repair_summary(day) 1155 1240 if day_is_complete(day): 1156 - backlog_days.append(_complete_backlog_day(day)) 1241 + backlog_days.append(_backlog_day_for_complete(day, repair)) 1157 1242 continue 1158 1243 1159 1244 try: ··· 1237 1322 state = BACKLOG_STATE_PENDING 1238 1323 else: 1239 1324 state = BACKLOG_STATE_COMPLETE 1325 + state, reason, reason_code, sr_error = _escalate_for_repair( 1326 + state, reason, reason_code, None, day, repair 1327 + ) 1240 1328 1241 1329 backlog_days.append( 1242 1330 BacklogDay( ··· 1250 1338 reason_code=reason_code, 1251 1339 provider=representative.provider if representative else None, 1252 1340 model=representative.model if representative else None, 1253 - error=None, 1341 + error=sr_error, 1254 1342 backoff_stuck=bool(backoff), 1255 1343 backoff_attempts=backoff["attempts"] if backoff else 0, 1256 1344 backoff_consecutive_non_completion=( ··· 1258 1346 ), 1259 1347 backoff_last_outcome=backoff["last_outcome"] if backoff else None, 1260 1348 backoff_next_retry_at=backoff["next_retry_at"] if backoff else None, 1349 + **_segment_repair_fields(repair), 1261 1350 ) 1262 1351 ) 1263 1352 ··· 1275 1364 stuck_days=stuck_days, 1276 1365 oldest_pending_day=min(outstanding) if outstanding else None, 1277 1366 errors=tuple(errors), 1367 + degraded=bool(errors) 1368 + or any(day.state == BACKLOG_STATE_UNKNOWN for day in backlog_days), 1278 1369 ) 1279 1370 1280 1371
+4 -1
solstone/think/supervisor.py
··· 33 33 from solstone.think.callosum import CallosumConnection, CallosumServer 34 34 from solstone.think.catchup_state import ( 35 35 KIND_DAILY_CATCHUP, 36 + KIND_SEGMENT_REPAIR, 36 37 STUCK_THRESHOLD, 37 38 day_eligible_to_drain, 38 39 reconcile_interrupted_attempts, ··· 2222 2223 2223 2224 def _eligible(day: str) -> bool: 2224 2225 try: 2225 - return day_eligible_to_drain(day, KIND_DAILY_CATCHUP) 2226 + return day_eligible_to_drain( 2227 + day, KIND_DAILY_CATCHUP 2228 + ) and day_eligible_to_drain(day, KIND_SEGMENT_REPAIR) 2226 2229 except Exception: 2227 2230 logging.warning( 2228 2231 "Catchup eligibility check failed for %s; treating as eligible",
+12
solstone/think/thinking.py
··· 29 29 ) 30 30 from solstone.think.activity_state_machine import ActivityStateMachine 31 31 from solstone.think.callosum import CallosumConnection 32 + from solstone.think.catchup_state import ( 33 + record_segment_repair_attempt, 34 + record_segment_repair_outcome, 35 + ) 32 36 from solstone.think.change_detection import detect_segment_change, resolve_predecessor 33 37 from solstone.think.cluster import cluster_segments 34 38 from solstone.think.cogitate_policy import DETERMINISTIC_FAILURE_THRESHOLD ··· 3758 3762 day_log(day, f"starting: {' '.join(cmd)}") 3759 3763 _jsonl_log("phase.start", mode=_run_mode, day=day, phase="segment_think") 3760 3764 _phase_start = time.time() 3765 + record_segment_repair_attempt(day, started_at=_phase_start) 3761 3766 phase_ok, phase_timed_out = run_bounded_phase( 3762 3767 cmd, day, DEFAULT_TASK_MAX_RUNTIME 3763 3768 ) ··· 3775 3780 bounded=True, 3776 3781 ) 3777 3782 _jsonl_log("phase.complete", **phase_complete) 3783 + record_segment_repair_outcome( 3784 + day, 3785 + success=phase_ok, 3786 + timed_out=phase_timed_out, 3787 + timeout_seconds=DEFAULT_TASK_MAX_RUNTIME, 3788 + ended_at=time.time(), 3789 + ) 3778 3790 if not phase_ok: 3779 3791 if phase_timed_out: 3780 3792 logging.warning(
+42
tests/test_backlog_view.py
··· 160 160 ) 161 161 162 162 163 + def test_verdict_degraded_backlog_does_not_claim_caught_up(): 164 + assert ( 165 + verdict({"pending_days": 0, "stuck_days": 0, "degraded": True}) 166 + != backlog_copy.BACKLOG_VERDICT_CAUGHT_UP 167 + ) 168 + 169 + 170 + def test_stuck_rows_includes_segment_repair_stuck_day(): 171 + rows = stuck_rows( 172 + { 173 + "days": [ 174 + { 175 + "day": "20260601", 176 + "state": "stuck", 177 + "segments": 0, 178 + "units": 0, 179 + "reason": "segment_repair_stuck", 180 + "reason_code": "segment_repair_stuck", 181 + "segment_repair_status": "stuck", 182 + "segment_repair_attempts": 3, 183 + "segment_repair_consecutive_non_completion": 3, 184 + "segment_repair_last_outcome": "timeout", 185 + "segment_repair_next_retry_at": 1600.0, 186 + "segment_repair_reason_code": "wall_clock_exceeded", 187 + "segment_repair_timeout_seconds": 300, 188 + "segment_repair_bounded": True, 189 + } 190 + ], 191 + "errors": [], 192 + } 193 + ) 194 + 195 + assert rows == [ 196 + { 197 + "day": "20260601", 198 + "reason": backlog_copy.BACKLOG_REASON_FAILING_STEP, 199 + "depth": None, 200 + "reason_code": "segment_repair_stuck", 201 + } 202 + ] 203 + 204 + 163 205 def test_verdict_mixed_copy_uses_independent_arms(): 164 206 assert ( 165 207 verdict({"pending_days": 1, "stuck_days": 1})
+379
tests/test_catchup_state.py
··· 77 77 entered_backoff_at: float | None = None, 78 78 active: dict | None = None, 79 79 fingerprint: str | None = None, 80 + reason_code: str | None = None, 81 + timeout_seconds: float | None = None, 82 + bounded: bool | None = None, 80 83 ) -> dict: 81 84 return { 82 85 "day": day, ··· 90 93 "notified_at": entered_backoff_at, 91 94 "fingerprint": fingerprint, 92 95 "active": active, 96 + "reason_code": reason_code, 97 + "timeout_seconds": timeout_seconds, 98 + "bounded": bounded, 93 99 } 94 100 95 101 ··· 352 358 assert catchup_state.day_eligible_to_drain(DAY, catchup_state.KIND_DAILY_CATCHUP) 353 359 354 360 361 + def test_record_segment_repair_attempt_sets_active_and_resets_on_fingerprint_change( 362 + journal, 363 + ): 364 + raw = _segment(journal).joinpath("audio.jsonl") 365 + raw.write_text("one\n", encoding="utf-8") 366 + key = f"{DAY}:{catchup_state.KIND_SEGMENT_REPAIR}" 367 + 368 + catchup_state.record_segment_repair_attempt(DAY, started_at=10) 369 + 370 + record = catchup_state.read_day_record(DAY, catchup_state.KIND_SEGMENT_REPAIR) 371 + fingerprint = catchup_state.read_raw_input_fingerprint(DAY) 372 + assert record["attempts"] == 1 373 + assert record["fingerprint"] == fingerprint 374 + assert record["active"] == {"ref": "segment-repair", "started_at": 10} 375 + 376 + record.update( 377 + consecutive_non_completion=2, 378 + entered_backoff_at=100, 379 + notified_at=100, 380 + next_retry_at=999, 381 + reason_code=catchup_state.SEGMENT_REPAIR_TIMEOUT_REASON, 382 + timeout_seconds=600, 383 + bounded=True, 384 + active=None, 385 + ) 386 + _write_state(journal, {key: record}) 387 + raw.write_text("two\n", encoding="utf-8") 388 + 389 + catchup_state.record_segment_repair_attempt(DAY, started_at=20) 390 + 391 + reset_record = catchup_state.read_day_record(DAY, catchup_state.KIND_SEGMENT_REPAIR) 392 + assert reset_record["attempts"] == 2 393 + assert reset_record["fingerprint"] != fingerprint 394 + assert reset_record["consecutive_non_completion"] == 0 395 + assert reset_record["entered_backoff_at"] is None 396 + assert reset_record["notified_at"] is None 397 + assert reset_record["next_retry_at"] == 0 398 + assert reset_record["reason_code"] is None 399 + assert reset_record["timeout_seconds"] is None 400 + assert reset_record["bounded"] is None 401 + assert reset_record["active"] == {"ref": "segment-repair", "started_at": 20} 402 + 403 + 404 + def test_record_segment_repair_outcome_records_failure_metadata_and_backoff(journal): 405 + _segment(journal).joinpath("audio.jsonl").write_text("one\n", encoding="utf-8") 406 + 407 + catchup_state.record_segment_repair_attempt(DAY, started_at=10) 408 + catchup_state.record_segment_repair_outcome( 409 + DAY, 410 + success=False, 411 + timed_out=True, 412 + timeout_seconds=300, 413 + ended_at=1000, 414 + ) 415 + 416 + timeout_record = catchup_state.read_day_record( 417 + DAY, catchup_state.KIND_SEGMENT_REPAIR 418 + ) 419 + assert timeout_record["active"] is None 420 + assert timeout_record["last_outcome"] == "timeout" 421 + assert timeout_record["reason_code"] == catchup_state.SEGMENT_REPAIR_TIMEOUT_REASON 422 + assert timeout_record["timeout_seconds"] == 300 423 + assert timeout_record["bounded"] is True 424 + assert timeout_record["consecutive_non_completion"] == 1 425 + assert timeout_record["next_retry_at"] == 1600 426 + assert timeout_record["entered_backoff_at"] is None 427 + 428 + catchup_state.record_segment_repair_attempt(DAY, started_at=20) 429 + catchup_state.record_segment_repair_outcome( 430 + DAY, 431 + success=False, 432 + timed_out=False, 433 + timeout_seconds=300, 434 + ended_at=2000, 435 + ) 436 + error_record = catchup_state.read_day_record(DAY, catchup_state.KIND_SEGMENT_REPAIR) 437 + assert error_record["last_outcome"] == "error" 438 + assert error_record["reason_code"] == catchup_state.SEGMENT_REPAIR_FAILED_REASON 439 + assert error_record["timeout_seconds"] is None 440 + assert error_record["bounded"] is False 441 + assert error_record["consecutive_non_completion"] == 2 442 + 443 + catchup_state.record_segment_repair_attempt(DAY, started_at=30) 444 + catchup_state.record_segment_repair_outcome( 445 + DAY, 446 + success=False, 447 + timed_out=False, 448 + timeout_seconds=300, 449 + ended_at=3000, 450 + ) 451 + stuck_record = catchup_state.read_day_record(DAY, catchup_state.KIND_SEGMENT_REPAIR) 452 + assert stuck_record["consecutive_non_completion"] == catchup_state.STUCK_THRESHOLD 453 + assert stuck_record["entered_backoff_at"] == 3000 454 + assert stuck_record["notified_at"] == 3000 455 + 456 + 457 + def test_record_segment_repair_outcome_success_pops_record(journal): 458 + _segment(journal).joinpath("audio.jsonl").write_text("one\n", encoding="utf-8") 459 + catchup_state.record_segment_repair_attempt(DAY, started_at=10) 460 + catchup_state.record_segment_repair_outcome( 461 + DAY, 462 + success=False, 463 + timed_out=False, 464 + timeout_seconds=None, 465 + ended_at=1000, 466 + ) 467 + assert catchup_state.read_segment_repair_summary(DAY) is not None 468 + 469 + catchup_state.record_segment_repair_outcome( 470 + DAY, 471 + success=True, 472 + timed_out=False, 473 + timeout_seconds=None, 474 + ended_at=2000, 475 + ) 476 + 477 + assert catchup_state.read_day_record(DAY, catchup_state.KIND_SEGMENT_REPAIR) is None 478 + assert catchup_state.read_segment_repair_summary(DAY) is None 479 + 480 + 481 + def test_read_segment_repair_summary_states(journal): 482 + raw = _segment(journal).joinpath("audio.jsonl") 483 + raw.write_text("one\n", encoding="utf-8") 484 + fingerprint = catchup_state.read_raw_input_fingerprint(DAY) 485 + key = f"{DAY}:{catchup_state.KIND_SEGMENT_REPAIR}" 486 + 487 + assert catchup_state.read_segment_repair_summary(DAY) is None 488 + 489 + _write_state( 490 + journal, 491 + {key: _record(DAY, catchup_state.KIND_SEGMENT_REPAIR, fingerprint=fingerprint)}, 492 + ) 493 + assert catchup_state.read_segment_repair_summary(DAY) is None 494 + 495 + _write_state( 496 + journal, 497 + { 498 + key: _record( 499 + DAY, 500 + catchup_state.KIND_SEGMENT_REPAIR, 501 + consecutive=1, 502 + fingerprint="stale", 503 + ) 504 + }, 505 + ) 506 + assert catchup_state.read_segment_repair_summary(DAY) is None 507 + 508 + _write_state( 509 + journal, 510 + { 511 + key: _record( 512 + DAY, 513 + catchup_state.KIND_SEGMENT_REPAIR, 514 + attempts=2, 515 + consecutive=1, 516 + last_outcome="error", 517 + next_retry_at=999, 518 + fingerprint=fingerprint, 519 + reason_code=catchup_state.SEGMENT_REPAIR_FAILED_REASON, 520 + bounded=False, 521 + ) 522 + }, 523 + ) 524 + degraded = catchup_state.read_segment_repair_summary(DAY) 525 + assert degraded == { 526 + "status": "degraded", 527 + "attempts": 2, 528 + "consecutive_non_completion": 1, 529 + "last_outcome": "error", 530 + "next_retry_at": 999, 531 + "repair_reason_code": catchup_state.SEGMENT_REPAIR_FAILED_REASON, 532 + "timeout_seconds": None, 533 + "bounded": False, 534 + } 535 + 536 + _write_state( 537 + journal, 538 + { 539 + key: _record( 540 + DAY, 541 + catchup_state.KIND_SEGMENT_REPAIR, 542 + attempts=3, 543 + consecutive=3, 544 + last_outcome="timeout", 545 + next_retry_at=1200, 546 + entered_backoff_at=600, 547 + fingerprint=fingerprint, 548 + reason_code=catchup_state.SEGMENT_REPAIR_TIMEOUT_REASON, 549 + timeout_seconds=300, 550 + bounded=True, 551 + ) 552 + }, 553 + ) 554 + stuck = catchup_state.read_segment_repair_summary(DAY) 555 + assert stuck["status"] == "stuck" 556 + assert stuck["repair_reason_code"] == catchup_state.SEGMENT_REPAIR_TIMEOUT_REASON 557 + assert stuck["timeout_seconds"] == 300 558 + assert stuck["bounded"] is True 559 + 560 + _state_file(journal).write_text("{not json", encoding="utf-8") 561 + assert catchup_state.read_segment_repair_summary(DAY) == {"status": "unknown"} 562 + 563 + 564 + def test_day_eligible_to_drain_segment_repair(journal, monkeypatch): 565 + raw = _segment(journal).joinpath("audio.jsonl") 566 + raw.write_text("one\n", encoding="utf-8") 567 + fingerprint = catchup_state.read_raw_input_fingerprint(DAY) 568 + key = f"{DAY}:{catchup_state.KIND_SEGMENT_REPAIR}" 569 + monkeypatch.setattr(catchup_state.time, "time", lambda: 100) 570 + 571 + assert catchup_state.day_eligible_to_drain(DAY, catchup_state.KIND_SEGMENT_REPAIR) 572 + 573 + _write_state( 574 + journal, 575 + { 576 + key: _record( 577 + DAY, 578 + catchup_state.KIND_SEGMENT_REPAIR, 579 + next_retry_at=1000, 580 + fingerprint=fingerprint, 581 + active={"ref": "segment-repair", "started_at": 1}, 582 + ) 583 + }, 584 + ) 585 + assert not catchup_state.day_eligible_to_drain( 586 + DAY, catchup_state.KIND_SEGMENT_REPAIR 587 + ) 588 + 589 + _write_state( 590 + journal, 591 + { 592 + key: _record( 593 + DAY, 594 + catchup_state.KIND_SEGMENT_REPAIR, 595 + next_retry_at=1000, 596 + fingerprint=fingerprint, 597 + ) 598 + }, 599 + ) 600 + assert not catchup_state.day_eligible_to_drain( 601 + DAY, catchup_state.KIND_SEGMENT_REPAIR 602 + ) 603 + 604 + raw.write_text("two\n", encoding="utf-8") 605 + assert catchup_state.day_eligible_to_drain(DAY, catchup_state.KIND_SEGMENT_REPAIR) 606 + 607 + 355 608 def test_reconcile_interrupted_attempts(journal, monkeypatch): 356 609 incomplete = "20260101" 357 610 complete = "20260102" ··· 403 656 assert entries[incomplete_key]["consecutive_non_completion"] == 3 404 657 assert entries[incomplete_key]["active"] is None 405 658 assert complete_key not in entries 659 + 660 + 661 + def test_reconcile_interrupted_segment_repair_on_complete_day_is_not_popped( 662 + journal, monkeypatch 663 + ): 664 + incomplete_daily = "20260101" 665 + complete_day = "20260102" 666 + _segment(journal, day=incomplete_daily).joinpath("audio.jsonl").write_text( 667 + "{}\n", encoding="utf-8" 668 + ) 669 + _touch_marker(journal, incomplete_daily, "stream.updated", 100) 670 + _segment(journal, day=complete_day).joinpath("audio.jsonl").write_text( 671 + "{}\n", encoding="utf-8" 672 + ) 673 + _touch_marker(journal, complete_day, "stream.updated", 100) 674 + _touch_marker(journal, complete_day, "daily.updated", 200) 675 + daily_incomplete_key = f"{incomplete_daily}:{catchup_state.KIND_DAILY_CATCHUP}" 676 + daily_complete_key = f"{complete_day}:{catchup_state.KIND_DAILY_CATCHUP}" 677 + repair_key = f"{complete_day}:{catchup_state.KIND_SEGMENT_REPAIR}" 678 + _write_state( 679 + journal, 680 + { 681 + daily_incomplete_key: _record( 682 + incomplete_daily, 683 + catchup_state.KIND_DAILY_CATCHUP, 684 + attempts=3, 685 + consecutive=2, 686 + active={ 687 + "ref": "daily-incomplete", 688 + "started_at": 1, 689 + "marker_mtime_at_start": None, 690 + }, 691 + ), 692 + daily_complete_key: _record( 693 + complete_day, 694 + catchup_state.KIND_DAILY_CATCHUP, 695 + attempts=1, 696 + active={ 697 + "ref": "daily-complete", 698 + "started_at": 1, 699 + "marker_mtime_at_start": None, 700 + }, 701 + ), 702 + repair_key: _record( 703 + complete_day, 704 + catchup_state.KIND_SEGMENT_REPAIR, 705 + attempts=2, 706 + consecutive=2, 707 + active={"ref": "segment-repair", "started_at": 1}, 708 + ), 709 + }, 710 + ) 711 + monkeypatch.setattr(catchup_state.time, "time", lambda: 1000) 712 + 713 + transitions = catchup_state.reconcile_interrupted_attempts() 714 + 715 + entries = _read_entries(journal) 716 + assert daily_complete_key not in entries 717 + assert entries[repair_key]["last_outcome"] == "interrupted" 718 + assert entries[repair_key]["consecutive_non_completion"] == 3 719 + assert entries[repair_key]["entered_backoff_at"] == 1000 720 + assert entries[repair_key]["active"] is None 721 + assert len(transitions) == 1 722 + assert transitions[0].day == incomplete_daily 723 + assert transitions[0].command_kind == catchup_state.KIND_DAILY_CATCHUP 724 + 725 + 726 + def test_daily_completion_and_clear_day_backoff_leave_segment_repair_record(journal): 727 + raw = _segment(journal).joinpath("audio.jsonl") 728 + raw.write_text("{}\n", encoding="utf-8") 729 + _touch_marker(journal, DAY, "stream.updated", 100) 730 + _touch_marker(journal, DAY, "daily.updated", 100) 731 + fingerprint = catchup_state.read_raw_input_fingerprint(DAY) 732 + daily_key = f"{DAY}:{catchup_state.KIND_DAILY_CATCHUP}" 733 + scratch_key = f"{DAY}:{catchup_state.KIND_DAILY_FROM_SCRATCH}" 734 + repair_key = f"{DAY}:{catchup_state.KIND_SEGMENT_REPAIR}" 735 + repair_record = _record( 736 + DAY, 737 + catchup_state.KIND_SEGMENT_REPAIR, 738 + consecutive=1, 739 + last_outcome="error", 740 + next_retry_at=999, 741 + fingerprint=fingerprint, 742 + reason_code=catchup_state.SEGMENT_REPAIR_FAILED_REASON, 743 + ) 744 + _write_state( 745 + journal, 746 + { 747 + daily_key: _record( 748 + DAY, 749 + catchup_state.KIND_DAILY_CATCHUP, 750 + active={ 751 + "ref": "daily", 752 + "started_at": 1, 753 + "marker_mtime_at_start": 100, 754 + }, 755 + ), 756 + repair_key: repair_record, 757 + }, 758 + ) 759 + _touch_marker(journal, DAY, "daily.updated", 200) 760 + 761 + result = catchup_state.record_outcome( 762 + CMD_DAILY, DAY, "daily", exit_status="ok", ended_at=300 763 + ) 764 + 765 + entries = _read_entries(journal) 766 + assert result.completed is True 767 + assert daily_key not in entries 768 + assert entries[repair_key]["command_kind"] == catchup_state.KIND_SEGMENT_REPAIR 769 + 770 + _write_state( 771 + journal, 772 + { 773 + daily_key: _record(DAY, catchup_state.KIND_DAILY_CATCHUP), 774 + scratch_key: _record(DAY, catchup_state.KIND_DAILY_FROM_SCRATCH), 775 + repair_key: entries[repair_key], 776 + }, 777 + ) 778 + 779 + catchup_state.clear_day_backoff(DAY) 780 + 781 + entries = _read_entries(journal) 782 + assert daily_key not in entries 783 + assert scratch_key not in entries 784 + assert repair_key in entries 406 785 407 786 408 787 def test_prune_removes_old_cleared_entries_but_keeps_old_stuck(journal):
+95
tests/test_journal_caught_up.py
··· 11 11 12 12 from solstone.think import catchup_state, pipeline_health 13 13 from solstone.think.pipeline_health import ( 14 + BACKLOG_STATE_COMPLETE, 15 + BACKLOG_STATE_PENDING, 14 16 BACKLOG_STATE_UNKNOWN, 15 17 BacklogDay, 16 18 BacklogError, ··· 132 134 assert "1 day(s) stuck" in result.detail 133 135 assert "oldest outstanding 20200229" in result.detail 134 136 assert str(2 + 1) not in result.detail 137 + 138 + 139 + def test_journal_caught_up_warns_for_degraded_segment_repair_day( 140 + doctor, 141 + monkeypatch, 142 + ): 143 + day = BacklogDay( 144 + day="20200229", 145 + state=BACKLOG_STATE_PENDING, 146 + segments=0, 147 + units=0, 148 + not_sensed=0, 149 + why=(), 150 + reason="segment_repair_degraded", 151 + reason_code="segment_repair_degraded", 152 + provider=None, 153 + model=None, 154 + error=None, 155 + segment_repair_status="degraded", 156 + ) 157 + view = BacklogView( 158 + window=30, 159 + days=(day,), 160 + pending_days=1, 161 + stuck_days=0, 162 + oldest_pending_day="20200229", 163 + errors=(), 164 + ) 165 + monkeypatch.setattr(pipeline_health, "read_backlog_view", lambda: view) 166 + 167 + result = doctor.journal_caught_up_check(args(doctor)) 168 + 169 + assert result.status == "warn" 170 + assert result.status != "ok" 171 + assert "1 day(s) pending" in result.detail 172 + 173 + 174 + def test_journal_caught_up_warns_for_unknown_segment_repair_day( 175 + doctor, 176 + monkeypatch, 177 + ): 178 + error = BacklogError( 179 + day="20200229", 180 + stage="segment_repair", 181 + message="segment-repair state unreadable", 182 + ) 183 + day = BacklogDay( 184 + day="20200229", 185 + state=BACKLOG_STATE_UNKNOWN, 186 + segments=0, 187 + units=0, 188 + not_sensed=0, 189 + why=(), 190 + reason="segment_repair_unknown", 191 + reason_code="segment_repair_unknown", 192 + provider=None, 193 + model=None, 194 + error=error, 195 + segment_repair_status="unknown", 196 + ) 197 + view = BacklogView( 198 + window=30, 199 + days=(day,), 200 + pending_days=0, 201 + stuck_days=0, 202 + oldest_pending_day=None, 203 + errors=(), 204 + degraded=True, 205 + ) 206 + monkeypatch.setattr(pipeline_health, "read_backlog_view", lambda: view) 207 + 208 + result = doctor.journal_caught_up_check(args(doctor)) 209 + 210 + assert result.status == "warn" 211 + assert result.status != "ok" 212 + assert "1 day(s) unknown" in result.detail 213 + 214 + 215 + def test_journal_caught_up_ok_for_all_complete_view(doctor, monkeypatch): 216 + view = BacklogView( 217 + window=30, 218 + days=(backlog_day("20200229", BACKLOG_STATE_COMPLETE),), 219 + pending_days=0, 220 + stuck_days=0, 221 + oldest_pending_day=None, 222 + errors=(), 223 + ) 224 + monkeypatch.setattr(pipeline_health, "read_backlog_view", lambda: view) 225 + 226 + result = doctor.journal_caught_up_check(args(doctor)) 227 + 228 + assert result.status == "ok" 229 + assert result.detail == "caught up" 135 230 136 231 137 232 def test_journal_caught_up_reports_backoff_stuck_day(doctor, tmp_path, monkeypatch):
+95
tests/test_journal_stats.py
··· 115 115 assert stuck_data["backoff_next_retry_at"] == 1600.0 116 116 117 117 118 + def test_serialize_backlog_day_includes_segment_repair_fields_only_when_present(): 119 + stats_mod = importlib.import_module("solstone.think.journal_stats") 120 + health_mod = importlib.import_module("solstone.think.pipeline_health") 121 + 122 + normal = health_mod.BacklogDay( 123 + day="20240101", 124 + state=health_mod.BACKLOG_STATE_COMPLETE, 125 + segments=0, 126 + units=0, 127 + not_sensed=0, 128 + why=(), 129 + reason=None, 130 + reason_code=None, 131 + provider=None, 132 + model=None, 133 + error=None, 134 + ) 135 + repair = health_mod.BacklogDay( 136 + day="20240102", 137 + state=health_mod.BACKLOG_STATE_STUCK, 138 + segments=0, 139 + units=0, 140 + not_sensed=0, 141 + why=(), 142 + reason=health_mod.REASON_SEGMENT_REPAIR_STUCK, 143 + reason_code=health_mod.REASON_SEGMENT_REPAIR_STUCK, 144 + provider=None, 145 + model=None, 146 + error=None, 147 + segment_repair_status="stuck", 148 + segment_repair_attempts=3, 149 + segment_repair_consecutive_non_completion=3, 150 + segment_repair_last_outcome="timeout", 151 + segment_repair_next_retry_at=1600.0, 152 + segment_repair_reason_code="wall_clock_exceeded", 153 + segment_repair_timeout_seconds=300, 154 + segment_repair_bounded=True, 155 + ) 156 + 157 + normal_data = stats_mod._serialize_backlog_day(normal) 158 + repair_data = stats_mod._serialize_backlog_day(repair) 159 + 160 + assert not any(key.startswith("segment_repair_") for key in normal_data) 161 + assert repair_data["segment_repair_status"] == "stuck" 162 + assert repair_data["segment_repair_attempts"] == 3 163 + assert repair_data["segment_repair_consecutive_non_completion"] == 3 164 + assert repair_data["segment_repair_last_outcome"] == "timeout" 165 + assert repair_data["segment_repair_next_retry_at"] == 1600.0 166 + assert repair_data["segment_repair_reason_code"] == "wall_clock_exceeded" 167 + assert repair_data["segment_repair_timeout_seconds"] == 300 168 + assert repair_data["segment_repair_bounded"] is True 169 + 170 + 118 171 def test_scan_day(tmp_path, monkeypatch): 119 172 stats_mod = importlib.import_module("solstone.think.journal_stats") 120 173 journal = tmp_path ··· 518 571 "errors": [], 519 572 "degraded": False, 520 573 } 574 + 575 + 576 + def test_stats_schema_v7_accepts_segment_repair_backlog_day(): 577 + stats_mod = importlib.import_module("solstone.think.journal_stats") 578 + schema_mod = importlib.import_module("solstone.think.stats_schema") 579 + 580 + repair_day = stats_mod.BacklogDay( 581 + day="20990410", 582 + state="pending", 583 + segments=0, 584 + units=0, 585 + not_sensed=0, 586 + why=(), 587 + reason="segment_repair_degraded", 588 + reason_code="segment_repair_degraded", 589 + provider=None, 590 + model=None, 591 + error=None, 592 + segment_repair_status="degraded", 593 + segment_repair_attempts=1, 594 + segment_repair_consecutive_non_completion=1, 595 + segment_repair_last_outcome="error", 596 + segment_repair_next_retry_at=1600.0, 597 + segment_repair_reason_code="repair_failed", 598 + segment_repair_timeout_seconds=None, 599 + segment_repair_bounded=False, 600 + ) 601 + js = stats_mod.JournalStats() 602 + js.backlog_view = stats_mod.BacklogView( 603 + window=30, 604 + days=(repair_day,), 605 + pending_days=1, 606 + stuck_days=0, 607 + oldest_pending_day="20990410", 608 + errors=(), 609 + ) 610 + 611 + data = js.to_dict() 612 + 613 + assert schema_mod.SCHEMA_VERSION == 7 614 + assert schema_mod.validate(data) == [] 615 + assert data["backlog"]["days"][0]["segment_repair_status"] == "degraded" 521 616 522 617 523 618 def test_backlog_day_serialization_includes_reason():
+130
tests/test_pipeline_health.py
··· 16 16 17 17 from solstone.think import catchup_state 18 18 from solstone.think.pipeline_health import ( 19 + BACKLOG_STATE_COMPLETE, 20 + BACKLOG_STATE_PENDING, 19 21 BACKLOG_STATE_STUCK, 22 + BACKLOG_STATE_UNKNOWN, 20 23 REASON_CATCHUP_BACKOFF, 24 + REASON_SEGMENT_REPAIR_DEGRADED, 25 + REASON_SEGMENT_REPAIR_STUCK, 21 26 STUCK_FAIL_THRESHOLD, 22 27 TERMINAL_FAIL, 23 28 CompletionsSince, ··· 165 170 return path 166 171 167 172 173 + def _seed_complete_day_with_raw(journal: Path, day: str) -> str: 174 + _seed_screen_segment(journal, day, "090000_300") 175 + _touch_marker(journal, day, "stream.updated", mtime_ms=100_000) 176 + _touch_marker(journal, day, "daily.updated", mtime_ms=200_000) 177 + return catchup_state.read_raw_input_fingerprint(day) 178 + 179 + 180 + def _write_segment_repair_state( 181 + journal: Path, 182 + day: str, 183 + *, 184 + fingerprint: str, 185 + consecutive: int = 1, 186 + entered_backoff_at: float | None = None, 187 + ) -> None: 188 + state_path = journal / "health" / "catchup-state.json" 189 + state_path.parent.mkdir(parents=True, exist_ok=True) 190 + state_path.write_text( 191 + json.dumps( 192 + { 193 + "version": catchup_state.STATE_VERSION, 194 + "entries": { 195 + f"{day}:{catchup_state.KIND_SEGMENT_REPAIR}": { 196 + "day": day, 197 + "command_kind": catchup_state.KIND_SEGMENT_REPAIR, 198 + "attempts": 3, 199 + "consecutive_non_completion": consecutive, 200 + "last_attempt_at": 1000.0, 201 + "last_outcome": "timeout" 202 + if entered_backoff_at is not None 203 + else "error", 204 + "next_retry_at": 1600.0, 205 + "entered_backoff_at": entered_backoff_at, 206 + "notified_at": entered_backoff_at, 207 + "fingerprint": fingerprint, 208 + "active": None, 209 + "reason_code": "wall_clock_exceeded" 210 + if entered_backoff_at is not None 211 + else "repair_failed", 212 + "timeout_seconds": 300 213 + if entered_backoff_at is not None 214 + else None, 215 + "bounded": entered_backoff_at is not None, 216 + } 217 + }, 218 + } 219 + ), 220 + encoding="utf-8", 221 + ) 222 + 223 + 168 224 @pytest.fixture 169 225 def pipeline_journal(tmp_path, monkeypatch): 170 226 journal = tmp_path / "journal" ··· 221 277 assert backlog_day.backoff_consecutive_non_completion == 3 222 278 assert backlog_day.backoff_last_outcome == "timeout" 223 279 assert backlog_day.backoff_next_retry_at == 1600.0 280 + 281 + 282 + def test_read_backlog_view_marks_complete_day_with_degraded_segment_repair_pending( 283 + pipeline_journal, 284 + ): 285 + day = "20990302" 286 + fingerprint = _seed_complete_day_with_raw(pipeline_journal, day) 287 + _write_segment_repair_state(pipeline_journal, day, fingerprint=fingerprint) 288 + 289 + view = read_backlog_view() 290 + 291 + backlog_day = next(item for item in view.days if item.day == day) 292 + assert backlog_day.state == BACKLOG_STATE_PENDING 293 + assert backlog_day.reason_code == REASON_SEGMENT_REPAIR_DEGRADED 294 + assert backlog_day.segment_repair_status == "degraded" 295 + assert backlog_day.segment_repair_attempts == 3 296 + assert view.pending_days == 1 297 + 298 + 299 + def test_read_backlog_view_marks_complete_day_with_stuck_segment_repair_stuck( 300 + pipeline_journal, 301 + ): 302 + day = "20990303" 303 + fingerprint = _seed_complete_day_with_raw(pipeline_journal, day) 304 + _write_segment_repair_state( 305 + pipeline_journal, 306 + day, 307 + fingerprint=fingerprint, 308 + consecutive=3, 309 + entered_backoff_at=1200.0, 310 + ) 311 + 312 + view = read_backlog_view() 313 + 314 + backlog_day = next(item for item in view.days if item.day == day) 315 + assert backlog_day.state == BACKLOG_STATE_STUCK 316 + assert backlog_day.reason_code == REASON_SEGMENT_REPAIR_STUCK 317 + assert backlog_day.segment_repair_status == "stuck" 318 + assert view.stuck_days == 1 319 + 320 + 321 + def test_read_backlog_view_marks_malformed_segment_repair_state_unknown( 322 + pipeline_journal, 323 + ): 324 + day = "20990304" 325 + _seed_complete_day_with_raw(pipeline_journal, day) 326 + state_path = pipeline_journal / "health" / "catchup-state.json" 327 + state_path.parent.mkdir(parents=True, exist_ok=True) 328 + state_path.write_bytes(b"{not json") 329 + 330 + view = read_backlog_view() 331 + 332 + backlog_day = next(item for item in view.days if item.day == day) 333 + assert backlog_day.state == BACKLOG_STATE_UNKNOWN 334 + assert backlog_day.error is not None 335 + assert backlog_day.error.stage == "segment_repair" 336 + assert view.degraded is True 337 + assert view.errors == () 338 + 339 + 340 + def test_read_backlog_view_suppresses_stale_segment_repair_fingerprint( 341 + pipeline_journal, 342 + ): 343 + day = "20990305" 344 + _seed_complete_day_with_raw(pipeline_journal, day) 345 + _write_segment_repair_state(pipeline_journal, day, fingerprint="stale") 346 + 347 + view = read_backlog_view() 348 + 349 + backlog_day = next(item for item in view.days if item.day == day) 350 + assert backlog_day.state == BACKLOG_STATE_COMPLETE 351 + assert backlog_day.segment_repair_status is None 352 + assert view.pending_days == 0 353 + assert view.stuck_days == 0 224 354 225 355 226 356 def test_read_completed_units_terminal_presence(pipeline_journal):
+23
tests/test_supervisor_schedule.py
··· 166 166 assert submit_mock.call_args_list == _daily_think_calls(submitted) 167 167 168 168 169 + def test_run_catchup_drain_skips_segment_repair_poison_day( 170 + mock_callosum, monkeypatch, submit_mock 171 + ): 172 + monkeypatch.setattr( 173 + mod, 174 + "updated_days", 175 + lambda **kwargs: ["20250101", "20250102", "20250103"], 176 + ) 177 + calls = [] 178 + 179 + def eligible(day, kind): 180 + calls.append((day, kind)) 181 + return (day, kind) != ("20250102", mod.KIND_SEGMENT_REPAIR) 182 + 183 + monkeypatch.setattr(mod, "day_eligible_to_drain", eligible) 184 + 185 + submitted = mod.run_catchup_drain() 186 + 187 + assert submitted == ["20250101", "20250103"] 188 + assert submit_mock.call_args_list == _daily_think_calls(submitted) 189 + assert ("20250102", mod.KIND_SEGMENT_REPAIR) in calls 190 + 191 + 169 192 def test_force_day_blocked_when_not_eligible(mock_callosum, monkeypatch, submit_mock): 170 193 monkeypatch.setattr( 171 194 mod,