381 lines
13 KiB
Rust
381 lines
13 KiB
Rust
//! The Honcho worker, stepped by hand against the fake Honcho (M5c spec, section 5): a turn sent
|
|
//! once as one batch, the speaker's parts then the agent's, neutralised, in parts of 25,000
|
|
//! characters, to the `-2` session for a new speaker; an outage waited out with a doubling wait
|
|
//! and logged twice in all; a refused send marked failed; a refused key stopping its workspace
|
|
//! only; a turn the log no longer holds marked failed; other actions left waiting. The waits and
|
|
//! a restart are in `honcho_worker_more.rs`. Do not edit.
|
|
|
|
#[path = "support/honcho.rs"]
|
|
mod honcho;
|
|
#[path = "support/logs.rs"]
|
|
mod logs;
|
|
#[path = "support/tmp.rs"]
|
|
mod tmp;
|
|
#[path = "support/turns.rs"]
|
|
mod turns;
|
|
#[path = "support/worker.rs"]
|
|
mod worker;
|
|
|
|
use std::time::Duration;
|
|
|
|
use bxcore::sanitize::TAGS;
|
|
use bxcore::{DataClass, SessionStateReply};
|
|
use honcho::{FakeHoncho, Mode, Request};
|
|
use logs::assistant;
|
|
use memoryd::honcho::ledger::{Item, Record};
|
|
use memoryd::honcho::worker::{MAX_BATCH, MAX_PART, Step, parts};
|
|
use serde_json::{Value, json};
|
|
use turns::{sid, spoken};
|
|
use worker::{FAST, Home};
|
|
|
|
/// The ids fixture 08 answers with.
|
|
const IDS: [&str; 2] = ["7HqDWzXcKAdw9tnElCNFg", "mjsyhwuGsXTh1_zG_tWRx"];
|
|
|
|
fn message_posts(fake: &FakeHoncho) -> Vec<Request> {
|
|
fake.requests()
|
|
.into_iter()
|
|
.filter(|r| r.path.ends_with("/messages"))
|
|
.collect()
|
|
}
|
|
|
|
fn turn(text: &str, speaker: &str, reply: &str) -> Vec<bxcore::LogRecord> {
|
|
vec![spoken(text, Some(speaker)), assistant(2, Some(reply), &[])]
|
|
}
|
|
|
|
fn meta(session: &str, turn: u64) -> Value {
|
|
json!({ "agent": "boxmaker0", "boxmaker_session": session, "turn": turn })
|
|
}
|
|
|
|
fn sent(session: &str, turn: u64) -> impl Fn(&Record) -> bool {
|
|
let item = Item::Turn {
|
|
session: sid(session),
|
|
turn,
|
|
};
|
|
move |r| matches!(r, Record::Sent { item: i, .. } if *i == item)
|
|
}
|
|
|
|
#[test]
|
|
fn the_limits() {
|
|
assert_eq!(MAX_PART, 25_000);
|
|
assert_eq!(MAX_BATCH, 100);
|
|
assert!(parts("").is_empty());
|
|
assert_eq!(parts("abc"), vec!["abc".to_string()]);
|
|
let long = "é".repeat(MAX_PART);
|
|
assert_eq!(parts(&long), vec![long.clone()], "characters, not bytes");
|
|
let over = format!("{long}xy");
|
|
assert_eq!(parts(&over), vec![long, "xy".to_string()]);
|
|
}
|
|
|
|
#[test]
|
|
fn a_turn_is_sent_once_as_one_batch() {
|
|
let fake = FakeHoncho::start();
|
|
let home = Home::new("worker-once");
|
|
home.log("mm-a", &[turn("hello", "kyle", "hi")]);
|
|
home.report("mm-a", 1);
|
|
let mut w = home.worker(&fake, FAST);
|
|
assert_eq!(w.step(), Step::Done);
|
|
let posts = message_posts(&fake);
|
|
assert_eq!(posts.len(), 1);
|
|
assert_eq!(
|
|
posts[0].path,
|
|
"/v3/workspaces/boxmaker0-trusted/sessions/mm-a/messages"
|
|
);
|
|
assert_eq!(
|
|
posts[0].body,
|
|
Some(json!({ "messages": [
|
|
{ "content": "hello", "peer_id": "kyle", "metadata": meta("mm-a", 1) },
|
|
{ "content": "hi", "peer_id": "boxmaker0", "metadata": meta("mm-a", 1) },
|
|
] }))
|
|
);
|
|
// The session was made first, with both peers.
|
|
assert_eq!(fake.requests().len(), 5);
|
|
assert!(fake.requests()[3].path.ends_with("/sessions"));
|
|
let records = home.records();
|
|
assert_eq!(records.len(), 2);
|
|
assert!(matches!(&records[1], Record::Sent { ids, .. } if *ids == IDS));
|
|
fake.clear();
|
|
assert_eq!(w.step(), Step::Idle);
|
|
assert!(fake.requests().is_empty());
|
|
assert!(home.lines().is_empty(), "{:?}", home.lines());
|
|
}
|
|
|
|
#[test]
|
|
fn a_text_over_the_limit_is_sent_in_parts() {
|
|
let fake = FakeHoncho::start();
|
|
let home = Home::new("worker-parts");
|
|
let text = "a".repeat(MAX_PART + 1);
|
|
let reply = "é".repeat(MAX_PART);
|
|
home.log("mm-a", &[turn(&text, "kyle", &reply)]);
|
|
home.report("mm-a", 1);
|
|
let mut w = home.worker(&fake, FAST);
|
|
assert_eq!(w.step(), Step::Done);
|
|
let body = message_posts(&fake)[0].body.clone().unwrap();
|
|
let got: Vec<(usize, String)> = body["messages"]
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.map(|m| {
|
|
(
|
|
m["content"].as_str().unwrap().chars().count(),
|
|
m["peer_id"].as_str().unwrap().to_string(),
|
|
)
|
|
})
|
|
.collect();
|
|
assert_eq!(
|
|
got,
|
|
vec![
|
|
(MAX_PART, "kyle".to_string()),
|
|
(1, "kyle".to_string()),
|
|
(MAX_PART, "boxmaker0".to_string()),
|
|
]
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn both_texts_are_neutralised() {
|
|
let fake = FakeHoncho::start();
|
|
let home = Home::new("worker-neutral");
|
|
let text = format!("a{}b", TAGS[0].text);
|
|
let reply = format!("c{}d", TAGS[1].text);
|
|
home.log("mm-a", &[turn(&text, "kyle", &reply)]);
|
|
home.report("mm-a", 1);
|
|
let mut w = home.worker(&fake, FAST);
|
|
assert_eq!(w.step(), Step::Done);
|
|
let body = message_posts(&fake)[0].body.clone().unwrap();
|
|
let contents: Vec<String> = body["messages"]
|
|
.as_array()
|
|
.unwrap()
|
|
.iter()
|
|
.map(|m| m["content"].as_str().unwrap().to_string())
|
|
.collect();
|
|
assert_eq!(contents.len(), 2);
|
|
for (content, tag) in contents.iter().zip([TAGS[0].text, TAGS[1].text]) {
|
|
assert!(!content.contains(tag));
|
|
assert!(content.contains("<"), "not neutralised");
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn a_new_speaker_goes_to_the_next_session() {
|
|
let fake = FakeHoncho::start();
|
|
let home = Home::new("worker-second");
|
|
home.log(
|
|
"mm-a",
|
|
&[turn("one", "kyle", "a"), turn("two", "ana_b", "b")],
|
|
);
|
|
home.report("mm-a", 2);
|
|
let mut w = home.worker(&fake, FAST);
|
|
assert_eq!(w.step(), Step::Done);
|
|
assert_eq!(w.step(), Step::Done);
|
|
let posts = message_posts(&fake);
|
|
assert_eq!(posts.len(), 2);
|
|
assert!(posts[0].path.ends_with("/sessions/mm-a/messages"));
|
|
assert!(posts[1].path.ends_with("/sessions/mm-a-2/messages"));
|
|
let second = fake
|
|
.requests()
|
|
.into_iter()
|
|
.find(|r| r.path.ends_with("/sessions") && r.body.as_ref().unwrap()["id"] == "mm-a-2")
|
|
.unwrap();
|
|
let peers = second.body.unwrap()["peers"].clone();
|
|
assert_eq!(peers["kyle"]["observe_me"], true);
|
|
assert_eq!(peers["ana_b"]["observe_me"], true);
|
|
assert_eq!(peers["boxmaker0"]["observe_me"], false);
|
|
}
|
|
|
|
#[test]
|
|
fn an_outage_is_waited_out_and_logged_twice_in_all() {
|
|
let fake = FakeHoncho::start();
|
|
let home = Home::new("worker-outage");
|
|
home.log("mm-a", &[turn("private-alpha", "kyle", "a")]);
|
|
home.log("mm-b", &[turn("private-beta", "kyle", "b")]);
|
|
home.report("mm-a", 1);
|
|
home.report("mm-b", 1);
|
|
let mut w = home.worker(&fake, FAST);
|
|
fake.set(Mode::Status(503));
|
|
let waits: Vec<Step> = (0..6).map(|_| w.step()).collect();
|
|
let ms = |n| Step::Backoff(Duration::from_millis(n));
|
|
assert_eq!(waits, vec![ms(1), ms(2), ms(4), ms(8), ms(8), ms(8)]);
|
|
assert_eq!(home.lines().len(), 1, "{:?}", home.lines());
|
|
fake.set(Mode::Fixtures);
|
|
assert_eq!(w.step(), Step::Done);
|
|
assert_eq!(w.step(), Step::Done);
|
|
let lines = home.lines();
|
|
assert_eq!(lines.len(), 2, "{lines:?}");
|
|
assert_eq!(
|
|
lines[0],
|
|
"memoryd: Honcho is unreachable (server, HTTP 503); sends wait and are retried; see docs/runbook.md#honcho-unreachable"
|
|
);
|
|
assert_eq!(
|
|
lines[1],
|
|
"memoryd: Honcho answers again; waiting sends go on"
|
|
);
|
|
// In order: mm-a, then mm-b.
|
|
let records = home.records();
|
|
let a = records.iter().position(sent("mm-a", 1)).unwrap();
|
|
let b = records.iter().position(sent("mm-b", 1)).unwrap();
|
|
assert!(a < b);
|
|
// A new outage starts from the first wait again.
|
|
home.log("mm-c", &[turn("c", "kyle", "c")]);
|
|
home.report("mm-c", 1);
|
|
fake.set(Mode::Hang);
|
|
assert_eq!(w.step(), ms(1));
|
|
assert!(home.lines()[2].contains("(timeout)"), "{:?}", home.lines());
|
|
}
|
|
|
|
#[test]
|
|
fn a_refused_send_is_marked_failed_and_the_next_goes_on() {
|
|
let fake = FakeHoncho::start();
|
|
let home = Home::new("worker-refused");
|
|
home.log("mm-a", &[turn("private-alpha", "kyle", "a")]);
|
|
home.log("mm-b", &[turn("private-beta", "kyle", "b")]);
|
|
home.report("mm-a", 1);
|
|
home.report("mm-b", 1);
|
|
let mut w = home.worker(&fake, FAST);
|
|
fake.set(Mode::Status(422));
|
|
assert_eq!(w.step(), Step::Done);
|
|
let failed = home.records().into_iter().find_map(|r| match r {
|
|
Record::Failed {
|
|
item, status, kind, ..
|
|
} => Some((item, status, kind)),
|
|
_ => None,
|
|
});
|
|
assert_eq!(
|
|
failed,
|
|
Some((
|
|
Item::Turn {
|
|
session: sid("mm-a"),
|
|
turn: 1
|
|
},
|
|
Some(422),
|
|
"status".to_string()
|
|
))
|
|
);
|
|
let lines = home.lines();
|
|
assert_eq!(
|
|
lines,
|
|
vec![
|
|
"memoryd: turn 1 of session mm-a was not sent: Honcho refused the request (status, HTTP 422); it is marked failed; see docs/runbook.md#honcho-send-refused"
|
|
.to_string()
|
|
]
|
|
);
|
|
fake.set(Mode::Fixtures);
|
|
assert_eq!(w.step(), Step::Done);
|
|
assert!(home.records().iter().any(sent("mm-b", 1)));
|
|
assert_eq!(w.step(), Step::Idle, "the failed turn is not tried again");
|
|
}
|
|
|
|
#[test]
|
|
fn a_refused_key_stops_its_workspace_only() {
|
|
let fake = FakeHoncho::start();
|
|
let home = Home::new("worker-key");
|
|
home.log("mm-a", &[turn("a", "kyle", "a")]);
|
|
home.log("mm-b", &[turn("b", "kyle", "b")]);
|
|
home.report("mm-a", 1);
|
|
*home.answer.lock().unwrap() = Ok(SessionStateReply {
|
|
class: DataClass::Private,
|
|
untrusted: true,
|
|
});
|
|
home.report("mm-b", 1);
|
|
let mut w = home.worker(&fake, FAST);
|
|
fake.set(Mode::Status(401));
|
|
assert_eq!(w.step(), Step::Done);
|
|
assert_eq!(
|
|
home.lines(),
|
|
vec![
|
|
"memoryd: Honcho refused the key of workspace boxmaker0-trusted (HTTP 401); nothing more is sent to it until memoryd restarts; see docs/runbook.md#honcho-key-refused"
|
|
.to_string()
|
|
]
|
|
);
|
|
fake.set(Mode::Fixtures);
|
|
fake.clear();
|
|
assert_eq!(w.step(), Step::Done);
|
|
assert!(home.records().iter().any(sent("mm-b", 1)));
|
|
assert!(
|
|
fake.requests()
|
|
.iter()
|
|
.all(|r| !r.path.contains("boxmaker0-trusted"))
|
|
);
|
|
assert_eq!(w.step(), Step::Idle);
|
|
// The trusted turn waits: neither sent nor failed.
|
|
let sync = home.sync.lock().unwrap();
|
|
assert_eq!(sync.ledger().outbox().len(), 1);
|
|
assert_eq!(sync.ledger().counts().failed, 0);
|
|
}
|
|
|
|
#[test]
|
|
fn a_turn_the_log_no_longer_holds_is_marked_failed() {
|
|
let fake = FakeHoncho::start();
|
|
let home = Home::new("worker-gone");
|
|
home.log("mm-a", &[turn("a", "kyle", "a")]);
|
|
home.report("mm-a", 1);
|
|
let path = home.dir.path().join("sessions/mm-a/0.jsonl");
|
|
std::fs::write(&path, "not json\n").unwrap();
|
|
let mut w = home.worker(&fake, FAST);
|
|
assert_eq!(w.step(), Step::Done);
|
|
assert!(matches!(
|
|
home.records().last(),
|
|
Some(Record::Failed { status: None, kind, .. }) if kind == "log"
|
|
));
|
|
assert_eq!(
|
|
home.lines(),
|
|
vec![
|
|
"memoryd: turn 1 of session mm-a cannot be sent (log); it is marked failed; see docs/runbook.md#session-log-damaged"
|
|
.to_string()
|
|
]
|
|
);
|
|
assert!(message_posts(&fake).is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn a_turn_without_its_reply_any_more_is_marked_failed() {
|
|
let fake = FakeHoncho::start();
|
|
let home = Home::new("worker-noreply");
|
|
home.log("mm-a", &[turn("a", "kyle", "a")]);
|
|
home.report("mm-a", 1);
|
|
// The log now ends in a tool round: the reply is gone.
|
|
home.log("mm-a", &[turns::running(Some("kyle"))]);
|
|
let mut w = home.worker(&fake, FAST);
|
|
assert_eq!(w.step(), Step::Done);
|
|
assert!(matches!(
|
|
home.records().last(),
|
|
Some(Record::Failed { status: None, kind, .. }) if kind == "log"
|
|
));
|
|
assert!(message_posts(&fake).is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn a_refused_key_ends_an_outage() {
|
|
let fake = FakeHoncho::start();
|
|
let home = Home::new("worker-key-outage");
|
|
home.log("mm-a", &[turn("a", "kyle", "a")]);
|
|
home.report("mm-a", 1);
|
|
let mut w = home.worker(&fake, FAST);
|
|
fake.set(Mode::Status(503));
|
|
assert!(matches!(w.step(), Step::Backoff(_)));
|
|
fake.set(Mode::Status(403));
|
|
assert_eq!(w.step(), Step::Done);
|
|
let lines = home.lines();
|
|
assert_eq!(lines.len(), 3, "{lines:?}");
|
|
assert_eq!(
|
|
lines[1],
|
|
"memoryd: Honcho answers again; waiting sends go on"
|
|
);
|
|
assert!(lines[2].ends_with("see docs/runbook.md#honcho-key-refused"));
|
|
}
|
|
|
|
#[test]
|
|
fn a_turn_too_long_for_one_batch_is_marked_failed() {
|
|
let fake = FakeHoncho::start();
|
|
let home = Home::new("worker-long");
|
|
let text = "a".repeat(MAX_PART * MAX_BATCH);
|
|
home.log("mm-a", &[turn(&text, "kyle", "a")]);
|
|
home.report("mm-a", 1);
|
|
let mut w = home.worker(&fake, FAST);
|
|
assert_eq!(w.step(), Step::Done);
|
|
assert!(matches!(
|
|
home.records().last(),
|
|
Some(Record::Failed { status: None, kind, .. }) if kind == "too-long"
|
|
));
|
|
assert!(fake.requests().is_empty());
|
|
}
|