feat(vent): friction capture + per-turn consolidation → auto-filed items - #259
Merged
Conversation
PromptSubmit hook calls vent::consolidate best-effort; emits stderr note when items are filed. Pre-existing clippy -D warnings in dependent crates unrelated.
…se; normalize CLI severity; fmt
…tore agentflare serve dashboard command dropped by a bad merge
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughChangesVent feature
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant VentCapture
participant VentConsolidate
participant BackendDB
participant ItemStore
Client->>VentCapture: submit vent message
VentCapture->>VentCapture: append JSONL event
VentConsolidate->>VentCapture: read unread events
VentConsolidate->>BackendDB: upsert aggregated vent
VentConsolidate->>ItemStore: create and link actionable item
Possibly related PRs
Suggested labels: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (2)
src/hook.rs (1)
36-48: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicate panic-safe consolidate block.
The same 8-line
catch_unwind-wrappedconsolidate()+eprintln!block is duplicated verbatim insession_startandprompt_submit. Extract into a shared helper to avoid drift.♻️ Proposed fix
+fn flush_vents() { + 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() + ); + } + }); +} + 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() - ); - } - }); + flush_vents();(and similarly at the
prompt_submitcall site)Also applies to: 358-369
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/hook.rs` around lines 36 - 48, Extract the duplicated panic-safe consolidate-and-report logic from session_start and prompt_submit into a shared helper in src/hook.rs. Have both call sites invoke that helper, preserving best-effort panic handling, consolidation, and the existing eprintln output for created items.src/vent/consolidate.rs (1)
171-184: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winDuplicated project-link-file parsing in
consolidate()andvent list. Both sites read.agentflare/project.jsonand JSON-parse it to extractproject_id, with no shared helper.
src/vent/consolidate.rs#L171-L184: extract this parsing into a shared helper (e.g.paths::resolve_linked_project_id() -> Option<String>) and call it here.src/cli/vent.rs#L70-L83: replace the inlinestd::fs::read/serde_json::from_sliceparsing in theListarm with the same shared helper.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/vent/consolidate.rs` around lines 171 - 184, Extract the shared `.agentflare/project.json` parsing into a `paths::resolve_linked_project_id() -> Option<String>` helper. In src/vent/consolidate.rs lines 171-184, replace the inline read and JSON extraction with this helper while preserving the existing missing-file buffering behavior; in src/cli/vent.rs lines 70-83, replace the `List` arm’s duplicate parsing with the same helper.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@CHANGELOG.md`:
- Line 12: Update the changelog entry’s raw-log path description to state that
logs are stored as per-repository files under state_dir()/vents/, such as
<repo_slug>.jsonl, rather than a single vents.jsonl file.
In `@crates/agentflare-backend/src/vent.rs`:
- Around line 39-86: Make the vent upsert atomic by replacing the existing
SELECT/update/insert flow with a single INSERT ... ON CONFLICT(project_id,
topic_key) DO UPDATE statement. Increment seen_count using the database value
plus seen_delta, while preserving the existing field updates and returning the
same UpsertOutcome data, including prior item_id and actionable state. Use the
upsert result to distinguish newly inserted versus existing vents without
allowing concurrent writes to lose increments or violate the unique constraint.
In `@src/vent/capture.rs`:
- Around line 16-40: Update capture::append to make persistence best-effort at
the interface boundary: handle filesystem and serialization failures internally,
emit diagnostic information through stderr logging or telemetry, and avoid
propagating failures to MCP callers. Preserve an optional mechanism for local
callers to observe whether persistence succeeded, while keeping successful
appends returning the generated event ID.
- Around line 35-39: Update the append logic around the OpenOptions handle and
JSON serialization to acquire a cross-process file lock before writing each
complete JSONL record, holding it through the newline flush and releasing it
afterward. Ensure concurrent CLI and MCP writers cannot interleave records, and
add a parallel-writer test covering valid, lossless reads through
read_new_lines.
In `@src/vent/consolidate.rs`:
- Around line 74-119: Preserve VentLine.tags during consolidation by adding the
tags field to Group, initializing it from the first line, and merging tags from
subsequent lines without duplicates. Replace the hardcoded tags_json in the
consolidation upsert flow with serialized Group tags, and include the same tags
in the created item's metadata so both vents.tags and metadata retain
CLI/MCP-provided tags.
- Around line 16-46: Add an exclusive cross-process advisory lock around the
complete consolidation transaction in consolidate_core, covering log reading,
vent-row upsert/item creation, seen_count updates, and cursor persistence.
Ensure competing runs cannot enter this read-to-write sequence simultaneously,
and release the lock on every return path.
In `@src/vent/paths.rs`:
- Around line 28-45: Update repo_slug() to generate a collision-resistant
filename key from repo_key(), using unambiguous encoding for non-alphanumeric
content or a stable hash; preserve its String return type so log_path() and
cursor_path() continue addressing repository-specific files.
---
Nitpick comments:
In `@src/hook.rs`:
- Around line 36-48: Extract the duplicated panic-safe consolidate-and-report
logic from session_start and prompt_submit into a shared helper in src/hook.rs.
Have both call sites invoke that helper, preserving best-effort panic handling,
consolidation, and the existing eprintln output for created items.
In `@src/vent/consolidate.rs`:
- Around line 171-184: Extract the shared `.agentflare/project.json` parsing
into a `paths::resolve_linked_project_id() -> Option<String>` helper. In
src/vent/consolidate.rs lines 171-184, replace the inline read and JSON
extraction with this helper while preserving the existing missing-file buffering
behavior; in src/cli/vent.rs lines 70-83, replace the `List` arm’s duplicate
parsing with the same helper.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 029c97d9-876e-4d1e-b614-e79a149c8755
📒 Files selected for processing (16)
CHANGELOG.mdcrates/agentflare-backend/src/db.rscrates/agentflare-backend/src/lib.rscrates/agentflare-backend/src/migrations/0006_vents.sqlcrates/agentflare-backend/src/vent.rssrc/cli/mod.rssrc/cli/vent.rssrc/hook.rssrc/main.rssrc/mcp_server.rssrc/mcp_server/types.rssrc/vent/capture.rssrc/vent/classify.rssrc/vent/consolidate.rssrc/vent/mod.rssrc/vent/paths.rs
|
|
||
| ### 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`). |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Correct the raw-log location.
The implementation stores logs as per-repository files under state_dir()/vents/ (for example, <repo_slug>.jsonl), not a single vents.jsonl. Update this user-facing path description.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@CHANGELOG.md` at line 12, Update the changelog entry’s raw-log path
description to state that logs are stored as per-repository files under
state_dir()/vents/, such as <repo_slug>.jsonl, rather than a single vents.jsonl
file.
Comment on lines
+39
to
+86
| 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<String>>(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 | ||
| ], | ||
| )?; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== vent.rs ==\n'
wc -l crates/agentflare-backend/src/vent.rs
sed -n '1,220p' crates/agentflare-backend/src/vent.rs | cat -n
printf '\n== search vents schema / upsert callers ==\n'
rg -n --hidden --glob '!target' --glob '!node_modules' \
'CREATE TABLE vents|UNIQUE\\s*\\(project_id,\\s*topic_key\\)|upsert|seen_count|topic_key' \
crates/agentflare-backend -SRepository: getappz/agentflare
Length of output: 10102
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== callers of vent::upsert ==\n'
rg -n --hidden --glob '!target' --glob '!node_modules' 'vent::upsert|upsert\(' crates/agentflare-backend/src -S
printf '\n== migration 0006_vents.sql ==\n'
cat -n crates/agentflare-backend/src/migrations/0006_vents.sqlRepository: getappz/agentflare
Length of output: 1322
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== transaction / lock / retry usage ==\n'
rg -n --hidden --glob '!target' --glob '!node_modules' \
'BEGIN IMMEDIATE|BEGIN EXCLUSIVE|transaction\(|immediate_transaction|busy_timeout|retry|ConstraintViolation|UNIQUE\(project_id, topic_key\)' \
crates/agentflare-backend/src -S
printf '\n== db connection setup ==\n'
rg -n --hidden --glob '!target' --glob '!node_modules' \
'open_in_memory|open\(|Connection::open|pragma|journal_mode|foreign_keys|busy_timeout' \
crates/agentflare-backend/src -SRepository: getappz/agentflare
Length of output: 9315
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== db.rs ==\n'
wc -l crates/agentflare-backend/src/db.rs
sed -n '1,240p' crates/agentflare-backend/src/db.rs | cat -n
printf '\n== connection sharing / mutex / pool ==\n'
rg -n --hidden --glob '!target' --glob '!node_modules' \
'Mutex<Connection>|Arc<Mutex<Connection>>|Pool<|r2d2|deadpool|Connection>|Db|AppState|State|open_db\(' \
crates/agentflare-backend/src -SRepository: getappz/agentflare
Length of output: 10691
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== vent module exports / runtime references ==\n'
rg -n --hidden --glob '!target' --glob '!node_modules' \
'\bmod vent\b|\bpub use vent\b|\bvent::upsert\b|\bupsert\(' \
crates/agentflare-backend/src crates -S
printf '\n== where open_db is used outside tests ==\n'
rg -n --hidden --glob '!target' --glob '!node_modules' \
'\bopen_db\(' \
. -SRepository: getappz/agentflare
Length of output: 3298
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== src/vent/consolidate.rs ==\n'
wc -l src/vent/consolidate.rs
sed -n '1,260p' src/vent/consolidate.rs | cat -n
printf '\n== src/cli/vent.rs ==\n'
wc -l src/cli/vent.rs
sed -n '1,180p' src/cli/vent.rs | cat -n
printf '\n== src/hook.rs vent references ==\n'
rg -n --hidden --glob '!target' --glob '!node_modules' 'vent::|consolidat|upsert\(' src/hook.rs src/mcp_server.rs -SRepository: getappz/agentflare
Length of output: 18040
Make the vent upsert atomic
SELECT + separate UPDATE/INSERT can lose seen_count increments under concurrent consolidations, and first-write races can trip the UNIQUE(project_id, topic_key) constraint. Use a single INSERT ... ON CONFLICT(project_id, topic_key) DO UPDATE that increments seen_count in SQL, or wrap the sequence in an immediate transaction.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/agentflare-backend/src/vent.rs` around lines 39 - 86, Make the vent
upsert atomic by replacing the existing SELECT/update/insert flow with a single
INSERT ... ON CONFLICT(project_id, topic_key) DO UPDATE statement. Increment
seen_count using the database value plus seen_delta, while preserving the
existing field updates and returning the same UpsertOutcome data, including
prior item_id and actionable state. Use the upsert result to distinguish newly
inserted versus existing vents without allowing concurrent writes to lose
increments or violate the unique constraint.
Comment on lines
+16
to
+40
| pub fn append( | ||
| log_path: &Path, | ||
| session: Option<&str>, | ||
| severity: &str, | ||
| tags: &[String], | ||
| message: &str, | ||
| ) -> std::io::Result<String> { | ||
| 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) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Keep capture failures non-fatal at the interface boundary.
Every filesystem failure is returned here, and the MCP caller surfaces it to the agent. That conflicts with the stated append-only capture requirement that callers are not failed. Make this a best-effort operation at the capture boundary (with stderr logging/telemetry), while preserving a way for local callers to observe failed persistence if needed.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/vent/capture.rs` around lines 16 - 40, Update capture::append to make
persistence best-effort at the interface boundary: handle filesystem and
serialization failures internally, emit diagnostic information through stderr
logging or telemetry, and avoid propagating failures to MCP callers. Preserve an
optional mechanism for local callers to observe whether persistence succeeded,
while keeping successful appends returning the generated event ID.
Comment on lines
+35
to
+39
| let mut f = std::fs::OpenOptions::new() | ||
| .create(true) | ||
| .append(true) | ||
| .open(log_path)?; | ||
| writeln!(f, "{}", serde_json::to_string(&line)?)?; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the target file and find the consolidation logic that consumes the JSONL file.
printf '\n== src/vent/capture.rs ==\n'
sed -n '1,220p' src/vent/capture.rs | cat -n
printf '\n== search for JSONL parsing / consolidation ==\n'
rg -n "serde_json::from_str|from_reader|lines\\(|BufRead|capture|vent" src/vent -SRepository: getappz/agentflare
Length of output: 5739
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Broaden the search to find any locking or append helpers used around the same file.
rg -n "OpenOptions::new\\(|append\\(true\\)|lock|flock|fs2|FileExt|fcntl|O_APPEND|jsonl" src -SRepository: getappz/agentflare
Length of output: 23301
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== src/vent/consolidate.rs (reader + tests) ==\n'
sed -n '1,340p' src/vent/consolidate.rs | cat -nRepository: getappz/agentflare
Length of output: 12487
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '\n== vent appends / callers ==\n'
rg -n "vent::capture::append|capture::append\\(|crate::vent::capture::append|append\\(&log" src -SRepository: getappz/agentflare
Length of output: 735
Serialize the JSONL append. src/vent/capture.rs:35-39 is written by both the CLI and MCP server; if two processes interleave here, read_new_lines will skip the malformed JSON line and lose vents. Guard the full record append with a cross-process lock and add a parallel-writer test.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/vent/capture.rs` around lines 35 - 39, Update the append logic around the
OpenOptions handle and JSON serialization to acquire a cross-process file lock
before writing each complete JSONL record, holding it through the newline flush
and releasing it afterward. Ensure concurrent CLI and MCP writers cannot
interleave records, and add a parallel-writer test covering valid, lossless
reads through read_new_lines.
Comment on lines
+16
to
+46
| pub fn read_new_lines(log_path: &Path, cursor_path: &Path) -> (Vec<VentLine>, u64) { | ||
| let mut offset = std::fs::read_to_string(cursor_path) | ||
| .ok() | ||
| .and_then(|s| s.trim().parse::<u64>().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::<VentLine>(&line) { | ||
| lines.push(v); | ||
| } | ||
| } | ||
| (lines, offset + bytes_read) | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the consolidation flow around the cited lines.
sed -n '1,240p' src/vent/consolidate.rs | nl -ba | sed -n '1,220p'
# Inspect the vent upsert implementation referenced in the comment.
sed -n '1,260p' crates/agentflare-backend/src/vent.rs | nl -ba | sed -n '1,260p'
# Find relevant tests or locking mechanisms.
rg -n "consolidate_is_idempotent_via_cursor|advisory|lock|seen_count|upsert" src crates -g '!target'Repository: getappz/agentflare
Length of output: 196
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== src/vent/consolidate.rs =="
awk '{printf "%5d\t%s\n", NR, $0}' src/vent/consolidate.rs | sed -n '1,220p'
echo
echo "== crates/agentflare-backend/src/vent.rs =="
awk '{printf "%5d\t%s\n", NR, $0}' crates/agentflare-backend/src/vent.rs | sed -n '1,260p'
echo
echo "== search =="
rg -n "consolidate_is_idempotent_via_cursor|advisory|lock|seen_count|upsert|consolidate_core|session_start|prompt_submit|agentflare vent consolidate" src crates -g '!target'Repository: getappz/agentflare
Length of output: 50376
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== vents schema / migrations =="
rg -n "CREATE TABLE vents|CREATE UNIQUE INDEX|UNIQUE.*topic_key|topic_key|seen_count" crates src -g '!target' | head -n 120
echo
echo "== schema file slices =="
for f in $(rg -l "CREATE TABLE vents|topic_key" crates src -g '!target' | head -n 20); do
echo "--- $f ---"
awk '{printf "%5d\t%s\n", NR, $0}' "$f" | sed -n '1,220p'
echo
done
echo
echo "== consolidate tests around cursor/idempotency =="
awk '{printf "%5d\t%s\n", NR, $0}' src/vent/consolidate.rs | sed -n '240,340p'Repository: getappz/agentflare
Length of output: 50374
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== item schema/constraints =="
rg -n "CREATE TABLE items|UNIQUE\\(|external_source|external_id|item_id" crates/agentflare-backend/src src -g '!target' | head -n 200
echo
echo "== item create implementation =="
rg -n "pub fn create\\(|CreateItem" crates/agentflare-backend/src src -g '!target' | head -n 80
echo
echo "== consolidate tests around item creation =="
awk '{printf "%5d\t%s\n", NR, $0}' src/vent/consolidate.rs | sed -n '240,330p'Repository: getappz/agentflare
Length of output: 26409
🏁 Script executed:
#!/bin/bash
set -euo pipefail
echo "== crates/agentflare-backend/src/item.rs (create + query bits) =="
awk 'NR>=1 && NR<=220 {printf "%5d\t%s\n", NR, $0}' crates/agentflare-backend/src/item.rs
echo
echo "== item table schema in migrations =="
awk 'NR>=1 && NR<=220 {printf "%5d\t%s\n", NR, $0}' crates/agentflare-backend/src/migrations/0001_initial.sql | sed -n '1,220p'Repository: getappz/agentflare
Length of output: 15703
Serialize vent consolidation. consolidate_core still has a cross-process race: two runs can read the same log window before the cursor advances, and the vent row is only deduped by UNIQUE(project_id, topic_key). The second run can observe item_id = NULL and create a second item for the same topic, and seen_count can be incremented twice. An exclusive advisory lock around the read → upsert/create → cursor write path would prevent overlapping runs.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/vent/consolidate.rs` around lines 16 - 46, Add an exclusive cross-process
advisory lock around the complete consolidation transaction in consolidate_core,
covering log reading, vent-row upsert/item creation, seen_count updates, and
cursor persistence. Ensure competing runs cannot enter this read-to-write
sequence simultaneously, and release the lock on every return path.
Comment on lines
+74
to
+119
| struct Group { | ||
| message: String, | ||
| severity: String, | ||
| count: i64, | ||
| first_event: String, | ||
| } | ||
| let mut groups: BTreeMap<String, Group> = 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; |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Tags are silently dropped during consolidation.
Group never captures VentLine.tags, and tags_json is hardcoded to "[]" at Line 98. The --tag/tags inputs accepted by the CLI (say --tag ...) and MCP vent tool are persisted in the raw JSONL, but consolidation throws them away before they reach agentflare_backend::vent::upsert's tags_json — so they never populate vents.tags, and the created item's metadata (Lines 125-132) also omits them. Tags are effectively write-only today.
🐛 Proposed fix
struct Group {
message: String,
severity: String,
count: i64,
first_event: String,
+ tags: Vec<String>,
}
let mut groups: BTreeMap<String, Group> = 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(),
+ tags: Vec::new(),
});
g.count += 1;
g.message = l.message.clone();
if severity_rank(&l.severity) > severity_rank(&g.severity) {
g.severity = l.severity.clone();
}
+ for t in &l.tags {
+ if !g.tags.contains(t) {
+ g.tags.push(t.clone());
+ }
+ }
}
for (key, g) in groups {
report.consolidated += g.count as usize;
- let tags_json = "[]";
+ let tags_json = serde_json::to_string(&g.tags).unwrap_or_else(|_| "[]".to_string());📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| struct Group { | |
| message: String, | |
| severity: String, | |
| count: i64, | |
| first_event: String, | |
| } | |
| let mut groups: BTreeMap<String, Group> = 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; | |
| struct Group { | |
| message: String, | |
| severity: String, | |
| count: i64, | |
| first_event: String, | |
| tags: Vec<String>, | |
| } | |
| let mut groups: BTreeMap<String, Group> = 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(), | |
| tags: Vec::new(), | |
| }); | |
| g.count += 1; | |
| g.message = l.message.clone(); | |
| if severity_rank(&l.severity) > severity_rank(&g.severity) { | |
| g.severity = l.severity.clone(); | |
| } | |
| for t in &l.tags { | |
| if !g.tags.contains(t) { | |
| g.tags.push(t.clone()); | |
| } | |
| } | |
| } | |
| for (key, g) in groups { | |
| report.consolidated += g.count as usize; | |
| let tags_json = serde_json::to_string(&g.tags).unwrap_or_else(|_| "[]".to_string()); | |
| 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; |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/vent/consolidate.rs` around lines 74 - 119, Preserve VentLine.tags during
consolidation by adding the tags field to Group, initializing it from the first
line, and merging tags from subsequent lines without duplicates. Replace the
hardcoded tags_json in the consolidation upsert flow with serialized Group tags,
and include the same tags in the created item's metadata so both vents.tags and
metadata retain CLI/MCP-provided tags.
Comment on lines
+28
to
+45
| 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())) | ||
| } |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Use a collision-resistant repository filename key.
repo_slug() maps distinct repository identities to the same filename. Those repositories then share both JSONL input and cursor state, so vents can be consolidated into the wrong project. Encode every non-alphanumeric byte unambiguously or use a stable hash of repo_key().
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/vent/paths.rs` around lines 28 - 45, Update repo_slug() to generate a
collision-resistant filename key from repo_key(), using unambiguous encoding for
non-alphanumeric content or a stable hash; preserve its String return type so
log_path() and cursor_path() continue addressing repository-specific files.
…aces log_and_cursor_are_siblings_under_state_dir read cwd/AGENTFLARE_HOME_OVERRIDE via repo_key() without holding the PATH_LOCK other tests use to serialize that same global state, so a concurrent with_temp_cwd/with_temp_home test could change it between the two repo_key() calls and break the assertion. Flaked on Linux CI's build job (CI Green required-check failure on PR #259).
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
ventMCP tool +agentflare ventCLI (say/consolidate/list): agents log tooling friction to an append-only per-repo JSONL, never failing the caller.ventstable (backend.db) storing dedup/seen-count/actionable state per epic refactor(mcp): split item_inner's dispatch arms into src/mcp_server/item.rs #187's two-layer rule.Review fixes folded in (see item #204 for the full review thread)
pre_tool_use(fires every tool call, and only behind an unrelated nudge condition) instead ofprompt_submit/session_start— fixed.vent say --severitydidn't normalize input like the MCP tool did — extracted a sharednormalize_severity().vent::paths::repo_root()reimplemented repo-root resolution and dropped the non-git ROOT_MARKERS fallback used elsewhere — now delegates toAgentflareMcp::repo_root().VentCLI command had replacedmod serve;/Commands::Serveinstead of being added alongside it (restored the dashboardagentflare servecommand), anduuid::Uuid::now_v7()no longer compiled after feat(items): accept sequence_id in item/claim tools; switch id generation to nanoid #257 moved id generation todb_kit::ids::new_id()(swapped to match).manual_repeat_nnit.Test plan
cargo fmt --checkcleancargo clippy --workspace --all-targets --all-features -- -D warningsclean (only the pre-existing, plan-allowed Windows-only handoff: assign items + attach versioned assets instead of raw artifacts #169 note)cargo test --workspace— 652 passed, 0 failed, 1 pre-existing ignoredSummary by CodeRabbit
--features semantic, with a backfill command to index existing observations.ventsdatabase table and migration to support reliable vent storage and querying.