A full session transcript, a week late, with nothing taken out.

No summary, no cleanup. Tool calls included.

This one is a code review. The setup matters. The code was written by one AI agent, Codex. The review was done by a different one, from a different vendor, with no tools and no ability to read anything beyond the diffs pasted into the prompt.

The rule in my setup is simple. Nobody reviews their own work. That includes agents reviewing their own work.

The change under review is a distributed lock. Two agents share one directory. The lease stops them writing the same file at once. One owner wins. The loser gets BUSY and the name of the holder. This version adds fencing epochs, so a stale process cannot release a lease that has since been handed to someone else.

The verdict was REQUEST_CHANGES. It rests on one finding, and I think it is a genuinely good catch.

The lock had a fallback for old database rows. If the exact session and epoch pair misses, delete where the epoch is empty. The docstring calls this a one time schema handoff. Nothing in the code scopes it to migration time.

So it runs on every release call. Forever. A permanent wildcard match. Any caller legitimately holding a lease under the old empty epoch, and the code's own comment names one, can have that lease silently released by an unrelated call that happens to share the session string.

The part I find most useful is what the reviewer said about the tests. The suite does prove the intended direction. It also proves that an untokened release cannot touch a tokened lease. It never tests the reverse: a tokened release wiping a live lease belonging to a different logical owner. A test suite that covers one direction of an asymmetric fallback reads as coverage and is not.

Two more findings are here. Both flagged low, not inflated. A contract test that will break if the hook config is reorganised into an equally valid shape. And a skill duplication guard that quietly vanished in a file rename, with no replacement.

One note on language. My working sessions are usually in Russian, with English code and tooling. This one is almost entirely English, since the review prompt and the verdict contract are English.

The whole corpus of these transcripts lives at github.com/tonydzi/agent-sessions, which is where this one is canonical.

The transcript

Антон:

You are an INDEPENDENT senior code reviewer from a DIFFERENT vendor than the author. The author (Codex) wrote the two-repository change below. Review ONLY the supplied diffs; do not use tools or read other files. Be concrete and skeptical.

Task: local Windows Claude/Codex shared-file write protection. Required properties: two concurrent writers of one absolute file produce one owner and BUSY rc=7 naming the holder; foreign release and delayed old-turn release are harmless; old-schema migration is race-safe and token-aware Stop can release a migrated legacy row; real Codex shell/apply_patch hook wiring is tested; counters are attributable; child-script, relative-path, Windows-only, per-checkout and local-only boundaries are honest.

Check correctness, security bypasses, caller breakage, races, and whether tests really kill their named mutations. Prioritize real MED/HIGH defects; do not block on style.

CRITICAL OUTPUT CONTRACT: the VERY LAST line of your reply must be exactly
===VERDICT===
where is one bare word: APPROVE (no blocking issues) or REQUEST_CHANGES (one or more MED/HIGH issues). No backticks, no bold, no brackets, nothing else on that line. State it exactly once, on that final line only.

Format before the final line:
SUMMARY:
FINDINGS:

  • file:line | severity | issue | fix (or none)

--- SCRIPTS DIFF START ---diff --git a/_test_codex_shared_workspace_safety.py b/_test_codex_shared_workspace_safety.py
deleted file mode 100644
index 4d7d597e5..000000000
--- a/_test_codex_shared_workspace_safety.py
+++ /dev/null
@@ -1,273 +0,0 @@
-# -- coding: utf-8 --

-"""Integration acceptance for Claude/Codex shared-workspace safety.

-Purpose: prove the active Codex registry has one local shelf, both harness configs
-route writes through the same guard, canonical apply_patch is really parsed, and a
-Stop event releases the turn lease. Input: live configs plus an isolated temp DB.
-Output: PASS/FAIL and evidence; never edits a real workspace file.
-Caller: /tt after shared-workspace changes. Rail: local Python + codex debug, 0 LLM.
-updated: 2026-09-11
-"""

-from future import annotations

-import json
-import os
-import re
-import shutil
-import subprocess
-import sys
-import tempfile

-import time

-HOME = os.path.expanduser("~")
-SCRIPTS = os.path.join(HOME, ".claude", "scripts")
-HOOKS = os.path.join(HOME, ".claude", "hooks")
-GUARD = os.path.join(HOOKS, "workspace_write_guard.py")
-CONSTITUTION = os.path.join(HOOKS, "constitution_guard.py")
-ORPHAN = os.path.join(HOOKS, "orphan_check_hook.py")
-TURNSTATE = os.path.join(HOOKS, "turnstate_hook.py")
-CODEX_HOOKS = os.path.join(HOME, ".codex", "hooks.json")
-CLAUDE_SETTINGS = os.path.join(HOME, ".claude", "settings.json")
-R1 = os.path.join(HOME, ".agents", "skills")

-RESULTS = []

-
-def check(name, ok, detail=""):

  • RESULTS.append((name, bool(ok), detail))
  • print(" %s %s%s" % ("OK " if ok else "FAIL", name,

- (" -- " + str(detail)) if detail else ""))

-
-def _json(path):

  • with open(path, encoding="utf-8") as fh:

- return json.load(fh)

-
-def _commands(entries):

  • out = []
  • for group in entries or []:
  • for hook in group.get("hooks") or []:
  • out.append((group.get("matcher") or "", hook.get("command") or ""))

- return out

-
-def _run_guard(payload, phase="pre", env=None):

  • return subprocess.run([sys.executable, GUARD, "--phase", phase],
  • input=json.dumps(payload), capture_output=True, text=True,

- encoding="utf-8", errors="replace", env=env, timeout=30)

-
-def _run_constitution(payload, env=None):

  • return subprocess.run([sys.executable, CONSTITUTION], input=json.dumps(payload),
  • capture_output=True, text=True, encoding="utf-8",

- errors="replace", env=env, timeout=30)

-
-def _wait_until(ts):

  • while True:
  • left = float(ts) - time.time()
  • if left <= 0:
  • return

- time.sleep(min(0.02, left / 2))

-
-def _guard_child(root, start, session, target):

  • env = dict(os.environ)
  • env["WORKSPACE_WRITE_LEASE_DB"] = os.path.join(root, "race.sqlite3")
  • env["WORKSPACE_WRITE_LEASE_COUNTER"] = os.path.join(root, "race-counter.jsonl")
  • _wait_until(start)
  • payload = {"session_id": session, "cwd": root, "tool_name": "Write",
  • "tool_input": {"file_path": target}}
  • run = _run_guard(payload, env=env)
  • print("WIN" if run.returncode == 0 else "LOSE")

