From 292b382868c21ed52f0774f4a126dafcd893c4f2 Mon Sep 17 00:00:00 2001 From: Tom Softreck Date: Wed, 2 Sep 2026 00:58:43 +0200 Subject: [PATCH] feat: persist resumable delivery plan DAGs --- CHANGELOG.md | 5 + docs/PUBLIC_CONTRACTS.md | 36 + planfile/__init__.py | 50 ++ planfile/delivery_plan.py | 822 ++++++++++++++++++ planfile/delivery_plan_contracts.py | 429 +++++++++ .../delivery-plan-state.schema.v1.json | 224 +++++ pyproject.toml | 2 +- tests/test_delivery_plan.py | 315 +++++++ 8 files changed, 1882 insertions(+), 1 deletion(-) create mode 100644 planfile/delivery_plan.py create mode 100644 planfile/delivery_plan_contracts.py create mode 100644 planfile/schemas/delivery-plan-state.schema.v1.json create mode 100644 tests/test_delivery_plan.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 8c7856d..dc31d38 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,11 @@ frequently changing full-history query retains one serialized body, not four. ### Added +- Added a durable `planfile.delivery-plan-state/v1` runtime for materializing + Strategy's inert ticket DAG, preserving parent/child split links, explicit + continuation checkpoints and deduplicated protected terminal receipts. The + resume projection survives a missing final state replace by recovering exact + candidate ticket IDs, and never derives lifecycle state from chat prose. - Added a public, append-only `PLOG/1` forensic timeline in `.planfile/events/logs.dsl.txt`, daily history partitions, bounded text/JSON APIs, streaming backfill from SODL, and durable coverage for evidence and diff --git a/docs/PUBLIC_CONTRACTS.md b/docs/PUBLIC_CONTRACTS.md index bcda52f..fd2a7d6 100644 --- a/docs/PUBLIC_CONTRACTS.md +++ b/docs/PUBLIC_CONTRACTS.md @@ -101,6 +101,42 @@ the storage transport. Compatibility policy: add a new version instead of weakening strict validation or changing the meaning of an existing result code. +## Resumable delivery-plan DAG v1 + +`Planfile.materialize_delivery_plan` accepts the closed, execution-inert +`subactor.compiled-work-plan/v1` DTO emitted by Strategy. It atomically assigns +stable Planfile ticket IDs to its candidates and persists the runtime projection +under `.planfile/delivery-plans/.json`. Replaying the same DTO returns +the same IDs; if power fails after the ticket batch lands but before the state +replace, the exact candidate hash bindings recover that batch without creating +duplicates. + +```python +state = planfile.materialize_delivery_plan(compiled_plan) +state = planfile.checkpoint_delivery_candidate(checkpoint_v1) +state = planfile.record_delivery_split(split_v1) +state = planfile.record_delivery_terminal_receipt(plan_id, terminal_receipt_v1) +frontier = planfile.resume_delivery_plan(plan_id) +``` + +Checkpoints have an explicit sequence and phase and bind the plan hash, +candidate digest and evidence references. Splits bind a `split_required` parent +to at least two already-materialized bounded children and rewire its dependents. +Terminal receipts are immutable and deduplicated by their complete protected +binding. A conflicting checkpoint, receipt, plan revision or candidate digest +fails closed. + +The returned resume frontier is a deterministic projection of those structured +records. Planfile does not infer completion from chat prose and the contracts +reject command, executor, grant and tool-authority fields. Materialization +creates tickets with `executor=None`; selecting an agent or executing tools +remains a separate authorized runtime concern. + +The persisted format is published as +`planfile/schemas/delivery-plan-state.schema.v1.json`. The state file and its +ticket YAML/event journals are authoritative continuity data and must be backed +up together; disposable SQLite and fast-JSON indexes are not substitutes. + ## Atomic external evidence append `POST /tickets/{ticket_id}/evidence` is the retry-safe write contract for an diff --git a/planfile/__init__.py b/planfile/__init__.py index 33061e5..2f7dacd 100644 --- a/planfile/__init__.py +++ b/planfile/__init__.py @@ -38,6 +38,14 @@ ) from planfile.core.store import PlanfileStore from planfile.delegation import DelegationActor, load_delegation_actors +from planfile.delivery_plan import DeliveryPlanError, DeliveryPlanRepository +from planfile.delivery_plan_contracts import ( + CompiledWorkPlanV1, + DeliveryCheckpointV1, + DeliveryPlanSplitV1, + TicketCandidateV1, + WorkPlanTerminalReceiptV1, +) from planfile.dsl import DSLExecutor, DSLParser, DSLResult from planfile.testql_integration import ( build_testql_tickets, @@ -719,6 +727,45 @@ def create_tickets_bulk( tickets.append(Ticket(id=ticket_id, **data)) return self.store._create_tickets_bulk_unlocked(tickets) + @property + def delivery_plans(self) -> DeliveryPlanRepository: + """Return the durable, execution-inert delivery-plan repository.""" + + return DeliveryPlanRepository(self.store) + + def materialize_delivery_plan( + self, + compiled_plan: dict, + *, + terminal_receipts: list[dict] | tuple[dict, ...] = (), + ) -> dict: + """Atomically materialize a Strategy DAG or recover its exact prior IDs.""" + + return self.delivery_plans.materialize( + compiled_plan, + terminal_receipts=terminal_receipts, + ) + + def checkpoint_delivery_candidate(self, checkpoint: dict) -> dict: + """Append one explicit, hash-bound continuation checkpoint.""" + + return self.delivery_plans.record_checkpoint(checkpoint) + + def record_delivery_terminal_receipt(self, plan_id: str, receipt: dict) -> dict: + """Deduplicate a protected terminal receipt and close its exact slice.""" + + return self.delivery_plans.record_terminal_receipt(plan_id, receipt) + + def record_delivery_split(self, split: dict) -> dict: + """Link a split-required parent to already materialized bounded children.""" + + return self.delivery_plans.record_split(split) + + def resume_delivery_plan(self, plan_id: str) -> dict: + """Read the deterministic continuation frontier without executing tools.""" + + return self.delivery_plans.resume(plan_id) + def quick_ticket(name: str, tool: str = "unknown", **kwargs) -> Ticket: """One-liner ticket creation for tools.""" @@ -736,6 +783,9 @@ def quick_ticket(name: str, tool: str = "unknown", **kwargs) -> Ticket: # Tickets "Ticket", "TicketStatus", "TicketSource", "TicketExecutor", "TicketExecution", "TicketInputs", "TicketOutputs", + "CompiledWorkPlanV1", "DeliveryCheckpointV1", "DeliveryPlanError", + "DeliveryPlanRepository", "DeliveryPlanSplitV1", "TicketCandidateV1", + "WorkPlanTerminalReceiptV1", # Store & API "PlanfileStore", "Planfile", "quick_ticket", # Executors (lazy loaded) diff --git a/planfile/delivery_plan.py b/planfile/delivery_plan.py new file mode 100644 index 0000000..bd58d49 --- /dev/null +++ b/planfile/delivery_plan.py @@ -0,0 +1,822 @@ +"""Durable, idempotent materialization of inert Strategy delivery plans.""" + +from __future__ import annotations + +import copy +import json +import re +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, NoReturn, cast + +from planfile.core.fastio import _atomic_write_text +from planfile.core.models import Ticket, TicketExecution, TicketOutputs, TicketSource, TicketStatus +from planfile.delivery_plan_contracts import ( + DELIVERY_PLAN_RESUME_SCHEMA, + DELIVERY_PLAN_STATE_SCHEMA, + CompiledWorkPlanV1, + DeliveryCheckpointV1, + DeliveryPlacementV1, + DeliveryPlanSplitV1, + TicketCandidateV1, + WorkPlanTerminalReceiptV1, +) + + +class DeliveryPlanError(ValueError): + """A stable delivery-plan contract or persistence failure.""" + + +def _fail(code: str) -> NoReturn: + raise DeliveryPlanError(code) + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _dump(model: Any) -> dict[str, Any]: + return cast( + dict[str, Any], + model.model_dump(mode="json", by_alias=True, exclude_none=False), + ) + + +class DeliveryPlanRepository: + """Persist delivery DAG state next to Planfile's authoritative ticket store. + + Every mutation shares Planfile's cross-process store lock and replaces one + complete JSON state file. Ticket creation is replay-safe: candidate source + bindings are sufficient to recover when the ticket batch landed but a power + loss happened before the state file replace. + """ + + def __init__(self, store: Any): + self.store = store + self.base_dir = Path(store.base_dir) / "delivery-plans" + + @staticmethod + def _plan_id(value: str) -> str: + if not re.fullmatch(r"[a-z][a-z0-9-]{1,63}", str(value or "")): + _fail("delivery_plan_id_invalid") + return str(value) + + def _path(self, plan_id: str) -> Path: + return self.base_dir / f"{self._plan_id(plan_id)}.json" + + def _read_unlocked(self, plan_id: str) -> dict[str, Any] | None: + try: + value = json.loads(self._path(plan_id).read_text(encoding="utf-8")) + except FileNotFoundError: + return None + except (OSError, ValueError) as exc: + raise DeliveryPlanError("delivery_plan_state_unreadable") from exc + self._validate_state(value) + return cast(dict[str, Any], value) + + def _write_unlocked(self, state: dict[str, Any]) -> None: + self._validate_state(state) + content = json.dumps(state, ensure_ascii=False, indent=2, sort_keys=True) + "\n" + _atomic_write_text(self._path(str(state["plan_id"])), content) + + @staticmethod + def _validate_state(value: object) -> None: + if not isinstance(value, dict): + _fail("delivery_plan_state_invalid") + state = cast(dict[str, Any], value) + expected = { + "schema", + "revision", + "plan_id", + "plan_ref", + "plan_hash", + "repository", + "accepted_base_sha", + "target_branch", + "placement", + "created_at", + "updated_at", + "candidates", + } + if set(state) != expected or state.get("schema") != DELIVERY_PLAN_STATE_SCHEMA: + _fail("delivery_plan_state_invalid") + if not isinstance(state.get("revision"), int) or state["revision"] < 1: + _fail("delivery_plan_state_revision_invalid") + if not re.fullmatch(r"[a-z][a-z0-9-]{1,63}", str(state.get("plan_id") or "")): + _fail("delivery_plan_state_identity_invalid") + if not re.fullmatch( + r"(?:artifact|knowledge)://[^\s?#]+", str(state.get("plan_ref") or "") + ): + _fail("delivery_plan_state_identity_invalid") + if not re.fullmatch(r"sha256:[a-f0-9]{64}", str(state.get("plan_hash") or "")): + _fail("delivery_plan_state_identity_invalid") + if not re.fullmatch( + r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", str(state.get("repository") or "") + ): + _fail("delivery_plan_state_identity_invalid") + if not re.fullmatch(r"[a-f0-9]{40}", str(state.get("accepted_base_sha") or "")): + _fail("delivery_plan_state_identity_invalid") + if not re.fullmatch( + r"(?!/)(?!.*(?:\.\.|//|@\{|[~^:?*\[\\]))[A-Za-z0-9._/-]+", + str(state.get("target_branch") or ""), + ): + _fail("delivery_plan_state_identity_invalid") + DeliveryPlacementV1.model_validate(state.get("placement")) + try: + created_at = datetime.fromisoformat(str(state["created_at"])) + updated_at = datetime.fromisoformat(str(state["updated_at"])) + except ValueError as exc: + raise DeliveryPlanError("delivery_plan_state_time_invalid") from exc + if created_at.tzinfo is None or updated_at.tzinfo is None or updated_at < created_at: + _fail("delivery_plan_state_time_invalid") + candidates = state.get("candidates") + if not isinstance(candidates, dict) or not candidates: + _fail("delivery_plan_state_candidates_invalid") + ticket_ids: set[str] = set() + checkpoint_refs: set[str] = set() + receipt_refs: set[str] = set() + split_refs: set[str] = set() + for key, node in candidates.items(): + if not isinstance(node, dict) or set(node) != { + "candidate", + "ticket_id", + "status", + "parent_candidate_key", + "child_candidate_keys", + "depends_on", + "checkpoints", + "terminal_receipt", + "split", + }: + _fail("delivery_plan_state_candidate_invalid") + candidate = TicketCandidateV1.model_validate(node["candidate"]) + if ( + key != candidate.candidate_key + or candidate.plan_hash != state["plan_hash"] + or candidate.placement.model_dump(mode="json") != state["placement"] + ): + _fail("delivery_plan_state_candidate_binding_mismatch") + ticket_id = str(node.get("ticket_id") or "") + if not ticket_id or ticket_id in ticket_ids: + _fail("delivery_plan_state_ticket_binding_invalid") + ticket_ids.add(ticket_id) + if node.get("status") not in { + "materialized", + "checkpointed", + "split_required", + "split", + "terminal", + }: + _fail("delivery_plan_state_candidate_status_invalid") + children = node.get("child_candidate_keys") + dependencies = node.get("depends_on") + checkpoints = node.get("checkpoints") + if ( + not isinstance(children, list) + or len(set(children)) != len(children) + or not all(isinstance(item, str) for item in children) + ): + _fail("delivery_plan_state_children_invalid") + if ( + not isinstance(dependencies, list) + or len(set(dependencies)) != len(dependencies) + or not all(isinstance(item, str) for item in dependencies) + ): + _fail("delivery_plan_state_dependencies_invalid") + if not isinstance(checkpoints, list): + _fail("delivery_plan_state_checkpoints_invalid") + for sequence, checkpoint in enumerate(checkpoints, start=1): + parsed = DeliveryCheckpointV1.model_validate(checkpoint) + if ( + parsed.candidate_key != key + or parsed.plan_id != state["plan_id"] + or parsed.plan_hash != state["plan_hash"] + or parsed.candidate_digest != candidate.candidate_digest + or parsed.sequence != sequence + or parsed.checkpoint_ref in checkpoint_refs + ): + _fail("delivery_plan_state_checkpoint_binding_mismatch") + checkpoint_refs.add(parsed.checkpoint_ref) + receipt = node.get("terminal_receipt") + if receipt is not None: + parsed_receipt = WorkPlanTerminalReceiptV1.model_validate(receipt) + if ( + parsed_receipt.candidate_key != key + or parsed_receipt.plan_hash != state["plan_hash"] + or parsed_receipt.accepted_base_sha != state["accepted_base_sha"] + or parsed_receipt.candidate_digest != candidate.candidate_digest + or parsed_receipt.receipt_ref in receipt_refs + ): + _fail("delivery_plan_state_receipt_binding_mismatch") + receipt_refs.add(parsed_receipt.receipt_ref) + split = node.get("split") + if split is not None: + parsed_split = DeliveryPlanSplitV1.model_validate(split) + if ( + parsed_split.parent_candidate_key != key + or parsed_split.plan_id != state["plan_id"] + or parsed_split.plan_hash != state["plan_hash"] + or parsed_split.parent_candidate_digest != candidate.candidate_digest + or tuple(node["child_candidate_keys"]) != parsed_split.child_candidate_keys + or parsed_split.split_ref in split_refs + ): + _fail("delivery_plan_state_split_binding_mismatch") + split_refs.add(parsed_split.split_ref) + status = node["status"] + if status == "materialized" and (checkpoints or receipt is not None or split is not None): + _fail("delivery_plan_state_candidate_status_invalid") + if status == "checkpointed" and (not checkpoints or receipt is not None or split is not None): + _fail("delivery_plan_state_candidate_status_invalid") + if status == "split_required" and ( + candidate.state != "split_required" + or checkpoints + or receipt is not None + or split is not None + ): + _fail("delivery_plan_state_candidate_status_invalid") + if status == "split" and ( + candidate.state != "split_required" or checkpoints or split is None + ): + _fail("delivery_plan_state_candidate_status_invalid") + if status == "terminal" and (receipt is None or split is not None): + _fail("delivery_plan_state_candidate_status_invalid") + known = set(candidates) + for key, node in candidates.items(): + parent = node.get("parent_candidate_key") + if parent is not None and parent not in known: + _fail("delivery_plan_state_parent_unknown") + if any(child not in known for child in node["child_candidate_keys"]): + _fail("delivery_plan_state_child_unknown") + if any(dependency not in known for dependency in node["depends_on"]): + _fail("delivery_plan_state_dependency_unknown") + if key in node["depends_on"] or key in node["child_candidate_keys"]: + _fail("delivery_plan_state_cycle_invalid") + for child in node["child_candidate_keys"]: + if candidates[child]["parent_candidate_key"] != key: + _fail("delivery_plan_state_child_binding_mismatch") + parent = node["parent_candidate_key"] + if parent is not None and key not in candidates[parent]["child_candidate_keys"]: + _fail("delivery_plan_state_parent_binding_mismatch") + DeliveryPlanRepository._validate_dependency_dag(candidates) + + @staticmethod + def _validate_dependency_dag(candidates: dict[str, dict[str, Any]]) -> None: + pending = { + key: {str(dependency) for dependency in node["depends_on"]} + for key, node in candidates.items() + } + while pending: + ready = {key for key, dependencies in pending.items() if not dependencies} + if not ready: + _fail("delivery_plan_state_dependency_cycle") + for key in ready: + pending.pop(key) + for dependencies in pending.values(): + dependencies.difference_update(ready) + + @staticmethod + def _receipt_map( + plan: CompiledWorkPlanV1, + receipts: list[dict[str, Any]] | tuple[dict[str, Any], ...], + ) -> dict[str, WorkPlanTerminalReceiptV1]: + candidates = {candidate.candidate_key: candidate for candidate in plan.candidates} + by_candidate: dict[str, WorkPlanTerminalReceiptV1] = {} + by_ref: dict[str, WorkPlanTerminalReceiptV1] = {} + for raw in receipts: + receipt = WorkPlanTerminalReceiptV1.model_validate(raw) + candidate = candidates.get(receipt.candidate_key) + if candidate is None: + _fail("delivery_plan_receipt_candidate_unknown") + if ( + receipt.plan_hash != plan.plan_hash + or receipt.candidate_digest != candidate.candidate_digest + or receipt.accepted_base_sha != plan.accepted_base_sha + ): + _fail("delivery_plan_receipt_stale") + previous = by_candidate.get(receipt.candidate_key) + if previous is not None and previous != receipt: + _fail("delivery_plan_receipt_conflict") + ref_owner = by_ref.get(receipt.receipt_ref) + if ref_owner is not None and ref_owner != receipt: + _fail("delivery_plan_receipt_ref_conflict") + by_candidate[receipt.candidate_key] = receipt + by_ref[receipt.receipt_ref] = receipt + for candidate in plan.candidates: + if candidate.state == "terminal": + terminal_receipt = by_candidate.get(candidate.candidate_key) + if ( + terminal_receipt is not None + and terminal_receipt.receipt_ref != candidate.terminal_receipt_ref + ): + _fail("delivery_plan_terminal_receipt_conflict") + elif candidate.candidate_key in by_candidate: + _fail("delivery_plan_receipt_for_nonterminal_candidate") + return by_candidate + + @staticmethod + def _candidate_context(record: dict[str, Any]) -> dict[str, Any]: + source = record.get("source") + if not isinstance(source, dict): + return {} + context = source.get("context") + return context if isinstance(context, dict) else {} + + def _existing_ticket_bindings_unlocked( + self, + plan: CompiledWorkPlanV1, + ) -> dict[str, dict[str, Any]]: + expected = {candidate.candidate_key: candidate for candidate in plan.candidates} + found: dict[str, dict[str, Any]] = {} + for record in self.store.ticket_records(sprint="all"): + context = self._candidate_context(record) + if context.get("delivery_plan_id") != plan.plan_id: + continue + source = record.get("source") + if not isinstance(source, dict) or source.get("tool") != "subactor.strategy": + _fail("delivery_plan_existing_candidate_conflict") + key = str(context.get("candidate_key") or "") + candidate = expected.get(key) + if candidate is None: + _fail("delivery_plan_existing_candidate_unknown") + if ( + context.get("plan_hash") != plan.plan_hash + or context.get("candidate_digest") != candidate.candidate_digest + or context.get("idempotency_key") != candidate.idempotency_key + ): + _fail("delivery_plan_existing_candidate_conflict") + if key in found: + _fail("delivery_plan_existing_candidate_duplicate") + found[key] = record + return found + + @staticmethod + def _ticket_for( + candidate: TicketCandidateV1, + ticket_id: str, + dependency_ids: list[str], + receipt: WorkPlanTerminalReceiptV1 | None, + plan: CompiledWorkPlanV1, + ) -> Ticket: + context = { + "delivery_plan_id": plan.plan_id, + "delivery_plan_ref": plan.plan_ref, + "plan_hash": plan.plan_hash, + "candidate_key": candidate.candidate_key, + "candidate_digest": candidate.candidate_digest, + "idempotency_key": candidate.idempotency_key, + "workstream": candidate.workstream, + "execution": "inert", + "authority": "none", + } + labels = [ + "delivery-plan", + f"plan:{plan.plan_id}", + f"candidate:{candidate.candidate_key}", + f"workstream:{candidate.workstream}", + ] + execution_state = "pending" + status = TicketStatus.open + outputs = None + if candidate.state == "split_required": + execution_state = "waiting_input" + labels.append("split-required") + elif candidate.state == "terminal": + execution_state = "done" + status = TicketStatus.done + if receipt is None: + _fail("delivery_plan_terminal_receipt_missing") + outputs = TicketOutputs(completion_receipt=_dump(receipt)) + return Ticket( + id=ticket_id, + name=candidate.title, + status=status, + description=( + f"Materialized from {plan.plan_ref}; candidate {candidate.candidate_key}. " + "Execution authority is intentionally absent." + ), + labels=labels, + files=list(candidate.allowed_paths), + acceptance_criteria=[criterion.statement for criterion in candidate.acceptance], + blocked_by=dependency_ids, + source=TicketSource(tool="subactor.strategy", version="work-plan/v1", context=context), + execution=TicketExecution(state=execution_state), + outputs=outputs, + ) + + @staticmethod + def _node( + candidate: TicketCandidateV1, + ticket_id: str, + receipt: WorkPlanTerminalReceiptV1 | None, + ) -> dict[str, Any]: + status = { + "pending": "materialized", + "split_required": "split_required", + "terminal": "terminal", + }[candidate.state] + return { + "candidate": _dump(candidate), + "ticket_id": ticket_id, + "status": status, + "parent_candidate_key": None, + "child_candidate_keys": [], + "depends_on": list(candidate.depends_on), + "checkpoints": [], + "terminal_receipt": _dump(receipt) if receipt is not None else None, + "split": None, + } + + def materialize( + self, + compiled_plan: dict[str, Any], + *, + terminal_receipts: list[dict[str, Any]] | tuple[dict[str, Any], ...] = (), + ) -> dict[str, Any]: + """Create or recover the exact ticket DAG and return its durable state.""" + + plan = CompiledWorkPlanV1.model_validate(compiled_plan) + receipts = self._receipt_map(plan, terminal_receipts) + with self.store.mutation_lock(): + current = self._read_unlocked(plan.plan_id) + if current is not None: + return self._reconcile_unlocked(current, plan, receipts) + + if any( + candidate.state == "terminal" and candidate.candidate_key not in receipts + for candidate in plan.candidates + ): + _fail("delivery_plan_terminal_receipt_missing") + + existing = self._existing_ticket_bindings_unlocked(plan) + missing = [c for c in plan.candidates if c.candidate_key not in existing] + reserved = iter(self.store._reserve_ids_unlocked(len(missing))) + ticket_ids = { + candidate.candidate_key: str(existing[candidate.candidate_key]["id"]) + if candidate.candidate_key in existing + else next(reserved) + for candidate in plan.candidates + } + created = [ + self._ticket_for( + candidate, + ticket_ids[candidate.candidate_key], + [ticket_ids[key] for key in candidate.depends_on], + receipts.get(candidate.candidate_key), + plan, + ) + for candidate in missing + ] + if created: + self.store._create_tickets_bulk_unlocked(created) + timestamp = _now() + state = { + "schema": DELIVERY_PLAN_STATE_SCHEMA, + "revision": 1, + "plan_id": plan.plan_id, + "plan_ref": plan.plan_ref, + "plan_hash": plan.plan_hash, + "repository": plan.repository, + "accepted_base_sha": plan.accepted_base_sha, + "target_branch": plan.target_branch, + "placement": plan.placement.model_dump(mode="json"), + "created_at": timestamp, + "updated_at": timestamp, + "candidates": { + candidate.candidate_key: self._node( + candidate, + ticket_ids[candidate.candidate_key], + receipts.get(candidate.candidate_key), + ) + for candidate in plan.candidates + }, + } + self._write_unlocked(state) + return copy.deepcopy(state) + + def _reconcile_unlocked( + self, + current: dict[str, Any], + plan: CompiledWorkPlanV1, + receipts: dict[str, WorkPlanTerminalReceiptV1], + ) -> dict[str, Any]: + identity = ( + current["plan_hash"], + current["plan_ref"], + current["repository"], + current["accepted_base_sha"], + current["target_branch"], + ) + expected = ( + plan.plan_hash, + plan.plan_ref, + plan.repository, + plan.accepted_base_sha, + plan.target_branch, + ) + if identity != expected: + _fail("delivery_plan_identity_conflict") + candidates = {candidate.candidate_key: candidate for candidate in plan.candidates} + if set(candidates) != set(current["candidates"]): + _fail("delivery_plan_candidate_set_conflict") + changed = False + for key, candidate in candidates.items(): + node = current["candidates"][key] + stored = TicketCandidateV1.model_validate(node["candidate"]) + if stored.candidate_digest != candidate.candidate_digest: + _fail("delivery_plan_candidate_conflict") + receipt = receipts.get(key) + if receipt is None: + if candidate.state == "terminal": + stored_receipt = node.get("terminal_receipt") + if stored_receipt is None: + _fail("delivery_plan_terminal_receipt_missing") + parsed_receipt = WorkPlanTerminalReceiptV1.model_validate(stored_receipt) + if parsed_receipt.receipt_ref != candidate.terminal_receipt_ref: + _fail("delivery_plan_terminal_receipt_conflict") + continue + if node["status"] in {"split", "split_required"}: + _fail("delivery_plan_receipt_for_split_candidate") + previous = node.get("terminal_receipt") + if previous is not None: + if WorkPlanTerminalReceiptV1.model_validate(previous) != receipt: + _fail("delivery_plan_receipt_conflict") + continue + if any( + other.get("terminal_receipt", {}).get("receipt_ref") == receipt.receipt_ref + for other in current["candidates"].values() + if isinstance(other.get("terminal_receipt"), dict) + ): + _fail("delivery_plan_receipt_ref_conflict") + self._complete_ticket_unlocked(str(node["ticket_id"]), receipt) + node["terminal_receipt"] = _dump(receipt) + node["status"] = "terminal" + changed = True + if changed: + current["revision"] += 1 + current["updated_at"] = _now() + self._write_unlocked(current) + return copy.deepcopy(current) + + def _complete_ticket_unlocked( + self, + ticket_id: str, + receipt: WorkPlanTerminalReceiptV1, + ) -> None: + ticket = self.store.get_ticket(ticket_id) + if ticket is None: + _fail("delivery_plan_ticket_missing") + existing = ticket.outputs.completion_receipt if ticket.outputs else None + serialized = _dump(receipt) + if str(ticket.status.value) == "done": + if existing != serialized: + _fail("delivery_plan_ticket_terminal_conflict") + return + outputs_data = ticket.outputs.model_dump(mode="python") if ticket.outputs else {} + outputs_data["completion_receipt"] = serialized + updated = self.store._update_ticket_unlocked( + ticket_id, + status="done", + execution=TicketExecution(state="done"), + outputs=TicketOutputs(**outputs_data), + reason="delivery_plan_terminal_receipt", + actor="planfile.delivery-plan", + ) + if updated is None: + _fail("delivery_plan_ticket_missing") + + def get(self, plan_id: str) -> dict[str, Any] | None: + state = self._read_unlocked(plan_id) + return copy.deepcopy(state) if state is not None else None + + def record_checkpoint(self, checkpoint_value: dict[str, Any]) -> dict[str, Any]: + checkpoint = DeliveryCheckpointV1.model_validate(checkpoint_value) + with self.store.mutation_lock(): + state = self._read_unlocked(checkpoint.plan_id) + if state is None: + _fail("delivery_plan_not_found") + node = state["candidates"].get(checkpoint.candidate_key) + if node is None: + _fail("delivery_plan_checkpoint_candidate_unknown") + candidate = TicketCandidateV1.model_validate(node["candidate"]) + if ( + checkpoint.plan_hash != state["plan_hash"] + or checkpoint.candidate_digest != candidate.candidate_digest + ): + _fail("delivery_plan_checkpoint_stale") + if node["status"] not in {"materialized", "checkpointed"}: + _fail("delivery_plan_checkpoint_state_invalid") + for existing in node["checkpoints"]: + parsed = DeliveryCheckpointV1.model_validate(existing) + if parsed.checkpoint_ref != checkpoint.checkpoint_ref: + continue + if parsed != checkpoint: + _fail("delivery_plan_checkpoint_conflict") + return copy.deepcopy(state) + expected_sequence = len(node["checkpoints"]) + 1 + if checkpoint.sequence != expected_sequence: + _fail("delivery_plan_checkpoint_sequence_invalid") + node["checkpoints"].append(_dump(checkpoint)) + node["status"] = "checkpointed" + state["revision"] += 1 + state["updated_at"] = _now() + self._write_unlocked(state) + return copy.deepcopy(state) + + def record_terminal_receipt(self, plan_id: str, receipt_value: dict[str, Any]) -> dict[str, Any]: + receipt = WorkPlanTerminalReceiptV1.model_validate(receipt_value) + with self.store.mutation_lock(): + state = self._read_unlocked(plan_id) + if state is None: + _fail("delivery_plan_not_found") + node = state["candidates"].get(receipt.candidate_key) + if node is None: + _fail("delivery_plan_receipt_candidate_unknown") + candidate = TicketCandidateV1.model_validate(node["candidate"]) + if ( + receipt.plan_hash != state["plan_hash"] + or receipt.accepted_base_sha != state["accepted_base_sha"] + or receipt.candidate_digest != candidate.candidate_digest + ): + _fail("delivery_plan_receipt_stale") + if node["status"] in {"split", "split_required"}: + _fail("delivery_plan_receipt_for_split_candidate") + previous = node.get("terminal_receipt") + if previous is not None: + if WorkPlanTerminalReceiptV1.model_validate(previous) != receipt: + _fail("delivery_plan_receipt_conflict") + return copy.deepcopy(state) + if any( + other.get("terminal_receipt", {}).get("receipt_ref") == receipt.receipt_ref + for other in state["candidates"].values() + if isinstance(other.get("terminal_receipt"), dict) + ): + _fail("delivery_plan_receipt_ref_conflict") + self._complete_ticket_unlocked(str(node["ticket_id"]), receipt) + node["terminal_receipt"] = _dump(receipt) + node["status"] = "terminal" + state["revision"] += 1 + state["updated_at"] = _now() + self._write_unlocked(state) + return copy.deepcopy(state) + + def record_split(self, split_value: dict[str, Any]) -> dict[str, Any]: + """Link an oversized candidate to already materialized bounded children.""" + + split = DeliveryPlanSplitV1.model_validate(split_value) + with self.store.mutation_lock(): + state = self._read_unlocked(split.plan_id) + if state is None: + _fail("delivery_plan_not_found") + parent = state["candidates"].get(split.parent_candidate_key) + if parent is None: + _fail("delivery_plan_split_parent_unknown") + parent_candidate = TicketCandidateV1.model_validate(parent["candidate"]) + if ( + split.plan_hash != state["plan_hash"] + or split.parent_candidate_digest != parent_candidate.candidate_digest + ): + _fail("delivery_plan_split_stale") + if parent.get("split") is not None: + if DeliveryPlanSplitV1.model_validate(parent["split"]) != split: + _fail("delivery_plan_split_conflict") + return copy.deepcopy(state) + if parent["status"] != "split_required": + _fail("delivery_plan_split_state_invalid") + children = [state["candidates"].get(key) for key in split.child_candidate_keys] + if any(child is None for child in children): + _fail("delivery_plan_split_child_unknown") + for child in children: + assert child is not None + if child["status"] == "split_required" or child["parent_candidate_key"] not in { + None, + split.parent_candidate_key, + }: + _fail("delivery_plan_split_child_invalid") + if split.parent_candidate_key in child["depends_on"]: + _fail("delivery_plan_split_dependency_cycle") + child["parent_candidate_key"] = split.parent_candidate_key + + parent["status"] = "split" + parent["child_candidate_keys"] = list(split.child_candidate_keys) + parent["split"] = _dump(split) + for node in state["candidates"].values(): + if split.parent_candidate_key not in node["depends_on"]: + continue + node["depends_on"] = sorted( + { + *( + dependency + for dependency in node["depends_on"] + if dependency != split.parent_candidate_key + ), + *split.child_candidate_keys, + } + ) + self._validate_state(state) + child_ids = [str(child["ticket_id"]) for child in children if child is not None] + parent_ticket = self.store.get_ticket(str(parent["ticket_id"])) + if parent_ticket is None: + _fail("delivery_plan_ticket_missing") + updated_parent = self.store._update_ticket_unlocked( + str(parent["ticket_id"]), + children=child_ids, + blocked_by=child_ids, + execution=TicketExecution(state="waiting_input"), + reason="delivery_plan_candidate_split", + actor="planfile.delivery-plan", + ) + if updated_parent is None: + _fail("delivery_plan_ticket_missing") + for child in children: + assert child is not None + updated_child = self.store._update_ticket_unlocked( + str(child["ticket_id"]), + parent=str(parent["ticket_id"]), + reason="delivery_plan_candidate_split_child", + actor="planfile.delivery-plan", + ) + if updated_child is None: + _fail("delivery_plan_ticket_missing") + for node in state["candidates"].values(): + if split.parent_candidate_key not in node["candidate"]["depends_on"]: + continue + dependency_ids = [ + str(state["candidates"][dependency]["ticket_id"]) + for dependency in node["depends_on"] + ] + updated = self.store._update_ticket_unlocked( + str(node["ticket_id"]), + blocked_by=dependency_ids, + reason="delivery_plan_split_dependency_rewire", + actor="planfile.delivery-plan", + ) + if updated is None: + _fail("delivery_plan_ticket_missing") + state["revision"] += 1 + state["updated_at"] = _now() + self._write_unlocked(state) + return copy.deepcopy(state) + + def resume(self, plan_id: str) -> dict[str, Any]: + """Return a deterministic continuation frontier from persisted DSL state.""" + + state = self._read_unlocked(plan_id) + if state is None: + _fail("delivery_plan_not_found") + candidates = state["candidates"] + + def completed(key: str, seen: frozenset[str] = frozenset()) -> bool: + if key in seen: + _fail("delivery_plan_state_cycle_invalid") + node = candidates[key] + if node["status"] == "terminal": + return True + if node["status"] == "split": + return all(completed(child, seen | {key}) for child in node["child_candidate_keys"]) + return False + + ready: list[dict[str, Any]] = [] + waiting: list[dict[str, Any]] = [] + split_required: list[str] = [] + split_active: list[str] = [] + split_complete: list[str] = [] + terminal: list[str] = [] + for key, node in sorted( + candidates.items(), key=lambda item: int(item[1]["candidate"]["order"]) + ): + if node["status"] == "terminal": + terminal.append(key) + continue + if node["status"] == "split_required": + split_required.append(key) + continue + if node["status"] == "split": + (split_complete if completed(key) else split_active).append(key) + continue + checkpoint = node["checkpoints"][-1] if node["checkpoints"] else None + item = { + "candidate_key": key, + "ticket_id": node["ticket_id"], + "checkpoint": checkpoint, + "waiting_on": [ + dependency for dependency in node["depends_on"] if not completed(dependency) + ], + } + (waiting if item["waiting_on"] else ready).append(item) + return { + "schema": DELIVERY_PLAN_RESUME_SCHEMA, + "plan_id": state["plan_id"], + "plan_hash": state["plan_hash"], + "revision": state["revision"], + "ready": ready, + "waiting": waiting, + "split_required": split_required, + "split_active": split_active, + "split_complete": split_complete, + "terminal": sorted(terminal), + "execution": "inert", + "authority": "none", + } + + +__all__ = ["DeliveryPlanError", "DeliveryPlanRepository"] diff --git a/planfile/delivery_plan_contracts.py b/planfile/delivery_plan_contracts.py new file mode 100644 index 0000000..2760eb2 --- /dev/null +++ b/planfile/delivery_plan_contracts.py @@ -0,0 +1,429 @@ +"""Closed contracts for resumable Strategy delivery plans. + +The Strategy compiler is deliberately inert. These models preserve that +boundary while giving Planfile enough structured data to materialize tickets, +record explicit checkpoints and resume after an interrupted process. +""" + +from __future__ import annotations + +import hashlib +import json +import re +from datetime import datetime +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + +COMPILED_WORK_PLAN_SCHEMA = "subactor.compiled-work-plan/v1" +TICKET_CANDIDATE_SCHEMA = "subactor.ticket-candidate/v1" +TERMINAL_RECEIPT_SCHEMA = "subactor.work-plan-terminal-receipt/v1" +DELIVERY_CHECKPOINT_SCHEMA = "planfile.delivery-checkpoint/v1" +DELIVERY_SPLIT_SCHEMA = "planfile.delivery-plan-split/v1" +DELIVERY_PLAN_STATE_SCHEMA = "planfile.delivery-plan-state/v1" +DELIVERY_PLAN_RESUME_SCHEMA = "planfile.delivery-plan-resume/v1" + +_ID = re.compile(r"^[a-z][a-z0-9-]{1,63}$") +_REPOSITORY = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$") +_SHA256 = re.compile(r"^sha256:[a-f0-9]{64}$") +_GIT_SHA = re.compile(r"^[a-f0-9]{40}$") +_REFERENCE = re.compile(r"^(?:artifact|knowledge)://[^\s?#]+$") +_RECEIPT_REFERENCE = re.compile(r"^receipt://[^\s?#]+$") +_CHECKPOINT_REFERENCE = re.compile(r"^checkpoint://[^\s?#]+$") +_SAFE_PATH = re.compile( + r"^(?!/)(?!.*(?:^|/)\.\.(?:/|$))[A-Za-z0-9._-]+(?:/[A-Za-z0-9._-]+)*$" +) +_BRANCH = re.compile(r"^(?!/)(?!.*(?:\.\.|//|@\{|[~^:?*\[\\]))[A-Za-z0-9._/-]+$") + + +def _stable_json(value: object) -> str: + return json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + + +def _digest(value: object) -> str: + return f"sha256:{hashlib.sha256(_stable_json(value).encode('utf-8')).hexdigest()}" + + +def _tuple_of_strings(value: object) -> tuple[str, ...]: + if not isinstance(value, (list, tuple)): + raise TypeError("expected a list or tuple") + return tuple(str(item) for item in value) + + +def _timestamp(value: object) -> datetime: + if isinstance(value, str): + try: + value = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError("delivery_plan_timestamp_invalid") from exc + if not isinstance(value, datetime) or value.tzinfo is None: + raise ValueError("delivery_plan_timestamp_invalid") + return value + + +class _ClosedContract(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, populate_by_name=True, strict=True) + + +class DeliveryPlacementV1(_ClosedContract): + home: Literal["wellmanifest", "subactor", "semcod"] + shape: Literal["domain_pack", "runtime_service", "both"] + runtime_owner: Literal["wellmanifest", "subactor", "semcod"] + adopt: tuple[str, ...] + + @field_validator("adopt", mode="before") + @classmethod + def _normalize_adopt(cls, value: object) -> tuple[str, ...]: + result = _tuple_of_strings(value) + if not result or len(set(result)) != len(result): + raise ValueError("delivery_plan_adoption_invalid") + if any(not re.fullmatch(r"wellmanifest/[a-z0-9][a-z0-9-]{1,79}", item) for item in result): + raise ValueError("delivery_plan_adoption_invalid") + return tuple(sorted(result)) + + @model_validator(mode="after") + def _runtime_home(self) -> DeliveryPlacementV1: + if self.shape == "runtime_service" and ( + self.home == "wellmanifest" or self.runtime_owner == "wellmanifest" + ): + raise ValueError("delivery_plan_runtime_home_invalid") + return self + + +class DeliveryBudgetV1(_ClosedContract): + complexity: Literal["XS", "S", "M", "L"] + estimated_minutes: int = Field(ge=1, le=240) + max_implementation_files: int = Field(ge=1, le=30) + max_affected_components: int = Field(ge=1, le=10) + max_public_interface_changes: int = Field(ge=0, le=10) + max_runtime_dependencies: int = Field(ge=0, le=10) + + +class DeliveryTestBindingV1(_ClosedContract): + id: str = Field(pattern=r"^[a-z][a-z0-9-]{1,63}$") + kind: Literal["docker", "governance", "node-test", "python-test"] + target: str + + @field_validator("target") + @classmethod + def _safe_target(cls, value: str) -> str: + if not _SAFE_PATH.fullmatch(value): + raise ValueError("delivery_plan_test_target_invalid") + return value + + +class DeliveryAcceptanceV1(_ClosedContract): + id: str = Field(pattern=r"^AC-[0-9]{2}$") + statement: str = Field(min_length=1) + test_ids: tuple[str, ...] + + @field_validator("test_ids", mode="before") + @classmethod + def _normalize_test_ids(cls, value: object) -> tuple[str, ...]: + result = _tuple_of_strings(value) + if not result or len(set(result)) != len(result) or any(not _ID.fullmatch(v) for v in result): + raise ValueError("delivery_plan_acceptance_tests_invalid") + return tuple(sorted(result)) + + +class TicketCandidateV1(_ClosedContract): + schema_id: Literal["subactor.ticket-candidate/v1"] = Field( + default="subactor.ticket-candidate/v1", + alias="schema", + ) + plan_hash: str = Field(pattern=r"^sha256:[a-f0-9]{64}$") + candidate_key: str = Field(pattern=r"^[a-z][a-z0-9-]{1,63}$") + order: int = Field(ge=0, le=1023) + title: str = Field(min_length=1) + workstream: str = Field(pattern=r"^[a-z][a-z0-9-]{1,63}$") + depends_on: tuple[str, ...] + allowed_paths: tuple[str, ...] + placement: DeliveryPlacementV1 + delivery: DeliveryBudgetV1 + components: tuple[str, ...] + public_interfaces: tuple[str, ...] + runtime_dependencies: tuple[str, ...] + acceptance: tuple[DeliveryAcceptanceV1, ...] + tests: tuple[DeliveryTestBindingV1, ...] + execution: Literal["inert"] + candidate_digest: str = Field(pattern=r"^sha256:[a-f0-9]{64}$") + idempotency_key: str = Field(pattern=r"^sha256:[a-f0-9]{64}$") + state: Literal["pending", "split_required", "terminal"] + split_reasons: tuple[str, ...] + terminal_receipt_ref: str | None + + @field_validator( + "depends_on", + "allowed_paths", + "components", + "public_interfaces", + "runtime_dependencies", + "split_reasons", + mode="before", + ) + @classmethod + def _normalize_string_tuple(cls, value: object) -> tuple[str, ...]: + result = _tuple_of_strings(value) + if len(set(result)) != len(result): + raise ValueError("delivery_plan_candidate_set_duplicate") + return tuple(sorted(result)) + + @field_validator("acceptance", "tests", mode="before") + @classmethod + def _normalize_model_tuple(cls, value: object) -> tuple[object, ...]: + if not isinstance(value, (list, tuple)): + raise TypeError("expected a list or tuple") + return tuple(value) + + @model_validator(mode="after") + def _validate_candidate(self) -> TicketCandidateV1: + if not self.allowed_paths or any(not _SAFE_PATH.fullmatch(path) for path in self.allowed_paths): + raise ValueError("delivery_plan_candidate_paths_invalid") + if not self.components or not self.acceptance or not self.tests: + raise ValueError("delivery_plan_candidate_evidence_invalid") + test_ids = {binding.id for binding in self.tests} + if len(test_ids) != len(self.tests): + raise ValueError("delivery_plan_candidate_test_duplicate") + if len({criterion.id for criterion in self.acceptance}) != len(self.acceptance): + raise ValueError("delivery_plan_candidate_acceptance_duplicate") + if any(not set(criterion.test_ids).issubset(test_ids) for criterion in self.acceptance): + raise ValueError("delivery_plan_candidate_acceptance_test_unknown") + if self.state == "terminal": + if not self.terminal_receipt_ref or not _RECEIPT_REFERENCE.fullmatch( + self.terminal_receipt_ref + ): + raise ValueError("delivery_plan_candidate_terminal_receipt_invalid") + elif self.terminal_receipt_ref is not None: + raise ValueError("delivery_plan_candidate_terminal_receipt_unexpected") + if (self.state == "split_required") != bool(self.split_reasons): + raise ValueError("delivery_plan_candidate_split_state_invalid") + + core = self.model_dump( + mode="json", + by_alias=True, + include={ + "schema_id", + "plan_hash", + "candidate_key", + "order", + "title", + "workstream", + "depends_on", + "allowed_paths", + "placement", + "delivery", + "components", + "public_interfaces", + "runtime_dependencies", + "acceptance", + "tests", + "execution", + }, + ) + if _digest(core) != self.candidate_digest: + raise ValueError("delivery_plan_candidate_digest_mismatch") + expected_idempotency = _digest( + { + "plan_hash": self.plan_hash, + "candidate_key": self.candidate_key, + "candidate_digest": self.candidate_digest, + } + ) + if expected_idempotency != self.idempotency_key: + raise ValueError("delivery_plan_candidate_idempotency_mismatch") + return self + + +class CompiledWorkPlanV1(_ClosedContract): + schema_id: Literal["subactor.compiled-work-plan/v1"] = Field( + default="subactor.compiled-work-plan/v1", + alias="schema", + ) + status: Literal["ready", "split_required"] + execution: Literal["inert"] + authority: Literal["none"] + plan_id: str = Field(pattern=r"^[a-z][a-z0-9-]{1,63}$") + plan_ref: str + plan_hash: str = Field(pattern=r"^sha256:[a-f0-9]{64}$") + repository: str + accepted_base_sha: str = Field(pattern=r"^[a-f0-9]{40}$") + target_branch: str + placement: DeliveryPlacementV1 + candidate_count: int = Field(ge=1, le=12) + pending_count: int = Field(ge=0, le=12) + split_required: tuple[str, ...] + terminal_candidates: tuple[str, ...] + candidates: tuple[TicketCandidateV1, ...] + + @field_validator("split_required", "terminal_candidates", mode="before") + @classmethod + def _normalize_key_tuple(cls, value: object) -> tuple[str, ...]: + result = _tuple_of_strings(value) + if len(set(result)) != len(result) or any(not _ID.fullmatch(item) for item in result): + raise ValueError("delivery_plan_candidate_keys_invalid") + return tuple(sorted(result)) + + @field_validator("candidates", mode="before") + @classmethod + def _normalize_candidates(cls, value: object) -> tuple[object, ...]: + if not isinstance(value, (list, tuple)): + raise TypeError("expected a list or tuple") + return tuple(value) + + @field_validator("plan_ref") + @classmethod + def _plan_reference(cls, value: str) -> str: + if not _REFERENCE.fullmatch(value): + raise ValueError("delivery_plan_ref_invalid") + return value + + @field_validator("repository") + @classmethod + def _repository(cls, value: str) -> str: + if not _REPOSITORY.fullmatch(value): + raise ValueError("delivery_plan_repository_invalid") + return value + + @field_validator("target_branch") + @classmethod + def _branch(cls, value: str) -> str: + if not _BRANCH.fullmatch(value): + raise ValueError("delivery_plan_branch_invalid") + return value + + @model_validator(mode="after") + def _validate_graph(self) -> CompiledWorkPlanV1: + if len(self.candidates) != self.candidate_count: + raise ValueError("delivery_plan_candidate_count_mismatch") + keys = [candidate.candidate_key for candidate in self.candidates] + if len(set(keys)) != len(keys): + raise ValueError("delivery_plan_candidate_key_duplicate") + if [candidate.order for candidate in self.candidates] != list(range(len(self.candidates))): + raise ValueError("delivery_plan_candidate_order_invalid") + order = {candidate.candidate_key: candidate.order for candidate in self.candidates} + for candidate in self.candidates: + if candidate.plan_hash != self.plan_hash or candidate.placement != self.placement: + raise ValueError("delivery_plan_candidate_binding_mismatch") + if any(dependency not in order for dependency in candidate.depends_on): + raise ValueError("delivery_plan_dependency_unknown") + if any(order[dependency] >= candidate.order for dependency in candidate.depends_on): + raise ValueError("delivery_plan_dependency_order_invalid") + split = tuple(sorted(c.candidate_key for c in self.candidates if c.state == "split_required")) + terminal = tuple(sorted(c.candidate_key for c in self.candidates if c.state == "terminal")) + if split != self.split_required or terminal != self.terminal_candidates: + raise ValueError("delivery_plan_candidate_state_projection_mismatch") + if self.pending_count != len(self.candidates) - len(split) - len(terminal): + raise ValueError("delivery_plan_pending_count_mismatch") + if (self.status == "split_required") != bool(split): + raise ValueError("delivery_plan_status_mismatch") + return self + + +class WorkPlanTerminalReceiptV1(_ClosedContract): + schema_id: Literal["subactor.work-plan-terminal-receipt/v1"] = Field( + default="subactor.work-plan-terminal-receipt/v1", + alias="schema", + ) + receipt_ref: str + plan_hash: str = Field(pattern=r"^sha256:[a-f0-9]{64}$") + candidate_key: str = Field(pattern=r"^[a-z][a-z0-9-]{1,63}$") + candidate_digest: str = Field(pattern=r"^sha256:[a-f0-9]{64}$") + accepted_base_sha: str = Field(pattern=r"^[a-f0-9]{40}$") + outcome: Literal["merged", "no-change"] + terminal_sha: str = Field(pattern=r"^[a-f0-9]{40}$") + + @field_validator("receipt_ref") + @classmethod + def _receipt_reference(cls, value: str) -> str: + if not _RECEIPT_REFERENCE.fullmatch(value): + raise ValueError("delivery_plan_receipt_ref_invalid") + return value + + +class DeliveryCheckpointV1(_ClosedContract): + schema_id: Literal["planfile.delivery-checkpoint/v1"] = Field( + default="planfile.delivery-checkpoint/v1", + alias="schema", + ) + checkpoint_ref: str + plan_id: str = Field(pattern=r"^[a-z][a-z0-9-]{1,63}$") + plan_hash: str = Field(pattern=r"^sha256:[a-f0-9]{64}$") + candidate_key: str = Field(pattern=r"^[a-z][a-z0-9-]{1,63}$") + candidate_digest: str = Field(pattern=r"^sha256:[a-f0-9]{64}$") + sequence: int = Field(ge=1) + phase: Literal["materialized", "development", "testing", "review", "waiting"] + head_sha: str | None = Field(default=None, pattern=r"^[a-f0-9]{40}$") + evidence_refs: tuple[str, ...] = () + recorded_at: datetime + + @field_validator("recorded_at", mode="before") + @classmethod + def _recorded_timestamp(cls, value: object) -> datetime: + return _timestamp(value) + + @field_validator("checkpoint_ref") + @classmethod + def _checkpoint_reference(cls, value: str) -> str: + if not _CHECKPOINT_REFERENCE.fullmatch(value): + raise ValueError("delivery_plan_checkpoint_ref_invalid") + return value + + @field_validator("evidence_refs", mode="before") + @classmethod + def _evidence_references(cls, value: object) -> tuple[str, ...]: + result = _tuple_of_strings(value) + if len(set(result)) != len(result) or any(not _REFERENCE.fullmatch(item) for item in result): + raise ValueError("delivery_plan_checkpoint_evidence_invalid") + return tuple(sorted(result)) + + +class DeliveryPlanSplitV1(_ClosedContract): + schema_id: Literal["planfile.delivery-plan-split/v1"] = Field( + default="planfile.delivery-plan-split/v1", + alias="schema", + ) + split_ref: str + plan_id: str = Field(pattern=r"^[a-z][a-z0-9-]{1,63}$") + plan_hash: str = Field(pattern=r"^sha256:[a-f0-9]{64}$") + parent_candidate_key: str = Field(pattern=r"^[a-z][a-z0-9-]{1,63}$") + parent_candidate_digest: str = Field(pattern=r"^sha256:[a-f0-9]{64}$") + child_candidate_keys: tuple[str, ...] + recorded_at: datetime + + @field_validator("recorded_at", mode="before") + @classmethod + def _recorded_timestamp(cls, value: object) -> datetime: + return _timestamp(value) + + @field_validator("split_ref") + @classmethod + def _split_reference(cls, value: str) -> str: + if not value.startswith("split://") or any(character.isspace() for character in value): + raise ValueError("delivery_plan_split_ref_invalid") + return value + + @field_validator("child_candidate_keys", mode="before") + @classmethod + def _children(cls, value: object) -> tuple[str, ...]: + result = _tuple_of_strings(value) + if len(result) < 2 or len(set(result)) != len(result) or any( + not _ID.fullmatch(item) for item in result + ): + raise ValueError("delivery_plan_split_children_invalid") + return tuple(sorted(result)) + + +__all__ = [ + "COMPILED_WORK_PLAN_SCHEMA", + "DELIVERY_CHECKPOINT_SCHEMA", + "DELIVERY_PLAN_RESUME_SCHEMA", + "DELIVERY_PLAN_STATE_SCHEMA", + "DELIVERY_SPLIT_SCHEMA", + "TERMINAL_RECEIPT_SCHEMA", + "TICKET_CANDIDATE_SCHEMA", + "CompiledWorkPlanV1", + "DeliveryCheckpointV1", + "DeliveryPlanSplitV1", + "TicketCandidateV1", + "WorkPlanTerminalReceiptV1", +] diff --git a/planfile/schemas/delivery-plan-state.schema.v1.json b/planfile/schemas/delivery-plan-state.schema.v1.json new file mode 100644 index 0000000..43bd8dd --- /dev/null +++ b/planfile/schemas/delivery-plan-state.schema.v1.json @@ -0,0 +1,224 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://semcod.org/planfile/schemas/delivery-plan-state/v1", + "title": "Planfile resumable delivery-plan state", + "type": "object", + "additionalProperties": false, + "required": [ + "schema", + "revision", + "plan_id", + "plan_ref", + "plan_hash", + "repository", + "accepted_base_sha", + "target_branch", + "placement", + "created_at", + "updated_at", + "candidates" + ], + "properties": { + "schema": {"const": "planfile.delivery-plan-state/v1"}, + "revision": {"type": "integer", "minimum": 1}, + "plan_id": {"$ref": "#/$defs/id"}, + "plan_ref": {"type": "string", "pattern": "^(artifact|knowledge)://[^\\s?#]+$"}, + "plan_hash": {"$ref": "#/$defs/sha256"}, + "repository": {"type": "string", "pattern": "^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+$"}, + "accepted_base_sha": {"$ref": "#/$defs/gitSha"}, + "target_branch": {"type": "string", "minLength": 1}, + "placement": {"$ref": "#/$defs/placement"}, + "created_at": {"type": "string", "format": "date-time"}, + "updated_at": {"type": "string", "format": "date-time"}, + "candidates": { + "type": "object", + "minProperties": 1, + "propertyNames": {"$ref": "#/$defs/id"}, + "additionalProperties": {"$ref": "#/$defs/stateCandidate"} + } + }, + "$defs": { + "id": {"type": "string", "pattern": "^[a-z][a-z0-9-]{1,63}$"}, + "sha256": {"type": "string", "pattern": "^sha256:[a-f0-9]{64}$"}, + "gitSha": {"type": "string", "pattern": "^[a-f0-9]{40}$"}, + "safePath": { + "type": "string", + "pattern": "^(?!/)(?!.*(^|/)\\.\\.(/|$))[A-Za-z0-9._-]+(/[A-Za-z0-9._-]+)*$" + }, + "placement": { + "type": "object", + "additionalProperties": false, + "required": ["home", "shape", "runtime_owner", "adopt"], + "properties": { + "home": {"enum": ["wellmanifest", "subactor", "semcod"]}, + "shape": {"enum": ["domain_pack", "runtime_service", "both"]}, + "runtime_owner": {"enum": ["wellmanifest", "subactor", "semcod"]}, + "adopt": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": {"type": "string", "pattern": "^wellmanifest/[a-z0-9][a-z0-9-]{1,79}$"} + } + } + }, + "delivery": { + "type": "object", + "additionalProperties": false, + "required": [ + "complexity", + "estimated_minutes", + "max_implementation_files", + "max_affected_components", + "max_public_interface_changes", + "max_runtime_dependencies" + ], + "properties": { + "complexity": {"enum": ["XS", "S", "M", "L"]}, + "estimated_minutes": {"type": "integer", "minimum": 1, "maximum": 240}, + "max_implementation_files": {"type": "integer", "minimum": 1, "maximum": 30}, + "max_affected_components": {"type": "integer", "minimum": 1, "maximum": 10}, + "max_public_interface_changes": {"type": "integer", "minimum": 0, "maximum": 10}, + "max_runtime_dependencies": {"type": "integer", "minimum": 0, "maximum": 10} + } + }, + "acceptance": { + "type": "object", + "additionalProperties": false, + "required": ["id", "statement", "test_ids"], + "properties": { + "id": {"type": "string", "pattern": "^AC-[0-9]{2}$"}, + "statement": {"type": "string", "minLength": 1}, + "test_ids": { + "type": "array", + "minItems": 1, + "uniqueItems": true, + "items": {"$ref": "#/$defs/id"} + } + } + }, + "testBinding": { + "type": "object", + "additionalProperties": false, + "required": ["id", "kind", "target"], + "properties": { + "id": {"$ref": "#/$defs/id"}, + "kind": {"enum": ["docker", "governance", "node-test", "python-test"]}, + "target": {"$ref": "#/$defs/safePath"} + } + }, + "candidate": { + "type": "object", + "additionalProperties": false, + "required": [ + "schema", + "plan_hash", + "candidate_key", + "order", + "title", + "workstream", + "depends_on", + "allowed_paths", + "placement", + "delivery", + "components", + "public_interfaces", + "runtime_dependencies", + "acceptance", + "tests", + "execution", + "candidate_digest", + "idempotency_key", + "state", + "split_reasons", + "terminal_receipt_ref" + ], + "properties": { + "schema": {"const": "subactor.ticket-candidate/v1"}, + "plan_hash": {"$ref": "#/$defs/sha256"}, + "candidate_key": {"$ref": "#/$defs/id"}, + "order": {"type": "integer", "minimum": 0}, + "title": {"type": "string", "minLength": 1}, + "workstream": {"$ref": "#/$defs/id"}, + "depends_on": {"type": "array", "uniqueItems": true, "items": {"$ref": "#/$defs/id"}}, + "allowed_paths": {"type": "array", "minItems": 1, "uniqueItems": true, "items": {"$ref": "#/$defs/safePath"}}, + "placement": {"$ref": "#/$defs/placement"}, + "delivery": {"$ref": "#/$defs/delivery"}, + "components": {"type": "array", "minItems": 1, "uniqueItems": true, "items": {"type": "string", "minLength": 1}}, + "public_interfaces": {"type": "array", "uniqueItems": true, "items": {"type": "string", "minLength": 1}}, + "runtime_dependencies": {"type": "array", "uniqueItems": true, "items": {"type": "string", "minLength": 1}}, + "acceptance": {"type": "array", "minItems": 1, "items": {"$ref": "#/$defs/acceptance"}}, + "tests": {"type": "array", "minItems": 1, "items": {"$ref": "#/$defs/testBinding"}}, + "execution": {"const": "inert"}, + "candidate_digest": {"$ref": "#/$defs/sha256"}, + "idempotency_key": {"$ref": "#/$defs/sha256"}, + "state": {"enum": ["pending", "split_required", "terminal"]}, + "split_reasons": {"type": "array", "uniqueItems": true, "items": {"type": "string", "minLength": 1}}, + "terminal_receipt_ref": {"type": ["string", "null"], "pattern": "^receipt://[^\\s?#]+$"} + } + }, + "checkpoint": { + "type": "object", + "additionalProperties": false, + "required": ["schema", "checkpoint_ref", "plan_id", "plan_hash", "candidate_key", "candidate_digest", "sequence", "phase", "head_sha", "evidence_refs", "recorded_at"], + "properties": { + "schema": {"const": "planfile.delivery-checkpoint/v1"}, + "checkpoint_ref": {"type": "string", "pattern": "^checkpoint://[^\\s?#]+$"}, + "plan_id": {"$ref": "#/$defs/id"}, + "plan_hash": {"$ref": "#/$defs/sha256"}, + "candidate_key": {"$ref": "#/$defs/id"}, + "candidate_digest": {"$ref": "#/$defs/sha256"}, + "sequence": {"type": "integer", "minimum": 1}, + "phase": {"enum": ["materialized", "development", "testing", "review", "waiting"]}, + "head_sha": {"anyOf": [{"$ref": "#/$defs/gitSha"}, {"type": "null"}]}, + "evidence_refs": {"type": "array", "uniqueItems": true, "items": {"type": "string", "pattern": "^(artifact|knowledge)://[^\\s?#]+$"}}, + "recorded_at": {"type": "string", "format": "date-time"} + } + }, + "terminalReceipt": { + "type": "object", + "additionalProperties": false, + "required": ["schema", "receipt_ref", "plan_hash", "candidate_key", "candidate_digest", "accepted_base_sha", "outcome", "terminal_sha"], + "properties": { + "schema": {"const": "subactor.work-plan-terminal-receipt/v1"}, + "receipt_ref": {"type": "string", "pattern": "^receipt://[^\\s?#]+$"}, + "plan_hash": {"$ref": "#/$defs/sha256"}, + "candidate_key": {"$ref": "#/$defs/id"}, + "candidate_digest": {"$ref": "#/$defs/sha256"}, + "accepted_base_sha": {"$ref": "#/$defs/gitSha"}, + "outcome": {"enum": ["merged", "no-change"]}, + "terminal_sha": {"$ref": "#/$defs/gitSha"} + } + }, + "split": { + "type": "object", + "additionalProperties": false, + "required": ["schema", "split_ref", "plan_id", "plan_hash", "parent_candidate_key", "parent_candidate_digest", "child_candidate_keys", "recorded_at"], + "properties": { + "schema": {"const": "planfile.delivery-plan-split/v1"}, + "split_ref": {"type": "string", "pattern": "^split://[^\\s]+$"}, + "plan_id": {"$ref": "#/$defs/id"}, + "plan_hash": {"$ref": "#/$defs/sha256"}, + "parent_candidate_key": {"$ref": "#/$defs/id"}, + "parent_candidate_digest": {"$ref": "#/$defs/sha256"}, + "child_candidate_keys": {"type": "array", "minItems": 2, "uniqueItems": true, "items": {"$ref": "#/$defs/id"}}, + "recorded_at": {"type": "string", "format": "date-time"} + } + }, + "stateCandidate": { + "type": "object", + "additionalProperties": false, + "required": ["candidate", "ticket_id", "status", "parent_candidate_key", "child_candidate_keys", "depends_on", "checkpoints", "terminal_receipt", "split"], + "properties": { + "candidate": {"$ref": "#/$defs/candidate"}, + "ticket_id": {"type": "string", "minLength": 1}, + "status": {"enum": ["materialized", "checkpointed", "split_required", "split", "terminal"]}, + "parent_candidate_key": {"anyOf": [{"$ref": "#/$defs/id"}, {"type": "null"}]}, + "child_candidate_keys": {"type": "array", "uniqueItems": true, "items": {"$ref": "#/$defs/id"}}, + "depends_on": {"type": "array", "uniqueItems": true, "items": {"$ref": "#/$defs/id"}}, + "checkpoints": {"type": "array", "items": {"$ref": "#/$defs/checkpoint"}}, + "terminal_receipt": {"anyOf": [{"$ref": "#/$defs/terminalReceipt"}, {"type": "null"}]}, + "split": {"anyOf": [{"$ref": "#/$defs/split"}, {"type": "null"}]} + } + } + } +} diff --git a/pyproject.toml b/pyproject.toml index 33cd172..39eb198 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -82,7 +82,7 @@ where = ["."] include = ["planfile*"] [tool.setuptools.package-data] -"planfile" = ["examples/**/*.yaml", "examples/**/*.yml"] +"planfile" = ["examples/**/*.yaml", "examples/**/*.yml", "schemas/*.json"] [tool.ruff] line-length = 100 diff --git a/tests/test_delivery_plan.py b/tests/test_delivery_plan.py new file mode 100644 index 0000000..592e002 --- /dev/null +++ b/tests/test_delivery_plan.py @@ -0,0 +1,315 @@ +from __future__ import annotations + +import hashlib +import json +from datetime import UTC, datetime + +import pytest +from pydantic import ValidationError + +from planfile import Planfile +from planfile.delivery_plan import DeliveryPlanError +from planfile.delivery_plan_contracts import CompiledWorkPlanV1 + +PLAN_HASH = "sha256:" + "a" * 64 +BASE_SHA = "b" * 40 + + +def _digest(value: object) -> str: + text = json.dumps(value, ensure_ascii=False, separators=(",", ":"), sort_keys=True) + return f"sha256:{hashlib.sha256(text.encode()).hexdigest()}" + + +def _candidate( + key: str, + order: int, + *, + depends_on: list[str] | None = None, + state: str = "pending", + paths: list[str] | None = None, +) -> dict: + core = { + "schema": "subactor.ticket-candidate/v1", + "plan_hash": PLAN_HASH, + "candidate_key": key, + "order": order, + "title": f"Deliver {key}", + "workstream": "runtime", + "depends_on": sorted(depends_on or []), + "allowed_paths": sorted(paths or [f"src/{key}.py"]), + "placement": { + "home": "semcod", + "shape": "runtime_service", + "runtime_owner": "semcod", + "adopt": ["wellmanifest/dsl", "wellmanifest/new-project"], + }, + "delivery": { + "complexity": "S", + "estimated_minutes": 20, + "max_implementation_files": 2, + "max_affected_components": 1, + "max_public_interface_changes": 1, + "max_runtime_dependencies": 0, + }, + "components": ["delivery-plan"], + "public_interfaces": ["Planfile.materialize_delivery_plan"], + "runtime_dependencies": [], + "acceptance": [ + { + "id": "AC-01", + "statement": "The bounded slice has explicit verification.", + "test_ids": ["python-suite"], + } + ], + "tests": [ + {"id": "python-suite", "kind": "python-test", "target": "tests/test_delivery_plan.py"} + ], + "execution": "inert", + } + candidate_digest = _digest(core) + return { + **core, + "candidate_digest": candidate_digest, + "idempotency_key": _digest( + { + "plan_hash": PLAN_HASH, + "candidate_key": key, + "candidate_digest": candidate_digest, + } + ), + "state": state, + "split_reasons": ["complexity_time_limit"] if state == "split_required" else [], + "terminal_receipt_ref": None, + } + + +def _plan(candidates: list[dict]) -> dict: + split_required = sorted( + candidate["candidate_key"] + for candidate in candidates + if candidate["state"] == "split_required" + ) + terminal = sorted( + candidate["candidate_key"] + for candidate in candidates + if candidate["state"] == "terminal" + ) + return { + "schema": "subactor.compiled-work-plan/v1", + "status": "split_required" if split_required else "ready", + "execution": "inert", + "authority": "none", + "plan_id": "resumable-delivery", + "plan_ref": "artifact://semcod/planfile/delivery-plan-r1", + "plan_hash": PLAN_HASH, + "repository": "semcod/planfile", + "accepted_base_sha": BASE_SHA, + "target_branch": "main", + "placement": { + "home": "semcod", + "shape": "runtime_service", + "runtime_owner": "semcod", + "adopt": ["wellmanifest/dsl", "wellmanifest/new-project"], + }, + "candidate_count": len(candidates), + "pending_count": len(candidates) - len(split_required) - len(terminal), + "split_required": split_required, + "terminal_candidates": terminal, + "candidates": candidates, + } + + +def _receipt(candidate: dict, *, receipt_ref: str = "receipt://github/semcod/planfile/101") -> dict: + return { + "schema": "subactor.work-plan-terminal-receipt/v1", + "receipt_ref": receipt_ref, + "plan_hash": PLAN_HASH, + "candidate_key": candidate["candidate_key"], + "candidate_digest": candidate["candidate_digest"], + "accepted_base_sha": BASE_SHA, + "outcome": "merged", + "terminal_sha": "c" * 40, + } + + +def test_materialization_is_atomic_idempotent_and_execution_inert(tmp_path) -> None: + runtime = _candidate("runtime-core", 0) + facade = _candidate("package-facade", 1, depends_on=["runtime-core"]) + backend = Planfile(str(tmp_path)) + + first = backend.materialize_delivery_plan(_plan([runtime, facade])) + second = backend.materialize_delivery_plan(_plan([runtime, facade])) + + assert first == second + assert first["revision"] == 1 + assert len(backend.list_tickets(sprint="all")) == 2 + runtime_ticket = backend.get_ticket(first["candidates"]["runtime-core"]["ticket_id"]) + facade_ticket = backend.get_ticket(first["candidates"]["package-facade"]["ticket_id"]) + assert runtime_ticket is not None and runtime_ticket.executor is None + assert runtime_ticket.source.context["authority"] == "none" + assert facade_ticket is not None and facade_ticket.blocked_by == [runtime_ticket.id] + persisted = tmp_path / ".planfile" / "delivery-plans" / "resumable-delivery.json" + assert json.loads(persisted.read_text())["schema"] == "planfile.delivery-plan-state/v1" + + +def test_materialization_recovers_exact_ids_after_missing_state_replace(tmp_path) -> None: + candidates = [_candidate("runtime-core", 0), _candidate("package-facade", 1)] + backend = Planfile(str(tmp_path)) + first = backend.materialize_delivery_plan(_plan(candidates)) + state_path = tmp_path / ".planfile" / "delivery-plans" / "resumable-delivery.json" + state_path.unlink() + + recovered = backend.materialize_delivery_plan(_plan(candidates)) + + assert { + key: node["ticket_id"] for key, node in recovered["candidates"].items() + } == {key: node["ticket_id"] for key, node in first["candidates"].items()} + assert len(backend.list_tickets(sprint="all")) == 2 + + +def test_materialization_requires_and_deduplicates_compiled_terminal_receipt(tmp_path) -> None: + candidate = _candidate("already-delivered", 0) + receipt = _receipt(candidate) + candidate["state"] = "terminal" + candidate["terminal_receipt_ref"] = receipt["receipt_ref"] + plan = _plan([candidate]) + backend = Planfile(str(tmp_path)) + + with pytest.raises(DeliveryPlanError, match="delivery_plan_terminal_receipt_missing"): + backend.materialize_delivery_plan(plan) + state = backend.materialize_delivery_plan(plan, terminal_receipts=[receipt, receipt]) + + ticket = backend.get_ticket(state["candidates"]["already-delivered"]["ticket_id"]) + assert ticket is not None and ticket.status.value == "done" + assert ticket.outputs.completion_receipt == receipt + + +def test_checkpoint_and_terminal_receipt_resume_partial_completion(tmp_path) -> None: + runtime = _candidate("runtime-core", 0) + facade = _candidate("package-facade", 1, depends_on=["runtime-core"]) + backend = Planfile(str(tmp_path)) + state = backend.materialize_delivery_plan(_plan([runtime, facade])) + checkpoint = { + "schema": "planfile.delivery-checkpoint/v1", + "checkpoint_ref": "checkpoint://semcod/planfile/runtime-core/1", + "plan_id": state["plan_id"], + "plan_hash": state["plan_hash"], + "candidate_key": "runtime-core", + "candidate_digest": runtime["candidate_digest"], + "sequence": 1, + "phase": "testing", + "head_sha": "d" * 40, + "evidence_refs": ["artifact://semcod/planfile/test-report-r1"], + "recorded_at": datetime.now(UTC).isoformat(), + } + + checkpointed = backend.checkpoint_delivery_candidate(checkpoint) + duplicate = backend.checkpoint_delivery_candidate(checkpoint) + assert duplicate == checkpointed + resume = backend.resume_delivery_plan(state["plan_id"]) + assert resume["ready"][0]["checkpoint"]["checkpoint_ref"] == checkpoint["checkpoint_ref"] + assert resume["waiting"][0]["candidate_key"] == "package-facade" + assert resume["authority"] == "none" + + terminal = backend.record_delivery_terminal_receipt(state["plan_id"], _receipt(runtime)) + repeated = backend.record_delivery_terminal_receipt(state["plan_id"], _receipt(runtime)) + assert repeated == terminal + assert backend.materialize_delivery_plan(_plan([runtime, facade])) == terminal + resumed = backend.resume_delivery_plan(state["plan_id"]) + assert resumed["terminal"] == ["runtime-core"] + assert resumed["ready"][0]["candidate_key"] == "package-facade" + ticket = backend.get_ticket(terminal["candidates"]["runtime-core"]["ticket_id"]) + assert ticket is not None and ticket.status.value == "done" + assert ticket.outputs.completion_receipt["receipt_ref"].startswith("receipt://") + + +def test_conflicting_checkpoint_and_receipt_fail_closed(tmp_path) -> None: + candidate = _candidate("runtime-core", 0) + backend = Planfile(str(tmp_path)) + state = backend.materialize_delivery_plan(_plan([candidate])) + checkpoint = { + "schema": "planfile.delivery-checkpoint/v1", + "checkpoint_ref": "checkpoint://semcod/planfile/runtime-core/1", + "plan_id": state["plan_id"], + "plan_hash": state["plan_hash"], + "candidate_key": "runtime-core", + "candidate_digest": candidate["candidate_digest"], + "sequence": 1, + "phase": "testing", + "head_sha": None, + "evidence_refs": [], + "recorded_at": datetime.now(UTC).isoformat(), + } + backend.checkpoint_delivery_candidate(checkpoint) + with pytest.raises(DeliveryPlanError, match="delivery_plan_checkpoint_conflict"): + backend.checkpoint_delivery_candidate({**checkpoint, "phase": "review"}) + + backend.record_delivery_terminal_receipt(state["plan_id"], _receipt(candidate)) + with pytest.raises(DeliveryPlanError, match="delivery_plan_receipt_conflict"): + backend.record_delivery_terminal_receipt( + state["plan_id"], + _receipt(candidate, receipt_ref="receipt://github/semcod/planfile/other"), + ) + + +def test_split_state_links_children_and_rewires_successor(tmp_path) -> None: + parent = _candidate("oversized-parent", 0, state="split_required") + child_a = _candidate("bounded-child-a", 1) + child_b = _candidate("bounded-child-b", 2) + successor = _candidate("integration-slice", 3, depends_on=["oversized-parent"]) + backend = Planfile(str(tmp_path)) + state = backend.materialize_delivery_plan(_plan([parent, child_a, child_b, successor])) + split = { + "schema": "planfile.delivery-plan-split/v1", + "split_ref": "split://semcod/planfile/oversized-parent/1", + "plan_id": state["plan_id"], + "plan_hash": state["plan_hash"], + "parent_candidate_key": "oversized-parent", + "parent_candidate_digest": parent["candidate_digest"], + "child_candidate_keys": ["bounded-child-a", "bounded-child-b"], + "recorded_at": datetime.now(UTC).isoformat(), + } + + linked = backend.record_delivery_split(split) + assert backend.record_delivery_split(split) == linked + parent_node = linked["candidates"]["oversized-parent"] + assert parent_node["status"] == "split" + assert parent_node["child_candidate_keys"] == ["bounded-child-a", "bounded-child-b"] + assert linked["candidates"]["integration-slice"]["depends_on"] == [ + "bounded-child-a", + "bounded-child-b", + ] + parent_ticket_id = parent_node["ticket_id"] + child_ticket = backend.get_ticket(linked["candidates"]["bounded-child-a"]["ticket_id"]) + assert child_ticket is not None and child_ticket.parent == parent_ticket_id + resume = backend.resume_delivery_plan(state["plan_id"]) + assert {item["candidate_key"] for item in resume["ready"]} == { + "bounded-child-a", + "bounded-child-b", + } + assert resume["split_active"] == ["oversized-parent"] + assert resume["waiting"][0]["waiting_on"] == ["bounded-child-a", "bounded-child-b"] + + backend.record_delivery_terminal_receipt( + state["plan_id"], + _receipt(child_a, receipt_ref="receipt://github/semcod/planfile/child-a"), + ) + backend.record_delivery_terminal_receipt( + state["plan_id"], + _receipt(child_b, receipt_ref="receipt://github/semcod/planfile/child-b"), + ) + completed_split = backend.resume_delivery_plan(state["plan_id"]) + assert completed_split["split_complete"] == ["oversized-parent"] + assert completed_split["ready"][0]["candidate_key"] == "integration-slice" + + +def test_compiled_contract_rejects_embedded_tool_authority() -> None: + candidate = _candidate("runtime-core", 0) + candidate["command"] = "git push --force" + with pytest.raises(ValidationError): + CompiledWorkPlanV1.model_validate(_plan([candidate])) + + plan = _plan([_candidate("runtime-core", 0)]) + plan["authority"] = "shell" + with pytest.raises(ValidationError): + CompiledWorkPlanV1.model_validate(plan)