From 533365ad6ec2367eb2f06230a6cd37f65820fc35 Mon Sep 17 00:00:00 2001 From: "Shuhao Zhang (Tony)" Date: Sat, 15 Aug 2026 07:55:07 +0800 Subject: [PATCH 1/3] perf(runtime): add compiled workflow plan cache --- CHANGELOG.md | 1 + docs/compiled-workflow-plans.md | 36 ++++ src/sage/runtime/__init__.py | 14 ++ src/sage/runtime/compiled_plan_cache.py | 244 ++++++++++++++++++++++++ src/tests/test_compiled_plan_cache.py | 111 +++++++++++ 5 files changed, 406 insertions(+) create mode 100644 docs/compiled-workflow-plans.md create mode 100644 src/sage/runtime/compiled_plan_cache.py create mode 100644 src/tests/test_compiled_plan_cache.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 6318a3d12..58e573ea0 100644 --- a/CHANGELOG.md +++ b/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 fingerprint/cache foundation with bounded LRU/TTL, failure recovery, single-flight compilation, metrics, and request-neutral runtime binding. - 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. diff --git a/docs/compiled-workflow-plans.md b/docs/compiled-workflow-plans.md new file mode 100644 index 000000000..3293bbbc1 --- /dev/null +++ b/docs/compiled-workflow-plans.md @@ -0,0 +1,36 @@ +# 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. diff --git a/src/sage/runtime/__init__.py b/src/sage/runtime/__init__.py index 75c92903b..089ccf0bf 100644 --- a/src/sage/runtime/__init__.py +++ b/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", ] diff --git a/src/sage/runtime/compiled_plan_cache.py b/src/sage/runtime/compiled_plan_cache.py new file mode 100644 index 000000000..7e5a5e44a --- /dev/null +++ b/src/sage/runtime/compiled_plan_cache.py @@ -0,0 +1,244 @@ +"""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( + "plan fingerprint values must be JSON-like structural data; " + f"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", +] diff --git a/src/tests/test_compiled_plan_cache.py b/src/tests/test_compiled_plan_cache.py new file mode 100644 index 000000000..ebe484884 --- /dev/null +++ b/src/tests/test_compiled_plan_cache.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +import gc +import threading +import time +import weakref +from concurrent.futures import ThreadPoolExecutor + +import pytest + +from sage.runtime import CompiledPlanCache, PlanCompileError, PlanFingerprint, bind_compiled_plan + + +def _fingerprint(*, policy_version: str = "v1") -> PlanFingerprint: + return PlanFingerprint.build( + operator_dag=[{"id": "retrieve"}, {"id": "answer", "after": ["retrieve"]}], + schema={"input": "Question", "output": "Answer"}, + policy_version=policy_version, + capabilities={"model": "chat", "retrieval": True}, + retrieval_contract={"kind": "public-knowledge", "version": 2}, + resource_class="interactive-npu", + ) + + +def test_fingerprint_is_canonical_and_versioned() -> None: + first = _fingerprint() + same = _fingerprint() + changed = _fingerprint(policy_version="v2") + assert first == same + assert first.digest != changed.digest + + +def test_warm_lookup_compiles_once() -> None: + cache: CompiledPlanCache[tuple[str, ...]] = CompiledPlanCache() + calls = 0 + + def compile_plan() -> tuple[str, ...]: + nonlocal calls + calls += 1 + return ("retrieve", "answer") + + first = cache.get_or_compile(_fingerprint(), compile_plan) + second = cache.get_or_compile(_fingerprint(), compile_plan) + assert first is second + assert calls == 1 + assert cache.stats().hits == 1 + + +def test_twenty_concurrent_misses_are_single_flight() -> None: + cache: CompiledPlanCache[object] = CompiledPlanCache() + calls = 0 + lock = threading.Lock() + release = threading.Event() + + def compile_plan() -> object: + nonlocal calls + with lock: + calls += 1 + release.wait(timeout=2) + return object() + + with ThreadPoolExecutor(max_workers=20) as pool: + futures = [pool.submit(cache.get_or_compile, _fingerprint(), compile_plan) for _ in range(20)] + wait_deadline = time.monotonic() + 2.0 + while cache.stats().waits < 19 and time.monotonic() < wait_deadline: + time.sleep(0.001) + release.set() + plans = [future.result(timeout=2) for future in futures] + + assert calls == 1 + assert len({id(plan) for plan in plans}) == 1 + assert cache.stats().waits == 19 + + +def test_negative_cache_expires_and_recovers() -> None: + now = [10.0] + cache: CompiledPlanCache[str] = CompiledPlanCache( + negative_ttl_seconds=2.0, + clock=lambda: now[0], + ) + calls = 0 + + def compile_plan() -> str: + nonlocal calls + calls += 1 + if calls == 1: + raise ValueError("invalid graph") + return "compiled" + + with pytest.raises(ValueError, match="invalid graph"): + cache.get_or_compile(_fingerprint(), compile_plan) + with pytest.raises(PlanCompileError, match="invalid graph"): + cache.get_or_compile(_fingerprint(), compile_plan) + assert calls == 1 + now[0] += 2.1 + assert cache.get_or_compile(_fingerprint(), compile_plan).artifact == "compiled" + assert calls == 2 + + +def test_request_context_is_not_retained_after_bind() -> None: + class RequestContext: + pass + + cache = CompiledPlanCache[str]() + plan = cache.get_or_compile(_fingerprint(), lambda: "static-plan") + context = RequestContext() + context_ref = weakref.ref(context) + assert bind_compiled_plan(plan, context, lambda artifact, _: artifact) == "static-plan" + del context + gc.collect() + assert context_ref() is None From 2bd66bd55b3ffa2ffd188e4576e8df6abad2fcb6 Mon Sep 17 00:00:00 2001 From: "Shuhao Zhang (Tony)" Date: Sat, 15 Aug 2026 07:58:34 +0800 Subject: [PATCH 2/3] style(runtime): format compiled plan cache --- src/sage/runtime/compiled_plan_cache.py | 3 +-- src/tests/test_compiled_plan_cache.py | 4 +++- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/src/sage/runtime/compiled_plan_cache.py b/src/sage/runtime/compiled_plan_cache.py index 7e5a5e44a..c61c3cf44 100644 --- a/src/sage/runtime/compiled_plan_cache.py +++ b/src/sage/runtime/compiled_plan_cache.py @@ -32,8 +32,7 @@ def _canonical(value: Any) -> Any: normalized = [_canonical(item) for item in value] return sorted(normalized, key=lambda item: json.dumps(item, sort_keys=True)) raise TypeError( - "plan fingerprint values must be JSON-like structural data; " - f"got {type(value).__name__}" + f"plan fingerprint values must be JSON-like structural data; got {type(value).__name__}" ) diff --git a/src/tests/test_compiled_plan_cache.py b/src/tests/test_compiled_plan_cache.py index ebe484884..3fc98087b 100644 --- a/src/tests/test_compiled_plan_cache.py +++ b/src/tests/test_compiled_plan_cache.py @@ -60,7 +60,9 @@ def compile_plan() -> object: return object() with ThreadPoolExecutor(max_workers=20) as pool: - futures = [pool.submit(cache.get_or_compile, _fingerprint(), compile_plan) for _ in range(20)] + futures = [ + pool.submit(cache.get_or_compile, _fingerprint(), compile_plan) for _ in range(20) + ] wait_deadline = time.monotonic() + 2.0 while cache.stats().waits < 19 and time.monotonic() < wait_deadline: time.sleep(0.001) From 3bca67addf16d0cdab0af1f93f646d4e70820111 Mon Sep 17 00:00:00 2001 From: "Shuhao Zhang (Tony)" Date: Sat, 15 Aug 2026 08:27:57 +0800 Subject: [PATCH 3/3] perf(flownet): reuse compiled flow declarations --- CHANGELOG.md | 2 +- docs/compiled-workflow-plans.md | 21 ++++ src/sage/runtime/flownet/api/declarations.py | 104 +++++++++++++++++- .../test_flownet_compiled_plan_integration.py | 81 ++++++++++++++ 4 files changed, 206 insertions(+), 2 deletions(-) create mode 100644 src/tests/test_flownet_compiled_plan_integration.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 58e573ea0..7a25e2d43 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -16,7 +16,7 @@ Source: `https://pypi.org/pypi/isage/json` (checked on 2026-02-14, UTC). ### Changed -- Added a typed, versioned compiled-workflow plan fingerprint/cache foundation with bounded LRU/TTL, failure recovery, single-flight compilation, metrics, and request-neutral runtime binding. +- 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. diff --git a/docs/compiled-workflow-plans.md b/docs/compiled-workflow-plans.md index 3293bbbc1..05168c83b 100644 --- a/docs/compiled-workflow-plans.md +++ b/docs/compiled-workflow-plans.md @@ -34,3 +34,24 @@ 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. diff --git a/src/sage/runtime/flownet/api/declarations.py b/src/sage/runtime/flownet/api/declarations.py index f2240bf4b..8a5d89152 100644 --- a/src/sage/runtime/flownet/api/declarations.py +++ b/src/sage/runtime/flownet/api/declarations.py @@ -609,6 +609,7 @@ def __init__( policies=self.policies, ) self._default_program = None + self._compiled_plan_cache = None def __get__(self, instance, owner): if instance is not None: @@ -680,6 +681,102 @@ def _bound_dsl(init_stream): self._default_program = program return program + def compile_reusable( + self, + *, + structural_args: tuple[Any, ...] = (), + structural_kwargs: Mapping[str, Any] | None = None, + schema: Any, + policy_version: str, + capabilities: Any, + retrieval_contract: Any, + resource_class: str, + compiler_version: str = "1", + max_entries: int = 128, + ttl_seconds: float = 900.0, + negative_ttl_seconds: float = 2.0, + ): + """Compile or reuse one request-neutral structural flow plan. + + ``structural_args`` and ``structural_kwargs`` become part of the + fingerprint. Runtime input, identity, evidence, deadline, trace, and + cancellation state belong in :meth:`bind_reusable` instead. + """ + + from sage.runtime.compiled_plan_cache import CompiledPlanCache, PlanFingerprint + + resolved_args = tuple(structural_args) + resolved_kwargs = dict(structural_kwargs or {}) + fingerprint = PlanFingerprint.build( + operator_dag={ + "declaration_id": self.declaration_id, + "definition_hash": self.definition_hash, + "structural_args": resolved_args, + "structural_kwargs": resolved_kwargs, + }, + schema=schema, + policy_version=policy_version, + capabilities=capabilities, + retrieval_contract=retrieval_contract, + resource_class=resource_class, + compiler_version=compiler_version, + ) + if self._compiled_plan_cache is None: + self._compiled_plan_cache = CompiledPlanCache( + max_entries=max_entries, + ttl_seconds=ttl_seconds, + negative_ttl_seconds=negative_ttl_seconds, + ) + return self._compiled_plan_cache.get_or_compile( + fingerprint, + lambda: self.compile(*resolved_args, **resolved_kwargs), + ) + + def bind_reusable( + self, + *structural_args: Any, + in_: Any | None = None, + out: Any | None = None, + schema: Any, + policy_version: str, + capabilities: Any, + retrieval_contract: Any, + resource_class: str, + compiler_version: str = "1", + **structural_kwargs: Any, + ) -> BoundFlowDeclaration: + """Reuse static compilation, then attach per-request IO bindings.""" + + plan = self.compile_reusable( + structural_args=tuple(structural_args), + structural_kwargs=structural_kwargs, + schema=schema, + policy_version=policy_version, + capabilities=capabilities, + retrieval_contract=retrieval_contract, + resource_class=resource_class, + compiler_version=compiler_version, + ) + return BoundFlowDeclaration( + declaration=self, + flow_args=tuple(structural_args), + flow_kwargs=dict(structural_kwargs), + in_binding=in_, + out_binding=out, + _precompiled_flow_program=plan.artifact, + compiled_plan_fingerprint=plan.fingerprint.digest, + compiled_plan_duration_ms=plan.compile_duration_ms, + ) + + def compiled_plan_cache_stats(self): + if self._compiled_plan_cache is None: + return None + return self._compiled_plan_cache.stats() + + def clear_compiled_plan_cache(self) -> None: + if self._compiled_plan_cache is not None: + self._compiled_plan_cache.clear() + @property def pipeline(self) -> list[Any]: return list(self.compile().pipeline) @@ -905,6 +1002,9 @@ class BoundFlowDeclaration: flow_kwargs: dict[str, Any] = field(default_factory=dict) in_binding: Any = None out_binding: Any = None + _precompiled_flow_program: Any = field(default=None, repr=False, compare=False) + compiled_plan_fingerprint: str | None = None + compiled_plan_duration_ms: float | None = None _flow_program: Any = field(init=False, repr=False, compare=False) def __post_init__(self) -> None: @@ -913,7 +1013,9 @@ def __post_init__(self) -> None: object.__setattr__( self, "_flow_program", - self.declaration.compile(*self.flow_args, **self.flow_kwargs), + self._precompiled_flow_program + if self._precompiled_flow_program is not None + else self.declaration.compile(*self.flow_args, **self.flow_kwargs), ) @property diff --git a/src/tests/test_flownet_compiled_plan_integration.py b/src/tests/test_flownet_compiled_plan_integration.py new file mode 100644 index 000000000..4dfd3e9a1 --- /dev/null +++ b/src/tests/test_flownet_compiled_plan_integration.py @@ -0,0 +1,81 @@ +from __future__ import annotations + +from concurrent.futures import ThreadPoolExecutor + +from sage.runtime.flownet.api.declarations import FlowDeclaration + + +def _declaration() -> FlowDeclaration: + def demo_flow(stream, stage_name: str): + assert stage_name == "answer" + return stream + + return FlowDeclaration( + target=demo_flow, + uri="sage://tests/reusable-flow", + scheduler={}, + resources={}, + policies={}, + metadata={}, + dsl_name="demo_flow", + ) + + +def _compile(declaration: FlowDeclaration): + return declaration.compile_reusable( + structural_args=("answer",), + schema={"input": "Question", "output": "Answer"}, + policy_version="v1", + capabilities={"model": "chat"}, + retrieval_contract={"kind": "public"}, + resource_class="interactive", + ) + + +def test_real_flow_program_compile_is_reused() -> None: + declaration = _declaration() + first = _compile(declaration) + second = _compile(declaration) + + assert first is second + assert first.artifact is second.artifact + assert declaration.compiled_plan_cache_stats().compiles == 1 + assert declaration.compiled_plan_cache_stats().hits == 1 + + +def test_concurrent_real_flow_compile_is_single_flight() -> None: + declaration = _declaration() + with ThreadPoolExecutor(max_workers=20) as pool: + plans = list(pool.map(lambda _: _compile(declaration), range(20))) + + assert len({id(plan.artifact) for plan in plans}) == 1 + assert declaration.compiled_plan_cache_stats().compiles == 1 + + +def test_runtime_bindings_do_not_change_or_mutate_static_plan() -> None: + declaration = _declaration() + first = declaration.bind_reusable( + "answer", + in_="request-input-1", + out="request-output-1", + schema={"input": "Question", "output": "Answer"}, + policy_version="v1", + capabilities={"model": "chat"}, + retrieval_contract={"kind": "public"}, + resource_class="interactive", + ) + second = declaration.bind_reusable( + "answer", + in_="request-input-2", + out="request-output-2", + schema={"input": "Question", "output": "Answer"}, + policy_version="v1", + capabilities={"model": "chat"}, + retrieval_contract={"kind": "public"}, + resource_class="interactive", + ) + + assert first.flow_program is second.flow_program + assert first.compiled_plan_fingerprint == second.compiled_plan_fingerprint + assert first._resolve_io_topics() == ("request-input-1", "request-output-1") + assert second._resolve_io_topics() == ("request-input-2", "request-output-2")