887 lines
34 KiB
Python
887 lines
34 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Durable context and resume behavior over a test-only in-memory repository."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from band_aider.judgement.test_runner_strategy import TestRunOutcome
|
|
from dataclasses import dataclass
|
|
import json
|
|
import logging
|
|
from pathlib import Path
|
|
import pickle
|
|
from typing import Sequence, TextIO
|
|
|
|
import pytest
|
|
|
|
from band_aider.ai_client.openai_compatible_client import OpenAICompatibleClient
|
|
from band_aider.context.alpha_context import AlphaContext
|
|
from band_aider.context.beta_context import BetaContext
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.context.contex_bag import ContextBag
|
|
from band_aider.judgement.code_editor import EditorRecord
|
|
from band_aider.context.context_factory import _checkpoint_shape_is_current, context_factory
|
|
from band_aider.context.failing_test import FailingTest
|
|
from band_aider.context import provenance
|
|
from band_aider.context.provenance import FutureGuidance, Phase, Provenance
|
|
from band_aider.context.global_context import GlobalContext, StoryCursor
|
|
from band_aider.context.resume_state import configure_resume_settings, reset_resume_state_for_tests
|
|
from band_aider.context.stabilization_context import StabilizationContext
|
|
from band_aider.context.topological_liveliness import TopologicalLiveliness
|
|
from band_aider.context.traversed_frontier_history import TraversedFrontierHistory
|
|
from band_aider.context.work_completion import WorkCompletion
|
|
from band_aider.context.working_backlog import (
|
|
BackLog,
|
|
BacklogLane,
|
|
ContributorBacklog,
|
|
FailedQAItem,
|
|
InspectionBacklog,
|
|
QABacklog,
|
|
StagedInboundItem,
|
|
TestingBacklog,
|
|
)
|
|
from band_aider.judgement.code_parser import CodeParser
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.file_parser import FileParser
|
|
from band_aider.judgement.intent_check import IntentCheckFactory
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
|
|
from band_aider.judgement.test_runner_strategy import TestRunResult, TestRunnerStrategy
|
|
from band_aider.loop.loop_gate import LoopGate
|
|
from band_aider.loop.mechanical_loop_gate import MechanicalLoopGate
|
|
from band_aider.loop.proto_loop import ProtoLoop
|
|
from band_aider.loop.stabilization_loop import StabilizationLoop
|
|
from band_aider.strategies import LanguageStrategySet
|
|
from band_aider.strategies.python import build_python_language_strategy_set
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
PROMPT_FORMATTER = MarkdownPromptFormatterBuilder()
|
|
|
|
|
|
@dataclass
|
|
class _TinyContext(ContextBag):
|
|
"""Represent the smallest durable ContextBag used by checkpoint tests."""
|
|
|
|
marker: str = ""
|
|
|
|
|
|
@dataclass
|
|
class _SelectiveContext(ContextBag):
|
|
"""Persist an explicit durable subset of its state."""
|
|
|
|
keep: str = ""
|
|
runtime_callback: object | None = None
|
|
|
|
def __getstate__(self) -> dict[str, object]:
|
|
"""Omit the explicitly runtime-only callback from the stored payload."""
|
|
|
|
state = super().__getstate__()
|
|
state.pop("runtime_callback", None)
|
|
return state
|
|
|
|
|
|
class _OneShotGate(LoopGate):
|
|
"""Admit exactly one ProtoLoop pass without retry or continuation."""
|
|
|
|
def may_enter(self) -> bool:
|
|
"""Allow the initial iteration."""
|
|
|
|
return True
|
|
|
|
def should_retry(self) -> bool:
|
|
"""Never retry the completed tiny loop."""
|
|
|
|
return False
|
|
|
|
def should_continue(self) -> bool:
|
|
"""Never request another tiny-loop pass."""
|
|
|
|
return False
|
|
|
|
|
|
class _TinyLoop(ProtoLoop[LoopGate, _TinyContext]):
|
|
"""Exercise the normal ProtoLoop terminal path."""
|
|
|
|
def _prepare_work(self) -> ProtoLoop | None:
|
|
"""Return no child loop for the terminal scenario."""
|
|
|
|
return None
|
|
|
|
def _post_run(self) -> str:
|
|
"""Record one terminal result in the durable context."""
|
|
|
|
self.context.marker = "finished"
|
|
return "done"
|
|
|
|
|
|
class _Client(OpenAICompatibleClient):
|
|
"""Return stationary JSON for context metadata and intent decisions."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Create a nominal client whose overridden completion has no transport."""
|
|
|
|
super().__init__(model="test-model", api_base="http://model.invalid")
|
|
|
|
def complete(self, prompt: str, logger: logging.Logger | None = None) -> str:
|
|
"""Return the exact response shape requested by production collaborators."""
|
|
|
|
del logger
|
|
if "cot_tasks" in prompt:
|
|
return json.dumps({"cot_tasks": [{"task": "a.py: task", "locations": ["a.py"], "symbols": []}]})
|
|
if "IntentCheck" in prompt:
|
|
return json.dumps({"intent": "existing_files", "new_feature": False, "reason": "Existing source is in scope."})
|
|
return json.dumps({"cdc": "metadata changed"})
|
|
|
|
|
|
class _Parser(CodeParser):
|
|
"""Parse ClassMeta values from paths held by a MemoryRepoExec instance."""
|
|
|
|
def __init__(self, src_root: Path, test_root: Path, repo_exec: MemoryRepoExec) -> None:
|
|
"""Store lexical roots and the repository execution boundary."""
|
|
|
|
self.src_root = src_root
|
|
self.test_root = test_root
|
|
self.repo_exec = repo_exec
|
|
|
|
def parse_file(
|
|
self,
|
|
file_name: str,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Return deterministic path-derived metadata."""
|
|
|
|
path = Path(file_name)
|
|
root = self.src_root if path == self.src_root or self.src_root in path.parents else self.test_root
|
|
return ClassMeta(
|
|
location=str(path.relative_to(root)),
|
|
summary=f"summary {path.name}",
|
|
relevance=[prompt] if prompt else [],
|
|
relevant_symbols={},
|
|
content_type=content_type,
|
|
)
|
|
|
|
def parse_stream(
|
|
self,
|
|
file_name: str,
|
|
stream: TextIO,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Apply the same lexical rule to an already-open stream."""
|
|
|
|
del stream
|
|
return self.parse_file(file_name, prompt, content_type)
|
|
|
|
def parse_files(
|
|
self,
|
|
file_names: list[str],
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> list[ClassMeta]:
|
|
"""Parse every requested path in stable order."""
|
|
|
|
return [self.parse_file(file_name, prompt, content_type) for file_name in file_names]
|
|
|
|
|
|
class _Runner(TestRunnerStrategy):
|
|
"""Provide stationary passing test results for context-only workflows."""
|
|
|
|
def failure_reports(self, test_files, result):
|
|
"""Return the scripted known test scope without running report discovery."""
|
|
return [FailingTest.from_result(test_files[0], result)]
|
|
|
|
def matching_tests(self, test_files: list[ClassMeta], result: TestRunResult) -> dict[str, tuple[str, ...]]:
|
|
"""Reject unexpected final-exam discovery in this isolated runner scenario."""
|
|
raise AssertionError("This scenario does not provide failed-suite method discovery.")
|
|
|
|
def run_tests_for(
|
|
self,
|
|
test_files: list[ClassMeta],
|
|
max_failures: int | None = None,
|
|
test_names: Sequence[str] = (),
|
|
) -> TestRunResult:
|
|
"""Return a passing bounded targeted result."""
|
|
|
|
del test_files, max_failures, test_names
|
|
return TestRunResult("targeted", TestRunOutcome.Success, 0, "ok")
|
|
|
|
def run_all_tests_under(self, test_root: Path) -> TestRunResult:
|
|
"""Return a passing whole-suite result without a child process."""
|
|
|
|
del test_root
|
|
return TestRunResult("all", TestRunOutcome.Success, 0, "ok")
|
|
|
|
|
|
def _configure(memory_root: Path, executor: MemoryRepoExec, *, hydrate: bool = False, keep: int = 5) -> None:
|
|
"""Bind only the in-memory executor as this scenario's checkpoint storage."""
|
|
|
|
reset_resume_state_for_tests()
|
|
configure_resume_settings(
|
|
hydrate=hydrate,
|
|
checkpoint_dir=memory_root / ".band-aider-chkpts",
|
|
repo_exec=executor,
|
|
keep=keep,
|
|
)
|
|
|
|
|
|
def _file_manager(
|
|
memory_root: Path,
|
|
executor: MemoryRepoExec,
|
|
*,
|
|
source_files: dict[str, str] | None = None,
|
|
test_files: dict[str, str] | None = None,
|
|
) -> FileManager:
|
|
"""Build a real FileManager backed entirely by an isolated MemoryRepoExec."""
|
|
|
|
source_root = memory_root / "src"
|
|
test_root = memory_root / "tests"
|
|
executor.mkdir(source_root)
|
|
executor.mkdir(test_root)
|
|
for location, content in (source_files or {"a.py": "value = 1\n"}).items():
|
|
executor.write_text(source_root / location, content)
|
|
for location, content in (test_files or {}).items():
|
|
executor.write_text(test_root / location, content)
|
|
LanguageStrategySet.set_instance(build_python_language_strategy_set)
|
|
return FileManager(
|
|
parser=_Parser(source_root, test_root, executor),
|
|
client=_Client(),
|
|
repo_root=memory_root,
|
|
src_root=source_root,
|
|
test_root=test_root,
|
|
actor="test",
|
|
prompt_formatter_builder=PROMPT_FORMATTER,
|
|
prompt="goal",
|
|
repo_exec=executor,
|
|
)
|
|
|
|
|
|
def _global_context(memory_root: Path, executor: MemoryRepoExec) -> GlobalContext:
|
|
"""Build a normal GlobalContext that owns a memory-backed FileManager."""
|
|
|
|
source_root = memory_root / "src"
|
|
test_root = memory_root / "tests"
|
|
executor.mkdir(source_root)
|
|
executor.mkdir(test_root)
|
|
executor.write_text(source_root / "a.py", "value = 1\n")
|
|
parser = _Parser(source_root, test_root, executor)
|
|
context = GlobalContext(
|
|
goal="goal",
|
|
parser=parser,
|
|
client=_Client(),
|
|
repo_root=memory_root,
|
|
src_root=source_root,
|
|
test_root=test_root,
|
|
actor="test",
|
|
prompt_formatter_builder=PROMPT_FORMATTER,
|
|
intent_check_factory=IntentCheckFactory(_Client(), PROMPT_FORMATTER),
|
|
)
|
|
context.logger = logging.getLogger(__name__)
|
|
return context
|
|
|
|
|
|
def _checkpoint(executor: MemoryRepoExec, path: Path) -> dict[str, object]:
|
|
"""Decode one persisted checkpoint from the in-memory repository."""
|
|
|
|
value = pickle.loads(executor.read_bytes(path))
|
|
assert isinstance(value, dict)
|
|
return value
|
|
|
|
|
|
def _source(file_manager: FileManager, location: str) -> ClassMeta:
|
|
"""Return required current source metadata."""
|
|
|
|
source = file_manager.current_src(location)
|
|
assert source is not None
|
|
return source
|
|
|
|
|
|
def _test(file_manager: FileManager, location: str) -> ClassMeta:
|
|
"""Return required current test metadata."""
|
|
|
|
test = file_manager.current_test(location)
|
|
assert test is not None
|
|
return test
|
|
|
|
|
|
def _alpha_context(file_manager: FileManager, source: ClassMeta, test: ClassMeta) -> tuple[AlphaContext, BackLog]:
|
|
"""Build Alpha's normal backlog view over a real FileManager."""
|
|
|
|
raw = BackLog(completion={}, inspection=[InspectionBacklog(source, "inspect", downstream_classes=[source])])
|
|
strategy_set = build_python_language_strategy_set(file_manager)
|
|
strategy_set.bindings[TestRunnerStrategy] = _Runner()
|
|
backlog = ContributorBacklog(
|
|
backlog=raw,
|
|
get_tests=lambda _sources: [test],
|
|
file_manager=file_manager,
|
|
parser=file_manager.parser,
|
|
language_strategy_set=strategy_set,
|
|
)
|
|
return AlphaContext(goal="goal", classes=[source, test], file_manager=file_manager, backlog=backlog), raw
|
|
|
|
|
|
def _beta_context(file_manager: FileManager, source: ClassMeta, test: ClassMeta) -> tuple[BetaContext, BackLog]:
|
|
"""Build Beta's normal backlog view over a real FileManager."""
|
|
|
|
raw = BackLog(completion={}, testing=[TestingBacklog(focused_spotlight=test, summary="validate", downstream_classes=[source])])
|
|
backlog = QABacklog(raw, file_manager.client, file_manager, file_manager.parser)
|
|
return BetaContext(goal="goal", classes=[source, test], file_manager=file_manager, backlog=backlog), raw
|
|
|
|
|
|
def test_claimed_phase_items_start_provenance_before_loop_processing(memory_root: Path) -> None:
|
|
"""Alpha and Beta start their graph message at the committed backlog claim."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
manager = _file_manager(memory_root, executor, test_files={"test_a.py": "pass\n"})
|
|
source = _source(manager, "src/a.py")
|
|
test = _test(manager, "tests/test_a.py")
|
|
alpha, _ = _alpha_context(manager, source, test)
|
|
beta, _ = _beta_context(manager, source, test)
|
|
graph = Provenance.open(repo_root=memory_root, repo_exec=executor)
|
|
|
|
with provenance.bind(graph):
|
|
alpha_claim = alpha.move_inbound_from_backlog()
|
|
beta_claim = beta.move_inbound_from_backlog()
|
|
|
|
assert alpha_claim is not None
|
|
assert beta_claim is not None
|
|
assert alpha_claim.value.message_id in graph.messages
|
|
assert beta_claim.value.message_id in graph.messages
|
|
assert graph.messages[alpha_claim.value.message_id].phase is Phase.ALPHA
|
|
assert graph.messages[beta_claim.value.message_id].phase is Phase.BETA
|
|
|
|
|
|
def test_context_bag_persists_and_prunes_rolling_memory_checkpoints(memory_root: Path) -> None:
|
|
"""ContextBag writes and retains only the requested rolling checkpoint count."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor, keep=1)
|
|
context = _TinyContext(goal="goal", classes=[], marker="one")
|
|
first = context.persist()
|
|
context.marker = "two"
|
|
second = context.persist()
|
|
|
|
assert not executor.exists(first)
|
|
assert executor.exists(second)
|
|
assert [path.name for path in executor.list_directory(second.parent)] == [second.name]
|
|
|
|
|
|
def test_context_bag_uses_getstate_payload(memory_root: Path) -> None:
|
|
"""Persisted context wrapper retains durable fields and omits runtime state."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
context = _SelectiveContext(goal="goal", classes=[], keep="value", runtime_callback=object())
|
|
|
|
persisted = _checkpoint(executor, context.persist())
|
|
payload = persisted["payload"]
|
|
assert isinstance(payload, dict)
|
|
|
|
assert payload["goal"] == "goal"
|
|
assert payload["keep"] == "value"
|
|
assert "runtime_callback" not in payload
|
|
|
|
|
|
def test_protoloop_terminal_completion_does_not_write_checkpoint(memory_root: Path) -> None:
|
|
"""A completed ProtoLoop does not invent an after-the-fact durable checkpoint."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
context = _TinyContext(goal="goal", classes=[])
|
|
loop = _TinyLoop(
|
|
gate=MechanicalLoopGate(context, lambda: True, lambda: False),
|
|
context=context,
|
|
)
|
|
|
|
assert loop.run() == "done"
|
|
assert not executor.is_dir(memory_root / ".band-aider-chkpts")
|
|
|
|
|
|
def test_file_manager_resync_clears_dirty_maps(memory_root: Path) -> None:
|
|
"""A full metadata resync resets both source and test dirty maps."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
manager = _file_manager(memory_root, executor, test_files={"test_a.py": "from a import value\n"})
|
|
manager.src_dirty["src/a.py"] = True
|
|
manager.test_dirty["tests/test_a.py"] = True
|
|
|
|
manager.resync()
|
|
|
|
assert manager.src_dirty == {}
|
|
assert manager.test_dirty == {}
|
|
|
|
|
|
def test_global_context_sync_refreshes_current_memory_metadata(memory_root: Path) -> None:
|
|
"""Global sync rebuilds its owned FileManager through the bound memory executor."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
context = _global_context(memory_root, executor)
|
|
manager = context.file_manager
|
|
executor.write_text(manager.src_root / "b.py", "value = 2\n")
|
|
|
|
context.sync()
|
|
|
|
assert {source.location for source in context.file_manager.src_classes} == {"src/a.py", "src/b.py"}
|
|
assert context.file_manager.prompt == "goal"
|
|
|
|
|
|
def test_global_context_marks_root_lifecycle_in_durable_cursor(memory_root: Path) -> None:
|
|
"""Visited and stabilized root facts persist through the context's real checkpoint boundary."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
context = _global_context(memory_root, executor)
|
|
root = _source(context.file_manager, "src/a.py")
|
|
|
|
context.mark_root_visited(root)
|
|
context.mark_root_stabilized(root, "stable")
|
|
|
|
cursor = context.story_cursor.task_cursor
|
|
assert cursor.visited_locations_by_cot["0"] == {root.location}
|
|
assert cursor.stabilized_locations_by_cot["0"] == {root.location}
|
|
assert cursor.active_root is None
|
|
assert context.story_cursor.run_summaries == ["stable"]
|
|
|
|
|
|
def test_stabilization_started_transition_persists_only_on_change(memory_root: Path) -> None:
|
|
"""Repeated equivalent stabilization state does not create a redundant checkpoint."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
context = _global_context(memory_root, executor)
|
|
|
|
context.stabilization_started = True
|
|
once = tuple(executor.list_directory(memory_root / ".band-aider-chkpts"))
|
|
context.stabilization_started = True
|
|
context.stabilization_started = False
|
|
|
|
assert len(once) == 1
|
|
assert len(executor.list_directory(memory_root / ".band-aider-chkpts")) == 2
|
|
|
|
|
|
def test_file_manager_checkpoint_preserves_durable_and_strips_runtime_state(memory_root: Path) -> None:
|
|
"""Pickled FileManager retains metadata but deliberately loses parser/client wiring."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
manager = _file_manager(memory_root, executor, test_files={"test_a.py": "from a import value\n\ndef test_a():\n assert value == 1\n"})
|
|
source = _source(manager, "src/a.py")
|
|
test = _test(manager, "tests/test_a.py")
|
|
manager.src_dirty[source.location] = True
|
|
manager.test_dirty[test.location] = True
|
|
source.summary = "changed source"
|
|
test.summary = "changed test"
|
|
|
|
restored = pickle.loads(pickle.dumps(manager))
|
|
|
|
assert restored.src_dirty == {source.location: True}
|
|
assert restored.test_dirty == {test.location: True}
|
|
assert restored.current_src(source.location) is not None
|
|
assert restored.current_test(test.location) is not None
|
|
assert not hasattr(restored, "parser")
|
|
assert not hasattr(restored, "client")
|
|
|
|
|
|
def test_file_parser_checkpoint_strips_runtime_client(memory_root: Path) -> None:
|
|
"""FileParser serialization drops only its runtime completion client."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
parser = FileParser(client=_Client(), repo_exec=executor, prompt_formatter_builder=PROMPT_FORMATTER)
|
|
|
|
restored = pickle.loads(pickle.dumps(parser))
|
|
|
|
assert not hasattr(restored, "client")
|
|
|
|
|
|
def test_stabilization_initialization_rebinds_scoped_runtime_collaborators(memory_root: Path) -> None:
|
|
"""A scoped manager receives parser/client and dependency state from its upstream owner."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
upstream = _file_manager(memory_root, executor)
|
|
scoped = FileManager.from_manager(upstream)
|
|
root = _source(upstream, "src/a.py")
|
|
context = StabilizationContext(goal="goal", classes=[root], upstream_file_manager=upstream, file_manager=scoped, root_change=root)
|
|
strategies = build_python_language_strategy_set(upstream)
|
|
strategies.bindings[TestRunnerStrategy] = _Runner()
|
|
loop = StabilizationLoop(
|
|
gate=MechanicalLoopGate(context, lambda: True, lambda: False),
|
|
context=context,
|
|
language_strategy_set=strategies,
|
|
)
|
|
|
|
scoped_manager = loop.context.file_manager
|
|
assert scoped_manager is not None
|
|
assert scoped_manager.parser is upstream.parser
|
|
assert scoped_manager.client is upstream.client
|
|
assert scoped_manager.upstream_file_manager is upstream
|
|
|
|
|
|
def test_stabilization_context_thaw_rebinds_runtime_file_manager(memory_root: Path) -> None:
|
|
"""Thawing a stabilization checkpoint restores current upstream runtime collaborators."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
upstream = _file_manager(memory_root, executor)
|
|
root = _source(upstream, "src/a.py")
|
|
context = StabilizationContext(goal="goal", classes=[root], upstream_file_manager=upstream, file_manager=FileManager.from_manager(upstream), root_change=root)
|
|
context.persist()
|
|
_configure(memory_root, executor, hydrate=True)
|
|
|
|
resumed = context_factory.thaw(StabilizationContext, goal="goal", classes=[root], upstream_file_manager=upstream, root_change=root, cot_index=0)
|
|
|
|
assert resumed is not None
|
|
assert resumed.file_manager is not None
|
|
assert resumed.file_manager.parser is upstream.parser
|
|
assert resumed.file_manager.client is upstream.client
|
|
|
|
|
|
def test_global_context_thaw_preserves_repository_coordinates(memory_root: Path) -> None:
|
|
"""A global checkpoint restores current memory-root coordinates and a rehydrated manager."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
context = _global_context(memory_root, executor)
|
|
context.file_manager
|
|
context.persist()
|
|
_configure(memory_root, executor, hydrate=True)
|
|
source_root = memory_root / "src"
|
|
test_root = memory_root / "tests"
|
|
resumed = context_factory.thaw(
|
|
GlobalContext,
|
|
goal="fresh",
|
|
parser=_Parser(source_root, test_root, executor),
|
|
client=_Client(),
|
|
repo_root=memory_root,
|
|
src_root=source_root,
|
|
test_root=test_root,
|
|
actor="test",
|
|
prompt_formatter_builder=PROMPT_FORMATTER,
|
|
intent_check_factory=IntentCheckFactory(_Client(), PROMPT_FORMATTER),
|
|
)
|
|
|
|
assert resumed is not None
|
|
assert resumed.repo_root == memory_root
|
|
assert resumed.file_manager.repo_root == memory_root
|
|
|
|
|
|
def test_global_context_thaw_preserves_traversed_frontier_lineage(memory_root: Path) -> None:
|
|
"""A global checkpoint retains durable prior-COT frontier lineage."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
context = _global_context(memory_root, executor)
|
|
context.story_cursor.task_cursor.prior_traversed_frontiers = TraversedFrontierHistory(
|
|
cot_index=1,
|
|
root_location="src/b.py",
|
|
frontiers=(("src/b.py",),),
|
|
previous=TraversedFrontierHistory(cot_index=0, root_location="src/a.py", frontiers=(("src/a.py",),)),
|
|
)
|
|
context.persist()
|
|
_configure(memory_root, executor, hydrate=True)
|
|
resumed = context_factory.thaw(
|
|
GlobalContext,
|
|
goal="fresh",
|
|
parser=_Parser(memory_root / "src", memory_root / "tests", executor),
|
|
client=_Client(),
|
|
repo_root=memory_root,
|
|
src_root=memory_root / "src",
|
|
test_root=memory_root / "tests",
|
|
actor="test",
|
|
prompt_formatter_builder=PROMPT_FORMATTER,
|
|
intent_check_factory=IntentCheckFactory(_Client(), PROMPT_FORMATTER),
|
|
)
|
|
assert resumed is not None
|
|
history = resumed.story_cursor.task_cursor.prior_traversed_frontiers
|
|
assert history is not None
|
|
|
|
assert history.find_location("src/a.py") == (0, 0)
|
|
assert history.find_location("src/b.py") == (1, 0)
|
|
|
|
|
|
def test_alpha_resume_before_pop_releases_unclaimed_inbound(memory_root: Path) -> None:
|
|
"""An Alpha inbound staged before pop disappears without queue growth on recovery."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
manager = _file_manager(memory_root, executor, test_files={"test_a.py": "pass\n"})
|
|
source = _source(manager, "src/a.py")
|
|
test = _test(manager, "tests/test_a.py")
|
|
context, raw = _alpha_context(manager, source, test)
|
|
context.move_inbound(StagedInboundItem(lane=BacklogLane.Inspection, value=raw.inspection[0], inbound_id="before-pop"))
|
|
|
|
context.recover_active_inbound()
|
|
|
|
assert context.active_inbound is None
|
|
assert len(raw.inspection) == 1
|
|
assert not raw.testing
|
|
|
|
|
|
def test_alpha_resume_after_pop_replays_one_testing_item(memory_root: Path) -> None:
|
|
"""Alpha recovery discards abandoned outbound state before one replayed flush."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
manager = _file_manager(memory_root, executor, test_files={"test_a.py": "pass\n"})
|
|
source = _source(manager, "src/a.py")
|
|
test = _test(manager, "tests/test_a.py")
|
|
context, raw = _alpha_context(manager, source, test)
|
|
assert context.move_inbound_from_backlog() is not None
|
|
context.stage_testing_outbound(TestingBacklog(focused_spotlight=test, summary="first", downstream_classes=[source]))
|
|
|
|
context.recover_active_inbound()
|
|
context.stage_testing_outbound(TestingBacklog(focused_spotlight=test, summary="replayed", downstream_classes=[source]))
|
|
context.flush_outbound_with_wal()
|
|
|
|
assert [item.summary for item in raw.testing] == ["replayed"]
|
|
assert context.active_inbound is None
|
|
|
|
|
|
def test_beta_resume_before_pop_releases_unclaimed_inbound(memory_root: Path) -> None:
|
|
"""A Beta inbound staged before pop disappears without queue growth on recovery."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
manager = _file_manager(memory_root, executor, test_files={"test_a.py": "pass\n"})
|
|
source = _source(manager, "src/a.py")
|
|
test = _test(manager, "tests/test_a.py")
|
|
context, raw = _beta_context(manager, source, test)
|
|
context.move_inbound(StagedInboundItem(lane=BacklogLane.Testing, value=raw.testing[0], inbound_id="before-pop"))
|
|
|
|
context.recover_active_inbound()
|
|
|
|
assert context.active_inbound is None
|
|
assert len(raw.testing) == 1
|
|
assert not raw.remediation
|
|
|
|
|
|
def test_beta_claim_admits_testing_work_to_the_testing_lane(memory_root: Path) -> None:
|
|
"""The Beta queue pump admits only testing work on its testing lane."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
manager = _file_manager(memory_root, executor, test_files={"test_a.py": "pass\n"})
|
|
source = _source(manager, "src/a.py")
|
|
test = _test(manager, "tests/test_a.py")
|
|
context, _ = _beta_context(manager, source, test)
|
|
|
|
assert context.move_inbound_from_backlog() is not None
|
|
assert context.active_inbound is not None
|
|
assert context.active_inbound.lane is BacklogLane.Testing
|
|
assert isinstance(context.active_inbound.value, TestingBacklog)
|
|
|
|
|
|
def test_beta_resume_after_pop_replays_one_remediation(memory_root: Path) -> None:
|
|
"""Beta recovery discards abandoned outbound remediation before one replayed flush."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
manager = _file_manager(memory_root, executor, test_files={"test_a.py": "pass\n"})
|
|
source = _source(manager, "src/a.py")
|
|
test = _test(manager, "tests/test_a.py")
|
|
context, raw = _beta_context(manager, source, test)
|
|
assert context.move_inbound_from_backlog() is not None
|
|
first = FailedQAItem(
|
|
focused_spotlight=source,
|
|
summary="first",
|
|
downstream_classes=[source],
|
|
failed_test=FailingTest(
|
|
test=test,
|
|
error_message="failure",
|
|
stack_trace=[],
|
|
raw_output="failure",
|
|
),
|
|
)
|
|
context.stage_remediation_outbound(first)
|
|
|
|
context.recover_active_inbound()
|
|
replayed = FailedQAItem(
|
|
focused_spotlight=source,
|
|
summary="replayed",
|
|
downstream_classes=[source],
|
|
failed_test=FailingTest(
|
|
test=test,
|
|
error_message="failure",
|
|
stack_trace=[],
|
|
raw_output="failure",
|
|
),
|
|
)
|
|
context.stage_remediation_outbound(replayed)
|
|
context.flush_outbound_with_wal()
|
|
|
|
assert [item.summary for item in raw.remediation] == ["replayed"]
|
|
assert not raw.testing
|
|
|
|
|
|
def test_phase_transitions_drop_future_guidance_and_raw_work_logs(memory_root: Path) -> None:
|
|
"""Phase handoffs retain localized assignments but discard raw edit and revival history."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
manager = _file_manager(memory_root, executor, test_files={"test_a.py": "pass\n"})
|
|
source = _source(manager, "src/a.py")
|
|
test = _test(manager, "tests/test_a.py")
|
|
future_guidance = (FutureGuidance(
|
|
origin_message_id="later",
|
|
addressed_message_id="earlier",
|
|
message="Avoid the dead-end premise.",
|
|
justification="Later evidence disproved the prior route.",
|
|
),)
|
|
|
|
alpha, alpha_raw = _alpha_context(manager, source, test)
|
|
assert alpha.move_inbound_from_backlog() is not None
|
|
alpha_item = TestingBacklog(
|
|
focused_spotlight=test,
|
|
summary="verify source",
|
|
downstream_classes=[source],
|
|
future_guidance=future_guidance,
|
|
editor_records=[EditorRecord("source repair", "return 1", "return 2")],
|
|
)
|
|
alpha.stage_testing_outbound(alpha_item)
|
|
alpha.flush_outbound_with_wal()
|
|
|
|
beta, beta_raw = _beta_context(manager, source, test)
|
|
assert beta.move_inbound_from_backlog() is not None
|
|
beta_item = FailedQAItem(
|
|
focused_spotlight=source,
|
|
summary="repair source",
|
|
downstream_classes=[source],
|
|
future_guidance=future_guidance,
|
|
editor_records=[EditorRecord("test repair", "assert 1", "assert 2")],
|
|
failed_test=FailingTest(
|
|
test=test,
|
|
error_message="failure",
|
|
stack_trace=[],
|
|
raw_output="failure",
|
|
),
|
|
)
|
|
beta.stage_remediation_outbound(beta_item)
|
|
beta.flush_outbound_with_wal()
|
|
|
|
assert alpha_item.future_guidance == ()
|
|
assert alpha_raw.testing[0].future_guidance == ()
|
|
assert beta_item.future_guidance == ()
|
|
assert beta_raw.remediation[0].future_guidance == ()
|
|
assert alpha_item.editor_records == alpha_raw.testing[0].editor_records == []
|
|
assert beta_item.editor_records == beta_raw.remediation[0].editor_records == []
|
|
assert alpha_raw.testing[0].summary == "verify source"
|
|
assert beta_raw.remediation[0].summary == "repair source"
|
|
|
|
|
|
def test_alpha_and_beta_checkpoint_payloads_exclude_runtime_collaborators(memory_root: Path) -> None:
|
|
"""Child contexts keep durable queues while omitting their runtime FileManager and backlog views."""
|
|
|
|
executor = MemoryRepoExec(memory_root)
|
|
_configure(memory_root, executor)
|
|
manager = _file_manager(memory_root, executor, test_files={"test_a.py": "pass\n"})
|
|
source = _source(manager, "src/a.py")
|
|
test = _test(manager, "tests/test_a.py")
|
|
alpha, _ = _alpha_context(manager, source, test)
|
|
beta, _ = _beta_context(manager, source, test)
|
|
|
|
alpha_payload = alpha.__getstate__()
|
|
beta_payload = beta.__getstate__()
|
|
|
|
assert "file_manager" not in alpha_payload
|
|
assert "backlog" not in alpha_payload
|
|
assert "file_manager" not in beta_payload
|
|
assert "backlog" not in beta_payload
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"invalid_shape",
|
|
(
|
|
"missing_topology",
|
|
"missing_frontier_hashes",
|
|
"missing_prior_lineage",
|
|
"retired_payload_field",
|
|
"topology_not_liveliness",
|
|
"missing_frontiers",
|
|
"missing_bfs_depth",
|
|
"missing_visited",
|
|
"missing_whitelist",
|
|
"missing_deferred_locations",
|
|
"missing_unchanged_perimeter",
|
|
"negative_hash_depth",
|
|
"non_mapping_hashes",
|
|
"non_mapping_depth_hashes",
|
|
"non_string_hash_location",
|
|
"non_string_hash_value",
|
|
"frontier_depth_mismatch",
|
|
"missing_active_frontier_baseline",
|
|
"wrong_frontier_baseline_depth",
|
|
"foreign_payload_field",
|
|
),
|
|
)
|
|
def test_stabilization_checkpoint_shape_rejects_invalid_durable_state(
|
|
invalid_shape: str,
|
|
) -> None:
|
|
"""Every distinct invalid durable shape is rejected before runtime hydration."""
|
|
|
|
root = _meta("src/a.py")
|
|
payload: dict[str, object] = {
|
|
"goal": "goal",
|
|
"classes": [root],
|
|
"topology": TopologicalLiveliness(),
|
|
"frontier_open_hashes": {},
|
|
"prior_traversed_frontiers": None,
|
|
}
|
|
topology = payload["topology"]
|
|
assert isinstance(topology, TopologicalLiveliness)
|
|
if invalid_shape == "missing_topology":
|
|
payload.pop("topology")
|
|
elif invalid_shape == "missing_frontier_hashes":
|
|
payload.pop("frontier_open_hashes")
|
|
elif invalid_shape == "missing_prior_lineage":
|
|
payload.pop("prior_traversed_frontiers")
|
|
elif invalid_shape in {"retired_payload_field", "foreign_payload_field"}:
|
|
payload[invalid_shape] = "retired"
|
|
elif invalid_shape == "topology_not_liveliness":
|
|
payload["topology"] = "not-a-topology"
|
|
elif invalid_shape == "missing_frontiers":
|
|
del topology.frontiers
|
|
elif invalid_shape == "missing_bfs_depth":
|
|
del topology.bfs_depth
|
|
elif invalid_shape == "missing_visited":
|
|
del topology.visited
|
|
elif invalid_shape == "missing_whitelist":
|
|
del topology.whitelist
|
|
elif invalid_shape == "missing_deferred_locations":
|
|
del topology.deferred_topology_locations
|
|
elif invalid_shape == "missing_unchanged_perimeter":
|
|
del topology.unchanged_perimeter
|
|
elif invalid_shape == "negative_hash_depth":
|
|
payload["frontier_open_hashes"] = {-1: {root.location: "hash"}}
|
|
elif invalid_shape == "non_mapping_hashes":
|
|
payload["frontier_open_hashes"] = []
|
|
elif invalid_shape == "non_mapping_depth_hashes":
|
|
payload["frontier_open_hashes"] = {0: []}
|
|
elif invalid_shape == "non_string_hash_location":
|
|
payload["frontier_open_hashes"] = {0: {1: "hash"}}
|
|
elif invalid_shape == "non_string_hash_value":
|
|
payload["frontier_open_hashes"] = {0: {root.location: 1}}
|
|
elif invalid_shape == "frontier_depth_mismatch":
|
|
topology.frontiers = [[root]]
|
|
payload["frontier_open_hashes"] = {1: {root.location: "hash"}}
|
|
elif invalid_shape == "missing_active_frontier_baseline":
|
|
topology.frontiers = [[root]]
|
|
topology.bfs_depth = 1
|
|
elif invalid_shape == "wrong_frontier_baseline_depth":
|
|
topology.frontiers = [[root]]
|
|
topology.bfs_depth = 1
|
|
payload["frontier_open_hashes"] = {0: {root.location: "hash"}}
|
|
|
|
assert not _checkpoint_shape_is_current(StabilizationContext, payload)
|
|
|
|
|
|
def _meta(location: str) -> ClassMeta:
|
|
"""Return source or test metadata with its correct durable ownership kind."""
|
|
|
|
content_type = ContentType.Test if location.startswith("tests/") else ContentType.Source
|
|
return ClassMeta(location, location, [], {}, content_type=content_type)
|