mirror of
https://github.com/fabro-sh/fabro.git
synced 2026-10-01 02:04:24 +00:00
Cover the run tools inside a Petri run through the server and its worker
Three scenarios on the real binary: an agent creates a child run with `fabro_run_create` from inside a Petri run and the child carries the parent link; a `[[run.hooks]]` pre_tool_use hook blocks a run tool, the model reads the reason, and Petri's record holds the report and the denied call; a sub-agent calls an inherited run tool, recorded under the parent stage naming the parent session. The Petri scenario harness is shared: the server can start with extra settings and vault entries, and the detached run takes extra arguments. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
parent
49e70e481d
commit
23c422f9a2
4 changed files with 522 additions and 18 deletions
|
|
@ -125,6 +125,7 @@ fabro-mcp = { path = "../../components/fabro-mcp", features = ["test-support"] }
|
|||
fabro-build-support = { path = "../../foundation/build-support" }
|
||||
fabro-sandbox = { path = "../../components/fabro-sandbox", features = ["test-support"] }
|
||||
fabro-server = { path = "../fabro-server", features = ["test-support"] }
|
||||
fabro-petri = { path = "../../components/fabro-petri", features = ["test-support"] }
|
||||
fabro-workflow = { path = "../../components/fabro-workflow", features = ["test-support"] }
|
||||
fabro-types = { path = "../../foundation/fabro-types", features = ["clap", "test-support"] }
|
||||
insta = { workspace = true, features = ["filters"] }
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ mod auth;
|
|||
mod exec;
|
||||
mod lifecycle;
|
||||
mod petri;
|
||||
mod petri_tools;
|
||||
mod server_lifecycle;
|
||||
mod smoke;
|
||||
|
||||
|
|
|
|||
|
|
@ -10,6 +10,9 @@
|
|||
//! executable is not found, unless `FABRO_REQUIRE_SANDBOX_PLUGINS` is set.
|
||||
//! The plugin's path override crosses into the server and its workers the
|
||||
//! way `PATH` does.
|
||||
//!
|
||||
//! The harness here (the server, the detached run, the status and event
|
||||
//! reads) is shared with the run-tools scenarios in `petri_tools.rs`.
|
||||
|
||||
#![expect(
|
||||
clippy::disallowed_methods,
|
||||
|
|
@ -35,6 +38,7 @@ use fabro_static::EnvVars;
|
|||
use fabro_store::EventEnvelope;
|
||||
use fabro_test::{apply_test_isolation, expect_reqwest_json, isolated_storage_dir, test_context};
|
||||
use fabro_types::EventBody;
|
||||
use fabro_vault::{SecretType, Vault};
|
||||
|
||||
use crate::cmd::support::created_run_id;
|
||||
use crate::support::{
|
||||
|
|
@ -49,7 +53,7 @@ const POLL: Duration = Duration::from_millis(50);
|
|||
/// 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> {
|
||||
pub(super) fn host_plugin() -> Option<PathBuf> {
|
||||
let found = env::var_os(EnvVars::PETRI_SANDBOX_HOST_PLUGIN)
|
||||
.map(PathBuf::from)
|
||||
.or_else(|| {
|
||||
|
|
@ -73,18 +77,26 @@ fn host_plugin() -> Option<PathBuf> {
|
|||
|
||||
/// A foreground server on its own disk storage, dev-token auth, started
|
||||
/// from the compiled `fabro` binary. Dropping it kills the process.
|
||||
struct RunningServer {
|
||||
child: Option<Child>,
|
||||
home_root: tempfile::TempDir,
|
||||
_storage_root: tempfile::TempDir,
|
||||
storage_dir: PathBuf,
|
||||
config_path: PathBuf,
|
||||
port: u16,
|
||||
api_base_url: String,
|
||||
pub(super) struct RunningServer {
|
||||
child: Option<Child>,
|
||||
home_root: tempfile::TempDir,
|
||||
_storage_root: tempfile::TempDir,
|
||||
pub(super) storage_dir: PathBuf,
|
||||
config_path: PathBuf,
|
||||
port: u16,
|
||||
pub(super) api_base_url: String,
|
||||
}
|
||||
|
||||
impl RunningServer {
|
||||
async fn start() -> Self {
|
||||
pub(super) async fn start() -> Self {
|
||||
Self::start_with("", &[]).await
|
||||
}
|
||||
|
||||
/// Start with `settings` appended to the server's settings file (the
|
||||
/// workers read the same file through `FABRO_CONFIG`) and `secrets`
|
||||
/// in the vault before the first launch, so the server and its workers
|
||||
/// see them from the start.
|
||||
pub(super) async fn start_with(settings: &str, secrets: &[(&str, &str)]) -> Self {
|
||||
let home_root = tempfile::tempdir_in("/tmp").expect("home tempdir");
|
||||
let storage_root = isolated_storage_dir();
|
||||
let storage_dir = storage_root.path().join("storage");
|
||||
|
|
@ -92,9 +104,18 @@ impl RunningServer {
|
|||
let config_path = home_root.path().join("settings.toml");
|
||||
std::fs::write(
|
||||
&config_path,
|
||||
"_version = 1\n\n[server.auth]\nmethods = [\"dev-token\"]\n",
|
||||
format!("_version = 1\n\n[server.auth]\nmethods = [\"dev-token\"]\n{settings}"),
|
||||
)
|
||||
.expect("the server settings write");
|
||||
if !secrets.is_empty() {
|
||||
let mut vault = Vault::load(Storage::new(&storage_dir).secrets_path())
|
||||
.expect("the server vault loads");
|
||||
for (name, value) in secrets {
|
||||
vault
|
||||
.set(name, value, SecretType::Token, None)
|
||||
.expect("the secret stores in the server vault");
|
||||
}
|
||||
}
|
||||
let runtime_directory = Storage::new(&storage_dir).runtime_directory();
|
||||
envfile::merge_env_file(&runtime_directory.env_path(), [
|
||||
("SESSION_SECRET", TEST_SESSION_SECRET),
|
||||
|
|
@ -157,13 +178,13 @@ impl RunningServer {
|
|||
Stdio::from(file)
|
||||
}
|
||||
|
||||
fn stderr_text(&self) -> String {
|
||||
pub(super) fn stderr_text(&self) -> String {
|
||||
std::fs::read_to_string(self.storage_dir.with_file_name("server.stderr.log"))
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
/// The `--server` target a CLI command reaches this server at.
|
||||
fn target(&self) -> String {
|
||||
pub(super) fn target(&self) -> String {
|
||||
format!("{}/api/v1", self.api_base_url)
|
||||
}
|
||||
|
||||
|
|
@ -175,7 +196,7 @@ impl RunningServer {
|
|||
let _ = child.wait();
|
||||
}
|
||||
|
||||
fn shutdown(mut self) {
|
||||
pub(super) fn shutdown(mut self) {
|
||||
let mut stop = Command::new(env!("CARGO_BIN_EXE_fabro"));
|
||||
apply_test_isolation(&mut stop, self.home_root.path());
|
||||
stop.args(["server", "stop"])
|
||||
|
|
@ -203,7 +224,7 @@ impl RunningServer {
|
|||
|
||||
/// Petri's store over the server's database, read beside the server:
|
||||
/// what `petri inspect` would see.
|
||||
async fn petri_store(&self) -> SqliteRunStore {
|
||||
pub(super) async fn petri_store(&self) -> SqliteRunStore {
|
||||
let database = fabro_db::Database::connect(Storage::new(&self.storage_dir).sqlite_path())
|
||||
.await
|
||||
.expect("the server database opens");
|
||||
|
|
@ -277,6 +298,17 @@ fn run_detached(
|
|||
context: &fabro_test::TestContext,
|
||||
server: &RunningServer,
|
||||
workspace: &Path,
|
||||
) -> String {
|
||||
run_detached_with(context, server, workspace, &[])
|
||||
}
|
||||
|
||||
/// [`run_detached`] with `extra` arguments on the command, such as the
|
||||
/// model to run the workflow's agents on.
|
||||
pub(super) fn run_detached_with(
|
||||
context: &fabro_test::TestContext,
|
||||
server: &RunningServer,
|
||||
workspace: &Path,
|
||||
extra: &[&str],
|
||||
) -> String {
|
||||
let target = server.target();
|
||||
seed_dev_token_auth(
|
||||
|
|
@ -294,8 +326,9 @@ fn run_detached(
|
|||
"--auto-approve",
|
||||
"--environment",
|
||||
"local",
|
||||
"workflow.toml",
|
||||
])
|
||||
.args(extra)
|
||||
.arg("workflow.toml")
|
||||
.output()
|
||||
.expect("the detached run executes");
|
||||
assert!(
|
||||
|
|
@ -307,7 +340,7 @@ fn run_detached(
|
|||
created_run_id(&output)
|
||||
}
|
||||
|
||||
async fn run_json(server: &RunningServer, path: &str) -> serde_json::Value {
|
||||
pub(super) async fn run_json(server: &RunningServer, path: &str) -> serde_json::Value {
|
||||
let response = fabro_test::test_http_client()
|
||||
.get(format!("{}/api/v1/{path}", server.api_base_url))
|
||||
.bearer_auth(TEST_DEV_TOKEN)
|
||||
|
|
@ -329,7 +362,11 @@ async fn run_status(server: &RunningServer, run_id: &str) -> String {
|
|||
.to_string()
|
||||
}
|
||||
|
||||
async fn wait_for_status(server: &RunningServer, run_id: &str, expected: &[&str]) -> String {
|
||||
pub(super) async fn wait_for_status(
|
||||
server: &RunningServer,
|
||||
run_id: &str,
|
||||
expected: &[&str],
|
||||
) -> String {
|
||||
let deadline = Instant::now() + RUN_TIMEOUT;
|
||||
loop {
|
||||
let status = run_status(server, run_id).await;
|
||||
|
|
|
|||
465
lib/apps/fabro-cli/tests/it/scenario/petri_tools.rs
Normal file
465
lib/apps/fabro-cli/tests/it/scenario/petri_tools.rs
Normal file
|
|
@ -0,0 +1,465 @@
|
|||
//! Fabro's run tools inside a Petri run (integration plan item F3.4): a
|
||||
//! workflow that enables `[run.agent] fabro_tools` runs on Petri in the
|
||||
//! worker the server launched, and the agent stage's model, the twin, calls
|
||||
//! the run tools the worker registered through Petri's host tool
|
||||
//! capability. The agent creates a child run from inside the Petri run; a
|
||||
//! `[[run.hooks]]` hook blocks a run tool; a sub-agent calls an inherited
|
||||
//! run tool. Each call is read back from Petri's record of the run, under
|
||||
//! the stage it served.
|
||||
//!
|
||||
//! The harness is `petri.rs`'s: a foreground server on disk storage with
|
||||
//! the `openai` provider repointed at the twin, its key in the vault, and
|
||||
//! the run started with `fabro run --detach`. The runs take their host
|
||||
//! scope through the sandbox-driver host plugin, so the tests skip, and say
|
||||
//! why, when it is not found.
|
||||
|
||||
#![expect(
|
||||
clippy::disallowed_methods,
|
||||
reason = "these scenarios stage workspaces with sync std::fs and start a real server subprocess"
|
||||
)]
|
||||
#![expect(
|
||||
clippy::print_stderr,
|
||||
reason = "a scenario says where it is, and why it skipped"
|
||||
)]
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::path::PathBuf;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use fabro_petri::engine::{self, RunStatus};
|
||||
use fabro_petri::host_tools::recorded::{self, ExecutionId, InvocationId, ToolCall};
|
||||
use fabro_static::EnvVars;
|
||||
use fabro_test::{TwinScenario, TwinScenarios, TwinToolCall, test_context, twin_openai};
|
||||
use fabro_types::{WorkflowPath, WorkflowVersion};
|
||||
use serde_json::{Value, json};
|
||||
|
||||
use super::petri::{RunningServer, host_plugin, run_detached_with, run_json, wait_for_status};
|
||||
use crate::support::TEST_DEV_TOKEN;
|
||||
|
||||
const MODEL: &str = "gpt-5.4";
|
||||
const POLL: Duration = Duration::from_millis(50);
|
||||
const RUN_TIMEOUT: Duration = Duration::from_mins(1);
|
||||
|
||||
/// What the stage asks of its agent; every request of the stage's own
|
||||
/// session carries it, and no request of a sub-agent does.
|
||||
const PROMPT: &str = "Work with the Fabro run tools as instructed.";
|
||||
/// The task the stage hands a sub-agent; every request of the child's
|
||||
/// session carries it.
|
||||
const TASK: &str = "Helper: search the Fabro runs and report what you find.";
|
||||
|
||||
/// The child workflow the agent starts: one command stage, on Petri.
|
||||
const CHILD_DOT: &str = r#"digraph Child {
|
||||
graph [goal="Run one command", default_max_retries=0]
|
||||
start [shape=Mdiamond]
|
||||
exit [shape=Msquare]
|
||||
say [shape=parallelogram, script="echo hello from the child", max_retries=0]
|
||||
start -> say -> exit
|
||||
}"#;
|
||||
const CHILD_SETTINGS: &str =
|
||||
"_version = 1\n\n[workflow]\ngraph = \"workflow.fabro\"\nengine = \"petri\"\n";
|
||||
|
||||
/// A `[[run.hooks]]` entry that blocks every `fabro_run_search` call.
|
||||
const BLOCKING_HOOK: &str = r#"
|
||||
[[run.hooks]]
|
||||
name = "no-run-search"
|
||||
event = "pre_tool_use"
|
||||
script = '''if grep -q 'fabro_run_search' "$FABRO_HOOK_CONTEXT"; then echo '{"decision":"block","reason":"run tools are not allowed here"}'; exit 2; fi'''
|
||||
"#;
|
||||
|
||||
/// A server whose `openai` provider is the twin, keyed by `namespace`.
|
||||
async fn server_on_twin(twin_base_url: &str, namespace: &str) -> RunningServer {
|
||||
RunningServer::start_with(
|
||||
&format!("\n[llm.providers.openai]\nbase_url = \"{twin_base_url}\"\n"),
|
||||
&[(EnvVars::OPENAI_API_KEY, namespace)],
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
/// A workspace holding a one-stage agent workflow on Petri with the run
|
||||
/// tools enabled, and `extra_settings` appended to its `workflow.toml`.
|
||||
fn write_agent_workspace(context: &fabro_test::TestContext, extra_settings: &str) -> PathBuf {
|
||||
let workspace = context.temp_dir.join("tools-workspace");
|
||||
std::fs::create_dir_all(&workspace).expect("the workspace creates");
|
||||
std::fs::write(
|
||||
workspace.join("workflow.fabro"),
|
||||
format!(
|
||||
"digraph Tools {{\n graph [goal=\"Use the run tools\", default_max_retries=0]\n \
|
||||
start [shape=Mdiamond]\n exit [shape=Msquare]\n work [shape=box, \
|
||||
prompt=\"{PROMPT}\", max_retries=0]\n start -> work -> exit\n}}\n"
|
||||
),
|
||||
)
|
||||
.expect("the workflow writes");
|
||||
std::fs::write(
|
||||
workspace.join("workflow.toml"),
|
||||
format!(
|
||||
"_version = 1\n\n[workflow]\ngraph = \"workflow.fabro\"\nengine = \"petri\"\n\n[run]\n\
|
||||
goal = \"Use the run tools\"\n\n[run.agent]\nfabro_tools = true\n{extra_settings}"
|
||||
),
|
||||
)
|
||||
.expect("the settings write");
|
||||
workspace
|
||||
}
|
||||
|
||||
/// Register the child workflow as a version through the server's API; its
|
||||
/// id, for the agent's `fabro_run_create` call.
|
||||
async fn register_child_version(server: &RunningServer) -> String {
|
||||
let path = |name: &str| WorkflowPath::new(name).expect("the fixture path is valid");
|
||||
let version = WorkflowVersion::new(
|
||||
path("workflow.fabro"),
|
||||
BTreeMap::from([
|
||||
(path("workflow.fabro"), CHILD_DOT.to_string()),
|
||||
(path("workflow.toml"), CHILD_SETTINGS.to_string()),
|
||||
]),
|
||||
BTreeMap::new(),
|
||||
)
|
||||
.expect("the child version is valid");
|
||||
let response = fabro_test::test_http_client()
|
||||
.post(format!("{}/api/v1/workflow-versions", server.api_base_url))
|
||||
.bearer_auth(TEST_DEV_TOKEN)
|
||||
.json(&version)
|
||||
.send()
|
||||
.await
|
||||
.expect("the registration sends");
|
||||
let body = fabro_test::expect_reqwest_json(
|
||||
response,
|
||||
fabro_http::StatusCode::CREATED,
|
||||
"POST /api/v1/workflow-versions",
|
||||
)
|
||||
.await;
|
||||
body["workflow_version_id"]
|
||||
.as_str()
|
||||
.expect("the registration names the version")
|
||||
.to_string()
|
||||
}
|
||||
|
||||
fn run_stage_scenario() -> TwinScenario {
|
||||
TwinScenario::responses(MODEL).input_contains(PROMPT)
|
||||
}
|
||||
|
||||
fn child_scenario() -> TwinScenario {
|
||||
TwinScenario::responses(MODEL).input_contains(TASK)
|
||||
}
|
||||
|
||||
/// The run's Petri outcome: succeeded and whole, or the test says why not.
|
||||
async fn assert_petri_succeeded(server: &RunningServer, run_id: &str) {
|
||||
let store = server.petri_store().await;
|
||||
let outcome = engine::outcome_of(&store, run_id)
|
||||
.await
|
||||
.expect("the run's Petri record inspects");
|
||||
assert_eq!(outcome.status, RunStatus::Success, "{outcome:?}");
|
||||
assert!(outcome.complete, "{:?}", outcome.incomplete);
|
||||
}
|
||||
|
||||
/// Every completed call of `tool` in the run's Petri record.
|
||||
async fn recorded_calls(server: &RunningServer, run_id: &str, tool: &str) -> Vec<ToolCall> {
|
||||
let store = server.petri_store().await;
|
||||
recorded::tool_calls(&store, run_id, tool)
|
||||
.await
|
||||
.expect("the run's Petri record replays")
|
||||
}
|
||||
|
||||
/// The twin's request log for `namespace`: the input text of each request
|
||||
/// the stage's agent or its sub-agents made, in order. The server's own
|
||||
/// request for a run title goes to the same twin and is left out.
|
||||
async fn request_inputs(twin: &fabro_test::TwinOpenAi, namespace: &str) -> Vec<String> {
|
||||
let logs = twin.request_logs(namespace).await;
|
||||
logs["requests"]
|
||||
.as_array()
|
||||
.expect("the twin request log is an array")
|
||||
.iter()
|
||||
.map(|request| {
|
||||
request["input_text"]
|
||||
.as_str()
|
||||
.unwrap_or_default()
|
||||
.to_string()
|
||||
})
|
||||
.filter(|input| input.contains(PROMPT) || input.contains(TASK))
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Approve `run_id` as a user, through the server's API.
|
||||
async fn approve_run(server: &RunningServer, run_id: &str) {
|
||||
let response = fabro_test::test_http_client()
|
||||
.post(format!(
|
||||
"{}/api/v1/runs/{run_id}/approve",
|
||||
server.api_base_url
|
||||
))
|
||||
.bearer_auth(TEST_DEV_TOKEN)
|
||||
.send()
|
||||
.await
|
||||
.expect("the approval sends");
|
||||
fabro_test::expect_reqwest_json(
|
||||
response,
|
||||
fabro_http::StatusCode::OK,
|
||||
format!("POST /api/v1/runs/{run_id}/approve"),
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
/// The runs whose parent is `parent_id`.
|
||||
async fn children_of(server: &RunningServer, parent_id: &str) -> Vec<Value> {
|
||||
run_json(server, &format!("runs?parent_id={parent_id}")).await["data"]
|
||||
.as_array()
|
||||
.cloned()
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
async fn wait_for_children(server: &RunningServer, parent_id: &str) -> Vec<Value> {
|
||||
let deadline = Instant::now() + RUN_TIMEOUT;
|
||||
loop {
|
||||
let children = children_of(server, parent_id).await;
|
||||
if !children.is_empty() {
|
||||
return children;
|
||||
}
|
||||
assert!(
|
||||
Instant::now() < deadline,
|
||||
"no child run of {parent_id} appeared"
|
||||
);
|
||||
tokio::time::sleep(POLL).await;
|
||||
}
|
||||
}
|
||||
|
||||
/// The agent calls `fabro_run_create` from inside the Petri run: the child
|
||||
/// run is created under the Petri run as its parent and runs to its end,
|
||||
/// the model reads the tool's answer, the run succeeds, and the call is in
|
||||
/// Petri's record under the stage.
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn an_agent_starts_a_child_run_with_a_run_tool_inside_a_petri_run() {
|
||||
if host_plugin().is_none() {
|
||||
return;
|
||||
}
|
||||
let context = test_context!();
|
||||
let twin = twin_openai().await;
|
||||
let namespace = format!("{}::{}", module_path!(), line!());
|
||||
let server = server_on_twin(&twin.base_url, &namespace).await;
|
||||
let child_version = register_child_version(&server).await;
|
||||
// The child runs in the local environment, which serves a folder
|
||||
// target and not a `none` one.
|
||||
let child_workspace = context.temp_dir.join("child-workspace");
|
||||
std::fs::create_dir_all(&child_workspace).expect("the child workspace creates");
|
||||
TwinScenarios::new(namespace.clone())
|
||||
.scenario(run_stage_scenario().tool_call(TwinToolCall::new(
|
||||
"fabro_run_create",
|
||||
json!({
|
||||
"runs": [{
|
||||
"workflow_version_id": child_version,
|
||||
"target": {"kind": "folder", "path": child_workspace},
|
||||
"environment_id": "local",
|
||||
"args": {"auto_approve": true},
|
||||
}],
|
||||
}),
|
||||
)))
|
||||
.scenario(run_stage_scenario().text("The child run is on its way."))
|
||||
.load(twin)
|
||||
.await;
|
||||
let workspace = write_agent_workspace(&context, "");
|
||||
let run_id = run_detached_with(&context, &server, &workspace, &[
|
||||
"--provider",
|
||||
"openai",
|
||||
"--model",
|
||||
MODEL,
|
||||
]);
|
||||
|
||||
eprintln!("run {run_id} started");
|
||||
let status = wait_for_status(&server, &run_id, &["succeeded", "failed"]).await;
|
||||
eprintln!("run {run_id} is {status}");
|
||||
let run = run_json(&server, &format!("runs/{run_id}")).await;
|
||||
assert_eq!(
|
||||
status,
|
||||
"succeeded",
|
||||
"run: {run}\nserver stderr:\n{}",
|
||||
server.stderr_text()
|
||||
);
|
||||
assert_petri_succeeded(&server, &run_id).await;
|
||||
|
||||
let calls = recorded_calls(&server, &run_id, "fabro_run_create").await;
|
||||
assert_eq!(calls.len(), 1, "one call in the record: {calls:?}");
|
||||
let call = &calls[0];
|
||||
assert_eq!(call.node, "work", "recorded under the stage");
|
||||
assert_eq!(call.invocation, Some(InvocationId::ROOT));
|
||||
assert_eq!(call.execution, Some(ExecutionId::new(0)));
|
||||
assert!(
|
||||
call.parent_session.is_none(),
|
||||
"the stage's own session called"
|
||||
);
|
||||
assert_eq!(call.payload["is_error"], false, "{:?}", call.payload);
|
||||
|
||||
let children = wait_for_children(&server, &run_id).await;
|
||||
assert_eq!(children.len(), 1, "one child run: {children:?}");
|
||||
let child_id = children[0]["id"]
|
||||
.as_str()
|
||||
.expect("the child run has an id")
|
||||
.to_string();
|
||||
eprintln!("child run {child_id} found");
|
||||
assert_eq!(children[0]["parent_id"], run_id, "{:?}", children[0]);
|
||||
// A run a worker creates waits for a person's approval, as it does
|
||||
// when the legacy worker's agent creates one; the test is that person.
|
||||
assert_eq!(
|
||||
children[0]["lifecycle"]["status"]["reason"], "approval_required",
|
||||
"{:?}",
|
||||
children[0]["lifecycle"]
|
||||
);
|
||||
approve_run(&server, &child_id).await;
|
||||
let child_status = wait_for_status(&server, &child_id, &["succeeded", "failed"]).await;
|
||||
eprintln!("child run {child_id} is {child_status}");
|
||||
assert_eq!(
|
||||
child_status,
|
||||
"succeeded",
|
||||
"child: {}",
|
||||
run_json(&server, &format!("runs/{child_id}")).await
|
||||
);
|
||||
|
||||
let inputs = request_inputs(twin, &namespace).await;
|
||||
assert_eq!(inputs.len(), 2, "{inputs:?}");
|
||||
assert!(
|
||||
inputs[1].contains(&child_id),
|
||||
"the model read the tool's answer naming the child run: {}",
|
||||
inputs[1]
|
||||
);
|
||||
server.shutdown();
|
||||
}
|
||||
|
||||
/// A `pre_tool_use` hook from `[[run.hooks]]` blocks a run tool as it
|
||||
/// blocks Pebble's: the tool never runs, the model reads the reason, the
|
||||
/// run goes on, and Petri's record holds the hook's report and the denied
|
||||
/// call.
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn a_run_hook_blocks_a_run_tool_inside_a_petri_run() {
|
||||
if host_plugin().is_none() {
|
||||
return;
|
||||
}
|
||||
let context = test_context!();
|
||||
let twin = twin_openai().await;
|
||||
let namespace = format!("{}::{}", module_path!(), line!());
|
||||
let server = server_on_twin(&twin.base_url, &namespace).await;
|
||||
TwinScenarios::new(namespace.clone())
|
||||
.scenario(
|
||||
run_stage_scenario()
|
||||
.tool_call(TwinToolCall::new("fabro_run_search", json!({ "first": 5 }))),
|
||||
)
|
||||
.scenario(run_stage_scenario().text("The search was refused."))
|
||||
.load(twin)
|
||||
.await;
|
||||
let workspace = write_agent_workspace(&context, BLOCKING_HOOK);
|
||||
let run_id = run_detached_with(&context, &server, &workspace, &[
|
||||
"--provider",
|
||||
"openai",
|
||||
"--model",
|
||||
MODEL,
|
||||
]);
|
||||
|
||||
let status = wait_for_status(&server, &run_id, &["succeeded", "failed"]).await;
|
||||
let run = run_json(&server, &format!("runs/{run_id}")).await;
|
||||
assert_eq!(
|
||||
status,
|
||||
"succeeded",
|
||||
"run: {run}\nserver stderr:\n{}",
|
||||
server.stderr_text()
|
||||
);
|
||||
assert_petri_succeeded(&server, &run_id).await;
|
||||
|
||||
let calls = recorded_calls(&server, &run_id, "fabro_run_search").await;
|
||||
assert_eq!(calls.len(), 1, "{calls:?}");
|
||||
assert_eq!(calls[0].node, "work");
|
||||
assert_eq!(calls[0].payload["is_error"], true, "{:?}", calls[0].payload);
|
||||
assert_eq!(
|
||||
calls[0].payload["error_kind"], "denied",
|
||||
"{:?}",
|
||||
calls[0].payload
|
||||
);
|
||||
assert!(
|
||||
children_of(&server, &run_id).await.is_empty(),
|
||||
"the blocked tool created nothing"
|
||||
);
|
||||
|
||||
let store = server.petri_store().await;
|
||||
let reports = recorded::hook_reports(&store, &run_id, "pre_tool_use")
|
||||
.await
|
||||
.expect("the record replays");
|
||||
assert_eq!(reports.len(), 1, "one pre_tool_use report: {reports:?}");
|
||||
assert_eq!(reports[0]["node"], "work");
|
||||
let report = serde_json::to_string(&reports[0]["report"]).expect("the report serializes");
|
||||
assert!(
|
||||
report.contains("run tools are not allowed here"),
|
||||
"the report carries the block: {report}"
|
||||
);
|
||||
|
||||
let inputs = request_inputs(twin, &namespace).await;
|
||||
assert_eq!(inputs.len(), 2, "{inputs:?}");
|
||||
assert!(
|
||||
inputs[1].contains("run tools are not allowed here"),
|
||||
"the model saw the block reason: {}",
|
||||
inputs[1]
|
||||
);
|
||||
server.shutdown();
|
||||
}
|
||||
|
||||
/// A sub-agent the stage spawns inherits the run tools: the child's call
|
||||
/// runs under the stage, and Petri records it under the stage naming the
|
||||
/// parent session.
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn a_sub_agent_calls_an_inherited_run_tool_inside_a_petri_run() {
|
||||
if host_plugin().is_none() {
|
||||
return;
|
||||
}
|
||||
let context = test_context!();
|
||||
let twin = twin_openai().await;
|
||||
let namespace = format!("{}::{}", module_path!(), line!());
|
||||
let server = server_on_twin(&twin.base_url, &namespace).await;
|
||||
// The queue answers each request with its first unspent match: the
|
||||
// stage's requests carry `PROMPT` and the child's carry `TASK`, so the
|
||||
// stage spawns, then waits, then finishes, while the child searches
|
||||
// and reports, whichever order the two sessions ask in.
|
||||
TwinScenarios::new(namespace.clone())
|
||||
.scenario(
|
||||
run_stage_scenario()
|
||||
.tool_call(TwinToolCall::new("spawn_agent", json!({ "task": TASK }))),
|
||||
)
|
||||
.scenario(
|
||||
child_scenario()
|
||||
.tool_call(TwinToolCall::new("fabro_run_search", json!({ "first": 5 }))),
|
||||
)
|
||||
.scenario(child_scenario().text("Found the runs."))
|
||||
.scenario(run_stage_scenario().tool_call(TwinToolCall::new("wait", json!({}))))
|
||||
.scenario(run_stage_scenario().text("The helper searched."))
|
||||
.load(twin)
|
||||
.await;
|
||||
let workspace = write_agent_workspace(&context, "");
|
||||
let run_id = run_detached_with(&context, &server, &workspace, &[
|
||||
"--provider",
|
||||
"openai",
|
||||
"--model",
|
||||
MODEL,
|
||||
]);
|
||||
|
||||
let status = wait_for_status(&server, &run_id, &["succeeded", "failed"]).await;
|
||||
let run = run_json(&server, &format!("runs/{run_id}")).await;
|
||||
assert_eq!(
|
||||
status,
|
||||
"succeeded",
|
||||
"run: {run}\nserver stderr:\n{}",
|
||||
server.stderr_text()
|
||||
);
|
||||
assert_petri_succeeded(&server, &run_id).await;
|
||||
|
||||
let calls = recorded_calls(&server, &run_id, "fabro_run_search").await;
|
||||
assert_eq!(calls.len(), 1, "{calls:?}");
|
||||
let call = &calls[0];
|
||||
assert_eq!(call.node, "work", "recorded under the parent stage");
|
||||
assert_eq!(call.invocation, Some(InvocationId::ROOT));
|
||||
assert!(
|
||||
call.parent_session.is_some(),
|
||||
"the child's call names its parent session: {call:?}"
|
||||
);
|
||||
assert_eq!(call.payload["is_error"], false, "{:?}", call.payload);
|
||||
|
||||
let inputs = request_inputs(twin, &namespace).await;
|
||||
let child_inputs: Vec<&String> = inputs.iter().filter(|input| input.contains(TASK)).collect();
|
||||
assert_eq!(child_inputs.len(), 2, "the child asked twice: {inputs:?}");
|
||||
assert!(
|
||||
child_inputs[1].contains(&run_id),
|
||||
"the child read the search answer naming this run: {}",
|
||||
child_inputs[1]
|
||||
);
|
||||
server.shutdown();
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue