967 lines
42 KiB
Python
967 lines
42 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Verify phase-start provenance survives restart and independent recovery."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import replace
|
|
import logging
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import cast
|
|
from unittest.mock import Mock
|
|
|
|
import pytest
|
|
|
|
from band_aider.ai_client.completion_client import CompletionClient, OpenAIConfig
|
|
from band_aider.ai_client.escalation_attempts import EscalationAttempts, EscalationState
|
|
from band_aider.context.class_meta import ClassMeta
|
|
from band_aider.context.failing_test import FailingTest
|
|
from band_aider.context import provenance
|
|
from band_aider.context.provenance import DeadEndError, Direction, Phase, Provenance
|
|
from band_aider.context.working_backlog import (
|
|
FailedQAItem,
|
|
InspectionBacklog,
|
|
TestingBacklog as BetaWork,
|
|
)
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.code_editor import EditorRecord
|
|
from band_aider.judgement.dead_end_decision import DeadEndDecisionBuilder
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.bi_dependency_discovery_strategy import BiDependencyDiscoveryStrategy
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
|
|
from band_aider.loop.alpha_loop import AlphaLoop
|
|
from band_aider.loop.beta_loop import BetaLoop
|
|
from tests.support.memory_path import MemoryPath
|
|
|
|
|
|
def _inspection(location: str = "src/focus.py") -> InspectionBacklog:
|
|
"""Build one small Alpha phase-start item with no external collaborators."""
|
|
|
|
return InspectionBacklog(
|
|
focused_spotlight=ClassMeta(location, "focus", [], {}),
|
|
summary="inspect the focus",
|
|
)
|
|
|
|
|
|
def _testing(location: str = "tests/test_focus.py") -> BetaWork:
|
|
"""Build one small Beta phase-start item with no external collaborators."""
|
|
|
|
return BetaWork(
|
|
focused_spotlight=ClassMeta(location, "focus test", [], {}),
|
|
summary="test the focus",
|
|
)
|
|
|
|
|
|
def _graph(root: MemoryPath, *, resume: bool = False) -> Provenance:
|
|
"""Open a fresh or resumed repository-local graph through the real adapter."""
|
|
|
|
return Provenance.open(repo_root=root, repo_exec=root.repository, resume=resume)
|
|
|
|
|
|
def test_same_message_retry_preserves_escalation_and_history(tmp_path: MemoryPath) -> None:
|
|
"""Retrying work survives hydration without a new identity, self-edge, or reset attempt count."""
|
|
|
|
graph = _graph(tmp_path)
|
|
item = _testing()
|
|
original = graph.start(Phase.BETA, item)
|
|
attempts = EscalationAttempts(tmp_path, tmp_path.repository)
|
|
decision = "TargetedAlphaOrBetaRoutingDecision/tests/test_focus.py"
|
|
attempts.save(original, decision, EscalationState(attempts=12))
|
|
retry = replace(item, previous_beta_guidance="Repair the setup; production still blocks completion.",
|
|
editor_records=[EditorRecord("retained repair", "before", "after")])
|
|
before = tuple(graph.messages), graph.edges
|
|
|
|
assert graph.track_provenance(original, Phase.BETA, retry) == original
|
|
recovered = _graph(tmp_path, resume=True)
|
|
assert recovered.start(Phase.BETA, retry) == original
|
|
assert (tuple(recovered.messages), recovered.edges) == before
|
|
assert attempts.load(retry.message_id, decision).attempts == 12
|
|
assert retry.editor_records == [EditorRecord("retained repair", "before", "after")]
|
|
|
|
# A distinct new work item still starts a distinct message even with identical file evidence.
|
|
fresh = _testing()
|
|
assert recovered.start(Phase.BETA, fresh) != original
|
|
assert attempts.load(fresh.message_id, decision).attempts == 0
|
|
|
|
|
|
class _Client(CompletionClient):
|
|
"""Return ordered boundary-decision responses without contacting an endpoint."""
|
|
|
|
def __init__(self, *responses: object) -> None:
|
|
"""Store the complete response sequence consumed by one boundary check."""
|
|
|
|
self.responses = list(responses)
|
|
self.prompts: list[str] = []
|
|
|
|
@property
|
|
def inner(self) -> CompletionClient | None:
|
|
"""Return no nested client because this fake is the terminal completion boundary."""
|
|
|
|
return None
|
|
|
|
@property
|
|
def openai_configs(self) -> tuple[OpenAIConfig, ...]:
|
|
"""Expose no escalation configurations for fixed test responses."""
|
|
|
|
return ()
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Record one prompt and return the next configured JSON object."""
|
|
|
|
del logger
|
|
self.prompts.append(prompt)
|
|
return json.dumps(self.responses.pop(0))
|
|
|
|
|
|
class _DecisionFileManager(FileManager):
|
|
"""Supply real path normalization for provenance decisions; use _DecisionFileManager()."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Bind repository identity and the decision's prompt formatting contract."""
|
|
self.repo_root = Path("/testbed")
|
|
self.prompt_formatter_builder = MarkdownPromptFormatterBuilder()
|
|
self.prompt = "Preserve the selected contract."
|
|
|
|
|
|
@pytest.mark.parametrize("trace, error", ((["old()"], "old failed"), ([], "old failed"), (["old()"], "")))
|
|
def test_duplicate_remediation_finds_a_distinct_progressed_representation_after_resume(
|
|
tmp_path: MemoryPath, trace: list[str], error: str,
|
|
) -> None:
|
|
"""A matching remediation on another branch survives resume after reaching Beta."""
|
|
|
|
graph = _graph(tmp_path)
|
|
source = ClassMeta("src/focus.py", "", [], {})
|
|
failure = FailingTest(ClassMeta("tests/earlier.py", "", [], {}), trace, error)
|
|
root = graph.start(Phase.ALPHA, _inspection("src/root.py"))
|
|
earlier = FailedQAItem(source, "repair old call", failed_test=failure)
|
|
earlier_id = graph.track_provenance(root, Phase.ALPHA, earlier)
|
|
graph.track_provenance(earlier_id, Phase.BETA, _testing())
|
|
current = FailedQAItem(
|
|
source, "another test reports old call",
|
|
failed_test=FailingTest(ClassMeta("tests/later.py", "", [], {}), trace, error),
|
|
)
|
|
current_id = graph.track_provenance(root, Phase.ALPHA, current)
|
|
restored = _graph(tmp_path, resume=True)
|
|
|
|
match = restored.duplicate_remediation(current_id)
|
|
|
|
assert match is not None
|
|
assert match.message_id == earlier_id
|
|
assert restored.start(Phase.ALPHA, earlier) == earlier_id
|
|
assert restored.duplicate_remediation(earlier_id) is None
|
|
|
|
|
|
@pytest.mark.parametrize("invalid", (
|
|
"inspection", "other_file", "other_error", "other_trace", "other_format",
|
|
"blank_failure", "no_failure", "terminal", "restart",
|
|
))
|
|
def test_duplicate_remediation_excludes_nonqualifying_retained_records(
|
|
tmp_path: MemoryPath, invalid: str,
|
|
) -> None:
|
|
"""Retention alone cannot replace progression, focused file, or complete failure identity."""
|
|
|
|
graph = _graph(tmp_path)
|
|
source = ClassMeta("src/focus.py", "", [], {})
|
|
failure = FailingTest(ClassMeta("tests/focus.py", "", [], {}), ["old()"], "failed")
|
|
root = graph.start(Phase.ALPHA, InspectionBacklog(source, "root"))
|
|
earlier: InspectionBacklog = FailedQAItem(source, "repair", failed_test=failure)
|
|
# Vary one admission prerequisite while retaining the historical record.
|
|
if invalid == "inspection":
|
|
earlier = InspectionBacklog(source, "inspect")
|
|
elif invalid == "other_file":
|
|
earlier = FailedQAItem(ClassMeta("src/other.py", "", [], {}), "repair", failed_test=failure)
|
|
# The same traceback with a different error does not name the same failure.
|
|
elif invalid == "other_error":
|
|
earlier = FailedQAItem(source, "repair", failed_test=FailingTest(failure.test, ["old()"], "different"))
|
|
# The same generic error with a different traceback is also insufficient.
|
|
elif invalid == "other_trace":
|
|
earlier = FailedQAItem(source, "repair", failed_test=FailingTest(failure.test, ["other()"], "failed"))
|
|
elif invalid == "other_format":
|
|
earlier = FailedQAItem(source, "repair", failed_test=FailingTest(failure.test, [" old()"], "failed"))
|
|
elif invalid == "blank_failure":
|
|
earlier = FailedQAItem(source, "repair", failed_test=FailingTest(failure.test, ["", " "], " "))
|
|
elif invalid == "no_failure":
|
|
earlier = FailedQAItem(source, "repair", failed_test=None)
|
|
earlier_id = graph.track_provenance(root, Phase.ALPHA, earlier)
|
|
# Only a normal transition into another phase establishes completed progression.
|
|
if invalid not in {"terminal", "restart"}:
|
|
graph.track_provenance(earlier_id, Phase.BETA, _testing())
|
|
elif invalid == "restart":
|
|
graph.revive(DeadEndError(
|
|
direction=Direction.ROLLBACK, current_message_id=earlier_id,
|
|
message="return to root", justification="wrong branch",
|
|
))
|
|
current = graph.track_provenance(
|
|
root,
|
|
Phase.ALPHA,
|
|
FailedQAItem(source, "repair again", failed_test=failure),
|
|
)
|
|
|
|
assert graph.duplicate_remediation(current) is None
|
|
|
|
|
|
def test_duplicate_remediation_selects_newest_progressed_match_across_branches(tmp_path: MemoryPath) -> None:
|
|
"""The newest matching progressed representation wins without an ancestry requirement."""
|
|
|
|
graph = _graph(tmp_path)
|
|
source = ClassMeta("src/focus.py", "", [], {})
|
|
failure = FailingTest(ClassMeta("tests/focus.py", "", [], {}), ["old()"], "failed")
|
|
root = graph.start(Phase.ALPHA, _inspection("src/root.py"))
|
|
first = graph.track_provenance(root, Phase.ALPHA, FailedQAItem(source, "first repair", failed_test=failure))
|
|
graph.track_provenance(first, Phase.BETA, _testing("tests/first.py"))
|
|
newest = graph.track_provenance(root, Phase.ALPHA, FailedQAItem(source, "second repair", failed_test=failure))
|
|
graph.track_provenance(newest, Phase.BETA, _testing("tests/second.py"))
|
|
current = graph.track_provenance(root, Phase.ALPHA, FailedQAItem(source, "later repair", failed_test=failure))
|
|
|
|
match = graph.duplicate_remediation(current)
|
|
|
|
assert match is not None
|
|
assert match.message_id == newest
|
|
|
|
|
|
@pytest.mark.parametrize("current_kind", ("blank", "inspection", "no_failure"))
|
|
def test_empty_or_nonremediation_current_work_cannot_match_history(
|
|
tmp_path: MemoryPath, current_kind: str,
|
|
) -> None:
|
|
"""Equal blank failures and current inspection cannot borrow prior admission."""
|
|
|
|
graph = _graph(tmp_path)
|
|
source = ClassMeta("src/focus.py", "", [], {})
|
|
failure = FailingTest(ClassMeta("tests/focus.py", "", [], {}), ["", " "], " ")
|
|
earlier = graph.start(Phase.ALPHA, FailedQAItem(source, "repair", failed_test=failure))
|
|
graph.track_provenance(earlier, Phase.BETA, _testing())
|
|
candidates = {
|
|
"blank": FailedQAItem(source, "repeat", failed_test=failure),
|
|
"inspection": InspectionBacklog(source, "inspect"),
|
|
"no_failure": FailedQAItem(source, "repair", failed_test=None),
|
|
}
|
|
current = graph.start(Phase.ALPHA, candidates[current_kind])
|
|
|
|
assert graph.duplicate_remediation(current) is None
|
|
|
|
|
|
@pytest.mark.parametrize("progressed", [False, True])
|
|
@pytest.mark.parametrize("direction", [Direction.ROLLBACK, Direction.ROLLFORWARD])
|
|
def test_revived_remediation_uses_prior_handoff_after_resume(
|
|
tmp_path: MemoryPath, progressed: bool, direction: Direction,
|
|
) -> None:
|
|
"""Only a historical ordinary Beta handoff makes the fresh revival duplicate-eligible."""
|
|
|
|
graph = _graph(tmp_path)
|
|
root = graph.start(Phase.ALPHA, _inspection("src/root.py"))
|
|
failure = FailingTest(ClassMeta("tests/focus.py", "", [], {}), ["old()"], "old failed")
|
|
work = FailedQAItem(ClassMeta("src/focus.py", "", [], {}), "repair", failed_test=failure)
|
|
original = graph.track_provenance(root, Phase.ALPHA, work)
|
|
original_payload = graph.messages[original].payload
|
|
# Changing only the successor's phase distinguishes progress from another source obligation.
|
|
successor = graph.track_provenance(
|
|
original, Phase.BETA if progressed else Phase.ALPHA,
|
|
_testing() if progressed else _inspection("src/later.py"),
|
|
)
|
|
# Rollforward selects the same original work from one further descendant.
|
|
current = successor if direction is Direction.ROLLBACK else graph.track_provenance(
|
|
successor, Phase.ALPHA, _inspection("src/last.py"),
|
|
)
|
|
|
|
_, revived = graph.revive(DeadEndError(
|
|
direction=direction, current_message_id=current,
|
|
message="Recheck the source.", justification="The later attempt could not proceed.",
|
|
))
|
|
recovered = _graph(tmp_path, resume=True)
|
|
match = recovered.duplicate_remediation(revived.message_id)
|
|
|
|
assert revived.message_id not in {root, original, successor, current}
|
|
assert recovered.messages[revived.message_id].parent_id == root
|
|
assert (match.message_id if match else None) == (original if progressed else None)
|
|
assert recovered.duplicate_remediation(original) is None
|
|
assert recovered.messages[original].payload == original_payload
|
|
assert recovered.start(Phase.ALPHA, revived) == revived.message_id
|
|
payload = json.loads(recovered.messages[revived.message_id].payload)
|
|
assert payload["failed_test"] == json.loads(original_payload)["failed_test"]
|
|
assert payload["future_guidance"][-1]["addressed_message_id"] == revived.message_id
|
|
assert payload["future_guidance"][-1]["origin_message_id"] == current
|
|
assert recovered.messages[original].parent_id == root
|
|
assert any(edge.from_id == root and edge.to_id == original for edge in recovered.edges)
|
|
assert tuple(edge.from_id for edge in recovered.edges if edge.to_id == revived.message_id) == (root,)
|
|
assert recovered.edges[-1].from_id == root
|
|
assert recovered.edges[-1].to_id == revived.message_id
|
|
assert recovered.edges[-1].direction is direction
|
|
assert recovered.edges[-1].guidance == revived.future_guidance[-1]
|
|
|
|
|
|
def test_reviving_beta_does_not_manufacture_alpha_progress(tmp_path: MemoryPath) -> None:
|
|
"""A preserved Alpha parent is lineage, not a new ordinary handoff to revived Beta."""
|
|
|
|
graph = _graph(tmp_path)
|
|
failure = FailingTest(ClassMeta("tests/focus.py", "", [], {}), ["old()"], "old failed")
|
|
source = ClassMeta("src/focus.py", "", [], {})
|
|
original = graph.start(Phase.ALPHA, FailedQAItem(source, "repair", failed_test=failure))
|
|
beta = graph.start(Phase.BETA, _testing(), parent_id=original, direction=Direction.ROLLBACK)
|
|
later = graph.track_provenance(beta, Phase.ALPHA, _inspection("src/later.py"))
|
|
|
|
_, revived = graph.revive(DeadEndError(
|
|
direction=Direction.ROLLBACK, current_message_id=later,
|
|
message="Revisit testing.", justification="The selected source cannot address this failure.",
|
|
))
|
|
current = graph.start(Phase.ALPHA, FailedQAItem(source, "repeat", failed_test=failure))
|
|
|
|
assert graph.messages[revived.message_id].parent_id == original
|
|
assert graph.edges[-1].from_id == original
|
|
assert graph.edges[-1].to_id == revived.message_id
|
|
assert graph.edges[-1].direction is Direction.ROLLBACK
|
|
assert graph.duplicate_remediation(current) is None
|
|
|
|
|
|
def test_rollback_steps_to_the_immediately_prior_phase_start(tmp_path: MemoryPath) -> None:
|
|
"""Rollback creates a fresh attempt with the preceding turn's original parent."""
|
|
|
|
graph = _graph(tmp_path)
|
|
original = _inspection()
|
|
original.editor_records = [EditorRecord("old source repair", "before", "after")]
|
|
original_id = graph.start(Phase.ALPHA, original)
|
|
current = _testing()
|
|
current_id = graph.track_provenance(original_id, Phase.BETA, current)
|
|
|
|
phase, revived = graph.revive(DeadEndError(
|
|
direction=Direction.ROLLBACK,
|
|
current_message_id=current_id,
|
|
message="Reconsider the source instruction.",
|
|
justification="The editable source lacks the implicated operation.",
|
|
))
|
|
|
|
assert phase is Phase.ALPHA
|
|
assert isinstance(revived, InspectionBacklog)
|
|
assert revived.editor_records == []
|
|
assert revived.message_id not in {original_id, current_id}
|
|
assert graph.messages[revived.message_id].parent_id is None
|
|
assert revived.future_guidance[0].addressed_message_id == revived.message_id
|
|
assert revived.future_guidance[0].origin_message_id == current_id
|
|
assert set(graph.messages) == {original_id, current_id, revived.message_id}
|
|
assert not any(edge.to_id == revived.message_id for edge in graph.edges)
|
|
recovered = _graph(tmp_path, resume=True)
|
|
assert recovered.messages[revived.message_id].parent_id is None
|
|
assert json.loads(recovered.messages[revived.message_id].payload)["future_guidance"][0]["origin_message_id"] == current_id
|
|
assert not any(edge.to_id == revived.message_id for edge in recovered.edges)
|
|
|
|
|
|
def test_new_cot_selects_the_newest_visible_terminal_after_resume(tmp_path: MemoryPath) -> None:
|
|
"""A surrogate can link the next COT to the newest related Beta branch."""
|
|
|
|
graph = _graph(tmp_path)
|
|
first = graph.start(Phase.ALPHA, _inspection("src/first.py"))
|
|
beta = graph.track_provenance(first, Phase.BETA, _testing("tests/first.py"))
|
|
graph.start(Phase.ALPHA, _inspection("src/second.py"))
|
|
latest = graph.track_provenance(beta, Phase.BETA, _testing("tests/repeated.py"))
|
|
graph.start(Phase.ALPHA, _inspection("src/unrelated.py"))
|
|
recovered = _graph(tmp_path, resume=True)
|
|
strategy = Mock(spec=BiDependencyDiscoveryStrategy)
|
|
strategy.has_path.side_effect = lambda start, target: (start, target) in {
|
|
("src/surrogate.py", "src/first.py"),
|
|
("src/seed.py", "src/second.py"),
|
|
}
|
|
roots = [ClassMeta("src/seed.py", "", [], {}), ClassMeta("src/surrogate.py", "", [], {})]
|
|
before = tuple(recovered.messages)
|
|
|
|
parent = recovered.parent_for(roots, strategy)
|
|
|
|
assert parent == latest
|
|
assert tuple(recovered.messages) == before
|
|
item = _inspection("src/seed.py")
|
|
new_id = recovered.start(Phase.ALPHA, item, parent_id=parent)
|
|
assert recovered.messages[new_id].parent_id == latest
|
|
assert recovered.restart_options(new_id)[0]["message_id"] == latest
|
|
assert recovered.start(Phase.ALPHA, item) == new_id
|
|
|
|
|
|
@pytest.mark.parametrize("visible", [False, True])
|
|
def test_new_cot_links_only_a_visible_terminal_alpha(tmp_path: MemoryPath, visible: bool) -> None:
|
|
"""Changing only dependency visibility decides whether a COT gets a parent."""
|
|
|
|
graph = _graph(tmp_path)
|
|
original = graph.start(Phase.ALPHA, _inspection("src/original.py"))
|
|
strategy = Mock(spec=BiDependencyDiscoveryStrategy)
|
|
strategy.has_path.return_value = visible
|
|
roots = [ClassMeta("src/seed.py", "", [], {})]
|
|
|
|
parent = graph.parent_for(roots, strategy)
|
|
|
|
assert parent == (original if visible else None)
|
|
strategy.has_path.assert_called_once_with("src/seed.py", "src/original.py")
|
|
|
|
|
|
def test_restart_edges_do_not_erase_a_terminal_staging_branch(tmp_path: MemoryPath) -> None:
|
|
"""A fresh revival becomes visible without fabricating an ordinary staging successor."""
|
|
|
|
graph = _graph(tmp_path)
|
|
first = graph.start(Phase.ALPHA, _inspection("src/first.py"))
|
|
beta = graph.track_provenance(first, Phase.BETA, _testing())
|
|
_, revived = graph.revive(DeadEndError(
|
|
direction=Direction.ROLLBACK, current_message_id=beta,
|
|
message="Reconsider the earlier source.", justification="Its contract is incomplete.",
|
|
))
|
|
strategy = Mock(spec=BiDependencyDiscoveryStrategy)
|
|
strategy.has_path.return_value = True
|
|
|
|
assert graph.parent_for([ClassMeta("src/seed.py", "", [], {})], strategy) == revived.message_id
|
|
assert not any(edge.from_id == beta and edge.direction is None for edge in graph.edges)
|
|
assert len(graph.messages) == 3
|
|
|
|
|
|
def test_attribution_uses_recorded_context_and_traceback_in_its_own_lineage(tmp_path: MemoryPath) -> None:
|
|
"""File evidence need not be a phase focus, but unrelated branches cannot supply it."""
|
|
|
|
graph = _graph(tmp_path)
|
|
earlier = _testing()
|
|
earlier.failure_context_sources = [ClassMeta("src/context.py", "", [], {})]
|
|
earlier.failure_stack_trace = [f'File "{tmp_path}/src/caller.py", line 9']
|
|
earlier.failure_raw_output = "tests/src/suffix.py:22: failed"
|
|
parent = graph.start(Phase.BETA, earlier)
|
|
current = graph.track_provenance(parent, Phase.ALPHA, _inspection())
|
|
graph.start(Phase.ALPHA, _inspection("src/sibling.py"))
|
|
recovered = _graph(tmp_path, resume=True)
|
|
|
|
assert recovered.has_file(current, "src/context.py")
|
|
assert recovered.has_file(current, "src/caller.py")
|
|
assert not recovered.has_file(current, "src/sibling.py")
|
|
assert not recovered.has_file(current, "src/suffix.py")
|
|
assert not recovered.has_file(current, "src/missing.py")
|
|
|
|
|
|
def test_later_file_evidence_survives_successive_parent_restarts(tmp_path: MemoryPath) -> None:
|
|
"""A later attribution stays available after revival without rewriting earlier snapshots."""
|
|
|
|
graph = _graph(tmp_path)
|
|
first = graph.start(Phase.ALPHA, _inspection("src/first.py"))
|
|
beta = graph.track_provenance(first, Phase.BETA, _testing())
|
|
current = _testing("tests/later.py")
|
|
current.failure_stack_trace = ["src/later.py:12: stale call"]
|
|
last = graph.track_provenance(beta, Phase.BETA, current)
|
|
original_payload = graph.messages[beta].payload
|
|
assert graph.has_file(last, "src/later.py")
|
|
assert not graph.has_file(beta, "src/later.py")
|
|
_, revived = graph.revive(DeadEndError(
|
|
direction=Direction.ROLLBACK, current_message_id=last,
|
|
message="Attributed file: src/later.py", justification="The traceback exposes the stale call.",
|
|
))
|
|
recovered = _graph(tmp_path, resume=True)
|
|
|
|
assert recovered.has_file(revived.message_id, "src/later.py")
|
|
assert not recovered.has_file(beta, "src/later.py")
|
|
_, earlier = recovered.revive(DeadEndError(
|
|
direction=Direction.ROLLBACK, current_message_id=revived.message_id,
|
|
message="Attributed file: src/later.py", justification="The source plan must reconsider this call.",
|
|
))
|
|
assert earlier.message_id not in {first, beta, last, revived.message_id}
|
|
assert recovered.messages[earlier.message_id].parent_id is None
|
|
assert recovered.has_file(earlier.message_id, "src/later.py")
|
|
assert recovered.messages[beta].payload == original_payload
|
|
assert len(recovered.messages) == 5
|
|
|
|
|
|
def test_rollforward_skips_the_immediately_prior_phase_start(tmp_path: MemoryPath) -> None:
|
|
"""Rollforward retreats two turns and requeues the older phase start."""
|
|
|
|
graph = _graph(tmp_path)
|
|
alpha = _inspection()
|
|
alpha_id = graph.start(Phase.ALPHA, alpha)
|
|
beta = _testing()
|
|
beta_id = graph.track_provenance(alpha_id, Phase.BETA, beta)
|
|
current = _inspection("src/current.py")
|
|
current_id = graph.track_provenance(beta_id, Phase.ALPHA, current)
|
|
|
|
phase, revived = graph.revive(DeadEndError(
|
|
direction=Direction.ROLLFORWARD,
|
|
current_message_id=current_id,
|
|
message="Revisit the selected source work.",
|
|
justification="The test failure shows the preceding Alpha choice was wrong.",
|
|
))
|
|
|
|
assert phase is Phase.ALPHA
|
|
assert isinstance(revived, InspectionBacklog)
|
|
assert revived.message_id not in {alpha_id, beta_id, current_id}
|
|
assert graph.messages[revived.message_id].parent_id is None
|
|
assert revived.future_guidance[0].origin_message_id == current_id
|
|
assert revived.future_guidance[0].addressed_message_id == revived.message_id
|
|
assert not any(edge.to_id == revived.message_id for edge in graph.edges)
|
|
|
|
|
|
def test_repeated_restarts_keep_retreating_through_primary_history(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
"""Two rollbacks move from the head to its first, then second, predecessor."""
|
|
|
|
graph = _graph(tmp_path)
|
|
first_id = graph.start(Phase.ALPHA, _inspection("src/first.py"))
|
|
second_id = graph.track_provenance(first_id, Phase.BETA, _testing())
|
|
third = _inspection("src/third.py")
|
|
third_id = graph.track_provenance(second_id, Phase.ALPHA, third)
|
|
|
|
first_phase, first_restart = graph.revive(DeadEndError(
|
|
direction=Direction.ROLLBACK,
|
|
current_message_id=third_id,
|
|
message="The current source turn selected the wrong work.",
|
|
justification="The immediately preceding test turn must be reconsidered first.",
|
|
))
|
|
second_phase, second_restart = graph.revive(DeadEndError(
|
|
direction=Direction.ROLLBACK,
|
|
current_message_id=first_restart.message_id,
|
|
message="The returned test turn is also wrong.",
|
|
justification="The preceding source turn remains the next historical target.",
|
|
))
|
|
|
|
assert first_phase is Phase.BETA
|
|
assert first_restart.message_id not in {first_id, second_id, third_id}
|
|
assert graph.messages[first_restart.message_id].parent_id == first_id
|
|
assert second_phase is Phase.ALPHA
|
|
assert second_restart.message_id not in {first_id, second_id, third_id, first_restart.message_id}
|
|
assert graph.messages[second_restart.message_id].parent_id is None
|
|
assert len(graph.messages) == 5
|
|
assert graph.edges[-1].from_id == first_id
|
|
assert graph.edges[-1].to_id == first_restart.message_id
|
|
assert graph.edges[-1].direction is Direction.ROLLBACK
|
|
assert not any(edge.to_id == second_restart.message_id for edge in graph.edges)
|
|
assert second_restart.future_guidance[-1].origin_message_id == first_restart.message_id
|
|
|
|
new_fourth_id = graph.track_provenance(
|
|
second_restart.message_id,
|
|
Phase.BETA,
|
|
_testing(),
|
|
)
|
|
|
|
assert new_fourth_id not in {first_id, second_id, third_id}
|
|
assert graph.messages[third_id].parent_id == second_id
|
|
assert graph.messages[second_id].parent_id == first_id
|
|
assert graph.messages[new_fourth_id].parent_id == second_restart.message_id
|
|
assert len(graph.messages) == 6
|
|
|
|
|
|
def test_restart_options_require_the_history_depth_of_each_direction(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
"""Rollback needs one prior turn; rollforward needs two."""
|
|
|
|
graph = _graph(tmp_path)
|
|
first_id = graph.start(Phase.ALPHA, _inspection("src/first.py"))
|
|
second_id = graph.track_provenance(first_id, Phase.BETA, _testing())
|
|
third_id = graph.track_provenance(second_id, Phase.ALPHA, _inspection("src/third.py"))
|
|
|
|
assert graph.restart_options(first_id) == ()
|
|
assert [option["direction"] for option in graph.restart_options(second_id)] == [
|
|
Direction.ROLLBACK.value,
|
|
]
|
|
assert [option["direction"] for option in graph.restart_options(third_id)] == [
|
|
Direction.ROLLBACK.value,
|
|
Direction.ROLLFORWARD.value,
|
|
]
|
|
|
|
|
|
def test_resumed_graph_recovery_is_independent_of_context_hydration(tmp_path: MemoryPath) -> None:
|
|
"""An explicit resume reads selected phase history without a context checkpoint."""
|
|
|
|
graph = _graph(tmp_path)
|
|
original = _inspection()
|
|
original_id = graph.start(Phase.ALPHA, original)
|
|
|
|
recovered = _graph(tmp_path, resume=True)
|
|
|
|
assert original_id in recovered.messages
|
|
assert recovered.messages[original_id].phase is Phase.ALPHA
|
|
assert recovered.restart_options(original_id) == ()
|
|
|
|
|
|
def test_resumed_graph_revives_a_neutral_remediation_record(tmp_path: MemoryPath) -> None:
|
|
"""Recovery rebuilds the complete Alpha work record without opaque runtime state."""
|
|
|
|
failed_test = ClassMeta("tests/focus", "focused test", [], {})
|
|
remediation = FailedQAItem(
|
|
focused_spotlight=ClassMeta("src/focus", "focused source", ["direct"], {"run": "needed"}),
|
|
summary="repair focused source",
|
|
original_cot="The source must provide the focused behavior.",
|
|
previous_beta_guidance="The focused test identifies the missing behavior.",
|
|
failure_error_message="focused test fails",
|
|
failure_stack_trace=["tests/focus: assertion"],
|
|
failure_raw_output="failure output",
|
|
failure_context_sources=[ClassMeta("src/helper", "helper", [], {})],
|
|
failed_test=FailingTest(
|
|
test=failed_test,
|
|
stack_trace=["tests/focus: assertion"],
|
|
error_message="focused test fails",
|
|
raw_output="failure output",
|
|
),
|
|
)
|
|
graph = _graph(tmp_path)
|
|
message_id = graph.start(Phase.ALPHA, remediation)
|
|
current_id = graph.track_provenance(message_id, Phase.BETA, _testing())
|
|
|
|
segments = tmp_path.repository.list_files(
|
|
tmp_path / provenance.PROVENANCE_DIRNAME / graph.graph_id / "segments",
|
|
("*.jsonl",),
|
|
recursive=False,
|
|
)
|
|
message_record = json.loads(tmp_path.repository.read_text(segments[0]).splitlines()[0])
|
|
durable_record = json.loads(message_record["payload"])
|
|
recovered = _graph(tmp_path, resume=True)
|
|
phase, revived = recovered.revive(DeadEndError(
|
|
direction=Direction.ROLLBACK,
|
|
current_message_id=current_id,
|
|
message="Retry the source phase with the focused failure evidence.",
|
|
justification="The current remediation needs the original failure context.",
|
|
))
|
|
|
|
assert durable_record["kind"] == "remediation"
|
|
assert durable_record["focus"]["location"] == "src/focus"
|
|
assert durable_record["failed_test"]["test"]["location"] == "tests/focus"
|
|
assert phase is Phase.ALPHA
|
|
assert isinstance(revived, FailedQAItem)
|
|
assert revived.focused_spotlight is not None
|
|
assert revived.focused_spotlight.relevant_symbols == {"run": "needed"}
|
|
assert revived.failed_test is not None
|
|
assert revived.failed_test.test.location == "tests/focus"
|
|
assert revived.failure_context_sources[0].location == "src/helper"
|
|
assert revived.future_guidance[-1].message == (
|
|
"Retry the source phase with the focused failure evidence."
|
|
)
|
|
|
|
|
|
def test_fresh_task_resets_current_graph_without_writing_ui_logs(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
"""A fresh task selects an empty durable graph and leaves UI logging untouched."""
|
|
|
|
first = _graph(tmp_path)
|
|
first_id = first.start(Phase.ALPHA, _inspection())
|
|
|
|
fresh = _graph(tmp_path)
|
|
resumed = _graph(tmp_path, resume=True)
|
|
|
|
assert fresh.graph_id != first.graph_id
|
|
assert fresh.messages == {}
|
|
assert resumed.graph_id == fresh.graph_id
|
|
assert first_id not in resumed.messages
|
|
assert tmp_path.repository.exists(
|
|
tmp_path / provenance.PROVENANCE_DIRNAME / "current.json"
|
|
)
|
|
assert tmp_path.repository.exists(
|
|
tmp_path / provenance.PROVENANCE_DIRNAME / first.graph_id / "segments"
|
|
)
|
|
assert not tmp_path.repository.exists(tmp_path / ".band-aider-logs" / "provenance")
|
|
|
|
|
|
def test_resume_without_prior_provenance_keeps_checkpoint_hydration_runnable(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
"""A checkpoint created before provenance starts a durable graph on resume."""
|
|
|
|
graph = _graph(tmp_path, resume=True)
|
|
|
|
assert graph.messages == {}
|
|
assert tmp_path.repository.exists(
|
|
tmp_path / provenance.PROVENANCE_DIRNAME / "current.json"
|
|
)
|
|
|
|
|
|
def test_graph_records_emit_payload_free_runtime_log_events(
|
|
tmp_path: MemoryPath,
|
|
caplog,
|
|
) -> None:
|
|
"""Durable graph writes emit normal runtime logs without treating them as state."""
|
|
|
|
graph = _graph(tmp_path)
|
|
logger = logging.getLogger("test.provenance.events")
|
|
with caplog.at_level(logging.INFO, logger=logger.name):
|
|
with provenance.bind(graph, logger):
|
|
message_id = graph.start(Phase.ALPHA, _inspection())
|
|
|
|
event = next(
|
|
record.getMessage()
|
|
for record in caplog.records
|
|
if record.name == logger.name and record.getMessage().startswith("PROVENANCE ")
|
|
)
|
|
assert message_id in event
|
|
assert graph.messages[message_id].payload not in event
|
|
|
|
|
|
def test_each_parent_tracks_an_edge_when_a_message_is_deduplicated(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
"""Two phase starts can retain separate edges to one queued message."""
|
|
|
|
graph = _graph(tmp_path)
|
|
first_parent = graph.start(Phase.ALPHA, _inspection("src/one.py"))
|
|
second_parent = graph.start(Phase.ALPHA, _inspection("src/two.py"))
|
|
tracked = _testing()
|
|
tracked_id = graph.track_provenance(first_parent, Phase.BETA, tracked)
|
|
repeated_id = graph.track_provenance(second_parent, Phase.BETA, tracked)
|
|
|
|
assert repeated_id == tracked_id
|
|
assert len(graph.messages) == 3
|
|
assert graph.edges[-2].from_id == first_parent
|
|
assert graph.edges[-1].from_id == second_parent
|
|
assert graph.edges[-2].to_id == graph.edges[-1].to_id == tracked_id
|
|
|
|
|
|
def test_direct_editor_maps_each_file_bound_fault_to_its_phase(tmp_path: MemoryPath) -> None:
|
|
"""Direct-editor fault categories distinguish another file from an invalid new file."""
|
|
|
|
graph = _graph(tmp_path)
|
|
alpha_id = graph.start(Phase.ALPHA, _inspection())
|
|
beta_id = graph.track_provenance(alpha_id, Phase.BETA, _testing())
|
|
item = _inspection("src/current.py")
|
|
graph.track_provenance(beta_id, Phase.ALPHA, item)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="Repair the selected source.",
|
|
editable_files=[Path("src/current.py")],
|
|
readonly_files=[Path("src/readonly.py")],
|
|
)
|
|
completed_prompt = "\n".join((
|
|
"FULL EDIT CONTEXT",
|
|
"EDITABLE FILES",
|
|
"src/current.py",
|
|
"⋮",
|
|
"READONLY FILES",
|
|
"src/readonly.py",
|
|
"DEPENDENCY REFERENCE CARDS",
|
|
"No reference cards resolved.",
|
|
))
|
|
wrong_file_client = _Client({
|
|
"dead_end": True,
|
|
"scope_fault": "wrong_editable_file",
|
|
"message": "The implicated operation is outside this file.",
|
|
"justification": "The failure evidence indicates the selected scope cannot reach the operation.",
|
|
})
|
|
new_file_client = _Client({
|
|
"dead_end": True,
|
|
"scope_fault": "file_should_not_exist",
|
|
"message": "Reconsider whether this source file belongs in the task.",
|
|
"justification": "The task does not require the newly planned file.",
|
|
})
|
|
|
|
with provenance.bind(graph):
|
|
with provenance.item_scope(Phase.ALPHA, item):
|
|
with pytest.raises(DeadEndError) as wrong_file:
|
|
DeadEndDecisionBuilder.for_direct_code_editor(
|
|
client=wrong_file_client,
|
|
file_manager=_DecisionFileManager(),
|
|
request=request,
|
|
last_result={"candidate_compiler_errors": ("fault",)},
|
|
completed_prompt=completed_prompt,
|
|
).build()()
|
|
with pytest.raises(DeadEndError) as new_file:
|
|
DeadEndDecisionBuilder.for_direct_code_editor(
|
|
client=new_file_client,
|
|
file_manager=_DecisionFileManager(),
|
|
request=request,
|
|
last_result={"candidate_compiler_errors": ("fault",)},
|
|
completed_prompt=completed_prompt,
|
|
).build()()
|
|
|
|
assert wrong_file.value.direction is Direction.ROLLBACK
|
|
assert new_file.value.direction is Direction.ROLLFORWARD
|
|
assert completed_prompt in wrong_file_client.prompts[0]
|
|
assert "wrong_editable_file" in wrong_file_client.prompts[0]
|
|
assert "file_should_not_exist" in wrong_file_client.prompts[0]
|
|
assert "A fault that implicates a supplied readonly file is one example" in wrong_file_client.prompts[0]
|
|
assert "it is not a required proof" in wrong_file_client.prompts[0]
|
|
assert "ContextualFileRenderer omitted a relevant region of an editable file" in wrong_file_client.prompts[0]
|
|
assert "a dependency reference card is missing or incomplete" in wrong_file_client.prompts[0]
|
|
assert "not by themselves evidence of a wrong editable selection" in wrong_file_client.prompts[0]
|
|
assert "CURRENT PHASE" not in wrong_file_client.prompts[0]
|
|
assert "AVAILABLE RESTART DIRECTIONS" not in wrong_file_client.prompts[0]
|
|
|
|
|
|
def test_editor_dead_end_localizes_future_guidance_before_admission(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
"""An editor dead-end admission sees only guidance localized to its repair."""
|
|
|
|
graph = _graph(tmp_path)
|
|
beta_id = graph.start(Phase.BETA, _testing())
|
|
item = _inspection()
|
|
graph.track_provenance(beta_id, Phase.ALPHA, item)
|
|
client = _Client(
|
|
{
|
|
"applicable": True,
|
|
"reason": "This source attribution can reconsider its evidence.",
|
|
},
|
|
{
|
|
"localized_prompt": "Re-evaluate the failure evidence before rejecting source attribution.",
|
|
},
|
|
{
|
|
"dead_end": False,
|
|
"scope_fault": "none",
|
|
"message": "",
|
|
"justification": "",
|
|
},
|
|
)
|
|
request = AiderEditRequest(
|
|
authoritative_prompt="Repair the selected source.",
|
|
editable_files=[Path("src/focus.py")],
|
|
readonly_files=[],
|
|
failure_error_message="The focused source cannot satisfy the failing test.",
|
|
backlog_guidance=["Original source attribution guidance."],
|
|
future_guidance=(provenance.FutureGuidance(
|
|
origin_message_id="later",
|
|
addressed_message_id=item.message_id,
|
|
message="The prior source attribution reached a dead end.",
|
|
justification="The failure evidence was not reconsidered.",
|
|
),),
|
|
)
|
|
|
|
with provenance.bind(graph):
|
|
with provenance.item_scope(Phase.ALPHA, item):
|
|
assert DeadEndDecisionBuilder.for_direct_code_editor(
|
|
client=client,
|
|
file_manager=_DecisionFileManager(),
|
|
request=request,
|
|
last_result={
|
|
"decision": "DirectCodeEditor",
|
|
"file": "tests/test_focus.py",
|
|
"reason": "The test retains an obsolete assertion.",
|
|
},
|
|
completed_prompt="DIRECT EDITOR",
|
|
).build()() is False
|
|
|
|
assert request.backlog_guidance == [
|
|
"Original source attribution guidance.",
|
|
"Re-evaluate the failure evidence before rejecting source attribution.",
|
|
]
|
|
assert len(client.prompts) == 3
|
|
assert "FUTURE GUIDANCE APPLICABILITY" in client.prompts[0]
|
|
assert "FUTURE GUIDANCE RELOCALIZATION" in client.prompts[1]
|
|
assert "LOCALIZED BACKLOG GUIDANCE" in client.prompts[2]
|
|
assert "The prior source attribution reached a dead end." not in client.prompts[2]
|
|
|
|
|
|
def test_alpha_handler_rolls_back_by_prepending_the_prior_beta_start(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
"""An Alpha dead end abandons its item and requeues the prior Beta turn."""
|
|
|
|
graph = _graph(tmp_path)
|
|
beta = _testing()
|
|
beta_id = graph.start(Phase.BETA, beta)
|
|
alpha = _inspection()
|
|
alpha_id = graph.track_provenance(beta_id, Phase.ALPHA, alpha)
|
|
queued: list[tuple[BetaWork, bool]] = []
|
|
discarded: list[bool] = []
|
|
|
|
def queue_testing(item: BetaWork, *, prepend: bool, logger: logging.Logger) -> None:
|
|
del logger
|
|
queued.append((item, prepend))
|
|
|
|
context = SimpleNamespace(
|
|
discard_active_inbound=lambda: discarded.append(True),
|
|
backlog=SimpleNamespace(queue_testing=queue_testing),
|
|
)
|
|
loop = cast(AlphaLoop, SimpleNamespace(
|
|
context=context,
|
|
provenance=graph,
|
|
logger=logging.getLogger(__name__),
|
|
))
|
|
|
|
AlphaLoop._restart_item(loop, DeadEndError(
|
|
direction=Direction.ROLLBACK,
|
|
current_message_id=alpha_id,
|
|
message="Resume the focused test phase.",
|
|
justification="The current Alpha item selected the wrong source file.",
|
|
))
|
|
|
|
assert discarded == [True]
|
|
assert len(queued) == 1
|
|
restarted, prepend = queued[0]
|
|
assert prepend
|
|
assert restarted.message_id not in {beta_id, alpha_id}
|
|
assert graph.messages[restarted.message_id].parent_id == graph.messages[beta_id].parent_id
|
|
assert restarted.future_guidance[-1].origin_message_id == alpha_id
|
|
|
|
|
|
def test_beta_handler_rolls_forward_by_prepending_the_two_turn_prior_beta_start(
|
|
tmp_path: MemoryPath,
|
|
) -> None:
|
|
"""A Beta dead end abandons its item and requeues the older Beta turn."""
|
|
|
|
graph = _graph(tmp_path)
|
|
beta = _testing()
|
|
beta_id = graph.start(Phase.BETA, beta)
|
|
alpha_id = graph.track_provenance(beta_id, Phase.ALPHA, _inspection())
|
|
current = _testing("tests/current.py")
|
|
current_id = graph.track_provenance(alpha_id, Phase.BETA, current)
|
|
queued: list[tuple[BetaWork, bool]] = []
|
|
discarded: list[bool] = []
|
|
|
|
def queue_testing(item: BetaWork, *, prepend: bool, logger: logging.Logger) -> None:
|
|
del logger
|
|
queued.append((item, prepend))
|
|
|
|
context = SimpleNamespace(
|
|
discard_active_inbound=lambda: discarded.append(True),
|
|
backlog=SimpleNamespace(queue_testing=queue_testing),
|
|
)
|
|
loop = cast(BetaLoop, SimpleNamespace(
|
|
context=context,
|
|
provenance=graph,
|
|
logger=logging.getLogger(__name__),
|
|
))
|
|
|
|
BetaLoop._restart_item(loop, DeadEndError(
|
|
direction=Direction.ROLLFORWARD,
|
|
current_message_id=current_id,
|
|
message="Retry the test phase from its start.",
|
|
justification="The selected test file cannot express the required coverage.",
|
|
))
|
|
|
|
assert discarded == [True]
|
|
assert len(queued) == 1
|
|
restarted, prepend = queued[0]
|
|
assert prepend
|
|
assert restarted.message_id not in {beta_id, alpha_id, current_id}
|
|
assert graph.messages[restarted.message_id].parent_id == graph.messages[beta_id].parent_id
|
|
assert restarted.future_guidance[-1].origin_message_id == current_id
|
|
|
|
|
|
def test_graph_rotates_the_next_append_without_rewriting_prior_records(
|
|
tmp_path: MemoryPath,
|
|
monkeypatch,
|
|
) -> None:
|
|
"""An oversized segment remains intact while the following record uses a new file."""
|
|
|
|
monkeypatch.setattr(provenance, "SEGMENT_BYTES", 1)
|
|
graph = _graph(tmp_path)
|
|
graph.start(Phase.ALPHA, _inspection("src/one.py"))
|
|
graph.start(Phase.ALPHA, _inspection("src/two.py"))
|
|
|
|
root = tmp_path / provenance.PROVENANCE_DIRNAME / graph.graph_id / "segments"
|
|
segments = tmp_path.repository.list_files(root, ("*.jsonl",), recursive=False)
|
|
|
|
assert len(segments) == 2
|
|
assert all(path.name.endswith(".jsonl") for path in segments)
|
|
|
|
|
|
def test_recovery_orders_timestamped_segments_before_rebuilding_edges(
|
|
tmp_path: MemoryPath,
|
|
monkeypatch,
|
|
) -> None:
|
|
"""A filesystem that returns segments backward cannot reverse durable edge history."""
|
|
|
|
monkeypatch.setattr(provenance, "SEGMENT_BYTES", 1)
|
|
graph = _graph(tmp_path)
|
|
alpha_id = graph.start(Phase.ALPHA, _inspection("src/one.py"))
|
|
beta = _testing("tests/test_one.py")
|
|
beta_id = graph.track_provenance(alpha_id, Phase.BETA, beta)
|
|
graph.track_provenance(beta_id, Phase.ALPHA, _inspection("src/two.py"))
|
|
list_files = tmp_path.repository.list_files
|
|
monkeypatch.setattr(
|
|
tmp_path.repository,
|
|
"list_files",
|
|
lambda *args, **kwargs: list(reversed(list_files(*args, **kwargs))),
|
|
)
|
|
|
|
recovered = _graph(tmp_path, resume=True)
|
|
|
|
assert [edge.sequence for edge in recovered.edges] == sorted(
|
|
edge.sequence for edge in recovered.edges
|
|
)
|