feat(persistence): P6b — câblage live du checkpoint conversationnel (best-effort)
ask_agent persiste désormais chaque paire dans son conversationId : tour Prompt à l'enqueue, tour Response au succès — best-effort (un échec de persistance ne dégrade jamais la délégation). - application : port RecordTurnProvider (matérialise un RecordTurn sur le bon project root — OrchestratorService est mono-instance multi-projets, le log est par root) ; wither with_record_turn(provider, Clock) ; helper record_turn_best_effort ; horodatage via domain::ports::Clock (pas d'horloge infra dans application) - app-tauri : AppRecordTurnProvider (Fs* sur le root du projet) câblé au composition root avec le SystemClock partagé - tests : 5 cas (paire Prompt→Response, fil A↔B vs User↔B, no-op sans provider, ask Ok même si record échoue) ; orchestrator_service 40, application + app-tauri verts Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@ -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<Vec<DomainTurn>>,
|
||||
fail_append: bool,
|
||||
}
|
||||
impl RecordingLog {
|
||||
fn failing() -> Self {
|
||||
Self {
|
||||
appends: Mutex::new(Vec::new()),
|
||||
fail_append: true,
|
||||
}
|
||||
}
|
||||
fn appends(&self) -> Vec<DomainTurn> {
|
||||
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<TurnId>,
|
||||
) -> Result<Vec<DomainTurn>, StoreError> {
|
||||
Ok(self.appends())
|
||||
}
|
||||
async fn last(
|
||||
&self,
|
||||
_conversation: ConversationId,
|
||||
n: usize,
|
||||
) -> Result<Vec<DomainTurn>, 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<HashMap<ConversationId, Handoff>>);
|
||||
#[async_trait]
|
||||
impl HandoffStore for NoopHandoffStore {
|
||||
async fn load(&self, conversation: ConversationId) -> Result<Option<Handoff>, 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<Handoff>, 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<RecordTurn>);
|
||||
impl RecordTurnProvider for SharedRecordProvider {
|
||||
fn record_turn_for(&self, _root: &ProjectPath) -> Option<Arc<RecordTurn>> {
|
||||
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<Arc<RecordTurn>> {
|
||||
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<dyn RecordTurnProvider>,
|
||||
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<dyn ProfileStore>,
|
||||
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<dyn SkillStore>,
|
||||
Arc::new(SeqIds::new()),
|
||||
));
|
||||
|
||||
let mediator = Arc::new(TestMediator::new(
|
||||
Arc::clone(&mailbox),
|
||||
Arc::new(pty.clone()) as Arc<dyn PtyPort>,
|
||||
));
|
||||
let conversations = Arc::new(TestConversations::new()) as Arc<dyn ConversationRegistry>;
|
||||
let service = Arc::new(
|
||||
OrchestratorService::new(
|
||||
create,
|
||||
launch,
|
||||
list,
|
||||
close,
|
||||
update,
|
||||
create_skill,
|
||||
Arc::clone(&profiles) as Arc<dyn ProfileStore>,
|
||||
Arc::clone(&sessions),
|
||||
)
|
||||
.with_input_mediator(
|
||||
Arc::clone(&mediator) as Arc<dyn InputMediator>,
|
||||
Arc::clone(&mailbox) as Arc<dyn AgentMailbox>,
|
||||
)
|
||||
.with_conversations(conversations)
|
||||
.with_events(Arc::new(bus.clone()))
|
||||
.with_record_turn(provider, Arc::new(FixedClock(clock_ms)) as Arc<dyn Clock>),
|
||||
);
|
||||
|
||||
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<dyn ConversationLog>,
|
||||
Arc::new(NoopHandoffStore::default()) as Arc<dyn HandoffStore>,
|
||||
Arc::new(ConcatSummarizer) as Arc<dyn HandoffSummarizer>,
|
||||
));
|
||||
let provider = Arc::new(SharedRecordProvider(record)) as Arc<dyn RecordTurnProvider>;
|
||||
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<dyn ConversationLog>,
|
||||
Arc::new(NoopHandoffStore::default()) as Arc<dyn HandoffStore>,
|
||||
Arc::new(ConcatSummarizer) as Arc<dyn HandoffSummarizer>,
|
||||
));
|
||||
let provider = Arc::new(SharedRecordProvider(record)) as Arc<dyn RecordTurnProvider>;
|
||||
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<dyn RecordTurnProvider>;
|
||||
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<dyn ConversationLog>,
|
||||
Arc::new(NoopHandoffStore::default()) as Arc<dyn HandoffStore>,
|
||||
Arc::new(ConcatSummarizer) as Arc<dyn HandoffSummarizer>,
|
||||
));
|
||||
let provider = Arc::new(SharedRecordProvider(record)) as Arc<dyn RecordTurnProvider>;
|
||||
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");
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user