Harden the POD job model and unify CLI/REST service interfaces

.github/workflows/ci.yml

@@ -148,7 +148,7 @@ jobs:
           for attempt in {1..60}; do
             status_json="$(curl --fail --silent "http://127.0.0.1:${port}/jobs/${job_id}")"
             state="$(printf '%s' "$status_json" | python3 -c 'import json,sys; print(json.load(sys.stdin)["state"])')"
-            if [[ "$state" == "finished" ]]; then
+            if [[ "$state" == "succeeded" ]]; then
               break
             fi
             if [[ "$state" == "failed" ]]; then
@@ -159,7 +159,7 @@ jobs:
           done
 
           state="$(printf '%s' "$status_json" | python3 -c 'import json,sys; print(json.load(sys.stdin)["state"])')"
-          [[ "$state" == "finished" ]]
+          [[ "$state" == "succeeded" ]]
 
           run_dir="$SPOD_REST_OUT/$job_id"
           request_path="$run_dir/request.json"

Cargo.toml

@@ -48,7 +48,7 @@ chrono = "0.4"
 
 # ── HTTP / REST ────────────────────────────────────────────────────────────────
 axum  = "0.7"
-tokio = { version = "1.39", features = ["macros", "rt-multi-thread"] }
+tokio = { version = "1.39", features = ["macros", "rt-multi-thread", "time"] }
 tower = "0.5"
 uuid  = { version = "1.10", features = ["v4"] }
 

docs/architecture/boundaries.md

@@ -4,7 +4,7 @@
 |---|---|---|
 | `siderust::formats` | Standard-format parsing and writing | Service orchestration and transport. |
 | `siderust::pod` | Force models, propagation, observations, estimation, QC, products, and run metadata. | Service configuration and transport. |
-| `spod::service` | Configuration and validation, workflow dispatch, run orchestration, input boundaries, artifact layout, and provenance finalisation using `siderust::pod`. | Reusable numerical algorithms and scientific compatibility facades. |
+| `spod::service` | Configuration and validation, canonical job lifecycle/status/error contracts, workflow dispatch, run orchestration, input boundaries, artifact layout, and provenance finalisation using `siderust::pod`. | Reusable numerical algorithms, HTTP semantics, and durable scheduling/persistence. |
 | `spod` binaries | `clap`-based commands and the REST server; delegates to `spod::service`. | Any numerical logic. |
 
 The full set of allowed and forbidden dependency edges is in

docs/architecture/overview.md

@@ -53,3 +53,18 @@ The CLI and REST binaries are thin front ends over `spod::service::Runner`.
 They do not assemble pipelines or call scientific components directly.
 Siderust's scientific APIs remain independently testable, while the service
 boundary is covered by synthetic end-to-end execution.
+
+## Shared job model
+
+`spod::service` owns the canonical `JobId`, `JobState`, `JobStatus`,
+`JobResult`, and `JobError` contracts. The lifecycle is `pending -> running ->
+succeeded` or `pending -> running -> failed`; terminal statuses carry either
+artifact references or a stable machine-readable error category. REST stores
+these snapshots in an in-memory `JobStore` and exposes them as JSON. The CLI
+uses the same model synchronously and formats its terminal status for
+operators.
+
+The REST adapter is intentionally asynchronous and experimental. It does not
+provide durable persistence, scheduling, retries, or distributed workers.
+Those concerns are deferred to later service/API work, including the stable
+artifact API tracked by #28.

src/bin/spod.rs

