perf(effect-runtime): reuse a private source digest across processes

docs/architecture/rfcs/typescript-control-plane-migration-v0.md

@@ -301,10 +301,16 @@ Canonical summaries prepare recovery history only for role groups that contain
 a resume condition. Succession still uses complete active and archived lineage.
 Role summaries reuse request-local lineage facts and the succession owner's
 existing context fields. That owner validates and indexes the complete graph
-before materializing requested evaluations. Source fingerprinting still reads
-every admitted byte and rechecks freshness; sequential reads avoid starting a
-thread pool in each process. Healthy read commands use canonical registry/path
-owners and defer mutation error imports until needed.
+before materializing requested evaluations. Source fingerprinting hashes every
+admitted byte and rechecks freshness; sequential reads avoid a thread pool. On
+POSIX, a private per-user digest keyed by release, interpreter and complete
+source metadata lets later processes skip reopening unchanged sources. Only
+sources unchanged for two seconds are recorded; a key mismatch or untrusted
+record rereads every byte. A different file matching size, mtime and ctime,
+such as a retargeted symlink, or a filesystem that does not maintain ctime can
+hide a change; other hosts reread in every process.
+Healthy read commands use canonical registry/path owners and defer mutation
+error imports until needed.
 Exact detail queries push selection into the existing source projection after
 complete source validation. Queries that report unfiltered counts retain the
 complete first projection.

docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md

@@ -246,7 +246,10 @@ Canonical 摘要仅在某角色组包含恢复条件时准备恢复历史。
 继任计算仍使用完整的活跃与归档血缘。
 角色摘要复用一次请求内的血缘事实和继任 owner 的既有上下文字段。
 该 owner 先校验并索引完整图，再生成需要返回的评估。
-源码指纹仍读取每个已准入字节并复查新鲜度；顺序读取省去每个进程的线程池创建。
+源码指纹对每个已准入字节求哈希并复查新鲜度；顺序读取省去线程池创建。
+在 POSIX 上，按 release、解释器与完整源码元数据建键的私有用户级摘要，使后续进程无需重新打开未变化的源码。
+仅记录两秒内未变化的源码；键不匹配或记录不可信时重读每个字节。
+size、mtime 与 ctime 均相同的另一文件（例如重新指向的 symlink）或不维护 ctime 的文件系统可能掩盖修改；其他平台每个进程仍全量重读。
 正常读取直接使用注册表和路径的 canonical owner，发生错误时才导入相关错误类型。
 精确详情查询在完整来源校验后，将筛选下推到既有来源投影。
 需报告未筛选数量的查询仍保留完整的首次投影。

loopx/control_plane/effect_runtime.py

@@ -86,6 +86,11 @@
 CANONICAL_AUTHORITY_READ_TIMEOUT_SECONDS = 15.0
 _RUNTIME_SOURCE_SUFFIXES = frozenset({".json", ".ts"})
 _RuntimeSourceSnapshot = tuple[tuple[str, int, int, int], ...]
+_SOURCE_DIGEST_SCHEMA_VERSION = "loopx_effect_runtime_source_digest_v1"
+_SOURCE_DIGEST_MAX_BYTES = 4096
+# A coarse ctime clock can give a rewrite the timestamp it already had. Sources
+# changed this recently are hashed but not recorded for other processes.
+_SOURCE_DIGEST_SETTLE_NS = 2_000_000_000
 
 
 def source_number_tokens(value: object) -> list[list[Any]]:
@@ -347,20 +352,130 @@ def _runtime_source_files(root: Path | None = None) -> tuple[str, ...]:
     return tuple(relative for relative, *_metadata in _runtime_source_snapshot(root))
 
 
