huskies: merge 1033

2026-05-14 13:01:01 +00:00
parent 72d79deec9
commit c353c0a6be
3 changed files with 62 additions and 1 deletions
@@ -51,6 +51,13 @@ pub struct BotContext {
    /// Used to proxy bot commands to the active project over WebSocket (`/ws`).
    /// Empty in standalone mode.
    pub gateway_project_urls: BTreeMap<String, String>,
+    /// Pipeline transition events buffered since the last LLM turn.
+    ///
+    /// A background task appends one compact audit line per real stage
+    /// transition.  `handle_message` drains this buffer and injects it as a
+    /// `<system-reminder>` block at the head of the next user prompt so Timmy
+    /// sees pipeline activity without requiring a separate message.
+    pub pending_pipeline_events: Arc<TokioMutex<Vec<String>>>,
 }

 impl BotContext {
@@ -236,6 +243,7 @@ mod tests {
            gateway_active_project,
            gateway_projects,
            gateway_project_urls,
+            pending_pipeline_events: Arc::new(TokioMutex::new(Vec::new())),
        }
    }

@@ -30,6 +30,30 @@ pub(in crate::chat::transport::matrix::bot) async fn handle_message(
        guard.get(&room_id).and_then(|conv| conv.session_id.clone())
    };

+    // Drain any pipeline transition events buffered since the last LLM turn and
+    // prepend them as a passive <system-reminder> block so Timmy sees pipeline
+    // activity without requiring a separate message.  Capped at 20 lines to
+    // keep context size bounded.
+    const MAX_PIPELINE_EVENTS: usize = 20;
+    let system_reminder_prefix = {
+        let mut guard = ctx.pending_pipeline_events.lock().await;
+        if guard.is_empty() {
+            String::new()
+        } else {
+            let total = guard.len();
+            let lines: Vec<String> = guard.drain(..).collect();
+            drop(guard);
+            let shown_count = total.min(MAX_PIPELINE_EVENTS);
+            let shown = lines[..shown_count].join("\n");
+            let tail = if total > MAX_PIPELINE_EVENTS {
+                format!("\n...and {} more", total - MAX_PIPELINE_EVENTS)
+            } else {
+                String::new()
+            };
+            format!("<system-reminder>\n{shown}{tail}\n</system-reminder>\n")
+        }
+    };
+
    // The prompt is just the current message with sender attribution.
    // Prior conversation context is carried by the Claude Code session.
    let bot_name = &ctx.services.bot_name;
@@ -40,7 +64,7 @@ pub(in crate::chat::transport::matrix::bot) async fn handle_message(
        String::new()
    };
    let prompt = format!(
-        "[Your name is {bot_name}. Refer to yourself as {bot_name}, not Claude.]\n{active_project_ctx}\n{}",
+        "{system_reminder_prefix}[Your name is {bot_name}. Refer to yourself as {bot_name}, not Claude.]\n{active_project_ctx}\n{}",
        format_user_prompt(&sender, &user_message)
    );

@@ -295,6 +295,34 @@ pub async fn run_bot(
        );
    }

+    // Subscribe to pipeline stage transitions and buffer compact audit lines
+    // between Timmy's turns.  Replay events (before == after stage label) are
+    // silently dropped — only real transitions are recorded.
+    let pending_pipeline_events: Arc<TokioMutex<Vec<String>>> =
+        Arc::new(TokioMutex::new(Vec::new()));
+    {
+        use crate::pipeline_state::{format_audit_entry, stage_label, subscribe_transitions};
+        let mut rx = subscribe_transitions();
+        let buf = Arc::clone(&pending_pipeline_events);
+        tokio::spawn(async move {
+            loop {
+                match rx.recv().await {
+                    Ok(fired) => {
+                        if stage_label(&fired.before) == stage_label(&fired.after) {
+                            continue;
+                        }
+                        let line = format_audit_entry(&fired);
+                        buf.lock().await.push(line);
+                    }
+                    Err(tokio::sync::broadcast::error::RecvError::Lagged(n)) => {
+                        slog!("[matrix-bot] pipeline event buffer lagged by {n} events");
+                    }
+                    Err(tokio::sync::broadcast::error::RecvError::Closed) => break,
+                }
+            }
+        });
+    }
+
    let ctx = BotContext {
        services,
        matrix_user_id: bot_user_id,
@@ -309,6 +337,7 @@ pub async fn run_bot(
        gateway_active_project,
        gateway_projects,
        gateway_project_urls,
+        pending_pipeline_events,
    };

    slog!(