mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
feat(rlm): language-based workflows — Rhai .ragsh REPL as a first-class rlm tool (#4528)
This commit is contained in:
Steven Enamakel
Generated
+99
@@ -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",
|
||||
|
||||
+3
-1
@@ -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
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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",
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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`:
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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 `<thread>:<session_id>`, 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.
|
||||
@@ -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: "<name>"})` hits the real backend with usage intact.
|
||||
//! - **Agents** — a [`SubagentCapability`] per entry in the parent's
|
||||
//! `allowed_subagent_ids`, so `agent_query("<id>", ...)` 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<Vec<Box<dyn OhTool>>>,
|
||||
name: String,
|
||||
description: String,
|
||||
schema: TaToolSchema,
|
||||
policy: TaToolPolicy,
|
||||
}
|
||||
|
||||
impl RlmToolAdapter {
|
||||
fn new(tools: Arc<Vec<Box<dyn OhTool>>>, 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<TaToolResult> {
|
||||
Ok(self.dispatch(call, None).await)
|
||||
}
|
||||
|
||||
async fn call_with_context(
|
||||
&self,
|
||||
_state: &(),
|
||||
call: TaToolCall,
|
||||
context: ToolExecutionContext,
|
||||
) -> tinyagents::Result<TaToolResult> {
|
||||
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("<id>", ...)` 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<SubAgentOutput> {
|
||||
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()
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
@@ -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<String>,
|
||||
agents: Vec<String>,
|
||||
models: Vec<String>,
|
||||
}
|
||||
|
||||
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<ReplResult, TinyAgentsError>),
|
||||
/// 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<RlmEvalResponse, RlmError> {
|
||||
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<RlmEvalResponse, RlmError> {
|
||||
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<RlmCallSummary> {
|
||||
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<TaToolResult> {
|
||||
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:?}");
|
||||
}
|
||||
}
|
||||
@@ -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<u64>,
|
||||
overrides: Option<&RlmLimitsOverride>,
|
||||
) -> Result<ReplPolicy, String> {
|
||||
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<usize>, 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<usize>, 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);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,317 @@
|
||||
//! Session manager: a process-global, bounded map of persistent `.ragsh`
|
||||
//! sessions.
|
||||
//!
|
||||
//! Sessions are keyed `<thread_id>:<session_id>` 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<Mutex<ReplSession>>,
|
||||
/// 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<Mutex<ReplSession>>,
|
||||
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<HashMap<String, SessionSlot>>,
|
||||
}
|
||||
|
||||
static MANAGER: OnceLock<RlmSessionManager> = 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<String, SessionSlot>> {
|
||||
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<CellStats> {
|
||||
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<String, SessionSlot>, 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);
|
||||
}
|
||||
}
|
||||
@@ -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<String> {
|
||||
Some("running RLM workflow".to_string())
|
||||
}
|
||||
|
||||
fn display_detail(&self, args: &Value) -> Option<String> {
|
||||
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::<String>()))
|
||||
} else {
|
||||
Some(first.to_string())
|
||||
}
|
||||
}
|
||||
|
||||
async fn execute(&self, args: Value) -> anyhow::Result<ToolResult> {
|
||||
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"]));
|
||||
}
|
||||
}
|
||||
@@ -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<usize>,
|
||||
/// Requested `max_agent_calls` for the session.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub max_agent_calls: Option<usize>,
|
||||
/// Requested `max_model_calls` for the session.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub max_model_calls: Option<usize>,
|
||||
/// Requested `max_concurrency` for batched calls (hard-capped at 8).
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
pub max_concurrency: Option<usize>,
|
||||
}
|
||||
|
||||
/// 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<String>,
|
||||
/// Per-cell wall-clock timeout in seconds (clamped 1–3600).
|
||||
#[serde(default)]
|
||||
pub timeout_secs: Option<u64>,
|
||||
/// Optional per-session limit overrides.
|
||||
#[serde(default)]
|
||||
pub limits: Option<RlmLimitsOverride>,
|
||||
/// 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<serde_json::Value>,
|
||||
/// Names of persistent variables the cell created or changed.
|
||||
pub variables_changed: Vec<String>,
|
||||
/// Summarized capability calls the cell performed.
|
||||
pub calls: Vec<RlmCallSummary>,
|
||||
/// The final answer, if the cell called `answer(...)`.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub final_answer: Option<String>,
|
||||
/// 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,
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
@@ -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<dyn Provider>,
|
||||
model: impl Into<String>,
|
||||
temperature: f64,
|
||||
) -> Arc<dyn ChatModel<()>> {
|
||||
Arc::new(ProviderModel::new(provider, model, temperature))
|
||||
}
|
||||
|
||||
impl ProviderModel {
|
||||
/// Build a model adapter for `provider`, pinned to `model`/`temperature`.
|
||||
///
|
||||
|
||||
@@ -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>,
|
||||
|
||||
@@ -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::*;
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
Vendored
+1
-1
Submodule vendor/tinyagents updated: 357bcc8587...df391c4689
Reference in New Issue
Block a user