feat: KV store support (duroxide 0.1.24)

- OrchestrationContext: set_value, get_value, clear_value, clear_all_values, get_value_from_instance
- Client: get_value, wait_for_value
- Constants: MAX_KV_KEYS, MAX_KV_VALUE_BYTES
- 2 KV e2e tests (request/response, cross-orchestration read)
- Updated README, CHANGELOG, user-guide, architecture docs
- Remove [patch.crates-io] — using published duroxide-pg 0.1.25

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: affandar

CHANGELOG.md

@@ -5,6 +5,15 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.1.14] - 2026-03-13
+
+- **KV store support:** Durable key-value store for per-instance state
+  - `OrchestrationContext`: `set_value()`, `get_value()`, `clear_value()`, `clear_all_values()`
+  - `Client`: `get_value()`, `wait_for_value()`
+  - Constants: `MAX_KV_KEYS`, `MAX_KV_VALUE_BYTES`
+- Added KV e2e tests (`test_kv_store.py`)
+- Bumped duroxide to 0.1.24, duroxide-pg to 0.1.25
+
 ## [0.1.13] - 2026-03-08
 
 ### Added

Cargo.toml

@@ -1,15 +1,15 @@
 [package]
 name = "duroxide-python"
-version = "0.1.13"
+version = "0.1.14"
 edition = "2021"
 
 [lib]
 name = "duroxide_python"
 crate-type = ["cdylib"]
 
 [dependencies]
-duroxide = { version = "0.1.23", features = ["sqlite"] }
-duroxide-pg = { version = "0.1.24" }
+duroxide = { version = "0.1.24", features = ["sqlite"] }
+duroxide-pg = { version = "0.1.25" }
 pyo3 = { version = "0.23", features = ["extension-module"] }
 async-trait = "0.1"
 tokio = { version = "1", features = ["full"] }

README.md

@@ -16,9 +16,10 @@ Write durable workflows as Python generators. The Rust runtime handles replay, p
 - **Deterministic replay** — safe resume after crashes
 - **SQLite & PostgreSQL** — pluggable storage providers
 - **Custom Status** — `ctx.set_custom_status()` / `ctx.reset_custom_status()` for orchestration progress reporting, `client.wait_for_status_change()` for efficient polling
+- **KV Store** — durable per-instance state via `ctx.set_value()` / `ctx.get_value()` / `ctx.clear_value()` / `ctx.clear_all_values()`, plus `client.get_value()` / `client.wait_for_value()`
 - **Event Queues** — `ctx.dequeue_event(queue_name)` for FIFO mailbox-style message passing, `client.enqueue_event()` to send messages
 - **Retry on Session** — `ctx.schedule_activity_with_retry_on_session()` for retry with session affinity
-- **Tag Routing** — worker tags for activity affinity (`MAX_WORKER_TAGS=5`, `MAX_TAG_NAME_BYTES=256`)
+- **Tag Routing** — worker tags for activity affinity (`MAX_WORKER_TAGS=5`, `MAX_TAG_NAME_BYTES=256`, `MAX_KV_KEYS=10`, `MAX_KV_VALUE_BYTES=16384`)
 - **Admin APIs** — instance management, metrics, pruning
 - **Activity client access** — `ctx.get_client()` lets activities start new orchestrations
 - **Runtime metrics** — `metrics_snapshot()` for orchestration/activity counters
