692 lines
32 KiB
Python
692 lines
32 KiB
Python
"""Strict adapters from published mdpolish artifacts to the reviewer API."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import dataclass
|
|
from hashlib import sha256
|
|
from pathlib import Path
|
|
from typing import NoReturn, TypeAlias, cast
|
|
|
|
from mdpolish._artifact_replay import (
|
|
LocatedChange,
|
|
ReplayChange,
|
|
ReplayComponent,
|
|
ReplayError,
|
|
ReplayResult,
|
|
replay_change_chain,
|
|
)
|
|
|
|
JsonValue: TypeAlias = bool | int | float | str | list["JsonValue"] | dict[str, "JsonValue"] | None
|
|
JsonObject: TypeAlias = dict[str, JsonValue]
|
|
_STATUSES = {"success", "failed", "unstable"}
|
|
_ERROR_STAGES = {"transform", "final_review"}
|
|
|
|
|
|
class ReviewArtifactError(ValueError):
|
|
"""Published artifacts cannot be exposed as a trustworthy review response."""
|
|
|
|
def __init__(self, code: str, message: str, http_status: int = 422) -> None:
|
|
super().__init__(message)
|
|
self.code = code
|
|
self.http_status = http_status
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ComponentRecord:
|
|
component_id: str
|
|
version: str
|
|
parameters: JsonValue
|
|
applicability: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ManifestDocument:
|
|
document_id: str
|
|
source_label: str
|
|
status: str
|
|
input_sha256: str
|
|
current_sha256: str
|
|
change_count: int
|
|
result_path: str
|
|
cleaned_path: str | None
|
|
diff_path: str | None
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class LocatorDocument:
|
|
document_id: str
|
|
source_path: Path
|
|
input_sha256: str
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ChangeRecord:
|
|
replay: ReplayChange
|
|
reason: str
|
|
recorded_line: int
|
|
recorded_column: int
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class ResultRecord:
|
|
status: str
|
|
input_sha256: str
|
|
current_sha256: str
|
|
changes: tuple[ChangeRecord, ...]
|
|
errors: tuple[JsonObject, ...]
|
|
residual_proposals: tuple[JsonObject, ...]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class DocumentRecord:
|
|
manifest: ManifestDocument
|
|
result: ResultRecord
|
|
locator: LocatorDocument | None
|
|
|
|
|
|
def _fail(code: str, message: str, http_status: int = 422) -> NoReturn:
|
|
raise ReviewArtifactError(code, message, http_status)
|
|
|
|
|
|
def _object(value: object, label: str) -> dict[str, object]:
|
|
if not isinstance(value, dict) or any(not isinstance(key, str) for key in value):
|
|
_fail("invalid_artifact", f"{label} 必须是 JSON 对象。")
|
|
return cast(dict[str, object], value)
|
|
|
|
|
|
def _array(value: object, label: str) -> list[object]:
|
|
if not isinstance(value, list):
|
|
_fail("invalid_artifact", f"{label} 必须是数组。")
|
|
return cast(list[object], value)
|
|
|
|
|
|
def _string(value: object, label: str, *, allow_empty: bool = False) -> str:
|
|
if not isinstance(value, str) or "\0" in value or (not allow_empty and not value):
|
|
_fail("invalid_artifact", f"{label} 必须是字符串。")
|
|
return value
|
|
|
|
|
|
def _integer(value: object, label: str, *, minimum: int = 0) -> int:
|
|
if isinstance(value, bool) or not isinstance(value, int) or value < minimum:
|
|
_fail("invalid_artifact", f"{label} 必须是不小于 {minimum} 的整数。")
|
|
return value
|
|
|
|
|
|
def _hash(value: object, label: str) -> str:
|
|
digest = _string(value, label)
|
|
if len(digest) != 64 or any(character not in "0123456789abcdef" for character in digest):
|
|
_fail("invalid_artifact", f"{label} 必须是小写 SHA-256 摘要。")
|
|
return digest
|
|
|
|
|
|
def _status(value: object, label: str) -> str:
|
|
status = _string(value, label)
|
|
if status not in _STATUSES:
|
|
_fail("invalid_artifact", f"{label} 不是已知状态。")
|
|
return status
|
|
|
|
|
|
def _json_value(value: object, label: str) -> JsonValue:
|
|
if value is None or isinstance(value, str | int | float | bool):
|
|
return value
|
|
if isinstance(value, list):
|
|
return [_json_value(item, label) for item in value]
|
|
if isinstance(value, dict) and all(isinstance(key, str) for key in value):
|
|
return {cast(str, key): _json_value(item, label) for key, item in value.items()}
|
|
_fail("invalid_artifact", f"{label} 包含不支持的 JSON 值。")
|
|
raise AssertionError("unreachable")
|
|
|
|
|
|
def _read_json(path: Path, label: str) -> dict[str, object]:
|
|
try:
|
|
content = path.read_bytes()
|
|
except OSError:
|
|
_fail("missing_artifact", f"无法读取{label}。")
|
|
if content.startswith(b"\xef\xbb\xbf"):
|
|
_fail("invalid_artifact", f"{label} 不能包含 UTF-8 BOM。")
|
|
if not content.endswith(b"\n"):
|
|
_fail("invalid_artifact", f"{label} 必须以换行结尾。")
|
|
try:
|
|
text = content.decode("utf-8", errors="strict")
|
|
except UnicodeDecodeError:
|
|
_fail("invalid_utf8", f"{label} 不是严格 UTF-8。")
|
|
try:
|
|
payload = cast(
|
|
object,
|
|
json.loads(
|
|
text,
|
|
parse_constant=lambda _value: _fail(
|
|
"invalid_artifact", f"{label} 不能包含非有限数值。"
|
|
),
|
|
),
|
|
)
|
|
except json.JSONDecodeError:
|
|
_fail("invalid_artifact", f"{label} 不是合法 JSON。")
|
|
return _object(payload, label)
|
|
|
|
|
|
def _read_markdown(path: Path, expected_hash: str, label: str, error_code: str) -> str:
|
|
try:
|
|
if path.is_symlink() or not path.is_file():
|
|
_fail(error_code, f"{label}不是普通文件。", 409)
|
|
content = path.read_bytes()
|
|
except OSError:
|
|
_fail(error_code, f"无法读取{label}。", 409)
|
|
if sha256(content).hexdigest() != expected_hash:
|
|
_fail("hash_mismatch", f"{label}的内容哈希已经变化。", 409)
|
|
try:
|
|
return content.decode("utf-8", errors="strict")
|
|
except UnicodeDecodeError:
|
|
_fail("invalid_utf8", f"{label}不是严格 UTF-8。", 409)
|
|
raise AssertionError("unreachable")
|
|
|
|
|
|
def _run_path(run_directory: Path, relative_path: str, label: str) -> Path:
|
|
relative = Path(relative_path)
|
|
if relative.is_absolute() or "\0" in relative_path:
|
|
_fail("unsafe_path", f"{label}必须是运行目录内的相对路径。")
|
|
unresolved = run_directory / relative
|
|
try:
|
|
if unresolved.is_symlink() or not unresolved.is_file():
|
|
_fail("missing_artifact", f"无法读取{label}。")
|
|
resolved = unresolved.resolve(strict=True)
|
|
except OSError:
|
|
_fail("missing_artifact", f"无法读取{label}。")
|
|
if not resolved.is_relative_to(run_directory):
|
|
_fail("unsafe_path", f"{label}越过了运行目录边界。")
|
|
return resolved
|
|
|
|
|
|
def _component(value: object, position: int) -> ComponentRecord:
|
|
item = _object(value, f"pipeline.components[{position}]")
|
|
return ComponentRecord(
|
|
component_id=_string(item.get("component_id"), "component_id"),
|
|
version=_string(item.get("version"), "component version"),
|
|
parameters=_json_value(item.get("parameters"), "component parameters"),
|
|
applicability=_string(item.get("applicability"), "component applicability"),
|
|
)
|
|
|
|
|
|
def _manifest_document(value: object, position: int) -> ManifestDocument:
|
|
item = _object(value, f"manifest.documents[{position}]")
|
|
document_id = _string(item.get("document_id"), "document_id")
|
|
status = _status(item.get("status"), "document status")
|
|
result_path = _string(item.get("result_path"), "result_path")
|
|
cleaned_value = item.get("cleaned_path")
|
|
diff_value = item.get("diff_path")
|
|
cleaned_path = None if cleaned_value is None else _string(cleaned_value, "cleaned_path")
|
|
diff_path = None if diff_value is None else _string(diff_value, "diff_path")
|
|
base = f"documents/{document_id}"
|
|
if (
|
|
result_path != f"{base}/result.json"
|
|
or cleaned_path != (f"{base}/cleaned.md" if status == "success" else None)
|
|
or diff_path != (f"{base}/changes.diff" if status == "success" else None)
|
|
):
|
|
_fail("invalid_artifact", "manifest 文档产物路径不符合 schema 1。")
|
|
return ManifestDocument(
|
|
document_id=document_id,
|
|
source_label=_string(item.get("source_label"), "source_label"),
|
|
status=status,
|
|
input_sha256=_hash(item.get("input_sha256"), "input_sha256"),
|
|
current_sha256=_hash(item.get("current_sha256"), "current_sha256"),
|
|
change_count=_integer(item.get("change_count"), "change_count"),
|
|
result_path=result_path,
|
|
cleaned_path=cleaned_path,
|
|
diff_path=diff_path,
|
|
)
|
|
|
|
|
|
def _parse_change(value: object, position: int) -> ChangeRecord:
|
|
item = _object(value, f"changes[{position}]")
|
|
proposal = _object(item.get("proposal_ref"), "proposal_ref")
|
|
span = _object(item.get("span"), "span")
|
|
location = _object(item.get("location"), "location")
|
|
start = _integer(span.get("start"), "span.start")
|
|
end = _integer(span.get("end"), "span.end")
|
|
if end < start:
|
|
_fail("invalid_artifact", "change span 必须满足 start <= end。")
|
|
replay = ReplayChange(
|
|
component_id=_string(item.get("component_id"), "change component_id"),
|
|
component_version=_string(item.get("component_version"), "change component_version"),
|
|
component_position=_integer(item.get("component_position"), "change component_position"),
|
|
proposal_component_position=_integer(
|
|
proposal.get("component_position"), "proposal component_position"
|
|
),
|
|
proposal_snapshot_sha256=_hash(proposal.get("snapshot_sha256"), "proposal snapshot_sha256"),
|
|
proposal_index=_integer(proposal.get("proposal_index"), "proposal_index"),
|
|
edit_index=_integer(item.get("edit_index"), "edit_index"),
|
|
start=start,
|
|
end=end,
|
|
before=_string(item.get("before"), "before", allow_empty=True),
|
|
after=_string(item.get("after"), "after", allow_empty=True),
|
|
before_sha256=_hash(item.get("before_sha256"), "before_sha256"),
|
|
after_sha256=_hash(item.get("after_sha256"), "after_sha256"),
|
|
)
|
|
return ChangeRecord(
|
|
replay=replay,
|
|
reason=_string(item.get("reason"), "change reason"),
|
|
recorded_line=_integer(location.get("line"), "location.line", minimum=1),
|
|
recorded_column=_integer(location.get("column"), "location.column", minimum=1),
|
|
)
|
|
|
|
|
|
def _parse_error(value: object, position: int) -> JsonObject:
|
|
item = _object(value, f"errors[{position}]")
|
|
stage = _string(item.get("stage"), "error stage")
|
|
if stage not in _ERROR_STAGES:
|
|
_fail("invalid_artifact", "error stage 不是已知阶段。")
|
|
return {
|
|
"component_id": _string(item.get("component_id"), "error component_id"),
|
|
"component_version": _string(item.get("component_version"), "error component_version"),
|
|
"component_position": _integer(item.get("component_position"), "error component_position"),
|
|
"stage": stage,
|
|
"error_type": _string(item.get("error_type"), "error_type"),
|
|
"message": _string(item.get("message"), "error message"),
|
|
}
|
|
|
|
|
|
def _parse_residual(value: object, position: int) -> JsonObject:
|
|
item = _object(value, f"residual_proposals[{position}]")
|
|
reference = _object(item.get("proposal_ref"), "residual proposal_ref")
|
|
proposal = _object(item.get("proposal"), "residual proposal")
|
|
component_position = _integer(item.get("component_position"), "residual component_position")
|
|
proposal_snapshot = _hash(proposal.get("snapshot_sha256"), "residual proposal snapshot_sha256")
|
|
if (
|
|
_integer(reference.get("component_position"), "residual reference component_position")
|
|
!= component_position
|
|
or _hash(reference.get("snapshot_sha256"), "residual reference snapshot_sha256")
|
|
!= proposal_snapshot
|
|
):
|
|
_fail("invalid_artifact", "residual proposal_ref 与候选身份不一致。")
|
|
_integer(reference.get("proposal_index"), "residual proposal_index")
|
|
edits = _array(proposal.get("edits"), "residual proposal edits")
|
|
if not edits:
|
|
_fail("invalid_artifact", "residual proposal edits 不能为空。")
|
|
for edit_position, edit_value in enumerate(edits):
|
|
edit = _object(edit_value, f"residual edit[{edit_position}]")
|
|
span = _object(edit.get("span"), "residual edit span")
|
|
if _hash(edit.get("snapshot_sha256"), "residual edit snapshot_sha256") != proposal_snapshot:
|
|
_fail("invalid_artifact", "residual edit 与候选快照不一致。")
|
|
start = _integer(span.get("start"), "residual span.start")
|
|
end = _integer(span.get("end"), "residual span.end")
|
|
expected = _string(edit.get("expected_text"), "residual expected_text", allow_empty=True)
|
|
_string(edit.get("replacement"), "residual replacement", allow_empty=True)
|
|
if end < start or len(expected) != end - start:
|
|
_fail("invalid_artifact", "residual edit 范围与 expected_text 不一致。")
|
|
return {
|
|
"component_id": _string(item.get("component_id"), "residual component_id"),
|
|
"component_version": _string(item.get("component_version"), "residual component_version"),
|
|
"component_position": component_position,
|
|
"reason": _string(proposal.get("reason"), "residual reason"),
|
|
"edit_count": len(edits),
|
|
}
|
|
|
|
|
|
def _component_json(component: ComponentRecord, position: int, change_count: int) -> JsonObject:
|
|
return {
|
|
"component_position": position,
|
|
"component_id": component.component_id,
|
|
"version": component.version,
|
|
"parameters": component.parameters,
|
|
"applicability": component.applicability,
|
|
"change_count": change_count,
|
|
}
|
|
|
|
|
|
class ReviewArtifacts:
|
|
"""Validated, read-only view over one explicitly selected run directory."""
|
|
|
|
def __init__(self, run_directory: str | Path) -> None:
|
|
requested = Path(run_directory)
|
|
try:
|
|
if requested.is_symlink() or not requested.is_dir():
|
|
_fail("missing_run", "指定的运行目录不存在或不是普通目录。", 400)
|
|
self.run_directory = requested.resolve(strict=True)
|
|
except OSError:
|
|
_fail("missing_run", "无法读取指定的运行目录。", 400)
|
|
|
|
manifest = _read_json(_run_path(self.run_directory, "manifest.json", "manifest.json"), "manifest.json")
|
|
if manifest.get("schema_version") != 1:
|
|
_fail("unsupported_schema", "只支持 manifest.json schema 1。", 409)
|
|
run = _object(manifest.get("run"), "manifest.run")
|
|
pipeline = _object(manifest.get("pipeline"), "manifest.pipeline")
|
|
summary = _object(manifest.get("summary"), "manifest.summary")
|
|
self.run_id = _string(run.get("run_id"), "run_id")
|
|
self.run_json: JsonObject = {
|
|
"run_id": self.run_id,
|
|
"run_date": _string(run.get("run_date"), "run_date"),
|
|
"status": _status(run.get("status"), "run status"),
|
|
"started_at_utc": _string(run.get("started_at_utc"), "started_at_utc"),
|
|
"completed_at_utc": _string(run.get("completed_at_utc"), "completed_at_utc"),
|
|
"retention_until": _string(run.get("retention_until"), "retention_until"),
|
|
}
|
|
self.components = tuple(
|
|
_component(value, position)
|
|
for position, value in enumerate(_array(pipeline.get("components"), "pipeline.components"))
|
|
)
|
|
manifest_documents = tuple(
|
|
_manifest_document(value, position)
|
|
for position, value in enumerate(_array(manifest.get("documents"), "manifest.documents"))
|
|
)
|
|
if len({item.document_id for item in manifest_documents}) != len(manifest_documents):
|
|
_fail("invalid_artifact", "manifest 中的 document_id 必须唯一。")
|
|
self.summary_json: JsonObject = {
|
|
"document_count": _integer(summary.get("document_count"), "summary.document_count"),
|
|
"success_count": _integer(summary.get("success_count"), "summary.success_count"),
|
|
"failed_count": _integer(summary.get("failed_count"), "summary.failed_count"),
|
|
"unstable_count": _integer(summary.get("unstable_count"), "summary.unstable_count"),
|
|
"change_count": _integer(summary.get("change_count"), "summary.change_count"),
|
|
}
|
|
expected_summary: JsonObject = {
|
|
"document_count": len(manifest_documents),
|
|
"success_count": sum(item.status == "success" for item in manifest_documents),
|
|
"failed_count": sum(item.status == "failed" for item in manifest_documents),
|
|
"unstable_count": sum(item.status == "unstable" for item in manifest_documents),
|
|
"change_count": sum(item.change_count for item in manifest_documents),
|
|
}
|
|
if self.summary_json != expected_summary:
|
|
_fail("invalid_artifact", "manifest 汇总与文档索引不一致。")
|
|
expected_status = (
|
|
"failed"
|
|
if expected_summary["failed_count"]
|
|
else "unstable"
|
|
if expected_summary["unstable_count"]
|
|
else "success"
|
|
)
|
|
if self.run_json["status"] != expected_status:
|
|
_fail("invalid_artifact", "manifest 运行状态与文档状态不一致。")
|
|
|
|
locator_path = self.run_directory / "review-locator.json"
|
|
locators: tuple[LocatorDocument, ...] | None = None
|
|
self.original_run_location_changed = False
|
|
if locator_path.exists():
|
|
locator = _read_json(
|
|
_run_path(self.run_directory, "review-locator.json", "review-locator.json"),
|
|
"review-locator.json",
|
|
)
|
|
if locator.get("schema_version") != 1:
|
|
_fail("unsupported_schema", "只支持 review-locator.json schema 1。", 409)
|
|
locator_run = _object(locator.get("run"), "review locator run")
|
|
if (
|
|
_string(locator_run.get("run_id"), "locator run_id") != self.run_id
|
|
or locator_run.get("manifest_path") != "manifest.json"
|
|
):
|
|
_fail("invalid_artifact", "review locator 与 manifest 运行身份不一致。")
|
|
recorded_directory_text = _string(locator_run.get("run_directory"), "locator run_directory")
|
|
recorded_directory = Path(recorded_directory_text)
|
|
if not recorded_directory.is_absolute() or str(recorded_directory) != str(recorded_directory.resolve()):
|
|
_fail("invalid_artifact", "locator run_directory 必须是绝对解析路径。")
|
|
self.original_run_location_changed = recorded_directory != self.run_directory
|
|
parsed_locators: list[LocatorDocument] = []
|
|
for position, value in enumerate(_array(locator.get("documents"), "locator documents")):
|
|
item = _object(value, f"locator.documents[{position}]")
|
|
source_text = _string(item.get("source_path"), "source_path")
|
|
source_path = Path(source_text)
|
|
if not source_path.is_absolute() or str(source_path) != str(source_path.resolve()):
|
|
_fail("invalid_artifact", "source_path 必须是绝对解析路径。")
|
|
parsed_locators.append(
|
|
LocatorDocument(
|
|
document_id=_string(item.get("document_id"), "locator document_id"),
|
|
source_path=source_path,
|
|
input_sha256=_hash(item.get("input_sha256"), "locator input_sha256"),
|
|
)
|
|
)
|
|
locators = tuple(parsed_locators)
|
|
if len(locators) != len(manifest_documents):
|
|
_fail("invalid_artifact", "review locator 文档数量与 manifest 不一致。")
|
|
for indexed, located in zip(manifest_documents, locators, strict=True):
|
|
if indexed.document_id != located.document_id or indexed.input_sha256 != located.input_sha256:
|
|
_fail("invalid_artifact", "review locator 文档身份与 manifest 不一致。")
|
|
|
|
records: list[DocumentRecord] = []
|
|
for position, manifest_document in enumerate(manifest_documents):
|
|
result = self._parse_result(manifest_document)
|
|
records.append(
|
|
DocumentRecord(
|
|
manifest=manifest_document,
|
|
result=result,
|
|
locator=None if locators is None else locators[position],
|
|
)
|
|
)
|
|
self.documents = tuple(records)
|
|
self.documents_by_id = {item.manifest.document_id: item for item in self.documents}
|
|
|
|
def _parse_result(self, manifest: ManifestDocument) -> ResultRecord:
|
|
payload = _read_json(_run_path(self.run_directory, manifest.result_path, "result.json"), "result.json")
|
|
if payload.get("schema_version") != 1:
|
|
_fail("unsupported_schema", "只支持 result.json schema 1。", 409)
|
|
document = _object(payload.get("document"), "result.document")
|
|
if (
|
|
_string(document.get("document_id"), "result document_id") != manifest.document_id
|
|
or _string(document.get("source_label"), "result source_label") != manifest.source_label
|
|
):
|
|
_fail("invalid_artifact", "result 与 manifest 文档身份不一致。")
|
|
status = _status(payload.get("status"), "result status")
|
|
input_hash = _hash(payload.get("input_sha256"), "result input_sha256")
|
|
current_hash = _hash(payload.get("current_sha256"), "result current_sha256")
|
|
changes = tuple(
|
|
_parse_change(value, position)
|
|
for position, value in enumerate(_array(payload.get("changes"), "result changes"))
|
|
)
|
|
errors = tuple(
|
|
_parse_error(value, position)
|
|
for position, value in enumerate(_array(payload.get("errors"), "result errors"))
|
|
)
|
|
residuals = tuple(
|
|
_parse_residual(value, position)
|
|
for position, value in enumerate(_array(payload.get("residual_proposals"), "result residual_proposals"))
|
|
)
|
|
output = _object(payload.get("output"), "result output")
|
|
cleaned = output.get("cleaned_path")
|
|
diff = output.get("diff_path")
|
|
expected_cleaned: object = "cleaned.md" if status == "success" else None
|
|
expected_diff: object = "changes.diff" if status == "success" else None
|
|
if cleaned != expected_cleaned or diff != expected_diff:
|
|
_fail("invalid_artifact", "result 状态与输出路径不一致。")
|
|
if (
|
|
status != manifest.status
|
|
or input_hash != manifest.input_sha256
|
|
or current_hash != manifest.current_sha256
|
|
or len(changes) != manifest.change_count
|
|
):
|
|
_fail("invalid_artifact", "result 与 manifest 文档索引不一致。")
|
|
if status == "success" and (errors or residuals):
|
|
_fail("invalid_artifact", "success 文档不能包含错误或残留候选。")
|
|
if status == "failed" and (not errors or residuals):
|
|
_fail("invalid_artifact", "failed 文档的错误或残留状态不合法。")
|
|
if status == "unstable" and (errors or not residuals):
|
|
_fail("invalid_artifact", "unstable 文档的错误或残留状态不合法。")
|
|
for change in changes:
|
|
position = change.replay.component_position
|
|
if position >= len(self.components):
|
|
_fail("invalid_artifact", "change 没有对应的流水线组件。")
|
|
component = self.components[position]
|
|
if (
|
|
change.replay.component_id != component.component_id
|
|
or change.replay.component_version != component.version
|
|
):
|
|
_fail("invalid_artifact", "change 身份与流水线组件不一致。")
|
|
for evidence in (*errors, *residuals):
|
|
position_value = evidence["component_position"]
|
|
component_id = evidence["component_id"]
|
|
component_version = evidence["component_version"]
|
|
if not isinstance(position_value, int) or position_value >= len(self.components):
|
|
_fail("invalid_artifact", "审计证据没有对应的流水线组件。")
|
|
component = self.components[position_value]
|
|
if component_id != component.component_id or component_version != component.version:
|
|
_fail("invalid_artifact", "审计证据身份与流水线组件不一致。")
|
|
if manifest.diff_path is not None:
|
|
_run_path(self.run_directory, manifest.diff_path, "changes.diff")
|
|
return ResultRecord(status, input_hash, current_hash, changes, errors, residuals)
|
|
|
|
def _source(self, record: DocumentRecord) -> tuple[str | None, str | None]:
|
|
if record.locator is None:
|
|
return None, "这次历史运行没有 review-locator.json，无法定位完整原文。"
|
|
try:
|
|
return (
|
|
_read_markdown(
|
|
record.locator.source_path,
|
|
record.manifest.input_sha256,
|
|
"原文",
|
|
"unavailable_source",
|
|
),
|
|
None,
|
|
)
|
|
except ReviewArtifactError as error:
|
|
return None, str(error)
|
|
|
|
def _cleaned(self, record: DocumentRecord) -> tuple[str | None, str | None]:
|
|
if record.manifest.status != "success" or record.manifest.cleaned_path is None:
|
|
return None, None
|
|
try:
|
|
path = _run_path(self.run_directory, record.manifest.cleaned_path, "cleaned.md")
|
|
return (
|
|
_read_markdown(path, record.manifest.current_sha256, "清洗结果", "missing_artifact"),
|
|
None,
|
|
)
|
|
except ReviewArtifactError as error:
|
|
return None, str(error)
|
|
|
|
def _replay(self, record: DocumentRecord, original: str, cleaned: str | None) -> ReplayResult:
|
|
try:
|
|
replayed = replay_change_chain(
|
|
input_markdown=original,
|
|
input_sha256=record.result.input_sha256,
|
|
components=tuple(ReplayComponent(item.component_id, item.version) for item in self.components),
|
|
changes=tuple(item.replay for item in record.result.changes),
|
|
current_sha256=record.result.current_sha256,
|
|
current_markdown=cleaned,
|
|
include_zero_change_stages=record.result.status == "success",
|
|
)
|
|
except ReplayError as error:
|
|
raise ReviewArtifactError("untrusted_replay", f"产物无法可信重放：{error}", 409) from error
|
|
for recorded, located in zip(record.result.changes, replayed.changes, strict=True):
|
|
if recorded.recorded_line != located.line or recorded.recorded_column != located.column:
|
|
_fail("untrusted_replay", "产物记录的位置与重放快照不一致。", 409)
|
|
return replayed
|
|
|
|
@staticmethod
|
|
def _summary_json(
|
|
record: DocumentRecord,
|
|
original: str | None,
|
|
source_error: str | None,
|
|
cleaned: str | None,
|
|
output_error: str | None,
|
|
) -> JsonObject:
|
|
return {
|
|
"document_id": record.manifest.document_id,
|
|
"source_label": record.manifest.source_label,
|
|
"status": record.manifest.status,
|
|
"input_sha256": record.manifest.input_sha256,
|
|
"current_sha256": record.manifest.current_sha256,
|
|
"change_count": record.manifest.change_count,
|
|
"source_available": original is not None,
|
|
"output_available": cleaned is not None,
|
|
"availability_error": source_error or output_error,
|
|
}
|
|
|
|
def _summary(self, record: DocumentRecord) -> JsonObject:
|
|
original, source_error = self._source(record)
|
|
cleaned, output_error = self._cleaned(record)
|
|
return self._summary_json(record, original, source_error, cleaned, output_error)
|
|
|
|
def _component_summaries(self, changes: tuple[ChangeRecord, ...]) -> list[JsonValue]:
|
|
counts = [0] * len(self.components)
|
|
for change in changes:
|
|
if 0 <= change.replay.component_position < len(counts):
|
|
counts[change.replay.component_position] += 1
|
|
return [_component_json(item, position, counts[position]) for position, item in enumerate(self.components)]
|
|
|
|
@staticmethod
|
|
def _change_json(recorded: ChangeRecord, located: LocatedChange | None) -> JsonObject:
|
|
change = recorded.replay
|
|
return {
|
|
"component_id": change.component_id,
|
|
"component_version": change.component_version,
|
|
"component_position": change.component_position,
|
|
"proposal_ref": {
|
|
"component_position": change.proposal_component_position,
|
|
"snapshot_sha256": change.proposal_snapshot_sha256,
|
|
"proposal_index": change.proposal_index,
|
|
},
|
|
"edit_index": change.edit_index,
|
|
"reason": recorded.reason,
|
|
"location": {
|
|
"line": recorded.recorded_line if located is None else located.line,
|
|
"column": recorded.recorded_column if located is None else located.column,
|
|
},
|
|
"editor_range": (
|
|
None if located is None else {"start": located.editor_start, "end": located.editor_end}
|
|
),
|
|
"before": change.before,
|
|
"after": change.after,
|
|
}
|
|
|
|
def run_summary(self) -> JsonObject:
|
|
all_changes = tuple(change for document in self.documents for change in document.result.changes)
|
|
return {
|
|
"schema_version": 1,
|
|
"run": self.run_json,
|
|
"components": self._component_summaries(all_changes),
|
|
"documents": [self._summary(item) for item in self.documents],
|
|
"summary": self.summary_json,
|
|
"original_run_location_changed": self.original_run_location_changed,
|
|
}
|
|
|
|
def _document(self, document_id: str) -> DocumentRecord:
|
|
record = self.documents_by_id.get(document_id)
|
|
if record is None:
|
|
_fail("unknown_document", "文档不存在。", 404)
|
|
return record
|
|
|
|
def document_comparison(self, document_id: str) -> JsonObject:
|
|
record = self._document(document_id)
|
|
original, source_error = self._source(record)
|
|
cleaned, output_error = self._cleaned(record)
|
|
replayed = None if original is None else self._replay(record, original, cleaned)
|
|
located = () if replayed is None else replayed.changes
|
|
return {
|
|
"schema_version": 1,
|
|
"document": self._summary_json(record, original, source_error, cleaned, output_error),
|
|
"components": self._component_summaries(record.result.changes),
|
|
"original_markdown": original,
|
|
"cleaned_markdown": cleaned,
|
|
"changes": [
|
|
self._change_json(change, located[position] if position < len(located) else None)
|
|
for position, change in enumerate(record.result.changes)
|
|
],
|
|
"errors": list(record.result.errors),
|
|
"residual_proposals": list(record.result.residual_proposals),
|
|
}
|
|
|
|
def component_stage(self, document_id: str, component_position: int) -> JsonObject:
|
|
record = self._document(document_id)
|
|
if record.manifest.status != "success":
|
|
_fail("stage_unavailable", "只有 success 文档具有完整组件阶段。", 409)
|
|
if component_position < 0 or component_position >= len(self.components):
|
|
_fail("unknown_component", "组件位置不存在。", 404)
|
|
original, source_error = self._source(record)
|
|
cleaned, output_error = self._cleaned(record)
|
|
if original is None or cleaned is None:
|
|
_fail("comparison_unavailable", source_error or output_error or "组件阶段不可用。", 409)
|
|
replayed = self._replay(record, original, cleaned)
|
|
stage = replayed.stages[component_position]
|
|
component = self.components[component_position]
|
|
recorded_changes = tuple(
|
|
item for item in record.result.changes if item.replay.component_position == component_position
|
|
)
|
|
return {
|
|
"schema_version": 1,
|
|
"document_id": record.manifest.document_id,
|
|
"component": _component_json(component, component_position, len(stage.changes)),
|
|
"before_sha256": stage.before_sha256,
|
|
"after_sha256": stage.after_sha256,
|
|
"before_markdown": stage.before_markdown,
|
|
"after_markdown": stage.after_markdown,
|
|
"changes": [
|
|
self._change_json(change, located)
|
|
for change, located in zip(recorded_changes, stage.changes, strict=True)
|
|
],
|
|
}
|