feat(orchestrator): backstop no-reply du rendez-vous inter-agents
Remédiation du wedge persistant après échec live du fix Finding A (77e62e5).
Détecte la fin de tour d'un agent sollicité qui n'a pas appelé idea_reply et
débloque l'agent demandeur au lieu de le laisser en attente indéfinie.
Ajoute le suivi de tour côté inspector Claude (claude_turn_watcher) et la
résolution des chemins de session (claude_paths), câblés dans le rendez-vous
idea_ask_agent ⇄ idea_reply.
Build workspace vert, suite complète verte, zéro warning.
Backstop NON prouvé levé en live : merge develop interdit tant que la levée
du wedge n'est pas validée en conditions réelles.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
File diff suppressed because it is too large
Load Diff
@ -25,11 +25,13 @@ use async_trait::async_trait;
|
||||
use serde::Deserialize;
|
||||
|
||||
use domain::ports::{
|
||||
ConversationDetails, FileSystem, FsError, InspectError, RemotePath, SessionInspector,
|
||||
ConversationDetails, FileSystem, FsError, InspectError, SessionInspector,
|
||||
};
|
||||
use domain::profile::{AgentProfile, ContextInjection};
|
||||
use domain::profile::AgentProfile;
|
||||
use domain::project::ProjectPath;
|
||||
|
||||
use super::claude_paths::{claude_transcript_path, supports_claude};
|
||||
|
||||
/// Max byte length of the extracted `last_topic` before it is truncated.
|
||||
///
|
||||
/// Kept small: the topic only labels a resume popup, it is not the message.
|
||||
@ -67,32 +69,6 @@ impl ClaudeTranscriptInspector {
|
||||
pub fn with_base_dir(fs: Arc<dyn FileSystem>, home_dir: impl Into<String>) -> Self {
|
||||
Self::new(fs, home_dir)
|
||||
}
|
||||
|
||||
/// `<home>/.claude/projects/<encoded-cwd>/<conversation-id>.jsonl`.
|
||||
fn transcript_path(&self, conversation_id: &str, cwd: &ProjectPath) -> RemotePath {
|
||||
let base = self.home_dir.trim_end_matches(['/', '\\']);
|
||||
let encoded = encode_cwd(cwd.as_str());
|
||||
RemotePath::new(format!(
|
||||
"{base}/.claude/projects/{encoded}/{conversation_id}.jsonl"
|
||||
))
|
||||
}
|
||||
}
|
||||
|
||||
/// Encodes a cwd the way Claude Code names its per-project transcript folder.
|
||||
///
|
||||
/// **Assumption (documented, isolated, and unit-tested):** Claude flattens the
|
||||
/// absolute cwd into a single directory segment by replacing each path
|
||||
/// separator (`/` or `\`) — and the leading separator — with `-`. So
|
||||
/// `/home/anthony/Documents/Projects/IdeA` becomes
|
||||
/// `-home-anthony-Documents-Projects-IdeA`. We keep this convention in one
|
||||
/// small, tested function so that if Claude's exact encoding differs in some
|
||||
/// edge case, only this seam needs to change. We do not attempt to encode `.`
|
||||
/// or other characters, as the common case (clean absolute project paths) is
|
||||
/// covered and over-encoding would risk pointing at the wrong folder.
|
||||
fn encode_cwd(cwd: &str) -> String {
|
||||
cwd.chars()
|
||||
.map(|c| if c == '/' || c == '\\' { '-' } else { c })
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// One transcript line, as far as we care about it. Everything is optional so
|
||||
@ -251,17 +227,7 @@ fn truncate_topic(s: &str) -> String {
|
||||
#[async_trait]
|
||||
impl SessionInspector for ClaudeTranscriptInspector {
|
||||
fn supports(&self, profile: &AgentProfile) -> bool {
|
||||
// Recognise Claude by its conventional context file `CLAUDE.md`,
|
||||
// mirroring the detection in `application::agent::lifecycle`.
|
||||
matches!(
|
||||
&profile.context_injection,
|
||||
ContextInjection::ConventionFile { target }
|
||||
if target
|
||||
.rsplit(['/', '\\'])
|
||||
.next()
|
||||
.unwrap_or(target)
|
||||
.eq_ignore_ascii_case("CLAUDE.md")
|
||||
)
|
||||
supports_claude(profile)
|
||||
}
|
||||
|
||||
async fn details(
|
||||
@ -270,7 +236,7 @@ impl SessionInspector for ClaudeTranscriptInspector {
|
||||
conversation_id: &str,
|
||||
cwd: &ProjectPath,
|
||||
) -> Result<ConversationDetails, InspectError> {
|
||||
let path = self.transcript_path(conversation_id, cwd);
|
||||
let path = claude_transcript_path(&self.home_dir, conversation_id, cwd);
|
||||
match self.fs.read(&path).await {
|
||||
Ok(bytes) => Ok(parse_transcript(&bytes)),
|
||||
Err(FsError::NotFound(_)) => Err(InspectError::NotFound),
|
||||
@ -283,17 +249,6 @@ impl SessionInspector for ClaudeTranscriptInspector {
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn encode_cwd_flattens_separators_to_dash() {
|
||||
assert_eq!(
|
||||
encode_cwd("/home/anthony/Documents/Projects/IdeA"),
|
||||
"-home-anthony-Documents-Projects-IdeA"
|
||||
);
|
||||
assert_eq!(encode_cwd("/a/b"), "-a-b");
|
||||
// Windows-style separators flatten too.
|
||||
assert_eq!(encode_cwd("C:\\Users\\me"), "C:-Users-me");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parse_extracts_last_user_topic_and_token_sum() {
|
||||
let body = concat!(
|
||||
|
||||
102
crates/infrastructure/src/inspector/claude_paths.rs
Normal file
102
crates/infrastructure/src/inspector/claude_paths.rs
Normal file
@ -0,0 +1,102 @@
|
||||
//! Shared Claude Code transcript-path helpers.
|
||||
//!
|
||||
//! Claude Code records each conversation as a JSONL transcript under the user's
|
||||
//! home directory:
|
||||
//!
|
||||
//! ```text
|
||||
//! <home>/.claude/projects/<encoded-cwd>/<engine-session-id>.jsonl
|
||||
//! ```
|
||||
//!
|
||||
//! where `<encoded-cwd>` is the agent's working directory with its path separators
|
||||
//! flattened to `-` (see [`encode_cwd`]). Both the [`ClaudeTranscriptInspector`] (which
|
||||
//! reads one known transcript) and the [`ClaudeTranscriptTurnWatcher`] (which watches the
|
||||
//! whole per-agent project dir) need to derive these paths, so the convention lives in
|
||||
//! **one** small, tested place: if Claude's exact encoding ever differs, only this seam
|
||||
//! changes.
|
||||
//!
|
||||
//! [`ClaudeTranscriptInspector`]: super::ClaudeTranscriptInspector
|
||||
//! [`ClaudeTranscriptTurnWatcher`]: super::ClaudeTranscriptTurnWatcher
|
||||
|
||||
use domain::ports::RemotePath;
|
||||
use domain::profile::{AgentProfile, ContextInjection};
|
||||
use domain::project::ProjectPath;
|
||||
|
||||
/// Encodes a cwd the way Claude Code names its per-project transcript folder.
|
||||
///
|
||||
/// **Assumption (documented, isolated, and unit-tested):** Claude flattens the
|
||||
/// absolute cwd into a single directory segment by replacing each path separator
|
||||
/// (`/` or `\`) — and the leading separator — with `-`. So
|
||||
/// `/home/anthony/Documents/Projects/IdeA` becomes
|
||||
/// `-home-anthony-Documents-Projects-IdeA`. We keep this convention in one small,
|
||||
/// tested function so that if Claude's exact encoding differs in some edge case, only
|
||||
/// this seam needs to change. We do not attempt to encode `.` or other characters, as
|
||||
/// the common case (clean absolute project paths) is covered and over-encoding would
|
||||
/// risk pointing at the wrong folder.
|
||||
pub(crate) fn encode_cwd(cwd: &str) -> String {
|
||||
cwd.chars()
|
||||
.map(|c| if c == '/' || c == '\\' { '-' } else { c })
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// `<home>/.claude/projects/<encoded-cwd>` — the directory holding **all** of one
|
||||
/// agent's conversation transcripts for `cwd` (its isolated run dir). Trailing path
|
||||
/// separators on `home_dir` are trimmed first.
|
||||
pub(crate) fn claude_project_dir(home_dir: &str, cwd: &ProjectPath) -> RemotePath {
|
||||
let base = home_dir.trim_end_matches(['/', '\\']);
|
||||
let encoded = encode_cwd(cwd.as_str());
|
||||
RemotePath::new(format!("{base}/.claude/projects/{encoded}"))
|
||||
}
|
||||
|
||||
/// `<home>/.claude/projects/<encoded-cwd>/<conversation-id>.jsonl`.
|
||||
pub(crate) fn claude_transcript_path(
|
||||
home_dir: &str,
|
||||
conversation_id: &str,
|
||||
cwd: &ProjectPath,
|
||||
) -> RemotePath {
|
||||
let dir = claude_project_dir(home_dir, cwd);
|
||||
RemotePath::new(format!("{}/{conversation_id}.jsonl", dir.0))
|
||||
}
|
||||
|
||||
/// Recognises a Claude profile by its conventional context file `CLAUDE.md`, mirroring
|
||||
/// the detection in `application::agent::lifecycle`. Shared by the inspector and the
|
||||
/// turn watcher so both agree on what they can read.
|
||||
pub(crate) fn supports_claude(profile: &AgentProfile) -> bool {
|
||||
matches!(
|
||||
&profile.context_injection,
|
||||
ContextInjection::ConventionFile { target }
|
||||
if target
|
||||
.rsplit(['/', '\\'])
|
||||
.next()
|
||||
.unwrap_or(target)
|
||||
.eq_ignore_ascii_case("CLAUDE.md")
|
||||
)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn encode_cwd_flattens_separators_to_dash() {
|
||||
assert_eq!(
|
||||
encode_cwd("/home/anthony/Documents/Projects/IdeA"),
|
||||
"-home-anthony-Documents-Projects-IdeA"
|
||||
);
|
||||
assert_eq!(encode_cwd("/a/b"), "-a-b");
|
||||
// Windows-style separators flatten too.
|
||||
assert_eq!(encode_cwd("C:\\Users\\me"), "C:-Users-me");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn project_dir_and_transcript_path_compose() {
|
||||
let cwd = ProjectPath::new("/a/b").expect("valid path");
|
||||
assert_eq!(
|
||||
claude_project_dir("/home/me/", &cwd).0,
|
||||
"/home/me/.claude/projects/-a-b"
|
||||
);
|
||||
assert_eq!(
|
||||
claude_transcript_path("/home/me", "conv-1", &cwd).0,
|
||||
"/home/me/.claude/projects/-a-b/conv-1.jsonl"
|
||||
);
|
||||
}
|
||||
}
|
||||
352
crates/infrastructure/src/inspector/claude_turn_watcher.rs
Normal file
352
crates/infrastructure/src/inspector/claude_turn_watcher.rs
Normal file
@ -0,0 +1,352 @@
|
||||
//! [`ClaudeTranscriptTurnWatcher`] — the [`TurnWatcher`] adapter for the **Claude Code**
|
||||
//! CLI (rendez-vous no-reply backstop).
|
||||
//!
|
||||
//! It replaces the dead PTY prompt-ready sniff as the **end-of-turn** authority. Claude
|
||||
//! records each completed turn as a `turn_duration` entry in its JSONL transcript; this
|
||||
//! adapter **tail-polls** the agent's per-run-dir transcript folder
|
||||
//! (`<home>/.claude/projects/<encoded-run-dir>/`) every [`POLL_INTERVAL`] and fires the
|
||||
//! [`OnTurnEnd`] callback each time the **aggregate** count of `turn_duration` records
|
||||
//! grows above the **baseline** captured at arm time.
|
||||
//!
|
||||
//! ## Why the whole folder, not one file
|
||||
//!
|
||||
//! The transcript file is named by Claude's **engine session id** (the resumable), which
|
||||
//! is unknown at cold start (the file does not exist yet) and is **not** the IdeA pair
|
||||
//! `conversation_id`. Since every IdeA agent runs in its **own isolated run dir**
|
||||
//! (`.ideai/run/<agent>/`), its `<encoded-run-dir>` folder belongs to exactly **one**
|
||||
//! agent: aggregating `turn_duration` across the `.jsonl` files in that folder is a sound
|
||||
//! per-agent turn-end signal that needs neither the engine id nor file mtimes (the
|
||||
//! [`domain::ports::FileSystem`] port exposes neither).
|
||||
//!
|
||||
//! ## Robustness (cadrage pièges)
|
||||
//!
|
||||
//! - **Baseline at arm** ⇒ a pre-existing transcript (relaunch / background wake) never
|
||||
//! fires a phantom turn end (cf. `mcp-e2e-findings-reply-wedge-phantom-busy`).
|
||||
//! - **Cold start**: the folder/file may not exist yet ⇒ treated as count `0`, the watch
|
||||
//! simply waits for it to appear (the first turn is also covered by the MCP-initialize
|
||||
//! cold-start release).
|
||||
//! - **Rotation / truncation**: if the aggregate count ever **drops**, the baseline is
|
||||
//! reset to the new (lower) value — best-effort, never fires spuriously.
|
||||
//! - **Poll, not inotify** ⇒ portable inside the AppImage.
|
||||
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
use domain::ids::AgentId;
|
||||
use domain::ports::{FileSystem, OnTurnEnd, RemotePath, TurnWatchHandle, TurnWatcher};
|
||||
use domain::profile::AgentProfile;
|
||||
use domain::project::ProjectPath;
|
||||
|
||||
use super::claude_paths::{claude_project_dir, supports_claude};
|
||||
|
||||
/// Tail-poll interval (cadrage : 250-500 ms). 300 ms balances latency vs. cost.
|
||||
const POLL_INTERVAL: Duration = Duration::from_millis(300);
|
||||
|
||||
/// Substring marking a completed turn in a Claude transcript line.
|
||||
const TURN_END_MARKER: &str = "turn_duration";
|
||||
|
||||
/// [`TurnWatcher`] reading Claude Code transcripts. Composes a [`FileSystem`] port (so it
|
||||
/// stays OS- and Tauri-agnostic and trivially testable) plus the base directory standing
|
||||
/// in for the user's home (`<home>/.claude/projects/`).
|
||||
#[derive(Clone)]
|
||||
pub struct ClaudeTranscriptTurnWatcher {
|
||||
fs: Arc<dyn FileSystem>,
|
||||
home_dir: String,
|
||||
}
|
||||
|
||||
impl ClaudeTranscriptTurnWatcher {
|
||||
/// Builds the watcher from an injected [`FileSystem`] and the user's home directory
|
||||
/// (resolved by the composition root, e.g. from `$HOME`).
|
||||
#[must_use]
|
||||
pub fn new(fs: Arc<dyn FileSystem>, home_dir: impl Into<String>) -> Self {
|
||||
Self {
|
||||
fs,
|
||||
home_dir: home_dir.into(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Handle whose [`Drop`] stops the polling task **effectively and idempotently**: it
|
||||
/// flips the shared stop flag (observed at the top *and* bottom of each loop iteration)
|
||||
/// **and** aborts the task (so an in-flight `await` cannot fire after the drop).
|
||||
struct ClaudeTurnWatchHandle {
|
||||
stop: Arc<AtomicBool>,
|
||||
task: tokio::task::JoinHandle<()>,
|
||||
}
|
||||
|
||||
impl TurnWatchHandle for ClaudeTurnWatchHandle {}
|
||||
|
||||
impl Drop for ClaudeTurnWatchHandle {
|
||||
fn drop(&mut self) {
|
||||
self.stop.store(true, Ordering::SeqCst);
|
||||
self.task.abort();
|
||||
}
|
||||
}
|
||||
|
||||
impl TurnWatcher for ClaudeTranscriptTurnWatcher {
|
||||
fn supports(&self, profile: &AgentProfile) -> bool {
|
||||
supports_claude(profile)
|
||||
}
|
||||
|
||||
fn watch(
|
||||
&self,
|
||||
agent: AgentId,
|
||||
conversation_id: Option<String>,
|
||||
cwd: ProjectPath,
|
||||
on_turn_end: OnTurnEnd,
|
||||
) -> Box<dyn TurnWatchHandle> {
|
||||
let fs = Arc::clone(&self.fs);
|
||||
let dir = claude_project_dir(&self.home_dir, &cwd);
|
||||
let stop = Arc::new(AtomicBool::new(false));
|
||||
let stop_task = Arc::clone(&stop);
|
||||
application::diag!(
|
||||
"[turn-watcher] arm agent={agent} dir={} conversation={}",
|
||||
dir.0,
|
||||
conversation_id.as_deref().unwrap_or("<none>"),
|
||||
);
|
||||
let task = tokio::spawn(async move {
|
||||
// Baseline = turn ends already present at arm time (never re-fired).
|
||||
let mut last = count_turn_ends(fs.as_ref(), &dir).await;
|
||||
application::diag!("[turn-watcher] baseline agent={agent} count={last}");
|
||||
loop {
|
||||
if stop_task.load(Ordering::SeqCst) {
|
||||
break;
|
||||
}
|
||||
tokio::time::sleep(POLL_INTERVAL).await;
|
||||
if stop_task.load(Ordering::SeqCst) {
|
||||
break;
|
||||
}
|
||||
let current = count_turn_ends(fs.as_ref(), &dir).await;
|
||||
if current < last {
|
||||
// Rotation / truncation : re-baseline (best-effort, no spurious fire).
|
||||
last = current;
|
||||
continue;
|
||||
}
|
||||
if current > last {
|
||||
application::diag!(
|
||||
"[turn-watcher] turn_duration detected agent={agent} count {last}->{current} -> turn_ended"
|
||||
);
|
||||
last = current;
|
||||
// Callback invoked from the polling task, holding no lock (it ends up
|
||||
// in `InputMediator::turn_ended`). One fire per detected increase: a
|
||||
// burst still ends the currently-busy turn; an idle agent is a no-op.
|
||||
on_turn_end(agent);
|
||||
}
|
||||
}
|
||||
});
|
||||
Box::new(ClaudeTurnWatchHandle { stop, task })
|
||||
}
|
||||
}
|
||||
|
||||
/// Aggregate count of `turn_duration` records across **all** `.jsonl` transcripts in the
|
||||
/// agent's per-run-dir folder. A missing folder/file (cold start) ⇒ `0`. Unreadable
|
||||
/// files are skipped (best-effort, never fatal).
|
||||
async fn count_turn_ends(fs: &dyn FileSystem, dir: &RemotePath) -> usize {
|
||||
let Ok(entries) = fs.list(dir).await else {
|
||||
return 0; // folder not created yet (cold start) or transient error.
|
||||
};
|
||||
let mut total = 0usize;
|
||||
for entry in entries {
|
||||
if entry.is_dir || !entry.name.ends_with(".jsonl") {
|
||||
continue;
|
||||
}
|
||||
let path = RemotePath::new(format!("{}/{}", dir.0, entry.name));
|
||||
let Ok(bytes) = fs.read(&path).await else {
|
||||
continue;
|
||||
};
|
||||
total += count_marker_lines(&bytes);
|
||||
}
|
||||
total
|
||||
}
|
||||
|
||||
/// Counts transcript **lines** containing the [`TURN_END_MARKER`] (one logical turn end
|
||||
/// per `turn_duration` record). Substring match on the raw bytes — cheap and tolerant of
|
||||
/// partial/unparseable lines.
|
||||
fn count_marker_lines(body: &[u8]) -> usize {
|
||||
String::from_utf8_lossy(body)
|
||||
.lines()
|
||||
.filter(|line| line.contains(TURN_END_MARKER))
|
||||
.count()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use async_trait::async_trait;
|
||||
use domain::ports::{DirEntry, FsError};
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Mutex;
|
||||
|
||||
/// In-memory [`FileSystem`] double: a flat map of path → bytes; `list` returns the
|
||||
/// basenames whose parent is the queried dir. Only `read`/`list` are exercised.
|
||||
#[derive(Default)]
|
||||
struct FakeFs {
|
||||
files: Mutex<HashMap<String, Vec<u8>>>,
|
||||
}
|
||||
impl FakeFs {
|
||||
fn set(&self, path: &str, body: &str) {
|
||||
self.files
|
||||
.lock()
|
||||
.unwrap()
|
||||
.insert(path.to_owned(), body.as_bytes().to_vec());
|
||||
}
|
||||
}
|
||||
#[async_trait]
|
||||
impl FileSystem for FakeFs {
|
||||
async fn read(&self, path: &RemotePath) -> Result<Vec<u8>, FsError> {
|
||||
self.files
|
||||
.lock()
|
||||
.unwrap()
|
||||
.get(&path.0)
|
||||
.cloned()
|
||||
.ok_or_else(|| FsError::NotFound(path.0.clone()))
|
||||
}
|
||||
async fn write(&self, path: &RemotePath, data: &[u8]) -> Result<(), FsError> {
|
||||
self.files.lock().unwrap().insert(path.0.clone(), data.to_vec());
|
||||
Ok(())
|
||||
}
|
||||
async fn exists(&self, path: &RemotePath) -> Result<bool, FsError> {
|
||||
Ok(self.files.lock().unwrap().contains_key(&path.0))
|
||||
}
|
||||
async fn create_dir_all(&self, _path: &RemotePath) -> Result<(), FsError> {
|
||||
Ok(())
|
||||
}
|
||||
async fn list(&self, path: &RemotePath) -> Result<Vec<DirEntry>, FsError> {
|
||||
let prefix = format!("{}/", path.0);
|
||||
let files = self.files.lock().unwrap();
|
||||
let mut out = Vec::new();
|
||||
for key in files.keys() {
|
||||
if let Some(rest) = key.strip_prefix(&prefix) {
|
||||
if !rest.contains('/') {
|
||||
out.push(DirEntry {
|
||||
name: rest.to_owned(),
|
||||
is_dir: false,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
if out.is_empty() && !files.keys().any(|k| k.starts_with(&prefix)) {
|
||||
return Err(FsError::NotFound(path.0.clone()));
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
async fn symlink(&self, _src: &RemotePath, _dst: &RemotePath) -> Result<(), FsError> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
fn agent(n: u128) -> AgentId {
|
||||
AgentId::from_uuid(uuid::Uuid::from_u128(n))
|
||||
}
|
||||
|
||||
fn line(extra: &str) -> String {
|
||||
format!("{{\"type\":\"result\",\"{extra}\":1}}\n")
|
||||
}
|
||||
|
||||
/// The run dir `/run/a` encodes to `-run-a` ⇒ folder `<home>/.claude/projects/-run-a`.
|
||||
fn transcript_path(name: &str) -> String {
|
||||
format!("/home/me/.claude/projects/-run-a/{name}")
|
||||
}
|
||||
|
||||
fn count_calls() -> (Arc<Mutex<Vec<AgentId>>>, OnTurnEnd) {
|
||||
let log: Arc<Mutex<Vec<AgentId>>> = Arc::new(Mutex::new(Vec::new()));
|
||||
let log2 = Arc::clone(&log);
|
||||
let cb: OnTurnEnd = Arc::new(move |a: AgentId| log2.lock().unwrap().push(a));
|
||||
(log, cb)
|
||||
}
|
||||
|
||||
async fn wait_until(mut cond: impl FnMut() -> bool) -> bool {
|
||||
for _ in 0..100 {
|
||||
if cond() {
|
||||
return true;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(20)).await;
|
||||
}
|
||||
cond()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn count_marker_lines_counts_only_turn_duration_lines() {
|
||||
let body = concat!(
|
||||
"{\"type\":\"user\"}\n",
|
||||
"{\"type\":\"result\",\"turn_duration\":12}\n",
|
||||
"garbage line\n",
|
||||
"{\"type\":\"result\",\"turn_duration\":7}\n",
|
||||
);
|
||||
assert_eq!(count_marker_lines(body.as_bytes()), 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn fires_on_new_turn_duration_above_baseline() {
|
||||
let fs = Arc::new(FakeFs::default());
|
||||
// Baseline: one pre-existing turn end ⇒ must NOT fire for it.
|
||||
fs.set(&transcript_path("engine-1.jsonl"), &line("turn_duration"));
|
||||
let watcher = ClaudeTranscriptTurnWatcher::new(Arc::clone(&fs) as Arc<dyn FileSystem>, "/home/me");
|
||||
let (log, cb) = count_calls();
|
||||
let cwd = ProjectPath::new("/run/a").expect("path");
|
||||
let _h = watcher.watch(agent(1), None, cwd, cb);
|
||||
|
||||
// Append a second turn end ⇒ exactly one fire.
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
fs.set(
|
||||
&transcript_path("engine-1.jsonl"),
|
||||
&format!("{}{}", line("turn_duration"), line("turn_duration")),
|
||||
);
|
||||
assert!(
|
||||
wait_until(|| !log.lock().unwrap().is_empty()).await,
|
||||
"a new turn_duration above baseline fires turn_ended"
|
||||
);
|
||||
assert_eq!(log.lock().unwrap().as_slice(), &[agent(1)]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn pre_existing_transcript_never_fires_phantom() {
|
||||
let fs = Arc::new(FakeFs::default());
|
||||
fs.set(
|
||||
&transcript_path("engine-1.jsonl"),
|
||||
&format!("{}{}", line("turn_duration"), line("turn_duration")),
|
||||
);
|
||||
let watcher = ClaudeTranscriptTurnWatcher::new(Arc::clone(&fs) as Arc<dyn FileSystem>, "/home/me");
|
||||
let (log, cb) = count_calls();
|
||||
let cwd = ProjectPath::new("/run/a").expect("path");
|
||||
let _h = watcher.watch(agent(1), None, cwd, cb);
|
||||
|
||||
// No new turn end appended ⇒ no fire, ever (baseline absorbs the existing ones).
|
||||
tokio::time::sleep(Duration::from_millis(120)).await;
|
||||
assert!(log.lock().unwrap().is_empty(), "baseline must absorb pre-existing turn ends");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn cold_start_missing_folder_then_first_turn_fires() {
|
||||
let fs = Arc::new(FakeFs::default());
|
||||
let watcher = ClaudeTranscriptTurnWatcher::new(Arc::clone(&fs) as Arc<dyn FileSystem>, "/home/me");
|
||||
let (log, cb) = count_calls();
|
||||
let cwd = ProjectPath::new("/run/a").expect("path");
|
||||
let _h = watcher.watch(agent(1), None, cwd, cb);
|
||||
|
||||
// Folder/file appear later (cold start), then a first turn end is written.
|
||||
tokio::time::sleep(Duration::from_millis(40)).await;
|
||||
fs.set(&transcript_path("engine-1.jsonl"), &line("turn_duration"));
|
||||
assert!(
|
||||
wait_until(|| !log.lock().unwrap().is_empty()).await,
|
||||
"a first turn end after a cold start fires (baseline was 0)"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn dropping_handle_stops_polling() {
|
||||
let fs = Arc::new(FakeFs::default());
|
||||
let watcher = ClaudeTranscriptTurnWatcher::new(Arc::clone(&fs) as Arc<dyn FileSystem>, "/home/me");
|
||||
let (log, cb) = count_calls();
|
||||
let cwd = ProjectPath::new("/run/a").expect("path");
|
||||
let h = watcher.watch(agent(1), None, cwd, cb);
|
||||
drop(h); // stop the watch.
|
||||
|
||||
// A turn end written after the drop must not fire.
|
||||
tokio::time::sleep(Duration::from_millis(40)).await;
|
||||
fs.set(&transcript_path("engine-1.jsonl"), &line("turn_duration"));
|
||||
tokio::time::sleep(Duration::from_millis(120)).await;
|
||||
assert!(log.lock().unwrap().is_empty(), "a dropped handle stops firing");
|
||||
}
|
||||
}
|
||||
@ -8,5 +8,8 @@
|
||||
//! missing or failing inspector must never block a resume.
|
||||
|
||||
mod claude;
|
||||
mod claude_paths;
|
||||
mod claude_turn_watcher;
|
||||
|
||||
pub use claude::ClaudeTranscriptInspector;
|
||||
pub use claude_turn_watcher::ClaudeTranscriptTurnWatcher;
|
||||
|
||||
@ -48,7 +48,7 @@ pub use fs::LocalFileSystem;
|
||||
pub use git::Git2Repository;
|
||||
pub use id::UuidGenerator;
|
||||
pub use input::{MediatedInbox, MillisClock, SystemMillisClock};
|
||||
pub use inspector::ClaudeTranscriptInspector;
|
||||
pub use inspector::{ClaudeTranscriptInspector, ClaudeTranscriptTurnWatcher};
|
||||
pub use mailbox::InMemoryMailbox;
|
||||
pub use orchestrator::mcp::{
|
||||
resolve_ask_rendezvous_timeout, McpServer, MemoryTransport, StdioTransport,
|
||||
@ -77,7 +77,7 @@ pub use store::{
|
||||
AdaptiveMemoryRecall, EmbedderEnvProbe, FsEmbedderProfileStore, FsEmbedderPromptStore,
|
||||
FsLiveStateStore, FsMemoryStore, FsPermissionStore, FsProfileStore, FsProjectStore,
|
||||
FsSkillStore, FsTemplateStore, HashEmbedder, IdeaiContextStore, NaiveMemoryRecall,
|
||||
OnnxModelInfo, ReferenceBackfillProfileStore, StubEmbedder, VectorMemoryRecall,
|
||||
OnnxModelInfo, StubEmbedder, VectorMemoryRecall,
|
||||
DEFAULT_OLLAMA_BASE_URL, ONNX_CACHE_SUBDIR,
|
||||
RECOMMENDED_ONNX_MODELS, VECTOR_HTTP_ENABLED, VECTOR_ONNX_ENABLED,
|
||||
};
|
||||
|
||||
@ -41,17 +41,22 @@ const MCP_PROTOCOL_VERSION: &str = "2024-11-05";
|
||||
/// layer). The application layer already bounds the rendezvous itself, but this
|
||||
/// adapter adds its own **finite** outer bound so a wedged target can never park a
|
||||
/// `tools/call` task forever: on expiry the call returns a clean JSON-RPC error
|
||||
/// instead of hanging. Generous on purpose (delegated turns can be very long), but
|
||||
/// never infinite. Every other tool (`idea_reply`, `idea_list_agents`, …) is left
|
||||
/// untouched — they don't rendezvous.
|
||||
const ASK_RENDEZVOUS_TIMEOUT: Duration = Duration::from_secs(24 * 60 * 60);
|
||||
/// instead of hanging.
|
||||
///
|
||||
/// **Plancher universel FINI (backstop no-reply).** The real turn-end signal now exists
|
||||
/// (the transcript `turn_duration` watcher), but only Claude emits it; this outer bound
|
||||
/// stays the last resort for any silent target that never emits it (Codex & co) or that
|
||||
/// wedges. It must therefore NEVER be (quasi-)infinite. 600 s by default, overridable per
|
||||
/// project via `IDEA_ASK_RENDEZVOUS_TIMEOUT_MS`. Every other tool (`idea_reply`,
|
||||
/// `idea_list_agents`, …) is left untouched — they don't rendezvous.
|
||||
const ASK_RENDEZVOUS_TIMEOUT: Duration = Duration::from_secs(600);
|
||||
|
||||
/// Resolves the effective `idea_ask_agent` rendezvous bound from an optional
|
||||
/// configured override in **milliseconds** (project/profile level), mirroring the
|
||||
/// application's [`resolve_turn_timeout`](application) shape: `Some(ms)` ⇒ that bound,
|
||||
/// `None`/absent ⇒ the generous [`ASK_RENDEZVOUS_TIMEOUT`] default (statu quo, zero
|
||||
/// regression). `Some(0)` is treated as "no override" so a misconfigured zero can never
|
||||
/// collapse the safety net to an instant expiry. Pure and unit-testable without wiring.
|
||||
/// `None`/absent ⇒ the finite [`ASK_RENDEZVOUS_TIMEOUT`] default (600 s). `Some(0)` is
|
||||
/// treated as "no override" so a misconfigured zero can never collapse the safety net to
|
||||
/// an instant expiry. Pure and unit-testable without wiring.
|
||||
#[must_use]
|
||||
pub fn resolve_ask_rendezvous_timeout(override_ms: Option<u32>) -> Duration {
|
||||
match override_ms {
|
||||
|
||||
@ -28,7 +28,7 @@ pub use embedder::{
|
||||
pub use live_state::FsLiveStateStore;
|
||||
pub use memory::{index_token_size, FsMemoryStore, NaiveMemoryRecall};
|
||||
pub use permission::FsPermissionStore;
|
||||
pub use profile::{FsEmbedderProfileStore, FsProfileStore, ReferenceBackfillProfileStore};
|
||||
pub use profile::{FsEmbedderProfileStore, FsProfileStore};
|
||||
pub use project::FsProjectStore;
|
||||
pub use skill::FsSkillStore;
|
||||
pub use template::FsTemplateStore;
|
||||
|
||||
@ -148,59 +148,6 @@ impl ProfileStore for FsProfileStore {
|
||||
}
|
||||
}
|
||||
|
||||
/// **Merge-on-read decorator** over any [`ProfileStore`]: backfills *internal*
|
||||
/// reference defaults that a stored profile is missing, without ever persisting
|
||||
/// anything (finding A — propagate the measured `prompt_ready_pattern` to profiles
|
||||
/// saved before that field existed).
|
||||
///
|
||||
/// Only [`list`](ProfileStore::list) is transformed — each returned profile is run
|
||||
/// through [`application::overlay_reference_defaults`] (pure, allowlisted, never
|
||||
/// clobbers a user value, idempotent). `save`/`delete`/`is_configured`/
|
||||
/// `mark_configured` **delegate verbatim** so persistence and first-run detection are
|
||||
/// byte-for-byte unchanged. Wrap the concrete [`FsProfileStore`] once at the
|
||||
/// composition root, before injecting the store into use cases and the orchestrator,
|
||||
/// so every reader (including direct `.list()` calls) sees the overlaid reads.
|
||||
pub struct ReferenceBackfillProfileStore {
|
||||
inner: Arc<dyn ProfileStore>,
|
||||
}
|
||||
|
||||
impl ReferenceBackfillProfileStore {
|
||||
/// Wraps an inner [`ProfileStore`]; reads are overlaid, writes pass through.
|
||||
#[must_use]
|
||||
pub fn new(inner: Arc<dyn ProfileStore>) -> Self {
|
||||
Self { inner }
|
||||
}
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ProfileStore for ReferenceBackfillProfileStore {
|
||||
async fn list(&self) -> Result<Vec<AgentProfile>, StoreError> {
|
||||
Ok(self
|
||||
.inner
|
||||
.list()
|
||||
.await?
|
||||
.into_iter()
|
||||
.map(|p| application::overlay_reference_defaults(&p))
|
||||
.collect())
|
||||
}
|
||||
|
||||
async fn save(&self, profile: &AgentProfile) -> Result<(), StoreError> {
|
||||
self.inner.save(profile).await
|
||||
}
|
||||
|
||||
async fn delete(&self, id: ProfileId) -> Result<(), StoreError> {
|
||||
self.inner.delete(id).await
|
||||
}
|
||||
|
||||
async fn is_configured(&self) -> Result<bool, StoreError> {
|
||||
self.inner.is_configured().await
|
||||
}
|
||||
|
||||
async fn mark_configured(&self) -> Result<(), StoreError> {
|
||||
self.inner.mark_configured().await
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// FsEmbedderProfileStore — declarative embedder profiles (`embedder.json`, LOT C).
|
||||
// ---------------------------------------------------------------------------
|
||||
@ -340,87 +287,3 @@ impl EmbedderProfileStore for FsEmbedderProfileStore {
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod backfill_tests {
|
||||
use super::*;
|
||||
use std::sync::Mutex;
|
||||
|
||||
/// In-memory [`ProfileStore`] fake recording mutating/query calls, so we can prove
|
||||
/// the decorator transforms ONLY `list` and delegates the rest verbatim.
|
||||
#[derive(Default)]
|
||||
struct FakeStore {
|
||||
profiles: Mutex<Vec<AgentProfile>>,
|
||||
saved: Mutex<Vec<AgentProfile>>,
|
||||
deleted: Mutex<Vec<ProfileId>>,
|
||||
is_configured_calls: Mutex<u32>,
|
||||
mark_configured_calls: Mutex<u32>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl ProfileStore for FakeStore {
|
||||
async fn list(&self) -> Result<Vec<AgentProfile>, StoreError> {
|
||||
Ok(self.profiles.lock().unwrap().clone())
|
||||
}
|
||||
async fn save(&self, profile: &AgentProfile) -> Result<(), StoreError> {
|
||||
self.saved.lock().unwrap().push(profile.clone());
|
||||
Ok(())
|
||||
}
|
||||
async fn delete(&self, id: ProfileId) -> Result<(), StoreError> {
|
||||
self.deleted.lock().unwrap().push(id);
|
||||
Ok(())
|
||||
}
|
||||
async fn is_configured(&self) -> Result<bool, StoreError> {
|
||||
*self.is_configured_calls.lock().unwrap() += 1;
|
||||
Ok(true)
|
||||
}
|
||||
async fn mark_configured(&self) -> Result<(), StoreError> {
|
||||
*self.mark_configured_calls.lock().unwrap() += 1;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Claude reference profile with the internal field cleared — the persisted-before
|
||||
/// shape the overlay backfills.
|
||||
fn claude_without_pattern() -> AgentProfile {
|
||||
let id = application::reference_profile_id("claude");
|
||||
let mut p = application::reference_profiles()
|
||||
.into_iter()
|
||||
.find(|p| p.id == id)
|
||||
.expect("claude reference exists");
|
||||
p.prompt_ready_pattern = None;
|
||||
p
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn list_overlays_reference_defaults() {
|
||||
let fake = Arc::new(FakeStore::default());
|
||||
fake.profiles.lock().unwrap().push(claude_without_pattern());
|
||||
let store = ReferenceBackfillProfileStore::new(Arc::clone(&fake) as Arc<dyn ProfileStore>);
|
||||
|
||||
let listed = store.list().await.expect("list ok");
|
||||
assert_eq!(listed.len(), 1);
|
||||
assert_eq!(
|
||||
listed[0].prompt_ready_pattern.as_deref(),
|
||||
Some("? for shortcuts"),
|
||||
"a stored Claude profile WITHOUT the pattern is backfilled on read"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn save_delete_and_config_delegate_verbatim() {
|
||||
let fake = Arc::new(FakeStore::default());
|
||||
let store = ReferenceBackfillProfileStore::new(Arc::clone(&fake) as Arc<dyn ProfileStore>);
|
||||
|
||||
let p = claude_without_pattern();
|
||||
store.save(&p).await.expect("save ok");
|
||||
store.delete(p.id).await.expect("delete ok");
|
||||
assert!(store.is_configured().await.expect("is_configured ok"));
|
||||
store.mark_configured().await.expect("mark_configured ok");
|
||||
|
||||
// save passes the profile through UNCHANGED (no overlay on the write path).
|
||||
assert_eq!(*fake.saved.lock().unwrap(), vec![p.clone()]);
|
||||
assert_eq!(*fake.deleted.lock().unwrap(), vec![p.id]);
|
||||
assert_eq!(*fake.is_configured_calls.lock().unwrap(), 1);
|
||||
assert_eq!(*fake.mark_configured_calls.lock().unwrap(), 1);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user