fix: integrate safe Drive and config failure diagnostics

crates/worker/src/feature/builtin.rs

@@ -4,6 +4,7 @@
 //! same descriptor-approved registry path used by feature modules. They are not
 //! an external plugin-loading surface.
 
+mod backend_diagnostic;
 pub mod backend_job_result;
 pub(crate) mod bash;
 

crates/worker/src/feature/builtin/backend_diagnostic.rs

@@ -0,0 +1,195 @@
+//! Internal adapter diagnostics are independent of public failure classifications.
+//!
+//! Only closed vocabularies, numeric status and opaque correlation tokens cross this boundary.
+//! In particular, a transport error's Display/Debug and server-controlled text are never recorded.
+use sha2::{Digest, Sha256};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum Adapter {
+    Drive,
+    WorkspaceConfig,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum Stage {
+    Request,
+    Response,
+    Completion,
+    Reconcile,
+    Attach,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(super) enum Reason {
+    Deadline,
+    TaskJoin,
+    TransportUnavailable,
+    InvalidPath,
+    Communication,
+    Decode,
+    RequestSize,
+    ResponseSize,
+    ResponseLength,
+    Identity,
+    InvalidMetadata,
+    BackendRejected,
+    AccessCeiling,
+    AttachConversion,
+    BackendUnknown,
+    UnrecognizedResponse,
+    NotCommittedSnapshot,
+    IncompleteReceipt,
+    Reconciled,
+}
+
+/// One logical call, including its original-request reconciliation. Request identities are hashed
+/// rather than logged verbatim: callers and responses need not contain safe log text.
+#[derive(Clone)]
+pub(super) struct Diagnostic {
+    adapter: Adapter,
+    operation: &'static str,
+    mutation: bool,
+    correlation: String,
+    dispatch: tracing::Dispatch,
+    stage_override: Option<Stage>,
+}
+
+impl Diagnostic {
+    pub(super) fn new(
+        adapter: Adapter,
+        operation: &'static str,
+        mutation: bool,
+        request_id: Option<&str>,
+    ) -> Self {
+        Self {
+            adapter,
+            operation,
+            mutation,
+            correlation: request_id
+                .map(|id| {
+                    let hash: String = Sha256::digest(id.as_bytes())
+                        .iter()
+                        .map(|b| format!("{b:02x}"))
+                        .collect();
+                    format!("request:{hash}")
+                })
+                .unwrap_or_else(|| uuid::Uuid::now_v7().to_string()),
+            dispatch: tracing::dispatcher::get_default(Clone::clone),
+            stage_override: None,
+        }
+    }
+
+    pub(super) fn at_stage(&self, stage: Stage) -> Self {
+        let mut diagnostic = self.clone();
+        diagnostic.stage_override = Some(stage);
+        diagnostic
+    }
+
+    pub(super) fn record(&self, stage: Stage, reason: Reason, status: Option<u16>) {
+        let stage = self.stage_override.unwrap_or(stage);
+        tracing::dispatcher::with_default(&self.dispatch, || {
+            tracing::warn!(
+                target: "worker::backend_diagnostic",
+                adapter = ?self.adapter,
+                operation = self.operation,
+                mutation = self.mutation,
+                correlation = %self.correlation,
+                stage = ?stage,
+                reason = ?reason,
+                status,
+                "Workspace adapter diagnostic"
+            );
+        });
+    }
+
+    pub(super) fn transport(&self, error: &crate::worker::WorkspaceClientError) {
+        use crate::worker::WorkspaceClientError;
+        self.record(
+            Stage::Request,
+            match error {
+                WorkspaceClientError::Unavailable(_) => Reason::TransportUnavailable,
+                WorkspaceClientError::InvalidPath(_) => Reason::InvalidPath,
+                WorkspaceClientError::Request(_) => Reason::Communication,
+            },
+            None,
+        );
+    }
+}
+
+#[cfg(test)]
+pub(super) mod tests {
+    use super::{Adapter, Diagnostic, Reason, Stage};
+    use std::sync::{Arc, Mutex};
+    use tracing::{Event, Metadata, Subscriber, field::Visit, span};
+
+    #[test]
+    fn unsafe_request_identity_is_correlated_without_logging_its_value() {
+        let capture = Capture::default();
+        tracing::dispatcher::with_default(&capture.dispatch(), || {
+            for id in [
+                "secret /host/path\ncredential",
+                "secret /host/path\ncredential",
+                "different",
+            ] {
+                Diagnostic::new(Adapter::Drive, "mutate", true, Some(id)).record(
+                    Stage::Request,
+                    Reason::Communication,
+                    None,
+                );
+            }
+        });
+        let text = capture.text();
+        for secret in ["secret", "/host/path", "credential", "different"] {
+            assert!(!text.contains(secret));
+        }
+        let correlations: Vec<_> = text
+            .lines()
+            .map(|line| {
+                line.split("correlation=")
+                    .nth(1)
+                    .unwrap()
+                    .split(';')
+                    .next()
+                    .unwrap()
+            })
+            .collect();
+        assert_eq!(correlations.len(), 3);
+        assert_eq!(correlations[0], correlations[1]);
+        assert_ne!(correlations[0], correlations[2]);
+    }
+
+    #[derive(Clone, Default)]
+    pub(crate) struct Capture(pub(crate) Arc<Mutex<Vec<String>>>);
+    impl Capture {
+        pub(crate) fn dispatch(&self) -> tracing::Dispatch {
+            tracing::Dispatch::new(self.clone())
+        }
+        pub(crate) fn text(&self) -> String {
+            self.0.lock().unwrap().join("\n")
+        }
+    }
+    struct Fields(String);
+    impl Visit for Fields {
+        fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
+            use std::fmt::Write;
+            write!(&mut self.0, "{}={value:?};", field.name()).unwrap();
+        }
+    }
+    impl Subscriber for Capture {
+        fn enabled(&self, metadata: &Metadata<'_>) -> bool {
+            metadata.target() == "worker::backend_diagnostic"
+        }
+        fn new_span(&self, _: &span::Attributes<'_>) -> span::Id {
+            span::Id::from_u64(1)
+        }
+        fn record(&self, _: &span::Id, _: &span::Record<'_>) {}
+        fn record_follows_from(&self, _: &span::Id, _: &span::Id) {}
+        fn event(&self, event: &Event<'_>) {
+            let mut fields = Fields(String::new());
+            event.record(&mut fields);
+            self.0.lock().unwrap().push(fields.0);
+        }
+        fn enter(&self, _: &span::Id) {}
+        fn exit(&self, _: &span::Id) {}
+    }
+}

crates/worker/src/feature/builtin/drive/backend.rs

@@ -11,6 +11,8 @@
 //! cases retain an unknown outcome with the original request ID. Never blindly resend with a new ID.
 use std::{sync::Arc, time::Duration};
 
+use super::super::backend_diagnostic::{Adapter, Diagnostic, Reason, Stage};
+
 use crate::worker::{
     WorkspaceBinaryRequest, WorkspaceClient, WorkspaceRequest, WorkspaceRequestMethod,
 };
@@ -23,6 +25,7 @@ const RESPONSE_LIMIT: usize = 2 * 1024 * 1024;
 #[derive(Clone)]
 pub struct DriveBackend {
     client: Arc<dyn WorkspaceClient>,
+    deadline: Duration,
 }
 
 #[derive(Debug, Clone, thiserror::Error)]
@@ -49,7 +52,10 @@ pub enum DriveError {
 
 impl DriveBackend {
     pub fn new(client: Arc<dyn WorkspaceClient>) -> Self {
-        Self { client }
+        Self {
+            client,
+            deadline: DEADLINE,
+        }
     }
     pub fn workspace(&self) -> Result<&str, DriveError> {
         self.client
@@ -61,15 +67,28 @@ impl DriveBackend {
         Ok(format!("/api/w/{}/drive", encode(self.workspace()?)))
     }
     pub(super) fn validate_entry(&self, entry: &DriveEntry) -> Result<(), DriveError> {
-        self.validate_ref(&entry.entry)?;
+        self.validate_entry_at(
+            entry,
+            &Diagnostic::new(Adapter::Drive, "validate_entry", false, None),
+        )
+    }
+    fn validate_entry_at(
+        &self,
+        entry: &DriveEntry,
+        diagnostic: &Diagnostic,
+    ) -> Result<(), DriveError> {
+        self.validate_ref(&entry.entry)
+            .inspect_err(|_| diagnostic.record(Stage::Completion, Reason::Identity, None))?;
         if let Some(parent) = &entry.parent {
-            self.validate_ref(parent)?;
+            self.validate_ref(parent)
+                .inspect_err(|_| diagnostic.record(Stage::Completion, Reason::Identity, None))?;
         }
         if (entry.last_mutation_id.len() > 128
             || entry.last_mutation_id.chars().any(char::is_control)
             || (entry.parent.is_some() && entry.last_mutation_id.is_empty()))
             || entry.size.is_some_and(|n| n > DRIVE_FILE_MAX_BYTES)
         {
+            diagnostic.record(Stage::Completion, Reason::InvalidMetadata, None);
             return Err(DriveError::Unavailable);
         }
         Ok(())
@@ -90,11 +109,13 @@ impl DriveBackend {
         request: WorkspaceRequest,
         mutation: bool,
         request_id: &str,
+        diagnostic: &Diagnostic,
     ) -> Result<T, DriveError> {
         let client = self.client.clone();
+        let deadline = self.deadline;
         let response = tokio::time::timeout(
-            DEADLINE,
-            tokio::task::spawn_blocking(move || client.execute_with_timeout(request, DEADLINE)),
+            deadline,
+            tokio::task::spawn_blocking(move || client.execute_with_timeout(request, deadline)),
         )
         .await;
         let unknown = || {
@@ -105,10 +126,20 @@ impl DriveBackend {
             }
         };
         let response = response
-            .map_err(|_| unknown())?
-            .map_err(|_| unknown())?
-            .map_err(|_| unknown())?;
+            .map_err(|_| {
+                diagnostic.record(Stage::Request, Reason::Deadline, None);
+                unknown()
+            })?
+            .map_err(|_| {
+                diagnostic.record(Stage::Request, Reason::TaskJoin, None);
+                unknown()
+            })?
+            .map_err(|error| {
+                diagnostic.transport(&error);
+                unknown()
+            })?;
         if response.body.len() > RESPONSE_LIMIT {
+            diagnostic.record(Stage::Response, Reason::ResponseSize, None);
             return Err(if mutation {
                 unknown()
             } else {
@@ -121,42 +152,57 @@ impl DriveBackend {
                 response.body.as_bytes(),
                 mutation,
                 request_id,
+                diagnostic,
             ));
         }
-        serde_json::from_str(&response.body).map_err(|_| unknown())
+        serde_json::from_str(&response.body).map_err(|_| {
+            diagnostic.record(Stage::Response, Reason::Decode, None);
+            unknown()
+        })
     }
     async fn get<T: DeserializeOwned + Send + 'static, Q: Serialize>(
         &self,
-        suffix: &str,
+        suffix: &'static str,
         query: &Q,
+        diagnostic: &Diagnostic,
     ) -> Result<T, DriveError> {
         let query = query_string(query)?;
         self.json(
             WorkspaceRequest::get(format!("{}{suffix}{query}", self.base()?)),
             false,
             "",
+            diagnostic,
         )
         .await
     }
     pub async fn capabilities(&self) -> Result<DriveCapabilities, DriveError> {
-        self.get("/capabilities", &serde_json::json!({})).await
+        self.get(
+            "/capabilities",
+            &serde_json::json!({}),
+            &Diagnostic::new(Adapter::Drive, "capabilities", false, None),
+        )
+        .await
     }
     pub async fn root(&self) -> Result<DriveEntry, DriveError> {
+        let diagnostic = Diagnostic::new(Adapter::Drive, "root", false, None);
         let entry: DriveEntry = self
             .json(
                 WorkspaceRequest::get(format!("{}/root", self.base()?)),
                 false,
                 "",
+                &diagnostic,
             )
             .await?;
-        self.validate_entry(&entry)?;
+        self.validate_entry_at(&entry, &diagnostic)?;
         if entry.kind != DriveEntryKind::Folder || entry.parent.is_some() {
+            diagnostic.record(Stage::Completion, Reason::InvalidMetadata, None);
             return Err(DriveError::Unavailable);
         }
         Ok(entry)
     }
     pub async fn metadata(&self, entry: &DriveEntryRef) -> Result<DriveEntry, DriveError> {
         self.validate_ref(entry)?;
+        let diagnostic = Diagnostic::new(Adapter::Drive, "metadata", false, None);
         let result: DriveEntry = self
             .get(
                 "/metadata",
@@ -164,10 +210,12 @@ impl DriveBackend {
                     entry_workspace_id: entry.workspace_id.clone(),
                     id: entry.node_id.clone(),
                 },
+                &diagnostic,
             )
             .await?;
-        self.validate_entry(&result)?;
+        self.validate_entry_at(&result, &diagnostic)?;
         if result.entry != *entry {
+            diagnostic.record(Stage::Completion, Reason::Identity, None);
             return Err(DriveError::Unavailable);
         }
         Ok(result)
@@ -179,6 +227,7 @@ impl DriveBackend {
         after: Option<String>,
     ) -> Result<DriveListResponse, DriveError> {
         self.validate_ref(parent)?;
+        let diagnostic = Diagnostic::new(Adapter::Drive, "list", false, None);
         let result: DriveListResponse = self
             .get(
                 "/list",
@@ -188,6 +237,7 @@ impl DriveBackend {
                     limit,
                     after,
                 },
+                &diagnostic,
             )
             .await?;
         self.validate_page(
@@ -195,33 +245,43 @@ impl DriveBackend {
             limit
                 .unwrap_or(DRIVE_PAGE_MAX_LIMIT)
                 .min(DRIVE_PAGE_MAX_LIMIT),
+            &diagnostic,
         )?;
         if result
             .entries
             .iter()
             .any(|entry| entry.parent.as_ref() != Some(parent))
         {
+            diagnostic.record(Stage::Completion, Reason::Identity, None);
             return Err(DriveError::Unavailable);
         }
         Ok(result)
     }
     pub async fn search(&self, query: DriveSearchQuery) -> Result<DriveListResponse, DriveError> {
-        let result: DriveListResponse = self.get("/search", &query).await?;
+        let diagnostic = Diagnostic::new(Adapter::Drive, "search", false, None);
+        let result: DriveListResponse = self.get("/search", &query, &diagnostic).await?;
         self.validate_page(
             &result,
             query
                 .limit
                 .unwrap_or(DRIVE_SEARCH_MAX_LIMIT)
                 .min(DRIVE_SEARCH_MAX_LIMIT),
+            &diagnostic,
         )?;
         Ok(result)
     }
-    fn validate_page(&self, page: &DriveListResponse, limit: u32) -> Result<(), DriveError> {
+    fn validate_page(
+        &self,
+        page: &DriveListResponse,
+        limit: u32,
+        diagnostic: &Diagnostic,
+    ) -> Result<(), DriveError> {
         if page.entries.len() > limit as usize {
+            diagnostic.record(Stage::Completion, Reason::ResponseSize, None);
             return Err(DriveError::Limit);
         }
         for entry in &page.entries {
-            self.validate_entry(entry)?;
+            self.validate_entry_at(entry, diagnostic)?;
         }
         Ok(())
     }
@@ -231,6 +291,7 @@ impl DriveBackend {
         max_bytes: u32,
     ) -> Result<DriveReadTextResponse, DriveError> {
         self.validate_ref(entry)?;
+        let diagnostic = Diagnostic::new(Adapter::Drive, "read_text", false, None);
         let result: DriveReadTextResponse = self
             .get(
                 "/read-text",
@@ -239,10 +300,16 @@ impl DriveBackend {
                     id: entry.node_id.clone(),
                     max_bytes,
                 },
+                &diagnostic,
             )
             .await?;
-        self.validate_entry(&result.entry)?;
-        if result.entry.entry != *entry || result.text.len() > max_bytes as usize {
+        self.validate_entry_at(&result.entry, &diagnostic)?;
+        if result.entry.entry != *entry {
+            diagnostic.record(Stage::Completion, Reason::Identity, None);
+            return Err(DriveError::Unavailable);
+        }
+        if result.text.len() > max_bytes as usize {
+            diagnostic.record(Stage::Completion, Reason::ResponseSize, None);
             return Err(DriveError::Unavailable);
         }
         Ok(result)
@@ -255,6 +322,7 @@ impl DriveBackend {
         if total as usize > limit {
             return Err(DriveError::Limit);
         }
+        let diagnostic = Diagnostic::new(Adapter::Drive, "read_chunk", false, None);
         let mut bytes = Vec::with_capacity(total as usize);
         // Even an empty file gets a read bound to the observed committed mutation and current authorization.
         loop {
@@ -274,21 +342,42 @@ impl DriveBackend {
                 max_response_bytes: DRIVE_CHUNK_MAX_BYTES as usize,
             };
             let client = self.client.clone();
+            let deadline = self.deadline;
             let response = tokio::time::timeout(
-                DEADLINE,
+                deadline,
                 tokio::task::spawn_blocking(move || {
-                    client.execute_binary_with_timeout(request, DEADLINE)
+                    client.execute_binary_with_timeout(request, deadline)
                 }),
             )
             .await
-            .map_err(|_| DriveError::Unavailable)?
-            .map_err(|_| DriveError::Unavailable)?
-            .map_err(|_| DriveError::Unavailable)?;
+            .map_err(|_| {
+                diagnostic.record(Stage::Request, Reason::Deadline, None);
+                DriveError::Unavailable
+            })?
+            .map_err(|_| {
+                diagnostic.record(Stage::Request, Reason::TaskJoin, None);
+                DriveError::Unavailable
+            })?
+            .map_err(|error| {
+                diagnostic.transport(&error);
+                DriveError::Unavailable
+            })?;
+            if response.body.len() > DRIVE_CHUNK_MAX_BYTES as usize {
+                diagnostic.record(Stage::Response, Reason::ResponseSize, None);
+                return Err(DriveError::Unavailable);
+            }
             if !(200..300).contains(&response.status) {
-                return Err(classify(response.status, &response.body, false, ""));
+                return Err(classify(
+                    response.status,
+                    &response.body,
+                    false,
+                    "",
+                    &diagnostic,
+                ));
             }
             let expected = total.saturating_sub(offset).min(length) as usize;
             if response.body.len() != expected {
+                diagnostic.record(Stage::Response, Reason::ResponseLength, None);
                 return Err(DriveError::Unavailable);
             }
             bytes.extend_from_slice(&response.body);
@@ -301,22 +390,40 @@ impl DriveBackend {
         &self,
         id: &str,
         result: Result<DriveMutationResponse, DriveError>,
+        diagnostic: &Diagnostic,
     ) -> Result<DriveMutationResponse, DriveError> {
         // A parseable success can still be an invalid completion. Classify it
         // before reconciliation so it also gets one original-request inquiry.
         let result = result.and_then(|response| {
-            self.validate_completion(id, &response)
+            self.validate_completion(id, &response, diagnostic)
                 .map_err(|_| DriveError::OutcomeUnknown(id.into()))?;
             Ok(response)
         });
         match result {
-            Err(DriveError::OutcomeUnknown(_)) => match self.status(id).await {
+            Err(DriveError::OutcomeUnknown(_)) => match self
+                .status_with_diagnostic(id, &diagnostic.at_stage(Stage::Reconcile))
+                .await
+            {
                 Ok(DriveRequestStatusResponse {
                     state: DriveRequestState::Committed,
                     response: Some(response),
                     ..
-                }) => Ok(response), // status validates both IDs and entry metadata
-                _ => Err(DriveError::OutcomeUnknown(id.into())),
+                }) => {
+                    diagnostic.record(Stage::Reconcile, Reason::Reconciled, None);
+                    Ok(response) // status validates both IDs and entry metadata
+                }
+                Ok(status) => {
+                    diagnostic.record(
+                        Stage::Reconcile,
+                        match status.state {
+                            DriveRequestState::Uncommitted => Reason::NotCommittedSnapshot,
+                            DriveRequestState::Committed => Reason::IncompleteReceipt,
+                        },
+                        None,
+                    );
+                    Err(DriveError::OutcomeUnknown(id.into()))
+                }
+                Err(_) => Err(DriveError::OutcomeUnknown(id.into())),
             },
             other => other,
         }
@@ -325,12 +432,14 @@ impl DriveBackend {
         &self,
         id: &str,
         response: &DriveMutationResponse,
+        diagnostic: &Diagnostic,
     ) -> Result<(), DriveError> {
         if response.request_id != id {
+            diagnostic.record(Stage::Completion, Reason::Identity, None);
             return Err(DriveError::Unavailable);
         }
         if let Some(entry) = &response.entry {
-            self.validate_entry(entry)?;
+            self.validate_entry_at(entry, diagnostic)?;
         }
         Ok(())
     }
@@ -338,9 +447,11 @@ impl DriveBackend {
         &self,
         request: DriveMutationRequest,
     ) -> Result<DriveMutationResponse, DriveError> {
+        let diagnostic = Diagnostic::new(Adapter::Drive, "mutate", true, Some(&request.request_id));
         let body = serde_json::to_string(&request)
             .map_err(|_| DriveError::Invalid("invalid mutation".into()))?;
         if body.len() > 512 * 1024 {
+            diagnostic.record(Stage::Request, Reason::RequestSize, None);
             return Err(DriveError::Limit);
         }
         let result = self
@@ -352,16 +463,20 @@ impl DriveBackend {
                 ),
                 true,
                 &request.request_id,
+                &diagnostic,
             )
             .await;
-        self.reconcile(&request.request_id, result).await
+        self.reconcile(&request.request_id, result, &diagnostic)
+            .await
     }
     pub async fn upload(
         &self,
         query: DriveUploadQuery,
         bytes: Vec<u8>,
     ) -> Result<DriveMutationResponse, DriveError> {
+        let diagnostic = Diagnostic::new(Adapter::Drive, "upload", true, Some(&query.request_id));
         if bytes.len() > DRIVE_FILE_MAX_BYTES as usize {
+            diagnostic.record(Stage::Request, Reason::RequestSize, None);
             return Err(DriveError::Limit);
         }
         let path = format!("{}/upload{}", self.base()?, query_string(&query)?);
@@ -372,29 +487,56 @@ impl DriveBackend {
             max_response_bytes: RESPONSE_LIMIT,
         };
         let client = self.client.clone();
+        let deadline = self.deadline;
         let result = tokio::time::timeout(
-            DEADLINE,
+            deadline,
             tokio::task::spawn_blocking(move || {
-                client.execute_binary_with_timeout(request, DEADLINE)
+                client.execute_binary_with_timeout(request, deadline)
             }),
         )
         .await;
         let result = match result {
+            Ok(Ok(Ok(response))) if response.body.len() > RESPONSE_LIMIT => {
+                diagnostic.record(Stage::Response, Reason::ResponseSize, None);
+                Err(DriveError::OutcomeUnknown(query.request_id.clone()))
+            }
             Ok(Ok(Ok(response))) if (200..300).contains(&response.status) => {
-                serde_json::from_slice(&response.body)
-                    .map_err(|_| DriveError::OutcomeUnknown(query.request_id.clone()))
+                serde_json::from_slice(&response.body).map_err(|_| {
+                    diagnostic.record(Stage::Response, Reason::Decode, None);
+                    DriveError::OutcomeUnknown(query.request_id.clone())
+                })
             }
             Ok(Ok(Ok(response))) => Err(classify(
                 response.status,
                 &response.body,
                 true,
                 &query.request_id,
+                &diagnostic,
             )),
-            _ => Err(DriveError::OutcomeUnknown(query.request_id.clone())),
+            Err(_) => {
+                diagnostic.record(Stage::Request, Reason::Deadline, None);
+                Err(DriveError::OutcomeUnknown(query.request_id.clone()))
+            }
+            Ok(Err(_)) => {
+                diagnostic.record(Stage::Request, Reason::TaskJoin, None);
+                Err(DriveError::OutcomeUnknown(query.request_id.clone()))
+            }
+            Ok(Ok(Err(error))) => {
+                diagnostic.transport(&error);
+                Err(DriveError::OutcomeUnknown(query.request_id.clone()))
+            }
         };
-        self.reconcile(&query.request_id, result).await
+        self.reconcile(&query.request_id, result, &diagnostic).await
     }
     pub async fn status(&self, request_id: &str) -> Result<DriveRequestStatusResponse, DriveError> {
+        let diagnostic = Diagnostic::new(Adapter::Drive, "request_status", false, Some(request_id));
+        self.status_with_diagnostic(request_id, &diagnostic).await
+    }
+    async fn status_with_diagnostic(
+        &self,
+        request_id: &str,
+        diagnostic: &Diagnostic,
+    ) -> Result<DriveRequestStatusResponse, DriveError> {
         if request_id.is_empty() || request_id.len() > 256 {
             return Err(DriveError::Invalid("invalid request ID".into()));
         }
@@ -403,13 +545,15 @@ impl DriveBackend {
                 WorkspaceRequest::get(format!("{}/requests/{}", self.base()?, encode(request_id))),
                 false,
                 "",
+                diagnostic,
             )
             .await?;
         if response.request_id != request_id {
+            diagnostic.record(Stage::Reconcile, Reason::Identity, None);
             return Err(DriveError::Unavailable);
         }
         if let Some(result) = &response.response {
-            self.validate_completion(request_id, result)?;
+            self.validate_completion(request_id, result, diagnostic)?;
         }
         Ok(response)
     }
@@ -463,11 +607,21 @@ fn query_string<T: Serialize>(value: &T) -> Result<String, DriveError> {
         format!("?{}", pairs.join("&"))
     })
 }
-fn classify(status: u16, body: &[u8], mutation: bool, id: &str) -> DriveError {
+fn classify(
+    status: u16,
+    body: &[u8],
+    mutation: bool,
+    id: &str,
+    diagnostic: &Diagnostic,
+) -> DriveError {
     if let Ok(error) = serde_json::from_slice::<DriveApiError>(body) {
-        if error.classification == DriveFailureClassification::Unknown {
+        if error.classification == DriveFailureClassification::Unknown
+            || error.code == DriveApiErrorCode::OutcomeUnknown
+        {
+            diagnostic.record(Stage::Response, Reason::BackendUnknown, Some(status));
             return DriveError::OutcomeUnknown(id.into());
         }
+        diagnostic.record(Stage::Response, Reason::BackendRejected, Some(status));
         return match error.code {
             DriveApiErrorCode::Denied => DriveError::Denied,
             DriveApiErrorCode::NotFound => DriveError::NotFound,
@@ -478,6 +632,7 @@ fn classify(status: u16, body: &[u8], mutation: bool, id: &str) -> DriveError {
             DriveApiErrorCode::OutcomeUnknown => DriveError::OutcomeUnknown(id.into()),
         };
     }
+    diagnostic.record(Stage::Response, Reason::UnrecognizedResponse, Some(status));
     if matches!(status, 401 | 403) {
         DriveError::Denied
     } else if mutation {
@@ -486,3 +641,7 @@ fn classify(status: u16, body: &[u8], mutation: bool, id: &str) -> DriveError {
         DriveError::Unavailable
     }
 }
+
+#[cfg(test)]
+#[path = "backend_tests.rs"]
+mod tests;

crates/worker/src/feature/builtin/drive/backend_tests.rs

@@ -0,0 +1,644 @@
+//! Production Drive adapter fault injection. Only the Workspace transport boundary is substituted;
+//! retained fixture effects and original-request inquiries are inspected independently of replies.
+//! The HTTP case additionally connects the adapter to reqwest and an isolated loopback router.
+use super::*;
+use crate::feature::builtin::backend_diagnostic::{Reason, tests::Capture};
+use crate::worker::{WorkspaceBinaryResponse, WorkspaceClientError, WorkspaceResponse};
+use std::sync::{Condvar, Mutex};
+use tracing::instrument::WithSubscriber;
+
+const SECRET: &str = "secret-token /private/drive credential=unsafe";
+const ID: &str = "host-request-1";
+
+fn entry() -> DriveEntry {
+    DriveEntry {
+        entry: DriveEntryRef {
+            workspace_id: "workspace".into(),
+            node_id: "2".into(),
+        },
+        parent: Some(DriveEntryRef {
+            workspace_id: "workspace".into(),
+            node_id: "1".into(),
+        }),
+        name: SECRET.into(),
+        kind: DriveEntryKind::File,
+        last_mutation_id: "old-request".into(),
+        size: Some(5),
+        content_type: Some("text/plain".into()),
+        updated_by: SECRET.into(),
+        updated_at: SECRET.into(),
+        latest_url: SECRET.into(),
+    }
+}
+fn mutation() -> DriveMutationRequest {
+    DriveMutationRequest {
+        request_id: ID.into(),
+        mutation: DriveMutation::UpdateText {
+            id: entry().entry,
+            expected_mutation_id: "old-request".into(),
+            text: SECRET.into(),
+            content_type: "text/plain".into(),
+        },
+    }
+}
+fn json(value: impl Serialize) -> WorkspaceResponse {
+    WorkspaceResponse {
+        status: 200,
+        body: serde_json::to_string(&value).unwrap(),
+    }
+}
+#[derive(Clone, Copy, Debug)]
+enum Fault {
+    Communication,
+    Unavailable,
+    InvalidPath,
+    Decode,
+    Size,
+    Length,
+    Identity,
+    Metadata,
+    Rejected,
+    Unknown,
+    Unrecognized,
+    Incomplete,
+    Panic,
+    Deadline,
+    None,
+}
+#[derive(Debug, Default)]
+struct State {
+    mutations: usize,
+    saved: Option<String>,
+    requests: Vec<WorkspaceRequest>,
+    binary: Vec<WorkspaceBinaryRequest>,
+    released: bool,
+    started: bool,
+    finished: bool,
+}
+#[derive(Debug)]
+struct Client {
+    fault: Fault,
+    receipt: Fault,
+    state: Mutex<State>,
+    changed: Condvar,
+    started: tokio::sync::Notify,
+}
+impl Client {
+    fn new(fault: Fault, receipt: Fault) -> Arc<Self> {
+        Arc::new(Self {
+            fault,
+            receipt,
+            state: Mutex::new(State::default()),
+            changed: Condvar::new(),
+            started: tokio::sync::Notify::new(),
+        })
+    }
+    fn fail(
+        &self,
+        fault: Fault,
+        success: WorkspaceResponse,
+    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+        match fault {
+            Fault::Communication => Err(WorkspaceClientError::Request(SECRET.into())),
+            Fault::Unavailable => Err(WorkspaceClientError::Unavailable(SECRET.into())),
+            Fault::InvalidPath => Err(WorkspaceClientError::InvalidPath(SECRET.into())),
+            Fault::Panic => panic!("injected adapter task panic"),
+            Fault::Decode => Ok(WorkspaceResponse {
+                status: 200,
+                body: SECRET.into(),
+            }),
+            Fault::Size => Ok(WorkspaceResponse {
+                status: 200,
+                body: SECRET.repeat(RESPONSE_LIMIT / SECRET.len() + 1),
+            }),
+            Fault::Rejected | Fault::Unknown => Ok(WorkspaceResponse {
+                status: 409,
+                body: serde_json::to_string(&DriveApiError {
+                    code: DriveApiErrorCode::Conflict,
+                    message: SECRET.into(),
+                    classification: if matches!(fault, Fault::Unknown) {
+                        DriveFailureClassification::Unknown
+                    } else {
+                        DriveFailureClassification::NotCommitted
+                    },
+                })
+                .unwrap(),
+            }),
+            Fault::Unrecognized => Ok(WorkspaceResponse {
+                status: 503,
+                body: SECRET.into(),
+            }),
+            _ => Ok(success),
+        }
+    }
+    fn release(&self) {
+        self.state.lock().unwrap().released = true;
+        self.changed.notify_all();
+    }
+    fn wait(&self, finished: bool) {
+        let (state, timeout) = self
+            .changed
+            .wait_timeout_while(self.state.lock().unwrap(), Duration::from_secs(5), |s| {
+                if finished { !s.finished } else { !s.started }
+            })
+            .unwrap();
+        assert!(!timeout.timed_out());
+        assert!(if finished {
+            state.finished
+        } else {
+            state.started
+        });
+    }
+}
+struct Release(Arc<Client>);
+impl Drop for Release {
+    fn drop(&mut self) {
+        self.0.release();
+    }
+}
+impl WorkspaceClient for Client {
+    fn workspace_id(&self) -> Option<&str> {
+        Some("workspace")
+    }
+    fn kind(&self) -> &str {
+        "drive-diagnostic-fixture"
+    }
+    fn is_available(&self) -> bool {
+        true
+    }
+    fn execute(
+        &self,
+        request: WorkspaceRequest,
+    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+        let is_status = request.path.contains("/requests/");
+        let fault = if is_status { self.receipt } else { self.fault };
+        let mutation = request.method == WorkspaceRequestMethod::Post;
+        {
+            let mut state = self.state.lock().unwrap();
+            if mutation
+                && !matches!(
+                    fault,
+                    Fault::Rejected | Fault::InvalidPath | Fault::Unavailable
+                )
+            {
+                let input: DriveMutationRequest =
+                    serde_json::from_str(request.body.as_ref().unwrap()).unwrap();
+                assert_eq!(input, super::tests::mutation());
+                state.mutations += 1;
+                if let DriveMutation::UpdateText { text, .. } = input.mutation {
+                    state.saved = Some(text);
+                }
+            }
+            state.requests.push(request);
+            state.started = true;
+            self.started.notify_one();
+            self.changed.notify_all();
+            if matches!(fault, Fault::Deadline) {
+                let (_state, timeout) = self
+                    .changed
+                    .wait_timeout_while(state, Duration::from_secs(5), |s| !s.released)
+                    .unwrap();
+                assert!(
+                    !timeout.timed_out(),
+                    "transport must be explicitly released"
+                );
+            }
+        }
+        let mut metadata = entry();
+        if matches!(fault, Fault::Identity) {
+            metadata.entry.node_id = "3".into();
+        }
+        if matches!(fault, Fault::Metadata) {
+            metadata.size = Some(DRIVE_FILE_MAX_BYTES + 1);
+        }
+        let completion = DriveMutationResponse {
+            request_id: if matches!(fault, Fault::Identity) {
+                "other-request"
+            } else {
+                ID
+            }
+            .into(),
+            entry: Some(metadata.clone()),
+        };
+        let value = if is_status {
+            json(DriveRequestStatusResponse {
+                request_id: if matches!(fault, Fault::Identity) {
+                    "other-request"
+                } else {
+                    ID
+                }
+                .into(),
+                state: if matches!(
+                    fault,
+                    Fault::None | Fault::Identity | Fault::Metadata | Fault::Incomplete
+                ) {
+                    DriveRequestState::Committed
+                } else {
+                    DriveRequestState::Uncommitted
+                },
+                response: if matches!(fault, Fault::None | Fault::Identity | Fault::Metadata) {
+                    Some(completion)
+                } else {
+                    None
+                },
+            })
+        } else if mutation {
+            json(completion)
+        } else {
+            json(metadata)
+        };
+        if !is_status {
+            self.state.lock().unwrap().finished = true;
+        }
+        self.changed.notify_all();
+        self.fail(fault, value)
+    }
+    fn execute_binary(
+        &self,
+        request: WorkspaceBinaryRequest,
+    ) -> Result<WorkspaceBinaryResponse, WorkspaceClientError> {
+        let upload = request.method == WorkspaceRequestMethod::Put;
+        {
+            let mut state = self.state.lock().unwrap();
+            if upload {
+                state.mutations += 1;
+                state.saved = Some(SECRET.into());
+            }
+            state.binary.push(request);
+        }
+        let success = if upload {
+            json(DriveMutationResponse {
+                request_id: ID.into(),
+                entry: Some(entry()),
+            })
+        } else {
+            WorkspaceResponse {
+                status: 200,
+                body: if matches!(self.fault, Fault::Length) {
+                    "x"
+                } else {
+                    "12345"
+                }
+                .into(),
+            }
+        };
+        self.fail(self.fault, success)
+            .map(|r| WorkspaceBinaryResponse {
+                status: r.status,
+                body: r.body.into_bytes(),
+            })
+    }
+}
+fn safe(capture: &Capture, reason: Reason, public: &str) {
+    let text = capture.text();
+    assert!(text.contains(&format!("reason={reason:?}")), "{text}");
+    assert!(text.contains("adapter=Drive"), "{text}");
+    assert!(text.contains("correlation="), "{text}");
+    for secret in [
+        SECRET,
+        "secret-token",
+        "/private/drive",
+        "credential=unsafe",
+    ] {
+        assert!(!text.contains(secret), "internal diagnostic leaked content");
+        assert!(!public.contains(secret), "public error leaked content");
+    }
+}
+fn reason(fault: Fault) -> Reason {
+    match fault {
+        Fault::Communication => Reason::Communication,
+        Fault::Unavailable => Reason::TransportUnavailable,
+        Fault::InvalidPath => Reason::InvalidPath,
+        Fault::Decode => Reason::Decode,
+        Fault::Size => Reason::ResponseSize,
+        Fault::Length => Reason::ResponseLength,
+        Fault::Identity => Reason::Identity,
+        Fault::Metadata => Reason::InvalidMetadata,
+        Fault::Rejected => Reason::BackendRejected,
+        Fault::Unknown => Reason::BackendUnknown,
+        Fault::Unrecognized => Reason::UnrecognizedResponse,
+        Fault::Panic => Reason::TaskJoin,
+        Fault::Deadline => Reason::Deadline,
+        Fault::Incomplete => Reason::IncompleteReceipt,
+        Fault::None => panic!("no fault"),
+    }
+}
+#[tokio::test]
+async fn production_read_failures_retain_public_classification_and_safe_cause() {
+    for fault in [
+        Fault::Communication,
+        Fault::Unavailable,
+        Fault::InvalidPath,
+        Fault::Decode,
+        Fault::Size,
+        Fault::Identity,
+        Fault::Metadata,
+        Fault::Rejected,
+        Fault::Unknown,
+        Fault::Unrecognized,
+        Fault::Panic,
+    ] {
+        let client = Client::new(fault, Fault::None);
+        let backend = DriveBackend::new(client.clone());
+        let capture = Capture::default();
+        let error = backend
+            .metadata(&entry().entry)
+            .with_subscriber(capture.dispatch())
+            .await
+            .unwrap_err();
+        match fault {
+            Fault::Size => assert!(matches!(error, DriveError::Limit)),
+            Fault::Rejected => assert!(matches!(error, DriveError::Conflict)),
+            Fault::Unknown => assert!(matches!(error, DriveError::OutcomeUnknown(_))),
+            _ => assert!(matches!(error, DriveError::Unavailable)),
+        }
+        safe(&capture, reason(fault), &error.to_string());
+        let state = client.state.lock().unwrap();
+        assert_eq!(state.requests.len(), 1);
+        assert_eq!(state.mutations, 0);
+    }
+}
+#[tokio::test]
+async fn production_mutation_loss_reconciles_only_original_identity_and_never_resends() {
+    for fault in [
+        Fault::Communication,
+        Fault::Decode,
+        Fault::Size,
+        Fault::Identity,
+        Fault::Metadata,
+        Fault::Unknown,
+        Fault::Unrecognized,
+        Fault::Panic,
+    ] {
+        for receipt in [
+            Fault::None,
+            Fault::Identity,
+            Fault::Metadata,
+            Fault::Decode,
+            Fault::Communication,
+            Fault::Unrecognized,
+            Fault::Length,
+            Fault::Incomplete,
+        ] {
+            let client = Client::new(fault, receipt);
+            let backend = DriveBackend::new(client.clone());
+            let capture = Capture::default();
+            let result = backend
+                .mutate(mutation())
+                .with_subscriber(capture.dispatch())
+                .await;
+            if matches!(receipt, Fault::None) {
+                assert_eq!(result.unwrap().request_id, ID);
+            } else {
+                assert!(matches!(result, Err(DriveError::OutcomeUnknown(ref id)) if id == ID));
+            }
+            safe(&capture, reason(fault), "");
+            let text = capture.text();
+            if matches!(receipt, Fault::Length | Fault::Incomplete) {
+                safe(
+                    &capture,
+                    if matches!(receipt, Fault::Length) {
+                        Reason::NotCommittedSnapshot
+                    } else {
+                        Reason::IncompleteReceipt
+                    },
+                    "",
+                );
+            }
+            assert!(text.contains("stage=Reconcile"), "{text}");
+            let correlations: std::collections::BTreeSet<_> = text
+                .lines()
+                .filter_map(|l| l.split("correlation=").nth(1))
+                .filter_map(|l| l.split(';').next())
+                .collect();
+            assert_eq!(
+                correlations.len(),
+                1,
+                "original loss and inquiry must correlate"
+            );
+            assert!(
+                !text.contains(ID),
+                "request identity must be opaque in logs"
+            );
+            let state = client.state.lock().unwrap();
+            assert_eq!(state.mutations, 1);
+            assert_eq!(state.saved.as_deref(), Some(SECRET));
+            assert_eq!(state.requests.len(), 2);
+            assert_eq!(state.requests[1].method, WorkspaceRequestMethod::Get);
+            assert!(state.requests[1].path.ends_with(&format!("/requests/{ID}")));
+        }
+    }
+}
+#[tokio::test]
+async fn production_rejection_has_no_effect_or_status_inquiry() {
+    let client = Client::new(Fault::Rejected, Fault::None);
+    let backend = DriveBackend::new(client.clone());
+    let capture = Capture::default();
+    let error = backend
+        .mutate(mutation())
+        .with_subscriber(capture.dispatch())
+        .await
+        .unwrap_err();
+    assert!(matches!(error, DriveError::Conflict));
+    safe(&capture, Reason::BackendRejected, &error.to_string());
+    let state = client.state.lock().unwrap();
+    assert_eq!(state.mutations, 0);
+    assert_eq!(state.requests.len(), 1);
+}
+#[tokio::test]
+async fn production_binary_length_transport_and_upload_loss_are_diagnosed() {
+    for fault in [
+        Fault::Length,
+        Fault::Communication,
+        Fault::Unavailable,
+        Fault::Panic,
+    ] {
+        let client = Client::new(fault, Fault::None);
+        let backend = DriveBackend::new(client.clone());
+        let capture = Capture::default();
+        let error = backend
+            .bytes(&entry(), 5)
+            .with_subscriber(capture.dispatch())
+            .await
+            .unwrap_err();
+        assert!(matches!(error, DriveError::Unavailable));
+        safe(&capture, reason(fault), &error.to_string());
+        assert_eq!(client.state.lock().unwrap().binary.len(), 1);
+    }
+    for fault in [Fault::Decode, Fault::Communication, Fault::Size] {
+        let client = Client::new(fault, Fault::Unrecognized);
+        let backend = DriveBackend::new(client.clone());
+        let capture = Capture::default();
+        let query = DriveUploadQuery {
+            operation: DriveUploadOperation::Update,
+            request_id: ID.into(),
+            entry_workspace_id: "workspace".into(),
+            parent_id: None,
+            name: None,
+            id: Some("2".into()),
+            expected_mutation_id: Some("old-request".into()),
+            content_type: "text/plain".into(),
+            size: 5,
+            sha256: "hash".into(),
+        };
+        let error = backend
+            .upload(query, b"12345".to_vec())
+            .with_subscriber(capture.dispatch())
+            .await
+            .unwrap_err();
+        assert!(matches!(error, DriveError::OutcomeUnknown(ref id) if id == ID));
+        safe(&capture, reason(fault), &error.to_string());
+        let state = client.state.lock().unwrap();
+        assert_eq!(state.mutations, 1);
+        assert_eq!(state.binary.len(), 1);
+        assert_eq!(state.requests.len(), 1);
+        assert!(state.requests[0].path.ends_with(&format!("/requests/{ID}")));
+    }
+}
+#[tokio::test]
+async fn production_deadline_does_not_cancel_effect_or_infer_uncommitted_from_snapshot() {
+    for mutation_call in [false, true] {
+        let client = Client::new(Fault::Deadline, Fault::Unrecognized);
+        let release = Release(client.clone());
+        let mut backend = DriveBackend::new(client.clone());
+        backend.deadline = Duration::from_secs(1);
+        let capture = Capture::default();
+        let request = async {
+            if mutation_call {
+                backend.mutate(mutation()).await.map(|_| ())
+            } else {
+                backend.metadata(&entry().entry).await.map(|_| ())
+            }
+        }
+        .with_subscriber(capture.dispatch());
+        tokio::pin!(request);
+        tokio::select! {
+            biased;
+            _ = client.started.notified() => {},
+            _ = &mut request => panic!("request must start before timeout"),
+        }
+        {
+            let state = client.state.lock().unwrap();
+            assert_eq!(state.mutations, usize::from(mutation_call));
+            assert!(!state.released && !state.finished);
+        }
+        let result = request.await;
+        let error = result.unwrap_err();
+        assert!(if mutation_call {
+            matches!(error, DriveError::OutcomeUnknown(ref id) if id == ID)
+        } else {
+            matches!(error, DriveError::Unavailable)
+        });
+        safe(&capture, Reason::Deadline, &error.to_string());
+        drop(release);
+        let waiting = client.clone();
+        tokio::task::spawn_blocking(move || waiting.wait(true))
+            .await
+            .unwrap();
+        let state = client.state.lock().unwrap();
+        assert_eq!(state.mutations, usize::from(mutation_call));
+        assert_eq!(state.saved.is_some(), mutation_call);
+        assert_eq!(state.requests.len(), if mutation_call { 2 } else { 1 });
+    }
+}
+
+#[tokio::test]
+async fn http_reply_loss_keeps_original_receipt_and_retained_effect_without_reexecution() {
+    use axum::{
+        Router,
+        extract::State as HttpState,
+        routing::{get, post},
+    };
+    #[derive(Default)]
+    struct Stored {
+        updates: usize,
+        content: String,
+        ids: Vec<String>,
+    }
+    struct Server(tokio::task::JoinHandle<()>);
+    impl Drop for Server {
+        fn drop(&mut self) {
+            self.0.abort();
+        }
+    }
+    let stored = Arc::new(Mutex::new(Stored::default()));
+    let router = Router::new()
+        .route("/api/w/workspace/drive/metadata", get(|| async { SECRET }))
+        .route(
+            "/api/w/workspace/drive/mutate",
+            post(
+                |HttpState(state): HttpState<Arc<Mutex<Stored>>>,
+                 body: axum::body::Bytes| async move {
+                    let request: DriveMutationRequest = serde_json::from_slice(&body).unwrap();
+                    let mut state = state.lock().unwrap();
+                    state.ids.push(request.request_id);
+                    if let DriveMutation::UpdateText {
+                        text,
+                        expected_mutation_id,
+                        ..
+                    } = request.mutation
+                    {
+                        assert_eq!(expected_mutation_id, "old-request");
+                        state.content = text;
+                        state.updates += 1;
+                    } else {
+                        panic!("unexpected mutation");
+                    }
+                    // The effect is retained; only the success response is malformed.
+                    SECRET
+                },
+            ),
+        )
+        .route(
+            "/api/w/workspace/drive/requests/{id}",
+            get(
+                |HttpState(state): HttpState<Arc<Mutex<Stored>>>,
+                 axum::extract::Path(id): axum::extract::Path<String>| async move {
+                    let state = state.lock().unwrap();
+                    assert_eq!(state.ids, [id.clone()]);
+                    axum::Json(DriveRequestStatusResponse {
+                        request_id: id.clone(),
+                        state: DriveRequestState::Committed,
+                        response: Some(DriveMutationResponse {
+                            request_id: id,
+                            entry: Some(entry()),
+                        }),
+                    })
+                },
+            ),
+        )
+        .with_state(stored.clone());
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+    let server = Server(tokio::spawn(async move {
+        axum::serve(listener, router).await.unwrap();
+    }));
+    let client = Arc::new(crate::worker::TestWorkspaceHttpClient::new(
+        "workspace",
+        format!("http://{address}"),
+    ));
+    let backend = DriveBackend::new(client);
+    let capture = Capture::default();
+    let error = backend
+        .metadata(&entry().entry)
+        .with_subscriber(capture.dispatch())
+        .await
+        .unwrap_err();
+    assert!(matches!(error, DriveError::Unavailable));
+    safe(&capture, Reason::Decode, &error.to_string());
+    let capture = Capture::default();
+    let response = backend
+        .mutate(mutation())
+        .with_subscriber(capture.dispatch())
+        .await
+        .unwrap();
+    assert_eq!(response.request_id, ID);
+    safe(&capture, Reason::Decode, "");
+    assert!(capture.text().contains("reason=Reconciled"));
+    let state = stored.lock().unwrap();
+    assert_eq!(state.updates, 1);
+    assert_eq!(state.content, SECRET);
+    assert_eq!(state.ids, [ID]);
+    drop(state);
+    drop(server);
+}

crates/worker/src/feature/builtin/workspace_config/backend.rs

@@ -7,6 +7,7 @@
 //! undecodable mutation responses do not prove non-commit; dispatched requests are never resent.
 //! This adapter does not reimplement Backend source-persistence transactions or activation.
 use std::sync::Arc;
+use std::time::Duration;
 
 use async_trait::async_trait;
 use serde::{Serialize, de::DeserializeOwned};
@@ -20,6 +21,7 @@ use super::{
     ATTACHMENT_ALIAS, CONTENT_ROOT, ConfigAccess, ConfigAttachError, ConfigAttachment, DEADLINE,
     WorkspaceConfigAttachmentBackend,
 };
+use crate::feature::builtin::backend_diagnostic::{Adapter, Diagnostic, Reason, Stage};
 use crate::wip::WipOperationError;
 use crate::worker::{
     WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod,
@@ -34,11 +36,21 @@ const MAX_RESPONSE_BYTES: usize = 2 * 1024 * 1024;
 #[derive(Clone)]
 pub struct WorkspaceConfigBackend {
     client: Arc<dyn WorkspaceClient>,
+    deadline: Duration,
 }
 
 impl WorkspaceConfigBackend {
     pub fn new(client: Arc<dyn WorkspaceClient>) -> Self {
-        Self { client }
+        Self {
+            client,
+            deadline: DEADLINE,
+        }
+    }
+
+    #[cfg(test)]
+    pub(super) fn with_deadline(mut self, deadline: Duration) -> Self {
+        self.deadline = deadline;
+        self
     }
 
     fn base(&self) -> Result<String, WipOperationError> {
@@ -47,6 +59,7 @@ impl WorkspaceConfigBackend {
             .workspace_id()
             .filter(|id| !id.is_empty())
             .ok_or_else(|| {
+                diagnostic("authority", false).record(Stage::Request, Reason::Identity, None);
                 failure(
                     ProtocolErrorCode::PermissionDenied,
                     "Workspace config authority unavailable",
@@ -69,11 +82,13 @@ impl WorkspaceConfigBackend {
         &self,
         request: WorkspaceRequest,
         mutation: bool,
+        diagnostic: &Diagnostic,
     ) -> Result<T, WipOperationError> {
         let response = self
             .client
-            .execute_with_timeout(request, DEADLINE)
+            .execute_with_timeout(request, self.deadline)
             .map_err(|error| {
+                diagnostic.transport(&error);
                 if mutation && !matches!(error, WorkspaceClientError::InvalidPath(_)) {
                     unknown()
                 } else {
@@ -84,6 +99,7 @@ impl WorkspaceConfigBackend {
                 }
             })?;
         if response.body.len() > MAX_RESPONSE_BYTES {
+            diagnostic.record(Stage::Response, Reason::ResponseSize, Some(response.status));
             return Err(if mutation {
                 unknown()
             } else {
@@ -96,8 +112,18 @@ impl WorkspaceConfigBackend {
         if !response.is_success() {
             if let Ok(error) = serde_json::from_str::<WorkspaceConfigApiError>(&response.body) {
                 if error.classification == WorkspaceConfigFailureClassification::Unknown {
+                    diagnostic.record(
+                        Stage::Response,
+                        Reason::BackendUnknown,
+                        Some(response.status),
+                    );
                     return Err(unknown());
                 }
+                diagnostic.record(
+                    Stage::Response,
+                    Reason::BackendRejected,
+                    Some(response.status),
+                );
                 // Do not reflect arbitrary transport/provider text into model output.
                 let code = match error.status {
                     401 | 403 | 404 => ProtocolErrorCode::PermissionDenied,
@@ -111,6 +137,11 @@ impl WorkspaceConfigBackend {
                     "Workspace config request rejected without commit",
                 ));
             }
+            diagnostic.record(
+                Stage::Response,
+                Reason::UnrecognizedResponse,
+                Some(response.status),
+            );
             return Err(if mutation {
                 unknown()
             } else {
@@ -125,6 +156,7 @@ impl WorkspaceConfigBackend {
             });
         }
         serde_json::from_str(&response.body).map_err(|_| {
+            diagnostic.record(Stage::Response, Reason::Decode, Some(response.status));
             if mutation {
                 unknown()
             } else {
@@ -138,10 +170,17 @@ impl WorkspaceConfigBackend {
 
     /// GET checks the current grant even when the Worker has not attached yet.
     pub fn current(&self) -> Result<Option<WorkspaceConfigAttachment>, WipOperationError> {
+        self.current_diagnosed(&diagnostic("current", false))
+    }
+
+    fn current_diagnosed(
+        &self,
+        diagnostic: &Diagnostic,
+    ) -> Result<Option<WorkspaceConfigAttachment>, WipOperationError> {
         let result: Option<WorkspaceConfigAttachment> =
-            self.request(WorkspaceRequest::get(self.base()?), false)?;
+            self.request(WorkspaceRequest::get(self.base()?), false, diagnostic)?;
         if let Some(attachment) = &result {
-            self.validate_attachment(attachment)?;
+            self.validate_attachment(attachment, diagnostic)?;
         }
         Ok(result)
     }
@@ -149,6 +188,7 @@ impl WorkspaceConfigBackend {
     fn validate_attachment(
         &self,
         attachment: &WorkspaceConfigAttachment,
+        diagnostic: &Diagnostic,
     ) -> Result<(), WipOperationError> {
         if Some(attachment.workspace_id.as_str()) != self.client.workspace_id()
             || attachment.alias != ATTACHMENT_ALIAS
@@ -160,6 +200,7 @@ impl WorkspaceConfigBackend {
                 .bytes()
                 .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_'))
         {
+            diagnostic.record(Stage::Response, Reason::Identity, None);
             return Err(failure(
                 ProtocolErrorCode::PermissionDenied,
                 "Invalid Workspace config attachment identity",
@@ -173,14 +214,34 @@ impl WorkspaceConfigBackend {
         suffix: &str,
         request: &R,
         mutation: bool,
+    ) -> Result<T, WipOperationError> {
+        let operation = match suffix {
+            "/observe" => "observe",
+            "/read" => "read",
+            "/commit" => "commit",
+            "" => "attach",
+            _ => "post",
+        };
+        self.post_diagnosed(suffix, request, mutation, diagnostic(operation, mutation))
+            .await
+    }
+
+    pub(super) async fn post_diagnosed<T: DeserializeOwned + Send + 'static, R: Serialize>(
+        &self,
+        suffix: &str,
+        request: &R,
+        mutation: bool,
+        diagnostic: Diagnostic,
     ) -> Result<T, WipOperationError> {
         let body = serde_json::to_string(request).map_err(|_| {
+            diagnostic.record(Stage::Request, Reason::Decode, None);
             failure(
                 ProtocolErrorCode::InvalidArguments,
                 "Invalid Workspace config request",
             )
         })?;
         if body.len() > MAX_RESPONSE_BYTES {
+            diagnostic.record(Stage::Request, Reason::RequestSize, None);
             return Err(failure(
                 ProtocolErrorCode::ResourceLimitExceeded,
                 "Workspace config request exceeds limit",
@@ -192,12 +253,16 @@ impl WorkspaceConfigBackend {
             body,
         );
         let this = self.clone();
+        let request_diagnostic = diagnostic.clone();
         tokio::time::timeout(
-            DEADLINE,
-            tokio::task::spawn_blocking(move || this.request(request, mutation)),
+            self.deadline,
+            tokio::task::spawn_blocking(move || {
+                this.request(request, mutation, &request_diagnostic)
+            }),
         )
         .await
         .map_err(|_| {
+            diagnostic.record(Stage::Completion, Reason::Deadline, None);
             if mutation {
                 unknown()
             } else {
@@ -208,6 +273,7 @@ impl WorkspaceConfigBackend {
             }
         })?
         .map_err(|_| {
+            diagnostic.record(Stage::Completion, Reason::TaskJoin, None);
             if mutation {
                 unknown()
             } else {
@@ -223,18 +289,22 @@ impl WorkspaceConfigBackend {
         &self,
     ) -> Result<Option<WorkspaceConfigAttachment>, WipOperationError> {
         let this = self.clone();
+        let diagnostic = diagnostic("current", false);
+        let request_diagnostic = diagnostic.clone();
         tokio::time::timeout(
-            DEADLINE,
-            tokio::task::spawn_blocking(move || this.current()),
+            self.deadline,
+            tokio::task::spawn_blocking(move || this.current_diagnosed(&request_diagnostic)),
         )
         .await
         .map_err(|_| {
+            diagnostic.record(Stage::Completion, Reason::Deadline, None);
             failure(
                 ProtocolErrorCode::ResourceLimitExceeded,
                 "Workspace config request deadline exceeded",
             )
         })?
         .map_err(|_| {
+            diagnostic.record(Stage::Completion, Reason::TaskJoin, None);
             failure(
                 ProtocolErrorCode::Internal,
                 "Workspace config transport unavailable",
@@ -249,10 +319,14 @@ impl WorkspaceConfigAttachmentBackend for WorkspaceConfigBackend {
         self.client.is_available()
     }
     fn validate_access(&self, access: Option<ConfigAccess>) -> Result<(), ConfigAttachError> {
-        let current = self.current().map_err(attach_error)?;
+        let diagnostic = diagnostic("validate_access", false);
+        let current = self
+            .current_diagnosed(&diagnostic)
+            .map_err(|error| attach_error(error, &diagnostic))?;
         if access == Some(ConfigAccess::ReadWrite)
             && current.is_some_and(|a| a.access == WorkspaceConfigAccess::ReadOnly)
         {
+            diagnostic.record(Stage::Attach, Reason::AccessCeiling, None);
             return Err(ConfigAttachError::Denied);
         }
         Ok(())
@@ -271,10 +345,16 @@ impl WorkspaceConfigAttachmentBackend for WorkspaceConfigBackend {
                 ConfigAccess::ReadWrite => WorkspaceConfigAccess::ReadWrite,
             }),
         };
-        let attachment: WorkspaceConfigAttachment =
-            self.post("", &request, true).await.map_err(attach_error)?;
-        self.validate_attachment(&attachment)
-            .map_err(|_| ConfigAttachError::OutcomeUnknown)?;
+        let diagnostic = diagnostic("attach", true);
+        let attachment: WorkspaceConfigAttachment = self
+            .post_diagnosed("", &request, true, diagnostic.clone())
+            .await
+            .map_err(|error| attach_error(error, &diagnostic))?;
+        self.validate_attachment(&attachment, &diagnostic)
+            .map_err(|_| {
+                diagnostic.record(Stage::Attach, Reason::Identity, None);
+                ConfigAttachError::OutcomeUnknown
+            })?;
         Ok(ConfigAttachment {
             connection_id: attachment.connection_id,
             access: match attachment.access {
@@ -286,7 +366,8 @@ impl WorkspaceConfigAttachmentBackend for WorkspaceConfigBackend {
     }
 }
 
-fn attach_error(error: WipOperationError) -> ConfigAttachError {
+fn attach_error(error: WipOperationError, diagnostic: &Diagnostic) -> ConfigAttachError {
+    diagnostic.record(Stage::Attach, Reason::AttachConversion, None);
     match error {
         WipOperationError::OutcomeUnknown(_) => ConfigAttachError::OutcomeUnknown,
         WipOperationError::Protocol(error) => match error.code {
@@ -298,6 +379,10 @@ fn attach_error(error: WipOperationError) -> ConfigAttachError {
     }
 }
 
+pub(super) fn diagnostic(operation: &'static str, mutation: bool) -> Diagnostic {
+    Diagnostic::new(Adapter::WorkspaceConfig, operation, mutation, None)
+}
+
 pub(super) fn failure(code: ProtocolErrorCode, message: &str) -> WipOperationError {
     WipOperationError::Protocol(ProtocolError {
         code,
@@ -307,3 +392,7 @@ pub(super) fn failure(code: ProtocolErrorCode, message: &str) -> WipOperationErr
 pub(super) fn unknown() -> WipOperationError {
     WipOperationError::OutcomeUnknown("Workspace config outcome unknown; refresh before any subsequent operation; no automatic retry".into())
 }
+
+#[cfg(test)]
+#[path = "backend_tests.rs"]
+mod tests;

crates/worker/src/feature/builtin/workspace_config/backend_tests.rs

@@ -0,0 +1,580 @@
+//! Production WorkspaceConfigBackend with only WorkspaceClient substituted. The transport fixture
+//! applies effects before injected reply loss, retains its state for rereading, and gates blocking
+//! completion explicitly. These ordinary worker lib tests do not claim Backend transaction coverage.
+use super::*;
+use crate::feature::builtin::backend_diagnostic::tests::Capture;
+use crate::worker::WorkspaceResponse;
+use std::sync::{Condvar, Mutex};
+use tracing::instrument::WithSubscriber;
+
+const UNSAFE: &str = "secret-token /private/config credential=unsafe";
+
+#[derive(Clone, Copy, Debug)]
+enum Fault {
+    Unavailable,
+    InvalidPath,
+    Communication,
+    Decode,
+    Size,
+    Rejected(u16),
+    Unknown,
+    Unrecognized,
+    Identity,
+    Panic,
+    Deadline,
+}
+
+#[derive(Debug, Default)]
+struct Effects {
+    requests: usize,
+    mutations: usize,
+    saved: Option<String>,
+    completed: bool,
+    released: bool,
+}
+
+#[derive(Debug)]
+struct Client {
+    fault: Fault,
+    state: Mutex<Effects>,
+    changed: Condvar,
+    started: tokio::sync::Notify,
+}
+impl Client {
+    fn new(fault: Fault) -> Arc<Self> {
+        Arc::new(Self {
+            fault,
+            state: Mutex::new(Effects::default()),
+            changed: Condvar::new(),
+            started: tokio::sync::Notify::new(),
+        })
+    }
+    fn finish(&self) {
+        self.state.lock().unwrap().completed = true;
+        self.changed.notify_all();
+    }
+    fn release(&self) {
+        self.state.lock().unwrap().released = true;
+        self.changed.notify_all();
+    }
+    fn wait_completed(&self) {
+        let (state, timeout) = self
+            .changed
+            .wait_timeout_while(self.state.lock().unwrap(), Duration::from_secs(5), |s| {
+                !s.completed
+            })
+            .unwrap();
+        assert!(
+            !timeout.timed_out() && state.completed,
+            "blocking request did not finish"
+        );
+    }
+}
+struct Release(Arc<Client>);
+impl Drop for Release {
+    fn drop(&mut self) {
+        self.0.release();
+    }
+}
+
+fn attachment() -> WorkspaceConfigAttachment {
+    WorkspaceConfigAttachment {
+        workspace_id: "workspace".into(),
+        connection_id: "connection-1".into(),
+        alias: ATTACHMENT_ALIAS.into(),
+        content_path: CONTENT_ROOT.into(),
+        working_directory_id: "grant".into(),
+        access: WorkspaceConfigAccess::ReadWrite,
+        name: UNSAFE.into(),
+        purpose: UNSAFE.into(),
+        already_attached: false,
+    }
+}
+
+impl WorkspaceClient for Client {
+    fn workspace_id(&self) -> Option<&str> {
+        Some("workspace")
+    }
+    fn kind(&self) -> &str {
+        "diagnostic-test-boundary"
+    }
+    fn is_available(&self) -> bool {
+        true
+    }
+    fn execute(
+        &self,
+        request: WorkspaceRequest,
+    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+        {
+            let mut state = self.state.lock().unwrap();
+            state.requests += 1;
+            if request.method == WorkspaceRequestMethod::Post
+                && !request.path.ends_with("/read")
+                && !matches!(
+                    self.fault,
+                    Fault::Unavailable | Fault::InvalidPath | Fault::Rejected(_)
+                )
+            {
+                state.mutations += 1;
+                state.saved = request.body.clone();
+            }
+            self.started.notify_one();
+            if matches!(self.fault, Fault::Deadline) {
+                let (_state, timeout) = self
+                    .changed
+                    .wait_timeout_while(state, Duration::from_secs(5), |s| !s.released)
+                    .unwrap();
+                assert!(!timeout.timed_out(), "test must release blocked transport");
+            }
+        }
+        self.finish();
+        match self.fault {
+            Fault::Unavailable => Err(WorkspaceClientError::Unavailable(UNSAFE.into())),
+            Fault::InvalidPath => Err(WorkspaceClientError::InvalidPath(UNSAFE.into())),
+            Fault::Communication => Err(WorkspaceClientError::Request(UNSAFE.into())),
+            Fault::Panic => panic!("injected blocking task failure"),
+            Fault::Decode => Ok(WorkspaceResponse {
+                status: 200,
+                body: UNSAFE.into(),
+            }),
+            Fault::Size => Ok(WorkspaceResponse {
+                status: 200,
+                body: UNSAFE.repeat(MAX_RESPONSE_BYTES / UNSAFE.len() + 1),
+            }),
+            Fault::Rejected(_) | Fault::Unknown => {
+                let status = match self.fault {
+                    Fault::Rejected(status) => status,
+                    _ => 503,
+                };
+                Ok(WorkspaceResponse {
+                    status,
+                    body: serde_json::to_string(&WorkspaceConfigApiError {
+                        status,
+                        code: UNSAFE.into(),
+                        message: UNSAFE.into(),
+                        classification: if matches!(self.fault, Fault::Unknown) {
+                            WorkspaceConfigFailureClassification::Unknown
+                        } else {
+                            WorkspaceConfigFailureClassification::NotCommitted
+                        },
+                    })
+                    .unwrap(),
+                })
+            }
+            Fault::Unrecognized => Ok(WorkspaceResponse {
+                status: 503,
+                body: UNSAFE.into(),
+            }),
+            Fault::Identity | Fault::Deadline => {
+                let mut value = attachment();
+                if matches!(self.fault, Fault::Identity) {
+                    value.workspace_id = UNSAFE.into();
+                }
+                Ok(WorkspaceResponse {
+                    status: 200,
+                    body: serde_json::to_string(&value).unwrap(),
+                })
+            }
+        }
+    }
+}
+
+fn assert_safe(capture: &Capture, reason: Reason, stage: Stage, public: &str) {
+    let text = capture.text();
+    assert!(
+        text.contains(&format!("reason={reason:?}")),
+        "missing reason: {text}"
+    );
+    assert!(
+        text.contains(&format!("stage={stage:?}")),
+        "missing stage: {text}"
+    );
+    assert!(text.contains("adapter=WorkspaceConfig"), "{text}");
+    assert!(text.contains("correlation="), "{text}");
+    for secret in [
+        UNSAFE,
+        "secret-token",
+        "/private/config",
+        "credential=unsafe",
+    ] {
+        assert!(
+            !text.contains(secret),
+            "diagnostics leaked unsafe boundary data"
+        );
+        assert!(
+            !public.contains(secret),
+            "public result leaked unsafe boundary data"
+        );
+    }
+}
+fn public_error(
+    result: Result<serde_json::Value, WipOperationError>,
+    code: Option<ProtocolErrorCode>,
+) -> String {
+    match (result, code) {
+        (Err(WipOperationError::OutcomeUnknown(message)), None) => {
+            assert!(message.contains("no automatic retry"));
+            message
+        }
+        (Err(WipOperationError::Protocol(error)), Some(expected)) => {
+            assert_eq!(error.code, expected);
+            error.message
+        }
+        _ => panic!("unexpected public error classification"),
+    }
+}
+
+#[tokio::test]
+async fn production_reply_failures_retain_reasons_without_retrying_mutations() {
+    let cases = [
+        (
+            Fault::Unavailable,
+            Reason::TransportUnavailable,
+            Stage::Request,
+            ProtocolErrorCode::Internal,
+            false,
+        ),
+        (
+            Fault::InvalidPath,
+            Reason::InvalidPath,
+            Stage::Request,
+            ProtocolErrorCode::Internal,
+            false,
+        ),
+        (
+            Fault::Communication,
+            Reason::Communication,
+            Stage::Request,
+            ProtocolErrorCode::Internal,
+            true,
+        ),
+        (
+            Fault::Decode,
+            Reason::Decode,
+            Stage::Response,
+            ProtocolErrorCode::Internal,
+            true,
+        ),
+        (
+            Fault::Size,
+            Reason::ResponseSize,
+            Stage::Response,
+            ProtocolErrorCode::ResourceLimitExceeded,
+            true,
+        ),
+        (
+            Fault::Unknown,
+            Reason::BackendUnknown,
+            Stage::Response,
+            ProtocolErrorCode::Internal,
+            true,
+        ),
+        (
+            Fault::Unrecognized,
+            Reason::UnrecognizedResponse,
+            Stage::Response,
+            ProtocolErrorCode::Internal,
+            true,
+        ),
+        (
+            Fault::Panic,
+            Reason::TaskJoin,
+            Stage::Completion,
+            ProtocolErrorCode::Internal,
+            true,
+        ),
+    ];
+    for (fault, reason, stage, read_code, applies) in cases {
+        for mutation in [false, true] {
+            let client = Client::new(fault);
+            let backend = WorkspaceConfigBackend::new(client.clone());
+            let capture = Capture::default();
+            let result = backend
+                .post::<serde_json::Value, _>(
+                    if mutation { "/commit" } else { "/read" },
+                    &serde_json::json!({"content": UNSAFE}),
+                    mutation,
+                )
+                .with_subscriber(capture.dispatch())
+                .await;
+            let code = if (mutation && !matches!(fault, Fault::InvalidPath))
+                || matches!(fault, Fault::Unknown)
+            {
+                None
+            } else {
+                Some(read_code)
+            };
+            let public = public_error(result, code);
+            assert_safe(&capture, reason, stage, &public);
+            let state = client.state.lock().unwrap();
+            assert_eq!(state.requests, 1);
+            assert_eq!(state.mutations, usize::from(mutation && applies));
+            assert_eq!(state.saved.is_some(), mutation && applies);
+        }
+    }
+}
+
+#[tokio::test]
+async fn production_typed_rejections_preserve_public_codes_and_never_apply() {
+    for (status, code) in [
+        (403, ProtocolErrorCode::PermissionDenied),
+        (409, ProtocolErrorCode::ValidatorMismatch),
+        (413, ProtocolErrorCode::ResourceLimitExceeded),
+        (422, ProtocolErrorCode::InvalidArguments),
+        (500, ProtocolErrorCode::Internal),
+    ] {
+        for mutation in [false, true] {
+            let client = Client::new(Fault::Rejected(status));
+            let backend = WorkspaceConfigBackend::new(client.clone());
+            let capture = Capture::default();
+            let public = public_error(
+                backend
+                    .post::<serde_json::Value, _>(
+                        if mutation { "/commit" } else { "/read" },
+                        &(),
+                        mutation,
+                    )
+                    .with_subscriber(capture.dispatch())
+                    .await,
+                Some(code),
+            );
+            assert_safe(&capture, Reason::BackendRejected, Stage::Response, &public);
+            assert!(capture.text().contains(&format!("status={status}")));
+            let state = client.state.lock().unwrap();
+            assert_eq!(state.requests, 1);
+            assert_eq!(state.mutations, 0);
+            assert!(state.saved.is_none());
+        }
+    }
+}
+
+#[tokio::test]
+async fn production_deadline_is_unknown_after_effect_and_blocking_completion_is_not_retried() {
+    for mutation in [false, true] {
+        let client = Client::new(Fault::Deadline);
+        let release = Release(client.clone());
+        let backend =
+            WorkspaceConfigBackend::new(client.clone()).with_deadline(Duration::from_secs(1));
+        let capture = Capture::default();
+        let body = serde_json::json!({"content": UNSAFE});
+        let request = backend
+            .post::<serde_json::Value, _>(
+                if mutation { "/commit" } else { "/read" },
+                &body,
+                mutation,
+            )
+            .with_subscriber(capture.dispatch());
+        tokio::pin!(request);
+        tokio::select! {
+            biased;
+            _ = client.started.notified() => {},
+            _ = &mut request => panic!("request must start and apply its effect before timeout is observed"),
+        }
+        {
+            let state = client.state.lock().unwrap();
+            assert_eq!(state.mutations, usize::from(mutation));
+            assert_eq!(state.saved.is_some(), mutation);
+            assert!(!state.completed && !state.released);
+        }
+        let result = request.await;
+        let public = public_error(
+            result,
+            if mutation {
+                None
+            } else {
+                Some(ProtocolErrorCode::ResourceLimitExceeded)
+            },
+        );
+        assert_safe(&capture, Reason::Deadline, Stage::Completion, &public);
+        drop(release);
+        let completed = client.clone();
+        tokio::task::spawn_blocking(move || completed.wait_completed())
+            .await
+            .unwrap();
+        let state = client.state.lock().unwrap();
+        assert_eq!(state.requests, 1);
+        assert_eq!(state.mutations, usize::from(mutation));
+        assert_eq!(state.saved.is_some(), mutation);
+    }
+}
+
+#[tokio::test]
+async fn production_current_identity_and_task_loss_remain_safe() {
+    for (fault, reason, stage, code) in [
+        (
+            Fault::Identity,
+            Reason::Identity,
+            Stage::Response,
+            ProtocolErrorCode::PermissionDenied,
+        ),
+        (
+            Fault::Panic,
+            Reason::TaskJoin,
+            Stage::Completion,
+            ProtocolErrorCode::Internal,
+        ),
+        (
+            Fault::Decode,
+            Reason::Decode,
+            Stage::Response,
+            ProtocolErrorCode::Internal,
+        ),
+    ] {
+        let client = Client::new(fault);
+        let backend = WorkspaceConfigBackend::new(client.clone());
+        let capture = Capture::default();
+        let result = backend
+            .current_async()
+            .with_subscriber(capture.dispatch())
+            .await;
+        let message = match result {
+            Err(WipOperationError::Protocol(error)) => {
+                assert_eq!(error.code, code);
+                error.message
+            }
+            _ => panic!("unexpected current classification"),
+        };
+        assert_safe(&capture, reason, stage, &message);
+        assert_eq!(client.state.lock().unwrap().mutations, 0);
+    }
+}
+
+#[tokio::test]
+async fn production_attach_conversion_retains_original_reason_and_effect_count() {
+    for (fault, reason, expected, mutations) in [
+        (
+            Fault::Communication,
+            Reason::Communication,
+            ConfigAttachError::OutcomeUnknown,
+            1,
+        ),
+        (
+            Fault::Identity,
+            Reason::Identity,
+            ConfigAttachError::OutcomeUnknown,
+            1,
+        ),
+        (
+            Fault::Rejected(403),
+            Reason::BackendRejected,
+            ConfigAttachError::Denied,
+            0,
+        ),
+        (
+            Fault::Rejected(422),
+            Reason::BackendRejected,
+            ConfigAttachError::InvalidRequest,
+            0,
+        ),
+        (
+            Fault::Rejected(500),
+            Reason::BackendRejected,
+            ConfigAttachError::Unavailable,
+            0,
+        ),
+    ] {
+        let client = Client::new(fault);
+        let backend = WorkspaceConfigBackend::new(client.clone());
+        let capture = Capture::default();
+        assert_eq!(
+            backend
+                .attach("host-key-not-logged", None)
+                .with_subscriber(capture.dispatch())
+                .await,
+            Err(expected)
+        );
+        assert_safe(&capture, reason, Stage::Attach, expected.message());
+        let text = capture.text();
+        if !matches!(fault, Fault::Identity) {
+            assert!(text.contains("reason=AttachConversion"), "{text}");
+        }
+        assert!(!text.contains("host-key-not-logged"));
+        let correlations: std::collections::BTreeSet<_> = text
+            .lines()
+            .filter_map(|line| line.split("correlation=").nth(1))
+            .filter_map(|tail| tail.split(';').next())
+            .collect();
+        assert_eq!(
+            correlations.len(),
+            1,
+            "request and attach conversion must correlate"
+        );
+        let state = client.state.lock().unwrap();
+        assert_eq!(state.requests, 1);
+        assert_eq!(state.mutations, mutations);
+        assert_eq!(state.saved.is_some(), mutations == 1);
+    }
+}
+
+#[tokio::test]
+async fn production_oversized_request_is_rejected_before_dispatch() {
+    let client = Client::new(Fault::Communication);
+    let backend = WorkspaceConfigBackend::new(client.clone());
+    let capture = Capture::default();
+    let public = public_error(
+        backend
+            .post::<serde_json::Value, _>("/commit", &"x".repeat(MAX_RESPONSE_BYTES), true)
+            .with_subscriber(capture.dispatch())
+            .await,
+        Some(ProtocolErrorCode::ResourceLimitExceeded),
+    );
+    assert_safe(&capture, Reason::RequestSize, Stage::Request, &public);
+    let state = client.state.lock().unwrap();
+    assert_eq!(state.requests, 0);
+    assert_eq!(state.mutations, 0);
+}
+
+#[tokio::test]
+async fn production_current_deadline_remains_bounded_and_has_no_mutation() {
+    let client = Client::new(Fault::Deadline);
+    let release = Release(client.clone());
+    let backend = WorkspaceConfigBackend::new(client.clone()).with_deadline(Duration::from_secs(1));
+    let capture = Capture::default();
+    let request = backend.current_async().with_subscriber(capture.dispatch());
+    tokio::pin!(request);
+    tokio::select! {
+        biased;
+        _ = client.started.notified() => {},
+        _ = &mut request => panic!("current request must enter the gate before timeout is observed"),
+    }
+    assert_eq!(client.state.lock().unwrap().mutations, 0);
+    let error = request.await;
+    let message = match error {
+        Err(WipOperationError::Protocol(error)) => {
+            assert_eq!(error.code, ProtocolErrorCode::ResourceLimitExceeded);
+            error.message
+        }
+        _ => panic!("current deadline must remain a public resource-limit error"),
+    };
+    assert_safe(&capture, Reason::Deadline, Stage::Completion, &message);
+    drop(release);
+    let completed = client.clone();
+    tokio::task::spawn_blocking(move || completed.wait_completed())
+        .await
+        .unwrap();
+    let state = client.state.lock().unwrap();
+    assert_eq!(state.requests, 1);
+    assert_eq!(state.mutations, 0);
+}
+
+#[tokio::test]
+async fn production_request_encoding_failure_is_safe_and_never_dispatched() {
+    struct Unencodable;
+    impl Serialize for Unencodable {
+        fn serialize<S: serde::Serializer>(&self, _: S) -> Result<S::Ok, S::Error> {
+            Err(serde::ser::Error::custom(UNSAFE))
+        }
+    }
+    let client = Client::new(Fault::Communication);
+    let backend = WorkspaceConfigBackend::new(client.clone());
+    let capture = Capture::default();
+    let result = backend
+        .post::<serde_json::Value, _>("/commit", &Unencodable, true)
+        .with_subscriber(capture.dispatch())
+        .await;
+    let message = public_error(result, Some(ProtocolErrorCode::InvalidArguments));
+    assert_safe(&capture, Reason::Decode, Stage::Request, &message);
+    let state = client.state.lock().unwrap();
+    assert_eq!(state.requests, 0);
+    assert_eq!(state.mutations, 0);
+}

crates/worker/src/feature/builtin/workspace_config/content.rs

@@ -36,9 +36,11 @@ use wip_protocol::{
 };
 
 use super::backend::{
-    MAX_DEPTH, MAX_NODES, MAX_PATH_BYTES, MAX_TEXT_BYTES, WorkspaceConfigBackend, failure, unknown,
+    MAX_DEPTH, MAX_NODES, MAX_PATH_BYTES, MAX_TEXT_BYTES, WorkspaceConfigBackend, diagnostic,
+    failure, unknown,
 };
 use super::{CONTENT_ROOT, WorkspaceConfigFeature};
+use crate::feature::builtin::backend_diagnostic::{Diagnostic, Reason, Stage};
 use crate::wip::{
     WipCallContext, WipOperationError, WipOperationHandler, WipOperationOutput, WipProjection,
     WipProjectionKind, WipSubtreeProvider, contextual_reference, json_to_wip,
@@ -138,6 +140,23 @@ async fn observe(
     connection: &str,
     paths: Vec<String>,
     depth: u32,
+) -> Result<WorkspaceConfigObserveResponse, WipOperationError> {
+    observe_diagnosed(
+        backend,
+        connection,
+        paths,
+        depth,
+        diagnostic("observe", false),
+    )
+    .await
+}
+
+async fn observe_diagnosed(
+    backend: &WorkspaceConfigBackend,
+    connection: &str,
+    paths: Vec<String>,
+    depth: u32,
+    diagnostic: Diagnostic,
 ) -> Result<WorkspaceConfigObserveResponse, WipOperationError> {
     if paths.len() > MAX_NODES || depth > MAX_DEPTH {
         return Err(invalid());
@@ -148,7 +167,7 @@ async fn observe(
         }
     }
     let result: WorkspaceConfigObserveResponse = backend
-        .post(
+        .post_diagnosed(
             "/observe",
             &WorkspaceConfigObserveRequest {
                 connection_id: connection.into(),
@@ -156,6 +175,7 @@ async fn observe(
                 depth,
             },
             false,
+            diagnostic.clone(),
         )
         .await?;
     if result.connection_id != connection
@@ -164,24 +184,37 @@ async fn observe(
         || result.validator.len() > 4096
         || result.digest.is_empty()
     {
+        diagnostic.record(
+            Stage::Response,
+            if result.connection_id != connection {
+                Reason::Identity
+            } else {
+                Reason::InvalidMetadata
+            },
+            None,
+        );
         return Err(failure(
             ProtocolErrorCode::Internal,
             "Invalid Workspace config metadata",
         ));
     }
     if result.entrypoints.is_empty() || result.entrypoints.len() > MAX_NODES {
+        diagnostic.record(Stage::Response, Reason::InvalidMetadata, None);
         return Err(failure(
             ProtocolErrorCode::Internal,
             "Invalid Workspace config entrypoints",
         ));
     }
     for path in &result.entrypoints {
-        validate_source(path)?;
+        validate_source(path)
+            .inspect_err(|_| diagnostic.record(Stage::Response, Reason::InvalidMetadata, None))?;
     }
     let mut seen = BTreeSet::new();
     for node in &result.nodes {
         if !node.path.is_empty() {
-            validate_source(&node.path)?;
+            validate_source(&node.path).inspect_err(|_| {
+                diagnostic.record(Stage::Response, Reason::InvalidMetadata, None)
+            })?;
         }
         if !seen.insert(&node.path)
             || node.validator.is_empty()
@@ -198,6 +231,7 @@ async fn observe(
                 suffix.is_some_and(|s| s.split('/').count() <= depth as usize)
             })
         {
+            diagnostic.record(Stage::Response, Reason::InvalidMetadata, None);
             return Err(failure(
                 ProtocolErrorCode::Internal,
                 "Invalid Workspace config metadata scope",
@@ -490,9 +524,10 @@ fn content_type(value: Option<&Value>) -> Result<ConfigContentType, WipOperation
 
 impl ConfigHandler {
     async fn read(&self) -> Result<WorkspaceConfigReadResponse, WipOperationError> {
+        let diagnostic = diagnostic("read", false);
         let response: WorkspaceConfigReadResponse = self
             .backend
-            .post(
+            .post_diagnosed(
                 "/read",
                 &WorkspaceConfigReadRequest {
                     connection_id: self.attachment.connection_id.clone(),
@@ -500,6 +535,7 @@ impl ConfigHandler {
                     validator: self.node.validator.clone(),
                 },
                 false,
+                diagnostic.clone(),
             )
             .await?;
         if response.path != self.node.path
@@ -507,12 +543,14 @@ impl ConfigHandler {
             || Some(&response.digest) != self.node.digest.as_ref()
             || Some(&response.content_type) != self.node.content_type.as_ref()
         {
+            diagnostic.record(Stage::Response, Reason::Identity, None);
             return Err(failure(
                 ProtocolErrorCode::Internal,
                 "Invalid Workspace config read response",
             ));
         }
         if response.content.len() > MAX_TEXT_BYTES {
+            diagnostic.record(Stage::Response, Reason::ResponseSize, None);
             return Err(failure(
                 ProtocolErrorCode::ResourceLimitExceeded,
                 "Workspace config source exceeds text limit",
@@ -630,12 +668,17 @@ impl WipOperationHandler for ConfigHandler {
         context: WipCallContext,
     ) -> Result<WipOperationOutput, WipOperationError> {
         let dispatched = std::sync::atomic::AtomicBool::new(false);
+        let diagnostic = diagnostic(
+            "operation",
+            !matches!(operation, "read" | "read_lines" | "search_text"),
+        );
         tokio::time::timeout(
             super::DEADLINE,
-            self.execute(operation, arguments, context, &dispatched),
+            self.execute(operation, arguments, context, &dispatched, &diagnostic),
         )
         .await
         .map_err(|_| {
+            diagnostic.record(Stage::Completion, Reason::Deadline, None);
             if dispatched.load(std::sync::atomic::Ordering::Acquire) {
                 unknown()
             } else {
@@ -655,6 +698,7 @@ impl ConfigHandler {
         arguments: &BTreeMap<String, Value>,
         context: WipCallContext,
         dispatched: &std::sync::atomic::AtomicBool,
+        operation_diagnostic: &Diagnostic,
     ) -> Result<WipOperationOutput, WipOperationError> {
         if !self.operations.iter().any(|name| name == operation) {
             return Err(failure(
@@ -798,9 +842,10 @@ impl ConfigHandler {
         };
         let changes = self.canonical_changes(changes).await?;
         dispatched.store(true, std::sync::atomic::Ordering::Release);
+        let diagnostic = operation_diagnostic.clone();
         let result: WorkspaceConfigCommitResponse = self
             .backend
-            .post(
+            .post_diagnosed(
                 "/commit",
                 &WorkspaceConfigCommitRequest {
                     connection_id: self.attachment.connection_id.clone(),
@@ -812,28 +857,37 @@ impl ConfigHandler {
                     },
                 },
                 true,
+                diagnostic.clone(),
             )
             .await?;
         // Commit response supplies ROOT state. Obtain the exact post-commit PATH
         // validator, never substitute the root validator for a file validator.
-        let updated = observe(
+        let updated = observe_diagnosed(
             &self.backend,
             &self.attachment.connection_id,
             vec![self.node.path.clone()],
             0,
+            diagnostic.at_stage(Stage::Reconcile),
         )
         .await
         .map_err(|_| unknown())?;
         if updated.digest != result.digest || updated.validator != result.validator {
+            diagnostic.record(Stage::Reconcile, Reason::Identity, None);
             return Err(unknown());
         }
         let node = updated
             .nodes
             .iter()
             .find(|n| n.path == self.node.path)
-            .ok_or_else(unknown)?;
+            .ok_or_else(|| {
+                diagnostic.record(Stage::Reconcile, Reason::InvalidMetadata, None);
+                unknown()
+            })?;
         let value = json_to_wip(&json!({"digest": result.digest,
-            "activation": "Saved to canonical Workspace configuration; existing Worker manifests are not rewritten"})).map_err(|_| unknown())?;
+            "activation": "Saved to canonical Workspace configuration; existing Worker manifests are not rewritten"})).map_err(|_| {
+                diagnostic.record(Stage::Completion, Reason::Decode, None);
+                unknown()
+            })?;
         Ok(WipOperationOutput::native_with_validator(
             value,
             object_validator(

crates/worker/src/feature/builtin/workspace_config/production_tests.rs

@@ -1,8 +1,10 @@
-//! Worker-side E2E across real Feature invocation, WIP Client/Host and typed
-//! Workspace transport. Canonical persistence/projection tests live in Backend;
-//! this deterministic router models its published auth/CAS/response contract.
+//! Worker-side integration across real Feature invocation, WIP Client/Host and the production
+//! WorkspaceConfigBackend. Only WorkspaceClient transport is substituted by a deterministic router
+//! retaining applied effects and modeling the published auth/CAS/response contract. These ordinary
+//! worker lib tests do not replace canonical persistence/projection tests owned by Backend.
 use super::*;
 use crate::feature::FeatureRegistryBuilder;
+use crate::feature::builtin::backend_diagnostic::tests::Capture;
 use crate::hook::HookRegistryBuilder;
 use crate::wip::{WipMountRegistry, WipRuntime};
 use crate::worker::{
@@ -16,6 +18,7 @@ use serde_json::{Value as Json, json};
 use server_api::*;
 use std::collections::{BTreeMap, BTreeSet};
 use std::sync::{Arc, Mutex};
+use tracing::instrument::WithSubscriber;
 
 #[derive(Debug)]
 struct State {
@@ -26,6 +29,10 @@ struct State {
     entries: BTreeMap<String, (ConfigContentType, String)>,
     commits: Vec<WorkspaceConfigCommitRequest>,
     requests: Vec<WorkspaceRequest>,
+    applied_commits: usize,
+    malformed_read_reply: Option<&'static str>,
+    malformed_observe_reply: Option<&'static str>,
+    lose_postcommit_observe_reply: bool,
     fail_commit: bool,
     fail_current: bool,
     lose_commit_reply: bool,
@@ -58,6 +65,10 @@ impl Router {
                 ]),
                 commits: Vec::new(),
                 requests: Vec::new(),
+                applied_commits: 0,
+                malformed_read_reply: None,
+                malformed_observe_reply: None,
+                lose_postcommit_observe_reply: false,
                 fail_commit: false,
                 fail_current: false,
                 lose_commit_reply: false,
@@ -222,6 +233,11 @@ impl WorkspaceClient for Router {
         }
         match suffix {
             "/observe" => {
+                if state.lose_postcommit_observe_reply && state.applied_commits > 0 {
+                    return Err(WorkspaceClientError::Request(
+                        "secret-token /private/host/config".into(),
+                    ));
+                }
                 let request: WorkspaceConfigObserveRequest = serde_json::from_value(input).unwrap();
                 assert!(request.depth <= 8 && request.paths.len() <= 256);
                 let mut kinds =
@@ -268,13 +284,23 @@ impl WorkspaceClient for Router {
                         )
                     })
                     .collect();
-                Ok(response(&WorkspaceConfigObserveResponse {
+                let mut result = WorkspaceConfigObserveResponse {
                     connection_id: state.active.as_ref().unwrap().connection_id.clone(),
                     validator: state.validator(""),
                     digest: state.tree_digest(),
                     entrypoints: vec!["main.dcdl".into()],
                     nodes,
-                }))
+                };
+                match state.malformed_observe_reply {
+                    Some("connection") => {
+                        result.connection_id = "secret-token /private/host/config".into()
+                    }
+                    Some("entrypoints") => result.entrypoints.clear(),
+                    Some("scope") => result.nodes[0].path = "outside/path.dcdl".into(),
+                    None => {}
+                    Some(_) => unreachable!(),
+                }
+                Ok(response(&result))
             }
             "/read" => {
                 let request: WorkspaceConfigReadRequest = serde_json::from_value(input).unwrap();
@@ -290,13 +316,20 @@ impl WorkspaceClient for Router {
                         WorkspaceConfigFailureClassification::NotCommitted,
                     ));
                 };
-                Ok(response(&WorkspaceConfigReadResponse {
+                let mut result = WorkspaceConfigReadResponse {
                     path: request.path.clone(),
                     content: content.clone(),
                     content_type: *content_type,
                     digest: digest(content),
                     validator: state.validator(&request.path),
-                }))
+                };
+                match state.malformed_read_reply {
+                    Some("identity") => result.path = "secret-token /private/host/config".into(),
+                    Some("size") => result.content = "x".repeat(256 * 1024 + 1),
+                    None => {}
+                    Some(_) => unreachable!(),
+                }
+                Ok(response(&result))
             }
             "/commit" => {
                 let request: WorkspaceConfigCommitRequest = serde_json::from_value(input).unwrap();
@@ -403,6 +436,7 @@ impl WorkspaceClient for Router {
                     ));
                 }
                 state.entries = entries;
+                state.applied_commits += 1;
                 if state.lose_commit_reply {
                     state.lose_commit_reply = false;
                     return Err(WorkspaceClientError::Request(
@@ -797,14 +831,20 @@ async fn post_commit_observation_mismatch_reports_unknown_without_replaying_writ
         let path = "/workspace-config/main.dcdl";
         observe_interface(&runtime, path, true).await;
         client.state.lock().unwrap().mismatched_commit_reply = Some(field);
+        let capture = Capture::default();
         let error = call(&runtime, path, "write", json!({"content":"saved"}))
+            .with_subscriber(capture.dispatch())
             .await
             .unwrap_err()
             .to_string();
         assert!(error.contains("unknown"), "{field}: {error}");
+        assert!(capture.text().contains("stage=Reconcile"));
+        assert!(capture.text().contains("reason=Identity"));
+        assert!(!capture.text().contains("saved"));
         let state = client.state.lock().unwrap();
         assert_eq!(state.commits.len(), 1);
         assert_eq!(state.entries["main.dcdl"].1, "saved");
+        assert_eq!(state.applied_commits, 1);
     }
 }
 
@@ -1155,3 +1195,110 @@ async fn workspace_config_backend_validators_reject_path_and_connection_transpla
     assert!(result.content.unwrap().contains("model = updated"));
     assert_eq!(client.state.lock().unwrap().commits.len(), 1);
 }
+
+#[tokio::test]
+async fn production_read_validation_retains_reason_without_committing() {
+    for (fault, reason) in [("identity", "Identity"), ("size", "ResponseSize")] {
+        let client = Router::new();
+        let (feature, runtime) = runtime(client.clone());
+        feature.attach("initial", None).await.unwrap();
+        let path = "/workspace-config/main.dcdl";
+        observe_interface(&runtime, path, true).await;
+        client.state.lock().unwrap().malformed_read_reply = Some(fault);
+        let capture = Capture::default();
+        let error = call(&runtime, path, "read", json!({}))
+            .with_subscriber(capture.dispatch())
+            .await
+            .unwrap_err()
+            .to_string();
+        let text = capture.text();
+        assert!(text.contains(&format!("reason={reason}")), "{text}");
+        assert!(text.contains("stage=Response"));
+        assert!(!error.contains("unknown"));
+        assert!(!text.contains("secret-token") && !text.contains("/private/host"));
+        assert!(!error.contains("secret-token") && !error.contains("/private/host"));
+        let state = client.state.lock().unwrap();
+        assert_eq!(state.applied_commits, 0);
+        assert_eq!(state.entries["main.dcdl"].1, "model = old");
+    }
+}
+
+#[tokio::test]
+async fn production_observation_validation_retains_reason_without_committing() {
+    for (fault, reason) in [
+        ("connection", "Identity"),
+        ("entrypoints", "InvalidMetadata"),
+        ("scope", "InvalidMetadata"),
+    ] {
+        let client = Router::new();
+        let (feature, runtime) = runtime(client.clone());
+        feature.attach("initial", None).await.unwrap();
+        client.state.lock().unwrap().malformed_observe_reply = Some(fault);
+        let capture = Capture::default();
+        let error = runtime
+            .tree("/workspace-config/main.dcdl".into(), 0, true)
+            .with_subscriber(capture.dispatch())
+            .await
+            .unwrap_err()
+            .to_string();
+        let text = capture.text();
+        assert!(text.contains(&format!("reason={reason}")), "{text}");
+        assert!(text.contains("stage=Response"));
+        assert!(!text.contains("secret-token") && !text.contains("/private/host"));
+        assert!(!error.contains("secret-token") && !error.contains("/private/host"));
+        assert_eq!(client.state.lock().unwrap().applied_commits, 0);
+    }
+}
+
+#[tokio::test]
+async fn production_postcommit_transport_loss_retains_cause_and_unknown_without_retry() {
+    let client = Router::new();
+    let (feature, runtime) = runtime(client.clone());
+    feature.attach("initial", None).await.unwrap();
+    let path = "/workspace-config/main.dcdl";
+    observe_interface(&runtime, path, true).await;
+    client.state.lock().unwrap().lose_postcommit_observe_reply = true;
+    let capture = Capture::default();
+    let error = call(
+        &runtime,
+        path,
+        "write",
+        json!({"content":"saved-before-observation-loss"}),
+    )
+    .with_subscriber(capture.dispatch())
+    .await
+    .unwrap_err()
+    .to_string();
+    assert!(error.contains("unknown"));
+    let text = capture.text();
+    assert!(
+        text.lines().any(
+            |event| event.contains("reason=Communication") && event.contains("stage=Reconcile")
+        ),
+        "{text}"
+    );
+    assert!(
+        !text.contains("reason=UnrecognizedResponse"),
+        "must not invent a response-decode cause for transport loss"
+    );
+    assert!(!text.contains("secret-token") && !text.contains("/private/host"));
+    assert!(!text.contains("saved-before-observation-loss"));
+    assert!(!error.contains("secret-token") && !error.contains("/private/host"));
+    let correlations: BTreeSet<_> = text
+        .lines()
+        .filter_map(|line| line.split("correlation=").nth(1))
+        .filter_map(|tail| tail.split(';').next())
+        .collect();
+    assert_eq!(
+        correlations.len(),
+        1,
+        "lost postcommit observation must correlate to original commit"
+    );
+    let state = client.state.lock().unwrap();
+    assert_eq!(state.commits.len(), 1);
+    assert_eq!(state.applied_commits, 1);
+    assert_eq!(
+        state.entries["main.dcdl"].1,
+        "saved-before-observation-loss"
+    );
+}

crates/worker/src/feature/builtin/workspace_config/wip.rs

@@ -161,7 +161,14 @@ impl WipOperationHandler for AttachHandler {
             "attachment_path": format!("/workdir-attachments/{}", encode_identity(&result.connection_id)),
             "content_path": CONTENT_ROOT,
             "already_attached": result.already_attached,
-        })).map_err(|_| WipOperationError::OutcomeUnknown("Completed config attachment response unavailable".into()))?;
+        })).map_err(|_| {
+            super::backend::diagnostic("attach_output", true).record(
+                crate::feature::builtin::backend_diagnostic::Stage::Attach,
+                crate::feature::builtin::backend_diagnostic::Reason::Decode,
+                None,
+            );
+            WipOperationError::OutcomeUnknown("Completed config attachment response unavailable".into())
+        })?;
         Ok(WipOperationOutput::native(value))
     }
 }