`07dbbb9`

Steer duplicate repair writes

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

SHA: 07dbbb9e97edcb0b44a7e8fe558e29bfbc6f327a
Parents: fdbe57a
Tree: 98b6131

4 changed files

Status	File	+	-
M	`src/loader/runtime/repair_focus.py`	19	5
M	`src/loader/runtime/tool_batches.py`	57	0
M	`tests/test_repair_focus.py`	26	0
M	`tests/test_tool_batches.py`	60	0

src/loader/runtime/repair_focus.pymodified

              if artifact_path:
                  if artifact_path not in absolute_paths:
                      absolute_paths.insert(0, artifact_path)
 -            allowed_paths = tuple(
 -                sorted(
 -                    absolute_paths,
 -                    key=lambda item: (not Path(item).exists(), item),
 -                )
 +            allowed_paths = _ordered_allowed_paths(
 +                absolute_paths,
 +                primary_path=artifact_path,
+             )
              allowed_roots = _collapse_roots(_path_roots(set(absolute_paths)))
              return ActiveRepairContext(
              continue
          collapsed.append(root)
      return tuple(collapsed)
++
++
 +def _ordered_allowed_paths(paths: list[str], *, primary_path: str) -> tuple[str, ...]:
 +    """Preserve repair-focus order with the immediate target first."""
++
 +    ordered: list[str] = []
++
 +    def add(path: str) -> None:
 +        if not path or path in ordered:
 +            return
 +        ordered.append(path)
++
 +    add(primary_path)
 +    for path in paths:
 +        add(path)
 +    return tuple(ordered)

src/loader/runtime/tool_batches.pymodified

              self.context.session.append(outcome.message)
              summary.tool_result_messages.append(outcome.message)
              if outcome.state == ToolExecutionState.DUPLICATE:
 +                self._queue_duplicate_mutation_nudge(tool_call, dod=dod)
                  self._queue_duplicate_observation_nudge(tool_call, dod=dod)
              elif outcome.state == ToolExecutionState.BLOCKED:
                  self._queue_blocked_invalid_mutation_nudge(
              "Choose a different next step that makes progress."
+         )
 +    def _queue_duplicate_mutation_nudge(
 +        self,
 +        tool_call: ToolCall,
 +        *,
 +        dod: DefinitionOfDone,
 +    ) -> None:
 +        """After a duplicate mutation, restate concrete repair deltas."""
++
 +        if tool_call.name not in {"write", "edit", "patch"}:
 +            return
++
 +        target = str(
 +            tool_call.arguments.get("file_path")
 +            or tool_call.arguments.get("path")
 +            or ""
 +        ).strip()
 +        repair = extract_active_repair_context(self.context.session.messages)
 +        if repair is not None:
 +            repair_preview = _active_repair_focus_preview(repair.repair_lines)
 +            target_label = f"`{target}`" if target else "that file"
 +            self.context.queue_steering_message(
 +                f"That {tool_call.name} was skipped because it would not change {target_label}. "
 +                "Do not submit the same content again. "
 +                f"Verification still requires these concrete repair deltas: {repair_preview} "
 +                "Use the current generated file as the source of truth and make one real edit, "
 +                "patch, or write that expands or changes the flagged artifact."
 +            )
 +            return
++
 +        if all_planned_artifact_outputs_exist(dod, project_root=self.context.project_root):
 +            target_label = f"`{target}`" if target else "the target file"
 +            self.context.queue_steering_message(
 +                f"That {tool_call.name} was skipped because it would not change {target_label}. "
 +                "All explicitly planned artifacts already exist, so do not rewrite the same content. "
 +                "If verification identified a mismatch, make a different concrete mutation that fixes it; "
 +                "otherwise finish so Loader can verify the files already on disk."
 +            )
++
      def _queue_post_mutation_self_audit_nudge(
          self,
          tool_call: ToolCall,
+     )
 +def _active_repair_focus_preview(repair_lines: list[str], *, max_lines: int = 4) -> str:
 +    """Compact repair-focus bullets for steering after no-op mutations."""
++
 +    preview: list[str] = []
 +    for raw_line in repair_lines:
 +        line = str(raw_line or "").strip()
 +        if not line.startswith("- "):
 +            continue
 +        if line.startswith("- Immediate next step:"):
 +            continue
 +        preview.append(line[2:].strip())
 +        if len(preview) >= max_lines:
 +            break
 +    if not preview:
 +        return "the active verifier repair focus"
 +    return "; ".join(preview)
++
++
  def _tool_call_label(tool_call: ToolCall) -> str:
      """Human-readable label for one tool call."""
      name = tool_call.name

tests/test_repair_focus.pymodified

      assert context is not None
      assert context.artifact_path == str(repair_target.resolve(strict=False))
      assert str(repair_target.resolve(strict=False)) in context.allowed_paths
++
++
 +def test_extract_active_repair_context_keeps_immediate_target_first(
 +    tmp_path: Path,
 +) -> None:
 +    index_path = tmp_path / "guides" / "nginx" / "index.html"
 +    chapter_path = tmp_path / "guides" / "nginx" / "chapters" / "02-installation.html"
++
 +    context = extract_active_repair_context(
 +        [
 +            Message(
 +                role=Role.USER,
 +                content=(
 +                    "Repair focus:\n"
 +                    f"- Improve `{chapter_path}`: thin content (526 text chars, expected at least 1758).\n"
 +                    f"- Immediate next step: edit `{index_path}`.\n"
 +                    f"- Improve `{index_path}`: insufficient structured content (9 blocks, expected at least 12).\n"
 +                ),
 +            )
 +        ]
 +    )
++
 +    assert context is not None
 +    assert context.artifact_path == str(index_path.resolve(strict=False))
 +    assert context.allowed_paths[0] == str(index_path.resolve(strict=False))
 +    assert str(chapter_path.resolve(strict=False)) in context.allowed_paths

tests/test_tool_batches.pymodified

      assert "before widening the change set" in queued[0]
 +def test_tool_batch_runner_duplicate_repair_mutation_restates_verifier_deltas(
 +    temp_dir: Path,
 +) -> None:
 +    async def assess_confidence(
 +        tool_name: str,
 +        tool_args: dict,
 +        context: str,
 +    ) -> ConfidenceAssessment:
 +        raise AssertionError("Confidence scoring should be disabled in this scenario")
++
 +    async def verify_action(
 +        tool_name: str,
 +        tool_args: dict,
 +        result: str,
 +        expected: str = "",
 +    ) -> ActionVerification:
 +        raise AssertionError("Verification should not run in this scenario")
++
 +    index_path = temp_dir / "guide" / "index.html"
 +    chapter_path = temp_dir / "guide" / "chapters" / "02-installation.html"
 +    context = build_context(
 +        temp_dir=temp_dir,
 +        messages=[
 +            Message(
 +                role=Role.USER,
 +                content=(
 +                    "Repair focus:\n"
 +                    f"- Improve `{index_path}`: insufficient structured content (9 blocks, expected at least 12).\n"
 +                    f"- Improve `{chapter_path}`: thin content (526 text chars, expected at least 1758).\n"
 +                    f"- Immediate next step: edit `{index_path}`.\n"
 +                    "- Update the listed generated artifacts directly; do not recreate the artifact set.\n"
 +                ),
 +            )
 +        ],
 +        safeguards=FakeSafeguards(),
 +        assess_confidence=assess_confidence,
 +        verify_action=verify_action,
 +    )
 +    queued: list[str] = []
 +    context.queue_steering_message_callback = queued.append
 +    runner = ToolBatchRunner(context, DefinitionOfDoneStore(temp_dir))
 +    dod = create_definition_of_done("Create a multi-file guide.")
++
 +    runner._queue_duplicate_mutation_nudge(  # type: ignore[attr-defined]
 +        ToolCall(
 +            id="dup-write",
 +            name="write",
 +            arguments={"file_path": str(index_path), "content": "<h1>same</h1>"},
 +        ),
 +        dod=dod,
 +    )
++
 +    assert queued
 +    assert "skipped because it would not change" in queued[0]
 +    assert "Do not submit the same content again" in queued[0]
 +    assert "insufficient structured content" in queued[0]
 +    assert "thin content" in queued[0]
 +    assert "make one real edit" in queued[0]
++
++
  @pytest.mark.asyncio
  async def test_tool_batch_runner_hands_off_after_active_repair_support_file_write(
      temp_dir: Path,