diff --git a/android/truapi-host/src/main/kotlin/io/parity/truapi/TrUAPIHost.kt b/android/truapi-host/src/main/kotlin/io/parity/truapi/TrUAPIHost.kt
index a8b0faa62..1208600b7 100644
--- a/android/truapi-host/src/main/kotlin/io/parity/truapi/TrUAPIHost.kt
+++ b/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)
diff --git a/docs/rfcs/0027-storage-subscriptions-pending-operations.md b/docs/rfcs/0027-storage-subscriptions-pending-operations.md
new file mode 100644
index 000000000..a3dcba056
--- /dev/null
+++ b/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.
diff --git a/docs/rfcs/_index.md b/docs/rfcs/_index.md
index daa061f95..1e31e769d 100644
--- a/docs/rfcs/_index.md
+++ b/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  | —                                                               |
diff --git a/ios/truapi-host/Sources/TrUAPIHost/TrUAPIHost.swift b/ios/truapi-host/Sources/TrUAPIHost/TrUAPIHost.swift
index 9a4559c2c..6fffd9af9 100644
--- a/ios/truapi-host/Sources/TrUAPIHost/TrUAPIHost.swift
+++ b/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)
     }
diff --git a/js/packages/truapi-host/src/test-support.ts b/js/packages/truapi-host/src/test-support.ts
index d2ca60740..1514f1fc5 100644
--- a/js/packages/truapi-host/src/test-support.ts
+++ b/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,
diff --git a/js/packages/truapi-host/src/web/create-worker-host-runtime.ts b/js/packages/truapi-host/src/web/create-worker-host-runtime.ts
index 451f0e541..ef28c466e 100644
--- a/js/packages/truapi-host/src/web/create-worker-host-runtime.ts
+++ b/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);
     },
   };
diff --git a/js/packages/truapi-host/src/web/worker-provider.test.ts b/js/packages/truapi-host/src/web/worker-provider.test.ts
index bf8e6cabf..84fa222dc 100644
--- a/js/packages/truapi-host/src/web/worker-provider.test.ts
+++ b/js/packages/truapi-host/src/web/worker-provider.test.ts
@@ -10,7 +10,11 @@ import { bytesToHex } from "@parity/truapi/scale";
 import type { GenericError, Result, ThemeVariant } from "@parity/truapi";
 
 import { createWasmRawCallbacks } from "../generated/host-callbacks-adapter.js";
-import { AuthState, CoreStorageKey } from "../generated/host-callbacks.js";
+import {
+  AuthState,
+  CoreStorageKey,
+  ProductContext,
+} from "../generated/host-callbacks.js";
 import type {
   AuthState as AuthStateValue,
   PreimageHost,
@@ -959,6 +963,97 @@ describe("createWebWorkerPairingHostRuntime", () => {
     provider.dispose();
   });
 
+  it("keeps the worker alive until pending operations end", async () => {
+    const worker = new FakeWorker();
+    const provider = await readyProvider(worker, {
+      runtimeConfig: runtimeConfig({ executionKind: "Worker" }),
+    });
+
+    const product = ProductContext.enc({
+      productId: "dotli.dot",
+      executionKind: "Worker",
+    });
+
+    // Open a pending operation.
+    worker.emit({
+      kind: "callbackRequest",
+      requestId: 1,
+      name: "beginOperation",
+      args: [product, "funding"],
+    });
+    await settle();
+
+    // Disposing while the operation is open defers the worker teardown.
+    provider.dispose();
+    await settle();
+    expect(worker.terminated).toBe(false);
+
+    // Ending the operation runs the deferred teardown, which terminates the
+    // worker on a zero-delay timer.
+    worker.emit({
+      kind: "callbackRequest",
+      requestId: 2,
+      name: "endOperation",
+      args: [product, 1],
+    });
+    await settle();
+    await new Promise((resolve) => setTimeout(resolve, 10));
+    expect(worker.terminated).toBe(true);
+  });
+
+  it("ending an unknown or already-ended operation releases no other hold", async () => {
+    const worker = new FakeWorker();
+    let nextId = 1;
+    const providerPromise = createProviderFromRuntime(
+      asWorker(worker),
+      makeHostCallbacks({
+        productOperations: { beginOperation: async () => ({ id: nextId++ }) },
+      }),
+      { runtimeConfig: runtimeConfig({ executionKind: "Worker" }) },
+    );
+    worker.emit({ kind: "loaded" });
+    worker.emit({ kind: "ready" });
+    const provider = await finishProviderReady(worker, providerPromise);
+
+    const product = ProductContext.enc({
+      productId: "dotli.dot",
+      executionKind: "Worker",
+    });
+    const begin = (requestId: number) =>
+      worker.emit({
+        kind: "callbackRequest",
+        requestId,
+        name: "beginOperation",
+        args: [product, ""],
+      });
+    const end = (requestId: number, id: number) =>
+      worker.emit({
+        kind: "callbackRequest",
+        requestId,
+        name: "endOperation",
+        args: [product, id],
+      });
+
+    begin(1);
+    begin(2);
+    await settle();
+    provider.dispose();
+
+    // Operation 1 ends twice and an unknown id ends once. Operation 2 still
+    // holds the worker.
+    end(3, 1);
+    end(4, 1);
+    end(5, 99);
+    await settle();
+    await new Promise((resolve) => setTimeout(resolve, 10));
+    expect(worker.terminated).toBe(false);
+
+    end(6, 2);
+    await settle();
+    await new Promise((resolve) => setTimeout(resolve, 10));
+    expect(worker.terminated).toBe(true);
+  });
+
   it("routes payload-carrying subscriptions by name", async () => {
     const worker = new FakeWorker();
     const keys: Uint8Array[] = [];
diff --git a/playground/tests/unit/chat-diagnosis.test.ts b/playground/tests/unit/chat-diagnosis.test.ts
index 866ca004d..b8a2e4800 100644
--- a/playground/tests/unit/chat-diagnosis.test.ts
+++ b/playground/tests/unit/chat-diagnosis.test.ts
@@ -7,7 +7,7 @@ describe("ChatDiagnosis", () => {
   // Expectation comes from codegen, so a missing method fails here.
   test("covers every generated Chat method", () => {
     const generated = servicesForExecution(generatedServices, "Worker")
-      .filter((service) => service.requiredExecution === "Worker")
+      .filter((service) => service.name === "Chat")
       .flatMap((service) =>
         service.methods.map((method) => `${service.name}/${method.name}`),
       );
diff --git a/rust/crates/truapi-codegen/tests/golden/dispatcher.rs b/rust/crates/truapi-codegen/tests/golden/dispatcher.rs
index 7b059b295..2129afbb7 100644
--- a/rust/crates/truapi-codegen/tests/golden/dispatcher.rs
+++ b/rust/crates/truapi-codegen/tests/golden/dispatcher.rs
@@ -29,6 +29,7 @@ use truapi::api::{
     StatementStore,
     System,
     Theme,
+    Worker,
 };
 use truapi::versioned::{self, Versioned};
 use truapi_platform::ProductExecutionKind;
@@ -63,7 +64,8 @@ where
     register_signing(dispatcher, host.clone());
     register_statement_store(dispatcher, host.clone());
     register_system(dispatcher, host.clone());
-    register_theme(dispatcher, host);
+    register_theme(dispatcher, host.clone());
+    register_worker(dispatcher, host);
 }
 
 /// Start the host-initiated `chat_custom_message_render` subscription.
@@ -1626,7 +1628,7 @@ where
         });
     }
     {
-        let host = host;
+        let host = host.clone();
         dispatcher.on_request(wire_table::LOCAL_STORAGE_CLEAR, move |request_id: String, bytes: Vec<u8>| {
             let host = host.clone();
             Box::pin(async move {
@@ -1664,6 +1666,21 @@ where
             })
         });
     }
+    {
+        let host = host;
+        dispatcher.on_subscription(wire_table::LOCAL_STORAGE_SUBSCRIBE, move |request_id: String, bytes: Vec<u8>| {
+            let host = host.clone();
+            Box::pin(async move {
+                let request: versioned::local_storage::HostLocalStorageSubscribeRequest = match Decode::decode(&mut &bytes[..]) {
+                    Ok(request) => request,
+                    Err(_) => return Err(Vec::new()),
+                };
+                let cx = CallContext::with_request_id(request_id.clone());
+                let stream = host.subscribe(&cx, request).await;
+                Ok(subscription_stream::<versioned::local_storage::HostLocalStorageChangeItem, _>(stream))
+            })
+        });
+    }
 }
 
 fn register_locale<P>(dispatcher: &mut Dispatcher, host: Arc<P>)
@@ -2700,3 +2717,99 @@ where
         });
     }
 }