+def _source_digest_location(
+    root: str,
+    snapshot: _RuntimeSourceSnapshot,
+) -> tuple[Path, str] | None:
+    """Name one private digest per release and interpreter, keyed by metadata.
+
+    Only POSIX ctime cannot be set by a writer, so other hosts keep full reads.
+    """
+
+    if os.name != "posix":
+        return None
+    try:
+        directory = _runtime_dir()
+    except OSError:
+        return None
+    identity = hashlib.sha256()
+    for part in (_SOURCE_DIGEST_SCHEMA_VERSION, sys.executable, str(sys.version_info[:3])):
+        identity.update(part.encode("utf-8") + b"\0")
+    identity.update(os.fsencode(os.path.normcase(root)) + b"\0")
+    path = directory / f"source-digest-{identity.hexdigest()[:32]}.json"
+    identity.update(json.dumps(snapshot, separators=(",", ":")).encode("ascii"))
+    return path, identity.hexdigest()
+
+
+def _private_directory(path: Path) -> bool:
+    try:
+        metadata = os.lstat(path)
+    except OSError:
+        return False
+    return (stat.S_ISDIR(metadata.st_mode) and metadata.st_uid == os.getuid()
+            and not metadata.st_mode & 0o077)
+
+
+def _read_source_digest(path: Path, key: str) -> str | None:
+    if not _private_directory(path.parent):
+        return None
+    try:
+        descriptor = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
+    except OSError:
+        return None
+    try:
+        with os.fdopen(descriptor, "rb") as file:
+            metadata = os.fstat(file.fileno())
+            if (not stat.S_ISREG(metadata.st_mode) or metadata.st_uid != os.getuid()
+                    or metadata.st_mode & 0o077):
+                return None
+            payload = json.loads(file.read(_SOURCE_DIGEST_MAX_BYTES + 1))
+    except (OSError, ValueError, RecursionError):
+        return None
+    if (not isinstance(payload, dict)
+            or payload.get("schema_version") != _SOURCE_DIGEST_SCHEMA_VERSION
+            or payload.get("key") != key):
+        return None
+    fingerprint = payload.get("fingerprint")
+    if not isinstance(fingerprint, str) or BARE_SHA256_PATTERN.fullmatch(fingerprint) is None:
+        return None
+    return fingerprint
+
+
+def _write_source_digest(
+    path: Path,
+    key: str,
+    fingerprint: str,
+    snapshot: _RuntimeSourceSnapshot,
+) -> None:
+    """Best-effort publication into an existing private runtime namespace."""
+
+    settled_before = time.time_ns() - _SOURCE_DIGEST_SETTLE_NS
+    if any(max(mtime, ctime) > settled_before for _name, mtime, ctime, _size in snapshot):
+        return
+    if not _private_directory(path.parent):
+        return
+    data = json.dumps({
+        "schema_version": _SOURCE_DIGEST_SCHEMA_VERSION,
+        "key": key,
+        "fingerprint": fingerprint,
+    }, separators=(",", ":")).encode("ascii")
+    temporary = path.with_name(f".{path.name}.{secrets.token_hex(8)}.tmp")
+    try:
+        descriptor = os.open(
+            temporary, os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW, 0o600,
+        )
+    except OSError:
+        return
+    try:
+        with os.fdopen(descriptor, "wb") as file:
+            file.write(data)
+        os.replace(temporary, path)
+    except OSError:
+        try:
+            temporary.unlink()
+        except OSError:
+            pass
+
+
+def _require_unchanged_snapshot(root: Path, snapshot: _RuntimeSourceSnapshot) -> None:
+    current_snapshot = _runtime_source_snapshot(root)
+    if current_snapshot != snapshot:
+        raise _RuntimeSourceChanged(current_snapshot)
+
+
 @lru_cache(maxsize=8)
 def _runtime_fingerprint_for_snapshot(
     root: str,
     snapshot: _RuntimeSourceSnapshot,
 ) -> str:
+    # A fresh CLI process would otherwise open every source; on hosts that scan
+    # each open this dominates short commands. The record is only a memo.
+    source_root = Path(root)
+    location = _source_digest_location(root, snapshot)
+    if location is not None:
+        recorded = _read_source_digest(*location)
+        if recorded is not None:
+            # A hit adopts the record only under the full hash's freshness check.
+            _require_unchanged_snapshot(source_root, snapshot)
+            return recorded
     digest = hashlib.sha256()
     # The daemon's fixed Python adapter context must follow the interpreter
     # selected by this caller, even when two environments share source files.
     digest.update(sys.executable.encode("utf-8"))
     digest.update(str(sys.version_info[:3]).encode("ascii"))
