feat: execute durable standalone Jobs through Host-owned Internal Workers (T-708)

crates/standalone/Cargo.toml

@@ -10,6 +10,8 @@ agen.workspace = true
 client.workspace = true
 fs4.workspace = true
 manifest.workspace = true
+job.workspace = true
+rusqlite.workspace = true
 protocol.workspace = true
 serde.workspace = true
 serde_json.workspace = true

crates/standalone/src/host.rs

@@ -43,6 +43,7 @@ pub struct StandaloneHost {
     worker_store: FsWorkerStore,
     record: StandaloneWorkerRecord,
     lease: Option<StandaloneWorkerLease>,
+    jobs: crate::jobs::StandaloneJobs,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Error)]
@@ -77,6 +78,8 @@ pub enum StandaloneStartupError {
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Error)]
 pub enum StandaloneShutdownError {
+    #[error("standalone Job execution cleanup or terminal persistence could not be confirmed")]
+    JobCleanup,
     #[error("the standalone Worker did not stop before the shutdown deadline")]
     DeadlineExceeded,
     #[error("the standalone Worker shutdown confirmation was lost")]
@@ -110,6 +113,16 @@ impl StandaloneHost {
             .allocate(&launch.cwd, StaleLeasePolicy::Reject)
             .map_err(classify_store_startup_error)?;
         let worker_id = allocation.worker_id();
+        let jobs = match crate::jobs::StandaloneJobs::open(
+            &store.jobs_path(worker_id),
+            launch.cwd.clone(),
+        ) {
+            Ok(jobs) => jobs,
+            Err(_) => {
+                let _ = store.abandon_allocation(allocation);
+                return Err(StandaloneStartupError::StateStore);
+            }
+        };
 
         // WorkerId is the stable identity. The current Worker store remains
         // name-keyed, so keep its derived storage key separate from the
@@ -181,6 +194,7 @@ impl StandaloneHost {
             worker_store,
             record,
             allocation.into_lease(),
+            jobs,
         ))
     }
 
@@ -216,6 +230,11 @@ impl StandaloneHost {
         let lease = store
             .acquire_lease(worker_id, StaleLeasePolicy::Recover)
             .map_err(classify_store_startup_error)?;
+        let jobs = crate::jobs::StandaloneJobs::open(
+            &store.jobs_path(worker_id),
+            record.cwd.canonical_path.clone(),
+        )
+        .map_err(|_| StandaloneStartupError::StateStore)?;
         let (backing_store, worker_store) = backing_store(&store, worker_id)?;
         let storage_key = record.storage_key.clone();
         let mut manifest = record.manifest.clone();
@@ -270,6 +289,7 @@ impl StandaloneHost {
             worker_store,
             record,
             lease,
+            jobs,
         ))
     }
 
