feat(automations): revision precondition and Idempotency-Key on mutations (APPS-248)

AutomationRecord gains a persisted integer revision (1 on create, +1 on
PATCH/pause/resume, legacy records read 0; scheduler bookkeeping does not
bump it). PATCH/run/pause/resume accept expected_revision (body on PATCH,
?expected_revision= on all four) and answer 409 revision_conflict with
current_revision on mismatch, checked inside the state transaction. POST
create/PATCH/run/pause/resume accept Idempotency-Key: first 2xx response is
persisted in idempotency.json (1024 entries, 24h TTL); an identical retry
replays it with Idempotency-Replayed: true, a different request on the same
key is 422 idempotency_key_reuse. Advertised as
capabilities.automation_mutation_preconditions on /v1/runtime/info. Absent
inputs behave as before. docs/RUNTIME_API.md updated. DELETE is not covered.

Evidence: cargo test -p codewhale-tui --lib automation (RUST_MIN_STACK=32MB):
128 passed, 0 failed, 1 ignored; protocol runtime_capabilities test passed.
One existing editor test updated to ignore the now-changing revision.
Limits: no new tests per repo policy, so the 409/replay/422 paths are
compiled but not exercised; no live HTTP run; full gate not run; the
automation filter overflows the default test-thread stack in one existing
test (pre-existing, unrelated to this slice).

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

crates/protocol/src/runtime/mod.rs

