+9

observe/sense.py

reviewed

··· 485 485 f"Segment fully observed{note_str}: {day}/{segment} ({duration}s)" 486 486 ) 487 487 488 488 + # Touch stream.updated marker for downstream consumers 489 489 + if day: 490 490 + try: 491 491 + health_dir = day_path(day) / "health" 492 492 + health_dir.mkdir(parents=True, exist_ok=True) 493 493 + (health_dir / "stream.updated").touch() 494 494 + except Exception: 495 495 + pass 496 496 + 488 497 # Cleanup segment tracking 489 498 del self.segment_files[segment] 490 499 del self.segment_start_time[segment]

+56

tests/test_markers.py

reviewed

··· 1 1 + # SPDX-License-Identifier: AGPL-3.0-only 2 2 + # Copyright (c) 2026 sol pbc 3 3 + 4 4 + """Tests for processing marker files (stream.updated, daily.updated).""" 5 5 + 6 6 + import time 7 7 + 8 8 + from think.utils import day_path 9 9 + 10 10 + 11 11 + def test_stream_updated_marker_created(tmp_path, monkeypatch): 12 12 + """stream.updated marker file is created in day health directory.""" 13 13 + monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) 14 14 + day = "20260215" 15 15 + health_dir = day_path(day) / "health" 16 16 + health_dir.mkdir(parents=True, exist_ok=True) 17 17 + (health_dir / "stream.updated").touch() 18 18 + assert (health_dir / "stream.updated").exists() 19 19 + 20 20 + 21 21 + def test_daily_updated_marker_created(tmp_path, monkeypatch): 22 22 + """daily.updated marker file is created in day health directory.""" 23 23 + monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) 24 24 + day = "20260215" 25 25 + health_dir = day_path(day) / "health" 26 26 + health_dir.mkdir(parents=True, exist_ok=True) 27 27 + (health_dir / "daily.updated").touch() 28 28 + assert (health_dir / "daily.updated").exists() 29 29 + 30 30 + 31 31 + def test_marker_mtime_updates(tmp_path, monkeypatch): 32 32 + """Touching a marker file again updates its mtime.""" 33 33 + monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) 34 34 + day = "20260215" 35 35 + health_dir = day_path(day) / "health" 36 36 + health_dir.mkdir(parents=True, exist_ok=True) 37 37 + marker = health_dir / "stream.updated" 38 38 + marker.touch() 39 39 + mtime1 = marker.stat().st_mtime 40 40 + time.sleep(0.05) 41 41 + marker.touch() 42 42 + mtime2 = marker.stat().st_mtime 43 43 + assert mtime2 > mtime1 44 44 + 45 45 + 46 46 + def test_marker_not_created_when_day_is_none(tmp_path, monkeypatch): 47 47 + """When day is None, no marker file should be created.""" 48 48 + monkeypatch.setenv("JOURNAL_PATH", str(tmp_path)) 49 49 + day = None 50 50 + if day: 51 51 + health_dir = day_path(day) / "health" 52 52 + health_dir.mkdir(parents=True, exist_ok=True) 53 53 + (health_dir / "stream.updated").touch() 54 54 + 55 55 + # No files should exist under tmp_path since day was None 56 56 + assert not list(tmp_path.rglob("stream.updated"))

+24

think/dream.py

reviewed

··· 1514 1514 max_concurrency=args.jobs, 1515 1515 stream=seg_stream, 1516 1516 ) 1517 1517 + # Touch stream.updated marker after each segment 1518 1518 + try: 1519 1519 + health_dir = day_path(day) / "health" 1520 1520 + health_dir.mkdir(parents=True, exist_ok=True) 1521 1521 + (health_dir / "stream.updated").touch() 1522 1522 + except Exception: 1523 1523 + pass 1517 1524 batch_success += success 1518 1525 batch_failed += failed 1519 1526 _update_status(segments_completed=i, segments_total=total) ··· 1571 1578 stream=args.stream, 1572 1579 ) 1573 1580 1581 1581 + # Touch stream.updated marker after segment processing 1582 1582 + if args.segment: 1583 1583 + try: 1584 1584 + health_dir = day_path(day) / "health" 1585 1585 + health_dir.mkdir(parents=True, exist_ok=True) 1586 1586 + (health_dir / "stream.updated").touch() 1587 1587 + except Exception: 1588 1588 + pass 1589 1589 + 1574 1590 # POST-PHASE: Final indexing and stats (daily only) 1575 1591 if not args.segment: 1576 1592 logging.info("Running post-phase: indexer rescan") ··· 1584 1600 if args.verbose: 1585 1601 stats_cmd.append("--verbose") 1586 1602 run_command(stats_cmd, day) 1603 1603 + 1604 1604 + # Touch daily.updated marker after daily schedule completion 1605 1605 + try: 1606 1606 + health_dir = day_path(day) / "health" 1607 1607 + health_dir.mkdir(parents=True, exist_ok=True) 1608 1608 + (health_dir / "daily.updated").touch() 1609 1609 + except Exception: 1610 1610 + pass 1587 1611 1588 1612 # Build log message 1589 1613 msg = "dream"