fix(sdk): separate delivered stream cursors from watermarks

crates/tracedecay-agent-hosts/src/agents/codex/tests.rs

@@ -1,6 +1,9 @@
 use super::*;
 use crate::agents::host_bundle::HostComponentV1;
-use crate::agents::safe_write_json_file;
+use crate::agents::{
+    observed_bundle_discovery_matches, safe_write_json_file, unexpected_bundle_entrypoints,
+};
+use sha2::{Digest, Sha256};
 #[cfg(unix)]
 use tracedecay_runtime_core::test_executable::write_executable_script;
 
@@ -742,12 +745,13 @@ fn install_ctx(home: &Path) -> InstallContext {
 /// Core install leaves them.
 fn install_codex_personal_bootstrap(home: &Path, tracedecay_bin: &str) -> Result<PathBuf> {
     let install_dir = codex_plugin_install_dir(home);
+    let profile = ProfileRoot::under_home(home);
     install_codex_plugin_bundle(
         &install_dir,
         tracedecay_bin,
         &ServeArgs::default(),
         InstallScope::Global,
-        home,
+        profile.data_dir(),
     )?;
     install_codex_marketplace_entry(
         &codex_personal_marketplace_path(home),
@@ -782,6 +786,97 @@ fn write_exact_native_activation(home: &Path, tracedecay_bin: &str) {
     copy_rendered_bundle_to_native_cache(home, tracedecay_bin);
 }
 
+// Only evaluated when an activation assertion fails: identify the failing
+// production stage and bounded byte differences without printing bundle contents.
+fn native_activation_diagnostics(
+    profile: &ProfileRoot,
+    home: &Path,
+    tracedecay_bin: &str,
+) -> String {
+    const MAX_DETAILS: usize = 8;
+    let serve_args = ServeArgs::default();
+    let mut details = vec![format!(
+        "marketplace={:?}; manifest={:?}; enabled={:?}; cache_matches={:?}",
+        codex_exact_personal_marketplace_name(home),
+        codex_source_manifest_matches_catalog_version(home),
+        codex_plugin_enabled(home),
+        codex_loaded_cache_matches_rendered_bundle(
+            profile,
+            home,
+            Some((tracedecay_bin, &serve_args))
+        ),
+    )];
+    let rendered = match rendered_global_plugin_files(tracedecay_bin, &serve_args) {
+        Ok(rendered) => rendered,
+        Err(error) => return format!("{}; render error={error}", details.join("\n")),
+    };
+    let source = codex_plugin_install_dir(home);
+    let cache = codex_plugin_current_cached_install_dir(home);
+    let fingerprint = |bytes: &[u8]| {
+        format!(
+            "len={} sha256={}",
+            bytes.len(),
+            hex::encode(Sha256::digest(bytes))
+        )
+    };
+    let mut differences = Vec::new();
+    let mut difference_count = 0;
+    for (relative, expected) in &rendered {
+        for (label, root) in [("source", &source), ("cache", &cache)] {
+            let observed = std::fs::read(root.join(relative));
+            if observed
+                .as_ref()
+                .is_ok_and(|bytes| bytes.as_slice() == expected.as_bytes())
+            {
+                continue;
+            }
+            difference_count += 1;
+            if differences.len() < MAX_DETAILS {
+                let observed = match observed {
+                    Ok(bytes) => fingerprint(&bytes),
+                    Err(error) => format!("error={:?}", error.kind()),
+                };
+                differences.push(format!(
+                    "{label}/{relative}: actual({observed}), expected({})",
+                    fingerprint(expected.as_bytes()),
+                ));
+            }
+        }
+    }
+    details.push(format!(
+        "differing_files={difference_count}; first {MAX_DETAILS}={differences:?}"
+    ));
+    let relatives = rendered
+        .iter()
+        .map(|(relative, _)| (*relative).to_string())
+        .collect();
+    match codex_expected_discovery_relatives(profile.data_dir(), home, relatives) {
+        Ok(expected) => {
+            details.push(format!(
+                "discovery_matches={:?}",
+                observed_bundle_discovery_matches(
+                    &source,
+                    &cache,
+                    &expected,
+                    CODEX_DISCOVERY_ROOTS
+                ),
+            ));
+            for (label, root) in [("source", &source), ("cache", &cache)] {
+                match unexpected_bundle_entrypoints(root, &expected, CODEX_DISCOVERY_ROOTS) {
+                    Ok(paths) => details.push(format!(
+                        "{label} unexpected_entrypoints={}; first {MAX_DETAILS}={:?}",
+                        paths.len(),
+                        &paths[..paths.len().min(MAX_DETAILS)],
+                    )),
+                    Err(error) => details.push(format!("{label} discovery error={error}")),
+                }
+            }
+        }
+        Err(()) => details.push("expected discovery error".to_string()),
+    }
+    details.join("\n")
+}
+
 #[test]
 fn native_activation_binds_enabled_key_to_exact_marketplace_and_cache() {
     let home = tempfile::tempdir().unwrap();
@@ -884,7 +979,9 @@ fn native_cache_content_drift_and_binary_relocation_require_refresh() {
             home.path(),
             Some((old_bin, &ServeArgs::default()))
         )
-        .unwrap()
+        .unwrap(),
+        "initial native activation: {}",
+        native_activation_diagnostics(profile, home.path(), old_bin),
     );
 
     let retired_skill =
@@ -897,7 +994,9 @@ fn native_cache_content_drift_and_binary_relocation_require_refresh() {
             home.path(),
             Some((old_bin, &ServeArgs::default()))
         )
-        .unwrap()
+        .unwrap(),
+        "retired cache entry must deactivate: {}",
+        native_activation_diagnostics(profile, home.path(), old_bin),
     );
     std::fs::remove_file(retired_skill).unwrap();
     assert!(
@@ -906,7 +1005,9 @@ fn native_cache_content_drift_and_binary_relocation_require_refresh() {
             home.path(),
             Some((old_bin, &ServeArgs::default()))
         )
-        .unwrap()
+        .unwrap(),
+        "removing retired entry must reactivate: {}",
+        native_activation_diagnostics(profile, home.path(), old_bin),
     );
 
     std::fs::write(
@@ -920,7 +1021,9 @@ fn native_cache_content_drift_and_binary_relocation_require_refresh() {
             home.path(),
             Some((old_bin, &ServeArgs::default()))
         )
-        .unwrap()
+        .unwrap(),
+        "cache byte drift must deactivate: {}",
+        native_activation_diagnostics(profile, home.path(), old_bin),
     );
     copy_rendered_bundle_to_native_cache(home.path(), old_bin);
     assert!(
@@ -929,7 +1032,9 @@ fn native_cache_content_drift_and_binary_relocation_require_refresh() {
             home.path(),
             Some((old_bin, &ServeArgs::default()))
         )
-        .unwrap()
+        .unwrap(),
+        "refreshing original bundle must reactivate: {}",
+        native_activation_diagnostics(profile, home.path(), old_bin),
     );
 
     install_codex_personal_bootstrap(home.path(), new_bin).unwrap();
@@ -939,7 +1044,9 @@ fn native_cache_content_drift_and_binary_relocation_require_refresh() {
             home.path(),
             Some((new_bin, &ServeArgs::default()))
         )
-        .unwrap()
+        .unwrap(),
+        "binary relocation must require refresh: {}",
+        native_activation_diagnostics(profile, home.path(), new_bin),
     );
     copy_rendered_bundle_to_native_cache(home.path(), new_bin);
     assert!(
@@ -948,7 +1055,9 @@ fn native_cache_content_drift_and_binary_relocation_require_refresh() {
             home.path(),
             Some((new_bin, &ServeArgs::default()))
         )
-        .unwrap()
+        .unwrap(),
+        "refreshing relocated bundle must reactivate: {}",
+        native_activation_diagnostics(profile, home.path(), new_bin),
     );
 }
 

