1351 lines
55 KiB
Python
1351 lines
55 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Mapping
|
|
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import random
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from band_aider.ai_client import llm_logging
|
|
from band_aider.context.bootstrap_context import BootstrapContext
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.alpha_blank_source_correction_check import (
|
|
AlphaBlankSourceCorrectionDecision,
|
|
)
|
|
from band_aider.judgement.bootstrap_acceptance_check import BootstrapAcceptanceCheck
|
|
from band_aider.judgement.code_edit_or_leave_it import (
|
|
consume_code_edit_leave_summary,
|
|
is_code_edit_leave_summary,
|
|
)
|
|
from band_aider.judgement.code_dependency_graph_maintainer import (
|
|
CodeDependencyGraphMaintainer,
|
|
)
|
|
from band_aider.judgement.code_editor import CodeEditor
|
|
from band_aider.judgement.contextual_file_renderer import ContextualFileRenderer
|
|
from band_aider.judgement.cross_cutting_concern_check import (
|
|
CrossCuttingConcernCheckFactory,
|
|
)
|
|
from band_aider.judgement.fidelity import Fidelity
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.source_preflight_decision import (
|
|
BootstrapSourcePreflightDecision,
|
|
)
|
|
from band_aider.judgement.repository_idiom_check import (
|
|
RepositoryIdiomCheckFactory,
|
|
idiom_entry_key,
|
|
)
|
|
from band_aider.judgement.tighten_design_work_check import (
|
|
TightenDesignWorkCheckFactory,
|
|
)
|
|
from band_aider.judgement.relevant_beat_selector import RelevantBeatSelector
|
|
from band_aider.judgement.source_discovery_strategy import (
|
|
SourceDiscoveryStrategy,
|
|
searchable_beat_texts,
|
|
)
|
|
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.strategies import LanguageStrategySet
|
|
from band_aider.ui.status import publish_bootstrap_status, publish_loop_status
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class _BootstrapEditorFailure:
|
|
"""_BootstrapEditorFailure adapts mechanical bootstrap rejection into retry guidance."""
|
|
|
|
reason: str
|
|
additional: dict[str, str]
|
|
|
|
|
|
@dataclass(init=False)
|
|
class BootstrapLoop(ProtoLoop[LoopGate, BootstrapContext]):
|
|
"""BootstrapLoop maps and generates reusable constructs before ordinary COT work."""
|
|
|
|
language_strategy_set: LanguageStrategySet
|
|
aider_worker: CodeEditor
|
|
cross_cutting_concern_check_factory: CrossCuttingConcernCheckFactory
|
|
repository_idiom_check_factory: RepositoryIdiomCheckFactory
|
|
tighten_design_work_check_factory: TightenDesignWorkCheckFactory
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
context: BootstrapContext,
|
|
language_strategy_set: LanguageStrategySet,
|
|
aider_worker: CodeEditor,
|
|
cross_cutting_concern_check_factory: CrossCuttingConcernCheckFactory | None = None,
|
|
repository_idiom_check_factory: RepositoryIdiomCheckFactory | None = None,
|
|
tighten_design_work_check_factory: TightenDesignWorkCheckFactory | None = None,
|
|
logger: logging.Logger | None = None,
|
|
) -> None:
|
|
"""Create a bootstrap mapping loop with deterministic structural gating."""
|
|
|
|
self.language_strategy_set = language_strategy_set
|
|
self.aider_worker = aider_worker
|
|
self.cross_cutting_concern_check_factory = (
|
|
cross_cutting_concern_check_factory
|
|
or CrossCuttingConcernCheckFactory(
|
|
context.upstream_file_manager.client,
|
|
context.upstream_file_manager.prompt_formatter_builder,
|
|
)
|
|
)
|
|
self.repository_idiom_check_factory = (
|
|
repository_idiom_check_factory
|
|
or RepositoryIdiomCheckFactory(
|
|
context.upstream_file_manager.client,
|
|
context.upstream_file_manager.prompt_formatter_builder,
|
|
)
|
|
)
|
|
self.tighten_design_work_check_factory = (
|
|
tighten_design_work_check_factory
|
|
or TightenDesignWorkCheckFactory(
|
|
context.upstream_file_manager.client,
|
|
context.upstream_file_manager.prompt_formatter_builder,
|
|
)
|
|
)
|
|
gate = MechanicalLoopGate(
|
|
context=context,
|
|
may_enter=lambda: (
|
|
not context.completed
|
|
and context.iteration < context.max_iterations
|
|
),
|
|
should_continue=lambda: (
|
|
not context.completed
|
|
and context.iteration < context.max_iterations
|
|
),
|
|
should_retry=lambda: False,
|
|
)
|
|
super().__init__(gate=gate, context=context, logger=logger)
|
|
self._bind_editor_file_manager()
|
|
|
|
def _prepare_work(self) -> ProtoLoop | None:
|
|
"""Return None because bootstrap owns its internal ordered stages directly."""
|
|
|
|
return None
|
|
|
|
def _post_run(self) -> str:
|
|
"""Return a concise bootstrap mapping summary."""
|
|
|
|
if self.context.blocked_summary:
|
|
return f"Bootstrap mapping stopped: {self.context.blocked_summary}"
|
|
if self.context.created_locations or self.context.involved_locations:
|
|
return (
|
|
"Bootstrap mapping completed across "
|
|
f"{len(self.context.created_locations)} created source file(s) and "
|
|
f"{len(self.context.involved_locations)} involved source file(s)."
|
|
)
|
|
return "Bootstrap mapping completed without creating reusable constructs."
|
|
|
|
def _publish_status(self, current_item: str | None = None) -> None:
|
|
"""Publish bootstrap progress in the terminal status panels."""
|
|
|
|
item = current_item or f"bootstrap: {self.context.stage.replace('_', ' ')}"
|
|
publish_loop_status(
|
|
previous_loop="SupremeLoop",
|
|
current_loop="BootstrapLoop",
|
|
next_loop="SupremeLoop",
|
|
current_item=item,
|
|
)
|
|
publish_bootstrap_status(self.context)
|
|
|
|
def run(self, logger_name: str | None = None) -> str:
|
|
"""Run the bootstrap mapping state machine until completion or the hard cap."""
|
|
|
|
if not self._logger_stack:
|
|
self._logger_stack.append(llm_logging.logger_for_loop(self))
|
|
if logger_name:
|
|
self._logger_stack.append(
|
|
self.logger.getChild(llm_logging.safe_log_segment(logger_name))
|
|
)
|
|
llm_logging.set_current_logger(self.logger)
|
|
try:
|
|
if not self.gate.may_enter():
|
|
if not self.context.completed:
|
|
self.context.mark_completed()
|
|
self._publish_status()
|
|
return self._post_run()
|
|
|
|
while self.gate.should_continue():
|
|
stage = self.context.stage
|
|
self._publish_status()
|
|
if stage in {"initialized", "extract_beats"}:
|
|
self._extract_visible_beats()
|
|
continue
|
|
if stage == "beats_extracted":
|
|
self._run_concern_check()
|
|
continue
|
|
if stage == "concern_selected":
|
|
self._prepare_idiom_pairs()
|
|
continue
|
|
if stage == "cataloging_idioms":
|
|
self._run_idiom_catalog()
|
|
continue
|
|
if stage == "designing":
|
|
self._run_design_check()
|
|
continue
|
|
if stage == "editing":
|
|
self._run_editor()
|
|
continue
|
|
if stage == "refreshing":
|
|
self.context.mark_completed()
|
|
continue
|
|
self.context.mark_blocked(f"Unknown bootstrap stage {stage}.")
|
|
if not self.context.completed:
|
|
self.context.mark_completed()
|
|
self._publish_status()
|
|
return self._post_run()
|
|
finally:
|
|
if logger_name:
|
|
self._logger_stack.pop()
|
|
llm_logging.set_current_logger(self.logger if self._logger_stack else None)
|
|
|
|
def _bind_editor_file_manager(self) -> None:
|
|
"""Bind the current global FileManager to the injected editor when supported."""
|
|
|
|
try:
|
|
setattr(self.aider_worker, "file_manager", self.context.upstream_file_manager)
|
|
except Exception:
|
|
return
|
|
|
|
def _visible_files(self) -> list[ClassMeta]:
|
|
"""Return current source metadata for visible bootstrap locations."""
|
|
|
|
files: list[ClassMeta] = []
|
|
for location in self.context.visible_locations:
|
|
current = self.context.upstream_file_manager.current_src(location)
|
|
if current is not None:
|
|
files.append(current)
|
|
return files
|
|
|
|
def _extract_visible_beats(self) -> None:
|
|
"""Extract complete selected searchable beats from the current visible source cohort."""
|
|
|
|
self._publish_status("Mapping: extracting searchable beats")
|
|
discovery = self.language_strategy_set[SourceDiscoveryStrategy]
|
|
beats_by_location: dict[str, list[str]] = {}
|
|
visible_files = self._visible_files()
|
|
if not visible_files:
|
|
self.context.beats_by_location = {}
|
|
self.context.visible_locations = []
|
|
self.context.selected_concern = self._standalone_feature_concern()
|
|
self.context.cross_cutting_concerns = [self.context.selected_concern]
|
|
self.context.pending_idiom_pairs = []
|
|
self.context.persist()
|
|
self._publish_status("Mapping: standalone feature concern")
|
|
self.context.mark_stage("concern_selected")
|
|
return
|
|
|
|
visible_locations = [
|
|
file.location
|
|
for file in visible_files
|
|
]
|
|
for file in visible_files:
|
|
try:
|
|
file_manager = self.context.upstream_file_manager
|
|
source = file_manager.repo_exec.read_text(
|
|
file_manager.source_path(file.location), encoding="utf-8"
|
|
)
|
|
raw_beats = discovery.searchable_beats(file.location, source)
|
|
beats = RelevantBeatSelector(
|
|
anchors=getattr(self.context, "search_anchors", ()),
|
|
request_text=self.context.goal,
|
|
)(searchable_beat_texts(raw_beats))
|
|
except Exception as error:
|
|
self.logger.warning(
|
|
"Bootstrap beat extraction failed for %s: %s",
|
|
file.location,
|
|
error,
|
|
)
|
|
beats = ()
|
|
beats_by_location[file.location] = list(beats)
|
|
|
|
self.context.beats_by_location = beats_by_location
|
|
self.context.visible_locations = visible_locations
|
|
self._publish_status("Mapping: beats extracted")
|
|
self.context.mark_stage("beats_extracted")
|
|
|
|
def _standalone_feature_concern(self) -> dict[str, object]:
|
|
"""Return a synthetic bootstrap concern for new-file feature seeding."""
|
|
|
|
exhibits = [
|
|
{
|
|
"location": location,
|
|
"beat": (self.context.beats_by_location.get(location) or [location])[0],
|
|
"symbol": "",
|
|
"reason": "Visible source participates in the requested standalone feature wiring.",
|
|
}
|
|
for location in self.context.visible_locations
|
|
]
|
|
return {
|
|
"name": "Standalone feature source",
|
|
"summary": (
|
|
"The feature request requires bootstrap to create a source file before "
|
|
"ordinary stabilization can proceed."
|
|
),
|
|
"exhibits": exhibits,
|
|
}
|
|
|
|
def _run_concern_check(self) -> None:
|
|
"""Run Prompt 1 and store the selected cross-cutting concern."""
|
|
|
|
self._publish_status("Mapping: selecting cross-cutting concern")
|
|
check = self.cross_cutting_concern_check_factory.create(
|
|
original_prompt=self.context.goal,
|
|
visible_files=self._visible_files(),
|
|
beats_by_location=self.context.beats_by_location,
|
|
)
|
|
if not check():
|
|
if getattr(check, "outcome", "") == "invalid_found":
|
|
self.logger.warning(
|
|
"Prompt 1 returned invalid bootstrap concern; falling back to standalone feature source: %s",
|
|
check.reason,
|
|
)
|
|
self.context.cross_cutting_concerns = []
|
|
self.context.selected_concern = self._standalone_feature_concern()
|
|
self.context.cross_cutting_concerns = [self.context.selected_concern]
|
|
self.context.pending_idiom_pairs = []
|
|
self.context.persist()
|
|
self._publish_status("Mapping: standalone feature concern")
|
|
self.context.mark_stage("concern_selected")
|
|
return
|
|
|
|
concerns = list(check.additional.get("concerns") or [])
|
|
selected_index = check.additional.get("selected_concern_index")
|
|
selected = (
|
|
concerns[selected_index]
|
|
if isinstance(selected_index, int)
|
|
and 0 <= selected_index < len(concerns)
|
|
else None
|
|
)
|
|
if not isinstance(selected, dict):
|
|
self.context.selected_concern = self._standalone_feature_concern()
|
|
self.context.cross_cutting_concerns = [self.context.selected_concern]
|
|
self.context.pending_idiom_pairs = []
|
|
self.context.persist()
|
|
self._publish_status("Mapping: standalone feature concern")
|
|
self.context.mark_stage("concern_selected")
|
|
return
|
|
self.context.cross_cutting_concerns = concerns
|
|
self.context.selected_concern = selected
|
|
self.context.pending_idiom_pairs = []
|
|
self.context.persist()
|
|
self._publish_status("Mapping: concern selected")
|
|
self.context.mark_stage("concern_selected")
|
|
|
|
def _prepare_idiom_pairs(self) -> None:
|
|
"""Persist deterministic Prompt 2 sample pairs before cataloging starts."""
|
|
|
|
if not self.context.pending_idiom_pairs:
|
|
self.context.pending_idiom_pairs = self._deterministic_pairs(
|
|
self.context.visible_locations
|
|
)
|
|
self.context.persist()
|
|
self._publish_status("Mapping: idiom samples selected")
|
|
self.context.mark_stage("cataloging_idioms")
|
|
|
|
def _run_idiom_catalog(self) -> None:
|
|
"""Run Prompt 2 for every remaining deterministic file sample pair."""
|
|
|
|
total = len(self.context.pending_idiom_pairs)
|
|
processed = 0
|
|
while self.context.pending_idiom_pairs:
|
|
pair = self.context.pending_idiom_pairs[0]
|
|
processed += 1
|
|
self._publish_status(
|
|
"Mapping: cataloging repository idioms "
|
|
f"{processed}/{total}"
|
|
)
|
|
sample_files = [
|
|
file
|
|
for file in self._visible_files()
|
|
if file.location in pair
|
|
]
|
|
renderer = ContextualFileRenderer.build(
|
|
file_manager=self.context.upstream_file_manager,
|
|
client=self.context.upstream_file_manager.client,
|
|
editable_files=[],
|
|
readonly_files=sample_files,
|
|
prompt=self.context.goal,
|
|
guidance=(
|
|
str(
|
|
(self.context.selected_concern or {}).get("summary")
|
|
or ""
|
|
),
|
|
),
|
|
logger=self.logger.getChild("contextual-file-renderer"),
|
|
prompt_formatter_builder=self.context.upstream_file_manager.prompt_formatter_builder,
|
|
)
|
|
sources_by_location = {
|
|
file.location: renderer.read(
|
|
file,
|
|
Fidelity.AUTO | Fidelity.CDC,
|
|
)
|
|
for file in sample_files
|
|
}
|
|
check = self.repository_idiom_check_factory.create(
|
|
original_prompt=self.context.goal,
|
|
selected_concern=self.context.selected_concern or {},
|
|
sample_files=sample_files,
|
|
sources_by_location=sources_by_location,
|
|
reference_cards=renderer.get_reference_card_evidence(),
|
|
)
|
|
check()
|
|
self._merge_idiom_entries(list(check.additional.get("entries") or []))
|
|
self.context.pending_idiom_pairs = self.context.pending_idiom_pairs[1:]
|
|
self.context.persist()
|
|
self._publish_status(
|
|
"Mapping: cataloged repository idioms "
|
|
f"{processed}/{total}"
|
|
)
|
|
self.context.mark_stage("designing")
|
|
|
|
def _run_design_check(self) -> None:
|
|
"""Run Prompt 3 and store the complete implementation plan."""
|
|
|
|
self._publish_status("Mapping: designing reusable construct")
|
|
check = self.tighten_design_work_check_factory.create(
|
|
file_manager=self.context.upstream_file_manager,
|
|
original_prompt=self.context.goal,
|
|
selected_concern=self.context.selected_concern or {},
|
|
idiom_catalog=self.context.idiom_catalog,
|
|
visible_files=self._visible_files(),
|
|
previous_construct_locations=self.context.created_locations,
|
|
)
|
|
check()
|
|
self.context.design_plan = dict(getattr(check, "additional", {}) or {})
|
|
if not self.context.design_plan:
|
|
raise ValueError("Prompt 3 produced no bootstrap design plan.")
|
|
self.context.implementation_prompt = str(
|
|
self.context.design_plan.get("implementation_prompt") or ""
|
|
)
|
|
self.context.persist()
|
|
self._publish_status("Mapping: construct design selected")
|
|
self.context.mark_stage("editing")
|
|
|
|
def _bootstrap_editable_sources(self) -> list[ClassMeta]:
|
|
"""Return bootstrap existing and planned-new editable source metadata."""
|
|
|
|
file_manager = self.context.upstream_file_manager
|
|
repo_exec = file_manager.repo_exec
|
|
locations: list[tuple[str, str]] = []
|
|
for location in self.context.design_plan.get("editable_locations", []):
|
|
location_text = str(location).strip()
|
|
if location_text:
|
|
locations.append((location_text, self.context.implementation_prompt))
|
|
for item in self.context.design_plan.get("new_files", []):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
location = str(item.get("location") or "").strip()
|
|
reason = str(item.get("reason") or "").strip()
|
|
if location:
|
|
locations.append((location, reason))
|
|
|
|
editable_sources: list[ClassMeta] = []
|
|
seen: set[str] = set()
|
|
for location, reason in locations:
|
|
if location in seen:
|
|
continue
|
|
seen.add(location)
|
|
editable_sources.append(
|
|
file_manager.current_src(location)
|
|
or ClassMeta(
|
|
location=location,
|
|
summary=reason or f"Bootstrap planned source {location}.",
|
|
relevance=[reason] if reason else [],
|
|
relevant_symbols={},
|
|
content_type=ContentType.Source,
|
|
)
|
|
)
|
|
return editable_sources
|
|
|
|
def _bootstrap_readonly_sources(self) -> list[ClassMeta]:
|
|
"""Return bootstrap source context that is visible but not editable."""
|
|
|
|
file_manager = self.context.upstream_file_manager
|
|
editable_locations = {
|
|
source.location
|
|
for source in self._bootstrap_editable_sources()
|
|
}
|
|
readonly_locations = [
|
|
*self.context.visible_locations,
|
|
*[
|
|
str(location)
|
|
for location in self.context.design_plan.get("readonly_locations", [])
|
|
],
|
|
*self._source_dependency_locations(editable_locations),
|
|
]
|
|
readonly_sources: list[ClassMeta] = []
|
|
seen: set[str] = set()
|
|
for location in readonly_locations:
|
|
location = str(location).strip()
|
|
if not location or location in editable_locations or location in seen:
|
|
continue
|
|
current = file_manager.current_src(location)
|
|
if current is None:
|
|
continue
|
|
seen.add(location)
|
|
readonly_sources.append(current)
|
|
return readonly_sources
|
|
|
|
def _source_dependency_locations(self, locations: set[str]) -> list[str]:
|
|
"""Return graph upstream source dependency locations for the given files."""
|
|
|
|
graph = getattr(self._dependency_graph_maintainer(), "graph", None)
|
|
dependencies = getattr(graph, "source_dependencies", {}) or {}
|
|
result: list[str] = []
|
|
for location in sorted(locations):
|
|
result.extend(sorted(dependencies.get(location, set())))
|
|
return self._dedupe(result)
|
|
|
|
def _bootstrap_work_needed_by_location(
|
|
self,
|
|
editable_sources: list[ClassMeta],
|
|
) -> dict[str, str]:
|
|
"""Return bootstrap work guidance keyed by editable source location."""
|
|
|
|
work_needed: dict[str, str] = {}
|
|
new_file_reasons = {
|
|
str(item.get("location") or "").strip(): str(item.get("reason") or "").strip()
|
|
for item in self.context.design_plan.get("new_files", [])
|
|
if isinstance(item, dict)
|
|
}
|
|
call_site_work = {
|
|
str(item.get("location") or "").strip(): " ".join(
|
|
part
|
|
for part in (
|
|
str(item.get("required_change") or "").strip(),
|
|
str(item.get("usage") or "").strip(),
|
|
)
|
|
if part
|
|
)
|
|
for item in self.context.design_plan.get("call_sites", [])
|
|
if isinstance(item, dict)
|
|
}
|
|
for source in editable_sources:
|
|
work_needed[source.location] = (
|
|
new_file_reasons.get(source.location)
|
|
or call_site_work.get(source.location)
|
|
or self.context.implementation_prompt
|
|
)
|
|
return work_needed
|
|
|
|
def _bootstrap_blank_source_check(
|
|
self,
|
|
*,
|
|
editable_sources: list[ClassMeta],
|
|
readonly_sources: list[ClassMeta],
|
|
readonly_tests: list[ClassMeta],
|
|
work_needed_by_location: dict[str, str],
|
|
previous_summary: str,
|
|
request: AiderEditRequest,
|
|
) -> AlphaBlankSourceCorrectionDecision:
|
|
"""Return a blank-source correction check for bootstrap editable source files."""
|
|
|
|
file_manager = self.context.upstream_file_manager
|
|
repo_exec = file_manager.repo_exec
|
|
blank_sources: list[ClassMeta] = []
|
|
for source in editable_sources:
|
|
path = file_manager.source_path(source.location)
|
|
try:
|
|
content = (
|
|
repo_exec.read_text(path, encoding="utf-8")
|
|
if repo_exec.exists(path)
|
|
else None
|
|
)
|
|
except OSError:
|
|
content = None
|
|
if content is not None and content.strip():
|
|
continue
|
|
blank_sources.append(source)
|
|
return AlphaBlankSourceCorrectionDecision(
|
|
client=file_manager.client,
|
|
file_manager=file_manager,
|
|
request=request,
|
|
alpha_task=self.context.implementation_prompt or self.context.goal,
|
|
work_needed_by_location=work_needed_by_location,
|
|
editable_sources=editable_sources,
|
|
readonly_sources=readonly_sources,
|
|
readonly_tests=readonly_tests,
|
|
blank_sources=blank_sources,
|
|
previous_summary=previous_summary,
|
|
)
|
|
|
|
def _bootstrap_acceptance_check(
|
|
self,
|
|
*,
|
|
edit_request: AiderEditRequest,
|
|
editable_sources: list[ClassMeta],
|
|
readonly_sources: list[ClassMeta],
|
|
readonly_tests: list[ClassMeta],
|
|
previous_summary: str,
|
|
baseline: Mapping[str, str],
|
|
) -> BootstrapAcceptanceCheck:
|
|
"""Return a bootstrap acceptance check for the pre-stabilization source edit."""
|
|
|
|
return BootstrapAcceptanceCheck(
|
|
client=self.context.upstream_file_manager.client,
|
|
file_manager=self.context.upstream_file_manager,
|
|
edit_request=edit_request,
|
|
editable_sources=editable_sources,
|
|
readonly_sources=readonly_sources,
|
|
readonly_tests=readonly_tests,
|
|
previous_summary=previous_summary,
|
|
baseline=baseline,
|
|
)
|
|
|
|
def _bootstrap_correction_guide(self, failed_checks: tuple[object, ...]) -> str:
|
|
"""Return retry guidance from failed bootstrap Alpha acceptance gates."""
|
|
|
|
failure_lines = [
|
|
"BOOTSTRAP EDIT ACCEPTANCE FAILED",
|
|
"",
|
|
"The previous bootstrap source edit was rejected, but its changes remain.",
|
|
"The current editable source files include the previous attempt's changes.",
|
|
"",
|
|
"Perform the next bootstrap source edit from the current mutated state.",
|
|
"Continue from the current mutated state. Revise, replace, remove, or extend",
|
|
"the existing changes as needed. Do not assume the previous attempt was",
|
|
"rolled back.",
|
|
"Do not leave planned bootstrap files blank.",
|
|
"Do not edit tests.",
|
|
"Do not create additional files outside the planned source scope.",
|
|
]
|
|
for index, failed_check in enumerate(failed_checks, start=1):
|
|
failure_lines.extend(
|
|
[
|
|
"",
|
|
f"Failure {index}:",
|
|
str(getattr(failed_check, "reason", "")).strip(),
|
|
"",
|
|
"Required correction:",
|
|
str(getattr(failed_check, "additional", {}).get("instruction", "")).strip(),
|
|
]
|
|
)
|
|
return "\n".join(failure_lines)
|
|
|
|
def _run_editor(self) -> None:
|
|
"""Run one coordinated editor request and validate the mechanical result."""
|
|
|
|
self._publish_status("Mapping: generating and wiring construct")
|
|
file_manager = self.context.upstream_file_manager
|
|
self._preflight_needed_source_files()
|
|
editable_sources = self._bootstrap_editable_sources()
|
|
self._preflight_planned_blank_sources()
|
|
self._register_preflight_participant_relevance()
|
|
editable_sources = self._bootstrap_editable_sources()
|
|
readonly_sources = self._bootstrap_readonly_sources()
|
|
readonly_tests: list[ClassMeta] = []
|
|
work_needed_by_location = self._bootstrap_work_needed_by_location(
|
|
editable_sources,
|
|
)
|
|
extra_guidance: list[str] | None = None
|
|
while self.context.iteration < self.context.max_iterations:
|
|
request = self._edit_request(extra_guidance=extra_guidance)
|
|
# Bootstrap keeps rejected edits, so each attempt owns its own PRE text without rollback mementos.
|
|
baseline = dict(map(lambda source: (
|
|
source.location, file_manager.repo_exec.read_bytes(source.resolve_path(file_manager)).decode("utf-8"),
|
|
), editable_sources))
|
|
result = self.aider_worker.edit(
|
|
request,
|
|
logger=self.logger,
|
|
)
|
|
changed_paths, summary = result.changed_paths, result.summary
|
|
request.editor_records = list(result.records)
|
|
self.context.editor_records = list(result.records)
|
|
leave_it_judged = is_code_edit_leave_summary(summary)
|
|
summary = consume_code_edit_leave_summary(summary)
|
|
self._refresh_changed_sources(changed_paths, summary)
|
|
editable_sources = self._bootstrap_editable_sources()
|
|
readonly_sources = self._bootstrap_readonly_sources()
|
|
work_needed_by_location = self._bootstrap_work_needed_by_location(
|
|
editable_sources,
|
|
)
|
|
self.context.editor_summaries.append(summary)
|
|
self.context.persist()
|
|
failed_checks: tuple[object, ...] = ()
|
|
require_changes = not leave_it_judged
|
|
if require_changes:
|
|
blank_check = self._bootstrap_blank_source_check(
|
|
editable_sources=editable_sources,
|
|
readonly_sources=readonly_sources,
|
|
readonly_tests=readonly_tests,
|
|
work_needed_by_location=work_needed_by_location,
|
|
previous_summary=summary,
|
|
request=request,
|
|
)
|
|
check_results = [(blank_check, blank_check())]
|
|
if check_results[0][1]:
|
|
acceptance_check = self._bootstrap_acceptance_check(
|
|
edit_request=request,
|
|
editable_sources=editable_sources,
|
|
readonly_sources=readonly_sources,
|
|
readonly_tests=readonly_tests,
|
|
previous_summary=summary,
|
|
baseline=baseline,
|
|
)
|
|
check_results.append((acceptance_check, acceptance_check()))
|
|
failed_checks = tuple(
|
|
check
|
|
for check, passed in check_results
|
|
if not passed
|
|
)
|
|
|
|
if not failed_checks:
|
|
self._record_accepted_editor_result(
|
|
changed_paths,
|
|
require_changes=require_changes,
|
|
)
|
|
self._publish_status(
|
|
"Mapping: construct already satisfied"
|
|
if not require_changes
|
|
else "Mapping: construct edit accepted"
|
|
)
|
|
self.context.mark_completed()
|
|
return
|
|
|
|
extra_guidance = [self._bootstrap_correction_guide(failed_checks)]
|
|
self.context.iteration += 1
|
|
self.context.persist()
|
|
|
|
self.context.mark_blocked(
|
|
"Bootstrap editor retry budget exhausted before an accepted edit."
|
|
)
|
|
|
|
def _refresh_changed_sources(
|
|
self,
|
|
changed_paths: list[Path],
|
|
editor_summary: str,
|
|
) -> None:
|
|
"""Refresh each editor-changed source before Bootstrap continues."""
|
|
|
|
file_manager = self.context.upstream_file_manager
|
|
cdc_prompt = editor_summary or self.context.implementation_prompt
|
|
for path in changed_paths:
|
|
try:
|
|
location = file_manager.source_location(path)
|
|
except ValueError:
|
|
continue
|
|
file_manager.refresh_src_relevance(
|
|
location,
|
|
relevance_prompt=self.context.goal,
|
|
cdc_prompt=cdc_prompt,
|
|
tailing_logs=editor_summary,
|
|
)
|
|
if location not in self.context.involved_locations:
|
|
self.context.involved_locations.append(location)
|
|
self.context.persist()
|
|
|
|
def _preflight_planned_blank_sources(self) -> None:
|
|
"""Create and register planned bootstrap source files before edit requests."""
|
|
|
|
file_manager = self.context.upstream_file_manager
|
|
repo_exec = file_manager.repo_exec
|
|
requested: dict[Path, str] = {}
|
|
created_paths: set[Path] = set()
|
|
for location, reason in self._planned_new_file_reasons().items():
|
|
path = file_manager.source_path(location)
|
|
current = file_manager.current_src(location)
|
|
if repo_exec.exists(path):
|
|
if current is not None:
|
|
continue
|
|
content = repo_exec.read_text(path, encoding="utf-8")
|
|
if content.strip():
|
|
requested[path] = reason
|
|
continue
|
|
else:
|
|
repo_exec.mkdir(path.parent, parents=True, exist_ok=True)
|
|
repo_exec.write_text(path, "", encoding="utf-8")
|
|
created_paths.add(path)
|
|
requested[path] = reason
|
|
if requested and not self._register_preflight_new_files(requested, created_paths):
|
|
raise ValueError("Bootstrap preflight new-file registration failed.")
|
|
|
|
def _register_preflight_participant_relevance(self) -> None:
|
|
"""Register temporary relevance handles for all bootstrap participant sources."""
|
|
|
|
file_manager = self.context.upstream_file_manager
|
|
editable_locations = {
|
|
source.location
|
|
for source in self._bootstrap_editable_sources()
|
|
}
|
|
readonly_locations = {
|
|
source.location
|
|
for source in self._bootstrap_readonly_sources()
|
|
}
|
|
for entry in self.context.idiom_catalog:
|
|
evidence = entry.get("evidence") if isinstance(entry, dict) else None
|
|
if not isinstance(evidence, list):
|
|
continue
|
|
for item in evidence:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
location = str(item.get("location") or "").strip()
|
|
if file_manager.current_src(location) is not None:
|
|
readonly_locations.add(location)
|
|
participant_locations = self._dedupe(
|
|
[
|
|
*sorted(editable_locations),
|
|
*sorted(readonly_locations),
|
|
]
|
|
)
|
|
for location in participant_locations:
|
|
current = file_manager.current_src(location)
|
|
if current is None:
|
|
continue
|
|
role = "editable" if location in editable_locations else "readonly"
|
|
handle = self._preflight_participant_relevance(location, role)
|
|
if not handle:
|
|
continue
|
|
if handle in current.relevance:
|
|
continue
|
|
updated = ClassMeta(
|
|
location=current.location,
|
|
summary=current.summary,
|
|
relevance=[*current.relevance, handle],
|
|
relevant_symbols=dict(current.relevant_symbols),
|
|
cdc=[
|
|
*current.cdc,
|
|
(
|
|
datetime.now(),
|
|
file_manager.actor,
|
|
"Bootstrap preflight registered temporary participant relevance.",
|
|
),
|
|
],
|
|
content_type=ContentType.Source,
|
|
)
|
|
file_manager._upsert(updated, file_manager.src_classes, file_manager.src_index)
|
|
file_manager.src_dirty[location] = True
|
|
self.context.persist()
|
|
|
|
def _preflight_participant_relevance(self, location: str, role: str) -> str:
|
|
"""Return one Step 3-derived temporary participant relevance handle."""
|
|
|
|
parts = self._step3_relevance_parts(location)
|
|
if not parts and role == "readonly":
|
|
parts = self._readonly_relevance_parts(location)
|
|
if not parts:
|
|
return ""
|
|
return " ".join(
|
|
[
|
|
f"Bootstrap preflight {role} relevance from Step 3 design.",
|
|
f"Location {location}.",
|
|
*parts,
|
|
]
|
|
).strip()
|
|
|
|
def _step3_relevance_parts(self, location: str) -> list[str]:
|
|
"""Return Step 3 design relevance text for one source location."""
|
|
|
|
parts: list[str] = []
|
|
for item in self.context.design_plan.get("new_files", []):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
new_location = str(item.get("location") or "").strip()
|
|
if new_location != location:
|
|
continue
|
|
reason = str(item.get("reason") or "").strip()
|
|
if reason:
|
|
parts.append(f"New file reason: {reason}.")
|
|
construct_parts = []
|
|
for construct in item.get("constructs", []):
|
|
if not isinstance(construct, dict):
|
|
continue
|
|
name = str(construct.get("name") or "").strip()
|
|
responsibility = str(construct.get("responsibility") or "").strip()
|
|
if name and responsibility:
|
|
construct_parts.append(f"{name}: {responsibility}")
|
|
elif name:
|
|
construct_parts.append(name)
|
|
elif responsibility:
|
|
construct_parts.append(responsibility)
|
|
if construct_parts:
|
|
parts.append(f"Planned constructs: {'; '.join(construct_parts)}.")
|
|
parts.append(f"Original request: {self.context.goal}")
|
|
for item in self.context.design_plan.get("call_sites", []):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
if str(item.get("location") or "").strip() != location:
|
|
continue
|
|
parts.append(self._site_relevance_text("Call site", item))
|
|
for item in self.context.design_plan.get("construction_sites", []):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
if str(item.get("location") or "").strip() != location:
|
|
continue
|
|
parts.append(self._site_relevance_text("Construction site", item))
|
|
for item in self.context.design_plan.get("expected_relationships", []):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
consumer = str(item.get("consumer") or "").strip()
|
|
dependency = str(item.get("dependency") or "").strip()
|
|
if location not in {consumer, dependency}:
|
|
continue
|
|
reason = str(item.get("reason") or "").strip()
|
|
parts.append(
|
|
" ".join(
|
|
part
|
|
for part in (
|
|
f"Design relationship: {consumer} depends on {dependency}.",
|
|
f"Reason: {reason}." if reason else "",
|
|
)
|
|
if part
|
|
)
|
|
)
|
|
return [part for part in parts if part]
|
|
|
|
def _site_relevance_text(self, label: str, item: dict[str, object]) -> str:
|
|
"""Return relevance text for one Step 3 site entry."""
|
|
|
|
symbol = str(item.get("symbol_or_beat") or "").strip()
|
|
required_change = str(item.get("required_change") or "").strip()
|
|
usage = str(item.get("usage") or "").strip()
|
|
return " ".join(
|
|
part
|
|
for part in (
|
|
f"{label}: {symbol}." if symbol else f"{label}.",
|
|
f"Required change: {required_change}." if required_change else "",
|
|
f"Usage: {usage}." if usage else "",
|
|
)
|
|
if part
|
|
)
|
|
|
|
def _readonly_relevance_parts(self, location: str) -> list[str]:
|
|
"""Return readonly context relevance from Step 2 idiom evidence."""
|
|
|
|
parts: list[str] = []
|
|
for entry in self.context.idiom_catalog:
|
|
evidence = entry.get("evidence")
|
|
if not isinstance(evidence, list):
|
|
continue
|
|
evidence_locations = {
|
|
str(item.get("location") or "").strip()
|
|
for item in evidence
|
|
if isinstance(item, dict)
|
|
}
|
|
if location not in evidence_locations:
|
|
continue
|
|
observation = str(entry.get("observation") or "").strip()
|
|
implication = str(entry.get("design_implication") or "").strip()
|
|
parts.append(
|
|
" ".join(
|
|
part
|
|
for part in (
|
|
f"Step 2 idiom evidence: {observation}." if observation else "",
|
|
f"Design implication: {implication}." if implication else "",
|
|
)
|
|
if part
|
|
)
|
|
)
|
|
return [part for part in parts if part]
|
|
|
|
def _preflight_needed_source_files(self) -> None:
|
|
"""Ask whether the bootstrap edit needs additional source files before editing."""
|
|
|
|
existing_editable_sources = self._bootstrap_existing_editable_sources()
|
|
readonly_sources = self._bootstrap_readonly_sources_for_locations(
|
|
{
|
|
source.location
|
|
for source in existing_editable_sources
|
|
}
|
|
)
|
|
if not existing_editable_sources and not readonly_sources:
|
|
return
|
|
file_manager = self.context.upstream_file_manager
|
|
work_needed_by_location = self._bootstrap_work_needed_by_location(
|
|
[*existing_editable_sources, *readonly_sources],
|
|
)
|
|
preflight_request = AiderEditRequest(
|
|
authoritative_prompt=self.context.implementation_prompt,
|
|
editable_files=[
|
|
file_manager.source_path(source.location)
|
|
for source in existing_editable_sources
|
|
],
|
|
readonly_files=[
|
|
file_manager.source_path(source.location)
|
|
for source in readonly_sources
|
|
],
|
|
)
|
|
decision = BootstrapSourcePreflightDecision(
|
|
client=file_manager.client,
|
|
file_manager=file_manager,
|
|
request=preflight_request,
|
|
original_request=self.context.goal,
|
|
implementation_prompt=self.context.implementation_prompt,
|
|
design_plan=self.context.design_plan,
|
|
editable_sources=existing_editable_sources,
|
|
readonly_sources=readonly_sources,
|
|
work_needed_by_location=work_needed_by_location,
|
|
logger=self.logger.getChild("source-preflight"),
|
|
)
|
|
decision()
|
|
requested = decision.additional["requested_files"]
|
|
|
|
requested_files = self._normalize_preflight_new_files(requested)
|
|
if requested_files:
|
|
self._merge_preflight_new_files(requested_files)
|
|
|
|
def _bootstrap_existing_editable_sources(self) -> list[ClassMeta]:
|
|
"""Return existing editable source files from the design plan."""
|
|
|
|
file_manager = self.context.upstream_file_manager
|
|
sources: list[ClassMeta] = []
|
|
seen: set[str] = set()
|
|
for location in self.context.design_plan.get("editable_locations", []):
|
|
location = str(location).strip()
|
|
if not location or location in seen:
|
|
continue
|
|
source = file_manager.current_src(location)
|
|
if source is None:
|
|
continue
|
|
seen.add(location)
|
|
sources.append(source)
|
|
return sources
|
|
|
|
def _bootstrap_readonly_sources_for_locations(
|
|
self,
|
|
editable_locations: set[str],
|
|
) -> list[ClassMeta]:
|
|
"""Return deterministic relational readonly source context for preflight."""
|
|
|
|
file_manager = self.context.upstream_file_manager
|
|
locations = self._dedupe(
|
|
[
|
|
*self.context.visible_locations,
|
|
*[
|
|
str(location)
|
|
for location in self.context.design_plan.get("readonly_locations", [])
|
|
],
|
|
*self._source_dependency_locations(editable_locations),
|
|
]
|
|
)
|
|
sources: list[ClassMeta] = []
|
|
seen: set[str] = set()
|
|
for location in locations:
|
|
if not location or location in editable_locations or location in seen:
|
|
continue
|
|
source = file_manager.current_src(location)
|
|
if source is None:
|
|
continue
|
|
seen.add(location)
|
|
sources.append(source)
|
|
return sources
|
|
|
|
def _normalize_preflight_new_files(
|
|
self,
|
|
requested: list[object],
|
|
) -> list[dict[str, str]]:
|
|
"""Return valid additional source files from preflight output."""
|
|
|
|
file_manager = self.context.upstream_file_manager
|
|
existing = {
|
|
source.location
|
|
for source in file_manager.src_classes
|
|
}
|
|
planned = set(self._planned_new_file_reasons())
|
|
editable = {
|
|
str(location).strip()
|
|
for location in self.context.design_plan.get("editable_locations", [])
|
|
}
|
|
readonly = {
|
|
source.location
|
|
for source in self._bootstrap_readonly_sources_for_locations(editable)
|
|
}
|
|
files: list[dict[str, str]] = []
|
|
seen: set[str] = set()
|
|
for item in requested:
|
|
if isinstance(item, str):
|
|
location = item.strip()
|
|
reason = "Required by bootstrap source preflight."
|
|
elif isinstance(item, dict):
|
|
location = str(item.get("location") or "").strip()
|
|
reason = str(item.get("reason") or "").strip()
|
|
else:
|
|
continue
|
|
if not location or not reason or location in seen:
|
|
continue
|
|
if location in existing or location in planned or location in editable or location in readonly:
|
|
continue
|
|
try:
|
|
normalized = file_manager.source_location(location)
|
|
file_manager.source_path(normalized)
|
|
except (OSError, ValueError):
|
|
continue
|
|
seen.add(normalized)
|
|
files.append({"location": normalized, "reason": reason})
|
|
return files
|
|
|
|
def _merge_preflight_new_files(self, requested_files: list[dict[str, str]]) -> None:
|
|
"""Merge preflight-requested files into the current bootstrap design plan."""
|
|
|
|
if not requested_files:
|
|
return
|
|
planned = [
|
|
item
|
|
for item in self.context.design_plan.get("new_files", [])
|
|
if isinstance(item, dict)
|
|
]
|
|
planned_locations = {
|
|
str(item.get("location") or "").strip()
|
|
for item in planned
|
|
}
|
|
added: list[dict[str, str]] = []
|
|
for item in requested_files:
|
|
location = str(item.get("location") or "").strip()
|
|
reason = str(item.get("reason") or "").strip()
|
|
if not location or not reason or location in planned_locations:
|
|
continue
|
|
planned.append({"location": location, "reason": reason})
|
|
planned_locations.add(location)
|
|
added.append({"location": location, "reason": reason})
|
|
if not added:
|
|
return
|
|
self.context.design_plan["new_files"] = planned
|
|
extra_instruction = " ".join(
|
|
f"Create and populate {item['location']}: {item['reason']}"
|
|
for item in added
|
|
)
|
|
self.context.implementation_prompt = " ".join(
|
|
part
|
|
for part in (self.context.implementation_prompt, extra_instruction)
|
|
if part
|
|
)
|
|
self.context.design_plan["implementation_prompt"] = self.context.implementation_prompt
|
|
self.context.persist()
|
|
|
|
def _refresh_pending_locations(self) -> list[str]:
|
|
"""Return bootstrap participant locations still needing metadata refresh."""
|
|
|
|
return [
|
|
location
|
|
for location in self._dedupe(
|
|
[
|
|
*self.context.created_locations,
|
|
*self.context.involved_locations,
|
|
]
|
|
)
|
|
if location not in self.context.refreshed_locations
|
|
]
|
|
|
|
def _deterministic_pairs(self, locations: list[str]) -> list[tuple[str, ...]]:
|
|
"""Return stable random-looking sample pairs with at most two locations each."""
|
|
|
|
shuffled = list(dict.fromkeys(locations))
|
|
seed_material = "\n".join(
|
|
[
|
|
self.context.goal,
|
|
str(self.context.iteration),
|
|
*sorted(shuffled),
|
|
]
|
|
)
|
|
seed = int(hashlib.sha256(seed_material.encode("utf-8")).hexdigest(), 16)
|
|
rng = random.Random(seed)
|
|
rng.shuffle(shuffled)
|
|
return [
|
|
tuple(shuffled[index:index + 2])
|
|
for index in range(0, len(shuffled), 2)
|
|
]
|
|
|
|
def _merge_idiom_entries(self, entries: list[dict[str, object]]) -> None:
|
|
"""Merge Prompt 2 entries into the stationary deduplicated catalog."""
|
|
|
|
existing = {
|
|
idiom_entry_key(entry)
|
|
for entry in self.context.idiom_catalog
|
|
}
|
|
for entry in entries:
|
|
key = idiom_entry_key(entry)
|
|
if key in existing:
|
|
continue
|
|
existing.add(key)
|
|
self.context.idiom_catalog.append(entry)
|
|
|
|
def _read_source(self, location: str) -> str:
|
|
"""Return the current source text for a repo-relative location."""
|
|
|
|
file_manager = self.context.upstream_file_manager
|
|
return file_manager.repo_exec.read_text(
|
|
file_manager.source_path(location), encoding="utf-8"
|
|
)
|
|
|
|
def _edit_request(self, extra_guidance: list[str] | None = None) -> AiderEditRequest:
|
|
"""Build one coordinated edit request from the stored design plan."""
|
|
|
|
plan = self.context.design_plan
|
|
file_manager = self.context.upstream_file_manager
|
|
editable_locations = [
|
|
str(location)
|
|
for location in plan.get("editable_locations", [])
|
|
]
|
|
editable_locations = self._dedupe(
|
|
[
|
|
*editable_locations,
|
|
*[
|
|
str(item.get("location") or "").strip()
|
|
for item in plan.get("new_files", [])
|
|
if isinstance(item, dict)
|
|
],
|
|
]
|
|
)
|
|
editable_location_set = set(editable_locations)
|
|
readonly_locations = [
|
|
location
|
|
for location in self._dedupe(
|
|
[
|
|
*[
|
|
str(location)
|
|
for location in plan.get("readonly_locations", [])
|
|
],
|
|
*self._source_dependency_locations(editable_location_set),
|
|
]
|
|
)
|
|
if location not in editable_location_set
|
|
]
|
|
prompt_parts = [
|
|
self.context.implementation_prompt,
|
|
*list(extra_guidance or []),
|
|
]
|
|
return AiderEditRequest(
|
|
authoritative_prompt="\n\n".join(
|
|
part
|
|
for part in prompt_parts
|
|
if part.strip()
|
|
),
|
|
editable_files=[
|
|
file_manager.source_path(location)
|
|
for location in editable_locations
|
|
],
|
|
readonly_files=[
|
|
file_manager.source_path(location)
|
|
for location in readonly_locations
|
|
],
|
|
allowed_additional_roots=[file_manager.src_root],
|
|
backlog_guidance=list(extra_guidance or []),
|
|
editor_records=list(self.context.editor_records),
|
|
context=BootstrapLoop,
|
|
)
|
|
|
|
def _register_preflight_new_files(
|
|
self,
|
|
files: dict[Path, str],
|
|
created_paths: set[Path],
|
|
) -> bool:
|
|
"""Register bootstrap planned source metadata before editing."""
|
|
|
|
file_manager = self.context.upstream_file_manager
|
|
planned_reasons = self._planned_new_file_reasons()
|
|
for path, reason in files.items():
|
|
try:
|
|
location = file_manager.source_location(path)
|
|
except (OSError, ValueError):
|
|
return False
|
|
if location not in planned_reasons:
|
|
return False
|
|
provisional_reason = planned_reasons.get(location) or str(reason)
|
|
current = file_manager.current_src(location)
|
|
if current is not None:
|
|
continue
|
|
cdc_message = (
|
|
"Bootstrap preflight created a planned blank source."
|
|
if path in created_paths
|
|
else "Bootstrap preflight registered planned source metadata."
|
|
)
|
|
meta = ClassMeta(
|
|
location=location,
|
|
summary=provisional_reason,
|
|
relevance=[provisional_reason],
|
|
relevant_symbols={},
|
|
cdc=[
|
|
(
|
|
datetime.now(),
|
|
file_manager.actor,
|
|
cdc_message,
|
|
)
|
|
],
|
|
content_type=ContentType.Source,
|
|
)
|
|
file_manager._upsert(meta, file_manager.src_classes, file_manager.src_index)
|
|
file_manager.src_dirty[location] = True
|
|
self.context.persist()
|
|
return True
|
|
|
|
def _planned_new_file_reasons(self) -> dict[str, str]:
|
|
"""Return planned new source locations and their provisional reasons."""
|
|
|
|
reasons: dict[str, str] = {}
|
|
for item in self.context.design_plan.get("new_files", []):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
location = str(item.get("location") or "").strip()
|
|
reason = str(item.get("reason") or "").strip()
|
|
if location:
|
|
reasons[location] = reason
|
|
return reasons
|
|
|
|
def _record_accepted_editor_result(
|
|
self,
|
|
changed_paths: list[Path],
|
|
*,
|
|
require_changes: bool,
|
|
) -> None:
|
|
"""Validate and record accepted bootstrap participant locations."""
|
|
|
|
accepted_paths = list(dict.fromkeys(changed_paths))
|
|
if require_changes and not accepted_paths:
|
|
raise ValueError("editor changed no files")
|
|
file_manager = self.context.upstream_file_manager
|
|
planned_new = set(self._planned_new_file_reasons())
|
|
planned_editable = {
|
|
str(location)
|
|
for location in self.context.design_plan.get("editable_locations", [])
|
|
}
|
|
allowed = planned_new | planned_editable
|
|
for path in accepted_paths:
|
|
location = file_manager.source_location(path)
|
|
if location not in allowed:
|
|
raise ValueError(f"changed path {location} was outside planned source scope")
|
|
|
|
for location in planned_new:
|
|
path = file_manager.source_path(location)
|
|
if not file_manager.repo_exec.exists(path):
|
|
raise ValueError(f"planned new file {location} was not created")
|
|
if not file_manager.repo_exec.read_text(path, encoding="utf-8").strip():
|
|
raise ValueError(f"planned new file {location} remained blank")
|
|
|
|
created = sorted(planned_new)
|
|
self.context.created_locations = self._dedupe(
|
|
[*self.context.created_locations, *created]
|
|
)
|
|
self.context.involved_locations = self._dedupe(
|
|
[
|
|
*getattr(self.context, "involved_locations", []),
|
|
*sorted(planned_new),
|
|
*sorted(planned_editable),
|
|
]
|
|
)
|
|
if require_changes:
|
|
self.context.refreshed_locations = []
|
|
self.context.persist()
|
|
|
|
def _dependency_graph_maintainer(self) -> CodeDependencyGraphMaintainer:
|
|
"""Return the mutable dependency graph maintainer for targeted refresh."""
|
|
|
|
file_manager = self.context.upstream_file_manager
|
|
if file_manager.dependency_graph_maintainer is None:
|
|
file_manager.dependency_graph_maintainer = self.language_strategy_set[
|
|
CodeDependencyGraphMaintainer
|
|
]
|
|
return file_manager.dependency_graph_maintainer
|
|
|
|
def _dedupe(self, locations: list[str]) -> list[str]:
|
|
"""Return locations in first-seen order without duplicates."""
|
|
|
|
return list(dict.fromkeys(location for location in locations if location))
|