- return 0

-
-def _guard_race(root, target, racers=8):

  • start = time.time() + 1.2
  • procs = [subprocess.Popen(
  • [sys.executable, os.path.abspath(file), "--guard-child", root,
  • "%.6f" % start, "hook-session-%d" % i, target],
  • stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  • for i in range(racers)]
  • wins, errors = 0, []
  • for proc in procs:
  • out, err = proc.communicate(timeout=60)
  • answer = out.decode("utf-8", "replace").strip()
  • if answer.endswith("WIN"):
  • wins += 1
  • elif not answer.endswith("LOSE"):
  • errors.append("rc=%s out=%r err=%r" %
  • (proc.returncode, answer, err.decode("utf-8", "replace")[:120]))

- return wins, errors

-
-def main():

  • root = tempfile.mkdtemp(prefix="codex-shared-safety-test-")
  • env = dict(os.environ)
  • env["WORKSPACE_WRITE_LEASE_DB"] = os.path.join(root, "leases.sqlite3")
  • env["WORKSPACE_WRITE_LEASE_COUNTER"] = os.path.join(root, "counter.jsonl")
  • target = os.path.join(root, "shared file.md")
  • try:
  • r1_skills = []
  • if os.path.isdir(R1):
  • r1_skills = [n for n in os.listdir(R1)
  • if os.path.isfile(os.path.join(R1, n, "SKILL.md"))]
  • check("r1/.agents не содержит активных копий SKILL.md", not r1_skills,

- "active=%d first=%s" % (len(r1_skills), r1_skills[:5]))

  • check("общий workspace guard существует", os.path.isfile(GUARD), GUARD)
  • codex = _json(CODEX_HOOKS)
  • claude = _json(CLAUDE_SETTINGS)
  • cpre = _commands(codex.get("hooks", {}).get("PreToolUse"))
  • clpre = _commands(claude.get("hooks", {}).get("PreToolUse"))
  • check("Codex PreToolUse явно матчится на canonical apply_patch",
  • any("constitution_guard" in cmd and "apply_patch" in matcher
  • for matcher, cmd in cpre), cpre[:3])
  • check("Codex shell тоже идёт через guard",
  • any("constitution_guard" in cmd and
  • ("Bash" in matcher or "PowerShell" in matcher) for matcher, cmd in cpre), cpre[:3])
  • check("Claude file tools идут через тот же guard",
  • any("constitution_guard" in cmd and "Write" in matcher for matcher, cmd in clpre))
  • check("Claude shell идёт через тот же guard",
  • any("constitution_guard" in cmd and
  • ("Bash" in matcher or "PowerShell" in matcher) for matcher, cmd in clpre))
  • cpost = _commands(codex.get("hooks", {}).get("PostToolUse"))
  • check("Codex PostToolUse явно матчится на canonical apply_patch",
  • any("orphan_check_hook" in cmd and "apply_patch" in matcher

- for matcher, cmd in cpost), cpost[:3])

  • if os.path.isfile(GUARD):
  • race_root = os.path.join(root, "parallel-hook")
  • os.makedirs(race_root, exist_ok=True)
  • race_target = os.path.join(race_root, "one-file.md")
  • wins, errors = _guard_race(race_root, race_target)
  • check("параллельный hook: 8 turn -> ровно один проходит", wins == 1,
  • "wins=%d errors=%s" % (wins, errors))
  • patch = "*** Begin Patch\n*** Update File: %s\n@@\n-old\n+new\n*** End Patch" % target
  • p1 = {"session_id": "codex-A", "cwd": root, "tool_name": "apply_patch",
  • "tool_input": {"command": patch}}
  • p2 = {"session_id": "claude-B", "cwd": root, "tool_name": "Edit",
  • "tool_input": {"file_path": target}}
  • a = _run_guard(p1, env=env)
  • b = _run_guard(p2, env=env)
  • check("canonical apply_patch реально получает lease", a.returncode == 0,
  • "rc=%s stderr=%s" % (a.returncode, a.stderr[:200]))
  • check("второй агент блокируется на том же файле", b.returncode == 2 and
  • "codex-A" in b.stderr, "rc=%s stderr=%s" % (b.returncode, b.stderr[:300]))
  • codex_b = dict(p2)
  • codex_b.update({"tool_name": "apply_patch", "turn_id": "turn-B",
  • "tool_use_id": "call-B",
  • "tool_input": {"command": patch}})
  • structured = _run_constitution(codex_b, env)
  • try:
  • structured_json = json.loads(structured.stdout)
  • permission = structured_json["hookSpecificOutput"]["permissionDecision"]
  • except Exception:
  • permission = ""
  • check("Codex 0.147 получает structured deny, не hook Failed",
  • structured.returncode == 0 and permission == "deny" and
  • "codex-A" in structured.stdout,
  • "rc=%s out=%s err=%s" % (structured.returncode,
  • structured.stdout[:300], structured.stderr[:120]))
  • stop_payload = {"session_id": "codex-A", "cwd": root,
  • "hook_event_name": "Stop"}
  • stop = subprocess.run([sys.executable, TURNSTATE], input=json.dumps(stop_payload),
  • capture_output=True, text=True, encoding="utf-8",
  • errors="replace", env=env, timeout=30)
  • after = _run_guard(p2, env=env)
  • check("реальный turnstate Stop освобождает turn-lease",
  • stop.returncode == 0 and after.returncode == 0,
  • "stop=%s after=%s" % (stop.returncode, after.returncode))
  • _run_guard({"session_id": "claude-B", "cwd": root}, "stop", env)
  • post_payload = dict(p1)
  • post_payload.update({"turn_id": "turn-post", "tool_use_id": "call-post",
  • "hook_event_name": "PostToolUse", "tool_response": {"ok": True}})
  • post = subprocess.run([sys.executable, ORPHAN], input=json.dumps(post_payload),
  • capture_output=True, text=True, encoding="utf-8",
  • errors="replace", env=env, timeout=30)
  • check("Codex PostToolUse no-op = пустой stdout, не invalid approve",
  • post.returncode == 0 and not post.stdout.strip(),
  • "rc=%s out=%r err=%r" % (post.returncode, post.stdout, post.stderr[:120]))
  • malformed = subprocess.run([sys.executable, GUARD, "--phase", "pre"],
  • input="not-json", capture_output=True, text=True,
  • env=env, timeout=30)
  • check("битый write-payload блокируется fail-closed", malformed.returncode == 2)
  • readonly = _run_guard({"session_id": "reader", "cwd": root,
  • "tool_name": "Bash",
  • "tool_input": {"command": "git status --short"}}, env=env)
  • check("read-only shell не берёт лишний замок", readonly.returncode == 0)
  • if HOOKS not in sys.path:
  • sys.path.insert(0, HOOKS)
  • import workspace_write_guard as guard_module
  • check("read-only whitelist не пропускает PowerShell script-block",
  • not guard_module.shell_is_read_only(
  • "Get-Content x | Where-Object { Remove-Item x; $true }"))
  • check("read-only whitelist не пропускает git --output",
  • not guard_module.shell_is_read_only("git diff --output=stolen.patch"))
  • # 14.09.2026, класс lease-storm. ЗДЕСЬ БЫЛ ОБРАТНЫЙ ИНВАРИАНТ: python mystery.py
  • # лизовал cwd ЦЕЛИКОМ, и тест требовал, чтобы сосед после этого не мог править
  • # СВОЙ файл в той же папке. Это и есть шторм, записанный как «защита»: замер 14.09
  • # -- 198 блокировок, 69% из них запрос ПАПКИ, 36 пострадавших сессий, среди жертв
  • # чтение файла. Защита была ещё и мнимой: путь цели в такой команде не назван, так
  • # что от РЕАЛЬНОЙ гонки за файл папочная лиза не спасала -- она лишь запирала
  • # непричастных. Новый инвариант: неизвестная команда без признака записи не лизует
  • # ничего, а гонку за конкретный файл по-прежнему ловит файловый инструмент.
  • unknown_writer = _run_guard({"session_id": "script-A", "cwd": root,
  • "tool_name": "Bash",
  • "tool_input": {"command": "python mystery.py"}}, env=env)
  • neighbour = _run_guard({"session_id": "script-B", "cwd": root,
  • "tool_name": "Edit",
  • "tool_input": {"file_path": target}}, env=env)
  • check("неизвестный shell-процесс НЕ запирает соседей по папке",
  • unknown_writer.returncode == 0 and neighbour.returncode == 0,
  • "writer=%s neighbour=%s" % (unknown_writer.returncode, neighbour.returncode))
  • same_file = _run_guard({"session_id": "script-C", "cwd": root,
  • "tool_name": "Edit",
  • "tool_input": {"file_path": target}}, env=env)
  • check("гонка за ОДИН файл по-прежнему отбивается",
  • same_file.returncode == 2, "same_file=%s" % same_file.returncode)
  • _run_guard({"session_id": "script-B", "cwd": root}, "stop", env)
  • _run_guard({"session_id": "script-A", "cwd": root}, "stop", env)
  • env_sid = dict(env)
  • env_sid["CODEX_THREAD_ID"] = "env-only-session"
  • env_target = os.path.join(root, "env-session.md")
  • env_acquire = _run_guard({"cwd": root, "tool_name": "Edit",
  • "tool_input": {"file_path": env_target}}, env=env_sid)
  • env_stop = subprocess.run([sys.executable, TURNSTATE],
  • input=json.dumps({"cwd": root, "hook_event_name": "Stop"}),
  • capture_output=True, text=True, encoding="utf-8",
  • errors="replace", env=env_sid, timeout=30)
  • env_after = _run_guard({"session_id": "after-env", "cwd": root,
  • "tool_name": "Edit",
  • "tool_input": {"file_path": env_target}}, env=env)
  • check("Stop использует тот же env fallback session-id",
  • env_acquire.returncode == 0 and env_stop.returncode == 0 and
  • env_after.returncode == 0,
  • "acquire=%s stop=%s after=%s" % (env_acquire.returncode,

- env_stop.returncode, env_after.returncode))

  • help_run = subprocess.run([sys.executable, GUARD, "--help"],
  • capture_output=True, text=True, timeout=30)
  • bad_flag = subprocess.run([sys.executable, GUARD, "--wat"],
  • capture_output=True, text=True, timeout=30)
  • check("guard --help", help_run.returncode == 0 and "usage:" in help_run.stdout)
  • check("guard отвергает неизвестный флаг", bad_flag.returncode == 2)
  • finally:

- shutil.rmtree(root, ignore_errors=True)

  • bad = [name for name, ok, _ in RESULTS if not ok]
  • print("\nВЕРДИКТ: %s" % ("PASS" if not bad else "FAIL: " + "; ".join(bad)))

- return 0 if not bad else 1

-
-if name == "main":

  • try:
  • sys.stdout.reconfigure(encoding="utf-8")
  • sys.stderr.reconfigure(encoding="utf-8")
  • except Exception:
  • pass
  • if len(sys.argv) > 1 and sys.argv[1] == "--guard-child":
  • sys.exit(_guard_child(sys.argv[2], sys.argv[3], sys.argv[4], sys.argv[5]))
  • sys.path.insert(0, SCRIPTS)
  • sys.exit(main()) diff --git a/_test_workspace_write_guard.py b/_test_workspace_write_guard.py new file mode 100644 index 000000000..e2c264e78 --- /dev/null +++ b/_test_workspace_write_guard.py @@ -0,0 +1,830 @@ +# -- coding: utf-8 -- +"""Integration acceptance for Claude/Codex shared-workspace safety. + +Purpose: prove both harness configs route canonical Bash/exec_command/shell and +apply_patch through one guard, Codex turn_id becomes a fencing epoch, and a delayed +old Stop cannot release the next turn. Input: live configs plus an isolated temp DB. +Output: PASS/FAIL and evidence; never edits a real workspace file. A separate fresh +Codex-session canary is still required to prove the harness actually invoked hooks. +Caller: nightly regress grid and /tt after shared-workspace changes. Rail: local +Python stdlib, 0 LLM/network. + +KILL-LIST (source/config mutation -> case that must fail): +- remove Bash/PowerShell/exec_command/shell from Codex PreToolUse matcher ->
  • mutation: удаление shell matcher обязано стать красным; +- disable/remove the persisted Codex PreToolUse trust state ->
  • Codex PreToolUse trust state включён; +- replace the configured hook by a missing command, dropped stdin, or echo ->
  • live hooks.json command исполняется и три command mutants красные; +- release by session while ignoring turn_id ->
  • поздний Stop старого turn_id не снимает новую эпоху; +- keep agent/<id> distinct from <id> ->
  • Stop alias agent/session снимает lease plain session; +- remove/misroute Codex Stop -> turnstate_hook registration ->
  • Codex Stop зарегистрирован на fenced release; +- return allow instead of Codex structured deny ->
  • Codex 0.147 получает structured deny, не hook Failed; +- let a read-only-looking PowerShell script block mutate -> the two read-only
  • whitelist checks fail. The fresh-session canary covers the runtime boundary that
  • this direct-hook suite cannot prove. +- scan quoted <input> or discard 2>$null as output redirection ->
  • literal rg repro не лизует входные absolute roots. +- hide a quoted child-shell body from mutation detection ->
  • quoted PowerShell/cmd writers блокируются holder. +- ignore a quoted executable path (including PowerShell call-operator) ->
  • quoted executable path child writers блокируются holder. +- treat PowerShell backslash as a quote escape, or stop at an option value before
  • -Command -> PowerShell option/backslash/call variants блокируются holder. +- lease the quoted shell executable itself ->
  • один pwsh executable не сериализует разные output files. +- ignore bash -c text inside an rg pattern, or miss bash/sh combined c flags ->
  • quoted bash -lc/sh -ec writers видны, rg pattern остаётся reader. +updated: 2026-09-21 +""" +from future import annotations + +import json +import os +import re +import shutil +import sqlite3 +import subprocess +import sys +import tempfile +import time + +HOME = os.path.expanduser("~") +SCRIPTS = os.environ.get("WORKSPACE_LEASE_TEST_SCRIPTS") or os.path.join(
  • HOME, ".claude", "scripts") +HOOKS = os.environ.get("WORKSPACE_LEASE_TEST_HOOKS") or os.path.join(
  • HOME, ".claude", "hooks") +GUARD = os.path.join(HOOKS, "workspace_write_guard.py") +CONSTITUTION = os.path.join(HOOKS, "constitution_guard.py") +ORPHAN = os.path.join(HOOKS, "orphan_check_hook.py") +TURNSTATE = os.path.join(HOOKS, "turnstate_hook.py") +CODEX_HOOKS = os.path.join(HOME, ".codex", "hooks.json") +CODEX_CONFIG = os.path.join(HOME, ".codex", "config.toml") +CLAUDE_SETTINGS = os.path.join(HOME, ".claude", "settings.json") +RESULTS = [] + + +def check(name, ok, detail=""):
  • RESULTS.append((name, bool(ok), detail))
  • print(" %s %s%s" % ("OK " if ok else "FAIL", name,
  • (" -- " + str(detail)) if detail else "")) + + +def _json(path):
  • with open(path, encoding="utf-8") as fh:
  • return json.load(fh) + + +def _commands(entries):
  • out = []
  • for group in entries or []:
  • for hook in group.get("hooks") or []:
  • out.append((group.get("matcher") or "", hook.get("command") or ""))
  • return out + + +def _codex_pretool_contract(commands):
  • required = {"apply_patch", "Bash", "PowerShell", "exec_command", "shell"}
  • for matcher, command in commands:
  • tokens = {part.strip() for part in matcher.split("|") if part.strip()}
  • if "constitution_guard" in command and required.issubset(tokens):
  • return True
  • return False + + +def _codex_pretool_trust_state(config_text):
  • """Read the exact persisted state; runtime canary still proves hash freshness."""
  • marker = "[hooks.state.'%s:pre_tool_use:0:0']" % CODEX_HOOKS
  • start = config_text.find(marker)
  • if start < 0:
  • return False
  • tail = config_text[start + len(marker):]
  • end = tail.find("\n[")
  • section = tail if end < 0 else tail[:end]
  • return (bool(re.search(r'(?m)^trusted_hash\s*=\s*"sha256:[0-9a-f]{64}"\s*$', section)) and
  • bool(re.search(r"(?m)^enabled\s*=\s*true\s*$", section))) + + +def _file_size(path):
  • return os.path.getsize(path) if os.path.isfile(path) else 0 + + +def _new_matching_lines(path, offset, markers):
  • """Allow concurrent legitimate writers; return only rows attributable here."""
  • if not os.path.isfile(path):
  • return []
  • if os.path.getsize(path) < offset:
  • return ["production audit log shrank/rotated during the test"]
  • with open(path, "rb") as fh:
  • fh.seek(offset)
  • suffix = fh.read().decode("utf-8", "replace")
  • lowered = [marker.lower() for marker in markers]
  • return [line for line in suffix.splitlines()
  • if any(marker in line.lower() for marker in lowered)] + + +def _lease_owner(db, target):
  • resource = os.path.normcase(os.path.realpath(os.path.abspath(target)))
  • con = sqlite3.connect(db)
  • try:
  • row = con.execute(
  • "SELECT session,epoch FROM leases WHERE resource=?", (resource,)).fetchone()
  • return {"session": row[0], "epoch": row[1]} if row else None
  • finally:
  • con.close() + + +def _run_guard(payload, phase="pre", env=None):
  • return subprocess.run([sys.executable, GUARD, "--phase", phase],
  • input=json.dumps(payload), capture_output=True, text=True,
  • encoding="utf-8", errors="replace", env=env, timeout=30) + + +def _run_constitution(payload, env=None):
  • return subprocess.run([sys.executable, CONSTITUTION], input=json.dumps(payload),
  • capture_output=True, text=True, encoding="utf-8",
  • errors="replace", env=env, timeout=30) + + +def _run_hook_command(command, payload, env=None):
  • """Execute the exact hooks.json command, including its real .cmd shim."""
  • if os.name == "nt":
  • argv = [os.environ.get("COMSPEC") or "cmd.exe", "/d", "/s", "/c", command]
  • else:
  • argv = ["sh", "-c", command]
  • return subprocess.run(argv, input=json.dumps(payload), capture_output=True,
  • text=True, encoding="utf-8", errors="replace",
  • env=env, timeout=30) + + +def _is_structured_deny(run, holder=""):
  • try:
  • parsed = json.loads(run.stdout)
  • output = parsed["hookSpecificOutput"]
  • reason = str(output.get("permissionDecisionReason") or "")
  • return (run.returncode == 0 and output.get("permissionDecision") == "deny" and
  • (not holder or holder in reason))
  • except Exception:
  • return False + + +def _wait_until(ts):
  • while True:
  • left = float(ts) - time.time()
  • if left <= 0:
  • return
  • time.sleep(min(0.02, left / 2)) + + +def _guard_child(root, start, session, target):
  • env = dict(os.environ)
  • env["WORKSPACE_WRITE_LEASE_DB"] = os.path.join(root, "race.sqlite3")
  • env["WORKSPACE_WRITE_LEASE_COUNTER"] = os.path.join(root, "race-counter.jsonl")
  • env["WORKSPACE_WRITE_LEASE_MODULE_DIR"] = SCRIPTS
  • _wait_until(start)
  • payload = {"session_id": session, "cwd": root, "tool_name": "Write",
  • "tool_input": {"file_path": target}}
  • run = _run_guard(payload, env=env)
  • if run.returncode == 0:
  • print("WIN %s" % session)
  • return 0
  • if (run.returncode == 2 and "BUSY:" in run.stderr and
  • "held by hook-session-" in run.stderr):
  • match = re.search(r"held by ([^.\r\n]+)", run.stderr)
  • if match:
  • print("LOSE BUSY held by %s" % match.group(1).strip())
  • return 0
  • print("ERROR BUSY without parseable holder: %r" % run.stderr[:500])
  • return 3
  • print("ERROR rc=%s out=%r err=%r" %
  • (run.returncode, run.stdout[:160], run.stderr[:500]))
  • return 3 + + +def _guard_race(root, target, racers=8):
  • start = time.time() + 1.2
  • procs = [subprocess.Popen(
  • [sys.executable, os.path.abspath(file), "--guard-child", root,
  • "%.6f" % start, "hook-session-%d" % i, target],
  • stdout=subprocess.PIPE, stderr=subprocess.PIPE)
  • for i in range(racers)]
  • winners, loser_holders, errors = [], [], []
  • for proc in procs:
  • out, err = proc.communicate(timeout=60)
  • answer = out.decode("utf-8", "replace").strip()
  • if answer.startswith("WIN "):
  • winners.append(answer[4:].strip())
  • elif answer.startswith("LOSE BUSY held by "):
  • loser_holders.append(answer[len("LOSE BUSY held by "):].strip())
  • else:
  • errors.append("rc=%s out=%r err=%r" %
  • (proc.returncode, answer, err.decode("utf-8", "replace")[:120]))
  • if len(winners) == 1:
  • wrong = [holder for holder in loser_holders if holder != winners[0]]
  • if len(loser_holders) != racers - 1 or wrong:
  • errors.append("winner=%r loser_holders=%r" % (winners, loser_holders))
  • return len(winners), errors + + +def main():
  • production_audit = os.path.join(HOME, ".claude", "hooks", "_constitution_guard.log")
  • production_audit_before = _file_size(production_audit)
  • production_usage = os.path.join(
  • os.environ.get("LOCALAPPDATA") or os.path.join(HOME, ".local"),
  • "AntonAgents", "workspace-write-leases", "usage.jsonl")
  • production_usage_before = _file_size(production_usage)
  • root = tempfile.mkdtemp(prefix="codex-shared-safety-test-")
  • env = dict(os.environ)
  • env["WORKSPACE_WRITE_LEASE_DB"] = os.path.join(root, "leases.sqlite3")
  • env["WORKSPACE_WRITE_LEASE_COUNTER"] = os.path.join(root, "counter.jsonl")
  • env["WORKSPACE_WRITE_LEASE_MODULE_DIR"] = SCRIPTS
  • env["USERPROFILE"] = root
  • env["HOME"] = root
  • isolated_audit_dir = os.path.join(root, ".claude", "hooks")
  • os.makedirs(isolated_audit_dir, exist_ok=True)
  • isolated_audit = os.path.join(isolated_audit_dir, "_constitution_guard.log")
  • env["CONSTITUTION_GUARD_LOG"] = isolated_audit
  • target = os.path.join(root, "shared file.md")
  • try:
  • check("общий workspace guard существует", os.path.isfile(GUARD), GUARD)
  • codex = _json(CODEX_HOOKS)
  • claude = _json(CLAUDE_SETTINGS)
  • with open(CODEX_CONFIG, encoding="utf-8") as fh:
  • codex_config = fh.read()
  • cpre = _commands(codex.get("hooks", {}).get("PreToolUse"))
  • clpre = _commands(claude.get("hooks", {}).get("PreToolUse"))
  • configured_pretool = next((command for matcher, command in cpre
  • if "constitution_guard" in command and
  • "apply_patch" in matcher), "")
  • check("Codex PreToolUse покрывает apply_patch и shell aliases",
  • _codex_pretool_contract(cpre), cpre[:3])
  • matcher_mutation_survivors = []
  • for removed in ("Bash", "PowerShell", "exec_command", "shell"):
  • mutated_cpre = []
  • for matcher, command in cpre:
  • kept = [part for part in matcher.split("|") if part != removed]
  • mutated_cpre.append(("|".join(kept), command))
  • if _codex_pretool_contract(mutated_cpre):
  • matcher_mutation_survivors.append(removed)
  • check("mutation: каждый shell matcher по одному обязан стать красным",
  • not matcher_mutation_survivors,
  • "survivors=%r" % matcher_mutation_survivors)
  • trust_state = _codex_pretool_trust_state(codex_config)
  • check("persisted PreToolUse state имеет enabled+hash; freshness проверяет canary",
  • trust_state, CODEX_CONFIG)
  • mutated_trust = codex_config.replace(
  • "[hooks.state.'%s:pre_tool_use:0:0']" % CODEX_HOOKS,
  • "[hooks.state.'%s:pre_tool_use:mutated:0']" % CODEX_HOOKS, 1)
  • check("mutation: удаление PreToolUse trust state обязано стать красным",
  • trust_state and not _codex_pretool_trust_state(mutated_trust))
  • check("Claude file tools идут через тот же guard",
  • any("constitution_guard" in cmd and "Write" in matcher for matcher, cmd in clpre))
  • check("Claude shell идёт через тот же guard",
  • any("constitution_guard" in cmd and
  • ("Bash" in matcher or "PowerShell" in matcher) for matcher, cmd in clpre))
  • cpost = _commands(codex.get("hooks", {}).get("PostToolUse"))
  • check("Codex PostToolUse явно матчится на canonical apply_patch",
  • any("orphan_check_hook" in cmd and "apply_patch" in matcher
  • for matcher, cmd in cpost), cpost[:3])
  • cstop = _commands(codex.get("hooks", {}).get("Stop"))
  • stop_registered = any("turnstate_hook" in command for _, command in cstop)
  • check("Codex Stop зарегистрирован на fenced release",
  • stop_registered, cstop[:3])
  • mutated_cstop = [(matcher, command) for matcher, command in cstop
  • if "turnstate_hook" not in command]
  • check("mutation: удаление Codex Stop registration обязано стать красным",
  • stop_registered and not any(
  • "turnstate_hook" in command for _, command in mutated_cstop),
  • mutated_cstop[:3]) +
  • if os.path.isfile(GUARD):
  • race_root = os.path.join(root, "parallel-hook")
  • os.makedirs(race_root, exist_ok=True)
  • race_target = os.path.join(race_root, "one-file.md")
  • wins, errors = _guard_race(race_root, race_target)
  • check("параллельный hook: 8 turn -> один win, 7 доказанных BUSY",
  • wins == 1 and not errors,
  • "wins=%d errors=%s" % (wins, errors))
  • patch = "*** Begin Patch\n*** Update File: %s\n@@\n-old\n+new\n*** End Patch" % target
  • p1 = {"session_id": "codex-A", "turn_id": "turn-A",
  • "tool_use_id": "call-A", "cwd": root, "tool_name": "apply_patch",
  • "tool_input": {"command": patch}}
  • p2 = {"session_id": "claude-B", "cwd": root, "tool_name": "Edit",
  • "tool_input": {"file_path": target}}
  • a = _run_guard(p1, env=env)
  • b = _run_guard(p2, env=env)
  • check("canonical apply_patch реально получает lease", a.returncode == 0,
  • "rc=%s stderr=%s" % (a.returncode, a.stderr[:200]))
  • check("второй агент блокируется на том же файле", b.returncode == 2 and
  • "codex-A" in b.stderr, "rc=%s stderr=%s" % (b.returncode, b.stderr[:300]))
  • codex_b = dict(p2)
  • codex_b.update({"tool_name": "apply_patch", "turn_id": "turn-B",
  • "tool_use_id": "call-B",
  • "tool_input": {"command": patch}})
  • structured = _run_constitution(codex_b, env)
  • try:
  • structured_json = json.loads(structured.stdout)
  • permission = structured_json["hookSpecificOutput"]["permissionDecision"]
  • except Exception:
  • permission = ""
  • check("Codex 0.147 получает structured deny, не hook Failed",
  • structured.returncode == 0 and permission == "deny" and
  • "codex-A" in structured.stdout,
  • "rc=%s out=%s err=%s" % (structured.returncode,
  • structured.stdout[:300], structured.stderr[:120]))
  • live_hooks = os.path.normcase(os.path.realpath(
  • os.path.join(HOME, ".claude", "hooks")))
  • if os.path.normcase(os.path.realpath(HOOKS)) == live_hooks:
  • exact_env = dict(env)
  • exact_env["USERPROFILE"] = HOME
  • exact_env["HOME"] = HOME
  • exact = _run_hook_command(configured_pretool, codex_b, exact_env)
  • missing = _run_hook_command(
  • configured_pretool + ".missing", codex_b, exact_env)
  • dropped_stdin = _run_hook_command(
  • configured_pretool +
  • (" < NUL" if os.name == "nt" else " < /dev/null"),
  • codex_b, exact_env)
  • echoed = _run_hook_command(
  • "echo constitution_guard", codex_b, exact_env)
  • check("live hooks.json command исполняется и три command mutants красные",
  • _is_structured_deny(exact, "codex-A") and
  • not _is_structured_deny(missing, "codex-A") and
  • not _is_structured_deny(dropped_stdin, "codex-A") and
  • not _is_structured_deny(echoed, "codex-A"),
  • "exact=%s/%r missing=%s/%r no-stdin=%s/%r echo=%s/%r" %
  • (exact.returncode, exact.stdout[:120],
  • missing.returncode, missing.stdout[:80],
  • dropped_stdin.returncode, dropped_stdin.stdout[:80],
  • echoed.returncode, echoed.stdout[:80]))
  • else:
  • print(" INFO live hooks.json command deferred until mandatory live run")
  • check("constitution audit пишет только в тестовый профиль",
  • os.path.isfile(isolated_audit) and
  • "codex-A" in open(isolated_audit, encoding="utf-8").read(),
  • isolated_audit)
  • stop_payload = {"session_id": "codex-A", "turn_id": "turn-A", "cwd": root,
  • "hook_event_name": "Stop"}
  • stop = subprocess.run([sys.executable, TURNSTATE], input=json.dumps(stop_payload),
  • capture_output=True, text=True, encoding="utf-8",
  • errors="replace", env=env, timeout=30)
  • after = _run_guard(p2, env=env)
  • check("реальный turnstate Stop освобождает turn-lease",
  • stop.returncode == 0 and after.returncode == 0,
  • "stop=%s after=%s" % (stop.returncode, after.returncode))
  • _run_guard({"session_id": "claude-B", "cwd": root}, "stop", env) +
  • alias_target = os.path.join(root, "session-alias.md")
  • alias_patch = ("*** Begin Patch\n*** Update File: %s\n@@\n-old\n+new\n"
  • "*** End Patch" % alias_target)
  • alias_acquire = _run_guard({
  • "session_id": "codex-alias", "turn_id": "turn-alias",
  • "tool_use_id": "call-alias", "cwd": root, "tool_name": "apply_patch",
  • "tool_input": {"command": alias_patch}}, env=env)
  • alias_owner_before = _lease_owner(
  • env["WORKSPACE_WRITE_LEASE_DB"], alias_target)
  • alias_stop = subprocess.run(
  • [sys.executable, TURNSTATE],
  • input=json.dumps({"session_id": "agent/codex-alias",
  • "turn_id": "turn-alias", "cwd": root,
  • "hook_event_name": "Stop"}),
  • capture_output=True, text=True, encoding="utf-8", errors="replace",
  • env=env, timeout=30)
  • alias_owner_after = _lease_owner(
  • env["WORKSPACE_WRITE_LEASE_DB"], alias_target)
  • alias_challenger = _run_guard({
  • "session_id": "alias-challenger", "cwd": root, "tool_name": "Edit",
  • "tool_input": {"file_path": alias_target}}, env=env)
  • check("Stop alias agent/session снимает lease plain session",
  • alias_acquire.returncode == 0 and not alias_acquire.stdout.strip() and
  • alias_owner_before and
  • alias_owner_before.get("session") == "codex-alias" and
  • alias_stop.returncode == 0 and alias_owner_after is None and
  • alias_challenger.returncode == 0,
  • "acquire=%s/%r before=%r stop=%s after=%r challenger=%s" %
  • (alias_acquire.returncode, alias_acquire.stdout[:100],
  • alias_owner_before, alias_stop.returncode, alias_owner_after,
  • alias_challenger.returncode))
  • _run_guard({"session_id": "alias-challenger", "cwd": root}, "stop", env) +
  • epoch_target = os.path.join(root, "epoch-fence.md")
  • epoch_patch = "*** Begin Patch\n*** Update File: %s\n@@\n-old\n+new\n*** End Patch" % epoch_target
  • old_turn = {"session_id": "codex-same", "turn_id": "turn-old",
  • "tool_use_id": "call-old", "cwd": root,
  • "tool_name": "apply_patch",
  • "tool_input": {"command": epoch_patch}}
  • new_turn = dict(old_turn)
  • new_turn.update({"turn_id": "turn-new", "tool_use_id": "call-new"})
  • first_old = _run_guard(old_turn, env=env)
  • old_stop_payload = {"session_id": "codex-same", "turn_id": "turn-old",
  • "cwd": root, "hook_event_name": "Stop"}
  • first_old_stop = subprocess.run(
  • [sys.executable, TURNSTATE], input=json.dumps(old_stop_payload),
  • capture_output=True, text=True, encoding="utf-8", errors="replace",
  • env=env, timeout=30)
  • acquired_new = _run_guard(new_turn, env=env)
  • owner_after_new = _lease_owner(
  • env["WORKSPACE_WRITE_LEASE_DB"], epoch_target)
  • delayed_old_stop = subprocess.run(
  • [sys.executable, TURNSTATE], input=json.dumps(old_stop_payload),
  • capture_output=True, text=True, encoding="utf-8", errors="replace",
  • env=env, timeout=30)
  • owner_after_late_stop = _lease_owner(
  • env["WORKSPACE_WRITE_LEASE_DB"], epoch_target)
  • challenger = _run_guard(
  • {"session_id": "claude-challenger", "cwd": root,
  • "tool_name": "Edit", "tool_input": {"file_path": epoch_target}},
  • env=env)
  • check("поздний Stop старого turn_id не снимает новую эпоху",
  • first_old.returncode == 0 and first_old_stop.returncode == 0 and
  • acquired_new.returncode == 0 and not acquired_new.stdout.strip() and
  • owner_after_new and owner_after_new.get("epoch") == "turn-new" and
  • delayed_old_stop.returncode == 0 and owner_after_late_stop and
  • owner_after_late_stop.get("epoch") == "turn-new" and
  • challenger.returncode == 2 and "codex-same" in challenger.stderr,
  • "old=%s stop=%s new=%s/%r owner-new=%r late=%s owner-late=%r "
  • "challenger=%s/%s" %
  • (first_old.returncode, first_old_stop.returncode,
  • acquired_new.returncode, acquired_new.stdout[:120], owner_after_new,
  • delayed_old_stop.returncode, owner_after_late_stop,
  • challenger.returncode, challenger.stderr[:160]))
  • _run_guard({"session_id": "codex-same", "turn_id": "turn-new",
  • "cwd": root}, "stop", env)
  • post_payload = dict(p1)
  • post_payload.update({"turn_id": "turn-post", "tool_use_id": "call-post",
  • "hook_event_name": "PostToolUse", "tool_response": {"ok": True}})
  • post = subprocess.run([sys.executable, ORPHAN], input=json.dumps(post_payload),
  • capture_output=True, text=True, encoding="utf-8",
  • errors="replace", env=env, timeout=30)
  • check("Codex PostToolUse no-op = пустой stdout, не invalid approve",
  • post.returncode == 0 and not post.stdout.strip(),
  • "rc=%s out=%r err=%r" % (post.returncode, post.stdout, post.stderr[:120]))
  • malformed = subprocess.run([sys.executable, GUARD, "--phase", "pre"],
  • input="not-json", capture_output=True, text=True,
  • env=env, timeout=30)
  • check("битый write-payload блокируется fail-closed", malformed.returncode == 2)
  • readonly = _run_guard({"session_id": "reader", "cwd": root,
  • "tool_name": "Bash",
  • "tool_input": {"command": "git status --short"}}, env=env)
  • check("read-only shell не берёт лишний замок", readonly.returncode == 0)
  • read_target = os.path.join(root, "read-only", "absolute.txt")
  • read_parent = os.path.dirname(read_target)
  • absolute_reader_a = _run_guard({
  • "session_id": "reader-A", "cwd": root, "tool_name": "PowerShell",
  • "tool_input": {"command":
  • 'Get-Content -LiteralPath "%s"' % read_target}}, env=env)
  • absolute_reader_b = _run_guard({
  • "session_id": "reader-B", "cwd": root, "tool_name": "exec_command",
  • "tool_input": {"command":
  • 'Get-FileHash -LiteralPath "%s" -Algorithm SHA256' % read_target}},
  • env=env)
  • rg_with_null = _run_guard({
  • "session_id": "reader-C", "cwd": root, "tool_name": "exec_command",
  • "tool_input": {"command":
  • ('rg -n --hidden "runtime.*(lease|hook)|" "%s" '
  • '2>$null | Select-Object -Last 160') % read_parent}}, env=env)
  • rg_with_quoted_angle = _run_guard({
  • "session_id": "reader-D", "cwd": root, "tool_name": "shell",
  • "tool_input": {"command":
  • "rg -n -l 'session' '%s'" % read_parent}}, env=env)
  • rg_with_child_text = _run_guard({
  • "session_id": "reader-E", "cwd": root, "tool_name": "Bash",
  • "tool_input": {"command":
  • 'rg -n "bash -c \'rm C:\tmp\x\'" "%s"' % read_parent}}, env=env)
  • writer_after_readers = _run_guard({
  • "session_id": "writer-after-readers", "cwd": root, "tool_name": "Edit",
  • "tool_input": {"file_path": read_target}}, env=env)
  • read_owner = _lease_owner(env["WORKSPACE_WRITE_LEASE_DB"], read_target)
  • parent_owner = _lease_owner(env["WORKSPACE_WRITE_LEASE_DB"], read_parent)
  • root_owner = _lease_owner(env["WORKSPACE_WRITE_LEASE_DB"], root)
  • check("absolute readers не лизуют файл или parent",
  • absolute_reader_a.returncode == 0 and
  • absolute_reader_b.returncode == 0 and
  • rg_with_null.returncode == 0 and
  • rg_with_quoted_angle.returncode == 0 and
  • rg_with_child_text.returncode == 0 and
  • writer_after_readers.returncode == 0 and
  • read_owner and read_owner.get("session") == "writer-after-readers" and
  • parent_owner is None and root_owner is None,
  • "readA=%s readB=%s rg-null=%s rg-angle=%s rg-child-text=%s writer=%s "
  • "file=%r parent=%r root=%r" %
  • (absolute_reader_a.returncode, absolute_reader_b.returncode,
  • rg_with_null.returncode, rg_with_quoted_angle.returncode,
  • rg_with_child_text.returncode, writer_after_readers.returncode,
  • read_owner, parent_owner, root_owner))
  • check("literal rg repro не лизует входные absolute roots",
  • rg_with_null.returncode == 0 and
  • rg_with_quoted_angle.returncode == 0 and
  • rg_with_child_text.returncode == 0 and
  • parent_owner is None and root_owner is None)
  • _run_guard({"session_id": "writer-after-readers", "cwd": root}, "stop", env)
  • if HOOKS not in sys.path:
  • sys.path.insert(0, HOOKS)
  • import workspace_write_guard as guard_module
  • check("read-only whitelist не пропускает PowerShell script-block",
  • not guard_module.shell_is_read_only(
  • "Get-Content x | Where-Object { Remove-Item x; $true }"))
  • check("read-only whitelist не пропускает git --output",
  • not guard_module.shell_is_read_only("git diff --output=stolen.patch"))
  • quoted_target = os.path.join(root, "quoted-child-writer.txt")
  • quoted_holder = _run_guard({
  • "session_id": "quoted-holder", "cwd": root, "tool_name": "Edit",
  • "tool_input": {"file_path": quoted_target}}, env=env)
  • quoted_powershell = _run_guard({
  • "session_id": "quoted-powershell", "cwd": root, "tool_name": "Bash",
  • "tool_input": {"command":
  • 'powershell -Command "Set-Content -LiteralPath \'%s\' -Value y"' %
  • quoted_target}}, env=env)
  • quoted_cmd = _run_guard({
  • "session_id": "quoted-cmd", "cwd": root, "tool_name": "exec_command",
  • "tool_input": {"command":
  • 'cmd /c "echo y > %s"' % quoted_target}}, env=env)
  • quoted_bash = _run_guard({
  • "session_id": "quoted-bash", "cwd": root, "tool_name": "Bash",
  • "tool_input": {"command":
  • 'bash -lc "echo y > %s"' % quoted_target}}, env=env)
  • quoted_sh = _run_guard({
  • "session_id": "quoted-sh", "cwd": root, "tool_name": "shell",
  • "tool_input": {"command":
  • 'sh -ec "echo y > %s"' % quoted_target}}, env=env)
  • quoted_path_pwsh = _run_guard({
  • "session_id": "quoted-path-pwsh", "cwd": root,
  • "tool_name": "PowerShell",
  • "tool_input": {"command":
  • '& "C:\Program Files\PowerShell\7\pwsh.exe" -Command '
  • '"Set-Content -LiteralPath \'%s\' -Value y"' % quoted_target}},
  • env=env)
  • quoted_path_cmd = _run_guard({
  • "session_id": "quoted-path-cmd", "cwd": root,
  • "tool_name": "exec_command",
  • "tool_input": {"command":
  • '"C:\Windows\System32\cmd.exe" /c "echo y > %s"' %
  • quoted_target}}, env=env)
  • quoted_path_pwsh_options = _run_guard({
  • "session_id": "quoted-path-pwsh-options", "cwd": root,
  • "tool_name": "PowerShell",
  • "tool_input": {"command":
  • '& "C:\Program Files\PowerShell\7\pwsh.exe" -NoProfile '
  • '-ExecutionPolicy Bypass -Command '
  • '"Set-Content -LiteralPath \'%s\' -Value y"' % quoted_target}},
  • env=env)
  • quoted_flag_pwsh = _run_guard({
  • "session_id": "quoted-flag-pwsh", "cwd": root,
  • "tool_name": "PowerShell",
  • "tool_input": {"command":
  • 'powershell "-Command" '
  • '"Set-Content -LiteralPath \'%s\' -Value y"' % quoted_target}},
  • env=env)
  • command_with_args = _run_guard({
  • "session_id": "command-with-args", "cwd": root,
  • "tool_name": "PowerShell",
  • "tool_input": {"command":
  • 'pwsh -CommandWithArgs '
  • '"Set-Content -LiteralPath \'%s\' -Value y"' % quoted_target}},
  • env=env)
  • working_directory = _run_guard({
  • "session_id": "working-directory", "cwd": root,
  • "tool_name": "PowerShell",
  • "tool_input": {"command":
  • 'pwsh -WorkingDirectory \'%s\' -Command '
  • '"Set-Content -LiteralPath \'%s\' -Value y"' %
  • (root, quoted_target)}}, env=env)
  • quoted_call_cmdlet = _run_guard({
  • "session_id": "quoted-call-cmdlet", "cwd": root,
  • "tool_name": "PowerShell",
  • "tool_input": {"command":
  • 'powershell -Command '
  • '"& \'Set-Content\' -LiteralPath \'%s\' -Value y"' %
  • quoted_target}}, env=env)
  • single_amp_chain = _run_guard({
  • "session_id": "single-amp-chain", "cwd": root,
  • "tool_name": "PowerShell",
  • "tool_input": {"command":
  • 'echo ok & powershell -Command '
  • '"Set-Content -LiteralPath \'%s\' -Value y"' % quoted_target}},
  • env=env)Source：https://dev.to/cryptopunk_dz2_bd054446a6/the-one-time-migration-fallback-that-runs-on-every-call-forever-4mb4
    My favorites