feltdb 0.4.0

//! Durable workload authority. Workers execute; this state machine decides.
use crate::{
    authorization::{SignedGrant, Subject},
    AtomicMutation, FeltDb, FlowError, RecordPrecondition, StoredRow,
};
use rand::{distributions::Alphanumeric, Rng};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use sha2::{Digest, Sha256};
use std::collections::{BTreeMap, BTreeSet};
use std::path::Path;
pub const WORKLOAD_CONTRACT_VERSION: u32 = 1;
pub const DURABLE_WORK_COLLECTION: &str = "system_durable_work";
/// Workloads admitted at this contract version or later have their `timeouts`
/// enforced. Earlier records were admitted when timeouts were stored and
/// ignored; they keep that behaviour, so an upgrade never times out work whose
/// requester was told nothing would.
pub const DEADLINE_CONTRACT_VERSION: u32 = 2;

#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, PartialOrd, Ord)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub enum WorkloadState {
    Created,
    Ready,
    Claimed,
    Running,
    Checkpointed,
    Recovering,
    CancellationRequested,
    Succeeded,
    Failed,
    Cancelled,
    TimedOut,
    DeadLetter,
    /// Admitted with an approval policy and not yet decided: not claimable.
    AwaitingApproval,
    /// An authorized approver rejected it: final, never claimable.
    Rejected,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum BackoffPolicy {
    Fixed,
    Exponential,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkloadRetryPolicy {
    pub max_attempts: u32,
    pub initial_delay_ms: u64,
    pub max_delay_ms: u64,
    pub backoff: BackoffPolicy,
    #[serde(default)]
    pub retry_on: BTreeSet<String>,
}
impl Default for WorkloadRetryPolicy {
    fn default() -> Self {
        Self {
            max_attempts: 3,
            initial_delay_ms: 1000,
            max_delay_ms: 300_000,
            backoff: BackoffPolicy::Exponential,
            retry_on: [
                "timeout".into(),
                "worker_failure".into(),
                "transient_connection".into(),
            ]
            .into(),
        }
    }
}
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct WorkloadTimeouts {
    pub queue_ms: Option<u64>,
    pub claim_ms: Option<u64>,
    pub execution_ms: Option<u64>,
    pub heartbeat_ms: Option<u64>,
    pub overall_ms: Option<u64>,
    /// How long work may wait in `AWAITING_APPROVAL`.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub approval_ms: Option<u64>,
}
/// Who may decide a workload that requires approval.
///
/// Presence on admission is the requirement. A decision always needs the
/// authenticated decider to hold a signed `workload:approve` grant for the
/// workload; this policy narrows who that may be.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ApprovalPolicy {
    /// Subject kinds that may decide: `human` and/or `service`; required, so
    /// the choice is explicit. Agents, workers, workloads and connections can
    /// never decide.
    pub approver_kinds: BTreeSet<String>,
    /// When non-empty, only these subjects may decide.
    #[serde(default)]
    pub approvers: BTreeSet<Subject>,
}
impl Default for ApprovalPolicy {
    /// People approve: `human` only.
    fn default() -> Self {
        Self {
            approver_kinds: ["human".into()].into(),
            approvers: BTreeSet::new(),
        }
    }
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ApprovalOutcome {
    Approved,
    Rejected,
}
/// A committed approval decision. It is never replaced.
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct ApprovalDecision {
    pub outcome: ApprovalOutcome,
    pub decided_by: Subject,
    pub decided_at: i64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub reason: Option<String>,
    /// The grant that authorized the decision.
    pub grant_id: String,
    /// The input the decider reviewed: the decision binds to it.
    pub input_hash: String,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkloadApproval {
    pub policy: ApprovalPolicy,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub decision: Option<ApprovalDecision>,
}
impl WorkloadApproval {
    fn approved(&self) -> bool {
        self.decision
            .as_ref()
            .is_some_and(|d| d.outcome == ApprovalOutcome::Approved)
    }
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkloadDependency {
    pub workload_id: String,
    #[serde(default = "succeeded_state")]
    pub required_state: WorkloadState,
}
fn succeeded_state() -> WorkloadState {
    WorkloadState::Succeeded
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkloadClaim {
    pub claim_id: String,
    pub workload_id: String,
    pub worker_id: String,
    pub lease_id: String,
    pub fencing_token: u64,
    pub capability_snapshot: BTreeSet<String>,
    pub claimed_at: i64,
    pub expires_at: i64,
    pub last_heartbeat: i64,
}
#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "SCREAMING_SNAKE_CASE")]
pub enum AttemptStatus {
    Claimed,
    Running,
    Checkpointed,
    Succeeded,
    Failed,
    TimedOut,
    Cancelled,
    WorkerLost,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkloadAttempt {
    pub attempt_id: String,
    pub workload_id: String,
    pub worker_id: String,
    pub claim_id: String,
    pub fencing_token: u64,
    pub started_at: Option<i64>,
    pub ended_at: Option<i64>,
    pub input_hash: String,
    pub result_hash: Option<String>,
    pub error: Option<String>,
    #[serde(default)]
    pub resource_usage: BTreeMap<String, u64>,
    pub status: AttemptStatus,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkloadCheckpoint {
    pub checkpoint_id: String,
    pub workload_id: String,
    pub attempt_id: String,
    pub sequence: u64,
    pub state: Value,
    pub state_hash: String,
    pub created_at: i64,
    pub produced_by: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkloadResult {
    pub result_id: String,
    pub workload_id: String,
    pub attempt_id: String,
    pub status: String,
    pub output: Value,
    pub output_hash: String,
    #[serde(default)]
    pub artifacts: Vec<String>,
    #[serde(default)]
    pub provenance: BTreeMap<String, String>,
    pub produced_by: String,
    pub created_at: i64,
    pub schema_version: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkloadExecutionContext {
    pub workload_id: String,
    pub attempt_id: String,
    pub worker_id: String,
    pub capabilities: BTreeSet<String>,
    pub connection_scopes: BTreeSet<String>,
    pub secret_references: Vec<String>,
    pub expires_at: i64,
    pub fencing_token: u64,
    /// The work's input, so the claiming worker can execute it.
    #[serde(default)]
    pub input: Value,
    /// Every committed checkpoint, from this and earlier attempts, in order. A
    /// resumed attempt reads these to learn how far lost attempts got.
    #[serde(default)]
    pub checkpoints: Vec<WorkloadCheckpoint>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkloadEvent {
    pub sequence: u64,
    pub event_type: String,
    pub from: Option<WorkloadState>,
    pub to: WorkloadState,
    pub actor: String,
    pub at: i64,
    #[serde(default)]
    pub evidence: BTreeMap<String, String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Workload {
    pub contract_version: u32,
    pub workload_id: String,
    pub tenant_id: String,
    pub application_id: String,
    pub environment: String,
    pub definition_id: String,
    pub definition_revision: String,
    pub trigger: String,
    pub input: Value,
    pub input_hash: String,
    pub requested_by: Subject,
    #[serde(default)]
    pub session_id: Option<String>,
    pub authorization_snapshot: SignedGrant,
    pub capability_snapshot: BTreeSet<String>,
    pub connection_scopes: BTreeSet<String>,
    pub priority: i32,
    pub idempotency_key: String,
    pub created_at: i64,
    pub updated_at: i64,
    pub state: WorkloadState,
    pub execution_history: Vec<WorkloadEvent>,
    pub attempts: Vec<WorkloadAttempt>,
    pub checkpoints: Vec<WorkloadCheckpoint>,
    pub result: Option<WorkloadResult>,
    pub retry_policy: WorkloadRetryPolicy,
    pub timeouts: WorkloadTimeouts,
    pub attempt_count: u32,
    pub retry_count: u32,
    /// The instant, in unix seconds, before which this workload must not be
    /// claimed. `None` means "eligible as soon as it is `Ready`".
    ///
    /// This is the single temporal eligibility concept in the workload model.
    /// It is what retry backoff sets after a failure, what a timeout sets before
    /// re-arming, and what [`CreateWorkload::not_before`] sets at creation, so a
    /// workload that is not yet executable is always not-yet-executable for the
    /// same reason and is refused by the same check in [`WorkloadStore::claim`].
    pub next_attempt_at: Option<i64>,
    pub last_failure: Option<String>,
    pub current_claim: Option<WorkloadClaim>,
    pub fencing_token: u64,
    #[serde(default)]
    pub dependencies: Vec<WorkloadDependency>,
    /// The approval this work requires and, once committed, its decision.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub approval: Option<WorkloadApproval>,
}

impl Workload {
    /// Has this workload's eligibility instant arrived?
    ///
    /// Eligibility is only half of claimability — the state must also be
    /// `Ready`, and [`WorkloadStore::claim`] checks both. This answers the
    /// temporal half alone so a scheduler can skip a workload that is merely
    /// early without attempting a claim it knows will be refused.
    pub fn is_eligible_at(&self, now: i64) -> bool {
        self.next_attempt_at.is_none_or(|instant| instant <= now)
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CreateWorkload {
    pub tenant_id: String,
    pub application_id: String,
    pub environment: String,
    pub definition_id: String,
    pub definition_revision: String,
    pub trigger: String,
    pub input: Value,
    pub requested_by: Subject,
    #[serde(default)]
    pub session_id: Option<String>,
    pub authorization_snapshot: SignedGrant,
    pub capability_snapshot: BTreeSet<String>,
    #[serde(default)]
    pub connection_scopes: BTreeSet<String>,
    #[serde(default)]
    pub priority: i32,
    pub idempotency_key: String,
    #[serde(default)]
    pub retry_policy: WorkloadRetryPolicy,
    #[serde(default)]
    pub timeouts: WorkloadTimeouts,
    #[serde(default)]
    pub dependencies: Vec<WorkloadDependency>,
    /// The instant, in unix seconds, before which this workload must not be
    /// claimed. Omitted or `None` means it is claimable as soon as it is
    /// `Ready`, which is what every workload did before this field existed.
    ///
    /// A workload created with `not_before` is durable immediately and
    /// executable later: it reaches `Ready`, survives restart, and is refused to
    /// every worker until the instant arrives. It is *not* a schedule — there is
    /// no recurrence, no expression, and no component that re-arms it. Creating
    /// the next occurrence is the caller's business.
    ///
    /// A `not_before` already in the past is not an error. It simply describes a
    /// workload that is eligible now, which is the same thing as omitting it.
    #[serde(default)]
    pub not_before: Option<i64>,
    pub created_at: i64,
    /// Require an approval decision before the work may be claimed.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub approval: Option<ApprovalPolicy>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct WorkloadError {
    pub code: String,
    pub message: String,
    /// Structured detail for the caller, e.g. which request fields conflict.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub details: Option<Value>,
}
impl WorkloadError {
    fn new(code: &str, message: impl Into<String>) -> Self {
        Self {
            code: code.into(),
            message: message.into(),
            details: None,
        }
    }
}
impl std::fmt::Display for WorkloadError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}: {}", self.code, self.message)
    }
}
impl std::error::Error for WorkloadError {}

#[derive(Clone)]
pub struct WorkloadStore {
    db: FeltDb,
    pub workloads: BTreeMap<String, Workload>,
    idempotency: BTreeMap<String, String>,
    result_ids: BTreeMap<String, String>,
    /// Rebuildable cache of FeltDB-owned document versions. This is never
    /// serialized and never advances independently of a canonical commit.
    document_versions: BTreeMap<String, u64>,
    /// The store revision this projection last reconciled with. A store that
    /// has moved past it (a commit by another writer, or a replicated
    /// operation) is re-read before the projection answers.
    synced_revision: u64,
    /// Always false outside tests, which set it to act on the projection
    /// without re-reading the store: that reproduces another writer
    /// committing between a refresh and a commit — the window only the storage
    /// guard closes.
    refresh_disabled: bool,
}

#[derive(Serialize, Deserialize)]
struct LegacyWorkloadStore {
    #[serde(default)]
    workloads: BTreeMap<String, Workload>,
    #[serde(default)]
    idempotency: BTreeMap<String, String>,
    #[serde(default)]
    result_ids: BTreeMap<String, String>,
    #[serde(default)]
    event_sequence: u64,
}

impl WorkloadStore {
    pub fn load(path: impl AsRef<Path>) -> Result<Self, WorkloadError> {
        Self::from_db(FeltDb::open(path).map_err(io)?)
    }

    pub fn from_db(db: FeltDb) -> Result<Self, WorkloadError> {
        // Read before listing: a commit that lands in between is re-read by
        // the next refresh rather than missed.
        let synced_revision = db.current_revision().map_err(io)?;
        let mut workloads = BTreeMap::new();
        let mut document_versions = BTreeMap::new();
        for row in db.list_collection(DURABLE_WORK_COLLECTION).map_err(io)? {
            let version = row
                .value
                .get("__version")
                .and_then(Value::as_u64)
                // Canonical FeltDB CAS defines a pre-versioned document as
                // version 1. The next guarded mutation writes version 2, which
                // upgrades rows created before Work adopted RecordPrecondition.
                .unwrap_or(1);
            let workload: Workload = serde_json::from_value(row.value).map_err(io)?;
            document_versions.insert(workload.workload_id.clone(), version);
            workloads.insert(workload.workload_id.clone(), workload);
        }
        let idempotency = idempotency_index(&workloads, idempotency_scope);
        let mut result_ids = BTreeMap::new();
        for workload in workloads.values() {
            if let Some(result) = &workload.result {
                result_ids.insert(result.result_id.clone(), workload.workload_id.clone());
            }
        }
        Ok(Self {
            db,
            workloads,
            idempotency,
            result_ids,
            document_versions,
            synced_revision,
            refresh_disabled: false,
        })
    }

    /// Reconcile the projection with the committed store.
    ///
    /// The store, not this projection, is authoritative. When its revision has
    /// moved since the last reconciliation, every workload row whose document
    /// version differs from the projection's is re-read; rows are borrowed and
    /// only changed ones are cloned. Commits by another writer of the store and
    /// operations replicated from a peer both become visible here, without a
    /// restart.
    pub fn refresh(&mut self) -> Result<(), WorkloadError> {
        if self.refresh_disabled {
            return Ok(());
        }
        let revision = self.db.current_revision().map_err(io)?;
        if revision == self.synced_revision {
            return Ok(());
        }
        let known = &self.document_versions;
        let changed = self
            .db
            .query_collection(DURABLE_WORK_COLLECTION, None, |row| {
                let version = row
                    .value
                    .get("__version")
                    .and_then(Value::as_u64)
                    .unwrap_or(1);
                let id = row.key.rsplit('/').next().unwrap_or_default();
                known.get(id) != Some(&version)
            })
            .map_err(io)?;
        for row in changed {
            self.adopt(row)?;
        }
        self.synced_revision = revision;
        Ok(())
    }

    /// Re-read one workload from the store when its committed version differs
    /// from the projection's: a keyed read, not a scan. An id the projection
    /// has never seen (admitted by another writer) falls back to [`Self::refresh`].
    fn refresh_id(&mut self, id: &str) -> Result<(), WorkloadError> {
        match self.workloads.get(id).map(work_key) {
            Some(key) => self.refresh_key(&key).map(|_| ()),
            None => self.refresh(),
        }
    }

    fn refresh_key(&mut self, key: &str) -> Result<Option<Workload>, WorkloadError> {
        if self.refresh_disabled {
            let id = key.rsplit('/').next().unwrap_or_default();
            return Ok(self.workloads.get(id).cloned());
        }
        let Some(row) = self.db.live_row(DURABLE_WORK_COLLECTION, key) else {
            return Ok(None);
        };
        let version = row
            .value
            .get("__version")
            .and_then(Value::as_u64)
            .unwrap_or(1);
        let id = key.rsplit('/').next().unwrap_or_default();
        if self.document_versions.get(id) == Some(&version) {
            return Ok(self.workloads.get(id).cloned());
        }
        self.adopt(row).map(Some)
    }

    /// Install one committed row in the projection and its derived indexes.
    fn adopt(&mut self, row: StoredRow) -> Result<Workload, WorkloadError> {
        let version = row
            .value
            .get("__version")
            .and_then(Value::as_u64)
            .unwrap_or(1);
        let workload: Workload = serde_json::from_value(row.value).map_err(io)?;
        if let Some(result) = &workload.result {
            self.result_ids
                .insert(result.result_id.clone(), workload.workload_id.clone());
        }
        self.idempotency.insert(
            idempotency_scope(
                &workload.tenant_id,
                &workload.application_id,
                &workload.requested_by,
                &workload.idempotency_key,
            ),
            workload.workload_id.clone(),
        );
        self.document_versions
            .insert(workload.workload_id.clone(), version);
        self.workloads
            .insert(workload.workload_id.clone(), workload.clone());
        Ok(workload)
    }

    /// A workload as committed in the store (see [`Self::refresh`]).
    pub fn fresh(&mut self, id: &str) -> Result<&Workload, WorkloadError> {
        self.refresh_id(id)?;
        self.get(id)
    }

    /// An application's workloads as committed in the store.
    pub fn fresh_list(
        &mut self,
        tenant: &str,
        application: &str,
    ) -> Result<Vec<Workload>, WorkloadError> {
        self.refresh()?;
        Ok(self.list(tenant, application))
    }

    pub fn migrate_legacy_file(db: FeltDb, path: &Path) -> Result<Self, WorkloadError> {
        let bytes = std::fs::read(path).map_err(io)?;
        let legacy: LegacyWorkloadStore = serde_json::from_slice(&bytes).map_err(io)?;
        let mut store = Self::from_db(db)?;
        for (id, workload) in legacy.workloads {
            if id != workload.workload_id {
                return Err(WorkloadError::new(
                    "WORKLOAD_MIGRATION_CONFLICT",
                    format!(
                        "legacy key {id} does not match workload identity {}",
                        workload.workload_id
                    ),
                ));
            }
            if let Some(existing) = store.workloads.get(&id) {
                if serde_json::to_value(existing).map_err(io)?
                    != serde_json::to_value(&workload).map_err(io)?
                {
                    return Err(WorkloadError::new(
                        "WORKLOAD_MIGRATION_CONFLICT",
                        format!("canonical workload {id} differs from legacy source"),
                    ));
                }
                continue;
            }
            store.workloads.insert(id.clone(), workload);
            store.persist_workload(&id, "workload_migration")?;
        }
        // The maps are verification evidence, not migrated authorities.
        let rebuilt = Self::from_db(store.db.clone())?;
        if legacy.idempotency != idempotency_index(&rebuilt.workloads, legacy_idempotency_scope)
            || legacy.result_ids != rebuilt.result_ids
            || legacy.event_sequence != derived_event_sequence(&rebuilt.workloads)
        {
            return Err(WorkloadError::new(
                "WORKLOAD_MIGRATION_CONFLICT",
                "legacy derived indexes do not match canonical work aggregates",
            ));
        }
        Ok(rebuilt)
    }

    fn persist_workload(&mut self, id: &str, action: &str) -> Result<(), WorkloadError> {
        let workload = self
            .workloads
            .get(id)
            .ok_or_else(|| WorkloadError::new("WORKLOAD_NOT_FOUND", "workload not found"))?;
        let key = work_key(workload);
        let expected_version = self.document_versions.get(id).copied();
        let transaction_id = runtime_id("work_tx");
        let aggregate_digest = hash(&serde_json::to_value(workload).map_err(io)?);
        let commit = self
            .db
            .apply_atomic_transaction_guarded(
                &transaction_id,
                None,
                None,
                &[],
                &[RecordPrecondition {
                    capability: DURABLE_WORK_COLLECTION.into(),
                    key: key.clone(),
                    require_absent: expected_version.is_none(),
                    expected_version,
                    expected_epoch: None,
                    expected_lease_id: None,
                }],
                &[AtomicMutation {
                    capability: DURABLE_WORK_COLLECTION.into(),
                    key,
                    value: Some(serde_json::to_value(workload).map_err(io)?),
                }],
                Some(serde_json::json!({
                    "kind": action,
                    "subject": workload.requested_by.id(),
                    "principal_id": workload.requested_by.id(),
                    "agent_id": (workload.requested_by.kind() == "agent").then(|| workload.requested_by.id()),
                    "session_id": workload.session_id,
                    "delegator_id": workload.authorization_snapshot.grant.issuer.id(),
                    "authorization_decision_id": workload.authorization_snapshot.grant.id,
                    "tenant": workload.tenant_id,
                    "application": workload.application_id,
                    "work_id": workload.workload_id,
                    "operation_id": transaction_id,
                    "attempt_id": workload.attempts.last().map(|attempt| attempt.attempt_id.as_str()),
                    "worker_id": workload.current_claim.as_ref().map(|claim| claim.worker_id.as_str()),
                    "result_id": workload.result.as_ref().map(|result| result.result_id.as_str()),
                    "source_kind": (action == "workload_migration").then_some("WorkloadStore"),
                    "source_digest": aggregate_digest,
                    "migration_outcome": (action == "workload_migration").then_some("imported"),
                })),
            )
            .map_err(workload_storage_error)?;
        let version = commit
            .rows
            .first()
            .and_then(|row| row.value.get("__version"))
            .and_then(Value::as_u64)
            .ok_or_else(|| io("canonical work commit has no document version"))?;
        self.document_versions.insert(id.into(), version);
        Ok(())
    }
    pub fn create(&mut self, input: CreateWorkload) -> Result<Workload, WorkloadError> {
        self.admit(input).map(|(workload, _)| workload)
    }

    /// Create a workload, or return the one this requester already created
    /// under the same key.
    ///
    /// The key is scoped to the tenant, the application **and the requester**,
    /// so one requester's key can never return, reveal or block another's
    /// workload. A repeated key is answered with the original workload only when
    /// the request is the same — definition, revision, environment, input and
    /// capability snapshot; otherwise it is refused with
    /// `IDEMPOTENCY_KEY_REUSED` and nothing is created. The requester's
    /// authorization snapshot is checked before any replay. The boolean is
    /// `true` when this call created the workload.
    pub fn admit(&mut self, input: CreateWorkload) -> Result<(Workload, bool), WorkloadError> {
        ensure_safe(&input.input)?;
        if input.idempotency_key.trim().is_empty() {
            return Err(WorkloadError::new(
                "WORKLOAD_INVALID",
                "idempotency key is required",
            ));
        }
        let t = &input.timeouts;
        if [
            t.queue_ms,
            t.claim_ms,
            t.execution_ms,
            t.heartbeat_ms,
            t.overall_ms,
            t.approval_ms,
        ]
        .contains(&Some(0))
        {
            return Err(WorkloadError::new(
                "WORKLOAD_INVALID",
                "a timeout must be at least 1 ms; omit it for no deadline",
            ));
        }
        if t.approval_ms.is_some() && input.approval.is_none() {
            return Err(WorkloadError::new(
                "WORKLOAD_INVALID",
                "timeouts.approval_ms requires an approval policy",
            ));
        }
        if let Some(policy) = &input.approval {
            validate_approval_policy(policy)?;
        }
        if input.authorization_snapshot.grant.subject != input.requested_by
            || input.authorization_snapshot.grant.tenant_id != input.tenant_id
            || input.authorization_snapshot.grant.application_id != input.application_id
        {
            return Err(WorkloadError::new(
                "WORKLOAD_UNAUTHORIZED",
                "authorization snapshot does not bind the requester",
            ));
        }
        let idempotency = idempotency_scope(
            &input.tenant_id,
            &input.application_id,
            &input.requested_by,
            &input.idempotency_key,
        );
        // The scoped idempotency identity is the record identity. Competing
        // admissions therefore contend on one canonical require-absent CAS
        // even when their payloads differ; no separate durable uniqueness
        // index is needed.
        let id = format!("workload_{}", &hex(&Sha256::digest(&idempotency))[..24]);
        // Another writer, or a peer, may have admitted this identity: the
        // store is asked by key, not the projection.
        let committed_key = format!("{}/{}/{}", input.tenant_id, input.application_id, id);
        if let Some(existing) = self.refresh_key(&committed_key)? {
            return replay(existing, &input);
        }
        if let Some(existing_id) = self.idempotency.get(&idempotency) {
            let existing = self.workloads[existing_id].clone();
            return replay(existing, &input);
        }
        let request = input.clone();
        let input_hash = hash(&input.input);
        let mut workload = Workload {
            contract_version: DEADLINE_CONTRACT_VERSION,
            workload_id: id.clone(),
            tenant_id: input.tenant_id,
            application_id: input.application_id,
            environment: input.environment,
            definition_id: input.definition_id,
            definition_revision: input.definition_revision,
            trigger: input.trigger,
            input: input.input,
            input_hash,
            requested_by: input.requested_by,
            session_id: input.session_id,
            authorization_snapshot: input.authorization_snapshot,
            capability_snapshot: input.capability_snapshot,
            connection_scopes: input.connection_scopes,
            priority: input.priority,
            idempotency_key: input.idempotency_key,
            created_at: input.created_at,
            updated_at: input.created_at,
            state: WorkloadState::Created,
            execution_history: vec![],
            attempts: vec![],
            checkpoints: vec![],
            result: None,
            retry_policy: input.retry_policy,
            timeouts: input.timeouts,
            attempt_count: 0,
            retry_count: 0,
            // Creation-time eligibility and retry backoff are the same thing, so
            // they are the same field. A workload with `not_before` is created
            // exactly as any other and is simply not yet claimable.
            next_attempt_at: input.not_before,
            last_failure: None,
            current_claim: None,
            fencing_token: 0,
            dependencies: input.dependencies,
            approval: input.approval.map(|policy| WorkloadApproval {
                policy,
                decision: None,
            }),
        };
        // The durable history records the eligibility the workload was created
        // with, so "why was this Ready and unclaimed for an hour" is answerable
        // from the workload itself rather than from the caller's intent.
        let mut evidence: BTreeMap<String, String> = match workload.next_attempt_at {
            Some(instant) => [("not_before".into(), instant.to_string())].into(),
            None => BTreeMap::new(),
        };
        // Work that requires approval waits for it: it is never Ready, and so
        // never claimable, until an authorized decision is committed.
        let initial = if workload.approval.is_some() {
            evidence.insert("approval".into(), "required".into());
            WorkloadState::AwaitingApproval
        } else {
            WorkloadState::Ready
        };
        self.transition(&mut workload, initial, "rust", input.created_at, evidence)?;
        let key = work_key(&workload);
        self.idempotency.insert(idempotency.clone(), id.clone());
        self.workloads.insert(id.clone(), workload.clone());
        if let Err(error) = self.persist_workload(&id, "workload.created") {
            self.workloads.remove(&id);
            self.idempotency.remove(&idempotency);
            // Another writer of the same store (a second process, or a replica)
            // admitted this identity first. Converge on what it committed.
            if error.code == "WORKLOAD_CONFLICT" {
                if let Some(existing) = self.reload_committed(&key)? {
                    self.idempotency.insert(idempotency, id);
                    return replay(existing, &request);
                }
            }
            return Err(error);
        }
        Ok((workload, true))
    }
    pub fn get(&self, id: &str) -> Result<&Workload, WorkloadError> {
        self.workloads
            .get(id)
            .ok_or_else(|| WorkloadError::new("WORKLOAD_NOT_FOUND", "workload not found"))
    }
    pub fn list(&self, tenant: &str, application: &str) -> Vec<Workload> {
        let mut values = self
            .workloads
            .values()
            .filter(|v| v.tenant_id == tenant && v.application_id == application)
            .cloned()
            .collect::<Vec<_>>();
        values.sort_by_key(|v| (std::cmp::Reverse(v.priority), v.created_at));
        values
    }
    fn edit<R>(
        &mut self,
        id: &str,
        action: impl FnOnce(&mut Self, &mut Workload) -> Result<R, WorkloadError>,
    ) -> Result<R, WorkloadError> {
        // Act only on the committed version: a projection that missed another
        // writer's commit would otherwise lose its guarded commit at best.
        self.refresh_id(id)?;
        let workload = self
            .workloads
            .get(id)
            .cloned()
            .ok_or_else(|| WorkloadError::new("WORKLOAD_NOT_FOUND", "workload not found"))?;
        let mut edited = workload.clone();
        let result = action(self, &mut edited)?;
        self.workloads.insert(id.into(), edited);
        if let Err(error) = self.persist_workload(id, "workload.mutated") {
            self.workloads.insert(id.into(), workload.clone());
            // Another writer of the same store committed this workload first.
            // Nothing of this edit was committed; adopt what was, so this
            // projection does not keep acting on a version that no longer
            // exists (every later guarded commit from it would conflict).
            if error.code == "WORKLOAD_CONFLICT" {
                self.reload_committed(&work_key(&workload))?;
            }
            return Err(error);
        }
        Ok(result)
    }
    #[cfg(test)]
    pub(crate) fn db_for_tests(&self) -> FeltDb {
        self.db.clone()
    }
    /// See `refresh_disabled`.
    #[cfg(test)]
    pub(crate) fn without_refresh(mut self) -> Self {
        self.refresh_disabled = true;
        self
    }
    /// Replace this projection's copy of a workload with the committed row.
    fn reload_committed(&mut self, key: &str) -> Result<Option<Workload>, WorkloadError> {
        match self.db.live_row(DURABLE_WORK_COLLECTION, key) {
            Some(row) => self.adopt(row).map(Some),
            None => Ok(None),
        }
    }
    fn transition(
        &mut self,
        w: &mut Workload,
        to: WorkloadState,
        actor: &str,
        at: i64,
        evidence: BTreeMap<String, String>,
    ) -> Result<(), WorkloadError> {
        if !legal(w.state, to) {
            return Err(WorkloadError::new(
                "WORKLOAD_TRANSITION_DENIED",
                format!("{:?} cannot transition to {:?}", w.state, to),
            ));
        }
        let from = w.state;
        w.state = to;
        w.updated_at = at;
        let sequence = w
            .execution_history
            .last()
            .map_or(1, |event| event.sequence + 1);
        w.execution_history.push(WorkloadEvent {
            sequence,
            event_type: event_name(to).into(),
            from: Some(from),
            to,
            actor: actor.into(),
            at,
            evidence,
        });
        Ok(())
    }
    pub fn claim(
        &mut self,
        id: &str,
        worker: &str,
        capabilities: &BTreeSet<String>,
        now: i64,
        lease_ms: u64,
    ) -> Result<WorkloadClaim, WorkloadError> {
        self.enforce_deadline(id, now)?;
        // A Ready workload that has already used its budget (an operator
        // re-arm beyond it, or a record left by unbounded recovery before it
        // was bounded) is dead-lettered here rather than claimed again.
        let exhausted = self
            .get(id)
            .is_ok_and(|w| w.state == WorkloadState::Ready && w.attempt_count >= attempt_budget(w));
        if exhausted {
            self.edit(id, |store, w| {
                let evidence = [
                    ("reason".into(), "attempt_budget_exhausted".into()),
                    ("attempts_used".into(), w.attempt_count.to_string()),
                    ("attempt_budget".into(), attempt_budget(w).to_string()),
                ]
                .into();
                store.transition(w, WorkloadState::DeadLetter, "rust.claim", now, evidence)
            })?;
            return Err(WorkloadError::new(
                "WORKLOAD_ATTEMPTS_EXHAUSTED",
                "workload has used its attempt budget",
            ));
        }
        self.edit(id, |store, w| {
            // The approval gate, decided on the committed version: pending or
            // rejected work is refused, and — independently of its state — no
            // workload that requires approval is claimed without an approved
            // decision.
            match w.state {
                WorkloadState::AwaitingApproval => {
                    return Err(WorkloadError::new(
                        "WORKLOAD_APPROVAL_PENDING",
                        "workload is awaiting approval",
                    ))
                }
                WorkloadState::Rejected => {
                    return Err(WorkloadError::new(
                        "WORKLOAD_REJECTED",
                        "workload was rejected and will not run",
                    ))
                }
                _ => {}
            }
            if w.state != WorkloadState::Ready {
                return Err(WorkloadError::new(
                    "WORKLOAD_NOT_READY",
                    "workload is not eligible",
                ));
            }
            if w.approval.as_ref().is_some_and(|a| !a.approved()) {
                return Err(WorkloadError::new(
                    "WORKLOAD_APPROVAL_PENDING",
                    "workload has no approved decision",
                ));
            }
            // Separated from the state check because the two mean opposite
            // things to a worker: a workload in the wrong state may never become
            // claimable, while one that is merely early certainly will, at an
            // instant the error names. Collapsing them into one code left a
            // polling worker unable to tell "come back" from "give up".
            if !w.is_eligible_at(now) {
                return Err(WorkloadError::new(
                    "WORKLOAD_NOT_YET_ELIGIBLE",
                    format!(
                        "workload is not claimable until {}",
                        w.next_attempt_at.unwrap_or(now)
                    ),
                ));
            }
            if !w.capability_snapshot.is_subset(capabilities) {
                return Err(WorkloadError::new(
                    "WORKLOAD_CAPABILITY_DENIED",
                    "worker lacks required capability snapshot",
                ));
            }
            w.fencing_token += 1;
            w.attempt_count += 1;
            let seed = format!("{}:{}:{}", w.workload_id, w.fencing_token, worker);
            let claim = WorkloadClaim {
                claim_id: format!("claim_{}", &hex(&Sha256::digest(seed.as_bytes()))[..20]),
                workload_id: w.workload_id.clone(),
                worker_id: worker.into(),
                lease_id: format!("lease_{}", w.fencing_token),
                fencing_token: w.fencing_token,
                capability_snapshot: w.capability_snapshot.clone(),
                claimed_at: now,
                expires_at: now + (lease_ms / 1000) as i64,
                last_heartbeat: now,
            };
            w.attempts.push(WorkloadAttempt {
                attempt_id: format!("attempt_{}_{}", w.workload_id, w.attempt_count),
                workload_id: w.workload_id.clone(),
                worker_id: worker.into(),
                claim_id: claim.claim_id.clone(),
                fencing_token: claim.fencing_token,
                started_at: None,
                ended_at: None,
                input_hash: w.input_hash.clone(),
                result_hash: None,
                error: None,
                resource_usage: BTreeMap::new(),
                status: AttemptStatus::Claimed,
            });
            w.current_claim = Some(claim.clone());
            store.transition(
                w,
                WorkloadState::Claimed,
                worker,
                now,
                [("fencing_token".into(), claim.fencing_token.to_string())].into(),
            )?;
            Ok(claim)
        })
    }
    fn fence(
        w: &Workload,
        claim: &str,
        worker: &str,
        token: u64,
        now: i64,
    ) -> Result<(), WorkloadError> {
        let current = w
            .current_claim
            .as_ref()
            .ok_or_else(|| WorkloadError::new("WORKLOAD_CLAIM_INVALID", "no active claim"))?;
        if current.claim_id != claim
            || current.worker_id != worker
            || current.fencing_token != token
        {
            return Err(WorkloadError::new(
                "WORKLOAD_STALE_FENCE",
                "claim or fencing token is stale",
            ));
        }
        if current.expires_at <= now {
            return Err(WorkloadError::new(
                "WORKLOAD_LEASE_EXPIRED",
                "claim lease expired",
            ));
        }
        Ok(())
    }
    pub fn start(
        &mut self,
        id: &str,
        claim: &str,
        worker: &str,
        token: u64,
        now: i64,
    ) -> Result<(), WorkloadError> {
        self.enforce_deadline(id, now)?;
        self.edit(id, |store, w| {
            Self::fence(w, claim, worker, token, now)?;
            let attempt = w.attempts.last_mut().unwrap();
            attempt.status = AttemptStatus::Running;
            attempt.started_at = Some(now);
            store.transition(w, WorkloadState::Running, worker, now, BTreeMap::new())
        })
    }
    pub fn heartbeat(
        &mut self,
        id: &str,
        claim: &str,
        worker: &str,
        token: u64,
        now: i64,
        extend_ms: u64,
    ) -> Result<WorkloadClaim, WorkloadError> {
        self.enforce_deadline(id, now)?;
        self.edit(id, |_, w| {
            Self::fence(w, claim, worker, token, now)?;
            let current = w.current_claim.as_mut().unwrap();
            current.last_heartbeat = now;
            current.expires_at = now + (extend_ms / 1000) as i64;
            Ok(current.clone())
        })
    }
    pub fn execution_context(
        &mut self,
        id: &str,
        claim: &str,
        worker: &str,
        token: u64,
        now: i64,
        secret_references: Vec<String>,
    ) -> Result<WorkloadExecutionContext, WorkloadError> {
        self.enforce_deadline(id, now)?;
        let workload = self.fresh(id)?;
        Self::fence(workload, claim, worker, token, now)?;
        if secret_references
            .iter()
            .any(|v| !v.starts_with("secret://"))
        {
            return Err(WorkloadError::new(
                "WORKLOAD_SECRET_REJECTED",
                "execution context accepts secret references only",
            ));
        }
        let current = workload.current_claim.as_ref().unwrap();
        Ok(WorkloadExecutionContext {
            workload_id: id.into(),
            attempt_id: workload.attempts.last().unwrap().attempt_id.clone(),
            worker_id: worker.into(),
            capabilities: current.capability_snapshot.clone(),
            connection_scopes: workload.connection_scopes.clone(),
            secret_references,
            expires_at: current.expires_at,
            fencing_token: token,
            input: workload.input.clone(),
            checkpoints: workload.checkpoints.clone(),
        })
    }
    pub fn checkpoint(
        &mut self,
        id: &str,
        claim: &str,
        worker: &str,
        token: u64,
        state: Value,
        now: i64,
    ) -> Result<WorkloadCheckpoint, WorkloadError> {
        ensure_safe(&state)?;
        self.enforce_deadline(id, now)?;
        self.edit(id, |store, w| {
            Self::fence(w, claim, worker, token, now)?;
            if !matches!(
                w.state,
                WorkloadState::Running | WorkloadState::Checkpointed
            ) {
                return Err(WorkloadError::new(
                    "WORKLOAD_TRANSITION_DENIED",
                    "checkpoint requires a running workload",
                ));
            }
            let attempt = w.attempts.last_mut().unwrap();
            attempt.status = AttemptStatus::Checkpointed;
            let sequence = w.checkpoints.len() as u64 + 1;
            let checkpoint = WorkloadCheckpoint {
                checkpoint_id: format!("checkpoint_{}_{}", w.workload_id, sequence),
                workload_id: w.workload_id.clone(),
                attempt_id: attempt.attempt_id.clone(),
                sequence,
                state_hash: hash(&state),
                state,
                created_at: now,
                produced_by: worker.into(),
            };
            w.checkpoints.push(checkpoint.clone());
            if w.state == WorkloadState::Running {
                store.transition(w, WorkloadState::Checkpointed, worker, now, BTreeMap::new())?;
            }
            Ok(checkpoint)
        })
    }
    pub fn complete(
        &mut self,
        id: &str,
        claim: &str,
        worker: &str,
        token: u64,
        result_id: &str,
        output: Value,
        artifacts: Vec<String>,
        schema_version: u64,
        now: i64,
    ) -> Result<WorkloadResult, WorkloadError> {
        ensure_safe(&output)?;
        if let Some(workload_id) = self.result_ids.get(result_id) {
            if workload_id == id {
                // A replay is the reporting attempt retrying the same report
                // (e.g. after a lost response). Anything else naming this
                // result id did not produce it and must not be told it landed.
                let result = self.get(id)?.result.clone().unwrap();
                let same_report = result.produced_by == worker
                    && result.provenance.get("claim_id").map(String::as_str) == Some(claim)
                    && result.provenance.get("fencing_token") == Some(&token.to_string())
                    && result.output_hash == hash(&output);
                if same_report {
                    return Ok(result);
                }
                return Err(WorkloadError::new(
                    "WORKLOAD_RESULT_CONFLICT",
                    "the result id was already recorded by a different report",
                ));
            }
            return Err(WorkloadError::new(
                "WORKLOAD_RESULT_DUPLICATE",
                "result id belongs to another workload",
            ));
        }
        self.enforce_deadline(id, now)?;
        let result = self.edit(id, |store, w| {
            Self::fence(w, claim, worker, token, now)?;
            if !matches!(
                w.state,
                WorkloadState::Running | WorkloadState::Checkpointed
            ) {
                return Err(WorkloadError::new(
                    "WORKLOAD_TRANSITION_DENIED",
                    "completion requires running state",
                ));
            }
            let attempt = w.attempts.last_mut().unwrap();
            let output_hash = hash(&output);
            attempt.status = AttemptStatus::Succeeded;
            attempt.ended_at = Some(now);
            attempt.result_hash = Some(output_hash.clone());
            let result = WorkloadResult {
                result_id: result_id.into(),
                workload_id: w.workload_id.clone(),
                attempt_id: attempt.attempt_id.clone(),
                status: "succeeded".into(),
                output,
                output_hash,
                artifacts,
                provenance: [
                    ("claim_id".into(), claim.into()),
                    ("fencing_token".into(), token.to_string()),
                ]
                .into(),
                produced_by: worker.into(),
                created_at: now,
                schema_version,
            };
            w.result = Some(result.clone());
            w.current_claim = None;
            store.transition(w, WorkloadState::Succeeded, worker, now, BTreeMap::new())?;
            Ok(result)
        })?;
        self.result_ids.insert(result_id.into(), id.into());
        Ok(result)
    }
    pub fn fail(
        &mut self,
        id: &str,
        claim: &str,
        worker: &str,
        token: u64,
        kind: &str,
        error: &str,
        now: i64,
    ) -> Result<WorkloadState, WorkloadError> {
        self.fail_with_effect(id, claim, worker, token, kind, error, None, now)
    }
    /// A worker's failure report for its current attempt.
    ///
    /// Accepted from `CLAIMED` (the worker could not start) as well as from a
    /// started attempt, fenced like every worker call. `effect` says what the
    /// worker knows about external effects: `none` — it failed before causing
    /// any (implied for an attempt that never started); `unknown` — an effect
    /// may have happened. Without it, the policy decides by `kind`, as before.
    /// An `unknown` effect is decided like a lost worker: retried only when the
    /// policy retries `worker_failure`, whatever `kind` says. Repeating the same
    /// report replays its outcome; any other report from an attempt that
    /// already reported gets the fence's answer (`CLAIM_INVALID`/`STALE_FENCE`).
    #[allow(clippy::too_many_arguments)]
    pub fn fail_with_effect(
        &mut self,
        id: &str,
        claim: &str,
        worker: &str,
        token: u64,
        kind: &str,
        error: &str,
        effect: Option<&str>,
        now: i64,
    ) -> Result<WorkloadState, WorkloadError> {
        if sensitive_text(error) {
            return Err(WorkloadError::new(
                "WORKLOAD_SECRET_REJECTED",
                "errors may not contain secret material",
            ));
        }
        if effect.is_some_and(|e| !matches!(e, "none" | "unknown")) {
            return Err(WorkloadError::new(
                "WORKLOAD_INVALID",
                "effect must be \"none\" or \"unknown\"",
            ));
        }
        // The same report repeated (a lost response) replays its outcome. Any
        // other report from an attempt that already reported is answered by
        // the fence, as before: its claim is no longer current.
        let current = self.fresh(id)?.clone();
        if let Some(attempt) = reported_attempt(&current, claim, worker, token) {
            let same = attempt.status == AttemptStatus::Failed
                && attempt.error.as_deref() == Some(error)
                && failure_event(&current, &attempt.attempt_id).is_some_and(|event| {
                    let recorded_effect = event.evidence.get("effect").map(String::as_str);
                    event.evidence.get("failure_kind").map(String::as_str) == Some(kind)
                        && (effect.is_none() || effect == recorded_effect)
                });
            if same {
                return Ok(current.state);
            }
        }
        self.enforce_deadline(id, now)?;
        self.edit(id, |store, w| {
            Self::fence(w, claim, worker, token, now)?;
            let started = w.state != WorkloadState::Claimed;
            let effect = effect.or((!started).then_some("none"));
            let attempt = w.attempts.last_mut().unwrap();
            attempt.status = AttemptStatus::Failed;
            attempt.ended_at = Some(now);
            attempt.error = Some(error.into());
            let attempt_id = attempt.attempt_id.clone();
            w.last_failure = Some(format!("{kind}: {error}"));
            w.current_claim = None;
            let mut evidence: BTreeMap<String, String> = [
                ("failure_kind".into(), kind.into()),
                ("attempt_id".into(), attempt_id),
            ]
            .into();
            if let Some(effect) = effect {
                evidence.insert("effect".into(), effect.into());
            }
            store.transition(w, WorkloadState::Failed, worker, now, evidence)?;
            let unknown = effect == Some("unknown");
            let allowed =
                w.retry_policy
                    .retry_on
                    .contains(if unknown { "worker_failure" } else { kind });
            let decision = if !allowed && unknown {
                "effect_unknown_not_retryable"
            } else if !allowed {
                "failure_kind_not_retryable"
            } else if w.attempt_count >= attempt_budget(w) {
                "attempt_budget_exhausted"
            } else {
                "retry"
            };
            let retryable = decision == "retry";
            let evidence: BTreeMap<String, String> = [("decision".into(), decision.into())].into();
            if retryable {
                w.retry_count += 1;
                let multiplier = if w.retry_policy.backoff == BackoffPolicy::Exponential {
                    2u64.saturating_pow(w.retry_count.saturating_sub(1))
                } else {
                    1
                };
                let delay = w
                    .retry_policy
                    .initial_delay_ms
                    .saturating_mul(multiplier)
                    .min(w.retry_policy.max_delay_ms);
                w.next_attempt_at = Some(now + (delay / 1000) as i64);
                store.transition(w, WorkloadState::Ready, "rust.retry", now, evidence)?;
            } else {
                store.transition(w, WorkloadState::DeadLetter, "rust.retry", now, evidence)?;
            }
            Ok(w.state)
        })
    }
    pub fn cancel(
        &mut self,
        id: &str,
        actor: &str,
        now: i64,
    ) -> Result<WorkloadState, WorkloadError> {
        self.edit(id, |store, w| {
            match w.state {
                WorkloadState::Ready
                | WorkloadState::Created
                | WorkloadState::AwaitingApproval
                | WorkloadState::Failed
                | WorkloadState::DeadLetter => {
                    w.current_claim = None;
                    store.transition(w, WorkloadState::Cancelled, actor, now, BTreeMap::new())?
                }
                WorkloadState::Claimed | WorkloadState::Running | WorkloadState::Checkpointed => {
                    store.transition(
                        w,
                        WorkloadState::CancellationRequested,
                        actor,
                        now,
                        BTreeMap::new(),
                    )?
                }
                WorkloadState::CancellationRequested => {}
                _ => {
                    return Err(WorkloadError::new(
                        "WORKLOAD_TRANSITION_DENIED",
                        "terminal workload cannot be cancelled",
                    ))
                }
            }
            Ok(w.state)
        })
    }
    pub fn confirm_cancel(
        &mut self,
        id: &str,
        claim: &str,
        worker: &str,
        token: u64,
        now: i64,
    ) -> Result<(), WorkloadError> {
        self.edit(id, |store, w| {
            Self::fence(w, claim, worker, token, now)?;
            if w.state != WorkloadState::CancellationRequested {
                return Err(WorkloadError::new(
                    "WORKLOAD_TRANSITION_DENIED",
                    "cancellation was not requested",
                ));
            }
            if let Some(attempt) = w.attempts.last_mut() {
                attempt.status = AttemptStatus::Cancelled;
                attempt.ended_at = Some(now);
            }
            w.current_claim = None;
            store.transition(w, WorkloadState::Cancelled, worker, now, BTreeMap::new())
        })
    }
    pub fn timeout(
        &mut self,
        id: &str,
        actor: &str,
        now: i64,
    ) -> Result<WorkloadState, WorkloadError> {
        self.edit(id, |store, w| {
            if !matches!(
                w.state,
                WorkloadState::Ready
                    | WorkloadState::Claimed
                    | WorkloadState::Running
                    | WorkloadState::Checkpointed
            ) {
                return Err(WorkloadError::new(
                    "WORKLOAD_TRANSITION_DENIED",
                    "workload cannot time out in its current state",
                ));
            }
            // A Ready workload has no attempt in progress: the last attempt's
            // recorded outcome is not this timeout's to overwrite.
            if matches!(
                w.state,
                WorkloadState::Claimed | WorkloadState::Running | WorkloadState::Checkpointed
            ) {
                if let Some(attempt) = w.attempts.last_mut() {
                    attempt.status = AttemptStatus::TimedOut;
                    attempt.ended_at = Some(now);
                }
            }
            w.current_claim = None;
            store.transition(w, WorkloadState::TimedOut, actor, now, BTreeMap::new())?;
            if w.attempt_count < attempt_budget(w) && w.retry_policy.retry_on.contains("timeout")
            {
                w.retry_count += 1;
                w.next_attempt_at = Some(now + (w.retry_policy.initial_delay_ms / 1000) as i64);
                store.transition(w, WorkloadState::Ready, "rust.retry", now, BTreeMap::new())?;
            } else {
                store.transition(
                    w,
                    WorkloadState::DeadLetter,
                    "rust.timeout",
                    now,
                    BTreeMap::new(),
                )?;
            }
            Ok(w.state)
        })
    }
    pub fn retry(&mut self, id: &str, actor: &str, now: i64) -> Result<(), WorkloadError> {
        self.retry_with_reason(id, actor, now, None)
    }
    /// An operator's re-arm of failed, dead-lettered or timed-out work.
    ///
    /// Each re-arm grants one more attempt, as it always has, and is recorded
    /// in the history with its actor and reason. A workload may be re-armed at
    /// most `retry_policy.max_attempts` times, so operator retries cannot
    /// become an unbounded loop.
    pub fn retry_with_reason(
        &mut self,
        id: &str,
        actor: &str,
        now: i64,
        reason: Option<&str>,
    ) -> Result<(), WorkloadError> {
        self.edit(id, |store, w| {
            if !matches!(
                w.state,
                WorkloadState::Failed | WorkloadState::DeadLetter | WorkloadState::TimedOut
            ) {
                return Err(WorkloadError::new(
                    "WORKLOAD_TRANSITION_DENIED",
                    "workload is not retryable",
                ));
            }
            let previous = operator_retries(w);
            if previous >= w.retry_policy.max_attempts {
                return Err(WorkloadError::new(
                    "WORKLOAD_RETRY_EXHAUSTED",
                    "workload has used every operator retry its policy allows",
                ));
            }
            w.next_attempt_at = Some(now);
            w.current_claim = None;
            let mut evidence: BTreeMap<String, String> = [
                ("retry".into(), "operator".into()),
                ("operator_retry".into(), (previous + 1).to_string()),
                (
                    "attempt_budget".into(),
                    w.retry_policy
                        .max_attempts
                        .saturating_add(previous + 1)
                        .to_string(),
                ),
            ]
            .into();
            if let Some(reason) = reason.filter(|reason| !reason.trim().is_empty()) {
                evidence.insert("reason".into(), reason.chars().take(256).collect());
            }
            // A re-arm never stands in for approval: work that was never
            // approved (it timed out waiting) waits for a decision again.
            let to = if w.approval.as_ref().is_some_and(|a| !a.approved()) {
                WorkloadState::AwaitingApproval
            } else {
                WorkloadState::Ready
            };
            store.transition(w, to, actor, now, evidence)
        })
    }
    /// Commit an approval decision for work that requires one.
    ///
    /// `decider` is the authenticated subject, already verified by the caller
    /// to hold a `workload:approve` grant for this workload. Here the policy is
    /// applied: the decider's kind must be one the policy allows (never an
    /// agent, worker, workload or connection), it must be listed when the
    /// policy lists approvers, and it may not be the requester. The decision
    /// binds to `input_hash`, the input the decider reviewed.
    ///
    /// The first committed decision is final. The same outcome again returns
    /// it (`true`: a replay, nothing committed); the opposite outcome is
    /// `WORKLOAD_APPROVAL_DECIDED`. Work that is no longer awaiting approval
    /// (timed out, cancelled) cannot be decided.
    #[allow(clippy::too_many_arguments)]
    pub fn decide(
        &mut self,
        id: &str,
        decider: &Subject,
        outcome: ApprovalOutcome,
        reason: Option<&str>,
        grant_id: &str,
        input_hash: &str,
        now: i64,
    ) -> Result<(ApprovalDecision, bool), WorkloadError> {
        let reason = reason.map(str::trim).filter(|r| !r.is_empty());
        if outcome == ApprovalOutcome::Rejected && reason.is_none() {
            return Err(WorkloadError::new(
                "WORKLOAD_INVALID",
                "a rejection requires a reason",
            ));
        }
        if reason.is_some_and(sensitive_text) {
            return Err(WorkloadError::new(
                "WORKLOAD_SECRET_REJECTED",
                "reasons may not contain secret material",
            ));
        }
        let current = self.fresh(id)?.clone();
        let Some(approval) = &current.approval else {
            return Err(WorkloadError::new(
                "WORKLOAD_APPROVAL_NOT_REQUIRED",
                "this workload was not admitted with an approval requirement",
            ));
        };
        approver_permitted(&approval.policy, &current.requested_by, decider)?;
        if current.input_hash != input_hash {
            let mut error = WorkloadError::new(
                "WORKLOAD_APPROVAL_STALE",
                "the decision names input this workload does not have",
            );
            error.details = Some(serde_json::json!({ "input_hash": current.input_hash }));
            return Err(error);
        }
        if let Some(decided) = &approval.decision {
            return decided_answer(decided, outcome);
        }
        self.enforce_deadline(id, now)?;
        let decision = self.edit(id, |store, w| {
            // Decide on the committed version: a decision, timeout or
            // cancellation that committed first stands.
            if w.approval.as_ref().is_some_and(|a| a.decision.is_some()) {
                return Err(WorkloadError::new(
                    "WORKLOAD_CONFLICT",
                    "another writer committed a decision first; ask again for it",
                ));
            }
            if w.state != WorkloadState::AwaitingApproval {
                let mut error = WorkloadError::new(
                    "WORKLOAD_APPROVAL_NOT_ACTIONABLE",
                    "the workload is no longer awaiting approval",
                );
                error.details = Some(serde_json::json!({ "state": w.state }));
                return Err(error);
            }
            let decision = ApprovalDecision {
                outcome,
                decided_by: decider.clone(),
                decided_at: now,
                reason: reason.map(|r| r.chars().take(512).collect()),
                grant_id: grant_id.into(),
                input_hash: input_hash.into(),
            };
            w.approval.as_mut().unwrap().decision = Some(decision.clone());
            let mut evidence: BTreeMap<String, String> = [
                ("decided_by".into(), decider.id().into()),
                ("decider_kind".into(), decider.kind().into()),
                ("grant_id".into(), grant_id.into()),
            ]
            .into();
            if let Some(reason) = &decision.reason {
                evidence.insert("reason".into(), reason.clone());
            }
            let to = match outcome {
                ApprovalOutcome::Approved => WorkloadState::Ready,
                ApprovalOutcome::Rejected => WorkloadState::Rejected,
            };
            store.transition(w, to, decider.id(), now, evidence)?;
            Ok(decision)
        })?;
        Ok((decision, false))
    }
    /// Time out every workload past one of its deadlines (see
    /// [`deadline_breach`]), as the recovery pass does each second. Deadlines
    /// are computed from the workload's committed timestamps, so the decision
    /// is the same after a restart and on any writer of the store.
    pub fn enforce_deadlines(&mut self, now: i64) -> Result<Vec<String>, WorkloadError> {
        self.refresh()?;
        let due = self
            .workloads
            .values()
            .filter_map(|w| deadline_breach(w, now).map(|breach| (w.workload_id.clone(), breach)))
            .collect::<Vec<_>>();
        let mut expired = Vec::with_capacity(due.len());
        for (id, (deadline, at)) in due {
            match self.expire(&id, deadline, at, now) {
                Ok(()) => expired.push(id),
                Err(error) if error.code == "WORKLOAD_CONFLICT" => {}
                Err(error) => return Err(error),
            }
        }
        Ok(expired)
    }
    /// Refuse a worker call on a workload past a deadline, timing it out
    /// first, in the same way the recovery pass would have.
    fn enforce_deadline(&mut self, id: &str, now: i64) -> Result<(), WorkloadError> {
        let Some((deadline, at)) = deadline_breach(self.fresh(id)?, now) else {
            return Ok(());
        };
        self.expire(id, deadline, at, now)?;
        let mut error = WorkloadError::new(
            "WORKLOAD_TIMED_OUT",
            format!("the workload's {deadline} deadline passed"),
        );
        error.details = Some(serde_json::json!({ "deadline": deadline, "deadline_at": at }));
        Err(error)
    }
    fn expire(&mut self, id: &str, deadline: &str, at: i64, now: i64) -> Result<(), WorkloadError> {
        self.edit(id, |store, w| {
            // Re-check on the committed version: a heartbeat or completion that
            // committed first decides, and this changes nothing.
            if deadline_breach(w, now).map(|(kind, _)| kind) != Some(deadline) {
                return Err(WorkloadError::new(
                    "WORKLOAD_CONFLICT",
                    "the workload changed before its deadline was enforced",
                ));
            }
            let lost = w.current_claim.take();
            if matches!(
                w.state,
                WorkloadState::Claimed | WorkloadState::Running | WorkloadState::Checkpointed
            ) {
                if let Some(attempt) = w.attempts.last_mut() {
                    attempt.status = AttemptStatus::TimedOut;
                    attempt.ended_at = Some(now);
                    attempt.error = Some(format!("timeout: {deadline} deadline passed"));
                }
            }
            let mut evidence: BTreeMap<String, String> = [
                ("deadline".into(), deadline.into()),
                ("deadline_at".into(), at.to_string()),
            ]
            .into();
            if let Some(claim) = &lost {
                evidence.insert("claim_id".into(), claim.claim_id.clone());
                evidence.insert("worker_id".into(), claim.worker_id.clone());
            }
            store.transition(w, WorkloadState::TimedOut, "rust.deadline", now, evidence)?;
            // The overall, queue and approval deadlines bound the work, not an
            // attempt: another attempt cannot meet them, and work whose approval
            // wait expired never becomes runnable by a late decision.
            let decision = if matches!(deadline, "overall" | "queue" | "approval") {
                "deadline_not_retryable"
            } else if !w.retry_policy.retry_on.contains("timeout") {
                "timeout_not_retryable"
            } else if w.attempt_count >= attempt_budget(w) {
                "attempt_budget_exhausted"
            } else {
                "retry"
            };
            let evidence: BTreeMap<String, String> = [("decision".into(), decision.into())].into();
            if decision == "retry" {
                w.retry_count += 1;
                w.next_attempt_at = Some(now + (w.retry_policy.initial_delay_ms / 1000) as i64);
                store.transition(w, WorkloadState::Ready, "rust.deadline", now, evidence)?;
            } else {
                store.transition(w, WorkloadState::DeadLetter, "rust.deadline", now, evidence)?;
            }
            Ok(())
        })
    }
    pub fn recover_expired(&mut self, now: i64) -> Result<Vec<String>, WorkloadError> {
        self.refresh()?;
        let ids = self
            .workloads
            .values()
            .filter(|w| {
                w.current_claim
                    .as_ref()
                    .is_some_and(|c| c.expires_at <= now)
                    && matches!(
                        w.state,
                        WorkloadState::Claimed
                            | WorkloadState::Running
                            | WorkloadState::Checkpointed
                            | WorkloadState::CancellationRequested
                    )
            })
            .map(|w| w.workload_id.clone())
            .collect::<Vec<_>>();
        let mut recovered = Vec::with_capacity(ids.len());
        for id in ids {
            let outcome = self.edit(&id, |store, w| {
                // Decide on the committed version: another writer may have
                // recovered (or the worker completed) it since it was listed.
                let still_expired = w
                    .current_claim
                    .as_ref()
                    .is_some_and(|c| c.expires_at <= now)
                    && matches!(
                        w.state,
                        WorkloadState::Claimed
                            | WorkloadState::Running
                            | WorkloadState::Checkpointed
                            | WorkloadState::CancellationRequested
                    );
                if !still_expired {
                    return Err(WorkloadError::new(
                        "WORKLOAD_CONFLICT",
                        "the workload changed before it was recovered",
                    ));
                }
                let lost = w.current_claim.take();
                if let Some(attempt) = w.attempts.last_mut() {
                    attempt.status = AttemptStatus::WorkerLost;
                    attempt.ended_at = Some(now);
                    attempt.error = Some("worker_lost: claim lease expired".into());
                }
                let mut evidence: BTreeMap<String, String> = [
                    ("recovery_reason".into(), "lease_expired".into()),
                    ("attempts_used".into(), w.attempt_count.to_string()),
                    ("attempt_budget".into(), attempt_budget(w).to_string()),
                ]
                .into();
                if let Some(claim) = &lost {
                    evidence.insert("claim_id".into(), claim.claim_id.clone());
                    evidence.insert("worker_id".into(), claim.worker_id.clone());
                    evidence.insert("lease_expired_at".into(), claim.expires_at.to_string());
                }
                if let Some(attempt) = w.attempts.last() {
                    evidence.insert("attempt_id".into(), attempt.attempt_id.clone());
                }
                if w.state == WorkloadState::CancellationRequested {
                    store.transition(w, WorkloadState::Cancelled, "rust.recovery", now, evidence)?;
                } else {
                    store.transition(
                        w,
                        WorkloadState::Recovering,
                        "rust.recovery",
                        now,
                        evidence,
                    )?;
                    // A lost worker is the policy's `worker_failure`. Whether
                    // the lost attempt had effects is unknown to FeltDB, so a
                    // retry re-executes it: at-least-once, never exactly-once.
                    let (to, decision) = if !w.retry_policy.retry_on.contains("worker_failure") {
                        (WorkloadState::DeadLetter, "worker_failure_not_retryable")
                    } else if w.attempt_count >= attempt_budget(w) {
                        (WorkloadState::DeadLetter, "attempt_budget_exhausted")
                    } else {
                        (WorkloadState::Ready, "retry")
                    };
                    store.transition(
                        w,
                        to,
                        "rust.recovery",
                        now,
                        [("decision".into(), decision.into())].into(),
                    )?;
                }
                Ok(())
            });
            match outcome {
                Ok(()) => recovered.push(id),
                // Another writer of this store committed the workload first —
                // in a race, its own recovery. Nothing of this attempt was
                // committed and `edit` has adopted the committed workload, so
                // the attempt is recovered once, by that writer. If it still
                // needs recovery, the next pass sees it.
                Err(error) if error.code == "WORKLOAD_CONFLICT" => {}
                Err(error) => return Err(error),
            }
        }
        Ok(recovered)
    }
}

pub trait WorkScheduler {
    fn enqueue(&mut self, workload_id: &str);
    fn claim(
        &mut self,
        store: &mut WorkloadStore,
        tenant: &str,
        application: &str,
        worker: &str,
        capabilities: &BTreeSet<String>,
        now: i64,
        lease_ms: u64,
    ) -> Result<Option<WorkloadClaim>, WorkloadError>;
    fn release(&mut self, claim_id: &str);
}
#[derive(Default)]
pub struct LocalDurableScheduler {
    queue: BTreeSet<String>,
}
impl WorkScheduler for LocalDurableScheduler {
    fn enqueue(&mut self, id: &str) {
        self.queue.insert(id.into());
    }
    fn claim(
        &mut self,
        store: &mut WorkloadStore,
        tenant: &str,
        application: &str,
        worker: &str,
        capabilities: &BTreeSet<String>,
        now: i64,
        lease_ms: u64,
    ) -> Result<Option<WorkloadClaim>, WorkloadError> {
        let selected = store
            .list(tenant, application)
            .into_iter()
            .filter(|w| {
                // Eligibility is part of selection, not a reason to fail. A
                // workload that is merely early is skipped so the scheduler can
                // hand out work that is ready now; without this it would pick an
                // ineligible workload, have `store.claim` refuse it, and return
                // an error while claimable work sat behind it.
                w.state == WorkloadState::Ready
                    && w.is_eligible_at(now)
                    && w.capability_snapshot.is_subset(capabilities)
            })
            .find(|w| {
                w.dependencies.iter().all(|d| {
                    store
                        .workloads
                        .get(&d.workload_id)
                        .is_some_and(|dependency| dependency.state == d.required_state)
                })
            })
            .map(|w| w.workload_id);
        match selected {
            Some(id) => store
                .claim(&id, worker, capabilities, now, lease_ms)
                .map(Some),
            None => Ok(None),
        }
    }
    fn release(&mut self, _: &str) {}
}
fn legal(from: WorkloadState, to: WorkloadState) -> bool {
    use WorkloadState::*;
    matches!(
        (from, to),
        (Created, Ready)
            | (Ready, Claimed)
            | (Claimed, Running)
            | (Running, Checkpointed)
            | (Checkpointed, Checkpointed)
            | (Running, Succeeded)
            | (Checkpointed, Succeeded)
            | (Claimed, CancellationRequested)
            | (Running, CancellationRequested)
            | (Checkpointed, CancellationRequested)
            | (CancellationRequested, Cancelled)
            | (Ready, Cancelled)
            | (Created, Cancelled)
            | (Failed, Cancelled)
            | (DeadLetter, Cancelled)
            | (Claimed, Failed)
            | (Running, Failed)
            | (Checkpointed, Failed)
            | (Claimed, Recovering)
            | (Running, Recovering)
            | (Checkpointed, Recovering)
            | (Recovering, Ready)
            | (Failed, Ready)
            | (TimedOut, Ready)
            | (DeadLetter, Ready)
            | (Failed, DeadLetter)
            | (Running, TimedOut)
            | (Claimed, TimedOut)
            | (Checkpointed, TimedOut)
            | (Ready, TimedOut)
            | (TimedOut, DeadLetter)
            | (Recovering, DeadLetter)
            | (Ready, DeadLetter)
            | (Created, AwaitingApproval)
            | (AwaitingApproval, Ready)
            | (AwaitingApproval, Rejected)
            | (AwaitingApproval, Cancelled)
            | (AwaitingApproval, TimedOut)
            | (TimedOut, AwaitingApproval)
            | (DeadLetter, AwaitingApproval)
            | (Failed, AwaitingApproval)
    )
}
pub fn can_transition(from: WorkloadState, to: WorkloadState) -> bool {
    legal(from, to)
}
fn event_name(state: WorkloadState) -> &'static str {
    use WorkloadState::*;
    match state {
        Created => "workload.created",
        Ready => "workload.ready",
        Claimed => "workload.claimed",
        Running => "workload.started",
        Checkpointed => "workload.checkpointed",
        Recovering => "workload.recovered",
        CancellationRequested => "workload.cancellation_requested",
        Succeeded => "workload.completed",
        Failed => "workload.failed",
        Cancelled => "workload.cancelled",
        TimedOut => "workload.timed_out",
        DeadLetter => "workload.dead_lettered",
        AwaitingApproval => "workload.awaiting_approval",
        Rejected => "workload.rejected",
    }
}
fn ensure_safe(value: &Value) -> Result<(), WorkloadError> {
    fn walk(value: &Value) -> bool {
        match value {
            Value::Object(map) => map.iter().any(|(k, v)| {
                matches!(
                    k.to_ascii_lowercase().as_str(),
                    "password" | "secret" | "token" | "api_key" | "private_key"
                ) || walk(v)
            }),
            Value::Array(values) => values.iter().any(walk),
            Value::String(v) => sensitive_text(v),
            _ => false,
        }
    }
    if walk(value) {
        Err(WorkloadError::new(
            "WORKLOAD_SECRET_REJECTED",
            "plaintext secrets are forbidden in workload state",
        ))
    } else {
        Ok(())
    }
}
fn sensitive_text(v: &str) -> bool {
    v.contains("BEGIN PRIVATE KEY") || v.starts_with("sk-") || v.starts_with("fdb_live_")
}
/// The identity a caller's idempotency key is unique within: tenant,
/// application and requester. Encoded as a JSON array, so no choice of key or
/// identifier can make two different scopes coincide.
fn idempotency_scope(tenant: &str, application: &str, requester: &Subject, key: &str) -> String {
    serde_json::json!([tenant, application, requester.kind(), requester.id(), key]).to_string()
}
/// The scope before keys were bound to their requester, kept only to verify
/// legacy store files during migration.
fn legacy_idempotency_scope(
    tenant: &str,
    application: &str,
    _requester: &Subject,
    key: &str,
) -> String {
    format!("{tenant}:{application}:{key}")
}
fn idempotency_index(
    workloads: &BTreeMap<String, Workload>,
    scope: fn(&str, &str, &Subject, &str) -> String,
) -> BTreeMap<String, String> {
    workloads
        .values()
        .map(|w| {
            (
                scope(&w.tenant_id, &w.application_id, &w.requested_by, &w.idempotency_key),
                w.workload_id.clone(),
            )
        })
        .collect()
}
/// Answer a repeated key: the original workload for the same request, a
/// refusal for a different one. Never the original for a different request.
/// The execution-affecting part of an admission, normalized, field by field.
///
/// A replay under the same key is answered with the original workload only when
/// every one of these is equal. Optional fields are compared after their
/// defaults apply (serde fills an omitted `retry_policy`, `timeouts`,
/// `priority`, `connection_scopes` or `dependencies`); sets compare as sets and
/// dependencies regardless of order; the input compares by its canonical hash,
/// so member order is not part of it. Provenance that does not change what runs
/// — `trigger`, `session_id`, `created_at`, and the authorization snapshot,
/// which is verified before any replay — is not compared.
fn execution_fields(
    definition_id: &str,
    definition_revision: &str,
    environment: &str,
    input_hash: &str,
    capability_snapshot: &BTreeSet<String>,
    connection_scopes: &BTreeSet<String>,
    priority: i32,
    retry_policy: &WorkloadRetryPolicy,
    timeouts: &WorkloadTimeouts,
    dependencies: &[WorkloadDependency],
    not_before: Option<i64>,
    approval: Option<&ApprovalPolicy>,
) -> Vec<(&'static str, Value)> {
    let mut dependencies = dependencies
        .iter()
        .map(|d| serde_json::json!([d.workload_id, d.required_state]))
        .collect::<Vec<_>>();
    dependencies.sort_by_key(|d| d.to_string());
    vec![
        ("definition_id", definition_id.into()),
        ("definition_revision", definition_revision.into()),
        ("environment", environment.into()),
        ("input", input_hash.into()),
        (
            "capability_snapshot",
            serde_json::json!(capability_snapshot),
        ),
        ("connection_scopes", serde_json::json!(connection_scopes)),
        ("priority", priority.into()),
        ("retry_policy", serde_json::json!(retry_policy)),
        ("timeouts", serde_json::json!(timeouts)),
        ("dependencies", Value::Array(dependencies)),
        ("not_before", serde_json::json!(not_before)),
        ("approval", serde_json::json!(approval)),
    ]
}
/// The `not_before` a workload was admitted with, from its creation event
/// (`next_attempt_at` moves with retries; the admission evidence does not).
fn admitted_not_before(w: &Workload) -> Option<i64> {
    w.execution_history
        .iter()
        .find(|event| event.to == WorkloadState::Ready)
        .and_then(|event| event.evidence.get("not_before"))
        .and_then(|value| value.parse().ok())
}
fn replay(existing: Workload, input: &CreateWorkload) -> Result<(Workload, bool), WorkloadError> {
    let recorded = execution_fields(
        &existing.definition_id,
        &existing.definition_revision,
        &existing.environment,
        &existing.input_hash,
        &existing.capability_snapshot,
        &existing.connection_scopes,
        existing.priority,
        &existing.retry_policy,
        &existing.timeouts,
        &existing.dependencies,
        admitted_not_before(&existing),
        existing.approval.as_ref().map(|a| &a.policy),
    );
    let requested = execution_fields(
        &input.definition_id,
        &input.definition_revision,
        &input.environment,
        &hash(&input.input),
        &input.capability_snapshot,
        &input.connection_scopes,
        input.priority,
        &input.retry_policy,
        &input.timeouts,
        &input.dependencies,
        input.not_before,
        input.approval.as_ref(),
    );
    let mut conflicting = recorded
        .iter()
        .zip(&requested)
        .filter(|((_, left), (_, right))| left != right)
        .map(|((name, _), _)| *name)
        .collect::<Vec<_>>();
    if existing.requested_by != input.requested_by {
        conflicting.insert(0, "requested_by");
    }
    if conflicting.is_empty() {
        return Ok((existing, false));
    }
    let mut error = WorkloadError::new(
        "IDEMPOTENCY_KEY_REUSED",
        "idempotency key was already used for a different request",
    );
    error.details = Some(serde_json::json!({ "fields": conflicting }));
    Err(error)
}
fn validate_approval_policy(policy: &ApprovalPolicy) -> Result<(), WorkloadError> {
    if policy.approver_kinds.is_empty()
        || policy
            .approver_kinds
            .iter()
            .any(|kind| !matches!(kind.as_str(), "human" | "service"))
    {
        return Err(WorkloadError::new(
            "WORKLOAD_INVALID",
            "approval.approver_kinds must be one or both of \"human\" and \"service\"",
        ));
    }
    if policy
        .approvers
        .iter()
        .any(|subject| !policy.approver_kinds.contains(subject.kind()))
    {
        return Err(WorkloadError::new(
            "WORKLOAD_INVALID",
            "every listed approver must be of a permitted approver kind",
        ));
    }
    Ok(())
}
/// Whether `decider` may decide under `policy`. Refusals are authorization
/// failures (`WORKLOAD_APPROVER_NOT_PERMITTED`).
fn approver_permitted(
    policy: &ApprovalPolicy,
    requested_by: &Subject,
    decider: &Subject,
) -> Result<(), WorkloadError> {
    let reason = if !policy.approver_kinds.contains(decider.kind()) {
        Some("this kind of subject may not approve this workload")
    } else if !policy.approvers.is_empty() && !policy.approvers.contains(decider) {
        Some("the decider is not one of this workload's approvers")
    } else if decider == requested_by {
        Some("the requester may not decide its own work")
    } else {
        None
    };
    match reason {
        Some(reason) => Err(WorkloadError::new(
            "WORKLOAD_APPROVER_NOT_PERMITTED",
            reason,
        )),
        None => Ok(()),
    }
}
/// The answer to a decision on work already decided: the same outcome
/// replays the committed decision; the opposite one is a conflict.
fn decided_answer(
    decided: &ApprovalDecision,
    outcome: ApprovalOutcome,
) -> Result<(ApprovalDecision, bool), WorkloadError> {
    if decided.outcome == outcome {
        return Ok((decided.clone(), true));
    }
    let mut error = WorkloadError::new(
        "WORKLOAD_APPROVAL_DECIDED",
        "a different decision was already committed",
    );
    error.details = Some(serde_json::json!({
        "outcome": decided.outcome,
        "decided_by": decided.decided_by,
        "decided_at": decided.decided_at,
    }));
    Err(error)
}
/// Seconds a millisecond timeout lasts, rounded up so a deadline is never early.
fn deadline_seconds(ms: u64) -> i64 {
    ms.div_ceil(1000) as i64
}
/// The first deadline a workload has passed at `now`, and when it fell.
///
/// | Deadline | Measured from | While |
/// | --- | --- | --- |
/// | `overall` | admission (`created_at`) | not terminal |
/// | `queue` | becoming `READY` (or its `next_attempt_at`, if later) | `READY` |
/// | `claim` | the claim | `CLAIMED` |
/// | `execution` | the attempt's start | `RUNNING`, `CHECKPOINTED` |
/// | `heartbeat` | the last heartbeat (or the claim) | claimed |
///
/// Only workloads admitted at [`DEADLINE_CONTRACT_VERSION`] or later.
pub fn deadline_breach(w: &Workload, now: i64) -> Option<(&'static str, i64)> {
    use WorkloadState::*;
    if w.contract_version < DEADLINE_CONTRACT_VERSION {
        return None;
    }
    let active = matches!(
        w.state,
        Ready | Claimed | Running | Checkpointed | AwaitingApproval
    );
    if !active {
        return None;
    }
    let t = &w.timeouts;
    let claim = w.current_claim.as_ref();
    let candidates = [
        (
            "approval",
            t.approval_ms
                .filter(|_| w.state == AwaitingApproval)
                .map(|ms| {
                    let waiting_since = w
                        .execution_history
                        .iter()
                        .rev()
                        .find(|event| event.to == AwaitingApproval)
                        .map_or(w.created_at, |event| event.at);
                    waiting_since + deadline_seconds(ms)
                }),
        ),
        (
            "overall",
            t.overall_ms.map(|ms| w.created_at + deadline_seconds(ms)),
        ),
        (
            "queue",
            t.queue_ms.filter(|_| w.state == Ready).map(|ms| {
                let ready_since = w
                    .execution_history
                    .iter()
                    .rev()
                    .find(|event| event.to == Ready)
                    .map_or(w.created_at, |event| event.at);
                ready_since.max(w.next_attempt_at.unwrap_or(i64::MIN)) + deadline_seconds(ms)
            }),
        ),
        (
            "claim",
            t.claim_ms
                .filter(|_| w.state == Claimed)
                .and_then(|ms| claim.map(|c| c.claimed_at + deadline_seconds(ms))),
        ),
        (
            "execution",
            t.execution_ms
                .filter(|_| matches!(w.state, Running | Checkpointed))
                .and_then(|ms| {
                    w.attempts
                        .last()
                        .and_then(|a| a.started_at)
                        .map(|started| started + deadline_seconds(ms))
                }),
        ),
        (
            "heartbeat",
            t.heartbeat_ms
                .and_then(|ms| claim.map(|c| c.last_heartbeat + deadline_seconds(ms))),
        ),
    ];
    candidates
        .into_iter()
        .find_map(|(kind, at)| at.filter(|at| now >= *at).map(|at| (kind, at)))
}
/// The attempt this fence named, when that attempt has already reported its
/// outcome (failed or succeeded). Lost and timed-out attempts are not reports.
fn reported_attempt<'w>(
    w: &'w Workload,
    claim: &str,
    worker: &str,
    token: u64,
) -> Option<&'w WorkloadAttempt> {
    w.attempts.iter().find(|attempt| {
        attempt.claim_id == claim
            && attempt.worker_id == worker
            && attempt.fencing_token == token
            && matches!(
                attempt.status,
                AttemptStatus::Failed | AttemptStatus::Succeeded
            )
    })
}
/// The `FAILED` event a failure report of `attempt_id` recorded.
fn failure_event<'w>(w: &'w Workload, attempt_id: &str) -> Option<&'w WorkloadEvent> {
    w.execution_history.iter().rev().find(|event| {
        event.to == WorkloadState::Failed
            && event.evidence.get("attempt_id").map(String::as_str) == Some(attempt_id)
    })
}
/// Operator re-arms recorded in the workload's own history.
fn operator_retries(w: &Workload) -> u32 {
    w.execution_history
        .iter()
        .filter(|event| event.evidence.get("retry").map(String::as_str) == Some("operator"))
        .count() as u32
}
/// How many claims the workload may have in total: its policy's
/// `max_attempts`, plus one for each operator re-arm (which is what an operator
/// retry has always granted).
fn attempt_budget(w: &Workload) -> u32 {
    w.retry_policy
        .max_attempts
        .saturating_add(operator_retries(w))
}
fn work_key(workload: &Workload) -> String {
    format!(
        "{}/{}/{}",
        workload.tenant_id, workload.application_id, workload.workload_id
    )
}
fn derived_event_sequence(workloads: &BTreeMap<String, Workload>) -> u64 {
    workloads
        .values()
        .flat_map(|workload| workload.execution_history.iter())
        .map(|event| event.sequence)
        .max()
        .unwrap_or_default()
}
fn runtime_id(prefix: &str) -> String {
    let value: String = rand::thread_rng()
        .sample_iter(&Alphanumeric)
        .take(24)
        .map(char::from)
        .collect();
    format!("{prefix}_{value}")
}
fn io(e: impl std::fmt::Display) -> WorkloadError {
    WorkloadError::new("WORKLOAD_STORAGE_FAILURE", e.to_string())
}
fn workload_storage_error(error: FlowError) -> WorkloadError {
    match error {
        FlowError::PreconditionFailed(_) | FlowError::RevisionConflict { .. } => {
            WorkloadError::new("WORKLOAD_CONFLICT", error.to_string())
        }
        FlowError::CapabilityError(ref message) if message.starts_with("PRECONDITION_FAILED:") => {
            WorkloadError::new("WORKLOAD_CONFLICT", error.to_string())
        }
        _ => io(error),
    }
}
fn hash(value: &Value) -> String {
    hex(&Sha256::digest(
        serde_json::to_vec(value).unwrap_or_default(),
    ))
}
fn hex(bytes: &[u8]) -> String {
    bytes.iter().map(|v| format!("{v:02x}")).collect()
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::authorization::{
        Grant, GrantConstraints, GrantSigner, GrantStatus, ResourceScope, ResourceScopeKind,
    };
    use std::{fs, path::PathBuf};
    fn create() -> CreateWorkload {
        create_as("a1")
    }
    /// The fixture request, made by agent `agent`.
    fn create_as(agent: &str) -> CreateWorkload {
        let signer = GrantSigner::new("k", vec![1; 32]).unwrap();
        let subject = Subject::Agent(agent.into());
        let grant = Grant {
            contract_version: 1,
            id: "g".into(),
            issuer: Subject::Human("h".into()),
            subject: subject.clone(),
            tenant_id: "t".into(),
            application_id: "app".into(),
            capabilities: ["workload:execute".into()].into(),
            resources: vec![ResourceScope {
                kind: ResourceScopeKind::Workload,
                uri: "flow://app/workloads/*".into(),
            }],
            constraints: GrantConstraints::default(),
            issued_at: 1,
            expires_at: 9999,
            parent_grant: None,
            revision: 1,
            status: GrantStatus::Active,
        };
        CreateWorkload {
            tenant_id: "t".into(),
            application_id: "app".into(),
            environment: "production".into(),
            definition_id: "send".into(),
            definition_revision: "r1".into(),
            trigger: "manual".into(),
            input: serde_json::json!({"id":1}),
            requested_by: subject,
            session_id: Some("session-1".into()),
            authorization_snapshot: signer.sign(grant).unwrap(),
            capability_snapshot: ["email:send".into()].into(),
            connection_scopes: BTreeSet::new(),
            priority: 0,
            idempotency_key: "one".into(),
            retry_policy: Default::default(),
            timeouts: Default::default(),
            dependencies: vec![],
            not_before: None,
            created_at: 10,
            approval: None,
        }
    }
    fn capabilities() -> BTreeSet<String> {
        ["email:send".into()].into()
    }

    #[test]
    fn idempotency_keys_are_bound_to_their_requester_and_request() {
        let (path, mut store) = store("requester-scoped-keys");
        let (original, created) = store.admit(create_as("a1")).unwrap();
        assert!(created);

        // The same requester and request: the original, not a second workload.
        let (replayed, created) = store.admit(create_as("a1")).unwrap();
        assert!(!created);
        assert_eq!(replayed.workload_id, original.workload_id);

        // The same key for a different request is refused and creates nothing.
        let mut different = create_as("a1");
        different.input = serde_json::json!({"id": 2});
        assert_eq!(store.admit(different).unwrap_err().code, "IDEMPOTENCY_KEY_REUSED");
        let mut other_definition = create_as("a1");
        other_definition.definition_id = "other".into();
        assert_eq!(
            store.admit(other_definition).unwrap_err().code,
            "IDEMPOTENCY_KEY_REUSED"
        );

        // Another requester's identical key is its own workload: it neither
        // returns, reveals nor blocks the first requester's.
        let (theirs, created) = store.admit(create_as("a2")).unwrap();
        assert!(created);
        assert_ne!(theirs.workload_id, original.workload_id);
        assert_eq!(store.workloads.len(), 2);

        // A snapshot that does not bind the requester is refused before any
        // replay, so a key cannot be used to fetch a workload.
        let mut unbound = create_as("a1");
        unbound.requested_by = Subject::Agent("a2".into());
        assert_eq!(store.admit(unbound).unwrap_err().code, "WORKLOAD_UNAUTHORIZED");
        let _ = fs::remove_file(path);
    }

    #[test]
    fn lease_recovery_records_the_lost_worker_and_is_bounded_by_the_attempt_budget() {
        let (path, mut store) = store("bounded-recovery");
        let workload = store.create(create()).unwrap();
        let id = workload.workload_id.clone();
        for attempt in 1..=3u32 {
            let now = 100 * attempt as i64;
            let claim = store.claim(&id, "worker-a", &capabilities(), now, 10_000).unwrap();
            assert_eq!(store.recover_expired(claim.expires_at).unwrap(), vec![id.clone()]);
            let recovered = store.get(&id).unwrap();
            let lost = recovered.attempts.last().unwrap();
            assert_eq!(lost.status, AttemptStatus::WorkerLost);
            assert_eq!(lost.claim_id, claim.claim_id);
            let recovering = recovered
                .execution_history
                .iter()
                .rev()
                .find(|event| event.to == WorkloadState::Recovering)
                .unwrap();
            assert_eq!(recovering.evidence["recovery_reason"], "lease_expired");
            assert_eq!(recovering.evidence["worker_id"], "worker-a");
            assert_eq!(recovering.evidence["claim_id"], claim.claim_id);
            assert_eq!(recovering.evidence["attempts_used"], attempt.to_string());
            let decision = &recovered.execution_history.last().unwrap();
            if attempt < 3 {
                assert_eq!(recovered.state, WorkloadState::Ready);
                assert_eq!(decision.evidence["decision"], "retry");
            } else {
                assert_eq!(recovered.state, WorkloadState::DeadLetter);
                assert_eq!(decision.evidence["decision"], "attempt_budget_exhausted");
            }
        }
        assert_eq!(store.get(&id).unwrap().attempt_count, 3);
        assert_eq!(
            store.claim(&id, "worker-b", &capabilities(), 1_000, 10_000).unwrap_err().code,
            "WORKLOAD_NOT_READY"
        );
        let _ = fs::remove_file(path);
    }

    /// The recovery history of `id` as committed, read by a fresh projection.
    fn committed_workload(db: &FeltDb, id: &str) -> Workload {
        WorkloadStore::from_db(db.clone()).unwrap().get(id).unwrap().clone()
    }

    fn recoveries(workload: &Workload) -> usize {
        workload
            .execution_history
            .iter()
            .filter(|event| event.to == WorkloadState::Recovering)
            .count()
    }

    /// Two writers of one store recover the same expired workload at once.
    fn race_recovery(db: &FeltDb, now: i64) -> (WorkloadStore, WorkloadStore, Vec<Vec<String>>) {
        let stores = [
            WorkloadStore::from_db(db.clone()).unwrap(),
            WorkloadStore::from_db(db.clone()).unwrap(),
        ];
        let barrier = std::sync::Barrier::new(2);
        let results = std::thread::scope(|scope| {
            let handles = stores
                .into_iter()
                .map(|mut store| {
                    let barrier = &barrier;
                    scope.spawn(move || {
                        barrier.wait();
                        let recovered = store.recover_expired(now).unwrap();
                        (store, recovered)
                    })
                })
                .collect::<Vec<_>>();
            handles
                .into_iter()
                .map(|handle| handle.join().unwrap())
                .collect::<Vec<_>>()
        });
        let mut results = results.into_iter();
        let (first, first_recovered) = results.next().unwrap();
        let (second, second_recovered) = results.next().unwrap();
        (first, second, vec![first_recovered, second_recovered])
    }

    #[test]
    fn concurrent_recovery_commits_one_transition_and_the_loser_converges() {
        for round in 0..20 {
            let (path, mut store) = store(&format!("concurrent-recovery-{round}"));
            let db = store.db.clone();
            let id = store.create(create()).unwrap().workload_id;
            let claim = store.claim(&id, "worker-a", &capabilities(), 100, 10_000).unwrap();
            // A writer that loaded the workload before either recovery.
            // Commits land in this projection's window between refresh and
            // commit: only the storage guard can refuse them.
            let mut stale = WorkloadStore::from_db(db.clone())
                .unwrap()
                .without_refresh();

            let (mut first, mut second, recovered) = race_recovery(&db, claim.expires_at);
            // Exactly one writer committed the recovery; the other reported none.
            let mut outcomes = recovered.clone();
            outcomes.sort();
            assert_eq!(outcomes, vec![vec![], vec![id.clone()]], "round {round}");

            let committed = committed_workload(&db, &id);
            assert_eq!(recoveries(&committed), 1, "one recovery transition");
            assert_eq!(committed.state, WorkloadState::Ready);
            assert_eq!(committed.attempt_count, 1);
            assert_eq!(committed.attempts.len(), 1);
            assert_eq!(committed.attempts[0].status, AttemptStatus::WorkerLost);
            assert!(committed.current_claim.is_none());
            let recovering = committed
                .execution_history
                .iter()
                .find(|event| event.to == WorkloadState::Recovering)
                .unwrap();
            assert_eq!(recovering.evidence["claim_id"], claim.claim_id);
            assert_eq!(recovering.evidence["attempts_used"], "1");

            // The losing writer adopted the committed workload instead of
            // keeping its own, uncommitted view.
            let expected = serde_json::to_value(&committed).unwrap();
            assert_eq!(serde_json::to_value(first.get(&id).unwrap()).unwrap(), expected);
            assert_eq!(serde_json::to_value(second.get(&id).unwrap()).unwrap(), expected);
            // And neither recovers it again, or commits anything more.
            assert!(first.recover_expired(claim.expires_at + 1).unwrap().is_empty());
            assert!(second.recover_expired(claim.expires_at + 1).unwrap().is_empty());
            assert_eq!(
                serde_json::to_value(committed_workload(&db, &id)).unwrap(),
                expected,
                "no further commit"
            );

            // The lost claimant cannot complete, even through a projection
            // that still shows its claim as current and unexpired: the guarded
            // commit refuses the stale version.
            let refused = stale
                .start(&id, &claim.claim_id, "worker-a", claim.fencing_token, claim.expires_at - 1)
                .unwrap_err();
            assert_eq!(refused.code, "WORKLOAD_CONFLICT");
            assert_eq!(stale.get(&id).unwrap().state, WorkloadState::Ready);
            let refused = stale
                .complete(
                    &id,
                    &claim.claim_id,
                    "worker-a",
                    claim.fencing_token,
                    "late-result",
                    serde_json::json!({"late": true}),
                    vec![],
                    1,
                    claim.expires_at - 1,
                )
                .unwrap_err();
            assert_eq!(refused.code, "WORKLOAD_CLAIM_INVALID");
            // Through a current projection its fence no longer matches.
            let refused = first
                .heartbeat(&id, &claim.claim_id, "worker-a", claim.fencing_token, claim.expires_at - 1, 10_000)
                .unwrap_err();
            assert_eq!(refused.code, "WORKLOAD_CLAIM_INVALID");
            assert_eq!(serde_json::to_value(committed_workload(&db, &id)).unwrap(), expected);
            drop((store, first, second, stale, db));
            let _ = fs::remove_file(path);
        }
    }

    #[test]
    fn concurrent_recovery_never_exceeds_the_attempt_budget() {
        let (path, mut store) = store("concurrent-recovery-budget");
        let db = store.db.clone();
        let id = store.create(create()).unwrap().workload_id;
        for attempt in 1..=3i64 {
            // Each attempt is claimed by an up-to-date writer and lost.
            let mut claimer = WorkloadStore::from_db(db.clone()).unwrap();
            let claim = claimer
                .claim(&id, "worker-a", &capabilities(), 100 * attempt, 10_000)
                .unwrap();
            let (_, _, recovered) = race_recovery(&db, claim.expires_at);
            assert_eq!(recovered.concat(), vec![id.clone()], "attempt {attempt}");
            let committed = committed_workload(&db, &id);
            assert_eq!(recoveries(&committed), attempt as usize);
            assert_eq!(committed.attempt_count, attempt as u32);
        }
        let committed = committed_workload(&db, &id);
        assert_eq!(committed.state, WorkloadState::DeadLetter);
        assert_eq!(
            committed.execution_history.last().unwrap().evidence["decision"],
            "attempt_budget_exhausted"
        );
        assert!(committed
            .attempts
            .iter()
            .all(|attempt| attempt.status == AttemptStatus::WorkerLost));
        let refused = WorkloadStore::from_db(db.clone())
            .unwrap()
            .claim(&id, "worker-b", &capabilities(), 10_000, 10_000)
            .unwrap_err();
        assert_eq!(refused.code, "WORKLOAD_NOT_READY");
        assert_eq!(committed_workload(&db, &id).attempt_count, 3);
        drop((store, db));
        let _ = fs::remove_file(path);
    }

    #[test]
    fn a_policy_that_does_not_retry_worker_failure_dead_letters_a_lost_attempt() {
        let (path, mut store) = store("no-worker-failure-retry");
        let mut request = create();
        request.retry_policy.retry_on = ["timeout".into()].into();
        let id = store.create(request).unwrap().workload_id;
        let claim = store.claim(&id, "worker-a", &capabilities(), 100, 10_000).unwrap();
        store.recover_expired(claim.expires_at).unwrap();
        let workload = store.get(&id).unwrap();
        assert_eq!(workload.state, WorkloadState::DeadLetter);
        assert_eq!(
            workload.execution_history.last().unwrap().evidence["decision"],
            "worker_failure_not_retryable"
        );
        let _ = fs::remove_file(path);
    }

    #[test]
    fn claiming_exhausted_ready_work_dead_letters_it_instead() {
        let (path, mut store) = store("exhausted-claim");
        let id = store.create(create()).unwrap().workload_id;
        // What unbounded recovery used to leave behind: Ready, budget spent.
        let mut legacy = store.get(&id).unwrap().clone();
        legacy.attempt_count = 7;
        store.workloads.insert(id.clone(), legacy);
        store.persist_workload(&id, "fixture.legacy").unwrap();
        let refused = store.claim(&id, "worker-a", &capabilities(), 100, 10_000).unwrap_err();
        assert_eq!(refused.code, "WORKLOAD_ATTEMPTS_EXHAUSTED");
        let workload = store.get(&id).unwrap();
        assert_eq!(workload.state, WorkloadState::DeadLetter);
        assert_eq!(
            workload.execution_history.last().unwrap().evidence["reason"],
            "attempt_budget_exhausted"
        );
        // Durable: a restart sees the same outcome.
        let reopened = WorkloadStore::from_db(store.db.clone()).unwrap();
        assert_eq!(reopened.get(&id).unwrap().state, WorkloadState::DeadLetter);
        let _ = fs::remove_file(path);
    }

    #[test]
    fn operator_retry_grants_one_attempt_is_recorded_and_is_bounded() {
        let (path, mut store) = store("operator-retry");
        let id = store.create(create()).unwrap().workload_id;
        // Claim and lose leases until the workload stops being claimable.
        fn exhaust(store: &mut WorkloadStore, id: &str, now: &mut i64) {
            loop {
                *now += 100;
                match store.claim(id, "worker-a", &capabilities(), *now, 10_000) {
                    Ok(claim) => {
                        store.recover_expired(claim.expires_at).unwrap();
                    }
                    Err(error) => {
                        assert!(
                            matches!(
                                error.code.as_str(),
                                "WORKLOAD_NOT_READY" | "WORKLOAD_ATTEMPTS_EXHAUSTED"
                            ),
                            "{error:?}"
                        );
                        return;
                    }
                }
            }
        }
        let mut now = 100;
        exhaust(&mut store, &id, &mut now);
        assert_eq!(store.get(&id).unwrap().attempt_count, 3);
        for retry in 1..=3u32 {
            now += 100;
            store
                .retry_with_reason(&id, "operator-1", now, Some("dependency restored"))
                .unwrap();
            let event = store.get(&id).unwrap().execution_history.last().unwrap().clone();
            assert_eq!(event.actor, "operator-1");
            assert_eq!(event.evidence["retry"], "operator");
            assert_eq!(event.evidence["reason"], "dependency restored");
            assert_eq!(event.evidence["operator_retry"], retry.to_string());
            exhaust(&mut store, &id, &mut now);
            assert_eq!(
                store.get(&id).unwrap().attempt_count,
                3 + retry,
                "each operator retry grants exactly one attempt"
            );
            assert_eq!(store.get(&id).unwrap().state, WorkloadState::DeadLetter);
        }
        assert_eq!(
            store.retry(&id, "operator-1", now + 100).unwrap_err().code,
            "WORKLOAD_RETRY_EXHAUSTED"
        );
        assert_eq!(store.get(&id).unwrap().state, WorkloadState::DeadLetter);
        let _ = fs::remove_file(path);
    }

    fn store(name: &str) -> (PathBuf, WorkloadStore) {
        let path =
            std::env::temp_dir().join(format!("workload-{name}-{}.json", std::process::id()));
        let store = WorkloadStore::load(&path).unwrap();
        (path, store)
    }
    #[test]
    fn legacy_migration_is_idempotent_conflict_detecting_and_source_independent() {
        let (source_db, mut source) = store("legacy-source-builder");
        let workload = source.create(create()).unwrap();
        let legacy = LegacyWorkloadStore {
            workloads: source.workloads.clone(),
            // What a pre-requester-scope release wrote.
            idempotency: idempotency_index(&source.workloads, legacy_idempotency_scope),
            result_ids: source.result_ids.clone(),
            event_sequence: derived_event_sequence(&source.workloads),
        };
        let legacy_path = source_db.with_extension("legacy-workloads.json");
        fs::write(&legacy_path, serde_json::to_vec(&legacy).unwrap()).unwrap();
        let canonical_path = source_db.with_extension("canonical.log");
        let db = FeltDb::open(&canonical_path).unwrap();
        let migrated = WorkloadStore::migrate_legacy_file(db.clone(), &legacy_path).unwrap();
        assert_eq!(
            migrated.get(&workload.workload_id).unwrap().input_hash,
            workload.input_hash
        );
        WorkloadStore::migrate_legacy_file(db.clone(), &legacy_path).unwrap();
        fs::remove_file(&legacy_path).unwrap();
        let reopened = WorkloadStore::from_db(FeltDb::open(&canonical_path).unwrap()).unwrap();
        assert_eq!(
            reopened.get(&workload.workload_id).unwrap().workload_id,
            workload.workload_id
        );

        let mut conflicting = legacy;
        conflicting
            .workloads
            .get_mut(&workload.workload_id)
            .unwrap()
            .priority += 1;
        fs::write(&legacy_path, serde_json::to_vec(&conflicting).unwrap()).unwrap();
        assert_eq!(
            WorkloadStore::migrate_legacy_file(db, &legacy_path)
                .err()
                .unwrap()
                .code,
            "WORKLOAD_MIGRATION_CONFLICT"
        );
        let _ = fs::remove_file(source_db);
        let _ = fs::remove_file(canonical_path);
        let _ = fs::remove_file(legacy_path);
    }

    #[test]
    fn canonical_history_reconstructs_work_identity_provenance() {
        let (path, mut store) = store("provenance");
        let workload = store.create(create()).unwrap();
        let transaction = store
            .db
            .committed_transactions_after(0, usize::MAX)
            .unwrap()
            .into_iter()
            .last()
            .unwrap();
        let audit = transaction.audit.unwrap();
        assert_eq!(audit["work_id"], workload.workload_id);
        assert_eq!(audit["session_id"], workload.session_id.as_deref().unwrap());
        assert_eq!(audit["agent_id"], workload.requested_by.id());
        assert_eq!(
            audit["delegator_id"],
            workload.authorization_snapshot.grant.issuer.id()
        );
        assert!(transaction.commit_revision > transaction.base_revision);
        assert_eq!(transaction.work_id.as_deref(), Some(workload.workload_id.as_str()));
        assert_eq!(transaction.agent_id.as_deref(), Some(workload.requested_by.id()));
        assert_eq!(transaction.session_id, workload.session_id);
        assert_eq!(
            transaction.delegator_id.as_deref(),
            Some(workload.authorization_snapshot.grant.issuer.id())
        );
        assert_eq!(transaction.operation_ids.len(), 1);
        assert_ne!(transaction.operation_ids[0], transaction.transaction_id);
        drop(store);
        let reopened = WorkloadStore::load(&path).unwrap();
        assert_eq!(
            reopened.get(&workload.workload_id).unwrap().workload_id,
            workload.workload_id
        );
        let _ = fs::remove_file(path);
    }

    #[test]
    fn competing_writers_use_canonical_document_cas_before_and_after_restart() {
        let (path, mut seed) = store("canonical-cas");
        let workload = seed.create(create()).unwrap();
        let id = workload.workload_id.clone();
        let db = seed.db.clone();
        let caps = ["email:send".into()].into();

        let mut winner = WorkloadStore::from_db(db.clone()).unwrap();
        // The loser's claim races the winner's inside the window between its
        // refresh and its commit, which only the storage guard closes.
        let mut loser = WorkloadStore::from_db(db.clone())
            .unwrap()
            .without_refresh();
        winner.claim(&id, "worker-a", &caps, 11, 1_000).unwrap();
        let conflict = loser.claim(&id, "worker-b", &caps, 11, 1_000).unwrap_err();
        assert_eq!(conflict.code, "WORKLOAD_CONFLICT");
        // The loser committed nothing and now projects the winner's claim.
        let projected = loser.get(&id).unwrap();
        assert_eq!(projected.state, WorkloadState::Claimed);
        assert_eq!(projected.current_claim.as_ref().unwrap().worker_id, "worker-a");

        let row = db
            .list_collection(DURABLE_WORK_COLLECTION)
            .unwrap()
            .into_iter()
            .find(|row| row.key.ends_with(&id))
            .unwrap();
        assert_eq!(row.value["__version"], 2);

        drop((seed, winner, loser, db));
        let restarted_db = FeltDb::open(&path).unwrap();
        let mut recovery = WorkloadStore::from_db(restarted_db.clone()).unwrap();
        recovery.recover_expired(13).unwrap();
        let mut restarted_winner = WorkloadStore::from_db(restarted_db.clone()).unwrap();
        let mut restarted_loser = WorkloadStore::from_db(restarted_db.clone())
            .unwrap()
            .without_refresh();
        restarted_winner
            .claim(&id, "worker-c", &caps, 13, 10_000)
            .unwrap();
        let conflict = restarted_loser
            .claim(&id, "worker-d", &caps, 13, 10_000)
            .unwrap_err();
        assert_eq!(conflict.code, "WORKLOAD_CONFLICT");
        let projected = restarted_loser.get(&id).unwrap();
        assert_eq!(projected.state, WorkloadState::Claimed);
        assert_eq!(projected.current_claim.as_ref().unwrap().worker_id, "worker-c");
        drop((recovery, restarted_winner, restarted_loser, restarted_db));
        let _ = fs::remove_file(path);
    }

    #[test]
    fn competing_admissions_share_one_canonical_idempotency_identity() {
        let path = std::env::temp_dir().join(format!(
            "workload-idempotency-cas-{}.log",
            std::process::id()
        ));
        let db = FeltDb::open(&path).unwrap();
        let mut first = WorkloadStore::from_db(db.clone()).unwrap();
        let mut second = WorkloadStore::from_db(db.clone()).unwrap();
        let accepted = first.create(create()).unwrap();
        // A second writer of the same store loses the require-absent CAS and
        // converges on what was committed: a different request is refused as a
        // reused key, never answered with the original.
        let mut competing = create();
        competing.input = serde_json::json!({"id": 2});
        let conflict = second.create(competing).unwrap_err();
        assert_eq!(conflict.code, "IDEMPOTENCY_KEY_REUSED");
        assert_eq!(
            second.workloads.keys().collect::<Vec<_>>(),
            vec![&accepted.workload_id],
            "only the committed workload is projected"
        );
        // The same request from a third, equally stale writer converges on it.
        let mut third = WorkloadStore::from_db(db.clone()).unwrap();
        third.workloads.clear();
        third.idempotency.clear();
        third.document_versions.clear();
        let (converged, created) = third.admit(create()).unwrap();
        assert!(!created);
        assert_eq!(converged.workload_id, accepted.workload_id);

        drop((first, second, third, db));
        let restarted = WorkloadStore::from_db(FeltDb::open(&path).unwrap()).unwrap();
        assert_eq!(restarted.workloads.len(), 1);
        assert_eq!(
            restarted.idempotency[&idempotency_scope("t", "app", &accepted.requested_by, "one")],
            accepted.workload_id
        );
        let _ = fs::remove_file(path);
    }

    #[test]
    fn idempotency_scopes_cannot_be_made_to_coincide() {
        let (path, mut store) = store("scope-separators");
        let mut first = create_as("x:k1");
        first.idempotency_key = "k2".into();
        let mut second = create_as("x");
        second.idempotency_key = "k1:k2".into();
        let first = store.create(first).unwrap();
        let second = store.create(second).unwrap();
        assert_ne!(first.workload_id, second.workload_id);
        let _ = fs::remove_file(path);
    }

    #[test]
    fn concurrent_admissions_by_competing_writers_converge_or_conflict() {
        for round in 0..20 {
            let (path, seed) = store(&format!("concurrent-admission-{round}"));
            let db = seed.db.clone();
            let admit_concurrently = |requests: [CreateWorkload; 2]| {
                let barrier = std::sync::Barrier::new(2);
                std::thread::scope(|scope| {
                    requests
                        .into_iter()
                        .map(|request| {
                            let mut writer = WorkloadStore::from_db(db.clone()).unwrap();
                            let barrier = &barrier;
                            scope.spawn(move || {
                                barrier.wait();
                                writer.admit(request)
                            })
                        })
                        .collect::<Vec<_>>()
                        .into_iter()
                        .map(|handle| handle.join().unwrap())
                        .collect::<Vec<_>>()
                })
            };
            // The same request: both writers answer with one workload, and
            // exactly one of them created it.
            let same = admit_concurrently([create(), create()]);
            let (first, first_created) = same[0].as_ref().unwrap().clone();
            let (second, second_created) = same[1].as_ref().unwrap().clone();
            assert_eq!(first.workload_id, second.workload_id);
            assert!(first_created ^ second_created, "round {round}");

            // Different requests under one key: one is admitted, the other is
            // refused as a reused key and never answered with the winner.
            let mut left = create();
            left.idempotency_key = "contested".into();
            let mut right = left.clone();
            right.input = serde_json::json!({"id": 2});
            let contested = admit_concurrently([left, right]);
            let admitted = contested.iter().filter(|outcome| outcome.is_ok()).count();
            assert_eq!(admitted, 1, "round {round}");
            assert!(contested.iter().any(|outcome| outcome
                .as_ref()
                .is_err_and(|error| error.code == "IDEMPOTENCY_KEY_REUSED")));
            let (winner, _) = contested.iter().find_map(|o| o.as_ref().ok()).unwrap();
            let stored = WorkloadStore::from_db(db.clone()).unwrap();
            assert_eq!(stored.workloads.len(), 2);
            assert_eq!(stored.get(&winner.workload_id).unwrap().input, winner.input);
            drop((seed, stored, db));
            let _ = fs::remove_file(path);
        }
    }

    #[test]
    fn pre_versioned_canonical_rows_upgrade_through_the_standard_cas_contract() {
        let (path, mut store) = store("pre-versioned-upgrade");
        let workload = store.create(create()).unwrap();
        let db = store.db.clone();
        db.apply_atomic_transaction(
            "fixture:pre-versioned-work",
            None,
            &[],
            &[AtomicMutation {
                capability: DURABLE_WORK_COLLECTION.into(),
                key: work_key(&workload),
                value: Some(serde_json::to_value(&workload).unwrap()),
            }],
            None,
        )
        .unwrap();

        let mut upgraded = WorkloadStore::from_db(db.clone()).unwrap();
        upgraded
            .claim(
                &workload.workload_id,
                "worker",
                &["email:send".into()].into(),
                11,
                10_000,
            )
            .unwrap();
        let row = db
            .list_collection(DURABLE_WORK_COLLECTION)
            .unwrap()
            .into_iter()
            .find(|row| row.key.ends_with(&workload.workload_id))
            .unwrap();
        assert_eq!(row.value["__version"], 2);
        let _ = fs::remove_file(path);
    }

    #[test]
    fn derived_indexes_rebuild_exclusively_from_canonical_work() {
        let (path, mut store) = store("derived-index-rebuild");
        let workload = store.create(create()).unwrap();
        let caps = ["email:send".into()].into();
        let claim = store
            .claim(&workload.workload_id, "worker", &caps, 11, 10_000)
            .unwrap();
        store
            .start(
                &workload.workload_id,
                &claim.claim_id,
                "worker",
                claim.fencing_token,
                12,
            )
            .unwrap();
        store
            .complete(
                &workload.workload_id,
                &claim.claim_id,
                "worker",
                claim.fencing_token,
                "result-rebuilt",
                serde_json::json!({"ok": true}),
                vec![],
                1,
                13,
            )
            .unwrap();

        store.idempotency.clear();
        store.result_ids.clear();
        store.document_versions.clear();
        let rebuilt = WorkloadStore::from_db(store.db.clone()).unwrap();
        assert_eq!(
            rebuilt
                .idempotency
                .get(&idempotency_scope("t", "app", &workload.requested_by, "one")),
            Some(&workload.workload_id)
        );
        assert_eq!(
            rebuilt.result_ids.get("result-rebuilt"),
            Some(&workload.workload_id)
        );
        assert!(rebuilt.document_versions[&workload.workload_id] > 1);
        assert_eq!(
            rebuilt
                .get(&workload.workload_id)
                .unwrap()
                .execution_history
                .iter()
                .map(|event| event.sequence)
                .collect::<Vec<_>>(),
            vec![1, 2, 3, 4]
        );
        let duplicate = WorkloadStore::from_db(store.db.clone())
            .unwrap()
            .create(create())
            .unwrap();
        assert_eq!(duplicate.workload_id, workload.workload_id);
        let _ = fs::remove_file(path);
    }
    #[test]
    fn idempotent_create_and_legal_lifecycle() {
        let (path, mut store) = store("lifecycle");
        let one = store.create(create()).unwrap();
        let two = store.create(create()).unwrap();
        assert_eq!(one.workload_id, two.workload_id);
        let caps = ["email:send".into()].into();
        let claim = store
            .claim(&one.workload_id, "worker", &caps, 11, 10_000)
            .unwrap();
        store
            .start(
                &one.workload_id,
                &claim.claim_id,
                "worker",
                claim.fencing_token,
                12,
            )
            .unwrap();
        let result = store
            .complete(
                &one.workload_id,
                &claim.claim_id,
                "worker",
                claim.fencing_token,
                "result-1",
                serde_json::json!({"ok":true}),
                vec![],
                1,
                13,
            )
            .unwrap();
        assert_eq!(result.result_id, "result-1");
        assert!(store
            .start(
                &one.workload_id,
                &claim.claim_id,
                "worker",
                claim.fencing_token,
                14
            )
            .is_err());
        let recovered = WorkloadStore::load(&path).unwrap();
        assert_eq!(
            recovered.get(&one.workload_id).unwrap().state,
            WorkloadState::Succeeded
        );
        let _ = fs::remove_file(path);
    }
    #[test]
    fn stale_fence_and_lease_recovery() {
        let (path, mut store) = store("fence");
        let w = store.create(create()).unwrap();
        let caps = ["email:send".into()].into();
        let old = store.claim(&w.workload_id, "a", &caps, 11, 1000).unwrap();
        store
            .start(&w.workload_id, &old.claim_id, "a", old.fencing_token, 11)
            .unwrap();
        store.recover_expired(13).unwrap();
        let new = store.claim(&w.workload_id, "b", &caps, 13, 10_000).unwrap();
        assert!(new.fencing_token > old.fencing_token);
        assert_eq!(
            store
                .complete(
                    &w.workload_id,
                    &old.claim_id,
                    "a",
                    old.fencing_token,
                    "stale",
                    Value::Null,
                    vec![],
                    1,
                    14
                )
                .unwrap_err()
                .code,
            "WORKLOAD_STALE_FENCE"
        );
        let _ = fs::remove_file(path);
    }
    #[test]
    fn retry_dead_letter_checkpoint_and_secret_safety() {
        let (path, mut store) = store("retry");
        let mut input = create();
        input.retry_policy.max_attempts = 1;
        let w = store.create(input).unwrap();
        let caps = ["email:send".into()].into();
        let claim = store.claim(&w.workload_id, "a", &caps, 11, 10_000).unwrap();
        store
            .start(
                &w.workload_id,
                &claim.claim_id,
                "a",
                claim.fencing_token,
                12,
            )
            .unwrap();
        assert!(store
            .checkpoint(
                &w.workload_id,
                &claim.claim_id,
                "a",
                claim.fencing_token,
                serde_json::json!({"password":"bad"}),
                13
            )
            .is_err());
        assert_eq!(
            store
                .fail(
                    &w.workload_id,
                    &claim.claim_id,
                    "a",
                    claim.fencing_token,
                    "worker_failure",
                    "crashed",
                    13
                )
                .unwrap(),
            WorkloadState::DeadLetter
        );
        let _ = fs::remove_file(path);
    }
    #[test]
    fn a_repeated_completion_replays_only_for_the_claim_and_output_that_produced_it() {
        let (path, mut store) = store("complete-replay");
        let w = store.create(create()).unwrap();
        let id = w.workload_id.clone();
        let caps = capabilities();
        // Worker a loses its lease; worker b completes the work.
        let lost = store.claim(&id, "a", &caps, 11, 1000).unwrap();
        store
            .start(&id, &lost.claim_id, "a", lost.fencing_token, 11)
            .unwrap();
        store.recover_expired(13).unwrap();
        let won = store.claim(&id, "b", &caps, 13, 10_000).unwrap();
        store
            .start(&id, &won.claim_id, "b", won.fencing_token, 13)
            .unwrap();
        let output = serde_json::json!({"sent": 1});
        let first = store
            .complete(
                &id,
                &won.claim_id,
                "b",
                won.fencing_token,
                "r1",
                output.clone(),
                vec![],
                1,
                14,
            )
            .unwrap();
        // The reporting attempt retrying after a lost response gets the same result.
        let replay = store
            .complete(
                &id,
                &won.claim_id,
                "b",
                won.fencing_token,
                "r1",
                output.clone(),
                vec![],
                1,
                15,
            )
            .unwrap();
        assert_eq!(replay.output_hash, first.output_hash);
        assert_eq!(replay.attempt_id, first.attempt_id);
        // The same result id with a different output is a conflict, not a replay.
        assert_eq!(
            store
                .complete(
                    &id,
                    &won.claim_id,
                    "b",
                    won.fencing_token,
                    "r1",
                    serde_json::json!({"sent": 2}),
                    vec![],
                    1,
                    15
                )
                .unwrap_err()
                .code,
            "WORKLOAD_RESULT_CONFLICT"
        );
        // A stale worker naming the winner's result id learns nothing landed for it.
        assert_eq!(
            store
                .complete(
                    &id,
                    &lost.claim_id,
                    "a",
                    lost.fencing_token,
                    "r1",
                    output.clone(),
                    vec![],
                    1,
                    15
                )
                .unwrap_err()
                .code,
            "WORKLOAD_RESULT_CONFLICT"
        );
        let committed = store.get(&id).unwrap();
        assert_eq!(committed.state, WorkloadState::Succeeded);
        assert_eq!(committed.result.as_ref().unwrap().output, output);
        assert_eq!(committed.result.as_ref().unwrap().produced_by, "b");
        let _ = fs::remove_file(path);
    }
    #[test]
    fn a_resumed_attempt_reads_the_input_and_the_lost_attempts_checkpoints_through_its_fence() {
        let (path, mut store) = store("resume-context");
        let w = store.create(create()).unwrap();
        let id = w.workload_id.clone();
        let caps = capabilities();
        let lost = store.claim(&id, "a", &caps, 11, 1000).unwrap();
        store
            .start(&id, &lost.claim_id, "a", lost.fencing_token, 11)
            .unwrap();
        store
            .checkpoint(
                &id,
                &lost.claim_id,
                "a",
                lost.fencing_token,
                serde_json::json!({"step": "reserve", "status": "done"}),
                11,
            )
            .unwrap();
        store.recover_expired(13).unwrap();
        let next = store.claim(&id, "b", &caps, 13, 10_000).unwrap();
        let context = store
            .execution_context(&id, &next.claim_id, "b", next.fencing_token, 13, vec![])
            .unwrap();
        assert_eq!(context.input, serde_json::json!({"id": 1}));
        assert_eq!(context.checkpoints.len(), 1);
        assert_eq!(context.checkpoints[0].state["step"], "reserve");
        assert_ne!(
            context.checkpoints[0].attempt_id, context.attempt_id,
            "written by the lost attempt"
        );
        // The lost worker's fence reads nothing.
        assert_eq!(
            store
                .execution_context(&id, &lost.claim_id, "a", lost.fencing_token, 13, vec![])
                .unwrap_err()
                .code,
            "WORKLOAD_STALE_FENCE"
        );
        let _ = fs::remove_file(path);
    }
    #[test]
    fn cancellation_race_and_duplicate_completion_are_deterministic() {
        let (path, mut store) = store("cancel");
        let w = store.create(create()).unwrap();
        let caps = ["email:send".into()].into();
        let claim = store.claim(&w.workload_id, "a", &caps, 11, 10_000).unwrap();
        store
            .start(
                &w.workload_id,
                &claim.claim_id,
                "a",
                claim.fencing_token,
                12,
            )
            .unwrap();
        assert_eq!(
            store.cancel(&w.workload_id, "user", 13).unwrap(),
            WorkloadState::CancellationRequested
        );
        assert_eq!(
            store
                .complete(
                    &w.workload_id,
                    &claim.claim_id,
                    "a",
                    claim.fencing_token,
                    "r",
                    Value::Null,
                    vec![],
                    1,
                    14
                )
                .unwrap_err()
                .code,
            "WORKLOAD_TRANSITION_DENIED"
        );
        store
            .confirm_cancel(
                &w.workload_id,
                &claim.claim_id,
                "a",
                claim.fencing_token,
                14,
            )
            .unwrap();
        assert_eq!(
            store.get(&w.workload_id).unwrap().state,
            WorkloadState::Cancelled
        );
        let _ = fs::remove_file(path);
    }
    #[test]
    fn scheduler_respects_dependencies_and_tenant_application_isolation() {
        let (path, mut store) = store("dependency");
        let parent = store.create(create()).unwrap();
        let mut child = create();
        child.idempotency_key = "child".into();
        child.dependencies = vec![WorkloadDependency {
            workload_id: parent.workload_id.clone(),
            required_state: WorkloadState::Succeeded,
        }];
        let child = store.create(child).unwrap();
        assert_eq!(store.list("other", "app").len(), 0);
        let caps = ["email:send".into()].into();
        let mut scheduler = LocalDurableScheduler::default();
        let first = scheduler
            .claim(&mut store, "t", "app", "worker", &caps, 11, 10_000)
            .unwrap()
            .unwrap();
        assert_eq!(first.workload_id, parent.workload_id);
        assert_ne!(first.workload_id, child.workload_id);
        let _ = fs::remove_file(path);
    }

    // ------------------------------------------------------------------
    // Temporal work eligibility.
    //
    // The invariant under test: a workload may exist durably before it is
    // executable. Every test below drives the real store — creating, claiming,
    // reloading from disk — rather than asserting that a field or a function
    // name exists.
    // ------------------------------------------------------------------

    /// A store on a path unique to this test, cleared first so a recycled
    /// process id cannot hand a test a previous run's file.
    fn temporal_store(name: &str) -> (PathBuf, WorkloadStore) {
        let path = std::env::temp_dir().join(format!(
            "workload-temporal-{name}-{}-{}.json",
            std::process::id(),
            rand::random::<u64>()
        ));
        let _ = fs::remove_file(&path);
        (path.clone(), WorkloadStore::load(&path).unwrap())
    }

    fn scheduled(not_before: i64) -> CreateWorkload {
        let mut input = create();
        input.not_before = Some(not_before);
        input
    }

    fn caps() -> BTreeSet<String> {
        ["email:send".into()].into()
    }

    #[test]
    fn scheduled_work_is_not_claimable_before_not_before() {
        let (path, mut store) = temporal_store("before");
        let workload = store.create(scheduled(2_000)).unwrap();

        // Durable and Ready immediately: existing before it is executable is
        // the whole point.
        assert_eq!(workload.state, WorkloadState::Ready);
        assert_eq!(workload.next_attempt_at, Some(2_000));

        let refused = store
            .claim(&workload.workload_id, "worker", &caps(), 1_999, 10_000)
            .unwrap_err();
        assert_eq!(refused.code, "WORKLOAD_NOT_YET_ELIGIBLE");
        assert!(
            refused.message.contains("2000"),
            "the refusal names the instant the worker should return at: {}",
            refused.message
        );

        // Refusing a claim is not a state change. Nothing executed.
        let after = store.get(&workload.workload_id).unwrap();
        assert_eq!(after.state, WorkloadState::Ready);
        assert_eq!(after.attempt_count, 0);
        assert!(after.attempts.is_empty());
        assert!(after.current_claim.is_none());
        let _ = fs::remove_file(path);
    }

    #[test]
    fn scheduled_work_becomes_claimable_after_not_before() {
        let (path, mut store) = temporal_store("after");
        let workload = store.create(scheduled(2_000)).unwrap();

        // The boundary belongs to the eligible side: at exactly `not_before`
        // the workload is claimable, so a worker polling on the second does not
        // have to wait a further tick.
        let claim = store
            .claim(&workload.workload_id, "worker", &caps(), 2_000, 10_000)
            .expect("eligible at the instant itself");
        assert_eq!(claim.worker_id, "worker");
        assert_eq!(
            store.get(&workload.workload_id).unwrap().state,
            WorkloadState::Claimed
        );
        let _ = fs::remove_file(path);
    }

    #[test]
    fn scheduled_work_survives_restart_before_eligibility() {
        let (path, mut store) = temporal_store("restart-before");
        let id = store.create(scheduled(2_000)).unwrap().workload_id;
        drop(store);

        // Restart the authority. The workload is still here, still Ready, still
        // carrying its eligibility, and still refused.
        let mut reopened = WorkloadStore::load(&path).unwrap();
        let recovered = reopened.get(&id).unwrap();
        assert_eq!(recovered.state, WorkloadState::Ready);
        assert_eq!(recovered.next_attempt_at, Some(2_000));

        assert_eq!(
            reopened
                .claim(&id, "worker", &caps(), 1_999, 10_000)
                .unwrap_err()
                .code,
            "WORKLOAD_NOT_YET_ELIGIBLE"
        );
        let _ = fs::remove_file(path);
    }

    #[test]
    fn scheduled_work_executes_after_restart() {
        let (path, mut store) = temporal_store("restart-after");
        let id = store.create(scheduled(2_000)).unwrap().workload_id;
        drop(store);

        let mut reopened = WorkloadStore::load(&path).unwrap();
        let claim = reopened
            .claim(&id, "worker", &caps(), 2_001, 10_000)
            .expect("claimable after restart once eligible");
        reopened
            .start(&id, &claim.claim_id, "worker", claim.fencing_token, 2_002)
            .unwrap();
        reopened
            .complete(
                &id,
                &claim.claim_id,
                "worker",
                claim.fencing_token,
                "result-scheduled",
                serde_json::json!({"ok": true}),
                vec![],
                1,
                2_003,
            )
            .unwrap();

        // And the outcome is durable across a second restart.
        let again = WorkloadStore::load(&path).unwrap();
        let finished = again.get(&id).unwrap();
        assert_eq!(finished.state, WorkloadState::Succeeded);
        assert_eq!(
            finished.result.as_ref().map(|r| r.result_id.as_str()),
            Some("result-scheduled")
        );
        let _ = fs::remove_file(path);
    }

    #[test]
    fn immediate_work_behavior_is_unchanged() {
        let (path, mut store) = temporal_store("immediate");
        // `create()` sets no `not_before`, which is what every caller did
        // before this field existed.
        let workload = store.create(create()).unwrap();
        assert_eq!(workload.next_attempt_at, None);
        assert!(workload.is_eligible_at(workload.created_at));

        // Claimable at its own creation instant, with no waiting.
        store
            .claim(
                &workload.workload_id,
                "worker",
                &caps(),
                workload.created_at,
                10_000,
            )
            .expect("work with no eligibility instant is claimable immediately");
        let _ = fs::remove_file(path);
    }

    #[test]
    fn multiple_workers_claim_temporally_eligible_work_once() {
        let (path, mut store) = temporal_store("contention");
        let id = store.create(scheduled(2_000)).unwrap().workload_id;

        // Both workers are polling and both see eligibility open at the same
        // instant. Exactly one wins.
        let winner = store
            .claim(&id, "worker-a", &caps(), 2_000, 10_000)
            .unwrap();
        let loser = store
            .claim(&id, "worker-b", &caps(), 2_000, 10_000)
            .unwrap_err();
        assert_eq!(
            loser.code, "WORKLOAD_NOT_READY",
            "the second worker is refused on state, not on time: the workload is claimed"
        );

        // And the loser cannot complete work it never owned.
        assert_eq!(
            store
                .complete(
                    &id,
                    &winner.claim_id,
                    "worker-b",
                    winner.fencing_token,
                    "duplicate",
                    Value::Null,
                    vec![],
                    1,
                    2_001,
                )
                .unwrap_err()
                .code,
            "WORKLOAD_STALE_FENCE"
        );
        assert_eq!(store.get(&id).unwrap().attempt_count, 1);
        let _ = fs::remove_file(path);
    }

    #[test]
    fn retry_sets_next_eligibility_without_duplicate_execution() {
        let (path, mut store) = temporal_store("retry");
        let id = store.create(scheduled(2_000)).unwrap().workload_id;

        let first = store
            .claim(&id, "worker-a", &caps(), 2_000, 10_000)
            .unwrap();
        store
            .start(&id, &first.claim_id, "worker-a", first.fencing_token, 2_001)
            .unwrap();
        let state = store
            .fail(
                &id,
                &first.claim_id,
                "worker-a",
                first.fencing_token,
                "timeout",
                "upstream did not answer",
                2_002,
            )
            .unwrap();

        // Retry re-arms the same eligibility field creation used, so there is
        // one temporal concept rather than two competing ones.
        assert_eq!(state, WorkloadState::Ready);
        let backing_off = store.get(&id).unwrap();
        let next = backing_off.next_attempt_at.expect("retry set eligibility");
        assert!(
            next > 2_002,
            "backoff pushed eligibility into the future: {next}"
        );

        assert_eq!(
            store
                .claim(&id, "worker-b", &caps(), next - 1, 10_000)
                .unwrap_err()
                .code,
            "WORKLOAD_NOT_YET_ELIGIBLE"
        );

        let second = store.claim(&id, "worker-b", &caps(), next, 10_000).unwrap();
        assert!(second.fencing_token > first.fencing_token);

        // The first attempt cannot land a result after the retry was handed on.
        assert_eq!(
            store
                .complete(
                    &id,
                    &first.claim_id,
                    "worker-a",
                    first.fencing_token,
                    "stale",
                    Value::Null,
                    vec![],
                    1,
                    next,
                )
                .unwrap_err()
                .code,
            "WORKLOAD_STALE_FENCE"
        );
        assert_eq!(store.get(&id).unwrap().attempt_count, 2);
        let _ = fs::remove_file(path);
    }

    #[test]
    fn cancellation_before_not_before_prevents_execution() {
        let (path, mut store) = temporal_store("cancel");
        let id = store.create(scheduled(2_000)).unwrap().workload_id;

        // Cancelling work that is durable but not yet executable is the whole
        // reason to be able to see it before it runs.
        assert_eq!(
            store.cancel(&id, "operator", 1_500).unwrap(),
            WorkloadState::Cancelled
        );

        // Eligibility arriving does not resurrect it.
        assert_eq!(
            store
                .claim(&id, "worker", &caps(), 2_001, 10_000)
                .unwrap_err()
                .code,
            "WORKLOAD_NOT_READY"
        );

        let reopened = WorkloadStore::load(&path).unwrap();
        let cancelled = reopened.get(&id).unwrap();
        assert_eq!(cancelled.state, WorkloadState::Cancelled);
        assert!(cancelled.attempts.is_empty(), "nothing ever executed");
        let _ = fs::remove_file(path);
    }

    #[test]
    fn a_worker_dying_after_a_claim_returns_the_work_and_it_executes_once() {
        let (path, mut store) = temporal_store("worker-death");
        let id = store.create(create()).unwrap().workload_id;

        // Worker A claims a short lease, starts, and dies without completing.
        let dead = store.claim(&id, "worker-a", &caps(), 10, 1_000).unwrap();
        store
            .start(&id, &dead.claim_id, "worker-a", dead.fencing_token, 10)
            .unwrap();
        assert_eq!(
            store.get(&id).unwrap().state,
            WorkloadState::Running,
            "the work was in flight when its worker died"
        );

        // The lease expiring is the only thing that can notice.
        let recovered = store.recover_expired(12).unwrap();
        assert_eq!(recovered, vec![id.clone()]);
        let back = store.get(&id).unwrap();
        assert_eq!(back.state, WorkloadState::Ready, "returned to the pool");
        assert_eq!(
            back.attempts.last().map(|attempt| attempt.status),
            Some(AttemptStatus::WorkerLost),
            "and the dead attempt is recorded as lost rather than erased"
        );

        // Worker B picks it up and finishes it.
        let live = store.claim(&id, "worker-b", &caps(), 12, 10_000).unwrap();
        assert!(live.fencing_token > dead.fencing_token);
        store
            .start(&id, &live.claim_id, "worker-b", live.fencing_token, 13)
            .unwrap();
        store
            .complete(
                &id,
                &live.claim_id,
                "worker-b",
                live.fencing_token,
                "result-recovered",
                serde_json::json!({"ok": true}),
                vec![],
                1,
                14,
            )
            .unwrap();

        // Exactly one execution succeeded, and the dead worker cannot add another.
        let finished = store.get(&id).unwrap();
        assert_eq!(finished.state, WorkloadState::Succeeded);
        assert_eq!(
            finished
                .attempts
                .iter()
                .filter(|attempt| attempt.status == AttemptStatus::Succeeded)
                .count(),
            1
        );
        // The dead worker cannot add a second result. Its claim was released by
        // the recovery, and completion released the successor's, so it is
        // refused for having no claim at all rather than a stale one.
        assert_eq!(
            store
                .complete(
                    &id,
                    &dead.claim_id,
                    "worker-a",
                    dead.fencing_token,
                    "resurrected",
                    Value::Null,
                    vec![],
                    1,
                    15,
                )
                .unwrap_err()
                .code,
            "WORKLOAD_CLAIM_INVALID"
        );
        let _ = fs::remove_file(path);
    }

    #[test]
    fn a_recovered_workload_refuses_its_dead_worker_while_the_successor_still_holds_it() {
        let (path, mut store) = temporal_store("worker-death-inflight");
        let id = store.create(create()).unwrap().workload_id;

        let dead = store.claim(&id, "worker-a", &caps(), 10, 1_000).unwrap();
        store
            .start(&id, &dead.claim_id, "worker-a", dead.fencing_token, 10)
            .unwrap();
        store.recover_expired(12).unwrap();
        let live = store.claim(&id, "worker-b", &caps(), 12, 10_000).unwrap();

        // While the successor is mid-flight, the resurrected worker is refused
        // on the fence itself — the case a lease exists to decide.
        assert_eq!(
            store
                .complete(
                    &id,
                    &dead.claim_id,
                    "worker-a",
                    dead.fencing_token,
                    "resurrected",
                    Value::Null,
                    vec![],
                    1,
                    13,
                )
                .unwrap_err()
                .code,
            "WORKLOAD_STALE_FENCE"
        );
        assert_eq!(store.get(&id).unwrap().state, WorkloadState::Claimed);
        assert!(live.fencing_token > dead.fencing_token);
        let _ = fs::remove_file(path);
    }

    #[test]
    fn the_scheduler_skips_ineligible_work_rather_than_failing_on_it() {
        let (path, mut store) = temporal_store("scheduler");
        let mut early = create();
        early.idempotency_key = "early".into();
        early.not_before = Some(9_000);
        let early = store.create(early).unwrap();

        let mut ready = create();
        ready.idempotency_key = "ready".into();
        let ready = store.create(ready).unwrap();

        // The ineligible workload sorts first by id in some orderings and not
        // others, so what matters is that the scheduler never returns it and
        // never fails because of it.
        let mut scheduler = LocalDurableScheduler::default();
        let claim = scheduler
            .claim(&mut store, "t", "app", "worker", &caps(), 1_000, 10_000)
            .expect("an ineligible workload must not fail the whole pass")
            .expect("the eligible workload is still handed out");
        assert_eq!(claim.workload_id, ready.workload_id);
        assert_eq!(
            store.get(&early.workload_id).unwrap().state,
            WorkloadState::Ready
        );
        let _ = fs::remove_file(path);
    }

    // ---- PR 2: authoritative reads, canonical admission, failure reports, deadlines ----

    fn version_of(store: &WorkloadStore, id: &str) -> u64 {
        let db = store.db_for_tests();
        let w = store.get(id).unwrap();
        db.live_row(DURABLE_WORK_COLLECTION, &work_key(w))
            .and_then(|row| row.value.get("__version").and_then(Value::as_u64))
            .unwrap()
    }

    #[test]
    fn another_writers_commits_are_visible_and_acted_on_without_restart() {
        let (path, mut a) = store("two-writers");
        let mut b = WorkloadStore::from_db(a.db_for_tests()).unwrap();
        let id = a.create(create()).unwrap().workload_id;
        // B never saw the admission, yet reads it from the store.
        assert!(b.get(&id).is_err(), "the projection alone is stale");
        assert_eq!(b.fresh(&id).unwrap().state, WorkloadState::Ready);
        assert_eq!(b.fresh_list("t", "app").unwrap().len(), 1);
        // A claims; B's claim decides on the committed claim, not its own view.
        let claim = a
            .claim(&id, "worker-a", &capabilities(), 11, 10_000)
            .unwrap();
        assert_eq!(
            b.claim(&id, "worker-b", &capabilities(), 11, 10_000)
                .unwrap_err()
                .code,
            "WORKLOAD_NOT_READY"
        );
        // A transition committed through A is what B fences against.
        b.start(&id, &claim.claim_id, "worker-a", claim.fencing_token, 12)
            .unwrap();
        assert_eq!(a.fresh(&id).unwrap().state, WorkloadState::Running);
        // B replays an admission A committed, by key, from the store.
        let mut c = WorkloadStore::from_db(a.db_for_tests()).unwrap();
        c.workloads.clear();
        c.idempotency.clear();
        c.document_versions.clear();
        let (replayed, created) = c.admit(create()).unwrap();
        assert!(!created);
        assert_eq!(replayed.workload_id, id);
        let _ = fs::remove_file(path);
    }

    #[test]
    fn a_replay_matches_only_an_equivalent_request_and_names_every_conflicting_field() {
        let (path, mut store) = store("canonical-admission");
        let mut original = create();
        original.input = serde_json::json!({"a": 1, "b": 2});
        original.dependencies = vec![
            WorkloadDependency {
                workload_id: "w1".into(),
                required_state: WorkloadState::Succeeded,
            },
            WorkloadDependency {
                workload_id: "w2".into(),
                required_state: WorkloadState::Succeeded,
            },
        ];
        let first = store.create(original.clone()).unwrap();

        // Equivalent after normalization: member order, dependency order, the
        // defaults written out, and provenance that does not change what runs.
        let mut equivalent = original.clone();
        equivalent.input = serde_json::json!({"b": 2, "a": 1});
        equivalent.dependencies.reverse();
        equivalent.retry_policy = WorkloadRetryPolicy::default();
        equivalent.timeouts = WorkloadTimeouts::default();
        equivalent.priority = 0;
        equivalent.trigger = "retried-by-client".into();
        equivalent.session_id = Some("another-session".into());
        equivalent.created_at = 99;
        let (replayed, created) = store.admit(equivalent).unwrap();
        assert!(!created);
        assert_eq!(replayed.workload_id, first.workload_id);

        let mutations: Vec<(&str, Box<dyn Fn(&mut CreateWorkload)>)> = vec![
            (
                "definition_id",
                Box::new(|r| r.definition_id = "other".into()),
            ),
            (
                "definition_revision",
                Box::new(|r| r.definition_revision = "r2".into()),
            ),
            (
                "environment",
                Box::new(|r| r.environment = "staging".into()),
            ),
            (
                "input",
                Box::new(|r| r.input = serde_json::json!({"a": 1, "b": 3})),
            ),
            (
                "capability_snapshot",
                Box::new(|r| {
                    r.capability_snapshot.insert("sms:send".into());
                }),
            ),
            (
                "connection_scopes",
                Box::new(|r| {
                    r.connection_scopes.insert("crm".into());
                }),
            ),
            ("priority", Box::new(|r| r.priority = 5)),
            (
                "retry_policy",
                Box::new(|r| r.retry_policy.max_attempts = 9),
            ),
            (
                "retry_policy",
                Box::new(|r| {
                    r.retry_policy.retry_on.remove("worker_failure");
                }),
            ),
            (
                "timeouts",
                Box::new(|r| r.timeouts.execution_ms = Some(60_000)),
            ),
            (
                "dependencies",
                Box::new(|r| r.dependencies.pop().map(|_| ()).unwrap()),
            ),
            ("not_before", Box::new(|r| r.not_before = Some(5_000))),
        ];
        for (field, mutate) in mutations {
            let mut request = original.clone();
            mutate(&mut request);
            let error = store.admit(request).unwrap_err();
            assert_eq!(error.code, "IDEMPOTENCY_KEY_REUSED", "{field}");
            assert_eq!(
                error.details.unwrap()["fields"],
                serde_json::json!([field]),
                "{field}"
            );
        }
        // Several at once are all named.
        let mut request = original.clone();
        request.priority = 1;
        request.environment = "staging".into();
        let fields = store.admit(request).unwrap_err().details.unwrap()["fields"].clone();
        assert_eq!(fields, serde_json::json!(["environment", "priority"]));

        // After a restart: the same answers from the committed record.
        drop(store);
        let mut reopened = WorkloadStore::load(&path).unwrap();
        assert!(!reopened.admit(original.clone()).unwrap().1);
        let mut changed = original.clone();
        changed.retry_policy.initial_delay_ms = 5;
        assert_eq!(
            reopened.admit(changed).unwrap_err().code,
            "IDEMPOTENCY_KEY_REUSED"
        );
        assert_eq!(reopened.workloads.len(), 1, "nothing was created");
        let _ = fs::remove_file(path);
    }

    #[test]
    fn concurrent_conflicting_admissions_admit_one_and_refuse_the_other() {
        for round in 0..10 {
            let path = std::env::temp_dir().join(format!(
                "workload-admission-race-{round}-{}.log",
                std::process::id()
            ));
            let _ = fs::remove_file(&path);
            let db = FeltDb::open(&path).unwrap();
            let stores = [
                WorkloadStore::from_db(db.clone()).unwrap(),
                WorkloadStore::from_db(db.clone()).unwrap(),
            ];
            let barrier = std::sync::Barrier::new(2);
            let outcomes = std::thread::scope(|scope| {
                stores
                    .into_iter()
                    .enumerate()
                    .map(|(i, mut store)| {
                        let barrier = &barrier;
                        scope.spawn(move || {
                            let mut request = create();
                            request.priority = i as i32;
                            barrier.wait();
                            store
                                .admit(request)
                                .map(|(w, created)| (w.priority, created))
                        })
                    })
                    .collect::<Vec<_>>()
                    .into_iter()
                    .map(|handle| handle.join().unwrap())
                    .collect::<Vec<_>>()
            });
            let admitted = outcomes
                .iter()
                .filter(|o| matches!(o, Ok((_, true))))
                .count();
            let refused = outcomes
                .iter()
                .filter(|o| {
                    o.as_ref()
                        .is_err_and(|e| e.code == "IDEMPOTENCY_KEY_REUSED")
                })
                .count();
            assert_eq!((admitted, refused), (1, 1), "round {round}: {outcomes:?}");
            let _ = fs::remove_file(path);
        }
    }

    fn claimed(
        name: &str,
        policy: impl FnOnce(&mut CreateWorkload),
    ) -> (PathBuf, WorkloadStore, String, WorkloadClaim) {
        let (path, mut store) = store(name);
        let mut request = create();
        policy(&mut request);
        let id = store.create(request).unwrap().workload_id;
        let claim = store.claim(&id, "a", &capabilities(), 11, 10_000).unwrap();
        (path, store, id, claim)
    }

    #[test]
    fn a_worker_reports_failure_from_claimed_fenced_and_by_policy() {
        // Could not start: a known failure before any effect, retried by kind.
        let (path, mut store, id, claim) = claimed("fail-claimed", |_| {});
        let state = store
            .fail_with_effect(
                &id,
                &claim.claim_id,
                "a",
                claim.fencing_token,
                "transient_connection",
                "no route",
                None,
                12,
            )
            .unwrap();
        assert_eq!(state, WorkloadState::Ready);
        let w = store.get(&id).unwrap();
        let failed = w
            .execution_history
            .iter()
            .find(|e| e.to == WorkloadState::Failed)
            .unwrap();
        assert_eq!(
            failed.evidence["effect"], "none",
            "never started, so no effect"
        );
        assert_eq!(w.attempts[0].status, AttemptStatus::Failed);
        // The identical report again (lost response) replays and commits nothing.
        let version = version_of(&store, &id);
        assert_eq!(
            store
                .fail_with_effect(
                    &id,
                    &claim.claim_id,
                    "a",
                    claim.fencing_token,
                    "transient_connection",
                    "no route",
                    None,
                    13
                )
                .unwrap(),
            WorkloadState::Ready
        );
        assert_eq!(version_of(&store, &id), version);
        // A different report from the same attempt is answered by its fence.
        assert_eq!(
            store
                .fail_with_effect(
                    &id,
                    &claim.claim_id,
                    "a",
                    claim.fencing_token,
                    "timeout",
                    "other",
                    None,
                    13
                )
                .unwrap_err()
                .code,
            "WORKLOAD_CLAIM_INVALID"
        );
        // So is a completion: the failure stands.
        assert_eq!(
            store
                .complete(
                    &id,
                    &claim.claim_id,
                    "a",
                    claim.fencing_token,
                    "r",
                    Value::Null,
                    vec![],
                    1,
                    13
                )
                .unwrap_err()
                .code,
            "WORKLOAD_CLAIM_INVALID"
        );
        // Another worker's claim fences the old attempt out.
        let next = store.claim(&id, "b", &capabilities(), 14, 10_000).unwrap();
        assert_eq!(
            store
                .fail_with_effect(
                    &id,
                    &claim.claim_id,
                    "a",
                    claim.fencing_token,
                    "timeout",
                    "late",
                    None,
                    15
                )
                .unwrap_err()
                .code,
            "WORKLOAD_STALE_FENCE"
        );
        assert_eq!(
            store
                .get(&id)
                .unwrap()
                .current_claim
                .as_ref()
                .unwrap()
                .claim_id,
            next.claim_id
        );
        let _ = fs::remove_file(path);
    }

    #[test]
    fn failure_policy_decides_retry_and_an_unknown_effect_is_decided_like_a_lost_worker() {
        // A kind the policy does not retry dead-letters.
        let (path, mut store, id, claim) = claimed("fail-kind", |_| {});
        store
            .start(&id, &claim.claim_id, "a", claim.fencing_token, 12)
            .unwrap();
        assert_eq!(
            store
                .fail_with_effect(
                    &id,
                    &claim.claim_id,
                    "a",
                    claim.fencing_token,
                    "validation",
                    "bad input",
                    Some("none"),
                    13
                )
                .unwrap(),
            WorkloadState::DeadLetter
        );
        assert_eq!(
            store
                .get(&id)
                .unwrap()
                .execution_history
                .last()
                .unwrap()
                .evidence["decision"],
            "failure_kind_not_retryable"
        );
        let _ = fs::remove_file(path);

        // A retryable kind with an unknown effect, under a policy that does not
        // re-run lost work: not retried automatically.
        let (path, mut store, id, claim) = claimed("fail-unknown", |r| {
            r.retry_policy.retry_on = ["timeout".into(), "transient_connection".into()].into();
        });
        store
            .start(&id, &claim.claim_id, "a", claim.fencing_token, 12)
            .unwrap();
        assert_eq!(
            store
                .fail_with_effect(
                    &id,
                    &claim.claim_id,
                    "a",
                    claim.fencing_token,
                    "timeout",
                    "charge sent, no answer",
                    Some("unknown"),
                    13
                )
                .unwrap(),
            WorkloadState::DeadLetter
        );
        let w = store.get(&id).unwrap();
        assert_eq!(
            w.execution_history.last().unwrap().evidence["decision"],
            "effect_unknown_not_retryable"
        );
        // An operator can still re-arm it, with a reason.
        store
            .retry_with_reason(&id, "operator", 14, Some("verified: not charged"))
            .unwrap();
        assert_eq!(store.get(&id).unwrap().state, WorkloadState::Ready);
        let _ = fs::remove_file(path);

        // The same report under a policy that re-runs lost work is retried.
        let (path, mut store, id, claim) = claimed("fail-unknown-retry", |_| {});
        store
            .start(&id, &claim.claim_id, "a", claim.fencing_token, 12)
            .unwrap();
        assert_eq!(
            store
                .fail_with_effect(
                    &id,
                    &claim.claim_id,
                    "a",
                    claim.fencing_token,
                    "validation",
                    "x",
                    Some("unknown"),
                    13
                )
                .unwrap(),
            WorkloadState::Ready
        );
        // An invalid effect is refused before anything changes.
        assert_eq!(
            store
                .fail_with_effect(&id, "c", "a", 9, "x", "y", Some("maybe"), 14)
                .unwrap_err()
                .code,
            "WORKLOAD_INVALID"
        );
        let _ = fs::remove_file(path);

        // The attempt budget bounds retries.
        let (path, mut store, id, mut claim) =
            claimed("fail-budget", |r| r.retry_policy.max_attempts = 2);
        assert_eq!(
            store
                .fail(
                    &id,
                    &claim.claim_id,
                    "a",
                    claim.fencing_token,
                    "timeout",
                    "x",
                    12
                )
                .unwrap(),
            WorkloadState::Ready
        );
        let next = store.get(&id).unwrap().next_attempt_at.unwrap();
        claim = store
            .claim(&id, "a", &capabilities(), next, 10_000)
            .unwrap();
        assert_eq!(
            store
                .fail(
                    &id,
                    &claim.claim_id,
                    "a",
                    claim.fencing_token,
                    "timeout",
                    "x",
                    next
                )
                .unwrap(),
            WorkloadState::DeadLetter
        );
        assert_eq!(
            store
                .get(&id)
                .unwrap()
                .execution_history
                .last()
                .unwrap()
                .evidence["decision"],
            "attempt_budget_exhausted"
        );
        let _ = fs::remove_file(path);
    }

    #[test]
    fn failure_racing_completion_commits_exactly_one_outcome() {
        for completion_first in [true, false] {
            let (path, mut store, id, claim) = claimed(&format!("race-{completion_first}"), |_| {});
            store
                .start(&id, &claim.claim_id, "a", claim.fencing_token, 12)
                .unwrap();
            let complete = |s: &mut WorkloadStore| {
                s.complete(
                    &id,
                    &claim.claim_id,
                    "a",
                    claim.fencing_token,
                    "r",
                    Value::Null,
                    vec![],
                    1,
                    13,
                )
                .map(|_| ())
            };
            let fail = |s: &mut WorkloadStore| {
                s.fail(
                    &id,
                    &claim.claim_id,
                    "a",
                    claim.fencing_token,
                    "timeout",
                    "x",
                    13,
                )
                .map(|_| ())
            };
            let (first, second) = if completion_first {
                (complete(&mut store), fail(&mut store))
            } else {
                (fail(&mut store), complete(&mut store))
            };
            assert!(first.is_ok());
            assert_eq!(second.unwrap_err().code, "WORKLOAD_CLAIM_INVALID");
            let w = store.get(&id).unwrap();
            assert_eq!(
                w.state,
                if completion_first {
                    WorkloadState::Succeeded
                } else {
                    WorkloadState::Ready
                }
            );
            assert_eq!(w.attempts.len(), 1);
            let _ = fs::remove_file(path);
        }
    }

    #[test]
    fn deadlines_are_enforced_from_committed_timestamps_and_race_safely() {
        // execution: a heartbeat inside the deadline keeps the attempt; after it,
        // every worker call is refused and the attempt is timed out, once.
        let (path, mut store, id, claim) =
            claimed("deadline-exec", |r| r.timeouts.execution_ms = Some(5_000));
        store
            .start(&id, &claim.claim_id, "a", claim.fencing_token, 12)
            .unwrap();
        store
            .heartbeat(&id, &claim.claim_id, "a", claim.fencing_token, 16, 10_000)
            .unwrap();
        assert!(store.enforce_deadlines(16).unwrap().is_empty());
        let refused = store
            .complete(
                &id,
                &claim.claim_id,
                "a",
                claim.fencing_token,
                "late",
                Value::Null,
                vec![],
                1,
                17,
            )
            .unwrap_err();
        assert_eq!(refused.code, "WORKLOAD_TIMED_OUT");
        assert_eq!(refused.details.unwrap()["deadline"], "execution");
        let w = store.get(&id).unwrap().clone();
        assert_eq!(
            w.state,
            WorkloadState::Ready,
            "timeout is retryable by default"
        );
        assert_eq!(w.attempts[0].status, AttemptStatus::TimedOut);
        assert!(w.result.is_none(), "a late completion never lands");
        assert!(
            store.enforce_deadlines(30).unwrap().is_empty(),
            "timed out once"
        );
        let _ = fs::remove_file(path);

        // heartbeat: the gap since the last heartbeat; a heartbeat that commits
        // first moves the deadline, the recovery pass after it changes nothing.
        let (path, mut store, id, claim) = claimed("deadline-heartbeat", |r| {
            r.timeouts.heartbeat_ms = Some(3_000)
        });
        store
            .start(&id, &claim.claim_id, "a", claim.fencing_token, 12)
            .unwrap();
        store
            .heartbeat(&id, &claim.claim_id, "a", claim.fencing_token, 13, 10_000)
            .unwrap();
        assert!(store.enforce_deadlines(15).unwrap().is_empty());
        assert_eq!(store.enforce_deadlines(16).unwrap(), vec![id.clone()]);
        assert_eq!(
            store
                .heartbeat(&id, &claim.claim_id, "a", claim.fencing_token, 16, 10_000)
                .unwrap_err()
                .code,
            "WORKLOAD_CLAIM_INVALID",
            "the timed-out claim is gone"
        );
        let _ = fs::remove_file(path);

        // claim: claimed but never started; not retried when the policy says so.
        let (path, mut store, id, claim) = claimed("deadline-claim", |r| {
            r.timeouts.claim_ms = Some(2_000);
            r.retry_policy.retry_on = ["worker_failure".into()].into();
        });
        let refused = store
            .start(&id, &claim.claim_id, "a", claim.fencing_token, 13)
            .unwrap_err();
        assert_eq!(refused.code, "WORKLOAD_TIMED_OUT");
        let w = store.get(&id).unwrap();
        assert_eq!(w.state, WorkloadState::DeadLetter);
        assert_eq!(
            w.execution_history.last().unwrap().evidence["decision"],
            "timeout_not_retryable"
        );
        let _ = fs::remove_file(path);

        // queue and overall bound the work: dead-lettered, never retried, and a
        // queued workload's previous attempt is not rewritten.
        let (path, mut store) = super::tests::store("deadline-queue");
        let mut request = create();
        request.timeouts.queue_ms = Some(4_000);
        let id = store.create(request).unwrap().workload_id;
        assert_eq!(
            store
                .claim(&id, "a", &capabilities(), 14, 10_000)
                .unwrap_err()
                .code,
            "WORKLOAD_TIMED_OUT"
        );
        let w = store.get(&id).unwrap();
        assert_eq!(w.state, WorkloadState::DeadLetter);
        assert_eq!(
            w.execution_history.last().unwrap().evidence["decision"],
            "deadline_not_retryable"
        );
        assert!(w.attempts.is_empty());
        let _ = fs::remove_file(path);

        let (path, mut store, id, claim) =
            claimed("deadline-overall", |r| r.timeouts.overall_ms = Some(10_000));
        store
            .start(&id, &claim.claim_id, "a", claim.fencing_token, 12)
            .unwrap();
        // Restart: the deadline comes from the committed record.
        drop(store);
        let mut store = WorkloadStore::load(&path).unwrap();
        assert!(store.enforce_deadlines(19).unwrap().is_empty());
        assert_eq!(store.enforce_deadlines(20).unwrap(), vec![id.clone()]);
        assert_eq!(store.get(&id).unwrap().state, WorkloadState::DeadLetter);
        let _ = fs::remove_file(path);
    }

    #[test]
    fn deadlines_are_versioned_and_zero_is_refused() {
        let (path, mut store) = store("deadline-version");
        let mut request = create();
        request.timeouts.queue_ms = Some(0);
        assert_eq!(store.create(request).unwrap_err().code, "WORKLOAD_INVALID");
        // A record admitted before deadlines were enforced keeps its contract.
        let mut request = create();
        request.timeouts.queue_ms = Some(1_000);
        let id = store.create(request).unwrap().workload_id;
        assert_eq!(
            store.get(&id).unwrap().contract_version,
            DEADLINE_CONTRACT_VERSION
        );
        let mut legacy = store.get(&id).unwrap().clone();
        legacy.contract_version = 1;
        assert_eq!(deadline_breach(&legacy, 1_000_000), None);
        assert!(deadline_breach(store.get(&id).unwrap(), 1_000_000).is_some());
        let _ = fs::remove_file(path);
    }

    // ---- PR 3: the approval gate ----

    fn gated(name: &str, policy: ApprovalPolicy) -> (PathBuf, WorkloadStore, String, String) {
        let (path, mut store) = store(name);
        let mut request = create();
        request.approval = Some(policy);
        let w = store.create(request).unwrap();
        (path, store, w.workload_id, w.input_hash)
    }
    fn human(id: &str) -> Subject {
        Subject::Human(id.into())
    }

    #[test]
    fn work_requiring_approval_is_never_claimable_before_an_authorized_decision_commits() {
        let (path, mut store, id, input_hash) = gated("approval-gate", ApprovalPolicy::default());
        assert_eq!(
            store.get(&id).unwrap().state,
            WorkloadState::AwaitingApproval
        );
        assert_eq!(
            store
                .claim(&id, "w", &capabilities(), 11, 10_000)
                .unwrap_err()
                .code,
            "WORKLOAD_APPROVAL_PENDING"
        );
        // Who may not decide: an agent (model output), a worker, a service
        // under a human-only policy, and the requester itself.
        for decider in [
            Subject::Agent("model".into()),
            Subject::Worker("w".into()),
            Subject::Service("svc".into()),
            Subject::Agent("a1".into()),
        ] {
            assert_eq!(
                store
                    .decide(
                        &id,
                        &decider,
                        ApprovalOutcome::Approved,
                        None,
                        "g",
                        &input_hash,
                        11
                    )
                    .unwrap_err()
                    .code,
                "WORKLOAD_APPROVER_NOT_PERMITTED",
                "{decider:?}"
            );
        }
        // A decision about other input (a stale review, or another workload's) is refused.
        let stale = store
            .decide(
                &id,
                &human("ada"),
                ApprovalOutcome::Approved,
                None,
                "g",
                "sha256:other",
                11,
            )
            .unwrap_err();
        assert_eq!(stale.code, "WORKLOAD_APPROVAL_STALE");
        assert_eq!(stale.details.unwrap()["input_hash"], input_hash.as_str());
        assert_eq!(
            store.get(&id).unwrap().state,
            WorkloadState::AwaitingApproval,
            "refusals change nothing"
        );

        let (decision, replayed) = store
            .decide(
                &id,
                &human("ada"),
                ApprovalOutcome::Approved,
                Some("reviewed"),
                "g",
                &input_hash,
                12,
            )
            .unwrap();
        assert!(!replayed);
        assert_eq!(decision.decided_by, human("ada"));
        assert_eq!(store.get(&id).unwrap().state, WorkloadState::Ready);
        // The same decision again: a replay that commits nothing.
        let version = version_of(&store, &id);
        let (again, replayed) = store
            .decide(
                &id,
                &human("bob"),
                ApprovalOutcome::Approved,
                None,
                "g2",
                &input_hash,
                13,
            )
            .unwrap();
        assert!(replayed);
        assert_eq!(
            again, decision,
            "the committed decision is what is returned"
        );
        assert_eq!(version_of(&store, &id), version);
        // The opposite decision cannot overwrite it.
        let conflict = store
            .decide(
                &id,
                &human("bob"),
                ApprovalOutcome::Rejected,
                Some("no"),
                "g2",
                &input_hash,
                13,
            )
            .unwrap_err();
        assert_eq!(conflict.code, "WORKLOAD_APPROVAL_DECIDED");
        assert_eq!(conflict.details.unwrap()["outcome"], "approved");
        // Now it runs.
        store.claim(&id, "w", &capabilities(), 14, 10_000).unwrap();
        // After a restart the decision and the state are what was committed.
        drop(store);
        let reopened = WorkloadStore::load(&path).unwrap();
        let w = reopened.get(&id).unwrap();
        assert_eq!(w.state, WorkloadState::Claimed);
        assert_eq!(
            w.approval.as_ref().unwrap().decision.as_ref().unwrap(),
            &decision
        );
        let order: Vec<_> = w.execution_history.iter().map(|e| e.to).collect();
        assert_eq!(
            order,
            vec![
                WorkloadState::AwaitingApproval,
                WorkloadState::Ready,
                WorkloadState::Claimed
            ]
        );
        let _ = fs::remove_file(path);
    }

    #[test]
    fn rejected_work_never_runs_and_a_rejection_is_final() {
        let (path, mut store, id, input_hash) = gated("approval-reject", ApprovalPolicy::default());
        assert_eq!(
            store
                .decide(
                    &id,
                    &human("ada"),
                    ApprovalOutcome::Rejected,
                    None,
                    "g",
                    &input_hash,
                    11
                )
                .unwrap_err()
                .code,
            "WORKLOAD_INVALID",
            "a rejection needs a reason"
        );
        store
            .decide(
                &id,
                &human("ada"),
                ApprovalOutcome::Rejected,
                Some("wrong vendor"),
                "g",
                &input_hash,
                11,
            )
            .unwrap();
        let w = store.get(&id).unwrap();
        assert_eq!(w.state, WorkloadState::Rejected);
        assert_eq!(
            w.execution_history.last().unwrap().evidence["reason"],
            "wrong vendor"
        );
        assert_eq!(
            store
                .claim(&id, "w", &capabilities(), 12, 10_000)
                .unwrap_err()
                .code,
            "WORKLOAD_REJECTED"
        );
        assert_eq!(
            store
                .retry_with_reason(&id, "op", 12, Some("again"))
                .unwrap_err()
                .code,
            "WORKLOAD_TRANSITION_DENIED"
        );
        assert_eq!(
            store.cancel(&id, "op", 12).unwrap_err().code,
            "WORKLOAD_TRANSITION_DENIED"
        );
        assert_eq!(
            store
                .decide(
                    &id,
                    &human("bob"),
                    ApprovalOutcome::Approved,
                    None,
                    "g",
                    &input_hash,
                    12
                )
                .unwrap_err()
                .code,
            "WORKLOAD_APPROVAL_DECIDED"
        );
        assert!(
            store.get(&id).unwrap().attempts.is_empty(),
            "nothing ever ran"
        );
        let _ = fs::remove_file(path);
    }

    #[test]
    fn a_policy_narrows_approvers_and_excludes_the_requester() {
        let policy = ApprovalPolicy {
            approver_kinds: ["service".into(), "human".into()].into(),
            approvers: [Subject::Service("finance-bot".into()), human("ada")].into(),
        };
        let (path, mut store, id, input_hash) = gated("approval-policy", policy);
        for (decider, code) in [
            (human("eve"), "WORKLOAD_APPROVER_NOT_PERMITTED"),
            (
                Subject::Service("other-bot".into()),
                "WORKLOAD_APPROVER_NOT_PERMITTED",
            ),
        ] {
            assert_eq!(
                store
                    .decide(
                        &id,
                        &decider,
                        ApprovalOutcome::Approved,
                        None,
                        "g",
                        &input_hash,
                        11
                    )
                    .unwrap_err()
                    .code,
                code
            );
        }
        store
            .decide(
                &id,
                &Subject::Service("finance-bot".into()),
                ApprovalOutcome::Approved,
                None,
                "g",
                &input_hash,
                11,
            )
            .unwrap();
        let _ = fs::remove_file(path);

        // The requester never decides its own work, even when its kind may.
        let (path, mut store) = super::tests::store("approval-requester");
        let mut request = create_as("svc");
        request.requested_by = Subject::Service("svc".into());
        request.authorization_snapshot.grant.subject = request.requested_by.clone();
        // A re-signed snapshot is not needed for the store's own checks.
        request.approval = Some(ApprovalPolicy {
            approver_kinds: ["service".into()].into(),
            approvers: BTreeSet::new(),
        });
        let w = store.create(request).unwrap();
        assert_eq!(
            store
                .decide(
                    &w.workload_id,
                    &Subject::Service("svc".into()),
                    ApprovalOutcome::Approved,
                    None,
                    "g",
                    &w.input_hash,
                    11
                )
                .unwrap_err()
                .code,
            "WORKLOAD_APPROVER_NOT_PERMITTED"
        );
        let _ = fs::remove_file(path);
    }

    #[test]
    fn an_approval_policy_is_validated_compared_on_replay_and_required_for_a_decision() {
        let (path, mut store) = store("approval-validate");
        for policy in [
            ApprovalPolicy {
                approver_kinds: ["agent".into()].into(),
                approvers: BTreeSet::new(),
            },
            ApprovalPolicy {
                approver_kinds: BTreeSet::new(),
                approvers: BTreeSet::new(),
            },
            ApprovalPolicy {
                approver_kinds: ["human".into()].into(),
                approvers: [Subject::Service("s".into())].into(),
            },
        ] {
            let mut request = create();
            request.approval = Some(policy);
            assert_eq!(store.create(request).unwrap_err().code, "WORKLOAD_INVALID");
        }
        let mut request = create();
        request.timeouts.approval_ms = Some(1_000);
        assert_eq!(
            store.create(request).unwrap_err().code,
            "WORKLOAD_INVALID",
            "approval_ms needs a policy"
        );
        // Work without a policy behaves exactly as before.
        let plain = store.create(create()).unwrap();
        assert_eq!(plain.state, WorkloadState::Ready);
        assert!(plain.approval.is_none());
        assert_eq!(
            store
                .decide(
                    &plain.workload_id,
                    &human("ada"),
                    ApprovalOutcome::Approved,
                    None,
                    "g",
                    &plain.input_hash,
                    11
                )
                .unwrap_err()
                .code,
            "WORKLOAD_APPROVAL_NOT_REQUIRED"
        );
        // Requiring approval is execution-affecting: the same key cannot add it.
        let mut gated = create();
        gated.approval = Some(ApprovalPolicy::default());
        let error = store.admit(gated).unwrap_err();
        assert_eq!(error.code, "IDEMPOTENCY_KEY_REUSED");
        assert_eq!(
            error.details.unwrap()["fields"],
            serde_json::json!(["approval"])
        );
        let _ = fs::remove_file(path);
    }

    #[test]
    fn work_whose_approval_wait_expired_never_runs_on_a_late_approval() {
        let mut policy_request = create();
        policy_request.approval = Some(ApprovalPolicy::default());
        policy_request.timeouts.approval_ms = Some(2_000);
        let (path, mut store) = store("approval-deadline");
        let w = store.create(policy_request).unwrap();
        let (id, input_hash) = (w.workload_id, w.input_hash);
        // created_at 10: the wait ends at 12.
        assert!(store.enforce_deadlines(11).unwrap().is_empty());
        // A late decision is refused as expired, and expires the work.
        let late = store
            .decide(
                &id,
                &human("ada"),
                ApprovalOutcome::Approved,
                None,
                "g",
                &input_hash,
                12,
            )
            .unwrap_err();
        assert_eq!(late.code, "WORKLOAD_TIMED_OUT");
        assert_eq!(late.details.unwrap()["deadline"], "approval");
        let w = store.get(&id).unwrap();
        assert_eq!(w.state, WorkloadState::DeadLetter);
        assert!(
            w.approval.as_ref().unwrap().decision.is_none(),
            "no decision was recorded"
        );
        assert_eq!(
            store
                .decide(
                    &id,
                    &human("ada"),
                    ApprovalOutcome::Approved,
                    None,
                    "g",
                    &input_hash,
                    13
                )
                .unwrap_err()
                .code,
            "WORKLOAD_APPROVAL_NOT_ACTIONABLE"
        );
        assert_eq!(
            store
                .claim(&id, "w", &capabilities(), 13, 10_000)
                .unwrap_err()
                .code,
            "WORKLOAD_NOT_READY"
        );
        // An operator re-arm does not stand in for approval: it waits again.
        store
            .retry_with_reason(&id, "op", 14, Some("ask finance again"))
            .unwrap();
        assert_eq!(
            store.get(&id).unwrap().state,
            WorkloadState::AwaitingApproval
        );
        assert_eq!(
            store
                .claim(&id, "w", &capabilities(), 14, 10_000)
                .unwrap_err()
                .code,
            "WORKLOAD_APPROVAL_PENDING"
        );
        store
            .decide(
                &id,
                &human("ada"),
                ApprovalOutcome::Approved,
                None,
                "g",
                &input_hash,
                15,
            )
            .unwrap();
        store.claim(&id, "w", &capabilities(), 15, 10_000).unwrap();
        let _ = fs::remove_file(path);
    }

    #[test]
    fn approval_stands_across_attempts_and_cancellation_ends_the_wait() {
        let (path, mut store, id, input_hash) =
            gated("approval-attempts", ApprovalPolicy::default());
        store
            .decide(
                &id,
                &human("ada"),
                ApprovalOutcome::Approved,
                None,
                "g",
                &input_hash,
                11,
            )
            .unwrap();
        let claim = store.claim(&id, "w", &capabilities(), 11, 10_000).unwrap();
        assert_eq!(
            store
                .fail(
                    &id,
                    &claim.claim_id,
                    "w",
                    claim.fencing_token,
                    "transient_connection",
                    "x",
                    12
                )
                .unwrap(),
            WorkloadState::Ready
        );
        // The retried attempt does not need a second approval.
        let next = store.get(&id).unwrap().next_attempt_at.unwrap();
        store
            .claim(&id, "w", &capabilities(), next, 10_000)
            .unwrap();
        let _ = fs::remove_file(path);

        let (path, mut store, id, input_hash) = gated("approval-cancel", ApprovalPolicy::default());
        assert_eq!(
            store.cancel(&id, "requester", 11).unwrap(),
            WorkloadState::Cancelled
        );
        assert_eq!(
            store
                .decide(
                    &id,
                    &human("ada"),
                    ApprovalOutcome::Approved,
                    None,
                    "g",
                    &input_hash,
                    12
                )
                .unwrap_err()
                .code,
            "WORKLOAD_APPROVAL_NOT_ACTIONABLE"
        );
        let _ = fs::remove_file(path);
    }

    #[test]
    fn concurrent_approval_and_claim_never_claim_unapproved_work() {
        for round in 0..20 {
            let path = std::env::temp_dir().join(format!(
                "workload-approval-race-{round}-{}.log",
                std::process::id()
            ));
            let _ = fs::remove_file(&path);
            let db = FeltDb::open(&path).unwrap();
            let mut seed = WorkloadStore::from_db(db.clone()).unwrap();
            let mut request = create();
            request.approval = Some(ApprovalPolicy::default());
            let w = seed.create(request).unwrap();
            let (id, input_hash) = (w.workload_id.clone(), w.input_hash.clone());
            let barrier = std::sync::Barrier::new(3);
            std::thread::scope(|scope| {
                let mut approver = WorkloadStore::from_db(db.clone()).unwrap();
                let claimers = [
                    WorkloadStore::from_db(db.clone()).unwrap(),
                    WorkloadStore::from_db(db.clone()).unwrap(),
                ];
                let (b, i, h) = (&barrier, &id, &input_hash);
                scope.spawn(move || {
                    b.wait();
                    let _ = approver.decide(
                        i,
                        &human("ada"),
                        ApprovalOutcome::Approved,
                        None,
                        "g",
                        h,
                        11,
                    );
                });
                for (n, mut claimer) in claimers.into_iter().enumerate() {
                    scope.spawn(move || {
                        b.wait();
                        for _ in 0..50 {
                            match claimer.claim(i, &format!("w{n}"), &capabilities(), 12, 10_000) {
                                Ok(_) => return,
                                Err(e) => assert!(
                                    matches!(
                                        e.code.as_str(),
                                        "WORKLOAD_APPROVAL_PENDING"
                                            | "WORKLOAD_NOT_READY"
                                            | "WORKLOAD_CONFLICT"
                                    ),
                                    "{}",
                                    e.code
                                ),
                            }
                        }
                    });
                }
            });
            let committed = committed_workload(&db, &id);
            // Whatever interleaving happened: if any attempt exists, the
            // approval was committed before it, and at most one claim landed.
            assert!(committed.attempts.len() <= 1, "round {round}");
            if !committed.attempts.is_empty() {
                let order: Vec<_> = committed.execution_history.iter().map(|e| e.to).collect();
                let ready = order
                    .iter()
                    .position(|s| *s == WorkloadState::Ready)
                    .unwrap();
                let claimed = order
                    .iter()
                    .position(|s| *s == WorkloadState::Claimed)
                    .unwrap();
                assert!(ready < claimed, "round {round}: {order:?}");
                assert!(committed.approval.as_ref().unwrap().approved());
            }
            let _ = fs::remove_file(path);
        }
    }
}