529 lines
19 KiB
Python
529 lines
19 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Sovereign-loop ownership exercised through nominal memory-only collaborators."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from pathlib import Path
|
|
from typing import TextIO
|
|
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.context.context_factory import _checkpoint_shape_is_current
|
|
from band_aider.context.global_context import GlobalContext, SovereignState, StoryCursor
|
|
from band_aider.context.provenance import Phase
|
|
from band_aider.context.resume_state import configure_resume_settings
|
|
from band_aider.context.working_backlog import InspectionBacklog
|
|
from band_aider.judgement.broad_or_narrow_check import SearchPolicy
|
|
from band_aider.judgement.code_parser import CodeParser
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.intent_check import IntentCheckFactory
|
|
from band_aider.judgement.parser_factory import ParserFactory
|
|
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
|
|
from band_aider.loop.mechanical_loop_gate import MechanicalLoopGate
|
|
from band_aider.loop.proto_loop import ProtoLoop
|
|
from band_aider.loop.sovereign_loop import SovereignLoop
|
|
from band_aider.loop.supreme_loop import SupremeLoop
|
|
from tests.support.completion_clients import (
|
|
StaticCompletionClient,
|
|
StaticOpenAICompatibleClient,
|
|
)
|
|
from tests.support.memory_repo_exec import MemoryRepoExec
|
|
|
|
|
|
class _Parser(CodeParser):
|
|
"""Supply deterministic metadata without reading a host-backed parser pipeline."""
|
|
|
|
def parse_file(
|
|
self,
|
|
file_name: str,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Return minimal metadata for one requested memory-only location."""
|
|
|
|
del prompt
|
|
return ClassMeta(file_name, "fixture", [], {}, content_type=content_type)
|
|
|
|
def parse_stream(
|
|
self,
|
|
file_name: str,
|
|
stream: TextIO,
|
|
prompt: str,
|
|
content_type: ContentType = ContentType.Source,
|
|
) -> ClassMeta:
|
|
"""Delegate stream parsing to the same explicit file metadata fixture."""
|
|
|
|
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]:
|
|
"""Return one stationary metadata record per requested location."""
|
|
|
|
return [self.parse_file(file_name, prompt, content_type) for file_name in file_names]
|
|
|
|
|
|
class _Context(GlobalContext):
|
|
"""Own the real GlobalContext cursor shape with a memory-backed manager."""
|
|
|
|
def __init__(
|
|
self,
|
|
stories: list[str],
|
|
current_story_index: int,
|
|
active_story: str,
|
|
sovereign_phase: SovereignState = SovereignState.BETWEEN_STORIES,
|
|
) -> None:
|
|
"""Build durable story coordinates without checkpoint or filesystem activity."""
|
|
|
|
self.goal = active_story
|
|
self.original_prompt = active_story
|
|
self.stories = list(stories)
|
|
self.story_cursor = StoryCursor(
|
|
current_story_index=current_story_index,
|
|
active_story=active_story,
|
|
sovereign_phase=sovereign_phase,
|
|
)
|
|
self.supreme_completion_state = SovereignState.SUPREME_ACTIVE
|
|
self.supreme_result = ""
|
|
self.supreme_reported = False
|
|
self.intent = None
|
|
self.persisted: list[tuple[int, str, SovereignState]] = []
|
|
self.sync_calls = 0
|
|
self.parser = _Parser()
|
|
self.client = StaticOpenAICompatibleClient('{"summary": "fixture"}')
|
|
self.repo_root = Path("/testbed")
|
|
self.src_root = self.repo_root / "src"
|
|
self.test_root = self.repo_root / "tests"
|
|
self.actor = "test"
|
|
self.prompt_formatter_builder = MarkdownPromptFormatterBuilder()
|
|
self.intent_check_factory = IntentCheckFactory(
|
|
StaticCompletionClient('{"intent": "existing_files"}'),
|
|
self.prompt_formatter_builder,
|
|
)
|
|
self.search_anchors: tuple[str, ...] = ()
|
|
self.parser_factory: object | None = None
|
|
self.logger = logging.getLogger("test.sovereign.context")
|
|
self.relevant_files: list[ClassMeta] = []
|
|
self.marginal_result: list[str] = []
|
|
self.repo_exec = MemoryRepoExec(self.repo_root)
|
|
self._file_manager = FileManager(
|
|
parser=self.parser,
|
|
client=self.client,
|
|
repo_root=self.repo_root,
|
|
src_root=self.src_root,
|
|
test_root=self.test_root,
|
|
actor=self.actor,
|
|
prompt_formatter_builder=self.prompt_formatter_builder,
|
|
repo_exec=self.repo_exec,
|
|
)
|
|
|
|
def sync(self) -> None:
|
|
"""Record fresh context synchronization without any host repository operation."""
|
|
|
|
self.sync_calls += 1
|
|
|
|
def persist(self, keep: int | None = None) -> Path:
|
|
"""Record durable coordinate transitions without writing a checkpoint."""
|
|
|
|
del keep
|
|
self.persisted.append((
|
|
self.story_cursor.current_story_index,
|
|
self.story_cursor.active_story,
|
|
self.story_cursor.sovereign_phase,
|
|
))
|
|
return self.repo_root / "GlobalContext-test.pkl"
|
|
|
|
|
|
class _StationarySupremeLoop(SupremeLoop):
|
|
"""Represent a created Supreme child without starting workers or a process."""
|
|
|
|
def __init__(self, context: GlobalContext) -> None:
|
|
"""Bind the actual selected context as the child-loop owner."""
|
|
|
|
self.context = context
|
|
self._logger_stack = []
|
|
|
|
def run(self, logger_name: str | None = None) -> str:
|
|
"""Return the context-owned result without performing external work."""
|
|
|
|
del logger_name
|
|
return self.context.supreme_result
|
|
|
|
|
|
class _PreparationHarness(SovereignLoop):
|
|
"""Provide nominal fresh-context construction for the inherited preparation workflow."""
|
|
|
|
def __init__(self, fresh_context: _Context) -> None:
|
|
"""Build the real sovereign loop and retain one explicit new-story context."""
|
|
|
|
super().__init__(
|
|
original_prompt="original",
|
|
client=StaticOpenAICompatibleClient('{"summary": "fixture"}'),
|
|
repo_root=Path("/testbed"),
|
|
src_root=Path("/testbed/src"),
|
|
test_root=Path("/testbed/tests"),
|
|
actor="test",
|
|
intent_check_factory=IntentCheckFactory(
|
|
StaticCompletionClient('{"intent": "existing_files"}'),
|
|
MarkdownPromptFormatterBuilder(),
|
|
),
|
|
aider_worker=None,
|
|
repo_exec=MemoryRepoExec(),
|
|
language="python",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
)
|
|
self.context = _Context(["fresh story"], 0, "fresh story")
|
|
self.gate.context = self.context
|
|
self._fresh_context = fresh_context
|
|
self.created_children: list[GlobalContext] = []
|
|
|
|
def _stories_for_new_run(self) -> list[str]:
|
|
"""Expose one pre-approved fresh story without invoking an LLM decomposition."""
|
|
|
|
return ["fresh story"]
|
|
|
|
def _classify_story(self, story: str) -> SearchPolicy:
|
|
"""Return the stationary search policy owned by this fresh-story fixture."""
|
|
|
|
assert story == "fresh story"
|
|
return SearchPolicy("narrow", "named target", ("Target",))
|
|
|
|
def _create_supreme_context(
|
|
self,
|
|
*,
|
|
story: str,
|
|
stories: list[str],
|
|
story_index: int,
|
|
policy: SearchPolicy,
|
|
parser: CodeParser,
|
|
parser_factory: ParserFactory,
|
|
) -> GlobalContext:
|
|
"""Return the nominal memory context selected by inherited preparation."""
|
|
|
|
del story, stories, story_index, policy, parser, parser_factory
|
|
return self._fresh_context
|
|
|
|
def _create_supreme_loop(
|
|
self,
|
|
context: GlobalContext,
|
|
parser: CodeParser,
|
|
) -> SupremeLoop:
|
|
"""Return a nominal child bound to the exact context prepared by the parent."""
|
|
|
|
del parser
|
|
self.created_children.append(context)
|
|
return _StationarySupremeLoop(context)
|
|
|
|
|
|
class _CompletingSupremeLoop(_StationarySupremeLoop):
|
|
"""Complete one fixed story result through the actual Sovereign child boundary."""
|
|
|
|
def __init__(self, context: GlobalContext, result: str) -> None:
|
|
"""Bind one stationary story result to the context selected by Sovereign."""
|
|
|
|
super().__init__(context)
|
|
self._result = result
|
|
|
|
def run(self, logger_name: str | None = None) -> str:
|
|
"""Publish completion state before returning the fixed child summary."""
|
|
|
|
del logger_name
|
|
self.context.supreme_result = self._result
|
|
self.context.supreme_completion_state = SovereignState.BETWEEN_STORIES
|
|
return self._result
|
|
|
|
|
|
class _RunHarness(SovereignLoop):
|
|
"""Exercise real Sovereign story transitions with stationary child completions."""
|
|
|
|
def __init__(self) -> None:
|
|
"""Build a two-story memory context and retain its final report evidence."""
|
|
|
|
super().__init__(
|
|
original_prompt="ORIGINAL_STORY_PROMPT_SENTINEL",
|
|
client=StaticOpenAICompatibleClient('{"summary": "fixture"}'),
|
|
repo_root=Path("/testbed"),
|
|
src_root=Path("/testbed/src"),
|
|
test_root=Path("/testbed/tests"),
|
|
actor="test",
|
|
intent_check_factory=IntentCheckFactory(
|
|
StaticCompletionClient('{"intent": "existing_files"}'),
|
|
MarkdownPromptFormatterBuilder(),
|
|
),
|
|
aider_worker=None,
|
|
repo_exec=MemoryRepoExec(),
|
|
language="python",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
)
|
|
context = _Context(["first story", "second story"], 0, "first story")
|
|
self.context = context
|
|
self.gate = MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: False,
|
|
)
|
|
self._results = ["SUCCESS: first story", "SUCCESS: second story"]
|
|
self.reports: list[str] = []
|
|
|
|
def _prepare_work(self) -> ProtoLoop | None:
|
|
"""Return one stationary child while preserving the parent-owned transition."""
|
|
|
|
if not self._results:
|
|
return None
|
|
return _CompletingSupremeLoop(self.context, self._results.pop(0))
|
|
|
|
def _write_final_report(self, result: str) -> None:
|
|
"""Record report authority without writing a host-backed report fixture."""
|
|
|
|
self.reports.append(result)
|
|
|
|
|
|
def _loop(*, repo_exec: MemoryRepoExec | None = None) -> SovereignLoop:
|
|
"""Build a fully nominal SovereignLoop with memory-only runtime collaborators."""
|
|
|
|
executor = repo_exec or MemoryRepoExec()
|
|
loop = SovereignLoop(
|
|
original_prompt="original",
|
|
client=StaticOpenAICompatibleClient('{"stories": ["first"]}'),
|
|
repo_root=Path("/testbed"),
|
|
src_root=Path("/testbed/src"),
|
|
test_root=Path("/testbed/tests"),
|
|
actor="test",
|
|
intent_check_factory=IntentCheckFactory(
|
|
StaticCompletionClient('{"intent": "existing_files"}'),
|
|
MarkdownPromptFormatterBuilder(),
|
|
),
|
|
aider_worker=None,
|
|
repo_exec=executor,
|
|
language="python",
|
|
prompt_formatter_builder=MarkdownPromptFormatterBuilder(),
|
|
)
|
|
context = _Context(["first", "second"], 0, "first")
|
|
loop.context = context
|
|
loop.gate = MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: True,
|
|
should_continue=lambda: False,
|
|
)
|
|
loop._logger_stack = []
|
|
loop._prepared = False
|
|
return loop
|
|
|
|
|
|
def test_sovereign_is_a_proto_loop() -> None:
|
|
"""Sovereign participates in the normal parent-loop hierarchy."""
|
|
|
|
assert issubclass(SovereignLoop, ProtoLoop)
|
|
|
|
|
|
def test_fresh_story_parser_factory_uses_search_policy() -> None:
|
|
"""Fresh story policy creates a narrow parser factory with its exact anchors."""
|
|
|
|
loop = _loop()
|
|
|
|
parser_factory = loop._parser_factory_for_policy(
|
|
SearchPolicy("narrow", "named target", ("Target",)),
|
|
)
|
|
|
|
assert parser_factory.scope == "narrow"
|
|
assert parser_factory.policy_anchors == ("Target",)
|
|
assert isinstance(parser_factory(), CodeParser)
|
|
|
|
|
|
def test_sovereign_shares_one_graph_with_its_supreme_child() -> None:
|
|
"""One outer task owns the graph used by each created Supreme story."""
|
|
|
|
loop = _loop()
|
|
|
|
child = SovereignLoop._create_supreme_loop(loop, loop.context, loop.context.parser)
|
|
|
|
assert child.provenance is loop.provenance
|
|
|
|
|
|
def test_sovereign_resets_or_restores_provenance_from_resume_settings() -> None:
|
|
"""A fresh task selects a graph, while a resumed task restores that graph."""
|
|
|
|
executor = MemoryRepoExec()
|
|
checkpoint_dir = executor.root / "checkpoints"
|
|
configure_resume_settings(
|
|
hydrate=False,
|
|
checkpoint_dir=checkpoint_dir,
|
|
repo_exec=executor,
|
|
)
|
|
first = _loop(repo_exec=executor)
|
|
message_id = first.provenance.start(
|
|
Phase.ALPHA,
|
|
InspectionBacklog(ClassMeta("src/history.py", "fixture", [], {}), "history"),
|
|
)
|
|
|
|
configure_resume_settings(
|
|
hydrate=True,
|
|
checkpoint_dir=checkpoint_dir,
|
|
repo_exec=executor,
|
|
)
|
|
resumed = _loop(repo_exec=executor)
|
|
|
|
assert resumed.provenance.graph_id == first.provenance.graph_id
|
|
assert message_id in resumed.provenance.messages
|
|
|
|
configure_resume_settings(
|
|
hydrate=False,
|
|
checkpoint_dir=checkpoint_dir,
|
|
repo_exec=executor,
|
|
)
|
|
fresh = _loop(repo_exec=executor)
|
|
|
|
assert fresh.provenance.graph_id != first.provenance.graph_id
|
|
assert fresh.provenance.messages == {}
|
|
|
|
|
|
def test_fresh_prepare_work_creates_parser_without_checkpoint_shape() -> None:
|
|
"""Inherited preparation synchronizes the supplied fresh context once."""
|
|
|
|
fresh_context = _Context(["fresh story"], 0, "fresh story")
|
|
loop = _PreparationHarness(fresh_context)
|
|
|
|
child = loop._prepare_work()
|
|
|
|
assert isinstance(child, _StationarySupremeLoop)
|
|
assert fresh_context.sync_calls == 1
|
|
assert loop._prepared is True
|
|
assert loop.created_children == [fresh_context]
|
|
|
|
|
|
def test_story_decomposition_persists_before_story_setup() -> None:
|
|
"""Persisted story lists normalize to a fresh ordered sequence."""
|
|
|
|
loop = _loop()
|
|
|
|
assert loop._normalized_stories({"stories": ["first", "second"]}) == [
|
|
"first",
|
|
"second",
|
|
]
|
|
assert loop._normalized_stories({"stories": []}) == []
|
|
|
|
|
|
def test_advance_story_only_mutates_and_persists() -> None:
|
|
"""Story advancement changes only the durable coordinate and persists it."""
|
|
|
|
loop = _loop()
|
|
|
|
loop._advance_story()
|
|
|
|
assert loop.context.story_cursor.current_story_index == 1
|
|
assert loop.context.story_cursor.active_story == "second"
|
|
assert loop.context.story_cursor.sovereign_phase == SovereignState.BETWEEN_STORIES
|
|
assert isinstance(loop.context, _Context)
|
|
assert loop.context.persisted == [(1, "second", SovereignState.BETWEEN_STORIES)]
|
|
|
|
|
|
def test_final_story_completion_does_not_reenter_outer_gate() -> None:
|
|
"""Aggregating final story results is independent of parent gate re-entry."""
|
|
|
|
loop = _loop()
|
|
loop.story_results = ["SUCCESS: first", "FAILED: second"]
|
|
|
|
assert loop._aggregate_story_results() == "FAILED: SUCCESS: first | FAILED: second"
|
|
|
|
|
|
def test_sovereign_run_advances_every_memory_owned_story_before_aggregation() -> None:
|
|
"""Two completed children advance the durable cursor and produce one final result."""
|
|
|
|
loop = _RunHarness()
|
|
|
|
result = loop.run()
|
|
|
|
assert result == "SUCCESS: SUCCESS: first story | SUCCESS: second story"
|
|
assert loop.story_results == ["SUCCESS: first story", "SUCCESS: second story"]
|
|
assert loop.reports == [result]
|
|
assert loop.context.story_cursor.current_story_index == 2
|
|
assert loop.context.story_cursor.sovereign_phase is SovereignState.COMPLETED
|
|
assert loop.context.supreme_completion_state is SovereignState.SUPREME_FINISHED_UNREPORTED
|
|
assert loop.context.supreme_reported is True
|
|
|
|
|
|
def test_between_story_hydration_does_not_consume_stale_completion_twice() -> None:
|
|
"""A between-story payload is not considered an active Supreme checkpoint."""
|
|
|
|
loop = _loop()
|
|
payload = {
|
|
"stories": ["first", "second", "third"],
|
|
"story_cursor": StoryCursor(
|
|
current_story_index=1,
|
|
active_story="second",
|
|
sovereign_phase=SovereignState.BETWEEN_STORIES,
|
|
),
|
|
"goal": "second",
|
|
"supreme_completion_state": SovereignState.SUPREME_FINISHED_UNREPORTED,
|
|
}
|
|
|
|
assert not loop._active_supreme_checkpoint_matches(payload)
|
|
assert loop._normalized_stories(payload) == ["first", "second", "third"]
|
|
|
|
|
|
def test_current_shape_policy_rejects_pre_sovereign_context() -> None:
|
|
"""Temporary checkpoints without the current cursor shape are discarded by design."""
|
|
|
|
assert not _checkpoint_shape_is_current(
|
|
GlobalContext,
|
|
{"goal": "old prompt", "classes": []},
|
|
)
|
|
|
|
|
|
def test_active_checkpoint_matching_is_explicit() -> None:
|
|
"""Only a matching durable active-story coordinate is resumable as Supreme."""
|
|
|
|
loop = _loop()
|
|
payload = {
|
|
"story_cursor": StoryCursor(
|
|
current_story_index=0,
|
|
active_story="first",
|
|
sovereign_phase=SovereignState.SUPREME_ACTIVE,
|
|
),
|
|
"goal": "first",
|
|
"stories": ["first", "second"],
|
|
"supreme_completion_state": SovereignState.SUPREME_ACTIVE,
|
|
}
|
|
|
|
assert loop._active_supreme_checkpoint_matches(payload)
|
|
payload["goal"] = "other"
|
|
assert not loop._active_supreme_checkpoint_matches(payload)
|
|
|
|
|
|
def test_hydration_thaws_once_and_runs_active_child_immediately() -> None:
|
|
"""The active checkpoint predicate admits one matching durable child coordinate."""
|
|
|
|
loop = _loop()
|
|
payload = {
|
|
"story_cursor": StoryCursor(
|
|
current_story_index=0,
|
|
active_story="first",
|
|
sovereign_phase=SovereignState.SUPREME_ACTIVE,
|
|
search_policy_scope="strict",
|
|
search_policy_reason="named",
|
|
),
|
|
"goal": "first",
|
|
"stories": ["first", "second"],
|
|
"search_anchors": ("Target",),
|
|
"supreme_completion_state": SovereignState.SUPREME_ACTIVE,
|
|
}
|
|
|
|
assert loop._active_supreme_checkpoint_matches(payload)
|
|
assert payload["story_cursor"].search_policy_scope == "strict"
|
|
assert payload["search_anchors"] == ("Target",)
|
|
|
|
|
|
def test_sovereign_and_supreme_have_one_way_ownership() -> None:
|
|
"""Only Sovereign owns story progression and final-result aggregation."""
|
|
|
|
assert hasattr(SovereignLoop, "_advance_story")
|
|
assert hasattr(SovereignLoop, "_aggregate_story_results")
|
|
assert not hasattr(SupremeLoop, "_advance_story")
|
|
assert not hasattr(SupremeLoop, "_aggregate_story_results")
|