`eabd30b`

Track repair progress in verification retries

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

SHA: eabd30b108d131d2a72a451ead95f1091df9df34
Parents: 549ed74
Tree: 305ee5c

4 changed files

Status	File	+	-
M	`src/loader/runtime/dod.py`	43	5
M	`src/loader/runtime/finalization.py`	40	0
M	`tests/test_dod.py`	31	0
M	`tests/test_finalization.py`	45	0

src/loader/runtime/dod.pymodified

  from typing import Any, Literal
  from ..llm.base import Message, ToolCall
 +from ..tools.fs_safety import StructuredPatchHunk, coerce_structured_patch_payload
  from ..tools.shell_tools import BashTool
  from .verification_observations import VerificationAttempt, verification_attempt_id
      storage_path: str | None = None
      last_verification_result: str | None = None
      last_verification_signature: str | None = None
 +    last_failed_verification_issue_signature: str | None = None
      verification_attempt_counter: int = 0
      active_verification_attempt_id: str | None = None
      active_verification_attempt_number: int | None = None
              storage_path=data.get("storage_path"),
              last_verification_result=data.get("last_verification_result"),
              last_verification_signature=data.get("last_verification_signature"),
 +            last_failed_verification_issue_signature=data.get(
 +                "last_failed_verification_issue_signature"
 +            ),
              verification_attempt_counter=int(data.get("verification_attempt_counter", 0)),
              active_verification_attempt_id=data.get("active_verification_attempt_id"),
              active_verification_attempt_number=(
          file_path = _resolve_touched_path(tool_call.arguments.get("file_path", ""))
          if file_path:
              _append_unique(dod.touched_files, file_path)
 -        for hunk in tool_call.arguments.get("hunks", []):
 -            if not isinstance(hunk, dict):
 -                continue
 -            old_lines = int(hunk.get("old_lines", 0))
 -            new_lines = int(hunk.get("new_lines", 0))
 +        for hunk in _coerce_patch_hunks_for_accounting(tool_call.arguments):
 +            old_lines, new_lines = _patch_hunk_line_counts(hunk)
              dod.line_changes += max(old_lines, new_lines)
      elif tool_call.name == "bash":
          command = str(tool_call.arguments.get("command", "")).strip()
      return content.count("\n") + 1
 +def _coerce_patch_hunks_for_accounting(
 +    arguments: dict[str, object],
 +) -> list[dict[str, object] | StructuredPatchHunk]:
 +    for key in ("hunks", "structured_patch", "structuredPatch"):
 +        hunks = coerce_structured_patch_payload(arguments.get(key))
 +        if hunks:
 +            return hunks
 +    return []
++
++
 +def _patch_hunk_line_counts(
 +    hunk: dict[str, object] | StructuredPatchHunk,
 +) -> tuple[int, int]:
 +    if isinstance(hunk, StructuredPatchHunk):
 +        return hunk.old_lines, hunk.new_lines
++
 +    old_lines = _coerce_int(hunk.get("old_lines"), default=0)
 +    new_lines_value = hunk.get("new_lines")
 +    if isinstance(new_lines_value, list):
 +        new_lines = len(new_lines_value)
 +        old_start = _coerce_int(hunk.get("old_start"), default=0)
 +        old_end = _coerce_int(hunk.get("old_end"), default=old_start - 1)
 +        if old_lines <= 0 and old_start > 0:
 +            old_lines = max(0, old_end - old_start + 1)
 +    else:
 +        new_lines = _coerce_int(new_lines_value, default=0)
 +    return max(0, old_lines), max(0, new_lines)
++
++
 +def _coerce_int(value: object, *, default: int) -> int:
 +    try:
 +        return int(value)  # type: ignore[arg-type]
 +    except (TypeError, ValueError):
 +        return default
++
++
  def _is_verification_command(command: str) -> bool:
      command_lower = command.lower()
      signals = (

src/loader/runtime/finalization.pymodified

              dod.pending_items = []
              dod.status = "done"
              dod.last_verification_result = "passed"
 +            dod.last_failed_verification_issue_signature = None
 +            dod.retry_count = 0
              dod.confidence = "high"
              summary.verification_status = "passed"
              summary.definition_of_done = dod
+             )
          dod.last_verification_result = "failed"
 +        failure_issue_signature = _verification_failure_issue_signature(dod)
 +        if _verification_failure_issue_changed(dod, failure_issue_signature):
 +            dod.retry_count = 0
 +        if failure_issue_signature:
 +            dod.last_failed_verification_issue_signature = failure_issue_signature
          summary.verification_status = "failed"
          summary.definition_of_done = dod
          failed_provenance = _verification_result_provenance(dod, passed=False)
+     )
 +def _verification_failure_issue_signature(dod: DefinitionOfDone) -> str:
 +    parts: list[str] = []
 +    for evidence in dod.evidence:
 +        if evidence.passed or evidence.skipped:
 +            continue
 +        detail = "\n".join(
 +            str(value).strip()
 +            for value in (
 +                evidence.command,
 +                evidence.stdout,
 +                evidence.stderr,
 +                evidence.output,
 +            )
 +            if str(value).strip()
 +        )
 +        normalized_lines = [
 +            " ".join(line.strip().split())
 +            for line in detail.splitlines()
 +            if line.strip()
 +        ]
 +        if normalized_lines:
 +            parts.append("\n".join(normalized_lines))
 +    return "\n---\n".join(sorted(parts))
