diff --git a/crates/adaptive/examples/service.rs b/crates/adaptive/examples/service.rs index a2ac2c3..5c57a58 100644 --- a/crates/adaptive/examples/service.rs +++ b/crates/adaptive/examples/service.rs @@ -20,9 +20,16 @@ //! unique wire id, sends the frame, and awaits the report with a deadline — //! the exact shape a Socket.IO handler pair implements; //! 3. the device side: one call to [`serve`] between deserialize and reply; -//! 4. the success gate: the learned workflow reaches the vault only because +//! 4. the STORAGE relay — the device-master arrangement over a wire. The +//! device's own store is the durable workflow home; the service holds no +//! workflow durably. [`DeviceVault`] is a stateless adapter implementing +//! [`Vault`]: `load` fetches the device's catalogue at episode start, +//! and the success-gated flush relays each kept record back as a put the +//! device writes into its store — where its own surfaces list it; +//! 5. the success gate: the learned workflow reaches the DEVICE only because //! the goal run satisfied; -//! 5. the second goal run selecting what the first one learned. +//! 6. the second goal run selecting what the first one learned — read back +//! from the device, not from anything the service kept. use std::collections::HashMap; use std::sync::atomic::{AtomicU64, Ordering}; @@ -34,7 +41,7 @@ use serde_json::{Value, json}; use tinyflows::caps::mock::mock_capabilities; use tinyflows::caps::{Capabilities, LlmProvider}; use tinyflows::error::Result as EngineResult; -use tinyflows::model::{Edge, InputType, Node, NodeKind, WorkflowGraph, WorkflowInput}; +use tinyflows::store::types::{WorkflowError, WorkflowRecord}; use tinyflows::store::{HostPolicy, WorkflowStore}; use tinyflows_adaptive::contracts::{Budget, Goal}; use tinyflows_adaptive::driver::{Clock, Loop}; @@ -43,8 +50,8 @@ use tinyflows_adaptive::host::HostFacts; use tinyflows_adaptive::inventory; use tinyflows_adaptive::ledger::memory::MemoryLedger; use tinyflows_adaptive::ledger::{EpisodeStatus, Ledger}; -use tinyflows_adaptive::workflows::Snapshot; use tinyflows_adaptive::workflows::memory::MemoryVault; +use tinyflows_adaptive::workflows::{Snapshot, Vault}; use tokio::sync::{mpsc, oneshot}; // --------------------------------------------------------------------------- @@ -212,10 +219,20 @@ impl LlmProvider for TierRouter { "inputs": { "repo": "acme/rust-lib" }, }) } + // The recipe surface: steps in, never graph syntax — the + // lowering writes the graph. The ask names no concrete repo; + // the declared input carries it, which is what lets `keep` file + // the plan for the next repository. "author" => json!({ - "graph": review_graph(), "why": "nothing stored fits yet", + "declared": [ + { "name": "repo", "description": "the repository to review", "required": true } + ], "inputs": { "repo": "acme/thing" }, + "steps": [{ + "id": "review", + "ask": "Review the open pull requests and post a summary." + }], }), "judge" => json!({ "satisfied": true, "gap": "" }), "generalise" => json!({ @@ -230,43 +247,225 @@ impl LlmProvider for TierRouter { } } -/// The graph the "author" writes: parameterised, which is what lets `keep` -/// file it — the repo arrives through a declared input, never pasted in. -fn review_graph() -> Value { - serde_json::to_value(WorkflowGraph { - schema_version: 1, - id: None, - name: "review-prs".into(), - inputs: vec![WorkflowInput::new("repo", InputType::String).required()], - agents: Vec::new(), - nodes: vec![ - Node { - id: "start".into(), - kind: NodeKind::Trigger, - type_version: 1, - name: "manual".into(), - config: json!({ "trigger_kind": "manual" }), - ports: Vec::new(), - position: None, - }, - Node { - id: "report".into(), - kind: NodeKind::Transform, - type_version: 1, - name: "report".into(), - config: json!({ "set": { "target": "=run.inputs.repo" } }), - ports: Vec::new(), - position: None, - }, - ], - edges: vec![Edge { - from_node: "start".into(), - from_port: "main".into(), - to_node: "report".into(), - to_port: "main".into(), - }], - }) - .expect("a graph serializes") +// --------------------------------------------------------------------------- +// The storage relay — device-master over a wire. +// --------------------------------------------------------------------------- + +/// One message of the vault wire, either direction. +/// +/// The service sends `Load` / `Put` / `Remove`; the device answers `Records` +/// or `Ack` under the echoed wire id. Same discipline as the run relay: ids +/// are minted service-side, so a late reply can never resolve the wrong +/// waiter. +#[derive(serde::Serialize, serde::Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +enum VaultFrame { + /// Service → device: send me your whole catalogue. + Load { wire_id: String }, + /// Device → service: the catalogue. + Records { + wire_id: String, + records: Vec, + }, + /// Service → device: write this record into YOUR store. + /// + /// Boxed: a record dwarfs every other variant, and the frame travels + /// through queues sized for the small ones. + Put { + wire_id: String, + record: Box, + }, + /// Service → device: remove this id from your store. + Remove { wire_id: String, id: String }, + /// Device → service: the write or removal settled. + Ack { + wire_id: String, + error: Option, + }, +} + +/// What a vault waiter resolves to. +enum VaultReply { + Records(Vec), + Ack(Option), +} + +/// The service's view of the DEVICE's workflow store. +/// +/// Stateless on purpose: it holds a sender, a deadline and a counter — +/// never a record. During an episode the [`Snapshot`] is the only +/// service-side copy, and it is memory; when the success gate flushes, each +/// kept workflow crosses this wire and comes to rest in the device's own +/// store, which is the one durable home the design allows it. +struct DeviceVault { + to_device: mpsc::Sender, + waiting: Mutex>>, + sequence: AtomicU64, + deadline: Duration, +} + +impl DeviceVault { + fn new(to_device: mpsc::Sender, deadline: Duration) -> Arc { + Arc::new(Self { + to_device, + waiting: Mutex::new(HashMap::new()), + sequence: AtomicU64::new(0), + deadline, + }) + } + + /// HOST: the body of your `socket.on("tinyflows:vault_reply", …)` handler. + fn deliver(&self, frame: &str) { + let Ok(frame) = serde_json::from_str::(frame) else { + eprintln!(" ! dropped an unparseable vault frame"); + return; + }; + let (wire_id, reply) = match frame { + VaultFrame::Records { wire_id, records } => (wire_id, VaultReply::Records(records)), + VaultFrame::Ack { wire_id, error } => (wire_id, VaultReply::Ack(error)), + // Requests only travel the other way. + _ => return, + }; + if let Some(tx) = self.waiting.lock().expect("vault waiters").remove(&wire_id) { + let _ = tx.send(reply); + } else { + eprintln!(" ! late or unknown vault reply `{wire_id}`"); + } + } + + async fn exchange(&self, mut frame: VaultFrame) -> Result { + let wire_id = format!("vault#{}", self.sequence.fetch_add(1, Ordering::Relaxed)); + match &mut frame { + VaultFrame::Load { wire_id: id } + | VaultFrame::Put { wire_id: id, .. } + | VaultFrame::Remove { wire_id: id, .. } => *id = wire_id.clone(), + _ => unreachable!("the service only sends requests"), + } + let payload = serde_json::to_string(&frame) + .map_err(|e| WorkflowError::Engine(format!("vault frame: {e}")))?; + + let (tx, rx) = oneshot::channel(); + self.waiting + .lock() + .expect("vault waiters") + .insert(wire_id.clone(), tx); + + if self.to_device.send(payload).await.is_err() { + self.waiting.lock().expect("vault waiters").remove(&wire_id); + return Err(WorkflowError::Engine("no device connected".to_string())); + } + match tokio::time::timeout(self.deadline, rx).await { + Ok(Ok(reply)) => Ok(reply), + Ok(Err(_)) => Err(WorkflowError::Engine( + "the delivery side dropped the vault waiter".to_string(), + )), + Err(_) => { + self.waiting.lock().expect("vault waiters").remove(&wire_id); + // Said with the consequence: a flush that cannot reach the + // device is a REPORTED failure — the learnings ledger is + // untouched, and nothing pretends the workflow landed. + Err(WorkflowError::Engine(format!( + "the device did not answer within {:?} — the record was NOT stored", + self.deadline + ))) + } + } + } +} + +#[async_trait] +impl Vault for DeviceVault { + async fn load(&self) -> Result, WorkflowError> { + match self + .exchange(VaultFrame::Load { + wire_id: String::new(), + }) + .await? + { + VaultReply::Records(records) => Ok(records), + VaultReply::Ack(_) => Err(WorkflowError::Engine( + "the device answered a load with an ack".to_string(), + )), + } + } + + async fn put(&self, record: &WorkflowRecord) -> Result<(), WorkflowError> { + println!(" → VaultPut {}", record.id); + match self + .exchange(VaultFrame::Put { + wire_id: String::new(), + record: Box::new(record.clone()), + }) + .await? + { + VaultReply::Ack(None) => Ok(()), + VaultReply::Ack(Some(error)) => Err(WorkflowError::Engine(error)), + VaultReply::Records(_) => Err(WorkflowError::Engine( + "the device answered a put with records".to_string(), + )), + } + } + + async fn remove(&self, id: &str) -> Result<(), WorkflowError> { + match self + .exchange(VaultFrame::Remove { + wire_id: String::new(), + id: id.to_string(), + }) + .await? + { + VaultReply::Ack(None) => Ok(()), + VaultReply::Ack(Some(error)) => Err(WorkflowError::Engine(error)), + VaultReply::Records(_) => Err(WorkflowError::Engine( + "the device answered a remove with records".to_string(), + )), + } + } +} + +/// The device's side of the storage relay. +/// +/// The whole obligation: deserialize, apply to the DEVICE's own store, reply. +/// In production the store is the device's real tinyflows store — the one its +/// workflow surfaces already list — and this task is your socket handler. +fn spawn_device_store( + store: Arc, + mut from_server: mpsc::Receiver, + to_server: mpsc::Sender, +) { + tokio::spawn(async move { + while let Some(frame) = from_server.recv().await { + let Ok(frame) = serde_json::from_str::(&frame) else { + continue; + }; + let reply = match frame { + VaultFrame::Load { wire_id } => match store.load().await { + Ok(records) => VaultFrame::Records { wire_id, records }, + Err(e) => VaultFrame::Ack { + wire_id, + error: Some(e.to_string()), + }, + }, + VaultFrame::Put { wire_id, record } => { + println!(" ← device stored `{}` in ITS store", record.id); + VaultFrame::Ack { + wire_id, + error: store.put(&record).await.err().map(|e| e.to_string()), + } + } + VaultFrame::Remove { wire_id, id } => VaultFrame::Ack { + wire_id, + error: store.remove(&id).await.err().map(|e| e.to_string()), + }, + // Replies only travel the other way. + _ => continue, + }; + let Ok(payload) = serde_json::to_string(&reply) else { + continue; + }; + let _ = to_server.send(payload).await; + } + }); } // --------------------------------------------------------------------------- @@ -303,14 +502,17 @@ async fn main() { // HOST: MongoLedger::connect / SqliteLedger::at_default_location, a real // HTTP-backed LlmProvider, real HostFacts from the device's probe. let ledger_root = MemoryLedger::new(); - let vault_root = MemoryVault::new(); + // The DEVICE's workflow store — the one durable home for workflows in + // this whole topology. In production: the device's real tinyflows store. + let device_store = Arc::new(MemoryVault::new()); let caps = Capabilities { llm: Arc::new(TierRouter), ..mock_capabilities() }; let facts = HostFacts::unknown(); - // The wire: two channels where production has one socket. + // The wire: two channel pairs where production has one socket — one for + // runs, one for the vault. let (to_device_tx, to_device_rx) = mpsc::channel::(16); let (to_server_tx, mut to_server_rx) = mpsc::channel::(16); let relay = ChannelRelay::new(to_device_tx, Duration::from_secs(30)); @@ -324,11 +526,26 @@ async fn main() { } }); } + let (vault_tx, vault_rx) = mpsc::channel::(16); + let (vault_reply_tx, mut vault_reply_rx) = mpsc::channel::(16); + let vault = DeviceVault::new(vault_tx, Duration::from_secs(30)); + spawn_device_store(Arc::clone(&device_store), vault_rx, vault_reply_tx); + { + // HOST: and this one is your vault-reply handler. + let vault = Arc::clone(&vault); + tokio::spawn(async move { + while let Some(frame) = vault_reply_rx.recv().await { + vault.deliver(&frame); + } + }); + } // ---- tenant scope: per request, free -------------------------------- + // The ledger stays tenant-scoped on the service: learnings are the + // service's property. Workflows are the DEVICE's — the vault handle IS + // the device, so tenancy is which device you are talking to. let tenant = "user-7"; let ledger = ledger_root.for_tenant(tenant); - let vault = vault_root.for_tenant(tenant); // ---- goal run 1: a cold catalogue, so the loop authors -------------- println!("── goal run 1 · cold start ──"); @@ -336,7 +553,7 @@ async fn main() { "ep-1", &Goal::new("review the open pull requests on acme/thing"), &ledger, - &vault, + vault.as_ref(), &caps, &facts, &relay, @@ -344,21 +561,31 @@ async fn main() { .await; // ---- goal run 2: the catalogue now holds what run 1 learned --------- - println!("\n── goal run 2 · the loop reuses what it learned ──"); + // Learned from the DEVICE: the service kept nothing between the runs. + println!("\n── goal run 2 · the loop reuses what the DEVICE now holds ──"); run_goal( "ep-2", &Goal::new("review the open pull requests on acme/rust-lib"), &ledger, - &vault, + vault.as_ref(), &caps, &facts, &relay, ) .await; - // ---- what is on the shelf, and what the trail says ------------------ - println!("\n── the tenant's shelf ──"); - let snapshot = Snapshot::load(&vault, permissive()).await.expect("load"); + // ---- what the DEVICE holds, and what the trail says ----------------- + // Listed from the device store directly: proof the records came to rest + // on the device, not in anything the service kept. + println!("\n── the DEVICE's store ──"); + for record in device_store.load().await.expect("device load") { + println!(" {} · {}", record.id, record.name); + } + + println!("\n── the tenant's shelf, read back over the wire ──"); + let snapshot = Snapshot::load(vault.as_ref(), permissive()) + .await + .expect("load"); let store: Arc = Arc::new(snapshot); for listing in inventory::shelf(&store, &ledger).await.expect("shelf") { println!( @@ -388,13 +615,13 @@ async fn run_goal( episode: &str, goal: &Goal, ledger: &MemoryLedger, - vault: &MemoryVault, + vault: &DeviceVault, caps: &Capabilities, facts: &HostFacts, relay: &Arc, ) { - // Fetched fresh each goal run. HOST: a Layered vault puts a device - // catalogue (read-only, degrading) in front of this one. + // Fetched fresh each goal run — FROM THE DEVICE, over the wire: the + // service holds no catalogue of its own to consult. let snapshot = Snapshot::load(vault, permissive()).await.expect("load"); let store: Arc = Arc::new(snapshot.clone()); @@ -419,10 +646,11 @@ async fn run_goal( finished.status, finished.attempts ); - // The success gate: the vault — and through it a device — only ever - // receives workflows from goal runs that succeeded. + // The success gate: the DEVICE only ever receives workflows from goal + // runs that succeeded — each pending record crosses the wire as a put + // the device writes into its own store. if finished.status == EpisodeStatus::Satisfied && snapshot.pending() > 0 { let landed = snapshot.flush(vault).await.expect("flush"); - println!(" flushed {landed} learned workflow(s) to the vault"); + println!(" flushed {landed} learned workflow(s) to the DEVICE"); } }