mirror of
https://github.com/OthmanAdi/planning-with-files.git
synced 2026-09-14 16:09:40 +08:00
OthmanAdi
472 lines
20 KiB
Python
472 lines
20 KiB
Python
"""Nested plan-root isolation for the Codex hook route (issue #212).
|
|
|
|
tests/test_nested_plan_isolation.py covers the canonical inject-plan.sh
|
|
dispatcher. This file covers the OTHER route with the same bug: the
|
|
independently written .codex/hooks.json integration, which is the route
|
|
Codex actually executes. Three semantics, ported from inject-plan.sh:
|
|
|
|
* PWF_PLAN_ROOT — absolute plan-root binding, highest precedence.
|
|
.codex/hooks/user-prompt-submit.sh prefixes every planning-state read
|
|
with it, the shared resolver resolves under it, and the Python adapter
|
|
(pre_tool_use / post_tool_use / stop / permission_request) routes its
|
|
shell helpers and the session-attachment check through it. A broken pin
|
|
fails CLOSED: the once-per-turn hook emits one notice, the per-tool-call
|
|
hooks refuse silently.
|
|
* Session-attachment notice — the guard itself predates this change; the
|
|
refusal now announces itself (once per turn, user-prompt-submit only)
|
|
with the same wording as inject-plan.sh.
|
|
* Nested-root conflict detection — a cwd GUESS (active-plan pointer,
|
|
newest-by-mtime, or legacy root task_plan.md) refuses to inject when a
|
|
direct child of the effective root carries its own competing .planning.
|
|
Explicit resolution (valid PLAN_ID, valid pin, attached session) skips
|
|
the check.
|
|
|
|
Tree used throughout (the reporter's shape):
|
|
|
|
workspace/.planning/plan-a (active via .active_plan)
|
|
workspace/project/.planning/plan-b (active via its own .active_plan)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import hashlib
|
|
import os
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
HOOKS_DIR = REPO_ROOT / ".codex" / "hooks"
|
|
|
|
PLAN_A_TITLE = "PLAN-ALPHA-PARENT-WORKSPACE"
|
|
PLAN_B_TITLE = "PLAN-BRAVO-NESTED-PROJECT"
|
|
|
|
SCRUB_VARS = (
|
|
"PLAN_ID",
|
|
"PWF_PLAN_ROOT",
|
|
"PWF_SESSION_ID",
|
|
"PWF_SESSION_KEY",
|
|
"PWF_INJECT",
|
|
"PLANNING_DISABLED",
|
|
)
|
|
|
|
|
|
def session_key(root: Path, session_id: str) -> str:
|
|
digest = hashlib.sha256()
|
|
project = os.path.normcase(os.path.realpath(os.path.abspath(root))).replace("\\", "/")
|
|
for value in ("codex", project, session_id):
|
|
encoded = value.encode("utf-8", errors="surrogatepass")
|
|
digest.update(len(encoded).to_bytes(8, "big"))
|
|
digest.update(encoded)
|
|
return digest.hexdigest()
|
|
|
|
|
|
def shell_and_env() -> tuple[str, dict[str, str]]:
|
|
"""POSIX sh, or the real Git-for-Windows sh plus its coreutils PATH."""
|
|
env = os.environ.copy()
|
|
if os.name != "nt":
|
|
shell = shutil.which("sh")
|
|
if shell is None:
|
|
raise unittest.SkipTest("POSIX sh is unavailable")
|
|
return shell, env
|
|
|
|
sys.path.insert(0, str(HOOKS_DIR))
|
|
try:
|
|
import codex_hook_adapter as adapter
|
|
|
|
shell, extra_path_dirs = adapter._windows_git_bash()
|
|
finally:
|
|
sys.path.pop(0)
|
|
if shell is None:
|
|
raise unittest.SkipTest("Git for Windows sh.exe is unavailable")
|
|
if extra_path_dirs:
|
|
env["PATH"] = os.pathsep.join([*extra_path_dirs, env.get("PATH", "")])
|
|
env["PWF_TRUSTED_PYTHON"] = sys.executable
|
|
return shell, env
|
|
|
|
|
|
class NestedTreeMixin(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.tmp = Path(tempfile.mkdtemp(prefix="pwf-codex-nested-"))
|
|
self.workspace = self.tmp / "workspace"
|
|
self.project = self.workspace / "project"
|
|
|
|
def tearDown(self) -> None:
|
|
shutil.rmtree(self.tmp, ignore_errors=True)
|
|
|
|
def build_tree(self, nested: bool = True, phased_nested: bool = False) -> None:
|
|
plan_a = self.workspace / ".planning" / "plan-a"
|
|
plan_a.mkdir(parents=True)
|
|
(plan_a / "task_plan.md").write_text(f"# {PLAN_A_TITLE}\n", encoding="utf-8")
|
|
(plan_a / "progress.md").write_text("# progress a\n", encoding="utf-8")
|
|
(self.workspace / ".planning" / ".active_plan").write_text(
|
|
"plan-a\n", encoding="utf-8"
|
|
)
|
|
if nested:
|
|
plan_b = self.project / ".planning" / "plan-b"
|
|
plan_b.mkdir(parents=True)
|
|
body = f"# {PLAN_B_TITLE}\n"
|
|
if phased_nested:
|
|
body += "\n### Phase 1: Work\n- **Status:** in_progress\n"
|
|
(plan_b / "task_plan.md").write_text(body, encoding="utf-8")
|
|
(plan_b / "progress.md").write_text("# progress b\n", encoding="utf-8")
|
|
(self.project / ".planning" / ".active_plan").write_text(
|
|
"plan-b\n", encoding="utf-8"
|
|
)
|
|
|
|
|
|
class CodexUserPromptSubmitNestedRootTests(NestedTreeMixin):
|
|
"""The once-per-turn shell hook: notices allowed, fail closed on ambiguity."""
|
|
|
|
def _run(self, env_extra: dict[str, str] | None = None) -> subprocess.CompletedProcess[str]:
|
|
shell, env = shell_and_env()
|
|
for var in SCRUB_VARS:
|
|
env.pop(var, None)
|
|
if env_extra:
|
|
env.update(env_extra)
|
|
return subprocess.run(
|
|
[shell, str(HOOKS_DIR / "user-prompt-submit.sh")],
|
|
cwd=str(self.workspace),
|
|
text=True,
|
|
encoding="utf-8",
|
|
capture_output=True,
|
|
env=env,
|
|
check=False,
|
|
)
|
|
|
|
# -- 1. ambiguous parent cwd: fail closed ------------------------------
|
|
def test_parent_cwd_with_nested_plan_injects_nothing(self) -> None:
|
|
self.build_tree(nested=True)
|
|
result = self._run()
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertNotIn(PLAN_A_TITLE, result.stdout, "parent plan body must not inject")
|
|
self.assertNotIn(PLAN_B_TITLE, result.stdout, "nested plan body must not inject")
|
|
self.assertIn("Ambiguous plan", result.stdout)
|
|
self.assertIn("project", result.stdout, "warning must name the nested root")
|
|
self.assertIn("PWF_PLAN_ROOT", result.stdout)
|
|
self.assertIn("PLAN_ID", result.stdout)
|
|
|
|
# -- 2. PWF_PLAN_ROOT pins the nested project --------------------------
|
|
def test_plan_root_pin_from_parent_cwd_injects_nested_plan(self) -> None:
|
|
self.build_tree(nested=True)
|
|
result = self._run({"PWF_PLAN_ROOT": str(self.project)})
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertIn(PLAN_B_TITLE, result.stdout, "pinned root's plan must inject")
|
|
self.assertNotIn(PLAN_A_TITLE, result.stdout, "parent plan must not leak through")
|
|
self.assertIn("ACTIVE PLAN", result.stdout)
|
|
|
|
# -- 3. broken pin fails closed, never falls back ----------------------
|
|
def test_plan_root_broken_pin_notices_and_injects_nothing(self) -> None:
|
|
self.build_tree(nested=True)
|
|
result = self._run({"PWF_PLAN_ROOT": str(self.workspace / "does-not-exist")})
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertNotEqual("", result.stdout.strip(), "broken pin must emit a notice")
|
|
self.assertIn("PWF_PLAN_ROOT", result.stdout)
|
|
self.assertNotIn(PLAN_A_TITLE, result.stdout, "must not fall back to cwd plan")
|
|
self.assertNotIn(PLAN_B_TITLE, result.stdout)
|
|
|
|
def test_relative_and_sibling_shell_pins_fail_closed(self) -> None:
|
|
self.build_tree(nested=True)
|
|
sibling = self.tmp / "sibling"
|
|
sibling.mkdir()
|
|
for pin in ("project", str(sibling)):
|
|
with self.subTest(pin=pin):
|
|
result = self._run({"PWF_PLAN_ROOT": pin})
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertIn("PWF_PLAN_ROOT", result.stdout)
|
|
self.assertNotIn(PLAN_A_TITLE, result.stdout)
|
|
self.assertNotIn(PLAN_B_TITLE, result.stdout)
|
|
|
|
# -- 4. explicit PLAN_ID beats the conflict check ----------------------
|
|
def test_valid_plan_id_at_parent_still_injects_despite_nested_plan(self) -> None:
|
|
self.build_tree(nested=True)
|
|
result = self._run({"PLAN_ID": "plan-a"})
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertIn(
|
|
PLAN_A_TITLE,
|
|
result.stdout,
|
|
"explicit PLAN_ID names the plan deliberately; conflict check must not fire",
|
|
)
|
|
self.assertNotIn("Ambiguous plan", result.stdout)
|
|
|
|
# -- 5. session guard now says why it refuses --------------------------
|
|
def test_unattached_session_refuses_with_notice(self) -> None:
|
|
self.build_tree(nested=False)
|
|
(self.workspace / ".planning" / "sessions").mkdir()
|
|
result = self._run({"PWF_SESSION_ID": "sess-1"})
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertNotIn(PLAN_A_TITLE, result.stdout)
|
|
self.assertIn("Session isolation is armed", result.stdout)
|
|
# The notice must name both recovery routes.
|
|
self.assertIn("PWF_SESSION_ID", result.stdout)
|
|
self.assertIn("delete", result.stdout)
|
|
|
|
def test_sessions_dir_with_no_session_id_says_why(self) -> None:
|
|
self.build_tree(nested=False)
|
|
(self.workspace / ".planning" / "sessions").mkdir()
|
|
result = self._run()
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertNotIn(PLAN_A_TITLE, result.stdout)
|
|
self.assertIn("Session isolation is armed", result.stdout)
|
|
|
|
# -- 6. attached session injects and beats the conflict check ----------
|
|
def test_attached_session_is_explicit_and_beats_conflict_check(self) -> None:
|
|
self.build_tree(nested=True)
|
|
sessions = self.workspace / ".planning" / "sessions"
|
|
sessions.mkdir()
|
|
(sessions / f"{session_key(self.workspace, 'sess-1')}.attached").write_text("", encoding="utf-8")
|
|
result = self._run({"PWF_SESSION_ID": "sess-1"})
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertIn(PLAN_A_TITLE, result.stdout)
|
|
self.assertNotIn("Ambiguous plan", result.stdout)
|
|
|
|
# -- 7. legacy invariant: no nesting = unchanged injection -------------
|
|
def test_no_nested_plan_injects_exactly_as_before(self) -> None:
|
|
self.build_tree(nested=False)
|
|
result = self._run()
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertIn("ACTIVE PLAN", result.stdout)
|
|
self.assertIn(PLAN_A_TITLE, result.stdout)
|
|
self.assertNotIn("Ambiguous plan", result.stdout)
|
|
self.assertNotIn("Session isolation", result.stdout)
|
|
|
|
def test_bare_nested_planning_is_no_conflict(self) -> None:
|
|
# A nested .planning without .active_plan and without any
|
|
# <slug>/task_plan.md is not a competing plan.
|
|
self.build_tree(nested=False)
|
|
(self.project / ".planning").mkdir(parents=True)
|
|
result = self._run()
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertIn(PLAN_A_TITLE, result.stdout)
|
|
self.assertNotIn("Ambiguous plan", result.stdout)
|
|
|
|
def test_nested_slug_plan_without_pointer_still_competes(self) -> None:
|
|
# Competing = .active_plan OR at least one <slug>/task_plan.md.
|
|
self.build_tree(nested=True)
|
|
(self.project / ".planning" / ".active_plan").unlink()
|
|
result = self._run()
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertIn("Ambiguous plan", result.stdout)
|
|
self.assertNotIn(PLAN_A_TITLE, result.stdout)
|
|
|
|
# -- 8. PLANNING_DISABLED still silences everything --------------------
|
|
def test_planning_disabled_beats_broken_pin(self) -> None:
|
|
self.build_tree(nested=True)
|
|
result = self._run(
|
|
{
|
|
"PLANNING_DISABLED": "1",
|
|
"PWF_PLAN_ROOT": str(self.workspace / "does-not-exist"),
|
|
}
|
|
)
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertEqual("", result.stdout.strip())
|
|
|
|
|
|
class CodexAdapterPlanRootPinTests(NestedTreeMixin):
|
|
"""The Python adapter route: per-tool-call hooks fail closed SILENTLY."""
|
|
|
|
def run_python_hook(
|
|
self, script_name: str, payload: dict, env_extra: dict[str, str] | None = None
|
|
) -> subprocess.CompletedProcess[str]:
|
|
env = os.environ.copy()
|
|
for var in SCRUB_VARS:
|
|
env.pop(var, None)
|
|
if env_extra:
|
|
env.update(env_extra)
|
|
return subprocess.run(
|
|
[sys.executable, str(HOOKS_DIR / script_name)],
|
|
input=json.dumps(payload),
|
|
text=True,
|
|
encoding="utf-8",
|
|
capture_output=True,
|
|
cwd=str(self.workspace),
|
|
env=env,
|
|
check=False,
|
|
)
|
|
|
|
def test_effective_plan_root_unit_semantics(self) -> None:
|
|
sys.path.insert(0, str(HOOKS_DIR))
|
|
try:
|
|
import codex_hook_adapter as adapter
|
|
|
|
self.build_tree(nested=True)
|
|
for var in SCRUB_VARS:
|
|
os.environ.pop(var, None)
|
|
try:
|
|
# Unset: cwd passes through untouched (legacy invariant).
|
|
self.assertEqual(
|
|
self.workspace, adapter.effective_plan_root(self.workspace)
|
|
)
|
|
# Valid pin: the pinned root wins.
|
|
os.environ["PWF_PLAN_ROOT"] = str(self.project)
|
|
self.assertEqual(
|
|
self.project.resolve(),
|
|
adapter.effective_plan_root(self.workspace),
|
|
)
|
|
# Broken pin: fail closed.
|
|
os.environ["PWF_PLAN_ROOT"] = str(self.workspace / "missing")
|
|
self.assertIsNone(adapter.effective_plan_root(self.workspace))
|
|
finally:
|
|
os.environ.pop("PWF_PLAN_ROOT", None)
|
|
finally:
|
|
sys.path.pop(0)
|
|
|
|
def test_relative_and_sibling_plan_root_pins_fail_closed(self) -> None:
|
|
sys.path.insert(0, str(HOOKS_DIR))
|
|
try:
|
|
import codex_hook_adapter as adapter
|
|
|
|
self.build_tree(nested=True)
|
|
sibling = self.tmp / "sibling"
|
|
sibling.mkdir()
|
|
try:
|
|
os.environ["PWF_PLAN_ROOT"] = "project"
|
|
self.assertIsNone(adapter.effective_plan_root(self.workspace))
|
|
os.environ["PWF_PLAN_ROOT"] = str(sibling)
|
|
self.assertIsNone(adapter.effective_plan_root(self.workspace))
|
|
finally:
|
|
os.environ.pop("PWF_PLAN_ROOT", None)
|
|
finally:
|
|
sys.path.pop(0)
|
|
|
|
def test_run_sh_valid_pin_uses_nested_effective_root(self) -> None:
|
|
self.build_tree(nested=True)
|
|
# run_sh requires the shell script name as argv[1], so exercise it
|
|
# directly here rather than through run_python_hook's fixed argv.
|
|
env = os.environ.copy()
|
|
for var in SCRUB_VARS:
|
|
env.pop(var, None)
|
|
env["PWF_PLAN_ROOT"] = str(self.project)
|
|
result = subprocess.run(
|
|
[sys.executable, str(HOOKS_DIR / "run_sh.py"), "user-prompt-submit.sh"],
|
|
input=json.dumps({"cwd": str(self.workspace), "session_id": "sess-x"}),
|
|
text=True,
|
|
encoding="utf-8",
|
|
capture_output=True,
|
|
cwd=self.workspace,
|
|
env=env,
|
|
check=False,
|
|
)
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
if os.name == "nt" and not result.stdout.strip():
|
|
self.skipTest("Git for Windows sh.exe not resolvable on this runner")
|
|
payload = json.loads(result.stdout)
|
|
context = payload["hookSpecificOutput"]["additionalContext"]
|
|
self.assertIn(PLAN_B_TITLE, context)
|
|
self.assertNotIn(PLAN_A_TITLE, context)
|
|
|
|
def test_pre_tool_use_broken_pin_is_silent(self) -> None:
|
|
self.build_tree(nested=True)
|
|
result = self.run_python_hook(
|
|
"pre_tool_use.py",
|
|
{"cwd": str(self.workspace), "tool_input": {"command": "pwd"}},
|
|
{"PWF_PLAN_ROOT": str(self.workspace / "does-not-exist")},
|
|
)
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertEqual("", result.stdout.strip(), "per-tool-call hooks refuse silently")
|
|
|
|
def test_post_tool_use_broken_pin_is_silent(self) -> None:
|
|
self.build_tree(nested=True)
|
|
result = self.run_python_hook(
|
|
"post_tool_use.py",
|
|
{"cwd": str(self.workspace), "tool_response": "ok"},
|
|
{"PWF_PLAN_ROOT": str(self.workspace / "does-not-exist")},
|
|
)
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertEqual("", result.stdout.strip())
|
|
|
|
def test_stop_broken_pin_is_silent(self) -> None:
|
|
self.build_tree(nested=True, phased_nested=True)
|
|
result = self.run_python_hook(
|
|
"stop.py",
|
|
{"cwd": str(self.workspace), "stop_hook_active": False},
|
|
{"PWF_PLAN_ROOT": str(self.workspace / "does-not-exist")},
|
|
)
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertEqual("", result.stdout.strip())
|
|
|
|
def test_permission_request_broken_pin_is_silent(self) -> None:
|
|
self.build_tree(nested=True)
|
|
result = self.run_python_hook(
|
|
"permission_request.py",
|
|
{"cwd": str(self.workspace), "tool_name": "Bash"},
|
|
{"PWF_PLAN_ROOT": str(self.workspace / "does-not-exist")},
|
|
)
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertEqual("", result.stdout.strip())
|
|
|
|
def test_pre_tool_use_valid_pin_injects_nested_context(self) -> None:
|
|
self.build_tree(nested=True)
|
|
result = self.run_python_hook(
|
|
"pre_tool_use.py",
|
|
{"cwd": str(self.workspace), "tool_input": {"command": "pwd"}},
|
|
{"PWF_PLAN_ROOT": str(self.project)},
|
|
)
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
if os.name == "nt" and not result.stdout.strip():
|
|
self.skipTest("Git for Windows sh.exe not resolvable on this runner")
|
|
payload = json.loads(result.stdout)
|
|
context = payload["hookSpecificOutput"]["additionalContext"]
|
|
self.assertIn(PLAN_B_TITLE, context, "pinned root's plan must reach the model")
|
|
self.assertNotIn(PLAN_A_TITLE, context, "parent plan must not leak through")
|
|
|
|
def test_stop_valid_pin_reports_nested_plan_phases(self) -> None:
|
|
self.build_tree(nested=True, phased_nested=True)
|
|
result = self.run_python_hook(
|
|
"stop.py",
|
|
{"cwd": str(self.workspace), "stop_hook_active": False},
|
|
{"PWF_PLAN_ROOT": str(self.project)},
|
|
)
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
if os.name == "nt" and not result.stdout.strip():
|
|
self.skipTest("Git for Windows sh.exe not resolvable on this runner")
|
|
payload = json.loads(result.stdout)
|
|
self.assertIn("Task in progress", payload["systemMessage"])
|
|
|
|
def test_permission_request_valid_pin_emits_reminder(self) -> None:
|
|
self.build_tree(nested=True)
|
|
result = self.run_python_hook(
|
|
"permission_request.py",
|
|
{"cwd": str(self.workspace), "tool_name": "Bash"},
|
|
{"PWF_PLAN_ROOT": str(self.project)},
|
|
)
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
if os.name == "nt" and not result.stdout.strip():
|
|
self.skipTest("Git for Windows sh.exe not resolvable on this runner")
|
|
payload = json.loads(result.stdout)
|
|
self.assertIn("Active plan", payload["systemMessage"])
|
|
|
|
def test_pinned_sessions_guard_reads_the_pinned_root(self) -> None:
|
|
# Session isolation armed at the PIN, not at the cwd: an unattached
|
|
# session must refuse even though the cwd has no sessions dir.
|
|
self.build_tree(nested=True)
|
|
(self.project / ".planning" / "sessions").mkdir()
|
|
result = self.run_python_hook(
|
|
"pre_tool_use.py",
|
|
{"cwd": str(self.workspace), "session_id": "sess-x", "tool_input": {}},
|
|
{"PWF_PLAN_ROOT": str(self.project)},
|
|
)
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
self.assertEqual("", result.stdout.strip())
|
|
|
|
def test_unpinned_adapter_behavior_unchanged(self) -> None:
|
|
# Legacy invariant at the adapter layer: no pin, cwd resolution.
|
|
self.build_tree(nested=False)
|
|
result = self.run_python_hook(
|
|
"pre_tool_use.py",
|
|
{"cwd": str(self.workspace), "tool_input": {"command": "pwd"}},
|
|
)
|
|
self.assertEqual(0, result.returncode, result.stderr)
|
|
if os.name == "nt" and not result.stdout.strip():
|
|
self.skipTest("Git for Windows sh.exe not resolvable on this runner")
|
|
payload = json.loads(result.stdout)
|
|
self.assertIn(PLAN_A_TITLE, payload["hookSpecificOutput"]["additionalContext"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|