solpbc.org / solstone
personal memory agent
0
Fork 0

Configure Feed

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

fix(observe): re-enter record-less screen analysis outputs

Make processing_record the shared source of truth for analysis-output evidence by moving the bounded JSONL row-key detector there with audio and screen row-key constants, then route cluster through that helper.

An existing analysis output now blocks re-entry only when it carries evidence: analyzed rows or a processing record. Outputs with neither are indeterminate and re-enter until the record ledger can govern them; the record-less clause is describe/screen-only.

Decode-determined corrupt_input and empty verdicts terminalize regardless of provider availability, while runs that still have qualified frames to describe continue to defer when no brain is configured.

author
Jer Miller date (Jul 24, 2026, 1:22 PM -0600) commit 413cfba1 parent ace29f83
+694 -75
+14 -4
solstone/observe/describe.py
··· 51 51 build_processing_record, 52 52 read_processing_record_header, 53 53 record_attempts, 54 - should_reenter_failed_describe, 54 + should_reenter_analysis_output, 55 55 ) 56 56 from solstone.observe.utils import get_segment_key, resize_for_vlm 57 57 from solstone.think.callosum import callosum_send ··· 952 952 print(result_line.rstrip("\n"), flush=True) 953 953 954 954 try: 955 + if not had_qualified_frames: 956 + # Decoded frames are real work; do not discard them before description. 957 + if self.decode_failed: 958 + _promote(STATE_FAILED, REASON_CORRUPT_INPUT) 959 + else: 960 + _promote(STATE_EMPTY, REASON_NO_DECODABLE_FRAMES) 961 + return 962 + 955 963 frame_provider, frame_model = resolve_provider("generate") 956 964 if frame_provider == NO_BRAIN_PROVIDER: 957 965 logger.info("No thinking engine selected; deferring frame description") ··· 1452 1460 1453 1461 if self.decode_failed: 1454 1462 state, reason_code = STATE_FAILED, REASON_CORRUPT_INPUT 1455 - elif not had_qualified_frames: 1456 - state, reason_code = STATE_EMPTY, REASON_NO_DECODABLE_FRAMES 1457 1463 elif not emitted_row_has_error and emitted_frame_ids == qualified_ids: 1458 1464 state, reason_code = STATE_ANALYZED, REASON_OK 1459 1465 else: ··· 1556 1562 # Skip if already processed (unless redo mode) 1557 1563 if not args.redo and output_path.exists(): 1558 1564 record = read_processing_record_header(output_path) 1559 - if not should_reenter_failed_describe(record): 1565 + if not should_reenter_analysis_output( 1566 + record=record, 1567 + output_path=output_path, 1568 + handler=HANDLER_DESCRIBE, 1569 + ): 1560 1570 logger.info(f"Already processed: {video_path}") 1561 1571 return 1562 1572 previous_attempts = record_attempts(record)
+59 -12
solstone/observe/processing_record.py
··· 1 1 # SPDX-License-Identifier: AGPL-3.0-only 2 2 # Copyright (c) 2026 sol pbc 3 3 4 - """Shared `_solstone_processing` record vocabulary for media-analysis handlers. 4 + """Shared evidence vocabulary for media-analysis outputs. 5 5 6 - The screen (`describe`) and audio (`transcribe`) handlers stamp one of these 7 - records into the metadata header (row 1) of the JSONL they already produce, so 8 - a downstream reader lode can derive per-segment processing state without 9 - re-deriving it from raw media. This module is the one authoritative source of 10 - the closed state / reason_code / handler / schema vocabulary; neither handler 11 - may carry these literals inline. Failed describe records may also carry an 12 - ``attempts`` counter; absent attempts means 0, and 13 - ``FAILED_ATTEMPT_BOUND`` is the shared retry exhaustion bound. 6 + This module is the source of truth for the two forms of output evidence: 7 + ``_solstone_processing`` metadata-header records and the JSONL row keys that 8 + prove audio or screen analysis rows exist. The screen (``describe``) and audio 9 + (``transcribe``) handlers stamp processing records into the metadata header of 10 + the JSONL they produce, while row-key detection gives bounded evidence for 11 + legacy or record-less outputs. 12 + 13 + ``FileSensor.scan_unprocessed``, ``describe.async_main``, and 14 + ``derive_modality_state`` consume this vocabulary so capture re-entry, 15 + describe-side skipping, and downstream state derivation share the same reading 16 + of whether an output is useful, terminal, retryable, or indeterminate. 14 17 """ 15 18 16 19 import json ··· 21 24 FAILED_ATTEMPT_BOUND = 3 22 25 ATTEMPTS_KEY = "attempts" 23 26 MAX_FIRST_ROW_BYTES = 64 * 1024 27 + SCREEN_ANALYSIS_ROW_KEY = "timestamp" 28 + AUDIO_TRANSCRIPT_ROW_KEY = "start" 24 29 25 30 # state values (closed set) 26 31 STATE_ANALYZED = "analyzed" ··· 84 89 return record if isinstance(record, dict) else None 85 90 86 91 87 - def should_reenter_failed_describe(record: dict | None) -> bool: 88 - """Return whether an existing failed describe output should be retried.""" 89 - return ( 92 + def jsonl_has_row_with_key(path: Path, row_key: str) -> bool: 93 + """Return whether an early JSONL object has the row key. 94 + 95 + Key-based membership test on at most the first two nonblank lines. 96 + """ 97 + try: 98 + lines = [] 99 + with path.open("r", encoding="utf-8") as handle: 100 + for line in handle: 101 + if not line.strip(): 102 + continue 103 + lines.append(line) 104 + if len(lines) == 2: 105 + break 106 + except OSError: 107 + return False 108 + for line in lines: 109 + try: 110 + parsed = json.loads(line) 111 + except (json.JSONDecodeError, ValueError): 112 + continue 113 + if isinstance(parsed, dict) and row_key in parsed: 114 + return True 115 + return False 116 + 117 + 118 + # an existing analysis output only blocks re-entry when it actually carries 119 + # evidence — analyzed rows or a processing record. An output with neither is 120 + # indeterminate, and indeterminate work re-enters until the record ledger can 121 + # govern it. Decode-determined verdicts terminalize regardless of provider 122 + # availability. 123 + def should_reenter_analysis_output( 124 + *, 125 + record: dict | None, 126 + output_path: Path, 127 + handler: str, 128 + ) -> bool: 129 + """Return whether an existing analysis output should be retried.""" 130 + if ( 90 131 isinstance(record, dict) 91 132 and record.get("state") == STATE_FAILED 92 133 and record.get("handler") == HANDLER_DESCRIBE 93 134 and not is_failure_exhausted(record) 135 + ): 136 + return True 137 + return ( 138 + record is None 139 + and handler == HANDLER_DESCRIBE 140 + and not jsonl_has_row_with_key(output_path, SCREEN_ANALYSIS_ROW_KEY) 94 141 ) 95 142 96 143
+11 -8
solstone/observe/sense.py
··· 27 27 from solstone.observe.exit_codes import EXIT_PROVIDER_BLOCKED, WATCHDOG_TIMEOUT 28 28 from solstone.observe.processing_record import ( 29 29 read_processing_record_header, 30 - should_reenter_failed_describe, 30 + should_reenter_analysis_output, 31 31 ) 32 32 from solstone.observe.utils import ( 33 33 AUDIO_EXTENSIONS, ··· 1048 1048 if modality_filter == "screen" and suffix not in VIDEO_EXTENSIONS: 1049 1049 continue 1050 1050 1051 - # Check if output JSONL exists (already processed) 1052 - output_path = file_path.with_suffix(".jsonl") 1053 - if output_path.exists(): 1054 - record = read_processing_record_header(output_path) 1055 - if not should_reenter_failed_describe(record): 1056 - continue 1057 - 1058 1051 handler_info = self._match_pattern(file_path) 1059 1052 if handler_info: 1060 1053 handler_name, command = handler_info 1054 + # Check if output JSONL exists (already processed) 1055 + output_path = file_path.with_suffix(".jsonl") 1056 + if output_path.exists(): 1057 + record = read_processing_record_header(output_path) 1058 + if not should_reenter_analysis_output( 1059 + record=record, 1060 + output_path=output_path, 1061 + handler=handler_name, 1062 + ): 1063 + continue 1061 1064 if handler_name == "depict" and is_import_stream(stream_name): 1062 1065 continue 1063 1066 to_process.append((file_path, handler_name, command))
+9 -28
solstone/think/cluster.py
··· 11 11 from pathlib import Path 12 12 from typing import Any 13 13 14 + from solstone.observe.processing_record import ( 15 + AUDIO_TRANSCRIPT_ROW_KEY, 16 + SCREEN_ANALYSIS_ROW_KEY, 17 + jsonl_has_row_with_key, 18 + ) 14 19 from solstone.observe.screen import format_screen_text 15 20 from solstone.think.browser_formatter import format_browser_text 16 21 from solstone.think.data_state import ( ··· 478 483 return ranges 479 484 480 485 481 - def _jsonl_has_marker_row(path: Path, marker_key: str) -> bool: 482 - """Return whether an early JSONL object has the marker key. 483 - 484 - Key-based membership test on at most the first two nonblank lines. 485 - """ 486 - try: 487 - lines = [] 488 - with path.open("r", encoding="utf-8") as handle: 489 - for line in handle: 490 - if not line.strip(): 491 - continue 492 - lines.append(line) 493 - if len(lines) == 2: 494 - break 495 - except OSError: 496 - return False 497 - for line in lines: 498 - try: 499 - parsed = json.loads(line) 500 - except (json.JSONDecodeError, ValueError): 501 - continue 502 - if isinstance(parsed, dict) and marker_key in parsed: 503 - return True 504 - return False 505 - 506 - 507 486 def _read_processing_record(jsonl_files: list[Path]) -> dict | None: 508 487 """Return the first processing header record from sorted JSONL files.""" 509 488 for path in jsonl_files: ··· 593 572 ) 594 573 audio_md_files = _markdown_transcript_files(seg_path) 595 574 audio_analyzed = any( 596 - _jsonl_has_marker_row(path, "start") for path in audio_jsonl_files 575 + jsonl_has_row_with_key(path, AUDIO_TRANSCRIPT_ROW_KEY) 576 + for path in audio_jsonl_files 597 577 ) or any(_has_nonempty_text(path) for path in audio_md_files) 598 578 audio_record = _read_processing_record(audio_jsonl_files) 599 579 audio_has_raw = _has_raw_media(raw_media_paths, AUDIO_EXTENSIONS) ··· 617 597 } 618 598 ) 619 599 screen_analyzed = any( 620 - _jsonl_has_marker_row(path, "timestamp") for path in screen_jsonl_files 600 + jsonl_has_row_with_key(path, SCREEN_ANALYSIS_ROW_KEY) 601 + for path in screen_jsonl_files 621 602 ) 622 603 screen_record = _read_processing_record(screen_jsonl_files) 623 604 screen_has_raw = _has_raw_media(raw_media_paths, VIDEO_EXTENSIONS)
+165 -14
tests/test_bad_media_corpus.py
··· 35 35 STATE_ANALYZED, 36 36 STATE_EMPTY, 37 37 STATE_FAILED, 38 + is_failure_exhausted, 39 + should_reenter_analysis_output, 38 40 ) 39 41 from solstone.observe.utils import SAMPLE_RATE, AudioDecodeError 40 42 from solstone.observe.vad import VadResult ··· 44 46 ) 45 47 from solstone.think.data_state import DataState 46 48 from solstone.think.pipeline_health import ( 49 + SegmentProgress, 47 50 classify_segment_completion, 48 51 read_segment_progress, 52 + segment_fully_sensed, 49 53 ) 50 54 51 55 DAY = "20990501" ··· 290 294 agenerate_response: str = "{}", 291 295 agenerate_finish_reason: str = "stop", 292 296 expect_runtime_error: bool = False, 293 - ) -> tuple[dict[str, Any], dict[str, Any], AsyncMock]: 297 + provider_result: tuple[str, str] = ("google", "gemini-test"), 298 + expect_record: bool = True, 299 + ) -> tuple[dict[str, Any], dict[str, Any] | None, AsyncMock]: 294 300 from solstone.observe import describe, processing_record 295 301 296 302 agenerate = AsyncMock( ··· 298 304 ) 299 305 monkeypatch.setattr( 300 306 "solstone.think.models.resolve_provider", 301 - lambda _interface: ("google", "gemini-test"), 307 + lambda _interface: provider_result, 302 308 ) 303 309 monkeypatch.setattr(describe, "callosum_send", lambda *args, **kwargs: None) 304 310 monkeypatch.setattr(describe, "select_frames_for_extraction", lambda *a, **k: []) ··· 325 331 ) 326 332 327 333 header = _read_header(output_path) 328 - record = header["_solstone_processing"] 329 - assert isinstance(record, dict) 334 + record = header.get("_solstone_processing") 335 + if expect_record: 336 + assert isinstance(record, dict) 337 + else: 338 + assert record is None 330 339 return header, record, agenerate 331 340 332 341 ··· 641 650 monkeypatch, 642 651 ): 643 652 av = pytest.importorskip("av") 644 - from solstone.observe.describe import should_reenter_failed_describe 645 653 646 654 segment = _segment_dir(segment_journal) 647 655 video_path = segment / "screen.webm" ··· 671 679 assert read_segment_data_state(DAY, SEGMENT) == { 672 680 "screen": DataState.FAILED_FINAL.value 673 681 } 674 - assert should_reenter_failed_describe(record) is False 682 + assert ( 683 + should_reenter_analysis_output( 684 + record=record, 685 + output_path=output_path, 686 + handler=HANDLER_DESCRIBE, 687 + ) 688 + is False 689 + ) 690 + 691 + 692 + def test_ac4_no_provider_truncated_recordless_screen_converges_corrupt_input( 693 + segment_journal, 694 + monkeypatch, 695 + ): 696 + from solstone.think.models import NO_BRAIN_PROVIDER 697 + 698 + segment = _segment_dir(segment_journal, segment="123500_300") 699 + video_path = segment / "screen.webm" 700 + output_path = segment / "screen.jsonl" 701 + _build_truncated_webm(video_path) 702 + output_path.write_text( 703 + json.dumps({"raw": video_path.name}) + "\n", 704 + encoding="utf-8", 705 + ) 706 + 707 + _header, record, agenerate = _drive_describe( 708 + monkeypatch, 709 + video_path, 710 + output_path, 711 + provider_result=(NO_BRAIN_PROVIDER, ""), 712 + ) 713 + 714 + _assert_processing_record( 715 + record, 716 + state=STATE_FAILED, 717 + reason_code=REASON_CORRUPT_INPUT, 718 + handler=HANDLER_DESCRIBE, 719 + ) 720 + assert agenerate.call_count == 0 721 + assert is_failure_exhausted(record) is True 722 + assert ( 723 + should_reenter_analysis_output( 724 + record=record, 725 + output_path=output_path, 726 + handler=HANDLER_DESCRIBE, 727 + ) 728 + is False 729 + ) 730 + 731 + 732 + def test_ac6_failed_final_screen_record_unblocks_day_with_failed_marker( 733 + segment_journal, 734 + monkeypatch, 735 + ): 736 + from solstone.think.models import NO_BRAIN_PROVIDER 737 + 738 + segment_key = "123600_300" 739 + segment = _segment_dir(segment_journal, segment=segment_key) 740 + video_path = segment / "screen.webm" 741 + output_path = segment / "screen.jsonl" 742 + _build_truncated_webm(video_path) 743 + output_path.write_text( 744 + json.dumps({"raw": video_path.name}) + "\n", 745 + encoding="utf-8", 746 + ) 747 + 748 + _header, record, agenerate = _drive_describe( 749 + monkeypatch, 750 + video_path, 751 + output_path, 752 + provider_result=(NO_BRAIN_PROVIDER, ""), 753 + ) 754 + (segment / ".analyze_failed_screen").write_text("{}\n", encoding="utf-8") 755 + 756 + _assert_processing_record( 757 + record, 758 + state=STATE_FAILED, 759 + reason_code=REASON_CORRUPT_INPUT, 760 + handler=HANDLER_DESCRIBE, 761 + ) 762 + assert agenerate.call_count == 0 763 + data_state = read_segment_data_state(DAY, segment_key) 764 + assert data_state == {"screen": DataState.FAILED_FINAL.value} 765 + assert segment_fully_sensed(data_state) is True 766 + 767 + progress = { 768 + (STREAM, segment_key): SegmentProgress( 769 + sensed=True, 770 + density="idle", 771 + change_class=None, 772 + dispatched=frozenset(), 773 + completed=frozenset(), 774 + unconfigured=frozenset(), 775 + capped=frozenset(), 776 + ) 777 + } 778 + completion = classify_segment_completion(cluster_segments(DAY), progress) 779 + assert completion.blockers == [] 780 + assert completion.exhausted == (segment_key,) 675 781 676 782 677 783 def test_no_video_stream_screen_terminalizes_corrupt_input( ··· 707 813 } 708 814 709 815 710 - def test_partial_decode_failure_preserves_qualified_count( 711 - segment_journal, 712 - monkeypatch, 713 - ): 816 + def _install_partial_decode_failure(monkeypatch, video_path: Path) -> None: 714 817 from fractions import Fraction 715 818 716 819 av = pytest.importorskip("av") ··· 756 859 yield frame 757 860 raise decode_error 758 861 862 + video_path.write_bytes(b"fake container bytes") 863 + monkeypatch.setattr(av, "open", lambda *args, **kwargs: FakeContainer()) 864 + monkeypatch.setattr(aruco, "detect_markers", lambda _image: None) 865 + 866 + 867 + def test_partial_decode_failure_preserves_qualified_count( 868 + segment_journal, 869 + monkeypatch, 870 + ): 759 871 segment_key = "125000_300" 760 872 segment = _segment_dir(segment_journal, segment=segment_key) 761 873 video_path = segment / "screen.mp4" 762 874 output_path = segment / "screen.jsonl" 763 - video_path.write_bytes(b"fake container bytes") 764 - 765 - monkeypatch.setattr(av, "open", lambda *args, **kwargs: FakeContainer()) 766 - monkeypatch.setattr(aruco, "detect_markers", lambda _image: None) 875 + _install_partial_decode_failure(monkeypatch, video_path) 767 876 768 877 header, record, agenerate = _drive_describe( 769 878 monkeypatch, ··· 783 892 rows = _read_jsonl(output_path) 784 893 assert len(rows) == 2 785 894 assert rows[1]["frame_id"] == 1 895 + 896 + 897 + def test_partial_decode_failure_without_provider_reenters_recordless_output( 898 + segment_journal, 899 + monkeypatch, 900 + ): 901 + from solstone.think.models import NO_BRAIN_PROVIDER 902 + 903 + segment_key = "125500_300" 904 + segment = _segment_dir(segment_journal, segment=segment_key) 905 + video_path = segment / "screen.mp4" 906 + output_path = segment / "screen.jsonl" 907 + _install_partial_decode_failure(monkeypatch, video_path) 908 + output_path.write_text( 909 + json.dumps({"raw": video_path.name}) + "\n", 910 + encoding="utf-8", 911 + ) 912 + original_output = output_path.read_bytes() 913 + 914 + header, record, agenerate = _drive_describe( 915 + monkeypatch, 916 + video_path, 917 + output_path, 918 + provider_result=(NO_BRAIN_PROVIDER, ""), 919 + expect_record=False, 920 + ) 921 + 922 + assert header == {"raw": video_path.name} 923 + assert record is None 924 + assert output_path.read_bytes() == original_output 925 + assert agenerate.call_count == 0 926 + assert read_segment_data_state(DAY, segment_key) == { 927 + "screen": DataState.PENDING.value 928 + } 929 + assert ( 930 + should_reenter_analysis_output( 931 + record=record, 932 + output_path=output_path, 933 + handler=HANDLER_DESCRIBE, 934 + ) 935 + is True 936 + ) 786 937 787 938 788 939 def test_aruco_frame_body_index_error_still_propagates(
+367
tests/test_data_state.py
··· 1 1 # SPDX-License-Identifier: AGPL-3.0-only 2 2 # Copyright (c) 2026 sol pbc 3 3 4 + import json 4 5 import os 5 6 import time 6 7 7 8 from solstone.observe.processing_record import ( 8 9 FAILED_ATTEMPT_BOUND, 10 + HANDLER_DESCRIBE, 11 + HANDLER_TRANSCRIBE, 9 12 REASON_ANALYSIS_FAILED, 10 13 REASON_CORRUPT_INPUT, 14 + SCREEN_ANALYSIS_ROW_KEY, 15 + STATE_ANALYZED, 11 16 STATE_EMPTY, 12 17 STATE_FAILED, 13 18 is_failure_exhausted, 14 19 record_attempts, 20 + should_reenter_analysis_output, 15 21 ) 16 22 from solstone.think.data_state import ( 17 23 DataState, ··· 66 72 assert record_attempts({"attempts": None}) == 0 67 73 assert record_attempts({"attempts": "3"}) == 0 68 74 assert record_attempts({"attempts": True}) == 0 75 + 76 + 77 + def test_ac1_should_reenter_analysis_output_table(tmp_path) -> None: 78 + output_path = tmp_path / "screen.jsonl" 79 + 80 + def write_output(record: dict | None, rows: list[dict] | None = None) -> None: 81 + header = {"raw": "screen.webm"} 82 + if record is not None: 83 + header["_solstone_processing"] = record 84 + lines = [json.dumps(header)] 85 + if rows: 86 + lines.extend(json.dumps(row) for row in rows) 87 + output_path.write_text("\n".join(lines) + "\n", encoding="utf-8") 88 + 89 + cases = [ 90 + ( 91 + "retryable_describe_failure", 92 + { 93 + "state": STATE_FAILED, 94 + "handler": HANDLER_DESCRIBE, 95 + "reason_code": REASON_ANALYSIS_FAILED, 96 + "attempts": FAILED_ATTEMPT_BOUND - 1, 97 + }, 98 + None, 99 + HANDLER_DESCRIBE, 100 + True, 101 + ), 102 + ( 103 + "corrupt_describe_failure", 104 + { 105 + "state": STATE_FAILED, 106 + "handler": HANDLER_DESCRIBE, 107 + "reason_code": REASON_CORRUPT_INPUT, 108 + }, 109 + None, 110 + HANDLER_DESCRIBE, 111 + False, 112 + ), 113 + ( 114 + "exhausted_describe_failure", 115 + { 116 + "state": STATE_FAILED, 117 + "handler": HANDLER_DESCRIBE, 118 + "reason_code": REASON_ANALYSIS_FAILED, 119 + "attempts": FAILED_ATTEMPT_BOUND, 120 + }, 121 + None, 122 + HANDLER_DESCRIBE, 123 + False, 124 + ), 125 + ( 126 + "transcribe_failure", 127 + { 128 + "state": STATE_FAILED, 129 + "handler": HANDLER_TRANSCRIBE, 130 + "reason_code": REASON_ANALYSIS_FAILED, 131 + "attempts": 1, 132 + }, 133 + None, 134 + HANDLER_TRANSCRIBE, 135 + False, 136 + ), 137 + ("recordless_screen_no_rows", None, None, HANDLER_DESCRIBE, True), 138 + ( 139 + "recordless_screen_with_rows", 140 + None, 141 + [{"frame_id": 1, SCREEN_ANALYSIS_ROW_KEY: 0.0}], 142 + HANDLER_DESCRIBE, 143 + False, 144 + ), 145 + ("recordless_audio_no_rows", None, None, HANDLER_TRANSCRIBE, False), 146 + ] 147 + 148 + for _name, record, rows, handler, expected in cases: 149 + write_output(record, rows) 150 + assert ( 151 + should_reenter_analysis_output( 152 + record=record, 153 + output_path=output_path, 154 + handler=handler, 155 + ) 156 + is expected 157 + ) 158 + 159 + 160 + def test_ac8_screen_disagreement_table_has_no_silent_nonterminal_wedge( 161 + tmp_path, 162 + ) -> None: 163 + terminal_states = { 164 + DataState.ANALYZED.value, 165 + DataState.EMPTY.value, 166 + DataState.FAILED_FINAL.value, 167 + } 168 + retryable_record = { 169 + "state": STATE_FAILED, 170 + "handler": HANDLER_DESCRIBE, 171 + "reason_code": REASON_ANALYSIS_FAILED, 172 + "attempts": 1, 173 + } 174 + final_record = { 175 + "state": STATE_FAILED, 176 + "handler": HANDLER_DESCRIBE, 177 + "reason_code": REASON_CORRUPT_INPUT, 178 + } 179 + empty_record = {"state": STATE_EMPTY, "handler": HANDLER_DESCRIBE} 180 + analyzed_record = {"state": STATE_ANALYZED, "handler": HANDLER_DESCRIBE} 181 + row = {"frame_id": 1, SCREEN_ANALYSIS_ROW_KEY: 0.0} 182 + cases = [ 183 + { 184 + "name": "no_jsonl_pending_open_row", 185 + "jsonl": False, 186 + "record": None, 187 + "rows": False, 188 + "marker": False, 189 + "state": DataState.PENDING.value, 190 + "reentry": True, 191 + }, 192 + { 193 + "name": "no_jsonl_failed_marker_open_row", 194 + "jsonl": False, 195 + "record": None, 196 + "rows": False, 197 + "marker": True, 198 + "state": DataState.FAILED.value, 199 + "reentry": True, 200 + }, 201 + { 202 + "name": "no_jsonl_rows_impossible", 203 + "jsonl": False, 204 + "record": None, 205 + "rows": True, 206 + "marker": False, 207 + "impossible": "analyzed rows require a JSONL file", 208 + }, 209 + { 210 + "name": "no_jsonl_rows_marker_impossible", 211 + "jsonl": False, 212 + "record": None, 213 + "rows": True, 214 + "marker": True, 215 + "impossible": "analyzed rows require a JSONL file", 216 + }, 217 + { 218 + "name": "no_jsonl_record_impossible", 219 + "jsonl": False, 220 + "record": final_record, 221 + "rows": False, 222 + "marker": False, 223 + "impossible": "a processing record requires a JSONL header", 224 + }, 225 + { 226 + "name": "no_jsonl_record_marker_impossible", 227 + "jsonl": False, 228 + "record": final_record, 229 + "rows": False, 230 + "marker": True, 231 + "impossible": "a processing record requires a JSONL header", 232 + }, 233 + { 234 + "name": "no_jsonl_record_rows_impossible", 235 + "jsonl": False, 236 + "record": final_record, 237 + "rows": True, 238 + "marker": False, 239 + "impossible": "recorded analyzed rows require a JSONL file", 240 + }, 241 + { 242 + "name": "no_jsonl_record_rows_marker_impossible", 243 + "jsonl": False, 244 + "record": final_record, 245 + "rows": True, 246 + "marker": True, 247 + "impossible": "recorded analyzed rows require a JSONL file", 248 + }, 249 + { 250 + "name": "header_only_pending", 251 + "jsonl": True, 252 + "record": None, 253 + "rows": False, 254 + "marker": False, 255 + "state": DataState.PENDING.value, 256 + "reentry": True, 257 + }, 258 + { 259 + "name": "header_only_failed_marker", 260 + "jsonl": True, 261 + "record": None, 262 + "rows": False, 263 + "marker": True, 264 + "state": DataState.FAILED.value, 265 + "reentry": True, 266 + }, 267 + { 268 + "name": "recordless_rows_analyzed", 269 + "jsonl": True, 270 + "record": None, 271 + "rows": True, 272 + "marker": False, 273 + "state": DataState.ANALYZED.value, 274 + "reentry": False, 275 + }, 276 + { 277 + "name": "recordless_rows_marker_analyzed", 278 + "jsonl": True, 279 + "record": None, 280 + "rows": True, 281 + "marker": True, 282 + "state": DataState.ANALYZED.value, 283 + "reentry": False, 284 + }, 285 + { 286 + "name": "retryable_record_no_rows", 287 + "jsonl": True, 288 + "record": retryable_record, 289 + "rows": False, 290 + "marker": False, 291 + "state": DataState.FAILED.value, 292 + "reentry": True, 293 + }, 294 + { 295 + "name": "retryable_record_marker_no_rows", 296 + "jsonl": True, 297 + "record": retryable_record, 298 + "rows": False, 299 + "marker": True, 300 + "state": DataState.FAILED.value, 301 + "reentry": True, 302 + }, 303 + { 304 + "name": "retryable_record_rows", 305 + "jsonl": True, 306 + "record": retryable_record, 307 + "rows": True, 308 + "marker": False, 309 + "state": DataState.FAILED.value, 310 + "reentry": True, 311 + }, 312 + { 313 + "name": "retryable_record_marker_rows", 314 + "jsonl": True, 315 + "record": retryable_record, 316 + "rows": True, 317 + "marker": True, 318 + "state": DataState.FAILED.value, 319 + "reentry": True, 320 + }, 321 + { 322 + "name": "final_record_no_rows", 323 + "jsonl": True, 324 + "record": final_record, 325 + "rows": False, 326 + "marker": False, 327 + "state": DataState.FAILED_FINAL.value, 328 + "reentry": False, 329 + }, 330 + { 331 + "name": "final_record_marker_no_rows", 332 + "jsonl": True, 333 + "record": final_record, 334 + "rows": False, 335 + "marker": True, 336 + "state": DataState.FAILED_FINAL.value, 337 + "reentry": False, 338 + }, 339 + { 340 + "name": "empty_record_no_rows", 341 + "jsonl": True, 342 + "record": empty_record, 343 + "rows": False, 344 + "marker": False, 345 + "state": DataState.EMPTY.value, 346 + "reentry": False, 347 + }, 348 + { 349 + "name": "empty_record_marker_no_rows", 350 + "jsonl": True, 351 + "record": empty_record, 352 + "rows": False, 353 + "marker": True, 354 + "state": DataState.EMPTY.value, 355 + "reentry": False, 356 + }, 357 + { 358 + "name": "empty_record_rows_impossible", 359 + "jsonl": True, 360 + "record": empty_record, 361 + "rows": True, 362 + "marker": False, 363 + "impossible": "empty verdicts contain no analyzed screen rows", 364 + }, 365 + { 366 + "name": "analyzed_record_rows", 367 + "jsonl": True, 368 + "record": analyzed_record, 369 + "rows": True, 370 + "marker": False, 371 + "state": DataState.ANALYZED.value, 372 + "reentry": False, 373 + }, 374 + { 375 + "name": "analyzed_record_marker_rows", 376 + "jsonl": True, 377 + "record": analyzed_record, 378 + "rows": True, 379 + "marker": True, 380 + "state": DataState.ANALYZED.value, 381 + "reentry": False, 382 + }, 383 + { 384 + "name": "analyzed_record_no_rows_impossible", 385 + "jsonl": True, 386 + "record": analyzed_record, 387 + "rows": False, 388 + "marker": False, 389 + "impossible": "analyzed verdicts are emitted with analyzed rows", 390 + }, 391 + ] 392 + 393 + for index, case in enumerate(cases): 394 + if case.get("impossible"): 395 + assert isinstance(case["impossible"], str) 396 + continue 397 + assert case["jsonl"] or not case["record"] 398 + 399 + segment = tmp_path / f"0900{index:02d}_300" 400 + segment.mkdir() 401 + output_path = segment / "screen.jsonl" 402 + if case["marker"]: 403 + (segment / ".analyze_failed_screen").write_text("{}\n", encoding="utf-8") 404 + if case["jsonl"]: 405 + header = {"raw": "screen.webm"} 406 + record = case["record"] 407 + if record is not None: 408 + header["_solstone_processing"] = record 409 + lines = [json.dumps(header)] 410 + if case["rows"]: 411 + lines.append(json.dumps(row)) 412 + output_path.write_text("\n".join(lines) + "\n", encoding="utf-8") 413 + 414 + record = case["record"] 415 + state = derive_modality_state( 416 + segment, 417 + "screen", 418 + has_chunks=bool(case["rows"]), 419 + has_jsonl=bool(case["jsonl"]), 420 + has_raw=True, 421 + record=record, 422 + ) 423 + assert state == case["state"], case["name"] 424 + if case["jsonl"]: 425 + reentry = should_reenter_analysis_output( 426 + record=record, 427 + output_path=output_path, 428 + handler=HANDLER_DESCRIBE, 429 + ) 430 + else: 431 + # no JSONL + handler exits nonzero without output -> non-terminal but 432 + # re-attempted each cycle, a visible blocker rather than a silent wedge. 433 + reentry = True 434 + assert reentry is case["reentry"], case["name"] 435 + assert not (state not in terminal_states and not reentry), case["name"] 69 436 70 437 71 438 def test_derive_chunks_win_beats_processing_record(tmp_path) -> None:
+2 -2
tests/test_describe_promote.py
··· 1481 1481 False, 1482 1482 None, 1483 1483 ), 1484 - ("no_record", None, False, None), 1484 + ("no_record", None, True, 0), 1485 1485 ( 1486 1486 "corrupt_input", 1487 1487 { ··· 1576 1576 else: 1577 1577 assert constructed == [] 1578 1578 assert observed_previous_attempts == [] 1579 - assert output_path.read_bytes() == original 1579 + assert output_path.read_bytes() == original 1580 1580 1581 1581 1582 1582 @pytest.mark.asyncio
+67 -7
tests/test_sense.py
··· 27 27 REASON_ANALYSIS_FAILED, 28 28 REASON_CORRUPT_INPUT, 29 29 REASON_OK, 30 + SCREEN_ANALYSIS_ROW_KEY, 30 31 STATE_ANALYZED, 31 32 STATE_EMPTY, 32 33 STATE_FAILED, ··· 156 157 def _write_processing_output( 157 158 media_path: Path, 158 159 record: dict | None, 160 + rows: list[dict] | None = None, 159 161 ) -> Path: 160 162 header = {"raw": media_path.name} 161 163 if record is not None: 162 164 header["_solstone_processing"] = record 163 165 output_path = media_path.with_suffix(".jsonl") 164 - output_path.write_text(json.dumps(header) + "\n", encoding="utf-8") 166 + lines = [json.dumps(header)] 167 + if rows is not None: 168 + lines.extend(json.dumps(row) for row in rows) 169 + output_path.write_text("\n".join(lines) + "\n", encoding="utf-8") 165 170 return output_path 166 171 167 172 ··· 811 816 ( 812 817 "143000_300", 813 818 _processing_record(state=STATE_ANALYZED, reason_code="ok"), 819 + None, 814 820 False, 815 821 ), 816 822 ( ··· 819 825 state=STATE_EMPTY, 820 826 reason_code="no_decodable_frames", 821 827 ), 828 + None, 829 + False, 830 + ), 831 + ("143002_300", None, None, True), 832 + ( 833 + "143008_300", 834 + None, 835 + [{"frame_id": 1, SCREEN_ANALYSIS_ROW_KEY: 0.0, "analysis": {}}], 822 836 False, 823 837 ), 824 - ("143002_300", None, False), 825 838 ( 826 839 "143003_300", 827 840 _processing_record( 828 841 state=STATE_FAILED, 829 842 reason_code=REASON_CORRUPT_INPUT, 830 843 ), 844 + None, 831 845 False, 832 846 ), 833 847 ( ··· 836 850 state=STATE_FAILED, 837 851 attempts=FAILED_ATTEMPT_BOUND, 838 852 ), 853 + None, 839 854 False, 840 855 ), 841 856 ( ··· 845 860 handler=HANDLER_TRANSCRIBE, 846 861 attempts=1, 847 862 ), 863 + None, 848 864 False, 849 865 ), 850 866 ( 851 867 "143006_300", 852 868 _processing_record(state=STATE_FAILED), 869 + None, 853 870 True, 854 871 ), 855 872 ( ··· 858 875 state=STATE_FAILED, 859 876 attempts=FAILED_ATTEMPT_BOUND - 1, 860 877 ), 878 + None, 861 879 True, 862 880 ), 863 881 ] 864 - for segment, record, _expected in cases: 882 + for segment, record, rows, _expected in cases: 865 883 media_path = make_segment_file(tmp_path, segment=segment) 866 - _write_processing_output(media_path, record) 884 + _write_processing_output(media_path, record, rows=rows) 867 885 868 886 sensor = FileSensor(tmp_path) 869 887 sensor.register("*.webm", "describe", ["journal", "describe", "{file}"]) ··· 876 894 877 895 sensor.process_day("20250101", max_jobs=1) 878 896 879 - assert processed == [segment for segment, _record, expected in cases if expected] 897 + assert processed == [case[0] for case in cases if case[-1]] 880 898 881 899 882 900 def test_process_day_reentry_uses_bounded_first_window(tmp_path, monkeypatch): 883 901 monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 884 - media_path = make_segment_file(tmp_path) 902 + media_path = make_segment_file(tmp_path, segment="143022_300") 885 903 record = _processing_record(state=STATE_FAILED) 886 904 output_path = media_path.with_suffix(".jsonl") 887 905 output_path.write_bytes( ··· 891 909 + json.dumps(record).encode("utf-8") 892 910 + b"}\n" 893 911 ) 912 + # A record past the bounded first-row window is indeterminate: it re-enters 913 + # unless analyzed rows provide evidence that the output is already useful. 914 + row_media_path = make_segment_file(tmp_path, segment="143023_300") 915 + row_output_path = row_media_path.with_suffix(".jsonl") 916 + row_output_path.write_bytes( 917 + b'{"pad":"' 918 + + (b"x" * MAX_FIRST_ROW_BYTES) 919 + + b'","_solstone_processing":' 920 + + json.dumps(record).encode("utf-8") 921 + + b"}\n" 922 + + json.dumps( 923 + {"frame_id": 1, SCREEN_ANALYSIS_ROW_KEY: 0.0, "analysis": {}} 924 + ).encode("utf-8") 925 + + b"\n" 926 + ) 894 927 895 928 sensor = FileSensor(tmp_path) 896 929 sensor.register("*.webm", "describe", ["journal", "describe", "{file}"]) ··· 903 936 904 937 sensor.process_day("20250101", max_jobs=1) 905 938 906 - assert processed == [] 939 + assert processed == [media_path] 940 + 941 + 942 + def test_ac5_recordless_audio_output_does_not_reenter(tmp_path, monkeypatch): 943 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 944 + audio_path = make_segment_file(tmp_path, filename="audio.flac") 945 + _write_processing_output(audio_path, None) 946 + 947 + sensor = FileSensor(tmp_path) 948 + command = ["journal", "transcribe", "{file}"] 949 + sensor.register("*.flac", "transcribe", command) 950 + 951 + to_process, _ = sensor.scan_unprocessed("20250101") 952 + 953 + assert to_process == [] 954 + 955 + 956 + def test_ac7_missing_screen_output_still_queues_raw_video(tmp_path, monkeypatch): 957 + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) 958 + media_path = make_segment_file(tmp_path) 959 + 960 + sensor = FileSensor(tmp_path) 961 + command = ["journal", "describe", "{file}"] 962 + sensor.register("*.webm", "describe", command) 963 + 964 + to_process, _ = sensor.scan_unprocessed("20250101") 965 + 966 + assert to_process == [(media_path, "describe", command)] 907 967 908 968 909 969 def test_process_day_retries_failed_describe_until_attempt_bound(tmp_path, monkeypatch):