hanzoai/dev
2
0
Fork 0
mirror of https://github.com/hanzoai/dev.git synced 2026-10-17 15:34:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(dev): refuse a delivery without consuming what it names

Browse Source 
observed() and answered() removed the outstanding entry before comparing the
family it was dispatched to, so a result from the wrong family — refused,
answering with nothing — deleted the entry the real result needed: the turn
wedged forever, or resumed early and asked the model with an effect still in
flight. Compare, then remove.

Message becomes User, Agent{text, calls} and Tool{id, text, failed}, so the
transcript keeps the call each result answers instead of throwing it away; a
provider that needs the assistant tool call ahead of the result can be driven
from this state. The prelude moves to Ask, where a system prompt belongs.

A Turn carries the host's id, and the core refuses one at or below the highest
it has accepted, so a redelivered prompt runs once. A Timer asks for a Save only
when the sequence has moved. Done names the turn that asked, a prompt queued
mid-turn answers with a Save, and a cancel says what became of each — so N
prompts produce N Dones and nothing is accepted in silence.

A path in a call is workspace-relative; the core refuses one that is absolute or
climbs out rather than leaving every host to confine /etc/passwd itself. Patch
names the one file it edits. What may run and what may be fetched stays policy,
and policy is /v1/dev's.

A snapshot carries its ABI and a checksum of its body, so a truncated, foreign
or edited one is refused rather than restoring a session that re-mints ids the
host has already dispatched under. It is not a signature: there is no key here
and no identity.

Slab::put swallowed a failed lookup, so a dev_drop landing inside a step
answered DEV_OK and dropped the session it had just stepped. put hands the value
back, enter answers DEV_HANDLE and lends nothing out, and remove keeps a slot
whose value is on loan out of the free list until the call lets go. encode
returns a Result instead of panicking inside a library, and the doc no longer
claims dev_abi runs inside catch_unwind.

