`6925565`

Recover blocked HTML quality writes

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

SHA: 6925565bebf9c73a3c3c4da83c439f38d4f8be85
Parents: 2c94bf4
Tree: fa6e89a

2 changed files

Status	File	+	-
M	`src/loader/runtime/tool_batches.py`	103	3
M	`tests/test_tool_batches.py`	101	0

src/loader/runtime/tool_batches.pymodified

                      tool_call,
                      outcome.event_content,
+                 )
 +                self._queue_blocked_html_content_quality_nudge(
 +                    tool_call,
 +                    outcome.event_content,
 +                    dod=dod,
 +                )
                  self._queue_blocked_active_repair_nudge(outcome.event_content)
                  self._queue_blocked_active_repair_mutation_nudge(outcome.event_content)
                  self._queue_blocked_completed_artifact_scope_nudge(
              "not claim completion until the blocked file write succeeds."
+         )
 +    def _queue_blocked_html_content_quality_nudge(
 +        self,
 +        tool_call: ToolCall,
 +        event_content: str,
 +        *,
 +        dod: DefinitionOfDone,
 +    ) -> None:
 +        """Keep blocked HTML chapter-quality retries on the same concrete target."""
++
 +        if tool_call.name not in {"write", "edit", "patch"}:
 +            return
 +        if (
 +            "HTML content contains placeholder or stub text" not in event_content
 +            and "HTML guide chapter content is too thin" not in event_content
 +        ):
 +            return
++
 +        target = str(
 +            tool_call.arguments.get("file_path")
 +            or tool_call.arguments.get("path")
 +            or ""
 +        ).strip()
 +        if not target:
 +            return
++
 +        quality_notes: list[str] = []
 +        placeholder_phrases = _extract_blocked_html_target_list(
 +            event_content,
 +            "Placeholder phrase(s):",
 +        )
 +        if placeholder_phrases:
 +            quality_notes.append(
 +                "Do not reuse placeholder pattern(s): "
 +                + ", ".join(f"`{phrase}`" for phrase in placeholder_phrases[:4])
 +                + "."
 +            )
 +        thin_match = re.search(
 +            r"Current content has ([^.]+?); expected at least ([^.]+?)\.",
 +            event_content,
 +        )
 +        if thin_match:
 +            quality_notes.append(
 +                f"The blocked draft had {thin_match.group(1)}; the floor is {thin_match.group(2)}."
 +            )
++
 +        missing_artifact = _next_missing_planned_artifact(
 +            dod,
 +            project_root=self.context.project_root,
 +            messages=list(getattr(self.context.session, "messages", []) or []),
 +        )
 +        missing_suffix = ""
 +        if missing_artifact is not None:
 +            missing_target, _ = missing_artifact
 +            target_path = Path(target).expanduser().resolve(strict=False)
 +            if target_path == missing_target.expanduser().resolve(strict=False):
 +                missing_suffix = (
 +                    " "
 +                    + _missing_artifact_resume_suffix(
 +                        missing_artifact,
 +                        project_root=self.context.project_root,
 +                        messages=list(getattr(self.context.session, "messages", []) or []),
 +                    ).strip()
 +                )
++
 +        quality_detail = (
 +            " " + " ".join(quality_notes)
 +            if quality_notes
 +            else ""
 +        )
 +        self.context.queue_steering_message(
 +            f"The last HTML mutation for `{target}` was blocked, so the file was "
 +            "not created or updated. Retry that same target with one concrete "
 +            "`write`, `edit`, or `patch` call containing finished user-facing HTML, "
 +            "not a scaffold or outline."
 +            f"{quality_detail}"
 +            " Include specific explanations, commands/configuration examples, "
 +            "lists, and troubleshooting or operational details as appropriate for "
 +            "the artifact."
 +            f"{missing_suffix}"
 +            " Do not switch to a different sibling file, do not claim completion, "
 +            "and do not reopen unrelated reference materials before this blocked "
 +            "mutation succeeds."
 +        )