-    source_root = Path(root)
     # Lazy imports and Python bridges resolve from this physical release. Equal
     # bytes in another checkout must not reuse a loader whose release may retire.
-    # Resolved aliases still share one owner; source bytes remain fully checked.
+    # Resolved aliases still share one owner; the digest covers every source byte.
     digest.update(b"loopx_effect_runtime_source_instance_v1\0")
     digest.update(os.fsencode(os.path.normcase(root)))
     digest.update(b"\0")
@@ -377,7 +492,10 @@ def _runtime_fingerprint_for_snapshot(
     current_snapshot = _runtime_source_snapshot(source_root)
     if current_snapshot != snapshot:
         raise _RuntimeSourceChanged(current_snapshot)
-    return digest.hexdigest()
+    fingerprint = digest.hexdigest()
+    if location is not None:
+        _write_source_digest(*location, fingerprint, snapshot)
+    return fingerprint
 
 
 def _runtime_fingerprint() -> str:

tests/control_plane/test_runtime_source_read_batching.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import hashlib
+import json
 import sys
 import os
 from pathlib import Path
@@ -80,3 +81,153 @@ def read(path):
 
     monkeypatch.setattr(Path, "read_bytes", read)
     assert runtime._runtime_fingerprint() == serial_fingerprint(tmp_path)
+
+
+posix_only = pytest.mark.skipif(os.name != "posix", reason="POSIX ctime keys the memo")
+
+
+def _private_namespace(tmp_path, monkeypatch, *, settle_ns=0):
+    namespace = tmp_path / "runtime"
+    namespace.mkdir(mode=0o700)
+    monkeypatch.setattr(runtime, "_runtime_dir", lambda: namespace)
+    monkeypatch.setattr(runtime, "_SOURCE_DIGEST_SETTLE_NS", settle_ns)
+    return namespace
+
+
+def _fresh_process_fingerprint():
+    runtime._runtime_fingerprint_for_snapshot.cache_clear()
+    return runtime._runtime_fingerprint()
+
+
+def _count_source_reads(monkeypatch):
+    original, reads = runtime.iter_binary_file_reads, []
+
+    def counted(paths, **kwargs):
+        for read in original(paths, **kwargs):
+            reads.append(read.path.name)
+            yield read
+
+    monkeypatch.setattr(runtime, "iter_binary_file_reads", counted)
+    return reads
+
+
+@posix_only
+def test_fresh_process_reuses_the_private_digest_until_sources_change(tmp_path, monkeypatch):
+    root = tmp_path / "release"
+    (root / "nested").mkdir(parents=True)
+    (root / "a.ts").write_bytes(b"export {};\n")
+    (root / "nested/b.json").write_bytes(b"{}")
+    monkeypatch.setattr(runtime, "_control_plane_root", lambda: root)
+    namespace = _private_namespace(tmp_path, monkeypatch)
+    reads = _count_source_reads(monkeypatch)
+
+    first = _fresh_process_fingerprint()
+    assert first == serial_fingerprint(root) and len(reads) == 2
+    [record] = namespace.iterdir()
+    # Locator retirement and fixture cleanup own only runtime-*.json names.
+    assert record.name.startswith("source-digest-") and not record.match("runtime-*.json")
+    assert record.stat().st_mode & 0o777 == 0o600
+    assert _fresh_process_fingerprint() == first and len(reads) == 2
+
+    (root / "a.ts").write_bytes(b"export const edited = true;\n")
+    edited = _fresh_process_fingerprint()
+    assert edited == serial_fingerprint(root) != first and len(reads) == 4
+    (root / "c.ts").write_bytes(b"export {};\n")
+    added = _fresh_process_fingerprint()
+    assert added == serial_fingerprint(root) != edited
+    (root / "nested/b.json").unlink()
+    removed = _fresh_process_fingerprint()
+    assert removed == serial_fingerprint(root) != added
+    assert [path.name for path in namespace.iterdir()] == [record.name]
+
+    copy = tmp_path / "copy"
+    copy.mkdir()
+    for name in ("a.ts", "c.ts"):
+        (copy / name).write_bytes((root / name).read_bytes())
+    monkeypatch.setattr(runtime, "_control_plane_root", lambda: copy)
+    assert _fresh_process_fingerprint() == serial_fingerprint(copy) != removed
+
+
+@posix_only
+def test_source_change_before_a_digest_hit_is_not_adopted(tmp_path, monkeypatch):
+    root = tmp_path / "release"
+    root.mkdir()
+    edited = root / "a.ts"
+    edited.write_bytes(b"export {};\n")
+    (root / "b.json").write_bytes(b"{}")
+    monkeypatch.setattr(runtime, "_control_plane_root", lambda: root)
+    namespace = _private_namespace(tmp_path, monkeypatch)
+    first = _fresh_process_fingerprint()
+    assert first == serial_fingerprint(root) and len(list(namespace.iterdir())) == 1
+    original, hits = runtime._read_source_digest, []
+
+    def edit_after_snapshot(path, key):
+        # The caller's snapshot is already taken; the stored key still matches it.
+        recorded = original(path, key)
+        if not hits:
+            edited.write_bytes(b"export const edited = true;\n")
+        hits.append(recorded)
+        return recorded
+
+    monkeypatch.setattr(runtime, "_read_source_digest", edit_after_snapshot)
+    reads = _count_source_reads(monkeypatch)
+
+    assert _fresh_process_fingerprint() == serial_fingerprint(root) != first
+    assert hits[0] == first and hits[1:] == [None] and reads == ["a.ts", "b.json"]
+
+
+@posix_only
+@pytest.mark.parametrize("defect", ["foreign_mode", "symlink", "shared_namespace",
+                                    "corrupt", "stale_key", "invalid_digest"])
+def test_untrusted_digest_record_is_recomputed(tmp_path, monkeypatch, defect):
+    root = tmp_path / "release"
+    root.mkdir()
+    (root / "a.ts").write_bytes(b"export {};\n")
+    monkeypatch.setattr(runtime, "_control_plane_root", lambda: root)
+    namespace = _private_namespace(tmp_path, monkeypatch)
+    path, key = runtime._source_digest_location(
+        os.fspath(root.resolve()), runtime._runtime_source_snapshot(root))
+    forged = {"schema_version": runtime._SOURCE_DIGEST_SCHEMA_VERSION,
+              "key": "0" * 64 if defect == "stale_key" else key,
+              "fingerprint": "F" * 64 if defect == "invalid_digest" else "0" * 64}
+    target = tmp_path / "elsewhere.json" if defect == "symlink" else path
+    target.write_text("{" if defect == "corrupt" else json.dumps(forged))
+    target.chmod(0o644 if defect == "foreign_mode" else 0o600)
+    if defect == "symlink":
+        path.symlink_to(target)
+    if defect == "shared_namespace":
+        namespace.chmod(0o755)
+    reads = _count_source_reads(monkeypatch)
+
+    assert _fresh_process_fingerprint() == serial_fingerprint(root)
+    assert reads == ["a.ts"]
+
+
+@posix_only
+def test_unsettled_or_failed_hash_records_no_digest(tmp_path, monkeypatch):
+    root = tmp_path / "release"
+    root.mkdir()
+    for name in ("a.ts", "b.ts"):
+        (root / name).write_bytes(b"export {};\n")
+    monkeypatch.setattr(runtime, "_control_plane_root", lambda: root)
+    namespace = _private_namespace(tmp_path, monkeypatch, settle_ns=3600 * 10**9)
+    assert _fresh_process_fingerprint() == serial_fingerprint(root)
+    assert list(namespace.iterdir()) == []
+
+    monkeypatch.setattr(runtime, "_SOURCE_DIGEST_SETTLE_NS", 0)
+    original = Path.read_bytes
+
+    def read(path):
+        if path.name == "b.ts":
+            raise PermissionError("denied")
+        return original(path)
+
+    monkeypatch.setattr(Path, "read_bytes", read)
+    with pytest.raises(PermissionError):
+        _fresh_process_fingerprint()
+    assert list(namespace.iterdir()) == []
+
+    namespace.rmdir()
+    monkeypatch.setattr(Path, "read_bytes", original)
+    assert _fresh_process_fingerprint() == serial_fingerprint(root)
+    assert not namespace.exists(), "the memo never creates the runtime namespace"