Persist workspace job receipts and shorten implicit sockets [skip ci]

Use the existing Runtime thread store and shell manager for keyed setup, run, archive and command admission. Persist intent before dispatch; exact retries return the same receipt, changed requests refuse, and unconfirmed outcomes fence new work and removal. Cache a positively observed child exit before output cleanup so Stop can persist its terminal receipt. Retain small terminal witnesses across output eviction.

Bind client recovery to the complete request fingerprint and canonical workspace/cwd, validate project policy against the approved root, and avoid nested blocking-pool filesystem work while holding admission. Shorten overlong implicit control sockets into a private stable endpoint without moving user data or changing explicit socket choices. The shared Runtime API also carries the reviewed catalog refresh projection from the already-landed model catalog source.

Verified a captured macOS development CLI build and exact source matching for 15 of these 16 paths; the newer jobs.rs was compiled and exercised in the Linux Boat candidate. Actual HTTP observations passed concurrent single-spawn/replay, mismatch/store/workspace guards, Stop and subsequent Run, prelaunch PTY refusal with restored mode, restart fences and private receipt content checks. Actual Engine plus transpiled TypeScript passed canonical alias binding, one shell spawn, same-key replay and a discarded real changed-env 409 that stayed unconfirmed. Separate prior paired GUI Run/output/Stop/reload and real output-eviction/deletion observations passed. Existing model API checks passed nine controlled cases.

No new product tests or code comments. These are scoped development/runtime receipts, not final-candidate packaging, Windows, provider inference or release qualification. Parent exit confirmation does not prove detached descendants are gone; path checks do not provide a directory-handle replacement guarantee, and archive recovery remains multi-phase work.

crates/app-server-protocol/src/runtime.rs

