huskies: merge 1236 story Ask "what happened with X" and get a paged, subject-scoped history

2026-07-20 14:44:32 +00:00
parent 30ca3ad463
commit c9804ecfdf
18 changed files with 911 additions and 25 deletions
@@ -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<Utc>,
+    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<HistoryEntry>,
+    pub next_cursor: Option<String>,
+}
+
+/// 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<i64>,
+    until: Option<i64>,
+    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::HistoryEntryRaw> =
+        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<String> {
+    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);
+    }
+}