Struct SopEngine 

pub struct SopEngine { /* private fields */ }

Central SOP orchestrator: loads SOPs, matches triggers, manages run lifecycle.

Implementations

impl SopEngine

pub fn new(config: SopConfig) -> SopEngine

Create a new engine with the given config. Call reload() to load SOPs.

pub fn with_decision_models( self, models: HashMap<String, Arc<dyn DecisionModel>>, ) -> SopEngine

Register decision models by alias (see super::decision).

pub fn decision_model(&self, alias: &str) -> Option<Arc<dyn DecisionModel>>

The decision model an SOP selected by alias, if configured.

pub fn with_store(self, store: Arc<dyn SopRunStore>) -> SopEngine

Inject a durable run-state store (used by build_sop_engine). Default is an ephemeral in-memory store, so callers that don’t set one keep today’s behavior exactly.

pub fn with_metrics(self, metrics: Arc<SopMetricsCollector>) -> SopEngine

Inject the metrics collector. build_sop_engine passes the process-shared collector so the engine’s completion metrics and the SOP tools’ reports observe one set; the default per-engine collector keeps tests isolated.

pub fn with_run_notifier(self, tx: Sender<SopRunSummary>) -> SopEngine

Attach a live run-change notifier. build_sop_engine wires the gateway’s sender here so run transitions push to the Runs WebSocket. Returns the engine unchanged when never called (tests, headless embedders).

pub fn subscribe_run_changes(&self) -> Option<Receiver<SopRunSummary>>

Subscribe to the live run-change feed if a notifier is attached. Each item is a fresh SopRunSummary for the run that just transitioned.

pub fn with_capabilities( self, capabilities: Arc<SopCapabilityRegistry>, ) -> SopEngine

Inject a deterministic capability registry. Tests and future daemon wiring can replace the built-ins without adding another execution path.

pub fn with_execution_capability( self, execution_capability: AgentExecutionCapability, ) -> SopEngine

Bind executable SOP actions to the daemon-owned authority. The binding must be installed before managed producers can create actions.

pub fn has_execution_capability(&self) -> bool

pub fn with_approval_broker(self, broker: Arc<ApprovalBroker>) -> SopEngine

Inject the approval broker (built from [sop.approval] config). Defaults to a pass-through; build_sop_engine replaces it with the configured broker.

pub fn approval_broker(&self) -> Arc<ApprovalBroker>

The approval broker (membership + quorum authorization). Callers that must deliver an escalation to a policy’s second route read it here.

pub fn resolve_via_broker( &mut self, run_id: &str, decision: ApprovalDecision, principal: ApprovalPrincipal, ) -> Result<BrokerOutcome, Error>

Resolve a gate or deterministic checkpoint THROUGH the broker (membership + quorum), then its single transition owner. This is the entry point out-of-band surfaces (gateway / CLI / tools) should call so a [sop.approval] policy is enforced; with no policy it is exactly resolve_gate for a WaitingApproval run or the historical checkpoint resolver for a PausedCheckpoint run. The broker is cloned out first so it does not borrow self while self is mutated by the chokepoint.

This shared entry point does not consume a resumed deterministic capability tail inline: the caller receives the resumed action and hands it to the shared async executor, which reacquires the engine lock once per capability. That is what lets an operator cancellation persist CancelRequested between post-checkpoint capabilities instead of blocking on the engine mutex until the whole tail has already run.

A caller that genuinely owns an exclusive engine handle and must observe the fully-driven tail before returning wants Self::resolve_via_broker_inline instead — but note that no shared transport qualifies.

pub fn resolve_via_broker_inline( &mut self, run_id: &str, decision: ApprovalDecision, principal: ApprovalPrincipal, ) -> Result<BrokerOutcome, Error>

Resolve through the broker and synchronously drive the resumed deterministic capability tail while the engine mutex is still held.

⚠️ Cancellation cannot interleave with a tail driven this way. Reserved for genuinely exclusive callers (single-owner engine handles and tests that assert on the fully-driven tail). Shared transports — admin HTTP, WebSocket, RPC, channels, and the sop_approve tool — must use Self::resolve_via_broker, which defers the tail.

pub fn resolve_via_broker_deferred( &mut self, run_id: &str, decision: ApprovalDecision, principal: ApprovalPrincipal, ) -> Result<BrokerOutcome, Error>

