|
57 | 57 | import time |
58 | 58 | from concurrent.futures import Future, ThreadPoolExecutor |
59 | 59 | from dataclasses import dataclass |
| 60 | +from pathlib import Path |
60 | 61 | from typing import TYPE_CHECKING, Any, Callable, Dict, List, Mapping, Optional, Sequence, Set, cast |
61 | 62 |
|
62 | 63 | from . import librarian as librarian_mod |
63 | 64 | from . import repair_queue |
64 | 65 | from . import scholar as scholar_mod |
| 66 | +from . import transcript as transcript_mod |
65 | 67 | from .buffer import Event, RollingBuffer |
66 | 68 | from .librarian import EvidencePacket |
| 69 | +from .paths import transcript_marker_path |
67 | 70 |
|
68 | 71 | if TYPE_CHECKING: |
69 | 72 | from .ingest import JsonlTailer |
|
81 | 84 | DEFAULT_QUEUE_CEILING = 100 # bounded queue capacity |
82 | 85 | DEFAULT_SWEEP_INTERVAL_SECS = 300.0 # buffer.sweep cadence |
83 | 86 |
|
| 87 | +# Role tag stamped on synthetic events carrying assistant natural-language |
| 88 | +# prose pulled from the Claude Code transcript on a Stop/SessionEnd. Distinct |
| 89 | +# from "assistant" (the default role for tool-I/O events) so the Librarian can |
| 90 | +# tell the model's REASONING apart from tool calls. ``flatten_buffer`` renders |
| 91 | +# it via the explicit-``payload.role`` path, so it reaches the prompt verbatim. |
| 92 | +ASSISTANT_PROSE_ROLE = "assistant-prose" |
| 93 | + |
84 | 94 |
|
85 | 95 | LibrarianFn = Callable[[Dict[str, Any]], EvidencePacket] |
86 | 96 | # The scholar callable normally returns None; on an agent TIMEOUT it returns |
@@ -438,12 +448,17 @@ def __init__( |
438 | 448 | config: Optional[SchedulerConfig] = None, |
439 | 449 | librarian: LibrarianFn = librarian_mod.scan, |
440 | 450 | scholar: ScholarFn = scholar_mod.review, |
| 451 | + transcript_marker: Optional[Path] = None, |
441 | 452 | ) -> None: |
442 | 453 | self.buffer = buffer |
443 | 454 | self.config = config or SchedulerConfig() |
444 | 455 | self._librarian = librarian |
445 | 456 | self._scholar = scholar |
446 | 457 | self.stats = SchedulerStats() |
| 458 | + # Where assistant-prose transcript reads persist their per-session |
| 459 | + # seen-uuid marker (incremental reads). Defaults to the canonical |
| 460 | + # daemon-home path; tests inject a tmp path. |
| 461 | + self._transcript_marker = transcript_marker or transcript_marker_path() |
447 | 462 |
|
448 | 463 | # Bounded queues — capacity == queue_ceiling, per scope brief. |
449 | 464 | # We use two separate queues so a slow Scholar doesn't choke |
@@ -572,7 +587,17 @@ def stop(self) -> None: |
572 | 587 |
|
573 | 588 | def on_event(self, ev: Event) -> None: |
574 | 589 | """Fold an Event into the buffer; arm the debounce timer when |
575 | | - the event seals a turn.""" |
| 590 | + the event seals a turn. |
| 591 | +
|
| 592 | + On a sealing event (``Stop`` / ``SessionEnd``) that carries a |
| 593 | + ``transcript_path``, we first read the assistant's natural-language |
| 594 | + turns from the Claude Code transcript and fold them into the OPEN |
| 595 | + turn as synthetic ``assistant-prose`` events — so the same Stop |
| 596 | + seals the model's reasoning alongside the tool I/O the Librarian |
| 597 | + already sees. Without this the Librarian never sees the assistant's |
| 598 | + prose (events.jsonl carries only prompts + tool calls).""" |
| 599 | + if ev.type in ("Stop", "SessionEnd"): |
| 600 | + self._inject_assistant_prose(ev) |
576 | 601 | sealed_sess = self.buffer.ingest(ev) |
577 | 602 | if sealed_sess is None: |
578 | 603 | return |
@@ -606,6 +631,64 @@ def on_event(self, ev: Event) -> None: |
606 | 631 | self.stats.librarian_debounced += 1 |
607 | 632 | # ``needs_librarian`` is cleared once the snapshot is enqueued. |
608 | 633 |
|
| 634 | + # ---- assistant-prose injection (transcript → buffer) ----------- |
| 635 | + |
| 636 | + @staticmethod |
| 637 | + def _transcript_path_of(ev: Event) -> Optional[str]: |
| 638 | + """Pull ``transcript_path`` off a sealing event's payload (or, as a |
| 639 | + fallback, the raw event dict the hook author may put it on directly). |
| 640 | + Returns ``None`` when absent — the common case until Phase 2 wires the |
| 641 | + hook to forward it, so the rest of the daemon degrades to no prose.""" |
| 642 | + for src in (ev.payload, ev.raw): |
| 643 | + cand = src.get("transcript_path") |
| 644 | + if isinstance(cand, str) and cand.strip(): |
| 645 | + return cand |
| 646 | + return None |
| 647 | + |
| 648 | + def _inject_assistant_prose(self, sealing_ev: Event) -> None: |
| 649 | + """Read NEW assistant prose for this session's transcript and fold it |
| 650 | + into the buffer as synthetic ``assistant-prose`` events BEFORE the |
| 651 | + sealing event seals the open turn. |
| 652 | +
|
| 653 | + Fail-soft on every axis: no ``transcript_path`` → no-op; unreadable |
| 654 | + transcript → empty list (logged DEBUG in the transcript module); |
| 655 | + already-seen turns → skipped via the per-session marker. Any |
| 656 | + unexpected error is swallowed with ``log.exception`` so transcript |
| 657 | + trouble can never break ingestion.""" |
| 658 | + transcript_path = self._transcript_path_of(sealing_ev) |
| 659 | + if not transcript_path: |
| 660 | + return |
| 661 | + try: |
| 662 | + prose = transcript_mod.read_new_assistant_prose( |
| 663 | + Path(transcript_path), |
| 664 | + session_id=sealing_ev.session_id, |
| 665 | + marker_path=self._transcript_marker, |
| 666 | + ) |
| 667 | + except Exception: |
| 668 | + log.exception( |
| 669 | + "assistant-prose read failed for session=%s; continuing without prose", |
| 670 | + sealing_ev.session_id, |
| 671 | + ) |
| 672 | + return |
| 673 | + if not prose: |
| 674 | + return |
| 675 | + for turn in prose: |
| 676 | + synthetic = Event( |
| 677 | + ts=sealing_ev.ts, |
| 678 | + session_id=sealing_ev.session_id, |
| 679 | + type="AssistantProse", |
| 680 | + cwd=sealing_ev.cwd, |
| 681 | + payload={"role": ASSISTANT_PROSE_ROLE, "text": turn.text}, |
| 682 | + ) |
| 683 | + # Fold into the OPEN turn (not the Stop itself) so it seals into |
| 684 | + # the same turn and flatten_buffer renders it as a quotable line. |
| 685 | + self.buffer.ingest(synthetic) |
| 686 | + log.info( |
| 687 | + "scheduler: folded %d assistant-prose turn(s) into buffer (session=%s)", |
| 688 | + len(prose), |
| 689 | + sealing_ev.session_id, |
| 690 | + ) |
| 691 | + |
609 | 692 | # ---- debounce → librarian queue -------------------------------- |
610 | 693 |
|
611 | 694 | def _on_debounce_fire(self, session_id: str) -> None: |
|
0 commit comments