`5b0f1ee`

Tighten post-build guide convergence

Authored by

espadonne 2 weeks ago

SHA: 5b0f1ee901534e254c02ad9640b9ff17dae4311d
Parents: df5639d
Tree: 3089c60

4 changed files

Status	File	+
M	`src/loader/runtime/dod.py`	108
M	`src/loader/runtime/hooks.py`	58
M	`tests/test_dod.py`	38
M	`tests/test_permissions.py`	69

src/loader/runtime/dod.pymodified

      if html_link_command:
          _append_unique(commands, html_link_command)
 +    html_quality_command = _derive_multi_page_html_quality_command(
 +        dod,
 +        project_root=project_root,
 +        task_statement=task_statement,
 +    )
 +    if html_quality_command:
 +        _append_unique(commands, html_quality_command)
      for command in _build_planned_artifact_verification_commands(planned_artifact_targets):
          _append_unique(commands, command)
      return _build_local_html_link_verification_command(resolved_paths)
 +def _derive_multi_page_html_quality_command(
 +    dod: DefinitionOfDone,
 +    *,
 +    project_root: Path,
 +    task_statement: str,
 +) -> str | None:
 +    html_paths = _multi_page_html_quality_paths(dod, project_root=project_root)
 +    if len(html_paths) < 4:
 +        return None
 +    if not _task_requires_substantive_html_guide_quality(task_statement):
 +        return None
++
 +    path_literals = ", ".join(repr(str(path)) for path in html_paths)
 +    return "\n".join(
 +        [
 +            "python3 - <<'PY'",
 +            "from pathlib import Path",
 +            "import re",
 +            "",
 +            f"paths = [{path_literals}]",
 +            "tag_pattern = re.compile(r'<[^>]+>')",
 +            "content_block_pattern = re.compile(r'<(p|li|pre|code|section|article|table|h2|h3|h4)\\b', re.IGNORECASE)",
 +            "issues = []",
 +            "checked = 0",
 +            "for raw_path in paths:",
 +            "    path = Path(raw_path)",
 +            "    if not path.exists():",
 +            "        continue",
 +            "    checked += 1",
 +            "    text = path.read_text()",
 +            "    plain = tag_pattern.sub(' ', text)",
 +            "    plain = re.sub(r'\\s+', ' ', plain).strip()",
 +            "    content_blocks = len(content_block_pattern.findall(text))",
 +            "    has_h1 = bool(re.search(r'<h1\\b', text, re.IGNORECASE))",
 +            "    minimum_chars = 180 if path.name.lower() == 'index.html' else 220",
 +            "    minimum_blocks = 2 if path.name.lower() == 'index.html' else 3",
 +            "    if not has_h1:",
 +            "        issues.append(f'{path}: missing <h1>')",
 +            "    if len(plain) < minimum_chars:",
 +            "        issues.append(",
 +            "            f'{path}: thin content ({len(plain)} text chars, expected at least {minimum_chars})'",
 +            "        )",
 +            "    if content_blocks < minimum_blocks:",
 +            "        issues.append(",
 +            "            f'{path}: insufficient structured content ({content_blocks} blocks, expected at least {minimum_blocks})'",
 +            "        )",
 +            "if issues:",
 +            "    print('HTML guide content quality issues:')",
 +            "    print('\\n'.join(issues))",
 +            "    raise SystemExit(1)",
 +            "print(f'Checked HTML guide content quality across {checked} file(s).')",
 +            "PY",
 +        ]
 +    )