Explicit alias for Self::resolve_via_broker, kept for call sites that want the deferral to be obvious at the point of use.

pub fn restore_runs(&mut self)

Reconstruct in-flight runs from the store at startup (durable backends). No-op for the in-memory default. Does not overwrite already-present runs.

pub fn is_gate_reference_superseded( &self, run_id: &str, reference_revision: u32, ) -> bool

A prompt becomes stale only after a replacement presentation is durable. A gate can update its in-memory revision before its parked snapshot saves; finalizing the old prompt in that window would leave operators without a recoverable replacement after a crash.

pub fn reload(&mut self, install_root: &Path)

Load/reload SOPs from the configured directory, resolved against install_root (the install root, config_path’s parent).

pub fn sops(&self) -> &[Sop]

Return all loaded SOP definitions.

pub fn active_runs(&self) -> &HashMap<String, SopRun>

Return all active (in-flight) runs.

pub fn get_run(&self, run_id: &str) -> Option<&SopRun>

Look up a run by ID (active or finished).

pub fn get_sop(&self, name: &str) -> Option<&Sop>

Look up an SOP by name.

pub fn match_trigger(&self, event: &SopEvent) -> Vec<&Sop>

Match an incoming event against all loaded SOPs and return the names of SOPs whose triggers match.

pub fn wants_source(&self, source: SopTriggerSource) -> bool

True when any loaded SOP has a trigger of this source. Fan-in callers use this as a cheap pre-filter before building and dispatching an event.

pub fn can_start(&self, sop_name: &str) -> bool

Check whether a new run can be started for the given SOP (respects cooldown and concurrency limits).

pub fn evaluate_admission(&self, sop_name: &str) -> SopAdmission

A2: decide how to admit a matched trigger for sop_name under its SopAdmissionPolicy. Admit still passes through the authoritative CAS in start_run; the other outcomes are surfaced by the dispatch layer so a non-admitted trigger is never silently lost. A cooldown or unknown SOP drops regardless of policy (a cooldown is a deliberate rate limit, not backpressure).

AUTHORITY: within a SINGLE daemon this decision is authoritative - the engine Mutex serializes evaluate_admission + the CAS claim, so two triggers cannot both admit past the policy. The exec-slot cap is additionally CAS-authoritative via the shared store even ACROSS engines. The pending-approval pool (max_pending_approvals), however, is only ADVISORY across engines: a run parks at approval only AFTER it has executed, so its pending slot cannot be atomically pre-reserved at admission time, and two engines sharing a store can each admit a run that later parks. Making the pending cap cross-engine- authoritative requires a store-level two-phase reservation (a follow-up); the single-daemon deployment - the common case - is fully authoritative today.

pub fn forget_active_dispatch_dedup_for_run(&mut self, run_id: &str)

Drop any active producer key pointing at this run.

The active-key lookup coalesces a later producer onto whatever run the key names for as long as that run stays in active_runs. A caller that starts a run it will not drive therefore has to withdraw the key, or the next producer is handed a run nothing is advancing instead of doing the work itself — the key would suppress real work rather than duplicate it.

Only the key is withdrawn. The run is untouched, because a caller that cannot drive it is also not the right place to decide its fate.

pub fn start_run( &mut self, sop_name: &str, event: SopEvent, ) -> Result<SopRunAction, Error>

pub fn start_run_owned( &mut self, sop_name: &str, event: SopEvent, initiator: Option<&str>, ) -> Result<SopRunAction, Error>

Self::start_run for a run started INSIDE an agent turn, recording that agent on the run.

An unowned procedure borrows its owner from the calling turn, which is enough right up until the run parks at an approval: the approved step resumes on the headless driver, with no turn to borrow from. Recording the initiator here is what lets that resume still run as the agent that started it. None for every headless trigger, which has no initiating turn to record.

pub fn start_run_with_mode( &mut self, sop_name: &str, event: SopEvent, decided_mode: Option<SopExecutionMode>, decisions: BTreeMap<u32, f64>, ) -> Result<SopRunAction, Error>

Start a headless-triggered run with a dispatch-decided execution mode. Dispatch has no initiating agent turn, so the initiator is None, as it is for every headless trigger in Self::start_run.

pub fn advance_step( &mut self, run_id: &str, result: SopStepResult, ) -> Result<SopRunAction, Error>

pub fn cancel_run(&mut self, run_id: &str) -> Result<(), Error>

