Add async runtime and concurrent vault support

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased
+
+- Added an `AsyncRuntime` façade and reusable thread-safe `CancellationToken` without changing the synchronous runtime API.
+- Added cancellable asyncio RTK command rewriting and aggressive reads that kill and reap subprocesses on task or token cancellation.
+- Enabled WAL-backed concurrent artifact-vault access and added deterministic thread/process contention and lease coverage.
+
 ## 0.3.1 - 2026-08-24
 
 - Kept collapsed-turn summaries inside the first retained user message so strict provider role sequencing remains valid.

README.md

@@ -29,6 +29,8 @@ The engine has four cooperating reduction paths:
 
 The reduction core is not intrinsically tied to Hermes: it operates on Python dictionaries, strings, stable request/session identifiers, and a local SQLite vault. The repository includes a turnkey Hermes plugin because Hermes exposes the required lifecycle hooks. Other agent runtimes need a small adapter that presents the same boundaries; they do not need a fork of the reduction engine.
 
+Async agent frameworks can use the included `AsyncRuntime` façade. It keeps provider loops responsive by moving compiler, vault, and telemetry work to an executor, propagates task or token cancellation, and uses native cancellable subprocess paths for RTK command rewriting and aggressive reads. The synchronous `Runtime` API remains unchanged.
+
 It does **not** replace the host's context engine, memory system, transcript store, or provider client. It does not add an MCP server or standing prompt text. If storage, recovery, middleware, or compilation is unavailable or unsafe, the host receives the original request or result unchanged.
 
 ## What it does
@@ -228,6 +230,39 @@ def reduce_provider_request(request, *, session_id, request_id):
 
 An adapter must preserve four contracts: stable request/session identity, original-object immutability, pass-through on `None` or error, and model access to `artifact_get`. The `Runtime` surface is usable today; framework-specific one-command adapters beyond Hermes are not yet shipped.
 
+### Async runtimes, cancellation, and concurrency
+
+Wrap the same synchronous runtime when the host owns an asyncio event loop:
+
+```python
+from rtk_hermes_plus import AsyncRuntime, CancellationToken, Runtime
+
+terminator = AsyncRuntime(Runtime(config, profile_name="my-agent"))
+cancellation = CancellationToken()
+
+reduced = await terminator.transform_tool_result(
+    tool_name="search_files",
+    args={"pattern": "ContextEngine"},
+    result=large_result,
+    session_id=session_id,
+    tool_call_id=call_id,
+    cancellation=cancellation,
+)
+
+compiled = await terminator.llm_request_middleware(
+    request=provider_request,
+    session_id=session_id,
+    request_id=request_id,
+    cancellation=cancellation,
+)
+```
+
+`AsyncRuntime` mirrors the adapter-facing tool, result, request, recovery, and session methods. Pass a custom `concurrent.futures.Executor` to `AsyncRuntime(runtime, executor=...)` when the host needs a dedicated worker pool.
+
+Cancelling the awaiting asyncio task, or calling the thread-safe `cancellation.cancel()`, raises `asyncio.CancelledError` at the adapter boundary. Active RTK subprocesses are killed and reaped. Compiler and SQLite operations already running in an executor remain atomic and may finish in that worker after the caller has stopped waiting; their result is discarded. Token Terminator does not intercept or buffer provider response streams, so adapters compile immediately before dispatch and leave streaming responses under host control.
+
+The artifact vault enables SQLite WAL mode and uses `synchronous=NORMAL`, a ten-second busy timeout, short-lived connections, and `BEGIN IMMEDIATE` writes. Independent runtime instances and agent processes may share one local vault while preserving content deduplication, lease limits, and observation provenance. Keep the database on a local filesystem: SQLite WAL is not a network-filesystem coordination protocol.
+
 ## Exact recovery
 
 Compressed results and request receipts contain an artifact identifier. The model can recover an exact page through the registered tool:
@@ -302,7 +337,7 @@ A valid comparison requires separate fresh sessions with stable modes, the same
 ## Security and privacy
 
 - Exact raw artifacts and their private provenance are stored locally because recovery is part of the product contract.
-- The vault enforces per-artifact and total-capacity limits, SQLite foreign keys, busy timeouts, schema-version checks, and short-lived transactions.
+- The vault enforces per-artifact and total-capacity limits, SQLite WAL, foreign keys, busy timeouts, schema-version checks, short-lived transactions, and serialized writes.
 - POSIX storage uses `0700` parent directories and `0600` databases. Windows storage inherits the user's profile ACLs.
 - RTK subprocesses use argument arrays with `shell=False`.
 - Remote terminal backends are disabled by default.

src/rtk_hermes_plus/__init__.py

