mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-10-02 02:13:49 +00:00
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 <noreply@anthropic.com>
This commit is contained in:
parent
a745d0b75d
commit
49e70e481d
9 changed files with 546 additions and 9 deletions
5
Cargo.lock
generated
5
Cargo.lock
generated
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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<FabroRunToolServices> {
|
||||
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<RuntimeSpec> {
|
||||
/// 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<FabroRunToolServices>,
|
||||
) -> Result<RuntimeSpec> {
|
||||
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<R
|
|||
model_client,
|
||||
dry_run: run_state.spec.settings.run.execution.mode == RunMode::DryRun,
|
||||
fabro_home: None,
|
||||
run_tools,
|
||||
})
|
||||
}
|
||||
|
|
|
|||
|
|
@ -212,7 +212,7 @@ struct WorkerTokenScopeClaim {
|
|||
scope: String,
|
||||
}
|
||||
|
||||
fn fabro_run_tools_enabled_from_worker_token(worker_token: &str) -> 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::<WorkerTokenScopeClaim>(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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -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"] }
|
||||
|
|
|
|||
167
lib/components/fabro-petri/src/host_tools.rs
Normal file
167
lib/components/fabro-petri/src/host_tools.rs
Normal file
|
|
@ -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<RegisteredTool> {
|
||||
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<InvocationId>,
|
||||
pub execution: Option<ExecutionId>,
|
||||
/// The parent session of a sub-agent's call; `None` for a call of
|
||||
/// the stage's own session.
|
||||
pub parent_session: Option<String>,
|
||||
/// 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<Vec<RunEvent>> {
|
||||
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<Vec<ToolCall>> {
|
||||
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<Vec<Value>> {
|
||||
Ok(events(store, run_id)
|
||||
.await?
|
||||
.iter()
|
||||
.filter_map(RunEvent::custom)
|
||||
.filter(|value| value["kind"] == REPORT_EVENT && value["event"] == event)
|
||||
.cloned()
|
||||
.collect())
|
||||
}
|
||||
}
|
||||
|
|
@ -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;
|
||||
|
|
|
|||
|
|
@ -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<PathBuf>,
|
||||
/// 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<FabroRunToolServices>,
|
||||
}
|
||||
|
||||
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 {
|
||||
|
|
|
|||
300
lib/components/fabro-petri/tests/host_tools.rs
Normal file
300
lib/components/fabro-petri/tests/host_tools.rs
Normal file
|
|
@ -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<PathBuf> {
|
||||
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 `<root>/.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<ScriptedProvider>) {
|
||||
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<FabroRunToolServices>,
|
||||
store: &Arc<MemoryRunStore>,
|
||||
) -> 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<dyn RunStore>)
|
||||
.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;
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue