fix(automations): harden idempotency store after review of 774689ada (APPS-248)

- Capability doc comment no longer claims delete; DELETE honors neither
  expected_revision nor Idempotency-Key (docs already said so).
- A failed idempotency_store after the mutation committed is logged and the
  success response is returned instead of a 500.
- Unparseable idempotency.json is renamed to idempotency.corrupt-<ts>.json
  and the request fails with an error instead of silently starting empty.
- Keys are scoped by route and automation id (operation + key).
- The process-wide gate is replaced by a per-scoped-key lock, so unrelated
  keyed mutations no longer wait behind a slow run enqueue.
- Rejected: revision not bumped by adopt_for_run/scheduler writes (documented
  bookkeeping exemption; out of this slice).

Evidence: RUST_MIN_STACK=32MB cargo test -p codewhale-tui --lib automation:
128 passed, 0 failed, 1 ignored. Limits: no new tests per policy, so the
quarantine and scoped-key paths are compiled but not exercised; full gate and
CI not run.

Co-Authored-By: Claude Sonnet 5.5 <[email protected]>

crates/protocol/src/runtime/mod.rs

@@ -74,9 +74,10 @@ pub struct RuntimeCapabilities {
     /// Read-only exact accepted-turn lookup by thread and operation key.
     #[serde(default)]
     pub turn_operation_lookup: bool,
-    /// Automation create/patch/delete/run/pause/resume honor `expected_revision`
-    /// (409 `revision_conflict` carrying `current_revision`) and a bounded,
-    /// persisted `Idempotency-Key`.
+    /// Automation patch/run/pause/resume honor `expected_revision` (409
+    /// `revision_conflict` carrying `current_revision`); create/patch/run/pause/
+    /// resume honor a bounded, persisted `Idempotency-Key`. Delete honors
+    /// neither.
     #[serde(default)]
     pub automation_mutation_preconditions: bool,
     /// Bounded inline image inputs, persisted and replayed with their turn.

crates/tui/src/automation_manager.rs

@@ -1463,7 +1463,22 @@ impl AutomationManager {
     fn read_idempotency_unlocked(&self, now: DateTime<Utc>) -> Result<Vec<IdempotencyEntry>> {
         let path = self.idempotency_path()?;
         let mut entries: Vec<IdempotencyEntry> = match fs::read(&path) {
-            Ok(bytes) => serde_json::from_slice(&bytes).unwrap_or_default(),
+            Ok(bytes) => match serde_json::from_slice(&bytes) {
+                Ok(entries) => entries,
+                Err(parse_error) => {
+                    let quarantine = path.with_file_name(format!(
+                        "idempotency.corrupt-{}.json",
+                        now.format("%Y%m%dT%H%M%S%.3f")
+                    ));
+                    fs::rename(&path, &quarantine)
+                        .with_context(|| format!("quarantine corrupt {}", path.display()))?;
+                    anyhow::bail!(
+                        "idempotency store {} was unparseable ({parse_error}); moved to {}",
+                        path.display(),
+                        quarantine.display()
+                    );
+                }
+            },
             Err(error) if error.kind() == std::io::ErrorKind::NotFound => Vec::new(),
             Err(error) => return Err(error).with_context(|| format!("read {}", path.display())),
         };

crates/tui/src/runtime_api.rs

@@ -6733,7 +6733,20 @@ struct AutomationPreconditionQuery {
     expected_revision: Option<u64>,
 }
 
-static AUTOMATION_IDEMPOTENCY_GATE: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+static AUTOMATION_IDEMPOTENCY_GATES: std::sync::LazyLock<
+    std::sync::Mutex<std::collections::HashMap<String, std::sync::Arc<tokio::sync::Mutex<()>>>>,
+> = std::sync::LazyLock::new(Default::default);
+
+async fn lock_automation_idempotency_key(scoped_key: &str) -> tokio::sync::OwnedMutexGuard<()> {
+    let gate = {
+        let mut gates = AUTOMATION_IDEMPOTENCY_GATES
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        gates.retain(|_, gate| std::sync::Arc::strong_count(gate) > 1);
+        gates.entry(scoped_key.to_string()).or_default().clone()
+    };
+    gate.lock_owned().await
+}
 
 fn automation_idempotency_key(headers: &HeaderMap) -> Result<Option<String>, ApiError> {
     let Some(value) = headers.get("idempotency-key") else {
@@ -6819,7 +6832,8 @@ where
             .map(|b| format!("{b:02x}"))
             .collect::<String>()
     };
-    let _gate = AUTOMATION_IDEMPOTENCY_GATE.lock().await;
+    let key = format!("{operation}\n{key}");
+    let _gate = lock_automation_idempotency_key(&key).await;
     let lookup = state
         .automations
         .lock()
@@ -6845,12 +6859,18 @@ where
     };
     let body = serde_json::to_value(value)
         .map_err(|e| ApiError::internal(format!("Failed to encode response: {e}")))?;
-    state
-        .automations
-        .lock()
-        .await
-        .idempotency_store(&key, &fingerprint, ok_status.as_u16(), body.clone())
-        .map_err(|e| ApiError::internal(format!("Idempotency store unavailable: {e}")))?;
+    if let Err(error) = state.automations.lock().await.idempotency_store(
+        &key,
+        &fingerprint,
+        ok_status.as_u16(),
+        body.clone(),
+    ) {
+        tracing::error!(
+            %error,
+            operation,
+            "automation mutation committed but its idempotency record could not be stored"
+        );
+    }
     Ok(respond(ok_status, body, false))
 }
 

docs/RUNTIME_API.md

@@ -1723,7 +1723,13 @@ relying on the behavior below (older Engines ignore both inputs).
   and identical request returns the stored status and body with
   `Idempotency-Replayed: true` and performs no second create or run. The same
   key with a different request is `422` `idempotency_key_reuse`. Failed
-  attempts are not stored, so they can be retried. `DELETE` is not covered.
+  attempts are not stored, so they can be retried. Keys are scoped to the
+  route and automation id. If the record cannot be persisted after the
+  mutation committed, the success response is still returned (the failure is
+  logged), so a retry of that request is not deduplicated. An unparseable
+  `idempotency.json` is renamed to `idempotency.corrupt-<timestamp>.json` and
+  that request fails with `500`; later requests start from an empty store.
+  `DELETE` is not covered.
 
 **Operate** (always-on named operation; same `OperateRecord` as CWC
 `20de981` / PR #284)