From 81ec8baec70e34a92f4794d596b3aa89aa9cf8c3 Mon Sep 17 00:00:00 2001 From: ymote <151983+ymote@users.noreply.github.com> Date: Tue, 29 Sep 2026 22:09:38 -0700 Subject: [PATCH] Ask panel follow-ups: lane order, one Stop notice, peer slugs, a flaky test Four follow-ups from the live run of "Ask " (#184). 1. Order across lanes. Each lane counts its own cursor.seq (the peer's session is long, the person's context new), so sorting both by seq drew the system agent's request below the person's later question. The kernel also sends a turn's user message only when the turn ENDS, and never for a stopped turn, so the request line sat after its answer and a stopped request never showed. Now: within a lane, ledger order; across lanes, one clock (the order events reached the panel: stream events carry no time; history comes merged by persisted_at). A turn's request line stands where the turn started: the broker records the words it sent (the person's message, the system agent's peer/input) and hands them over with turn/started (`request`), and the conversation's history adds a stopped turn's request the kernel never recorded, at the time it started. 2. Stop said "turn interrupted by client" twice: once from the turn's end, once from the send's error. The send's error is shown only when the turn never started. 3. The system agent tried the app id (os.news) with peer_send_input. The note and agents.list now give the peer slug (os-news-22a12f90; OctosAppService::peer_slug) and say to use it, or to find it with peer_list, never the app id. 4. the_person_and_the_system_agent_talk_in_parallel_lanes_that_share_history failed under load with turn_in_progress: the test drives the system session itself and started its second turn while the first was still finishing (after peer_send_input the system agent still writes its own answer). Not a broker race: the test now waits for its session, as the system chat does. Reproduced by slowing that answer in the scripted model; 20 runs at --test-threads=4 pass. Tests: two lanes replayed with interleaved arrival (out-of-order envelopes, user messages at the end, a stopped turn); the panel's Stop shows one notice; the broker hands a turn's words over at its start and keeps a stopped turn's request in history; the note and agents.list carry the slug. Co-Authored-By: Claude Opus 5.5 (1M context) --- crates/ai-host/src/contained.rs | 6 ++ crates/app-peers/src/broker.rs | 122 +++++++++++++++++++++++++- crates/app-peers/src/contract.rs | 7 ++ crates/app-peers/tests/broker.rs | 49 +++++++++++ crates/app-peers/tests/real_kernel.rs | 28 ++++-- crates/shell/src/agents.rs | 69 +++++++++++---- crates/shell/src/app_chat/mod.rs | 11 ++- crates/shell/src/app_chat/model.rs | 103 +++++++++++++++++++--- crates/shell/src/app_chat/tests.rs | 102 ++++++++++++++++++++- 9 files changed, 453 insertions(+), 44 deletions(-) diff --git a/crates/ai-host/src/contained.rs b/crates/ai-host/src/contained.rs index e14d7a91..edc7c931 100644 --- a/crates/ai-host/src/contained.rs +++ b/crates/ai-host/src/contained.rs @@ -147,6 +147,12 @@ pub fn prepare(app_id: &str) -> Result<(), String> { service.prepare() } +/// The kernel's slug of `app_id`'s peer, once one is live and bound (what +/// `peer_list` shows and `peer_send_input` takes). +pub fn peer_slug(app_id: &str) -> Option { + live(|l| l.get(app_id).cloned())?.peer_slug() +} + /// Whether `app_id` has a live peer (prepared, or opened by the app). pub fn is_live(app_id: &str) -> bool { live(|l| l.contains_key(app_id)) diff --git a/crates/app-peers/src/broker.rs b/crates/app-peers/src/broker.rs index a06ba181..83b266a6 100644 --- a/crates/app-peers/src/broker.rs +++ b/crates/app-peers/src/broker.rs @@ -441,6 +441,66 @@ impl State { } } +/// The words that started one turn of the app's conversation, as this +/// broker sent them: which session (lane), who spoke and when it started. +#[derive(Clone, Debug)] +struct Request { + turn: String, + session: String, + lane: &'static str, + text: String, + speaker: Speaker, + /// RFC 3339 (UTC), the host's clock at the start. + at: String, +} + +impl Request { + fn to_json(&self) -> Value { + json!({"text": self.text, "speaker": self.speaker.to_json()}) + } +} + +impl State { + fn note_request(&mut self, request: Request) { + if request.text.trim().is_empty() || self.requests.iter().any(|r| r.turn == request.turn) { + return; + } + self.requests.push_back(request); + while self.requests.len() > REMEMBERED { + self.requests.pop_front(); + } + } + + fn request_of(&self, turn: &str) -> Option<&Request> { + self.requests.iter().find(|r| r.turn == turn) + } +} + +/// Now as RFC 3339 in UTC with microseconds, like the kernel's +/// `persisted_at` (so [`time_key`] orders both). +fn rfc3339_now() -> String { + let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default(); + let secs = now.as_secs() as i64; + let (days, rem) = (secs.div_euclid(86_400), secs.rem_euclid(86_400)); + // Days since 1970-01-01 to a civil date (H. Hinnant's algorithm). + let z = days + 719_468; + let era = z.div_euclid(146_097); + let doe = z - era * 146_097; + let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); + let mp = (5 * doy + 2) / 153; + let day = doy - (153 * mp + 2) / 5 + 1; + let month = if mp < 10 { mp + 3 } else { mp - 9 }; + let year = yoe + era * 400 + i64::from(month <= 2); + format!( + "{year:04}-{month:02}-{day:02}T{:02}:{:02}:{:02}.{:06}Z", + rem / 3600, + rem % 3600 / 60, + rem % 60, + now.subsec_micros() + ) +} + fn remember(list: &mut VecDeque, item: String) { if !list.contains(&item) { list.push_back(item); @@ -501,6 +561,12 @@ struct State { /// Who speaks in each person-lane turn this broker started (the origin /// it sent), by turn id. speakers: HashMap, + /// The words that started each turn of the app's conversation this + /// broker started (a person's message, the system agent's input), in + /// start order. The kernel records a turn's user message only when the + /// turn ends (and never for an interrupted one), so a follower is told + /// them when the turn starts, and history keeps a stopped turn's. + requests: VecDeque, /// The system agent's inputs waiting for the peer's running turn, in /// order (one queue per peer; the person's lane has its own session). queue: VecDeque, @@ -568,6 +634,7 @@ impl Broker { turn_triggers: HashMap::new(), trigger_order: VecDeque::new(), speakers: HashMap::new(), + requests: VecDeque::new(), queue: VecDeque::new(), conversations: 0, cwds: HashMap::new(), @@ -1626,6 +1693,14 @@ impl Inner { let mut st = self.lock(); st.peer_turn = Some(input.turn_id.clone()); remember(&mut st.input_turns, input.turn_id.clone()); + st.note_request(Request { + turn: input.turn_id.clone(), + session: input.session_id.clone(), + lane: LANE_SYSTEM_AGENT, + text: input.text.clone(), + speaker: Speaker { kind: host_tools::TurnOrigin::SystemAgent, label: None }, + at: rfc3339_now(), + }); } let inner = self.clone(); self.rt().spawn(async move { @@ -2135,8 +2210,11 @@ fn time_key(row: &Value) -> Option { /// (its sharing context's transcript) and the system agent's lane (the /// peer's), each row with its `lane` and, for a marked user message, its /// `speaker` and `display_text`, merged by `persisted_at` (a row without one -/// keeps its place in its own lane's order). -fn merged_history(person: Value, system_agent: Value) -> Value { +/// keeps its place in its own lane's order). A turn this broker started +/// whose user message the kernel never recorded (it was stopped) gets its +/// request as a user row (`requests`: this conversation's lanes), at the +/// time it started. +fn merged_history(person: Value, system_agent: Value, requests: &[Request]) -> Value { let mut out = speakers_in_history(person); let lane_rows = |history: Value, lane: &str| -> Vec { let mut rows = match speakers_in_history(history)["messages"].take() { @@ -2150,6 +2228,22 @@ fn merged_history(person: Value, system_agent: Value) -> Value { }; let mut rows = lane_rows(json!({"messages": out["messages"].take()}), LANE_PERSON); rows.extend(lane_rows(system_agent, LANE_SYSTEM_AGENT)); + for request in requests { + let recorded = rows.iter().any(|row| { + row["role"] == "user" + && row["lane"] == request.lane + && (row["turn_id"] == request.turn.as_str() + || row["thread_id"] == request.turn.as_str() + || row["display_text"].as_str().or_else(|| row["content"].as_str()).is_some_and(|t| t.trim() == request.text.trim())) + }); + if !recorded { + rows.push(json!({ + "role": "user", "content": request.text, "display_text": request.text, + "speaker": request.speaker.to_json(), "lane": request.lane, + "turn_id": request.turn, "persisted_at": request.at, "unrecorded": true, + })); + } + } // A stable sort: equal or missing times keep their lane order. let mut last = String::new(); let mut keyed: Vec<(String, Value)> = Vec::with_capacity(rows.len()); @@ -2373,6 +2467,13 @@ impl ContextInner { if let Some(speaker) = speaker { data["speaker"] = speaker.to_json(); } + // The turn's words, when it starts: the kernel sends its user + // message only when the turn ends (never for a stopped one). + if method == "turn/started" { + if let Some(request) = turn_id.and_then(|t| inner.lock().request_of(t).map(Request::to_json)) { + data["request"] = request; + } + } } // A conversation's caller hears only its own turn; a request // context's session is its caller's alone. @@ -2512,6 +2613,14 @@ impl ContextInner { let mut st = inner.lock(); st.note_trigger(&turn_id, trigger); st.speakers.insert(turn_id.clone(), speaker.clone()); + st.note_request(Request { + turn: turn_id.clone(), + session: session.to_owned(), + lane: LANE_PERSON, + text: text.clone(), + speaker: speaker.clone(), + at: rfc3339_now(), + }); } let params = json!({ "session_id": session, @@ -2603,11 +2712,12 @@ impl ContextInner { if self.conversation { // Both lanes: the person's context and the peer's session. let peer_session = inner.lock().peer.as_ref().map(|(_, p)| p.session.clone()); - let system_agent = match peer_session { + let system_agent = match peer_session.clone() { Some(peer) => inner.request("session/hydrate", json!({"session_id": peer, "include": ["messages"]})).await?, None => json!({"messages": []}), }; - history.map(|person| merged_history(person, system_agent)) + let requests: Vec = inner.lock().requests.iter().filter(|r| r.session == session || peer_session.as_deref() == Some(r.session.as_str())).cloned().collect(); + history.map(|person| merged_history(person, system_agent, &requests)) } else { history } @@ -2807,6 +2917,10 @@ impl OctosAppService for Broker { self.bind() } + fn peer_slug(&self) -> Option { + self.peer().map(|(slug, _)| slug) + } + fn model(&self) -> Option { self.0.lock().model.clone() } diff --git a/crates/app-peers/src/contract.rs b/crates/app-peers/src/contract.rs index aae35e8d..26bcb47e 100644 --- a/crates/app-peers/src/contract.rs +++ b/crates/app-peers/src/contract.rs @@ -314,6 +314,13 @@ pub trait OctosAppService: Send + Sync { fn prepare(&self) -> Result<(), String> { Ok(()) } + /// The kernel's slug of the app's peer once it is bound (what the + /// system agent's `peer_list` shows and `peer_send_input` takes, e.g. + /// `os-news-22a12f90`: never the app id). `None` before, or for a + /// service without a peer of its own. + fn peer_slug(&self) -> Option { + None + } /// The app is closing: close every context and release subscriptions. /// Does not stop a shared kernel or other apps' work. fn release(&self); diff --git a/crates/app-peers/tests/broker.rs b/crates/app-peers/tests/broker.rs index b53f6484..be873028 100644 --- a/crates/app-peers/tests/broker.rs +++ b/crates/app-peers/tests/broker.rs @@ -1330,6 +1330,10 @@ fn a_conversation_follows_both_lanes_and_merges_their_history() { assert!(sa.iter().all(|d| d["speaker"]["kind"] == "system_agent" && d["lane"] == "system_agent"), "{sa:?}"); let user = sa.iter().find(|d| d["method"] == "projection/envelope").unwrap(); assert_eq!(user["display_text"], "check the inbox"); + // The kernel sends a turn's user message when the turn ends: its words + // come with its start, from what the broker sent. + let started = sa.iter().find(|d| d["method"] == "turn/started").unwrap(); + assert_eq!(started["request"], json!({"text": "check the inbox", "speaker": {"kind": "system_agent"}})); // The person's own message: the follower sees it too, in the person's // lane, with its speaker. let (sink, rx) = collect(); @@ -1340,6 +1344,10 @@ fn a_conversation_follows_both_lanes_and_merges_their_history() { seen.iter().any(|d| d["method"] == "turn/completed" && d["speaker"]["kind"] == "person" && d["speaker"]["label"] == "Rinx" && d["lane"] == "person" && d["params"]["session_id"] == lane.as_str()), "{seen:?}" ); + assert!( + seen.iter().any(|d| d["method"] == "turn/started" && d["lane"] == "person" && d["request"] == json!({"text": "thanks", "speaker": {"kind": "person", "label": "Rinx"}})), + "{seen:?}" + ); // A caller's sink hears only its own turn: nothing of the system agent's. let (sink, rx) = collect(); chat.call(ContextOp::History, sink).unwrap(); @@ -1396,6 +1404,47 @@ fn a_conversation_follows_both_lanes_and_merges_their_history() { drop(broker); } +/// A stopped turn's user message is never recorded by the kernel (it +/// writes a turn's rows when the turn ends): the conversation's history +/// keeps its request from the broker, in its lane, at the time it started; +/// a recorded one is not repeated. +#[test] +fn a_stopped_turns_request_stays_in_the_conversations_history() { + let host = Arc::new(RecordingHost::default()); + let (broker, script) = new_broker_with(&ALL, Some(host), None); + { + let mut s = script.lock().unwrap(); + s.hold_turns = true; + s.interrupts_end = true; + } + broker.set_account(Some("@a:x")); + let chat = broker.open_conversation(spec("@a:x", "ui", &ALL)).unwrap(); + let (sink, rx) = collect(); + chat.call(ContextOp::TurnFrom { text: "a long digest please".into(), trigger: TurnTrigger::Person }, sink).unwrap(); + wait_for("the person's turn", || calls_of(&script, "turn/start").len() == 1); + let (sink, stop_rx) = collect(); + chat.call(ContextOp::Interrupt, sink).unwrap(); + assert_eq!(complete(&stop_rx).unwrap()["turns"][0]["lane"], "person"); + assert!(complete(&rx).is_err(), "the stopped turn ends with its error"); + { + let mut s = script.lock().unwrap(); + s.history = vec![json!({"role": "user", "content": "[from the person: Rinx] earlier", "persisted_at": "2020-01-01T00:00:00Z"})]; + s.peer_history = vec![]; + } + let (sink, rx) = collect(); + chat.call(ContextOp::History, sink).unwrap(); + let rows = complete(&rx).unwrap()["messages"].as_array().unwrap().clone(); + let shown: Vec<(String, String)> = rows.iter().map(|r| (r["lane"].as_str().unwrap().to_owned(), r["display_text"].as_str().unwrap_or("").to_owned())).collect(); + assert_eq!(shown, [("person".to_owned(), "earlier".to_owned()), ("person".to_owned(), "a long digest please".to_owned())]); + assert_eq!(rows[1]["speaker"], json!({"kind": "person", "label": "Rinx"})); + // Once the kernel has the row, it is not added twice. + script.lock().unwrap().history.push(json!({"role": "user", "content": "[from the person: Rinx] a long digest please", "persisted_at": "2020-01-01T00:00:01Z"})); + let (sink, rx) = collect(); + chat.call(ContextOp::History, sink).unwrap(); + assert_eq!(complete(&rx).unwrap()["messages"].as_array().unwrap().len(), 2); + drop(broker); +} + /// A kernel that ignores `share_history` would give the person a plain /// context the system agent never sees: the conversation is refused. #[test] diff --git a/crates/app-peers/tests/real_kernel.rs b/crates/app-peers/tests/real_kernel.rs index a59026ca..1ac0c3ea 100644 --- a/crates/app-peers/tests/real_kernel.rs +++ b/crates/app-peers/tests/real_kernel.rs @@ -984,14 +984,28 @@ fn the_person_and_the_system_agent_talk_in_parallel_lanes_that_share_history() { assert_eq!(answer["lane"], "person"); assert!(answer["text"].as_str().unwrap().contains("hello from the app"), "{answer}"); - // The system agent's input to the same peer, in its own lane. + // The system agent's input to the same peer, in its own lane. The + // system session may still be busy with the previous turn: after its + // `peer_send_input` it finishes its own answer, and the kernel may wake + // it with the peer's result. The test is the system chat here, so it + // waits for its session like the chat does (the kernel queues nothing: + // `turn_in_progress`). let system_turn = |text: String| { - rinx.host_request( - "turn/start", - json!({"session_id": "_main:api:octosense#system", "turn_id": uuid_like(), - "input": [{"kind": "text", "text": text}]}), - ) - .expect("system turn"); + let params = json!({"session_id": "_main:api:octosense#system", "turn_id": uuid_like(), + "input": [{"kind": "text", "text": text}]}); + for _ in 0..120 { + match rinx.host_request("turn/start", params.clone()) { + Err(e) if e.contains("turn_in_progress") => { + eprintln!("the system session is still busy: {e}"); + std::thread::sleep(Duration::from_millis(500)); + } + other => { + other.expect("system turn"); + return; + } + } + } + panic!("the system session stayed busy for a minute"); }; let history_until = |done: &dyn Fn(&[Value]) -> bool| -> Vec { let mut rows = Vec::new(); diff --git a/crates/shell/src/agents.rs b/crates/shell/src/agents.rs index 084f8339..2b8333fe 100644 --- a/crates/shell/src/agents.rs +++ b/crates/shell/src/agents.rs @@ -250,9 +250,42 @@ pub fn ask(app: &AgentApp) -> Access { // ------------------------------------------------------------ the system agent +/// The kernel's slug of `app`'s peer, once it is bound: what the system +/// agent's `peer_list` shows and `peer_send_input` takes (`os-news-22a12f90`, +/// never the app id `os.news`). +pub fn peer_slug(app: &AgentApp) -> Option { + if !app.native { + return crate::ai_host::contained::peer_slug(&app.id); + } + native_slug(&app.id) +} + +#[cfg(kernel)] +fn native_slug(app: &str) -> Option { + crate::ai_host::app_peers::broker::live(app)?.peer().map(|(slug, _)| slug) +} + +#[cfg(not(kernel))] +fn native_slug(_app: &str) -> Option { + None +} + +/// How the system agent reaches an allowed app's agent: by its peer slug +/// with `peer_send_input`, never by the app id. +fn reach(app: &AgentApp, slug: Option<&str>) -> String { + match slug { + Some(slug) => format!("Reach it with peer_send_input and the peer slug \"{slug}\" (not the app id {})", app.id), + None => format!( + "Find its peer slug with peer_list (the peer named \"{} \u{2026}\"), then use that slug with peer_send_input (not the app id {})", + peer_label(app), + app.id + ), + } +} + /// One app's line for the system agent: its name, its peer and where it /// stands, and what to do. -fn line(app: &AgentApp) -> Value { +pub(crate) fn line(app: &AgentApp) -> Value { let access = access(&app.id); let peer = match (access, prepared(&app.id)) { (Access::Allowed, Some(Prepared::Ready)) => "ready: in peer_list".to_string(), @@ -262,12 +295,13 @@ fn line(app: &AgentApp) -> Value { (Access::Allowed, None) => "not started yet".to_string(), _ => "none".to_string(), }; + let slug = if access == Access::Allowed { peer_slug(app) } else { None }; let what = match access { - Access::Allowed => format!("Reach it with peer_send_input (its peer is named \"{} …\" in peer_list).", peer_label(app)), + Access::Allowed => format!("{}.", reach(app, slug.as_deref())), Access::NotAsked => format!("{}'s assistant is not yet allowed. Call agents.ask to ask the person, or tell them to open {} and use Ask {}.", app.name, app.name, app.name), Access::Off => format!("{}'s assistant is off. Only the person can turn it on (Settings › Assistant › Approvals); tell them.", app.name), }; - json!({"app": app.id, "name": app.name, "access": access.as_str(), "peer": peer, "what_to_do": what}) + json!({"app": app.id, "name": app.name, "access": access.as_str(), "peer": peer, "peer_slug": slug, "what_to_do": what}) } /// The name the kernel gives the app's peer (`