Persist resumable delivery-plan DAGs and split state

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

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/<plan-id>.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

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)