@@ -123,6 +123,8 @@ pub struct RuntimeCapabilities {
     /// `GET`/`POST /v1/worktrees` and `POST /v1/worktrees/remove`.
     #[serde(default)]
     pub worktrees: bool,
+    #[serde(default)]
+    pub job_operations: bool,
     /// Explicit per-thread shell opt-in is checked against loaded policy and
     /// cannot broaden a conversation while it has an active turn.
     #[serde(default)]
@@ -537,6 +539,7 @@ mod tests {
             thread_scoped_git_writes: true,
             thread_scoped_snapshots: true,
             worktrees: true,
+            job_operations: true,
             thread_shell_consent: true,
             turns: true,
             turn_operation_idempotency: true,

crates/app-server/src/daemon_client.rs

@@ -41,6 +41,32 @@ mod platform {
         tokio::io::WriteHalf<Connection>,
     >;
 
+    fn same_socket_selection(left: &std::path::Path, right: &std::path::Path) -> bool {
+        if left == right {
+            return true;
+        }
+        #[cfg(unix)]
+        {
+            if left.file_name().is_none() || left.file_name() != right.file_name() {
+                return false;
+            }
+            match (
+                left.parent()
+                    .map(codewhale_paths::resolve_root_owned_ancestors),
+                right
+                    .parent()
+                    .map(codewhale_paths::resolve_root_owned_ancestors),
+            ) {
+                (Some(left), Some(right)) => left == right,
+                _ => false,
+            }
+        }
+        #[cfg(windows)]
+        {
+            false
+        }
+    }
+
     pub struct OwnerClient {
         read: BoundedLines<BufReader<tokio::io::ReadHalf<OwnerTransport>>>,
         write: tokio::io::WriteHalf<OwnerTransport>,
@@ -194,7 +220,7 @@ mod platform {
                 old.version == owner.version
                     && old.data_dir == owner.data_dir
                     && old.execution_scope == owner.execution_scope
-                    && old.socket_path == endpoint
+                    && same_socket_selection(&old.socket_path, endpoint)
                     && old.config_path == owner.config_path
                     && old.principal == owner.principal,
                 "captured stale owner belongs to another selected store/config/principal"
@@ -278,7 +304,7 @@ mod platform {
             anyhow::ensure!(
                 receipt.version == 1
                     && receipt.pid > 0
-                    && receipt.socket_path == socket_path
+                    && same_socket_selection(&receipt.socket_path, &socket_path)
                     && !receipt.lease_generation.is_empty(),
                 "selected owner receipt is invalid"
             );

crates/app-server/src/daemon_socket.rs

@@ -1663,7 +1663,7 @@ mod tests {
             explicit: None,
             codewhale_home_override: None,
             xdg_runtime_dir: None,
-            user_home: Some(PathBuf::from("/home/whale")),
+            user_home: Some(PathBuf::from("/cw-test-home/whale")),
             macos: false,
         }
     }
@@ -1724,7 +1724,7 @@ mod tests {
         let resolved = resolve_socket_path(&inputs()).expect("resolve");
         assert_eq!(
             resolved,
-            PathBuf::from("/home/whale/.codewhale/run/daemon.sock")
+            PathBuf::from("/cw-test-home/whale/.codewhale/run/daemon.sock")
         );
     }
 
@@ -1734,7 +1734,7 @@ mod tests {
         let resolved = store_selected_socket_path(None, &inputs()).expect("resolve");
         assert_eq!(
             resolved,
-            PathBuf::from("/home/whale/.codewhale/run/daemon.sock")
+            PathBuf::from("/cw-test-home/whale/.codewhale/run/daemon.sock")
         );
     }
 
@@ -1742,18 +1742,18 @@ mod tests {
     #[test]
     fn distinct_stores_derive_distinct_sibling_sockets() {
         let one = store_selected_socket_path(
-            Some(Path::new("/home/whale/.local/state/one/runtime")),
+            Some(Path::new("/cw-test-home/whale/.local/state/one/runtime")),
             &inputs(),
         )
         .expect("resolve");
         let two = store_selected_socket_path(
-            Some(Path::new("/home/whale/.local/state/two/runtime")),
+            Some(Path::new("/cw-test-home/whale/.local/state/two/runtime")),
             &inputs(),
         )
         .expect("resolve");
         assert_eq!(
             one.parent().unwrap(),
-            Path::new("/home/whale/.codewhale/run")
+            Path::new("/cw-test-home/whale/.codewhale/run")
         );
         assert_eq!(one.parent().unwrap(), two.parent().unwrap());
         assert_ne!(one, two);
@@ -1766,7 +1766,7 @@ mod tests {
     #[cfg(unix)]
     #[test]
     fn the_derived_socket_is_a_pure_function_of_the_store_root() {
-        let store = Path::new("/home/whale/.local/state/one/runtime");
+        let store = Path::new("/cw-test-home/whale/.local/state/one/runtime");
         let first = store_selected_socket_path(Some(store), &inputs()).expect("resolve");
         let second = store_selected_socket_path(Some(store), &inputs()).expect("resolve");
         assert_eq!(first, second);

crates/app-server/src/runtime_api.rs

@@ -1555,6 +1555,7 @@ struct RuntimeInfoResponse {
     auth_required: bool,
     transports: Vec<&'static str>,
     capabilities: RuntimeCapabilities,
+    job_operation_store_id: Option<String>,
     account: RuntimeAccountInfo,
     experimental: RuntimeExperimentalCapabilities,
     // Backward-compatible alias kept for existing clients.
@@ -1575,6 +1576,7 @@ fn default_runtime_capabilities() -> RuntimeCapabilities {
         thread_scoped_git_writes: true,
         thread_scoped_snapshots: true,
         worktrees: true,
+        job_operations: true,
         thread_shell_consent: true,
         turns: true,
         turn_operation_idempotency: true,
@@ -2546,7 +2548,8 @@ async fn store_selected_control_socket(
     let selected_store = selected_store.to_path_buf();
     codewhale_runtime::runtime_control::owner_work(move || {
         let store_override = codewhale_engine::runtime_threads::canonical_runtime_dir_override()?;
-        let inputs = codewhale_paths::daemon_socket::SocketPathInputs::from_environment(None)?;
+        let inputs =
+            codewhale_paths::daemon_socket::SocketPathInputs::from_environment(explicit.clone())?;
         store_selected_control_socket_with(
             explicit,
             &selected_store,
@@ -3035,6 +3038,15 @@ pub fn build_router(state: RuntimeApiState) -> Router {
             "/v1/threads/{id}/jobs",
             get(jobs::list_thread_jobs).post(jobs::create_thread_job),
         )
+        .route("/v1/job-operations", get(jobs::list_job_operations))
+        .route(
+            "/v1/threads/{id}/job-operations",
+            get(jobs::list_thread_job_operations),
+        )
+        .route(
+            "/v1/threads/{id}/job-operations/{operation_key}",
+            get(jobs::get_job_operation),
+        )
         .route("/v1/threads/{id}/jobs/{job_id}", get(jobs::get_thread_job))
         .route(
             "/v1/threads/{id}/jobs/{job_id}/output",
@@ -6940,6 +6952,8 @@ async fn runtime_info(
         auth_required: state.auth_required,
         transports: vec!["http", "sse"],
         capabilities,
+        job_operation_store_id: runtime_request_is_authorized(&request, &state)
+            .then(|| state.runtime_threads.job_operation_store_id().to_string()),
         account,
         experimental: RuntimeExperimentalCapabilities::default(),
         version,
@@ -12039,6 +12053,37 @@ struct ProvidersResponse {
     /// whichever entry happens to share the generic kind.
     current_provider_id: Option<String>,
     providers: Vec<ProviderEntry>,
+    catalog_revision: String,
+    public_catalog: PublicCatalogStatus,
+}
+
+#[derive(Debug, Clone, Serialize)]
+struct PublicCatalogStatus {
+    freshness: codewhale_engine::models_dev_live::ModelsDevFreshness,
+    fetched_at: Option<u64>,
+    outcome: &'static str,
+    error: Option<&'static str>,
+}
+
+fn public_catalog_status() -> PublicCatalogStatus {
+    let snapshot = codewhale_engine::models_dev_live::status();
+    PublicCatalogStatus {
+        freshness: snapshot.freshness,
+        fetched_at: snapshot.fetched_at,
+        outcome: if snapshot.last_error.is_some() {
+            "failed"
+        } else {
+            "cached"
+        },
+        error: snapshot.last_error.is_some().then_some("refresh_failed"),
+    }
+}
+
+#[derive(Serialize)]
+struct RefreshProviderModelsResponse {
+    #[serde(flatten)]
+    provider: codewhale_engine::provider_lake::CatalogUpdateReceipt,
+    public_catalog: PublicCatalogStatus,
 }
 
 /// Entry in `GET /v1/providers/{id}/models`.
@@ -12724,6 +12769,7 @@ async fn list_providers(
         #[cfg(any(test, feature = "test-support"))]
         let _membership = codewhale_engine::test_support::join_env_scope(env_ticket);
         let config = state.config.read().clone();
+        let (live_generation, cloud_generation) = codewhale_engine::provider_lake::catalog_revision();
         secrets::invalidate_stale_account_catalog(&config);
         let active_identity = config.active_provider_identity().ok();
         let current = active_identity.as_ref().map_or_else(|| config.provider.clone().unwrap_or_else(|| "unavailable".into()), |identity| identity.persisted_kind().to_string());
@@ -12739,6 +12785,8 @@ async fn list_providers(
             current_provider_id: active_identity.as_ref().and_then(|identity| identity.persisted_id()).map(str::to_string),
             current,
             providers,
+            catalog_revision: format!("{live_generation}:{cloud_generation}"),
+            public_catalog: public_catalog_status(),
         }))
     })
     .await
@@ -12831,13 +12879,15 @@ async fn list_provider_models(
 #[serde(deny_unknown_fields)]
 struct RefreshProviderModelsParams {
     model_provider_id: Option<String>,
+    #[serde(default)]
+    force_catalog: bool,
 }
 
 async fn refresh_provider_models(
     State(state): State<RuntimeApiState>,
     Path(id): Path<String>,
     Query(params): Query<RefreshProviderModelsParams>,
-) -> Result<Json<codewhale_engine::provider_lake::CatalogUpdateReceipt>, ApiError> {
+) -> Result<Json<RefreshProviderModelsResponse>, ApiError> {
     let runtime = tokio::runtime::Handle::current();
     #[cfg(any(test, feature = "test-support"))]
     let env_ticket = codewhale_engine::test_support::env_scope_ticket();
@@ -12846,9 +12896,40 @@ async fn refresh_provider_models(
         let _membership = codewhale_engine::test_support::join_env_scope(env_ticket);
         let config = state.config.read().clone();
         let identity = provider_models_identity(&config, &id, params.model_provider_id.as_deref())?;
-        Ok(Json(runtime.block_on(
-            codewhale_engine::provider_lake::update_provider_catalog(&config, &identity),
-        )))
+        Ok(Json(runtime.block_on(async {
+            let before = public_catalog_status();
+            let (public_result, provider) = tokio::join!(
+                codewhale_engine::models_dev_live::refresh(params.force_catalog),
+                codewhale_engine::provider_lake::update_provider_catalog(&config, &identity),
+            );
+            let mut public_catalog = public_catalog_status();
+            match public_result {
+                Ok(_)
+                    if public_catalog.error.is_none()
+                        && (params.force_catalog
+                            || before.fetched_at != public_catalog.fetched_at) =>
+                {
+                    public_catalog.outcome = "updated";
+                }
+                Ok(_) => {}
+                Err(error) => {
+                    use codewhale_engine::models_dev_live::ModelsDevRefreshError;
+                    public_catalog.outcome = "failed";
+                    public_catalog.error = Some(match error {
+                        ModelsDevRefreshError::Disabled => "fetch_disabled",
+                        ModelsDevRefreshError::Network(_) => "network",
+                        ModelsDevRefreshError::HttpStatus(_) => "http_status",
+                        ModelsDevRefreshError::InvalidResponse(_) => "invalid_response",
+                        ModelsDevRefreshError::EmptyCatalog => "empty_catalog",
+                        ModelsDevRefreshError::Io(_) => "cache_io",
+                    });
+                }
+            }
+            RefreshProviderModelsResponse {
+                provider,
+                public_catalog,
+            }
+        })))
     })
     .await
     .map_err(|_| ApiError::internal("Provider model refresh failed"))?

crates/app-server/src/runtime_api/git.rs

@@ -2617,13 +2617,82 @@ pub(super) async fn worktree_remove(
     }
     args.push("--".to_string());
     args.push(row.path.clone());
-    let run = tokio::spawn(async move {
+    let managers = state.runtime_threads.shell_managers_snapshot().await;
+    let generation = state
+        .runtime_threads
+        .job_operation_generation()
+        .map_err(super::jobs::operation_error)?;
+    let runtime_owner = state.runtime_threads.clone();
+    let store = state.runtime_threads.job_operation_store();
+    let runtime = tokio::runtime::Handle::current();
+    let fence_path = path.clone();
+    let run = tokio::task::spawn_blocking(move || {
         let _admission = admission;
         let _guard = guard;
-        git_write(&main, args).await
+        store.with_job_operations(|transaction| {
+            Ok((|| -> Result<GitRun, ApiError> {
+                let worktree = fence_path.canonicalize().unwrap_or(fence_path);
+                let worktree = if cfg!(windows) {
+                    let raw = worktree
+                        .to_str()
+                        .ok_or_else(|| ApiError::internal("worktree path unavailable"))?;
+                    PathBuf::from(super::jobs::plain_windows_path(raw))
+                } else {
+                    worktree
+                };
+                for mut record in transaction.list().map_err(super::jobs::operation_error)? {
+                    let (operation, _) = super::jobs::project_operation(
+                        transaction,
+                        &mut record,
+                        &managers,
+                        &generation,
+                    )?;
+                    if !matches!(operation.status.as_str(), "completed" | "rejected")
+                        && (super::jobs::overlaps(&worktree, &operation.workspace)
+                            || super::jobs::overlaps(&worktree, &operation.cwd))
+                    {
+                        return Err(ApiError::conflict(
+                            "a workspace job is running or unconfirmed",
+                        )
+                        .with_code("worktree_busy"));
+                    }
+                }
+                for (_, manager) in &managers {
+                    let mut manager = manager
+                        .lock()
+                        .map_err(|_| ApiError::internal("job listing failed"))?;
+                    for job in manager.list_jobs() {
+                        if !job.stale
+                            && job.status == codewhale_engine::tools::shell::ShellStatus::Running
+                        {
+                            let cwd = job.cwd.canonicalize().ok().and_then(|cwd| {
+                                if cfg!(windows) {
+                                    cwd.to_str().map(|raw| {
+                                        PathBuf::from(super::jobs::plain_windows_path(raw))
+                                    })
+                                } else {
+                                    Some(cwd)
+                                }
+                            });
+                            if cwd.is_none_or(|cwd| super::jobs::overlaps(&worktree, &cwd)) {
+                                return Err(ApiError::conflict("a workspace job is still running")
+                                    .with_code("worktree_busy"));
+                            }
+                        }
+                    }
+                }
+                if runtime_owner.job_operation_generation().ok().as_deref()
+                    != Some(generation.as_str())
+                {
+                    return Err(ApiError::internal("job operation owner changed"));
+                }
+                runtime.block_on(git_write(&main, args))
+            })())
+        })
     })
     .await
-    .map_err(|_| ApiError::internal("worktree removal failed"))??;
+    .map_err(|_| ApiError::internal("worktree removal failed"))?
+    .map_err(super::jobs::operation_error)??;
     if !run.status_success {
         return Err(ApiError::bad_request(output_tail(&run)).into());
     }

crates/app-server/src/runtime_api/jobs.rs

@@ -6,7 +6,7 @@
 //! per-session completion drain never claims client-launched work as model
 //! evidence — and the model's jobs never appear "client-owned" here.
 
-use std::collections::HashMap;
+use std::collections::{BTreeMap, HashMap};
 
 use axum::Json;
 use axum::extract::{Path, Query, State};
@@ -14,9 +14,15 @@ use axum::http::StatusCode;
 use base64::Engine as _;
 use serde::{Deserialize, Serialize};
 
+use codewhale_engine::runtime_threads::RuntimeThreadStore;
+use codewhale_engine::runtime_threads::job_operations::{
+    JobOperation, JobOperationRecord, JobOperationTransaction, JobOperationsBusy,
+    request_fingerprint, validate_key,
+};
+use codewhale_engine::tools::shell::SharedShellManager;
 use codewhale_engine::tools::shell::{
-    PtyDimensions, ShellJobSnapshot, ShellManager, ShellOutputChunk, ShellOutputStream,
-    ShellResult, ShellStatus,
+    PtyDimensions, ShellJobSnapshot, ShellLaunchNotStarted, ShellManager, ShellOutputChunk,
+    ShellOutputStream, ShellResult, ShellStatus,
 };
 
 use super::{ApiError, RuntimeApiState, map_thread_err};
@@ -170,11 +176,268 @@ pub(super) struct CreateJobRequest {
     tty: bool,
     #[serde(default)]
     env: HashMap<String, String>,
+    operation_key: Option<String>,
+    purpose: Option<String>,
+    expected_store_id: Option<String>,
+    expected_workspace: Option<String>,
 }
 
 #[derive(Debug, Serialize)]
 pub(super) struct CreateJobResponse {
-    job: JobView,
+    job: Option<JobView>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    operation: Option<JobOperation>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    request_fingerprint: Option<String>,
+}
+
+pub(super) type JobManagers = Vec<(String, SharedShellManager)>;
+
+pub(super) fn operation_error(error: anyhow::Error) -> ApiError {
+    if error.downcast_ref::<JobOperationsBusy>().is_some() {
+        ApiError::conflict("job operation admission is busy").with_code("job_operation_pending")
+    } else {
+        ApiError::internal("job operation storage unavailable")
+    }
+}
+
+pub(super) fn project_operation(
+    transaction: &JobOperationTransaction<'_>,
+    record: &mut JobOperationRecord,
+    managers: &JobManagers,
+    generation: &str,
+) -> Result<(JobOperation, Option<JobView>), ApiError> {
+    let mut operation = record.operation.clone();
+    let mut view = None;
+    if !matches!(operation.status.as_str(), "completed" | "rejected") {
+        operation.status = "unknown".to_string();
+        if record.owner_generation == generation
+            && let Some(job_id) = record.operation.job_id.clone()
+            && let Some((_, manager)) = managers
+                .iter()
+                .find(|(thread, _)| thread == &operation.thread_id)
+            && let Ok(mut manager) = manager.lock()
+            && let Some(witness) = manager.job_receipt_witness(&job_id)
+        {
+            if !witness.wait_failed
+                && (witness.status == ShellStatus::Running || witness.terminal_confirmed)
+                && witness.owner_session_id == api_job_scope(&operation.thread_id)
+                && witness.cwd == operation.cwd
+            {
+                operation.status = if witness.status == ShellStatus::Running {
+                    "running"
+                } else {
+                    "completed"
+                }
+                .to_string();
+                if operation.status == "completed" {
+                    operation.job_status = Some(format!("{:?}", witness.status));
+                    operation.exit_code = witness.exit_code;
+                    operation.updated_at_ms = chrono::Utc::now()
+                        .timestamp_millis()
+                        .max(operation.created_at_ms);
+                    record.operation = operation.clone();
+                    transaction.save(record).map_err(operation_error)?;
+                    manager.release_job_receipt(&job_id);
+                }
+                if let Ok(detail) = manager.inspect_job(&job_id)
+                    && !detail.snapshot.stale
+                {
+                    view = Some(JobView::new(
+                        detail.snapshot,
+                        operation.thread_id.clone(),
+                        &manager,
+                    ));
+                }
+            }
+        }
+    } else if record.owner_generation == generation
+        && let Some(job_id) = operation.job_id.as_deref()
+        && let Some((_, manager)) = managers
+            .iter()
+            .find(|(thread, _)| thread == &operation.thread_id)
+        && let Ok(mut manager) = manager.lock()
+        && let Ok(detail) = manager.inspect_job(job_id)
+        && !detail.snapshot.stale
+        && detail.snapshot.owner_session_id == api_job_scope(&operation.thread_id)
+        && detail.snapshot.cwd == operation.cwd
+    {
+        view = Some(JobView::new(
+            detail.snapshot,
+            operation.thread_id.clone(),
+            &manager,
+        ));
+    }
+    Ok((operation, view))
+}
+
+fn observe_operation(
+    store: RuntimeThreadStore,
+    runtime_owner: std::sync::Weak<
+        codewhale_engine::runtime_threads::manager::RuntimeThreadManager,
+    >,
+    manager: std::sync::Weak<std::sync::Mutex<ShellManager>>,
+    thread_id: String,
+    generation: String,
+    id: String,
+) {
+    tokio::spawn(async move {
+        let mut delay = std::time::Duration::from_millis(500);
+        loop {
+            tokio::time::sleep(delay).await;
+            let store = store.clone();
+            let runtime_owner = runtime_owner.clone();
+            let manager = manager.clone();
+            let thread_id = thread_id.clone();
+            let generation = generation.clone();
+            let id = id.clone();
+            let result = tokio::task::spawn_blocking(move || {
+                let Some(runtime_owner) = runtime_owner.upgrade() else {
+                    return Ok(None);
+                };
+                if runtime_owner.job_operation_generation().ok().as_deref()
+                    != Some(generation.as_str())
+                {
+                    return Ok(None);
+                }
+                let Some(manager) = manager.upgrade() else {
+                    return Ok(None);
+                };
+                let managers = vec![(thread_id, manager)];
+                store.with_job_operations(|transaction| {
+                    let Some(mut record) = transaction.get(&id)? else {
+                        return Ok(None);
+                    };
+                    let projected =
+                        project_operation(transaction, &mut record, &managers, &generation)
+                            .map_err(|_| anyhow::anyhow!("job observation unavailable"))?;
+                    match projected.0.status.as_str() {
+                        "running" => Ok(Some(true)),
+                        "unknown" => {
+                            let available =
+                                record.operation.job_id.as_deref().is_some_and(|job_id| {
+                                    managers[0].1.lock().ok().is_some_and(|mut manager| {
+                                        manager.job_receipt_witness(job_id).is_some()
+                                    })
+                                });
+                            Ok(available.then_some(false))
+                        }
+                        _ => Ok(None),
+                    }
+                })
+            })
+            .await;
+            match result {
+                Ok(Ok(Some(true))) => delay = std::time::Duration::from_millis(500),
+                Ok(Ok(Some(false))) => delay = (delay * 2).min(std::time::Duration::from_secs(10)),
+                Ok(Ok(None)) => break,
+                _ => delay = (delay * 2).min(std::time::Duration::from_secs(10)),
+            }
+        }
+    });
+}
+
+pub(super) fn overlaps(left: &std::path::Path, right: &std::path::Path) -> bool {
+    left.starts_with(right) || right.starts_with(left)
+}
+
+#[derive(Serialize)]
+pub(super) struct JobOperationsResponse {
+    operations: Vec<JobOperation>,
+}
+
+pub(super) async fn list_job_operations(
+    State(state): State<RuntimeApiState>,
+) -> Result<Json<JobOperationsResponse>, ApiError> {
+    list_operations(&state, None).await
+}
+
+pub(super) async fn list_thread_job_operations(
+    State(state): State<RuntimeApiState>,
+    Path(thread): Path<String>,
+) -> Result<Json<JobOperationsResponse>, ApiError> {
+    state
+        .runtime_threads
+        .get_thread(&thread)
+        .await
+        .map_err(map_thread_err)?;
+    list_operations(&state, Some(thread)).await
+}
+
+async fn list_operations(
+    state: &RuntimeApiState,
+    thread: Option<String>,
+) -> Result<Json<JobOperationsResponse>, ApiError> {
+    let managers = state.runtime_threads.shell_managers_snapshot().await;
+    let generation = state
+        .runtime_threads
+        .job_operation_generation()
+        .map_err(operation_error)?;
+    let store = state.runtime_threads.job_operation_store();
+    let operations = tokio::task::spawn_blocking(move || {
+        store.with_job_operations(|transaction| {
+            let mut operations = Vec::new();
+            for mut record in transaction.list()? {
+                if thread
+                    .as_ref()
+                    .is_none_or(|thread| thread == &record.operation.thread_id)
+                {
+                    let (operation, _) =
+                        project_operation(transaction, &mut record, &managers, &generation)
+                            .map_err(|_| anyhow::anyhow!("job operation projection failed"))?;
+                    operations.push(operation);
+                }
+            }
+            Ok(operations)
+        })
+    })
+    .await
+    .map_err(|_| ApiError::internal("job operation listing failed"))?
+    .map_err(operation_error)?;
+    Ok(Json(JobOperationsResponse { operations }))
+}
+
+pub(super) async fn get_job_operation(
+    State(state): State<RuntimeApiState>,
+    Path((thread, key)): Path<(String, String)>,
+) -> Result<Json<CreateJobResponse>, ApiError> {
+    validate_key(&key).map_err(|_| ApiError::bad_request("invalid job operation key"))?;
+    state
+        .runtime_threads
+        .get_thread(&thread)
+        .await
+        .map_err(map_thread_err)?;
+    let managers = state.runtime_threads.shell_managers_snapshot().await;
+    let generation = state
+        .runtime_threads
+        .job_operation_generation()
+        .map_err(operation_error)?;
+    let store = state.runtime_threads.job_operation_store();
+    let result = tokio::task::spawn_blocking(move || {
+        store.with_job_operations(|transaction| {
+            let id = transaction.operation_id(&thread, &key)?;
+            let Some(mut record) = transaction.get(&id)? else {
+                return Ok(None);
+            };
+            let result = project_operation(transaction, &mut record, &managers, &generation)
+                .map_err(|_| anyhow::anyhow!("job operation projection failed"))?;
+            Ok(Some((
+                result.0,
+                result.1,
+                record.request_fingerprint.clone(),
+            )))
+        })
+    })
+    .await
+    .map_err(|_| ApiError::internal("job operation lookup failed"))?
+    .map_err(operation_error)?;
+    let (operation, job, fingerprint) =
+        result.ok_or_else(|| ApiError::not_found("job operation not found"))?;
+    Ok(Json(CreateJobResponse {
+        job,
+        operation: Some(operation),
+        request_fingerprint: Some(fingerprint),
+    }))
 }
 
 /// Resolve a job `cwd` against the thread workspace, never the server's cwd.
@@ -189,9 +452,8 @@ pub(super) struct CreateJobResponse {
 /// whose name is not UTF-8 is refused rather than lossily renamed into a
 /// different one. On Windows the `\\?\` verbatim prefix `canonicalize` adds
 /// is dropped ([`plain_windows_path`]) because `cmd.exe` refuses it as a
-/// current directory. Runs on `tokio::fs`, so no path resolution blocks a
-/// runtime worker.
-async fn resolve_job_cwd(
+/// current directory.
+fn resolve_job_cwd(
     workspace: &std::path::Path,
     raw: &str,
     trust_mode: bool,
@@ -203,18 +465,12 @@ async fn resolve_job_cwd(
         workspace.join(requested)
     };
     let not_a_directory = || ApiError::bad_request("cwd must be an existing directory");
-    let resolved = tokio::fs::canonicalize(&candidate)
-        .await
-        .map_err(|_| not_a_directory())?;
-    if !tokio::fs::metadata(&resolved)
-        .await
-        .is_ok_and(|meta| meta.is_dir())
-    {
+    let resolved = std::fs::canonicalize(&candidate).map_err(|_| not_a_directory())?;
+    if !std::fs::metadata(&resolved).is_ok_and(|meta| meta.is_dir()) {
         return Err(not_a_directory());
     }
     if !trust_mode {
-        let root = tokio::fs::canonicalize(workspace)
-            .await
+        let root = std::fs::canonicalize(workspace)
             .map_err(|_| ApiError::internal("thread workspace is unavailable"))?;
         if !resolved.starts_with(&root) {
             return Err(ApiError::forbidden(
@@ -235,7 +491,7 @@ async fn resolve_job_cwd(
 
 /// `\\?\C:\dir` -> `C:\dir` and `\\?\UNC\host\share` -> `\\host\share`; any
 /// other spelling is returned unchanged. A string rule so every host tests it.
-fn plain_windows_path(path: &str) -> String {
+pub(super) fn plain_windows_path(path: &str) -> String {
     if let Some(rest) = path.strip_prefix(r"\\?\UNC\") {
         return format!(r"\\{rest}");
     }
@@ -250,6 +506,16 @@ fn plain_windows_path(path: &str) -> String {
     }
 }
 
+fn canonical_job_path(path: &std::path::Path) -> Option<std::path::PathBuf> {
+    let path = std::fs::canonicalize(path).ok()?;
+    if cfg!(windows) {
+        path.to_str()
+            .map(|raw| std::path::PathBuf::from(plain_windows_path(raw)))
+    } else {
+        Some(path)
+    }
+}
+
 /// `POST /v1/threads/{id}/jobs` — launch a client-owned background job under
 /// the thread's own sandbox policy. The client asking is the approval; the
 /// thread's posture still bounds what the job may touch.
@@ -262,6 +528,9 @@ pub(super) async fn create_thread_job(
     if command.is_empty() {
         return Err(ApiError::bad_request("command is required"));
     }
+    if command.contains('\0') {
+        return Err(ApiError::bad_request("command contains a NUL byte"));
+    }
     if command.len() > COMMAND_MAX_BYTES {
         return Err(ApiError::bad_request(format!(
             "command must be at most {COMMAND_MAX_BYTES} bytes"
@@ -273,29 +542,145 @@ pub(super) async fn create_thread_job(
         )));
     }
     for (key, value) in &request.env {
-        if key.len() > ENV_KEY_MAX_BYTES || key.contains(['=', '\0']) {
+        if key.is_empty() || key.len() > ENV_KEY_MAX_BYTES || key.contains(['=', '\0']) {
             return Err(ApiError::bad_request("invalid env key"));
         }
         if value.len() > ENV_VALUE_MAX_BYTES || value.contains('\0') {
             return Err(ApiError::bad_request("invalid env value"));
         }
     }
+    #[cfg(target_env = "ohos")]
+    if request.tty {
+        return Err(ApiError::bad_request(
+            "TTY shell mode is not supported on HarmonyOS/OpenHarmony",
+        ));
+    }
 
+    let keyed = match (
+        &request.operation_key,
+        &request.purpose,
+        &request.expected_store_id,
+        &request.expected_workspace,
+    ) {
+        (None, None, None, None) => false,
+        (Some(key), Some(purpose), Some(expected), Some(workspace)) => {
+            if workspace.trim().is_empty() {
+                return Err(ApiError::bad_request("expected_workspace is required"));
+            }
+            validate_key(key).map_err(|_| ApiError::bad_request("invalid job operation key"))?;
+            if !matches!(purpose.as_str(), "setup" | "run" | "archive" | "command") {
+                return Err(ApiError::bad_request("invalid job operation purpose"));
+            }
+            if expected != state.runtime_threads.job_operation_store_id() {
+                return Err(ApiError::conflict("job operation store changed")
+                    .with_code("job_operation_store_changed"));
+            }
+            true
+        }
+        _ => {
+            return Err(ApiError::bad_request(
+                "operation_key, purpose, expected_store_id and expected_workspace must be supplied together",
+            ));
+        }
+    };
     let admission = state.runtime_threads.session_checkpoint_guard().await;
     let thread = state
         .runtime_threads
         .get_thread(&thread_id)
         .await
         .map_err(map_thread_err)?;
+    let command = command.to_string();
+    let fingerprint = if let Some(key) = request.operation_key.as_deref() {
+        let env: BTreeMap<_, _> = request.env.iter().collect();
+        let env: Vec<_> = env.into_iter().collect();
+        let payload = serde_json::to_vec(&(
+            2,
+            &thread_id,
+            &request.expected_workspace,
+            &command,
+            &request.cwd,
+            request.timeout_ms.unwrap_or(120_000),
+            request.tty,
+            env,
+            &request.purpose,
+        ))
+        .map_err(|_| ApiError::internal("job operation validation failed"))?;
+        Some(
+            request_fingerprint(key, &payload)
+                .map_err(|_| ApiError::bad_request("invalid job operation key"))?,
+        )
+    } else {
+        None
+    };
+    let generation = state
+        .runtime_threads
+        .job_operation_generation()
+        .map_err(operation_error)?;
+    let store = state.runtime_threads.job_operation_store();
+    if let Some(key) = request.operation_key.clone() {
+        let lookup_store = store.clone();
+        let lookup_thread = thread_id.clone();
+        let lookup_fingerprint = fingerprint.clone();
+        let lookup_generation = generation.clone();
+        let managers = state.runtime_threads.shell_managers_snapshot().await;
+        let replay = tokio::task::spawn_blocking(move || {
+            lookup_store.with_job_operations(|transaction| {
+                let id = transaction.operation_id(&lookup_thread, &key)?;
+                let Some(mut record) = transaction.get(&id)? else {
+                    return Ok(None);
+                };
+                if Some(&record.request_fingerprint) != lookup_fingerprint.as_ref() {
+                    return Ok(Some(Err(ApiError::conflict(
+                        "job operation key belongs to a different request",
+                    )
+                    .with_code("job_operation_mismatch"))));
+                }
+                Ok(Some(
+                    project_operation(transaction, &mut record, &managers, &lookup_generation).map(
+                        |(operation, job)| (operation, job, record.request_fingerprint.clone()),
+                    ),
+                ))
+            })
+        })
+        .await
+        .map_err(|_| ApiError::internal("job operation lookup failed"))?
+        .map_err(operation_error)?;
+        if let Some(result) = replay {
+            let (operation, job, fingerprint) = result?;
+            return Ok((
+                StatusCode::OK,
+                Json(CreateJobResponse {
+                    job,
+                    operation: Some(operation),
+                    request_fingerprint: Some(fingerprint),
+                }),
+            ));
+        }
+    }
     if !thread.allow_shell {
         return Err(ApiError::forbidden(
             "this thread does not allow shell commands",
         ));
     }
+    let requested_workspace = thread.workspace.clone();
+    let expected_workspace = request.expected_workspace.clone();
+    let workspace = tokio::task::spawn_blocking(move || {
+        let workspace = canonical_job_path(&requested_workspace)
+            .ok_or_else(|| ApiError::internal("thread workspace is unavailable"))?;
+        if expected_workspace.as_deref().is_some_and(|expected| {
+            canonical_job_path(std::path::Path::new(expected)).as_ref() != Some(&workspace)
+        }) {
+            return Err(ApiError::conflict("job operation workspace changed")
+                .with_code("job_operation_workspace_changed"));
+        }
+        Ok(workspace)
+    })
+    .await
+    .map_err(|_| ApiError::internal("thread workspace resolution failed"))??;
     state
         .runtime_threads
         .validate_shell_access_policy(
-            &thread.workspace,
+            &workspace,
             state.config_path.as_deref(),
             state.config_profile.as_deref(),
         )
@@ -314,50 +699,238 @@ pub(super) async fn create_thread_job(
             "this server runs commands in an external sandbox backend, and a thread job starts a local process; run the command through the thread's shell tool instead",
         ));
     }
-    // The manager resolves a relative `working_dir` against the server's own
-    // cwd, so hand it the directory resolved against the thread workspace.
-    let request_cwd = match request.cwd.as_deref() {
-        Some(cwd) => Some(resolve_job_cwd(&thread.workspace, cwd, thread.trust_mode).await?),
-        None => None,
-    };
+    let mut policy_thread = thread.clone();
+    policy_thread.workspace = workspace.clone();
     let policy = state
         .runtime_threads
-        .thread_job_sandbox_policy(&thread)
+        .thread_job_sandbox_policy(&policy_thread)
         .await;
     let manager = thread_manager(&state, &thread_id, true).await?;
+    let managers = state.runtime_threads.shell_managers_snapshot().await;
+    let store_id = state.runtime_threads.job_operation_store_id().to_string();
+    let runtime_owner = state.runtime_threads.clone();
     let scope = api_job_scope(&thread_id);
-    let request_timeout = request.timeout_ms;
-    let request_tty = request.tty;
-    let request_env = request.env;
-    let command = command.to_string();
-    let job = tokio::task::spawn_blocking(move || -> Result<JobView, ApiError> {
+    let result = tokio::task::spawn_blocking(move || {
         let _admission = admission;
-        let mut guard = manager.lock().unwrap_or_else(|e| e.into_inner());
-        let result = guard
-            .execute_with_options_env_for_session(
-                &command,
-                request_cwd.as_deref(),
-                request_timeout.unwrap_or(120_000),
-                true,
-                None,
-                request_tty,
-                Some(policy),
-                request_env,
-                &scope,
-            )
-            .map_err(|error| ApiError::internal(format!("job launch failed: {error}")))?;
-        let task_id = result.task_id.clone().unwrap_or_default();
-        let snapshot = guard
-            .inspect_job(&task_id)
-            .map_err(|error| {
-                ApiError::internal(format!("job launched but is not tracked: {error}"))
-            })?
-            .snapshot;
-        Ok(JobView::new(snapshot, thread_id, &guard))
+        store.with_job_operations(|transaction| {
+            Ok((|| -> Result<(StatusCode, CreateJobResponse), ApiError> {
+                let id = request
+                    .operation_key
+                    .as_deref()
+                    .map(|key| transaction.operation_id(&thread_id, key))
+                    .transpose()
+                    .map_err(operation_error)?;
+                if let Some(id) = &id
+                    && let Some(mut existing) = transaction.get(id).map_err(operation_error)?
+                {
+                    if Some(&existing.request_fingerprint) != fingerprint.as_ref() {
+                        return Err(ApiError::conflict(
+                            "job operation key belongs to a different request",
+                        )
+                        .with_code("job_operation_mismatch"));
+                    }
+                    let (operation, job) =
+                        project_operation(transaction, &mut existing, &managers, &generation)?;
+                    return Ok((
+                        StatusCode::OK,
+                        CreateJobResponse {
+                            job,
+                            operation: Some(operation),
+                            request_fingerprint: Some(existing.request_fingerprint.clone()),
+                        },
+                    ));
+                }
+                if canonical_job_path(&thread.workspace).as_ref() != Some(&workspace)
+                    || request
+                        .expected_workspace
+                        .as_deref()
+                        .is_some_and(|expected| {
+                            canonical_job_path(std::path::Path::new(expected)).as_ref()
+                                != Some(&workspace)
+                        })
+                {
+                    return Err(ApiError::conflict("job operation workspace changed")
+                        .with_code("job_operation_workspace_changed"));
+                }
+                let raw_cwd = request.cwd.as_deref().unwrap_or(".");
+                let cwd = resolve_job_cwd(&workspace, raw_cwd, thread.trust_mode)?;
+                let cwd_path = std::path::PathBuf::from(&cwd);
+                for mut record in transaction.list().map_err(operation_error)? {
+                    let (operation, _) =
+                        project_operation(transaction, &mut record, &managers, &generation)?;
+                    if overlaps(&workspace, &operation.workspace)
+                        || overlaps(&workspace, &operation.cwd)
+                        || overlaps(&cwd_path, &operation.workspace)
+                        || overlaps(&cwd_path, &operation.cwd)
+                    {
+                        if operation.status == "unknown" {
+                            return Err(ApiError::conflict(
+                                "an earlier job operation is unconfirmed",
+                            )
+                            .with_code("job_operation_unconfirmed"));
+                        }
+                        if keyed
+                            && request.purpose.as_deref() != Some("command")
+                            && operation.status == "running"
+                            && operation.purpose != "command"
+                        {
+                            return Err(ApiError::conflict(
+                                "a workspace script is already running",
+                            )
+                            .with_code("job_operation_busy"));
+                        }
+                    }
+                }
+                runtime_owner
+                    .validate_job_operation_thread(&thread)
+                    .map_err(|_| {
+                        ApiError::conflict("job operation workspace or shell policy changed")
+                            .with_code("job_operation_workspace_changed")
+                    })?;
+                if runtime_owner.job_operation_generation().ok().as_deref()
+                    != Some(generation.as_str())
+                {
+                    return Err(ApiError::internal("job operation owner changed"));
+                }
+                let mut guard = manager
+                    .lock()
+                    .map_err(|_| ApiError::internal("job manager unavailable"))?;
+                guard
+                    .validate_background_job(
+                        &command,
+                        &cwd_path,
+                        request.timeout_ms.unwrap_or(120_000),
+                        &policy,
+                        &request.env,
+                    )
+                    .map_err(|error| ApiError::forbidden(error.to_string()))?;
+                if canonical_job_path(&thread.workspace).as_ref() != Some(&workspace)
+                    || canonical_job_path(&cwd_path).as_ref() != Some(&cwd_path)
+                {
+                    return Err(ApiError::conflict("job operation workspace changed")
+                        .with_code("job_operation_workspace_changed"));
+                }
+                let mut record = if let Some(id) = id {
+                    let now = chrono::Utc::now().timestamp_millis();
+                    let record = JobOperationRecord {
+                        schema_version: 1,
+                        operation: JobOperation {
+                            id,
+                            store_id,
+                            thread_id: thread_id.clone(),
+                            workspace,
+                            cwd: cwd_path,
+                            purpose: request.purpose.clone().expect("keyed purpose"),
+                            status: "unknown".to_string(),
+                            job_id: None,
+                            job_status: None,
+                            exit_code: None,
+                            created_at_ms: now,
+                            updated_at_ms: now,
+                        },
+                        request_fingerprint: fingerprint.expect("keyed fingerprint"),
+                        owner_generation: generation.clone(),
+                    };
+                    transaction.save(&record).map_err(operation_error)?;
+                    Some(record)
+                } else {
+                    None
+                };
+                if runtime_owner.job_operation_generation().ok().as_deref()
+                    != Some(generation.as_str())
+                {
+                    return Err(ApiError::internal("job operation owner changed")
+                        .with_code("job_operation_unconfirmed"));
+                }
+                runtime_owner
+                    .validate_job_operation_thread(&thread)
+                    .map_err(|_| {
+                        ApiError::internal("job operation thread changed")
+                            .with_code("job_operation_unconfirmed")
+                    })?;
+                let result = match guard.execute_with_options_env_for_session(
+                    &command,
+                    Some(&cwd),
+                    request.timeout_ms.unwrap_or(120_000),
+                    true,
+                    None,
+                    request.tty,
+                    Some(policy),
+                    request.env,
+                    &scope,
+                ) {
+                    Ok(result) => result,
+                    Err(error) if error.downcast_ref::<ShellLaunchNotStarted>().is_some() => {
+                        if let Some(record) = &mut record {
+                            record.operation.status = "rejected".to_string();
+                            record.operation.updated_at_ms = chrono::Utc::now()
+                                .timestamp_millis()
+                                .max(record.operation.created_at_ms);
+                            transaction.save(record).map_err(|_| {
+                                ApiError::internal("job launch could not be confirmed")
+                                    .with_code("job_operation_unconfirmed")
+                            })?;
+                            return Err(ApiError::conflict("job did not start")
+                                .with_code("job_operation_rejected"));
+                        }
+                        return Err(ApiError::internal("job did not start"));
+                    }
+                    Err(_) => {
+                        return Err(ApiError::internal("job launch could not be confirmed")
+                            .with_code("job_operation_unconfirmed"));
+                    }
+                };
+                let task_id = result.task_id.ok_or_else(|| {
+                    ApiError::internal("job launch could not be confirmed")
+                        .with_code("job_operation_unconfirmed")
+                })?;
+                if record.is_some() {
+                    guard.pin_job_receipt(&task_id).map_err(|_| {
+                        ApiError::internal("job launch could not be confirmed")
+                            .with_code("job_operation_unconfirmed")
+                    })?;
+                }
+                let snapshot = guard
+                    .inspect_job(&task_id)
+                    .map_err(|_| {
+                        ApiError::internal("job launch could not be confirmed")
+                            .with_code("job_operation_unconfirmed")
+                    })?
+                    .snapshot;
+                let job = JobView::new(snapshot, thread_id.clone(), &guard);
+                if let Some(record) = &mut record {
+                    record.operation.job_id = Some(task_id);
+                    record.operation.status = "running".to_string();
+                    record.operation.updated_at_ms = chrono::Utc::now()
+                        .timestamp_millis()
+                        .max(record.operation.created_at_ms);
+                    transaction.save(record).map_err(operation_error)?;
+                    observe_operation(
+                        store.clone(),
+                        std::sync::Arc::downgrade(&runtime_owner),
+                        std::sync::Arc::downgrade(&manager),
+                        thread_id,
+                        generation,
+                        record.operation.id.clone(),
+                    );
+                }
+                Ok((
+                    StatusCode::CREATED,
+                    CreateJobResponse {
+                        job: Some(job),
+                        request_fingerprint: record
+                            .as_ref()
+                            .map(|record| record.request_fingerprint.clone()),
+                        operation: record.map(|record| record.operation),
+                    },
+                ))
+            })())
+        })
     })
     .await
-    .map_err(|_| ApiError::internal("job launch failed"))??;
-    Ok((StatusCode::CREATED, Json(CreateJobResponse { job })))
+    .map_err(|_| ApiError::internal("job launch failed"))?
+    .map_err(operation_error)??;
+    Ok((result.0, Json(result.1)))
 }
 
 #[derive(Debug, Serialize)]
@@ -593,16 +1166,65 @@ pub(super) async fn kill_thread_job(
     if job_id.len() > JOB_ID_MAX_BYTES {
         return Err(ApiError::not_found("job not found"));
     }
+    let admission = state.runtime_threads.session_checkpoint_guard().await;
     let manager = thread_manager(&state, &thread_id, false).await?;
+    let store = state.runtime_threads.job_operation_store();
+    let generation = state
+        .runtime_threads
+        .job_operation_generation()
+        .map_err(operation_error)?;
+    let runtime_owner = state.runtime_threads.clone();
     tokio::task::spawn_blocking(move || {
+        let _admission = admission;
+        if runtime_owner.job_operation_generation().ok().as_deref() != Some(generation.as_str()) {
+            return Err(ApiError::internal("job operation owner changed"));
+        }
         let mut guard = manager.lock().unwrap_or_else(|e| e.into_inner());
         let result = guard.kill(&job_id).map_err(map_job_err)?;
         guard.mark_cancel_reported(&result);
         let snapshot = guard.inspect_job(&job_id).map_err(map_job_err)?.snapshot;
-        Ok(Json(KillJobResponse {
-            job: JobView::new(snapshot, thread_id, &guard),
-            result,
-        }))
+        let pinned = guard.job_receipt_witness(&job_id).is_some();
+        let job = JobView::new(snapshot, thread_id.clone(), &guard);
+        drop(guard);
+        if !pinned {
+            return Ok(Json(KillJobResponse { job, result }));
+        }
+        let managers = vec![(thread_id.clone(), manager)];
+        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
+        loop {
+            let settlement = store.with_job_operations(|transaction| {
+                let mut records: Vec<_> = transaction
+                    .list()?
+                    .into_iter()
+                    .filter(|record| {
+                        record.operation.thread_id == thread_id
+                            && record.operation.job_id.as_deref() == Some(job_id.as_str())
+                    })
+                    .collect();
+                anyhow::ensure!(records.len() <= 1, "duplicate job operation binding");
+                let Some(mut record) = records.pop() else {
+                    return Ok(false);
+                };
+                let (operation, _) =
+                    project_operation(transaction, &mut record, &managers, &generation)
+                        .map_err(|_| anyhow::anyhow!("job settlement unavailable"))?;
+                Ok(operation.status == "completed")
+            });
+            match settlement {
+                Ok(true) => return Ok(Json(KillJobResponse { job, result })),
+                Ok(false) => {}
+                Err(error) if error.downcast_ref::<JobOperationsBusy>().is_some() => {}
+                Err(_) => {
+                    return Err(ApiError::internal("job stop could not be confirmed")
+                        .with_code("job_operation_unconfirmed"));
+                }
+            }
+            if std::time::Instant::now() >= deadline {
+                return Err(ApiError::internal("job stop could not be confirmed")
+                    .with_code("job_operation_unconfirmed"));
+            }
+            std::thread::sleep(std::time::Duration::from_millis(50));
+        }
     })
     .await
     .map_err(|_| ApiError::internal("job kill failed"))?
@@ -659,8 +1281,8 @@ mod tests {
         }
     }
 
-    #[tokio::test]
-    async fn job_cwd_resolves_against_the_thread_workspace() {
+    #[test]
+    fn job_cwd_resolves_against_the_thread_workspace() {
         let tmp = tempfile::tempdir().unwrap();
         let workspace = tmp.path().join("workspace");
         std::fs::create_dir_all(workspace.join("packages/app")).unwrap();
@@ -669,40 +1291,30 @@ mod tests {
 
         // Relative and absolute requests both come back as the canonical
         // directory that was checked, never the caller's spelling.
-        let resolved = resolve_job_cwd(&workspace, "packages/app", false)
-            .await
-            .unwrap();
+        let resolved = resolve_job_cwd(&workspace, "packages/app", false).unwrap();
         assert!(std::path::Path::new(&resolved).is_absolute(), "{resolved}");
         assert_eq!(resolved, canonical(&workspace.join("packages/app")));
         let absolute = workspace.join("packages");
         assert_eq!(
-            resolve_job_cwd(&workspace, &absolute.to_string_lossy(), false)
-                .await
-                .unwrap(),
+            resolve_job_cwd(&workspace, &absolute.to_string_lossy(), false).unwrap(),
             canonical(&absolute)
         );
 
-        let missing = resolve_job_cwd(&workspace, "packages/none", false)
-            .await
-            .unwrap_err();
+        let missing = resolve_job_cwd(&workspace, "packages/none", false).unwrap_err();
         assert_eq!(missing.status, StatusCode::BAD_REQUEST);
         let outside_raw = outside.to_string_lossy().into_owned();
         for raw in [outside_raw.as_str(), "../outside"] {
-            let error = resolve_job_cwd(&workspace, raw, false).await.unwrap_err();
+            let error = resolve_job_cwd(&workspace, raw, false).unwrap_err();
             assert_eq!(error.status, StatusCode::FORBIDDEN, "{raw}");
         }
         #[cfg(unix)]
         {
             std::os::unix::fs::symlink(&outside, workspace.join("escape")).unwrap();
-            let error = resolve_job_cwd(&workspace, "escape", false)
-                .await
-                .unwrap_err();
+            let error = resolve_job_cwd(&workspace, "escape", false).unwrap_err();
             assert_eq!(error.status, StatusCode::FORBIDDEN);
         }
         assert_eq!(
-            resolve_job_cwd(&workspace, "../outside", true)
-                .await
-                .unwrap(),
+            resolve_job_cwd(&workspace, "../outside", true).unwrap(),
             canonical(&outside)
         );
     }
@@ -711,8 +1323,8 @@ mod tests {
     /// checked: retargeting the requested symlink afterwards does not move
     /// the child out of the workspace.
     #[cfg(unix)]
-    #[tokio::test]
-    async fn job_cwd_launches_in_the_checked_directory_after_a_retarget() {
+    #[test]
+    fn job_cwd_launches_in_the_checked_directory_after_a_retarget() {
         let tmp = tempfile::tempdir().unwrap();
         let workspace = tmp.path().join("workspace");
         let inside = workspace.join("inside");
@@ -722,7 +1334,7 @@ mod tests {
         let link = workspace.join("link");
         std::os::unix::fs::symlink(&inside, &link).unwrap();
 
-        let resolved = resolve_job_cwd(&workspace, "link", false).await.unwrap();
+        let resolved = resolve_job_cwd(&workspace, "link", false).unwrap();
         assert_eq!(resolved, canonical(&inside));
 
         std::fs::remove_file(&link).unwrap();
@@ -743,8 +1355,8 @@ mod tests {
     /// than lossily renamed into a `U+FFFD` sibling. Linux only: macOS and
     /// Windows filesystems do not store non-UTF-8 names.
     #[cfg(target_os = "linux")]
-    #[tokio::test]
-    async fn job_cwd_refuses_a_directory_name_that_is_not_utf8() {
+    #[test]
+    fn job_cwd_refuses_a_directory_name_that_is_not_utf8() {
         use std::os::unix::ffi::OsStrExt as _;
         let tmp = tempfile::tempdir().unwrap();
         let workspace = tmp.path().join("workspace");
@@ -753,9 +1365,7 @@ mod tests {
         std::fs::create_dir_all(workspace.join("dir-\u{fffd}")).unwrap();
         std::os::unix::fs::symlink(workspace.join(raw_name), workspace.join("alias")).unwrap();
 
-        let error = resolve_job_cwd(&workspace, "alias", false)
-            .await
-            .unwrap_err();
+        let error = resolve_job_cwd(&workspace, "alias", false).unwrap_err();
         assert_eq!(error.status, StatusCode::BAD_REQUEST);
     }
 

crates/app-server/src/runtime_api/tests.rs

@@ -24489,7 +24489,20 @@ async fn thread_notices_serve_list_and_ack() -> Result<()> {
 #[tokio::test]
 async fn provider_models_refresh_uses_exact_configured_route_without_switching() -> Result<()> {
     let _env = codewhale_engine::test_support::lock_test_env();
+    let _live = codewhale_engine::provider_lake::lock_live_snapshot();
+    codewhale_engine::provider_lake::clear_live_snapshot();
     let root = tempfile::tempdir()?;
+    let catalog_path = root.path().join("models-dev.json");
+    fs::write(
+        &catalog_path,
+        json!({"models": {}, "providers": {"togetherai": {"id": "togetherai", "models": {
+            "catalog-refresh-only": {"id": "catalog-refresh-only", "modalities": {"input": ["text"], "output": ["text"]}}
+        }}}}).to_string(),
+    )?;
+    let _catalog = codewhale_engine::test_support::EnvVarGuard::set(
+        "CODEWHALE_MODELS_DEV_PATH",
+        &catalog_path,
+    );
     let _home = codewhale_engine::test_support::EnvVarGuard::set("CODEWHALE_HOME", root.path());
     let _cli =
         codewhale_engine::test_support::EnvVarGuard::remove(codewhale_config::CLI_API_KEY_ENV);
@@ -24549,7 +24562,7 @@ api_key = "refresh-fixture-key"
     }
     let response = client
         .post(format!(
-            "http://{addr}/v1/providers/custom/models/refresh?model_provider_id=second"
+            "http://{addr}/v1/providers/custom/models/refresh?model_provider_id=second&force_catalog=true"
         ))
         .send()
         .await?;
@@ -24557,6 +24570,9 @@ api_key = "refresh-fixture-key"
     let receipt: Value = response.json().await?;
     assert_eq!(receipt["outcome"], "updated");
     assert_eq!(receipt["model_count"], 1);
+    assert_eq!(receipt["public_catalog"]["freshness"], "live");
+    assert_eq!(receipt["public_catalog"]["outcome"], "updated");
+    assert!(receipt["public_catalog"]["error"].is_null());
     assert!(!receipt.to_string().contains("refresh-fixture-key"));
     let catalog: Value = client
         .get(format!(
@@ -24575,6 +24591,7 @@ api_key = "refresh-fixture-key"
     );
     assert_eq!(get_config(&client, &addr).await["provider"], "first");
     handle.abort();
+    codewhale_engine::provider_lake::clear_live_snapshot();
     Ok(())
 }
 

crates/app-server/src/runtime_api/tests/control_socket_selection.rs

@@ -12,13 +12,13 @@ fn inputs() -> codewhale_paths::daemon_socket::SocketPathInputs {
         explicit: None,
         codewhale_home_override: None,
         xdg_runtime_dir: None,
-        user_home: Some(PathBuf::from("/home/whale")),
+        user_home: Some(PathBuf::from("/cw-test-home/whale")),
         macos: false,
     }
 }
 
 fn default_socket() -> PathBuf {
-    PathBuf::from("/home/whale/.codewhale/run/daemon.sock")
+    PathBuf::from("/cw-test-home/whale/.codewhale/run/daemon.sock")
 }
 
 fn temp_root(tag: &str) -> PathBuf {
@@ -153,7 +153,7 @@ async fn the_runtime_dir_override_reaches_the_control_endpoint() -> Result<()> {
         resolved.display()
     );
     assert!(
-        resolved.starts_with(&home),
+        resolved.starts_with(codewhale_paths::resolve_root_owned_ancestors(&home)),
         "the derived endpoint stays beside the well-known one, got {}",
         resolved.display()
     );

crates/config/src/private_directory.rs

@@ -6,6 +6,8 @@
 use crate::windows_identity::{CurrentWindowsUser, WindowsLocalAllocation};
 use anyhow::{Context, Result, bail};
 #[cfg(unix)]
+use codewhale_paths::resolve_root_owned_ancestors as resolve_ancestors;
+#[cfg(unix)]
 use std::ffi::CString;
 use std::fs::{self, File};
 use std::io::Write as _;
@@ -572,22 +574,6 @@ fn validate_private_basename(name: &str) -> Result<()> {
     Ok(())
 }
 
-#[cfg(unix)]
-fn resolve_ancestors(directory: &Path) -> std::path::PathBuf {
-    use std::os::unix::fs::MetadataExt as _;
-
-    let mut resolved = std::path::PathBuf::new();
-    for component in directory.components() {
-        resolved.push(component);
-        let root_owned_link = std::fs::symlink_metadata(&resolved)
-            .is_ok_and(|meta| meta.file_type().is_symlink() && meta.uid() == 0);
-        if root_owned_link && let Ok(real) = resolved.canonicalize() {
-            resolved = real;
-        }
-    }
-    resolved
-}
-
 #[cfg(unix)]
 fn open_owned_directory(
     directory: &Path,

crates/engine/src/runtime_threads.rs

@@ -84,7 +84,7 @@ const AGENT_MAIL_OWNER_FILE: &str = "owner.json";
 /// across all of them is what lets a switch adopt an existing store (#6207);
 /// `adoptable_empty_store_reports_nothing_to_abandon` pins this list against
 /// `open` by asserting each directory is load-bearing on its own.
-const RUNTIME_STORE_WORK_DIRS: [&str; 7] = [
+const RUNTIME_STORE_WORK_DIRS: [&str; 8] = [
     "threads",
     "turns",
     "items",
@@ -92,6 +92,7 @@ const RUNTIME_STORE_WORK_DIRS: [&str; 7] = [
     "goals",
     "agent-mail",
     "turn-operations",
+    "job-operations",
 ];
 const TURN_OPERATION_BINDING_SCHEMA_VERSION: u32 = 2;
 const RUNTIME_USAGE_PARTITION_SUMMARY: &str = "Provider usage receipt partition";
@@ -1586,6 +1587,7 @@ pub struct RuntimeThreadStore {
     goal_mutation: Arc<parking_lot::Mutex<()>>,
     mail_dir: PathBuf,
     turn_operations_dir: PathBuf,
+    job_operations_dir: PathBuf,
     owner_id: String,
     state_path: PathBuf,
     event_lock_path: PathBuf,
@@ -1665,6 +1667,7 @@ impl RuntimeThreadStore {
         let goals_dir = root.join("goals");
         let mail_dir = root.join("agent-mail");
         let turn_operations_dir = root.join("turn-operations");
+        let job_operations_dir = root.join("job-operations");
         ensure_runtime_store_dir(&threads_dir)?;
         ensure_runtime_store_dir(&turns_dir)?;
         ensure_runtime_store_dir(&items_dir)?;
@@ -1672,6 +1675,7 @@ impl RuntimeThreadStore {
         ensure_runtime_store_dir(&goals_dir)?;
         ensure_runtime_store_dir(&mail_dir)?;
         ensure_runtime_store_dir(&turn_operations_dir)?;
+        ensure_runtime_store_dir(&job_operations_dir)?;
         let state_path = root.join("state.json");
         let owner_path = root.join(AGENT_MAIL_OWNER_FILE);
         let event_lock_path = root.join(EVENT_TRANSACTION_LOCK_FILE);
@@ -1689,6 +1693,7 @@ impl RuntimeThreadStore {
             goals_dir,
             mail_dir,
             turn_operations_dir,
+            job_operations_dir,
             owner_id,
             state_path,
             event_lock_path,
@@ -1745,6 +1750,7 @@ impl RuntimeThreadStore {
             goals_dir: root.join("goals"),
             mail_dir: root.join("agent-mail"),
             turn_operations_dir: root.join("turn-operations"),
+            job_operations_dir: root.join("job-operations"),
             owner_id: String::new(),
             state_path: root.join("state.json"),
             event_lock_path: root.join(EVENT_TRANSACTION_LOCK_FILE),
@@ -5975,6 +5981,7 @@ fn remove_file_if_exists(path: &Path) -> Result<()> {
     }
 }
 
+pub mod job_operations;
 pub mod manager;
 mod turn_artifacts;
 pub use turn_artifacts::{

crates/engine/src/runtime_threads/job_operations.rs

@@ -0,0 +1,236 @@
+use super::*;
+use ring::hmac;
+use std::sync::{Mutex as StdMutex, TryLockError};
+
+static JOB_OPERATION_MUTEX: StdMutex<()> = StdMutex::new(());
+
+#[derive(Debug, thiserror::Error)]
+#[error("job operation admission is busy")]
+pub struct JobOperationsBusy;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct JobOperation {
+    pub id: String,
+    pub store_id: String,
+    pub thread_id: String,
+    pub workspace: PathBuf,
+    pub cwd: PathBuf,
+    pub purpose: String,
+    pub status: String,
+    pub job_id: Option<String>,
+    pub job_status: Option<String>,
+    pub exit_code: Option<i64>,
+    pub created_at_ms: i64,
+    pub updated_at_ms: i64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct JobOperationRecord {
+    pub schema_version: u32,
+    pub operation: JobOperation,
+    pub request_fingerprint: String,
+    pub owner_generation: String,
+}
+
+pub struct JobOperationTransaction<'a> {
+    store: &'a RuntimeThreadStore,
+}
+
+pub fn validate_key(key: &str) -> Result<()> {
+    let id = Uuid::parse_str(key).context("invalid job operation key")?;
+    anyhow::ensure!(
+        id.get_version_num() == 4 && id.to_string() == key,
+        "invalid job operation key"
+    );
+    Ok(())
+}
+
+pub fn request_fingerprint(key: &str, request: &[u8]) -> Result<String> {
+    validate_key(key)?;
+    let key = hmac::Key::new(hmac::HMAC_SHA256, key.as_bytes());
+    Ok(hex_digest(hmac::sign(&key, request).as_ref()))
+}
+
+impl RuntimeThreadStore {
+    pub fn with_job_operations<T>(
+        &self,
+        action: impl FnOnce(&JobOperationTransaction<'_>) -> Result<T>,
+    ) -> Result<T> {
+        let _local = match JOB_OPERATION_MUTEX.try_lock() {
+            Ok(guard) => guard,
+            Err(TryLockError::WouldBlock) => return Err(JobOperationsBusy.into()),
+            Err(TryLockError::Poisoned(_)) => bail!("job operation admission unavailable"),
+        };
+        let root = self
+            .job_operations_dir
+            .parent()
+            .context("missing runtime store root")?;
+        checked_existing_runtime_store_dir(&self.job_operations_dir)?;
+        let path = root.join("job-operations.lock");
+        let file = open_runtime_store_file(&path, "job operation lock", |options| {
+            options.create(true).truncate(false).read(true).write(true);
+        })?;
+        let mut lock = fd_lock::RwLock::new(file);
+        let guard = match lock.try_write() {
+            Ok(guard) => guard,
+            Err(error)
+                if matches!(
+                    error.kind(),
+                    io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted
+                ) =>
+            {
+                return Err(JobOperationsBusy.into());
+            }
+            Err(error) => return Err(error).context("job operation lock unavailable"),
+        };
+        anyhow::ensure!(
+            owner_lock_is_at_path(&guard, &path)?,
+            "job operation lock changed"
+        );
+        action(&JobOperationTransaction { store: self })
+    }
+}
+
+impl JobOperationTransaction<'_> {
+    pub fn operation_id(&self, thread_id: &str, key: &str) -> Result<String> {
+        validated_record_id(thread_id, "job operation thread")?;
+        validate_key(key)?;
+        Ok(codewhale_runtime::hashing::sha256_hex(format!(
+            "runtime-job-operation\u{1f}{}\u{1f}{thread_id}\u{1f}{key}",
+            self.store.owner_id
+        )))
+    }
+
+    fn path(&self, id: &str) -> Result<PathBuf> {
+        validate_sha256_fingerprint(id, "job operation id")?;
+        RuntimeThreadStore::record_path(
+            &self.store.job_operations_dir,
+            id,
+            "json",
+            "job operation id",
+        )
+    }
+
+    fn validate(&self, record: &JobOperationRecord) -> Result<()> {
+        let operation = &record.operation;
+        anyhow::ensure!(
+            record.schema_version == 1,
+            "unsupported job operation schema"
+        );
+        validate_sha256_fingerprint(&operation.id, "job operation id")?;
+        validate_sha256_fingerprint(&record.request_fingerprint, "job request fingerprint")?;
+        validated_record_id(&operation.thread_id, "job operation thread")?;
+        anyhow::ensure!(
+            operation.store_id == self.store.owner_id,
+            "job operation store mismatch"
+        );
+        anyhow::ensure!(
+            operation.workspace.is_absolute() && operation.cwd.is_absolute(),
+            "invalid job operation workspace"
+        );
+        anyhow::ensure!(
+            matches!(
+                operation.purpose.as_str(),
+                "setup" | "run" | "archive" | "command"
+            ),
+            "invalid job purpose"
+        );
+        anyhow::ensure!(
+            matches!(
+                operation.status.as_str(),
+                "running" | "completed" | "rejected" | "unknown"
+            ),
+            "invalid job status"
+        );
+        if let Some(id) = &operation.job_id {
+            validated_record_id(id, "job id")?;
+            anyhow::ensure!(
+                id.starts_with("shell_") && id.len() <= 128,
+                "invalid job id"
+            );
+        }
+        anyhow::ensure!(
+            operation.created_at_ms > 0 && operation.updated_at_ms >= operation.created_at_ms,
+            "invalid job operation timestamps"
+        );
+        anyhow::ensure!(
+            operation.status == "completed"
+                || (operation.job_status.is_none() && operation.exit_code.is_none()),
+            "nonterminal job operation has a terminal result"
+        );
+        anyhow::ensure!(
+            operation.status != "rejected" || operation.job_id.is_none(),
+            "rejected job operation has a job id"
+        );
+        anyhow::ensure!(
+            operation.status != "running" || operation.job_id.is_some(),
+            "running job operation has no job id"
+        );
+        anyhow::ensure!(
+            operation.status != "completed"
+                || (operation.job_id.is_some()
+                    && operation
+                        .job_status
+                        .as_deref()
+                        .is_some_and(|status| matches!(
+                            status,
+                            "Completed" | "Failed" | "Killed" | "TimedOut"
+                        ))),
+            "completed job operation has no terminal result"
+        );
+        anyhow::ensure!(
+            Uuid::parse_str(&record.owner_generation).is_ok_and(|id| id.get_version_num() == 4),
+            "invalid job owner generation"
+        );
+        Ok(())
+    }
+
+    pub fn get(&self, id: &str) -> Result<Option<JobOperationRecord>> {
+        let path = self.path(id)?;
+        let raw = match read_store_file(&path) {
+            Ok(raw) => raw,
+            Err(error)
+                if error
+                    .downcast_ref::<io::Error>()
+                    .is_some_and(|error| error.kind() == io::ErrorKind::NotFound) =>
+            {
+                return Ok(None);
+            }
+            Err(error) => return Err(error),
+        };
+        let record: JobOperationRecord = serde_json::from_str(&raw)?;
+        self.validate(&record)?;
+        anyhow::ensure!(record.operation.id == id, "job operation path mismatch");
+        Ok(Some(record))
+    }
+
+    pub fn list(&self) -> Result<Vec<JobOperationRecord>> {
+        let directory = checked_existing_runtime_store_dir(&self.store.job_operations_dir)?;
+        let mut records = Vec::new();
+        for entry in fs::read_dir(directory)? {
+            let path = entry?.path();
+            if path.extension().is_none_or(|ext| ext != "json") {
+                continue;
+            }
+            let id = path
+                .file_stem()
+                .and_then(|name| name.to_str())
+                .context("invalid job operation filename")?;
+            records.push(self.get(id)?.context("job operation disappeared")?);
+        }
+        records.sort_by(|a, b| {
+            a.operation
+                .created_at_ms
+                .cmp(&b.operation.created_at_ms)
+                .then_with(|| a.operation.id.cmp(&b.operation.id))
+        });
+        Ok(records)
+    }
+
+    pub fn save(&self, record: &JobOperationRecord) -> Result<()> {
+        self.validate(record)?;
+        write_json_atomic(&self.path(&record.operation.id)?, record)
+    }
+}

crates/engine/src/runtime_threads/manager.rs

@@ -4343,6 +4343,36 @@ impl RuntimeThreadManager {
         &self.workspace
     }
 
+    pub fn job_operation_store_id(&self) -> &str {
+        &self.store.owner_id
+    }
+
+    pub fn job_operation_store(&self) -> RuntimeThreadStore {
+        self.store.clone()
+    }
+
+    pub fn job_operation_generation(&self) -> Result<String> {
+        #[cfg(any(unix, windows))]
+        {
+            Ok(self.capture_control_owner()?.1)
+        }
+        #[cfg(not(any(unix, windows)))]
+        {
+            bail!("job operation owner unavailable")
+        }
+    }
+
+    pub fn validate_job_operation_thread(&self, thread: &ThreadRecord) -> Result<()> {
+        let current = self.store.load_thread(&thread.id)?;
+        anyhow::ensure!(
+            current.workspace == thread.workspace
+                && current.allow_shell == thread.allow_shell
+                && current.trust_mode == thread.trust_mode,
+            "thread workspace or shell policy changed"
+        );
+        Ok(())
+    }
+
     pub fn mcp_runtime_servers(&self, workspace: &Path) -> crate::mcp::SharedRuntimeMcpServers {
         self.runtime_mcp_servers
             .lock()
@@ -7242,34 +7272,42 @@ impl RuntimeThreadManager {
             )
             .await
         {
-            self.active.lock().await.shell_managers.remove(&thread.id);
-            let _ = self.store.remove_thread(&thread.id);
             return Err(error);
         }
         Ok(thread)
     }
 
     pub async fn discard_empty_thread(&self, thread_id: &str) -> Result<()> {
+        let _admission = self.session_checkpoint_guard().await;
         let mut active = self.active.lock().await;
         if active.engines.contains_key(thread_id) {
             bail!("cannot discard a loaded Runtime thread");
         }
-        let _thread_mutation = self.store.thread_mutation.lock();
-        let thread = self.store.load_thread(thread_id)?;
-        if thread.latest_turn_id.is_some() {
-            bail!("cannot discard a Runtime thread that owns turns");
-        }
-        // Drop the thread's shell authority with it: the manager owns any
-        // API-created jobs, and dropping the last handle kills them.
-        active.shell_managers.remove(thread_id);
+        self.store.with_job_operations(|transaction| {
+            anyhow::ensure!(
+                !transaction
+                    .list()?
+                    .iter()
+                    .any(|record| record.operation.thread_id == thread_id
+                        && !matches!(record.operation.status.as_str(), "completed" | "rejected")),
+                "thread has a running or unconfirmed job operation"
+            );
+            let _thread_mutation = self.store.thread_mutation.lock();
+            let thread = self.store.load_thread(thread_id)?;
+            if thread.latest_turn_id.is_some() {
+                bail!("cannot discard a Runtime thread that owns turns");
+            }
+            active.shell_managers.remove(thread_id);
+            self.store.remove_thread(thread_id)?;
+            Ok(())
+        })?;
         drop(active);
-        self.store.remove_thread(thread_id)?;
-        // A deleted conversation keeps no session grants behind it.
         self.take_approval_grants(thread_id);
         Ok(())
     }
 
     pub async fn delete_thread(&self, thread_id: &str) -> Result<ThreadRecord> {
+        let _admission = self.session_checkpoint_guard().await;
         let mut active = self.active.lock().await;
         if active
             .engines
@@ -7278,21 +7316,32 @@ impl RuntimeThreadManager {
         {
             bail!("thread {thread_id} already has an active turn; stop it before deleting");
         }
-        let _thread_mutation = self.store.thread_mutation.lock();
-        let thread = self.store.load_thread(thread_id)?;
-        active.lru.retain(|id| id != thread_id);
-        active.shell_managers.remove(thread_id);
-        let evicted = active.engines.remove(thread_id).map(|state| state.engine);
-        for turn in self.store.list_turns_for_thread(thread_id)? {
-            for item_id in &turn.item_ids {
-                self.store.remove_item(item_id)?;
+        let (thread, evicted) = self.store.with_job_operations(|transaction| {
+            anyhow::ensure!(
+                !transaction
+                    .list()?
+                    .iter()
+                    .any(|record| record.operation.thread_id == thread_id
+                        && !matches!(record.operation.status.as_str(), "completed" | "rejected")),
+                "thread has a running or unconfirmed job operation"
+            );
+            let _thread_mutation = self.store.thread_mutation.lock();
+            let thread = self.store.load_thread(thread_id)?;
+            active.lru.retain(|id| id != thread_id);
+            active.shell_managers.remove(thread_id);
+            let evicted = active.engines.remove(thread_id).map(|state| state.engine);
+            for turn in self.store.list_turns_for_thread(thread_id)? {
+                for item_id in &turn.item_ids {
+                    self.store.remove_item(item_id)?;
+                }
+                self.store.remove_turn(&turn.id)?;
             }
-            self.store.remove_turn(&turn.id)?;
-        }
-        remove_file_if_exists(&self.store.events_path(thread_id)?)?;
-        remove_file_if_exists(&self.store.goal_path(thread_id)?)?;
-        self.store.remove_seed_journal(thread_id)?;
-        self.store.remove_thread(thread_id)?;
+            remove_file_if_exists(&self.store.events_path(thread_id)?)?;
+            remove_file_if_exists(&self.store.goal_path(thread_id)?)?;
+            self.store.remove_seed_journal(thread_id)?;
+            self.store.remove_thread(thread_id)?;
+            Ok((thread, evicted))
+        })?;
         drop(active);
         if let Some(engine) = evicted {
             engine.cancel_with_reason(crate::core::engine::CancelReason::Internal);
@@ -10547,10 +10596,17 @@ impl RuntimeThreadManager {
         })();
 
         if let Err(persistence_error) = persistence {
-            let mut cleanup_errors = Vec::new();
-            if let Err(error) = self.store.remove_thread(&thread.id) {
-                cleanup_errors.push(format!("remove thread: {error}"));
+            let thread_path = self.store.thread_path(&thread.id)?;
+            match thread_path.try_exists() {
+                Ok(false) => {}
+                Ok(true) => bail!(
+                    "Fork publication could not be confirmed; retained its records for recovery: {persistence_error}"
+                ),
+                Err(error) => bail!(
+                    "Fork publication could not be checked; retained its records for recovery: {error}"
+                ),
             }
+            let mut cleanup_errors = Vec::new();
             for turn_id in saved_turn_ids.iter().rev() {
                 if let Err(error) = self.store.remove_turn(turn_id) {
                     cleanup_errors.push(format!("remove turn {turn_id}: {error}"));

crates/engine/src/tools/shell.rs

@@ -11,7 +11,7 @@
 use anyhow::{Context, Result, anyhow};
 use base64::Engine as _;
 use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::fs::File;
 use std::io;
 use std::io::{Read, Write};
@@ -147,6 +147,10 @@ fn validate_shell_working_dir(path: &Path, inherited_session_workspace: bool) ->
     Ok(())
 }
 
+#[derive(Debug, thiserror::Error)]
+#[error("shell launch did not start")]
+pub struct ShellLaunchNotStarted;
+
 /// Status of a shell process.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub enum ShellStatus {
@@ -251,6 +255,32 @@ pub struct ShellJobSnapshot {
     pub owner_session_id: String,
 }
 
+#[derive(Debug, Clone)]
+pub struct ShellReceiptWitness {
+    pub cwd: PathBuf,
+    pub status: ShellStatus,
+    pub exit_code: Option<i64>,
+    pub owner_session_id: String,
+    pub wait_failed: bool,
+    pub terminal_confirmed: bool,
+}
+
+impl ShellReceiptWitness {
+    fn from_shell(shell: &mut BackgroundShell) -> Self {
+        let terminal_confirmed = shell.status != ShellStatus::Running
+            && !shell.wait_failed
+            && shell.terminal_exit_confirmed;
+        Self {
+            cwd: shell.working_dir.clone(),
+            status: shell.status.clone(),
+            exit_code: shell.exit_code,
+            owner_session_id: shell.owner_session_id.clone(),
+            wait_failed: shell.wait_failed,
+            terminal_confirmed,
+        }
+    }
+}
+
 /// Once-only completion event for a tracked background shell job.
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub struct ShellCompletionEvent {
@@ -1152,6 +1182,7 @@ pub struct BackgroundShell {
     /// The terminal status came from a failed `try_wait`, not an observed
     /// exit or signal. Such a run gets no execution receipt (#6689).
     wait_failed: bool,
+    terminal_exit_confirmed: bool,
 }
 
 #[derive(Clone)]
@@ -1315,6 +1346,7 @@ impl BackgroundShell {
         let completed = if let Some(ref mut child) = self.child {
             match child.try_wait() {
                 Ok(Some(status)) => {
+                    self.terminal_exit_confirmed = true;
                     self.exit_code = status.code;
                     self.status = if status.success {
                         ShellStatus::Completed
@@ -1621,7 +1653,7 @@ impl BackgroundShell {
             unregister_pending_persistent_process_group(process_group_id);
         }
         if let Some(ref mut child) = self.child {
-            match child {
+            match &mut *child {
                 ShellChild::Process(proc) => {
                     #[cfg(windows)]
                     {
@@ -1648,6 +1680,9 @@ impl BackgroundShell {
                     let _ = child.wait();
                 }
             }
+            if child.try_wait().is_ok_and(|status| status.is_some()) {
+                self.terminal_exit_confirmed = true;
+            }
         }
         self.status = ShellStatus::Killed;
         self.mark_finished();
@@ -1962,6 +1997,8 @@ pub fn inherited_interactive_terminal_refusal() -> Option<&'static str> {
 pub struct ShellManager {
     processes: HashMap<String, BackgroundShell>,
     stale_jobs: HashMap<String, ShellJobSnapshot>,
+    receipt_pins: HashSet<String>,
+    receipt_witnesses: HashMap<String, ShellReceiptWitness>,
     default_workspace: PathBuf,
     sandbox_manager: SandboxManager,
     sandbox_policy: ExecutionSandboxPolicy,
@@ -1994,6 +2031,28 @@ impl std::fmt::Debug for ShellManager {
 }
 
 impl ShellManager {
+    pub fn pin_job_receipt(&mut self, task_id: &str) -> Result<()> {
+        anyhow::ensure!(self.processes.contains_key(task_id), "job not tracked");
+        self.receipt_pins.insert(task_id.to_string());
+        Ok(())
+    }
+
+    pub fn job_receipt_witness(&mut self, task_id: &str) -> Option<ShellReceiptWitness> {
+        if !self.receipt_pins.contains(task_id) {
+            return None;
+        }
+        if let Some(shell) = self.processes.get_mut(task_id) {
+            shell.poll();
+            return Some(ShellReceiptWitness::from_shell(shell));
+        }
+        self.receipt_witnesses.get(task_id).cloned()
+    }
+
+    pub fn release_job_receipt(&mut self, task_id: &str) {
+        self.receipt_pins.remove(task_id);
+        self.receipt_witnesses.remove(task_id);
+    }
+
     fn require_session_owner(&self, task_id: &str, active_session_id: &str) -> Result<()> {
         let owned = self.processes.get(task_id).is_some_and(|shell| {
             !active_session_id.is_empty() && shell.owner_session_id == active_session_id
@@ -2013,6 +2072,8 @@ impl ShellManager {
         Self {
             processes: HashMap::new(),
             stale_jobs: HashMap::new(),
+            receipt_pins: HashSet::new(),
+            receipt_witnesses: HashMap::new(),
             default_workspace: workspace,
             sandbox_manager: SandboxManager::new(),
             sandbox_policy: ExecutionSandboxPolicy::default(),
@@ -2088,6 +2149,7 @@ impl ShellManager {
                 last_lifecycle_status: None,
                 last_lifecycle_bytes: 0,
                 wait_failed: false,
+                terminal_exit_confirmed: false,
             },
         );
     }
@@ -2148,6 +2210,29 @@ impl ShellManager {
         self.sandbox_manager.prepare(&spec)
     }
 
+    pub fn validate_background_job(
+        &self,
+        command: &str,
+        cwd: &Path,
+        timeout_ms: u64,
+        policy: &ExecutionSandboxPolicy,
+        env: &HashMap<String, String>,
+    ) -> Result<()> {
+        validate_shell_working_dir(cwd, false)?;
+        let spec = CommandSpec::shell(
+            command,
+            cwd.to_path_buf(),
+            Duration::from_millis(timeout_ms.clamp(1_000, 600_000)),
+        )
+        .with_policy(policy.clone())
+        .with_env(env.clone());
+        let prepared = self.sandbox_manager.prepare(&spec);
+        if matches!(spec.sandbox_policy, ExecutionSandboxPolicy::ReadOnly { .. }) {
+            require_native_readonly_execution(&prepared)?;
+        }
+        Ok(())
+    }
+
     fn register_shell_wait(&mut self, session_id: &str) -> Arc<AtomicBool> {
         self.active_waits
             .retain(|(_, wait)| wait.strong_count() > 0);
@@ -2346,7 +2431,10 @@ impl ShellManager {
         codewhale_sandbox::shell_dispatcher::ShellDispatcher::log_exec(command);
 
         let work_dir = working_dir.map_or_else(|| self.default_workspace.clone(), PathBuf::from);
-        validate_shell_working_dir(&work_dir, working_dir.is_none())?;
+        validate_shell_working_dir(&work_dir, working_dir.is_none()).map_err(|error| {
+            let message = error.to_string();
+            error.context(ShellLaunchNotStarted).context(message)
+        })?;
 
         let timeout_ms = timeout_ms.clamp(timeout_bounds_ms.0, timeout_bounds_ms.1);
 
@@ -2380,7 +2468,10 @@ impl ShellManager {
         {
             // Arbitrary code with a read-only policy needs kernel enforcement;
             // only the separately hardened argv subset may run without it.
-            require_native_readonly_execution(&exec_env)?;
+            require_native_readonly_execution(&exec_env).map_err(|error| {
+                let message = error.to_string();
+                error.context(ShellLaunchNotStarted).context(message)
+            })?;
         }
 
         if background {
@@ -2733,7 +2824,7 @@ impl ShellManager {
             origin_turn_id,
             work_lifecycle,
         } = spawn_context;
-        let task_id = format!("shell_{}", &Uuid::new_v4().to_string()[..8]);
+        let task_id = format!("shell_{}", Uuid::new_v4().simple());
         let mut spawn_guard =
             ShellSpawnIntentGuard::new(work_lifecycle, &task_id, original_command);
         // The guard owns the registration outcome: when bookkeeping could not
@@ -2791,6 +2882,7 @@ impl ShellManager {
                         pixel_width: 0,
                         pixel_height: 0,
                     })
+                    .context(ShellLaunchNotStarted)
                     .context("Failed to open PTY")?;
 
                 let mut child = exec_env
@@ -2978,6 +3070,7 @@ impl ShellManager {
             last_lifecycle_status: None,
             last_lifecycle_bytes: 0,
             wait_failed: false,
+            terminal_exit_confirmed: false,
         };
 
         #[cfg(unix)]
@@ -3755,14 +3848,20 @@ impl ShellManager {
     /// age out too: jobs launched over the Runtime API, or owned by a session
     /// that is no longer active, are never drained.
     pub fn cleanup(&mut self, max_age: Duration) {
-        self.processes.retain(|_, shell| {
+        let pins = &self.receipt_pins;
+        let witnesses = &mut self.receipt_witnesses;
+        self.processes.retain(|id, shell| {
             if shell.status == ShellStatus::Running {
                 return true;
             }
             shell.mark_finished();
-            shell
+            let retain = shell
                 .finished_at
-                .is_none_or(|finished| finished.elapsed() < max_age)
+                .is_none_or(|finished| finished.elapsed() < max_age);
+            if !retain && pins.contains(id) {
+                witnesses.insert(id.clone(), ShellReceiptWitness::from_shell(shell));
+            }
+            retain
         });
         self.enforce_finished_job_bounds();
     }
@@ -3819,7 +3918,12 @@ impl ShellManager {
             {
                 break;
             }
-            self.processes.remove(&id);
+            if let Some(mut shell) = self.processes.remove(&id)
+                && self.receipt_pins.contains(&id)
+            {
+                self.receipt_witnesses
+                    .insert(id, ShellReceiptWitness::from_shell(&mut shell));
+            }
             remaining_count -= 1;
             remaining_bytes = remaining_bytes.saturating_sub(bytes);
         }
@@ -6464,7 +6568,7 @@ impl ToolSpec for BashTool {
                 .lock()
                 .map_err(|_| ToolError::execution_failed("shell manager lock poisoned"))?;
             let work_lifecycle = shell_work_lifecycle_from_context(context);
-            let task_id = format!("shell_{}", &Uuid::new_v4().to_string()[..8]);
+            let task_id = format!("shell_{}", Uuid::new_v4().simple());
             let mut spawn_guard = ShellSpawnIntentGuard::new(work_lifecycle, &task_id, command);
             // Only a registered operation is observed (#6435).
             let work_lifecycle = spawn_guard.lifecycle.clone();

crates/engine/src/tools/shell/tests.rs

@@ -4308,6 +4308,7 @@ fn killed_shell_does_not_wait_for_blocked_reader_threads() {
         last_lifecycle_status: None,
         last_lifecycle_bytes: 0,
         wait_failed: false,
+        terminal_exit_confirmed: false,
     };
 
     let started = std::time::Instant::now();

crates/paths/src/daemon_socket.rs

@@ -54,6 +54,11 @@ pub enum DaemonSocketError {
     /// `CODEWHALE_HOME` is set but not a usable absolute path.
     #[error("invalid CODEWHALE_HOME override: {0}")]
     InvalidHomeOverride(String),
+    #[error("cannot use {variable} for the runtime endpoint: {reason}")]
+    InvalidRuntimeRoot {
+        variable: &'static str,
+        reason: &'static str,
+    },
     /// Unix socket paths are limited to roughly one hundred bytes.
     #[error("daemon socket path {} is {len} bytes; this platform allows at most {max}", path.display())]
     PathTooLong {
@@ -110,7 +115,7 @@ pub enum DaemonSocketError {
 }
 
 /// Inputs to [`resolve_socket_path`], separated from the environment so the
-/// precedence rules are a pure, testable function.
+/// precedence rules can be verified independently of process-wide variables.
 #[derive(Debug, Clone, Default)]
 pub struct SocketPathInputs {
     /// `--socket-path`.
@@ -133,30 +138,79 @@ impl SocketPathInputs {
         let xdg_runtime_dir = std::env::var_os("XDG_RUNTIME_DIR")
             .filter(|value| !value.is_empty())
             .map(PathBuf::from);
+        let user_home = crate::user_home();
         Ok(Self {
             explicit,
             codewhale_home_override,
             xdg_runtime_dir,
-            user_home: crate::user_home(),
+            user_home,
             macos: cfg!(target_os = "macos"),
         })
     }
 }
 
-/// Apply the precedence rules documented at the module level and enforce the
-/// kernel's path-length limit.
+fn implicit_endpoint_root(
+    _variable: &'static str,
+    root: PathBuf,
+) -> Result<PathBuf, DaemonSocketError> {
+    #[cfg(unix)]
+    {
+        if !root.is_absolute()
+            || root
+                .components()
+                .any(|part| matches!(part, std::path::Component::ParentDir))
+        {
+            return Err(DaemonSocketError::InvalidRuntimeRoot {
+                variable: _variable,
+                reason: "expected an absolute directory path without '..' components",
+            });
+        }
+        let root = crate::resolve_root_owned_ancestors(&root);
+        for ancestor in root.ancestors() {
+            match std::fs::symlink_metadata(ancestor) {
+                Ok(metadata) if metadata.file_type().is_symlink() => {
+                    return Err(DaemonSocketError::InvalidRuntimeRoot {
+                        variable: _variable,
+                        reason: "a non-system symbolic link is not allowed",
+                    });
+                }
+                Err(error) if error.kind() != std::io::ErrorKind::NotFound => {
+                    return Err(DaemonSocketError::Io {
+                        context: "cannot inspect runtime endpoint directory",
+                        path: ancestor.to_path_buf(),
+                        source: error,
+                    });
+                }
+                _ => {}
+            }
+        }
+        Ok(root)
+    }
+    #[cfg(not(unix))]
+    {
+        Ok(root)
+    }
+}
+
+/// Apply the precedence rules documented at the module level, shortening only
+/// implicit endpoints to leave room for store-specific control sockets.
 pub fn resolve_socket_path(inputs: &SocketPathInputs) -> Result<PathBuf, DaemonSocketError> {
     let path = if let Some(explicit) = inputs.explicit.clone() {
         explicit
     } else if let Some(home) = inputs.codewhale_home_override.clone() {
-        home.join("run").join(DAEMON_SOCKET_FILE_NAME)
+        implicit_endpoint_root("CODEWHALE_HOME", home)?
+            .join("run")
+            .join(DAEMON_SOCKET_FILE_NAME)
     } else if let Some(runtime_dir) = inputs.xdg_runtime_dir.clone() {
-        runtime_dir.join("codewhale").join(DAEMON_SOCKET_FILE_NAME)
+        implicit_endpoint_root("XDG_RUNTIME_DIR", runtime_dir)?
+            .join("codewhale")
+            .join(DAEMON_SOCKET_FILE_NAME)
     } else {
         let user_home = inputs
             .user_home
             .clone()
             .ok_or(DaemonSocketError::RuntimeDirUnavailable)?;
+        let user_home = implicit_endpoint_root("HOME", user_home)?;
         if inputs.macos {
             user_home
                 .join("Library")
@@ -170,6 +224,13 @@ pub fn resolve_socket_path(inputs: &SocketPathInputs) -> Result<PathBuf, DaemonS
                 .join(DAEMON_SOCKET_FILE_NAME)
         }
     };
+    #[cfg(unix)]
+    if inputs.explicit.is_none() {
+        let store_name = format!("store-{}.sock", "0".repeat(DERIVED_STORE_HASH_BYTES * 2));
+        if path.with_file_name(store_name).as_os_str().len() > MAX_SOCKET_PATH_BYTES {
+            return Ok(short_socket_directory(&path).join(DAEMON_SOCKET_FILE_NAME));
+        }
+    }
     let len = path.as_os_str().len();
     if len > MAX_SOCKET_PATH_BYTES {
         return Err(DaemonSocketError::PathTooLong {
@@ -181,6 +242,25 @@ pub fn resolve_socket_path(inputs: &SocketPathInputs) -> Result<PathBuf, DaemonS
     Ok(path)
 }
 
+#[cfg(unix)]
+fn short_socket_directory(scope: &Path) -> PathBuf {
+    use sha2::Digest as _;
+
+    let mut digest = sha2::Sha256::new();
+    digest.update(b"Codewhale/short-socket/v1\0");
+    digest.update(scope.as_os_str().as_encoded_bytes());
+    let suffix: String = digest.finalize()[..16]
+        .iter()
+        .map(|byte| format!("{byte:02x}"))
+        .collect();
+    let root = if cfg!(target_os = "macos") {
+        "/private/tmp"
+    } else {
+        "/tmp"
+    };
+    PathBuf::from(root).join(format!("codewhale-{suffix}"))
+}
+
 /// The socket path this host would use with no explicit override.
 #[cfg(unix)]
 pub fn default_socket_path() -> Result<PathBuf, DaemonSocketError> {

crates/paths/src/lib.rs

@@ -11,6 +11,28 @@ use std::path::{Path, PathBuf};
 
 pub mod daemon_socket;
 
+#[cfg(unix)]
+#[doc(hidden)]
+pub fn resolve_root_owned_ancestors(directory: &Path) -> PathBuf {
+    use std::os::unix::fs::MetadataExt as _;
+
+    let mut resolved = PathBuf::new();
+    for component in directory.components() {
+        resolved.push(component);
+        if let Ok(metadata) = std::fs::symlink_metadata(&resolved)
+            && metadata.file_type().is_symlink()
+        {
+            if metadata.uid() != 0 {
+                return directory.to_path_buf();
+            }
+            if let Ok(real) = resolved.canonicalize() {
+                resolved = real;
+            }
+        }
+    }
+    resolved
+}
+
 /// Canonical Codewhale app directory name under the user home.
 pub const CODEWHALE_APP_DIR: &str = ".codewhale";