304 lines
12 KiB
Python
304 lines
12 KiB
Python
"""Sequential orchestration and final stability review."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Iterable
|
|
|
|
from mdpolish.component import Component, ComponentContractError
|
|
from mdpolish.edits import apply_component_batch, validate_component_batch
|
|
from mdpolish.models import (
|
|
Change,
|
|
ComponentInfo,
|
|
DocumentSnapshot,
|
|
ErrorStage,
|
|
ProposalReference,
|
|
ProposedChange,
|
|
ResidualProposal,
|
|
RunError,
|
|
RunStatus,
|
|
TransformResult,
|
|
)
|
|
|
|
|
|
class PipelineContractError(ValueError):
|
|
"""A pipeline composition does not satisfy the approved contract."""
|
|
|
|
|
|
class Pipeline:
|
|
"""Run selected components once, then review the final snapshot for stability."""
|
|
|
|
def __init__(self, components: Iterable[Component]) -> None:
|
|
self._components = tuple(components)
|
|
|
|
@property
|
|
def components(self) -> tuple[Component, ...]:
|
|
return self._components
|
|
|
|
def transform(self, markdown: str) -> TransformResult:
|
|
input_snapshot = DocumentSnapshot(markdown)
|
|
component_infos, preflight_error = self._preflight_components()
|
|
if preflight_error is not None:
|
|
return self._failed_result(
|
|
input_snapshot=input_snapshot,
|
|
current_snapshot=input_snapshot,
|
|
component_infos=component_infos,
|
|
changes=(),
|
|
errors=(preflight_error,),
|
|
)
|
|
|
|
current_snapshot = input_snapshot
|
|
changes: list[Change] = []
|
|
for position, (component, expected_info) in enumerate(zip(self._components, component_infos, strict=True)):
|
|
current_info, metadata_error = self._current_component_info(
|
|
component=component,
|
|
expected_info=expected_info,
|
|
position=position,
|
|
stage=ErrorStage.TRANSFORM,
|
|
)
|
|
if metadata_error is not None:
|
|
return self._failed_result(
|
|
input_snapshot=input_snapshot,
|
|
current_snapshot=current_snapshot,
|
|
component_infos=component_infos,
|
|
changes=tuple(changes),
|
|
errors=(metadata_error,),
|
|
)
|
|
|
|
proposals, proposal_error = self._proposals(
|
|
component=component,
|
|
component_info=current_info,
|
|
position=position,
|
|
stage=ErrorStage.TRANSFORM,
|
|
snapshot=current_snapshot,
|
|
)
|
|
if proposal_error is not None:
|
|
return self._failed_result(
|
|
input_snapshot=input_snapshot,
|
|
current_snapshot=current_snapshot,
|
|
component_infos=component_infos,
|
|
changes=tuple(changes),
|
|
errors=(proposal_error,),
|
|
)
|
|
|
|
try:
|
|
applied_batch = apply_component_batch(
|
|
snapshot=current_snapshot,
|
|
proposals=proposals,
|
|
component=current_info,
|
|
component_position=position,
|
|
)
|
|
except Exception as error:
|
|
return self._failed_result(
|
|
input_snapshot=input_snapshot,
|
|
current_snapshot=current_snapshot,
|
|
component_infos=component_infos,
|
|
changes=tuple(changes),
|
|
errors=(
|
|
self._run_error(
|
|
component_info=current_info,
|
|
position=position,
|
|
stage=ErrorStage.TRANSFORM,
|
|
error=error,
|
|
unexpected_message="component edit batch does not satisfy the edit contract",
|
|
),
|
|
),
|
|
)
|
|
|
|
current_snapshot = applied_batch.snapshot
|
|
changes.extend(applied_batch.changes)
|
|
|
|
review_errors: list[RunError] = []
|
|
residual_proposals: list[ResidualProposal] = []
|
|
for position, (component, expected_info) in enumerate(zip(self._components, component_infos, strict=True)):
|
|
current_info, metadata_error = self._current_component_info(
|
|
component=component,
|
|
expected_info=expected_info,
|
|
position=position,
|
|
stage=ErrorStage.FINAL_REVIEW,
|
|
)
|
|
if metadata_error is not None:
|
|
review_errors.append(metadata_error)
|
|
continue
|
|
|
|
proposals, proposal_error = self._proposals(
|
|
component=component,
|
|
component_info=current_info,
|
|
position=position,
|
|
stage=ErrorStage.FINAL_REVIEW,
|
|
snapshot=current_snapshot,
|
|
)
|
|
if proposal_error is not None:
|
|
review_errors.append(proposal_error)
|
|
continue
|
|
|
|
try:
|
|
validate_component_batch(current_snapshot, proposals)
|
|
except Exception as error:
|
|
review_errors.append(
|
|
self._run_error(
|
|
component_info=current_info,
|
|
position=position,
|
|
stage=ErrorStage.FINAL_REVIEW,
|
|
error=error,
|
|
unexpected_message="component edit batch does not satisfy the edit contract",
|
|
)
|
|
)
|
|
continue
|
|
|
|
residual_proposals.extend(
|
|
ResidualProposal(
|
|
component_id=current_info.component_id,
|
|
component_version=current_info.version,
|
|
component_position=position,
|
|
proposal_ref=ProposalReference(
|
|
component_position=position,
|
|
snapshot_sha256=current_snapshot.sha256,
|
|
proposal_index=proposal_index,
|
|
),
|
|
proposal=proposal,
|
|
)
|
|
for proposal_index, proposal in enumerate(proposals)
|
|
)
|
|
|
|
if review_errors:
|
|
return self._failed_result(
|
|
input_snapshot=input_snapshot,
|
|
current_snapshot=current_snapshot,
|
|
component_infos=component_infos,
|
|
changes=tuple(changes),
|
|
errors=tuple(review_errors),
|
|
residual_proposals=tuple(residual_proposals),
|
|
)
|
|
if residual_proposals:
|
|
return TransformResult(
|
|
status=RunStatus.UNSTABLE,
|
|
input_sha256=input_snapshot.sha256,
|
|
current_sha256=current_snapshot.sha256,
|
|
components=component_infos,
|
|
changes=tuple(changes),
|
|
residual_proposals=tuple(residual_proposals),
|
|
partial_markdown=current_snapshot.markdown,
|
|
)
|
|
return TransformResult(
|
|
status=RunStatus.SUCCESS,
|
|
input_sha256=input_snapshot.sha256,
|
|
current_sha256=current_snapshot.sha256,
|
|
components=component_infos,
|
|
changes=tuple(changes),
|
|
output_markdown=current_snapshot.markdown,
|
|
)
|
|
|
|
def _preflight_components(self) -> tuple[tuple[ComponentInfo, ...], RunError | None]:
|
|
component_infos: list[ComponentInfo] = []
|
|
seen_ids: set[str] = set()
|
|
for position, component in enumerate(self._components):
|
|
if not isinstance(component, Component):
|
|
error = ComponentContractError("pipeline entries must be Component instances")
|
|
return tuple(component_infos), self._run_error(
|
|
component_info=None,
|
|
position=position,
|
|
stage=ErrorStage.TRANSFORM,
|
|
error=error,
|
|
unexpected_message="a pipeline entry is not a Component instance",
|
|
)
|
|
try:
|
|
component_info = component._component_info()
|
|
except Exception as error:
|
|
return tuple(component_infos), self._run_error(
|
|
component_info=None,
|
|
position=position,
|
|
stage=ErrorStage.TRANSFORM,
|
|
error=error,
|
|
unexpected_message="component metadata does not satisfy the component contract",
|
|
)
|
|
component_infos.append(component_info)
|
|
if component_info.component_id in seen_ids:
|
|
duplicate_error = PipelineContractError("pipeline component_id values must be unique")
|
|
return tuple(component_infos), self._run_error(
|
|
component_info=component_info,
|
|
position=position,
|
|
stage=ErrorStage.TRANSFORM,
|
|
error=duplicate_error,
|
|
unexpected_message="pipeline component_id values must be unique",
|
|
)
|
|
seen_ids.add(component_info.component_id)
|
|
return tuple(component_infos), None
|
|
|
|
def _current_component_info(
|
|
self,
|
|
component: Component,
|
|
expected_info: ComponentInfo,
|
|
position: int,
|
|
stage: ErrorStage,
|
|
) -> tuple[ComponentInfo, RunError | None]:
|
|
try:
|
|
current_info = component._component_info()
|
|
if current_info != expected_info:
|
|
raise ComponentContractError("component metadata changed during pipeline execution")
|
|
except Exception as error:
|
|
return expected_info, self._run_error(
|
|
component_info=expected_info,
|
|
position=position,
|
|
stage=stage,
|
|
error=error,
|
|
unexpected_message="component metadata does not satisfy the component contract",
|
|
)
|
|
return current_info, None
|
|
|
|
def _proposals(
|
|
self,
|
|
component: Component,
|
|
component_info: ComponentInfo,
|
|
position: int,
|
|
stage: ErrorStage,
|
|
snapshot: DocumentSnapshot,
|
|
) -> tuple[tuple[ProposedChange, ...], RunError | None]:
|
|
try:
|
|
proposals = component._collect_proposals(snapshot)
|
|
except Exception as error:
|
|
return (), self._run_error(
|
|
component_info=component_info,
|
|
position=position,
|
|
stage=stage,
|
|
error=error,
|
|
unexpected_message="component could not produce contract-valid proposals",
|
|
)
|
|
return proposals, None
|
|
|
|
@staticmethod
|
|
def _run_error(
|
|
component_info: ComponentInfo | None,
|
|
position: int,
|
|
stage: ErrorStage,
|
|
error: Exception,
|
|
unexpected_message: str,
|
|
) -> RunError:
|
|
return RunError(
|
|
component_id=component_info.component_id if component_info is not None else "<invalid>",
|
|
component_version=component_info.version if component_info is not None else "<invalid>",
|
|
component_position=position,
|
|
stage=stage,
|
|
error_type=type(error).__name__,
|
|
message=unexpected_message,
|
|
)
|
|
|
|
@staticmethod
|
|
def _failed_result(
|
|
input_snapshot: DocumentSnapshot,
|
|
current_snapshot: DocumentSnapshot,
|
|
component_infos: tuple[ComponentInfo, ...],
|
|
changes: tuple[Change, ...],
|
|
errors: tuple[RunError, ...],
|
|
residual_proposals: tuple[ResidualProposal, ...] = (),
|
|
) -> TransformResult:
|
|
return TransformResult(
|
|
status=RunStatus.FAILED,
|
|
input_sha256=input_snapshot.sha256,
|
|
current_sha256=current_snapshot.sha256,
|
|
components=component_infos,
|
|
changes=changes,
|
|
errors=errors,
|
|
residual_proposals=residual_proposals,
|
|
partial_markdown=current_snapshot.markdown,
|
|
)
|