Merge pull request #1488 from ScriptedAlchemy/agent/sweep-host-crates

fix(hosts): release advisory writer locks explicitly

Author: ScriptedAlchemy

crates/tracedecay-agent-hosts/src/agents/host_bundle/writer.rs

@@ -54,6 +54,14 @@ pub struct HostBundleWriterV1 {
     _writer_lock: fs::File,
 }
 
+impl Drop for HostBundleWriterV1 {
+    fn drop(&mut self) {
+        if let Err(error) = self._writer_lock.unlock() {
+            tracing::warn!(error = %error, "host bundle writer lock could not be released");
+        }
+    }
+}
+
 impl HostBundleWriterV1 {
     pub fn open(root_path: impl Into<PathBuf>) -> Result<Self, HostBundleError> {
         let root_path = root_path.into();

crates/tracedecay-code-index-runtime/src/code_index_scheduler/publication_store.rs

@@ -478,7 +478,6 @@ pub struct DaemonCodeIndexPublicationStoreV1 {
     /// Test-only: observes each batched `segments_root` directory fsync so a
     /// test can assert a single publish syncs the directory once regardless
     /// of how many new file segments it wrote.
-    /// CI retrigger marker (no behavior).
     #[cfg(test)]
     segments_dir_sync_observer: Option<Arc<dyn Fn() + Send + Sync>>,
     /// Last generation handed to `publish_atomically`. A transient store
@@ -2633,6 +2632,20 @@ impl CodeChunkProjectionSink for DaemonProjectionSinkV1 {
                     output_digest: None,
                 }),
         );
+        decisions.extend(
+            request
+                .changes
+                .reused
+                .iter()
+                .map(|change| ChunkProjectionDecisionV1 {
+                    chunk_id: change.chunk_id.clone(),
+                    prior_chunk_digest: change.prior_digest.clone(),
+                    current_chunk_digest: change.current_digest.clone(),
+                    operation: ProjectionOperationV1::Reused,
+                    outcome: ProjectionOutcomeV1::Reused,
+                    output_digest: None,
+                }),
+        );
         decisions.sort_by(|left, right| left.chunk_id.cmp(&right.chunk_id));
         receipt_builder
             .build(&decisions)

crates/tracedecay-code-index-runtime/src/code_index_scheduler/query_runtime.rs

@@ -1039,5 +1039,3 @@ mod tests {
             .expect("fallback policy is accepted by the fallback authority mode");
     }
 }
-
-// CI retrigger marker (no behavior).

crates/tracedecay-code-index/src/production/lexical_page_source.rs

@@ -54,7 +54,6 @@ pub(super) const LEXICAL_FILE_PREFETCH_BYTES_V1: u64 = 64 * 1024 * 1024;
 /// byte budget above (`LEXICAL_FILE_PREFETCH_BYTES_V1`) be the binding
 /// constraint far more often, without changing how many bytes are held in
 /// flight at once (the byte cap still applies on top of this file cap).
-/// CI retrigger marker (no behavior).
 pub(super) const LEXICAL_DECODE_WINDOW_FILES_PER_WORKER_V1: usize = 4;
 
 type PersistedSealedLexicalCursorFields = (

crates/tracedecay-code-index/tests/code_index_suite/chunk_incremental.rs

@@ -353,3 +353,37 @@ fn duplicate_file_occurrences_are_rejected_before_manifest_flattening() {
         )))
     );
 }
+
+#[test]
+fn standalone_pool_failure_is_parallelism_not_identity_validation() {
+    struct ClearForce;
+    impl Drop for ClearForce {
+        fn drop(&mut self) {
+            tracedecay_code_index::parallelism::force_install_failure_for_test(false);
+        }
+    }
+    let _clear = ClearForce;
+    tracedecay_code_index::parallelism::force_install_failure_for_test(true);
+
+    let expected_generation = generation(2);
+    let file = baseline_file(&expected_generation, "file.ok", "src/lib.rs");
+    let err = GenerationChunkManifestV1::new(expected_generation, vec![file]).unwrap_err();
+
+    match err {
+        ChunkIncrementErrorV1::Parallelism(
+            tracedecay_code_index::parallelism::CodeIndexParallelismErrorV1::PoolBuild { message },
+        ) => {
+            assert!(
+                message.contains("forced"),
+                "expected forced pool failure, got {message}"
+            );
+        }
+        ChunkIncrementErrorV1::NonCanonical(cause) => {
+            panic!(
+                "operational pool failure must not be NonCanonical ({})",
+                cause.reason_code()
+            );
+        }
+        other => panic!("unexpected increment error: {other}"),
+    }
+}

crates/tracedecay-hooks/src/delivery_spool.rs

@@ -120,6 +120,14 @@ pub struct HookDeliveryReceiptSpoolV1 {
     _lock: File,
 }
 
