diff --git a/solstone/observe/describe.py b/solstone/observe/describe.py index d5fb6c687..9935b2d67 100644 --- a/solstone/observe/describe.py +++ b/solstone/observe/describe.py @@ -51,7 +51,7 @@ from solstone.observe.processing_record import ( build_processing_record, read_processing_record_header, record_attempts, - should_reenter_failed_describe, + should_reenter_analysis_output, ) from solstone.observe.utils import get_segment_key, resize_for_vlm from solstone.think.callosum import callosum_send @@ -952,6 +952,14 @@ class VideoProcessor: print(result_line.rstrip("\n"), flush=True) try: + if not had_qualified_frames: + # Decoded frames are real work; do not discard them before description. + if self.decode_failed: + _promote(STATE_FAILED, REASON_CORRUPT_INPUT) + else: + _promote(STATE_EMPTY, REASON_NO_DECODABLE_FRAMES) + return + frame_provider, frame_model = resolve_provider("generate") if frame_provider == NO_BRAIN_PROVIDER: logger.info("No thinking engine selected; deferring frame description") @@ -1452,8 +1460,6 @@ class VideoProcessor: if self.decode_failed: state, reason_code = STATE_FAILED, REASON_CORRUPT_INPUT - elif not had_qualified_frames: - state, reason_code = STATE_EMPTY, REASON_NO_DECODABLE_FRAMES elif not emitted_row_has_error and emitted_frame_ids == qualified_ids: state, reason_code = STATE_ANALYZED, REASON_OK else: @@ -1556,7 +1562,11 @@ async def async_main(): # Skip if already processed (unless redo mode) if not args.redo and output_path.exists(): record = read_processing_record_header(output_path) - if not should_reenter_failed_describe(record): + if not should_reenter_analysis_output( + record=record, + output_path=output_path, + handler=HANDLER_DESCRIBE, + ): logger.info(f"Already processed: {video_path}") return previous_attempts = record_attempts(record) diff --git a/solstone/observe/processing_record.py b/solstone/observe/processing_record.py index 073c88d4a..34df778dc 100644 --- a/solstone/observe/processing_record.py +++ b/solstone/observe/processing_record.py @@ -1,16 +1,19 @@ # SPDX-License-Identifier: AGPL-3.0-only # Copyright (c) 2026 sol pbc -"""Shared `_solstone_processing` record vocabulary for media-analysis handlers. - -The screen (`describe`) and audio (`transcribe`) handlers stamp one of these -records into the metadata header (row 1) of the JSONL they already produce, so -a downstream reader lode can derive per-segment processing state without -re-deriving it from raw media. This module is the one authoritative source of -the closed state / reason_code / handler / schema vocabulary; neither handler -may carry these literals inline. Failed describe records may also carry an -``attempts`` counter; absent attempts means 0, and -``FAILED_ATTEMPT_BOUND`` is the shared retry exhaustion bound. +"""Shared evidence vocabulary for media-analysis outputs. + +This module is the source of truth for the two forms of output evidence: +``_solstone_processing`` metadata-header records and the JSONL row keys that +prove audio or screen analysis rows exist. The screen (``describe``) and audio +(``transcribe``) handlers stamp processing records into the metadata header of +the JSONL they produce, while row-key detection gives bounded evidence for +legacy or record-less outputs. + +``FileSensor.scan_unprocessed``, ``describe.async_main``, and +``derive_modality_state`` consume this vocabulary so capture re-entry, +describe-side skipping, and downstream state derivation share the same reading +of whether an output is useful, terminal, retryable, or indeterminate. """ import json @@ -21,6 +24,8 @@ SCHEMA = "solstone.processing.v1" FAILED_ATTEMPT_BOUND = 3 ATTEMPTS_KEY = "attempts" MAX_FIRST_ROW_BYTES = 64 * 1024 +SCREEN_ANALYSIS_ROW_KEY = "timestamp" +AUDIO_TRANSCRIPT_ROW_KEY = "start" # state values (closed set) STATE_ANALYZED = "analyzed" @@ -84,13 +89,55 @@ def read_processing_record_header(path: Path) -> dict | None: return record if isinstance(record, dict) else None -def should_reenter_failed_describe(record: dict | None) -> bool: - """Return whether an existing failed describe output should be retried.""" - return ( +def jsonl_has_row_with_key(path: Path, row_key: str) -> bool: + """Return whether an early JSONL object has the row key. + + Key-based membership test on at most the first two nonblank lines. + """ + try: + lines = [] + with path.open("r", encoding="utf-8") as handle: + for line in handle: + if not line.strip(): + continue + lines.append(line) + if len(lines) == 2: + break + except OSError: + return False + for line in lines: + try: + parsed = json.loads(line) + except (json.JSONDecodeError, ValueError): + continue + if isinstance(parsed, dict) and row_key in parsed: + return True + return False + + +# an existing analysis output only blocks re-entry when it actually carries +# evidence — analyzed rows or a processing record. An output with neither is +# indeterminate, and indeterminate work re-enters until the record ledger can +# govern it. Decode-determined verdicts terminalize regardless of provider +# availability. +def should_reenter_analysis_output( + *, + record: dict | None, + output_path: Path, + handler: str, +) -> bool: + """Return whether an existing analysis output should be retried.""" + if ( isinstance(record, dict) and record.get("state") == STATE_FAILED and record.get("handler") == HANDLER_DESCRIBE and not is_failure_exhausted(record) + ): + return True + return ( + record is None + and handler == HANDLER_DESCRIBE + and not jsonl_has_row_with_key(output_path, SCREEN_ANALYSIS_ROW_KEY) ) diff --git a/solstone/observe/sense.py b/solstone/observe/sense.py index 2dab3ea0a..2d47771e9 100644 --- a/solstone/observe/sense.py +++ b/solstone/observe/sense.py @@ -27,7 +27,7 @@ from solstone import __version__ from solstone.observe.exit_codes import EXIT_PROVIDER_BLOCKED, WATCHDOG_TIMEOUT from solstone.observe.processing_record import ( read_processing_record_header, - should_reenter_failed_describe, + should_reenter_analysis_output, ) from solstone.observe.utils import ( AUDIO_EXTENSIONS, @@ -1048,16 +1048,19 @@ class FileSensor: if modality_filter == "screen" and suffix not in VIDEO_EXTENSIONS: continue - # Check if output JSONL exists (already processed) - output_path = file_path.with_suffix(".jsonl") - if output_path.exists(): - record = read_processing_record_header(output_path) - if not should_reenter_failed_describe(record): - continue - handler_info = self._match_pattern(file_path) if handler_info: handler_name, command = handler_info + # Check if output JSONL exists (already processed) + output_path = file_path.with_suffix(".jsonl") + if output_path.exists(): + record = read_processing_record_header(output_path) + if not should_reenter_analysis_output( + record=record, + output_path=output_path, + handler=handler_name, + ): + continue if handler_name == "depict" and is_import_stream(stream_name): continue to_process.append((file_path, handler_name, command)) diff --git a/solstone/think/cluster.py b/solstone/think/cluster.py index b9a9b6a69..e69d00465 100644 --- a/solstone/think/cluster.py +++ b/solstone/think/cluster.py @@ -11,6 +11,11 @@ from datetime import datetime, timedelta from pathlib import Path from typing import Any +from solstone.observe.processing_record import ( + AUDIO_TRANSCRIPT_ROW_KEY, + SCREEN_ANALYSIS_ROW_KEY, + jsonl_has_row_with_key, +) from solstone.observe.screen import format_screen_text from solstone.think.browser_formatter import format_browser_text from solstone.think.data_state import ( @@ -478,32 +483,6 @@ def _slots_to_ranges(slots: list[datetime]) -> list[tuple[str, str]]: return ranges -def _jsonl_has_marker_row(path: Path, marker_key: str) -> bool: - """Return whether an early JSONL object has the marker key. - - Key-based membership test on at most the first two nonblank lines. - """ - try: - lines = [] - with path.open("r", encoding="utf-8") as handle: - for line in handle: - if not line.strip(): - continue - lines.append(line) - if len(lines) == 2: - break - except OSError: - return False - for line in lines: - try: - parsed = json.loads(line) - except (json.JSONDecodeError, ValueError): - continue - if isinstance(parsed, dict) and marker_key in parsed: - return True - return False - - def _read_processing_record(jsonl_files: list[Path]) -> dict | None: """Return the first processing header record from sorted JSONL files.""" for path in jsonl_files: @@ -593,7 +572,8 @@ def _detect_data_state(seg_path: Path) -> dict[str, str]: ) audio_md_files = _markdown_transcript_files(seg_path) audio_analyzed = any( - _jsonl_has_marker_row(path, "start") for path in audio_jsonl_files + jsonl_has_row_with_key(path, AUDIO_TRANSCRIPT_ROW_KEY) + for path in audio_jsonl_files ) or any(_has_nonempty_text(path) for path in audio_md_files) audio_record = _read_processing_record(audio_jsonl_files) audio_has_raw = _has_raw_media(raw_media_paths, AUDIO_EXTENSIONS) @@ -617,7 +597,8 @@ def _detect_data_state(seg_path: Path) -> dict[str, str]: } ) screen_analyzed = any( - _jsonl_has_marker_row(path, "timestamp") for path in screen_jsonl_files + jsonl_has_row_with_key(path, SCREEN_ANALYSIS_ROW_KEY) + for path in screen_jsonl_files ) screen_record = _read_processing_record(screen_jsonl_files) screen_has_raw = _has_raw_media(raw_media_paths, VIDEO_EXTENSIONS) diff --git a/tests/test_bad_media_corpus.py b/tests/test_bad_media_corpus.py index f2fadc1dd..45f8b26ef 100644 --- a/tests/test_bad_media_corpus.py +++ b/tests/test_bad_media_corpus.py @@ -35,6 +35,8 @@ from solstone.observe.processing_record import ( STATE_ANALYZED, STATE_EMPTY, STATE_FAILED, + is_failure_exhausted, + should_reenter_analysis_output, ) from solstone.observe.utils import SAMPLE_RATE, AudioDecodeError from solstone.observe.vad import VadResult @@ -44,8 +46,10 @@ from solstone.think.cluster import ( ) from solstone.think.data_state import DataState from solstone.think.pipeline_health import ( + SegmentProgress, classify_segment_completion, read_segment_progress, + segment_fully_sensed, ) DAY = "20990501" @@ -290,7 +294,9 @@ def _drive_describe( agenerate_response: str = "{}", agenerate_finish_reason: str = "stop", expect_runtime_error: bool = False, -) -> tuple[dict[str, Any], dict[str, Any], AsyncMock]: + provider_result: tuple[str, str] = ("google", "gemini-test"), + expect_record: bool = True, +) -> tuple[dict[str, Any], dict[str, Any] | None, AsyncMock]: from solstone.observe import describe, processing_record agenerate = AsyncMock( @@ -298,7 +304,7 @@ def _drive_describe( ) monkeypatch.setattr( "solstone.think.models.resolve_provider", - lambda _interface: ("google", "gemini-test"), + lambda _interface: provider_result, ) monkeypatch.setattr(describe, "callosum_send", lambda *args, **kwargs: None) monkeypatch.setattr(describe, "select_frames_for_extraction", lambda *a, **k: []) @@ -325,8 +331,11 @@ def _drive_describe( ) header = _read_header(output_path) - record = header["_solstone_processing"] - assert isinstance(record, dict) + record = header.get("_solstone_processing") + if expect_record: + assert isinstance(record, dict) + else: + assert record is None return header, record, agenerate @@ -641,7 +650,6 @@ def test_eof_truncated_screen_terminalizes_corrupt_input( monkeypatch, ): av = pytest.importorskip("av") - from solstone.observe.describe import should_reenter_failed_describe segment = _segment_dir(segment_journal) video_path = segment / "screen.webm" @@ -671,7 +679,105 @@ def test_eof_truncated_screen_terminalizes_corrupt_input( assert read_segment_data_state(DAY, SEGMENT) == { "screen": DataState.FAILED_FINAL.value } - assert should_reenter_failed_describe(record) is False + assert ( + should_reenter_analysis_output( + record=record, + output_path=output_path, + handler=HANDLER_DESCRIBE, + ) + is False + ) + + +def test_ac4_no_provider_truncated_recordless_screen_converges_corrupt_input( + segment_journal, + monkeypatch, +): + from solstone.think.models import NO_BRAIN_PROVIDER + + segment = _segment_dir(segment_journal, segment="123500_300") + video_path = segment / "screen.webm" + output_path = segment / "screen.jsonl" + _build_truncated_webm(video_path) + output_path.write_text( + json.dumps({"raw": video_path.name}) + "\n", + encoding="utf-8", + ) + + _header, record, agenerate = _drive_describe( + monkeypatch, + video_path, + output_path, + provider_result=(NO_BRAIN_PROVIDER, ""), + ) + + _assert_processing_record( + record, + state=STATE_FAILED, + reason_code=REASON_CORRUPT_INPUT, + handler=HANDLER_DESCRIBE, + ) + assert agenerate.call_count == 0 + assert is_failure_exhausted(record) is True + assert ( + should_reenter_analysis_output( + record=record, + output_path=output_path, + handler=HANDLER_DESCRIBE, + ) + is False + ) + + +def test_ac6_failed_final_screen_record_unblocks_day_with_failed_marker( + segment_journal, + monkeypatch, +): + from solstone.think.models import NO_BRAIN_PROVIDER + + segment_key = "123600_300" + segment = _segment_dir(segment_journal, segment=segment_key) + video_path = segment / "screen.webm" + output_path = segment / "screen.jsonl" + _build_truncated_webm(video_path) + output_path.write_text( + json.dumps({"raw": video_path.name}) + "\n", + encoding="utf-8", + ) + + _header, record, agenerate = _drive_describe( + monkeypatch, + video_path, + output_path, + provider_result=(NO_BRAIN_PROVIDER, ""), + ) + (segment / ".analyze_failed_screen").write_text("{}\n", encoding="utf-8") + + _assert_processing_record( + record, + state=STATE_FAILED, + reason_code=REASON_CORRUPT_INPUT, + handler=HANDLER_DESCRIBE, + ) + assert agenerate.call_count == 0 + data_state = read_segment_data_state(DAY, segment_key) + assert data_state == {"screen": DataState.FAILED_FINAL.value} + assert segment_fully_sensed(data_state) is True + + progress = { + (STREAM, segment_key): SegmentProgress( + sensed=True, + density="idle", + change_class=None, + dispatched=frozenset(), + completed=frozenset(), + unconfigured=frozenset(), + capped=frozenset(), + ) + } + completion = classify_segment_completion(cluster_segments(DAY), progress) + assert completion.blockers == [] + assert completion.exhausted == (segment_key,) def test_no_video_stream_screen_terminalizes_corrupt_input( @@ -707,10 +813,7 @@ def test_no_video_stream_screen_terminalizes_corrupt_input( } -def test_partial_decode_failure_preserves_qualified_count( - segment_journal, - monkeypatch, -): +def _install_partial_decode_failure(monkeypatch, video_path: Path) -> None: from fractions import Fraction av = pytest.importorskip("av") @@ -756,14 +859,20 @@ def test_partial_decode_failure_preserves_qualified_count( yield frame raise decode_error + video_path.write_bytes(b"fake container bytes") + monkeypatch.setattr(av, "open", lambda *args, **kwargs: FakeContainer()) + monkeypatch.setattr(aruco, "detect_markers", lambda _image: None) + + +def test_partial_decode_failure_preserves_qualified_count( + segment_journal, + monkeypatch, +): segment_key = "125000_300" segment = _segment_dir(segment_journal, segment=segment_key) video_path = segment / "screen.mp4" output_path = segment / "screen.jsonl" - video_path.write_bytes(b"fake container bytes") - - monkeypatch.setattr(av, "open", lambda *args, **kwargs: FakeContainer()) - monkeypatch.setattr(aruco, "detect_markers", lambda _image: None) + _install_partial_decode_failure(monkeypatch, video_path) header, record, agenerate = _drive_describe( monkeypatch, @@ -785,6 +894,48 @@ def test_partial_decode_failure_preserves_qualified_count( assert rows[1]["frame_id"] == 1 +def test_partial_decode_failure_without_provider_reenters_recordless_output( + segment_journal, + monkeypatch, +): + from solstone.think.models import NO_BRAIN_PROVIDER + + segment_key = "125500_300" + segment = _segment_dir(segment_journal, segment=segment_key) + video_path = segment / "screen.mp4" + output_path = segment / "screen.jsonl" + _install_partial_decode_failure(monkeypatch, video_path) + output_path.write_text( + json.dumps({"raw": video_path.name}) + "\n", + encoding="utf-8", + ) + original_output = output_path.read_bytes() + + header, record, agenerate = _drive_describe( + monkeypatch, + video_path, + output_path, + provider_result=(NO_BRAIN_PROVIDER, ""), + expect_record=False, + ) + + assert header == {"raw": video_path.name} + assert record is None + assert output_path.read_bytes() == original_output + assert agenerate.call_count == 0 + assert read_segment_data_state(DAY, segment_key) == { + "screen": DataState.PENDING.value + } + assert ( + should_reenter_analysis_output( + record=record, + output_path=output_path, + handler=HANDLER_DESCRIBE, + ) + is True + ) + + def test_aruco_frame_body_index_error_still_propagates( segment_journal, monkeypatch, diff --git a/tests/test_data_state.py b/tests/test_data_state.py index 97cecb436..8ec190bba 100644 --- a/tests/test_data_state.py +++ b/tests/test_data_state.py @@ -1,17 +1,23 @@ # SPDX-License-Identifier: AGPL-3.0-only # Copyright (c) 2026 sol pbc +import json import os import time from solstone.observe.processing_record import ( FAILED_ATTEMPT_BOUND, + HANDLER_DESCRIBE, + HANDLER_TRANSCRIBE, REASON_ANALYSIS_FAILED, REASON_CORRUPT_INPUT, + SCREEN_ANALYSIS_ROW_KEY, + STATE_ANALYZED, STATE_EMPTY, STATE_FAILED, is_failure_exhausted, record_attempts, + should_reenter_analysis_output, ) from solstone.think.data_state import ( DataState, @@ -68,6 +74,367 @@ def test_record_attempts_coerces_absent_or_malformed_to_zero() -> None: assert record_attempts({"attempts": True}) == 0 +def test_ac1_should_reenter_analysis_output_table(tmp_path) -> None: + output_path = tmp_path / "screen.jsonl" + + def write_output(record: dict | None, rows: list[dict] | None = None) -> None: + header = {"raw": "screen.webm"} + if record is not None: + header["_solstone_processing"] = record + lines = [json.dumps(header)] + if rows: + lines.extend(json.dumps(row) for row in rows) + output_path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + cases = [ + ( + "retryable_describe_failure", + { + "state": STATE_FAILED, + "handler": HANDLER_DESCRIBE, + "reason_code": REASON_ANALYSIS_FAILED, + "attempts": FAILED_ATTEMPT_BOUND - 1, + }, + None, + HANDLER_DESCRIBE, + True, + ), + ( + "corrupt_describe_failure", + { + "state": STATE_FAILED, + "handler": HANDLER_DESCRIBE, + "reason_code": REASON_CORRUPT_INPUT, + }, + None, + HANDLER_DESCRIBE, + False, + ), + ( + "exhausted_describe_failure", + { + "state": STATE_FAILED, + "handler": HANDLER_DESCRIBE, + "reason_code": REASON_ANALYSIS_FAILED, + "attempts": FAILED_ATTEMPT_BOUND, + }, + None, + HANDLER_DESCRIBE, + False, + ), + ( + "transcribe_failure", + { + "state": STATE_FAILED, + "handler": HANDLER_TRANSCRIBE, + "reason_code": REASON_ANALYSIS_FAILED, + "attempts": 1, + }, + None, + HANDLER_TRANSCRIBE, + False, + ), + ("recordless_screen_no_rows", None, None, HANDLER_DESCRIBE, True), + ( + "recordless_screen_with_rows", + None, + [{"frame_id": 1, SCREEN_ANALYSIS_ROW_KEY: 0.0}], + HANDLER_DESCRIBE, + False, + ), + ("recordless_audio_no_rows", None, None, HANDLER_TRANSCRIBE, False), + ] + + for _name, record, rows, handler, expected in cases: + write_output(record, rows) + assert ( + should_reenter_analysis_output( + record=record, + output_path=output_path, + handler=handler, + ) + is expected + ) + + +def test_ac8_screen_disagreement_table_has_no_silent_nonterminal_wedge( + tmp_path, +) -> None: + terminal_states = { + DataState.ANALYZED.value, + DataState.EMPTY.value, + DataState.FAILED_FINAL.value, + } + retryable_record = { + "state": STATE_FAILED, + "handler": HANDLER_DESCRIBE, + "reason_code": REASON_ANALYSIS_FAILED, + "attempts": 1, + } + final_record = { + "state": STATE_FAILED, + "handler": HANDLER_DESCRIBE, + "reason_code": REASON_CORRUPT_INPUT, + } + empty_record = {"state": STATE_EMPTY, "handler": HANDLER_DESCRIBE} + analyzed_record = {"state": STATE_ANALYZED, "handler": HANDLER_DESCRIBE} + row = {"frame_id": 1, SCREEN_ANALYSIS_ROW_KEY: 0.0} + cases = [ + { + "name": "no_jsonl_pending_open_row", + "jsonl": False, + "record": None, + "rows": False, + "marker": False, + "state": DataState.PENDING.value, + "reentry": True, + }, + { + "name": "no_jsonl_failed_marker_open_row", + "jsonl": False, + "record": None, + "rows": False, + "marker": True, + "state": DataState.FAILED.value, + "reentry": True, + }, + { + "name": "no_jsonl_rows_impossible", + "jsonl": False, + "record": None, + "rows": True, + "marker": False, + "impossible": "analyzed rows require a JSONL file", + }, + { + "name": "no_jsonl_rows_marker_impossible", + "jsonl": False, + "record": None, + "rows": True, + "marker": True, + "impossible": "analyzed rows require a JSONL file", + }, + { + "name": "no_jsonl_record_impossible", + "jsonl": False, + "record": final_record, + "rows": False, + "marker": False, + "impossible": "a processing record requires a JSONL header", + }, + { + "name": "no_jsonl_record_marker_impossible", + "jsonl": False, + "record": final_record, + "rows": False, + "marker": True, + "impossible": "a processing record requires a JSONL header", + }, + { + "name": "no_jsonl_record_rows_impossible", + "jsonl": False, + "record": final_record, + "rows": True, + "marker": False, + "impossible": "recorded analyzed rows require a JSONL file", + }, + { + "name": "no_jsonl_record_rows_marker_impossible", + "jsonl": False, + "record": final_record, + "rows": True, + "marker": True, + "impossible": "recorded analyzed rows require a JSONL file", + }, + { + "name": "header_only_pending", + "jsonl": True, + "record": None, + "rows": False, + "marker": False, + "state": DataState.PENDING.value, + "reentry": True, + }, + { + "name": "header_only_failed_marker", + "jsonl": True, + "record": None, + "rows": False, + "marker": True, + "state": DataState.FAILED.value, + "reentry": True, + }, + { + "name": "recordless_rows_analyzed", + "jsonl": True, + "record": None, + "rows": True, + "marker": False, + "state": DataState.ANALYZED.value, + "reentry": False, + }, + { + "name": "recordless_rows_marker_analyzed", + "jsonl": True, + "record": None, + "rows": True, + "marker": True, + "state": DataState.ANALYZED.value, + "reentry": False, + }, + { + "name": "retryable_record_no_rows", + "jsonl": True, + "record": retryable_record, + "rows": False, + "marker": False, + "state": DataState.FAILED.value, + "reentry": True, + }, + { + "name": "retryable_record_marker_no_rows", + "jsonl": True, + "record": retryable_record, + "rows": False, + "marker": True, + "state": DataState.FAILED.value, + "reentry": True, + }, + { + "name": "retryable_record_rows", + "jsonl": True, + "record": retryable_record, + "rows": True, + "marker": False, + "state": DataState.FAILED.value, + "reentry": True, + }, + { + "name": "retryable_record_marker_rows", + "jsonl": True, + "record": retryable_record, + "rows": True, + "marker": True, + "state": DataState.FAILED.value, + "reentry": True, + }, + { + "name": "final_record_no_rows", + "jsonl": True, + "record": final_record, + "rows": False, + "marker": False, + "state": DataState.FAILED_FINAL.value, + "reentry": False, + }, + { + "name": "final_record_marker_no_rows", + "jsonl": True, + "record": final_record, + "rows": False, + "marker": True, + "state": DataState.FAILED_FINAL.value, + "reentry": False, + }, + { + "name": "empty_record_no_rows", + "jsonl": True, + "record": empty_record, + "rows": False, + "marker": False, + "state": DataState.EMPTY.value, + "reentry": False, + }, + { + "name": "empty_record_marker_no_rows", + "jsonl": True, + "record": empty_record, + "rows": False, + "marker": True, + "state": DataState.EMPTY.value, + "reentry": False, + }, + { + "name": "empty_record_rows_impossible", + "jsonl": True, + "record": empty_record, + "rows": True, + "marker": False, + "impossible": "empty verdicts contain no analyzed screen rows", + }, + { + "name": "analyzed_record_rows", + "jsonl": True, + "record": analyzed_record, + "rows": True, + "marker": False, + "state": DataState.ANALYZED.value, + "reentry": False, + }, + { + "name": "analyzed_record_marker_rows", + "jsonl": True, + "record": analyzed_record, + "rows": True, + "marker": True, + "state": DataState.ANALYZED.value, + "reentry": False, + }, + { + "name": "analyzed_record_no_rows_impossible", + "jsonl": True, + "record": analyzed_record, + "rows": False, + "marker": False, + "impossible": "analyzed verdicts are emitted with analyzed rows", + }, + ] + + for index, case in enumerate(cases): + if case.get("impossible"): + assert isinstance(case["impossible"], str) + continue + assert case["jsonl"] or not case["record"] + + segment = tmp_path / f"0900{index:02d}_300" + segment.mkdir() + output_path = segment / "screen.jsonl" + if case["marker"]: + (segment / ".analyze_failed_screen").write_text("{}\n", encoding="utf-8") + if case["jsonl"]: + header = {"raw": "screen.webm"} + record = case["record"] + if record is not None: + header["_solstone_processing"] = record + lines = [json.dumps(header)] + if case["rows"]: + lines.append(json.dumps(row)) + output_path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + record = case["record"] + state = derive_modality_state( + segment, + "screen", + has_chunks=bool(case["rows"]), + has_jsonl=bool(case["jsonl"]), + has_raw=True, + record=record, + ) + assert state == case["state"], case["name"] + if case["jsonl"]: + reentry = should_reenter_analysis_output( + record=record, + output_path=output_path, + handler=HANDLER_DESCRIBE, + ) + else: + # no JSONL + handler exits nonzero without output -> non-terminal but + # re-attempted each cycle, a visible blocker rather than a silent wedge. + reentry = True + assert reentry is case["reentry"], case["name"] + assert not (state not in terminal_states and not reentry), case["name"] + + def test_derive_chunks_win_beats_processing_record(tmp_path) -> None: segment = tmp_path / "090000_300" segment.mkdir() diff --git a/tests/test_describe_promote.py b/tests/test_describe_promote.py index 5a56dcf4d..688ab7cae 100644 --- a/tests/test_describe_promote.py +++ b/tests/test_describe_promote.py @@ -1481,7 +1481,7 @@ async def test_failed_promote_increments_previous_attempts(tmp_path, monkeypatch False, None, ), - ("no_record", None, False, None), + ("no_record", None, True, 0), ( "corrupt_input", { @@ -1576,7 +1576,7 @@ async def test_existing_output_reenters_only_retryable_describe_failures( else: assert constructed == [] assert observed_previous_attempts == [] - assert output_path.read_bytes() == original + assert output_path.read_bytes() == original @pytest.mark.asyncio diff --git a/tests/test_sense.py b/tests/test_sense.py index 2328f4e29..748d873c4 100644 --- a/tests/test_sense.py +++ b/tests/test_sense.py @@ -27,6 +27,7 @@ from solstone.observe.processing_record import ( REASON_ANALYSIS_FAILED, REASON_CORRUPT_INPUT, REASON_OK, + SCREEN_ANALYSIS_ROW_KEY, STATE_ANALYZED, STATE_EMPTY, STATE_FAILED, @@ -156,12 +157,16 @@ def _processing_record( def _write_processing_output( media_path: Path, record: dict | None, + rows: list[dict] | None = None, ) -> Path: header = {"raw": media_path.name} if record is not None: header["_solstone_processing"] = record output_path = media_path.with_suffix(".jsonl") - output_path.write_text(json.dumps(header) + "\n", encoding="utf-8") + lines = [json.dumps(header)] + if rows is not None: + lines.extend(json.dumps(row) for row in rows) + output_path.write_text("\n".join(lines) + "\n", encoding="utf-8") return output_path @@ -811,6 +816,7 @@ def test_process_day_reenters_only_retryable_describe_failures(tmp_path, monkeyp ( "143000_300", _processing_record(state=STATE_ANALYZED, reason_code="ok"), + None, False, ), ( @@ -819,15 +825,23 @@ def test_process_day_reenters_only_retryable_describe_failures(tmp_path, monkeyp state=STATE_EMPTY, reason_code="no_decodable_frames", ), + None, + False, + ), + ("143002_300", None, None, True), + ( + "143008_300", + None, + [{"frame_id": 1, SCREEN_ANALYSIS_ROW_KEY: 0.0, "analysis": {}}], False, ), - ("143002_300", None, False), ( "143003_300", _processing_record( state=STATE_FAILED, reason_code=REASON_CORRUPT_INPUT, ), + None, False, ), ( @@ -836,6 +850,7 @@ def test_process_day_reenters_only_retryable_describe_failures(tmp_path, monkeyp state=STATE_FAILED, attempts=FAILED_ATTEMPT_BOUND, ), + None, False, ), ( @@ -845,11 +860,13 @@ def test_process_day_reenters_only_retryable_describe_failures(tmp_path, monkeyp handler=HANDLER_TRANSCRIBE, attempts=1, ), + None, False, ), ( "143006_300", _processing_record(state=STATE_FAILED), + None, True, ), ( @@ -858,12 +875,13 @@ def test_process_day_reenters_only_retryable_describe_failures(tmp_path, monkeyp state=STATE_FAILED, attempts=FAILED_ATTEMPT_BOUND - 1, ), + None, True, ), ] - for segment, record, _expected in cases: + for segment, record, rows, _expected in cases: media_path = make_segment_file(tmp_path, segment=segment) - _write_processing_output(media_path, record) + _write_processing_output(media_path, record, rows=rows) sensor = FileSensor(tmp_path) sensor.register("*.webm", "describe", ["journal", "describe", "{file}"]) @@ -876,12 +894,12 @@ def test_process_day_reenters_only_retryable_describe_failures(tmp_path, monkeyp sensor.process_day("20250101", max_jobs=1) - assert processed == [segment for segment, _record, expected in cases if expected] + assert processed == [case[0] for case in cases if case[-1]] def test_process_day_reentry_uses_bounded_first_window(tmp_path, monkeypatch): monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) - media_path = make_segment_file(tmp_path) + media_path = make_segment_file(tmp_path, segment="143022_300") record = _processing_record(state=STATE_FAILED) output_path = media_path.with_suffix(".jsonl") output_path.write_bytes( @@ -891,6 +909,21 @@ def test_process_day_reentry_uses_bounded_first_window(tmp_path, monkeypatch): + json.dumps(record).encode("utf-8") + b"}\n" ) + # A record past the bounded first-row window is indeterminate: it re-enters + # unless analyzed rows provide evidence that the output is already useful. + row_media_path = make_segment_file(tmp_path, segment="143023_300") + row_output_path = row_media_path.with_suffix(".jsonl") + row_output_path.write_bytes( + b'{"pad":"' + + (b"x" * MAX_FIRST_ROW_BYTES) + + b'","_solstone_processing":' + + json.dumps(record).encode("utf-8") + + b"}\n" + + json.dumps( + {"frame_id": 1, SCREEN_ANALYSIS_ROW_KEY: 0.0, "analysis": {}} + ).encode("utf-8") + + b"\n" + ) sensor = FileSensor(tmp_path) sensor.register("*.webm", "describe", ["journal", "describe", "{file}"]) @@ -903,7 +936,34 @@ def test_process_day_reentry_uses_bounded_first_window(tmp_path, monkeypatch): sensor.process_day("20250101", max_jobs=1) - assert processed == [] + assert processed == [media_path] + + +def test_ac5_recordless_audio_output_does_not_reenter(tmp_path, monkeypatch): + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + audio_path = make_segment_file(tmp_path, filename="audio.flac") + _write_processing_output(audio_path, None) + + sensor = FileSensor(tmp_path) + command = ["journal", "transcribe", "{file}"] + sensor.register("*.flac", "transcribe", command) + + to_process, _ = sensor.scan_unprocessed("20250101") + + assert to_process == [] + + +def test_ac7_missing_screen_output_still_queues_raw_video(tmp_path, monkeypatch): + monkeypatch.setenv("SOLSTONE_JOURNAL", str(tmp_path)) + media_path = make_segment_file(tmp_path) + + sensor = FileSensor(tmp_path) + command = ["journal", "describe", "{file}"] + sensor.register("*.webm", "describe", command) + + to_process, _ = sensor.scan_unprocessed("20250101") + + assert to_process == [(media_path, "describe", command)] def test_process_day_retries_failed_describe_until_attempt_bound(tmp_path, monkeypatch):