Module reducer 

Reducer task body. Sole nixfleet_state_machine::step caller per invariant (1) in runtime::mod.

State held in-task:

• per-rollout HostRolloutState keyed by rollout_id (the agent owns its own host’s state, so the key is just the rollout id)
• cached SignedManifestSet (refreshed by the manifest_poll worker per RFC-0004 §1 invariant #1 — single signed source of truth fetched + verified once per tick; the reducer reads rollout policy from it for step() calls)

Seq assignment: workers emit events with seq = 0. The reducer rewrites it to state.last_event_seq + 1 before calling step() — single mutator owns the per-rollout monotonic counter, so cross-worker ordering can’t race.

Constants

SUSTAINED_FAILURE_THRESHOLD_SECS 🔒

Sustained-failure window cap. RFC-0005 §6 — the agent transitions Soaking → Failed when a probe has been failing continuously past this threshold.

Functions

all_enforce_probes_pass 🔒

All enforce-mode probes have status Pass. Observe and Disabled are ignored per RFC-0007 §3.3 (ProbeMode docstring, state.rs); they do not gate convergence. Mirror of collect_failing_enforce_probes on the Soaking → Converged exit path. Empty enforce set trivially satisfies — matches the shared verifier’s “empty probe map acceptable” semantic in verify_converged_invariants.

apply_bootstrap_snapshot 🔒

LIFT #3 + LIFT #4: apply a CP-supplied HostRolloutSnapshot to the agent’s in-memory reducer cache, then emit the worker re-priming effects the rehydrated state demands.

bootstrap_pending_state 🔒

First-touch bootstrap for a fresh LocalActivate event. Pure: derives channel from the canonical RolloutId composite (RFC-0008 §6.3); the caller threads in the manifest-looked-up target_closure for this host (selecting by hostname == cfg.machine_id) and the CP-resolved soak_due_at carried by the LocalActivate event from DispatchResponse.soak_due_at (CP is the single source of truth for the policy-resolved soak window per RFC-0004 §1 invariant 1). Caller also threads now so the helper stays clock-injection-free.

collect_failing_enforce_probes 🔒

Collect probe names that are currently failing AND declared with mode = Enforce. Per RFC-0007 §3.4, only Enforce-mode probes participate in the soak gate; Observe and Disabled records events but does not gate. The pre-fix builder filtered only by status == Fail, which silently included failing Observe-mode probes in LocalSustainedFailureCrossed.failing_probes and gated soak promotion against the documented contract.

handle_input 🔒

merge_snapshot_into_state 🔒

Pure merge of a wire snapshot with the existing in-memory entry. Canonical fields (state, target_closure, dispatch/activation timestamps, last_event_seq) come from the snapshot. Agent-local fields not carried in the wire shape are preserved from existing when present, defaulted when not.

resolve_policy 🔒

run

run_advance_tick 🔒

run_host_event 🔒

with_seq 🔒

Rewrite the seq field on a Local* event. The reducer owns the monotonic counter (single mutator) so workers can emit with seq = 0 and let this function fill it in.