+impl Drop for HookDeliveryReceiptSpoolV1 {
+    fn drop(&mut self) {
+        if let Err(error) = self._lock.unlock() {
+            tracing::warn!(error = %error, "hook delivery receipt spool lock could not be released");
+        }
+    }
+}
+
 impl HookDeliveryReceiptSpoolV1 {
     /// Opens the spool without waiting for a held writer lock. Native callbacks
     /// use `open_within` with one synchronous budget for the lock wait.

crates/tracedecay-hooks/src/spool/mod.rs

@@ -187,7 +187,7 @@ impl HookSpoolV1 {
         hotpath::measure_block!("hooks.spool.fsync.directory", {
             shared_sync_directory(&root, DIRECTORY_POLICY).map_err(|_| HookSpoolError::Io)
         })?;
-        drop(lease_file);
+        lease_file.unlock().map_err(|_| HookSpoolError::Io)?;
         Ok(())
     }
 

crates/tracedecay-query/src/retrieval/lexical/projection/artifact/builder.rs

@@ -24,7 +24,8 @@ use tracedecay_code_index::production::{
     VerifiedSealedLexicalSourceReceiptV1, VerifiedSealedLexicalSymbolDisplayV1,
 };
 use tracedecay_domain::{
-    CodeSearchChunkAnchorV1, CodeSearchChunkV1, ExactTechnicalTermV1, ManifestDigest,
+    CodeSearchChunkAnchorV1, CodeSearchChunkV1, ExactTechnicalTermV1, FileOccurrenceId,
+    ManifestDigest,
 };
 use tracedecay_private_fs::framed_log::{DirectorySyncPolicy, sync_parent_directory};
 use tracedecay_private_fs::{create_private_file_retained, open_private_file};
@@ -2455,8 +2456,7 @@ fn validated_fixed_ledger_charge(
     let serialized_bytes = metadata_serialized_upper_bound(metadata);
     let fixed = ARTIFACT_SQLITE_CACHE_BYTES
         .checked_add(
-            metadata
-                .retained_owned_bytes()
+            metadata_retained_bytes(metadata)
                 .checked_mul(2)
                 .ok_or_else(|| {
                     CodeLexicalArtifactErrorV1::Contract(
@@ -2490,13 +2490,50 @@ fn metadata_serialized_upper_bound(metadata: &CodeLexicalProjectionMetadataV1) -
                 .saturating_add(path.len())
                 .saturating_add(32)
         });
-    metadata
-        .retained_owned_bytes()
+    metadata_retained_bytes(metadata)
         .saturating_add(path_bytes)
         .saturating_mul(6)
         .saturating_add(512)
 }
 
+/// Owned bytes one projection metadata structure retains: logical paths at
+/// capacity with per-entry b-tree node overhead, and every scalar identity
+/// string charged as its `String` header plus payload length.
+fn metadata_retained_bytes(metadata: &CodeLexicalProjectionMetadataV1) -> usize {
+    let path_bytes = metadata.logical_paths.iter().fold(
+        metadata.logical_paths.len().saturating_mul(
+            std::mem::size_of::<(FileOccurrenceId, String)>()
+                .saturating_add(BTREE_MAP_ENTRY_OVERHEAD_BYTES),
+        ),
+        |bytes, (file, path)| {
+            bytes
+                .saturating_add(file.as_str().len())
+                .saturating_add(path.capacity())
+        },
+    );
+    let scalar_identities = [
+        Some(metadata.generation.as_str()),
+        metadata
+            .repository_id
+            .as_ref()
+            .map(|repository| repository.as_str()),
+        Some(metadata.freshness.source_namespace.as_str()),
+        Some(metadata.freshness.source_instance.as_str()),
+        Some(metadata.freshness.policy_revision.as_str()),
+        Some(metadata.exact_retriever_revision.as_str()),
+        Some(metadata.lexical_retriever_revision.as_str()),
+        Some(metadata.exact_score_domain.as_str()),
+    ];
+    scalar_identities
+        .into_iter()
+        .flatten()
+        .fold(path_bytes, |bytes, identity| {
+            bytes
+                .saturating_add(std::mem::size_of::<String>())
+                .saturating_add(identity.len())
+        })
+}
+
 fn page_batch_ledger_charge_bytes(
     metadata: &CodeLexicalProjectionMetadataV1,
     pages: &[VerifiedSealedLexicalPageV1],
@@ -3802,7 +3839,6 @@ fn verify_clone_payload_digests<'body>(
     Ok(())
 }
 
-/// CI retrigger marker (no behavior).
 fn append_prepared_clone_bodies(
     transaction: &Transaction<'_>,
     pages: &[PreparedCodeLexicalArtifactPageV1],