feat(shutdown): a stop that SAVES, and says what reached disk

core/continuum-core/src/cognition/mod.rs

@@ -59,6 +59,7 @@ pub mod deliberation_parse;
 pub mod deliberation_prompt;
 pub mod dispatch_listener;
 pub mod activity_gate;
+pub mod turn_ingress;
 pub mod dream_consolidation;
 pub mod embedding;
 pub mod eval;

core/continuum-core/src/cognition/turn_ingress.rs

@@ -0,0 +1,112 @@
+//! THE DOOR A STOPPING NODE CLOSES.
+//!
+//! One process-wide flag: may a persona's service loop begin a NEW turn?
+//!
+//! # Why this is not `quiesced`
+//!
+//! `PersonaAircRuntimeRegistry`'s per-persona `quiesced` flag suspends the autonomic
+//! SELF-TICK and deliberately leaves her reachable — its contract is "she stops wandering
+//! and still picks up the phone", because a measurement lease must not make a citizen
+//! unreachable. That is the right behaviour for a benchmark and the wrong one for a stop:
+//! a quiesced citizen still starts turns when addressed, so a `save_state` taken after a
+//! quiesce can still land underneath a turn halfway through writing.
+//!
+//! So this is a different question with a different answer, and folding it into `quiesced`
+//! would break the lease's promise. Closing this door stops EVERY new turn — directed,
+//! self-directed, forked — and lets the ones already running finish.
+//!
+//! # One-way, on purpose
+//!
+//! There is no `open()`. The only caller is the shutdown drain, and the process exits
+//! immediately afterwards; a reopen verb would exist solely to be called by mistake. A
+//! node that wants turns again starts a core.
+
+use crate::runtime::AdmissionGate;
+
+/// The process's turn gate. One [`AdmissionGate`]: open/closed and the in-flight count in
+/// one word, so an admission cannot slip between a close and the drain's read.
+///
+/// The gate logic lives in `runtime::admission_gate` rather than here because the log
+/// queue needs exactly the same invariant, and when it was written twice the second copy
+/// reintroduced the race the first had just removed.
+static GATE: AdmissionGate = AdmissionGate::new();
+
+/// SERVICE-LOOP turns running right now, across every persona in this process.
+///
+/// # What this number does NOT include, stated because a drain keys on it
+///
+/// Only work that passed through [`admit`] is counted, and today that is the persona
+/// service loop. A `cognition/eval` fork, or any caller that reaches the cognition
+/// faculties directly rather than through a citizen's loop, runs UNCOUNTED — so a drain
+/// can report the node quiet while such a call is mid-flight.
+///
+/// That is a narrowed contract, not an oversight, and it is narrowed rather than widened
+/// because the alternative is worse: an eval fork holding a permit would keep the whole
+/// node from draining for the length of a benchmark, and a permit taken somewhere that
+/// does not release it on every path leaks a phantom turn that no drain can ever clear.
+/// The service loop is the one caller whose entry and exit are a single bounded scope.
+///
+/// The consequence a reader must not be surprised by: a stop taken DURING an eval saves a
+/// consistent citizen but may cut the eval. Widening this is a real piece of work —
+/// permits at the faculty boundary, with the same RAII discipline — and it belongs to
+/// whoever gives `cognition/eval` a lifecycle, not to a shutdown rail. Raised in review
+/// by Astra.
+pub fn in_flight() -> u64 {
+    GATE.in_flight()
+}
+
+/// ADMIT a turn, or refuse because the node is stopping.
+///
+/// # Why not read `activity_gate`'s engaged flag
+///
+/// `persona_engaged` is stamped where a serving LANE is acquired, deliberately — a room
+/// wake alone is not wakefulness, or a busy room would cancel every dream. So a turn that
+/// has been admitted and is still composing context, or is queued for a lane, is not yet
+/// `engaged`. Draining on that flag would walk past exactly the turns that have taken
+/// input and not yet written anything, which are the ones whose loss is invisible.
+#[must_use = "the permit IS the turn's admission; dropping it immediately ends the turn"]
+pub fn admit() -> Option<crate::runtime::Permit<'static>> {
+    GATE.admit()
+}
+
+/// Shut the door. Returns whether THIS call closed it, so a second drain — a signal
+/// racing the `system/shutdown` verb, since both reach the same broadcast — can tell it is
+/// re-entering rather than report a fresh close.
+pub fn close() -> bool {
+    GATE.close()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // what this catches: the process gate being wired to something other than an open
+    // AdmissionGate — a gate that defaulted closed would make every citizen refuse to take
+    // a turn, a total silent outage no shutdown test would exercise.
+    //
+    // The gate's BEHAVIOUR — the admit/close race, the CAS retry, the shared word — is
+    // tested against the real type in `runtime::admission_gate`, on instances a test can
+    // close without poisoning the process. This module's job is only to hold one and hand
+    // it to the service loop, so that is all this asserts.
+    #[test]
+    fn the_process_turn_gate_starts_open_and_empty() {
+        // Openness is asserted by ADMITTING, not by a separate reader: a bare `is_open`
+        // on this module had no caller at all — the service loop takes a permit — so it
+        // was dead weight and is gone.
+        //
+        // It was NOT the reachability-ratchet failure, though I first claimed it was. The
+        // scanner greps `pub struct ` / `pub enum ` (production_reachability.rs:125) and
+        // does not count functions at all, so removing one could never have moved the
+        // number. The counted item was `pub struct ShutdownOperation`, now private.
+        // Corrected by Astra, who READ the scanner rather than inferring what it measures
+        // — which is what I should have done before claiming a fix for it.
+        let permit = admit().expect("an open gate admits");
+        assert_eq!(
+            in_flight(),
+            1,
+            "an admitted turn must be visible to a drain"
+        );
+        drop(permit);
+        assert_eq!(in_flight(), 0, "and invisible once it ends");
+    }
+}

