674 lines
35 KiB
Python
674 lines
35 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Verify escalation identity, response ownership, and portable progression."""
|
|
|
|
from dataclasses import replace
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
from urllib.parse import quote
|
|
|
|
import pytest
|
|
|
|
from band_aider.ai_client.cached_completion_client import CachedCompletionClient
|
|
from band_aider.ai_client.completion_client import CompletionClient, HumanResponse, OpenAIConfig
|
|
from band_aider.ai_client.cycle_jacker import CycleJacker
|
|
from band_aider.ai_client.escalation_attempts import EscalationAttempts, EscalationState
|
|
from band_aider.ai_client.stabilization_escalation import StabilizationEscalation, STABILIZATION_TIER_ATTEMPTS
|
|
from band_aider.ai_client.cycle_jacker import cycle
|
|
from band_aider.ai_client.fence_crashing_client import FenceCrashingClient
|
|
from band_aider.ai_client.human_feedback_store import HumanFeedbackStore
|
|
from band_aider.ai_client.json_guarding_client import JsonGuardingClient
|
|
from band_aider.ai_client.token_jacker import TokenJacker
|
|
from band_aider.context import provenance
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.context.provenance import Phase, Provenance
|
|
from band_aider.context.working_backlog import InspectionBacklog
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
|
|
from band_aider.judgement.repo_exec import HostRepoExec
|
|
from band_aider.ui.dto.human_intervention import HumanInterventionRequestDTO, HumanInterventionResponseDTO
|
|
from band_aider.ui.status_publisher import HumanCommandReceiver, StatusPublisher
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
class _Provider(CompletionClient):
|
|
"""Record actual provider calls while returning independently supplied text."""
|
|
|
|
def __init__(self, name: str, *responses: str) -> None:
|
|
"""Bind one named tier and its explicit responses."""
|
|
|
|
self.name = name
|
|
self.responses = iter(responses)
|
|
self.prompts: list[str] = []
|
|
|
|
@property
|
|
def inner(self) -> CompletionClient | None:
|
|
"""Act as the terminal provider boundary."""
|
|
|
|
return None
|
|
|
|
@property
|
|
def openai_configs(self) -> tuple[OpenAIConfig, ...]:
|
|
"""Describe one provider without exposing live credentials."""
|
|
|
|
return (OpenAIConfig(model=self.name, api_base="http://unused"),)
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Return the next response and fail if an undeclared call occurs."""
|
|
|
|
self.prompts.append(prompt)
|
|
return next(self.responses)
|
|
|
|
|
|
class _Human(StatusPublisher, HumanCommandReceiver):
|
|
"""Correlate scripted human commands to newly published intervention IDs."""
|
|
|
|
def __init__(self, *replies: HumanInterventionResponseDTO) -> None:
|
|
"""Keep explicit input and published requests in independent per-test state."""
|
|
|
|
self.replies = iter(replies)
|
|
self.requests: list[HumanInterventionRequestDTO] = []
|
|
|
|
def publish(self, event: object) -> None:
|
|
"""Retain only the human interaction events for inspection."""
|
|
|
|
if isinstance(event, HumanInterventionRequestDTO):
|
|
self.requests.append(event)
|
|
|
|
def consume_pending(self) -> None:
|
|
"""Provide no out-of-band interruption in these stationary scenarios."""
|
|
|
|
def wait_for(self, request_id: str) -> HumanInterventionResponseDTO:
|
|
"""Return one command for the exact freshly published request."""
|
|
|
|
return replace(next(self.replies), request_id=request_id)
|
|
|
|
|
|
def test_model_ladder_then_one_hint_then_verbatim_human_responses() -> None:
|
|
"""Changing prompts and response bytes cannot restart a decision's escalation ladder."""
|
|
|
|
local = _Provider("local", *("local invalid",) * 12)
|
|
cloud1 = _Provider("cloud1", *("cloud1 invalid",) * 12)
|
|
cloud2 = _Provider("cloud2", *("cloud2 invalid",) * 12, "hinted bad")
|
|
human = _Human(
|
|
HumanInterventionResponseDTO("", "reconstruct PRE + DIFF\n保持", "resend"),
|
|
HumanInterventionResponseDTO("", " malformed human\n", "resend"),
|
|
HumanInterventionResponseDTO("", ' {"outcome":"pass"}\n', "resend"),
|
|
)
|
|
workflow = StabilizationEscalation(
|
|
(local, cloud1, cloud2), check_name="Acceptance", prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
event_publisher=human, command_receiver=human,
|
|
)
|
|
workflow.ask = 'Complete ask with {"outcome":"pass","reason":["line"]}'
|
|
|
|
responses = tuple(map(lambda index: workflow.complete(f"prompt {index}"), range(39)))
|
|
|
|
assert responses == ("local invalid",) * 12 + ("cloud1 invalid",) * 12 + ("cloud2 invalid",) * 12 + ("hinted bad", " malformed human\n", ' {"outcome":"pass"}\n')
|
|
assert isinstance(responses[-1], HumanResponse)
|
|
assert local.prompts == list(map(lambda index: f"prompt {index}", range(12)))
|
|
assert cloud1.prompts == list(map(lambda index: f"prompt {index}", range(12, 24)))
|
|
assert len(cloud2.prompts) == 13
|
|
assert "reconstruct PRE + DIFF\n保持" in cloud2.prompts[-1]
|
|
assert tuple(map(lambda request: request.mode, human.requests)) == ("hint", "direct_response", "direct_response")
|
|
assert len(set(map(lambda request: request.request_id, human.requests))) == 3
|
|
assert human.requests[-1].draft == " malformed human\n"
|
|
assert human.requests[-1].original_prompt == workflow.ask
|
|
assert human.requests[-1].repeated_response_tail == ""
|
|
|
|
|
|
def test_execution_conflict_requests_strongest_without_spending_weaker_retries() -> None:
|
|
"""Explicit receipt conflict uses the existing ladder and reports the answering tier."""
|
|
local, cloud = _Provider("local", "reject"), _Provider("cloud", "invalid", "accept")
|
|
middle = _Provider("middle")
|
|
workflow = StabilizationEscalation((local, middle, cloud), check_name="Acceptance", prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
|
|
assert workflow.complete("judge") == "reject"
|
|
assert workflow.last_client_index == 0
|
|
workflow.request_strongest()
|
|
assert workflow.complete("receipt conflicts with rejection") == "invalid"
|
|
assert workflow.last_client_index == 2
|
|
assert workflow.complete("correct cloud output") == "accept"
|
|
assert workflow.last_client_index == 2
|
|
assert local.prompts == ["judge"]
|
|
assert middle.prompts == []
|
|
|
|
|
|
def test_strongest_request_with_one_provider_uses_that_provider() -> None:
|
|
"""Strongest means the configured endpoint, not an invented cloud or human tier."""
|
|
local = _Provider("local", "reject", "hinted local")
|
|
human = _Human(HumanInterventionResponseDTO("", "Inspect the fresh execution.", "resend"))
|
|
workflow = StabilizationEscalation((local,), check_name="Acceptance", prompt_formatter_builder=MarkdownPromptFormatterBuilder(), event_publisher=human, command_receiver=human)
|
|
|
|
assert workflow.complete("judge") == "reject"
|
|
workflow.request_strongest()
|
|
assert workflow.complete("receipt conflict") == "hinted local"
|
|
assert workflow.last_client_index == 0
|
|
assert len(human.requests) == 0
|
|
|
|
|
|
@pytest.mark.parametrize("suppressed", (False, True))
|
|
def test_highest_tier_contradiction_uses_available_unsuppressed_human(
|
|
monkeypatch: pytest.MonkeyPatch, suppressed: bool,
|
|
) -> None:
|
|
"""A highest-tier finding requests available human help unless intervention is suppressed."""
|
|
|
|
monkeypatch.setattr(cycle, "intervention_suppressed", suppressed)
|
|
provider = _Provider("highest", "finding", "confirmed")
|
|
human = _Human(HumanInterventionResponseDTO("", "Check the competing assertions.", "resend"))
|
|
workflow = StabilizationEscalation(
|
|
(provider,), check_name="FileBlame", prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
event_publisher=human, command_receiver=human,
|
|
)
|
|
assert workflow.complete("same evidence") == "finding"
|
|
|
|
workflow.request_strongest(human_if_already_strongest=True)
|
|
assert workflow.complete("same evidence plus conflicting assertion") == "confirmed"
|
|
|
|
assert len(human.requests) == (0 if suppressed else 1)
|
|
assert len(provider.prompts) == 2
|
|
assert "same evidence plus conflicting assertion" in provider.prompts[-1]
|
|
assert ("Check the competing assertions." in provider.prompts[-1]) is (not suppressed)
|
|
|
|
|
|
def test_highest_tier_contradiction_without_human_keeps_model_correction(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Unavailable human help leaves review on the highest model through repeated invalid answers."""
|
|
|
|
monkeypatch.setattr(cycle, "intervention_suppressed", False)
|
|
attempts = STABILIZATION_TIER_ATTEMPTS * 3
|
|
local = _Provider("local")
|
|
provider = _Provider("highest", "finding", *("invalid",) * attempts, "confirmed")
|
|
workflow = StabilizationEscalation(
|
|
(local, provider), check_name="FileBlame", prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
)
|
|
workflow.request_strongest()
|
|
assert workflow.complete("same evidence") == "finding"
|
|
|
|
workflow.request_strongest(human_if_already_strongest=True)
|
|
assert tuple(map(lambda _: workflow.complete("same evidence plus assertion"), range(attempts))) == ("invalid",) * attempts
|
|
assert workflow.complete("same evidence plus assertion") == "confirmed"
|
|
|
|
assert len(provider.prompts) == attempts + 2
|
|
assert workflow.last_client_index == 1
|
|
assert local.prompts == []
|
|
|
|
|
|
def test_strongest_review_outlasts_the_model_and_human_allowances() -> None:
|
|
"""Correctable strongest-tier responses cannot fall through to human or weaker tiers."""
|
|
attempts = STABILIZATION_TIER_ATTEMPTS * 4 + 1
|
|
local, middle = _Provider("local", "reject"), _Provider("middle")
|
|
strongest = _Provider("strongest", *("invalid",) * attempts, "valid")
|
|
human = _Human()
|
|
workflow = StabilizationEscalation(
|
|
(local, middle, strongest), check_name="Acceptance",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
event_publisher=human, command_receiver=human,
|
|
)
|
|
|
|
assert workflow.complete("same evidence") == "reject"
|
|
workflow.request_strongest()
|
|
assert tuple(map(lambda _: workflow.complete("same evidence"), range(attempts))) == ("invalid",) * attempts
|
|
assert workflow.complete("same evidence") == "valid"
|
|
|
|
assert local.prompts == ["same evidence"]
|
|
assert middle.prompts == []
|
|
assert strongest.prompts == ["same evidence"] * (attempts + 1)
|
|
assert human.requests == []
|
|
assert workflow.last_client_index == 2
|
|
assert workflow.last_response == "valid"
|
|
|
|
|
|
def test_resume_preserves_node_and_class_progression_without_prompt_hashes() -> None:
|
|
"""An accepted response consumes an attempt across shutdown; new node or class starts fresh."""
|
|
|
|
repo = MemoryRepoExec()
|
|
graph = Provenance.open(repo_root=repo.root, repo_exec=repo, resume=False)
|
|
item = InspectionBacklog(focused_spotlight=ClassMeta("src/a.py", "", [], {}), summary="repair")
|
|
graph.start(Phase.ALPHA, item)
|
|
local = _Provider("local", "valid fail", "valid pass", *("invalid",) * 10, "other class", "new node")
|
|
cloud = _Provider("cloud", "cloud judgment")
|
|
builder = MarkdownPromptFormatterBuilder()
|
|
first = StabilizationEscalation((local, cloud), check_name="Acceptance", prompt_formatter_builder=builder)
|
|
with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
|
|
assert first.complete("one") == "valid fail"
|
|
assert first.complete("two") == "valid pass"
|
|
|
|
resumed = Provenance.open(repo_root=repo.root, repo_exec=repo, resume=True)
|
|
second = StabilizationEscalation((local, cloud), check_name="Acceptance", prompt_formatter_builder=builder)
|
|
other = StabilizationEscalation((local, cloud), check_name="Applicability", prompt_formatter_builder=builder)
|
|
with provenance.bind(resumed), provenance.item_scope(Phase.ALPHA, item):
|
|
assert tuple(map(lambda _: second.complete("changed after resume"), range(10))) == ("invalid",) * 10
|
|
assert second.complete("thirteenth") == "cloud judgment"
|
|
assert other.complete("thirteenth") == "other class"
|
|
next_item = InspectionBacklog(focused_spotlight=ClassMeta("src/b.py", "", [], {}), summary="other")
|
|
resumed.start(Phase.ALPHA, next_item)
|
|
with provenance.bind(resumed), provenance.item_scope(Phase.ALPHA, next_item):
|
|
assert second.complete("thirteenth") == "new node"
|
|
|
|
paths = EscalationAttempts(repo.root, repo).records(item.message_id, "Acceptance")
|
|
assert paths[0].name == f"{item.message_id}_Acceptance.json"
|
|
assert paths[-1].name == f"{item.message_id}_Acceptance(12).json"
|
|
assert len(paths) == 13
|
|
assert EscalationAttempts(repo.root, repo).load(item.message_id, "Acceptance").attempts == 13
|
|
assert len(local.prompts) == 14
|
|
assert cloud.prompts == ["thirteenth"]
|
|
|
|
|
|
def test_attempt_files_move_with_the_repository_and_ignore_incomplete_writes() -> None:
|
|
"""Portable append-only records require neither the old root path nor the provenance process."""
|
|
|
|
original = MemoryRepoExec(Path("/original"))
|
|
store = EscalationAttempts(original.root, original)
|
|
store.save("node", "Acceptance", EscalationState(9, "last", "human draft"))
|
|
moved = MemoryRepoExec(Path("/moved"))
|
|
path = store.records("node", "Acceptance")[0]
|
|
moved.write_text(moved.root / path.relative_to(original.root), original.read_text(path))
|
|
moved.write_text(moved.root / ".band-aider-attempts/node_Acceptance(1).tmp", "interrupted")
|
|
resumed = EscalationAttempts(moved.root, moved)
|
|
|
|
assert resumed.load("node", "Acceptance") == EscalationState(9, "last", "human draft")
|
|
resumed.save("node", "Acceptance", EscalationState(10, "corrected", "corrected"))
|
|
assert resumed.load("node", "Acceptance").attempts == 10
|
|
assert original.read_text(path) == moved.read_text(moved.root / path.relative_to(original.root))
|
|
|
|
|
|
def test_displayed_file_counters_survive_resume_without_crossing_files(caplog: pytest.LogCaptureFixture) -> None:
|
|
"""Different same-basename files start locally; a resumed file keeps its spent tier."""
|
|
|
|
repo = MemoryRepoExec()
|
|
graph = Provenance.open(repo_root=repo.root, repo_exec=repo, resume=False)
|
|
item = InspectionBacklog(focused_spotlight=ClassMeta("src/owner.py", "", [], {}), summary="repair")
|
|
graph.start(Phase.ALPHA, item)
|
|
local = _Provider("local", *("invalid",) * 12, "file B", "legacy", "literal percent")
|
|
raw = ' {"include":false}\n'
|
|
cloud = _Provider("cloud", raw)
|
|
builder = MarkdownPromptFormatterBuilder()
|
|
first = StabilizationEscalation((local, cloud), check_name="Selection", prompt_formatter_builder=builder)
|
|
with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
|
|
assert tuple(map(lambda _: first.complete("judge A", disambiguation="src/a/file.py"), range(12))) == ("invalid",) * 12
|
|
|
|
resumed = Provenance.open(repo_root=repo.root, repo_exec=repo, resume=True)
|
|
second = StabilizationEscalation((local, cloud), check_name="Selection", prompt_formatter_builder=builder)
|
|
with caplog.at_level(logging.INFO), provenance.bind(resumed), provenance.item_scope(Phase.ALPHA, item):
|
|
assert second.complete("judge B", disambiguation="src/b/file.py") == "file B"
|
|
assert second.complete("changed source and prompt", disambiguation="src/a/file.py") == '{"include":false}'
|
|
assert second.last_response == raw
|
|
assert second.complete("undisambiguated") == "legacy"
|
|
assert second.complete("literal percent", disambiguation="src%2Fa%2Ffile.py") == "literal percent"
|
|
|
|
assert cloud.prompts == ["changed source and prompt"]
|
|
assert "disambiguation=src/a/file.py" in caplog.text
|
|
store = EscalationAttempts(repo.root, repo)
|
|
assert store.load(item.message_id, "Selection/src/a/file.py").attempts == 13
|
|
assert store.load(item.message_id, "Selection/src/b/file.py").attempts == 1
|
|
assert store.load(item.message_id, "Selection").attempts == 1
|
|
assert store.load(item.message_id, "Selection/src%252Fa%252Ffile.py").attempts == 1
|
|
|
|
|
|
def test_instance_local_disambiguation_separates_ordered_candidate_pairs() -> None:
|
|
"""Changing either displayed candidate or its comparison order cannot spend another pair's attempts."""
|
|
|
|
local = _Provider("local", *("invalid",) * 12, "different pair", "reversed pair")
|
|
cloud = _Provider("cloud", "original pair")
|
|
workflow = StabilizationEscalation(
|
|
(local, cloud), check_name="Precedence", prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
)
|
|
pair = '["src/a.py", "src/b.py"]'
|
|
assert tuple(map(lambda _: workflow.complete("compare", disambiguation=pair), range(12))) == ("invalid",) * 12
|
|
assert workflow.complete("compare", disambiguation='["src/a.py", "src/c.py"]') == "different pair"
|
|
assert workflow.complete("compare", disambiguation='["src/b.py", "src/a.py"]') == "reversed pair"
|
|
assert workflow.complete("updated evidence", disambiguation=pair) == "original pair"
|
|
assert workflow.last_response == "original pair"
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_multi_file_identity_fits_host_storage_without_hashing_or_truncation(tmp_path_factory: pytest.TempPathFactory) -> None:
|
|
"""Several displayed paths remain readable and resumable without becoming one oversized filename."""
|
|
|
|
tmp_path = tmp_path_factory.mktemp("escalation")
|
|
repo = HostRepoExec(tmp_path)
|
|
graph = Provenance.open(repo_root=tmp_path, repo_exec=repo, resume=False)
|
|
item = InspectionBacklog(focused_spotlight=ClassMeta("src/owner.py", "", [], {}), summary="repair")
|
|
graph.start(Phase.ALPHA, item)
|
|
files = json.dumps((
|
|
"band_aider/judgement/predictive_failure_source_candidate_selection_strategy.py",
|
|
"band_aider/judgement/source_needed_to_understand_failure_check.py",
|
|
"band_aider/context/targeted_test_collection_failure_owner_check.py",
|
|
"band_aider/context/file_blame_check.py",
|
|
))
|
|
workflow = StabilizationEscalation(
|
|
(_Provider("local", "first", "second"),), check_name="FileBlameDecision",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
)
|
|
with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
|
|
assert workflow.complete("judge", disambiguation=files) == "first"
|
|
assert workflow.complete("judge again", disambiguation=files) == "second"
|
|
store = EscalationAttempts(tmp_path, repo)
|
|
key = f"FileBlameDecision/{quote(files, safe='/')}"
|
|
assert store.load(item.message_id, key).attempts == 2
|
|
assert len(store.records(item.message_id, key)) == 2
|
|
assert json.loads(repo.read_text(store.records(item.message_id, key)[-1]))["decision"] == key
|
|
|
|
|
|
def test_cycle_recovery_does_not_spend_stabilization_attempts() -> None:
|
|
"""Interleaved cycle recovery cannot advance or reset the same node's stabilization ladder."""
|
|
|
|
repo = MemoryRepoExec()
|
|
graph = Provenance.open(repo_root=repo.root, repo_exec=repo, resume=False)
|
|
item = InspectionBacklog(focused_spotlight=ClassMeta("src/a.py", "", [], {}), summary="repair")
|
|
graph.start(Phase.ALPHA, item)
|
|
local = _Provider("local", "first judgment", "repeated cycle answer", "repeated cycle answer", *("remaining local judgment",) * 11)
|
|
cloud = _Provider("cloud", "cycle recovered", "thirteenth judgment")
|
|
builder = MarkdownPromptFormatterBuilder()
|
|
first = StabilizationEscalation((local, cloud), check_name="Acceptance", prompt_formatter_builder=builder)
|
|
# Cycle recovery must remain usable when constructing stabilization state is forbidden.
|
|
with patch.object(StabilizationEscalation, "__init__", side_effect=AssertionError("cycle constructed stabilization state")):
|
|
detector = CycleJacker(
|
|
local, escalation_clients=(cloud,), repeat_threshold=1,
|
|
feedback_store=HumanFeedbackStore(repo.root, repo_exec=repo), prompt_formatter_builder=builder,
|
|
)
|
|
with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
|
|
assert first.complete("judge") == "first judgment"
|
|
assert detector.complete("ordinary completion") == "repeated cycle answer"
|
|
assert detector.complete("ordinary completion") == "cycle recovered"
|
|
store = EscalationAttempts(repo.root, repo)
|
|
assert store.load(item.message_id, "Acceptance").attempts == 1
|
|
assert store.records(item.message_id, "CycleJacker") == ()
|
|
|
|
resumed = StabilizationEscalation((local, cloud), check_name="Acceptance", prompt_formatter_builder=builder)
|
|
assert tuple(map(lambda _: resumed.complete("judge"), range(11))) == ("remaining local judgment",) * 11
|
|
assert resumed.complete("judge") == "thirteenth judgment"
|
|
assert store.load(item.message_id, "Acceptance").attempts == 13
|
|
|
|
assert local.prompts == ["judge", "ordinary completion", "ordinary completion"] + ["judge"] * 11
|
|
assert cloud.prompts == ["ordinary completion", "judge"]
|
|
|
|
|
|
def test_stabilization_bypasses_cache_cycle_and_hidden_json_repairs() -> None:
|
|
"""Judgments clean fences while bypassing caches and hidden repairs on every tier."""
|
|
|
|
from band_aider.context.token_budget_state import TokenBudgetSettings
|
|
|
|
repo = MemoryRepoExec()
|
|
fenced = '```json\n{"outcome":"pass"}\n```'
|
|
local = _Provider("local", "cached local", "not json", fenced, *("still invalid",) * 10)
|
|
cloud = _Provider("cloud", "cached cloud", fenced)
|
|
cached_local = CachedCompletionClient(local)
|
|
cached_cloud = CachedCompletionClient(cloud)
|
|
assert cached_local.complete("return strict JSON") == "cached local"
|
|
assert cached_cloud.complete("return strict JSON") == "cached cloud"
|
|
budget = TokenJacker(
|
|
cached_local, settings=TokenBudgetSettings(max_prompt_tokens=10_000),
|
|
escalation_clients=(cached_cloud,),
|
|
)
|
|
runtime = CycleJacker(
|
|
FenceCrashingClient(budget), escalation_clients=(cached_cloud,),
|
|
feedback_store=HumanFeedbackStore(repo.root, repo_exec=repo),
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
)
|
|
workflow = StabilizationEscalation.for_stabilization(
|
|
runtime, check_name="Acceptance", prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
)
|
|
guarded = JsonGuardingClient(workflow, MarkdownPromptFormatterBuilder())
|
|
with patch.object(runtime, "complete", side_effect=AssertionError("cycle decorator called")):
|
|
responses = tuple(map(lambda _: guarded.complete("return strict JSON"), range(13)))
|
|
|
|
assert responses == ("not json", '{"outcome":"pass"}') + ("still invalid",) * 10 + ('{"outcome":"pass"}',)
|
|
assert workflow.last_response == fenced
|
|
assert isinstance(workflow.inner, TokenJacker)
|
|
assert runtime.complete("return strict JSON") == "cached local"
|
|
assert cached_cloud.complete("return strict JSON") == "cached cloud"
|
|
assert len(local.prompts) == 13
|
|
assert len(cloud.prompts) == 2
|
|
|
|
|
|
def test_rendering_retains_runtime_cache_and_cycle_policy_outside_judgment_attempts() -> None:
|
|
"""Repeated COT selection uses the ordinary runtime while only judgments spend the local allowance."""
|
|
|
|
from band_aider.context.cot_task_generator import CotTaskRecord
|
|
from band_aider.context.provenance import FutureGuidance
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.stabilization_decision import FutureGuidanceApplicability
|
|
from tests.band_aider_loop_impl import _file_manager
|
|
|
|
manager = _file_manager(Path("/testbed"))
|
|
source = ClassMeta("src/a.py", "", [], {})
|
|
item = InspectionBacklog(focused_spotlight=source, summary="repair")
|
|
graph = Provenance.open(repo_root=manager.repo_root, repo_exec=manager.repo_exec, resume=False)
|
|
graph.start(Phase.ALPHA, item)
|
|
answer = '{"applicable":false,"reason":"The later observation concerns a different obligation."}'
|
|
local = _Provider("local", "KEEP: COT-0000", answer, answer, "invalid JSON")
|
|
cloud = _Provider("cloud", answer)
|
|
runtime = CycleJacker(
|
|
CachedCompletionClient(local), escalation_clients=(cloud,),
|
|
feedback_store=HumanFeedbackStore(manager.repo_root, repo_exec=manager.repo_exec),
|
|
prompt_formatter_builder=manager.prompt_formatter_builder,
|
|
)
|
|
request = AiderEditRequest(
|
|
"Current obligation", [], [],
|
|
previous_cot_records=(CotTaskRecord("Preserve the caller contract.", source.location, source),),
|
|
)
|
|
with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
|
|
EscalationAttempts(manager.repo_root, manager.repo_exec).save(
|
|
item.message_id, "FutureGuidanceApplicability", EscalationState(9),
|
|
)
|
|
with patch.object(runtime, "complete", wraps=runtime.complete) as ordinary_calls:
|
|
# Separate decisions repeat identical rendering while sharing one obligation's progression.
|
|
for _ in range(3):
|
|
decision = FutureGuidanceApplicability(
|
|
client=runtime, file_manager=manager, request=request,
|
|
future_guidance=FutureGuidance("later", item.message_id, "Later evidence", "Other obligation"),
|
|
current_prompt="Judge current work.", logger=logging.getLogger(__name__),
|
|
)
|
|
assert decision() is False
|
|
assert ordinary_calls.call_count == 3
|
|
assert all(map(lambda call: "CONTEXTUAL COT SELECTION" in call.args[0], ordinary_calls.call_args_list))
|
|
assert EscalationAttempts(manager.repo_root, manager.repo_exec).load(
|
|
item.message_id, "FutureGuidanceApplicability",
|
|
).attempts == 13
|
|
|
|
assert len(local.prompts) == 4
|
|
assert sum(map(lambda prompt: prompt.startswith("== BEGIN CONTEXTUAL COT SELECTION =="), local.prompts)) == 1
|
|
assert len(cloud.prompts) == 1
|
|
assert "invalid JSON" in cloud.prompts[0]
|
|
|
|
|
|
def test_direct_human_skips_response_adapters() -> None:
|
|
"""Invalid human text is never cleaned up or sent for model repair."""
|
|
|
|
local = _Provider("local", *("invalid",) * 12)
|
|
human = _Human(
|
|
HumanInterventionResponseDTO("", "", "retry"),
|
|
HumanInterventionResponseDTO("", " ```json\ninvalid\n``` ", "resend"),
|
|
)
|
|
workflow = StabilizationEscalation((local,), check_name="Acceptance", prompt_formatter_builder=MarkdownPromptFormatterBuilder(), event_publisher=human, command_receiver=human)
|
|
assert tuple(map(lambda _: workflow.complete("judge"), range(12))) == ("invalid",) * 12
|
|
guarded = JsonGuardingClient(FenceCrashingClient(workflow), MarkdownPromptFormatterBuilder())
|
|
|
|
response = guarded.complete("return strict JSON")
|
|
|
|
assert response == " ```json\ninvalid\n``` "
|
|
assert isinstance(response, HumanResponse)
|
|
assert len(local.prompts) == 12
|
|
assert tuple(map(lambda request: request.mode, human.requests)) == ("hint", "direct_response")
|
|
|
|
|
|
def test_unattended_exhaustion_restarts_models_without_admitting_a_response() -> None:
|
|
"""An unavailable human cannot stop correction or turn exhaustion into acceptance."""
|
|
|
|
local = _Provider("local", *("invalid",) * 24, "corrected")
|
|
workflow = StabilizationEscalation((local,), check_name="Acceptance", prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
assert tuple(map(lambda _: workflow.complete("judge"), range(24))) == ("invalid",) * 24
|
|
assert workflow.complete("judge") == "corrected"
|
|
assert len(local.prompts) == 25
|
|
|
|
|
|
def test_suppression_repeats_full_model_ladders_without_human_prompts(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Every model retains twelve responses on successive circuits while F9 suppresses human work."""
|
|
|
|
monkeypatch.setattr(cycle, "intervention_suppressed", True)
|
|
local = _Provider("local", *("local",) * 25)
|
|
cloud1 = _Provider("cloud1", *("cloud1",) * 24)
|
|
cloud2 = _Provider("cloud2", *("cloud2",) * 24)
|
|
human = _Human()
|
|
workflow = StabilizationEscalation(
|
|
(local, cloud1, cloud2), check_name="Acceptance", prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
event_publisher=human, command_receiver=human,
|
|
)
|
|
|
|
replies = tuple(map(lambda index: workflow.complete(f"judgment {index}"), range(73)))
|
|
|
|
assert replies == (("local",) * 12 + ("cloud1",) * 12 + ("cloud2",) * 12) * 2 + ("local",)
|
|
assert human.requests == []
|
|
assert len(local.prompts) == 25
|
|
assert len(cloud1.prompts) == len(cloud2.prompts) == 24
|
|
|
|
|
|
def test_human_deferral_resumes_local_and_explicit_abort_still_stops() -> None:
|
|
"""Skipping a hint and declining dictation preserve correction; explicit abort retains operator authority."""
|
|
|
|
from band_aider.ai_client.escalation_utils import EscalationAborted
|
|
|
|
local = _Provider("local", *("invalid",) * 12, "retry locally")
|
|
human = _Human(HumanInterventionResponseDTO("", "", "retry"), HumanInterventionResponseDTO("", "", "defer"))
|
|
workflow = StabilizationEscalation(
|
|
(local,), check_name="Acceptance", prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
event_publisher=human, command_receiver=human,
|
|
)
|
|
tuple(map(lambda _: workflow.complete("judge"), range(12)))
|
|
|
|
assert workflow.complete("judge again") == "retry locally"
|
|
assert tuple(map(lambda request: request.mode, human.requests)) == ("hint", "direct_response")
|
|
assert len(local.prompts) == 13
|
|
|
|
aborted = _Human(HumanInterventionResponseDTO("", "", "abort"))
|
|
stopping = StabilizationEscalation(
|
|
(_Provider("local", *("invalid",) * 12),), check_name="Acceptance",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(), event_publisher=aborted, command_receiver=aborted,
|
|
)
|
|
tuple(map(lambda _: stopping.complete("judge"), range(12)))
|
|
with pytest.raises(EscalationAborted):
|
|
stopping.complete("judge")
|
|
|
|
|
|
def test_twelve_human_responses_return_to_models_without_repairing_dictation() -> None:
|
|
"""Human responses remain verbatim, but repeated invalid dictation cannot trap the model circuit."""
|
|
|
|
local = _Provider("local", *("invalid",) * 12, "model retry")
|
|
human = _Human(
|
|
HumanInterventionResponseDTO("", "", "retry"),
|
|
*(HumanInterventionResponseDTO("", " invalid human\n", "resend"),) * 12,
|
|
)
|
|
workflow = StabilizationEscalation(
|
|
(local,), check_name="Acceptance", prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
event_publisher=human, command_receiver=human,
|
|
)
|
|
tuple(map(lambda _: workflow.complete("judge"), range(12)))
|
|
|
|
replies = tuple(map(lambda _: workflow.complete("correct response"), range(12)))
|
|
|
|
assert all(map(lambda reply: isinstance(reply, HumanResponse), replies))
|
|
assert replies == (" invalid human\n",) * 12
|
|
assert workflow.complete("correct response") == "model retry"
|
|
assert len(local.prompts) == 13
|
|
assert len(human.requests) == 13
|
|
|
|
|
|
def test_suppressing_open_intervention_resumes_models_and_skips_later_prompts(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Suppression from an open screen survives another entire model circuit without asking again."""
|
|
|
|
monkeypatch.setattr(cycle, "intervention_suppressed", False)
|
|
local = _Provider("local", *("invalid",) * 24, "corrected response")
|
|
human = _Human(HumanInterventionResponseDTO("", "unfinished draft", "suppress"))
|
|
workflow = StabilizationEscalation(
|
|
(local,), check_name="Acceptance", prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
event_publisher=human, command_receiver=human,
|
|
)
|
|
|
|
replies = tuple(map(lambda _: workflow.complete("judge"), range(25)))
|
|
|
|
assert replies == ("invalid",) * 24 + ("corrected response",)
|
|
assert len(local.prompts) == 25
|
|
assert len(human.requests) == 1
|
|
assert cycle.intervention_suppressed is True
|
|
|
|
|
|
def test_resume_at_last_cloud_preserves_its_remaining_attempt_then_wraps(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""A resumed last-cloud boundary spends its twelfth response before returning to the lowest tier."""
|
|
|
|
monkeypatch.setattr(cycle, "intervention_suppressed", True)
|
|
repo = MemoryRepoExec()
|
|
graph = Provenance.open(repo_root=repo.root, repo_exec=repo, resume=False)
|
|
item = InspectionBacklog(focused_spotlight=ClassMeta("src/a.py", "", [], {}), summary="repair")
|
|
graph.start(Phase.ALPHA, item)
|
|
store = EscalationAttempts(repo.root, repo)
|
|
store.save(item.message_id, "Acceptance", EscalationState(23, "last cloud response"))
|
|
local = _Provider("local", "next circuit")
|
|
cloud = _Provider("cloud", "twelfth cloud")
|
|
resumed = StabilizationEscalation((local, cloud), check_name="Acceptance", prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
|
|
with provenance.bind(graph), provenance.item_scope(Phase.ALPHA, item):
|
|
assert resumed.complete("same obligation") == "twelfth cloud"
|
|
assert store.load(item.message_id, "Acceptance").attempts == 24
|
|
assert resumed.complete("same obligation") == "next circuit"
|
|
assert store.load(item.message_id, "Acceptance").attempts == 37
|
|
assert local.prompts == cloud.prompts == ["same obligation"]
|
|
|
|
|
|
def test_oversized_stabilization_prompts_keep_token_jacker_routing_without_caching() -> None:
|
|
"""Uncached stabilization retains full oversized prompts and provider token accounting."""
|
|
|
|
from band_aider.ai_client.token_jacker import TokenJacker
|
|
from band_aider.context.token_budget_state import TokenBudgetSettings
|
|
from band_aider.context.band_aider_facts import get_band_aider_facts
|
|
|
|
local = _Provider("local")
|
|
cloud = _Provider("cloud", "first cloud")
|
|
next_cloud = _Provider("next_cloud", *("second cloud",) * 12)
|
|
runtime = TokenJacker(CachedCompletionClient(local), settings=TokenBudgetSettings(max_prompt_tokens=1), escalation_clients=(CachedCompletionClient(cloud), CachedCompletionClient(next_cloud)))
|
|
workflow = StabilizationEscalation.for_stabilization(runtime, check_name="Acceptance", prompt_formatter_builder=MarkdownPromptFormatterBuilder())
|
|
before = get_band_aider_facts().total_tokens_spent
|
|
|
|
assert workflow.complete("complete oversized evidence") == "first cloud"
|
|
assert tuple(map(lambda _: workflow.complete("complete oversized evidence"), range(12))) == ("second cloud",) * 12
|
|
|
|
assert local.prompts == []
|
|
assert cloud.prompts == ["complete oversized evidence"]
|
|
assert next_cloud.prompts == ["complete oversized evidence"] * 12
|
|
assert get_band_aider_facts().total_tokens_spent > before
|
|
|
|
|
|
def test_operator_interrupt_uses_the_shared_human_channel_without_cycle_detection(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""F5 can request an early human hint even though stabilization bypasses CycleJacker."""
|
|
|
|
monkeypatch.setattr(cycle, "intervention_suppressed", True)
|
|
local = _Provider("local")
|
|
cloud = _Provider("cloud", "hinted")
|
|
human = _Human(HumanInterventionResponseDTO("", "operator hint", "resend"))
|
|
workflow = StabilizationEscalation((local, cloud), check_name="Acceptance", prompt_formatter_builder=MarkdownPromptFormatterBuilder(), event_publisher=human, command_receiver=human)
|
|
cycle.human_interrupt = True
|
|
try:
|
|
assert workflow.complete("judge") == "hinted"
|
|
finally:
|
|
cycle.human_interrupt = False
|
|
assert local.prompts == []
|
|
assert cycle.intervention_suppressed is False
|
|
assert len(cloud.prompts) == 1
|
|
assert human.requests[0].mode == "hint"
|
|
|
|
|
|
def test_human_source_survives_the_last_ordinary_json_repair_without_cloud_fallback() -> None:
|
|
"""Even the final repair slot must preserve human text and stop automatic model escalation."""
|
|
|
|
dictated = HumanResponse(" ```json\nnot JSON\n``` ")
|
|
provider = _Provider("local", "malformed", "malformed again", dictated)
|
|
guarded = FenceCrashingClient(JsonGuardingClient(provider, MarkdownPromptFormatterBuilder()))
|
|
with patch("band_aider.ai_client.json_guarding_client.OpenAIProxyFactory.from_configs", side_effect=AssertionError("human dictation triggered fallback")):
|
|
response = guarded.complete("return strict JSON")
|
|
|
|
assert response is dictated
|
|
assert len(provider.prompts) == 3
|