340 lines
13 KiB
Python
340 lines
13 KiB
Python
# SPDX-FileCopyrightText: 2026 Art Sea LLC
|
|
#
|
|
# SPDX-License-Identifier: AGPL-3.0-only
|
|
|
|
"""Translate executable phase starts to and from neutral durable records."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
|
|
from band_aider.context.class_meta import ClassMeta, ContentType
|
|
from band_aider.context.failing_test import FailingTest
|
|
from band_aider.context.provenance import FutureGuidance
|
|
from band_aider.context.working_backlog import (
|
|
ChangeType,
|
|
FailedQAItem,
|
|
InspectionBacklog,
|
|
TestingBacklog,
|
|
)
|
|
|
|
|
|
def phase_start_record(item: InspectionBacklog) -> dict[str, object]:
|
|
"""Return the complete language-neutral durable record for one phase start."""
|
|
|
|
if isinstance(item, FailedQAItem):
|
|
record = _testing_phase_start_record(item)
|
|
record["kind"] = "remediation"
|
|
record["failed_test"] = _failing_test_record(item.failed_test)
|
|
return record
|
|
if isinstance(item, TestingBacklog):
|
|
record = _testing_phase_start_record(item)
|
|
record["kind"] = "testing"
|
|
return record
|
|
if isinstance(item, InspectionBacklog):
|
|
record = _base_phase_start_record(item)
|
|
record["kind"] = "inspection"
|
|
return record
|
|
raise TypeError(f"Provenance cannot persist unsupported phase start {type(item).__name__}.")
|
|
|
|
|
|
def phase_start_from_record(record: dict[str, object]) -> InspectionBacklog:
|
|
"""Restore one executable phase start from its language-neutral durable record."""
|
|
|
|
kind = str(record.get("kind", ""))
|
|
# Rebuild the exact phase-start kind that owns the restored work.
|
|
if kind == "inspection":
|
|
return _inspection_from_record(record)
|
|
elif kind == "testing":
|
|
return _testing_from_record(record)
|
|
elif kind == "remediation":
|
|
return _remediation_from_record(record)
|
|
else:
|
|
raise ValueError(f"Provenance phase-start kind is unsupported: {kind!r}.")
|
|
|
|
|
|
def _base_phase_start_record(item: InspectionBacklog) -> dict[str, object]:
|
|
"""Return durable fields shared by every Alpha and Beta phase start."""
|
|
|
|
return {
|
|
"focus": _class_meta_record(item.focused_spotlight),
|
|
"summary": item.summary,
|
|
"upstream": [_class_meta_record(source) for source in item.upstream_classes],
|
|
"downstream": [_class_meta_record(source) for source in item.downstream_classes],
|
|
"future_guidance": [_future_guidance_record(guidance) for guidance in item.future_guidance],
|
|
}
|
|
|
|
|
|
def _testing_phase_start_record(item: TestingBacklog) -> dict[str, object]:
|
|
"""Return the durable test-specific extension of one phase-start record."""
|
|
|
|
record = _base_phase_start_record(item)
|
|
record.update({
|
|
"original_cot": item.original_cot,
|
|
"previous_beta_guidance": item.previous_beta_guidance,
|
|
"change": _change_record(item.change_type),
|
|
"failure_error_message": item.failure_error_message,
|
|
"failure_stack_trace": list(item.failure_stack_trace),
|
|
"failure_raw_output": item.failure_raw_output,
|
|
"failure_run_info": item.failure_run_info,
|
|
"failure_context": [_class_meta_record(source) for source in item.failure_context_sources],
|
|
"last_exam": item.last_exam,
|
|
})
|
|
# A suite sentinel has no originating test; concrete failures use the shared codec.
|
|
if item.focused_spotlight is not None:
|
|
failure = FailingTest(
|
|
test=item.focused_spotlight,
|
|
error_message=item.failure_error_message,
|
|
stack_trace=list(item.failure_stack_trace),
|
|
raw_output=item.failure_raw_output,
|
|
test_name=item.failure_test_name,
|
|
outcome=item.failure_outcome,
|
|
run_info=item.failure_run_info,
|
|
)
|
|
record.update(map(lambda entry: (f"failure_{entry[0]}", entry[1]), failure.to_record().items()))
|
|
return record
|
|
|
|
|
|
def _inspection_from_record(record: dict[str, object]) -> InspectionBacklog:
|
|
"""Restore the fields shared by every Alpha and Beta phase start."""
|
|
|
|
return InspectionBacklog(
|
|
focused_spotlight=_class_meta_from_record(record.get("focus")),
|
|
summary=str(record.get("summary", "")),
|
|
upstream_classes=_class_meta_list(record.get("upstream"), "upstream"),
|
|
downstream_classes=_class_meta_list(record.get("downstream"), "downstream"),
|
|
future_guidance=_future_guidance_list(record.get("future_guidance")),
|
|
)
|
|
|
|
|
|
def _testing_from_record(record: dict[str, object]) -> TestingBacklog:
|
|
"""Restore one Beta phase start from its durable fields."""
|
|
|
|
base = _inspection_from_record(record)
|
|
run_info = record.get("failure_run_info", "")
|
|
# Diagnostic execution records are optional in old checkpoints, never coerced.
|
|
if not isinstance(run_info, str):
|
|
raise ValueError("failure_run_info must be a string.")
|
|
item = TestingBacklog(
|
|
focused_spotlight=base.focused_spotlight,
|
|
summary=base.summary,
|
|
upstream_classes=base.upstream_classes,
|
|
downstream_classes=base.downstream_classes,
|
|
future_guidance=base.future_guidance,
|
|
original_cot=str(record.get("original_cot", "")),
|
|
previous_beta_guidance=str(record.get("previous_beta_guidance", "")),
|
|
change_type=_change_from_record(record.get("change")),
|
|
failure_error_message=str(record.get("failure_error_message", "")),
|
|
failure_stack_trace=_string_list(record.get("failure_stack_trace"), "failure_stack_trace"),
|
|
failure_raw_output=str(record.get("failure_raw_output", "")),
|
|
failure_run_info=run_info,
|
|
failure_context_sources=_class_meta_list(record.get("failure_context"), "failure_context"),
|
|
last_exam=bool(record.get("last_exam", False)),
|
|
)
|
|
# Suite sentinels retain their existing schema without inventing a failing test owner.
|
|
if base.focused_spotlight is not None:
|
|
failure = FailingTest.from_record(base.focused_spotlight, {
|
|
"error_message": record.get("failure_error_message", ""),
|
|
"stack_trace": record.get("failure_stack_trace", []),
|
|
"raw_output": record.get("failure_raw_output", ""),
|
|
"test_name": record.get("failure_test_name", ""),
|
|
"outcome": record.get("failure_outcome"),
|
|
"run_info": run_info,
|
|
})
|
|
item.failure_error_message = failure.error_message
|
|
item.failure_stack_trace = list(failure.stack_trace)
|
|
item.failure_raw_output = failure.raw_output
|
|
item.failure_test_name = failure.test_name
|
|
item.failure_outcome = failure.outcome
|
|
item.failure_run_info = failure.run_info
|
|
return item
|
|
|
|
|
|
def _remediation_from_record(record: dict[str, object]) -> FailedQAItem:
|
|
"""Restore one Alpha remediation phase start from its durable fields."""
|
|
|
|
testing = _testing_from_record(record)
|
|
return FailedQAItem(
|
|
focused_spotlight=testing.focused_spotlight,
|
|
summary=testing.summary,
|
|
upstream_classes=testing.upstream_classes,
|
|
downstream_classes=testing.downstream_classes,
|
|
future_guidance=testing.future_guidance,
|
|
original_cot=testing.original_cot,
|
|
previous_beta_guidance=testing.previous_beta_guidance,
|
|
change_type=testing.change_type,
|
|
failure_error_message=testing.failure_error_message,
|
|
failure_stack_trace=testing.failure_stack_trace,
|
|
failure_raw_output=testing.failure_raw_output,
|
|
failure_test_name=testing.failure_test_name,
|
|
failure_outcome=testing.failure_outcome,
|
|
failure_run_info=testing.failure_run_info,
|
|
failure_context_sources=testing.failure_context_sources,
|
|
last_exam=testing.last_exam,
|
|
failed_test=_failing_test_from_record(record.get("failed_test")),
|
|
)
|
|
|
|
|
|
def _class_meta_record(meta: ClassMeta | None) -> dict[str, object] | None:
|
|
"""Return durable file evidence without retaining runtime collaborators."""
|
|
|
|
if meta is None:
|
|
return None
|
|
history: list[dict[str, str]] = []
|
|
# Preserve every prior change observation in its recorded order.
|
|
for timestamp, actor, summary in meta.cdc:
|
|
history.append({
|
|
"timestamp": timestamp.isoformat(),
|
|
"actor": actor,
|
|
"summary": summary,
|
|
})
|
|
return {
|
|
"location": meta.location,
|
|
"summary": meta.summary,
|
|
"relevance": list(meta.relevance),
|
|
"relevant_symbols": dict(meta.relevant_symbols),
|
|
"change_history": history,
|
|
"content_type": meta.content_type.value,
|
|
}
|
|
|
|
|
|
def _class_meta_from_record(value: object) -> ClassMeta | None:
|
|
"""Restore durable file evidence as current metadata without runtime collaborators."""
|
|
|
|
if value is None:
|
|
return None
|
|
record = _object_record(value, "file evidence")
|
|
history: list[tuple[datetime, str, str]] = []
|
|
# Rebuild every change observation in its original order.
|
|
for entry in _record_list(record.get("change_history"), "change_history"):
|
|
change = _object_record(entry, "change_history entry")
|
|
history.append((
|
|
datetime.fromisoformat(str(change.get("timestamp", ""))),
|
|
str(change.get("actor", "")),
|
|
str(change.get("summary", "")),
|
|
))
|
|
return ClassMeta(
|
|
location=str(record.get("location", "")),
|
|
summary=str(record.get("summary", "")),
|
|
relevance=_string_list(record.get("relevance"), "relevance"),
|
|
relevant_symbols=_string_mapping(record.get("relevant_symbols"), "relevant_symbols"),
|
|
cdc=history,
|
|
content_type=ContentType(str(record.get("content_type", ContentType.Source.value))),
|
|
)
|
|
|
|
|
|
def _class_meta_list(value: object, name: str) -> list[ClassMeta]:
|
|
"""Restore one ordered file-evidence collection from a durable record."""
|
|
|
|
result: list[ClassMeta] = []
|
|
# Require each collection member to describe concrete file evidence.
|
|
for entry in _record_list(value, name):
|
|
metadata = _class_meta_from_record(entry)
|
|
if metadata is None:
|
|
raise ValueError(f"Provenance {name} cannot contain null file evidence.")
|
|
result.append(metadata)
|
|
return result
|
|
|
|
|
|
def _future_guidance_record(guidance: FutureGuidance) -> dict[str, str]:
|
|
"""Return one later observation attached to a revived phase start."""
|
|
|
|
return {
|
|
"origin_message_id": guidance.origin_message_id,
|
|
"addressed_message_id": guidance.addressed_message_id,
|
|
"message": guidance.message,
|
|
"justification": guidance.justification,
|
|
}
|
|
|
|
|
|
def _future_guidance_list(value: object) -> tuple[FutureGuidance, ...]:
|
|
"""Restore every later observation attached to a revived phase start."""
|
|
|
|
result: list[FutureGuidance] = []
|
|
# Preserve all later observations in the order they were attached.
|
|
for entry in _record_list(value, "future_guidance"):
|
|
guidance = _object_record(entry, "future_guidance entry")
|
|
result.append(FutureGuidance(
|
|
origin_message_id=str(guidance.get("origin_message_id", "")),
|
|
addressed_message_id=str(guidance.get("addressed_message_id", "")),
|
|
message=str(guidance.get("message", "")),
|
|
justification=str(guidance.get("justification", "")),
|
|
))
|
|
return tuple(result)
|
|
|
|
|
|
def _failing_test_record(failure: FailingTest | None) -> dict[str, object] | None:
|
|
"""Return current failure evidence when Alpha remediation owns it."""
|
|
|
|
if failure is None:
|
|
return None
|
|
return {
|
|
"test": _class_meta_record(failure.test),
|
|
**failure.to_record(),
|
|
}
|
|
|
|
|
|
def _failing_test_from_record(value: object) -> FailingTest | None:
|
|
"""Restore current failure evidence for one Alpha remediation phase start."""
|
|
|
|
if value is None:
|
|
return None
|
|
record = _object_record(value, "failed_test")
|
|
test = _class_meta_from_record(record.get("test"))
|
|
if test is None:
|
|
raise ValueError("Provenance failed_test requires file evidence.")
|
|
return FailingTest.from_record(test, record)
|
|
|
|
|
|
def _change_record(change: ChangeType) -> str:
|
|
"""Return the neutral lifecycle word for one test-work change."""
|
|
|
|
names = {
|
|
ChangeType.Create: "create",
|
|
ChangeType.Update: "update",
|
|
ChangeType.Delete: "delete",
|
|
}
|
|
return names[change]
|
|
|
|
|
|
def _change_from_record(value: object) -> ChangeType:
|
|
"""Restore one test-work change from its neutral lifecycle word."""
|
|
|
|
changes = {
|
|
"create": ChangeType.Create,
|
|
"update": ChangeType.Update,
|
|
"delete": ChangeType.Delete,
|
|
}
|
|
name = str(value)
|
|
if name not in changes:
|
|
raise ValueError(f"Provenance change is unsupported: {name!r}.")
|
|
return changes[name]
|
|
|
|
|
|
def _object_record(value: object, name: str) -> dict[str, object]:
|
|
"""Require one JSON object from the durable phase-start record."""
|
|
|
|
if not isinstance(value, dict):
|
|
raise ValueError(f"Provenance {name} must be a JSON object.")
|
|
return value
|
|
|
|
|
|
def _record_list(value: object, name: str) -> list[object]:
|
|
"""Require one JSON array from the durable phase-start record."""
|
|
|
|
if not isinstance(value, list):
|
|
raise ValueError(f"Provenance {name} must be a JSON array.")
|
|
return value
|
|
|
|
|
|
def _string_list(value: object, name: str) -> list[str]:
|
|
"""Restore one ordered text collection from the durable phase-start record."""
|
|
|
|
return [str(item) for item in _record_list(value, name)]
|
|
|
|
|
|
def _string_mapping(value: object, name: str) -> dict[str, str]:
|
|
"""Restore one text mapping from the durable phase-start record."""
|
|
|
|
record = _object_record(value, name)
|
|
return {str(key): str(item) for key, item in record.items()}
|