@@ -178,6 +179,25 @@ if status:
     print(status.custom_status_version)  # monotonically increasing counter
 ```
 
+## KV Store
+
+Durable per-instance key-value state for orchestration coordination and request/response patterns:
+
+```python
+@runtime.register_orchestration("KvWorkflow")
+def kv_workflow(ctx, input):
+    ctx.set_value("status", "running")
+    result = yield ctx.schedule_activity("Compute", input)
+    ctx.set_value("result", str(result))
+    return result
+
+# External reads
+status = client.wait_for_value("instance-1", "status", 10000)
+result = client.get_value("instance-1", "result")
+```
+
+KV entries are scoped to a single orchestration instance and remain readable after completion until the instance is deleted or pruned.
+
 ## Event Queues
 
 Persistent FIFO message passing between clients and orchestrations:

docs/architecture.md

@@ -352,6 +352,38 @@ client.wait_for_status_change(id, last_version, poll_ms, timeout_ms)
 
 The `custom_status_version` is a monotonically increasing counter incremented on every `set_custom_status()` call. Clients pass their last-seen version to avoid redundant reads.
 
+## KV Store Data Path
+
+KV operations are fire-and-forget orchestration context calls backed by provider state for the current instance.
+
+```
+Python                              Rust (PyO3)                        Provider (DB)
+──────                              ───────────                        ─────────────
+ctx.set_value("status", "ready")
+  │
+  └─► orchestration_set_value(instance_id, "status", "ready")
+        │
+        └─► ORCHESTRATION_CTXS.get(instance_id)
+              │
+              └─► ctx.set_value("status", "ready")
+                    │
+                    └─► provider.upsert KV row for (instance_id, key)
+
+client.get_value(id, "status")
+  │
+  └─► py.allow_threads(|| TOKIO_RT.block_on(client.get_value(id, "status")))
+        │
+        └─► provider.get_value(id, "status")
+
+client.wait_for_value(id, "status", timeout_ms)
+  │
+  └─► py.allow_threads(|| TOKIO_RT.block_on(client.wait_for_value(...)))
+        │
+        └─► repeated provider.get_value(...) polling until key exists or timeout
+```
+
+Cross-orchestration reads use the built-in `__duroxide_syscall:get_kv_value` activity. In Python this is exposed as `ctx.get_value_from_instance(instance_id, key)` and replays like any other scheduled activity.
+
 ## Event Queue Data Flow
 
 Event queues provide persistent FIFO message passing between external clients and orchestrations. Unlike `wait_for_event()` which matches a single named event, event queues support multiple messages on named queues with guaranteed ordering. Messages survive `continue_as_new`.

docs/user-guide.md

@@ -380,6 +380,38 @@ while True:
 - `result.custom_status` — the custom status string, or `None` if not set
 - `result.custom_status_version` — monotonically increasing version counter
 
+## KV Store — Durable Per-Instance State
+
+KV entries are durable metadata scoped to a single orchestration instance. They can be updated from inside the orchestration without yielding and read externally through the client.
+
+```python
+@runtime.register_orchestration("RequestServer")
+def request_server(ctx, input):
+    ctx.set_value("status", "ready")
+    request = yield ctx.wait_for_event("request")
+    response = request["command"][::-1]
+    ctx.set_value(f"response:{request['op_id']}", response)
+    return "done"
+
+status = client.wait_for_value("server-1", "status", 10000)
+response = client.get_value("server-1", "response:op-1")
+```
+
+**OrchestrationContext methods:**
+- `ctx.set_value(key, value)` — set or overwrite a key
+- `ctx.get_value(key)` — read the current value for a key in the active instance
+- `ctx.clear_value(key)` — remove a single key
+- `ctx.clear_all_values()` — clear all keys for the active instance
+- `ctx.get_value_from_instance(instance_id, key)` — read another instance's KV via the built-in syscall activity
+
+**Client methods:**
+- `client.get_value(instance_id, key)` — read a key immediately
+- `client.wait_for_value(instance_id, key, timeout_ms)` — block until the key exists or timeout
+
+**Limits:**
+- `MAX_KV_KEYS = 10`
+- `MAX_KV_VALUE_BYTES = 16384`
+
 ## Event Queues — Persistent FIFO Message Passing
 
 Event queues provide durable, ordered message passing between external clients and orchestrations. Unlike `wait_for_event()` which waits for a single named event, event queues support FIFO ordering with multiple messages on named queues. Messages survive `continue_as_new`.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "maturin"
 
 [project]
 name = "duroxide"
-version = "0.1.13"
+version = "0.1.14"
 description = "Python SDK for the Duroxide durable execution runtime"
 readme = "README.md"
 license = { text = "MIT" }

python/duroxide/__init__.py

@@ -183,6 +183,21 @@ def wait_for_status_change(
         )
         return _parse_status(result)
 
+    def get_value(self, instance_id: str, key: str) -> "Optional[str]":
+        """Read a single KV entry for the given instance."""
+        return self._native.get_value(instance_id, key)
+
+    def wait_for_value(self, instance_id: str, key: str, timeout_ms: int = 30000) -> str:
+        """Wait for a KV value to be set on an instance."""
+        try:
+            return self._native.wait_for_value(instance_id, key, timeout_ms)
+        except RuntimeError as exc:
+            if "timed out" in str(exc).lower():
+                raise TimeoutError(
+                    f"Timed out waiting for value '{key}' on instance '{instance_id}'"
+                ) from exc
+            raise
+
     def cancel_instance(self, instance_id: str, reason: str = None):
         self._native.cancel_instance(instance_id, reason)
 
@@ -424,6 +439,8 @@ def metrics_snapshot(self):
 # Tag limits
 MAX_WORKER_TAGS = 5
 MAX_TAG_NAME_BYTES = 256
+MAX_KV_KEYS = 10
+MAX_KV_VALUE_BYTES = 16384
 
 
 class TagFilter:
@@ -478,4 +495,6 @@ def default_and(tags: list) -> str:
     "parse_result",
     "MAX_WORKER_TAGS",
     "MAX_TAG_NAME_BYTES",
+    "MAX_KV_KEYS",
+    "MAX_KV_VALUE_BYTES",
 ]

python/duroxide/context.py

@@ -16,6 +16,10 @@
     orchestration_set_custom_status,
     orchestration_reset_custom_status,
     orchestration_get_custom_status,
+    orchestration_set_value,
+    orchestration_get_value,
+    orchestration_clear_value,
+    orchestration_clear_all_values,
     activity_trace_log,
     activity_is_cancelled,
     activity_tag,
@@ -455,6 +459,30 @@ def get_custom_status(self) -> Optional[str]:
         """
         return orchestration_get_custom_status(self.instance_id)
 
+    def set_value(self, key: str, value: str):
+        """Set a key-value pair scoped to this orchestration instance."""
+        orchestration_set_value(self.instance_id, str(key), str(value))
+
+    def get_value(self, key: str) -> Optional[str]:
+        """Get the current value for a key. Returns None if not set."""
+        return orchestration_get_value(self.instance_id, str(key))
+
+    def clear_value(self, key: str):
+        """Remove a single key from the KV store."""
+        orchestration_clear_value(self.instance_id, str(key))
+
+    def clear_all_values(self):
+        """Clear ALL key-value pairs for this orchestration instance."""
+        orchestration_clear_all_values(self.instance_id)
+
+    def get_value_from_instance(self, instance_id: str, key: str) -> ScheduledTask:
+        """Read a KV value from another orchestration instance via the built-in syscall activity."""
+        return ScheduledTask({
+            "type": "activity",
+            "name": "__duroxide_syscall:get_kv_value",
+            "input": json.dumps({"instance_id": instance_id, "key": key}),
+        })
+
     # ─── Logging (fire-and-forget, delegates to Rust ctx.trace()) ───
 
     def trace_info(self, message: str):

src/client.rs

@@ -271,6 +271,41 @@ impl PyClient {
         .map_err(|e: String| pyo3::exceptions::PyRuntimeError::new_err(e))
     }
 
+    /// Read a single KV entry for an orchestration instance.
+    fn get_value(
+        &self,
+        py: Python<'_>,
+        instance_id: String,
+        key: String,
+    ) -> PyResult<Option<String>> {
+        let client = self.inner.clone();
+        py.allow_threads(|| {
+            TOKIO_RT.block_on(async { client.get_value(&instance_id, &key).await.map_err(|e| format!("{e}")) })
+        })
+        .map_err(|e: String| pyo3::exceptions::PyRuntimeError::new_err(e))
+    }
+
+    /// Wait for a KV entry to become available on an orchestration instance.
+    fn wait_for_value(
+        &self,
+        py: Python<'_>,
+        instance_id: String,
+        key: String,
+        timeout_ms: u64,
+    ) -> PyResult<String> {
+        let client = self.inner.clone();
+        let timeout = Duration::from_millis(timeout_ms);
+        py.allow_threads(|| {
+            TOKIO_RT.block_on(async {
+                client
+                    .wait_for_value(&instance_id, &key, timeout)
+                    .await
+                    .map_err(|e| format!("{e}"))
+            })
+        })
+        .map_err(|e: String| pyo3::exceptions::PyRuntimeError::new_err(e))
+    }
+
     /// Get system metrics (if provider supports management).
     fn get_system_metrics(&self, py: Python<'_>) -> PyResult<PySystemMetrics> {
         let client = self.inner.clone();

src/handlers.rs

@@ -139,6 +139,36 @@ pub fn orchestration_get_custom_status(instance_id: &str) -> Option<String> {
     map.get(instance_id).and_then(|ctx| ctx.get_custom_status())
 }
 
+/// Called from Python to set a KV value on the OrchestrationContext.
+pub fn orchestration_set_value(instance_id: &str, key: &str, value: &str) {
+    let map = ORCHESTRATION_CTXS.lock();
+    if let Some(ctx) = map.get(instance_id) {
+        ctx.set_value(key, value);
+    }
+}
+
+/// Called from Python to read a KV value from the OrchestrationContext.
+pub fn orchestration_get_value(instance_id: &str, key: &str) -> Option<String> {
+    let map = ORCHESTRATION_CTXS.lock();
+    map.get(instance_id).and_then(|ctx| ctx.get_value(key))
+}
+
+/// Called from Python to clear a single KV value on the OrchestrationContext.
+pub fn orchestration_clear_value(instance_id: &str, key: &str) {
+    let map = ORCHESTRATION_CTXS.lock();
+    if let Some(ctx) = map.get(instance_id) {
+        ctx.clear_value(key);
+    }
+}
+
+/// Called from Python to clear all KV values on the OrchestrationContext.
+pub fn orchestration_clear_all_values(instance_id: &str) {
+    let map = ORCHESTRATION_CTXS.lock();
+    if let Some(ctx) = map.get(instance_id) {
+        ctx.clear_all_values();
+    }
+}
+
 // ─── Activity Bridge ─────────────────────────────────────────────
 
 /// Wraps a Python callable as a Rust ActivityHandler.