solpbc.org
personal memory agent
1# SPDX-License-Identifier: AGPL-3.0-only
2# Copyright (c) 2026 sol pbc
3
4"""Offline fuzz corpus for structurally bad media.
5
6These tests drive real ingest, handler, state, completion, and idle-gate paths
7on synthesized pathological media, asserting terminal classes without server,
8sandbox, network, or model calls. AC10 is enforced as a whole-suite property.
9"""
10
11from __future__ import annotations
12
13import argparse
14import asyncio
15import importlib
16import io
17import json
18import sys
19from pathlib import Path
20from typing import Any
21from unittest.mock import AsyncMock, Mock
22
23import numpy as np
24import pytest
25import soundfile as sf
26from PIL import Image
27
28from solstone.observe.processing_record import (
29 HANDLER_DESCRIBE,
30 HANDLER_TRANSCRIBE,
31 REASON_ANALYSIS_FAILED,
32 REASON_CORRUPT_INPUT,
33 REASON_NO_DECODABLE_AUDIO,
34 REASON_NO_DECODABLE_FRAMES,
35 STATE_ANALYZED,
36 STATE_EMPTY,
37 STATE_FAILED,
38)
39from solstone.observe.utils import SAMPLE_RATE, AudioDecodeError
40from solstone.observe.vad import VadResult
41from solstone.think.cluster import (
42 cluster_segments,
43 read_segment_data_state,
44)
45from solstone.think.data_state import DataState
46from solstone.think.pipeline_health import (
47 classify_segment_completion,
48 read_segment_progress,
49)
50
51DAY = "20990501"
52STREAM = "default"
53SEGMENT = "120000_300"
54FIXED_NOW = "2026-06-30T12:00:00Z"
55
56
57def _generate_result(text: str, finish_reason: str = "stop") -> dict[str, Any]:
58 return {"text": text, "finish_reason": finish_reason}
59
60
61@pytest.fixture
62def observer_env(tmp_path, monkeypatch):
63 """Temp journal + Flask test client factory.
64
65 Self-contained mirror of the observer app-test fixture. Defined inline
66 rather than reused via ``pytest_plugins`` pointing at
67 ``solstone/apps/observer/tests/conftest.py``: that conftest is also
68 auto-registered by path during a full-suite run, so naming it as a plugin
69 double-registers the same module under two names and aborts collection
70 (passes in isolation, fails the full ``make ci``).
71 """
72
73 def _create():
74 journal = tmp_path / "journal"
75 journal.mkdir()
76
77 config_dir = journal / "config"
78 config_dir.mkdir(parents=True, exist_ok=True)
79 (config_dir / "journal.json").write_text(
80 json.dumps({"setup": {"completed_at": 1700000000000}}, indent=2)
81 )
82
83 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
84
85 from solstone.convey import create_app
86
87 app = create_app(journal=str(journal))
88
89 class Env:
90 def __init__(self):
91 self.journal = journal
92 self.client = app.test_client()
93 self.app = app
94
95 return Env()
96
97 return _create
98
99
100@pytest.fixture
101def segment_journal(tmp_path, monkeypatch):
102 journal = tmp_path / "journal"
103 journal.mkdir()
104 monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal))
105 monkeypatch.setenv("SOL_SKIP_SUPERVISOR_CHECK", "1")
106 return journal
107
108
109def _segment_dir(
110 journal: Path,
111 day: str = DAY,
112 segment: str = SEGMENT,
113 stream: str = STREAM,
114) -> Path:
115 path = journal / "chronicle" / day / stream / segment
116 path.mkdir(parents=True, exist_ok=True)
117 return path
118
119
120def _read_jsonl(path: Path) -> list[dict[str, Any]]:
121 return [
122 json.loads(line)
123 for line in path.read_text(encoding="utf-8").splitlines()
124 if line.strip()
125 ]
126
127
128def _read_header(path: Path) -> dict[str, Any]:
129 return _read_jsonl(path)[0]
130
131
132def _read_processing_record(path: Path) -> dict[str, Any]:
133 record = _read_header(path)["_solstone_processing"]
134 assert isinstance(record, dict)
135 return record
136
137
138def _assert_processing_record(
139 record: dict[str, Any],
140 *,
141 state: str,
142 reason_code: str,
143 handler: str,
144) -> None:
145 assert record["state"] == state
146 assert record["reason_code"] == reason_code
147 assert record["handler"] == handler
148 assert record["attempted_at"] == FIXED_NOW
149
150
151def _write_silent_flac(path: Path, seconds: float = 0.5) -> None:
152 sf.write(
153 path,
154 np.zeros(int(seconds * SAMPLE_RATE), np.float32),
155 SAMPLE_RATE,
156 format="FLAC",
157 )
158
159
160def _silent_flac_bytes(seconds: float = 0.5) -> bytes:
161 buf = io.BytesIO()
162 sf.write(
163 buf,
164 np.zeros(int(seconds * SAMPLE_RATE), np.float32),
165 SAMPLE_RATE,
166 format="FLAC",
167 )
168 return buf.getvalue()
169
170
171def _require_real_cv2(monkeypatch):
172 module = sys.modules.get("cv2")
173 if module is not None and not hasattr(module, "aruco"):
174 # Root tests/conftest.py installs an aruco-less cv2 stub when cv2 has
175 # not been imported yet. Evict it via monkeypatch so the real OpenCV
176 # is restored (and the stub reinstated) at test teardown — no leak
177 # into other tests in a full run.
178 monkeypatch.delitem(sys.modules, "cv2", raising=False)
179
180 cv2 = pytest.importorskip("cv2")
181 if not hasattr(cv2, "aruco"):
182 pytest.skip("OpenCV ArUco support is unavailable")
183 return cv2
184
185
186def _build_convey_covered_mp4(monkeypatch, path: Path) -> None:
187 pytest.importorskip("av")
188 cv2 = _require_real_cv2(monkeypatch)
189 import av
190
191 from solstone.observe.aruco import CORNER_TAG_IDS
192
193 # PyAV 16.1.0 does not mux a true zero-frame MP4: closing without
194 # encoding frames leaves no output file. This valid one-frame MP4 instead
195 # drives the real decode loop and is skipped by the production Convey-mask
196 # gate, producing zero qualified frames.
197 dictionary = cv2.aruco.getPredefinedDictionary(cv2.aruco.DICT_4X4_50)
198 size = 512
199 marker_size = 80
200 margin = 4
201 positions_by_id = {
202 6: (margin, margin),
203 7: (size - marker_size - margin, margin),
204 4: (margin, size - marker_size - margin),
205 2: (size - marker_size - margin, size - marker_size - margin),
206 }
207 assert set(positions_by_id) == CORNER_TAG_IDS
208
209 image = Image.new("RGB", (size, size), "white")
210 for tag_id, pos in positions_by_id.items():
211 marker = cv2.aruco.generateImageMarker(dictionary, tag_id, marker_size)
212 image.paste(Image.fromarray(marker).convert("RGB"), pos)
213
214 with av.open(str(path), "w", format="mp4") as container:
215 stream = container.add_stream("mpeg4", rate=1)
216 stream.width = size
217 stream.height = size
218 stream.pix_fmt = "yuv420p"
219 frame = av.VideoFrame.from_ndarray(np.asarray(image), format="rgb24")
220 for packet in stream.encode(frame):
221 container.mux(packet)
222 for packet in stream.encode():
223 container.mux(packet)
224
225
226def _build_one_frame_mp4(path: Path) -> None:
227 pytest.importorskip("av")
228 import av
229
230 with av.open(str(path), "w", format="mp4") as container:
231 stream = container.add_stream("mpeg4", rate=1)
232 stream.width = 64
233 stream.height = 64
234 stream.pix_fmt = "yuv420p"
235 frame = av.VideoFrame.from_ndarray(
236 np.zeros((64, 64, 3), dtype=np.uint8),
237 format="rgb24",
238 )
239 for packet in stream.encode(frame):
240 container.mux(packet)
241 for packet in stream.encode():
242 container.mux(packet)
243
244
245def _drive_describe(
246 monkeypatch,
247 video_path: Path,
248 output_path: Path,
249 *,
250 agenerate_response: str = "{}",
251 agenerate_finish_reason: str = "stop",
252 expect_runtime_error: bool = False,
253) -> tuple[dict[str, Any], dict[str, Any], AsyncMock]:
254 from solstone.observe import describe, processing_record
255
256 agenerate = AsyncMock(
257 return_value=_generate_result(agenerate_response, agenerate_finish_reason)
258 )
259 monkeypatch.setattr(
260 "solstone.think.models.resolve_provider",
261 lambda _context, _interface: ("google", "gemini-test"),
262 )
263 monkeypatch.setattr(describe, "callosum_send", lambda *args, **kwargs: None)
264 monkeypatch.setattr(describe, "select_frames_for_extraction", lambda *a, **k: [])
265 monkeypatch.setattr(processing_record, "now_iso_utc", lambda: FIXED_NOW)
266 monkeypatch.setattr("solstone.think.batch.agenerate_with_result", agenerate)
267
268 processor = describe.VideoProcessor(video_path)
269 if expect_runtime_error:
270 with pytest.raises(RuntimeError):
271 asyncio.run(
272 processor.process_with_vision(
273 max_concurrent=1,
274 output_path=output_path,
275 work_key=f"{DAY}/{video_path.parent.name}/{video_path.stem}",
276 )
277 )
278 else:
279 asyncio.run(
280 processor.process_with_vision(
281 max_concurrent=1,
282 output_path=output_path,
283 work_key=f"{DAY}/{video_path.parent.name}/{video_path.stem}",
284 )
285 )
286
287 header = _read_header(output_path)
288 record = header["_solstone_processing"]
289 assert isinstance(record, dict)
290 return header, record, agenerate
291
292
293def _drive_transcribe(
294 monkeypatch,
295 audio_path: Path,
296 *,
297 preserve_all: bool,
298) -> tuple[dict[str, Any] | None, Mock, Path]:
299 from solstone.observe import processing_record
300
301 transcribe_main = importlib.import_module("solstone.observe.transcribe.main")
302 vad_module = importlib.import_module("solstone.observe.vad")
303 stt_spy = Mock(return_value=[])
304 monkeypatch.setattr(processing_record, "now_iso_utc", lambda: FIXED_NOW)
305 monkeypatch.setattr(
306 transcribe_main,
307 "callosum_send",
308 lambda *args, **kwargs: None,
309 )
310 monkeypatch.setattr(transcribe_main, "stt_transcribe", stt_spy)
311 monkeypatch.setattr(
312 vad_module,
313 "run_vad",
314 lambda _audio, **_kwargs: VadResult(
315 duration=0.5,
316 speech_duration=0.0,
317 has_speech=False,
318 ),
319 )
320 monkeypatch.setattr(transcribe_main, "tag_audio", lambda *_args, **_kwargs: None)
321
322 transcribe_main._process_one(
323 audio_path,
324 argparse.Namespace(backend=None, cpu=False, model=None, redo=False),
325 {"preserve_all": preserve_all},
326 "parakeet",
327 [],
328 )
329
330 jsonl_path = audio_path.with_suffix(".jsonl")
331 record = _read_processing_record(jsonl_path) if jsonl_path.exists() else None
332 return record, stt_spy, jsonl_path
333
334
335def _sense_config_with_load() -> dict[str, dict[str, Any]]:
336 return {
337 "sense": {
338 "priority": 10,
339 "type": "generate",
340 "output": "json",
341 "schedule": "segment",
342 "load": {
343 "transcripts": True,
344 "percepts": True,
345 "talents": False,
346 },
347 },
348 "documents": {
349 "priority": 20,
350 "type": "cogitate",
351 "schedule": "segment",
352 },
353 "screen": {
354 "priority": 20,
355 "type": "generate",
356 "output": "md",
357 "schedule": "segment",
358 },
359 }
360
361
362def _run_idle_gate(
363 monkeypatch,
364 journal: Path,
365 day: str,
366 segment: str,
367 *,
368 agenerate_spy: AsyncMock | None = None,
369) -> tuple[tuple[int, int, list[str]], list[str], list[dict[str, Any]], AsyncMock]:
370 from solstone.think import thinking as think
371 from solstone.think.thinking import ThinkingJSONLWriter
372
373 spawned: list[str] = []
374 writer_path = journal / "chronicle" / day / "health" / f"idle_{segment}.jsonl"
375 writer = ThinkingJSONLWriter(str(writer_path))
376 agenerate = agenerate_spy or AsyncMock(return_value=_generate_result("{}"))
377 original_callosum = think._callosum
378 original_jsonl = think._jsonl
379 try:
380 monkeypatch.setattr(
381 think,
382 "get_talent_configs",
383 lambda schedule=None, **kwargs: _sense_config_with_load(),
384 )
385 monkeypatch.setattr(
386 think,
387 "cortex_request",
388 lambda prompt, name, config=None, **kwargs: (
389 spawned.append(name) or f"agent-{name}"
390 ),
391 )
392 monkeypatch.setattr(
393 think,
394 "wait_for_uses",
395 lambda agent_ids, timeout=600: ({aid: "finish" for aid in agent_ids}, []),
396 )
397 monkeypatch.setattr("solstone.think.batch.agenerate_with_result", agenerate)
398 think._callosum = None
399 think._jsonl = writer
400 result = think.run_segment_sense(
401 day,
402 segment,
403 refresh=False,
404 verbose=False,
405 stream=STREAM,
406 )
407 finally:
408 writer.close()
409 think._callosum = original_callosum
410 think._jsonl = original_jsonl
411
412 return result, spawned, _read_jsonl(writer_path), agenerate
413
414
415def _completion(day: str):
416 return classify_segment_completion(
417 cluster_segments(day),
418 read_segment_progress(day),
419 )
420
421
422def _create_observer(env, name: str) -> str:
423 response = env.client.post(
424 "/app/observer/api/create",
425 json={"name": name},
426 content_type="application/json",
427 )
428 assert response.status_code == 200
429 return response.get_json()["key"]
430
431
432def test_ac1_ingest_drops_zero_byte_keeps_valid_media(observer_env):
433 # Cross-ref:
434 # solstone/apps/observer/tests/test_routes.py::test_ingest_zero_byte_file_rejected
435 # solstone/apps/observer/tests/test_routes.py::test_ingest_mixed_zero_byte_files
436 env = observer_env()
437 key = _create_observer(env, "bad-media-observer")
438 valid_data = _silent_flac_bytes()
439
440 response = env.client.post(
441 "/app/observer/ingest",
442 headers={"Authorization": f"Bearer {key}"},
443 data={
444 "day": DAY,
445 "segment": SEGMENT,
446 "files": [
447 (io.BytesIO(b""), "empty.flac"),
448 (io.BytesIO(valid_data), "audio.flac"),
449 ],
450 },
451 )
452
453 assert response.status_code == 200
454 data = response.get_json()
455 assert data["files"] == ["audio.flac"]
456 assert data["bytes"] == len(valid_data)
457 segment = env.journal / "chronicle" / DAY / "bad-media-observer" / SEGMENT
458 assert (segment / "audio.flac").read_bytes() == valid_data
459 assert not (segment / "empty.flac").exists()
460
461
462def test_ac2_empty_screen_terminalizes_to_idle_quietly(segment_journal, monkeypatch):
463 segment = _segment_dir(segment_journal)
464 video_path = segment / "screen.mp4"
465 output_path = segment / "screen.jsonl"
466 _build_convey_covered_mp4(monkeypatch, video_path)
467
468 header, record, agenerate = _drive_describe(monkeypatch, video_path, output_path)
469
470 _assert_processing_record(
471 record,
472 state=STATE_EMPTY,
473 reason_code=REASON_NO_DECODABLE_FRAMES,
474 handler=HANDLER_DESCRIBE,
475 )
476 assert header["qualified_count"] == 0
477 assert agenerate.call_count == 0
478 assert read_segment_data_state(DAY, SEGMENT) == {"screen": DataState.EMPTY.value}
479
480 result, spawned, _events, idle_agenerate = _run_idle_gate(
481 monkeypatch,
482 segment_journal,
483 DAY,
484 SEGMENT,
485 agenerate_spy=agenerate,
486 )
487
488 assert result == (0, 0, [])
489 assert spawned == []
490 assert idle_agenerate.call_count == 0
491 assert _completion(DAY).blockers == []
492
493
494def test_ac3_silent_audio_records_empty_no_stt(segment_journal, monkeypatch):
495 segment = _segment_dir(segment_journal)
496 preserve_audio = segment / "audio.flac"
497 _write_silent_flac(preserve_audio)
498
499 record, stt_spy, jsonl_path = _drive_transcribe(
500 monkeypatch,
501 preserve_audio,
502 preserve_all=True,
503 )
504
505 assert record is not None
506 _assert_processing_record(
507 record,
508 state=STATE_EMPTY,
509 reason_code=REASON_NO_DECODABLE_AUDIO,
510 handler=HANDLER_TRANSCRIBE,
511 )
512 assert stt_spy.call_count == 0
513 assert len(jsonl_path.read_text(encoding="utf-8").splitlines()) == 1
514
515 filter_audio = segment / "filtered.flac"
516 _write_silent_flac(filter_audio)
517 filtered_record, filtered_stt, filtered_jsonl = _drive_transcribe(
518 monkeypatch,
519 filter_audio,
520 preserve_all=False,
521 )
522
523 assert filtered_record is None
524 assert filtered_stt.call_count == 0
525 assert not filter_audio.exists()
526 assert not filtered_jsonl.exists()
527
528
529def test_corrupt_audio_decode_records_failed_without_vad_or_stt(
530 segment_journal,
531 monkeypatch,
532):
533 from solstone.observe import processing_record
534
535 transcribe_main = importlib.import_module("solstone.observe.transcribe.main")
536 vad_module = importlib.import_module("solstone.observe.vad")
537 segment = _segment_dir(segment_journal)
538 audio_path = segment / "audio.m4a"
539 audio_path.write_bytes(b"truncated")
540
541 stt_spy = Mock(return_value=[])
542 vad_spy = Mock()
543 monkeypatch.setattr(processing_record, "now_iso_utc", lambda: FIXED_NOW)
544 monkeypatch.setattr(
545 transcribe_main,
546 "callosum_send",
547 lambda *args, **kwargs: None,
548 )
549 load_audio_spy = Mock(side_effect=AudioDecodeError("worker exited from signal 11"))
550 monkeypatch.setattr(transcribe_main, "load_audio", load_audio_spy)
551 monkeypatch.setattr(transcribe_main, "stt_transcribe", stt_spy)
552 monkeypatch.setattr(vad_module, "run_vad", vad_spy)
553
554 transcribe_main._process_one(
555 audio_path,
556 argparse.Namespace(backend=None, cpu=False, model=None, redo=False),
557 {"preserve_all": True},
558 "parakeet",
559 [],
560 )
561
562 jsonl_path = audio_path.with_suffix(".jsonl")
563 record = _read_processing_record(jsonl_path)
564 _assert_processing_record(
565 record,
566 state=STATE_FAILED,
567 reason_code=REASON_CORRUPT_INPUT,
568 handler=HANDLER_TRANSCRIBE,
569 )
570 assert len(jsonl_path.read_text(encoding="utf-8").splitlines()) == 1
571 assert stt_spy.call_count == 0
572 assert vad_spy.call_count == 0
573 assert load_audio_spy.call_count == 1
574 assert read_segment_data_state(DAY, SEGMENT) == {"audio": DataState.FAILED.value}
575
576 load_audio_spy.reset_mock()
577 stt_spy.reset_mock()
578 vad_spy.reset_mock()
579
580 transcribe_main._process_one(
581 audio_path,
582 argparse.Namespace(backend=None, cpu=False, model=None, redo=False),
583 {"preserve_all": True},
584 "parakeet",
585 [],
586 )
587
588 assert load_audio_spy.call_count == 0
589 assert stt_spy.call_count == 0
590 assert vad_spy.call_count == 0
591
592
593def test_ac4_corrupt_screen_is_failed_distinct_from_empty(segment_journal, monkeypatch):
594 empty_segment = _segment_dir(segment_journal, segment="121000_300")
595 empty_video = empty_segment / "screen.mp4"
596 _build_convey_covered_mp4(monkeypatch, empty_video)
597 _empty_header, empty_record, _empty_agenerate = _drive_describe(
598 monkeypatch,
599 empty_video,
600 empty_segment / "screen.jsonl",
601 )
602
603 corrupt_segment = _segment_dir(segment_journal, segment="122000_300")
604 corrupt_video = corrupt_segment / "screen.mp4"
605 corrupt_video.write_bytes(b"not a real mp4 file at all")
606 _corrupt_header, corrupt_record, corrupt_agenerate = _drive_describe(
607 monkeypatch,
608 corrupt_video,
609 corrupt_segment / "screen.jsonl",
610 )
611
612 empty_pair = (empty_record["state"], empty_record["reason_code"])
613 corrupt_pair = (corrupt_record["state"], corrupt_record["reason_code"])
614 assert empty_pair == (STATE_EMPTY, REASON_NO_DECODABLE_FRAMES)
615 assert corrupt_pair == (STATE_FAILED, REASON_CORRUPT_INPUT)
616 assert corrupt_pair != empty_pair
617 assert corrupt_agenerate.call_count == 0
618 assert read_segment_data_state(DAY, "121000_300") == {
619 "screen": DataState.EMPTY.value
620 }
621 assert read_segment_data_state(DAY, "122000_300") == {
622 "screen": DataState.FAILED.value
623 }
624 assert DataState.FAILED != DataState.EMPTY
625
626
627def test_ac5_all_frames_fail_is_analysis_failed_distinct(
628 segment_journal,
629 monkeypatch,
630):
631 segment = _segment_dir(segment_journal)
632 video_path = segment / "screen.mp4"
633 output_path = segment / "screen.jsonl"
634 _build_one_frame_mp4(video_path)
635
636 _header, record, agenerate = _drive_describe(
637 monkeypatch,
638 video_path,
639 output_path,
640 agenerate_response="not json",
641 expect_runtime_error=True,
642 )
643
644 _assert_processing_record(
645 record,
646 state=STATE_FAILED,
647 reason_code=REASON_ANALYSIS_FAILED,
648 handler=HANDLER_DESCRIBE,
649 )
650 assert agenerate.call_count == 5 # 1 qualified frame + 4 retries.
651 assert record["reason_code"] != REASON_CORRUPT_INPUT
652 assert record["reason_code"] != REASON_NO_DECODABLE_FRAMES
653 assert record["state"] != STATE_ANALYZED
654 # Both corrupt_input and analysis_failed derive DataState.FAILED; the
655 # distinction survives only in the processing-record reason_code.
656 assert read_segment_data_state(DAY, SEGMENT) == {"screen": DataState.FAILED.value}
657
658
659def test_truncated_frame_categorization_retries_and_promotes_no_frame_artifact(
660 segment_journal,
661 monkeypatch,
662):
663 segment = _segment_dir(segment_journal)
664 video_path = segment / "screen.mp4"
665 output_path = segment / "screen.jsonl"
666 _build_one_frame_mp4(video_path)
667
668 _header, record, agenerate = _drive_describe(
669 monkeypatch,
670 video_path,
671 output_path,
672 agenerate_response='{"visual_description":"partial"',
673 agenerate_finish_reason="max_tokens",
674 expect_runtime_error=True,
675 )
676
677 _assert_processing_record(
678 record,
679 state=STATE_FAILED,
680 reason_code=REASON_ANALYSIS_FAILED,
681 handler=HANDLER_DESCRIBE,
682 )
683 assert agenerate.call_count == 5
684 assert len(_read_jsonl(output_path)) == 1
685
686
687def test_ac6_no_model_calls_on_all_empty_segment(segment_journal, monkeypatch):
688 segment = _segment_dir(segment_journal)
689 screen_path = segment / "screen.mp4"
690 audio_path = segment / "audio.flac"
691 _build_convey_covered_mp4(monkeypatch, screen_path)
692 _write_silent_flac(audio_path)
693
694 _screen_header, screen_record, agenerate = _drive_describe(
695 monkeypatch,
696 screen_path,
697 segment / "screen.jsonl",
698 )
699 audio_record, stt_spy, _audio_jsonl = _drive_transcribe(
700 monkeypatch,
701 audio_path,
702 preserve_all=True,
703 )
704 result, spawned, _events, idle_agenerate = _run_idle_gate(
705 monkeypatch,
706 segment_journal,
707 DAY,
708 SEGMENT,
709 agenerate_spy=agenerate,
710 )
711
712 _assert_processing_record(
713 screen_record,
714 state=STATE_EMPTY,
715 reason_code=REASON_NO_DECODABLE_FRAMES,
716 handler=HANDLER_DESCRIBE,
717 )
718 assert audio_record is not None
719 _assert_processing_record(
720 audio_record,
721 state=STATE_EMPTY,
722 reason_code=REASON_NO_DECODABLE_AUDIO,
723 handler=HANDLER_TRANSCRIBE,
724 )
725 assert result == (0, 0, [])
726 assert spawned == []
727 assert idle_agenerate.call_count == 0
728 assert stt_spy.call_count == 0
729 assert _completion(DAY).blockers == []
730
731
732def test_ac7_terminal_empty_day_has_no_churn(segment_journal, monkeypatch):
733 segment = _segment_dir(segment_journal)
734 video_path = segment / "screen.mp4"
735 _build_convey_covered_mp4(monkeypatch, video_path)
736 _header, _record, agenerate = _drive_describe(
737 monkeypatch,
738 video_path,
739 segment / "screen.jsonl",
740 )
741 result, _spawned, _events, idle_agenerate = _run_idle_gate(
742 monkeypatch,
743 segment_journal,
744 DAY,
745 SEGMENT,
746 agenerate_spy=agenerate,
747 )
748 assert result == (0, 0, [])
749
750 for _ in range(3):
751 assert read_segment_data_state(DAY, SEGMENT) == {
752 "screen": DataState.EMPTY.value
753 }
754 assert _completion(DAY).blockers == []
755 assert idle_agenerate.call_count == 0
756
757
758def test_ac8_reprocess_terminalized_is_idempotent(segment_journal, monkeypatch):
759 segment = _segment_dir(segment_journal)
760 screen_path = segment / "screen.mp4"
761 audio_path = segment / "audio.flac"
762 screen_jsonl = segment / "screen.jsonl"
763 _build_convey_covered_mp4(monkeypatch, screen_path)
764 _write_silent_flac(audio_path)
765
766 _header, _screen_record, agenerate = _drive_describe(
767 monkeypatch,
768 screen_path,
769 screen_jsonl,
770 )
771 _audio_record, stt_spy, audio_jsonl = _drive_transcribe(
772 monkeypatch,
773 audio_path,
774 preserve_all=True,
775 )
776 result, _spawned, _events, idle_agenerate = _run_idle_gate(
777 monkeypatch,
778 segment_journal,
779 DAY,
780 SEGMENT,
781 agenerate_spy=agenerate,
782 )
783 assert result == (0, 0, [])
784 screen_lines = len(screen_jsonl.read_text(encoding="utf-8").splitlines())
785 audio_lines = len(audio_jsonl.read_text(encoding="utf-8").splitlines())
786
787 rerun_result, rerun_spawned, _rerun_events, rerun_agenerate = _run_idle_gate(
788 monkeypatch,
789 segment_journal,
790 DAY,
791 SEGMENT,
792 agenerate_spy=idle_agenerate,
793 )
794 assert rerun_result == (0, 0, [])
795 assert rerun_spawned == []
796 assert rerun_agenerate.call_count == 0
797 assert len(screen_jsonl.read_text(encoding="utf-8").splitlines()) == screen_lines
798 assert read_segment_data_state(DAY, SEGMENT) == {
799 "audio": DataState.EMPTY.value,
800 "screen": DataState.EMPTY.value,
801 }
802 assert _completion(DAY).blockers == []
803
804 _second_record, second_stt_spy, _second_audio_jsonl = _drive_transcribe(
805 monkeypatch,
806 audio_path,
807 preserve_all=True,
808 )
809 assert second_stt_spy.call_count == 0
810 assert len(audio_jsonl.read_text(encoding="utf-8").splitlines()) == audio_lines
811 assert stt_spy.call_count == 0
812
813
814def test_ac9_corrupt_output_jsonl_derives_pending(segment_journal):
815 segment = _segment_dir(segment_journal)
816 (segment / "screen.jsonl").write_bytes(b"\x00\x01 not json\n")
817
818 assert read_segment_data_state(DAY, SEGMENT) == {"screen": DataState.PENDING.value}
819 completion = _completion(DAY)
820 assert completion.blockers == [
821 {
822 "segment": SEGMENT,
823 "dimension": "not_sensed",
824 "detail": f"screen={DataState.PENDING.value}",
825 }
826 ]