diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index a9932bc1..99da2938 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -316,7 +316,7 @@ jobs: - name: Check run: cargo check --manifest-path "$MANIFEST" --all-targets - name: Check with the language features (rutis-bridge 0.9) - run: cargo check --manifest-path "$MANIFEST" --all-targets --features node,python,bun,go,websocket + run: cargo check --manifest-path "$MANIFEST" --all-targets --features node,python,bun,go,websocket,probe # TypeScript (Node and Bun 1.4), Python and Go (1.24) plugins through rutis-bridge 0.9, end to end: real # Node 22 (the floor: pi needs 22.19, rutis 0.9 22) @@ -381,13 +381,15 @@ jobs: "$PLUGINS/.venv/bin/pip" install -r "$PLUGINS/requirements.txt" - name: Clippy (each language feature, and all) run: | - for features in node python bun go websocket node,python,bun,go,websocket; do + for features in node python bun go websocket probe node,python,bun,go,websocket,probe; do cargo clippy --manifest-path "$MANIFEST" --all-targets --features "$features" done - name: Documentation - run: RUSTDOCFLAGS="-Dwarnings" cargo doc --manifest-path "$MANIFEST" --no-deps --features node,python,bun,go,websocket + run: RUSTDOCFLAGS="-Dwarnings" cargo doc --manifest-path "$MANIFEST" --no-deps --features node,python,bun,go,websocket,probe + # `probe` adds tests/probe_test.rs: the pi and DSH adapters' fixtures + # through `yoagent_rutis::probe`. - name: Test - run: cargo test --manifest-path "$MANIFEST" --features node,python,bun,go,websocket + run: cargo test --manifest-path "$MANIFEST" --features node,python,bun,go,websocket,probe - name: Example run: cargo run --manifest-path "$MANIFEST" --features node,python --example language_plugins - name: pi extensions example (scripted, the fixture extension) diff --git a/CHANGELOG.md b/CHANGELOG.md index 442da229..132f39ad 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,23 @@ All notable changes to `yoagent` are documented here. The format loosely follows [Keep a Changelog](https://keepachangelog.com/), and the project adheres to [Semantic Versioning](https://semver.org/). +## Unreleased + +### yoagent-rutis (0.4.0) + +- **`yoagent_rutis::adapters`: where the pi and DSH adapters are.** `adapters()` returns the folders of the adapter files the crate ships (`plugins/pi`, `plugins/dsh`, `plugins/yoagent.d.ts`), so a host can copy them and `npm ci` the copy instead of keeping its own copy of them in step with the crate's. +- **`yoagent_rutis::probe` (feature `probe`): vet a plugin package without running it as an agent.** `probe::pi` and `probe::dsh` load a pi extension or DSH plugin rows through the adapter, as an agent would, and report what an agent gets — tools, prompt additions, which DSH services it needs and whether they exist, skills, and what the plugin logged — without calling a tool or a model. Both take `plugins`, a copy of `adapters().root` with `npm ci` run in its `pi/` and `dsh/`. Each report's `problem()` is the verdict (`None` = a clean load). Probes in one process run one at a time, so each report carries its own plugin logs. It is what `examples/ecosystem_probe.rs` (the compatibility matrix's probe, now with `--plugins`) ran; the example now calls it, with the same verdicts. The feature adds `rutis-loader`, `tempfile` and `tracing-subscriber`; the default build is unchanged. +- **The probe reports what used to read as clean.** + - A pi extension the adapter refuses after load (a handler for an event yoagent never fires, or a same-name tool added later) is `loaded: false` with the refusal, where it read as loaded with no tools. A late same-name tool is retried with `withoutBuiltins`, like one at load. + - A DSH plugin's injected services are read from its modules inside the runtime, so a missing one is found without the caller listing it (`inject` adds to them). Every key counts: Cordis requires them all, so one declared `required: false` and missing used to read clean (also in `inspect-package.mjs` and the matrix). A module whose `inject` could not be read is a problem, not a clean verdict. + - The probe copies its two DSH files (`dsh-tools-adapter.ts`, `probe-services.ts`) into `base` when they are missing: they must resolve the base's Cordis, and a base with only one of them failed every service check. + - DSH rows that are not an array of rows with string ids are an error. + - A bare `SKILL.md` path is a skill in the current folder (it was dropped). + - pi's early failures keep their logs. + - Logs include the bridge's reports of failed hooks (`yoagent_rutis::extension`). + - A DSH report's rows are only the plugin's (the base's host-service rows were listed too). + - On error paths the reports now carry empty fields (`rows: []`, `services: null`) where the old example printed nothing, and an unreadable `host-services.json` is an error, not a panic. + ## 0.25.5 (2026-10-16) ### yoagent diff --git a/integrations/yoagent-rutis/CLAUDE.md b/integrations/yoagent-rutis/CLAUDE.md index 3d4af21d..f02ec510 100644 --- a/integrations/yoagent-rutis/CLAUDE.md +++ b/integrations/yoagent-rutis/CLAUDE.md @@ -10,4 +10,4 @@ Its own `Cargo.toml` (not a workspace member; cargo leaves the nested package ou ### Plugin compatibility matrix -`scripts/ecosystem-matrix.mjs` (repo root) → `docs/evals/plugin-compatibility.{md,json}`: top N (default 20) npm packages by `downloads.monthly` among the keyword search results (`pi-package`, `dsh-plugin`; the API returns ≤1000), each probed by `examples/ecosystem_probe.rs` (feature `node`; one JSON line; nothing called, no model). **Harness errors are classified `failed` with the reason, never as a verdict** (a failed resolve, a service check that did not report, rows whose `inject` could not be read). pi: installed alone `--legacy-peer-deps --ignore-scripts` (pi supplies its own packages), resolved by `plugins/pi/resolve-package.mjs` (pi's `DefaultPackageManager.resolveExtensionSources`, in-memory settings, a temp agent dir removed after — as `pi -e `), loaded through the adapter; the probe waits for the adapter's handler, then re-reads the row (a TS row reads `Active` before its async `apply` ends); a same-name refusal is retried with that built-in in `withoutBuiltins`; an "adapter stopped" warning → `refused`; skills loaded with yoagent's `SkillSet` (paths compared canonicalized — the loader canonicalizes, and macOS `/var` is `/private/var`; rejected ones reported); the adapter's debug line `[pi] loaded : events …; tools …` gives the hooks. Logs via a **global** tracing subscriber (the bridge logs from worker threads). DSH: base = `plugins/dsh` lock `npm ci`'d once, copied per package; the plugin is installed in its own dir **with** its peers (its `@deepseek-ai/dsh-*` peers the base lacks at the base's release, other `@deepseek-ai/` peers at their range; retried without peers only if that fails, noted on the row), then **merged without npm** (`mergeInto`: the plugin first, replacing any base copy; the base's `@deepseek-ai`/`@arcships` packages win; any other name the base has is nested under the plugin — `npm install` in the copy re-plans and prunes/swaps DSH packages); the adapter and `probe-services.ts` are copied into it (one Cordis); `plugins/dsh/inspect-package.mjs` reads the manifest directly, the bundle patch's `insert` rows, and every row's `inject` (`importErrors` per row); the probe loads the base, then the rows, and diffs tools / guidance; `probe-services.ts` (plain Cordis plugin, row id `probe-services`) reports which injected services exist — a Node row waiting on a missing service reads `Active` to the host. A DSH plugin with a web client is "works, partly" (its UI ignored). `--from-json` re-labels earlier results without probing; `--only` keeps the other ecosystem's results and dates (`runs.{pi,dsh}`). Manual workflow `plugin-matrix.yml` (artifact only, read-only token, `persist-credentials: false`: loading runs third-party code). +`scripts/ecosystem-matrix.mjs` (repo root) → `docs/evals/plugin-compatibility.{md,json}`: top N (default 20) npm packages by `downloads.monthly` among the keyword search results (`pi-package`, `dsh-plugin`; the API returns ≤1000), each probed by `examples/ecosystem_probe.rs` (feature `probe`; one JSON line; nothing called, no model), the command-line face of `src/probe.rs` (`yoagent_rutis::probe`, feature `probe` = `node` + optional `rutis-loader`/`tempfile`/`tracing-subscriber`: `pi(plugins, cwd, extensions, skills, logs) -> PiReport`, `dsh(plugins, base, rows, inject, logs) -> DshReport` (`plugins` = a copy of `adapters().root` with `npm ci` in `pi/` and `dsh/`; the plugin's rows only, by id), typed, `#[non_exhaustive]`, `Serialize` to the JSON the matrix reads, `problem()` = the verdict; probes in one process serialized by a static tokio mutex (`Logs` is one buffer; it collects targets `yoagent_rutis::plugin` and `::extension`); a refusal after load (`before_model` → `Stop`/`Fail`, read in `offered`) is `loaded: false` + the refusal, and a late same-name refusal retries `withoutBuiltins`; `probe-services.ts` reads each row module's `inject` itself (`modules` + `base`, as `inspect-package.mjs` does; **every** key — Cordis requires them all, `required: false` included) plus the caller's `inject`; `dsh_file` copies `dsh-tools-adapter.ts` / `probe-services.ts` into `base` when missing (they must resolve base's Cordis); `importErrors` count in `problem()`; rows must be an array with string ids (`row_ids`); tests `tests/probe_test.rs` on the adapters' fixtures incl. `fixture-late-refusal.ts`, `fixture-shadow.ts`, `fixture-missing-service.ts`, unit tests pin the report JSON keys). `src/adapters.rs`: `adapters() -> Adapters { root, pi, dsh, types }`, the crate's packaged `plugins/` files (`env!("CARGO_MANIFEST_DIR")`; read or copy, never install into). **Harness errors are classified `failed` with the reason, never as a verdict** (a failed resolve, a service check that did not report, rows whose `inject` could not be read). pi: installed alone `--legacy-peer-deps --ignore-scripts` (pi supplies its own packages), resolved by `plugins/pi/resolve-package.mjs` (pi's `DefaultPackageManager.resolveExtensionSources`, in-memory settings, a temp agent dir removed after — as `pi -e `), loaded through the adapter; the probe waits for the adapter's handler, then re-reads the row (a TS row reads `Active` before its async `apply` ends); a same-name refusal is retried with that built-in in `withoutBuiltins`; an "adapter stopped" warning, or an `error` with it (the probe's refusal after load), → `refused`; skills loaded with yoagent's `SkillSet` (paths compared canonicalized — the loader canonicalizes, and macOS `/var` is `/private/var`; rejected ones reported); the adapter's debug line `[pi] loaded : events …; tools …` gives the hooks. Logs via a **global** tracing subscriber (the bridge logs from worker threads). DSH: base = `plugins/dsh` lock `npm ci`'d once, copied per package; the plugin is installed in its own dir **with** its peers (its `@deepseek-ai/dsh-*` peers the base lacks at the base's release, other `@deepseek-ai/` peers at their range; retried without peers only if that fails, noted on the row), then **merged without npm** (`mergeInto`: the plugin first, replacing any base copy; the base's `@deepseek-ai`/`@arcships` packages win; any other name the base has is nested under the plugin — `npm install` in the copy re-plans and prunes/swaps DSH packages); the adapter and `probe-services.ts` are copied into it (one Cordis); `plugins/dsh/inspect-package.mjs` reads the manifest directly, the bundle patch's `insert` rows, and every row's `inject` (`importErrors` per row); the probe loads the base, then the rows, and diffs tools / guidance; `probe-services.ts` (plain Cordis plugin, row id `probe-services`) reports which injected services exist — a Node row waiting on a missing service reads `Active` to the host. A DSH plugin with a web client is "works, partly" (its UI ignored). `--from-json` re-labels earlier results without probing; `--only` keeps the other ecosystem's results and dates (`runs.{pi,dsh}`). Manual workflow `plugin-matrix.yml` (artifact only, read-only token, `persist-credentials: false`: loading runs third-party code). diff --git a/integrations/yoagent-rutis/Cargo.toml b/integrations/yoagent-rutis/Cargo.toml index 212a3851..e9752a99 100644 --- a/integrations/yoagent-rutis/Cargo.toml +++ b/integrations/yoagent-rutis/Cargo.toml @@ -46,9 +46,14 @@ tokio = { version = "1", features = ["rt", "sync", "time", "macros"] } # Same crate rutis hands its cancellation tokens out as. tokio-util = "0.7" tracing = "0.1" +# The `probe` feature: loading plugin rows (the bridge itself never loads +# plugins), a DSH home per probe, and collecting what plugins log. +rutis-loader = { version = "0.9", optional = true, features = ["node"] } +tempfile = { version = "3", optional = true } +tracing-subscriber = { version = "0.3", optional = true, default-features = false, features = ["registry"] } [package.metadata.docs.rs] -features = ["node", "python", "bun", "go", "websocket"] +features = ["node", "python", "bun", "go", "websocket", "probe"] rustdoc-args = ["--cfg", "docsrs"] [features] @@ -66,6 +71,9 @@ python = ["rutis-bridge/python"] bun = ["rutis-bridge/bun"] # Remote nodes and runtimes over WebSocket (rustls). websocket = ["rutis-bridge/websocket"] +# `yoagent_rutis::probe`: load a pi or DSH package through its adapter and +# report what an agent gets, without calling a tool or a model (Unix). +probe = ["node", "dep:rutis-loader", "dep:tempfile", "dep:tracing-subscriber"] [dev-dependencies] tokio = { version = "1", features = ["rt-multi-thread", "macros", "time", "sync"] } @@ -100,6 +108,10 @@ required-features = ["node"] name = "pi_log_test" required-features = ["node"] +[[test]] +name = "probe_test" +required-features = ["probe"] + [[example]] name = "language_plugins" required-features = ["node", "python"] @@ -114,4 +126,4 @@ required-features = ["node"] [[example]] name = "ecosystem_probe" -required-features = ["node"] +required-features = ["probe"] diff --git a/integrations/yoagent-rutis/README.md b/integrations/yoagent-rutis/README.md index f7257f3a..51237a10 100644 --- a/integrations/yoagent-rutis/README.md +++ b/integrations/yoagent-rutis/README.md @@ -303,6 +303,36 @@ The end-to-end tests cover local Node and Python runtimes; a remote node over ## Tools from other agent ecosystems +The adapters below ship in the crate (`plugins/pi`, `plugins/dsh`, +`plugins/yoagent.d.ts`, without `node_modules`), and +`yoagent_rutis::adapters::adapters()` says where they are: the crate's own +files, where it was built (cargo's registry or a path dependency). Copy +the whole `root` (the adapters import `../yoagent.d.ts`) and run `npm ci` +in the copy's `pi/` and `dsh/`; never install into the registry's. + +With the `probe` feature, `yoagent_rutis::probe` loads a pi or DSH package +through its adapter and reports what an agent gets — tools, prompt +additions, the DSH services it needs and whether they exist, skills, and +what the plugin logged — without calling a tool or a model (loading runs +the package's code, as its own host would). Each report's `problem()` is +the verdict: `None` for a clean load, else what kept the package from +giving an agent what it offers (a refusal after load included). Probes in +one process run one at a time. + +```rust,no_run +# async fn check() { +use yoagent_rutis::probe; +// A copy of `adapters().root`, with `npm ci` run in its `pi/` and `dsh/`. +let plugins = std::path::Path::new("/path/to/plugins"); +let cwd = std::path::Path::new("/path/to/project"); +let report = probe::pi(plugins, cwd, &["/path/to/extension.ts".into()], &[], None).await; +match report.problem() { + None => println!("usable: {:?}", report.tools), + Some(why) => println!("not usable: {why}"), +} +# } +``` + ### dsh (DeepSeek Harness) tool plugins dsh plugins are Cordis plugins, and rutis's Node runtime runs them @@ -552,7 +582,7 @@ downloads, among npm's first 1000 search results) load through these adapters is measured in [plugin compatibility](../../docs/evals/plugin-compatibility.md) (`scripts/ecosystem-matrix.mjs`, which runs `examples/ecosystem_probe.rs` -per package). +— `yoagent_rutis::probe` on the command line — per package). The point is that **a plugin's safety policy never silently stops applying**. Not every rule arises for every adapter: the DSH adapter maps diff --git a/integrations/yoagent-rutis/examples/ecosystem_probe.rs b/integrations/yoagent-rutis/examples/ecosystem_probe.rs index 47bd0139..ba33c22c 100644 --- a/integrations/yoagent-rutis/examples/ecosystem_probe.rs +++ b/integrations/yoagent-rutis/examples/ecosystem_probe.rs @@ -1,502 +1,31 @@ //! Loads one third-party plugin package through its adapter, as an agent -//! would, and prints what yoagent gets from it as one JSON line. The -//! compatibility matrix (`scripts/ecosystem-matrix.mjs`) runs it per package; -//! it installs each package and resolves what is in it first. +//! would, and prints what yoagent gets from it as one JSON line: the +//! command-line face of [`yoagent_rutis::probe`]. The compatibility matrix +//! (`scripts/ecosystem-matrix.mjs`) runs it per package; it installs each +//! package and resolves what is in it first. //! -//! - `pi --cwd DIR [--skill PATH]... EXTENSION...`: pi extension files (as -//! pi's package manager resolved them) through the pi adapter. When the -//! adapter refuses a tool named like a yoagent built-in, the load is tried -//! again with that built-in left out (`withoutBuiltins`), as a host would. -//! Skills are loaded with yoagent's own skills loader. -//! - `dsh --base DIR --rows JSON [--inject SERVICE]...`: DSH plugin rows (its bundle patch, or one -//! row naming the package) over DSH's base packages and the tools adapter, -//! in a Node runtime whose `package.json` is `DIR`'s (with a copy of -//! `plugins/dsh/dsh-tools-adapter.ts` in `DIR` when it is not this crate's -//! `plugins/dsh`, so both resolve one Cordis). Reports what the rows add -//! to the base: tools and prompt guidance, and which of the services the -//! plugin injects exist (`plugins/dsh/probe-services.ts`, from inside the -//! runtime: a plugin missing one never starts, though its row reads as -//! started to the host). +//! - `pi --cwd DIR [--skill PATH]... EXTENSION...`: [`probe::pi`]. +//! - `dsh --base DIR --rows JSON [--inject SERVICE]...`: [`probe::dsh`]. +//! +//! `--plugins DIR` names the installed adapters (default: this crate's own +//! `plugins/`, with `npm ci` run in its `pi/` and `dsh/`). //! //! Nothing is called: tools are listed, not run, and the model is never //! asked. Loading a package runs its code, as its own host would. //! -//! Run: `cargo run --manifest-path integrations/yoagent-rutis/Cargo.toml --features node --example ecosystem_probe -- pi --cwd /tmp/x ext.ts` +//! Run: `cargo run --manifest-path integrations/yoagent-rutis/Cargo.toml --features probe --example ecosystem_probe -- pi --cwd /tmp/x ext.ts` -use std::path::{Path, PathBuf}; -use std::sync::{Arc, Mutex}; -use std::time::Duration; +use std::path::PathBuf; -use rutis::{Ctx, FiberState}; -use rutis_bridge::runtime::{Launcher, LocalRuntime}; -use rutis_loader::{ - Chain, EntryStatus, Layer, Loader, LoaderOptions, LoaderPlugin, Patch, RuntimeResolver, - RuntimeRowsPlugin, ServiceCatalog, -}; -use serde_json::{json, Value}; -use tokio_util::sync::CancellationToken; +use serde_json::Value; use tracing_subscriber::layer::SubscriberExt; -use yoagent::extension::{RunContext, RunEnd, RunOutcome, TurnDecision}; -use yoagent::{Extension, Message, SkillSet, TurnContext}; -use yoagent_rutis::RutisBridge; +use yoagent_rutis::adapters::adapters; +use yoagent_rutis::probe::{self, Logs}; type BoxError = Box; -/// How long rows may take to start, and the adapter to register. -const SETTLE: Duration = Duration::from_secs(60); - -fn plugins_dir() -> PathBuf { - Path::new(env!("CARGO_MANIFEST_DIR")).join("plugins") -} - -/// Every plugin log line (the adapters report through the bridge's `log`). -#[derive(Clone, Default)] -struct Logs(Arc>>); - -impl tracing_subscriber::Layer for Logs { - fn on_event(&self, event: &tracing::Event<'_>, _: tracing_subscriber::layer::Context<'_, S>) { - if event.metadata().target() != "yoagent_rutis::plugin" { - return; - } - #[derive(Default)] - struct Message(String); - impl tracing::field::Visit for Message { - fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { - if field.name() == "message" { - self.0 = format!("{value:?}"); - } - } - } - let mut message = Message::default(); - event.record(&mut message); - let level = event.metadata().level().to_string().to_lowercase(); - self.0 - .lock() - .unwrap() - .push(json!({ "level": level, "message": message.0 })); - } -} - -struct Host { - root: Ctx, - bridge: RutisBridge, - loader: Loader, -} - -/// A rutis root with the bridge and one Node runtime whose `package.json` is -/// `package_json` (where row names resolve). -async fn host(runtime: &Path, package_json: &Path) -> Result { - host_with(runtime, package_json, &[]).await -} - -/// [`host`], the runtime's process given `env` (DSH's `DSH_HOME`). -async fn host_with( - runtime: &Path, - package_json: &Path, - env: &[(&str, &Path)], -) -> Result { - let root = Ctx::root()?; - let bridge = RutisBridge::install(&root)?; - let mut catalog = ServiceCatalog::new(); - catalog.register_shared("yoagent"); - let launcher = env - .iter() - .fold(Launcher::node(runtime), |launcher, (name, value)| { - launcher.env(*name, *value) - }); - let node = LocalRuntime::launcher("node", launcher, package_json); - let resolver = Arc::new(RuntimeResolver::node(node.handle()).with_catalog(&catalog)); - root.plugin(node); - let plugin = LoaderPlugin::new( - Chain::new().with_shared(resolver.clone()), - LoaderOptions { - catalog, - ..LoaderOptions::default() - }, - ); - let loader = plugin.handle(); - root.plugin(plugin).await?; - root.plugin(RuntimeRowsPlugin::new(resolver)); - Ok(Host { - root, - bridge, - loader, - }) -} - -/// One row's state once settled: `active`, `failed`, `unresolved`, or -/// `waiting` (still pending after the wait: a dependency never came). -fn row_state(loader: &Loader, id: &str) -> (String, Option) { - match loader.get(id).map(|info| info.status) { - Some(EntryStatus::Running(s)) => match s.state { - FiberState::Active => ("active".into(), None), - FiberState::Failed => ("failed".into(), s.error.map(|e| e.to_string())), - FiberState::Pending => ("waiting".into(), None), - other => (format!("{other:?}").to_lowercase(), None), - }, - Some(EntryStatus::Unresolved(e)) => ("unresolved".into(), Some(e.to_string())), - Some(other) => (format!("{other:?}").to_lowercase(), None), - None => ("missing".into(), None), - } -} - -/// Load the given layers in one reconcile and wait until each row is -/// active or failed (or the wait ends); each row's state. -async fn load(host: &Host, layers: Vec<(&str, Value)>) -> Result, BoxError> { - let mut ids = Vec::new(); - let mut built = Vec::new(); - for (name, rows) in layers { - ids.extend( - rows.as_array() - .into_iter() - .flatten() - .filter_map(|row| row["id"].as_str().map(str::to_owned)), - ); - let patches: Vec = serde_json::from_value(json!([{ "insert": rows }]))?; - built.push(Layer::new(name, patches)); - } - host.loader.reconcile(built, None).await?; - // Rows still pending when the wait ends are reported as `waiting`. - let _ = tokio::time::timeout(SETTLE, async { - while !ids.iter().all(|id| { - matches!( - row_state(&host.loader, id).0.as_str(), - "active" | "failed" | "unresolved" - ) - }) { - tokio::time::sleep(Duration::from_millis(50)).await; - } - }) - .await; - Ok(ids - .iter() - .map(|id| { - let (state, error) = row_state(&host.loader, id); - json!({ "id": id, "state": state, "error": error }) - }) - .collect()) -} - -/// Wait until a handler named `name` is registered. -async fn handler(host: &Host, name: &str) -> bool { - tokio::time::timeout(SETTLE, async { - while !host - .bridge - .registry() - .handlers() - .iter() - .any(|h| h.name() == name) - { - tokio::time::sleep(Duration::from_millis(50)).await; - } - }) - .await - .is_ok() -} - -/// The tools a run would be offered, and the note its first request would carry. -async fn offered(host: &Host) -> Result<(Vec, Option), BoxError> { - let cancel = CancellationToken::new(); - let prompt = [Message::user("hello")]; - let run = RunContext::new("probe", &prompt, &cancel); - let mut hooks = host.bridge.extension().start_run(&run).await?; - let mut tools: Vec = hooks - .tools(&run) - .await - .iter() - .map(|t| t.name().to_string()) - .collect(); - tools.sort(); - let turn = TurnContext::new("", &prompt, &[], "probe").with_run_prompts(&prompt); - let note = match hooks.before_model(&turn).await { - TurnDecision::Note(note) => Some(note), - _ => None, - }; - hooks - .finish(&RunOutcome::new(RunEnd::Completed, None)) - .await; - Ok((tools, note)) -} - -async fn shutdown(host: Host) { - let _ = tokio::time::timeout(Duration::from_secs(10), host.root.shutdown()).await; -} - -/// `SKILL.md` paths or skill directories, loaded with yoagent's loader. -fn skills(paths: &[PathBuf]) -> Value { - let mut loaded = Vec::new(); - let mut errors = Vec::new(); - for path in paths { - let dir = if path.ends_with("SKILL.md") { - path.parent().unwrap_or(path) - } else { - path - }; - let Some(parent) = dir.parent() else { continue }; - let (set, errs) = SkillSet::load_dir_resilient(parent, "probe"); - // The loader keeps canonical paths (on macOS /var is /private/var). - let dir = std::fs::canonicalize(dir).unwrap_or_else(|_| dir.to_path_buf()); - match set.skills().iter().find(|s| s.base_dir == dir) { - Some(skill) => loaded.push(skill.name.clone()), - None => { - // The loader's error about this skill, not a sibling's. - let name = dir.file_name().map(|n| n.to_string_lossy().into_owned()); - let error = errs - .iter() - .map(|e| e.to_string()) - .find(|e| name.as_deref().is_some_and(|n| e.contains(n))); - errors.push(format!( - "{}: {}", - dir.display(), - error.unwrap_or_else(|| "not loaded, no error given".into()) - )); - } - } - } - json!({ "found": paths.len(), "loaded": loaded, "errors": errors }) -} - -/// The built-in yoagent's tool `name` a refusal names, if it is a -/// same-name refusal. -fn shadowed(error: &str) -> Option { - let rest = error.split("named like yoagent's built-in \"").nth(1)?; - Some(rest.split('"').next()?.to_string()) -} - -async fn probe_pi( - logs: Logs, - cwd: PathBuf, - extensions: Vec, - skill_paths: Vec, -) -> Value { - let runtime = plugins_dir().join("pi/node_modules/@arcships/rutis-runtime"); - let mut without: Vec = Vec::new(); - let mut result = json!({ "ecosystem": "pi", "skills": skills(&skill_paths) }); - if extensions.is_empty() { - result["loaded"] = json!(null); - return result; - } - // A same-name refusal is retried with that built-in left out, a few times. - for _ in 0..6 { - logs.0.lock().unwrap().clear(); - let host = match host(&runtime, &plugins_dir().join("pi/package.json")).await { - Ok(host) => host, - Err(e) => { - result["loaded"] = json!(false); - result["error"] = json!(format!("host: {e}")); - return result; - } - }; - let row = json!([{ - "id": "pi", - "name": plugins_dir().join("pi/pi-extensions-adapter.ts"), - "config": { "extensions": extensions, "cwd": cwd, "withoutBuiltins": without }, - }]); - let rows = match load(&host, vec![("probe", row)]).await { - Ok(rows) => rows, - Err(e) => { - result["loaded"] = json!(false); - result["error"] = json!(e.to_string()); - shutdown(host).await; - return result; - } - }; - // A TS row reads active before its async `apply` ends: wait for the - // adapter's handler, then read the row again (a refusal fails it late). - let registered = rows[0]["state"] == "active" && handler(&host, "pi-extensions").await; - let (state, error) = row_state(&host.loader, "pi"); - if let Some(name) = error.as_deref().and_then(shadowed) { - if !without.contains(&name) { - without.push(name); - shutdown(host).await; - continue; - } - } - result["rows"] = json!([{ "id": "pi", "state": state, "error": error }]); - result["without"] = json!(without); - if registered && state == "active" { - match offered(&host).await { - Ok((tools, note)) => { - result["loaded"] = json!(true); - result["tools"] = json!(tools); - result["note_chars"] = json!(note.map_or(0, |n| n.len())); - } - Err(e) => { - result["loaded"] = json!(false); - result["error"] = json!(format!("starting a run: {e}")); - } - } - } else { - result["loaded"] = json!(false); - result["error"] = json!( - error.unwrap_or_else(|| format!("the adapter did not register (row {state})")) - ); - } - // Let late log lines arrive (a warning decides "works, partly"), - // then again after shutdown, which flushes the runtime. - tokio::time::sleep(Duration::from_millis(500)).await; - shutdown(host).await; - result["logs"] = json!(logs.0.lock().unwrap().clone()); - return result; - } - result["loaded"] = json!(false); - result["error"] = json!("too many same-name tools"); - result -} - -/// A file of `plugins/dsh`, its copy in `base` when there is one: it must -/// resolve the base's Cordis. -fn dsh_file(base: &Path, name: &str) -> PathBuf { - let copy = base.join(name); - if copy.exists() { - copy - } else { - plugins_dir().join("dsh").join(name) - } -} - -/// The row id of `plugins/dsh/probe-services.ts`; a failure of it is -/// reported under `services.error`. -const SERVICE_CHECK: &str = "probe-services"; - -/// DSH's base packages and the tools adapter: what every DSH plugin gets. -fn dsh_base_rows(base: &Path) -> Value { - let adapter = dsh_file(base, "dsh-tools-adapter.ts"); - // The services a headless DSH host runs (fs, attachments, credentials, - // storage, session persistence, skills): DSH's own packages, listed in - // `plugins/dsh/host-services.json` for every host to load. - let path = plugins_dir().join("dsh/host-services.json"); - let services: Vec = std::fs::read_to_string(&path) - .map_err(|e| e.to_string()) - .and_then(|text| serde_json::from_str(&text).map_err(|e| e.to_string())) - .unwrap_or_else(|e| panic!("{}: {e}", path.display())); - let mut rows = services; - rows.extend([ - json!({ "id": "base-web", "name": "@deepseek-ai/dsh-web" }), - json!({ "id": "base-system-prompt", "name": "@deepseek-ai/dsh-system-prompt" }), - json!({ "id": "base-tools", "name": "@deepseek-ai/dsh-tools" }), - json!({ "id": "base-user-questions", "name": "@deepseek-ai/dsh-user-questions" }), - json!({ "id": "base-adapter", "name": adapter }), - ]); - Value::Array(rows) -} - -async fn probe_dsh(logs: Logs, base: PathBuf, rows: Value, inject: Vec) -> Value { - let mut result = json!({ "ecosystem": "dsh" }); - let runtime = base.join("node_modules/@arcships/rutis-runtime"); - // A DSH home (and skills home) of its own: `~/.dsh` and `~/.agents` are - // not handed to the plugins probed (the process environment still is). - let dsh_home = tempfile::tempdir().expect("a temp DSH_HOME"); - let agents_home = dsh_home.path().join("agents"); - let host = match host_with( - &runtime, - &base.join("package.json"), - &[ - ("DSH_HOME", dsh_home.path()), - ("DSH_AGENTS_HOME", &agents_home), - ], - ) - .await - { - Ok(host) => host, - Err(e) => { - result["error"] = json!(format!("host: {e}")); - return result; - } - }; - // The base alone first, then the plugin's rows on top: the difference is - // what the plugin adds. - let base_state = load(&host, vec![("base", dsh_base_rows(&base))]).await; - let ready = matches!(&base_state, Ok(rows) if rows.iter().all(|r| r["state"] == "active")) - && handler(&host, "dsh-tools").await; - if !ready { - result["error"] = json!(format!("the DSH base did not start: {base_state:?}")); - shutdown(host).await; - return result; - } - let before = offered(&host).await; - // The plugin's rows, and a check (from inside the runtime) of which of - // the services it injects exist: without one it never starts. - let out = std::env::temp_dir().join(format!( - "yoagent-probe-services-{}.json", - std::process::id() - )); - let _ = std::fs::remove_file(&out); - let mut plugin_rows = rows.as_array().cloned().unwrap_or_default(); - plugin_rows.push(json!({ - "id": SERVICE_CHECK, - "name": dsh_file(&base, "probe-services.ts"), - "config": { "names": inject, "out": out, "waitMs": 3000 }, - })); - let state = load( - &host, - vec![ - ("base", dsh_base_rows(&base)), - ("plugin", json!(plugin_rows)), - ], - ) - .await; - let services = tokio::time::timeout(Duration::from_secs(20), async { - loop { - if let Some(found) = std::fs::read_to_string(&out) - .ok() - .and_then(|text| serde_json::from_str::(&text).ok()) - { - return found; - } - tokio::time::sleep(Duration::from_millis(100)).await; - } - }) - .await - .unwrap_or_else(|_| { - let (state, error) = row_state(&host.loader, SERVICE_CHECK); - json!({ "error": format!("no report within 20 s (row {state}{})", error.map(|e| format!(": {e}")).unwrap_or_default()) }) - }); - let _ = std::fs::remove_file(&out); - result["services"] = services; - let after = offered(&host).await; - match (state, before, after) { - (Ok(state), Ok((tools_before, note_before)), Ok((tools_after, note_after))) => { - let plugin_rows: Vec = state - .into_iter() - .filter(|r| { - let id = r["id"].as_str().unwrap_or(""); - !id.starts_with("base-") && id != SERVICE_CHECK - }) - .collect(); - let tools: Vec = tools_after - .into_iter() - .filter(|t| !tools_before.contains(t)) - .collect(); - let (before, after) = ( - note_before.unwrap_or_default(), - note_after.unwrap_or_default(), - ); - let guidance: Vec<&str> = after - .lines() - .filter(|line| !line.trim().is_empty() && !before.contains(line)) - .collect(); - result["rows"] = json!(plugin_rows); - result["tools"] = json!(tools); - result["guidance_chars"] = json!(guidance.iter().map(|l| l.len() + 1).sum::()); - } - (state, before, after) => { - result["error"] = json!(format!( - "{:?} {:?} {:?}", - state.err(), - before.err(), - after.err() - )); - } - } - tokio::time::sleep(Duration::from_millis(200)).await; - shutdown(host).await; - result["logs"] = json!(logs.0.lock().unwrap().clone()); - result -} - fn usage() -> BoxError { - "usage: ecosystem_probe pi --cwd DIR [--skill PATH]... EXTENSION... | dsh --base DIR --rows JSON [--inject SERVICE]..." + "usage: ecosystem_probe pi [--plugins DIR] --cwd DIR [--skill PATH]... EXTENSION... | dsh [--plugins DIR] --base DIR --rows JSON [--inject SERVICE]..." .into() } @@ -504,14 +33,16 @@ fn usage() -> BoxError { async fn main() -> Result<(), BoxError> { let mut args = std::env::args().skip(1); let mode = args.next().ok_or_else(usage)?; + let mut plugins = adapters().root; let mut cwd = None; let mut base = None; let mut rows = None; let mut inject = Vec::new(); - let mut skill_paths = Vec::new(); + let mut skills = Vec::new(); let mut extensions = Vec::new(); while let Some(arg) = args.next() { match arg.as_str() { + "--plugins" => plugins = args.next().map(PathBuf::from).ok_or_else(usage)?, "--cwd" => cwd = args.next().map(PathBuf::from), "--base" => base = args.next().map(PathBuf::from), "--rows" => { @@ -519,7 +50,7 @@ async fn main() -> Result<(), BoxError> { &args.next().ok_or_else(usage)?, )?) } - "--skill" => skill_paths.extend(args.next().map(PathBuf::from)), + "--skill" => skills.extend(args.next().map(PathBuf::from)), "--inject" => inject.extend(args.next()), _ => extensions.push(arg), } @@ -527,19 +58,20 @@ async fn main() -> Result<(), BoxError> { // Global: the bridge logs from the runtime's worker threads. let logs = Logs::default(); tracing::subscriber::set_global_default(tracing_subscriber::registry().with(logs.clone()))?; - let result = match mode.as_str() { - "pi" => probe_pi(logs, cwd.ok_or_else(usage)?, extensions, skill_paths).await, + let mut result = match mode.as_str() { + "pi" => { + let cwd = cwd.ok_or_else(usage)?; + serde_json::to_value( + probe::pi(&plugins, &cwd, &extensions, &skills, Some(&logs)).await, + )? + } "dsh" => { - probe_dsh( - logs, - base.ok_or_else(usage)?, - rows.ok_or_else(usage)?, - inject, - ) - .await + let (base, rows) = (base.ok_or_else(usage)?, rows.ok_or_else(usage)?); + serde_json::to_value(probe::dsh(&plugins, &base, &rows, &inject, Some(&logs)).await)? } _ => return Err(usage()), }; + result["ecosystem"] = mode.into(); println!("{}", serde_json::to_string(&result)?); Ok(()) } diff --git a/integrations/yoagent-rutis/plugins/dsh/fixture-missing-service.ts b/integrations/yoagent-rutis/plugins/dsh/fixture-missing-service.ts new file mode 100644 index 00000000..a9ae378d --- /dev/null +++ b/integrations/yoagent-rutis/plugins/dsh/fixture-missing-service.ts @@ -0,0 +1,9 @@ +// A test fixture for the probe (`tests/probe_test.rs`): a Cordis plugin that +// injects a service no DSH host provides. It never starts, while its row +// reads as started to the host; the probe finds the missing service by +// reading `inject` from this module. Not for production use. + +export const name = 'yoagent-fixture-missing-service' +export const inject = ['yoagentNoSuchService'] + +export function apply(_ctx: unknown) {} diff --git a/integrations/yoagent-rutis/plugins/dsh/fixture-optional-service.ts b/integrations/yoagent-rutis/plugins/dsh/fixture-optional-service.ts new file mode 100644 index 00000000..b67ad8d2 --- /dev/null +++ b/integrations/yoagent-rutis/plugins/dsh/fixture-optional-service.ts @@ -0,0 +1,9 @@ +// A test fixture for the probe (`tests/probe_test.rs`): a Cordis plugin that +// injects a service no DSH host provides, declared `required: false`. Cordis +// requires it all the same, so it never starts; the probe must not take +// `required: false` as optional. Not for production use. + +export const name = 'yoagent-fixture-optional-service' +export const inject = { yoagentNoSuchOptional: { required: false } } + +export function apply(_ctx: unknown) {} diff --git a/integrations/yoagent-rutis/plugins/dsh/inspect-package.mjs b/integrations/yoagent-rutis/plugins/dsh/inspect-package.mjs index 753dbb1c..241b7155 100644 --- a/integrations/yoagent-rutis/plugins/dsh/inspect-package.mjs +++ b/integrations/yoagent-rutis/plugins/dsh/inspect-package.mjs @@ -55,7 +55,7 @@ const injectOf = async (spec) => { return Array.isArray(inject) ? inject : inject && typeof inject === 'object' - ? Object.entries(inject).map(([key, value]) => (value?.required === false ? `${key}?` : key)) + ? Object.keys(inject) // Cordis treats every key as required : [] } const injected = new Set() diff --git a/integrations/yoagent-rutis/plugins/dsh/probe-services.ts b/integrations/yoagent-rutis/plugins/dsh/probe-services.ts index 77e37677..8f3271d1 100644 --- a/integrations/yoagent-rutis/plugins/dsh/probe-services.ts +++ b/integrations/yoagent-rutis/plugins/dsh/probe-services.ts @@ -1,34 +1,73 @@ -// For the ecosystem probe (examples/ecosystem_probe.rs): which of the named -// Cordis services exist in this DSH runtime. A plugin injecting one that does -// not never starts, while its row still reads as started to the host, so -// the probe asks from inside. Writes `{ present, missing }` to `config.out` -// after `config.waitMs` (default 3000), giving late providers their chance. +// For the probe (`yoagent_rutis::probe::dsh`): which of the Cordis services +// a DSH plugin's rows inject exist in this DSH runtime. A plugin injecting +// one that does not never starts, while its row still reads as started to +// the host, so the probe asks from inside. What the rows inject is read +// from their modules here (`config.modules`, resolved from `config.base`'s +// package.json, as DSH would), plus any names in `config.names`. Writes +// `{ present, missing, importErrors }` to `config.out` after `config.waitMs` +// (default 3000), giving late providers their chance. A module that cannot +// be imported here is listed in `importErrors`: what it injects is unknown. +// Every injected service is checked: Cordis requires them all, even one +// declared `required: false`. // // A plain Cordis plugin, injecting nothing, so it always starts. import { writeFileSync } from 'node:fs' +import { createRequire } from 'node:module' +import { join, resolve } from 'node:path' +import { pathToFileURL } from 'node:url' export const name = 'yoagent-probe-services' export interface Config { - names: string[] + names?: string[] + /** The plugin rows' module names (packages or files). */ + modules?: string[] + /** The folder whose package.json the modules resolve from. */ + base?: string out: string waitMs?: number } +/** What a module's plugin injects: `inject` as a list, or the keys of an object. Cordis treats every key as + * required (`required: false` included): a plugin missing any of them never starts. */ +async function injectOf(spec: string, base: string): Promise { + const require = createRequire(join(resolve(base), 'package.json')) + const mod = await import(pathToFileURL(require.resolve(spec)).href) + const plugin = mod.default ?? mod + const inject = plugin.inject ?? mod.inject + return Array.isArray(inject) + ? inject + : inject && typeof inject === 'object' + ? Object.keys(inject) + : [] +} + export function apply(ctx: any, config: Config) { - const timer = setTimeout(() => { + const names = new Set(config.names ?? []) + const importErrors: Record = {} + const read = Promise.all( + (config.modules ?? []).map(async (spec) => { + try { + for (const name of await injectOf(spec, config.base ?? process.cwd())) names.add(name) + } catch (error: any) { + importErrors[spec] = String(error?.message ?? error).slice(0, 300) + } + }), + ) + const timer = setTimeout(async () => { + await read const present: string[] = [] const missing: string[] = [] - for (const name of config.names ?? []) { + for (const entry of names) { let found = false try { - found = ctx.get(name) !== undefined + found = ctx.get(entry) !== undefined } catch { found = false } - ;(found ? present : missing).push(name) + ;(found ? present : missing).push(entry) } - writeFileSync(config.out, JSON.stringify({ present, missing })) + writeFileSync(config.out, JSON.stringify({ present, missing, importErrors })) }, config.waitMs ?? 3000) ctx.effect(() => () => clearTimeout(timer)) } diff --git a/integrations/yoagent-rutis/plugins/pi/fixture-late-refusal.ts b/integrations/yoagent-rutis/plugins/pi/fixture-late-refusal.ts new file mode 100644 index 00000000..c081f11b --- /dev/null +++ b/integrations/yoagent-rutis/plugins/pi/fixture-late-refusal.ts @@ -0,0 +1,25 @@ +// A pi extension for the probe's tests (`tests/probe_test.rs`): it loads +// cleanly, then registers a handler for an event yoagent never fires +// (`before_provider_request`, which could rewrite the model request). The +// adapter refuses it from then on; the probe must report that, not a clean +// load. + +import { defineTool, type ExtensionAPI } from '@earendil-works/pi-coding-agent' +import { Type } from 'typebox' + +export default function (pi: ExtensionAPI) { + pi.registerTool( + defineTool({ + name: 'pi_late_ok', + label: 'Late', + description: 'Registered at load.', + parameters: Type.Object({}), + async execute() { + return { content: [{ type: 'text', text: 'ok' }], details: {} } + }, + }), + ) + setTimeout(() => { + pi.on('before_provider_request', (event: any) => event) + }, 0) +} diff --git a/integrations/yoagent-rutis/plugins/pi/fixture-late-shadow.ts b/integrations/yoagent-rutis/plugins/pi/fixture-late-shadow.ts new file mode 100644 index 00000000..fc079769 --- /dev/null +++ b/integrations/yoagent-rutis/plugins/pi/fixture-late-shadow.ts @@ -0,0 +1,23 @@ +// A pi extension for the probe's tests (`tests/probe_test.rs`): after load, +// it registers a tool named like yoagent's built-in `read_file`. The adapter +// refuses it then; the probe retries with that built-in left out +// (`withoutBuiltins`), as it does for one registered at load. + +import { defineTool, type ExtensionAPI } from '@earendil-works/pi-coding-agent' +import { Type } from 'typebox' + +export default function (pi: ExtensionAPI) { + setTimeout(() => { + pi.registerTool( + defineTool({ + name: 'read_file', + label: 'Read', + description: 'A pi read_file, registered late.', + parameters: Type.Object({ path: Type.String() }), + async execute(_id, params) { + return { content: [{ type: 'text', text: `pi read ${params.path}` }], details: {} } + }, + }), + ) + }, 0) +} diff --git a/integrations/yoagent-rutis/plugins/pi/fixture-shadow.ts b/integrations/yoagent-rutis/plugins/pi/fixture-shadow.ts new file mode 100644 index 00000000..7b9b0164 --- /dev/null +++ b/integrations/yoagent-rutis/plugins/pi/fixture-shadow.ts @@ -0,0 +1,20 @@ +// A pi extension for the probe's tests (`tests/probe_test.rs`): a tool named +// like yoagent's built-in `read_file`. The adapter refuses it unless the host +// leaves its own `read_file` out (`withoutBuiltins`), which the probe tries. + +import { defineTool, type ExtensionAPI } from '@earendil-works/pi-coding-agent' +import { Type } from 'typebox' + +export default function (pi: ExtensionAPI) { + pi.registerTool( + defineTool({ + name: 'read_file', + label: 'Read', + description: 'A pi read_file.', + parameters: Type.Object({ path: Type.String() }), + async execute(_id, params) { + return { content: [{ type: 'text', text: `pi read ${params.path}` }], details: {} } + }, + }), + ) +} diff --git a/integrations/yoagent-rutis/src/adapters.rs b/integrations/yoagent-rutis/src/adapters.rs new file mode 100644 index 00000000..6cd32bac --- /dev/null +++ b/integrations/yoagent-rutis/src/adapters.rs @@ -0,0 +1,70 @@ +//! Where this crate's adapter files are: the pi and DSH adapters and the +//! TypeScript declarations of the `yoagent` host service, as published in +//! the crate (`plugins/`, without `node_modules`). +//! +//! The files are this crate's own, in cargo's registry (or the path +//! dependency's folder): read them, or copy them to install their npm +//! dependencies — never install into them. Copy the whole `root` (the +//! adapters import `../yoagent.d.ts`) and run `npm ci` in the copy's `pi/` +//! and `dsh/`; that copy is what the `probe` module (feature `probe`) takes. The paths are where +//! the crate was built: a binary run on another machine must ship a copy. +//! +//! ```no_run +//! let adapters = yoagent_rutis::adapters::adapters(); +//! let adapter = adapters.pi.join("pi-extensions-adapter.ts"); +//! # let _ = adapter; +//! ``` + +use std::path::{Path, PathBuf}; + +/// The adapter folders and files of this crate. +#[derive(Debug, Clone, PartialEq, Eq)] +#[non_exhaustive] +pub struct Adapters { + /// `plugins/`: everything below. + pub root: PathBuf, + /// `plugins/pi`: the pi extensions adapter (`pi-extensions-adapter.ts`), + /// `host-ui.ts`, `resolve-package.mjs`, and their `package.json` and lock. + pub pi: PathBuf, + /// `plugins/dsh`: the DSH tools adapter (`dsh-tools-adapter.ts`), the + /// host services a headless DSH host loads (`host-services.json`), and + /// their `package.json` and lock. + pub dsh: PathBuf, + /// `plugins/yoagent.d.ts`: the `yoagent` host service's TypeScript types. + pub types: PathBuf, +} + +/// This crate's adapter files. +pub fn adapters() -> Adapters { + let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("plugins"); + Adapters { + pi: root.join("pi"), + dsh: root.join("dsh"), + types: root.join("yoagent.d.ts"), + root, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn the_adapter_files_are_there() { + let a = adapters(); + for file in [ + a.pi.join("pi-extensions-adapter.ts"), + a.pi.join("package.json"), + a.dsh.join("dsh-tools-adapter.ts"), + a.pi.join("package-lock.json"), + a.dsh.join("host-services.json"), + a.dsh.join("probe-services.ts"), + a.dsh.join("package.json"), + a.dsh.join("package-lock.json"), + a.types.clone(), + ] { + assert!(file.is_file(), "{}", file.display()); + } + assert!(a.pi.starts_with(&a.root) && a.dsh.starts_with(&a.root)); + } +} diff --git a/integrations/yoagent-rutis/src/lib.rs b/integrations/yoagent-rutis/src/lib.rs index 2f2b6d34..a6402402 100644 --- a/integrations/yoagent-rutis/src/lib.rs +++ b/integrations/yoagent-rutis/src/lib.rs @@ -30,9 +30,10 @@ //! can return images in tool results and write logs into the host's //! `tracing` output (target `yoagent_rutis::plugin`). Plugins from other //! agent ecosystems — DSH tool plugins, pi extensions — plug in through -//! adapters shipped in the repository's `plugins/` directory; -//! see the [README](https://github.com/yologdev/yoagent/tree/main/integrations/yoagent-rutis) -//! and its adapter contract. +//! adapters shipped in the crate's `plugins/` directory ([`adapters`] says +//! where); see the [README](https://github.com/yologdev/yoagent/tree/main/integrations/yoagent-rutis) +//! and its adapter contract. With the `probe` feature, `probe` loads such a +//! package through its adapter and reports what an agent gets from it. //! //! # Host //! @@ -142,6 +143,7 @@ //! - **Route rutis's `ErrorSink`** (default: stderr) into your logging with //! `Ctx::root_with_sink`. +pub mod adapters; pub mod bridge; pub mod events; pub mod extension; @@ -166,6 +168,9 @@ mod host; )] pub mod languages; pub mod plugin; +#[cfg(feature = "probe")] +#[cfg_attr(docsrs, doc(cfg(feature = "probe")))] +pub mod probe; pub mod registry; pub use bridge::RutisBridge; diff --git a/integrations/yoagent-rutis/src/probe.rs b/integrations/yoagent-rutis/src/probe.rs new file mode 100644 index 00000000..f147911c --- /dev/null +++ b/integrations/yoagent-rutis/src/probe.rs @@ -0,0 +1,901 @@ +//! Loads one third-party plugin package through its adapter, as an agent +//! would, and reports what yoagent gets from it — without calling a tool or +//! asking a model. The compatibility matrix (`scripts/ecosystem-matrix.mjs`, +//! through the `ecosystem_probe` example) and hosts that vet a plugin before +//! using it (`yobuild check`) share it. +//! +//! - [`pi`]: pi extension files (as pi's package manager resolved them) +//! through the pi adapter. A tool the adapter refuses because it is named +//! like a yoagent built-in is tried again with that built-in left out +//! (`withoutBuiltins`), as a host would. Skills are loaded with yoagent's +//! own skills loader. +//! - [`dsh`]: DSH plugin rows over DSH's base packages and the tools +//! adapter, in a Node runtime whose `package.json` is the base folder's. +//! Reports what the rows add to the base — tools and prompt guidance — and +//! which of the services the rows' plugins inject exist (a plugin missing +//! one never starts, though its row reads as started to the host). +//! +//! **The adapters, installed.** Both take `plugins`: a folder laid out like +//! [`adapters`](crate::adapters)' `root` (the pi and DSH adapters beside +//! `yoagent.d.ts`, which they import), with `npm ci` run in its `pi/` and +//! `dsh/`. Copy `root` there — never install into the crate's own files, in +//! cargo's registry — or, for a path dependency, pass `root` itself. +//! +//! **A verdict.** Each report's `problem()` says what kept the package from +//! giving an agent what it offers, if anything; `None` is a clean load. +//! +//! **One at a time.** Probes in one process run one after another (a second +//! waits): [`Logs`] is one buffer for the process, and each report carries +//! the lines of its own load. Loading a package runs its code, as its own +//! host would. + +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use rutis::{Ctx, FiberState}; +use rutis_bridge::runtime::{Launcher, LocalRuntime}; +use rutis_loader::{ + Chain, EntryStatus, Layer, Loader, LoaderOptions, LoaderPlugin, Patch, RuntimeResolver, + RuntimeRowsPlugin, ServiceCatalog, +}; +use serde::Serialize; +use serde_json::{json, Value}; +use tokio_util::sync::CancellationToken; +use yoagent::extension::{RunContext, RunEnd, RunOutcome, TurnDecision}; +use yoagent::{Extension, Message, SkillSet, TurnContext}; + +use crate::RutisBridge; + +type BoxError = Box; + +/// How long rows may take to start, and the adapter to register. +const SETTLE: Duration = Duration::from_secs(60); + +/// Probes in this process, one at a time (their logs share [`Logs`]). +static ONE_AT_A_TIME: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(()); + +/// The `tracing` targets [`Logs`] collects: what plugins log, and the +/// bridge's own reports of a hook that failed. +const LOG_TARGETS: [&str; 2] = ["yoagent_rutis::plugin", "yoagent_rutis::extension"]; + +/// One line a plugin (or the bridge, about one) logged. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[non_exhaustive] +pub struct LogLine { + /// `error`, `warn`, `info`, `debug` or `trace`. + pub level: String, + pub message: String, +} + +/// Collects what plugins log (target `yoagent_rutis::plugin`) and the +/// bridge's reports of their failed hooks (`yoagent_rutis::extension`), as +/// a `tracing` layer. Install it globally (the runtime logs from its own +/// threads) and pass it to [`pi`] or [`dsh`]: each report then carries the +/// lines of its own load. +#[derive(Debug, Clone, Default)] +pub struct Logs(Arc>>); + +impl Logs { + fn clear(&self) { + self.0.lock().unwrap().clear(); + } + + fn lines(&self) -> Vec { + self.0.lock().unwrap().clone() + } +} + +impl tracing_subscriber::Layer for Logs { + fn on_event(&self, event: &tracing::Event<'_>, _: tracing_subscriber::layer::Context<'_, S>) { + if !LOG_TARGETS.contains(&event.metadata().target()) { + return; + } + #[derive(Default)] + struct Text(String); + impl tracing::field::Visit for Text { + fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) { + if field.name() == "message" { + self.0 = format!("{value:?}"); + } + } + } + let mut text = Text::default(); + event.record(&mut text); + self.0.lock().unwrap().push(LogLine { + level: event.metadata().level().to_string().to_lowercase(), + message: text.0, + }); + } +} + +/// A row's state once settled. +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[non_exhaustive] +pub struct Row { + pub id: String, + /// `active`, `failed`, `unresolved`, `waiting` (still pending after the + /// wait: a dependency never came), or another rutis state. + pub state: String, + pub error: Option, +} + +/// Skills, loaded with yoagent's skills loader. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)] +#[non_exhaustive] +pub struct Skills { + /// How many were given. + pub found: usize, + /// The names of those that loaded. + pub loaded: Vec, + /// Why the others did not. + pub errors: Vec, +} + +/// What the pi adapter made of extension files. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)] +#[non_exhaustive] +pub struct PiReport { + pub skills: Skills, + /// Whether the adapter loaded the extensions and a run would get them; + /// `None` when there were none. + pub loaded: Option, + /// Why not: the load's error, or the adapter's refusal at the first + /// model request (a handler it cannot honour, found after load). + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, + pub rows: Vec, + /// The yoagent built-ins left out because an extension's tool has + /// their name. + pub without: Vec, + /// The tools a run is offered. + pub tools: Vec, + /// The length of the note the first request carries (prompt additions). + pub note_chars: usize, + /// What the plugins logged during the load. + pub logs: Vec, +} + +impl PiReport { + /// What kept the extensions from loading, if anything: `None` when they + /// loaded (or there were none). + pub fn problem(&self) -> Option { + match self.loaded { + Some(false) => Some( + self.error + .clone() + .unwrap_or_else(|| "not loaded, no error given".into()), + ), + _ => None, + } + } +} + +/// What DSH plugin rows add to DSH's base. +#[derive(Debug, Clone, Default, PartialEq, Serialize)] +#[non_exhaustive] +pub struct DshReport { + /// Which services the plugin's rows inject, as the runtime saw it: + /// `{ "present": [...], "missing": [...], "importErrors": {...} }` + /// (a row whose module could not be imported: what it injects is + /// unknown), or `{ "error": … }`; `null` when the probe did not get + /// that far (see `error`). + pub services: Value, + /// The plugin's rows (not the base's). + pub rows: Vec, + /// The tools the rows add. + pub tools: Vec, + /// The length of the prompt guidance the rows add. + pub guidance_chars: usize, + /// Why the probe could not tell. + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, + pub logs: Vec, +} + +impl DshReport { + /// What kept the plugin from giving an agent what it offers, if + /// anything: the probe's error, a service check that did not report, a + /// service the plugin needs and the host lacks, or a row not active. + /// `None` is a clean load. + pub fn problem(&self) -> Option { + if let Some(error) = &self.error { + return Some(error.clone()); + } + if let Some(error) = self.services.get("error") { + return Some(format!("the service check: {error}")); + } + let missing = self.services["missing"] + .as_array() + .cloned() + .unwrap_or_default(); + if !missing.is_empty() { + return Some(format!("needs services the host lacks: {}", json!(missing))); + } + if self.services["missing"].is_null() { + return Some("the service check did not report".into()); + } + if let Some(errors) = self.services["importErrors"].as_object() { + if !errors.is_empty() { + return Some(format!( + "could not read what these rows inject: {}", + Value::Object(errors.clone()) + )); + } + } + self.rows + .iter() + .find(|r| r.state != "active") + .map(|r| match &r.error { + Some(e) => format!("row {} {}: {e}", r.id, r.state), + None => format!("row {} {}", r.id, r.state), + }) + } +} + +struct Host { + root: Ctx, + bridge: RutisBridge, + loader: Loader, +} + +/// A rutis root with the bridge and one Node runtime (`runtime`, the +/// `@arcships/rutis-runtime` package) whose `package.json` is +/// `package_json`, where row names resolve; its process given `env`. +async fn host( + runtime: &Path, + package_json: &Path, + env: &[(&str, &Path)], +) -> Result { + let root = Ctx::root()?; + let bridge = RutisBridge::install(&root)?; + let mut catalog = ServiceCatalog::new(); + catalog.register_shared("yoagent"); + let launcher = env + .iter() + .fold(Launcher::node(runtime), |launcher, (name, value)| { + launcher.env(*name, *value) + }); + let node = LocalRuntime::launcher("node", launcher, package_json); + let resolver = Arc::new(RuntimeResolver::node(node.handle()).with_catalog(&catalog)); + root.plugin(node); + let plugin = LoaderPlugin::new( + Chain::new().with_shared(resolver.clone()), + LoaderOptions { + catalog, + ..LoaderOptions::default() + }, + ); + let loader = plugin.handle(); + root.plugin(plugin).await?; + root.plugin(RuntimeRowsPlugin::new(resolver)); + Ok(Host { + root, + bridge, + loader, + }) +} + +fn row(loader: &Loader, id: &str) -> Row { + let (state, error) = match loader.get(id).map(|info| info.status) { + Some(EntryStatus::Running(s)) => match s.state { + FiberState::Active => ("active".into(), None), + FiberState::Failed => ("failed".into(), s.error.map(|e| e.to_string())), + FiberState::Pending => ("waiting".into(), None), + other => (format!("{other:?}").to_lowercase(), None), + }, + Some(EntryStatus::Unresolved(e)) => ("unresolved".into(), Some(e.to_string())), + Some(other) => (format!("{other:?}").to_lowercase(), None), + None => ("missing".into(), None), + }; + Row { + id: id.to_owned(), + state, + error, + } +} + +/// The ids of `rows`: an array of rows, each with a string `id` (a row +/// without one would load but could be neither waited for nor reported). +fn row_ids(rows: &Value) -> Result, BoxError> { + let rows = rows.as_array().ok_or("rows: an array of rows")?; + rows.iter() + .map(|row| { + row["id"] + .as_str() + .map(str::to_owned) + .ok_or_else(|| format!("a row without a string id: {row}").into()) + }) + .collect() +} + +/// Load the given layers in one reconcile and wait until each row is +/// active or failed (or the wait ends); each row's state. +async fn load(host: &Host, layers: Vec<(&str, Value)>) -> Result, BoxError> { + let mut ids = Vec::new(); + let mut built = Vec::new(); + for (name, rows) in layers { + ids.extend(row_ids(&rows)?); + let patches: Vec = serde_json::from_value(json!([{ "insert": rows }]))?; + built.push(Layer::new(name, patches)); + } + host.loader.reconcile(built, None).await?; + // Rows still pending when the wait ends are reported as `waiting`. + let settled = |id: &String| { + matches!( + row(&host.loader, id).state.as_str(), + "active" | "failed" | "unresolved" + ) + }; + let _ = tokio::time::timeout(SETTLE, async { + while !ids.iter().all(settled) { + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await; + Ok(ids.iter().map(|id| row(&host.loader, id)).collect()) +} + +/// Wait until a handler named `name` is registered. +async fn handler(host: &Host, name: &str) -> bool { + tokio::time::timeout(SETTLE, async { + while !host + .bridge + .registry() + .handlers() + .iter() + .any(|h| h.name() == name) + { + tokio::time::sleep(Duration::from_millis(50)).await; + } + }) + .await + .is_ok() +} + +/// What a run gets: the tools it is offered, and what its first model +/// request meets. +struct Offered { + tools: Vec, + /// The note the request carries. + note: Option, + /// A run stopped (or failed) before its first request: why. + stopped: Option, +} + +async fn offered(host: &Host) -> Result { + let cancel = CancellationToken::new(); + let prompt = [Message::user("hello")]; + let run = RunContext::new("probe", &prompt, &cancel); + let mut hooks = host.bridge.extension().start_run(&run).await?; + let mut tools: Vec = hooks + .tools(&run) + .await + .iter() + .map(|t| t.name().to_string()) + .collect(); + tools.sort(); + let turn = TurnContext::new("", &prompt, &[], "probe").with_run_prompts(&prompt); + let (note, stopped) = match hooks.before_model(&turn).await { + TurnDecision::Note(note) => (Some(note), None), + TurnDecision::Stop(why) | TurnDecision::Fail(why) => (None, Some(why)), + _ => (None, None), + }; + hooks + .finish(&RunOutcome::new(RunEnd::Completed, None)) + .await; + Ok(Offered { + tools, + note, + stopped, + }) +} + +async fn shutdown(host: Host) { + let _ = tokio::time::timeout(Duration::from_secs(10), host.root.shutdown()).await; +} + +/// `SKILL.md` paths or skill directories, loaded with yoagent's loader. +fn skills(paths: &[PathBuf]) -> Skills { + let mut out = Skills { + found: paths.len(), + ..Skills::default() + }; + for path in paths { + let dir = if path.ends_with("SKILL.md") { + path.parent().unwrap_or(path) + } else { + path + }; + // A bare `SKILL.md`: the current folder. + let dir = if dir.as_os_str().is_empty() { + Path::new(".") + } else { + dir + }; + // The loader keeps canonical paths (on macOS /var is /private/var). + let dir = std::fs::canonicalize(dir).unwrap_or_else(|_| dir.to_path_buf()); + let Some(parent) = dir.parent() else { + out.errors + .push(format!("{}: not a skill folder", dir.display())); + continue; + }; + let (set, errs) = SkillSet::load_dir_resilient(parent, "probe"); + match set.skills().iter().find(|s| s.base_dir == dir) { + Some(skill) => out.loaded.push(skill.name.clone()), + None => { + // The loader's error about this skill, not a sibling's. + let name = dir.file_name().map(|n| n.to_string_lossy().into_owned()); + let error = errs + .iter() + .map(|e| e.to_string()) + .find(|e| name.as_deref().is_some_and(|n| e.contains(n))); + out.errors.push(format!( + "{}: {}", + dir.display(), + error.unwrap_or_else(|| "not loaded, no error given".into()) + )); + } + } + } + out +} + +/// The built-in yoagent tool's name a refusal names, if it is a same-name +/// refusal (the pi adapter's wording: `named like yoagent's built-in "…"`). +fn shadowed(error: &str) -> Option { + let rest = error.split("named like yoagent's built-in \"").nth(1)?; + Some(rest.split('"').next()?.to_string()) +} + +/// Loads pi `extensions` (files, as pi's package manager resolved them) with +/// `cwd` as their working folder, and `skills` (`SKILL.md` paths or skill +/// folders) with yoagent's loader. `plugins`: the adapters, installed (see +/// the module's doc). +pub async fn pi( + plugins: &Path, + cwd: &Path, + extensions: &[String], + skill_paths: &[PathBuf], + logs: Option<&Logs>, +) -> PiReport { + let _one = ONE_AT_A_TIME.lock().await; + let mut report = PiReport { + skills: skills(skill_paths), + ..PiReport::default() + }; + if extensions.is_empty() { + return report; + } + pi_load(&mut report, &plugins.join("pi"), cwd, extensions, logs).await; + report.logs = logs.map(Logs::lines).unwrap_or_default(); + report +} + +async fn pi_load( + report: &mut PiReport, + pi_dir: &Path, + cwd: &Path, + extensions: &[String], + logs: Option<&Logs>, +) { + let runtime = pi_dir.join("node_modules/@arcships/rutis-runtime"); + let mut without: Vec = Vec::new(); + // A same-name refusal (at load, or at the first request for a tool + // added later) is retried with that built-in left out, a few times. + for _ in 0..6 { + if let Some(logs) = logs { + logs.clear(); + } + report.without = without.clone(); + let host = match host(&runtime, &pi_dir.join("package.json"), &[]).await { + Ok(host) => host, + Err(e) => { + report.loaded = Some(false); + report.error = Some(format!("host: {e}")); + return; + } + }; + let rows = json!([{ + "id": "pi", + "name": pi_dir.join("pi-extensions-adapter.ts"), + "config": { "extensions": extensions, "cwd": cwd, "withoutBuiltins": without }, + }]); + let rows = match load(&host, vec![("probe", rows)]).await { + Ok(rows) => rows, + Err(e) => { + report.loaded = Some(false); + report.error = Some(e.to_string()); + shutdown(host).await; + return; + } + }; + // A TS row reads active before its async `apply` ends: wait for the + // adapter's handler, then read the row again (a refusal fails it late). + let registered = rows[0].state == "active" && handler(&host, "pi-extensions").await; + let pi_row = row(&host.loader, "pi"); + let run = match (registered && pi_row.state == "active").then_some(()) { + Some(()) => Some(offered(&host).await), + None => None, + }; + // Retried: a same-name refusal at load, or at the first request. + let refusal = pi_row.error.clone().or_else(|| match &run { + Some(Ok(offered)) => offered.stopped.clone(), + _ => None, + }); + if let Some(name) = refusal.as_deref().and_then(shadowed) { + if !without.contains(&name) { + without.push(name); + shutdown(host).await; + continue; + } + } + match run { + Some(Ok(offered)) => { + report.tools = offered.tools; + report.note_chars = offered.note.map_or(0, |n| n.len()); + match offered.stopped { + // The adapter refused after load: a run would stop there. + Some(why) => { + report.loaded = Some(false); + report.error = Some(why); + } + None => report.loaded = Some(true), + } + } + Some(Err(e)) => { + report.loaded = Some(false); + report.error = Some(format!("starting a run: {e}")); + } + None => { + report.loaded = Some(false); + report.error = Some(pi_row.error.clone().unwrap_or_else(|| { + format!("the adapter did not register (row {})", pi_row.state) + })); + } + } + report.rows = vec![pi_row]; + // Let late log lines arrive (a warning decides "works, partly"), + // then again after shutdown, which flushes the runtime. + tokio::time::sleep(Duration::from_millis(500)).await; + shutdown(host).await; + return; + } + report.loaded = Some(false); + report.error = Some("too many same-name tools".into()); +} + +/// The row id of `dsh/probe-services.ts`; a failure of it is reported under +/// `services.error`. +const SERVICE_CHECK: &str = "probe-services"; + +/// A file of the DSH adapter folder, as loaded from `base`: it must resolve +/// the base's Cordis (a copy elsewhere resolves another one, and the runtime +/// refuses it), so it is copied into `base` when not there yet. +fn dsh_file(dsh_dir: &Path, base: &Path, name: &str) -> Result { + let copy = base.join(name); + if !copy.exists() { + std::fs::copy(dsh_dir.join(name), ©) + .map_err(|e| format!("copying {name} into {}: {e}", base.display()))?; + } + Ok(copy) +} + +/// DSH's base packages and the tools adapter: what every DSH plugin gets. +fn dsh_base_rows(dsh_dir: &Path, base: &Path) -> Result { + // The services a headless DSH host runs (fs, attachments, credentials, + // storage, session persistence, skills): DSH's own packages, listed in + // `host-services.json` for every host to load. + let path = dsh_dir.join("host-services.json"); + let text = std::fs::read_to_string(&path).map_err(|e| format!("{}: {e}", path.display()))?; + let mut rows: Vec = + serde_json::from_str(&text).map_err(|e| format!("{}: {e}", path.display()))?; + rows.extend([ + json!({ "id": "base-web", "name": "@deepseek-ai/dsh-web" }), + json!({ "id": "base-system-prompt", "name": "@deepseek-ai/dsh-system-prompt" }), + json!({ "id": "base-tools", "name": "@deepseek-ai/dsh-tools" }), + json!({ "id": "base-user-questions", "name": "@deepseek-ai/dsh-user-questions" }), + json!({ "id": "base-adapter", "name": dsh_file(dsh_dir, base, "dsh-tools-adapter.ts")? }), + ]); + Ok(Value::Array(rows)) +} + +/// Loads DSH plugin `rows` (its bundle patch, or one row naming the +/// package; an array of rows, each with a string `id`) over DSH's base +/// packages and the tools adapter, in a Node runtime whose `package.json` +/// is `base`'s (`base` holds DSH's base packages and the plugin), and checks +/// which of the services the rows' plugins inject exist — read from their +/// modules inside the runtime (every key: Cordis requires them all), plus any +/// named in `inject`. The probe's two DSH files (`dsh-tools-adapter.ts`, +/// `probe-services.ts`) must resolve `base`'s Cordis, so they are copied +/// into `base` when they are not there. `plugins`: the adapters, installed (see the module's doc). +/// Each probe gets a DSH home of its own: `~/.dsh` and `~/.agents` are not +/// handed to the plugins (the process environment still is). +pub async fn dsh( + plugins: &Path, + base: &Path, + rows: &Value, + inject: &[String], + logs: Option<&Logs>, +) -> DshReport { + let _one = ONE_AT_A_TIME.lock().await; + if let Some(logs) = logs { + logs.clear(); + } + let mut report = match dsh_load(&plugins.join("dsh"), base, rows, inject).await { + Ok(report) => report, + Err(e) => DshReport { + error: Some(e.to_string()), + ..DshReport::default() + }, + }; + report.logs = logs.map(Logs::lines).unwrap_or_default(); + report +} + +async fn dsh_load( + dsh_dir: &Path, + base: &Path, + rows: &Value, + inject: &[String], +) -> Result { + let plugin_ids = row_ids(rows)?; + let mut report = DshReport::default(); + let runtime = base.join("node_modules/@arcships/rutis-runtime"); + let dsh_home = tempfile::tempdir()?; + let agents_home = dsh_home.path().join("agents"); + let base_rows = dsh_base_rows(dsh_dir, base)?; + let host = host( + &runtime, + &base.join("package.json"), + &[ + ("DSH_HOME", dsh_home.path()), + ("DSH_AGENTS_HOME", &agents_home), + ], + ) + .await + .map_err(|e| format!("host: {e}"))?; + // The base alone first, then the plugin's rows on top: the difference is + // what the plugin adds. + let base_state = load(&host, vec![("base", base_rows.clone())]).await; + let ready = matches!(&base_state, Ok(rows) if rows.iter().all(|r| r.state == "active")) + && handler(&host, "dsh-tools").await; + if !ready { + shutdown(host).await; + return Err(format!("the DSH base did not start: {base_state:?}").into()); + } + let before = offered(&host).await; + // The plugin's rows, and a check (from inside the runtime) of which of + // the services they inject exist: without one a plugin never starts. + let out = dsh_home.path().join("probe-services.json"); + let modules: Vec<&Value> = rows + .as_array() + .into_iter() + .flatten() + .filter_map(|row| row.get("name").filter(|n| n.is_string())) + .collect(); + let mut plugin_rows = rows.as_array().cloned().unwrap_or_default(); + plugin_rows.push(json!({ + "id": SERVICE_CHECK, + "name": dsh_file(dsh_dir, base, "probe-services.ts")?, + "config": { "names": inject, "modules": modules, "base": base, "out": out, "waitMs": 3000 }, + })); + let state = load( + &host, + vec![("base", base_rows), ("plugin", json!(plugin_rows))], + ) + .await; + report.services = tokio::time::timeout(Duration::from_secs(20), async { + loop { + if let Some(found) = std::fs::read_to_string(&out) + .ok() + .and_then(|text| serde_json::from_str::(&text).ok()) + { + return found; + } + tokio::time::sleep(Duration::from_millis(100)).await; + } + }) + .await + .unwrap_or_else(|_| { + let check = row(&host.loader, SERVICE_CHECK); + let error = check.error.map(|e| format!(": {e}")).unwrap_or_default(); + json!({ "error": format!("no report within 20 s (row {}{error})", check.state) }) + }); + let after = offered(&host).await; + match (state, before, after) { + (Ok(state), Ok(before), Ok(after)) => { + report.rows = state + .into_iter() + .filter(|r| plugin_ids.contains(&r.id)) + .collect(); + report.tools = after + .tools + .into_iter() + .filter(|t| !before.tools.contains(t)) + .collect(); + let (before, after) = ( + before.note.unwrap_or_default(), + after.note.unwrap_or_default(), + ); + report.guidance_chars = after + .lines() + .filter(|line| !line.trim().is_empty() && !before.contains(line)) + .map(|line| line.len() + 1) + .sum(); + } + (state, before, after) => { + report.error = Some(format!( + "{:?} {:?} {:?}", + state.err(), + before.err(), + after.err() + )); + } + } + tokio::time::sleep(Duration::from_millis(200)).await; + shutdown(host).await; + Ok(report) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn keys(value: &Value) -> Vec<&str> { + let mut keys: Vec<&str> = value + .as_object() + .unwrap() + .keys() + .map(String::as_str) + .collect(); + keys.sort(); + keys + } + + #[test] + fn the_reports_serialize_to_the_json_the_matrix_reads() { + let pi = PiReport { + rows: vec![Row { + id: "pi".into(), + state: "active".into(), + error: None, + }], + logs: vec![LogLine { + level: "warn".into(), + message: "m".into(), + }], + ..PiReport::default() + }; + let json = serde_json::to_value(&pi).unwrap(); + assert_eq!( + keys(&json), + [ + "loaded", + "logs", + "note_chars", + "rows", + "skills", + "tools", + "without" + ] + ); + assert_eq!(keys(&json["skills"]), ["errors", "found", "loaded"]); + assert_eq!(keys(&json["rows"][0]), ["error", "id", "state"]); + assert_eq!(keys(&json["logs"][0]), ["level", "message"]); + // `error` only when there is one. + let failed = PiReport { + error: Some("x".into()), + ..PiReport::default() + }; + assert_eq!(serde_json::to_value(&failed).unwrap()["error"], "x"); + + let dsh = DshReport { + services: json!({ "present": [], "missing": [] }), + ..DshReport::default() + }; + let json = serde_json::to_value(&dsh).unwrap(); + assert_eq!( + keys(&json), + ["guidance_chars", "logs", "rows", "services", "tools"] + ); + assert_eq!(keys(&json["services"]), ["missing", "present"]); + let json = serde_json::to_value(DshReport::default()).unwrap(); + assert!(json["services"].is_null()); + } + + #[test] + fn a_same_name_refusal_names_the_built_in() { + // The pi adapter's wording (pi-extensions-adapter.ts, checkShadowing). + let error = "the pi extensions adapter stopped: a pi extension's tool \"read_file\" is \ + named like yoagent's built-in \"read_file\"; if the host installs that one \ + too, it runs instead of the pi tool."; + assert_eq!(shadowed(error).as_deref(), Some("read_file")); + assert_eq!(shadowed("something else"), None); + } + + #[test] + fn a_verdict_from_each_report() { + let clean = DshReport { + services: json!({ "present": ["fs"], "missing": [] }), + rows: vec![Row { + id: "fx".into(), + state: "active".into(), + error: None, + }], + ..DshReport::default() + }; + assert_eq!(clean.problem(), None); + let missing = DshReport { + services: json!({ "missing": ["db"] }), + ..clean.clone() + }; + assert_eq!( + missing.problem().as_deref(), + Some("needs services the host lacks: [\"db\"]") + ); + let waiting = DshReport { + rows: vec![Row { + id: "fx".into(), + state: "waiting".into(), + error: None, + }], + ..clean.clone() + }; + assert_eq!(waiting.problem().as_deref(), Some("row fx waiting")); + let unchecked = DshReport { + services: json!({ "error": "no report" }), + ..clean.clone() + }; + assert!(unchecked + .problem() + .unwrap() + .starts_with("the service check")); + assert_eq!( + DshReport::default().problem().as_deref(), + Some("the service check did not report") + ); + let unread = DshReport { + services: json!({ "missing": [], "importErrors": { "fx": "cannot import" } }), + ..clean.clone() + }; + assert!(unread + .problem() + .unwrap() + .contains("could not read what these rows inject")); + + let pi = PiReport { + loaded: Some(false), + error: Some("refused".into()), + ..PiReport::default() + }; + assert_eq!(pi.problem().as_deref(), Some("refused")); + assert_eq!(PiReport::default().problem(), None); + } + + #[test] + fn rows_need_an_array_and_ids() { + assert!(row_ids(&json!({ "id": "x" })).is_err()); + assert!(row_ids(&json!([{ "name": "x" }])).is_err()); + assert_eq!(row_ids(&json!([{ "id": "a" }])).unwrap(), ["a"]); + } + + #[test] + fn skills_from_folders_and_files() { + let dir = tempfile::tempdir().unwrap(); + let skill = dir.path().join("greet"); + std::fs::create_dir(&skill).unwrap(); + std::fs::write( + skill.join("SKILL.md"), + "---\nname: greet\ndescription: Says hello.\n---\nSay hello.\n", + ) + .unwrap(); + let empty = dir.path().join("empty"); + std::fs::create_dir(&empty).unwrap(); + let out = skills(&[skill.join("SKILL.md"), skill.clone(), empty.clone()]); + assert_eq!(out.found, 3); + assert_eq!(out.loaded, ["greet", "greet"]); + assert_eq!(out.errors.len(), 1, "{out:?}"); + assert!(out.errors[0].contains("empty"), "{out:?}"); + } +} diff --git a/integrations/yoagent-rutis/tests/probe_test.rs b/integrations/yoagent-rutis/tests/probe_test.rs new file mode 100644 index 00000000..c3ae2993 --- /dev/null +++ b/integrations/yoagent-rutis/tests/probe_test.rs @@ -0,0 +1,238 @@ +//! The probe (`yoagent_rutis::probe`, feature `probe`) on this crate's own +//! fixtures: what the pi and DSH adapters give an agent, and what plugins +//! log. Needs `npm ci` in `plugins/pi/` and `plugins/dsh/`; skipped +//! without, unless `YOAGENT_RUTIS_REQUIRE_RUNTIMES=1`. + +use std::path::Path; +use std::sync::OnceLock; + +use serde_json::json; +use tracing_subscriber::layer::SubscriberExt; +use yoagent_rutis::adapters::adapters; +use yoagent_rutis::probe::{self, Logs}; + +/// Installed once for the test binary: the runtime logs from its own threads. +fn logs() -> &'static Logs { + static LOGS: OnceLock = OnceLock::new(); + LOGS.get_or_init(|| { + let logs = Logs::default(); + tracing::subscriber::set_global_default(tracing_subscriber::registry().with(logs.clone())) + .unwrap(); + logs + }) +} + +fn installed(dir: &Path) -> bool { + if dir + .join("node_modules/@arcships/rutis-runtime/package.json") + .exists() + { + return true; + } + if std::env::var_os("YOAGENT_RUTIS_REQUIRE_RUNTIMES").is_some_and(|v| v == "1") { + panic!( + "the runtime is required but unavailable: run `npm ci` in {}", + dir.display() + ); + } + eprintln!("SKIPPED: run `npm ci` in {}", dir.display()); + false +} + +#[tokio::test(flavor = "multi_thread")] +async fn pi_extensions_are_loaded_and_their_tools_listed() { + let pi = adapters().pi; + if !installed(&pi) { + return; + } + let cwd = tempfile::tempdir().unwrap(); + let extension = pi.join("fixture-extension.ts").display().to_string(); + let report = probe::pi( + &adapters().root, + cwd.path(), + &[extension], + &[], + Some(logs()), + ) + .await; + assert_eq!(report.loaded, Some(true), "{report:?}"); + assert!(report.tools.iter().any(|t| t == "pi_echo"), "{report:?}"); + assert_eq!(report.rows.len(), 1); + assert_eq!(report.rows[0].state, "active"); + assert_eq!(report.problem(), None); + // The adapter's load line, which the matrix reads the events from. + assert!( + report.logs.iter().any(|l| l.level == "debug" + && l.message.contains("[pi] loaded") + && l.message.contains("events")), + "{:?}", + report.logs + ); + + // No extensions: only skills (none here), nothing loaded. + let report = probe::pi(&adapters().root, cwd.path(), &[], &[], None).await; + assert_eq!(report.loaded, None); + assert_eq!(report.skills.found, 0); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_pi_extension_that_does_not_load_says_why() { + let pi = adapters().pi; + if !installed(&pi) { + return; + } + let cwd = tempfile::tempdir().unwrap(); + let missing = cwd + .path() + .join("no-such-extension.ts") + .display() + .to_string(); + let report = probe::pi(&adapters().root, cwd.path(), &[missing], &[], None).await; + assert_eq!(report.loaded, Some(false), "{report:?}"); + assert!(report.error.is_some(), "{report:?}"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn dsh_rows_report_the_tools_they_add_and_the_services_they_need() { + let dsh = adapters().dsh; + if !installed(&dsh) { + return; + } + let rows = json!([{ "id": "fx", "name": dsh.join("fixture-tools.ts") }]); + let report = probe::dsh(&adapters().root, &dsh, &rows, &[], None).await; + assert_eq!(report.error, None, "{report:?}"); + assert!( + report.tools.iter().any(|t| t == "fixture_echo"), + "{report:?}" + ); + assert_eq!(report.rows.len(), 1); + assert_eq!(report.services["missing"], json!([]), "{report:?}"); + assert_eq!(report.problem(), None, "{report:?}"); + + // A service nobody provides is reported missing. + let report = probe::dsh( + &adapters().root, + &dsh, + &rows, + &["no-such-service".into()], + None, + ) + .await; + assert_eq!( + report.services["missing"], + json!(["no-such-service"]), + "{report:?}" + ); + assert!(report.problem().unwrap().contains("no-such-service")); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_dsh_plugin_missing_a_service_it_injects_is_not_clean() { + let dsh = adapters().dsh; + if !installed(&dsh) { + return; + } + // Its module injects a service no host provides: it never starts, though + // its row reads as started. `inject` is empty: the probe reads what the + // module injects itself. + let rows = json!([{ "id": "needs", "name": dsh.join("fixture-missing-service.ts") }]); + let report = probe::dsh(&adapters().root, &dsh, &rows, &[], None).await; + assert_eq!( + report.services["missing"], + json!(["yoagentNoSuchService"]), + "{report:?}" + ); + assert!(report.problem().is_some(), "{report:?}"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn dsh_rows_must_be_an_array_with_ids() { + let dsh = adapters().dsh; + let report = probe::dsh(&adapters().root, &dsh, &json!({ "id": "x" }), &[], None).await; + assert!(report.error.is_some(), "{report:?}"); + let report = probe::dsh(&adapters().root, &dsh, &json!([{ "name": "x" }]), &[], None).await; + assert!(report.error.is_some(), "{report:?}"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_pi_extension_that_refuses_after_load_is_not_clean() { + let pi = adapters().pi; + if !installed(&pi) { + return; + } + let cwd = tempfile::tempdir().unwrap(); + let extension = pi.join("fixture-late-refusal.ts").display().to_string(); + let report = probe::pi( + &adapters().root, + cwd.path(), + &[extension], + &[], + Some(logs()), + ) + .await; + assert_eq!(report.loaded, Some(false), "{report:?}"); + let error = report.error.as_deref().unwrap_or_default(); + assert!(error.contains("adapter stopped"), "{report:?}"); + assert!(report.problem().is_some()); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_tool_named_like_a_built_in_is_retried_without_it() { + let pi = adapters().pi; + if !installed(&pi) { + return; + } + let cwd = tempfile::tempdir().unwrap(); + let extension = pi.join("fixture-shadow.ts").display().to_string(); + let report = probe::pi(&adapters().root, cwd.path(), &[extension], &[], None).await; + assert_eq!(report.loaded, Some(true), "{report:?}"); + assert_eq!(report.without, ["read_file"], "{report:?}"); + assert!(report.tools.iter().any(|t| t == "read_file"), "{report:?}"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_service_declared_required_false_is_still_required() { + let dsh = adapters().dsh; + if !installed(&dsh) { + return; + } + let rows = json!([{ "id": "opt", "name": dsh.join("fixture-optional-service.ts") }]); + let report = probe::dsh(&adapters().root, &dsh, &rows, &[], None).await; + assert_eq!( + report.services["missing"], + json!(["yoagentNoSuchOptional"]), + "{report:?}" + ); + assert!(report.problem().is_some(), "{report:?}"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_late_tool_named_like_a_built_in_is_retried_without_it() { + let pi = adapters().pi; + if !installed(&pi) { + return; + } + let cwd = tempfile::tempdir().unwrap(); + let extension = pi.join("fixture-late-shadow.ts").display().to_string(); + let report = probe::pi(&adapters().root, cwd.path(), &[extension], &[], None).await; + assert_eq!(report.loaded, Some(true), "{report:?}"); + assert_eq!(report.without, ["read_file"], "{report:?}"); +} + +#[tokio::test(flavor = "multi_thread")] +async fn a_base_of_its_own_gets_the_probe_files_copied_in() { + let dsh = adapters().dsh; + if !installed(&dsh) { + return; + } + // A base like a host's: DSH's packages and a package.json, without the + // probe's two files. Node follows the symlink, so one Cordis. + let base = tempfile::tempdir().unwrap(); + std::fs::copy(dsh.join("package.json"), base.path().join("package.json")).unwrap(); + std::os::unix::fs::symlink(dsh.join("node_modules"), base.path().join("node_modules")).unwrap(); + let rows = json!([{ "id": "fx", "name": dsh.join("fixture-tools.ts") }]); + let report = probe::dsh(&adapters().root, base.path(), &rows, &[], None).await; + assert_eq!(report.problem(), None, "{report:?}"); + assert!(base.path().join("probe-services.ts").is_file()); + assert!(base.path().join("dsh-tools-adapter.ts").is_file()); +} diff --git a/scripts/ecosystem-matrix.mjs b/scripts/ecosystem-matrix.mjs index 48b5b0c7..984c4bf6 100644 --- a/scripts/ecosystem-matrix.mjs +++ b/scripts/ecosystem-matrix.mjs @@ -79,7 +79,7 @@ async function top(keyword) { let probe if (!opts.fromJson) { log('building the probe…') -const build = run('cargo', ['build', '--quiet', '--manifest-path', join(rutis, 'Cargo.toml'), '--features', 'node', '--example', 'ecosystem_probe', '--message-format=json'], { timeout: 1_800_000 }) +const build = run('cargo', ['build', '--quiet', '--manifest-path', join(rutis, 'Cargo.toml'), '--features', 'probe', '--example', 'ecosystem_probe', '--message-format=json'], { timeout: 1_800_000 }) if (!build.ok) throw new Error(`cargo build failed:\n${tail(build.stderr, 4000)}`) probe = build.stdout.split('\n').filter(Boolean).map((l) => { try { return JSON.parse(l) } catch { return {} } }) .find((m) => m.reason === 'compiler-artifact' && m.target?.name === 'ecosystem_probe')?.executable @@ -154,6 +154,8 @@ function classifyPi(r) { const error = r.error ?? 'unknown' // Its skills do not go through the adapter: yoagent's own loader takes them. const usable = skills.length > 0 ? ` Its ${plural(skills.length, 'skill')} still load with yoagent's skills loader.` : '' + // Refused after load (the probe reports the stop a run would meet). + if (/adapter stopped/.test(error)) return { result: 'refused', detail: error.replace(/^.*?adapter stopped: /s, '') + usable } if (/never fires/.test(error)) return { result: 'refused', detail: error.replace(/^.*?Error: /s, '') + usable } // A throwing session_start refuses the load too (fail closed), typically // reaching for pi app internals the adapter's context does not have. @@ -313,7 +315,8 @@ async function runDsh(packages) { if (!info) { r = { inspectError: `could not read the package: ${tail(ins.stderr, 300)}` } } else { - const inject = (info.inject ?? []).filter((n) => !n.endsWith('?')) + // Every key is required in Cordis, `required: false` included. + const inject = info.inject ?? [] r = { ...probeJson(['dsh', '--base', dir, '--rows', JSON.stringify(info.rows), ...inject.flatMap((n) => ['--inject', n])]), client: info.client, peers: info.peers, inject: info.inject, importErrors: info.importErrors } } if (notes.length > 0) r.installNote = notes.join('; ')