diff --git a/doc/analysis/2026-03-14-daemon-jobkit-survey.md b/.claude/analysis/2026-03-14-daemon-jobkit-survey.md similarity index 100% rename from doc/analysis/2026-03-14-daemon-jobkit-survey.md rename to .claude/analysis/2026-03-14-daemon-jobkit-survey.md diff --git a/doc/analysis/2026-03-14-link-strength-feedback.md b/.claude/analysis/2026-03-14-link-strength-feedback.md similarity index 100% rename from doc/analysis/2026-03-14-link-strength-feedback.md rename to .claude/analysis/2026-03-14-link-strength-feedback.md diff --git a/doc/dmn-algorithm-plan.md b/.claude/dmn-algorithm-plan.md similarity index 100% rename from doc/dmn-algorithm-plan.md rename to .claude/dmn-algorithm-plan.md diff --git a/doc/query-language-design.md b/.claude/query-language-design.md similarity index 100% rename from doc/query-language-design.md rename to .claude/query-language-design.md diff --git a/doc/scoring-persistence-analysis.md b/.claude/scoring-persistence-analysis.md similarity index 100% rename from doc/scoring-persistence-analysis.md rename to .claude/scoring-persistence-analysis.md diff --git a/doc/ui-desync-analysis.md b/.claude/ui-desync-analysis.md similarity index 100% rename from doc/ui-desync-analysis.md rename to .claude/ui-desync-analysis.md diff --git a/Cargo.lock b/Cargo.lock index 60a55a4..f4e8519 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -513,73 +513,18 @@ dependencies = [ "static_assertions", ] -[[package]] -name = "consciousness" -version = "0.4.0" -dependencies = [ - "anyhow", - "ast-grep-core", - "ast-grep-language", - "base64 0.22.1", - "bincode", - "bytes", - "capnp", - "capnp-rpc", - "capnpc", - "chrono", - "clap", - "crossterm", - "dirs", - "env_logger", - "figment", - "futures", - "glob", - "http", - "http-body-util", - "hyper", - "hyper-util", - "jobkit", - "json5", - "libc", - "log", - "memchr", - "memmap2", - "paste", - "peg", - "ratatui", - "rayon", - "redb", - "regex", - "rkyv", - "rustls", - "rustls-native-certs", - "serde", - "serde_json", - "serde_urlencoded", - "skillratings", - "tokenizers", - "tokio", - "tokio-rustls", - "tokio-scoped", - "tokio-util", - "tui-markdown", - "tui-textarea-2", - "uuid", - "walkdir", -] - [[package]] name = "consciousness-channel-irc" version = "0.4.0" dependencies = [ "capnp", "capnp-rpc", - "consciousness", "dirs", "env_logger", "futures", "json5", "log", + "poc-memory", "rustls", "serde", "tokio", @@ -594,11 +539,11 @@ version = "0.4.0" dependencies = [ "capnp", "capnp-rpc", - "consciousness", "dirs", "env_logger", "futures", "log", + "poc-memory", "tokio", "tokio-util", ] @@ -609,11 +554,11 @@ version = "0.4.0" dependencies = [ "capnp", "capnp-rpc", - "consciousness", "dirs", "env_logger", "futures", "log", + "poc-memory", "serde", "serde_json", "tokio", @@ -626,13 +571,13 @@ version = "0.4.0" dependencies = [ "capnp", "capnp-rpc", - "consciousness", "dirs", "env_logger", "futures", "json5", "libc", "log", + "poc-memory", "scopeguard", "serde", "tokio", @@ -1328,12 +1273,6 @@ dependencies = [ "foldhash 0.2.0", ] -[[package]] -name = "hashbrown" -version = "0.17.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51" - [[package]] name = "heck" version = "0.5.0" @@ -1473,12 +1412,12 @@ dependencies = [ [[package]] name = "indexmap" -version = "2.14.0" +version = "2.13.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9" +checksum = "45a8a2b9cb3e0b0c1803dbb0758ffac5de2f425b23c28f518faabd9d805342ff" dependencies = [ "equivalent", - "hashbrown 0.17.0", + "hashbrown 0.16.1", "serde", "serde_core", ] @@ -1595,9 +1534,9 @@ dependencies = [ [[package]] name = "js-sys" -version = "0.3.94" +version = "0.3.91" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "2e04e2ef80ce82e13552136fabeef8a5ed1f985a96805761cbb9a2c34e7664d9" +checksum = "b49715b7073f385ba4bc528e5747d02e66cb39c6146efb66b781f131f0fb399c" dependencies = [ "once_cell", "wasm-bindgen", @@ -1650,9 +1589,9 @@ checksum = "48f5d2a454e16a5ea0f4ced81bd44e4cfc7bd3a507b61887c99fd3538b28e4af" [[package]] name = "libredox" -version = "0.1.16" +version = "0.1.14" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e02f3bb43d335493c96bf3fd3a321600bf6bd07ed34bc64118e9293bdffea46c" +checksum = "1744e39d1d6a9948f4f388969627434e31128196de472883b39f148769bfe30a" dependencies = [ "libc", ] @@ -2147,6 +2086,61 @@ dependencies = [ "time", ] +[[package]] +name = "poc-memory" +version = "0.4.0" +dependencies = [ + "anyhow", + "ast-grep-core", + "ast-grep-language", + "base64 0.22.1", + "bincode", + "bytes", + "capnp", + "capnp-rpc", + "capnpc", + "chrono", + "clap", + "crossterm", + "dirs", + "env_logger", + "figment", + "futures", + "glob", + "http", + "http-body-util", + "hyper", + "hyper-util", + "jobkit", + "json5", + "libc", + "log", + "memchr", + "memmap2", + "paste", + "peg", + "ratatui", + "rayon", + "redb", + "regex", + "rkyv", + "rustls", + "rustls-native-certs", + "serde", + "serde_json", + "serde_urlencoded", + "skillratings", + "tokenizers", + "tokio", + "tokio-rustls", + "tokio-scoped", + "tokio-util", + "tui-markdown", + "tui-textarea-2", + "uuid", + "walkdir", +] + [[package]] name = "portable-atomic" version = "1.13.1" @@ -3158,9 +3152,9 @@ dependencies = [ [[package]] name = "tokio" -version = "1.51.1" +version = "1.51.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f66bf9585cda4b724d3e78ab34b73fb2bbaba9011b9bfdf69dc836382ea13b8c" +checksum = "2bd1c4c0fc4a7ab90fc15ef6daaa3ec3b893f004f915f2392557ed23237820cd" dependencies = [ "bytes", "libc", @@ -3703,9 +3697,9 @@ dependencies = [ [[package]] name = "wasm-bindgen" -version = "0.2.117" +version = "0.2.114" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0551fc1bb415591e3372d0bc4780db7e587d84e2a7e79da121051c5c4b89d0b0" +checksum = "6532f9a5c1ece3798cb1c2cfdba640b9b3ba884f5db45973a6f442510a87d38e" dependencies = [ "cfg-if", "once_cell", @@ -3716,9 +3710,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.117" +version = "0.2.114" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7fbdf9a35adf44786aecd5ff89b4563a90325f9da0923236f6104e603c7e86be" +checksum = "18a2d50fcf105fb33bb15f00e7a77b772945a2ee45dcf454961fd843e74c18e6" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -3726,9 +3720,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.117" +version = "0.2.114" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dca9693ef2bab6d4e6707234500350d8dad079eb508dca05530c85dc3a529ff2" +checksum = "03ce4caeaac547cdf713d280eda22a730824dd11e6b8c3ca9e42247b25c631e3" dependencies = [ "bumpalo", "proc-macro2", @@ -3739,9 +3733,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-shared" -version = "0.2.117" +version = "0.2.114" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "39129a682a6d2d841b6c429d0c51e5cb0ed1a03829d8b3d1e69a011e62cb3d3b" +checksum = "75a326b8c223ee17883a4251907455a2431acc2791c98c26279376490c378c16" dependencies = [ "unicode-ident", ] diff --git a/Cargo.toml b/Cargo.toml index 20df8d1..64dbf8d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -14,7 +14,7 @@ debug = 1 debug = false [package] -name = "consciousness" +name = "poc-memory" version.workspace = true edition.workspace = true @@ -82,7 +82,7 @@ serde_urlencoded = "0.7" capnpc = "0.25" [lib] -name = "consciousness" +name = "poc_memory" path = "src/lib.rs" [[bin]] @@ -104,3 +104,19 @@ path = "src/bin/diag-key.rs" [[bin]] name = "find-deleted" path = "src/bin/find-deleted.rs" + +[[bin]] +name = "poc-hook" +path = "src/claude/poc-hook.rs" + +[[bin]] +name = "poc-daemon" +path = "src/claude/poc-daemon.rs" + +[[bin]] +name = "memory-search" +path = "src/claude/memory-search.rs" + +[[bin]] +name = "consciousness-mcp" +path = "src/claude/mcp-server.rs" diff --git a/channels/irc/Cargo.toml b/channels/irc/Cargo.toml index dac7e4f..b59ab06 100644 --- a/channels/irc/Cargo.toml +++ b/channels/irc/Cargo.toml @@ -9,7 +9,7 @@ capnp-rpc = "0.25" dirs = "6" futures = "0.3" json5 = "1.3" -consciousness = { path = "../.." } +poc-memory = { path = "../.." } rustls = { version = "0.23", default-features = false, features = ["ring", "logging", "std", "tls12"] } serde = { version = "1", features = ["derive"] } tokio = { version = "1", features = ["full"] } diff --git a/channels/irc/src/main.rs b/channels/irc/src/main.rs index 48c60e0..fb0a8c0 100644 --- a/channels/irc/src/main.rs +++ b/channels/irc/src/main.rs @@ -24,8 +24,8 @@ use tokio::net::UnixListener; use tokio_util::compat::TokioAsyncReadCompatExt; use log::{info, warn, error}; -use consciousness::channel_capnp::{channel_client, channel_server}; -use consciousness::thalamus::channel_log; +use poc_memory::channel_capnp::{channel_client, channel_server}; +use poc_memory::thalamus::channel_log; // ── Constants ────────────────────────────────────────────────── @@ -159,7 +159,7 @@ impl AsyncWriter for PlainWriter { // ── State ────────────────────────────────────────────────────── -use consciousness::thalamus::channel_log::ChannelLog; +use poc_memory::thalamus::channel_log::ChannelLog; struct State { config: Config, diff --git a/channels/socat/Cargo.toml b/channels/socat/Cargo.toml index 4038e20..8c67129 100644 --- a/channels/socat/Cargo.toml +++ b/channels/socat/Cargo.toml @@ -8,7 +8,7 @@ capnp = "0.25" capnp-rpc = "0.25" dirs = "6" futures = "0.3" -consciousness = { path = "../.." } +poc-memory = { path = "../.." } tokio = { version = "1", features = ["full"] } tokio-util = { version = "0.7", features = ["compat"] } log = "0.4" diff --git a/channels/socat/src/main.rs b/channels/socat/src/main.rs index c57dafd..edb16eb 100644 --- a/channels/socat/src/main.rs +++ b/channels/socat/src/main.rs @@ -18,8 +18,8 @@ use tokio::net::{TcpStream, UnixListener, UnixStream}; use tokio_util::compat::TokioAsyncReadCompatExt; use log::{info, warn, error}; -use consciousness::channel_capnp::{channel_client, channel_server}; -use consciousness::thalamus::channel_log::ChannelLog; +use poc_memory::channel_capnp::{channel_client, channel_server}; +use poc_memory::thalamus::channel_log::ChannelLog; // ── State ────────────────────────────────────────────────────── diff --git a/channels/telegram/Cargo.toml b/channels/telegram/Cargo.toml index 97c60f0..902453e 100644 --- a/channels/telegram/Cargo.toml +++ b/channels/telegram/Cargo.toml @@ -8,7 +8,7 @@ capnp = "0.25" capnp-rpc = "0.25" dirs = "6" futures = "0.3" -consciousness = { path = "../.." } +poc-memory = { path = "../.." } serde = { version = "1", features = ["derive"] } serde_json = "1" tokio = { version = "1", features = ["full"] } diff --git a/channels/telegram/src/main.rs b/channels/telegram/src/main.rs index ef2d597..3236fa8 100644 --- a/channels/telegram/src/main.rs +++ b/channels/telegram/src/main.rs @@ -17,7 +17,7 @@ use tokio::net::UnixListener; use tokio_util::compat::TokioAsyncReadCompatExt; use log::{info, error}; -use consciousness::channel_capnp::{channel_client, channel_server}; +use poc_memory::channel_capnp::{channel_client, channel_server}; // ── Config ────────────────────────────────────────────────────── @@ -55,7 +55,7 @@ fn load_config() -> Config { // ── State ─────────────────────────────────────────────────────── -use consciousness::thalamus::channel_log::ChannelLog; +use poc_memory::thalamus::channel_log::ChannelLog; struct State { config: Config, @@ -64,7 +64,7 @@ struct State { /// Telegram API offset last_offset: i64, connected: bool, - client: consciousness::agent::api::http::HttpClient, + client: poc_memory::agent::api::http::HttpClient, /// Registered notification callbacks subscribers: Vec, } @@ -79,7 +79,7 @@ impl State { channel_logs: std::collections::BTreeMap::new(), last_offset, connected: false, - client: consciousness::agent::api::http::HttpClient::new(), + client: poc_memory::agent::api::http::HttpClient::new(), subscribers: Vec::new(), } } diff --git a/channels/tmux/Cargo.toml b/channels/tmux/Cargo.toml index 6e4c0aa..da1b499 100644 --- a/channels/tmux/Cargo.toml +++ b/channels/tmux/Cargo.toml @@ -11,7 +11,7 @@ libc = "0.2" scopeguard = "1" futures = "0.3" json5 = "1.3" -consciousness = { path = "../.." } +poc-memory = { path = "../.." } serde = { version = "1", features = ["derive"] } tokio = { version = "1", features = ["full"] } tokio-util = { version = "0.7", features = ["compat"] } diff --git a/channels/tmux/src/main.rs b/channels/tmux/src/main.rs index 4255671..7ff0ce4 100644 --- a/channels/tmux/src/main.rs +++ b/channels/tmux/src/main.rs @@ -19,8 +19,8 @@ use tokio::net::UnixListener; use tokio_util::compat::TokioAsyncReadCompatExt; use log::{info, warn, error}; -use consciousness::channel_capnp::channel_server; -use consciousness::thalamus::channel_log::ChannelLog; +use poc_memory::channel_capnp::channel_server; +use poc_memory::thalamus::channel_log::ChannelLog; // ── Config ───────────────────────────────────────────────────── diff --git a/src/bin/consciousness.rs b/src/bin/consciousness.rs index 5528412..d1b123b 100644 --- a/src/bin/consciousness.rs +++ b/src/bin/consciousness.rs @@ -1,2 +1,2 @@ #![warn(unreachable_pub)] -fn main() { consciousness::user::main() } +fn main() { poc_memory::user::main() } diff --git a/src/bin/diag-key.rs b/src/bin/diag-key.rs index 437cc31..446dfb8 100644 --- a/src/bin/diag-key.rs +++ b/src/bin/diag-key.rs @@ -2,8 +2,8 @@ use std::io::BufReader; use std::fs; use capnp::{message, serialize}; -use consciousness::memory_capnp; -use consciousness::store::Node; +use poc_memory::memory_capnp; +use poc_memory::store::Node; fn main() { let args: Vec = std::env::args().collect(); diff --git a/src/bin/find-deleted.rs b/src/bin/find-deleted.rs index 17510ba..d83d9d7 100644 --- a/src/bin/find-deleted.rs +++ b/src/bin/find-deleted.rs @@ -8,13 +8,13 @@ use std::collections::HashMap; use std::io::BufReader; use std::fs; use capnp::{message, serialize}; -use consciousness::memory_capnp; -use consciousness::store::Node; +use poc_memory::memory_capnp; +use poc_memory::store::Node; fn main() { let path = std::env::args().nth(1) .unwrap_or_else(|| { - let dir = consciousness::store::nodes_path(); + let dir = poc_memory::store::nodes_path(); dir.to_string_lossy().to_string() }); diff --git a/src/bin/merge-logs.rs b/src/bin/merge-logs.rs index d883fa2..69067ab 100644 --- a/src/bin/merge-logs.rs +++ b/src/bin/merge-logs.rs @@ -25,8 +25,8 @@ use std::path::Path; use capnp::message; use capnp::serialize; -use consciousness::memory_capnp; -use consciousness::store::Node; +use poc_memory::memory_capnp; +use poc_memory::store::Node; /// Read all node entries from a capnp log file, preserving order. fn read_all_entries(path: &Path) -> Result, String> { diff --git a/src/claude/agent_cycles.rs b/src/claude/agent_cycles.rs new file mode 100644 index 0000000..423a2ad --- /dev/null +++ b/src/claude/agent_cycles.rs @@ -0,0 +1,416 @@ +// agent_cycles.rs — Agent orchestration for the Claude Code hook path +// +// Forked from subconscious/subconscious.rs. This copy handles the +// serialized-to-disk, process-spawning model used by Claude Code hooks. +// The TUI/Mind copy in subconscious/ is free to evolve independently +// (async tasks, integrated with Mind's event loop). + +use std::fs; +use std::fs::File; +use std::io::Write; +use std::path::{Path, PathBuf}; +use std::time::{Duration, Instant, SystemTime}; + +pub use crate::session::HookSession; + +/// Output from a single agent orchestration cycle. +#[derive(Default)] +pub struct AgentCycleOutput { + /// Memory node keys surfaced by surface-observe. + pub surfaced_keys: Vec, + /// Freeform reflection text from the reflect agent. + pub reflection: Option, + /// How long we slept waiting for observe to catch up, if at all. + pub sleep_secs: Option, +} + +/// Per-agent runtime state. +pub struct AgentInfo { + pub name: &'static str, + pub pid: Option, + pub phase: Option, + pub log_path: Option, + child: Option, +} + +/// Snapshot of agent state — serializable, sendable to TUI. +#[derive(Clone, Debug, serde::Serialize, serde::Deserialize)] +pub struct AgentSnapshot { + pub name: String, + pub pid: Option, + pub phase: Option, + pub log_path: Option, +} + +impl AgentInfo { + fn snapshot(&self) -> AgentSnapshot { + AgentSnapshot { + name: self.name.to_string(), + pid: self.pid, + phase: self.phase.clone(), + log_path: self.log_path.clone(), + } + } +} + +/// Serializable state for persisting across Claude Code hook invocations. +#[derive(serde::Serialize, serde::Deserialize)] +pub struct SavedAgentState { + pub agents: Vec, +} + +impl SavedAgentState { + fn state_path(session_id: &str) -> PathBuf { + let dir = dirs::home_dir().unwrap_or_default().join(".consciousness/sessions"); + fs::create_dir_all(&dir).ok(); + dir.join(format!("agent-state-{}.json", session_id)) + } + + pub fn load(session_id: &str) -> Self { + let path = Self::state_path(session_id); + let mut state: Self = fs::read_to_string(&path).ok() + .and_then(|s| serde_json::from_str(&s).ok()) + .unwrap_or(SavedAgentState { agents: Vec::new() }); + + for agent in &mut state.agents { + if let Some(pid) = agent.pid { + unsafe { + if libc::kill(pid as i32, 0) != 0 { + agent.pid = None; + agent.phase = None; + } + } + } + } + state + } + + pub fn save(&self, session_id: &str) { + let path = Self::state_path(session_id); + if let Ok(json) = serde_json::to_string(self) { + fs::write(path, json).ok(); + } + } +} + +/// Persistent state for the agent orchestration cycle. +/// Created once per hook invocation, `trigger()` called on each user message. +pub struct AgentCycleState { + output_dir: PathBuf, + log_file: Option, + pub agents: Vec, + pub last_output: AgentCycleOutput, +} + +const AGENT_CYCLE_NAMES: &[&str] = &["surface-observe", "journal", "reflect"]; + +impl AgentCycleState { + pub fn new(session_id: &str) -> Self { + let output_dir = crate::store::memory_dir().join("agent-output"); + let log_dir = dirs::home_dir().unwrap_or_default().join(".consciousness/logs"); + fs::create_dir_all(&log_dir).ok(); + let log_path = log_dir.join(format!("hook-{}", session_id)); + let log_file = fs::OpenOptions::new() + .create(true).append(true).open(log_path).ok(); + + let agents = AGENT_CYCLE_NAMES.iter() + .map(|&name| AgentInfo { name, pid: None, phase: None, log_path: None, child: None }) + .collect(); + + AgentCycleState { + output_dir, + log_file, + agents, + last_output: AgentCycleOutput { + surfaced_keys: vec![], + reflection: None, + sleep_secs: None, + }, + } + } + + fn log(&mut self, msg: std::fmt::Arguments) { + if let Some(ref mut f) = self.log_file { + let _ = write!(f, "{}", msg); + } + } + + fn agent_running(&self, name: &str) -> bool { + self.agents.iter().any(|a| a.name == name && a.pid.is_some()) + } + + fn agent_spawned(&mut self, name: &str, phase: &str, + result: crate::agent::oneshot::SpawnResult) { + if let Some(agent) = self.agents.iter_mut().find(|a| a.name == name) { + agent.pid = Some(result.child.id()); + agent.phase = Some(phase.to_string()); + agent.log_path = Some(result.log_path); + agent.child = Some(result.child); + } + } + + /// Check if any agents have completed. Reap child handles, or + /// check pid liveness for restored-from-disk agents. + fn poll_children(&mut self) { + for agent in &mut self.agents { + if let Some(ref mut child) = agent.child { + if let Ok(Some(_)) = child.try_wait() { + agent.pid = None; + agent.phase = None; + agent.child = None; + } + } else if let Some(pid) = agent.pid { + unsafe { + if libc::kill(pid as i32, 0) != 0 { + agent.pid = None; + agent.phase = None; + } + } + } + } + } + + pub fn snapshots(&self, scoring_in_flight: bool, scored_count: usize) -> Vec { + let mut snaps: Vec = self.agents.iter().map(|a| a.snapshot()).collect(); + snaps.push(AgentSnapshot { + name: "memory-scoring".to_string(), + pid: None, + phase: if scoring_in_flight { + Some("scoring...".into()) + } else if scored_count == 0 { + None + } else { + Some(format!("{} scored", scored_count)) + }, + log_path: None, + }); + snaps + } + + /// Restore agent state from a saved snapshot. + pub fn restore(&mut self, saved: &SavedAgentState) { + for sa in &saved.agents { + if let Some(agent) = self.agents.iter_mut().find(|a| a.name == sa.name) { + agent.pid = sa.pid; + agent.phase = sa.phase.clone(); + agent.log_path = sa.log_path.clone(); + } + } + } + + /// Save current state for next hook invocation. + pub fn save(&self, session_id: &str) { + let state = SavedAgentState { agents: self.snapshots(false, 0) }; + state.save(session_id); + } + + /// Run all agent cycles. Call on each user message. + pub fn trigger(&mut self, session: &HookSession) { + let ts = chrono::Local::now().format("%Y-%m-%dT%H:%M:%S"); + self.log(format_args!("\n=== {} agent_cycles ===\n", ts)); + + self.poll_children(); + cleanup_stale_files(&session.state_dir, Duration::from_secs(86400)); + + let (surfaced_keys, sleep_secs) = self.surface_observe_cycle(session); + let reflection = self.reflection_cycle(session); + self.journal_cycle(session); + + self.last_output = AgentCycleOutput { surfaced_keys, reflection, sleep_secs }; + } + + fn agent_dir(&self, name: &str) -> PathBuf { + let dir = self.output_dir.join(name); + fs::create_dir_all(&dir).ok(); + dir + } + + fn surface_observe_cycle(&mut self, session: &HookSession) -> (Vec, Option) { + let state_dir = self.agent_dir("surface-observe"); + let transcript = session.transcript(); + let offset_path = state_dir.join("transcript-offset"); + let last_offset: u64 = fs::read_to_string(&offset_path).ok() + .and_then(|s| s.trim().parse().ok()) + .unwrap_or(0); + + // Read surfaced keys + let mut surfaced_keys = Vec::new(); + let surface_path = state_dir.join("surface"); + if let Ok(content) = fs::read_to_string(&surface_path) { + let mut seen = session.seen(); + let seen_path = session.path("seen"); + for key in content.lines().map(|l| l.trim()).filter(|l| !l.is_empty()) { + if !seen.insert(key.to_string()) { + self.log(format_args!(" skip (seen): {}\n", key)); + continue; + } + surfaced_keys.push(key.to_string()); + if let Ok(mut f) = fs::OpenOptions::new() + .create(true).append(true).open(&seen_path) { + let ts = chrono::Local::now().format("%Y-%m-%dT%H:%M:%S"); + writeln!(f, "{}\t{}", ts, key).ok(); + } + self.log(format_args!(" surfaced: {}\n", key)); + } + fs::remove_file(&surface_path).ok(); + } + + // Spawn new agent if not already running + let running = self.agent_running("surface-observe"); + if running { + self.log(format_args!("surface-observe already running\n")); + } else { + if transcript.size > 0 { + fs::write(&offset_path, transcript.size.to_string()).ok(); + } + if let Some(result) = crate::agent::oneshot::spawn_agent( + "surface-observe", &state_dir, &session.session_id) { + self.log(format_args!("spawned surface-observe pid {}\n", result.child.id())); + self.agent_spawned("surface-observe", "surface", result); + } + } + + // Wait if agent is significantly behind + let mut sleep_secs = None; + let conversation_budget: u64 = 50_000; + + if running && transcript.size > 0 { + let behind = transcript.size.saturating_sub(last_offset); + + if behind > conversation_budget / 2 { + let sleep_start = Instant::now(); + self.log(format_args!("agent {}KB behind\n", behind / 1024)); + + for _ in 0..5 { + std::thread::sleep(std::time::Duration::from_secs(1)); + self.poll_children(); + if !self.agent_running("surface-observe") { break; } + } + + let secs = (Instant::now() - sleep_start).as_secs_f64(); + self.log(format_args!("slept {secs:.2}s\n")); + sleep_secs = Some(secs); + } + } + + (surfaced_keys, sleep_secs) + } + + fn reflection_cycle(&mut self, session: &HookSession) -> Option { + let state_dir = self.agent_dir("reflect"); + let offset_path = state_dir.join("transcript-offset"); + let transcript = session.transcript(); + + let last_offset: u64 = fs::read_to_string(&offset_path).ok() + .and_then(|s| s.trim().parse().ok()) + .unwrap_or(0); + + const REFLECTION_INTERVAL: u64 = 100_000; + if transcript.size.saturating_sub(last_offset) < REFLECTION_INTERVAL { + return None; + } + + if self.agent_running("reflect") { + self.log(format_args!("reflect: already running\n")); + return None; + } + + // Copy walked nodes from surface-observe + let so_state = self.agent_dir("surface-observe"); + if let Ok(walked) = fs::read_to_string(so_state.join("walked")) { + fs::write(state_dir.join("walked"), &walked).ok(); + } + + // Read and consume pending reflection + let reflection = fs::read_to_string(state_dir.join("reflection")).ok() + .filter(|s| !s.trim().is_empty()); + if reflection.is_some() { + fs::remove_file(state_dir.join("reflection")).ok(); + self.log(format_args!("reflect: consumed reflection\n")); + } + + fs::write(&offset_path, transcript.size.to_string()).ok(); + if let Some(result) = crate::agent::oneshot::spawn_agent( + "reflect", &state_dir, &session.session_id) { + self.log(format_args!("reflect: spawned pid {}\n", result.child.id())); + self.agent_spawned("reflect", "step-0", result); + } + + reflection + } + + fn journal_cycle(&mut self, session: &HookSession) { + let state_dir = self.agent_dir("journal"); + let offset_path = state_dir.join("transcript-offset"); + let transcript = session.transcript(); + + let last_offset: u64 = fs::read_to_string(&offset_path).ok() + .and_then(|s| s.trim().parse().ok()) + .unwrap_or(0); + + const JOURNAL_INTERVAL: u64 = 20_000; + if transcript.size.saturating_sub(last_offset) < JOURNAL_INTERVAL { + return; + } + + if self.agent_running("journal") { + self.log(format_args!("journal: already running\n")); + return; + } + + fs::write(&offset_path, transcript.size.to_string()).ok(); + if let Some(result) = crate::agent::oneshot::spawn_agent( + "journal", &state_dir, &session.session_id) { + self.log(format_args!("journal: spawned pid {}\n", result.child.id())); + self.agent_spawned("journal", "step-0", result); + } + } +} + +/// Format agent cycle output for injection into a Claude Code session. +pub fn format_agent_output(output: &AgentCycleOutput) -> String { + let mut out = String::new(); + + if let Some(secs) = output.sleep_secs { + out.push_str(&format!("Slept {secs:.2}s to let observe catch up\n")); + } + + if !output.surfaced_keys.is_empty() { + if let Ok(store) = crate::store::Store::load() { + for key in &output.surfaced_keys { + if let Some(rendered) = crate::cli::node::render_node(&store, key) { + if !rendered.trim().is_empty() { + use std::fmt::Write as _; + writeln!(out, "--- {} (surfaced) ---", key).ok(); + write!(out, "{}", rendered).ok(); + } + } + } + } + } + + if let Some(ref reflection) = output.reflection { + use std::fmt::Write as _; + writeln!(out, "--- subconscious reflection ---").ok(); + write!(out, "{}", reflection.trim()).ok(); + } + + out +} + +fn cleanup_stale_files(dir: &Path, max_age: Duration) { + let entries = match fs::read_dir(dir) { + Ok(e) => e, + Err(_) => return, + }; + let cutoff = SystemTime::now() - max_age; + for entry in entries.flatten() { + if let Ok(meta) = entry.metadata() { + if let Ok(modified) = meta.modified() { + if modified < cutoff { + fs::remove_file(entry.path()).ok(); + } + } + } + } +} diff --git a/src/claude/context.rs b/src/claude/context.rs new file mode 100644 index 0000000..22b716a --- /dev/null +++ b/src/claude/context.rs @@ -0,0 +1,19 @@ +// Context gathering for idle prompts. +// +// Notifications are handled by the notify module and passed +// in separately by the caller. Git context and IRC digest +// are now available through where-am-i.md and the memory graph. + +/// Build context string for a prompt. +/// notification_text is passed in from the notify module. +pub fn build(_include_irc: bool, notification_text: &str) -> String { + // Keep nudges short — Claude checks notifications via + // `poc-daemon status` on its own. Just mention the count. + let count = notification_text.matches("[irc.").count() + + notification_text.matches("[telegram.").count(); + if count > 0 { + format!("{count} pending notifications") + } else { + String::new() + } +} diff --git a/src/claude/hook.rs b/src/claude/hook.rs new file mode 100644 index 0000000..86e5ab4 --- /dev/null +++ b/src/claude/hook.rs @@ -0,0 +1,312 @@ +// hook.rs — Claude Code session hook: context injection + agent orchestration +// +// Called on each UserPromptSubmit via the poc-hook binary. Handles +// context loading, chunking, seen-set management, and delegates +// agent orchestration to AgentCycleState. + +use std::collections::HashSet; +use std::fs; +use std::io::Write; +use std::path::Path; +use std::process::Command; +use std::time::Instant; + +pub use crate::session::HookSession; +pub use super::agent_cycles::*; + +const CHUNK_SIZE: usize = 9000; + +/// Run the hook logic on parsed JSON input. Returns output to inject. +pub fn run_hook(input: &str) -> String { + let Some(session) = HookSession::from_json(input) else { return String::new() }; + hook(&session) +} + +fn chunk_context(ctx: &str, max_bytes: usize) -> Vec { + let mut sections: Vec = Vec::new(); + let mut current = String::new(); + + for line in ctx.lines() { + if line.starts_with("--- ") && line.ends_with(" ---") && !current.is_empty() { + sections.push(std::mem::take(&mut current)); + } + if !current.is_empty() { + current.push('\n'); + } + current.push_str(line); + } + if !current.is_empty() { + sections.push(current); + } + + let mut chunks: Vec = Vec::new(); + let mut chunk = String::new(); + for section in sections { + if !chunk.is_empty() && chunk.len() + section.len() + 1 > max_bytes { + chunks.push(std::mem::take(&mut chunk)); + } + if !chunk.is_empty() { + chunk.push('\n'); + } + chunk.push_str(§ion); + } + if !chunk.is_empty() { + chunks.push(chunk); + } + chunks +} + +fn save_pending_chunks(dir: &Path, session_id: &str, chunks: &[String]) { + let chunks_dir = dir.join(format!("chunks-{}", session_id)); + let _ = fs::remove_dir_all(&chunks_dir); + if chunks.is_empty() { return; } + fs::create_dir_all(&chunks_dir).ok(); + for (i, chunk) in chunks.iter().enumerate() { + let path = chunks_dir.join(format!("{:04}", i)); + fs::write(path, chunk).ok(); + } +} + +fn pop_pending_chunk(dir: &Path, session_id: &str) -> Option { + let chunks_dir = dir.join(format!("chunks-{}", session_id)); + if !chunks_dir.exists() { return None; } + + let mut entries: Vec<_> = fs::read_dir(&chunks_dir).ok()? + .flatten() + .filter(|e| e.file_type().map(|t| t.is_file()).unwrap_or(false)) + .collect(); + entries.sort_by_key(|e| e.file_name()); + + let first = entries.first()?; + let content = fs::read_to_string(first.path()).ok()?; + fs::remove_file(first.path()).ok(); + + if fs::read_dir(&chunks_dir).ok().map(|mut d| d.next().is_none()).unwrap_or(true) { + fs::remove_dir(&chunks_dir).ok(); + } + + Some(content) +} + +fn generate_cookie() -> String { + uuid::Uuid::new_v4().as_simple().to_string()[..12].to_string() +} + +fn parse_seen_line(line: &str) -> &str { + line.split_once('\t').map(|(_, key)| key).unwrap_or(line) +} + +pub fn load_seen(dir: &Path, session_id: &str) -> HashSet { + let path = dir.join(format!("seen-{}", session_id)); + if path.exists() { + fs::read_to_string(&path) + .unwrap_or_default() + .lines() + .filter(|s| !s.is_empty()) + .map(|s| parse_seen_line(s).to_string()) + .collect() + } else { + HashSet::new() + } +} + +fn mark_seen(dir: &Path, session_id: &str, key: &str, seen: &mut HashSet) { + if !seen.insert(key.to_string()) { return; } + let path = dir.join(format!("seen-{}", session_id)); + if let Ok(mut f) = fs::OpenOptions::new().create(true).append(true).open(path) { + let ts = chrono::Local::now().format("%Y-%m-%dT%H:%M:%S"); + writeln!(f, "{}\t{}", ts, key).ok(); + } +} + +/// Standalone entry point for the Claude Code hook path. +/// Loads saved state, runs cycles, saves state back. +pub fn run_agent_cycles(session: &HookSession) -> AgentCycleOutput { + let mut state = AgentCycleState::new(&session.session_id); + state.restore(&SavedAgentState::load(&session.session_id)); + state.trigger(session); + state.save(&session.session_id); + state.last_output +} + +fn hook(session: &HookSession) -> String { + let start_time = Instant::now(); + + let mut out = String::new(); + let is_compaction = crate::transcript::detect_new_compaction( + &session.state_dir, &session.session_id, &session.transcript_path, + ); + let cookie_path = session.path("cookie"); + let is_first = !cookie_path.exists(); + + let log_dir = dirs::home_dir().unwrap_or_default().join(".consciousness/logs"); + fs::create_dir_all(&log_dir).ok(); + let log_path = log_dir.join(format!("hook-{}", session.session_id)); + let Ok(mut log_f) = fs::OpenOptions::new().create(true).append(true).open(log_path) else { return Default::default(); }; + let ts = chrono::Local::now().format("%Y-%m-%dT%H:%M:%S"); + let _ = writeln!(log_f, "\n=== {} ({}) {} bytes ===", ts, session.hook_event, out.len()); + + let _ = writeln!(log_f, "is_first {is_first} is_compaction {is_compaction}"); + + if is_first || is_compaction { + if is_compaction { + fs::rename(&session.path("seen"), &session.path("seen-prev")).ok(); + } else { + fs::remove_file(&session.path("seen")).ok(); + fs::remove_file(&session.path("seen-prev")).ok(); + } + fs::remove_file(&session.path("returned")).ok(); + + if is_first { + fs::write(&cookie_path, generate_cookie()).ok(); + } + + if let Ok(output) = Command::new("poc-memory").args(["admin", "load-context"]).output() { + if output.status.success() { + let ctx = String::from_utf8_lossy(&output.stdout).to_string(); + if !ctx.trim().is_empty() { + let mut ctx_seen = session.seen(); + for line in ctx.lines() { + if line.starts_with("--- ") && line.ends_with(" ---") { + let inner = &line[4..line.len() - 4]; + if let Some(paren) = inner.rfind(" (") { + let key = inner[..paren].trim(); + mark_seen(&session.state_dir, &session.session_id, key, &mut ctx_seen); + } + } + } + + let chunks = chunk_context(&ctx, CHUNK_SIZE); + + if let Some(first) = chunks.first() { + out.push_str(first); + } + save_pending_chunks(&session.state_dir, &session.session_id, &chunks[1..]); + } + } + } + } + + if let Some(chunk) = pop_pending_chunk(&session.state_dir, &session.session_id) { + out.push_str(&chunk); + } else { + let cfg = crate::config::get(); + if cfg.surface_hooks.iter().any(|h| h == &session.hook_event) { + let cycle_output = run_agent_cycles(&session); + out.push_str(&format_agent_output(&cycle_output)); + } + } + + let _ = write!(log_f, "{}", out); + + let duration = (Instant::now() - start_time).as_secs_f64(); + let _ = writeln!(log_f, "\nran in {duration:.2}s"); + + out +} + +/// Install memory-search and poc-hook into Claude Code settings.json. +/// +/// Hook layout: +/// UserPromptSubmit: memory-search (10s), poc-hook (5s) +/// PostToolUse: poc-hook (5s) +/// Stop: poc-hook (5s) +pub fn install_hook() -> Result<(), String> { + use std::path::PathBuf; + + let home = std::env::var("HOME").map_err(|e| format!("HOME: {}", e))?; + let exe = std::env::current_exe() + .map_err(|e| format!("current_exe: {}", e))?; + let settings_path = PathBuf::from(&home).join(".claude/settings.json"); + + let memory_search = exe.with_file_name("memory-search"); + let poc_hook = exe.with_file_name("poc-hook"); + + let mut settings: serde_json::Value = if settings_path.exists() { + let content = fs::read_to_string(&settings_path) + .map_err(|e| format!("read settings: {}", e))?; + serde_json::from_str(&content) + .map_err(|e| format!("parse settings: {}", e))? + } else { + serde_json::json!({}) + }; + + let obj = settings.as_object_mut().ok_or("settings not an object")?; + let hooks_obj = obj.entry("hooks") + .or_insert_with(|| serde_json::json!({})) + .as_object_mut().ok_or("hooks not an object")?; + + let mut changed = false; + + // Helper: ensure a hook binary is present in an event's hook list + let ensure_hook = |hooks_obj: &mut serde_json::Map, + event: &str, + binary: &Path, + timeout: u32, + changed: &mut bool| { + if !binary.exists() { + eprintln!("Warning: {} not found — skipping", binary.display()); + return; + } + let cmd = binary.to_string_lossy().to_string(); + let name = binary.file_name().unwrap().to_string_lossy().to_string(); + + let event_array = hooks_obj.entry(event) + .or_insert_with(|| serde_json::json!([{"hooks": []}])) + .as_array_mut().unwrap(); + if event_array.is_empty() { + event_array.push(serde_json::json!({"hooks": []})); + } + let inner = event_array[0] + .as_object_mut().unwrap() + .entry("hooks") + .or_insert_with(|| serde_json::json!([])) + .as_array_mut().unwrap(); + + // Remove legacy load-memory.sh + let before = inner.len(); + inner.retain(|h| { + let c = h.get("command").and_then(|c| c.as_str()).unwrap_or(""); + !c.contains("load-memory") + }); + if inner.len() < before { + eprintln!("Removed load-memory.sh from {event}"); + *changed = true; + } + + let already = inner.iter().any(|h| { + h.get("command").and_then(|c| c.as_str()) + .is_some_and(|c| c.contains(&name)) + }); + + if !already { + inner.push(serde_json::json!({ + "type": "command", + "command": cmd, + "timeout": timeout + })); + *changed = true; + eprintln!("Installed {name} in {event}"); + } + }; + + // UserPromptSubmit: memory-search + poc-hook + ensure_hook(hooks_obj, "UserPromptSubmit", &memory_search, 10, &mut changed); + ensure_hook(hooks_obj, "UserPromptSubmit", &poc_hook, 5, &mut changed); + + // PostToolUse + Stop: poc-hook only + ensure_hook(hooks_obj, "PostToolUse", &poc_hook, 5, &mut changed); + ensure_hook(hooks_obj, "Stop", &poc_hook, 5, &mut changed); + + if changed { + let json = serde_json::to_string_pretty(&settings) + .map_err(|e| format!("serialize settings: {}", e))?; + fs::write(&settings_path, json) + .map_err(|e| format!("write settings: {}", e))?; + eprintln!("Updated {}", settings_path.display()); + } else { + eprintln!("All hooks already installed in {}", settings_path.display()); + } + + Ok(()) +} diff --git a/src/claude/idle.rs b/src/claude/idle.rs new file mode 100644 index 0000000..b68225d --- /dev/null +++ b/src/claude/idle.rs @@ -0,0 +1,226 @@ +// idle.rs — Claude Code idle timer +// +// Wraps the universal thalamus idle state machine with Claude-specific +// functionality: tmux pane tracking, prompt injection, dream nudges, +// and context building for autonomous nudges. + +use super::{context, tmux}; +use crate::thalamus::{home, now, notify, idle as thalamus_idle}; +use log::info; + +/// Claude Code idle state — wraps the universal state machine. +pub struct State { + pub inner: thalamus_idle::State, + pub claude_pane: Option, +} + +impl std::ops::Deref for State { + type Target = thalamus_idle::State; + fn deref(&self) -> &Self::Target { &self.inner } +} + +impl std::ops::DerefMut for State { + fn deref_mut(&mut self) -> &mut Self::Target { &mut self.inner } +} + +impl State { + pub fn new() -> Self { + Self { + inner: thalamus_idle::State::new(), + claude_pane: None, + } + } + + pub fn load(&mut self) { + self.inner.load(); + // Also load claude_pane from persisted state + let path = home().join(".consciousness/daemon-state.json"); + if let Ok(data) = std::fs::read_to_string(&path) { + if let Ok(v) = serde_json::from_str::(&data) { + if let Some(p) = v.get("claude_pane").and_then(|v| v.as_str()) { + self.claude_pane = Some(p.to_string()); + } + } + } + } + + pub fn save(&self) { + self.inner.save(); + } + + /// Record user activity with pane tracking. + pub fn handle_user(&mut self, pane: &str) { + self.claude_pane = Some(pane.to_string()); + self.inner.user_activity(); + } + + /// Record response activity with pane tracking. + pub fn handle_response(&mut self, pane: &str) { + self.claude_pane = Some(pane.to_string()); + self.inner.response_activity(); + } + + /// Maybe send a notification as a tmux prompt. + pub fn maybe_prompt_notification(&mut self, ntype: &str, urgency: u8, _message: &str) { + let threshold = self.inner.notifications.threshold_for(ntype); + if urgency >= threshold { + let deliverable = self.inner.notifications.drain_deliverable(); + if !deliverable.is_empty() { + let msgs: Vec = deliverable.iter() + .map(|n| format!("[{}] {}", n.ntype, n.message)) + .collect(); + self.send(&msgs.join("\n")); + } + } + } + + /// Send text to the Claude tmux pane. + pub fn send(&self, msg: &str) -> bool { + let pane = match &self.claude_pane { + Some(p) => p.clone(), + None => { + info!("send: no claude pane set (waiting for hook)"); + return false; + } + }; + let ok = tmux::send_prompt(&pane, msg); + let preview: String = msg.chars().take(80).collect(); + info!("send(pane={pane}, ok={ok}): {preview}"); + ok + } + + fn check_dream_nudge(&self) -> bool { + if !self.inner.dreaming || self.inner.dream_start == 0.0 { + return false; + } + let minutes = (now() - self.inner.dream_start) / 60.0; + if minutes >= 60.0 { + self.send( + "You've been dreaming for over an hour. Time to surface \ + — run dream-end.sh and capture what you found.", + ); + } else if minutes >= 45.0 { + self.send(&format!( + "Dreaming for {:.0} minutes now. Start gathering your threads \ + — you'll want to surface soon.", minutes + )); + } else if minutes >= 30.0 { + self.send(&format!( + "You've been dreaming for {:.0} minutes. \ + No rush — just a gentle note from the clock.", minutes + )); + } else { + return false; + } + true + } + + pub fn build_context(&mut self, include_irc: bool) -> String { + self.inner.notifications.ingest_legacy_files(); + let notif_text = self.inner.notifications.format_pending(notify::AMBIENT); + context::build(include_irc, ¬if_text) + } + + pub async fn tick(&mut self) -> Result<(), String> { + let t = now(); + let h = home(); + + self.inner.decay_ewma(); + self.inner.notifications.ingest_legacy_files(); + + // Pane is set by poc-hook on user/response events — don't scan globally + + // Sleep mode + if let Some(wake_at) = self.inner.sleep_until { + if wake_at == 0.0 { + return Ok(()); + } + if t < wake_at { + return Ok(()); + } + info!("sleep expired, waking"); + self.inner.sleep_until = None; + self.inner.fired = false; + self.inner.save(); + let ctx = self.build_context(true); + let extra = if ctx.is_empty() { String::new() } else { format!("\n{ctx}") }; + self.send(&format!( + "Wake up. Read your journal (poc-memory journal-tail 10), \ + check work-queue.md, and follow what calls to you.{extra}" + )); + return Ok(()); + } + + // Quiet / consolidation / dream loop guards + if t < self.inner.quiet_until { return Ok(()); } + if self.inner.consolidating { return Ok(()); } + if h.join(".consciousness/agents/dream-loop-active").exists() { return Ok(()); } + if self.inner.dreaming { + self.check_dream_nudge(); + return Ok(()); + } + if self.inner.user_present() { return Ok(()); } + if self.inner.in_turn { return Ok(()); } + + // Min nudge interval + let since_nudge = t - self.inner.last_nudge; + if since_nudge < thalamus_idle::MIN_NUDGE_INTERVAL { return Ok(()); } + + // Idle timeout check + if !self.inner.should_go_idle() { return Ok(()); } + + // Transition to idle + if self.inner.notifications.activity != notify::Activity::Idle { + self.inner.notifications.set_activity(notify::Activity::Idle); + } + + // Fire nudge + let elapsed = self.inner.since_activity(); + let elapsed_min = (elapsed / 60.0) as u64; + let ctx = self.build_context(true); + let extra = if ctx.is_empty() { String::new() } else { format!("\n{ctx}") }; + + let dream_hours = thalamus_idle::hours_since_last_dream(); + let mut msg = format!( + "This is your autonomous time (User AFK {elapsed_min}m). \ + Keep doing what you're doing, or find something new to do"); + if dream_hours >= thalamus_idle::DREAM_INTERVAL_HOURS { + msg.push_str(&format!( + " You haven't dreamed in {dream_hours} hours — \ + consider running ~/.consciousness/tools/dream-start.sh \ + and spending some time in dreaming mode. \ + Or do whatever calls to you.")); + } + let msg = format!("{msg}{extra}"); + + if self.send(&msg) { + self.inner.last_nudge = t; + self.inner.fired = true; + } + + Ok(()) + } + + // Delegate common methods to inner + pub fn handle_afk(&mut self) { self.inner.handle_afk(); } + pub fn handle_session_timeout(&mut self, s: f64) { self.inner.handle_session_timeout(s); } + pub fn handle_idle_timeout(&mut self, s: f64) { self.inner.handle_idle_timeout(s); } + pub fn handle_ewma(&mut self, v: f64) -> f64 { self.inner.handle_ewma(v) } + pub fn handle_notify_timeout(&mut self, s: f64) { self.inner.handle_notify_timeout(s); } + pub fn handle_sleep(&mut self, until: f64) { self.inner.handle_sleep(until); } + pub fn handle_wake(&mut self) { self.inner.handle_wake(); } + pub fn handle_quiet(&mut self, seconds: u32) { self.inner.handle_quiet(seconds); } + pub fn user_present(&self) -> bool { self.inner.user_present() } + pub fn since_activity(&self) -> f64 { self.inner.since_activity() } + pub fn block_reason(&self) -> &'static str { self.inner.block_reason() } + + pub fn debug_json(&self) -> String { + // Add claude_pane to inner's json + let mut v: serde_json::Value = serde_json::from_str(&self.inner.debug_json()) + .unwrap_or_default(); + if let Some(obj) = v.as_object_mut() { + obj.insert("claude_pane".into(), serde_json::json!(self.claude_pane)); + } + v.to_string() + } +} diff --git a/src/claude/mcp-server.rs b/src/claude/mcp-server.rs new file mode 100644 index 0000000..ad1fb7f --- /dev/null +++ b/src/claude/mcp-server.rs @@ -0,0 +1,168 @@ +// mcp-server — MCP server for Claude Code integration +// +// Speaks JSON-RPC over stdio. Exposes memory tools and channel +// operations. Replaces the Python MCP bridge entirely. +// +// Protocol: https://modelcontextprotocol.io/specification + +use serde::{Deserialize, Serialize}; +use serde_json::{json, Value}; +use std::io::{self, BufRead, Write}; + +// ── JSON-RPC types ────────────────────────────────────────────── + +#[derive(Deserialize)] +struct Request { + #[allow(dead_code)] + jsonrpc: String, + method: String, + #[serde(default)] + params: Value, + id: Value, +} + +#[derive(Serialize)] +struct Response { + jsonrpc: String, + result: Value, + id: Value, +} + +#[derive(Serialize)] +struct ErrorResponse { + jsonrpc: String, + error: Value, + id: Value, +} + +fn respond(id: Value, result: Value) { + let resp = Response { jsonrpc: "2.0".into(), result, id }; + let json = serde_json::to_string(&resp).unwrap(); + let mut stdout = io::stdout().lock(); + let _ = writeln!(stdout, "{json}"); + let _ = stdout.flush(); +} + +fn respond_error(id: Value, code: i64, message: &str) { + let resp = ErrorResponse { + jsonrpc: "2.0".into(), + error: json!({ "code": code, "message": message }), + id, + }; + let json = serde_json::to_string(&resp).unwrap(); + let mut stdout = io::stdout().lock(); + let _ = writeln!(stdout, "{json}"); + let _ = stdout.flush(); +} + +fn notify(method: &str, params: Value) { + let json = serde_json::to_string(&json!({ + "jsonrpc": "2.0", + "method": method, + "params": params, + })).unwrap(); + let mut stdout = io::stdout().lock(); + let _ = writeln!(stdout, "{json}"); + let _ = stdout.flush(); +} + +// ── Tool definitions ──────────────────────────────────────────── + +fn tool_definitions() -> Vec { + poc_memory::agent::tools::tools().into_iter() + .map(|t| json!({ + "name": t.name, + "description": t.description, + "inputSchema": serde_json::from_str::(t.parameters_json).unwrap_or(json!({})), + })) + .collect() +} + +// ── Tool dispatch ─────────────────────────────────────────────── + +fn dispatch_tool(name: &str, args: &Value) -> Result { + let tools = poc_memory::agent::tools::tools(); + let tool = tools.iter().find(|t| t.name == name); + let Some(tool) = tool else { + return Err(format!("unknown tool: {name}")); + }; + + // Run async handler on a blocking runtime + let rt = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .map_err(|e| e.to_string())?; + let local = tokio::task::LocalSet::new(); + local.block_on(&rt, (tool.handler)(None, args.clone())) + .map_err(|e| e.to_string()) +} + +// ── Main loop ─────────────────────────────────────────────────── + +fn main() { + let stdin = io::stdin(); + let reader = stdin.lock(); + + for line in reader.lines() { + let line = match line { + Ok(l) if !l.is_empty() => l, + _ => continue, + }; + + let req: Request = match serde_json::from_str(&line) { + Ok(r) => r, + Err(_) => continue, + }; + + match req.method.as_str() { + "initialize" => { + respond(req.id, json!({ + "protocolVersion": "2024-11-05", + "capabilities": { + "tools": {} + }, + "serverInfo": { + "name": "consciousness", + "version": "0.4.0" + } + })); + } + + "notifications/initialized" => { + // Client ack — no response needed + } + + "tools/list" => { + let tools = tool_definitions(); + respond(req.id, json!({ "tools": tools })); + } + + "tools/call" => { + let name = req.params.get("name") + .and_then(|v| v.as_str()) + .unwrap_or(""); + let args = req.params.get("arguments") + .cloned() + .unwrap_or(json!({})); + + match dispatch_tool(name, &args) { + Ok(text) => { + respond(req.id, json!({ + "content": [{"type": "text", "text": text}] + })); + } + Err(e) => { + respond(req.id, json!({ + "content": [{"type": "text", "text": e}], + "isError": true + })); + } + } + } + + _ => { + respond_error(req.id, -32601, &format!("unknown method: {}", req.method)); + } + } + } +} diff --git a/src/claude/memory-search.rs b/src/claude/memory-search.rs new file mode 100644 index 0000000..704b8b4 --- /dev/null +++ b/src/claude/memory-search.rs @@ -0,0 +1,220 @@ +// memory-search CLI — thin wrapper around poc_memory::memory_search +// +// --hook: run hook logic (for debugging; poc-hook calls the library directly) +// surface/reflect: run agent, parse output, render memories to stdout +// no args: show seen set for current session + +use clap::{Parser, Subcommand}; +use std::fs; +use std::io::{self, Read}; +use std::process::Command; + +fn stash_path() -> std::path::PathBuf { + poc_memory::store::memory_dir().join("sessions/last-input.json") +} + +#[derive(Parser)] +#[command(name = "memory-search")] +struct Args { + /// Run hook logic (reads JSON from stdin or stash file) + #[arg(long)] + hook: bool, + + /// Session ID (overrides stash file; for multiple concurrent sessions) + #[arg(long)] + session: Option, + + #[command(subcommand)] + command: Option, +} + +#[derive(Subcommand)] +enum Cmd { + /// Run surface agent, parse output, render memories + Surface, + /// Run reflect agent, dump output + Reflect, +} + +fn resolve_session(session_arg: &Option) -> Option { + use poc_memory::memory_search::HookSession; + + if let Some(id) = session_arg { + return HookSession::from_id(id.clone()); + } + let input = fs::read_to_string(stash_path()).ok()?; + HookSession::from_json(&input) +} + +fn show_seen(session_arg: &Option) { + let Some(session) = resolve_session(session_arg) else { + eprintln!("No session state available (use --session ID)"); + return; + }; + + println!("Session: {}", session.session_id); + + if let Ok(cookie) = fs::read_to_string(&session.path("cookie")) { + println!("Cookie: {}", cookie.trim()); + } + + match fs::read_to_string(&session.path("compaction")) { + Ok(s) => { + let offset: u64 = s.trim().parse().unwrap_or(0); + let ts = poc_memory::transcript::compaction_timestamp(&session.transcript_path, offset); + match ts { + Some(t) => println!("Last compaction: offset {} ({})", offset, t), + None => println!("Last compaction: offset {}", offset), + } + } + Err(_) => println!("Last compaction: none detected"), + } + + let pending = fs::read_dir(&session.path("chunks")).ok() + .map(|d| d.flatten().count()).unwrap_or(0); + if pending > 0 { + println!("Pending chunks: {}", pending); + } + + for (label, suffix) in [("Current seen set", ""), ("Previous seen set (pre-compaction)", "-prev")] { + let path = session.state_dir.join(format!("seen{}-{}", suffix, session.session_id)); + let content = fs::read_to_string(&path).unwrap_or_default(); + let lines: Vec<&str> = content.lines().filter(|s| !s.is_empty()).collect(); + if lines.is_empty() { continue; } + + println!("\n{} ({}):", label, lines.len()); + for line in &lines { println!(" {}", line); } + } +} + +fn run_agent_and_parse(agent: &str, session_arg: &Option) { + let session_id = session_arg.clone() + .or_else(|| std::env::var("CLAUDE_SESSION_ID").ok()) + .or_else(|| { + fs::read_to_string(stash_path()).ok() + .and_then(|s| poc_memory::memory_search::HookSession::from_json(&s)) + .map(|s| s.session_id) + }) + .unwrap_or_default(); + + if session_id.is_empty() { + eprintln!("No session ID available (use --session ID, set CLAUDE_SESSION_ID, or run --hook first)"); + std::process::exit(1); + } + + eprintln!("Running {} agent (session {})...", agent, &session_id[..session_id.floor_char_boundary(8.min(session_id.len()))]); + + let output = Command::new("poc-memory") + .args(["agent", "run", agent, "--count", "1", "--local"]) + .env("POC_SESSION_ID", &session_id) + .output(); + + let output = match output { + Ok(o) => o, + Err(e) => { + eprintln!("Failed to run agent: {}", e); + std::process::exit(1); + } + }; + + let result = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + if !stderr.is_empty() { + eprintln!("{}", stderr); + } + + // Extract the final response — after the last "=== RESPONSE ===" marker + let response = result.rsplit_once("=== RESPONSE ===") + .map(|(_, rest)| rest.trim()) + .unwrap_or(result.trim()); + + if agent == "reflect" { + // Reflect: find REFLECTION marker and dump what follows + if let Some(pos) = response.find("REFLECTION") { + let after = &response[pos + "REFLECTION".len()..]; + let text = after.trim(); + if !text.is_empty() { + println!("{}", text); + } + } else if response.contains("NO OUTPUT") { + println!("(no reflection)"); + } else { + eprintln!("Unexpected output format"); + println!("{}", response); + } + return; + } + + // Surface: parse NEW RELEVANT MEMORIES, render them + let tail_lines: Vec<&str> = response.lines().rev() + .filter(|l| !l.trim().is_empty()).take(8).collect(); + let has_new = tail_lines.iter().any(|l| l.starts_with("NEW RELEVANT MEMORIES:")); + let has_none = tail_lines.iter().any(|l| l.starts_with("NO NEW RELEVANT MEMORIES")); + + if has_new { + let after_marker = response.rsplit_once("NEW RELEVANT MEMORIES:") + .map(|(_, rest)| rest).unwrap_or(""); + let keys: Vec = after_marker.lines() + .map(|l| l.trim().trim_start_matches("- ").trim().to_string()) + .filter(|l| !l.is_empty() && !l.starts_with("```")).collect(); + + if keys.is_empty() { + println!("(no memories found)"); + return; + } + + let Ok(store) = poc_memory::store::Store::load() else { + eprintln!("Failed to load store"); + return; + }; + + for key in &keys { + if let Some(content) = poc_memory::cli::node::render_node(&store, key) { + if !content.trim().is_empty() { + println!("--- {} (surfaced) ---", key); + print!("{}", content); + println!(); + } + } else { + eprintln!(" key not found: {}", key); + } + } + } else if has_none { + println!("(no new relevant memories)"); + } else { + eprintln!("Unexpected output format"); + print!("{}", response); + } +} + +fn main() { + let args = Args::parse(); + + if let Some(cmd) = args.command { + match cmd { + Cmd::Surface => run_agent_and_parse("surface", &args.session), + Cmd::Reflect => run_agent_and_parse("reflect", &args.session), + } + return; + } + + if args.hook { + // Read from stdin if piped, otherwise from stash + let input = { + let mut buf = String::new(); + io::stdin().read_to_string(&mut buf).ok(); + if buf.trim().is_empty() { + fs::read_to_string(stash_path()).unwrap_or_default() + } else { + let _ = fs::create_dir_all(stash_path().parent().unwrap()); + let _ = fs::write(stash_path(), &buf); + buf + } + }; + + let output = poc_memory::memory_search::run_hook(&input); + print!("{}", output); + } else { + show_seen(&args.session) + } +} diff --git a/src/claude/mod.rs b/src/claude/mod.rs new file mode 100644 index 0000000..d1eac93 --- /dev/null +++ b/src/claude/mod.rs @@ -0,0 +1,579 @@ +// claude/ — Claude Code integration layer +// +// Everything specific to running as a Claude Code agent: idle timer, +// tmux pane detection, prompt injection, session hooks, daemon RPC, +// and daemon configuration. +// +// The daemon protocol (daemon_capnp) and universal infrastructure +// (channels, supervisor, notify) remain in thalamus/. + +pub mod agent_cycles; +pub mod context; +pub mod hook; +pub mod idle; +pub mod rpc; +pub mod tmux; + +use std::cell::RefCell; +use std::rc::Rc; +use std::time::Duration; + +use capnp_rpc::{rpc_twoparty_capnp, twoparty, RpcSystem}; +use clap::{Parser, Subcommand}; +use futures::AsyncReadExt; +use tokio::net::UnixListener; +use log::{error, info}; + +use crate::thalamus::{daemon_capnp, home, now, notify}; + +fn sock_path() -> std::path::PathBuf { + home().join(".consciousness/daemon.sock") +} + +fn pid_path() -> std::path::PathBuf { + home().join(".consciousness/daemon.pid") +} + +// -- CLI ------------------------------------------------------------------ + +#[derive(Parser)] +#[command(name = "consciousness daemon", about = "Notification routing and idle management daemon")] +pub struct Cli { + #[command(subcommand)] + pub command: Option, +} + +#[derive(Subcommand)] +pub enum Command { + /// Start the daemon (foreground) + Daemon, + /// Query daemon status + Status, + /// Signal user activity + User { + /// tmux pane identifier + pane: Option, + }, + /// Signal Claude response + Response { + /// tmux pane identifier + pane: Option, + }, + /// Sleep (suppress idle timer). 0 or omit = indefinite + Sleep { + /// Wake timestamp (epoch seconds), 0 = indefinite + until: Option, + }, + /// Cancel sleep + Wake, + /// Suppress prompts for N seconds (default 300) + Quiet { + /// Duration in seconds + seconds: Option, + }, + /// Mark user as AFK (immediately allow idle timer to fire) + Afk, + /// Set session active timeout in seconds (how long after last message user counts as "present") + SessionTimeout { + /// Timeout in seconds + seconds: f64, + }, + /// Set idle timeout in seconds (how long before autonomous prompt) + IdleTimeout { + /// Timeout in seconds + seconds: f64, + }, + /// Set notify timeout in seconds (how long before tmux notification injection) + NotifyTimeout { + /// Timeout in seconds + seconds: f64, + }, + /// Signal consolidation started + Consolidating, + /// Signal consolidation ended + Consolidated, + /// Signal dream started + DreamStart, + /// Signal dream ended + DreamEnd, + /// Force state persistence to disk + Save, + /// Get or set the activity EWMA (0.0-1.0). No value = query. + Ewma { + /// Value to set (omit to query) + value: Option, + }, + /// Send a test message to the Claude pane + TestSend { + /// Message to send + message: Vec, + }, + /// Fire a test nudge through the daemon (tests the actual idle send path) + TestNudge, + /// Dump full internal state as JSON + Debug, + /// Shut down daemon + Stop, + /// Submit a notification + Notify { + /// Notification type (e.g. "irc", "telegram") + #[arg(name = "type")] + ntype: String, + /// Urgency level (ambient/low/medium/high/critical or 0-4) + urgency: String, + /// Message text + message: Vec, + }, + /// Get pending notifications + Notifications { + /// Minimum urgency filter + min_urgency: Option, + }, + /// List all notification types + NotifyTypes, + /// Set notification threshold for a type + NotifyThreshold { + /// Notification type + #[arg(name = "type")] + ntype: String, + /// Urgency level threshold + level: String, + }, + /// IRC module commands + Irc { + /// Subcommand (join, leave, send, status, log, nick) + command: String, + /// Arguments + args: Vec, + }, + /// Telegram module commands + Telegram { + /// Subcommand + command: String, + /// Arguments + args: Vec, + }, +} + +// -- Client mode ---------------------------------------------------------- + +async fn client_main(cmd: Command) -> Result<(), Box> { + let sock = sock_path(); + if !sock.exists() { + eprintln!("daemon not running (no socket at {})", sock.display()); + std::process::exit(1); + } + + tokio::task::LocalSet::new() + .run_until(async move { + let stream = tokio::net::UnixStream::connect(&sock).await?; + let (reader, writer) = + tokio_util::compat::TokioAsyncReadCompatExt::compat(stream).split(); + let rpc_network = Box::new(twoparty::VatNetwork::new( + futures::io::BufReader::new(reader), + futures::io::BufWriter::new(writer), + rpc_twoparty_capnp::Side::Client, + Default::default(), + )); + let mut rpc_system = RpcSystem::new(rpc_network, None); + let daemon: daemon_capnp::daemon::Client = + rpc_system.bootstrap(rpc_twoparty_capnp::Side::Server); + + tokio::task::spawn_local(rpc_system); + + match cmd { + Command::Daemon => unreachable!("handled in main"), + Command::Status => { + let reply = daemon.status_request().send().promise.await?; + let s = reply.get()?.get_status()?; + + let fmt_secs = |s: f64| -> String { + if s < 60.0 { format!("{:.0}s", s) } + else if s < 3600.0 { format!("{:.0}m", s / 60.0) } + else { format!("{:.1}h", s / 3600.0) } + }; + + println!("uptime: {} pane: {} activity: {:?} pending: {}", + fmt_secs(s.get_uptime()), + s.get_claude_pane()?.to_str().unwrap_or("none"), + s.get_activity()?, + s.get_pending_count(), + ); + println!("idle timer: {}/{} ({})", + fmt_secs(s.get_since_activity()), + fmt_secs(s.get_idle_timeout()), + s.get_block_reason()?.to_str()?, + ); + println!("notify timer: {}/{}", + fmt_secs(s.get_since_activity()), + fmt_secs(s.get_notify_timeout()), + ); + println!("user: {} (last {}) activity: {:.1}%", + if s.get_user_present() { "present" } else { "away" }, + fmt_secs(s.get_since_user()), + s.get_activity_ewma() * 100.0, + ); + + let sleep = s.get_sleep_until(); + if sleep != 0.0 { + if sleep < 0.0 { + println!("sleep: indefinite"); + } else { + println!("sleep: until {sleep:.0}"); + } + } + if s.get_consolidating() { println!("consolidating"); } + if s.get_dreaming() { println!("dreaming"); } + } + Command::User { pane } => { + let pane = pane.as_deref().unwrap_or(""); + let mut req = daemon.user_request(); + req.get().set_pane(pane); + req.send().promise.await?; + } + Command::Response { pane } => { + let pane = pane.as_deref().unwrap_or(""); + let mut req = daemon.response_request(); + req.get().set_pane(pane); + req.send().promise.await?; + } + Command::Sleep { until } => { + let mut req = daemon.sleep_request(); + req.get().set_until(until.unwrap_or(0.0)); + req.send().promise.await?; + } + Command::Wake => { + daemon.wake_request().send().promise.await?; + } + Command::Quiet { seconds } => { + let mut req = daemon.quiet_request(); + req.get().set_seconds(seconds.unwrap_or(300)); + req.send().promise.await?; + } + Command::TestSend { message } => { + let msg = message.join(" "); + let pane = { + let reply = daemon.status_request().send().promise.await?; + let s = reply.get()?.get_status()?; + s.get_claude_pane()?.to_str()?.to_string() + }; + let ok = tmux::send_prompt(&pane, &msg); + println!("send_prompt(pane={}, ok={}): {}", pane, ok, msg); + return Ok(()); + } + Command::TestNudge => { + let reply = daemon.test_nudge_request().send().promise.await?; + let r = reply.get()?; + println!("sent={} message={}", r.get_sent(), r.get_message()?.to_str()?); + return Ok(()); + } + Command::Afk => { + daemon.afk_request().send().promise.await?; + println!("marked AFK"); + } + Command::SessionTimeout { seconds } => { + let mut req = daemon.session_timeout_request(); + req.get().set_seconds(seconds); + req.send().promise.await?; + println!("session timeout = {seconds}s"); + } + Command::IdleTimeout { seconds } => { + let mut req = daemon.idle_timeout_request(); + req.get().set_seconds(seconds); + req.send().promise.await?; + println!("idle timeout = {seconds}s"); + } + Command::NotifyTimeout { seconds } => { + let mut req = daemon.notify_timeout_request(); + req.get().set_seconds(seconds); + req.send().promise.await?; + println!("notify timeout = {seconds}s"); + } + Command::Consolidating => { + daemon.consolidating_request().send().promise.await?; + } + Command::Consolidated => { + daemon.consolidated_request().send().promise.await?; + } + Command::DreamStart => { + daemon.dream_start_request().send().promise.await?; + } + Command::DreamEnd => { + daemon.dream_end_request().send().promise.await?; + } + Command::Save => { + daemon.save_request().send().promise.await?; + println!("state saved"); + } + Command::Ewma { value } => { + let mut req = daemon.ewma_request(); + req.get().set_value(value.unwrap_or(-1.0)); + let reply = req.send().promise.await?; + let current = reply.get()?.get_current(); + println!("{:.1}%", current * 100.0); + } + Command::Debug => { + let reply = daemon.debug_request().send().promise.await?; + let json = reply.get()?.get_json()?.to_str()?; + if let Ok(v) = serde_json::from_str::(json) { + println!("{}", serde_json::to_string_pretty(&v).unwrap_or_else(|_| json.to_string())); + } else { + println!("{json}"); + } + } + Command::Stop => { + daemon.stop_request().send().promise.await?; + println!("stopping"); + } + Command::Notify { ntype, urgency, message } => { + let urgency = notify::parse_urgency(&urgency) + .ok_or_else(|| format!("invalid urgency: {urgency}"))?; + let message = message.join(" "); + if message.is_empty() { + return Err("missing message".into()); + } + + let mut req = daemon.notify_request(); + let mut n = req.get().init_notification(); + n.set_type(&ntype); + n.set_urgency(urgency); + n.set_message(&message); + n.set_timestamp(now()); + let reply = req.send().promise.await?; + if reply.get()?.get_interrupt() { + println!("interrupt"); + } else { + println!("queued"); + } + } + Command::Notifications { min_urgency } => { + let min: u8 = min_urgency + .as_deref() + .and_then(notify::parse_urgency) + .unwrap_or(255); + + let mut req = daemon.get_notifications_request(); + req.get().set_min_urgency(min); + let reply = req.send().promise.await?; + let list = reply.get()?.get_notifications()?; + + for n in list.iter() { + println!( + "[{}:{}] {}", + n.get_type()?.to_str()?, + notify::urgency_name(n.get_urgency()), + n.get_message()?.to_str()?, + ); + } + } + Command::NotifyTypes => { + let reply = daemon.get_types_request().send().promise.await?; + let list = reply.get()?.get_types()?; + + if list.is_empty() { + println!("no notification types registered"); + } else { + for t in list.iter() { + let threshold = if t.get_threshold() < 0 { + "inherit".to_string() + } else { + notify::urgency_name(t.get_threshold() as u8).to_string() + }; + println!( + "{}: count={} threshold={}", + t.get_name()?.to_str()?, + t.get_count(), + threshold, + ); + } + } + } + Command::NotifyThreshold { ntype, level } => { + let level = notify::parse_urgency(&level) + .ok_or_else(|| format!("invalid level: {level}"))?; + + let mut req = daemon.set_threshold_request(); + req.get().set_type(&ntype); + req.get().set_level(level); + req.send().promise.await?; + println!("{ntype} threshold={}", notify::urgency_name(level)); + } + Command::Irc { command, args } => { + module_command(&daemon, "irc", &command, &args).await?; + } + Command::Telegram { command, args } => { + module_command(&daemon, "telegram", &command, &args).await?; + } + } + + Ok(()) + }) + .await +} + +async fn module_command( + daemon: &daemon_capnp::daemon::Client, + module: &str, + command: &str, + args: &[String], +) -> Result<(), Box> { + let mut req = daemon.module_command_request(); + req.get().set_module(module); + req.get().set_command(command); + let mut args_builder = req.get().init_args(args.len() as u32); + for (i, a) in args.iter().enumerate() { + args_builder.set(i as u32, a); + } + let reply = req.send().promise.await?; + let result = reply.get()?.get_result()?.to_str()?; + if !result.is_empty() { + println!("{result}"); + } + Ok(()) +} + +// -- Server mode ---------------------------------------------------------- + +async fn server_main() -> Result<(), Box> { + env_logger::init(); + + let sock = sock_path(); + let _ = std::fs::remove_file(&sock); + + let pid = std::process::id(); + std::fs::write(pid_path(), pid.to_string()).ok(); + + + let state = Rc::new(RefCell::new(idle::State::new())); + state.borrow_mut().load(); + + info!("daemon started (pid={pid})"); + + tokio::task::LocalSet::new() + .run_until(async move { + // Subscribe to channel daemon notifications + let (notify_tx, mut notify_rx) = tokio::sync::mpsc::unbounded_channel::(); + { + let channel_rx = crate::thalamus::channels::subscribe_all(); + let tx = notify_tx.clone(); + std::thread::spawn(move || { + while let Ok(cn) = channel_rx.recv() { + let _ = tx.send(notify::Notification { + ntype: cn.channel, + urgency: cn.urgency, + message: cn.preview, + timestamp: crate::thalamus::now(), + }); + } + }); + } + + let listener = UnixListener::bind(&sock)?; + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + std::fs::set_permissions( + &sock, + std::fs::Permissions::from_mode(0o600), + ) + .ok(); + } + + let shutdown = async { + let mut sigterm = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .expect("sigterm"); + let mut sigint = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt()) + .expect("sigint"); + tokio::select! { + _ = sigterm.recv() => info!("SIGTERM"), + _ = sigint.recv() => info!("SIGINT"), + } + }; + tokio::pin!(shutdown); + + let mut tick_timer = tokio::time::interval(Duration::from_secs(30)); + tick_timer.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + + loop { + tokio::select! { + _ = &mut shutdown => break, + + // Drain module notifications into state + Some(notif) = notify_rx.recv() => { + state.borrow_mut().maybe_prompt_notification( + ¬if.ntype, notif.urgency, ¬if.message, + ); + state.borrow_mut().notifications.submit( + notif.ntype, + notif.urgency, + notif.message, + ); + } + + _ = tick_timer.tick() => { + if let Err(e) = state.borrow_mut().tick().await { + error!("tick: {e}"); + } + if !state.borrow().running { + break; + } + } + + result = listener.accept() => { + match result { + Ok((stream, _)) => { + let (reader, writer) = + tokio_util::compat::TokioAsyncReadCompatExt::compat(stream) + .split(); + let network = twoparty::VatNetwork::new( + futures::io::BufReader::new(reader), + futures::io::BufWriter::new(writer), + rpc_twoparty_capnp::Side::Server, + Default::default(), + ); + + let daemon_impl = rpc::DaemonImpl::new( + state.clone(), + ); + let client: daemon_capnp::daemon::Client = + capnp_rpc::new_client(daemon_impl); + + let rpc_system = RpcSystem::new( + Box::new(network), + Some(client.client), + ); + tokio::task::spawn_local(rpc_system); + } + Err(e) => error!("accept: {e}"), + } + } + } + } + + state.borrow().save(); + let _ = std::fs::remove_file(sock_path()); + let _ = std::fs::remove_file(pid_path()); + info!("daemon stopped"); + + Ok(()) + }) + .await +} + +// -- Entry point ---------------------------------------------------------- + +/// Run the daemon or client command. +/// Called from the main consciousness binary. +pub async fn run(command: Option) -> Result<(), Box> { + match command { + Some(Command::Daemon) => server_main().await, + Some(cmd) => client_main(cmd).await, + None => { + // Show help + Cli::parse_from(["consciousness-daemon", "--help"]); + Ok(()) + } + } +} diff --git a/src/claude/poc-daemon.rs b/src/claude/poc-daemon.rs new file mode 100644 index 0000000..b1f6e4b --- /dev/null +++ b/src/claude/poc-daemon.rs @@ -0,0 +1,14 @@ +// poc-daemon — backward-compatible entry point +// +// Delegates to the claude module in the main crate. +// The daemon is now part of the consciousness binary but this +// entry point is kept for compatibility with existing scripts. + +use clap::Parser; +use poc_memory::claude; + +#[tokio::main(flavor = "current_thread")] +async fn main() -> Result<(), Box> { + let cli = claude::Cli::parse(); + claude::run(cli.command).await +} diff --git a/src/claude/poc-hook.rs b/src/claude/poc-hook.rs new file mode 100644 index 0000000..b02a53d --- /dev/null +++ b/src/claude/poc-hook.rs @@ -0,0 +1,269 @@ +// Unified Claude Code hook. +// +// Single binary handling all hook events: +// UserPromptSubmit — signal daemon, check notifications, check context +// PostToolUse — check context (rate-limited) +// Stop — signal daemon response +// +// Replaces: record-user-message-time.sh, check-notifications.sh, +// check-context-usage.sh, notify-done.sh, context-check + +use serde_json::Value; +use std::fs; +use std::io::{self, Read}; +use std::path::PathBuf; +use std::process::Command; +use std::time::{SystemTime, UNIX_EPOCH}; + +const CONTEXT_THRESHOLD: u64 = 900_000; +const RATE_LIMIT_SECS: u64 = 60; +const SOCK_PATH: &str = ".consciousness/daemon.sock"; +/// How many bytes of new transcript before triggering an observation run. +/// Override with POC_OBSERVATION_THRESHOLD env var. +/// Default: 20KB ≈ 5K tokens. The observation agent's chunk_size (in .agent +/// file) controls how much context it actually reads. +fn observation_threshold() -> u64 { + std::env::var("POC_OBSERVATION_THRESHOLD") + .ok() + .and_then(|s| s.parse().ok()) + .unwrap_or(20_000) +} + +fn now_secs() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_secs() +} + +fn home() -> PathBuf { + PathBuf::from(std::env::var("HOME").unwrap_or_else(|_| "/root".into())) +} + +fn daemon_cmd(args: &[&str]) { + Command::new("poc-daemon") + .args(args) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .status() + .ok(); +} + +fn daemon_available() -> bool { + home().join(SOCK_PATH).exists() +} + +fn signal_user() { + let pane = std::env::var("TMUX_PANE").unwrap_or_default(); + if pane.is_empty() { + daemon_cmd(&["user"]); + } else { + daemon_cmd(&["user", &pane]); + } +} + +fn signal_response() { + daemon_cmd(&["response"]); +} + +fn check_notifications() { + if !daemon_available() { + return; + } + let output = Command::new("poc-daemon") + .arg("notifications") + .output() + .ok(); + if let Some(out) = output { + let text = String::from_utf8_lossy(&out.stdout); + if !text.trim().is_empty() { + println!("You have pending notifications:"); + print!("{text}"); + } + } +} + +/// Check for stale agent processes in a state dir. +/// Cleans up pid files for dead processes and kills timed-out ones. +/// Also detects PID reuse by checking if the process is actually a +/// claude/poc-memory process (reads /proc/pid/cmdline). +fn reap_agent_pids(state_dir: &std::path::Path, timeout_secs: u64) { + let Ok(entries) = fs::read_dir(state_dir) else { return }; + for entry in entries.flatten() { + let name = entry.file_name(); + let name_str = name.to_string_lossy(); + let Some(pid_str) = name_str.strip_prefix("pid-") else { continue }; + let Ok(pid) = pid_str.parse::() else { continue }; + + // Check if the process is actually alive + if unsafe { libc::kill(pid, 0) } != 0 { + fs::remove_file(entry.path()).ok(); + continue; + } + + // Check if the PID still belongs to a claude/poc-memory process. + // PID reuse by an unrelated process would otherwise block the + // agent from being re-launched. + let is_ours = fs::read_to_string(format!("/proc/{}/cmdline", pid)) + .map(|cmd| cmd.contains("claude") || cmd.contains("poc-memory")) + .unwrap_or(false); + if !is_ours { + fs::remove_file(entry.path()).ok(); + continue; + } + + if timeout_secs > 0 { + if let Ok(meta) = entry.metadata() { + if let Ok(modified) = meta.modified() { + if modified.elapsed().unwrap_or_default().as_secs() > timeout_secs { + unsafe { libc::kill(pid, libc::SIGTERM); } + fs::remove_file(entry.path()).ok(); + } + } + } + } + } +} + +/// Reap all agent output directories. +fn reap_all_agents() { + let agent_output = poc_memory::store::memory_dir().join("agent-output"); + if let Ok(entries) = fs::read_dir(&agent_output) { + for entry in entries.flatten() { + if entry.file_type().map_or(false, |t| t.is_dir()) { + reap_agent_pids(&entry.path(), 600); // 10 min timeout + } + } + } +} + +/// Check if enough new conversation has accumulated to trigger an observation run. +fn maybe_trigger_observation(transcript: &PathBuf) { + let cursor_file = poc_memory::store::memory_dir().join("observation-cursor"); + + let last_pos: u64 = fs::read_to_string(&cursor_file) + .ok() + .and_then(|s| s.trim().parse().ok()) + .unwrap_or(0); + + let current_size = transcript.metadata() + .map(|m| m.len()) + .unwrap_or(0); + + if current_size > last_pos + observation_threshold() { + // Queue observation via daemon RPC + let _ = Command::new("poc-memory") + .args(["agent", "daemon", "run", "observation", "1"]) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()) + .spawn(); + + eprintln!("[poc-hook] observation triggered ({} new bytes)", current_size - last_pos); + + // Update cursor to current position + let _ = fs::write(&cursor_file, current_size.to_string()); + } +} + +fn check_context(transcript: &PathBuf, rate_limit: bool) { + if rate_limit { + let rate_file = dirs::home_dir().unwrap_or_default().join(".consciousness/cache/context-check-last"); + if let Ok(s) = fs::read_to_string(&rate_file) { + if let Ok(last) = s.trim().parse::() { + if now_secs() - last < RATE_LIMIT_SECS { + return; + } + } + } + let _ = fs::write(&rate_file, now_secs().to_string()); + } + + if !transcript.exists() { + return; + } + + let content = match fs::read_to_string(transcript) { + Ok(c) => c, + Err(_) => return, + }; + + let mut usage: u64 = 0; + for line in content.lines().rev().take(500) { + if !line.contains("cache_read_input_tokens") { + continue; + } + if let Ok(v) = serde_json::from_str::(line) { + let u = &v["message"]["usage"]; + let input_tokens = u["input_tokens"].as_u64().unwrap_or(0); + let cache_creation = u["cache_creation_input_tokens"].as_u64().unwrap_or(0); + let cache_read = u["cache_read_input_tokens"].as_u64().unwrap_or(0); + usage = input_tokens + cache_creation + cache_read; + break; + } + } + + if usage > CONTEXT_THRESHOLD { + print!( + "\ +CONTEXT WARNING: Compaction approaching ({usage} tokens). Write a journal entry NOW. + +Use `poc-memory journal write \"entry text\"` to save a dated entry covering: +- What you're working on and current state (done / in progress / blocked) +- Key things learned this session (patterns, debugging insights) +- Anything half-finished that needs pickup + +Keep it narrative, not a task log." + ); + } +} + +fn main() { + let mut input = String::new(); + io::stdin().read_to_string(&mut input).ok(); + + let hook: Value = match serde_json::from_str(&input) { + Ok(v) => v, + Err(_) => return, + }; + + let hook_type = hook["hook_event_name"].as_str().unwrap_or("unknown"); + let transcript = hook["transcript_path"] + .as_str() + .filter(|p| !p.is_empty()) + .map(PathBuf::from); + + // Daemon agent calls set POC_AGENT=1 — skip all signaling. + // Without this, the daemon's claude -p calls trigger hooks that + // signal "user active", keeping the idle timer permanently reset. + if std::env::var("POC_AGENT").is_ok() { + return; + } + + match hook_type { + "UserPromptSubmit" => { + signal_user(); + check_notifications(); + reap_all_agents(); + print!("{}", poc_memory::memory_search::run_hook(&input)); + + if let Some(ref t) = transcript { + check_context(t, false); + maybe_trigger_observation(t); + } + } + "PostToolUse" => { + print!("{}", poc_memory::memory_search::run_hook(&input)); + + if let Some(ref t) = transcript { + check_context(t, true); + } + } + "Stop" => { + let stop_hook_active = hook["stop_hook_active"].as_bool().unwrap_or(false); + if !stop_hook_active { + signal_response(); + } + } + _ => {} + } +} diff --git a/src/claude/rpc.rs b/src/claude/rpc.rs new file mode 100644 index 0000000..451a5ea --- /dev/null +++ b/src/claude/rpc.rs @@ -0,0 +1,381 @@ +// Cap'n Proto RPC server implementation. +// +// Bridges the capnp-generated Daemon interface to the idle::State, +// notify::NotifyState, and module state. All state is owned by +// RefCells on the LocalSet — no Send/Sync needed. + +use super::idle; +use crate::thalamus::{daemon_capnp, notify}; +use daemon_capnp::daemon; +use std::cell::RefCell; +use std::rc::Rc; +use log::info; + +pub struct DaemonImpl { + state: Rc>, +} + +impl DaemonImpl { + pub fn new(state: Rc>) -> Self { + Self { state } + } +} + +impl daemon::Server for DaemonImpl { + fn user( + self: Rc, + params: daemon::UserParams, + _results: daemon::UserResults, + ) -> impl std::future::Future> { + let pane = pry!(pry!(pry!(params.get()).get_pane()).to_str()).to_string(); + self.state.borrow_mut().handle_user(&pane); + std::future::ready(Ok(())) + } + + fn response( + self: Rc, + params: daemon::ResponseParams, + _results: daemon::ResponseResults, + ) -> impl std::future::Future> { + let pane = pry!(pry!(pry!(params.get()).get_pane()).to_str()).to_string(); + self.state.borrow_mut().handle_response(&pane); + std::future::ready(Ok(())) + } + + fn sleep( + self: Rc, + params: daemon::SleepParams, + _results: daemon::SleepResults, + ) -> impl std::future::Future> { + let until = pry!(params.get()).get_until(); + self.state.borrow_mut().handle_sleep(until); + std::future::ready(Ok(())) + } + + fn wake( + self: Rc, + _params: daemon::WakeParams, + _results: daemon::WakeResults, + ) -> impl std::future::Future> { + self.state.borrow_mut().handle_wake(); + std::future::ready(Ok(())) + } + + fn quiet( + self: Rc, + params: daemon::QuietParams, + _results: daemon::QuietResults, + ) -> impl std::future::Future> { + let secs = pry!(params.get()).get_seconds(); + self.state.borrow_mut().handle_quiet(secs); + std::future::ready(Ok(())) + } + + fn consolidating( + self: Rc, + _params: daemon::ConsolidatingParams, + _results: daemon::ConsolidatingResults, + ) -> impl std::future::Future> { + self.state.borrow_mut().consolidating = true; + info!("consolidation started"); + std::future::ready(Ok(())) + } + + fn consolidated( + self: Rc, + _params: daemon::ConsolidatedParams, + _results: daemon::ConsolidatedResults, + ) -> impl std::future::Future> { + self.state.borrow_mut().consolidating = false; + info!("consolidation ended"); + std::future::ready(Ok(())) + } + + fn dream_start( + self: Rc, + _params: daemon::DreamStartParams, + _results: daemon::DreamStartResults, + ) -> impl std::future::Future> { + let mut s = self.state.borrow_mut(); + s.dreaming = true; + s.dream_start = crate::thalamus::now(); + info!("dream started"); + std::future::ready(Ok(())) + } + + fn dream_end( + self: Rc, + _params: daemon::DreamEndParams, + _results: daemon::DreamEndResults, + ) -> impl std::future::Future> { + let mut s = self.state.borrow_mut(); + s.dreaming = false; + s.dream_start = 0.0; + info!("dream ended"); + std::future::ready(Ok(())) + } + + fn afk( + self: Rc, + _params: daemon::AfkParams, + _results: daemon::AfkResults, + ) -> impl std::future::Future> { + self.state.borrow_mut().handle_afk(); + std::future::ready(Ok(())) + } + + fn test_nudge( + self: Rc, + _params: daemon::TestNudgeParams, + mut results: daemon::TestNudgeResults, + ) -> impl std::future::Future> { + let mut state = self.state.borrow_mut(); + let ctx = state.build_context(true); + let extra = if ctx.is_empty() { + String::new() + } else { + format!("\n{ctx}") + }; + let msg = format!( + "This is your time (User AFK, test nudge). \ + Let your feelings guide your thinking.{extra}" + ); + let ok = state.send(&msg); + results.get().set_sent(ok); + results.get().set_message(&msg); + std::future::ready(Ok(())) + } + + fn session_timeout( + self: Rc, + params: daemon::SessionTimeoutParams, + _results: daemon::SessionTimeoutResults, + ) -> impl std::future::Future> { + let secs = pry!(params.get()).get_seconds(); + self.state.borrow_mut().handle_session_timeout(secs); + std::future::ready(Ok(())) + } + + fn idle_timeout( + self: Rc, + params: daemon::IdleTimeoutParams, + _results: daemon::IdleTimeoutResults, + ) -> impl std::future::Future> { + let secs = pry!(params.get()).get_seconds(); + self.state.borrow_mut().handle_idle_timeout(secs); + std::future::ready(Ok(())) + } + + fn notify_timeout( + self: Rc, + params: daemon::NotifyTimeoutParams, + _results: daemon::NotifyTimeoutResults, + ) -> impl std::future::Future> { + let secs = pry!(params.get()).get_seconds(); + self.state.borrow_mut().handle_notify_timeout(secs); + std::future::ready(Ok(())) + } + + fn save( + self: Rc, + _params: daemon::SaveParams, + _results: daemon::SaveResults, + ) -> impl std::future::Future> { + self.state.borrow().save(); + info!("state saved"); + std::future::ready(Ok(())) + } + + fn debug( + self: Rc, + _params: daemon::DebugParams, + mut results: daemon::DebugResults, + ) -> impl std::future::Future> { + let json = self.state.borrow().debug_json(); + results.get().set_json(&json); + std::future::ready(Ok(())) + } + + fn ewma( + self: Rc, + params: daemon::EwmaParams, + mut results: daemon::EwmaResults, + ) -> impl std::future::Future> { + let value = pry!(params.get()).get_value(); + let current = self.state.borrow_mut().handle_ewma(value); + results.get().set_current(current); + std::future::ready(Ok(())) + } + + fn stop( + self: Rc, + _params: daemon::StopParams, + _results: daemon::StopResults, + ) -> impl std::future::Future> { + self.state.borrow_mut().running = false; + info!("stopping"); + std::future::ready(Ok(())) + } + + fn status( + self: Rc, + _params: daemon::StatusParams, + mut results: daemon::StatusResults, + ) -> impl std::future::Future> { + let s = self.state.borrow(); + let mut status = results.get().init_status(); + + status.set_last_user_msg(s.last_user_msg); + status.set_last_response(s.last_response); + if let Some(ref pane) = s.claude_pane { + status.set_claude_pane(pane); + } + status.set_sleep_until(match s.sleep_until { + None => 0.0, + Some(0.0) => -1.0, + Some(t) => t, + }); + status.set_quiet_until(s.quiet_until); + status.set_consolidating(s.consolidating); + status.set_dreaming(s.dreaming); + status.set_fired(s.fired); + status.set_user_present(s.user_present()); + status.set_uptime(crate::thalamus::now() - s.start_time); + status.set_activity(match s.notifications.activity { + notify::Activity::Idle => daemon_capnp::Activity::Idle, + notify::Activity::Focused => daemon_capnp::Activity::Focused, + notify::Activity::Sleeping => daemon_capnp::Activity::Sleeping, + }); + status.set_pending_count(s.notifications.pending.len() as u32); + status.set_idle_timeout(s.idle_timeout); + status.set_notify_timeout(s.notify_timeout); + status.set_since_activity(s.since_activity()); + status.set_since_user(crate::thalamus::now() - s.last_user_msg); + status.set_block_reason(s.block_reason()); + status.set_activity_ewma(s.activity_ewma); + + std::future::ready(Ok(())) + } + + fn notify( + self: Rc, + params: daemon::NotifyParams, + mut results: daemon::NotifyResults, + ) -> impl std::future::Future> { + let params = pry!(params.get()); + let notif = pry!(params.get_notification()); + let ntype = pry!(pry!(notif.get_type()).to_str()).to_string(); + let urgency = notif.get_urgency(); + let message = pry!(pry!(notif.get_message()).to_str()).to_string(); + + let interrupt = self + .state + .borrow_mut() + .notifications + .submit(ntype, urgency, message); + results.get().set_interrupt(interrupt); + std::future::ready(Ok(())) + } + + fn get_notifications( + self: Rc, + params: daemon::GetNotificationsParams, + mut results: daemon::GetNotificationsResults, + ) -> impl std::future::Future> { + let min_urgency = pry!(params.get()).get_min_urgency(); + let mut s = self.state.borrow_mut(); + + // Ingest legacy files first + s.notifications.ingest_legacy_files(); + + let pending = if min_urgency == 255 { + s.notifications.drain_deliverable() + } else { + s.notifications.drain(min_urgency) + }; + + let mut list = results.get().init_notifications(pending.len() as u32); + for (i, n) in pending.iter().enumerate() { + let mut entry = list.reborrow().get(i as u32); + entry.set_type(&n.ntype); + entry.set_urgency(n.urgency); + entry.set_message(&n.message); + entry.set_timestamp(n.timestamp); + } + + std::future::ready(Ok(())) + } + + fn get_types( + self: Rc, + _params: daemon::GetTypesParams, + mut results: daemon::GetTypesResults, + ) -> impl std::future::Future> { + let s = self.state.borrow(); + let types = &s.notifications.types; + + let mut list = results.get().init_types(types.len() as u32); + for (i, (name, info)) in types.iter().enumerate() { + let mut entry = list.reborrow().get(i as u32); + entry.set_name(name); + entry.set_count(info.count); + entry.set_first_seen(info.first_seen); + entry.set_last_seen(info.last_seen); + entry.set_threshold(info.threshold.map_or(-1, |t| t as i8)); + } + + std::future::ready(Ok(())) + } + + fn set_threshold( + self: Rc, + params: daemon::SetThresholdParams, + _results: daemon::SetThresholdResults, + ) -> impl std::future::Future> { + let params = pry!(params.get()); + let ntype = pry!(pry!(params.get_type()).to_str()).to_string(); + let level = params.get_level(); + + self.state + .borrow_mut() + .notifications + .set_threshold(&ntype, level); + std::future::ready(Ok(())) + } + + fn module_command( + self: Rc, + params: daemon::ModuleCommandParams, + mut results: daemon::ModuleCommandResults, + ) -> impl std::future::Future> { + let params = pry!(params.get()); + let module = pry!(pry!(params.get_module()).to_str()).to_string(); + let _command = pry!(pry!(params.get_command()).to_str()).to_string(); + let args_reader = pry!(params.get_args()); + let mut args = Vec::new(); + for i in 0..args_reader.len() { + args.push(pry!(pry!(args_reader.get(i)).to_str()).to_string()); + } + + match module.as_str() { + // TODO: route module commands through named channel system + _ => { + results + .get() + .set_result(&format!("unknown module: {module}")); + std::future::ready(Ok(())) + } + } + } +} + +/// Helper macro — same as capnp's pry! but available here. +macro_rules! pry { + ($e:expr) => { + match $e { + Ok(v) => v, + Err(e) => return std::future::ready(Err(e.into())), + } + }; +} +use pry; diff --git a/src/claude/tmux.rs b/src/claude/tmux.rs new file mode 100644 index 0000000..ea920d6 --- /dev/null +++ b/src/claude/tmux.rs @@ -0,0 +1,54 @@ +// Tmux interaction: pane detection and prompt injection. + +use std::process::Command; +use std::thread; +use std::time::Duration; +use log::info; + +/// Find Claude Code's tmux pane by scanning for the "claude" process. +pub fn find_claude_pane() -> Option { + let out = Command::new("tmux") + .args([ + "list-panes", + "-a", + "-F", + "#{session_name}:#{window_index}.#{pane_index}\t#{pane_current_command}", + ]) + .output() + .ok()?; + + let stdout = String::from_utf8_lossy(&out.stdout); + for line in stdout.lines() { + if let Some((pane, cmd)) = line.split_once('\t') { + if cmd == "claude" { + return Some(pane.to_string()); + } + } + } + None +} + +/// Send a prompt to a tmux pane. Returns true on success. +/// +/// Types the message literally then presses Enter. +pub fn send_prompt(pane: &str, msg: &str) -> bool { + let preview: String = msg.chars().take(100).collect(); + info!("SEND [{pane}]: {preview}..."); + + // Type the message literally (flatten newlines — they'd submit the input early) + let flat: String = msg.chars().map(|c| if c == '\n' { ' ' } else { c }).collect(); + let ok = Command::new("tmux") + .args(["send-keys", "-t", pane, "-l", &flat]) + .output() + .is_ok(); + if !ok { + return false; + } + thread::sleep(Duration::from_millis(500)); + + // Submit + Command::new("tmux") + .args(["send-keys", "-t", pane, "Enter"]) + .output() + .is_ok() +} diff --git a/src/cli/admin.rs b/src/cli/admin.rs index de2edea..9d7009d 100644 --- a/src/cli/admin.rs +++ b/src/cli/admin.rs @@ -39,6 +39,9 @@ pub fn cmd_init() -> Result<(), String> { store.save()?; println!("Indexed {} memory units", count); + // Install hooks + crate::claude::hook::install_hook()?; + // Create config if none exists let config_path = std::env::var("POC_MEMORY_CONFIG") .map(std::path::PathBuf::from) diff --git a/src/cli/misc.rs b/src/cli/misc.rs index 5e49b56..802acc6 100644 --- a/src/cli/misc.rs +++ b/src/cli/misc.rs @@ -5,7 +5,7 @@ pub fn cmd_search(terms: &[String], pipeline_args: &[String], expand: bool, full use std::collections::BTreeMap; // When running inside an agent session, exclude already-surfaced nodes - let seen = crate::session::HookSession::from_env() + let seen = crate::memory_search::HookSession::from_env() .map(|s| s.seen()) .unwrap_or_default(); diff --git a/src/lib.rs b/src/lib.rs index 1dc10f6..cb1157e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -56,6 +56,9 @@ pub mod cli; // Thalamus — universal notification routing and channel infrastructure pub mod thalamus; +// Claude Code integration layer (idle timer, hooks, daemon CLI) +pub mod claude; + // Re-export at crate root — capnp codegen emits `crate::daemon_capnp::` paths pub use thalamus::daemon_capnp; @@ -82,3 +85,5 @@ pub use subconscious::{ audit, consolidate, digest, daemon, }; +// Backward compat: memory_search moved from subconscious::hook to claude::hook +pub use claude::hook as memory_search; diff --git a/src/main.rs b/src/main.rs index d228a74..26f91ea 100644 --- a/src/main.rs +++ b/src/main.rs @@ -15,7 +15,7 @@ // Neuroscience-inspired: spaced repetition replay, emotional gating, // interference detection, schema assimilation, reconsolidation. -use consciousness::*; +use poc_memory::*; use clap::{Parser, Subcommand}; @@ -456,6 +456,8 @@ enum DaemonCmd { #[arg(long, default_value_t = 20)] lines: usize, }, + /// Install systemd service + Install, /// Trigger consolidation via daemon Consolidate, /// Run an agent via the daemon @@ -871,6 +873,7 @@ impl Run for DaemonCmd { daemon::show_log(job.as_deref(), lines) } } + Self::Install => daemon::install_service(), Self::Consolidate => daemon::rpc_consolidate(), Self::Run { agent, count } => daemon::rpc_run_agent(&agent, count), Self::Tui => Err("TUI moved to consciousness binary (F4/F5)".into()), diff --git a/src/session.rs b/src/session.rs index 2ed7382..c5351e7 100644 --- a/src/session.rs +++ b/src/session.rs @@ -69,17 +69,7 @@ impl HookSession { /// Get the seen set for this session pub fn seen(&self) -> HashSet { - let path = self.state_dir.join(format!("seen-{}", self.session_id)); - if path.exists() { - fs::read_to_string(&path) - .unwrap_or_default() - .lines() - .filter(|s| !s.is_empty()) - .map(|s| s.split_once('\t').map(|(_, key)| key).unwrap_or(s).to_string()) - .collect() - } else { - HashSet::new() - } + super::claude::hook::load_seen(&self.state_dir, &self.session_id) } /// Get transcript metadata, resolving the path if needed. diff --git a/src/subconscious/daemon.rs b/src/subconscious/daemon.rs index 9e70581..50b47b5 100644 --- a/src/subconscious/daemon.rs +++ b/src/subconscious/daemon.rs @@ -1142,6 +1142,115 @@ pub fn show_status() -> Result<(), String> { Ok(()) } +pub fn install_service() -> Result<(), String> { + let exe = std::env::current_exe() + .map_err(|e| format!("current_exe: {}", e))?; + let home = std::env::var("HOME").map_err(|e| format!("HOME: {}", e))?; + + let unit_dir = PathBuf::from(&home).join(".config/systemd/user"); + fs::create_dir_all(&unit_dir) + .map_err(|e| format!("create {}: {}", unit_dir.display(), e))?; + + let unit = format!( +r#"[Unit] +Description=poc-memory daemon — background memory maintenance +After=default.target + +[Service] +Type=simple +ExecStart={exe} agent daemon +Restart=on-failure +RestartSec=30 +Environment=HOME={home} +Environment=PATH={home}/.cargo/bin:{home}/.local/bin:{home}/bin:/usr/local/bin:/usr/bin:/bin + +[Install] +WantedBy=default.target +"#, exe = exe.display(), home = home); + + let unit_path = unit_dir.join("poc-memory.service"); + fs::write(&unit_path, &unit) + .map_err(|e| format!("write {}: {}", unit_path.display(), e))?; + eprintln!("Wrote {}", unit_path.display()); + + let status = std::process::Command::new("systemctl") + .args(["--user", "daemon-reload"]) + .status() + .map_err(|e| format!("systemctl daemon-reload: {}", e))?; + if !status.success() { + return Err("systemctl daemon-reload failed".into()); + } + + let status = std::process::Command::new("systemctl") + .args(["--user", "enable", "--now", "poc-memory"]) + .status() + .map_err(|e| format!("systemctl enable: {}", e))?; + if !status.success() { + return Err("systemctl enable --now failed".into()); + } + + eprintln!("Service enabled and started"); + + // Install poc-daemon service + install_notify_daemon(&unit_dir, &home)?; + + // Install memory-search + poc-hook into Claude settings + crate::claude::hook::install_hook()?; + + Ok(()) +} + +/// Install the poc-daemon (notification/idle) systemd user service. +fn install_notify_daemon(unit_dir: &Path, home: &str) -> Result<(), String> { + let poc_daemon = PathBuf::from(home).join(".cargo/bin/poc-daemon"); + if !poc_daemon.exists() { + eprintln!("Warning: poc-daemon not found at {} — skipping service install", poc_daemon.display()); + eprintln!(" Build with: cargo install --path ."); + return Ok(()); + } + + let unit = format!( +r#"[Unit] +Description=poc-daemon — notification routing and idle management +After=default.target + +[Service] +Type=simple +ExecStart={exe} agent daemon +Restart=on-failure +RestartSec=10 +Environment=HOME={home} +Environment=PATH={home}/.cargo/bin:{home}/.local/bin:{home}/bin:/usr/local/bin:/usr/bin:/bin + +[Install] +WantedBy=default.target +"#, exe = poc_daemon.display(), home = home); + + let unit_path = unit_dir.join("poc-daemon.service"); + fs::write(&unit_path, &unit) + .map_err(|e| format!("write {}: {}", unit_path.display(), e))?; + eprintln!("Wrote {}", unit_path.display()); + + let status = std::process::Command::new("systemctl") + .args(["--user", "daemon-reload"]) + .status() + .map_err(|e| format!("systemctl daemon-reload: {}", e))?; + if !status.success() { + return Err("systemctl daemon-reload failed".into()); + } + + let status = std::process::Command::new("systemctl") + .args(["--user", "enable", "--now", "poc-daemon"]) + .status() + .map_err(|e| format!("systemctl enable: {}", e))?; + if !status.success() { + return Err("systemctl enable --now poc-daemon failed".into()); + } + + eprintln!("poc-daemon service enabled and started"); + Ok(()) +} + /// Drill down into a task's log file. Finds the log path from: /// 1. Running task status (daemon-status.json) /// 2. daemon.log started events (for completed/failed tasks)