feat(a2ui): bind live data to worker-owned trigger types, deprecate stream bindings (MOT-3619)

a2ui/Cargo.toml

@@ -30,3 +30,6 @@ tracing-subscriber = { version = "0.3", features = ["fmt", "env-filter"] }
 clap = { version = "4", features = ["derive", "env"] }
 schemars = "0.8"
 uuid = { version = "1", features = ["v4"] }
+
+[dev-dependencies]
+async-trait = "0.1"

a2ui/README.md

@@ -38,7 +38,33 @@ Patch an existing surface with another natural-language request. Supplying `expe
 
 Interactive surfaces automatically submit their complete bound data model with button actions, so form values are persisted and forwarded to the originating Harness turn. The page also supports bounded revision history and undo, pinning, duplication, JSON import/export, and a per-session template library.
 
-`a2ui::binding::set` can bind an exact `state`, `stream`, or `shell::changed` event to a JSON Pointer in the surface data model. Browser events are deliberately excluded because Console-side registration would bypass Browser approval boundaries. Bindings are declarative and cannot invoke arbitrary functions.
+### Live bindings
+
+`a2ui::binding::set` binds a JSON Pointer in the surface data model to live data. The Console page registers the binding for its own view and stores each delivered value, so a binding is active while the surface is open. Bindings are declarative and cannot invoke arbitrary functions. A surface holds at most 32 bindings.
+
+- **Worker-owned trigger types** (recommended): `trigger_type` is any trigger type a worker registers, for example `orders::changed`. The type must be registered when the binding is set, and `config` is validated against the configuration schema the provider registered (with `trigger_request_format`). In event mode, each event payload (or `event_path` inside it) is applied. With a `query`, the event is only a notification. The page reads the provider's query when it opens the surface (after binding) and again after each notification, through the Console-only `a2ui::binding::refresh`. That covers the initial read, recovery after a reconnect and latest-state semantics. The query function must be registered by the same worker that owns the trigger type, with metadata `{"read_only": true}`, and its payload is validated against the function's request schema.
+- **`state`** (`scope`, `key`) and **`shell::changed`** (`path`) keep their exact configs.
+- Reserved types are rejected: `engine::*`, `harness::*`, `browser::*` (Console-side registration would bypass Browser approval boundaries), `a2ui::*`, `iii::*`, any `hook` type, `http`, `cron`, `queue`, `durable:subscriber`, `subscribe`, `stream:join` and `stream:leave`.
+- **`stream`** (`stream_name`, `group_id`, optional `item_id`) is deprecated legacy compatibility. It is still accepted and keeps working on engines that run iii-stream. The receipt carries a `deprecation` notice, the worker logs a warning, and the page marks the surface. Move these bindings to a worker-owned trigger type; see the guide "Migrate from iii-stream and pubsub".
+
+Example binding to a worker-owned trigger type served by the runnable provider in `examples/owned_trigger_provider.rs` (`cargo run --example owned_trigger_provider -- ws://127.0.0.1:49134`):
+
+```json
+{
+  "surface_id": "clicks",
+  "binding": {
+    "id": "clicks-count",
+    "trigger_type": "demo::counter-changed",
+    "config": { "counter": "clicks" },
+    "target_path": "/count",
+    "query": {
+      "function_id": "demo::counter::get",
+      "payload": { "counter": "clicks" },
+      "result_path": "/value"
+    }
+  }
+}
+```
 
 The A2UI page does not replace or embed itself in Shell or Browser. Its workspace action materializes a complete runnable React project under `generated/a2ui/<surface>-r<revision>` in the active Harness working directory. Shell then shows the real source files and Git diffs for editing. Run the generated Vite app in Shell and open its local URL in the Browser worker for preview. The same runnable project is available as a React app ZIP through `a2ui::surface::export-code`; JSON exports remain portable through `a2ui::surface::import`.
 

a2ui/examples/owned_trigger_provider.rs

