fix(planner): harness fixes F1–F6

skills/scripts/skills/planner/README.md

@@ -17,7 +17,7 @@ QR PASS/FAIL is determined by LLM reading QR output, not Python. Gate routing is
 
 ## State Files
 
-All state mutations (except initial context.json) happen via Python CLI commands. State directory created via `tempfile.mkdtemp()` in `/tmp`.
+All state mutations (except initial context.json) happen via Python CLI commands. Step 1 honors an explicit `--state-dir` (resumable); otherwise it mints a fresh persistent directory under `<project>/.claude/planner-state/` (git-ignored) -- never `/tmp`, which a session-limit restart would wipe (F1).
 
 | File              | Schema         | Created     | Mutated By     | Lifecycle              |
 | ----------------- | -------------- | ----------- | -------------- | ---------------------- |

skills/scripts/skills/planner/architect/plan_design_execute.py

@@ -281,14 +281,23 @@ def get_step_guidance(
                 "  GOOD: M1=auth stack, M2=users stack, M3=posts stack (parallel)",
                 "  If file overlap: extract to M0 (foundation) or consolidate",
                 "",
-                "VALIDATION: After building plan.json, run:",
+                "VALIDATION: After building plan.json, run BOTH:",
                 "  python3 -m skills.planner.cli.plan validate --phase plan-design",
+                "  python3 -m skills.planner.cli.plan validate-planning-context",
+                "",
+                "PLANNING_CONTEXT SHAPE (F6 -- self-correct before PASS):",
+                "  - constraints is a list of plain STRINGS, not objects.",
+                "  - each rejected_alternatives entry needs both a rejection_reason",
+                "    and a decision_ref (DL-XXX).",
+                "  If validate-planning-context reports <planning_context_errors>,",
+                "  fix the shape via CLI and re-run until it passes -- do NOT hand a",
+                "  malformed planning_context to a later step.",
                 "",
                 "REFERENCE SCHEMA:",
                 "",
                 plan_json_schema,
                 "",
-                "When plan.json written and validation passes, output: PASS",
+                "When plan.json written and BOTH validations pass, output: PASS",
             ],
             "next": "",
         }

skills/scripts/skills/planner/architect/plan_design_qr_fix.py

@@ -23,6 +23,7 @@
     get_context_path,
     render_context_file,
 )
