fix(webhooks,cron): wire trigger pipeline so agents auto-respond reliably (#747)

2026-07-27 21:08:00 +00:00 · 2026-04-24 00:35:27 +05:30
parent 3b83fba93a
commit c78f496686
22 changed files with 1229 additions and 104 deletions
@@ -116,6 +116,15 @@ Quick reference for anyone starting with Claude on this project. Updated by the
 - **Rust-side HTTP timeout is separate** — `src/openhuman/providers/compatible.rs` sets a 120s `reqwest` client timeout on LLM calls. Not changed in #715; relevant if a single LLM round-trip itself stalls for >2 min.
 - **Manual cancel path** — `chatCancel()` in `app/src/services/chatService.ts` → `openhuman.channel_web_cancel` RPC → `cancel_chat()` in `src/openhuman/channels/providers/web.rs`. Fully implemented; the silence timer is an automatic fallback.

+## Webhook & Cron Triggers (Issue #726)
+
+- **Webhook bus was hardcoded 410** — `src/openhuman/webhooks/bus.rs` `WebhookRequestSubscriber::handle()` returned 410 "skill runtime removed" for ALL incoming webhooks. Now routes to echo/agent/skill/404 based on `TunnelRegistration.target_kind`.
+- **WebhookRouter access from bus.rs** — Router lives in `SocketManager::shared.webhook_router` (was `pub(super)`). Added `pub fn webhook_router(&self)` accessor on `SocketManager`; bus.rs reaches it via `global_socket_manager().webhook_router()`.
+- **`TriggerSource` enum: three update points** — Adding new variants requires updating: (a) `slug()` match in `envelope.rs`, (b) exhaustive test match, (c) `handle_triage_evaluate` string match in `agent/schemas.rs` (uses `p.source.as_str()`, not the enum directly).
+- **`CronJobTriggered/CronJobCompleted` were never published** — Defined in `events.rs` and used in tests but never emitted. Now published by `execute_and_persist_job()` in `scheduler.rs`. Adding fields to these variants requires updating ~5 construction sites: `cron/bus.rs`, `composio/bus.rs`, `tree_summarizer/bus.rs`, `channels/proactive.rs`, and `events.rs` tests.
+- **Webhook ops were all stubs** — `list_registrations`, `list_logs`, `clear_logs`, `register_echo`, `unregister_echo` in `ops.rs` all returned empty. Now backed by the real router via a `get_router()` helper.
+- **`GGML_NATIVE=OFF` for cargo check** — Sidestepping the whisper-rs macOS Tahoe build blocker for `cargo check`: `GGML_NATIVE=OFF cargo check --manifest-path Cargo.toml`. Allows compilation checks without the cmake failure.
+
 ## Environment

 - **Core sidecar port** — `7788` (default). Check with `lsof -i :7788`.
@@ -8,6 +8,8 @@ export interface TunnelRegistration {
  skill_id: string;
  tunnel_name: string | null;
  backend_tunnel_id: string | null;
+  /** Optional agent definition ID for agent-type tunnels. */
+  agent_id?: string | null;
 }

 export interface WebhookActivityEntry {
@@ -328,6 +328,7 @@ mod tests {
        // This should pass through (domain = "cron")
        bus.publish(DomainEvent::CronJobTriggered {
            job_id: "j1".into(),
+            job_name: "test-job".into(),
            job_type: "shell".into(),
        });

@@ -116,9 +116,17 @@ pub enum DomainEvent {

    // ── Cron ────────────────────────────────────────────────────────────
    /// A cron job was triggered for execution.
-    CronJobTriggered { job_id: String, job_type: String },
+    CronJobTriggered {
+        job_id: String,
+        job_name: String,
+        job_type: String,
+    },
    /// A cron job completed execution.
-    CronJobCompleted { job_id: String, success: bool },
+    CronJobCompleted {
+        job_id: String,
+        success: bool,
+        output: String,
+    },
    /// A cron job requests delivery of its output to a channel.
    CronDeliveryRequested {
        job_id: String,
@@ -527,6 +535,7 @@ mod tests {
            (
                DomainEvent::CronJobTriggered {
                    job_id: "j".into(),
+                    job_name: "my-job".into(),
                    job_type: "t".into(),
                },
                "cron",
@@ -535,6 +544,7 @@ mod tests {
                DomainEvent::CronJobCompleted {
                    job_id: "j".into(),
                    success: true,
+                    output: "ok".into(),
                },
                "cron",
            ),
@@ -251,11 +251,18 @@ fn handle_triage_evaluate(params: Map<String, Value>) -> ControllerFuture {
    Box::pin(async move {
        let p = deserialize_params::<TriageEvaluateParams>(params)?;

+        tracing::debug!(
+            source = %p.source,
+            dry_run = p.dry_run.unwrap_or(false),
+            has_external_id = p.external_id.is_some(),
+            "[rpc][agent] triage_evaluate received"
+        );
+
        // Build a TriggerEnvelope from the RPC params. Source-specific
-        // variants are discriminated by `p.source`; composio is the
-        // only one today.
+        // variants are discriminated by `p.source`.
        let envelope = match p.source.as_str() {
            "composio" => {
+                tracing::trace!("[rpc][agent] building composio trigger envelope");
                let toolkit = p.toolkit.as_deref().unwrap_or("unknown");
                let trigger = p.trigger.as_deref().unwrap_or("unknown");
                let eid = p.external_id.as_deref().unwrap_or("rpc");
@@ -263,13 +270,52 @@ fn handle_triage_evaluate(params: Map<String, Value>) -> ControllerFuture {
                    toolkit, trigger, "rpc", eid, p.payload,
                )
            }
+            "webhook" => {
+                tracing::trace!("[rpc][agent] building webhook trigger envelope");
+                let tunnel_id = p.external_id.as_deref().unwrap_or("unknown");
+                let method = p.toolkit.as_deref().unwrap_or("POST");
+                let path = p.trigger.as_deref().unwrap_or("/");
+                crate::openhuman::agent::triage::TriggerEnvelope::from_webhook(
+                    tunnel_id, method, path, p.payload,
+                )
+            }
+            "cron" => {
+                tracing::trace!("[rpc][agent] building cron trigger envelope");
+                let job_id = p.external_id.as_deref().unwrap_or("unknown");
+                let job_name = p.display_label.as_str();
+                // Preserve the structured payload — extract the output string
+                // for the envelope label but keep the full JSON for triage.
+                let output = p
+                    .payload
+                    .get("output")
+                    .and_then(Value::as_str)
+                    .unwrap_or(job_name);
+                crate::openhuman::agent::triage::TriggerEnvelope::from_cron(
+                    job_id, job_name, output,
+                )
+            }
+            "external" => {
+                tracing::trace!("[rpc][agent] building external trigger envelope");
+                let caller_id = p.external_id.as_deref().unwrap_or("unknown");
+                let reason = p.display_label.as_str();
+                crate::openhuman::agent::triage::TriggerEnvelope::from_external(
+                    caller_id, reason, p.payload,
+                )
+            }
            other => {
+                tracing::warn!(source = %other, "[rpc][agent] unsupported trigger source");
                return Err(format!(
-                    "unsupported trigger source `{other}` — only `composio` is supported today"
+                    "unsupported trigger source `{other}` — supported: composio, webhook, cron, external"
                ));
            }
        };

+        tracing::debug!(
+            source = %envelope.source.slug(),
+            external_id_len = envelope.external_id.len(),
+            "[rpc][agent] running triage pipeline"
+        );
+
        let run = crate::openhuman::agent::triage::run_triage(&envelope)
            .await
            .map_err(|e| format!("triage evaluation failed: {e}"))?;
@@ -445,7 +491,7 @@ mod tests {
    #[tokio::test]
    async fn triage_handler_rejects_unknown_source_and_to_json_maps_outcome() {
        let err = handle_triage_evaluate(Map::from_iter([
-            ("source".into(), Value::String("webhook".into())),
+            ("source".into(), Value::String("__unknown_source__".into())),
            ("display_label".into(), Value::String("lbl".into())),
            ("payload".into(), json!({})),
        ]))
@@ -27,8 +27,17 @@ pub enum TriggerSource {
        provider: String,
        account_id: String,
    },
-    // Cron / Webhook / … variants will be added in later commits as
-    // those callers wire up the triage pipeline.
+    /// An incoming webhook request routed through the webhook tunnel system.
+    Webhook {
+        tunnel_id: String,
+        method: String,
+        path: String,
+    },
+    /// A cron job that completed and whose output feeds the triage pipeline.
+    Cron { job_id: String, job_name: String },
+    /// An external caller (e.g. another service or RPC client) requesting
+    /// an agent trigger directly.
+    External { caller_id: String, reason: String },
 }

 impl TriggerSource {
@@ -38,6 +47,9 @@ impl TriggerSource {
        match self {
            Self::Composio { .. } => "composio",
            Self::WebviewIntegration { .. } => "webview",
+            Self::Webhook { .. } => "webhook",
+            Self::Cron { .. } => "cron",
+            Self::External { .. } => "external",
        }
    }
 }
@@ -108,6 +120,59 @@ impl TriggerEnvelope {
            received_at: Utc::now(),
        }
    }
+
+    /// Build a `TriggerEnvelope` from an incoming webhook request.
+    ///
+    /// `tunnel_id` is used as the correlation id so webhook responses
+    /// can be matched back to their trigger envelope.
+    pub fn from_webhook(tunnel_id: &str, method: &str, path: &str, payload: Value) -> Self {
+        Self {
+            source: TriggerSource::Webhook {
+                tunnel_id: tunnel_id.to_string(),
+                method: method.to_string(),
+                path: path.to_string(),
+            },
+            external_id: tunnel_id.to_string(),
+            display_label: format!("webhook/{method}/{path}"),
+            payload,
+            received_at: Utc::now(),
+        }
+    }
+
+    /// Build a `TriggerEnvelope` from a completed cron job.
+    ///
+    /// `job_id` is used as the correlation id; `output` is embedded in
+    /// the payload so the triage LLM can see what the job produced.
+    pub fn from_cron(job_id: &str, job_name: &str, output: &str) -> Self {
+        Self {
+            source: TriggerSource::Cron {
+                job_id: job_id.to_string(),
+                job_name: job_name.to_string(),
+            },
+            external_id: job_id.to_string(),
+            display_label: format!("cron/{job_name}"),
+            payload: serde_json::json!({ "output": output }),
+            received_at: Utc::now(),
+        }
+    }
+
+    /// Build a `TriggerEnvelope` from an external caller.
+    ///
+    /// `caller_id` is used as the correlation id. `reason` is a short
+    /// human-readable label explaining what prompted the trigger (e.g.
+    /// `"manual_rpc_test"`, `"ci_pipeline"`, …).
+    pub fn from_external(caller_id: &str, reason: &str, payload: Value) -> Self {
+        Self {
+            source: TriggerSource::External {
+                caller_id: caller_id.to_string(),
+                reason: reason.to_string(),
+            },
+            external_id: caller_id.to_string(),
+            display_label: format!("external/{caller_id}"),
+            payload,
+            received_at: Utc::now(),
+        }
+    }
 }

 #[cfg(test)]
@@ -132,7 +197,7 @@ mod tests {
                assert_eq!(toolkit, "gmail");
                assert_eq!(trigger, "GMAIL_NEW_GMAIL_MESSAGE");
            }
-            _ => panic!("expected Composio source"),
+            _ => panic!("expected Composio variant"),
        }
        assert_eq!(env.payload["from"], "a@b.com");
    }
@@ -148,4 +213,63 @@ mod tests {
        );
        assert_eq!(env.external_id, "trig-fallback");
    }
+
+    #[test]
+    fn webhook_envelope_builds_expected_label_and_slug() {
+        let env = TriggerEnvelope::from_webhook(
+            "tunnel-uuid-1",
+            "POST",
+            "/hooks/test",
+            json!({ "event": "push" }),
+        );
+        assert_eq!(env.display_label, "webhook/POST//hooks/test");
+        assert_eq!(env.external_id, "tunnel-uuid-1");
+        assert_eq!(env.source.slug(), "webhook");
+        match env.source {
+            TriggerSource::Webhook {
+                tunnel_id,
+                method,
+                path,
+            } => {
+                assert_eq!(tunnel_id, "tunnel-uuid-1");
+                assert_eq!(method, "POST");
+                assert_eq!(path, "/hooks/test");
+            }
+            _ => panic!("expected Webhook variant"),
+        }
+        assert_eq!(env.payload["event"], "push");
+    }
+
+    #[test]
+    fn cron_envelope_builds_expected_label_and_slug() {
+        let env = TriggerEnvelope::from_cron("job-1", "morning_briefing", "Briefing complete");
+        assert_eq!(env.display_label, "cron/morning_briefing");
+        assert_eq!(env.external_id, "job-1");
+        assert_eq!(env.source.slug(), "cron");
+        match env.source {
+            TriggerSource::Cron { job_id, job_name } => {
+                assert_eq!(job_id, "job-1");
+                assert_eq!(job_name, "morning_briefing");
+            }
+            _ => panic!("expected Cron variant"),
+        }
+        assert_eq!(env.payload["output"], "Briefing complete");
+    }
+
+    #[test]
+    fn external_envelope_builds_expected_label_and_slug() {
+        let env =
+            TriggerEnvelope::from_external("caller-abc", "ci_pipeline", json!({ "ref": "main" }));
+        assert_eq!(env.display_label, "external/caller-abc");
+        assert_eq!(env.external_id, "caller-abc");
+        assert_eq!(env.source.slug(), "external");
+        match env.source {
+            TriggerSource::External { caller_id, reason } => {
+                assert_eq!(caller_id, "caller-abc");
+                assert_eq!(reason, "ci_pipeline");
+            }
+            _ => panic!("expected External variant"),
+        }
+        assert_eq!(env.payload["ref"], "main");
+    }
 }
@@ -309,6 +309,7 @@ mod tests {

        sub.handle(&DomainEvent::CronJobTriggered {
            job_id: "j".into(),
+            job_name: "test-job".into(),
            job_type: "agent".into(),
        })
        .await;
@@ -502,6 +502,7 @@ mod tests {
        let sub = ComposioTriggerSubscriber::new();
        sub.handle(&DomainEvent::CronJobTriggered {
            job_id: "j1".into(),
+            job_name: "test-job".into(),
            job_type: "shell".into(),
        })
        .await;
@@ -143,6 +143,7 @@ mod tests {

        sub.handle(&DomainEvent::CronJobTriggered {
            job_id: "j1".into(),
+            job_name: "test-job".into(),
            job_type: "shell".into(),
        })
        .await;
@@ -9,7 +9,7 @@ mod types;
 pub mod scheduler;

 pub use ops as rpc;
-pub use ops::{add_once, add_once_at, pause_job, resume_job, update_cron_job};
+pub use ops::{add_once, add_once_at, parse_human_delay, pause_job, resume_job, update_cron_job};
 #[allow(unused_imports)]
 pub use schedule::{
    next_run_for_schedule, normalize_expression, schedule_cron_expression, validate_schedule,
@@ -8,7 +8,7 @@ use anyhow::Result;
 use serde_json::json;

 pub fn add_once(config: &Config, delay: &str, command: &str) -> Result<CronJob> {
-    let duration = parse_delay(delay)?;
+    let duration = parse_human_delay(delay)?;
    let at = chrono::Utc::now() + duration;
    add_once_at(config, at, command)
 }
@@ -93,7 +93,9 @@ pub fn update_cron_job(
    update_job(config, id, patch)
 }

-fn parse_delay(input: &str) -> Result<chrono::Duration> {
+/// Parse a human-friendly delay string (e.g. "5m", "2h", "30s") into a
+/// `chrono::Duration`. Defaults to minutes when no unit is given.
+pub fn parse_human_delay(input: &str) -> Result<chrono::Duration> {
    let input = input.trim();
    if input.is_empty() {
        anyhow::bail!("delay must not be empty");
@@ -195,6 +197,10 @@ pub async fn cron_run(
    );
    let _ = cron::record_last_run(config, &job.id, finished_at, success, &output);

+    // Deliver via the same path as the scheduler loop so proactive
+    // messages and alerts are sent on "Run Now" too.
+    cron::scheduler::deliver_job(config, &job, &output).await;
+
    Ok(RpcOutcome::new(
        json!({
            "job_id": job.id,
@@ -383,43 +389,55 @@ mod tests {

    #[test]
    fn parse_delay_accepts_seconds_minutes_hours_days() {
-        assert_eq!(parse_delay("5s").unwrap(), chrono::Duration::seconds(5));
-        assert_eq!(parse_delay("10m").unwrap(), chrono::Duration::minutes(10));
-        assert_eq!(parse_delay("2h").unwrap(), chrono::Duration::hours(2));
-        assert_eq!(parse_delay("3d").unwrap(), chrono::Duration::days(3));
+        assert_eq!(
+            parse_human_delay("5s").unwrap(),
+            chrono::Duration::seconds(5)
+        );
+        assert_eq!(
+            parse_human_delay("10m").unwrap(),
+            chrono::Duration::minutes(10)
+        );
+        assert_eq!(parse_human_delay("2h").unwrap(), chrono::Duration::hours(2));
+        assert_eq!(parse_human_delay("3d").unwrap(), chrono::Duration::days(3));
    }

    #[test]
    fn parse_delay_defaults_to_minutes_when_no_unit() {
-        assert_eq!(parse_delay("15").unwrap(), chrono::Duration::minutes(15));
+        assert_eq!(
+            parse_human_delay("15").unwrap(),
+            chrono::Duration::minutes(15)
+        );
    }

    #[test]
    fn parse_delay_trims_whitespace() {
-        assert_eq!(parse_delay("  7m  ").unwrap(), chrono::Duration::minutes(7));
+        assert_eq!(
+            parse_human_delay("  7m  ").unwrap(),
+            chrono::Duration::minutes(7)
+        );
    }

    #[test]
    fn parse_delay_rejects_empty_input() {
-        let err = parse_delay("").unwrap_err();
+        let err = parse_human_delay("").unwrap_err();
        assert!(err.to_string().contains("delay must not be empty"));
-        let err = parse_delay("   ").unwrap_err();
+        let err = parse_human_delay("   ").unwrap_err();
        assert!(err.to_string().contains("delay must not be empty"));
    }

    #[test]
    fn parse_delay_rejects_unsupported_unit() {
-        let err = parse_delay("5x").unwrap_err();
+        let err = parse_human_delay("5x").unwrap_err();
        assert!(err.to_string().contains("unsupported delay unit"));
        // Multi-char unit not matched in the parse branch either.
-        let err = parse_delay("5wk").unwrap_err();
+        let err = parse_human_delay("5wk").unwrap_err();
        assert!(err.to_string().contains("unsupported delay unit"));
    }

    #[test]
    fn parse_delay_rejects_non_numeric_prefix() {
        // No ascii-digit prefix at all → empty num, parse() fails.
-        assert!(parse_delay("abc").is_err());
+        assert!(parse_human_delay("abc").is_err());
    }

    // ── add_once ────────────────────────────────────────────────────
@@ -53,6 +53,19 @@ pub async fn run(config: Config) -> Result<()> {
    }
 }

+/// Public entry point for delivering a job's output via the configured
+/// delivery mode (proactive / announce). Called by `cron_run` ("Run Now")
+/// so manual runs also push notifications and alerts.
+pub async fn deliver_job(config: &Config, job: &CronJob, output: &str) {
+    if let Err(e) = deliver_if_configured(config, job, output).await {
+        if job.delivery.best_effort {
+            tracing::warn!("[cron] delivery failed (best_effort, Run Now): {e}");
+        } else {
+            tracing::warn!("[cron] delivery failed (Run Now): {e}");
+        }
+    }
+}
+
 pub async fn execute_job_now(config: &Config, job: &CronJob) -> (bool, String) {
    let security = SecurityPolicy::from_config(&config.autonomy, &config.workspace_dir);
    execute_job_with_retry(config, &security, job).await
@@ -127,9 +140,30 @@ async fn execute_and_persist_job(
    warn_if_high_frequency_agent_job(job);

    let started_at = Utc::now();
-    let (success, output) = execute_job_with_retry(config, security, job).await;
+
+    publish_global(DomainEvent::CronJobTriggered {
+        job_id: job.id.clone(),
+        job_name: job.name.clone().unwrap_or_default(),
+        job_type: format!("{:?}", job.job_type),
+    });
+
+    let (execution_success, output) = execute_job_with_retry(config, security, job).await;
    let finished_at = Utc::now();
-    let success = persist_job_result(config, job, success, &output, started_at, finished_at).await;
+    let success = persist_job_result(
+        config,
+        job,
+        execution_success,
+        &output,
+        started_at,
+        finished_at,
+    )
+    .await;
+
+    publish_global(DomainEvent::CronJobCompleted {
+        job_id: job.id.clone(),
+        success,
+        output: crate::openhuman::util::truncate_with_ellipsis(&output, 512),
+    });
    let failure_message =
        (!success).then(|| crate::openhuman::util::truncate_with_ellipsis(&output, 256));

@@ -308,7 +342,7 @@ fn warn_if_high_frequency_agent_job(job: &CronJob) {
    }
 }

-async fn deliver_if_configured(_config: &Config, job: &CronJob, output: &str) -> Result<()> {
+async fn deliver_if_configured(config: &Config, job: &CronJob, output: &str) -> Result<()> {
    let delivery: &DeliveryConfig = &job.delivery;

    let mode = delivery.mode.trim().to_ascii_lowercase();
@@ -328,6 +362,9 @@ async fn deliver_if_configured(_config: &Config, job: &CronJob, output: &str) ->
                message: output.to_string(),
                job_name: job.name.clone(),
            });
+
+            // Also push to the alerts tab so the user sees it in /notifications.
+            push_cron_alert(config, job, output);
        }

        // Announce delivery — the cron job specifies the exact channel
@@ -354,6 +391,8 @@ async fn deliver_if_configured(_config: &Config, job: &CronJob, output: &str) ->
                target: target.to_string(),
                output: output.to_string(),
            });
+
+            push_cron_alert(config, job, output);
        }

        // No delivery configured — output is stored in last_output only.
@@ -363,6 +402,47 @@ async fn deliver_if_configured(_config: &Config, job: &CronJob, output: &str) ->
    Ok(())
 }

+/// Insert a notification into the alerts tab for a completed cron job.
+fn push_cron_alert(config: &Config, job: &CronJob, output: &str) {
+    use crate::openhuman::notifications::store as notif_store;
+    use crate::openhuman::notifications::types::{IntegrationNotification, NotificationStatus};
+
+    let name = job.name.as_deref().unwrap_or("Cron job");
+    let truncated = crate::openhuman::util::truncate_with_ellipsis(output, 512);
+
+    let notification = IntegrationNotification {
+        id: uuid::Uuid::new_v4().to_string(),
+        provider: "cron".to_string(),
+        account_id: Some(job.id.clone()),
+        title: name.to_string(),
+        body: truncated,
+        raw_payload: serde_json::json!({
+            "job_id": job.id,
+            "job_name": job.name,
+            "delivery_mode": job.delivery.mode,
+        }),
+        importance_score: Some(0.65),
+        triage_action: Some("react".to_string()),
+        triage_reason: Some("Scheduled delivery".to_string()),
+        status: NotificationStatus::Unread,
+        received_at: Utc::now(),
+        scored_at: Some(Utc::now()),
+    };
+
+    if let Err(e) = notif_store::insert(config, &notification) {
+        tracing::warn!(
+            job_id = %job.id,
+            error = %e,
+            "[cron] failed to push notification alert"
+        );
+    } else {
+        tracing::debug!(
+            job_id = %job.id,
+            "[cron] pushed notification alert to alerts tab"
+        );
+    }
+}
+
 fn is_env_assignment(word: &str) -> bool {
    word.contains('=')
        && word
@@ -62,7 +62,9 @@ fn now_ms() -> u64 {
 pub fn event_to_notification(event: &DomainEvent) -> Option<CoreNotificationEvent> {
    let ts = now_ms();
    match event {
-        DomainEvent::CronJobCompleted { job_id, success } => Some(CoreNotificationEvent {
+        DomainEvent::CronJobCompleted {
+            job_id, success, ..
+        } => Some(CoreNotificationEvent {
            id: format!("cron:{}:{}", job_id, ts),
            category: CoreNotificationCategory::Agents,
            title: if *success {
@@ -184,6 +186,7 @@ mod tests {
        let ev = DomainEvent::CronJobCompleted {
            job_id: "job-1".into(),
            success: true,
+            output: "done".into(),
        };
        let n = event_to_notification(&ev).expect("should produce notification");
        assert_eq!(n.category, CoreNotificationCategory::Agents);
@@ -196,6 +199,7 @@ mod tests {
        let ev = DomainEvent::CronJobCompleted {
            job_id: "job-1".into(),
            success: false,
+            output: "error".into(),
        };
        let n = event_to_notification(&ev).unwrap();
        assert_eq!(n.title, "Cron job failed");
@@ -95,6 +95,11 @@ impl SocketManager {
        *self.shared.webhook_router.write() = Some(router);
    }

+    /// Get the webhook router, if one has been set.
+    pub fn webhook_router(&self) -> Option<Arc<WebhookRouter>> {
+        self.shared.webhook_router.read().clone()
+    }
+
    /// Get the current socket state (status, ID, error).
    pub fn get_state(&self) -> SocketState {
        SocketState {
@@ -31,7 +31,7 @@ impl Tool for CronAddTool {
        json!({
            "type": "object",
            "properties": {
-                "name": { "type": "string" },
+                "name": { "type": "string", "description": "Short human-readable name for the job (e.g. 'drink_water_reminder'). Always provide a name." },
                "schedule": {
                    "type": "object",
                    "description": "Schedule object: {kind:'cron',expr,tz?} | {kind:'at',at} | {kind:'every',every_ms}"
@@ -41,10 +41,19 @@ impl Tool for CronAddTool {
                "prompt": { "type": "string" },
                "session_target": { "type": "string", "enum": ["isolated", "main"] },
                "model": { "type": "string" },
-                "delivery": { "type": "object" },
+                "delivery": {
+                    "type": "object",
+                    "description": "Delivery config. Defaults to proactive (notifies user). Modes: proactive, announce (needs channel+to), none (silent).",
+                    "properties": {
+                        "mode": { "type": "string", "enum": ["proactive", "announce", "none"] },
+                        "channel": { "type": "string", "description": "Required for announce mode" },
+                        "to": { "type": "string", "description": "Required for announce mode" },
+                        "best_effort": { "type": "boolean", "default": true }
+                    }
+                },
                "delete_after_run": { "type": "boolean" }
            },
-            "required": ["schedule"]
+            "required": ["name", "schedule"]
        })
    }

@@ -72,7 +81,27 @@ impl Tool for CronAddTool {
        let name = args
            .get("name")
            .and_then(serde_json::Value::as_str)
-            .map(str::to_string);
+            .map(str::to_string)
+            .or_else(|| {
+                // Derive a name from the prompt so cron jobs are never unnamed.
+                args.get("prompt")
+                    .and_then(serde_json::Value::as_str)
+                    .map(|p| {
+                        let slug: String = p
+                            .chars()
+                            .map(|c| {
+                                if c.is_alphanumeric() {
+                                    c.to_ascii_lowercase()
+                                } else {
+                                    '_'
+                                }
+                            })
+                            .take(48)
+                            .collect();
+                        slug.trim_matches('_').to_string()
+                    })
+                    .filter(|s| !s.is_empty())
+            });

        let job_type = match args.get("job_type").and_then(serde_json::Value::as_str) {
            Some("agent") => JobType::Agent,
@@ -146,7 +175,12 @@ impl Tool for CronAddTool {
                            return Ok(ToolResult::error(format!("Invalid delivery config: {e}")));
                        }
                    },
-                    None => None,
+                    None => Some(DeliveryConfig {
+                        mode: "proactive".to_string(),
+                        channel: None,
+                        to: None,
+                        best_effort: true,
+                    }),
                };

                cron::add_agent_job(
@@ -268,6 +302,47 @@ mod tests {
        assert!(result.output().contains("every_ms must be > 0"));
    }

+    #[tokio::test]
+    async fn agent_job_defaults_to_proactive_delivery() {
+        let tmp = TempDir::new().unwrap();
+        let cfg = test_config(&tmp).await;
+        let tool = CronAddTool::new(cfg.clone(), test_security(&cfg));
+        let result = tool
+            .execute(json!({
+                "schedule": { "kind": "every", "every_ms": 300000 },
+                "job_type": "agent",
+                "prompt": "remind me to drink water"
+            }))
+            .await
+            .unwrap();
+
+        assert!(!result.is_error, "{:?}", result.output());
+        let jobs = cron::list_jobs(&cfg).unwrap();
+        assert_eq!(jobs.len(), 1);
+        assert_eq!(jobs[0].delivery.mode, "proactive");
+    }
+
+    #[tokio::test]
+    async fn agent_job_respects_explicit_none_delivery() {
+        let tmp = TempDir::new().unwrap();
+        let cfg = test_config(&tmp).await;
+        let tool = CronAddTool::new(cfg.clone(), test_security(&cfg));
+        let result = tool
+            .execute(json!({
+                "schedule": { "kind": "every", "every_ms": 300000 },
+                "job_type": "agent",
+                "prompt": "silent background task",
+                "delivery": { "mode": "none" }
+            }))
+            .await
+            .unwrap();
+
+        assert!(!result.is_error, "{:?}", result.output());
+        let jobs = cron::list_jobs(&cfg).unwrap();
+        assert_eq!(jobs.len(), 1);
+        assert_eq!(jobs[0].delivery.mode, "none");
+    }
+
    #[tokio::test]
    async fn agent_job_requires_prompt() {
        let tmp = TempDir::new().unwrap();
@@ -1,5 +1,5 @@
 use crate::openhuman::config::Config;
-use crate::openhuman::cron;
+use crate::openhuman::cron::{self, DeliveryConfig, Schedule, SessionTarget};
 use crate::openhuman::security::SecurityPolicy;
 use crate::openhuman::tools::traits::{Tool, ToolResult};
 use anyhow::Result;
@@ -53,7 +53,15 @@ impl Tool for ScheduleTool {
                },
                "command": {
                    "type": "string",
-                    "description": "Shell command to execute. Required for create/add/once."
+                    "description": "Shell command to execute. Use 'command' for shell jobs OR 'prompt' for agent jobs."
+                },
+                "prompt": {
+                    "type": "string",
+                    "description": "Agent prompt for recurring agent tasks (e.g. reminders, briefings). Use this instead of 'command' for user-facing notifications."
+                },
+                "name": {
+                    "type": "string",
+                    "description": "Short human-readable name for the job (e.g. 'drink_water_reminder'). Always provide a name."
                },
                "id": {
                    "type": "string",
@@ -201,8 +209,26 @@ impl ScheduleTool {
        let command = args
            .get("command")
            .and_then(|value| value.as_str())
-            .filter(|value| !value.trim().is_empty())
-            .ok_or_else(|| anyhow::anyhow!("Missing or empty 'command' parameter"))?;
+            .filter(|value| !value.trim().is_empty());
+        let prompt = args
+            .get("prompt")
+            .and_then(|value| value.as_str())
+            .filter(|value| !value.trim().is_empty());
+
+        // If the LLM passed a "command" that isn't a real shell command,
+        // treat it as an agent prompt instead. This handles the common case
+        // where the LLM puts "remind me to drink water" in the command field.
+        let (command, prompt) = match (command, prompt) {
+            (Some(cmd), None) if !looks_like_shell_command(cmd) => (None, Some(cmd)),
+            other => other,
+        };
+
+        // Must have either command (shell) or prompt (agent).
+        if command.is_none() && prompt.is_none() {
+            return Ok(ToolResult::error(
+                "Provide 'command' for shell jobs or 'prompt' for agent jobs.".to_string(),
+            ));
+        }

        let expression = args.get("expression").and_then(|value| value.as_str());
        let delay = args.get("delay").and_then(|value| value.as_str());
@@ -241,6 +267,89 @@ impl ScheduleTool {
            }
        }

+        // ── Agent job (prompt provided) ──────────────────────────────
+        if let Some(prompt_text) = prompt {
+            tracing::debug!(
+                action = %action,
+                prompt_len = prompt_text.len(),
+                "[schedule] creating agent job"
+            );
+            let schedule = if let Some(expr) = expression {
+                Schedule::Cron {
+                    expr: expr.to_string(),
+                    tz: None,
+                }
+            } else if let Some(delay_str) = delay {
+                let at = Utc::now() + cron::parse_human_delay(delay_str)?;
+                Schedule::At { at }
+            } else if let Some(at_str) = run_at {
+                let at: DateTime<Utc> = DateTime::parse_from_rfc3339(at_str)
+                    .map_err(|e| anyhow::anyhow!("Invalid run_at timestamp: {e}"))?
+                    .with_timezone(&Utc);
+                Schedule::At { at }
+            } else {
+                return Ok(ToolResult::error("Missing scheduling parameters"));
+            };
+
+            let name = args
+                .get("name")
+                .and_then(|v| v.as_str())
+                .map(str::to_string)
+                .or_else(|| {
+                    // Derive a slug from the prompt so jobs are never unnamed.
+                    Some(
+                        prompt_text
+                            .chars()
+                            .map(|c| {
+                                if c.is_alphanumeric() {
+                                    c.to_ascii_lowercase()
+                                } else {
+                                    '_'
+                                }
+                            })
+                            .take(48)
+                            .collect::<String>()
+                            .trim_matches('_')
+                            .to_string(),
+                    )
+                    .filter(|s| !s.is_empty())
+                });
+
+            let delete_after_run = matches!(schedule, Schedule::At { .. });
+            let delivery = Some(DeliveryConfig {
+                mode: "proactive".to_string(),
+                channel: None,
+                to: None,
+                best_effort: true,
+            });
+
+            let job = cron::add_agent_job(
+                &self.config,
+                name,
+                schedule,
+                prompt_text,
+                SessionTarget::Isolated,
+                None,
+                delivery,
+                delete_after_run,
+            )?;
+
+            let job_name = job.name.as_deref().unwrap_or("(unnamed)");
+            tracing::debug!(
+                job_id = %job.id,
+                job_name = %job_name,
+                next_run = %job.next_run,
+                "[schedule] agent job created"
+            );
+            return Ok(ToolResult::success(format!(
+                "Created agent job '{}' (id: {}, next: {})",
+                job_name, job.id, job.next_run,
+            )));
+        }
+
+        // ── Shell job (command provided) ─────────────────────────────
+        let command = command.unwrap();
+
        if let Some(value) = expression {
            let job = cron::add_job(&self.config, value, command)?;
            return Ok(ToolResult::success(format!(
@@ -301,6 +410,34 @@ impl ScheduleTool {
    }
 }

+/// Heuristic: does this look like a shell command rather than a natural
+/// language prompt? Shell commands typically start with an executable name
+/// or path and contain shell metacharacters.
+fn looks_like_shell_command(input: &str) -> bool {
+    let trimmed = input.trim();
+    if trimmed.is_empty() {
+        return false;
+    }
+    // Starts with a path or known shell built-in
+    if trimmed.starts_with('/') || trimmed.starts_with("./") || trimmed.starts_with("~/") {
+        return true;
+    }
+    // Contains shell operators
+    if trimmed.contains('|') || trimmed.contains("&&") || trimmed.contains(">>") {
+        return true;
+    }
+    // First word is a common CLI executable
+    let first_word = trimmed.split_whitespace().next().unwrap_or("");
+    // Exclude ambiguous words (test, find, make, source, head, sort) that
+    // are common English verbs and would misclassify natural-language prompts.
+    const SHELL_COMMANDS: &[&str] = &[
+        "echo", "cat", "ls", "cd", "cp", "mv", "rm", "mkdir", "grep", "sed", "awk", "curl", "wget",
+        "python", "python3", "node", "npm", "yarn", "cargo", "bash", "sh", "zsh", "git", "docker",
+        "kubectl", "env", "export", "tail", "wc", "tar", "zip", "unzip",
+    ];
+    SHELL_COMMANDS.contains(&first_word)
+}
+
 #[cfg(test)]
 mod tests {
    use super::*;
@@ -123,6 +123,7 @@ mod tests {
        let sub = TreeSummarizerEventSubscriber::new();
        sub.handle(&DomainEvent::CronJobTriggered {
            job_id: "j1".into(),
+            job_name: "test-job".into(),
            job_type: "shell".into(),
        })
        .await;
@@ -7,8 +7,10 @@

 use crate::core::event_bus::{publish_global, DomainEvent, EventHandler};
 use crate::openhuman::socket::global_socket_manager;
+use crate::openhuman::webhooks::WebhookResponseData;
 use async_trait::async_trait;
 use serde_json::json;
+use std::collections::HashMap;
 use std::time::Instant;

 /// Base64-encode a string (for webhook response bodies).
@@ -74,21 +76,142 @@ impl EventHandler for WebhookRequestSubscriber {
            correlation_id,
        );

-        tracing::debug!(
-            "[webhook] skill runtime removed; rejecting tunnel {} ({})",
-            tunnel_uuid,
-            tunnel_name
-        );
-        let response = crate::openhuman::webhooks::WebhookResponseData {
-            correlation_id: correlation_id.clone(),
-            status_code: 410,
-            headers: std::collections::HashMap::new(),
-            body: error_body("Webhook skill runtime has been removed"),
-        };
-        let resolved_skill_id: Option<String> = None;
-        let response_error = Some("webhook skill runtime removed".to_string());
+        // Retrieve the router from the global socket manager.
+        let router = global_socket_manager().and_then(|mgr| mgr.webhook_router());

-        // Publish notification events
+        // Look up the registration for this tunnel.
+        let registration = router.as_ref().and_then(|r| r.registration(&tunnel_uuid));
+
+        let (response, resolved_skill_id, response_error) = match registration {
+            Some(ref reg) if reg.target_kind == "echo" => {
+                tracing::debug!(
+                    "[webhook] echo tunnel {} — returning echo response",
+                    tunnel_uuid
+                );
+                let resp = crate::openhuman::webhooks::ops::build_echo_response(request);
+                (resp, Some("echo".to_string()), None)
+            }
+            Some(ref reg) if reg.target_kind == "agent" => {
+                tracing::info!(
+                    "[webhook] agent tunnel {} — routing to triage pipeline",
+                    tunnel_uuid
+                );
+                let decoded = decode_webhook_body(&request.body);
+                if let Err(e) = &decoded {
+                    tracing::error!("[webhook] rejecting — failed to decode body: {}", e);
+                    let resp = WebhookResponseData {
+                        correlation_id: correlation_id.clone(),
+                        status_code: 400,
+                        headers: HashMap::new(),
+                        body: error_body(&format!("Invalid request body: {e}")),
+                    };
+                    (resp, None, Some(e.to_string()))
+                } else {
+                    let payload = decoded.unwrap();
+                    let envelope = crate::openhuman::agent::triage::TriggerEnvelope::from_webhook(
+                        &tunnel_uuid,
+                        &method,
+                        &path,
+                        payload,
+                    );
+                    // Spawn the triage pipeline so we don't block the
+                    // broadcast channel's dispatch task during LLM calls.
+                    let corr = correlation_id.clone();
+                    let skill = reg.agent_id.clone().or_else(|| Some(reg.skill_id.clone()));
+                    tokio::spawn(async move {
+                        let result =
+                            tokio::time::timeout(std::time::Duration::from_secs(60), async {
+                                run_agent_trigger(&envelope).await
+                            })
+                            .await;
+                        let (resp, err) = match result {
+                            Ok(Ok(output)) => (build_agent_response(&corr, 200, &output), None),
+                            Ok(Err(e)) => {
+                                tracing::error!("[webhook] agent trigger failed: {}", e);
+                                (
+                                    build_agent_response(&corr, 500, &format!("Agent error: {e}")),
+                                    Some(e),
+                                )
+                            }
+                            Err(_) => {
+                                tracing::error!("[webhook] agent trigger timed out (60s)");
+                                (
+                                    build_agent_response(&corr, 504, "Agent triage timed out"),
+                                    Some("timed out after 60s".to_string()),
+                                )
+                            }
+                        };
+                        // Emit response from the spawned task.
+                        if let Some(mgr) = global_socket_manager() {
+                            let response_data = serde_json::json!({
+                                "correlationId": resp.correlation_id,
+                                "statusCode": resp.status_code,
+                                "headers": resp.headers,
+                                "body": resp.body,
+                            });
+                            if let Err(e) = mgr.emit("webhook:response", response_data).await {
+                                tracing::error!("[webhook] failed to emit spawned response: {}", e);
+                            }
+                        }
+                        if let Some(e) = err {
+                            tracing::warn!("[webhook] agent trigger error: {}", e);
+                        }
+                    });
+                    // Return 202 Accepted immediately so the event handler
+                    // doesn't block the broadcast channel.
+                    let resp = WebhookResponseData {
+                        correlation_id: correlation_id.clone(),
+                        status_code: 202,
+                        headers: HashMap::new(),
+                        body: serde_json::json!({"status": "accepted", "message": "Agent triage started"}).to_string(),
+                    };
+                    let skill_id = reg.agent_id.clone().or_else(|| Some(reg.skill_id.clone()));
+                    (resp, skill_id, None)
+                }
+            }
+            Some(ref reg) => {
+                // skill target kind or any other unrecognised kind — skill runtime not available
+                tracing::debug!(
+                    "[webhook] skill tunnel {} (kind={}) — skill runtime not available",
+                    tunnel_uuid,
+                    reg.target_kind,
+                );
+                let resp = WebhookResponseData {
+                    correlation_id: correlation_id.clone(),
+                    status_code: 501,
+                    headers: HashMap::new(),
+                    body: error_body("Skill runtime not available for direct dispatch"),
+                };
+                (
+                    resp,
+                    Some(reg.skill_id.clone()),
+                    Some("skill runtime not available".to_string()),
+                )
+            }
+            None => {
+                tracing::debug!("[webhook] no registration for tunnel {}", tunnel_uuid);
+                let resp = WebhookResponseData {
+                    correlation_id: correlation_id.clone(),
+                    status_code: 404,
+                    headers: HashMap::new(),
+                    body: error_body("No tunnel registration found"),
+                };
+                (resp, None, Some("no tunnel registration".to_string()))
+            }
+        };
+
+        // Record request and response in the router debug logs.
+        if let Some(ref r) = router {
+            r.record_request(request, resolved_skill_id.clone());
+            r.record_response(
+                request,
+                &response,
+                resolved_skill_id.clone(),
+                response_error.clone(),
+            );
+        }
+
+        // Publish notification events.
        if let Some(ref sid) = resolved_skill_id {
            publish_global(DomainEvent::WebhookReceived {
                tunnel_id: tunnel_uuid.clone(),
@@ -109,7 +232,7 @@ impl EventHandler for WebhookRequestSubscriber {
            error: response_error.clone(),
        });

-        // Emit response back through the socket
+        // Emit response back through the socket.
        if let Some(mgr) = global_socket_manager() {
            let response_data = json!({
                "correlationId": response.correlation_id,
@@ -125,16 +248,71 @@ impl EventHandler for WebhookRequestSubscriber {
        }

        tracing::info!(
-            "[webhook] {} {} → status={}, skill={:?}, tunnel={}",
+            "[webhook] {} {} → status={}, skill={:?}, tunnel={} ({}ms)",
            method,
            path,
            response.status_code,
            resolved_skill_id,
            tunnel_name,
+            started_at.elapsed().as_millis(),
        );
    }
 }

+/// Decode a base64-encoded webhook request body into a JSON value.
+///
+/// Returns an empty object when the body is absent, empty, or not valid
+/// UTF-8 JSON. If the body is valid UTF-8 but not valid JSON, the raw
+/// text is wrapped under the `"raw"` key so callers still have access
+/// to the original content.
+fn decode_webhook_body(base64_body: &str) -> Result<serde_json::Value, String> {
+    if base64_body.is_empty() {
+        return Ok(serde_json::json!({}));
+    }
+    use base64::Engine;
+    let decoded = base64::engine::general_purpose::STANDARD
+        .decode(base64_body.as_bytes())
+        .map_err(|e| format!("invalid base64 body: {e}"))?;
+    let text = std::str::from_utf8(&decoded).map_err(|e| format!("invalid utf-8 body: {e}"))?;
+    Ok(serde_json::from_str(text).unwrap_or_else(|_| serde_json::json!({ "raw": text })))
+}
+
+/// Run the triage pipeline for a trigger envelope and return the
+/// human-readable decision summary on success.
+async fn run_agent_trigger(
+    envelope: &crate::openhuman::agent::triage::TriggerEnvelope,
+) -> Result<String, String> {
+    let run = crate::openhuman::agent::triage::run_triage(envelope)
+        .await
+        .map_err(|e| format!("triage evaluation failed: {e}"))?;
+
+    crate::openhuman::agent::triage::apply_decision(run.clone(), envelope)
+        .await
+        .map_err(|e| format!("apply_decision failed: {e}"))?;
+
+    Ok(format!(
+        "Triage decision: {} (agent: {:?})",
+        run.decision.action.as_str(),
+        run.decision.target_agent
+    ))
+}
+
+/// Build a base64-encoded JSON response body for an agent trigger result.
+fn build_agent_response(
+    correlation_id: &str,
+    status_code: u16,
+    body_text: &str,
+) -> WebhookResponseData {
+    let mut headers = HashMap::new();
+    headers.insert("content-type".to_string(), "application/json".to_string());
+    WebhookResponseData {
+        correlation_id: correlation_id.to_string(),
+        status_code,
+        headers,
+        body: base64_encode(&serde_json::json!({ "result": body_text }).to_string()),
+    }
+}
+
 #[cfg(test)]
 mod tests {
    use super::*;
@@ -202,16 +380,15 @@ mod tests {

    #[tokio::test]
    async fn handle_processes_incoming_webhook_without_socket_manager() {
-        // When the socket-manager singleton isn't initialised, the
-        // handler should log "no socket manager available" and return
-        // cleanly rather than panicking. We exercise the full routing
-        // path (currently the "skill runtime removed" branch) to lock
-        // in that contract for future refactors.
+        // When the socket-manager singleton isn't initialised, the router
+        // lookup returns None (no registration), so the handler takes the
+        // "no tunnel registration → 404" path and then logs "no socket
+        // manager available" before returning cleanly.
        let subscriber = WebhookRequestSubscriber::new();
        let request = WebhookRequest {
            correlation_id: "wh_test_1".into(),
            tunnel_id: "tid-1".into(),
-            tunnel_uuid: "uuid-1".into(),
+            tunnel_uuid: "uuid-unregistered".into(),
            tunnel_name: "my-hook".into(),
            method: "POST".into(),
            path: "/hook".into(),
@@ -223,6 +400,57 @@ mod tests {
            request,
            raw_data: serde_json::json!({}),
        };
+        // Must not panic — even without any singletons initialised.
        subscriber.handle(&event).await;
    }
+
+    // ── decode_webhook_body ───────────────────────────────────────
+
+    #[test]
+    fn decode_webhook_body_empty_returns_empty_object() {
+        let v = decode_webhook_body("").unwrap();
+        assert!(v.as_object().map(|o| o.is_empty()).unwrap_or(false));
+    }
+
+    #[test]
+    fn decode_webhook_body_parses_valid_json() {
+        use base64::Engine;
+        let encoded =
+            base64::engine::general_purpose::STANDARD.encode(r#"{"key":"value"}"#.as_bytes());
+        let v = decode_webhook_body(&encoded).unwrap();
+        assert_eq!(v["key"].as_str(), Some("value"));
+    }
+
+    #[test]
+    fn decode_webhook_body_wraps_non_json_in_raw_field() {
+        use base64::Engine;
+        let encoded = base64::engine::general_purpose::STANDARD.encode("plain text".as_bytes());
+        let v = decode_webhook_body(&encoded).unwrap();
+        assert_eq!(v["raw"].as_str(), Some("plain text"));
+    }
+
+    #[test]
+    fn decode_webhook_body_rejects_invalid_base64() {
+        let err = decode_webhook_body("not-valid-base64!!!").unwrap_err();
+        assert!(err.contains("invalid base64"));
+    }
+
+    // ── build_agent_response ──────────────────────────────────────
+
+    #[test]
+    fn build_agent_response_sets_status_and_body() {
+        let resp = build_agent_response("corr-1", 200, "Triage decision: drop");
+        assert_eq!(resp.correlation_id, "corr-1");
+        assert_eq!(resp.status_code, 200);
+        assert_eq!(
+            resp.headers.get("content-type").map(String::as_str),
+            Some("application/json")
+        );
+        // Body must be base64-encoded JSON with a "result" key.
+        let decoded = base64::engine::general_purpose::STANDARD
+            .decode(resp.body.as_bytes())
+            .expect("valid base64");
+        let v: serde_json::Value = serde_json::from_slice(&decoded).expect("valid json");
+        assert_eq!(v["result"].as_str(), Some("Triage decision: drop"));
+    }
 }
@@ -40,36 +40,71 @@ async fn get_authed_value(
        .map_err(|e| e.to_string())
 }

-pub async fn list_registrations() -> Result<RpcOutcome<WebhookDebugRegistrationsResult>, String> {
-    let registrations = Vec::new();
-    let count = 0usize;
+/// Retrieve the global webhook router, returning an error if the socket
+/// manager or router is not yet initialised.
+fn get_router() -> Result<std::sync::Arc<crate::openhuman::webhooks::WebhookRouter>, String> {
+    crate::openhuman::socket::global_socket_manager()
+        .ok_or_else(|| "socket manager not initialized".to_string())?
+        .webhook_router()
+        .ok_or_else(|| "webhook router not initialized".to_string())
+}

-    Ok(RpcOutcome::single_log(
-        WebhookDebugRegistrationsResult { registrations },
-        format!("webhooks.list_registrations returned {count} registration(s)"),
-    ))
+pub async fn list_registrations() -> Result<RpcOutcome<WebhookDebugRegistrationsResult>, String> {
+    match get_router() {
+        Ok(router) => {
+            let registrations = router.list_all();
+            let count = registrations.len();
+            Ok(RpcOutcome::single_log(
+                WebhookDebugRegistrationsResult { registrations },
+                format!("webhooks.list_registrations returned {count} registration(s)"),
+            ))
+        }
+        Err(_) => {
+            // Router not yet initialized — return empty list (not an error in RPC).
+            Ok(RpcOutcome::single_log(
+                WebhookDebugRegistrationsResult {
+                    registrations: Vec::new(),
+                },
+                "webhooks.list_registrations returned 0 registration(s) (router not initialized)"
+                    .to_string(),
+            ))
+        }
+    }
 }

 pub async fn list_logs(
    limit: Option<usize>,
 ) -> Result<RpcOutcome<WebhookDebugLogListResult>, String> {
-    let _ = limit;
-    let logs = Vec::new();
-    let count = 0usize;
-
-    Ok(RpcOutcome::single_log(
-        WebhookDebugLogListResult { logs },
-        format!("webhooks.list_logs returned {count} log entrie(s)"),
-    ))
+    match get_router() {
+        Ok(router) => {
+            let logs = router.list_logs(limit);
+            let count = logs.len();
+            Ok(RpcOutcome::single_log(
+                WebhookDebugLogListResult { logs },
+                format!("webhooks.list_logs returned {count} log entrie(s)"),
+            ))
+        }
+        Err(_) => Ok(RpcOutcome::single_log(
+            WebhookDebugLogListResult { logs: Vec::new() },
+            "webhooks.list_logs returned 0 log entrie(s) (router not initialized)".to_string(),
+        )),
+    }
 }

 pub async fn clear_logs() -> Result<RpcOutcome<WebhookDebugLogsClearedResult>, String> {
-    let cleared = 0usize;
-
-    Ok(RpcOutcome::single_log(
-        WebhookDebugLogsClearedResult { cleared },
-        format!("webhooks.clear_logs removed {cleared} log entrie(s)"),
-    ))
+    match get_router() {
+        Ok(router) => {
+            let cleared = router.clear_logs();
+            Ok(RpcOutcome::single_log(
+                WebhookDebugLogsClearedResult { cleared },
+                format!("webhooks.clear_logs removed {cleared} log entrie(s)"),
+            ))
+        }
+        Err(_) => Ok(RpcOutcome::single_log(
+            WebhookDebugLogsClearedResult { cleared: 0 },
+            "webhooks.clear_logs removed 0 log entrie(s) (router not initialized)".to_string(),
+        )),
+    }
 }

 pub async fn register_echo(
@@ -77,9 +112,9 @@ pub async fn register_echo(
    tunnel_name: Option<String>,
    backend_tunnel_id: Option<String>,
 ) -> Result<RpcOutcome<WebhookDebugRegistrationsResult>, String> {
-    let _ = (tunnel_name, backend_tunnel_id);
-    let registrations = Vec::new();
-
+    let router = get_router().map_err(|e| format!("webhooks.register_echo failed: {e}"))?;
+    router.register_echo(tunnel_uuid, tunnel_name, backend_tunnel_id)?;
+    let registrations = router.list_all();
    Ok(RpcOutcome::single_log(
        WebhookDebugRegistrationsResult { registrations },
        format!("webhooks.register_echo registered tunnel {tunnel_uuid}"),
@@ -89,14 +124,88 @@ pub async fn register_echo(
 pub async fn unregister_echo(
    tunnel_uuid: &str,
 ) -> Result<RpcOutcome<WebhookDebugRegistrationsResult>, String> {
-    let registrations = Vec::new();
-
+    let router = get_router().map_err(|e| format!("webhooks.unregister_echo failed: {e}"))?;
+    router.unregister(tunnel_uuid, "echo")?;
+    let registrations = router.list_all();
    Ok(RpcOutcome::single_log(
        WebhookDebugRegistrationsResult { registrations },
        format!("webhooks.unregister_echo removed tunnel {tunnel_uuid}"),
    ))
 }

+/// Register an agent-backed webhook tunnel.
+///
+/// Incoming requests on this tunnel will be routed to the triage
+/// pipeline instead of the (removed) skill runtime.
+pub async fn register_agent(
+    tunnel_uuid: &str,
+    agent_id: Option<String>,
+    tunnel_name: Option<String>,
+    backend_tunnel_id: Option<String>,
+) -> Result<RpcOutcome<WebhookDebugRegistrationsResult>, String> {
+    let router = get_router().map_err(|e| format!("webhooks.register_agent failed: {e}"))?;
+    router.register_agent(tunnel_uuid, agent_id, tunnel_name, backend_tunnel_id)?;
+    let registrations = router.list_all();
+    Ok(RpcOutcome::single_log(
+        WebhookDebugRegistrationsResult { registrations },
+        format!("webhooks.register_agent registered agent tunnel {tunnel_uuid}"),
+    ))
+}
+
+/// Trigger the triage/agent pipeline directly via RPC without requiring
+/// an incoming webhook request. Useful for testing and manual escalation.
+pub async fn trigger_agent(
+    source: &str,
+    caller_id: &str,
+    reason: &str,
+    payload: serde_json::Value,
+) -> Result<RpcOutcome<serde_json::Value>, String> {
+    use crate::openhuman::agent::triage::TriggerEnvelope;
+
+    let envelope = match source {
+        "webhook" => TriggerEnvelope::from_webhook(caller_id, "POST", "/trigger", payload),
+        "cron" => {
+            let output = payload
+                .get("output")
+                .and_then(serde_json::Value::as_str)
+                .unwrap_or(reason);
+            TriggerEnvelope::from_cron(caller_id, reason, output)
+        }
+        "external" => TriggerEnvelope::from_external(caller_id, reason, payload),
+        other => {
+            return Err(format!(
+                "unsupported trigger source `{other}` — supported: webhook, cron, external"
+            ))
+        }
+    };
+
+    let run = tokio::time::timeout(
+        std::time::Duration::from_secs(60),
+        crate::openhuman::agent::triage::run_triage(&envelope),
+    )
+    .await
+    .map_err(|_| "triage timed out after 60s".to_string())?
+    .map_err(|e| format!("triage failed: {e}"))?;
+
+    tokio::time::timeout(
+        std::time::Duration::from_secs(60),
+        crate::openhuman::agent::triage::apply_decision(run.clone(), &envelope),
+    )
+    .await
+    .map_err(|_| "apply_decision timed out after 60s".to_string())?
+    .map_err(|e| format!("apply_decision failed: {e}"))?;
+
+    Ok(RpcOutcome::single_log(
+        serde_json::json!({
+            "decision": run.decision.action.as_str(),
+            "target_agent": run.decision.target_agent,
+            "prompt": run.decision.prompt,
+            "reason": run.decision.reason,
+        }),
+        format!("webhooks.trigger_agent completed for {source}/{caller_id}"),
+    ))
+}
+
 pub fn build_echo_response(request: &WebhookRequest) -> WebhookResponseData {
    let response_body = serde_json::json!({
        "ok": true,
@@ -312,50 +421,58 @@ mod tests {
        assert!(err.contains("no backend session token"));
    }

-    // ── Stub ops (list/clear/register/unregister) ────────────────
+    // ── Router-not-initialized fallback paths ─────────────────────
+    // These tests run without a global SocketManager so the router
+    // accessor returns an error and the ops fall back gracefully.

    #[tokio::test]
-    async fn list_registrations_returns_empty_payload_with_count_log() {
+    async fn list_registrations_returns_empty_when_router_not_initialized() {
+        // No global socket manager → graceful empty response.
        let out = list_registrations().await.unwrap();
        assert!(out.value.registrations.is_empty());
        assert!(out.logs.iter().any(|l| l.contains("returned 0")));
    }

    #[tokio::test]
-    async fn list_logs_returns_empty_and_ignores_limit() {
+    async fn list_logs_returns_empty_when_router_not_initialized() {
        let out = list_logs(Some(50)).await.unwrap();
        assert!(out.value.logs.is_empty());
        assert!(out.logs.iter().any(|l| l.contains("returned 0")));

-        // `None` limit path.
        let out2 = list_logs(None).await.unwrap();
        assert!(out2.value.logs.is_empty());
    }

    #[tokio::test]
-    async fn clear_logs_reports_zero_cleared() {
+    async fn clear_logs_reports_zero_when_router_not_initialized() {
        let out = clear_logs().await.unwrap();
        assert_eq!(out.value.cleared, 0);
        assert!(out.logs.iter().any(|l| l.contains("removed 0")));
    }

    #[tokio::test]
-    async fn register_echo_surfaces_tunnel_uuid_in_log() {
-        let out = register_echo("uuid-1", Some("name".into()), Some("btid-1".into()))
+    async fn register_echo_errors_when_router_not_initialized() {
+        // Without the router, register_echo must return an Err.
+        let err = register_echo("uuid-1", Some("name".into()), Some("btid-1".into()))
            .await
-            .unwrap();
-        assert!(out.value.registrations.is_empty());
-        assert!(out
-            .logs
-            .iter()
-            .any(|l| l.contains("registered tunnel uuid-1")));
+            .unwrap_err();
+        assert!(
+            err.contains("socket manager not initialized")
+                || err.contains("webhook router not initialized")
+                || err.contains("register_echo failed"),
+            "unexpected error: {err}"
+        );
    }

    #[tokio::test]
-    async fn unregister_echo_surfaces_tunnel_uuid_in_log() {
-        let out = unregister_echo("uuid-1").await.unwrap();
-        assert!(out.value.registrations.is_empty());
-        assert!(out.logs.iter().any(|l| l.contains("removed tunnel uuid-1")));
+    async fn unregister_echo_errors_when_router_not_initialized() {
+        let err = unregister_echo("uuid-1").await.unwrap_err();
+        assert!(
+            err.contains("socket manager not initialized")
+                || err.contains("webhook router not initialized")
+                || err.contains("unregister_echo failed"),
+            "unexpected error: {err}"
+        );
    }

    // ── build_echo_response ───────────────────────────────────────
@@ -104,6 +104,7 @@ impl WebhookRouter {
            skill_id,
            tunnel_name,
            backend_tunnel_id,
+            None,
        )
    }

@@ -114,7 +115,38 @@ impl WebhookRouter {
        tunnel_name: Option<String>,
        backend_tunnel_id: Option<String>,
    ) -> Result<(), String> {
-        self.register_target(tunnel_uuid, "echo", "echo", tunnel_name, backend_tunnel_id)
+        self.register_target(
+            tunnel_uuid,
+            "echo",
+            "echo",
+            tunnel_name,
+            backend_tunnel_id,
+            None,
+        )
+    }
+
+    /// Register an agent-backed webhook tunnel.
+    ///
+    /// Requests arriving on this tunnel are routed into the triage
+    /// pipeline rather than the skill runtime. `agent_id` is stored
+    /// for observability and rebind validation; the triage evaluator
+    /// currently selects the target agent dynamically regardless of
+    /// this value.
+    pub fn register_agent(
+        &self,
+        tunnel_uuid: &str,
+        agent_id: Option<String>,
+        tunnel_name: Option<String>,
+        backend_tunnel_id: Option<String>,
+    ) -> Result<(), String> {
+        self.register_target(
+            tunnel_uuid,
+            "agent",
+            "agent",
+            tunnel_name,
+            backend_tunnel_id,
+            agent_id,
+        )
    }

    fn register_target(
@@ -124,6 +156,7 @@ impl WebhookRouter {
        skill_id: &str,
        tunnel_name: Option<String>,
        backend_tunnel_id: Option<String>,
+        agent_id: Option<String>,
    ) -> Result<(), String> {
        let mut routes = self.routes.write().map_err(|e| e.to_string())?;

@@ -134,11 +167,24 @@ impl WebhookRouter {
                    tunnel_uuid, existing.target_kind, existing.skill_id, target_kind, skill_id
                ));
            }
+            // Prevent silent agent_id rebinding on agent tunnels.
+            if target_kind == "agent" && existing.agent_id.as_deref() != agent_id.as_deref() {
+                tracing::warn!(
+                    tunnel = %tunnel_uuid,
+                    existing_agent = ?existing.agent_id,
+                    requested_agent = ?agent_id,
+                    "[webhooks] rejecting agent tunnel rebind"
+                );
+                return Err(format!(
+                    "Tunnel {} is already bound to agent {:?}; cannot rebind to {:?}",
+                    tunnel_uuid, existing.agent_id, agent_id
+                ));
+            }
        }

        debug!(
-            "[webhooks] Registering tunnel {} → {} '{}'",
-            tunnel_uuid, target_kind, skill_id
+            "[webhooks] Registering tunnel {} → {} '{}' (agent={:?})",
+            tunnel_uuid, target_kind, skill_id, agent_id,
        );

        let tunnel_name_clone = tunnel_name.clone();
@@ -150,6 +196,7 @@ impl WebhookRouter {
                skill_id: skill_id.to_string(),
                tunnel_name,
                backend_tunnel_id,
+                agent_id,
            },
        );

@@ -842,4 +889,60 @@ mod tests {
        let logs = router.list_logs(Some(MAX_DEBUG_LOG_ENTRIES + 100));
        assert!(logs.len() <= MAX_DEBUG_LOG_ENTRIES);
    }
+
+    #[test]
+    fn register_agent_persists_agent_id_and_name() {
+        let router = WebhookRouter::new(None);
+        router
+            .register_agent(
+                "uuid-a1",
+                Some("agent-42".into()),
+                Some("My Agent".into()),
+                None,
+            )
+            .unwrap();
+
+        let reg = router.registration("uuid-a1").unwrap();
+        assert_eq!(reg.target_kind, "agent");
+        assert_eq!(reg.agent_id.as_deref(), Some("agent-42"));
+        assert_eq!(reg.tunnel_name.as_deref(), Some("My Agent"));
+    }
+
+    #[test]
+    fn register_agent_same_id_succeeds() {
+        let router = WebhookRouter::new(None);
+        router
+            .register_agent("uuid-a2", Some("agent-1".into()), None, None)
+            .unwrap();
+        // Re-register with the same agent_id should succeed.
+        router
+            .register_agent(
+                "uuid-a2",
+                Some("agent-1".into()),
+                Some("Updated".into()),
+                None,
+            )
+            .unwrap();
+
+        let reg = router.registration("uuid-a2").unwrap();
+        assert_eq!(reg.agent_id.as_deref(), Some("agent-1"));
+        assert_eq!(reg.tunnel_name.as_deref(), Some("Updated"));
+    }
+
+    #[test]
+    fn register_agent_rejects_different_agent_id() {
+        let router = WebhookRouter::new(None);
+        router
+            .register_agent("uuid-a3", Some("agent-A".into()), None, None)
+            .unwrap();
+
+        let err = router
+            .register_agent("uuid-a3", Some("agent-B".into()), None, None)
+            .unwrap_err();
+        assert!(err.contains("already bound"));
+
+        // Original agent_id is preserved.
+        let reg = router.registration("uuid-a3").unwrap();
+        assert_eq!(reg.agent_id.as_deref(), Some("agent-A"));
+    }
 }
@@ -23,6 +23,26 @@ struct WebhookUnregisterEchoParams {
    tunnel_uuid: String,
 }

+#[derive(Debug, Deserialize)]
+struct WebhookRegisterAgentParams {
+    tunnel_uuid: String,
+    agent_id: Option<String>,
+    tunnel_name: Option<String>,
+    backend_tunnel_id: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+struct WebhookTriggerAgentParams {
+    /// Trigger source slug: `"webhook"`, `"cron"`, or `"external"`.
+    source: Option<String>,
+    /// Stable identifier for the caller (tunnel UUID, job ID, etc.).
+    caller_id: String,
+    /// Human-readable reason / label for the trigger.
+    reason: Option<String>,
+    /// Trigger payload forwarded to the triage pipeline.
+    payload: Option<Value>,
+}
+
 #[derive(Debug, Deserialize)]
 struct WebhookCreateTunnelParams {
    name: String,
@@ -50,6 +70,8 @@ pub fn all_controller_schemas() -> Vec<ControllerSchema> {
        schemas("clear_logs"),
        schemas("register_echo"),
        schemas("unregister_echo"),
+        schemas("register_agent"),
+        schemas("trigger_agent"),
        schemas("list_tunnels"),
        schemas("create_tunnel"),
        schemas("get_tunnel"),
@@ -81,6 +103,14 @@ pub fn all_registered_controllers() -> Vec<RegisteredController> {
            schema: schemas("unregister_echo"),
            handler: handle_unregister_echo,
        },
+        RegisteredController {
+            schema: schemas("register_agent"),
+            handler: handle_register_agent,
+        },
+        RegisteredController {
+            schema: schemas("trigger_agent"),
+            handler: handle_trigger_agent,
+        },
        RegisteredController {
            schema: schemas("list_tunnels"),
            handler: handle_list_tunnels,
@@ -175,6 +205,73 @@ pub fn schemas(function: &str) -> ControllerSchema {
            }],
            outputs: vec![json_output("result", "Updated webhook registrations.")],
        },
+        "register_agent" => ControllerSchema {
+            namespace: "webhooks",
+            function: "register_agent",
+            description:
+                "Register an agent-backed webhook tunnel. Incoming requests on this tunnel \
+                 are routed to the triage pipeline instead of the skill runtime.",
+            inputs: vec![
+                FieldSchema {
+                    name: "tunnel_uuid",
+                    ty: TypeSchema::String,
+                    comment: "Tunnel UUID from the backend.",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "agent_id",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::String)),
+                    comment: "Optional agent definition id to pin for this tunnel.",
+                    required: false,
+                },
+                FieldSchema {
+                    name: "tunnel_name",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::String)),
+                    comment: "Optional human-readable tunnel name.",
+                    required: false,
+                },
+                FieldSchema {
+                    name: "backend_tunnel_id",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::String)),
+                    comment: "Optional backend tunnel id.",
+                    required: false,
+                },
+            ],
+            outputs: vec![json_output("result", "Updated webhook registrations.")],
+        },
+        "trigger_agent" => ControllerSchema {
+            namespace: "webhooks",
+            function: "trigger_agent",
+            description: "Trigger the triage/agent pipeline directly via RPC without requiring an \
+                 incoming webhook request. Useful for testing and manual escalation.",
+            inputs: vec![
+                FieldSchema {
+                    name: "caller_id",
+                    ty: TypeSchema::String,
+                    comment: "Stable identifier for the caller (tunnel UUID, job ID, etc.).",
+                    required: true,
+                },
+                FieldSchema {
+                    name: "source",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::String)),
+                    comment: "Trigger source slug: 'webhook', 'cron', or 'external' (default).",
+                    required: false,
+                },
+                FieldSchema {
+                    name: "reason",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::String)),
+                    comment: "Human-readable reason or label for the trigger.",
+                    required: false,
+                },
+                FieldSchema {
+                    name: "payload",
+                    ty: TypeSchema::Option(Box::new(TypeSchema::Json)),
+                    comment: "Optional trigger payload forwarded to the triage pipeline.",
+                    required: false,
+                },
+            ],
+            outputs: vec![json_output("result", "Triage decision result.")],
+        },
        "list_tunnels" => ControllerSchema {
            namespace: "webhooks",
            function: "list_tunnels",
@@ -316,6 +413,39 @@ fn handle_unregister_echo(params: Map<String, Value>) -> ControllerFuture {
    })
 }

+fn handle_register_agent(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let payload = deserialize_params::<WebhookRegisterAgentParams>(params)?;
+        to_json(
+            crate::openhuman::webhooks::ops::register_agent(
+                &payload.tunnel_uuid,
+                payload.agent_id,
+                payload.tunnel_name,
+                payload.backend_tunnel_id,
+            )
+            .await?,
+        )
+    })
+}
+
+fn handle_trigger_agent(params: Map<String, Value>) -> ControllerFuture {
+    Box::pin(async move {
+        let payload = deserialize_params::<WebhookTriggerAgentParams>(params)?;
+        let source = payload.source.as_deref().unwrap_or("external");
+        let reason = payload.reason.as_deref().unwrap_or("rpc_trigger");
+        let trigger_payload = payload.payload.unwrap_or_else(|| serde_json::json!({}));
+        to_json(
+            crate::openhuman::webhooks::ops::trigger_agent(
+                source,
+                &payload.caller_id,
+                reason,
+                trigger_payload,
+            )
+            .await?,
+        )
+    })
+}
+
 fn handle_list_tunnels(_params: Map<String, Value>) -> ControllerFuture {
    Box::pin(async move {
        let config = crate::openhuman::config::rpc::load_config_with_timeout().await?;
@@ -413,6 +543,8 @@ mod tests {
        "clear_logs",
        "register_echo",
        "unregister_echo",
+        "register_agent",
+        "trigger_agent",
        "list_tunnels",
        "create_tunnel",
        "get_tunnel",
@@ -543,6 +675,30 @@ mod tests {
        assert_eq!(required_input_names(&s), vec!["tunnel_uuid"]);
    }

+    #[test]
+    fn register_agent_requires_tunnel_uuid_and_has_optional_fields() {
+        let s = schemas("register_agent");
+        assert_eq!(required_input_names(&s), vec!["tunnel_uuid"]);
+        for optional in ["agent_id", "tunnel_name", "backend_tunnel_id"] {
+            assert!(
+                s.inputs.iter().any(|f| f.name == optional && !f.required),
+                "`register_agent` must accept optional `{optional}`"
+            );
+        }
+    }
+
+    #[test]
+    fn trigger_agent_requires_caller_id_only() {
+        let s = schemas("trigger_agent");
+        assert_eq!(required_input_names(&s), vec!["caller_id"]);
+        for optional in ["source", "reason", "payload"] {
+            assert!(
+                s.inputs.iter().any(|f| f.name == optional && !f.required),
+                "`trigger_agent` must accept optional `{optional}`"
+            );
+        }
+    }
+
    #[test]
    fn list_tunnels_has_no_inputs() {
        assert!(schemas("list_tunnels").inputs.is_empty());
@@ -64,6 +64,11 @@ pub struct TunnelRegistration {
    /// Backend MongoDB `_id` for CRUD operations.
    #[serde(default)]
    pub backend_tunnel_id: Option<String>,
+    /// Optional agent ID for agent-type tunnels. Set when
+    /// `target_kind == "agent"` to identify which agent definition
+    /// should handle incoming requests on this tunnel.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub agent_id: Option<String>,
 }

 fn default_webhook_target_kind() -> String {
@@ -356,6 +361,7 @@ mod tests {
                skill_id: "s".into(),
                tunnel_name: None,
                backend_tunnel_id: None,
+                agent_id: None,
            }],
        };
        let back: WebhookDebugRegistrationsResult =