crates/tracedecay-lsp/src/protocol/diagnostic_snapshots.rs

@@ -412,9 +412,21 @@ where
     }
 
     pub(super) fn clear_document_diagnostics(&mut self) {
-        let documents: Vec<_> = self.diagnostics.active_refreshes.keys().cloned().collect();
+        let documents: BTreeSet<_> = self
+            .diagnostics
+            .active_refreshes
+            .keys()
+            .chain(self.outbound.queue.iter().filter_map(|frame| {
+                frame
+                    .publication
+                    .as_ref()
+                    .map(|publication| &publication.uri)
+            }))
+            .cloned()
+            .collect();
         for uri in documents {
             self.retire_diagnostic_refresh(&uri);
+            self.discard_queued_document_publications(&uri);
         }
         self.diagnostics.document_snapshots.clear();
     }

crates/tracedecay-lsp/src/protocol/tests/diagnostic_snapshots.rs

@@ -853,6 +853,74 @@ fn dependency_invalidation_discards_queued_stale_publications() {
     assert_eq!(harness.pull(None)["error"]["code"], -32802);
 }
 
+#[test]
+fn workspace_mutation_discards_stale_publications_and_preserves_in_flight_delivery() {
+    for in_flight in [false, true] {
+        let mut harness = Harness::new(true);
+        harness.open(1);
+        harness.start_refresh();
+        harness.runtime.complete_next();
+        harness.session.flush_due(harness.now_ms);
+        let pending = if in_flight {
+            Some(harness.session.poll_outbound().unwrap().to_vec())
+        } else {
+            None
+        };
+        harness
+            .session
+            .publish_diagnostics(DOCUMENT, 1, 2, Vec::new())
+            .unwrap();
+        let response = json!({ "jsonrpc": "2.0", "id": 900, "result": null });
+        assert!(harness.session.enqueue_value(response.clone()));
+        harness.session.handle_payload(
+            &serde_json::to_vec(&json!({
+                "jsonrpc": "2.0", "method": "workspace/didChangeWorkspaceFolders",
+                "params": { "event": {
+                    "added": [{ "uri": "file:///other", "name": "other" }],
+                    "removed": [],
+                } },
+            }))
+            .unwrap(),
+            harness.now_ms + 1,
+        );
+        let mutation = harness.session.pending_workspace_folder_mutation().unwrap();
+        harness
+            .session
+            .apply_workspace_folder_mutation(
+                &mutation,
+                AuthorizedLspWorkspace::new(
+                    Some(digest('d')),
+                    vec![
+                        AdmittedRoot::authorized("file:///root", digest('b')),
+                        AdmittedRoot::authorized("file:///other", digest('e')),
+                    ],
+                )
+                .unwrap(),
+            )
+            .unwrap();
+        if let Some(pending) = pending {
+            assert_eq!(harness.session.poll_outbound().unwrap(), pending);
+            assert!(harness.session.acknowledge_outbound());
+        }
+        assert_eq!(harness.messages(), vec![response], "in_flight={in_flight}");
+        harness
+            .session
+            .publish_diagnostics(DOCUMENT, 1, 2, Vec::new())
+            .expect("the discarded report was never delivered");
+        assert!(
+            harness
+                .session
+                .publish_diagnostics(DOCUMENT, 1, 2, Vec::new())
+                .is_err(),
+            "an identical queued publication must still be deduplicated"
+        );
+        let delivered = harness.messages();
+        assert_eq!(delivered.len(), 1);
+        assert_eq!(delivered[0]["method"], "textDocument/publishDiagnostics");
+        assert_eq!(delivered[0]["params"]["uri"], DOCUMENT);
+    }
+}
+
 #[test]
 fn late_completion_cannot_answer_a_reopened_document_lifetime() {
     let mut harness = Harness::new(false);

crates/tracedecay-sdk/README.md

@@ -48,3 +48,16 @@ its request/result schemas, route, effect, idempotency, cancellation, and
 deadline contract. `operations::UNAVAILABLE_OPERATIONS` is generated from the
 same registry for cataloged operations that still lack a canonical schema or
 SDK transport; those entries do not acquire callable methods.
+
+## Resuming an operation stream
+
+An `open` event's `frontier.next_sequence` reports how far the server has
+published. It is not the subscriber's delivered position. The client refreshes
+the frontier's resume token while holding its delivered cursor until an event
+or explicit gap is consumed. A gap advances to `last_missing_sequence + 1` and
+reports the missing range instead of silently skipping it.
+
+For manual resume, set `StreamResume.next_sequence` to one past the last event
+or gap your application consumed, or zero if it consumed only `open`. Resuming
+exactly one past completion delivers the retained terminal receipt once and
+ends. Automatic reconnect remains opt-in through `StreamOptions.max_reconnects`.