++
++
 +def _verification_failure_issue_changed(
 +    dod: DefinitionOfDone,
 +    failure_issue_signature: str,
 +) -> bool:
 +    previous = dod.last_failed_verification_issue_signature
 +    return bool(previous and failure_issue_signature and previous != failure_issue_signature)
++
++
  def _normalize_pending_statement(value: str) -> str:
      return " ".join(value.strip().lower().split())

tests/test_dod.pymodified

  """Tests for definition-of-done state and persistence."""
 +import json
  import subprocess
  from pathlib import Path
      assert dod.touched_files == [str(absolute_path)]
 +def test_record_successful_tool_call_counts_json_string_patch_hunks(
 +    tmp_path: Path,
 +) -> None:
 +    dod = create_definition_of_done("Patch generated HTML content.")
 +    target = tmp_path / "chapter.html"
 +    hunks = json.dumps(
 +        [
 +            {
 +                "old_start": 10,
 +                "old_lines": 2,
 +                "new_start": 10,
 +                "new_lines": 8,
 +                "lines": ["-old", "-body", "+new", "+expanded"],
 +            }
 +        ]
 +    )[:-1]
++
 +    record_successful_tool_call(
 +        dod,
 +        ToolCall(
 +            id="patch-1",
 +            name="patch",
 +            arguments={"file_path": str(target), "hunks": hunks},
 +        ),
 +    )
++
 +    assert dod.touched_files == [str(target)]
 +    assert dod.line_changes == 8
++
++
  def test_derive_verification_commands_adds_semantic_html_toc_check(tmp_path: Path) -> None:
      chapters = tmp_path / "chapters"
      chapters.mkdir()

tests/test_finalization.pymodified

      assert session.messages[-1].content.startswith("[DEFINITION OF DONE CHECK STILL FAILING]")
 +@pytest.mark.asyncio
 +async def test_turn_finalizer_extends_retry_budget_when_failures_change(
 +    temp_dir: Path,
 +) -> None:
 +    session = FakeSession()
 +    context = build_context(temp_dir, session)
 +    finalizer = TurnFinalizer(
 +        context,
 +        RuntimeTracer(),
 +        DefinitionOfDoneStore(temp_dir),
 +        set_workflow_mode=_noop_set_workflow_mode,
 +    )
 +    target = temp_dir / "chapter.html"
 +    target.write_text("<h1>Chapter</h1>\n")
 +    dod = create_definition_of_done("Expand the generated chapter.")
 +    dod.retry_count = dod.retry_budget
 +    dod.mutating_actions.append("patch")
 +    dod.touched_files.append(str(target))
 +    dod.line_changes = 20
 +    dod.last_verification_result = "failed"
 +    dod.last_verification_signature = "lines=10;touched=chapter.html;actions=1;commands="
 +    dod.last_failed_verification_issue_signature = "old failing artifact set"
 +    dod.verification_commands = ["python check_quality.py"]
 +    summary = TurnSummary(final_response="")
 +    executor = SelectiveRecordingExecutor("check_quality.py")
++
 +    async def capture(event) -> None:
 +        return None
++
 +    result = await finalizer.run_definition_of_done_gate(
 +        dod=dod,
 +        candidate_response="I expanded one failing file.",
 +        emit=capture,
 +        summary=summary,
 +        executor=executor,  # type: ignore[arg-type]
 +    )
++
 +    assert result.should_continue is True
 +    assert result.reason_code == "verification_failed_reentry"
 +    assert dod.retry_count == 1
 +    assert dod.status == "fixing"
 +    assert "python check_quality.py" in executor.commands
 +    assert session.messages[-1].content.startswith("[DEFINITION OF DONE CHECK FAILED]")
++
++
  @pytest.mark.asyncio
  async def test_turn_finalizer_accepts_missing_optional_html5validator_when_semantic_check_passes(
      temp_dir: Path,