From 78ad3bc15335dcedafd0d01eaa36b352b5eb4c15 Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Mon, 10 Aug 2026 15:59:28 +0200 Subject: [PATCH 1/2] fix(cli): Re-read the conversation index for each plugin read A plugin host loads the conversation index once at startup and does not own the store. A `jp query` in another terminal, or another plugin, appends events to a conversation whose metadata and stream this process cached and would otherwise keep serving for its lifetime. A conversation started elsewhere never appeared at all. `list_conversations` and `read_events` re-read the index first, which drops both caches so the read that follows comes from disk. The scan is a directory listing per storage root; metadata and streams stay lazy, so only what the request actually reads is loaded again. Not a sanitize pass. That repairs a store on startup and can move broken conversations aside, which is not something a page view should do. Signed-off-by: Jean Mertz --- crates/jp_cli/src/cmd/plugin/dispatch.rs | 22 ++++++++ .../jp_cli/src/cmd/plugin/dispatch_tests.rs | 50 +++++++++++++++++++ 2 files changed, 72 insertions(+) diff --git a/crates/jp_cli/src/cmd/plugin/dispatch.rs b/crates/jp_cli/src/cmd/plugin/dispatch.rs index 24203e51a..02612e36f 100644 --- a/crates/jp_cli/src/cmd/plugin/dispatch.rs +++ b/crates/jp_cli/src/cmd/plugin/dispatch.rs @@ -1076,11 +1076,13 @@ fn handle_request( } PluginToHost::ListConversations(req) => { + refresh_conversations(workspace); let response = handle_list_conversations(workspace, req.id); write_message(writer, &response)?; } PluginToHost::ReadEvents(req) => { + refresh_conversations(workspace); let response = handle_read_events(workspace, &req.conversation, req.id); write_message(writer, &response)?; } @@ -1563,6 +1565,26 @@ fn handle_list_configs( HostToPlugin::Configs(ConfigsResponse { id: req_id, data }) } +/// Re-read the conversation index, dropping what this process has cached. +/// +/// A plugin host is long-lived and does not own the store: a `jp query` in +/// another terminal, or another plugin, appends events to a conversation whose +/// metadata and stream this process loaded once and would otherwise keep +/// serving forever. +/// Re-reading the index clears both caches, so the read that follows comes from +/// disk, and conversations created or deleted since startup appear and +/// disappear. +/// +/// The scan is a directory listing per storage root; metadata and streams stay +/// lazy, so only what the request actually reads is loaded again. +/// +/// Deliberately not a sanitize pass: that repairs a store on startup and can +/// move broken conversations aside, which is not a thing a page view should do. +fn refresh_conversations(workspace: &mut Workspace) { + trace!("Re-reading the conversation index for a plugin request."); + workspace.load_conversation_index(); +} + fn handle_list_conversations(workspace: &Workspace, req_id: Option) -> HostToPlugin { let data: Vec = workspace .conversations() diff --git a/crates/jp_cli/src/cmd/plugin/dispatch_tests.rs b/crates/jp_cli/src/cmd/plugin/dispatch_tests.rs index b11f2b7d7..26cef116c 100644 --- a/crates/jp_cli/src/cmd/plugin/dispatch_tests.rs +++ b/crates/jp_cli/src/cmd/plugin/dispatch_tests.rs @@ -561,6 +561,56 @@ fn an_unknown_conversation_fails_against_its_request() { } } +/// A long-running host sees what another process wrote after it started. +/// +/// The host loads the index once at startup. +/// Without re-reading it, a plugin asking for the conversation list is served +/// that snapshot for the life of the process, so a conversation started in a +/// terminal never appears. +#[tokio::test] +async fn a_conversation_written_after_startup_is_listed() { + let (mut ws, first, fs, tmp) = workspace_with_drafts(); + let mut sink: Vec = Vec::new(); + + // The host's view, taken at startup. + ws.load_conversation_index(); + assert_eq!(ws.conversations().count(), 1); + + // Another process writes a second conversation. Same store, its own handle, + // which is what a `jp query` in a terminal amounts to. + let second = ConversationId::try_from( + chrono::DateTime::::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_001), + ) + .unwrap(); + fs.write_test_conversation(&second, &Conversation::default()); + + let response = handle_request( + PluginToHost::ListConversations(jp_plugin::message::OptionalId { id: None }), + &mut sink, + &mut ws, + &json!({}), + None, + None, + &AppConfig::new_test(), + &router(), + ) + .unwrap(); + assert_eq!(response, Flow::Continue); + + let listed: Vec = ws.conversations().map(|(id, _)| id.to_string()).collect(); + + assert!( + listed.contains(&second.to_string()), + "a conversation written after startup must be listed: {listed:?}" + ); + assert!( + listed.contains(&first.to_string()), + "and the one from startup is still there: {listed:?}" + ); + + drop(tmp); +} + /// The canonical spelling is what JP prints, and bare deciseconds still resolve /// because that is what the wire carried before. #[test] From 76a21a17b31394989e974b91f6c5a0a00acdabbf Mon Sep 17 00:00:00 2001 From: Jean Mertz Date: Wed, 16 Sep 2026 19:18:33 +0200 Subject: [PATCH 2/2] review feedback Signed-off-by: Jean Mertz --- .../jp_cli/src/cmd/plugin/dispatch_tests.rs | 113 +++++++++++++++--- 1 file changed, 99 insertions(+), 14 deletions(-) diff --git a/crates/jp_cli/src/cmd/plugin/dispatch_tests.rs b/crates/jp_cli/src/cmd/plugin/dispatch_tests.rs index 26cef116c..4d3fbbf11 100644 --- a/crates/jp_cli/src/cmd/plugin/dispatch_tests.rs +++ b/crates/jp_cli/src/cmd/plugin/dispatch_tests.rs @@ -1,6 +1,8 @@ use camino_tempfile::{Utf8TempDir, tempdir}; use jp_conversation::{Conversation, ConversationId}; -use jp_plugin::message::{ExitMessage, InterruptRequest, ReadyMessage}; +use jp_plugin::message::{ + ExitMessage, InterruptRequest, OptionalId, ReadEventsRequest, ReadyMessage, +}; use jp_storage::backend::{FsStorageBackend, PersistBackend as _}; use relative_path::RelativePathBuf; use serde_json::json; @@ -561,6 +563,15 @@ fn an_unknown_conversation_fails_against_its_request() { } } +/// The messages the host wrote back, in the order the plugin receives them. +fn replies(sink: &[u8]) -> Vec { + String::from_utf8(sink.to_vec()) + .expect("the host writes utf-8") + .lines() + .map(|line| serde_json::from_str(line).expect("a host message")) + .collect() +} + /// A long-running host sees what another process wrote after it started. /// /// The host loads the index once at startup. @@ -569,7 +580,7 @@ fn an_unknown_conversation_fails_against_its_request() { /// terminal never appears. #[tokio::test] async fn a_conversation_written_after_startup_is_listed() { - let (mut ws, first, fs, tmp) = workspace_with_drafts(); + let (mut ws, _first, fs, tmp) = workspace_with_drafts(); let mut sink: Vec = Vec::new(); // The host's view, taken at startup. @@ -578,14 +589,11 @@ async fn a_conversation_written_after_startup_is_listed() { // Another process writes a second conversation. Same store, its own handle, // which is what a `jp query` in a terminal amounts to. - let second = ConversationId::try_from( - chrono::DateTime::::UNIX_EPOCH + std::time::Duration::from_secs(1_700_000_001), - ) - .unwrap(); + let second = conversation_id(1_700_000_001); fs.write_test_conversation(&second, &Conversation::default()); let response = handle_request( - PluginToHost::ListConversations(jp_plugin::message::OptionalId { id: None }), + PluginToHost::ListConversations(OptionalId { id: None }), &mut sink, &mut ws, &json!({}), @@ -597,17 +605,94 @@ async fn a_conversation_written_after_startup_is_listed() { .unwrap(); assert_eq!(response, Flow::Continue); - let listed: Vec = ws.conversations().map(|(id, _)| id.to_string()).collect(); + // Asserted against what reached the plugin rather than the workspace: a + // refresh running after the response is serialized leaves the workspace + // right and the plugin holding the stale list. + let sent = replies(&sink); + let [HostToPlugin::Conversations(listed)] = sent.as_slice() else { + panic!("expected one conversations response, got {sent:?}"); + }; + + // Sorted because the index is a map, and the order it iterates in is not + // what this test is about. + let mut ids: Vec<&str> = listed.data.iter().map(|c| c.id.as_str()).collect(); + ids.sort_unstable(); + assert_eq!(ids, ["jp-c17000000000", "jp-c17000000010"]); + + drop(tmp); +} + +/// A conversation the host has already read is read again, not served from the +/// copy it kept. +/// +/// The first read loads the stream and caches it. +/// A `jp query` in a terminal appends to the same conversation, and without +/// dropping that cache the plugin is served the events as they stood before +/// that turn ran, for the life of the process. +#[tokio::test] +async fn an_event_written_after_a_read_is_served_by_the_next_read() { + let (mut ws, id, fs, tmp) = workspace_with_drafts(); + let mut sink: Vec = Vec::new(); + + let request = || { + PluginToHost::ReadEvents(ReadEventsRequest { + id: None, + conversation: wire_id(id), + }) + }; + + // The read that populates the host's stream cache. + handle_request( + request(), + &mut sink, + &mut ws, + &json!({}), + None, + None, + &AppConfig::new_test(), + &router(), + ) + .unwrap(); + + // Another process runs a turn on that same conversation. + let stream = ConversationStream::new_test().with_turn("what the other terminal asked"); + fs.write( + &id, + &Conversation::default(), + &stream, + Projection::Projected, + ) + .unwrap(); + + handle_request( + request(), + &mut sink, + &mut ws, + &json!({}), + None, + None, + &AppConfig::new_test(), + &router(), + ) + .unwrap(); + + let sent = replies(&sink); + let [HostToPlugin::Events(before), HostToPlugin::Events(after)] = sent.as_slice() else { + panic!("expected two events responses, got {sent:?}"); + }; assert!( - listed.contains(&second.to_string()), - "a conversation written after startup must be listed: {listed:?}" - ); - assert!( - listed.contains(&first.to_string()), - "and the one from startup is still there: {listed:?}" + before.data.is_empty(), + "the conversation had no events when it was first read: {before:?}" ); + let [turn_start, chat_request] = after.data.as_slice() else { + panic!("expected the turn the other process wrote, got {after:?}"); + }; + assert_eq!(turn_start["type"], "turn_start"); + assert_eq!(chat_request["type"], "chat_request"); + assert_eq!(chat_request["content"], "what the other terminal asked"); + drop(tmp); }