feat(truapi): localStorage.subscribe and worker pending operations (RFC 0027)

android/truapi-host/src/main/kotlin/io/parity/truapi/TrUAPIHost.kt

@@ -361,6 +361,19 @@ interface HostBridge {
     @Throws(HostRejection::class)
     fun supportedChains(): HostChainSet = HostChainSet(network = "", chains = emptyList())
 
+    /**
+     * Begin a pending operation for a product's worker, returning a
+     * host-assigned id. The host keeps the product's worker alive while it holds
+     * at least one open operation. [label] is a log/UI hint, empty when the
+     * product gave none. Default: no worker keep-alive.
+     */
+    @Throws(HostRejection::class)
+    suspend fun beginOperation(productId: String, label: String): UInt = 0u
+
+    /** End a pending operation. Idempotent: an unknown or already-ended id succeeds. */
+    @Throws(HostRejection::class)
+    suspend fun endOperation(productId: String, id: UInt) {}
+
     /** Product-scoped key-value storage for the Rust core. */
     val storage: HostStorage
 
@@ -502,6 +515,12 @@ private class HostCallbackAdapter(private val bridge: HostBridge) : HostCallback
 
     override fun localStorageClear(key: String) =
         withStorageException { bridge.storage.clear(key) }
+
+    override suspend fun beginOperation(productId: String, label: String): UInt =
+        withHostRejection { bridge.beginOperation(productId, label) }
+
+    override suspend fun endOperation(productId: String, id: UInt) =
+        withHostRejection { bridge.endOperation(productId, id) }
 }
 
 // A host that throws an exception type its callback does not declare crosses
@@ -951,6 +970,14 @@ class TrUAPIProductExecution internal constructor(
         inner.notifyLocaleChanged(locale)
     }
 
