fix(review): close executable and fork identity gaps

Author: Coding-Dev-Tools

engraphis/core/ids.py

@@ -9,6 +9,7 @@
 """
 from __future__ import annotations
 
+import os
 import secrets
 import threading
 import time
@@ -20,6 +21,20 @@
 _ULID_LOCK = threading.Lock()
 _LAST_TIMESTAMP_MS = -1
 _LAST_RANDOM = -1
+_LAST_PID = os.getpid()
+
+
+def _reset_ulid_state_after_fork() -> None:
+    """Drop inherited sequence state and locks in a forked child."""
+    global _LAST_PID, _LAST_RANDOM, _LAST_TIMESTAMP_MS, _ULID_LOCK
+    _ULID_LOCK = threading.Lock()
+    _LAST_PID = os.getpid()
+    _LAST_TIMESTAMP_MS = -1
+    _LAST_RANDOM = -1
+
+
+if hasattr(os, "register_at_fork"):
+    os.register_at_fork(after_in_child=_reset_ulid_state_after_fork)
 
 
 # Canonical prefixes for each entity kind.
@@ -59,8 +74,13 @@ def ulid(timestamp_ms: Optional[int] = None) -> str:
         ts = timestamp_ms
     if not 0 <= ts < _MAX_TIMESTAMP_MS:
         raise ValueError("timestamp_ms must be an integer in range [0, 2**48)")
-    global _LAST_RANDOM, _LAST_TIMESTAMP_MS
+    global _LAST_PID, _LAST_RANDOM, _LAST_TIMESTAMP_MS
     with _ULID_LOCK:
+        pid = os.getpid()
+        if pid != _LAST_PID:
+            _LAST_PID = pid
+            _LAST_TIMESTAMP_MS = -1
+            _LAST_RANDOM = -1
         if ts == _LAST_TIMESTAMP_MS:
             if _LAST_RANDOM >= _MAX_RANDOM:
                 raise RuntimeError("ULID entropy exhausted within one millisecond")

scripts/check_release_readiness.py

@@ -12,6 +12,7 @@
 import math
 import os
 import re
+import stat
 import subprocess
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
@@ -36,8 +37,8 @@
     # ``sitecustomize.py`` executes before the candidate package and can alter
     # imports even when the tracked tree is clean.
     ".bat", ".cjs", ".cmd", ".css", ".dll", ".dylib", ".exe", ".html",
-    ".js", ".jsx", ".mjs", ".node", ".ps1", ".pyd", ".py", ".pyc", ".pyo",
-    ".pyw", ".sh", ".so", ".ts", ".tsx",
+    ".egg", ".js", ".jsx", ".mjs", ".node", ".pth", ".ps1", ".pyd", ".py",
+    ".pyc", ".pyo", ".pyw", ".sh", ".so", ".ts", ".tsx", ".whl", ".zip",
 })
 _IGNORED_RUNTIME_DIRS = frozenset({
     ".codex-pytest-tmp", ".hosted-eval-results", ".playwright", ".private-eval",
@@ -47,6 +48,24 @@
 })
 
 
+def _is_ignored_runtime_artifact(root: Path, relative: Path) -> bool:
+    """Return whether an ignored path can execute or affect imports."""
+    if relative.suffix.lower() in _EXECUTABLE_SUFFIXES:
+        return True
+    path = root / relative
+    try:
+        if path.is_symlink():
+            return True
+        mode = path.stat().st_mode
+        if mode & (stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH):
+            return True
+        with path.open("rb") as source:
+            return source.read(2) == b"#!"
+    except OSError:
+        # An unreadable ignored artifact cannot be proven harmless.
+        return True
+
+
 def canonical_bytes(value: Any) -> bytes:
     return json.dumps(value, sort_keys=True, separators=(",", ":"),
                       ensure_ascii=True, allow_nan=False).encode("utf-8")
@@ -107,12 +126,11 @@ def _ignored_executable_paths(root: Path) -> list[str]:
         if not entry:
             continue
         relative = Path(os.fsdecode(entry))
-        if relative.suffix.lower() not in _EXECUTABLE_SUFFIXES:
-            continue
         parts = {part.lower() for part in relative.parts}
         if parts & _IGNORED_RUNTIME_DIRS:
             continue
-        suspicious.append(relative.as_posix())
+        if _is_ignored_runtime_artifact(root, relative):
+            suspicious.append(relative.as_posix())
     return suspicious
 
 

tests/test_core_ids.py

@@ -26,6 +26,22 @@ def test_ulids_are_monotonic_within_one_timestamp():
     assert values == sorted(values)
 
 
+def test_ulid_sequence_resets_when_process_id_changes(monkeypatch):
+    random_values = iter((100, 200))
+    process_id = 1000
+    monkeypatch.setattr(ids.os, "getpid", lambda: process_id)
+    monkeypatch.setattr(ids.secrets, "randbits", lambda _: next(random_values))
+
+    first = ids.ulid(timestamp_ms=42_000)
+    second = ids.ulid(timestamp_ms=42_000)
+    process_id = 1001
+    third = ids.ulid(timestamp_ms=42_000)
+
+    assert first.endswith(ids._encode(100, 16))
+    assert second.endswith(ids._encode(101, 16))
+    assert third.endswith(ids._encode(200, 16))
+
+
 _CROCKFORD = {
     char: index for index, char in enumerate("0123456789ABCDEFGHJKMNPQRSTVWXYZ")
 }

tests/test_product_release_readiness.py

@@ -321,6 +321,27 @@ def test_unchecked_hash_bytecode_cannot_qualify_candidate(ledger, tmp_path):
     assert not result["valid"]
 
 
+def test_extensionless_ignored_script_cannot_qualify_candidate(ledger, tmp_path):
+    repository = tmp_path / "extensionless-script-repo"
+    repository.mkdir()
+    (repository / ".gitignore").write_text("release-helper\n", encoding="utf-8")
+    (repository / "engine.py").write_text("original", encoding="utf-8")
+    subprocess.run(["git", "init", "--quiet", str(repository)], check=True)
+    subprocess.run(["git", "add", ".gitignore", "engine.py"], cwd=repository, check=True)
+    subprocess.run(["git", "-c", "user.name=Fixture", "-c", "user.email=fixture@example.invalid",
+                    "commit", "--quiet", "-m", "fixture"], cwd=repository, check=True)
+    ledger["components"]["engine"]["commit"] = subprocess.check_output(
+        ["git", "rev-parse", "HEAD"], cwd=repository, text=True).strip()
+    ledger["candidate_id"] = candidate_id(ledger["components"])
+    for name in RELEASE_GATES:
+        pass_gate(ledger, tmp_path, name)
+    (repository / "release-helper").write_text("#!/bin/sh\necho ignored\n", encoding="utf-8")
+    result = validate(ledger, tmp_path, engine_root=repository)
+    assert not result["engine_checkout_verified"]
+    assert any("release-helper" in error for error in result["errors"])
+    assert not result["valid"]
+
+
 @pytest.mark.parametrize("flag", ["--assume-unchanged", "--skip-worktree"])
 def test_hidden_git_changes_cannot_qualify_candidate(ledger, tmp_path, flag):
     repository = tmp_path / "hidden-repo"