From af8b4fbd33572b8e0917c97cc357e62b811b2633 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Tue, 13 Oct 2026 15:15:02 +0300 Subject: [PATCH 1/2] feat(tinycomputer): StartTask reads and writes per-site memory through SiteMemoryHost (contract 2.17) StartTask takes an optional site; the module loads what finished tasks there left, hands the task a saved plan and grounding hints, follows it, and saves what it learned once it ends. Load/Save carry an epoch so a site forgotten while a task runs is not written back. Co-authored-by: Medulla --- Cargo.lock | 1 + .../src/agent/types/request.rs | 7 + .../tinycomputer-bus/src/site_memory/mod.rs | 14 +- .../src/site_memory/site_memory_tests.rs | 23 ++ crates/tinycomputer-bus/src/version/mod.rs | 4 +- .../src/version/version_tests.rs | 6 +- .../src/bin/task_live/main.rs | 1 + crates/tinycomputer/Cargo.toml | 2 + .../src/tinybus_module/dispatch/mod.rs | 25 +- .../src/tinybus_module/dispatch/service.rs | 57 ++- crates/tinycomputer/src/tinybus_module/mod.rs | 6 +- .../src/tinybus_module/site_memory/mod.rs | 300 ++++++++++++++ .../site_memory/site_memory_tests.rs | 377 ++++++++++++++++++ crates/tinycomputer/tests/public_api_tests.rs | 2 +- docs/crates/tinycomputer-bus/versioning.md | 12 +- 15 files changed, 818 insertions(+), 19 deletions(-) create mode 100644 crates/tinycomputer/src/tinybus_module/site_memory/mod.rs create mode 100644 crates/tinycomputer/src/tinybus_module/site_memory/site_memory_tests.rs diff --git a/Cargo.lock b/Cargo.lock index 54ac2a22..41eb0eb4 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2623,6 +2623,7 @@ dependencies = [ "tinycomputer-accessibility", "tinycomputer-browser", "tinycomputer-bus", + "tinycomputer-core", "tinycomputer-desktop", "tinycomputer-engine", "tokio", diff --git a/crates/tinycomputer-bus/src/agent/types/request.rs b/crates/tinycomputer-bus/src/agent/types/request.rs index acad9b9f..18e0b79a 100644 --- a/crates/tinycomputer-bus/src/agent/types/request.rs +++ b/crates/tinycomputer-bus/src/agent/types/request.rs @@ -77,6 +77,13 @@ pub struct StartTaskRequest { /// records only. #[serde(skip_serializing_if = "Option::is_none")] pub output: Option, + /// The site the task starts on, a plain host name. When set and the host + /// serves `SiteMemoryHost`, the module hands the task what finished tasks + /// on that site left (a saved plan for the same goal and facts, grounding + /// hints) and, once the task ends, keeps what it learned there. Absent, + /// the task neither reads nor writes site memory. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub site: Option, } /// What a finished task returns, beyond its raw records. diff --git a/crates/tinycomputer-bus/src/site_memory/mod.rs b/crates/tinycomputer-bus/src/site_memory/mod.rs index 4d1a12c1..45b20262 100644 --- a/crates/tinycomputer-bus/src/site_memory/mod.rs +++ b/crates/tinycomputer-bus/src/site_memory/mod.rs @@ -101,6 +101,13 @@ pub struct LoadSiteResponse { /// What is kept; empty for a site with nothing. #[serde(default)] pub memory: SiteMemory, + /// Counts the times the site was forgotten (`Forget`). The module sends + /// back the value it loaded with its `Save`, and the host refuses a save + /// whose epoch is no longer current, so a task that was running when its + /// site was forgotten cannot write it back. A host that never forgets + /// under a running task may always answer `0`. + #[serde(default)] + pub epoch: u64, } /// `Save`: replaces one site's memory. @@ -110,12 +117,17 @@ pub struct SaveSiteRequest { pub site: String, /// What to keep; empty removes the site. pub memory: SiteMemory, + /// The `epoch` the site was loaded with; see [`LoadSiteResponse::epoch`]. + #[serde(default)] + pub epoch: u64, } /// The reply to `Save`. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] pub struct SaveSiteResponse { - /// Whether the host holds the memory as sent. + /// Whether the host holds the memory as sent. `false` when the host + /// refused it: too large, learning switched off, or the site forgotten + /// since it was loaded (a stale `epoch`). pub saved: bool, } diff --git a/crates/tinycomputer-bus/src/site_memory/site_memory_tests.rs b/crates/tinycomputer-bus/src/site_memory/site_memory_tests.rs index bcbb3287..dee0d343 100644 --- a/crates/tinycomputer-bus/src/site_memory/site_memory_tests.rs +++ b/crates/tinycomputer-bus/src/site_memory/site_memory_tests.rs @@ -38,3 +38,26 @@ fn forgetting_without_a_site_means_every_site() { let request: ForgetSiteRequest = serde_json::from_str("{}").unwrap(); assert_eq!(request.site, None); } + +#[test] +fn an_epoch_is_optional_on_the_wire() { + let load: LoadSiteResponse = serde_json::from_str(r#"{"memory":{}}"#).unwrap(); + assert_eq!(load.epoch, 0); + let save: SaveSiteRequest = + serde_json::from_str(r#"{"site":"example.com","memory":{}}"#).unwrap(); + assert_eq!(save.epoch, 0); + let save: SaveSiteRequest = + serde_json::from_str(r#"{"site":"example.com","memory":{},"epoch":3}"#).unwrap(); + assert_eq!(save.epoch, 3); +} + +#[test] +fn a_start_task_site_is_optional_and_left_out_when_absent() { + use crate::agent::StartTaskRequest; + let request: StartTaskRequest = serde_json::from_str("{}").unwrap(); + assert_eq!(request.site, None); + let json = serde_json::to_value(&request).unwrap(); + assert!(json.get("site").is_none()); + let request: StartTaskRequest = serde_json::from_str(r#"{"site":"example.com"}"#).unwrap(); + assert_eq!(request.site.as_deref(), Some("example.com")); +} diff --git a/crates/tinycomputer-bus/src/version/mod.rs b/crates/tinycomputer-bus/src/version/mod.rs index 0751f8d0..5a94b741 100644 --- a/crates/tinycomputer-bus/src/version/mod.rs +++ b/crates/tinycomputer-bus/src/version/mod.rs @@ -8,7 +8,7 @@ //! release workflow bumps, which tracks the shipped artifact. /// The wire contract version this crate defines. -pub const CONTRACT_VERSION: (u32, u32) = (2, 16); +pub const CONTRACT_VERSION: (u32, u32) = (2, 17); /// Returns whether a host holding [`CONTRACT_VERSION`] can bind to a module /// reporting `module`. @@ -22,7 +22,7 @@ pub const CONTRACT_VERSION: (u32, u32) = (2, 16); /// ``` /// # use tinycomputer_bus::{is_compatible, CONTRACT_VERSION}; /// assert!(is_compatible(CONTRACT_VERSION)); -/// assert!(is_compatible((2, 16))); +/// assert!(is_compatible((2, 17))); /// assert!(!is_compatible((2, 11))); /// assert!(!is_compatible((1, 8))); /// ``` diff --git a/crates/tinycomputer-bus/src/version/version_tests.rs b/crates/tinycomputer-bus/src/version/version_tests.rs index 893bf036..25603b63 100644 --- a/crates/tinycomputer-bus/src/version/version_tests.rs +++ b/crates/tinycomputer-bus/src/version/version_tests.rs @@ -6,7 +6,7 @@ use super::{CONTRACT_VERSION, binds, is_compatible}; #[test] fn the_shipped_contract_version_is_pinned() { - assert_eq!(CONTRACT_VERSION, (2, 16)); + assert_eq!(CONTRACT_VERSION, (2, 17)); } #[test] @@ -16,7 +16,7 @@ fn the_contract_binds_to_itself() { #[test] fn a_newer_minor_on_the_module_side_binds() { - assert!(is_compatible((2, 16))); + assert!(is_compatible((2, 17))); assert!(is_compatible((2, 97))); // 2.12 adds replayable GlobeRead and joined terminal GlobeShutdown. // A 2.11 module does not implement those members. @@ -28,6 +28,8 @@ fn a_newer_minor_on_the_module_side_binds() { // 2.14 adds BrowserClaimProfile and BrowserReleaseProfile. // 2.15 adds browser::tool: the recorded `browser` tool schema and ToolArgs. // 2.16 adds desktop_tool: the eight desktop tool declarations and request builders. + // 2.17 adds `StartTask.site` and the `epoch` on `SiteMemoryHost` Load/Save. + assert!(!is_compatible((2, 16))); assert!(!is_compatible((2, 15))); assert!(!is_compatible((2, 14))); assert!(!is_compatible((2, 13))); diff --git a/crates/tinycomputer-examples/src/bin/task_live/main.rs b/crates/tinycomputer-examples/src/bin/task_live/main.rs index 4dd05f6a..53a9a7df 100644 --- a/crates/tinycomputer-examples/src/bin/task_live/main.rs +++ b/crates/tinycomputer-examples/src/bin/task_live/main.rs @@ -161,6 +161,7 @@ async fn main() -> Result<(), LabError> { trace: true, output, memory, + site: None, }) .await?; let limit = Duration::from_secs( diff --git a/crates/tinycomputer/Cargo.toml b/crates/tinycomputer/Cargo.toml index 43689cbe..7499b060 100644 --- a/crates/tinycomputer/Cargo.toml +++ b/crates/tinycomputer/Cargo.toml @@ -27,6 +27,8 @@ getrandom = "0.4" # takes one dependency rather than two, and so `tinycomputer::SnapshotRequest` # and `tinycomputer_bus::SnapshotRequest` are the same type. tinycomputer-bus = { workspace = true } +# The per-site memory rules (reuse, fact filtering, expiry) a task is run under. +tinycomputer-core = { workspace = true } # The agent-desktop adapter. `Desktop`, `Error`, and `Result` are re-exported. tinycomputer-desktop = { workspace = true } # The agent runtime behind the agentic and task members: Jev loops, goals, diff --git a/crates/tinycomputer/src/tinybus_module/dispatch/mod.rs b/crates/tinycomputer/src/tinybus_module/dispatch/mod.rs index 1c698f23..ff2d16ee 100644 --- a/crates/tinycomputer/src/tinybus_module/dispatch/mod.rs +++ b/crates/tinycomputer/src/tinybus_module/dispatch/mod.rs @@ -48,6 +48,7 @@ use tinycomputer_bus::agent::{ }; use super::runner::jev_not_configured; +use super::site_memory::SiteLink; use crate::Desktop; use tinycomputer_browser::{ Action, Browser, DownloadWaitRequest, EvaluateRequest, NavigateRequest, OutputRequest, @@ -75,6 +76,8 @@ pub(crate) struct DesktopService { /// `BrowserOpenSession` wherever the caller left them unset. browser_defaults: crate::tinybus_module::config::BrowserDefaults, accessibility: Arc, + /// The host's per-site memory, when the module was set up on a bus. + site_link: Option>, } #[tinybus::interface(name = "ai.tinyhumans.tinycomputer.Desktop")] @@ -149,7 +152,7 @@ impl DesktopService { &self, request: StartTaskRequest, ) -> TinyBusResult> { - self.on_tasks(move |tasks| tasks.start(&request)).await + self.start_remembering(request).await } /// Waits until a task needs something, finishes, or the timeout passes. @@ -157,7 +160,9 @@ impl DesktopService { &self, request: AwaitTaskRequest, ) -> TinyBusResult> { - Ok(self.tasks.await_task(request).await) + let reply = self.tasks.await_task(request).await; + self.learn_from(&reply).await; + Ok(reply) } /// Answers what a paused task asked for, and resumes it. @@ -166,13 +171,23 @@ impl DesktopService { &self, request: ContinueTaskRequest, ) -> TinyBusResult> { - self.on_tasks(move |tasks| tasks.continue_task(request)) - .await + if let Some(link) = &self.site_link { + link.note_inputs(&request.id, &request.inputs); + } + let reply = self + .on_tasks(move |tasks| tasks.continue_task(request)) + .await?; + self.learn_from(&reply).await; + Ok(reply) } /// Stops a task. async fn cancel_task(&self, request: TaskRef) -> TinyBusResult> { - self.on_tasks(move |tasks| tasks.cancel(&request.id)).await + let reply = self + .on_tasks(move |tasks| tasks.cancel(&request.id)) + .await?; + self.learn_from(&reply).await; + Ok(reply) } /// Everything a task did. diff --git a/crates/tinycomputer/src/tinybus_module/dispatch/service.rs b/crates/tinycomputer/src/tinybus_module/dispatch/service.rs index f53fb1ba..14d033b9 100644 --- a/crates/tinycomputer/src/tinybus_module/dispatch/service.rs +++ b/crates/tinycomputer/src/tinybus_module/dispatch/service.rs @@ -5,13 +5,16 @@ use std::sync::Arc; use tinybus::{Error as TinyBusError, Result as TinyBusResult}; use tinycomputer_browser::{AgentBrowser, Browser}; -use tinycomputer_bus::agent::{SurfaceAvailability, SurfaceKind}; +use tinycomputer_bus::agent::{ + AgentResponse, StartTaskRequest, SurfaceAvailability, SurfaceKind, TaskView, +}; use tinycomputer_bus::{DesktopResponse, JevConfig}; use tinycomputer_engine as agentic; use super::DesktopService; use crate::tinybus_module::config::{BrowserDefaults, cursor_config}; use crate::tinybus_module::runner::WorkspaceRunner; +use crate::tinybus_module::site_memory::{SiteLink, SiteMemoryPort, refused_flow}; use crate::{Desktop, Result}; impl DesktopService { @@ -87,6 +90,7 @@ impl DesktopService { browser, browser_defaults, accessibility: Arc::new(super::accessibility::Access::default()), + site_link: None, }) } @@ -200,3 +204,54 @@ impl DesktopService { self } } + +impl DesktopService { + /// Keeps what browser tasks learn about a site through `port`, the host's + /// `SiteMemoryHost`. + pub(crate) fn with_site_memory(mut self, port: Box) -> Self { + self.site_link = Some(Arc::new(SiteLink::new(port))); + self + } + + /// Starts a task with what earlier tasks on its `site` left, and follows + /// it for what it may teach. A saved plan the engine refuses is + /// forgotten and the task started again to be planned afresh. + pub(super) async fn start_remembering( + &self, + mut request: StartTaskRequest, + ) -> TinyBusResult> { + let Some(link) = self.site_link.clone() else { + return self.on_tasks(move |tasks| tasks.start(&request)).await; + }; + let mut prepared = link.prepare(&mut request).await; + let mut reply = self.start_once(request.clone()).await?; + if refused_flow(&reply) + && let Some(reusing) = prepared.as_mut().filter(|reusing| reusing.reuses_plan()) + { + link.refused(reusing).await; + request.flow = None; + reply = self.start_once(request).await?; + } + if let (Some(prepared), Some(view)) = (prepared, reply.data.as_ref()) { + link.follow(&view.id, prepared); + } + self.learn_from(&reply).await; + Ok(reply) + } + + async fn start_once( + &self, + request: StartTaskRequest, + ) -> TinyBusResult> { + self.on_tasks(move |tasks| tasks.start(&request)).await + } + + /// Learns from `reply`'s task once it has ended. + pub(super) async fn learn_from(&self, reply: &AgentResponse) { + let (Some(link), Some(view)) = (&self.site_link, reply.data.as_ref()) else { + return; + }; + let tasks = self.tasks.clone(); + link.learn(view, |id| tasks.report(id)).await; + } +} diff --git a/crates/tinycomputer/src/tinybus_module/mod.rs b/crates/tinycomputer/src/tinybus_module/mod.rs index 9921924f..0ac330e2 100644 --- a/crates/tinycomputer/src/tinybus_module/mod.rs +++ b/crates/tinycomputer/src/tinybus_module/mod.rs @@ -22,6 +22,7 @@ mod config; mod dispatch; mod runner; +mod site_memory; use serde_json::Value; use tinybus::{Connection, Result as TinyBusResult}; @@ -38,6 +39,9 @@ pub(crate) use dispatch::DesktopService; async fn setup(connection: Connection, config: Value) -> TinyBusResult<()> { let service = DesktopService::from_config(&config) .map_err(|error| tinybus::Error::failed(error.to_string()))?; + // The host's per-site memory; a host that does not serve it only costs + // tasks what earlier ones learned. + let service = service.with_site_memory(Box::new(site_memory::BusPort::new(connection.clone()))); #[cfg(test)] let service = service.with_native_fixture(); service.sweep_outputs(); @@ -82,7 +86,7 @@ tinybus_module::module_export_optional_static! { ], signals = [], requires = [], - optional = [], + optional = ["ai.tinyhumans.tinycomputer.SiteMemoryHost"], lazy = false, } diff --git a/crates/tinycomputer/src/tinybus_module/site_memory/mod.rs b/crates/tinycomputer/src/tinybus_module/site_memory/mod.rs new file mode 100644 index 00000000..565d8afe --- /dev/null +++ b/crates/tinycomputer/src/tinybus_module/site_memory/mod.rs @@ -0,0 +1,300 @@ +//! What a browser task learns about its site, kept through the host's +//! `SiteMemoryHost` interface. +//! +//! The module keeps nothing between tasks itself. A `StartTask` that names a +//! `site` asks the host for what finished tasks there left (`Load`), hands the +//! task a saved plan for the same goal and facts and the elements earlier +//! tasks found, and follows the task. Once it ends, the module reads the +//! report, applies the rules in [`tinycomputer_core::sites`] and gives the +//! result back (`Save`). The host decides only where the bytes live. +//! +//! A host that does not serve the interface, or answers an error, costs the +//! task nothing but the memory: it runs as if the site were new. + +use std::collections::{BTreeMap, VecDeque}; +use std::future::Future; +use std::pin::Pin; +use std::sync::Mutex; +use std::time::{SystemTime, UNIX_EPOCH}; + +use tinybus::Connection; +use tinycomputer_bus::agent::{ + AgentResponse, StartTaskRequest, TaskId, TaskReport, TaskStatus, TaskView, +}; +use tinycomputer_bus::site_memory::{ + INTERFACE, LoadSiteRequest, LoadSiteResponse, OBJECT_PATH, SaveSiteRequest, SaveSiteResponse, + SiteMemory, methods, +}; +use tinycomputer_core::sites::{self, Following, SiteMemoryRules}; + +/// How many started tasks are followed at once; the oldest is let go first. +const FOLLOWED: usize = 64; +/// The deadline for one call to the host. +const CALL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); + +/// A pending call to the host. +pub(crate) type PortFuture<'a, T> = Pin> + Send + 'a>>; + +/// The host's `SiteMemoryHost`, as the module calls it. +pub(crate) trait SiteMemoryPort: Send + Sync + std::fmt::Debug { + /// `Load`: one site's memory. + fn load<'a>(&'a self, site: &'a str) -> PortFuture<'a, LoadSiteResponse>; + /// `Save`: replaces one site's memory. + fn save(&self, request: SaveSiteRequest) -> PortFuture<'_, SaveSiteResponse>; +} + +/// The port over the bus connection the module was set up on. +#[derive(Clone)] +pub(crate) struct BusPort { + connection: Connection, +} + +impl std::fmt::Debug for BusPort { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.debug_struct("BusPort").finish_non_exhaustive() + } +} + +impl BusPort { + pub(crate) fn new(connection: Connection) -> Self { + Self { connection } + } + + fn proxy(&self) -> Result { + self.connection + .proxy(INTERFACE, OBJECT_PATH, INTERFACE) + .map(|proxy| proxy.with_timeout(CALL_TIMEOUT)) + .map_err(|error| error.to_string()) + } +} + +impl SiteMemoryPort for BusPort { + fn load<'a>(&'a self, site: &'a str) -> PortFuture<'a, LoadSiteResponse> { + Box::pin(async move { + let request = LoadSiteRequest { + site: site.to_owned(), + }; + self.proxy()? + .call(methods::LOAD, (request,)) + .await + .map_err(|error| error.to_string()) + }) + } + + fn save(&self, request: SaveSiteRequest) -> PortFuture<'_, SaveSiteResponse> { + Box::pin(async move { + self.proxy()? + .call(methods::SAVE, (request,)) + .await + .map_err(|error| error.to_string()) + }) + } +} + +/// A task about to start, with the memory it was handed. +#[derive(Debug)] +pub(crate) struct Prepared { + following: Following, + memory: SiteMemory, + epoch: u64, +} + +impl Prepared { + /// Whether the task runs a plan saved for its goal and facts. + pub(crate) fn reuses_plan(&self) -> bool { + self.following.reuses_plan() + } +} + +/// A followed task: what it may teach, and the epoch it started under. +#[derive(Debug)] +struct Followed { + following: Following, + epoch: u64, +} + +/// The module's link to the host's site memory, and the tasks it follows. +#[derive(Debug)] +pub(crate) struct SiteLink { + port: Box, + followed: Mutex>, +} + +fn now() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map_or(0, |elapsed| elapsed.as_secs()) +} + +impl SiteLink { + pub(crate) fn new(port: Box) -> Self { + Self { + port, + followed: Mutex::new(VecDeque::new()), + } + } + + fn followed(&self) -> std::sync::MutexGuard<'_, VecDeque<(TaskId, Followed)>> { + self.followed + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner) + } + + /// Fills `request` from what finished tasks on its `site` left: the plan + /// saved for its goal and facts, unless the caller brought a flow of its + /// own, and the elements found there. `None` when the request names no + /// site or goal, or the host has nothing to say. + pub(crate) async fn prepare(&self, request: &mut StartTaskRequest) -> Option { + let site = sites::site_named(request.site.as_deref()?)?; + let goal = request.task.as_deref()?; + let Ok(loaded) = self.port.load(&site).await else { + return None; + }; + let mut memory = loaded.memory; + memory.expire(now()); + let mut following = Following::new(site, goal, &request.facts, false); + if request.flow.is_none() + && let Some(plan) = memory.plan_for(following.goal(), following.facts_id()) + { + request.flow = Some(plan.clone()); + following = Following::new(following.site().to_owned(), goal, &request.facts, true); + } + if request.memory.is_empty() { + request.memory = memory + .hints + .iter() + .map(|saved| saved.hint.clone()) + .collect(); + } + Some(Prepared { + following, + memory, + epoch: loaded.epoch, + }) + } + + /// Forgets the saved plan the module refused, so the task is planned + /// afresh. When it cannot be saved, the task still counts as reusing it, + /// and a failure forgets it then. + pub(crate) async fn refused(&self, prepared: &mut Prepared) { + prepared.memory.forget_plan(&prepared.following); + if self + .save(&prepared.following, &mut prepared.memory, prepared.epoch) + .await + { + prepared.following.plan_refused(); + } + } + + /// Follows task `id` until it ends, to learn from how it did. + pub(crate) fn follow(&self, id: &TaskId, prepared: Prepared) { + let mut followed = self.followed(); + followed.retain(|(known, _)| known != id); + followed.push_back(( + id.clone(), + Followed { + following: prepared.following, + epoch: prepared.epoch, + }, + )); + while followed.len() > FOLLOWED { + followed.pop_front(); + } + } + + /// Adds `inputs`, the values a paused task `id` is given, to the values + /// it may not keep. + pub(crate) fn note_inputs(&self, id: &TaskId, inputs: &BTreeMap) { + if inputs.is_empty() { + return; + } + if let Some((_, followed)) = self.followed().iter_mut().find(|(known, _)| known == id) { + followed.following.note_inputs(inputs.values()); + } + } + + /// Learns from a followed task that has ended: what it found and the plan + /// it ran when it finished, or that the plan it reused failed. `report` + /// fetches the task's record. A task still running or paused stays + /// followed. + pub(crate) async fn learn( + &self, + view: &TaskView, + report: impl FnOnce(&TaskId) -> AgentResponse, + ) { + match &view.status { + TaskStatus::Done { .. } | TaskStatus::Checkpoint { .. } => { + let Some(followed) = self.stop(&view.id) else { + return; + }; + let Some(report) = report(&view.id).data else { + return; + }; + self.update(&followed, |memory, following| { + memory.learn(following, &report, now()); + }) + .await; + } + TaskStatus::Failed { .. } => { + let Some(followed) = self.stop(&view.id) else { + return; + }; + if followed.following.reuses_plan() { + self.update(&followed, SiteMemory::forget_plan).await; + } + } + TaskStatus::Cancelled | TaskStatus::NeedsPlan { .. } => { + self.stop(&view.id); + } + _ => {} + } + } + + fn stop(&self, id: &TaskId) -> Option { + let mut followed = self.followed(); + let index = followed.iter().position(|(known, _)| known == id)?; + followed.remove(index).map(|(_, followed)| followed) + } + + /// Loads the site's current memory, applies `change`, and saves it, unless + /// the site was forgotten since the task started. + async fn update(&self, followed: &Followed, change: impl FnOnce(&mut SiteMemory, &Following)) { + let Ok(loaded) = self.port.load(followed.following.site()).await else { + return; + }; + if loaded.epoch != followed.epoch { + return; + } + let mut memory = loaded.memory; + memory.expire(now()); + change(&mut memory, &followed.following); + self.save(&followed.following, &mut memory, followed.epoch) + .await; + } + + async fn save(&self, following: &Following, memory: &mut SiteMemory, epoch: u64) -> bool { + if memory.fit().is_err() { + return false; + } + let request = SaveSiteRequest { + site: following.site().to_owned(), + memory: std::mem::take(memory), + epoch, + }; + let kept = request.memory.clone(); + let saved = self.port.save(request).await.is_ok_and(|reply| reply.saved); + *memory = kept; + saved + } +} + +/// Whether `reply` is the module refusing the flow a task was handed. +pub(crate) fn refused_flow(reply: &AgentResponse) -> bool { + reply + .error + .as_ref() + .is_some_and(|error| error.code == "INVALID_FLOW") +} + +#[cfg(test)] +mod site_memory_tests; diff --git a/crates/tinycomputer/src/tinybus_module/site_memory/site_memory_tests.rs b/crates/tinycomputer/src/tinybus_module/site_memory/site_memory_tests.rs new file mode 100644 index 00000000..b383b5a7 --- /dev/null +++ b/crates/tinycomputer/src/tinybus_module/site_memory/site_memory_tests.rs @@ -0,0 +1,377 @@ +//! A task that names a site is handed what earlier tasks there left, through +//! the host's `SiteMemoryHost`, and leaves what it learned the same way. + +#![allow( + clippy::unwrap_used, + clippy::expect_used, + clippy::panic, + clippy::needless_pass_by_value +)] + +use std::sync::{Arc, Mutex}; + +use serde_json::{Value, json}; + +use tinybus::Connection; + +use super::*; + +/// A host that keeps sites in memory. +#[derive(Debug, Default)] +struct Store { + sites: Mutex>, + epoch: Mutex, + refuse_saves: Mutex, + unreachable: Mutex, + saves: Mutex, +} + +#[derive(Debug, Clone, Default)] +struct FakeHost(Arc); + +impl FakeHost { + fn memory(&self, site: &str) -> SiteMemory { + self.0 + .sites + .lock() + .unwrap() + .get(site) + .cloned() + .unwrap_or_default() + } + + fn forget(&self, site: &str) { + self.0.sites.lock().unwrap().remove(site); + *self.0.epoch.lock().unwrap() += 1; + } +} + +impl SiteMemoryPort for FakeHost { + fn load<'a>(&'a self, site: &'a str) -> PortFuture<'a, LoadSiteResponse> { + Box::pin(async move { + if *self.0.unreachable.lock().unwrap() { + return Err("no such interface".to_owned()); + } + Ok(LoadSiteResponse { + memory: self.memory(site), + epoch: *self.0.epoch.lock().unwrap(), + }) + }) + } + + fn save(&self, request: SaveSiteRequest) -> PortFuture<'_, SaveSiteResponse> { + Box::pin(async move { + *self.0.saves.lock().unwrap() += 1; + let stale = request.epoch != *self.0.epoch.lock().unwrap(); + if *self.0.refuse_saves.lock().unwrap() || stale { + return Ok(SaveSiteResponse { saved: false }); + } + self.0 + .sites + .lock() + .unwrap() + .insert(request.site, request.memory); + Ok(SaveSiteResponse { saved: true }) + }) + } +} + +fn link(host: &FakeHost) -> SiteLink { + SiteLink::new(Box::new(host.clone())) +} + +fn request(goal: &str) -> StartTaskRequest { + StartTaskRequest { + task: Some(goal.to_owned()), + site: Some("shop.test".to_owned()), + facts: BTreeMap::from([("name".into(), "Asha Raina".into())]), + ..StartTaskRequest::default() + } +} + +fn view(id: &str, status: Value) -> TaskView { + serde_json::from_value(json!({ + "id": id, "status": status, "summary": "", "progress": 1.0, "next": [] + })) + .unwrap() +} + +fn done(id: &str) -> TaskView { + view( + id, + json!({"state": "done", "answer": "ordered", "records": {}}), + ) +} + +fn report_of(id: &str, learned: Value, rescues: Value) -> AgentResponse { + AgentResponse::ok( + serde_json::from_value(json!({ + "view": {"id": id, "status": {"state": "done", "answer": "ordered", "records": {}}, + "summary": "", "progress": 1.0, "next": []}, + "flow": {"app": "browser", "steps": ["open the shop"]}, + "steps": [], "records": {}, "artifacts": [], + "learned": learned, "trace": [], "rescues": rescues + })) + .unwrap(), + ) +} + +fn hint(key: &str, name: &str) -> Value { + json!({"app": "browser", "key": key, "role": "button", "name": name}) +} + +async fn run_to_the_end( + link: &SiteLink, + mut request: StartTaskRequest, + id: &str, + report: AgentResponse, +) { + let prepared = link.prepare(&mut request).await.unwrap(); + link.follow(&TaskId::new(id), prepared); + link.learn(&done(id), |_| report).await; +} + +#[tokio::test] +async fn a_finished_task_leaves_its_plan_and_elements_for_the_next() { + let host = FakeHost::default(); + let link = link(&host); + run_to_the_end( + &link, + request("order milk"), + "t1", + report_of("t1", json!([hint("k1", "Add to cart")]), json!([])), + ) + .await; + let kept = host.memory("shop.test"); + assert_eq!(kept.plans.len(), 1); + assert_eq!(kept.hints.len(), 1); + + let mut next = request("Order MILK"); + let prepared = link.prepare(&mut next).await.unwrap(); + assert!(prepared.reuses_plan()); + assert!(next.flow.is_some()); + assert_eq!(next.memory.len(), 1); +} + +#[tokio::test] +async fn a_task_without_a_site_or_a_goal_reads_and_writes_nothing() { + let host = FakeHost::default(); + let link = link(&host); + let mut no_site = request("order milk"); + no_site.site = None; + assert!(link.prepare(&mut no_site).await.is_none()); + let mut no_goal = request("order milk"); + no_goal.task = None; + assert!(link.prepare(&mut no_goal).await.is_none()); + let mut odd_site = request("order milk"); + odd_site.site = Some("../etc".to_owned()); + assert!(link.prepare(&mut odd_site).await.is_none()); + assert_eq!(*host.0.saves.lock().unwrap(), 0); +} + +#[tokio::test] +async fn a_host_that_does_not_serve_the_interface_costs_the_task_only_the_memory() { + let host = FakeHost::default(); + *host.0.unreachable.lock().unwrap() = true; + let mut request = request("order milk"); + assert!(link(&host).prepare(&mut request).await.is_none()); + assert!(request.flow.is_none()); +} + +#[tokio::test] +async fn a_caller_flow_is_kept_over_a_saved_plan() { + let host = FakeHost::default(); + let link = link(&host); + run_to_the_end( + &link, + request("order milk"), + "t1", + report_of("t1", json!([]), json!([])), + ) + .await; + let mut own = request("order milk"); + let flow: tinycomputer_bus::Flow = + serde_json::from_value(json!({"app": "browser", "steps": ["my own"]})).unwrap(); + own.flow = Some(flow.clone()); + let prepared = link.prepare(&mut own).await.unwrap(); + assert!(!prepared.reuses_plan()); + assert_eq!(own.flow, Some(flow)); +} + +#[tokio::test] +async fn a_site_forgotten_while_a_task_runs_is_not_written_back() { + let host = FakeHost::default(); + let link = link(&host); + let mut start = request("order milk"); + let prepared = link.prepare(&mut start).await.unwrap(); + link.follow(&TaskId::new("t1"), prepared); + host.forget("shop.test"); + let report = report_of("t1", json!([hint("k1", "Add to cart")]), json!([])); + link.learn(&done("t1"), |_| report).await; + assert!(host.memory("shop.test").is_empty()); +} + +#[tokio::test] +async fn a_save_the_host_refuses_leaves_the_site_as_it_was() { + let host = FakeHost::default(); + *host.0.refuse_saves.lock().unwrap() = true; + run_to_the_end( + &link(&host), + request("order milk"), + "t1", + report_of("t1", json!([]), json!([])), + ) + .await; + assert!(host.memory("shop.test").is_empty()); +} + +#[tokio::test] +async fn a_reused_plan_that_fails_is_forgotten_and_a_running_task_stays_followed() { + let host = FakeHost::default(); + let link = link(&host); + run_to_the_end( + &link, + request("order milk"), + "t1", + report_of("t1", json!([]), json!([])), + ) + .await; + assert_eq!(host.memory("shop.test").plans.len(), 1); + + let mut again = request("order milk"); + let prepared = link.prepare(&mut again).await.unwrap(); + link.follow(&TaskId::new("t2"), prepared); + link.learn(&view("t2", json!({"state": "running"})), |_| { + panic!("no report for a running task") + }) + .await; + assert_eq!(host.memory("shop.test").plans.len(), 1); + link.learn( + &view("t2", json!({"state": "failed", "step": null, "reason": "gone", "hint": "", "recoverable": false})), + |_| panic!("no report for a failed task"), + ) + .await; + assert!(host.memory("shop.test").plans.is_empty()); +} + +#[tokio::test] +async fn a_refused_plan_is_forgotten_so_the_task_is_planned_afresh() { + let host = FakeHost::default(); + let link = link(&host); + run_to_the_end( + &link, + request("order milk"), + "t1", + report_of("t1", json!([]), json!([])), + ) + .await; + let mut again = request("order milk"); + let mut prepared = link.prepare(&mut again).await.unwrap(); + assert!(prepared.reuses_plan()); + link.refused(&mut prepared).await; + assert!(!prepared.reuses_plan()); + assert!(host.memory("shop.test").plans.is_empty()); +} + +#[tokio::test] +async fn values_given_to_a_paused_task_are_not_kept() { + let host = FakeHost::default(); + let link = link(&host); + let mut start = request("order milk"); + let prepared = link.prepare(&mut start).await.unwrap(); + let id = TaskId::new("t1"); + link.follow(&id, prepared); + link.note_inputs(&id, &BTreeMap::from([("code".into(), "Zanzibar77".into())])); + let report = report_of("t1", json!([hint("k1", "Zanzibar77 button")]), json!([])); + link.learn(&done("t1"), |_| report).await; + assert!(host.memory("shop.test").hints.is_empty()); +} + +#[tokio::test] +async fn a_cancelled_task_stops_being_followed_and_teaches_nothing() { + let host = FakeHost::default(); + let link = link(&host); + let mut start = request("order milk"); + let prepared = link.prepare(&mut start).await.unwrap(); + link.follow(&TaskId::new("t1"), prepared); + link.learn(&view("t1", json!({"state": "cancelled"})), |_| { + panic!("no report") + }) + .await; + link.learn(&done("t1"), |_| panic!("no longer followed")) + .await; + assert_eq!(*host.0.saves.lock().unwrap(), 0); +} + +#[test] +fn only_the_engines_invalid_flow_error_counts_as_a_refused_flow() { + let refused: AgentResponse = AgentResponse::err( + tinycomputer_bus::agent::AgentError::new("INVALID_FLOW", "no", "fix", false), + ); + assert!(refused_flow(&refused)); + let other: AgentResponse = AgentResponse::err( + tinycomputer_bus::agent::AgentError::new("NO_SUCH_TASK", "INVALID_FLOW", "x", false), + ); + assert!(!refused_flow(&other)); +} + +/// A host serving `SiteMemoryHost` over a real bus. +#[derive(Clone)] +struct BusHost(FakeHost); + +#[tinybus::interface(name = "ai.tinyhumans.tinycomputer.SiteMemoryHost")] +impl BusHost { + async fn load(&self, request: LoadSiteRequest) -> tinybus::Result { + self.0 + .load(&request.site) + .await + .map_err(tinybus::Error::failed) + } + + async fn save(&self, request: SaveSiteRequest) -> tinybus::Result { + self.0.save(request).await.map_err(tinybus::Error::failed) + } +} + +#[tokio::test] +async fn the_bus_port_loads_and_saves_through_the_hosts_interface() -> tinybus::Result<()> { + use tinybus::broker::Broker; + use tinybus::transport::memory::MemoryBus; + + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let host_connection = Connection::connect(bus.connect().await?).await?; + let fake = FakeHost::default(); + host_connection + .serve_at(OBJECT_PATH.try_into()?, BusHost(fake.clone())) + .await?; + host_connection.request_name(INTERFACE).await?; + let module = Connection::connect(bus.connect().await?).await?; + + let link = SiteLink::new(Box::new(BusPort::new(module))); + run_to_the_end( + &link, + request("order milk"), + "t1", + report_of("t1", json!([hint("k1", "Add to cart")]), json!([])), + ) + .await; + assert_eq!(fake.memory("shop.test").plans.len(), 1); + let mut next = request("order milk"); + assert!(link.prepare(&mut next).await.unwrap().reuses_plan()); + Ok(()) +} + +#[tokio::test] +async fn the_bus_port_reports_a_host_that_serves_nothing() -> tinybus::Result<()> { + use tinybus::broker::Broker; + use tinybus::transport::memory::MemoryBus; + + let bus = MemoryBus::new(); + Broker::new().spawn(bus.clone()); + let module = Connection::connect(bus.connect().await?).await?; + let link = SiteLink::new(Box::new(BusPort::new(module))); + let mut start = request("order milk"); + assert!(link.prepare(&mut start).await.is_none()); + Ok(()) +} diff --git a/crates/tinycomputer/tests/public_api_tests.rs b/crates/tinycomputer/tests/public_api_tests.rs index 29f4c689..1a377bc2 100644 --- a/crates/tinycomputer/tests/public_api_tests.rs +++ b/crates/tinycomputer/tests/public_api_tests.rs @@ -34,7 +34,7 @@ fn the_crate_re_exports_the_contract_rather_than_redefining_it() { #[test] fn the_shipped_contract_binds_to_itself() { assert!(is_compatible(CONTRACT_VERSION)); - assert_eq!(CONTRACT_VERSION, (2, 16)); + assert_eq!(CONTRACT_VERSION, (2, 17)); } #[test] diff --git a/docs/crates/tinycomputer-bus/versioning.md b/docs/crates/tinycomputer-bus/versioning.md index 5d5291f2..d1727119 100644 --- a/docs/crates/tinycomputer-bus/versioning.md +++ b/docs/crates/tinycomputer-bus/versioning.md @@ -9,7 +9,7 @@ code, so this page exists mostly to keep them apart. ## `CONTRACT_VERSION`: does this host understand this module ```rust,ignore -pub const CONTRACT_VERSION: (u32, u32) = (2, 16); +pub const CONTRACT_VERSION: (u32, u32) = (2, 17); ``` This describes the *vocabulary*: the member set and the payload shapes, @@ -45,15 +45,15 @@ Two conditions, both necessary: relies on changed shape or disappeared, so a mismatched major is refused outright, in either direction. 2. **The module must be at least as new, minor-wise, as the host.** A host - built against `(2, 16)` expects at least the members and fields that - existed at `(2, 16)`. A module reporting `(2, 11)` might be missing one of - them, so it is rejected. A module reporting `(2, 16)` has everything the + built against `(2, 17)` expects at least the members and fields that + existed at `(2, 17)`. A module reporting `(2, 11)` might be missing one of + them, so it is rejected. A module reporting `(2, 17)` has everything the host expects, plus something newer the host simply does not use yet, so it is accepted. ```rust,ignore -assert!(is_compatible(CONTRACT_VERSION)); // (2, 16), the exact version this crate ships -assert!(is_compatible((2, 16))); // a newer, still-compatible module +assert!(is_compatible(CONTRACT_VERSION)); // (2, 17), the exact version this crate ships +assert!(is_compatible((2, 17))); // a newer, still-compatible module assert!(!is_compatible((2, 11))); // an older module, missing something assert!(!is_compatible((1, 10))); // a different major, all bets off ``` From d74b9ce72cbfc1a8a9770c53fe53281857078ed2 Mon Sep 17 00:00:00 2001 From: Steven Enamakel Date: Tue, 13 Oct 2026 15:38:20 +0300 Subject: [PATCH 2/2] test: satisfy assert_is_empty lint Co-authored-by: Medulla --- .../src/tinybus_module/site_memory/site_memory_tests.rs | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/crates/tinycomputer/src/tinybus_module/site_memory/site_memory_tests.rs b/crates/tinycomputer/src/tinybus_module/site_memory/site_memory_tests.rs index b383b5a7..d81ec574 100644 --- a/crates/tinycomputer/src/tinybus_module/site_memory/site_memory_tests.rs +++ b/crates/tinycomputer/src/tinybus_module/site_memory/site_memory_tests.rs @@ -251,7 +251,7 @@ async fn a_reused_plan_that_fails_is_forgotten_and_a_running_task_stays_followed |_| panic!("no report for a failed task"), ) .await; - assert!(host.memory("shop.test").plans.is_empty()); + assert_eq!(host.memory("shop.test").plans.len(), 0); } #[tokio::test] @@ -270,7 +270,7 @@ async fn a_refused_plan_is_forgotten_so_the_task_is_planned_afresh() { assert!(prepared.reuses_plan()); link.refused(&mut prepared).await; assert!(!prepared.reuses_plan()); - assert!(host.memory("shop.test").plans.is_empty()); + assert_eq!(host.memory("shop.test").plans.len(), 0); } #[tokio::test] @@ -284,7 +284,7 @@ async fn values_given_to_a_paused_task_are_not_kept() { link.note_inputs(&id, &BTreeMap::from([("code".into(), "Zanzibar77".into())])); let report = report_of("t1", json!([hint("k1", "Zanzibar77 button")]), json!([])); link.learn(&done("t1"), |_| report).await; - assert!(host.memory("shop.test").hints.is_empty()); + assert_eq!(host.memory("shop.test").hints.len(), 0); } #[tokio::test]