@@ -6,6 +6,14 @@
 """
 
 from ._version import __version__
+from .async_runtime import AsyncRuntime
+from .cancellation import CancellationToken
 from .plugin import Runtime, register
 
-__all__ = ["Runtime", "__version__", "register"]
+__all__ = [
+    "AsyncRuntime",
+    "CancellationToken",
+    "Runtime",
+    "__version__",
+    "register",
+]

src/rtk_hermes_plus/async_runtime.py

@@ -0,0 +1,242 @@
+from __future__ import annotations
+
+import asyncio
+from collections.abc import Callable
+from concurrent.futures import Executor
+from contextlib import suppress
+from functools import partial
+from typing import Any, TypeVar
+
+from .cancellation import CancellationToken
+from .compress import compact_text
+from .plugin import Runtime
+
+_T = TypeVar("_T")
+
+
+class AsyncRuntime:
+    """Non-blocking façade over :class:`Runtime` for async agent frameworks.
+
+    CPU-bound and SQLite-backed sync operations run in an executor. Cancelling
+    the awaiting task propagates immediately to the caller; already-running
+    sync work remains atomic and may finish in its worker thread. RTK command
+    rewrites use a native asyncio subprocess path and are killed and reaped on
+    cancellation.
+    """
+
+    def __init__(self, runtime: Runtime, *, executor: Executor | None = None) -> None:
+        self.runtime = runtime
+        self.executor = executor
+
+    def _raise_if_cancelled(self, cancellation: CancellationToken | None) -> None:
+        if cancellation is not None and cancellation.cancelled:
+            self.runtime.metrics.add("async_cancelled")
+            raise asyncio.CancelledError
+
+    async def _run_sync(
+        self,
+        function: Callable[..., _T],
+        /,
+        *args: Any,
+        cancellation: CancellationToken | None = None,
+        **kwargs: Any,
+    ) -> _T:
+        self._raise_if_cancelled(cancellation)
+        loop = asyncio.get_running_loop()
+        future = loop.run_in_executor(self.executor, partial(function, *args, **kwargs))
+        cancellation_waiter: asyncio.Task[None] | None = None
+        try:
+            if cancellation is None:
+                return await future
+            cancellation_waiter = asyncio.create_task(cancellation.wait())
+            done, _pending = await asyncio.wait(
+                (future, cancellation_waiter),
+                return_when=asyncio.FIRST_COMPLETED,
+            )
+            if future in done:
+                return future.result()
+            if cancellation_waiter in done:
+                future.cancel()
+                raise asyncio.CancelledError
+            raise RuntimeError("async runtime wait completed without a result")
+        except asyncio.CancelledError:
+            if cancellation is not None:
+                cancellation.cancel()
+            future.cancel()
+            self.runtime.metrics.add("async_cancelled")
+            raise
+        finally:
+            if cancellation_waiter is not None:
+                cancellation_waiter.cancel()
+                with suppress(asyncio.CancelledError):
+                    await cancellation_waiter
+
+    async def tool_request_middleware(
+        self,
+        *,
+        tool_name: str,
+        args: dict,
+        cancellation: CancellationToken | None = None,
+        **kwargs: Any,
+    ) -> dict | None:
+        self._raise_if_cancelled(cancellation)
+        if tool_name != "terminal" or not isinstance(args, dict):
+            return None
+        prepared = await self._run_sync(
+            self.runtime._prepare_rewrite,
+            args,
+            cancellation=cancellation,
+        )
+        if prepared is None:
+            return None
+        command, cwd = prepared
+        try:
+            result = await self.runtime.rewriter.rewrite_async(
+                command,
+                cwd=cwd,
+                cancellation=cancellation,
+            )
+        except asyncio.CancelledError:
+            if cancellation is not None:
+                cancellation.cancel()
+            self.runtime.metrics.add("async_cancelled")
+            raise
+        rewritten = self.runtime._apply_rewrite_result(args, result)
+        if rewritten is None:
+            return None
+        await self._run_sync(
+            self.runtime._record_rewrite,
+            session_id=str(kwargs.get("session_id") or ""),
+            turn_id=str(kwargs.get("turn_id") or ""),
+            cancellation=cancellation,
+        )
+        return {
+            "args": rewritten,
+            "source": "token-terminator",
+            "reason": "strict token reduction",
+        }
+
+    async def observe_tool_call(
+        self, *, cancellation: CancellationToken | None = None, **kwargs: Any
+    ) -> None:
+        await self._run_sync(
+            self.runtime.observe_tool_call, cancellation=cancellation, **kwargs
+        )
+
+    async def transform_tool_result(
+        self,
+        *,
+        tool_name: str,
+        args: dict,
+        result: str,
+        cancellation: CancellationToken | None = None,
+        **kwargs: Any,
+    ) -> Any:
+        self._raise_if_cancelled(cancellation)
+        compressor = self.runtime.compressor
+        if not compressor._eligible(tool_name=tool_name, args=args, result=result):
+            return None
+        compressor.metrics.add("native_attempted")
+        compact = None
+        try:
+            if tool_name == "read_file":
+                compact = await compressor._rtk_read_async(
+                    args,
+                    cancellation=cancellation,
+                )
+        except asyncio.CancelledError:
+            if cancellation is not None:
+                cancellation.cancel()
+            self.runtime.metrics.add("async_cancelled")
+            raise
+        if compact is None:
+            compact = await self._run_sync(
+                compact_text,
+                result,
+                compressor.config.native_max_chars,
+                cancellation=cancellation,
+            )
+        transformed = await self._run_sync(
+            compressor._finalize_transform,
+            tool_name=tool_name,
+            args=args,
+            result=result,
+            compact=compact,
+            session_id=str(kwargs.get("session_id") or ""),
+            tool_call_id=str(kwargs.get("tool_call_id") or ""),
+            cancellation=cancellation,
+        )
+        if transformed is not None:
+            await self._run_sync(
+                self.runtime._record_native,
+                session_id=str(kwargs.get("session_id") or ""),
+                turn_id=str(kwargs.get("turn_id") or ""),
+                raw_chars=len(result),
+                output_chars=len(transformed),
+                cancellation=cancellation,
+            )
+        return transformed
+
+    async def post_tool_call(
+        self, *, cancellation: CancellationToken | None = None, **kwargs: Any
+    ) -> None:
+        await self._run_sync(
+            self.runtime.post_tool_call, cancellation=cancellation, **kwargs
+        )
+
+    async def llm_request_middleware(
+        self, *, cancellation: CancellationToken | None = None, **kwargs: Any
+    ) -> dict | None:
+        return await self._run_sync(
+            self.runtime.llm_request_middleware, cancellation=cancellation, **kwargs
+        )
+
+    async def on_session_start(
+        self,
+        *,
+        session_id: str,
+        cancellation: CancellationToken | None = None,
+    ) -> None:
+        await self._run_sync(
+            self.runtime.on_session_start,
+            session_id=session_id,
+            cancellation=cancellation,
+        )
+
+    async def pre_llm_call(
+        self, *, cancellation: CancellationToken | None = None, **kwargs: Any
+    ) -> None:
+        await self._run_sync(
+            self.runtime.pre_llm_call, cancellation=cancellation, **kwargs
+        )
+
+    async def on_session_end(
+        self, *, cancellation: CancellationToken | None = None, **kwargs: Any
+    ) -> None:
+        await self._run_sync(
+            self.runtime.on_session_end, cancellation=cancellation, **kwargs
+        )
+
+    async def on_session_finalize(
+        self,
+        *,
+        session_id: str,
+        cancellation: CancellationToken | None = None,
+    ) -> None:
+        await self._run_sync(
+            self.runtime.on_session_finalize,
+            session_id=session_id,
+            cancellation=cancellation,
+        )
+
+    async def tool(
+        self, *, cancellation: CancellationToken | None = None, **kwargs: Any
+    ) -> str:
+        return await self._run_sync(
+            self.runtime.tool, cancellation=cancellation, **kwargs
+        )
+
+    async def status(
+        self, *, cancellation: CancellationToken | None = None
+    ) -> dict[str, Any]:
+        return await self._run_sync(self.runtime.status, cancellation=cancellation)

src/rtk_hermes_plus/cancellation.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+import asyncio
+import threading
+
+
+class CancellationToken:
+    """Thread-safe cancellation signal for async runtime adapters.
+
+    Cancelling a token stops cancellable async work such as RTK subprocesses.
+    SQLite and compiler operations already running in a worker thread remain
+    atomic and may finish after the awaiting task has been cancelled.
+    """
+
+    def __init__(self) -> None:
+        self._cancelled = False
+        self._lock = threading.Lock()
+        self._waiters: set[tuple[asyncio.AbstractEventLoop, asyncio.Future[None]]] = (
+            set()
+        )
+
+    @property
+    def cancelled(self) -> bool:
+        with self._lock:
+            return self._cancelled
+
+    @staticmethod
+    def _resolve_waiter(waiter: asyncio.Future[None]) -> None:
+        if not waiter.done():
+            waiter.set_result(None)
+
+    def cancel(self) -> None:
+        with self._lock:
+            if self._cancelled:
+                return
+            self._cancelled = True
+            waiters = tuple(self._waiters)
+            self._waiters.clear()
+        for loop, waiter in waiters:
+            try:
+                loop.call_soon_threadsafe(self._resolve_waiter, waiter)
+            except RuntimeError:
+                # The owning loop closed while cancellation crossed threads.
+                # No active coroutine remains to wake in that loop.
+                continue
+
+    async def wait(self) -> None:
+        loop = asyncio.get_running_loop()
+        waiter = loop.create_future()
+        entry = (loop, waiter)
+        with self._lock:
+            if self._cancelled:
+                return
+            self._waiters.add(entry)
+        try:
+            await waiter
+        finally:
+            with self._lock:
+                self._waiters.discard(entry)
+
+    def raise_if_cancelled(self) -> None:
+        if self.cancelled:
+            raise asyncio.CancelledError