[auto] #86 feat: webhook triggers — GitHub/Slack events start agent workflows

src/automation/mod.rs

@@ -3907,6 +3907,7 @@ mod tests {
             handoff: None,
             observe: None,
             budget: None,
+            webhooks: vec![],
         }
     }
 
@@ -5186,6 +5187,7 @@ mod tests {
             handoff: None,
             observe: None,
             budget: None,
+            webhooks: vec![],
         };
 
         let dir = std::env::temp_dir().join("tutti-test-resolver-control-steps");

src/budget/mod.rs

@@ -270,6 +270,7 @@ mod tests {
                 workspace_weekly_tokens: Some(200),
                 agent_weekly_tokens: caps,
             }),
+            webhooks: vec![],
         }
     }
 

src/cli/doctor.rs

@@ -599,6 +599,7 @@ mod tests {
             handoff: None,
             observe: None,
             budget: None,
+            webhooks: vec![],
         }
     }
 

src/cli/handoff.rs

@@ -585,6 +585,7 @@ mod tests {
             }),
             observe: None,
             budget: None,
+            webhooks: vec![],
         }
     }
 

src/cli/serve.rs

@@ -7,14 +7,15 @@ use crate::scheduler;
 use crate::session::TmuxSession;
 use crate::state;
 use crate::state::ControlEvent;
+use crate::webhook;
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
 use serde_json::{Value, json};
 use std::collections::{HashMap, HashSet};
 use std::io::{self, Read};
 use std::path::{Path, PathBuf};
-use std::sync::Arc;
 use std::sync::mpsc;
+use std::sync::{Arc, Mutex};
 use std::thread;
 use std::time::{Duration, Instant};
 use tiny_http::{Header, Method, Request, Response, Server, StatusCode};
@@ -506,6 +507,14 @@ fn route_request(request: &mut Request, targets: &[WorkspaceTarget]) -> (StatusC
                 api_err("action", "action_failed", e.to_string()).to_string(),
             ),
         }
+    } else if method == Method::Post && path == "/v1/webhooks/generic" {
+        match route_webhook(request, targets) {
+            Ok(value) => (StatusCode(200), value.to_string()),
+            Err(e) => (
+                StatusCode(400),
+                api_err("webhook", "webhook_failed", e.to_string()).to_string(),
+            ),
+        }
     } else {
         (
             StatusCode(404),
@@ -753,6 +762,161 @@ fn execute_action(action: &str, body: &Value, target: &WorkspaceTarget) -> Resul
     }
 }
 