43 tests, and each fix is pinned: putting the defect back fails the test that
names it, 14 for 14.

Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
zeekay
2026-09-18 17:44:35 -07:00
co-authored by Hanzo Dev
parent ca4cfbefae
commit 4ddd7b6667
7 changed files with 1046 additions and 246 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/core/src/lib.rs
+143 -78
View File
@@ -6,15 +6,36 @@
//! `std::net` and no async runtime anywhere below this line, which is what lets
//! the same core run in cloud, in a terminal, and inside wasm.
//!
//! Two rules make the loop replayable:
//! # What is keyed, and what that buys
//!
//! - Every action gets the next id from a counter that is part of the
//! snapshot, so a restored session continues the sequence rather than
//! repeating it.
//! - An observation is accepted only while its id is outstanding. A duplicate
//! delivery of the same result changes nothing and produces no action, so
//! at-least-once delivery of an event is not at-least-once execution of an
//! effect.
//! Every action gets the next id from a counter that is part of the snapshot,
//! so a restored session continues the sequence rather than repeating it. On
//! top of that, each event class is keyed differently, and the difference is
//! the whole replay story:
//!
//! - A **result** ([`Event::Exec`], [`Event::File`], [`Event::Git`],
//! [`Event::Browse`], [`Event::Model`]) is accepted only while its id is
//! outstanding *and* was dispatched to that family. A second delivery, or
//! one from the wrong family, changes nothing and produces no action — and
//! in particular does not consume the entry the genuine result still needs.
//! - A **[`Event::Turn`]** carries the host's id. The core keeps the highest
//! it has accepted and refuses anything at or below it, so a redelivered
//! prompt is not run twice.
//! - A **[`Event::Timer`]** carries no id and needs none: it asks only for
//! [`Op::Save`], and only when the sequence has moved since the last one, so
//! a redelivered timer asks for nothing.
//! - A **[`Event::Cancel`]** is idempotent: there is one turn to stop.
//!
//! So at-least-once delivery of an event is at-most-once execution of an
//! effect, for every event this protocol has.
//!
//! # Every accepted prompt ends in a [`Done`] that names it
//!
//! A prompt that arrives mid-turn is queued, and queueing answers with an
//! [`Op::Save`] because the queue is state the host must not lose. A prompt
//! the core refuses answers with nothing at all, so the two are distinct at
//! the call. Whichever way the turn ends — its own answer, a chained queue, a
//! cancel — exactly one [`Op::Done`] names the turn id that asked for it.
//!
//! # What upstream this does not reuse, and why
//!
@@ -44,13 +65,15 @@ use dev_protocol::Answer;
use dev_protocol::Ask;
use dev_protocol::Call;
use dev_protocol::Config;
use dev_protocol::Dispatch;
use dev_protocol::Done;
use dev_protocol::Event;
use dev_protocol::Malformed;
use dev_protocol::Message;
use dev_protocol::Op;
use dev_protocol::Outcome;
use dev_protocol::Output;
use dev_protocol::Role;
use dev_protocol::Turn;
use serde::Deserialize;
use serde::Serialize;
@@ -71,30 +94,29 @@ pub struct Session {
messages: Vec<Message>,
/// The id the next action will carry.
next: u64,
/// The turn in flight, by the id the host gave it.
turn: Option<u64>,
/// The highest turn id accepted. One at or below it is a redelivery.
high: u64,
/// Actions dispatched and not yet answered.
outstanding: BTreeMap<u64, Wants>,
/// Prompts that arrived while a turn was in flight.
queue: VecDeque<String>,
/// A turn is in flight.
busy: bool,
queue: VecDeque<Turn>,
/// `next` as it stood when the last [`Op::Save`] was asked for.
saved: u64,
}
impl Session {
pub fn new(config: Config) -> Self {
let mut messages = Vec::new();
if let Some(prelude) = config.prelude.clone() {
messages.push(Message {
role: Role::Agent,
text: prelude,
});
}
Self {
config,
messages,
messages: Vec::new(),
next: 1,
turn: None,
high: 0,
outstanding: BTreeMap::new(),
queue: VecDeque::new(),
busy: false,
saved: 1,
}
}
@@ -111,7 +133,7 @@ impl Session {
/// Advance the machine.
pub fn step(&mut self, event: Event) -> Vec<Action> {
match event {
Event::Turn(turn) => self.turn(turn.prompt),
Event::Turn(turn) => self.turn(turn),
Event::Model(answer) => self.answered(answer),
Event::Exec(output) => self.observed(Wants::Exec, output),
Event::File(output) => self.observed(Wants::File, output),
@@ -123,57 +145,87 @@ impl Session {
}
/// The core's own snapshot — one of the three facts a session is.
pub fn snapshot(&self) -> Vec<u8> {
dev_protocol::encode(self)
pub fn snapshot(&self) -> Result<Vec<u8>, Malformed> {
dev_protocol::pack(self)
}
/// Rebuild a session from its snapshot.
pub fn restore(state: &[u8]) -> Result<Self, Malformed> {
dev_protocol::decode(state)
dev_protocol::unpack(state)
}
fn turn(&mut self, prompt: String) -> Vec<Action> {
if self.busy {
self.queue.push_back(prompt);
fn turn(&mut self, turn: Turn) -> Vec<Action> {
if turn.id <= self.high {
return Vec::new();
}
self.messages.push(Message {
role: Role::User,
text: prompt,
});
self.busy = true;
self.high = turn.id;
if self.turn.is_some() {
self.queue.push_back(turn);
return vec![self.save()];
}
self.turn = Some(turn.id);
self.messages.push(Message::User(turn.prompt));
vec![self.ask()]
}
fn answered(&mut self, answer: Answer) -> Vec<Action> {
if self.outstanding.remove(&answer.id) != Some(Wants::Model) {
let Some(turn) = self.turn else {
return Vec::new();
};
if self.outstanding.get(&answer.id) != Some(&Wants::Model) {
return Vec::new();
}
if let Some(text) = answer.reply.text.clone() {
self.messages.push(Message {
role: Role::Agent,
text,
});
}
self.outstanding.remove(&answer.id);
if answer.reply.calls.is_empty() {
return self.finish(answer.reply.text);
if let Some(text) = answer.reply.text.clone() {
self.messages.push(Message::Agent {
text: Some(text),
calls: Vec::new(),
});
}
return self.finish(turn, answer.reply.text);
}
answer
.reply
.calls
.into_iter()
.map(|call| self.dispatch(call))
.collect()
// Mint an id for every call, whether it goes out or not: the transcript
// records the call the model made and the answer it gets, and the two
// agree on the id even when the answer is a refusal.
let mut actions = Vec::new();
let mut calls = Vec::new();
let mut refused = Vec::new();
for call in answer.reply.calls {
let id = self.mint();
match call.escape() {
None => actions.push(self.dispatch(id, call.clone())),
Some(path) => refused.push(Message::Tool {
id,
text: format!("refused: {path} is not inside the workspace"),
failed: true,
}),
}
calls.push(Dispatch { id, call });
}
self.messages.push(Message::Agent {
text: answer.reply.text,
calls,
});
self.messages.extend(refused);
if self.outstanding.is_empty() {
// Every call was refused, so nothing will arrive to resume the
// turn: tell the model now, with the refusals in the transcript.
actions.push(self.ask());
}
actions
}
fn observed(&mut self, wants: Wants, output: Output) -> Vec<Action> {
if self.outstanding.remove(&output.id) != Some(wants) {
if self.outstanding.get(&output.id) != Some(&wants) {
return Vec::new();
}
let mark = if output.failed { "failed" } else { "ok" };
self.messages.push(Message {
role: Role::Tool,
text: format!("{} {mark}: {}", output.id, output.text),
self.outstanding.remove(&output.id);
self.messages.push(Message::Tool {
id: output.id,
text: output.text,
failed: output.failed,
});
if self.outstanding.is_empty() {
vec![self.ask()]
@@ -183,45 +235,41 @@ impl Session {
}
fn tick(&mut self) -> Vec<Action> {
if self.busy {
vec![self.action(Op::Save)]
if self.turn.is_some() && self.next != self.saved {
vec![self.save()]
} else {
Vec::new()
}
}
fn cancel(&mut self) -> Vec<Action> {
if !self.busy {
let Some(turn) = self.turn.take() else {
return Vec::new();
}
};
self.outstanding.clear();
self.queue.clear();
self.busy = false;
vec![
self.action(Op::Save),
self.action(Op::Done(Outcome::Cancelled)),
]
let dropped: Vec<u64> = self.queue.drain(..).map(|turn| turn.id).collect();
let mut actions = vec![self.save(), self.done(turn, Outcome::Cancelled)];
for queued in dropped {
actions.push(self.done(queued, Outcome::Cancelled));
}
actions
}
/// End the turn, or start the next queued prompt instead.
fn finish(&mut self, text: Option<String>) -> Vec<Action> {
/// End the turn, then start the next queued prompt if there is one.
fn finish(&mut self, turn: u64, text: Option<String>) -> Vec<Action> {
let mut actions = Vec::new();
if let Some(text) = text {
actions.push(self.action(Op::Emit(text)));
}
actions.push(self.action(Op::Save));
actions.push(self.save());
actions.push(self.done(turn, Outcome::Complete));
match self.queue.pop_front() {
Some(prompt) => {
self.messages.push(Message {
role: Role::User,
text: prompt,
});
Some(next) => {
self.turn = Some(next.id);
self.messages.push(Message::User(next.prompt));
actions.push(self.ask());
}
None => {
self.busy = false;
actions.push(self.action(Op::Done(Outcome::Complete)));
}
None => self.turn = None,
}
actions
}
@@ -229,6 +277,7 @@ impl Session {
fn ask(&mut self) -> Action {
let op = Op::Model(Ask {
model: self.config.model.clone(),
prelude: self.config.prelude.clone(),
messages: self.messages.clone(),
});
let action = self.action(op);
@@ -236,7 +285,7 @@ impl Session {
action
}
fn dispatch(&mut self, call: Call) -> Action {
fn dispatch(&mut self, id: u64, call: Call) -> Action {
let (op, wants) = match call {
Call::Read(read) => (Op::Read(read), Wants::File),
Call::Write(write) => (Op::Write(write), Wants::File),
@@ -245,14 +294,30 @@ impl Session {
Call::Git(git) => (Op::Git(git), Wants::Git),
Call::Browse(browse) => (Op::Browse(browse), Wants::Browse),
};
let action = self.action(op);
self.outstanding.insert(action.id, wants);
self.outstanding.insert(id, wants);
Action { id, op }
}
fn save(&mut self) -> Action {
let action = self.action(Op::Save);
self.saved = self.next;
action
}
fn done(&mut self, turn: u64, outcome: Outcome) -> Action {
self.action(Op::Done(Done { turn, outcome }))
}
fn action(&mut self, op: Op) -> Action {
Action {
id: self.mint(),
op,
}
}
fn mint(&mut self) -> u64 {
let id = self.next;
self.next += 1;
Action { id, op }
id
}
}
crates/core/tests/session.rs
+371 -83
View File
@@ -3,18 +3,24 @@
use dev_core::Session;
use dev_protocol::Action;
use dev_protocol::Answer;
use dev_protocol::Ask;
use dev_protocol::Call;
use dev_protocol::Config;
use dev_protocol::Dispatch;
use dev_protocol::Done;
use dev_protocol::Event;
use dev_protocol::Exec;
use dev_protocol::Message;
use dev_protocol::Op;
use dev_protocol::Outcome;
use dev_protocol::Output;
use dev_protocol::Read;
use dev_protocol::Reply;
use dev_protocol::Turn;
fn turn(prompt: &str) -> Event {
fn turn(id: u64, prompt: &str) -> Event {
Event::Turn(Turn {
id,
prompt: prompt.to_string(),
})
}
@@ -43,50 +49,73 @@ fn test() -> Call {
})
}
fn read(path: &str) -> Call {
Call::Read(Read {
path: path.to_string(),
})
}
fn result(id: u64, text: &str) -> Output {
Output {
id,
text: text.to_string(),
failed: false,
}
}
fn done(turn: u64, outcome: Outcome) -> Op {
Op::Done(Done { turn, outcome })
}
fn ops(actions: &[Action]) -> Vec<&Op> {
actions.iter().map(|a| &a.op).collect()
}
/// The request an [`Op::Model`] action carries, or a failure naming what it
/// carried instead.
fn asked(action: &Action) -> &Ask {
match &action.op {
Op::Model(ask) => ask,
op => panic!("asked for {op:?}"),
}
}
#[test]
fn a_turn_asks_a_model_then_runs_the_tool_then_asks_again() {
let mut session = Session::new(Config::default());
let asked = session.step(turn("fix the build"));
assert_eq!(asked.len(), 1);
assert_eq!(asked[0].id, 1);
assert!(matches!(asked[0].op, Op::Model(_)));
let opened = session.step(turn(1, "fix the build"));
assert_eq!(opened.len(), 1);
assert_eq!(opened[0].id, 1);
assert!(matches!(opened[0].op, Op::Model(_)));
let dispatched = session.step(tools(vec![test()], 1));
assert_eq!(dispatched.len(), 1);
assert_eq!(dispatched[0].id, 2);
assert!(matches!(dispatched[0].op, Op::Exec(_)));
let again = session.step(Event::Exec(Output {
id: 2,
text: "ok".to_string(),
failed: false,
}));
let again = session.step(Event::Exec(result(2, "ok")));
assert_eq!(again.len(), 1);
assert_eq!(again[0].id, 3);
assert!(matches!(again[0].op, Op::Model(_)));
let done = session.step(says("built", 3));
let ended = session.step(says("built", 3));
assert_eq!(
ops(&done),
ops(&ended),
vec![
&Op::Emit("built".to_string()),
&Op::Save,
&Op::Done(Outcome::Complete),
&done(1, Outcome::Complete),
]
);
assert_eq!(done[0].id, 4);
assert_eq!(done[2].id, 6);
assert_eq!(ended[0].id, 4);
assert_eq!(ended[2].id, 6);
}
#[test]
fn every_call_of_one_reply_gets_its_own_id() {
let mut session = Session::new(Config::default());
session.step(turn("run the suite twice"));
session.step(turn(1, "run the suite twice"));
let dispatched = session.step(tools(vec![test(), test()], 1));
let ids: Vec<u64> = dispatched.iter().map(|a| a.id).collect();
@@ -94,52 +123,34 @@ fn every_call_of_one_reply_gets_its_own_id() {
// One answer is not enough: the model is asked again only when the last
// outstanding effect has been observed.
let quiet = session.step(Event::Exec(Output {
id: 2,
text: "ok".to_string(),
failed: false,
}));
assert!(quiet.is_empty());
let asked = session.step(Event::Exec(Output {
id: 3,
text: "ok".to_string(),
failed: false,
}));
assert_eq!(asked.len(), 1);
assert!(matches!(asked[0].op, Op::Model(_)));
assert!(session.step(Event::Exec(result(2, "ok"))).is_empty());
let again = session.step(Event::Exec(result(3, "ok")));
assert_eq!(again.len(), 1);
assert!(matches!(again[0].op, Op::Model(_)));
}
#[test]
fn a_replayed_result_runs_nothing_twice() {
let mut session = Session::new(Config::default());
session.step(turn("fix the build"));
session.step(tools(vec![test()], 1));
let answered = Event::Exec(Output {
id: 2,
text: "ok".to_string(),
failed: false,
fn the_prelude_rides_beside_the_conversation_rather_than_inside_it() {
let mut session = Session::new(Config {
model: Some("zen5.8".to_string()),
prelude: Some("you are a coding agent".to_string()),
});
assert_eq!(session.step(answered.clone()).len(), 1);
assert!(session.step(answered).is_empty());
// The same id delivered by the wrong family is refused too.
assert!(
session
.step(Event::Git(Output {
id: 2,
text: "ok".to_string(),
failed: false,
}))
.is_empty()
let opened = session.step(turn(1, "fix the build"));
let ask = asked(&opened[0]);
assert_eq!(ask.model.as_deref(), Some("zen5.8"));
assert_eq!(ask.prelude.as_deref(), Some("you are a coding agent"));
assert_eq!(
ask.messages,
vec![Message::User("fix the build".to_string())]
);
}
#[test]
fn snapshot_and_restore_continue_at_the_same_id() {
let mut session = Session::new(Config::default());
session.step(turn("fix the build"));
session.step(turn(1, "fix the build"));
let state = session.snapshot();
let state = session.snapshot().expect("snapshot");
let mut restored = Session::restore(&state).expect("restore");
assert_eq!(restored.next_id(), session.next_id());
assert_eq!(restored.messages(), session.messages());
@@ -150,28 +161,12 @@ fn snapshot_and_restore_continue_at_the_same_id() {
assert_eq!(there[0].id, 2);
}
#[test]
fn a_prompt_mid_turn_waits_for_the_turn() {
let mut session = Session::new(Config::default());
session.step(turn("first"));
assert!(session.step(turn("second")).is_empty());
let actions = session.step(says("first done", 1));
assert_eq!(actions.len(), 3);
assert_eq!(actions[0].op, Op::Emit("first done".to_string()));
assert_eq!(actions[1].op, Op::Save);
assert!(matches!(actions[2].op, Op::Model(_)));
let closed = session.step(says("second done", actions[2].id));
assert!(matches!(closed.last().map(|a| &a.op), Some(Op::Done(_))));
}
#[test]
fn a_timer_checkpoints_only_inside_a_turn() {
let mut session = Session::new(Config::default());
assert!(session.step(Event::Timer).is_empty());
session.step(turn("fix the build"));
session.step(turn(1, "fix the build"));
let saved = session.step(Event::Timer);
assert_eq!(ops(&saved), vec![&Op::Save]);
}
@@ -179,30 +174,323 @@ fn a_timer_checkpoints_only_inside_a_turn() {
#[test]
fn cancel_ends_the_turn_and_drops_what_was_outstanding() {
let mut session = Session::new(Config::default());
session.step(turn("fix the build"));
session.step(turn(1, "fix the build"));
session.step(tools(vec![test()], 1));
let stopped = session.step(Event::Cancel);
assert_eq!(
ops(&stopped),
vec![&Op::Save, &Op::Done(Outcome::Cancelled)]
);
assert_eq!(ops(&stopped), vec![&Op::Save, &done(1, Outcome::Cancelled)]);
// A result for the cancelled effect is no longer outstanding.
assert!(
session
.step(Event::Exec(Output {
id: 2,
text: "ok".to_string(),
failed: false,
}))
.is_empty()
);
assert!(session.step(Event::Exec(result(2, "ok"))).is_empty());
assert!(session.step(Event::Cancel).is_empty());
}
#[test]
fn an_answer_to_an_id_that_was_never_asked_is_refused() {
let mut session = Session::new(Config::default());
session.step(turn("fix the build"));
session.step(turn(1, "fix the build"));
assert!(session.step(says("hello", 99)).is_empty());
}
#[test]
fn a_failed_result_reaches_the_model_as_a_failure() {
let mut session = Session::new(Config::default());
session.step(turn(1, "fix the build"));
session.step(tools(vec![test()], 1));
let again = session.step(Event::Exec(Output {
id: 2,
text: "2 tests failed".to_string(),
failed: true,
}));
assert_eq!(
asked(&again[0]).messages.last(),
Some(&Message::Tool {
id: 2,
text: "2 tests failed".to_string(),
failed: true,
})
);
}
// Everything below is a delivery the core must refuse. A refusal answers with
// no action, and — the part that took a wedged turn to learn — changes nothing.
#[test]
fn a_wrong_family_result_leaves_the_entry_for_the_right_one() {
let mut session = Session::new(Config::default());
session.step(turn(1, "fix the build"));
session.step(tools(vec![test()], 1));
// Id 2 was dispatched as an Exec. A Git result naming it is refused...
assert!(session.step(Event::Git(result(2, "ok"))).is_empty());
// ...and the Exec result it was standing in front of still lands.
let again = session.step(Event::Exec(result(2, "ok")));
assert_eq!(again.len(), 1);
assert!(matches!(again[0].op, Op::Model(_)));
let ended = session.step(says("built", again[0].id));
assert_eq!(
ended.last().map(|a| &a.op),
Some(&done(1, Outcome::Complete))
);
}
#[test]
fn a_wrong_family_result_does_not_resume_the_turn_early() {
let mut session = Session::new(Config::default());
session.step(turn(1, "run the suite twice"));
session.step(tools(vec![test(), test()], 1));
assert!(session.step(Event::Git(result(2, "ok"))).is_empty());
// Two effects are still in flight, so the second result is not the last.
assert!(session.step(Event::Exec(result(3, "second"))).is_empty());
let again = session.step(Event::Exec(result(2, "first")));
assert_eq!(again.len(), 1);
// The model is asked once both results are in the transcript, never with
// one of them missing.
let results: Vec<&Message> = asked(&again[0])
.messages
.iter()
.filter(|message| matches!(message, Message::Tool { .. }))
.collect();
assert_eq!(
results,
vec![
&Message::Tool {
id: 3,
text: "second".to_string(),
failed: false,
},
&Message::Tool {
id: 2,
text: "first".to_string(),
failed: false,
},
]
);
}
#[test]
fn a_model_event_naming_a_tool_id_leaves_the_entry_alone() {
let mut session = Session::new(Config::default());
session.step(turn(1, "fix the build"));
session.step(tools(vec![test()], 1));
// Id 2 is an Exec, not the model request.
assert!(session.step(says("here you go", 2)).is_empty());
assert_eq!(session.step(Event::Exec(result(2, "ok"))).len(), 1);
}
#[test]
fn a_replayed_result_runs_nothing_twice() {
let mut session = Session::new(Config::default());
session.step(turn(1, "fix the build"));
session.step(tools(vec![test()], 1));
assert_eq!(session.step(Event::Exec(result(2, "ok"))).len(), 1);
assert!(session.step(Event::Exec(result(2, "ok"))).is_empty());
}
#[test]
fn a_redelivered_turn_runs_the_prompt_once() {
let mut session = Session::new(Config::default());
let opened = session.step(turn(1, "ship it"));
assert_eq!(opened.len(), 1);
assert!(session.step(turn(1, "ship it")).is_empty());
// The model is asked once, with one prompt in front of it.
let ended = session.step(says("shipped", 1));
assert_eq!(
ended.last().map(|a| &a.op),
Some(&done(1, Outcome::Complete))
);
let again = session.step(turn(2, "and again"));
assert_eq!(
asked(&again[0]).messages,
vec![
Message::User("ship it".to_string()),
Message::Agent {
text: Some("shipped".to_string()),
calls: Vec::new(),
},
Message::User("and again".to_string()),
]
);
}
#[test]
fn a_redelivered_timer_asks_for_nothing() {
let mut session = Session::new(Config::default());
session.step(turn(1, "fix the build"));
assert_eq!(ops(&session.step(Event::Timer)), vec![&Op::Save]);
assert!(session.step(Event::Timer).is_empty());
assert!(session.step(Event::Timer).is_empty());
// The next one checkpoints what the turn has moved on to since.
session.step(tools(vec![test()], 1));
assert_eq!(ops(&session.step(Event::Timer)), vec![&Op::Save]);
}
#[test]
fn the_transcript_keeps_the_call_each_result_answers() {
let mut session = Session::new(Config::default());
session.step(turn(1, "fix the build"));
session.step(tools(vec![test()], 1));
let again = session.step(Event::Exec(result(2, "ok")));
assert_eq!(
asked(&again[0]).messages,
vec![
Message::User("fix the build".to_string()),
Message::Agent {
text: None,
calls: vec![Dispatch {
id: 2,
call: test(),
}],
},
Message::Tool {
id: 2,
text: "ok".to_string(),
failed: false,
},
]
);
}
#[test]
fn a_prompt_that_arrives_mid_turn_is_acknowledged_and_ends_in_its_own_done() {
let mut session = Session::new(Config::default());
session.step(turn(1, "first"));
// Accepted, so the queue is state worth persisting — and the host can tell
// this apart from the empty answer a refusal gives.
assert_eq!(ops(&session.step(turn(2, "second"))), vec![&Op::Save]);
let chained = session.step(says("first done", 1));
assert_eq!(chained.len(), 4);
assert_eq!(chained[0].op, Op::Emit("first done".to_string()));
assert_eq!(chained[1].op, Op::Save);
assert_eq!(chained[2].op, done(1, Outcome::Complete));
assert!(matches!(chained[3].op, Op::Model(_)));
let closed = session.step(says("second done", chained[3].id));
assert_eq!(
closed.last().map(|a| &a.op),
Some(&done(2, Outcome::Complete))
);
}
#[test]
fn a_cancel_says_what_became_of_every_prompt_it_discards() {
let mut session = Session::new(Config::default());
session.step(turn(1, "first"));
session.step(turn(2, "second"));
session.step(turn(3, "third"));
let stopped = session.step(Event::Cancel);
assert_eq!(
ops(&stopped),
vec![
&Op::Save,
&done(1, Outcome::Cancelled),
&done(2, Outcome::Cancelled),
&done(3, Outcome::Cancelled),
]
);
}
#[test]
fn a_path_that_leaves_the_workspace_is_refused_rather_than_dispatched() {
let mut session = Session::new(Config::default());
session.step(turn(1, "read my keys"));
let dispatched = session.step(tools(
vec![read("../../../../etc/passwd"), read("src/lib.rs")],
1,
));
// Only the confined read goes out; the other never becomes an action.
assert_eq!(dispatched.len(), 1);
assert_eq!(
ops(&dispatched),
vec![&Op::Read(Read {
path: "src/lib.rs".to_string()
})]
);
assert_eq!(dispatched[0].id, 3);
// The model is told, under the id its own call was recorded with.
let again = session.step(Event::File(result(3, "the file")));
assert_eq!(
asked(&again[0]).messages[1..],
[
Message::Agent {
text: None,
calls: vec![
Dispatch {
id: 2,
call: read("../../../../etc/passwd"),
},
Dispatch {
id: 3,
call: read("src/lib.rs"),
},
],
},
Message::Tool {
id: 2,
text: "refused: ../../../../etc/passwd is not inside the workspace".to_string(),
failed: true,
},
Message::Tool {
id: 3,
text: "the file".to_string(),
failed: false,
},
]
);
}
#[test]
fn a_turn_whose_every_call_is_refused_asks_again_instead_of_waiting() {
let mut session = Session::new(Config::default());
session.step(turn(1, "read my keys"));
let answered = session.step(tools(vec![read("/Users/z/.ssh/authorized_keys")], 1));
assert_eq!(answered.len(), 1, "{:?}", ops(&answered));
let ask = asked(&answered[0]);
assert_eq!(
ask.messages.last(),
Some(&Message::Tool {
id: 2,
text: "refused: /Users/z/.ssh/authorized_keys is not inside the workspace".to_string(),
failed: true,
})
);
// And the turn is still live: it ends the ordinary way.
let ended = session.step(says("I cannot read that", answered[0].id));
assert_eq!(
ended.last().map(|a| &a.op),
Some(&done(1, Outcome::Complete))
);
}
#[test]
fn a_snapshot_the_store_handed_back_changed_is_refused() {
let mut session = Session::new(Config::default());
session.step(turn(1, "fix the build"));
let state = session.snapshot().expect("snapshot");
// Re-mint an id the host has already dispatched under.
let stamp = 12;
let body = String::from_utf8(state[stamp..].to_vec()).expect("a json body");
let forged = body.replace("\"next\":2", "\"next\":1");
assert_ne!(forged, body, "the snapshot no longer spells the next id");
let mut bytes = state[..stamp].to_vec();
bytes.extend_from_slice(forged.as_bytes());
let err = Session::restore(&bytes).expect_err("a forged snapshot was restored");
assert!(err.to_string().contains("checksum"), "{err}");
}
crates/ffi/include/dev.h
+6 -1
View File
@@ -9,9 +9,14 @@
* - Every dev_buf handed back is a Rust allocation; release it with
* dev_free and never with free(3).
* - A session handle is opaque and generational: dev_drop retires it, and a
* retired handle answers DEV_HANDLE instead of naming a live session.
* retired handle answers DEV_HANDLE instead of naming a live session. A
* dev_drop that lands while a call on that session is running makes the
* call answer DEV_HANDLE and hand back no dev_buf: there is no session
* left for its answer to belong to.
* - A call that panics answers DEV_PANIC and poisons that session; its later
* calls answer DEV_POISON until dev_drop. Other sessions are untouched.
* - A dev_buf is written only on DEV_OK. On any other status the out
* parameter is left as the caller passed it.
*
* Payloads are dev-protocol messages (Config, Event, Action list, snapshot) in
* the encoding dev_abi() names.
crates/ffi/src/lib.rs
+138 -26
View File
@@ -9,11 +9,16 @@
//! a Rust allocation; `free(3)` on it is wrong.
//! - **A handle is generational.** Dropping a session retires its handle, and
//! a stale handle is refused with [`DEV_HANDLE`] rather than resolving to
//! whatever took the slot next.
//! - **A panic does not cross the boundary.** Every entry point runs inside
//! `catch_unwind`; an escaped panic answers [`DEV_PANIC`] and poisons that
//! session, whose every later call answers [`DEV_POISON`] until it is
//! dropped. The process stays up and the other sessions are untouched.
//! whatever took the slot next. A session dropped while one of its calls is
//! running makes that call answer [`DEV_HANDLE`] too, and lend nothing out:
//! there is no session left for the answer to belong to.
//! - **A panic does not cross the boundary.** Every entry point that has a
//! status to answer with runs inside `catch_unwind`; an escaped panic answers
//! [`DEV_PANIC`] and poisons that session, whose every later call answers
//! [`DEV_POISON`] until it is dropped. The process stays up and the other
//! sessions are untouched. [`dev_abi`] is the exception and has no status
//! channel to answer on: it reads a constant, so there is nothing in it to
//! panic.
mod slab;
@@ -124,15 +129,21 @@ pub unsafe extern "C" fn dev_step(
let Some(bytes) = (unsafe { copy(event, len) }) else {
return DEV_NULL;
};
enter(session, |session| {
let (status, answer) = enter(session, |session| {
let event: Event = match dev_protocol::decode(&bytes) {
Ok(event) => event,
Err(_) => return DEV_MALFORMED,
Err(_) => return (DEV_MALFORMED, Vec::new()),
};
let actions = session.step(event);
unsafe { out.write(lend(dev_protocol::encode(&actions))) };
DEV_OK
})
match dev_protocol::encode(&actions) {
Ok(bytes) => (DEV_OK, bytes),
Err(_) => (DEV_MALFORMED, Vec::new()),
}
});
if status == DEV_OK {
unsafe { out.write(lend(answer)) };
}
status
})
}
@@ -147,10 +158,14 @@ pub unsafe extern "C" fn dev_snapshot(session: u64, out: *mut Buf) -> i32 {
if out.is_null() {
return DEV_NULL;
}
enter(session, |session| {
unsafe { out.write(lend(session.snapshot())) };
DEV_OK
})
let (status, state) = enter(session, |session| match session.snapshot() {
Ok(bytes) => (DEV_OK, bytes),
Err(_) => (DEV_MALFORMED, Vec::new()),
});
if status == DEV_OK {
unsafe { out.write(lend(state)) };
}
status
})
}
@@ -207,20 +222,28 @@ pub unsafe extern "C" fn dev_free(buf: Buf) {
/// Run a call with the session lifted out of the slab, so no lock is held
/// while it runs and a panic leaves the slot poisoned rather than occupied.
fn enter(handle: u64, call: impl FnOnce(&mut Session) -> i32) -> i32 {
///
/// The bytes the call produced reach the caller only once the session is back
/// in its slot. A [`dev_drop`] that lands in between retires the handle, the
/// slab refuses the session, and the call answers [`DEV_HANDLE`] with nothing:
/// a host that has dropped a session is never handed actions it would have to
/// account for against one.
fn enter(handle: u64, call: impl FnOnce(&mut Session) -> (i32, Vec<u8>)) -> (i32, Vec<u8>) {
let mut session = match sessions().take(handle) {
Ok(session) => session,
Err(fault) => return status(fault),
Err(fault) => return (status(fault), Vec::new()),
};
match catch_unwind(AssertUnwindSafe(|| call(&mut session))) {
Ok(status) => {
sessions().put(handle, session);
status
}
Ok(answer) => match sessions().put(handle, session) {
Ok(()) => answer,
// The slab handed the session back because the handle is retired.
// It dies here, with its answer, which is what dropping it meant.
Err(_) => (DEV_HANDLE, Vec::new()),
},
Err(_) => {
drop(session);
sessions().poison(handle);
DEV_PANIC
(DEV_PANIC, Vec::new())
}
}
}
@@ -286,6 +309,8 @@ mod tests {
use dev_protocol::Config;
use dev_protocol::Op;
use dev_protocol::Turn;
use std::sync::atomic::AtomicU64;
use std::sync::atomic::Ordering;
fn start() -> u64 {
let mut handle = 0u64;
@@ -298,10 +323,11 @@ mod tests {
}
fn step(handle: u64, event: &Event) -> Result<Vec<Action>, i32> {
let bytes = dev_protocol::encode(event);
let bytes = dev_protocol::encode(event).expect("encode");
let mut out = Buf::empty();
let status = unsafe { dev_step(handle, bytes.as_ptr(), bytes.len(), &raw mut out) };
if status != DEV_OK {
assert!(out.ptr.is_null(), "a refused call lent a buffer out");
return Err(status);
}
let actions = dev_protocol::decode(unsafe {
@@ -313,7 +339,11 @@ mod tests {
}
fn turn(prompt: &str) -> Event {
// Each turn needs an id the session has not accepted before; a session
// refuses a redelivery, which is the point of the id.
static NEXT: AtomicU64 = AtomicU64::new(1);
Event::Turn(Turn {
id: NEXT.fetch_add(1, Ordering::Relaxed),
prompt: prompt.to_string(),
})
}
@@ -338,7 +368,8 @@ mod tests {
let cfg = dev_protocol::encode(&Config {
model: Some("zen5.8".to_string()),
prelude: Some("you are a coding agent".to_string()),
});
})
.expect("encode");
let mut handle = 0u64;
assert_eq!(
unsafe { dev_new(cfg.as_ptr(), cfg.len(), &raw mut handle) },
@@ -348,7 +379,8 @@ mod tests {
match &actions[0].op {
Op::Model(ask) => {
assert_eq!(ask.model.as_deref(), Some("zen5.8"));
assert_eq!(ask.messages.len(), 2);
assert_eq!(ask.prelude.as_deref(), Some("you are a coding agent"));
assert_eq!(ask.messages.len(), 1);
}
op => panic!("asked for {op:?}"),
}
@@ -385,9 +417,10 @@ mod tests {
let handle = start();
let quiet = std::panic::take_hook();
std::panic::set_hook(Box::new(|_| {}));
let status = enter(handle, |_| panic!("the loop gave up"));
let (status, answer) = enter(handle, |_| panic!("the loop gave up"));
std::panic::set_hook(quiet);
assert_eq!(status, DEV_PANIC);
assert!(answer.is_empty());
assert_eq!(step(handle, &turn("fix the build")), Err(DEV_POISON));
let mut out = Buf::empty();
@@ -417,7 +450,7 @@ mod tests {
#[test]
fn a_null_out_pointer_is_refused() {
let handle = start();
let bytes = dev_protocol::encode(&turn("fix the build"));
let bytes = dev_protocol::encode(&turn("fix the build")).expect("encode");
assert_eq!(
unsafe { dev_step(handle, bytes.as_ptr(), bytes.len(), std::ptr::null_mut()) },
DEV_NULL
@@ -460,4 +493,83 @@ mod tests {
dev_drop(handle);
dev_drop(restored);
}
#[test]
fn an_edited_snapshot_is_refused() {
let handle = start();
step(handle, &turn("fix the build")).expect("step");
let mut state = Buf::empty();
assert_eq!(unsafe { dev_snapshot(handle, &raw mut state) }, DEV_OK);
let mut bytes =
unsafe { std::slice::from_raw_parts(state.ptr.cast_const(), state.len) }.to_vec();
unsafe { dev_free(state) };
dev_drop(handle);
let body = String::from_utf8(bytes.split_off(12)).expect("a json body");
let forged = body.replace("\"next\":2", "\"next\":1");
assert_ne!(forged, body, "the snapshot no longer spells the next id");
bytes.extend_from_slice(forged.as_bytes());
let mut restored = 0u64;
assert_eq!(
unsafe { dev_restore(bytes.as_ptr(), bytes.len(), &raw mut restored) },
DEV_MALFORMED
);
assert_eq!(restored, 0);
}
/// The window `enter` opens on purpose: the lock is not held while the
/// session runs, so a `dev_drop` can land in the middle of a step.
#[test]
fn a_drop_during_a_step_answers_for_a_handle_that_is_gone() {
let handle = start();
let (status, answer) = enter(handle, |session| {
dev_drop(handle);
let actions = session.step(turn("fix the build"));
assert_eq!(actions.len(), 1, "the session did step");
(DEV_OK, dev_protocol::encode(&actions).expect("encode"))
});
assert_eq!(status, DEV_HANDLE, "a lost session answered OK");
assert!(answer.is_empty(), "actions for a session that is gone");
assert_eq!(step(handle, &turn("fix the build")), Err(DEV_HANDLE));
}
/// Two threads, one racing `dev_drop` against the other's `dev_step`, over
/// and over. This is the slab under contention: take, put, remove and the
/// free list all interleaving. It pins the answers — only [`DEV_OK`] or
/// [`DEV_HANDLE`], a buffer only on `DEV_OK`, a live session for anything
/// that claims one — and it is not what pins the lost-session fix, because
/// a handle is dead afterwards either way. That is
/// `a_drop_during_a_step_answers_for_a_handle_that_is_gone`, which times the
/// drop inside the window instead of racing for it.
#[test]
fn a_race_between_a_step_and_a_drop_never_reports_a_lost_session() {
for _ in 0..500 {
let handle = start();
let stepping = std::thread::spawn(move || step(handle, &turn("fix the build")));
let dropping = std::thread::spawn(move || dev_drop(handle));
let stepped = stepping.join().expect("stepping thread");
dropping.join().expect("dropping thread");
match stepped {
Ok(actions) => {
assert_eq!(actions.len(), 1);
assert!(matches!(actions[0].op, Op::Model(_)));
// The step won the race, so its session must still be there
// for the host to snapshot — or already dropped by the
// other thread, never a slot handed to someone else.
let mut out = Buf::empty();
let status = unsafe { dev_snapshot(handle, &raw mut out) };
assert!(
status == DEV_OK || status == DEV_HANDLE,
"snapshot answered {status}"
);
if status == DEV_OK {
unsafe { dev_free(out) };
}
}
Err(status) => assert_eq!(status, DEV_HANDLE, "the step answered {status}"),
}
dev_drop(handle);
}
}
}
crates/ffi/src/slab.rs
+94 -8
View File
@@ -1,6 +1,18 @@
//! A generational slab: a handle names a slot and the generation it was born
//! in, so a handle that outlives its session is refused instead of hitting
//! whatever took the slot next.
//!
//! A call lifts its value out of the slot for as long as it runs ([`Slab::take`]
//! then [`Slab::put`]), which is the window in which the handle can be retired
//! under it. Two rules close that window:
//!
//! - [`Slab::remove`] keeps a slot whose value is out on loan out of the free
//! list, so no [`Slab::insert`] can hand the slot to a new session while the
//! older call is still running.
//! - [`Slab::put`] refuses the value when the handle has been retired and hands
//! it back, so the caller answers for a session that no longer exists instead
//! of dropping its work on the floor. That is also where the slot rejoins the
//! free list.
/// Why a handle did not resolve.
#[derive(Clone, Copy, Debug, PartialEq)]
@@ -19,6 +31,9 @@ struct Slot<T> {
generation: u32,
value: Option<T>,
poisoned: bool,
/// Retired while its value was on loan to a running call. The slot stays
/// out of the free list until that call hands the value back.
orphan: bool,
}
pub struct Slab<T> {
@@ -54,6 +69,7 @@ impl<T> Slab<T> {
generation: 1,
value: Some(value),
poisoned: false,
orphan: false,
});
Ok(handle(index, 1))
}
@@ -70,19 +86,30 @@ impl<T> Slab<T> {
slot.value.take().ok_or(Fault::Busy)
}
/// Put the value back after the call returned.
pub fn put(&mut self, handle: u64, value: T) {
if let Ok(slot) = self.slot(handle) {
slot.value = Some(value);
/// Put the value back after the call returned, or hand it back when the
/// handle was retired while the call ran.
pub fn put(&mut self, handle: u64, value: T) -> Result<(), T> {
match self.slot(handle) {
Ok(slot) => {
slot.value = Some(value);
Ok(())
}
Err(_) => {
self.release(handle);
Err(value)
}
}
}
/// Mark the slot as poisoned: the value is gone and the handle answers
/// [`Fault::Poisoned`] until it is dropped.
pub fn poison(&mut self, handle: u64) {
if let Ok(slot) = self.slot(handle) {
slot.value = None;
slot.poisoned = true;
match self.slot(handle) {
Ok(slot) => {
slot.value = None;
slot.poisoned = true;
}
Err(_) => self.release(handle),
}
}
@@ -90,18 +117,35 @@ impl<T> Slab<T> {
pub fn remove(&mut self, handle: u64) -> Result<(), Fault> {
let (index, _) = parts(handle);
let slot = self.slot(handle)?;
let lent = slot.value.is_none() && !slot.poisoned;
slot.value = None;
slot.poisoned = false;
slot.orphan = lent;
slot.generation = slot.generation.wrapping_add(1);
if slot.generation == 0 {
// Generation zero would make a handle collide with the reserved
// zero handle; skip it.
slot.generation = 1;
}
self.free.push(index);
if !lent {
self.free.push(index);
}
Ok(())
}
/// Return a slot that was retired mid-call to the free list, now that the
/// call has let go of it.
fn release(&mut self, handle: u64) {
let (index, _) = parts(handle);
let Some(slot) = self.slots.get_mut(index as usize) else {
return;
};
if slot.orphan {
slot.orphan = false;
self.free.push(index);
}
}
fn slot(&mut self, handle: u64) -> Result<&mut Slot<T>, Fault> {
let (index, generation) = parts(handle);
let slot = self.slots.get_mut(index as usize).ok_or(Fault::Stale)?;
@@ -156,4 +200,46 @@ mod tests {
slab.remove(handle).expect("remove");
assert_eq!(slab.take(handle), Err(Fault::Stale));
}
#[test]
fn a_slot_retired_mid_call_is_not_reused_until_the_call_lets_go() {
let mut slab: Slab<u32> = Slab::new();
let handle = slab.insert(7).expect("insert");
let lent = slab.take(handle).expect("take");
slab.remove(handle).expect("remove");
// The slot the running call still holds is not handed to a new session.
let other = slab.insert(9).expect("insert");
assert_ne!(parts(other).0, parts(handle).0);
// The value comes back to the caller, which is how it learns the
// session it stepped is gone.
assert_eq!(slab.put(handle, lent), Err(7));
// Only now is the slot free again, and its handles are all stale.
let next = slab.insert(11).expect("insert");
assert_eq!(parts(next).0, parts(handle).0);
assert_ne!(next, handle);
assert_eq!(slab.take(handle), Err(Fault::Stale));
assert_eq!(slab.take(next), Ok(11));
assert_eq!(slab.take(other), Ok(9));
}
#[test]
fn a_slot_retired_under_a_panicking_call_is_reused_once_and_only_once() {
let mut slab: Slab<u32> = Slab::new();
let handle = slab.insert(7).expect("insert");
// A panicking call drops the value where it stands and poisons after.
slab.take(handle).expect("take");
slab.remove(handle).expect("remove");
slab.poison(handle);
let next = slab.insert(11).expect("insert");
assert_eq!(parts(next).0, parts(handle).0);
// A second late arrival on the retired handle does not free the slot
// the new session is using.
slab.poison(handle);
assert_eq!(slab.put(handle, 13), Err(13));
assert_eq!(slab.insert(17).map(|h| parts(h).0), Ok(parts(next).0 + 1));
assert_eq!(slab.take(next), Ok(11));
}
}
crates/ffi/tests/abi.c
+28 -3
View File
@@ -22,7 +22,7 @@ static int failures = 0;
} \
} while (0)
static const char *TURN = "{\"Turn\":{\"prompt\":\"fix the build\"}}";
static const char *TURN = "{\"Turn\":{\"id\":1,\"prompt\":\"fix the build\"}}";
/* The buffer Rust lends is not terminated; copy it before reading it as text. */
static char *text(dev_buf buf) {
@@ -56,6 +56,18 @@ int main(void) {
}
dev_free(actions);
/* The same turn again is a redelivery: no action, and the prompt is not
* asked a second time. */
dev_buf twice = {0};
CHECK(dev_step(session, (const uint8_t *)TURN, strlen(TURN), &twice) == DEV_OK);
shown = text(twice);
CHECK(shown != NULL);
if (shown != NULL) {
CHECK(strcmp(shown, "[]") == 0);
free(shown);
}
dev_free(twice);
/* A snapshot restores into a second session that continues the sequence. */
dev_buf state = {0};
CHECK(dev_snapshot(session, &state) == DEV_OK);
@@ -73,16 +85,29 @@ int main(void) {
shown = text(ended);
CHECK(shown != NULL);
if (shown != NULL) {
/* Emit, Save, then the Done that names the turn that asked. */
CHECK(strstr(shown, "\"id\":2") != NULL);
CHECK(strstr(shown, "Done") != NULL);
CHECK(strstr(shown, "\"id\":4") != NULL);
CHECK(strstr(shown, "{\"Done\":{\"turn\":1,\"outcome\":\"Complete\"}}") != NULL);
free(shown);
}
dev_free(ended);
dev_drop(restored);
/* Garbage is refused, and the session survives it. */
/* Garbage is refused, the session survives it, and nothing is lent out. */
dev_buf refused = {0};
CHECK(dev_step(session, (const uint8_t *)"garbage", 7, &refused) == DEV_MALFORMED);
CHECK(refused.ptr == NULL);
/* A snapshot the store handed back changed is refused, not restored. */
dev_buf tampered = {0};
CHECK(dev_snapshot(session, &tampered) == DEV_OK);
CHECK(tampered.len > 13);
tampered.ptr[tampered.len - 1] ^= 0x20;
uint64_t forged = 0;
CHECK(dev_restore(tampered.ptr, tampered.len, &forged) == DEV_MALFORMED);
CHECK(forged == 0);
dev_free(tampered);
/* A dropped handle names nothing. */
dev_drop(session);
crates/protocol/src/lib.rs
+266 -47
View File
@@ -1,10 +1,13 @@
//! Events in, actions out — the only language the loop and its host share.
//!
//! HIP-1330 splits a coding agent at the seam between reasoning and effect.
//! This crate is the alphabet of that seam: the host observes the world and
//! sends an [`Event`]; the core reasons and answers with [`Action`]s. Every
//! action carries an `id`, so the host's ledger can record dispatch and
//! completion under it and replay a result instead of repeating an effect.
//! HIP-1330 splits a coding agent between reasoning and effect. This crate is
//! the alphabet of that split: the host observes the world and sends an
//! [`Event`]; the core reasons and answers with [`Action`]s. Every action
//! carries an `id`, so the host's ledger can record dispatch and completion
//! under it and replay a result instead of repeating an effect. A [`Turn`]
//! carries the host's own id for the same reason, in the other direction: the
//! core refuses a turn it has already accepted, and names that id in the
//! [`Done`] that ends it.
//!
//! # Encoding
//!
@@ -13,15 +16,27 @@
//! Today it is `serde_json`.
//!
//! It becomes ZAP (HIP-0114) when the ZAP schema compiler is a build input
//! this repo has. The Rust side is published — `zap-proto` on crates.io
//! re-exports `zap-schema`, the same runtime that sits in `zap/rust/zap` —
//! but it is a runtime for generated code, and generating that code takes the
//! `zap` schema compiler, which is not installed here and is not a build
//! input of this repo or its CI. Pointing the generator at `capnp` instead
//! would be a shim across a fork boundary, and hand-writing the wire format
//! would be worse than either. So the two functions below stay serde until
//! the compiler is a dependency, and nothing here imitates ZAP framing in the
//! meantime.
//! this repo has. The Rust side exists — `zap` 1.0.0 in `zap/rust/zap` is the
//! runtime and `zapc` 1.0.0 is the codegen plugin, both published from the
//! `zap-proto` GitHub org — but the runtime is a runtime for generated code,
//! and generating that code takes the `zap` schema compiler, a C++ binary
//! `zapc` shells out to that is not installed here and is not a build input of
//! this repo or its CI. Pointing the generator at `capnp` instead would be a
//! shim across a fork boundary, and hand-writing the format would be worse
//! than either. So the two functions below stay serde until the compiler is a
//! dependency, and nothing here imitates ZAP framing in the meantime.
//!
//! # Paths
//!
//! A path in this protocol is workspace-relative, and [`inside`] is the one
//! statement of what that means. The host joins a path onto the tree it
//! mounted for the session, so `/etc/passwd` names no workspace and
//! `../../etc/passwd` climbs out of the one it was given. The core refuses
//! such a call rather than emitting it, so no host re-derives confinement.
//!
//! Which effects may happen at all — which argv, which URL — is policy, and
//! policy belongs to `/v1/dev` outside the sandbox (HIP-1330 §Security). This
//! crate takes no position on it.
use serde::Deserialize;
use serde::Serialize;
@@ -30,10 +45,16 @@ use serde::de::DeserializeOwned;
/// Version of the byte format and of the C ABI that carries it.
pub const ABI: u32 = 1;
/// A decode failure, with the reason the bytes were refused.
/// A payload this protocol cannot carry, with the reason.
#[derive(Debug)]
pub struct Malformed(String);
impl Malformed {
fn new(reason: impl core::fmt::Display) -> Self {
Self(reason.to_string())
}
}
impl core::fmt::Display for Malformed {
fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
write!(f, "malformed payload: {}", self.0)
@@ -42,19 +63,99 @@ impl core::fmt::Display for Malformed {
impl std::error::Error for Malformed {}
/// Encode a payload for the seam.
pub fn encode<T: Serialize>(value: &T) -> Vec<u8> {
// serde_json only fails on a serializer that refuses a value (a map with
// non-string keys, a float that is not finite). No protocol type has one.
match serde_json::to_vec(value) {
Ok(bytes) => bytes,
Err(e) => panic!("protocol type is not encodable: {e}"),
}
/// Encode a payload.
pub fn encode<T: Serialize>(value: &T) -> Result<Vec<u8>, Malformed> {
serde_json::to_vec(value).map_err(Malformed::new)
}
/// Decode a payload from the seam.
/// Decode a payload.
pub fn decode<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, Malformed> {
serde_json::from_slice(bytes).map_err(|e| Malformed(e.to_string()))
serde_json::from_slice(bytes).map_err(Malformed::new)
}
/// Pack a payload that will be stored and handed back later: the ABI it was
/// written under, a checksum of the body, then the body.
///
/// The checksum is not a signature, and nothing here is a place to put one:
/// this crate holds no key and knows no identity. It catches a truncated,
/// corrupted, hand-edited or foreign snapshot, which is what a store gets
/// wrong. Whose snapshot this is, and whether the sequence it restores agrees
/// with the ledger, is the host's: IAM is the only identity (HIP-0026) and the
/// journal — not the snapshot — is the record of which ids were dispatched
/// (HIP-1330 §A session is three facts).
pub fn pack<T: Serialize>(value: &T) -> Result<Vec<u8>, Malformed> {
let body = encode(value)?;
let mut bytes = Vec::with_capacity(body.len() + STAMP);
bytes.extend_from_slice(&ABI.to_le_bytes());
bytes.extend_from_slice(&sum(&body).to_le_bytes());
bytes.extend_from_slice(&body);
Ok(bytes)
}
/// Open a packed payload, refusing one written under another ABI and one whose
/// bytes no longer match their checksum.
pub fn unpack<T: DeserializeOwned>(bytes: &[u8]) -> Result<T, Malformed> {
let (stamp, body) = bytes
.split_at_checked(STAMP)
.ok_or_else(|| Malformed::new("shorter than its own stamp"))?;
let (abi, checksum) = stamp.split_at(size_of::<u32>());
let abi = u32::from_le_bytes(abi.try_into().map_err(Malformed::new)?);
if abi != ABI {
return Err(Malformed::new(format!(
"written under ABI {abi}, not {ABI}"
)));
}
let checksum = u64::from_le_bytes(checksum.try_into().map_err(Malformed::new)?);
if checksum != sum(body) {
return Err(Malformed::new("checksum does not match the body"));
}
decode(body)
}
/// Bytes of ABI and checksum in front of a packed body.
const STAMP: usize = size_of::<u32>() + size_of::<u64>();
/// FNV-1a, 64-bit: enough to catch a store that lost or changed bytes, and
/// deliberately not enough to pass for authentication.
fn sum(bytes: &[u8]) -> u64 {
let mut hash: u64 = 0xcbf2_9ce4_8422_2325;
for byte in bytes {
hash ^= u64::from(*byte);
hash = hash.wrapping_mul(0x0000_0100_0000_01b3);
}
hash
}
/// Whether a path a tool call names stays inside the workspace.
///
/// Absolute is out, whatever spelling: a leading separator, a drive letter, a
/// UNC prefix. Climbing out is out: the parts are counted, and a `..` that
/// would pass the root refuses the path. An interior NUL is out, because the
/// path crosses a C ABI and would be cut there. What is left is a relative
/// path the host can join onto the workspace it mounted.
pub fn inside(path: &str) -> bool {
if path.is_empty() || path.contains('\0') {
return false;
}
if path.starts_with('/') || path.starts_with('\\') {
return false;
}
let bytes = path.as_bytes();
if bytes.len() >= 2 && bytes[0].is_ascii_alphabetic() && bytes[1] == b':' {
return false;
}
let mut depth: usize = 0;
for part in path.split(['/', '\\']) {
match part {
"" | "." => {}
".." => match depth.checked_sub(1) {
Some(up) => depth = up,
None => return false,
},
_ => depth += 1,
}
}
true
}
/// What the host tells the core.
@@ -100,17 +201,32 @@ pub enum Op {
/// Persist the core's snapshot at this point in the sequence.
Save,
/// The turn is over.
Done(Outcome),
Done(Done),
}
/// A prompt, under the id the host's journal gave the request.
///
/// Ids rise and are never zero. The core keeps the highest it has accepted and
/// refuses anything at or below it, so a redelivered turn produces no action
/// and runs the prompt once.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Turn {
pub id: u64,
pub prompt: String,
}
/// The end of the turn `turn` asked for.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Done {
pub turn: u64,
pub outcome: Outcome,
}
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Ask {
pub model: Option<String>,
/// The session's standing instructions, ahead of the conversation.
pub prelude: Option<String>,
pub messages: Vec<Message>,
}
@@ -125,8 +241,14 @@ pub struct Write {
pub text: String,
}
/// A diff, and the one file it edits.
///
/// The paths inside a diff are visible only to the parser that applies it, so
/// the action names the file instead: the core confines `path`, and the host
/// applies `diff` to that file and nothing else.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Patch {
pub path: String,
pub diff: String,
}
@@ -148,7 +270,7 @@ pub struct Browse {
}
/// Why a turn ended.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[derive(Clone, Copy, Debug, PartialEq, Serialize, Deserialize)]
pub enum Outcome {
Complete,
Cancelled,
@@ -180,6 +302,31 @@ pub enum Call {
Browse(Browse),
}
impl Call {
/// The path this call would reach outside the workspace, if it names one.
///
/// A `Git` argv and a `Browse` url are not paths and are not judged here:
/// what may be run and what may be fetched is policy, and policy is
/// `/v1/dev`'s.
pub fn escape(&self) -> Option<&str> {
let path = match self {
Self::Read(read) => Some(read.path.as_str()),
Self::Write(write) => Some(write.path.as_str()),
Self::Patch(patch) => Some(patch.path.as_str()),
Self::Exec(exec) => exec.cwd.as_deref(),
Self::Git(_) | Self::Browse(_) => None,
};
path.filter(|path| !inside(path))
}
}
/// A call the core sent out, under the id its result will name.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Dispatch {
pub id: u64,
pub call: Call,
}
/// The result of an effect, naming the action it answers.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Output {
@@ -188,19 +335,23 @@ pub struct Output {
pub failed: bool,
}
/// Who said something.
#[derive(Clone, Copy, Debug, PartialEq, Serialize, Deserialize)]
pub enum Role {
User,
Agent,
Tool,
}
/// One entry of the conversation the core carries.
///
/// An agent entry carries the calls it asked for, each under the id it was
/// dispatched with, and a tool entry names the id it answers. So a provider
/// that requires the assistant's tool call ahead of the tool result can be
/// driven straight from this: the pair is in the transcript, not inferred.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct Message {
pub role: Role,
pub text: String,
pub enum Message {
/// A person asked for work.
User(String),
/// The agent spoke, asked for tools, or both.
Agent {
text: Option<String>,
calls: Vec<Dispatch>,
},
/// A tool answered the call dispatched under `id`.
Tool { id: u64, text: String, failed: bool },
}
/// What a session is born with.
@@ -208,7 +359,7 @@ pub struct Message {
pub struct Config {
/// Model the core names in every [`Op::Model`]; the host routes it.
pub model: Option<String>,
/// First message of the conversation, if the host wants one.
/// Standing instructions every ask carries, if the host wants some.
pub prelude: Option<String>,
}
@@ -220,12 +371,10 @@ mod tests {
#[test]
fn event_round_trips() {
let event = Event::Turn(Turn {
id: 1,
prompt: "fix the build".to_string(),
});
let back: Event = match decode(&encode(&event)) {
Ok(e) => e,
Err(e) => panic!("{e}"),
};
let back: Event = decode(&encode(&event).expect("encode")).expect("decode");
assert_eq!(event, back);
}
@@ -238,10 +387,7 @@ mod tests {
cwd: None,
}),
};
let back: Action = match decode(&encode(&action)) {
Ok(a) => a,
Err(e) => panic!("{e}"),
};
let back: Action = decode(&encode(&action).expect("encode")).expect("decode");
assert_eq!(back.id, 7821);
assert_eq!(action, back);
}
@@ -251,4 +397,77 @@ mod tests {
let err = decode::<Event>(b"not a payload").expect_err("garbage decoded");
assert!(err.to_string().starts_with("malformed payload:"));
}
#[test]
fn a_packed_payload_survives_the_trip_and_an_edited_one_does_not() {
let turn = Turn {
id: 9,
prompt: "ship it".to_string(),
};
let packed = pack(&turn).expect("pack");
assert_eq!(unpack::<Turn>(&packed).expect("unpack"), turn);
for cut in 0..packed.len() {
assert!(unpack::<Turn>(&packed[..cut]).is_err(), "{cut} bytes");
}
let mut edited = packed.clone();
let last = edited.len() - 1;
edited[last] ^= 0x20;
let err = unpack::<Turn>(&edited).expect_err("an edited body was opened");
assert!(err.to_string().contains("checksum"), "{err}");
let mut older = packed;
older[0] = older[0].wrapping_sub(1);
let err = unpack::<Turn>(&older).expect_err("a foreign ABI was opened");
assert!(err.to_string().contains("ABI"), "{err}");
}
#[test]
fn a_path_is_inside_only_when_it_is_relative_and_stays_down() {
for path in ["src/lib.rs", "a/../b", "./x", "deep/./er/../er/x"] {
assert!(inside(path), "{path}");
}
for path in [
"",
"/etc/passwd",
"..",
"../x",
"a/../../x",
"C:\\Windows",
"\\\\host\\share",
"\\etc",
"a\0b",
] {
assert!(!inside(path), "{path}");
}
}
#[test]
fn a_call_names_the_path_it_would_escape_with() {
let read = Call::Read(Read {
path: "../../etc/passwd".to_string(),
});
assert_eq!(read.escape(), Some("../../etc/passwd"));
let write = Call::Write(Write {
path: "src/lib.rs".to_string(),
text: String::new(),
});
assert_eq!(write.escape(), None);
let patch = Call::Patch(Patch {
path: "/etc/hosts".to_string(),
diff: String::new(),
});
assert_eq!(patch.escape(), Some("/etc/hosts"));
let exec = Call::Exec(Exec {
argv: vec!["ls".to_string()],
cwd: Some("/".to_string()),
});
assert_eq!(exec.escape(), Some("/"));
let anywhere = Call::Exec(Exec {
argv: vec!["ls".to_string()],
cwd: None,
});
assert_eq!(anywhere.escape(), None);
}
}