mirror of
https://github.com/hanzoai/dev.git
synced 2026-10-17 15:34:54 +00:00
Kai decides in the dev loop, and dev flow runs Decision Programs whose steps are tools, Kai or Zen.
hanzo-loop is the contract the loop applies: a tool shortlist per step, a command verdict joined with the policy on allow < ask < deny, and routing hints on the turn metadata. hanzo-kai links hanzoai/decision's control and program at ec8207d: one controller per thread answers those three through seven anchored upstream edits, and extension contributors feed it context, progress and completion. kai.toml turns it on, every operation in shadow by default. Each decision is one trace line in the shape Enso's control writes. dev flow runs a Decision Program over the repository. The shipped fix flow retrieves candidate files, Kai picks one, Zen writes a diff, the tests run, and Kai says done, retry or escalate within what the tests and the budget permit. go/dev.wasm is rebuilt because Cargo.toml, Cargo.lock and the Makefile are among its recorded inputs. Claude-Session: https://claude.ai/code/session_018Gt5TvrE2rxXuZikWSauv3
This commit is contained in:
@@ -0,0 +1,4 @@
|
||||
[net]
|
||||
# hanzoai/decision is a private git dependency: the git CLI carries the host's credentials and
|
||||
# URL rewrites, which cargo's built-in fetcher does not.
|
||||
git-fetch-with-cli = true
|
||||
Generated
+1212
-13
File diff suppressed because it is too large
Load Diff
@@ -3,6 +3,8 @@ members = [
|
||||
"crates/hanzo-dev",
|
||||
"crates/hanzo-config",
|
||||
"crates/hanzo-tui",
|
||||
"crates/hanzo-kai",
|
||||
"crates/hanzo-loop",
|
||||
"crates/protocol",
|
||||
"crates/core",
|
||||
"crates/ffi",
|
||||
|
||||
@@ -40,6 +40,37 @@ Changes go in a `hanzo-*` crate, or in `patches/upstream.json` when upstream
|
||||
hard-codes something that must be ours. A hand edit inside `upstream/` is lost
|
||||
at the next `make bump`.
|
||||
|
||||
## Kai
|
||||
|
||||
Kai decides inside the loop; Zen only writes. Two crates:
|
||||
|
||||
- `crates/hanzo-loop` is what the loop asks a controller and applies: a tool shortlist per
|
||||
step, a command verdict joined with the policy on `allow < ask < deny` (`Handle::command`
|
||||
joins again, so no controller can loosen it), and routing hints for the turn's request
|
||||
metadata. It is small and stable on purpose: `codex-core` depends on it and rebuilds when
|
||||
it changes.
|
||||
- `crates/hanzo-kai` links hanzoai/decision's `control` (Kai in process, batched) and
|
||||
`program` (the Decision Program executor), pinned by rev. `agent` is the controller per
|
||||
thread plus the extension contributors; `flow` runs `dev flow`; `trace` writes the decision
|
||||
lines; `tools` are the deterministic steps; `zen` is the only model call a flow makes.
|
||||
|
||||
The upstream delta is seven anchored edits in `patches/upstream.json`: `hanzo-loop` into
|
||||
core's manifest; the shortlist asked in `built_tools` and applied in `build_tool_router`; the
|
||||
verdict joined in the orchestrator right after the exec policy's requirement; the routing
|
||||
hints set on the turn metadata in `run_turn`; `hanzo-kai` into app-server's manifest and
|
||||
`hanzo_kai::install` before Guardian in `extensions.rs`.
|
||||
|
||||
`kai.toml` in the product home turns Kai on; without it nothing runs. Operations take Enso's
|
||||
names (`tools`, `risk`, `model`, `reasoning`, `context`, `progress`, `complete`), each with a
|
||||
`program`, a `mode` (default `shadow`), `thresholds` and `k`. Only `enforced` acts, and an
|
||||
enforced decision is waited for at most 30 s. The trace (`kai/decisions.jsonl`) is Enso's
|
||||
line shape, one line per decision, written once its outcome is known.
|
||||
|
||||
`dev flow fix --test "<cmd>"` runs `crates/hanzo-kai/flows/fix.json`: retrieval ranks the
|
||||
repository's files against the failure, Kai picks one, Zen writes a diff, the tests run, and
|
||||
Kai says done, retry or escalate within what the tests and the budget permit. A retried or
|
||||
escalated attempt is undone. `--control rule` takes Kai out of the steering.
|
||||
|
||||
## The Code fork
|
||||
|
||||
`upstream/code` is reference, not a build input. Its crates depend on its own
|
||||
|
||||
@@ -100,7 +100,7 @@ v8:
|
||||
|
||||
## fmt: format the crates Hanzo owns
|
||||
fmt:
|
||||
@$(CARGO) fmt -p hanzo-dev -p hanzo-config -p hanzo-tui -p dev-protocol -p dev-core -p dev-ffi
|
||||
@$(CARGO) fmt -p hanzo-dev -p hanzo-config -p hanzo-tui -p hanzo-kai -p hanzo-loop -p dev-protocol -p dev-core -p dev-ffi
|
||||
@$(CARGO) fmt --manifest-path crates/hanzo-upstream/Cargo.toml
|
||||
|
||||
## clean: discard build output
|
||||
|
||||
@@ -7,8 +7,7 @@ description = "Hanzo Dev config integration"
|
||||
repository = "https://github.com/hanzoai/dev"
|
||||
|
||||
[dependencies]
|
||||
serde = { workspace = true, features = ["derive"] }
|
||||
tempfile = { workspace = true }
|
||||
toml_edit = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
toml = { workspace = true }
|
||||
toml_edit = { workspace = true }
|
||||
|
||||
@@ -0,0 +1,219 @@
|
||||
//! Kai's settings: `kai.toml` in the product home.
|
||||
//!
|
||||
//! Each operation names the Decision Program it runs, the mode it runs in, the thresholds
|
||||
//! merged over the program's, and for a selection how many candidates it keeps. `shadow`
|
||||
//! records the decision, `advisory` also shows it, and only `enforced` lets it act, and then
|
||||
//! only toward the stricter outcome. A missing key is its default; every operation defaults
|
||||
//! to `shadow`. Without the file Kai does not run in the agent loop.
|
||||
|
||||
use serde::Deserialize;
|
||||
use std::collections::BTreeMap;
|
||||
use std::io;
|
||||
use std::path::Path;
|
||||
use std::path::PathBuf;
|
||||
|
||||
/// The file, under the product home.
|
||||
pub const FILE: &str = "kai.toml";
|
||||
|
||||
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Deserialize)]
|
||||
#[serde(rename_all = "lowercase")]
|
||||
pub enum Mode {
|
||||
/// Recorded; never shown, never applied.
|
||||
#[default]
|
||||
Shadow,
|
||||
/// Recorded and shown; the deterministic rule still decides.
|
||||
Advisory,
|
||||
/// Joined with the deterministic rule: it may tighten, never loosen.
|
||||
Enforced,
|
||||
}
|
||||
|
||||
impl Mode {
|
||||
pub fn name(self) -> &'static str {
|
||||
match self {
|
||||
Mode::Shadow => "shadow",
|
||||
Mode::Advisory => "advisory",
|
||||
Mode::Enforced => "enforced",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The operations Kai decides in the agent loop. The names are Enso's, so one pipeline reads
|
||||
/// both traces.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
|
||||
pub enum Operation {
|
||||
/// Which tools the model sees directly.
|
||||
Tools,
|
||||
/// Whether a command runs, asks first or is refused.
|
||||
Risk,
|
||||
/// Which model tier serves the turn.
|
||||
Model,
|
||||
/// How much the serving model reasons.
|
||||
Reasoning,
|
||||
/// Which files the turn starts from.
|
||||
Context,
|
||||
/// Whether the agent is advancing.
|
||||
Progress,
|
||||
/// Whether the request is done.
|
||||
Complete,
|
||||
}
|
||||
|
||||
impl Operation {
|
||||
pub const ALL: [Operation; 7] = [
|
||||
Operation::Tools,
|
||||
Operation::Risk,
|
||||
Operation::Model,
|
||||
Operation::Reasoning,
|
||||
Operation::Context,
|
||||
Operation::Progress,
|
||||
Operation::Complete,
|
||||
];
|
||||
|
||||
pub fn name(self) -> &'static str {
|
||||
match self {
|
||||
Operation::Tools => "tools",
|
||||
Operation::Risk => "risk",
|
||||
Operation::Model => "model",
|
||||
Operation::Reasoning => "reasoning",
|
||||
Operation::Context => "context",
|
||||
Operation::Progress => "progress",
|
||||
Operation::Complete => "complete",
|
||||
}
|
||||
}
|
||||
|
||||
/// The shipped program an operation runs unless `kai.toml` names another.
|
||||
pub fn program(self) -> &'static str {
|
||||
match self {
|
||||
Operation::Tools => "tools.select@1",
|
||||
Operation::Risk => "agent.command-risk@1",
|
||||
Operation::Model => "router.model@1",
|
||||
Operation::Reasoning => "reasoning.budget@1",
|
||||
Operation::Context => "context.select@1",
|
||||
Operation::Progress => "agent.progress@1",
|
||||
Operation::Complete => "agent.complete@1",
|
||||
}
|
||||
}
|
||||
|
||||
/// How many candidates a selection keeps unless `kai.toml` says.
|
||||
fn k(self) -> Option<usize> {
|
||||
match self {
|
||||
Operation::Tools => Some(8),
|
||||
Operation::Context => Some(4),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// One operation's settings as written.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Deserialize)]
|
||||
#[serde(default, deny_unknown_fields)]
|
||||
struct Written {
|
||||
program: Option<String>,
|
||||
mode: Mode,
|
||||
thresholds: BTreeMap<String, f64>,
|
||||
k: Option<usize>,
|
||||
}
|
||||
|
||||
/// One operation's settings, defaults applied.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct Op {
|
||||
/// A shipped program id (`agent.command-risk@1`) or a path to a program file.
|
||||
pub program: String,
|
||||
pub mode: Mode,
|
||||
/// Merged over the program's own, by question.
|
||||
pub thresholds: BTreeMap<String, f64>,
|
||||
/// Candidates a selection keeps.
|
||||
pub k: Option<usize>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, PartialEq, Deserialize)]
|
||||
#[serde(default, deny_unknown_fields)]
|
||||
struct File {
|
||||
models: Option<Vec<String>>,
|
||||
device: Option<String>,
|
||||
trace: Option<PathBuf>,
|
||||
zen: Option<String>,
|
||||
ops: BTreeMap<String, Written>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct Kai {
|
||||
/// Decision runtime checkpoints Kai answers from, in process.
|
||||
pub models: Vec<String>,
|
||||
/// `cpu`, `metal` or `cuda`.
|
||||
pub device: String,
|
||||
/// The decision trace, JSON lines, one per decision.
|
||||
pub trace: PathBuf,
|
||||
/// The Zen model a flow's generation nodes call.
|
||||
pub zen: String,
|
||||
ops: BTreeMap<&'static str, Op>,
|
||||
}
|
||||
|
||||
impl Kai {
|
||||
/// `home/kai.toml`, or `None` when there is none.
|
||||
pub fn load(home: &Path) -> io::Result<Option<Kai>> {
|
||||
match std::fs::read_to_string(home.join(FILE)) {
|
||||
Ok(text) => Kai::parse(&text, home).map(Some),
|
||||
Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None),
|
||||
Err(e) => Err(e),
|
||||
}
|
||||
}
|
||||
|
||||
/// `home/kai.toml`, or the defaults when there is none.
|
||||
pub fn load_or_default(home: &Path) -> io::Result<Kai> {
|
||||
match Kai::load(home)? {
|
||||
Some(kai) => Ok(kai),
|
||||
None => Kai::parse("", home),
|
||||
}
|
||||
}
|
||||
|
||||
/// Settings from `text`; a relative trace path is under `home`.
|
||||
pub fn parse(text: &str, home: &Path) -> io::Result<Kai> {
|
||||
let invalid =
|
||||
|e: String| io::Error::new(io::ErrorKind::InvalidData, format!("{FILE}: {e}"));
|
||||
let file: File = toml::from_str(text).map_err(|e| invalid(e.to_string()))?;
|
||||
if let Some(name) = file
|
||||
.ops
|
||||
.keys()
|
||||
.find(|name| !Operation::ALL.iter().any(|op| op.name() == name.as_str()))
|
||||
{
|
||||
let known: Vec<&str> = Operation::ALL.iter().map(|op| op.name()).collect();
|
||||
return Err(invalid(format!(
|
||||
"no operation {name:?}; the operations are {}",
|
||||
known.join(", ")
|
||||
)));
|
||||
}
|
||||
let ops = Operation::ALL
|
||||
.iter()
|
||||
.map(|op| {
|
||||
let written = file.ops.get(op.name()).cloned().unwrap_or_default();
|
||||
let settings = Op {
|
||||
program: written.program.unwrap_or_else(|| op.program().to_string()),
|
||||
mode: written.mode,
|
||||
thresholds: written.thresholds,
|
||||
k: written.k.or(op.k()),
|
||||
};
|
||||
(op.name(), settings)
|
||||
})
|
||||
.collect();
|
||||
Ok(Kai {
|
||||
models: file
|
||||
.models
|
||||
.unwrap_or_else(|| vec!["laya-agent".to_string()]),
|
||||
device: file.device.unwrap_or_else(|| "cpu".to_string()),
|
||||
trace: home.join(
|
||||
file.trace
|
||||
.unwrap_or_else(|| PathBuf::from("kai/decisions.jsonl")),
|
||||
),
|
||||
zen: file.zen.unwrap_or_else(|| "zen6-coder".to_string()),
|
||||
ops,
|
||||
})
|
||||
}
|
||||
|
||||
pub fn op(&self, op: Operation) -> &Op {
|
||||
&self.ops[op.name()]
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "kai_tests.rs"]
|
||||
mod tests;
|
||||
@@ -0,0 +1,66 @@
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn no_file_means_no_kai_in_the_loop() {
|
||||
let home = tempfile::tempdir().unwrap();
|
||||
assert_eq!(Kai::load(home.path()).unwrap(), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn every_operation_defaults_to_shadow_on_its_shipped_program() {
|
||||
let home = tempfile::tempdir().unwrap();
|
||||
let kai = Kai::parse("", home.path()).unwrap();
|
||||
for op in Operation::ALL {
|
||||
assert_eq!(kai.op(op).mode, Mode::Shadow, "{}", op.name());
|
||||
assert_eq!(kai.op(op).program, op.program());
|
||||
}
|
||||
assert_eq!(kai.op(Operation::Tools).k, Some(8));
|
||||
assert_eq!(kai.op(Operation::Context).k, Some(4));
|
||||
assert_eq!(kai.op(Operation::Risk).k, None);
|
||||
assert_eq!(kai.models, ["laya-agent"]);
|
||||
assert_eq!(kai.trace, home.path().join("kai/decisions.jsonl"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_operation_takes_its_program_mode_thresholds_and_k() {
|
||||
let home = tempfile::tempdir().unwrap();
|
||||
std::fs::write(
|
||||
home.path().join(FILE),
|
||||
r#"
|
||||
models = ["kai@main"]
|
||||
trace = "/var/log/kai.jsonl"
|
||||
|
||||
[ops.risk]
|
||||
program = "action.risk@1"
|
||||
mode = "enforced"
|
||||
thresholds = { verdict = 0.7 }
|
||||
|
||||
[ops.tools]
|
||||
mode = "advisory"
|
||||
k = 3
|
||||
"#,
|
||||
)
|
||||
.unwrap();
|
||||
let kai = Kai::load(home.path()).unwrap().unwrap();
|
||||
let risk = kai.op(Operation::Risk);
|
||||
assert_eq!(risk.program, "action.risk@1");
|
||||
assert_eq!(risk.mode, Mode::Enforced);
|
||||
assert_eq!(risk.thresholds.get("verdict"), Some(&0.7));
|
||||
assert_eq!(kai.op(Operation::Tools).mode, Mode::Advisory);
|
||||
assert_eq!(kai.op(Operation::Tools).k, Some(3));
|
||||
assert_eq!(kai.op(Operation::Model).mode, Mode::Shadow);
|
||||
assert_eq!(kai.models, ["kai@main"]);
|
||||
assert_eq!(kai.trace, PathBuf::from("/var/log/kai.jsonl"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn an_unknown_operation_or_mode_is_refused() {
|
||||
let home = tempfile::tempdir().unwrap();
|
||||
let error = Kai::parse("[ops.approve]\nmode = \"enforced\"", home.path()).unwrap_err();
|
||||
assert!(
|
||||
error.to_string().contains("no operation \"approve\""),
|
||||
"{error}"
|
||||
);
|
||||
assert!(Kai::parse("[ops.risk]\nmode = \"strict\"", home.path()).is_err());
|
||||
assert!(Kai::parse("[ops.risk]\nmodel = \"x\"", home.path()).is_err());
|
||||
}
|
||||
@@ -4,6 +4,7 @@ use std::io::Write;
|
||||
use std::path::Path;
|
||||
use std::path::PathBuf;
|
||||
|
||||
pub mod kai;
|
||||
mod provider;
|
||||
pub use provider::activate_provider;
|
||||
pub use provider::clear_hanzo_api_key;
|
||||
|
||||
@@ -16,6 +16,7 @@ workspace = true
|
||||
|
||||
[dependencies]
|
||||
hanzo-config = { path = "../hanzo-config", version = "0.7.0-dev.1" }
|
||||
hanzo-kai = { path = "../hanzo-kai", version = "0.7.0-dev.1" }
|
||||
hanzo-tui = { path = "../hanzo-tui", version = "0.7.0-dev.1" }
|
||||
anyhow = { workspace = true }
|
||||
chrono = { workspace = true }
|
||||
|
||||
@@ -113,6 +113,9 @@ enum Command {
|
||||
#[cfg(any(target_os = "macos", target_os = "windows"))]
|
||||
App(codex_cli::app_cmd::AppCommand),
|
||||
|
||||
/// Run a flow: a Decision Program over this repository, its steps tools, Kai or Zen.
|
||||
Flow(Flow),
|
||||
|
||||
/// Run a command inside the agent sandbox.
|
||||
Sandbox(HostSandboxArgs),
|
||||
|
||||
@@ -133,6 +136,43 @@ enum FeatureCommand {
|
||||
Disable { feature: String },
|
||||
}
|
||||
|
||||
/// A flow and the task it runs for.
|
||||
#[derive(Parser)]
|
||||
struct Flow {
|
||||
/// A flow file, or the name of a shipped flow (`fix`).
|
||||
program: String,
|
||||
|
||||
/// The command whose failure is the task, run in this repository.
|
||||
#[arg(long)]
|
||||
test: String,
|
||||
|
||||
/// What the change is for, in words.
|
||||
#[arg(long, default_value = "")]
|
||||
goal: String,
|
||||
|
||||
/// Most attempts.
|
||||
#[arg(long, default_value_t = 3)]
|
||||
max: u32,
|
||||
|
||||
/// Who steers: Kai within what the tests and the budget permit, or the rule alone.
|
||||
#[arg(long, value_enum, default_value_t = Steer::Kai)]
|
||||
control: Steer,
|
||||
|
||||
/// The longest one test run may take, in seconds.
|
||||
#[arg(long, default_value_t = 600)]
|
||||
timeout: u64,
|
||||
|
||||
/// Write the last attempt's Decision Package to this file.
|
||||
#[arg(long)]
|
||||
out: Option<std::path::PathBuf>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, clap::ValueEnum)]
|
||||
enum Steer {
|
||||
Kai,
|
||||
Rule,
|
||||
}
|
||||
|
||||
/// Which earlier session to act on. With neither, a picker opens.
|
||||
#[derive(Parser)]
|
||||
struct Pick {
|
||||
@@ -246,6 +286,7 @@ async fn run(paths: Arg0DispatchPaths) -> Result<()> {
|
||||
}
|
||||
#[cfg(any(target_os = "macos", target_os = "windows"))]
|
||||
Some(Command::App(cli)) => codex_cli::app_cmd::run_app(cli).await?,
|
||||
Some(Command::Flow(flow)) => run_flow(flow).await?,
|
||||
Some(Command::Sandbox(args)) => {
|
||||
run_sandbox(args, sandbox, overrides).await?;
|
||||
}
|
||||
@@ -261,6 +302,38 @@ async fn run(paths: Arg0DispatchPaths) -> Result<()> {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn run_flow(args: Flow) -> Result<()> {
|
||||
use hanzo_kai::flow;
|
||||
let program = flow::load(&args.program).map_err(anyhow::Error::msg)?;
|
||||
let setup = flow::Setup::new(
|
||||
&hanzo_config::home(),
|
||||
hanzo_config::API_BASE,
|
||||
hanzo_config::hanzo_credential(),
|
||||
)
|
||||
.map_err(anyhow::Error::msg)?;
|
||||
let options = flow::Options {
|
||||
task: serde_json::json!({"test": args.test, "goal": args.goal}),
|
||||
max: args.max,
|
||||
control: match args.control {
|
||||
Steer::Kai => flow::Control::Kai,
|
||||
Steer::Rule => flow::Control::Rule,
|
||||
},
|
||||
limit: std::time::Duration::from_secs(args.timeout),
|
||||
};
|
||||
let root = std::env::current_dir()?;
|
||||
let report = flow::run_with(setup, program, root, options)
|
||||
.await
|
||||
.map_err(anyhow::Error::msg)?;
|
||||
print!("{}", report.summary());
|
||||
if let (Some(out), Some(package)) = (args.out, &report.package) {
|
||||
std::fs::write(out, serde_json::to_vec_pretty(package)?)?;
|
||||
}
|
||||
if report.outcome == "escalate" {
|
||||
std::process::exit(1);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn run_features(command: FeatureCommand, overrides: CliConfigOverrides) -> Result<()> {
|
||||
use codex_features::FEATURES;
|
||||
let set = |feature: &str, on: bool| -> Result<()> {
|
||||
|
||||
@@ -12,7 +12,8 @@ use std::process::Output;
|
||||
/// clap only rejects a duplicate argument when it assembles that subcommand.
|
||||
const COMMANDS: &[&str] = &[
|
||||
"exec", "login", "logout", "serve", "resume", "fork", "review", "agents", "queue", "archive",
|
||||
"unarchive", "delete", "cloud", "mcp", "plugin", "doctor", "features", "sandbox", "completion",
|
||||
"unarchive", "delete", "cloud", "mcp", "plugin", "doctor", "features", "flow", "sandbox",
|
||||
"completion",
|
||||
];
|
||||
|
||||
fn dev(home: &Path, arguments: &[&str]) -> Output {
|
||||
|
||||
@@ -0,0 +1,33 @@
|
||||
[package]
|
||||
name = "hanzo-kai"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
description = "Kai in the Dev loop: decisions per turn, their trace, and flows over Decision Programs."
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
|
||||
[dependencies]
|
||||
# Kai in process (`decide`, batched) and the Decision Program executor. The repository is
|
||||
# private: .cargo/config.toml fetches it with the git CLI, which carries the host's
|
||||
# credentials.
|
||||
control = { git = "https://github.com/hanzoai/decision", rev = "ec8207d55bf4de506775f9d734053c8d9fa8a344" }
|
||||
program = { git = "https://github.com/hanzoai/decision", rev = "ec8207d55bf4de506775f9d734053c8d9fa8a344" }
|
||||
hanzo-config = { path = "../hanzo-config", version = "0.7.0-dev.1" }
|
||||
hanzo-loop = { path = "../hanzo-loop", version = "0.7.0-dev.1" }
|
||||
chrono = { workspace = true }
|
||||
codex-core = { workspace = true }
|
||||
codex-extension-api = { workspace = true }
|
||||
codex-protocol = { workspace = true }
|
||||
indexmap = { workspace = true, features = ["serde"] }
|
||||
reqwest = { workspace = true, features = ["json"] }
|
||||
serde = { workspace = true, features = ["derive"] }
|
||||
serde_json = { workspace = true }
|
||||
tokio = { workspace = true, features = ["macros", "process", "rt", "rt-multi-thread", "sync", "time"] }
|
||||
tracing = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
anyhow = { workspace = true }
|
||||
pretty_assertions = { workspace = true }
|
||||
tempfile = { workspace = true }
|
||||
@@ -0,0 +1,49 @@
|
||||
{
|
||||
"id": "flow.fix",
|
||||
"version": 1,
|
||||
"description": "A failing test: Kai picks the file among those retrieval ranks, Zen writes a patch, the tests run, and Kai decides done, retry or escalate.",
|
||||
"evidence": [
|
||||
{"id": "task", "uri": "dev:task", "description": "The task: {test, goal}."},
|
||||
{"id": "failure", "uri": "dev:test", "description": "The failing test run the attempt starts from: {passed, code, output}."},
|
||||
{"id": "last", "uri": "dev:last", "description": "The previous attempt: {patch, apply, test}; null on the first."}
|
||||
],
|
||||
"retrieve": {"graph": "dev:files", "top": 8},
|
||||
"nodes": [
|
||||
{
|
||||
"id": "pick",
|
||||
"inputs": ["failure", "last"],
|
||||
"kind": "kai",
|
||||
"model": "laya-agent",
|
||||
"scope": "option",
|
||||
"question": {
|
||||
"type": "noul",
|
||||
"instructions": "A test fails with this output. Is this file the one to change to make it pass?"
|
||||
}
|
||||
},
|
||||
{"id": "read", "inputs": ["pick"], "kind": "solver", "solver": "read"},
|
||||
{
|
||||
"id": "patch",
|
||||
"inputs": ["task", "failure", "read", "last"],
|
||||
"kind": "zen",
|
||||
"prompt": "The test command in `task` fails with the output in `failure`. The file most likely at fault is `read.path`; its content is `read.content`. When `last` is not null, it is the previous attempt: its patch and the test output after it. Write the smallest change that makes the test pass without weakening the test, as one unified diff against the repository root."
|
||||
},
|
||||
{"id": "apply", "inputs": ["patch"], "kind": "solver", "solver": "patch"},
|
||||
{"id": "test", "inputs": ["task", "apply"], "kind": "solver", "solver": "test"},
|
||||
{"id": "diff", "inputs": ["test"], "kind": "solver", "solver": "git", "params": {"args": ["diff", "--stat"]}},
|
||||
{
|
||||
"id": "next",
|
||||
"inputs": ["test", "last"],
|
||||
"kind": "kai",
|
||||
"model": "laya-agent",
|
||||
"question": {
|
||||
"type": "choice",
|
||||
"instructions": "An attempt to fix a failing test ran the tests again. What should happen next?",
|
||||
"criteria": {
|
||||
"done": "the tests pass: the fix is finished",
|
||||
"retry": "the tests still fail, and another attempt could fix them",
|
||||
"escalate": "the tests still fail, and more attempts will not help: a person must look"
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,339 @@
|
||||
use super::*;
|
||||
use crate::fake;
|
||||
use crate::fake::on;
|
||||
use pretty_assertions::assert_eq;
|
||||
use program::Kind;
|
||||
use std::sync::atomic::Ordering;
|
||||
|
||||
/// Kai for tests: `kai.toml` is `settings`, answers come from `rule`.
|
||||
fn kai(
|
||||
settings: &str,
|
||||
rule: impl Fn(&str, &program::Question, &Value) -> Vec<f64> + Send + Sync + 'static,
|
||||
) -> (tempfile::TempDir, Arc<fake::Kai>, Arc<Kai>) {
|
||||
let home = tempfile::tempdir().unwrap();
|
||||
let settings = crate::Settings::parse(settings, home.path()).unwrap();
|
||||
let ask = Arc::new(fake::Kai::new(rule));
|
||||
let kai = Arc::new(Kai::with(
|
||||
settings,
|
||||
crate::Decider::with(ask.clone() as crate::decide::Shared),
|
||||
));
|
||||
(home, ask, kai)
|
||||
}
|
||||
|
||||
fn thread(kai: &Arc<Kai>) -> (Arc<Thread>, Arc<fake::Thread>) {
|
||||
let act = Arc::new(fake::Thread::default());
|
||||
let thread = Arc::new(Thread::new(
|
||||
Arc::clone(kai),
|
||||
"enso-auto".into(),
|
||||
act.clone(),
|
||||
));
|
||||
(thread, act)
|
||||
}
|
||||
|
||||
/// Deny what removes files recursively, allow the rest; no escalation.
|
||||
fn risky(_: &str, q: &program::Question, state: &Value) -> Vec<f64> {
|
||||
match q.kind {
|
||||
Kind::Choice => {
|
||||
let rm = state["command"]
|
||||
.as_str()
|
||||
.unwrap_or_default()
|
||||
.contains("rm -rf");
|
||||
on(q, if rm { "deny" } else { "allow" }, 0.9)
|
||||
}
|
||||
_ => vec![0.95, 0.05],
|
||||
}
|
||||
}
|
||||
|
||||
fn exec(command: &str) -> Value {
|
||||
json!({"tool": "exec_command", "command": ["bash", "-lc", command], "cwd": "/repo"})
|
||||
}
|
||||
|
||||
async fn start(thread: &Arc<Thread>, turn: &str, text: &str) {
|
||||
let request = Request {
|
||||
text: text.into(),
|
||||
images: false,
|
||||
};
|
||||
thread.route(turn, request).await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn an_enforced_verdict_tightens_the_policy_and_never_loosens_it() {
|
||||
let (home, _, kai) = kai("[ops.risk]\nmode = \"enforced\"", risky);
|
||||
let (thread, _) = thread(&kai);
|
||||
start(&thread, "t1", "clean the build").await;
|
||||
let refused = thread
|
||||
.risk("t1", "c1", Verdict::Allow, exec("rm -rf /"))
|
||||
.await;
|
||||
let allowed = thread.risk("t1", "c2", Verdict::Allow, exec("ls")).await;
|
||||
let denied = thread.risk("t1", "c3", Verdict::Deny, exec("ls")).await;
|
||||
assert_eq!(
|
||||
(refused, allowed, denied),
|
||||
(Verdict::Deny, Verdict::Allow, Verdict::Deny)
|
||||
);
|
||||
|
||||
thread.finished("t1", "c1", "shell".into(), "blocked").await;
|
||||
let risk: Vec<Value> = fake::lines(&home.path().join("kai/decisions.jsonl"), 1)
|
||||
.into_iter()
|
||||
.filter(|l| l["ops"].get("risk").is_some())
|
||||
.collect();
|
||||
let line = &risk[0];
|
||||
assert_eq!(line["family"], "dev");
|
||||
assert_eq!(line["request"], "t1");
|
||||
assert_eq!(line["sku"], "enso-auto");
|
||||
assert_eq!(line["outcome"]["finish"], "blocked");
|
||||
let step = &line["ops"]["risk"];
|
||||
assert_eq!(step["program"], "agent.command-risk@1");
|
||||
assert_eq!(step["mode"], "enforced");
|
||||
assert_eq!(step["gates"], json!(["enforced"]));
|
||||
assert_eq!(step["kai"], "deny");
|
||||
assert_eq!(step["taken"], "deny");
|
||||
assert_eq!(step["applied"], true);
|
||||
assert_eq!(step["model"], "laya-agent@fake");
|
||||
assert!(step["states"][0].as_str().unwrap().starts_with("sha256:"));
|
||||
let verdict = &step["answers"][0]["verdict"];
|
||||
assert!(
|
||||
(verdict["deny"].as_f64().unwrap() - 0.9).abs() < 1e-9,
|
||||
"{verdict}"
|
||||
);
|
||||
assert!(
|
||||
!line.to_string().contains("rm -rf"),
|
||||
"a line holds hashes, never the command"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn a_shadow_verdict_is_recorded_and_changes_nothing() {
|
||||
let (home, _, kai) = kai("", risky);
|
||||
let (thread, act) = thread(&kai);
|
||||
start(&thread, "t1", "clean").await;
|
||||
let verdict = thread
|
||||
.risk("t1", "c1", Verdict::Allow, exec("rm -rf target"))
|
||||
.await;
|
||||
assert_eq!(verdict, Verdict::Allow);
|
||||
thread.finished("t1", "c1", "shell".into(), "ok").await;
|
||||
thread.turn_ended("completed", String::new());
|
||||
let lines = fake::lines(&home.path().join("kai/decisions.jsonl"), 3);
|
||||
let risk = lines
|
||||
.iter()
|
||||
.find(|l| l["ops"].get("risk").is_some())
|
||||
.unwrap();
|
||||
assert_eq!(risk["ops"]["risk"]["mode"], "shadow");
|
||||
assert_eq!(risk["ops"]["risk"]["gates"], json!(["shadow"]));
|
||||
assert_eq!(risk["ops"]["risk"]["kai"], "deny");
|
||||
assert_eq!(risk["ops"]["risk"]["taken"], "allow");
|
||||
assert_eq!(risk["ops"]["risk"]["applied"], false);
|
||||
assert!(
|
||||
act.warned.lock().unwrap().is_empty(),
|
||||
"shadow shows nothing"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn an_advisory_verdict_is_shown_and_the_policy_decides() {
|
||||
let (_home, _, kai) = kai("[ops.risk]\nmode = \"advisory\"", risky);
|
||||
let (thread, act) = thread(&kai);
|
||||
start(&thread, "t1", "clean").await;
|
||||
let verdict = thread
|
||||
.risk("t1", "c1", Verdict::Allow, exec("rm -rf target"))
|
||||
.await;
|
||||
assert_eq!(verdict, Verdict::Allow);
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
|
||||
while act.warned.lock().unwrap().is_empty() && std::time::Instant::now() < deadline {
|
||||
tokio::time::sleep(std::time::Duration::from_millis(20)).await;
|
||||
}
|
||||
let warned = act.warned.lock().unwrap().clone();
|
||||
assert_eq!(warned, ["Kai (advisory): would refuse `rm -rf target`"]);
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn an_enforced_shortlist_keeps_the_top_k_tools_once_a_turn() {
|
||||
let (home, ask, kai) = kai("[ops.tools]\nmode = \"enforced\"\nk = 2", |_, _, state| {
|
||||
let p = match state["tool"]["name"].as_str().unwrap_or_default() {
|
||||
"hanzo.git" => 0.9,
|
||||
"hanzo.fs" => 0.8,
|
||||
"hanzo.search" => 0.6,
|
||||
_ => 0.1,
|
||||
};
|
||||
vec![1.0 - p, p]
|
||||
});
|
||||
let (thread, _) = thread(&kai);
|
||||
let cards: Vec<Card> = ["hanzo.browser", "hanzo.fs", "hanzo.git", "hanzo.search"]
|
||||
.iter()
|
||||
.map(|n| Card {
|
||||
name: n.to_string(),
|
||||
description: format!("the {n} tool"),
|
||||
})
|
||||
.collect();
|
||||
let before = ask.calls.load(Ordering::SeqCst);
|
||||
let keep = thread.shortlist("t1", cards.clone()).await.unwrap();
|
||||
let again = thread.shortlist("t1", cards).await.unwrap();
|
||||
assert_eq!(
|
||||
ask.calls.load(Ordering::SeqCst) - before,
|
||||
1,
|
||||
"one decision a turn"
|
||||
);
|
||||
let expected: HashSet<String> = ["hanzo.git", "hanzo.fs"]
|
||||
.iter()
|
||||
.map(|s| s.to_string())
|
||||
.collect();
|
||||
assert_eq!(keep, expected);
|
||||
assert_eq!(again, expected);
|
||||
|
||||
thread.finished("t1", "c1", "hanzo.git".into(), "ok").await;
|
||||
thread.turn_ended("completed", String::new());
|
||||
let lines = fake::lines(&home.path().join("kai/decisions.jsonl"), 1);
|
||||
let tools = lines
|
||||
.iter()
|
||||
.find(|l| l["ops"].get("tools").is_some())
|
||||
.unwrap();
|
||||
assert_eq!(tools["ops"]["tools"]["kai"], "hanzo.git,hanzo.fs");
|
||||
assert_eq!(tools["ops"]["tools"]["applied"], true);
|
||||
assert_eq!(tools["ops"]["tools"]["states"].as_array().unwrap().len(), 4);
|
||||
assert_eq!(tools["outcome"]["result"]["called"], json!(["hanzo.git"]));
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn enforced_routing_hints_tier_and_effort() {
|
||||
let settings = "[ops.model]\nmode = \"enforced\"\n[ops.reasoning]\nmode = \"enforced\"";
|
||||
let (_home, _, kai) = kai(settings, |id, q, _| match id {
|
||||
"tier" => on(q, "large", 0.9),
|
||||
"budget" => vec![0.02, 0.03, 0.05, 0.9],
|
||||
_ => vec![0.9, 0.1],
|
||||
});
|
||||
let (thread, _) = thread(&kai);
|
||||
let hints = thread
|
||||
.route(
|
||||
"t1",
|
||||
Request {
|
||||
text: "prove the lemma".into(),
|
||||
images: false,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
let expected: BTreeMap<String, String> = [("kai.tier", "large"), ("kai.effort", "high")]
|
||||
.iter()
|
||||
.map(|(k, v)| (k.to_string(), v.to_string()))
|
||||
.collect();
|
||||
assert_eq!(hints, expected);
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn shadow_routing_sends_no_hints() {
|
||||
let (_home, _, kai) = kai("", |id, q, _| match id {
|
||||
"tier" => on(q, "large", 0.9),
|
||||
_ => on(q, &q.keys()[0], 0.9),
|
||||
});
|
||||
let (thread, _) = thread(&kai);
|
||||
let hints = thread
|
||||
.route(
|
||||
"t1",
|
||||
Request {
|
||||
text: "hello".into(),
|
||||
images: false,
|
||||
},
|
||||
)
|
||||
.await;
|
||||
assert!(hints.is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn an_unfinished_request_is_continued_twice_then_handed_back() {
|
||||
let (home, _, kai) = kai("[ops.complete]\nmode = \"enforced\"", |id, q, _| match id {
|
||||
"done" => vec![0.9, 0.1],
|
||||
"verified" => vec![0.8, 0.2],
|
||||
_ => on(q, "2", 0.8),
|
||||
});
|
||||
let (thread, act) = thread(&kai);
|
||||
start(&thread, "t1", "add the feature and its tests").await;
|
||||
thread.answered("I added the feature.".into());
|
||||
thread.idle().await;
|
||||
// A turn Kai started carries no user text: the request and the count stay.
|
||||
start(&thread, "t2", "").await;
|
||||
thread.idle().await;
|
||||
start(&thread, "t3", "").await;
|
||||
thread.idle().await;
|
||||
assert_eq!(act.resumed.lock().unwrap().len(), 2);
|
||||
assert_eq!(
|
||||
act.warned.lock().unwrap().clone(),
|
||||
["Kai: this request does not look finished after two more tries; over to you."]
|
||||
);
|
||||
let lines = fake::lines(&home.path().join("kai/decisions.jsonl"), 3);
|
||||
let taken: Vec<&str> = lines
|
||||
.iter()
|
||||
.filter_map(|l| l["ops"]["complete"]["taken"].as_str())
|
||||
.collect();
|
||||
assert_eq!(taken, ["retry", "retry", "escalate"]);
|
||||
assert_eq!(lines.last().unwrap()["outcome"]["escalated"], true);
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn a_new_request_resets_the_continuations() {
|
||||
let (_home, _, kai) = kai("[ops.complete]\nmode = \"enforced\"", |id, q, _| match id {
|
||||
"done" => vec![0.9, 0.1],
|
||||
_ => on(q, &q.keys()[0], 0.9),
|
||||
});
|
||||
let (thread, act) = thread(&kai);
|
||||
for turn in ["t1", "t2", "t3"] {
|
||||
start(&thread, turn, "a new request").await;
|
||||
thread.idle().await;
|
||||
}
|
||||
assert_eq!(act.resumed.lock().unwrap().len(), 3);
|
||||
assert!(act.warned.lock().unwrap().is_empty());
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn progress_steers_a_looping_agent_once_a_turn() {
|
||||
let (_home, _, kai) = kai("[ops.progress]\nmode = \"enforced\"", |id, q, _| match id {
|
||||
"status" => on(q, "looping", 0.85),
|
||||
_ => vec![0.9, 0.1],
|
||||
});
|
||||
let (thread, act) = thread(&kai);
|
||||
start(&thread, "t1", "fix the build").await;
|
||||
for (i, result) in ["failed", "failed", "failed", "failed"].iter().enumerate() {
|
||||
thread
|
||||
.finished("t1", &format!("c{i}"), "shell".into(), result)
|
||||
.await;
|
||||
}
|
||||
let steered = act.steered.lock().unwrap().clone();
|
||||
assert_eq!(steered.len(), 1, "{steered:?}");
|
||||
assert!(steered[0].contains("looping"), "{}", steered[0]);
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread")]
|
||||
async fn enforced_context_names_the_files_to_read_first() {
|
||||
let (_home, _, kai) = kai(
|
||||
"[ops.context]\nmode = \"enforced\"\nk = 1",
|
||||
|_, _, state| {
|
||||
let p = if state["source"] == "src/parser.rs" {
|
||||
0.9
|
||||
} else {
|
||||
0.2
|
||||
};
|
||||
vec![1.0 - p, p]
|
||||
},
|
||||
);
|
||||
let (thread, _) = thread(&kai);
|
||||
let repo = tempfile::tempdir().unwrap();
|
||||
std::fs::create_dir_all(repo.path().join("src")).unwrap();
|
||||
std::fs::write(repo.path().join("src/parser.rs"), "fn parse() {}").unwrap();
|
||||
std::fs::write(repo.path().join("src/lexer.rs"), "fn lex() { parse }").unwrap();
|
||||
let body = thread
|
||||
.context(
|
||||
"t1",
|
||||
"the parser rejects empty input".into(),
|
||||
repo.path().to_path_buf(),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(body.contains("- src/parser.rs"), "{body}");
|
||||
assert!(!body.contains("lexer"), "{body}");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_command_reads_as_its_script() {
|
||||
let state = command_state(&exec("cargo test -p foo"));
|
||||
assert_eq!(
|
||||
state,
|
||||
json!({"tool": "exec_command", "command": "cargo test -p foo", "cwd": "/repo"})
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,147 @@
|
||||
//! Kai answering: the decision runtime's checkpoints in process, opened on first use, or any
|
||||
//! other [`Ask`] (a fake, in tests).
|
||||
|
||||
use control::Ask;
|
||||
use control::Call;
|
||||
use control::Decision;
|
||||
use control::Row;
|
||||
use hanzo_config::kai::Mode;
|
||||
use hanzo_config::kai::Op;
|
||||
use program::policy::Signal;
|
||||
use serde_json::Value;
|
||||
use std::sync::Arc;
|
||||
use std::sync::OnceLock;
|
||||
use std::time::Duration;
|
||||
|
||||
pub type Shared = Arc<dyn Ask + Send + Sync>;
|
||||
|
||||
type Open = Box<dyn Fn() -> Result<Shared, String> + Send + Sync>;
|
||||
|
||||
/// Kai, opened once, answering from any thread.
|
||||
pub struct Decider {
|
||||
open: Open,
|
||||
ask: OnceLock<Result<Shared, String>>,
|
||||
}
|
||||
|
||||
impl Decider {
|
||||
/// Kai in process: `models` are decision runtime checkpoints, loaded on the first
|
||||
/// decision, on `device`.
|
||||
pub fn open(models: Vec<String>, device: String) -> Decider {
|
||||
Decider {
|
||||
open: Box::new(move || {
|
||||
let device = control::device(&device).map_err(|e| format!("{e:#}"))?;
|
||||
let kai = control::Kai::open(&models, &device).map_err(|e| format!("{e:#}"))?;
|
||||
Ok(Arc::new(kai) as Shared)
|
||||
}),
|
||||
ask: OnceLock::new(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Any answerer.
|
||||
pub fn with(ask: Shared) -> Decider {
|
||||
let ask = OnceLock::from(Ok(ask));
|
||||
Decider {
|
||||
open: Box::new(|| Err("unreachable".into())),
|
||||
ask,
|
||||
}
|
||||
}
|
||||
|
||||
/// The answerer, opening it on first use. Blocks while it opens.
|
||||
pub fn ask(&self) -> Result<Shared, String> {
|
||||
self.ask.get_or_init(|| (self.open)()).clone()
|
||||
}
|
||||
|
||||
/// Runs `call`. Blocks: call it off the async runtime.
|
||||
pub fn decide_blocking(&self, call: &Call) -> Result<Decision, String> {
|
||||
let ask = self.ask()?;
|
||||
control::decide(ask.as_ref(), call).map_err(|e| format!("{e:#}"))
|
||||
}
|
||||
|
||||
/// Runs `call` on a blocking thread, waiting at most `wait` for it.
|
||||
pub async fn decide(self: &Arc<Self>, call: Call, wait: Duration) -> Result<Decision, String> {
|
||||
let me = Arc::clone(self);
|
||||
let run = tokio::task::spawn_blocking(move || me.decide_blocking(&call));
|
||||
match tokio::time::timeout(wait, run).await {
|
||||
Ok(Ok(decision)) => decision,
|
||||
Ok(Err(e)) => Err(format!("kai: {e}")),
|
||||
Err(_) => Err("late".into()),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn mode(mode: Mode) -> program::Mode {
|
||||
match mode {
|
||||
Mode::Shadow => program::Mode::Shadow,
|
||||
Mode::Advisory => program::Mode::Advisory,
|
||||
Mode::Enforced => program::Mode::Enforced,
|
||||
}
|
||||
}
|
||||
|
||||
/// A call for `op` over `states`: its program, mode, thresholds and k.
|
||||
pub fn call(op: &Op, states: Vec<Value>, base: Option<program::Verdict>) -> Call {
|
||||
let program = match std::fs::read_to_string(&op.program) {
|
||||
// A path names a program file; anything else is a shipped id.
|
||||
Ok(text) => serde_json::from_str(&text).unwrap_or(Value::String(op.program.clone())),
|
||||
Err(_) => Value::String(op.program.clone()),
|
||||
};
|
||||
Call {
|
||||
program,
|
||||
states,
|
||||
mode: Some(mode(op.mode)),
|
||||
thresholds: op.thresholds.iter().map(|(q, t)| (q.clone(), *t)).collect(),
|
||||
k: op.k,
|
||||
base,
|
||||
}
|
||||
}
|
||||
|
||||
/// A state's ruling, when Kai may act on it: its gate ran enforced.
|
||||
pub fn enforced(decision: &Decision, index: usize) -> Option<&control::Ruled> {
|
||||
match decision.results.get(index)? {
|
||||
Row::Ruled(r) if r.mode == program::Mode::Enforced => Some(r),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Every state ruled under an enforced gate.
|
||||
pub fn all_enforced(decision: &Decision) -> bool {
|
||||
!decision.results.is_empty()
|
||||
&& (0..decision.results.len()).all(|i| enforced(decision, i).is_some())
|
||||
}
|
||||
|
||||
/// The first state's ruling, whatever its mode.
|
||||
pub fn first(decision: &Decision) -> Option<&control::Ruled> {
|
||||
match decision.results.first()? {
|
||||
Row::Ruled(r) => Some(r),
|
||||
Row::Failed { .. } => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// A signal's answer when it is accepted (a choice or score) or holds (a noul).
|
||||
pub fn accepted(signal: &Signal) -> Option<&str> {
|
||||
let yes = signal.accepted.or(signal.holds)?;
|
||||
yes.then_some(signal.answer.as_str())
|
||||
}
|
||||
|
||||
pub fn verdict(v: program::Verdict) -> hanzo_loop::Verdict {
|
||||
match v {
|
||||
program::Verdict::Allow => hanzo_loop::Verdict::Allow,
|
||||
program::Verdict::Ask => hanzo_loop::Verdict::Ask,
|
||||
program::Verdict::Deny => hanzo_loop::Verdict::Deny,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn policy(v: hanzo_loop::Verdict) -> program::Verdict {
|
||||
match v {
|
||||
hanzo_loop::Verdict::Allow => program::Verdict::Allow,
|
||||
hanzo_loop::Verdict::Ask => program::Verdict::Ask,
|
||||
hanzo_loop::Verdict::Deny => program::Verdict::Deny,
|
||||
}
|
||||
}
|
||||
|
||||
/// `text` cut to at most `n` characters.
|
||||
pub fn cut(text: &str, n: usize) -> String {
|
||||
match text.char_indices().nth(n) {
|
||||
Some((at, _)) => text[..at].to_string(),
|
||||
None => text.to_string(),
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
//! Fakes for tests: Kai answering by a rule, Zen answering by a rule, a thread that records
|
||||
//! what Kai did to it.
|
||||
|
||||
use crate::agent::Act;
|
||||
use control::Answers;
|
||||
use control::Job;
|
||||
use hanzo_loop::Answer;
|
||||
use program::Question;
|
||||
use program::Revision;
|
||||
use serde_json::Value;
|
||||
use std::sync::Mutex;
|
||||
use std::sync::atomic::AtomicUsize;
|
||||
use std::sync::atomic::Ordering;
|
||||
|
||||
/// The calibration the shipped agent programs pin: an answer under another runs in shadow.
|
||||
pub const CALIBRATION: &str = "cal_6dbb072f8109a222";
|
||||
|
||||
type Rule = Box<dyn Fn(&str, &Question, &Value) -> Vec<f64> + Send + Sync>;
|
||||
|
||||
/// Kai answering each question by `rule(question id, question, state)`.
|
||||
pub struct Kai {
|
||||
rule: Rule,
|
||||
pub calls: AtomicUsize,
|
||||
}
|
||||
|
||||
impl Kai {
|
||||
pub fn new(rule: impl Fn(&str, &Question, &Value) -> Vec<f64> + Send + Sync + 'static) -> Kai {
|
||||
Kai {
|
||||
rule: Box::new(rule),
|
||||
calls: AtomicUsize::new(0),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl control::Ask for Kai {
|
||||
fn ask(&self, _model: &str, jobs: &[Job]) -> anyhow::Result<Vec<anyhow::Result<Answers>>> {
|
||||
self.calls.fetch_add(1, Ordering::SeqCst);
|
||||
Ok(jobs
|
||||
.iter()
|
||||
.map(|job| {
|
||||
Ok(job
|
||||
.questions
|
||||
.iter()
|
||||
.map(|(id, q)| (id.clone(), (self.rule)(id, q, &job.state)))
|
||||
.collect())
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
fn revision(&self, model: &str) -> anyhow::Result<Revision> {
|
||||
Ok(Revision {
|
||||
model: format!("{model}@fake"),
|
||||
calibration: Some(CALIBRATION.into()),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Probabilities putting `p` on `label` of `q`'s options and the rest evenly.
|
||||
pub fn on(q: &Question, label: &str, p: f64) -> Vec<f64> {
|
||||
let keys = q.keys();
|
||||
let rest = (1.0 - p) / (keys.len() as f64 - 1.0).max(1.0);
|
||||
keys.iter()
|
||||
.map(|k| if k == label { p } else { rest })
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// Zen answering by `rule(prompt, context)`.
|
||||
pub struct Zen(pub Box<dyn Fn(&str, &Value) -> String + Send + Sync>);
|
||||
|
||||
impl program::Zen for Zen {
|
||||
fn generate(&self, prompt: &str, context: &Value) -> program::Result<String> {
|
||||
Ok((self.0)(prompt, context))
|
||||
}
|
||||
|
||||
fn revision(&self) -> Revision {
|
||||
Revision {
|
||||
model: "zen@fake".into(),
|
||||
calibration: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// What Kai did to a thread.
|
||||
#[derive(Default)]
|
||||
pub struct Thread {
|
||||
pub warned: Mutex<Vec<String>>,
|
||||
pub steered: Mutex<Vec<String>>,
|
||||
pub resumed: Mutex<Vec<String>>,
|
||||
}
|
||||
|
||||
impl Act for Thread {
|
||||
fn warn(&self, _turn: &str, message: String) {
|
||||
self.warned.lock().unwrap().push(message);
|
||||
}
|
||||
|
||||
fn steer(&self, text: String) -> Answer<'_, ()> {
|
||||
self.steered.lock().unwrap().push(text);
|
||||
Box::pin(async {})
|
||||
}
|
||||
|
||||
fn resume(&self, text: String) -> Answer<'_, bool> {
|
||||
self.resumed.lock().unwrap().push(text);
|
||||
Box::pin(async { true })
|
||||
}
|
||||
}
|
||||
|
||||
/// The trace's lines, once there are `n`, waiting at most five seconds for them.
|
||||
pub fn lines(path: &std::path::Path, n: usize) -> Vec<Value> {
|
||||
let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
|
||||
loop {
|
||||
let lines: Vec<Value> = std::fs::read_to_string(path)
|
||||
.unwrap_or_default()
|
||||
.lines()
|
||||
.map(|l| serde_json::from_str(l).expect("a trace line is JSON"))
|
||||
.collect();
|
||||
if lines.len() >= n || std::time::Instant::now() > deadline {
|
||||
return lines;
|
||||
}
|
||||
std::thread::sleep(std::time::Duration::from_millis(20));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,628 @@
|
||||
//! Flows: a Decision Program run over a repository, attempt after attempt.
|
||||
//!
|
||||
//! A flow is a Decision Program in the `program` crate's schema, run by its executor. Its
|
||||
//! solver nodes are Dev's deterministic tools (`search`, `read`, `patch`, `test`, `git`), its
|
||||
//! kai nodes are Kai, its zen nodes are Zen, and nothing else calls a model. Evidence is
|
||||
//! resolved by URI: `dev:task` is the task given on the command line, `dev:test` the failing
|
||||
//! test run the attempt starts from, `dev:last` the previous attempt (null on the first) and
|
||||
//! `git:status` HEAD with the working diff. A program that retrieves its options from
|
||||
//! `dev:files` is bound, each attempt, to the repository files that share the most words with
|
||||
//! its evidence: Kai chooses among those, never among every file.
|
||||
//!
|
||||
//! Two nodes steer the loop: `test`, whose output says whether the tests pass, and `next`, a
|
||||
//! Kai choice over `done`, `retry` and `escalate`. Under Kai control Kai's answer acts when
|
||||
//! its probability reaches [`CERTAIN`] and the tests and the budget permit it: `done` only once
|
||||
//! the tests pass, `retry` only while attempts remain, `escalate` always. Under rule control
|
||||
//! the tests and the budget decide alone. A `read` node reads the file Kai rated most likely
|
||||
//! under Kai control, and the best retrieval match under rule control.
|
||||
//!
|
||||
//! An attempt that is retried or escalated is undone, so the next starts from the original
|
||||
//! tree and a flow that ends anywhere but `done` leaves the tree as it found it. Attempts
|
||||
//! share the node cache: a node whose inputs did not change is not run again.
|
||||
|
||||
use crate::decide::Decider;
|
||||
use crate::tools;
|
||||
use crate::trace::Line;
|
||||
use crate::trace::Step;
|
||||
use crate::trace::Trace;
|
||||
use indexmap::IndexMap;
|
||||
use program::Package;
|
||||
use program::Program;
|
||||
use program::Question;
|
||||
use program::Revision;
|
||||
use program::Snapshot;
|
||||
use program::program::Alternative;
|
||||
use serde_json::Value;
|
||||
use serde_json::json;
|
||||
use std::path::Path;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::time::Duration;
|
||||
|
||||
/// The probability at which Kai's answer to `next` acts.
|
||||
pub const CERTAIN: f64 = 0.5;
|
||||
|
||||
/// The flows Dev ships, by name.
|
||||
pub const SHIPPED: &[(&str, &str)] = &[("fix", include_str!("../flows/fix.json"))];
|
||||
|
||||
/// A shipped flow by name, or a flow file.
|
||||
pub fn load(name: &str) -> Result<Program, String> {
|
||||
let text = match SHIPPED.iter().find(|(n, _)| *n == name) {
|
||||
Some((_, text)) => (*text).to_string(),
|
||||
None => std::fs::read_to_string(name).map_err(|e| format!("{name}: {e}"))?,
|
||||
};
|
||||
Program::parse(&text).map_err(|e| format!("{name}: {e}"))
|
||||
}
|
||||
|
||||
/// Who steers the loop and picks the file: Kai, or the tests, the budget and the ranking.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub enum Control {
|
||||
Rule,
|
||||
Kai,
|
||||
}
|
||||
|
||||
pub struct Options {
|
||||
/// `dev:task`: `{test, goal}`.
|
||||
pub task: Value,
|
||||
/// Most attempts.
|
||||
pub max: u32,
|
||||
pub control: Control,
|
||||
/// The longest one test run may take.
|
||||
pub limit: Duration,
|
||||
}
|
||||
|
||||
/// One attempt, as the report shows it.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct Attempt {
|
||||
pub n: u32,
|
||||
/// The file read, and who picked it.
|
||||
pub file: Option<String>,
|
||||
pub by: String,
|
||||
pub applied: bool,
|
||||
pub passed: bool,
|
||||
/// Kai's answer to `next` and its probability.
|
||||
pub kai: Option<(String, f64)>,
|
||||
pub rule: String,
|
||||
pub taken: String,
|
||||
/// Nodes run, of all; the rest came from the cache.
|
||||
pub ran: usize,
|
||||
pub nodes: usize,
|
||||
}
|
||||
|
||||
pub struct Report {
|
||||
pub attempts: Vec<Attempt>,
|
||||
/// `done`, `escalate`, or `passing` when the tests passed before any attempt.
|
||||
pub outcome: String,
|
||||
/// The last attempt's Decision Package.
|
||||
pub package: Option<Package>,
|
||||
/// The last attempt's patch, as Zen wrote it.
|
||||
pub patch: String,
|
||||
}
|
||||
|
||||
impl Report {
|
||||
pub fn summary(&self) -> String {
|
||||
let mut out = String::new();
|
||||
if self.outcome == "passing" {
|
||||
out.push_str("The tests already pass; nothing to do.\n");
|
||||
return out;
|
||||
}
|
||||
for a in &self.attempts {
|
||||
let kai = a
|
||||
.kai
|
||||
.as_ref()
|
||||
.map(|(answer, p)| format!("{answer} ({p:.2})"))
|
||||
.unwrap_or_else(|| "none".into());
|
||||
out.push_str(&format!(
|
||||
"attempt {}: file {} (by {}), patch {}, tests {}; next: kai {kai}, rule {} -> {}; {} of {} nodes ran\n",
|
||||
a.n,
|
||||
a.file.as_deref().unwrap_or("-"),
|
||||
a.by,
|
||||
if a.applied { "applied" } else { "not applied" },
|
||||
if a.passed { "pass" } else { "fail" },
|
||||
a.rule,
|
||||
a.taken,
|
||||
a.ran,
|
||||
a.nodes,
|
||||
));
|
||||
}
|
||||
out.push_str(&format!("outcome: {}\n", self.outcome));
|
||||
if self.outcome == "escalate" && !self.patch.is_empty() {
|
||||
out.push_str("the tree is as it was; the last patch tried:\n");
|
||||
out.push_str(&tools::diff_of(&self.patch));
|
||||
}
|
||||
out
|
||||
}
|
||||
}
|
||||
|
||||
/// Kai answering a flow's kai nodes, one question at a time.
|
||||
pub struct Judge(pub Arc<Decider>);
|
||||
|
||||
impl program::Judge for Judge {
|
||||
fn judge(&self, model: &str, question: &Question, state: &Value) -> program::Result<Vec<f64>> {
|
||||
let backend = |e: String| program::Error::Backend(e);
|
||||
let ask = self.0.ask().map_err(backend)?;
|
||||
let job = control::Job {
|
||||
state: state.clone(),
|
||||
questions: [("q".to_string(), question.clone())].into_iter().collect(),
|
||||
};
|
||||
let mut out = ask
|
||||
.ask(model, &[job])
|
||||
.map_err(|e| backend(format!("{e:#}")))?;
|
||||
let answers = out
|
||||
.pop()
|
||||
.ok_or_else(|| backend("no answer".into()))?
|
||||
.map_err(|e| backend(format!("{e:#}")))?;
|
||||
answers
|
||||
.into_values()
|
||||
.next()
|
||||
.ok_or_else(|| backend("no answer".into()))
|
||||
}
|
||||
|
||||
fn revision(&self, model: &str) -> program::Result<Revision> {
|
||||
let ask = self.0.ask().map_err(program::Error::Backend)?;
|
||||
ask.revision(model)
|
||||
.map_err(|e| program::Error::Backend(format!("{e:#}")))
|
||||
}
|
||||
}
|
||||
|
||||
/// Dev's deterministic tools as a flow's solvers.
|
||||
struct Tools<'a> {
|
||||
root: &'a Path,
|
||||
control: Control,
|
||||
options: &'a Options,
|
||||
}
|
||||
|
||||
impl Tools<'_> {
|
||||
/// The file to read: Kai's likeliest under Kai control, the best match under rule control.
|
||||
fn read(&self, context: &Value) -> Result<Value, String> {
|
||||
let options = context["options"].as_array().cloned().unwrap_or_default();
|
||||
if options.is_empty() {
|
||||
return Err("read: no candidate files".into());
|
||||
}
|
||||
// A kai node's output keyed by option id, `{id: {"false": p, "true": q}}`.
|
||||
let rated = context["inputs"]
|
||||
.as_object()
|
||||
.into_iter()
|
||||
.flat_map(|m| m.values())
|
||||
.find(|v| {
|
||||
options
|
||||
.iter()
|
||||
.all(|o| v.get(o["id"].as_str().unwrap_or_default()).is_some())
|
||||
});
|
||||
let (option, by, p) = match (self.control, rated) {
|
||||
(Control::Kai, Some(rated)) => {
|
||||
let p = |o: &Value| {
|
||||
rated[o["id"].as_str().unwrap_or_default()]["true"]
|
||||
.as_f64()
|
||||
.unwrap_or(0.0)
|
||||
};
|
||||
let Some(best) = options.iter().max_by(|a, b| p(a).total_cmp(&p(b))) else {
|
||||
return Err("read: no candidate files".into());
|
||||
};
|
||||
(best, "kai", Some(p(best)))
|
||||
}
|
||||
_ => (&options[0], "rule", None),
|
||||
};
|
||||
let path = option["label"].as_str().unwrap_or_default();
|
||||
let content = tools::head(self.root, path, tools::HEAD)
|
||||
.ok_or_else(|| format!("read: {path} is not a text file"))?;
|
||||
Ok(json!({"path": path, "content": content, "by": by, "p": p}))
|
||||
}
|
||||
}
|
||||
|
||||
/// The text of a node's inputs: strings as they are, anything else as JSON.
|
||||
fn text(value: &Value) -> String {
|
||||
match value {
|
||||
Value::String(s) => s.clone(),
|
||||
Value::Null => String::new(),
|
||||
Value::Object(m) => m.values().map(text).collect::<Vec<_>>().join("\n"),
|
||||
Value::Array(a) => a.iter().map(text).collect::<Vec<_>>().join("\n"),
|
||||
other => other.to_string(),
|
||||
}
|
||||
}
|
||||
|
||||
impl program::Solver for Tools<'_> {
|
||||
fn solve(&self, solver: &str, params: &Value, context: &Value) -> program::Result<Value> {
|
||||
let inputs = &context["inputs"];
|
||||
let out = match solver {
|
||||
"read" => self.read(context),
|
||||
"patch" => {
|
||||
let answer = inputs
|
||||
.as_object()
|
||||
.into_iter()
|
||||
.flat_map(|m| m.values())
|
||||
.find_map(Value::as_str)
|
||||
.unwrap_or_default();
|
||||
Ok(tools::patch(self.root, answer))
|
||||
}
|
||||
"test" => {
|
||||
let command = params["command"]
|
||||
.as_str()
|
||||
.or_else(|| self.options.task["test"].as_str())
|
||||
.unwrap_or_default();
|
||||
if command.is_empty() {
|
||||
Err("test: no test command".to_string())
|
||||
} else {
|
||||
Ok(tools::test(self.root, command, self.options.limit))
|
||||
}
|
||||
}
|
||||
"git" => {
|
||||
let args: Vec<String> = params["args"]
|
||||
.as_array()
|
||||
.into_iter()
|
||||
.flatten()
|
||||
.filter_map(|a| a.as_str().map(str::to_string))
|
||||
.collect();
|
||||
tools::git(self.root, &args)
|
||||
}
|
||||
"search" => {
|
||||
let query = params["query"]
|
||||
.as_str()
|
||||
.map(str::to_string)
|
||||
.unwrap_or_else(|| text(inputs));
|
||||
let top = params["top"].as_u64().unwrap_or(8) as usize;
|
||||
let hits = tools::search(self.root, &query, top, tools::HEAD);
|
||||
Ok(json!({"files": hits
|
||||
.iter()
|
||||
.map(|h| json!({"path": h.path, "score": h.score}))
|
||||
.collect::<Vec<_>>()}))
|
||||
}
|
||||
other => Err(format!(
|
||||
"no tool {other:?}; a flow's tools are read, patch, test, git and search"
|
||||
)),
|
||||
};
|
||||
out.map_err(program::Error::Backend)
|
||||
}
|
||||
|
||||
fn revision(&self, solver: &str) -> program::Result<Revision> {
|
||||
Ok(Revision {
|
||||
model: format!("dev.{solver}@1"),
|
||||
calibration: None,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// The evidence a flow reads, by URI.
|
||||
fn evidence(
|
||||
uri: &str,
|
||||
task: &Value,
|
||||
failure: &Value,
|
||||
last: &Value,
|
||||
root: &Path,
|
||||
) -> Result<Value, String> {
|
||||
Ok(match uri {
|
||||
"dev:task" => task.clone(),
|
||||
"dev:test" => failure.clone(),
|
||||
"dev:last" => last.clone(),
|
||||
"git:status" => {
|
||||
let head = tools::git(root, &["rev-parse".into(), "HEAD".into()])?;
|
||||
let diff = tools::git(root, &["diff".into()])?;
|
||||
json!({"head": head["output"].as_str().unwrap_or_default().trim(), "diff": diff["output"]})
|
||||
}
|
||||
other => {
|
||||
return Err(format!(
|
||||
"no evidence source {other:?}; a flow reads dev:task, dev:test, dev:last and git:status"
|
||||
));
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
/// A Kai node's distribution: its likeliest label and probability.
|
||||
fn likeliest(dist: &Value) -> Option<(String, f64)> {
|
||||
dist.as_object()?
|
||||
.iter()
|
||||
.filter_map(|(k, v)| Some((k.clone(), v.as_f64()?)))
|
||||
.max_by(|a, b| a.1.total_cmp(&b.1))
|
||||
}
|
||||
|
||||
/// Runs `flow` over the repository at `root`. Blocks: call it off the async runtime.
|
||||
pub fn run(
|
||||
flow: &Program,
|
||||
root: &Path,
|
||||
options: &Options,
|
||||
judge: &dyn program::Judge,
|
||||
zen: &dyn program::Zen,
|
||||
trace: &Trace,
|
||||
) -> Result<Report, String> {
|
||||
for id in ["test", "next"] {
|
||||
if flow.node(id).is_none() {
|
||||
return Err(format!("{}: a flow needs a node {id:?}", flow.id));
|
||||
}
|
||||
}
|
||||
let command = options.task["test"].as_str().unwrap_or_default();
|
||||
if command.is_empty() {
|
||||
return Err("the task names no test command".into());
|
||||
}
|
||||
let failure = tools::test(root, command, options.limit);
|
||||
if failure["passed"] == true {
|
||||
return Ok(Report {
|
||||
attempts: Vec::new(),
|
||||
outcome: "passing".into(),
|
||||
package: None,
|
||||
patch: String::new(),
|
||||
});
|
||||
}
|
||||
let solvers = Tools {
|
||||
root,
|
||||
control: options.control,
|
||||
options,
|
||||
};
|
||||
let absent = program::Absent::default();
|
||||
let rt = program::Runtime {
|
||||
judge,
|
||||
zen,
|
||||
solver: &solvers,
|
||||
human: &absent,
|
||||
};
|
||||
let run_id = crate::trace::now();
|
||||
let mut cache = program::Cache::default();
|
||||
let mut last = Value::Null;
|
||||
let mut report = Report {
|
||||
attempts: Vec::new(),
|
||||
outcome: String::new(),
|
||||
package: None,
|
||||
patch: String::new(),
|
||||
};
|
||||
for n in 1..=options.max.max(1) {
|
||||
let time = chrono::Utc::now().timestamp();
|
||||
let mut snapshots = IndexMap::new();
|
||||
for e in &flow.evidence {
|
||||
let content = evidence(&e.uri, &options.task, &failure, &last, root)?;
|
||||
snapshots.insert(e.id.clone(), Snapshot { content, time });
|
||||
}
|
||||
let bound = match &flow.retrieve {
|
||||
Some(r) if r.graph == "dev:files" => {
|
||||
let query = snapshots
|
||||
.values()
|
||||
.map(|s| text(&s.content))
|
||||
.collect::<Vec<_>>()
|
||||
.join("\n");
|
||||
let options: Vec<Alternative> =
|
||||
tools::search(root, &query, r.top as usize, tools::HEAD)
|
||||
.into_iter()
|
||||
.enumerate()
|
||||
.map(|(i, hit)| Alternative {
|
||||
id: format!("f{i}"),
|
||||
label: hit.path,
|
||||
attributes: [("score".to_string(), hit.score)].into_iter().collect(),
|
||||
})
|
||||
.collect();
|
||||
if options.is_empty() {
|
||||
return Err("no file in the repository shares a word with the failure".into());
|
||||
}
|
||||
flow.bind(options).map_err(|e| e.to_string())?
|
||||
}
|
||||
Some(r) => {
|
||||
return Err(format!(
|
||||
"no graph {:?}; a flow retrieves from dev:files",
|
||||
r.graph
|
||||
));
|
||||
}
|
||||
None => flow.clone(),
|
||||
};
|
||||
let package = program::execute(&bound, &snapshots, &[], &rt, &mut cache)
|
||||
.map_err(|e| e.to_string())?;
|
||||
let output = |id: &str| {
|
||||
package
|
||||
.results
|
||||
.get(id)
|
||||
.map(|o| o.output.clone())
|
||||
.unwrap_or(Value::Null)
|
||||
};
|
||||
let test = output("test");
|
||||
let passed = test["passed"] == true;
|
||||
let read = bound
|
||||
.nodes
|
||||
.iter()
|
||||
.find(|n| matches!(&n.op, program::Op::Solver(s) if s.solver == "read"))
|
||||
.map(|n| output(&n.id));
|
||||
let applied = bound
|
||||
.nodes
|
||||
.iter()
|
||||
.find(|n| matches!(&n.op, program::Op::Solver(s) if s.solver == "patch"))
|
||||
.map(|n| output(&n.id));
|
||||
let zen_answer = bound
|
||||
.nodes
|
||||
.iter()
|
||||
.find(|n| matches!(n.op, program::Op::Zen(_)))
|
||||
.map(|n| output(&n.id))
|
||||
.and_then(|v| v.as_str().map(str::to_string))
|
||||
.unwrap_or_default();
|
||||
let kai = likeliest(&output("next"));
|
||||
let rule = match (passed, n < options.max) {
|
||||
(true, _) => "done",
|
||||
(false, true) => "retry",
|
||||
(false, false) => "escalate",
|
||||
};
|
||||
let permitted: &[&str] = match rule {
|
||||
"done" => &["done", "escalate"],
|
||||
"retry" => &["retry", "escalate"],
|
||||
_ => &["escalate"],
|
||||
};
|
||||
let taken = match (&kai, options.control) {
|
||||
(Some((answer, p)), Control::Kai)
|
||||
if *p >= CERTAIN && permitted.contains(&answer.as_str()) =>
|
||||
{
|
||||
answer.clone()
|
||||
}
|
||||
_ => rule.to_string(),
|
||||
};
|
||||
let was_applied = applied.as_ref().is_some_and(|a| a["applied"] == true);
|
||||
let attempt = Attempt {
|
||||
n,
|
||||
file: read
|
||||
.as_ref()
|
||||
.and_then(|r| r["path"].as_str().map(str::to_string)),
|
||||
by: read
|
||||
.as_ref()
|
||||
.and_then(|r| r["by"].as_str().map(str::to_string))
|
||||
.unwrap_or_default(),
|
||||
applied: was_applied,
|
||||
passed,
|
||||
kai: kai.clone(),
|
||||
rule: rule.into(),
|
||||
taken: taken.clone(),
|
||||
ran: package
|
||||
.trace
|
||||
.iter()
|
||||
.filter(|s| s.source == program::Source::Run)
|
||||
.count(),
|
||||
nodes: package.trace.len(),
|
||||
};
|
||||
trace.write(&line(
|
||||
flow,
|
||||
&bound,
|
||||
&package,
|
||||
&attempt,
|
||||
&run_id,
|
||||
options.control,
|
||||
));
|
||||
report.attempts.push(attempt);
|
||||
report.patch = zen_answer.clone();
|
||||
report.package = Some(package);
|
||||
if taken != "done" && was_applied {
|
||||
// Undo the attempt: the next starts from the tree as the flow found it.
|
||||
tools::revert(root, &zen_answer);
|
||||
}
|
||||
match taken.as_str() {
|
||||
"retry" => {
|
||||
last =
|
||||
json!({"patch": tools::diff_of(&zen_answer), "apply": applied, "test": test});
|
||||
}
|
||||
_ => {
|
||||
report.outcome = taken;
|
||||
return Ok(report);
|
||||
}
|
||||
}
|
||||
}
|
||||
report.outcome = "escalate".into();
|
||||
Ok(report)
|
||||
}
|
||||
|
||||
/// An attempt's trace line: Kai's pick and its `next`, with what was done and what followed.
|
||||
fn line(
|
||||
flow: &Program,
|
||||
bound: &Program,
|
||||
package: &Package,
|
||||
attempt: &Attempt,
|
||||
run: &str,
|
||||
control: Control,
|
||||
) -> Line {
|
||||
let mode = match control {
|
||||
Control::Kai => "enforced",
|
||||
Control::Rule => "shadow",
|
||||
};
|
||||
let mut ops = std::collections::BTreeMap::new();
|
||||
for node in &bound.nodes {
|
||||
let program::Op::Kai(q) = &node.op else {
|
||||
continue;
|
||||
};
|
||||
let Some(out) = package.results.get(&node.id) else {
|
||||
continue;
|
||||
};
|
||||
let mut step = Step::new(&format!("{}@{}#{}", flow.id, flow.version, node.id), mode);
|
||||
step.hash = bound.hash();
|
||||
step.model = out
|
||||
.revision
|
||||
.as_ref()
|
||||
.map(|r| r.model.clone())
|
||||
.unwrap_or_default();
|
||||
match q.scope {
|
||||
program::program::Scope::Option => {
|
||||
for o in &bound.options {
|
||||
step.states
|
||||
.push(program::canon::hash(&json!({"source": o.label})));
|
||||
step.gates.push(mode.into());
|
||||
let mut answers = IndexMap::new();
|
||||
answers.insert(node.id.clone(), out.output[&o.id].clone());
|
||||
step.answers.push(answers);
|
||||
}
|
||||
step.kai = bound
|
||||
.options
|
||||
.iter()
|
||||
.max_by(|a, b| {
|
||||
let p = |o: &Alternative| out.output[&o.id]["true"].as_f64().unwrap_or(0.0);
|
||||
p(a).total_cmp(&p(b))
|
||||
})
|
||||
.map(|o| o.label.clone())
|
||||
.unwrap_or_default();
|
||||
step.taken = attempt.file.clone().unwrap_or_default();
|
||||
step.applied = attempt.by == "kai";
|
||||
}
|
||||
program::program::Scope::Program => {
|
||||
step.states.push(package.results[&node.id].inputs.clone());
|
||||
step.gates.push(mode.into());
|
||||
let mut answers = IndexMap::new();
|
||||
answers.insert(node.id.clone(), out.output.clone());
|
||||
step.answers.push(answers);
|
||||
step.kai = likeliest(&out.output).map(|(a, _)| a).unwrap_or_default();
|
||||
if node.id == "next" {
|
||||
step.taken = attempt.taken.clone();
|
||||
step.applied = attempt
|
||||
.kai
|
||||
.as_ref()
|
||||
.is_some_and(|(a, _)| *a == attempt.taken)
|
||||
&& attempt.taken != attempt.rule;
|
||||
}
|
||||
}
|
||||
}
|
||||
ops.insert(node.id.clone(), step);
|
||||
}
|
||||
Line {
|
||||
time: crate::trace::now(),
|
||||
request: format!("{run}#{}", attempt.n),
|
||||
family: "dev.flow".into(),
|
||||
sku: format!("{}@{}", flow.id, flow.version),
|
||||
ms: package.trace.iter().map(|s| s.ms).sum(),
|
||||
outcome: crate::trace::Outcome {
|
||||
status: i64::from(!attempt.passed),
|
||||
finish: attempt.taken.clone(),
|
||||
escalated: attempt.taken == "escalate",
|
||||
result: json!({"passed": attempt.passed, "applied": attempt.applied, "attempt": attempt.n}),
|
||||
..Default::default()
|
||||
},
|
||||
ops,
|
||||
..Line::default()
|
||||
}
|
||||
}
|
||||
|
||||
/// A flow's backends and settings from the profile at `home`: Kai in process, Zen through
|
||||
/// the Hanzo API.
|
||||
pub struct Setup {
|
||||
pub decider: Arc<Decider>,
|
||||
pub zen: crate::zen::Zen,
|
||||
pub trace: Trace,
|
||||
}
|
||||
|
||||
impl Setup {
|
||||
pub fn new(home: &Path, base: &str, key: Option<String>) -> Result<Setup, String> {
|
||||
let settings = hanzo_config::kai::Kai::load_or_default(home).map_err(|e| e.to_string())?;
|
||||
Ok(Setup {
|
||||
decider: Arc::new(Decider::open(
|
||||
settings.models.clone(),
|
||||
settings.device.clone(),
|
||||
)),
|
||||
zen: crate::zen::Zen::new(base, key, &settings.zen),
|
||||
trace: Trace::new(settings.trace),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Runs `flow` at `root` with Kai and Zen from `setup`, off the async runtime.
|
||||
pub async fn run_with(
|
||||
setup: Setup,
|
||||
flow: Program,
|
||||
root: PathBuf,
|
||||
options: Options,
|
||||
) -> Result<Report, String> {
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let judge = Judge(Arc::clone(&setup.decider));
|
||||
run(&flow, &root, &options, &judge, &setup.zen, &setup.trace)
|
||||
})
|
||||
.await
|
||||
.map_err(|e| e.to_string())?
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[path = "flow_tests.rs"]
|
||||
mod tests;
|
||||
@@ -0,0 +1,276 @@
|
||||
use super::*;
|
||||
use crate::fake;
|
||||
use crate::fake::on;
|
||||
use pretty_assertions::assert_eq;
|
||||
use program::Kind;
|
||||
use std::process::Command;
|
||||
|
||||
const BUGGY: &str = "echo $(( $1 - $2 ))\n";
|
||||
const FIXED: &str = "echo $(( $1 + $2 ))\n";
|
||||
|
||||
/// A repository whose test fails: `src/add.sh` subtracts.
|
||||
fn repo() -> tempfile::TempDir {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let root = dir.path();
|
||||
std::fs::create_dir_all(root.join("src")).unwrap();
|
||||
std::fs::write(root.join("src/add.sh"), BUGGY).unwrap();
|
||||
std::fs::write(root.join("src/mul.sh"), "echo $(( $1 * $2 ))\n").unwrap();
|
||||
std::fs::write(root.join("README.md"), "Shell arithmetic.\n").unwrap();
|
||||
std::fs::write(
|
||||
root.join("test.sh"),
|
||||
"r=$(sh src/add.sh 2 3)\n[ \"$r\" = 5 ] || { echo \"FAIL src/add.sh: add 2 3 gave $r, want 5\"; exit 1; }\necho ok\n",
|
||||
)
|
||||
.unwrap();
|
||||
let git = |args: &[&str]| {
|
||||
let ok = Command::new("git")
|
||||
.args([
|
||||
"-c",
|
||||
"user.name=t",
|
||||
"-c",
|
||||
"user.email=t@t",
|
||||
"-c",
|
||||
"commit.gpgsign=false",
|
||||
])
|
||||
.args(args)
|
||||
.current_dir(root)
|
||||
.output()
|
||||
.unwrap()
|
||||
.status
|
||||
.success();
|
||||
assert!(ok, "git {args:?}");
|
||||
};
|
||||
git(&["init", "-q"]);
|
||||
git(&["add", "."]);
|
||||
git(&["commit", "-q", "-m", "start"]);
|
||||
dir
|
||||
}
|
||||
|
||||
fn diff(from: &str, to: &str) -> String {
|
||||
format!(
|
||||
"```diff\n--- a/src/add.sh\n+++ b/src/add.sh\n@@ -1 +1 @@\n-{}\n+{}\n```\n",
|
||||
from.trim_end(),
|
||||
to.trim_end()
|
||||
)
|
||||
}
|
||||
|
||||
/// Kai rating `src/add.sh` the file to change, and answering `next` by `next(test output)`.
|
||||
fn kai(next: impl Fn(&Value) -> (&'static str, f64) + Send + Sync + 'static) -> Judge {
|
||||
let ask = fake::Kai::new(move |_, q, state| match q.kind {
|
||||
Kind::Noul => {
|
||||
let p = if state["option"]["label"] == "src/add.sh" {
|
||||
0.9
|
||||
} else {
|
||||
0.2
|
||||
};
|
||||
vec![1.0 - p, p]
|
||||
}
|
||||
Kind::Choice => {
|
||||
let (label, p) = next(&state["test"]);
|
||||
on(q, label, p)
|
||||
}
|
||||
Kind::Score => on(q, "0", 0.9),
|
||||
});
|
||||
Judge(Arc::new(crate::Decider::with(Arc::new(ask))))
|
||||
}
|
||||
|
||||
/// Kai as an honest judge of the tests.
|
||||
fn honest() -> Judge {
|
||||
kai(|test| {
|
||||
if test["passed"] == true {
|
||||
("done", 0.9)
|
||||
} else {
|
||||
("retry", 0.8)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
fn zen(answer: impl Fn(&Value) -> String + Send + Sync + 'static) -> fake::Zen {
|
||||
fake::Zen(Box::new(move |_, context| answer(context)))
|
||||
}
|
||||
|
||||
fn options(max: u32, control: Control) -> Options {
|
||||
Options {
|
||||
task: json!({"test": "sh test.sh", "goal": "make the test pass"}),
|
||||
max,
|
||||
control,
|
||||
limit: Duration::from_secs(20),
|
||||
}
|
||||
}
|
||||
|
||||
fn run_flow(
|
||||
root: &Path,
|
||||
judge: &Judge,
|
||||
zen: &fake::Zen,
|
||||
options: &Options,
|
||||
trace: &Trace,
|
||||
) -> Report {
|
||||
run(&load("fix").unwrap(), root, options, judge, zen, trace).unwrap()
|
||||
}
|
||||
|
||||
fn read(root: &Path) -> String {
|
||||
std::fs::read_to_string(root.join("src/add.sh")).unwrap()
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_shipped_flow_is_a_valid_program() {
|
||||
let flow = load("fix").unwrap();
|
||||
assert_eq!(flow.id, "flow.fix");
|
||||
assert!(flow.node("test").is_some() && flow.node("next").is_some());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn kai_picks_the_file_zen_patches_it_and_the_flow_is_done() {
|
||||
let repo = repo();
|
||||
let trace = Trace::new(repo.path().join(".git/kai.jsonl"));
|
||||
let zen = zen(|context| {
|
||||
assert_eq!(
|
||||
context["read"]["path"], "src/add.sh",
|
||||
"Zen sees the file Kai picked"
|
||||
);
|
||||
assert!(
|
||||
context["failure"]["output"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.contains("gave -1")
|
||||
);
|
||||
diff(BUGGY, FIXED)
|
||||
});
|
||||
let report = run_flow(
|
||||
repo.path(),
|
||||
&honest(),
|
||||
&zen,
|
||||
&options(3, Control::Kai),
|
||||
&trace,
|
||||
);
|
||||
assert_eq!(report.outcome, "done");
|
||||
assert_eq!(report.attempts.len(), 1);
|
||||
let a = &report.attempts[0];
|
||||
assert_eq!(a.file.as_deref(), Some("src/add.sh"));
|
||||
assert_eq!((a.by.as_str(), a.applied, a.passed), ("kai", true, true));
|
||||
let (answer, p) = a.kai.clone().unwrap();
|
||||
assert_eq!(answer, "done");
|
||||
assert!((p - 0.9).abs() < 1e-6, "{p}");
|
||||
assert_eq!(read(repo.path()), FIXED);
|
||||
assert!(
|
||||
report.summary().contains("outcome: done"),
|
||||
"{}",
|
||||
report.summary()
|
||||
);
|
||||
|
||||
let lines = fake::lines(trace.path(), 1);
|
||||
let line = &lines[0];
|
||||
assert_eq!(line["family"], "dev.flow");
|
||||
assert_eq!(line["sku"], "flow.fix@1");
|
||||
assert_eq!(line["outcome"]["finish"], "done");
|
||||
assert_eq!(line["outcome"]["result"]["passed"], true);
|
||||
assert_eq!(line["ops"]["pick"]["kai"], "src/add.sh");
|
||||
assert_eq!(line["ops"]["pick"]["applied"], true);
|
||||
assert_eq!(line["ops"]["next"]["kai"], "done");
|
||||
assert_eq!(line["ops"]["next"]["taken"], "done");
|
||||
assert!(
|
||||
line["ops"]["pick"]["answers"].as_array().unwrap().len() >= 2,
|
||||
"one per candidate"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_wrong_patch_is_undone_and_the_retry_sees_it() {
|
||||
let repo = repo();
|
||||
let trace = Trace::new(repo.path().join(".git/kai.jsonl"));
|
||||
let zen = zen(|context| {
|
||||
if context["last"].is_null() {
|
||||
diff(BUGGY, "echo $(( $1 * $2 ))")
|
||||
} else {
|
||||
assert!(
|
||||
context["last"]["test"]["output"]
|
||||
.as_str()
|
||||
.unwrap()
|
||||
.contains("gave 6")
|
||||
);
|
||||
diff(BUGGY, FIXED)
|
||||
}
|
||||
});
|
||||
let report = run_flow(
|
||||
repo.path(),
|
||||
&honest(),
|
||||
&zen,
|
||||
&options(3, Control::Kai),
|
||||
&trace,
|
||||
);
|
||||
assert_eq!(report.outcome, "done");
|
||||
let taken: Vec<&str> = report.attempts.iter().map(|a| a.taken.as_str()).collect();
|
||||
assert_eq!(taken, ["retry", "done"]);
|
||||
assert!(report.attempts[0].applied && !report.attempts[0].passed);
|
||||
assert_eq!(read(repo.path()), FIXED);
|
||||
assert_eq!(fake::lines(trace.path(), 2).len(), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn out_of_attempts_the_flow_escalates_and_leaves_the_tree_as_it_was() {
|
||||
let repo = repo();
|
||||
let trace = Trace::new(repo.path().join(".git/kai.jsonl"));
|
||||
let zen = zen(|_| diff(BUGGY, "echo 0"));
|
||||
let report = run_flow(
|
||||
repo.path(),
|
||||
&honest(),
|
||||
&zen,
|
||||
&options(2, Control::Kai),
|
||||
&trace,
|
||||
);
|
||||
assert_eq!(report.outcome, "escalate");
|
||||
let taken: Vec<&str> = report.attempts.iter().map(|a| a.taken.as_str()).collect();
|
||||
assert_eq!(taken, ["retry", "escalate"]);
|
||||
assert_eq!(read(repo.path()), BUGGY);
|
||||
assert!(report.summary().contains("+echo 0"), "{}", report.summary());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn kai_cannot_call_a_failing_attempt_done() {
|
||||
let repo = repo();
|
||||
let trace = Trace::new(repo.path().join(".git/kai.jsonl"));
|
||||
let judge = kai(|_| ("done", 0.95));
|
||||
let zen = zen(|_| diff(BUGGY, "echo 4"));
|
||||
let report = run_flow(repo.path(), &judge, &zen, &options(2, Control::Kai), &trace);
|
||||
let taken: Vec<&str> = report.attempts.iter().map(|a| a.taken.as_str()).collect();
|
||||
assert_eq!(taken, ["retry", "escalate"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn kai_may_escalate_early_and_rule_control_ignores_it() {
|
||||
let repo = repo();
|
||||
let trace = Trace::new(repo.path().join(".git/kai.jsonl"));
|
||||
let judge = kai(|_| ("escalate", 0.9));
|
||||
let zen = zen(|_| diff(BUGGY, "echo 0"));
|
||||
let report = run_flow(repo.path(), &judge, &zen, &options(3, Control::Kai), &trace);
|
||||
assert_eq!(report.attempts.len(), 1);
|
||||
assert_eq!(report.outcome, "escalate");
|
||||
|
||||
let repo = self::repo();
|
||||
let report = run_flow(
|
||||
repo.path(),
|
||||
&judge,
|
||||
&zen,
|
||||
&options(3, Control::Rule),
|
||||
&trace,
|
||||
);
|
||||
let taken: Vec<&str> = report.attempts.iter().map(|a| a.taken.as_str()).collect();
|
||||
assert_eq!(taken, ["retry", "retry", "escalate"]);
|
||||
assert!(report.attempts.iter().all(|a| a.by == "rule"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn passing_tests_need_no_attempt() {
|
||||
let repo = repo();
|
||||
std::fs::write(repo.path().join("src/add.sh"), FIXED).unwrap();
|
||||
let trace = Trace::new(repo.path().join(".git/kai.jsonl"));
|
||||
let zen = zen(|_| unreachable!("no attempt, no Zen"));
|
||||
let report = run_flow(
|
||||
repo.path(),
|
||||
&honest(),
|
||||
&zen,
|
||||
&options(3, Control::Kai),
|
||||
&trace,
|
||||
);
|
||||
assert_eq!(report.outcome, "passing");
|
||||
assert!(report.attempts.is_empty());
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
//! Kai in Hanzo Dev.
|
||||
//!
|
||||
//! Zen generates, Enso routes, Kai decides, policy governs. In the agent loop ([`agent`]) Kai
|
||||
//! makes the loop's decisions per turn, each a Decision Program run through `control` in
|
||||
//! process, each in the mode `kai.toml` gives it. A flow ([`flow`]) is a Decision Program run
|
||||
//! over a repository, its steps deterministic tools, Kai decisions or Zen generation, under
|
||||
//! Kai's control or the tests' and the budget's. Every decision is a line in the trace
|
||||
//! ([`trace`]), the shape Enso's control writes.
|
||||
|
||||
pub mod agent;
|
||||
pub mod decide;
|
||||
#[cfg(test)]
|
||||
mod fake;
|
||||
pub mod flow;
|
||||
pub mod tools;
|
||||
pub mod trace;
|
||||
pub mod zen;
|
||||
|
||||
pub use agent::extension::install;
|
||||
pub use decide::Decider;
|
||||
pub use hanzo_config::kai::Kai as Settings;
|
||||
|
||||
use std::sync::Arc;
|
||||
use trace::Trace;
|
||||
|
||||
/// Kai for one profile: its settings, the checkpoint it answers from, and the trace.
|
||||
pub struct Kai {
|
||||
pub settings: Settings,
|
||||
pub decider: Arc<Decider>,
|
||||
pub trace: Arc<Trace>,
|
||||
}
|
||||
|
||||
impl Kai {
|
||||
/// Kai in process, from `settings`; the checkpoint loads on the first decision.
|
||||
pub fn open(settings: Settings) -> Kai {
|
||||
let decider = Decider::open(settings.models.clone(), settings.device.clone());
|
||||
Kai::with(settings, decider)
|
||||
}
|
||||
|
||||
pub fn with(settings: Settings, decider: Decider) -> Kai {
|
||||
let trace = Arc::new(Trace::new(settings.trace.clone()));
|
||||
Kai {
|
||||
settings,
|
||||
decider: Arc::new(decider),
|
||||
trace,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,358 @@
|
||||
//! Deterministic tools over a repository: search, read, apply a patch, run the tests, git.
|
||||
//! A flow's solver nodes call them; the loop's context selection uses [`search`].
|
||||
|
||||
use serde_json::Value;
|
||||
use serde_json::json;
|
||||
use std::collections::HashSet;
|
||||
use std::io::Read;
|
||||
use std::path::Path;
|
||||
use std::process::Command;
|
||||
use std::process::Stdio;
|
||||
use std::time::Duration;
|
||||
use std::time::Instant;
|
||||
|
||||
/// How much of a file a search reads.
|
||||
pub const HEAD: usize = 64 * 1024;
|
||||
/// How many files a search scores.
|
||||
const FILES: usize = 20_000;
|
||||
/// How much of a tool's output is kept: its tail, where failures print.
|
||||
pub const TAIL: usize = 6000;
|
||||
|
||||
/// A file a search found, and how well it matched.
|
||||
#[derive(Debug, Clone, PartialEq)]
|
||||
pub struct Hit {
|
||||
pub path: String,
|
||||
pub score: f64,
|
||||
}
|
||||
|
||||
/// Lowercase words of three or more characters: letters, digits and underscores.
|
||||
pub fn words(text: &str) -> HashSet<String> {
|
||||
text.split(|c: char| !(c.is_alphanumeric() || c == '_'))
|
||||
.filter(|w| w.chars().count() >= 3)
|
||||
.map(str::to_lowercase)
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// The repository's files, as git lists them; every file under `root` when it is not a
|
||||
/// repository.
|
||||
pub fn files(root: &Path) -> Vec<String> {
|
||||
let listed = Command::new("git")
|
||||
.args([
|
||||
"ls-files",
|
||||
"--cached",
|
||||
"--others",
|
||||
"--exclude-standard",
|
||||
"-z",
|
||||
])
|
||||
.current_dir(root)
|
||||
.stderr(Stdio::null())
|
||||
.output();
|
||||
if let Ok(out) = listed
|
||||
&& out.status.success()
|
||||
{
|
||||
return out
|
||||
.stdout
|
||||
.split(|b| *b == 0)
|
||||
.filter(|p| !p.is_empty())
|
||||
.map(|p| String::from_utf8_lossy(p).into_owned())
|
||||
.take(FILES)
|
||||
.collect();
|
||||
}
|
||||
let mut out = Vec::new();
|
||||
let mut stack = vec![root.to_path_buf()];
|
||||
while let Some(dir) = stack.pop() {
|
||||
let Ok(entries) = std::fs::read_dir(&dir) else {
|
||||
continue;
|
||||
};
|
||||
for entry in entries.flatten() {
|
||||
let path = entry.path();
|
||||
let name = entry.file_name();
|
||||
if name.to_string_lossy().starts_with('.') || name == "target" || name == "node_modules"
|
||||
{
|
||||
continue;
|
||||
}
|
||||
if path.is_dir() {
|
||||
stack.push(path);
|
||||
} else if let Ok(rel) = path.strip_prefix(root) {
|
||||
out.push(rel.to_string_lossy().into_owned());
|
||||
if out.len() >= FILES {
|
||||
return out;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
out.sort();
|
||||
out
|
||||
}
|
||||
|
||||
/// The first `n` bytes of `path` under `root` as text; `None` for a binary or unreadable file.
|
||||
pub fn head(root: &Path, path: &str, n: usize) -> Option<String> {
|
||||
let mut buf = Vec::with_capacity(n.min(1 << 16));
|
||||
std::fs::File::open(root.join(path))
|
||||
.ok()?
|
||||
.take(n as u64)
|
||||
.read_to_end(&mut buf)
|
||||
.ok()?;
|
||||
if buf.contains(&0) {
|
||||
return None;
|
||||
}
|
||||
Some(String::from_utf8_lossy(&buf).into_owned())
|
||||
}
|
||||
|
||||
/// Files under `root` ranked by the words they share with `query`: a word in the path counts
|
||||
/// three, one in the first `read` bytes counts one. At most `top`, best first; none that
|
||||
/// share nothing.
|
||||
pub fn search(root: &Path, query: &str, top: usize, read: usize) -> Vec<Hit> {
|
||||
let wanted = words(query);
|
||||
if wanted.is_empty() {
|
||||
return Vec::new();
|
||||
}
|
||||
let mut hits: Vec<Hit> = files(root)
|
||||
.into_iter()
|
||||
.filter_map(|path| {
|
||||
let in_path = words(&path).intersection(&wanted).count() as f64;
|
||||
let in_text = if read == 0 {
|
||||
0.0
|
||||
} else {
|
||||
let text = head(root, &path, read)?;
|
||||
words(&text).intersection(&wanted).count() as f64
|
||||
};
|
||||
let score = 3.0 * in_path + in_text;
|
||||
(score > 0.0).then_some(Hit { path, score })
|
||||
})
|
||||
.collect();
|
||||
hits.sort_by(|a, b| {
|
||||
b.score
|
||||
.total_cmp(&a.score)
|
||||
.then(a.path.len().cmp(&b.path.len()))
|
||||
.then(a.path.cmp(&b.path))
|
||||
});
|
||||
hits.truncate(top);
|
||||
hits
|
||||
}
|
||||
|
||||
/// The last `n` characters of `text`.
|
||||
pub fn tail(text: &str, n: usize) -> String {
|
||||
let count = text.chars().count();
|
||||
if count <= n {
|
||||
return text.to_string();
|
||||
}
|
||||
text.chars().skip(count - n).collect()
|
||||
}
|
||||
|
||||
/// Runs `sh -c script` in `root`, killing it after `limit`: `{code, output, ms}`, output
|
||||
/// being the tail of stdout and stderr together.
|
||||
pub fn shell(root: &Path, script: &str, limit: Duration) -> Value {
|
||||
let started = Instant::now();
|
||||
let mut command = Command::new("sh");
|
||||
command
|
||||
.arg("-c")
|
||||
.arg(format!("exec 2>&1\n{script}"))
|
||||
.current_dir(root)
|
||||
.stdin(Stdio::null())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::null());
|
||||
// Its own process group, so a timeout stops everything the script started.
|
||||
#[cfg(unix)]
|
||||
std::os::unix::process::CommandExt::process_group(&mut command, 0);
|
||||
let child = command.spawn();
|
||||
let mut child = match child {
|
||||
Ok(c) => c,
|
||||
Err(e) => return json!({"code": -1, "output": e.to_string(), "ms": 0}),
|
||||
};
|
||||
let Some(mut stdout) = child.stdout.take() else {
|
||||
let _ = child.kill();
|
||||
return json!({"code": -1, "output": "no output pipe", "ms": 0});
|
||||
};
|
||||
let reader = std::thread::spawn(move || {
|
||||
let mut out = Vec::new();
|
||||
let _ = stdout.read_to_end(&mut out);
|
||||
out
|
||||
});
|
||||
let code = loop {
|
||||
match child.try_wait() {
|
||||
Ok(Some(status)) => break status.code().unwrap_or(-1),
|
||||
Ok(None) if started.elapsed() > limit => {
|
||||
#[cfg(unix)]
|
||||
let _ = Command::new("kill")
|
||||
.args(["-9", &format!("-{}", child.id())])
|
||||
.status();
|
||||
let _ = child.kill();
|
||||
let _ = child.wait();
|
||||
break -2;
|
||||
}
|
||||
Ok(None) => std::thread::sleep(Duration::from_millis(20)),
|
||||
Err(_) => break -1,
|
||||
}
|
||||
};
|
||||
let out = reader.join().unwrap_or_default();
|
||||
let mut output = tail(&String::from_utf8_lossy(&out), TAIL);
|
||||
if code == -2 {
|
||||
output.push_str(&format!("\n[killed after {}s]", limit.as_secs()));
|
||||
}
|
||||
json!({"code": code, "output": output, "ms": started.elapsed().as_millis() as u64})
|
||||
}
|
||||
|
||||
/// Runs the tests: `{passed, code, output, ms}`.
|
||||
pub fn test(root: &Path, command: &str, limit: Duration) -> Value {
|
||||
let mut run = shell(root, command, limit);
|
||||
let passed = run["code"] == 0;
|
||||
run["passed"] = Value::Bool(passed);
|
||||
run
|
||||
}
|
||||
|
||||
/// The unified diff inside a model's answer: a fenced block when there is one, else the
|
||||
/// text from the first `diff --git` or `--- ` line.
|
||||
pub fn diff_of(answer: &str) -> String {
|
||||
let mut fenced = String::new();
|
||||
let mut inside = false;
|
||||
for line in answer.lines() {
|
||||
if line.trim_start().starts_with("```") {
|
||||
if inside {
|
||||
break;
|
||||
}
|
||||
inside = true;
|
||||
continue;
|
||||
}
|
||||
if inside {
|
||||
fenced.push_str(line);
|
||||
fenced.push('\n');
|
||||
}
|
||||
}
|
||||
let text = if fenced.trim().is_empty() {
|
||||
answer
|
||||
} else {
|
||||
fenced.as_str()
|
||||
};
|
||||
let start = text
|
||||
.find("diff --git")
|
||||
.or_else(|| text.find("--- "))
|
||||
.unwrap_or(0);
|
||||
let mut diff = text[start..].to_string();
|
||||
if !diff.ends_with('\n') {
|
||||
diff.push('\n');
|
||||
}
|
||||
diff
|
||||
}
|
||||
|
||||
/// Applies a model's patch with `git apply`: `{applied, files, error}`.
|
||||
pub fn patch(root: &Path, answer: &str) -> Value {
|
||||
let diff = diff_of(answer);
|
||||
let files: Vec<String> = diff
|
||||
.lines()
|
||||
.filter_map(|l| l.strip_prefix("+++ "))
|
||||
.map(|p| p.trim().trim_start_matches("b/").to_string())
|
||||
.filter(|p| p != "/dev/null")
|
||||
.collect();
|
||||
if files.is_empty() {
|
||||
return json!({"applied": false, "files": [], "error": "no diff in the answer"});
|
||||
}
|
||||
let mut child = match Command::new("git")
|
||||
.args(["apply", "--recount", "--whitespace=nowarn", "-"])
|
||||
.current_dir(root)
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::null())
|
||||
.stderr(Stdio::piped())
|
||||
.spawn()
|
||||
{
|
||||
Ok(c) => c,
|
||||
Err(e) => return json!({"applied": false, "files": files, "error": e.to_string()}),
|
||||
};
|
||||
if let Some(mut stdin) = child.stdin.take() {
|
||||
use std::io::Write;
|
||||
let _ = stdin.write_all(diff.as_bytes());
|
||||
}
|
||||
match child.wait_with_output() {
|
||||
Ok(out) if out.status.success() => json!({"applied": true, "files": files}),
|
||||
Ok(out) => json!({
|
||||
"applied": false,
|
||||
"files": files,
|
||||
"error": tail(&String::from_utf8_lossy(&out.stderr), 2000),
|
||||
}),
|
||||
Err(e) => json!({"applied": false, "files": files, "error": e.to_string()}),
|
||||
}
|
||||
}
|
||||
|
||||
/// Undoes a patch [`patch`] applied from `answer`: `git apply -R`.
|
||||
pub fn revert(root: &Path, answer: &str) -> bool {
|
||||
let diff = diff_of(answer);
|
||||
let Ok(mut child) = Command::new("git")
|
||||
.args(["apply", "-R", "--recount", "--whitespace=nowarn", "-"])
|
||||
.current_dir(root)
|
||||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::null())
|
||||
.stderr(Stdio::null())
|
||||
.spawn()
|
||||
else {
|
||||
return false;
|
||||
};
|
||||
if let Some(mut stdin) = child.stdin.take() {
|
||||
use std::io::Write;
|
||||
let _ = stdin.write_all(diff.as_bytes());
|
||||
}
|
||||
child.wait().is_ok_and(|s| s.success())
|
||||
}
|
||||
|
||||
/// Git commands a flow may run: those that only read.
|
||||
const GIT: &[&str] = &["diff", "status", "log", "show", "rev-parse", "ls-files"];
|
||||
|
||||
/// Runs `git args` in `root` when the subcommand only reads: `{code, output}`.
|
||||
pub fn git(root: &Path, args: &[String]) -> Result<Value, String> {
|
||||
let sub = args.first().map(String::as_str).unwrap_or_default();
|
||||
if !GIT.contains(&sub) {
|
||||
return Err(format!("git {sub}: a flow runs only {}", GIT.join(", ")));
|
||||
}
|
||||
let out = Command::new("git")
|
||||
.args(args)
|
||||
.current_dir(root)
|
||||
.stdin(Stdio::null())
|
||||
.output()
|
||||
.map_err(|e| e.to_string())?;
|
||||
let mut text = String::from_utf8_lossy(&out.stdout).into_owned();
|
||||
text.push_str(&String::from_utf8_lossy(&out.stderr));
|
||||
Ok(json!({"code": out.status.code().unwrap_or(-1), "output": tail(&text, TAIL)}))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn a_fenced_diff_is_taken_from_the_answer() {
|
||||
let answer =
|
||||
"Here is the fix:\n```diff\n--- a/x.txt\n+++ b/x.txt\n@@ -1 +1 @@\n-1\n+2\n```\nDone.";
|
||||
assert_eq!(
|
||||
diff_of(answer),
|
||||
"--- a/x.txt\n+++ b/x.txt\n@@ -1 +1 @@\n-1\n+2\n"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn search_ranks_a_path_named_in_the_query_first() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
std::fs::create_dir_all(dir.path().join("src")).unwrap();
|
||||
std::fs::write(dir.path().join("src/math.rs"), "pub fn add() {}").unwrap();
|
||||
std::fs::write(dir.path().join("src/text.rs"), "pub fn words() {}").unwrap();
|
||||
std::fs::write(dir.path().join("README.md"), "math notes").unwrap();
|
||||
let hits = search(dir.path(), "panicked at src/math.rs:5: add failed", 8, HEAD);
|
||||
assert_eq!(hits[0].path, "src/math.rs");
|
||||
assert!(
|
||||
hits.iter()
|
||||
.all(|h| h.path != "src/text.rs" || h.score > 0.0)
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn git_runs_only_what_reads() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
assert!(git(dir.path(), &["push".to_string()]).is_err());
|
||||
assert!(git(dir.path(), &["status".to_string()]).is_ok());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn a_slow_command_is_killed() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let run = shell(dir.path(), "sleep 5", Duration::from_millis(200));
|
||||
assert_eq!(run["code"], -2);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,274 @@
|
||||
//! The decision trace: one JSON line per decision, for training.
|
||||
//!
|
||||
//! The shape is the line Enso's control writes per request (`time`, `request`, `family`,
|
||||
//! `sku`, token counts, `ms`, `outcome`, and `ops` keyed by operation, each with its program,
|
||||
//! hash, mode, the gate's mode per state, state hashes, Kai's distributions and signals, what
|
||||
//! Kai would do, what was done and whether Kai's answer was applied), so one pipeline reads
|
||||
//! both. A line carries state hashes and distributions, never the text of a request, a
|
||||
//! command or an answer. Dev adds `outcome.result`, what followed the decision.
|
||||
|
||||
use control::Decision;
|
||||
use control::Row;
|
||||
use indexmap::IndexMap;
|
||||
use program::policy::Signal;
|
||||
use serde::Deserialize;
|
||||
use serde::Serialize;
|
||||
use serde_json::Value;
|
||||
use std::collections::BTreeMap;
|
||||
use std::io::Write;
|
||||
use std::path::PathBuf;
|
||||
use std::sync::Arc;
|
||||
use std::sync::Mutex;
|
||||
|
||||
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
|
||||
pub struct Line {
|
||||
pub time: String,
|
||||
#[serde(default, skip_serializing_if = "String::is_empty")]
|
||||
pub request: String,
|
||||
pub family: String,
|
||||
pub sku: String,
|
||||
#[serde(default, skip_serializing_if = "String::is_empty")]
|
||||
pub served: String,
|
||||
pub prompt_tokens: i64,
|
||||
pub cached_tokens: i64,
|
||||
pub completion_tokens: i64,
|
||||
pub reasoning_tokens: i64,
|
||||
pub ms: f64,
|
||||
#[serde(default, skip_serializing_if = "String::is_empty")]
|
||||
pub cost: String,
|
||||
pub outcome: Outcome,
|
||||
pub ops: BTreeMap<String, Step>,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
|
||||
pub struct Outcome {
|
||||
pub status: i64,
|
||||
#[serde(default, skip_serializing_if = "String::is_empty")]
|
||||
pub finish: String,
|
||||
#[serde(default, skip_serializing_if = "std::ops::Not::not")]
|
||||
pub escalated: bool,
|
||||
#[serde(default, skip_serializing_if = "String::is_empty")]
|
||||
pub error: String,
|
||||
/// What followed the decision: the tools called, a command's result, a flow's tests.
|
||||
#[serde(default, skip_serializing_if = "Value::is_null")]
|
||||
pub result: Value,
|
||||
}
|
||||
|
||||
/// One operation's decision.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
|
||||
pub struct Step {
|
||||
pub program: String,
|
||||
#[serde(default, skip_serializing_if = "String::is_empty")]
|
||||
pub hash: String,
|
||||
/// As configured.
|
||||
pub mode: String,
|
||||
/// Per state, as the gate ran it.
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
pub gates: Vec<String>,
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
pub states: Vec<String>,
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
pub answers: Vec<IndexMap<String, Value>>,
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
pub signals: Vec<IndexMap<String, Signal>>,
|
||||
/// The checkpoint that answered.
|
||||
#[serde(default, skip_serializing_if = "String::is_empty")]
|
||||
pub model: String,
|
||||
/// What Kai would do.
|
||||
#[serde(default, skip_serializing_if = "String::is_empty")]
|
||||
pub kai: String,
|
||||
/// What was done.
|
||||
#[serde(default, skip_serializing_if = "String::is_empty")]
|
||||
pub taken: String,
|
||||
pub applied: bool,
|
||||
#[serde(default, skip_serializing_if = "is_zero")]
|
||||
pub ms: f64,
|
||||
#[serde(default, skip_serializing_if = "String::is_empty")]
|
||||
pub error: String,
|
||||
}
|
||||
|
||||
fn is_zero(ms: &f64) -> bool {
|
||||
*ms == 0.0
|
||||
}
|
||||
|
||||
impl Step {
|
||||
pub fn new(program: &str, mode: &str) -> Step {
|
||||
Step {
|
||||
program: program.to_string(),
|
||||
mode: mode.to_string(),
|
||||
..Step::default()
|
||||
}
|
||||
}
|
||||
|
||||
/// Fills the step from Kai's decision: every state's gate, hash, distributions and
|
||||
/// signals, in order.
|
||||
pub fn fill(&mut self, decision: &Decision) {
|
||||
self.program = decision.program.clone();
|
||||
self.hash = decision.hash.clone();
|
||||
self.model = decision.revision.model.clone();
|
||||
self.ms = decision.ms;
|
||||
for row in &decision.results {
|
||||
match row {
|
||||
Row::Ruled(r) => {
|
||||
self.states.push(r.state.clone());
|
||||
self.gates.push(mode_name(r.mode).to_string());
|
||||
self.answers.push(r.answers.clone());
|
||||
self.signals.push(r.signals.clone());
|
||||
}
|
||||
Row::Failed { state, error } => {
|
||||
self.states.push(state.clone());
|
||||
self.gates.push(String::new());
|
||||
self.answers.push(IndexMap::new());
|
||||
self.signals.push(IndexMap::new());
|
||||
if self.error.is_empty() {
|
||||
self.error = error.clone();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub fn mode_name(mode: program::Mode) -> &'static str {
|
||||
match mode {
|
||||
program::Mode::Shadow => "shadow",
|
||||
program::Mode::Advisory => "advisory",
|
||||
program::Mode::Enforced => "enforced",
|
||||
}
|
||||
}
|
||||
|
||||
/// Now, as Enso writes it: RFC 3339, UTC, nanoseconds.
|
||||
pub fn now() -> String {
|
||||
chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Nanos, true)
|
||||
}
|
||||
|
||||
/// Appends lines to one file. Writes are whole lines, one at a time.
|
||||
pub struct Trace {
|
||||
path: PathBuf,
|
||||
file: Mutex<Option<std::fs::File>>,
|
||||
}
|
||||
|
||||
impl Trace {
|
||||
pub fn new(path: PathBuf) -> Trace {
|
||||
Trace {
|
||||
path,
|
||||
file: Mutex::new(None),
|
||||
}
|
||||
}
|
||||
|
||||
pub fn path(&self) -> &std::path::Path {
|
||||
&self.path
|
||||
}
|
||||
|
||||
/// Appends `line`. A trace that cannot be written is logged and skipped: the loop never
|
||||
/// waits on it.
|
||||
pub fn write(&self, line: &Line) {
|
||||
let Ok(mut text) = serde_json::to_vec(line) else {
|
||||
return;
|
||||
};
|
||||
text.push(b'\n');
|
||||
let mut file = self
|
||||
.file
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
if file.is_none() {
|
||||
if let Some(dir) = self.path.parent() {
|
||||
let _ = std::fs::create_dir_all(dir);
|
||||
}
|
||||
match std::fs::OpenOptions::new()
|
||||
.create(true)
|
||||
.append(true)
|
||||
.open(&self.path)
|
||||
{
|
||||
Ok(f) => *file = Some(f),
|
||||
Err(e) => {
|
||||
tracing::warn!("kai trace {}: {e}", self.path.display());
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
if let Some(f) = file.as_mut()
|
||||
&& let Err(e) = f.write_all(&text)
|
||||
{
|
||||
tracing::warn!("kai trace {}: {e}", self.path.display());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A line that is written once both its decision and its outcome are in, in either order.
|
||||
pub struct Record {
|
||||
trace: Arc<Trace>,
|
||||
inner: Mutex<Pending>,
|
||||
}
|
||||
|
||||
struct Pending {
|
||||
line: Line,
|
||||
op: String,
|
||||
decided: bool,
|
||||
closed: bool,
|
||||
}
|
||||
|
||||
impl Record {
|
||||
pub fn new(trace: Arc<Trace>, line: Line, op: &str, step: Step) -> Arc<Record> {
|
||||
let mut line = line;
|
||||
line.ops.insert(op.to_string(), step);
|
||||
Arc::new(Record {
|
||||
trace,
|
||||
inner: Mutex::new(Pending {
|
||||
line,
|
||||
op: op.to_string(),
|
||||
decided: false,
|
||||
closed: false,
|
||||
}),
|
||||
})
|
||||
}
|
||||
|
||||
fn with(&self, f: impl FnOnce(&mut Pending)) {
|
||||
let mut p = self
|
||||
.inner
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let was = p.decided && p.closed;
|
||||
f(&mut p);
|
||||
if !was && p.decided && p.closed {
|
||||
self.trace.write(&p.line);
|
||||
}
|
||||
}
|
||||
|
||||
/// The decision is in: `f` fills the step, and the outcome when the decision is its own.
|
||||
pub fn decided(&self, f: impl FnOnce(&mut Step, &mut Outcome)) {
|
||||
self.with(|p| {
|
||||
if let Some(step) = p.line.ops.get_mut(&p.op) {
|
||||
f(step, &mut p.line.outcome);
|
||||
p.line.ms = step.ms;
|
||||
}
|
||||
p.decided = true;
|
||||
});
|
||||
}
|
||||
|
||||
/// Updates the step before or after the decision is in, without deciding it.
|
||||
pub fn step(&self, f: impl FnOnce(&mut Step)) {
|
||||
self.with(|p| {
|
||||
if let Some(step) = p.line.ops.get_mut(&p.op) {
|
||||
f(step);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// The outcome is in: `f` fills the line. Only the first close counts.
|
||||
pub fn closed(&self, f: impl FnOnce(&mut Line)) {
|
||||
self.with(|p| {
|
||||
if !p.closed {
|
||||
f(&mut p.line);
|
||||
p.closed = true;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
pub fn is_closed(&self) -> bool {
|
||||
self.inner
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
.closed
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
//! Zen through the Hanzo API: the only model calls a flow makes, and only to write code or
|
||||
//! text.
|
||||
|
||||
use program::Revision;
|
||||
use serde_json::Value;
|
||||
use serde_json::json;
|
||||
use std::time::Duration;
|
||||
|
||||
const SYSTEM: &str = "You are Zen, writing code for Hanzo Dev. Answer with exactly what the \
|
||||
request asks for and nothing else. When it asks for a change, answer with one unified diff \
|
||||
against the repository root that `git apply` accepts, in a single ```diff block.";
|
||||
|
||||
/// Zen over `POST /v1/chat/completions`.
|
||||
pub struct Zen {
|
||||
base: String,
|
||||
key: Option<String>,
|
||||
model: String,
|
||||
client: reqwest::Client,
|
||||
runtime: tokio::runtime::Handle,
|
||||
}
|
||||
|
||||
impl Zen {
|
||||
/// Zen at `base` (`https://api.hanzo.ai/v1`) as `model`, signed with `key`. Call from
|
||||
/// within the async runtime; generation then blocks a thread off it.
|
||||
pub fn new(base: &str, key: Option<String>, model: &str) -> Zen {
|
||||
Zen {
|
||||
base: base.trim_end_matches('/').to_string(),
|
||||
key,
|
||||
model: model.to_string(),
|
||||
client: reqwest::Client::builder()
|
||||
.timeout(Duration::from_secs(600))
|
||||
.build()
|
||||
.unwrap_or_default(),
|
||||
runtime: tokio::runtime::Handle::current(),
|
||||
}
|
||||
}
|
||||
|
||||
async fn complete(&self, prompt: &str, context: &Value) -> Result<String, String> {
|
||||
let body = json!({
|
||||
"model": self.model,
|
||||
"temperature": 0,
|
||||
"messages": [
|
||||
{"role": "system", "content": SYSTEM},
|
||||
{"role": "user", "content": format!(
|
||||
"{prompt}\n\nInputs, as JSON:\n{}",
|
||||
serde_json::to_string_pretty(context).unwrap_or_default()
|
||||
)},
|
||||
],
|
||||
});
|
||||
let mut request = self
|
||||
.client
|
||||
.post(format!("{}/chat/completions", self.base))
|
||||
.json(&body);
|
||||
if let Some(key) = &self.key {
|
||||
request = request.bearer_auth(key);
|
||||
}
|
||||
let response = request.send().await.map_err(|e| format!("zen: {e}"))?;
|
||||
let status = response.status();
|
||||
let answer: Value = response.json().await.map_err(|e| format!("zen: {e}"))?;
|
||||
if !status.is_success() {
|
||||
let why = answer["error"]["message"]
|
||||
.as_str()
|
||||
.unwrap_or("no reason given");
|
||||
return Err(format!("zen: {status}: {why}"));
|
||||
}
|
||||
answer["choices"][0]["message"]["content"]
|
||||
.as_str()
|
||||
.map(str::to_string)
|
||||
.ok_or_else(|| "zen: an answer without text".to_string())
|
||||
}
|
||||
}
|
||||
|
||||
impl program::Zen for Zen {
|
||||
fn generate(&self, prompt: &str, context: &Value) -> program::Result<String> {
|
||||
self.runtime
|
||||
.block_on(self.complete(prompt, context))
|
||||
.map_err(program::Error::Backend)
|
||||
}
|
||||
|
||||
fn revision(&self) -> Revision {
|
||||
Revision {
|
||||
model: self.model.clone(),
|
||||
calibration: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
[package]
|
||||
name = "hanzo-loop"
|
||||
version.workspace = true
|
||||
edition.workspace = true
|
||||
license.workspace = true
|
||||
description = "What a controller decides inside the agent loop, as data the loop applies."
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
|
||||
# The agent loop depends on this crate, so it stays small and changes rarely: the loop is
|
||||
# rebuilt whenever it does.
|
||||
[dependencies]
|
||||
serde_json = { workspace = true }
|
||||
|
||||
[dev-dependencies]
|
||||
tokio = { workspace = true, features = ["macros", "rt"] }
|
||||
@@ -0,0 +1,158 @@
|
||||
//! What a controller decides inside the agent loop, as data the loop applies.
|
||||
//!
|
||||
//! The loop asks at three points and applies the answer: which tools the model sees directly
|
||||
//! this step, whether a command may run, and which routing hints ride the turn's requests. A
|
||||
//! controller installed on the thread answers. Nothing it says can loosen what the loop already
|
||||
//! decided: a verdict joins the policy's on `allow < ask < deny`, a shortlist only moves tools
|
||||
//! out of direct view, and a hint is metadata the gateway may read.
|
||||
//!
|
||||
//! The loop finds the controller as a [`Handle`] in the thread's extension data. The patched
|
||||
//! upstream calls in `patches/upstream.json` are the only readers.
|
||||
|
||||
use std::collections::BTreeMap;
|
||||
use std::collections::HashSet;
|
||||
use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
|
||||
/// A controller's answer, computed asynchronously.
|
||||
pub type Answer<'a, T> = Pin<Box<dyn Future<Output = T> + Send + 'a>>;
|
||||
|
||||
/// Whether an action may proceed. The order is the lattice: `Allow < Ask < Deny`.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
|
||||
pub enum Verdict {
|
||||
Allow,
|
||||
Ask,
|
||||
Deny,
|
||||
}
|
||||
|
||||
impl Verdict {
|
||||
/// The stricter of the two.
|
||||
pub fn join(self, other: Verdict) -> Verdict {
|
||||
self.max(other)
|
||||
}
|
||||
|
||||
pub fn name(self) -> &'static str {
|
||||
match self {
|
||||
Verdict::Allow => "allow",
|
||||
Verdict::Ask => "ask",
|
||||
Verdict::Deny => "deny",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A tool on offer: its model-visible name and what it says it does.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct Card {
|
||||
pub name: String,
|
||||
pub description: String,
|
||||
}
|
||||
|
||||
/// A turn's request as the loop received it.
|
||||
#[derive(Debug, Clone, Default, PartialEq, Eq)]
|
||||
pub struct Request {
|
||||
/// The user's text.
|
||||
pub text: String,
|
||||
/// Whether it carries images.
|
||||
pub images: bool,
|
||||
}
|
||||
|
||||
pub trait Controller: Send + Sync {
|
||||
/// Of `cards`, the tools to keep in direct view for a step of `turn`; `None` keeps all.
|
||||
fn tools<'a>(&'a self, turn: &'a str, cards: Vec<Card>) -> Answer<'a, Option<HashSet<String>>>;
|
||||
|
||||
/// The verdict for the command `call` in `turn` would run, which the policy rates
|
||||
/// `policy`. `action` is the command as the approval path describes it.
|
||||
fn command<'a>(
|
||||
&'a self,
|
||||
turn: &'a str,
|
||||
call: &'a str,
|
||||
policy: Verdict,
|
||||
action: serde_json::Value,
|
||||
) -> Answer<'a, Verdict>;
|
||||
|
||||
/// Routing hints for `turn`'s model requests.
|
||||
fn route<'a>(&'a self, turn: &'a str, request: Request)
|
||||
-> Answer<'a, BTreeMap<String, String>>;
|
||||
}
|
||||
|
||||
/// The controller installed on a thread.
|
||||
#[derive(Clone)]
|
||||
pub struct Handle(pub Arc<dyn Controller>);
|
||||
|
||||
impl Handle {
|
||||
/// The command's verdict: the policy's joined with the controller's, never looser.
|
||||
pub async fn command(
|
||||
&self,
|
||||
turn: &str,
|
||||
call: &str,
|
||||
policy: Verdict,
|
||||
action: serde_json::Value,
|
||||
) -> Verdict {
|
||||
self.0
|
||||
.command(turn, call, policy, action)
|
||||
.await
|
||||
.join(policy)
|
||||
}
|
||||
|
||||
pub async fn tools(&self, turn: &str, cards: Vec<Card>) -> Option<Shortlist> {
|
||||
self.0.tools(turn, cards).await.map(Shortlist)
|
||||
}
|
||||
|
||||
pub async fn route(&self, turn: &str, request: Request) -> BTreeMap<String, String> {
|
||||
self.0.route(turn, request).await
|
||||
}
|
||||
}
|
||||
|
||||
/// A step's shortlist: the tools that stay in direct view.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct Shortlist(pub HashSet<String>);
|
||||
|
||||
impl Shortlist {
|
||||
pub fn keeps(&self, tool: &str) -> bool {
|
||||
self.0.contains(tool)
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
struct Loose;
|
||||
|
||||
impl Controller for Loose {
|
||||
fn tools<'a>(&'a self, _: &'a str, _: Vec<Card>) -> Answer<'a, Option<HashSet<String>>> {
|
||||
Box::pin(async { None })
|
||||
}
|
||||
fn command<'a>(
|
||||
&'a self,
|
||||
_: &'a str,
|
||||
_: &'a str,
|
||||
_: Verdict,
|
||||
_: serde_json::Value,
|
||||
) -> Answer<'a, Verdict> {
|
||||
Box::pin(async { Verdict::Allow })
|
||||
}
|
||||
fn route<'a>(&'a self, _: &'a str, _: Request) -> Answer<'a, BTreeMap<String, String>> {
|
||||
Box::pin(async { BTreeMap::new() })
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn the_lattice_orders_allow_ask_deny() {
|
||||
assert_eq!(Verdict::Allow.join(Verdict::Ask), Verdict::Ask);
|
||||
assert_eq!(Verdict::Deny.join(Verdict::Ask), Verdict::Deny);
|
||||
assert_eq!(Verdict::Allow.join(Verdict::Allow), Verdict::Allow);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn a_controller_that_allows_cannot_loosen_the_policy() {
|
||||
let handle = Handle(Arc::new(Loose));
|
||||
for policy in [Verdict::Allow, Verdict::Ask, Verdict::Deny] {
|
||||
let verdict = handle
|
||||
.command("t", "c", policy, serde_json::Value::Null)
|
||||
.await;
|
||||
assert_eq!(verdict, policy);
|
||||
}
|
||||
}
|
||||
}
|
||||
BIN
Binary file not shown.
+4
-4
@@ -1,6 +1,6 @@
|
||||
1e6b85f098f58c700a6b655ba2a9874196178e5195583dba04b198ad462bb13f Cargo.lock
|
||||
fde9ae0773128d76b07cfc611ce4ea1e7aa8d0147d479a8c91c405b3835fe9d4 Cargo.toml
|
||||
1f12a27d2828cdd2ffcc446acaba3f8f52dfb5df257a8520335eb33c1ac7e954 Makefile
|
||||
9cd6d43e934741b191d3f9f96e419980129eb9c35658f4af43b6cc8c3956c5a8 Cargo.lock
|
||||
5c6a159d3f8f18739ac24defccf6f81d0ecd12640ac1b82e9b9e91578e75f55f Cargo.toml
|
||||
270bfcb09db9c1d01120060326c9158219e3b001a75502f9f54f5441188fb614 Makefile
|
||||
8c35d8e2c8526a931766898c3fcdf0ef1ed7dfaf261c45a154c1dfccc2010c61 crates/core/Cargo.toml
|
||||
3a04e2e31602a642becff242ecd295078812ec30c3e073d2debb336c266d97f4 crates/core/src/lib.rs
|
||||
a139b95fd31492f8f8ae08fabf2d0f5ddf7eaf69f10eab46f511cc9d57686759 crates/core/tests/session.rs
|
||||
@@ -12,5 +12,5 @@ e7523851f6ad52e7bb016e269f16d9b4738a7c7416316a133250b7095b6cc21a crates/ffi/src
|
||||
803b9f0198a7cc71dc59831bbcdd25f4cbaf55575fd523f93faa4710edbde5df crates/ffi/tests/abi.rs
|
||||
dcc7a0c0b23d8b48686201d1019437ca40a4959d43d4c91e8e75b5992efc8582 crates/protocol/Cargo.toml
|
||||
667faea58072a615039e9374a8beffe481aa99d88903ab0888d972e8d1c3de3e crates/protocol/src/lib.rs
|
||||
9d28969b2bbf6dad2e30602840b5ef9cd240cbb37f3d0b1028c19127fa0c7e5a go/dev.wasm
|
||||
5bee5265160bd47693a66da06e026a3458fc0fdc8a7cff0d57410b6f7314f7ac go/dev.wasm
|
||||
9dc3df952bc2a2f54de1e8e35d68eb5356ca245a9f479f9e5ad4b133afa356f0 rust-toolchain.toml
|
||||
|
||||
@@ -226,5 +226,47 @@
|
||||
"old": "JSON value from `codex/sandbox-state-meta` to apply directly",
|
||||
"new": "JSON value from the MCP sandbox state capability to apply directly",
|
||||
"count": 1
|
||||
},
|
||||
{
|
||||
"file": "core/Cargo.toml",
|
||||
"old": "[dependencies]",
|
||||
"new": "[dependencies]\nhanzo-loop = { path = \"../../../../crates/hanzo-loop\" }",
|
||||
"count": 1
|
||||
},
|
||||
{
|
||||
"file": "core/src/session/turn.rs",
|
||||
"old": " Ok(Arc::new(build_tool_router(\n",
|
||||
"new": " // Hanzo: the thread's controller shortlists this step's MCP tools.\n if let Some(handle) = sess\n .services\n .thread_extension_data\n .get::<hanzo_loop::Handle>()\n {\n let cards = all_mcp_tools\n .iter()\n .map(|tool| hanzo_loop::Card {\n name: tool.canonical_tool_name().to_string(),\n description: tool.tool.description.as_deref().unwrap_or_default().to_string(),\n })\n .collect();\n if let Some(shortlist) = handle.tools(&turn_context.sub_id, cards).await {\n step_store.insert(shortlist);\n }\n }\n Ok(Arc::new(build_tool_router(\n",
|
||||
"count": 1
|
||||
},
|
||||
{
|
||||
"file": "core/src/tools/spec_plan.rs",
|
||||
"old": " apply_mcp_tool_exposure_policy(\n turn_context,\n model_info,\n mcp,\n ®istered_mcp_tools,\n &mut registry,\n );\n",
|
||||
"new": " apply_mcp_tool_exposure_policy(\n turn_context,\n model_info,\n mcp,\n ®istered_mcp_tools,\n &mut registry,\n );\n // Hanzo: MCP tools off the controller's shortlist leave direct view; tool search\n // still finds them when the model supports it.\n if let Some(shortlist) = step_store.get::<hanzo_loop::Shortlist>() {\n let search = search_tool_enabled(turn_context, model_info);\n for tool in registry.entries_mut() {\n let name = tool.runtime.tool_name();\n if !registered_mcp_tools.contains(&name) || shortlist.keeps(&name.to_string()) {\n continue;\n }\n tool.exposure = match (tool.exposure, search) {\n (ToolExposure::Direct, true) => ToolExposure::Deferred,\n (ToolExposure::DirectModelOnly, true) => ToolExposure::DeferredModelOnly,\n (exposure, _) if exposure.is_direct() => ToolExposure::Hidden,\n (exposure, _) => exposure,\n };\n }\n }\n",
|
||||
"count": 1
|
||||
},
|
||||
{
|
||||
"file": "core/src/tools/orchestrator.rs",
|
||||
"old": " let requirement = tool.exec_approval_requirement(req).unwrap_or_else(|| {\n default_exec_approval_requirement(approval_policy, &file_system_sandbox_policy)\n });\n",
|
||||
"new": " let requirement = tool.exec_approval_requirement(req).unwrap_or_else(|| {\n default_exec_approval_requirement(approval_policy, &file_system_sandbox_policy)\n });\n // Hanzo: the thread's controller joins the policy on allow < ask < deny; it never\n // loosens it.\n let requirement = match tool_ctx\n .session\n .services\n .thread_extension_data\n .get::<hanzo_loop::Handle>()\n {\n None => requirement,\n Some(handle) => {\n let policy = match &requirement {\n ExecApprovalRequirement::Skip { .. } => hanzo_loop::Verdict::Allow,\n ExecApprovalRequirement::NeedsApproval { .. } => hanzo_loop::Verdict::Ask,\n ExecApprovalRequirement::Forbidden { .. } => hanzo_loop::Verdict::Deny,\n };\n let action = tool\n .approval_action(req, &tool_ctx.call_id)\n .ok()\n .and_then(|action| serde_json::to_value(action).ok())\n .unwrap_or_default();\n let turn = &tool_ctx.step_context.turn.sub_id;\n match handle.command(turn, &tool_ctx.call_id, policy, action).await {\n verdict if verdict == policy => requirement,\n hanzo_loop::Verdict::Deny => ExecApprovalRequirement::Forbidden {\n reason: \"Kai refused this command.\".to_string(),\n },\n _ => ExecApprovalRequirement::NeedsApproval {\n reason: Some(\"Kai asks before this command runs.\".to_string()),\n proposed_execpolicy_amendment: None,\n },\n }\n }\n };\n",
|
||||
"count": 1
|
||||
},
|
||||
{
|
||||
"file": "core/src/session/turn.rs",
|
||||
"old": " let user_input = turn_user_input(&input);\n",
|
||||
"new": " let user_input = turn_user_input(&input);\n // Hanzo: the thread's controller adds routing hints to the turn's request metadata.\n if let Some(handle) = sess\n .services\n .thread_extension_data\n .get::<hanzo_loop::Handle>()\n {\n let request = hanzo_loop::Request {\n text: user_input\n .iter()\n .filter_map(|item| match item {\n UserInput::Text { text, .. } => Some(text.as_str()),\n _ => None,\n })\n .collect::<Vec<_>>()\n .join(\"\\n\"),\n images: user_input\n .iter()\n .any(|item| matches!(item, UserInput::Image { .. } | UserInput::LocalImage { .. })),\n };\n let hints = handle.route(&turn_context.sub_id, request).await;\n if !hints.is_empty() {\n let mut metadata = turn_context.config.responses_api_metadata.clone();\n metadata.extend(hints);\n turn_context\n .turn_metadata_state\n .set_responses_api_metadata(metadata);\n }\n }\n",
|
||||
"count": 1
|
||||
},
|
||||
{
|
||||
"file": "app-server/Cargo.toml",
|
||||
"old": "[dependencies]",
|
||||
"new": "[dependencies]\nhanzo-kai = { path = \"../../../../crates/hanzo-kai\" }",
|
||||
"count": 1
|
||||
},
|
||||
{
|
||||
"file": "app-server/src/extensions.rs",
|
||||
"old": " codex_guardian_v2::install(\n &mut builder,",
|
||||
"new": " // Hanzo: Kai decides in the loop; it runs only for a profile with a kai.toml.\n hanzo_kai::install(&mut builder, thread_manager.clone());\n codex_guardian_v2::install(\n &mut builder,",
|
||||
"count": 1
|
||||
}
|
||||
]
|
||||
|
||||
Reference in New Issue
Block a user