+from skills.planner.shared.fix_mode import CLASS_SWEEP_DIRECTIVE
 from skills.planner.shared.qr.utils import (
     load_qr_state,
     format_failed_items_for_fix,
@@ -66,19 +67,21 @@ def get_step_guidance(
                 "",
                 failed_items_block if failed_items_block else "Read QR report from: STATE_DIR/qr-plan-design.json",
                 "",
+                CLASS_SWEEP_DIRECTIVE,
+                "",
                 "PLANNING CONTEXT (reference for semantic validation):",
                 "",
                 context_display,
                 "",
                 "For EACH failed item:",
-                "  1. Read the 'finding' field to understand the issue",
-                "  2. Identify what in plan.json needs to change",
+                "  1. Read the 'finding' field to understand the issue and its CLASS",
+                "  2. Identify what in plan.json needs to change -- and every other",
+                "     instance of that class across the plan",
                 "  3. Note the fix approach for step 2",
                 "",
                 "CONTEXT PRESERVATION:",
                 "  - Do NOT remove valid decision_log entries",
-                "  - Do NOT change milestones unnecessarily",
-                "  - Focus ONLY on addressing the specific failures",
+                "  - Do NOT change milestones outside the flagged classes",
                 "",
                 "CONTEXT.JSON CONTRACT: READ-ONLY.",
                 "  - context.json is owned by the orchestrator",
@@ -130,7 +133,8 @@ def get_step_guidance(
                 "  - Add decision_log entry explaining user confirmation",
                 "  - Or use <needs_user_input> to get confirmation NOW",
                 "",
-                "CONSTRAINT: Fix ONLY the failing items. Don't refactor passing items.",
+                "CONSTRAINT: Fix EVERY instance of the flagged classes (not just the",
+                "named findings). Don't refactor unrelated, passing items.",
             ],
             "next": f"python3 -m {MODULE_PATH} --step 3 --state-dir {state_dir}",
         }
@@ -142,19 +146,28 @@ def get_step_guidance(
                 "VALIDATE your fixes before returning to orchestrator.",
                 "",
                 "Run structural validation:",
-                "  python3 -m skills.planner.cli.plan validate --phase plan-design --state-dir {state_dir}",
+                f"  python3 -m skills.planner.cli.plan validate --phase plan-design --state-dir {state_dir}",
+                "",
+                "Validate planning_context SHAPE (F6 -- self-correct before PASS):",
+                f"  python3 -m skills.planner.cli.plan validate-planning-context --state-dir {state_dir}",
+                "  - constraints must be plain strings (not objects).",
+                "  - each rejected_alternatives entry needs rejection_reason + decision_ref.",
+                "  If it reports <planning_context_errors>, fix the shape and re-run;",
+                "  never hand a malformed planning_context to a downstream step.",
                 "",
                 "SELF-CHECK each fixed item:",
                 "  For each FAIL item you addressed:",
                 "    - Does the fix address the specific finding?",
+                "    - Did you sweep the WHOLE plan for siblings of that class,",
+                "      or only patch the named instance?",
                 "    - Does the fix introduce new issues?",
                 "    - Is the reasoning chain multi-step (not single assertion)?",
                 "",
-                "If validation fails or self-check fails:",
+                "If validation, the planning_context shape check, or self-check fails:",
                 "  - Apply additional fixes",
-                "  - Re-run validation",
+                "  - Re-run both validations",
                 "",
-                "If validation passes:",
+                "If both validations pass:",
                 "  Your complete response must be exactly: PASS",
                 "  Do not add summaries, explanations, or any other text.",
             ],

skills/scripts/skills/planner/cli/plan.py

@@ -634,14 +634,37 @@ def add_arguments(cls, p: argparse.ArgumentParser) -> None:
         p.add_argument("--milestone", required=True, help="Parent milestone ID")
         p.add_argument("--intent-ref", help="Intent ID this implements")
         p.add_argument("--file", help="Changed file path (required for create)")
-        p.add_argument("--diff", help="Diff content (required for create)")
+        p.add_argument("--diff", help="Diff content (required for create unless snippets given)")
+        p.add_argument("--old-snippet", help="F4: exact current text to replace; @@ headers computed by tool")
+        p.add_argument("--new-snippet", help="F4: replacement text; pair with --old-snippet")
         p.add_argument("--comments", help="Change-level comments")
 
+    @staticmethod
+    def _diff_from_snippets(file: str, old_snippet: str, new_snippet: str) -> str:
+        """F4: compute the unified diff deterministically from snippets.
+
+        Reads the current file from cwd when present so the @@ headers anchor to
+        real line numbers; the LLM never authors them.
+        """
+        from ..shared.diffgen import build_unified_diff
+        try:
+            file_text = Path(file).read_text()
+        except (OSError, ValueError):
+            file_text = None  # plan-time: file may not exist / be readable
+        return build_unified_diff(file, old_snippet, new_snippet, file_text)
+
     @classmethod
     def run(cls, args: argparse.Namespace) -> None:
         state_dir = get_state_dir()
         plan = load_plan(state_dir)
 
+        # F4: snippets are the source of truth when provided; tool computes diff.
+        use_snippets = args.old_snippet is not None or args.new_snippet is not None
+        if use_snippets and (args.old_snippet is None or args.new_snippet is None):
+            error_exit("--old-snippet and --new-snippet must be provided together")
+        if use_snippets and args.diff:
+            error_exit("pass snippets OR --diff, not both (diff is computed from snippets)")
+
         ms = plan.get_milestone(args.milestone)
         if not ms:
             ids = [m.id for m in plan.milestones]
@@ -675,7 +698,11 @@ def run(cls, args: argparse.Namespace) -> None:
                 cc.intent_ref = args.intent_ref if args.intent_ref else None
             if args.file:
                 cc.file = args.file
-            if args.diff:
+            if use_snippets:
+                cc.old_snippet = args.old_snippet
+                cc.new_snippet = args.new_snippet
+                cc.diff = cls._diff_from_snippets(cc.file, args.old_snippet, args.new_snippet)
+            elif args.diff:
                 cc.diff = args.diff
             if args.comments is not None:
                 cc.comments = args.comments
@@ -688,10 +715,13 @@ def run(cls, args: argparse.Namespace) -> None:
             # CREATE path
             if args.version is not None:
                 error_exit("--version only valid for updates (when --id provided)")
-            if not args.file or not args.diff:
-                error_exit("--file and --diff required for create")
+            if not args.file or not (args.diff or use_snippets):
+                error_exit("--file and (--diff OR --old-snippet/--new-snippet) required for create")
 
-            diff_content = args.diff
+            if use_snippets:
+                diff_content = cls._diff_from_snippets(args.file, args.old_snippet, args.new_snippet)
+            else:
+                diff_content = args.diff
 
             num = len(ms.code_changes) + 1
             ccid = f"CC-{ms.id}-{num:03d}"
@@ -702,6 +732,8 @@ def run(cls, args: argparse.Namespace) -> None:
                 intent_ref=args.intent_ref,
                 file=args.file,
                 diff=diff_content,
+                old_snippet=args.old_snippet or "" if use_snippets else "",
+                new_snippet=args.new_snippet or "" if use_snippets else "",
                 comments=args.comments or "",
             )
             ms.code_changes.append(cc)
@@ -1221,6 +1253,64 @@ def run(cls, args: argparse.Namespace) -> None:
             success(f"Validation passed for phase {args.phase}")
 
 