@@ -279,6 +299,7 @@ impl StandaloneHost {
         worker_store: FsWorkerStore,
         record: StandaloneWorkerRecord,
         lease: StandaloneWorkerLease,
+        jobs: crate::jobs::StandaloneJobs,
     ) -> Self {
         Self {
             handle: started.handle,
@@ -288,9 +309,16 @@ impl StandaloneHost {
             worker_store,
             record,
             lease: Some(lease),
+            jobs,
         }
     }
 
+    /// Explicit Host execution capability for local Features, separate from the
+    /// interactive Worker protocol and its model-created SubWorkers.
+    pub fn jobs(&self) -> crate::jobs::StandaloneJobs {
+        self.jobs.clone()
+    }
+
     #[must_use]
     pub fn worker_id(&self) -> WorkerId {
         self.record.worker_id
@@ -318,6 +346,7 @@ impl StandaloneHost {
     }
 
     pub async fn shutdown(mut self) -> Result<(), StandaloneShutdownError> {
+        let jobs_result = self.jobs.shutdown().await;
         let command = protocol::WorkerCommandEnvelope::new(u64::MAX);
         let _ = self.handle.send(Method::Shutdown { command }).await;
         let Some(shutdown) = self.shutdown.take() else {
@@ -335,6 +364,10 @@ impl StandaloneHost {
                 return Err(StandaloneShutdownError::DeadlineExceeded);
             }
         }
+        if jobs_result.is_err() {
+            self.retain_lease();
+            return Err(StandaloneShutdownError::JobCleanup);
+        }
         let active = match active_pointer(&self.worker_store, &self.record.storage_key) {
             Ok(active) => active,
             Err(_) => {
@@ -370,6 +403,16 @@ impl StandaloneHost {
     }
 }
 
+impl Drop for StandaloneHost {
+    fn drop(&mut self) {
+        self.jobs.close();
+        // A dropped Host cannot synchronously confirm either controller or Job
+        // cleanup. Keep its lease until explicit shutdown or process recovery,
+        // preventing another Host from reopening a still-live execution store.
+        self.retain_lease();
+    }
+}
+
 async fn run_protocol_session(
     handle: worker::WorkerHandle,
     streams: WorkerProtocolSessionStreams,

crates/standalone/src/jobs.rs

@@ -0,0 +1,393 @@
+//! Host-provided local Jobs. No Runtime registry, Workspace client or dialog-session state.
+#[cfg(test)]
+mod tests;
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use agen::llm_client::LlmClient;
+use job::{JobRequest, JobResultSubmission};
+use tokio::sync::watch;
+
+use crate::job_store::JobStore;
+pub use crate::job_store::{JobAttempt, JobAttemptState, JobSnapshot, JobState, JobStoreError};
+
+#[derive(Debug, thiserror::Error)]
+pub enum JobServiceError {
+    #[error(transparent)]
+    Store(#[from] JobStoreError),
+    #[error("standalone Job execution requirements are unavailable: {0}")]
+    Configuration(String),
+    #[error("the standalone Job Host is shutting down")]
+    Closed,
+    #[error("standalone Job execution capacity is exhausted")]
+    Capacity,
+    #[error("standalone Job cleanup failed: {0}")]
+    Cleanup(String),
+}
+
+struct Execution {
+    cancel: watch::Sender<bool>,
+    complete: watch::Receiver<Option<Result<(), String>>>,
+    attempt_id: String,
+    resource_key: Option<String>,
+    max_concurrent: u16,
+    task: tokio::task::JoinHandle<Result<(), JobServiceError>>,
+}
+struct Executions {
+    closed: bool,
+    jobs: HashMap<String, Execution>,
+}
+
+/// Cloneable Feature-facing access to a Host-owned service. Cloning this handle
+/// neither extends the Host's authority nor permits opening/recovering its store.
+#[derive(Clone)]
+pub struct StandaloneJobs {
+    store: Arc<Mutex<Option<JobStore>>>,
+    executions: Arc<Mutex<Executions>>,
+    cwd: PathBuf,
+    capacity: u16,
+}
+
+impl StandaloneJobs {
+    pub(crate) fn open(path: &Path, cwd: PathBuf) -> Result<Self, JobServiceError> {
+        Ok(Self {
+            store: Arc::new(Mutex::new(Some(JobStore::open(path)?))),
+            executions: Arc::new(Mutex::new(Executions {
+                closed: false,
+                jobs: HashMap::new(),
+            })),
+            cwd,
+            capacity: job::DEFAULT_MAX_CONCURRENT_JOBS,
+        })
+    }
+
+    fn with_store<T>(
+        &self,
+        operation: impl FnOnce(&mut JobStore) -> Result<T, JobStoreError>,
+    ) -> Result<T, JobServiceError> {
+        let mut slot = self.store.lock().expect("Job store poisoned");
+        Ok(operation(slot.as_mut().ok_or(JobServiceError::Closed)?)?)
+    }
+
+    /// Persist intent without starting a model. Exact resend returns the same
+    /// current attempt, including after acknowledgement or cancellation.
+    pub fn request(&self, request: JobRequest) -> Result<JobSnapshot, JobServiceError> {
+        let executions = self.executions.lock().expect("Job executions poisoned");
+        if executions.closed {
+            return Err(JobServiceError::Closed);
+        }
+        Ok(self.with_store(|store| store.reserve(request))?)
+    }
+
+    pub fn get(&self, job_id: &str) -> Result<JobSnapshot, JobServiceError> {
+        Ok(self.with_store(|store| store.get(job_id))?)
+    }
+
+    /// Pending attempts survive shutdown. The consumer chooses when to resume
+    /// them; opening a Host never implicitly replays unknown side effects.
+    pub fn pending(&self) -> Result<Vec<JobSnapshot>, JobServiceError> {
+        Ok(self.with_store(|store| store.pending())?)
+    }
+
+    /// Explicit new attempt, bounded by immutable intent limits. Unknown
+    /// outcomes cannot be retried without domain reconciliation/new intent.
+    pub fn retry(&self, job_id: &str) -> Result<JobSnapshot, JobServiceError> {
+        let executions = self.executions.lock().expect("Job executions poisoned");
+        if executions.closed {
+            return Err(JobServiceError::Closed);
+        }
+        if executions
+            .jobs
+            .get(job_id)
+            .is_some_and(|run| !run.task.is_finished())
+        {
+            return Err(JobServiceError::Capacity);
+        }
+        Ok(self.with_store(|store| store.retry(job_id))?)
+    }
+
+    /// Successful model processing and consumer postprocessing have independent
+    /// lifetimes. A failed delivery simply leaves this acknowledgement pending.
+    pub fn acknowledge(&self, job_id: &str) -> Result<JobSnapshot, JobServiceError> {
+        let executions = self.executions.lock().expect("Job executions poisoned");
+        if executions.closed {
+            return Err(JobServiceError::Closed);
+        }
+        Ok(self.with_store(|store| store.ack(job_id))?)
+    }
+
+    pub fn cancel(&self, job_id: &str) -> Result<JobSnapshot, JobServiceError> {
+        let executions = self.executions.lock().expect("Job executions poisoned");
+        if executions.closed {
+            return Err(JobServiceError::Closed);
+        }
+        let snapshot = self.with_store(|store| store.cancel(job_id))?;
+        if let Some(execution) = executions.jobs.get(job_id) {
+            execution.cancel.send_replace(true);
+        }
+        Ok(snapshot)
+    }
+
+    pub async fn start(&self, job_id: &str) -> Result<(), JobServiceError> {
+        self.start_with_optional_model_client(job_id, None).await
+    }
+
+    /// Injectable model transport for embedders and scripted execution tests.
+    /// The requested registry Profile still resolves/validates normally; the
+    /// transport grants no filesystem, Feature or result-submission authority.
+    pub async fn start_with_model_client<C>(
+        &self,
+        job_id: &str,
+        client: C,
+    ) -> Result<(), JobServiceError>
+    where
+        C: LlmClient + 'static,
+    {
+        self.start_with_optional_model_client(job_id, Some(Box::new(client)))
+            .await
+    }
+
+    async fn start_with_optional_model_client(
+        &self,
+        job_id: &str,
+        client: Option<Box<dyn LlmClient>>,
+    ) -> Result<(), JobServiceError> {
+        let snapshot = self.get(job_id)?;
+        let prepared = worker::job::prepare_job(&snapshot.request, &self.cwd, client)
+            .map_err(|error| JobServiceError::Configuration(error.to_string()))?;
+        let mut executions = self.executions.lock().expect("Job executions poisoned");
+        if executions.closed {
+            return Err(JobServiceError::Closed);
+        }
+        executions.jobs.retain(|_, run| {
+            !(run.task.is_finished() && matches!(run.complete.borrow().as_ref(), Some(Ok(()))))
+        });
+        // Count live resources, not only result states: acceptance can precede
+        // the Internal Worker's cleanup and must not free a concurrency slot.
+        let live = || {
+            executions
+                .jobs
+                .values()
+                .filter(|run| !run.task.is_finished())
+        };
+        let count = live().count();
+        let active_limit = live()
+            .map(|run| run.max_concurrent)
+            .min()
+            .unwrap_or(self.capacity);
+        if count >= usize::from(self.capacity.min(active_limit))
+            || count >= usize::from(snapshot.request.limits.max_concurrent_jobs)
+            || executions
+                .jobs
+                .get(job_id)
+                .is_some_and(|run| !run.task.is_finished())
+        {
+            return Err(JobServiceError::Capacity);
+        }
+        let resource_key = snapshot.request.serialization_key.clone();
+        if let Some(key) = &resource_key {
+            if executions
+                .jobs
+                .values()
+                .any(|run| !run.task.is_finished() && run.resource_key.as_ref() == Some(key))
+            {
+                return Err(JobServiceError::Capacity);
+            }
+        }
+        // Retain errors from finished tasks until a caller observes them rather
+        // than silently swallowing store/cleanup failures.
+        if let Some(previous) = executions.jobs.get(job_id) {
+            if !previous.task.is_finished() {
+                return Err(JobServiceError::Capacity);
+            }
+        }
+        let snapshot = self.with_store(|store| {
+            store.dispatch(job_id, &snapshot.attempt.attempt_id, self.capacity)
+        })?;
+        let store = self.store.clone();
+        let id = job_id.to_string();
+        let attempt_id = snapshot.attempt.attempt_id.clone();
+        let timeout = Duration::from_secs(u64::from(snapshot.request.limits.timeout_seconds));
+        let sink = Arc::new(BoundResultSink {
+            store: store.clone(),
+            job_id: id.clone(),
+            attempt_id: attempt_id.clone(),
+            input_revision: snapshot.request.input_revision.clone(),
+            deadline: tokio::time::Instant::now() + timeout,
+        });
+        let (cancel, mut cancellation) = watch::channel(false);
+        let (complete_tx, complete) = watch::channel(None);
+        let execution_attempt = attempt_id.clone();
+        let task = tokio::spawn(async move {
+            let outcome: Result<(), JobServiceError> = async {
+            let (sender_tx, mut sender_rx) = tokio::sync::oneshot::channel();
+            let run = prepared.run(attempt_id.clone(), sink, move |sender| { let _ = sender_tx.send(sender); });
+            tokio::pin!(run);
+            let reason = tokio::select! {
+                result = &mut run => {
+                    let (state, detail) = match result {
+                        Ok(_) => (JobAttemptState::Failed, "Worker finished without an accepted structured result".to_string()),
+                        Err(error @ worker::job::JobExecutionError::ResultUnconfirmed(_)) => (JobAttemptState::Unknown, error.to_string()),
+                        Err(error) => (JobAttemptState::Failed, error.to_string()),
+                    };
+                    finish_if_pending(&store, &id, &attempt_id, state, &detail)?;
+                    return Ok(());
+                }
+                _ = tokio::time::sleep(timeout) => "timeout",
+                _ = cancellation.wait_for(|cancelled| *cancelled) => "Host cancellation/shutdown",
+            };
+            // Fence timeout/shutdown before cancelling so a late tool call can
+            // never turn an interrupted unaccepted execution into success.
+            let fence_result = finish_if_pending(&store, &id, &attempt_id, JobAttemptState::Unknown, reason);
+            // Cancellation is routed through Engine; keep the future alive to
+            // allow normal Feature cleanup. Hard stop is a fenced unknown
+            // outcome, never an automatic retry. Result acceptance wins races.
+            if let Ok(sender) = sender_rx.try_recv() {
+                let _ = sender.try_send(());
+            } else if let Ok(Ok(sender)) = tokio::time::timeout(Duration::from_secs(1), &mut sender_rx).await {
+                let _ = sender.try_send(());
+            }
+            let cleanup = tokio::time::timeout(Duration::from_secs(2), &mut run).await;
+            fence_result?;
+            if cleanup.is_err() { return Err(JobServiceError::Cleanup("Internal Worker cancellation deadline exceeded; outcome fenced unknown".into())); }
+            Ok(())
+            }.await;
+            complete_tx.send_replace(Some(
+                outcome.as_ref().map(|_| ()).map_err(ToString::to_string),
+            ));
+            outcome
+        });
+        executions.jobs.insert(
+            job_id.to_string(),
+            Execution {
+                cancel,
+                complete,
+                task,
+                attempt_id: execution_attempt,
+                resource_key,
+                max_concurrent: snapshot.request.limits.max_concurrent_jobs,
+            },
+        );
+        Ok(())
+    }
+
+    /// Observe execution/cleanup, not success: read the durable result separately.
+    pub async fn wait(&self, job_id: &str) -> Result<JobSnapshot, JobServiceError> {
+        let execution = self
+            .executions
+            .lock()
+            .expect("Job executions poisoned")
+            .jobs
+            .get(job_id)
+            .map(|run| (run.attempt_id.clone(), run.complete.clone()));
+        if let Some((attempt, mut completion)) = execution {
+            let result = completion
+                .wait_for(Option::is_some)
+                .await
+                .map_err(|error| JobServiceError::Cleanup(error.to_string()))?
+                .clone()
+                .expect("completion");
+            result.map_err(JobServiceError::Cleanup)?;
+            let task = {
+                let mut executions = self.executions.lock().expect("Job executions poisoned");
+                if executions
+                    .jobs
+                    .get(job_id)
+                    .is_some_and(|run| run.attempt_id == attempt)
+                {
+                    executions.jobs.remove(job_id)
+                } else {
+                    None
+                }
+            };
+            if let Some(execution) = task {
+                execution
+                    .task
+                    .await
+                    .map_err(|error| JobServiceError::Cleanup(error.to_string()))??;
+            }
+        }
+        self.get(job_id)
+    }
+
+    pub(crate) fn close(&self) {
+        let mut executions = self.executions.lock().expect("Job executions poisoned");
+        executions.closed = true;
+        for execution in executions.jobs.values() {
+            execution.cancel.send_replace(true);
+        }
+    }
+
+    pub(crate) async fn shutdown(&self) -> Result<(), JobServiceError> {
+        self.close();
+        let tasks = {
+            let mut executions = self.executions.lock().expect("Job executions poisoned");
+            std::mem::take(&mut executions.jobs)
+        };
+        let mut error = None;
+        for execution in tasks.into_values() {
+            match execution.task.await {
+                Ok(Ok(())) => {}
+                Ok(Err(failure)) => {
+                    error = Some(failure);
+                }
+                Err(failure) => {
+                    error = Some(JobServiceError::Cleanup(failure.to_string()));
+                }
+            }
+        }
+        // Close SQLite before the Host releases its lease. Cloned handles may
+        // outlive the Host but cannot mutate or retain the live connection.
+        self.store.lock().expect("Job store poisoned").take();
+        error.map_or(Ok(()), Err)
+    }
+}
+
+fn finish_if_pending(
+    store: &Mutex<Option<JobStore>>,
+    id: &str,
+    attempt: &str,
+    state: JobAttemptState,
+    detail: &str,
+) -> Result<(), JobServiceError> {
+    let mut slot = store.lock().expect("Job store poisoned");
+    let store = slot.as_mut().ok_or(JobServiceError::Closed)?;
+    let snapshot = store.get(id)?;
+    if snapshot.state == JobState::Pending {
+        store.finish(id, attempt, state, Some(detail))?;
+    }
+    Ok(())
+}
+
+struct BoundResultSink {
+    store: Arc<Mutex<Option<JobStore>>>,
+    job_id: String,
+    attempt_id: String,
+    input_revision: String,
+    deadline: tokio::time::Instant,
+}
+impl worker::job::JobResultSink for BoundResultSink {
+    fn submit(&self, submission: JobResultSubmission) -> Result<(), String> {
+        if submission.job_id != self.job_id
+            || submission.attempt_id != self.attempt_id
+            || submission.input_revision != self.input_revision
+        {
+            return Err("Job result capability binding mismatch".into());
+        }
+        let mut slot = self
+            .store
+            .lock()
+            .map_err(|_| "Job store poisoned".to_string())?;
+        let store = slot.as_mut().ok_or_else(|| "Job Host closed".to_string())?;
+        let snapshot = store.get(&self.job_id).map_err(|error| error.to_string())?;
+        if tokio::time::Instant::now() >= self.deadline && snapshot.state != JobState::Completed {
+            return Err("Job result deadline expired".into());
+        }
+        store
+            .accept(&submission)
+            .map(|_| ())
+            .map_err(|error| error.to_string())
+    }
+}

crates/standalone/src/jobs/tests.rs

@@ -0,0 +1,126 @@
+use super::*;
+use worker::job::JobResultSink;
+
+fn request(id: &str) -> JobRequest {
+    JobRequest {
+        job_id: id.into(),
+        purpose: "test".into(),
+        input_revision: "r1".into(),
+        input_ref: "fixture:input".into(),
+        input: serde_json::json!({"data":42}),
+        instruction: "Return a structured result".into(),
+        profile: "builtin:job".into(),
+        serialization_key: None,
+        limits: job::JobLimits::default(),
+    }
+}
+fn sink(jobs: &StandaloneJobs, id: &str, attempt: &str) -> BoundResultSink {
+    BoundResultSink {
+        store: jobs.store.clone(),
+        job_id: id.into(),
+        attempt_id: attempt.into(),
+        input_revision: "r1".into(),
+        deadline: tokio::time::Instant::now() + Duration::from_secs(10),
+    }
+}
+fn result(id: &str, attempt: &str) -> JobResultSubmission {
+    JobResultSubmission {
+        job_id: id.into(),
+        attempt_id: attempt.into(),
+        input_revision: "r1".into(),
+        result: serde_json::json!({"valid":true}),
+    }
+}
+
+#[tokio::test]
+async fn host_result_capability_cannot_cross_job_attempt_revision_or_deadline() {
+    let temp = tempfile::tempdir().unwrap();
+    let jobs =
+        StandaloneJobs::open(&temp.path().join("jobs.sqlite3"), temp.path().to_path_buf()).unwrap();
+    let first = jobs.request(request("first")).unwrap();
+    let second = jobs.request(request("second")).unwrap();
+    for snapshot in [&first, &second] {
+        jobs.with_store(|store| {
+            store.dispatch(&snapshot.request.job_id, &snapshot.attempt.attempt_id, 8)
+        })
+        .unwrap();
+    }
+    let bound = sink(&jobs, "first", &first.attempt.attempt_id);
+    assert!(
+        bound
+            .submit(result("second", &second.attempt.attempt_id))
+            .is_err()
+    );
+    assert!(bound.submit(result("first", "first:attempt:2")).is_err());
+    let mut stale = result("first", &first.attempt.attempt_id);
+    stale.input_revision = "r0".into();
+    assert!(bound.submit(stale).is_err());
+    let mut expired = sink(&jobs, "first", &first.attempt.attempt_id);
+    expired.deadline = tokio::time::Instant::now() - Duration::from_secs(1);
+    assert!(
+        expired
+            .submit(result("first", &first.attempt.attempt_id))
+            .is_err()
+    );
+    assert!(jobs.get("first").unwrap().result.is_none());
+    bound
+        .submit(result("first", &first.attempt.attempt_id))
+        .unwrap();
+    // Accepted exact replay is still allowed after the deadline, never a new result.
+    expired
+        .submit(result("first", &first.attempt.attempt_id))
+        .unwrap();
+    let mut changed = result("first", &first.attempt.attempt_id);
+    changed.result = serde_json::json!({"valid":false});
+    assert!(expired.submit(changed).is_err());
+    assert_eq!(jobs.get("second").unwrap().state, JobState::Pending);
+    jobs.shutdown().await.unwrap();
+    assert!(
+        bound
+            .submit(result("first", &first.attempt.attempt_id))
+            .is_err()
+    );
+}
+
+#[tokio::test]
+async fn cleanup_cannot_reclassify_success_or_durable_cancellation_and_stale_handles_are_closed() {
+    let temp = tempfile::tempdir().unwrap();
+    let jobs =
+        StandaloneJobs::open(&temp.path().join("jobs.sqlite3"), temp.path().to_path_buf()).unwrap();
+    let cancelled = jobs.request(request("cancelled")).unwrap();
+    jobs.cancel("cancelled").unwrap();
+    finish_if_pending(
+        &jobs.store,
+        "cancelled",
+        &cancelled.attempt.attempt_id,
+        JobAttemptState::Unknown,
+        "cleanup",
+    )
+    .unwrap();
+    assert_eq!(jobs.get("cancelled").unwrap().state, JobState::Cancelled);
+    let accepted = jobs.request(request("accepted")).unwrap();
+    jobs.with_store(|store| store.dispatch("accepted", &accepted.attempt.attempt_id, 8))
+        .unwrap();
+    sink(&jobs, "accepted", &accepted.attempt.attempt_id)
+        .submit(result("accepted", &accepted.attempt.attempt_id))
+        .unwrap();
+    finish_if_pending(
+        &jobs.store,
+        "accepted",
+        &accepted.attempt.attempt_id,
+        JobAttemptState::Failed,
+        "model failed later",
+    )
+    .unwrap();
+    assert_eq!(jobs.get("accepted").unwrap().state, JobState::Completed);
+    jobs.shutdown().await.unwrap();
+    assert!(jobs.store.lock().unwrap().is_none());
+    for operation in [
+        jobs.cancel("cancelled"),
+        jobs.acknowledge("accepted"),
+        jobs.retry("cancelled"),
+        jobs.request(request("new")),
+    ] {
+        assert!(matches!(operation, Err(JobServiceError::Closed)));
+    }
+}

crates/standalone/src/lib.rs

@@ -5,6 +5,8 @@
 //! Server, HTTP, WebSocket, subprocess Worker, or alternative execution path.
 
 pub mod host;
+mod job_store;
+pub mod jobs;
 pub mod launch;
 pub mod store;
 

crates/standalone/src/store.rs

@@ -352,6 +352,10 @@ impl StandaloneWorkerStore {
     }
 
     #[must_use]
+    pub(crate) fn jobs_path(&self, id: WorkerId) -> PathBuf {
+        self.worker_dir(id).join("jobs.sqlite3")
+    }
+
     pub(crate) fn runtime_dir(&self, id: WorkerId) -> PathBuf {
         self.worker_dir(id).join("runtime")
     }

crates/standalone/tests/jobs.rs

@@ -0,0 +1,478 @@
+use std::collections::VecDeque;
+use std::pin::Pin;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use agen::llm_client::{
+    LlmClient,
+    error::ClientError,
+    event::{Event as LlmEvent, StopReason},
+    types::Request,
+};
+use async_trait::async_trait;
+use futures::{Stream, stream};
+use job::{JobLimits, JobRequest};
+use standalone::jobs::{JobServiceError, JobState};
+use standalone::{StandaloneHost, StandaloneLaunchConfig};
+
+#[derive(Clone)]
+struct ScriptedClient {
+    responses: Arc<Mutex<VecDeque<Vec<LlmEvent>>>>,
+    requests: Arc<Mutex<Vec<Request>>>,
+    started: Arc<tokio::sync::Notify>,
+    stall: bool,
+    fail: bool,
+}
+impl ScriptedClient {
+    fn new(responses: Vec<Vec<LlmEvent>>) -> Self {
+        Self {
+            responses: Arc::new(Mutex::new(responses.into())),
+            requests: Arc::new(Mutex::new(Vec::new())),
+            started: Arc::new(tokio::sync::Notify::new()),
+            stall: false,
+            fail: false,
+        }
+    }
+    fn stall() -> Self {
+        Self {
+            stall: true,
+            ..Self::new(Vec::new())
+        }
+    }
+    fn requests(&self) -> Vec<Request> {
+        self.requests.lock().unwrap().clone()
+    }
+}
+#[async_trait]
+impl LlmClient for ScriptedClient {
+    async fn stream(
+        &self,
+        request: Request,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<LlmEvent, ClientError>> + Send>>, ClientError>
+    {
+        self.requests.lock().unwrap().push(request);
+        self.started.notify_one();
+        if self.fail {
+            return Err(ClientError::Config("scripted model failure".into()));
+        }
+        if self.stall {
+            return Ok(Box::pin(stream::pending()));
+        }
+        let events = self
+            .responses
+            .lock()
+            .unwrap()
+            .pop_front()
+            .expect("scripted response");
+        Ok(Box::pin(stream::iter(events.into_iter().map(Ok))))
+    }
+    fn clone_boxed(&self) -> Box<dyn LlmClient> {
+        Box::new(self.clone())
+    }
+}
+fn prose() -> Vec<LlmEvent> {
+    vec![
+        LlmEvent::text_block_start(0),
+        LlmEvent::text_delta(0, "done"),
+        LlmEvent::text_block_stop(0, Some(StopReason::EndTurn)),
+    ]
+}
+fn success() -> ScriptedClient {
+    ScriptedClient::new(vec![
+        vec![
+            LlmEvent::tool_use_start(0, "result-1", "SubmitJobResult"),
+            LlmEvent::tool_input_delta(0, r#"{"result":{"valid":true}}"#),
+            LlmEvent::tool_use_stop(0),
+        ],
+        prose(),
+    ])
+}
+fn request(id: &str) -> JobRequest {
+    JobRequest {
+        job_id: id.into(),
+        purpose: "generic_test".into(),
+        input_revision: "r1".into(),
+        input_ref: "test:immutable".into(),
+        input: serde_json::json!({"test_input":42}),
+        instruction: "Return a structured validity result".into(),
+        profile: "builtin:job".into(),
+        serialization_key: None,
+        limits: JobLimits::default(),
+    }
+}
+async fn host(temp: &tempfile::TempDir) -> (StandaloneHost, ScriptedClient) {
+    let launch = StandaloneLaunchConfig::new(
+        temp.path(),
+        temp.path().join("state"),
+        manifest::ProfileSelector::Default,
+        "job-host",
+    )
+    .resolve()
+    .unwrap();
+    let client = ScriptedClient::new(Vec::new());
+    let host = StandaloneHost::start_with_model_client(launch, client.clone())
+        .await
+        .unwrap();
+    (host, client)
+}
+async fn wait_started(client: &ScriptedClient) {
+    tokio::time::timeout(Duration::from_secs(10), client.started.notified())
+        .await
+        .unwrap();
+}
+
+#[tokio::test]
+async fn host_job_runs_without_dialog_and_result_survives_restore_and_ack() {
+    let temp = tempfile::tempdir().unwrap();
+    let (host, dialog) = host(&temp).await;
+    let id = host.worker_id();
+    let jobs = host.jobs();
+    let intent = request("success");
+    let reserved = jobs.request(intent.clone()).unwrap();
+    assert_eq!(jobs.request(intent.clone()).unwrap(), reserved);
+    let model = success();
+    jobs.start_with_model_client("success", model.clone())
+        .await
+        .unwrap();
+    let completed = jobs.wait("success").await.unwrap();
+    assert_eq!(completed.state, JobState::Completed);
+    assert_eq!(completed.result, Some(serde_json::json!({"valid":true})));
+    assert!(!completed.acknowledged);
+    let neutral = completed.outcome();
+    assert_eq!(neutral.state, job::JobState::Completed);
+    assert_eq!(neutral.result, completed.result);
+    assert_eq!(neutral.attempt_state, job::JobAttemptState::Completed);
+    assert!(
+        dialog.requests().is_empty(),
+        "Job never runs on the interactive model/session"
+    );
+    assert_eq!(
+        model.requests()[0]
+            .tools
+            .iter()
+            .map(|tool| tool.name.as_str())
+            .collect::<Vec<_>>(),
+        ["SubmitJobResult"]
+    );
+    assert_eq!(jobs.request(intent.clone()).unwrap(), completed);
+    assert!(
+        jobs.start_with_model_client("success", success())
+            .await
+            .is_err()
+    );
+    assert!(jobs.retry("success").is_err());
+    let db = temp
+        .path()
+        .join("state")
+        .join(id.to_string())
+        .join("jobs.sqlite3");
+    assert!(db.exists());
+    host.shutdown().await.unwrap();
+    assert!(matches!(
+        jobs.request(request("after-shutdown")),
+        Err(JobServiceError::Closed)
+    ));
+    assert!(matches!(
+        jobs.cancel("success"),
+        Err(JobServiceError::Closed)
+    ));
+    assert!(matches!(
+        jobs.acknowledge("success"),
+        Err(JobServiceError::Closed)
+    ));
+    assert!(matches!(
+        jobs.start_with_model_client("success", success()).await,
+        Err(JobServiceError::Closed)
+    ));
+    let restored = StandaloneHost::restore_with_model_client(
+        temp.path().join("state"),
+        id,
+        ScriptedClient::new(Vec::new()),
+    )
+    .await
+    .unwrap();
+    let restored_jobs = restored.jobs();
+    assert_eq!(restored_jobs.get("success").unwrap(), completed);
+    assert!(restored_jobs.acknowledge("success").unwrap().acknowledged);
+    restored.shutdown().await.unwrap();
+}
+
+#[tokio::test]
+async fn live_resource_limits_cancellation_and_wait_are_host_owned() {
+    let temp = tempfile::tempdir().unwrap();
+    let (host, _) = host(&temp).await;
+    let jobs = host.jobs();
+    let mut first = request("first");
+    first.limits.max_concurrent_jobs = 1;
+    jobs.request(first).unwrap();
+    jobs.request(request("second")).unwrap();
+    let stalled = ScriptedClient::stall();
+    jobs.start_with_model_client("first", stalled.clone())
+        .await
+        .unwrap();
+    wait_started(&stalled).await;
+    // Waiting must not remove the Host-owned task/cancellation capability.
+    let waiting_jobs = jobs.clone();
+    let waiter = tokio::spawn(async move { waiting_jobs.wait("first").await });
+    assert!(matches!(
+        jobs.start_with_model_client("second", success()).await,
+        Err(JobServiceError::Capacity)
+    ));
+    jobs.cancel("first").unwrap();
+    let cancelled = tokio::time::timeout(Duration::from_secs(10), waiter)
+        .await
+        .unwrap()
+        .unwrap()
+        .unwrap();
+    assert_eq!(cancelled.state, JobState::Cancelled);
+    assert!(jobs.retry("first").is_err());
+    jobs.start_with_model_client("second", success())
+        .await
+        .unwrap();
+    assert_eq!(
+        jobs.wait("second").await.unwrap().state,
+        JobState::Completed
+    );
+    host.shutdown().await.unwrap();
+}
+
+#[tokio::test]
+async fn prose_failure_requires_explicit_bounded_new_attempt_and_exact_replay_does_not_retry() {
+    let temp = tempfile::tempdir().unwrap();
+    let (host, _) = host(&temp).await;
+    let jobs = host.jobs();
+    let intent = request("prose");
+    jobs.request(intent.clone()).unwrap();
+    jobs.start_with_model_client("prose", ScriptedClient::new(vec![prose()]))
+        .await
+        .unwrap();
+    let failed = jobs.wait("prose").await.unwrap();
+    assert_eq!(failed.state, JobState::Failed);
+    assert!(failed.result.is_none());
+    assert_eq!(jobs.request(intent).unwrap(), failed);
+    let retry = jobs.retry("prose").unwrap();
+    assert_eq!(retry.attempt.number, 2);
+    assert_ne!(retry.attempt.attempt_id, failed.attempt.attempt_id);
+    jobs.start_with_model_client("prose", ScriptedClient::new(vec![prose()]))
+        .await
+        .unwrap();
+    assert_eq!(jobs.wait("prose").await.unwrap().state, JobState::Failed);
+    assert!(jobs.retry("prose").is_err());
+    host.shutdown().await.unwrap();
+}
+
+#[tokio::test]
+async fn timeout_is_unknown_not_prose_success_or_automatic_retry() {
+    let temp = tempfile::tempdir().unwrap();
+    let (host, _) = host(&temp).await;
+    let jobs = host.jobs();
+    let mut intent = request("timeout");
+    intent.limits.timeout_seconds = 1;
+    jobs.request(intent).unwrap();
+    let stalled = ScriptedClient::stall();
+    jobs.start_with_model_client("timeout", stalled.clone())
+        .await
+        .unwrap();
+    wait_started(&stalled).await;
+    let outcome = tokio::time::timeout(Duration::from_secs(10), jobs.wait("timeout"))
+        .await
+        .unwrap()
+        .unwrap();
+    assert_eq!(outcome.state, JobState::Unknown);
+    assert!(outcome.result.is_none());
+    assert!(jobs.retry("timeout").is_err());
+    host.shutdown().await.unwrap();
+}
+
+#[tokio::test]
+async fn shutdown_keeps_reserved_results_and_fences_running_attempts_on_reopen() {
+    let temp = tempfile::tempdir().unwrap();
+    let (host, _) = host(&temp).await;
+    let id = host.worker_id();
+    let jobs = host.jobs();
+    jobs.request(request("reserved")).unwrap();
+    jobs.request(request("cancelled")).unwrap();
+    jobs.cancel("cancelled").unwrap();
+    jobs.request(request("running")).unwrap();
+    let stalled = ScriptedClient::stall();
+    jobs.start_with_model_client("running", stalled.clone())
+        .await
+        .unwrap();
+    wait_started(&stalled).await;
+    host.shutdown().await.unwrap();
+    assert!(matches!(jobs.get("running"), Err(JobServiceError::Closed)));
+    let restored = StandaloneHost::restore_with_model_client(
+        temp.path().join("state"),
+        id,
+        ScriptedClient::new(Vec::new()),
+    )
+    .await
+    .unwrap();
+    let jobs = restored.jobs();
+    assert_eq!(
+        jobs.pending()
+            .unwrap()
+            .iter()
+            .map(|snapshot| snapshot.request.job_id.as_str())
+            .collect::<Vec<_>>(),
+        ["reserved"]
+    );
+    assert_eq!(jobs.get("cancelled").unwrap().state, JobState::Cancelled);
+    assert_eq!(jobs.get("running").unwrap().state, JobState::Unknown);
+    assert!(jobs.retry("running").is_err());
+    jobs.start_with_model_client("reserved", success())
+        .await
+        .unwrap();
+    assert_eq!(
+        jobs.wait("reserved").await.unwrap().state,
+        JobState::Completed
+    );
+    restored.shutdown().await.unwrap();
+}
+
+#[tokio::test]
+async fn profile_requirements_are_rejected_without_model_execution_or_implicit_fallback() {
+    let temp = tempfile::tempdir().unwrap();
+    let (host, _) = host(&temp).await;
+    let jobs = host.jobs();
+    for profile in [
+        "builtin:coder",
+        "builtin:backend-job",
+        "builtin:does-not-exist",
+    ] {
+        let mut intent = request(profile);
+        intent.profile = profile.into();
+        jobs.request(intent).unwrap();
+        let model = success();
+        assert!(matches!(
+            jobs.start_with_model_client(profile, model.clone()).await,
+            Err(JobServiceError::Configuration(_))
+        ));
+        assert!(model.requests().is_empty());
+        assert_eq!(jobs.get(profile).unwrap().state, JobState::Pending);
+    }
+    host.shutdown().await.unwrap();
+}
+
+// Only this test harness starts a subprocess: production Jobs remain in-process.
+// The child exits deliberately without destructors to prove actual process-loss
+// recovery, rather than treating a hand-edited SQLite fixture as that evidence.
+#[tokio::test]
+#[ignore = "entry point for process_loss_recovers_durable_jobs_without_replaying_unknown"]
+async fn process_job_fixture() {
+    let root = std::path::PathBuf::from(
+        std::env::var_os("YOI_JOB_PROCESS_FIXTURE_DIR").expect("fixture root"),
+    );
+    let launch = StandaloneLaunchConfig::new(
+        &root,
+        root.join("state"),
+        manifest::ProfileSelector::Default,
+        "job-process-fixture",
+    )
+    .resolve()
+    .unwrap();
+    let host = StandaloneHost::start_with_model_client(launch, ScriptedClient::new(Vec::new()))
+        .await
+        .unwrap();
+    std::fs::write(root.join("worker-id"), host.worker_id().to_string()).unwrap();
+    let jobs = host.jobs();
+    jobs.request(request("reserved")).unwrap();
+    jobs.request(request("cancelled")).unwrap();
+    jobs.cancel("cancelled").unwrap();
+    jobs.request(request("accepted")).unwrap();
+    jobs.start_with_model_client("accepted", success())
+        .await
+        .unwrap();
+    assert_eq!(
+        jobs.wait("accepted").await.unwrap().state,
+        JobState::Completed
+    );
+    jobs.request(request("interrupted")).unwrap();
+    let stalled = ScriptedClient::stall();
+    jobs.start_with_model_client("interrupted", stalled.clone())
+        .await
+        .unwrap();
+    wait_started(&stalled).await;
+    // Test-owned process only, never the operating Yoi Worker or dogfood daemon.
+    std::process::exit(23);
+}
+
+#[tokio::test]
+async fn process_loss_recovers_durable_jobs_without_replaying_unknown() {
+    let temp = tempfile::tempdir().unwrap();
+    let output = std::process::Command::new(std::env::current_exe().unwrap())
+        .args(["--exact", "process_job_fixture", "--ignored", "--nocapture"])
+        .env("YOI_JOB_PROCESS_FIXTURE_DIR", temp.path())
+        .output()
+        .unwrap();
+    assert_eq!(
+        output.status.code(),
+        Some(23),
+        "fixture failed: {}\n{}",
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr)
+    );
+    let worker_id = std::fs::read_to_string(temp.path().join("worker-id"))
+        .unwrap()
+        .parse()
+        .unwrap();
+    let host = StandaloneHost::restore_with_model_client(
+        temp.path().join("state"),
+        worker_id,
+        ScriptedClient::new(Vec::new()),
+    )
+    .await
+    .unwrap();
+    let jobs = host.jobs();
+    assert_eq!(jobs.get("interrupted").unwrap().state, JobState::Unknown);
+    assert!(jobs.retry("interrupted").is_err());
+    let completed = jobs.get("accepted").unwrap();
+    assert_eq!(completed.state, JobState::Completed);
+    assert_eq!(completed.result, Some(serde_json::json!({"valid":true})));
+    assert!(!completed.acknowledged);
+    assert_eq!(jobs.get("cancelled").unwrap().state, JobState::Cancelled);
+    assert_eq!(jobs.pending().unwrap().len(), 1);
+    jobs.start_with_model_client("reserved", success())
+        .await
+        .unwrap();
+    assert_eq!(
+        jobs.wait("reserved").await.unwrap().state,
+        JobState::Completed
+    );
+    host.shutdown().await.unwrap();
+}
+
+#[tokio::test]
+async fn model_failure_is_durable_known_failure_with_explicit_successful_retry() {
+    let temp = tempfile::tempdir().unwrap();
+    let (host, _) = host(&temp).await;
+    let jobs = host.jobs();
+    jobs.request(request("model-failure")).unwrap();
+    let client = ScriptedClient {
+        fail: true,
+        ..ScriptedClient::new(Vec::new())
+    };
+    jobs.start_with_model_client("model-failure", client)
+        .await
+        .unwrap();
+    let failure = jobs.wait("model-failure").await.unwrap();
+    assert_eq!(failure.state, JobState::Failed);
+    assert!(failure.result.is_none());
+    assert!(
+        failure
+            .attempt
+            .failure
+            .unwrap()
+            .contains("scripted model failure")
+    );
+    let next = jobs.retry("model-failure").unwrap();
+    assert_eq!(next.attempt.number, 2);
+    jobs.start_with_model_client("model-failure", success())
+        .await
+        .unwrap();
+    assert_eq!(
+        jobs.wait("model-failure").await.unwrap().state,
+        JobState::Completed
+    );
+    host.shutdown().await.unwrap();
+}