+/// Maximum webhook payload size (1 MB)
+const WEBHOOK_MAX_BODY_BYTES: usize = 1_048_576;
+
+/// Mutex to serialize webhook dispatches so that `with_project_root` cwd changes
+/// do not race across concurrent request threads.
+static WEBHOOK_DISPATCH_LOCK: std::sync::LazyLock<Mutex<()>> =
+    std::sync::LazyLock::new(|| Mutex::new(()));
+
+/// Handle POST /v1/webhooks/generic
+fn route_webhook(request: &mut Request, targets: &[WorkspaceTarget]) -> Result<Value> {
+    // Body size is enforced inside read_json_body via take()
+    let body = read_json_body(request)?;
+
+    let source = body
+        .get("source")
+        .and_then(Value::as_str)
+        .unwrap_or("generic");
+    let event_type = body
+        .get("event")
+        .and_then(Value::as_str)
+        .unwrap_or("unknown");
+    let payload = body.get("payload").cloned().unwrap_or_else(|| body.clone());
+    let workspace_hint = body.get("workspace").and_then(Value::as_str);
+
+    let target = resolve_action_workspace(targets, workspace_hint)?;
+
+    // Deduplication: derive an idempotency key from the delivery ID header,
+    // Idempotency-Key header, or a hash of the payload to prevent duplicate
+    // workflow runs from GitHub retries or concurrent identical deliveries.
+    let dedup_key = webhook_dedup_key(request, &body);
+    if let Some(cached) = idempotency_lookup(target, &dedup_key)? {
+        return Ok(cached.response);
+    }
+
+    let matched = webhook::match_triggers(&target.config.webhooks, source, event_type);
+
+    if matched.is_empty() {
+        webhook::log_event(&target.project_root, source, event_type, None, "no_match");
+        let response = api_ok(
+            "webhook.received",
+            json!({
+                "source": source,
+                "event": event_type,
+                "matched": false,
+                "triggers_fired": 0
+            }),
+        );
+        idempotency_save(target, &dedup_key, "webhook.received", response.clone())?;
+        return Ok(response);
+    }
+
+    // Serialize webhook dispatches behind a mutex so that concurrent requests
+    // cannot race on the process-wide cwd changed by `with_project_root`.
+    let _guard = WEBHOOK_DISPATCH_LOCK
+        .lock()
+        .unwrap_or_else(|e| e.into_inner());
+
+    let mut dispatched = Vec::new();
+    for wh in &matched {
+        if let Some(workflow) = &wh.workflow {
+            with_project_root(&target.project_root, || {
+                super::run::run(
+                    Some(workflow),
+                    None,
+                    false,
+                    wh.agent.as_deref(),
+                    false,
+                    false,
+                    false,
+                )
+            })?;
+            webhook::log_event(
+                &target.project_root,
+                source,
+                event_type,
+                Some(workflow),
+                "dispatched_workflow",
+            );
+            dispatched.push(json!({
+                "type": "workflow",
+                "workflow": workflow,
+                "agent": wh.agent,
+            }));
+        } else if let Some(agent) = &wh.agent {
+            let raw_prompt = wh
+                .prompt
+                .as_deref()
+                .unwrap_or("Webhook event received — check recent events for details.");
+            let prompt = webhook::expand_template(raw_prompt, &payload);
+            let options = super::send::SendOptions {
+                auto_up: false,
+                wait: false,
+                timeout_secs: 900,
+                idle_stable_secs: 5,
+                output: false,
+                output_lines: 200,
+            };
+            with_project_root(&target.project_root, || {
+                super::send::run(agent, std::slice::from_ref(&prompt), options).map(|_| ())
+            })?;
+            webhook::log_event(
+                &target.project_root,
+                source,
+                event_type,
+                Some(agent),
+                "dispatched_send",
+            );
+            dispatched.push(json!({
+                "type": "send",
+                "agent": agent,
+            }));
+        }
+    }
+
+    let response = api_ok(
+        "webhook.received",
+        json!({
+            "source": source,
+            "event": event_type,
+            "matched": true,
+            "triggers_fired": dispatched.len(),
+            "dispatched": dispatched
+        }),
+    );
+    idempotency_save(target, &dedup_key, "webhook.received", response.clone())?;
+    Ok(response)
+}
+
+/// Derive a deduplication key for a webhook request.
+/// Prefers X-GitHub-Delivery or Idempotency-Key headers; falls back to a
+/// hash of the payload body for generic webhook sources.
+fn webhook_dedup_key(request: &Request, body: &Value) -> String {
+    // Check X-GitHub-Delivery header (unique per GitHub webhook delivery)
+    for header in request.headers() {
+        if header.field.equiv("X-GitHub-Delivery") {
+            let val = header.value.as_str().trim();
+            if !val.is_empty() {
+                return format!("webhook:{val}");
+            }
+        }
+    }
+    // Check Idempotency-Key header or body field
+    if let Some(key) = read_idempotency_key(request, body) {
+        return format!("webhook:{key}");
+    }
+    // Fall back to a stable content hash (FNV-1a, deterministic across restarts)
+    let serialized = body.to_string();
+    let mut hash: u64 = 0xcbf29ce484222325;
+    for byte in serialized.bytes() {
+        hash ^= byte as u64;
+        hash = hash.wrapping_mul(0x100000001b3);
+    }
+    format!("webhook:hash:{hash:016x}")
+}
+
 /// Resolve the target workspace for an action, defaulting to the only one if singular
 fn resolve_action_workspace<'a>(
     targets: &'a [WorkspaceTarget],
@@ -773,10 +937,20 @@ fn resolve_action_workspace<'a>(
     ))
 }
 
-/// Read and parse the request body as JSON, defaulting to an empty object
+/// Read and parse the request body as JSON, defaulting to an empty object.
+/// Enforces the body size cap while reading via `take()` to prevent memory exhaustion
+/// from oversized payloads (a malicious client could send more bytes than Content-Length).
 fn read_json_body(request: &mut Request) -> Result<Value> {
     let mut body = String::new();
-    request.as_reader().read_to_string(&mut body)?;
+    request
+        .as_reader()
+        .take((WEBHOOK_MAX_BODY_BYTES + 1) as u64)
+        .read_to_string(&mut body)?;
+    if body.len() > WEBHOOK_MAX_BODY_BYTES {
+        return Err(TuttiError::ConfigValidation(format!(
+            "webhook payload too large (>{WEBHOOK_MAX_BODY_BYTES} bytes)"
+        )));
+    }
     if body.trim().is_empty() {
         return Ok(json!({}));
     }