From c9804ecfdf804c1da3ba49addc1772451828c907 Mon Sep 17 00:00:00 2001 From: Huskies Agent Date: Mon, 20 Jul 2026 14:37:52 +0000 Subject: [PATCH] huskies: merge 1236 story Ask "what happened with X" and get a paged, subject-scoped history --- .../agents/pool/pipeline/completion/server.rs | 8 + server/src/chatbot_log.rs | 189 ++++++++++ server/src/crdt_state/lww_maps/history_log.rs | 153 ++++++++ server/src/crdt_state/lww_maps/mod.rs | 2 + server/src/crdt_state/mod.rs | 17 +- server/src/crdt_state/types.rs | 31 ++ server/src/history/mod.rs | 357 ++++++++++++++++++ server/src/http/mcp/dispatch.rs | 7 +- server/src/http/mcp/history_tools.rs | 66 ++++ server/src/http/mcp/mod.rs | 2 + server/src/http/mcp/tools_list/mod.rs | 4 +- .../src/http/mcp/tools_list/system_tools.rs | 48 +++ server/src/llm/chat/run.rs | 9 + .../llm/providers/claude_code/events/mod.rs | 5 +- server/src/llm/providers/claude_code/mod.rs | 27 +- server/src/main.rs | 6 + server/src/startup/project.rs | 1 + server/src/startup/tick_loop.rs | 4 + 18 files changed, 911 insertions(+), 25 deletions(-) create mode 100644 server/src/chatbot_log.rs create mode 100644 server/src/crdt_state/lww_maps/history_log.rs create mode 100644 server/src/history/mod.rs create mode 100644 server/src/http/mcp/history_tools.rs diff --git a/server/src/agents/pool/pipeline/completion/server.rs b/server/src/agents/pool/pipeline/completion/server.rs index 56223fc0..73aed9d7 100644 --- a/server/src/agents/pool/pipeline/completion/server.rs +++ b/server/src/agents/pool/pipeline/completion/server.rs @@ -170,6 +170,14 @@ pub(in crate::agents::pool) async fn run_server_owned_completion( "[agents] Server-owned completion for '{story_id}:{agent_name}': gates_passed={gates_passed}" ); + crate::history::record_agent_run( + story_id, + agent_name, + session_id.as_deref(), + gates_passed, + &gate_output, + ); + // Notify chat transports of the agent completion result. let _ = watcher_tx.send(WatcherEvent::AgentCompleted { story_id: story_id.to_string(), diff --git a/server/src/chatbot_log.rs b/server/src/chatbot_log.rs new file mode 100644 index 00000000..bb219c77 --- /dev/null +++ b/server/src/chatbot_log.rs @@ -0,0 +1,189 @@ +//! Rotated log sink for the chat bot's PTY output. +//! +//! The chat bot runs Claude Code CLI in a PTY (see +//! [`crate::llm::providers::claude_code`]) and previously logged every raw +//! PTY line — spawn commands, reader-thread lifecycle, and truncated +//! passthrough of each NDJSON line — via [`crate::slog!`], which meant this +//! high-volume, low-signal output shared the bounded operational ring buffer +//! and `server.log` with everything else, displacing genuinely operational +//! lines. This sink gives that PTY output its own daily-rotated file +//! (`chatbot-YYYY-MM-DD.log`) instead. + +use std::fs::OpenOptions; +use std::io::Write; +use std::path::{Path, PathBuf}; +use std::sync::{Mutex, OnceLock}; + +/// Number of daily log files to keep on disk before pruning older ones. +const KEEP_DAYS: u64 = 7; + +/// Internal state for the on-disk log: directory and last-written date. +struct ChatBotLogState { + dir: Option, + /// `YYYY-MM-DD` of the last written entry — used to detect day rollover. + last_date: String, +} + +/// Daily-rotated on-disk sink for the chat bot's PTY output. +pub struct ChatBotLog { + state: Mutex, +} + +impl ChatBotLog { + fn new() -> Self { + Self { + state: Mutex::new(ChatBotLogState { + dir: None, + last_date: String::new(), + }), + } + } + + /// Set the directory for daily-rotated chat bot log files. + /// + /// Files are written as `chatbot-YYYY-MM-DD.log` inside `dir`. Files + /// older than [`KEEP_DAYS`] are pruned immediately and again on each day + /// rollover. Call once at startup after the project root is known. + pub fn set_log_dir(&self, dir: PathBuf) { + prune_old_logs(&dir, KEEP_DAYS); + if let Ok(mut state) = self.state.lock() { + state.dir = Some(dir); + } + } + + /// Append a line to today's chat bot log file, prefixed with an ISO 8601 + /// UTC timestamp. No-ops silently until [`set_log_dir`] has been called. + pub fn push_line(&self, message: &str) { + let (log_path, prune_dir) = match self.state.lock() { + Ok(mut state) => { + if let Some(dir) = state.dir.clone() { + let today = chrono::Utc::now().format("%Y-%m-%d").to_string(); + let path = dir.join(format!("chatbot-{today}.log")); + let maybe_prune = if state.last_date != today { + state.last_date = today; + Some(dir) + } else { + None + }; + (Some(path), maybe_prune) + } else { + (None, None) + } + } + Err(_) => (None, None), + }; + + if let Some(ref dir) = prune_dir { + prune_old_logs(dir, KEEP_DAYS); + } + + if let Some(ref path) = log_path { + let timestamp = chrono::Utc::now().format("%Y-%m-%dT%H:%M:%SZ").to_string(); + if let Ok(mut file) = OpenOptions::new().create(true).append(true).open(path) { + let _ = writeln!(file, "{timestamp} [pty-debug] {message}"); + } + } + } +} + +static GLOBAL: OnceLock = OnceLock::new(); + +/// Access the process-wide chat bot PTY log sink. +pub fn global() -> &'static ChatBotLog { + GLOBAL.get_or_init(ChatBotLog::new) +} + +/// Delete daily `chatbot-*.log` files older than `keep_days` from `dir`. +fn prune_old_logs(dir: &Path, keep_days: u64) { + let cutoff = chrono::Utc::now() + .checked_sub_signed(chrono::Duration::days(keep_days as i64)) + .map(|t| t.format("%Y-%m-%d").to_string()) + .unwrap_or_default(); + + let Ok(entries) = std::fs::read_dir(dir) else { + return; + }; + for entry in entries.filter_map(|e| e.ok()) { + let path = entry.path(); + let Some(name) = path.file_name().and_then(|n| n.to_str()) else { + continue; + }; + // Match "chatbot-YYYY-MM-DD.log" + if name.starts_with("chatbot-") && name.ends_with(".log") && name.len() == 22 { + // SAFETY: "chatbot-" is 8 ASCII bytes, ".log" is 4, total 22 chars + // means the middle 10 bytes are the date "YYYY-MM-DD" — all + // ASCII, safe to slice. + if let Some(date_part) = name.get(8..18) + && date_part < cutoff.as_str() + { + let _ = std::fs::remove_file(&path); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn fresh_sink() -> ChatBotLog { + ChatBotLog::new() + } + + #[test] + fn push_line_before_set_log_dir_is_a_noop() { + let sink = fresh_sink(); + // Must not panic when no directory has been configured yet. + sink.push_line("hello"); + } + + #[test] + fn push_line_writes_to_rotated_file() { + let tmp = std::env::temp_dir().join(format!( + "huskies_chatbot_log_test_{}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + )); + std::fs::create_dir_all(&tmp).unwrap(); + + let sink = fresh_sink(); + sink.set_log_dir(tmp.clone()); + sink.push_line("raw line: {\"type\":\"assistant\"}"); + + let today = chrono::Utc::now().format("%Y-%m-%d").to_string(); + let path = tmp.join(format!("chatbot-{today}.log")); + let contents = std::fs::read_to_string(&path).unwrap(); + + let _ = std::fs::remove_dir_all(&tmp); + + assert!(contents.contains("[pty-debug]")); + assert!(contents.contains("raw line: {\"type\":\"assistant\"}")); + } + + #[test] + fn does_not_write_into_shared_ring_buffer() { + let tmp = std::env::temp_dir().join(format!( + "huskies_chatbot_log_isolation_test_{}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_nanos() + )); + std::fs::create_dir_all(&tmp).unwrap(); + + let sink = fresh_sink(); + sink.set_log_dir(tmp.clone()); + let marker = "chatbot_isolation_marker_9f31a"; + sink.push_line(marker); + + let _ = std::fs::remove_dir_all(&tmp); + + let ring_hits = crate::log_buffer::global().get_recent(1000, Some(marker), None); + assert!( + ring_hits.is_empty(), + "chat bot PTY output must not land in the shared server log ring buffer" + ); + } +} diff --git a/server/src/crdt_state/lww_maps/history_log.rs b/server/src/crdt_state/lww_maps/history_log.rs new file mode 100644 index 00000000..5c87a656 --- /dev/null +++ b/server/src/crdt_state/lww_maps/history_log.rs @@ -0,0 +1,153 @@ +//! Read/write helpers for the `history_log` append-only list in the CRDT document. +//! +//! Every chat turn, agent run, and pipeline transition is appended as a +//! [`HistoryEntryCrdt`][super::super::types::HistoryEntryCrdt] entry, scoped +//! to a subject (story, sled, or project). Entries are never updated or +//! tombstoned — the list is strictly grow-only, mirroring `event_log.rs`. + +use bft_json_crdt::json_crdt::{JsonValue, *}; +use bft_json_crdt::op::ROOT_ID; +use serde_json::json; + +use super::super::state::{apply_and_persist, get_crdt}; +use super::super::types::HistoryEntryCrdt; + +/// Raw history entry extracted from the CRDT document. +pub struct HistoryEntryRaw { + /// Monotonic sequence number for the recording sled (0-based). + pub event_seq: u64, + /// Hex-encoded Ed25519 public key of the sled that wrote this entry. + pub sled_id: String, + /// Unix timestamp (seconds) when the entry was recorded. + pub timestamp: f64, + /// Subject kind: `"story"`, `"sled"`, or `"project"`. + pub subject_type: String, + /// Subject identifier. + pub subject_id: String, + /// Entry kind: `"pipeline_transition"`, `"chat_turn"`, or `"agent_run"`. + pub kind: String, + /// Short human-readable summary. + pub summary: String, + /// JSON-encoded full payload. + pub detail: String, +} + +/// Append a new history entry to the CRDT, computing the monotonic `event_seq` +/// atomically while the CRDT lock is held. No-ops silently when the CRDT is +/// not yet initialised. +#[allow(clippy::too_many_arguments)] +pub fn append_history_entry( + sled_id: &str, + timestamp: f64, + subject_type: &str, + subject_id: &str, + kind: &str, + summary: &str, + detail: &str, +) { + let Some(state_mutex) = get_crdt() else { + return; + }; + let Ok(mut state) = state_mutex.lock() else { + return; + }; + + // Count existing entries for this sled while holding the lock so the seq + // is computed and used in the same critical section — no TOCTOU gap. + let event_seq = state + .crdt + .doc + .history_log + .iter() + .filter(|e| matches!(e.sled_id.view(), JsonValue::String(s) if s == sled_id)) + .count() as f64; + + // Append after the last existing entry so the list stays in insertion order. + let total_len = state.crdt.doc.history_log.view().len(); + let after = if total_len > 0 { + super::list_id_at(&state.crdt.doc.history_log, total_len - 1).unwrap_or(ROOT_ID) + } else { + ROOT_ID + }; + + let entry: JsonValue = json!({ + "event_seq": event_seq, + "sled_id": sled_id, + "timestamp": timestamp, + "subject_type": subject_type, + "subject_id": subject_id, + "kind": kind, + "summary": summary, + "detail": detail, + }) + .into(); + + apply_and_persist(&mut state, |s| s.crdt.doc.history_log.insert(after, entry)); +} + +/// Read all history entries from the CRDT document. +/// +/// Entries with a missing or empty `sled_id` are silently skipped. Order +/// reflects CRDT insertion order (RGA list semantics) — callers that need a +/// deterministic global order should sort by `(timestamp, sled_id, event_seq)`. +pub fn read_all_history_entries() -> Vec { + let Some(state_mutex) = get_crdt() else { + return Vec::new(); + }; + let Ok(state) = state_mutex.lock() else { + return Vec::new(); + }; + state + .crdt + .doc + .history_log + .iter() + .filter_map(extract_entry) + .collect() +} + +/// Convert a CRDT history entry to its read-side representation. +fn extract_entry(e: &HistoryEntryCrdt) -> Option { + let event_seq = match e.event_seq.view() { + JsonValue::Number(n) => n as u64, + _ => return None, + }; + let sled_id = match e.sled_id.view() { + JsonValue::String(s) if !s.is_empty() => s, + _ => return None, + }; + let timestamp = match e.timestamp.view() { + JsonValue::Number(n) => n, + _ => 0.0, + }; + let subject_type = match e.subject_type.view() { + JsonValue::String(s) => s, + _ => String::new(), + }; + let subject_id = match e.subject_id.view() { + JsonValue::String(s) => s, + _ => String::new(), + }; + let kind = match e.kind.view() { + JsonValue::String(s) => s, + _ => String::new(), + }; + let summary = match e.summary.view() { + JsonValue::String(s) => s, + _ => String::new(), + }; + let detail = match e.detail.view() { + JsonValue::String(s) => s, + _ => String::new(), + }; + Some(HistoryEntryRaw { + event_seq, + sled_id, + timestamp, + subject_type, + subject_id, + kind, + summary, + detail, + }) +} diff --git a/server/src/crdt_state/lww_maps/mod.rs b/server/src/crdt_state/lww_maps/mod.rs index 840ee191..432fccfa 100644 --- a/server/src/crdt_state/lww_maps/mod.rs +++ b/server/src/crdt_state/lww_maps/mod.rs @@ -16,6 +16,7 @@ mod active_agents; mod agent_throttle; mod event_log; mod gateway_projects; +mod history_log; mod llm_sessions; mod merge_jobs; mod test_jobs; @@ -37,6 +38,7 @@ pub use event_log::{ pub use gateway_projects::{ delete_gateway_project, read_all_gateway_projects, read_gateway_project, write_gateway_project, }; +pub use history_log::{HistoryEntryRaw, append_history_entry, read_all_history_entries}; pub use llm_sessions::{assemble_and_advance_session, read_llm_session, write_llm_session}; pub use merge_jobs::{delete_merge_job, read_all_merge_jobs, read_merge_job, write_merge_job}; pub use test_jobs::{delete_test_job, read_all_test_jobs, read_test_job, write_test_job}; diff --git a/server/src/crdt_state/mod.rs b/server/src/crdt_state/mod.rs index 1e9a14ca..bd16b4b3 100644 --- a/server/src/crdt_state/mod.rs +++ b/server/src/crdt_state/mod.rs @@ -28,14 +28,15 @@ mod write; pub use gateway_config::{read_gateway_active_project, write_gateway_active_project}; pub use lww_maps::{ - EventLogEntryRaw, GAP_PIPELINE_EVENT, append_event_log_entry, append_gap_log_entry, - assemble_and_advance_session, delete_active_agent, delete_agent_throttle, - delete_gateway_project, delete_merge_job, delete_test_job, delete_token_usage, - read_active_agent, read_agent_throttle, read_all_active_agents, read_all_agent_throttles, - read_all_event_log_entries, read_all_gateway_projects, read_all_merge_jobs, read_all_test_jobs, - read_all_token_usage, read_gateway_project, read_llm_session, read_merge_job, read_test_job, - read_token_usage, write_active_agent, write_agent_throttle, write_gateway_project, - write_llm_session, write_merge_job, write_test_job, write_token_usage, + EventLogEntryRaw, GAP_PIPELINE_EVENT, HistoryEntryRaw, append_event_log_entry, + append_gap_log_entry, append_history_entry, assemble_and_advance_session, delete_active_agent, + delete_agent_throttle, delete_gateway_project, delete_merge_job, delete_test_job, + delete_token_usage, read_active_agent, read_agent_throttle, read_all_active_agents, + read_all_agent_throttles, read_all_event_log_entries, read_all_gateway_projects, + read_all_history_entries, read_all_merge_jobs, read_all_test_jobs, read_all_token_usage, + read_gateway_project, read_llm_session, read_merge_job, read_test_job, read_token_usage, + write_active_agent, write_agent_throttle, write_gateway_project, write_llm_session, + write_merge_job, write_test_job, write_token_usage, }; pub use ops::{all_ops_json, apply_remote_op, ops_since, our_vector_clock, subscribe_ops}; pub use presence::{ diff --git a/server/src/crdt_state/types.rs b/server/src/crdt_state/types.rs index 10ee89c8..d7e289d6 100644 --- a/server/src/crdt_state/types.rs +++ b/server/src/crdt_state/types.rs @@ -51,6 +51,9 @@ pub struct PipelineDoc { pub event_log: ListCrdt, /// Per-session LLM context state (high-water marks for event log injection). pub llm_sessions: ListCrdt, + /// Append-only, subject-scoped log of chat turns, agent runs, and + /// pipeline transitions, persisted as CRDT ops (story 1236). + pub history_log: ListCrdt, } /// CRDT entry representing a single persisted pipeline stage-transition event. @@ -79,6 +82,34 @@ pub struct EventLogEntryCrdt { pub pipeline_event: LwwRegisterCrdt, } +/// CRDT entry representing a single persisted history entry — a chat turn, +/// agent run, or pipeline transition — scoped to a subject (story, sled, or +/// project) for the `get_history` / `get_history_entry` MCP tools (story 1236). +/// +/// Entries are append-only, mirroring [`EventLogEntryCrdt`]'s per-sled +/// monotonic `event_seq` scheme so `"{sled_id}:{event_seq}"` is a stable, +/// re-readable pagination cursor and payload ref. +#[add_crdt_fields] +#[derive(Clone, CrdtNode, Debug, Serialize, Deserialize)] +pub struct HistoryEntryCrdt { + /// Monotonic sequence number for this sled (0, 1, 2, …). + pub event_seq: LwwRegisterCrdt, + /// Hex-encoded Ed25519 public key of the sled that recorded this entry. + pub sled_id: LwwRegisterCrdt, + /// Unix timestamp (seconds) when the entry was recorded. + pub timestamp: LwwRegisterCrdt, + /// Subject kind: `"story"`, `"sled"`, or `"project"`. + pub subject_type: LwwRegisterCrdt, + /// Subject identifier (story ID, sled hex ID, or project name/persona). + pub subject_id: LwwRegisterCrdt, + /// Entry kind: `"pipeline_transition"`, `"chat_turn"`, or `"agent_run"`. + pub kind: LwwRegisterCrdt, + /// Short human-readable summary shown in a paged listing. + pub summary: LwwRegisterCrdt, + /// JSON-encoded full payload returned by `get_history_entry`. + pub detail: LwwRegisterCrdt, +} + /// CRDT entry tracking an LLM session's event-log injection state. /// /// Each session (keyed by `session_id`, typically a Matrix room ID) records the diff --git a/server/src/history/mod.rs b/server/src/history/mod.rs new file mode 100644 index 00000000..f5304b7f --- /dev/null +++ b/server/src/history/mod.rs @@ -0,0 +1,357 @@ +//! History — subject-scoped, cursor-paged timeline over chat turns, agent +//! runs, and pipeline transitions (story 1236). +//! +//! Every entry is persisted to the CRDT `history_log` list (see +//! [`crate::crdt_state`]) so it survives restarts and replicates across +//! sleds. Callers query via [`get_history`], which returns short typed +//! summaries plus an opaque `ref` string; the full payload for a single +//! entry is fetched on demand via [`get_history_entry`]. + +use chrono::Utc; + +/// One entry in a paged history listing: a short typed summary plus a `ref` +/// that [`get_history_entry`] can resolve to the full payload. +pub struct HistoryEntry { + pub kind: String, + pub subject_type: String, + pub subject_id: String, + pub at: chrono::DateTime, + pub summary: String, + /// Opaque cursor-safe reference: `"{sled_id}:{event_seq}"`. + pub entry_ref: String, +} + +/// A page of history entries plus an opaque cursor for the next page. +pub struct HistoryPage { + pub entries: Vec, + pub next_cursor: Option, +} + +/// Sort key used both for global ordering and for the pagination cursor. +fn sort_key(e: &crate::crdt_state::HistoryEntryRaw) -> (i64, String, u64) { + (e.timestamp as i64, e.sled_id.clone(), e.event_seq) +} + +fn encode_cursor(key: &(i64, String, u64)) -> String { + format!("{}:{}:{}", key.0, key.1, key.2) +} + +fn decode_cursor(cursor: &str) -> Option<(i64, String, u64)> { + let mut parts = cursor.splitn(3, ':'); + let ts: i64 = parts.next()?.parse().ok()?; + let sled_id = parts.next()?.to_string(); + let seq: u64 = parts.next()?.parse().ok()?; + Some((ts, sled_id, seq)) +} + +/// Query a time-ordered, paged history for a subject. +/// +/// `since`/`until` are inclusive Unix-second bounds (`None` = unbounded). +/// `cursor` resumes after the last entry returned by a previous call; +/// `limit` is clamped to `[1, 500]`. +pub fn get_history( + subject_type: &str, + subject_id: &str, + since: Option, + until: Option, + cursor: Option<&str>, + limit: usize, +) -> HistoryPage { + let limit = limit.clamp(1, 500); + let after_key = cursor.and_then(decode_cursor); + + let mut matching: Vec = + crate::crdt_state::read_all_history_entries() + .into_iter() + .filter(|e| subject_matches(e, subject_type, subject_id)) + .filter(|e| since.is_none_or(|s| e.timestamp as i64 >= s)) + .filter(|e| until.is_none_or(|u| e.timestamp as i64 <= u)) + .collect(); + + matching.sort_by_key(sort_key); + + let start = match after_key { + Some(after) => matching + .iter() + .position(|e| sort_key(e) > after) + .unwrap_or(matching.len()), + None => 0, + }; + + let page: Vec<_> = matching[start..].iter().take(limit).collect(); + let next_cursor = if start + page.len() < matching.len() { + page.last().map(|e| encode_cursor(&sort_key(e))) + } else { + None + }; + + let entries = page + .into_iter() + .map(|e| HistoryEntry { + kind: e.kind.clone(), + subject_type: e.subject_type.clone(), + subject_id: e.subject_id.clone(), + at: chrono::DateTime::from_timestamp(e.timestamp as i64, 0).unwrap_or_default(), + summary: e.summary.clone(), + entry_ref: format!("{}:{}", e.sled_id, e.event_seq), + }) + .collect(); + + HistoryPage { + entries, + next_cursor, + } +} + +/// Return true when `entry` belongs to the requested subject. +/// +/// `"sled"` subjects match on the recording sled's own ID (who did it); +/// `"story"` subjects match on the entry's declared subject dimension (what +/// it happened to). A `"project"` query returns every entry recorded by this +/// server instance, since each huskies server is scoped to a single project. +fn subject_matches( + e: &crate::crdt_state::HistoryEntryRaw, + subject_type: &str, + subject_id: &str, +) -> bool { + match subject_type { + "sled" | "robot" => e.sled_id == subject_id, + "project" => true, + _ => e.subject_type == subject_type && e.subject_id == subject_id, + } +} + +/// Resolve a `ref` string returned by [`get_history`] into the full payload +/// for that entry. Returns `None` when the ref does not resolve to any known +/// entry. +pub fn get_history_entry(entry_ref: &str) -> Option { + let (sled_id, seq_str) = entry_ref.split_once(':')?; + let seq: u64 = seq_str.parse().ok()?; + crate::crdt_state::read_all_history_entries() + .into_iter() + .find(|e| e.sled_id == sled_id && e.event_seq == seq) + .map(|e| e.detail) +} + +/// Record a pipeline stage transition into the unified history log. +/// +/// Called from a dedicated broadcast subscriber (see +/// [`spawn_history_subscriber`]) so it runs independently of +/// [`crate::event_log`]'s own transition log. +pub(crate) fn record_pipeline_transition(fired: &crate::pipeline_state::TransitionFired) { + let sled_id = crate::crdt_state::our_node_id().unwrap_or_default(); + let timestamp = fired.at.timestamp() as f64; + let from_stage = crate::pipeline_state::stage_label(&fired.before); + let to_stage = crate::pipeline_state::stage_label(&fired.after); + let pipeline_event = crate::pipeline_state::event_label(&fired.event); + let summary = format!( + "{} moved {from_stage} -> {to_stage} ({pipeline_event})", + fired.story_id.0 + ); + let detail = serde_json::json!({ + "story_id": fired.story_id.0, + "from_stage": from_stage, + "to_stage": to_stage, + "pipeline_event": pipeline_event, + }) + .to_string(); + + crate::crdt_state::append_history_entry( + &sled_id, + timestamp, + "story", + &fired.story_id.0, + "pipeline_transition", + &summary, + &detail, + ); +} + +/// Record a completed chat turn into the unified history log. +/// +/// `subject_id` is the persona/session the turn belongs to (e.g. `"timmy"`). +pub(crate) fn record_chat_turn(subject_id: &str, user_message: &str, assistant_reply: &str) { + let sled_id = crate::crdt_state::our_node_id().unwrap_or_default(); + let timestamp = Utc::now().timestamp() as f64; + let summary = truncate(user_message, 120); + let detail = serde_json::json!({ + "user": user_message, + "assistant": assistant_reply, + }) + .to_string(); + + crate::crdt_state::append_history_entry( + &sled_id, + timestamp, + "project", + subject_id, + "chat_turn", + &summary, + &detail, + ); +} + +/// Record a completed agent run into the unified history log. +/// +/// The gate output is truncated to keep the CRDT entry bounded; the full +/// transcript remains on disk under `.huskies/logs/{story_id}/` for deeper +/// inspection via `get_agent_output`. +pub(crate) fn record_agent_run( + story_id: &str, + agent_name: &str, + session_id: Option<&str>, + gates_passed: bool, + gate_output: &str, +) { + let sled_id = crate::crdt_state::our_node_id().unwrap_or_default(); + let timestamp = Utc::now().timestamp() as f64; + let outcome = if gates_passed { "passed" } else { "failed" }; + let summary = format!( + "{agent_name} run on {story_id} {outcome}: {}", + truncate(gate_output, 100) + ); + let detail = serde_json::json!({ + "story_id": story_id, + "agent_name": agent_name, + "session_id": session_id, + "gates_passed": gates_passed, + "gate_output": truncate(gate_output, 4000), + }) + .to_string(); + + crate::crdt_state::append_history_entry( + &sled_id, + timestamp, + "story", + story_id, + "agent_run", + &summary, + &detail, + ); +} + +fn truncate(s: &str, max_chars: usize) -> String { + if s.chars().count() <= max_chars { + s.to_string() + } else { + let truncated: String = s.chars().take(max_chars).collect(); + format!("{truncated}…") + } +} + +/// Spawn a background task that persists every `TransitionFired` event to the +/// unified history log, independently of [`crate::event_log`]'s subscriber. +pub fn spawn_history_subscriber() { + let mut rx = crate::pipeline_state::subscribe_transitions(); + tokio::spawn(async move { + loop { + match rx.recv().await { + Ok(fired) => record_pipeline_transition(&fired), + Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue, + Err(tokio::sync::broadcast::error::RecvError::Closed) => break, + } + } + }); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn record_and_query_pipeline_transition() { + crate::crdt_state::init_for_test(); + let fired = crate::pipeline_state::TransitionFired { + story_id: crate::pipeline_state::StoryId("42_story_test".to_string()), + before: crate::pipeline_state::Stage::Backlog, + after: crate::pipeline_state::Stage::Coding { + claim: None, + plan: crate::pipeline_state::PlanState::Missing, + retries: 0, + }, + event: crate::pipeline_state::PipelineEvent::DepsMet, + at: chrono::Utc::now(), + }; + record_pipeline_transition(&fired); + + let page = get_history("story", "42_story_test", None, None, None, 50); + assert_eq!(page.entries.len(), 1); + assert_eq!(page.entries[0].kind, "pipeline_transition"); + assert!(page.entries[0].summary.contains("42_story_test")); + + let full = get_history_entry(&page.entries[0].entry_ref).unwrap(); + assert!(full.contains("DepsMet")); + } + + #[test] + fn record_and_query_chat_turn() { + crate::crdt_state::init_for_test(); + record_chat_turn("timmy", "what happened with 42?", "here's the history"); + + let page = get_history("project", "timmy", None, None, None, 50); + assert_eq!(page.entries.len(), 1); + assert_eq!(page.entries[0].kind, "chat_turn"); + + let full = get_history_entry(&page.entries[0].entry_ref).unwrap(); + assert!(full.contains("here's the history")); + } + + #[test] + fn record_and_query_agent_run() { + crate::crdt_state::init_for_test(); + record_agent_run( + "42_story_test", + "coder-1", + Some("sess-1"), + true, + "all gates passed", + ); + + let page = get_history("story", "42_story_test", None, None, None, 50); + assert_eq!(page.entries.len(), 1); + assert_eq!(page.entries[0].kind, "agent_run"); + } + + #[test] + fn pagination_cursor_advances() { + crate::crdt_state::init_for_test(); + for i in 0..5 { + record_chat_turn("timmy", &format!("msg {i}"), &format!("reply {i}")); + } + + let page1 = get_history("project", "timmy", None, None, None, 2); + assert_eq!(page1.entries.len(), 2); + assert!(page1.next_cursor.is_some()); + + let page2 = get_history( + "project", + "timmy", + None, + None, + page1.next_cursor.as_deref(), + 2, + ); + assert_eq!(page2.entries.len(), 2); + + let page3 = get_history( + "project", + "timmy", + None, + None, + page2.next_cursor.as_deref(), + 2, + ); + assert_eq!(page3.entries.len(), 1); + assert!(page3.next_cursor.is_none()); + } + + #[test] + fn sled_subject_matches_recording_sled() { + crate::crdt_state::init_for_test(); + record_chat_turn("timmy", "hi", "hello"); + let sled_id = crate::crdt_state::our_node_id().unwrap_or_default(); + + let page = get_history("sled", &sled_id, None, None, None, 50); + assert_eq!(page.entries.len(), 1); + } +} diff --git a/server/src/http/mcp/dispatch.rs b/server/src/http/mcp/dispatch.rs index 201e2449..290894ef 100644 --- a/server/src/http/mcp/dispatch.rs +++ b/server/src/http/mcp/dispatch.rs @@ -3,8 +3,8 @@ use serde_json::Value; use super::{ - agent_tools, diagnostics, git_tools, merge_tools, qa_tools, shell_tools, status_tools, - story_tools, timer_tools, trigger_tools, wizard_tools, + agent_tools, diagnostics, git_tools, history_tools, merge_tools, qa_tools, shell_tools, + status_tools, story_tools, timer_tools, trigger_tools, wizard_tools, }; use crate::http::context::AppContext; @@ -91,6 +91,9 @@ pub async fn dispatch_tool_call( "get_token_usage" => diagnostics::tool_get_token_usage(&args, ctx), // Chat turn telemetry (story 1209) "chat_telemetry" => diagnostics::tool_chat_telemetry(&args), + // Subject-scoped history (story 1236) + "get_history" => history_tools::tool_get_history(&args), + "get_history_entry" => history_tools::tool_get_history_entry(&args), // Delete story "delete_story" => story_tools::tool_delete_story(&args, ctx).await, // Purge story (CRDT tombstone — story 521) diff --git a/server/src/http/mcp/history_tools.rs b/server/src/http/mcp/history_tools.rs new file mode 100644 index 00000000..24dcd74d --- /dev/null +++ b/server/src/http/mcp/history_tools.rs @@ -0,0 +1,66 @@ +//! MCP tools for subject-scoped history queries (story 1236). +//! +//! `get_history` returns a time-ordered, cursor-paged list of short typed +//! summaries for a subject (story, sled, or project); `get_history_entry` +//! resolves the `ref` from one of those summaries into its full payload. + +use serde_json::{Value, json}; + +/// MCP tool: return a paged, subject-scoped history listing. +pub(crate) fn tool_get_history(args: &Value) -> Result { + let subject_type = args + .get("subject_type") + .and_then(|v| v.as_str()) + .ok_or("subject_type is required (one of: story, robot, project)")?; + let subject_id = args + .get("subject_id") + .and_then(|v| v.as_str()) + .ok_or("subject_id is required")?; + if !matches!(subject_type, "story" | "sled" | "robot" | "project") { + return Err(format!( + "subject_type must be one of: story, robot, project (got '{subject_type}')" + )); + } + let since = args.get("since").and_then(|v| v.as_i64()); + let until = args.get("until").and_then(|v| v.as_i64()); + let cursor = args.get("cursor").and_then(|v| v.as_str()); + let limit = args + .get("limit") + .and_then(|v| v.as_u64()) + .map(|n| n as usize) + .unwrap_or(50); + + let page = crate::history::get_history(subject_type, subject_id, since, until, cursor, limit); + + let entries: Vec = page + .entries + .iter() + .map(|e| { + json!({ + "ref": e.entry_ref, + "kind": e.kind, + "subject_type": e.subject_type, + "subject_id": e.subject_id, + "at": e.at.format("%Y-%m-%dT%H:%M:%SZ").to_string(), + "summary": e.summary, + }) + }) + .collect(); + + serde_json::to_string_pretty(&json!({ + "entries": entries, + "next_cursor": page.next_cursor, + })) + .map_err(|e| format!("Serialization error: {e}")) +} + +/// MCP tool: resolve a `ref` from `get_history` into its full payload. +pub(crate) fn tool_get_history_entry(args: &Value) -> Result { + let entry_ref = args + .get("ref") + .and_then(|v| v.as_str()) + .ok_or("ref is required (from a get_history entry)")?; + + crate::history::get_history_entry(entry_ref) + .ok_or_else(|| format!("No history entry found for ref '{entry_ref}'")) +} diff --git a/server/src/http/mcp/mod.rs b/server/src/http/mcp/mod.rs index d3babc0b..d83039fa 100644 --- a/server/src/http/mcp/mod.rs +++ b/server/src/http/mcp/mod.rs @@ -12,6 +12,8 @@ pub mod diagnostics; pub mod dispatch; /// MCP tools for git operations scoped to agent worktrees. pub mod git_tools; +/// MCP tools for subject-scoped history queries (`get_history`, `get_history_entry`). +pub mod history_tools; /// MCP tools for merge status and merge-to-master operations. pub mod merge_tools; /// Task-local progress emitter used to deliver `notifications/progress` diff --git a/server/src/http/mcp/tools_list/mod.rs b/server/src/http/mcp/tools_list/mod.rs index 9aff4232..6edb98e0 100644 --- a/server/src/http/mcp/tools_list/mod.rs +++ b/server/src/http/mcp/tools_list/mod.rs @@ -121,7 +121,9 @@ mod tests { assert!(names.contains(&"gc")); assert!(names.contains(&"chat_telemetry")); assert!(names.contains(&"ask_question")); - assert_eq!(tools.len(), 89); + assert!(names.contains(&"get_history")); + assert!(names.contains(&"get_history_entry")); + assert_eq!(tools.len(), 91); } #[test] diff --git a/server/src/http/mcp/tools_list/system_tools.rs b/server/src/http/mcp/tools_list/system_tools.rs index f62c8661..57d29bf2 100644 --- a/server/src/http/mcp/tools_list/system_tools.rs +++ b/server/src/http/mcp/tools_list/system_tools.rs @@ -120,6 +120,54 @@ pub(super) fn system_tools() -> Vec { } } }), + json!({ + "name": "get_history", + "description": "Return a time-ordered, cursor-paged history for a subject (story, robot, or project) over an optional time range. Each entry is a short typed summary (chat_turn, agent_run, or pipeline_transition) plus a 'ref' string; fetch the full payload for one entry with get_history_entry.", + "inputSchema": { + "type": "object", + "properties": { + "subject_type": { + "type": "string", + "description": "One of: story, robot, project" + }, + "subject_id": { + "type": "string", + "description": "Story ID, robot (sled) hex ID, or project/persona name" + }, + "since": { + "type": "integer", + "description": "Optional Unix-second lower bound (inclusive)" + }, + "until": { + "type": "integer", + "description": "Optional Unix-second upper bound (inclusive)" + }, + "cursor": { + "type": "string", + "description": "Opaque cursor from a previous page's next_cursor to resume from" + }, + "limit": { + "type": "integer", + "description": "Maximum number of entries to return (default 50, max 500)" + } + }, + "required": ["subject_type", "subject_id"] + } + }), + json!({ + "name": "get_history_entry", + "description": "Resolve a 'ref' string returned by get_history into its full payload.", + "inputSchema": { + "type": "object", + "properties": { + "ref": { + "type": "string", + "description": "The 'ref' value from a get_history entry" + } + }, + "required": ["ref"] + } + }), json!({ "name": "run_command", "description": "Execute a shell command in an agent's worktree directory. The working_dir must be inside .huskies/worktrees/. Returns stdout, stderr, exit_code, and timed_out. Supports SSE streaming (send Accept: text/event-stream) for long-running commands. Dangerous commands (rm -rf /, sudo, etc.) are blocked.", diff --git a/server/src/llm/chat/run.rs b/server/src/llm/chat/run.rs index 3647e9cf..4a2ee434 100644 --- a/server/src/llm/chat/run.rs +++ b/server/src/llm/chat/run.rs @@ -257,6 +257,15 @@ where } else { result.extend(cc_messages); } + + let assistant_reply = result + .iter() + .rev() + .find(|m| m.role == Role::Assistant) + .map(|m| m.content.as_str()) + .unwrap_or_default(); + crate::history::record_chat_turn(persona, &user_message, assistant_reply); + on_update(&result); return Ok(ChatResult { messages: result, diff --git a/server/src/llm/providers/claude_code/events/mod.rs b/server/src/llm/providers/claude_code/events/mod.rs index 74433fc2..6eac3d00 100644 --- a/server/src/llm/providers/claude_code/events/mod.rs +++ b/server/src/llm/providers/claude_code/events/mod.rs @@ -8,7 +8,6 @@ mod tests; use super::parse::{parse_assistant_message, parse_tool_results}; use crate::agents::TokenUsage; use crate::llm::types::Message; -use crate::slog; use std::sync::atomic::{AtomicBool, Ordering}; use stream::handle_stream_event; @@ -36,7 +35,7 @@ pub(super) fn process_json_event( // Capture session_id from the first event that carries it if let Some(tx) = sid_tx.take() { if let Some(sid) = json.get("session_id").and_then(|s| s.as_str()) { - slog!("[pty-debug] CAPTURED session_id: {}", sid); + crate::chatbot_log::global().push_line(&format!("CAPTURED session_id: {sid}")); let _ = tx.send(sid.to_string()); } else { *sid_tx = Some(tx); @@ -45,7 +44,7 @@ pub(super) fn process_json_event( // Detect authentication_failed at the top level of any event. if json.get("error").and_then(|e| e.as_str()) == Some("authentication_failed") { - slog!("[pty-debug] Detected authentication_failed error"); + crate::chatbot_log::global().push_line("Detected authentication_failed error"); auth_failed.store(true, Ordering::Relaxed); } diff --git a/server/src/llm/providers/claude_code/mod.rs b/server/src/llm/providers/claude_code/mod.rs index 6cb3642a..218c7848 100644 --- a/server/src/llm/providers/claude_code/mod.rs +++ b/server/src/llm/providers/claude_code/mod.rs @@ -180,7 +180,8 @@ impl ClaudeCodeProvider { } let captured_session_id = sid_rx.await.ok(); - slog!("[pty-debug] RECEIVED session_id: {:?}", captured_session_id); + crate::chatbot_log::global() + .push_line(&format!("RECEIVED session_id: {captured_session_id:?}")); let usage = usage_rx.await.ok(); let structured_messages: Vec = msg_rx.try_iter().collect(); @@ -308,21 +309,22 @@ fn run_pty_session( // Allow nested spawning when the server itself runs inside Claude Code cmd.env("CLAUDECODE", ""); - slog!( - "[pty-debug] Spawning: claude -p \"{}\" {} {} --output-format stream-json --verbose --include-partial-messages --permission-prompt-tool mcp__huskies__prompt_permission", + crate::chatbot_log::global().push_line(&format!( + "Spawning: claude -p \"{}\" {} {} --output-format stream-json --verbose --include-partial-messages --permission-prompt-tool mcp__huskies__prompt_permission", user_message, resume_session_id .map(|s| format!("--resume {s}")) .unwrap_or_default(), model.map(|m| format!("--model {m}")).unwrap_or_default() - ); + )); let mut child = pair .slave .spawn_command(cmd) .map_err(|e| format!("Failed to spawn claude: {e}"))?; - slog!("[pty-debug] Process spawned, pid: {:?}", child.process_id()); + crate::chatbot_log::global() + .push_line(&format!("Process spawned, pid: {:?}", child.process_id())); drop(pair.slave); let reader = pair @@ -339,23 +341,23 @@ fn run_pty_session( let reader_handle = std::thread::spawn(move || { let buf_reader = BufReader::new(reader); - slog!("[pty-debug] Reader thread started"); + crate::chatbot_log::global().push_line("Reader thread started"); for line in buf_reader.lines() { match line { Ok(l) => { - slog!("[pty-debug] raw line: {}", l); + crate::chatbot_log::global().push_line(&format!("raw line: {l}")); if line_tx.send(Some(l)).is_err() { break; } } Err(e) => { - slog!("[pty-debug] read error: {e}"); + crate::chatbot_log::global().push_line(&format!("read error: {e}")); let _ = line_tx.send(None); break; } } } - slog!("[pty-debug] Reader thread done"); + crate::chatbot_log::global().push_line("Reader thread done"); let _ = line_tx.send(None); }); @@ -382,7 +384,8 @@ fn run_pty_session( while !trimmed.is_char_boundary(end) { end -= 1; } - slog!("[pty-debug] processing: {}...", &trimmed[..end]); + crate::chatbot_log::global() + .push_line(&format!("processing: {}...", &trimmed[..end])); // Try to parse as JSON if let Ok(json) = serde_json::from_str::(trimmed) @@ -477,7 +480,9 @@ fn run_pty_session( && let Some(ref status) = exit_status && !status.success() { - slog!("[pty-debug] Claude Code exited with non-zero status: {status}"); + crate::chatbot_log::global().push_line(&format!( + "Claude Code exited with non-zero status: {status}" + )); return Err(format!("Claude Code crashed (exit status: {status})")); } diff --git a/server/src/main.rs b/server/src/main.rs index 3dc45a3b..73799d40 100644 --- a/server/src/main.rs +++ b/server/src/main.rs @@ -10,6 +10,9 @@ mod agent_log; mod agent_mode; mod agents; mod chat; +/// Chat bot log — daily-rotated sink for the chat bot's PTY output, kept +/// separate from the shared operational ring buffer and `server.log`. +pub mod chatbot_log; #[cfg(test)] mod ci_publish_artifact; mod config; @@ -27,6 +30,9 @@ pub(crate) mod event_log; /// Gateway mode — multi-project reverse proxy that fronts multiple project containers. pub mod gateway; mod gateway_relay; +/// History — subject-scoped, cursor-paged timeline over chat turns, agent +/// runs, and pipeline transitions. +pub(crate) mod history; mod http; mod io; mod llm; diff --git a/server/src/startup/project.rs b/server/src/startup/project.rs index 3c1d2dec..9274f9e6 100644 --- a/server/src/startup/project.rs +++ b/server/src/startup/project.rs @@ -228,6 +228,7 @@ pub(crate) async fn init_subsystems(app_state: &Arc, cwd: &Path, i if let Some(ref root) = *app_state.project_root.lock().unwrap() { let log_dir = root.join(".huskies").join("logs"); let _ = std::fs::create_dir_all(&log_dir); + crate::chatbot_log::global().set_log_dir(log_dir.clone()); log_buffer::global().set_log_dir(log_dir); } diff --git a/server/src/startup/tick_loop.rs b/server/src/startup/tick_loop.rs index f9545972..f74cc159 100644 --- a/server/src/startup/tick_loop.rs +++ b/server/src/startup/tick_loop.rs @@ -42,6 +42,10 @@ pub(crate) fn spawn_event_bridges( // the history survives rebuild_and_restart and replicates across nodes. crate::event_log::spawn_event_log_subscriber(); + // Unified history subscriber: persist every transition into the + // subject-scoped history log alongside chat turns and agent runs (story 1236). + crate::history::spawn_history_subscriber(); + // CRDT → watcher bridge: translate CRDT stage-transition events into // WatcherEvent::WorkItem so downstream consumers (WebSocket, auto-assign) // see a uniform stream regardless of whether the event originated from the