622 lines
25 KiB
Python
622 lines
25 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Semantic keyframes for prompt-local prior-COT authority."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from copy import deepcopy
|
|
import logging
|
|
from pathlib import Path
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from band_aider.ai_client.completion_client import CompletionClient, OpenAIConfig
|
|
from band_aider.ai_client.cached_completion_client import CachedCompletionClient
|
|
from band_aider.ai_client.cycle_jacker import CycleJacker
|
|
from band_aider.ai_client.human_feedback_store import HumanFeedbackStore
|
|
from band_aider.ai_client.token_jacker import TokenJacker
|
|
from band_aider.ai_client.token_budget_estimator import TokenBudgetEstimator
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.context.cot_task_generator import CotTaskRecord
|
|
from band_aider.context.provenance import FutureGuidance
|
|
from band_aider.context import token_budget_state
|
|
from band_aider.context.token_budget_state import TokenBudgetSettings
|
|
from band_aider.judgement.contextual_cot_renderer import ContextualCotRenderer
|
|
from band_aider.judgement.edit_request import (
|
|
AiderEditRequest,
|
|
AiderEditRequestMask,
|
|
AiderFailureMode,
|
|
)
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.stabilization_decision import FutureGuidanceApplicability
|
|
from band_aider.ui.status_publisher import StatusPublisher
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
class _Client(CompletionClient):
|
|
"""Return exact selector responses and retain every selection prompt.
|
|
|
|
Example: ``_Client("NO_KEEP").complete("prompt")`` returns the queued result.
|
|
"""
|
|
|
|
def __init__(self, *responses: str) -> None:
|
|
"""Store the ordered selector responses."""
|
|
|
|
self.responses = list(responses)
|
|
self.prompts: list[str] = []
|
|
|
|
@property
|
|
def inner(self) -> CompletionClient | None:
|
|
"""Expose this fake as the terminal completion client."""
|
|
|
|
return None
|
|
|
|
@property
|
|
def openai_configs(self) -> tuple[OpenAIConfig, ...]:
|
|
"""Expose no escalation configurations for deterministic selection."""
|
|
|
|
return ()
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Record one prompt and return its configured identity response."""
|
|
|
|
del logger
|
|
self.prompts.append(prompt)
|
|
return self.responses.pop(0)
|
|
|
|
|
|
def _record(task: str, location: str) -> CotTaskRecord:
|
|
"""Create one prior COT record with matching source metadata."""
|
|
|
|
return CotTaskRecord(task, location, ClassMeta(location, task, [], {}))
|
|
|
|
|
|
def _request(*records: CotTaskRecord) -> AiderEditRequest:
|
|
"""Create one semantic edit request carrying prior COT candidates."""
|
|
|
|
return AiderEditRequest(
|
|
authoritative_prompt="Repair the stale caller.",
|
|
editable_files=[Path("src/caller.py")],
|
|
readonly_files=[],
|
|
previous_cot_records=records,
|
|
failure_error_message="Memento has no attribute clear.",
|
|
failure_stack_trace=["src/caller.py:18: memento.clear()"],
|
|
)
|
|
|
|
|
|
def test_selected_identity_copies_exact_prior_cot_without_mutating_request() -> None:
|
|
"""A selected identity adds its stored task while the original request stays intact."""
|
|
|
|
remove_clear = _record(
|
|
"Remove Memento.clear; stale callers must stop using it.",
|
|
"src/memento.py",
|
|
)
|
|
unrelated = _record("Update cache ownership.", "src/cache.py")
|
|
request = _request(remove_clear, unrelated)
|
|
original = deepcopy(request)
|
|
client = _Client("KEEP: COT-0000")
|
|
renderer = ContextualCotRenderer(
|
|
client=client,
|
|
current_cot="Adopt managed cache lifecycle.",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
logger=logging.getLogger(__name__),
|
|
)
|
|
|
|
rendered = renderer.render(
|
|
AiderEditRequestMask(request, AiderFailureMode.SEMANTIC),
|
|
"Decide who owns the missing clear failure.",
|
|
)
|
|
|
|
assert "COT-0000" in rendered
|
|
assert remove_clear.task in rendered
|
|
assert unrelated.task not in rendered
|
|
assert "CURRENT COT — PRIMARY AUTHORITY AND EDIT SCOPE" in rendered
|
|
assert "Adopt managed cache lifecycle." in rendered
|
|
assert request == original
|
|
assert "COT-0001" in client.prompts[0]
|
|
|
|
|
|
def test_selector_neutralizes_beta_response_shape_but_preserves_real_beta_prompt() -> None:
|
|
"""Prior-COT selection cannot inherit Beta's response shape, which the real turn retains."""
|
|
|
|
request = _request(_record("Preserve the accepted source contract.", "src/widget.py"))
|
|
client = _Client("KEEP: COT-0000")
|
|
renderer = ContextualCotRenderer(
|
|
client=client,
|
|
current_cot="Add focused regression coverage.",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
logger=logging.getLogger(__name__),
|
|
)
|
|
baseline = (
|
|
"BETA TEST PROMPT REFINEMENT\n"
|
|
'Return strict JSON: {"action":"adapt","beta_instruction":"repair test"}.'
|
|
)
|
|
|
|
rendered = renderer.render(
|
|
AiderEditRequestMask(request, AiderFailureMode.SEMANTIC),
|
|
baseline,
|
|
)
|
|
|
|
assert '{"action":"adapt"' not in client.prompts[0]
|
|
assert 'BETA TEST PROMPT REFINEMENT\n---REDACTED---' in client.prompts[0]
|
|
assert '"action"' not in client.prompts[0]
|
|
assert 'beta_instruction' not in client.prompts[0]
|
|
assert 'Return strict JSON:' not in client.prompts[0]
|
|
assert 'KEEP:' in client.prompts[0]
|
|
assert baseline in rendered
|
|
|
|
|
|
def test_selection_reuses_identical_turn_and_refreshes_for_new_failure() -> None:
|
|
"""An exact turn reuses selection while changed failure evidence gets a fresh gate."""
|
|
|
|
request = _request(_record("Remove Memento.clear.", "src/memento.py"))
|
|
client = _Client("NO_KEEP", "KEEP: COT-0000")
|
|
renderer = ContextualCotRenderer(
|
|
client=client,
|
|
current_cot="Repair the cache.",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
logger=logging.getLogger(__name__),
|
|
)
|
|
mask = AiderEditRequestMask(request, AiderFailureMode.SEMANTIC)
|
|
|
|
first = renderer.render(mask, "Failure: cache key mismatch.")
|
|
repeated = renderer.render(mask, "Failure: cache key mismatch.")
|
|
changed = renderer.render(mask, "Failure: Memento.clear is absent.")
|
|
|
|
assert first == "Failure: cache key mismatch."
|
|
assert repeated == first
|
|
assert "Remove Memento.clear." in changed
|
|
assert len(client.prompts) == 2
|
|
|
|
|
|
@pytest.mark.parametrize("batch_target_ratio, estimated_tokens", ((0.5, 400), (0.5, 401), (0.75, 600), (0.75, 601)))
|
|
def test_selection_includes_all_candidates_at_or_above_batch_target(
|
|
batch_target_ratio: float,
|
|
estimated_tokens: int,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Expensive COT judgement retains every candidate without a model response."""
|
|
|
|
settings = TokenBudgetSettings(
|
|
max_prompt_tokens=1000, max_output_tokens=200, batch_target_ratio=batch_target_ratio,
|
|
)
|
|
monkeypatch.setattr(token_budget_state, "get_token_budget_settings", lambda: settings)
|
|
request = _request(
|
|
_record("Prior zero constraint.", "src/zero.py"),
|
|
_record("Prior one constraint.", "src/one.py"),
|
|
)
|
|
original = deepcopy(request)
|
|
client = _Client()
|
|
estimator = Mock(spec=TokenBudgetEstimator)
|
|
estimator.estimate.return_value = estimated_tokens
|
|
renderer = ContextualCotRenderer(
|
|
client=client,
|
|
current_cot="Repair the current source.",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
logger=logging.getLogger(__name__),
|
|
estimator=estimator,
|
|
)
|
|
|
|
rendered = renderer.render(
|
|
AiderEditRequestMask(request, AiderFailureMode.SEMANTIC),
|
|
"Decide who owns the current failure.",
|
|
)
|
|
|
|
assert "Prior zero constraint." in rendered
|
|
assert "Prior one constraint." in rendered
|
|
assert rendered.index("COT-0000") < rendered.index("COT-0001")
|
|
assert "CURRENT COT — PRIMARY AUTHORITY AND EDIT SCOPE" in rendered
|
|
assert "Repair the current source." in rendered
|
|
assert request == original
|
|
assert client.prompts == []
|
|
|
|
|
|
@pytest.mark.parametrize("batch_target_ratio, estimated_tokens", ((0.5, 399), (0.75, 599)))
|
|
def test_selection_below_batch_target_still_uses_model_judgement(
|
|
batch_target_ratio: float,
|
|
estimated_tokens: int,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A cheap COT judgement can still reject every prior candidate."""
|
|
|
|
settings = TokenBudgetSettings(
|
|
max_prompt_tokens=1000, max_output_tokens=200, batch_target_ratio=batch_target_ratio,
|
|
)
|
|
monkeypatch.setattr(token_budget_state, "get_token_budget_settings", lambda: settings)
|
|
request = _request(
|
|
_record("Prior zero constraint.", "src/zero.py"),
|
|
_record("Prior one constraint.", "src/one.py"),
|
|
)
|
|
client = _Client("NO_KEEP")
|
|
estimator = Mock(spec=TokenBudgetEstimator)
|
|
estimator.estimate.return_value = estimated_tokens
|
|
renderer = ContextualCotRenderer(
|
|
client=client,
|
|
current_cot="Repair the current source.",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
logger=logging.getLogger(__name__),
|
|
estimator=estimator,
|
|
)
|
|
baseline = "Decide who owns the current failure."
|
|
|
|
rendered = renderer.render(AiderEditRequestMask(request, AiderFailureMode.SEMANTIC), baseline)
|
|
|
|
assert rendered == baseline
|
|
assert len(client.prompts) == 1
|
|
assert "Prior zero constraint." in client.prompts[0]
|
|
assert "Prior one constraint." in client.prompts[0]
|
|
|
|
|
|
def test_combined_catalog_uses_budget_inclusion_instead_of_smaller_model_calls(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""A costly full catalog must not turn into several affordable selector calls."""
|
|
|
|
settings = TokenBudgetSettings(max_prompt_tokens=2000, max_output_tokens=1000, batch_target_ratio=1.0)
|
|
monkeypatch.setattr(token_budget_state, "get_token_budget_settings", lambda: settings)
|
|
first = _record("FIRST_CONSTRAINT " * 400, "src/first.py")
|
|
second = _record("SECOND_CONSTRAINT " * 400, "src/second.py")
|
|
single_client = _Client("NO_KEEP")
|
|
combined_client = _Client()
|
|
single = ContextualCotRenderer(
|
|
client=single_client,
|
|
current_cot="Repair the current source.",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
logger=logging.getLogger(__name__),
|
|
estimator=TokenBudgetEstimator(tokens_per_word=1.0),
|
|
)
|
|
combined = ContextualCotRenderer(
|
|
client=combined_client,
|
|
current_cot="Repair the current source.",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
logger=logging.getLogger(__name__),
|
|
estimator=TokenBudgetEstimator(tokens_per_word=1.0),
|
|
)
|
|
baseline = "Decide who owns the current failure."
|
|
|
|
single_rendered = single.render(AiderEditRequestMask(_request(first), AiderFailureMode.SEMANTIC), baseline)
|
|
combined_rendered = combined.render(AiderEditRequestMask(_request(first, second), AiderFailureMode.SEMANTIC), baseline)
|
|
|
|
assert single_rendered == baseline
|
|
assert len(single_client.prompts) == 1
|
|
assert first.task in combined_rendered
|
|
assert second.task in combined_rendered
|
|
assert combined_client.prompts == []
|
|
|
|
|
|
def test_no_prior_cots_preserves_prompt_without_model_judgement() -> None:
|
|
"""An empty prior catalog needs no selection or extra authority."""
|
|
|
|
request = _request()
|
|
client = _Client()
|
|
renderer = ContextualCotRenderer(
|
|
client=client,
|
|
current_cot="Repair the current source.",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
logger=logging.getLogger(__name__),
|
|
)
|
|
baseline = "Decide who owns the current failure."
|
|
|
|
rendered = renderer.render(AiderEditRequestMask(request, AiderFailureMode.SEMANTIC), baseline)
|
|
|
|
assert rendered == baseline
|
|
assert client.prompts == []
|
|
|
|
|
|
def test_unknown_identities_retry_with_rejection_and_valid_menu(
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""Invalid identities remain unadmitted until corrected with the supplied menu."""
|
|
|
|
request = _request(_record("Remove Memento.clear.", "src/memento.py"))
|
|
client = _Client(
|
|
"KEEP: COT-9999",
|
|
"KEEP: COT-9999",
|
|
"KEEP: COT-9999",
|
|
"KEEP: COT-0000\nREASON: The removed API constrains caller repair.",
|
|
)
|
|
renderer = ContextualCotRenderer(
|
|
client=client,
|
|
current_cot="Repair the cache.",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
logger=logging.getLogger(__name__),
|
|
)
|
|
|
|
with caplog.at_level(logging.INFO):
|
|
rendered = renderer.render(
|
|
AiderEditRequestMask(request, AiderFailureMode.SEMANTIC),
|
|
"Failure: Memento.clear is absent.",
|
|
)
|
|
|
|
assert "Remove Memento.clear." in rendered
|
|
assert len(client.prompts) == 4
|
|
assert client.prompts[0] != client.prompts[1]
|
|
assert "unknown prior COT identities" in client.prompts[-1]
|
|
assert "Your previous selector response was rejected." in client.prompts[-1]
|
|
menu = client.prompts[-1].split("==== BEGIN REQUIRED RESPONSE FORMAT ====", 1)[1]
|
|
assert "KEEP: COT-0000" in menu
|
|
assert "COT-9999" not in menu
|
|
assert "NO_KEEP" in menu
|
|
assert "REASON:" in menu
|
|
assert caplog.text.count("selector rejected response:") == 3
|
|
assert "Rejected output:\nKEEP: COT-9999" in caplog.text
|
|
assert "reason=The removed API constrains caller repair." in caplog.text
|
|
|
|
|
|
def test_empty_response_gets_feedback_before_explicit_no_keep(
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""An empty reply must be corrected before the renderer admits no prior authority."""
|
|
|
|
request = _request(_record("Remove Memento.clear.", "src/memento.py"))
|
|
client = _Client("", "NO_KEEP\nREASON: This failure concerns an unrelated cache key.")
|
|
renderer = ContextualCotRenderer(
|
|
client=client,
|
|
current_cot="Repair the cache.",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
logger=logging.getLogger(__name__),
|
|
)
|
|
|
|
with caplog.at_level(logging.INFO):
|
|
rendered = renderer.render(
|
|
AiderEditRequestMask(request, AiderFailureMode.SEMANTIC),
|
|
"Failure: cache key mismatch.",
|
|
)
|
|
|
|
assert rendered == "Failure: cache key mismatch."
|
|
assert "SELECTOR VALIDATION ERROR" in client.prompts[1]
|
|
assert "Your previous selector response was rejected." in client.prompts[1]
|
|
assert "KEEP: COT-0000" in client.prompts[1]
|
|
assert "reason=This failure concerns an unrelated cache key." in caplog.text
|
|
|
|
|
|
@pytest.mark.parametrize("repeat_threshold", (1, 3, 5))
|
|
def test_cached_invalid_selection_reaches_last_cloud_model(
|
|
repeat_threshold: int,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""Stationary rejected replies reach CycleJacker's last tier before selection completes."""
|
|
|
|
invalid = '{"outcome":"source","additional":{"test_work_needed":""}}'
|
|
primary = _Client(invalid, invalid)
|
|
cloud_one = _Client(*(invalid,) * repeat_threshold)
|
|
cloud_two = _Client("KEEP: COT-0000\nREASON: Preserve the removed API contract.")
|
|
repo_exec = MemoryRepoExec()
|
|
client = CycleJacker(
|
|
inner=CachedCompletionClient(primary),
|
|
feedback_store=HumanFeedbackStore(repo_exec.root, repo_exec=repo_exec),
|
|
escalation_clients=(cloud_one, cloud_two),
|
|
repeat_threshold=repeat_threshold,
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
)
|
|
request = _request(_record("Remove Memento.clear.", "src/memento.py"))
|
|
renderer = ContextualCotRenderer(
|
|
client=client,
|
|
current_cot="Return inert Memento descriptors.",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
logger=logging.getLogger(__name__),
|
|
)
|
|
|
|
with caplog.at_level(logging.INFO):
|
|
rendered = renderer.render(
|
|
AiderEditRequestMask(request, AiderFailureMode.SEMANTIC),
|
|
'Return strict JSON: {"outcome":"source|test"}.',
|
|
)
|
|
|
|
assert "Remove Memento.clear." in rendered
|
|
assert len(primary.prompts) == 2
|
|
assert len(cloud_one.prompts) == repeat_threshold
|
|
assert len(cloud_two.prompts) == 1
|
|
assert primary.prompts[0] != primary.prompts[1]
|
|
assert cloud_two.prompts[0] == primary.prompts[1]
|
|
assert "embedded task is not a COT selection" in cloud_two.prompts[0]
|
|
assert "KEEP: COT-0000" in cloud_two.prompts[0]
|
|
assert "CachedCompletionClient returned a cached response." in caplog.text
|
|
assert "CycleJacker detected repeated completion" in caplog.text
|
|
|
|
|
|
def test_oversized_selection_skips_model_but_downstream_judgement_uses_overflow(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Budget inclusion skips only COT judgement; the downstream model still runs."""
|
|
|
|
settings = TokenBudgetSettings(max_prompt_tokens=1)
|
|
monkeypatch.setattr(token_budget_state, "get_token_budget_settings", lambda: settings)
|
|
primary = _Client()
|
|
cloud = _Client("routed source")
|
|
client = TokenJacker(
|
|
inner=primary,
|
|
settings=settings,
|
|
escalation_clients=(cloud,),
|
|
event_publisher=Mock(spec=StatusPublisher),
|
|
)
|
|
first = _record("Remove Memento.clear.", "src/memento.py")
|
|
second = _record("Callers own managed Memento scopes.", "src/caller.py")
|
|
request = _request(first, second)
|
|
renderer = ContextualCotRenderer(
|
|
client=client,
|
|
current_cot="Repair the cache lifecycle.",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
logger=logging.getLogger(__name__),
|
|
)
|
|
|
|
rendered = renderer.render(
|
|
AiderEditRequestMask(request, AiderFailureMode.SEMANTIC),
|
|
"Decide who owns the stale clear call.",
|
|
)
|
|
assert cloud.prompts == []
|
|
result = client.complete(rendered)
|
|
|
|
assert result == "routed source"
|
|
assert primary.prompts == []
|
|
assert len(cloud.prompts) == 1
|
|
assert first.task in cloud.prompts[0]
|
|
assert second.task in cloud.prompts[0]
|
|
assert cloud.prompts[0] == rendered
|
|
|
|
|
|
@pytest.mark.parametrize("cloud_count", (0, 1, 2))
|
|
@pytest.mark.parametrize("repeat_threshold", (1, 3))
|
|
def test_correction_continues_beyond_configured_cycle_allowance(
|
|
cloud_count: int,
|
|
repeat_threshold: int,
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
"""A complete model circuit cannot authorize returning without a valid selection."""
|
|
|
|
retries = repeat_threshold * (cloud_count + 2)
|
|
primary = _Client(*map(lambda index: f"invalid answer {index}", range(retries + 1)), "KEEP: COT-0000")
|
|
clouds = tuple(_Client() for _index in range(cloud_count))
|
|
repo_exec = MemoryRepoExec()
|
|
cycle_client = CycleJacker(
|
|
inner=primary,
|
|
feedback_store=HumanFeedbackStore(repo_exec.root, repo_exec=repo_exec),
|
|
escalation_clients=clouds,
|
|
repeat_threshold=repeat_threshold,
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
)
|
|
client = TokenJacker(inner=cycle_client, settings=TokenBudgetSettings())
|
|
request = _request(_record("Remove Memento.clear.", "src/memento.py"))
|
|
original = deepcopy(request)
|
|
renderer = ContextualCotRenderer(
|
|
client=client,
|
|
current_cot="Return inert descriptors.",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
logger=logging.getLogger(__name__),
|
|
)
|
|
baseline = "Route the stale clear caller for repair."
|
|
|
|
with caplog.at_level(logging.INFO):
|
|
rendered = renderer.render(AiderEditRequestMask(request, AiderFailureMode.SEMANTIC), baseline)
|
|
|
|
assert "Remove Memento.clear." in rendered
|
|
assert request == original
|
|
assert len(primary.prompts) == retries + 2
|
|
assert "exhausted" not in caplog.text
|
|
assert caplog.text.count("selector rejected response:") == retries + 1
|
|
|
|
|
|
def test_selection_correction_at_batch_target_includes_all_without_another_call(
|
|
caplog: pytest.LogCaptureFixture,
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
"""Correction growth retains every COT before another model call can be spent."""
|
|
|
|
settings = TokenBudgetSettings(max_prompt_tokens=1000, max_output_tokens=200, batch_target_ratio=0.5)
|
|
monkeypatch.setattr(token_budget_state, "get_token_budget_settings", lambda: settings)
|
|
first = _record("Prior zero constraint.", "src/zero.py")
|
|
second = _record("Prior one constraint.", "src/one.py")
|
|
request = _request(first, second)
|
|
original = deepcopy(request)
|
|
client = _Client("KEEP: COT-9999")
|
|
estimator = Mock(spec=TokenBudgetEstimator)
|
|
estimator.estimate.side_effect = (399, 400)
|
|
renderer = ContextualCotRenderer(
|
|
client=client,
|
|
current_cot="Repair the caller.",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
logger=logging.getLogger(__name__),
|
|
estimator=estimator,
|
|
)
|
|
mask = AiderEditRequestMask(request, AiderFailureMode.SEMANTIC)
|
|
baseline = "Route the current failure."
|
|
|
|
with caplog.at_level(logging.INFO):
|
|
corrected = renderer.render(mask, baseline)
|
|
recovered = renderer.render(mask, baseline)
|
|
|
|
assert corrected == recovered
|
|
assert len(client.prompts) == 1
|
|
assert first.task in recovered
|
|
assert second.task in recovered
|
|
assert "COT-9999" not in recovered
|
|
assert request == original
|
|
assert "unknown prior COT identities" in caplog.text
|
|
|
|
|
|
class _DecisionFileManager(FileManager):
|
|
"""Supply real decision path normalization; use _DecisionFileManager()."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Bind the repository, governing goal, and formatter used in this scenario."""
|
|
self.repo_root = Path("/testbed")
|
|
self.prompt = "Repair the cache lifecycle."
|
|
self.prompt_formatter_builder = MarkdownPromptFormatterBuilder()
|
|
|
|
|
|
def test_stabilization_decision_renders_selected_prior_cot_for_its_llm_turn() -> None:
|
|
"""The shared decision boundary adds prior authority before child completion."""
|
|
|
|
request = _request(_record("Remove Memento.clear.", "src/memento.py"))
|
|
client = _Client(
|
|
"KEEP: COT-0000",
|
|
'{"applicable": false, "reason": "The observation is unrelated."}',
|
|
)
|
|
file_manager = _DecisionFileManager()
|
|
decision = FutureGuidanceApplicability(
|
|
client=client,
|
|
file_manager=file_manager,
|
|
request=request,
|
|
future_guidance=FutureGuidance(
|
|
origin_message_id="later",
|
|
addressed_message_id="earlier",
|
|
message="A later observation.",
|
|
justification="Later evidence.",
|
|
),
|
|
current_prompt="Decide whether the caller owns the missing clear failure.",
|
|
logger=logging.getLogger(__name__),
|
|
)
|
|
|
|
assert decision() is False
|
|
assert "CONTEXTUAL COT SELECTION" in client.prompts[0]
|
|
assert "Remove Memento.clear." in client.prompts[1]
|
|
assert "CURRENT COT — PRIMARY AUTHORITY AND EDIT SCOPE" in client.prompts[1]
|
|
|
|
|
|
@pytest.mark.parametrize("estimated_tokens, selector_calls", ((1, 1), (10**30, 0)))
|
|
def test_acceptance_projection_removes_old_diagnosis_without_mutating_stored_cots(
|
|
estimated_tokens: int,
|
|
selector_calls: int,
|
|
) -> None:
|
|
"""Model selection and budget inclusion preserve the same acceptance-only projection."""
|
|
|
|
from band_aider.judgement import contextual_cot_renderer
|
|
|
|
task = "\n".join((
|
|
"Task introduction", "FILE-OWNED CONTRACT", "Honor the inert lifecycle.",
|
|
"CURRENT SOURCE EVIDENCE", "HISTORICAL_CLEAR_CALL",
|
|
"REQUIRED CHANGES", "OLD_REPAIR_INSTRUCTION", "COORDINATION / GUARDRAILS",
|
|
"Preserve callers.", "VERIFICATION", "Verify the lifecycle.",
|
|
))
|
|
record = _record(task, "src/memento.py")
|
|
request = _request(record)
|
|
client = _Client("KEEP: COT-0000\nREASON: lifecycle contract")
|
|
estimator = Mock(spec=TokenBudgetEstimator)
|
|
estimator.estimate.return_value = estimated_tokens
|
|
renderer = ContextualCotRenderer(
|
|
client=client, current_cot="CURRENT_CONTRACT", prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
logger=logging.getLogger(__name__), task_projection=contextual_cot_renderer.acceptance_cot,
|
|
current_cot_in_prompt=True, estimator=estimator,
|
|
)
|
|
|
|
rendered = renderer.render(AiderEditRequestMask(request, AiderFailureMode.SEMANTIC), "CURRENT_CONTRACT\nReturn pass or fail.")
|
|
|
|
assert rendered.count("CURRENT_CONTRACT") == 1
|
|
assert "Honor the inert lifecycle." in rendered
|
|
assert "Preserve callers." in rendered
|
|
assert "Verify the lifecycle." in rendered
|
|
assert "HISTORICAL_CLEAR_CALL" not in rendered
|
|
assert "OLD_REPAIR_INSTRUCTION" not in rendered
|
|
assert len(client.prompts) == selector_calls
|
|
assert all(map(lambda prompt: "HISTORICAL_CLEAR_CALL" not in prompt, client.prompts))
|
|
assert record.task == task
|
|
ordinary = ContextualCotRenderer(
|
|
client=_Client("KEEP: COT-0000\nREASON: lifecycle contract"), current_cot="CURRENT_CONTRACT",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(), logger=logging.getLogger(__name__),
|
|
)
|
|
assert task in ordinary.render(AiderEditRequestMask(request, AiderFailureMode.SEMANTIC), "Judge work")
|
|
assert contextual_cot_renderer.acceptance_cot("Unstructured exact task") == "Unstructured exact task"
|