@@ -0,0 +1,230 @@
+//! Minimal provider of a worker-owned trigger type that an A2UI live binding
+//! can target. Run it next to an engine (no iii-stream needed):
+//!
+//! ```bash
+//! cargo run --example owned_trigger_provider -- ws://127.0.0.1:49134
+//! ```
+//!
+//! Then bind a surface to it with `a2ui::binding::set`:
+//!
+//! ```json
+//! {
+//!   "surface_id": "clicks",
+//!   "binding": {
+//!     "id": "clicks-count",
+//!     "trigger_type": "demo::counter-changed",
+//!     "config": { "counter": "clicks" },
+//!     "target_path": "/count",
+//!     "query": {
+//!       "function_id": "demo::counter::get",
+//!       "payload": { "counter": "clicks" },
+//!       "result_path": "/value"
+//!     }
+//!   }
+//! }
+//! ```
+//!
+//! `demo::counter::bump { "counter": "clicks" }` commits a new value and then
+//! notifies matching bindings with `{ counter, revision }`; the A2UI page
+//! re-reads through the read-only `demo::counter::get` query.
+
+use std::collections::HashMap;
+use std::sync::{Arc, Mutex};
+
+use async_trait::async_trait;
+use iii_sdk::protocol::TriggerRequest;
+use iii_sdk::runtime::WorkerMetadata;
+use iii_sdk::trigger::{TriggerConfig, TriggerHandler};
+use iii_sdk::{
+    register_worker, Error, IIIClient, InitOptions, RegisterFunction, RegisterTriggerType,
+};
+use schemars::JsonSchema;
+use serde::{Deserialize, Serialize};
+use serde_json::{json, Value};
+
+const TRIGGER_TYPE: &str = "demo::counter-changed";
+/// Bound fan-out: bindings beyond this are rejected at registration.
+const MAX_BINDINGS: usize = 64;
+
+/// Binding configuration for `demo::counter-changed`.
+#[derive(Debug, Serialize, Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+struct CounterConfig {
+    /// Counter whose changes are delivered.
+    counter: String,
+}
+
+/// Request naming one counter.
+#[derive(Debug, Deserialize, JsonSchema)]
+struct CounterRequest {
+    /// Counter name.
+    counter: String,
+}
+
+#[derive(Debug, Deserialize, JsonSchema)]
+struct Empty {}
+
+#[derive(Default)]
+struct Provider {
+    /// counter -> (value, revision); the source of truth the query reads.
+    counters: Mutex<HashMap<String, (u64, u64)>>,
+    /// trigger id -> binding.
+    bindings: Mutex<HashMap<String, TriggerConfig>>,
+}
+
+struct Handler(Arc<Provider>);
+
+#[async_trait]
+impl TriggerHandler for Handler {
+    async fn register_trigger(&self, binding: TriggerConfig) -> Result<(), Error> {
+        // Validate the filter: throwing rejects the binding.
+        let config: CounterConfig = serde_json::from_value(binding.config.clone())
+            .map_err(|error| Error::Handler(format!("invalid {TRIGGER_TYPE} config: {error}")))?;
+        if config.counter.trim().is_empty() || config.counter.len() > 64 {
+            return Err(Error::Handler("counter must be 1-64 bytes".into()));
+        }
+        let mut bindings = self.0.bindings.lock().unwrap();
+        if bindings.len() >= MAX_BINDINGS && !bindings.contains_key(&binding.id) {
+            return Err(Error::Handler(format!("too many {TRIGGER_TYPE} bindings")));
+        }
+        bindings.insert(binding.id.clone(), binding);
+        Ok(())
+    }
+
+    async fn unregister_trigger(&self, binding: TriggerConfig) -> Result<(), Error> {
+        self.0.bindings.lock().unwrap().remove(&binding.id);
+        Ok(())
+    }
+}
+
+impl Provider {
+    fn read(&self, counter: &str) -> Value {
+        let (value, revision) = self
+            .counters
+            .lock()
+            .unwrap()
+            .get(counter)
+            .copied()
+            .unwrap_or_default();
+        json!({ "counter": counter, "value": value, "revision": revision })
+    }
+
+    /// Deliver after commit, sequentially with a timeout per binding: no task
+    /// per event and no queue for slow consumers.
+    async fn notify(&self, iii: &IIIClient, counter: &str, revision: u64) -> usize {
+        let targets: Vec<TriggerConfig> = self
+            .bindings
+            .lock()
+            .unwrap()
+            .values()
+            .filter(|binding| {
+                binding.config.get("counter").and_then(Value::as_str) == Some(counter)
+            })
+            .cloned()
+            .collect();
+        let mut delivered = 0;
+        for binding in targets {
+            let request = TriggerRequest {
+                function_id: binding.function_id.clone(),
+                payload: json!({ "counter": counter, "revision": revision }),
+                action: None,
+                timeout_ms: Some(5_000),
+            };
+            // Preserve the consumer's namespace and metadata.
+            let mut request = request.metadata(binding.metadata.clone().unwrap_or(Value::Null));
+            if let Some(namespace) = binding.namespace.clone() {
+                request = request.namespace(namespace);
+            }
+            match iii.trigger(request).await {
+                Ok(_) => delivered += 1,
+                Err(error) => {
+                    eprintln!("{TRIGGER_TYPE} delivery to {} failed: {error}", binding.id)
+                }
+            }
+        }
+        delivered
+    }
+}
+
+#[tokio::main]
+async fn main() {
+    let url = std::env::args()
+        .nth(1)
+        .or_else(|| std::env::var("III_URL").ok())
+        .unwrap_or_else(|| "ws://127.0.0.1:49134".into());
+    let iii = Arc::new(register_worker(
+        &url,
+        InitOptions {
+            metadata: Some(WorkerMetadata {
+                runtime: "rust".into(),
+                name: "a2ui-demo-provider".into(),
+                ..WorkerMetadata::default()
+            }),
+            ..InitOptions::default()
+        },
+    ));
+    let provider = Arc::new(Provider::default());
+
+    iii.register_trigger_type(
+        RegisterTriggerType::new(
+            TRIGGER_TYPE,
+            "Demo counter changed: { counter, revision } after each committed bump.",
+            Handler(provider.clone()),
+        )
+        .trigger_request_format::<CounterConfig>(),
+    );
+
+    let get = provider.clone();
+    iii.register_function(
+        "demo::counter::get",
+        RegisterFunction::new_async(move |request: CounterRequest| {
+            let provider = get.clone();
+            async move { Ok::<Value, Error>(provider.read(&request.counter)) }
+        })
+        .description("Read one demo counter: { counter, value, revision }.")
+        .metadata(json!({ "read_only": true })),
+    );
+
+    let bump = provider.clone();
+    let bump_iii = iii.clone();
+    iii.register_function(
+        "demo::counter::bump",
+        RegisterFunction::new_async(move |request: CounterRequest| {
+            let provider = bump.clone();
+            let iii = bump_iii.clone();
+            async move {
+                let revision = {
+                    let mut counters = provider.counters.lock().unwrap();
+                    let entry = counters.entry(request.counter.clone()).or_default();
+                    entry.0 += 1;
+                    entry.1 += 1;
+                    entry.1
+                };
+                let delivered = provider.notify(&iii, &request.counter, revision).await;
+                let mut current = provider.read(&request.counter);
+                current["delivered"] = json!(delivered);
+                Ok::<Value, Error>(current)
+            }
+        })
+        .description(
+            "Increment a demo counter, then notify matching demo::counter-changed bindings.",
+        ),
+    );
+
+    let count = provider.clone();
+    iii.register_function(
+        "demo::bindings::count",
+        RegisterFunction::new_async(move |_: Empty| {
+            let provider = count.clone();
+            async move {
+                Ok::<Value, Error>(json!({ "count": provider.bindings.lock().unwrap().len() }))
+            }
+        })
+        .description("Number of active demo::counter-changed bindings.")
+        .metadata(json!({ "read_only": true })),
+    );
+
+    println!("a2ui-demo-provider ready on {url}");
+    let _ = tokio::signal::ctrl_c().await;
+    iii.shutdown_async().await;
+}

