diff --git a/.claude/memory.md b/.claude/memory.md index 2afa89e0a..8072887e4 100644 --- a/.claude/memory.md +++ b/.claude/memory.md @@ -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`. diff --git a/app/src/features/webhooks/types.ts b/app/src/features/webhooks/types.ts index 612edae7b..92b8cf410 100644 --- a/app/src/features/webhooks/types.ts +++ b/app/src/features/webhooks/types.ts @@ -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 { diff --git a/src/core/event_bus/bus.rs b/src/core/event_bus/bus.rs index 508ece046..bacc2aad0 100644 --- a/src/core/event_bus/bus.rs +++ b/src/core/event_bus/bus.rs @@ -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(), }); diff --git a/src/core/event_bus/events.rs b/src/core/event_bus/events.rs index 14d5d76b3..fc6aa6fa6 100644 --- a/src/core/event_bus/events.rs +++ b/src/core/event_bus/events.rs @@ -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", ), diff --git a/src/openhuman/agent/schemas.rs b/src/openhuman/agent/schemas.rs index ecc70e35a..f88233238 100644 --- a/src/openhuman/agent/schemas.rs +++ b/src/openhuman/agent/schemas.rs @@ -251,11 +251,18 @@ fn handle_triage_evaluate(params: Map) -> ControllerFuture { Box::pin(async move { let p = deserialize_params::(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) -> 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!({})), ])) diff --git a/src/openhuman/agent/triage/envelope.rs b/src/openhuman/agent/triage/envelope.rs index 5754e3ae5..6eb16f6b6 100644 --- a/src/openhuman/agent/triage/envelope.rs +++ b/src/openhuman/agent/triage/envelope.rs @@ -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"); + } } diff --git a/src/openhuman/channels/proactive.rs b/src/openhuman/channels/proactive.rs index b6c2872d3..e546f1b94 100644 --- a/src/openhuman/channels/proactive.rs +++ b/src/openhuman/channels/proactive.rs @@ -309,6 +309,7 @@ mod tests { sub.handle(&DomainEvent::CronJobTriggered { job_id: "j".into(), + job_name: "test-job".into(), job_type: "agent".into(), }) .await; diff --git a/src/openhuman/composio/bus.rs b/src/openhuman/composio/bus.rs index 1603de6bb..22970d36d 100644 --- a/src/openhuman/composio/bus.rs +++ b/src/openhuman/composio/bus.rs @@ -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; diff --git a/src/openhuman/cron/bus.rs b/src/openhuman/cron/bus.rs index cf0a170fa..4a525038f 100644 --- a/src/openhuman/cron/bus.rs +++ b/src/openhuman/cron/bus.rs @@ -143,6 +143,7 @@ mod tests { sub.handle(&DomainEvent::CronJobTriggered { job_id: "j1".into(), + job_name: "test-job".into(), job_type: "shell".into(), }) .await; diff --git a/src/openhuman/cron/mod.rs b/src/openhuman/cron/mod.rs index e6c1e59e3..3145cc398 100644 --- a/src/openhuman/cron/mod.rs +++ b/src/openhuman/cron/mod.rs @@ -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, diff --git a/src/openhuman/cron/ops.rs b/src/openhuman/cron/ops.rs index 090aeee21..1670a9e92 100644 --- a/src/openhuman/cron/ops.rs +++ b/src/openhuman/cron/ops.rs @@ -8,7 +8,7 @@ use anyhow::Result; use serde_json::json; pub fn add_once(config: &Config, delay: &str, command: &str) -> Result { - 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 { +/// 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 { 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 ──────────────────────────────────────────────────── diff --git a/src/openhuman/cron/scheduler.rs b/src/openhuman/cron/scheduler.rs index e2b528134..9fb1f3614 100644 --- a/src/openhuman/cron/scheduler.rs +++ b/src/openhuman/cron/scheduler.rs @@ -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, ¬ification) { + 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 diff --git a/src/openhuman/notifications/bus.rs b/src/openhuman/notifications/bus.rs index 64767b4a9..9a031fe55 100644 --- a/src/openhuman/notifications/bus.rs +++ b/src/openhuman/notifications/bus.rs @@ -62,7 +62,9 @@ fn now_ms() -> u64 { pub fn event_to_notification(event: &DomainEvent) -> Option { 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"); diff --git a/src/openhuman/socket/manager.rs b/src/openhuman/socket/manager.rs index 7ca9c4ecf..0959bfe41 100644 --- a/src/openhuman/socket/manager.rs +++ b/src/openhuman/socket/manager.rs @@ -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> { + self.shared.webhook_router.read().clone() + } + /// Get the current socket state (status, ID, error). pub fn get_state(&self) -> SocketState { SocketState { diff --git a/src/openhuman/tools/impl/cron/add.rs b/src/openhuman/tools/impl/cron/add.rs index 0eed8c18b..e275d2ce0 100644 --- a/src/openhuman/tools/impl/cron/add.rs +++ b/src/openhuman/tools/impl/cron/add.rs @@ -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(); diff --git a/src/openhuman/tools/impl/system/schedule.rs b/src/openhuman/tools/impl/system/schedule.rs index 4eced4d5d..4203b7f6b 100644 --- a/src/openhuman/tools/impl/system/schedule.rs +++ b/src/openhuman/tools/impl/system/schedule.rs @@ -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 = 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::() + .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::*; diff --git a/src/openhuman/tree_summarizer/bus.rs b/src/openhuman/tree_summarizer/bus.rs index 5ce7f885f..6efacd97b 100644 --- a/src/openhuman/tree_summarizer/bus.rs +++ b/src/openhuman/tree_summarizer/bus.rs @@ -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; diff --git a/src/openhuman/webhooks/bus.rs b/src/openhuman/webhooks/bus.rs index 409c69bf1..cf0134e04 100644 --- a/src/openhuman/webhooks/bus.rs +++ b/src/openhuman/webhooks/bus.rs @@ -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 = 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 { + 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 { + 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")); + } } diff --git a/src/openhuman/webhooks/ops.rs b/src/openhuman/webhooks/ops.rs index 7e73e6618..918aedfbc 100644 --- a/src/openhuman/webhooks/ops.rs +++ b/src/openhuman/webhooks/ops.rs @@ -40,36 +40,71 @@ async fn get_authed_value( .map_err(|e| e.to_string()) } -pub async fn list_registrations() -> Result, 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, 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, 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, ) -> Result, 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, 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, backend_tunnel_id: Option, ) -> Result, 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, 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, + tunnel_name: Option, + backend_tunnel_id: Option, +) -> Result, 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, 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 ─────────────────────────────────────── diff --git a/src/openhuman/webhooks/router.rs b/src/openhuman/webhooks/router.rs index af0b485ff..ffaffb8fa 100644 --- a/src/openhuman/webhooks/router.rs +++ b/src/openhuman/webhooks/router.rs @@ -104,6 +104,7 @@ impl WebhookRouter { skill_id, tunnel_name, backend_tunnel_id, + None, ) } @@ -114,7 +115,38 @@ impl WebhookRouter { tunnel_name: Option, backend_tunnel_id: Option, ) -> 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, + tunnel_name: Option, + backend_tunnel_id: Option, + ) -> 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, backend_tunnel_id: Option, + agent_id: Option, ) -> 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")); + } } diff --git a/src/openhuman/webhooks/schemas.rs b/src/openhuman/webhooks/schemas.rs index af2c3b615..3db0307c5 100644 --- a/src/openhuman/webhooks/schemas.rs +++ b/src/openhuman/webhooks/schemas.rs @@ -23,6 +23,26 @@ struct WebhookUnregisterEchoParams { tunnel_uuid: String, } +#[derive(Debug, Deserialize)] +struct WebhookRegisterAgentParams { + tunnel_uuid: String, + agent_id: Option, + tunnel_name: Option, + backend_tunnel_id: Option, +} + +#[derive(Debug, Deserialize)] +struct WebhookTriggerAgentParams { + /// Trigger source slug: `"webhook"`, `"cron"`, or `"external"`. + source: Option, + /// Stable identifier for the caller (tunnel UUID, job ID, etc.). + caller_id: String, + /// Human-readable reason / label for the trigger. + reason: Option, + /// Trigger payload forwarded to the triage pipeline. + payload: Option, +} + #[derive(Debug, Deserialize)] struct WebhookCreateTunnelParams { name: String, @@ -50,6 +70,8 @@ pub fn all_controller_schemas() -> Vec { 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 { 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) -> ControllerFuture { }) } +fn handle_register_agent(params: Map) -> ControllerFuture { + Box::pin(async move { + let payload = deserialize_params::(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) -> ControllerFuture { + Box::pin(async move { + let payload = deserialize_params::(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) -> 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()); diff --git a/src/openhuman/webhooks/types.rs b/src/openhuman/webhooks/types.rs index 66e163c27..9fd094f2b 100644 --- a/src/openhuman/webhooks/types.rs +++ b/src/openhuman/webhooks/types.rs @@ -64,6 +64,11 @@ pub struct TunnelRegistration { /// Backend MongoDB `_id` for CRUD operations. #[serde(default)] pub backend_tunnel_id: Option, + /// 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, } 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 =