553 lines
40 KiB
Python
553 lines
40 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Replace task instructions while splicing exact source sections into the localized prompt."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, replace
|
|
from itertools import chain, groupby
|
|
import json
|
|
import logging
|
|
from typing import ClassVar, cast
|
|
|
|
from band_aider.ai_client import completion_client, json_response_decoder
|
|
from band_aider.ai_client.completion_client import CompletionClient
|
|
from band_aider.context import validation_utils
|
|
from band_aider.context.validation_utils import (
|
|
MechanicalValidationError, SemanticValidationError, ValidationLocation, ValidationResult,
|
|
)
|
|
from band_aider.judgement.composite_check import CompositeCheck
|
|
from band_aider.judgement.edit_request import AiderEditRequest
|
|
from band_aider.judgement.file_manager import FileManager
|
|
from band_aider.judgement.mechanical_check import MechanicalCheck
|
|
from band_aider.judgement.prompt_formatter import PromptFormatterBuilder
|
|
from band_aider.judgement.stabilization_decision import StabilizationDecision
|
|
|
|
|
|
_AUTHORITY_FIELDS = ("relevant cot", "relevant backlog guidance", "relevant future guidance")
|
|
_CITATION_REFERENCE = {"relevant code": [[1, 2]], "relevant stack": [[3, 4]],
|
|
"relevant cot": [5], "relevant backlog guidance": [6], "relevant future guidance": [7]}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LocalSource:
|
|
"""Identify original input lines by inclusive, one-based coordinates."""
|
|
|
|
start_line: int
|
|
end_line: int
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LocalizedPrompt:
|
|
"""Keep one actual localized prompt with its original code and diagnostic ranges."""
|
|
|
|
evidence: str
|
|
prompt: str
|
|
code: tuple[LocalSource, ...] = ()
|
|
stack: tuple[LocalSource, ...] = ()
|
|
|
|
def render(self) -> str:
|
|
"""Insert exact selected code without reinterpreting markers inside that code."""
|
|
lines = self.evidence.splitlines(keepends=True)
|
|
selected = sorted(set(chain.from_iterable(map(
|
|
lambda span: range(span.start_line, span.end_line + 1), self.code))))
|
|
# Union removes duplicate/overlapping selections; consecutive buckets preserve exact source blocks.
|
|
blocks = groupby(enumerate(selected), key=lambda pair: pair[1] - pair[0])
|
|
source = "\n\n".join(map(lambda block: "".join(map(lambda pair: lines[pair[1] - 1], block[1])), blocks))
|
|
# A supplied placeholder controls placement; otherwise code follows this item's instructions.
|
|
if "{{SOURCE}}" in self.prompt:
|
|
return self.prompt.replace("{{SOURCE}}", source, 1)
|
|
return self.prompt + ("\n\n" + source if source else "")
|
|
|
|
|
|
class LocalizationEvidenceCheck(MechanicalCheck):
|
|
"""Validate selected original ranges and diagnostic quotations before source insertion."""
|
|
|
|
def __init__(self, original: str, candidates: tuple[LocalizedPrompt, ...]) -> None:
|
|
"""Bind the original input and the localized items whose evidence is being checked."""
|
|
super().__init__(changes=[])
|
|
self._original, self._candidates = original, candidates
|
|
self.validation = ValidationResult()
|
|
|
|
def __call__(self, *args, **kwargs) -> bool:
|
|
"""Require each claimed diagnostic to match the original input at its declared range."""
|
|
lines = self._original.splitlines(keepends=True)
|
|
errors = []
|
|
# Validate each item independently; relevant evidence may support more than one instruction.
|
|
for index, candidate in enumerate(self._candidates):
|
|
location = ValidationLocation([str(index)])
|
|
ranges = candidate.code + candidate.stack
|
|
valid = all(type(value.start_line) is int and type(value.end_line) is int
|
|
and 1 <= value.start_line <= value.end_line <= len(lines) for value in ranges)
|
|
# No external coordinate is rendered until it is known to address original input.
|
|
if not valid or candidate.evidence != self._original:
|
|
errors.append(MechanicalValidationError(location,
|
|
"relevant code and relevant stack must select existing original-input lines: lists of [start, end] ranges, 1-based inclusive."))
|
|
continue
|
|
# The prompt carries actual instructions, with at most one optional source insertion point.
|
|
if candidate.prompt.count("{{SOURCE}}") > 1 or ("{{SOURCE}}" in candidate.prompt and not candidate.code):
|
|
errors.append(MechanicalValidationError(location,
|
|
"Use at most one {{SOURCE}} with relevant code; otherwise omit the marker and code follows your prompt."))
|
|
# Diagnostic provenance is checked against the original prompt, not separate request fields.
|
|
for stack in candidate.stack:
|
|
diagnostic = "".join(lines[stack.start_line - 1:stack.end_line]).removesuffix("\n").removesuffix("\r")
|
|
# Source insertion must not change or remove the original diagnostic quotation.
|
|
if not diagnostic.strip() or diagnostic not in candidate.prompt or diagnostic not in candidate.render():
|
|
errors.append(MechanicalValidationError(location,
|
|
"The quoted failure diagnostic must exist verbatim in the original prompt at relevant stack; "
|
|
"copy that exact diagnostic into this item's prompt without changing its words or indentation."))
|
|
self.validation = ValidationResult(errors=tuple(errors))
|
|
return self.validation.valid
|
|
|
|
|
|
def correction_prompt(prompt: str, response: object, validation: ValidationResult | None,
|
|
formatter: PromptFormatterBuilder) -> str:
|
|
"""Keep only the immediately rejected localization response and its validation feedback."""
|
|
# The initial attempt receives the same complete contract without invented correction history.
|
|
if validation is None:
|
|
return prompt
|
|
section = formatter.generate
|
|
return "\n\n".join((prompt, section("LOCALIZATION VALIDATION ERROR")(validation.stringify()),
|
|
section("REJECTED PREVIOUS RESPONSE")(response if isinstance(response, str) else json.dumps(response)),
|
|
"Correct the response using the unchanged input and original response contract."))
|
|
|
|
|
|
class LocalizationRequirements(StabilizationDecision):
|
|
"""Select applicable input lines before seeing any localized candidate; call requirements()."""
|
|
|
|
_REFERENCE: list[object] = [_CITATION_REFERENCE]
|
|
|
|
def __init__(self, *, client: CompletionClient, file_manager: FileManager, request: AiderEditRequest,
|
|
prompt: str, original: str, evidence: str, logger: logging.Logger) -> None:
|
|
"""Bind the independent applicability question and its original coordinate boundaries."""
|
|
super().__init__(client=client, file_manager=file_manager,
|
|
request=replace(request, future_guidance=()), prompt=prompt, logger=logger)
|
|
self._original = original
|
|
self._evidence = evidence
|
|
|
|
@property
|
|
def valid_response_references(self) -> list[object]:
|
|
"""Require explicit selections for every authority, source and diagnostic category."""
|
|
return self._REFERENCE
|
|
|
|
def _decide(self, response: object | None = None,
|
|
validation: ValidationResult | None = None) -> tuple[object | None, ValidationResult]:
|
|
"""Admit original-input selections without access to any generated instruction or review."""
|
|
prompt = correction_prompt(self.prompt, response, validation, self._prompt_formatter_builder)
|
|
with completion_client.scoped_completion(type(self).__name__, self.escalation.files):
|
|
raw = self._client.complete(prompt, logger=self.logger)
|
|
data = json.loads(json_response_decoder.strip_json_fence(raw))
|
|
checked = validation_utils.validate_shape(data, self.valid_response_references, location=ValidationLocation([]))
|
|
# Malformed selections are corrected here, before the producer has any candidate to anchor on.
|
|
if not checked.valid:
|
|
return data, checked
|
|
result = cast(dict[str, object], data)
|
|
code, stack = map(lambda key: cast(list[list[int]], result[key]), ("relevant code", "relevant stack"))
|
|
line_count, evidence_count = len(self._original.splitlines()), len(self._evidence.splitlines())
|
|
authority = dict(map(lambda key: (key, frozenset(cast(list[int], result[key]))), _AUTHORITY_FIELDS))
|
|
# Code and diagnostics address original evidence; additional numbered authority is not source.
|
|
if any(len(span) != 2 or not 1 <= span[0] <= span[1] <= evidence_count for span in chain(code, stack)) or any(
|
|
not 1 <= line <= line_count for line in chain.from_iterable(authority.values())):
|
|
return data, ValidationResult(errors=(MechanicalValidationError(ValidationLocation([]),
|
|
"Select existing input lines: authority uses individual line numbers; code/stack use lists of inclusive "
|
|
"[start, end] ranges within original evidence, not appended authority."),))
|
|
evidence_lines = self._evidence.splitlines(keepends=True)
|
|
# A blank diagnostic selection cannot be fulfilled by any exact quotation.
|
|
if any(not "".join(evidence_lines[span[0] - 1:span[1]]).strip() for span in stack):
|
|
return data, ValidationResult(errors=(MechanicalValidationError(ValidationLocation(["relevant stack"]),
|
|
"Select complete nonblank diagnostic blocks, or [] when none applies."),))
|
|
required = dict(authority)
|
|
# Fixed applicability is expressed as line identities, so duplicate selections cannot mask omissions.
|
|
for key, spans in (("relevant code", code), ("relevant stack", stack)):
|
|
required[key] = frozenset(chain.from_iterable(map(lambda span: range(span[0], span[1] + 1), spans)))
|
|
self.additional = {"required": required, "stacks": tuple(map(lambda span: LocalSource(*span), stack))}
|
|
self.changes.append((self, True))
|
|
return data, ValidationResult()
|
|
|
|
|
|
class LocalizationReview(StabilizationDecision):
|
|
"""Review one instruction without recursively localizing its reviewer: review()."""
|
|
|
|
_REFERENCE: list[object] = [{"outcome": "pass", "reason": "evidence-based explanation", "instruction": ""}]
|
|
|
|
def __init__(self, *, client: CompletionClient, file_manager: FileManager,
|
|
request: AiderEditRequest, prompt: str, alignment: bool,
|
|
logger: logging.Logger, disambiguation: str) -> None:
|
|
"""Bind one semantic question; only contradiction review can override ordinary escalation."""
|
|
super().__init__(client=client, file_manager=file_manager,
|
|
request=replace(request, future_guidance=()), prompt=prompt, logger=logger, disambiguation=disambiguation)
|
|
self._alignment = alignment
|
|
self._completion: CompletionClient = client
|
|
self.validation = ValidationResult()
|
|
|
|
@property
|
|
def valid_response_references(self) -> list[object]:
|
|
"""Both reviews use the same closed verdict and correction contract."""
|
|
return self._REFERENCE
|
|
|
|
def _decide(self, response: object | None = None,
|
|
validation: ValidationResult | None = None) -> tuple[object | None, ValidationResult]:
|
|
"""Admit a review verdict, confirming a COT contradiction at the strongest tier."""
|
|
prompt = correction_prompt(self.prompt, response, validation, self._prompt_formatter_builder)
|
|
# Ordinary corrections retain CycleJacker; only a COT contradiction forces the strongest tier.
|
|
while True:
|
|
with completion_client.scoped_completion(type(self).__name__, self.escalation.files):
|
|
raw = self._completion.complete(prompt, logger=self.logger)
|
|
data = json.loads(json_response_decoder.strip_json_fence(raw))
|
|
checked = validation_utils.validate_shape(data, self.valid_response_references, location=ValidationLocation([]))
|
|
# Probabilistic structure errors return to the parent's stationary correction loop.
|
|
if not checked.valid:
|
|
return data, checked
|
|
result = cast(dict[str, str], data)
|
|
outcome, reason, instruction = result["outcome"], result["reason"], result["instruction"]
|
|
# A rejection must tell the producer what to correct; a pass must not conceal a correction.
|
|
if outcome not in {"pass", "fail"} or not reason.strip() or bool(instruction.strip()) != (outcome == "fail"):
|
|
return data, ValidationResult(errors=(SemanticValidationError(ValidationLocation([]),
|
|
"Require pass/fail, a nonblank reason, and corrective instruction only for fail."),))
|
|
# Review the identical candidate before allowing a weaker alignment rejection to control repair.
|
|
if self._alignment and outcome == "fail" and self._completion is not self.escalation:
|
|
self.escalation.request_strongest()
|
|
self._completion = self.escalation
|
|
continue
|
|
self.outcome, self.reason = outcome, reason
|
|
self.additional = {"instruction": instruction}
|
|
self.validation = ValidationResult() if outcome == "pass" else ValidationResult(errors=(SemanticValidationError(
|
|
ValidationLocation(["alignment" if self._alignment else "detail"]), reason + "\n" + instruction,
|
|
),))
|
|
self.changes.append((self, outcome == "pass"))
|
|
return data, ValidationResult()
|
|
|
|
|
|
class LocalizationFeedback(StabilizationDecision):
|
|
"""Explain mechanically established gaps without changing admission; call feedback() then read reason."""
|
|
|
|
@property
|
|
def valid_response_references(self) -> list[object]:
|
|
"""The feedback call supplies guidance, not another admission verdict."""
|
|
return [{"summary": "Explain the missing obligation and what needs reconsideration."}]
|
|
|
|
def _decide(self, response: object | None = None,
|
|
validation: ValidationResult | None = None) -> tuple[object | None, ValidationResult]:
|
|
"""Use the existing correction owner to obtain a usable explanation of fixed coverage gaps."""
|
|
prompt = correction_prompt(self.prompt, response, validation, self._prompt_formatter_builder)
|
|
with completion_client.scoped_completion(type(self).__name__, self.escalation.files):
|
|
raw = self._client.complete(prompt, logger=self.logger)
|
|
data = json.loads(json_response_decoder.strip_json_fence(raw))
|
|
checked = validation_utils.validate_shape(data, self.valid_response_references, location=ValidationLocation([]))
|
|
# An explanation must be valid before the producer is asked to reason from it.
|
|
if not checked.valid:
|
|
return data, checked
|
|
summary = cast(dict[str, str], data)["summary"]
|
|
# Empty feedback would leave the requested semantic explanation absent.
|
|
if not summary.strip():
|
|
return data, ValidationResult(errors=(SemanticValidationError(ValidationLocation(["summary"]),
|
|
"Summarize the missing obligations and why the localized instructions do not cover them."),))
|
|
self.reason = summary
|
|
self.changes.append((self, True))
|
|
return data, ValidationResult()
|
|
|
|
|
|
class PromptLocalization(StabilizationDecision):
|
|
"""Rewrite one invocation as localized prompts with original source and diagnostic ranges."""
|
|
|
|
# Quarantined after lost patch contracts and costly helper retries; see docs/localization.md.
|
|
# Work Logs use existing calls only and must not implicitly enable full-prompt localization.
|
|
enabled: ClassVar[bool] = False
|
|
|
|
_REFERENCE: list[object] = [[{"prompt": "Localized instructions, exact diagnostic, then {{SOURCE}}",
|
|
**_CITATION_REFERENCE}]]
|
|
|
|
def __init__(self, *, client: CompletionClient, file_manager: FileManager, request: AiderEditRequest,
|
|
owner: str, logger: logging.Logger | None = None, current_cot: str | None = None) -> None:
|
|
"""Bind the existing owner and authority without making its decision or performing edits."""
|
|
super().__init__(client=client, file_manager=file_manager, request=replace(request, future_guidance=()), logger=logger)
|
|
self._owner = owner
|
|
self._cot = file_manager.prompt if current_cot is None else current_cot
|
|
self._future_guidance = request.future_guidance
|
|
self._completion: CompletionClient = client
|
|
self._original = ""
|
|
self._input = ""
|
|
self._required: dict[str, frozenset[int]] = {}
|
|
self._stacks: tuple[LocalSource, ...] = ()
|
|
|
|
@property
|
|
def valid_response_references(self) -> list[object]:
|
|
"""Declare the array of actual prompts and optional original-input ranges."""
|
|
return self._REFERENCE
|
|
|
|
def localize(self, prompt: str) -> str:
|
|
"""Present localization instructions first, then the complete numbered original input."""
|
|
if not self.enabled:
|
|
return prompt
|
|
section = self._prompt_formatter_builder.generate
|
|
future = "\n\n".join(map(lambda value: value.message + "\n" + value.justification, self._future_guidance))
|
|
self._original = prompt + ("\n\n" + section("FUTURE GUIDANCE")(future) if future else "")
|
|
self._input = self._original
|
|
# Number additional authority without changing the original source or its final newline.
|
|
for title, value in (("GOVERNING COT", self._cot),
|
|
("BACKLOG GUIDANCE", "\n".join(self.request.guidance))):
|
|
# Already-rendered authority retains its original coordinates.
|
|
if value.strip() and value not in self._input:
|
|
self._input += "\n\n" + section(title)(value)
|
|
numbered = "".join(map(lambda entry: f"{entry[0]} | {entry[1]}",
|
|
enumerate(self._input.splitlines(keepends=True), 1)))
|
|
requirements = LocalizationRequirements(client=self._client, file_manager=self._file_manager,
|
|
request=self.request, original=self._input, evidence=self._original, logger=self.logger,
|
|
prompt="\n\n".join((section("LOCALIZATION APPLICABILITY")(
|
|
"Identify everything this invocation must address before any localized prompt is generated. "
|
|
"You have only original input, never a candidate summary. Do not perform the downstream task "
|
|
"or pre-answer whether work is required. Select applicable COT requirements, including "
|
|
"verification obligations; current backlog guidance; relevant future guidance consistent "
|
|
"with the governing COT; and the source needed for a coherent decision or repair. "
|
|
"COT controls the overall goal; backlog and failure identify current work. Do not narrow "
|
|
"the goal merely because its primary owner is another file. Preserve all applicable explicit "
|
|
"observable conditions and verification requirements. Extra source is "
|
|
"acceptable; unrelated source need not be retained. For diagnostics, select COMPLETE "
|
|
"failure/stack-trace blocks, including frames, assertion details and terminal errors; "
|
|
"never select only an illustrative excerpt. Stack text will be matched mechanically, "
|
|
"not accepted through semantic similarity. Do not invent diagnostics or missing sections. "
|
|
"Return one JSON object with all five citation fields, without instruction prose. "
|
|
"Code/stack use lists of inclusive [start, end] ranges, such as [[1, 10], [20, 30]]. "
|
|
"COT/backlog/future are lists of individual input line numbers, not ranges. Use [] when "
|
|
"nothing applies. All numbers refer to the supplied numbered input, not source-file lines. "
|
|
"Do not reproduce source bodies. These selections will remain fixed through localization "
|
|
"corrections. Return JSON only: " + json.dumps(LocalizationRequirements._REFERENCE[0])),
|
|
section("CURRENT WORK")("Responsible agent: " + self._owner),
|
|
section("GOVERNING COT")(self._cot),
|
|
section("EDITABLE FILES")("\n".join(map(self._file_manager.repo_location, self.request.editable_files))),
|
|
section("READONLY FILES")("\n".join(map(self._file_manager.repo_location, self.request.readonly_files))),
|
|
section("NUMBERED ORIGINAL INPUT")("N | prefixes are coordinates, not content.\n" + numbered))))
|
|
requirements()
|
|
self._required = cast(dict[str, frozenset[int]], requirements.additional["required"])
|
|
self._stacks = cast(tuple[LocalSource, ...], requirements.additional["stacks"])
|
|
instructions = "\n\n".join((
|
|
section("YOUR TASK")(
|
|
"You are a task summarizer. Turn the supplied work into concise, comprehensive localized "
|
|
"instructions for the responsible agent. Each prompt you return is the actual instruction "
|
|
"text for its relevant source, not a token-only template. It replaces the old task wrapper. "
|
|
"Write what the agent must do; do not perform the work or return its final verdict."),
|
|
section("HOW TO USE EACH INPUT SECTION")(
|
|
"1. COT: Preserve the overall goal, constraints and verification obligations. Never contradict "
|
|
"the goal. Adapt it to the current backlog assignment and the available code. Relevant prior "
|
|
"COTs supply additional constraints without replacing the current goal. Identify the COT lines "
|
|
"this item addresses in relevant cot, including applicable verification requirements.\n"
|
|
"2. BACKLOG AND FAILURE: These identify the work to do now. Break the goal into concrete "
|
|
"steps addressing the current failure: name exact files and symbols, explain the observed "
|
|
"behavior, describe the repair or decision criteria, then specify focused verification. "
|
|
"Check factual claims against the code; if the cause is uncertain, name the check needed.\n"
|
|
"3. DIAGNOSTICS: Repeat relevant failure diagnostics verbatim in prompt, including indentation, "
|
|
"line numbers and code excerpts within the diagnostic. Use relevant stack to identify their "
|
|
"original-input lines. A quoted diagnostic must exist exactly in the original prompt; never "
|
|
"invent or paraphrase one. When no diagnostic applies, omit relevant stack.\n"
|
|
"4. SOURCE: Read current code, baseline code, diffs and reference cards. Preserve file paths "
|
|
"and symbol names exactly. Do not reproduce code bodies in your response. Select complete "
|
|
"source sections, including headers and renderer metadata, through relevant code. Put "
|
|
"{{SOURCE}} where that exact code should be inserted, or leave it out to append code after "
|
|
"this prompt. Use additional items for additional source sections. Keep the source needed for "
|
|
"a coherent task; unrelated source may be omitted. Do not copy the old task wrapper. "
|
|
"Dependency proximity or a removed passage "
|
|
"alone does not establish a defect.\n"
|
|
"5. GUIDANCE AND FEEDBACK: Apply current backlog guidance and the latest correction; consider "
|
|
"relevant future guidance without contradicting the COT. Distinguish an unapplied patch "
|
|
"from a code failure. Rejected answers are evidence, not authority. Localization validation "
|
|
"feedback asks you to repair this response, not to invent a source defect. Identify the backlog "
|
|
"and future guidance lines addressed by this item in relevant backlog guidance and "
|
|
"relevant future guidance. Listing a line does not substitute for addressing it in prompt.\n"
|
|
"6. OWNER AND SCOPE: Keep editable and readonly boundaries. For a decision owner, ask it "
|
|
"to decide whether repairs are needed and describe them when required; it does not execute "
|
|
"edits itself. For an editor, give repair and verification instructions. Preserve existing "
|
|
"offered exits without choosing another file, discharging work or restarting execution.\n"
|
|
"7. DOWNSTREAM RESPONSE: Include the owner's response type, exact fields, value types, "
|
|
"allowed values, field-coherence rules, citations and acceptable alternatives. Repeat "
|
|
"supplied format examples exactly and identify their values as examples, not a verdict. "
|
|
"Preserve required reasons even for no-work answers. This contract belongs inside prompt; "
|
|
"your own response uses the array below.\n"
|
|
"8. ORDER: Put prerequisites before dependent actions and verification. Cover each applicable "
|
|
"goal requirement and failure. Remove repetition, not obligations. Already-satisfied work "
|
|
"still needs decision criteria; do not return an empty response."),
|
|
section("CURRENT WORK")("Responsible agent: " + self._owner),
|
|
section("GOVERNING COT")(self._cot),
|
|
section("EDITABLE FILES")("\n".join(map(self._file_manager.repo_location, self.request.editable_files))),
|
|
section("READONLY FILES")("\n".join(map(self._file_manager.repo_location, self.request.readonly_files))),
|
|
section("RESPONSE CONTRACT")(
|
|
"Return a nonempty JSON array. Each item has prompt (actual localized instruction text), "
|
|
"optional relevant code, relevant stack, relevant cot, relevant backlog guidance and relevant "
|
|
"future guidance. Code/stack each use a list of [start, end] ranges, such as [[1, 10], [20, 30]], 1-based "
|
|
"inclusive in the numbered original input, not source-file coordinates. Omitted ranges "
|
|
"or [] mean none. COT/backlog/future fields list individual input line numbers, not ranges. "
|
|
"Independently identify all applicable requirements from the original input. Your selections "
|
|
"will be compared with a separate applicability check whose answer is not supplied here. "
|
|
"Split, overlapping and repeated ranges are compared by their union of covered input lines. "
|
|
"Extra coverage is allowed. Required goals and guidance must actually be addressed in "
|
|
"prompt, not merely listed. Required stack traces must be quoted in FULL, verbatim, with "
|
|
"their input ranges cited; semantic paraphrases or partial traces will not pass. "
|
|
"A diagnostic selected by relevant stack must be quoted exactly in that "
|
|
"item's prompt. Source is inserted by code, never generated again. There are no steps, "
|
|
"reason or instruction-placeholder fields. The source token is optional, at most one per "
|
|
"item, and requires relevant code. The numbers in this example are illustrative:\n" +
|
|
json.dumps(self._REFERENCE[0])),
|
|
))
|
|
self.prompt = section("PROMPT LOCALIZATION")(instructions) + "\n\n" + section("NUMBERED ORIGINAL INPUT")(
|
|
"Each N | prefix labels an input line and is not part of its content. Supplied COT/guidance absent from the original is appended below it.\n" + numbered)
|
|
self()
|
|
return cast(str, self.additional["localized_prompt"])
|
|
|
|
def _decide(self, response: object | None = None,
|
|
validation: ValidationResult | None = None) -> tuple[object | None, ValidationResult]:
|
|
"""Validate provenance and semantic reviews before delivering the assembled prompt array."""
|
|
prompt = correction_prompt(self.prompt, response, validation, self._prompt_formatter_builder)
|
|
with completion_client.scoped_completion(type(self).__name__, self.escalation.files):
|
|
raw = self._completion.complete(prompt, logger=self.logger)
|
|
data = json.loads(json_response_decoder.strip_json_fence(raw))
|
|
# Optional ranges have an empty meaning; preserve unknown fields for the existing shape validator.
|
|
normalized = list(map(lambda item: {**dict.fromkeys(_AUTHORITY_FIELDS, []), "relevant code": [], "relevant stack": [], **item}
|
|
if isinstance(item, dict) else item, data)) if isinstance(data, list) else data
|
|
checked = validation_utils.validate_shape(normalized, self.valid_response_references, location=ValidationLocation([]))
|
|
# Malformed responses return to the existing correction loop before evidence is interpreted.
|
|
if not checked.valid:
|
|
return data, checked
|
|
items = cast(list[dict[str, object]], normalized)
|
|
# An empty answer or blank prompt cannot discharge the owner's assignment.
|
|
if not items or any(not cast(str, item["prompt"]).strip() for item in items):
|
|
return data, ValidationResult(errors=(SemanticValidationError(ValidationLocation([]),
|
|
"Return a nonempty array of actual, nonblank localized prompts."),))
|
|
# Reference shape validates element types; ranges additionally require exactly two coordinates.
|
|
if any(len(span) != 2 for item in items for key in ("relevant code", "relevant stack")
|
|
for span in cast(list[list[int]], item[key])):
|
|
return data, ValidationResult(errors=(MechanicalValidationError(ValidationLocation([]),
|
|
"Each relevant code or relevant stack must be a list of [start, end] ranges, or omitted/empty."),))
|
|
lines = self._input.splitlines()
|
|
selected = dict(map(lambda key: (key, frozenset(chain.from_iterable(map(
|
|
lambda item: cast(list[int], item[key]), items)))), _AUTHORITY_FIELDS))
|
|
# Claimed obligations must address the same input used for independent applicability selection.
|
|
if any(not 1 <= line <= len(lines) for line in chain.from_iterable(selected.values())):
|
|
return data, ValidationResult(errors=(MechanicalValidationError(ValidationLocation([]),
|
|
"COT and guidance fields must list existing individual numbered-input lines."),))
|
|
candidates = []
|
|
# Bind each instruction to its selected original evidence without copying source into model output.
|
|
for item in items:
|
|
code, stack = cast(list[list[int]], item["relevant code"]), cast(list[list[int]], item["relevant stack"])
|
|
candidates.append(LocalizedPrompt(self._original, cast(str, item["prompt"]),
|
|
tuple(map(lambda span: LocalSource(*span), code)), tuple(map(lambda span: LocalSource(*span), stack))))
|
|
evidence = LocalizationEvidenceCheck(self._original, tuple(candidates))
|
|
# Only original-input evidence can proceed to semantic review and downstream delivery.
|
|
if not evidence():
|
|
return data, evidence.validation
|
|
delivered = "\n\n".join(map(LocalizedPrompt.render, candidates))
|
|
# Source and trace ranges participate in the same identity-based coverage comparison.
|
|
for key in ("relevant code", "relevant stack"):
|
|
spans = chain.from_iterable(map(lambda item: cast(list[list[int]], item[key]), items))
|
|
selected[key] = frozenset(chain.from_iterable(map(lambda span: range(span[0], span[1] + 1), spans)))
|
|
errors = []
|
|
missing_coverage = {}
|
|
# Applicability is frozen before generation; no candidate-aware review can weaken it.
|
|
for key, required in self._required.items():
|
|
missing = required - selected[key]
|
|
# Equal coverage and supersets pass; duplicates cannot compensate for missing identities.
|
|
if missing:
|
|
missing_coverage[key] = sorted(missing)
|
|
instruction = "Include and cite the missing original source/diagnostic lines."
|
|
# Spell out omitted goals and guidance so regeneration can highlight the actual obligation.
|
|
if key in _AUTHORITY_FIELDS:
|
|
instruction = "Reassess these omitted goal/guidance obligations against the supplied code. " \
|
|
"Explain the needed work in your own instructions; do not pre-answer the downstream decision. " \
|
|
"Original requirements for reference:\n" + "\n".join(map(
|
|
lambda line: f"{line} | {lines[line - 1]}", sorted(missing)))
|
|
errors.append(MechanicalValidationError(ValidationLocation([key]),
|
|
"Missing independently required input lines " + json.dumps(sorted(missing)) + ". " + instruction))
|
|
evidence_lines = self._original.splitlines(keepends=True)
|
|
# Full trace matching is deterministic even if individually cited fragments are otherwise valid.
|
|
for stack in self._stacks:
|
|
diagnostic = "".join(evidence_lines[stack.start_line - 1:stack.end_line]).removesuffix("\n").removesuffix("\r")
|
|
# The complete trace must be quoted and cited together, not incidentally inserted as code.
|
|
quoted = any(diagnostic in candidate.prompt and set(range(stack.start_line, stack.end_line + 1)) <= set(
|
|
chain.from_iterable(map(lambda span: range(span.start_line, span.end_line + 1), candidate.stack)))
|
|
for candidate in candidates)
|
|
if not quoted or diagnostic not in delivered:
|
|
errors.append(MechanicalValidationError(ValidationLocation(["relevant stack"]),
|
|
f"Repeat the complete diagnostic from input lines {stack.start_line}-{stack.end_line} "
|
|
"verbatim as one uninterrupted block and cite its range; partial or paraphrased traces are invalid."))
|
|
# Missing evidence returns to the producer before semantic review can approve incomplete work.
|
|
if errors:
|
|
# A separate semantic explanation guides repair but cannot waive fixed mechanical gaps.
|
|
if missing_coverage.keys() - {"relevant stack"}:
|
|
section = self._prompt_formatter_builder.generate
|
|
feedback = LocalizationFeedback(client=self._client, file_manager=self._file_manager,
|
|
request=replace(self.request, future_guidance=()), logger=self.logger,
|
|
prompt="\n\n".join((section("LOCALIZATION COVERAGE FEEDBACK")(
|
|
"Mechanical coverage validation has already failed. You explain the failure; you cannot "
|
|
"override it, change applicability, or decide the downstream task. Summarize in your own "
|
|
"words what COT, guidance, source or diagnostic obligation is missing, why the current "
|
|
"localized instruction does not cover it, and what the localizer needs to reconsider. "
|
|
"Distinguish a missing citation from missing instruction content: the content may already "
|
|
"address an obligation whose citation is absent. Use the original input and current "
|
|
"candidate as evidence; do not invent an omission. Refer to the relevant input lines and "
|
|
"symbols. Guide its reasoning; do not write replacement localized prose, supply a patch, "
|
|
"or give the downstream answer. Return JSON only: {\"summary\": \"your explanation\"}"),
|
|
section("CURRENT WORK")("Responsible agent: " + self._owner),
|
|
section("NUMBERED ORIGINAL INPUT")("".join(map(lambda entry: f"{entry[0]} | {entry[1]}",
|
|
enumerate(self._input.splitlines(keepends=True), 1)))),
|
|
section("MISSING REQUIRED COVERAGE")(json.dumps(missing_coverage)),
|
|
section("RETURNED LOCALIZED PROMPTS")(json.dumps(data)),
|
|
section("COMPLETE DOWNSTREAM PROMPT")(delivered))))
|
|
feedback()
|
|
errors.append(SemanticValidationError(ValidationLocation(["coverage explanation"]), feedback.reason))
|
|
return data, ValidationResult(errors=tuple(errors))
|
|
section = self._prompt_formatter_builder.generate
|
|
common = "\n\n".join(("Current decision/action: " + self._owner, "GOVERNING COT\n" + self._cot,
|
|
section("EDITABLE FILES")("\n".join(map(self._file_manager.repo_location, self.request.editable_files))),
|
|
section("READONLY FILES")("\n".join(map(self._file_manager.repo_location, self.request.readonly_files))),
|
|
"NUMBERED ORIGINAL INPUT — COMPARISON ONLY\n" + "".join(map(
|
|
lambda entry: f"{entry[0]} | {entry[1]}", enumerate(self._input.splitlines(keepends=True), 1))),
|
|
"INDEPENDENT REQUIRED COVERAGE — FIXED BEFORE LOCALIZATION\n" + json.dumps(dict(map(
|
|
lambda item: (item[0], sorted(item[1])), self._required.items()))),
|
|
"RETURNED LOCALIZED PROMPTS AND INPUT RANGES\n" + json.dumps(data),
|
|
"COMPLETE DOWNSTREAM PROMPT\n" + delivered,
|
|
"Ranges use the numbered input, not source-file line numbers. Mechanical checks already "
|
|
"validated coordinates, required line coverage and complete exact diagnostic quotations. "
|
|
"Do not reselect or shrink applicability after seeing the candidate. Judge whether the "
|
|
"actual instructions address every independently selected COT/backlog/future obligation "
|
|
"in substance; listing a line number is not sufficient. Detail review must fail if a "
|
|
"required goal or guidance is absent from the prose. In correction, name its input lines and "
|
|
"summarize the missing obligation or relationship. Ask the localizer to reason from the original "
|
|
"code about how to address it; do not draft replacement localized prose, a patch, or a downstream "
|
|
"verdict. Alignment still judges direct contradictions. "
|
|
"Do not substitute semantic similarity for the mechanical diagnostic checks.",
|
|
"Return strict JSON: " + json.dumps(LocalizationReview._REFERENCE[0]),
|
|
"Use outcome pass or fail, a nonblank reason, and instruction empty for pass or corrective for fail."))
|
|
alignment = LocalizationReview(client=self._completion, file_manager=self._file_manager, request=self.request,
|
|
prompt=section("LOCALIZATION ALIGNMENT REVIEW")(
|
|
"Reject direct contradictions of the governing goal or constraints. A decision owner "
|
|
"must decide whether edits are required without executing them; that is not a prohibition "
|
|
"on recommending necessary repairs. Different file focus or specificity is not a contradiction.\n" + common),
|
|
alignment=True, logger=self.logger, disambiguation="LocalizationAlignment/" + self._owner)
|
|
detail = LocalizationReview(client=self._client, file_manager=self._file_manager, request=self.request,
|
|
prompt=section("LOCALIZATION DETAIL REVIEW")(
|
|
"Check that the localized prompts together are self-contained: governing goal, current "
|
|
"backlog/failure, relevant future guidance, exact symbols, scope, concrete tasks and "
|
|
"verification in dependency order. Ensure selected source sections are complete and contain "
|
|
"source/diffs/reference rendering rather than the old task wrapper. Source beyond the independently "
|
|
"required lines may be omitted when the task remains coherent. Every quoted failure "
|
|
"diagnostic must exist verbatim in the original "
|
|
"input and be identified by relevant stack; reject invented diagnostics or irrelevant "
|
|
"stack selections. Do not demand reproduction of absent diagnostics or an empty section. "
|
|
"Preserve the downstream response type, schema, exact examples, allowed values, coherence "
|
|
"rules and acceptable alternatives including citations. The localized array is not that "
|
|
"downstream answer. Reject invented work, widened scope, changed exits or lost corrections. "
|
|
"Evaluate the proposed instructions, not whether the current source already has the fix.\n" + common),
|
|
alignment=False, logger=self.logger, disambiguation="LocalizationDetail/" + self._owner)
|
|
checks = (alignment, detail)
|
|
admitted = CompositeCheck(changes=[], aggregate_checks=list(checks))()
|
|
# A rejected review returns to the existing correction owner without admitting a downstream task.
|
|
if not admitted:
|
|
# Confirmed goal contradictions retain the existing strongest-model regeneration policy.
|
|
if not alignment.validation.valid:
|
|
self.escalation.request_strongest()
|
|
self._completion = self.escalation
|
|
rejected = next(filter(lambda check: not check.validation.valid, checks))
|
|
return data, rejected.validation
|
|
self.additional = {"localized_prompt": delivered}
|
|
self.changes.append((self, True))
|
|
return data, ValidationResult()
|