++
++
  def collect_planned_artifact_targets(
      dod: DefinitionOfDone,
      *,
      return commands
 +def _multi_page_html_quality_paths(
 +    dod: DefinitionOfDone,
 +    *,
 +    project_root: Path,
 +) -> list[Path]:
 +    planned_targets = collect_planned_artifact_targets(
 +        dod,
 +        project_root=project_root,
 +        max_paths=24,
 +    )
 +    planned_html = [
 +        target
 +        for target, expect_directory in planned_targets
 +        if not expect_directory and target.suffix.lower() in {".html", ".htm"}
 +    ]
 +    if planned_html:
 +        return planned_html
++
 +    touched_html = []
 +    for path_str in dod.touched_files:
 +        path = Path(path_str)
 +        effective_path = path if path.is_absolute() else (project_root / path)
 +        if effective_path.suffix.lower() in {".html", ".htm"}:
 +            touched_html.append(effective_path)
 +    return list(dict.fromkeys(touched_html))
++
++
 +def _task_requires_substantive_html_guide_quality(task_statement: str) -> bool:
 +    lowered = task_statement.lower()
 +    if not any(token in lowered for token in ("guide", "tutorial", "documentation", "docs")):
 +        return False
 +    return any(
 +        token in lowered
 +        for token in (
 +            "thorough",
 +            "comprehensive",
 +            "detailed",
 +            "equally",
 +            "cadence",
 +            "chapter",
 +            "chapters",
 +        )
 +    )
++
++
  def _extract_markdown_section_lines(markdown: str, heading: str) -> list[str]:
      current_heading: str | None = None
      collected: list[str] = []

src/loader/runtime/hooks.pymodified

      """Block reopening old reference paths once planned artifacts are well underway."""
      _MIN_COMPLETED_FILES = 3
 +    _MAX_COMPLETED_SCOPE_OBSERVATIONS = 4
      def __init__(self, *, dod_store: DefinitionOfDoneStore, project_root: Path, session: Any) -> None:
          self.dod_store = dod_store
          self.project_root = project_root
          self.session = session
 +        self._completed_scope_key: tuple[str, ...] | None = None
 +        self._completed_scope_observation_count = 0
      async def pre_tool_use(self, context: HookContext) -> HookResult:
          if context.tool_call.name not in _OBSERVATION_TOOLS:
              if not observed_paths:
                  return HookResult()
              if all(path_within_allowed_roots(path, completed_scope) for path in observed_paths):
 +                self._sync_completed_scope_state(completed_scope)
 +                if (
 +                    context.source != "verification"
 +                    and self._completed_scope_observation_count
 +                    >= self._MAX_COMPLETED_SCOPE_OBSERVATIONS
 +                ):
 +                    roots_preview = ", ".join(f"`{root}`" for root in completed_scope[:2])
 +                    if len(completed_scope) > 2:
 +                        roots_preview += ", ..."
 +                    return HookResult(
 +                        decision=HookDecision.DENY,
 +                        message=(
 +                            "[Blocked - post-build audit loop: all explicitly planned artifacts "
 +                            "already exist and the current output set has already been inspected "
 +                            "several times.] Suggestion: move to verification now or make one "
 +                            "concrete edit for a specific mismatch inside "
 +                            f"{roots_preview} instead of more rereads."
 +                        ),
 +                        terminal_state="blocked",
 +                    )
                  return HookResult()
              roots_preview = ", ".join(f"`{root}`" for root in completed_scope[:2])
              return None
          return missing_label, tuple(planned_roots)
 +    async def post_tool_use(self, context: HookContext) -> HookResult:
 +        if context.tool_call.name in _MUTATION_TOOLS:
 +            self._reset_completed_scope_state()
 +            return HookResult()
 +        if context.tool_call.name not in _OBSERVATION_TOOLS:
 +            return HookResult()
 +        if context.source == "verification":
 +            return HookResult()
++
 +        completed_scope = self._completed_artifact_scope()
 +        if completed_scope is None:
 +            self._reset_completed_scope_state()
 +            return HookResult()
++
 +        observed_paths = _extract_observation_paths(context.tool_call)
 +        if not observed_paths:
 +            return HookResult()
 +        if not all(path_within_allowed_roots(path, completed_scope) for path in observed_paths):
 +            return HookResult()
++
 +        self._sync_completed_scope_state(completed_scope)
 +        self._completed_scope_observation_count += 1
 +        return HookResult()
++
      def _completed_artifact_scope(self) -> tuple[str, ...] | None:
          dod_path = getattr(self.session, "active_dod_path", None)
          if not dod_path:
              planned_roots.append(root)
          return tuple(planned_roots)
 +    def _sync_completed_scope_state(self, completed_scope: tuple[str, ...]) -> None:
 +        normalized = tuple(sorted(completed_scope))
 +        if self._completed_scope_key == normalized:
 +            return
 +        self._completed_scope_key = normalized
 +        self._completed_scope_observation_count = 0
++
 +    def _reset_completed_scope_state(self) -> None:
 +        self._completed_scope_key = None
 +        self._completed_scope_observation_count = 0
++
  class HookManager:
      """Runs tool hooks across Loader's three lifecycle events."""

tests/test_dod.pymodified

      assert f"test -d {tmp_path / 'docs/chapters'}" in commands
 +def test_derive_verification_commands_adds_html_guide_quality_check_for_thorough_guides(
 +    tmp_path: Path,
 +) -> None:
 +    docs = tmp_path / "docs"
 +    chapters = docs / "chapters"
 +    chapters.mkdir(parents=True)
 +    implementation_plan = tmp_path / "implementation.md"
 +    implementation_plan.write_text(
 +        "\n".join(
 +            [
 +                "# Implementation Plan",
 +                "",
 +                "## File Changes",
 +                f"- `{docs / 'index.html'}`",
 +                f"- `{chapters / '01-introduction.html'}`",
 +                f"- `{chapters / '02-installation.html'}`",
 +                f"- `{chapters / '03-configuration.html'}`",
 +                f"- `{chapters / '04-troubleshooting.html'}`",
 +                "",
 +            ]
 +        )
 +    )
++
 +    dod = create_definition_of_done(
 +        "Create an equally thorough multi-page HTML guide with chapter files."
 +    )
 +    dod.implementation_plan = str(implementation_plan)
++
 +    commands = derive_verification_commands(
 +        dod,
 +        project_root=tmp_path,
 +        task_statement=dod.task_statement,
 +        supplement_existing=True,
 +    )
++
 +    assert any("HTML guide content quality issues:" in command for command in commands)
++
++
  def test_collect_planned_artifact_targets_ignores_prose_path_fragments_in_refreshed_plan(
      tmp_path: Path,
  ) -> None:

tests/test_permissions.pymodified

      assert str(temp_dir / "guide") in result.message
 +@pytest.mark.asyncio
 +async def test_late_reference_drift_hook_blocks_excessive_post_build_self_audits(
 +    temp_dir: Path,
 +) -> None:
 +    registry = create_default_registry(temp_dir)
 +    policy = build_permission_policy(
 +        active_mode=PermissionMode.WORKSPACE_WRITE,
 +        workspace_root=temp_dir,
 +        tool_requirements=registry.get_tool_requirements(),
 +    )
 +    dod_store = DefinitionOfDoneStore(temp_dir)
 +    dod = create_definition_of_done("Create a multi-file guide from a reference")
 +    dod.status = "in_progress"
 +    plan_path = temp_dir / "implementation.md"
 +    plan_path.write_text(
 +        "\n".join(
 +            [
 +                "# Implementation Plan",
 +                "",
 +                "## File Changes",
 +                f"- `{temp_dir / 'guide' / 'index.html'}`",
 +                f"- `{temp_dir / 'guide' / 'chapters' / '01-getting-started.html'}`",
 +                f"- `{temp_dir / 'guide' / 'chapters' / '02-installation.html'}`",
 +                "",
 +            ]
 +        )
 +    )
 +    dod.implementation_plan = str(plan_path)
 +    guide_dir = temp_dir / "guide" / "chapters"
 +    guide_dir.mkdir(parents=True, exist_ok=True)
 +    target = guide_dir / "02-installation.html"
 +    (temp_dir / "guide" / "index.html").write_text("<h1>Nginx Guide</h1>\n")
 +    (guide_dir / "01-getting-started.html").write_text("<h1>Getting Started</h1>\n")
 +    target.write_text("<h1>Installation</h1>\n")
 +    dod_path = dod_store.save(dod)
 +    session = FakeSession(active_dod_path=str(dod_path), messages=[])
 +    hook = LateReferenceDriftHook(
 +        dod_store=dod_store,
 +        project_root=temp_dir,
 +        session=session,
 +    )
++
 +    def make_context(index: int) -> HookContext:
 +        return HookContext(
 +            tool_call=ToolCall(
 +                id=f"read-{index}",
 +                name="read",
 +                arguments={"file_path": str(target)},
 +            ),
 +            tool=registry.get("read"),
 +            registry=registry,
 +            permission_policy=policy,
 +            source="native",
 +        )
++
 +    for index in range(1, 5):
 +        context = make_context(index)
 +        result = await hook.pre_tool_use(context)
 +        assert result.decision == HookDecision.CONTINUE
 +        await hook.post_tool_use(context)
++
 +    blocked = await hook.pre_tool_use(make_context(5))
++
 +    assert blocked.decision == HookDecision.DENY
 +    assert blocked.terminal_state == "blocked"
 +    assert blocked.message is not None
 +    assert "post-build audit loop" in blocked.message
++
++
  @pytest.mark.asyncio
  async def test_late_reference_drift_hook_does_not_treat_empty_output_dir_as_complete_artifact_set(
      temp_dir: Path,