Cancel an active run immediately without an operator-supplied reason.

Internal callers use this only when no step is executing concurrently. Remotely initiated cancellation must use cancel_run_idempotent, which requests cancellation for a running step and preserves its claim until the driver reaches a boundary.

pub fn finish_requested_cancellation( &mut self, run_id: &str, ) -> Result<Option<SopRunAction>, Error>

Finish a previously requested cancellation at a driver-owned execution boundary. Actor and reason are resolved from the durable request event, avoiding a second in-memory source of truth.

pub fn cancel_run_idempotent( &mut self, run_id: &str, reason: Option<String>, actor: Option<String>, ) -> Result<Option<CancelOutcome>, Error>

Cancel a run by id, idempotently. Classifies the run under a single borrow of engine state (no intervening lookup that a concurrent completion could race) and reports:

• Ok(Some(Cancelled)) - the run was active and is now Cancelled.
• Ok(Some(AlreadyTerminal(status))) - the run had already reached a terminal status (a repeat cancel, or it finished normally first); no second cancellation or audit event is recorded.
• Ok(None) - no run with this id is known to the engine.
• Err(_) - the terminal persist failed; see err_is_terminal_persistence_retained to tell a retryable store fault (the run stays active and claimed) from any other error.

pub fn settle_run_for_drained_generation( &mut self, run_id: &str, ) -> Result<Option<SopRunStatus>, Error>

Settle a run whose driver was refused because the generation that owned it had already drained.

Deliberately not Self::cancel_run_idempotent. That path is cooperative: a Running run becomes CancelRequested and stays active and claimed until a driver reaches its next boundary. This is the one case where no driver exists and none ever will, so waiting for a boundary is precisely the stall this prevents. The run therefore goes terminal immediately, through the same persistence path a normal cancellation uses — which is what releases the execution claim and stops a later engine rebuild from restoring it as Running and renewing the claim forever.

Cancelled rather than Failed: the work was withdrawn before it ran, not attempted and failed. The reason travels on the durable run_generation_drained event rather than failure_reason, which the terminal path stamps only for genuine failures.

Returns the status the run held before it was settled, or None when no active run carries this id — already terminal, or never started.

pub fn settle_run_for_aborted_driver( &mut self, run_id: &str, ) -> Result<Option<SopRunStatus>, Error>

Settle a run whose driver a reload aborted before it finished.

Teardown aborts a driver that overruns the drain deadline. Its run is still Running and claimed, and no driver will exist for it: the replacement generation restores active runs but does not start drivers for them, so without this the run would hold a concurrency slot indefinitely. It goes terminal here through the normal claim-releasing path, Failed because the step was underway and did not finish, with a durable run_driver_aborted event recording why. A run that had already asked to cancel is settled Cancelled instead, honoring that request.

A run parked at a gate, or already terminal, needs no settlement: parked runs are waiting on an operator, not a driver. Returns the prior status of a run it settled, or None when there was nothing to settle.

pub fn settle_orphaned_run( &mut self, run_id: &str, kind: OrphanedRunSettlement, ) -> Result<Option<SopRunStatus>, Error>

Settle a run no driver will advance, keeping ownership of the retry.

On success any pending retry for the run is cleared. On failure the run is recorded for Self::run_maintenance_tick to settle again, so a transient store error cannot leave it Running and claimed with nobody left to write its terminal state. The error is still returned so the caller can report it.

pub fn adopt_orphaned_run_settlements( &mut self, runs: impl IntoIterator<Item = (String, OrphanedRunSettlement)>, )

Hand this engine the retry for runs a previous engine could not settle.

A reload retires the old generation, but its stragglers may still be settling through the shared queue. If their terminal write failed, the durable row is still Running, and restoring it here would renew its claim with no driver to advance it. Recording the runs here makes this engine’s maintenance the owner of that terminal write.

pub fn orphaned_run_settlement_queue(&self) -> OrphanedRunSettlementQueue

Retain the canonical retry queue even after its drivers were pruned.

pub fn adopt_orphaned_run_settlement_queue( &mut self, queue: OrphanedRunSettlementQueue, )

Transfer retry ownership without losing writes from late old producers.

pub fn has_pending_orphan_settlement(&self, run_id: &str) -> bool

Runs awaiting a retried terminal write, for callers and tests that need to see whether maintenance still owns one.

pub fn approve_step(&mut self, run_id: &str) -> Result<SopRunAction, Error>