+
+fn register_worker<P>(dispatcher: &mut Dispatcher, host: Arc<P>)
+where
+    P: Worker + Send + Sync + 'static,
+{
+    {
+        let execution_allowed = dispatcher.allows_execution(ProductExecutionKind::Worker);
+        let host = host.clone();
+        dispatcher.on_request(wire_table::WORKER_BEGIN_OPERATION, move |request_id: String, bytes: Vec<u8>| {
+            let host = host.clone();
+            Box::pin(async move {
+                let request: versioned::worker::HostWorkerBeginOperationRequest = match Decode::decode(&mut &bytes[..]) {
+                    Ok(request) => request,
+                    Err(err) => {
+                        let error: truapi::CallError<versioned::worker::HostWorkerBeginOperationError> =
+                            truapi::CallError::MalformedFrame { reason: err.to_string() };
+                        return Ok(encode_versioned_err_payload(
+                            error,
+                            <versioned::worker::HostWorkerBeginOperationError as Versioned>::LATEST,
+                        ));
+                    }
+                };
+                let target_version = request.version();
+                let cx = CallContext::with_request_id(request_id.clone());
+                if !execution_allowed {
+                    let error: truapi::CallError<versioned::worker::HostWorkerBeginOperationError> =
+                        truapi::CallError::Denied;
+                    return Ok(encode_versioned_err_payload(error, target_version));
+                }
+                let response: versioned::worker::HostWorkerBeginOperationResponse = match host.begin_operation(&cx, request).await {
+                    Ok(value) => value,
+                    Err(err) => {
+                        return Ok(encode_versioned_err_payload(
+                            downgrade_call_error(err, target_version),
+                            target_version,
+                        ));
+                    }
+                };
+                // Downgraded to the caller's version: a handler answers in
+                // latest terms, and a peer that asked in an older version
+                // cannot decode a newer variant.
+                Ok(encode_versioned_ok_payload(
+                    <versioned::worker::HostWorkerBeginOperationResponse as truapi::versioned::FromLatest>::from_latest(
+                        truapi::versioned::IntoLatest::into_latest(response),
+                        target_version,
+                    ),
+                ))
+            })
+        });
+    }
+    {
+        let execution_allowed = dispatcher.allows_execution(ProductExecutionKind::Worker);
+        let host = host;
+        dispatcher.on_request(wire_table::WORKER_END_OPERATION, move |request_id: String, bytes: Vec<u8>| {
+            let host = host.clone();
+            Box::pin(async move {
+                let request: versioned::worker::HostWorkerEndOperationRequest = match Decode::decode(&mut &bytes[..]) {
+                    Ok(request) => request,
+                    Err(err) => {
+                        let error: truapi::CallError<versioned::worker::HostWorkerEndOperationError> =
+                            truapi::CallError::MalformedFrame { reason: err.to_string() };
+                        return Ok(encode_versioned_err_payload(
+                            error,
+                            <versioned::worker::HostWorkerEndOperationError as Versioned>::LATEST,
+                        ));
+                    }
+                };
+                let target_version = request.version();
+                let cx = CallContext::with_request_id(request_id.clone());
+                if !execution_allowed {
+                    let error: truapi::CallError<versioned::worker::HostWorkerEndOperationError> =
+                        truapi::CallError::Denied;
+                    return Ok(encode_versioned_err_payload(error, target_version));
+                }
+                let response: versioned::worker::HostWorkerEndOperationResponse = match host.end_operation(&cx, request).await {
+                    Ok(value) => value,
+                    Err(err) => {
+                        return Ok(encode_versioned_err_payload(
+                            downgrade_call_error(err, target_version),
+                            target_version,
+                        ));
+                    }
+                };
+                // Downgraded to the caller's version: a handler answers in
+                // latest terms, and a peer that asked in an older version
+                // cannot decode a newer variant.
+                Ok(encode_versioned_ok_payload(
+                    <versioned::worker::HostWorkerEndOperationResponse as truapi::versioned::FromLatest>::from_latest(
+                        truapi::versioned::IntoLatest::into_latest(response),
+                        target_version,
+                    ),
+                ))
+            })
+        });
+    }
+}
diff --git a/rust/crates/truapi-codegen/tests/golden/host-callbacks-adapter.ts b/rust/crates/truapi-codegen/tests/golden/host-callbacks-adapter.ts
index be3d7e146..a3f25365f 100644
--- a/rust/crates/truapi-codegen/tests/golden/host-callbacks-adapter.ts
+++ b/rust/crates/truapi-codegen/tests/golden/host-callbacks-adapter.ts
@@ -17,10 +17,12 @@ import {
   HostDevicePermissionResponse,
   HostFeatureSupportedRequest,
   HostFeatureSupportedResponse,
+  HostLocalStorageChangeItem,
   HostLocaleSubscribeItem,
   HostPushNotificationRequest,
   HostPushNotificationResponse,
   HostThemeSubscribeItem,
+  HostWorkerBeginOperationResponse,
   RemotePermissionRequest,
   RemotePermissionResponse,
 } from "@parity/truapi";
@@ -83,9 +85,16 @@ export interface RawCallbacks {
     sendItem: (item?: Uint8Array) => void,
     sendError: (error: GenericError) => void,
   ): (() => void) | void;
+  beginOperation(product: Uint8Array, label: string): Promise<Uint8Array>;
+  endOperation(product: Uint8Array, id: number): Promise<void>;
   read(key: string): Promise<Uint8Array | null | undefined>;
   write(key: string, value: Uint8Array): Promise<void>;
   clear(key: string): Promise<void>;
