fix: bind Plugin receipts to durable Host Tool-call history identity

crates/workspace-server/src/server_plugin_controller_tests.rs

@@ -13,13 +13,14 @@
 // Run: cargo test -p yoi-workspace-server --lib plugin_controller_tests
 // Prerequisites: openssl and the normal connected-restore test environment.
 mod plugin_controller_tests {
-    use super::plugin_host_tests::plugin_runtime_tests::{Model, assert_https_history};
+    use super::plugin_host_tests::plugin_runtime_tests::{
+        Model, assert_https_history, persisted_call_consumers,
+    };
     use super::plugin_tests::request;
     use super::*;
     use crate::plugin_https::test_support::{HOST, Peer, response};
     use crate::plugin_plan::PlanPinState;
     use session_store::Store;
-    use sha2::Digest;
     use std::sync::{
         Mutex,
         atomic::{AtomicUsize, Ordering},
@@ -41,6 +42,10 @@ mod plugin_controller_tests {
 
     impl Fixture {
         async fn new(peer: &Peer) -> Self {
+            Self::with_request_count(peer, 1).await
+        }
+
+        async fn with_request_count(peer: &Peer, request_count: usize) -> Self {
             // Use the established short-path/environment guard for real restore
             // (not a mock controller or a replacement execution backend).
             let directory = super::restored_parent_control_tests::connected_restore_tempdir();
@@ -57,7 +62,7 @@ mod plugin_controller_tests {
             let clients = model_clients.clone();
             let observed = requests.clone();
             let tool = model_tool.clone();
-            let transport = Arc::new(peer.transport());
+            let transport = Arc::new(peer.transport_for_requests(request_count));
             let api = WorkspaceApi::new_with_embedded_configurators(
                 config,
                 store,
@@ -345,6 +350,14 @@ mod plugin_controller_tests {
             &self,
             worker: &RuntimeWorkerRef,
         ) -> (session_store::SessionId, Vec<agen::Item>) {
+            let (session, saved) = self.saved_state(worker);
+            (session, saved.history)
+        }
+
+        fn saved_state(
+            &self,
+            worker: &RuntimeWorkerRef,
+        ) -> (session_store::SessionId, session_store::RestoredState) {
             let reopened =
                 session_store::WorkerSessionStore::new(self.aggregate(worker).join("session"))
                     .unwrap();
@@ -352,7 +365,7 @@ mod plugin_controller_tests {
             let segments = reopened.list_segments(session).unwrap();
             assert_eq!(segments.len(), 1);
             let saved = session_store::restore(&reopened, session, segments[0]).unwrap();
-            (session, saved.history)
+            (session, saved)
         }
 
         fn count(&self, table: &str) -> i64 {
@@ -444,10 +457,14 @@ mod plugin_controller_tests {
     }
 
     async fn submit(connection: &mut WorkspaceWorkerProtocolConnection) {
+        submit_id(connection, "controller-https-submit").await;
+    }
+
+    async fn submit_id(connection: &mut WorkspaceWorkerProtocolConnection, request_id: &str) {
         connection
             .methods
             .send(protocol::Method::Submit {
-                submission_request_id: "controller-https-submit".into(),
+                submission_request_id: request_id.into(),
                 input: vec![protocol::Segment::Text {
                     content: "Check the service status.".into(),
                 }],
@@ -464,6 +481,14 @@ mod plugin_controller_tests {
     async fn finish(
         connection: &mut WorkspaceWorkerProtocolConnection,
         expected: protocol::RunResult,
+    ) {
+        finish_id(connection, expected, "controller-https-submit").await;
+    }
+
+    async fn finish_id(
+        connection: &mut WorkspaceWorkerProtocolConnection,
+        expected: protocol::RunResult,
+        request_id: &str,
     ) {
         tokio::time::timeout(WATCHDOG, async {
             let mut accepted = false;
@@ -474,7 +499,7 @@ mod plugin_controller_tests {
                         submission_request_id,
                         ..
                     } => {
-                        assert_eq!(submission_request_id, "controller-https-submit");
+                        assert_eq!(submission_request_id, request_id);
                         accepted = true;
                     }
                     protocol::Event::SubmissionRejected { message, .. } => {
@@ -489,7 +514,7 @@ mod plugin_controller_tests {
                         if finished
                             && snapshot.state == protocol::WorkerState::Idle
                             && snapshot.last_finished_submission_request_id.as_deref()
-                                == Some("controller-https-submit") =>
+                                == Some(request_id) =>
                     {
                         return;
                     }
@@ -501,6 +526,116 @@ mod plugin_controller_tests {
         .expect("real Controller HTTPS submission did not finish");
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn repeated_provider_call_id_has_distinct_durable_receipts_and_completed_restore_never_reruns()
+     {
+        // Gemini also reuses a response-local id for the same Tool. Deliberately
+        // emit the identical provider id in two explicit user submissions.
+        let peer = Peer::new(
+            HOST,
+            vec![response(503, &[], b"old"), response(200, &[], b"fresh")],
+        )
+        .await;
+        let f = Fixture::with_request_count(&peer, 2).await;
+        f.install(&peer).await;
+        let worker = f.create(PROFILE).await;
+        let public_name = f.model_tool.lock().unwrap().clone();
+        let mut connection = connect_workspace_worker_protocol(&f.api, &worker, None)
+            .await
+            .unwrap();
+        snapshot(&mut connection).await;
+        for request in ["first-explicit-call", "second-explicit-call"] {
+            submit_id(&mut connection, request).await;
+            finish_id(&mut connection, protocol::RunResult::Finished, request).await;
+        }
+        assert_eq!(
+            peer.connection_count(),
+            2,
+            "a new logical call must not replay the first receipt; model requests: {:?}",
+            f.requests.lock().unwrap().iter().map(|r| &r.items).collect::<Vec<_>>()
+        );
+        assert_eq!(f.count("plugin_invocations"), 2);
+        assert_eq!(f.count("plugin_effect_receipts"), 2);
+        let statuses = |history: &[agen::Item]| -> Vec<u16> {
+            history
+                .iter()
+                .filter_map(|item| match item {
+                    agen::Item::ToolResult {
+                        call_id,
+                        is_error: false,
+                        content: Some(content),
+                        ..
+                    } if call_id == "https-call" => Some(
+                        serde_json::from_str::<Value>(content).unwrap()["structured"]["status"]
+                            .as_u64()
+                            .unwrap() as u16,
+                    ),
+                    _ => None,
+                })
+                .collect()
+        };
+        {
+            let requests = f.requests.lock().unwrap();
+            assert_eq!(requests.len(), 4);
+            assert_eq!(statuses(&requests[3].items), [503, 200]);
+        }
+        drop(connection);
+        f.stop(&worker).await;
+        let (session, saved) = f.saved_state(&worker);
+        assert_eq!(statuses(&saved.history), [503, 200]);
+        let calls: Vec<_> = saved
+            .annotated_history
+            .iter()
+            .filter_map(|entry| match &entry.item {
+                session_store::LoggedItem::ToolCall { call_id, name, .. }
+                    if name == &public_name =>
+                {
+                    Some((call_id, &entry.metadata.entry_id))
+                }
+                _ => None,
+            })
+            .collect();
+        assert_eq!(calls.len(), 2);
+        assert_eq!(
+            calls[0].0, calls[1].0,
+            "provider id must actually be reused in the fixture"
+        );
+        assert_ne!(calls[0].1, calls[1].1);
+        let mut expected = persisted_call_consumers(session, &saved, &public_name);
+        expected.sort();
+        let consumers: Vec<String> = f.api.config_store.with_conn(|conn| {
+            let mut statement = conn.prepare("SELECT json_extract(consumer_json,'$.tool_call_id') FROM plugin_invocations WHERE workspace_id=?1 AND worker_id=?2 ORDER BY 1")?;
+            Ok(statement.query_map(rusqlite::params![TEST_WORKSPACE_ID, worker.worker_id], |row| row.get(0))?.collect::<std::result::Result<Vec<_>,_>>()?)
+        }).unwrap();
+        assert_eq!(
+            consumers, expected,
+            "receipts must bind committed Host annotation ids, not response-local ids"
+        );
+        let restored = f.restore(&worker, "completed-reused-id-restore").await;
+        assert_eq!(
+            restored.state,
+            server_api::WorkerRestoreState::Accepted,
+            "{restored:?}"
+        );
+        assert_eq!(f.requests.lock().unwrap().len(), 4);
+        let mut connection = connect_workspace_worker_protocol(&f.api, &worker, None)
+            .await
+            .unwrap();
+        assert!(snapshot(&mut connection).await.tools.contains(&public_name));
+        assert_eq!(peer.connection_count(), 2);
+        assert_eq!(f.count("plugin_invocations"), 2);
+        assert_eq!(f.count("plugin_effect_receipts"), 2);
+        drop(connection);
+        f.stop(&worker).await;
+        let (restored_session, saved) = f.saved_state(&worker);
+        assert_eq!(restored_session, session);
+        let mut restored = persisted_call_consumers(session, &saved, &public_name);
+        restored.sort();
+        assert_eq!(restored, expected);
+        assert_eq!(statuses(&saved.history), [503, 200]);
+        f.close().await;
+    }
+
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
     async fn owner_diagnostics_observes_saved_pin_and_live_refusal_without_execution_or_paths() {
         let peer = Peer::new(HOST, vec![]).await;
@@ -735,14 +870,9 @@ mod plugin_controller_tests {
                 rusqlite::params![TEST_WORKSPACE_ID, saved.plan_id, worker.worker_id],
                 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)))?)
         }).unwrap();
-        let identity = serde_json::to_vec(&(session.to_string(), "https-call")).unwrap();
-        let consumer_id = format!(
-            "tool-{}",
-            sha2::Sha256::digest(identity)
-                .iter()
-                .map(|byte| format!("{byte:02x}"))
-                .collect::<String>()
-        );
+        let consumers = persisted_call_consumers(session, &f.saved_state(&worker).1, &public_name);
+        assert_eq!(consumers.len(), 1);
+        let consumer_id = consumers[0].clone();
         assert_eq!(
             serde_json::from_str::<server_api::PluginConsumer>(&consumer).unwrap(),
             server_api::PluginConsumer::ToolCall {

crates/workspace-server/src/server_plugin_remote_controller_tests.rs

@@ -7,7 +7,9 @@
 // Run: cargo test -p yoi-workspace-server --lib plugin_remote_controller_tests
 // Prerequisites: openssl and the existing connected Runtime test environment.
 mod plugin_remote_controller_tests {
-    use super::plugin_host_tests::plugin_runtime_tests::{Model, assert_https_history};
+    use super::plugin_host_tests::plugin_runtime_tests::{
+        Model, assert_https_history, persisted_call_consumers,
+    };
     use super::plugin_tests::request;
     use super::subject_authorization_tests::{CONNECTED_RUNTIME, ConnectedWorkerHarness};
     use super::*;
@@ -15,7 +17,6 @@ mod plugin_remote_controller_tests {
     use crate::plugin_plan::PlanPinState;
     use agen::llm_client::{ClientError, LlmClient, Request as ModelRequest};
     use session_store::Store;
-    use sha2::Digest;
     use std::sync::Mutex;
     use std::time::Duration;
 
@@ -380,14 +381,9 @@ mod plugin_remote_controller_tests {
                 rusqlite::params![TEST_WORKSPACE_ID, saved.plan_id, worker.worker_id],
                 |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)))?)
         }).unwrap();
-        let identity = serde_json::to_vec(&(session.to_string(), "https-call")).unwrap();
-        let consumer_id = format!(
-            "tool-{}",
-            sha2::Sha256::digest(identity)
-                .iter()
-                .map(|byte| format!("{byte:02x}"))
-                .collect::<String>()
-        );
+        let consumers = persisted_call_consumers(session, &history, &public_name);
+        assert_eq!(consumers.len(), 1);
+        let consumer_id = consumers[0].clone();
         assert_eq!(
             serde_json::from_str::<server_api::PluginConsumer>(&consumer).unwrap(),
             server_api::PluginConsumer::ToolCall {

crates/workspace-server/src/server_plugin_runtime_tests.rs

@@ -147,7 +147,7 @@ pub(super) mod plugin_runtime_tests {
                 turn
             };
             let events = match turn {
-                0 => vec![
+                0 | 2 => vec![
                     ModelEvent::tool_use_start(0, "https-call", &self.tool_name),
                     ModelEvent::tool_input_delta(
                         0,
@@ -159,7 +159,7 @@ pub(super) mod plugin_runtime_tests {
                     ),
                     ModelEvent::tool_use_stop(0),
                 ],
-                1 => vec![
+                1 | 3 => vec![
                     ModelEvent::text_block_start(0),
                     ModelEvent::text_delta(0, "Observed the HTTPS tool result."),
                     ModelEvent::text_block_stop(0, None),
@@ -177,6 +177,32 @@ pub(super) mod plugin_runtime_tests {
         }
     }
 
+    pub(in crate::server::tests) fn persisted_call_consumers(
+        session: session_store::SessionId,
+        saved: &session_store::RestoredState,
+        public_name: &str,
+    ) -> Vec<String> {
+        saved
+            .annotated_history
+            .iter()
+            .filter_map(|entry| match &entry.item {
+                session_store::LoggedItem::ToolCall { name, .. } if name == public_name => {
+                    let bytes =
+                        serde_json::to_vec(&(session.to_string(), &entry.metadata.entry_id.0))
+                            .unwrap();
+                    Some(format!(
+                        "tool-{}",
+                        Sha256::digest(bytes)
+                            .iter()
+                            .map(|byte| format!("{byte:02x}"))
+                            .collect::<String>()
+                    ))
+                }
+                _ => None,
+            })
+            .collect()
+    }
+
     pub(in crate::server::tests) fn assert_https_history(items: &[agen::Item], public_name: &str) {
         let calls = items
             .iter()
@@ -234,7 +260,19 @@ pub(super) mod plugin_runtime_tests {
         assert_eq!(output["structured"], json!({"status":503,"body_bytes":9}));
     }
 
-    async fn https_through_runtime_and_history(remote: bool) {
+    struct PauseBeforeFirstCall;
+
+    #[async_trait::async_trait]
+    impl worker::hook::Hook<worker::hook::PreToolCall> for PauseBeforeFirstCall {
+        async fn call(
+            &self,
+            _: &worker::hook::ToolCallSummary,
+        ) -> std::result::Result<worker::hook::HookPreToolAction, worker::hook::HookError> {
+            Ok(worker::hook::HookPreToolAction::Pause)
+        }
+    }
+
+    async fn https_through_runtime_and_history(remote: bool, restore_pending: bool) {
         let mut peer = Peer::new(HOST, vec![response(503, &[], b"peer data")]).await;
         let mut f = Fixture::new(Some(peer_binding(&peer))).await;
         f.api.plugin_https_transport = Some(Arc::new(peer.transport()));
@@ -263,17 +301,34 @@ pub(super) mod plugin_runtime_tests {
             tool_name: public_name.clone(),
             requests: Arc::default(),
         };
-        let manifest = worker::WorkerManifest::from_toml(&format!(
+        let mut manifest = worker::WorkerManifest::from_toml(&format!(
             "[worker]\nname=\"{}\"\n[model]\nscheme=\"anthropic\"\nmodel_id=\"scripted\"\n[engine]\n[scope]\nallow=[]\n",
             f.source.worker_id
         ))
         .unwrap();
+        if restore_pending {
+            manifest
+                .plugins
+                .selections
+                .push(manifest::PluginProfileSelection {
+                    installation_id: f.installed.installation.installation_id.clone(),
+                    tools: vec![f.tool.into()],
+                    logging: false,
+                    https_slots: vec!["api".into()],
+                    config: Default::default(),
+                });
+        }
         let store_root = f._directory.path().join("tool-history");
-        let mut host = worker::Worker::new(
-            manifest,
-            agen::Engine::new_annotated(model.clone())
-                .system_prompt("Use the selected HTTPS tool."),
+        let metadata_root = f._directory.path().join("worker-metadata");
+        let store = session_store::CombinedStore::new(
             session_store::FsStore::new(&store_root).unwrap(),
+            session_store::FsWorkerStore::new(&metadata_root).unwrap(),
+        );
+        let mut host = worker::Worker::new(
+            manifest.clone(),
+            agen::Engine::new_annotated(Box::new(model.clone()) as Box<dyn LlmClient>)
+                .system_prompt("Use the selected HTTPS tool."),
+            store,
             worker::WorkerWorkspaceContext::unavailable(
                 Some(worker::WorkspaceId::new(TEST_WORKSPACE_ID).unwrap()),
                 "unrelated Workspace operations are not bound by this fixture",
@@ -330,6 +385,10 @@ pub(super) mod plugin_runtime_tests {
             tools.get_tool(f.tool).is_none(),
             "bare guest names are not public tools"
         );
+        if restore_pending {
+            host.enable_worker_metadata_write_through().unwrap();
+            host.add_pre_tool_call_hook(PauseBeforeFirstCall);
+        }
         host.wire_history_persistence();
         let result = tokio::time::timeout(
             std::time::Duration::from_secs(30),
@@ -340,7 +399,63 @@ pub(super) mod plugin_runtime_tests {
         .await
         .expect("HTTPS Tool/Engine must finish within the fixture deadline")
         .unwrap();
-        assert!(matches!(result, worker::WorkerRunResult::Finished));
+        if restore_pending {
+            assert!(matches!(result, worker::WorkerRunResult::Paused));
+            assert_eq!(peer.connection_count(), 0);
+            assert_eq!(f.count("plugin_invocations"), 0);
+            let reopened = session_store::FsStore::new(&store_root).unwrap();
+            let segments = reopened.list_segments(session_id).unwrap();
+            let pending = session_store::restore(&reopened, session_id, segments[0]).unwrap();
+            let expected = persisted_call_consumers(session_id, &pending, &public_name);
+            assert_eq!(
+                expected.len(),
+                1,
+                "paused call must already have a durable Host identity"
+            );
+            assert!(
+                !pending
+                    .history
+                    .iter()
+                    .any(|item| matches!(item, agen::Item::ToolResult { .. }))
+            );
+            drop(host);
+            let mut restored = worker::Worker::prepare_restored_from_worker_metadata_with_context_and_model_client(
+                &manifest.worker.name, manifest.clone(), session_store::CombinedStore::new(
+                    reopened, session_store::FsWorkerStore::new(&metadata_root).unwrap(),
+                ),
+                worker::PromptCatalogSource::builtins_only(),
+                worker::WorkerWorkspaceContext::unavailable(Some(worker::WorkspaceId::new(TEST_WORKSPACE_ID).unwrap()), "unrelated Workspace operations unavailable"),
+                worker::WorkerFilesystemAuthority::None, Some(Box::new(model.clone())),
+            ).await.unwrap().activate().await.unwrap();
+            assert_eq!(restored.session_id(), session_id);
+            let reloaded = session_store::restore(
+                &session_store::FsStore::new(&store_root).unwrap(),
+                session_id,
+                segments[0],
+            )
+            .unwrap();
+            assert_eq!(
+                persisted_call_consumers(session_id, &reloaded, &public_name),
+                expected
+            );
+            let feature = bridge
+                .prepare(&worker_ref, &snapshot, &session_id.to_string())
+                .await
+                .unwrap();
+            let report =
+                restored.install_features(FeatureRegistryBuilder::new().with_module(feature));
+            assert!(!report.has_errors(), "{}", report.error_message());
+            restored.wire_history_persistence();
+            let result =
+                tokio::time::timeout(std::time::Duration::from_secs(30), restored.resume())
+                    .await
+                    .unwrap()
+                    .unwrap();
+            assert!(matches!(result, worker::WorkerRunResult::Finished));
+            host = restored;
+        } else {
+            assert!(matches!(result, worker::WorkerRunResult::Finished));
+        }
         assert_https_history(&host.history(), &public_name);
         {
             let requests = model.requests.lock().unwrap();
@@ -371,14 +486,9 @@ pub(super) mod plugin_runtime_tests {
         assert_eq!(segments.len(), 1);
         let saved = session_store::restore(&reopened, session_id, segments[0]).unwrap();
         assert_https_history(&saved.history, &public_name);
-        let identity_bytes = serde_json::to_vec(&(session_id.to_string(), "https-call")).unwrap();
-        let consumer_id = format!(
-            "tool-{}",
-            Sha256::digest(identity_bytes)
-                .iter()
-                .map(|byte| format!("{byte:02x}"))
-                .collect::<String>()
-        );
+        let consumers = persisted_call_consumers(session_id, &saved, &public_name);
+        assert_eq!(consumers.len(), 1);
+        let consumer_id = consumers[0].clone();
         let (consumer, state, status, body): (String, String, i64, Vec<u8>) = f
             .api
             .config_store
@@ -418,7 +528,8 @@ pub(super) mod plugin_runtime_tests {
                 .call_tool(
                     &public_name,
                     "{}",
-                    ToolExecutionContext::new("denied-call", "denied-batch", 0),
+                    ToolExecutionContext::new("denied-call", "denied-batch", 0)
+                        .with_logical_call_id("denied-persisted-call"),
                 )
                 .await
                 .is_err()
@@ -444,12 +555,17 @@ pub(super) mod plugin_runtime_tests {
 
     #[tokio::test]
     async fn remote_pinned_https_tool_commits_normal_history_and_receipt_then_fences_live_denial() {
-        https_through_runtime_and_history(true).await;
+        https_through_runtime_and_history(true, false).await;
     }
 
     #[tokio::test]
     async fn embedded_pinned_https_tool_commits_normal_history_and_receipt_then_fences_live_denial()
     {
-        https_through_runtime_and_history(false).await;
+        https_through_runtime_and_history(false, false).await;
+    }
+
+    #[tokio::test]
+    async fn pending_tool_call_restores_its_committed_host_identity_before_first_effect() {
+        https_through_runtime_and_history(false, true).await;
     }
 }