AI-2146: fix stale notes.md and silent permission errors in init_task_record.py

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,54 +21,81 @@ 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()
 
 
 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)
 

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)