diff --git a/CHANGELOG.md b/CHANGELOG.md index 5f046c0f..135e3fc9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- `vent` MCP tool + `agentflare vent` CLI: agents log tooling friction to an append-only per-repo JSONL; a deterministic classifier consolidates them once per turn (via the PromptSubmit hook) and auto-files actionable vents as backlog items. No new dependencies; fully auditable (raw `vents.jsonl` + `vent list`). + ### Changed - *(memory)* brain.db now opens through the shared db-kit engine (versioned migrations, WAL, FK enforcement); recall gains optional hybrid semantic search (BM25+vector merge, 30-day temporal decay, MMR) behind `--features semantic`, with `agentflare memory backfill-embeddings` to index existing observations. FTS-only behavior is byte-identical without an embedding model. diff --git a/crates/agentflare-backend/src/db.rs b/crates/agentflare-backend/src/db.rs index e0cba300..cd300d3b 100644 --- a/crates/agentflare-backend/src/db.rs +++ b/crates/agentflare-backend/src/db.rs @@ -14,6 +14,7 @@ const MIGRATION_LIST: &[M<'static>] = &[ M::up(include_str!("migrations/0003_asset_versioning.sql")), M::up(include_str!("migrations/0004_item_comments.sql")), M::up(include_str!("migrations/0005_items_fts.sql")), + M::up(include_str!("migrations/0006_vents.sql")), ]; const MIGRATIONS: Migrations = Migrations::from_slice(MIGRATION_LIST); diff --git a/crates/agentflare-backend/src/lib.rs b/crates/agentflare-backend/src/lib.rs index 7b3dd9b7..3bf64839 100644 --- a/crates/agentflare-backend/src/lib.rs +++ b/crates/agentflare-backend/src/lib.rs @@ -8,6 +8,7 @@ pub mod item; pub mod label; pub mod project; pub mod state; +pub mod vent; pub mod webhook; pub mod workspace; diff --git a/crates/agentflare-backend/src/migrations/0006_vents.sql b/crates/agentflare-backend/src/migrations/0006_vents.sql new file mode 100644 index 00000000..1b6650b3 --- /dev/null +++ b/crates/agentflare-backend/src/migrations/0006_vents.sql @@ -0,0 +1,16 @@ +CREATE TABLE IF NOT EXISTS vents ( + id TEXT PRIMARY KEY, + project_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE, + message TEXT NOT NULL, + severity TEXT NOT NULL DEFAULT 'medium', + tags TEXT NOT NULL DEFAULT '[]', + topic_key TEXT NOT NULL, + seen_count INTEGER NOT NULL DEFAULT 1, + actionable INTEGER NOT NULL DEFAULT 0, + item_id TEXT REFERENCES items(id) ON DELETE SET NULL, + first_event_id TEXT NOT NULL, + created_at INTEGER NOT NULL, + updated_at INTEGER NOT NULL, + UNIQUE(project_id, topic_key) +); +CREATE INDEX IF NOT EXISTS idx_vents_project_actionable ON vents(project_id, actionable); diff --git a/crates/agentflare-backend/src/vent.rs b/crates/agentflare-backend/src/vent.rs new file mode 100644 index 00000000..47a0c73d --- /dev/null +++ b/crates/agentflare-backend/src/vent.rs @@ -0,0 +1,198 @@ +use crate::error::Result; +use rusqlite::{Connection, OptionalExtension, params}; + +#[derive(Debug, Clone, serde::Serialize)] +pub struct Vent { + pub id: String, + pub project_id: String, + pub message: String, + pub severity: String, + pub tags: String, + pub topic_key: String, + pub seen_count: i64, + pub actionable: bool, + pub item_id: Option, + pub first_event_id: String, + pub created_at: i64, + pub updated_at: i64, +} + +pub struct UpsertOutcome { + pub id: String, + pub seen_count: i64, + pub existing_item_id: Option, + pub was_actionable: bool, +} + +#[allow(clippy::too_many_arguments)] +pub fn upsert( + conn: &Connection, + project_id: &str, + message: &str, + severity: &str, + tags_json: &str, + topic_key: &str, + first_event_id: &str, + seen_delta: i64, + now: i64, +) -> Result { + let existing = conn + .query_row( + "SELECT id, seen_count, item_id, actionable FROM vents + WHERE project_id = ?1 AND topic_key = ?2", + params![project_id, topic_key], + |r| { + Ok(( + r.get::<_, String>(0)?, + r.get::<_, i64>(1)?, + r.get::<_, Option>(2)?, + r.get::<_, i64>(3)? != 0, + )) + }, + ) + .optional()?; + + if let Some((id, seen, item_id, actionable)) = existing { + let new_seen = seen + seen_delta; + conn.execute( + "UPDATE vents SET seen_count = ?2, message = ?3, severity = ?4, + tags = ?5, updated_at = ?6 WHERE id = ?1", + params![id, new_seen, message, severity, tags_json, now], + )?; + return Ok(UpsertOutcome { + id, + seen_count: new_seen, + existing_item_id: item_id, + was_actionable: actionable, + }); + } + + let id = db_kit::ids::new_id(); + conn.execute( + "INSERT INTO vents (id, project_id, message, severity, tags, topic_key, + seen_count, actionable, item_id, first_event_id, created_at, updated_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 0, NULL, ?8, ?9, ?9)", + params![ + id, + project_id, + message, + severity, + tags_json, + topic_key, + seen_delta, + first_event_id, + now + ], + )?; + Ok(UpsertOutcome { + id, + seen_count: seen_delta, + existing_item_id: None, + was_actionable: false, + }) +} + +pub fn link_item(conn: &Connection, vent_id: &str, item_id: &str) -> Result<()> { + conn.execute( + "UPDATE vents SET item_id = ?2 WHERE id = ?1", + params![vent_id, item_id], + )?; + Ok(()) +} + +pub fn set_actionable(conn: &Connection, vent_id: &str, actionable: bool) -> Result<()> { + conn.execute( + "UPDATE vents SET actionable = ?2 WHERE id = ?1", + params![vent_id, i64::from(actionable)], + )?; + Ok(()) +} + +pub fn list(conn: &Connection, project_id: &str, actionable_only: bool) -> Result> { + let mut stmt = conn.prepare( + "SELECT id, project_id, message, severity, tags, topic_key, seen_count, + actionable, item_id, first_event_id, created_at, updated_at + FROM vents WHERE project_id = ?1 AND (?2 = 0 OR actionable = 1) + ORDER BY updated_at DESC", + )?; + let rows = stmt.query_map(params![project_id, i64::from(actionable_only)], |r| { + Ok(Vent { + id: r.get(0)?, + project_id: r.get(1)?, + message: r.get(2)?, + severity: r.get(3)?, + tags: r.get(4)?, + topic_key: r.get(5)?, + seen_count: r.get(6)?, + actionable: r.get::<_, i64>(7)? != 0, + item_id: r.get(8)?, + first_event_id: r.get(9)?, + created_at: r.get(10)?, + updated_at: r.get(11)?, + }) + })?; + rows.collect::>>() + .map_err(Into::into) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::db::open_in_memory; + + fn seed_project(conn: &rusqlite::Connection) -> String { + conn.execute("INSERT INTO workspaces (id,name,slug,item_label,created_at,updated_at) VALUES ('w','W','w','Item',1,1)", []).unwrap(); + conn.execute("INSERT INTO projects (id,workspace_id,name,identifier,created_at,updated_at) VALUES ('p','w','P','P',1,1)", []).unwrap(); + "p".to_string() + } + + #[test] + fn upsert_dedups_by_topic_and_accumulates_seen_count() { + let conn = open_in_memory().unwrap(); + let p = seed_project(&conn); + let a = upsert( + &conn, + &p, + "disk full", + "medium", + "[]", + "disk full", + "ev1", + 1, + 100, + ) + .unwrap(); + assert_eq!(a.seen_count, 1); + assert!(a.existing_item_id.is_none()); + let b = upsert( + &conn, + &p, + "disk full", + "high", + "[]", + "disk full", + "ev2", + 5, + 200, + ) + .unwrap(); + assert_eq!(a.id, b.id, "same topic → same row"); + assert_eq!(b.seen_count, 6, "1 + delta 5"); + assert_eq!(list(&conn, &p, false).unwrap().len(), 1); + } + + #[test] + fn link_item_and_actionable_filter() { + let conn = open_in_memory().unwrap(); + let p = seed_project(&conn); + conn.execute("INSERT INTO states (id,project_id,name,group_name,sequence,is_default,created_at,updated_at) VALUES ('s','p','Backlog','backlog',1.0,1,1,1)", []).unwrap(); + conn.execute("INSERT INTO items (id,project_id,state_id,name,created_at,updated_at) VALUES ('it','p','s','x',1,1)", []).unwrap(); + let v = upsert(&conn, &p, "m", "low", "[]", "m", "ev", 1, 1).unwrap(); + set_actionable(&conn, &v.id, true).unwrap(); + link_item(&conn, &v.id, "it").unwrap(); + let only = list(&conn, &p, true).unwrap(); + assert_eq!(only.len(), 1); + assert_eq!(only[0].item_id.as_deref(), Some("it")); + assert!(only[0].actionable); + } +} diff --git a/src/cli/mod.rs b/src/cli/mod.rs index beb11bce..60b8fd5b 100644 --- a/src/cli/mod.rs +++ b/src/cli/mod.rs @@ -20,6 +20,7 @@ mod run; mod serve; mod uninstall; mod update; +mod vent; use clap::{Parser, Subcommand}; use std::sync::LazyLock; @@ -67,6 +68,7 @@ pub enum Commands { Review(review::ReviewArgs), Memory(memory::MemoryArgs), Serve(serve::ServeArgs), + Vent(vent::VentArgs), } impl Commands { @@ -95,6 +97,7 @@ impl Commands { Self::Review(cmd) => cmd.run(), Self::Memory(cmd) => cmd.run(), Self::Serve(cmd) => cmd.run(), + Self::Vent(cmd) => vent::run(cmd), } } } diff --git a/src/cli/vent.rs b/src/cli/vent.rs new file mode 100644 index 00000000..f0dfe4da --- /dev/null +++ b/src/cli/vent.rs @@ -0,0 +1,117 @@ +use clap::{Args, Subcommand}; + +#[derive(Args)] +pub struct VentArgs { + #[command(subcommand)] + pub cmd: VentCmd, +} + +#[derive(Subcommand)] +pub enum VentCmd { + /// Log a friction vent (append-only; triaged at the next turn). + Say { + message: String, + #[arg(long, default_value = "medium")] + severity: String, + #[arg(long = "tag")] + tags: Vec, + }, + /// Triage buffered vents now (also run automatically once per turn). + Consolidate, + /// List triaged vents for this repo's project. + List { + #[arg(long)] + actionable: bool, + }, +} + +pub fn run(args: VentArgs) { + match args.cmd { + VentCmd::Say { + message, + severity, + tags, + } => { + let severity = crate::vent::classify::normalize_severity(Some(&severity)); + let log = crate::vent::paths::log_path(); + match crate::vent::capture::append(&log, None, severity, &tags, message.trim()) { + Ok(id) => println!("vented {id}"), + Err(e) => eprintln!("vent failed: {e}"), + } + } + VentCmd::Consolidate => { + let r = crate::vent::consolidate::consolidate(); + println!( + "consolidated {} vent(s) → {} item(s){}", + r.consolidated, + r.items_created.len(), + if r.buffered_no_project > 0 { + format!( + " ({} buffered — no linked project yet)", + r.buffered_no_project + ) + } else { + String::new() + } + ); + for id in r.items_created { + println!(" filed item {id}"); + } + } + VentCmd::List { actionable } => { + let conn = match agentflare_backend::db::open_db(&crate::vent::paths::backend_db_path()) + { + Ok(c) => c, + Err(e) => { + eprintln!("cannot open backend: {e}"); + return; + } + }; + let link = crate::vent::paths::repo_root() + .join(".agentflare") + .join("project.json"); + let project_id = std::fs::read(&link) + .ok() + .and_then(|b| serde_json::from_slice::(&b).ok()) + .and_then(|v| { + v.get("project_id") + .and_then(|p| p.as_str().map(String::from)) + }); + let Some(pid) = project_id else { + println!("no linked project — run an agentflare item/memory command here first"); + return; + }; + match agentflare_backend::vent::list(&conn, &pid, actionable) { + Ok(vents) => { + for v in vents { + println!( + "{} seen×{} {} {}{}", + if v.actionable { "●" } else { "○" }, + v.seen_count, + v.severity, + v.message.split('\n').next().unwrap_or(""), + v.item_id + .map(|i| format!(" → item {i}")) + .unwrap_or_default(), + ); + } + } + Err(e) => eprintln!("list failed: {e}"), + } + } + } +} + +#[cfg(test)] +mod tests { + #[test] + fn append_then_read_back_roundtrips() { + let dir = tempfile::tempdir().unwrap(); + let log = dir.path().join("v.jsonl"); + crate::vent::capture::append(&log, None, "high", &[], "roundtrip check").unwrap(); + let (lines, _) = + crate::vent::consolidate::read_new_lines(&log, &dir.path().join("v.cursor")); + assert_eq!(lines.len(), 1); + assert_eq!(lines[0].message, "roundtrip check"); + } +} diff --git a/src/hook.rs b/src/hook.rs index 89c3315d..e89b8ea0 100644 --- a/src/hook.rs +++ b/src/hook.rs @@ -33,6 +33,19 @@ fn read_stdin_or_skip(label: &str) -> Option { pub fn session_start(agent: &str) { let msg = session_start_message(agent); + + // Flush any vents buffered since the last turn/session (best-effort; + // never blocks the hook or surfaces errors to the agent). + let _ = std::panic::catch_unwind(|| { + let r = crate::vent::consolidate::consolidate(); + if !r.items_created.is_empty() { + eprintln!( + "[agentflare] vent: filed {} item(s) from friction", + r.items_created.len() + ); + } + }); + // Plain stdout reaches Claude's context for this event (see module // comment) but is NOT shown to the user in the terminal. `systemMessage` // is the only field that renders visibly, so emit both: the user sees @@ -342,6 +355,18 @@ pub fn prompt_submit(agent: &str) { return; } + // Triage the previous turn's buffered vents once per turn (best-effort; + // never blocks the hook or surfaces errors to the agent). + let _ = std::panic::catch_unwind(|| { + let r = crate::vent::consolidate::consolidate(); + if !r.items_created.is_empty() { + eprintln!( + "[agentflare] vent: filed {} item(s) from friction", + r.items_created.len() + ); + } + }); + let mut bits = vec![ "AGENTFLARE ACTIVE.".to_string(), "Prefer lean-ctx ctx_* tools over native Read/Grep/Bash/Glob.".to_string(), diff --git a/src/main.rs b/src/main.rs index 15b8b176..486dae9e 100644 --- a/src/main.rs +++ b/src/main.rs @@ -49,6 +49,7 @@ mod tool_install; mod ui; mod uninstall; mod update; +mod vent; mod worktree; use clap::Parser; diff --git a/src/mcp_server.rs b/src/mcp_server.rs index cf884429..335ccb55 100644 --- a/src/mcp_server.rs +++ b/src/mcp_server.rs @@ -473,7 +473,7 @@ impl AgentflareMcp { /// across multiple linked projects depending on which subdirectory a /// tool was invoked from. Falls back to raw cwd only when nothing is /// found anywhere above it. - fn repo_root() -> std::path::PathBuf { + pub(crate) fn repo_root() -> std::path::PathBuf { let cwd = std::env::current_dir().unwrap_or_default(); if let Some(root) = crate::git::repo_toplevel(&cwd) { return root; @@ -1005,6 +1005,23 @@ impl AgentflareMcp { self.memory_impl(req) } + #[tool( + description = "Vent friction when the TOOLING blocks you (not the task) — a wrong/missing tool, a fabricated assumption, an environment gap. Actionable vents auto-file a DX item once per turn; noise is just logged. Use sparingly, exactly when you're genuinely blocked. DO: \"The $CLAUDE_JOB_DIR I assumed exists is empty — I fabricated it; there's no such env var and my temp writes went to /.\" DON'T: \"This build is slow to compile.\" Inputs: message (required), severity (low|medium|high), tags." + )] + fn vent(&self, Parameters(req): Parameters) -> Result { + if req.message.trim().is_empty() { + return Err(ErrorData::invalid_params("message is required", None)); + } + let severity = crate::vent::classify::normalize_severity(req.severity.as_deref()); + let tags = req.tags.unwrap_or_default(); + let log = crate::vent::paths::log_path(); + let event_id = + crate::vent::capture::append(&log, None, severity, &tags, req.message.trim()).map_err( + |e| ErrorData::internal_error(format!("vent capture failed: {e}"), None), + )?; + Ok(serde_json::json!({ "ok": true, "event_id": event_id }).to_string()) + } + /// Rejects PR titles that don't start with a conventional-commit type, /// mirroring `.github/workflows/pr-title.yml`'s /// `amannn/action-semantic-pull-request` config so the check fires here diff --git a/src/mcp_server/types.rs b/src/mcp_server/types.rs index 0a080899..38ca0a19 100644 --- a/src/mcp_server/types.rs +++ b/src/mcp_server/types.rs @@ -263,6 +263,13 @@ pub(crate) struct OptimizeRequest { pub(crate) id: Option, } +#[derive(Debug, Deserialize, schemars::JsonSchema)] +pub(crate) struct VentRequest { + pub(crate) message: String, + pub(crate) severity: Option, + pub(crate) tags: Option>, +} + #[derive(Debug, Default, Deserialize, schemars::JsonSchema)] pub(crate) struct MemoryRequest { #[schemars(description = "Action: compact|context|curate|handoff|recall|relate|remember")] diff --git a/src/vent/capture.rs b/src/vent/capture.rs new file mode 100644 index 00000000..08ba703f --- /dev/null +++ b/src/vent/capture.rs @@ -0,0 +1,62 @@ +use std::io::Write; +use std::path::Path; + +#[derive(serde::Serialize, serde::Deserialize)] +pub struct VentLine { + pub event_id: String, + pub ts: String, + #[serde(default)] + pub session: Option, + pub severity: String, + #[serde(default)] + pub tags: Vec, + pub message: String, +} + +pub fn append( + log_path: &Path, + session: Option<&str>, + severity: &str, + tags: &[String], + message: &str, +) -> std::io::Result { + if let Some(parent) = log_path.parent() { + std::fs::create_dir_all(parent)?; + } + let id = crate::vent::event_id(message); + let line = VentLine { + event_id: id.clone(), + ts: chrono::Utc::now().to_rfc3339(), + session: session.map(str::to_string), + severity: severity.to_string(), + tags: tags.to_vec(), + message: message.to_string(), + }; + let mut f = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(log_path)?; + writeln!(f, "{}", serde_json::to_string(&line)?)?; + Ok(id) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn append_writes_one_parseable_line_per_call() { + let dir = tempfile::tempdir().unwrap(); + let log = dir.path().join("v.jsonl"); + let id1 = append(&log, Some("s1"), "high", &["dx".into()], "boom").unwrap(); + let _id2 = append(&log, None, "medium", &[], "again").unwrap(); + let text = std::fs::read_to_string(&log).unwrap(); + let lines: Vec<&str> = text.lines().collect(); + assert_eq!(lines.len(), 2); + let first: VentLine = serde_json::from_str(lines[0]).unwrap(); + assert_eq!(first.event_id, id1); + assert_eq!(first.severity, "high"); + assert_eq!(first.message, "boom"); + assert_eq!(first.tags, vec!["dx".to_string()]); + } +} diff --git a/src/vent/classify.rs b/src/vent/classify.rs new file mode 100644 index 00000000..3a1fe5b0 --- /dev/null +++ b/src/vent/classify.rs @@ -0,0 +1,86 @@ +use regex::Regex; +use std::sync::LazyLock; + +pub const ACTIONABLE_SEEN_THRESHOLD: i64 = 3; + +static MARKER_RE: LazyLock = LazyLock::new(|| { + Regex::new( + r"(?i)\b(broke|broken|fails?|failing|wrong|should|missing|can'?t|cannot|error|panic|crash|hang|stuck|nonexistent|fabricat\w*)\b|[\w./-]+\.\w{1,6}", + ) + .expect("static marker regex is valid") +}); + +pub fn topic_key(message: &str) -> String { + message + .to_lowercase() + .split_whitespace() + .map(|w| { + w.chars() + .filter(|c| c.is_alphanumeric()) + .collect::() + }) + .filter(|w| !w.is_empty()) + .collect::>() + .join(" ") +} + +pub fn severity_rank(severity: &str) -> u8 { + match severity { + "high" => 2, + "medium" => 1, + _ => 0, + } +} + +/// Normalize user-supplied severity to exactly "low" | "medium" | "high" +/// (case-insensitive), defaulting to "medium" for anything else. Shared by +/// the MCP `vent` tool and the `agentflare vent say` CLI so both entry +/// points classify identically. +pub fn normalize_severity(input: Option<&str>) -> &'static str { + match input.map(str::to_lowercase).as_deref() { + Some("low") => "low", + Some("high") => "high", + _ => "medium", + } +} + +pub fn classify(severity: &str, seen_count: i64, message: &str) -> bool { + severity == "high" || seen_count >= ACTIONABLE_SEEN_THRESHOLD || MARKER_RE.is_match(message) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn topic_key_is_stable_across_case_ws_punct() { + assert_eq!(topic_key("Disk FULL!!"), topic_key("disk full")); + assert_eq!(topic_key(" a,b. c "), "ab c"); + } + + #[test] + fn classify_truth_table() { + assert!(classify("high", 1, "all good")); + assert!(!classify("low", 2, "all good")); + assert!(classify("low", 3, "all good")); + assert!(classify("low", 1, "the build fails on windows")); + assert!(classify("low", 1, "I fabricated $CLAUDE_JOB_DIR")); + assert!(classify("low", 1, "cannot open config.toml")); + assert!(!classify("low", 1, "this is a normal note")); + } + + #[test] + fn severity_rank_orders_low_medium_high() { + assert!(severity_rank("high") > severity_rank("medium")); + assert!(severity_rank("medium") > severity_rank("low")); + assert_eq!(severity_rank("garbage"), severity_rank("low")); + } + + #[test] + fn normalize_severity_is_case_insensitive_and_defaults_to_medium() { + assert_eq!(normalize_severity(Some("High")), "high"); + assert_eq!(normalize_severity(Some("LOW")), "low"); + assert_eq!(normalize_severity(Some("garbage")), "medium"); + assert_eq!(normalize_severity(None), "medium"); + } +} diff --git a/src/vent/consolidate.rs b/src/vent/consolidate.rs new file mode 100644 index 00000000..88ac5bae --- /dev/null +++ b/src/vent/consolidate.rs @@ -0,0 +1,303 @@ +use crate::vent::capture::VentLine; +use crate::vent::classify::{classify, severity_rank, topic_key}; +use crate::vent::paths; +use std::collections::BTreeMap; +use std::io::{BufRead, BufReader, Seek, SeekFrom}; +use std::path::Path; + +#[derive(Debug, Default)] +pub struct ConsolidateReport { + pub consolidated: usize, + pub items_created: Vec, + pub noise: usize, + pub buffered_no_project: usize, +} + +pub fn read_new_lines(log_path: &Path, cursor_path: &Path) -> (Vec, u64) { + let mut offset = std::fs::read_to_string(cursor_path) + .ok() + .and_then(|s| s.trim().parse::().ok()) + .unwrap_or(0); + let Ok(mut file) = std::fs::File::open(log_path) else { + return (Vec::new(), offset); + }; + let meta_len = file.metadata().map_or(0, |m| m.len()); + if meta_len < offset { + offset = 0; + } + if meta_len == offset { + return (Vec::new(), offset); + } + let _ = file.seek(SeekFrom::Start(offset)); + let reader = BufReader::new(&file); + let mut lines = Vec::new(); + let mut bytes_read: u64 = 0; + for line in reader.lines() { + let Ok(line) = line else { break }; + bytes_read += line.len() as u64 + 1; + if line.trim().is_empty() { + continue; + } + if let Ok(v) = serde_json::from_str::(&line) { + lines.push(v); + } + } + (lines, offset + bytes_read) +} + +fn now() -> i64 { + chrono::Utc::now().timestamp() +} + +fn truncate(s: &str, max: usize) -> String { + let one_line = s.split('\n').next().unwrap_or(s).trim(); + if one_line.chars().count() <= max { + one_line.to_string() + } else { + format!("{}…", one_line.chars().take(max).collect::()) + } +} + +pub fn consolidate_core( + conn: &rusqlite::Connection, + project_id: &str, + default_state_id: &str, + log_path: &Path, + cursor_path: &Path, +) -> std::io::Result { + let (lines, new_offset) = read_new_lines(log_path, cursor_path); + let mut report = ConsolidateReport::default(); + if lines.is_empty() { + return Ok(report); + } + + struct Group { + message: String, + severity: String, + count: i64, + first_event: String, + } + let mut groups: BTreeMap = BTreeMap::new(); + for l in &lines { + let key = topic_key(&l.message); + let g = groups.entry(key).or_insert_with(|| Group { + message: l.message.clone(), + severity: l.severity.clone(), + count: 0, + first_event: l.event_id.clone(), + }); + g.count += 1; + g.message = l.message.clone(); + if severity_rank(&l.severity) > severity_rank(&g.severity) { + g.severity = l.severity.clone(); + } + } + + for (key, g) in groups { + report.consolidated += g.count as usize; + let tags_json = "[]"; + let out = match agentflare_backend::vent::upsert( + conn, + project_id, + &g.message, + &g.severity, + tags_json, + &key, + &g.first_event, + g.count, + now(), + ) { + Ok(o) => o, + Err(e) => { + eprintln!("[vent] upsert failed: {e}"); + continue; + } + }; + let actionable = classify(&g.severity, out.seen_count, &g.message); + let _ = agentflare_backend::vent::set_actionable(conn, &out.id, actionable); + if !actionable { + report.noise += g.count as usize; + continue; + } + if out.existing_item_id.is_some() { + continue; + } + let metadata = serde_json::json!({ + "source": "vent", + "topic_key": key, + "severity": g.severity, + "seen_count": out.seen_count, + "first_event_id": g.first_event, + }) + .to_string(); + let input = agentflare_backend::item::CreateItem { + project_id: project_id.to_string(), + state_id: default_state_id.to_string(), + name: format!("[vent] {}", truncate(&g.message, 72)), + description: Some(format!( + "{}\n\n---\nsource: vent · severity: {} · seen: {}", + g.message, g.severity, out.seen_count + )), + priority: None, + parent_id: None, + assignee_agent: None, + sort_order: None, + external_source: None, + external_id: None, + metadata: Some(metadata), + label_ids: vec![], + assignee_ids: vec![], + dependency_ids: vec![], + }; + match agentflare_backend::item::create(conn, input) { + Ok(item) => { + let _ = agentflare_backend::vent::link_item(conn, &out.id, &item.id); + report.items_created.push(item.id); + } + Err(e) => eprintln!("[vent] item::create failed: {e}"), + } + } + + let _ = std::fs::write(cursor_path, new_offset.to_string()); + Ok(report) +} + +pub fn consolidate() -> ConsolidateReport { + let empty = ConsolidateReport::default(); + let conn = match agentflare_backend::db::open_db(&paths::backend_db_path()) { + Ok(c) => c, + Err(_) => return empty, + }; + let link_file = paths::repo_root().join(".agentflare").join("project.json"); + let Ok(bytes) = std::fs::read(&link_file) else { + let (lines, _) = read_new_lines(&paths::log_path(), &paths::cursor_path()); + return ConsolidateReport { + buffered_no_project: lines.len(), + ..empty + }; + }; + let Ok(link) = serde_json::from_slice::(&bytes) else { + return empty; + }; + let Some(project_id) = link.get("project_id").and_then(|v| v.as_str()) else { + return empty; + }; + let Ok(project) = agentflare_backend::project::get(&conn, project_id) else { + return empty; + }; + let default_state = agentflare_backend::state::list_by_project(&conn, &project.id) + .ok() + .and_then(|states| states.into_iter().find(|s| s.is_default)); + let Some(state) = default_state else { + return empty; + }; + consolidate_core( + &conn, + &project.id, + &state.id, + &paths::log_path(), + &paths::cursor_path(), + ) + .unwrap_or_default() +} + +#[cfg(test)] +mod tests { + use super::*; + use agentflare_backend::db::open_in_memory; + + fn seed(conn: &rusqlite::Connection) -> (String, String) { + conn.execute( + "INSERT INTO workspaces (id,name,slug,item_label,created_at,updated_at) VALUES ('w','W','w','Item',1,1)", + [], + ) + .unwrap(); + conn.execute( + "INSERT INTO projects (id,workspace_id,name,identifier,created_at,updated_at) VALUES ('p','w','P','P',1,1)", + [], + ) + .unwrap(); + conn.execute( + "INSERT INTO states (id,project_id,name,group_name,sequence,is_default,created_at,updated_at) VALUES ('s','p','Backlog','backlog',1.0,1,1,1)", + [], + ) + .unwrap(); + ("p".into(), "s".into()) + } + + fn write_lines(log: &std::path::Path, msgs: &[(&str, &str)]) { + use std::io::Write; + let mut f = std::fs::OpenOptions::new() + .create(true) + .append(true) + .open(log) + .unwrap(); + for (sev, m) in msgs { + let line = crate::vent::capture::VentLine { + event_id: crate::vent::event_id(m), + ts: "t".into(), + session: None, + severity: (*sev).to_string(), + tags: vec![], + message: (*m).to_string(), + }; + writeln!(f, "{}", serde_json::to_string(&line).unwrap()).unwrap(); + } + } + + #[test] + fn actionable_creates_one_item_noise_creates_none() { + let conn = open_in_memory().unwrap(); + let (p, s) = seed(&conn); + let dir = tempfile::tempdir().unwrap(); + let (log, cur) = (dir.path().join("v.jsonl"), dir.path().join("v.cursor")); + write_lines( + &log, + &[ + ("high", "server crash on boot"), + ("low", "just a calm note"), + ], + ); + let rep = consolidate_core(&conn, &p, &s, &log, &cur).unwrap(); + assert_eq!(rep.items_created.len(), 1, "only the actionable one"); + assert_eq!(rep.noise, 1); + let items: i64 = conn + .query_row("SELECT count(*) FROM items", [], |r| r.get(0)) + .unwrap(); + assert_eq!(items, 1); + } + + #[test] + fn consolidate_is_idempotent_via_cursor() { + let conn = open_in_memory().unwrap(); + let (p, s) = seed(&conn); + let dir = tempfile::tempdir().unwrap(); + let (log, cur) = (dir.path().join("v.jsonl"), dir.path().join("v.cursor")); + write_lines(&log, &[("high", "boom happened")]); + let r1 = consolidate_core(&conn, &p, &s, &log, &cur).unwrap(); + let r2 = consolidate_core(&conn, &p, &s, &log, &cur).unwrap(); + assert_eq!(r1.items_created.len(), 1); + assert_eq!(r2.consolidated, 0, "cursor consumed the line"); + let items: i64 = conn + .query_row("SELECT count(*) FROM items", [], |r| r.get(0)) + .unwrap(); + assert_eq!(items, 1, "no duplicate item on rerun"); + } + + #[test] + fn within_turn_spiral_collapses_to_one_item() { + let conn = open_in_memory().unwrap(); + let (p, s) = seed(&conn); + let dir = tempfile::tempdir().unwrap(); + let (log, cur) = (dir.path().join("v.jsonl"), dir.path().join("v.cursor")); + let spiral: Vec<(&str, &str)> = + std::iter::repeat_n(("high", "the same broken thing"), 43).collect(); + write_lines(&log, &spiral); + let rep = consolidate_core(&conn, &p, &s, &log, &cur).unwrap(); + assert_eq!(rep.items_created.len(), 1, "43 identical vents → one item"); + let seen: i64 = conn + .query_row("SELECT seen_count FROM vents", [], |r| r.get(0)) + .unwrap(); + assert_eq!(seen, 43); + } +} diff --git a/src/vent/mod.rs b/src/vent/mod.rs new file mode 100644 index 00000000..eb8abefc --- /dev/null +++ b/src/vent/mod.rs @@ -0,0 +1,16 @@ +pub mod capture; +pub mod classify; +pub mod consolidate; +pub mod paths; + +pub fn event_id(message: &str) -> String { + use std::hash::{Hash, Hasher}; + let mut h = std::collections::hash_map::DefaultHasher::new(); + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + nanos.hash(&mut h); + message.hash(&mut h); + format!("{:08x}", h.finish() as u32) +} diff --git a/src/vent/paths.rs b/src/vent/paths.rs new file mode 100644 index 00000000..3625100b --- /dev/null +++ b/src/vent/paths.rs @@ -0,0 +1,78 @@ +use std::path::PathBuf; + +pub fn repo_key() -> String { + if let Ok(out) = std::process::Command::new("git") + .args(["remote", "get-url", "origin"]) + .output() + && out.status.success() + { + let remote = String::from_utf8_lossy(&out.stdout); + let remote = remote.trim(); + if !remote.is_empty() { + return format!("git:{}", crate::claims::normalize_repo(remote)); + } + } + let root = repo_root(); + let canonical = std::fs::canonicalize(&root).unwrap_or(root); + format!("path:{}", canonical.to_string_lossy()) +} + +/// Same resolution `AgentflareMcp` uses for `.agentflare/project.json` — +/// git toplevel, else a marker-based walk-up for non-git projects. Vent must +/// resolve to the identical root or `consolidate()` looks for the project +/// link in the wrong directory. +pub fn repo_root() -> PathBuf { + crate::mcp_server::AgentflareMcp::repo_root() +} + +fn repo_slug() -> String { + repo_key() + .chars() + .map(|c| if c.is_ascii_alphanumeric() { c } else { '_' }) + .collect() +} + +pub fn log_path() -> PathBuf { + crate::state::state_dir() + .join("vents") + .join(format!("{}.jsonl", repo_slug())) +} + +pub fn cursor_path() -> PathBuf { + crate::state::state_dir() + .join("vents") + .join(format!("{}.cursor", repo_slug())) +} + +pub fn backend_db_path() -> PathBuf { + crate::paths::home().join(".agentflare").join("backend.db") +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn log_and_cursor_are_siblings_under_state_dir() { + // repo_key() reads cwd (via repo_root()) and AGENTFLARE_HOME_OVERRIDE, + // both process-global and mutated by with_temp_cwd/with_temp_home + // elsewhere in this binary -- without this lock, a concurrent test can + // change either between the two repo_key() calls below (log_path() + // then cursor_path()), producing mismatched slugs. Flaked on Linux CI + // under higher default test parallelism; not reliably reproduced locally. + let _guard = agent_registry::detect::PATH_LOCK + .lock() + .unwrap_or_else(|e| e.into_inner()); + let log = log_path(); + let cur = cursor_path(); + assert!(log.to_string_lossy().contains("vents")); + assert_eq!(cur.extension().unwrap(), "cursor"); + assert_eq!(log.parent(), cur.parent()); + assert_eq!(log.file_stem(), cur.file_stem()); + } + + #[test] + fn repo_key_is_stable() { + assert_eq!(repo_key(), repo_key()); + } +}