@@ -74,6 +74,11 @@ 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`.
+    #[serde(default)]
+    pub automation_mutation_preconditions: bool,
     /// Bounded inline image inputs, persisted and replayed with their turn.
     #[serde(default)]
     pub turn_image_inputs: bool,
@@ -443,6 +448,7 @@ mod tests {
             turns: true,
             turn_operation_idempotency: true,
             turn_operation_lookup: true,
+            automation_mutation_preconditions: true,
             turn_image_inputs: true,
             turn_steer: true,
             turn_interrupt: true,

crates/tui/src/automation_manager.rs

@@ -197,6 +197,9 @@ pub struct AutomationRecord {
     /// Bound by the trusted service, independently of visibility ownership.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub execution_scope: Option<String>,
+    /// Increments on every definition mutation; absent on legacy records (0).
+    #[serde(default)]
+    pub revision: u64,
 }
 
 impl AutomationRecord {
@@ -318,6 +321,54 @@ pub struct UpdateAutomationRequest {
     pub auto_approve: Option<bool>,
     pub delivery_mode: Option<AutomationDeliveryMode>,
     pub status: Option<AutomationStatus>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub expected_revision: Option<u64>,
+}
+
+#[derive(Debug)]
+pub struct RevisionConflict {
+    pub current: u64,
+}
+
+impl std::fmt::Display for RevisionConflict {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "Automation revision conflict: current revision is {}",
+            self.current
+        )
+    }
+}
+
+impl std::error::Error for RevisionConflict {}
+
+fn check_expected_revision(record: &AutomationRecord, expected: Option<u64>) -> Result<()> {
+    match expected {
+        Some(expected) if expected != record.revision => Err(RevisionConflict {
+            current: record.revision,
+        }
+        .into()),
+        _ => Ok(()),
+    }
+}
+
+pub const IDEMPOTENCY_TTL_HOURS: i64 = 24;
+pub const IDEMPOTENCY_MAX_ENTRIES: usize = 1024;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct IdempotencyEntry {
+    key: String,
+    fingerprint: String,
+    status: u16,
+    body: serde_json::Value,
+    created_at: DateTime<Utc>,
+}
+
+#[derive(Debug)]
+pub enum IdempotencyLookup {
+    Miss,
+    Replay { status: u16, body: serde_json::Value },
+    KeyReused,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -1254,6 +1305,7 @@ impl AutomationManager {
             updated_at: now,
             next_run_at,
             last_run_at: None,
+            revision: 1,
         };
 
         self.save_automation(&record)?;
@@ -1316,6 +1368,7 @@ impl AutomationManager {
         req: UpdateAutomationRequest,
     ) -> Result<AutomationRecord> {
         let mut existing = self.get_automation(id)?;
+        check_expected_revision(&existing, req.expected_revision)?;
         let adopting = existing.execution_scope.is_none()
             && self.execution_scope.is_some()
             && req.status != Some(AutomationStatus::Paused);
@@ -1391,10 +1444,71 @@ impl AutomationManager {
         }
 
         existing.updated_at = Utc::now();
+        existing.revision = existing.revision.saturating_add(1);
         self.save_automation_unlocked(&existing)?;
         Ok(existing)
     }
 
+    fn idempotency_path(&self) -> Result<PathBuf> {
+        Ok(self
+            .automations_dir
+            .parent()
+            .context("automation root")?
+            .join("idempotency.json"))
+    }
+
+    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(),
+            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Vec::new(),
+            Err(error) => return Err(error).with_context(|| format!("read {}", path.display())),
+        };
+        let cutoff = now - Duration::hours(IDEMPOTENCY_TTL_HOURS);
+        entries.retain(|entry| entry.created_at > cutoff);
+        Ok(entries)
+    }
+
+    pub fn idempotency_lookup(&self, key: &str, fingerprint: &str) -> Result<IdempotencyLookup> {
+        self.with_transaction(|| {
+            let entries = self.read_idempotency_unlocked(Utc::now())?;
+            Ok(match entries.into_iter().find(|entry| entry.key == key) {
+                None => IdempotencyLookup::Miss,
+                Some(entry) if entry.fingerprint == fingerprint => IdempotencyLookup::Replay {
+                    status: entry.status,
+                    body: entry.body,
+                },
+                Some(_) => IdempotencyLookup::KeyReused,
+            })
+        })
+    }
+
+    pub fn idempotency_store(
+        &self,
+        key: &str,
+        fingerprint: &str,
+        status: u16,
+        body: serde_json::Value,
+    ) -> Result<()> {
+        self.with_transaction(|| {
+            let now = Utc::now();
+            let mut entries = self.read_idempotency_unlocked(now)?;
+            entries.retain(|entry| entry.key != key);
+            entries.push(IdempotencyEntry {
+                key: key.to_string(),
+                fingerprint: fingerprint.to_string(),
+                status,
+                body,
+                created_at: now,
+            });
+            if entries.len() > IDEMPOTENCY_MAX_ENTRIES {
+                let excess = entries.len() - IDEMPOTENCY_MAX_ENTRIES;
+                entries.drain(..excess);
+            }
+            write_json_atomic(&self.idempotency_path()?, &entries)
+        })
+    }
+
     pub fn pause_automation(&self, id: &str) -> Result<AutomationRecord> {
         self.update_automation(
             id,
@@ -2286,13 +2400,23 @@ pub async fn run_now_shared(
     automations: &SharedAutomationManager,
     automation_id: &str,
     task_manager: &SharedTaskManager,
+) -> Result<AutomationRunRecord> {
+    run_now_shared_if(automations, automation_id, None, task_manager).await
+}
+
+pub async fn run_now_shared_if(
+    automations: &SharedAutomationManager,
+    automation_id: &str,
+    expected_revision: Option<u64>,
+    task_manager: &SharedTaskManager,
 ) -> Result<AutomationRunRecord> {
     automations.lock().await.bind_task_manager(task_manager)?;
     let task_manager = Arc::clone(task_manager);
     let task_data_dir = task_manager.data_dir();
-    run_now_with(
+    run_now_inner(
         automations,
         automation_id,
+        expected_revision,
         &task_data_dir,
         move |_, mut run| async move {
             enqueue_run_task(&mut run, &task_manager).await;
@@ -2302,13 +2426,27 @@ pub async fn run_now_shared(
     .await
 }
 
-/// Keep the process-wide dispatch claim over the await, never the manager mutex.
 async fn run_now_with<F, Fut>(
     automations: &SharedAutomationManager,
     automation_id: &str,
     task_data_dir: &Path,
     enqueue: F,
 ) -> Result<AutomationRunRecord>
+where
+    F: FnOnce(AutomationRecord, AutomationRunRecord) -> Fut,
+    Fut: Future<Output = AutomationRunRecord>,
+{
+    run_now_inner(automations, automation_id, None, task_data_dir, enqueue).await
+}
+
+/// Keep the process-wide dispatch claim over the await, never the manager mutex.
+async fn run_now_inner<F, Fut>(
+    automations: &SharedAutomationManager,
+    automation_id: &str,
+    expected_revision: Option<u64>,
+    task_data_dir: &Path,
+    enqueue: F,
+) -> Result<AutomationRunRecord>
 where
     F: FnOnce(AutomationRecord, AutomationRunRecord) -> Fut,
     Fut: Future<Output = AutomationRunRecord>,
@@ -2321,6 +2459,7 @@ where
         let manager = automations.lock().await;
         manager.with_transaction(|| {
             let mut automation = manager.get_automation(automation_id)?;
+            check_expected_revision(&automation, expected_revision)?;
             manager.adopt_for_run(&mut automation)?;
             let now = Utc::now();
             let mut run = new_run_record(&automation.id, now, now);
@@ -3903,6 +4042,7 @@ mod tests {
             updated_at: now,
             next_run_at: None,
             last_run_at: None,
+            revision: 0,
         }
     }
 

crates/tui/src/operate.rs

@@ -1298,6 +1298,7 @@ pub(crate) fn upsert_keepalive(
             updated_at: now,
             next_run_at: None,
             last_run_at: None,
+            revision: 0,
         });
         record.schema_version = crate::automation_manager::CURRENT_AUTOMATION_SCHEMA_VERSION;
         if record.execution_scope.is_none() {

crates/tui/src/runtime_api.rs

@@ -1242,6 +1242,7 @@ fn default_runtime_capabilities() -> RuntimeCapabilities {
         turns: true,
         turn_operation_idempotency: true,
         turn_operation_lookup: true,
+        automation_mutation_preconditions: true,
         turn_image_inputs: true,
         turn_output_token_limit: true,
         profile_constitution: true,
@@ -6719,15 +6720,144 @@ async fn list_automations(
     Ok(Json(automations))
 }
 
+#[derive(Debug, Deserialize, Default)]
+struct AutomationPreconditionQuery {
+    #[serde(default)]
+    expected_revision: Option<u64>,
+}
+
+static AUTOMATION_IDEMPOTENCY_GATE: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+
+fn automation_idempotency_key(headers: &HeaderMap) -> Result<Option<String>, ApiError> {
+    let Some(value) = headers.get("idempotency-key") else {
+        return Ok(None);
+    };
+    let key = value
+        .to_str()
+        .map_err(|_| ApiError::bad_request("Idempotency-Key must be visible ASCII"))?
+        .trim();
+    if key.is_empty() || key.len() > 255 || !key.bytes().all(|b| b.is_ascii_graphic()) {
+        return Err(ApiError::bad_request(
+            "Idempotency-Key must be 1-255 visible ASCII characters",
+        ));
+    }
+    Ok(Some(key.to_string()))
+}
+
+async fn automation_mutation<T, Fut>(
+    state: &RuntimeApiState,
+    headers: &HeaderMap,
+    operation: &str,
+    request: &impl Serialize,
+    ok_status: StatusCode,
+    map_err: fn(anyhow::Error) -> ApiError,
+    run: Fut,
+) -> Result<Response, ApiError>
+where
+    T: Serialize,
+    Fut: std::future::Future<Output = anyhow::Result<T>>,
+{
+    let run = async {
+        match run.await {
+            Ok(value) => Ok(Ok(value)),
+            Err(err) => match err.downcast_ref::<crate::automation_manager::RevisionConflict>() {
+                Some(conflict) => Ok(Err(conflict.current)),
+                None => Err(map_err(err)),
+            },
+        }
+    };
+    let conflict_response = |current: u64| {
+        (
+            StatusCode::CONFLICT,
+            Json(json!({
+                "error": {
+                    "message": format!(
+                        "Automation revision conflict: current revision is {current}"
+                    ),
+                    "status": 409,
+                    "code": "revision_conflict",
+                    "current_revision": current,
+                }
+            })),
+        )
+            .into_response()
+    };
+    let respond = |status: StatusCode, body: serde_json::Value, replayed: bool| {
+        let mut response = (status, Json(body)).into_response();
+        if replayed {
+            response
+                .headers_mut()
+                .insert("idempotency-replayed", HeaderValue::from_static("true"));
+        }
+        response
+    };
+    let Some(key) = automation_idempotency_key(headers)? else {
+        let value = match run.await? {
+            Ok(value) => value,
+            Err(current) => return Ok(conflict_response(current)),
+        };
+        let body = serde_json::to_value(value)
+            .map_err(|e| ApiError::internal(format!("Failed to encode response: {e}")))?;
+        return Ok(respond(ok_status, body, false));
+    };
+    let fingerprint = {
+        use sha2::{Digest as _, Sha256};
+        let mut hasher = Sha256::new();
+        hasher.update(operation.as_bytes());
+        hasher.update([0]);
+        hasher.update(serde_json::to_vec(request).unwrap_or_default());
+        hasher.finalize().iter().map(|b| format!("{b:02x}")).collect::<String>()
+    };
+    let _gate = AUTOMATION_IDEMPOTENCY_GATE.lock().await;
+    let lookup = state
+        .automations
+        .lock()
+        .await
+        .idempotency_lookup(&key, &fingerprint)
+        .map_err(|e| ApiError::internal(format!("Idempotency store unavailable: {e}")))?;
+    match lookup {
+        crate::automation_manager::IdempotencyLookup::Replay { status, body } => {
+            let status = StatusCode::from_u16(status).unwrap_or(ok_status);
+            return Ok(respond(status, body, true));
+        }
+        crate::automation_manager::IdempotencyLookup::KeyReused => {
+            return Err(ApiError::unprocessable(
+                "Idempotency-Key was already used for a different request",
+            )
+            .with_code("idempotency_key_reuse"));
+        }
+        crate::automation_manager::IdempotencyLookup::Miss => {}
+    }
+    let value = match run.await? {
+        Ok(value) => value,
+        Err(current) => return Ok(conflict_response(current)),
+    };
+    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}")))?;
+    Ok(respond(ok_status, body, false))
+}
+
 async fn create_automation(
     State(state): State<RuntimeApiState>,
+    headers: HeaderMap,
     Json(req): Json<CreateAutomationRequest>,
-) -> Result<(StatusCode, Json<AutomationRecord>), ApiError> {
-    let manager = state.automations.lock().await;
-    let automation = manager
-        .create_automation(req)
-        .map_err(|e| ApiError::bad_request(e.to_string()))?;
-    Ok((StatusCode::CREATED, Json(automation)))
+) -> Result<Response, ApiError> {
+    automation_mutation(
+        &state,
+        &headers,
+        "POST /v1/automations",
+        &req,
+        StatusCode::CREATED,
+        |e| ApiError::bad_request(e.to_string()),
+        async { state.automations.lock().await.create_automation(req.clone()) },
+    )
+    .await
 }
 
 async fn get_automation(
@@ -6742,13 +6872,21 @@ async fn get_automation(
 async fn update_automation(
     State(state): State<RuntimeApiState>,
     Path(id): Path<String>,
-    Json(req): Json<UpdateAutomationRequest>,
-) -> Result<Json<AutomationRecord>, ApiError> {
-    let manager = state.automations.lock().await;
-    let automation = manager
-        .update_automation(&id, req)
-        .map_err(map_automation_err)?;
-    Ok(Json(automation))
+    Query(query): Query<AutomationPreconditionQuery>,
+    headers: HeaderMap,
+    Json(mut req): Json<UpdateAutomationRequest>,
+) -> Result<Response, ApiError> {
+    req.expected_revision = req.expected_revision.or(query.expected_revision);
+    automation_mutation(
+        &state,
+        &headers,
+        &format!("PATCH /v1/automations/{id}"),
+        &req,
+        StatusCode::OK,
+        map_automation_err,
+        async { state.automations.lock().await.update_automation(&id, req.clone()) },
+    )
+    .await
 }
 
 async fn delete_automation(
@@ -6763,32 +6901,83 @@ async fn delete_automation(
 async fn run_automation(
     State(state): State<RuntimeApiState>,
     Path(id): Path<String>,
-) -> Result<Json<AutomationRunRecord>, ApiError> {
-    // run_now_shared drops the manager mutex across the task-manager await so
-    // other automation endpoints stay responsive behind a slow enqueue.
-    let run =
-        crate::automation_manager::run_now_shared(&state.automations, &id, &state.task_manager)
-            .await
-            .map_err(map_automation_err)?;
-    Ok(Json(run))
+    Query(query): Query<AutomationPreconditionQuery>,
+    headers: HeaderMap,
+) -> Result<Response, ApiError> {
+    automation_mutation(
+        &state,
+        &headers,
+        &format!("POST /v1/automations/{id}/run"),
+        &query.expected_revision,
+        StatusCode::OK,
+        map_automation_err,
+        crate::automation_manager::run_now_shared_if(
+            &state.automations,
+            &id,
+            query.expected_revision,
+            &state.task_manager,
+        ),
+    )
+    .await
+}
+
+async fn set_automation_status(
+    state: RuntimeApiState,
+    id: String,
+    query: AutomationPreconditionQuery,
+    headers: HeaderMap,
+    action: &str,
+    status: crate::automation_manager::AutomationStatus,
+) -> Result<Response, ApiError> {
+    let req = UpdateAutomationRequest {
+        status: Some(status),
+        expected_revision: query.expected_revision,
+        ..UpdateAutomationRequest::default()
+    };
+    automation_mutation(
+        &state,
+        &headers,
+        &format!("POST /v1/automations/{id}/{action}"),
+        &query.expected_revision,
+        StatusCode::OK,
+        map_automation_err,
+        async { state.automations.lock().await.update_automation(&id, req) },
+    )
+    .await
 }
 
 async fn pause_automation(
     State(state): State<RuntimeApiState>,
     Path(id): Path<String>,
-) -> Result<Json<AutomationRecord>, ApiError> {
-    let manager = state.automations.lock().await;
-    let automation = manager.pause_automation(&id).map_err(map_automation_err)?;
-    Ok(Json(automation))
+    Query(query): Query<AutomationPreconditionQuery>,
+    headers: HeaderMap,
+) -> Result<Response, ApiError> {
+    set_automation_status(
+        state,
+        id,
+        query,
+        headers,
+        "pause",
+        crate::automation_manager::AutomationStatus::Paused,
+    )
+    .await
 }
 
 async fn resume_automation(
     State(state): State<RuntimeApiState>,
     Path(id): Path<String>,
-) -> Result<Json<AutomationRecord>, ApiError> {
-    let manager = state.automations.lock().await;
-    let automation = manager.resume_automation(&id).map_err(map_automation_err)?;
-    Ok(Json(automation))
+    Query(query): Query<AutomationPreconditionQuery>,
+    headers: HeaderMap,
+) -> Result<Response, ApiError> {
+    set_automation_status(
+        state,
+        id,
+        query,
+        headers,
+        "resume",
+        crate::automation_manager::AutomationStatus::Active,
+    )
+    .await
 }
 
 async fn list_automation_runs(
@@ -12520,6 +12709,14 @@ impl ApiError {
         }
     }
 
+    fn unprocessable(message: impl Into<String>) -> Self {
+        Self {
+            status: StatusCode::UNPROCESSABLE_ENTITY,
+            message: message.into(),
+            code: None,
+        }
+    }
+
     fn with_code(mut self, code: &'static str) -> Self {
         self.code = Some(code);
         self

crates/tui/src/tools/automation.rs

@@ -474,6 +474,7 @@ impl AutomationTool {
             auto_approve,
             delivery_mode: optional_delivery_mode(input)?,
             status,
+            expected_revision: None,
         };
         let automation = manager
             .update_automation(id, req)

crates/tui/src/tui/automation_panel.rs

@@ -332,6 +332,7 @@ mod tests {
             updated_at: now,
             next_run_at: None,
             last_run_at: None,
+            revision: 0,
         }
     }
 

crates/tui/src/tui/automation_routing.rs

@@ -615,6 +615,7 @@ mod tests {
             updated_at: now,
             next_run_at: None,
             last_run_at: None,
+            revision: 0,
         }
     }
 

crates/tui/src/tui/views/automations.rs

@@ -881,6 +881,7 @@ mod tests {
             updated_at: at,
             next_run_at: Some(at),
             last_run_at: None,
+            revision: 0,
         }
     }
 

crates/tui/src/tui/views/automations/editor.rs

@@ -1286,7 +1286,7 @@ mod tests {
         assert_eq!(saved.name, "renamed");
         let mut before = serde_json::to_value(original).unwrap();
         let mut after = serde_json::to_value(saved).unwrap();
-        for key in ["name", "updated_at"] {
+        for key in ["name", "updated_at", "revision"] {
             before.as_object_mut().unwrap().remove(key);
             after.as_object_mut().unwrap().remove(key);
         }

docs/RUNTIME_API.md

@@ -1694,6 +1694,28 @@ Create and update requests accept an optional `model`. When present, each
 scheduled or manually triggered run uses that model; omitting it keeps the
 runtime's default task model.
 
+**Automation mutation preconditions.** `GET /v1/runtime/info` advertises
+`capabilities.automation_mutation_preconditions: true`; require it before
+relying on the behavior below (older Engines ignore both inputs).
+
+- Every `AutomationRecord` carries an integer `revision`: `1` on create, `+1`
+  on each `PATCH`, `pause` and `resume`; legacy records read as `0`. Scheduler
+  bookkeeping (`next_run_at`, `last_run_at`) does not change it.
+- `expected_revision` is an optional precondition on `PATCH`, `run`, `pause`
+  and `resume`: a JSON body field on `PATCH`, or the `?expected_revision=N`
+  query parameter on any of the four. On mismatch nothing is written and the
+  response is `409` with
+  `{"error":{"code":"revision_conflict","current_revision":N,...}}`. Omitting
+  it behaves as before.
+- `Idempotency-Key` (1-255 visible ASCII characters) is optional on `POST
+  /v1/automations`, `PATCH`, `run`, `pause` and `resume`. The first successful
+  response (2xx) is persisted in `idempotency.json` beside the automation
+  store, bounded to 1024 entries and a 24 hour TTL. A retry with the same key
+  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.
+
 **Operate** (always-on named operation; same `OperateRecord` as CWC
 `20de981` / PR #284)