core/continuum-core/src/commands/log/write.rs

@@ -40,9 +40,9 @@ crate::action_command! {
     params: WriteLogPayload,
     output: WriteLogResult,
     run(this, _ctx, p) => {
-        this.state
-            .log_tx
-            .send(p)
+        // Through the counting choke point, not the raw sender: the shutdown drain waits
+        // on that depth, and an entry enqueued around it is one the drain cannot see.
+        crate::modules::logger::enqueue_log_blocking(&this.state.log_tx, p)
             .map_err(|e| format!("Queue send failed: {e}"))?;
         this.state.requests_processed.fetch_add(1, Ordering::Relaxed);
         Ok(WriteLogResult { bytes_written: 0 })

core/continuum-core/src/commands/log/write_batch.rs

@@ -51,7 +51,9 @@ crate::action_command! {
         let entries_queued = p.entries.len();
         for entry in p.entries {
             // Best-effort: drop on a full queue rather than block the submitter.
-            let _ = this.state.log_tx.try_send(entry);
+            // Counted: see `enqueue_log`. A batch that bypassed it hid its whole size
+            // from the drain.
+            let _ = crate::modules::logger::enqueue_log(&this.state.log_tx, entry);
         }
         this.state.requests_processed.fetch_add(1, Ordering::Relaxed);
         Ok(WriteLogBatchResult { entries_queued })

core/continuum-core/src/commands/system.rs

@@ -28,6 +28,10 @@ pub mod memory_gate;
 pub mod pressure;
 pub mod pressure_broker_state;
 pub mod resources;
+// `system/shutdown` holds no deps, so `action_command!`'s stateless arm registers it
+// onto the one registry by itself — it is deliberately NOT in `command_objects` below,
+// which is only for commands that need construction with a service.
+pub mod shutdown;
 
 /// Shared params for the no-argument `system/*` reads (cpu, memory, pressure,
 /// memory-gate, memory-budget, docker-tier-stats). One empty contract reused across

core/continuum-core/src/commands/system/shutdown.rs

@@ -0,0 +1,147 @@
+//! `system/shutdown` — ask the RUNNING core to stop itself gracefully and say what
+//! reached disk.
+//!
+//! # Why a resident verb rather than a signal
+//!
+//! `continuum stop` reached for `taskkill /F` on Windows and a kill tree elsewhere. A
+//! forced kill runs no module's `save_state`, so a stop that looked identical to a clean
+//! one lost every module's volatile state — and the CLI printed a success line and exited
+//! 0 either way, because nothing came back from the kill except "the process is gone".
+//!
+//! Signals do exist and `main.rs` wires them, but they are not a request/response: the
+//! handler cannot hand an exit code or a receipt back to the operator who typed `stop`,
+//! and on Windows the console-control arms are best-effort. This verb travels the socket
+//! request path every other command already uses, so the answer comes back the same way
+//! `ping`'s does — no new transport, no Windows-specific rail.
+//!
+//! `Privileged`, never `AiSafe`: stopping the node is not a thing a persona reasoning
+//! about its own workload should be able to reach for.
+
+use serde::{Deserialize, Serialize};
+use ts_rs::TS;
+
+use crate::runtime::ShutdownReceipt;
+
+use super::SystemQuery;
+
+/// How the core answered a stop request. Carries the receipt so the caller can exit
+/// non-zero and NAME what did not save, rather than reporting a success it never
+/// established.
+#[derive(Debug, Clone, Serialize, Deserialize, TS)]
+#[ts(
+    export,
+    export_to = "../../../protocol/typescript/system/ShutdownResult.ts"
+)]
+pub struct ShutdownResult {
+    /// Every module's durable state reached disk. The one field a caller may key an exit
+    /// code on.
+    pub state_is_durable: bool,
+    /// One line naming what did not save, if anything.
+    pub summary: String,
+    pub receipt: ShutdownReceipt,
+}
+
+crate::action_command! {
+    /// Drain, save and join every module. Return the retained shutdown receipt before
+    /// the calling CLI tears down the process.
+    pub struct SystemShutdown;
+    name: "system/shutdown",
+    access: Privileged,
+    params: SystemQuery,
+    output: ShutdownResult,
+    run(_this, _ctx, _p) => {
+        // TRIGGER AND OBSERVE — the operation itself belongs to the runtime.
+        //
+        // This handler used to run the broadcast inline, and a socket handler is
+        // cancelled when its client goes away. A CLI that died or was interrupted
+        // mid-request therefore left the node with turn ingress CLOSED and no shutdown:
+        // every citizen refusing work, nothing saved, nothing to restart it. Strictly
+        // worse than the forced kill this verb exists to replace. `begin_shutdown` runs
+        // it in a task no connection owns, and is idempotent, so a signal racing this
+        // request joins the same broadcast rather than saving twice over one state.
+        // Refuse BEFORE triggering when no runtime is installed. `begin_shutdown`
+        // publishes an empty receipt in that case so no observer hangs — but an empty
+        // receipt is "all zero modules were durable", which is true and says nothing, and
+        // the caller reads a success here as permission to stop hard-killing. A core in
+        // its boot window has state at stake and no one to save it; that is a refusal,
+        // not a clean stop.
+        if crate::runtime::signal_runtime().is_none() {
+            return Err(crate::sdk_codegen::CommandError::Internal(
+                "no runtime is installed yet — the core cannot stop gracefully, so nothing                  here says its state is durable"
+                    .to_string(),
+            ));
+        }
+        let rx = crate::runtime::begin_shutdown();
+
+        // Bounded so a wedged module cannot hold the connection open forever. A timeout
+        // here does NOT cancel the shutdown — it is still running, un-cancelled, in the
+        // runtime's task — so the honest answer is that durability is unknown, not that
+        // it failed.
+        const OBSERVE: std::time::Duration = std::time::Duration::from_secs(15);
+        let Some(receipt) = crate::runtime::await_shutdown(rx, OBSERVE).await else {
+            return Err(crate::sdk_codegen::CommandError::Internal(format!(
+                "shutdown is still running after {}s — it was NOT cancelled, and whether                  every module saved is unknown from here",
+                OBSERVE.as_secs()
+            )));
+        };
+
+        // The process is deliberately still alive. Every module has drained, saved and
+        // joined; what a stopped node is FOR is the caller's decision — the CLI tears it
+        // down having read this receipt, and the signal path exits because it must.
+        //
+        // The previous version exited here on a 400ms timer, which proved nothing about
+        // whether this response had reached the wire: a handler that kills its own
+        // process cannot also answer, and a timer guessing when its answer flushed is a
+        // guess. Handing the teardown back to the caller removes the question instead of
+        // estimating it.
+        Ok(ShutdownResult {
+            state_is_durable: receipt.state_is_durable(),
+            summary: receipt.summary(),
+            receipt,
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::sdk_codegen::{ActionCommand, Ctx};
+
+    // what this catches: the access level. `system/shutdown` on the AiSafe surface would
+    // let any persona stop the node it is thinking inside.
+    #[test]
+    fn stopping_the_node_is_privileged_not_ai_safe() {
+        assert_eq!(SystemShutdown::NAME, "system/shutdown");
+        assert!(matches!(
+            SystemShutdown::ACCESS,
+            crate::sdk_codegen::AccessLevel::Privileged
+        ));
+    }
+
+    // what this catches: a core with no runtime answering "stopped cleanly". The caller
+    // treats a successful response as permission to stop hard-killing, so a success here
+    // would be read as "state is safe" about a core that never ran a save.
+    #[tokio::test]
+    async fn without_a_runtime_it_refuses_rather_than_claiming_a_clean_stop() {
+        // `signal_runtime()` is a process-wide OnceLock. In a test binary that has not
+        // installed one this is None; if another test in the same process installed it,
+        // this assertion would be about a live runtime, so it checks the branch it can.
+        if crate::runtime::signal_runtime().is_none() {
+            let err = SystemShutdown
+                .run(&Ctx::default(), SystemQuery {})
+                .await
+                .expect_err("no runtime must refuse");
+            // Matched on the VARIANT, not on formatted text: `Internal` is the category
+            // that says "the node could not do this", and a future refactor that turned
+            // this into `Invalid` (a caller error) would be a real behaviour change the
+            // test should catch rather than paper over with a substring match.
+            let crate::sdk_codegen::CommandError::Internal(msg) = err else {
+                panic!("a core that cannot stop gracefully is an Internal failure, not a caller error");
+            };
+            assert!(
+                msg.contains("durable"),
+                "the refusal must say what is unestablished, got: {msg}"
+            );
+        }
+    }
+}

core/continuum-core/src/main.rs

@@ -140,9 +140,20 @@ fn install_shutdown_handlers() {
                 _ = close_fut => {}
                 _ = shutdown_fut => {}
             }
-            eprintln!("[continuum-core] shutdown signal — killing sentinel process groups");
+            eprintln!("[continuum-core] shutdown signal — save-and-join broadcast, then exit");
+            // THE SAME BROADCAST THE UNIX ARMS RUN. This arm killed sentinels, slept a
+            // flat 2 seconds, and `_exit`ed — which is precisely the behaviour the SIGTERM
+            // arm's own comment says was replaced on 2026-09-02 ("a flat 2s sleep during
+            // which NOTHING saved"). The unix half was fixed and this half was not, so on
+            // Windows no module has ever saved on a signal stop: not the citizens'
+            // workspaces, not the log queue, nothing. The node that runs the citizens is a
+            // Windows node.
+            //
+            // Ordering matches unix deliberately: modules drain and save FIRST, sentinels
+            // are killed after (they are children, not modules), then the fast `_exit`
+            // that skips llama.cpp's double-free-prone static destructors.
+            continuum_core::runtime::run_signal_shutdown().await;
             continuum_core::modules::sentinel::shutdown_all_sentinels();
-            tokio::time::sleep(std::time::Duration::from_secs(2)).await;
             unsafe { libc::_exit(0) };
         });
     }

core/continuum-core/src/modules/cognition.rs

@@ -216,6 +216,52 @@ impl ServiceModule for CognitionModule {
         }
     }
 
