diff --git a/crates/app-tauri/src/state.rs b/crates/app-tauri/src/state.rs index e8dcd8d..4618f88 100644 --- a/crates/app-tauri/src/state.rs +++ b/crates/app-tauri/src/state.rs @@ -22,7 +22,8 @@ use application::{ ListTemplates, LiveAgentRegistry, LoadLayout, MoveTabToNewWindow, MutateLayout, OnnxModelView, OpenProject, OpenTerminal, OrchestratorService, ReadAgentContext, ReadMemoryIndex, ReadProjectContext, - RecallMemory, ReconcileLayouts, ReferenceProfiles, RenameLayout, ResizeTerminal, + RecallMemory, ReconcileLayouts, RecordTurn, RecordTurnProvider, ReferenceProfiles, + RenameLayout, ResizeTerminal, ResolveMemoryLinks, SaveEmbedderProfile, SaveProfile, SetActiveLayout, SnapshotRunningAgents, StructuredSessions, SuggestedThisSession, @@ -40,7 +41,8 @@ use domain::{DomainEvent, EmbedderProfile, Project, ProjectId}; use infrastructure::{ embedder_from_profile, AdaptiveMemoryRecall, ClaudeTranscriptInspector, CliAgentRuntime, - EmbedderEnvProbe, FsEmbedderProfileStore, FsEmbedderPromptStore, FsMemoryStore, + EmbedderEnvProbe, FsConversationLog, FsEmbedderProfileStore, FsEmbedderPromptStore, + FsHandoffStore, FsMemoryStore, HeuristicHandoffSummarizer, FsOrchestratorWatcher, FsProfileStore, FsProjectStore, FsSkillStore, FsTemplateStore, Git2Repository, IdeaiContextStore, InMemoryConversationRegistry, InMemoryMailbox, LocalFileSystem, LocalProcessSpawner, McpServer, MediatedInbox, SystemMillisClock, @@ -61,6 +63,25 @@ use interprocess::local_socket::traits::tokio::Listener as _; use interprocess::local_socket::{GenericFilePath, ListenerOptions, ToFsName}; use tokio::io::{AsyncBufReadExt, BufReader}; +/// Implémente [`RecordTurnProvider`] (lot P6b) en matérialisant un [`RecordTurn`] +/// ciblant le **project root** du tour en cours. +/// +/// L'[`OrchestratorService`] est unique pour tous les projets, alors que le log/handoff +/// conversationnel est **par project root** (`/.ideai/conversations/`). Les +/// adapters `Fs*` fixent leur racine à la construction et ne sont que des jointures de +/// chemin : on en construit donc un jeu frais par tour, ciblant le bon dossier. Sans +/// état (zéro champ), partagé via un simple `Arc`. +struct AppRecordTurnProvider; + +impl RecordTurnProvider for AppRecordTurnProvider { + fn record_turn_for(&self, root: &domain::project::ProjectPath) -> Option> { + let log = Arc::new(FsConversationLog::new(root)); + let handoffs = Arc::new(FsHandoffStore::new(root)); + let summarizer = Arc::new(HeuristicHandoffSummarizer::new()); + Some(Arc::new(RecordTurn::new(log, handoffs, summarizer))) + } +} + /// Everything the IPC layer needs at runtime, managed by Tauri. /// /// Use cases are stored behind `Arc` so handlers clone cheaply. The concrete @@ -785,6 +806,14 @@ impl AppState { // réelle ⇒ le pont MCP de la cible se spawne et `idea_reply` débloque le round-trip. .with_mcp_runtime_provider( Arc::new(AppMcpRuntimeProvider) as Arc + ) + // Persistance conversationnelle best-effort (lot P6b) : Prompt + Response de + // chaque paire déléguée écrits dans `/.ideai/conversations/` via + // le provider per-root + l'horloge millis (port Clock) déjà câblée. Un échec + // n'affecte jamais la délégation live. + .with_record_turn( + Arc::new(AppRecordTurnProvider) as Arc, + Arc::clone(&clock) as Arc, ), ); diff --git a/crates/application/src/lib.rs b/crates/application/src/lib.rs index 042ce85..9cb380d 100644 --- a/crates/application/src/lib.rs +++ b/crates/application/src/lib.rs @@ -73,7 +73,9 @@ pub use memory::{ RecallMemoryInput, RecallMemoryOutput, ResolveMemoryLinks, ResolveMemoryLinksInput, ResolveMemoryLinksOutput, UpdateMemory, UpdateMemoryInput, UpdateMemoryOutput, }; -pub use orchestrator::{McpRuntimeProvider, OrchestratorOutcome, OrchestratorService}; +pub use orchestrator::{ + McpRuntimeProvider, OrchestratorOutcome, OrchestratorService, RecordTurnProvider, +}; pub use project::{ CloseProject, CloseProjectInput, CloseProjectOutput, CloseTab, CloseTabInput, CreateProject, CreateProjectInput, CreateProjectOutput, ListProjects, ListProjectsOutput, OpenProject, diff --git a/crates/application/src/orchestrator/mod.rs b/crates/application/src/orchestrator/mod.rs index bf77476..0d13586 100644 --- a/crates/application/src/orchestrator/mod.rs +++ b/crates/application/src/orchestrator/mod.rs @@ -11,4 +11,6 @@ pub use context_guard::{ ProposeContext, ProposeContextInput, ProposeOutcome, ReadContext, ReadContextInput, ReadMemory, ReadMemoryInput, WriteMemory, WriteMemoryInput, }; -pub use service::{McpRuntimeProvider, OrchestratorOutcome, OrchestratorService}; +pub use service::{ + McpRuntimeProvider, OrchestratorOutcome, OrchestratorService, RecordTurnProvider, +}; diff --git a/crates/application/src/orchestrator/service.rs b/crates/application/src/orchestrator/service.rs index 0ecefff..6eb13f8 100644 --- a/crates/application/src/orchestrator/service.rs +++ b/crates/application/src/orchestrator/service.rs @@ -22,11 +22,15 @@ use tokio::sync::Mutex as AsyncMutex; use domain::conversation::{ ConversationParty, ConversationRegistry, SessionRef, WaitForGraph, }; -use domain::input::InputMediator; +use domain::conversation_log::{ConversationTurn, TurnId, TurnRole}; +use domain::input::{InputMediator, InputSource}; use domain::mailbox::{Ticket, TicketId}; -use domain::ports::{EventBus, ProfileStore, PtyHandle}; +use domain::ports::{Clock, EventBus, ProfileStore, PtyHandle}; +use domain::project::ProjectPath; use domain::{AgentId, DomainEvent, OrchestratorCommand, OrchestratorVisibility, ProfileId, Project}; +use crate::conversation::RecordTurn; + use crate::agent::{ CreateAgentFromScratch, CreateAgentInput, LaunchAgent, LaunchAgentInput, ListAgents, ListAgentsInput, McpRuntime, ReattachDecision, UpdateAgentContext, UpdateAgentContextInput, @@ -83,6 +87,27 @@ pub trait McpRuntimeProvider: Send + Sync { fn runtime_for(&self, project: &Project, agent_id: AgentId) -> Option; } +/// Fournit le use case [`RecordTurn`] **lié au project root** de la délégation en +/// cours (lot P6b). +/// +/// L'[`OrchestratorService`] est **unique et partagé par tous les projets ouverts** +/// (un seul `Arc` au composition root), alors que le log/handoff conversationnel est +/// **par project root** (`/.ideai/conversations/`, comme la mémoire). Les +/// adapters `Fs*` de P6a fixent leur racine à la construction et leur port ne porte +/// pas le root par appel ; on ne peut donc pas figer un `RecordTurn` global. Ce +/// **port** lève la tension : `ask_agent` connaît le `project.root` du tour et demande +/// au provider de matérialiser un `RecordTurn` ciblant le **bon** dossier (les adapters +/// `Fs*` ne font que des jointures de chemin, leur construction est triviale). +/// +/// `None` ⇒ aucune persistance (best-effort absente) : zéro régression pour les call +/// sites/tests qui ne le branchent pas. Implémenté dans app-tauri (seul détenteur des +/// adapters `Fs*`). +pub trait RecordTurnProvider: Send + Sync { + /// Construit le [`RecordTurn`] dont le log/handoff ciblent `root`. Appelé une fois + /// par checkpoint best-effort ; `None` ⇒ on saute silencieusement la persistance. + fn record_turn_for(&self, root: &ProjectPath) -> Option>; +} + /// Dispatches validated orchestrator commands to the agent/terminal use cases. pub struct OrchestratorService { create_agent: Arc, @@ -147,6 +172,16 @@ pub struct OrchestratorService { /// [`Self::with_context_guard`] ; `None` ⇒ les commandes `context.*`/`memory.*` /// renvoient une erreur typée (call sites/tests legacy restent verts). context_guard: Option>, + /// Provider du use case [`RecordTurn`] **par project root** (lot P6b), pour + /// persister **best-effort** le Prompt et la Response de chaque paire déléguée dans + /// le bon `conversationId`. Injecté avec son horloge via [`Self::with_record_turn`] ; + /// `None` ⇒ aucune persistance (zéro régression pour les call sites/tests legacy). + /// Un échec de persistance ne transforme **jamais** un succès de délégation en erreur. + record_turn: Option>, + /// Horloge millis (port [`Clock`]) pour estampiller `at_ms` des tours persistés — + /// injectée avec [`Self::record_turn`]. La couche `application` reste **pure** : pas + /// de `SystemTime::now()` brut ici, le temps vient du port injecté au composition root. + clock: Option>, } /// Bundle des quatre use cases C7 sous [`domain::fileguard::FileGuard`], injectés @@ -207,6 +242,8 @@ impl OrchestratorService { ask_locks: StdMutex::new(HashMap::new()), mcp_runtime_provider: None, context_guard: None, + record_turn: None, + clock: None, } } @@ -279,6 +316,58 @@ impl OrchestratorService { self } + /// Branche le provider de [`RecordTurn`] **par project root** + son horloge (lot + /// P6b) pour persister **best-effort** le Prompt et la Response de chaque paire + /// déléguée. Builder additif : signature de [`Self::new`] **inchangée** (les + /// tests/call sites legacy restent verts ; `None` ⇒ aucune persistance, donc aucune + /// régression). Un échec de persistance ne casse jamais la délégation live. + #[must_use] + pub fn with_record_turn( + mut self, + record_turn: Arc, + clock: Arc, + ) -> Self { + self.record_turn = Some(record_turn); + self.clock = Some(clock); + self + } + + /// Persiste **best-effort** un tour (`Prompt`/`Response`) dans `conversation`. + /// + /// No-op silencieux quand le provider/horloge ne sont pas câblés, quand le provider + /// ne rend pas de [`RecordTurn`] pour ce root, ou quand l'`append`/`save` échoue : la + /// persistance ne doit **jamais** transformer un succès de délégation en erreur, ni + /// paniquer (contrat P6b). N'ajoute pas de latence inutile (un seul `await` borné par + /// les adapters `Fs*`, déjà sérialisés par le verrou de tour de la cible). + async fn record_turn_best_effort( + &self, + root: &ProjectPath, + conversation: domain::conversation::ConversationId, + source: InputSource, + role: TurnRole, + text: String, + ) { + let (Some(provider), Some(clock)) = (&self.record_turn, &self.clock) else { + return; + }; + let Some(record) = provider.record_turn_for(root) else { + return; + }; + let at_ms = u64::try_from(clock.now_millis()).unwrap_or(0); + let turn = ConversationTurn::new( + TurnId::new_random(), + conversation, + at_ms, + source, + role, + text, + ); + // Best-effort : un échec de persistance est avalé (le contrat P6b interdit qu'il + // remonte). Pas de framework de log dans `application` ; on reste cohérent avec + // les autres effets best-effort du service (cf. publication `AgentReplied`). + let _ = record.record(conversation, turn).await; + } + /// Dispatches a validated command against `project`. /// /// # Errors @@ -674,6 +763,21 @@ impl OrchestratorService { // porte la source (Human/Agent) et la conversation cible. let requester_label = self.requester_label(project, requester).await; let ticket_id = TicketId::new_random(); + // Checkpoint Prompt (P6b, best-effort) : persister l'invite AVANT que `task` ne + // soit déplacé dans le `Ticket`. Source = origine de la requête (agent demandeur + // `Some(from)` ⇒ Agent, sinon Humain) — **même** `InputSource` que le ticket. + let prompt_source = match requester { + Some(from) => InputSource::agent(from), + None => InputSource::Human, + }; + self.record_turn_best_effort( + &project.root, + conversation_id, + prompt_source, + TurnRole::Prompt, + task.clone(), + ) + .await; let ticket = match requester { Some(from) => { Ticket::from_agent(ticket_id, from, conversation_id, requester_label, task) @@ -688,7 +792,20 @@ impl OrchestratorService { // 3. Attendre la réponse, bornée. Timeout/canal fermé ⇒ retirer le ticket // (cible laissée vivante) et renvoyer une erreur typée. match tokio::time::timeout(ASK_AGENT_TIMEOUT, pending).await { - Ok(Ok(result)) => Ok(self.reply_outcome(agent_id, &target, result)), + Ok(Ok(result)) => { + // Checkpoint Response (P6b, best-effort) : persister la réponse AVANT de + // déplacer `result` dans `reply_outcome`. Source = la **cible** (c'est + // elle qui a rendu le tour) ; même conversation que le Prompt. + self.record_turn_best_effort( + &project.root, + conversation_id, + InputSource::agent(agent_id), + TurnRole::Response, + result.clone(), + ) + .await; + Ok(self.reply_outcome(agent_id, &target, result)) + } Ok(Err(_cancelled)) => { mailbox.cancel_head(agent_id, ticket_id); Err(AppError::Process(format!( diff --git a/crates/application/tests/orchestrator_service.rs b/crates/application/tests/orchestrator_service.rs index 60c5a2c..907c27b 100644 --- a/crates/application/tests/orchestrator_service.rs +++ b/crates/application/tests/orchestrator_service.rs @@ -2306,3 +2306,392 @@ async fn interrupt_unknown_agent_is_not_found() { assert_eq!(err.code(), "NOT_FOUND", "unknown agent interrupt: {err:?}"); assert!(fx.mediator.preempts().is_empty(), "no preempt on unknown agent"); } + +// --------------------------------------------------------------------------- +// P6b — live wiring of `RecordTurn` in `ask_agent` (conversation persistence). +// +// On a successful inter-agent delegation, `ask_agent` persists the **pair** +// best-effort: a `Prompt` turn at enqueue (text = task, source = requester) and a +// `Response` turn on the `Ok(Ok(result))` branch (text = result, source = target), +// both on the SAME resolved `conversation_id`. Persistence is best-effort: a missing +// provider/clock, a provider returning `None`, or a failing append must NEVER turn a +// successful ask into an error. +// +// These tests reuse the full `ask`/`reply` rendezvous harness above (real mailbox + +// mediated PTY delivery + conversation registry) and add a recording in-memory +// `ConversationLog` / `RecordTurnProvider` / fixed `Clock`, mirroring the P6a fakes +// in `conversation_record.rs`. +// --------------------------------------------------------------------------- + +use domain::{ + ConversationLog, ConversationTurn as DomainTurn, Handoff, HandoffStore, HandoffSummarizer, + TurnId, TurnRole, +}; +use application::{OrchestratorOutcome, RecordTurn, RecordTurnProvider}; +use domain::ports::Clock; +use domain::InputSource; + +/// A deterministic [`Clock`]: every persisted turn is stamped with this constant. +struct FixedClock(i64); +impl Clock for FixedClock { + fn now_millis(&self) -> i64 { + self.0 + } +} + +/// In-memory [`ConversationLog`] that **records the order and content** of every +/// append so a test can assert the exact `[Prompt, Response]` sequence and inspect +/// each turn's `text` / `role` / `source` / `conversation`. Optionally fails on +/// `append` (best-effort path). +#[derive(Default)] +struct RecordingLog { + appends: Mutex>, + fail_append: bool, +} +impl RecordingLog { + fn failing() -> Self { + Self { + appends: Mutex::new(Vec::new()), + fail_append: true, + } + } + fn appends(&self) -> Vec { + self.appends.lock().unwrap().clone() + } +} +#[async_trait] +impl ConversationLog for RecordingLog { + async fn append( + &self, + _conversation: ConversationId, + turn: DomainTurn, + ) -> Result<(), StoreError> { + if self.fail_append { + return Err(StoreError::Io("recording log: forced append failure".to_owned())); + } + self.appends.lock().unwrap().push(turn); + Ok(()) + } + async fn read( + &self, + _conversation: ConversationId, + _since: Option, + ) -> Result, StoreError> { + Ok(self.appends()) + } + async fn last( + &self, + _conversation: ConversationId, + n: usize, + ) -> Result, StoreError> { + let all = self.appends(); + let start = all.len().saturating_sub(n); + Ok(all[start..].to_vec()) + } +} + +/// Trivial in-memory [`HandoffStore`] — the P6b tests only assert on the log; the +/// handoff is exercised end-to-end by P6a. Kept minimal. +#[derive(Default)] +struct NoopHandoffStore(Mutex>); +#[async_trait] +impl HandoffStore for NoopHandoffStore { + async fn load(&self, conversation: ConversationId) -> Result, StoreError> { + Ok(self.0.lock().unwrap().get(&conversation).cloned()) + } + async fn save( + &self, + conversation: ConversationId, + handoff: Handoff, + ) -> Result<(), StoreError> { + self.0.lock().unwrap().insert(conversation, handoff); + Ok(()) + } +} + +/// Deterministic summarizer: folds the new turns' text onto the previous summary. +#[derive(Default)] +struct ConcatSummarizer; +#[async_trait] +impl HandoffSummarizer for ConcatSummarizer { + async fn fold(&self, prev: Option, new_turns: &[DomainTurn]) -> Handoff { + let mut summary = prev.map(|h| h.summary_md).unwrap_or_default(); + for t in new_turns { + summary.push_str(&t.text); + } + let up_to = new_turns.last().map(|t| t.id).expect("at least one new turn"); + Handoff::new(summary, up_to, None) + } +} + +/// A [`RecordTurnProvider`] that always materialises the **same** shared +/// [`RecordTurn`] (its log is observable by the test), ignoring the root — the +/// fixture pins a single project, so root-routing is out of scope here. +struct SharedRecordProvider(Arc); +impl RecordTurnProvider for SharedRecordProvider { + fn record_turn_for(&self, _root: &ProjectPath) -> Option> { + Some(Arc::clone(&self.0)) + } +} + +/// A [`RecordTurnProvider`] that always declines (no `RecordTurn` for this root) — +/// drives the "provider returns None ⇒ silent no-op" branch. +struct NoneRecordProvider; +impl RecordTurnProvider for NoneRecordProvider { + fn record_turn_for(&self, _root: &ProjectPath) -> Option> { + None + } +} + +/// Builds the same `ask` harness as [`ask_fixture`] but additionally wires +/// `with_record_turn(provider, clock)`. Returns the [`AskFixture`] plus the shared +/// recording log (when one was provided) for assertion. +fn ask_fixture_with_record( + contexts: FakeContexts, + provider: Arc, + clock_ms: i64, +) -> AskFixture { + let profiles = Arc::new(FakeProfiles::new(vec![claude_profile()])); + let sessions = Arc::new(TerminalSessions::new()); + let pty = FakePty::new(sid(777)); + let bus = SpyBus::default(); + let mailbox = Arc::new(TestMailbox::new()); + + let create = Arc::new(CreateAgentFromScratch::new( + Arc::new(contexts.clone()), + Arc::new(SeqIds::new()), + Arc::new(bus.clone()), + )); + let launch = Arc::new(LaunchAgent::new( + Arc::new(contexts.clone()), + Arc::clone(&profiles) as Arc, + Arc::new(FakeRuntime), + Arc::new(FakeFs), + Arc::new(pty.clone()), + Arc::new(FakeSkills), + Arc::clone(&sessions), + Arc::new(bus.clone()), + Arc::new(SeqIds::new()), + Arc::new(FakeRecall), + None, + )); + let list = Arc::new(ListAgents::new(Arc::new(contexts.clone()))); + let close = Arc::new(CloseTerminal::new( + Arc::new(pty.clone()), + Arc::clone(&sessions), + )); + let update = Arc::new(UpdateAgentContext::new(Arc::new(contexts.clone()))); + let create_skill = Arc::new(CreateSkill::new( + Arc::new(RecordingSkills::default()) as Arc, + Arc::new(SeqIds::new()), + )); + + let mediator = Arc::new(TestMediator::new( + Arc::clone(&mailbox), + Arc::new(pty.clone()) as Arc, + )); + let conversations = Arc::new(TestConversations::new()) as Arc; + let service = Arc::new( + OrchestratorService::new( + create, + launch, + list, + close, + update, + create_skill, + Arc::clone(&profiles) as Arc, + Arc::clone(&sessions), + ) + .with_input_mediator( + Arc::clone(&mediator) as Arc, + Arc::clone(&mailbox) as Arc, + ) + .with_conversations(conversations) + .with_events(Arc::new(bus.clone())) + .with_record_turn(provider, Arc::new(FixedClock(clock_ms)) as Arc), + ); + + AskFixture { + service, + mailbox, + sessions, + bus, + pty, + mediator, + } +} + +/// Drives a full `ask` round-trip (requester `from`, target architect) on the given +/// service, replying `result`. Returns the dispatch outcome. +async fn run_ask_roundtrip( + fx: &AskFixture, + ask_json: &str, + reply_from: AgentId, + result: &str, +) -> OrchestratorOutcome { + let svc = Arc::clone(&fx.service); + let json = ask_json.to_owned(); + let ask = tokio::spawn(async move { svc.dispatch(&project(), cmd(&json)).await }); + await_until(|| fx.mailbox.pending(&reply_from) == 1).await; + fx.service + .dispatch(&project(), reply_cmd(reply_from, result)) + .await + .expect("reply ok"); + timeout(TEST_GUARD, ask) + .await + .expect("ask completes once replied") + .expect("join ok") + .expect("ask ok") +} + +/// Round-trip from the **User** (requester `None`) writes exactly two turns on the +/// SAME conversation, in order Prompt then Response, with the right text/role/source. +#[tokio::test] +async fn p6b_user_ask_records_prompt_then_response_pair() { + let agent = scratch_agent(aid(1), "architect", "agents/architect.md"); + let log = Arc::new(RecordingLog::default()); + let record = Arc::new(RecordTurn::new( + Arc::clone(&log) as Arc, + Arc::new(NoopHandoffStore::default()) as Arc, + Arc::new(ConcatSummarizer) as Arc, + )); + let provider = Arc::new(SharedRecordProvider(record)) as Arc; + let fx = ask_fixture_with_record(FakeContexts::with_agent(&agent, "# persona"), provider, 123); + seed_live_pty(&fx.sessions, aid(1), sid(800)); + + // ASK_JSON uses `requestedBy:"Main"` but no agent requester id ⇒ User-sourced. + let out = run_ask_roundtrip(&fx, ASK_JSON, aid(1), "the §17 answer").await; + assert_eq!(out.reply.as_deref(), Some("the §17 answer")); + + let appends = log.appends(); + assert_eq!(appends.len(), 2, "exactly two turns persisted (Prompt + Response)"); + + let prompt = &appends[0]; + let response = &appends[1]; + // Same conversation for both turns. + assert_eq!( + prompt.conversation, response.conversation, + "both turns share the resolved conversation id" + ); + // Order + role. + assert_eq!(prompt.role, TurnRole::Prompt, "first turn is the Prompt"); + assert_eq!(response.role, TurnRole::Response, "second turn is the Response"); + // Text: task then result. + assert_eq!(prompt.text, "Analyse §17", "prompt text is the delegated task"); + assert_eq!(response.text, "the §17 answer", "response text is the reply result"); + // Source: Human prompt (no agent requester), target-sourced response. + assert_eq!(prompt.source, InputSource::Human, "User ask ⇒ Human prompt source"); + assert_eq!( + response.source, + InputSource::agent(aid(1)), + "response is sourced from the target (architect)" + ); + // Clock stamp came from the injected Clock. + assert_eq!(prompt.at_ms, 123); + assert_eq!(response.at_ms, 123); +} + +/// Round-trip with an **agent** requester (A) resolves the `A↔B` conversation and the +/// Prompt source is `agent(A)`; the Response source remains the target B. +#[tokio::test] +async fn p6b_agent_requester_records_pair_on_a_b_thread() { + let architect = scratch_agent(aid(1), "architect", "agents/architect.md"); + let log = Arc::new(RecordingLog::default()); + let record = Arc::new(RecordTurn::new( + Arc::clone(&log) as Arc, + Arc::new(NoopHandoffStore::default()) as Arc, + Arc::new(ConcatSummarizer) as Arc, + )); + let provider = Arc::new(SharedRecordProvider(record)) as Arc; + let fx = + ask_fixture_with_record(FakeContexts::with_agent(&architect, "# persona"), provider, 0); + seed_live_pty(&fx.sessions, aid(1), sid(800)); + + // Requester is agent A (id 7). The orchestrator threads `requester` from the + // `requestedBy` field of `agent.message` (parsed as a uuid when well-formed). + let a = aid(7); + let a_uuid = uuid::Uuid::from_u128(7); + let ask_json = format!( + r#"{{ "type":"agent.message", "requestedBy":"{a_uuid}", "targetAgent":"architect", "task":"Analyse §17" }}"# + ); + let out = run_ask_roundtrip(&fx, &ask_json, aid(1), "ack").await; + assert_eq!(out.reply.as_deref(), Some("ack")); + + let appends = log.appends(); + assert_eq!(appends.len(), 2, "two turns persisted"); + let (prompt, response) = (&appends[0], &appends[1]); + + // Independently resolve the A↔B conversation the SAME way the service does and + // assert both turns landed on it. + let convs = TestConversations::new(); + let expected = convs + .resolve(ConversationParty::agent(a), ConversationParty::agent(aid(1))) + .id; + // (Resolution is deterministic per pair within one registry; we instead assert the + // turns share a thread and the prompt source carries A — the load-bearing facts.) + let _ = expected; + assert_eq!(prompt.conversation, response.conversation, "shared thread"); + assert_eq!( + prompt.source, + InputSource::agent(a), + "agent requester ⇒ Prompt sourced from A" + ); + assert_eq!(prompt.role, TurnRole::Prompt); + assert_eq!(response.role, TurnRole::Response); + assert_eq!( + response.source, + InputSource::agent(aid(1)), + "Response sourced from the target B" + ); +} + +/// No provider wired (plain `ask_fixture`) ⇒ the round-trip still succeeds and nothing +/// is persisted (no panic, no error). +#[tokio::test] +async fn p6b_no_provider_is_silent_no_op() { + let agent = scratch_agent(aid(1), "architect", "agents/architect.md"); + // `ask_fixture` does NOT call `with_record_turn`. + let fx = ask_fixture(FakeContexts::with_agent(&agent, "# persona")); + seed_live_pty(&fx.sessions, aid(1), sid(800)); + + let out = run_ask_roundtrip(&fx, ASK_JSON, aid(1), "ok").await; + assert_eq!(out.reply.as_deref(), Some("ok"), "ask succeeds without persistence"); +} + +/// Provider wired but returns `None` for the root ⇒ ask still succeeds (best-effort +/// skip). +#[tokio::test] +async fn p6b_provider_returns_none_is_silent_no_op() { + let agent = scratch_agent(aid(1), "architect", "agents/architect.md"); + let provider = Arc::new(NoneRecordProvider) as Arc; + let fx = ask_fixture_with_record(FakeContexts::with_agent(&agent, "# persona"), provider, 0); + seed_live_pty(&fx.sessions, aid(1), sid(800)); + + let out = run_ask_roundtrip(&fx, ASK_JSON, aid(1), "ok").await; + assert_eq!(out.reply.as_deref(), Some("ok"), "ask succeeds when provider declines"); +} + +/// `RecordTurn` built on a log whose `append` always fails ⇒ persistence errors are +/// swallowed; the delegation result is NOT degraded (ask returns `Ok`). +#[tokio::test] +async fn p6b_failing_record_does_not_degrade_ask() { + let agent = scratch_agent(aid(1), "architect", "agents/architect.md"); + let failing_log = Arc::new(RecordingLog::failing()); + let record = Arc::new(RecordTurn::new( + Arc::clone(&failing_log) as Arc, + Arc::new(NoopHandoffStore::default()) as Arc, + Arc::new(ConcatSummarizer) as Arc, + )); + let provider = Arc::new(SharedRecordProvider(record)) as Arc; + let fx = ask_fixture_with_record(FakeContexts::with_agent(&agent, "# persona"), provider, 0); + seed_live_pty(&fx.sessions, aid(1), sid(800)); + + let out = run_ask_roundtrip(&fx, ASK_JSON, aid(1), "still ok").await; + assert_eq!( + out.reply.as_deref(), + Some("still ok"), + "a failing append must not turn a successful delegation into an error" + ); + // The failing log recorded nothing (every append errored). + assert!(failing_log.appends().is_empty(), "no turns stored when append fails"); +}