From 9415beb1acc16cd39ae758b1bf3916fef12c770e Mon Sep 17 00:00:00 2001 From: ymote <151983+ymote@users.noreply.github.com> Date: Wed, 30 Sep 2026 11:55:11 -0700 Subject: [PATCH] peer link: prove a process app hears the system agent's lane live; ADR 0004's two lanes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A real-kernel test for the process path (ADR 0004 §6): a process app opens its conversation over the peer link (`octos.session.open` without a client); the system agent sends the app's agent a message (`peer_send_input`); the turn reaches the process's link live as `conversation` frames with lane `system_agent` and its speaker: `turn/started` with the request, the user message without the kernel's marker, the answer (`assistant_delta`) and the end (`turn_terminal`), with no history request. The recorded frames are the fixtures of Makepad's client (makepad#77), which now reads them and queues a second message on a handle. The framing test also reads a conversation frame as Makepad writes it. ADR 0004: the two lanes' client items are done in Makepad; process apps get them once the runtime pin reaches it (the repin is pending). Co-Authored-By: Claude Opus 5.5 --- .../shell/src/host_tools/real_kernel_tests.rs | 149 ++++++++++++++++++ crates/shell/src/peer_link/tests.rs | 5 + crates/shell/src/peer_link/wire.rs | 4 +- .../adr/0004-native-apps-hosting-and-peers.md | 4 +- 4 files changed, 159 insertions(+), 3 deletions(-) diff --git a/crates/shell/src/host_tools/real_kernel_tests.rs b/crates/shell/src/host_tools/real_kernel_tests.rs index e79ebdb7..0ead5e95 100644 --- a/crates/shell/src/host_tools/real_kernel_tests.rs +++ b/crates/shell/src/host_tools/real_kernel_tests.rs @@ -311,6 +311,155 @@ fn real_kernel_an_allowed_apps_agent_is_prepared_listed_and_shares_its_conversat drop(model); } +/// The shell's side of a process app's peer link, for the test: News's +/// agent granted and allowed, its service the test's broker. +struct ProcessHost(Broker); + +impl crate::peer_link::PeerHost for ProcessHost { + fn granted(&self, _app: &str) -> std::collections::BTreeSet { + OCTOS_SERVICES.iter().map(|s| s.to_string()).collect() + } + fn keeps_accounts(&self, _app: &str) -> bool { + false + } + fn consent(&mut self, _app: &str) -> bool { + true + } + fn service(&mut self, _app: &str, _services: &std::collections::BTreeSet) -> Option> { + Some(Arc::new(self.0.clone())) + } + fn tool_rule(&self, _app: &str, _tool: &str) -> Option<(crate::native_apps::Confirm, bool)> { + None + } + fn request_approval(&mut self, _app: &str, _tool: crate::approvals::ToolSpec, _args: Value, _caller: crate::approvals::Caller, _context: crate::approvals::RequestContext) -> crate::approvals::Route { + crate::approvals::Route::Refused("no approvals in this test".into()) + } + fn take_decisions(&mut self) -> Vec<(crate::approvals::RequestId, crate::approvals::Decision, String)> { + Vec::new() + } + fn app_confirm_answered(&mut self, _id: &crate::approvals::RequestId, _approved: bool, _reason: &str) {} + fn link_opened(&mut self, _app: &str) {} + fn link_closed(&mut self, _app: &str) {} +} + +/// ADR 0004 §6 on a REAL kernel, for a PROCESS app: the app opens its +/// conversation over the peer link (`octos.session.open` without a +/// `client`), and when the system agent sends the app's agent a message +/// (`peer_send_input`), that turn (the system agent's lane) reaches the +/// app's process live as `conversation` frames, each with its lane and +/// speaker: `turn/started` with the request, streamed text, and the end, +/// without the app asking for history. The frames are printed +/// (`[peer-link frame]`) as the fixtures of Makepad's client. +#[test] +fn real_kernel_a_process_app_hears_the_system_agents_lane_live() { + let Some(program) = kernel() else { return }; + let _factory = crate::agents::FACTORY_TESTS.lock().unwrap_or_else(|e| e.into_inner()); + let dir = std::env::temp_dir().join(format!("octosense-shell-peer-link-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&dir); + std::fs::create_dir_all(&dir).unwrap(); + let offered_log = dir.join("offered.jsonl"); + let model = start_model(&offered_log); + let core_dir = dir.join("octos-home/.octos"); + write_profile(&core_dir, model.1); + let core = Core::new(Options::default().core_dir(&core_dir).program(&program)); + let home = dir.join("home"); + std::fs::create_dir_all(&home).unwrap(); + if crate::approvals::with(|_| ()).is_none() { + crate::approvals::init(&home); + } + // The person allowed News's agent (else its peer refuses `peer/input`). + crate::approvals::with(|a| a.consent.set(&crate::approvals::rules::ApprovalGesture::sheet_tap(), "os.news", true, crate::approvals::now())); + // News's tools (so its peer has a host route for `peer/input`). + let host_dir = dir.join("apps/.host"); + std::fs::create_dir_all(&host_dir).unwrap(); + octosense_news_service::register_with(octosense_news_service::Options::default().host_dir(&host_dir).timer(false)); + let bundle = super::script_apps::tests::stamped_bundle("news", "peer-link", |_, _| {}); + super::script_apps::install("os.news", super::script_apps::from_bundle(&bundle).unwrap(), host_dir.clone()); + let services = OCTOS_SERVICES.iter().map(|s| s.to_string()).collect(); + let mut cfg = BrokerConfig::new(Deployment::Hosted, "_main", "_main:api:octosense#system", "card.os.news", "News", services); + cfg.state_dir = Some(dir.join("host-state")); + cfg.tool_host = Some(ToolHostHandle(Arc::new(super::ShellToolHost) as Arc)); + let broker = Broker::new(cfg, Arc::new(CoreConnector::shared(core.clone()))); + + // The app's process connects its hub socket; its frames are recorded. + let frames: Arc>> = Arc::default(); + let f = frames.clone(); + let out: crate::peer_link::FrameOut = Arc::new(move |json: String| f.lock().unwrap().push(json)); + let mut links = crate::peer_link::PeerLinks::new(Box::new(ProcessHost(broker.clone())), Box::new(crate::peer_link::RecordingToolRelay::default())); + assert!(links.connected(7, "os.news", out)); + let downs = |frames: &Arc>>| frames.lock().unwrap().iter().filter_map(|f| crate::peer_link::wire::Down::parse(f)).collect::>(); + // Its conversation, as Makepad's client opens it (no `client`). + assert!(links.on_frame(7, "os.news", r#"{"octos_peer":{"up":"request","req_id":1,"method":"octos.session.open","args":{"client":null}}}"#, None)); + let deadline = Instant::now() + Duration::from_secs(90); + let context = loop { + super::pump(); + if let Some(reply) = downs(&frames).into_iter().find_map(|d| match d { + crate::peer_link::wire::Down::Reply { req_id: 1, result } => Some(result), + _ => None, + }) { + break reply.expect("the conversation opened")["context"].as_str().unwrap().to_string(); + } + assert!(Instant::now() < deadline, "the conversation did not open: {:?}", frames.lock().unwrap()); + std::thread::sleep(Duration::from_millis(50)); + }; + let (slug, _) = broker.peer().expect("the peer is bound"); + + // The system agent sends News's agent a message. + let mut chat = crate::system_chat::session::Driver::new(Box::new(SystemLink(core.clone()))); + chat.command(crate::system_chat::session::Command::Open); + let deadline = Instant::now() + Duration::from_secs(60); + while chat.model.phase() != &crate::system_chat::model::Phase::Ready { + chat.step(Duration::from_millis(50)); + assert!(Instant::now() < deadline, "the system session did not open: {:?}", chat.model.phase()); + } + chat.command(crate::system_chat::session::Command::Send(format!("TELL_PEER_SHOW:{slug}"))); + let deadline = Instant::now() + Duration::from_secs(120); + let lane = loop { + chat.step(Duration::from_millis(50)); + super::pump(); + let lane: Vec = downs(&frames) + .into_iter() + .filter_map(|d| match d { + crate::peer_link::wire::Down::Conversation { context: c, event } if c == context && event["lane"] == "system_agent" => Some(event), + _ => None, + }) + .collect(); + // The kernel's v2 envelopes: the turn's end (`turn_terminal`) and + // its answer (`assistant_delta`), in either order. + let envelope = |kind: &str| lane.iter().any(|e| e["params"]["payload"]["type"] == kind); + if envelope("turn_terminal") && envelope("assistant_delta") { + break lane; + } + assert!(Instant::now() < deadline, "the process never heard the system agent's lane: {:?}", frames.lock().unwrap()); + }; + for frame in frames.lock().unwrap().iter() { + eprintln!("[peer-link frame] {frame}"); + } + links.process_gone(7); + drop(chat); + broker.release(); + let _ = std::fs::remove_dir_all(&bundle); + core.shutdown_within(Duration::from_secs(5)); + let _ = std::fs::remove_dir_all(&dir); + drop(model); + + // Live: the turn's start (with its request and speaker) came before its + // end, the system agent's words and the answer streamed, each event with + // its speaker; no history was asked for. + let started = lane.iter().position(|e| e["method"] == "turn/started").expect("turn/started reached the app"); + let ended = lane.iter().position(|e| e["params"]["payload"]["type"] == "turn_terminal").unwrap(); + assert!(started < ended, "{lane:?}"); + assert_eq!(lane[started]["speaker"]["kind"], "system_agent", "{}", lane[started]); + assert_eq!(lane[started]["request"]["text"], "SHOW_SHARED", "the system agent's words: {}", lane[started]); + assert!(lane.iter().any(|e| e["params"]["payload"]["type"] == "user_message" && e["display_text"] == "SHOW_SHARED"), "{lane:?}"); + assert!( + lane.iter().any(|e| e["params"]["payload"]["type"] == "assistant_delta" && e["params"]["payload"]["data"]["text"].as_str().is_some_and(|t| t.starts_with("SHARED"))), + "the answer streamed to the app: {lane:?}" + ); + assert!(lane.iter().filter(|e| e["method"] != "session/orchestration" && e["method"] != "context/normalization_reported").all(|e| e["speaker"]["kind"] == "system_agent"), "{lane:?}"); + assert!(!frames.lock().unwrap().iter().any(|f| f.contains("octos.session.history")), "no history request"); +} + struct SystemLink(Core); impl crate::system_chat::session::Connector for SystemLink { fn connect(&mut self) -> Result, crate::system_chat::session::Unavailable> { diff --git a/crates/shell/src/peer_link/tests.rs b/crates/shell/src/peer_link/tests.rs index 4de344eb..5c80eeae 100644 --- a/crates/shell/src/peer_link/tests.rs +++ b/crates/shell/src/peer_link/tests.rs @@ -245,6 +245,11 @@ fn the_shell_reads_the_makepad_clients_frames() { assert_eq!(Up::parse(r#"{"octos_peer":{"up":"request","req_id":1,"method":"octos.turn.start"},"from":"rinx"}"#), None); let big = request(1, "octos.turn.start", json!({"text": "a".repeat(wire::MAX_FRAME_BYTES)})); assert_eq!(Up::parse(&big), None); + // A conversation frame as Makepad's client writes it (its own key + // order) reads back here; the client reads the shell's (its fixtures + // are frames recorded from this link on a real kernel). + let makepad_conversation = r#"{"octos_peer":{"down":"conversation","context":"pl7-1","event":{"method":"turn/started"}}}"#; + assert_eq!(Down::parse(makepad_conversation), Some(Down::Conversation { context: "pl7-1".into(), event: json!({"method": "turn/started"}) })); let call = Down::ToolCall(wire::ToolCallDown { call_id: "k2".into(), name: "send".into(), diff --git a/crates/shell/src/peer_link/wire.rs b/crates/shell/src/peer_link/wire.rs index f2dc0250..6f545f78 100644 --- a/crates/shell/src/peer_link/wire.rs +++ b/crates/shell/src/peer_link/wire.rs @@ -83,7 +83,9 @@ pub enum Down { /// `person` or `system_agent`), for a context opened without a /// `client` (the app's conversation): every turn, whoever speaks /// (`event.speaker`), after the request that opened it answered. - /// A client that does not know the frame ignores it. + /// Makepad's client (`makepad_ai_services::peer`) surfaces it as + /// `PeerEvent::Conversation`; a client that does not know the frame + /// ignores it. Conversation { context: String, event: Value }, } diff --git a/docs/adr/0004-native-apps-hosting-and-peers.md b/docs/adr/0004-native-apps-hosting-and-peers.md index d2bb89f2..ec58b202 100644 --- a/docs/adr/0004-native-apps-hosting-and-peers.md +++ b/docs/adr/0004-native-apps-hosting-and-peers.md @@ -354,7 +354,7 @@ How it is turned on and kept from leaking into normal use: - [x] **An agent for every app that declares one** (section 4, 2026-09-29): a live run on `625d6cc` found News (tools.json, no `octos.*`) without a peer, an empty `peer_list` for the system agent, Settings listing only Rinx, and no place for the person to talk to an app's agent. Now the shell lists every app with an agent, prepares a consented app's peer, tells the system agent about the ones it cannot list, and draws "Ask ". Still open: a native app's peer exists only while the app is open; the speaker label of a script app's person turns is its app id (the peer's name, kept for resume), not its display name; prepared peers hold a kernel connection, so the kernel's idle stop waits for them. - [x] **The "Ask " panel's live run, follow-ups** (2026-09-29, [#189](https://github.com/OctoSense-org/OctoSense/pull/189) and the App Hub pin after it): the panel orders both lanes by one clock (each lane's `cursor.seq` is its own; a turn's request line stands where the turn started, handed over by the broker with `turn/started`, since octos writes a turn's user message only when it ends and never for a stopped turn); Stop says "interrupted" once; the system agent is given each peer's slug for `peer_send_input`; and a script app's agent may keep octos's `ask_user_question` (App Hub's one kernel tool for contained apps, `KERNEL_TOOLS`, "Ask you questions" in the store), handed to its peer's `generic_tools` by G3. News's agent keeps it, and its questions of the person's lane are asked in "Ask News". -*Still open after the implementation (2026-09-29):* the Windows AppContainer sandbox ([#137](https://github.com/OctoSense-org/OctoSense/issues/137)); the Linux sandbox has run only in CI; process apps see the system agent's lane only through history until Makepad's peer client reads `conversation` frames; a second message on one conversation handle is refused rather than queued; Rinx still has to send turn triggers and hand over its send sheet; the OpenHarmony build with octos#2619's BoringSSL (`btls-sys`) is untested. +*Still open after the implementation (2026-09-29):* the Windows AppContainer sandbox ([#137](https://github.com/OctoSense-org/OctoSense/issues/137)); the Linux sandbox has run only in CI; ~~process apps see the system agent's lane only through history until Makepad's peer client reads `conversation` frames; a second message on one conversation handle is refused rather than queued~~ (done in Makepad's client, [makepad#77](https://github.com/OctoSense-org/makepad/pull/77); in the build once OctoSense's runtime pin reaches it; see the gaps below); Rinx still has to send turn triggers and hand over its send sheet; the OpenHarmony build with octos#2619's BoringSSL (`btls-sys`) is untested. Gaps between this decision and `main` found after it was accepted (2026-09-28; see [the architecture overview](../architecture.md#where-the-code-and-the-adrs-disagree)): @@ -369,7 +369,7 @@ Gaps between this decision and `main` found after it was accepted (2026-09-28; s - **Contained script apps' cards** (`crates/ai-host/src/contained.rs`): streamed text is dropped (the app gets only the final answer, and reads the rest of the conversation, both lanes merged, with `octos.session.history`: App Hub's service calls push no events), an approval the router does not take is declined (listed in `denied_approvals`), and every peer uses the fixed account `device`. *(Until #166 each app had one request context shared by all its cards and UI; #166 moved them to the peer's shared conversation; now they talk in the person's lane, a sharing context per handle generation, section 6.)* - [x] **`user_question/requested` for cards**: by design a card never answers its agent's question; the broker hands every question on the peer's session or a context to the shell's request model (`crates/shell/src/questions/`), routed by the turn's origin, and the person answers on a shell surface (the app's conversation, or the system chat). - **A failed host `turn/start` for a `peer/input` is only logged** (`start_input` in `broker.rs`); the system agent is not told. Since #166 a start refused `turn_in_progress` is retried with the same turn id; any other refusal is still only logged. - - **The two lanes, left open**: the Makepad peer client (`makepad_ai_services::peer`) does not read the `conversation` frames yet (it drops events of a request that has answered), so a process app shows the system agent's turns only through `octos.session.history` until it does; a conversation handle runs one turn of its own at a time (a second message on the same handle is refused, not queued); the shared history block is the kernel's defaults (the shell sends `share_history: {}`); and the accepted hazards of section 6 (unlocked file edits across lanes, shared unknown-outcome marks, a slight budget overshoot). *(octos [#2627](https://github.com/octos-org/octos/issues/2627), `peer_gather` refusing the kernel's own `result.md`, is fixed by octos #2630.)* + - **The two lanes, left open**: ~~the Makepad peer client (`makepad_ai_services::peer`) does not read the `conversation` frames yet~~ *(done 2026-09-30, [makepad#77](https://github.com/OctoSense-org/makepad/pull/77): the client surfaces every `conversation` frame of a context it opened as `PeerEvent::Conversation`, with its lane, speaker, envelope type and answer delta, so a process app follows the system agent's turns live; `real_kernel_a_process_app_hears_the_system_agents_lane_live` in `crates/shell/src/host_tools/real_kernel_tests.rs` proves on a real kernel that the shell streams that lane to a process app's link, and those recorded frames are the client's test fixtures)*; ~~a conversation handle runs one turn of its own at a time (a second message on the same handle is refused, not queued)~~ *(a handle still runs one turn at a time, but Makepad's client now queues a second message behind the running turn and sends it when that turn answers: at most 8 wait per handle, beyond that `queue_full`; closing the handle, by the app or the shell, answers the running and waiting ones `cancelled`, and Stop cancels what waits. An in-process module calling `OctosContext` directly is still refused a second turn by the broker)*; process apps get both only once OctoSense's runtime pin reaches a Makepad with #77 (pending: that repin waits while another repin of the chain is open, Octoscript-Makepad [#50](https://github.com/OctoSense-org/Octoscript-Makepad/pull/50)); the shared history block is the kernel's defaults (the shell sends `share_history: {}`); and the accepted hazards of section 6 (unlocked file edits across lanes, shared unknown-outcome marks, a slight budget overshoot). *(octos [#2627](https://github.com/octos-org/octos/issues/2627), `peer_gather` refusing the kernel's own `result.md`, is fixed by octos #2630.)* - **The `peer/input` route belongs to whichever connection registered the peer's tools last**: octos `set_host_route` (`crates/octos-cli/src/peers/host_tools.rs`) replaces any earlier route. Untested with more than one instance of an app. - [x] **A shell-native chat with the system agent**: done. `crates/shell/src/system_chat/` (Setup → Assistant → Assistant chat, F8; full screen on a phone) talks to `_main:api:octosense#system` over the shell's own kernel connection, and its approvals go through the approval router, batched per request.