From 7aca03a4fa9d829498b7d3b2496a8964a58be2cd Mon Sep 17 00:00:00 2001
From: chriscoveries <212684707+chriscoveries@users.noreply.github.com>
Date: Thu, 22 Oct 2026 16:54:59 +0000
Subject: [PATCH 1/7] test(ci): route current ChatGPT browser fixtures to
 nightly

---
 .config/nextest.toml | 10 ++++++++++
 1 file changed, 10 insertions(+)

diff --git a/.config/nextest.toml b/.config/nextest.toml
index 1e34fa948..3a9c6ab96 100644
--- a/.config/nextest.toml
+++ b/.config/nextest.toml
@@ -36,8 +36,13 @@ not (package(cccc-pair-web) & (
   | test(=browser_surface::browser_surface_tests::shared_owner::shared_windows_submit_independent_batches_while_viewers_capture)
   | test(=browser_surface::browser_surface_tests::shutdown_closes_all_browser_processes)
   | test(=browser_surface::browser_surface_tests::special_key_command_applies_native_input_behavior)
+  | test(=browser_surface::pairing::tests::current_chatgpt_pairing_receipt_keeps_exact_user_and_assistant_boundaries)
   | test(=browser_surface::prompt_submission::grok_tests::grok_controls_busy_state_drafts_and_user_receipts)
   | test(=browser_surface::prompt_submission::readiness_tests::chatgpt_auth_and_landing_inputs_are_not_conversation_composers)
+  | test(=browser_surface::prompt_submission::readiness_tests::current_chatgpt_composer_is_not_an_auth_or_search_input)
+  | test(=browser_surface::prompt_submission::readiness_tests::current_chatgpt_layout_submits_once_and_protects_drafts_and_generation)
+  | test(=browser_surface::prompt_submission::readiness_tests::current_chatgpt_user_bubble_is_required_for_delivery_receipt)
+  | test(=browser_surface::prompt_submission::readiness_tests::failed_page_inspection_invalidates_earlier_readiness)
   | test(=browser_surface::prompt_submission::readiness_tests::image_compat_submission_waits_for_enabled_button_and_sends_once)
   | test(=browser_surface::prompt_submission::readiness_tests::owned_draft_recovery_rechecks_composer_and_page_before_replacement)
   | test(=browser_surface::prompt_submission::readiness_tests::send_activation_rechecks_ownership_and_uses_the_button_handler_once)
@@ -91,8 +96,13 @@ package(cccc-pair-web) & (
   | test(=browser_surface::browser_surface_tests::shared_owner::shared_windows_submit_independent_batches_while_viewers_capture)
   | test(=browser_surface::browser_surface_tests::shutdown_closes_all_browser_processes)
   | test(=browser_surface::browser_surface_tests::special_key_command_applies_native_input_behavior)
+  | test(=browser_surface::pairing::tests::current_chatgpt_pairing_receipt_keeps_exact_user_and_assistant_boundaries)
   | test(=browser_surface::prompt_submission::grok_tests::grok_controls_busy_state_drafts_and_user_receipts)
   | test(=browser_surface::prompt_submission::readiness_tests::chatgpt_auth_and_landing_inputs_are_not_conversation_composers)
+  | test(=browser_surface::prompt_submission::readiness_tests::current_chatgpt_composer_is_not_an_auth_or_search_input)
+  | test(=browser_surface::prompt_submission::readiness_tests::current_chatgpt_layout_submits_once_and_protects_drafts_and_generation)
+  | test(=browser_surface::prompt_submission::readiness_tests::current_chatgpt_user_bubble_is_required_for_delivery_receipt)
+  | test(=browser_surface::prompt_submission::readiness_tests::failed_page_inspection_invalidates_earlier_readiness)
   | test(=browser_surface::prompt_submission::readiness_tests::image_compat_submission_waits_for_enabled_button_and_sends_once)
   | test(=browser_surface::prompt_submission::readiness_tests::owned_draft_recovery_rechecks_composer_and_page_before_replacement)
   | test(=browser_surface::prompt_submission::readiness_tests::send_activation_rechecks_ownership_and_uses_the_button_handler_once)

From 8e5b539d80cab0c0fa128708712206e7006e4b68 Mon Sep 17 00:00:00 2001
From: chriscoveries <212684707+chriscoveries@users.noreply.github.com>
Date: Thu, 22 Oct 2026 19:46:28 +0000
Subject: [PATCH 2/7] feat(runtime): observe Hermes terminal turns from owned
 leases

---
 Cargo.lock                                    |  62 ++++
 crates/cccc-cli/src/args/actor.rs             |   6 +
 crates/cccc-cli/src/commands/actor.rs         |   5 +-
 crates/cccc-contracts/src/actor.rs            |  19 ++
 crates/cccc-contracts/src/lib.rs              |   2 +-
 crates/cccc-core/src/profiles.rs              |  19 +-
 crates/cccc-daemon/Cargo.toml                 |   3 +-
 crates/cccc-daemon/src/ops/actor_activity.rs  |   1 +
 .../src/ops/actor_profile_runtime.rs          |  10 +-
 crates/cccc-daemon/src/ops/actor_runtime.rs   |   6 +-
 crates/cccc-daemon/src/ops/mod.rs             |   1 +
 .../cccc-daemon/src/ops/turn_probe/hermes.rs  | 169 +++++++++++
 crates/cccc-daemon/src/ops/turn_probe/mod.rs  | 157 ++++++++++
 .../cccc-daemon/src/ops/turn_probe/process.rs | 135 +++++++++
 .../cccc-daemon/src/ops/turn_probe/tests.rs   | 283 ++++++++++++++++++
 crates/cccc-daemon/src/ops/working_state.rs   |  23 +-
 .../fixtures/hermes-turn-probe/README.md      |  22 ++
 .../fixtures/hermes-turn-probe/capture.json   |  51 ++++
 .../fixtures/hermes-turn-probe/during.sqlite3 | Bin 0 -> 32768 bytes
 docs/runtime-turn-probes.md                   |  40 +++
 20 files changed, 1005 insertions(+), 9 deletions(-)
 create mode 100644 crates/cccc-daemon/src/ops/turn_probe/hermes.rs
 create mode 100644 crates/cccc-daemon/src/ops/turn_probe/mod.rs
 create mode 100644 crates/cccc-daemon/src/ops/turn_probe/process.rs
 create mode 100644 crates/cccc-daemon/src/ops/turn_probe/tests.rs
 create mode 100644 crates/cccc-daemon/tests/fixtures/hermes-turn-probe/README.md
 create mode 100644 crates/cccc-daemon/tests/fixtures/hermes-turn-probe/capture.json
 create mode 100644 crates/cccc-daemon/tests/fixtures/hermes-turn-probe/during.sqlite3
 create mode 100644 docs/runtime-turn-probes.md

diff --git a/Cargo.lock b/Cargo.lock
index edefad3fe..2ced96bf9 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -19,6 +19,18 @@ dependencies = [
  "cpufeatures 0.2.17",
 ]
 
+[[package]]
+name = "ahash"
+version = "0.8.12"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75"
+dependencies = [
+ "cfg-if",
+ "once_cell",
+ "version_check",
+ "zerocopy",
+]
+
 [[package]]
 name = "aho-corasick"
 version = "1.1.4"
@@ -592,6 +604,7 @@ dependencies = [
  "percent-encoding",
  "regex",
  "reqwest 0.12.28",
+ "rusqlite",
  "rustls 0.23.37",
  "same-file",
  "serde",
@@ -1409,6 +1422,18 @@ dependencies = [
  "windows-sys 0.61.2",
 ]
 
+[[package]]
+name = "fallible-iterator"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2acce4a10f12dc2fb14a218589d4f1f62ef011b2d0cc4b3cb1bba8e94da14649"
+
+[[package]]
+name = "fallible-streaming-iterator"
+version = "0.1.9"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7360491ce676a36bf9bb3c56c1aa791658183a54d2744120f27285738d90465a"
+
 [[package]]
 name = "fastrand"
 version = "2.4.1"
@@ -1712,6 +1737,9 @@ name = "hashbrown"
 version = "0.14.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1"
+dependencies = [
+ "ahash",
+]
 
 [[package]]
 name = "hashbrown"
@@ -1719,6 +1747,15 @@ version = "0.17.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
 
+[[package]]
+name = "hashlink"
+version = "0.9.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6ba4ff7128dee98c7dc9794b6a411377e1404dba1c97deb8d1a55297bd25d8af"
+dependencies = [
+ "hashbrown 0.14.5",
+]
+
 [[package]]
 name = "heck"
 version = "0.4.1"
@@ -2261,6 +2298,17 @@ dependencies = [
  "libc",
 ]
 
+[[package]]
+name = "libsqlite3-sys"
+version = "0.30.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2e99fb7a497b1e3339bc746195567ed8d3e24945ecd636e3619d20b9de9e9149"
+dependencies = [
+ "cc",
+ "pkg-config",
+ "vcpkg",
+]
+
 [[package]]
 name = "linux-raw-sys"
 version = "0.12.1"
@@ -3122,6 +3170,20 @@ dependencies = [
  "windows-sys 0.52.0",
 ]
 
+[[package]]
+name = "rusqlite"
+version = "0.32.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7753b721174eb8ff87a9a0e799e2d7bc3749323e773db92e0984debb00019d6e"
+dependencies = [
+ "bitflags 2.13.0",
+ "fallible-iterator",
+ "fallible-streaming-iterator",
+ "hashlink",
+ "libsqlite3-sys",
+ "smallvec",
+]
+
 [[package]]
 name = "rust-embed"
 version = "8.12.0"
diff --git a/crates/cccc-cli/src/args/actor.rs b/crates/cccc-cli/src/args/actor.rs
index 840ea81e0..fc8a87c25 100644
--- a/crates/cccc-cli/src/args/actor.rs
+++ b/crates/cccc-cli/src/args/actor.rs
@@ -20,6 +20,9 @@ pub enum ActorAction {
         runtime: String,
         #[arg(long, default_value = "default", value_parser = ["default", "acp"])]
         runtime_mode: String,
+        /// Read-only native turn observation; takes effect on the next start.
+        #[arg(long, default_value = "off", value_parser = ["off", "auto", "hermes_turn_lease"])]
+        turn_probe: String,
         #[arg(long, default_value = "")]
         command: String,
         #[arg(long = "env")]
@@ -47,6 +50,9 @@ pub enum ActorAction {
         runtime: Option<String>,
         #[arg(long, value_parser = ["default", "acp"])]
         runtime_mode: Option<String>,
+        /// Read-only native turn observation; takes effect on the next start.
+        #[arg(long, value_parser = ["off", "auto", "hermes_turn_lease"])]
+        turn_probe: Option<String>,
         #[arg(long)]
         scope: Option<String>,
         #[arg(long)]
diff --git a/crates/cccc-cli/src/commands/actor.rs b/crates/cccc-cli/src/commands/actor.rs
index 20fb86e68..172ea1e14 100644
--- a/crates/cccc-cli/src/commands/actor.rs
+++ b/crates/cccc-cli/src/commands/actor.rs
@@ -22,6 +22,7 @@ pub async fn run(client: &DaemonClient, home: &HomeLayout, args: ActorArgs) -> R
             title,
             runtime,
             runtime_mode,
+            turn_probe,
             command,
             env: raw_env,
             scope,
@@ -36,7 +37,7 @@ pub async fn run(client: &DaemonClient, home: &HomeLayout, args: ActorArgs) -> R
                 "actor_add",
                 json!({
                     "group_id":group(home,group_id)?,"actor_id":actor_id,"title":title,
-                    "runtime":runtime,"runtime_mode":runtime_mode,"command":command,"env":env(raw_env)?,
+                    "runtime":runtime,"runtime_mode":runtime_mode,"turn_probe":turn_probe,"command":command,"env":env(raw_env)?,
                     "default_scope_key":scope,"submit":submit,"by":by
                 }),
             )
@@ -52,6 +53,7 @@ pub async fn run(client: &DaemonClient, home: &HomeLayout, args: ActorArgs) -> R
             title,
             runtime,
             runtime_mode,
+            turn_probe,
             scope,
             command,
             env: raw_env,
@@ -63,6 +65,7 @@ pub async fn run(client: &DaemonClient, home: &HomeLayout, args: ActorArgs) -> R
             optional(&mut patch, "title", title);
             optional(&mut patch, "runtime", runtime);
             optional(&mut patch, "runtime_mode", runtime_mode);
+            optional(&mut patch, "turn_probe", turn_probe);
             if let Some(scope) = scope {
                 patch.insert(
                     "default_scope_key".into(),
diff --git a/crates/cccc-contracts/src/actor.rs b/crates/cccc-contracts/src/actor.rs
index 810fdc7eb..312117d4f 100644
--- a/crates/cccc-contracts/src/actor.rs
+++ b/crates/cccc-contracts/src/actor.rs
@@ -46,6 +46,22 @@ pub enum RuntimeMode {
     Acp,
 }
 
+/// Optional authoritative observation for a native terminal, applied on next start.
+#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)]
+#[serde(rename_all = "snake_case")]
+pub enum TurnProbe {
+    #[default]
+    Off,
+    Auto,
+    HermesTurnLease,
+}
+
+impl TurnProbe {
+    pub const fn is_off(&self) -> bool {
+        matches!(self, Self::Off)
+    }
+}
+
 impl RuntimeMode {
     pub const fn is_default(&self) -> bool {
         matches!(self, Self::Default)
@@ -213,6 +229,8 @@ pub struct Actor {
     pub runtime: ActorRuntime,
     #[serde(default, skip_serializing_if = "RuntimeMode::is_default")]
     pub runtime_mode: RuntimeMode,
+    #[serde(default, skip_serializing_if = "TurnProbe::is_off")]
+    pub turn_probe: TurnProbe,
     #[serde(default)]
     pub runtime_state_source: RuntimeStateSource,
     #[serde(default)]
@@ -255,6 +273,7 @@ impl Actor {
             runner: RunnerKind::default(),
             runtime: ActorRuntime::default(),
             runtime_mode: RuntimeMode::default(),
+            turn_probe: TurnProbe::Off,
             runtime_state_source: RuntimeStateSource::default(),
             internal_kind: None,
             avatar_asset_path: String::new(),
diff --git a/crates/cccc-contracts/src/lib.rs b/crates/cccc-contracts/src/lib.rs
index 488c1a730..0d546b216 100644
--- a/crates/cccc-contracts/src/lib.rs
+++ b/crates/cccc-contracts/src/lib.rs
@@ -13,7 +13,7 @@ pub mod voice_secretary;
 
 pub use actor::{
     Actor, ActorRole, ActorRuntime, ActorSubmit, GroupState, RunnerKind, RuntimeMode,
-    RuntimeStateSource,
+    RuntimeStateSource, TurnProbe,
 };
 pub use codex_voice::{AgentRuntimeSettings, CodexVoiceAnalystSettings, CodexVoiceSettings};
 pub use deepseek::{
diff --git a/crates/cccc-core/src/profiles.rs b/crates/cccc-core/src/profiles.rs
index 4923502c4..84518ac05 100644
--- a/crates/cccc-core/src/profiles.rs
+++ b/crates/cccc-core/src/profiles.rs
@@ -1,4 +1,4 @@
-use cccc_contracts::{ActorRuntime, ActorSubmit, RunnerKind, RuntimeMode, utc_now};
+use cccc_contracts::{ActorRuntime, ActorSubmit, RunnerKind, RuntimeMode, TurnProbe, utc_now};
 use serde::{Deserialize, Serialize};
 use serde_json::{Map, Value, json};
 use sha2::{Digest, Sha256};
@@ -52,6 +52,7 @@ struct SecretDoc {
 pub struct RuntimeProfileConfig {
     pub runtime: ActorRuntime,
     pub runtime_mode: RuntimeMode,
+    pub turn_probe: TurnProbe,
     pub runner: RunnerKind,
     pub command: Vec<String>,
     pub submit: ActorSubmit,
@@ -150,12 +151,14 @@ impl ProfileStore {
         let runtime_mode = parse_profile_field(&profile, "runtime_mode", RuntimeMode::default())?;
         runtime_mode.validate(runtime).map_err(io::Error::other)?;
         let runner = runtime.runner_for(runtime_mode);
+        let turn_probe = parse_profile_field(&profile, "turn_probe", TurnProbe::Off)?;
         let submit = parse_profile_field(&profile, "submit", ActorSubmit::default())?;
         let command = parse_profile_command(&profile)?;
         let environment = self.secret_values_ref(profile_id, scope, owner_id)?;
         Ok(Some(RuntimeProfileConfig {
             runtime,
             runtime_mode,
+            turn_probe,
             runner,
             command,
             submit,
@@ -234,6 +237,19 @@ impl ProfileStore {
             "runtime_mode".into(),
             serde_json::to_value(runtime_mode).map_err(io::Error::other)?,
         );
+        let turn_probe: TurnProbe = parse_profile_field(
+            &Value::Object(profile.clone()),
+            "turn_probe",
+            TurnProbe::Off,
+        )?;
+        if turn_probe.is_off() {
+            profile.remove("turn_probe");
+        } else {
+            profile.insert(
+                "turn_probe".into(),
+                serde_json::to_value(turn_probe).map_err(io::Error::other)?,
+            );
+        }
         profile.insert(
             "runner".into(),
             serde_json::to_value(runtime.runner_for(runtime_mode)).map_err(io::Error::other)?,
@@ -262,6 +278,7 @@ impl ProfileStore {
                     | "owner_id"
                     | "runtime"
                     | "runtime_mode"
+                    | "turn_probe"
                     | "runner"
                     | "command"
                     | "submit"
diff --git a/crates/cccc-daemon/Cargo.toml b/crates/cccc-daemon/Cargo.toml
index 31438bf0e..27aaee1b5 100644
--- a/crates/cccc-daemon/Cargo.toml
+++ b/crates/cccc-daemon/Cargo.toml
@@ -20,12 +20,13 @@ chrono.workspace = true
 fs2.workspace = true
 futures-util.workspace = true
 mime_guess.workspace = true
-nix = { workspace = true, features = ["hostname", "signal", "process"] }
+nix = { workspace = true, features = ["hostname", "signal", "process", "time", "feature"] }
 percent-encoding.workspace = true
 regex.workspace = true
 same-file.workspace = true
 reqwest.workspace = true
 rustls.workspace = true
+rusqlite = { version = "0.32.1", features = ["bundled", "hooks"] }
 tokio-rustls.workspace = true
 serde.workspace = true
 serde_json.workspace = true
diff --git a/crates/cccc-daemon/src/ops/actor_activity.rs b/crates/cccc-daemon/src/ops/actor_activity.rs
index e348140c6..cf0e3c5a0 100644
--- a/crates/cccc-daemon/src/ops/actor_activity.rs
+++ b/crates/cccc-daemon/src/ops/actor_activity.rs
@@ -37,6 +37,7 @@ impl Publisher {
         let mut payloads = Vec::new();
         let mut snapshot = BTreeMap::new();
         for actor in &group.actors {
+            super::turn_probe::refresh(&group.group_id, actor);
             if let Some(payload) = running_payload(store.home(), group, actor) {
                 let runner = text(&payload, "runner_effective", "pty");
                 let state = text(&payload, "effective_working_state", "waiting");
diff --git a/crates/cccc-daemon/src/ops/actor_profile_runtime.rs b/crates/cccc-daemon/src/ops/actor_profile_runtime.rs
index 55bbdc0f3..653042e16 100644
--- a/crates/cccc-daemon/src/ops/actor_profile_runtime.rs
+++ b/crates/cccc-daemon/src/ops/actor_profile_runtime.rs
@@ -7,7 +7,14 @@ use std::collections::{BTreeMap, BTreeSet};
 use crate::dispatch::OpError;
 
 // Actor capability autoload is an independent, additive baseline, even when linked.
-const CONTROLLED_FIELDS: &[&str] = &["runtime", "runtime_mode", "command", "submit", "env"];
+const CONTROLLED_FIELDS: &[&str] = &[
+    "runtime",
+    "runtime_mode",
+    "turn_probe",
+    "command",
+    "submit",
+    "env",
+];
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct CapabilityDefaults {
@@ -106,6 +113,7 @@ fn apply(home: &HomeLayout, actor: &Actor, profile_id: &str) -> Result<Actor, Op
     let mut resolved = actor.clone();
     resolved.runtime = profile.runtime;
     resolved.runtime_mode = profile.runtime_mode;
+    resolved.turn_probe = profile.turn_probe;
     resolved.runner = profile.runner;
     resolved.submit = profile.submit;
     resolved.command = profile.command;
diff --git a/crates/cccc-daemon/src/ops/actor_runtime.rs b/crates/cccc-daemon/src/ops/actor_runtime.rs
index 59fe106e8..c0c10da3e 100644
--- a/crates/cccc-daemon/src/ops/actor_runtime.rs
+++ b/crates/cccc-daemon/src/ops/actor_runtime.rs
@@ -163,7 +163,7 @@ fn start(home: &HomeLayout, group: &GroupDoc, actor: &Actor) -> Result<SessionSt
     let command = cccc_runtime::resolve_command_executable(&command, &env);
     let history =
         terminal_history::config(home, &group.group_id, &actor.id).map_err(OpError::io)?;
-    cccc_runtime::start_with_history(
+    let status = cccc_runtime::start_with_history(
         LaunchSpec {
             group_id: group.group_id.clone(),
             actor_id: actor.id.clone(),
@@ -176,7 +176,9 @@ fn start(home: &HomeLayout, group: &GroupDoc, actor: &Actor) -> Result<SessionSt
         },
         history,
     )
-    .map_err(runtime_error)
+    .map_err(runtime_error)?;
+    super::turn_probe::started(&actor, &status);
+    Ok(status)
 }
 
 pub(super) fn stop(group: &GroupDoc, actor_id: &str) -> Result<Option<SessionStatus>, OpError> {
diff --git a/crates/cccc-daemon/src/ops/mod.rs b/crates/cccc-daemon/src/ops/mod.rs
index 1e72e6a88..ce7970989 100644
--- a/crates/cccc-daemon/src/ops/mod.rs
+++ b/crates/cccc-daemon/src/ops/mod.rs
@@ -78,6 +78,7 @@ mod task_list;
 mod terminal;
 mod terminal_history_source;
 mod terminal_text;
+mod turn_probe;
 mod voice_notifications;
 pub(crate) mod voice_secretary;
 mod web_model_connectors;
diff --git a/crates/cccc-daemon/src/ops/turn_probe/hermes.rs b/crates/cccc-daemon/src/ops/turn_probe/hermes.rs
new file mode 100644
index 000000000..b1b799f8f
--- /dev/null
+++ b/crates/cccc-daemon/src/ops/turn_probe/hermes.rs
@@ -0,0 +1,169 @@
+use super::TurnObservation;
+use super::process::{Process, Processes, owned};
+use rusqlite::{Connection, OpenFlags, OptionalExtension};
+use std::collections::BTreeMap;
+use std::path::Path;
+use std::time::{Duration, Instant};
+
+#[derive(Clone, Default)]
+pub(super) struct Hermes {
+    // Learned only from a live owned lease, never from session id/cwd similarity.
+    sessions: BTreeMap<String, Process>,
+}
+
+impl Hermes {
+    pub fn observe(
+        &mut self,
+        path: &Path,
+        processes: &impl Processes,
+        root: &Process,
+        now: f64,
+    ) -> TurnObservation {
+        match self.read(path, processes, root, now) {
+            Ok(value) => value,
+            Err(why) => TurnObservation::Unknown { why },
+        }
+    }
+
+    fn read(
+        &mut self,
+        path: &Path,
+        processes: &impl Processes,
+        root: &Process,
+        now: f64,
+    ) -> Result<TurnObservation, &'static str> {
+        if !now.is_finite() {
+            return Err("clock_invalid");
+        }
+        let db = Connection::open_with_flags(
+            path,
+            OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
+        )
+        .map_err(|_| "database_unavailable")?;
+        db.busy_timeout(Duration::from_millis(25))
+            .map_err(|_| "database_unavailable")?;
+        let deadline = Instant::now() + Duration::from_millis(50);
+        db.progress_handler(1000, Some(move || Instant::now() >= deadline));
+        db.execute_batch("PRAGMA query_only=ON; PRAGMA trusted_schema=OFF; BEGIN")
+            .map_err(|_| "database_unavailable")?;
+        let versions = db
+            .prepare("SELECT version FROM schema_version LIMIT 2")
+            .and_then(|mut q| {
+                q.query_map([], |row| row.get::<_, i64>(0))?
+                    .collect::<rusqlite::Result<Vec<_>>>()
+            })
+            .map_err(|_| "schema_unavailable")?;
+        if versions != [31] {
+            return Err("schema_unsupported");
+        }
+        let mut query = db.prepare(
+            "SELECT conversation_id, holder, acquired_at, expires_at FROM session_turn_leases LIMIT 257")
+            .map_err(|_| "schema_unavailable")?;
+        let leases = query
+            .query_map([], |r| {
+                Ok((
+                    r.get::<_, String>(0)?,
+                    r.get::<_, String>(1)?,
+                    r.get::<_, f64>(2)?,
+                    r.get::<_, f64>(3)?,
+                ))
+            })
+            .and_then(|rows| rows.collect::<rusqlite::Result<Vec<_>>>())
+            .map_err(|_| "database_unavailable")?;
+        if leases.len() > 256 {
+            return Err("lease_limit");
+        }
+        let mut working_since: Option<f64> = None;
+        let mut learned = self.sessions.clone();
+        for (session, holder, acquired, expiry) in leases {
+            if !acquired.is_finite() || !expiry.is_finite() || expiry < acquired {
+                return Err("lease_invalid");
+            }
+            if expiry <= now {
+                continue;
+            }
+            let (pid, namespace, cli) = parse_holder(&holder).ok_or("holder_unsupported")?;
+            // A foreign namespace cannot establish membership in this process tree.
+            if namespace != root.namespace {
+                return Err("namespace_mismatch");
+            }
+            let Some(process) = owned(processes, root, pid)? else {
+                continue;
+            };
+            if !cli {
+                return Err("platform_unsupported");
+            }
+            if acquired < process.born_before || acquired > now {
+                return Err("holder_generation_mismatch");
+            }
+            if learned
+                .get(&session)
+                .is_some_and(|old| old.pid != process.pid || old.start != process.start)
+            {
+                return Err("session_binding_changed");
+            }
+            learned.insert(session, process);
+            working_since = Some(working_since.map_or(acquired, |old| old.min(acquired)));
+        }
+        if learned.len() > 256 {
+            return Err("session_limit");
+        }
+        // Recheck every provider identity. A dead/reused holder is never proof of idle.
+        let mut live_sessions = 0;
+        for (session, bound) in &learned {
+            let current = owned(processes, root, bound.pid)?.ok_or("holder_left_tree")?;
+            if current.start != bound.start || current.namespace != bound.namespace {
+                return Err("holder_generation_mismatch");
+            }
+            let ended = db
+                .query_row(
+                    "SELECT ended_at FROM sessions WHERE id=?1",
+                    [session],
+                    |row| row.get::<_, Option<f64>>(0),
+                )
+                .optional()
+                .map_err(|_| "session_unavailable")?;
+            if ended == Some(None) {
+                live_sessions += 1;
+            }
+        }
+        let current_root = processes.read(root.pid)?;
+        if current_root.start != root.start || current_root.namespace != root.namespace {
+            return Err("generation_changed");
+        }
+        db.execute_batch("ROLLBACK")
+            .map_err(|_| "database_unavailable")?;
+        if Instant::now() >= deadline {
+            return Err("probe_budget");
+        }
+        self.sessions = learned;
+        if let Some(since) = working_since {
+            Ok(TurnObservation::Working {
+                since: timestamp(since),
+            })
+        } else if live_sessions > 0 {
+            Ok(TurnObservation::Idle)
+        } else {
+            Err("session_not_bound")
+        }
+    }
+}
+
+fn timestamp(seconds: f64) -> Option<String> {
+    chrono::DateTime::from_timestamp_millis((seconds * 1000.0) as i64)
+        .map(|time| time.to_rfc3339_opts(chrono::SecondsFormat::Millis, true))
+}
+
+fn parse_holder(holder: &str) -> Option<(u32, u64, bool)> {
+    let (pid, rest) = holder.strip_prefix("pid=")?.split_once(":pidns=")?;
+    let (namespace, turn) = rest.split_once(":turn=")?;
+    let (turn, platform) = turn.rsplit_once(":platform=")?;
+    if turn.is_empty() || pid.is_empty() || namespace.is_empty() {
+        return None;
+    }
+    Some((
+        pid.parse().ok()?,
+        namespace.parse().ok()?,
+        platform == "cli",
+    ))
+}
diff --git a/crates/cccc-daemon/src/ops/turn_probe/mod.rs b/crates/cccc-daemon/src/ops/turn_probe/mod.rs
new file mode 100644
index 000000000..02f7664f8
--- /dev/null
+++ b/crates/cccc-daemon/src/ops/turn_probe/mod.rs
@@ -0,0 +1,157 @@
+//! Provider-owned turn state for a single PTY generation. No output inference.
+mod hermes;
+mod process;
+
+use cccc_contracts::{Actor, ActorRuntime, TurnProbe};
+use cccc_runtime::SessionStatus;
+use process::{ProcFs, Process, Processes};
+use std::collections::BTreeMap;
+use std::sync::{Mutex, OnceLock};
+use std::time::{SystemTime, UNIX_EPOCH};
+
+#[derive(Debug, Clone, PartialEq)]
+pub(super) enum TurnObservation {
+    Working { since: Option<String> },
+    Idle,
+    Unknown { why: &'static str },
+}
+
+#[derive(Clone)]
+struct Entry {
+    started_at: String,
+    root: Process,
+    hermes: hermes::Hermes,
+    observation: TurnObservation,
+}
+
+type Key = (String, String);
+fn entries() -> &'static Mutex<BTreeMap<Key, Entry>> {
+    static ENTRIES: OnceLock<Mutex<BTreeMap<Key, Entry>>> = OnceLock::new();
+    ENTRIES.get_or_init(Mutex::default)
+}
+
+/// Called once at a successful launch: saved configuration changes wait for restart.
+pub(super) fn started(actor: &Actor, status: &SessionStatus) {
+    let key = (status.group_id.clone(), status.actor_id.clone());
+    let Ok(mut entries) = entries().lock() else {
+        return;
+    };
+    entries.remove(&key);
+    if actor.runtime != ActorRuntime::Hermes || actor.turn_probe == TurnProbe::Off {
+        return;
+    }
+    let Some(pid) = status.pid.filter(|_| status.running) else {
+        return;
+    };
+    let Ok(root) = ProcFs.read(pid) else {
+        return;
+    };
+    entries.insert(
+        key,
+        Entry {
+            started_at: status.started_at.clone(),
+            root,
+            hermes: hermes::Hermes::default(),
+            observation: TurnObservation::Unknown {
+                why: "not_observed",
+            },
+        },
+    );
+}
+
+/// Runs only on the ordinary activity publisher tick, never a mutating request.
+pub(super) fn refresh(group: &str, actor: &Actor) {
+    let key = (group.to_owned(), actor.id.clone());
+    let entry = entries()
+        .lock()
+        .ok()
+        .and_then(|entries| entries.get(&key).cloned());
+    let Some(mut entry) = entry else {
+        return;
+    };
+    let Some(status) = generation(group, &actor.id, &entry) else {
+        if let Ok(mut entries) = entries().lock() {
+            entries.remove(&key);
+        }
+        return;
+    };
+    // Registered managed authority always wins, including late registration.
+    entry.observation = if super::local_headless::running(group, &actor.id)
+        || super::local_headless::status(group, &actor.id).is_some()
+    {
+        TurnObservation::Unknown {
+            why: "managed_session",
+        }
+    } else {
+        observe(&mut entry, group, &actor.id)
+    };
+    if let Ok(mut entries) = entries().lock()
+        && let Some(current) = entries.get_mut(&key)
+        && current.started_at == status.started_at
+        && current.root == entry.root
+    {
+        *current = entry;
+    }
+}
+
+fn observe(entry: &mut Entry, group: &str, actor: &str) -> TurnObservation {
+    let result = (|| {
+        let home = process::provider_home(entry.root.pid, group, actor)?;
+        let now = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .map_err(|_| "clock_unavailable")?
+            .as_secs_f64();
+        Ok(entry
+            .hermes
+            .observe(&home.join("state.db"), &ProcFs, &entry.root, now))
+    })();
+    result.unwrap_or_else(|why| TurnObservation::Unknown { why })
+}
+
+fn generation(group: &str, actor: &str, entry: &Entry) -> Option<SessionStatus> {
+    let status = cccc_runtime::status(group, actor).ok()?;
+    if !status.running
+        || status.pid != Some(entry.root.pid)
+        || status.started_at != entry.started_at
+    {
+        return None;
+    }
+    let root = ProcFs.read(entry.root.pid).ok()?;
+    (root.start == entry.root.start && root.namespace == entry.root.namespace).then_some(status)
+}
+
+/// Cached state only. A new/reused PTY identity cannot inherit an observation.
+pub(super) fn cached(group: &str, actor: &str) -> Option<TurnObservation> {
+    let entry = entries()
+        .lock()
+        .ok()?
+        .get(&(group.into(), actor.into()))
+        .cloned()?;
+    generation(group, actor, &entry)?;
+    match entry.observation {
+        TurnObservation::Unknown { .. } => None,
+        observation => Some(observation),
+    }
+}
+
+pub(super) fn fields(
+    observation: TurnObservation,
+) -> (String, String, Option<String>, Option<String>) {
+    let (state, since) = match observation {
+        TurnObservation::Working { since } => ("working", since),
+        TurnObservation::Idle => ("idle", Some(cccc_contracts::utc_now())),
+        TurnObservation::Unknown { .. } => {
+            unreachable!("Unknown falls through before field construction")
+        }
+    };
+    // A provider conversation/turn id is not a CCCC task id.
+    (
+        state.into(),
+        "turn_probe:hermes_turn_lease".into(),
+        since,
+        None,
+    )
+}
+
+#[cfg(all(test, target_os = "linux"))]
+mod tests;
diff --git a/crates/cccc-daemon/src/ops/turn_probe/process.rs b/crates/cccc-daemon/src/ops/turn_probe/process.rs
new file mode 100644
index 000000000..9e596fcdb
--- /dev/null
+++ b/crates/cccc-daemon/src/ops/turn_probe/process.rs
@@ -0,0 +1,135 @@
+use std::path::PathBuf;
+
+#[derive(Debug, Clone, PartialEq)]
+pub(super) struct Process {
+    pub pid: u32,
+    pub parent: u32,
+    pub start: u64,
+    pub namespace: u64,
+    // Upper bound on process birth, including one kernel clock tick.
+    pub born_before: f64,
+}
+
+pub(super) trait Processes {
+    fn read(&self, pid: u32) -> Result<Process, &'static str>;
+}
+
+pub(super) struct ProcFs;
+
+#[cfg(target_os = "linux")]
+impl Processes for ProcFs {
+    fn read(&self, pid: u32) -> Result<Process, &'static str> {
+        use nix::time::{ClockId, clock_gettime};
+        use nix::unistd::{SysconfVar, sysconf};
+        let base = PathBuf::from(format!("/proc/{pid}"));
+        let stat = std::fs::read_to_string(base.join("stat")).map_err(|_| "process_unavailable")?;
+        let (_, fields) = stat.rsplit_once(')').ok_or("process_invalid")?;
+        let fields = fields.split_whitespace().collect::<Vec<_>>();
+        if fields.len() < 20 || matches!(fields[0], "Z" | "X") {
+            return Err("process_not_live");
+        }
+        let parent = fields[1].parse().map_err(|_| "process_invalid")?;
+        let start: u64 = fields[19].parse().map_err(|_| "process_invalid")?;
+        let namespace =
+            std::fs::read_link(base.join("ns/pid")).map_err(|_| "namespace_unavailable")?;
+        let namespace = namespace
+            .to_str()
+            .and_then(|s| s.strip_prefix("pid:["))
+            .and_then(|s| s.strip_suffix(']'))
+            .and_then(|s| s.parse().ok())
+            .ok_or("namespace_invalid")?;
+        let hz = sysconf(SysconfVar::CLK_TCK)
+            .map_err(|_| "clock_unavailable")?
+            .filter(|hz| *hz > 0)
+            .ok_or("clock_unavailable")? as f64;
+        let boot = clock_gettime(ClockId::CLOCK_BOOTTIME).map_err(|_| "clock_unavailable")?;
+        let real = clock_gettime(ClockId::CLOCK_REALTIME).map_err(|_| "clock_unavailable")?;
+        let seconds = |ts: nix::sys::time::TimeSpec| ts.tv_sec() as f64 + ts.tv_nsec() as f64 / 1e9;
+        Ok(Process {
+            pid,
+            parent,
+            start,
+            namespace,
+            born_before: seconds(real) - seconds(boot) + (start as f64 + 1.0) / hz,
+        })
+    }
+}
+
+#[cfg(not(target_os = "linux"))]
+impl Processes for ProcFs {
+    fn read(&self, _pid: u32) -> Result<Process, &'static str> {
+        Err("process_identity_unsupported")
+    }
+}
+
+pub(super) fn owned(
+    processes: &impl Processes,
+    root: &Process,
+    holder: u32,
+) -> Result<Option<Process>, &'static str> {
+    let original = processes.read(holder)?;
+    let mut current = original.clone();
+    // A bounded ancestry walk, never a scan of the machine's process table.
+    for _ in 0..64 {
+        if current.pid == root.pid {
+            return if current.start == root.start && current.namespace == root.namespace {
+                Ok(Some(original))
+            } else {
+                Err("generation_changed")
+            };
+        }
+        if current.parent <= 1 {
+            return Ok(None);
+        }
+        current = processes.read(current.parent)?;
+    }
+    Err("process_tree_limit")
+}
+
+#[cfg(target_os = "linux")]
+pub(super) fn provider_home(pid: u32, group: &str, actor: &str) -> Result<PathBuf, &'static str> {
+    use std::io::Read;
+    let file = std::fs::File::open(format!("/proc/{pid}/environ"))
+        .map_err(|_| "environment_unavailable")?;
+    let mut bytes = Vec::new();
+    file.take(131_073)
+        .read_to_end(&mut bytes)
+        .map_err(|_| "environment_unavailable")?;
+    if bytes.len() > 131_072 {
+        return Err("environment_limit");
+    }
+    // Only identity/location keys escape this function; never log environment data.
+    let find = |name: &[u8]| {
+        bytes.split(|b| *b == 0).find_map(|entry| {
+            let (key, value) = entry.split_at(entry.iter().position(|b| *b == b'=')?);
+            (key == name)
+                .then(|| std::str::from_utf8(&value[1..]).ok())
+                .flatten()
+        })
+    };
+    if find(b"CCCC_GROUP_ID") != Some(group) || find(b"CCCC_ACTOR_ID") != Some(actor) {
+        return Err("actor_identity_mismatch");
+    }
+    // Named profiles can relocate the database. They require a separately verified adapter.
+    if find(b"HERMES_PROFILE").is_some_and(|s| !s.is_empty() && s != "default") {
+        return Err("profile_unsupported");
+    }
+    let home = if let Some(home) = find(b"HERMES_HOME").filter(|s| !s.is_empty()) {
+        PathBuf::from(home)
+    } else {
+        PathBuf::from(find(b"HOME").ok_or("home_unavailable")?).join(".hermes")
+    };
+    if !home.is_absolute() {
+        return Err("home_not_absolute");
+    }
+    Ok(home)
+}
+
+#[cfg(not(target_os = "linux"))]
+pub(super) fn provider_home(
+    _pid: u32,
+    _group: &str,
+    _actor: &str,
+) -> Result<PathBuf, &'static str> {
+    Err("process_identity_unsupported")
+}
diff --git a/crates/cccc-daemon/src/ops/turn_probe/tests.rs b/crates/cccc-daemon/src/ops/turn_probe/tests.rs
new file mode 100644
index 000000000..07db81ae5
--- /dev/null
+++ b/crates/cccc-daemon/src/ops/turn_probe/tests.rs
@@ -0,0 +1,283 @@
+use super::*;
+use process::Processes;
+use rusqlite::Connection;
+use serde_json::Value;
+use std::path::{Path, PathBuf};
+
+fn fixture(name: &str) -> PathBuf {
+    Path::new(env!("CARGO_MANIFEST_DIR"))
+        .join("tests/fixtures/hermes-turn-probe")
+        .join(name)
+}
+
+struct Capture {
+    metadata: Value,
+    processes: BTreeMap<u32, Process>,
+}
+
+impl Capture {
+    fn load() -> Self {
+        let metadata: Value =
+            serde_json::from_slice(&std::fs::read(fixture("capture.json")).expect("capture"))
+                .expect("metadata");
+        let processes = metadata["processes"]
+            .as_object()
+            .expect("processes")
+            .values()
+            .map(|p| {
+                let process = Process {
+                    pid: p["pid"].as_u64().expect("pid") as u32,
+                    parent: p["ppid"].as_u64().expect("parent") as u32,
+                    start: p["start_ticks"].as_u64().expect("start"),
+                    namespace: p["pid_namespace"].as_u64().expect("namespace"),
+                    born_before: p["born_before"].as_f64().expect("birth"),
+                };
+                (process.pid, process)
+            })
+            .collect();
+        Self {
+            metadata,
+            processes,
+        }
+    }
+
+    fn root(&self) -> Process {
+        self.read(self.metadata["target"]["root_pid"].as_u64().expect("root") as u32)
+            .expect("root process")
+    }
+
+    fn now(&self) -> f64 {
+        self.metadata["captured_at_unix"]
+            .as_f64()
+            .expect("capture time")
+    }
+}
+
+impl Processes for Capture {
+    fn read(&self, pid: u32) -> Result<Process, &'static str> {
+        self.processes
+            .get(&pid)
+            .cloned()
+            .ok_or("process_unavailable")
+    }
+}
+
+fn copy_database() -> (tempfile::TempDir, PathBuf) {
+    let temp = tempfile::tempdir().expect("tempdir");
+    let path = temp.path().join("state.db");
+    std::fs::copy(fixture("during.sqlite3"), &path).expect("copy real metadata");
+    (temp, path)
+}
+
+#[test]
+fn real_lease_is_working_then_provider_expiry_is_idle_only_after_binding() {
+    let capture = Capture::load();
+    let root = capture.root();
+    let path = fixture("during.sqlite3");
+    let before = std::fs::read(&path).expect("before");
+    let mut probe = hermes::Hermes::default();
+    assert!(
+        matches!(
+            probe.observe(&path, &capture, &root, capture.now()),
+            TurnObservation::Working { .. }
+        ),
+        "a real owned unexpired Hermes lease must be working"
+    );
+    let last_expiry: f64 = Connection::open(&path)
+        .expect("fixture")
+        .query_row(
+            "SELECT MAX(expires_at) FROM session_turn_leases",
+            [],
+            |row| row.get(0),
+        )
+        .expect("expiry");
+    // Replay at the provider's own expiry; no invented lease or session rows.
+    assert_eq!(
+        probe.observe(&path, &capture, &root, last_expiry + 1.0),
+        TurnObservation::Idle
+    );
+    assert!(
+        matches!(
+            hermes::Hermes::default().observe(&path, &capture, &root, last_expiry + 1.0),
+            TurnObservation::Unknown { .. }
+        ),
+        "no lease alone cannot bind an idle session"
+    );
+    assert_eq!(
+        std::fs::read(&path).expect("after"),
+        before,
+        "probe must not write provider bytes"
+    );
+}
+
+#[test]
+fn foreign_lease_cannot_establish_an_actor_binding() {
+    let capture = Capture::load();
+    let root = capture.root();
+    let (_temp, path) = copy_database();
+    // Negative control on a copy of the capture: remove this Actor's real lease.
+    Connection::open(&path)
+        .expect("copy")
+        .execute(
+            "DELETE FROM session_turn_leases WHERE conversation_id=?1",
+            [capture.metadata["target"]["conversation_id"]
+                .as_str()
+                .expect("session")],
+        )
+        .expect("remove owned lease");
+    assert!(matches!(
+        hermes::Hermes::default().observe(&path, &capture, &root, capture.now()),
+        TurnObservation::Unknown { .. }
+    ));
+}
+
+#[test]
+fn unavailable_schema_lock_and_missing_session_fail_closed() {
+    let capture = Capture::load();
+    for alteration in [
+        "DROP TABLE session_turn_leases",
+        "UPDATE schema_version SET version=32",
+    ] {
+        let (_temp, path) = copy_database();
+        Connection::open(&path)
+            .expect("copy")
+            .execute_batch(alteration)
+            .expect("negative control");
+        assert!(matches!(
+            hermes::Hermes::default().observe(&path, &capture, &capture.root(), capture.now()),
+            TurnObservation::Unknown { .. }
+        ));
+    }
+    let (_temp, path) = copy_database();
+    let writer = Connection::open(&path).expect("copy");
+    writer
+        .execute_batch("PRAGMA journal_mode=DELETE; BEGIN EXCLUSIVE")
+        .expect("lock");
+    let start = std::time::Instant::now();
+    assert!(matches!(
+        hermes::Hermes::default().observe(&path, &capture, &capture.root(), capture.now()),
+        TurnObservation::Unknown { .. }
+    ));
+    assert!(start.elapsed() < std::time::Duration::from_secs(1));
+    writer.execute_batch("ROLLBACK").expect("unlock");
+    let mut probe = hermes::Hermes::default();
+    assert!(matches!(
+        probe.observe(&path, &capture, &capture.root(), capture.now()),
+        TurnObservation::Working { .. }
+    ));
+    writer
+        .execute_batch("DELETE FROM sessions")
+        .expect("missing sessions control");
+    let later: f64 = writer
+        .query_row(
+            "SELECT MAX(expires_at)+1 FROM session_turn_leases",
+            [],
+            |row| row.get(0),
+        )
+        .expect("expiry");
+    assert!(matches!(
+        probe.observe(&path, &capture, &capture.root(), later),
+        TurnObservation::Unknown { .. }
+    ));
+}
+
+#[test]
+fn pid_reuse_namespace_and_generation_changes_never_reuse_authority() {
+    let mut capture = Capture::load();
+    let old_root = capture.root();
+    let path = fixture("during.sqlite3");
+    let mut probe = hermes::Hermes::default();
+    assert!(matches!(
+        probe.observe(&path, &capture, &old_root, capture.now()),
+        TurnObservation::Working { .. }
+    ));
+    capture
+        .processes
+        .get_mut(&old_root.pid)
+        .expect("root")
+        .start += 1;
+    assert!(matches!(
+        probe.observe(&path, &capture, &old_root, capture.now()),
+        TurnObservation::Unknown { .. }
+    ));
+    let mut reused = capture.root();
+    reused.born_before = capture.now() + 1.0;
+    capture.processes.insert(reused.pid, reused.clone());
+    assert!(matches!(
+        hermes::Hermes::default().observe(&path, &capture, &reused, capture.now()),
+        TurnObservation::Unknown { .. }
+    ));
+    reused.namespace += 1;
+    capture.processes.insert(reused.pid, reused.clone());
+    assert!(matches!(
+        hermes::Hermes::default().observe(&path, &capture, &reused, capture.now()),
+        TurnObservation::Unknown { .. }
+    ));
+}
+
+#[test]
+fn defaults_are_absent_and_provider_conversation_is_not_a_task_id() {
+    let actor = Actor::new("peer");
+    let mut value = serde_json::to_value(&actor).expect("serialize");
+    assert!(value.get("turn_probe").is_none());
+    value["turn_probe"] = serde_json::json!("auto");
+    let enabled: Actor = serde_json::from_value(value).expect("opt in");
+    assert_eq!(enabled.turn_probe, TurnProbe::Auto);
+    let (_, reason, _, task) = fields(TurnObservation::Working { since: None });
+    assert_eq!(reason, "turn_probe:hermes_turn_lease");
+    assert_eq!(task, None);
+}
+
+#[test]
+fn captured_hermes_lease_populates_existing_working_fields() {
+    let capture = Capture::load();
+    let observation = hermes::Hermes::default().observe(
+        &fixture("during.sqlite3"),
+        &capture,
+        &capture.root(),
+        capture.now(),
+    );
+    assert!(matches!(observation, TurnObservation::Working { .. }));
+    let temp = tempfile::tempdir().expect("tempdir");
+    let home = cccc_core::HomeLayout::from_path(temp.path()).expect("home");
+    let mut actor = Actor::new("peer");
+    actor.runtime = ActorRuntime::Hermes;
+    actor.turn_probe = TurnProbe::Auto;
+    let state = super::super::working_state::fields_with_probe(
+        &home,
+        &actor,
+        "fixture",
+        true,
+        "pty",
+        || Some(observation),
+    );
+    assert_eq!(
+        state["effective_working_state"], "working",
+        "a captured owned Hermes turn must replace the native-terminal unknown fallback"
+    );
+    assert_eq!(
+        state["effective_working_reason"],
+        "turn_probe:hermes_turn_lease"
+    );
+    assert_eq!(state["effective_active_task_id"], serde_json::Value::Null);
+    assert!(!state.contains_key("turn_probe"));
+}
+
+#[test]
+fn stopped_and_headless_states_never_consult_the_probe() {
+    let temp = tempfile::tempdir().expect("tempdir");
+    let home = cccc_core::HomeLayout::from_path(temp.path()).expect("home");
+    let mut actor = Actor::new("peer");
+    actor.runtime = ActorRuntime::Custom;
+    for (running, runner, expected) in [(false, "pty", "stopped"), (true, "headless", "idle")] {
+        let state = super::super::working_state::fields_with_probe(
+            &home,
+            &actor,
+            "fixture",
+            running,
+            runner,
+            || panic!("probe must not run on this surface"),
+        );
+        assert_eq!(state["effective_working_state"], expected);
+    }
+}
diff --git a/crates/cccc-daemon/src/ops/working_state.rs b/crates/cccc-daemon/src/ops/working_state.rs
index f04590470..599c82764 100644
--- a/crates/cccc-daemon/src/ops/working_state.rs
+++ b/crates/cccc-daemon/src/ops/working_state.rs
@@ -30,11 +30,28 @@ pub(super) fn fields(
     running: bool,
     runner_effective: &str,
 ) -> Map<String, Value> {
-    let managed_session = super::local_headless::running(group_id, &actor.id)
-        || super::local_headless::uses_managed_session(actor);
+    fields_with_probe(home, actor, group_id, running, runner_effective, || {
+        super::turn_probe::cached(group_id, &actor.id)
+    })
+}
+
+pub(super) fn fields_with_probe(
+    home: &HomeLayout,
+    actor: &Actor,
+    group_id: &str,
+    running: bool,
+    runner_effective: &str,
+    probe: impl FnOnce() -> Option<super::turn_probe::TurnObservation>,
+) -> Map<String, Value> {
+    let registered_managed = super::local_headless::running(group_id, &actor.id);
+    let managed_session = registered_managed || super::local_headless::uses_managed_session(actor);
     let local_state = running
         .then(|| super::local_headless::status(group_id, &actor.id))
         .flatten();
+    let probe =
+        (running && !registered_managed && local_state.is_none() && runner_effective == "pty")
+            .then(probe)
+            .flatten();
     let (state, reason, updated_at, active_task_id) = if !running {
         (
             "stopped".to_owned(),
@@ -55,6 +72,8 @@ pub(super) fn fields(
             Some(local_state.updated_at),
             local_state.task_id,
         )
+    } else if let Some(probe) = probe {
+        super::turn_probe::fields(probe)
     } else if managed_session {
         (
             "waiting".to_owned(),
diff --git a/crates/cccc-daemon/tests/fixtures/hermes-turn-probe/README.md b/crates/cccc-daemon/tests/fixtures/hermes-turn-probe/README.md
new file mode 100644
index 000000000..b0a395679
--- /dev/null
+++ b/crates/cccc-daemon/tests/fixtures/hermes-turn-probe/README.md
@@ -0,0 +1,22 @@
+# Hermes turn-lease capture
+
+`during.sqlite3` is a metadata-only projection captured read-only from a real
+Hermes CLI state database on 2026-10-22 (schema version 31). It retains the
+original DDL for `schema_version`, `session_turn_leases` and `sessions`, all 30
+observed lease rows, and the 30 corresponding session rows. Only `id`, `source`,
+`started_at` and `ended_at` were selected from sessions; all remaining columns
+are NULL or the schema's defaults. No message tables, bodies, prompts,
+credentials, environment values, actor/group identifiers or filesystem paths
+are included. No provider-state row was generated for these tests.
+
+`capture.json` records the capture clock, original kernel PID/start/namespace
+metadata and ancestry for the live CLI holders, and the database digest. Numeric
+process identities, timestamps and provider session identifiers are unmodified.
+The process snapshot is replayed through a read-only test view: tests never
+contact the original processes or provider installation.
+
+Expiry tests advance the replay clock past the captured provider expiry; this
+does not create a provider record or claim that a real turn ended at that time.
+Negative controls delete a table/row or change a schema version on disposable
+copies of this capture. They are deliberately damaged captures, not additional
+provider observations. Database contention uses an exclusive lock on a copy.
diff --git a/crates/cccc-daemon/tests/fixtures/hermes-turn-probe/capture.json b/crates/cccc-daemon/tests/fixtures/hermes-turn-probe/capture.json
new file mode 100644
index 000000000..9881efcff
--- /dev/null
+++ b/crates/cccc-daemon/tests/fixtures/hermes-turn-probe/capture.json
@@ -0,0 +1,51 @@
+{
+  "schema_version": 31,
+  "captured_at_unix": 1792697799.0665095,
+  "clock_ticks_per_second": 100,
+  "processes": {
+    "2790164": {
+      "pid": 2790164,
+      "ppid": 2789094,
+      "start_ticks": 273132507,
+      "pid_namespace": 4026531836,
+      "born_before": 1792695985.5952091
+    },
+    "2790240": {
+      "pid": 2790240,
+      "ppid": 2789094,
+      "start_ticks": 273132518,
+      "pid_namespace": 4026531836,
+      "born_before": 1792695985.7052088
+    },
+    "2796583": {
+      "pid": 2796583,
+      "ppid": 2789094,
+      "start_ticks": 273134480,
+      "pid_namespace": 4026531836,
+      "born_before": 1792696005.325209
+    },
+    "2950773": {
+      "pid": 2950773,
+      "ppid": 2789094,
+      "start_ticks": 273270134,
+      "pid_namespace": 4026531836,
+      "born_before": 1792697361.8652086
+    },
+    "2789094": {
+      "pid": 2789094,
+      "ppid": 1,
+      "start_ticks": 273131084,
+      "pid_namespace": 4026531836,
+      "born_before": 1792695971.3652089
+    }
+  },
+  "target": {
+    "root_pid": 2790164,
+    "holder_pid": 2790164,
+    "conversation_id": "20261022_190630_6c7743"
+  },
+  "lease_count": 30,
+  "session_count": 30,
+  "database_sha256": "eaea613770e45a788b0c5aea65aff1718be8a35bd9b5d789150336a1d1792a50",
+  "redaction": "Only id, source, started_at and ended_at session columns were selected. All other session columns are NULL or schema defaults. No message table, bodies, prompts, environment values, actor/group identifiers or filesystem paths are included. Lease rows and process identity numbers are unmodified."
+}
diff --git a/crates/cccc-daemon/tests/fixtures/hermes-turn-probe/during.sqlite3 b/crates/cccc-daemon/tests/fixtures/hermes-turn-probe/during.sqlite3
new file mode 100644
index 0000000000000000000000000000000000000000..500a591ba7a2925cd63a482d659bfc06b238ca2f
GIT binary patch
literal 32768
zcmeHPd2Afj8DF1kJKkMyATf#Kumlle6gP7pRwD?B11=$e<Ip5PX6Ljg+3eao<0K|+
z>9jPQGPL4~5CtmH0u?2HRH;-IQABCDuQmlyRG~r(6s4l2^Z<(b&AuJ`&Fsc&q?LNT
zRh;#H-}~P8d%y3U^F8mDOE%fAnXwARJ*t~wV_mU^hS)ioOe_{_hMzY030_U`i!Tda
z4Z$ySyf?=_xA(e4>YK5a)|Ii;p`@GoLhH)p`Z+9!iZDPJAPf)&{(~5}d1-UJySKOD
zp17-OInyy6$1dcZ;Js!2@Zh?U!OY0Ib2klU0%>MdM+RQDkr^4>Hj=q`c<91)!`m|#
z3~patQaFXlqHZGP=8Hx$o40P-gfx?mSyak)<+8}5vfG?086c@Xrn=#BC5byVj;cka
z+_9pm8sTz=?TqKtDJ8G&!HX7(_Nbj#b~%MS=EmGVZWpH%%g(Dg`%2Tu3~e46Ja2F~
zvte-Kx~-c=G8A&Or(l>l%&zoV(F=LY9>v7Yl;fIvl<{I=&$xRcxiY3YW0-PWEt+{(
zDRUccjN__B*EAGZz#$HsGGQ{!yfLdoKtZGcC*fA^0gBa86Yx&v-4iv@EfjK!uIAvs
zxr*8O@kv*43%kv{bE2vWlWy%ox~h+vK31bfZK&YBB5Y=Altc};^P@G&>HAQ}MU;)&
zuA&vyyp9LO7Z%O&f>JCL@JRadJ*KM~s;eraX5K8SE{vBMCV~J(JCJ7Qay~HV8hdTS
zEMoBwRGR9T%48AmW`WA@ZkUDXxHjzRAohB}ag|BODD@81994HGRX>Oonvw{5lsoBQ
zl_BZufJL>87j50gyVorG^8r@gwcQ*Jru13CMm(w@r-FaLV(+!xsi6N>QyHeC7wz#9
zi~ej<t&Go_nTHVJj@5B{9{qJp%^L;FQv7z~)`1-63fJEvV=W9+X0cen1Es^LE|zC2
zOSN;8MSn;Yaz<faUV)Ki=ddtTD?E3Fxtylzy8$S4<J~no6kflbH!aKbxsHzU6Lm-2
zx@hXaF3f@kj78|ACe{id-m-IMbTV~exH#j5qU-l(zW~Nw(cEXlBov-RPJ}}(>SKU<
zcHTWv!ErmEpGUDV+c3=9qp{REOH)sIVnv;<GzI$WMtOW>R_IX3A7`k0uRzTNG1e1&
z?s5f~#Zcf@%oVxJ3U$&gC_$^|EL&PgH(oS6ICS1-e`U?As#qxcAW|C#hX*&WAKWst
zHm|acKGd9$acF5OEkR*xeC2Xh`Abw^N8g&3*6#Ja4KWC)x!;+{`BSjpg3{+;#c~un
z*cY^VTN>ltEZgw43-R5A;!YOxN)DDK7(L|ygo&D}?J;nh@^%B3TYow#O}N1_S-JNZ
zg8^BdUn9$<s!vSX#n}}zqAu+POedJDX0#oBgH5g7+u0g4;DBcs!G$x2ZnGiY-PhOP
zwIGgQ8Qh}=6Ri{a)EH(aq$`sWOwhr`B^W>*eF@lQHl#j|!9VgM3=jqg1B3y>0AYYI
zKo}ql5C#YXgaN_;Vc>s{fu@GWhErRhUH<tWUTx$V!2e#5Ns9;rgaN_;VSq3|7$6J~
z1_%R$0m1-bfG`k?#d|t{hUJ~F#8SUcJ(D_|x;1rWN=a=<^`(-@50bAYA4z^Ec}+5x
zyd)_kdy=ul?-EZZ9!}hnm`dzStWT^;Bs%{L#^gm9APf)&2m^!x!T@1_FhCd}3=jtX
zeGDvZk{cL`5on5G6k29vMo>7KXL){kL`4altVn{+%35!Qilr2jwscBbR!t@7qR6p5
zVU@sgw4_Lq&Wgs;u!?6yhEr%!Vp(cQSj8|5%__3Uh#Y@PSVd7XFDpDKbg4U}q9jh>
zMTORA+M*XnR5aA3$k1Y1TNF`I949KAX;6Z3a#+RD6iX|TL1_|yQZ*INYPuyX467Jc
zq9{evSdmrJ5f#Nlg}TgG4A)huq7+LMEs9HpRWvI?yi}T(RXrJ2Q4~+nifqZIY9*?v
zL|UT^v9p>=&^d|G7hoz{VQ7wFD23G+gXcOzDq7}f0MD`(t8wiWDn?cyCJb-3MN|~S
zK`2>XRP}gRCD1&>f=ffN#MZEi<2YVYI73inqa~uEB$ihgjx#B)xl%<bqG3sf+7wlZ
zJf)~Wugo_tX_DK^!+}<4juL5E5j0CsRsa6~^5%(H>f5Q$CtpkMPcBaUFfrWue&=-O
z>IF|M*xB*7j$1p}_LtfV?TNPUw++SLj(;P5dh27Ym$m$(<@Of7`4?bJUW5U{0AXN0
z3}ly|EuT~_h`wDU4p)!R3&i*rjs&0De4$^W^j3XY*K_u|*Nz;DC+|9*?X8~&Y<GY>
z+;PBp@%iksdU}}IWFQZ>zrJ^MZ?>n-9?)K;?BRu@?bF$%b@PC4d60+cqxY;_V=Osa
z?x|0t@tsp4Z#rWI7ovUrE0<rkE_+Jd+JSGvP&=+3z3SQ9vfXv_fbZpyhy7Q)v~Npx
zalJf*wuH!oW5l~J$u6p&2W)4FJWM=r=MV17o?Jf<*ao%iVcWHT`a|}lx_Q9&xyXZ&
zxPQUh*@boXfVRuXgFa+m@Mtz&HxKx(8+o|mi6@4>l<lgY2W<0>JY0T`asMgVR9!s;
zX8`13=R3zf(6h<9dBBen$ioi#y0?;M;%vFME^`xwpGcfZ4ZcDn?@vx6&rHfKW@nw<
zpwmsr&BVHw%>{LKgHB2zH}3Xh*^c^zC35OQVcFdO^pCb@+w0{aba+D^Hg@P2AIrAY
z)5FXeFz}#Vd{#DIKM&YZ5_#AlZh8F7Y-`;-;3rb#Vc?aYZ9OB~Qa2Cy!4`R7Kf3M7
zCE4csdBD!c$OHZ6lC}@BO?CAU9J9-9czJb>$u`!}!|Z7td0?8aK7T0NP)`pthyJpM
zWj%L)h2;Oss{nZ;3=jqg1B3y>0AYYIKo}ql5C;BR8NhO(B>zwH|L7SV$^XN3m@4`I
zp4eC{m4n;<8xn^T%Q~-J@Zkcr<GBvG{lWI0wj1Mrj~lHowyte?pk-0B*L1vTN8`^M
znTDHVpMcrCUtYSWRrbCZk7MaON`TZCN=%L0##%;{DO%tL;P<?<mV?_naCbyvg#rIt
zAlFv68#sZOqiGXzA;v)P&8=(EGmkqyxyCu>CkhW=f7f@t^wKCMJeB2Rq9u`$D7hvk
z+z!JO64Qodu`I_&F-;%6#rqy$x@jT2d#5e|Ob*Typ8P6dVkt%vg_@Xfy8)AtG>$Tu
zeu1HdexS$d*K|Yd=S)uGRf^>WSsyUrZjY(SY82u1ubjJ)1%x;J<@{~G_0p$AS;G@&
zK0>IClV!RlLfj5xEhN?r*<^JMLQwt=-f|?b9xc3k7%&|?>yZba^wQl?On9mdMUoOI
zsWv9u4#N}@vxcP#qF52h{a5_-(Q^UQRWIybXnS3Y0n=72lA+|BkBOsLsWzmL*({{6
zSQB6{2^92-Fiq13H&cHENYkHQ{LHVt^r8q&z{u=9u@WSjm1$nA6}pfejtQ%7KvOB2
zQyHi%g6YaLAKVF;ZtChgMtSLzD==aCKSUG931ZFAQ4(f{W5UHM!>g)DX;Do3xy*a*
zfNB3#@b0BgiekbuhNbSr)11uIoUAC^4#N}@YlfjymS9CO?Q+(<^)O)C^&ETe$zFP4
zbxbJ7SjNQDLhaZg=g6)CQ(4UFrl#>!1txuHM_(Ucn&|)N<xO5!8Zd>rIG&aCF|icf
zcZ7j5_nc1QW*DT9nAUl?Z>d&5nm%~x#`giz^rzcCddf?8MQLKNG^P&;?q~A!Jcb=^
zhhxI38{ioVXcC!d=(1<#pNs(}z3}xnKK9b7YM6q=XBiVG&^6Z?%EuJgMKH~XMV*l?
zSUIDZcD{3T+mw%K_gmVJy>v2)2~U-lLI<};1-5qVa61fBNGuB+qfwk5#WdR0yKfLM
zId}f`y$8I`1Yo)lpE2>|s*g$FSin@XJK<&nj@nQ4(~`!kk|qu?yw0$?icLzPH0Pyj
z_5qx&Pn*qayw1+*a8MHVGs8h<J{-E=)J4|NA>h1aNQNX~8<jAe8+Mf77@rP%oeQeN
zLEGBT3<sI{aF~9RGFVGD^Z|=u1zM+Jt_`N_Fq}92`3Trq^0=4o2z8)HQn%EBXov$3
z{cx_qAF&HMP*kk)6ldr{#gaR6{JzKh_3iQ%kN@(Rmu`<>3Z;R4Of)5m6jP&9!yXyG
z`#Xy%EQVXkCQK!;2n1xIP3G3&Bc~k%Ogj!7c>NJC-BtzDOyanV36?jQL+0kmcRphm
z#xyIoU^3EJs$%eu{L_ByL%_7N>BgO&*A)j$v-7s!UM$t@W17!Ei<se%u-XP>OQcPv
zaxcE+&9_#r^t;pJ?;Uv6OSeX8!gJF;Bo^+bOLC3n4KU$$7^aYzmEhvJZdK4U^yLfA
zZUju5j!gaL953Ax#e`?GOQB<6-&JcH0GMz)3{yyK%Ay9#Tg4nXedX!*P5`Exy3V*^
zqnB>3z=Y+$5hnOE1HQ)A#K(l$;h1o-s<E0zYcS)6yZFRAeZT4eOvOvYhEKe7Q#DLM
z#=Ojv%nCK9f)b{{E`n)BY{)wNv4m8y)wSMz|C74`)5IH(tbWK#H&$Yr&9Ng)tRT+|
z)2v+x6DBrvOI8KBVtTsl*zvJlfa&rKynE?}Dwt-n@(2^$&98OdEX8j23V#^WOmzdA
zsPejDRPe-U6Hk2$Fb&@I)#iu%`~OYJ4dwIyypQvwAYp(oKo}ql5C#YXgaN_;VSq3|
z7$6M%A2Q&jmxs=^ksBBOxt5W6c<@&9!ENZ98iI3eq`Co?mSuR7Tk+WKka+s@55c+i
z(8FhJTjr&ED==Yy%vr)D$^tLV0~2P4W5UH0Wg3R8Riu^fIG{B950nob+5Nz2UV2#-
zOf!G}S;hp9&ZxN`1^R@Pa_8u~DZ-d$svBS#O=KC-j6S=$<JlL6U-vOR^wO@M{0j_}
BIY9sb

literal 0
HcmV?d00001

diff --git a/docs/runtime-turn-probes.md b/docs/runtime-turn-probes.md
new file mode 100644
index 000000000..6963cdc27
--- /dev/null
+++ b/docs/runtime-turn-probes.md
@@ -0,0 +1,40 @@
+# Native terminal turn observation
+
+Actors normally running in a terminal have no authoritative turn state. Their
+working state remains `waiting` / `pty_running_state_unknown`; terminal output
+and quiet time do not determine whether a turn is active.
+
+The optional per-Actor `turn_probe` setting accepts `off` (the default), `auto`,
+or `hermes_turn_lease`. Set it with `cccc actor update <actor> --turn-probe auto`.
+It is also a runtime-profile field; linked Actors use the profile's setting.
+Changes apply on the **next start**, including changes back to `off`.
+
+The first adapter supports Hermes CLI metadata at database schema version 31 on
+Linux. `auto` selects it only for Hermes. Unsupported runtimes/platforms, named
+Hermes profiles, unreadable or locked databases, unknown schemas, ambiguous
+holders and process identity mismatches all preserve the existing fallback.
+Managed sessions always take precedence. There is no output-based inference,
+hook installation, provider launch, global setting, interval or idle threshold.
+
+The ordinary activity publisher tick reads only lease and session metadata from
+the provider's database, opened read-only with `query_only`. Queries have a
+25 ms busy timeout, 50 ms progress budget and bounded rows/process ancestry.
+Request handlers consume only the generation-validated cached result. No
+message, prompt or response columns are selected.
+
+An unexpired CLI lease counts as working only when its holder's PID namespace
+and ancestry match the owned PTY generation. PID/start-time checks reject
+reused identities; a lease acquired before that process was born cannot bind
+it. Bindings are in memory and are discarded on a new terminal generation.
+
+Idle requires a previously bound session row that is still open, a still-live
+holder with the same process identity, and no active owned lease. A freshly
+started Actor with no observed lease stays unknown until a real turn provides
+that positive binding. An absent lease alone never establishes idle.
+
+Successful observations fill the existing `effective_working_state`,
+`effective_working_reason` (`turn_probe:hermes_turn_lease`) and timestamp fields.
+The runtime state source remains `terminal`. A provider conversation identifier
+is not a CCCC task identifier, so it does not fill `effective_active_task_id`.
+Unknown preserves the existing fields. Probe-off adds no serialized Actor field
+and makes no provider-state reads.

From f8bea7cf61d332333103aab1a72cf5e585e120ba Mon Sep 17 00:00:00 2001
From: chriscoveries <212684707+chriscoveries@users.noreply.github.com>
Date: Thu, 22 Oct 2026 19:50:02 +0000
Subject: [PATCH 3/7] test(runtime): cover turn probe authority and launch
 generations

---
 crates/cccc-core/src/profiles.rs              | 53 ++++++++++++++++
 .../ops/local_headless/notice_gate_tests.rs   | 37 ++++++++++++
 .../cccc-daemon/src/ops/turn_probe/hermes.rs  | 15 ++++-
 crates/cccc-daemon/src/ops/turn_probe/mod.rs  |  2 +-
 .../cccc-daemon/src/ops/turn_probe/tests.rs   | 60 +++++++++++++++++++
 crates/cccc-daemon/src/ops/working_state.rs   | 15 +++--
 6 files changed, 174 insertions(+), 8 deletions(-)

diff --git a/crates/cccc-core/src/profiles.rs b/crates/cccc-core/src/profiles.rs
index 84518ac05..d91f721da 100644
--- a/crates/cccc-core/src/profiles.rs
+++ b/crates/cccc-core/src/profiles.rs
@@ -795,6 +795,59 @@ fn python_repr(value: &Value) -> String {
 mod tests {
     use super::*;
 
+    #[test]
+    fn turn_probe_is_opt_in_and_profile_controlled() {
+        let temp = tempfile::tempdir().expect("tempdir");
+        let home = HomeLayout::from_path(temp.path()).expect("home");
+        let store = ProfileStore::new(home).expect("store");
+        let off = store
+            .upsert(
+                json!({"id":"native", "runtime":"hermes"})
+                    .as_object()
+                    .expect("profile")
+                    .clone(),
+                None,
+            )
+            .expect("default profile");
+        assert!(off.get("turn_probe").is_none());
+        assert_eq!(
+            store
+                .runtime_ref("native", "global", "")
+                .expect("runtime")
+                .expect("profile")
+                .turn_probe,
+            TurnProbe::Off
+        );
+        store
+            .upsert(
+                json!({"id":"native", "runtime":"hermes", "turn_probe":"auto"})
+                    .as_object()
+                    .expect("profile")
+                    .clone(),
+                None,
+            )
+            .expect("opt in");
+        assert_eq!(
+            store
+                .runtime_ref("native", "global", "")
+                .expect("runtime")
+                .expect("profile")
+                .turn_probe,
+            TurnProbe::Auto
+        );
+        assert!(
+            store
+                .upsert(
+                    json!({"id":"native", "turn_probe":"guess_from_output"})
+                        .as_object()
+                        .expect("profile")
+                        .clone(),
+                    None
+                )
+                .is_err()
+        );
+    }
+
     #[test]
     fn profile_upsert_canonicalizes_shell_command_strings() {
         let temp = tempfile::tempdir().expect("tempdir");
diff --git a/crates/cccc-daemon/src/ops/local_headless/notice_gate_tests.rs b/crates/cccc-daemon/src/ops/local_headless/notice_gate_tests.rs
index 30eff71c7..1e68a23f9 100644
--- a/crates/cccc-daemon/src/ops/local_headless/notice_gate_tests.rs
+++ b/crates/cccc-daemon/src/ops/local_headless/notice_gate_tests.rs
@@ -60,6 +60,43 @@ fn send() -> Event {
     event("chat.message", json!({"text":"work"}))
 }
 
+#[test]
+fn registered_managed_authority_never_consults_a_native_turn_probe() {
+    let temp = tempfile::tempdir().expect("tempdir");
+    let (mut item, _control) = session(temp.path());
+    item.group_id = format!("managed-probe-{}", uuid::Uuid::new_v4());
+    item.managed = Arc::new(AnalystSession::claude_for_shutdown_test(
+        temp.path(),
+        "abcdef09",
+        Vec::new(),
+        true,
+    ));
+    let key = (item.group_id.clone(), item.actor_id.clone());
+    let item = Arc::new(item);
+    sessions()
+        .write()
+        .expect("registry")
+        .insert(key.clone(), Arc::clone(&item));
+    let mut actor = Actor::new(&item.actor_id);
+    // Saved settings may already describe the next launch, not the current owner.
+    actor.runtime = ActorRuntime::Hermes;
+    actor.turn_probe = cccc_contracts::TurnProbe::Auto;
+    for expected in ["working", "idle"] {
+        item.set_status(expected, None);
+        let state = crate::ops::working_state::fields_with_probe(
+            &item.home,
+            &actor,
+            &item.group_id,
+            true,
+            "pty",
+            || panic!("managed authority must skip the probe"),
+        );
+        assert_eq!(state["effective_working_state"], expected);
+        assert_eq!(state["effective_working_reason"], "managed_agent_session");
+    }
+    sessions().write().expect("registry").remove(&key);
+}
+
 fn turn(session: &Session, method: &str, id: &str) {
     super::super::output::handle_message(
         session,
diff --git a/crates/cccc-daemon/src/ops/turn_probe/hermes.rs b/crates/cccc-daemon/src/ops/turn_probe/hermes.rs
index b1b799f8f..f3e5de62d 100644
--- a/crates/cccc-daemon/src/ops/turn_probe/hermes.rs
+++ b/crates/cccc-daemon/src/ops/turn_probe/hermes.rs
@@ -21,7 +21,20 @@ impl Hermes {
     ) -> TurnObservation {
         match self.read(path, processes, root, now) {
             Ok(value) => value,
-            Err(why) => TurnObservation::Unknown { why },
+            Err(why) => {
+                if matches!(
+                    why,
+                    "holder_generation_mismatch"
+                        | "session_binding_changed"
+                        | "holder_left_tree"
+                        | "generation_changed"
+                        | "process_unavailable"
+                        | "process_not_live"
+                ) {
+                    self.sessions.clear();
+                }
+                TurnObservation::Unknown { why }
+            }
         }
     }
 
diff --git a/crates/cccc-daemon/src/ops/turn_probe/mod.rs b/crates/cccc-daemon/src/ops/turn_probe/mod.rs
index 02f7664f8..9dc62acb9 100644
--- a/crates/cccc-daemon/src/ops/turn_probe/mod.rs
+++ b/crates/cccc-daemon/src/ops/turn_probe/mod.rs
@@ -76,7 +76,7 @@ pub(super) fn refresh(group: &str, actor: &Actor) {
         return;
     };
     // Registered managed authority always wins, including late registration.
-    entry.observation = if super::local_headless::running(group, &actor.id)
+    entry.observation = if super::local_headless::registered_running(group, &actor.id).is_some()
         || super::local_headless::status(group, &actor.id).is_some()
     {
         TurnObservation::Unknown {
diff --git a/crates/cccc-daemon/src/ops/turn_probe/tests.rs b/crates/cccc-daemon/src/ops/turn_probe/tests.rs
index 07db81ae5..fd9deecf2 100644
--- a/crates/cccc-daemon/src/ops/turn_probe/tests.rs
+++ b/crates/cccc-daemon/src/ops/turn_probe/tests.rs
@@ -281,3 +281,63 @@ fn stopped_and_headless_states_never_consult_the_probe() {
         assert_eq!(state["effective_working_state"], expected);
     }
 }
+
+#[test]
+fn launch_policy_waits_for_restart_and_cache_never_crosses_a_generation() {
+    let temp = tempfile::tempdir().expect("tempdir");
+    let group = format!("probe-{}", uuid::Uuid::new_v4());
+    let mut actor = Actor::new("peer");
+    actor.runtime = ActorRuntime::Hermes;
+    actor.turn_probe = TurnProbe::Auto;
+    let launch = || {
+        cccc_runtime::start(cccc_runtime::LaunchSpec {
+            group_id: group.clone(),
+            actor_id: actor.id.clone(),
+            runner: cccc_contracts::RunnerKind::Pty,
+            command: vec!["/bin/cat".into()],
+            cwd: temp.path().into(),
+            env: BTreeMap::new(),
+            cols: 80,
+            rows: 24,
+        })
+        .expect("owned terminal")
+    };
+    let first = launch();
+    started(&actor, &first);
+    let key = (group.clone(), actor.id.clone());
+    entries()
+        .lock()
+        .expect("entries")
+        .get_mut(&key)
+        .expect("opted-in generation")
+        .observation = TurnObservation::Working { since: None };
+    assert!(matches!(
+        cached(&group, &actor.id),
+        Some(TurnObservation::Working { .. })
+    ));
+    let mut saved = actor.clone();
+    saved.turn_probe = TurnProbe::Off;
+    assert!(
+        matches!(
+            cached(&group, &saved.id),
+            Some(TurnObservation::Working { .. })
+        ),
+        "saving off must not alter the running launch policy"
+    );
+    cccc_runtime::stop(&group, &actor.id).expect("stop first");
+    assert_eq!(cached(&group, &actor.id), None);
+    let second = launch();
+    assert_eq!(
+        cached(&group, &actor.id),
+        None,
+        "new generation cannot inherit working"
+    );
+    started(&saved, &second);
+    assert!(
+        !entries().lock().expect("entries").contains_key(&key),
+        "off launch has no provider reader"
+    );
+    // Saving auto without a restart cannot enable a previously off generation.
+    assert_eq!(cached(&group, &actor.id), None);
+    cccc_runtime::stop(&group, &actor.id).expect("stop second");
+}
diff --git a/crates/cccc-daemon/src/ops/working_state.rs b/crates/cccc-daemon/src/ops/working_state.rs
index 599c82764..1255ac98a 100644
--- a/crates/cccc-daemon/src/ops/working_state.rs
+++ b/crates/cccc-daemon/src/ops/working_state.rs
@@ -43,15 +43,18 @@ pub(super) fn fields_with_probe(
     runner_effective: &str,
     probe: impl FnOnce() -> Option<super::turn_probe::TurnObservation>,
 ) -> Map<String, Value> {
-    let registered_managed = super::local_headless::running(group_id, &actor.id);
-    let managed_session = registered_managed || super::local_headless::uses_managed_session(actor);
+    let registered_managed = super::local_headless::registered_running(group_id, &actor.id);
+    let managed_session =
+        registered_managed.unwrap_or(false) || super::local_headless::uses_managed_session(actor);
     let local_state = running
         .then(|| super::local_headless::status(group_id, &actor.id))
         .flatten();
-    let probe =
-        (running && !registered_managed && local_state.is_none() && runner_effective == "pty")
-            .then(probe)
-            .flatten();
+    let probe = (running
+        && registered_managed.is_none()
+        && local_state.is_none()
+        && runner_effective == "pty")
+        .then(probe)
+        .flatten();
     let (state, reason, updated_at, active_task_id) = if !running {
         (
             "stopped".to_owned(),

From cb4ba5e19ec7815aa88d1114ec24b779f0f99bb7 Mon Sep 17 00:00:00 2001
From: chriscoveries <212684707+chriscoveries@users.noreply.github.com>
Date: Thu, 22 Oct 2026 19:56:59 +0000
Subject: [PATCH 4/7] fix(runtime): invalidate probe bindings when provider
 storage changes

---
 crates/cccc-daemon/Cargo.toml                 |  2 +-
 .../cccc-daemon/src/ops/turn_probe/hermes.rs  | 37 ++++++++-
 crates/cccc-daemon/src/ops/turn_probe/mod.rs  |  4 +-
 .../cccc-daemon/src/ops/turn_probe/tests.rs   | 76 ++++++++++++++++++-
 docs/runtime-turn-probes.md                   |  3 +-
 5 files changed, 114 insertions(+), 8 deletions(-)

diff --git a/crates/cccc-daemon/Cargo.toml b/crates/cccc-daemon/Cargo.toml
index 27aaee1b5..48f0707fb 100644
--- a/crates/cccc-daemon/Cargo.toml
+++ b/crates/cccc-daemon/Cargo.toml
@@ -26,7 +26,7 @@ regex.workspace = true
 same-file.workspace = true
 reqwest.workspace = true
 rustls.workspace = true
-rusqlite = { version = "0.32.1", features = ["bundled", "hooks"] }
+rusqlite = { version = "0.32.1", features = ["bundled", "hooks", "limits"] }
 tokio-rustls.workspace = true
 serde.workspace = true
 serde_json.workspace = true
diff --git a/crates/cccc-daemon/src/ops/turn_probe/hermes.rs b/crates/cccc-daemon/src/ops/turn_probe/hermes.rs
index f3e5de62d..315377fd8 100644
--- a/crates/cccc-daemon/src/ops/turn_probe/hermes.rs
+++ b/crates/cccc-daemon/src/ops/turn_probe/hermes.rs
@@ -9,6 +9,7 @@ use std::time::{Duration, Instant};
 pub(super) struct Hermes {
     // Learned only from a live owned lease, never from session id/cwd similarity.
     sessions: BTreeMap<String, Process>,
+    database_identity: Option<(u64, u64)>,
 }
 
 impl Hermes {
@@ -30,6 +31,7 @@ impl Hermes {
                         | "generation_changed"
                         | "process_unavailable"
                         | "process_not_live"
+                        | "database_replaced"
                 ) {
                     self.sessions.clear();
                 }
@@ -48,11 +50,17 @@ impl Hermes {
         if !now.is_finite() {
             return Err("clock_invalid");
         }
+        let identity = database_identity(path)?;
+        if self.database_identity.is_some_and(|old| old != identity) {
+            self.database_identity = Some(identity);
+            return Err("database_replaced");
+        }
         let db = Connection::open_with_flags(
             path,
             OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
         )
         .map_err(|_| "database_unavailable")?;
+        db.set_limit(rusqlite::limits::Limit::SQLITE_LIMIT_LENGTH, 8192);
         db.busy_timeout(Duration::from_millis(25))
             .map_err(|_| "database_unavailable")?;
         let deadline = Instant::now() + Duration::from_millis(50);
@@ -89,6 +97,9 @@ impl Hermes {
         let mut working_since: Option<f64> = None;
         let mut learned = self.sessions.clone();
         for (session, holder, acquired, expiry) in leases {
+            if Instant::now() >= deadline {
+                return Err("probe_budget");
+            }
             if !acquired.is_finite() || !expiry.is_finite() || expiry < acquired {
                 return Err("lease_invalid");
             }
@@ -124,6 +135,9 @@ impl Hermes {
         // Recheck every provider identity. A dead/reused holder is never proof of idle.
         let mut live_sessions = 0;
         for (session, bound) in &learned {
+            if Instant::now() >= deadline {
+                return Err("probe_budget");
+            }
             let current = owned(processes, root, bound.pid)?.ok_or("holder_left_tree")?;
             if current.start != bound.start || current.namespace != bound.namespace {
                 return Err("holder_generation_mismatch");
@@ -146,22 +160,43 @@ impl Hermes {
         }
         db.execute_batch("ROLLBACK")
             .map_err(|_| "database_unavailable")?;
+        if database_identity(path)? != identity {
+            return Err("database_replaced");
+        }
         if Instant::now() >= deadline {
             return Err("probe_budget");
         }
         self.sessions = learned;
+        self.database_identity = Some(identity);
         if let Some(since) = working_since {
             Ok(TurnObservation::Working {
                 since: timestamp(since),
             })
         } else if live_sessions > 0 {
-            Ok(TurnObservation::Idle)
+            Ok(TurnObservation::Idle {
+                since: timestamp(now),
+            })
         } else {
             Err("session_not_bound")
         }
     }
 }
 
+#[cfg(target_os = "linux")]
+fn database_identity(path: &Path) -> Result<(u64, u64), &'static str> {
+    use std::os::unix::fs::MetadataExt;
+    let metadata = std::fs::metadata(path).map_err(|_| "database_unavailable")?;
+    if !metadata.is_file() {
+        return Err("database_not_regular");
+    }
+    Ok((metadata.dev(), metadata.ino()))
+}
+
+#[cfg(not(target_os = "linux"))]
+fn database_identity(_path: &Path) -> Result<(u64, u64), &'static str> {
+    Err("database_identity_unsupported")
+}
+
 fn timestamp(seconds: f64) -> Option<String> {
     chrono::DateTime::from_timestamp_millis((seconds * 1000.0) as i64)
         .map(|time| time.to_rfc3339_opts(chrono::SecondsFormat::Millis, true))
diff --git a/crates/cccc-daemon/src/ops/turn_probe/mod.rs b/crates/cccc-daemon/src/ops/turn_probe/mod.rs
index 9dc62acb9..33c51e2c4 100644
--- a/crates/cccc-daemon/src/ops/turn_probe/mod.rs
+++ b/crates/cccc-daemon/src/ops/turn_probe/mod.rs
@@ -12,7 +12,7 @@ use std::time::{SystemTime, UNIX_EPOCH};
 #[derive(Debug, Clone, PartialEq)]
 pub(super) enum TurnObservation {
     Working { since: Option<String> },
-    Idle,
+    Idle { since: Option<String> },
     Unknown { why: &'static str },
 }
 
@@ -139,7 +139,7 @@ pub(super) fn fields(
 ) -> (String, String, Option<String>, Option<String>) {
     let (state, since) = match observation {
         TurnObservation::Working { since } => ("working", since),
-        TurnObservation::Idle => ("idle", Some(cccc_contracts::utc_now())),
+        TurnObservation::Idle { since } => ("idle", since),
         TurnObservation::Unknown { .. } => {
             unreachable!("Unknown falls through before field construction")
         }
diff --git a/crates/cccc-daemon/src/ops/turn_probe/tests.rs b/crates/cccc-daemon/src/ops/turn_probe/tests.rs
index fd9deecf2..43c416402 100644
--- a/crates/cccc-daemon/src/ops/turn_probe/tests.rs
+++ b/crates/cccc-daemon/src/ops/turn_probe/tests.rs
@@ -92,10 +92,10 @@ fn real_lease_is_working_then_provider_expiry_is_idle_only_after_binding() {
         )
         .expect("expiry");
     // Replay at the provider's own expiry; no invented lease or session rows.
-    assert_eq!(
+    assert!(matches!(
         probe.observe(&path, &capture, &root, last_expiry + 1.0),
-        TurnObservation::Idle
-    );
+        TurnObservation::Idle { .. }
+    ));
     assert!(
         matches!(
             hermes::Hermes::default().observe(&path, &capture, &root, last_expiry + 1.0),
@@ -131,6 +131,76 @@ fn foreign_lease_cannot_establish_an_actor_binding() {
     ));
 }
 
+#[test]
+fn several_real_sessions_require_every_owned_lease_to_expire_before_idle() {
+    let capture = Capture::load();
+    let parent = capture
+        .read(capture.root().parent)
+        .expect("captured common ancestor");
+    let path = fixture("during.sqlite3");
+    let mut probe = hermes::Hermes::default();
+    assert!(matches!(
+        probe.observe(&path, &capture, &parent, capture.now()),
+        TurnObservation::Working { .. }
+    ));
+    let db = Connection::open(&path).expect("fixture");
+    let expiry: (f64, f64) = db
+        .query_row(
+            "SELECT MIN(expires_at),MAX(expires_at) FROM session_turn_leases WHERE expires_at>?1",
+            [capture.now()],
+            |row| Ok((row.get(0)?, row.get(1)?)),
+        )
+        .expect("provider expiries");
+    assert!(
+        expiry.0 < expiry.1,
+        "capture includes distinct live lease expiries"
+    );
+    assert!(
+        matches!(
+            probe.observe(&path, &capture, &parent, (expiry.0 + expiry.1) / 2.0),
+            TurnObservation::Working { .. }
+        ),
+        "one expired lease cannot make several bound sessions idle"
+    );
+    assert!(matches!(
+        probe.observe(&path, &capture, &parent, expiry.1 + 1.0),
+        TurnObservation::Idle { .. }
+    ));
+}
+
+#[test]
+fn replacing_the_provider_database_discards_the_prior_session_binding() {
+    let capture = Capture::load();
+    let (_temp, path) = copy_database();
+    let mut probe = hermes::Hermes::default();
+    assert!(matches!(
+        probe.observe(&path, &capture, &capture.root(), capture.now()),
+        TurnObservation::Working { .. }
+    ));
+    let later: f64 = Connection::open(&path)
+        .expect("copy")
+        .query_row(
+            "SELECT MAX(expires_at)+1 FROM session_turn_leases",
+            [],
+            |row| row.get(0),
+        )
+        .expect("expiry");
+    let replacement = path.with_extension("replacement");
+    std::fs::copy(fixture("during.sqlite3"), &replacement).expect("real capture copy");
+    std::fs::rename(replacement, &path).expect("replace database");
+    assert!(matches!(
+        probe.observe(&path, &capture, &capture.root(), later),
+        TurnObservation::Unknown { .. }
+    ));
+    assert!(
+        matches!(
+            probe.observe(&path, &capture, &capture.root(), later),
+            TurnObservation::Unknown { .. }
+        ),
+        "matching old rows in a different database are not a positive process binding"
+    );
+}
+
 #[test]
 fn unavailable_schema_lock_and_missing_session_fail_closed() {
     let capture = Capture::load();
diff --git a/docs/runtime-turn-probes.md b/docs/runtime-turn-probes.md
index 6963cdc27..9a0d08e70 100644
--- a/docs/runtime-turn-probes.md
+++ b/docs/runtime-turn-probes.md
@@ -25,7 +25,8 @@ message, prompt or response columns are selected.
 An unexpired CLI lease counts as working only when its holder's PID namespace
 and ancestry match the owned PTY generation. PID/start-time checks reject
 reused identities; a lease acquired before that process was born cannot bind
-it. Bindings are in memory and are discarded on a new terminal generation.
+it. Bindings are in memory and are discarded on a new terminal generation or
+replacement of the provider's database file. Oversized metadata is rejected.
 
 Idle requires a previously bound session row that is still open, a still-live
 holder with the same process identity, and no active owned lease. A freshly

From c68a3f5ef874a2d7fb73b6658df33238c82a0102 Mon Sep 17 00:00:00 2001
From: chriscoveries <212684707+chriscoveries@users.noreply.github.com>
Date: Thu, 22 Oct 2026 20:07:05 +0000
Subject: [PATCH 5/7] fix(runtime): require confirmed managed-session absence
 before probing

---
 crates/cccc-daemon/src/ops/local_headless/mod.rs        | 2 +-
 crates/cccc-daemon/src/ops/local_headless/supervisor.rs | 7 +++++++
 crates/cccc-daemon/src/ops/turn_probe/mod.rs            | 5 ++++-
 3 files changed, 12 insertions(+), 2 deletions(-)

diff --git a/crates/cccc-daemon/src/ops/local_headless/mod.rs b/crates/cccc-daemon/src/ops/local_headless/mod.rs
index 8c0849e2d..d56ca74d8 100644
--- a/crates/cccc-daemon/src/ops/local_headless/mod.rs
+++ b/crates/cccc-daemon/src/ops/local_headless/mod.rs
@@ -22,10 +22,10 @@ pub(crate) use events::{
     append as append_event, append_with_dedupe as append_event_with_dedupe,
     contains_dedupe as contains_event_dedupe,
 };
-pub(crate) use supervisor::registered_running;
 pub(crate) use supervisor::{
     cancel_turn, respond_interaction, respond_permission, structured_state,
 };
+pub(crate) use supervisor::{has_no_registered_session, registered_running};
 
 use cccc_core::HomeLayout;
 use serde::Serialize;
diff --git a/crates/cccc-daemon/src/ops/local_headless/supervisor.rs b/crates/cccc-daemon/src/ops/local_headless/supervisor.rs
index 793a82358..a732fba65 100644
--- a/crates/cccc-daemon/src/ops/local_headless/supervisor.rs
+++ b/crates/cccc-daemon/src/ops/local_headless/supervisor.rs
@@ -391,6 +391,13 @@ pub(crate) fn registered_running(group_id: &str, actor_id: &str) -> Option<bool>
     lookup(&(group_id.to_owned(), actor_id.to_owned())).map(|item| item.running())
 }
 
+/// Native observations require confirmed absence, including under registry errors.
+pub(crate) fn has_no_registered_session(group_id: &str, actor_id: &str) -> bool {
+    sessions()
+        .read()
+        .is_ok_and(|sessions| !sessions.contains_key(&(group_id.to_owned(), actor_id.to_owned())))
+}
+
 #[must_use]
 pub fn status(group_id: &str, actor_id: &str) -> Option<HeadlessStatus> {
     let item = lookup(&(group_id.to_owned(), actor_id.to_owned()))?;
diff --git a/crates/cccc-daemon/src/ops/turn_probe/mod.rs b/crates/cccc-daemon/src/ops/turn_probe/mod.rs
index 33c51e2c4..d7cd59224 100644
--- a/crates/cccc-daemon/src/ops/turn_probe/mod.rs
+++ b/crates/cccc-daemon/src/ops/turn_probe/mod.rs
@@ -76,7 +76,7 @@ pub(super) fn refresh(group: &str, actor: &Actor) {
         return;
     };
     // Registered managed authority always wins, including late registration.
-    entry.observation = if super::local_headless::registered_running(group, &actor.id).is_some()
+    entry.observation = if !super::local_headless::has_no_registered_session(group, &actor.id)
         || super::local_headless::status(group, &actor.id).is_some()
     {
         TurnObservation::Unknown {
@@ -127,6 +127,9 @@ pub(super) fn cached(group: &str, actor: &str) -> Option<TurnObservation> {
         .ok()?
         .get(&(group.into(), actor.into()))
         .cloned()?;
+    if !super::local_headless::has_no_registered_session(group, actor) {
+        return None;
+    }
     generation(group, actor, &entry)?;
     match entry.observation {
         TurnObservation::Unknown { .. } => None,

From ccc69e331fbb57904ca45442157e7434d421d62f Mon Sep 17 00:00:00 2001
From: chriscoveries <212684707+chriscoveries@users.noreply.github.com>
Date: Thu, 22 Oct 2026 20:09:37 +0000
Subject: [PATCH 6/7] fix(runtime): retire turn probe bindings after confirmed
 stop

---
 crates/cccc-daemon/src/ops/actor_runtime.rs    | 6 +++++-
 crates/cccc-daemon/src/ops/turn_probe/mod.rs   | 7 +++++++
 crates/cccc-daemon/src/ops/turn_probe/tests.rs | 2 ++
 3 files changed, 14 insertions(+), 1 deletion(-)

diff --git a/crates/cccc-daemon/src/ops/actor_runtime.rs b/crates/cccc-daemon/src/ops/actor_runtime.rs
index c0c10da3e..91ef0f009 100644
--- a/crates/cccc-daemon/src/ops/actor_runtime.rs
+++ b/crates/cccc-daemon/src/ops/actor_runtime.rs
@@ -182,11 +182,15 @@ fn start(home: &HomeLayout, group: &GroupDoc, actor: &Actor) -> Result<SessionSt
 }
 
 pub(super) fn stop(group: &GroupDoc, actor_id: &str) -> Result<Option<SessionStatus>, OpError> {
-    match cccc_runtime::stop(&group.group_id, actor_id) {
+    let result = match cccc_runtime::stop(&group.group_id, actor_id) {
         Ok(status) => Ok(Some(status)),
         Err(cccc_runtime::RuntimeError::NotFound(_, _)) => Ok(None),
         Err(error) => Err(runtime_error(error)),
+    };
+    if result.is_ok() {
+        super::turn_probe::stopped(&group.group_id, actor_id);
     }
+    result
 }
 
 fn stop_registered(group: &GroupDoc, actor_id: &str) -> Result<Option<SessionStatus>, OpError> {
diff --git a/crates/cccc-daemon/src/ops/turn_probe/mod.rs b/crates/cccc-daemon/src/ops/turn_probe/mod.rs
index d7cd59224..0851c6e3d 100644
--- a/crates/cccc-daemon/src/ops/turn_probe/mod.rs
+++ b/crates/cccc-daemon/src/ops/turn_probe/mod.rs
@@ -30,6 +30,13 @@ fn entries() -> &'static Mutex<BTreeMap<Key, Entry>> {
     ENTRIES.get_or_init(Mutex::default)
 }
 
+/// Retire bounded provider bindings after confirmed stop, including Actor deletion.
+pub(super) fn stopped(group: &str, actor: &str) {
+    if let Ok(mut entries) = entries().lock() {
+        entries.remove(&(group.to_owned(), actor.to_owned()));
+    }
+}
+
 /// Called once at a successful launch: saved configuration changes wait for restart.
 pub(super) fn started(actor: &Actor, status: &SessionStatus) {
     let key = (status.group_id.clone(), status.actor_id.clone());
diff --git a/crates/cccc-daemon/src/ops/turn_probe/tests.rs b/crates/cccc-daemon/src/ops/turn_probe/tests.rs
index 43c416402..5b44bf75b 100644
--- a/crates/cccc-daemon/src/ops/turn_probe/tests.rs
+++ b/crates/cccc-daemon/src/ops/turn_probe/tests.rs
@@ -410,4 +410,6 @@ fn launch_policy_waits_for_restart_and_cache_never_crosses_a_generation() {
     // Saving auto without a restart cannot enable a previously off generation.
     assert_eq!(cached(&group, &actor.id), None);
     cccc_runtime::stop(&group, &actor.id).expect("stop second");
+    stopped(&group, &actor.id);
+    assert!(!entries().lock().expect("entries").contains_key(&key));
 }

From d3cb0464c5e2c564a028aa1e9afff2467b62457a Mon Sep 17 00:00:00 2001
From: chriscoveries <212684707+chriscoveries@users.noreply.github.com>
Date: Thu, 22 Oct 2026 20:27:55 +0000
Subject: [PATCH 7/7] test(runtime): initialize probe authority and corruption
 controls correctly

---
 .../cccc-daemon/src/ops/local_headless/notice_gate_tests.rs   | 4 ++--
 crates/cccc-daemon/src/ops/turn_probe/tests.rs                | 2 +-
 crates/cccc-daemon/tests/fixtures/hermes-turn-probe/README.md | 3 +++
 3 files changed, 6 insertions(+), 3 deletions(-)

diff --git a/crates/cccc-daemon/src/ops/local_headless/notice_gate_tests.rs b/crates/cccc-daemon/src/ops/local_headless/notice_gate_tests.rs
index 1e68a23f9..1559e30c4 100644
--- a/crates/cccc-daemon/src/ops/local_headless/notice_gate_tests.rs
+++ b/crates/cccc-daemon/src/ops/local_headless/notice_gate_tests.rs
@@ -60,8 +60,8 @@ fn send() -> Event {
     event("chat.message", json!({"text":"work"}))
 }
 
-#[test]
-fn registered_managed_authority_never_consults_a_native_turn_probe() {
+#[tokio::test]
+async fn registered_managed_authority_never_consults_a_native_turn_probe() {
     let temp = tempfile::tempdir().expect("tempdir");
     let (mut item, _control) = session(temp.path());
     item.group_id = format!("managed-probe-{}", uuid::Uuid::new_v4());
diff --git a/crates/cccc-daemon/src/ops/turn_probe/tests.rs b/crates/cccc-daemon/src/ops/turn_probe/tests.rs
index 5b44bf75b..53455e835 100644
--- a/crates/cccc-daemon/src/ops/turn_probe/tests.rs
+++ b/crates/cccc-daemon/src/ops/turn_probe/tests.rs
@@ -236,7 +236,7 @@ fn unavailable_schema_lock_and_missing_session_fail_closed() {
         TurnObservation::Working { .. }
     ));
     writer
-        .execute_batch("DELETE FROM sessions")
+        .execute_batch("PRAGMA foreign_keys=OFF; DELETE FROM sessions")
         .expect("missing sessions control");
     let later: f64 = writer
         .query_row(
diff --git a/crates/cccc-daemon/tests/fixtures/hermes-turn-probe/README.md b/crates/cccc-daemon/tests/fixtures/hermes-turn-probe/README.md
index b0a395679..2d166bbb3 100644
--- a/crates/cccc-daemon/tests/fixtures/hermes-turn-probe/README.md
+++ b/crates/cccc-daemon/tests/fixtures/hermes-turn-probe/README.md
@@ -20,3 +20,6 @@ does not create a provider record or claim that a real turn ended at that time.
 Negative controls delete a table/row or change a schema version on disposable
 copies of this capture. They are deliberately damaged captures, not additional
 provider observations. Database contention uses an exclusive lock on a copy.
+The original session DDL references content tables that are intentionally absent
+from this projection. Mutation controls disable foreign-key enforcement on their
+disposable copy; the adapter performs only read queries.