`e82c3e2`

Normalize home-rooted path prompts

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

SHA: e82c3e2841eee6899eea021a770304dc3566e908
Parents: ab475b4
Tree: 475a1d4

5 changed files

Status	File	+	-
A	`src/loader/runtime/path_display.py`	25	0
M	`src/loader/runtime/repair.py`	30	21
M	`src/loader/runtime/tool_batches.py`	11	8
M	`tests/test_repair.py`	75	12
M	`tests/test_tool_batches.py`	109	2

src/loader/runtime/path_display.pyadded

 +"""Helpers for consistent user-facing runtime path rendering."""
++
 +from __future__ import annotations
++
 +from pathlib import Path
++
++
 +def display_runtime_path(path_value: str | Path) -> str:
 +    """Render a stable user-facing path for prompts and steering messages.
++
 +    Paths under the active HOME are shown as `~/...` to avoid leaking
 +    platform-specific symlink expansions such as `/private/tmp/...`.
 +    Other paths keep the current normalized absolute rendering.
 +    """
++
 +    path = Path(path_value).expanduser()
 +    resolved = path.resolve(strict=False)
 +    home = Path.home().expanduser().resolve(strict=False)
 +    try:
 +        relative = resolved.relative_to(home)
 +    except ValueError:
 +        return str(resolved)
 +    if not relative.parts:
 +        return "~"
 +    return f"~/{relative.as_posix()}"

src/loader/runtime/repair.pymodified

      planned_artifact_target_satisfied,
+ )
  from .parsing import parse_tool_calls
 +from .path_display import display_runtime_path
  from .recovery import detect_missing_mutation_payload
  from .workflow import (
      infer_output_outline_label,
          target = fix["file_path"] or self._preferred_retry_target(dod)
          invalid = ", ".join(f"`{field}`" for field in fix["invalid_fields"])
 +        display_target = display_runtime_path(target) if target else None
          if fix.get("kind") == "missing_target":
              if attempt.tool_name == "write":
                  target_line = (
 -                    f"Last tool failure: resend `write` for `{target}` with a valid `file_path` and real `content`."
 -                    if target
 +                    f"Last tool failure: resend `write` for `{display_target}` with a valid `file_path` and real `content`."
 +                    if display_target
                      else "Last tool failure: resend `write` with a valid `file_path` and real `content`."
+                 )
                  return [
+                 ]
              if attempt.tool_name == "edit":
                  target_line = (
 -                    f"Last tool failure: resend `edit` for `{target}` with a valid `file_path` plus real `old_string`/`new_string`."
 -                    if target
 +                    f"Last tool failure: resend `edit` for `{display_target}` with a valid `file_path` plus real `old_string`/`new_string`."
 +                    if display_target
                      else "Last tool failure: resend `edit` with a valid `file_path` plus real `old_string`/`new_string`."
+                 )
                  return [
+                 ]
              if attempt.tool_name == "patch":
                  target_line = (
 -                    f"Last tool failure: resend `patch` for `{target}` with a valid `file_path` and real patch text or `hunks`."
 -                    if target
 +                    f"Last tool failure: resend `patch` for `{display_target}` with a valid `file_path` and real patch text or `hunks`."
 +                    if display_target
                      else "Last tool failure: resend `patch` with a valid `file_path` and real patch text or `hunks`."
+                 )
                  return [
          if attempt.tool_name == "write":
              lines = [
+                 (
 -                    f"Last tool failure: resend `write` for `{target}` with real `content`, not just summary fields."
 -                    if target
 +                    f"Last tool failure: resend `write` for `{display_target}` with real `content`, not just summary fields."
 +                    if display_target
                      else "Last tool failure: resend `write` with real `content`, not just summary fields."
                  ),
+             ]
          if attempt.tool_name == "edit":
              lines = [
+                 (
 -                    f"Last tool failure: resend `edit` for `{target}` with the real text payload."
 -                    if target
 +                    f"Last tool failure: resend `edit` for `{display_target}` with the real text payload."
 +                    if display_target
                      else "Last tool failure: resend `edit` with the real text payload."
                  ),
                  f"Do not use {invalid} in place of `old_string`/`new_string`.",
          if attempt.tool_name == "patch":
              lines = [
+                 (
 -                    f"Last tool failure: resend `patch` for `{target}` with real patch text or structured hunks."
 -                    if target
 +                    f"Last tool failure: resend `patch` for `{display_target}` with real patch text or structured hunks."
 +                    if display_target
                      else "Last tool failure: resend `patch` with real patch text or structured hunks."
                  ),
                  f"Do not use {invalid} in place of the real patch payload.",
              lines = [
                  f"Resume with this exact next step: create `{concrete_target.name}`.",
                  f"It is the next concrete output needed to continue `{next_pending}`.",
 -                f"Prefer one `write(content=...)` call for `{concrete_target}` before more research.",
 +                "Prefer one `write(content=...)` call for "
 +                f"`{display_runtime_path(concrete_target)}` before more research.",
                  self._mutation_tool_scaffold(
                      concrete_target,
                      tool_name="write",
+             ]
              if inferred_is_directory:
                  lines.append(
 -                    f"Prefer one concrete directory-creation step for `{inferred_pending_target}` before more research."
 +                    "Prefer one concrete directory-creation step for "
 +                    f"`{display_runtime_path(inferred_pending_target)}` before more research."
+                 )
                  lines.append(
                      self._directory_creation_scaffold(inferred_pending_target)
+                 )
              else:
                  lines.append(
 -                    f"Prefer one `write(content=...)` call for `{inferred_pending_target}` before more research."
 +                    "Prefer one `write(content=...)` call for "
 +                    f"`{display_runtime_path(inferred_pending_target)}` before more research."
+                 )
                  lines.append(
                      self._mutation_tool_scaffold(
+                         )
+                     )
                      lines.append(
 -                        f"Prefer one `write` call for `{next_output_file}` before more research."
 +                        "Prefer one `write` call for "
 +                        f"`{display_runtime_path(next_output_file)}` before more research."
+                     )
                      lines.append(
                          self._mutation_tool_scaffold(
                          f"under {label}."
+                     ]
                  lines.append(
 -                    f"Prefer one concrete `write` call for a file inside `{target}` before more research."
 +                    "Prefer one concrete `write` call for a file inside "
 +                    f"`{display_runtime_path(target)}` before more research."
+                 )
              else:
                  lines = [f"Resume with this exact next step: create {label}."]
              if expect_directory and not target.is_dir():
                  lines.append(
 -                    f"Prefer one concrete directory-creation step for `{target}` before more research."
 +                    "Prefer one concrete directory-creation step for "
 +                    f"`{display_runtime_path(target)}` before more research."
+                 )
              elif not expect_directory:
                  lines.append(
 -                    f"Prefer one `write` call for `{target}` before any more reference reads."
 +                    "Prefer one `write` call for "
 +                    f"`{display_runtime_path(target)}` before any more reference reads."
+                 )
                  if not target.parent.exists():
                      lines.append(
      @staticmethod
      def _mutation_tool_scaffold(path: Path, *, tool_name: str) -> str:
 -        normalized_path = json.dumps(str(path.expanduser().resolve(strict=False)))
 +        normalized_path = json.dumps(display_runtime_path(path))
          if tool_name == "edit":
              signature = (
                  f"edit(file_path={normalized_path}, old_string=\"...\", "
      @staticmethod
      def _directory_creation_scaffold(path: Path) -> str:
 -        command = f"mkdir -p {shlex.quote(str(path.expanduser().resolve(strict=False)))}"
 +        command = f"mkdir -p {shlex.quote(display_runtime_path(path))}"
          return f"Emit this tool shape now: `bash(command={json.dumps(command)})`."

src/loader/runtime/tool_batches.pymodified

  from .evidence_provenance import EvidenceProvenance, EvidenceProvenanceStatus
  from .executor import ToolExecutionState, ToolExecutor
  from .logging import get_runtime_logger
 +from .path_display import display_runtime_path
  from .policy_timeline import append_verification_timeline_entry
  from .recovery import RecoveryContext, detect_missing_mutation_payload
  from .repair_focus import extract_active_repair_context
      allow_inferred_child: bool = True,
  ) -> str:
      label = target.name or str(target)
 +    display_target = display_runtime_path(target)
      if expect_directory and not label.endswith("/"):
          label += "/"
      if expect_directory:
                  guidance = (
                      f" Resume by creating `{next_output_file.name}` now. {guidance_origin} "
                      f"Prefer one `write` call for "
 -                    f"`{next_output_file}` instead of more rereads."
 +                    f"`{display_runtime_path(next_output_file)}` instead of more rereads."
+                 )
                  if not next_output_file.parent.exists():
                      guidance += (
          if target.is_dir():
              return (
                  f" Resume by creating the next output file under `{label}` now. Prefer one "
 -                f"concrete `write` call for a file inside `{target}` instead of more rereads."
 +                f"concrete `write` call for a file inside `{display_target}` instead of more rereads."
                  " Make your next response the concrete mutation tool call itself, not another"
                  " bookkeeping-only turn."
+             )
          return (
              f" Resume by creating `{label}` now. Prefer one concrete directory-creation "
 -            f"step for `{target}` instead of more rereads."
 +            f"step for `{display_target}` instead of more rereads."
+         )
      guidance = (
 -        f" Resume by creating `{label}` now. Prefer one `write` call for `{target}` "
 +        f" Resume by creating `{label}` now. Prefer one `write` call for `{display_target}` "
          "instead of more rereads."
+     )
      if not target.parent.exists():
      target, expect_directory = missing_artifact
      label = target.name or str(target)
 +    display_target = display_runtime_path(target)
      if expect_directory and not label.endswith("/"):
          label += "/"
      if expect_directory:
              if target.is_dir():
                  return (
                      f"Next step: create the next output file under `{label}`. Prefer one "
 -                    f"concrete `write` call inside `{target}` now."
 +                    f"concrete `write` call inside `{display_target}` now."
+                 )
              return (
                  f"Next step: create `{label}`. Prefer one concrete directory-creation step "
 -                f"for `{target}` now."
 +                f"for `{display_target}` now."
+             )
          guidance = (
              f"Next step: create `{next_output_file.name}`. Prefer one "
 -            f"`write(file_path=..., content=...)` call for `{next_output_file}` now."
 +            f"`write(file_path=..., content=...)` call for `{display_runtime_path(next_output_file)}` now."
+         )
          if not next_output_file.parent.exists():
              guidance += (
      guidance = (
          f"Next step: create `{label}`. Prefer one "
 -        f"`write(file_path=..., content=...)` call for `{target}` now."
 +        f"`write(file_path=..., content=...)` call for `{display_target}` now."
+     )
      if not target.parent.exists():
          guidance += (

tests/test_repair.pymodified

  from pathlib import Path
  from types import SimpleNamespace
 +import pytest
++
  from loader.llm.base import Message, Role, ToolCall
  from loader.runtime.context import RuntimeContext
  from loader.runtime.dod import create_definition_of_done
 +from loader.runtime.path_display import display_runtime_path
  from loader.runtime.permissions import (
      PermissionMode,
      build_permission_policy,
          in decision.retry_message
+     )
      assert (
 -        f"Prefer one `write` call for `{index_path}` before any more reference reads."
 +        "Prefer one `write` call for "
 +        f"`{display_runtime_path(index_path)}` before any more reference reads."
          in decision.retry_message
+     )
      assert (
          in decision.retry_message
+     )
      assert (
 -        f'Emit this tool shape now: `write(file_path="{index_path.resolve(strict=False)}", content="...")`.'
 +        f'Emit this tool shape now: `write(file_path="{display_runtime_path(index_path)}", content="...")`.'
          in decision.retry_message
+     )
      assert "Do not restart discovery unless one specific missing fact blocks this step." in decision.retry_message
          in decision.retry_message
+     )
      assert (
 -        f"Prefer one concrete directory-creation step for `{chapters_path}` before more research."
 +        "Prefer one concrete directory-creation step for "
 +        f"`{display_runtime_path(chapters_path)}` before more research."
          in decision.retry_message
+     )
 -    expected_command = f"mkdir -p {chapters_path.resolve(strict=False)}"
 +    expected_command = f"mkdir -p {display_runtime_path(chapters_path)}"
      assert (
          f'Emit this tool shape now: `bash(command="{expected_command}")`.'
          in decision.retry_message
+     )
 -    assert f'write(file_path="{chapters_path.resolve(strict=False)}"' not in decision.retry_message
 +    assert f'write(file_path="{display_runtime_path(chapters_path)}"' not in decision.retry_message
++
++
 +def test_empty_response_retry_uses_home_relative_path_for_home_artifacts(
 +    temp_dir: Path,
 +    monkeypatch: pytest.MonkeyPatch,
 +) -> None:
 +    monkeypatch.setenv("HOME", str(temp_dir.resolve(strict=False)))
 +    context = build_context(
 +        temp_dir=temp_dir,
 +        use_react=False,
 +    )
 +    repairer = ResponseRepairer(context)
++
 +    guide_root = temp_dir / "Loader" / "guides" / "nginx"
 +    index_path = guide_root / "index.html"
++
 +    implementation_plan = temp_dir / "implementation.md"
 +    implementation_plan.write_text(
 +        "\n".join(
 +            [
 +                "# Implementation Plan",
 +                "",
 +                "## File Changes",
 +                f"- `{index_path}`",
 +                "",
 +            ]
 +        )
 +    )
++
 +    dod = create_definition_of_done("Create a multi-file nginx guide.")
 +    dod.implementation_plan = str(implementation_plan)
 +    dod.pending_items.extend(
 +        [
 +            "Create nginx guide directory structure",
 +            "Write main index.html for nginx guide",
 +        ]
 +    )
++
 +    decision = repairer.handle_empty_response(
 +        task="Create a multi-file nginx guide.",
 +        original_task=None,
 +        empty_retry_count=1,
 +        max_empty_retries=2,
 +        dod=dod,
 +    )
++
 +    assert decision.should_continue is True
 +    assert decision.retry_message is not None
 +    assert "`~/Loader/guides/nginx/index.html`" in decision.retry_message
 +    assert (
 +        'Emit this tool shape now: `write(file_path="~/Loader/guides/nginx/index.html", content="...")`.'
 +        in decision.retry_message
 +    )
  def test_empty_response_retry_recovers_blocked_empty_file_path_to_concrete_target(
      assert decision.should_continue is True
      assert decision.retry_message is not None
      assert (
 -        f"Last tool failure: resend `write` for `{second_chapter}` with a valid `file_path` and real `content`."
 +        "Last tool failure: resend `write` for "
 +        f"`{display_runtime_path(second_chapter)}` with a valid `file_path` and real `content`."
          in decision.retry_message
+     )
      assert "Do not leave `file_path` empty" in decision.retry_message
      assert (
 -        f'Emit this tool shape now: `write(file_path="{second_chapter.resolve(strict=False)}", content="...")`.'
 +        f'Emit this tool shape now: `write(file_path="{display_runtime_path(second_chapter)}", content="...")`.'
          in decision.retry_message
+     )
          in decision.retry_message
+     )
      assert (
 -        f"Prefer one concrete `write` call for a file inside `{chapters}` before more research."
 +        "Prefer one concrete `write` call for a file inside "
 +        f"`{display_runtime_path(chapters)}` before more research."
          in decision.retry_message
+     )
      assert decision.should_continue is True
      assert decision.retry_message is not None
      assert (
 -        f"Prefer one `write(content=...)` call for `{output_index}` before more research."
 +        "Prefer one `write(content=...)` call for "
 +        f"`{display_runtime_path(output_index)}` before more research."
          in decision.retry_message
+     )
      assert str(reference_index) not in decision.retry_message
          in decision.retry_message
+     )
      assert (
 -        f"Prefer one `write(content=...)` call for `{chapters / '02-installation.html'}` "
 +        "Prefer one `write(content=...)` call for "
 +        f"`{display_runtime_path(chapters / '02-installation.html')}` "
          "before more research."
          in decision.retry_message
+     )
          in decision.retry_message
+     )
      assert (
 -        f"Prefer one `write(content=...)` call for `{(chapters / '02-installation.html').resolve(strict=False)}` "
 +        "Prefer one `write(content=...)` call for "
 +        f"`{display_runtime_path(chapters / '02-installation.html')}` "
          "before more research."
          in decision.retry_message
+     )
      assert (
 -        f'Emit this tool shape now: `write(file_path="{(chapters / "02-installation.html").resolve(strict=False)}", content="...")`.'
 +        f'Emit this tool shape now: `write(file_path="{display_runtime_path(chapters / "02-installation.html")}", content="...")`.'
          in decision.retry_message
+     )
      assert (

tests/test_tool_batches.pymodified

+ )
  from loader.runtime.events import AgentEvent, TurnSummary
  from loader.runtime.executor import ToolExecutionOutcome, ToolExecutionState
 +from loader.runtime.path_display import display_runtime_path
  from loader.runtime.permissions import (
      PermissionMode,
      build_permission_policy,
      assert "A declared output artifact is still missing." in persistent_messages[0]
      assert "Resume by creating `04-variables.html` now." in persistent_messages[0]
      assert (
 -        f"Prefer one `write` call for `{temp_dir / 'chapters' / '04-variables.html'}` instead of more rereads."
 +        "Prefer one `write` call for "
 +        f"`{display_runtime_path(temp_dir / 'chapters' / '04-variables.html')}` instead of more rereads."
          in persistent_messages[0]
+     )
      assert ephemeral_messages == []
      assert "Do not reread reference material or spend the next turn on bookkeeping." in message
 +@pytest.mark.asyncio
 +async def test_tool_batch_runner_directory_handoff_uses_home_relative_path(
 +    temp_dir: Path,
 +    monkeypatch: pytest.MonkeyPatch,
 +) -> None:
 +    monkeypatch.setenv("HOME", str(temp_dir.resolve(strict=False)))
++
 +    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 for this scenario")
++
 +    nginx_root = temp_dir / "Loader" / "guides" / "nginx"
 +    chapters = nginx_root / "chapters"
 +    index_path = nginx_root / "index.html"
++
 +    implementation_plan = temp_dir / "implementation.md"
 +    implementation_plan.write_text(
 +        "\n".join(
 +            [
 +                "# Implementation Plan",
 +                "",
 +                "## File Changes",
 +                f"- `{chapters}/`",
 +                f"- `{index_path}`",
 +                "",
 +            ]
 +        )
 +    )
++
 +    context = build_context(
 +        temp_dir=temp_dir,
 +        messages=[],
 +        safeguards=FakeSafeguards(),
 +        assess_confidence=assess_confidence,
 +        verify_action=verify_action,
 +        auto_recover=False,
 +    )
 +    persistent_messages: list[str] = []
 +    context.queue_steering_message_callback = persistent_messages.append
 +    runner = ToolBatchRunner(context, DefinitionOfDoneStore(temp_dir))
 +    dod = create_definition_of_done("Create a multi-file nginx guide.")
 +    dod.implementation_plan = str(implementation_plan)
 +    sync_todos_to_definition_of_done(
 +        dod,
 +        [
 +            {
 +                "content": "Create the nginx directory structure",
 +                "active_form": "Creating the nginx directory structure",
 +                "status": "pending",
 +            },
 +            {
 +                "content": "Develop the main index.html file with proper structure",
 +                "active_form": "Developing the main index.html file with proper structure",
 +                "status": "pending",
 +            },
 +        ],
 +    )
++
 +    tool_call = ToolCall(
 +        id="mkdir-nginx-home",
 +        name="bash",
 +        arguments={"command": f"mkdir -p {chapters}"},
 +    )
 +    executor = FakeExecutor(
 +        [
 +            tool_outcome(
 +                tool_call=tool_call,
 +                output="",
 +                is_error=False,
 +            )
 +        ]
 +    )
++
 +    summary = TurnSummary(final_response="")
 +    await runner.execute_batch(
 +        tool_calls=[tool_call],
 +        tool_source="assistant",
 +        pending_tool_calls_seen=set(),
 +        emit=_noop_emit,
 +        summary=summary,
 +        dod=dod,
 +        executor=executor,  # type: ignore[arg-type]
 +        on_confirmation=None,
 +        on_user_question=None,
 +        emit_confirmation=None,
 +        consecutive_errors=0,
 +    )
++
 +    assert persistent_messages
 +    message = persistent_messages[-1]
 +    assert "Next step: create `index.html`." in message
 +    assert "`~/Loader/guides/nginx/index.html`" in message
++
++
  @pytest.mark.asyncio
  async def test_tool_batch_runner_redirects_post_write_self_audit_to_next_missing_artifact(
      temp_dir: Path,
      assert "did not provide a valid `file_path`" in queued[0]
      assert "Resume by creating `02-installation.html` now." in queued[0]
      assert (
 -        f"Prefer one `write` call for `{chapter_two}` instead of more rereads."
 +        f"Prefer one `write` call for `{display_runtime_path(chapter_two)}` instead of more rereads."
          in queued[0]
+     )
      assert context.recovery_context is not None