feat(runtime): pool node/python runtime processes across agents and skill runs (#5106) (#5129)

Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>
This commit is contained in:
YellowSnnowmann
2026-07-23 09:55:10 +03:00
committed by GitHub
co-authored by Claude Opus 4.8 Steven Enamakel
parent 08f28de147
commit 87fe420fdd
33 changed files with 3802 additions and 44 deletions
+19
View File
@@ -336,6 +336,25 @@ OPENHUMAN_SKILLS_WORKING_MEMORY_ENABLED=true
# [optional] Reserved cache directory for future managed CPython installs.
# OPENHUMAN_RUNTIME_PYTHON_CACHE_DIR=
# ---------------------------------------------------------------------------
# Shared language-runtime pool (#5106)
# ---------------------------------------------------------------------------
# Reuse a small, bounded set of warm node/python worker processes across skill
# runs and node_exec instead of forking one interpreter child per run. Full
# tuning lives in config.toml `[runtime_pool]`; these are the operator knobs.
# Node pooling is ON by default (each job runs in an isolated worker_thread).
# Python pooling is OFF by default (jobs share one interpreter, so reuse can
# leak globals across runs); opt in with `[runtime_pool.python] enabled = true`.
# [optional] Master kill switch. Default: true. `false` reverts every caller to
# the legacy per-call spawn (no behavioural change, just no reuse).
# OPENHUMAN_RUNTIME_POOL_ENABLED=true
# [optional] Max concurrently-resident node workers (default 2). Work beyond
# this queues rather than forking a new interpreter.
# OPENHUMAN_RUNTIME_POOL_NODE_MAX_WORKERS=2
# [optional] Max concurrently-resident python workers (default 2). Only takes
# effect when python pooling is explicitly enabled (see above).
# OPENHUMAN_RUNTIME_POOL_PYTHON_MAX_WORKERS=2
# ---------------------------------------------------------------------------
# TokenJuice — content-aware tool-output compaction (the content router)
# ---------------------------------------------------------------------------
+34 -1
View File
@@ -38,7 +38,7 @@ stderr). Each models a distinct embedding use case:
## How to run
Five scripts under `scripts/profile/` (each has `-h`/`--help`):
Six scripts under `scripts/profile/` (each has `-h`/`--help`):
- **`library-bench.sh`** — the primary RSS/duration benchmark. Builds the
binaries, runs each scenario N fresh-process repeats (default 5), and
@@ -88,6 +88,14 @@ Five scripts under `scripts/profile/` (each has `-h`/`--help`):
./scripts/profile/library-instances.sh --instances "10,25,50" --hold-secs 30
```
- **`library-pool-gate.sh`** — the runtime-pool regression gate (#5106). Runs
`skill-run` with K parallel skill runs and asserts the process tree grows by
~one pooled worker, not K interpreters; reports pooled vs unpooled.
```bash
./scripts/profile/library-pool-gate.sh --concurrency 8 --workers 1
```
### Default vs slim builds
Default-feature builds link every compile-time domain gate (`voice`, `web3`,
@@ -115,6 +123,9 @@ behavior, not linked code size.
| `OPENHUMAN_PROFILE_FORCE_UTC=1` | Skip `iana_time_zone`/CoreFoundation timezone resolution. |
| `OPENHUMAN_PROFILE_HOLD_SECS` / `HOLD_BEFORE_SECS` | Pause the process at settled/baseline state for external inspection (`vmmap`, `heap`, `malloc_history`, Instruments). |
| `OPENHUMAN_PROFILE_DHAT_OUT` | Output path for dhat JSON (set by `library-heap.sh`). |
| `OPENHUMAN_PROFILE_SKILL_RUN_CONCURRENCY` | `skill-run`: number of parallel `code_executor` turns (K), each spawning a `node_exec` job (default 1). |
| `OPENHUMAN_PROFILE_SKILL_RUN_POOL` | `skill-run`: `off` disables the shared runtime pool (legacy per-call spawn; tree then shows ~K resident `node` children). Default on. |
| `OPENHUMAN_PROFILE_SKILL_RUN_POOL_WORKERS` | `skill-run`: pool size W when pooling is on (default 1). The scenario asserts `child_count <= W` for K > 1 — the #5106 regression gate. |
## Metrics and interpretation
@@ -334,6 +345,28 @@ Runtime/subagent scenarios: `skill-run` measures a real `node` child at
runtime-pooling issue (tinyhumansai/openhuman#5106); `subagent-storm` shows
~0.78 MiB marginal per additional parallel subagent (K=8→32 cross-width).
Runtime pool (#5106): with the shared pool on (the default), a batch of K
concurrent `skill-run` turns shares a bounded set of warm `node` workers, so
the process tree grows by ~one pooled worker instead of K interpreters.
Measured at **K=8** (`max_workers=1`, system node v24): pooled tree
`child_count=1` at **~202 MiB**, vs unpooled `child_count=8` at **~690 MiB**
(eight `node` children of ~7375 MB each) — a ~490 MiB / 3.4× reduction that
grows with K. Compare the two regimes directly:
```bash
# Legacy: K interpreters resident at peak.
OPENHUMAN_PROFILE_SKILL_RUN_CONCURRENCY=8 OPENHUMAN_PROFILE_SKILL_RUN_POOL=off \
target/release/library-profile skill-run
# Pooled: child_count stays at the pool size (asserted), not K.
OPENHUMAN_PROFILE_SKILL_RUN_CONCURRENCY=8 OPENHUMAN_PROFILE_SKILL_RUN_POOL_WORKERS=1 \
target/release/library-profile skill-run
```
The pool is configured in `[runtime_pool]` (master switch + per-language
`node`/`python` `max_workers`, `idle_ttl_secs`, `recycle_after_jobs`,
`max_queue_depth`); `enabled = false` reverts every caller to the legacy
per-call spawn.
Watch-items from the sweep: thread count grows ~0.35/agent (needs
attribution + cap before real 1000-agent runs), and p95 latency at N=500 on
2 workers shows CPU saturation is the load constraint, not memory.
+1 -1
View File
@@ -39,5 +39,5 @@ Filesystem tools respect a workspace boundary - the agent can't read or write ou
## See also
* [System & Utilities](system-and-utilities.md) - `shell`, `node_exec`, `npm_exec` for the rest of the dev loop.
* [System & Utilities](system-and-utilities.md) - `shell`, `node_exec`, `npm_exec`, `python_exec` for the rest of the dev loop.
* [Agent Coordination](agent-coordination.md) - `todo_write`, `spawn_subagent` for larger refactors.
@@ -14,6 +14,7 @@ The catch-all family. Small, sharp tools the agent reaches for to round out a ta
| `shell` | Run a shell command. Bounded output, captured exit code. |
| `node_exec` | Run a Node.js snippet - useful for one-off scripting. |
| `npm_exec` | Run an `npm`/`pnpm`/`yarn` script. |
| `python_exec` | Run a Python 3 snippet or `.py` script - one-off scripting in Python. |
| `current_time` | Get the current time in any timezone, with formatting options. |
| `schedule` | One-shot "do this once at time T" - for recurring jobs see [Cron](cron.md). |
| `pushover` | Send a push notification to your devices. |
+17
View File
@@ -94,6 +94,23 @@ exit or missing/invalid JSON); default is report-only. See
[`docs/library-benchmarking.md`](../../docs/library-benchmarking.md#fleet-one-process-vs-instances-many-processes)
for the fleet-vs-instances framing.
### `library-pool-gate.sh` — runtime pool regression gate (#5106)
Drives the `skill-run` scenario with K parallel skill runs and asserts the DoD:
with the shared runtime pool ON, the process tree grows by ~one pooled worker,
**not** K interpreters. The scenario hard-asserts `tree.child_count <= max_workers`
for K > 1 (nonzero exit); this script runs it and reports a pooled-vs-unpooled
comparison. A regression that reintroduces per-run forking fails the gate.
```bash
./scripts/profile/library-pool-gate.sh # K=8, max_workers=1
./scripts/profile/library-pool-gate.sh --concurrency 16 --skip-build
```
Exits 0 = pass, 1 = regression. SKIPs (exit 0) when no system `node` is on
`PATH` (the scenario must never download an interpreter), so node-less CI
runners don't false-fail.
## Quick start
```bash
+133
View File
@@ -0,0 +1,133 @@
#!/usr/bin/env bash
# library-pool-gate.sh — regression gate for the shared runtime pool (#5106).
#
# Runs the `skill-run` scenario with K parallel skill runs and asserts the DoD:
# with the pool ON, the process tree grows by ~one pooled worker, NOT K
# interpreters. The scenario itself hard-asserts `tree.child_count <= max_workers`
# for K > 1 (nonzero exit on failure); this script drives it in the profiling
# suite and adds a pooled-vs-unpooled comparison for the report.
#
# A regression that reintroduces per-run interpreter forking makes the pooled
# run's child_count scale with K → the scenario exits nonzero → this gate fails.
#
# Usage:
# ./scripts/profile/library-pool-gate.sh [--concurrency N] [--workers W] [--skip-build] [--out DIR]
#
# Exits 0 = pass, 1 = regression/failure, 0 (with SKIP notice) = no system node.
set -euo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
CONCURRENCY=8
POOL_WORKERS=1
SKIP_BUILD=0
OUT_DIR=""
# Reject non-positive-integer values up front so a bad --concurrency/--workers
# fails loudly instead of silently changing the workload.
require_pos_int() {
case "$2" in
''|*[!0-9]*) echo "ERROR: $1 must be a positive integer, got: $2" >&2; exit 1 ;;
esac
[ "$2" -ge 1 ] || { echo "ERROR: $1 must be >= 1, got: $2" >&2; exit 1; }
}
while [[ $# -gt 0 ]]; do
case "$1" in
--concurrency) CONCURRENCY="${2:?--concurrency requires a value}"; require_pos_int --concurrency "$CONCURRENCY"; shift 2 ;;
--workers) POOL_WORKERS="${2:?--workers requires a value}"; require_pos_int --workers "$POOL_WORKERS"; shift 2 ;;
--skip-build) SKIP_BUILD=1; shift ;;
--out) OUT_DIR="${2:?--out requires a value}"; shift 2 ;;
-h|--help) sed -n '2,18p' "${BASH_SOURCE[0]}" | sed 's/^# \{0,1\}//'; exit 0 ;;
*) echo "ERROR: unknown argument: $1" >&2; exit 1 ;;
esac
done
require_pos_int --concurrency "$CONCURRENCY"
require_pos_int --workers "$POOL_WORKERS"
log() { echo "[pool-gate] $*" >&2; }
if ! command -v jq >/dev/null 2>&1; then
echo "ERROR: jq is required. Install it (e.g. 'brew install jq')." >&2
exit 1
fi
# The scenario requires a real system node (it must never download one). No node
# ⇒ SKIP (exit 0) so node-less CI runners don't false-fail; a node-equipped
# runner is where this gate is meaningful.
if ! command -v node >/dev/null 2>&1; then
log "SKIP: no system 'node' on PATH — pool gate needs a real interpreter."
exit 0
fi
[[ -z "$OUT_DIR" ]] && OUT_DIR="$REPO_ROOT/target/profile/rust-library/pool-gate-$(date +%Y%m%d-%H%M%S)"
mkdir -p "$OUT_DIR"
BIN="$REPO_ROOT/target/release/library-profile"
if [[ "$SKIP_BUILD" -eq 0 ]]; then
log "building library-profile (rss-bench, GGML_NATIVE=OFF)"
( cd "$REPO_ROOT" && GGML_NATIVE=OFF cargo build --release --features rss-bench --bin library-profile )
fi
[[ -x "$BIN" ]] || { echo "ERROR: $BIN not found (drop --skip-build to build it)." >&2; exit 1; }
pooled_json="$OUT_DIR/pooled-k$CONCURRENCY.json"
unpooled_json="$OUT_DIR/unpooled-k$CONCURRENCY.json"
# --- Pooled run (the gate) ------------------------------------------------
# The scenario hard-asserts child_count <= POOL_WORKERS; a nonzero exit here
# (via set -e) fails the gate.
log "pooled run: K=$CONCURRENCY, max_workers=$POOL_WORKERS (asserts child_count <= $POOL_WORKERS)"
# Force the pool ON explicitly: an inherited OPENHUMAN_PROFILE_SKILL_RUN_POOL=off
# would run the legacy path and make the scenario skip its pool assertion.
OPENHUMAN_PROFILE_SKILL_RUN_POOL=on \
OPENHUMAN_PROFILE_SKILL_RUN_CONCURRENCY="$CONCURRENCY" \
OPENHUMAN_PROFILE_SKILL_RUN_POOL_WORKERS="$POOL_WORKERS" \
"$BIN" skill-run >"$pooled_json"
# --- Unpooled baseline (report only) --------------------------------------
log "unpooled baseline: K=$CONCURRENCY, pool OFF (expect ~$CONCURRENCY interpreters)"
OPENHUMAN_PROFILE_SKILL_RUN_CONCURRENCY="$CONCURRENCY" \
OPENHUMAN_PROFILE_SKILL_RUN_POOL=off \
"$BIN" skill-run >"$unpooled_json"
# Pooled run: a missing/null tree means nothing was measured — that must FAIL,
# not silently coerce to 0 (which would let the gate pass without observing an
# interpreter). Unpooled is report-only, so its tree may default to 0.
pooled_cc="$(jq -r '.tree.child_count // "null"' "$pooled_json")"
unpooled_cc="$(jq -r '.tree.child_count // 0' "$unpooled_json")"
pooled_rss="$(jq -r '.tree.tree_rss_kib // "null"' "$pooled_json")"
unpooled_rss="$(jq -r '.tree.tree_rss_kib // 0' "$unpooled_json")"
echo
echo "=== runtime pool gate (#5106) — K=$CONCURRENCY ==="
echo " pooled (max_workers=$POOL_WORKERS): child_count=$pooled_cc tree_rss_kib=$pooled_rss"
echo " unpooled (legacy spawn) : child_count=$unpooled_cc tree_rss_kib=$unpooled_rss"
if [[ "$pooled_cc" == "null" ]]; then
echo "FAIL: pooled run captured no process-tree sample — cannot verify the pooled worker." >&2
exit 1
fi
# The pooled worker must actually be observed (>= 1) and must not exceed the
# pool size.
if [[ "$pooled_cc" -lt 1 ]]; then
echo "FAIL: pooled run observed zero interpreter children — the pooled worker was not sampled." >&2
exit 1
fi
if [[ "$pooled_cc" -gt "$POOL_WORKERS" ]]; then
echo "FAIL: pooled child_count=$pooled_cc exceeds max_workers=$POOL_WORKERS — pool is forking per run." >&2
exit 1
fi
# Sanity: the unpooled baseline should fork more than the pool. If it didn't,
# the comparison is inconclusive (likely a sampling miss on a fast machine) —
# warn rather than hard-fail, since the pooled assertion above is the real gate.
if [[ "$unpooled_cc" -le "$pooled_cc" ]]; then
echo "WARN: unpooled baseline child_count=$unpooled_cc did not exceed pooled=$pooled_cc" >&2
echo " (baseline sampling inconclusive; the pooled assertion still passed)." >&2
else
echo "PASS: pool bounded interpreters to $pooled_cc for K=$CONCURRENCY concurrent skill runs (unpooled forked $unpooled_cc)."
fi
echo " artifacts: $OUT_DIR"
exit 0
+126 -29
View File
@@ -1,5 +1,6 @@
//! `skill-run`: the true, process-*tree* cost of a skill step that executes on
//! a real language runtime — the interpreter child process included.
//! a real language runtime — the interpreter child process included — and the
//! regression instrument for the shared runtime pool (issue #5106).
//!
//! ## What it actually runs
//!
@@ -10,16 +11,24 @@
//! carries `node_exec` / `npm_exec`). So the agent that genuinely spawns the
//! language runtime *is* `code_executor`. This scenario drives that specialist
//! directly — one turn, one scripted `node_exec` call — which is the real
//! node-executing path, not a bare `std::process` spawn. Measuring the
//! orchestrator→specialist delegation on top would add in-process agent cost
//! without changing the runtime-child cost this scenario exists to capture;
//! the compromise is documented here on purpose.
//! node-executing path, not a bare `std::process` spawn.
//!
//! The mock ([`SkillRunMock`]) emits a `node_exec` call whose inline JavaScript
//! does real work, allocates, and busy-waits ~1.2 s so the child stays resident
//! long enough for the harness tree sampler (15 ms poll) to attribute it, then
//! prints JSON carrying [`NODE_MARKER`]. When that output rides back into the
//! turn the mock returns a plain final answer and the turn completes.
//! ## Concurrency knob (K parallel skill runs)
//!
//! `OPENHUMAN_PROFILE_SKILL_RUN_CONCURRENCY=K` (default 1) drives **K**
//! `code_executor` turns in parallel, each emitting its own `node_exec` call.
//! The point of #5106: with the runtime pool **on**, K concurrent skill runs
//! share a small bounded set of warm `node` workers, so the process tree grows
//! by ~one pooled worker — **not** K interpreters. This scenario asserts that
//! (`tree.child_count <= max_workers`) whenever pooling is on and K > 1, so a
//! regression that reintroduces per-run forking fails the profiling suite.
//!
//! Toggle for an A/B baseline:
//!
//! * `OPENHUMAN_PROFILE_SKILL_RUN_POOL=off` — disable the pool (legacy per-call
//! spawn); the tree then shows ~K resident `node` children at peak.
//! * `OPENHUMAN_PROFILE_SKILL_RUN_POOL_WORKERS=W` — pool size (default 1, for a
//! tight, deterministic bound).
//!
//! ## No interpreter download
//!
@@ -29,15 +38,15 @@
//! nonzero exit if none is on `PATH` — it must never pull a runtime.
//!
//! The measured cost lands in `result.tree` (`tree_rss_kib`, `child_count`,
//! per-child RSS) captured at the workload peak, since the `node` child has
//! already exited by settle time.
//! per-child RSS), captured at the workload peak.
use std::sync::Arc;
use anyhow::Result;
use anyhow::{Context, Result};
use openhuman_core::core::event_bus::init_global;
use openhuman_core::openhuman::agent::harness::AgentDefinitionRegistry;
use openhuman_core::openhuman::agent::Agent;
use openhuman_core::openhuman::config::Config;
use openhuman_core::openhuman::inference::provider::factory::test_provider_override;
use openhuman_core::openhuman::inference::provider::Provider;
use openhuman_core::openhuman::security::AutonomyLevel;
@@ -76,14 +85,49 @@ fn require_system_node() -> Result<String> {
}
}
/// Read a `>= 1` usize from the environment. Absent ⇒ `default`; present but not
/// a positive integer ⇒ a hard error (silently coercing a bad `0`/garbage value
/// would change the workload and bypass the `K > 1` pool gate).
fn env_usize(key: &str, default: usize) -> Result<usize> {
match std::env::var(key) {
Err(_) => Ok(default),
Ok(raw) => {
let n: usize = raw
.trim()
.parse()
.with_context(|| format!("{key}={raw:?} is not a valid integer"))?;
anyhow::ensure!(n >= 1, "{key}={raw:?} must be >= 1");
Ok(n)
}
}
}
/// Apply the `[runtime_pool]` settings to the in-memory fixture config. The exec
/// tools snapshot `config.runtime_pool` at construction (the pool config is
/// injected, never re-read from disk on the hot path), so mutating the config
/// the agent is built from is what toggles pooling for the scenario.
fn apply_pool_config(config: &mut Config, pool_enabled: bool, workers: usize) {
config.runtime_pool.enabled = pool_enabled;
config.runtime_pool.node.max_workers = workers;
config.runtime_pool.node.idle_ttl_secs = 300;
config.runtime_pool.node.recycle_after_jobs = 0;
}
pub async fn run() -> Result<ProfileResult> {
// Hard requirement: a system node must be present (no download).
require_system_node()?;
let mut fixture = fixture()?;
let concurrency = env_usize("OPENHUMAN_PROFILE_SKILL_RUN_CONCURRENCY", 1)?;
let pool_enabled = std::env::var("OPENHUMAN_PROFILE_SKILL_RUN_POOL")
.map(|v| !v.trim().eq_ignore_ascii_case("off"))
.unwrap_or(true);
let pool_workers = env_usize("OPENHUMAN_PROFILE_SKILL_RUN_POOL_WORKERS", 1)?;
// `node_exec` is a Write-class acting tool. Full autonomy keeps the gate
// from parking the turn on approval; the gate is also opted out explicitly.
let mut fixture = fixture()?;
fixture.config.autonomy.level = AutonomyLevel::Full;
apply_pool_config(&mut fixture.config, pool_enabled, pool_workers);
let _approval_env = EnvGuard::set("OPENHUMAN_APPROVAL_GATE", "0");
let _ = init_global(256);
@@ -95,21 +139,37 @@ pub async fn run() -> Result<ProfileResult> {
let _provider = test_provider_override::install(provider);
eprintln!(
"[library-profile] skill-run: registries ready, node_exec mock installed \
(agent={CODE_AGENT})"
(agent={CODE_AGENT}, concurrency={concurrency}, pool={}, pool_workers={pool_workers})",
if pool_enabled { "on" } else { "off" }
);
let mock_for_workload = mock.clone();
let config = fixture.config.clone();
let mut result = measure_with_tree("skill-run", 1, None, move |rec| async move {
let mut result = measure_with_tree("skill-run", concurrency, None, move |rec| async move {
rec.checkpoint("turn-start")?;
let mut agent = Agent::from_config_for_agent(&config, CODE_AGENT)?;
let reply = agent
.run_single(
"Run a short JavaScript computation with node_exec and report the JSON it prints.",
)
.await?;
// Drive K code_executor turns concurrently. `join_all` gives real
// process-level parallelism (each node_exec awaits its own child /
// pooled job) without requiring the agent future to be `Send`.
let futures = (0..concurrency).map(|idx| {
let config = config.clone();
async move {
let mut agent = Agent::from_config_for_agent(&config, CODE_AGENT)
.with_context(|| format!("building code_executor agent #{idx}"))?;
let reply = agent
.run_single(
"Run a short JavaScript computation with node_exec and report the JSON it prints.",
)
.await
.with_context(|| format!("code_executor turn #{idx}"))?;
anyhow::ensure!(!reply.trim().is_empty(), "empty code_executor reply #{idx}");
Ok::<(), anyhow::Error>(())
}
});
let outcomes = futures::future::join_all(futures).await;
for outcome in outcomes {
outcome?;
}
rec.checkpoint("turn-done")?;
anyhow::ensure!(!reply.trim().is_empty(), "empty code_executor reply");
anyhow::ensure!(
mock_for_workload.node_call_emitted(),
"the node_exec tool call was never emitted"
@@ -126,15 +186,18 @@ pub async fn run() -> Result<ProfileResult> {
Some(tree) if tree.child_count >= 1 => {
eprintln!(
"[library-profile] skill-run: captured tree_rss_kib={} child_count={} \
children={:?}",
tree.tree_rss_kib, tree.child_count, tree.children
concurrency={concurrency} pool={} children={:?}",
tree.tree_rss_kib,
tree.child_count,
if pool_enabled { "on" } else { "off" },
tree.children
);
}
Some(tree) => {
eprintln!(
"[library-profile] skill-run: WARNING tree captured but no child was resident at \
peak (tree_rss_kib={}). The node child may have been too short-lived; \
the busy-wait should have kept it alive.",
"[library-profile] skill-run: tree captured but no child resident at peak \
(tree_rss_kib={}). With the pool on this is expected between jobs; with the \
pool off the node child may have been too short-lived.",
tree.tree_rss_kib
);
}
@@ -143,7 +206,41 @@ pub async fn run() -> Result<ProfileResult> {
}
}
// DoD gate (#5106): with pooling on, K concurrent skill runs must not fork K
// interpreters — the tree grows by ~one pooled worker. A regression that
// reintroduces per-run forking makes child_count scale with K and fails here.
if pool_enabled && concurrency > 1 {
// Require a real measurement: a missing tree or zero children would let
// the gate "pass" without ever observing an interpreter. With the
// scenario's 300 s idle TTL the pooled worker is resident at sample time,
// so this must hold.
let tree = result.tree.as_ref().ok_or_else(|| {
anyhow::anyhow!(
"pool gate: no process-tree sample captured for K={concurrency}; cannot verify \
the pooled worker (tree sampling is required for this gate)"
)
})?;
anyhow::ensure!(
tree.child_count >= 1,
"pool gate: process tree captured zero interpreter children at K={concurrency}; \
the pooled worker was not observed"
);
anyhow::ensure!(
tree.child_count <= pool_workers,
"runtime pool failed to bound interpreters: child_count={} exceeds max_workers={} \
at concurrency K={} — expected warm-worker reuse, not K forked children",
tree.child_count,
pool_workers,
concurrency
);
eprintln!(
"[library-profile] skill-run: POOL OK — {} node worker(s) served {concurrency} \
concurrent skill runs (would be ~{concurrency} interpreters unpooled)",
tree.child_count
);
}
// Fold in scenario-visible fields (schema stays additive).
result.workload_units = 1;
result.workload_units = concurrency;
Ok(result)
}
@@ -37,6 +37,7 @@ named = [
"git_operations",
"node_exec",
"npm_exec",
"python_exec",
"curl",
"grep",
"glob",
@@ -23,6 +23,7 @@ named = [
"git_operations",
"node_exec",
"npm_exec",
"python_exec",
"grep",
"glob",
"list",
@@ -13,4 +13,4 @@ omit_skills_catalog = true
hint = "coding"
[tools]
named = ["file_write", "shell", "node_exec", "npm_exec"]
named = ["file_write", "shell", "node_exec", "npm_exec", "python_exec"]
+12 -11
View File
@@ -43,17 +43,18 @@ pub use schema::{
MeetConfig, MemoryConfig, MemoryTreeConfig, ModelRouteConfig, MultimodalConfig,
MultimodalFileConfig, ObservabilityConfig, OrchestratorModelConfig, PolymarketClobCredentials,
PolymarketConfig, PrivacyConfig, PrivacyMode, ProxyConfig, ProxyScope, ReflectionSource,
ReliabilityConfig, ResourceLimitsConfig, RuntimeConfig, SandboxBackend, SandboxConfig,
SchedulerConfig, SchedulerGateConfig, SchedulerGateMode, ScreenIntelligenceConfig,
SearchConfig, SearchEngine, SearchEngineCredentials, SearxngConfig, SecretsConfig,
SecurityConfig, ShellConfig, SlackConfig, StorageConfig, StorageProviderConfig,
StorageProviderSection, StreamMode, TeamModelConfig, TelegramConfig, TokenjuiceConfig,
UpdateConfig, UpdateRestartStrategy, VoiceActivationMode, VoiceServerConfig, WebSearchConfig,
WebhookConfig, YuanbaoConfig, DEFAULT_CLOUD_LLM_MODEL, DEFAULT_MEMORY_SYNC_INTERVAL_SECS,
DEFAULT_MODEL, MEMORY_SYNC_INTERVAL_PRESETS_SECS, MODEL_AGENTIC_V1, MODEL_BURST_V1,
MODEL_CHAT_V1, MODEL_CODING_V1, MODEL_REASONING_QUICK_V1, MODEL_REASONING_V1,
MODEL_SUMMARIZATION_V1, MODEL_VISION_V1, SEARCH_ENGINE_BRAVE, SEARCH_ENGINE_DISABLED,
SEARCH_ENGINE_MANAGED, SEARCH_ENGINE_PARALLEL, SEARCH_ENGINE_QUERIT,
ReliabilityConfig, ResourceLimitsConfig, RuntimeConfig, RuntimePoolConfig,
RuntimePoolLangConfig, SandboxBackend, SandboxConfig, SchedulerConfig, SchedulerGateConfig,
SchedulerGateMode, ScreenIntelligenceConfig, SearchConfig, SearchEngine,
SearchEngineCredentials, SearxngConfig, SecretsConfig, SecurityConfig, ShellConfig,
SlackConfig, StorageConfig, StorageProviderConfig, StorageProviderSection, StreamMode,
TeamModelConfig, TelegramConfig, TokenjuiceConfig, UpdateConfig, UpdateRestartStrategy,
VoiceActivationMode, VoiceServerConfig, WebSearchConfig, WebhookConfig, YuanbaoConfig,
DEFAULT_CLOUD_LLM_MODEL, DEFAULT_MEMORY_SYNC_INTERVAL_SECS, DEFAULT_MODEL,
MEMORY_SYNC_INTERVAL_PRESETS_SECS, MODEL_AGENTIC_V1, MODEL_BURST_V1, MODEL_CHAT_V1,
MODEL_CODING_V1, MODEL_REASONING_QUICK_V1, MODEL_REASONING_V1, MODEL_SUMMARIZATION_V1,
MODEL_VISION_V1, SEARCH_ENGINE_BRAVE, SEARCH_ENGINE_DISABLED, SEARCH_ENGINE_MANAGED,
SEARCH_ENGINE_PARALLEL, SEARCH_ENGINE_QUERIT,
};
// Kept as a separate re-export (issue #4117) so the large alphabetized group
// above stays byte-identical and rustfmt-stable.
@@ -449,6 +449,33 @@ impl Config {
self.runtime_python.preferred_command = command.trim().to_string();
}
// --- Shared language-runtime pool (#5106) --------------------------
if let Some(flag) = env.get("OPENHUMAN_RUNTIME_POOL_ENABLED") {
if let Some(enabled) = parse_env_bool("OPENHUMAN_RUNTIME_POOL_ENABLED", &flag) {
self.runtime_pool.enabled = enabled;
}
}
if let Some(raw) = env.get("OPENHUMAN_RUNTIME_POOL_NODE_MAX_WORKERS") {
match raw.trim().parse::<usize>() {
Ok(n) => self.runtime_pool.node.max_workers = n,
Err(e) => tracing::warn!(
value = %raw,
error = %e,
"[config] ignoring invalid OPENHUMAN_RUNTIME_POOL_NODE_MAX_WORKERS"
),
}
}
if let Some(raw) = env.get("OPENHUMAN_RUNTIME_POOL_PYTHON_MAX_WORKERS") {
match raw.trim().parse::<usize>() {
Ok(n) => self.runtime_pool.python.max_workers = n,
Err(e) => tracing::warn!(
value = %raw,
error = %e,
"[config] ignoring invalid OPENHUMAN_RUNTIME_POOL_PYTHON_MAX_WORKERS"
),
}
}
// --- TokenJuice content router -------------------------------------
if let Some(flag) = env.get("OPENHUMAN_TOKENJUICE_ENABLED") {
if let Some(v) = parse_env_bool("OPENHUMAN_TOKENJUICE_ENABLED", &flag) {
+36
View File
@@ -593,6 +593,42 @@ fn env_overlay_toggles_agent_tracing_capture_content() {
assert!(cfg.observability.agent_tracing.capture_content);
}
#[test]
fn env_overlay_runtime_pool_workers_and_enabled() {
// Baseline: master switch on, both pools at the default worker count.
let mut cfg = Config::default();
assert!(cfg.runtime_pool.enabled, "master switch defaults on");
assert_eq!(cfg.runtime_pool.node.max_workers, 2);
assert_eq!(cfg.runtime_pool.python.max_workers, 2);
// Valid overrides land; `enabled` parses via the shared bool parser.
cfg.apply_env_overlay_with(
&HashMapEnv::new()
.with("OPENHUMAN_RUNTIME_POOL_ENABLED", "off")
.with("OPENHUMAN_RUNTIME_POOL_NODE_MAX_WORKERS", "7")
.with("OPENHUMAN_RUNTIME_POOL_PYTHON_MAX_WORKERS", "3"),
);
assert!(!cfg.runtime_pool.enabled, "explicit off disables the pool");
assert_eq!(cfg.runtime_pool.node.max_workers, 7);
assert_eq!(cfg.runtime_pool.python.max_workers, 3);
// Unparseable worker counts are ignored (the warn arm) — the previously
// applied values survive rather than resetting to a default or zero.
cfg.apply_env_overlay_with(
&HashMapEnv::new()
.with("OPENHUMAN_RUNTIME_POOL_NODE_MAX_WORKERS", "not-a-number")
.with("OPENHUMAN_RUNTIME_POOL_PYTHON_MAX_WORKERS", ""),
);
assert_eq!(
cfg.runtime_pool.node.max_workers, 7,
"invalid node worker count keeps the prior value"
);
assert_eq!(
cfg.runtime_pool.python.max_workers, 3,
"empty python worker count keeps the prior value"
);
}
#[test]
fn env_overlay_model_only_honours_namespaced_var() {
// Both set → OPENHUMAN_MODEL wins; bare MODEL is ignored even when
+2
View File
@@ -41,6 +41,7 @@ mod privacy;
mod proxy;
mod routes;
mod runtime;
mod runtime_pool;
mod runtime_python;
mod scheduler_gate;
mod storage_memory;
@@ -87,6 +88,7 @@ pub use routes::{EmbeddingRouteConfig, ModelRouteConfig};
pub use runtime::{
DockerRuntimeConfig, ReliabilityConfig, RuntimeConfig, SchedulerConfig, ShellConfig,
};
pub use runtime_pool::{RuntimePoolConfig, RuntimePoolLangConfig};
pub use runtime_python::RuntimePythonConfig;
pub use scheduler_gate::{SchedulerGateConfig, SchedulerGateMode};
pub use storage_memory::{
+199
View File
@@ -0,0 +1,199 @@
//! Shared language-runtime pool configuration (`[runtime_pool]`).
//!
//! Controls the bounded pools of long-lived `node` / `python` worker processes
//! that execute inline code jobs for skill runs and the `node_exec` agent tool
//! instead of forking one interpreter child per execution (issue #5106).
//!
//! ## Why this exists
//!
//! At the opencompany deployment target (1001000 live agents in a
//! 2 GB / 2 vCPU box) a per-run interpreter child is the single biggest budget
//! breaker: a single JS skill step spawns a `node` child at ~7275 MB RSS.
//! Sharing a small, bounded pool of warm workers turns "K concurrent skill runs
//! → K interpreters" into "K concurrent skill runs → ~one pooled worker", at
//! the cost of serialising work beyond the pool size (surfaced as queue wait).
//!
//! ## Kill switch
//!
//! `enabled = false` (globally, or per language) reverts callers to the legacy
//! per-call spawn path with **no** behavioural change — the pool is purely an
//! optimisation seam and must always be safe to turn off.
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
/// `[runtime_pool]` — top-level switch plus per-language pool tuning.
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(default)]
pub struct RuntimePoolConfig {
/// Master switch. When `false`, no pool is started and every caller falls
/// back to spawning a fresh interpreter child per execution (legacy
/// behaviour). Per-language `enabled` flags gate each language on top of
/// this.
#[serde(default = "default_true")]
pub enabled: bool,
/// Node.js worker pool.
#[serde(default)]
pub node: RuntimePoolLangConfig,
/// Python worker pool.
#[serde(default)]
pub python: RuntimePoolLangConfig,
}
/// Per-language pool tuning. Applies identically to the node and python pools.
#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
#[serde(default)]
pub struct RuntimePoolLangConfig {
/// Whether this language routes through the pool. `None` (unset) means
/// "use the per-language default" — resolved via [`Self::is_enabled`]:
/// **node defaults on** (worker_thread isolation makes reuse safe),
/// **python defaults off** (in-process reuse can leak globals across jobs,
/// so it stays opt-in until stronger isolation lands). `false`/`true`
/// override explicitly.
#[serde(default)]
pub enabled: Option<bool>,
/// Maximum number of concurrently-resident worker processes. Concurrent
/// jobs beyond this bound **queue** rather than fork a new interpreter —
/// this is the whole point of the pool. Clamped to at least 1 at read time.
#[serde(default = "default_max_workers")]
pub max_workers: usize,
/// Idle time-to-live (seconds). A worker that has served no job for this
/// long is reaped so an idle fleet pays zero interpreter RSS. `0` disables
/// idle reaping (workers live until recycled or the process exits).
#[serde(default = "default_idle_ttl_secs")]
pub idle_ttl_secs: u64,
/// Recycle a worker after it has completed this many jobs, bounding
/// state/heap contamination across otherwise-isolated runs. `0` disables
/// job-count recycling.
#[serde(default = "default_recycle_after_jobs")]
pub recycle_after_jobs: u64,
/// Maximum number of jobs allowed to wait in the queue for a free worker
/// before new submissions are rejected with backpressure (rather than
/// growing memory unboundedly). Clamped to at least 1 at read time.
#[serde(default = "default_max_queue_depth")]
pub max_queue_depth: usize,
}
impl RuntimePoolLangConfig {
/// Whether this language routes through the pool, resolving an unset
/// `enabled` to the caller-supplied per-language default (node → `true`,
/// python → `false`). An explicit `enabled = true/false` always wins.
pub fn is_enabled(&self, default: bool) -> bool {
self.enabled.unwrap_or(default)
}
/// Effective worker count, never zero.
pub fn effective_max_workers(&self) -> usize {
self.max_workers.max(1)
}
/// Effective queue depth, never zero.
pub fn effective_max_queue_depth(&self) -> usize {
self.max_queue_depth.max(1)
}
}
fn default_true() -> bool {
true
}
fn default_max_workers() -> usize {
2
}
fn default_idle_ttl_secs() -> u64 {
60
}
fn default_recycle_after_jobs() -> u64 {
100
}
fn default_max_queue_depth() -> usize {
256
}
impl Default for RuntimePoolConfig {
fn default() -> Self {
Self {
enabled: default_true(),
node: RuntimePoolLangConfig::default(),
python: RuntimePoolLangConfig::default(),
}
}
}
impl Default for RuntimePoolLangConfig {
fn default() -> Self {
Self {
enabled: None,
max_workers: default_max_workers(),
idle_ttl_secs: default_idle_ttl_secs(),
recycle_after_jobs: default_recycle_after_jobs(),
max_queue_depth: default_max_queue_depth(),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn defaults_master_on_node_on_python_off() {
let cfg = RuntimePoolConfig::default();
assert!(cfg.enabled, "master switch defaults on");
// Per-language `enabled` is unset by default; node resolves on, python off.
assert_eq!(cfg.node.enabled, None);
assert_eq!(cfg.python.enabled, None);
assert!(cfg.node.is_enabled(true), "node default is on");
assert!(!cfg.python.is_enabled(false), "python default is off");
assert_eq!(cfg.node.max_workers, 2);
assert_eq!(cfg.node.idle_ttl_secs, 60);
assert_eq!(cfg.node.recycle_after_jobs, 100);
assert_eq!(cfg.node.max_queue_depth, 256);
}
#[test]
fn effective_getters_never_zero() {
let cfg = RuntimePoolLangConfig {
enabled: Some(true),
max_workers: 0,
idle_ttl_secs: 0,
recycle_after_jobs: 0,
max_queue_depth: 0,
};
assert_eq!(cfg.effective_max_workers(), 1);
assert_eq!(cfg.effective_max_queue_depth(), 1);
}
#[test]
fn explicit_enabled_overrides_language_default() {
// A partial python table without `enabled` keeps the python-off default.
let cfg: RuntimePoolConfig =
toml::from_str("[python]\nmax_workers = 4\n").expect("partial parses");
assert_eq!(cfg.python.enabled, None);
assert!(
!cfg.python.is_enabled(false),
"python stays off on a partial table"
);
// Explicit opt-in wins.
let on: RuntimePoolConfig =
toml::from_str("[python]\nenabled = true\n").expect("explicit parses");
assert!(
on.python.is_enabled(false),
"explicit enabled=true turns python on"
);
}
#[test]
fn deserializes_partial_toml_with_defaults() {
let cfg: RuntimePoolConfig = toml::from_str("enabled = true\n[node]\nmax_workers = 4\n")
.expect("partial runtime_pool config parses");
assert!(cfg.enabled);
assert_eq!(cfg.node.max_workers, 4);
// Unspecified fields fall back to defaults.
assert_eq!(cfg.node.idle_ttl_secs, 60);
assert_eq!(cfg.python.max_workers, 2);
}
}
+6
View File
@@ -415,6 +415,11 @@ pub struct Config {
#[serde(default)]
pub runtime_python: RuntimePythonConfig,
/// Shared language-runtime pool (long-lived `node`/`python` workers reused
/// across skill runs and `node_exec` instead of one child per run, #5106).
#[serde(default)]
pub runtime_pool: RuntimePoolConfig,
/// TokenJuice content-router / compaction configuration.
#[serde(default)]
pub tokenjuice: TokenjuiceConfig,
@@ -813,6 +818,7 @@ impl Default for Config {
subconscious_provider: None,
node: NodeConfig::default(),
runtime_python: RuntimePythonConfig::default(),
runtime_pool: RuntimePoolConfig::default(),
tokenjuice: TokenjuiceConfig::default(),
voice_server: VoiceServerConfig::default(),
voice_providers: Vec::new(),
+1
View File
@@ -113,6 +113,7 @@ pub mod referral;
pub mod rhai_workflows;
pub mod routing;
pub mod runtime_node;
pub mod runtime_pool;
pub mod runtime_python;
pub mod runtime_python_server;
pub mod sandbox;
+126
View File
@@ -0,0 +1,126 @@
//! Worker environment + harness-script materialisation helpers.
//!
//! Split out of `mod.rs` so the module root stays export-focused. Owns the
//! allow-listed child environment (`base_env`) and the once-per-process harness
//! write (`ensure_worker_script`).
use std::path::{Path, PathBuf};
use anyhow::{Context, Result};
use tokio::sync::OnceCell;
/// Env vars forwarded (allow-listed) into pooled workers. Mirrors the
/// `node_exec` / shell hygiene: secrets never leak into a worker's environment;
/// `PATH` is rebuilt separately with the managed interpreter's bin dir first.
const SAFE_ENV_VARS: &[&str] = &[
"HOME",
"TERM",
"LANG",
"LC_ALL",
"LC_CTYPE",
"USER",
"SHELL",
"TMPDIR",
// Windows process creation + child command lookup after env_clear().
"SystemRoot",
"WINDIR",
"COMSPEC",
"PATHEXT",
"TEMP",
"TMP",
"USERPROFILE",
"APPDATA",
"LOCALAPPDATA",
"ProgramFiles",
"ProgramFiles(x86)",
"ProgramW6432",
];
/// Build the allow-listed environment for a worker, with `bin_dir` prepended to
/// `PATH` so the child resolves the managed interpreter (and its tools).
pub(crate) fn base_env(bin_dir: &Path) -> Vec<(String, String)> {
let mut env: Vec<(String, String)> = Vec::new();
let host_path = std::env::var("PATH").unwrap_or_default();
let sep = if cfg!(windows) { ";" } else { ":" };
let path = if host_path.is_empty() {
bin_dir.to_string_lossy().into_owned()
} else {
format!("{}{}{}", bin_dir.display(), sep, host_path)
};
env.push(("PATH".to_string(), path));
for var in SAFE_ENV_VARS {
if let Ok(val) = std::env::var(var) {
env.push(((*var).to_string(), val));
}
}
env
}
/// Materialise a bundled harness script into a stable per-workspace cache path
/// and return it.
async fn write_worker_script(
workspace_dir: &Path,
filename: &str,
contents: &str,
) -> Result<PathBuf> {
let root = workspace_dir.join("runtime_pool");
tracing::debug!(dir = %root.display(), filename, "[runtime_pool] writing worker harness");
tokio::fs::create_dir_all(&root)
.await
.with_context(|| format!("creating runtime_pool cache {}", root.display()))?;
let path = root.join(filename);
tokio::fs::write(&path, contents)
.await
.with_context(|| format!("writing worker script {}", path.display()))?;
tracing::debug!(path = %path.display(), bytes = contents.len(), "[runtime_pool] worker harness ready");
Ok(path)
}
/// Return the harness script path, writing it **once per process** (a hot-path
/// `node_exec`/`python_exec` must not touch disk on every call — the point of
/// #5106 is to *reduce* per-run cost). The script is written on the first inline
/// exec and cached; a core upgrade is a fresh process, so it re-materialises
/// then, keeping the shipped harness current.
pub(crate) async fn ensure_worker_script(
cell: &'static OnceCell<PathBuf>,
workspace_dir: &Path,
filename: &str,
contents: &str,
) -> Result<PathBuf> {
Ok(cell
.get_or_try_init(|| write_worker_script(workspace_dir, filename, contents))
.await?
.clone())
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn base_env_prepends_bin_dir_to_path() {
let env = base_env(Path::new("/managed/bin"));
let path = env
.iter()
.find(|(k, _)| k == "PATH")
.map(|(_, v)| v.clone())
.expect("PATH present");
assert!(
path.starts_with("/managed/bin"),
"bin dir must be first on PATH; got {path}"
);
}
#[tokio::test]
async fn write_worker_script_roundtrips() {
let tmp = std::env::temp_dir().join(format!("rt-pool-test-{}", std::process::id()));
let path = write_worker_script(&tmp, "probe.js", "console.log('hi')")
.await
.expect("script written");
let read = tokio::fs::read_to_string(&path).await.unwrap();
assert_eq!(read, "console.log('hi')");
let _ = tokio::fs::remove_dir_all(&tmp).await;
}
}
+38
View File
@@ -0,0 +1,38 @@
//! Shared, bounded pools of long-lived `node` / `python` worker processes that
//! execute inline code jobs for skill runs and the `node_exec` agent tool —
//! instead of forking one interpreter child per execution (issue #5106).
//!
//! ## Why
//!
//! A single JS skill step spawns a `node` child at ~7275 MB RSS. At the
//! opencompany target (1001000 live agents in 2 GB / 2 vCPU) those per-run
//! interpreter children are the biggest budget breaker. Sharing a small bounded
//! pool of warm workers turns *K concurrent skill runs → K interpreters* into
//! *K concurrent skill runs → ~one pooled worker*, trading a little latency
//! (work beyond the pool size queues) for a large, flat memory floor.
//!
//! ## Shape
//!
//! * [`worker`] — one warm interpreter child speaking newline-delimited JSON.
//! * [`pool`] — the bounded [`LangPool`](pool::LangPool): semaphore-gated
//! concurrency, queue backpressure, idle-TTL reaping, recycle-after-N-jobs,
//! plus the process-global registry keyed per language.
//! * [`node`] / [`python`] — language backends that resolve the interpreter,
//! materialise the harness script, and submit inline jobs.
//! * [`env`] — allow-listed worker environment + once-per-process harness write.
//!
//! The whole subsystem is an **optimisation seam**: `runtime_pool.enabled =
//! false` (or a per-language flag) reverts callers to their legacy per-call
//! spawn with no behavioural change.
pub mod env;
pub mod node;
pub mod pool;
pub mod protocol;
pub mod python;
pub mod types;
pub mod worker;
pub(crate) use env::{base_env, ensure_worker_script};
pub use pool::{all_stats, LangPool, PoolRunError, PoolStats};
pub use types::{PoolExecOutcome, PoolLang, PoolSettings};
+364
View File
@@ -0,0 +1,364 @@
//! Node.js pool backend: resolve the interpreter, materialise the JS harness,
//! and submit inline jobs to the shared [`LangPool`](super::pool::LangPool).
use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::Result;
use tokio::sync::OnceCell;
use super::pool;
use super::types::{PoolExecOutcome, PoolLang, PoolSettings};
use super::worker::WorkerLaunch;
use crate::openhuman::config::{RuntimePoolConfig, RuntimePoolLangConfig};
/// The bundled Node worker harness (runs each inline job in an isolated
/// `worker_thread`, capturing its stdout/stderr and honouring a soft deadline).
const WORKER_JS: &str = include_str!("pool_worker.js");
/// Written once per process (see [`super::ensure_worker_script`]).
static NODE_SCRIPT: OnceCell<PathBuf> = OnceCell::const_new();
/// Whether inline `node` jobs should route through the pool. `node.enabled` is
/// already implied by the tool only being constructed when the node runtime is
/// enabled, so only the pool switches are checked here.
pub fn enabled(pool: &RuntimePoolConfig) -> bool {
// Node defaults ON: each job runs in an isolated worker_thread, so reuse is
// safe (fresh module graph + globals per job).
pool.enabled && pool.node.is_enabled(true)
}
/// Run inline JavaScript on a pooled, warm `node` worker.
///
/// `node_bin` / `bin_dir` come from the caller's already-resolved
/// [`ResolvedNode`](crate::openhuman::runtime_node::ResolvedNode). `workspace_dir`
/// and `lang_cfg` are injected at tool construction so this hot path never
/// re-reads config or re-writes the harness. `cwd` is the job's working
/// directory; `timeout` is the soft per-job deadline (`None` ⇒ run to completion).
pub async fn run_inline(
workspace_dir: &Path,
lang_cfg: &RuntimePoolLangConfig,
node_bin: &Path,
bin_dir: &Path,
code: String,
cwd: Option<PathBuf>,
timeout: Option<Duration>,
) -> Result<PoolExecOutcome, super::pool::PoolRunError> {
let script =
super::ensure_worker_script(&NODE_SCRIPT, workspace_dir, "pool_worker.js", WORKER_JS)
.await
.map_err(super::pool::PoolRunError::PreDispatch)?
.to_string_lossy()
.into_owned();
let env = super::base_env(bin_dir);
let launch = WorkerLaunch {
lang: PoolLang::Node,
bin: node_bin.to_path_buf(),
// `--experimental-vm-modules` lets the harness root dynamic `import()` at
// the job cwd (parity with `node -e`); the flag propagates to the per-job
// worker_thread via inherited `execArgv`.
args: vec![
"--experimental-vm-modules".to_string(),
"--experimental-import-meta-resolve".to_string(),
script,
],
env,
isolated_protocol: true,
};
let settings = PoolSettings::from_lang_config(lang_cfg);
let pool = pool::ensure_pool(launch, settings).await;
let cwd = cwd.map(|p| p.to_string_lossy().into_owned());
pool.run_inline(code, cwd, timeout).await
}
#[cfg(test)]
mod tests {
use super::*;
use crate::openhuman::config::Config;
use crate::openhuman::runtime_pool::{all_stats, PoolLang};
/// Resolve the host `node` binary + its bin dir, or `None` to skip.
fn system_node() -> Option<(std::path::PathBuf, std::path::PathBuf)> {
let out = std::process::Command::new("node")
.args(["-e", "process.stdout.write(process.execPath)"])
.output()
.ok()?;
if !out.status.success() {
return None;
}
let node_bin = std::path::PathBuf::from(String::from_utf8_lossy(&out.stdout).trim());
let bin_dir = node_bin.parent()?.to_path_buf();
Some((node_bin, bin_dir))
}
async fn node_spawns() -> u64 {
all_stats()
.await
.into_iter()
.find(|(lang, _)| *lang == PoolLang::Node)
.map(|(_, stats)| stats.worker_spawns)
.unwrap_or(0)
}
/// End-to-end: two inline jobs run on the pool, share ONE warm worker, and
/// surface stdout / exit codes exactly like the legacy path. Skips when no
/// system `node` is available (keeps CI hermetic on node-less runners).
#[tokio::test]
async fn pooled_node_runs_inline_and_reuses_worker() {
let Some((node_bin, bin_dir)) = system_node() else {
eprintln!("[runtime_pool] test skipped: no system node on PATH");
return;
};
let tmp = std::env::temp_dir().join(format!("rt-pool-node-e2e-{}", std::process::id()));
std::fs::create_dir_all(&tmp).unwrap();
let mut config = Config::default();
config.workspace_dir = tmp.clone();
config.runtime_pool.node.max_workers = 1;
config.runtime_pool.node.recycle_after_jobs = 0; // no recycle mid-test
let lang = config.runtime_pool.node.clone();
let spawns_before = node_spawns().await;
let out1 = run_inline(
&config.workspace_dir,
&lang,
&node_bin,
&bin_dir,
"console.log(JSON.stringify({ v: 6 * 7 }))".to_string(),
Some(tmp.clone()),
None,
)
.await
.expect("job 1 runs");
assert!(out1.success(), "job 1 should succeed: {out1:?}");
assert!(
out1.stdout.contains("\"v\":42"),
"stdout was {:?}",
out1.stdout
);
let out2 = run_inline(
&config.workspace_dir,
&lang,
&node_bin,
&bin_dir,
"throw new Error('nope')".to_string(),
Some(tmp.clone()),
None,
)
.await
.expect("job 2 runs");
assert!(!out2.success(), "throwing job should fail");
assert!(out2.stderr.contains("nope"), "stderr was {:?}", out2.stderr);
// cwd correctness: relative fs must resolve against the job's cwd, not
// the worker process's launch dir. Guards the host-chdir fix (a worker
// thread cannot chdir itself). Regression here = broken action sandbox.
std::fs::write(tmp.join("probe.txt"), "REL_OK").unwrap();
let out3 = run_inline(
&config.workspace_dir,
&lang,
&node_bin,
&bin_dir,
"const fs=require('fs'); process.stdout.write(fs.readFileSync('./probe.txt','utf8'))"
.to_string(),
Some(tmp.clone()),
None,
)
.await
.expect("job 3 runs");
assert!(out3.success(), "cwd-relative read should succeed: {out3:?}");
assert_eq!(out3.stdout, "REL_OK", "relative read resolved wrong cwd");
// fd-level writes must never share the protocol transport. This forged
// frame uses the real request id; on stdout-based framing Rust would
// accept it instead of the harness response and desynchronise the next
// job.
let out4 = run_inline(
&config.workspace_dir,
&lang,
&node_bin,
&bin_dir,
r#"
const fs = require('fs');
const { workerData } = require('worker_threads');
fs.writeSync(1, JSON.stringify({
id: workerData.id,
ok: true,
stdout: 'FORGED',
stderr: '',
exit_code: 0,
elapsed_ms: 0
}) + '\n');
console.log('REAL_RESPONSE');
"#
.to_string(),
Some(tmp.clone()),
None,
)
.await
.expect("fd-level output job runs");
assert!(
out4.success(),
"fd-level output job should succeed: {out4:?}"
);
assert_eq!(out4.stdout, "REAL_RESPONSE\n");
// Bare dynamic imports retain node -e semantics rather than being
// rewritten to a cwd-relative file URL.
let out5 = run_inline(
&config.workspace_dir,
&lang,
&node_bin,
&bin_dir,
"const fs = await import('fs'); console.log(typeof fs.readFileSync)".to_string(),
Some(tmp.clone()),
None,
)
.await
.expect("bare import job runs");
assert!(out5.success(), "bare import should succeed: {out5:?}");
assert_eq!(out5.stdout, "function\n");
// ESM-only packages expose an `import` condition but no CommonJS
// `require` condition. Bare imports must therefore use Node's ESM
// resolver rooted at the job cwd.
let esm_package = tmp.join("node_modules/esm-only");
std::fs::create_dir_all(&esm_package).unwrap();
std::fs::write(
esm_package.join("package.json"),
r#"{"name":"esm-only","type":"module","exports":{"import":"./index.mjs"}}"#,
)
.unwrap();
std::fs::write(
esm_package.join("index.mjs"),
"export const marker = 'ESM_ONLY_OK';",
)
.unwrap();
let esm_out = run_inline(
&config.workspace_dir,
&lang,
&node_bin,
&bin_dir,
"const { marker } = await import('esm-only'); console.log(marker)".to_string(),
Some(tmp.clone()),
None,
)
.await
.expect("ESM-only package import runs");
assert!(
esm_out.success(),
"ESM-only package import should succeed: {esm_out:?}"
);
assert_eq!(esm_out.stdout, "ESM_ONLY_OK\n");
// vm dynamic-import hooks receive attributes separately; forward them
// so JSON modules preserve legacy node -e behavior.
std::fs::write(tmp.join("data.json"), r#"{"answer":42}"#).unwrap();
let json_out = run_inline(
&config.workspace_dir,
&lang,
&node_bin,
&bin_dir,
"const data = await import('./data.json', { with: { type: 'json' } }); console.log(data.default.answer)"
.to_string(),
Some(tmp.clone()),
None,
)
.await
.expect("JSON import with attributes runs");
assert!(
json_out.success(),
"JSON import with attributes should succeed: {json_out:?}"
);
assert_eq!(json_out.stdout, "42\n");
// User warnings are tool output, not harness noise. Never suppress them
// globally on the pooled worker.
let warning_out = run_inline(
&config.workspace_dir,
&lang,
&node_bin,
&bin_dir,
"process.emitWarning('POOL_WARNING')".to_string(),
Some(tmp.clone()),
None,
)
.await
.expect("warning job runs");
assert!(warning_out.success(), "warning job failed: {warning_out:?}");
assert!(
warning_out.stderr.contains("POOL_WARNING"),
"user warning was hidden: {:?}",
warning_out.stderr
);
// The protocol never shares fd 0 with user code. Legacy Command::output
// supplies EOF on stdin, so pooled code must do the same rather than
// blocking on or consuming the next NDJSON request.
let stdin_out = run_inline(
&config.workspace_dir,
&lang,
&node_bin,
&bin_dir,
"const fs=require('fs'); console.log(JSON.stringify(fs.readFileSync(0,'utf8')))"
.to_string(),
Some(tmp.clone()),
Some(Duration::from_secs(2)),
)
.await
.expect("stdin EOF job runs");
assert!(stdin_out.success(), "stdin should be EOF: {stdin_out:?}");
assert_eq!(stdin_out.stdout, "\"\"\n");
// A missing cwd is a harness error. Running in the worker's inherited
// cwd would escape the requested action root and diverge from legacy
// Command::current_dir behavior.
let missing_cwd = tmp.join("deleted-action-root");
let err = run_inline(
&config.workspace_dir,
&lang,
&node_bin,
&bin_dir,
"require('fs').writeFileSync('must-not-exist.txt', 'bad')".to_string(),
Some(missing_cwd.clone()),
None,
)
.await
.expect_err("missing cwd must fail closed");
assert!(
err.to_string().contains("failed to set worker cwd"),
"unexpected missing-cwd error: {err}"
);
assert!(!tmp.join("must-not-exist.txt").exists());
// The harness-level cwd error must not poison framing for the next job.
let out6 = run_inline(
&config.workspace_dir,
&lang,
&node_bin,
&bin_dir,
"console.log('AFTER_CWD_ERROR')".to_string(),
Some(tmp.clone()),
None,
)
.await
.expect("job after cwd error runs");
assert_eq!(out6.stdout, "AFTER_CWD_ERROR\n");
// At most one NEW worker spawned for all jobs ⇒ the warm worker was
// reused. Measured as a delta so prior global pool state can't skew it.
let spawns_after = node_spawns().await;
assert!(
spawns_after - spawns_before <= 1,
"expected warm-worker reuse: {} new spawns",
spawns_after - spawns_before
);
let _ = std::fs::remove_dir_all(&tmp);
}
}
+435
View File
@@ -0,0 +1,435 @@
//! The bounded per-language worker pool and its process-global registry.
//!
//! One [`LangPool`] owns up to `max_workers` warm [`PoolWorker`]s for a single
//! language. Concurrency is bounded by a semaphore; submissions beyond the pool
//! size **queue** on that semaphore (the intended backpressure), and
//! submissions beyond `max_workers + max_queue_depth` in flight are rejected so
//! memory can't grow without bound. Idle workers are reaped after a TTL; busy
//! workers are recycled after N jobs.
use std::collections::HashMap;
use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
use std::sync::{Arc, OnceLock, Weak};
use std::time::{Duration, Instant};
use anyhow::Result;
use tokio::sync::{Mutex, Semaphore};
use super::protocol::{PoolJobRequest, PoolJobResponse};
use super::types::{PoolExecOutcome, PoolLang, PoolSettings};
use super::worker::{PoolWorker, WorkerLaunch};
/// Why a pooled run failed, classified so callers know whether a retry or a
/// legacy per-call-spawn fallback is safe.
#[derive(Debug)]
pub enum PoolRunError {
/// In-flight work exceeded `max_workers + max_queue_depth`; the pool shed
/// load rather than buffering unbounded. Callers must **not** fall back to a
/// per-call spawn — that reintroduces the very RSS the pool caps (#5106) —
/// but surface a busy error or retry later.
Saturated,
/// Failure before the job reached a worker (serialise / spawn / write). The
/// job never ran, so a retry or a legacy-spawn fallback is safe.
PreDispatch(anyhow::Error),
/// Failure after the job was dispatched (it may have executed). Terminal —
/// the caller must not re-run it (would duplicate side effects).
PostDispatch(anyhow::Error),
}
impl std::fmt::Display for PoolRunError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
PoolRunError::Saturated => write!(f, "runtime pool at capacity"),
PoolRunError::PreDispatch(e) => write!(f, "pre-dispatch pool failure: {e:#}"),
PoolRunError::PostDispatch(e) => write!(f, "post-dispatch pool failure: {e:#}"),
}
}
}
/// Extra grace added to a job's soft deadline before the Rust side treats the
/// worker as wedged and kills it. The worker should always self-abort first.
const HARD_TIMEOUT_GRACE: Duration = Duration::from_secs(10);
/// Lower bound on how often the idle reaper wakes, so a tiny TTL doesn't busy-loop.
const MIN_REAP_INTERVAL: Duration = Duration::from_secs(5);
/// Lightweight, cloneable snapshot of a pool's counters (for status/tests).
#[derive(Debug, Clone, Default)]
pub struct PoolStats {
pub jobs_total: u64,
pub worker_spawns: u64,
pub rejected_saturated: u64,
pub idle_workers: usize,
pub max_workers: usize,
}
pub struct LangPool {
launch: WorkerLaunch,
settings: PoolSettings,
/// Permits == max_workers. Acquiring one is the queue/backpressure gate.
permits: Arc<Semaphore>,
/// Warm, idle workers available for reuse.
idle: Mutex<Vec<PoolWorker>>,
/// Jobs currently in flight or waiting for a permit (for saturation guard).
inflight: AtomicUsize,
job_seq: AtomicU64,
jobs_total: AtomicU64,
worker_spawns: AtomicU64,
rejected_saturated: AtomicU64,
}
impl LangPool {
/// Build a pool and start its idle reaper. The reaper holds only a `Weak`
/// ref, so the pool is still dropped normally when the registry evicts it.
pub fn start(launch: WorkerLaunch, settings: PoolSettings) -> Arc<Self> {
let permits = Arc::new(Semaphore::new(settings.max_workers));
let pool = Arc::new(Self {
launch,
settings,
permits,
idle: Mutex::new(Vec::new()),
inflight: AtomicUsize::new(0),
job_seq: AtomicU64::new(0),
jobs_total: AtomicU64::new(0),
worker_spawns: AtomicU64::new(0),
rejected_saturated: AtomicU64::new(0),
});
if let Some(ttl) = pool.settings.idle_ttl {
spawn_reaper(Arc::downgrade(&pool), ttl);
}
pool
}
pub fn lang(&self) -> PoolLang {
self.launch.lang
}
pub async fn stats(&self) -> PoolStats {
PoolStats {
jobs_total: self.jobs_total.load(Ordering::Relaxed),
worker_spawns: self.worker_spawns.load(Ordering::Relaxed),
rejected_saturated: self.rejected_saturated.load(Ordering::Relaxed),
idle_workers: self.idle.lock().await.len(),
max_workers: self.settings.max_workers,
}
}
/// Run one inline job, blocking (asynchronously) until a worker is free.
pub async fn run_inline(
&self,
code: String,
cwd: Option<String>,
timeout: Option<Duration>,
) -> Result<PoolExecOutcome, PoolRunError> {
// Saturation guard: bound total in-flight (running + queued) work so a
// stampede queues up to a point, then sheds load instead of buffering
// unbounded. Capacity = worker slots + allowed queue depth.
let capacity = self.settings.max_workers + self.settings.max_queue_depth;
let inflight_now = self.inflight.fetch_add(1, Ordering::AcqRel) + 1;
if inflight_now > capacity {
self.inflight.fetch_sub(1, Ordering::AcqRel);
self.rejected_saturated.fetch_add(1, Ordering::Relaxed);
tracing::warn!(
lang = self.launch.lang.id(),
inflight = inflight_now,
capacity,
"[runtime_pool] saturated; shedding load (no spawn fallback)"
);
return Err(PoolRunError::Saturated);
}
// Ensure the in-flight counter is released on every exit path below.
let _inflight_guard = InflightGuard(&self.inflight);
let wait_start = Instant::now();
let _permit = self
.permits
.acquire()
.await
.expect("runtime pool semaphore never closed");
let queue_wait = wait_start.elapsed();
let id = self.job_seq.fetch_add(1, Ordering::Relaxed).to_string();
let req = PoolJobRequest {
id,
kind: "inline".to_string(),
code: Some(code),
cwd,
timeout_ms: timeout.map(|t| t.as_millis() as u64),
};
let hard_timeout = timeout.map(|t| t + HARD_TIMEOUT_GRACE);
let job_start = Instant::now();
let (response, worker) = self.submit_with_retry(&req, hard_timeout).await?;
let elapsed = job_start.elapsed();
self.jobs_total.fetch_add(1, Ordering::Relaxed);
// Recycle after N jobs, otherwise return the warm worker to the pool.
if worker.should_recycle(self.settings.recycle_after_jobs) {
tracing::debug!(
lang = self.launch.lang.id(),
jobs = worker.jobs_done(),
"[runtime_pool] recycling worker after job budget"
);
worker.shutdown();
} else {
self.idle.lock().await.push(worker);
}
if let Some(err) = response.error {
// The worker replied with a harness-level error: the job was
// dispatched (and may have run), so this is terminal.
return Err(PoolRunError::PostDispatch(anyhow::anyhow!(
"{} worker error: {err}",
self.launch.lang.id()
)));
}
Ok(PoolExecOutcome {
stdout: response.stdout,
stderr: response.stderr,
exit_code: response.exit_code,
timed_out: response.timed_out,
elapsed,
queue_wait,
})
}
/// Submit on a warm-or-fresh worker. A **pre-dispatch** failure (e.g. a
/// reused idle worker that died on write) respawns once — the job never ran,
/// so that is safe. A **post-dispatch** failure is terminal: the job may have
/// executed, so it is never re-run (no duplicate side effects). Returns the
/// surviving worker so the caller can recycle or re-pool it.
async fn submit_with_retry(
&self,
req: &PoolJobRequest,
hard_timeout: Option<Duration>,
) -> Result<(PoolJobResponse, PoolWorker), PoolRunError> {
let mut worker = self
.take_or_spawn()
.await
.map_err(PoolRunError::PreDispatch)?;
match worker.submit(req, hard_timeout).await {
Ok(resp) => Ok((resp, worker)),
Err(e) if !e.dispatched => {
tracing::warn!(
lang = self.launch.lang.id(),
"[runtime_pool] pre-dispatch submit failure ({e}); respawning once"
);
worker.shutdown();
let mut fresh = self
.spawn_worker()
.await
.map_err(PoolRunError::PreDispatch)?;
match fresh.submit(req, hard_timeout).await {
Ok(resp) => Ok((resp, fresh)),
Err(e2) if !e2.dispatched => Err(PoolRunError::PreDispatch(e2.err)),
Err(e2) => Err(PoolRunError::PostDispatch(e2.err)),
}
}
Err(e) => {
tracing::warn!(
lang = self.launch.lang.id(),
"[runtime_pool] post-dispatch submit failure ({e}); terminal, not retrying"
);
worker.shutdown();
Err(PoolRunError::PostDispatch(e.err))
}
}
}
/// Pop a still-fresh idle worker, or spawn a new one.
async fn take_or_spawn(&self) -> Result<PoolWorker> {
{
let mut idle = self.idle.lock().await;
while let Some(worker) = idle.pop() {
if let Some(ttl) = self.settings.idle_ttl {
if worker.idle_expired(ttl) {
worker.shutdown();
continue;
}
}
return Ok(worker);
}
}
self.spawn_worker().await
}
async fn spawn_worker(&self) -> Result<PoolWorker> {
self.worker_spawns.fetch_add(1, Ordering::Relaxed);
PoolWorker::spawn(&self.launch).await
}
/// Drop workers idle beyond the TTL. Called by the background reaper.
async fn reap_idle(&self) {
let Some(ttl) = self.settings.idle_ttl else {
return;
};
let mut idle = self.idle.lock().await;
let before = idle.len();
let mut kept = Vec::with_capacity(before);
for worker in idle.drain(..) {
if worker.idle_expired(ttl) {
worker.shutdown();
} else {
kept.push(worker);
}
}
let reaped = before - kept.len();
*idle = kept;
if reaped > 0 {
tracing::debug!(
lang = self.launch.lang.id(),
reaped,
"[runtime_pool] idle reaper retired workers"
);
}
}
}
/// Decrements the in-flight counter on drop so every early return is covered.
struct InflightGuard<'a>(&'a AtomicUsize);
impl Drop for InflightGuard<'_> {
fn drop(&mut self) {
self.0.fetch_sub(1, Ordering::AcqRel);
}
}
fn spawn_reaper(pool: Weak<LangPool>, ttl: Duration) {
let interval = ttl.max(MIN_REAP_INTERVAL);
tokio::spawn(async move {
loop {
tokio::time::sleep(interval).await;
match pool.upgrade() {
Some(pool) => pool.reap_idle().await,
None => break, // pool dropped — stop reaping
}
}
});
}
// ---------------------------------------------------------------------------
// Process-global registry
// ---------------------------------------------------------------------------
struct CachedPool {
key: String,
pool: Arc<LangPool>,
}
static REGISTRY: OnceLock<Mutex<HashMap<PoolLangKey, CachedPool>>> = OnceLock::new();
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
enum PoolLangKey {
Node,
Python,
}
impl From<PoolLang> for PoolLangKey {
fn from(lang: PoolLang) -> Self {
match lang {
PoolLang::Node => PoolLangKey::Node,
PoolLang::Python => PoolLangKey::Python,
}
}
}
fn registry() -> &'static Mutex<HashMap<PoolLangKey, CachedPool>> {
REGISTRY.get_or_init(|| Mutex::new(HashMap::new()))
}
/// Fingerprint that decides whether a cached pool can be reused. Any change to
/// the interpreter path, args, or tuning rebuilds the pool.
fn launch_key(launch: &WorkerLaunch, settings: &PoolSettings) -> String {
format!(
"{}|{}|{:?}|isolated={}|w={}|ttl={:?}|recycle={}|q={}",
launch.lang.id(),
launch.bin.display(),
launch.args,
launch.isolated_protocol,
settings.max_workers,
settings.idle_ttl,
settings.recycle_after_jobs,
settings.max_queue_depth,
)
}
/// Get (or build) the process-global pool for a language, keyed by its launch
/// fingerprint. A config or interpreter change transparently rebuilds it.
pub async fn ensure_pool(launch: WorkerLaunch, settings: PoolSettings) -> Arc<LangPool> {
let key = launch_key(&launch, &settings);
let lang_key = PoolLangKey::from(launch.lang);
let mut reg = registry().lock().await;
if let Some(cached) = reg.get(&lang_key) {
if cached.key == key {
return cached.pool.clone();
}
tracing::info!(
lang = launch.lang.id(),
"[runtime_pool] launch spec changed; rebuilding pool"
);
}
let pool = LangPool::start(launch, settings);
reg.insert(
lang_key,
CachedPool {
key,
pool: pool.clone(),
},
);
pool
}
/// Snapshot every live pool's stats (for a status surface / debugging).
pub async fn all_stats() -> Vec<(PoolLang, PoolStats)> {
let reg = registry().lock().await;
let mut out = Vec::new();
for cached in reg.values() {
out.push((cached.pool.lang(), cached.pool.stats().await));
}
out
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn launch_key_changes_with_tuning() {
let launch = WorkerLaunch {
lang: PoolLang::Node,
bin: "/usr/bin/node".into(),
args: vec!["worker.js".into()],
env: vec![],
isolated_protocol: true,
};
let a = PoolSettings {
max_workers: 2,
idle_ttl: Some(Duration::from_secs(60)),
recycle_after_jobs: 100,
max_queue_depth: 256,
};
let mut b = a.clone();
b.max_workers = 4;
assert_ne!(launch_key(&launch, &a), launch_key(&launch, &b));
assert_eq!(launch_key(&launch, &a), launch_key(&launch, &a));
}
#[test]
fn pool_run_error_display_is_classified() {
// The three arms drive distinct caller behaviour (retry / fall back /
// give up), so their rendered messages must stay distinguishable.
assert_eq!(
PoolRunError::Saturated.to_string(),
"runtime pool at capacity"
);
let pre = PoolRunError::PreDispatch(anyhow::anyhow!("spawn failed")).to_string();
assert!(pre.starts_with("pre-dispatch pool failure:"), "got {pre}");
assert!(pre.contains("spawn failed"));
let post = PoolRunError::PostDispatch(anyhow::anyhow!("read wedged")).to_string();
assert!(
post.starts_with("post-dispatch pool failure:"),
"got {post}"
);
assert!(post.contains("read wedged"));
}
}
+292
View File
@@ -0,0 +1,292 @@
// OpenHuman runtime-pool Node worker harness (issue #5106).
//
// A single long-lived `node` process that executes inline JavaScript jobs for
// many skill runs / node_exec calls, so the fleet pays one warm interpreter
// instead of one child per run.
//
// Protocol (newline-delimited JSON over an authenticated loopback socket,
// see runtime_pool/protocol.rs):
// 1. Print exactly one ready line: {"ready":true,"protocol":1,"lang":"node"}
// 2. For each request line {id,kind:"inline",code,cwd,timeout_ms} reply with
// {id,ok,stdout,stderr,exit_code,timed_out,elapsed_ms,error}.
//
// Each job runs in its own `worker_thread` for isolation (fresh module graph +
// globals per run) and safe termination (a runaway or process.exit()-y job is
// killed with worker.terminate() without taking down this host process). The
// job's stdout/stderr are isolated pipes (stdout:true/stderr:true). Protocol
// replies use a separate authenticated loopback socket so fd-level writes
// (`fs.writeSync`, inherited child stdio) cannot forge or corrupt frames.
'use strict';
const { Worker, isMainThread, parentPort, workerData } = require('worker_threads');
const PROTOCOL_VERSION = 1;
// ---------------------------------------------------------------------------
// Worker-thread mode: execute one job's code, then exit (flushing its pipes).
// ---------------------------------------------------------------------------
if (!isMainThread) {
const path = require('path');
const vm = require('vm');
const { createRequire } = require('module');
const { pathToFileURL } = require('url');
async function runUserCode(code, cwd) {
// NOTE: do NOT `process.chdir()` here — it throws ERR_WORKER_UNSUPPORTED_
// OPERATION inside a worker thread. The host chdirs before spawning this
// worker (jobs serialize per worker process), so `process.cwd()` is already
// the job's directory; `cwd` roots require/__dirname and the import base.
const dir = cwd || process.cwd();
const filename = path.join(dir, 'inline.js');
const req = createRequire(filename);
const base = pathToFileURL(filename).href;
// Use the ESM resolver (not createRequire.resolve) for bare dynamic
// imports so import-only package exports retain `node -e` semantics.
const { default: resolveEsm } = await import(
'data:text/javascript,export default (specifier, parent) => import.meta.resolve(specifier, parent)'
);
const importFromJob = (specifier, _referrer, importAttributes) => {
const options =
importAttributes && Object.keys(importAttributes).length > 0
? { with: importAttributes }
: undefined;
if (
specifier.startsWith('.') ||
specifier.startsWith('/') ||
specifier.startsWith('file:') ||
specifier.startsWith('data:')
) {
return import(new URL(specifier, base).href, options);
}
return import(resolveEsm(specifier, base), options);
};
// Mimic `node -e`: CommonJS-ish sloppy scope with require/__dirname, wrapped
// in an async IIFE so top-level `await` works. `vm.compileFunction` (over a
// bare `new Function`) lets us root dynamic `import()` at the job cwd via
// `importModuleDynamically`, so `await import('./rel.mjs')` resolves like
// `node -e` instead of relative to this harness file. Needs
// `--experimental-vm-modules` (passed on the worker launch).
const fn = vm.compileFunction(
'return (async () => {\n' + code + '\n})();',
['require', '__filename', '__dirname', 'module', 'exports'],
{
filename,
importModuleDynamically: importFromJob,
}
);
const mod = { exports: {} };
await fn(req, filename, dir, mod, mod.exports);
}
const code = (workerData && workerData.code) || '';
const cwd = (workerData && workerData.cwd) || null;
runUserCode(code, cwd).then(
() => {
// Resolve → let the thread exit naturally once its loop drains, which
// flushes the stdout/stderr pipes before the 'exit' event fires.
},
(err) => {
const msg = err && err.stack ? err.stack : String(err);
process.stderr.write(msg + '\n');
process.exitCode = 1;
}
);
return;
}
// ---------------------------------------------------------------------------
// Main (host) mode: read jobs, run each in a worker thread, reply per job.
// ---------------------------------------------------------------------------
function collect(stream) {
return new Promise((resolve) => {
let buf = '';
stream.setEncoding('utf8');
stream.on('data', (d) => {
buf += d;
});
const done = () => resolve(buf);
stream.on('end', done);
stream.on('close', done);
stream.on('error', done);
});
}
function runJob(job) {
return new Promise((resolve) => {
const start = Date.now();
// Set the job's working directory on the HOST before spawning the worker:
// a worker thread inherits the parent's cwd at creation and cannot chdir
// itself. The worker captures cwd synchronously at construction, so we
// restore the host's prior cwd immediately after — otherwise a later job
// without `cwd` (or whose chdir failed) would silently inherit this job's
// directory instead of the worker's original one.
const priorCwd = process.cwd();
if (job.cwd) {
try {
process.chdir(job.cwd);
} catch (e) {
resolve({
id: job.id,
ok: false,
stdout: '',
stderr: '',
exit_code: null,
timed_out: false,
elapsed_ms: Date.now() - start,
error: 'failed to set worker cwd: ' + (e && e.stack ? e.stack : String(e)),
});
return;
}
}
let worker;
try {
worker = new Worker(__filename, {
workerData: { id: job.id, code: job.code || '', cwd: job.cwd || null },
stdout: true,
stderr: true,
// Propagate host node flags (e.g. --experimental-vm-modules) so the
// worker's vm.compileFunction dynamic-import hook is enabled.
execArgv: process.execArgv,
});
} catch (e) {
try {
process.chdir(priorCwd);
} catch (_e) {
/* best-effort restore */
}
resolve({
id: job.id,
ok: false,
stdout: '',
stderr: '',
exit_code: null,
timed_out: false,
elapsed_ms: Date.now() - start,
error: 'failed to spawn worker thread: ' + (e && e.stack ? e.stack : String(e)),
});
return;
}
const outP = collect(worker.stdout);
const errP = collect(worker.stderr);
let exitCode = 0;
let timedOut = false;
let extraErr = '';
let timer = null;
if (job.timeout_ms && job.timeout_ms > 0) {
timer = setTimeout(() => {
timedOut = true;
worker.terminate();
}, job.timeout_ms);
}
worker.on('error', (e) => {
extraErr += (e && e.stack ? e.stack : String(e)) + '\n';
if (!exitCode) exitCode = 1;
});
worker.on('exit', async (code) => {
if (timer) clearTimeout(timer);
// Restore the host cwd only now: a worker thread reads its cwd
// asynchronously as it initializes, so the host must stay at `job.cwd`
// for the worker's whole life. Jobs are serialized, so the next job
// starts from this restored (prior) directory rather than inheriting
// this one's.
try {
process.chdir(priorCwd);
} catch (_e) {
/* best-effort restore */
}
if (code && !exitCode) exitCode = code;
const stdout = await outP;
const stderr = (await errP) + extraErr;
resolve({
id: job.id,
ok: true,
stdout,
stderr,
exit_code: timedOut ? null : exitCode,
timed_out: timedOut,
elapsed_ms: Date.now() - start,
error: null,
});
});
});
}
let protocolInput = process.stdin;
let protocolStream = process.stdout;
function reply(obj) {
protocolStream.write(JSON.stringify(obj) + '\n');
}
function serve() {
// Announce readiness, then serve jobs one at a time (the Rust pool already
// sends at most one outstanding job per worker; the chain keeps ordering).
reply({
ready: true,
protocol: PROTOCOL_VERSION,
lang: 'node',
protocol_token: process.env.OPENHUMAN_RUNTIME_POOL_PROTOCOL_TOKEN || null,
});
const readline = require('readline');
const rl = readline.createInterface({ input: protocolInput });
let chain = Promise.resolve();
rl.on('line', (line) => {
const trimmed = line.trim();
if (!trimmed) return;
let job;
try {
job = JSON.parse(trimmed);
} catch (_e) {
return; // ignore unparseable lines
}
chain = chain
.then(() => runJob(job))
.then((res) => reply(res))
.catch((e) => {
reply({
id: job && job.id,
ok: false,
stdout: '',
stderr: '',
exit_code: null,
timed_out: false,
elapsed_ms: 0,
error: String((e && e.stack) || e),
});
});
});
rl.on('close', () => {
// Drain any in-flight job before exiting so a closed stdin doesn't drop work
// that was already accepted onto the chain.
Promise.resolve(chain).finally(() => process.exit(0));
});
}
const protocolAddr = process.env.OPENHUMAN_RUNTIME_POOL_PROTOCOL_ADDR;
if (protocolAddr) {
const net = require('net');
const split = protocolAddr.lastIndexOf(':');
const host = protocolAddr.slice(0, split);
const port = Number(protocolAddr.slice(split + 1));
const socket = net.createConnection({ host, port });
socket.once('connect', () => {
protocolInput = socket;
protocolStream = socket;
serve();
});
socket.once('error', (e) => {
process.stderr.write('runtime pool protocol connection failed: ' + String(e) + '\n');
process.exit(1);
});
} else {
// Backward-compatible developer launch; production Node workers always use
// the isolated socket configured by Rust.
serve();
}
+212
View File
@@ -0,0 +1,212 @@
# OpenHuman runtime-pool Python worker harness (issue #5106).
#
# A single long-lived `python` process that executes inline Python jobs for many
# skill runs, so the fleet pays one warm interpreter instead of one child per
# run.
#
# Protocol (newline-delimited JSON over an authenticated loopback socket,
# see runtime_pool/protocol.rs):
# 1. Print exactly one ready line: {"ready":true,"protocol":1,"lang":"python"}
# 2. For each request line {id,kind:"inline",code,cwd,timeout_ms} reply with
# {id,ok,stdout,stderr,exit_code,timed_out,elapsed_ms,error}.
#
# Each job runs in this interpreter with stdout/stderr redirected into buffers so
# a job's prints never corrupt the protocol stream. Isolation is per-job globals
# plus the pool's recycle-after-N-jobs; CPython cannot safely kill a running
# thread, so the soft deadline is best-effort SIGALRM on Unix and otherwise the
# Rust side's hard deadline kills + respawns the worker.
import sys
import os
import json
import time
import tempfile
import traceback
PROTOCOL_VERSION = 1
# Production workers use one authenticated duplex socket for protocol traffic,
# leaving fd 0 at EOF and fd 1/2 entirely available to job capture. The stdio
# fallback keeps the harness convenient to launch by hand.
_PROTOCOL_TOKEN = os.environ.get("OPENHUMAN_RUNTIME_POOL_PROTOCOL_TOKEN")
_PROTOCOL_ADDR = os.environ.get("OPENHUMAN_RUNTIME_POOL_PROTOCOL_ADDR")
_PROTOCOL_SOCKET = None
if _PROTOCOL_ADDR:
import socket
_host, _port = _PROTOCOL_ADDR.rsplit(":", 1)
_PROTOCOL_SOCKET = socket.create_connection((_host, int(_port)))
_PROTO_IN = _PROTOCOL_SOCKET.makefile("r")
_PROTO = _PROTOCOL_SOCKET.makefile("w", buffering=1)
else:
# Private duplicates prevent per-job fd redirection from touching protocol.
_PROTO_IN = os.fdopen(os.dup(0), "r", buffering=1)
_PROTO = os.fdopen(os.dup(1), "w", buffering=1)
try:
import signal
_HAVE_ALARM = hasattr(signal, "SIGALRM") and hasattr(signal, "setitimer")
except Exception: # pragma: no cover - platform without signal
signal = None
_HAVE_ALARM = False
class _JobTimeout(Exception):
pass
def _run_job(job):
code = job.get("code") or ""
cwd = job.get("cwd")
timeout_ms = job.get("timeout_ms")
start = time.time()
exit_code = 0
timed_out = False
extra_err = ""
old_cwd = None
if cwd:
try:
old_cwd = os.getcwd()
os.chdir(cwd)
except Exception as exc:
# Match subprocess cwd semantics: if the requested action root
# cannot be entered, user code must not run in this long-lived
# worker's inherited directory.
return {
"id": job.get("id"),
"ok": False,
"stdout": "",
"stderr": "",
"exit_code": None,
"timed_out": False,
"elapsed_ms": int((time.time() - start) * 1000),
"error": f"failed to set worker cwd: {exc!r}",
}
# Capture at the FILE-DESCRIPTOR level (not just `sys.stdout`) so
# `os.write(1, ...)`, subprocesses, and native extensions are captured too —
# otherwise they would leak onto the real stdout, which is the NDJSON
# protocol channel. Temp files (vs pipes) avoid buffer-deadlock on large
# output.
in_f = tempfile.TemporaryFile(mode="w+b")
out_f = tempfile.TemporaryFile(mode="w+b")
err_f = tempfile.TemporaryFile(mode="w+b")
saved_in = os.dup(0)
saved_out = os.dup(1)
saved_err = os.dup(2)
os.dup2(in_f.fileno(), 0)
os.dup2(out_f.fileno(), 1)
os.dup2(err_f.fileno(), 2)
armed = False
if _HAVE_ALARM and timeout_ms and timeout_ms > 0:
def _on_alarm(_signum, _frame):
raise _JobTimeout()
signal.signal(signal.SIGALRM, _on_alarm)
signal.setitimer(signal.ITIMER_REAL, timeout_ms / 1000.0)
armed = True
try:
# Fresh globals per job so top-level names don't leak between runs.
g = {"__name__": "__main__", "__builtins__": __builtins__}
exec(compile(code, "<inline>", "exec"), g, g)
except _JobTimeout:
timed_out = True
except SystemExit as e: # honour sys.exit(n)
if e.code is None:
exit_code = 0
elif isinstance(e.code, int):
exit_code = e.code
else:
exit_code = 1
extra_err = str(e.code) + "\n"
except BaseException: # noqa: BLE001 - surface any job failure to the caller
exit_code = 1
extra_err = traceback.format_exc()
finally:
if armed:
signal.setitimer(signal.ITIMER_REAL, 0)
# Flush Python's buffers to the redirected fds, then restore the real
# stdout/stderr before reading the captures. A flush failure is surfaced
# in the job's stderr rather than silently discarded.
try:
sys.stdout.flush()
except Exception as flush_err: # noqa: BLE001
extra_err += f"[harness] stdout flush failed: {flush_err!r}\n"
try:
sys.stderr.flush()
except Exception as flush_err: # noqa: BLE001
extra_err += f"[harness] stderr flush failed: {flush_err!r}\n"
os.dup2(saved_in, 0)
os.dup2(saved_out, 1)
os.dup2(saved_err, 2)
os.close(saved_in)
os.close(saved_out)
os.close(saved_err)
if old_cwd is not None:
try:
os.chdir(old_cwd)
except Exception:
pass
in_f.close()
out_f.seek(0)
err_f.seek(0)
stdout = out_f.read().decode("utf-8", "replace")
stderr = err_f.read().decode("utf-8", "replace") + extra_err
out_f.close()
err_f.close()
return {
"id": job.get("id"),
"ok": True,
"stdout": stdout,
"stderr": stderr,
"exit_code": None if timed_out else exit_code,
"timed_out": timed_out,
"elapsed_ms": int((time.time() - start) * 1000),
"error": None,
}
def _reply(obj):
_PROTO.write(json.dumps(obj) + "\n")
_PROTO.flush()
def main():
_reply({
"ready": True,
"protocol": PROTOCOL_VERSION,
"lang": "python",
"protocol_token": _PROTOCOL_TOKEN,
})
for line in _PROTO_IN:
line = line.strip()
if not line:
continue
try:
job = json.loads(line)
except Exception:
continue # ignore unparseable lines
try:
res = _run_job(job)
except Exception as e: # harness-level failure
res = {
"id": job.get("id") if isinstance(job, dict) else None,
"ok": False,
"stdout": "",
"stderr": "",
"exit_code": None,
"timed_out": False,
"elapsed_ms": 0,
"error": repr(e),
}
_reply(res)
if __name__ == "__main__":
main()
+125
View File
@@ -0,0 +1,125 @@
//! Wire protocol between the core and a pooled language worker.
//!
//! Newline-delimited JSON over a per-worker duplex transport, mirroring the
//! [`runtime_python_server`](crate::openhuman::runtime_python_server) protocol.
//! Production workers use an authenticated loopback socket so job fd 0/1/2
//! cannot consume or corrupt protocol traffic:
//!
//! 1. On startup the worker prints exactly one [`PoolReadyLine`].
//! 2. The core writes one [`PoolJobRequest`] per line; the worker replies with
//! one [`PoolJobResponse`] per line, correlated by `id`.
//!
//! Child stdout/stderr are drained separately and never carry protocol frames.
use serde::{Deserialize, Serialize};
/// Bumped whenever the request/response shape changes incompatibly. The worker
/// echoes the version it speaks in its ready line; a mismatch fails the launch.
pub const PROTOCOL_VERSION: u32 = 1;
/// Handshake line printed once by the worker on startup.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PoolReadyLine {
#[serde(default)]
pub ready: bool,
#[serde(default)]
pub protocol: Option<u32>,
/// `"node"` / `"python"` — a sanity check that the right harness launched.
#[serde(default)]
pub lang: Option<String>,
#[serde(default)]
pub error: Option<String>,
/// Optional per-launch secret used when the protocol travels over an
/// isolated loopback socket instead of stdout.
#[serde(default)]
pub protocol_token: Option<String>,
}
/// A single unit of work sent to a worker.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PoolJobRequest {
/// Correlation id — the worker echoes it back in the response.
pub id: String,
/// Job kind. Today only `"inline"` (evaluate `code`); reserved for future
/// `"script"` support.
pub kind: String,
/// Inline source to evaluate (for `kind == "inline"`).
#[serde(default, skip_serializing_if = "Option::is_none")]
pub code: Option<String>,
/// Working directory for the job. The worker `chdir`s per job so relative
/// paths resolve against the caller's action sandbox.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cwd: Option<String>,
/// Soft per-job deadline in milliseconds. The worker aborts the job when it
/// elapses and replies with `timed_out = true`. Absent ⇒ run to completion.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub timeout_ms: Option<u64>,
}
/// A worker's reply to a [`PoolJobRequest`].
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PoolJobResponse {
pub id: Option<String>,
/// `true` when the harness ran the job to a normal conclusion (the user
/// code may still have thrown — see `exit_code`/`stderr`). `false` only for
/// harness-level failures described in `error`.
#[serde(default)]
pub ok: bool,
#[serde(default)]
pub stdout: String,
#[serde(default)]
pub stderr: String,
/// `0` on clean completion, non-zero when the job threw/exited non-zero,
/// `None` when not applicable.
#[serde(default)]
pub exit_code: Option<i32>,
/// Set when the worker aborted the job at its soft deadline.
#[serde(default)]
pub timed_out: bool,
#[serde(default)]
pub elapsed_ms: u64,
/// Harness-level error (worker could not run the job at all).
#[serde(default)]
pub error: Option<String>,
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn ready_line_parses() {
let ready: PoolReadyLine =
serde_json::from_str(r#"{"ready":true,"protocol":1,"lang":"node"}"#).unwrap();
assert!(ready.ready);
assert_eq!(ready.protocol, Some(PROTOCOL_VERSION));
assert_eq!(ready.lang.as_deref(), Some("node"));
}
#[test]
fn request_omits_absent_optional_fields() {
let req = PoolJobRequest {
id: "3".to_string(),
kind: "inline".to_string(),
code: Some("console.log(1)".to_string()),
cwd: None,
timeout_ms: None,
};
let line = serde_json::to_string(&req).unwrap();
assert!(line.contains("\"kind\":\"inline\""));
assert!(!line.contains("cwd"));
assert!(!line.contains("timeout_ms"));
}
#[test]
fn response_parses_failure_envelope() {
let resp: PoolJobResponse = serde_json::from_str(
r#"{"id":"7","ok":true,"stdout":"","stderr":"boom","exit_code":1,"elapsed_ms":12}"#,
)
.unwrap();
assert!(resp.ok);
assert_eq!(resp.exit_code, Some(1));
assert_eq!(resp.stderr, "boom");
assert!(!resp.timed_out);
}
}
+240
View File
@@ -0,0 +1,240 @@
//! Python pool backend: materialise the Python harness and submit inline jobs
//! to the shared [`LangPool`](super::pool::LangPool).
//!
//! Unlike the node backend, a job runs **in the worker's own interpreter** (no
//! per-job thread isolation — CPython can't safely kill a running thread), so a
//! Python job's soft deadline is enforced best-effort via `SIGALRM` on Unix and
//! otherwise falls back to the Rust-side hard deadline (which kills + respawns
//! the worker). `recycle_after_jobs` bounds cross-job module-state leakage.
use std::path::{Path, PathBuf};
use std::time::Duration;
use anyhow::Result;
use tokio::sync::OnceCell;
use super::pool;
use super::types::{PoolExecOutcome, PoolLang, PoolSettings};
use super::worker::WorkerLaunch;
use crate::openhuman::config::{RuntimePoolConfig, RuntimePoolLangConfig};
/// The bundled Python worker harness.
const WORKER_PY: &str = include_str!("pool_worker.py");
/// Written once per process (see [`super::ensure_worker_script`]).
static PYTHON_SCRIPT: OnceCell<PathBuf> = OnceCell::const_new();
/// Whether inline `python` jobs should route through the pool. `runtime_python.
/// enabled` is already implied by the tool only being constructed when the
/// python runtime is enabled, so only the pool switches are checked here.
pub fn enabled(pool: &RuntimePoolConfig) -> bool {
// Python defaults OFF: jobs share one interpreter (no worker_thread
// equivalent), so reuse can leak process-global state (`sys.modules`,
// `os.environ`, logging handlers, threads) across otherwise-unrelated runs.
// Opt in explicitly (`[runtime_pool.python] enabled = true`) to accept that
// in exchange for the warm-worker memory win.
pool.enabled && pool.python.is_enabled(false)
}
/// Run inline Python on a pooled, warm `python` worker.
///
/// `python_bin` / `bin_dir` come from the caller's already-resolved
/// [`ResolvedPython`](crate::openhuman::runtime_python::ResolvedPython).
/// `workspace_dir` and `lang_cfg` are injected at tool construction so this hot
/// path never re-reads config or re-writes the harness. `timeout` is the soft
/// per-job deadline (best-effort on Unix, hard-enforced by the Rust side).
pub async fn run_inline(
workspace_dir: &Path,
lang_cfg: &RuntimePoolLangConfig,
python_bin: &Path,
bin_dir: &Path,
code: String,
cwd: Option<PathBuf>,
timeout: Option<Duration>,
) -> Result<PoolExecOutcome, super::pool::PoolRunError> {
let script =
super::ensure_worker_script(&PYTHON_SCRIPT, workspace_dir, "pool_worker.py", WORKER_PY)
.await
.map_err(super::pool::PoolRunError::PreDispatch)?
.to_string_lossy()
.into_owned();
let mut env = super::base_env(bin_dir);
// Line-buffered stdio so protocol frames flush promptly.
env.push(("PYTHONUNBUFFERED".to_string(), "1".to_string()));
let launch = WorkerLaunch {
lang: PoolLang::Python,
bin: python_bin.to_path_buf(),
// `-u` unbuffered mirrors the runtime_python_server launch contract.
args: vec!["-u".to_string(), script],
env,
isolated_protocol: true,
};
let settings = PoolSettings::from_lang_config(lang_cfg);
let pool = pool::ensure_pool(launch, settings).await;
let cwd = cwd.map(|p| p.to_string_lossy().into_owned());
pool.run_inline(code, cwd, timeout).await
}
#[cfg(test)]
mod tests {
use super::*;
use crate::openhuman::config::Config;
use crate::openhuman::runtime_pool::{all_stats, PoolLang};
/// Resolve the host `python3` binary + its bin dir, or `None` to skip.
fn system_python() -> Option<(std::path::PathBuf, std::path::PathBuf)> {
for cmd in ["python3", "python"] {
if let Ok(out) = std::process::Command::new(cmd)
.args(["-c", "import sys; sys.stdout.write(sys.executable)"])
.output()
{
if out.status.success() {
let bin = std::path::PathBuf::from(String::from_utf8_lossy(&out.stdout).trim());
if let Some(dir) = bin.parent() {
return Some((bin.clone(), dir.to_path_buf()));
}
}
}
}
None
}
async fn python_spawns() -> u64 {
all_stats()
.await
.into_iter()
.find(|(lang, _)| *lang == PoolLang::Python)
.map(|(_, stats)| stats.worker_spawns)
.unwrap_or(0)
}
/// End-to-end: inline Python runs on the pool, reuses a warm worker, and
/// surfaces stdout / exit codes / cwd. Skips when no system python.
#[tokio::test]
async fn pooled_python_runs_inline_and_reuses_worker() {
let Some((python_bin, bin_dir)) = system_python() else {
eprintln!("[runtime_pool] test skipped: no system python on PATH");
return;
};
let tmp = std::env::temp_dir().join(format!("rt-pool-py-e2e-{}", std::process::id()));
std::fs::create_dir_all(&tmp).unwrap();
let mut config = Config::default();
config.workspace_dir = tmp.clone();
config.runtime_pool.python.max_workers = 1;
config.runtime_pool.python.recycle_after_jobs = 0;
let lang = config.runtime_pool.python.clone();
let spawns_before = python_spawns().await;
let out1 = run_inline(
&config.workspace_dir,
&lang,
&python_bin,
&bin_dir,
"print(6 * 7)".to_string(),
Some(tmp.clone()),
None,
)
.await
.expect("py job 1 runs");
assert!(out1.success(), "job 1 should succeed: {out1:?}");
assert_eq!(out1.stdout.trim(), "42");
// Raising surfaces a non-zero exit + traceback on stderr.
let out2 = run_inline(
&config.workspace_dir,
&lang,
&python_bin,
&bin_dir,
"raise ValueError('boom')".to_string(),
Some(tmp.clone()),
None,
)
.await
.expect("py job 2 runs");
assert!(!out2.success(), "raising job should fail");
assert!(out2.stderr.contains("boom"), "stderr was {:?}", out2.stderr);
// cwd-relative read resolves against the job cwd.
std::fs::write(tmp.join("probe.txt"), "PY_REL_OK").unwrap();
let out3 = run_inline(
&config.workspace_dir,
&lang,
&python_bin,
&bin_dir,
"print(open('./probe.txt').read(), end='')".to_string(),
Some(tmp.clone()),
None,
)
.await
.expect("py job 3 runs");
assert!(out3.success(), "cwd read should succeed: {out3:?}");
assert_eq!(out3.stdout, "PY_REL_OK");
// User stdin is an isolated EOF stream, not the long-lived worker's
// NDJSON request pipe.
let stdin_out = run_inline(
&config.workspace_dir,
&lang,
&python_bin,
&bin_dir,
"import os, sys\nprint(repr(sys.stdin.read()))\nprint(os.read(0, 1))".to_string(),
Some(tmp.clone()),
Some(Duration::from_secs(2)),
)
.await
.expect("stdin EOF job runs");
assert!(
stdin_out.success(),
"python stdin should be EOF: {stdin_out:?}"
);
assert_eq!(stdin_out.stdout, "''\nb''\n");
// A missing cwd must fail before executing user code. Continuing in the
// worker's inherited cwd would escape the requested action root.
let missing_cwd = tmp.join("deleted-action-root");
let err = run_inline(
&config.workspace_dir,
&lang,
&python_bin,
&bin_dir,
"open('must-not-exist.txt', 'w').write('bad')".to_string(),
Some(missing_cwd),
None,
)
.await
.expect_err("missing cwd must fail closed");
assert!(
err.to_string().contains("failed to set worker cwd"),
"unexpected missing-cwd error: {err}"
);
assert!(!tmp.join("must-not-exist.txt").exists());
// The harness-level error must not poison framing or worker reuse.
let out4 = run_inline(
&config.workspace_dir,
&lang,
&python_bin,
&bin_dir,
"print('AFTER_CWD_ERROR')".to_string(),
Some(tmp.clone()),
None,
)
.await
.expect("job after cwd error runs");
assert_eq!(out4.stdout, "AFTER_CWD_ERROR\n");
let spawns_after = python_spawns().await;
assert!(
spawns_after - spawns_before <= 1,
"expected warm-worker reuse: {} new spawns",
spawns_after - spawns_before
);
let _ = std::fs::remove_dir_all(&tmp);
}
}
+117
View File
@@ -0,0 +1,117 @@
//! Public types callers of the runtime pool see.
use std::time::Duration;
use crate::openhuman::config::RuntimePoolLangConfig;
/// Language a pool serves. Drives which harness script + interpreter launches.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PoolLang {
Node,
Python,
}
impl PoolLang {
pub fn id(self) -> &'static str {
match self {
PoolLang::Node => "node",
PoolLang::Python => "python",
}
}
}
/// The result of running one job on a pooled worker. Mirrors the fields a
/// per-call `std::process` spawn would have exposed, plus `queue_wait` so
/// callers can surface backpressure in run logs (a DoD requirement of #5106).
#[derive(Debug, Clone)]
pub struct PoolExecOutcome {
pub stdout: String,
pub stderr: String,
/// `0` on success, non-zero when the job threw / exited non-zero.
pub exit_code: Option<i32>,
/// The job hit its soft deadline and was aborted.
pub timed_out: bool,
/// Wall-clock the job itself took inside the worker.
pub elapsed: Duration,
/// How long the submission waited for a free worker (queue backpressure).
pub queue_wait: Duration,
}
impl PoolExecOutcome {
/// A job "succeeded" when it ran to completion with a zero (or absent) exit
/// code and did not time out.
pub fn success(&self) -> bool {
!self.timed_out && matches!(self.exit_code, None | Some(0))
}
}
/// The knobs a single language pool reads from config, snapshotted at pool
/// construction. Kept as a plain owned struct so the pool never re-reads config
/// mid-flight.
#[derive(Debug, Clone)]
pub struct PoolSettings {
pub max_workers: usize,
pub idle_ttl: Option<Duration>,
pub recycle_after_jobs: u64,
pub max_queue_depth: usize,
}
impl PoolSettings {
/// Derive the effective settings from a per-language config block, applying
/// the same "never zero" clamps the config getters use.
pub fn from_lang_config(cfg: &RuntimePoolLangConfig) -> Self {
Self {
max_workers: cfg.effective_max_workers(),
idle_ttl: if cfg.idle_ttl_secs == 0 {
None
} else {
Some(Duration::from_secs(cfg.idle_ttl_secs))
},
recycle_after_jobs: cfg.recycle_after_jobs,
max_queue_depth: cfg.effective_max_queue_depth(),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn outcome_success_semantics() {
let base = PoolExecOutcome {
stdout: String::new(),
stderr: String::new(),
exit_code: Some(0),
timed_out: false,
elapsed: Duration::ZERO,
queue_wait: Duration::ZERO,
};
assert!(base.success());
assert!(!PoolExecOutcome {
exit_code: Some(1),
..base.clone()
}
.success());
assert!(!PoolExecOutcome {
timed_out: true,
..base.clone()
}
.success());
}
#[test]
fn settings_disable_idle_reap_on_zero() {
let cfg = RuntimePoolLangConfig {
enabled: Some(true),
max_workers: 3,
idle_ttl_secs: 0,
recycle_after_jobs: 5,
max_queue_depth: 10,
};
let s = PoolSettings::from_lang_config(&cfg);
assert_eq!(s.max_workers, 3);
assert!(s.idle_ttl.is_none());
assert_eq!(s.recycle_after_jobs, 5);
}
}
+378
View File
@@ -0,0 +1,378 @@
//! A single pooled worker: one long-lived interpreter child speaking the
//! newline-delimited JSON [`protocol`](super::protocol) over an isolated
//! loopback socket (with a stdio fallback for development harnesses).
//!
//! One worker runs **one job at a time**; concurrency comes from the
//! [`LangPool`](super::pool::LangPool) holding several workers. A worker stays
//! warm between jobs (the whole point — no per-run interpreter spawn) until it
//! is idle-reaped or recycled after N jobs.
use std::path::PathBuf;
use std::process::Stdio;
use std::time::{Duration, Instant};
use anyhow::{bail, Context, Result};
use tokio::io::{AsyncBufReadExt, AsyncRead, AsyncWrite, AsyncWriteExt, BufReader, Lines};
use tokio::net::TcpListener;
use tokio::process::{Child, ChildStderr, ChildStdout, Command};
use super::protocol::{PoolJobRequest, PoolJobResponse, PoolReadyLine, PROTOCOL_VERSION};
use super::types::PoolLang;
/// How long to wait for a freshly-spawned worker to print its ready line.
const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(30);
/// Everything needed to (re)spawn a worker for one language. Cheap to clone so
/// the pool can respawn on demand.
#[derive(Debug, Clone)]
pub struct WorkerLaunch {
pub lang: PoolLang,
/// Interpreter binary (`node` / `python`).
pub bin: PathBuf,
/// Args after the binary — typically `[harness_script_path]`.
pub args: Vec<String>,
/// Full environment for the child (already allow-listed by the backend).
/// The child's env is cleared first, so this is the complete set.
pub env: Vec<(String, String)>,
/// Keep user fd 0/1/2 away from the NDJSON request/response stream by
/// serving the protocol over a per-launch authenticated loopback socket.
pub isolated_protocol: bool,
}
/// Failure from [`PoolWorker::submit`], tagged with whether the job was already
/// dispatched to the worker. A retry / legacy fallback is only safe when the job
/// was **not** dispatched (it never ran); a post-dispatch failure is terminal so
/// the same job is never executed twice.
#[derive(Debug)]
pub struct SubmitError {
pub err: anyhow::Error,
pub dispatched: bool,
}
impl SubmitError {
fn pre(err: anyhow::Error) -> Self {
Self {
err,
dispatched: false,
}
}
fn post(err: anyhow::Error) -> Self {
Self {
err,
dispatched: true,
}
}
}
impl std::fmt::Display for SubmitError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{:#}", self.err)
}
}
/// A warm interpreter child plus its bookkeeping.
pub struct PoolWorker {
launch: WorkerLaunch,
_child: Child,
stdin: Box<dyn AsyncWrite + Send + Unpin>,
responses: Lines<BufReader<Box<dyn AsyncRead + Send + Unpin>>>,
jobs_done: u64,
last_used: Instant,
}
impl PoolWorker {
pub fn jobs_done(&self) -> u64 {
self.jobs_done
}
pub fn last_used(&self) -> Instant {
self.last_used
}
/// Spawn a new worker and complete the readiness handshake.
pub async fn spawn(launch: &WorkerLaunch) -> Result<Self> {
tracing::info!(
lang = launch.lang.id(),
bin = %launch.bin.display(),
"[runtime_pool] spawning worker"
);
let mut cmd = Command::new(&launch.bin);
cmd.args(&launch.args);
cmd.env_clear();
for (key, value) in &launch.env {
cmd.env(key, value);
}
let isolated_protocol = if launch.isolated_protocol {
let listener = TcpListener::bind(("127.0.0.1", 0))
.await
.context("binding isolated worker protocol listener")?;
let addr = listener
.local_addr()
.context("reading isolated worker protocol address")?;
let token = uuid::Uuid::new_v4().to_string();
cmd.env("OPENHUMAN_RUNTIME_POOL_PROTOCOL_ADDR", addr.to_string());
cmd.env("OPENHUMAN_RUNTIME_POOL_PROTOCOL_TOKEN", &token);
Some((listener, token))
} else {
None
};
if isolated_protocol.is_some() {
// Jobs inherit EOF on fd 0, matching Command::output(), while the
// harness receives requests over the isolated duplex socket.
cmd.stdin(Stdio::null());
} else {
cmd.stdin(Stdio::piped());
}
cmd.stdout(Stdio::piped())
.stderr(Stdio::piped())
.kill_on_drop(true);
// Suppress the Windows console flash for each spawned worker.
#[cfg(windows)]
{
use std::os::windows::process::CommandExt;
const CREATE_NO_WINDOW: u32 = 0x0800_0000;
cmd.creation_flags(CREATE_NO_WINDOW);
}
let mut child = cmd
.spawn()
.with_context(|| format!("spawning {} worker", launch.lang.id()))?;
let child_stdin = child.stdin.take();
let stdout = child.stdout.take().context("worker stdout missing")?;
if let Some(stderr) = child.stderr.take() {
drain_stderr(launch.lang, stderr);
}
let (stdin, reader, expected_token): (
Box<dyn AsyncWrite + Send + Unpin>,
Box<dyn AsyncRead + Send + Unpin>,
Option<String>,
) = if let Some((listener, token)) = isolated_protocol {
// stdout is now exclusively user fd-level output. Drain it so
// chatty jobs cannot block; protocol frames use the socket.
drain_stdout(launch.lang, stdout);
let (stream, _) = tokio::time::timeout(HANDSHAKE_TIMEOUT, listener.accept())
.await
.map_err(|_| {
anyhow::anyhow!("{} worker protocol connection timed out", launch.lang.id())
})?
.context("accepting isolated worker protocol connection")?;
let (reader, writer) = tokio::io::split(stream);
(Box::new(writer), Box::new(reader), Some(token))
} else {
(
Box::new(child_stdin.context("worker stdin missing")?),
Box::new(stdout),
None,
)
};
let mut lines = BufReader::new(reader).lines();
let ready_line = match tokio::time::timeout(HANDSHAKE_TIMEOUT, lines.next_line()).await {
Ok(Ok(Some(line))) => line,
Ok(Ok(None)) => bail!(
"{} worker exited before readiness handshake",
launch.lang.id()
),
Ok(Err(error)) => {
return Err(error).context("reading worker handshake");
}
Err(_) => bail!("{} worker readiness handshake timed out", launch.lang.id()),
};
let ready: PoolReadyLine = serde_json::from_str(&ready_line)
.with_context(|| format!("parsing worker ready line: {ready_line}"))?;
if !ready.ready {
bail!(
"{} worker failed to start: {}",
launch.lang.id(),
ready.error.unwrap_or_else(|| "unknown".to_string())
);
}
if ready.protocol != Some(PROTOCOL_VERSION) {
bail!(
"{} worker protocol mismatch: expected {}, got {:?}",
launch.lang.id(),
PROTOCOL_VERSION,
ready.protocol
);
}
if ready.protocol_token != expected_token {
bail!("{} worker protocol authentication failed", launch.lang.id());
}
tracing::info!(lang = launch.lang.id(), "[runtime_pool] worker ready");
Ok(Self {
launch: launch.clone(),
_child: child,
stdin,
responses: lines,
jobs_done: 0,
last_used: Instant::now(),
})
}
/// Submit one job and await its response.
///
/// `hard_timeout` is a **safety net** above the worker's own soft deadline:
/// the worker aborts a job at `req.timeout_ms` and still replies, so this
/// only fires if the worker itself has wedged. On `Err` the caller must
/// discard this worker — its stdio framing can no longer be trusted.
pub async fn submit(
&mut self,
req: &PoolJobRequest,
hard_timeout: Option<Duration>,
) -> std::result::Result<PoolJobResponse, SubmitError> {
let mut line = serde_json::to_string(req)
.map_err(|e| SubmitError::pre(anyhow::Error::new(e).context("serialising pool job")))?;
line.push('\n');
// A write failure means the bytes never reached the worker (e.g. a
// reused idle worker died) → the job did not run → safe to retry.
self.stdin.write_all(line.as_bytes()).await.map_err(|e| {
SubmitError::pre(anyhow::Error::new(e).context("writing pool job request"))
})?;
// Past this point the request bytes are in the pipe: the job may execute,
// so any later failure is terminal (never re-run the same job).
self.stdin.flush().await.map_err(|e| {
SubmitError::post(anyhow::Error::new(e).context("flushing pool job request"))
})?;
// Fixed deadline: `continue`ing over unparseable / mismatched-id lines
// must NOT reset the wedged-worker timeout, so it bounds the total wait.
let deadline = hard_timeout.map(|t| tokio::time::Instant::now() + t);
loop {
let next = match deadline {
Some(dl) => match tokio::time::timeout_at(dl, self.responses.next_line()).await {
Ok(inner) => inner,
Err(_) => {
return Err(SubmitError::post(anyhow::anyhow!(
"pool worker job timed out (hard deadline; worker wedged)"
)))
}
},
None => self.responses.next_line().await,
};
let line = match next {
Ok(Some(line)) => line,
Ok(None) => {
return Err(SubmitError::post(anyhow::anyhow!(
"pool worker closed stdout"
)))
}
Err(error) => {
return Err(SubmitError::post(
anyhow::Error::new(error).context("reading pool job response"),
))
}
};
let response: PoolJobResponse = match serde_json::from_str(&line) {
Ok(response) => response,
Err(error) => {
tracing::warn!(
lang = self.launch.lang.id(),
"[runtime_pool] unparseable worker line skipped: {error}"
);
continue;
}
};
if response.id.as_deref() != Some(req.id.as_str()) {
tracing::debug!(
lang = self.launch.lang.id(),
"[runtime_pool] skipped response for different id={:?}",
response.id
);
continue;
}
self.jobs_done += 1;
self.last_used = Instant::now();
return Ok(response);
}
}
/// Whether this worker has served enough jobs to be recycled. `0` disables.
pub fn should_recycle(&self, recycle_after: u64) -> bool {
recycle_due(self.jobs_done, recycle_after)
}
/// Whether this worker has been idle at least `ttl`.
pub fn idle_expired(&self, ttl: Duration) -> bool {
idle_due(self.last_used.elapsed(), ttl)
}
/// Signal the child to exit. Best-effort; `kill_on_drop` is the backstop.
pub fn shutdown(mut self) {
if let Err(error) = self._child.start_kill() {
tracing::debug!(
lang = self.launch.lang.id(),
"[runtime_pool] failed to signal worker shutdown: {error}"
);
}
}
}
/// Continuously drain a worker's stderr so a chatty child never blocks on a
/// full pipe. Lines are logged at trace; never parsed as protocol.
fn drain_stderr(lang: PoolLang, stderr: ChildStderr) {
tokio::spawn(async move {
let mut reader = BufReader::new(stderr).lines();
while let Ok(Some(line)) = reader.next_line().await {
tracing::trace!(lang = lang.id(), "[runtime_pool] worker stderr: {line}");
}
});
}
/// Drain fd-level stdout from workers whose protocol uses an isolated socket.
/// This output is deliberately never parsed as NDJSON, so user code cannot
/// forge a response frame or desynchronise subsequent jobs.
fn drain_stdout(lang: PoolLang, stdout: ChildStdout) {
tokio::spawn(async move {
let mut reader = BufReader::new(stdout).lines();
while let Ok(Some(line)) = reader.next_line().await {
tracing::trace!(lang = lang.id(), "[runtime_pool] worker fd stdout: {line}");
}
});
}
/// Pure recycle predicate: a worker is due for recycling once it has served
/// `recycle_after` jobs (`0` disables recycling).
fn recycle_due(jobs_done: u64, recycle_after: u64) -> bool {
recycle_after > 0 && jobs_done >= recycle_after
}
/// Pure idle-expiry predicate: idle for at least `ttl`.
fn idle_due(idle_elapsed: Duration, ttl: Duration) -> bool {
idle_elapsed >= ttl
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn recycle_due_respects_budget_and_disable() {
assert!(!recycle_due(0, 0), "recycle_after=0 disables recycling");
assert!(!recycle_due(100, 0), "recycle_after=0 never recycles");
assert!(!recycle_due(4, 5), "below budget");
assert!(recycle_due(5, 5), "at budget");
assert!(recycle_due(6, 5), "past budget");
}
#[test]
fn idle_due_is_inclusive_at_ttl() {
assert!(!idle_due(Duration::from_secs(4), Duration::from_secs(5)));
assert!(idle_due(Duration::from_secs(5), Duration::from_secs(5)));
assert!(idle_due(Duration::from_secs(6), Duration::from_secs(5)));
}
#[test]
fn submit_error_tags_dispatch_state() {
let pre = SubmitError::pre(anyhow::anyhow!("write failed"));
assert!(
!pre.dispatched,
"write failures are pre-dispatch (retryable)"
);
let post = SubmitError::post(anyhow::anyhow!("read timed out"));
assert!(
post.dispatched,
"read failures are post-dispatch (terminal)"
);
}
}
+2
View File
@@ -9,6 +9,7 @@ mod node_exec;
mod npm_exec;
mod proxy_config;
mod pushover;
mod python_exec;
mod resolve_time;
mod retrieve_tool_output;
mod schedule;
@@ -37,6 +38,7 @@ pub use node_exec::NodeExecTool;
pub use npm_exec::NpmExecTool;
pub use proxy_config::ProxyConfigTool;
pub use pushover::PushoverTool;
pub use python_exec::PythonExecTool;
pub use resolve_time::ResolveTimeTool;
pub use retrieve_tool_output::RetrieveToolOutputTool;
pub use schedule::ScheduleTool;
@@ -73,6 +73,10 @@ pub struct NodeExecTool {
security: Arc<SecurityPolicy>,
runtime: Arc<dyn RuntimeAdapter>,
bootstrap: Arc<NodeBootstrap>,
/// Runtime-pool config + workspace, snapshotted at construction so the hot
/// inline path never re-reads config from disk (#5106 is a perf feature).
pool_cfg: crate::openhuman::config::RuntimePoolConfig,
workspace_dir: std::path::PathBuf,
}
impl NodeExecTool {
@@ -80,11 +84,15 @@ impl NodeExecTool {
security: Arc<SecurityPolicy>,
runtime: Arc<dyn RuntimeAdapter>,
bootstrap: Arc<NodeBootstrap>,
pool_cfg: crate::openhuman::config::RuntimePoolConfig,
workspace_dir: std::path::PathBuf,
) -> Self {
Self {
security,
runtime,
bootstrap,
pool_cfg,
workspace_dir,
}
}
}
@@ -289,6 +297,20 @@ impl NodeExecTool {
.await);
}
// Route inline JS through the shared runtime pool when enabled (#5106):
// a warm, bounded set of `node` workers replaces one `node -e` child per
// call, so a fleet pays ~one interpreter instead of one per skill run.
// `script_path` and sandboxed runs keep the legacy per-call spawn; a
// pool infrastructure failure also transparently falls back below.
if let Some(code) = inline_code.as_deref() {
if let Some(result) = self
.try_pool_inline(code, &resolved, &path_policy.action_dir, explicit_timeout)
.await
{
return Ok(result);
}
}
let mut cmd = match self
.runtime
.build_shell_command(&command, &path_policy.action_dir)
@@ -370,6 +392,81 @@ impl NodeExecTool {
}
}
impl NodeExecTool {
/// Attempt to run inline JS on the shared runtime pool (#5106).
///
/// Returns `Some(result)` when the pool handled the job — success, non-zero
/// exit, or timeout, all mapped to the same `ToolResult` shape as the legacy
/// path. Returns `None` when pooling is disabled or the pool infrastructure
/// failed, so the caller transparently falls back to a per-call spawn.
async fn try_pool_inline(
&self,
code: &str,
resolved: &crate::openhuman::runtime_node::ResolvedNode,
action_dir: &std::path::Path,
timeout: Option<Duration>,
) -> Option<ToolResult> {
if !crate::openhuman::runtime_pool::node::enabled(&self.pool_cfg) {
return None;
}
// Node forbids process.chdir() inside worker_threads. Preserve the
// legacy `node -e` contract for any statically apparent chdir use
// instead of dispatching code that the pooled worker cannot execute.
// False positives are safe: they only give up the pooling optimisation.
if inline_requires_process_chdir_compat(code) {
tracing::debug!("[node_exec] pool: process.chdir-compatible code uses legacy spawn");
return None;
}
match crate::openhuman::runtime_pool::node::run_inline(
&self.workspace_dir,
&self.pool_cfg.node,
&resolved.node_bin,
&resolved.bin_dir,
code.to_string(),
Some(action_dir.to_path_buf()),
timeout,
)
.await
{
Ok(outcome) => {
tracing::info!(
queue_wait_ms = outcome.queue_wait.as_millis() as u64,
elapsed_ms = outcome.elapsed.as_millis() as u64,
timed_out = outcome.timed_out,
"[node_exec] pool: inline job completed on a warm worker"
);
Some(pool_outcome_to_result(outcome, timeout))
}
// Job never ran → safe to fall back to a per-call spawn.
Err(crate::openhuman::runtime_pool::PoolRunError::PreDispatch(error)) => {
tracing::warn!(
error = %error,
"[node_exec] pool: pre-dispatch failure; falling back to legacy spawn"
);
None
}
// Load-shed: do NOT spawn (that reintroduces the per-run RSS the pool
// caps). Surface a retryable busy error instead.
Err(crate::openhuman::runtime_pool::PoolRunError::Saturated) => {
tracing::warn!("[node_exec] pool: saturated; shedding load");
Some(ToolResult::error(
"Node runtime pool is at capacity; retry shortly.",
))
}
// The job may already have executed → terminal, never re-run it.
Err(crate::openhuman::runtime_pool::PoolRunError::PostDispatch(error)) => {
tracing::warn!(
error = %error,
"[node_exec] pool: post-dispatch failure; not retried to avoid duplicate execution"
);
Some(ToolResult::error(format!(
"node_exec failed after the code was dispatched to a pooled worker; not retried to avoid running it twice: {error}"
)))
}
}
}
}
impl NodeExecTool {
/// Execute a node command through the sandbox backend. Called from
/// `execute()` when the agent's `SandboxMode` is `Sandboxed`.
@@ -481,6 +578,50 @@ impl NodeExecTool {
}
}
/// Map a runtime-pool outcome onto the same `ToolResult` shape the legacy
/// `node -e` path produces: 1 MB stdout/stderr caps, exit-code surfacing on
/// failure, and the identical timeout message. Keeps pooled and legacy runs
/// indistinguishable to the agent.
fn pool_outcome_to_result(
outcome: crate::openhuman::runtime_pool::PoolExecOutcome,
timeout: Option<Duration>,
) -> ToolResult {
if outcome.timed_out {
return ToolResult::error(format!(
"node_exec timed out after {}s and was killed",
timeout.map(|d| d.as_secs()).unwrap_or(0)
));
}
let mut stdout = outcome.stdout;
let mut stderr = outcome.stderr;
if stdout.len() > MAX_OUTPUT_BYTES {
stdout.truncate(crate::openhuman::util::floor_char_boundary(
&stdout,
MAX_OUTPUT_BYTES,
));
stdout.push_str("\n... [stdout truncated at 1MB]");
}
if stderr.len() > MAX_OUTPUT_BYTES {
stderr.truncate(crate::openhuman::util::floor_char_boundary(
&stderr,
MAX_OUTPUT_BYTES,
));
stderr.push_str("\n... [stderr truncated at 1MB]");
}
let success = matches!(outcome.exit_code, None | Some(0));
if success {
if stderr.is_empty() {
ToolResult::success(stdout)
} else {
ToolResult::success(format!("{stdout}\n[stderr]\n{stderr}"))
}
} else {
super::command_output::command_failure(outcome.exit_code, &stdout, &stderr)
}
}
/// Resolve the wall-clock policy for a `node_exec` call from its args.
///
/// No `timeout_secs` (or `0`) ⇒ run unbounded; a positive value ⇒ enforce it,
@@ -493,6 +634,17 @@ fn node_timeout_policy(args: &serde_json::Value) -> ToolTimeout {
}
}
/// Whether inline JavaScript needs the legacy main-thread process so
/// `process.chdir()` remains available.
///
/// Matching the property name anywhere deliberately catches direct calls,
/// aliases, destructuring, and bracket notation. Comments or string literals
/// may route an otherwise pool-safe snippet through the legacy path, which is
/// preferable to executing a cwd-mutating snippet with changed semantics.
fn inline_requires_process_chdir_compat(code: &str) -> bool {
code.contains("chdir")
}
/// POSIX-safe single-quote escaping. Wraps `s` in `'…'`, turning any embedded
/// single-quote into the four-char sequence `'\''`. Node bin paths and user
/// code pass through untouched semantically, but no shell metacharacter can
@@ -573,6 +725,24 @@ mod tests {
);
}
#[test]
fn process_chdir_snippets_use_legacy_node_spawn() {
for code in [
"process.chdir('subdir'); console.log(process.cwd())",
"const move = process.chdir; move('subdir')",
"const { chdir } = process; chdir('subdir')",
"process['chdir']('subdir')",
] {
assert!(
inline_requires_process_chdir_compat(code),
"expected legacy fallback for {code:?}"
);
}
assert!(!inline_requires_process_chdir_compat(
"console.log(process.cwd())"
));
}
#[test]
fn shell_quote_escapes_single_quotes() {
assert_eq!(shell_quote("it's"), "'it'\\''s'");
@@ -0,0 +1,620 @@
//! `python_exec` — execute Python 3 via the managed (or system) interpreter.
//!
//! Sibling to [`node_exec`](super::node_exec) with identical security gates and
//! env hygiene; the command is pinned to the `python` binary resolved by
//! [`crate::openhuman::runtime_python::PythonBootstrap`].
//!
//! Two input modes:
//!
//! | Mode | Params | Resulting invocation |
//! |---------------|-------------------------------------------|--------------------------------|
//! | Inline code | `inline_code: "print(1+1)"` | `python -c '<code>'` |
//! | Script path | `script_path: "scripts/run.py"`, `args` | `python <path> <args...>` |
//!
//! Inline code routes through the shared runtime pool (#5106) when enabled — a
//! warm, bounded set of `python` workers replaces one interpreter child per
//! call — and transparently falls back to a per-call spawn otherwise. Script
//! paths and sandboxed runs always use the per-call spawn.
use crate::openhuman::agent::host_runtime::RuntimeAdapter;
use crate::openhuman::runtime_python::{PythonBootstrap, ResolvedPython};
use crate::openhuman::security::{CommandClass, GateDecision, SecurityPolicy};
use crate::openhuman::tools::traits::{
PermissionLevel, Tool, ToolCallOptions, ToolResult, ToolTimeout,
};
use async_trait::async_trait;
use serde_json::json;
use std::sync::Arc;
use std::time::Duration;
use tinyagents::harness::tool::ToolExecutionContext;
/// Absolute ceiling a caller may request via `timeout_secs`. No default timeout —
/// Python scripts legitimately take minutes; a deadline applies only when
/// `timeout_secs` is supplied. Mirrors `node_exec` (issue #4023).
const PYTHON_TIMEOUT_MAX_SECS: u64 = 1800;
/// Maximum combined stdout/stderr size (1 MB each) — same cap as shell/node.
const MAX_OUTPUT_BYTES: usize = 1_048_576;
/// Env allow-list for child processes. Matches node_exec/shell — secrets never
/// leak into spawned python processes. `PATH` gets a prepend of the managed bin
/// dir before being forwarded.
const SAFE_ENV_VARS: &[&str] = &[
"HOME",
"TERM",
"LANG",
"LC_ALL",
"LC_CTYPE",
"USER",
"SHELL",
"TMPDIR",
"SystemRoot",
"WINDIR",
"COMSPEC",
"PATHEXT",
"TEMP",
"TMP",
"USERPROFILE",
"APPDATA",
"LOCALAPPDATA",
"ProgramFiles",
"ProgramFiles(x86)",
"ProgramW6432",
];
/// `python_exec` — execute Python through the resolved Python 3 runtime.
pub struct PythonExecTool {
security: Arc<SecurityPolicy>,
runtime: Arc<dyn RuntimeAdapter>,
bootstrap: Arc<PythonBootstrap>,
/// Runtime-pool config + workspace, snapshotted at construction so the hot
/// inline path never re-reads config from disk (#5106 is a perf feature).
pool_cfg: crate::openhuman::config::RuntimePoolConfig,
workspace_dir: std::path::PathBuf,
}
impl PythonExecTool {
pub fn new(
security: Arc<SecurityPolicy>,
runtime: Arc<dyn RuntimeAdapter>,
bootstrap: Arc<PythonBootstrap>,
pool_cfg: crate::openhuman::config::RuntimePoolConfig,
workspace_dir: std::path::PathBuf,
) -> Self {
Self {
security,
runtime,
bootstrap,
pool_cfg,
workspace_dir,
}
}
}
#[async_trait]
impl Tool for PythonExecTool {
fn name(&self) -> &str {
"python_exec"
}
fn description(&self) -> &str {
"Execute Python 3 through the managed interpreter. Pass either `inline_code` (runs via `python -c`) or `script_path` (runs a .py file in your working directory, the action sandbox). Optional `args` forwards positional arguments to the script. Only the program's stdout/stderr is captured and returned to you — a value you do not `print` is invisible. Always print the output you need (e.g. `print(json.dumps(result))`)."
}
fn parameters_schema(&self) -> serde_json::Value {
json!({
"type": "object",
"properties": {
"inline_code": {
"type": "string",
"description": "Python source passed to `python -c`. Mutually exclusive with script_path."
},
"script_path": {
"type": "string",
"description": "Path (relative to workspace) to a .py file. Mutually exclusive with inline_code."
},
"args": {
"type": "array",
"items": { "type": "string" },
"description": "Positional arguments appended after the script. Ignored for inline_code."
},
"timeout_secs": {
"type": "integer",
"description": "Optional wall-clock timeout (seconds) before the process is killed. No timeout by default. Capped at 1800s; 0 disables."
}
}
})
}
fn permission_level(&self) -> PermissionLevel {
PermissionLevel::Execute
}
fn timeout_policy(&self, args: &serde_json::Value) -> ToolTimeout {
python_timeout_policy(args)
}
/// Running Python is arbitrary code execution → the `Write` bucket, same as
/// `node_exec`. In ask-before-edit this routes through the human approval
/// gate; in Full it runs; in read-only it refuses below.
fn external_effect_with_args(&self, _args: &serde_json::Value) -> bool {
self.security.gate_decision(CommandClass::Write) == GateDecision::Prompt
}
async fn execute(&self, args: serde_json::Value) -> anyhow::Result<ToolResult> {
self.execute_in_context(args, None).await
}
async fn execute_with_context(
&self,
args: serde_json::Value,
_options: ToolCallOptions,
context: Option<&ToolExecutionContext>,
) -> anyhow::Result<ToolResult> {
self.execute_in_context(args, context).await
}
}
impl PythonExecTool {
async fn execute_in_context(
&self,
args: serde_json::Value,
context: Option<&ToolExecutionContext>,
) -> anyhow::Result<ToolResult> {
let inline_code = args
.get("inline_code")
.and_then(|v| v.as_str())
.map(str::to_string);
let script_path = args
.get("script_path")
.and_then(|v| v.as_str())
.map(str::to_string);
let extra_args: Vec<String> = args
.get("args")
.and_then(|v| v.as_array())
.map(|arr| {
arr.iter()
.filter_map(|v| v.as_str().map(str::to_string))
.collect()
})
.unwrap_or_default();
let explicit_timeout = crate::openhuman::tool_timeout::explicit_call_timeout_duration(
args.get("timeout_secs").and_then(|v| v.as_u64()),
PYTHON_TIMEOUT_MAX_SECS,
);
if inline_code.is_some() == script_path.is_some() {
return Ok(ToolResult::error(
"python_exec requires exactly one of `inline_code` or `script_path`",
));
}
if !self.security.can_act() {
return Ok(ToolResult::error(
"[policy-blocked] Action blocked: the agent is in read-only mode and cannot execute code.",
));
}
if self.security.is_rate_limited() {
return Ok(ToolResult::error(
"Rate limit exceeded: too many actions in the last hour",
));
}
if !self.security.record_action() {
return Ok(ToolResult::error(
"Rate limit exceeded: action budget exhausted",
));
}
let resolved = match self.bootstrap.resolve().await {
Ok(r) => r,
Err(e) => {
tracing::error!(error = %e, "[python_exec] failed to resolve python runtime");
return Ok(ToolResult::error(format!(
"Python runtime unavailable: {e}"
)));
}
};
tracing::info!(
version = %resolved.version,
python_bin = %resolved.python_bin.display(),
"[python_exec] starting invocation"
);
let path_policy = super::security_for_tool_context(&self.security, context, "python_exec");
let command = if let Some(code) = inline_code.as_deref() {
format!(
"{} -c {}",
shell_quote(&resolved.python_bin.to_string_lossy()),
shell_quote(code)
)
} else if let Some(path) = script_path.as_deref() {
let resolved_script = match resolve_script_path(&path_policy.action_dir, path) {
Ok(p) => p,
Err(msg) => return Ok(ToolResult::error(msg)),
};
let mut parts: Vec<String> = Vec::with_capacity(extra_args.len() + 2);
parts.push(shell_quote(&resolved.python_bin.to_string_lossy()));
parts.push(shell_quote(&resolved_script.to_string_lossy()));
for a in &extra_args {
parts.push(shell_quote(a));
}
parts.join(" ")
} else {
unreachable!("guarded above")
};
// Sandboxed agents route through the sandbox backend, mirroring node_exec.
if matches!(
crate::openhuman::agent::harness::current_sandbox_mode(),
Some(crate::openhuman::agent::harness::definition::SandboxMode::Sandboxed)
) {
let bin_dir = resolved
.python_bin
.parent()
.map(std::path::Path::to_path_buf)
.unwrap_or_default();
return Ok(self
.run_sandboxed(&path_policy, &command, &bin_dir, explicit_timeout)
.await);
}
// Route inline Python through the shared runtime pool when enabled
// (#5106). script_path and sandboxed runs keep the per-call spawn; a
// pool infrastructure failure also transparently falls back below.
if let Some(code) = inline_code.as_deref() {
if let Some(result) = self
.try_pool_inline(code, &resolved, &path_policy.action_dir, explicit_timeout)
.await
{
return Ok(result);
}
}
let mut cmd = match self
.runtime
.build_shell_command(&command, &path_policy.action_dir)
{
Ok(cmd) => cmd,
Err(e) => {
return Ok(ToolResult::error(format!(
"Failed to build runtime command: {e}"
)));
}
};
cmd.env_clear();
let host_path = std::env::var("PATH").unwrap_or_default();
let sep = if cfg!(windows) { ";" } else { ":" };
let prepended_path = if host_path.is_empty() {
resolved.bin_dir.to_string_lossy().into_owned()
} else {
format!("{}{}{}", resolved.bin_dir.display(), sep, host_path)
};
cmd.env("PATH", &prepended_path);
// Unbuffered stdio so partial output survives a kill on timeout.
cmd.env("PYTHONUNBUFFERED", "1");
for var in SAFE_ENV_VARS {
if let Ok(val) = std::env::var(var) {
cmd.env(var, val);
}
}
let result = match explicit_timeout {
Some(timeout) => tokio::time::timeout(timeout, cmd.output()).await,
None => Ok(cmd.output().await),
};
match result {
Ok(Ok(output)) => {
let mut stdout = String::from_utf8_lossy(&output.stdout).to_string();
let mut stderr = String::from_utf8_lossy(&output.stderr).to_string();
if stdout.len() > MAX_OUTPUT_BYTES {
stdout.truncate(crate::openhuman::util::floor_char_boundary(
&stdout,
MAX_OUTPUT_BYTES,
));
stdout.push_str("\n... [stdout truncated at 1MB]");
}
if stderr.len() > MAX_OUTPUT_BYTES {
stderr.truncate(crate::openhuman::util::floor_char_boundary(
&stderr,
MAX_OUTPUT_BYTES,
));
stderr.push_str("\n... [stderr truncated at 1MB]");
}
if output.status.success() {
if stderr.is_empty() {
Ok(ToolResult::success(stdout))
} else {
Ok(ToolResult::success(format!("{stdout}\n[stderr]\n{stderr}")))
}
} else {
Ok(super::command_output::command_failure(
output.status.code(),
&stdout,
&stderr,
))
}
}
Ok(Err(e)) => Ok(ToolResult::error(format!("Failed to execute python: {e}"))),
Err(_) => Ok(ToolResult::error(format!(
"python_exec timed out after {}s and was killed",
explicit_timeout.map(|d| d.as_secs()).unwrap_or(0)
))),
}
}
/// Attempt to run inline Python on the shared runtime pool (#5106). Returns
/// `Some(result)` when the pool handled the job; `None` when pooling is
/// disabled or the pool infrastructure failed (caller falls back to spawn).
async fn try_pool_inline(
&self,
code: &str,
resolved: &ResolvedPython,
action_dir: &std::path::Path,
timeout: Option<Duration>,
) -> Option<ToolResult> {
if !crate::openhuman::runtime_pool::python::enabled(&self.pool_cfg) {
return None;
}
match crate::openhuman::runtime_pool::python::run_inline(
&self.workspace_dir,
&self.pool_cfg.python,
&resolved.python_bin,
&resolved.bin_dir,
code.to_string(),
Some(action_dir.to_path_buf()),
timeout,
)
.await
{
Ok(outcome) => {
tracing::info!(
queue_wait_ms = outcome.queue_wait.as_millis() as u64,
elapsed_ms = outcome.elapsed.as_millis() as u64,
timed_out = outcome.timed_out,
"[python_exec] pool: inline job completed on a warm worker"
);
Some(pool_outcome_to_result(outcome, timeout))
}
// Job never ran → safe to fall back to a per-call spawn.
Err(crate::openhuman::runtime_pool::PoolRunError::PreDispatch(error)) => {
tracing::warn!(
error = %error,
"[python_exec] pool: pre-dispatch failure; falling back to legacy spawn"
);
None
}
// Load-shed: do NOT spawn (that reintroduces per-run RSS). Surface busy.
Err(crate::openhuman::runtime_pool::PoolRunError::Saturated) => {
tracing::warn!("[python_exec] pool: saturated; shedding load");
Some(ToolResult::error(
"Python runtime pool is at capacity; retry shortly.",
))
}
// The job may already have executed → terminal, never re-run it.
Err(crate::openhuman::runtime_pool::PoolRunError::PostDispatch(error)) => {
tracing::warn!(
error = %error,
"[python_exec] pool: post-dispatch failure; not retried to avoid duplicate execution"
);
Some(ToolResult::error(format!(
"python_exec failed after the code was dispatched to a pooled worker; not retried to avoid running it twice: {error}"
)))
}
}
}
/// Execute a python command through the sandbox backend. Mirrors
/// `NodeExecTool::run_sandboxed`.
async fn run_sandboxed(
&self,
security: &SecurityPolicy,
command: &str,
bin_dir: &std::path::Path,
timeout: Option<Duration>,
) -> ToolResult {
use crate::openhuman::sandbox;
let effective = timeout.unwrap_or_else(|| {
Duration::from_secs(crate::openhuman::tool_timeout::SANDBOX_UNBOUNDED_CAP_SECS)
});
let runtime_cfg = match crate::openhuman::config::ops::load_config_with_timeout().await {
Ok(cfg) => cfg.runtime,
Err(err) => {
tracing::warn!(
error = %err,
"[python_exec] failed to load live RuntimeConfig — falling back to defaults"
);
crate::openhuman::config::RuntimeConfig::default()
}
};
let policy = sandbox::resolve_sandbox_policy(
crate::openhuman::agent::harness::definition::SandboxMode::Sandboxed,
&security.action_dir,
&runtime_cfg,
false,
);
let mut extra_env = std::collections::HashMap::new();
let host_path = std::env::var("PATH").unwrap_or_default();
let sep = if cfg!(windows) { ";" } else { ":" };
let prepended = if host_path.is_empty() {
bin_dir.to_string_lossy().into_owned()
} else {
format!("{}{}{}", bin_dir.display(), sep, host_path)
};
extra_env.insert("PATH".to_string(), prepended);
extra_env.insert("PYTHONUNBUFFERED".to_string(), "1".to_string());
match sandbox::execute_in_sandbox(
&policy,
command,
&security.action_dir,
extra_env,
effective,
)
.await
{
Ok(result) => {
if result.timed_out {
ToolResult::error(format!(
"python_exec timed out after {}s and was killed",
effective.as_secs()
))
} else if result.success() {
if result.stderr.is_empty() {
ToolResult::success(result.stdout)
} else {
ToolResult::success(format!(
"{}\n[stderr]\n{}",
result.stdout, result.stderr
))
}
} else {
super::command_output::command_failure(
super::command_output::sandbox_exit_code(result.exit_code),
&result.stdout,
&result.stderr,
)
}
}
Err(e) => ToolResult::error(format!("Sandbox execution failed: {e}")),
}
}
}
/// Map a runtime-pool outcome onto the same `ToolResult` shape the legacy
/// `python -c` path produces.
fn pool_outcome_to_result(
outcome: crate::openhuman::runtime_pool::PoolExecOutcome,
timeout: Option<Duration>,
) -> ToolResult {
if outcome.timed_out {
return ToolResult::error(format!(
"python_exec timed out after {}s and was killed",
timeout.map(|d| d.as_secs()).unwrap_or(0)
));
}
let mut stdout = outcome.stdout;
let mut stderr = outcome.stderr;
if stdout.len() > MAX_OUTPUT_BYTES {
stdout.truncate(crate::openhuman::util::floor_char_boundary(
&stdout,
MAX_OUTPUT_BYTES,
));
stdout.push_str("\n... [stdout truncated at 1MB]");
}
if stderr.len() > MAX_OUTPUT_BYTES {
stderr.truncate(crate::openhuman::util::floor_char_boundary(
&stderr,
MAX_OUTPUT_BYTES,
));
stderr.push_str("\n... [stderr truncated at 1MB]");
}
let success = matches!(outcome.exit_code, None | Some(0));
if success {
if stderr.is_empty() {
ToolResult::success(stdout)
} else {
ToolResult::success(format!("{stdout}\n[stderr]\n{stderr}"))
}
} else {
super::command_output::command_failure(outcome.exit_code, &stdout, &stderr)
}
}
/// Resolve the wall-clock policy for a `python_exec` call from its args.
fn python_timeout_policy(args: &serde_json::Value) -> ToolTimeout {
match args.get("timeout_secs").and_then(|v| v.as_u64()) {
None | Some(0) => ToolTimeout::Unbounded,
Some(secs) => ToolTimeout::Secs(secs.min(PYTHON_TIMEOUT_MAX_SECS)),
}
}
/// POSIX-safe single-quote escaping (mirrors node_exec::shell_quote).
fn shell_quote(s: &str) -> String {
let escaped = s.replace('\'', "'\\''");
format!("'{escaped}'")
}
/// Resolve a caller-supplied `script_path` against the workspace. Rejects
/// absolute paths and any component that could escape the workspace.
fn resolve_script_path(
workspace: &std::path::Path,
raw: &str,
) -> Result<std::path::PathBuf, String> {
let raw = raw.trim();
if raw.is_empty() {
return Err("python_exec `script_path` cannot be empty".to_string());
}
let candidate = std::path::Path::new(raw);
if candidate.is_absolute() {
return Err(format!(
"python_exec `script_path` must be relative to workspace; got absolute {raw:?}"
));
}
if candidate.components().any(|c| {
matches!(
c,
std::path::Component::ParentDir | std::path::Component::Prefix(_)
)
}) {
return Err(format!(
"python_exec `script_path` must not escape workspace; got {raw:?}"
));
}
Ok(workspace.join(candidate))
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn python_timeout_policy_unbounded_by_default() {
assert_eq!(python_timeout_policy(&json!({})), ToolTimeout::Unbounded);
assert_eq!(
python_timeout_policy(&json!({"timeout_secs": 0})),
ToolTimeout::Unbounded
);
}
#[test]
fn python_timeout_policy_enforces_and_caps_explicit() {
assert_eq!(
python_timeout_policy(&json!({"timeout_secs": 120})),
ToolTimeout::Secs(120)
);
assert_eq!(
python_timeout_policy(&json!({"timeout_secs": 99999})),
ToolTimeout::Secs(PYTHON_TIMEOUT_MAX_SECS)
);
}
#[test]
fn shell_quote_escapes_single_quotes() {
assert_eq!(shell_quote("it's"), "'it'\\''s'");
assert_eq!(shell_quote("print('hi')"), "'print('\\''hi'\\'')'");
}
#[test]
fn resolve_script_path_rejects_escapes() {
let ws = std::path::Path::new("/ws");
assert!(resolve_script_path(ws, "").is_err());
assert!(resolve_script_path(ws, "../evil.py").is_err());
assert_eq!(
resolve_script_path(ws, "scripts/run.py").unwrap(),
std::path::Path::new("/ws/scripts/run.py")
);
}
}
+16
View File
@@ -1062,6 +1062,8 @@ pub fn all_tools_with_runtime(
security.clone(),
Arc::clone(&runtime),
Arc::clone(bootstrap),
root_config.runtime_pool.clone(),
root_config.workspace_dir.clone(),
)));
tools.push(Box::new(NpmExecTool::new(
security.clone(),
@@ -1071,6 +1073,20 @@ pub fn all_tools_with_runtime(
tracing::debug!("[tools::ops] registered node_exec + npm_exec");
}
// Managed Python exec tool — gated on `root_config.runtime_python.enabled`.
// Shares the same `PythonBootstrap` as ShellTool. Inline code routes through
// the shared runtime pool (#5106) when enabled.
if let Some(bootstrap) = python_bootstrap.as_ref() {
tools.push(Box::new(PythonExecTool::new(
security.clone(),
Arc::clone(&runtime),
Arc::clone(bootstrap),
root_config.runtime_pool.clone(),
root_config.workspace_dir.clone(),
)));
tracing::debug!("[tools::ops] registered python_exec");
}
// Vision tools are always available
tools.push(Box::new(ScreenshotTool::new(security.clone())));
tools.push(Box::new(ImageInfoTool::new(security.clone())));
+35
View File
@@ -1048,6 +1048,41 @@ fn all_tools_registers_node_exec_when_node_enabled() {
);
}
#[test]
fn all_tools_registers_python_exec_when_python_enabled() {
// Default RuntimePythonConfig has `enabled = true`, so `python_exec` must
// appear in the registry (routes inline code through the runtime pool, #5106).
let tmp = TempDir::new().unwrap();
let security = Arc::new(SecurityPolicy::default());
let mem_cfg = MemoryConfig {
backend: "markdown".into(),
..MemoryConfig::default()
};
let mem: Arc<dyn Memory> =
Arc::from(crate::openhuman::memory_store::create_memory(&mem_cfg, tmp.path()).unwrap());
let browser = BrowserConfig::default();
let http = crate::openhuman::config::HttpRequestConfig::default();
let cfg = test_config(&tmp);
let tools = all_tools(
Arc::new(Config::default()),
&security,
AuditLogger::disabled(),
mem,
&browser,
&http,
tmp.path(),
&HashMap::new(),
&cfg,
);
let names: Vec<&str> = tools.iter().map(|t| t.name()).collect();
assert!(
names.contains(&"python_exec"),
"python_exec must be registered when runtime_python.enabled=true; got: {names:?}"
);
}
#[test]
fn all_tools_excludes_node_exec_when_node_disabled() {
let tmp = TempDir::new().unwrap();
@@ -3587,7 +3587,16 @@ async fn node_and_npm_exec_tools_cover_validation_policy_and_disabled_runtime_pa
reqwest::Client::new(),
));
let node = NodeExecTool::new(full_security.clone(), runtime.clone(), bootstrap.clone());
let node = NodeExecTool::new(
full_security.clone(),
runtime.clone(),
bootstrap.clone(),
openhuman_core::openhuman::config::RuntimePoolConfig {
enabled: false,
..Default::default()
},
config.workspace_dir.clone(),
);
assert_eq!(node.name(), "node_exec");
assert_eq!(node.permission_level(), PermissionLevel::Execute);
assert!(node.description().contains("Execute JavaScript"));
@@ -3614,6 +3623,11 @@ async fn node_and_npm_exec_tools_cover_validation_policy_and_disabled_runtime_pa
readonly_security.clone(),
runtime.clone(),
bootstrap.clone(),
openhuman_core::openhuman::config::RuntimePoolConfig {
enabled: false,
..Default::default()
},
config.workspace_dir.clone(),
);
let blocked = readonly_node
.execute(json!({ "inline_code": "console.log('blocked')" }))