From 49e70e481d8014c3e6224d7825437c7299f73fc4 Mon Sep 17 00:00:00 2001 From: Bryan Helmkamp Date: Fri, 18 Sep 2026 01:52:46 -0400 Subject: [PATCH] Register Fabro's run tools on a Petri run through the host tool capability Plan item F3.4. `fabro_petri::host_tools` adapts Petri's `HostTools` capability to `register_fabro_run_tools`: every native agent session of a run gets the tools the legacy worker registers, bound to the worker's client and the run id, so a child run a stage creates is parented to the Petri run. The tools run under the run's tool hooks, are recorded under the stage, and reach sub-agents through Pebble's inheritance. `RuntimeSpec::run_tools` installs the capability; the worker sets it when the run's settings enable `[run.agent] fabro_tools` and the worker token carries `agent:run_tools`, the legacy worker's gate. The server's in-process test path runs without them, like the legacy one. The identity the tools need is the run id alone; no run tool records a stage on an effect, so nothing derives Fabro's `node@visit` label. A context for another run gets no tools. Co-Authored-By: Claude Fable 5.1 --- Cargo.lock | 5 + .../src/commands/run/petri_worker.rs | 48 ++- lib/apps/fabro-cli/src/commands/run/runner.rs | 4 +- .../fabro-server/src/server/petri_runs.rs | 3 + lib/components/fabro-petri/Cargo.toml | 6 + lib/components/fabro-petri/src/host_tools.rs | 167 ++++++++++ lib/components/fabro-petri/src/lib.rs | 5 +- lib/components/fabro-petri/src/runtime.rs | 17 +- .../fabro-petri/tests/host_tools.rs | 300 ++++++++++++++++++ 9 files changed, 546 insertions(+), 9 deletions(-) create mode 100644 lib/components/fabro-petri/src/host_tools.rs create mode 100644 lib/components/fabro-petri/tests/host_tools.rs diff --git a/Cargo.lock b/Cargo.lock index 2ab668dd9..3ad06b72d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2897,9 +2897,14 @@ dependencies = [ "fabro-db", "fabro-http", "fabro-llm", + "fabro-petri", "fabro-store", + "fabro-tool", "fabro-types", + "fabro-workflow", + "httpmock", "lithos-llm", + "pebble-coding-agent", "petri-attractor-steps", "petri-execution", "petri-frontend-attractor", diff --git a/lib/apps/fabro-cli/src/commands/run/petri_worker.rs b/lib/apps/fabro-cli/src/commands/run/petri_worker.rs index 2ec71756c..98e04683b 100644 --- a/lib/apps/fabro-cli/src/commands/run/petri_worker.rs +++ b/lib/apps/fabro-cli/src/commands/run/petri_worker.rs @@ -28,7 +28,12 @@ //! lowered and admitted at create time with the server's layer, and nothing //! lowers again at execution. The model client is built from the worker's //! catalog and vault snapshot for the providers whose credentials resolve, -//! the same eligible set the legacy worker's LLM backend uses. +//! the same eligible set the legacy worker's LLM backend uses. Fabro's run +//! tools go to every agent session of the run when the run's settings +//! enable them (`[run.agent] fabro_tools`) and the worker token carries the +//! `agent:run_tools` scope the server issues for such a run, the same gate +//! the legacy worker applies; they bind to the worker's client and the run +//! id, as the legacy worker binds them. use std::path::{Path, PathBuf}; use std::sync::Arc; @@ -50,6 +55,7 @@ use fabro_workflow::Error as WorkflowError; use fabro_workflow::event::{self as workflow_event, Emitter, Event, RunEventSink}; use fabro_workflow::run_control::RunControlState; use fabro_workflow::runtime_store::RunStoreHandle; +use fabro_workflow::services::FabroRunToolServices; use tokio_util::sync::CancellationToken; use tracing::{info, warn}; @@ -117,7 +123,8 @@ pub(super) async fn execute(worker: PetriWorker<'_>) -> Result<()> { RunEventSink::backend(worker.run_store.clone()), ); - let runtime = runtime_spec(worker.storage_dir, &worker.run_state).await?; + let run_tools = run_tool_services(&worker); + let runtime = runtime_spec(worker.storage_dir, &worker.run_state, run_tools).await?; let execution = match worker.mode { RunWorkerMode::Start => { let client = worker.client.clone_for_reuse(); @@ -227,10 +234,42 @@ pub(super) async fn execute(worker: PetriWorker<'_>) -> Result<()> { } } +/// Fabro's run tools for the run's agent sessions, when the run's settings +/// enable them and the worker token carries the scope; `None` otherwise. +/// The server issues the scope from the same setting, so the two agree +/// unless the token was issued for another run. +fn run_tool_services(worker: &PetriWorker<'_>) -> Option { + let enabled = worker.run_state.spec.settings.run.agent.fabro_tools; + let scoped = runner::fabro_run_tools_enabled_from_worker_token(worker.worker_token); + if !enabled || !scoped { + info!( + run_id = %worker.run_id, + enabled, + scoped, + "Fabro's run tools are not registered on this Petri run" + ); + return None; + } + let services = runner::build_fabro_run_tool_services( + worker.worker_token, + worker.client.clone_for_reuse(), + worker.run_id, + ); + if services.is_some() { + info!(run_id = %worker.run_id, "Fabro's run tools are registered on this Petri run"); + } + services +} + /// The runtime the worker hands Petri: no settings layer (nothing lowers /// at execution), the model client over the worker's catalog and vault for -/// the providers whose credentials resolve, and the run's mode. -async fn runtime_spec(storage_dir: &Path, run_state: &RunProjection) -> Result { +/// the providers whose credentials resolve, the run's mode, and the run +/// tools when the run has them. +async fn runtime_spec( + storage_dir: &Path, + run_state: &RunProjection, + run_tools: Option, +) -> Result { let catalog = command_context::load_cli_catalog().context("failed to build worker LLM catalog")?; let vault = runner::load_worker_vault(storage_dir).await?; @@ -251,5 +290,6 @@ async fn runtime_spec(storage_dir: &Path, run_state: &RunProjection) -> Result bool { +pub(super) fn fabro_run_tools_enabled_from_worker_token(worker_token: &str) -> bool { // Local tool registration only. The server validates the token signature and // scopes. insecure_decode::(worker_token) @@ -232,7 +232,7 @@ fn worker_scope_has_run_tools(scope_claim: &str) -> bool { has_run_worker && has_agent_run_tools } -fn build_fabro_run_tool_services( +pub(super) fn build_fabro_run_tool_services( worker_token: &str, client: fabro_client::Client, current_run_id: RunId, diff --git a/lib/apps/fabro-server/src/server/petri_runs.rs b/lib/apps/fabro-server/src/server/petri_runs.rs index 931a8a459..6e2043c65 100644 --- a/lib/apps/fabro-server/src/server/petri_runs.rs +++ b/lib/apps/fabro-server/src/server/petri_runs.rs @@ -90,6 +90,9 @@ pub(crate) fn runtime_spec( model_client, dry_run, fabro_home: None, + // The in-process test path has no worker client to bind the run + // tools to; like the legacy in-process path, it runs without them. + run_tools: None, } } diff --git a/lib/components/fabro-petri/Cargo.toml b/lib/components/fabro-petri/Cargo.toml index 3061d5f7a..6823236a6 100644 --- a/lib/components/fabro-petri/Cargo.toml +++ b/lib/components/fabro-petri/Cargo.toml @@ -25,6 +25,8 @@ fabro-db = { path = "../../foundation/fabro-db" } fabro-http.workspace = true fabro-store = { path = "../fabro-store" } fabro-types = { path = "../../foundation/fabro-types" } +fabro-workflow = { path = "../fabro-workflow" } +pebble-coding-agent.workspace = true petri_runtime.workspace = true petri_execution.workspace = true petri_store.workspace = true @@ -45,6 +47,10 @@ tokio-util.workspace = true tracing.workspace = true [dev-dependencies] +fabro-petri = { path = ".", features = ["test-support"] } +fabro-tool = { path = "../fabro-tool" } +httpmock = "0.8" +pebble-coding-agent = { workspace = true, features = ["test-util"] } fabro-auth = { path = "../../foundation/fabro-auth", features = ["test-support"] } fabro-llm = { path = "../fabro-llm", features = ["test-support"] } fabro-store = { path = "../fabro-store", features = ["test-support"] } diff --git a/lib/components/fabro-petri/src/host_tools.rs b/lib/components/fabro-petri/src/host_tools.rs new file mode 100644 index 000000000..8745edee4 --- /dev/null +++ b/lib/components/fabro-petri/src/host_tools.rs @@ -0,0 +1,167 @@ +//! Fabro's run tools inside a Petri run: the adapter from Petri's host tool +//! capability to `register_fabro_run_tools` (integration plan item F3.4). +//! +//! Petri's native agent step asks the [`HostTools`] capability for the +//! host's tools once per agent session, with a [`HostToolContext`] naming +//! the stage the session serves: the run key, the invocation and execution, +//! the node, the firing and the attempt. This module answers with the same +//! tools the legacy worker registers on a stage's Pebble builder, +//! `fabro_run_create`, `fabro_run_get` and the rest, built by +//! `register_fabro_run_tools` over the same [`FabroRunToolServices`]: the +//! worker's authenticated client and the run id every child run is parented +//! to. `fabro exec` and Ask Fabro sessions keep registering the tools on +//! their builders directly; this adapter is only for a run Petri executes. +//! +//! From there Petri treats the tools as any other: the model sees their +//! definitions beside Pebble's, every call passes through the run's tool +//! hooks (a `pre_tool_use` hook from `[[run.hooks]]` can block one), Pebble +//! reports the call on its event stream, and Petri records it under the +//! stage. A sub-agent inherits them through Pebble's own rule, since the +//! registration marks every run tool `allow_in_subagents`. +//! +//! # Identity +//! +//! The run tools need one identity: the Fabro run id, which is Petri's run +//! key for the run (`RunRequest::run_id`) and `FabroRunToolServices:: +//! current_run_id`. It is the parent link of every child run a stage +//! creates. No run tool records a stage on the effects it creates, so +//! nothing here derives Fabro's old `StageId` (`node@visit`); the stage a +//! call came from is Petri's own record of the call, under the stage key +//! `(run, execution, firing)`, and this adapter logs that key with the node +//! and attempt when it builds a session's tools. +//! +//! The builder refuses a context whose run key is not the run the services +//! were built for: the tools would parent child runs to the wrong run. That +//! cannot happen in the worker, which builds both from one run id, so it is +//! logged as an error and the session gets no run tools rather than the +//! wrong ones. + +use fabro_workflow::handler::llm::register_fabro_run_tools; +use fabro_workflow::services::FabroRunToolServices; +use pebble_coding_agent::tools::RegisteredTool; +use petri_attractor_steps::host_tools::{HostToolContext, HostTools}; +use tracing::{debug, error}; + +/// The `HostTools` capability that gives every native agent session of the +/// run Fabro's run tools, bound to `services`. Register it on the runtime +/// the run executes with; `RuntimeSpec::run_tools` does. +#[must_use] +pub fn capability(services: FabroRunToolServices) -> HostTools { + HostTools::new().with(move |context| tools_for_stage(&services, context)) +} + +/// The run tools for the session `context` names: what +/// `register_fabro_run_tools` builds for the legacy worker, or nothing when +/// the context's run is not the one `services` serves. +#[must_use] +pub fn tools_for_stage( + services: &FabroRunToolServices, + context: &HostToolContext, +) -> Vec { + let run_id = services.current_run_id.to_string(); + if context.run.as_str() != run_id { + error!( + run = %context.run, + services_run_id = %run_id, + node = %context.node, + "the Petri run key is not the run the Fabro run tools serve; the session gets no run tools" + ); + return Vec::new(); + } + debug!( + run = %context.run, + invocation = %context.invocation, + execution = %context.execution, + firing = %context.firing, + node = %context.node, + attempt = ?context.attempt, + "registering Fabro's run tools on a Petri agent session" + ); + register_fabro_run_tools(services) +} + +/// What a test reads back from a Petri run's record about the run tools, +/// without depending on the Petri packages itself. +#[cfg(feature = "test-support")] +pub mod recorded { + use petri_attractor_steps::hooks::REPORT_EVENT; + use petri_execution::events::{RunEvent, replay_run}; + use petri_execution::{Access, RunKey, RunStore}; + pub use petri_execution::{ExecutionId, InvocationId}; + use serde_json::Value; + + /// One completed tool call as Petri recorded it: the stage it was + /// recorded under and Pebble's completion payload. + #[derive(Clone, Debug)] + pub struct ToolCall { + /// The node's instance name. + pub node: String, + pub invocation: Option, + pub execution: Option, + /// The parent session of a sub-agent's call; `None` for a call of + /// the stage's own session. + pub parent_session: Option, + /// Pebble's `ToolCallCompleted` payload (`tool_name`, `is_error`, + /// `error_kind`, the output). + pub payload: Value, + } + + /// Every event of the run, replayed from its record. + async fn events(store: &dyn RunStore, run_id: &str) -> anyhow::Result> { + let logs = store + .open(&RunKey::new(run_id), Access::Read) + .await + .map_err(anyhow::Error::new)?; + replay_run(&*logs).await.map_err(anyhow::Error::new) + } + + /// Every completed call of `tool` in the run's record, in record order. + pub async fn tool_calls( + store: &dyn RunStore, + run_id: &str, + tool: &str, + ) -> anyhow::Result> { + Ok(events(store, run_id) + .await? + .iter() + .filter_map(|event| { + let custom = event.custom()?; + if custom["kind"] != "pebble" { + return None; + } + let envelope = custom.get("event")?; + let payload = envelope["event"].get("ToolCallCompleted")?; + if payload["tool_name"] != tool { + return None; + } + Some(ToolCall { + node: event + .subject + .as_ref() + .map(|subject| subject.node.name.to_string()) + .unwrap_or_default(), + invocation: event.context.invocation, + execution: event.context.execution, + parent_session: envelope["parent_session_id"].as_str().map(str::to_owned), + payload: payload.clone(), + }) + }) + .collect()) + } + + /// Every hook report for `event` (`pre_tool_use`, say) in the run's + /// record, as Petri's hook service recorded it. + pub async fn hook_reports( + store: &dyn RunStore, + run_id: &str, + event: &str, + ) -> anyhow::Result> { + Ok(events(store, run_id) + .await? + .iter() + .filter_map(RunEvent::custom) + .filter(|value| value["kind"] == REPORT_EVENT && value["event"] == event) + .cloned() + .collect()) + } +} diff --git a/lib/components/fabro-petri/src/lib.rs b/lib/components/fabro-petri/src/lib.rs index 443032ac8..767e05b3a 100644 --- a/lib/components/fabro-petri/src/lib.rs +++ b/lib/components/fabro-petri/src/lib.rs @@ -23,8 +23,10 @@ //! - [`HttpRunStore`]: the same store as a run's worker process reaches it, //! over the server's API with the worker's token and its launch id as the //! lease owner; +//! - [`host_tools`]: Fabro's run tools on every native agent session of a run, +//! through Petri's `HostTools` capability; //! - the platform adapters still to come: hooks, interviews over Fabro's API, -//! secrets, output storage, the run tools, the event projection. +//! secrets, output storage, the event projection. //! //! The Petri packages are pinned by revision in the workspace `Cargo.toml` //! under `petri_*` keys. @@ -32,6 +34,7 @@ pub mod admission; pub mod check; pub mod engine; +pub mod host_tools; pub mod http_store; pub mod interviewer; pub mod petri; diff --git a/lib/components/fabro-petri/src/runtime.rs b/lib/components/fabro-petri/src/runtime.rs index b63509c84..ae8ba0752 100644 --- a/lib/components/fabro-petri/src/runtime.rs +++ b/lib/components/fabro-petri/src/runtime.rs @@ -5,13 +5,16 @@ //! carrying the server's settings layer, the Attractor step kinds (the real //! ones, or the simulated registry for a dry run), the model client as the //! `PebbleClient` capability so Petri's admission pass pins every LLM node's -//! route, and the Fabro home for the skills step. Nothing here knows about a -//! run: the store and the run options are added by the caller. +//! route, the Fabro home for the skills step, and, at execution, Fabro's run +//! tools as the `HostTools` capability when the run enables them. Nothing +//! here knows about a run's record: the store and the run options are added +//! by the caller. use std::path::PathBuf; use std::sync::Arc; use fabro_http::HttpClient; +use fabro_workflow::services::FabroRunToolServices; use lithos_llm::Client; use lithos_llm::catalog::{Catalog, ProviderId}; use lithos_llm::client::ClientBuildError; @@ -22,6 +25,8 @@ use petri_frontend_fabro::Fabro; use petri_runtime::Runtime; use tracing::debug; +use crate::host_tools; + /// What every Petri runtime Fabro builds is configured with. #[derive(Clone, Default)] pub struct RuntimeSpec { @@ -39,6 +44,11 @@ pub struct RuntimeSpec { /// The Fabro home the skills step reads; `None` leaves it to Petri's /// own lookup (`FABRO_HOME`, else `$HOME/.fabro`). pub fabro_home: Option, + /// Fabro's run tools for every native agent session of the run, when + /// the run enables them (`[run.agent] fabro_tools` and the worker + /// token's `agent:run_tools` scope); `None` gives the sessions Pebble's + /// tools alone. See [`crate::host_tools`]. + pub run_tools: Option, } impl RuntimeSpec { @@ -60,6 +70,9 @@ impl RuntimeSpec { if let Some(home) = home { runtime = runtime.capability(home); } + if let Some(services) = &self.run_tools { + runtime = runtime.capability(host_tools::capability(services.clone())); + } if for_execution && self.dry_run { petri_attractor_steps::register_stubs(runtime) } else { diff --git a/lib/components/fabro-petri/tests/host_tools.rs b/lib/components/fabro-petri/tests/host_tools.rs new file mode 100644 index 000000000..33c301d11 --- /dev/null +++ b/lib/components/fabro-petri/tests/host_tools.rs @@ -0,0 +1,300 @@ +//! Fabro's run tools on a Petri run from this crate (integration plan item +//! F3.4): `RuntimeSpec::run_tools` installs the adapter as Petri's host +//! tool capability, a workflow with one agent stage runs on the real step +//! registry against a scripted model, and the stage's session gets the +//! tools the legacy worker registers, bound to the run: the model is +//! advertised every run tool, its `fabro_run_create` call reaches Fabro's +//! API with the Petri run as the child's parent, the API's answer comes +//! back to the model, and the call is in the run's record under the stage. +//! +//! Every run takes its scope through the sandbox-driver host plugin, so +//! the tests skip when that executable is not found, unless +//! `FABRO_REQUIRE_SANDBOX_PLUGINS` is set. + +#![expect( + clippy::disallowed_methods, + reason = "the tests locate the plugin executable through the process environment" +)] +#![expect(clippy::print_stderr, reason = "a skipped test says why on its stderr")] + +use std::env; +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use fabro_petri::host_tools::recorded::{self, ExecutionId, InvocationId}; +use fabro_petri::runtime::RuntimeSpec; +use fabro_tool::fabro_client::ClientBackend; +use fabro_types::{BlobHash, RunId, WorkflowVersionId}; +use fabro_workflow::handler::llm::register_fabro_run_tools; +use fabro_workflow::services::FabroRunToolServices; +use httpmock::{Method, MockServer}; +use lithos_llm::types::Request; +use pebble_coding_agent::test_support::{ + ScriptedCall, ScriptedProvider, scripted_client, text_response, tool_call_response, +}; +use petri_execution::host::{self, HostRun}; +use petri_runtime::executor::Retention; +use petri_runtime::frontend::CompileInputs; +use petri_runtime::ir::RunStatus; +use petri_runtime::{RunOptions, Runtime}; +use petri_store::{MemoryRunStore, RunKey, RunStore}; +use serde_json::json; +use tokio::fs; + +const HOST_PLUGIN: &str = "sandbox-driver-host"; +const HOST_PLUGIN_OVERRIDE: &str = "PETRI_SANDBOX_HOST_PLUGIN"; +const REQUIRE_ENV: &str = "FABRO_REQUIRE_SANDBOX_PLUGINS"; + +/// One agent stage on the native backend, pinned to the scripted model. +const AGENT_WORKFLOW: &str = r#"digraph Agent { + graph [goal="Start a child run", backend="api", default_max_retries=0] + start [shape=Mdiamond] + exit [shape=Msquare] + work [shape=box, prompt="Start the child run", model="test/model", max_retries=0] + start -> work -> exit +}"#; + +const AGENT_SETTINGS: &str = "_version = 1\n\n[workflow]\ngraph = \"workflow.fabro\"\n"; + +/// The host plugin as Petri's lookup finds it: the override variable, else +/// the executable on `PATH`. `None`, after saying so, when the test should +/// skip; a panic when the environment forbids a skip. +fn host_plugin() -> Option { + let found = env::var_os(HOST_PLUGIN_OVERRIDE) + .map(PathBuf::from) + .or_else(|| { + env::split_paths(&env::var_os("PATH")?) + .map(|dir| dir.join(HOST_PLUGIN)) + .find(|candidate| candidate.is_file()) + }); + if found.is_none() { + assert!( + env::var_os(REQUIRE_ENV).is_none(), + "{REQUIRE_ENV} is set, but {HOST_PLUGIN} is not on PATH and {HOST_PLUGIN_OVERRIDE} is unset" + ); + eprintln!("skipping: {HOST_PLUGIN} is not on PATH and {HOST_PLUGIN_OVERRIDE} is unset"); + } + found +} + +/// Write the agent bundle into `/.fabro/workflows/agent`; the +/// workflow file. +async fn install_bundle(root: &Path) -> PathBuf { + let bundle = root.join(".fabro").join("workflows").join("agent"); + fs::create_dir_all(&bundle) + .await + .expect("the bundle directory is creatable"); + fs::write(bundle.join("workflow.fabro"), AGENT_WORKFLOW) + .await + .expect("the workflow is writable"); + fs::write(bundle.join("workflow.toml"), AGENT_SETTINGS) + .await + .expect("the settings are writable"); + bundle.join("workflow.fabro") +} + +/// The run tools' services over `server`, as the worker binds them: the +/// client backend and the run the tools serve. +fn services(server: &MockServer, run_id: RunId) -> FabroRunToolServices { + let client = fabro_client::Client::new_no_proxy(&server.url("")).expect("the client builds"); + FabroRunToolServices { + backend: Arc::new(ClientBackend::new(Arc::new(client))), + current_run_id: run_id, + } +} + +/// The scripted model: one `fabro_run_create` call, then a closing line. +fn scripted_model(version_id: &str) -> (lithos_llm::Client, Arc) { + scripted_client(vec![ + ScriptedCall::response(tool_call_response( + "fabro_run_create", + "create", + json!({ + "runs": [{ + "workflow_version_id": version_id, + "target": {"kind": "none"}, + "args": {"auto_approve": false}, + }], + }), + )), + ScriptedCall::response(text_response("Asked for the child run.")), + ]) +} + +/// The runtime the worker would build for the run: the scripted model as +/// the model client and `run_tools` as the run tools. +fn runtime( + run_dir: &Path, + run_id: &str, + model_client: lithos_llm::Client, + run_tools: Option, + store: &Arc, +) -> Runtime { + let mut options = RunOptions::new(run_dir); + options.grace = Duration::from_secs(2); + options.retention = Retention::Never; + options.echo = false; + options.run_key = Some(RunKey::new(run_id)); + RuntimeSpec { + model_client: Some(model_client), + run_tools, + ..RuntimeSpec::default() + } + .runtime(true) + .store(Arc::clone(store) as Arc) + .options(options) +} + +/// Lower the bundle and run it to its end. +async fn run(rt: &Runtime, workflow: &Path) { + let lowered = rt + .check(workflow, None, None, &CompileInputs::new()) + .expect("the workflow file loads"); + let graph = lowered + .graph + .unwrap_or_else(|| panic!("the workflow lowers: {:?}", lowered.diagnostics)); + let report = host::run_configured(rt, HostRun::new(graph), |_, _| {}) + .await + .expect("the run completes"); + assert_eq!( + report.status, + RunStatus::Success, + "errors: {:?}; history: {:#?}", + report.state.errors(), + report.state.history() + ); +} + +/// The tools a request advertised, as `(name, description)`. +fn advertised(request: &Request) -> Vec<(String, String)> { + request + .tools() + .iter() + .map(|tool| (tool.name.clone(), tool.description.clone())) + .collect() +} + +/// The stage's session is given every run tool the legacy worker +/// registers, by the same names and descriptions; its `fabro_run_create` +/// call reaches Fabro's API with the Petri run as the parent; the API's +/// answer reaches the model; the call is in the record under the stage. +#[tokio::test] +async fn a_petri_stage_calls_a_run_tool_bound_to_the_run() { + if host_plugin().is_none() { + return; + } + let root = tempfile::tempdir().expect("a temp dir"); + let workflow = install_bundle(root.path()).await; + let run_id = RunId::new(); + let version_id: WorkflowVersionId = BlobHash::new(b"child workflow").into(); + let server = MockServer::start_async().await; + // Admission rejection proves the tool reached the canonical API with + // the Petri run as the child's parent, and nothing was created. + let create = server + .mock_async(|when, then| { + when.method(Method::POST) + .path("/api/v1/runs") + .json_body(json!({ + "workflow_version_id": version_id, + "target": {"kind": "none"}, + "parent_id": run_id, + "args": {"auto_approve": false}, + })); + then.status(422).body("native admission rejection"); + }) + .await; + let services = services(&server, run_id); + let legacy: Vec<(String, String)> = register_fabro_run_tools(&services) + .iter() + .map(|tool| { + ( + tool.definition().name.clone(), + tool.definition().description.clone(), + ) + }) + .collect(); + let (client, provider) = scripted_model(&version_id.to_string()); + let store = Arc::new(MemoryRunStore::new()); + let rt = runtime( + &root.path().join("run"), + &run_id.to_string(), + client, + Some(services), + &store, + ); + + run(&rt, &workflow).await; + + let requests = provider.requests(); + assert_eq!(requests.len(), 2, "one tool call, one closing turn"); + let tools = advertised(&requests[0]); + assert!(!legacy.is_empty()); + for tool in &legacy { + assert!( + tools.contains(tool), + "the model was advertised {tool:?} as the legacy worker registers it: {tools:?}" + ); + } + assert!( + tools.iter().any(|(name, _)| name == "shell"), + "Pebble's own tools stay: {tools:?}" + ); + let answer = serde_json::to_string(&requests[1]).expect("the request serializes"); + assert!( + answer.contains("native admission rejection"), + "the model read the API's answer: {answer}" + ); + create.assert_calls_async(1).await; + + let calls = recorded::tool_calls(store.as_ref(), &run_id.to_string(), "fabro_run_create") + .await + .expect("the record replays"); + assert_eq!(calls.len(), 1, "{calls:?}"); + assert_eq!(calls[0].node, "work", "recorded under the stage"); + assert_eq!(calls[0].invocation, Some(InvocationId::ROOT)); + assert_eq!(calls[0].execution, Some(ExecutionId::new(0))); + assert!(calls[0].parent_session.is_none()); + assert_eq!(calls[0].payload["is_error"], true, "{:?}", calls[0].payload); +} + +/// Services bound to another run give the stage no run tools: the model +/// is not advertised them, and its call is refused as an unknown tool +/// rather than parenting a child run to the wrong run. +#[tokio::test] +async fn services_for_another_run_give_the_stage_no_run_tools() { + if host_plugin().is_none() { + return; + } + let root = tempfile::tempdir().expect("a temp dir"); + let workflow = install_bundle(root.path()).await; + let run_id = RunId::new(); + let server = MockServer::start_async().await; + let create = server + .mock_async(|when, then| { + when.method(Method::POST).path("/api/v1/runs"); + then.status(500); + }) + .await; + let services = services(&server, RunId::new()); + let (client, provider) = scripted_model("0000"); + let store = Arc::new(MemoryRunStore::new()); + let rt = runtime( + &root.path().join("run"), + &run_id.to_string(), + client, + Some(services), + &store, + ); + + run(&rt, &workflow).await; + + let requests = provider.requests(); + assert_eq!(requests.len(), 2); + let tools = advertised(&requests[0]); + assert!( + !tools.iter().any(|(name, _)| name.starts_with("fabro_")), + "no run tool was advertised: {tools:?}" + ); + create.assert_calls_async(0).await; +}