+    /// CLOSE THE DOOR AND WAIT FOR THE CITIZENS TO FINISH THEIR TURNS.
+    ///
+    /// This is the module the drain phase exists for. Everything else in a stopping node
+    /// either holds no work or holds a buffer; cognition holds a citizen mid-thought, and
+    /// a `save_state` taken across that boundary writes a working memory that is half of
+    /// one turn and half of the next.
+    ///
+    /// Two steps, in this order:
+    ///
+    /// 1. `turn_ingress::close()` — no service loop admits another turn. NOT
+    ///    `quiesce_all`, whose contract is the opposite: a quiesced citizen stops
+    ///    wandering and deliberately STAYS reachable, so she would keep starting turns
+    ///    while we tried to save her.
+    /// 2. Wait for the turns already admitted to finish, bounded.
+    ///
+    /// The count comes from the permits the loops hold, not from `activity_gate`'s
+    /// `engaged` flag: engagement is stamped at serving-lane acquisition, so a turn that
+    /// has taken a message and is still composing context is not engaged, and draining on
+    /// that flag would walk past exactly the turns whose loss leaves no trace.
+    ///
+    /// Returns the number still running when the budget expired — a citizen who is still
+    /// thinking is a fact the receipt reports, not one it waits forever for.
+    async fn drain(&self) -> Result<u32, String> {
+        const POLL: std::time::Duration = std::time::Duration::from_millis(50);
+        // Inside the runtime's 2s phase bound, so the deadline that fires is this one,
+        // with a count, rather than the outer timeout, which produces no number.
+        const BUDGET: std::time::Duration = std::time::Duration::from_millis(1_800);
+
+        let first_to_close = crate::cognition::turn_ingress::close();
+        let deadline = std::time::Instant::now() + BUDGET;
+        while std::time::Instant::now() < deadline {
+            if crate::cognition::turn_ingress::in_flight() == 0 {
+                return Ok(0);
+            }
+            tokio::time::sleep(POLL).await;
+        }
+        let left = crate::cognition::turn_ingress::in_flight();
+        crate::probe!(
+            class = "cognition.drain.incomplete",
+            in_flight = left,
+            first_to_close = first_to_close,
+            "citizens were still inside turns when the drain budget expired — whatever is saved next is a snapshot taken mid-turn, and the receipt says so"
+        );
+        Ok(left.min(u32::MAX as u64) as u32)
+    }
+
     async fn initialize(&self, _ctx: &ModuleContext) -> Result<(), String> {
         // No init needed. Recipes are JSON data walked by the host
         // (TS recipe loader for the chat path today; future Rust