a2ui/skills/SKILL.md

@@ -20,14 +20,16 @@ Surfaces belong to the current Harness session. The Harness pre-trigger hook sta
 - Export a finished surface as portable A2UI JSON for reuse outside the current conversation.
 - Reuse, undo, pin, duplicate, import, or export a surface as a runnable React app or a data-serving iii worker template.
 - Materialize the React app into the selected workspace when the person wants to inspect and edit its files in Shell, then preview it through Browser.
-- Bind surface data to exact state, stream, or Shell change events when the UI should stay live.
+- Bind surface data to a worker-owned trigger type (optionally with its provider's read-only query), exact state, or Shell change events when the UI should stay live.
 
 ## Boundaries
 
 - Do not use A2UI for plain answers that are clearer as a short message.
 - Generated surfaces use the fixed Console catalog; they cannot execute HTML, JavaScript, CSS, or arbitrary components.
 - Use `canvas` for diagrams and freeform drawing, and use a domain worker directly when no visual surface is needed.
-- `a2ui::action`, `a2ui::binding::apply`, `a2ui::stamp-session`, and `a2ui::on-config-change` are internal Console/Harness lifecycle functions, not agent tools.
+- `a2ui::action`, `a2ui::binding::apply`, `a2ui::binding::refresh`, `a2ui::stamp-session`, and `a2ui::on-config-change` are internal Console/Harness lifecycle functions, not agent tools.
+- Live bindings never bind `browser::*`, `harness::*`, `engine::*`, hook, `http`, `cron`, or queue trigger types. A binding `query` must be a function registered by the trigger type's own worker with metadata `{"read_only": true}`.
+- Do not create new `stream` bindings: `stream` is deprecated (iii-stream) and will be removed in an upcoming release. Existing ones keep working while iii-stream runs. Bind the producing worker's own trigger type instead.
 
 ## Functions
 
@@ -40,7 +42,7 @@ Surfaces belong to the current Harness session. The Harness pre-trigger hook sta
 - `a2ui::surface::export` — return a session-free, replayable A2UI JSON package.
 - `a2ui::surface::history`, `undo`, `duplicate`, and `pin` — manage the surface library and revisions.
 - `a2ui::surface::import` and `export-code` — move surfaces between sessions or into source code.
-- `a2ui::binding::set` and `delete` — manage safe declarative live-data bindings.
+- `a2ui::binding::set` and `delete` — manage safe declarative live-data bindings. Example: `{ "id": "clicks-count", "trigger_type": "demo::counter-changed", "config": { "counter": "clicks" }, "target_path": "/count", "query": { "function_id": "demo::counter::get", "payload": { "counter": "clicks" }, "result_path": "/value" } }`.
 - `a2ui::template::*` — save, list, read, apply, and delete reusable session templates.
 - `a2ui::action` — internal Console action ingress and optional Harness forwarding.
 - `a2ui::stamp-session` — internal Harness hook that stamps authoritative turn context.