perf(runtime): add compiled workflow plan cache

CHANGELOG.md

@@ -16,6 +16,7 @@ Source: `https://pypi.org/pypi/isage/json` (checked on 2026-02-14, UTC).
 
 ### Changed
 
+- Added a typed, versioned compiled-workflow plan cache with bounded LRU/TTL, failure recovery, single-flight compilation, metrics, and real `FlowDeclaration.compile_reusable()` / `bind_reusable()` integration that keeps request IO outside static plans.
 - Promoted the main repository to the `0.3` product line after the stream/runtime/serving consolidation.
 - Reworked the owned `sage` CLI surface around the in-tree core boundary: `version`, `status`, `doctor`, `verify`, `runtime nodes`, `serve gateway`, `chat`, and `index ingest`.
 - Switched `sage chat` to a `sagellm`-first integration model: direct `sagellm` CLI or external gateway, with no in-repo mock fallback.

docs/compiled-workflow-plans.md

@@ -0,0 +1,57 @@
+# Compiled workflow plans
+
+`sage.runtime.CompiledPlanCache` separates immutable workflow construction from
+per-request binding. Use it for interactive workloads that repeatedly execute
+the same operator DAG under different inputs, identities, deadlines, evidence,
+or cancellation tokens.
+
+```python
+from sage.runtime import CompiledPlanCache, PlanFingerprint, bind_compiled_plan
+
+cache = CompiledPlanCache(max_entries=128, ttl_seconds=900)
+fingerprint = PlanFingerprint.build(
+    operator_dag=[{"id": "load"}, {"id": "transform", "after": ["load"]}],
+    schema={"input": "Record", "output": "Result"},
+    policy_version="2026-08",
+    capabilities={"transform": "v2"},
+    retrieval_contract={"kind": "none"},
+    resource_class="interactive",
+)
+plan = cache.get_or_compile(fingerprint, compile_static_dag)
+run = bind_compiled_plan(plan, request_context, bind_request)
+result = run.execute()
+```
+
+Only structural data belongs in `PlanFingerprint`. Request input, user identity,
+trace ID, deadline, retrieved evidence, and cancellation tokens must be passed
+to `bind_compiled_plan` and must not be captured by `compile_static_dag`.
+
+The cache provides bounded LRU capacity, positive and negative TTLs,
+single-flight compilation, and a `stats()` snapshot. A policy, schema,
+capability, retrieval-contract, resource-class, or compiler-version change
+produces a different digest and therefore invalidates the old plan naturally.
+
+This API is the cache foundation. Runtime-specific compiler integration should
+wrap an immutable compiled artifact and keep execution handles or request state
+outside the cached object.
+
+Flow declarations provide that integration directly:
+
+```python
+bound = declared_flow.bind_reusable(
+    "structural-stage-variant",
+    in_=request_input_topic,
+    out=request_output_topic,
+    schema={"input": "Record", "output": "Result"},
+    policy_version="2026-08",
+    capabilities={"transform": "v2"},
+    retrieval_contract={"kind": "none"},
+    resource_class="interactive",
+)
+```
+
+The structural arguments participate in the fingerprint. The IO bindings do
+not: each returned `BoundFlowDeclaration` points to the same immutable
+`FlowProgram` while retaining its own request topics. Use
+`compiled_plan_cache_stats()` for hit/miss/compile telemetry and
+`clear_compiled_plan_cache()` for explicit operational invalidation.

src/sage/runtime/__init__.py

@@ -14,6 +14,14 @@
 from sage.stream._runtime_kernel_types import Packet, StopSignal
 
 from .backend import get_runtime_backend
+from .compiled_plan_cache import (
+    CompiledPlanCache,
+    CompiledWorkflowPlan,
+    PlanCacheStats,
+    PlanCompileError,
+    PlanFingerprint,
+    bind_compiled_plan,
+)
 from .environments import FlowNetEnvironment, LocalEnvironment
 from .job_manager import JobManager
 from .pipeline_compiler import CompiledActorGraph, PipelineCompiler
