diff --git a/src/commands/launch.rs b/src/commands/launch.rs index 8773ebb3..05e35236 100644 --- a/src/commands/launch.rs +++ b/src/commands/launch.rs @@ -173,6 +173,7 @@ pub fn run(argv: &[String], flags: &GlobalFlags) -> Result { args: merged_args, persisted_args: None, prior_session_id: None, + retained_principal: None, tag, system_prompt, initial_prompt, diff --git a/src/commands/list.rs b/src/commands/list.rs index d6c2655e..d6f1a4b1 100644 --- a/src/commands/list.rs +++ b/src/commands/list.rs @@ -26,6 +26,9 @@ pub struct ListArgs { pub name: Option, /// Field to extract (used with name) pub field: Option, + /// Look up the exact instance recorded for a durable principal ID + #[arg(long, value_name = "ID", conflicts_with = "name")] + pub principal: Option, /// Show recently stopped agents #[arg(long)] pub stopped: bool, @@ -99,7 +102,8 @@ pub fn cmd_list(db: &HcomDb, args: &ListArgs, ctx: Option<&CommandContext>) -> i let names_output = args.names; let sh_output = args.sh; let format_template = args.format.clone(); - let target_name = args.name.as_deref(); + let principal_target = args.principal.as_deref(); + let target_name = principal_target.or(args.name.as_deref()); let field_name = args.field.as_deref(); // Resolve current instance identity @@ -134,11 +138,67 @@ pub fn cmd_list(db: &HcomDb, args: &ListArgs, ctx: Option<&CommandContext>) -> i return 1; } + let mut queried_principal = None; let lookup_name = if is_self { current_name.clone().unwrap_or_default() } else { - let resolved = resolve_display_name(db, target); - resolved.unwrap_or_else(|| target.to_string()) + let resolved = principal_target + .is_none() + .then(|| resolve_display_name(db, target)) + .flatten(); + if let Some(name) = resolved { + name + } else { + match db.lookup_principal(target) { + Ok(crate::db::PrincipalLookup::Resolved { instance_name, .. }) => { + queried_principal = Some(target.to_string()); + instance_name + } + Ok(crate::db::PrincipalLookup::Unresolved { instance_name }) => { + let payload = serde_json::json!({ + "name": instance_name, + "principal": target, + "session_id": null, + "status": "unresolved", + }); + if json_output { + println!("{}", serde_json::to_string(&payload).unwrap_or_default()); + return 0; + } + eprintln!( + "Principal {target} is unresolved (recorded instance: {instance_name})" + ); + return 1; + } + Ok(crate::db::PrincipalLookup::MissingBinding { claiming_instances }) => { + let payload = serde_json::json!({ + "name": null, + "principal": target, + "session_id": null, + "status": "unresolved", + "reason": "missing_binding", + "claiming_instances": claiming_instances, + }); + if json_output { + println!("{}", serde_json::to_string(&payload).unwrap_or_default()); + return 0; + } + eprintln!( + "Principal {target} is unresolved (binding missing; instance claims are diagnostic only)" + ); + return 1; + } + Ok(crate::db::PrincipalLookup::Unknown) if principal_target.is_some() => { + eprintln!("Error: unknown principal: {target}"); + return 1; + } + Ok(crate::db::PrincipalLookup::Unknown) => target.to_string(), + Err(error) => { + eprintln!("Error: principal lookup failed: {error}"); + return 1; + } + } + } }; if lookup_name.is_empty() { @@ -150,6 +210,7 @@ pub fn cmd_list(db: &HcomDb, args: &ListArgs, ctx: Option<&CommandContext>) -> i Ok(Some(data)) => { let mut payload = serde_json::json!({ "name": lookup_name, + "principal": queried_principal.as_deref().or(data.principal.as_deref()), "session_id": data.session_id, "status": data.status, "directory": data.directory, @@ -181,6 +242,7 @@ pub fn cmd_list(db: &HcomDb, args: &ListArgs, ctx: Option<&CommandContext>) -> i if is_self { let payload = serde_json::json!({ "name": lookup_name, + "principal": null, "session_id": sender_identity.as_ref().and_then(|id| id.session_id.as_deref()).unwrap_or(""), }); if let Some(field) = field_name { @@ -250,6 +312,7 @@ pub fn cmd_list(db: &HcomDb, args: &ListArgs, ctx: Option<&CommandContext>) -> i "unread_count": unread_counts.get(&data.name).copied().unwrap_or(0), "headless": data.background != 0, "session_id": data.session_id.as_deref().unwrap_or(""), + "principal": data.principal, "directory": data.directory, "parent_name": data.parent_name, "agent_id": data.agent_id, diff --git a/src/commands/resume.rs b/src/commands/resume.rs index ad2e3e31..e0426a18 100644 --- a/src/commands/resume.rs +++ b/src/commands/resume.rs @@ -512,6 +512,19 @@ fn prepare_resume_plan_from_source( } else { Some(display_name.clone()) }; + let retained_principal = if !fork && !is_adoption { + Some( + db.principal_for_tracked_resume(&display_name) + .map_err(|error| { + anyhow::anyhow!( + "cannot retain tracked resume identity for '{}': {error}", + display_name + ) + })?, + ) + } else { + None + }; let tracked_fork_identity = if fork && !is_adoption { Some(TrackedForkIdentity { parent_name: display_name.clone(), @@ -541,6 +554,7 @@ fn prepare_resume_plan_from_source( // Forks start a new session on first turn; only plain resume // inherits the prior id so kill-before-bind stays resumable. prior_session_id: (!fork).then(|| session_id.clone()), + retained_principal, tag: launch_tag, system_prompt: effective_system_prompt, initial_prompt: fork_initial_prompt, @@ -2831,6 +2845,7 @@ mod tests { data.insert("status".into(), json!(ST_INACTIVE)); data.insert("created_at".into(), json!(1.0)); db.save_instance_named("zeno", &data).unwrap(); + db.create_principal_binding("p-zeno", "zeno").unwrap(); // Emit a stopped life event so load_stopped_snapshot can find the snapshot let snapshot = serde_json::json!({ @@ -3125,6 +3140,71 @@ mod tests { ); } + #[test] + fn tracked_fork_route_mints_a_principal_distinct_from_parent() { + let db = test_db(); + let mut data = serde_json::Map::new(); + data.insert("session_id".into(), json!("session-123")); + data.insert("tool".into(), json!("codex")); + data.insert("status".into(), json!("listening")); + data.insert("created_at".into(), json!(1.0)); + db.save_instance_named("luna", &data).unwrap(); + db.create_principal_binding("p-parent", "luna").unwrap(); + + let plan = prepare_resume_plan(&db, "luna", true, &[], &GlobalFlags::default()).unwrap(); + let launch = prepare_launch_for_execution(&db, &plan).unwrap(); + let child_name = launch.name.expect("tracked fork reserved child name"); + let mut child_env = std::collections::HashMap::new(); + let child_principal = + crate::launcher::attach_launch_principal(&db, &child_name, &mut child_env).unwrap(); + + assert_ne!(child_principal, "p-parent"); + assert_eq!( + db.principal_for_instance(&child_name).unwrap().as_deref(), + Some(child_principal.as_str()) + ); + } + + #[test] + fn adoption_route_mints_a_new_principal_for_each_lifecycle() { + let db = test_db(); + let source = || ResumeSource::Disk { + session_id: "019f6550-1111-7222-8333-123456789abc".to_string(), + tool: "codex".to_string(), + cwd_hint: Some("/tmp".to_string()), + }; + + let first_plan = + prepare_resume_plan_from_source(&db, source(), false, &[], &GlobalFlags::default()) + .unwrap(); + assert!( + first_plan.launch.name.is_none(), + "adoption allocates at launch" + ); + let first_name = crate::instance_names::generate_unique_name(&db).unwrap(); + let first = crate::launcher::attach_launch_principal( + &db, + &first_name, + &mut std::collections::HashMap::new(), + ) + .unwrap(); + + let second_plan = + prepare_resume_plan_from_source(&db, source(), false, &[], &GlobalFlags::default()) + .unwrap(); + assert!(second_plan.launch.name.is_none()); + let second_name = crate::instance_names::generate_unique_name(&db).unwrap(); + let second = crate::launcher::attach_launch_principal( + &db, + &second_name, + &mut std::collections::HashMap::new(), + ) + .unwrap(); + + assert_ne!(first_name, second_name); + assert_ne!(first, second); + } + #[test] fn test_resume_inherits_prior_session_id_fork_does_not() { let db = test_db(); @@ -3134,6 +3214,7 @@ mod tests { data.insert("status".into(), json!(ST_INACTIVE)); data.insert("created_at".into(), json!(1.0)); db.save_instance_named("luna", &data).unwrap(); + db.create_principal_binding("p-luna", "luna").unwrap(); // Inactive rows resolve via the stopped-snapshot life event. let snapshot = serde_json::json!({ @@ -3156,6 +3237,11 @@ mod tests { .unwrap(); let resume = prepare_resume_plan(&db, "luna", false, &[], &GlobalFlags::default()).unwrap(); + assert_eq!( + resume.launch.retained_principal.as_deref(), + Some("p-luna"), + "tracked resume must carry the prior lifecycle principal into launch" + ); assert_eq!( resume.launch.prior_session_id.as_deref(), Some("session-123"), @@ -3164,6 +3250,7 @@ mod tests { ); let fork = prepare_resume_plan(&db, "luna", true, &[], &GlobalFlags::default()).unwrap(); + assert_eq!(fork.launch.retained_principal, None); assert_eq!( fork.launch.prior_session_id, None, "forks bind a fresh session on first turn; must not inherit the parent's" diff --git a/src/commands/start.rs b/src/commands/start.rs index 4f05bd54..104fed38 100644 --- a/src/commands/start.rs +++ b/src/commands/start.rs @@ -241,8 +241,8 @@ fn start_subagent(db: &HcomDb, info: &SubagentInfo) -> Result { ) .ok(); - let (subagent_name, was_announced) = match existing { - Some((name, announced)) => (name, announced != 0), + let (subagent_name, was_announced, created_fallback) = match existing { + Some((name, announced)) => (name, announced != 0, false), None => { let alloc = instance_names::SubagentAllocation { agent_id: &info.agent_id, @@ -254,10 +254,17 @@ fn start_subagent(db: &HcomDb, info: &SubagentInfo) -> Result { status_context: Some("tool:start"), }; let name = instance_names::allocate_subagent_instance(db, &alloc)?; - (name, false) + (name, false, true) } }; + if let Err(error) = crate::hooks::claude::ensure_subagent_principal(db, &subagent_name) { + if created_fallback { + let _ = db.delete_instance(&subagent_name); + } + return Err(error); + } + // Flip to active + emit life event so TUI/watchers see the state change. lifecycle::set_status( db, @@ -362,7 +369,10 @@ fn start_from_orphan( }; // Core DB registration - let _ = pidtrack::recover_single_orphan_to_db(db, orphan, &name); + pidtrack::recover_single_orphan_to_db(db, orphan, &name).map_err(anyhow::Error::msg)?; + let principal = db + .principal_for_instance(&name)? + .ok_or_else(|| anyhow::anyhow!("recovered orphan '{name}' has no principal"))?; db.log_event( "life", @@ -372,6 +382,7 @@ fn start_from_orphan( "by": "cli", "reason": "orphan_recover", "orphan_pid": pid, + "principal": principal, }), ) .ok(); @@ -481,7 +492,7 @@ fn start_rebind( // Create fresh instance with the target name let tool = ctx.tool.as_str(); let cwd_override = ctx.cwd.to_string_lossy().to_string(); - instance_binding::initialize_instance_in_position_file( + if !instance_binding::initialize_instance_in_position_file( db, &target_name, session_id.as_deref(), @@ -496,7 +507,16 @@ fn start_rebind( None, // subagent_timeout None, // hints Some(&cwd_override), - ); + ) { + bail!("failed to initialize reclaimed instance '{target_name}'"); + } + + let principal = crate::launcher::generate_principal_id(); + if let Err(error) = db.create_principal_binding(&principal, &target_name) { + eprintln!("[hcom] warn: principal setup failed for {target_name}: {error}"); + let _ = db.delete_instance(&target_name); + return Err(error); + } // Restore cursor position + mark as announced { @@ -652,6 +672,16 @@ fn load_rebind_target_metadata(db: &HcomDb, name: &str) -> Result Option { + if !ctx.is_launched + || ctx.process_id.as_deref().is_none_or(str::is_empty) + || ctx.launch_batch_id.as_deref().is_none_or(str::is_empty) + { + return None; + } + ctx.principal_id.clone().filter(|value| !value.is_empty()) +} + /// Path C: Bare start — detect tool or create adhoc instance. fn start_bare( db: &HcomDb, @@ -735,7 +765,7 @@ fn start_bare( return Ok(0); } - instance_binding::initialize_instance_in_position_file( + if !instance_binding::initialize_instance_in_position_file( db, &name, None, // session_id @@ -750,7 +780,17 @@ fn start_bare( None, // subagent_timeout None, // hints None, // cwd_override - ); + ) { + bail!("failed to initialize instance '{name}'"); + } + + let principal = + reusable_launch_principal(ctx).unwrap_or_else(crate::launcher::generate_principal_id); + if let Err(error) = db.create_principal_binding(&principal, &name) { + eprintln!("[hcom] warn: principal setup failed for {name}: {error}"); + let _ = db.delete_instance(&name); + return Err(error); + } // Bind process if we have a process_id if let Some(ref process_id) = ctx.process_id @@ -791,6 +831,7 @@ fn start_bare( "action": "started", "tool": tool, "name": name, + "principal": principal, }), ) .ok(); @@ -965,6 +1006,79 @@ mod tests { "/tmp/dasha-code/.worktrees/layer1-basic-conversation-fixes" ); assert_eq!(inst.last_event_id, 77); + assert!( + inst.principal + .as_ref() + .is_some_and(|value| !value.is_empty()) + ); + } + + #[test] + #[serial] + fn name_reclaim_route_mints_a_principal_distinct_from_prior_lifecycle() { + let (_dir, _hcom_dir, _home, _guard) = crate::hooks::test_helpers::isolated_test_env(); + let db = HcomDb::open().unwrap(); + db.conn() + .execute( + "INSERT INTO instances (name, tool, directory, created_at) + VALUES ('nova', 'claude', '/tmp/reclaim', 1.0)", + [], + ) + .unwrap(); + db.create_principal_binding("p-prior", "nova").unwrap(); + db.delete_instance("nova").unwrap(); + log_stopped_snapshot(&db, "nova", "claude", "/tmp/reclaim", "sid-nova", 77); + let ctx = make_ctx(&[("CLAUDECODE", "1")], "/tmp/reclaim"); + + assert_eq!(start_rebind(&db, "nova", &ctx, None).unwrap(), 0); + let current = db + .principal_for_instance("nova") + .unwrap() + .expect("reclaimed principal"); + assert_ne!(current, "p-prior"); + assert!(db.lookup_principal("p-prior").unwrap().is_unresolved()); + } + + #[test] + #[serial] + fn bare_start_reuses_only_complete_launch_envelope_principal() { + let (_dir, _hcom_dir, _home, _guard) = crate::hooks::test_helpers::isolated_test_env(); + let db = HcomDb::open().unwrap(); + let launched = make_ctx( + &[ + ("HCOM_LAUNCHED", "1"), + ("HCOM_PROCESS_ID", "proc-1"), + ("HCOM_LAUNCH_BATCH_ID", "batch-1"), + ("HCOM_PRINCIPAL_ID", "p-from-launch"), + ], + "/tmp", + ); + assert_eq!( + start_bare(&db, &paths::hcom_dir(), &launched, Some("mira")).unwrap(), + 0 + ); + assert_eq!( + db.principal_for_instance("mira").unwrap().as_deref(), + Some("p-from-launch") + ); + + let incomplete = make_ctx( + &[ + ("HCOM_LAUNCHED", "1"), + ("HCOM_PROCESS_ID", "proc-2"), + ("HCOM_LAUNCH_BATCH_ID", ""), + ("HCOM_PRINCIPAL_ID", "p-spoofed"), + ], + "/tmp", + ); + assert_eq!( + start_bare(&db, &paths::hcom_dir(), &incomplete, Some("kira")).unwrap(), + 0 + ); + assert_ne!( + db.principal_for_instance("kira").unwrap().as_deref(), + Some("p-spoofed") + ); } #[test] diff --git a/src/db/instances.rs b/src/db/instances.rs index 36d7c15d..f5795480 100644 --- a/src/db/instances.rs +++ b/src/db/instances.rs @@ -19,6 +19,7 @@ pub struct InstanceStatus { #[derive(Debug, Clone)] pub struct InstanceRow { pub name: String, + pub principal: Option, pub session_id: Option, pub parent_session_id: Option, pub parent_name: Option, @@ -55,6 +56,9 @@ impl InstanceRow { fn from_row(row: &rusqlite::Row<'_>) -> rusqlite::Result { Ok(Self { name: row.get("name")?, + principal: row + .get::<_, Option>("principal")? + .filter(|s| !s.is_empty()), session_id: row .get::<_, Option>("session_id")? .filter(|s| !s.is_empty()), @@ -358,7 +362,8 @@ impl HcomDb { let mut stmt = self.conn.prepare_cached( "SELECT transcript_path, session_id, tool, directory, parent_name, tag, wait_timeout, subagent_timeout, hints, pid, created_at, background, - agent_id, launch_args, origin_device_id, background_log_file, last_event_id + agent_id, launch_args, origin_device_id, background_log_file, last_event_id, + principal FROM instances WHERE name = ?", )?; @@ -381,6 +386,7 @@ impl HcomDb { "origin_device_id": row.get::<_, String>(14).unwrap_or_default(), "background_log_file": row.get::<_, String>(15).unwrap_or_default(), "last_event_id": row.get::<_, i64>(16).unwrap_or(0), + "principal": row.get::<_, String>(17).unwrap_or_default(), })) }) { Ok(snapshot) => Ok(Some(snapshot)), @@ -786,6 +792,7 @@ impl HcomDb { "idle_since", "terminal_preset_requested", "terminal_preset_effective", + "principal", ]; if VALID_COLUMNS.contains(&key) { Ok(key) @@ -952,6 +959,23 @@ mod tests { cleanup_test_db(db_path); } + #[test] + fn test_get_instance_snapshot_carries_principal_lifecycle_evidence() { + let (db, db_path) = setup_full_test_db(); + db.conn() + .execute( + "INSERT INTO instances (name, tool, created_at, principal) + VALUES ('luna', 'codex', 1.0, 'p-luna')", + [], + ) + .unwrap(); + + let snapshot = db.get_instance_snapshot("luna").unwrap().expect("snapshot"); + assert_eq!(snapshot["principal"], "p-luna"); + + cleanup_test_db(db_path); + } + #[test] fn test_set_status_does_not_emit_launch_ready() { let (db, db_path) = setup_full_test_db(); diff --git a/src/db/mod.rs b/src/db/mod.rs index d3589040..492eabd5 100644 --- a/src/db/mod.rs +++ b/src/db/mod.rs @@ -16,7 +16,7 @@ //! - Reading instance status //! - Registering notify endpoints -use anyhow::{Context, Result}; +use anyhow::{Context, Result, bail}; use chrono::Utc; use rusqlite::Connection; @@ -26,6 +26,7 @@ mod events; mod instances; mod kv; mod notify; +mod principals; pub(crate) mod reqwatch_policy; mod sessions; pub(crate) mod subscriptions; @@ -34,18 +35,33 @@ pub use events::Message; pub use instances::InstanceRow; #[allow(unused_imports)] pub use instances::InstanceStatus; +pub use principals::PrincipalLookup; /// Schema version - bump on any schema change. -const SCHEMA_VERSION: i32 = 17; +const SCHEMA_VERSION: i32 = 18; pub const DEV_ROOT_KV_KEY: &str = "config:dev_root"; -const MIGRATIONS: &[(i32, &str)] = &[( - 17, - "ALTER TABLE instances ADD COLUMN terminal_preset_requested TEXT DEFAULT ''; +const MIGRATIONS: &[(i32, &str)] = &[ + ( + 17, + "ALTER TABLE instances ADD COLUMN terminal_preset_requested TEXT DEFAULT ''; ALTER TABLE instances ADD COLUMN terminal_preset_effective TEXT DEFAULT ''; UPDATE instances SET terminal_preset_effective = json_extract(launch_context, '$.terminal_preset') WHERE launch_context != '' AND json_valid(launch_context) AND json_extract(launch_context, '$.terminal_preset') IS NOT NULL;", -)]; + ), + ( + 18, + "ALTER TABLE instances ADD COLUMN principal TEXT; + CREATE TABLE principal_bindings ( + principal TEXT PRIMARY KEY, + instance_name TEXT NOT NULL, + epoch INTEGER NOT NULL DEFAULT 0, + created_at REAL NOT NULL, + updated_at REAL NOT NULL + ); + CREATE INDEX idx_principal_bindings_instance ON principal_bindings(instance_name);", + ), +]; /// Schema compatibility check result enum SchemaCompat { @@ -205,6 +221,17 @@ impl HcomDb { CREATE INDEX IF NOT EXISTS idx_process_bindings_instance ON process_bindings(instance_name); CREATE INDEX IF NOT EXISTS idx_process_bindings_session ON process_bindings(session_id); + -- Durable hcom-owned identity. No foreign key: a stale binding must + -- remain observable as unresolved rather than disappearing by cascade. + CREATE TABLE IF NOT EXISTS principal_bindings ( + principal TEXT PRIMARY KEY, + instance_name TEXT NOT NULL, + epoch INTEGER NOT NULL DEFAULT 0, + created_at REAL NOT NULL, + updated_at REAL NOT NULL + ); + CREATE INDEX IF NOT EXISTS idx_principal_bindings_instance ON principal_bindings(instance_name); + -- Session bindings CREATE TABLE IF NOT EXISTS session_bindings ( session_id TEXT PRIMARY KEY, @@ -247,6 +274,7 @@ impl HcomDb { idle_since TEXT DEFAULT '', pid INTEGER DEFAULT NULL, launch_context TEXT DEFAULT '', + principal TEXT, FOREIGN KEY (parent_session_id) REFERENCES instances(session_id) ON DELETE SET NULL ); @@ -333,7 +361,22 @@ impl HcomDb { /// Checks schema version, archives DB if mismatched, reconnects, and reinitializes. /// Call after open() for production use. pub fn ensure_schema(&mut self) -> Result<()> { - match self.check_schema_compat()? { + let version: i32 = self + .conn + .query_row("PRAGMA user_version", [], |row| row.get(0)) + .unwrap_or(0); + if version == SCHEMA_VERSION { + self.require_principal_schema()?; + } + let compat = self.check_schema_compat()?; + if version == SCHEMA_VERSION + && let SchemaCompat::NeedsArchive(reason, _) = &compat + { + bail!( + "DB v18 schema is incompatible ({reason}); refusing to archive or rebuild durable principal data" + ); + } + match compat { SchemaCompat::Ok => { self.init_db()?; Ok(()) @@ -412,6 +455,53 @@ impl HcomDb { } } + /// A database already stamped v18 may contain durable principal rows from + /// another build. Never archive or reconstruct it: accept the exact contract, + /// otherwise fail explicitly and leave every byte in place. + fn require_principal_schema(&self) -> Result<()> { + let table_exists: bool = self.conn.query_row( + "SELECT EXISTS(SELECT 1 FROM sqlite_master WHERE type='table' AND name='principal_bindings')", + [], + |row| row.get(0), + )?; + if !table_exists { + bail!( + "DB v18 is missing required table principal_bindings; refusing to archive or rebuild durable principal data" + ); + } + + let instance_columns: std::collections::HashSet = self + .conn + .prepare("PRAGMA table_info(instances)")? + .query_map([], |row| row.get(1))? + .collect::>()?; + if !instance_columns.contains("principal") { + bail!( + "DB v18 is missing required column instances.principal; refusing to archive or rebuild durable principal data" + ); + } + + let binding_columns: std::collections::HashSet = self + .conn + .prepare("PRAGMA table_info(principal_bindings)")? + .query_map([], |row| row.get(1))? + .collect::>()?; + for required in [ + "principal", + "instance_name", + "epoch", + "created_at", + "updated_at", + ] { + if !binding_columns.contains(required) { + bail!( + "DB v18 principal_bindings is missing required column {required}; refusing to archive or rebuild durable principal data" + ); + } + } + Ok(()) + } + /// Internal: check schema compatibility without taking action. fn check_schema_compat(&self) -> Result { let version: i32 = self @@ -433,6 +523,7 @@ impl HcomDb { "instances", "kv", "notify_endpoints", + "principal_bindings", "session_bindings", ] .into_iter() @@ -541,6 +632,7 @@ impl HcomDb { "tool", "terminal_preset_requested", "terminal_preset_effective", + "principal", ]; Ok(required .iter() @@ -881,6 +973,7 @@ pub(super) mod tests { assert!(tables.contains(&"kv".to_string())); assert!(tables.contains(&"notify_endpoints".to_string())); assert!(tables.contains(&"process_bindings".to_string())); + assert!(tables.contains(&"principal_bindings".to_string())); assert!(tables.contains(&"session_bindings".to_string())); cleanup_test_db(db_path); @@ -1139,27 +1232,147 @@ pub(super) mod tests { } #[test] - fn test_ensure_schema_column_guard() { - use std::sync::atomic::{AtomicU64, Ordering}; - static COUNTER: AtomicU64 = AtomicU64::new(3000); + fn test_ensure_schema_migrates_v17_to_v18_in_place() { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("hcom.db"); + { + let conn = Connection::open(&db_path).unwrap(); + conn.execute_batch( + "CREATE TABLE events (id INTEGER PRIMARY KEY, timestamp TEXT, type TEXT, instance TEXT, data TEXT); + CREATE TABLE instances (name TEXT PRIMARY KEY, tool TEXT DEFAULT 'claude', created_at REAL NOT NULL, launch_context TEXT DEFAULT '', terminal_preset_requested TEXT DEFAULT '', terminal_preset_effective TEXT DEFAULT ''); + CREATE TABLE kv (key TEXT PRIMARY KEY, value TEXT); + CREATE TABLE notify_endpoints (instance TEXT, kind TEXT, port INTEGER, updated_at REAL, PRIMARY KEY(instance, kind)); + CREATE TABLE session_bindings (session_id TEXT PRIMARY KEY, instance_name TEXT NOT NULL, created_at REAL NOT NULL); + PRAGMA user_version = 17; + INSERT INTO instances (name, tool, created_at) VALUES ('luna', 'claude', 1.0);", + ) + .unwrap(); + } - let temp_dir = std::env::temp_dir(); - let test_id = COUNTER.fetch_add(1, Ordering::Relaxed); - let db_path = temp_dir.join(format!( - "test_hcom_colguard_{}_{}.db", - std::process::id(), - test_id - )); + let mut db = HcomDb::open_raw(&db_path).unwrap(); + db.ensure_schema().unwrap(); + + let (version, name, principal): (i32, String, Option) = ( + db.conn + .query_row("PRAGMA user_version", [], |row| row.get(0)) + .unwrap(), + db.conn + .query_row("SELECT name FROM instances", [], |row| row.get(0)) + .unwrap(), + db.conn + .query_row("SELECT principal FROM instances", [], |row| row.get(0)) + .unwrap(), + ); + assert_eq!(version, 18); + assert_eq!(name, "luna"); + assert_eq!( + principal, None, + "migration must not invent identity for old rows" + ); + + cleanup_test_db(db_path); + } + + #[test] + fn test_existing_v18_principal_rows_are_preserved_verbatim() { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("hcom.db"); + { + let conn = Connection::open(&db_path).unwrap(); + conn.execute_batch( + "CREATE TABLE events (id INTEGER PRIMARY KEY, timestamp TEXT, type TEXT, instance TEXT, data TEXT); + CREATE TABLE instances (name TEXT PRIMARY KEY, tool TEXT DEFAULT 'claude', created_at REAL NOT NULL, launch_context TEXT DEFAULT '', terminal_preset_requested TEXT DEFAULT '', terminal_preset_effective TEXT DEFAULT '', principal TEXT); + CREATE TABLE kv (key TEXT PRIMARY KEY, value TEXT); + CREATE TABLE notify_endpoints (instance TEXT, kind TEXT, port INTEGER, updated_at REAL, PRIMARY KEY(instance, kind)); + CREATE TABLE session_bindings (session_id TEXT PRIMARY KEY, instance_name TEXT NOT NULL, created_at REAL NOT NULL); + CREATE TABLE principal_bindings (principal TEXT PRIMARY KEY, instance_name TEXT NOT NULL, epoch INTEGER NOT NULL DEFAULT 0, created_at REAL NOT NULL, updated_at REAL NOT NULL); + PRAGMA user_version = 18; + INSERT INTO instances (name, tool, created_at, principal) VALUES ('luna', 'claude', 1.0, 'p-existing'); + INSERT INTO principal_bindings (principal, instance_name, epoch, created_at, updated_at) VALUES ('p-existing', 'luna', 7, 2.0, 3.0);", + ) + .unwrap(); + } + + let mut db = HcomDb::open_raw(&db_path).unwrap(); + db.ensure_schema().unwrap(); + let row: (String, String, i64, f64, f64) = db + .conn + .query_row( + "SELECT principal, instance_name, epoch, created_at, updated_at FROM principal_bindings", + [], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?, row.get(4)?)), + ) + .unwrap(); + assert_eq!(row, ("p-existing".into(), "luna".into(), 7, 2.0, 3.0)); + + // An older writer that names only its known columns remains compatible. + db.conn + .execute( + "INSERT INTO instances (name, tool, created_at) VALUES ('nova', 'codex', 4.0)", + [], + ) + .unwrap(); + let nova: Option = db + .conn + .query_row( + "SELECT principal FROM instances WHERE name='nova'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(nova, None); + + cleanup_test_db(db_path); + } + + #[test] + fn test_malformed_v18_fails_without_archiving_or_deleting_rows() { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("hcom.db"); + { + let conn = Connection::open(&db_path).unwrap(); + conn.execute_batch( + "CREATE TABLE events (id INTEGER PRIMARY KEY, timestamp TEXT, type TEXT, instance TEXT, data TEXT); + CREATE TABLE instances (name TEXT PRIMARY KEY, tool TEXT DEFAULT 'claude', created_at REAL NOT NULL, launch_context TEXT DEFAULT '', terminal_preset_requested TEXT DEFAULT '', terminal_preset_effective TEXT DEFAULT '', principal TEXT); + CREATE TABLE kv (key TEXT PRIMARY KEY, value TEXT); + CREATE TABLE notify_endpoints (instance TEXT, kind TEXT, port INTEGER, updated_at REAL, PRIMARY KEY(instance, kind)); + CREATE TABLE session_bindings (session_id TEXT PRIMARY KEY, instance_name TEXT NOT NULL, created_at REAL NOT NULL); + PRAGMA user_version = 18; + INSERT INTO instances (name, tool, created_at, principal) VALUES ('luna', 'claude', 1.0, 'p-existing');", + ) + .unwrap(); + } + + let mut db = HcomDb::open_raw(&db_path).unwrap(); + let err = db.ensure_schema().unwrap_err(); + assert!(err.to_string().contains("principal_bindings"), "{err:#}"); + assert_eq!( + db.conn + .query_row("SELECT name FROM instances", [], |row| row + .get::<_, String>(0)) + .unwrap(), + "luna" + ); + assert!(!db_path.parent().unwrap().join("archive").exists()); + + cleanup_test_db(db_path); + } + + #[test] + fn test_ensure_schema_column_guard() { + let temp_dir = tempfile::tempdir().unwrap(); + let db_path = temp_dir.path().join("hcom.db"); // Create a DB at current version but missing 'tool' column { let conn = Connection::open(&db_path).unwrap(); conn.execute_batch(&format!( "CREATE TABLE events (id INTEGER PRIMARY KEY, timestamp TEXT, type TEXT, instance TEXT, data TEXT); - CREATE TABLE instances (name TEXT PRIMARY KEY, created_at REAL NOT NULL); + CREATE TABLE instances (name TEXT PRIMARY KEY, created_at REAL NOT NULL, principal TEXT); CREATE TABLE kv (key TEXT PRIMARY KEY, value TEXT); CREATE TABLE notify_endpoints (instance TEXT, kind TEXT, port INTEGER, updated_at REAL, PRIMARY KEY(instance, kind)); CREATE TABLE session_bindings (session_id TEXT PRIMARY KEY, instance_name TEXT NOT NULL, created_at REAL NOT NULL); + CREATE TABLE principal_bindings (principal TEXT PRIMARY KEY, instance_name TEXT NOT NULL, epoch INTEGER NOT NULL DEFAULT 0, created_at REAL NOT NULL, updated_at REAL NOT NULL); PRAGMA user_version = {};", SCHEMA_VERSION )) @@ -1176,21 +1389,16 @@ pub(super) mod tests { _ => panic!("Expected NeedsArchive for missing tool column"), } - // ensure_schema should fix it - db.ensure_schema().unwrap(); - - let version: i32 = db - .conn - .query_row("PRAGMA user_version", [], |row| row.get(0)) - .unwrap(); - assert_eq!(version, SCHEMA_VERSION); + // Once durable principal data exists, current-version damage is never + // repaired by archiving the database. + assert!(db.ensure_schema().is_err()); + assert!(!db_path.parent().unwrap().join("archive").exists()); cleanup_test_db(db_path); } - /// Regression test for issue #16: init_db() stamped user_version=17 without - /// actually adding the terminal_preset_* columns. ensure_schema must repair - /// this via migration instead of archiving (which would lose data). + /// Regression test for issue #16's one-version repair path: a database + /// stamped one version behind gains only the latest principal schema in place. #[test] fn test_ensure_schema_repairs_stamped_but_not_migrated_db() { use std::sync::atomic::{AtomicU64, Ordering}; @@ -1204,9 +1412,7 @@ pub(super) mod tests { test_id )); - // Simulate the bug: create a v16-style DB but stamp it as v17 - // (this is what init_db() did — CREATE IF NOT EXISTS is a no-op on - // existing tables, then it unconditionally set user_version = 17) + // Complete v17 shape, without v18's principal column/table. { let conn = Connection::open(&db_path).unwrap(); conn.execute_batch( @@ -1238,6 +1444,8 @@ pub(super) mod tests { subagent_timeout INTEGER, tool TEXT DEFAULT 'claude', launch_args TEXT DEFAULT '', + terminal_preset_requested TEXT DEFAULT '', + terminal_preset_effective TEXT DEFAULT '', idle_since TEXT DEFAULT '', pid INTEGER DEFAULT NULL, launch_context TEXT DEFAULT '' @@ -1257,7 +1465,7 @@ pub(super) mod tests { .unwrap(); } - // Verify columns are missing before repair + // Verify the latest column is missing before repair. { let conn = Connection::open(&db_path).unwrap(); let cols: Vec = conn @@ -1268,7 +1476,7 @@ pub(super) mod tests { .filter_map(|r| r.ok()) .collect(); assert!( - !cols.contains(&"terminal_preset_requested".to_string()), + !cols.contains(&"principal".to_string()), "column should be missing before repair" ); } @@ -1283,7 +1491,7 @@ pub(super) mod tests { .unwrap(); assert_eq!(version, SCHEMA_VERSION); - // Columns should now exist + // Latest column/table should now exist. let cols: Vec = db .conn .prepare("PRAGMA table_info(instances)") @@ -1293,12 +1501,8 @@ pub(super) mod tests { .filter_map(|r| r.ok()) .collect(); assert!( - cols.contains(&"terminal_preset_requested".to_string()), - "terminal_preset_requested column should exist after repair" - ); - assert!( - cols.contains(&"terminal_preset_effective".to_string()), - "terminal_preset_effective column should exist after repair" + cols.contains(&"principal".to_string()), + "principal column should exist after repair" ); // Test data should have survived (not archived) diff --git a/src/db/principals.rs b/src/db/principals.rs new file mode 100644 index 00000000..9a3d027d --- /dev/null +++ b/src/db/principals.rs @@ -0,0 +1,790 @@ +//! Durable principal binding primitives. + +use anyhow::{Result, bail}; +use rusqlite::{OptionalExtension, params}; + +use super::HcomDb; +use crate::shared::time::now_epoch_f64; + +/// Exact principal lookup result. `Unresolved` retains only the recorded target; +/// it never guesses another instance from name or recency. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum PrincipalLookup { + Resolved { + instance_name: String, + session_id: Option, + }, + Unresolved { + instance_name: String, + }, + /// One or more instance rows claim the principal, but the authoritative + /// binding row is absent. Claims are diagnostic only and never routable. + MissingBinding { + claiming_instances: Vec, + }, + Unknown, +} + +impl PrincipalLookup { + pub fn instance_name(&self) -> Option<&str> { + match self { + Self::Resolved { instance_name, .. } | Self::Unresolved { instance_name } => { + Some(instance_name) + } + Self::MissingBinding { .. } => None, + Self::Unknown => None, + } + } + + pub fn is_unknown(&self) -> bool { + matches!(self, Self::Unknown) + } + + pub fn is_unresolved(&self) -> bool { + matches!(self, Self::Unresolved { .. } | Self::MissingBinding { .. }) + } +} + +impl HcomDb { + /// Atomically attach one immutable principal to one existing instance. + /// Replaying the same pair is a no-op; every attempted reassignment fails. + pub fn create_principal_binding(&self, principal: &str, instance_name: &str) -> Result { + if principal.is_empty() || instance_name.is_empty() { + bail!("principal and instance_name must not be empty"); + } + + self.conn.execute_batch("BEGIN IMMEDIATE")?; + let result = (|| -> Result { + let binding: Option = self + .conn + .query_row( + "SELECT instance_name FROM principal_bindings WHERE principal=?", + params![principal], + |row| row.get(0), + ) + .optional()?; + let column: Option> = self + .conn + .query_row( + "SELECT principal FROM instances WHERE name=?", + params![instance_name], + |row| row.get(0), + ) + .optional()?; + + let Some(column) = column else { + bail!("cannot bind principal to missing instance '{instance_name}'"); + }; + let foreign_claim: Option = self + .conn + .query_row( + "SELECT name FROM instances WHERE principal=? AND name!=? LIMIT 1", + params![principal, instance_name], + |row| row.get(0), + ) + .optional()?; + if let Some(claiming_instance) = foreign_claim { + bail!( + "principal '{principal}' is already claimed by instance '{claiming_instance}' without a consistent binding" + ); + } + if let Some(bound) = binding { + if bound == instance_name && column.as_deref() == Some(principal) { + return Ok(false); + } + bail!("principal '{principal}' is already bound to '{bound}'"); + } + if let Some(existing) = column.filter(|value| !value.is_empty()) { + bail!("instance '{instance_name}' already carries principal '{existing}'"); + } + + let now = now_epoch_f64(); + self.conn.execute( + "INSERT INTO principal_bindings (principal, instance_name, epoch, created_at, updated_at) + VALUES (?, ?, 0, ?, ?)", + params![principal, instance_name, now, now], + )?; + let updated = self.conn.execute( + "UPDATE instances SET principal=? WHERE name=? AND principal IS NULL", + params![principal, instance_name], + )?; + if updated != 1 { + bail!("instance '{instance_name}' changed while assigning principal"); + } + Ok(true) + })(); + + match result { + Ok(created) => { + self.conn.execute_batch("COMMIT")?; + Ok(created) + } + Err(error) => { + let _ = self.conn.execute_batch("ROLLBACK"); + Err(error) + } + } + } + + pub fn principal_for_instance(&self, instance_name: &str) -> Result> { + Ok(self + .conn + .query_row( + "SELECT principal FROM instances WHERE name=?", + params![instance_name], + |row| row.get::<_, Option>(0), + ) + .optional()? + .flatten() + .filter(|value| !value.is_empty())) + } + + /// Resolve the principal eligible for a tracked `hcom resume `. + /// + /// A current instance row is authoritative only when its principal column + /// and durable binding agree. After the row has been removed, only the + /// uniquely newest binding for the name is eligible. Ambiguous or damaged + /// state fails closed instead of guessing an older lifecycle. + pub(crate) fn principal_for_tracked_resume(&self, instance_name: &str) -> Result { + let current: Option> = self + .conn + .query_row( + "SELECT principal FROM instances WHERE name=?", + params![instance_name], + |row| row.get(0), + ) + .optional()?; + + if let Some(column) = current { + let principal = column.filter(|value| !value.is_empty()).ok_or_else(|| { + anyhow::anyhow!( + "tracked resume identity for '{instance_name}' is missing its principal" + ) + })?; + let bound_name: Option = self + .conn + .query_row( + "SELECT instance_name FROM principal_bindings WHERE principal=?", + params![principal], + |row| row.get(0), + ) + .optional()?; + if bound_name.as_deref() != Some(instance_name) { + bail!( + "tracked resume identity for '{instance_name}' has an inconsistent principal binding" + ); + } + let foreign_claim: bool = self.conn.query_row( + "SELECT EXISTS(SELECT 1 FROM instances WHERE principal=? AND name!=?)", + params![principal, instance_name], + |row| row.get(0), + )?; + if foreign_claim { + bail!( + "tracked resume identity for '{instance_name}' is claimed by another instance" + ); + } + return Ok(principal); + } + + let mut stmt = self.conn.prepare( + "SELECT principal, created_at FROM principal_bindings + WHERE instance_name=? + ORDER BY created_at DESC, principal ASC LIMIT 2", + )?; + let rows: Vec<(String, f64)> = stmt + .query_map(params![instance_name], |row| Ok((row.get(0)?, row.get(1)?)))? + .collect::>()?; + let Some((principal, newest_at)) = rows.first() else { + bail!("tracked resume identity for '{instance_name}' has no durable binding"); + }; + if principal.is_empty() { + bail!("tracked resume identity for '{instance_name}' has an empty principal"); + } + if rows.get(1).is_some_and(|(_, next_at)| next_at == newest_at) { + bail!( + "tracked resume identity for '{instance_name}' is ambiguous: newest bindings are tied" + ); + } + let lifecycle: Option<(Option, Option, Option)> = self + .conn + .query_row( + "SELECT json_extract(data, '$.snapshot.principal'), + json_extract(data, '$.snapshot.created_at'), + (julianday(timestamp) - 2440587.5) * 86400.0 + FROM events + WHERE type='life' + AND instance=? + AND json_extract(data, '$.action')='stopped' + ORDER BY id DESC LIMIT 1", + params![instance_name], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ) + .optional()?; + let Some((snapshot_principal, Some(lifecycle_created_at), Some(stopped_at))) = lifecycle + else { + bail!( + "tracked resume identity for '{instance_name}' has no complete stopped lifecycle evidence" + ); + }; + if let Some(snapshot_principal) = snapshot_principal.filter(|value| !value.is_empty()) { + if snapshot_principal != *principal { + bail!( + "tracked resume identity for '{instance_name}' does not match its latest stopped lifecycle" + ); + } + } else { + // Pre-P2 stopped snapshots do not carry the principal. Bound the + // newest binding to that lifecycle's creation/stop interval so a + // later crashed reclaim cannot be paired with an older transcript. + // SQLite's julianday conversion is millisecond-granular on some + // versions, hence the representation-only tolerance. + const TIMESTAMP_EPSILON_SECS: f64 = 0.001; + if *newest_at + TIMESTAMP_EPSILON_SECS < lifecycle_created_at + || *newest_at > stopped_at + TIMESTAMP_EPSILON_SECS + { + bail!( + "tracked resume identity for '{instance_name}' does not belong to its latest stopped lifecycle" + ); + } + } + if stopped_at + 0.001 < lifecycle_created_at { + bail!( + "tracked resume identity for '{instance_name}' has invalid stopped lifecycle timing" + ); + } + let foreign_claim: bool = self.conn.query_row( + "SELECT EXISTS(SELECT 1 FROM instances WHERE principal=? AND name!=?)", + params![principal, instance_name], + |row| row.get(0), + )?; + if foreign_claim { + bail!("tracked resume identity for '{instance_name}' is claimed by another instance"); + } + Ok(principal.clone()) + } + + /// Atomically recheck and reserve a tracked resume lifecycle. + /// + /// The pending row prevents a concurrent name reclaim after the eligibility + /// check. The durable binding is reused, not rewritten, so a later launch + /// failure can remove only the reservation while keeping resume evidence. + pub(crate) fn reserve_tracked_resume_principal( + &self, + instance_name: &str, + expected_principal: &str, + reservation_id: &str, + ) -> Result<()> { + if reservation_id.is_empty() { + bail!("tracked resume reservation id must not be empty"); + } + self.conn.execute_batch("BEGIN IMMEDIATE")?; + let result = (|| -> Result<()> { + let current = self.principal_for_tracked_resume(instance_name)?; + if current != expected_principal { + bail!("tracked resume identity for '{instance_name}' changed before launch"); + } + + let status: Option = self + .conn + .query_row( + "SELECT status FROM instances WHERE name=?", + params![instance_name], + |row| row.get(0), + ) + .optional()?; + if status.as_deref().is_some_and(|value| value != "inactive") { + bail!("tracked resume identity for '{instance_name}' changed before launch"); + } + + self.conn + .execute("DELETE FROM instances WHERE name=?", params![instance_name])?; + self.conn.execute( + "INSERT INTO instances + (name, principal, status, status_context, created_at, launch_context) + VALUES (?, ?, 'pending', 'new', ?, ?)", + params![ + instance_name, + expected_principal, + now_epoch_f64(), + serde_json::json!({"resume_reservation_id": reservation_id}).to_string() + ], + )?; + Ok(()) + })(); + + match result { + Ok(()) => { + self.conn.execute_batch("COMMIT")?; + Ok(()) + } + Err(error) => { + let _ = self.conn.execute_batch("ROLLBACK"); + Err(error) + } + } + } + + /// Claim exactly the tracked-resume reservation created for this process. + /// + /// Rechecking the row, durable binding, and reservation token in one write + /// transaction prevents an older launch attempt from attaching to a newer + /// retry that retained the same principal. + pub(crate) fn claim_tracked_resume_principal( + &self, + instance_name: &str, + expected_principal: &str, + reservation_id: &str, + ) -> Result<()> { + self.conn.execute_batch("BEGIN IMMEDIATE")?; + let result = (|| -> Result<()> { + let reserved: Option<(Option, String, Option)> = self + .conn + .query_row( + "SELECT principal, status, + CASE WHEN json_valid(launch_context) + THEN json_extract(launch_context, '$.resume_reservation_id') + ELSE NULL END + FROM instances WHERE name=?", + params![instance_name], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ) + .optional()?; + let Some((principal, status, token)) = reserved else { + bail!("tracked resume reservation for '{instance_name}' disappeared before launch"); + }; + if principal.as_deref() != Some(expected_principal) + || status != "pending" + || token.as_deref() != Some(reservation_id) + { + bail!("tracked resume reservation for '{instance_name}' changed before launch"); + } + + let bound_name: Option = self + .conn + .query_row( + "SELECT instance_name FROM principal_bindings WHERE principal=?", + params![expected_principal], + |row| row.get(0), + ) + .optional()?; + if bound_name.as_deref() != Some(instance_name) { + bail!( + "tracked resume identity for '{instance_name}' has an inconsistent principal binding" + ); + } + let foreign_claim: bool = self.conn.query_row( + "SELECT EXISTS(SELECT 1 FROM instances WHERE principal=? AND name!=?)", + params![expected_principal, instance_name], + |row| row.get(0), + )?; + if foreign_claim { + bail!( + "tracked resume identity for '{instance_name}' is claimed by another instance" + ); + } + + self.set_process_binding(reservation_id, "", instance_name)?; + Ok(()) + })(); + + match result { + Ok(()) => { + self.conn.execute_batch("COMMIT")?; + Ok(()) + } + Err(error) => { + let _ = self.conn.execute_batch("ROLLBACK"); + Err(error) + } + } + } + + /// Remove only the reservation owned by this failed tracked-resume launch. + pub(crate) fn cleanup_tracked_resume_reservation( + &self, + instance_name: &str, + expected_principal: &str, + reservation_id: &str, + ) -> Result { + self.conn.execute_batch("BEGIN IMMEDIATE")?; + let result = (|| -> Result { + let deleted = self.conn.execute( + "DELETE FROM instances + WHERE name=? AND principal=? + AND CASE WHEN json_valid(launch_context) + THEN json_extract(launch_context, '$.resume_reservation_id') + ELSE NULL END = ?", + params![instance_name, expected_principal, reservation_id], + )?; + self.conn.execute( + "DELETE FROM process_bindings WHERE process_id=?", + params![reservation_id], + )?; + Ok(deleted == 1) + })(); + + match result { + Ok(deleted) => { + self.conn.execute_batch("COMMIT")?; + Ok(deleted) + } + Err(error) => { + let _ = self.conn.execute_batch("ROLLBACK"); + Err(error) + } + } + } + + /// Roll back an identity minted for a launch that never became live. + /// Normal stop paths deliberately do not call this: once a launch succeeds, + /// its durable binding survives instance-row deletion for exact lookup. + pub(crate) fn rollback_provisional_principal_binding( + &self, + principal: &str, + instance_name: &str, + ) -> Result { + let tx = self.conn.unchecked_transaction()?; + let deleted = tx.execute( + "DELETE FROM principal_bindings WHERE principal=? AND instance_name=?", + params![principal, instance_name], + )?; + let cleared = tx.execute( + "UPDATE instances SET principal=NULL WHERE name=? AND principal=?", + params![instance_name, principal], + )?; + tx.commit()?; + Ok(deleted > 0 || cleared > 0) + } + + /// Resolve only the stored principal row and its exact claimed instance. + pub fn lookup_principal(&self, principal: &str) -> Result { + let row: Option<(String, Option, Option, bool)> = self + .conn + .query_row( + "SELECT b.instance_name, i.principal, i.session_id, + EXISTS(SELECT 1 FROM instances other + WHERE other.principal=b.principal AND other.name!=b.instance_name) + FROM principal_bindings b LEFT JOIN instances i ON i.name=b.instance_name + WHERE b.principal=?", + params![principal], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?, row.get(3)?)), + ) + .optional()?; + if row.is_none() { + let claiming_instances: Vec = self + .conn + .prepare("SELECT name FROM instances WHERE principal=? ORDER BY name")? + .query_map(params![principal], |row| row.get(0))? + .collect::>()?; + return Ok(if claiming_instances.is_empty() { + PrincipalLookup::Unknown + } else { + PrincipalLookup::MissingBinding { claiming_instances } + }); + } + Ok(match row { + Some((instance_name, Some(column), session_id, false)) if column == principal => { + PrincipalLookup::Resolved { + instance_name, + session_id: session_id.filter(|value| !value.is_empty()), + } + } + Some((instance_name, _, _, _)) => PrincipalLookup::Unresolved { instance_name }, + None => unreachable!("missing binding handled above"), + }) + } +} + +#[cfg(test)] +mod tests { + use super::super::tests::{cleanup_test_db, setup_full_test_db}; + use rusqlite::params; + + fn insert_instance(db: &super::super::HcomDb, name: &str) { + db.conn() + .execute( + "INSERT INTO instances (name, created_at) VALUES (?, 1.0)", + params![name], + ) + .unwrap(); + } + + fn insert_stopped_lifecycle( + db: &super::super::HcomDb, + name: &str, + created_at: f64, + stopped_at: &str, + ) { + let data = serde_json::json!({ + "action": "stopped", + "snapshot": {"created_at": created_at} + }); + db.conn() + .execute( + "INSERT INTO events (timestamp, type, instance, data) + VALUES (?, 'life', ?, ?)", + params![stopped_at, name, data.to_string()], + ) + .unwrap(); + } + + #[test] + fn create_binding_is_atomic_idempotent_and_conflict_safe() { + let (db, path) = setup_full_test_db(); + insert_instance(&db, "luna"); + insert_instance(&db, "nova"); + + assert!(db.create_principal_binding("p-1", "luna").unwrap()); + assert!(!db.create_principal_binding("p-1", "luna").unwrap()); + assert!(db.create_principal_binding("p-1", "nova").is_err()); + assert!(db.create_principal_binding("p-2", "luna").is_err()); + + let row: (String, String, i64) = db + .conn() + .query_row( + "SELECT principal, instance_name, epoch FROM principal_bindings", + [], + |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)), + ) + .unwrap(); + assert_eq!(row, ("p-1".into(), "luna".into(), 0)); + let column: Option = db + .conn() + .query_row( + "SELECT principal FROM instances WHERE name='luna'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(column.as_deref(), Some("p-1")); + + cleanup_test_db(path); + } + + #[test] + fn create_binding_rolls_back_when_second_write_fails() { + let (db, path) = setup_full_test_db(); + insert_instance(&db, "luna"); + db.conn() + .execute_batch( + "CREATE TRIGGER reject_principal BEFORE UPDATE OF principal ON instances + BEGIN SELECT RAISE(ABORT, 'reject principal'); END;", + ) + .unwrap(); + + assert!(db.create_principal_binding("p-1", "luna").is_err()); + let bindings: i64 = db + .conn() + .query_row("SELECT COUNT(*) FROM principal_bindings", [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(bindings, 0); + + cleanup_test_db(path); + } + + #[test] + fn principal_lookup_is_exact_and_reports_unresolved_state() { + let (db, path) = setup_full_test_db(); + insert_instance(&db, "luna"); + db.create_principal_binding("p-1", "luna").unwrap(); + + let found = db.lookup_principal("p-1").unwrap(); + assert_eq!(found.instance_name(), Some("luna")); + assert!(db.lookup_principal("p-missing").unwrap().is_unknown()); + + db.conn() + .execute("DELETE FROM instances WHERE name='luna'", []) + .unwrap(); + let unresolved = db.lookup_principal("p-1").unwrap(); + assert_eq!(unresolved.instance_name(), Some("luna")); + assert!(unresolved.is_unresolved()); + + cleanup_test_db(path); + } + + #[test] + fn principal_lookup_reports_missing_binding_as_unresolved_without_guessing() { + let (db, path) = setup_full_test_db(); + insert_instance(&db, "luna"); + db.create_principal_binding("p-1", "luna").unwrap(); + db.conn() + .execute("DELETE FROM principal_bindings WHERE principal='p-1'", []) + .unwrap(); + + let missing_binding = db.lookup_principal("p-1").unwrap(); + assert!(missing_binding.is_unresolved()); + assert_eq!( + missing_binding.instance_name(), + None, + "an instance-side claim is diagnostic evidence, not a routing target" + ); + + cleanup_test_db(path); + } + + #[test] + fn create_binding_rejects_foreign_instance_claim_and_preserves_state() { + let (db, path) = setup_full_test_db(); + insert_instance(&db, "luna"); + insert_instance(&db, "nova"); + db.conn() + .execute("UPDATE instances SET principal='p-1' WHERE name='luna'", []) + .unwrap(); + + assert!(db.create_principal_binding("p-1", "nova").is_err()); + let bindings: i64 = db + .conn() + .query_row("SELECT COUNT(*) FROM principal_bindings", [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(bindings, 0); + assert_eq!( + db.principal_for_instance("luna").unwrap().as_deref(), + Some("p-1") + ); + assert_eq!(db.principal_for_instance("nova").unwrap(), None); + + cleanup_test_db(path); + } + + #[test] + fn tracked_resume_uses_consistent_live_principal() { + let (db, path) = setup_full_test_db(); + insert_instance(&db, "luna"); + db.create_principal_binding("p-live", "luna").unwrap(); + + assert_eq!(db.principal_for_tracked_resume("luna").unwrap(), "p-live"); + + cleanup_test_db(path); + } + + #[test] + fn tracked_resume_uses_only_uniquely_newest_stopped_binding() { + let (db, path) = setup_full_test_db(); + insert_instance(&db, "luna"); + db.create_principal_binding("p-old", "luna").unwrap(); + db.delete_instance("luna").unwrap(); + db.conn() + .execute( + "INSERT INTO principal_bindings + (principal, instance_name, epoch, created_at, updated_at) + VALUES ('p-new', 'luna', 0, 2.0, 2.0)", + [], + ) + .unwrap(); + db.conn() + .execute( + "UPDATE principal_bindings SET created_at=1.0, updated_at=1.0 + WHERE principal='p-old'", + [], + ) + .unwrap(); + insert_stopped_lifecycle(&db, "luna", 1.5, "1970-01-01T00:00:04Z"); + + assert_eq!(db.principal_for_tracked_resume("luna").unwrap(), "p-new"); + + cleanup_test_db(path); + } + + #[test] + fn tracked_resume_rejects_missing_corrupt_and_tied_identity() { + let (db, path) = setup_full_test_db(); + assert!(db.principal_for_tracked_resume("missing").is_err()); + + insert_instance(&db, "luna"); + db.conn() + .execute( + "UPDATE instances SET principal='p-without-binding' WHERE name='luna'", + [], + ) + .unwrap(); + assert!(db.principal_for_tracked_resume("luna").is_err()); + + db.delete_instance("luna").unwrap(); + db.conn() + .execute_batch( + "INSERT INTO principal_bindings + (principal, instance_name, epoch, created_at, updated_at) + VALUES ('p-a', 'luna', 0, 3.0, 3.0), + ('p-b', 'luna', 0, 3.0, 3.0);", + ) + .unwrap(); + assert!(db.principal_for_tracked_resume("luna").is_err()); + + cleanup_test_db(path); + } + + #[test] + fn tracked_resume_reservation_rechecks_reclaim_and_keeps_binding() { + let (db, path) = setup_full_test_db(); + insert_instance(&db, "luna"); + db.create_principal_binding("p-old", "luna").unwrap(); + db.delete_instance("luna").unwrap(); + + insert_instance(&db, "luna"); + db.create_principal_binding("p-reclaimed", "luna").unwrap(); + + let error = db + .reserve_tracked_resume_principal("luna", "p-old", "proc-stale") + .unwrap_err() + .to_string(); + assert!(error.contains("changed"), "unexpected error: {error}"); + assert_eq!( + db.principal_for_instance("luna").unwrap().as_deref(), + Some("p-reclaimed") + ); + assert_eq!( + db.lookup_principal("p-old").unwrap().instance_name(), + Some("luna") + ); + + cleanup_test_db(path); + } + + #[test] + fn tracked_resume_never_pairs_old_snapshot_with_newer_unstopped_binding() { + let (db, path) = setup_full_test_db(); + db.conn() + .execute_batch( + "INSERT INTO principal_bindings + (principal, instance_name, epoch, created_at, updated_at) + VALUES ('p-old', 'luna', 0, 1.0, 1.0), + ('p-crashed-reclaim', 'luna', 0, 3.0, 3.0);", + ) + .unwrap(); + insert_stopped_lifecycle(&db, "luna", 0.5, "1970-01-01T00:00:02Z"); + + let error = db + .principal_for_tracked_resume("luna") + .unwrap_err() + .to_string(); + assert!(error.contains("lifecycle"), "unexpected error: {error}"); + + cleanup_test_db(path); + } + + #[test] + fn tracked_resume_after_stopped_reclaim_keeps_reclaimed_principal() { + let (db, path) = setup_full_test_db(); + db.conn() + .execute_batch( + "INSERT INTO principal_bindings + (principal, instance_name, epoch, created_at, updated_at) + VALUES ('p-old', 'luna', 0, 1.0, 1.0), + ('p-reclaimed', 'luna', 0, 3.0, 3.0);", + ) + .unwrap(); + insert_stopped_lifecycle(&db, "luna", 2.5, "1970-01-01T00:00:04Z"); + + assert_eq!( + db.principal_for_tracked_resume("luna").unwrap(), + "p-reclaimed" + ); + + cleanup_test_db(path); + } +} diff --git a/src/hooks/claude.rs b/src/hooks/claude.rs index 1cebccdb..de8b1228 100644 --- a/src/hooks/claude.rs +++ b/src/hooks/claude.rs @@ -1634,6 +1634,15 @@ fn ensure_subagent_row(db: &HcomDb, parent_session_id: &str, agent_id: &str, age match instance_names::allocate_subagent_instance(db, &alloc) { Ok(name) => { + if let Err(error) = ensure_subagent_principal(db, &name) { + let _ = db.delete_instance(&name); + log::log_warn( + "hooks", + "subagent.principal_failed", + &format!("name={name} agent_id={agent_id} err={error}"), + ); + return; + } log::log_info( "hooks", "subagent.row.ensured", @@ -1653,6 +1662,14 @@ fn ensure_subagent_row(db: &HcomDb, parent_session_id: &str, agent_id: &str, age } } +pub(crate) fn ensure_subagent_principal(db: &HcomDb, instance_name: &str) -> anyhow::Result<()> { + if db.principal_for_instance(instance_name)?.is_some() { + return Ok(()); + } + db.create_principal_binding(&crate::launcher::generate_principal_id(), instance_name)?; + Ok(()) +} + /// SubagentStop: message polling using agent_id, cleanup on exit. /// /// Returns (exit_code, stdout). exit_code=2 means message delivered @@ -2652,6 +2669,48 @@ mod tests { .unwrap() } + #[test] + fn dormant_subagent_gets_one_flat_principal_across_replay() { + let (_dir, db) = make_delivery_test_db(); + db.conn() + .execute( + "UPDATE instances SET session_id='sess-parent' WHERE name='nova'", + [], + ) + .unwrap(); + db.rebind_session("sess-parent", "nova").unwrap(); + db.create_principal_binding("p-parent", "nova").unwrap(); + + ensure_subagent_row(&db, "sess-parent", "agent-77", "general"); + let first: String = db + .conn() + .query_row( + "SELECT principal FROM instances WHERE agent_id='agent-77'", + [], + |row| row.get(0), + ) + .unwrap(); + ensure_subagent_row(&db, "sess-parent", "agent-77", "general"); + let second: String = db + .conn() + .query_row( + "SELECT principal FROM instances WHERE agent_id='agent-77'", + [], + |row| row.get(0), + ) + .unwrap(); + + assert_eq!( + first, second, + "SubagentStart replay must not rotate identity" + ); + assert_ne!( + first, "p-parent", + "child identity is not derived or inherited" + ); + assert!(!first.contains("agent-77")); + } + struct FailingWriter; impl std::io::Write for FailingWriter { diff --git a/src/hooks/common.rs b/src/hooks/common.rs index fb521806..8ffaf478 100644 --- a/src/hooks/common.rs +++ b/src/hooks/common.rs @@ -1079,6 +1079,7 @@ fn stop_instance_inner( "origin_device_id": instance_data.origin_device_id, "background_log_file": instance_data.background_log_file, "last_event_id": instance_data.last_event_id, + "principal": instance_data.principal, }); // Clean session bindings + process bindings + stop subagents for this session @@ -1226,6 +1227,7 @@ pub fn soft_finalize_session( "origin_device_id": instance_data.origin_device_id, "background_log_file": instance_data.background_log_file, "last_event_id": instance_data.last_event_id, + "principal": instance_data.principal, }); if let Some(ref session_id) = instance_data.session_id { diff --git a/src/identity.rs b/src/identity.rs index 7c481fdf..b1b1eb38 100644 --- a/src/identity.rs +++ b/src/identity.rs @@ -938,6 +938,7 @@ mod tests { fn default_instance() -> InstanceRow { InstanceRow { name: String::new(), + principal: None, session_id: None, parent_session_id: None, parent_name: None, diff --git a/src/instance_binding.rs b/src/instance_binding.rs index 77f442e8..58ce2105 100644 --- a/src/instance_binding.rs +++ b/src/instance_binding.rs @@ -874,6 +874,20 @@ pub fn create_orphaned_pty_identity( return None; } + // A fresh orphaned PTY session is a new lifecycle. Mint its durable identity + // before changing any routing bindings so principal failure can leave the + // prior process/session routes untouched. + let principal = crate::launcher::generate_principal_id(); + if let Err(e) = db.create_principal_binding(&principal, &name) { + crate::log::log_error( + "instances", + "create_orphaned_pty_identity.principal", + &e.to_string(), + ); + let _ = db.delete_instance(&name); + return None; + } + if let Err(e) = db.rebind_session(session_id, &name) { eprintln!("[hcom] warn: rebind_session failed for {name}: {e}"); } @@ -1272,6 +1286,17 @@ mod tests { let inst = db.get_instance_full(&name).unwrap().unwrap(); assert_eq!(inst.session_id.as_deref(), Some("sess-orphan")); assert_eq!(inst.tool, "claude"); + let principal = inst + .principal + .as_deref() + .expect("orphaned PTY lifecycle should carry a principal"); + assert_eq!( + db.lookup_principal(principal).unwrap(), + crate::db::PrincipalLookup::Resolved { + instance_name: name.clone(), + session_id: Some("sess-orphan".to_string()), + } + ); assert_eq!( db.get_session_binding("sess-orphan").unwrap(), @@ -1293,11 +1318,84 @@ mod tests { let name = result.unwrap(); let inst = db.get_instance_full(&name).unwrap().unwrap(); assert_eq!(inst.tool, "gemini"); + let principal = inst + .principal + .as_deref() + .expect("orphaned PTY lifecycle should carry a principal"); + assert_eq!( + db.lookup_principal(principal).unwrap(), + crate::db::PrincipalLookup::Resolved { + instance_name: name.clone(), + session_id: Some("sess-orphan2".to_string()), + } + ); assert_eq!(db.get_session_binding("sess-orphan2").unwrap(), Some(name)); cleanup(path); } + #[test] + fn test_create_orphaned_pty_identity_mints_a_new_principal_per_lifecycle() { + crate::config::Config::init(); + let (db, path) = setup_test_db(); + + let first = create_orphaned_pty_identity(&db, "sess-first", None, "claude").unwrap(); + let second = create_orphaned_pty_identity(&db, "sess-second", None, "claude").unwrap(); + + let first_principal = db.principal_for_instance(&first).unwrap().unwrap(); + let second_principal = db.principal_for_instance(&second).unwrap().unwrap(); + assert_ne!(first_principal, second_principal); + + cleanup(path); + } + + #[test] + fn test_create_orphaned_pty_identity_principal_failure_rolls_back_new_identity() { + crate::config::Config::init(); + let (db, path) = setup_test_db(); + db.conn() + .execute( + "INSERT INTO process_bindings (process_id, session_id, instance_name, updated_at)\ + VALUES ('pid-orphan', 'sess-old', 'old-name', 1.0)", + [], + ) + .unwrap(); + db.conn() + .execute_batch( + "CREATE TRIGGER fail_orphan_principal + BEFORE INSERT ON principal_bindings + BEGIN SELECT RAISE(ABORT, 'injected principal failure'); END;", + ) + .unwrap(); + + let result = create_orphaned_pty_identity(&db, "sess-failed", Some("pid-orphan"), "claude"); + + assert_eq!(result, None); + assert_eq!(db.get_session_binding("sess-failed").unwrap(), None); + assert_eq!( + db.get_process_binding_full("pid-orphan").unwrap(), + Some((Some("sess-old".to_string()), "old-name".to_string())) + ); + let new_instances: i64 = db + .conn() + .query_row( + "SELECT COUNT(*) FROM instances WHERE session_id='sess-failed'", + [], + |row| row.get(0), + ) + .unwrap(); + let principals: i64 = db + .conn() + .query_row("SELECT COUNT(*) FROM principal_bindings", [], |row| { + row.get(0) + }) + .unwrap(); + assert_eq!(new_instances, 0); + assert_eq!(principals, 0); + + cleanup(path); + } + #[test] fn test_resolve_from_binding_process_binding() { crate::config::Config::init(); diff --git a/src/instance_lifecycle.rs b/src/instance_lifecycle.rs index da229a78..90384892 100644 --- a/src/instance_lifecycle.rs +++ b/src/instance_lifecycle.rs @@ -736,6 +736,7 @@ mod tests { fn default_instance() -> InstanceRow { InstanceRow { name: String::new(), + principal: None, session_id: None, parent_session_id: None, parent_name: None, diff --git a/src/instances.rs b/src/instances.rs index 5b00e773..2832b8fb 100644 --- a/src/instances.rs +++ b/src/instances.rs @@ -355,6 +355,7 @@ mod tests { fn default_instance() -> InstanceRow { InstanceRow { name: String::new(), + principal: None, session_id: None, parent_session_id: None, parent_name: None, diff --git a/src/launcher.rs b/src/launcher.rs index 45a94b6d..b31bbf68 100644 --- a/src/launcher.rs +++ b/src/launcher.rs @@ -9,7 +9,7 @@ use std::fs; use std::io::Write; use std::path::Path; -use anyhow::{Result, bail}; +use anyhow::{Context, Result, bail}; use rand::RngExt; use serde_json::json; @@ -186,6 +186,9 @@ pub struct LaunchParams { /// Session id being resumed, inherited by the recreated instance row so a /// kill before the tool's first turn (no hook re-bind yet) stays resumable. pub prior_session_id: Option, + /// Existing durable principal retained only by a tracked named resume. + /// All ordinary launch, adoption, fork, and reclaim paths leave this unset. + pub(crate) retained_principal: Option, pub tag: Option, pub system_prompt: Option, pub initial_prompt: Option, @@ -209,6 +212,7 @@ impl Default for LaunchParams { args: Vec::new(), persisted_args: None, prior_session_id: None, + retained_principal: None, tag: None, system_prompt: None, initial_prompt: None, @@ -449,6 +453,50 @@ fn generate_process_id() -> String { format!("{:08x}-{:04x}-{:04x}-{:04x}-{:012x}", a, b, c, d, e) } +pub(crate) fn generate_principal_id() -> String { + uuid::Uuid::new_v4().to_string() +} + +pub(crate) fn attach_launch_principal( + db: &HcomDb, + instance_name: &str, + env: &mut HashMap, +) -> Result { + let principal = generate_principal_id(); + db.create_principal_binding(&principal, instance_name)?; + env.insert("HCOM_PRINCIPAL_ID".to_string(), principal.clone()); + Ok(principal) +} + +fn attach_provisional_launch_identity( + db: &HcomDb, + instance_name: &str, + process_id: &str, + env: &mut HashMap, + retained_principal: Option<&str>, +) -> Result { + if let Some(principal) = retained_principal { + db.claim_tracked_resume_principal(instance_name, principal, process_id)?; + env.insert("HCOM_PRINCIPAL_ID".to_string(), principal.to_string()); + return Ok(principal.to_string()); + } + + if let Err(error) = db.set_process_binding(process_id, "", instance_name) { + let _ = db.delete_instance(instance_name); + return Err(error); + } + let attached = attach_launch_principal(db, instance_name, env); + match attached { + Ok(principal) => Ok(principal), + Err(error) => { + let _ = db.delete_process_binding(process_id); + let _ = db.delete_instance(instance_name); + env.remove("HCOM_PRINCIPAL_ID"); + Err(error) + } + } +} + fn install_diag_context(tool: &LaunchTool, paths: &[(&str, std::path::PathBuf)]) -> String { use std::fmt::Write as _; let mut out = String::new(); @@ -1657,7 +1705,9 @@ pub fn launch(db: &HcomDb, mut params: LaunchParams) -> Result { params.count ); } - resolve_explicit_name_conflict(db, name)?; + if params.retained_principal.is_none() { + resolve_explicit_name_conflict(db, name)?; + } } // System prompt file for Gemini/Codex @@ -1839,7 +1889,11 @@ pub fn launch(db: &HcomDb, mut params: LaunchParams) -> Result { ); // Pre-register instance + let mut provisional_principal = None; if let Err(e) = (|| -> Result<()> { + if let Some(principal) = params.retained_principal.as_deref() { + db.reserve_tracked_resume_principal(&instance_name, principal, &process_id)?; + } instance_binding::initialize_instance_in_position_file( db, &instance_name, @@ -1860,9 +1914,25 @@ pub fn launch(db: &HcomDb, mut params: LaunchParams) -> Result { None, // hints Some(working_dir), // cwd_override: use launch params cwd, not current_dir() ); - db.set_process_binding(&process_id, "", &instance_name)?; + provisional_principal = Some( + attach_provisional_launch_identity( + db, + &instance_name, + &process_id, + &mut instance_env, + params.retained_principal.as_deref(), + ) + .with_context(|| format!("principal setup failed for '{instance_name}'"))?, + ); Ok(()) })() { + cleanup_instance( + db, + &instance_name, + &process_id, + provisional_principal.as_deref(), + params.retained_principal.as_deref(), + ); errors.push(json!({"tool": base_tool, "error": e.to_string()})); continue; } @@ -2252,10 +2322,22 @@ pub fn launch(db: &HcomDb, mut params: LaunchParams) -> Result { match launch_result { Ok(true) => launched += 1, Ok(false) => { - cleanup_instance(db, &instance_name, &process_id); + cleanup_instance( + db, + &instance_name, + &process_id, + provisional_principal.as_deref(), + params.retained_principal.as_deref(), + ); } Err(e) => { - cleanup_instance(db, &instance_name, &process_id); + cleanup_instance( + db, + &instance_name, + &process_id, + provisional_principal.as_deref(), + params.retained_principal.as_deref(), + ); errors.push(json!({"tool": base_tool, "error": e.to_string()})); } } @@ -2356,7 +2438,33 @@ pub(crate) fn validate_tool_args(tool: &LaunchTool, args: &[String]) -> Vec, + retained_principal: Option<&str>, +) { + if let Some(principal) = retained_principal { + if let Err(error) = db.cleanup_tracked_resume_reservation(name, principal, process_id) { + crate::log::log_warn( + "launcher", + "principal.resume_cleanup_failed", + &format!("name={name} err={error}"), + ); + } + return; + } + + if let Some(principal) = provisional_principal + && let Err(error) = db.rollback_provisional_principal_binding(principal, name) + { + crate::log::log_warn( + "launcher", + "principal.rollback_failed", + &format!("name={name} principal={principal} err={error}"), + ); + } db.delete_instance(name).ok(); db.delete_process_binding(process_id).ok(); } @@ -3183,6 +3291,319 @@ mod tests { db } + #[test] + fn launch_principal_is_fresh_per_lifecycle_and_matches_child_env() { + let db = launcher_test_db(); + for name in ["luna", "nova"] { + db.conn() + .execute( + "INSERT INTO instances (name, created_at) VALUES (?, 1.0)", + rusqlite::params![name], + ) + .unwrap(); + } + let mut first_env = HashMap::new(); + let first = attach_launch_principal(&db, "luna", &mut first_env).unwrap(); + let mut second_env = HashMap::new(); + let second = attach_launch_principal(&db, "nova", &mut second_env).unwrap(); + + assert_ne!( + first, second, + "fork/adoption/reclaim launches are new lifecycles" + ); + assert_eq!(first_env.get("HCOM_PRINCIPAL_ID"), Some(&first)); + assert_eq!(second_env.get("HCOM_PRINCIPAL_ID"), Some(&second)); + assert_eq!(db.principal_for_instance("luna").unwrap(), Some(first)); + } + + #[test] + fn provisional_launch_identity_rolls_back_when_process_binding_fails() { + let db = launcher_test_db(); + db.conn() + .execute( + "INSERT INTO instances (name, created_at) VALUES ('luna', 1.0)", + [], + ) + .unwrap(); + db.conn() + .execute_batch( + "CREATE TRIGGER reject_process_binding BEFORE INSERT ON process_bindings + BEGIN SELECT RAISE(ABORT, 'reject process binding'); END;", + ) + .unwrap(); + let mut env = HashMap::new(); + + assert!(attach_provisional_launch_identity(&db, "luna", "proc-1", &mut env, None).is_err()); + assert!(db.get_instance_full("luna").unwrap().is_none()); + assert_eq!( + db.conn() + .query_row("SELECT COUNT(*) FROM principal_bindings", [], |row| row + .get::<_, i64>(0)) + .unwrap(), + 0 + ); + assert!(!env.contains_key("HCOM_PRINCIPAL_ID")); + } + + #[test] + fn failed_backend_cleanup_removes_only_provisional_launch_identity() { + let db = launcher_test_db(); + db.conn() + .execute( + "INSERT INTO instances (name, created_at) VALUES ('luna', 1.0)", + [], + ) + .unwrap(); + let mut env = HashMap::new(); + let principal = attach_launch_principal(&db, "luna", &mut env).unwrap(); + db.set_process_binding("proc-1", "", "luna").unwrap(); + + cleanup_instance(&db, "luna", "proc-1", Some(&principal), None); + + assert!(db.get_instance_full("luna").unwrap().is_none()); + assert_eq!( + db.conn() + .query_row("SELECT COUNT(*) FROM principal_bindings", [], |row| row + .get::<_, i64>(0)) + .unwrap(), + 0 + ); + assert_eq!(db.get_process_binding("proc-1").unwrap(), None); + } + + #[test] + fn tracked_resume_reuses_principal_and_remints_process_binding() { + let db = launcher_test_db(); + db.conn() + .execute( + "INSERT INTO instances (name, status, created_at) + VALUES ('luna', 'inactive', 1.0)", + [], + ) + .unwrap(); + db.create_principal_binding("p-luna", "luna").unwrap(); + + for process_id in ["proc-first", "proc-second"] { + db.reserve_tracked_resume_principal("luna", "p-luna", process_id) + .unwrap(); + assert!(instance_binding::initialize_instance_in_position_file( + &db, + "luna", + Some("session-1"), + None, + None, + None, + None, + Some("codex"), + false, + None, + None, + None, + None, + Some("/tmp"), + )); + let mut env = HashMap::new(); + let principal = attach_provisional_launch_identity( + &db, + "luna", + process_id, + &mut env, + Some("p-luna"), + ) + .unwrap(); + assert_eq!(principal, "p-luna"); + assert_eq!( + env.get("HCOM_PRINCIPAL_ID").map(String::as_str), + Some("p-luna") + ); + assert_eq!( + db.get_process_binding(process_id).unwrap().as_deref(), + Some("luna") + ); + + let created_at = db + .get_instance_full("luna") + .unwrap() + .expect("resumed row") + .created_at; + db.log_event( + "life", + "luna", + &serde_json::json!({ + "action": "stopped", + "snapshot": { + "created_at": created_at, + "principal": "p-luna" + } + }), + ) + .unwrap(); + db.delete_process_binding(process_id).unwrap(); + db.delete_instance("luna").unwrap(); + } + + let binding_count: i64 = db + .conn() + .query_row( + "SELECT COUNT(*) FROM principal_bindings WHERE principal='p-luna'", + [], + |row| row.get(0), + ) + .unwrap(); + assert_eq!(binding_count, 1); + } + + #[test] + fn failed_tracked_resume_cleanup_keeps_durable_binding() { + let db = launcher_test_db(); + db.conn() + .execute( + "INSERT INTO instances (name, status, created_at) + VALUES ('luna', 'inactive', 1.0)", + [], + ) + .unwrap(); + db.create_principal_binding("p-luna", "luna").unwrap(); + db.reserve_tracked_resume_principal("luna", "p-luna", "proc-resume") + .unwrap(); + let mut env = HashMap::new(); + let principal = attach_provisional_launch_identity( + &db, + "luna", + "proc-resume", + &mut env, + Some("p-luna"), + ) + .unwrap(); + + cleanup_instance(&db, "luna", "proc-resume", Some(&principal), Some("p-luna")); + + assert!(db.get_instance_full("luna").unwrap().is_none()); + assert_eq!( + db.lookup_principal("p-luna").unwrap().instance_name(), + Some("luna") + ); + assert_eq!(db.get_process_binding("proc-resume").unwrap(), None); + } + + #[test] + fn stale_tracked_resume_cannot_claim_or_cleanup_a_replacement_reservation() { + let db = launcher_test_db(); + db.conn() + .execute( + "INSERT INTO instances (name, principal, status, created_at) + VALUES ('luna', 'p-luna', 'inactive', 1.0)", + [], + ) + .unwrap(); + db.conn() + .execute( + "INSERT INTO principal_bindings + (principal, instance_name, epoch, created_at, updated_at) + VALUES ('p-luna', 'luna', 0, 1.0, 1.0)", + [], + ) + .unwrap(); + + db.reserve_tracked_resume_principal("luna", "p-luna", "proc-stale") + .unwrap(); + db.delete_instance("luna").unwrap(); + db.conn() + .execute( + "INSERT INTO instances (name, principal, status, created_at) + VALUES ('luna', 'p-luna', 'inactive', 2.0)", + [], + ) + .unwrap(); + db.reserve_tracked_resume_principal("luna", "p-luna", "proc-replacement") + .unwrap(); + + let mut stale_env = HashMap::new(); + assert!( + attach_provisional_launch_identity( + &db, + "luna", + "proc-stale", + &mut stale_env, + Some("p-luna"), + ) + .is_err(), + "a stale launch must not claim a replacement reservation" + ); + cleanup_instance(&db, "luna", "proc-stale", None, Some("p-luna")); + + assert_eq!( + db.principal_for_instance("luna").unwrap().as_deref(), + Some("p-luna"), + "stale cleanup must leave the replacement reservation intact" + ); + let mut replacement_env = HashMap::new(); + assert_eq!( + attach_provisional_launch_identity( + &db, + "luna", + "proc-replacement", + &mut replacement_env, + Some("p-luna"), + ) + .unwrap(), + "p-luna" + ); + assert_eq!( + db.get_process_binding("proc-replacement") + .unwrap() + .as_deref(), + Some("luna") + ); + } + + #[test] + fn stale_tracked_resume_cleanup_does_not_delete_a_reclaimed_lifecycle() { + let db = launcher_test_db(); + db.conn() + .execute( + "INSERT INTO instances (name, principal, status, created_at) + VALUES ('luna', 'p-old', 'inactive', 1.0)", + [], + ) + .unwrap(); + db.conn() + .execute( + "INSERT INTO principal_bindings + (principal, instance_name, epoch, created_at, updated_at) + VALUES ('p-old', 'luna', 0, 1.0, 1.0)", + [], + ) + .unwrap(); + db.reserve_tracked_resume_principal("luna", "p-old", "proc-stale") + .unwrap(); + + db.delete_instance("luna").unwrap(); + db.conn() + .execute( + "INSERT INTO instances (name, principal, status, created_at) + VALUES ('luna', 'p-new', 'listening', 2.0)", + [], + ) + .unwrap(); + db.conn() + .execute( + "INSERT INTO principal_bindings + (principal, instance_name, epoch, created_at, updated_at) + VALUES ('p-new', 'luna', 0, 2.0, 2.0)", + [], + ) + .unwrap(); + + cleanup_instance(&db, "luna", "proc-stale", None, Some("p-old")); + + assert_eq!( + db.principal_for_instance("luna").unwrap().as_deref(), + Some("p-new"), + "stale cleanup must not delete a concurrently reclaimed lifecycle" + ); + } + fn insert_test_instance(db: &crate::db::HcomDb, name: &str, status: &str) { let now = chrono::Utc::now().timestamp() as f64; db.conn() diff --git a/src/pidtrack.rs b/src/pidtrack.rs index 8ab3f1e2..66bb8d79 100644 --- a/src/pidtrack.rs +++ b/src/pidtrack.rs @@ -332,6 +332,15 @@ pub fn recover_single_orphan_to_db( ) .map_err(|e| format!("failed to insert instance '{}': {}", instance_name, e))?; + if db + .principal_for_instance(instance_name) + .map_err(|e| format!("failed to read principal: {e}"))? + .is_none() + { + db.create_principal_binding(&crate::launcher::generate_principal_id(), instance_name) + .map_err(|e| format!("failed to create principal: {e}"))?; + } + // Update PID and directory let mut updates = serde_json::Map::new(); updates.insert("pid".into(), serde_json::json!(orphan.pid)); @@ -618,4 +627,46 @@ mod tests { "expected error when DB has no instances table" ); } + + #[test] + fn test_recover_single_orphan_mints_one_stable_principal_across_retry() { + let dir = tempfile::tempdir().unwrap(); + let db_path = dir.path().join("test.db"); + let db = crate::db::HcomDb::open_raw(&db_path).unwrap(); + db.init_db().unwrap(); + let orphan = OrphanProcess { + pid: std::process::id(), + tool: "claude".into(), + names: vec!["luna".into()], + directory: "/tmp".into(), + process_id: "pid-1".into(), + terminal_preset: String::new(), + pane_id: String::new(), + terminal_id: String::new(), + kitty_listen_on: String::new(), + zellij_session_name: String::new(), + session_id: String::new(), + notify_port: 0, + inject_port: 0, + tag: String::new(), + }; + + recover_single_orphan_to_db(&db, &orphan, "luna").unwrap(); + let first = db + .principal_for_instance("luna") + .unwrap() + .expect("recovered orphan principal"); + recover_single_orphan_to_db(&db, &orphan, "luna").unwrap(); + assert_eq!( + db.principal_for_instance("luna").unwrap().as_deref(), + Some(first.as_str()) + ); + assert_eq!( + db.conn() + .query_row("SELECT COUNT(*) FROM principal_bindings", [], |row| row + .get::<_, i64>(0)) + .unwrap(), + 1 + ); + } } diff --git a/src/relay/control.rs b/src/relay/control.rs index 025e63de..af02edd6 100644 --- a/src/relay/control.rs +++ b/src/relay/control.rs @@ -779,6 +779,7 @@ fn handle_remote_launch( args: prepared.args, persisted_args: None, prior_session_id: None, + retained_principal: None, tag: request.tag, system_prompt: request.system_prompt, initial_prompt: request.initial_prompt, diff --git a/src/shared/constants.rs b/src/shared/constants.rs index d36212ab..6e43afa0 100644 --- a/src/shared/constants.rs +++ b/src/shared/constants.rs @@ -53,6 +53,7 @@ pub static BIND_MARKER_RE: LazyLock = /// HCOM identity vars — set per-instance, cleared to prevent parent identity leakage. pub const HCOM_IDENTITY_VARS: &[&str] = &[ "HCOM_PROCESS_ID", + "HCOM_PRINCIPAL_ID", "HCOM_LAUNCHED", // HCOM_LAUNCHED_PRESET excluded — must survive into Rust binary for hook forwarding "HCOM_PTY_MODE", diff --git a/src/shared/context.rs b/src/shared/context.rs index b3e67e6a..b33859dc 100644 --- a/src/shared/context.rs +++ b/src/shared/context.rs @@ -21,6 +21,8 @@ pub struct HcomContext { // === Identity === /// HCOM_PROCESS_ID — identifies launched instances. pub process_id: Option, + /// HCOM_PRINCIPAL_ID — durable hcom-owned identity for this lifecycle. + pub principal_id: Option, /// HCOM_LAUNCHED=1 — true if launched by hcom. pub is_launched: bool, /// HCOM_PTY_MODE=1 — running in PTY wrapper. @@ -89,6 +91,7 @@ impl HcomContext { Self { process_id: get_nonempty("HCOM_PROCESS_ID"), + principal_id: get_nonempty("HCOM_PRINCIPAL_ID"), is_launched: is_eq("HCOM_LAUNCHED", "1"), is_pty_mode: is_eq("HCOM_PTY_MODE", "1"), is_background: get_nonempty("HCOM_BACKGROUND").is_some(), diff --git a/tests/cli_smoke.rs b/tests/cli_smoke.rs index 15facba2..c3042c7a 100644 --- a/tests/cli_smoke.rs +++ b/tests/cli_smoke.rs @@ -82,6 +82,66 @@ fn list_json_empty() { assert!(arr.is_empty(), "expected empty list, got {stdout}"); } +#[test] +fn list_json_exposes_principal_and_accepts_exact_principal_lookup() { + let h = Hcom::new(); + let name = h.start(); + + let (code, stdout, stderr) = h.run(["list", &name, "--json"]); + assert_eq!(code, 0, "stderr={stderr}"); + let by_name: serde_json::Value = serde_json::from_str(&stdout).unwrap(); + let principal = by_name["principal"] + .as_str() + .expect("new instance principal"); + + let (code, stdout, stderr) = h.run(["list", "--principal", principal, "--json"]); + assert_eq!(code, 0, "stderr={stderr}"); + let by_principal: serde_json::Value = serde_json::from_str(&stdout).unwrap(); + assert_eq!(by_principal["name"], name); + assert_eq!(by_principal["principal"], principal); + + assert_eq!(h.run(["stop", &name]).0, 0); + let (code, stdout, stderr) = h.run(["list", "--principal", principal, "--json"]); + assert_eq!(code, 0, "stderr={stderr}"); + let stopped: serde_json::Value = serde_json::from_str(&stdout).unwrap(); + assert_eq!(stopped["status"], "unresolved"); + assert_eq!(stopped["name"], name); +} + +#[test] +fn subagent_start_fallback_mints_a_principal() { + let h = Hcom::new(); + let parent = h.start(); + let db = rusqlite::Connection::open(h.path().join("hcom.db")).unwrap(); + db.execute( + "UPDATE instances SET running_tasks=? WHERE name=?", + rusqlite::params![ + serde_json::json!({ + "active": true, + "subagents": [{"agent_id": "agent-77", "type": "general"}] + }) + .to_string(), + parent + ], + ) + .unwrap(); + + let (code, _stdout, stderr) = h.run(["start", "--name", "agent-77"]); + assert_eq!(code, 0, "stderr={stderr}"); + let (name, principal): (String, Option) = db + .query_row( + "SELECT name, principal FROM instances WHERE agent_id='agent-77'", + [], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .unwrap(); + assert!(name.starts_with(&format!("{parent}_general_"))); + assert!( + principal.as_ref().is_some_and(|value| !value.is_empty()), + "successful subagent fallback must not leave a principal-less row" + ); +} + #[test] fn events_empty_in_fresh_dir() { let h = Hcom::new();