diff --git a/Cargo.lock b/Cargo.lock index 8ad1c25cc9..e58e0f4f60 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -61,6 +61,20 @@ dependencies = [ "subtle", ] +[[package]] +name = "ahash" +version = "0.8.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" +dependencies = [ + "cfg-if", + "const-random", + "getrandom 0.3.4", + "once_cell", + "version_check", + "zerocopy", +] + [[package]] name = "aho-corasick" version = "1.1.4" @@ -1085,6 +1099,26 @@ version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" +[[package]] +name = "const-random" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87e00182fe74b066627d63b85fd550ac2998d4b0bd86bfed477a0ae4c7c71359" +dependencies = [ + "const-random-macro", +] + +[[package]] +name = "const-random-macro" +version = "0.1.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9d839f2a20b0aee515dc581a6172f2321f96cab76c1a38a4c584a194955390e" +dependencies = [ + "getrandom 0.2.17", + "once_cell", + "tiny-keccak", +] + [[package]] name = "const_format" version = "0.2.36" @@ -3790,6 +3824,15 @@ dependencies = [ "libc", ] +[[package]] +name = "no-std-compat" +version = "0.4.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b93853da6d84c2e3c7d730d6473e8817692dd89be387eb01b94d7f108ecb5b8c" +dependencies = [ + "spin", +] + [[package]] name = "nom" version = "7.1.3" @@ -4255,6 +4298,9 @@ name = "once_cell" version = "1.21.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" +dependencies = [ + "portable-atomic", +] [[package]] name = "once_cell_polyfill" @@ -5539,6 +5585,35 @@ dependencies = [ "subtle", ] +[[package]] +name = "rhai" +version = "1.25.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dd4dd0f8c36625202a4ba553c416c19b719947cd2a31d1bda06126e4a5727daf" +dependencies = [ + "ahash", + "bitflags 2.11.1", + "no-std-compat", + "num-traits", + "once_cell", + "rhai_codegen", + "smallvec", + "smartstring", + "thin-vec", + "web-time", +] + +[[package]] +name = "rhai_codegen" +version = "3.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3cd3a7535e50bf36857e7be7bec276d334e8c2dfa469c2201226fd01638ea5ca" +dependencies = [ + "proc-macro2", + "quote", + "syn 2.0.117", +] + [[package]] name = "ring" version = "0.17.14" @@ -6300,6 +6375,17 @@ version = "1.15.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" +[[package]] +name = "smartstring" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3fb72c633efbaa2dd666986505016c32c3044395ceaf881518399d2f4127ee29" +dependencies = [ + "autocfg", + "static_assertions", + "version_check", +] + [[package]] name = "socket2" version = "0.6.3" @@ -6360,6 +6446,12 @@ dependencies = [ "socketioxide-core", ] +[[package]] +name = "spin" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6e63cff320ae2c57904679ba7cb63280a3dc4613885beafb148ee7bf9aa9042d" + [[package]] name = "spki" version = "0.7.3" @@ -6605,6 +6697,12 @@ dependencies = [ "windows-sys 0.61.2", ] +[[package]] +name = "thin-vec" +version = "0.2.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b0f7e269b48f0a7dd0146680fa24b50cc67fc0373f086a5b2f99bd084639b482" + [[package]] name = "thiserror" version = "1.0.69" @@ -6715,6 +6813,7 @@ dependencies = [ "async-trait", "futures", "reqwest 0.12.28", + "rhai", "rusqlite", "serde", "serde_json", diff --git a/Cargo.toml b/Cargo.toml index 4a3bdf813d..b9c72ea412 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -61,7 +61,9 @@ tinyjuice = { version = "0.1", default-features = false } # aligned to 0.40, avoiding duplicate `links = "sqlite3"` native bindings. # Durable graph checkpoints still use `SqlRunLedgerCheckpointer` until the # migration re-points those rows to the crate checkpointer. -tinyagents = { version = "1.5.0", features = ["sqlite"] } +# The `repl` feature adds the embedded Rhai `.ragsh` session runtime powering +# the `rlm` language-workflow tool (`src/openhuman/rlm/`). +tinyagents = { version = "1.5.0", features = ["sqlite", "repl"] } # TinyCortex — Rust core for the memory engine (store/chunks/tree/retrieval/ # queue/ingest/score + long tail), vendored as a git submodule and patched # below to `vendor/tinycortex`. OpenHuman's memory subsystem migrates onto this diff --git a/gitbooks/developing/architecture/agent-harness.md b/gitbooks/developing/architecture/agent-harness.md index efa6ac34e4..f32e2cc3ed 100644 --- a/gitbooks/developing/architecture/agent-harness.md +++ b/gitbooks/developing/architecture/agent-harness.md @@ -49,12 +49,12 @@ icon: layer-group ## TinyAgents crate: features & compatibility -OpenHuman pins `tinyagents = { version = "1.5.0", features = ["sqlite"] }` (see [`Cargo.toml`](../../../Cargo.toml)). The rationale, so future upgrades don't silently regress it: +OpenHuman pins `tinyagents = { version = "1.5.0", features = ["sqlite", "repl"] }` (see [`Cargo.toml`](../../../Cargo.toml)). The rationale, so future upgrades don't silently regress it: - **OpenHuman-owned providers only.** We do **not** enable any bundled provider feature. OpenHuman owns provider transport, credentials, OAuth, and billing classification, so the live model is always OpenHuman's `Provider` wrapped as [`ProviderModel`](../../../src/openhuman/tinyagents/model.rs), never an SDK-owned provider client. The `ChatModel` adapter is the seam that replaces feature-gated SDK providers. - **`sqlite` feature enabled with one native sqlite chain.** OpenHuman's root and Tauri Cargo worlds pin `rusqlite = "=0.40.0"` and patch `rusqlite` / `libsqlite3-sys` locally to avoid the upstream `cfg_select!` build break on the current toolchain. Both worlds resolve to a single `libsqlite3-sys v0.38.0` chain. Durable graph checkpoints still run through [`SqlRunLedgerCheckpointer`](../../../src/openhuman/tinyagents/checkpoint.rs) until the migration re-points those rows to the crate checkpointer. - **WhatsApp Web storage bridge.** `whatsapp-rust`'s Diesel-backed `sqlite-storage` feature links sqlite separately from rusqlite 0.40, so the optional `whatsapp-web` feature currently builds against `wacore::store::InMemoryBackend` and logs that sessions are not durable. A rusqlite-backed durable WhatsApp store is required before treating Web sessions as persistent again. -- **`repl` / expressive-language features unused.** OpenHuman drives graphs from Rust (`GraphBuilder`), not the crate's `.rag` REPL language. +- **`repl` feature enabled for language workflows; `.rag` expressive language unused.** OpenHuman still drives *graphs* from Rust (`GraphBuilder`), not the declarative `.rag` language. But the `repl` feature (the imperative Rhai `.ragsh` session runtime) is enabled to power the `rlm` language-workflow tool ([`openhuman::rlm`](../../../src/openhuman/rlm/README.md), see "Language workflows (`rlm`)" below). - **Adapter map (feature-gated SDK piece → OpenHuman replacement):** provider clients → `ProviderModel`; crate SQLite checkpointer rows not yet adopted → `SqlRunLedgerCheckpointer`; task/status stores not yet controller-canonical → OpenHuman SQL/JSON run ledgers (`running_subagents`, `workflow_runs`, `agent_teams`, `command_center`). The generic harness/graph/middleware/event primitives are used as-is. The agent harness is the runtime that turns a user message (or a webhook fire, or a cron tick) into a complete, tool-using LLM interaction. It owns the tool-call loop, sub-agent dispatch, the trigger-triage pipeline, and the hook surface around them. It does **not** own provider HTTP transport, tool implementations, prompt-section assembly, or memory storage - those are separate domains the harness composes. @@ -282,6 +282,21 @@ Each `AgentDefinition` carries an `agent_tier` field (`chat` / `reasoning` / `wo For Composio toolkits with hundreds of actions (GitHub alone has 500+), loading every action into the sub-agent's tool set balloons prompt size. The harness ranks the toolkit's actions against the parent-refined task prompt with a cheap CPU-only filter (verb detection, token overlap, verb-alignment boost) and only loads the top-ranked subset into the sub-agent. No model call, pure heuristic - fast and explainable. +## Language workflows (`rlm`) + +The fixed delegation primitives (`spawn_subagent`, `spawn_parallel_agents`, `run_workflow`) can't express *ad-hoc control flow* — "spawn N readers, dedupe their findings, verify each survivor with 3 refuters, loop until dry". The **`rlm` tool** closes that gap: it exposes TinyAgents' Rhai-backed `.ragsh` REPL (the `repl` cargo feature) so the orchestrator can author and run its own workflow scripts. + +**One tool call = one `eval_cell`.** The orchestrator's normal tool-call loop *is* the CodeAct driver loop: the model writes a Rhai cell, the cell runs against a persistent per-session namespace (top-level `let` bindings survive into the next cell via an optional `session_id`), and the structured result flows back as the tool result. Scripts reach the host only through capability functions — `tool_call`, `agent_query`, `model_query`, their `*_batched` fan-out variants, `emit`, and `answer`. + +The domain lives in [`src/openhuman/rlm/`](../../../src/openhuman/rlm/README.md): + +- **`policy.rs`** maps the autonomy tier + `tool_timeout` clamps onto a `tinyagents::ReplPolicy` (always bounded, never unbounded; `readonly` refused; `full` may raise call-count limits to a hard 2× ceiling). +- **`bridge.rs`** builds the `CapabilityRegistry`: the parent's visible tools (each re-wrapped so the **approval gate runs in the bridge** — it is *not* on the repl path, which bypasses the harness `wrap_tool` middleware), the turn's provider model, and a sub-agent capability per `allowed_subagent_ids`. Recursion/duplication hazards (`rlm`, `spawn_*`, workflow tools, `CliRpcOnly`-scoped tools) are excluded. Because `eval_cell` runs on `spawn_blocking` + `block_on`, the `agent_query` adapter re-installs the `PARENT_CONTEXT` task-local that `run_subagent` resolves. +- **`sessions.rs`** is a bounded (LRU + idle-TTL) manager of persistent sessions, one cell at a time (a concurrent call on a busy session returns a typed "busy" error). +- **`ops.rs`** runs the cell on `spawn_blocking` under a layered time bound (rhai `on_progress` deadline → `bridge_block_on` timer race → outer `tokio::timeout` backstop → harness `ToolTimeout`), wires the run-cancellation token to a fresh per-cell `ReplCancelFlag`, and maps every failure mode to a model-consumable result. + +The tool is registered for the orchestrator on `supervised`/`full` tiers only, behind the `OPENHUMAN_RLM=0` kill switch. The TinyAgents-side host-embedding support (external cancellation, live capability events) landed in that crate's `repl` feature. + ## Triage - handling external triggers When a webhook fires, a cron ticks, or a Composio event arrives, the system can't just hand it straight to the orchestrator. Most triggers are noise; some warrant a notification; only a few deserve a full agent turn. The **trigger-triage pipeline** is the gate. diff --git a/src/openhuman/about_app/catalog_data.rs b/src/openhuman/about_app/catalog_data.rs index 255efc757a..b784b5dae8 100644 --- a/src/openhuman/about_app/catalog_data.rs +++ b/src/openhuman/about_app/catalog_data.rs @@ -620,6 +620,16 @@ pub(super) const CAPABILITIES: &[Capability] = &[ status: CapabilityStatus::Beta, privacy: DERIVED_TO_BACKEND, }, + Capability { + id: "intelligence.language_workflows", + name: "Language Workflows (RLM)", + domain: "intelligence", + category: CapabilityCategory::Intelligence, + description: "The orchestrator can author and run small Rhai workflow scripts to express ad-hoc control flow over delegated work — parallel fan-out, loops, and dedup-then-verify pipelines that fixed spawn/parallel primitives cannot. Each script runs bounded and fail-closed (per-cell timeout, per-session caps on tool/model/agent calls and recursion depth), and every effectful step still passes the same approval and permission gates as a direct tool call. Progress rides the existing tool-call timeline.", + how_to: "Runs automatically when the orchestrator chooses the `rlm` tool; disable with OPENHUMAN_RLM=0 or the read-only autonomy tier", + status: CapabilityStatus::Beta, + privacy: DERIVED_TO_BACKEND, + }, Capability { id: "intelligence.agent_library", name: "Agents Library", diff --git a/src/openhuman/agent_orchestration/tools/spawn_subagent.rs b/src/openhuman/agent_orchestration/tools/spawn_subagent.rs index a90f57556b..29da6ada6a 100644 --- a/src/openhuman/agent_orchestration/tools/spawn_subagent.rs +++ b/src/openhuman/agent_orchestration/tools/spawn_subagent.rs @@ -936,7 +936,8 @@ mod tests { #[test] fn build_worker_thread_title_collapses_whitespace_and_caps_length() { - let prompt = " draft\n a very long\tplan that\nrambles ".to_string() + &"x".repeat(200); + let prompt = + " draft\n a very long\tplan that\nrambles ".to_string() + "x".repeat(200).as_str(); let title = build_worker_thread_title(&prompt); assert!(title.starts_with("draft a very long plan")); assert!(title.chars().count() <= WORKER_THREAD_TITLE_MAX_CHARS + 1); diff --git a/src/openhuman/agent_registry/agents/orchestrator/prompt.md b/src/openhuman/agent_registry/agents/orchestrator/prompt.md index 527ce2bd1c..9d1518b63d 100644 --- a/src/openhuman/agent_registry/agents/orchestrator/prompt.md +++ b/src/openhuman/agent_registry/agents/orchestrator/prompt.md @@ -103,6 +103,28 @@ their refs separate by `subagent_session_id` or `task_id` (`agentId` is only the worker type), tick or wait on each independently, and synthesise only completed outputs. Never fabricate a result for a worker that is still running or failed. +## Language workflows (rlm) + +When a task needs **ad-hoc control flow** over delegated work — loops, conditionals, a +dedup-then-verify pipeline, "spawn N, filter, then verify each survivor with M checks" — that +the fixed `spawn_parallel_agents` / `delegate_*` primitives can't express, use the `rlm` tool. +It evaluates a small **Rhai workflow cell** whose only side effects are capability calls: +`tool_call`, `agent_query`, `model_query`, and their `*_batched` fan-out variants (plus +`emit`/`answer`/`print`). + +- **One call = one cell.** Top-level `let` bindings persist within a `session_id`, so pass the + same `session_id` back to continue a namespace across calls (`let findings = …` in cell 1, + reference it in cell 2). Omit `session_id` for a fresh session; set `close_session: true` when done. +- **Prefer `rlm`** over `spawn_parallel_agents` when you need iteration, branching, or a + reduce/verify step over results — not for a single delegation (use the matching `delegate_*` + or `spawn_subagent` for that). +- **It stays inside the gates.** Every effectful inner `tool_call` still hits the approval gate; + `agent_query` only reaches sub-agents already in your allowlist. `rlm` itself, `spawn_*`, and + workflow tools are not callable from a cell. +- **It is bounded and fail-closed.** Cells have a wall-clock timeout and per-session caps on + model/tool/agent calls and recursion depth. Exceeding one returns an error you can fix and + retry in the same session; the result reports `limits_remaining` so you can plan fan-out. + ## Connecting external services When the user asks to connect a service (Gmail, Notion, WhatsApp, Calendar, Drive, etc.) or a sub-agent reports `Connection error, try to authenticate`: diff --git a/src/openhuman/mod.rs b/src/openhuman/mod.rs index 975e43f58d..e786d7754c 100644 --- a/src/openhuman/mod.rs +++ b/src/openhuman/mod.rs @@ -98,6 +98,7 @@ pub mod provider_surfaces; pub mod recall_calendar; pub mod redirect_links; pub mod referral; +pub mod rlm; pub mod routing; pub mod runtime_node; pub mod runtime_python; diff --git a/src/openhuman/rlm/README.md b/src/openhuman/rlm/README.md new file mode 100644 index 0000000000..6e687df682 --- /dev/null +++ b/src/openhuman/rlm/README.md @@ -0,0 +1,70 @@ +# `rlm` — language-based workflows (Rhai `.ragsh` REPL) + +Exposes TinyAgents' Rhai-backed `.ragsh` session runtime (the `repl` cargo +feature, `tinyagents::ReplSession`) as a first-class **`rlm` tool** so the +orchestrator model can author and execute its own workflow scripts — fan-out +over subagents, batched tool/model calls, loops, dedup/verify pipelines — and +run them deterministically, in the spirit of Claude Code Workflows and +Recursive Language Models. + +One orchestrator tool call maps to **one `eval_cell`**: the model writes a Rhai +cell, the cell runs against a persistent per-session namespace (top-level `let` +bindings survive into the next cell), and the structured result flows back as +the tool result. The orchestrator's own turn loop *is* the CodeAct driver loop. + +## Module shape + +| File | Role | +| ---- | ---- | +| `mod.rs` | Exports only (no controller schemas in v1). | +| `types.rs` | `RlmSessionId`, `RlmEvalRequest`/`RlmEvalResponse`, `RlmLimitsOverride`, `RlmCallSummary`, serde types. | +| `policy.rs` | Maps openhuman autonomy tier + `tool_timeout` clamps → `tinyagents::ReplPolicy` (fail-closed, bounded). | +| `bridge.rs` | Builds the `CapabilityRegistry<()>`: openhuman tools (approval-gated, scope-filtered) + provider models + subagents. | +| `sessions.rs` | `RlmSessionManager`: LRU + idle-TTL bounded map of persistent `ReplSession`s, keyed `:`, one cell at a time. | +| `ops.rs` | `eval_rlm_cell`: spawn_blocking + outer timeout, cancellation wiring, event forwarding, error → model-consumable result. | +| `tools.rs` | `RlmTool` (the `rlm` tool: schema, permission, scope, timeout, display). | + +## Fail-closed guarantees + +Every failure mode returns a **model-consumable tool result** — never a panic, +never a hung turn: + +- **Layered time bounds:** (1) rhai `on_progress` deadline for pure script + loops; (2) `bridge_block_on` timer race for hung capability futures; + (3) an outer `tokio::time::timeout(policy.timeout + 5s)` around + `spawn_blocking`; (4) the harness `ToolTimeout::Secs` backstop above all of + them. The tool's own timeout is always set below the harness backstop. +- **Bounded sessions:** LRU cap (16) + idle TTL (30 min); a second concurrent + call on a busy session returns a typed "session busy" error rather than + deadlocking; a poisoned/errored session is dropped, never reused. +- **Bounded work per session:** `ReplPolicy` caps on operations, output bytes, + script bytes, and per-kind call counts. `full` tier may raise call-count + limits up to a hard 2× ceiling via the tool's `limits` arg; `readonly` tier + does not get the tool at all. +- **Cancellation end-to-end:** the turn's run-cancellation token drives a + `ReplCancelFlag` watcher, so a user cancel drops an in-flight cell (script or + capability call) promptly; the session is left resumable. + +## Approval & security (bridged tools keep their own gates) + +The RLM bridge restricts callable tools to the parent turn's +`visible_tool_names`, and **excludes** recursion/duplication hazards: `rlm` +itself, `spawn_subagent`/`spawn_parallel_agents` (use `agent_query` instead), +and `run_workflow`/`await_workflow`. `ToolScope::CliRpcOnly` tools are denied. + +Approval gating is **not** on the tinyagents repl bridge path (it lives in the +harness `wrap_tool` middleware, which the REPL bypasses), so the bridge itself +invokes `ApprovalGate::intercept_audited` (+ `record_execution`) for any tool +whose `external_effect_with_args` is true, failing closed on denial. + +## Capability surface exposed to scripts + +`model_query`, `tool_call`, `agent_query`, their `*_batched` variants, `emit`, +and `answer`. Graph authoring/execution (`graph_*`) is **out of scope for v1** +(the REPL's `graph_run` returns a reference, not an execution). + +## Kill switch & rollout + +The tool is **not registered** when the autonomy tier is `readonly` or when +`OPENHUMAN_RLM=0`; default-on for `supervised`/`full`. Reverting the +registration line disables the surface without touching the domain. diff --git a/src/openhuman/rlm/bridge.rs b/src/openhuman/rlm/bridge.rs new file mode 100644 index 0000000000..1891e6116e --- /dev/null +++ b/src/openhuman/rlm/bridge.rs @@ -0,0 +1,326 @@ +//! Capability bridge: projects openhuman's tools, provider model, and +//! sub-agents into a `tinyagents` [`CapabilityRegistry`] a `.ragsh` session +//! binds against. +//! +//! Three capability kinds are wired, each keeping openhuman's own gates: +//! +//! - **Tools** — one [`RlmToolAdapter`] per visible, non-excluded, agent-scoped +//! tool. Approval is **not** on the tinyagents repl path (it lives in the +//! harness `wrap_tool` middleware the REPL bypasses), so the adapter itself +//! invokes the [`ApprovalGate`] for any tool whose `external_effect_with_args` +//! is true, failing closed on denial and recording the terminal outcome. +//! - **Model** — the turn's provider, registered under its model name so +//! `model_query(#{model: ""})` hits the real backend with usage intact. +//! - **Agents** — a [`SubagentCapability`] per entry in the parent's +//! `allowed_subagent_ids`, so `agent_query("", ...)` spawns a real +//! openhuman sub-agent through `run_subagent`. +//! +//! Recursion/duplication hazards are **excluded** from the tool surface: `rlm` +//! itself (no REPL-in-REPL), `spawn_*` (use `agent_query`), and +//! `run_workflow`/`await_workflow`. `ToolScope::CliRpcOnly` tools are excluded +//! too. + +use std::sync::Arc; + +use async_trait::async_trait; +use tinyagents::graph::subagent_node::{HarnessAgent, SubAgentInput, SubAgentOutput}; +use tinyagents::harness::events::EventSink; +use tinyagents::harness::tool::{ + Tool as TaTool, ToolCall as TaToolCall, ToolExecutionContext, ToolPolicy as TaToolPolicy, + ToolResult as TaToolResult, ToolSchema as TaToolSchema, +}; +use tinyagents::registry::CapabilityRegistry; +use tinyagents::TinyAgentsError; + +use crate::openhuman::agent::harness::definition::AgentDefinitionRegistry; +use crate::openhuman::agent::harness::fork_context::{with_parent_context, ParentExecutionContext}; +use crate::openhuman::agent::harness::subagent_runner::{run_subagent, SubagentRunOptions}; +use crate::openhuman::approval::{ + redact_args, summarize_action, ApprovalGate, ExecutionOutcome, GateOutcome, +}; +use crate::openhuman::tinyagents::model::provider_chat_model; +use crate::openhuman::tinyagents::tools::{ + execute_openhuman_tool, tool_policy_from_openhuman_tool, +}; +use crate::openhuman::tools::traits::ToolScope; +use crate::openhuman::tools::Tool as OhTool; + +/// Tools never exposed to a `.ragsh` script, to prevent recursion (a script +/// re-entering the REPL) and capability duplication (spawn/workflow primitives +/// the script models with `agent_query` instead). +fn is_excluded_tool(name: &str) -> bool { + name == "rlm" + || name == "run_workflow" + || name == "await_workflow" + || name.starts_with("spawn_") +} + +#[cfg(test)] +mod tests { + use super::is_excluded_tool; + + #[test] + fn recursion_and_duplication_hazards_are_excluded() { + for name in [ + "rlm", + "run_workflow", + "await_workflow", + "spawn_subagent", + "spawn_parallel_agents", + "spawn_async_subagent", + ] { + assert!(is_excluded_tool(name), "{name} should be excluded"); + } + for name in ["read_file", "grep", "edit_file", "web_search"] { + assert!(!is_excluded_tool(name), "{name} should be callable"); + } + } +} + +/// Builds the `CapabilityRegistry<()>` a session binds against from the parent +/// turn's execution context. +/// +/// Reads the parent's visible tool set, provider/model, and sub-agent +/// allowlist. The returned registry carries no `rlm`, `spawn_*`, or workflow +/// tools, and no `CliRpcOnly`-scoped tools. +pub(super) fn build_capability_registry(parent: &ParentExecutionContext) -> CapabilityRegistry<()> { + let mut registry = CapabilityRegistry::<()>::new(); + + // ── Model: the turn's provider, under its registered name. ── + let model = provider_chat_model( + parent.provider.clone(), + parent.model_name.clone(), + parent.temperature, + ); + registry.replace_model(parent.model_name.clone(), model); + + // ── Tools: visible, non-excluded, agent-scoped only. ── + let mut tool_count = 0usize; + for tool in parent.all_tools.iter() { + let name = tool.name(); + if !parent.visible_tool_names.is_empty() && !parent.visible_tool_names.contains(name) { + continue; + } + if is_excluded_tool(name) { + continue; + } + if matches!(tool.scope(), ToolScope::CliRpcOnly) { + continue; + } + registry.replace_tool(Arc::new(RlmToolAdapter::new( + parent.all_tools.clone(), + tool.as_ref(), + ))); + tool_count += 1; + } + + // ── Agents: one capability per allowed sub-agent id. ── + let mut agent_count = 0usize; + for agent_id in &parent.allowed_subagent_ids { + registry.replace_agent(Arc::new(SubagentCapability { + agent_id: agent_id.clone(), + parent: parent.clone(), + })); + agent_count += 1; + } + + tracing::debug!( + tools = tool_count, + agents = agent_count, + model = %parent.model_name, + "[rlm] built capability registry" + ); + registry +} + +/// A `tinyagents` tool backed by an openhuman [`Tool`](OhTool), located by name +/// in the parent's shared tool set on each call (the set is `Arc`-shared, not +/// cloned). Adds the approval gate the harness `wrap_tool` middleware would +/// otherwise apply — absent on the repl bridge path. +pub(super) struct RlmToolAdapter { + tools: Arc>>, + name: String, + description: String, + schema: TaToolSchema, + policy: TaToolPolicy, +} + +impl RlmToolAdapter { + fn new(tools: Arc>>, tool: &dyn OhTool) -> Self { + let schema = TaToolSchema { + name: tool.name().to_string(), + description: tool.description().to_string(), + parameters: tool.parameters_schema(), + format: Default::default(), + }; + Self { + name: tool.name().to_string(), + description: tool.description().to_string(), + schema, + policy: tool_policy_from_openhuman_tool(tool), + tools, + } + } + + async fn dispatch( + &self, + call: TaToolCall, + context: Option<&ToolExecutionContext>, + ) -> TaToolResult { + let found = self.tools.iter().find(|t| t.name() == self.name); + match found { + Some(tool) => gated_execute(tool.as_ref(), call, context).await, + None => { + tracing::warn!(tool = %self.name, "[rlm] bridged tool not found at call time"); + TaToolResult { + call_id: call.id, + name: call.name, + content: format!("Error: unknown tool '{}'", self.name), + raw: None, + error: Some("unknown tool".to_string()), + elapsed_ms: 0, + } + } + } + } +} + +#[async_trait] +impl TaTool<()> for RlmToolAdapter { + fn name(&self) -> &str { + &self.name + } + + fn description(&self) -> &str { + &self.description + } + + fn schema(&self) -> TaToolSchema { + self.schema.clone() + } + + fn policy(&self) -> TaToolPolicy { + self.policy.clone() + } + + async fn call(&self, _state: &(), call: TaToolCall) -> tinyagents::Result { + Ok(self.dispatch(call, None).await) + } + + async fn call_with_context( + &self, + _state: &(), + call: TaToolCall, + context: ToolExecutionContext, + ) -> tinyagents::Result { + Ok(self.dispatch(call, Some(&context)).await) + } +} + +/// Runs an openhuman tool for a `.ragsh` `tool_call`, routing any external-effect +/// tool through the [`ApprovalGate`] first (fail-closed on denial) since the +/// repl bridge sits outside the harness approval middleware. +async fn gated_execute( + tool: &dyn OhTool, + call: TaToolCall, + context: Option<&ToolExecutionContext>, +) -> TaToolResult { + if tool.external_effect_with_args(&call.arguments) { + if let Some(gate) = ApprovalGate::try_global() { + let summary = summarize_action(&call.name, &call.arguments); + let redacted = redact_args(&call.arguments); + tracing::debug!(tool = %call.name, "[rlm] external-effect tool — routing through approval gate"); + let (outcome, request_id) = + gate.intercept_audited(&call.name, &summary, redacted).await; + match outcome { + GateOutcome::Deny { reason } => { + tracing::info!(tool = %call.name, %reason, "[rlm] tool denied by approval gate"); + return TaToolResult { + content: format!("Denied by approval gate: {reason}"), + error: Some(format!("approval denied: {reason}")), + raw: None, + elapsed_ms: 0, + call_id: call.id, + name: call.name, + }; + } + GateOutcome::Allow => { + let result = execute_openhuman_tool(tool, call, context).await; + if let Some(id) = request_id { + let terminal = if result.error.is_none() { + ExecutionOutcome::Success + } else { + ExecutionOutcome::Failure + }; + gate.record_execution(&id, terminal, result.error.as_deref()); + } + return result; + } + } + } + // No global gate installed (e.g. gate disabled): fall through and + // execute — the harness-level env kill-switch owns that decision. + } + execute_openhuman_tool(tool, call, context).await +} + +/// A `.ragsh` `agent_query("", ...)` capability that spawns a real openhuman +/// sub-agent via `run_subagent`. +/// +/// Captures the parent [`ParentExecutionContext`] at bridge-build time and +/// **re-installs it** with [`with_parent_context`] before calling +/// `run_subagent`: the session's `eval_cell` runs on `spawn_blocking` + +/// `futures::executor::block_on`, which does not carry the `PARENT_CONTEXT` +/// task-local `run_subagent` resolves — without this the spawn would fail with +/// `NoParentContext`. +struct SubagentCapability { + agent_id: String, + parent: ParentExecutionContext, +} + +#[async_trait] +impl HarnessAgent for SubagentCapability { + fn name(&self) -> &str { + &self.agent_id + } + + async fn run( + &self, + input: SubAgentInput, + _events: EventSink, + ) -> tinyagents::Result { + let registry = AgentDefinitionRegistry::global().ok_or_else(|| { + TinyAgentsError::Capability("agent registry not initialised".to_string()) + })?; + let definition = registry.get(&self.agent_id).ok_or_else(|| { + TinyAgentsError::Capability(format!("agent `{}` is not registered", self.agent_id)) + })?; + // Defensive re-check of the parent allowlist (the bridge only registers + // allowed agents, but never trust that a script cannot reach further). + if !self.parent.allowed_subagent_ids.contains(&definition.id) { + return Err(TinyAgentsError::Capability(format!( + "agent `{}` is not in the parent's subagent allowlist", + definition.id + ))); + } + + let options = SubagentRunOptions { + workspace_descriptor: self.parent.workspace_descriptor.clone(), + ..Default::default() + }; + + tracing::debug!(agent = %self.agent_id, "[rlm] agent_query — spawning sub-agent"); + let outcome = with_parent_context( + self.parent.clone(), + run_subagent(definition, &input.prompt, options), + ) + .await + .map_err(|e| TinyAgentsError::Tool(format!("sub-agent `{}` failed: {e}", self.agent_id)))?; + + Ok(SubAgentOutput { + text: outcome.output, + model_calls: outcome.iterations, + ..Default::default() + }) + } +} diff --git a/src/openhuman/rlm/mod.rs b/src/openhuman/rlm/mod.rs new file mode 100644 index 0000000000..5207f2cde0 --- /dev/null +++ b/src/openhuman/rlm/mod.rs @@ -0,0 +1,20 @@ +//! `rlm` — language-based workflows over the TinyAgents Rhai `.ragsh` REPL. +//! +//! Exposes `tinyagents::ReplSession` (the `repl` feature) as a first-class +//! [`RlmTool`], so the orchestrator can author and run its own workflow scripts +//! — fan-out, batched calls, loops, dedup/verify pipelines — bounded and +//! fail-closed. See [`README.md`](https://github.com/tinyhumansai/openhuman) +//! (module `README.md`) for the design. +//! +//! Module shape (canonical): `mod.rs` exports only; logic lives in the +//! siblings. No controller schemas in v1 (`scope() = AgentOnly`). + +mod bridge; +mod ops; +mod policy; +mod sessions; +mod types; + +pub mod tools; + +pub use tools::RlmTool; diff --git a/src/openhuman/rlm/ops.rs b/src/openhuman/rlm/ops.rs new file mode 100644 index 0000000000..105be0e8fe --- /dev/null +++ b/src/openhuman/rlm/ops.rs @@ -0,0 +1,675 @@ +//! Cell evaluation orchestration: resolve the session, run the cell off the +//! async workers (`spawn_blocking`) under layered time bounds, forward +//! cancellation, and map every failure mode to a model-consumable result. + +use std::sync::{Arc, TryLockError}; +use std::time::Duration; + +use tinyagents::registry::{CapabilityRegistry, ComponentKind}; +use tinyagents::{ + ReplCallKind, ReplCallRecord, ReplCancelFlag, ReplCapabilities, ReplResult, ReplSession, + TinyAgentsError, +}; + +use crate::openhuman::agent::harness::fork_context::current_parent; +use crate::openhuman::security::live_policy; +use crate::openhuman::security::policy::AutonomyLevel; +use crate::openhuman::tinyagents::run_cancellation_context::current_run_cancellation; + +use super::bridge::build_capability_registry; +use super::policy::{resolve_policy, DEFAULT_RLM_TIMEOUT_SECS}; +use super::sessions::RlmSessionManager; +use super::types::{RlmCallSummary, RlmEvalRequest, RlmEvalResponse, RlmLimitsRemaining}; + +/// Grace added to the inner policy timeout for the outer `spawn_blocking` +/// backstop — the inner deadline should always fire first; this defends against +/// bugs in the inner layers, and if it ever fires the session is dropped. +const OUTER_TIMEOUT_GRACE: Duration = Duration::from_secs(5); + +/// Hard cap on the characters returned to the model in `stdout`, so a runaway +/// print cannot flood the context window even within the byte policy. +const MAX_RLM_RESULT_CHARS: usize = 24_000; + +/// A typed RLM failure, each mapped to a model-consumable tool result by +/// [`RlmError::message`] (with a fix hint) and [`RlmError::kind`]. +#[derive(Debug, Clone)] +pub(crate) enum RlmError { + /// The autonomy tier or policy refused the session. + Denied(String), + /// Script failed to compile or raised a runtime error. + Script(String), + /// A wall-clock deadline elapsed (script or in-flight capability call). + Timeout(String), + /// An op/output/script-size/call-count limit tripped. + LimitExceeded(String), + /// A script referenced an unknown tool/model/agent. + UnknownCapability(String), + /// Sub-agent recursion depth exceeded. + Depth(String), + /// A capability call itself returned an error. + CapabilityError(String), + /// The user cancelled the run. + Cancelled, + /// The session is already evaluating another cell. + SessionBusy(String), + /// An internal failure (join panic, poisoned session, missing context). + Internal(String), +} + +impl RlmError { + /// A stable, snake_case kind tag for logging and the tool result. + pub(crate) fn kind(&self) -> &'static str { + match self { + RlmError::Denied(_) => "denied", + RlmError::Script(_) => "script_error", + RlmError::Timeout(_) => "timeout", + RlmError::LimitExceeded(_) => "limit_exceeded", + RlmError::UnknownCapability(_) => "unknown_capability", + RlmError::Depth(_) => "recursion_depth", + RlmError::CapabilityError(_) => "capability_error", + RlmError::Cancelled => "cancelled", + RlmError::SessionBusy(_) => "session_busy", + RlmError::Internal(_) => "internal", + } + } + + /// The model-facing message, including a concrete fix hint. + pub(crate) fn message(&self) -> String { + match self { + RlmError::Denied(m) => m.clone(), + RlmError::Script(m) => { + format!("rlm script error: {m}\nFix the script and retry — reuse the same session_id to keep your bindings.") + } + RlmError::Timeout(m) => { + format!("rlm cell timed out: {m}\nSplit the work across cells, lower fan-out, or raise timeout_secs.") + } + RlmError::LimitExceeded(m) => { + format!("rlm limit exceeded: {m}\nSplit the work across cells, or (full tier) raise the relevant limit in `limits`.") + } + RlmError::UnknownCapability(m) => format!("rlm unknown capability: {m}"), + RlmError::Depth(m) => format!("rlm recursion depth exceeded: {m}"), + RlmError::CapabilityError(m) => { + format!("rlm capability call failed: {m}\nInspect the failing call's arguments and retry.") + } + RlmError::Cancelled => { + "rlm cell was cancelled by the user. The session is intact and resumable." + .to_string() + } + RlmError::SessionBusy(m) => m.clone(), + RlmError::Internal(m) => format!("rlm internal error: {m}"), + } + } +} + +/// The registered capability names, snapshotted for unknown-capability error +/// messages so the model sees the live surface it can actually call. +struct RlmAvailable { + tools: Vec, + agents: Vec, + models: Vec, +} + +impl RlmAvailable { + fn from_registry(registry: &CapabilityRegistry<()>) -> Self { + Self { + tools: registry.names(ComponentKind::Tool), + agents: registry.names(ComponentKind::Agent), + models: registry.names(ComponentKind::Model), + } + } +} + +/// The internal result of the blocking cell task, distinguishing a lock +/// contention / poison from an actual evaluation outcome. +enum CellRun { + /// The cell ran; carries the raw evaluation result. + Completed(Result), + /// Another cell holds the session lock. + Busy, + /// The session mutex was poisoned by a prior panic. + Poisoned, +} + +/// Evaluates one RLM cell against a (possibly new) session, fail-closed. +/// +/// Resolves the parent turn context, maps autonomy/timeout to a `ReplPolicy`, +/// resolves or creates the session, wires user cancellation to a fresh per-cell +/// flag, runs `eval_cell` on `spawn_blocking` under an outer `tokio::timeout` +/// backstop, and maps every outcome to [`RlmEvalResponse`] or a typed +/// [`RlmError`]. +pub(crate) async fn eval_rlm_cell(req: RlmEvalRequest) -> Result { + let parent = current_parent().ok_or_else(|| { + RlmError::Internal( + "no parent execution context — the rlm tool must run inside an agent turn".to_string(), + ) + })?; + + let tier = live_policy::current() + .map(|p| p.autonomy) + .unwrap_or(AutonomyLevel::Supervised); + let policy = + resolve_policy(tier, req.timeout_secs, req.limits.as_ref()).map_err(RlmError::Denied)?; + + let registry = build_capability_registry(&parent); + run_cell( + req, + policy, + registry, + &parent.session_id, + RlmSessionManager::global(), + ) + .await +} + +/// The parent-independent core of [`eval_rlm_cell`]: given a resolved policy and +/// a prebuilt capability registry, resolves/creates the session, runs the cell +/// under the layered time bounds and cancellation wiring, and maps the outcome. +/// +/// Split out so the full evaluation path is unit-testable with a hand-built +/// registry, without constructing a whole `ParentExecutionContext`. +async fn run_cell( + req: RlmEvalRequest, + policy: tinyagents::ReplPolicy, + registry: CapabilityRegistry<()>, + thread_scope: &str, + manager: &RlmSessionManager, +) -> Result { + let session_id = req + .session_id + .clone() + .filter(|s| !s.trim().is_empty()) + .unwrap_or_else(|| format!("rlm-{}", uuid::Uuid::new_v4())); + let key = RlmSessionManager::session_key(thread_scope, &session_id); + + // Snapshot the registry's names for error messages, then hand it to the + // session builder (used only on a fresh session; a reused session keeps its + // own registry). + let available = RlmAvailable::from_registry(®istry); + let policy_for_build = policy.clone(); + + tracing::info!( + session_key = %key, + thread_id = %thread_scope, + session_id = %session_id, + script_bytes = req.script.len(), + "[rlm] eval_rlm_cell: start" + ); + + let handle = manager.get_or_create(&key, move || { + let capabilities = ReplCapabilities::new(Arc::new(registry)); + ReplSession::<()>::new() + .with_capabilities(capabilities) + .with_policy(policy_for_build) + }); + + // Fresh per-cell cancel flag (sticky flags would leave a reused session + // permanently cancelled), wired to the turn's run-cancellation token. + let cell_flag = ReplCancelFlag::new(); + if let Some(token) = current_run_cancellation() { + let flag = cell_flag.clone(); + tokio::spawn(async move { + token.cancelled().await; + tracing::debug!("[rlm] run cancellation observed — tripping cell cancel flag"); + flag.cancel(); + }); + } + + let session = handle.session.clone(); + let script = req.script.clone(); + let flag_for_cell = cell_flag.clone(); + let outer_bound = policy + .timeout + .unwrap_or(Duration::from_secs(DEFAULT_RLM_TIMEOUT_SECS)) + + OUTER_TIMEOUT_GRACE; + + let join = tokio::task::spawn_blocking(move || { + let mut guard = match session.try_lock() { + Ok(guard) => guard, + Err(TryLockError::WouldBlock) => return CellRun::Busy, + Err(TryLockError::Poisoned(_)) => return CellRun::Poisoned, + }; + // Install this cell's fresh cancel flag before running. + guard.set_cancel_flag(flag_for_cell); + CellRun::Completed(guard.eval_cell(&script)) + }); + + let run = match tokio::time::timeout(outer_bound, join).await { + Ok(Ok(run)) => run, + Ok(Err(join_err)) => { + // The blocking task panicked — the session may be poisoned; drop it. + manager.close(&key); + tracing::error!(session_key = %key, %join_err, "[rlm] cell task panicked — session dropped"); + return Err(RlmError::Internal(format!( + "rlm cell task failed: {join_err}" + ))); + } + Err(_elapsed) => { + // The inner deadline should always fire first; if the outer backstop + // fires, the blocking thread may still be unwinding — never reuse it. + manager.close(&key); + tracing::error!( + session_key = %key, + outer_secs = outer_bound.as_secs(), + "[rlm] outer wall-clock backstop fired — session dropped (inner deadline should have fired first)" + ); + return Err(RlmError::Timeout( + "the rlm cell exceeded its outer wall-clock backstop".to_string(), + )); + } + }; + + let eval = match run { + CellRun::Completed(eval) => eval, + CellRun::Busy => { + tracing::info!(session_key = %key, "[rlm] session busy — concurrent cell rejected"); + return Err(RlmError::SessionBusy(format!( + "rlm session '{session_id}' is already evaluating a cell; wait for it to finish or use a different session_id" + ))); + } + CellRun::Poisoned => { + manager.close(&key); + return Err(RlmError::Internal( + "the rlm session was poisoned by a prior panic and has been dropped; start a fresh session".to_string(), + )); + } + }; + + let result = match eval { + Ok(result) => result, + Err(err) => { + let mapped = map_eval_error(err, &available); + tracing::info!(session_key = %key, kind = mapped.kind(), "[rlm] cell failed"); + return Err(mapped); + } + }; + + let stats = manager.finish_cell(&key, &result).unwrap_or_default(); + if req.close_session { + manager.close(&key); + } + + let response = RlmEvalResponse { + session_id, + stdout: truncate_chars(&result.stdout, MAX_RLM_RESULT_CHARS), + value: result.value.as_ref().map(|v| v.to_json()), + variables_changed: result.variables_changed, + calls: summarize_calls(&result.calls), + final_answer: result.final_answer, + elapsed_ms: result.elapsed.as_millis() as u64, + cells_used: stats.cells, + limits_remaining: RlmLimitsRemaining { + cells: policy.max_iterations.saturating_sub(stats.cells), + model_calls: policy.max_model_calls.saturating_sub(stats.model_calls), + tool_calls: policy.max_tool_calls.saturating_sub(stats.tool_calls), + agent_calls: policy.max_agent_calls.saturating_sub(stats.agent_calls), + }, + closed: req.close_session, + }; + tracing::info!( + session_key = %key, + elapsed_ms = response.elapsed_ms, + calls = response.calls.len(), + cells_used = response.cells_used, + "[rlm] eval_rlm_cell: done" + ); + Ok(response) +} + +/// Maps a raw `TinyAgentsError` from `eval_cell` onto a typed [`RlmError`], +/// enriching unknown-capability errors with the live registered names. +fn map_eval_error(err: TinyAgentsError, available: &RlmAvailable) -> RlmError { + match err { + TinyAgentsError::Validation(m) => RlmError::Script(m), + TinyAgentsError::Timeout(m) => RlmError::Timeout(m), + TinyAgentsError::LimitExceeded(m) => RlmError::LimitExceeded(m), + TinyAgentsError::Cancelled => RlmError::Cancelled, + TinyAgentsError::SubAgentDepth(n) => RlmError::Depth(format!( + "sub-agent recursion exceeded the maximum depth of {n}" + )), + TinyAgentsError::ModelNotFound(name) => RlmError::UnknownCapability(format!( + "model `{name}` is not registered. Available models: {}", + join_or_none(&available.models) + )), + TinyAgentsError::ToolNotFound(name) => RlmError::UnknownCapability(format!( + "tool `{name}` is not registered. Available tools: {}", + join_or_none(&available.tools) + )), + TinyAgentsError::Capability(m) => RlmError::UnknownCapability(format!( + "{m}. Available agents: {}", + join_or_none(&available.agents) + )), + TinyAgentsError::Tool(m) => RlmError::CapabilityError(m), + other => RlmError::Script(other.to_string()), + } +} + +/// Summarizes recorded calls into the model-visible shape — kind, name, timing, +/// success only, never raw arguments/payloads. +fn summarize_calls(calls: &[ReplCallRecord]) -> Vec { + calls + .iter() + .map(|c| RlmCallSummary { + kind: call_kind_str(c.kind).to_string(), + name: c.name.clone(), + elapsed_ms: c.elapsed.as_millis() as u64, + ok: true, + }) + .collect() +} + +fn call_kind_str(kind: ReplCallKind) -> &'static str { + match kind { + ReplCallKind::Model => "model", + ReplCallKind::Tool => "tool", + ReplCallKind::Graph => "graph", + ReplCallKind::Agent => "agent", + ReplCallKind::Emit => "emit", + } +} + +/// Renders a name list for an error message, or a placeholder when empty. +fn join_or_none(names: &[String]) -> String { + if names.is_empty() { + "(none)".to_string() + } else { + names.join(", ") + } +} + +/// Truncates `s` to at most `max` characters (char-boundary safe), appending a +/// marker when it had to cut. +fn truncate_chars(s: &str, max: usize) -> String { + if s.chars().count() <= max { + return s.to_string(); + } + let mut out: String = s.chars().take(max).collect(); + out.push_str("\n… (rlm output truncated)"); + out +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use tinyagents::harness::tool::{ + Tool as TaTool, ToolCall as TaToolCall, ToolResult as TaToolResult, ToolSchema, + }; + use tinyagents::registry::CapabilityRegistry; + use tinyagents::ReplPolicy; + + // ── Pure-function tests (mapping / helpers) ────────────────────────────── + + #[test] + fn truncate_bounds_output() { + let s = "a".repeat(100); + let out = truncate_chars(&s, 10); + assert!(out.starts_with(&"a".repeat(10))); + assert!(out.contains("truncated")); + assert!(truncate_chars("short", 10) == "short"); + } + + #[test] + fn error_kinds_are_stable() { + assert_eq!(RlmError::Cancelled.kind(), "cancelled"); + assert_eq!(RlmError::Timeout("x".into()).kind(), "timeout"); + assert!(RlmError::Script("boom".into()) + .message() + .contains("Fix the script")); + } + + #[test] + fn every_taxonomy_variant_maps_to_the_right_kind() { + let avail = RlmAvailable { + tools: vec!["read_file".into(), "grep".into()], + agents: vec!["researcher".into()], + models: vec!["chat".into()], + }; + let cases: Vec<(TinyAgentsError, &str)> = vec![ + (TinyAgentsError::Validation("parse".into()), "script_error"), + (TinyAgentsError::Timeout("t".into()), "timeout"), + (TinyAgentsError::LimitExceeded("l".into()), "limit_exceeded"), + (TinyAgentsError::Cancelled, "cancelled"), + (TinyAgentsError::SubAgentDepth(3), "recursion_depth"), + ( + TinyAgentsError::ModelNotFound("m".into()), + "unknown_capability", + ), + ( + TinyAgentsError::ToolNotFound("x".into()), + "unknown_capability", + ), + ( + TinyAgentsError::Capability("agent nope".into()), + "unknown_capability", + ), + (TinyAgentsError::Tool("boom".into()), "capability_error"), + ]; + for (err, expected) in cases { + assert_eq!(map_eval_error(err, &avail).kind(), expected); + } + // Unknown-tool errors list the live tool names for the model. + let msg = map_eval_error(TinyAgentsError::ToolNotFound("nope".into()), &avail).message(); + assert!(msg.contains("read_file") && msg.contains("grep"), "{msg}"); + } + + // ── End-to-end `run_cell` matrix (with a hand-built registry) ──────────── + + struct EchoTool; + + #[async_trait::async_trait] + impl TaTool<()> for EchoTool { + fn name(&self) -> &str { + "echo" + } + fn description(&self) -> &str { + "echoes its msg argument" + } + fn schema(&self) -> ToolSchema { + ToolSchema { + name: "echo".into(), + description: "echoes".into(), + parameters: serde_json::json!({ "type": "object" }), + format: Default::default(), + } + } + async fn call(&self, _s: &(), call: TaToolCall) -> tinyagents::Result { + let msg = call + .arguments + .get("msg") + .and_then(|v| v.as_str()) + .unwrap_or(""); + Ok(TaToolResult::text( + call.id, + call.name, + format!("echo:{msg}"), + )) + } + } + + fn echo_registry() -> CapabilityRegistry<()> { + let mut registry = CapabilityRegistry::<()>::new(); + registry + .register_tool(Arc::new(EchoTool)) + .expect("register echo"); + registry + } + + fn req(script: &str) -> RlmEvalRequest { + RlmEvalRequest { + script: script.to_string(), + ..Default::default() + } + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn happy_path_runs_a_tool_call_and_reports_usage() { + let manager = RlmSessionManager::new_for_test(); + let resp = run_cell( + req(r#"tool_call(#{ tool: "echo", arguments: #{ msg: "hi" } })"#), + ReplPolicy::default(), + echo_registry(), + "t", + &manager, + ) + .await + .expect("cell ok"); + assert_eq!(resp.cells_used, 1); + assert_eq!(resp.calls.len(), 1); + assert_eq!(resp.calls[0].kind, "tool"); + assert_eq!(resp.calls[0].name, "echo"); + assert_eq!( + resp.limits_remaining.tool_calls, + ReplPolicy::default().max_tool_calls - 1 + ); + assert!(!resp.session_id.is_empty()); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn parse_error_is_a_script_error() { + let manager = RlmSessionManager::new_for_test(); + let err = run_cell( + req("let x = ;"), + ReplPolicy::default(), + echo_registry(), + "t", + &manager, + ) + .await + .expect_err("parse error"); + assert_eq!(err.kind(), "script_error"); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn unknown_tool_lists_registered_names() { + let manager = RlmSessionManager::new_for_test(); + let err = run_cell( + req(r#"tool_call(#{ tool: "does_not_exist" })"#), + ReplPolicy::default(), + echo_registry(), + "t", + &manager, + ) + .await + .expect_err("unknown tool"); + assert_eq!(err.kind(), "unknown_capability"); + assert!(err.message().contains("echo"), "{}", err.message()); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn call_count_limit_fails_closed() { + let manager = RlmSessionManager::new_for_test(); + let policy = ReplPolicy { + max_tool_calls: 1, + ..ReplPolicy::default() + }; + let err = run_cell( + req(r#"tool_call(#{tool:"echo"}); tool_call(#{tool:"echo"})"#), + policy, + echo_registry(), + "t", + &manager, + ) + .await + .expect_err("limit"); + assert_eq!(err.kind(), "limit_exceeded"); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn script_loop_times_out_within_the_bound() { + let manager = RlmSessionManager::new_for_test(); + let policy = ReplPolicy { + timeout: Some(Duration::from_millis(200)), + max_operations: 0, + ..ReplPolicy::default() + }; + let start = std::time::Instant::now(); + let err = run_cell(req("loop {}"), policy, echo_registry(), "t", &manager) + .await + .expect_err("timeout"); + assert_eq!(err.kind(), "timeout"); + assert!( + start.elapsed() < Duration::from_secs(4), + "took {:?}", + start.elapsed() + ); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn oversized_output_fails_closed() { + let manager = RlmSessionManager::new_for_test(); + let policy = ReplPolicy { + max_output_bytes: 64, + ..ReplPolicy::default() + }; + let err = run_cell( + req(r#"for i in 0..1000 { print("xxxxxxxxxxxxxxxx"); }"#), + policy, + echo_registry(), + "t", + &manager, + ) + .await + .expect_err("output limit"); + assert_eq!(err.kind(), "limit_exceeded"); + } + + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn namespace_persists_and_close_session_drops_it() { + let manager = RlmSessionManager::new_for_test(); + let first = run_cell( + RlmEvalRequest { + script: "let acc = 10; acc".into(), + session_id: Some("keep".into()), + ..Default::default() + }, + ReplPolicy::default(), + echo_registry(), + "t", + &manager, + ) + .await + .expect("cell 1"); + assert_eq!(first.session_id, "keep"); + + let second = run_cell( + RlmEvalRequest { + script: "acc + 5".into(), + session_id: Some("keep".into()), + close_session: true, + ..Default::default() + }, + ReplPolicy::default(), + echo_registry(), + "t", + &manager, + ) + .await + .expect("cell 2"); + assert_eq!(second.value, Some(serde_json::json!(15))); + assert!(second.closed); + assert_eq!(manager.len(), 0, "close_session dropped the session"); + } + + #[tokio::test(flavor = "current_thread")] + async fn concurrent_cell_on_a_busy_session_is_rejected() { + let manager = RlmSessionManager::new_for_test(); + let key = RlmSessionManager::session_key("t", "busy"); + let handle = manager.get_or_create(&key, || tinyagents::ReplSession::<()>::new()); + // Hold the session lock to simulate an in-flight cell. + let _guard = handle.session.lock().unwrap(); + + let res = run_cell( + RlmEvalRequest { + script: "1".into(), + session_id: Some("busy".into()), + ..Default::default() + }, + ReplPolicy::default(), + CapabilityRegistry::new(), + "t", + &manager, + ) + .await; + assert!(matches!(res, Err(RlmError::SessionBusy(_))), "got {res:?}"); + } +} diff --git a/src/openhuman/rlm/policy.rs b/src/openhuman/rlm/policy.rs new file mode 100644 index 0000000000..d0780f2367 --- /dev/null +++ b/src/openhuman/rlm/policy.rs @@ -0,0 +1,203 @@ +//! Maps openhuman's autonomy tier and tool-timeout clamps onto a +//! [`tinyagents::ReplPolicy`], fail-closed. +//! +//! Every RLM session is bounded: the resolved policy is always based on the +//! conservative crate default, the wall-clock timeout is always set (never +//! unbounded), the `readonly` tier is refused outright, and caller-supplied +//! limit overrides are clamped — a `full`-tier caller may raise call-count +//! limits up to a hard 2× ceiling, everyone else may only tighten them. + +use std::time::Duration; + +use tinyagents::ReplPolicy; + +use crate::openhuman::security::policy::AutonomyLevel; +use crate::openhuman::tool_timeout; + +use super::types::RlmLimitsOverride; + +/// Default per-cell wall-clock timeout when the caller does not specify one. +/// Matches the `rlm` tool's default `ToolTimeout::Secs(300)` so the inner +/// deadline and the harness backstop agree. +pub const DEFAULT_RLM_TIMEOUT_SECS: u64 = 300; + +/// Hard upper bound on batched concurrency, regardless of tier or override. +pub const MAX_RLM_CONCURRENCY: usize = 8; + +/// Ceiling multiplier applied to the crate default call-count limits when a +/// `full`-tier caller raises them via the tool's `limits` argument. +pub const LIMIT_CEILING_MULTIPLIER: usize = 2; + +/// Maximum sub-agent recursion depth an RLM session may drive. Mirrors the +/// harness `MAX_SPAWN_DEPTH` (kept in lock-step by intent; a divergence only +/// tightens one side, never opens an unbounded path). +pub const RLM_MAX_DEPTH: usize = 3; + +/// Resolves a [`ReplPolicy`] for a session opened at autonomy `tier`, with an +/// optional caller `timeout_secs` and `overrides`. +/// +/// Returns `Err(reason)` — a model-consumable string — when the tier forbids +/// RLM entirely (`readonly`). The `reason` is surfaced verbatim to the model. +pub fn resolve_policy( + tier: AutonomyLevel, + timeout_secs: Option, + overrides: Option<&RlmLimitsOverride>, +) -> Result { + if tier == AutonomyLevel::ReadOnly { + return Err( + "the `rlm` tool is disabled at the read-only autonomy tier (it can drive tools, \ + models, and sub-agents); raise autonomy to `supervised` or `full` to use it" + .to_string(), + ); + } + let allow_raise = tier == AutonomyLevel::Full; + + let base = ReplPolicy::default(); + + // Wall-clock timeout: clamp the caller's request to [1, 3600] and cap it, + // defaulting to DEFAULT_RLM_TIMEOUT_SECS. Never unbounded. + let secs = + tool_timeout::explicit_call_timeout_secs(timeout_secs, tool_timeout::MAX_TIMEOUT_SECS) + .unwrap_or(DEFAULT_RLM_TIMEOUT_SECS); + + let ov = overrides.cloned().unwrap_or_default(); + let policy = ReplPolicy { + timeout: Some(Duration::from_secs(secs)), + max_depth: base.max_depth.min(RLM_MAX_DEPTH), + max_model_calls: clamp_limit(base.max_model_calls, ov.max_model_calls, allow_raise), + max_agent_calls: clamp_limit(base.max_agent_calls, ov.max_agent_calls, allow_raise), + max_tool_calls: clamp_limit(base.max_tool_calls, ov.max_tool_calls, allow_raise), + max_concurrency: clamp_concurrency(base.max_concurrency, ov.max_concurrency, allow_raise), + ..base + }; + + tracing::debug!( + tier = ?tier, + timeout_secs = secs, + max_model_calls = policy.max_model_calls, + max_agent_calls = policy.max_agent_calls, + max_tool_calls = policy.max_tool_calls, + max_concurrency = policy.max_concurrency, + max_depth = policy.max_depth, + "[rlm] resolved ReplPolicy" + ); + Ok(policy) +} + +/// Clamps a caller-requested call-count limit. +/// +/// - No request → the crate default. +/// - `allow_raise` (full tier) → clamped to `[1, default * ceiling]`. +/// - Otherwise (supervised) → may only *tighten*: clamped to `[1, default]`. +fn clamp_limit(default: usize, requested: Option, allow_raise: bool) -> usize { + match requested { + None => default, + Some(n) => { + let ceiling = if allow_raise { + default.saturating_mul(LIMIT_CEILING_MULTIPLIER) + } else { + default + }; + n.clamp(1, ceiling.max(1)) + } + } +} + +/// Clamps batched concurrency to `[1, MAX_RLM_CONCURRENCY]`, and to the default +/// unless the tier allows raising it. +fn clamp_concurrency(default: usize, requested: Option, allow_raise: bool) -> usize { + let ceiling = if allow_raise { + MAX_RLM_CONCURRENCY + } else { + default.min(MAX_RLM_CONCURRENCY) + }; + match requested { + None => default.min(MAX_RLM_CONCURRENCY), + Some(n) => n.clamp(1, ceiling.max(1)), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn readonly_tier_is_refused() { + let err = + resolve_policy(AutonomyLevel::ReadOnly, None, None).expect_err("readonly refused"); + assert!( + err.contains("read-only"), + "reason should name the tier: {err}" + ); + } + + #[test] + fn default_timeout_when_unset_and_bounded_when_set() { + let p = resolve_policy(AutonomyLevel::Supervised, None, None).expect("policy"); + assert_eq!( + p.timeout, + Some(Duration::from_secs(DEFAULT_RLM_TIMEOUT_SECS)) + ); + + // A caller request is clamped to [1, 3600]. + let p = resolve_policy(AutonomyLevel::Supervised, Some(10_000), None).expect("policy"); + assert_eq!(p.timeout, Some(Duration::from_secs(3600))); + let p = resolve_policy(AutonomyLevel::Supervised, Some(0), None).expect("policy"); + assert_eq!( + p.timeout, + Some(Duration::from_secs(DEFAULT_RLM_TIMEOUT_SECS)) + ); + } + + #[test] + fn supervised_may_only_tighten_limits() { + let base = ReplPolicy::default(); + let ov = RlmLimitsOverride { + max_tool_calls: Some(base.max_tool_calls * 10), // request a raise + ..Default::default() + }; + let p = resolve_policy(AutonomyLevel::Supervised, None, Some(&ov)).expect("policy"); + assert_eq!( + p.max_tool_calls, base.max_tool_calls, + "supervised cannot raise above the default" + ); + + let ov = RlmLimitsOverride { + max_tool_calls: Some(1), + ..Default::default() + }; + let p = resolve_policy(AutonomyLevel::Supervised, None, Some(&ov)).expect("policy"); + assert_eq!(p.max_tool_calls, 1, "supervised may tighten"); + } + + #[test] + fn full_may_raise_up_to_the_ceiling() { + let base = ReplPolicy::default(); + let ov = RlmLimitsOverride { + max_model_calls: Some(base.max_model_calls * 100), // way over ceiling + ..Default::default() + }; + let p = resolve_policy(AutonomyLevel::Full, None, Some(&ov)).expect("policy"); + assert_eq!( + p.max_model_calls, + base.max_model_calls * LIMIT_CEILING_MULTIPLIER, + "full is capped at the 2x ceiling" + ); + } + + #[test] + fn concurrency_is_hard_capped() { + let ov = RlmLimitsOverride { + max_concurrency: Some(1000), + ..Default::default() + }; + let p = resolve_policy(AutonomyLevel::Full, None, Some(&ov)).expect("policy"); + assert_eq!(p.max_concurrency, MAX_RLM_CONCURRENCY); + } + + #[test] + fn depth_respects_the_spawn_ceiling() { + let p = resolve_policy(AutonomyLevel::Full, None, None).expect("policy"); + assert!(p.max_depth <= RLM_MAX_DEPTH); + } +} diff --git a/src/openhuman/rlm/sessions.rs b/src/openhuman/rlm/sessions.rs new file mode 100644 index 0000000000..65eabfbdfa --- /dev/null +++ b/src/openhuman/rlm/sessions.rs @@ -0,0 +1,317 @@ +//! Session manager: a process-global, bounded map of persistent `.ragsh` +//! sessions. +//! +//! Sessions are keyed `:` so parallel chats never share +//! a namespace. Each session is `Send` but `eval_cell` takes `&mut self`, so a +//! session runs **one cell at a time**, serialized by a per-session +//! [`std::sync::Mutex`]; a second concurrent call on a busy session sees a +//! `try_lock` failure and returns a typed "busy" error rather than queueing +//! (see [`super::ops`]). The map is bounded fail-closed: an idle-TTL sweep plus +//! an LRU cap keep the number of live namespaces finite. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex, OnceLock, PoisonError}; +use std::time::{Duration, Instant}; + +use tinyagents::{ReplCallKind, ReplCancelFlag, ReplResult, ReplSession}; + +/// Maximum number of live sessions before the least-recently-used one is +/// evicted on the next access. +pub(super) const MAX_SESSIONS: usize = 16; + +/// Idle time after which a session is evicted on the next access. +pub(super) const IDLE_TTL: Duration = Duration::from_secs(30 * 60); + +/// A live session and its bookkeeping. +struct SessionSlot { + /// The session, behind a `Mutex` so only one cell runs at a time; a busy + /// `try_lock` maps to a typed "session busy" error. + session: Arc>, + /// The session's cancel flag (a clone of the one installed on the session), + /// so a run-cancellation watcher can abort an in-flight cell. + cancel: ReplCancelFlag, + /// Last time the session was accessed, for idle-TTL and LRU eviction. + last_access: Instant, + /// Cells evaluated so far (for `cells_used`). + cells: usize, + /// Cumulative capability-call counts (for `limits_remaining`, since the + /// crate does not expose a session's internal counters). + model_calls: usize, + tool_calls: usize, + agent_calls: usize, +} + +/// A handle to a resolved session: the shared session and its cancel flag, plus +/// whether it was newly created this call. +pub(super) struct SlotHandle { + pub(super) session: Arc>, + pub(super) cancel: ReplCancelFlag, + pub(super) fresh: bool, +} + +/// A snapshot of a session's cumulative usage after a cell, used to compute +/// `limits_remaining`. +#[derive(Debug, Clone, Copy, Default)] +pub(super) struct CellStats { + pub(super) cells: usize, + pub(super) model_calls: usize, + pub(super) tool_calls: usize, + pub(super) agent_calls: usize, +} + +/// The process-global session manager. +pub(super) struct RlmSessionManager { + inner: Mutex>, +} + +static MANAGER: OnceLock = OnceLock::new(); + +impl RlmSessionManager { + /// Returns the process-global manager, initialising it on first use. + pub(super) fn global() -> &'static RlmSessionManager { + MANAGER.get_or_init(|| RlmSessionManager { + inner: Mutex::new(HashMap::new()), + }) + } + + /// Composes the map key from the parent thread scope and the session id. + pub(super) fn session_key(thread_scope: &str, session_id: &str) -> String { + format!("{thread_scope}:{session_id}") + } + + fn lock(&self) -> std::sync::MutexGuard<'_, HashMap> { + self.inner.lock().unwrap_or_else(PoisonError::into_inner) + } + + /// Resolves the session for `key`, building a fresh one with `build` if + /// absent. Runs an eviction sweep first so idle/over-cap sessions are + /// reclaimed. The `build` closure must produce a session that already has + /// its cancel flag installed (read back via `cancel_flag()`). + pub(super) fn get_or_create( + &self, + key: &str, + build: impl FnOnce() -> ReplSession, + ) -> SlotHandle { + let mut map = self.lock(); + Self::evict(&mut map, key); + let now = Instant::now(); + if let Some(slot) = map.get_mut(key) { + slot.last_access = now; + tracing::debug!(session_key = key, "[rlm] reusing existing session"); + return SlotHandle { + session: slot.session.clone(), + cancel: slot.cancel.clone(), + fresh: false, + }; + } + let session = build(); + let cancel = session.cancel_flag(); + let session = Arc::new(Mutex::new(session)); + map.insert( + key.to_string(), + SessionSlot { + session: session.clone(), + cancel: cancel.clone(), + last_access: now, + cells: 0, + model_calls: 0, + tool_calls: 0, + agent_calls: 0, + }, + ); + tracing::debug!( + session_key = key, + live_sessions = map.len(), + "[rlm] created new session" + ); + SlotHandle { + session, + cancel, + fresh: true, + } + } + + /// Records a completed cell against `key`: bumps the cell count, accumulates + /// capability-call counts from `result`, and returns the cumulative + /// snapshot. Returns `None` if the slot was evicted mid-cell. + pub(super) fn finish_cell(&self, key: &str, result: &ReplResult) -> Option { + let mut map = self.lock(); + let slot = map.get_mut(key)?; + slot.cells += 1; + slot.last_access = Instant::now(); + for call in &result.calls { + match call.kind { + ReplCallKind::Model => slot.model_calls += 1, + ReplCallKind::Tool => slot.tool_calls += 1, + ReplCallKind::Agent => slot.agent_calls += 1, + ReplCallKind::Graph | ReplCallKind::Emit => {} + } + } + Some(CellStats { + cells: slot.cells, + model_calls: slot.model_calls, + tool_calls: slot.tool_calls, + agent_calls: slot.agent_calls, + }) + } + + /// Drops the session for `key` (explicit close, or on a poisoned/errored + /// session that must never be reused). + pub(super) fn close(&self, key: &str) { + if self.lock().remove(key).is_some() { + tracing::debug!(session_key = key, "[rlm] closed session"); + } + } + + /// Evicts idle (past [`IDLE_TTL`]) sessions, then — if still at or above the + /// [`MAX_SESSIONS`] cap and `incoming` is not already present — the + /// least-recently-used session, so inserting `incoming` stays within cap. + fn evict(map: &mut HashMap, incoming: &str) { + let now = Instant::now(); + map.retain(|k, slot| { + let keep = now.duration_since(slot.last_access) < IDLE_TTL; + if !keep { + tracing::debug!(session_key = %k, "[rlm] evicting idle session (TTL)"); + } + keep + }); + if map.contains_key(incoming) || map.len() < MAX_SESSIONS { + return; + } + // Drop the least-recently-used slot to make room for the incoming one. + if let Some(lru_key) = map + .iter() + .min_by_key(|(_, slot)| slot.last_access) + .map(|(k, _)| k.clone()) + { + map.remove(&lru_key); + tracing::debug!(session_key = %lru_key, "[rlm] evicting LRU session (cap)"); + } + } + + /// Number of live sessions (for tests/diagnostics). + #[cfg(test)] + pub(super) fn len(&self) -> usize { + self.lock().len() + } + + /// A standalone (non-global) manager instance, so tests are isolated from + /// the process-global singleton and from each other. + #[cfg(test)] + pub(super) fn new_for_test() -> Self { + RlmSessionManager { + inner: Mutex::new(HashMap::new()), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use tinyagents::{ReplPolicy, ReplSession, ReplValue}; + + fn build_session() -> ReplSession { + ReplSession::<()>::new() + } + + #[test] + fn namespace_persists_across_cells_in_one_session() { + let manager = RlmSessionManager::new_for_test(); + let key = RlmSessionManager::session_key("t", "s1"); + + let handle = manager.get_or_create(&key, build_session); + assert!(handle.fresh); + handle + .session + .lock() + .unwrap() + .eval_cell("let n = 7;") + .expect("cell 1"); + + // Reusing the same key returns the same (non-fresh) session, so the + // binding is still visible. + let handle = manager.get_or_create(&key, build_session); + assert!(!handle.fresh); + let result = handle + .session + .lock() + .unwrap() + .eval_cell("n + 1") + .expect("cell 2"); + assert_eq!(result.value, Some(ReplValue::Int(8))); + } + + #[test] + fn distinct_session_keys_are_isolated() { + let manager = RlmSessionManager::new_for_test(); + let a = manager.get_or_create(&RlmSessionManager::session_key("t", "a"), build_session); + a.session.lock().unwrap().eval_cell("let x = 1;").unwrap(); + + // A different key has its own namespace, so `x` is undefined there. + let b = manager.get_or_create(&RlmSessionManager::session_key("t", "b"), build_session); + assert!(b.session.lock().unwrap().eval_cell("x").is_err()); + } + + #[test] + fn thread_scope_isolates_the_same_session_id() { + let manager = RlmSessionManager::new_for_test(); + let k1 = RlmSessionManager::session_key("thread-1", "shared"); + let k2 = RlmSessionManager::session_key("thread-2", "shared"); + assert_ne!(k1, k2); + manager.get_or_create(&k1, build_session); + let h2 = manager.get_or_create(&k2, build_session); + assert!( + h2.fresh, + "same session_id under a different thread is a fresh namespace" + ); + } + + #[test] + fn lru_cap_bounds_the_number_of_live_sessions() { + let manager = RlmSessionManager::new_for_test(); + for i in 0..(MAX_SESSIONS + 5) { + manager.get_or_create( + &RlmSessionManager::session_key("t", &format!("s{i}")), + build_session, + ); + } + assert!( + manager.len() <= MAX_SESSIONS, + "live sessions {} exceeded the cap {MAX_SESSIONS}", + manager.len() + ); + } + + #[test] + fn close_drops_the_session() { + let manager = RlmSessionManager::new_for_test(); + let key = RlmSessionManager::session_key("t", "closeme"); + manager.get_or_create(&key, build_session); + assert_eq!(manager.len(), 1); + manager.close(&key); + assert_eq!(manager.len(), 0); + // Re-creating after close is a fresh session. + assert!(manager.get_or_create(&key, build_session).fresh); + } + + #[test] + fn finish_cell_accumulates_and_bounds_are_reported() { + let manager = RlmSessionManager::new_for_test(); + let key = RlmSessionManager::session_key("t", "stats"); + let policy = ReplPolicy::default(); + let handle = manager.get_or_create(&key, || { + ReplSession::<()>::new().with_policy(policy.clone()) + }); + let result = handle + .session + .lock() + .unwrap() + .eval_cell("emit(\"hi\"); 1") + .expect("cell"); + let stats = manager.finish_cell(&key, &result).expect("stats"); + assert_eq!(stats.cells, 1); + // `emit` is not a model/tool/agent call, so those stay zero. + assert_eq!(stats.tool_calls, 0); + assert_eq!(stats.model_calls, 0); + } +} diff --git a/src/openhuman/rlm/tools.rs b/src/openhuman/rlm/tools.rs new file mode 100644 index 0000000000..f31bbc7b71 --- /dev/null +++ b/src/openhuman/rlm/tools.rs @@ -0,0 +1,255 @@ +//! The first-class `rlm` tool — the orchestrator's language-workflow surface. +//! +//! One tool call evaluates one Rhai cell against a persistent session +//! namespace. All effectful work a cell performs goes through the bridged inner +//! tools/models/sub-agents (each carrying its own approval/permission gates in +//! [`super::bridge`]), so the `rlm` tool itself declares no external effect. + +use async_trait::async_trait; +use serde_json::{json, Value}; + +use crate::openhuman::tools::traits::{PermissionLevel, Tool, ToolResult, ToolScope, ToolTimeout}; + +use super::policy::DEFAULT_RLM_TIMEOUT_SECS; +use super::types::RlmEvalRequest; + +/// The `rlm` language-workflow tool. Stateless: it resolves the parent turn +/// context, autonomy tier, and cancellation per call. +pub struct RlmTool; + +impl RlmTool { + /// Builds the tool. + pub fn new() -> Self { + Self + } +} + +impl Default for RlmTool { + fn default() -> Self { + Self::new() + } +} + +#[async_trait] +impl Tool for RlmTool { + fn name(&self) -> &str { + "rlm" + } + + fn description(&self) -> &str { + "Author and run a small Rhai workflow *cell* to orchestrate ad-hoc control flow — \ +fan-out, loops, dedup/verify pipelines — that fixed spawn/parallel tools cannot express. \ +One call evaluates one cell; top-level `let` bindings persist within a `session_id` so a \ +later cell continues the same namespace.\n\ +\n\ +In-cell built-ins:\n\ +• tool_call(#{tool, arguments}) — call one of your visible tools; returns its content.\n\ +• agent_query(#{agent, prompt}) — spawn a sub-agent (by id from your allowlist); returns its text.\n\ +• model_query(#{model, system?, prompt?}) — one model completion; returns the text.\n\ +• tool_call_batched([...]) / agent_query_batched([...]) / model_query_batched([...]) — bounded-concurrency fan-out; each returns an array aligned with its input.\n\ +• emit(name, #{..}) — record a progress event. answer(text) — set the cell's final answer.\n\ +• print(x) — captured into stdout. Ordinary Rhai: let, if, for/while, arrays #[..], maps #{..}.\n\ +\n\ +Bounds (fail-closed): per-cell wall-clock timeout, and per-session caps on model/tool/agent \ +calls, iterations, output bytes, and recursion depth. Exceeding one returns an error you can \ +fix and retry in the same session. Excluded from cells: `rlm` itself, `spawn_*`, and workflow tools.\n\ +\n\ +Example — parallel fan-out over sub-agents, then reduce:\n\ +```\n\ +let topics = [\"auth\", \"billing\", \"search\"];\n\ +let findings = agent_query_batched(topics.map(|t| #{ agent: \"researcher\", prompt: `Investigate ${t}` }));\n\ +for f in findings { print(f); }\n\ +answer(`Reviewed ${findings.len()} areas`);\n\ +```\n\ +\n\ +Example — batched tool calls, keep only the hits:\n\ +```\n\ +let files = [\"a.rs\", \"b.rs\", \"c.rs\"];\n\ +let reads = tool_call_batched(files.map(|p| #{ tool: \"read_file\", arguments: #{ path: p } }));\n\ +let hits = [];\n\ +for r in reads { if r.ok && r.content.contains(\"TODO\") { hits.push(r.content); } }\n\ +hits\n\ +```\n\ +\n\ +Prefer `rlm` over spawn_parallel_agents when you need loops, conditionals, or a dedup/verify \ +pipeline over results. Pass a `session_id` to continue a prior cell's bindings; set \ +`close_session: true` when done." + } + + fn parameters_schema(&self) -> Value { + json!({ + "type": "object", + "properties": { + "script": { + "type": "string", + "description": "Rhai workflow cell to evaluate. Top-level `let` bindings persist within a session_id." + }, + "session_id": { + "type": "string", + "description": "Continue a prior RLM session's namespace; omit for a fresh session (its generated id is returned)." + }, + "timeout_secs": { + "type": "integer", + "minimum": 1, + "maximum": 3600, + "description": "Per-cell wall-clock timeout in seconds (default 300)." + }, + "limits": { + "type": "object", + "description": "Optional per-session limit overrides (clamped; only the `full` autonomy tier may raise above defaults).", + "properties": { + "max_tool_calls": { "type": "integer" }, + "max_agent_calls": { "type": "integer" }, + "max_model_calls": { "type": "integer" }, + "max_concurrency": { "type": "integer" } + } + }, + "close_session": { + "type": "boolean", + "description": "Close (drop) the session after this cell." + } + }, + "required": ["script"] + }) + } + + fn permission_level(&self) -> PermissionLevel { + // Matches spawn_subagent: an orchestration primitive that can drive + // effectful inner tools (each of which re-gates itself). + PermissionLevel::Execute + } + + fn scope(&self) -> ToolScope { + // Orchestrator surface only; not exposed over CLI/RPC in v1. + ToolScope::AgentOnly + } + + fn timeout_policy(&self, args: &Value) -> ToolTimeout { + // Always an explicit bound (never Inherit/Unbounded): a legitimate + // fan-out outlives the default inherit budget, but must still be capped. + let requested = args.get("timeout_secs").and_then(Value::as_u64); + let secs = crate::openhuman::tool_timeout::explicit_call_timeout_secs( + requested, + crate::openhuman::tool_timeout::MAX_TIMEOUT_SECS, + ) + .unwrap_or(DEFAULT_RLM_TIMEOUT_SECS); + ToolTimeout::Secs(secs) + } + + fn display_label(&self, _args: &Value) -> Option { + Some("running RLM workflow".to_string()) + } + + fn display_detail(&self, args: &Value) -> Option { + let script = args.get("script").and_then(Value::as_str)?; + let first = script + .lines() + .find(|line| !line.trim().is_empty()) + .unwrap_or("") + .trim(); + if first.chars().count() > 80 { + Some(format!("{}…", first.chars().take(79).collect::())) + } else { + Some(first.to_string()) + } + } + + async fn execute(&self, args: Value) -> anyhow::Result { + let req: RlmEvalRequest = match serde_json::from_value(args) { + Ok(req) => req, + Err(err) => { + return Ok(ToolResult::error(format!( + "rlm: invalid arguments: {err}. Required: `script` (string). \ + Optional: session_id, timeout_secs (1–3600), limits, close_session." + ))); + } + }; + if req.script.trim().is_empty() { + return Ok(ToolResult::error( + "rlm: `script` is required and must be a non-empty Rhai cell.", + )); + } + + match super::ops::eval_rlm_cell(req).await { + Ok(response) => match serde_json::to_value(&response) { + Ok(value) => Ok(ToolResult::json(value)), + Err(err) => Ok(ToolResult::error(format!( + "rlm: internal error rendering result: {err}" + ))), + }, + Err(err) => { + log::info!("[rlm] tool returning error result (kind={})", err.kind()); + Ok(ToolResult::error(err.message())) + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn metadata_is_stable() { + let tool = RlmTool::new(); + assert_eq!(tool.name(), "rlm"); + assert_eq!(tool.permission_level(), PermissionLevel::Execute); + assert!(matches!(tool.scope(), ToolScope::AgentOnly)); + assert!(!tool.external_effect()); + } + + #[test] + fn timeout_is_always_bounded() { + let tool = RlmTool::new(); + assert_eq!( + tool.timeout_policy(&json!({})), + ToolTimeout::Secs(DEFAULT_RLM_TIMEOUT_SECS) + ); + assert_eq!( + tool.timeout_policy(&json!({ "timeout_secs": 42 })), + ToolTimeout::Secs(42) + ); + // Out-of-range requests are clamped, never unbounded. + assert_eq!( + tool.timeout_policy(&json!({ "timeout_secs": 100000 })), + ToolTimeout::Secs(3600) + ); + } + + #[test] + fn display_detail_elides_first_script_line() { + let tool = RlmTool::new(); + let detail = tool + .display_detail(&json!({ "script": "\n\nlet x = 1;\nlet y = 2;" })) + .expect("detail"); + assert_eq!(detail, "let x = 1;"); + } + + #[tokio::test] + async fn empty_script_is_a_model_consumable_error() { + let tool = RlmTool::new(); + let result = tool + .execute(json!({ "script": " " })) + .await + .expect("ok result"); + assert!(result.is_error); + } + + #[tokio::test] + async fn missing_script_is_a_model_consumable_error() { + let tool = RlmTool::new(); + // No `script` field at all → invalid arguments, not a panic. + let result = tool + .execute(json!({ "session_id": "x" })) + .await + .expect("ok result"); + assert!(result.is_error); + assert!(result.output_for_llm(false).contains("script")); + } + + #[test] + fn schema_requires_script() { + let schema = RlmTool::new().parameters_schema(); + assert_eq!(schema["required"], json!(["script"])); + } +} diff --git a/src/openhuman/rlm/types.rs b/src/openhuman/rlm/types.rs new file mode 100644 index 0000000000..25c94dbfbd --- /dev/null +++ b/src/openhuman/rlm/types.rs @@ -0,0 +1,129 @@ +//! Serde domain types for the `rlm` language-workflow tool. +//! +//! These cross the tool boundary (parsed from the `rlm` tool call, and rendered +//! back into its [`ToolResult`](crate::openhuman::tools::ToolResult)). The +//! runtime types that wire a session to openhuman's tools/models/subagents live +//! in [`super::bridge`] and [`super::sessions`]; the tinyagents-side limit and +//! result types live in that crate. + +use serde::{Deserialize, Serialize}; + +/// A caller-supplied RLM session id. Continuing a prior `session_id` reuses that +/// session's persistent namespace (`let` bindings survive across cells); an +/// absent id starts a fresh session. Namespaces are additionally scoped by the +/// parent thread inside [`super::sessions`], so two chats never collide. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct RlmSessionId(pub String); + +impl RlmSessionId { + /// The session id as a string slice. + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl std::fmt::Display for RlmSessionId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.0) + } +} + +/// Per-call limit overrides a caller may pass in the `rlm` tool's `limits` +/// argument. Each is clamped in [`super::policy`] to a hard ceiling (never +/// unbounded); a `full`-tier caller may raise them, others are capped at the +/// conservative defaults. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct RlmLimitsOverride { + /// Requested `max_tool_calls` for the session. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_tool_calls: Option, + /// Requested `max_agent_calls` for the session. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_agent_calls: Option, + /// Requested `max_model_calls` for the session. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_model_calls: Option, + /// Requested `max_concurrency` for batched calls (hard-capped at 8). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub max_concurrency: Option, +} + +/// A parsed `rlm` tool call — one cell to evaluate against a session. +#[derive(Debug, Clone, Default, Deserialize)] +pub struct RlmEvalRequest { + /// The Rhai workflow cell to evaluate. + pub script: String, + /// Continue a prior session's namespace; `None` starts a fresh session. + #[serde(default)] + pub session_id: Option, + /// Per-cell wall-clock timeout in seconds (clamped 1–3600). + #[serde(default)] + pub timeout_secs: Option, + /// Optional per-session limit overrides. + #[serde(default)] + pub limits: Option, + /// Close (drop) the session after this cell. + #[serde(default)] + pub close_session: bool, +} + +/// A summarized capability call a cell performed — kind, name, timing, and +/// success only. Never the raw arguments or payloads (those stay at `debug` log +/// level and on the live event stream), so a model-visible result cannot leak a +/// large or sensitive payload back into the context window. +#[derive(Debug, Clone, PartialEq, Serialize)] +pub struct RlmCallSummary { + /// `model` | `tool` | `agent` | `graph` | `emit`. + pub kind: String, + /// The capability or event name. + pub name: String, + /// Wall-clock time the call took, in milliseconds. + pub elapsed_ms: u64, + /// Whether the call was recorded (calls that errored abort the cell, so a + /// recorded call is a completed one). + pub ok: bool, +} + +/// The remaining per-session budget after a cell, surfaced so the model can plan +/// how much more fan-out it can do before splitting work across sessions. +#[derive(Debug, Clone, Default, PartialEq, Serialize)] +pub struct RlmLimitsRemaining { + /// Cells left before `max_iterations`. + pub cells: usize, + /// `model_query` calls left. + pub model_calls: usize, + /// `tool_call` calls left. + pub tool_calls: usize, + /// `agent_query` calls left. + pub agent_calls: usize, +} + +/// The structured result of evaluating one RLM cell, rendered into the JSON +/// content of the tool result. +#[derive(Debug, Clone, PartialEq, Serialize)] +pub struct RlmEvalResponse { + /// The session the cell ran in (echoed so a fresh session's generated id is + /// discoverable, and a later cell can continue it). + pub session_id: String, + /// Captured `print`/`debug` output. + pub stdout: String, + /// The cell's final expression value, if any. + #[serde(skip_serializing_if = "Option::is_none")] + pub value: Option, + /// Names of persistent variables the cell created or changed. + pub variables_changed: Vec, + /// Summarized capability calls the cell performed. + pub calls: Vec, + /// The final answer, if the cell called `answer(...)`. + #[serde(skip_serializing_if = "Option::is_none")] + pub final_answer: Option, + /// Wall-clock time the cell took, in milliseconds. + pub elapsed_ms: u64, + /// Number of cells this session has evaluated so far. + pub cells_used: usize, + /// Remaining per-session budget. + pub limits_remaining: RlmLimitsRemaining, + /// Whether the session was closed after this cell. + #[serde(default)] + pub closed: bool, +} diff --git a/src/openhuman/tinyagents/mod.rs b/src/openhuman/tinyagents/mod.rs index 38d6a172eb..785201571c 100644 --- a/src/openhuman/tinyagents/mod.rs +++ b/src/openhuman/tinyagents/mod.rs @@ -25,7 +25,7 @@ pub(crate) mod delegation; mod embeddings; pub(crate) mod journal; pub(crate) mod middleware; -mod model; +pub(crate) mod model; pub(crate) mod observability; pub(crate) mod orchestration; pub(crate) mod payload_summarizer; @@ -33,12 +33,12 @@ mod policy_denial; pub(crate) mod replay; pub(crate) mod retriever; mod routes; -mod run_cancellation_context; +pub(crate) mod run_cancellation_context; mod steering_forwarder; pub(crate) mod stop_hooks; pub(crate) mod subagent_graph; mod summarize; -mod tools; +pub(crate) mod tools; mod topology; use std::sync::Arc; diff --git a/src/openhuman/tinyagents/model.rs b/src/openhuman/tinyagents/model.rs index 31341fb78b..f7644809c8 100644 --- a/src/openhuman/tinyagents/model.rs +++ b/src/openhuman/tinyagents/model.rs @@ -364,6 +364,19 @@ pub(super) struct ProviderModel { profile: ModelProfile, } +/// Builds a `ChatModel<()>` capability over an openhuman [`Provider`], pinned to +/// `model`/`temperature`, ready to register into a `tinyagents` +/// [`CapabilityRegistry`](tinyagents::registry::CapabilityRegistry) — e.g. the +/// `.ragsh` REPL bridge (`crate::openhuman::rlm`). [`ProviderModel::new`] is +/// `pub(super)`; this is the crate-visible factory for that one bridge use. +pub(crate) fn provider_chat_model( + provider: Arc, + model: impl Into, + temperature: f64, +) -> Arc> { + Arc::new(ProviderModel::new(provider, model, temperature)) +} + impl ProviderModel { /// Build a model adapter for `provider`, pinned to `model`/`temperature`. /// diff --git a/src/openhuman/tinyagents/tools.rs b/src/openhuman/tinyagents/tools.rs index 981716478b..dcddc1cb17 100644 --- a/src/openhuman/tinyagents/tools.rs +++ b/src/openhuman/tinyagents/tools.rs @@ -120,7 +120,9 @@ impl Tool<()> for ToolAdapter { } } -fn tool_policy_from_openhuman_tool(tool: &dyn crate::openhuman::tools::Tool) -> ToolPolicy { +pub(crate) fn tool_policy_from_openhuman_tool( + tool: &dyn crate::openhuman::tools::Tool, +) -> ToolPolicy { use crate::openhuman::tools::traits::ToolTimeout; use crate::openhuman::tools::PermissionLevel; @@ -181,7 +183,7 @@ const TINYAGENTS_TOOL_SESSION: &str = "tinyagents"; /// Execute an openhuman [`Tool`](crate::openhuman::tools::Tool) for a harness /// [`TaToolCall`] and render the [`TaToolResult`] the way the LLM should see it /// (mirrors the live-path `HarnessToolExecutor`). -async fn execute_openhuman_tool( +pub(crate) async fn execute_openhuman_tool( tool: &dyn crate::openhuman::tools::Tool, call: TaToolCall, context: Option<&ToolExecutionContext>, diff --git a/src/openhuman/tools/mod.rs b/src/openhuman/tools/mod.rs index 8d86253889..6c2bbbd997 100644 --- a/src/openhuman/tools/mod.rs +++ b/src/openhuman/tools/mod.rs @@ -38,6 +38,7 @@ pub use crate::openhuman::monitor::tools::*; pub use crate::openhuman::orchestration::tools::*; pub use crate::openhuman::people::tools::*; pub use crate::openhuman::referral::tools::*; +pub use crate::openhuman::rlm::tools::*; pub use crate::openhuman::screen_intelligence::tools::*; pub use crate::openhuman::search::tools::*; pub use crate::openhuman::security::tools::*; diff --git a/src/openhuman/tools/ops.rs b/src/openhuman/tools/ops.rs index 12c855ea33..b716c7bb20 100644 --- a/src/openhuman/tools/ops.rs +++ b/src/openhuman/tools/ops.rs @@ -966,6 +966,29 @@ pub fn all_tools_with_runtime( tracing::debug!("[lsp] capability gate off (set OPENHUMAN_LSP_ENABLED=1 to register)"); } + // Language-workflow `rlm` tool (RLM / `.ragsh` REPL, `openhuman::rlm`): lets + // the orchestrator author and run its own Rhai workflow cells (fan-out, + // loops, dedup/verify pipelines). Registered on the `supervised`/`full` + // tiers only — dark on `readonly` (it can drive effectful tools/sub-agents) + // and behind the `OPENHUMAN_RLM=0` kill switch. Every effectful inner call + // still re-gates itself in the RLM bridge, so this surface adds no new + // ungated capability. + let rlm_enabled = std::env::var("OPENHUMAN_RLM") + .map(|v| v != "0") + .unwrap_or(true); + if rlm_enabled + && security.autonomy != crate::openhuman::security::policy::AutonomyLevel::ReadOnly + { + tools.push(Box::new(crate::openhuman::rlm::RlmTool::new())); + tracing::debug!("[rlm] registered rlm language-workflow tool"); + } else { + tracing::debug!( + rlm_enabled, + tier = ?security.autonomy, + "[rlm] rlm tool not registered (readonly tier or OPENHUMAN_RLM=0)" + ); + } + tools } diff --git a/vendor/tinyagents b/vendor/tinyagents index 357bcc8587..df391c4689 160000 --- a/vendor/tinyagents +++ b/vendor/tinyagents @@ -1 +1 @@ -Subproject commit 357bcc85871b40c03a7dd81b61d1b3253bfb33e6 +Subproject commit df391c46891cb69ffd39f69023f497a5d95bec0e