`8b65eda`

Filter reference verification commands

Authored by mfwolffe <wolffemf@dukes.jmu.edu> 1 week ago

SHA: 8b65edab3e1a3c3365ec7b1d070398052d310d17
Parents: add1107
Tree: 0a61056

3 changed files

Status	File	+	-
M	`src/loader/runtime/dod.py`	72	0
M	`src/loader/runtime/finalization.py`	25	7
M	`tests/test_finalization.py`	78	0

src/loader/runtime/dod.pymodified

      return commands
 +def sanitize_verification_commands(
 +    commands: list[str],
 +    *,
 +    dod: DefinitionOfDone,
 +    project_root: Path,
 +) -> list[str]:
 +    """Drop verification commands that reopen reference paths outside the planned output roots."""
++
 +    if not commands:
 +        return []
++
 +    planned_targets = collect_planned_artifact_targets(
 +        dod,
 +        project_root=project_root,
 +    )
 +    if not planned_targets:
 +        return list(dict.fromkeys(command for command in commands if command))
++
 +    allowed_roots = {
 +        str(target if expect_directory else target.parent)
 +        for target, expect_directory in planned_targets
 +    }
 +    sanitized: list[str] = []
 +    for command in commands:
 +        normalized = str(command or "").strip()
 +        if not normalized:
 +            continue
 +        detected_paths = _extract_verification_command_paths(normalized)
 +        if detected_paths and any(
 +            not _path_is_within_allowed_verification_roots(path, allowed_roots)
 +            for path in detected_paths
 +        ):
 +            continue
 +        _append_unique(sanitized, normalized)
 +    return sanitized
++
++
  def build_verification_summary(evidence: list[VerificationEvidence]) -> str:
      """Create a short evidence summary for the final user response."""
          return False
 +def _extract_verification_command_paths(command: str) -> list[Path]:
 +    try:
 +        parts = shlex.split(command)
 +    except ValueError:
 +        return []
++
 +    detected: list[Path] = []
 +    for token in parts:
 +        candidate = str(token or "").strip().strip("\"'")
 +        if not candidate or not candidate.startswith(("/", "~")):
 +            continue
 +        try:
 +            path = Path(candidate).expanduser()
 +        except (OSError, RuntimeError, ValueError):
 +            continue
 +        detected.append(path)
 +    return detected
++
++
 +def _path_is_within_allowed_verification_roots(path: Path, allowed_roots: set[str]) -> bool:
 +    try:
 +        candidate = path.expanduser().resolve(strict=False)
 +    except (OSError, RuntimeError, ValueError):
 +        candidate = path.expanduser()
++
 +    for raw_root in allowed_roots:
 +        try:
 +            root = Path(raw_root).expanduser().resolve(strict=False)
 +        except (OSError, RuntimeError, ValueError):
 +            root = Path(raw_root).expanduser()
 +        if candidate == root or str(candidate).startswith(f"{root}/"):
 +            return True
 +    return False