+    /**
+     * Push a host storage change for [key] to active TrUAPI storage
+     * subscriptions. A null [value] represents a cleared or absent key.
+     */
+    fun notifyStorageChanged(key: String, value: ByteArray?) {
+        inner.notifyStorageChanged(key, value)
+    }
+
     /** Push a preimage lookup update to active subscriptions for [key]. */
     fun notifyPreimageChanged(key: ByteArray, value: ByteArray?) {
         inner.notifyPreimageChanged(key, value)

docs/rfcs/0027-storage-subscriptions-pending-operations.md

@@ -0,0 +1,130 @@
+---
+title: "Product storage subscriptions and worker pending operations"
+owner: "Sergey Zhuravlev"
+status: draft
+---
+
+# RFC 0027: Product storage subscriptions and worker pending operations
+
+|                 |                                          |
+| --------------- | ---------------------------------------- |
+| **RFC Number**  | 27                                       |
+| **Start Date**  | 2026-08-25                               |
+| **Description** | Two TrUAPI additions so a background worker can finish a multi-step task and coordinate with the app through storage. |
+| **Authors**     | Sergey Zhuravlev                         |
+
+## Summary
+
+- `localStorage.subscribe(key)` streams a key's value on every change, within the product's own namespace.
+- `worker.beginOperation()` / `worker.endOperation(id)` declare a pending operation. The host keeps the worker running while any operation is open.
+
+## Motivation
+
+A funding operation, part of a safety-net release, builds a transaction, submits it, waits for inclusion and records the result. Steps hit a backend, so one run takes tens of seconds with polling in between. It runs in a worker so it outlives the product's screen. Two things make that unsafe today.
+
+The host disposes a worker once its on-screen surface is gone (the worker's own `dispose` in `worker-runtime.ts` is a no-op that defers to the main thread), and the in-flight submission dies with it. The product has no way to say it is mid-operation.
+
+The screen and the worker are separate runtimes over one storage namespace, and a write in one is invisible to the other until it re-reads. A progress view fed by the worker can only poll.
+
+## Detailed Design
+
+### localStorage.subscribe
+
+On the `LocalStorage` trait next to `read`, `write` and `clear`:
+
+```rust
+async fn subscribe(
+    &self,
+    cx: &CallContext,
+    request: HostLocalStorageSubscribeRequest, // { key: String }
+) -> Subscription<HostLocalStorageChangeItem>;
+
+pub struct HostLocalStorageChangeItem {
+    /// `Some` on write, `None` after clear.
+    pub value: Option<Vec<u8>>,
+}
+```
+
+The host owns the store both runtimes write to, so the host emits the changes and the core forwards its stream to the subscriber. This adds one method to the required host `ProductStorage` trait:
+
+```rust
+fn subscribe_storage(
+    &self,
+    product: &ProductContext,
+    key: String,
+) -> BoxStream<'static, Result<HostLocalStorageChangeItem, GenericError>>;
+```
+
+The core passes the calling product's context, the same scoping `read` and `write` use, so a product only sees its own keys. The first item is the current value, so there is no read-then-subscribe gap. After that a write emits `Some(value)` and a clear emits `None`. A burst of distinct values emits one item per value; nothing coalesces.
+
+If `write` gets the bytes the key already holds, the core skips the store write, so the host never sees it and nothing is emitted. `clear` always reaches the host and always emits `None`, even on an absent key.
+
+### Pending operations
+
+A `begin`/`end` pair on a new `Worker` trait, gated to the Worker execution kind:
+
+```rust
+async fn begin_operation(
+    &self,
+    cx: &CallContext,
+    request: HostWorkerBeginOperationRequest, // { label: Option<String> } for host logs and UI
+) -> Result<HostWorkerBeginOperationResponse, CallError<HostWorkerOperationError>>;
+
+/// Idempotent: an unknown or already-ended id returns `Ok`.
+async fn end_operation(
+    &self,
+    cx: &CallContext,
+    request: HostWorkerEndOperationRequest, // { id: OperationId }
+) -> Result<HostWorkerEndOperationResponse, CallError<HostWorkerOperationError>>;
+
+/// Host-assigned, unique per product. Like `NotificationId`.
+pub type OperationId = u32;
+
+pub struct HostWorkerBeginOperationResponse {
+    pub id: OperationId,
+}
+
+pub enum HostWorkerOperationError {
+    /// The product is at the host's per-product cap of open operations.
+    /// `end` never returns this.
+    TooManyOpen,
+    Unknown { reason: String },
+}
+```
+
+The id is session-scoped and never persisted. No host stores operations, so if the host dies the worker, the id and the operation die together and the next launch starts with none open.
+
+The host owns operations and the lifecycle. The core forwards each call to a host trait scoped to the calling product, and the operation existing is the keep-alive signal. Every host implements the trait; one with no worker lifecycle, such as the headless CLI, hands out ids and tracks nothing.
+
+```rust
+#[async_trait]
+pub trait ProductOperations: Send + Sync {
+    /// `label` is empty when the product gave none.
+    async fn begin_operation(
+        &self,
+        product: &ProductContext,
+        label: String,
+    ) -> Result<HostWorkerBeginOperationResponse, HostWorkerOperationError>;
+
+    /// Idempotent.
+    async fn end_operation(
+        &self,
+        product: &ProductContext,
+        id: OperationId,
+    ) -> Result<(), HostWorkerOperationError>;
+}
+```
+
+Operations are keyed by id and scoped to the product. Two tasks each begin one and the worker stays alive until both end. Ending an id twice, or one that was never begun, releases nothing.
+
+Keep-alive is best-effort. iOS and Android can kill a backgrounded worker. An open operation lets the host request the background time the platform allows (a background task assertion on iOS, a foreground service or WorkManager on Android), and the worker resumes from saved state after a kill.
+
+An operation is opaque: an optional label, no funding or deposit typing.
+
+## Compatibility
+
+All three wire methods are additive. `subscribe_storage` and `ProductOperations` are required host capabilities. Target is v0.2 / latest.
+
+## Future directions
+
+Operations become one term of a general liveness rule: `can_execute = has pending operations || has chats || has pocket cards || ...`. On top of that, a host UI listing active operations with `status`, a `list_operations` read to feed it, and user cancel. A prefix subscription if products want to watch a set of keys.

docs/rfcs/_index.md

@@ -26,3 +26,4 @@ created: 2026-03-13
 | 0022   | [Account key derivations](0022-account-derivations.md)                                                   | draft    | Valentin Sergeev  | —                                                               |
 | 0023   | [sr25519 VRF signing for product accounts](0023-account-sign-vrf.md)                                     | draft    | Valentin Sergeev  | —                                                               |
 | 0026   | [Host chain discovery and name resolution](0026-supported-chains.md)                                     | draft    | Valentin Fernandez | [#354](https://github.com/paritytech/host-rust-core/pull/354)          |
+| 0027   | [Product storage subscriptions and worker pending operations](0027-storage-subscriptions-pending-operations.md) | draft    | Sergey Zhuravlev  | —                                                               |

ios/truapi-host/Sources/TrUAPIHost/TrUAPIHost.swift

@@ -342,6 +342,16 @@ public protocol HostBridge: AnyObject, Sendable {
     /// per chain role. Invoked on the dispatcher thread; must return promptly.
     func supportedChains() throws -> HostChainSet
 
+    /// Begin a pending operation for a product's worker, returning a
+    /// host-assigned id. The host keeps the product's worker alive while it
+    /// holds at least one open operation. `label` is a log/UI hint, empty when
+    /// the product gave none.
+    func beginOperation(productId: String, label: String) async throws -> UInt32
+
+    /// End a pending operation. Idempotent: an unknown or already-ended id
+    /// succeeds.
+    func endOperation(productId: String, id: UInt32) async throws
+
     /// Scoped key-value storage for the Rust core.
     var storage: HostStorageBackend { get }
 
@@ -413,6 +423,10 @@ public extension HostBridge {
     func supportedChains() throws -> HostChainSet { HostChainSet(network: "", chains: []) }
     func devicePermissionStatus(request: HostDevicePermissionRequest) async throws
         -> NativeDevicePermissionStatus { .notApplicable }
+    /// Default: no worker keep-alive. Override to run background operations to
+    /// completion after the product's surface goes away.
+    func beginOperation(productId: String, label: String) async throws -> UInt32 { 0 }
+    func endOperation(productId: String, id: UInt32) async throws {}
 }
 
 /// Adapter that bridges the public `ChatHostBridge` to the generated UniFFI
@@ -611,6 +625,18 @@ private final class HostCallbackAdapter: HostCallbacks, @unchecked Sendable {
         }
     }
 
+    func beginOperation(productId: String, label: String) async throws -> UInt32 {
+        try await withHostRejection {
+            try await bridge.beginOperation(productId: productId, label: label)
+        }
+    }
+
+    func endOperation(productId: String, id: UInt32) async throws {
+        try await withHostRejection {
+            try await bridge.endOperation(productId: productId, id: id)
+        }
+    }
+
     private func withHostRejection<T>(_ operation: () throws -> T) throws -> T {
         do {
             return try operation()
@@ -828,6 +854,7 @@ public protocol TrUAPIProductExecutionProtocol: AnyObject, Sendable {
     ) throws
     func notifyThemeChanged(theme: HostThemeSubscribeItem)
     func notifyLocaleChanged(locale: HostLocaleSubscribeItem)
+    func notifyStorageChanged(key: String, value: Data?)
     func notifyPreimageChanged(key: Data, value: Data?)
     func notifyChainResponse(connectionId: UInt32, json: String)
     func notifyChainClosed(connectionId: UInt32)
@@ -907,6 +934,12 @@ public final class TrUAPIProductExecution: TrUAPIProductExecutionProtocol, @unch
         inner.notifyLocaleChanged(locale: locale)
     }
 
+    /// Push a host storage change for `key` to active TrUAPI storage
+    /// subscriptions. `value == nil` represents a cleared or absent key.
+    public func notifyStorageChanged(key: String, value: Data?) {
+        inner.notifyStorageChanged(key: key, value: value)
+    }
+
     public func notifyPreimageChanged(key: Data, value: Data?) {
         inner.notifyPreimageChanged(key: key, value: value)
     }

js/packages/truapi-host/src/test-support.ts

@@ -29,6 +29,11 @@ export function makeHostCallbacks(
       read: async () => undefined,
       write: async () => {},
       clear: async () => {},
+      async *subscribeStorage() {},
+    },
+    productOperations: {
+      beginOperation: async () => ({ id: 1 }),
+      endOperation: async () => {},
     },
     coreStorage: {
       readCoreStorage: async () => undefined,
@@ -65,6 +70,10 @@ export function makeHostCallbacks(
       ...defaults.productStorage,
       ...overrides.productStorage,
     },
+    productOperations: {
+      ...defaults.productOperations,
+      ...overrides.productOperations,
+    },
     coreStorage: {
       ...defaults.coreStorage,
       ...overrides.coreStorage,

js/packages/truapi-host/src/web/create-worker-host-runtime.ts

@@ -17,6 +17,7 @@ import type {
 import {
   CustomRendererNode as CustomRendererNodeCodec,
   HostChatActionSubscribeItem as HostChatActionSubscribeItemCodec,
+  HostWorkerBeginOperationResponse as HostWorkerBeginOperationResponseCodec,
 } from "@parity/truapi";
 import { PermissionAuthorizationRequest as PermissionAuthorizationRequestCodec } from "../generated/host-callbacks.js";
 import { createWasmRawCallbacks } from "../generated/host-callbacks-adapter.js";
@@ -118,6 +119,18 @@ interface RuntimeState {
     }
   >;
   subscriptionDisposers: Map<number, () => void>;
+  /**
+   * Ids of open worker pending operations (`worker.beginOperation`). While
+   * this is non-empty the worker is kept alive: a `dispose()` is deferred
+   * until the last operation ends. Keyed by id so ending an unknown or
+   * already-ended id releases nothing. Worker-global, not per-core, because a
+   * `callbackRequest` carries no core id and "keep the worker alive" is
+   * worker-scoped. ponytail: no cap on how long an operation may hold the
+   * worker; add a timeout ceiling here if a stuck operation becomes a problem.
+   */
+  openOperations: Set<number>;
+  /** A dispose() arrived while operations were open; run it once they drain. */
+  disposePending: boolean;
   chainConnections: Map<number, ChainConnection>;
   pendingDisconnects: Map<
     number,
@@ -372,6 +385,23 @@ function handleCallbackRequest(
     .then(() => fn(...msg.args))
     .then(
       (value) => {
+        // Keep the worker alive across an open pending operation: track ids
+        // only on success, so a rejected begin never leaves a stuck hold.
+        // When the last operation ends and a dispose is pending, run it.
+        if (msg.name === "beginOperation") {
+          state.openOperations.add(
+            HostWorkerBeginOperationResponseCodec.dec(value as Uint8Array).id,
+          );
+        } else if (msg.name === "endOperation") {
+          const id = msg.args[1];
+          if (typeof id === "number") {
+            state.openOperations.delete(id);
+          }
+          if (state.openOperations.size === 0 && state.disposePending) {
+            state.disposePending = false;
+            teardown(state, new Error("runtime disposed"), false);
+          }
+        }
         state.worker.postMessage({
           kind: "callbackResponse",
           requestId: msg.requestId,
@@ -781,6 +811,8 @@ export function createWebWorkerPairingHostRuntime(
       cores: new Map(),
       pendingCores: new Map(),
       subscriptionDisposers: new Map(),
+      openOperations: new Set(),
+      disposePending: false,
       chainConnections: new Map(),
       pendingDisconnects: new Map(),
       pendingSessionActivations: new Map(),
@@ -1220,6 +1252,14 @@ function buildRuntime(state: RuntimeState): WorkerPairingHostRuntime {
     },
     dispose(): void {
       devGlobalTargets.delete(runtime);
+      // Defer a clean dispose while the worker holds an open operation, so a
+      // background task (e.g. a funding transaction) runs to completion. The
+      // last endOperation runs the deferred teardown. Fault teardown is never
+      // deferred.
+      if (state.openOperations.size > 0) {
+        state.disposePending = true;
+        return;
+      }
       teardown(state, new Error("runtime disposed"), false);
     },
   };