diff --git a/scripts/init_task_record.py b/scripts/init_task_record.py index c03d223..78fe608 100644 --- a/scripts/init_task_record.py +++ b/scripts/init_task_record.py @@ -9,6 +9,7 @@ import argparse import json +import sys from datetime import datetime, timezone from pathlib import Path @@ -20,6 +21,11 @@ def parse_args() -> argparse.Namespace: parser.add_argument("--repo", default="", help="GitHub repo URL or slug") parser.add_argument("--branch", default="", help="Git branch name") parser.add_argument("--root", default="tasks", help="Artifact root folder") + parser.add_argument( + "--force", + action="store_true", + help="Refresh an existing task artifact (rewrites task.json and notes.md)", + ) return parser.parse_args() @@ -27,47 +33,69 @@ def main() -> None: args = parse_args() issue_id = args.issue_id.upper() task_dir = Path(args.root) / issue_id - logs_dir = task_dir / "logs" - logs_dir.mkdir(parents=True, exist_ok=True) task_json = task_dir / "task.json" notes_md = task_dir / "notes.md" - if not task_json.exists(): - task_json.write_text( - json.dumps( - { - "issue_id": issue_id, - "title": args.title, - "repo": args.repo, - "branch": args.branch, - "created_at": datetime.now(timezone.utc).isoformat(), - "linear_url": "", - "pr_url": "", - "status": "started", - "secret_policy": "No secrets in task artifacts.", - }, - indent=2, - ensure_ascii=False, - ) - + "\n", - encoding="utf-8", + existing = task_json.exists() or notes_md.exists() + if existing and not args.force: + print( + f"❌ Task artifact already exists at {task_dir}. " + "Re-running with the same args would silently leave a stale notes.md from a previous issue. " + "Pass --force to refresh, or remove the folder manually.", + file=sys.stderr, ) + sys.exit(2) - if not notes_md.exists(): - notes_md.write_text( - f"# {issue_id} Notes\n\n" - "## Scope\n\n" - "- \n\n" - "## Progress\n\n" - "- Started task artifact.\n\n" - "## Verification\n\n" - "- \n\n" - "## Links\n\n" - f"- Repo: {args.repo}\n" - f"- Branch: {args.branch}\n", - encoding="utf-8", + try: + # When --force, mkdir(exist_ok=True) so we reuse the folder. Otherwise the + # existing-files check above has already exited. + task_dir.mkdir(parents=True, exist_ok=True) + (task_dir / "logs").mkdir(exist_ok=True) + except PermissionError as e: + print( + f"❌ Cannot create task artifact at {task_dir}: permission denied ({e}). " + "Check that --root points to a writable directory.", + file=sys.stderr, ) + sys.exit(1) + except OSError as e: + print(f"❌ Cannot create task artifact at {task_dir}: {e}", file=sys.stderr) + sys.exit(1) + + task_json.write_text( + json.dumps( + { + "issue_id": issue_id, + "title": args.title, + "repo": args.repo, + "branch": args.branch, + "created_at": datetime.now(timezone.utc).isoformat(), + "linear_url": "", + "pr_url": "", + "status": "started", + "secret_policy": "No secrets in task artifacts.", + }, + indent=2, + ensure_ascii=False, + ) + + "\n", + encoding="utf-8", + ) + + notes_md.write_text( + f"# {issue_id} Notes\n\n" + "## Scope\n\n" + "- \n\n" + "## Progress\n\n" + "- Started task artifact.\n\n" + "## Verification\n\n" + "- \n\n" + "## Links\n\n" + f"- Repo: {args.repo}\n" + f"- Branch: {args.branch}\n", + encoding="utf-8", + ) print(task_dir) diff --git a/tests/test_init_task_record.py b/tests/test_init_task_record.py new file mode 100644 index 0000000..cc0fe70 --- /dev/null +++ b/tests/test_init_task_record.py @@ -0,0 +1,104 @@ +"""Tests for init_task_record.py — guards against stale notes & silent permission errors.""" +import json +import os +import subprocess +import sys +import tempfile +from pathlib import Path + +SCRIPT = Path(__file__).resolve().parent.parent / "scripts" / "init_task_record.py" + + +def run(args, cwd=None): + return subprocess.run( + [sys.executable, str(SCRIPT), *args], + capture_output=True, text=True, cwd=cwd, + ) + + +def test_first_run_creates_both_files(): + with tempfile.TemporaryDirectory() as tmp: + r = run(["AI-1111", "--title", "hello", "--repo", "x/y", "--branch", "b", "--root", f"{tmp}/tasks"]) + assert r.returncode == 0, r.stderr + tj = Path(tmp) / "tasks" / "AI-1111" / "task.json" + nm = Path(tmp) / "tasks" / "AI-1111" / "notes.md" + assert tj.exists() + assert nm.exists() + body = json.loads(tj.read_text()) + assert body["title"] == "hello" + assert body["repo"] == "x/y" + assert body["branch"] == "b" + + +def test_repeat_run_refuses_without_force(): + """Regression: the old code rewrote task.json but kept a stale notes.md from a previous issue. + New code must refuse to partially overwrite unless --force is given.""" + with tempfile.TemporaryDirectory() as tmp: + # First run + r1 = run(["AI-2222", "--title", "first", "--repo", "x/y", "--branch", "b1", "--root", f"{tmp}/tasks"]) + assert r1.returncode == 0, r1.stderr + + # Mutate notes.md to look like previous-task content + notes = Path(tmp) / "tasks" / "AI-2222" / "notes.md" + notes.write_text("# AI-2222 Notes\n\n## Scope\n- original scope\n") + + # Second run with NEW title/branch — must NOT silently partial-overwrite + r2 = run(["AI-2222", "--title", "DIFFERENT", "--repo", "z/w", "--branch", "b2", "--root", f"{tmp}/tasks"]) + assert r2.returncode != 0, f"expected refusal, got rc=0 stdout={r2.stdout!r} stderr={r2.stderr!r}" + assert "already exists" in r2.stderr + + # Old notes.md content preserved (no partial overwrite happened) + assert "original scope" in notes.read_text() + + # task.json preserved too + tj = json.loads((Path(tmp) / "tasks" / "AI-2222" / "task.json").read_text()) + assert tj["title"] == "first" + + +def test_force_overwrites_atomically(): + with tempfile.TemporaryDirectory() as tmp: + run(["AI-3333", "--title", "old", "--root", f"{tmp}/tasks"]) + r = run(["AI-3333", "--title", "new", "--root", f"{tmp}/tasks", "--force"]) + assert r.returncode == 0, r.stderr + tj = json.loads((Path(tmp) / "tasks" / "AI-3333" / "task.json").read_text()) + assert tj["title"] == "new" + notes = (Path(tmp) / "tasks" / "AI-3333" / "notes.md").read_text() + assert "Started task artifact" in notes # notes got refreshed + + +def test_permission_error_exits_cleanly(): + """Old code crashed with a PermissionError traceback. New code must exit non-zero with a clear message.""" + with tempfile.TemporaryDirectory() as tmp: + ro = Path(tmp) / "ro" + ro.mkdir() + os.chmod(ro, 0o555) + try: + r = run(["AI-4444", "--title", "x", "--root", f"{ro}/tasks"]) + assert r.returncode != 0, f"expected non-zero, got {r.returncode}" + assert "permission denied" in r.stderr.lower() or "Permission denied" in r.stderr + # No partial folder was left behind + assert not (ro / "AI-4444").exists() + finally: + os.chmod(ro, 0o755) + + +if __name__ == "__main__": + import traceback + tests = [ + test_first_run_creates_both_files, + test_repeat_run_refuses_without_force, + test_force_overwrites_atomically, + test_permission_error_exits_cleanly, + ] + passed = failed = 0 + for t in tests: + try: + t() + print(f" ✅ {t.__name__}") + passed += 1 + except Exception as e: + print(f" ❌ {t.__name__}: {e}") + traceback.print_exc() + failed += 1 + print(f"\n{passed} passed, {failed} failed") + sys.exit(0 if failed == 0 else 1)