feat: bulk feedback collection and thread locking (#111)

listen now returns ALL pending events at once (batch mode) and auto-claims
their threads so multiple agents can work in parallel without conflicts.

- Add claimed_by/claimed_at fields to Thread struct
- Add ThreadClaimed/ThreadReleased event types
- Add claim_thread/release_thread/is_claimed methods to FeedbackStore
- Change listen to batch mode by default (--no-batch for legacy)
- Auto-claim threads on listen, skip already-claimed threads
- Add --agent flag (default: agent-<pid>) for agent identity
- Add release subcommand with atomic comment + release
- Add POST /feedback/api/threads/{id}/release HTTP endpoint
- Add claim badges and release button to browser overlay
- Handle thread_claimed/thread_released events in frontend polling
- Extract process_batch() for testable batch orchestration logic
- 18 new tests (7 store-level, 11 batch-logic)

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: boennemann

Cargo.lock

@@ -90,6 +90,12 @@ pub struct Thread {
     /// The seq of the last message the human has viewed in this thread.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub last_human_seen_seq: Option<u64>,
+    /// Which agent has claimed this thread (agent name/id), if any.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub claimed_by: Option<String>,
+    /// When the claim was created.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub claimed_at: Option<DateTime<Utc>>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub viewport_width: Option<u32>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -127,6 +133,8 @@ pub enum EventType {
     AgentThreadCreated { thread: Thread },
     AgentListening,
     AgentStopped,
+    ThreadClaimed { thread_id: String, agent_id: String },
+    ThreadReleased { thread_id: String, agent_id: String },
 }
 
 // ---------------------------------------------------------------------------
@@ -304,6 +312,72 @@ impl FeedbackStore {
         Ok(())
     }
 
+    /// Claim a thread atomically. Succeeds if unclaimed or already claimed by the same agent.
+    /// Returns Err if already claimed by a different agent.
+    pub fn claim_thread(&self, thread_id: &str, agent_id: &str) -> anyhow::Result<Thread> {
+        let aid = agent_id.to_owned();
+        self.modify_thread(thread_id, move |thread| {
+            if let Some(ref existing) = thread.claimed_by {
+                if existing != &aid {
+                    // Will be caught after modify_thread returns — see below.
+                    return;
+                }
+            }
+            thread.claimed_by = Some(aid);
+            thread.claimed_at = Some(Utc::now());
+        })
+        .and_then(|thread| {
+            // Check if the claim actually took effect.
+            if thread.claimed_by.as_deref() == Some(agent_id) {
+                Ok(thread)
+            } else {
+                anyhow::bail!(
+                    "thread {} already claimed by {}",
+                    thread_id,
+                    thread.claimed_by.as_deref().unwrap_or("unknown")
+                )
+            }
+        })
+    }
+
+    /// Release a claim. If `agent_id` is Some, only releases if it matches the claimer.
+    /// If `agent_id` is None, force-releases (for UI button use).
+    pub fn release_thread(
+        &self,
+        thread_id: &str,
+        agent_id: Option<&str>,
+    ) -> anyhow::Result<Thread> {
+        let aid = agent_id.map(|s| s.to_owned());
+        self.modify_thread(thread_id, move |thread| {
+            if let Some(ref required) = aid {
+                if thread.claimed_by.as_deref() != Some(required.as_str()) {
+                    return; // Don't release — wrong agent.
+                }
+            }
+            thread.claimed_by = None;
+            thread.claimed_at = None;
+        })
+        .and_then(|thread| {
+            if let Some(required) = agent_id {
+                // If the caller specified an agent_id, verify the release happened.
+                if thread.claimed_by.is_some() {
+                    anyhow::bail!(
+                        "thread {} is claimed by {}, not {}",
+                        thread_id,
+                        thread.claimed_by.as_deref().unwrap_or("unknown"),
+                        required
+                    );
+                }
+            }
+            Ok(thread)
+        })
+    }
+
+    /// Check if a thread is currently claimed.
+    pub fn is_claimed(thread: &Thread) -> bool {
+        thread.claimed_by.is_some()
+    }
+
     // -- Event log ------------------------------------------------------------
 
     /// Atomically increment the sequence counter and return the new value.
@@ -509,6 +583,8 @@ pub fn new_thread(
         status: ThreadStatus::Open,
         messages: vec![initial_message],
         last_human_seen_seq: None,
+        claimed_by: None,
+        claimed_at: None,
         viewport_width,
         viewport_height,
         created_at: now,
@@ -889,4 +965,138 @@ mod tests {
         assert_eq!(all_seqs[0], 1);
         assert_eq!(*all_seqs.last().unwrap(), 100);
     }
+
+    #[test]
+    fn test_claim_thread() {
+        let tmp = TempDir::new().unwrap();
+        let store = make_store(&tmp);
+
+        let t = make_thread("Needs fixing");
+        store.save_thread(&t).unwrap();
+
+        let claimed = store.claim_thread(&t.id, "agent-1").unwrap();
+        assert_eq!(claimed.claimed_by.as_deref(), Some("agent-1"));
+        assert!(claimed.claimed_at.is_some());
+
+        // Same agent can re-claim (idempotent).
+        let reclaimed = store.claim_thread(&t.id, "agent-1").unwrap();
+        assert_eq!(reclaimed.claimed_by.as_deref(), Some("agent-1"));
+    }
+
+    #[test]
+    fn test_claim_already_claimed() {
+        let tmp = TempDir::new().unwrap();
+        let store = make_store(&tmp);
+
+        let t = make_thread("Contested");
+        store.save_thread(&t).unwrap();
+
+        store.claim_thread(&t.id, "agent-1").unwrap();
+
+        // Different agent cannot claim.
+        let err = store.claim_thread(&t.id, "agent-2").unwrap_err();
+        assert!(err.to_string().contains("already claimed by agent-1"));
+    }
+
+    #[test]
+    fn test_claim_no_expiry() {
+        let tmp = TempDir::new().unwrap();
+        let store = make_store(&tmp);
+
+        let t = make_thread("Long task");
+        store.save_thread(&t).unwrap();
+
+        store.claim_thread(&t.id, "agent-1").unwrap();
+
+        // Claim persists — another agent still cannot claim.
+        let err = store.claim_thread(&t.id, "agent-2").unwrap_err();
+        assert!(err.to_string().contains("already claimed"));
+    }
+
+    #[test]
+    fn test_release_thread() {
+        let tmp = TempDir::new().unwrap();
+        let store = make_store(&tmp);
+
+        let t = make_thread("Will release");
+        store.save_thread(&t).unwrap();
+
+        store.claim_thread(&t.id, "agent-1").unwrap();
+
+        let released = store.release_thread(&t.id, Some("agent-1")).unwrap();
+        assert!(released.claimed_by.is_none());
+        assert!(released.claimed_at.is_none());
+
+        // Now another agent can claim.
+        let claimed = store.claim_thread(&t.id, "agent-2").unwrap();
+        assert_eq!(claimed.claimed_by.as_deref(), Some("agent-2"));
+    }
+
+    #[test]
+    fn test_release_wrong_agent() {
+        let tmp = TempDir::new().unwrap();
+        let store = make_store(&tmp);
+
+        let t = make_thread("Guarded");
+        store.save_thread(&t).unwrap();
+
+        store.claim_thread(&t.id, "agent-1").unwrap();
+
+        let err = store.release_thread(&t.id, Some("agent-2")).unwrap_err();
+        assert!(err.to_string().contains("claimed by agent-1"));
+    }
+
+    #[test]
+    fn test_force_release() {
+        let tmp = TempDir::new().unwrap();
+        let store = make_store(&tmp);
+
+        let t = make_thread("Force release");
+        store.save_thread(&t).unwrap();
+
+        store.claim_thread(&t.id, "agent-1").unwrap();
+
+        // Force release (no agent_id check) — for UI button.
+        let released = store.release_thread(&t.id, None).unwrap();
+        assert!(released.claimed_by.is_none());
+    }
+
+    #[test]
+    fn test_is_claimed() {
+        let t = make_thread("Check claimed");
+        assert!(!FeedbackStore::is_claimed(&t));
+
+        let mut t2 = t;
+        t2.claimed_by = Some("agent-1".into());
+        t2.claimed_at = Some(Utc::now());
+        assert!(FeedbackStore::is_claimed(&t2));
+    }
+
+    #[test]
+    fn test_claim_event_types_serde() {
+        let event = Event {
+            seq: 1,
+            event_type: EventType::ThreadClaimed {
+                thread_id: "t_abc".into(),
+                agent_id: "agent-1".into(),
+            },
+            timestamp: Utc::now(),
+        };
+        let json = serde_json::to_string(&event).unwrap();
+        assert!(json.contains(r#""event":"thread_claimed"#));
+        assert!(json.contains(r#""agent_id":"agent-1"#));
+        let _: Event = serde_json::from_str(&json).unwrap();
+
+        let event = Event {
+            seq: 2,
+            event_type: EventType::ThreadReleased {
+                thread_id: "t_abc".into(),
+                agent_id: "agent-1".into(),
+            },
+            timestamp: Utc::now(),
+        };
+        let json = serde_json::to_string(&event).unwrap();
+        assert!(json.contains(r#""event":"thread_released"#));
+        let _: Event = serde_json::from_str(&json).unwrap();
+    }
 }

crates/veld-core/src/feedback.rs

@@ -525,6 +525,22 @@
   font-size: 10px; color: var(--vf-text-muted); margin-top: 3px;
   white-space: nowrap; overflow: hidden; text-overflow: ellipsis;
 }
+.veld-feedback-thread-card-claim-badge {
+  font-size: 10px; color: var(--vf-primary); margin-top: 4px;
+  display: flex; align-items: center; gap: 4px;
+}
+.veld-feedback-thread-card-claimed {
+  border-left: 3px solid var(--vf-primary);
+}
+.veld-feedback-thread-detail-claim {
+  display: flex; align-items: center; justify-content: space-between;
+  padding: 6px 10px; margin-bottom: 8px;
+  background: color-mix(in srgb, var(--vf-primary) 10%, transparent);
+  border-radius: 6px; font-size: 11px;
+}
+.veld-feedback-thread-detail-claim-text {
+  color: var(--vf-primary); font-weight: 500;
+}
 
 /* --- Thread messages --- */
 .veld-feedback-thread-messages {
@@ -578,6 +594,7 @@
   align-items: center; gap: 8px;
 }
 .veld-feedback-listening-dot {
+  display: block;
   width: 10px; height: 10px; border-radius: 50%;
   background: var(--vf-accent); flex-shrink: 0;
   animation: veld-feedback-pulse-dot 2s ease-in-out infinite;

crates/veld-daemon/frontend/src/feedback-overlay.css

@@ -160,6 +160,23 @@ function renderThreadDetail(thread: Thread): void {
 
   refs.panelBody.appendChild(header);
 
+  if (thread.claimed_by) {
+    const claimRow = mkEl("div", "thread-detail-claim");
+    claimRow.appendChild(mkEl("span", "thread-detail-claim-text", "\u2699 Being worked on by " + thread.claimed_by));
+    const releaseBtn = mkEl("button", "btn btn-secondary btn-sm", "Release");
+    releaseBtn.addEventListener("click", function () {
+      api("POST", "/threads/" + thread.id + "/release").then(function () {
+        thread.claimed_by = null;
+        thread.claimed_at = null;
+        dispatch({ type: "SET_THREADS", threads: [...getState().threads] });
+        renderPanel();
+        toast("Thread released");
+      });
+    });
+    claimRow.appendChild(releaseBtn);
+    refs.panelBody.appendChild(claimRow);
+  }
+
   if (thread.status === "resolved") {
     const msgList = mkEl("div", "thread-messages-list");
     thread.messages.forEach(function (msg: Message) {
@@ -262,6 +279,12 @@ function makeThreadCard(thread: Thread, isResolved: boolean): HTMLElement {
     card.appendChild(mkEl("div", "thread-card-selector", thread.scope.selector));
   }
 
+  if (thread.claimed_by) {
+    const claimBadge = mkEl("div", "thread-card-claim-badge", "\u2699 " + thread.claimed_by);
+    card.appendChild(claimBadge);
+    card.classList.add(PREFIX + "thread-card-claimed");
+  }
+
   card.addEventListener("click", function () { showThreadDetail(thread.id); });
   return card;
 }

crates/veld-daemon/frontend/src/feedback-overlay/panel.ts

@@ -57,6 +57,28 @@ function handleEvent(event: FeedbackEvent): void {
       dispatch({ type: "SET_LISTENING", listening: false });
       updateListeningModule();
       break;
+    case "thread_claimed":
+      if (event.thread_id) {
+        const clThread = findThread(getState().threads, event.thread_id);
+        if (clThread) {
+          clThread.claimed_by = event.agent_id || null;
+          clThread.claimed_at = new Date().toISOString();
+          dispatch({ type: "SET_THREADS", threads: [...getState().threads] });
+          if (getState().panelOpen) deps().renderPanel();
+        }
+      }
+      break;
+    case "thread_released":
+      if (event.thread_id) {
+        const rlThread = findThread(getState().threads, event.thread_id);
+        if (rlThread) {
+          rlThread.claimed_by = null;
+          rlThread.claimed_at = null;
+          dispatch({ type: "SET_THREADS", threads: [...getState().threads] });
+          if (getState().panelOpen) deps().renderPanel();
+        }
+      }
+      break;
     case "thread_created":
       if (event.thread && !findThread(getState().threads, event.thread.id)) {
         dispatch({ type: "ADD_THREAD", thread: event.thread });