From e6ca837c1ab48b1df45405317cabf6d19ee04421 Mon Sep 17 00:00:00 2001 From: sanil-23 Date: Wed, 17 Jun 2026 15:29:32 +0530 Subject: [PATCH] feat(skills): catalog-driven run_skill flow + mid-session skill refresh (#3722) Co-authored-by: Claude --- src/core/event_bus/events.rs | 9 +- src/core/event_bus/events_tests.rs | 9 + .../agent/harness/session/builder/setters.rs | 3 + src/openhuman/agent/harness/session/tests.rs | 120 +++ .../agent/harness/session/turn/core.rs | 79 +- .../agent/harness/session/turn/mod.rs | 18 + .../agent/harness/session/turn/tools.rs | 156 ++++ .../agent/harness/session/turn_tests.rs | 19 + src/openhuman/agent/harness/session/types.rs | 17 + .../agents/orchestrator/prompt.md | 2 +- .../agents/orchestrator/prompt.rs | 41 +- .../agent/skill_executor/prompt.md | 28 +- src/openhuman/workflows/README.md | 7 +- src/openhuman/workflows/inject.rs | 830 ------------------ src/openhuman/workflows/mod.rs | 3 +- src/openhuman/workflows/ops_create.rs | 54 ++ src/openhuman/workflows/ops_install.rs | 16 + 17 files changed, 530 insertions(+), 881 deletions(-) delete mode 100644 src/openhuman/workflows/inject.rs diff --git a/src/core/event_bus/events.rs b/src/core/event_bus/events.rs index b151f719a..1f1c64fcc 100644 --- a/src/core/event_bus/events.rs +++ b/src/core/event_bus/events.rs @@ -388,6 +388,11 @@ pub enum DomainEvent { success: bool, elapsed_ms: u64, }, + /// The set of installed skills/workflows changed (install / uninstall / + /// create). Lets a live agent session refresh its `## Installed Skills` + /// catalogue mid-conversation instead of waiting for a restart. `reason` + /// is a short tag for logs (e.g. `"install"`, `"uninstall"`, `"create"`). + WorkflowsChanged { reason: String }, // ── Tools ─────────────────────────────────────────────────────────── /// A tool execution started. @@ -1110,7 +1115,8 @@ impl DomainEvent { Self::WorkflowLoaded { .. } | Self::WorkflowStopped { .. } | Self::WorkflowStartFailed { .. } - | Self::WorkflowExecuted { .. } => "workflow", + | Self::WorkflowExecuted { .. } + | Self::WorkflowsChanged { .. } => "workflow", Self::ToolExecutionStarted { .. } | Self::ToolExecutionCompleted { .. } => "tool", @@ -1245,6 +1251,7 @@ impl DomainEvent { Self::WorkflowStopped { .. } => "WorkflowStopped", Self::WorkflowStartFailed { .. } => "WorkflowStartFailed", Self::WorkflowExecuted { .. } => "WorkflowExecuted", + Self::WorkflowsChanged { .. } => "WorkflowsChanged", Self::ToolExecutionStarted { .. } => "ToolExecutionStarted", Self::ToolExecutionCompleted { .. } => "ToolExecutionCompleted", Self::WebhookIncomingRequest { .. } => "WebhookIncomingRequest", diff --git a/src/core/event_bus/events_tests.rs b/src/core/event_bus/events_tests.rs index 218610c99..631ba8ca9 100644 --- a/src/core/event_bus/events_tests.rs +++ b/src/core/event_bus/events_tests.rs @@ -532,3 +532,12 @@ fn approval_requested_does_not_surface_session_id() { "ApprovalRequested Debug must not surface session_id: {dbg}" ); } + +#[test] +fn workflows_changed_domain_and_name() { + let event = DomainEvent::WorkflowsChanged { + reason: "install".into(), + }; + assert_eq!(event.domain(), "workflow"); + assert_eq!(event.variant_name(), "WorkflowsChanged"); +} diff --git a/src/openhuman/agent/harness/session/builder/setters.rs b/src/openhuman/agent/harness/session/builder/setters.rs index 3867da699..a89f19c9d 100644 --- a/src/openhuman/agent/harness/session/builder/setters.rs +++ b/src/openhuman/agent/harness/session/builder/setters.rs @@ -582,10 +582,13 @@ impl AgentBuilder { }), last_seen_integrations_hash: 0, composio_integrations_rx: None, + skill_events_rx: None, announced_integrations: std::collections::HashSet::new(), pending_integration_announcement: Vec::new(), announced_mcp_servers: std::collections::HashSet::new(), pending_mcp_announcement: Vec::new(), + announced_skills: std::collections::HashSet::new(), + pending_skill_announcement: Vec::new(), archivist_hook: self.archivist_hook, synthesized_tool_names: std::collections::HashSet::new(), pending_synthesized_tools_mask: std::collections::HashSet::new(), diff --git a/src/openhuman/agent/harness/session/tests.rs b/src/openhuman/agent/harness/session/tests.rs index c4a69e6af..2076a89e7 100644 --- a/src/openhuman/agent/harness/session/tests.rs +++ b/src/openhuman/agent/harness/session/tests.rs @@ -413,6 +413,126 @@ fn composio_listener_drains_integrations_changed_events() { ); } +#[test] +fn skill_listener_drains_workflows_changed_events() { + let _ = init_global(64); + let mut agent = build_minimal_agent_with_definition_name(Some("orchestrator")); + agent.ensure_skill_events_listener(); + publish_global(DomainEvent::WorkflowsChanged { + reason: "install".into(), + }); + assert!( + agent.drain_skill_events(), + "a WorkflowsChanged event should be observed" + ); + assert!( + !agent.drain_skill_events(), + "event queue should be drained after one pass" + ); +} + +#[test] +fn skill_listener_treats_lag_as_signal() { + let _ = init_global(64); + let mut agent = build_minimal_agent_with_definition_name(Some("orchestrator")); + agent.ensure_skill_events_listener(); + // Flood well past the 64-slot bounded bus so the receiver lags. The + // `Lagged` arm must still report a signal (returns true) so a refresh + // isn't silently dropped under load. + for _ in 0..256 { + publish_global(DomainEvent::WorkflowsChanged { + reason: "install".into(), + }); + } + assert!( + agent.drain_skill_events(), + "a lagged listener must be treated as a signal" + ); +} + +#[test] +fn skill_listener_closed_channel_nulls_rx_and_is_not_a_signal() { + let mut agent = build_minimal_agent_with_definition_name(Some("orchestrator")); + // A receiver whose sender has been dropped → `try_recv` yields `Closed`. + let (tx, rx) = tokio::sync::broadcast::channel::(4); + drop(tx); + agent.set_skill_events_rx_for_test(rx); + assert!( + !agent.drain_skill_events(), + "a closed channel is not a signal" + ); + assert!( + !agent.has_skill_events_rx(), + "a closed receiver should be dropped so the next drain re-arms" + ); +} + +#[test] +fn refresh_workflows_picks_up_skill_installed_on_disk() { + use crate::openhuman::workflows::ops_types::{SKILL_MD, TRUST_MARKER}; + + // Isolated, trusted workspace with one project-scope skill on disk. + let ws = tempfile::TempDir::new().expect("temp workspace"); + let wsp = ws.path().to_path_buf(); + std::fs::create_dir_all(wsp.join(".openhuman")).unwrap(); + std::fs::write(wsp.join(".openhuman").join(TRUST_MARKER), "").unwrap(); + let skill_dir = wsp + .join(".openhuman") + .join("skills") + .join("zz-refresh-test"); + std::fs::create_dir_all(&skill_dir).unwrap(); + std::fs::write( + skill_dir.join(SKILL_MD), + "---\nname: zz-refresh-test\ndescription: a refresh test skill\n---\n# body\n", + ) + .unwrap(); + + let memory_cfg = crate::openhuman::config::MemoryConfig { + backend: "none".into(), + ..crate::openhuman::config::MemoryConfig::default() + }; + let mem: Arc = + Arc::from(crate::openhuman::memory_store::create_memory(&memory_cfg, &wsp).unwrap()); + let provider = Box::new(MockProvider { + responses: Mutex::new(vec![]), + }); + let mut agent = Agent::builder() + .provider(provider) + .tools(vec![Box::new(MockTool)]) + .memory(mem) + .tool_dispatcher(Box::new(NativeToolDispatcher)) + .workspace_dir(wsp.clone()) + .build() + .expect("agent build should succeed"); + + // Starts with no skills; refresh discovers the on-disk one and parks it + // for announcement. + assert!(agent.test_workflow_ids().is_empty()); + assert!( + agent.refresh_workflows("test"), + "installing a skill on disk should change the set" + ); + assert!( + agent + .test_workflow_ids() + .iter() + .any(|id| id == "zz-refresh-test"), + "the new skill should be discoverable" + ); + assert!( + agent + .test_pending_skill_announcement() + .iter() + .any(|id| id == "zz-refresh-test"), + "the new skill should be parked for announcement" + ); + // Idempotent: no new install -> no change. + assert!( + !agent.refresh_workflows("test"), + "no install since last refresh -> no change" + ); +} + #[tokio::test] async fn turn_without_tools_returns_text() { let workspace = tempfile::TempDir::new().expect("temp workspace"); diff --git a/src/openhuman/agent/harness/session/turn/core.rs b/src/openhuman/agent/harness/session/turn/core.rs index 46d47609d..6b6418776 100644 --- a/src/openhuman/agent/harness/session/turn/core.rs +++ b/src/openhuman/agent/harness/session/turn/core.rs @@ -5,7 +5,7 @@ use super::super::turn_engine_adapter::{AgentCheckpoint, AgentObserver, AgentToo use super::super::types::Agent; use super::{ integration_announcement_note, mcp_announcement_note, newly_connected_slugs, - normalize_tool_call, + skill_announcement_note, }; use crate::openhuman::agent::harness; use crate::openhuman::agent::harness::definition::TriggerMemoryAgent; @@ -56,6 +56,11 @@ impl Agent { self.config.max_tool_iterations ); self.ensure_composio_integrations_listener(); + // Arm the installed-skills listener at turn start (not lazily inside + // `drain_skill_events`, which is only reached after the first turn) — + // broadcast subscriptions are not retroactive, so a skill installed + // during turn 1 would otherwise be missed until a later subscribe. + self.ensure_skill_events_listener(); // ── Session transcript resume ───────────────────────────────── // On a fresh session (empty history), look for a previous // transcript to pre-populate the exact provider messages for @@ -171,6 +176,20 @@ impl Agent { // old `Config::load_or_init()` round-trip on every turn. // let _ = self.refresh_delegation_tools_from_cached_integrations("turn-boundary"); + // Same idea for installed skills. The system-prompt + // `## Installed Skills` block is frozen at turn 1 for KV-cache + // stability (history is non-empty here, so it is never rebuilt + // mid-session), so — exactly like the MCP mechanism — the + // user-turn announcement below is what surfaces a mid-session + // install to the model. `refresh_workflows` updates the tracked + // set (so the next refresh diffs correctly and a future fresh + // session renders the new catalogue) and parks the announcement. + // Event-driven (mirror of the composio path): only re-scan disk + // when a `WorkflowsChanged` event was published since the last + // turn — no per-turn filesystem walk on the steady-state hot path. + if self.drain_skill_events() { + let _ = self.refresh_workflows("event"); + } // Cache empty/expired or config unavailable => no signal. // We leave the current tool surface alone and pick up any // real change on the next turn after the UI's 5 s poll has @@ -327,42 +346,19 @@ impl Agent { .inject_agent_experience_context(user_message, enriched) .await; - // ── SKILL.md body injection (#781) ─────────────────────────── - // Match installed SKILL.md skills against the user message and - // prepend their bodies ahead of the memory-context block so the - // LLM sees them at the top of the user turn. See the module - // docs on [`crate::openhuman::workflows::inject`] for the matching - // heuristic and size cap rationale. - let enriched = { - use crate::openhuman::workflows::inject; - let matches = inject::match_workflows(&self.workflows, user_message); - if matches.is_empty() { - log::debug!( - "[workflows:inject] no skill matches for user message (skill_catalog_len={})", - self.workflows.len() - ); - enriched - } else { - let injection = inject::render_injection( - &matches, - inject::DEFAULT_MAX_INJECTION_BYTES, - |skill| skill.read_body(), - ); - let matched_count = injection.decisions.iter().filter(|d| d.matched).count(); - log::info!( - "[workflows:inject] summary candidates={} matched={} injected_bytes={} truncated_any={}", - injection.decisions.len(), - matched_count, - injection.injected_bytes, - injection.truncated - ); - if injection.rendered.is_empty() { - enriched - } else { - format!("{}\n{}", injection.rendered, enriched) - } - } - }; + // ── SKILL.md body injection: REMOVED (was #781) ────────────── + // We used to keyword-match installed skills against the user message + // and prepend their full SKILL.md bodies onto the user turn. That + // brittle name/description/tag match fired unintentionally and — by + // baking the body into the stored user message — left full skill text + // permanently in chat history (microcompact only clears tool results, + // not user messages). + // + // Skills are now surfaced via the compact `## Installed Skills` + // catalog in the orchestrator prompt and executed via `run_skill`, + // which loads and follows the SKILL.md inside an isolated worker, so + // the full body never enters this conversation. `self.workflows` still + // feeds the catalog through `PromptContext`. // Consume any one-shot mid-session connect announcement parked by // `refresh_delegation_tools_from_cached_integrations`. It rides on the @@ -383,6 +379,15 @@ impl Agent { None => enriched, }; + // Same one-shot pattern for skills installed mid-session (parked by + // `refresh_workflows` above). Rides the user turn so the KV-cache + // prefix stays stable; `.take()` fires it exactly once. + let pending_skills = std::mem::take(&mut self.pending_skill_announcement); + let enriched = match skill_announcement_note(&pending_skills) { + Some(note) => format!("{note}\n\n{enriched}"), + None => enriched, + }; + // Pin the main agent to its configured model for the lifetime of // the session. Per-turn classification used to run here, but it // would flip `effective_model` mid-conversation (e.g. reasoning → diff --git a/src/openhuman/agent/harness/session/turn/mod.rs b/src/openhuman/agent/harness/session/turn/mod.rs index d14ec8513..04ae83850 100644 --- a/src/openhuman/agent/harness/session/turn/mod.rs +++ b/src/openhuman/agent/harness/session/turn/mod.rs @@ -118,6 +118,24 @@ Use the use_mcp_server delegate to act on them immediately — do not tell the u )) } +/// One-shot note prepended to the next user turn when skills are installed +/// mid-session. Mirrors [`integration_announcement_note`] for the +/// `## Installed Skills` catalogue: tells the model the freshly-installed +/// skills are usable now (via `run_skill`) so it acts instead of claiming +/// they aren't installed from stale context. Returns `None` when nothing is +/// pending. Rides the user turn (not the system prompt) to keep the KV-cache +/// prefix stable. +pub(super) fn skill_announcement_note(skill_ids: &[String]) -> Option { + if skill_ids.is_empty() { + return None; + } + Some(format!( + "[skills update] These skill(s) were installed during this conversation and are available right now: {}. \ +They are in your `## Installed Skills` list — run one with `run_skill` immediately; do not tell the user to reinstall or restart.", + skill_ids.join(", ") + )) +} + /// Wrapper around /// [`crate::openhuman::memory_tree::tree_runtime::store::collect_root_summaries_with_caps`] /// that takes user-resolved per-namespace and total caps. The actual diff --git a/src/openhuman/agent/harness/session/turn/tools.rs b/src/openhuman/agent/harness/session/turn/tools.rs index 64ae82dc1..735f0d9e0 100644 --- a/src/openhuman/agent/harness/session/turn/tools.rs +++ b/src/openhuman/agent/harness/session/turn/tools.rs @@ -246,6 +246,64 @@ impl Agent { saw_signal } + /// Lazily attach this session to the global event bus so it can observe + /// [`crate::core::event_bus::DomainEvent::WorkflowsChanged`] (skill + /// install / uninstall / create). Mirror of + /// [`Self::ensure_composio_integrations_listener`]. + pub(in super::super) fn ensure_skill_events_listener(&mut self) { + if self.skill_events_rx.is_some() { + return; + } + if let Some(bus) = crate::core::event_bus::global() { + self.skill_events_rx = Some(bus.raw_receiver()); + log::debug!( + "[agent_loop] armed installed-skills listener for session='{}'", + self.event_session_id + ); + } + } + + /// Drain pending [`crate::core::event_bus::DomainEvent::WorkflowsChanged`] + /// events. Returns `true` when at least one was observed (or the listener + /// lagged) and the caller should re-scan the installed skill set via + /// [`Self::refresh_workflows`]. Mirror of + /// [`Self::drain_composio_integrations_changed_events`]. + pub(in super::super) fn drain_skill_events(&mut self) -> bool { + self.ensure_skill_events_listener(); + let Some(rx) = self.skill_events_rx.as_mut() else { + return false; + }; + use tokio::sync::broadcast::error::TryRecvError; + + let mut saw_signal = false; + let mut closed = false; + loop { + match rx.try_recv() { + Ok(crate::core::event_bus::DomainEvent::WorkflowsChanged { reason }) => { + saw_signal = true; + log::info!("[agent_loop] received installed-skills changed event ({reason})"); + } + Ok(_) => {} + Err(TryRecvError::Empty) => break, + Err(TryRecvError::Lagged(skipped)) => { + saw_signal = true; + log::warn!( + "[agent_loop] installed-skills listener lagged by {} event(s); forcing catalogue re-check", + skipped + ); + } + Err(TryRecvError::Closed) => { + closed = true; + break; + } + } + } + if closed { + self.skill_events_rx = None; + } + saw_signal + } + /// Reconcile the session's delegation schema against the latest cached /// integrations snapshot. Returns `true` only when a refresh applied. pub(in super::super) fn refresh_delegation_tools_from_cached_integrations( @@ -299,6 +357,104 @@ impl Agent { } } + /// Reconcile the tracked installed-skill set ([`Self::workflows`]) against + /// what is on disk, so a skill installed/uninstalled mid-session can be + /// surfaced to the model without a session restart. + /// + /// Note the system-prompt `## Installed Skills` block is frozen at turn 1 + /// (KV-cache stability — it is only built when history is empty), so this + /// does NOT rebuild that block for the live session. Instead — exactly like + /// [`Self::refresh_delegation_tools_from_cached_integrations`] / the MCP + /// mid-session mechanism — genuinely-new skill ids (present on disk but not + /// in the prior snapshot) are parked in [`Self::pending_skill_announcement`] + /// (announced once via [`Self::announced_skills`]) and surfaced on the next + /// user turn; `run_skill` then loads/runs them fresh from disk. Updating the + /// tracked slice keeps the next diff correct and feeds a *fresh* session's + /// rendered catalogue. + /// + /// Returns `true` when the installed set changed. Cheap no-op when it + /// hasn't: a directory scan plus an id-set comparison, no prompt rebuild. + pub(in super::super) fn refresh_workflows(&mut self, trigger: &str) -> bool { + let id_of = |w: &crate::openhuman::workflows::Workflow| -> String { + if w.dir_name.is_empty() { + w.name.clone() + } else { + w.dir_name.clone() + } + }; + let latest = crate::openhuman::workflows::load_workflow_metadata(&self.workspace_dir); + let current_ids: std::collections::HashSet = + self.workflows.iter().map(&id_of).collect(); + let latest_ids: std::collections::HashSet = latest.iter().map(&id_of).collect(); + if current_ids == latest_ids { + return false; + } + // Newly-present skills (on disk now, absent from the prior snapshot), + // announced at most once this session. Removals are detected by the + // id-set diff above but are NOT yet retracted from the frozen + // catalogue (symmetric to the integration/MCP announcements, which + // also only announce additions) — tracked in + // tinyhumansai/openhuman#3738. + let newly: Vec = latest_ids + .difference(¤t_ids) + .filter(|id| self.announced_skills.insert((*id).clone())) + .cloned() + .collect(); + log::info!( + "[agent_loop] installed-skills set changed ({trigger}): {} -> {} skills; updating tracked set + parking announcement (system-prompt catalogue is frozen mid-session; the user-turn note surfaces the change)", + self.workflows.len(), + latest.len() + ); + self.workflows = latest; + for id in newly { + if !self.pending_skill_announcement.contains(&id) { + self.pending_skill_announcement.push(id); + } + } + true + } + + /// Test-only: installed-skill ids currently in the catalogue snapshot + /// (`dir_name`, falling back to `name`). Lets `refresh_workflows` tests + /// assert through a method instead of touching private fields. + #[cfg(test)] + pub(in super::super) fn test_workflow_ids(&self) -> Vec { + self.workflows + .iter() + .map(|w| { + if w.dir_name.is_empty() { + w.name.clone() + } else { + w.dir_name.clone() + } + }) + .collect() + } + + /// Test-only: skill ids parked for the next-turn `[skills update]` + /// announcement by `refresh_workflows`. + #[cfg(test)] + pub(in super::super) fn test_pending_skill_announcement(&self) -> &[String] { + &self.pending_skill_announcement + } + + /// Test-only: inject a specific skill-events receiver (e.g. one whose + /// sender has been dropped) so `drain_skill_events`' `Closed` arm is + /// reachable without the global bus singleton. + #[cfg(test)] + pub(in super::super) fn set_skill_events_rx_for_test( + &mut self, + rx: tokio::sync::broadcast::Receiver, + ) { + self.skill_events_rx = Some(rx); + } + + /// Test-only: whether the skill-events listener is currently armed. + #[cfg(test)] + pub(in super::super) fn has_skill_events_rx(&self) -> bool { + self.skill_events_rx.is_some() + } + /// Re-synthesise `delegate_*` tools for the orchestrator's `subagents` /// declaration using the live `connected_integrations` slice, and /// reconcile the resulting set into `self.tools` / `self.tool_specs` / diff --git a/src/openhuman/agent/harness/session/turn_tests.rs b/src/openhuman/agent/harness/session/turn_tests.rs index b9a53d50b..8507f048c 100644 --- a/src/openhuman/agent/harness/session/turn_tests.rs +++ b/src/openhuman/agent/harness/session/turn_tests.rs @@ -1858,3 +1858,22 @@ fn integration_announcement_accumulates_two_connects_in_one_note() { "startup slug must not re-announce: {note}" ); } + +#[test] +fn skill_announcement_note_empty_yields_none() { + assert!(super::skill_announcement_note(&[]).is_none()); +} + +#[test] +fn skill_announcement_note_mentions_ids_and_run_skill() { + let note = + super::skill_announcement_note(&["ascii-art".to_string(), "github-issues".to_string()]) + .expect("non-empty input should yield a note"); + assert!(note.contains("[skills update]")); + assert!(note.contains("ascii-art")); + assert!(note.contains("github-issues")); + assert!( + note.contains("run_skill"), + "note must steer the model to run_skill: {note}" + ); +} diff --git a/src/openhuman/agent/harness/session/types.rs b/src/openhuman/agent/harness/session/types.rs index 96d06c80a..fa9c54e0c 100644 --- a/src/openhuman/agent/harness/session/types.rs +++ b/src/openhuman/agent/harness/session/types.rs @@ -201,6 +201,12 @@ pub struct Agent { /// ACTIVE mid-turn can refresh the delegation schema in the same thread. pub(super) composio_integrations_rx: Option>, + /// Lazily-armed global-bus receiver for [`DomainEvent::WorkflowsChanged`] + /// (skill install / uninstall / create). Drained at each turn boundary so + /// `refresh_workflows` only re-scans disk when the installed set actually + /// changed — no per-turn filesystem walk on the steady-state hot path. + pub(super) skill_events_rx: + Option>, /// Toolkit slugs already surfaced to the model as freshly-connected /// this session. Seeded at turn 1 with the startup connected set, then /// extended whenever a mid-session connect is announced — so each new @@ -232,6 +238,17 @@ pub struct Agent { /// note rides the user turn (NOT the system prompt) so the KV-cache prefix /// stays byte-identical. Order-preserving + de-duped on insert. pub(super) pending_mcp_announcement: Vec, + /// Skill ids discovered mid-session (installed after session build) that + /// still need announcing on the next user message. Mirrors + /// [`Self::pending_integration_announcement`] for the `## Installed Skills` + /// catalogue: parked by `refresh_workflows`, rendered + cleared when the + /// next user message is built so the note rides the user turn (NOT the + /// system prompt) and the KV-cache prefix stays byte-identical. + pub(super) pending_skill_announcement: Vec, + /// Skill ids already surfaced to the model as installed this session, so + /// each newly-installed skill is announced exactly once and never + /// re-announced per turn. Seeded from the session-build catalogue. + pub(super) announced_skills: std::collections::HashSet, /// Optional reference to the `ArchivistHook` registered in /// `post_turn_hooks`. Kept separately so the turn loop can call /// `flush_open_segment` at session-memory-extraction time (the diff --git a/src/openhuman/agent_registry/agents/orchestrator/prompt.md b/src/openhuman/agent_registry/agents/orchestrator/prompt.md index c7798366c..c9efebc7f 100644 --- a/src/openhuman/agent_registry/agents/orchestrator/prompt.md +++ b/src/openhuman/agent_registry/agents/orchestrator/prompt.md @@ -34,7 +34,7 @@ Follow this sequence for every user message: - **Any task that touches a code repository — cloning, exploring, locating files, modifying, building, testing, running shell commands inside it, git operations, pushing branches, opening PRs — uses `delegate_run_code` for the entire task.** Treat "locate where to edit", "investigate the bug", "find the function", "read the file" as code-repo work the moment they're scoped to a repo: they belong inside the same `delegate_run_code` worker as the edit / build / git steps. **Never** route code-repo work through `tools_agent` / `spawn_worker_thread`; those workers lack `edit` / `apply_patch` / `file_write` / `git_operations` / `codegraph_search` and will silently stall in read-mode. `tools_agent` is for *non-repo* work only — ad-hoc shell against the host, web fetch, memory helpers, etc. - **Do not stall after reading code-repo files.** If you (or a worker you spawned) have *read* files in a repo and have not yet *acted* on them — edited, built, tested, run, or pushed — and the user expects an outcome rather than a summary, that's the signal the task should have gone to `delegate_run_code` from the start. Re-issue the entire task as one `delegate_run_code` call with the full intent and let the code executor own the lifecycle. Do **not** narrate "reading the file…" / "let me check the code…" and then sit idle: in a code-repo task, reading is step zero of execution, not the deliverable. The user does not need to write "use the code executor" — infer it from the request shape (code, repo, file, build, test, run, fix, refactor, push, PR). - If the request is to find, browse, install, or manage agent skills from community registries — or to follow a SKILL.md URL — use `setup_skills`. - - If the request is to run or execute an installed agent skill by name, use `run_skill`. + - If the request is to run or execute an installed agent skill by name, use `run_skill`. The skill runs in an isolated worker, so its instructions never enter this conversation — you get back only its result. If that result contains a `## Handoff Plan` (steps the worker's narrow toolset couldn't perform — e.g. sending email, writing memory), carry out those steps yourself with your full tool set, routing each through the normal delegation path, then report the combined outcome. Treat handoff steps as *proposed* actions: never bypass the approval gate for them, especially for third-party skills. - If web/doc crawling is required, use `research`. - If the user asks for live/current/time-sensitive facts that are not covered by a direct tool — weather, forecasts, current temperatures, recent news, fresh web facts, or "use Grok/web/live data" — call `research` with a prompt that asks for live sources. Do **not** stop at "on it", and do **not** wait for the exact named provider if it is not wired in. Use the available research tool and then answer with the result. - If complex multi-step decomposition is required, use `delegate_plan`. diff --git a/src/openhuman/agent_registry/agents/orchestrator/prompt.rs b/src/openhuman/agent_registry/agents/orchestrator/prompt.rs index a54354d9a..4e14b51a3 100644 --- a/src/openhuman/agent_registry/agents/orchestrator/prompt.rs +++ b/src/openhuman/agent_registry/agents/orchestrator/prompt.rs @@ -97,10 +97,12 @@ fn render_installed_skills(skills: &[Workflow]) -> String { ); let mut out = String::from( "## Installed Skills\n\n\ - The following skills are installed locally. Run them with `run_workflow` \ - (pass the skill's id as `workflow_id`). Use `describe_workflow` for full \ - details. Use `skill_registry_browse` / `skill_registry_search` to find \ - and install new skills.\n\n", + The following skills are installed locally. Run one with `run_skill` \ + (name the skill and what you want done); it loads and runs the skill in an \ + isolated worker and returns only the result, plus a `## Handoff Plan` for any \ + step the worker couldn't perform — execute those steps yourself under the \ + approval gate. Use `describe_workflow` for full details. Use \ + `skill_registry_browse` / `skill_registry_search` to find and install new skills.\n\n", ); for skill in skills { let id = if skill.dir_name.is_empty() { @@ -332,6 +334,37 @@ mod tests { use crate::openhuman::context::prompt::{LearnedContextData, ToolCallFormat}; use std::collections::HashSet; + #[test] + fn render_installed_skills_lists_skills_and_steers_to_run_skill() { + let skills = vec![ + Workflow { + dir_name: "ascii-art".into(), + description: "ASCII art via pyfiglet".into(), + ..Default::default() + }, + // dir_name empty -> id falls back to name; empty description -> + // "(no description)". + Workflow { + name: "no-dir".into(), + ..Default::default() + }, + ]; + let out = render_installed_skills(&skills); + assert!(out.contains("## Installed Skills")); + assert!( + out.contains("run_skill"), + "catalogue must steer to run_skill" + ); + assert!(out.contains("Handoff Plan")); + assert!(out.contains("- **ascii-art**: ASCII art via pyfiglet")); + assert!(out.contains("- **no-dir**: (no description)")); + } + + #[test] + fn render_installed_skills_empty_is_omitted() { + assert_eq!(render_installed_skills(&[]), ""); + } + fn ctx_with<'a>(integrations: &'a [ConnectedIntegration]) -> PromptContext<'a> { use std::sync::OnceLock; static EMPTY_VISIBLE: OnceLock> = OnceLock::new(); diff --git a/src/openhuman/skill_runtime/agent/skill_executor/prompt.md b/src/openhuman/skill_runtime/agent/skill_executor/prompt.md index 4c39739d1..661318c97 100644 --- a/src/openhuman/skill_runtime/agent/skill_executor/prompt.md +++ b/src/openhuman/skill_runtime/agent/skill_executor/prompt.md @@ -9,11 +9,12 @@ You execute agent skills that have been installed on this system. Skills are def 1. **Load** the skill's SKILL.md using `describe_workflow` to read its instructions. 2. **Read** any referenced resources using `read_workflow_resource` (scripts, references, etc.). 3. **Resolve runtimes** with `skill_runtime_resolve_runtimes` when the skill references Node.js, npm, npx, Python, or bundled `.js` / `.py` scripts. -4. **Follow** the skill's instructions step by step. +4. **Follow** the skill's instructions step by step, performing each step you have the tools to perform. 5. **Execute** any shell commands or scripts as directed by the skill. - Node.js scripts must use the OpenHuman Node runtime (`runtime_node`) rather than assuming the host PATH. - Python scripts must use the OpenHuman Python runtime (`runtime_python`) rather than assuming the host PATH. -6. **Report** results back to the user. +6. **Hand off** any step you cannot complete with your available tools instead of failing the whole skill — see "When a step needs a tool you don't have" below. +7. **Report** what you completed, plus a handoff plan for anything you delegated upward. > **Output contract:** only a command's stdout/stderr is captured back to you. A Python/Node > script that finishes without printing returns an *empty* result — that is "no output captured", @@ -21,6 +22,28 @@ You execute agent skills that have been installed on this system. Skills are def > `print(...)` / `console.log(...)`); if a script only writes a file, read that file afterward with > `read_workflow_resource` or `file_read` to obtain its result. +## When a step needs a tool you don't have + +Your toolset is intentionally narrow (shell, files, skill loading, runtimes). Some skills need capabilities you don't have — connected integrations (email, chat, calendars), user memory, or other typed tools. When a step requires one of these: + +- **Do not** invent a result, fake success, or try to fake the capability with unrelated shell commands. **Do not** abort the whole skill because one step is out of reach. +- **Do** finish every step you *can* with your own tools first. +- **Do** end your output with a `## Handoff Plan` describing only the remaining steps, so the calling agent — which has the full toolset and runs under user supervision — can finish them. + +Format your final report exactly like this: + +```text +## Completed +- → + +## Handoff Plan +- + - needs: + - inputs: +``` + +Keep the Handoff Plan compact and concrete — it is a list of actions for the caller to run, not a copy of the SKILL.md. Resolve inputs to real values so the caller never has to re-read the skill. Each handoff step is *proposed*, not performed by you: the caller executes it through the approval gate, so describe honestly and precisely what each step does. Never push a step upward just to avoid work — only hand up what genuinely needs a tool you lack. If you completed everything yourself, omit the Handoff Plan entirely. + ## Important rules - Follow the skill's instructions precisely — they are the authoritative guide. @@ -30,3 +53,4 @@ You execute agent skills that have been installed on this system. Skills are def - If a shell command fails, report the error and ask whether to retry or abort. - Respect the skill's `allowed-tools` declaration if present. - When the skill is read-only (no shell commands), do not use the shell tool. +- Prefer doing the work yourself. Only hand a step up via the `## Handoff Plan` when it truly needs a tool you lack — the caller runs those steps under the approval gate, so be precise and honest about what each one does. diff --git a/src/openhuman/workflows/README.md b/src/openhuman/workflows/README.md index 2464c3684..7c9a9430a 100644 --- a/src/openhuman/workflows/README.md +++ b/src/openhuman/workflows/README.md @@ -1,6 +1,6 @@ # Skills -Discovery, parsing, and per-turn injection of agentskills.io-style skills (a directory containing `SKILL.md` with YAML frontmatter and Markdown instructions). Owns scope resolution (User vs Project vs Legacy), trust-marker enforcement, resource reading, install / uninstall, and the matching heuristic that decides which `SKILL.md` body to splice into a chat turn. Does NOT own runtime execution internals or general tool execution (`tools/` / `javascript/`). +Discovery and parsing of agentskills.io-style skills (a directory containing `SKILL.md` with YAML frontmatter and Markdown instructions). Owns scope resolution (User vs Project vs Legacy), trust-marker enforcement, resource reading, and install / uninstall. Skills are surfaced to agents via the compact `## Installed Skills` catalog and executed via `run_skill` in an isolated worker — bodies are no longer spliced into chat turns. Does NOT own runtime execution internals or general tool execution (`tools/` / `javascript/`). ## Public surface @@ -8,14 +8,13 @@ Discovery, parsing, and per-turn injection of agentskills.io-style skills (a dir - `pub const MAX_SKILL_RESOURCE_BYTES: u64 = 128 * 1024` — `ops.rs:39` — bound on per-resource RPC payload. - `pub use ops::*` — `mod.rs:9` — re-exports skill discovery, parsing, install, uninstall, resource reading, and frontmatter types. - `pub struct ToolResult` / `pub enum ToolContent` — `types.rs:7-60` — content blocks returned by skill / tool execution. -- `pub mod inject` — `inject.rs` — per-turn `SKILL.md` body matching + injection into the user prompt (explicit `@name`, tag / description / name substring, with an 8 KiB injected-byte cap). - `pub mod bus` — `bus.rs` — emits skill events on the global event bus. - RPC `skills.{skills_list, skills_read_resource, skills_create, skills_install_from_url, skills_uninstall}` — `schemas.rs` (re-exported `all_skills_controller_schemas` / `all_skills_registered_controllers` via `mod.rs:10`). ## Calls into - `src/openhuman/config/` — workspace path resolution and trust-marker location. -- `src/openhuman/agent/` — injection consumers in `agent/prompts/` and `agent/harness/session/turn.rs`. +- `src/openhuman/agent/` — the `## Installed Skills` catalog rendered in `agent_registry/agents/orchestrator/prompt.rs`, fed by the skill list on `PromptContext` (`agent/harness/session/turn/context.rs`). - `src/openhuman/workspace/` — workspace-relative skill paths. - `src/core/event_bus/` — emits `DomainEvent::Skill(*)` on install / uninstall. @@ -31,5 +30,5 @@ Discovery, parsing, and per-turn injection of agentskills.io-style skills (a dir ## Tests -- Unit: tests live alongside `ops.rs`, `inject.rs`, `schemas.rs`, and `types.rs` as `#[cfg(test)] mod tests` blocks (no separate `*_tests.rs` files in this domain). +- Unit: tests live alongside `ops.rs`, `schemas.rs`, and `types.rs` as `#[cfg(test)] mod tests` blocks (no separate `*_tests.rs` files in this domain). - Cross-cutting agent + skill behavior is covered indirectly by `src/openhuman/agent/harness/session/{turn,runtime}_tests.rs`. diff --git a/src/openhuman/workflows/inject.rs b/src/openhuman/workflows/inject.rs deleted file mode 100644 index d60a62986..000000000 --- a/src/openhuman/workflows/inject.rs +++ /dev/null @@ -1,830 +0,0 @@ -//! SKILL.md body injection into the agent inference loop. -//! -//! This module wires the installed `SKILL.md` catalog into each user -//! turn so the LLM can see a matched skill's instruction body in -//! context. The plain-text catalog section that the prompt builder -//! already renders (`## Available Skills` — name + description only) -//! tells the model **what** skills exist; this injection step gives it -//! the actual instruction bodies for the specific skill(s) relevant to -//! the current message. -//! -//! ## Matching heuristic (v1) -//! -//! For each skill we emit a `matched` decision: -//! -//! 1. **Explicit `@` mention** in the user message — always -//! force-injects. Takes precedence over everything else. Names are -//! matched case-insensitively; `@foo bar` matches skill name -//! `foo-bar` after normalising `-`/`_`/whitespace → `_`. -//! 2. Otherwise, when the skill does **not** declare -//! `user-invocable: false` (default = invocable = true): -//! - `matched = true` when the skill's `description` appears as a -//! case-insensitive substring of the user message, OR any of its -//! `tags` appears as a whole-word case-insensitive substring, OR -//! the skill's `name` appears as a whole-word match. -//! 3. Skills with `user-invocable: false` **only** ever inject on an -//! explicit `@` mention — the auto-match path is disabled for them. -//! -//! The heuristic is intentionally narrow: exact + case-insensitive -//! substring is cheap, predictable for reviewers, and keeps false -//! positives bounded by the 8 KiB total injected-byte cap enforced -//! downstream in [`render_injection`]. More sophisticated ranking -//! (embeddings, LLM-rerank) can replace this later without touching -//! the calling site in `Agent::turn`. -//! -//! ## Ordering -//! -//! Matched skills are returned in this stable order: -//! -//! 1. Explicit `@` mentions in the order they appear in the message. -//! 2. Auto-matched skills by description length (longer first), then -//! by skill name alphabetically as a deterministic tiebreaker. -//! -//! ## Size cap -//! -//! Total injected payload (sum of all `[SKILL:] … [/SKILL]` -//! blocks) is capped at [`DEFAULT_MAX_INJECTION_BYTES`] = 8 KiB. When -//! a single body would push the total over the cap, it is truncated -//! and a `[SKILL::truncated]` marker replaces the closer so the -//! LLM knows the content was cut short. Any subsequent matched skills -//! that would exceed the cap are skipped with `SkipReason::BudgetExhausted` -//! and logged. -//! -//! ## Logging -//! -//! Every candidate emits a grep-friendly `[workflows:inject]` log line -//! with `matched=`, reason, and injected bytes (see -//! [`render_injection`]). A summary line lives in the caller -//! (`Agent::turn`). - -use super::Workflow; -use std::collections::HashSet; - -/// Upper bound on total bytes injected per turn. Matches the umbrella -/// issue #781 acceptance criterion ("≤ 8 KiB"). -pub const DEFAULT_MAX_INJECTION_BYTES: usize = 8 * 1024; - -/// Why a candidate skill was skipped. Kept on the match record for -/// both logging and unit-test assertions. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum SkipReason { - /// `user-invocable: false` skill without an explicit `@` mention. - NotUserInvocable, - /// No match in description / tags / name, and no `@` mention. - NoMatch, - /// Workflow body could not be read from disk (legacy manifest or I/O - /// failure). - BodyUnavailable, - /// Workflow body would push the running total past the size cap. - BudgetExhausted, -} - -/// How a matched skill was selected. Preserved on `WorkflowMatch` so the -/// logger can explain *why* each injection happened. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum MatchReason { - /// Selected via an explicit `@` mention. - AtMention, - /// Description substring matched the user message. - DescriptionSubstring, - /// A tag matched as a whole-word substring. - TagMatch, - /// The skill name itself appeared in the message. - NameMatch, -} - -impl MatchReason { - fn as_str(self) -> &'static str { - match self { - MatchReason::AtMention => "at_mention", - MatchReason::DescriptionSubstring => "description_substring", - MatchReason::TagMatch => "tag_match", - MatchReason::NameMatch => "name_match", - } - } -} - -/// A skill that passed the matcher. The caller resolves its body at -/// render time. -#[derive(Debug, Clone)] -pub struct WorkflowMatch<'a> { - pub skill: &'a Workflow, - pub reason: MatchReason, - /// Position in the user message for `@`-mention matches. Used to - /// preserve message order. Auto-matches get `usize::MAX` so they - /// sort after explicit mentions. - pub mention_index: usize, -} - -/// Per-skill decision returned to the caller for logging. Covers both -/// matched and skipped candidates so there is a single source of truth -/// for what happened this turn. -#[derive(Debug, Clone)] -pub struct WorkflowDecision { - pub name: String, - pub matched: bool, - pub reason: String, - pub injected_bytes: usize, - pub truncated: bool, -} - -/// Result of [`render_injection`] — the rendered block plus machine- -/// readable stats for logging. -#[derive(Debug, Clone, Default)] -pub struct Injection { - /// Concatenated `[SKILL:] … [/SKILL]` blocks. Empty when - /// nothing matched (or every match was skipped). - pub rendered: String, - /// Total bytes in `rendered`. - pub injected_bytes: usize, - /// Whether at least one body was truncated to fit the cap. - pub truncated: bool, - /// Per-candidate decisions (both matched and skipped) for logging. - pub decisions: Vec, -} - -/// Read the `user-invocable` flag from a skill's frontmatter. Defaults -/// to `true` (opt-out) when absent or unparseable. Accepts both the -/// spec-compliant `metadata.user-invocable` location and the deprecated -/// top-level `user-invocable` key (emitted with a migration warning by -/// the catalog loader). -pub fn is_user_invocable(skill: &Workflow) -> bool { - let lookup_bool = |key: &str| -> Option { - if let Some(v) = skill.frontmatter.metadata.get(key) { - if let Some(b) = v.as_bool() { - return Some(b); - } - } - if let Some(v) = skill.frontmatter.extra.get(key) { - if let Some(b) = v.as_bool() { - return Some(b); - } - } - None - }; - lookup_bool("user-invocable") - .or_else(|| lookup_bool("user_invocable")) - .unwrap_or(true) -} - -/// Normalise a skill name for case-insensitive `@` matching: -/// lowercase, collapse `-`/`_` runs to single `-`. -fn normalise(name: &str) -> String { - let mut out = String::with_capacity(name.len()); - let mut prev_sep = false; - for ch in name.chars().flat_map(|c| c.to_lowercase()) { - if ch == '-' || ch == '_' { - if !prev_sep && !out.is_empty() { - out.push('-'); - } - prev_sep = true; - } else { - out.push(ch); - prev_sep = false; - } - } - // Trim trailing separator if any. - if out.ends_with('-') { - out.pop(); - } - out -} - -/// Scan the user message for `@` patterns. Returns the -/// normalised skill name plus the byte index at which the `@` appears -/// (used later to preserve the original message order across mentions). -/// -/// A token qualifies as an `@` mention when: -/// - it starts with `@` (not preceded by an alphanumeric character so -/// email addresses don't accidentally trigger) -/// - and the following run of `[A-Za-z0-9_-]+` is non-empty -pub fn extract_mentions(user_message: &str) -> Vec<(String, usize)> { - let bytes = user_message.as_bytes(); - let mut out = Vec::new(); - let mut i = 0; - while i < bytes.len() { - if bytes[i] == b'@' { - let preceded_by_alnum = i > 0 - && (bytes[i - 1].is_ascii_alphanumeric() || bytes[i - 1] == b'.') - && !bytes[i - 1].is_ascii_whitespace(); - if preceded_by_alnum { - i += 1; - continue; - } - let start = i + 1; - let mut end = start; - while end < bytes.len() { - let c = bytes[end]; - if c.is_ascii_alphanumeric() || c == b'_' || c == b'-' { - end += 1; - } else { - break; - } - } - if end > start { - let name = &user_message[start..end]; - out.push((normalise(name), i)); - i = end; - continue; - } - } - i += 1; - } - out -} - -fn contains_whole_word(haystack_lower: &str, needle_lower: &str) -> bool { - if needle_lower.is_empty() { - return false; - } - // Whole-word = surrounding chars are NOT alphanumeric/_-. Simple - // loop over match positions rather than pulling in a regex crate. - let hay = haystack_lower.as_bytes(); - let ndl = needle_lower.as_bytes(); - if ndl.len() > hay.len() { - return false; - } - // `@` counts as a word character so a name/tag that happens to sit - // inside an email or `@mention` (`foo@alice.example.com`, `@gmail`) - // does not slip through the whole-word gate. Explicit mentions are - // handled separately by [`extract_mentions`]. - let is_word = |c: u8| c.is_ascii_alphanumeric() || c == b'_' || c == b'-' || c == b'@'; - let mut i = 0; - while i + ndl.len() <= hay.len() { - if &hay[i..i + ndl.len()] == ndl { - let left_ok = i == 0 || !is_word(hay[i - 1]); - let right_ok = i + ndl.len() == hay.len() || !is_word(hay[i + ndl.len()]); - if left_ok && right_ok { - return true; - } - } - i += 1; - } - false -} - -/// Match installed skills against a user message per the heuristic -/// documented at the top of this module. -pub fn match_workflows<'a>( - workflows: &'a [Workflow], - user_message: &str, -) -> Vec> { - let mentions = extract_mentions(user_message); - let mention_set: HashSet = mentions.iter().map(|(n, _)| n.clone()).collect(); - let mention_index = |skill_norm: &str| -> Option { - mentions - .iter() - .find(|(n, _)| n == skill_norm) - .map(|(_, idx)| *idx) - }; - - let lower_msg = user_message.to_lowercase(); - - let mut matches: Vec> = Vec::new(); - for skill in workflows { - let normalised_name = normalise(&skill.name); - let user_invocable = is_user_invocable(skill); - - // 1. `@` mention always wins. - if mention_set.contains(&normalised_name) { - let idx = mention_index(&normalised_name).unwrap_or(usize::MAX); - matches.push(WorkflowMatch { - skill, - reason: MatchReason::AtMention, - mention_index: idx, - }); - continue; - } - - // 2. Auto-match only when skill allows user invocation. - if !user_invocable { - continue; - } - - let desc_lower = skill.description.to_lowercase(); - if !desc_lower.is_empty() && lower_msg.contains(&desc_lower) { - matches.push(WorkflowMatch { - skill, - reason: MatchReason::DescriptionSubstring, - mention_index: usize::MAX, - }); - continue; - } - - let mut tag_hit = false; - for tag in &skill.tags { - let tag_lower = tag.to_lowercase(); - if contains_whole_word(&lower_msg, &tag_lower) { - tag_hit = true; - break; - } - } - if tag_hit { - matches.push(WorkflowMatch { - skill, - reason: MatchReason::TagMatch, - mention_index: usize::MAX, - }); - continue; - } - - // Name-as-whole-word fallback (e.g. user says "run the - // pdf-cruncher skill"). Skipped when the name is a very short - // token that would over-match (<= 2 chars). - let name_lower = skill.name.to_lowercase(); - if name_lower.chars().count() > 2 && contains_whole_word(&lower_msg, &name_lower) { - matches.push(WorkflowMatch { - skill, - reason: MatchReason::NameMatch, - mention_index: usize::MAX, - }); - } - } - - // Stable ordering: `@` mentions by message index first; auto-matches - // by description length descending, tie-breaking on skill name. - matches.sort_by(|a, b| match (a.reason, b.reason) { - (MatchReason::AtMention, MatchReason::AtMention) => a.mention_index.cmp(&b.mention_index), - (MatchReason::AtMention, _) => std::cmp::Ordering::Less, - (_, MatchReason::AtMention) => std::cmp::Ordering::Greater, - _ => { - let len_cmp = b.skill.description.len().cmp(&a.skill.description.len()); - if len_cmp != std::cmp::Ordering::Equal { - len_cmp - } else { - a.skill.name.cmp(&b.skill.name) - } - } - }); - - matches -} - -/// Build the injection block. Resolves each match's body via -/// `body_resolver` so callers can swap in a fake reader for tests. -/// -/// `max_bytes` caps the total rendered size. When a body would exceed -/// the remaining budget it is truncated on a UTF-8 boundary and -/// emitted with a `[SKILL::truncated]` close marker. -pub fn render_injection<'a, F>( - matches: &[WorkflowMatch<'a>], - max_bytes: usize, - mut body_resolver: F, -) -> Injection -where - F: FnMut(&Workflow) -> Option, -{ - const SKILL_OPEN_FMT: &str = "[SKILL:{}]\n"; - const SKILL_CLOSE_FMT: &str = "\n[/SKILL]\n"; - const SKILL_CLOSE_TRUNC_FMT: &str = "\n[/SKILL:truncated]\n"; - - let mut rendered = String::new(); - let mut decisions: Vec = Vec::new(); - let mut truncated_any = false; - - for m in matches { - let name = &m.skill.name; - let body = match body_resolver(m.skill) { - Some(b) => b, - None => { - log::warn!( - "[workflows:inject] matched={} reason={} name={} skipped=body_unavailable", - false, - "body_unavailable", - name - ); - decisions.push(WorkflowDecision { - name: name.clone(), - matched: false, - reason: format!("skipped:{:?}", SkipReason::BodyUnavailable), - injected_bytes: 0, - truncated: false, - }); - continue; - } - }; - - let header = SKILL_OPEN_FMT.replacen("{}", name, 1); - let footer_full = SKILL_CLOSE_FMT.to_string(); - let footer_trunc = SKILL_CLOSE_TRUNC_FMT.to_string(); - - let remaining = max_bytes.saturating_sub(rendered.len()); - let header_len = header.len(); - let footer_full_len = footer_full.len(); - let footer_trunc_len = footer_trunc.len(); - - // Minimum we need to emit anything meaningful: header + at - // least 1 byte of body + truncation footer. - let min_truncated = header_len + footer_trunc_len + 1; - if remaining < min_truncated { - log::info!( - "[workflows:inject] matched={} reason={} name={} skipped=budget_exhausted remaining_bytes={}", - false, - "budget_exhausted", - name, - remaining - ); - decisions.push(WorkflowDecision { - name: name.clone(), - matched: false, - reason: format!("skipped:{:?}", SkipReason::BudgetExhausted), - injected_bytes: 0, - truncated: false, - }); - continue; - } - - // Can we fit the whole body + full footer? - let full_len = header_len + body.len() + footer_full_len; - if full_len <= remaining { - rendered.push_str(&header); - rendered.push_str(&body); - rendered.push_str(&footer_full); - let injected = header_len + body.len() + footer_full_len; - log::debug!( - "[workflows:inject] matched={} reason={} name={} injected_bytes={} truncated={}", - true, - m.reason.as_str(), - name, - injected, - false - ); - decisions.push(WorkflowDecision { - name: name.clone(), - matched: true, - reason: m.reason.as_str().to_string(), - injected_bytes: injected, - truncated: false, - }); - continue; - } - - // Truncate: how many body bytes can we fit with the truncated - // footer? - let max_body = remaining.saturating_sub(header_len + footer_trunc_len); - // Round down to a char boundary. - let cut = crate::openhuman::util::floor_char_boundary(&body, max_body); - let truncated_body = &body[..cut]; - - rendered.push_str(&header); - rendered.push_str(truncated_body); - rendered.push_str(&footer_trunc); - truncated_any = true; - let injected = header_len + truncated_body.len() + footer_trunc_len; - log::warn!( - "[workflows:inject] matched={} reason={} name={} injected_bytes={} truncated={} body_bytes_total={} body_bytes_kept={}", - true, - m.reason.as_str(), - name, - injected, - true, - body.len(), - truncated_body.len() - ); - decisions.push(WorkflowDecision { - name: name.clone(), - matched: true, - reason: m.reason.as_str().to_string(), - injected_bytes: injected, - truncated: true, - }); - } - - let injected_bytes = rendered.len(); - Injection { - rendered, - injected_bytes, - truncated: truncated_any, - decisions, - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::openhuman::workflows::{Workflow, WorkflowFrontmatter}; - use std::collections::HashMap; - - fn skill(name: &str, description: &str) -> Workflow { - Workflow { - name: name.to_string(), - dir_name: name.to_string(), - description: description.to_string(), - version: "0.1.0".into(), - author: None, - tags: Vec::new(), - platforms: Vec::new(), - related_skills: Vec::new(), - source_format: "openhuman".to_string(), - tools: Vec::new(), - prompts: Vec::new(), - location: None, - frontmatter: WorkflowFrontmatter::default(), - resources: Vec::new(), - scope: Default::default(), - legacy: false, - warnings: Vec::new(), - } - } - - fn skill_with_tags(name: &str, description: &str, tags: &[&str]) -> Workflow { - let mut s = skill(name, description); - s.tags = tags.iter().map(|t| t.to_string()).collect(); - s - } - - fn skill_with_flag(name: &str, description: &str, flag_key: &str, flag: bool) -> Workflow { - let mut s = skill(name, description); - let mut map: HashMap = HashMap::new(); - map.insert(flag_key.to_string(), serde_yaml::Value::Bool(flag)); - s.frontmatter.metadata = map; - s - } - - #[test] - fn matches_skill_by_description_substring() { - let skills = vec![skill("email", "send email via gmail")]; - let m = match_workflows(&skills, "Please send email via gmail to alice."); - assert_eq!(m.len(), 1); - assert_eq!(m[0].reason, MatchReason::DescriptionSubstring); - } - - #[test] - fn matches_skill_by_tag_whole_word() { - let skills = vec![skill_with_tags("tp", "do things", &["pdf"])]; - let m = match_workflows(&skills, "Convert this pdf please."); - assert_eq!(m.len(), 1); - assert_eq!(m[0].reason, MatchReason::TagMatch); - } - - #[test] - fn tag_partial_word_does_not_match() { - let skills = vec![skill_with_tags("sk", "x", &["crypt"])]; - let m = match_workflows(&skills, "I like cryptography."); - // `crypt` is not a standalone word in `cryptography`. - assert!(m.is_empty(), "got: {:?}", m); - } - - #[test] - fn matches_skill_by_name_whole_word() { - let skills = vec![skill("pdf-crunch", "unrelated")]; - let m = match_workflows(&skills, "Run the pdf-crunch skill now"); - assert_eq!(m.len(), 1); - assert_eq!(m[0].reason, MatchReason::NameMatch); - } - - #[test] - fn explicit_at_mention_force_injects() { - let skills = vec![skill("notes", "completely unrelated description")]; - let m = match_workflows(&skills, "Hey can you @notes me the summary?"); - assert_eq!(m.len(), 1); - assert_eq!(m[0].reason, MatchReason::AtMention); - } - - #[test] - fn at_mention_case_insensitive_and_handles_dashes() { - let skills = vec![skill("pdf-crunch", "foo")]; - let m = match_workflows(&skills, "Use @Pdf-Crunch please"); - assert_eq!(m.len(), 1); - assert_eq!(m[0].reason, MatchReason::AtMention); - } - - #[test] - fn email_address_at_does_not_trigger_mention() { - let skills = vec![skill("alice", "nothing relevant")]; - let m = match_workflows(&skills, "Send email to foo@alice.example.com please"); - // `foo@alice` should not count because `o` precedes `@`. - assert!(m.is_empty(), "got: {:?}", m); - } - - #[test] - fn user_invocable_false_requires_at_mention() { - // description contains "summarize" so it would auto-match if - // invocable, but `user-invocable: false` blocks auto-matching. - let skills = vec![skill_with_flag( - "summary", - "summarize text", - "user-invocable", - false, - )]; - let m = match_workflows(&skills, "Please summarize text for me."); - assert!(m.is_empty(), "auto-match should be suppressed: {:?}", m); - - // But an explicit @ mention still force-injects. - let m2 = match_workflows(&skills, "Hey @summary for me"); - assert_eq!(m2.len(), 1); - assert_eq!(m2[0].reason, MatchReason::AtMention); - } - - #[test] - fn user_invocable_deprecated_underscore_alias() { - let skills = vec![skill_with_flag("x", "xx yy", "user_invocable", false)]; - let m = match_workflows(&skills, "xx yy please"); - assert!(m.is_empty()); - } - - #[test] - fn at_mention_overrides_non_match() { - let skills = vec![skill("bar", "zzz unrelated")]; - let m = match_workflows(&skills, "@bar do it"); - assert_eq!(m.len(), 1); - assert_eq!(m[0].reason, MatchReason::AtMention); - } - - #[test] - fn longer_description_ranks_higher_on_ties() { - let a = skill("aa", "short"); - let b = skill("bb", "this is a much longer description"); - // Both match on the word "description". - let msg = "I want to talk about description"; - // Use tags to guarantee both match. - let mut a = a; - a.tags.push("description".into()); - let mut b = b; - b.tags.push("description".into()); - let skills = [a, b]; - let m = match_workflows(&skills, msg); - assert_eq!(m.len(), 2); - // Longer description first. - assert_eq!(m[0].skill.name, "bb"); - assert_eq!(m[1].skill.name, "aa"); - } - - #[test] - fn at_mentions_sort_before_auto_matches() { - let a = skill("foo", "XXX YYY"); - let b = skill("bar", "XXX YYY"); - // `foo` auto-matches on description; `bar` is explicit via @. - let skills = [a, b]; - let m = match_workflows(&skills, "XXX YYY and @bar"); - assert_eq!(m.len(), 2); - assert_eq!(m[0].skill.name, "bar"); - assert_eq!(m[0].reason, MatchReason::AtMention); - } - - #[test] - fn render_injection_emits_full_block_when_under_budget() { - let s = skill("hello", "say hi"); - let skills = [s]; - let matches = match_workflows(&skills, "@hello please"); - let inj = render_injection(&matches, 1024, |sk| { - assert_eq!(sk.name, "hello"); - Some("instructions body".to_string()) - }); - assert!(inj.rendered.contains("[SKILL:hello]")); - assert!(inj.rendered.contains("instructions body")); - assert!(inj.rendered.contains("[/SKILL]")); - assert!(!inj.truncated); - assert_eq!(inj.decisions.len(), 1); - assert!(inj.decisions[0].matched); - } - - #[test] - fn size_cap_truncates_with_marker() { - let s = skill("big", "huge body"); - let skills = [s]; - let matches = match_workflows(&skills, "@big do it"); - // Force truncation by setting a tight cap. - let big_body = "X".repeat(4000); - let inj = render_injection(&matches, 200, |_| Some(big_body.clone())); - assert!(inj.truncated, "expected truncation: {:?}", inj); - assert!(inj.rendered.contains("[SKILL:big]")); - assert!(inj.rendered.contains("[/SKILL:truncated]")); - assert!(inj.injected_bytes <= 200); - assert!(inj.decisions[0].truncated); - } - - #[test] - fn test_render_injection_utf8_boundary() { - let s = skill("utf8", "d"); - let skills = [s]; - let matches = match_workflows(&skills, "@utf8"); - // Header: [SKILL:utf8]\n (13 bytes) - // Footer (full): \n[/SKILL]\n (10 bytes) - // Footer (trunc): \n[/SKILL:truncated]\n (20 bytes) - // Body: 🦀🦀 (8 bytes) - // Full length: 13 + 8 + 10 = 31 bytes. - // Min truncated: 13 + 20 + 1 = 34 bytes. - - // Cap at 35 bytes. Full (31) fits. - let inj = render_injection(&matches, 35, |_| Some("🦀🦀".to_string())); - assert!(!inj.truncated); - assert_eq!(inj.rendered.chars().filter(|c| *c == '🦀').count(), 2); - - // Cap at 35 bytes again, but with a body that just barely fits. - // (Just demonstrating it doesn't skip when cap >= min_truncated) - assert!(!inj.truncated); - - // To force truncation, body + full footer must exceed cap. - // Let body be 20 bytes. - // Full length: 13 + 20 + 10 = 43 bytes. - // Min truncated: 34 bytes. - // Cap at 40 bytes. - // max_body = 40 - 33 = 7 bytes. - // 7 bytes fits one 🦀 (4 bytes), but not two (8 bytes). - let inj = render_injection(&matches, 40, |_| Some("🦀".repeat(5))); - assert!(inj.truncated); - assert_eq!(inj.rendered.chars().filter(|c| *c == '🦀').count(), 1); - } - - #[test] - fn budget_exhausted_skips_later_candidates() { - let a = skill("first", "x"); - let b = skill("second", "x"); - let skills = [a, b]; - let matches = match_workflows(&skills, "@first @second"); - let body = "X".repeat(200); - // Cap just big enough for one block. - let inj = render_injection(&matches, 250, |_| Some(body.clone())); - assert_eq!(inj.decisions.len(), 2); - let matched_count = inj.decisions.iter().filter(|d| d.matched).count(); - assert_eq!(matched_count, 1); - let skipped = inj.decisions.iter().find(|d| !d.matched).unwrap(); - assert!( - skipped.reason.contains("BudgetExhausted"), - "got: {:?}", - skipped - ); - } - - #[test] - fn body_unavailable_logs_skip() { - let s = skill("ghost", "not on disk"); - let skills = [s]; - let matches = match_workflows(&skills, "@ghost"); - let inj = render_injection(&matches, 1024, |_| None); - assert!(inj.rendered.is_empty()); - assert_eq!(inj.decisions.len(), 1); - assert!(!inj.decisions[0].matched); - assert!(inj.decisions[0].reason.contains("BodyUnavailable")); - } - - #[test] - fn legacy_skill_read_body_returns_none() { - let mut s = skill("legacy", "d"); - s.legacy = true; - assert!(s.read_body().is_none()); - } - - #[test] - fn read_body_round_trip_from_tempfile() { - use std::io::Write; - let tmp = tempfile::tempdir().unwrap(); - let path = tmp.path().join("SKILL.md"); - let mut f = std::fs::File::create(&path).unwrap(); - writeln!( - f, - "---\nname: demo\ndescription: demo skill\n---\n\nThe actual body text.\n" - ) - .unwrap(); - drop(f); - - let mut s = skill("demo", "demo skill"); - s.location = Some(path); - let body = s.read_body().expect("should parse body"); - assert!(body.contains("The actual body text.")); - } - - #[test] - fn default_max_injection_bytes_matches_acceptance() { - // The #781 acceptance criterion is a hard 8 KiB cap. Lock the - // constant so future edits trip this test instead of silently - // relaxing the budget. - assert_eq!(DEFAULT_MAX_INJECTION_BYTES, 8192); - } - - #[test] - fn is_user_invocable_defaults_to_true() { - let s = skill("x", "d"); - assert!(is_user_invocable(&s)); - } - - #[test] - fn is_user_invocable_reads_extra_fallback() { - // Deprecated top-level key lands in `extra`. - let mut s = skill("x", "d"); - s.frontmatter - .extra - .insert("user-invocable".into(), serde_yaml::Value::Bool(false)); - assert!(!is_user_invocable(&s)); - } - - #[test] - fn extract_mentions_preserves_order() { - let m = extract_mentions("first @alpha, then @beta, then @gamma"); - let names: Vec<&str> = m.iter().map(|(n, _)| n.as_str()).collect(); - assert_eq!(names, vec!["alpha", "beta", "gamma"]); - } - - #[test] - fn extract_mentions_skips_bare_at() { - let m = extract_mentions("just an @ sign alone"); - assert!(m.is_empty(), "got: {:?}", m); - } - - #[test] - fn normalise_collapses_separators() { - assert_eq!(normalise("Foo_Bar-Baz"), "foo-bar-baz"); - assert_eq!(normalise("--foo--"), "foo"); - } -} diff --git a/src/openhuman/workflows/mod.rs b/src/openhuman/workflows/mod.rs index 2a20f880b..826724351 100644 --- a/src/openhuman/workflows/mod.rs +++ b/src/openhuman/workflows/mod.rs @@ -1,7 +1,6 @@ -//! Workflow metadata helpers and prompt-injection support. +//! Workflow metadata helpers. pub mod bus; -pub mod inject; pub mod ops; pub mod ops_create; pub mod ops_discover; diff --git a/src/openhuman/workflows/ops_create.rs b/src/openhuman/workflows/ops_create.rs index e0f90159f..813c9b51a 100644 --- a/src/openhuman/workflows/ops_create.rs +++ b/src/openhuman/workflows/ops_create.rs @@ -382,6 +382,15 @@ pub(crate) fn create_workflow_inner( .into_iter() .find(|s| s.name == slug) .ok_or_else(|| format!("created skill '{slug}' but failed to re-discover"))?; + + // Notify live agent sessions so they pick up the new skill in their + // `## Installed Skills` catalogue (see `Agent::refresh_workflows`). + let _ = crate::core::event_bus::publish_global( + crate::core::event_bus::DomainEvent::WorkflowsChanged { + reason: "create".to_string(), + }, + ); + Ok(created) } @@ -668,4 +677,49 @@ mod render_skill_toml_tests { Some("has \"quotes\" and \\ backslash\nand newline") ); } + + /// The trigger half of mid-session refresh: creating a workflow must + /// publish `DomainEvent::WorkflowsChanged` so live sessions re-scan. This + /// guards the `publish_global` emission line (the `refresh_workflows` test + /// writes to disk directly and bypasses create/install, so without this a + /// dropped emission would stay green while silently killing the feature). + #[test] + fn create_workflow_inner_emits_workflows_changed() { + use crate::core::event_bus::{global, init_global, DomainEvent}; + use tokio::sync::broadcast::error::TryRecvError; + + let _ = init_global(64); + let mut rx = global() + .expect("event bus should be initialized") + .raw_receiver(); + + let home = tempfile::TempDir::new().expect("temp home"); + let ws = tempfile::TempDir::new().expect("temp workspace"); + let params = CreateWorkflowParams { + name: "zz-emit-test".into(), + description: "emit test skill".into(), + scope: WorkflowScope::User, + ..Default::default() + }; + create_workflow_inner(Some(home.path()), ws.path(), params) + .expect("create_workflow_inner should succeed"); + + let mut saw = false; + loop { + match rx.try_recv() { + Ok(DomainEvent::WorkflowsChanged { reason }) => { + assert_eq!(reason, "create"); + saw = true; + break; + } + Ok(_) => continue, + Err(TryRecvError::Lagged(_)) => continue, + Err(TryRecvError::Empty) | Err(TryRecvError::Closed) => break, + } + } + assert!( + saw, + "create_workflow_inner must publish DomainEvent::WorkflowsChanged" + ); + } } diff --git a/src/openhuman/workflows/ops_install.rs b/src/openhuman/workflows/ops_install.rs index 120def08e..666d0470d 100644 --- a/src/openhuman/workflows/ops_install.rs +++ b/src/openhuman/workflows/ops_install.rs @@ -342,6 +342,14 @@ pub async fn install_workflow_from_url( ); let stderr = parse_warnings.join("\n"); + // Notify live agent sessions so they refresh their `## Installed Skills` + // catalogue mid-conversation (see `Agent::refresh_workflows`). + let _ = crate::core::event_bus::publish_global( + crate::core::event_bus::DomainEvent::WorkflowsChanged { + reason: "install".to_string(), + }, + ); + Ok(InstallWorkflowFromUrlOutcome { url: raw_url, stdout, @@ -508,6 +516,14 @@ pub fn uninstall_workflow( std::fs::remove_dir_all(&canonical_candidate) .map_err(|e| format!("remove {} failed: {e}", canonical_candidate.display()))?; + // Notify live agent sessions to drop the removed skill from their + // `## Installed Skills` catalogue (see `Agent::refresh_workflows`). + let _ = crate::core::event_bus::publish_global( + crate::core::event_bus::DomainEvent::WorkflowsChanged { + reason: "uninstall".to_string(), + }, + ); + Ok(UninstallWorkflowOutcome { name: trimmed, removed_path: canonical_candidate.display().to_string(),