@@ -41,6 +49,12 @@
     "get_runtime_backend",
     "PipelineCompiler",
     "CompiledActorGraph",
+    "CompiledPlanCache",
+    "CompiledWorkflowPlan",
+    "PlanCacheStats",
+    "PlanCompileError",
+    "PlanFingerprint",
+    "bind_compiled_plan",
 ]
 
 

src/sage/runtime/compiled_plan_cache.py

@@ -0,0 +1,243 @@
+"""Reusable, request-neutral compiled workflow plans.
+
+The cache is deliberately explicit: callers fingerprint only structural
+workflow contracts, compile an immutable artifact, and bind request state after
+lookup. This prevents identities, inputs, deadlines, evidence, and cancellation
+tokens from leaking into reusable plans.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from collections import OrderedDict
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass
+from threading import Event, RLock
+from time import monotonic
+from typing import Any, Generic, TypeVar
+
+PlanT = TypeVar("PlanT")
+BoundT = TypeVar("BoundT")
+
+
+def _canonical(value: Any) -> Any:
+    if value is None or isinstance(value, (bool, int, float, str)):
+        return value
+    if isinstance(value, Mapping):
+        return {str(key): _canonical(value[key]) for key in sorted(value, key=str)}
+    if isinstance(value, (list, tuple)):
+        return [_canonical(item) for item in value]
+    if isinstance(value, (set, frozenset)):
+        normalized = [_canonical(item) for item in value]
+        return sorted(normalized, key=lambda item: json.dumps(item, sort_keys=True))
+    raise TypeError(
+        f"plan fingerprint values must be JSON-like structural data; got {type(value).__name__}"
+    )
+
+
+@dataclass(frozen=True, slots=True)
+class PlanFingerprint:
+    digest: str
+    contract_json: str
+
+    @classmethod
+    def build(
+        cls,
+        *,
+        operator_dag: Any,
+        schema: Any,
+        policy_version: str,
+        capabilities: Any,
+        retrieval_contract: Any,
+        resource_class: str,
+        compiler_version: str = "1",
+    ) -> PlanFingerprint:
+        contract = _canonical(
+            {
+                "operator_dag": operator_dag,
+                "schema": schema,
+                "policy_version": str(policy_version),
+                "capabilities": capabilities,
+                "retrieval_contract": retrieval_contract,
+                "resource_class": str(resource_class),
+                "compiler_version": str(compiler_version),
+            }
+        )
+        contract_json = json.dumps(
+            contract,
+            ensure_ascii=False,
+            separators=(",", ":"),
+            sort_keys=True,
+        )
+        return cls(
+            digest=hashlib.sha256(contract_json.encode("utf-8")).hexdigest(),
+            contract_json=contract_json,
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class CompiledWorkflowPlan(Generic[PlanT]):
+    fingerprint: PlanFingerprint
+    artifact: PlanT
+    compile_duration_ms: float
+
+
+@dataclass(frozen=True, slots=True)
+class PlanCacheStats:
+    hits: int
+    misses: int
+    waits: int
+    compiles: int
+    compile_failures: int
+    evictions: int
+    entries: int
+    negative_entries: int
+
+
+class PlanCompileError(RuntimeError):
+    """A compile failure, including a bounded negative-cache hit."""
+
+
+@dataclass(slots=True)
+class _Entry(Generic[PlanT]):
+    plan: CompiledWorkflowPlan[PlanT] | None
+    error: str | None
+    expires_at: float
+
+
+class CompiledPlanCache(Generic[PlanT]):
+    def __init__(
+        self,
+        *,
+        max_entries: int = 128,
+        ttl_seconds: float = 900.0,
+        negative_ttl_seconds: float = 2.0,
+        clock: Callable[[], float] = monotonic,
+    ) -> None:
+        if max_entries < 1:
+            raise ValueError("max_entries must be >= 1")
+        if ttl_seconds <= 0 or negative_ttl_seconds <= 0:
+            raise ValueError("cache TTLs must be > 0")
+        self._max_entries = max_entries
+        self._ttl_seconds = ttl_seconds
+        self._negative_ttl_seconds = negative_ttl_seconds
+        self._clock = clock
+        self._entries: OrderedDict[str, _Entry[PlanT]] = OrderedDict()
+        self._inflight: dict[str, Event] = {}
+        self._lock = RLock()
+        self._hits = self._misses = self._waits = 0
+        self._compiles = self._compile_failures = self._evictions = 0
+
+    def get_or_compile(
+        self,
+        fingerprint: PlanFingerprint,
+        compiler: Callable[[], PlanT],
+    ) -> CompiledWorkflowPlan[PlanT]:
+        key = fingerprint.digest
+        while True:
+            owner = False
+            with self._lock:
+                entry = self._entries.get(key)
+                now = self._clock()
+                if entry is not None and entry.expires_at <= now:
+                    del self._entries[key]
+                    entry = None
+                if entry is not None:
+                    self._entries.move_to_end(key)
+                    self._hits += 1
+                    if entry.error is not None:
+                        raise PlanCompileError(entry.error)
+                    assert entry.plan is not None
+                    return entry.plan
+                event = self._inflight.get(key)
+                if event is None:
+                    event = Event()
+                    self._inflight[key] = event
+                    self._misses += 1
+                    owner = True
+                else:
+                    self._waits += 1
+            if owner:
+                break
+            event.wait()
+
+        started_at = self._clock()
+        try:
+            artifact = compiler()
+            plan = CompiledWorkflowPlan(
+                fingerprint=fingerprint,
+                artifact=artifact,
+                compile_duration_ms=max(0.0, (self._clock() - started_at) * 1000.0),
+            )
+        except Exception as exc:
+            with self._lock:
+                self._compile_failures += 1
+                self._insert(
+                    key,
+                    _Entry(
+                        plan=None,
+                        error=f"compiled plan failed for {key[:12]}: {exc}",
+                        expires_at=self._clock() + self._negative_ttl_seconds,
+                    ),
+                )
+                self._inflight.pop(key).set()
+            raise
+
+        with self._lock:
+            self._compiles += 1
+            self._insert(
+                key,
+                _Entry(
+                    plan=plan,
+                    error=None,
+                    expires_at=self._clock() + self._ttl_seconds,
+                ),
+            )
+            self._inflight.pop(key).set()
+        return plan
+
+    def _insert(self, key: str, entry: _Entry[PlanT]) -> None:
+        self._entries[key] = entry
+        self._entries.move_to_end(key)
+        while len(self._entries) > self._max_entries:
+            self._entries.popitem(last=False)
+            self._evictions += 1
+
+    def clear(self) -> None:
+        with self._lock:
+            self._entries.clear()
+
+    def stats(self) -> PlanCacheStats:
+        with self._lock:
+            negative_entries = sum(entry.error is not None for entry in self._entries.values())
+            return PlanCacheStats(
+                hits=self._hits,
+                misses=self._misses,
+                waits=self._waits,
+                compiles=self._compiles,
+                compile_failures=self._compile_failures,
+                evictions=self._evictions,
+                entries=len(self._entries),
+                negative_entries=negative_entries,
+            )
+
+
+def bind_compiled_plan(
+    plan: CompiledWorkflowPlan[PlanT],
+    request_context: Any,
+    binder: Callable[[PlanT, Any], BoundT],
+) -> BoundT:
+    """Bind request-scoped state without mutating or caching it on the plan."""
+
+    return binder(plan.artifact, request_context)
+
+
+__all__ = [
+    "CompiledPlanCache",
+    "CompiledWorkflowPlan",
+    "PlanCacheStats",
+    "PlanCompileError",
+    "PlanFingerprint",
+    "bind_compiled_plan",
+]