"]
+ ]
diff --git a/python/tests/test_local.py b/python/tests/test_local.py
new file mode 100644
index 00000000..be0e137b
--- /dev/null
+++ b/python/tests/test_local.py
@@ -0,0 +1,64 @@
+import pytest
+
+from dialcache.errors import ConfigError
+from dialcache.local import LocalCache
+
+
+class Clock:
+ now = 0.0
+
+ def monotonic_ms(self):
+ return self.now
+
+
+def test_fractional_insertion_and_read_use_whole_millisecond_grid():
+ clock = Clock()
+ cache = LocalCache(clock=clock)
+ clock.now = 0.7
+ cache.put("key", None, 1)
+ clock.now = 999.999
+ assert cache.read("key") == (True, None)
+ clock.now = 1000.0
+ assert cache.read("key") == (False, None)
+
+
+def test_hit_promotes_lru_without_renewing_expiry():
+ clock = Clock()
+ cache = LocalCache(max_size=2, clock=clock)
+ cache.put("a", 1, 1)
+ cache.put("b", 2, 2)
+ clock.now = 900
+ assert cache.read("a") == (True, 1)
+ cache.put("c", 3, 3)
+ assert cache.read("b") == (False, None)
+ clock.now = 1000
+ assert cache.read("a") == (False, None)
+ assert cache.read("c") == (True, 3)
+
+
+def test_new_publication_replaces_value_and_expiration():
+ clock = Clock()
+ cache = LocalCache(clock=clock)
+ cache.put("key", "first", 1)
+ clock.now = 900.8
+ cache.put("key", "second", 2)
+ clock.now = 2899.9
+ assert cache.read("key") == (True, "second")
+ clock.now = 2900
+ assert cache.read("key") == (False, None)
+
+
+def test_zero_capacity_and_large_sparse_capacity():
+ cache = LocalCache(max_size=0)
+ cache.put("key", 1, 1)
+ assert cache.read("key") == (False, None)
+ assert len(cache) == 0
+ large = LocalCache(max_size=9_007_199_254_740_991)
+ large.put("key", 1, 1)
+ assert len(large) == 1
+
+
+@pytest.mark.parametrize("capacity", [True, None, -1, 1.5, 9_007_199_254_740_992])
+def test_invalid_capacity_fails_at_construction(capacity):
+ with pytest.raises(ConfigError):
+ LocalCache(max_size=capacity)
diff --git a/python/tests/test_observer_recovery_boundaries.py b/python/tests/test_observer_recovery_boundaries.py
new file mode 100644
index 00000000..21dccd59
--- /dev/null
+++ b/python/tests/test_observer_recovery_boundaries.py
@@ -0,0 +1,210 @@
+"""Observer privacy and synchronous recovery ownership regressions."""
+
+import asyncio
+import inspect
+import json
+
+import pytest
+from formal.executor import Executor
+
+from dialcache import DialCache, Policy
+from dialcache.protocol import Frame
+
+
+@pytest.fixture
+def executor():
+ instance = Executor()
+ try:
+ yield instance
+ finally:
+ instance.close()
+
+
+@pytest.mark.parametrize("log", [False, True])
+@pytest.mark.parametrize("observer_failure", [False, True])
+def test_mismatch_details_are_logger_only(executor, log, observer_failure):
+ events, warnings = [], []
+
+ def observer(event):
+ events.append(event)
+ if observer_failure:
+ raise RuntimeError("metrics unavailable")
+
+ class Logger:
+ def warning(self, *args):
+ warnings.append(args)
+ if observer_failure:
+ raise RuntimeError("logger unavailable")
+
+ class Redis:
+ def read(self, request, context):
+ return Frame(int(executor.clock.wall_ms()), '"cached-synthetic-credential"')
+
+ cache = DialCache(redis=Redis(), clock=executor.clock, metrics=observer, logger=Logger())
+ policy = Policy(ttl_sec={"remote": 10}, shadow={"ramp": 100, "log_mismatches": log})
+ with cache.enable():
+ result = executor.finish(
+ cache.get_or_load(
+ lambda: "source-synthetic-credential",
+ key="entity-synthetic-id",
+ key_type="entity",
+ use_case="privacy",
+ default_config=policy,
+ )
+ )
+ executor.drain()
+ assert result == "cached-synthetic-credential"
+ assert [e["outcome"] for e in events if e["event"] == "shadowValidation"] == ["mismatch"]
+ assert [e["outcome"] for e in events if e["event"] == "shadowAge"] == ["mismatch"]
+ assert not any(e["event"] == "mismatchWarning" for e in events)
+ serialized = json.dumps(events)
+ for sensitive in [
+ "entity-synthetic-id",
+ "cached-synthetic-credential",
+ "source-synthetic-credential",
+ "cacheKey",
+ "cachedValueJson",
+ "sourceValueJson",
+ ]:
+ assert sensitive not in serialized
+ if log:
+ assert len(warnings) == 1
+ message, details = warnings[0]
+ assert message == "DialCache shadow validation mismatch: %s"
+ assert details["outcome"] == "mismatch"
+ assert "entity-synthetic-id" in details["cacheKey"]
+ assert details["cachedValueJson"] == '"cached-synthetic-credential"'
+ assert details["sourceValueJson"] == '"source-synthetic-credential"'
+ else:
+ assert warnings == []
+
+
+def failing_recovery_call(executor, cache, failure):
+ def source():
+ raise failure
+
+ with cache.enable():
+ with pytest.raises(ValueError) as caught:
+ executor.finish(
+ cache.get_or_load(
+ source,
+ key="id",
+ key_type="entity",
+ use_case="recovery",
+ default_config=Policy(ttl_sec={"remote": 1}, stale_on_error_max_age_sec=10),
+ )
+ )
+ assert caught.value is failure
+ executor.drain()
+
+
+def recovery_cache(executor, predicate):
+ class Redis:
+ def read(self, request, context):
+ return Frame(int(executor.clock.wall_ms()) - 2000, '"stale"')
+
+ return DialCache(redis=Redis(), clock=executor.clock, should_attempt_stale_recovery=predicate)
+
+
+def test_denied_async_recovery_starts_no_work(executor):
+ started, returned = [], []
+ gate = executor.future()
+
+ async def predicate(error):
+ started.append(error)
+ await gate
+ return True
+
+ def classify(error):
+ result = predicate(error)
+ returned.append(result)
+ return result
+
+ cache = recovery_cache(executor, classify)
+ failure = ValueError("source failure")
+ for _ in range(20):
+ failing_recovery_call(executor, cache, failure)
+ assert started == []
+ assert asyncio.all_tasks(executor.loop) == set()
+ assert all(inspect.getcoroutinestate(c) == inspect.CORO_CLOSED for c in returned)
+
+
+def test_denied_custom_recovery_awaitable_is_not_driven(executor):
+ attempts = []
+
+ class Awaitable:
+ def __await__(self):
+ attempts.append("started")
+ yield
+ return True
+
+ cache = recovery_cache(executor, lambda error: Awaitable())
+ failing_recovery_call(executor, cache, ValueError("source failure"))
+ assert attempts == []
+ assert asyncio.all_tasks(executor.loop) == set()
+
+
+def test_borrowed_started_coroutine_stays_application_owned(executor):
+ actions = []
+
+ async def work():
+ actions.append("started")
+ await asyncio.sleep(0)
+ actions.append("finished")
+ return True
+
+ coroutine = work()
+ coroutine.send(None)
+ cache = recovery_cache(executor, lambda error: coroutine)
+ failing_recovery_call(executor, cache, ValueError("source failure"))
+ assert actions == ["started"]
+ assert executor.finish(coroutine) is True
+ assert actions == ["started", "finished"]
+
+
+@pytest.mark.parametrize("kind", ["future", "task"])
+@pytest.mark.parametrize("settlement", ["failure", "cancel"])
+@pytest.mark.parametrize("already_done", [False, True])
+def test_denied_recovery_observes_borrowed_future_without_owning_it(executor, kind, settlement, already_done):
+ observed = []
+
+ class ObserveException:
+ def exception(self):
+ observed.append("observed")
+ return super().exception()
+
+ class Future(ObserveException, asyncio.Future):
+ pass
+
+ class Task(ObserveException, asyncio.Task):
+ pass
+
+ failure = RuntimeError("borrowed task failure")
+ gate = executor.future()
+
+ async def work():
+ await gate
+ raise failure
+
+ borrowed = Future(loop=executor.loop) if kind == "future" else Task(work(), loop=executor.loop)
+ executor.drain()
+
+ def settle():
+ if settlement == "cancel":
+ borrowed.cancel()
+ elif kind == "future":
+ borrowed.set_exception(failure)
+ else:
+ gate.set_result(None)
+ executor.drain()
+
+ if already_done:
+ settle()
+ cache = recovery_cache(executor, lambda error: borrowed)
+ failing_recovery_call(executor, cache, ValueError("source failure"))
+ assert asyncio.all_tasks(executor.loop) == ({borrowed} if kind == "task" and not already_done else set())
+ assert borrowed.done() is already_done
+ if not already_done:
+ settle()
+ assert borrowed.cancelled() is (settlement == "cancel")
+ assert bool(observed) is (settlement == "failure")
diff --git a/python/tests/test_protocol_native.py b/python/tests/test_protocol_native.py
new file mode 100644
index 00000000..66c0e681
--- /dev/null
+++ b/python/tests/test_protocol_native.py
@@ -0,0 +1,298 @@
+from __future__ import annotations
+
+import json
+import math
+import os
+import random
+import shutil
+import struct
+import subprocess
+from pathlib import Path
+
+import pytest
+import zstandard
+
+from dialcache.key import Key, normalize_args, ramp_hash, scalar_string
+from dialcache.protocol import (
+ Frame,
+ Miss,
+ RedisPayloadError,
+ RedisProtocolError,
+ compress_payload,
+ decode_read,
+ decode_tracked_read,
+ decompress_payload,
+ encode_frame,
+ validate_invalidation_reply,
+ validate_set_reply,
+)
+from dialcache.serializer import UNDEFINED, JsonSerializer
+
+ROOT = Path(__file__).resolve().parents[2]
+
+# Import the actual TypeScript module after stripping types with native Node.
+# No copied key/frame algorithm and no package build is used as the oracle.
+NODE_BRIDGE = r"""
+import fs from 'node:fs';
+import path from 'node:path';
+import { stripTypeScriptTypes } from 'node:module';
+const cache = new Map();
+function moduleUrl(file) {
+ if (cache.has(file)) return cache.get(file);
+ let source = stripTypeScriptTypes(fs.readFileSync(file, 'utf8'), {mode:'strip'});
+ source = source.replace(/(from\s+["'])(\.{1,2}\/[^"']+\.js)(["'])/g, (_, before, spec, after) => {
+ const resolved = path.resolve(path.dirname(file), spec.slice(0, -3) + '.ts');
+ return before + moduleUrl(resolved) + after;
+ });
+ const url = 'data:text/javascript;base64,' + Buffer.from(source).toString('base64');
+ cache.set(file, url); return url;
+}
+let text = ''; for await (const chunk of process.stdin) text += chunk;
+const input = JSON.parse(text);
+let output;
+if (input.op === 'numbers') output = input.values.map(String);
+else if (input.op === 'script') output = (await import(moduleUrl(path.resolve('typescript/src/internal/redis-scripts.ts')))).INVALIDATE_CACHE_SCRIPT;
+else {
+ const wire = await import(moduleUrl(path.resolve('typescript/src/internal/redis-payload.ts')));
+ if (input.op === 'encode') output = wire.encodeRedisFrame(input.binary ? Buffer.from(input.payload,'hex') : input.payload,input.at).toString('hex');
+ if (input.op === 'decode') {
+ const frame = Buffer.from(input.hex,'hex');
+ const result = input.tracked ? wire.decodeTrackedRedisReadResult(frame,input.watermark == null ? null : Buffer.from(input.watermark)) : wire.decodeRedisReadResult(frame);
+ output = result.kind === 'miss' ? result : { at:result.createdAtMs, binary:Buffer.isBuffer(result.payload), payload:Buffer.isBuffer(result.payload)?result.payload.toString('hex'):result.payload };
+ }
+}
+process.stdout.write(JSON.stringify(output));
+"""
+
+
+def node_bridge(message):
+ node = os.environ.get("NODE", shutil.which("node"))
+ if node is None:
+ pytest.fail("Node 24+ is required for native cross-language conformance")
+ result = subprocess.run(
+ [node, "--input-type=module", "-e", NODE_BRIDGE],
+ cwd=ROOT,
+ input=json.dumps(message),
+ text=True,
+ capture_output=True,
+ timeout=30,
+ check=False,
+ )
+ assert result.returncode == 0, result.stderr
+ return json.loads(result.stdout)
+
+
+def test_native_ieee754_scalar_identity_matches_javascript():
+ values = [
+ 0.0,
+ -0.0,
+ 1e-7,
+ 1e-6,
+ 1e20,
+ 1e21,
+ 1e23,
+ 1.0000000000000001e18,
+ 9007199254740992.0,
+ 1000000000000000100.0,
+ 5e-324,
+ float.fromhex("0x1.fffffffffffffp+1023"),
+ ]
+ for value in list(values):
+ values.extend(
+ candidate
+ for candidate in [math.nextafter(value, -math.inf), math.nextafter(value, math.inf)]
+ if math.isfinite(candidate)
+ )
+ generator = random.Random(31001)
+ while len(values) < 10000:
+ value = struct.unpack(">d", generator.randbytes(8))[0]
+ if math.isfinite(value):
+ values.append(value)
+ assert [scalar_string(value) for value in values] == node_bridge({"op": "numbers", "values": values})
+
+
+def test_arbitrary_bigints_and_scalar_domains():
+ assert scalar_string(10**5000 + 7) == "1" + "0" * 4999 + "7"
+ assert scalar_string(-(10**5000 + 7)) == "-1" + "0" * 4999 + "7"
+ assert scalar_string(-0.0) == "0"
+ assert scalar_string(math.nan) == "NaN"
+ assert scalar_string(math.inf) == "Infinity"
+ assert scalar_string(-math.inf) == "-Infinity"
+ with pytest.raises(TypeError):
+ scalar_string([])
+
+
+def test_utf16_order_pairs_and_payload_surrogates():
+ assert normalize_args({"\ue000": 1, "\U00010000": 2, "missing": UNDEFINED}) == (
+ ("\U00010000", "2"),
+ ("\ue000", "1"),
+ )
+ pair_key = Key("urn", "id", "\ud83d\ude00", "Get")
+ assert pair_key.logical == Key("urn", "id", "😀", "Get").logical
+ assert encode_frame("\ud83d\ude00", 1) == encode_frame("😀", 1)
+ assert encode_frame("\ud800", 1)[10:] == b"\xef\xbf\xbd"
+ assert ramp_hash("😀", "local") == ramp_hash("\ud83d\ude00", "local")
+
+
+@pytest.mark.parametrize("raw", ["text", bytearray(b"frame"), memoryview(b"frame"), 1, False, [], {}])
+def test_protocol_rejects_nonbulk_runtime_replies(raw):
+ with pytest.raises(RedisPayloadError):
+ decode_read(raw)
+ with pytest.raises(RedisPayloadError):
+ decode_tracked_read(None, raw)
+
+
+def test_fence_grammar_and_classification_precedence():
+ frame = encode_frame("x", 1)
+ assert decode_tracked_read(None, b"bad") == Miss("value_absent")
+ assert decode_tracked_read(frame, b"0" * 10000) == Frame(1, "x")
+ assert decode_tracked_read(frame, b"9" * 10000) == Miss("unclassified")
+ assert decode_tracked_read(frame[:9] + b"\xffx", b"1") == Miss("watermark_fenced", 1)
+ with pytest.raises(ValueError):
+ encode_frame("x", 10**5000)
+
+
+def test_mutation_replies_are_strict():
+ for reply in [None, False, "1", b"1", 1.0, True, 0, 2]:
+ with pytest.raises(RedisProtocolError):
+ validate_invalidation_reply(reply)
+ validate_invalidation_reply(1)
+ for reply in [None, False, "ok", 1, b"no"]:
+ with pytest.raises(RedisProtocolError):
+ validate_set_reply(reply)
+ for reply in ["OK", b"OK", True]:
+ validate_set_reply(reply)
+
+
+def test_compression_resource_and_native_stream_boundaries():
+ for known_size in [True, False]:
+ compressed = zstandard.ZstdCompressor(write_content_size=known_size).compress(b"a" * 10000)
+ raw = b"\x02" + compressed
+ assert decompress_payload(raw, 9999).outcome == "read_over_limit"
+ assert decompress_payload(raw, 10000).payload == b"a" * 10000
+ assert decompress_payload(raw[:-1], 10000).outcome == "fallback_raw"
+ assert decompress_payload(raw[:-1], 2).outcome == "fallback_raw"
+ first = zstandard.ZstdCompressor().compress(b"first")
+ second = zstandard.ZstdCompressor().compress(b"second")
+ # Match Node's native decoder: only the first completed stream is consumed.
+ assert decompress_payload(b"\x01" + first + second).payload == "first"
+ assert decompress_payload(b"\x01" + first + b"trailer").payload == "first"
+
+
+def test_json_roundtrips_and_undefined():
+ serializer = JsonSerializer()
+ values = [None, False, True, 0, -42, 1.5, "雪", {"k": [1, None, "x"]}, UNDEFINED]
+ for value in values:
+ assert serializer.load(serializer.dump(value)) == value
+ assert serializer.dump(UNDEFINED) == "__dialcache_json_undefined_v1__"
+ assert serializer.load('"__dialcache_json_undefined_v1__"') == "__dialcache_json_undefined_v1__"
+ assert serializer.load(b'"\xff"') == "\ufffd"
+ for value in [math.nan, math.inf, -math.inf, object()]:
+ with pytest.raises((TypeError, ValueError)):
+ serializer.dump(value)
+
+
+@pytest.mark.parametrize("value", ["\ud800", "\udc00", "A\ud800B", {"\ud800": ["\udc00", "雪"]}])
+def test_json_strings_survive_the_frame_utf8_boundary(value):
+ serializer = JsonSerializer()
+ decoded = decode_read(encode_frame(serializer.dump(value), 1))
+ assert isinstance(decoded, Frame)
+ assert serializer.load(decoded.payload) == value
+
+
+@pytest.mark.parametrize("size", [0, 1, 65535, 65536, 65537])
+@pytest.mark.parametrize("trailer", [b"garbage", zstandard.ZstdCompressor().compress(b"second" * 1000)])
+def test_unknown_size_decoder_stops_at_first_frame(size, trailer):
+ raw = b"a" * size
+ first = zstandard.ZstdCompressor(write_content_size=False).compress(raw)
+ result = decompress_payload(b"\x02" + first + trailer, maximum=size)
+ assert result.outcome == "decompressed"
+ assert result.payload == raw
+ assert decompress_payload(b"\x02" + first[:-1], maximum=size).outcome == "fallback_raw"
+
+
+@pytest.mark.parametrize("known_size", [False, True])
+def test_bad_checksum_preserves_output_limit_precedence(known_size):
+ encoded = zstandard.ZstdCompressor(write_content_size=known_size, write_checksum=True).compress(
+ b"a" * 100000
+ )
+ bad = b"\x02" + encoded[:-1] + bytes([encoded[-1] ^ 1])
+ assert decompress_payload(bad, maximum=65536).outcome == "read_over_limit"
+ assert decompress_payload(bad).outcome == "fallback_raw"
+
+
+@pytest.mark.parametrize("corrupt", [False, True])
+def test_unknown_size_decode_allocates_for_output_instead_of_ceiling(corrupt):
+ import tracemalloc
+
+ raw = b"a" * 10000
+ encoded = zstandard.ZstdCompressor(write_content_size=False, write_checksum=True).compress(raw)
+ if corrupt:
+ encoded = encoded[:-1] + bytes([encoded[-1] ^ 1])
+ payload = b"\x02" + encoded
+ tracemalloc.start()
+ try:
+ result = decompress_payload(payload)
+ _, peak = tracemalloc.get_traced_memory()
+ finally:
+ tracemalloc.stop()
+ assert result.outcome == ("fallback_raw" if corrupt else "decompressed")
+ assert result.payload == (payload if corrupt else raw)
+ # This small output used to allocate the 512 MiB decompression ceiling.
+ # Keep generous headroom for interpreter/dependency allocation differences.
+ assert peak < 8 * 1024 * 1024
+
+
+def test_known_oversized_decode_does_not_retain_unusable_output():
+ import tracemalloc
+
+ maximum = 8 * 1024 * 1024
+ payload = b"\x02" + zstandard.ZstdCompressor().compress(b"a" * (maximum + 1))
+ tracemalloc.start()
+ try:
+ result = decompress_payload(payload, maximum)
+ _, peak = tracemalloc.get_traced_memory()
+ finally:
+ tracemalloc.stop()
+ assert result.outcome == "read_over_limit"
+ assert result.payload is payload
+ # The frame header already rules out returning decoded bytes. Classification
+ # must use bounded chunks rather than retaining approximately the full cap.
+ assert peak < 2 * 1024 * 1024
+
+
+@pytest.mark.parametrize("text,normalized", [("雪", "雪"), ("\ud83d\ude00", "😀"), ("\ud800", "\ufffd")])
+def test_text_compression_uses_normalized_byte_sizes_without_changing_raw_text(text, normalized):
+ payload, decoded = text * 4096, normalized * 4096
+ size = len(decoded.encode("utf-8"))
+ result = compress_payload(payload, threshold_bytes=1, maximum=size)
+ assert result.outcome == "compressed"
+ assert result.original_bytes == size
+ assert result.stored_bytes == len(result.payload)
+ assert decompress_payload(result.payload).payload == decoded
+ for threshold, maximum, outcome in [
+ (size + 1, size, "below_threshold"),
+ (1, size - 1, "write_over_limit"),
+ ]:
+ result = compress_payload(payload, threshold_bytes=threshold, maximum=maximum)
+ assert result.outcome == outcome
+ assert result.payload is payload
+ assert result.original_bytes == result.stored_bytes == size
+
+
+def test_text_compression_does_not_allocate_redundant_full_payload_buffers():
+ import tracemalloc
+
+ payload = "abcd" * (1024 * 1024)
+ tracemalloc.start()
+ try:
+ result = compress_payload(payload)
+ _, peak = tracemalloc.get_traced_memory()
+ finally:
+ tracemalloc.stop()
+ assert result.outcome == "compressed"
+ assert decompress_payload(result.payload).payload == payload
+ # Normalizing text needs temporary UTF-16/UTF-8 buffers. Re-encoding while
+ # retaining a previous full buffer used over 4x the input; allow headroom
+ # above one normalization's 3x peak without allowing that extra copy.
+ assert peak < len(payload) * 7 // 2
diff --git a/python/tests/test_protocol_vectors.py b/python/tests/test_protocol_vectors.py
new file mode 100644
index 00000000..50336180
--- /dev/null
+++ b/python/tests/test_protocol_vectors.py
@@ -0,0 +1,226 @@
+"""Replay every fixed and generated portable wire vector against the real API."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+
+import pytest
+import zstandard
+
+from dialcache.key import Key, normalize_args, ramp_hash, ramp_sample
+from dialcache.protocol import (
+ Frame,
+ Miss,
+ RedisPayloadEncodingError,
+ ceil_supported_cache_ttl_ms,
+ compress_payload,
+ decode_read,
+ decode_tracked_read,
+ decompress_payload,
+ encode_frame,
+ escape_raw_payload,
+ utf8_bytes,
+)
+from dialcache.serializer import UNDEFINED
+
+ROOT = Path(__file__).resolve().parents[2]
+CORPORA = [
+ json.loads((ROOT / "formal" / name).read_text())
+ for name in (
+ "protocol-vectors.json",
+ "quint-key-vectors.json",
+ "quint-frame-vectors.json",
+ "quint-envelope-vectors.json",
+ )
+]
+
+
+def vectors(group):
+ return [pytest.param(vector, id=vector["name"]) for corpus in CORPORA for vector in corpus.get(group, [])]
+
+
+def payload(vector):
+ return bytes.fromhex(vector["payloadHex"]) if vector["payloadType"] == "binary" else vector["payloadUtf8"]
+
+
+def key_from(vector):
+ value = vector["input"]
+ return Key(
+ namespace=value["namespace"],
+ key_type=value["keyType"],
+ id=value["id"],
+ use_case=value["useCase"],
+ args=value["args"],
+ tracked=value["trackForInvalidation"],
+ )
+
+
+def test_schemas_provenance_and_inventory():
+ assert all(corpus["schemaVersion"] == 3 for corpus in CORPORA)
+ expected_counts = [134, 457, 589, 297]
+ for corpus, expected in zip(CORPORA, expected_counts, strict=True):
+ rows = [row for value in corpus.values() if isinstance(value, list) for row in value]
+ assert len(rows) == expected
+ for path, expected_hash in corpus.get("provenance", {}).get("sourceSha256", {}).items():
+ assert hashlib.sha256((ROOT / path).read_bytes()).hexdigest() == expected_hash, path
+
+
+@pytest.mark.parametrize("vector", vectors("keyVectors"))
+def test_keys(vector):
+ key = key_from(vector)
+ assert key.logical == vector["logicalKey"]
+ assert key.value_key == vector["valueKey"]
+ assert key.watermark_key == vector["watermarkKey"]
+
+
+@pytest.mark.parametrize("vector", vectors("invalidKeyVectors"))
+def test_invalid_keys(vector):
+ with pytest.raises((ValueError, TypeError)):
+ key_from(vector)
+
+
+@pytest.mark.parametrize("vector", vectors("normalizeArgsVectors"))
+def test_normalize_args(vector):
+ values = {
+ name: UNDEFINED if "undefinedSentinel" in vector and value == vector["undefinedSentinel"] else value
+ for name, value in vector["input"].items()
+ }
+ values.update({name: int(value) for name, value in vector.get("bigintArgs", {}).items()})
+ values.update({name: float(value) for name, value in vector.get("specialArgs", {}).items()})
+ assert normalize_args(values) == tuple(tuple(pair) for pair in vector["expected"])
+
+
+@pytest.mark.parametrize("vector", vectors("rampVectors"))
+def test_rollout(vector):
+ key = key_from(vector)
+ assert ramp_sample(key, vector["layer"]) == vector["sample"]
+ if "hashNumerator" in vector:
+ assert ramp_hash(key, vector["layer"]) == vector["hashNumerator"]
+
+
+@pytest.mark.parametrize("vector", vectors("frameVectors"))
+def test_encode(vector):
+ assert encode_frame(payload(vector), vector["createdAtMs"]).hex() == vector["frameHex"]
+
+
+@pytest.mark.parametrize("vector", vectors("invalidTimestampVectors"))
+def test_invalid_timestamp(vector):
+ value = float(vector["specialInput"]) if "specialInput" in vector else vector["input"]
+ with pytest.raises(ValueError):
+ encode_frame("value", value)
+
+
+@pytest.mark.parametrize("vector", vectors("durationVectors"))
+def test_duration(vector):
+ value = float(vector["specialInput"]) if "specialInput" in vector else vector["input"]
+ if vector["expected"] is None:
+ with pytest.raises(ValueError):
+ ceil_supported_cache_ttl_ms(value)
+ else:
+ assert ceil_supported_cache_ttl_ms(value) == vector["expected"]
+
+
+def assert_decode(vector, *, tracked):
+ raw = None if vector["frameHex"] is None else bytes.fromhex(vector["frameHex"])
+ watermark = vector.get("watermarkUtf8")
+
+ def decode():
+ return (
+ decode_tracked_read(raw, None if watermark is None else watermark.encode())
+ if tracked
+ else decode_read(raw)
+ )
+
+ expected = vector["expected"]
+ if expected["kind"] == "payload_encoding_error":
+ with pytest.raises(RedisPayloadEncodingError):
+ decode()
+ elif expected["kind"] == "miss":
+ assert decode() == Miss(expected["reason"], expected.get("observedWatermarkMs"))
+ else:
+ assert decode() == Frame(expected["createdAtMs"], payload(expected))
+
+
+@pytest.mark.parametrize("vector", vectors("trackedDecodeVectors"))
+def test_tracked_decode(vector):
+ assert_decode(vector, tracked=True)
+
+
+@pytest.mark.parametrize("vector", vectors("untrackedDecodeVectors"))
+def test_untracked_decode(vector):
+ assert_decode(vector, tracked=False)
+
+
+@pytest.mark.parametrize("vector", vectors("envelopeVectors"))
+def test_envelopes(vector):
+ raw = bytes.fromhex(vector["inputHex"])
+ escaped = bytes.fromhex(vector["escapedHex"])
+ assert escape_raw_payload(raw) == escaped
+ result = decompress_payload(raw)
+ assert result.payload == bytes.fromhex(vector["decodedHex"])
+ assert result.outcome == vector["outcome"]
+ assert decompress_payload(escaped).payload == raw
+
+
+@pytest.mark.parametrize("vector", vectors("compressedDecodeVectors"))
+def test_compressed_decode(vector):
+ raw = bytes.fromhex(vector["inputHex"])
+ if "codecFixture" in vector:
+ fixture = vector["codecFixture"]
+ if fixture["succeeds"]:
+ assert zstandard.ZstdDecompressor().decompress(raw[1:]).hex() == fixture["decodedHex"]
+ else:
+ with pytest.raises(zstandard.ZstdError):
+ zstandard.ZstdDecompressor().decompress(raw[1:])
+ result = decompress_payload(raw, vector.get("maxDecompressedBytes", 536870912))
+ assert result.payload == payload(vector)
+ assert result.outcome == vector.get("outcome", "decompressed")
+
+
+@pytest.mark.parametrize("vector", vectors("compressionWriteVectors"))
+def test_compression_selection(vector):
+ raw = payload(vector)
+ data = raw if isinstance(raw, bytes) else utf8_bytes(raw)
+ if "codecBytes" in vector:
+ # zstandard's level-3 binding matches Node's codec environment for this
+ # complete generated domain; establish that before comparing its model.
+ native = zstandard.ZstdCompressor(level=3).compress(data)
+ assert len(native) == vector["codecBytes"]["typescript"]
+ result = compress_payload(
+ raw, threshold_bytes=vector["thresholdBytes"], maximum=vector.get("maxDecompressedBytes", 536870912)
+ )
+ expected = vector.get("expectedByBinding", {}).get("typescript")
+ assert result.outcome == (expected["outcome"] if expected else vector["outcome"])
+ if expected:
+ assert result.stored_bytes == expected["storedBytes"]
+ assert result.original_bytes == vector["originalBytes"]
+ escaped = escape_raw_payload(raw)
+ assert (escaped if isinstance(escaped, bytes) else utf8_bytes(escaped)).hex() == vector["escapedHex"]
+ if result.outcome == "compressed":
+ assert result.payload[0] == expected["marker"]
+ assert decompress_payload(result.payload).payload == raw
+
+
+def assert_wire_vector(group, vector):
+ """Completion reporters call these same real assertions before granting credit."""
+ assertions = {
+ "keyVectors": test_keys,
+ "invalidKeyVectors": test_invalid_keys,
+ "normalizeArgsVectors": test_normalize_args,
+ "rampVectors": test_rollout,
+ "frameVectors": test_encode,
+ "invalidTimestampVectors": test_invalid_timestamp,
+ "durationVectors": test_duration,
+ "trackedDecodeVectors": test_tracked_decode,
+ "untrackedDecodeVectors": test_untracked_decode,
+ "envelopeVectors": test_envelopes,
+ "compressedDecodeVectors": test_compressed_decode,
+ "compressionWriteVectors": test_compression_selection,
+ }
+ try:
+ assertion = assertions[group]
+ except KeyError:
+ raise ValueError(f"Unsupported wire vector group: {group}") from None
+ assertion(vector)
diff --git a/python/tests/test_redis_adapter.py b/python/tests/test_redis_adapter.py
new file mode 100644
index 00000000..9cad057e
--- /dev/null
+++ b/python/tests/test_redis_adapter.py
@@ -0,0 +1,123 @@
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+from test_protocol_native import node_bridge
+
+from dialcache.protocol import Frame, Miss, RedisProtocolError, encode_frame
+from dialcache.redis import (
+ INVALIDATE_CACHE_SCRIPT,
+ INVALIDATE_CACHE_SCRIPT_SHA1,
+ InvalidationRequest,
+ ReadContext,
+ ReadRequest,
+ RedisAdapter,
+ WriteRequest,
+)
+
+
+class Client:
+ def __init__(self, replies):
+ self.replies = iter(replies)
+ self.calls = []
+
+ async def execute_command(self, *args, **kwargs):
+ self.calls.append((args, kwargs))
+ value = next(self.replies)
+ if isinstance(value, BaseException):
+ raise value
+ return value
+
+
+async def test_semantic_reads_and_single_complete_frame_set():
+ client = Client([None, [encode_frame("old", 1000), b"1000"], [encode_frame(b"new", 1001), b"1000"], True])
+ adapter = RedisAdapter(client)
+ assert await adapter.read(ReadRequest("value")) == Miss("value_absent")
+ assert await adapter.read(ReadRequest("value", "watermark")) == Miss("watermark_fenced", 1000)
+ assert await adapter.read(ReadRequest("value", "watermark")) == Frame(1001, b"new")
+ await adapter.write(WriteRequest("value", 1.25, b"data", 17))
+ assert client.calls == [
+ (("GET", "value"), {}),
+ (("MGET", "value", "watermark"), {}),
+ (("MGET", "value", "watermark"), {}),
+ (("SET", "value", encode_frame(b"data", 17), "PX", "2"), {}),
+ ]
+
+
+async def test_invalidation_retry_preserves_exact_arguments_and_error():
+ client = Client([ConnectionError("ambiguous"), 1])
+ await RedisAdapter(client).invalidate(InvalidationRequest("watermark", 200, 1000))
+ assert client.calls == [
+ (("EVALSHA", INVALIDATE_CACHE_SCRIPT_SHA1, "1", "watermark", "200", "1000"), {}),
+ (("EVAL", INVALIDATE_CACHE_SCRIPT, "1", "watermark", "200", "1000"), {}),
+ ]
+ error = RuntimeError("second rejection")
+ client = Client([ConnectionError(), error])
+ with pytest.raises(RuntimeError) as raised:
+ await RedisAdapter(client).invalidate(InvalidationRequest("watermark", 0, 1000))
+ assert raised.value is error
+
+
+async def test_accepted_bad_mutation_replies_are_never_retried():
+ for reply in [None, "1", True, 1.0]:
+ client = Client([reply])
+ with pytest.raises(RedisProtocolError):
+ await RedisAdapter(client).invalidate(InvalidationRequest("watermark", 0, 1000))
+ assert len(client.calls) == 1
+
+
+async def test_mutation_input_validation_precedes_dispatch():
+ client = Client([])
+ adapter = RedisAdapter(client)
+ for timestamp in [-1, True, 1.5, float("inf"), 9007199254740992]:
+ with pytest.raises(ValueError):
+ await adapter.write(WriteRequest("value", 10, "x", timestamp))
+ with pytest.raises(ValueError):
+ await adapter.invalidate(InvalidationRequest("watermark", 0, timestamp))
+ for ttl in [0, -1, True, float("inf"), 31536000001]:
+ with pytest.raises(ValueError):
+ await adapter.write(WriteRequest("value", ttl, "x", 1))
+ assert client.calls == []
+
+
+async def test_cluster_primary_routing_with_primary_only_connections():
+ class Cluster(Client):
+ read_from_replicas = False
+
+ def get_connection_kwargs(self):
+ return {}
+
+ async def initialize(self):
+ self.initialized = True
+
+ def get_node_from_key(self, key, replica=False):
+ assert self.initialized
+ assert replica is False
+ return ("primary", key)
+
+ client = Cluster([[encode_frame("value", 2), b"1"]])
+ assert await RedisAdapter(client).read(ReadRequest("{entity}#value", "{entity}#watermark")) == Frame(
+ 2, "value"
+ )
+ assert client.calls[0][1] == {"target_nodes": ("primary", "{entity}#value")}
+
+
+async def test_preaborted_read_does_not_dispatch():
+ class Signal:
+ aborted = True
+
+ client = Client([])
+ with pytest.raises(asyncio.CancelledError):
+ await RedisAdapter(client).read(ReadRequest("key"), ReadContext(10, Signal()))
+ assert client.calls == []
+
+
+@pytest.mark.parametrize("reply", [None, [], [b"a"], [b"a", b"b", b"c"], "ab"])
+async def test_invalid_atomic_snapshot_shape(reply):
+ with pytest.raises(RedisProtocolError):
+ await RedisAdapter(Client([reply])).read(ReadRequest("value", "watermark"))
+
+
+def test_lua_source_is_identical_to_current_typescript():
+ assert INVALIDATE_CACHE_SCRIPT == node_bridge({"op": "script"})
diff --git a/python/tests/test_redis_integration.py b/python/tests/test_redis_integration.py
new file mode 100644
index 00000000..e7964c66
--- /dev/null
+++ b/python/tests/test_redis_integration.py
@@ -0,0 +1,202 @@
+"""Real Redis/Valkey transition and Python/TypeScript interoperability evidence."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+import redis.asyncio as redis
+from redis.exceptions import ResponseError
+from test_protocol_native import node_bridge
+
+from dialcache.key import Key
+from dialcache.protocol import Frame, Miss, RedisProtocolError, encode_frame
+from dialcache.redis import (
+ INVALIDATE_CACHE_SCRIPT,
+ InvalidationRequest,
+ ReadRequest,
+ RedisAdapter,
+ WriteRequest,
+)
+
+pytestmark = pytest.mark.integration
+ROOT = Path(__file__).resolve().parents[2]
+INVALIDATION_CORPORA = [
+ json.loads((ROOT / "formal" / name).read_text())
+ for name in ["invalidation-vectors.json", "quint-invalidation-vectors.json"]
+]
+INVALIDATIONS = [
+ pytest.param(vector, id=vector["name"]) for corpus in INVALIDATION_CORPORA for vector in corpus["vectors"]
+]
+
+
+@pytest.fixture
+async def server():
+ url = os.environ.get("TEST_REDIS_URL")
+ if not url:
+ pytest.skip("Set TEST_REDIS_URL to run against a real Redis or Valkey server")
+ client = redis.Redis.from_url(url, decode_responses=False, socket_timeout=5, socket_connect_timeout=5)
+ await client.ping()
+ try:
+ yield client
+ finally:
+ await client.aclose()
+
+
+@pytest.fixture
+async def owned_key(server):
+ key = "dialcache-python-test:" + uuid4().hex
+ try:
+ yield key
+ finally:
+ await server.delete(key)
+
+
+def test_invalidation_corpus_inventory_and_provenance():
+ assert [len(corpus["vectors"]) for corpus in INVALIDATION_CORPORA] == [49, 288]
+ for corpus in INVALIDATION_CORPORA:
+ assert corpus["schemaVersion"] == 2
+ assert len({vector["name"] for vector in corpus["vectors"]}) == len(corpus["vectors"])
+ for path, expected in corpus.get("provenance", {}).get("sourceSha256", {}).items():
+ assert hashlib.sha256((ROOT / path).read_bytes()).hexdigest() == expected
+
+
+@pytest.mark.parametrize("vector", INVALIDATIONS)
+async def test_all_invalidation_transitions(server, owned_key, vector):
+ existing, expected = vector["existing"], vector["expected"]["state"]
+ commands = server.pipeline(transaction=True)
+ commands.time()
+ commands.delete(owned_key)
+ if existing["kind"] == "string":
+ commands.set(owned_key, existing["value"])
+ elif existing["kind"] == "list":
+ commands.rpush(owned_key, *existing["values"])
+ if existing["ttlMs"] > 0:
+ commands.pexpire(owned_key, existing["ttlMs"])
+ result_index = len(commands.command_stack)
+ commands.eval(INVALIDATE_CACHE_SCRIPT, 1, owned_key, vector["futureBufferMs"], vector["invalidatedAtMs"])
+ commands.type(owned_key)
+ content_index = len(commands.command_stack)
+ if expected["kind"] == "string":
+ commands.get(owned_key)
+ elif expected["kind"] == "list":
+ commands.lrange(owned_key, 0, -1)
+ commands.pttl(owned_key)
+ commands.time()
+ results = await commands.execute(raise_on_error=False)
+ result = results[result_index]
+ if vector["expected"].get("error"):
+ assert isinstance(result, ResponseError), result
+ else:
+ assert result == 1
+ assert (
+ results[result_index + 1]
+ == {"absent": b"none", "string": b"string", "list": b"list"}[expected["kind"]]
+ )
+ if expected["kind"] == "string":
+ assert results[content_index] == expected["value"].encode()
+ elif expected["kind"] == "list":
+ assert results[content_index] == [value.encode() for value in expected["values"]]
+ actual_ttl = results[-2]
+ expected_ttl = expected["ttlMs"]
+ if expected_ttl < 0:
+ assert actual_ttl == expected_ttl
+ else:
+ # Only the Redis clock measured around the atomic setup/transition/read
+ # widens the bound; no arbitrary network tolerance hides TTL defects.
+ before, after = results[0], results[-1]
+ elapsed_ms = (after[0] * 1_000_000 + after[1]) // 1000 - (before[0] * 1_000_000 + before[1]) // 1000
+ assert expected_ttl - elapsed_ms <= actual_ttl <= expected_ttl
+
+
+async def test_adapter_real_frames_fencing_and_retention(server):
+ key = Key("dialcache-python-test-" + uuid4().hex, "user", "1", "Get", tracked=True)
+ adapter = RedisAdapter(server)
+ try:
+ assert await adapter.read(ReadRequest(key.value_key, key.watermark_key)) == Miss("value_absent")
+ await adapter.write(WriteRequest(key.value_key, 60000, "value", 1000))
+ assert await server.get(key.value_key) == encode_frame("value", 1000)
+ assert await server.get(key.watermark_key) is None
+ before = await server.time()
+ await adapter.invalidate(InvalidationRequest(key.watermark_key, 200, 1000))
+ assert await adapter.read(ReadRequest(key.value_key, key.watermark_key)) == Miss(
+ "watermark_fenced", 1200
+ )
+ retention = await server.pttl(key.watermark_key)
+ after = await server.time()
+ elapsed = (after[0] * 1_000_000 + after[1]) // 1000 - (before[0] * 1_000_000 + before[1]) // 1000
+ assert 7_200_000 - elapsed <= retention <= 7_200_000
+ await adapter.write(WriteRequest(key.value_key, 60000, b"fresh", 1201))
+ assert await adapter.read(ReadRequest(key.value_key, key.watermark_key)) == Frame(1201, b"fresh")
+ assert await server.get(key.watermark_key) == b"1200"
+ await server.delete(key.value_key)
+ assert await adapter.read(ReadRequest(key.value_key, key.watermark_key)) == Miss("value_absent", 1200)
+ finally:
+ await server.delete(key.value_key, key.watermark_key)
+
+
+@pytest.mark.parametrize("value", ["snow: 雪 😀", b"\x00\xff\x01\x02"])
+async def test_bidirectional_typescript_protocol_through_real_redis(server, owned_key, value):
+ adapter = RedisAdapter(server)
+ await adapter.write(WriteRequest(owned_key, 60000, value, 1234))
+ raw = await server.get(owned_key)
+ result = node_bridge({"op": "decode", "hex": raw.hex(), "tracked": True, "watermark": "1233"})
+ assert result == {
+ "at": 1234,
+ "binary": isinstance(value, bytes),
+ "payload": value.hex() if isinstance(value, bytes) else value,
+ }
+ encoded = node_bridge(
+ {
+ "op": "encode",
+ "at": 1235,
+ "binary": isinstance(value, bytes),
+ "payload": value.hex() if isinstance(value, bytes) else value,
+ }
+ )
+ await server.set(owned_key, bytes.fromhex(encoded), px=60000)
+ assert await adapter.read(ReadRequest(owned_key)) == Frame(1235, value)
+
+
+async def test_cluster_atomic_primary_snapshot_and_native_writes():
+ url = os.environ.get("TEST_REDIS_CLUSTER_URL")
+ if not url:
+ pytest.skip("Set TEST_REDIS_CLUSTER_URL to test an actual Redis Cluster")
+ client = redis.RedisCluster.from_url(
+ url, decode_responses=False, socket_timeout=5, socket_connect_timeout=5
+ )
+ key = Key("dialcache-python-cluster-" + uuid4().hex, "entity", "1", "Get", tracked=True)
+ adapter = RedisAdapter(client)
+ try:
+ await client.initialize()
+ await adapter.write(WriteRequest(key.value_key, 60000, "before", 1000))
+ await adapter.invalidate(InvalidationRequest(key.watermark_key, 0, 1000))
+ assert await adapter.read(ReadRequest(key.value_key, key.watermark_key)) == Miss(
+ "watermark_fenced", 1000
+ )
+ await adapter.write(WriteRequest(key.value_key, 60000, "after", 1001))
+ for _ in range(20):
+ assert await adapter.read(ReadRequest(key.value_key, key.watermark_key)) == Frame(1001, "after")
+ primary = client.get_node_from_key(key.value_key, replica=False)
+ assert await client.execute_command("GET", key.value_key, target_nodes=primary) == encode_frame(
+ "after", 1001
+ )
+ assert await client.execute_command("GET", key.watermark_key, target_nodes=primary) == b"1000"
+ replica_client = redis.RedisCluster.from_url(
+ url, read_from_replicas=True, decode_responses=False, socket_timeout=5, socket_connect_timeout=5
+ )
+ try:
+ await replica_client.initialize()
+ with pytest.raises(RedisProtocolError, match="primary-only RedisCluster"):
+ await RedisAdapter(replica_client).read(ReadRequest(key.value_key, key.watermark_key))
+ finally:
+ await replica_client.aclose()
+ finally:
+ try:
+ await client.delete(key.value_key, key.watermark_key)
+ finally:
+ await client.aclose()
diff --git a/python/tests/test_review_regressions.py b/python/tests/test_review_regressions.py
new file mode 100644
index 00000000..6866f886
--- /dev/null
+++ b/python/tests/test_review_regressions.py
@@ -0,0 +1,506 @@
+"""Public boundary regressions from the independent Python PR review."""
+
+import asyncio
+
+import pytest
+from formal.executor import Executor
+
+from dialcache import DialCache, FallbackTimeoutError, Policy
+from dialcache.protocol import Frame, Miss
+
+
+@pytest.fixture
+def executor():
+ instance = Executor()
+ try:
+ yield instance
+ finally:
+ instance.close()
+
+
+@pytest.fixture(params=[False, True], ids=["ordinary", "eager"])
+def scheduled(request, executor):
+ if request.param:
+ if not hasattr(asyncio, "eager_task_factory"):
+ pytest.skip("Eager task factories require Python 3.12+")
+ executor.loop.set_task_factory(asyncio.eager_task_factory)
+ return executor
+
+
+@pytest.mark.parametrize("remote", [False, True], ids=["source", "read"])
+def test_callback_budget_excludes_queued_work(executor, remote):
+ events = []
+ invoked = []
+
+ class Redis:
+ def read(self, request, context):
+ invoked.append(("read", executor.clock.monotonic_ms()))
+ return Frame(int(executor.clock.wall_ms()), '"cached"')
+
+ def provider(key):
+ executor.loop.call_soon(executor.clock.consume, 20)
+
+ def source():
+ invoked.append(("source", executor.clock.monotonic_ms()))
+ return "source"
+
+ cache = DialCache(
+ clock=executor.clock,
+ redis=Redis() if remote else None,
+ policy_provider=provider,
+ metrics=events.append,
+ )
+ with cache.enable():
+ value = executor.finish(
+ cache.get_or_load(
+ source,
+ key="a",
+ key_type="entity",
+ use_case="origin",
+ fallback_timeout_ms=10,
+ default_config=Policy(ttl_sec={"remote": 10}, remote_read_timeout_ms=10, coalesce=False),
+ )
+ )
+ assert value == ("cached" if remote else "source")
+ assert invoked == [("read" if remote else "source", 20)]
+ assert not [e for e in events if e["event"] == "error"]
+ if not remote:
+ assert [e["seconds"] for e in events if e["event"] == "fallback"] == [0]
+
+
+@pytest.mark.parametrize("kind", ["success", "failure", "before_await"])
+@pytest.mark.parametrize("remote", [False, True], ids=["source", "read"])
+def test_callback_budget_counts_synchronous_work(scheduled, kind, remote):
+ executor = scheduled
+ events, signals, writes = [], [], []
+
+ def work():
+ executor.clock.consume(10)
+ if kind == "failure":
+ raise ValueError("raw callback failure")
+ return Frame(int(executor.clock.wall_ms()), '"late"') if remote else "late"
+
+ async def asynchronous_work():
+ value = work()
+ await asyncio.sleep(0)
+ return value
+
+ class Redis:
+ def read(self, request, context):
+ context.signal.add_callback(lambda: signals.append("abort"))
+ return asynchronous_work() if kind == "before_await" else work()
+
+ def write(self, request):
+ writes.append(request)
+
+ cache = DialCache(clock=executor.clock, redis=Redis() if remote else None, metrics=events.append)
+ with cache.enable():
+ call = cache.get_or_load(
+ (lambda: "source") if remote else (asynchronous_work if kind == "before_await" else work),
+ key="a",
+ key_type="entity",
+ use_case="budget",
+ fallback_timeout_ms=10,
+ default_config=Policy(ttl_sec={"remote": 10}, remote_read_timeout_ms=10),
+ )
+ if remote:
+ assert executor.finish(call) == "source"
+ assert signals == ["abort"]
+ assert writes == []
+ assert [e["error"] for e in events if e["event"] == "error"] == ["cache_read_timeout"]
+ else:
+ with pytest.raises(FallbackTimeoutError):
+ executor.finish(call)
+
+
+@pytest.mark.parametrize("ramp", [10**1000, -(10**1000)])
+@pytest.mark.parametrize("dark", [False, True])
+def test_invalid_shadow_ramp_preserves_normal_result(executor, ramp, dark):
+ events, writes = [], []
+
+ class Redis:
+ def read(self, request, context):
+ return Frame(int(executor.clock.wall_ms()), '"cached"')
+
+ def write(self, request):
+ writes.append(request)
+
+ cache = DialCache(
+ clock=executor.clock,
+ redis=Redis(),
+ metrics=events.append,
+ policy_provider=lambda key: {"shadow": {"ramp": ramp}},
+ )
+ with cache.enable():
+ result = executor.finish(
+ cache.get_or_load(
+ lambda: "source",
+ key="a",
+ key_type="entity",
+ use_case="invalid-shadow",
+ default_config=Policy(ttl_sec={"remote": 10}, ramp={"remote": 0 if dark else 100}),
+ )
+ )
+ assert result == ("source" if dark else "cached")
+ assert [e["error"] for e in events if e["event"] == "error"] == ["config_resolution"]
+ assert not [e for e in events if e["event"] == "shadowValidation"]
+ assert writes == []
+
+
+@pytest.mark.parametrize("policy", [Policy(request_local=True), Policy(ttl_sec={"local": 10})])
+async def test_explicit_self_adapter_separates_repository_instances(policy):
+ calls, adapters = [], []
+ cache = DialCache()
+
+ def adapt_self(repo):
+ adapters.append(repo.tenant)
+ return repo.tenant
+
+ class Repository:
+ def __init__(self, tenant):
+ self.tenant = tenant
+
+ @cache.cached(key_type="user", id_arg="uid", arg_adapters={"self": adapt_self}, default_config=policy)
+ async def get(self, uid):
+ calls.append(self.tenant)
+ return f"{self.tenant}:{uid}"
+
+ a, b = Repository("tenant-a"), Repository("tenant-b")
+ assert await a.get("42") == "tenant-a:42"
+ assert adapters == []
+ calls.clear()
+ with cache.enable():
+ assert await a.get("42") == "tenant-a:42"
+ assert await b.get("42") == "tenant-b:42"
+ assert await a.get("42") == "tenant-a:42"
+ assert calls == ["tenant-a", "tenant-b"]
+ assert adapters == ["tenant-a", "tenant-b", "tenant-a"]
+
+
+@pytest.mark.parametrize("ignore", [False, True])
+async def test_default_self_omission_and_explicit_ignore_are_preserved(ignore):
+ cache = DialCache()
+ calls = []
+
+ def forbidden(repo):
+ raise AssertionError("An explicitly ignored self must not be adapted")
+
+ class Repository:
+ @cache.cached(
+ key_type="user",
+ id_arg="uid",
+ arg_adapters={"self": forbidden} if ignore else None,
+ ignore_args=["self"] if ignore else (),
+ default_config=Policy(request_local=True),
+ )
+ async def get(self, uid):
+ calls.append(self)
+ return uid
+
+ with cache.enable():
+ assert await Repository().get("42") == await Repository().get("42") == "42"
+ assert len(calls) == 1
+
+
+@pytest.mark.parametrize("enabled", [False, True])
+@pytest.mark.parametrize("object_observer", [False, True])
+def test_async_observer_is_not_started_or_retained(executor, enabled, object_observer):
+ started = []
+ gate = executor.future()
+
+ async def observer(event):
+ started.append(event)
+ await gate
+
+ class Observer:
+ observe = staticmethod(observer)
+
+ cache = DialCache(clock=executor.clock, metrics=Observer() if object_observer else observer)
+ with cache.enable(enabled):
+ for _ in range(20):
+ assert (
+ executor.finish(
+ cache.get_or_load(
+ lambda: 1,
+ key="a",
+ key_type="entity",
+ use_case="observer",
+ default_config=Policy(request_local=True),
+ )
+ )
+ == 1
+ )
+ executor.drain()
+ assert started == []
+ assert not asyncio.all_tasks(executor.loop)
+
+
+@pytest.mark.parametrize("layers", ["none", "request", "process", "nested"])
+@pytest.mark.parametrize("mode", ["fill", "compare", "served"])
+def test_shadow_synchronous_phases_follow_caller_continuation(scheduled, layers, mode):
+ executor = scheduled
+ events = []
+ gate = executor.future()
+
+ class Redis:
+ def read(self, request, context):
+ events.append("read")
+ return Miss("value_absent") if mode == "fill" else Frame(int(executor.clock.wall_ms()), "1")
+
+ def write(self, request):
+ events.append("write")
+
+ class Serializer:
+ def load(self, payload):
+ events.append("decode")
+ return 1
+
+ def dump(self, value):
+ events.append("dump")
+ return "1"
+
+ async def source():
+ if mode != "served":
+ await gate
+ events.append("source")
+ return 1
+
+ def compare(a, b):
+ events.append("compare")
+ return a == b
+
+ policy = Policy(
+ request_local=layers in ("request", "nested"),
+ ttl_sec={"remote": 10, **({"local": 10} if layers in ("process", "nested") else {})},
+ ramp={"remote": 100 if mode == "served" else 0},
+ shadow={"ramp": 100},
+ )
+ cache = DialCache(clock=executor.clock, redis=Redis(), serializer=Serializer(), metrics=lambda e: None)
+
+ async def caller():
+ result = await cache.get_or_load(
+ source,
+ key="a",
+ key_type="entity",
+ use_case="defer",
+ default_config=policy,
+ shadow_comparator=compare,
+ )
+ events.append("returned")
+ return result
+
+ with cache.enable():
+ task = executor.task(caller())
+ executor.drain()
+ if mode != "served":
+ assert "read" in events and "source" not in events
+ gate.set_result(None)
+ executor.drain()
+ assert task.result() == 1
+ phase = "dump" if mode == "fill" else "compare"
+ assert events.index("returned") < events.index(phase)
+ if mode == "served":
+ assert events.index("returned") < events.index("source")
+
+
+@pytest.mark.parametrize("cancel_first", [False, True])
+def test_shadow_waits_for_cross_scope_and_late_followers(scheduled, cancel_first):
+ executor = scheduled
+ gate = executor.future()
+ events, followers = [], []
+ joining = False
+ spawned = False
+
+ class Redis:
+ def read(self, request, context):
+ return Miss("value_absent")
+
+ def write(self, request):
+ events.append("write")
+
+ class Serializer:
+ def dump(self, value):
+ events.append("dump")
+ return "1"
+
+ def observe(event):
+ nonlocal spawned
+ if joining and not spawned and event["event"] == "request" and event.get("layer") == "request_local":
+ spawned = True
+ followers.append(executor.task(caller("C")))
+
+ cache = DialCache(clock=executor.clock, redis=Redis(), serializer=Serializer(), metrics=observe)
+ policy = Policy(
+ request_local=True, ttl_sec={"local": 10, "remote": 10}, ramp={"remote": 0}, shadow={"ramp": 100}
+ )
+
+ async def source():
+ events.append("source")
+ await gate
+ return 1
+
+ async def caller(name):
+ value = await cache.get_or_load(
+ source, key="a", key_type="entity", use_case="followers", default_config=policy
+ )
+ events.append(name)
+ return value
+
+ with cache.enable():
+ first = executor.task(caller("A"))
+ executor.drain()
+ if cancel_first:
+ first.cancel()
+ executor.drain()
+ assert first.cancelled()
+ with cache.enable():
+ joining = True
+ second = executor.task(caller("B"))
+ executor.drain()
+ late = executor.task(caller("D"))
+ executor.drain()
+ assert len(followers) == 1
+ gate.set_result(None)
+ executor.drain()
+ assert second.result() == late.result() == followers[0].result() == 1
+ assert events.count("source") == 1
+ for name in ("B", "C", "D") if cancel_first else ("A", "B", "C", "D"):
+ assert events.index(name) < events.index("dump")
+ assert cache.get_coalescing_state()["process"]["active_leaders"] == 0
+
+
+def test_independent_same_key_source_does_not_delay_completed_shadow(executor):
+ gates = [executor.future(), executor.future()]
+ events = []
+
+ class Redis:
+ def read(self, request, context):
+ return Miss("value_absent")
+
+ def write(self, request):
+ events.append("write")
+
+ cache = DialCache(clock=executor.clock, redis=Redis(), metrics=lambda event: None)
+ policy = Policy(
+ request_local=True, coalesce=False, ttl_sec={"remote": 10}, ramp={"remote": 0}, shadow={"ramp": 100}
+ )
+
+ async def caller(index):
+ value = await cache.get_or_load(
+ lambda: gates[index], key="a", key_type="entity", use_case="independent", default_config=policy
+ )
+ events.append(index)
+ return value
+
+ with cache.enable():
+ first, second = executor.task(caller(0)), executor.task(caller(1))
+ executor.drain()
+ gates[0].set_result(1)
+ executor.drain()
+ assert first.result() == 1 and not second.done()
+ assert events == [0, "write"]
+ gates[1].set_result(2)
+ executor.drain()
+ assert second.result() == 2
+
+
+@pytest.mark.parametrize("served", [False, True])
+def test_shadow_delivery_wait_preserves_budget_origin(executor, served):
+ events = []
+
+ class Redis:
+ def read(self, request, context):
+ return Frame(int(executor.clock.wall_ms()), "1") if served else Miss("value_absent")
+
+ def write(self, request):
+ events.append({"event": "write"})
+
+ cache = DialCache(clock=executor.clock, redis=Redis(), metrics=events.append)
+
+ async def caller():
+ value = await cache.get_or_load(
+ lambda: 1,
+ key="a",
+ key_type="entity",
+ use_case="delivery-budget",
+ fallback_timeout_ms=10,
+ default_config=Policy(
+ ttl_sec={"remote": 10}, ramp={"remote": 100 if served else 0}, shadow={"ramp": 100}
+ ),
+ )
+ executor.clock.consume(10)
+ return value
+
+ with cache.enable():
+ assert executor.finish(caller()) == 1
+ assert [e["outcome"] for e in events if e["event"] == "shadowValidation"] == [
+ "match" if served else "timeout"
+ ]
+ assert not [e for e in events if e["event"] == "write"]
+
+
+@pytest.mark.parametrize("shadow", [False, True])
+def test_cancelled_follower_markers_are_collectible_during_unbounded_source(scheduled, monkeypatch, shadow):
+ import gc
+ import weakref
+
+ executor = scheduled
+ references = []
+ create_future = executor.loop.create_future
+
+ def observed_future():
+ future = create_future()
+ references.append(weakref.ref(future))
+ return future
+
+ monkeypatch.setattr(executor.loop, "create_future", observed_future)
+ gate = executor.future()
+
+ class Redis:
+ def read(self, request, context):
+ return Miss("value_absent")
+
+ def write(self, request):
+ pass
+
+ cache = DialCache(clock=executor.clock, redis=Redis(), metrics=lambda event: None)
+ policy = Policy(
+ request_local=True,
+ ttl_sec={"local": 10, "remote": 10},
+ ramp={"remote": 0},
+ shadow={"ramp": 100 if shadow else 0},
+ )
+
+ async def call():
+ return await cache.get_or_load(
+ lambda: gate,
+ key="a",
+ key_type="entity",
+ use_case="churn",
+ fallback_timeout_ms=None,
+ default_config=policy,
+ )
+
+ with cache.enable():
+ first = executor.task(call())
+ executor.drain()
+ with cache.enable():
+ # This request leader transfers its delivery group into the process
+ # flight. Later request followers must not retain completed markers in
+ # either the original group or its destination.
+ second = executor.task(call())
+ executor.drain()
+ gc.collect()
+ baseline = sum(ref() is not None for ref in references)
+ for _ in range(3):
+ for _ in range(100):
+ follower = executor.task(call())
+ executor.drain()
+ follower.cancel()
+ executor.drain()
+ assert follower.cancelled()
+ gc.collect()
+ assert sum(ref() is not None for ref in references) <= baseline + 2
+ gate.set_result(1)
+ executor.drain()
+ assert first.result() == second.result() == 1
diff --git a/python/tests/test_shadow_deadline_retention.py b/python/tests/test_shadow_deadline_retention.py
new file mode 100644
index 00000000..10e69186
--- /dev/null
+++ b/python/tests/test_shadow_deadline_retention.py
@@ -0,0 +1,51 @@
+"""Native ownership check: abandoned shadow work releases retained raw bytes."""
+
+import gc
+import weakref
+
+from formal.executor import Executor
+
+from dialcache import DialCache
+from dialcache.protocol import Frame
+
+
+def test_timed_out_shadow_releases_frame_while_source_is_still_held():
+ executor = Executor()
+ retained = []
+ events = []
+ source = executor.future()
+
+ class EphemeralRedis:
+ async def read(self, request, context=None):
+ # Unlike a storage fake, this adapter retains no strong reference
+ # after returning its frame, so only native cache ownership remains.
+ frame = Frame(int(executor.clock.wall_ms()), "1")
+ retained.append(weakref.ref(frame))
+ return frame
+
+ cache = DialCache(redis=EphemeralRedis(), clock=executor.clock, metrics=events.append)
+
+ async def call():
+ with cache.enable():
+ return await cache.get_or_load(
+ lambda: source,
+ key_type="id",
+ key="1",
+ use_case="Retention",
+ fallback_timeout_ms=10,
+ default_config={"ttlSec": {"remote": 60}, "shadow": {"ramp": 100}},
+ )
+
+ try:
+ assert executor.finish(call()) == 1
+ assert len(retained) == 1
+ assert retained[0]() is not None
+ assert not source.done()
+ executor.clock.advance(10)
+ executor.drain()
+ assert [event["outcome"] for event in events if event["event"] == "shadowValidation"] == ["timeout"]
+ assert not source.done()
+ gc.collect()
+ assert retained[0]() is None
+ finally:
+ executor.close()
diff --git a/python/tests/test_validation_environment.py b/python/tests/test_validation_environment.py
new file mode 100644
index 00000000..f0f959d3
--- /dev/null
+++ b/python/tests/test_validation_environment.py
@@ -0,0 +1,132 @@
+"""Native validation must import the selected checkout with any prepared interpreter."""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import site
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[2]
+VALIDATION = ROOT / "formal/validation.mjs"
+NODE_BRIDGE = r"""
+import { pathToFileURL } from 'node:url';
+let input = ''; for await (const chunk of process.stdin) input += chunk;
+const { module, directory, environment, boundary } = JSON.parse(input);
+const { checkPrerequisites, validationPlan, executeSteps } = await import(pathToFileURL(module));
+if (boundary === 'prerequisites') {
+ checkPrerequisites('check-python', { directory, environment });
+} else {
+ const steps = validationPlan(boundary, { directory, environment })
+ .filter(step => step.command === environment.PYTHON);
+ if (steps.length === 0) throw new Error('Expected direct Python validation steps');
+ await executeSteps(steps, { directory, environment });
+}
+"""
+NATIVE_TEST = """
+from pathlib import Path
+import dialcache
+import validation_caller_marker
+
+def test_selected_checkout_import():
+ assert Path(dialcache.__file__).resolve() == (Path.cwd() / 'python/dialcache/__init__.py').resolve()
+ assert dialcache.DialCache.__module__ == 'dialcache.cache'
+ assert validation_caller_marker.VALUE == 'caller path retained'
+"""
+
+
+@pytest.mark.parametrize("boundary", ["check-python", "smoke", "prerequisites"])
+def test_validation_selects_checkout_over_foreign_editable(tmp_path, boundary):
+ node = os.environ.get("NODE") or shutil.which("node")
+ assert node is not None, "Node 24 is required by Python validation"
+ checkout = tmp_path / "selected-checkout"
+ tests = checkout / "python/tests"
+ tests.mkdir(parents=True)
+ shutil.copytree(
+ ROOT / "python/dialcache", checkout / "python/dialcache", ignore=shutil.ignore_patterns("__pycache__")
+ )
+ (tests / "test_conformance.py").write_text(NATIVE_TEST)
+ shutil.copyfile(ROOT / "python/pyproject.toml", checkout / "python/pyproject.toml")
+ # Both generated native commands run their exact argv, but this small
+ # checkout contains only the sentinel test, so this test cannot recurse.
+ foreign = tmp_path / "foreign-editable"
+ (foreign / "dialcache").mkdir(parents=True)
+ (foreign / "dialcache/__init__.py").write_text(
+ "raise RuntimeError('foreign editable dialcache was imported')\n"
+ )
+ env_dir = tmp_path / "venv"
+ # venv uses this test process's sys.executable; no pip or network is needed.
+ subprocess.run(
+ [sys.executable, "-m", "venv", "--without-pip", str(env_dir)],
+ check=True,
+ capture_output=True,
+ text=True,
+ )
+ python = env_dir / ("Scripts/python.exe" if os.name == "nt" else "bin/python")
+ library = Path(
+ subprocess.check_output(
+ [str(python), "-c", "import sysconfig; print(sysconfig.get_path('purelib'))"], text=True
+ ).strip()
+ )
+ # Reuse installed dependencies without processing the source environment's
+ # editable-install .pth. The foreign package is the only editable on sys.path.
+ (library / "probe.pth").write_text("\n".join([str(foreign), *site.getsitepackages()]) + "\n")
+ inherited = tmp_path / "caller-path"
+ inherited.mkdir()
+ (inherited / "validation_caller_marker.py").write_text("VALUE = 'caller path retained'\n")
+ environment = {
+ **os.environ,
+ "PYTHON": str(python),
+ "PYTHONPATH": os.pathsep.join([str(foreign), str(inherited)]),
+ "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1",
+ "PYTEST_ADDOPTS": "",
+ }
+ # Prove this interpreter really imports the foreign package absent the fix.
+ control_env = {key: value for key, value in environment.items() if key != "PYTHONPATH"}
+ control = subprocess.run(
+ [str(python), "-c", "import dialcache"], cwd=checkout, env=control_env, text=True, capture_output=True
+ )
+ assert control.returncode != 0 and "foreign editable dialcache was imported" in control.stderr
+ # Satisfy unrelated Node/package-manager prerequisites without installations.
+ if boundary == "prerequisites":
+ (checkout / "node_modules/typescript").mkdir(parents=True)
+ (checkout / "node_modules/typescript/package.json").write_text("{}")
+ package = json.loads((ROOT / "package.json").read_text())
+ (checkout / "package.json").write_text(json.dumps({"packageManager": package["packageManager"]}))
+ tools = tmp_path / "bin"
+ tools.mkdir()
+ corepack = tools / "corepack"
+ corepack.write_text(
+ f"#!{node}\nconsole.log({json.dumps(package['packageManager'].removeprefix('pnpm@'))});\n"
+ )
+ corepack.chmod(0o755)
+ environment["PATH"] = str(tools) + os.pathsep + environment.get("PATH", "")
+ result = subprocess.run(
+ [node, "--input-type=module", "-e", NODE_BRIDGE],
+ cwd=ROOT,
+ env=environment,
+ input=json.dumps(
+ {
+ "module": str(VALIDATION),
+ "directory": str(checkout),
+ "environment": environment,
+ "boundary": boundary,
+ }
+ ),
+ text=True,
+ capture_output=True,
+ timeout=30,
+ )
+ assert result.returncode == 0, result.stdout + result.stderr
+ if boundary == "check-python":
+ report = (checkout / "coverage/python/native.lcov").read_text()
+ sources = [line.removeprefix("SF:") for line in report.splitlines() if line.startswith("SF:")]
+ assert "python/dialcache/cache.py" in sources
+ assert all(
+ source.startswith("python/dialcache/") and (checkout / source).is_file() for source in sources
+ )
diff --git a/scripts/check-docs.mjs b/scripts/check-docs.mjs
index 4061f0ca..a8bb7a67 100644
--- a/scripts/check-docs.mjs
+++ b/scripts/check-docs.mjs
@@ -58,8 +58,11 @@ export function checkDocsSources(directory = root) {
if (!port) { failures.push(`${at}: snippet must use a registered, executed example file`); continue; }
if (port.id !== selected) failures.push(`${at}: ${port.id} example requires its matching LanguageContent`);
const code = readFileSync(path, 'utf8');
- const starts = [...code.matchAll(new RegExp(`^\\s*//\\s*#region ${region}\\s*$`, 'gm'))];
- const ends = [...code.matchAll(new RegExp(`^\\s*//\\s*#endregion ${region}\\s*$`, 'gm'))];
+ // Match VitePress's native region syntax. In Python the leading # is
+ // already the region marker: '# region', never '# #region'.
+ const regionMarker = path.endsWith('.py') ? '# ?' : '//\\s*#';
+ const starts = [...code.matchAll(new RegExp(`^\\s*${regionMarker}region ${region}\\s*$`, 'gm'))];
+ const ends = [...code.matchAll(new RegExp(`^\\s*${regionMarker}endregion ${region}\\s*$`, 'gm'))];
if (starts.length !== 1 || ends.length !== 1 || starts[0].index >= ends[0].index) failures.push(`${at}: missing, duplicate or unclosed region ${region} in ${input}`);
if (!regions.has(region)) regions.set(region, new Set());
regions.get(region).add(port.id);
diff --git a/scripts/check-docs.test.mjs b/scripts/check-docs.test.mjs
index 81c30693..5b2d2e71 100644
--- a/scripts/check-docs.test.mjs
+++ b/scripts/check-docs.test.mjs
@@ -29,20 +29,26 @@ test('rejects missing regions instead of letting VitePress silently show the who
const root = mkdtempSync(join(tmpdir(), 'dialcache-docs-'));
t.after(() => rmSync(root, { recursive: true, force: true }));
mkdirSync(join(root, 'docs/languages'), { recursive: true });
- const ports = ['typescript', 'go', 'rust'].map(id => ({ id, guide: `/languages/${id}`, example: `${id}.txt` }));
+ const ports = ['typescript', 'go', 'rust', 'python'].map(id => ({ id, guide: `/languages/${id}`, example: id === 'python' ? 'python.py' : `${id}.txt` }));
writeFileSync(join(root, 'docs/ports.json'), JSON.stringify(ports));
for (const port of ports) {
writeFileSync(join(root, `docs${port.guide}.md`), '# Install\n');
- writeFileSync(join(root, port.example), '// #region scope\nreal_code();\n// #endregion scope\n');
+ const marker = port.id === 'python' ? '# ' : '// #';
+ writeFileSync(join(root, port.example), `${marker}region scope\nreal_code();\n${marker}endregion scope\n`);
}
- const section = id => `\n\n<<< @/../${id}.txt#scope\n\n\n`;
+ const section = id => `\n\n<<< @/../${ports.find(port => port.id === id).example}#scope\n\n\n`;
writeFileSync(join(root, 'docs/concepts.md'), ports.map(port => section(port.id)).join('\n'));
- assert.deepEqual(checkDocsSources(root), { ports: 3, imports: 3 });
+ assert.deepEqual(checkDocsSources(root), { ports: 4, imports: 4 });
+ writeFileSync(join(root, 'python.py'), '# #region scope\nreal_code()\n# #endregion scope\n');
+ assert.throws(() => checkDocsSources(root), /missing, duplicate or unclosed region/);
+ writeFileSync(join(root, 'python.py'), '# region scope\nreal_code()\n# endregion scope\n');
writeFileSync(join(root, 'rust.txt'), 'code_without_the_named_region();\n');
assert.throws(() => checkDocsSources(root), /missing, duplicate or unclosed region/);
writeFileSync(join(root, 'rust.txt'), '// #region scope\nreal_code();\n// #endregion scope\n');
writeFileSync(join(root, 'docs/concepts.md'), section('typescript') + section('go'));
assert.throws(() => checkDocsSources(root), /scope missing rust/);
+ writeFileSync(join(root, 'docs/concepts.md'), ports.filter(port => port.id !== 'python').map(port => section(port.id)).join('\n'));
+ assert.throws(() => checkDocsSources(root), /scope missing python/);
writeFileSync(join(root, 'docs/concepts.md'), section('typescript').replace('language="typescript"', 'language="ruby"'));
assert.throws(() => checkDocsSources(root), /unknown language ruby/);
});
diff --git a/scripts/generate-docs.mjs b/scripts/generate-docs.mjs
index 4969e2d9..ccd2554c 100644
--- a/scripts/generate-docs.mjs
+++ b/scripts/generate-docs.mjs
@@ -28,6 +28,22 @@ if (!process.argv.includes('--catalogue-only')) {
run('cargo', ['doc', '--locked', '--all-features', '--no-deps'], resolve(root, 'rust'));
const metadata = JSON.parse(execFileSync('cargo', ['metadata', '--format-version=1', '--no-deps'], { cwd: resolve(root, 'rust'), encoding: 'utf8' }));
cpSync(resolve(metadata.target_directory, 'doc'), resolve(destination, 'rust'), { recursive: true });
+ const python = process.env.PYTHON ?? resolve(root, 'python/.venv/bin/python');
+ // Use this checkout even when PYTHON points at an editable installation in
+ // another directory. pydoc is part of the standard library.
+ const pythonModules = JSON.parse(execFileSync(python, ['-c', `import importlib, json, pydoc, sys
+sys.path.insert(0, sys.argv[1])
+names = ['dialcache', 'dialcache.cache', 'dialcache.config', 'dialcache.key', 'dialcache.serializer', 'dialcache.redis', 'dialcache.protocol', 'dialcache.metrics', 'dialcache.clock', 'dialcache.errors']
+print(json.dumps({name: pydoc.render_doc(importlib.import_module(name), renderer=pydoc.plaintext) for name in names}))
+`, resolve(root, 'python')], { cwd: root, encoding: 'utf8', maxBuffer: 8 * 1024 * 1024 }));
+ mkdirSync(resolve(destination, 'python'), { recursive: true });
+ writeFileSync(resolve(destination, 'python/index.html'), `
+
+DialCache Python API
+
+DialCache Python API
Generated with pydoc from ${revision.slice(0, 7)}. Use your browser's Find command to locate a symbol.
+${Object.keys(pythonModules).map(name => `- ${name}
`).join('')}
+${Object.entries(pythonModules).map(([name, text]) => ``).join('\n')}\n`);
writeFileSync(resolve(destination, 'revision.json'), JSON.stringify({ revision }, null, 2) + '\n');
}
@@ -55,7 +71,7 @@ const pages = ['---', 'editLink: false', '---', '', '# Behavior catalogue', '',
'These links describe registered evidence and its scope. They are not a fresh test result or a claim of exhaustive coverage. ' +
'See the [validation guide](' + source('formal/VALIDATION.md') + ') for how a completed run is established.', '',
'All supported ports replay the shared histories through their native drivers. ' +
- [link('TypeScript replay tests', 'typescript/test/formal-features.test.ts'), link('Go replay tests', 'go/feature_replay_test.go'), link('Rust replay tests', 'rust/tests/conformance.rs')].join(' · ') + '.', '',
+ [link('TypeScript replay tests', 'typescript/test/formal-features.test.ts'), link('Go replay tests', 'go/feature_replay_test.go'), link('Rust replay tests', 'rust/tests/conformance.rs'), link('Python replay tests', 'python/tests/test_conformance.py')].join(' · ') + '.', '',
'| Case | Behavior | Model and regression evidence | Shared replay evidence |',
'| --- | --- | --- | --- |'];
for (const item of inventory.cases) {
diff --git a/typescript/test/formal-exploration.test.ts b/typescript/test/formal-exploration.test.ts
index 09a8a74f..5d7dc8ed 100644
--- a/typescript/test/formal-exploration.test.ts
+++ b/typescript/test/formal-exploration.test.ts
@@ -28,7 +28,8 @@ const inventory: Entry[] = [
];
const packageName = "example.com/exploration";
const context = (language: string) => ({ kind: "exploration", language, createdAt: 1, inventory,
- specification: {}, implementation: {}, corpus: {} });
+ specification: {}, implementation: {}, corpus: language === "python" ? Object.fromEntries(inventory.filter(entry => entry.path)
+ .map(entry => [entry.path!, createHash("sha256").update("synthetic exploratory history").digest("hex")])) : {} });
const { selectedProfiles } = await import(new URL("../../formal/witnesses.mjs", import.meta.url).href) as { selectedProfiles(selection: string): string[] };
function tsReport(directory: string, failure?: string) {
const ancestorTitles = ["generated recovery conformance"];
@@ -53,6 +54,19 @@ function rustReport(failure?: string) {
records.push({ kind: "finish", status: failure ? "failed" : "passed", finishedAt: 20, cases: inventory.length, failed: failure ? 1 : 0 });
return records.map(record => JSON.stringify(record)).join("\n");
}
+function pythonReport(failure?: string) {
+ const records: Record[] = [{ schemaVersion: 1, kind: "start", implementation: "python",
+ scope: "conformance", selection: "generated", partial: false, startedAt: 10 }];
+ for (const [index, entry] of inventory.entries()) {
+ const failed = entry.id === failure;
+ records.push({ kind: "case", id: nativeBinding(entry, "python"), status: failed ? "failed" : "passed",
+ startedAt: 11 + index, finishedAt: 12 + index,
+ ...(entry.path ? { historySha256: createHash("sha256").update("synthetic exploratory history").digest("hex") } : {}),
+ ...(failed ? { message: "Observation mismatch\nexpected: 1\nactual: 2" } : {}) });
+ }
+ records.push({ kind: "finish", status: failure ? "failed" : "passed", finishedAt: 20, cases: inventory.length, failed: failure ? 1 : 0 });
+ return records.map(record => JSON.stringify(record)).join("\n");
+}
function goReport(failure?: string) {
const events: Record[] = [];
const event = (Action: string, Test?: string) => events.push({ Action, ...(Test ? { Test } : {}), Package: packageName, Time: new Date(10 + events.length).toISOString() });
@@ -85,7 +99,7 @@ function writeWitnessInventory(directory: string, profiles: readonly string[]) {
writeFileSync(join(directory, "formal/coverage-witnesses.json"), JSON.stringify(Object.fromEntries(profiles.map(profile => [profile, ["required"]]))));
}
-function savedFixture(directory: string) {
+function savedFixture(directory: string, languages: readonly unknown[] = ["typescript", "go", "rust"], results: readonly unknown[] = languages) {
const saved = join(directory, "saved"), workspace = join(saved, "workspace");
mkdirSync(join(workspace, "formal"), { recursive: true });
mkdirSync(join(directory, "node_modules"));
@@ -101,12 +115,12 @@ function savedFixture(directory: string) {
"formal/execution.json": JSON.stringify({ models: [{ profile: "effects" }] }),
"formal/coverage-witnesses.json": JSON.stringify({ effects: ["required"] }),
"formal/explore.mjs": `import { writeFileSync } from 'node:fs';
- export const explorationPlan = (directory, seed) => [{ directory, seed, runner: 'saved' }];
+ export const explorationPlan = (directory, seed) => ${JSON.stringify(languages)}.map(nativeReport => ({ directory, seed, runner: 'saved', nativeReport }));
export async function runExplorationSteps(plan, options) {
writeFileSync(options.directory + '/.formal-traces/saved-runner.json', JSON.stringify(plan));
// A saved run's evaluator also has to leave a completed witness report behind.
writeFileSync(options.directory + '/.formal-traces/witness-report.json', ${JSON.stringify(JSON.stringify(completedWitnessReport("0x2a", ["effects"])))});
- return [{ language: 'typescript', status: 'passed' }, { language: 'go', status: 'passed' }, { language: 'rust', status: 'passed' }];
+ return ${JSON.stringify(results)}.map(language => ({ language, status: 'passed' }));
}`,
"formal/validation.mjs": `import { mkdirSync, writeFileSync } from 'node:fs';
export function checkPrerequisites(target, { directory }) {
@@ -154,13 +168,14 @@ describe("isolated exploratory validation", () => {
["formal/generated-fixtures.mjs", "--check"],
["formal/witnesses.mjs", "evaluate", "--profile", "all"],
["formal/check-go-parity.mjs"],
+ ["formal/run-python-replay.mjs", "--generated", "--scenarios", "--complete", "--report", ".formal-traces/python-replay.jsonl"],
]);
const replays = plan.filter(step => step.nativeReport);
- expect(replays.map(step => step.nativeReport)).toEqual(["typescript", "go", "rust"]);
- for (const step of replays) expect(step.env?.DIALCACHE_FEATURE_TRACE_DIR).toBe(`${directory}/.formal-traces/features`);
- expect(plan.filter(step => step.explorationContext).map(step => step.explorationContext)).toEqual(["typescript", "go", "rust"]);
+ expect(replays.map(step => step.nativeReport)).toEqual(["typescript", "go", "rust", "python"]);
+ for (const step of replays.filter(item => item.nativeReport !== "python")) expect(step.env?.DIALCACHE_FEATURE_TRACE_DIR).toBe(`${directory}/.formal-traces/features`);
+ expect(plan.filter(step => step.explorationContext).map(step => step.explorationContext)).toEqual(["typescript", "go", "rust", "python"]);
expect(plan.some(step => step.args?.includes("formal/check-go-replay.mjs") || step.args?.includes("formal/check-rust-replay.mjs")
- || step.args?.includes("formal/conformance-adapters.mjs"))).toBe(false);
+ || step.args?.includes("formal/check-python-replay.mjs") || step.args?.includes("formal/conformance-adapters.mjs"))).toBe(false);
expect(plan.some(step => step.args?.[0] === "formal/conformance.mjs" && step.args[1] === "check")).toBe(false);
});
@@ -187,9 +202,9 @@ describe("isolated exploratory validation", () => {
} finally { rmSync(directory, { recursive: true, force: true }); }
});
- it.each(["typescript", "go", "rust"])("classifies exact %s witness leaves separately from replay failures", language => {
+ it.each(["typescript", "go", "rust", "python"])("classifies exact %s witness leaves separately from replay failures", language => {
const native = (failure?: string) => language === "typescript" ? JSON.stringify(tsReport("/snapshot", failure))
- : language === "go" ? goReport(failure) : rustReport(failure);
+ : language === "go" ? goReport(failure) : language === "rust" ? rustReport(failure) : pythonReport(failure);
expect(nativeExplorationResult(language, native(), context(language), "/snapshot", packageName).status).toBe("passed");
expect(nativeExplorationResult(language, native("witness/recovery"), context(language), "/snapshot", packageName)).toMatchObject({
status: "witness-check-failure", witnessFailures: ["witness/recovery"], caseFailures: [],
@@ -267,7 +282,7 @@ describe("isolated exploratory validation", () => {
await expect(explore("42", { directory, run: async () => {
const output = join(directory, ".formal-traces/exploration", readdirSync(join(directory, ".formal-traces/exploration"))[0]!);
writeFileSync(join(output, "workspace/rule.qnt"), "changed");
- return [{ language: "typescript", status: "passed" }, { language: "go", status: "passed" }, { language: "rust", status: "passed" }];
+ return [{ language: "typescript", status: "passed" }, { language: "go", status: "passed" }, { language: "rust", status: "passed" }, { language: "python", status: "passed" }];
} })).rejects.toThrow(/changed/);
const output = join(directory, ".formal-traces/exploration", readdirSync(join(directory, ".formal-traces/exploration"))[0]!);
expect(readFileSync(join(directory, "rule.qnt"), "utf8")).toBe("original");
@@ -277,7 +292,7 @@ describe("isolated exploratory validation", () => {
} finally { rmSync(directory, { recursive: true, force: true }); }
});
- it("returns a nonzero failure after both ports report incomplete witness coverage", async () => {
+ it("returns a nonzero failure after every port reports incomplete witness coverage", async () => {
const directory = mkdtempSync(join(tmpdir(), "dialcache-exploration-coverage-"));
try {
execFileSync("git", ["init", "--quiet"], { cwd: directory });
@@ -285,46 +300,73 @@ describe("isolated exploratory validation", () => {
writeFileSync(join(directory, ".gitignore"), ".formal-traces/\n");
writeWitnessInventory(directory, selectedProfiles("all"));
await expect(explore("42", { directory, run: async () => [
- { language: "typescript", status: "witness-check-failure" }, { language: "go", status: "witness-check-failure" }, { language: "rust", status: "witness-check-failure" },
+ { language: "typescript", status: "witness-check-failure" }, { language: "go", status: "witness-check-failure" }, { language: "rust", status: "witness-check-failure" }, { language: "python", status: "witness-check-failure" },
] })).rejects.toThrow(/witness-check-failure/);
const output = join(directory, ".formal-traces/exploration", readdirSync(join(directory, ".formal-traces/exploration"))[0]!);
expect(JSON.parse(readFileSync(join(output, "report.json"), "utf8"))).toMatchObject({
kind: "exploration", acceptance: false, status: "witness-check-failure", sourcesUnchanged: true,
- native: [{ language: "typescript" }, { language: "go" }, { language: "rust" }],
+ native: [{ language: "typescript" }, { language: "go" }, { language: "rust" }, { language: "python" }],
});
expect(existsSync(join(output, "workspace/node_modules"))).toBe(false);
expect(existsSync(join(output, "workspace/typescript/node_modules"))).toBe(false);
} finally { rmSync(directory, { recursive: true, force: true }); }
});
- it("replays saved bytes with their own runner and prerequisites without Git, preserving original evidence", async () => {
- const directory = mkdtempSync(join(tmpdir(), "dialcache-exploration-reproduce-"));
+ it.each([{ languages: ["typescript", "go", "rust"] }, { languages: ["typescript", "go", "rust", "python"] }])(
+ "replays saved $languages bytes with their own runner and prerequisites without Git, preserving original evidence", async ({ languages }) => {
+ const directory = mkdtempSync(join(tmpdir(), "dialcache-exploration-reproduce-"));
+ try {
+ const saved = savedFixture(directory, languages), original = readFileSync(saved.path, "utf8");
+ mkdirSync(join(directory, "formal"));
+ writeFileSync(join(directory, "formal/explore.mjs"), 'throw new Error("new checkout runner must not execute")');
+ writeFileSync(join(directory, "formal/validation.mjs"), 'throw new Error("new checkout prerequisites must not execute")');
+ const output = await replayExploration(saved.path, { directory });
+ const report = JSON.parse(readFileSync(join(output, "report.json"), "utf8"));
+ expect(report).toMatchObject({ status: "passed", acceptance: false, seed: "0x2a", baseRevision: saved.report.baseRevision,
+ sources: saved.report.sources, replayOrigin: { path: realpathSync(saved.path),
+ reportSha256: createHash("sha256").update(original).digest("hex") } });
+ expect(JSON.parse(readFileSync(join(output, "workspace/.formal-traces/saved-runner.json"), "utf8"))).toEqual(
+ languages.map(nativeReport => ({ directory: join(output, "workspace"), seed: "0x2a", runner: "saved", nativeReport })),
+ );
+ expect(report.native.map((result: Result) => result.language)).toEqual(languages);
+ expect(readFileSync(join(output, "workspace/.formal-traces/saved-prerequisites.txt"), "utf8")).toBe("explore");
+ // The saved inventory has one profile while this checkout schedules many:
+ // the replay was judged against the snapshot's inventory, not the checkout's.
+ expect(selectedProfiles("all").length).toBeGreaterThan(1);
+ expect(Object.keys(report.witnesses.profiles)).toEqual(["effects"]);
+ expect(readFileSync(saved.path, "utf8")).toBe(original);
+ expect(readFileSync(join(saved.workspace, ".formal-traces/original-evidence.txt"), "utf8")).toBe("retain original native evidence");
+ for (const [path, content] of Object.entries(saved.sources)) expect(readFileSync(join(saved.workspace, path), "utf8")).toBe(content);
+ expect(existsSync(join(output, "workspace/node_modules"))).toBe(false);
+ expect(existsSync(join(output, "workspace/typescript/node_modules"))).toBe(false);
+ } finally { rmSync(directory, { recursive: true, force: true }); }
+ },
+ );
+
+ it.each([
+ { name: "missing", results: ["typescript", "go"] },
+ { name: "duplicate", results: ["typescript", "go", "go"] },
+ { name: "unexpected", results: ["typescript", "go", "rust", "python"] },
+ ])("rejects $name results against the saved plan's port inventory", async ({ results }) => {
+ const directory = mkdtempSync(join(tmpdir(), "dialcache-exploration-port-results-"));
try {
- const saved = savedFixture(directory), original = readFileSync(saved.path, "utf8");
- mkdirSync(join(directory, "formal"));
- writeFileSync(join(directory, "formal/explore.mjs"), 'throw new Error("new checkout runner must not execute")');
- writeFileSync(join(directory, "formal/validation.mjs"), 'throw new Error("new checkout prerequisites must not execute")');
- const output = await replayExploration(saved.path, { directory });
- const report = JSON.parse(readFileSync(join(output, "report.json"), "utf8"));
- expect(report).toMatchObject({ status: "passed", acceptance: false, seed: "0x2a", baseRevision: saved.report.baseRevision,
- sources: saved.report.sources, replayOrigin: { path: realpathSync(saved.path),
- reportSha256: createHash("sha256").update(original).digest("hex") } });
- expect(JSON.parse(readFileSync(join(output, "workspace/.formal-traces/saved-runner.json"), "utf8"))).toEqual([
- { directory: join(output, "workspace"), seed: "0x2a", runner: "saved" },
- ]);
- expect(readFileSync(join(output, "workspace/.formal-traces/saved-prerequisites.txt"), "utf8")).toBe("explore");
- // The saved inventory has one profile while this checkout schedules many:
- // the replay was judged against the snapshot's inventory, not the checkout's.
- expect(selectedProfiles("all").length).toBeGreaterThan(1);
- expect(Object.keys(report.witnesses.profiles)).toEqual(["effects"]);
- expect(readFileSync(saved.path, "utf8")).toBe(original);
- expect(readFileSync(join(saved.workspace, ".formal-traces/original-evidence.txt"), "utf8")).toBe("retain original native evidence");
- for (const [path, content] of Object.entries(saved.sources)) expect(readFileSync(join(saved.workspace, path), "utf8")).toBe(content);
- expect(existsSync(join(output, "workspace/node_modules"))).toBe(false);
- expect(existsSync(join(output, "workspace/typescript/node_modules"))).toBe(false);
+ const saved = savedFixture(directory, ["typescript", "go", "rust"], results);
+ await expect(replayExploration(saved.path, { directory })).rejects.toThrow(/did not finish every native port/);
} finally { rmSync(directory, { recursive: true, force: true }); }
});
+ it.each([{ languages: [] }, { languages: ["go", "go"] }, { languages: [""] }, { languages: [42] }])(
+ "rejects invalid saved port inventory $languages before native execution", async ({ languages }) => {
+ const directory = mkdtempSync(join(tmpdir(), "dialcache-exploration-port-plan-"));
+ try {
+ const saved = savedFixture(directory, languages);
+ await expect(replayExploration(saved.path, { directory })).rejects.toThrow(/invalid native port inventory/);
+ const output = join(directory, ".formal-traces/exploration", readdirSync(join(directory, ".formal-traces/exploration"))[0]!);
+ expect(existsSync(join(output, "workspace/.formal-traces/saved-runner.json"))).toBe(false);
+ } finally { rmSync(directory, { recursive: true, force: true }); }
+ },
+ );
+
it.each(["changed", "deleted"])("rejects a %s saved source before rerunning", async change => {
const directory = mkdtempSync(join(tmpdir(), "dialcache-exploration-drift-"));
try {
@@ -355,7 +397,7 @@ describe("isolated exploratory validation", () => {
expect(existsSync(join(output, "workspace/typescript/node_modules"))).toBe(false);
} finally { rmSync(directory, { recursive: true, force: true }); }
});
- it("fails exploration after both ports replay when the witness baseline gate tripped", async () => {
+ it("fails exploration after every port replays when the witness baseline gate tripped", async () => {
const directory = mkdtempSync(join(tmpdir(), "dialcache-exploration-witness-report-"));
try {
execFileSync("git", ["init", "--quiet"], { cwd: directory });
@@ -368,10 +410,10 @@ describe("isolated exploratory validation", () => {
const workspace = join(directory, ".formal-traces/exploration", readdirSync(join(directory, ".formal-traces/exploration"))[0]!, "workspace");
mkdirSync(join(workspace, ".formal-traces"), { recursive: true });
writeFileSync(join(workspace, ".formal-traces/witness-report.json"), JSON.stringify(witnesses));
- replays = 2;
- return [{ language: "typescript", status: "passed" }, { language: "go", status: "passed" }, { language: "rust", status: "passed" }];
+ replays = 4;
+ return [{ language: "typescript", status: "passed" }, { language: "go", status: "passed" }, { language: "rust", status: "passed" }, { language: "python", status: "passed" }];
} })).rejects.toThrow(/coverage-gate-failure[\s\S]*reply:13 reached by 1 sampled histories/);
- expect(replays).toBe(2);
+ expect(replays).toBe(4);
const output = join(directory, ".formal-traces/exploration", readdirSync(join(directory, ".formal-traces/exploration"))[0]!);
expect(JSON.parse(readFileSync(join(output, "report.json"), "utf8"))).toMatchObject({ status: "coverage-gate-failure", witnesses, sourcesUnchanged: true });
} finally { rmSync(directory, { recursive: true, force: true }); }
@@ -384,7 +426,7 @@ describe("isolated exploratory validation", () => {
["written by the report command", JSON.stringify({ ...completedWitnessReport("0x2a"), command: "report" }), /not a completed evaluation/],
["judged under the pinned seed", JSON.stringify(completedWitnessReport("0xd1a1ca")), /judged under seed 0xd1a1ca, not this exploration's 0x2a/],
["missing a scheduled profile", JSON.stringify({ ...completedWitnessReport("0x2a"), profiles: { effects: {} } }), /covers no evaluation of/],
- ])("fails as infrastructure when the witness report is %s although both ports passed", async (_name, contents, message) => {
+ ])("fails as infrastructure when the witness report is %s although every port passed", async (_name, contents, message) => {
const directory = mkdtempSync(join(tmpdir(), "dialcache-exploration-witness-missing-"));
try {
execFileSync("git", ["init", "--quiet"], { cwd: directory });
@@ -395,7 +437,7 @@ describe("isolated exploratory validation", () => {
const workspace = join(directory, ".formal-traces/exploration", readdirSync(join(directory, ".formal-traces/exploration"))[0]!, "workspace");
mkdirSync(join(workspace, ".formal-traces"), { recursive: true });
if (contents !== undefined) writeFileSync(join(workspace, ".formal-traces/witness-report.json"), contents);
- return [{ language: "typescript", status: "passed" }, { language: "go", status: "passed" }, { language: "rust", status: "passed" }];
+ return [{ language: "typescript", status: "passed" }, { language: "go", status: "passed" }, { language: "rust", status: "passed" }, { language: "python", status: "passed" }];
} })).rejects.toThrow(message);
const output = join(directory, ".formal-traces/exploration", readdirSync(join(directory, ".formal-traces/exploration"))[0]!);
expect(JSON.parse(readFileSync(join(output, "report.json"), "utf8")).status).toBe("infrastructure-failure");
@@ -422,7 +464,7 @@ describe("isolated exploratory validation", () => {
const workspace = join(directory, ".formal-traces/exploration", readdirSync(join(directory, ".formal-traces/exploration"))[0]!, "workspace");
mkdirSync(join(workspace, ".formal-traces"), { recursive: true });
writeFileSync(join(workspace, ".formal-traces/witness-report.json"), JSON.stringify(witnesses));
- return [{ language: "typescript", status: "passed" }, { language: "go", status: "passed" }, { language: "rust", status: "passed" }];
+ return [{ language: "typescript", status: "passed" }, { language: "go", status: "passed" }, { language: "rust", status: "passed" }, { language: "python", status: "passed" }];
} });
expect(JSON.parse(readFileSync(join(output, "report.json"), "utf8"))).toMatchObject({ status: "passed", witnesses });
} finally { rmSync(directory, { recursive: true, force: true }); }
diff --git a/typescript/test/formal-rust-replay.test.ts b/typescript/test/formal-rust-replay.test.ts
index 7c6cce51..ae1c28fd 100644
--- a/typescript/test/formal-rust-replay.test.ts
+++ b/typescript/test/formal-rust-replay.test.ts
@@ -43,7 +43,7 @@ describe("Rust replay report gate", () => {
it("binds every inventory entry to its own id", () => {
for (const entry of inventory) expect(nativeBinding(entry, "rust")).toBe(entry.id);
expect(nativeBinding(inventory[0]!, "go")).not.toBe(inventory[0]!.id);
- expect(() => nativeBinding(inventory[0]!, "zig")).toThrow(/TypeScript, Go and Rust/);
+ expect(() => nativeBinding(inventory[0]!, "zig")).toThrow(/TypeScript, Go, Rust and Python/);
});
it("accepts a complete report and summarizes it by category and profile", () => {
@@ -143,7 +143,7 @@ describe("Rust validation lanes", () => {
expect(step.env, step.label).toBeUndefined();
expect(step.cwd, step.label).toBe("rust");
}
- expect(validationPlan("check", { directory })).toEqual(["check-ts", "check-go", "check-rust", "docs", "audit"].flatMap(target => validationPlan(target, { directory })));
+ expect(validationPlan("check", { directory })).toEqual(["check-ts", "check-go", "check-rust", "check-python", "docs", "audit"].flatMap(target => validationPlan(target, { directory })));
});
it("replays the complete corpus against the shared evidence and adapts the harness report into a completion", () => {
@@ -170,19 +170,21 @@ describe("Rust validation lanes", () => {
expect(plan.some(step => step.args?.some(argument => /\.formal-traces\/(ts|go)-/.test(argument)))).toBe(false);
const go = validationPlan("formal-go", { directory }).find(step => step.command === "go" && step.env)!;
for (const key of Object.keys(go.env!)) expect(replay.env![key], key).toBe(go.env![key]);
- expect(validationPlan("formal", { directory }).slice(-plan.length)).toEqual(plan);
+ const aggregate = validationPlan("formal", { directory });
+ const rustStart = aggregate.findIndex(step => step.label === plan[0]!.label);
+ expect(rustStart).toBeGreaterThanOrEqual(0);
+ expect(aggregate.slice(rustStart, rustStart + plan.length)).toEqual(plan);
});
it("adds the smoke conformance run in default mode with no corpus selectors", () => {
const smoke = validationPlan("smoke", { directory });
- expect(smoke.at(-1)).toEqual({ label: "Replay committed Rust fixtures", command: "cargo", args: ["test", "--all-features", "--test", "conformance"], cwd: "rust" });
- expect(smoke.filter(step => step.command === "cargo")).toHaveLength(1);
+ expect(smoke.filter(step => step.command === "cargo")).toEqual([{ label: "Replay committed Rust fixtures", command: "cargo", args: ["test", "--all-features", "--test", "conformance"], cwd: "rust" }]);
});
it("runs the real-server integration binary only through the integration lane, which selects its ignored tests", () => {
const lane = validationPlan("integration-rust", { directory });
expect(lane).toEqual([{ label: "Run Rust Redis/Valkey/Cluster integrations", command: "cargo", args: ["test", "--all-features", "--test", "redis_integration", "--", "--ignored"], cwd: "rust" }]);
- expect(validationPlan("integration", { directory })).toEqual(["integration-ts", "integration-go", "integration-rust"].flatMap(target => validationPlan(target, { directory })));
+ expect(validationPlan("integration", { directory })).toEqual(["integration-ts", "integration-go", "integration-rust", "integration-python"].flatMap(target => validationPlan(target, { directory })));
for (const target of ["check-rust", "smoke", "formal-rust"]) expect(validationPlan(target, { directory }).some(step => step.args?.includes("--ignored")), target).toBe(false);
});
@@ -200,7 +202,8 @@ describe("Rust validation lanes", () => {
tool("corepack", 'console.log("10.33.0")');
tool("go", 'console.log("go version go1.27.1 test/test")');
tool("quint", 'console.log("0.32.0")');
- const environment = { ...process.env, PATH: `${join(temporary, "bin")}${delimiter}${process.env.PATH ?? ""}` };
+ tool("python", 'if (process.argv.includes("--version")) console.log("Python 3.11.9")');
+ const environment = { ...process.env, PYTHON: join(temporary, "bin", "python"), PATH: `${join(temporary, "bin")}${delimiter}${process.env.PATH ?? ""}` };
const options = { directory: temporary, environment, nodeVersion: "v24.20.0" };
tool("cargo", 'console.error("cargo: command not found"); process.exit(127)');
for (const target of ["check-rust", "formal-rust", "smoke", "check", "integration-rust", "mutations-rust", "mutations", "explore"]) expect(() => checkPrerequisites(target, options), target).toThrow(/Cannot run cargo/);
diff --git a/typescript/test/formal-validation.test.ts b/typescript/test/formal-validation.test.ts
index 67ae1e77..dbaaf445 100644
--- a/typescript/test/formal-validation.test.ts
+++ b/typescript/test/formal-validation.test.ts
@@ -64,6 +64,7 @@ process.exit(Number(process.argv[3] ?? 0));\n`);
fakeTool("corepack", 'console.log("10.33.0")');
fakeTool("go", 'console.log("go version go1.27.1 test/test")');
fakeTool("cargo", 'console.log("cargo 1.98.1 (test 2026-08-05)")');
+ environment.PYTHON = fakeTool("python", 'if (process.argv.includes("--version")) console.log("Python 3.11.14")');
fakeTool("quint", 'console.log("0.32.0")');
fakeTool("java", 'console.log("openjdk 21.0.11")');
fakeTool("tar", 'console.log("bsdtar 3.5.3")');
@@ -164,10 +165,10 @@ process.exit(Number(process.argv[3] ?? 0));\n`);
expect(plan.filter(step => step.args?.[0] === "formal/witnesses.mjs")).toHaveLength(1);
expect(plan.some(step => step.args?.[0] === "formal/generate-artifacts.mjs")).toBe(false);
expect(plan[0]!.args).toEqual(["formal/run-models.mjs", "check"]);
- expect(plan.find(step => step.remove)!.remove).toEqual([".formal-traces/ts-completion.json", ".formal-traces/go-completion.json", ".formal-traces/rust-completion.json"]);
+ expect(plan.find(step => step.remove)!.remove).toEqual([".formal-traces/ts-completion.json", ".formal-traces/go-completion.json", ".formal-traces/rust-completion.json", ".formal-traces/python-completion.json"]);
// The aggregate is exactly these lanes in order, so a CI job running
// one lane executes the same steps as the local sequential run.
- expect(plan).toEqual(["formal-check", "formal-generate", "formal-ts", "formal-go", "formal-rust"].flatMap(target => validationPlan(target, { directory })));
+ expect(plan).toEqual(["formal-check", "formal-generate", "formal-ts", "formal-go", "formal-rust", "formal-python"].flatMap(target => validationPlan(target, { directory })));
});
it("keeps the model check as its own lane that produces nothing the port lanes consume", () => {
@@ -182,7 +183,7 @@ process.exit(Number(process.argv[3] ?? 0));\n`);
const generate = validationPlan("formal-generate", { directory });
expect(generate.some(step => step.args?.[0] === "formal/run-models.mjs" && step.args[1] === "check")).toBe(false);
expect(generate.some(step => step.args?.[0] === "formal/check-model-properties.mjs")).toBe(false);
- for (const target of ["formal-ts", "formal-go", "formal-rust", "mutations"]) {
+ for (const target of ["formal-ts", "formal-go", "formal-rust", "formal-python", "mutations"]) {
expect(validationPlan(target, { directory }).some(step => step.args?.[0] === "formal/run-models.mjs")).toBe(false);
}
// Every acceptance entry point keeps one complete campaign, after all
@@ -256,6 +257,46 @@ process.exit(Number(process.argv[3] ?? 0));\n`);
expect(smoke.env).toBeUndefined();
});
+ it("runs Python from its prepared interpreter and checks complete evidence after replay", () => {
+ const native = validationPlan("check-python", { directory, environment });
+ expect(native[0]!.remove).toEqual(["coverage/python/.coverage-native", "coverage/python/native.lcov"]);
+ expect(native.slice(1).map(step => step.command)).toEqual([environment.PYTHON, environment.PYTHON]);
+ expect(native[1]!.args).toEqual(["-m", "coverage", "run", "--rcfile=python/pyproject.toml",
+ "--data-file=coverage/python/.coverage-native", "-m", "pytest", "python/tests", "-m", "not integration"]);
+ expect(native[2]!.args).toEqual(["-m", "coverage", "lcov", "--rcfile=python/pyproject.toml",
+ "--data-file=coverage/python/.coverage-native", "-o", "coverage/python/native.lcov"]);
+ const plan = validationPlan("formal-python", { directory, environment });
+ expect(plan[0]!.remove).toEqual([".formal-traces/python-completion.json"]);
+ expect(plan[1]!.args).toEqual(["formal/conformance.mjs", "prepare", "python", ".formal-traces/python-context.json"]);
+ expect(plan[2]!.args).toEqual(["formal/run-python-replay.mjs", "--generated", "--scenarios", "--complete", "--report", ".formal-traces/python-replay.jsonl"]);
+ expect(plan[2]!.env).toEqual({ PYTHON: environment.PYTHON });
+ expect(plan.at(-1)!.args).toEqual(["formal/conformance.mjs", "check", ".formal-traces/python-completion.json", ".formal-traces/python-context.json"]);
+ expect(validationPlan("smoke", { directory, environment }).at(-1)!.args).toEqual(["-m", "pytest", "python/tests/test_conformance.py"]);
+ expect(validationPlan("integration-python", { directory, environment })[0]!.args).toEqual(["formal/run-python-integration.mjs"]);
+ });
+
+ it("prepends the selected Python sources for both native commands and prerequisite imports", () => {
+ environment.PYTHONPATH = ["/foreign/checkout/python", "/caller/dependencies"].join(delimiter);
+ const expected = [join(directory, "python"), environment.PYTHONPATH].join(delimiter);
+ for (const target of ["check-python", "smoke"]) {
+ const step = validationPlan(target, { directory, environment }).find(item => item.command === environment.PYTHON)!;
+ expect(step.env).toEqual({ NODE: process.execPath, PYTHONPATH: expected });
+ }
+ environment.PYTHON = fakeTool("python", `if (process.argv.includes("--version")) console.log("Python 3.14.7");
+else if (process.env.PYTHONPATH !== ${JSON.stringify(expected)}) throw new Error("wrong checkout import path");`);
+ expect(() => checkPrerequisites("check-python", { directory, environment, nodeVersion: "v24.20.0" })).not.toThrow();
+ });
+
+ it("requires the Python floor and dependencies only for the Python lanes", () => {
+ environment.PYTHON = fakeTool("python", 'console.log("Python 3.10.16")');
+ for (const target of ["check-python", "formal-python", "integration-python", "smoke", "check"]) {
+ expect(() => checkPrerequisites(target, { directory, environment, nodeVersion: "v24.20.0" }), target).toThrow(/Python 3.11 or later/);
+ }
+ expect(() => checkPrerequisites("formal-rust", { directory, environment, nodeVersion: "v24.20.0" })).not.toThrow();
+ environment.PYTHON = fakeTool("python", 'if (process.argv.includes("--version")) console.log("Python 3.14.7"); else { console.error("missing pytest"); process.exit(1); }');
+ expect(() => checkPrerequisites("check-python", { directory, environment, nodeVersion: "v24.20.0" })).toThrow(/missing pytest/);
+ });
+
it("lets Go parity and every mutation measurement run off the generated corpus without completion checks", () => {
const isCompletionCheck = (step: Step) => step.args?.[0] === "formal/conformance.mjs" && step.args[1] === "check";
const go = validationPlan("formal-go", { directory });
@@ -382,7 +423,7 @@ process.exit(Number(process.argv[3] ?? 0));\n`);
for (const target of ["check-ts", "formal", "formal-check", "formal-generate", "formal-ts", "explore"]) {
expect(() => checkPrerequisites(target, { directory, environment, nodeVersion: "v24.20.0" })).not.toThrow();
}
- });
+ }, 15_000);
it("requires Docker for mutation measurements but not report merging", () => {
fakeTool("docker", 'console.error("Docker not running"); process.exit(1)');
@@ -424,7 +465,7 @@ else {
describe("full formal workflow shape", () => {
type Step = { name?: string; run?: string; uses?: string; if?: string; env?: Record; with?: Record };
type Job = { needs?: string | string[]; if?: string; env?: Record; strategy?: { "fail-fast"?: boolean; matrix?: Record }; "timeout-minutes"?: number; steps: Step[] };
- const lanes = ["typescript-parity", "go-parity", "rust-parity", "typescript-mutations", "go-mutations", "rust-mutations"];
+ const lanes = ["typescript-parity", "go-parity", "rust-parity", "python-parity", "typescript-mutations", "go-mutations", "rust-mutations"];
const needsOf = (job: Job) => (job.needs === undefined ? [] : [job.needs].flat());
let jobs: Record;
@@ -536,6 +577,21 @@ describe("full formal workflow shape", () => {
expect(summary.path).toContain("formal-summary/rust/rust-replay-summary.json");
});
+ it("requires Python complete replay and retains its actual completion evidence", () => {
+ const parity = jobs["python-parity"]!;
+ expect(parity.steps.find(step => step.uses === "./.github/actions/setup-validation")!.with).toEqual({ python: "true" });
+ expect(parity.steps.map(step => step.run).filter(Boolean)).toEqual(["make formal-python"]);
+ const aggregate = jobs["formal-full"]!;
+ expect(needsOf(aggregate)).toContain("python-parity");
+ const gate = aggregate.steps.find(step => step.run?.includes("_RESULT"))!;
+ expect(gate.env).toMatchObject({ PYTHON_RESULT: "$" + "{{ needs.python-parity.result }}" });
+ expect(gate.run).toMatch(/test "\$PYTHON_RESULT" = success/);
+ const summary = aggregate.steps.find(step => step.uses?.startsWith("actions/upload-artifact"))!.with!;
+ expect(summary.path).toContain("formal-summary/python/python-completion.json");
+ expect(summary.path).toContain("formal-summary/python/python-context.json");
+ expect(summary.path).toContain("formal-summary/python/python-replay-summary.json");
+ });
+
it("requires the model check in the aggregate and retains its report in the long-lived summary", () => {
const aggregate = jobs["formal-full"]!;
expect(needsOf(aggregate)).toEqual(expect.arrayContaining(["check-models", "generate", "typescript-parity", "go-parity"]));