@@ -77,17 +77,22 @@ fn main() -> anyhow::Result<()> {
         }
         Cmd::Run { config } => {
             let cfg = spod::service::RunConfig::from_yaml_file(&config)?;
-            let report = spod::service::run(&cfg, &config)?;
-            println!(
-                "OK: ran {} steps, final epoch JD={}, manifest at {}",
-                report.n_steps,
-                report
-                    .final_state
-                    .epoch
-                    .to::<siderust::tempoch::JD>()
-                    .value(),
-                report.artifacts.manifest.path.display()
-            );
+            let status = spod::service::execute(&cfg, &config);
+            match status.state {
+                spod::service::JobState::Succeeded => {
+                    let result = status.result.expect("succeeded status has a result");
+                    println!(
+                        "OK: run {} succeeded, manifest at {}",
+                        result.run_id,
+                        result.artifacts.manifest.path.display()
+                    );
+                }
+                spod::service::JobState::Failed => {
+                    let error = status.error.expect("failed status has an error");
+                    anyhow::bail!("{}: {}", serde_json::to_string(&error.code)?, error.message);
+                }
+                state => anyhow::bail!("run ended in unexpected state: {state:?}"),
+            }
         }
         Cmd::InspectManifest { manifest } => {
             let txt = std::fs::read_to_string(&manifest)?;

src/service/job.rs

@@ -1,15 +1,252 @@
-//! Application-level run requests and results.
+//! Application-level run and job contracts.
 
 use super::artifacts::Artifacts;
 use super::config::RunConfig;
+use super::error::ServiceError;
 use super::provenance::RunProvenance;
 use super::workflow::Workflow;
+use serde::{Deserialize, Serialize};
 use siderust::astro::dynamics::OrbitState;
+use std::collections::HashMap;
 use std::path::PathBuf;
+use std::sync::{Arc, Mutex};
 
 /// Stable identifier assigned by a caller to a run.
 pub type JobId = String;
 
+/// Stable lifecycle state for a submitted service job.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum JobState {
+    /// The service accepted the job but has not started execution.
+    Pending,
+    /// The runner is executing the job.
+    Running,
+    /// The runner completed successfully.
+    Succeeded,
+    /// The runner completed with a categorized failure.
+    Failed,
+}
+
+/// Stable, machine-readable failure exposed by job APIs.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct JobError {
+    /// Stable category/code for programmatic handling.
+    pub code: JobErrorCode,
+    /// Human-readable diagnostic, without implementation-specific details.
+    pub message: String,
+}
+
+/// Categories of failures that can be returned by a job API.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(rename_all = "snake_case")]
+pub enum JobErrorCode {
+    /// Configuration could not be parsed or validated.
+    Configuration,
+    /// An input could not be resolved.
+    Input,
+    /// The requested workflow or input is unsupported.
+    Unsupported,
+    /// An output artifact could not be written or finalized.
+    Artifact,
+    /// Scientific execution failed.
+    Scientific,
+    /// An unexpected service failure occurred.
+    Internal,
+}
+
+impl From<&ServiceError> for JobError {
+    fn from(error: &ServiceError) -> Self {
+        let code = match error {
+            ServiceError::Configuration { .. } => JobErrorCode::Configuration,
+            ServiceError::Unsupported { .. } => JobErrorCode::Unsupported,
+            ServiceError::Input { .. } => JobErrorCode::Input,
+            ServiceError::Artifact { .. } => JobErrorCode::Artifact,
+            ServiceError::Scientific { .. } => JobErrorCode::Scientific,
+        };
+        Self {
+            code,
+            message: error.to_string(),
+        }
+    }
+}
+
+/// Public result summary attached to a succeeded job.
+#[derive(Debug, Clone, Serialize)]
+pub struct JobResult {
+    /// Identifier of the completed run.
+    pub run_id: JobId,
+    /// Workflow that produced the result.
+    pub workflow: Workflow,
+    /// Generated files and their hashes.
+    pub artifacts: Artifacts,
+}
+
+impl From<&RunResult> for JobResult {
+    fn from(result: &RunResult) -> Self {
+        Self {
+            run_id: result.run_id.clone(),
+            workflow: result.workflow,
+            artifacts: result.artifacts.clone(),
+        }
+    }
+}
+
+/// Canonical status snapshot shared by CLI and REST consumers.
+#[derive(Debug, Clone, Serialize)]
+pub struct JobStatus {
+    /// Stable service-assigned or caller-provided identifier.
+    pub id: JobId,
+    /// Current lifecycle state.
+    pub state: JobState,
+    /// Workflow selected for this job.
+    pub workflow: Workflow,
+    /// Present after successful execution.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub result: Option<JobResult>,
+    /// Present after failed execution.
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub error: Option<JobError>,
+}
+
+impl JobStatus {
+    /// Construct a newly accepted job.
+    pub fn pending(id: JobId, workflow: Workflow) -> Self {
+        Self {
+            id,
+            state: JobState::Pending,
+            workflow,
+            result: None,
+            error: None,
+        }
+    }
+
+    /// Advance a pending job to execution.
+    pub fn start(&mut self) -> Result<(), JobTransitionError> {
+        if self.state != JobState::Pending {
+            return Err(JobTransitionError {
+                from: self.state,
+                to: JobState::Running,
+            });
+        }
+        self.state = JobState::Running;
+        Ok(())
+    }
+
+    /// Complete a running job successfully.
+    pub fn succeed(&mut self, result: JobResult) -> Result<(), JobTransitionError> {
+        self.require_running(JobState::Succeeded)?;
+        self.state = JobState::Succeeded;
+        self.result = Some(result);
+        self.error = None;
+        Ok(())
+    }
+
+    /// Complete a running job with a categorized failure.
+    pub fn fail(&mut self, error: JobError) -> Result<(), JobTransitionError> {
+        self.require_running(JobState::Failed)?;
+        self.state = JobState::Failed;
+        self.result = None;
+        self.error = Some(error);
+        Ok(())
+    }
+
+    fn require_running(&self, to: JobState) -> Result<(), JobTransitionError> {
+        if self.state != JobState::Running {
+            return Err(JobTransitionError {
+                from: self.state,
+                to,
+            });
+        }
+        Ok(())
+    }
+}
+
+impl JobStatus {
+    /// Construct a completed status from a synchronous runner result.
+    pub fn from_run(
+        id: JobId,
+        workflow: Workflow,
+        result: Result<RunResult, ServiceError>,
+    ) -> Self {
+        let mut status = Self::pending(id, workflow);
+        status
+            .start()
+            .expect("newly constructed job must transition to running");
+        match result {
+            Ok(result) => {
+                status
+                    .succeed(JobResult::from(&result))
+                    .expect("running job must transition to succeeded");
+            }
+            Err(error) => {
+                status
+                    .fail(JobError::from(&error))
+                    .expect("running job must transition to failed");
+            }
+        }
+        status
+    }
+}
+
+/// An attempted lifecycle transition that is not permitted.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct JobTransitionError {
+    /// State before the attempted transition.
+    pub from: JobState,
+    /// Requested destination state.
+    pub to: JobState,
+}
+
+/// Small synchronized store for the current in-memory job snapshots.
+#[derive(Clone, Default)]
+pub struct JobStore {
+    jobs: Arc<Mutex<HashMap<JobId, JobStatus>>>,
+}
+
+impl JobStore {
+    /// Insert an accepted job.
+    pub fn insert(&self, status: JobStatus) {
+        self.with_jobs(|jobs| {
+            jobs.insert(status.id.clone(), status);
+        });
+    }
+
+    /// Read a status snapshot, if the identifier is known.
+    pub fn get(&self, id: &str) -> Option<JobStatus> {
+        self.with_jobs(|jobs| jobs.get(id).cloned())
+    }
+
+    /// Apply one explicit transition while holding the store lock.
+    pub fn transition(
+        &self,
+        id: &str,
+        transition: impl FnOnce(&mut JobStatus) -> Result<(), JobTransitionError>,
+    ) -> Result<(), JobStoreError> {
+        self.with_jobs(|jobs| {
+            let status = jobs.get_mut(id).ok_or(JobStoreError::UnknownJob)?;
+            transition(status).map_err(JobStoreError::InvalidTransition)
+        })
+    }
+
+    fn with_jobs<T>(&self, operation: impl FnOnce(&mut HashMap<JobId, JobStatus>) -> T) -> T {
+        let mut jobs = self
+            .jobs
+            .lock()
+            .unwrap_or_else(|poisoned| poisoned.into_inner());
+        operation(&mut jobs)
+    }
+}
+
+/// Failure while updating the job store.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum JobStoreError {
+    /// No job exists for the requested identifier.
+    UnknownJob,
+    /// The requested state transition is invalid.
+    InvalidTransition(JobTransitionError),
+}
+
 /// Inputs to the high-level service runner.
 #[derive(Debug, Clone)]
 pub struct RunRequest {
@@ -42,3 +279,39 @@ pub struct RunResult {
 
 /// Application result name retained for callers of the previous service API.
 pub type RunReport = RunResult;
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn lifecycle_allows_success_only_from_running() {
+        let mut status = JobStatus::pending("job-1".into(), Workflow::Synthetic);
+        assert!(status.start().is_ok());
+        assert_eq!(status.state, JobState::Running);
+    }
+
+    #[test]
+    fn lifecycle_rejects_invalid_transitions() {
+        let mut status = JobStatus::pending("job-1".into(), Workflow::Synthetic);
+        assert!(status.start().is_ok());
+        assert!(status.start().is_err());
+        assert!(status
+            .fail(JobError {
+                code: JobErrorCode::Scientific,
+                message: "failed".into(),
+            })
+            .is_ok());
+        assert_eq!(status.state, JobState::Failed);
+        assert!(status.result.is_none());
+        assert!(status.error.is_some());
+        assert!(status.start().is_err());
+    }
+
+    #[test]
+    fn service_errors_map_to_stable_codes() {
+        let error = JobError::from(&ServiceError::unsupported("not available"));
+        assert_eq!(error.code, JobErrorCode::Unsupported);
+        assert_eq!(serde_json::to_value(error).unwrap()["code"], "unsupported");
+    }
+}

src/service/mod.rs

@@ -49,7 +49,10 @@ pub mod workflow;
 pub use artifacts::Artifacts;
 pub use config::{ForcesConfig, InputsConfig, RunConfig};
 pub use error::ServiceError;
-pub use job::{JobId, RunReport, RunRequest, RunResult};
+pub use job::{
+    JobError, JobErrorCode, JobId, JobResult, JobState, JobStatus, JobStore, JobStoreError,
+    JobTransitionError, RunReport, RunRequest, RunResult,
+};
 pub use provenance::RunProvenance;
-pub use runner::{run, Runner};
+pub use runner::{execute, run, Runner};
 pub use workflow::Workflow;