fix: vent tag union + flare-docs fetch-outcome observability

crates/agentflare-backend/src/vent.rs

@@ -38,7 +38,7 @@ pub fn upsert(
 ) -> Result<UpsertOutcome> {
     let existing = conn
         .query_row(
-            "SELECT id, seen_count, item_id, actionable FROM vents
+            "SELECT id, seen_count, item_id, actionable, tags FROM vents
              WHERE project_id = ?1 AND topic_key = ?2",
             params![project_id, topic_key],
             |r| {
@@ -47,17 +47,19 @@ pub fn upsert(
                     r.get::<_, i64>(1)?,
                     r.get::<_, Option<String>>(2)?,
                     r.get::<_, i64>(3)? != 0,
+                    r.get::<_, String>(4)?,
                 ))
             },
         )
         .optional()?;
 
-    if let Some((id, seen, item_id, actionable)) = existing {
+    if let Some((id, seen, item_id, actionable, old_tags_json)) = existing {
         let new_seen = seen + seen_delta;
+        let merged_tags_json = merge_tags_json(&old_tags_json, tags_json);
         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],
+            params![id, new_seen, message, severity, merged_tags_json, now],
         )?;
         return Ok(UpsertOutcome {
             id,
@@ -92,6 +94,21 @@ pub fn upsert(
     })
 }
 
+/// Unions `new_json`'s tags into `old_json`'s (both JSON string arrays),
+/// deduped and order-preserving. A later upsert with no tags (or a
+/// different subset) must not erase tags a prior upsert already persisted
+/// for the same topic_key.
+fn merge_tags_json(old_json: &str, new_json: &str) -> String {
+    let mut tags: Vec<String> = serde_json::from_str(old_json).unwrap_or_default();
+    let new_tags: Vec<String> = serde_json::from_str(new_json).unwrap_or_default();
+    for t in new_tags {
+        if !tags.contains(&t) {
+            tags.push(t);
+        }
+    }
+    serde_json::to_string(&tags).unwrap_or_else(|_| "[]".to_string())
+}
+
 pub fn link_item(conn: &Connection, vent_id: &str, item_id: &str) -> Result<()> {
     conn.execute(
         "UPDATE vents SET item_id = ?2 WHERE id = ?1",
@@ -181,6 +198,64 @@ mod tests {
         assert_eq!(list(&conn, &p, false).unwrap().len(), 1);
     }
 
+    #[test]
+    fn upsert_unions_tags_instead_of_replacing() {
+        let conn = open_in_memory().unwrap();
+        let p = seed_project(&conn);
+        upsert(
+            &conn,
+            &p,
+            "disk full",
+            "medium",
+            r#"["dx"]"#,
+            "disk full",
+            "ev1",
+            1,
+            100,
+        )
+        .unwrap();
+        // A later upsert with a different (or empty) tag set must not erase
+        // "dx" -- and a fresh tag must still be added.
+        upsert(
+            &conn,
+            &p,
+            "disk full",
+            "high",
+            r#"["perf"]"#,
+            "disk full",
+            "ev2",
+            1,
+            200,
+        )
+        .unwrap();
+        let stored = &list(&conn, &p, false).unwrap()[0].tags;
+        let mut tags: Vec<String> = serde_json::from_str(stored).unwrap();
+        tags.sort();
+        assert_eq!(tags, vec!["dx".to_string(), "perf".to_string()]);
+
+        // An untagged re-upsert must not erase what's already stored.
+        upsert(
+            &conn,
+            &p,
+            "disk full",
+            "high",
+            "[]",
+            "disk full",
+            "ev3",
+            1,
+            300,
+        )
+        .unwrap();
+        let after_untagged = &list(&conn, &p, false).unwrap()[0].tags;
+        let mut tags: Vec<String> = serde_json::from_str(after_untagged).unwrap();
+        tags.sort();
+        assert_eq!(
+            tags,
+            vec!["dx".to_string(), "perf".to_string()],
+            "an untagged upsert must not erase previously-stored tags"
+        );
+    }
+
     #[test]
     fn link_item_and_actionable_filter() {
         let conn = open_in_memory().unwrap();

src/vent/consolidate.rs

@@ -90,7 +90,11 @@ pub fn consolidate_core(
         });
         g.count += 1;
         g.message = l.message.clone();
-        g.tags = l.tags.clone();
+        for t in &l.tags {
+            if !g.tags.contains(t) {
+                g.tags.push(t.clone());
+            }
+        }
         if severity_rank(&l.severity) > severity_rank(&g.severity) {
             g.severity = l.severity.clone();
         }
@@ -335,4 +339,64 @@ mod tests {
             "tags must flow through, not be hardcoded to []"
         );
     }
+
+    fn write_tagged_line(log: &std::path::Path, message: &str, tags: &[&str]) {
+        use std::io::Write;
+        let mut f = std::fs::OpenOptions::new()
+            .create(true)
+            .append(true)
+            .open(log)
+            .unwrap();
+        let line = crate::vent::capture::VentLine {
+            event_id: crate::vent::event_id(&format!("{message}{}", tags.join(","))),
+            ts: "t".into(),
+            session: None,
+            severity: "high".to_string(),
+            tags: tags.iter().map(|t| (*t).to_string()).collect(),
+            message: message.to_string(),
+        };
+        writeln!(f, "{}", serde_json::to_string(&line).unwrap()).unwrap();
+    }
+
+    #[test]
+    fn same_topic_lines_union_tags_within_one_batch() {
+        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_tagged_line(&log, "same broken thing", &["dx"]);
+        write_tagged_line(&log, "same broken thing", &["perf"]);
+        consolidate_core(&conn, &p, &s, &log, &cur).unwrap();
+        let tags_json: String = conn
+            .query_row("SELECT tags FROM vents", [], |r| r.get(0))
+            .unwrap();
+        let mut tags: Vec<String> = serde_json::from_str(&tags_json).unwrap();
+        tags.sort();
+        assert_eq!(
+            tags,
+            vec!["dx".to_string(), "perf".to_string()],
+            "tags from every same-topic line in the batch must union, not just the last line's"
+        );
+    }
+
+    #[test]
+    fn later_untagged_consolidation_does_not_erase_prior_tags() {
+        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_tagged_line(&log, "recurring topic", &["dx"]);
+        consolidate_core(&conn, &p, &s, &log, &cur).unwrap();
+
+        write_tagged_line(&log, "recurring topic", &[]);
+        consolidate_core(&conn, &p, &s, &log, &cur).unwrap();
+
+        let tags_json: String = conn
+            .query_row("SELECT tags FROM vents", [], |r| r.get(0))
+            .unwrap();
+        assert_eq!(
+            tags_json, r#"["dx"]"#,
+            "an untagged run must not erase tags a prior run already persisted"
+        );
+    }
 }