Fix observe.observed not emitting for tmux-only segments

+98 -30

2 changed files

expand all

observe

sense.py

tests

test_sense.py

+46 -30

observe/sense.py

··· 399 399 item.file_path, handler_name, command, remote=item.remote 400 400 ) 401 401 402 + def _emit_segment_observed(self, segment: str, note: str = ""): 403 + """Emit observe.observed event and cleanup segment tracking. 404 + 405 + Must be called while holding self.lock. 406 + 407 + Args: 408 + segment: Segment key (HHMMSS_LEN format) 409 + note: Optional note for log message (e.g., "no handlers") 410 + """ 411 + duration = int(time.time() - self.segment_start_time[segment]) 412 + day = self.segment_day.get(segment) 413 + batch = self.segment_batch.get(segment, False) 414 + remote = self.segment_remote.get(segment) 415 + 416 + if self.callosum: 417 + event_fields = { 418 + "segment": segment, 419 + "day": day, 420 + "duration": duration, 421 + } 422 + if batch: 423 + event_fields["batch"] = True 424 + if remote: 425 + event_fields["remote"] = remote 426 + self.callosum.emit("observe", "observed", **event_fields) 427 + 428 + note_str = f" ({note})" if note else "" 429 + logger.info(f"Segment fully observed{note_str}: {day}/{segment} ({duration}s)") 430 + 431 + # Cleanup segment tracking 432 + del self.segment_files[segment] 433 + del self.segment_start_time[segment] 434 + if segment in self.segment_day: 435 + del self.segment_day[segment] 436 + if segment in self.segment_batch: 437 + del self.segment_batch[segment] 438 + if segment in self.segment_remote: 439 + del self.segment_remote[segment] 440 + 402 441 def _check_segment_observed(self, file_path: Path): 403 442 """Check if all files for this segment have completed processing.""" 404 443 from observe.utils import get_segment_key ··· 413 452 414 453 # If no more pending files, emit observed event 415 454 if not self.segment_files[segment]: 416 - # Calculate processing duration 417 - duration = int(time.time() - self.segment_start_time[segment]) 418 - day = self.segment_day.get(segment) 419 - batch = self.segment_batch.get(segment, False) 420 - remote = self.segment_remote.get(segment) 421 - 422 - if self.callosum: 423 - event_fields = { 424 - "segment": segment, 425 - "day": day, 426 - "duration": duration, 427 - } 428 - if batch: 429 - event_fields["batch"] = True 430 - if remote: 431 - event_fields["remote"] = remote 432 - self.callosum.emit("observe", "observed", **event_fields) 433 - logger.info( 434 - f"Segment fully observed: {day}/{segment} ({duration}s)" 435 - ) 436 - 437 - # Cleanup 438 - del self.segment_files[segment] 439 - del self.segment_start_time[segment] 440 - if segment in self.segment_day: 441 - del self.segment_day[segment] 442 - if segment in self.segment_batch: 443 - del self.segment_batch[segment] 444 - if segment in self.segment_remote: 445 - del self.segment_remote[segment] 455 + self._emit_segment_observed(segment) 446 456 447 457 def _handle_file( 448 458 self, ··· 513 523 # Process each file (pass segment context for env vars) 514 524 for file_path in file_paths: 515 525 self._handle_file(file_path, segment=segment, remote=remote) 526 + 527 + # If no files matched any handler patterns, emit observed immediately 528 + # (e.g., tmux-only segments with just .jsonl files) 529 + with self.lock: 530 + if segment in self.segment_files and not self.segment_files[segment]: 531 + self._emit_segment_observed(segment, note="no handlers") 516 532 517 533 def _emit_status(self): 518 534 """Emit observe.status event with current processing state (only when active)."""

+52

tests/test_sense.py

··· 568 568 assert observed_events[0].get("segment") == "143022_300" 569 569 570 570 571 + def test_file_sensor_segment_observed_no_handlers(tmp_path, mock_callosum): 572 + """Test that observe.observed is emitted immediately for segments with no matching handlers. 573 + 574 + This covers the case of tmux-only segments where files like .jsonl don't match 575 + any registered patterns (*.flac, *.webm, etc.). 576 + """ 577 + from think.callosum import CallosumConnection 578 + 579 + # Create journal/day/segment structure 580 + day_dir = tmp_path / "20250101" 581 + segment_dir = day_dir / "143022_300" 582 + segment_dir.mkdir(parents=True) 583 + 584 + sensor = FileSensor(tmp_path) 585 + # Only register handlers for audio/video (not .jsonl) 586 + sensor.register("*.flac", "transcribe", ["echo", "{file}"]) 587 + sensor.register("*.webm", "describe", ["echo", "{file}"]) 588 + 589 + # Set up callosum on sensor to capture emitted events 590 + emitted_events = [] 591 + sensor.callosum = CallosumConnection() 592 + sensor.callosum.start(callback=lambda msg: emitted_events.append(msg)) 593 + 594 + # Create test file that doesn't match any pattern (like tmux captures) 595 + jsonl_file = segment_dir / "tmux_0_screen.jsonl" 596 + jsonl_file.write_text('{"content": "terminal output"}') 597 + 598 + # Simulate observing event with only .jsonl file 599 + message = { 600 + "tract": "observe", 601 + "event": "observing", 602 + "day": "20250101", 603 + "segment": "143022_300", 604 + "files": ["tmux_0_screen.jsonl"], 605 + } 606 + sensor._handle_callosum_message(message) 607 + 608 + # Segment tracking should be cleaned up immediately (no handlers to wait for) 609 + assert "143022_300" not in sensor.segment_files 610 + assert "143022_300" not in sensor.segment_day 611 + 612 + # Check observe.observed event was emitted immediately 613 + observed_events = [ 614 + e 615 + for e in emitted_events 616 + if e.get("tract") == "observe" and e.get("event") == "observed" 617 + ] 618 + assert len(observed_events) == 1 619 + assert observed_events[0].get("day") == "20250101" 620 + assert observed_events[0].get("segment") == "143022_300" 621 + 622 + 571 623 def test_delete_outputs_screen(tmp_path): 572 624 """Test delete_outputs with screen type.""" 573 625 from observe.sense import delete_outputs

Configure Feed

Configure Feed