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) 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, #[arg(long, value_parser = ["default", "acp"])] runtime_mode: Option, + /// Read-only native turn observation; takes effect on the next start. + #[arg(long, value_parser = ["off", "auto", "hermes_turn_lease"])] + turn_probe: Option, #[arg(long)] scope: Option, #[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..d91f721da 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, 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" @@ -778,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/Cargo.toml b/crates/cccc-daemon/Cargo.toml index 31438bf0e..48f0707fb 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", "limits"] } 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 Result Result Result, 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, OpError> { 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/notice_gate_tests.rs b/crates/cccc-daemon/src/ops/local_headless/notice_gate_tests.rs index 30eff71c7..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,6 +60,43 @@ fn send() -> Event { event("chat.message", json!({"text":"work"})) } +#[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()); + 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/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 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 { let item = lookup(&(group_id.to_owned(), actor_id.to_owned()))?; 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..315377fd8 --- /dev/null +++ b/crates/cccc-daemon/src/ops/turn_probe/hermes.rs @@ -0,0 +1,217 @@ +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, + database_identity: Option<(u64, u64)>, +} + +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) => { + if matches!( + why, + "holder_generation_mismatch" + | "session_binding_changed" + | "holder_left_tree" + | "generation_changed" + | "process_unavailable" + | "process_not_live" + | "database_replaced" + ) { + self.sessions.clear(); + } + TurnObservation::Unknown { why } + } + } + } + + fn read( + &mut self, + path: &Path, + processes: &impl Processes, + root: &Process, + now: f64, + ) -> Result { + 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); + 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::>>() + }) + .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::>>()) + .map_err(|_| "database_unavailable")?; + if leases.len() > 256 { + return Err("lease_limit"); + } + let mut working_since: Option = 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"); + } + 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 { + 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"); + } + let ended = db + .query_row( + "SELECT ended_at FROM sessions WHERE id=?1", + [session], + |row| row.get::<_, Option>(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 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 { + 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 { + 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..0851c6e3d --- /dev/null +++ b/crates/cccc-daemon/src/ops/turn_probe/mod.rs @@ -0,0 +1,167 @@ +//! 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 }, + Idle { since: Option }, + 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> { + static ENTRIES: OnceLock>> = OnceLock::new(); + 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()); + 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::has_no_registered_session(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 { + 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 { + let entry = entries() + .lock() + .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, + observation => Some(observation), + } +} + +pub(super) fn fields( + observation: TurnObservation, +) -> (String, String, Option, Option) { + let (state, since) = match observation { + TurnObservation::Working { since } => ("working", since), + TurnObservation::Idle { since } => ("idle", since), + 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; +} + +pub(super) struct ProcFs; + +#[cfg(target_os = "linux")] +impl Processes for ProcFs { + fn read(&self, pid: u32) -> Result { + 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::>(); + 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 { + Err("process_identity_unsupported") + } +} + +pub(super) fn owned( + processes: &impl Processes, + root: &Process, + holder: u32, +) -> Result, &'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 { + 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 { + 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..53455e835 --- /dev/null +++ b/crates/cccc-daemon/src/ops/turn_probe/tests.rs @@ -0,0 +1,415 @@ +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, +} + +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 { + 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!(matches!( + 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 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(); + 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("PRAGMA foreign_keys=OFF; 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); + } +} + +#[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"); + stopped(&group, &actor.id); + assert!(!entries().lock().expect("entries").contains_key(&key)); +} diff --git a/crates/cccc-daemon/src/ops/working_state.rs b/crates/cccc-daemon/src/ops/working_state.rs index f04590470..1255ac98a 100644 --- a/crates/cccc-daemon/src/ops/working_state.rs +++ b/crates/cccc-daemon/src/ops/working_state.rs @@ -30,11 +30,31 @@ pub(super) fn fields( running: bool, runner_effective: &str, ) -> Map { - 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, +) -> Map { + 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.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(), @@ -55,6 +75,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..2d166bbb3 --- /dev/null +++ b/crates/cccc-daemon/tests/fixtures/hermes-turn-probe/README.md @@ -0,0 +1,25 @@ +# 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. +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. 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 000000000..500a591ba Binary files /dev/null and b/crates/cccc-daemon/tests/fixtures/hermes-turn-probe/during.sqlite3 differ diff --git a/docs/runtime-turn-probes.md b/docs/runtime-turn-probes.md new file mode 100644 index 000000000..9a0d08e70 --- /dev/null +++ b/docs/runtime-turn-probes.md @@ -0,0 +1,41 @@ +# 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 --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 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 +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.