pub fn decide_checkpoint( &mut self, run_id: &str, decision: ApprovalDecision, ) -> Result<SopRunAction, Error>

Resolve a checkpoint decision (PausedCheckpoint). Approve resumes the success path (records the checkpoint Completed, pipes forward down routing.next); Deny takes the failure path (records the checkpoint Failed and routes through the step’s on_failure, exactly like a step that failed execution). This is the single entry point for both outcomes; callers never branch on status. approve_step is the Approve-only alias.

pub fn finished_runs(&self, sop_name: Option<&str>) -> Vec<&SopRun>

List finished runs, optionally filtered by SOP name.

pub fn run_summaries(&self, sop_name: Option<&str>) -> Vec<SopRunSummary>

Summaries of every run the engine currently holds: live runs from the active set plus retained terminal runs, newest first by start time. This is the enumeration the Runs surface polls; it never touches the durable store directly, so it reflects exactly what the running engine knows (active set + max_finished_runs retention window).

pub fn deterministic_savings(&self) -> &DeterministicSavings

Return cumulative deterministic execution savings.

pub fn save_proposal(&self, proposal: &ProposalRecord) -> Result<(), StoreError>

Save a procedural-memory proposal into the shared SOP store. This is the production-facing engine surface EPIC F consumes for approval/write-back.

pub fn load_proposal( &self, id: &str, ) -> Result<Option<ProposalRecord>, StoreError>

Load a procedural-memory proposal by id from the shared SOP store.

pub fn list_proposals( &self, status: Option<ProposalStatus>, ) -> Result<Vec<ProposalRecord>, StoreError>

List procedural-memory proposals, optionally filtered by lifecycle status.

pub fn start_deterministic_run( &mut self, sop_name: &str, event: SopEvent, ) -> Result<SopRunAction, Error>

Start a deterministic SOP run. Steps execute sequentially without LLM round-trips. Returns the first action (DeterministicStep or CheckpointWait).

pub fn drive_headless_deterministic( &mut self, run_id: &str, first_action: SopRunAction, ) -> Result<SopRunAction, Error>

pub fn advance_headless_deterministic_step( &mut self, run_id: &str, action: SopRunAction, ) -> Result<SopRunAction, Error>

Execute at most one deterministic capability step. Async drivers call this once per engine-lock acquisition so an operator cancellation can acquire the lock and become visible before the next step dispatches.

pub fn advance_deterministic_step( &mut self, run_id: &str, step_output: Value, step_timestamps: Option<(String, Option<String>)>, ) -> Result<SopRunAction, Error>

Advance a deterministic run with the output of the current step. The output is piped as input to the next step.

pub fn resume_deterministic_run( &mut self, state: DeterministicRunState, ) -> Result<SopRunAction, Error>

Resume a deterministic run from persisted state.

pub fn load_deterministic_state( path: &Path, ) -> Result<DeterministicRunState, Error>

Load a persisted deterministic run state from a JSON file.

pub fn check_approval_timeouts(&mut self) -> Vec<SopRunAction>

pub fn run_maintenance_tick(&mut self) -> MaintenanceSummary

pub fn set_sops_for_test(&mut self, sops: Vec<Sop>)

Replace loaded SOPs (for testing from other modules).

pub fn last_finished_run(&self, sop_name: &str) -> Option<&SopRun>

pub fn finish_run( &mut self, run_id: &str, status: SopRunStatus, reason: Option<String>, ) -> Result<SopRunAction, Error>

pub fn config(&self) -> &SopConfig

Read-only config access for the approval resolver.

pub fn approval_config(&self) -> &SopApprovalConfig

The live [sop.approval] config - the single source of truth for approval groups and policies. The broker resolves membership/policy from this at use-time rather than holding a cloned copy that could drift on reload.

pub fn current_step_policy_name(&self, run_id: &str) -> Option<String>

The name of the approval policy that applies to the run’s current step, if that step names one. Read surfaces collapse unavailable live state to None; the broker uses the fallible lookup above to fail closed.

pub fn run_events( &self, run_id: &str, ) -> Result<Vec<SopEventRecord>, StoreError>

Ordered event/ledger history for a run (from the durable store).

pub fn resolve_gate( &mut self, run_id: &str, decision: ApprovalDecision, principal: ApprovalPrincipal, ) -> Result<ResolveOutcome, Error>