solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Regression witnesses for the daily segment-think pre-phase."""
5
6import asyncio
7import importlib
8import json
9import logging
10import subprocess
11import threading
12import time
13from pathlib import Path
14from unittest.mock import Mock
15
16import pytest
17
18from solstone.think.providers import fanout_policy
19from tests.test_think_segment import _segment_configs, _write_sense_output
20
21DAY = "20240115"
22STREAM = "default"
23ACTIVE_SEGMENT = "090000_300"
24IDLE_SEGMENT = "090500_300"
25FACET = "work"
26
27
28class NullCallosumConnection:
29 def __init__(self, *args, **kwargs) -> None:
30 pass
31
32 def start(self, callback=None) -> None:
33 return None
34
35 def emit(self, *args, **kwargs) -> None:
36 return None
37
38 def stop(self) -> None:
39 return None
40
41
42def _read_jsonl(path: Path) -> list[dict]:
43 return [
44 json.loads(line)
45 for line in path.read_text(encoding="utf-8").splitlines()
46 if line.strip()
47 ]
48
49
50def _write_jsonl(path: Path, events: list[dict]) -> None:
51 path.parent.mkdir(parents=True, exist_ok=True)
52 with path.open("w", encoding="utf-8") as handle:
53 for event in events:
54 handle.write(json.dumps(event) + "\n")
55
56
57def _event_name(event: dict) -> str:
58 return str(event.get("event") or event.get("type") or "")
59
60
61def _active_sense() -> dict:
62 return {
63 "density": "active",
64 "content_type": "coding",
65 "activity_summary": "Writing tests",
66 "entities": [],
67 "facets": [{"facet": FACET, "level": "high"}],
68 "recommend": {},
69 }
70
71
72def _idle_sense() -> dict:
73 return {
74 "density": "idle",
75 "content_type": "idle",
76 "activity_summary": "",
77 "entities": [],
78 "facets": [],
79 "recommend": {},
80 }
81
82
83def _seed_segment(
84 journal: Path,
85 day: str,
86 segment: str,
87 sense_json: dict | None = None,
88) -> Path:
89 segment_dir = journal / "chronicle" / day / STREAM / segment
90 (segment_dir / "talents").mkdir(parents=True, exist_ok=True)
91 (segment_dir / "screen.jsonl").write_text(
92 json.dumps({"timestamp": f"{day}T09:00:00"}) + "\n",
93 encoding="utf-8",
94 )
95 _write_sense_output(segment_dir, sense_json or _active_sense())
96 return segment_dir
97
98
99def _complete_segment_progress():
100 from solstone.think.pipeline_health import SEGMENT_FLOOR_TALENTS, SegmentProgress
101
102 floor = frozenset(SEGMENT_FLOOR_TALENTS)
103 return SegmentProgress(
104 sensed=True,
105 density="active",
106 change_class=None,
107 dispatched=floor,
108 completed=floor,
109 unconfigured=frozenset(),
110 capped=frozenset(),
111 )
112
113
114def _append_segment_terminal(
115 journal: Path,
116 *,
117 name: str,
118 event: str = "talent.complete",
119 ts: int = 100,
120 stream: str = STREAM,
121) -> None:
122 health_dir = journal / "chronicle" / DAY / "health"
123 health_dir.mkdir(parents=True, exist_ok=True)
124 path = health_dir / "001_segment.jsonl"
125 with path.open("a", encoding="utf-8") as handle:
126 handle.write(
127 json.dumps(
128 {
129 "event": event,
130 "ts": ts,
131 "mode": "segment",
132 "day": DAY,
133 "segment": ACTIVE_SEGMENT,
134 "stream": stream,
135 "name": name,
136 }
137 )
138 + "\n"
139 )
140
141
142def _patch_main_runtime(monkeypatch: pytest.MonkeyPatch) -> None:
143 from solstone.think import thinking as think
144
145 monkeypatch.setattr(think, "CallosumConnection", NullCallosumConnection)
146 monkeypatch.setattr(think, "check_callosum_available", lambda: True)
147
148
149def test_daily_health_log_keeps_segment_events_out(journal_copy, monkeypatch):
150 mod = importlib.import_module("solstone.think.thinking")
151
152 def mock_run_command(cmd, day):
153 return True
154
155 def mock_run_queued_command(cmd, day, timeout=600):
156 return True
157
158 def mock_run_daily_prompts(day, verbose, **kwargs):
159 return (5, 0, [], set())
160
161 _patch_main_runtime(monkeypatch)
162 monkeypatch.setattr(mod, "run_command", mock_run_command)
163 monkeypatch.setattr(
164 mod, "run_bounded_phase", lambda cmd, day, timeout=None: (True, False)
165 )
166 monkeypatch.setattr(mod, "run_queued_command", mock_run_queued_command)
167 monkeypatch.setattr(mod, "run_daily_prompts", mock_run_daily_prompts)
168 monkeypatch.setattr("sys.argv", ["sol think", "--day", "20240101"])
169
170 mod.main()
171
172 health_dir = journal_copy / "chronicle" / "20240101" / "health"
173 daily_files = sorted(health_dir.glob("*_daily.jsonl"))
174 assert len(daily_files) == 1
175
176 events = _read_jsonl(daily_files[0])
177 assert any(
178 event.get("phase") == "segment_think"
179 and _event_name(event).startswith("phase.")
180 for event in events
181 )
182 assert not [
183 event
184 for event in events
185 if _event_name(event).startswith(("talent.", "activity."))
186 ]
187
188
189def _forbid_slot_discovery(monkeypatch, mod):
190 """Assert the non-local path never probes the local server."""
191 del mod
192
193 def _unreachable() -> int:
194 raise AssertionError("slot discovery must not run for non-local defaults")
195
196 monkeypatch.setattr(fanout_policy, "read_server_parallel_slots", _unreachable)
197
198
199def _pin_describe_non_local(monkeypatch, mod):
200 """Pin the describe default to its CPU formula.
201
202 The fixture journal resolves observe.* to google, but these tests assert an
203 exact -j value; stub the predicate so they do not silently depend on that.
204 """
205 monkeypatch.setattr(fanout_policy, "_describe_uses_local", lambda: False)
206 _forbid_slot_discovery(monkeypatch, mod)
207
208
209def test_sense_repair_prephase_uses_default_describe_jobs(journal_copy, monkeypatch):
210 mod = importlib.import_module("solstone.think.thinking")
211 bounded_calls = []
212 daily_called = []
213
214 _pin_describe_non_local(monkeypatch, mod)
215
216 monkeypatch.setattr(fanout_policy.os, "cpu_count", lambda: 16)
217 assert fanout_policy.default_describe_jobs() == 4
218 monkeypatch.setattr(fanout_policy.os, "cpu_count", lambda: 7)
219 assert fanout_policy.default_describe_jobs() == 1
220 monkeypatch.setattr(fanout_policy.os, "cpu_count", lambda: 8)
221 assert fanout_policy.default_describe_jobs() == 2
222
223 def fake_bounded(cmd, day, timeout=None):
224 bounded_calls.append((cmd, day, timeout))
225 return (True, False)
226
227 def fake_daily(day, verbose, **kwargs):
228 daily_called.append(day)
229 return (5, 0, [], set())
230
231 _patch_main_runtime(monkeypatch)
232 monkeypatch.setattr(mod, "run_bounded_phase", fake_bounded)
233 monkeypatch.setattr(mod, "run_command", lambda cmd, day: True)
234 monkeypatch.setattr(mod, "run_queued_command", lambda cmd, day, timeout=600: True)
235 monkeypatch.setattr(mod, "run_daily_prompts", fake_daily)
236 monkeypatch.setattr("sys.argv", ["sol think", "--day", "20240101"])
237
238 mod.main()
239
240 assert daily_called == ["20240101"]
241 assert bounded_calls[0] == (
242 ["journal", "sense", "--day", "20240101", "-j", "2"],
243 "20240101",
244 mod.DEFAULT_TASK_MAX_RUNTIME,
245 )
246
247
248def test_daily_segment_prephase_timeout_is_nonfatal(journal_copy, monkeypatch):
249 mod = importlib.import_module("solstone.think.thinking")
250 bounded_calls = []
251 command_calls = []
252 daily_called = []
253
254 def fake_bounded(cmd, day, timeout=None):
255 bounded_calls.append((cmd, day, timeout))
256 return (False, True)
257
258 def fake_command(cmd, day):
259 command_calls.append(cmd)
260 return True
261
262 def fake_daily(day, verbose, **kwargs):
263 daily_called.append(day)
264 return (5, 0, [], set())
265
266 _patch_main_runtime(monkeypatch)
267 _pin_describe_non_local(monkeypatch, mod)
268 monkeypatch.setattr(fanout_policy.os, "cpu_count", lambda: 16)
269 monkeypatch.setattr(mod, "run_bounded_phase", fake_bounded)
270 monkeypatch.setattr(mod, "run_command", fake_command)
271 monkeypatch.setattr(mod, "run_queued_command", lambda cmd, day, timeout=600: True)
272 monkeypatch.setattr(mod, "run_daily_prompts", fake_daily)
273 monkeypatch.setattr("sys.argv", ["sol think", "--day", "20240101"])
274
275 mod.main()
276
277 health_dir = journal_copy / "chronicle" / "20240101" / "health"
278 daily_files = sorted(health_dir.glob("*_daily.jsonl"))
279 assert len(daily_files) == 1
280 events = _read_jsonl(daily_files[0])
281 segment_completes = [
282 event
283 for event in events
284 if _event_name(event) == "phase.complete"
285 and event.get("phase") == "segment_think"
286 ]
287 assert len(segment_completes) == 1
288 complete = segment_completes[0]
289 assert complete["success"] is False
290 assert complete["reason_code"] == "wall_clock_exceeded"
291 assert complete["timeout_seconds"] == 1800
292 assert complete["bounded"] is True
293
294 assert daily_called
295 assert bounded_calls == [
296 (
297 ["journal", "sense", "--day", "20240101", "-j", "4"],
298 "20240101",
299 mod.DEFAULT_TASK_MAX_RUNTIME,
300 ),
301 (
302 ["journal", "think", "--segments", "--day", "20240101"],
303 "20240101",
304 mod.DEFAULT_TASK_MAX_RUNTIME,
305 ),
306 (
307 ["journal", "journal-stats"],
308 "20240101",
309 mod.JOURNAL_STATS_MAX_RUNTIME,
310 ),
311 ]
312 assert mod.DEFAULT_TASK_MAX_RUNTIME == 1800
313 assert mod.JOURNAL_STATS_MAX_RUNTIME == 600
314 assert command_calls == []
315
316
317def test_daily_segment_prephase_records_yield_and_progressing_repair(
318 journal_copy, monkeypatch, caplog
319):
320 from solstone.think import catchup_state
321 from solstone.think import thinking as mod
322
323 day = "20240131"
324 _seed_segment(journal_copy, day, ACTIVE_SEGMENT)
325
326 def fake_bounded(cmd, day_arg, timeout=None):
327 if cmd[:2] == ["journal", "sense"]:
328 return (True, False)
329 if cmd[:3] == ["journal", "think", "--segments"]:
330 _write_jsonl(
331 journal_copy / "chronicle" / day / "health" / "001_segment.jsonl",
332 [
333 {
334 "event": "sense.complete",
335 "ts": 1,
336 "mode": "segment",
337 "day": day,
338 "segment": ACTIVE_SEGMENT,
339 "stream": STREAM,
340 "density": "active",
341 },
342 *[
343 {
344 "event": "talent.complete",
345 "ts": index + 2,
346 "mode": "segment",
347 "day": day,
348 "segment": ACTIVE_SEGMENT,
349 "stream": STREAM,
350 "name": name,
351 }
352 for index, name in enumerate(mod.SEGMENT_FLOOR_TALENTS)
353 ],
354 ],
355 )
356 return (False, True)
357 return (True, False)
358
359 def fake_daily(day, verbose, **kwargs):
360 return (1, 0, [], set())
361
362 _patch_main_runtime(monkeypatch)
363 monkeypatch.setattr(mod, "run_bounded_phase", fake_bounded)
364 monkeypatch.setattr(mod, "run_queued_command", lambda cmd, day, timeout=600: True)
365 monkeypatch.setattr(mod, "run_daily_prompts", fake_daily)
366 monkeypatch.setattr("sys.argv", ["sol think", "--day", day])
367
368 caplog.set_level(logging.WARNING)
369 mod.main()
370
371 health_dir = journal_copy / "chronicle" / day / "health"
372 daily_files = sorted(health_dir.glob("*_daily.jsonl"))
373 events = _read_jsonl(daily_files[0])
374 segment_complete = next(
375 event
376 for event in events
377 if _event_name(event) == "phase.complete"
378 and event.get("phase") == "segment_think"
379 )
380 assert segment_complete["success"] is False
381 assert segment_complete["cleared"] == 1
382 assert segment_complete["remaining"] == 0
383
384 record = catchup_state.read_day_record(day, catchup_state.KIND_SEGMENT_REPAIR)
385 assert record["last_outcome"] == catchup_state.PROGRESSING_OUTCOME
386 assert record["cleared"] == 1
387 assert record["remaining"] == 0
388 assert record["exit_reason"] == catchup_state.SEGMENT_REPAIR_TIMEOUT_REASON
389 assert "Segment-think repair exceeded its 1800s budget" in caplog.text
390 assert "yield: 1 cleared, 0 remaining" in caplog.text
391
392
393def test_daily_sense_prephase_timeout_records_disposition(journal_copy, monkeypatch):
394 mod = importlib.import_module("solstone.think.thinking")
395 daily_called = []
396
397 def fake_bounded(cmd, day, timeout=None):
398 if cmd[:2] == ["journal", "sense"]:
399 return (False, True)
400 return (True, False)
401
402 def fake_daily(day, verbose, **kwargs):
403 daily_called.append(day)
404 return (5, 0, [], set())
405
406 _patch_main_runtime(monkeypatch)
407 monkeypatch.setattr(mod, "run_bounded_phase", fake_bounded)
408 monkeypatch.setattr(mod, "run_command", lambda cmd, day: True)
409 monkeypatch.setattr(mod, "run_queued_command", lambda cmd, day, timeout=600: True)
410 monkeypatch.setattr(mod, "run_daily_prompts", fake_daily)
411 monkeypatch.setattr("sys.argv", ["sol think", "--day", "20240101"])
412
413 mod.main()
414
415 health_dir = journal_copy / "chronicle" / "20240101" / "health"
416 daily_files = sorted(health_dir.glob("*_daily.jsonl"))
417 assert len(daily_files) == 1
418 events = _read_jsonl(daily_files[0])
419 completes = [
420 e
421 for e in events
422 if _event_name(e) == "phase.complete" and e.get("phase") == "sense_repair"
423 ]
424 assert len(completes) == 1
425 complete = completes[0]
426 assert complete["success"] is False
427 assert complete["reason_code"] == "wall_clock_exceeded"
428 assert complete["timeout_seconds"] == mod.DEFAULT_TASK_MAX_RUNTIME
429 assert complete["bounded"] is True
430 # Pipeline continued past the sense pre-phase into the daily prompts.
431 assert daily_called
432
433
434def test_daily_journal_stats_postphase_timeout_records_disposition(
435 journal_copy, monkeypatch
436):
437 mod = importlib.import_module("solstone.think.thinking")
438 daily_called = []
439
440 def fake_bounded(cmd, day, timeout=None):
441 if cmd[:2] == ["journal", "journal-stats"]:
442 return (False, True)
443 return (True, False)
444
445 def fake_daily(day, verbose, **kwargs):
446 daily_called.append(day)
447 return (5, 0, [], set())
448
449 _patch_main_runtime(monkeypatch)
450 monkeypatch.setattr(mod, "run_bounded_phase", fake_bounded)
451 monkeypatch.setattr(mod, "run_command", lambda cmd, day: True)
452 monkeypatch.setattr(mod, "run_queued_command", lambda cmd, day, timeout=600: True)
453 monkeypatch.setattr(mod, "run_daily_prompts", fake_daily)
454 monkeypatch.setattr("sys.argv", ["sol think", "--day", "20240101"])
455
456 # main() returning normally proves the post-phase did not wedge the pipeline.
457 mod.main()
458
459 health_dir = journal_copy / "chronicle" / "20240101" / "health"
460 daily_files = sorted(health_dir.glob("*_daily.jsonl"))
461 assert len(daily_files) == 1
462 events = _read_jsonl(daily_files[0])
463 completes = [
464 e
465 for e in events
466 if _event_name(e) == "phase.complete" and e.get("phase") == "journal_stats"
467 ]
468 assert len(completes) == 1
469 complete = completes[0]
470 assert complete["success"] is False
471 assert complete["reason_code"] == "wall_clock_exceeded"
472 assert complete["timeout_seconds"] == mod.JOURNAL_STATS_MAX_RUNTIME
473 assert complete["bounded"] is True
474 assert daily_called
475
476
477def test_daily_postphase_emits_mixed_storage_warnings(journal_copy, monkeypatch):
478 mod = importlib.import_module("solstone.think.thinking")
479 callosum = importlib.import_module("solstone.think.callosum")
480 retention = importlib.import_module("solstone.think.retention")
481 sent = []
482 warnings = [
483 {
484 "level": "warning",
485 "type": "disk_percent",
486 "message": "disk warning",
487 "current": 95.0,
488 "threshold": 80,
489 },
490 {
491 "level": "warning",
492 "type": "offload_stalled",
493 "message": "offload warning",
494 "current": None,
495 "threshold": None,
496 },
497 ]
498
499 def fake_daily(day, verbose, **kwargs):
500 return (5, 0, [], set())
501
502 def fake_send(tract, event, **fields):
503 sent.append((tract, event, fields))
504 return True
505
506 _patch_main_runtime(monkeypatch)
507 monkeypatch.setattr(
508 mod, "run_bounded_phase", lambda cmd, day, timeout=None: (True, False)
509 )
510 monkeypatch.setattr(mod, "run_command", lambda cmd, day: True)
511 monkeypatch.setattr(mod, "run_queued_command", lambda cmd, day, timeout=600: True)
512 monkeypatch.setattr(mod, "run_daily_prompts", fake_daily)
513 monkeypatch.setattr(retention, "compute_storage_summary", lambda: object())
514 monkeypatch.setattr(
515 retention,
516 "check_storage_health",
517 lambda summary, journal_path: warnings,
518 )
519 monkeypatch.setattr(callosum, "callosum_send", fake_send)
520 monkeypatch.setattr("sys.argv", ["sol think", "--day", "20240101"])
521
522 mod.main()
523
524 storage = [item for item in sent if item[:2] == ("storage", "warning")]
525 notification = [item for item in sent if item[:2] == ("notification", "show")]
526 assert storage == [
527 (
528 "storage",
529 "warning",
530 {
531 "level": "warning",
532 "type": "disk_percent",
533 "message": "disk warning",
534 "current": 95.0,
535 "threshold": 80,
536 },
537 ),
538 (
539 "storage",
540 "warning",
541 {
542 "level": "warning",
543 "type": "offload_stalled",
544 "message": "offload warning",
545 "current": None,
546 "threshold": None,
547 },
548 ),
549 ]
550 assert notification == [
551 (
552 "notification",
553 "show",
554 {
555 "title": "Storage Warning",
556 "message": "disk warning",
557 "action": "/app/settings#storage",
558 },
559 )
560 ]
561
562
563def test_daily_segment_prephase_failure_has_no_timeout_reason(
564 journal_copy, monkeypatch
565):
566 mod = importlib.import_module("solstone.think.thinking")
567 daily_called = []
568
569 def fake_daily(day, verbose, **kwargs):
570 daily_called.append(day)
571 return (5, 0, [], set())
572
573 _patch_main_runtime(monkeypatch)
574 monkeypatch.setattr(
575 mod, "run_bounded_phase", lambda cmd, day, timeout=None: (False, False)
576 )
577 monkeypatch.setattr(mod, "run_command", lambda cmd, day: True)
578 monkeypatch.setattr(mod, "run_queued_command", lambda cmd, day, timeout=600: True)
579 monkeypatch.setattr(mod, "run_daily_prompts", fake_daily)
580 monkeypatch.setattr("sys.argv", ["sol think", "--day", "20240101"])
581
582 mod.main()
583
584 health_dir = journal_copy / "chronicle" / "20240101" / "health"
585 daily_files = sorted(health_dir.glob("*_daily.jsonl"))
586 assert len(daily_files) == 1
587 events = _read_jsonl(daily_files[0])
588 segment_completes = [
589 event
590 for event in events
591 if _event_name(event) == "phase.complete"
592 and event.get("phase") == "segment_think"
593 ]
594 assert len(segment_completes) == 1
595 complete = segment_completes[0]
596 assert complete["success"] is False
597 assert "reason_code" not in complete
598 assert "timeout_seconds" not in complete
599 assert "bounded" not in complete
600 assert daily_called
601
602
603def test_run_bounded_phase_timeout_with_exit_zero_propagates_failure(
604 journal_copy, monkeypatch
605):
606 mod = importlib.import_module("solstone.think.thinking")
607 log_path = journal_copy / "x.log"
608 fake = Mock()
609 fake.log_writer = Mock()
610 fake.log_writer.path = log_path
611 fake.name = "test"
612 fake.wait.side_effect = subprocess.TimeoutExpired(cmd=["x"], timeout=0.01)
613 fake.terminate.return_value = 0
614 fake.cleanup.return_value = None
615 monkeypatch.setattr(
616 "solstone.think.runner.ManagedProcess.spawn", lambda *a, **k: fake
617 )
618
619 result = mod.run_bounded_phase(
620 ["journal", "think", "--segments", "--day", DAY],
621 DAY,
622 timeout=0.01,
623 )
624
625 assert result == (False, True)
626
627
628def test_segment_health_log_receives_segment_talent_events(tmp_path, monkeypatch):
629 from solstone.think import thinking as think
630
631 journal = tmp_path / "journal"
632 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
633 _seed_segment(journal, DAY, ACTIVE_SEGMENT)
634
635 segments = think.cluster_segments(DAY)
636 assert any(segment["key"] == ACTIVE_SEGMENT for segment in segments)
637
638 _patch_main_runtime(monkeypatch)
639 # A tmp journal carries no local artifacts, so the real predicate resolves
640 # non-local and the default must never probe the server.
641 _forbid_slot_discovery(monkeypatch, think)
642 monkeypatch.setattr(
643 think,
644 "get_talent_configs",
645 lambda schedule=None, **kwargs: _segment_configs("sense"),
646 )
647 monkeypatch.setattr(
648 think,
649 "cortex_request",
650 lambda prompt, name, config=None, **kwargs: f"agent-{name}",
651 )
652 monkeypatch.setattr(
653 think,
654 "wait_for_uses",
655 lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
656 )
657 monkeypatch.setattr("sys.argv", ["sol think", "--segments", "--day", DAY])
658
659 with pytest.raises(SystemExit) as excinfo:
660 think.main()
661
662 assert excinfo.value.code == 0
663 health_dir = journal / "chronicle" / DAY / "health"
664 segment_files = sorted(health_dir.glob("*_segment.jsonl"))
665 assert len(segment_files) == 1
666 assert any(
667 _event_name(event).startswith("talent.")
668 for event in _read_jsonl(segment_files[0])
669 )
670 assert not list(health_dir.glob("*_daily.jsonl"))
671
672
673def test_select_segment_repair_targets_skips_complete_and_unsensed(monkeypatch):
674 from solstone.think import thinking as think
675
676 complete = {
677 "key": "090000_300",
678 "stream": STREAM,
679 "data_state": {"screen": "analyzed"},
680 }
681 incomplete = {
682 "key": "090500_300",
683 "stream": STREAM,
684 "data_state": {"screen": "analyzed"},
685 }
686 raw_blocked = {
687 "key": "091000_300",
688 "stream": STREAM,
689 "data_state": {"screen": "pending"},
690 }
691 segments = [complete, incomplete, raw_blocked]
692 monkeypatch.setattr(
693 think,
694 "read_segment_progress",
695 lambda day: {(STREAM, complete["key"]): _complete_segment_progress()},
696 )
697
698 selected, counts = think._select_segment_repair_targets(
699 DAY,
700 segments,
701 force_all=False,
702 )
703
704 assert selected == [incomplete]
705 assert counts == {
706 "total": 3,
707 "selected": 1,
708 "complete": 1,
709 "raw_blocked": 1,
710 }
711
712
713def test_select_segment_repair_targets_force_all_preserves_refresh_semantics(
714 monkeypatch,
715):
716 from solstone.think import thinking as think
717
718 segments = [
719 {
720 "key": "090000_300",
721 "stream": STREAM,
722 "data_state": {"screen": "analyzed"},
723 },
724 {
725 "key": "090500_300",
726 "stream": STREAM,
727 "data_state": {"screen": "pending"},
728 },
729 ]
730 monkeypatch.setattr(think, "read_segment_progress", lambda day: {})
731
732 selected, counts = think._select_segment_repair_targets(
733 DAY,
734 segments,
735 force_all=True,
736 )
737
738 assert selected == segments
739 assert counts == {
740 "total": 2,
741 "selected": 2,
742 "complete": 0,
743 "raw_blocked": 0,
744 }
745
746
747def test_raw_media_pending_skip_becomes_repair_selectable_after_describe_output(
748 monkeypatch,
749):
750 from solstone.think import thinking as think
751
752 pending = {
753 "key": "091000_300",
754 "stream": STREAM,
755 "data_state": {"screen": "pending"},
756 }
757 sensed = {
758 "key": pending["key"],
759 "stream": STREAM,
760 "data_state": {"screen": "analyzed"},
761 }
762 monkeypatch.setattr(think, "read_segment_progress", lambda day: {})
763
764 selected, counts = think._select_segment_repair_targets(
765 DAY,
766 [pending],
767 force_all=False,
768 )
769 assert selected == []
770 assert counts == {
771 "total": 1,
772 "selected": 0,
773 "complete": 0,
774 "raw_blocked": 1,
775 }
776
777 selected, counts = think._select_segment_repair_targets(
778 DAY,
779 [sensed],
780 force_all=False,
781 )
782 assert selected == [sensed]
783 assert counts == {
784 "total": 1,
785 "selected": 1,
786 "complete": 0,
787 "raw_blocked": 0,
788 }
789
790
791def test_media_terminal_no_sense_complete_is_selected_and_dispatched(monkeypatch):
792 from solstone.think import thinking as think
793 from solstone.think.pipeline_health import classify_segment_completion
794
795 segment = {
796 "key": "090000_300",
797 "stream": STREAM,
798 "start": "09:00",
799 "end": "09:05",
800 "data_state": {"screen": "analyzed"},
801 }
802 monkeypatch.setattr(think, "read_segment_progress", lambda day: {})
803
804 completion = classify_segment_completion([segment], {})
805 assert completion.blockers == [
806 {
807 "segment": segment["key"],
808 "dimension": "not_thought",
809 "detail": "no_sense_complete",
810 }
811 ]
812
813 selected, counts = think._select_segment_repair_targets(
814 DAY,
815 [segment],
816 force_all=False,
817 )
818 assert selected == [segment]
819 assert counts == {
820 "total": 1,
821 "selected": 1,
822 "complete": 0,
823 "raw_blocked": 0,
824 }
825
826 dispatched = []
827
828 def fake_run_segment_sense(**kwargs):
829 dispatched.append((kwargs["stream"], kwargs["segment"]))
830 return (1, 0, [])
831
832 monkeypatch.setattr(think, "run_segment_sense", fake_run_segment_sense)
833 monkeypatch.setattr(think, "resolve_predecessor", lambda *args: None)
834
835 success, failed = think._run_segment_repair_batch(
836 day=DAY,
837 segments=selected,
838 refresh=False,
839 verbose=False,
840 max_concurrency=1,
841 segment_workers=1,
842 timeout=None,
843 skip_activity_prompts=False,
844 skip_talents=frozenset(),
845 )
846
847 assert (success, failed) == (1, 0)
848 assert dispatched == [(STREAM, segment["key"])]
849
850
851def test_run_segment_repair_batch_respects_worker_bound(monkeypatch):
852 from solstone.think import thinking as think
853
854 segments = [
855 {"key": f"090{i}00_300", "stream": STREAM, "start": "09:00", "end": "09:05"}
856 for i in range(4)
857 ]
858 lock = threading.Lock()
859 barrier = threading.Barrier(2, timeout=1.0)
860 current = 0
861 peak = 0
862
863 def fake_run_segment_sense(**kwargs):
864 nonlocal current, peak
865 with lock:
866 current += 1
867 peak = max(peak, current)
868 try:
869 barrier.wait()
870 except threading.BrokenBarrierError:
871 pass
872 time.sleep(0.02)
873 with lock:
874 current -= 1
875 return (1, 0, [])
876
877 monkeypatch.setattr(think, "run_segment_sense", fake_run_segment_sense)
878 monkeypatch.setattr(think, "resolve_predecessor", lambda *args: None)
879
880 success, failed = think._run_segment_repair_batch(
881 day=DAY,
882 segments=segments,
883 refresh=False,
884 verbose=False,
885 max_concurrency=2,
886 segment_workers=2,
887 timeout=None,
888 skip_activity_prompts=False,
889 skip_talents=frozenset(),
890 )
891
892 assert (success, failed) == (4, 0)
893 assert peak == 2
894
895
896def test_segments_mode_targets_incomplete_tail_and_replays_full_day(
897 tmp_path,
898 monkeypatch,
899):
900 from solstone.think import thinking as think
901 from solstone.think import utils as think_utils
902
903 day = "20240117"
904 journal = tmp_path / "journal"
905 (journal / "chronicle" / day).mkdir(parents=True)
906 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
907 monkeypatch.setenv("SOL_SKIP_SUPERVISOR_CHECK", "1")
908 think_utils._journal_path_cache = None
909 _patch_main_runtime(monkeypatch)
910
911 complete = {
912 "key": "090000_300",
913 "stream": STREAM,
914 "start": "09:00",
915 "end": "09:05",
916 "data_state": {"screen": "analyzed"},
917 }
918 incomplete = {
919 "key": "090500_300",
920 "stream": STREAM,
921 "start": "09:05",
922 "end": "09:10",
923 "data_state": {"screen": "analyzed"},
924 }
925 raw_blocked = {
926 "key": "091000_300",
927 "stream": STREAM,
928 "start": "09:10",
929 "end": "09:15",
930 "data_state": {"screen": "pending"},
931 }
932 segments = [complete, incomplete, raw_blocked]
933 calls: list[dict] = []
934 replay_calls: list[dict] = []
935
936 def fake_run_segment_sense(**kwargs):
937 calls.append(kwargs)
938 return (1, 0, [])
939
940 monkeypatch.setattr(think, "cluster_segments", lambda day: segments)
941 monkeypatch.setattr(
942 think,
943 "read_segment_progress",
944 lambda day: {(STREAM, complete["key"]): _complete_segment_progress()},
945 )
946 monkeypatch.setattr(think, "run_segment_sense", fake_run_segment_sense)
947 monkeypatch.setattr(think, "resolve_predecessor", lambda *args: None)
948 monkeypatch.setattr(
949 think,
950 "_replay_activity_state_for_segments",
951 lambda **kwargs: replay_calls.append(kwargs),
952 )
953 monkeypatch.setattr(
954 "sys.argv",
955 ["sol think", "--segments", "--day", day, "--segment-workers", "1"],
956 )
957
958 with pytest.raises(SystemExit) as excinfo:
959 think.main()
960
961 assert excinfo.value.code == 0
962 assert [call["segment"] for call in calls] == [incomplete["key"]]
963 assert calls[0]["state_machine"] is None
964 assert calls[0]["skip_activity_prompts"] is True
965 assert replay_calls[0]["segments"] == segments
966
967
968def test_segments_mode_complete_day_noops_without_dispatch(tmp_path, monkeypatch):
969 from solstone.think import thinking as think
970 from solstone.think import utils as think_utils
971
972 day = "20240118"
973 journal = tmp_path / "journal"
974 (journal / "chronicle" / day).mkdir(parents=True)
975 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
976 monkeypatch.setenv("SOL_SKIP_SUPERVISOR_CHECK", "1")
977 think_utils._journal_path_cache = None
978 _patch_main_runtime(monkeypatch)
979
980 segment = {
981 "key": "090000_300",
982 "stream": STREAM,
983 "start": "09:00",
984 "end": "09:05",
985 "data_state": {"screen": "analyzed"},
986 }
987 calls: list[dict] = []
988 replay_calls: list[dict] = []
989 monkeypatch.setattr(think, "cluster_segments", lambda day: [segment])
990 monkeypatch.setattr(
991 think,
992 "read_segment_progress",
993 lambda day: {(STREAM, segment["key"]): _complete_segment_progress()},
994 )
995 monkeypatch.setattr(
996 think,
997 "run_segment_sense",
998 lambda **kwargs: calls.append(kwargs) or (1, 0, []),
999 )
1000 monkeypatch.setattr(
1001 think,
1002 "_replay_activity_state_for_segments",
1003 lambda **kwargs: replay_calls.append(kwargs),
1004 )
1005 monkeypatch.setattr(
1006 "sys.argv",
1007 ["sol think", "--segments", "--day", day, "--segment-workers", "1"],
1008 )
1009
1010 with pytest.raises(SystemExit) as excinfo:
1011 think.main()
1012
1013 assert excinfo.value.code == 0
1014 assert calls == []
1015 assert replay_calls == []
1016
1017
1018def test_existing_segment_talent_output_prevents_second_llm_run(
1019 tmp_path,
1020 monkeypatch,
1021):
1022 from solstone.think import models, talents
1023
1024 journal = tmp_path / "journal"
1025 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
1026 out = journal / "chronicle" / DAY / STREAM / ACTIVE_SEGMENT / "talents" / "flow.md"
1027 events: list[dict] = []
1028 called: list[str] = []
1029
1030 def fake_generate_with_result(**kwargs):
1031 called.append("generate")
1032 return {"text": "FRESH", "usage": {"output_tokens": 500}}
1033
1034 monkeypatch.setattr(models, "generate_with_result", fake_generate_with_result)
1035 monkeypatch.setattr(talents, "_run_pre_hooks", lambda config: {})
1036
1037 config = {
1038 "type": "generate",
1039 "name": "flow",
1040 "provider": "google",
1041 "model": "x",
1042 "prompt": "think about this segment",
1043 "day": DAY,
1044 "segment": ACTIVE_SEGMENT,
1045 "stream": STREAM,
1046 "output_path": str(out),
1047 "output": "md",
1048 "refresh": False,
1049 "schedule": "segment",
1050 }
1051
1052 # Lode-level analogue of test_talent_fallback's guard: the second run is
1053 # the healthy-day segment re-think and must use the cached output.
1054 asyncio.run(talents._run_talent(config, events.append, dry_run=False))
1055 assert len(called) == 1
1056 assert out.exists()
1057
1058 _append_segment_terminal(journal, name="flow")
1059
1060 second_events: list[dict] = []
1061 asyncio.run(talents._run_talent(config, second_events.append, dry_run=False))
1062
1063 finish_events = [event for event in second_events if event.get("event") == "finish"]
1064 assert len(called) == 1
1065 assert finish_events[-1]["result"] == "FRESH"
1066 assert finish_events[-1]["cache_hit"] is True
1067 assert finish_events[-1]["output_changed"] is False
1068 assert "usage" not in finish_events[-1]
1069
1070 changed_config = dict(config)
1071 changed_config["model"] = "y"
1072 third_events: list[dict] = []
1073 asyncio.run(talents._run_talent(changed_config, third_events.append, dry_run=False))
1074 assert len(called) == 2
1075 third_finish = [event for event in third_events if event.get("event") == "finish"][
1076 -1
1077 ]
1078 assert third_finish["cache_hit"] is False
1079
1080 transcript_changed_config = dict(changed_config)
1081 transcript_changed_config["transcript"] = "new source content from the segment"
1082 fourth_events: list[dict] = []
1083 asyncio.run(
1084 talents._run_talent(
1085 transcript_changed_config,
1086 fourth_events.append,
1087 dry_run=False,
1088 )
1089 )
1090 assert len(called) == 3
1091 fourth_finish = [
1092 event for event in fourth_events if event.get("event") == "finish"
1093 ][-1]
1094 assert fourth_finish["cache_hit"] is False
1095
1096
1097def test_provenance_reuse_requires_successful_latest_terminal(
1098 tmp_path,
1099 monkeypatch,
1100):
1101 from solstone.think import models, talents
1102
1103 journal = tmp_path / "journal"
1104 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
1105 out = (
1106 journal / "chronicle" / DAY / STREAM / ACTIVE_SEGMENT / "talents" / "screen.md"
1107 )
1108 called: list[str] = []
1109
1110 def fake_generate_with_result(**kwargs):
1111 called.append("generate")
1112 return {"text": "FRESH", "usage": {"output_tokens": 500}}
1113
1114 monkeypatch.setattr(models, "generate_with_result", fake_generate_with_result)
1115 monkeypatch.setattr(talents, "_run_pre_hooks", lambda config: {})
1116
1117 config = {
1118 "type": "generate",
1119 "name": "screen",
1120 "provider": "google",
1121 "model": "x",
1122 "prompt": "think about this segment",
1123 "day": DAY,
1124 "segment": ACTIVE_SEGMENT,
1125 "stream": STREAM,
1126 "output_path": str(out),
1127 "output": "md",
1128 "refresh": False,
1129 "schedule": "segment",
1130 }
1131
1132 asyncio.run(talents._run_talent(config, lambda event: None, dry_run=False))
1133 _append_segment_terminal(journal, name="screen", ts=100)
1134 _append_segment_terminal(journal, name="screen", event="talent.fail", ts=200)
1135
1136 events: list[dict] = []
1137 asyncio.run(talents._run_talent(config, events.append, dry_run=False))
1138
1139 assert len(called) == 2
1140 finish = [event for event in events if event.get("event") == "finish"][-1]
1141 assert finish["cache_hit"] is False
1142
1143
1144def test_refresh_identical_output_regenerates_without_output_changed(
1145 tmp_path,
1146 monkeypatch,
1147):
1148 from solstone.think import models, talents
1149
1150 journal = tmp_path / "journal"
1151 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
1152 out = (
1153 journal / "chronicle" / DAY / STREAM / ACTIVE_SEGMENT / "talents" / "screen.md"
1154 )
1155 out.parent.mkdir(parents=True, exist_ok=True)
1156 out.write_text("SAME", encoding="utf-8")
1157 called: list[str] = []
1158
1159 def fake_generate_with_result(**kwargs):
1160 called.append("generate")
1161 return {"text": "SAME", "usage": {"output_tokens": 500}}
1162
1163 monkeypatch.setattr(models, "generate_with_result", fake_generate_with_result)
1164 monkeypatch.setattr(talents, "_run_pre_hooks", lambda config: {})
1165
1166 events: list[dict] = []
1167 asyncio.run(
1168 talents._run_talent(
1169 {
1170 "type": "generate",
1171 "name": "screen",
1172 "provider": "google",
1173 "model": "x",
1174 "prompt": "think about this segment",
1175 "day": DAY,
1176 "segment": ACTIVE_SEGMENT,
1177 "stream": STREAM,
1178 "output_path": str(out),
1179 "output": "md",
1180 "refresh": True,
1181 "schedule": "segment",
1182 },
1183 events.append,
1184 dry_run=False,
1185 )
1186 )
1187
1188 finish = [event for event in events if event.get("event") == "finish"][-1]
1189 assert len(called) == 1
1190 assert finish["cache_hit"] is False
1191 assert finish["output_changed"] is False
1192
1193
1194def test_json_cache_reuse_requires_current_schema_validation(
1195 tmp_path,
1196 monkeypatch,
1197):
1198 from solstone.think import models, talents
1199 from solstone.think.talent_provenance import output_digest, write_provenance
1200
1201 journal = tmp_path / "journal"
1202 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
1203 out = (
1204 journal / "chronicle" / DAY / STREAM / ACTIVE_SEGMENT / "talents" / "sense.json"
1205 )
1206 out.parent.mkdir(parents=True, exist_ok=True)
1207 out.write_text('{"bad": true}', encoding="utf-8")
1208
1209 config = {
1210 "type": "generate",
1211 "name": "sense",
1212 "provider": "google",
1213 "model": "x",
1214 "prompt": "think about this segment",
1215 "day": DAY,
1216 "segment": ACTIVE_SEGMENT,
1217 "stream": STREAM,
1218 "output_path": str(out),
1219 "output": "json",
1220 "json_schema": {
1221 "type": "object",
1222 "required": ["ok"],
1223 "properties": {"ok": {"type": "boolean"}},
1224 },
1225 "refresh": False,
1226 "schedule": "segment",
1227 }
1228 runtime_schema = talents.hydrate_runtime_enums(config["json_schema"])
1229 output_sha256, output_size = output_digest(out)
1230 write_provenance(
1231 out,
1232 identity_hash=talents._identity_hash(config, runtime_schema),
1233 output_sha256=output_sha256,
1234 output_size=output_size,
1235 provider="google",
1236 model="x",
1237 generation_params=talents._generation_params(config),
1238 completed_at_ms=100,
1239 use_id="seed",
1240 identity_fields={"name": "sense"},
1241 )
1242 _append_segment_terminal(journal, name="sense", ts=100)
1243
1244 called: list[str] = []
1245
1246 def fake_generate_with_result(**kwargs):
1247 called.append("generate")
1248 return {
1249 "text": '{"ok": true}',
1250 "usage": {"output_tokens": 500},
1251 "schema_validation": {"valid": True, "errors": []},
1252 }
1253
1254 monkeypatch.setattr(models, "generate_with_result", fake_generate_with_result)
1255 monkeypatch.setattr(talents, "_run_pre_hooks", lambda config: {})
1256
1257 events: list[dict] = []
1258 asyncio.run(talents._run_talent(config, events.append, dry_run=False))
1259
1260 assert len(called) == 1
1261 finish = [event for event in events if event.get("event") == "finish"][-1]
1262 assert finish["cache_hit"] is False
1263 assert out.read_text(encoding="utf-8") == '{"ok": true}'
1264
1265
1266def test_identity_hash_ignores_source_count_vocabulary():
1267 from solstone.think import talents
1268
1269 config = {
1270 "type": "generate",
1271 "name": "sense",
1272 "provider": "google",
1273 "model": "x",
1274 "prompt": "think about this segment",
1275 "day": DAY,
1276 "segment": ACTIVE_SEGMENT,
1277 "stream": STREAM,
1278 "output": "json",
1279 "json_schema": {
1280 "type": "object",
1281 "required": ["ok"],
1282 "properties": {"ok": {"type": "boolean"}},
1283 },
1284 "schedule": "segment",
1285 "sources": {
1286 "transcripts": True,
1287 "percepts": False,
1288 "talents": {"sense": True},
1289 },
1290 "source_counts": {"transcripts": 1, "percepts": 0, "agents": 1},
1291 }
1292 runtime_schema = talents.hydrate_runtime_enums(config["json_schema"])
1293 renamed_counts = {
1294 **config,
1295 "source_counts": {"transcripts": 1, "percepts": 0, "talents": 1},
1296 }
1297
1298 assert talents._identity_hash(config, runtime_schema) == talents._identity_hash(
1299 renamed_counts,
1300 runtime_schema,
1301 )
1302
1303
1304def test_activity_replay_dedupes_records_and_preserves_non_refresh(
1305 tmp_path,
1306 monkeypatch,
1307):
1308 from solstone.think import thinking as think
1309 from solstone.think.activities import make_activity_id
1310 from solstone.think.activity_state_machine import ActivityStateMachine
1311
1312 journal = tmp_path / "journal"
1313 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
1314 _seed_segment(journal, DAY, ACTIVE_SEGMENT, _active_sense())
1315 _seed_segment(journal, DAY, IDLE_SEGMENT, _idle_sense())
1316
1317 activity_calls: list[dict] = []
1318
1319 monkeypatch.setattr(
1320 think,
1321 "get_talent_configs",
1322 lambda schedule=None, **kwargs: _segment_configs("sense"),
1323 )
1324 monkeypatch.setattr(
1325 think,
1326 "cortex_request",
1327 lambda prompt, name, config=None, **kwargs: f"agent-{name}",
1328 )
1329 monkeypatch.setattr(
1330 think,
1331 "wait_for_uses",
1332 lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
1333 )
1334 monkeypatch.setattr(
1335 think,
1336 "run_activity_prompts",
1337 lambda **kwargs: activity_calls.append(kwargs) or True,
1338 )
1339 monkeypatch.setattr(think, "_callosum", None)
1340 health_log = journal / "chronicle" / DAY / "health" / "activity_test.jsonl"
1341 writer = think.ThinkingJSONLWriter(str(health_log))
1342 monkeypatch.setattr(think, "_jsonl", writer)
1343
1344 try:
1345 for _ in range(2):
1346 state_machine = ActivityStateMachine()
1347 for segment in (ACTIVE_SEGMENT, IDLE_SEGMENT):
1348 think.run_segment_sense(
1349 DAY,
1350 segment,
1351 refresh=False,
1352 verbose=False,
1353 stream=STREAM,
1354 state_machine=state_machine,
1355 )
1356
1357 changed = _active_sense()
1358 changed["activity_summary"] = "Writing tests with new context"
1359 _write_sense_output(
1360 journal / "chronicle" / DAY / STREAM / ACTIVE_SEGMENT,
1361 changed,
1362 )
1363 state_machine = ActivityStateMachine()
1364 for segment in (ACTIVE_SEGMENT, IDLE_SEGMENT):
1365 think.run_segment_sense(
1366 DAY,
1367 segment,
1368 refresh=False,
1369 verbose=False,
1370 stream=STREAM,
1371 state_machine=state_machine,
1372 )
1373 finally:
1374 writer.close()
1375
1376 record_path = journal / "facets" / FACET / "activities" / f"{DAY}.jsonl"
1377 records = _read_jsonl(record_path)
1378 activity_id = make_activity_id("coding", ACTIVE_SEGMENT)
1379 matching = [record for record in records if record.get("id") == activity_id]
1380
1381 assert len(matching) == 1
1382 assert len(activity_calls) == 2
1383 assert activity_calls[0]["refresh"] is False
1384 assert activity_calls[1]["refresh"] is False
1385 health_events = _read_jsonl(health_log)
1386 assert any(event.get("event") == "activity.unchanged" for event in health_events)
1387
1388
1389def test_activity_replay_treats_malformed_sense_like_absent(
1390 tmp_path,
1391 monkeypatch,
1392 caplog,
1393):
1394 from solstone.think import thinking as think
1395 from solstone.think.activity_state_machine import ActivityStateMachine
1396
1397 ending_segment = "091000_300"
1398 segments = [
1399 {"stream": STREAM, "key": ACTIVE_SEGMENT},
1400 {"stream": STREAM, "key": IDLE_SEGMENT},
1401 {"stream": STREAM, "key": ending_segment},
1402 ]
1403
1404 monkeypatch.setattr(
1405 "solstone.think.activity_state_machine.time.time",
1406 lambda: 123.456,
1407 )
1408 monkeypatch.setattr(
1409 think,
1410 "run_activity_prompts",
1411 lambda **kwargs: True,
1412 )
1413 monkeypatch.setattr(think, "_callosum", None)
1414 state_machines: list[ActivityStateMachine] = []
1415
1416 class CapturingActivityStateMachine(ActivityStateMachine):
1417 def __init__(self, *args, **kwargs):
1418 super().__init__(*args, **kwargs)
1419 state_machines.append(self)
1420
1421 monkeypatch.setattr(think, "ActivityStateMachine", CapturingActivityStateMachine)
1422
1423 def run_case(journal: Path, *, malformed: bool) -> tuple[bytes, bytes]:
1424 start_index = len(state_machines)
1425 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
1426 _seed_segment(journal, DAY, ACTIVE_SEGMENT, _active_sense())
1427 middle_dir = _seed_segment(journal, DAY, IDLE_SEGMENT, _active_sense())
1428 _seed_segment(journal, DAY, ending_segment, _idle_sense())
1429 middle_sense = middle_dir / "talents" / "sense.json"
1430 if malformed:
1431 _write_sense_output(
1432 middle_dir,
1433 {
1434 "density": "active",
1435 "activity_summary": "Malformed cached output",
1436 "entities": [],
1437 "facets": [{"facet": FACET, "level": "high"}],
1438 "recommend": {},
1439 },
1440 )
1441 else:
1442 middle_sense.unlink()
1443
1444 think._replay_activity_state_for_segments(
1445 day=DAY,
1446 segments=segments,
1447 refresh=False,
1448 verbose=False,
1449 max_concurrency=2,
1450 skip_activity_prompts=False,
1451 )
1452 record_path = journal / "facets" / FACET / "activities" / f"{DAY}.jsonl"
1453 state_machine = state_machines[start_index]
1454 state_snapshot = {
1455 "last_segment_key": state_machine.last_segment_key,
1456 "last_segment_day": state_machine.last_segment_day,
1457 "active": {
1458 facet: {k: v for k, v in entry.items() if k != "_change"}
1459 for facet, entry in state_machine.state.items()
1460 },
1461 }
1462 return record_path.read_bytes(), json.dumps(
1463 state_snapshot, sort_keys=True
1464 ).encode()
1465
1466 caplog.set_level(logging.WARNING)
1467 malformed_records, malformed_state = run_case(
1468 tmp_path / "malformed", malformed=True
1469 )
1470 absent_records, absent_state = run_case(tmp_path / "absent", malformed=False)
1471
1472 assert malformed_records == absent_records
1473 assert malformed_state == absent_state
1474 assert "missing required keys: content_type" in caplog.text
1475
1476
1477@pytest.mark.parametrize("missing_key", ["density", "content_type"])
1478def test_run_segment_sense_reports_invalid_sense_output(
1479 tmp_path,
1480 monkeypatch,
1481 caplog,
1482 missing_key,
1483):
1484 from solstone.think import thinking as think
1485 from solstone.think.activity_state_machine import ActivityStateMachine
1486
1487 journal = tmp_path / "journal"
1488 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
1489 malformed = _active_sense()
1490 malformed.pop(missing_key)
1491 _seed_segment(journal, DAY, ACTIVE_SEGMENT, malformed)
1492
1493 monkeypatch.setattr(
1494 think,
1495 "get_talent_configs",
1496 lambda schedule=None, **kwargs: _segment_configs("sense"),
1497 )
1498 monkeypatch.setattr(
1499 think,
1500 "cortex_request",
1501 lambda prompt, name, config=None, **kwargs: f"agent-{name}",
1502 )
1503 monkeypatch.setattr(
1504 think,
1505 "wait_for_uses",
1506 lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
1507 )
1508 emitted: list[tuple[str, str, dict]] = []
1509
1510 class CaptureCallosum:
1511 def emit(self, tract, event, **fields):
1512 emitted.append((tract, event, fields))
1513
1514 monkeypatch.setattr(think, "_callosum", CaptureCallosum())
1515 caplog.set_level(logging.WARNING)
1516
1517 success, failed, failed_names = think.run_segment_sense(
1518 DAY,
1519 ACTIVE_SEGMENT,
1520 refresh=False,
1521 verbose=False,
1522 stream=STREAM,
1523 state_machine=ActivityStateMachine(),
1524 )
1525
1526 assert success == 1
1527 assert failed == 1
1528 assert failed_names == ["sense (output_invalid)"]
1529 assert f"missing required keys: {missing_key}" in caplog.text
1530 completed = [event for event in emitted if event[1] == "completed"]
1531 assert completed[-1][2]["failed"] == 1
1532 assert completed[-1][2]["failed_names"] == ["sense (output_invalid)"]
1533
1534
1535def test_cache_hit_priority_drain_suppresses_rescan_but_counts_segment_complete(
1536 tmp_path,
1537 monkeypatch,
1538):
1539 from solstone.think import thinking as think
1540 from solstone.think.pipeline_health import (
1541 read_completed_since,
1542 read_segment_progress,
1543 )
1544
1545 journal = tmp_path / "journal"
1546 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
1547 output_path = (
1548 journal / "chronicle" / DAY / STREAM / ACTIVE_SEGMENT / "talents" / "screen.md"
1549 )
1550 output_path.parent.mkdir(parents=True, exist_ok=True)
1551 output_path.write_text("cached", encoding="utf-8")
1552
1553 use_id = "1700000000000"
1554 use_dir = journal / "talents" / "screen"
1555 use_dir.mkdir(parents=True, exist_ok=True)
1556 (use_dir / f"{use_id}.jsonl").write_text(
1557 json.dumps(
1558 {
1559 "event": "finish",
1560 "ts": 300,
1561 "use_id": use_id,
1562 "name": "screen",
1563 "result": "cached",
1564 "cache_hit": True,
1565 "output_changed": False,
1566 "completed_at_ms": 100,
1567 }
1568 )
1569 + "\n",
1570 encoding="utf-8",
1571 )
1572
1573 health_log = journal / "chronicle" / DAY / "health" / "segment.jsonl"
1574 writer = think.ThinkingJSONLWriter(str(health_log))
1575 monkeypatch.setattr(think, "_jsonl", writer)
1576 monkeypatch.setattr(
1577 think,
1578 "wait_for_uses",
1579 lambda agent_ids, timeout=610: ({use_id: "finish"}, []),
1580 )
1581 rescan_calls: list[list[str]] = []
1582 monkeypatch.setattr(
1583 think,
1584 "run_queued_command",
1585 lambda cmd, day, timeout=600: rescan_calls.append(cmd) or True,
1586 )
1587
1588 try:
1589 success, failed, failed_names = think._drain_priority_batch(
1590 [(use_id, "screen", {"type": "generate", "output": "md"}, None)],
1591 "segment",
1592 DAY,
1593 ACTIVE_SEGMENT,
1594 STREAM,
1595 )
1596 finally:
1597 writer.close()
1598
1599 assert (success, failed, failed_names) == (1, 0, [])
1600 assert rescan_calls == []
1601 health_events = _read_jsonl(health_log)
1602 complete = [
1603 event for event in health_events if event.get("event") == "talent.complete"
1604 ][-1]
1605 assert complete["cache_hit"] is True
1606 assert complete["completed_at_ms"] == 100
1607 assert read_completed_since(DAY, 0).segments == ()
1608 assert "screen" in read_segment_progress(DAY)[(STREAM, ACTIVE_SEGMENT)].completed
1609
1610
1611def test_segments_mode_zero_segment_noop(tmp_path, monkeypatch, caplog):
1612 from solstone.think import thinking as think
1613
1614 day = "20240116"
1615 journal = tmp_path / "journal"
1616 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
1617 _patch_main_runtime(monkeypatch)
1618 _forbid_slot_discovery(monkeypatch, think)
1619 monkeypatch.setattr("sys.argv", ["sol think", "--segments", "--day", day])
1620
1621 caplog.set_level(logging.INFO)
1622 with pytest.raises(SystemExit) as excinfo:
1623 think.main()
1624
1625 assert excinfo.value.code == 0
1626 assert f"No segments found for {day}" in caplog.text
1627
1628
1629def _patch_segment_run(monkeypatch, think, journal):
1630 """Seed a runnable segment and stub the talent machinery around it."""
1631 _seed_segment(journal, DAY, ACTIVE_SEGMENT)
1632 _patch_main_runtime(monkeypatch)
1633 monkeypatch.setattr(
1634 think,
1635 "get_talent_configs",
1636 lambda schedule=None, **kwargs: _segment_configs("sense"),
1637 )
1638 monkeypatch.setattr(
1639 think,
1640 "cortex_request",
1641 lambda prompt, name, config=None, **kwargs: f"agent-{name}",
1642 )
1643 monkeypatch.setattr(
1644 think,
1645 "wait_for_uses",
1646 lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
1647 )
1648
1649
1650def test_segments_default_local_slot_fallback_logs_once_across_call_sites(
1651 tmp_path, monkeypatch, caplog
1652):
1653 """Discovery failure yields the floor value and logs the fallback once.
1654
1655 ``--segments`` calls ``default_segment_workers()`` twice in one process:
1656 once in argument validation and once in the run path. The tmp journal has
1657 no ``health/local.port``, so discovery fails without any network I/O.
1658 """
1659 from solstone.think import thinking as think
1660
1661 journal = tmp_path / "journal"
1662 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
1663 _patch_segment_run(monkeypatch, think, journal)
1664 monkeypatch.setattr(fanout_policy, "_segment_work_uses_local", lambda: True)
1665 monkeypatch.setattr(fanout_policy.os, "cpu_count", lambda: 12)
1666 monkeypatch.setattr("sys.argv", ["sol think", "--segments", "--day", DAY])
1667
1668 assert not (journal / "health" / "local.port").exists()
1669
1670 derived = []
1671 original_default = fanout_policy.default_segment_workers
1672
1673 def spy_default() -> int:
1674 value = original_default()
1675 derived.append(value)
1676 return value
1677
1678 monkeypatch.setattr(fanout_policy, "default_segment_workers", spy_default)
1679
1680 caplog.set_level(logging.INFO)
1681 with pytest.raises(SystemExit) as excinfo:
1682 think.main()
1683 assert excinfo.value.code == 0
1684
1685 # Validation (thinking.py:4161) and the run path (:4270) both call it.
1686 assert derived == [1, 1]
1687
1688 messages = [record.getMessage() for record in caplog.records]
1689 fallbacks = [m for m in messages if "local_server_parallel_slots fallback" in m]
1690 assert fallbacks == [
1691 "local_server_parallel_slots fallback slots=1 port=None "
1692 "context_tokens=None source=default"
1693 ]
1694
1695 caps = [m for m in messages if "default_segment_workers capped" in m]
1696 assert caps == [
1697 "default_segment_workers capped provider=local slots=1 formula=6 derived=1"
1698 ]
1699
1700
1701def test_segments_explicit_segment_workers_bypasses_local_default_at_call_site(
1702 tmp_path, monkeypatch
1703):
1704 """An explicit --segment-workers wins over a smaller derived default."""
1705 from solstone.think import thinking as think
1706
1707 journal = tmp_path / "journal"
1708 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
1709 _patch_segment_run(monkeypatch, think, journal)
1710
1711 # The derived default would be 1; the CLI asks for 6.
1712 monkeypatch.setattr(fanout_policy, "_segment_work_uses_local", lambda: True)
1713 monkeypatch.setattr(fanout_policy, "read_server_parallel_slots", lambda: 1)
1714 monkeypatch.setattr(fanout_policy.os, "cpu_count", lambda: 12)
1715 assert fanout_policy.default_segment_workers() == 1
1716
1717 observed = []
1718 original_batch = think._run_segment_repair_batch
1719
1720 def spy_batch(**kwargs):
1721 observed.append(kwargs["segment_workers"])
1722 return original_batch(**kwargs)
1723
1724 monkeypatch.setattr(think, "_run_segment_repair_batch", spy_batch)
1725 monkeypatch.setattr(
1726 "sys.argv",
1727 ["sol think", "--segments", "--day", DAY, "--segment-workers", "6"],
1728 )
1729
1730 with pytest.raises(SystemExit) as excinfo:
1731 think.main()
1732
1733 assert excinfo.value.code == 0
1734 assert observed == [6]