++
++
  def synthesize_todo_items(dod: DefinitionOfDone) -> list[dict[str, str]]:
      """Build a todo item list from the current DoD state.

src/loader/runtime/finalization.pymodified

      derive_verification_commands,
      ensure_active_verification_attempt,
      planned_artifact_target_satisfied,
 +    sanitize_verification_commands,
      synthesize_todo_items,
+ )
  from .events import AgentEvent, TurnSummary
              and dod.verification_plan
              and Path(dod.verification_plan).exists()
          ):
 -            dod.verification_commands = extract_verification_commands_from_markdown(
 -                Path(dod.verification_plan).read_text()
 +            dod.verification_commands = sanitize_verification_commands(
 +                extract_verification_commands_from_markdown(
 +                    Path(dod.verification_plan).read_text()
 +                ),
 +                dod=dod,
 +                project_root=self.context.project_root,
+             )
          if (
              and dod.implementation_plan
              and Path(dod.implementation_plan).exists()
          ):
 -            dod.verification_commands = extract_verification_commands_from_markdown(
 -                Path(dod.implementation_plan).read_text()
 +            dod.verification_commands = sanitize_verification_commands(
 +                extract_verification_commands_from_markdown(
 +                    Path(dod.implementation_plan).read_text()
 +                ),
 +                dod=dod,
 +                project_root=self.context.project_root,
+             )
          if not dod.verification_commands:
 -            dod.verification_commands = derive_verification_commands(
 -                dod,
 +            dod.verification_commands = sanitize_verification_commands(
 +                derive_verification_commands(
 +                    dod,
 +                    project_root=self.context.project_root,
 +                    task_statement=dod.task_statement,
 +                ),
 +                dod=dod,
                  project_root=self.context.project_root,
 -                task_statement=dod.task_statement,
+             )
          else:
              for command in derive_verification_commands(
              ):
                  if command not in dod.verification_commands:
                      dod.verification_commands.append(command)
 +            dod.verification_commands = sanitize_verification_commands(
 +                dod.verification_commands,
 +                dod=dod,
 +                project_root=self.context.project_root,
 +            )
          await self.set_workflow_mode(
              ModeDecision.transition(

tests/test_finalization.pymodified

+     ]
 +@pytest.mark.asyncio
 +async def test_turn_finalizer_filters_reference_side_verification_commands(
 +    temp_dir: Path,
 +) -> None:
 +    guide_root = temp_dir / "Loader" / "guides" / "nginx"
 +    chapters = guide_root / "chapters"
 +    chapters.mkdir(parents=True)
 +    index_path = guide_root / "index.html"
 +    chapter_one = chapters / "01-introduction.html"
 +    index_path.write_text("<html><body><h1>Guide</h1></body></html>\n")
 +    chapter_one.write_text("<html><body><h1>Intro</h1></body></html>\n")
++
 +    reference_root = temp_dir / "Loader" / "guides" / "fortran"
 +    reference_root.mkdir(parents=True)
++
 +    implementation_plan = temp_dir / "implementation.md"
 +    implementation_plan.write_text(
 +        "\n".join(
 +            [
 +                "# Implementation Plan",
 +                "",
 +                "## File Changes",
 +                f"- `{guide_root}`",
 +                f"- `{chapters}`",
 +                f"- `{index_path}`",
 +                f"- `{chapter_one}`",
 +                "",
 +            ]
 +        )
 +    )
 +    verification_plan = temp_dir / "verification.md"
 +    verification_plan.write_text(
 +        "\n".join(
 +            [
 +                "# Verification Plan",
 +                "",
 +                "## Verification Commands",
 +                "```bash",
 +                f"ls -la {guide_root}",
 +                f"ls -la {reference_root}",
 +                "```",
 +                "",
 +            ]
 +        )
 +    )
++
 +    session = FakeSession()
 +    context = build_context(temp_dir, session)
 +    finalizer = TurnFinalizer(
 +        context,
 +        RuntimeTracer(),
 +        DefinitionOfDoneStore(temp_dir),
 +        set_workflow_mode=_noop_set_workflow_mode,
 +    )
 +    dod = create_definition_of_done("Create an nginx guide from an external reference.")
 +    dod.mutating_actions.append("write")
 +    dod.touched_files.extend([str(index_path), str(chapter_one)])
 +    dod.implementation_plan = str(implementation_plan)
 +    dod.verification_plan = str(verification_plan)
 +    summary = TurnSummary(final_response="")
 +    executor = RecordingExecutor()
++
 +    async def capture(event) -> None:
 +        return None
++
 +    result = await finalizer.run_definition_of_done_gate(
 +        dod=dod,
 +        candidate_response="Created the nginx guide.",
 +        emit=capture,
 +        summary=summary,
 +        executor=executor,  # type: ignore[arg-type]
 +    )
++
 +    assert result.should_continue is False
 +    assert any(str(guide_root) in command for command in executor.commands)
 +    assert all(str(reference_root) not in command for command in executor.commands)
++
++
  @pytest.mark.asyncio
  async def test_turn_finalizer_blocks_completion_when_planned_artifacts_are_missing(
      temp_dir: Path,