+  subscribeStorage(
+    key: Uint8Array,
+    sendItem: (item?: Uint8Array) => void,
+    sendError: (error: GenericError) => void,
+  ): (() => void) | void;
   subscribeTheme(
     sendItem: (item?: Uint8Array) => void,
     sendError: (error: GenericError) => void,
@@ -194,10 +203,28 @@ export function createWasmRawCallbacks(
         sendItem,
         sendError,
       ),
+    beginOperation: async (product, label) =>
+      HostWorkerBeginOperationResponse.enc(
+        await callbacks.productOperations.beginOperation(
+          ProductContext.dec(product),
+          label,
+        ),
+      ),
+    endOperation: async (product, id) =>
+      await callbacks.productOperations.endOperation(
+        ProductContext.dec(product),
+        id,
+      ),
     read: async (key) => await callbacks.productStorage.read(key),
     write: async (key, value) =>
       await callbacks.productStorage.write(key, value),
     clear: async (key) => await callbacks.productStorage.clear(key),
+    subscribeStorage: (key, sendItem, sendError) =>
+      driveResultStream(
+        callbacks.productStorage.subscribeStorage(key),
+        (item) => sendItem(HostLocalStorageChangeItem.enc(item)),
+        sendError,
+      ),
     subscribeTheme: (sendItem, sendError) =>
       driveResultStream(
         callbacks.theme.subscribeTheme(),
diff --git a/rust/crates/truapi-codegen/tests/golden/host-callbacks.ts b/rust/crates/truapi-codegen/tests/golden/host-callbacks.ts
index b9ccb6ee7..a4f588d47 100644
--- a/rust/crates/truapi-codegen/tests/golden/host-callbacks.ts
+++ b/rust/crates/truapi-codegen/tests/golden/host-callbacks.ts
@@ -36,10 +36,12 @@ import type {
   HostDevicePermissionResponse,
   HostFeatureSupportedRequest,
   HostFeatureSupportedResponse,
+  HostLocalStorageChangeItem,
   HostLocaleSubscribeItem,
   HostPushNotificationRequest,
   HostPushNotificationResponse,
   HostThemeSubscribeItem,
+  HostWorkerBeginOperationResponse,
   NotificationId,
   RemotePermissionResponse,
   Result,
@@ -1276,6 +1278,29 @@ export interface PreimageHost {
   ): AsyncIterable<Result<Uint8Array | undefined, GenericError>>;
 }
 
+/**
+ * Host store for a product's pending operations. The host keeps the product's
+ * worker runtime alive while it holds at least one open operation. Worker
+ * products reach these through the `Worker` protocol trait, which is gated to
+ * the Worker execution kind, so non-worker products never call them.
+ */
+export interface ProductOperations {
+  /**
+   * Record a pending operation for this product. Returns its id. `label` is
+   * a host log and UI hint, empty when the product gave none.
+   */
+  beginOperation(
+    product: ProductContext,
+    label: string,
+  ): Promise<HostWorkerBeginOperationResponse>;
+
+  /**
+   * Remove a pending operation. Idempotent: an unknown or already-ended id
+   * returns `Ok`.
+   */
+  endOperation(product: ProductContext, id: number): Promise<void>;
+}
+
 /**
  * Product-scoped key-value storage.
  *
@@ -1298,6 +1323,16 @@ export interface ProductStorage {
    * Clear a value at a key.
    */
   clear(key: string): Promise<void>;
+
+  /**
+   * Emit the current value of a key, then each later change to it, from any
+   * of the product's runtimes. A write that leaves the stored bytes
+   * unchanged emits nothing. `key` is the UTF-8 bytes of the namespaced
+   * storage key, the same key passed to `Self::read` as a `String`.
+   */
+  subscribeStorage(
+    key: Uint8Array,
+  ): AsyncIterable<Result<HostLocalStorageChangeItem, GenericError>>;
 }
 
 /**
@@ -1339,6 +1374,7 @@ export interface HostCallbacks {
   theme: ThemeHost;
   locale: LocaleHost;
   preimage: PreimageHost;
+  productOperations: ProductOperations;
   chat?: ChatPlatform;
   permissionStatus?: PermissionStatusHost;
 }
@@ -1356,6 +1392,7 @@ export interface RequiredHostCallbacks {
   theme: Required<ThemeHost>;
   locale: Required<LocaleHost>;
   preimage: Required<PreimageHost>;
+  productOperations: Required<ProductOperations>;
   chat?: Required<ChatPlatform>;
   permissionStatus?: Required<PermissionStatusHost>;
 }
diff --git a/rust/crates/truapi-codegen/tests/golden/wasm_bridge.rs b/rust/crates/truapi-codegen/tests/golden/wasm_bridge.rs
index 329ce4058..806406e35 100644
--- a/rust/crates/truapi-codegen/tests/golden/wasm_bridge.rs
+++ b/rust/crates/truapi-codegen/tests/golden/wasm_bridge.rs
@@ -45,9 +45,12 @@ pub(super) struct JsBridge {
     pub(super) device_permission: Function,
     pub(super) remote_permission: Function,
     pub(super) lookup_preimage: Function,
+    pub(super) begin_operation: Function,
+    pub(super) end_operation: Function,
     pub(super) read: Function,
     pub(super) write: Function,
     pub(super) clear: Function,
+    pub(super) subscribe_storage: Function,
     pub(super) subscribe_theme: Function,
     pub(super) confirm_user_action: Function,
     pub(super) chat_present: bool,
@@ -81,9 +84,12 @@ impl JsBridge {
             device_permission: get_function(callbacks, "devicePermission")?,
             remote_permission: get_function(callbacks, "remotePermission")?,
             lookup_preimage: get_function(callbacks, "lookupPreimage")?,
+            begin_operation: get_function(callbacks, "beginOperation")?,
+            end_operation: get_function(callbacks, "endOperation")?,
             read: get_function(callbacks, "read")?,
             write: get_function(callbacks, "write")?,
             clear: get_function(callbacks, "clear")?,
+            subscribe_storage: get_function(callbacks, "subscribeStorage")?,
             subscribe_theme: get_function(callbacks, "subscribeTheme")?,
             confirm_user_action: get_function(callbacks, "confirmUserAction")?,
             chat_present: get_optional_function(callbacks, "createChatRoom")?.is_some()
@@ -389,6 +395,46 @@ impl truapi_platform::PreimageHost for WasmPlatform {
     }
 }
 
+#[truapi_platform::async_trait]
+impl truapi_platform::ProductOperations for WasmPlatform {
+    async fn begin_operation(
+        &self,
+        product: &truapi_platform::ProductContext,
+        label: String,
+    ) -> Result<v01::HostWorkerBeginOperationResponse, v01::HostWorkerOperationError> {
+        let bytes = invoke_bytes_return(
+            &self.bridge.begin_operation,
+            vec![
+                Uint8Array::from(product.encode().as_slice()).into(),
+                JsValue::from_str(&label),
+            ],
+        )
+        .await
+        .map_err(|reason| v01::HostWorkerOperationError::Unknown { reason })?;
+        decode_bytes::<v01::HostWorkerBeginOperationResponse>(
+            bytes,
+            "beginOperation response did not decode",
+        )
+        .map_err(|reason| v01::HostWorkerOperationError::Unknown { reason })
+    }
+
+    async fn end_operation(
+        &self,
+        product: &truapi_platform::ProductContext,
+        id: u32,
+    ) -> Result<(), v01::HostWorkerOperationError> {
+        invoke_unit(
+            &self.bridge.end_operation,
+            vec![
+                Uint8Array::from(product.encode().as_slice()).into(),
+                JsValue::from_f64(f64::from(id)),
+            ],
+        )
+        .await
+        .map_err(|reason| v01::HostWorkerOperationError::Unknown { reason })
+    }
+}
+
 #[truapi_platform::async_trait]
 impl truapi_platform::ProductStorage for WasmPlatform {
     async fn read(&self, key: String) -> Result<Option<Vec<u8>>, v01::HostLocalStorageReadError> {
@@ -422,6 +468,17 @@ impl truapi_platform::ProductStorage for WasmPlatform {
             .await
             .map_err(|reason| v01::HostLocalStorageReadError::Unknown { reason })
     }
+
+    fn subscribe_storage(
+        &self,
+        key: Vec<u8>,
+    ) -> BoxStream<'static, Result<v01::HostLocalStorageChangeItem, v01::GenericError>> {
+        invoke_js_subscription(
+            &self.bridge.subscribe_storage,
+            Some(key),
+            parse_host_local_storage_change_item_item,
+        )
+    }
 }
 
 impl truapi_platform::ThemeHost for WasmPlatform {
@@ -457,6 +514,12 @@ fn parse_host_chat_list_subscribe_item_item(
     decode_js_item::<v01::HostChatListSubscribeItem>(value, "HostChatListSubscribeItem")
 }
 
+fn parse_host_local_storage_change_item_item(
+    value: JsValue,
+) -> Result<v01::HostLocalStorageChangeItem, String> {
+    decode_js_item::<v01::HostLocalStorageChangeItem>(value, "HostLocalStorageChangeItem")
+}
+
 fn parse_host_locale_subscribe_item_item(
     value: JsValue,
 ) -> Result<v01::HostLocaleSubscribeItem, String> {
diff --git a/rust/crates/truapi-codegen/tests/golden/wire_table.rs b/rust/crates/truapi-codegen/tests/golden/wire_table.rs
index 5d59d7f93..4fbce8324 100644
--- a/rust/crates/truapi-codegen/tests/golden/wire_table.rs
+++ b/rust/crates/truapi-codegen/tests/golden/wire_table.rs
@@ -50,7 +50,7 @@ pub enum WireKind {
 /// `TRUAPI_WIRE_SCHEMA_HASH`. A host stamps it on each debug envelope so
 /// the debugger refuses to decode a frame whose contract differs from
 /// its own, even when the coarse handshake codec version is unchanged.
-pub const TRUAPI_WIRE_SCHEMA_HASH: &str = "0449982638d57658";
+pub const TRUAPI_WIRE_SCHEMA_HASH: &str = "e141a8b071605ab4";
 
 /// Wire discriminants for `system_handshake`.
 pub const SYSTEM_HANDSHAKE: RequestFrameIds = RequestFrameIds {
@@ -516,6 +516,26 @@ pub const LOCALE_SUBSCRIBE: SubscriptionFrameIds = SubscriptionFrameIds {
     receive_id: 197,
 };
 
+/// Wire discriminants for `local_storage_subscribe`.
+pub const LOCAL_STORAGE_SUBSCRIBE: SubscriptionFrameIds = SubscriptionFrameIds {
+    start_id: 198,
+    stop_id: 199,
+    interrupt_id: 200,
+    receive_id: 201,
+};
+
+/// Wire discriminants for `worker_begin_operation`.
+pub const WORKER_BEGIN_OPERATION: RequestFrameIds = RequestFrameIds {
+    request_id: 202,
+    response_id: 203,
+};
+
+/// Wire discriminants for `worker_end_operation`.
+pub const WORKER_END_OPERATION: RequestFrameIds = RequestFrameIds {
+    request_id: 204,
+    response_id: 205,
+};
+
 /// The full wire table. Ordering is part of the wire protocol;
 /// only ever append. Removed methods leave their slot empty.
 pub const WIRE_TABLE: &[WireEntry] = &[
@@ -807,4 +827,16 @@ pub const WIRE_TABLE: &[WireEntry] = &[
         method: "locale_subscribe",
         kind: WireKind::Subscription(LOCALE_SUBSCRIBE),
     },
+    WireEntry {
+        method: "local_storage_subscribe",
+        kind: WireKind::Subscription(LOCAL_STORAGE_SUBSCRIBE),
+    },
+    WireEntry {
+        method: "worker_begin_operation",
+        kind: WireKind::Request(WORKER_BEGIN_OPERATION),
+    },
+    WireEntry {
+        method: "worker_end_operation",
+        kind: WireKind::Request(WORKER_END_OPERATION),
+    },
 ];
diff --git a/rust/crates/truapi-codegen/tests/golden/worker-callbacks.ts b/rust/crates/truapi-codegen/tests/golden/worker-callbacks.ts
index 4fb2eb95b..8edccf8cc 100644
--- a/rust/crates/truapi-codegen/tests/golden/worker-callbacks.ts
+++ b/rust/crates/truapi-codegen/tests/golden/worker-callbacks.ts
@@ -26,6 +26,8 @@ export const CALLBACK_NAMES = [
   "devicePermissionStatus",
   "devicePermission",
   "remotePermission",
+  "beginOperation",
+  "endOperation",
   "read",
   "write",
   "clear",
@@ -37,6 +39,7 @@ export const SUBSCRIPTION_NAMES = [
   "subscribeChatRooms",
   "subscribeLocale",
   "lookupPreimage",
+  "subscribeStorage",
   "subscribeTheme",
 ] as const;
 export type SubscriptionName = (typeof SUBSCRIPTION_NAMES)[number];
@@ -71,6 +74,8 @@ function rawCallbacks(
     | "cancelNotification"
     | "devicePermission"
     | "remotePermission"
+    | "beginOperation"
+    | "endOperation"
     | "read"
     | "write"
     | "clear"
@@ -120,6 +125,14 @@ function rawCallbacks(
       bridge.callbackRequest("remotePermission", [request]) as ReturnType<
         Required<RawCallbacks>["remotePermission"]
       >,
+    beginOperation: (product, label) =>
+      bridge.callbackRequest("beginOperation", [product, label]) as ReturnType<
+        Required<RawCallbacks>["beginOperation"]
+      >,
+    endOperation: (product, id) =>
+      bridge.callbackRequest("endOperation", [product, id]) as ReturnType<
+        Required<RawCallbacks>["endOperation"]
+      >,
     read: (key) =>
       bridge.callbackRequest("read", [key]) as ReturnType<
         Required<RawCallbacks>["read"]
@@ -142,13 +155,18 @@ function rawCallbacks(
 function subscriptionRawCallbacks(
   bridge: WorkerCallbackBridge,
 ): Required<
-  Pick<RawCallbacks, "subscribeLocale" | "lookupPreimage" | "subscribeTheme">
+  Pick<
+    RawCallbacks,
+    "subscribeLocale" | "lookupPreimage" | "subscribeStorage" | "subscribeTheme"
+  >
 > {
   return {
     subscribeLocale: (sendItem, sendError) =>
       bridge.startSubscription("subscribeLocale", null, sendItem, sendError),
     lookupPreimage: (key, sendItem, sendError) =>
       bridge.startSubscription("lookupPreimage", key, sendItem, sendError),
+    subscribeStorage: (key, sendItem, sendError) =>
+      bridge.startSubscription("subscribeStorage", key, sendItem, sendError),
     subscribeTheme: (sendItem, sendError) =>
       bridge.startSubscription("subscribeTheme", null, sendItem, sendError),
   };
@@ -251,6 +269,12 @@ export function startRawSubscription(
         return undefined;
       }
       return callbacks.lookupPreimage(payload, sendItem, sendError);
+    case "subscribeStorage":
+      if (payload === null) {
+        console.warn(`[truapi worker] ${name} requires payload`);
+        return undefined;
+      }
+      return callbacks.subscribeStorage(payload, sendItem, sendError);
     case "subscribeTheme":
       return callbacks.subscribeTheme(sendItem, sendError);
   }
diff --git a/rust/crates/truapi-host-cli/src/platform.rs b/rust/crates/truapi-host-cli/src/platform.rs
index facad2333..9a65e26a0 100644
--- a/rust/crates/truapi-host-cli/src/platform.rs
+++ b/rust/crates/truapi-host-cli/src/platform.rs
@@ -24,14 +24,15 @@ use truapi::latest as api;
 use truapi_platform::{
     AuthState, ChainProvider, CoreStorage, CoreStorageKey, DevicePermissionStatus, Features,
     JsonRpcConnection, LocaleHost, Navigation, Notifications, PermissionStatusHost, Permissions,
-    PreimageHost, ProductStorage, ProductStorageKey, SessionUiInfo, ThemeHost, UserConfirmation,
-    UserConfirmationReview,
+    PreimageHost, ProductContext, ProductOperations, ProductStorage, ProductStorageKey,
+    SessionUiInfo, ThemeHost, UserConfirmation, UserConfirmationReview,
 };
 
 use crate::chain::WsChainProvider;
 use crate::terminal_ui::{SystemEvent, UiHandle};
 
 static NEXT_STORAGE_TEMP_ID: AtomicU32 = AtomicU32::new(0);
+static NEXT_OPERATION_ID: AtomicU32 = AtomicU32::new(1);
 
 /// How the host answers confirmation prompts (the web/iOS "sign?" modals).
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -478,6 +479,51 @@ impl ProductStorage for CliPlatform {
         self.persist_product_storage(scoped.product_id(), values)
             .map_err(|reason| api::HostLocalStorageReadError::Unknown { reason })
     }
+
+    fn subscribe_storage(
+        &self,
+        key: Vec<u8>,
+    ) -> BoxStream<'static, Result<api::HostLocalStorageChangeItem, api::GenericError>> {
+        // ponytail: emits the current value once; the CLI host does not push
+        // later changes. Wire a per-key broadcast off write/clear if e2e needs
+        // cross-context storage sync against the CLI signing host.
+        let key = String::from_utf8_lossy(&key).into_owned();
+        let value = ProductStorageKey::decode(&key).ok().and_then(|scoped| {
+            self.product_storage
+                .lock()
+                .expect("product storage mutex poisoned")
+                .get(scoped.product_id())
+                .and_then(|values| values.get(scoped.key()))
+                .cloned()
+        });
+        Box::pin(stream::once(async move {
+            Ok(api::HostLocalStorageChangeItem { value })
+        }))
+    }
+}
+
+#[async_trait]
+impl ProductOperations for CliPlatform {
+    async fn begin_operation(
+        &self,
+        _product: &ProductContext,
+        _label: String,
+    ) -> Result<api::HostWorkerBeginOperationResponse, api::HostWorkerOperationError> {
+        // ponytail: the headless CLI has no worker to keep alive; hand back a
+        // unique id so a product can pair begin/end. Add refcounting if the CLI
+        // ever backgrounds worker executions.
+        Ok(api::HostWorkerBeginOperationResponse {
+            id: NEXT_OPERATION_ID.fetch_add(1, Ordering::Relaxed),
+        })
+    }
+
+    async fn end_operation(
+        &self,
+        _product: &ProductContext,
+        _id: u32,
+    ) -> Result<(), api::HostWorkerOperationError> {
+        Ok(())
+    }
 }
 
 #[async_trait]
diff --git a/rust/crates/truapi-platform/src/lib.rs b/rust/crates/truapi-platform/src/lib.rs
index 170cbaf4e..b0d23defb 100644
--- a/rust/crates/truapi-platform/src/lib.rs
+++ b/rust/crates/truapi-platform/src/lib.rs
@@ -36,13 +36,14 @@ use truapi::latest::{
     HostChatListSubscribeItem, HostChatPostMessageError, HostChatPostMessageRequest,
     HostChatPostMessageResponse, HostChatRegisterBotError, HostChatRegisterBotRequest,
     HostChatRegisterBotResponse, HostDevicePermissionRequest, HostDevicePermissionResponse,
-    HostFeatureSupportedRequest, HostFeatureSupportedResponse, HostLocalStorageReadError,
-    HostLocaleSubscribeItem, HostNavigateToError, HostPlatform, HostPushNotificationRequest,
-    HostPushNotificationResponse, HostSignPayloadRequest, HostSignPayloadWithLegacyAccountRequest,
-    HostSignRawRequest, HostSignRawWithLegacyAccountRequest, HostThemeSubscribeItem,
-    LegacyAccountTxPayload, NotificationId, ProductAccountId, ProductAccountTxPayload,
-    ProductProofContext, RemotePermission, RemotePermissionRequest, RemotePermissionResponse,
-    RingLocation,
+    HostFeatureSupportedRequest, HostFeatureSupportedResponse, HostLocalStorageChangeItem,
+    HostLocalStorageReadError, HostLocaleSubscribeItem, HostNavigateToError, HostPlatform,
+    HostPushNotificationRequest, HostPushNotificationResponse, HostSignPayloadRequest,
+    HostSignPayloadWithLegacyAccountRequest, HostSignRawRequest,
+    HostSignRawWithLegacyAccountRequest, HostThemeSubscribeItem, HostWorkerBeginOperationResponse,
+    HostWorkerOperationError, LegacyAccountTxPayload, NotificationId, ProductAccountId,
+    ProductAccountTxPayload, ProductProofContext, RemotePermission, RemotePermissionRequest,
+    RemotePermissionResponse, RingLocation,
 };
 use truapi::v01::HostAccountSignVrfRequest;
 use url::{Host, Url};
@@ -971,6 +972,15 @@ pub trait ProductStorage: Send + Sync {
 
     /// Clear a value at a key.
     async fn clear(&self, key: String) -> Result<(), HostLocalStorageReadError>;
+
+    /// Emit the current value of a key, then each later change to it, from any
+    /// of the product's runtimes. A write that leaves the stored bytes
+    /// unchanged emits nothing. `key` is the UTF-8 bytes of the namespaced
+    /// storage key, the same key passed to [`Self::read`] as a `String`.
+    fn subscribe_storage(
+        &self,
+        key: Vec<u8>,
+    ) -> BoxStream<'static, Result<HostLocalStorageChangeItem, GenericError>>;
 }
 
 /// Open URLs in the system browser. Input is already trimmed, categorized,
@@ -2919,6 +2929,29 @@ pub trait PermissionStatusHost: Send + Sync {
     ) -> Result<DevicePermissionStatus, GenericError>;
 }
 
+/// Host store for a product's pending operations. The host keeps the product's
+/// worker runtime alive while it holds at least one open operation. Worker
+/// products reach these through the `Worker` protocol trait, which is gated to
+/// the Worker execution kind, so non-worker products never call them.
+#[async_trait]
+pub trait ProductOperations: Send + Sync {
+    /// Record a pending operation for this product. Returns its id. `label` is
+    /// a host log and UI hint, empty when the product gave none.
+    async fn begin_operation(
+        &self,
+        product: &ProductContext,
+        label: String,
+    ) -> Result<HostWorkerBeginOperationResponse, HostWorkerOperationError>;
+
+    /// Remove a pending operation. Idempotent: an unknown or already-ended id
+    /// returns `Ok`.
+    async fn end_operation(
+        &self,
+        product: &ProductContext,
+        id: u32,
+    ) -> Result<(), HostWorkerOperationError>;
+}
+
 /// Combined platform interface. A host must provide every capability trait
 /// listed here. Members marked optional may be omitted; the core answers their
 /// product calls with `Unsupported`. See [`OptionalPlatform`].
@@ -2935,6 +2968,7 @@ pub trait Platform:
     + ThemeHost
     + LocaleHost
     + PreimageHost
+    + ProductOperations
 {
 }
 
@@ -2951,6 +2985,7 @@ impl<T> Platform for T where
         + ThemeHost
         + LocaleHost
         + PreimageHost
+        + ProductOperations
 {
 }
 
diff --git a/rust/crates/truapi-server/src/native.rs b/rust/crates/truapi-server/src/native.rs
index ec240ab58..d93f5654e 100644
--- a/rust/crates/truapi-server/src/native.rs
+++ b/rust/crates/truapi-server/src/native.rs
@@ -24,7 +24,7 @@ use truapi_platform::{
     AuthPresenter, AuthState, ChainProvider, CoreAdmin, CoreStorage, CoreStorageKey, Features,
     HostInfo, JsonRpcConnection, LocaleHost, Navigation, Notifications,
     PermissionAuthorizationRequest, PermissionAuthorizationStatus, Permissions, PlatformInfo,
-    PreimageHost, ProductContext, ProductExecutionKind, ProductStorage,
+    PreimageHost, ProductContext, ProductExecutionKind, ProductOperations, ProductStorage,
     RuntimeConfigValidationError, SigningHostConfig, ThemeHost, UserConfirmation,
     UserConfirmationReview, async_trait, normalize_product_identifier,
 };
@@ -525,6 +525,18 @@ pub trait HostCallbacks: Send + Sync {
     fn local_storage_write(&self, key: String, value: Vec<u8>) -> Result<(), HostStorageError>;
     /// Clear a value from the host's scoped key-value store.
     fn local_storage_clear(&self, key: String) -> Result<(), HostStorageError>;
+
+    /// Record 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.
+    async fn begin_operation(
+        &self,
+        product_id: String,
+        label: String,
+    ) -> Result<u32, HostRejection>;
+    /// End a pending operation. Idempotent: an unknown or already-ended id
+    /// succeeds.
+    async fn end_operation(&self, product_id: String, id: u32) -> Result<(), HostRejection>;
 }
 
 /// Native Chat storage and UI adapter. Hosts that support the Chat modality
@@ -1061,6 +1073,11 @@ impl NativeProductExecution {
         self.events.notify_preimage_changed(&key, value);
     }
 
+    /// Push a host storage change to this execution's subscriptions for `key`.
+    pub fn notify_storage_changed(&self, key: String, value: Option<Vec<u8>>) {
+        self.events.notify_storage_changed(&key, value);
+    }
+
     /// Notify this execution's chain adapter of one JSON-RPC response.
     pub fn notify_chain_response(&self, connection_id: u32, json: String) {
         self.shared_events
@@ -1234,6 +1251,7 @@ struct NativeEventBus {
     locale_changes:
         Mutex<Vec<mpsc::UnboundedSender<Result<v01::HostLocaleSubscribeItem, v01::GenericError>>>>,
     preimage_changes: Mutex<Vec<PreimageSubscription>>,
+    storage_changes: Mutex<Vec<StorageSubscription>>,
     chain_responses: Mutex<HashMap<u32, mpsc::UnboundedSender<String>>>,
     chat_room_changes: Mutex<Vec<mpsc::UnboundedSender<v01::HostChatListSubscribeItem>>>,
 }
@@ -1243,6 +1261,11 @@ struct PreimageSubscription {
     tx: mpsc::UnboundedSender<Result<Option<Vec<u8>>, v01::GenericError>>,
 }
 
+struct StorageSubscription {
+    key: String,
+    tx: mpsc::UnboundedSender<Result<v01::HostLocalStorageChangeItem, v01::GenericError>>,
+}
+
 impl NativeEventBus {
     fn subscribe_theme(
         &self,
@@ -1306,6 +1329,31 @@ impl NativeEventBus {
             });
     }
 
+    fn subscribe_storage_changes(
+        &self,
+        key: String,
+    ) -> mpsc::UnboundedReceiver<Result<v01::HostLocalStorageChangeItem, v01::GenericError>> {
+        let (tx, rx) = mpsc::unbounded();
+        self.storage_changes
+            .lock()
+            .expect("native storage subscribers mutex poisoned")
+            .push(StorageSubscription { key, tx });
+        rx
+    }
+
+    fn notify_storage_changed(&self, key: &str, value: Option<Vec<u8>>) {
+        let item = v01::HostLocalStorageChangeItem { value };
+        self.storage_changes
+            .lock()
+            .expect("native storage subscribers mutex poisoned")
+            .retain(|sub| {
+                if sub.key != key {
+                    return true;
+                }
+                sub.tx.unbounded_send(Ok(item.clone())).is_ok()
+            });
+    }
+
     fn register_chain(&self, connection_id: u32) -> mpsc::UnboundedReceiver<String> {
         let (tx, rx) = mpsc::unbounded();
         self.chain_responses
@@ -1503,6 +1551,60 @@ impl ProductStorage for CallbackPlatform {
     async fn clear(&self, key: String) -> Result<(), v01::HostLocalStorageReadError> {
         self.callbacks.local_storage_clear(key).map_err(Into::into)
     }
+
+    fn subscribe_storage(
+        &self,
+        key: Vec<u8>,
+    ) -> BoxStream<'static, Result<v01::HostLocalStorageChangeItem, v01::GenericError>> {
+        // Register the change receiver first so no event between the read and
+        // the subscription is lost, then read the current value lazily. The
+        // host pushes later changes through `notify_storage_changed`.
+        let key = String::from_utf8_lossy(&key).into_owned();
+        let rx = self.events.subscribe_storage_changes(key.clone());
+        let callbacks = self.callbacks.clone();
+        let current = async move {
+            callbacks
+                .local_storage_read(key)
+                .map(|value| v01::HostLocalStorageChangeItem { value })
+                .map_err(|error| {
+                    let error: v01::HostLocalStorageReadError = error.into();
+                    v01::GenericError {
+                        reason: error.to_string(),
+                    }
+                })
+        };
+        stream::once(current).chain(rx).boxed()
+    }
+}
+
+#[async_trait]
+impl ProductOperations for CallbackPlatform {
+    async fn begin_operation(
+        &self,
+        product: &ProductContext,
+        label: String,
+    ) -> Result<v01::HostWorkerBeginOperationResponse, v01::HostWorkerOperationError> {
+        self.callbacks
+            .begin_operation(product.product_id.clone(), label)
+            .await
+            .map(|id| v01::HostWorkerBeginOperationResponse { id })
+            .map_err(|error| v01::HostWorkerOperationError::Unknown {
+                reason: error.to_string(),
+            })
+    }
+
+    async fn end_operation(
+        &self,
+        product: &ProductContext,
+        id: u32,
+    ) -> Result<(), v01::HostWorkerOperationError> {
+        self.callbacks
+            .end_operation(product.product_id.clone(), id)
+            .await
+            .map_err(|error| v01::HostWorkerOperationError::Unknown {
+                reason: error.to_string(),
+            })
+    }
 }
 
 #[async_trait]
@@ -2074,6 +2176,16 @@ mod tests {
         fn local_storage_clear(&self, _key: String) -> Result<(), HostStorageError> {
             Ok(())
         }
+        async fn begin_operation(
+            &self,
+            _product_id: String,
+            _label: String,
+        ) -> Result<u32, HostRejection> {
+            Ok(1)
+        }
+        async fn end_operation(&self, _product_id: String, _id: u32) -> Result<(), HostRejection> {
+            Ok(())
+        }
     }
 
     impl NativeChatCallbacks for EventCallbacks {
@@ -3178,6 +3290,20 @@ mod tests {
             fn local_storage_clear(&self, _key: String) -> Result<(), HostStorageError> {
                 Ok(())
             }
+            async fn begin_operation(
+                &self,
+                _product_id: String,
+                _label: String,
+            ) -> Result<u32, HostRejection> {
+                Ok(1)
+            }
+            async fn end_operation(
+                &self,
+                _product_id: String,
+                _id: u32,
+            ) -> Result<(), HostRejection> {
+                Ok(())
+            }
         }
 
         let execution = native_product_execution(Arc::new(Noop), "dotli.dot");
@@ -3332,6 +3458,20 @@ mod tests {
             fn local_storage_clear(&self, _key: String) -> Result<(), HostStorageError> {
                 Ok(())
             }
+            async fn begin_operation(
+                &self,
+                _product_id: String,
+                _label: String,
+            ) -> Result<u32, HostRejection> {
+                Ok(1)
+            }
+            async fn end_operation(
+                &self,
+                _product_id: String,
+                _id: u32,
+            ) -> Result<(), HostRejection> {
+                Ok(())
+            }
         }
 
         let (release_tx, release_rx) = tokio::sync::mpsc::channel::<()>(1);
diff --git a/rust/crates/truapi-server/src/runtime/capabilities/platform.rs b/rust/crates/truapi-server/src/runtime/capabilities/platform.rs
index 1b6429812..96660cd02 100644
--- a/rust/crates/truapi-server/src/runtime/capabilities/platform.rs
+++ b/rust/crates/truapi-server/src/runtime/capabilities/platform.rs
@@ -2,11 +2,12 @@
 
 use futures::StreamExt;
 use tracing::{instrument, warn};
-use truapi::api::{LocalStorage, Locale, Notifications, Permissions, System, Theme};
+use truapi::api::{LocalStorage, Locale, Notifications, Permissions, System, Theme, Worker};
 use truapi::versioned::local_storage::{
-    HostLocalStorageClearError, HostLocalStorageClearRequest, HostLocalStorageClearResponse,
-    HostLocalStorageReadError, HostLocalStorageReadRequest, HostLocalStorageReadResponse,
-    HostLocalStorageWriteError, HostLocalStorageWriteRequest, HostLocalStorageWriteResponse,
+    HostLocalStorageChangeItem, HostLocalStorageClearError, HostLocalStorageClearRequest,
+    HostLocalStorageClearResponse, HostLocalStorageReadError, HostLocalStorageReadRequest,
+    HostLocalStorageReadResponse, HostLocalStorageSubscribeRequest, HostLocalStorageWriteError,
+    HostLocalStorageWriteRequest, HostLocalStorageWriteResponse,
 };
 use truapi::versioned::locale::HostLocaleSubscribeItem;
 use truapi::versioned::notifications::{
@@ -25,6 +26,11 @@ use truapi::versioned::system::{
     HostNavigateToResponse,
 };
 use truapi::versioned::theme::HostThemeSubscribeItem;
+use truapi::versioned::worker::{
+    HostWorkerBeginOperationError, HostWorkerBeginOperationRequest,
+    HostWorkerBeginOperationResponse, HostWorkerEndOperationError, HostWorkerEndOperationRequest,
+    HostWorkerEndOperationResponse,
+};
 use truapi::{CallContext, CallError, Subscription, v01};
 use truapi_platform::PermissionAuthorizationStatus;
 
@@ -190,8 +196,18 @@ impl LocalStorage for ProductRuntimeHost {
     ) -> Result<HostLocalStorageWriteResponse, CallError<HostLocalStorageWriteError>> {
         let HostLocalStorageWriteRequest::V1(v01::HostLocalStorageWriteRequest { key, value }) =
             request;
+        let storage_key = self.product_storage_key(key);
+        // Skip the store write when the value is byte-identical to what the key
+        // already holds, so a subscriber sees only real changes and no
+        // redundant persistence happens. A failed pre-read falls through to the
+        // write rather than blocking it.
+        if let Ok(Some(current)) = self.platform.read(storage_key.clone()).await
+            && current == value
+        {
+            return Ok(HostLocalStorageWriteResponse::V1);
+        }
         self.platform
-            .write(self.product_storage_key(key), value)
+            .write(storage_key, value)
             .await
             .map(|()| HostLocalStorageWriteResponse::V1)
             .map_err(|err| CallError::Domain(HostLocalStorageWriteError::V1(err)))
@@ -210,6 +226,64 @@ impl LocalStorage for ProductRuntimeHost {
             .map(|()| HostLocalStorageClearResponse::V1)
             .map_err(|err| CallError::Domain(HostLocalStorageClearError::V1(err)))
     }
+
+    #[instrument(skip_all, fields(runtime.method = "local_storage.subscribe"))]
+    async fn subscribe(
+        &self,
+        _cx: &CallContext,
+        request: HostLocalStorageSubscribeRequest,
+    ) -> Subscription<HostLocalStorageChangeItem> {
+        let HostLocalStorageSubscribeRequest::V1(v01::HostLocalStorageSubscribeRequest { key }) =
+            request;
+        Subscription::new(Box::pin(
+            self.platform
+                .subscribe_storage(self.product_storage_key(key).into_bytes())
+                .filter_map(|item| async {
+                    match item {
+                        Ok(item) => Some(HostLocalStorageChangeItem::V1(item)),
+                        Err(error) => {
+                            warn!(
+                                reason = %error.reason,
+                                "local storage subscription platform stream failed"
+                            );
+                            None
+                        }
+                    }
+                }),
+        ))
+    }
+}
+
+#[truapi::async_trait]
+impl Worker for ProductRuntimeHost {
+    #[instrument(skip_all, fields(runtime.method = "worker.begin_operation"))]
+    async fn begin_operation(
+        &self,
+        _cx: &CallContext,
+        request: HostWorkerBeginOperationRequest,
+    ) -> Result<HostWorkerBeginOperationResponse, CallError<HostWorkerBeginOperationError>> {
+        let HostWorkerBeginOperationRequest::V1(v01::HostWorkerBeginOperationRequest { label }) =
+            request;
+        self.platform
+            .begin_operation(&self.product, label.unwrap_or_default())
+            .await
+            .map(HostWorkerBeginOperationResponse::V1)
+            .map_err(|error| CallError::Domain(HostWorkerBeginOperationError::V1(error)))
+    }
+
+    #[instrument(skip_all, fields(runtime.method = "worker.end_operation"))]
+    async fn end_operation(
+        &self,
+        _cx: &CallContext,
+        request: HostWorkerEndOperationRequest,
+    ) -> Result<HostWorkerEndOperationResponse, CallError<HostWorkerEndOperationError>> {
+        let HostWorkerEndOperationRequest::V1(v01::HostWorkerEndOperationRequest { id }) = request;
+        self.platform
+            .end_operation(&self.product, id)
+            .await
+            .map(|()| HostWorkerEndOperationResponse::V1)
+            .map_err(|error| CallError::Domain(HostWorkerEndOperationError::V1(error)))
+    }
 }
 
 #[truapi::async_trait]
diff --git a/rust/crates/truapi-server/src/runtime/tests.rs b/rust/crates/truapi-server/src/runtime/tests.rs
index f900e821b..8585b24a7 100644
--- a/rust/crates/truapi-server/src/runtime/tests.rs
+++ b/rust/crates/truapi-server/src/runtime/tests.rs
@@ -5,8 +5,8 @@ use std::sync::atomic::Ordering;
 
 use parity_scale_codec::Encode;
 use truapi::api::{
-    Account, Chain, Entropy, Notifications, Permissions, Preimage, ResourceAllocation, Signing,
-    System, Theme,
+    Account, Chain, Entropy, LocalStorage, Notifications, Permissions, Preimage,
+    ResourceAllocation, Signing, System, Theme, Worker,
 };
 use truapi::versioned::account::{
     HostAccountConnectionStatusSubscribeItem, HostAccountCreateProofError,
@@ -23,6 +23,10 @@ use truapi::versioned::chain::{
 use truapi::versioned::entropy::{
     HostDeriveEntropyError, HostDeriveEntropyRequest, HostDeriveEntropyResponse,
 };
+use truapi::versioned::local_storage::{
+    HostLocalStorageChangeItem, HostLocalStorageClearRequest, HostLocalStorageSubscribeRequest,
+    HostLocalStorageWriteRequest,
+};
 use truapi::versioned::notifications::{
     HostPushNotificationCancelRequest, HostPushNotificationCancelResponse,
     HostPushNotificationRequest, HostPushNotificationResponse,
@@ -51,6 +55,10 @@ use truapi::versioned::system::{
     HostNavigateToResponse,
 };
 use truapi::versioned::theme::HostThemeSubscribeItem;
+use truapi::versioned::worker::{
+    HostWorkerBeginOperationRequest, HostWorkerBeginOperationResponse,
+    HostWorkerEndOperationRequest,
+};
 use truapi_platform::{AuthState, CoreStorageKey, PermissionAuthorizationRequest};
 
 use super::*;
@@ -1754,6 +1762,170 @@ fn preimage_lookup_forged_host_bytes_downgraded_to_miss() {
     );
 }
 
+fn storage_item(value: Option<&[u8]>) -> HostLocalStorageChangeItem {
+    HostLocalStorageChangeItem::V1(v01::HostLocalStorageChangeItem {
+        value: value.map(<[u8]>::to_vec),
+    })
+}
+
+fn subscribe_storage_key(
+    host: &ProductRuntimeHost,
+    key: &str,
+) -> Subscription<HostLocalStorageChangeItem> {
+    futures::executor::block_on(LocalStorage::subscribe(
+        host,
+        &CallContext::default(),
+        HostLocalStorageSubscribeRequest::V1(v01::HostLocalStorageSubscribeRequest {
+            key: key.to_string(),
+        }),
+    ))
+}
+
+fn write_storage_key(host: &ProductRuntimeHost, key: &str, value: &[u8]) {
+    futures::executor::block_on(host.write(
+        &CallContext::default(),
+        HostLocalStorageWriteRequest::V1(v01::HostLocalStorageWriteRequest {
+            key: key.to_string(),
+            value: value.to_vec(),
+        }),
+    ))
+    .expect("storage write");
+}
+
+#[test]
+fn local_storage_subscribe_sees_writes_and_clears_but_not_identical_rewrites() {
+    let platform = stub_platform();
+    let host = ProductRuntimeHost::new(
+        platform.clone(),
+        runtime_config("myapp.dot"),
+        test_spawner(),
+    );
+    let mut subscription = subscribe_storage_key(&host, "progress");
+    let next = |subscription: &mut Subscription<HostLocalStorageChangeItem>| {
+        futures::executor::block_on(subscription.next()).expect("storage item")
+    };
+
+    assert_eq!(
+        next(&mut subscription),
+        storage_item(None),
+        "the first item is the current value"
+    );
+
+    write_storage_key(&host, "progress", b"1");
+    assert_eq!(next(&mut subscription), storage_item(Some(b"1")));
+
+    write_storage_key(&host, "progress", b"1");
+    assert!(
+        futures::FutureExt::now_or_never(subscription.next()).is_none(),
+        "a byte-identical rewrite emits nothing"
+    );
+    assert_eq!(
+        platform
+            .local_storage_writes
+            .lock()
+            .expect("local storage writes mutex poisoned")
+            .len(),
+        1,
+        "the core skips the identical write before it reaches the platform"
+    );
+
+    futures::executor::block_on(host.clear(
+        &CallContext::default(),
+        HostLocalStorageClearRequest::V1(v01::HostLocalStorageClearRequest {
+            key: "progress".to_string(),
+        }),
+    ))
+    .expect("storage clear");
+    assert_eq!(next(&mut subscription), storage_item(None));
+}
+
+#[test]
+fn local_storage_subscribe_is_scoped_to_the_calling_product() {
+    let platform = stub_platform();
+    let mine = ProductRuntimeHost::new(
+        platform.clone(),
+        runtime_config("myapp.dot"),
+        test_spawner(),
+    );
+    let other = ProductRuntimeHost::new(platform, runtime_config("other.dot"), test_spawner());
+    write_storage_key(&mine, "shared", b"mine");
+
+    let mut mine_items = subscribe_storage_key(&mine, "shared");
+    let mut other_items = subscribe_storage_key(&other, "shared");
+    assert_eq!(
+        futures::executor::block_on(mine_items.next()),
+        Some(storage_item(Some(b"mine")))
+    );
+    assert_eq!(
+        futures::executor::block_on(other_items.next()),
+        Some(storage_item(None)),
+        "the same key name in another product is a different key"
+    );
+
+    write_storage_key(&mine, "shared", b"again");
+    assert_eq!(
+        futures::executor::block_on(mine_items.next()),
+        Some(storage_item(Some(b"again")))
+    );
+    assert!(
+        futures::FutureExt::now_or_never(other_items.next()).is_none(),
+        "another product's write never reaches this subscriber"
+    );
+}
+
+#[test]
+fn worker_operations_reach_the_platform_scoped_to_the_calling_product() {
+    let platform = stub_platform();
+    let host = ProductRuntimeHost::new(
+        platform.clone(),
+        runtime_config("myapp.dot"),
+        test_spawner(),
+    );
+    let cx = CallContext::default();
+    let begin = |label: Option<&str>| {
+        let HostWorkerBeginOperationResponse::V1(response) =
+            futures::executor::block_on(host.begin_operation(
+                &cx,
+                HostWorkerBeginOperationRequest::V1(v01::HostWorkerBeginOperationRequest {
+                    label: label.map(str::to_string),
+                }),
+            ))
+            .expect("begin operation");
+        response.id
+    };
+    let end = |id: u32| {
+        futures::executor::block_on(host.end_operation(
+            &cx,
+            HostWorkerEndOperationRequest::V1(v01::HostWorkerEndOperationRequest { id }),
+        ))
+        .expect("end operation")
+    };
+
+    assert_eq!(begin(Some("funding")), 1);
+    assert_eq!(begin(None), 2);
+    assert_eq!(
+        *platform
+            .begun_operations
+            .lock()
+            .expect("begun operations mutex poisoned"),
+        vec![
+            ("myapp.dot".to_string(), "funding".to_string()),
+            ("myapp.dot".to_string(), String::new()),
+        ],
+        "begin reaches the platform under the calling product; a missing label is empty"
+    );
+
+    end(1);
+    end(99);
+    assert_eq!(
+        *platform
+            .ended_operations
+            .lock()
+            .expect("ended operations mutex poisoned"),
+        vec![("myapp.dot".to_string(), 1), ("myapp.dot".to_string(), 99)],
+        "end reaches the platform under the calling product, unknown ids included"
+    );
+}
 #[test]
 fn theme_subscribe_maps_platform_values() {
     let host = ProductRuntimeHost::new_compat(stub_platform(), test_spawner());
diff --git a/rust/crates/truapi-server/src/test_support.rs b/rust/crates/truapi-server/src/test_support.rs
index 62484a53d..6cbe1f519 100644
--- a/rust/crates/truapi-server/src/test_support.rs
+++ b/rust/crates/truapi-server/src/test_support.rs
@@ -18,7 +18,7 @@ use crate::subscription::Spawner;
 use crate::subscription::thread_per_subscription_spawner;
 
 use futures::Stream;
-use futures::stream::{self, BoxStream};
+use futures::stream::{self, BoxStream, StreamExt};
 use parity_scale_codec::{Decode, Encode};
 use schnorrkel::{ExpansionMode, MiniSecretKey};
 use truapi::v01;
@@ -29,9 +29,10 @@ use truapi_platform::{
     CoreStorage as PlatformCoreStorage, CoreStorageKey, Features as PlatformFeatures, HostInfo,
     JsonRpcConnection, LocaleHost, Navigation as PlatformNavigation,
     Notifications as PlatformNotifications, PairingHostConfig, Permissions as PlatformPermissions,
-    PlatformInfo, PreimageHost, ProductContext, ProductStorage as PlatformProductStorage,
-    ProductSubtreeReview, ResourceAllocationReview, SignVrfReview, StatementStoreProductSignReview,
-    ThemeHost, UserConfirmation, UserConfirmationReview,
+    PlatformInfo, PreimageHost, ProductContext, ProductOperations as PlatformProductOperations,
+    ProductStorage as PlatformProductStorage, ProductSubtreeReview, ResourceAllocationReview,
+    SignVrfReview, StatementStoreProductSignReview, ThemeHost, UserConfirmation,
+    UserConfirmationReview,
 };
 use x25519_dalek::{PublicKey as X25519PublicKey, StaticSecret as X25519SecretKey};
 
@@ -158,10 +159,45 @@ pub(crate) struct StubPlatform {
     /// forged value to exercise the in-core integrity check.
     pub(crate) preimage_lookup_value: Option<Vec<u8>>,
     pub(crate) local_storage: Arc<Mutex<std::collections::HashMap<String, Vec<u8>>>>,
+    /// Every product storage write that reached the platform, in order, so a
+    /// test can see which writes the core skipped.
+    pub(crate) local_storage_writes: Arc<Mutex<Vec<StorageWrite>>>,
+    /// Open `subscribe_storage` streams by namespaced key; `write` and
+    /// `clear` push each change to the matching ones.
+    pub(crate) storage_subscribers: Arc<Mutex<Vec<(String, StorageChangeSender)>>>,
+    /// Every `begin_operation` as `(product_id, label)`, in order. The
+    /// returned id is the call's 1-based position.
+    pub(crate) begun_operations: Arc<Mutex<Vec<(String, String)>>>,
+    /// Every `end_operation` as `(product_id, id)`, in order.
+    pub(crate) ended_operations: Arc<Mutex<Vec<(String, u32)>>>,
     /// When set, product/core storage reads fail with this reason.
     pub(crate) local_storage_error: Option<&'static str>,
 }
 
+/// One product storage write as the platform saw it: namespaced key and bytes.
+pub(crate) type StorageWrite = (String, Vec<u8>);
+
+/// Sender side of one stubbed storage subscription.
+pub(crate) type StorageChangeSender = futures::channel::mpsc::UnboundedSender<
+    Result<v01::HostLocalStorageChangeItem, v01::GenericError>,
+>;
+
+impl StubPlatform {
+    fn push_storage_change(&self, key: &str, value: Option<Vec<u8>>) {
+        self.storage_subscribers
+            .lock()
+            .expect("storage subscribers mutex poisoned")
+            .retain(|(subscribed, tx)| {
+                subscribed != key
+                    || tx
+                        .unbounded_send(Ok(v01::HostLocalStorageChangeItem {
+                            value: value.clone(),
+                        }))
+                        .is_ok()
+            });
+    }
+}
+
 /// Scripted peer behavior for the recording connection's SSO exchange.
 #[derive(Clone)]
 pub(crate) enum SsoResponseScript {
@@ -776,10 +812,15 @@ impl PlatformProductStorage for StubPlatform {
                 reason: reason.to_string(),
             });
         }
+        self.local_storage_writes
+            .lock()
+            .expect("local storage writes mutex poisoned")
+            .push((key.clone(), value.clone()));
         self.local_storage
             .lock()
             .expect("local storage mutex poisoned")
-            .insert(key, value);
+            .insert(key.clone(), value.clone());
+        self.push_storage_change(&key, Some(value));
         Ok(())
     }
     async fn clear(&self, key: String) -> Result<(), v01::HostLocalStorageReadError> {
@@ -792,6 +833,60 @@ impl PlatformProductStorage for StubPlatform {
             .lock()
             .expect("local storage mutex poisoned")
             .remove(&key);
+        self.push_storage_change(&key, None);
+        Ok(())
+    }
+
+    fn subscribe_storage(
+        &self,
+        key: Vec<u8>,
+    ) -> BoxStream<'static, Result<v01::HostLocalStorageChangeItem, v01::GenericError>> {
+        let key = String::from_utf8_lossy(&key).into_owned();
+        let (tx, rx) = futures::channel::mpsc::unbounded();
+        self.storage_subscribers
+            .lock()
+            .expect("storage subscribers mutex poisoned")
+            .push((key.clone(), tx));
+        let value = self
+            .local_storage
+            .lock()
+            .expect("local storage mutex poisoned")
+            .get(&key)
+            .cloned();
+        Box::pin(
+            stream::once(async move { Ok(v01::HostLocalStorageChangeItem { value }) }).chain(rx),
+        )
+    }
+}
+
+#[truapi_platform::async_trait]
+impl PlatformProductOperations for StubPlatform {
+    async fn begin_operation(
+        &self,
+        product: &ProductContext,
+        label: String,
+    ) -> Result<v01::HostWorkerBeginOperationResponse, v01::HostWorkerOperationError> {
+        // The stub has no worker lifecycle to keep alive; it only records the
+        // call and hands back its position as the id.
+        let mut begun = self
+            .begun_operations
+            .lock()
+            .expect("begun operations mutex poisoned");
+        begun.push((product.product_id.clone(), label));
+        Ok(v01::HostWorkerBeginOperationResponse {
+            id: begun.len() as u32,
+        })
+    }
+
+    async fn end_operation(
+        &self,
+        product: &ProductContext,
+        id: u32,
+    ) -> Result<(), v01::HostWorkerOperationError> {
+        self.ended_operations
+            .lock()
+            .expect("ended operations mutex poisoned")
+            .push((product.product_id.clone(), id));
         Ok(())
     }
 }
diff --git a/rust/crates/truapi-server/tests/common/mod.rs b/rust/crates/truapi-server/tests/common/mod.rs
index 387155156..2ecae7774 100644
--- a/rust/crates/truapi-server/tests/common/mod.rs
+++ b/rust/crates/truapi-server/tests/common/mod.rs
@@ -8,8 +8,8 @@ use truapi::v01;
 use truapi_platform::{
     AuthPresenter, ChainProvider, CoreStorage, CoreStorageKey, Features, HostInfo,
     JsonRpcConnection, LocaleHost, Navigation, Notifications, PairingHostConfig, Permissions,
-    PlatformInfo, PreimageHost, ProductContext, ProductStorage, ThemeHost, UserConfirmation,
-    UserConfirmationReview,
+    PlatformInfo, PreimageHost, ProductContext, ProductOperations, ProductStorage, ThemeHost,
+    UserConfirmation, UserConfirmationReview,
 };
 use truapi_server::frame::ProtocolMessage;
 use truapi_server::transport::Transport;
@@ -86,6 +86,30 @@ impl ProductStorage for WireShapePlatform {
     async fn clear(&self, _key: String) -> Result<(), v01::HostLocalStorageReadError> {
         Ok(())
     }
+    fn subscribe_storage(
+        &self,
+        _key: Vec<u8>,
+    ) -> BoxStream<'static, Result<v01::HostLocalStorageChangeItem, v01::GenericError>> {
+        Box::pin(stream::empty())
+    }
+}
+
+#[truapi_platform::async_trait]
+impl ProductOperations for WireShapePlatform {
+    async fn begin_operation(
+        &self,
+        _product: &ProductContext,
+        _label: String,
+    ) -> Result<v01::HostWorkerBeginOperationResponse, v01::HostWorkerOperationError> {
+        Ok(v01::HostWorkerBeginOperationResponse { id: 1 })
+    }
+    async fn end_operation(
+        &self,
+        _product: &ProductContext,
+        _id: u32,
+    ) -> Result<(), v01::HostWorkerOperationError> {
+        Ok(())
+    }
 }
 
 #[truapi_platform::async_trait]
diff --git a/rust/crates/truapi/src/api.rs b/rust/crates/truapi/src/api.rs
index a4a7bb03f..8cd3f3f02 100644
--- a/rust/crates/truapi/src/api.rs
+++ b/rust/crates/truapi/src/api.rs
@@ -16,6 +16,7 @@ pub mod signing;
 pub mod statement_store;
 pub mod system;
 pub mod theme;
+pub mod worker;
 
 pub use account::Account;
 pub use chain::Chain;
@@ -33,6 +34,7 @@ pub use signing::Signing;
 pub use statement_store::StatementStore;
 pub use system::System;
 pub use theme::Theme;
+pub use worker::Worker;
 
 /// The unified TrUAPI contract.
 pub trait TrUApi:
@@ -52,6 +54,7 @@ pub trait TrUApi:
     + StatementStore
     + System
     + Theme
+    + Worker
     + Send
     + Sync
 {
@@ -74,6 +77,7 @@ impl<T> TrUApi for T where
         + StatementStore
         + System
         + Theme
+        + Worker
         + Send
         + Sync
 {
diff --git a/rust/crates/truapi/src/api/local_storage.rs b/rust/crates/truapi/src/api/local_storage.rs
index 5c2057858..2c882bd66 100644
--- a/rust/crates/truapi/src/api/local_storage.rs
+++ b/rust/crates/truapi/src/api/local_storage.rs
@@ -1,12 +1,13 @@
 //! Unified [`LocalStorage`] trait.
 
 use crate::versioned::local_storage::{
-    HostLocalStorageClearError, HostLocalStorageClearRequest, HostLocalStorageClearResponse,
-    HostLocalStorageReadError, HostLocalStorageReadRequest, HostLocalStorageReadResponse,
-    HostLocalStorageWriteError, HostLocalStorageWriteRequest, HostLocalStorageWriteResponse,
+    HostLocalStorageChangeItem, HostLocalStorageClearError, HostLocalStorageClearRequest,
+    HostLocalStorageClearResponse, HostLocalStorageReadError, HostLocalStorageReadRequest,
+    HostLocalStorageReadResponse, HostLocalStorageSubscribeRequest, HostLocalStorageWriteError,
+    HostLocalStorageWriteRequest, HostLocalStorageWriteResponse,
 };
 use crate::wire;
-use crate::{CallContext, CallError};
+use crate::{CallContext, CallError, Subscription};
 
 /// Local key/value storage scoped to the calling product.
 #[crate::async_trait]
@@ -55,4 +56,27 @@ pub trait LocalStorage: Send + Sync {
         cx: &CallContext,
         request: HostLocalStorageClearRequest,
     ) -> Result<HostLocalStorageClearResponse, CallError<HostLocalStorageClearError>>;
+
+    /// Subscribe to changes of one key in the product's own storage namespace.
+    ///
+    /// Emits the current value immediately, then one item per later write or
+    /// clear of the key by any of the product's runtimes. A write that leaves
+    /// the stored bytes unchanged emits nothing.
+    ///
+    /// ```ts
+    /// import { firstValueFrom, from } from "rxjs";
+    ///
+    /// const item = await firstValueFrom(
+    ///   from(truapi.localStorage.subscribe({ request: { key: "test-key" } })),
+    /// );
+    /// console.log("storage change received:", item);
+    /// ```
+    #[wire(start_id = 198)]
+    async fn subscribe(
+        &self,
+        _cx: &CallContext,
+        _request: HostLocalStorageSubscribeRequest,
+    ) -> Subscription<HostLocalStorageChangeItem> {
+        Subscription::empty()
+    }
 }
diff --git a/rust/crates/truapi/src/api/worker.rs b/rust/crates/truapi/src/api/worker.rs
new file mode 100644
index 000000000..15274866f
--- /dev/null
+++ b/rust/crates/truapi/src/api/worker.rs
@@ -0,0 +1,54 @@
+//! Unified [`Worker`] trait.
+
+use crate::versioned::worker::{
+    HostWorkerBeginOperationError, HostWorkerBeginOperationRequest,
+    HostWorkerBeginOperationResponse, HostWorkerEndOperationError, HostWorkerEndOperationRequest,
+    HostWorkerEndOperationResponse,
+};
+use crate::wire;
+use crate::{CallContext, CallError};
+
+/// Worker background-operation APIs.
+///
+/// A worker holds itself alive by keeping an operation open. The host keeps
+/// the product's worker running while it has at least one open operation.
+#[crate::service(required_execution = Worker)]
+#[crate::async_trait]
+pub trait Worker: Send + Sync {
+    /// Begin a pending operation. The worker is kept alive while it has at
+    /// least one open operation. Returns an id for `end_operation`.
+    ///
+    /// ```ts
+    /// const result = await truapi.worker.beginOperation({ label: "funding" });
+    /// assert(result.isOk(), "beginOperation failed:", result);
+    /// console.log("operation started:", result.value.id);
+    /// await truapi.worker.endOperation({ id: result.value.id });
+    /// ```
+    #[wire(request_id = 202)]
+    async fn begin_operation(
+        &self,
+        _cx: &CallContext,
+        _request: HostWorkerBeginOperationRequest,
+    ) -> Result<HostWorkerBeginOperationResponse, CallError<HostWorkerBeginOperationError>> {
+        Err(CallError::unavailable())
+    }
+
+    /// End a pending operation. Idempotent: an unknown or already-ended id
+    /// returns success.
+    ///
+    /// ```ts
+    /// const begun = await truapi.worker.beginOperation({});
+    /// assert(begun.isOk(), "beginOperation failed:", begun);
+    /// const result = await truapi.worker.endOperation({ id: begun.value.id });
+    /// assert(result.isOk(), "endOperation failed:", result);
+    /// console.log("operation ended");
+    /// ```
+    #[wire(request_id = 204)]
+    async fn end_operation(
+        &self,
+        _cx: &CallContext,
+        _request: HostWorkerEndOperationRequest,
+    ) -> Result<HostWorkerEndOperationResponse, CallError<HostWorkerEndOperationError>> {
+        Err(CallError::unavailable())
+    }
+}
diff --git a/rust/crates/truapi/src/lib.rs b/rust/crates/truapi/src/lib.rs
index 06d3557f4..e7e3066d1 100644
--- a/rust/crates/truapi/src/lib.rs
+++ b/rust/crates/truapi/src/lib.rs
@@ -55,13 +55,14 @@ pub mod latest {
         ChatActionLayout, ChatActions, ChatBotRegistrationStatus, ChatCustomMessage, ChatFile,
         ChatMedia, ChatMessageContent, ChatReaction, ChatRichText, ChatRoomRegistrationStatus,
         ContextualAlias, DerivationIndex, GenericError, HostPlatform, HostSignPayloadData,
-        NotificationId, OperationStartedResult, ProductAccountId, ProductProofContext, RawPayload,
-        RegisteredRingVrfKey, RemotePermission, RemoteStatementStoreCreateProofError,
-        RemoteStatementStoreCreateProofRequest, RemoteStatementStoreCreateProofResponse,
-        RemoteStatementStoreSubscribeItem, RemoteStatementStoreSubscribeRequest, RingLocation,
-        RingVrfKeyDisclosure, RingVrfPublicKey, RuntimeApi, RuntimeSpec, RuntimeType,
-        SignedStatement, Statement, StatementProof, StorageQueryItem, StorageQueryType,
-        StorageResultItem, ThemeName, ThemeVariant, TxPayloadExtension,
+        HostWorkerOperationError, NotificationId, OperationId, OperationStartedResult,
+        ProductAccountId, ProductProofContext, RawPayload, RegisteredRingVrfKey, RemotePermission,
+        RemoteStatementStoreCreateProofError, RemoteStatementStoreCreateProofRequest,
+        RemoteStatementStoreCreateProofResponse, RemoteStatementStoreSubscribeItem,
+        RemoteStatementStoreSubscribeRequest, RingLocation, RingVrfKeyDisclosure, RingVrfPublicKey,
+        RuntimeApi, RuntimeSpec, RuntimeType, SignedStatement, Statement, StatementProof,
+        StorageQueryItem, StorageQueryType, StorageResultItem, ThemeName, ThemeVariant,
+        TxPayloadExtension,
     };
 
     /// Latest payload type of a versioned envelope.
@@ -130,6 +131,9 @@ pub mod latest {
     /// Product context bound to the current host runtime.
     pub type HostGetProductContextResponse =
         LatestOf<versioned::system::HostGetProductContextResponse>;
+    /// Storage key change pushed to a subscriber.
+    pub type HostLocalStorageChangeItem =
+        LatestOf<versioned::local_storage::HostLocalStorageChangeItem>;
     /// Local storage operation error.
     pub type HostLocalStorageReadError =
         LatestOf<versioned::local_storage::HostLocalStorageReadError>;
@@ -167,6 +171,9 @@ pub mod latest {
         LatestOf<versioned::signing::HostSignRawWithLegacyAccountRequest>;
     /// Current host theme pushed to subscribers.
     pub type HostThemeSubscribeItem = LatestOf<versioned::theme::HostThemeSubscribeItem>;
+    /// Result of beginning a worker pending operation.
+    pub type HostWorkerBeginOperationResponse =
+        LatestOf<versioned::worker::HostWorkerBeginOperationResponse>;
     /// Transaction creation payload for a legacy account.
     pub type LegacyAccountTxPayload =
         LatestOf<versioned::signing::HostCreateTransactionWithLegacyAccountRequest>;
diff --git a/rust/crates/truapi/src/v01.rs b/rust/crates/truapi/src/v01.rs
index afb43cf7c..0cd859d48 100644
--- a/rust/crates/truapi/src/v01.rs
+++ b/rust/crates/truapi/src/v01.rs
@@ -18,6 +18,7 @@ mod statement_store;
 mod system;
 mod theme;
 mod transaction;
+mod worker;
 
 pub use account::*;
 pub use chain::*;
@@ -37,3 +38,4 @@ pub use statement_store::*;
 pub use system::*;
 pub use theme::*;
 pub use transaction::*;
+pub use worker::*;
diff --git a/rust/crates/truapi/src/v01/local_storage.rs b/rust/crates/truapi/src/v01/local_storage.rs
index 3bd56307b..d3adf7997 100644
--- a/rust/crates/truapi/src/v01/local_storage.rs
+++ b/rust/crates/truapi/src/v01/local_storage.rs
@@ -45,3 +45,18 @@ pub struct HostLocalStorageClearRequest {
     /// Storage key to clear.
     pub key: String,
 }
+
+/// Request to subscribe to changes of one local storage key.
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+pub struct HostLocalStorageSubscribeRequest {
+    /// Storage key to observe.
+    pub key: String,
+}
+
+/// A change to a subscribed storage key, pushed to the subscriber.
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+#[cfg_attr(feature = "uniffi", derive(uniffi::Record))]
+pub struct HostLocalStorageChangeItem {
+    /// Value after the change. `Some` on write, `None` after clear.
+    pub value: Option<Vec<u8>>,
+}
diff --git a/rust/crates/truapi/src/v01/worker.rs b/rust/crates/truapi/src/v01/worker.rs
new file mode 100644
index 000000000..f02b4db51
--- /dev/null
+++ b/rust/crates/truapi/src/v01/worker.rs
@@ -0,0 +1,42 @@
+use derive_more::Display;
+use parity_scale_codec::{Decode, Encode};
+
+/// Opaque host-assigned pending-operation identifier, unique per product.
+pub type OperationId = u32;
+
+/// Request to begin a pending operation.
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+pub struct HostWorkerBeginOperationRequest {
+    /// Optional label for host logs and UI.
+    pub label: Option<String>,
+}
+
+/// Response carrying the id of a newly begun operation.
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+pub struct HostWorkerBeginOperationResponse {
+    /// Id to pass to `end_operation`.
+    pub id: OperationId,
+}
+
+/// Request to end a pending operation.
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+pub struct HostWorkerEndOperationRequest {
+    /// Id returned by `begin_operation`.
+    pub id: OperationId,
+}
+
+/// Pending-operation error.
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode, Display)]
+#[cfg_attr(feature = "uniffi", derive(uniffi::Enum))]
+pub enum HostWorkerOperationError {
+    /// The product is already at the host's per-product limit of open
+    /// operations.
+    #[display("too many open operations")]
+    TooManyOpen,
+    /// Catch-all host failure.
+    #[display("{reason}")]
+    Unknown {
+        /// Human-readable failure reason.
+        reason: String,
+    },
+}
diff --git a/rust/crates/truapi/src/versioned.rs b/rust/crates/truapi/src/versioned.rs
index 4d5e37c26..938b63fd3 100644
--- a/rust/crates/truapi/src/versioned.rs
+++ b/rust/crates/truapi/src/versioned.rs
@@ -46,6 +46,7 @@ pub mod signing;
 pub mod statement_store;
 pub mod system;
 pub mod theme;
+pub mod worker;
 
 #[cfg(test)]
 mod tests {
diff --git a/rust/crates/truapi/src/versioned/local_storage.rs b/rust/crates/truapi/src/versioned/local_storage.rs
index 708eb4c7b..9760cb43d 100644
--- a/rust/crates/truapi/src/versioned/local_storage.rs
+++ b/rust/crates/truapi/src/versioned/local_storage.rs
@@ -12,4 +12,6 @@ truapi_macros::versioned_type! {
     pub enum HostLocalStorageClearRequest { V1 => v01::HostLocalStorageClearRequest }
     pub enum HostLocalStorageClearResponse { V1 }
     pub enum HostLocalStorageClearError { V1 => v01::HostLocalStorageReadError }
+    pub enum HostLocalStorageSubscribeRequest { V1 => v01::HostLocalStorageSubscribeRequest }
+    pub enum HostLocalStorageChangeItem { V1 => v01::HostLocalStorageChangeItem }
 }
diff --git a/rust/crates/truapi/src/versioned/worker.rs b/rust/crates/truapi/src/versioned/worker.rs
new file mode 100644
index 000000000..9e2045a25
--- /dev/null
+++ b/rust/crates/truapi/src/versioned/worker.rs
@@ -0,0 +1,12 @@
+//! Versioned wrappers for [`Worker`](crate::api::Worker) methods.
+
+use crate::v01;
+
+truapi_macros::versioned_type! {
+    pub enum HostWorkerBeginOperationRequest { V1 => v01::HostWorkerBeginOperationRequest }
+    pub enum HostWorkerBeginOperationResponse { V1 => v01::HostWorkerBeginOperationResponse }
+    pub enum HostWorkerBeginOperationError { V1 => v01::HostWorkerOperationError }
+    pub enum HostWorkerEndOperationRequest { V1 => v01::HostWorkerEndOperationRequest }
+    pub enum HostWorkerEndOperationResponse { V1 }
+    pub enum HostWorkerEndOperationError { V1 => v01::HostWorkerOperationError }
+}