Fix plan-part contract binding, replan feedback, and E2E evidence

scripts/e2e/README.md

@@ -202,6 +202,15 @@ and are often `0600`, so a second container run copies `out/` and `data/logs/` i
 fails if any secret survives. Only `upload/` is uploaded, and only when redaction
 succeeded.
 
+When a test does not pass, the harness preserves plans from each pipeline clone under
+`out/pipeline-artifacts/<clone>/<make-plan|rectify>/` and Claude Code JSONL under
+`out/transcripts/`, retaining relative paths. The transcripts include the orchestrator,
+skill sessions and their subagents; the orchestrator's transcript holds each raw
+`bind_plan_set` response and its `plan_parts` argument. These files go through the same
+redaction and upload steps as the other output artifacts. Collection also runs for an
+expected failure or a harness exception. A container killed before or during collection
+may leave no evidence or only partial evidence.
+
 `clean-install` mounts no home directories and retains evidence under `out/`. Its redactor
 receives the same required `--secret-env` names with variables unset and skips the absent
 home-log tree. Redaction and upload still run after test failures.
@@ -330,6 +339,25 @@ docker run --rm \
   python3 /opt/e2e/e2e_harness.py run --test clean-install --out /artifacts
 ```
 
+### Check a real plan against the plan-set gate before dispatching a pipeline scenario
+
+After the plan-set changes have merged, perform this check before CI commissioning:
+
+1. Check out the #5255 branch revision containing both the `implementation` E2E scenario
+   and the evidence collector. Build and run that revision with the Docker
+   invocation above, selecting `--test implementation`. Use the same revision for
+   commissioning and record the model.
+2. If the run reaches implementation, the first bind succeeded. If the test does not
+   pass, inspect the orchestrator's `bind_plan_set` results in `out/transcripts/`.
+   An observed `success: true` on the first or second bind is the pass condition.
+   A `coverage_failed` result explains the rejection in `coverage_gaps`, and
+   `out/pipeline-artifacts/` retains the rejected plan. Missing evidence is inconclusive
+   and does not satisfy this gate: require an observed successful bind. When a run
+   replans, also confirm the second bind's `plan_parts` names only the replacement plan.
+3. Record the model, plan path and bind result in issue #5259's body using
+   `gh issue edit --body-file`, never as a comment. Only an observed successful bind
+   unblocks `gh workflow run e2e.yml -f tests=implementation`.
+
 ### Sandbox smoke with teardown
 
 Set `MINIMAX_API_KEY` and `E2E_SANDBOX_TOKEN` in the local environment first. Build the same

scripts/e2e/e2e_harness.py

@@ -5,6 +5,7 @@
 
 import argparse
 import ast
+import fnmatch
 import json
 import os
 import re
@@ -35,6 +36,7 @@
 REDACTED = b"[REDACTED]"
 CONFIG_TEMPLATE = Path(__file__).with_name("autoskillit-config.yaml")
 SANDBOX_CLONE = Path("/workspace/sandbox")
+PIPELINE_RUNS_DIR = "autoskillit-runs"
 
 _SCRUBBED_ENV = frozenset({"CLAUDE_CODE_OAUTH_TOKEN", *SECRET_ENV})
 _STDERR_TAIL_CHARS = 2000
@@ -1017,6 +1019,67 @@ def _run_test(
     return run_recipe(test, catalog.sandbox_repository, token, out, scrubbed, runner)
 
 
+def _copy_failure_files(
+    source: Path, dest: Path, boundary: Path, pattern: str, errors: list[str]
+) -> int:
+    relative = source.relative_to(boundary)
+    try:
+        if any(
+            boundary.joinpath(*relative.parts[:depth]).is_symlink()
+            for depth in range(len(relative.parts) + 1)
+        ):
+            return 0
+        if not source.exists():
+            return 0
+    except OSError as exc:
+        errors.append(f"{source}: {exc}")
+        return 0
+    copied = 0
+    for root, dirs, files in os.walk(
+        source, followlinks=False, onerror=lambda exc: errors.append(f"{source}: {exc}")
+    ):
+        dirs.sort()
+        for name in sorted(fnmatch.filter(files, pattern)):
+            path = Path(root) / name
+            try:
+                if path.is_symlink() or not path.is_file():
+                    continue
+                target = dest / path.relative_to(source)
+                target.parent.mkdir(parents=True, exist_ok=True)
+                shutil.copyfile(path, target)
+                copied += 1
+            except OSError as exc:
+                errors.append(f"{path}: {exc}")
+    return copied
+
+
+def collect_failure_evidence(out: Path, home: Path) -> dict[str, Any]:
+    """Keep pipeline plans and Claude transcripts for an unsuccessful E2E run."""
+    errors: list[str] = []
+    runs = SANDBOX_CLONE.parent / PIPELINE_RUNS_DIR
+    try:
+        clones = [] if runs.is_symlink() else sorted(runs.iterdir())
+    except FileNotFoundError:
+        clones = []
+    except OSError as exc:
+        errors.append(f"{runs}: {exc}")
+        clones = []
+    plan_files = 0
+    for clone in clones:
+        for kind in ("make-plan", "rectify"):
+            plan_files += _copy_failure_files(
+                clone / ".autoskillit" / "temp" / kind,
+                out / "pipeline-artifacts" / clone.name / kind,
+                clone,
+                "*",
+                errors,
+            )
+    transcripts = _copy_failure_files(
+        home / ".claude" / "projects", out / "transcripts", home, "*.jsonl", errors
+    )
+    return {"plan_files": plan_files, "transcripts": transcripts, "errors": errors}
+
+
 def run_test(
     test: e2e_catalog.CatalogTest,
     catalog: e2e_catalog.Catalog,
@@ -1043,6 +1106,18 @@ def run_test(
         result.update(
             outcome=outcome, passed=outcome == "passed", expected_findings=matched_findings
         )
+    if not result["passed"]:
+        try:
+            evidence = collect_failure_evidence(out, home)
+        except Exception as exc:
+            evidence = {
+                "plan_files": 0,
+                "transcripts": 0,
+                "errors": [f"{type(exc).__name__}: {exc}"],
+                "exception": traceback.format_exc(),
+            }
+        if any(evidence.values()):
+            result["evidence"] = evidence
     (out / "result.json").write_text(json.dumps(result, indent=2) + "\n", encoding="utf-8")
     return failures
 

src/autoskillit/core/io/markdown_sections.py

@@ -18,7 +18,9 @@ def extract_section(markdown: str, heading: str) -> str:
     pattern = re.compile(rf"^## {re.escape(heading)}[ \t]*$", re.MULTILINE)
     matches = tuple(pattern.finditer(markdown))
     if len(matches) != 1:
-        raise ValueError(f"expected exactly one ## {heading} section")
+        raise ValueError(
+            f"expected exactly one ## {heading} section; found {len(matches)} matching headings"
+        )
     start = matches[0].end()
     next_heading = re.search(r"^##\s+", markdown[start:], re.MULTILINE)
     end = start + next_heading.start() if next_heading is not None else len(markdown)
@@ -29,7 +31,7 @@ def split_table_row(line: str) -> tuple[str, ...]:
     """Parse one strictly pipe-delimited row."""
     stripped = line.strip()
     if not stripped.startswith("|") or not stripped.endswith("|"):
-        raise ValueError("table rows must be pipe-delimited")
+        raise ValueError(f'table rows must be pipe-delimited; offending line: "{stripped[:120]}"')
     return tuple(cell.strip() for cell in stripped[1:-1].split("|"))
 
 
@@ -41,15 +43,24 @@ def parse_pipe_table(
     lines = tuple(line for line in section.splitlines() if line.strip())
     header = tuple(expected_header)
     if len(lines) < 2:
-        raise ValueError("table must contain a header and separator")
+        line = lines[0].strip()[:120] if lines else ""
+        raise ValueError(f'table must contain a header and separator; offending line: "{line}"')
     if split_table_row(lines[0]) != header:
-        raise ValueError(f"table header must be {header!r}")
+        raise ValueError(
+            f'table header must be {header!r}; offending line: "{lines[0].strip()[:120]}"'
+        )
     separator = split_table_row(lines[1])
     if len(separator) != len(header) or any(
         re.fullmatch(r":?-{3,}:?", cell) is None for cell in separator
     ):
-        raise ValueError("table separator is invalid")
-    rows = tuple(split_table_row(line) for line in lines[2:])
-    if any(len(row) != len(header) for row in rows):
-        raise ValueError(f"table rows must have exactly {len(header)} columns")
-    return rows
+        raise ValueError(f'table separator is invalid; offending line: "{lines[1].strip()[:120]}"')
+    rows = []
+    for line in lines[2:]:
+        row = split_table_row(line)
+        if len(row) != len(header):
+            raise ValueError(
+                f"table rows must have exactly {len(header)} columns; "
+                f'offending line: "{line.strip()[:120]}"'
+            )
+        rows.append(row)
+    return tuple(rows)

src/autoskillit/core/planset/allocation.py

@@ -21,16 +21,29 @@ def parse_part_allocation(part_markdown: str, *, part_key: str) -> tuple[Allocat
     rows = parse_pipe_table(section, _HEADER)
     result: list[AllocationRowDef] = []
     seen: set[str] = set()
-    for requirement_id, allocation, implementation_step in rows:
+    raw_rows = tuple(line for line in section.splitlines() if line.strip())[2:]
+    for (requirement_id, allocation, implementation_step), raw_row in zip(
+        rows, raw_rows, strict=True
+    ):
+        if not requirement_id:
+            raise ValueError(
+                f'Requirement ID is empty in row "{raw_row.strip()[:120]}"; '
+                "use a non-empty bare Requirement ID"
+            )
         if requirement_id in seen:
             raise ValueError(f"duplicate allocated requirement {requirement_id}")
         seen.add(requirement_id)
         if not implementation_step:
-            raise ValueError("allocation implementation step is empty")
+            raise ValueError(
+                f"{requirement_id}: allocation implementation step is empty; "
+                "use one matching heading key such as Step 2.1"
+            )
         try:
             kind = AllocationKind(allocation.lower())
         except ValueError as exc:
-            raise ValueError(f"unknown allocation {allocation!r}") from exc
+            raise ValueError(
+                f'{requirement_id}: unknown allocation "{allocation}"; use owned or shared'
+            ) from exc
         result.append(
             AllocationRowDef(
                 requirement_id=requirement_id,
@@ -45,7 +58,7 @@ def parse_part_allocation(part_markdown: str, *, part_key: str) -> tuple[Allocat
 def verify_allocation_evidence(
     part_markdown: str,
     rows: tuple[AllocationRowDef, ...],
-) -> tuple[tuple[str, PlanSetRejectReason], ...]:
+) -> tuple[tuple[str, PlanSetRejectReason, str], ...]:
     """Ensure an allocation names a real step whose own text names the requirement."""
     matches = tuple(STEP_HEADING_RE.finditer(part_markdown))
     blocks: dict[str, str] = {}
@@ -57,11 +70,30 @@ def verify_allocation_evidence(
         )
         end = min(next_step, section_end)
         blocks[match.group(1)] = part_markdown[match.start() : end]
-    errors: list[tuple[str, PlanSetRejectReason]] = []
+    errors: list[tuple[str, PlanSetRejectReason, str]] = []
     for row in rows:
         block = blocks.get(row.implementation_step)
         if block is None:
-            errors.append((row.requirement_id, PlanSetRejectReason.ALLOCATION_STEP_MISSING))
+            found = (
+                f"step keys found: {', '.join(blocks)}" if blocks else "no step heading was found"
+            )
+            errors.append(
+                (
+                    row.requirement_id,
+                    PlanSetRejectReason.ALLOCATION_STEP_MISSING,
+                    f'{row.requirement_id}: Implementation Step cell "{row.implementation_step}" '
+                    f"does not match a step; {found}. Expected exactly one key such as Step 2.1, "
+                    "copied from a ### Step N: title heading (### Step N.M: title for sub-steps).",
+                )
+            )
         elif row.requirement_id not in block:
-            errors.append((row.requirement_id, PlanSetRejectReason.ALLOCATION_STEP_UNREFERENCED))
+            errors.append(
+                (
+                    row.requirement_id,
+                    PlanSetRejectReason.ALLOCATION_STEP_UNREFERENCED,
+                    f'Requirement ID cell "{row.requirement_id}" is not referenced in '
+                    f"{row.implementation_step}; the bare ID must appear in that step's own "
+                    "text, on the line under its heading.",
+                )
+            )
     return tuple(errors)

src/autoskillit/core/types/audit/_type_plan_set_authority.py

@@ -440,12 +440,21 @@ def _validate_collections(self) -> None:
         object.__setattr__(self, "unparsed_marker_lines", tuple(self.unparsed_marker_lines))
         if not isinstance(self.coverage, CoverageResultDef):
             raise ValueError("PlanSetAuthority.coverage is invalid")
+        part_keys = {part.part_key for part in self.parts}
+        if not set(self.coverage.parts_without_obligations) <= part_keys:
+            raise ValueError("coverage.parts_without_obligations contains an unknown part key")
         if self.state is PlanSetState.SEALED and self.coverage.status is not CoverageStatus.PASS:
             raise ValueError("sealed authority requires passing coverage")
         if tuple(part.ordinal for part in self.parts) != tuple(range(1, len(self.parts) + 1)):
             raise ValueError("PlanSetAuthority.parts must have consecutive ordinals")
         if len({Path(part.locator).resolve() for part in self.parts}) != len(self.parts):
             raise ValueError("PlanSetAuthority.parts contain duplicate locators")
+        if self.inventory_mode in {InventoryMode.UNENUMERATED, InventoryMode.NO_ISSUE}:
+            empty_parts = part_keys - {row.part_key for row in self.allocations}
+            if set(self.coverage.parts_without_obligations) != empty_parts:
+                raise ValueError(
+                    "coverage.parts_without_obligations must match actual empty parts"
+                )
 
     @classmethod
     def create(
@@ -622,6 +631,14 @@ class PlanSetBindResult:
     first_part_path: str = ""
     coverage_gaps: str = ""
 
+    def __post_init__(self) -> None:
+        if self.success != (self.reason is None):
+            raise ValueError("successful results require no reason; failures require a reason")
+        if self.reason is not PlanSetRejectReason.COVERAGE_FAILED and self.coverage_gaps:
+            raise ValueError("only coverage_failed results may contain coverage_gaps")
+        if self.reason is PlanSetRejectReason.COVERAGE_FAILED and not self.coverage_gaps.strip():
+            raise ValueError("coverage_failed requires non-empty coverage_gaps")
+
     def to_dict(self) -> dict[str, Any]:
         result = {name: getattr(self, name) for name in self.__dataclass_fields__}
         result["reason"] = self.reason.value if self.reason else ""

src/autoskillit/recipe/api_orchestration/_api_orchestration_match.py

@@ -17,6 +17,7 @@
     _LoadPipelineInputs,
     _ValidationResult,
 )
+from autoskillit.recipe.helpers._io_loading import _substitute_recipe_notes_in_yaml
 from autoskillit.recipe.io import (
     RecipeInfo,
     find_recipe_by_name,
@@ -45,6 +46,7 @@ def _resolve_recipe_match(
     raw_declared = match.content if match.content is not None else match.path.read_text()
     raw = substitute_temp_placeholder_yaml_safe(raw_declared, pipeline_inputs.temp_dir_relpath)
     raw = substitute_scripts_placeholder_yaml_safe(raw)
+    raw = _substitute_recipe_notes_in_yaml(raw)
 
     if match.source == RecipeSource.BUILTIN:
         recipes_dir = _orch.pkg_root() / "recipes"