+class ValidatePlanningContextCommand(Command):
+    name = "validate-planning-context"
+    help = "F6: validate planning_context shape with self-correctable errors"
+    role = None
+
+    @classmethod
+    def add_arguments(cls, p: argparse.ArgumentParser) -> None:
+        pass
+
+    @classmethod
+    def run(cls, args: argparse.Namespace) -> None:
+        # Read RAW plan.json (not load_plan) so a malformed planning_context
+        # yields friendly, targeted errors instead of a pydantic traceback --
+        # this is the whole point of F6: catch shape drift in the architect
+        # step and feed it back for self-correction.
+        import json
+        from ..shared.schema import validate_planning_context
+
+        state_dir = get_state_dir()
+        plan_raw = json.loads(get_plan_path(state_dir).read_text())
+        errors = validate_planning_context(plan_raw.get("planning_context", {}))
+
+        if errors:
+            print("<planning_context_errors>")
+            for err in errors:
+                print(f"  <error>{err}</error>")
+            print("</planning_context_errors>")
+            sys.exit(1)
+        success("planning_context shape is valid")
+
+
+class TemporalScanCommand(Command):
+    name = "temporal-scan"
+    help = "Deterministic temporal-contamination scan over the doc/comment surface"
+    role = None
+
+    @classmethod
+    def add_arguments(cls, p: argparse.ArgumentParser) -> None:
+        pass
+
+    @classmethod
+    def run(cls, args: argparse.Namespace) -> None:
+        # F2 post-fix gate: read raw plan.json (not the pydantic model) so the
+        # scan stays robust to deprecated/loose doc fields, then drive the
+        # whole temporal class to zero hits. Exit 1 on any hit so the fixer
+        # cannot report PASS while siblings of a named finding remain.
+        import json
+        from ..shared.temporal_detection import scan_plan_docs, format_scan_report
+
+        state_dir = get_state_dir()
+        plan_raw = json.loads(get_plan_path(state_dir).read_text())
+        hits = scan_plan_docs(plan_raw)
+
+        print(format_scan_report(hits))
+        if hits:
+            sys.exit(1)
+
+
 # =============================================================================
 # Commands: List Helpers (read-only, no role restriction)
 # =============================================================================
@@ -1327,6 +1417,8 @@ def run(cls, args: argparse.Namespace) -> None:
     SetDocDiffCommand,
     CreateDocChangeCommand,
     ValidateCommand,
+    ValidatePlanningContextCommand,
+    TemporalScanCommand,
     ListMilestonesCommand,
     ListIntentsCommand,
     ListChangesCommand,

skills/scripts/skills/planner/cli/qr.py

@@ -197,6 +197,10 @@ def cmd_update_item(state_dir: str, phase: str, args: list[str]):
         # Version increments on status change
         item["version"] = item.get("version", 1) + 1
         item["status"] = status
+        # F5: track per-item fix<->verify fail count so the convergence guard
+        # can stop an item looping more than the cap. Increment only on FAIL.
+        if status == "FAIL":
+            item["fail_count"] = item.get("fail_count", 0) + 1
         if finding:
             item["finding"] = finding
         elif "finding" in item and status == "PASS":
@@ -341,12 +345,59 @@ def cmd_assign_group(state_dir: str, phase: str, args: list[str]):
     ))
 
 
+def cmd_accept_item(state_dir: str, phase: str, args: list[str]):
+    """Override-accept a FAIL item with a recorded rationale (F5).
+
+    Used to resolve the loop-convergence escalation gate: a MUST item that has
+    failed past the per-item cap is accepted (accepted=True + acceptance_reason)
+    so it stops blocking the workflow, with the human's reason on record.
+    """
+    if not args:
+        error_exit("Usage: accept-item <id> --reason <text>")
+
+    item_id = args[0]
+    reason = None
+    i = 1
+    while i < len(args):
+        if args[i] == "--reason" and i + 1 < len(args):
+            reason = args[i + 1]
+            i += 2
+        else:
+            i += 1
+
+    if not reason:
+        error_exit("--reason required: record why this item is accepted despite failing")
+
+    qr_path = get_qr_path(state_dir, phase)
+    if not qr_path.exists():
+        error_exit(f"QR state file not found: {qr_path}")
+
+    with open(qr_path, "r+") as f:
+        fcntl.flock(f.fileno(), fcntl.LOCK_EX)
+        qr_state = load_qr_state_locked(f)
+        idx, item = find_item(qr_state, item_id)
+        if idx < 0:
+            error_exit(f"Item {item_id} not found in qr-{phase}.json")
+        item["accepted"] = True
+        item["acceptance_reason"] = reason
+        item["version"] = item.get("version", 1) + 1
+        qr_state["items"][idx] = item
+        save_qr_state_atomic(state_dir, phase, qr_state)
+
+    print_entity_result(EntityResult(
+        id=item_id,
+        version=item["version"],
+        operation="accepted"
+    ))
+
+
 COMMANDS = {
     "update-item": cmd_update_item,
     "get-item": cmd_get_item,
     "list-items": cmd_list_items,
     "summary": cmd_summary,
     "assign-group": cmd_assign_group,
+    "accept-item": cmd_accept_item,
 }