++
      def _queue_blocked_invalid_mutation_nudge(
          self,
          tool_call: ToolCall,
      """Return whether a blocked observation should steer without tripping fatal error limits."""
      normalized = str(event_content or "")
 -    return (
 -        "[Blocked - completed artifact set scope:" in normalized
 -        or "[Blocked - post-build audit loop:" in normalized
 -    )
 +    recoverable_markers = (
 +        "[Blocked - completed artifact set scope:",
 +        "[Blocked - post-build audit loop:",
 +        "[Blocked - active repair scope:",
 +        "[Blocked - active repair mutation scope:",
 +        "[Blocked - late reference drift:",
 +        "[Blocked - missing planned output artifact:",
 +        "[Blocked - HTML file creation falls outside the current declared artifact set]",
 +        "[Blocked - HTML page introduces new local targets outside the current declared artifact set]",
 +        "[Blocked - HTML local asset references do not exist]",
 +        "[Blocked - HTML content contains placeholder or stub text]",
 +        "[Blocked - HTML guide chapter content is too thin]",
 +        "[Blocked - Edited HTML links point to files that do not exist]",
 +    )
 +    return any(marker in normalized for marker in recoverable_markers)
  def _recent_edit_string_mismatch_target(recovery_context: RecoveryContext | None) -> str:

tests/test_tool_batches.pymodified

      assert "update the guide root" not in queued[0]
 +@pytest.mark.asyncio
 +async def test_tool_batch_runner_blocked_html_quality_guidance_does_not_halt(
 +    temp_dir: Path,
 +) -> None:
 +    async def assess_confidence(
 +        tool_name: str,
 +        tool_args: dict,
 +        context: str,
 +    ) -> ConfidenceAssessment:
 +        raise AssertionError("Confidence scoring should not run 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")
++
 +    target = temp_dir / "guide" / "chapters" / "06-security.html"
 +    implementation_plan = temp_dir / "implementation.md"
 +    implementation_plan.write_text(
 +        "\n".join(
 +            [
 +                "# Implementation Plan",
 +                "",
 +                "## File Changes",
 +                f"- `{target}`",
 +                "",
 +            ]
 +        )
 +    )
++
 +    context = build_context(
 +        temp_dir=temp_dir,
 +        messages=[],
 +        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 guide chapter.")
 +    dod.implementation_plan = str(implementation_plan)
++
 +    tool_calls = [
 +        ToolCall(
 +            id=f"write-quality-{index}",
 +            name="write",
 +            arguments={"file_path": str(target), "content": "<html></html>"},
 +        )
 +        for index in range(3)
 +    ]
 +    blocked_message = (
 +        "[Blocked - HTML content contains placeholder or stub text] "
 +        "Suggestion: Replace placeholder phrases with concrete user-facing content "
 +        "before writing the HTML artifact. Placeholder phrase(s): generic core "
 +        "concepts section, generic practical workflow section. Include specific "
 +        "explanations, examples, commands, or structured prose instead."
 +    )
 +    executor = FakeExecutor(
 +        [
 +            tool_outcome(
 +                tool_call=tool_call,
 +                output=blocked_message,
 +                is_error=True,
 +                state=ToolExecutionState.BLOCKED,
 +            )
 +            for tool_call in tool_calls
 +        ]
 +    )
 +    events: list[AgentEvent] = []
++
 +    async def emit(event: AgentEvent) -> None:
 +        events.append(event)
++
 +    result = await runner.execute_batch(
 +        tool_calls=tool_calls,
 +        tool_source="native",
 +        pending_tool_calls_seen=set(),
 +        emit=emit,
 +        summary=TurnSummary(final_response=""),
 +        dod=dod,
 +        executor=executor,
 +        on_confirmation=None,
 +        on_user_question=None,
 +        emit_confirmation=None,
 +        consecutive_errors=0,
 +    )
++
 +    assert result.halted is False
 +    assert result.consecutive_errors == 0
 +    assert queued
 +    assert str(target) in queued[-1]
 +    assert "Retry that same target" in queued[-1]
 +    assert "Do not reuse placeholder pattern(s)" in queued[-1]
 +    assert "generic core concepts section" in queued[-1]
 +    assert "not a scaffold or outline" in queued[-1]
++
++
  def test_tool_batch_runner_blocked_html_missing_target_after_outputs_exist_prefers_verify(
      temp_dir: Path,
  ) -> None: