"""Tests for response-repair helpers on RuntimeContext."""
from __future__ import annotations
import json
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,
load_permission_rules,
)
from loader.runtime.recovery import RecoveryContext
from loader.runtime.repair import ResponseRepairer
from loader.tools.base import create_default_registry
from tests.helpers.runtime_harness import ScriptedBackend
class FakeSession:
def __init__(self) -> None:
self.messages = []
def append(self, message) -> None:
self.messages.append(message)
class FakeCodeFilter:
def reset(self) -> None:
return None
class FakeSafeguards:
def __init__(self) -> None:
self.action_tracker = object()
self.validator = object()
self.code_filter = FakeCodeFilter()
def filter_stream_chunk(self, content: str) -> str:
return content
def filter_complete_content(self, content: str) -> str:
return content
def should_steer(self) -> bool:
return False
def get_steering_message(self) -> str | None:
return None
def record_response(self, content: str) -> None:
return None
def detect_text_loop(self, content: str) -> tuple[bool, str]:
return False, ""
def detect_loop(self) -> tuple[bool, str]:
return False, ""
def build_context(
*,
temp_dir: Path,
use_react: bool,
) -> RuntimeContext:
registry = create_default_registry(temp_dir)
registry.configure_workspace_root(temp_dir)
rule_status = load_permission_rules(temp_dir)
policy = build_permission_policy(
active_mode=PermissionMode.WORKSPACE_WRITE,
workspace_root=temp_dir,
tool_requirements=registry.get_tool_requirements(),
rules=rule_status.rules,
)
session = FakeSession()
return RuntimeContext(
project_root=temp_dir,
backend=ScriptedBackend(),
registry=registry,
session=session, # type: ignore[arg-type]
config=SimpleNamespace(force_react=use_react),
capability_profile=SimpleNamespace(supports_native_tools=not use_react), # type: ignore[arg-type]
project_context=None,
permission_policy=policy,
permission_config_status=rule_status,
workflow_mode="execute",
safeguards=FakeSafeguards(),
)
def test_response_repairer_uses_runtime_parser_for_bracket_tool_fallback(
temp_dir: Path,
) -> None:
context = build_context(
temp_dir=temp_dir,
use_react=False,
)
repairer = ResponseRepairer(context)
analysis = repairer.analyze_response(
content="I need clarification.",
response_content='[calls askuserquestion tool with: question="Which path?"]',
tool_calls=[],
extracted_iterations=0,
max_extracted_iterations=3,
)
assert analysis.tool_calls == [
ToolCall(
id="call_0",
name="AskUserQuestion",
arguments={"question": "Which path?"},
)
]
assert analysis.tool_source == "raw_text"
assert analysis.clear_stream is True
def test_response_repairer_recovers_todowrite_from_runtime_registry(
temp_dir: Path,
) -> None:
context = build_context(
temp_dir=temp_dir,
use_react=False,
)
repairer = ResponseRepairer(context)
analysis = repairer.analyze_response(
content="I'll track the work first.",
response_content=json.dumps(
{
"name": "TodoWrite",
"arguments": {
"todos": [
{
"content": "Run tests",
"active_form": "Running tests",
"status": "in_progress",
}
]
},
}
),
tool_calls=[],
extracted_iterations=0,
max_extracted_iterations=3,
)
assert analysis.tool_source == "raw_text"
assert analysis.clear_stream is True
assert analysis.tool_calls == [
ToolCall(
id="call_0",
name="TodoWrite",
arguments={
"todos": [
{
"content": "Run tests",
"active_form": "Running tests",
"status": "in_progress",
}
]
},
)
]
def test_response_repairer_fails_honestly_when_raw_tool_budget_is_exhausted(
temp_dir: Path,
) -> None:
context = build_context(
temp_dir=temp_dir,
use_react=False,
)
repairer = ResponseRepairer(context)
analysis = repairer.analyze_response(
content=json.dumps(
{
"name": "read",
"arguments": {"file_path": "README.md"},
}
),
response_content=json.dumps(
{
"name": "read",
"arguments": {"file_path": "README.md"},
}
),
tool_calls=[],
extracted_iterations=3,
max_extracted_iterations=3,
)
assert analysis.should_stop is True
assert analysis.final_response == (
"I couldn't safely continue because the model kept emitting raw-text "
"tool calls instead of proper tool invocations. Please try again or "
"switch to a different backend/model."
)
assert analysis.failure == "raw-text tool recovery budget exhausted"
assert "Let me know if you'd like me to continue" not in analysis.final_response
def test_empty_response_retry_message_surfaces_missing_planned_artifacts_and_working_note(
temp_dir: Path,
) -> None:
context = build_context(
temp_dir=temp_dir,
use_react=False,
)
repairer = ResponseRepairer(context)
implementation_plan = temp_dir / "implementation.md"
implementation_plan.write_text(
"\n".join(
[
"# Implementation Plan",
"",
"## File Changes",
f"- `{temp_dir / 'guides' / 'nginx' / 'index.html'}`",
f"- `{temp_dir / 'guides' / 'nginx' / 'chapters'}`",
"",
]
)
)
first_artifact = temp_dir / "guides" / "nginx" / "index.html"
first_artifact.parent.mkdir(parents=True)
first_artifact.write_text("\n")
dod = create_definition_of_done("Create a multi-file nginx guide.")
dod.implementation_plan = str(implementation_plan)
dod.touched_files.append(str(first_artifact))
dod.completed_items.append("Create the main index.html file")
dod.pending_items.append("Create each chapter file in sequence")
context.session.append(
SimpleNamespace(
role="tool",
content=(
"Observation [notepad_write_working]: Result: "
"- [2026-04-21T19:17:34Z] Creating fifth chapter file: Advanced configurations"
),
)
)
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 "Latest working note: Creating fifth chapter file: Advanced configurations" in decision.retry_message
assert "Confirmed touched files: `index.html`" in decision.retry_message
assert "Confirmed completed work: Create the main index.html file" in decision.retry_message
assert "Next pending item: Create each chapter file in sequence" in decision.retry_message
assert "Continue from the confirmed progress below instead of restarting." in decision.retry_message
def test_empty_response_retry_during_html_quality_repair_shrinks_mutation(
temp_dir: Path,
) -> None:
context = build_context(
temp_dir=temp_dir,
use_react=False,
)
repairer = ResponseRepairer(context)
guide = temp_dir / "guides" / "nginx"
chapters = guide / "chapters"
chapters.mkdir(parents=True)
first_chapter = chapters / "01-introduction.html"
second_chapter = chapters / "02-installation.html"
first_chapter.write_text("Intro
\n")
second_chapter.write_text("Install
\n")
context.session.append(
Message(
role=Role.USER,
content=(
"Repair focus:\n"
f"- Improve `{first_chapter}`: insufficient structured content "
"(13 blocks, expected at least 18).\n"
f"- Improve `{second_chapter}`: thin content "
"(514 text chars, expected at least 1758).\n"
f"- Immediate next step: edit `{first_chapter}` with a substantial "
"expansion or replacement that satisfies its listed quality issue.\n"
),
)
)
dod = create_definition_of_done("Create an equally thorough HTML guide.")
dod.touched_files = [str(first_chapter), str(second_chapter)]
dod.pending_items.append("Create chapter 1: Introduction to Nginx")
decision = repairer.handle_empty_response(
task="Create an equally thorough HTML 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 "while repairing HTML content quality" in decision.retry_message
assert (
f"Active quality repair target: `{first_chapter.resolve(strict=False)}`"
in decision.retry_message
)
assert "Shrink the next step" in decision.retry_message
assert "4-6 substantive sections" in decision.retry_message
assert "do not attempt a giant full-page rewrite from memory" in decision.retry_message
assert "No narration, no TodoWrite, no final summary" in decision.retry_message
assert f"`{second_chapter.resolve(strict=False)}`" in decision.retry_message
def test_empty_response_retry_uses_exact_anchor_after_stale_quality_repair_context(
temp_dir: Path,
) -> None:
context = build_context(
temp_dir=temp_dir,
use_react=False,
)
repairer = ResponseRepairer(context)
guide = temp_dir / "guides" / "nginx"
chapters = guide / "chapters"
chapters.mkdir(parents=True)
chapter = chapters / "05-load-balancing.html"
chapter.write_text("Load Balancing
\n")
context.session.append(
Message(
role=Role.USER,
content=(
"Repair focus:\n"
f"- Improve `{chapter}`: thin content "
"(846 text chars, expected at least 1758).\n"
f"- Immediate next step: edit `{chapter}`.\n"
),
)
)
context.session.append(
Message(
role=Role.TOOL,
content=(
"Observation [edit]: Error: Failed to complete the operation "
f"after 2 attempts for {chapter}. old_string not found in file."
),
)
)
dod = create_definition_of_done("Create an equally thorough HTML guide.")
dod.touched_files = [str(chapter)]
decision = repairer.handle_empty_response(
task="Create an equally thorough HTML 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 (
"Use exactly one `edit(file_path=..., old_string=..., new_string=...)`"
in decision.retry_message
)
assert "Use this exact `old_string` value from the current file" in decision.retry_message
assert "```html\n