Preserve failed implementation changes as unverified Git checkpoints

docs/checkpoint-verification.md

@@ -65,3 +65,67 @@ For an issue already linked through an authenticated parent receipt, teammates w
 This entry point requires a confirmed remote checkpoint whose factory branch still points to the selected revision. Local-only checkpoints, uncertain publications, changed remote heads, and unavailable source-control grants are refused. An older checkpoint is not silently replaced by the latest commit. Existing creation inputs without `source_checkpoint` keep their previous wire representation and request digest.
 
 Verification commands follow the selected runner policy. Cloning uses the existing controller-side Git command timeout and output bounds; this feature does not add clone disk-growth supervision. Ref-only admission adds no whole-project payload cap and does not download blobs as a preflight. It does not introduce distributed cloning, partial-clone fallback, automatic dependency provisioning, or a new natural-language permission to create test runs.
+
+## Retain source when a graph checkpoint cannot be stored
+
+A failed model session may have already written useful source. A **partial-source checkpoint** can retain that source without resuming the failed session. It stays marked `partial_unverified`: it does not mean the implementation finished or that its tests passed.
+
+This route is deliberately narrow. The latest stored native checkpoint must describe the complete current source. Every later consumed response must contain only supported source readers, and all earlier command and tool effects must have known, closed receipts. A pending response, a later write, changed source, unknown runner outcome, unsupported runtime contract, or revoked access makes the candidate unavailable. The error message alone cannot authorize recovery. This initial implementation refuses older runtime contracts. The API accepts no legacy override.
+
+<div class="forge-flow forge-flow--dependencies" role="region" aria-label="Partial source checkpoint lifecycle" aria-describedby="partial-source-flow-guide" tabindex="0" markdown="1">
+
+```mermaid
+---
+config:
+  flowchart:
+    wrappingWidth: 96
+    rankSpacing: 24
+---
+flowchart LR
+    A["Failed implementation"] --> B["Review stored source and receipts"]
+    B --> C["Unverified Git checkpoint"]
+    C --> D["Confirmed draft"]
+    D --> E["New test or maintain run"]
+    B -->|"Unknown or changed"| H["Keep source held"]
+```
+
+</div>
+
+<p class="forge-flow-caption" id="partial-source-flow-guide">Follow the recorded source into a checkpoint and a new run. Focus and scroll the diagram on a narrow screen.</p>
+
+Review `GET /api/v1/runs/{run_id}/git-deliveries/partial-source-candidate` while signed in with current project and repository execution access. Draft publication also requires release access and the run's original enabled Git-delivery consent to remain current. The response contains evidence hashes and source identifiers, without returning the model prompt or source contents.
+
+Use the exact candidate to submit one source-only request:
+
+```python
+from uuid import uuid4
+
+candidate = client.partial_source_candidate(failed_run_id)
+consent_fields = (
+    "policy_digest", "expected_run_digest", "journal_digest",
+    "source_digest", "frontier_record_digest",
+)
+request = {
+    "version": 1,
+    "request_id": uuid4().hex,
+    "expected_head": candidate["expected_head"],
+    "expected_source_digest": candidate["source_digest"],
+    "explicit_partial_source": {
+        "version": 1,
+        "candidate_digest": candidate["digest"],
+        **{name: candidate[name] for name in consent_fields},
+        "acknowledge_partial_unverified": True,
+        "acknowledge_preserve_failed_graph": True,
+        "acknowledge_no_execution_grant": True,
+    },
+}
+receipt = client.request_partial_source_checkpoint(failed_run_id, request)
+```
+
+The API queues this through `POST /api/v1/runs/{run_id}/git-deliveries/partial-source-requests`. The execution owner rechecks the source, current user, policy, journal and effect receipts before committing. The commit is labeled `[UNVERIFIED]`; the retained diff artifact and checkpoint receipt identify its exact source. When the original policy permits draft delivery, the ordinary source-control integration publishes that checkpoint as a draft.
+
+Read `client.git_delivery_request(failed_run_id, request["request_id"])` to follow the outcome. After a lost response, inspect that same request ID. Neither the SDK nor the worker automatically repeats an unknown Git or remote effect. In this initial partial-source route, an unknown local or remote result stays held for inspection. The existing remote-reconciliation route is not yet supported for this partial-source marker; do not submit a second request or treat the retained local commit as confirmed remote delivery.
+
+The original run remains failed, with its task, pending graph markers, consumed responses, tool receipts, command ordinals and spent budgets intact. Only its source-delivery history advances. After a confirmed remote checkpoint, select that exact `source_checkpoint` in a new test or maintain admission as described above. The child gets a new identity and independently admitted settings; it never receives the failed session's consumed results as execution authority.
+
+This first slice does not settle a write or command made after the latest stored source frontier. Such source remains quarantined until a supported native recovery path can prove its effects. It does not add a button to bypass a hold, reset a budget or declare the product complete.

src/firefly_forge/client.py

@@ -926,6 +926,25 @@ def request_git_delivery(self, run_id: str, request: GitDeliveryInputV1 | dict[s
     def git_delivery_request(self, run_id: str, request_id: str) -> dict[str, Any]:
         return dict(self.request("GET", f"runs/{self._id(run_id)}/git-deliveries/requests/{self._id(request_id)}"))
 
+    def partial_source_candidate(self, run_id: str) -> dict[str, Any]:
+        return dict(self.request("GET", f"runs/{self._id(run_id)}/git-deliveries/partial-source-candidate"))
+
+    def request_partial_source_checkpoint(
+        self, run_id: str, request: GitDeliveryInputV1 | dict[str, Any]
+    ) -> dict[str, Any]:
+        from firefly_forge.git_delivery_models import GitDeliveryInputV1
+
+        body = GitDeliveryInputV1.model_validate(request)
+        if body.explicit_partial_source is None:
+            raise ValueError("Partial source requires explicit source-only consent")
+        return dict(
+            self.request(
+                "POST",
+                f"runs/{self._id(run_id)}/git-deliveries/partial-source-requests",
+                json=body.model_dump(mode="json"),
+            )
+        )
+
     def _diff_path(self, run_id: str, expected_manifest: str, request_id: str | None) -> str:
         if not isinstance(expected_manifest, str) or re.fullmatch(r"[a-f0-9]{64}", expected_manifest) is None:
             raise ValueError("Use an exact diff manifest SHA-256")

src/firefly_forge/git_deliveries.py

@@ -89,7 +89,12 @@ async def candidate(self, run_id: str, *, actor: Data | None = None) -> Data:
             actor = ProjectAccess.actor_snapshot(user)
         run = await self.store.get_run(run_id)
         policy = GitDeliveryPolicyV1.model_validate(run.get("git_delivery_policy") or {})
-        if not policy.enabled or run["status"] not in {"paused", "failed"} or run.get("git_failed_unit_adoption"):
+        if (
+            not policy.enabled
+            or run["status"] not in {"paused", "failed"}
+            or run.get("git_failed_unit_adoption")
+            or run.get("git_partial_source_checkpoint")
+        ):
             raise ValueError("Review a quiescent run with its original enabled checkpoint policy")
         await self._authorize(run, actor, "release" if policy.remote == "draft_pr" else "execute")
         active = self.engine.active.get(run_id)
@@ -148,6 +153,13 @@ async def _guard(self, run_id: str, expected: Data, actor: Data | None, *, bound
         from firefly_forge.git_failed_unit_adoption import effective_policy
 
         policy = await effective_policy(self, current, actor=actor, writing=True)
+        partial = current.get("git_partial_source_checkpoint")
+        if partial:
+            from firefly_forge.git_partial_source_checkpoints import GitPartialSourceCheckpoints
+
+            if actor is None:
+                raise ValueError("Partial source requires its current explicit actor")
+            await GitPartialSourceCheckpoints(self).guard(current, actor)
         if not policy.enabled or self.engine.stopping:
             raise ValueError("Intermediate Git delivery is disabled or the execution owner is draining")
         if self.authorization is None:
@@ -158,8 +170,10 @@ async def _guard(self, run_id: str, expected: Data, actor: Data | None, *, bound
                 raise ValueError("Current project Git delivery consent is disabled")
             tasks = (await self.journal.plan(run_id))["tasks"]
             implementation = next((task for task in tasks if task["id"] == "implement"), None)
-            if not current.get("git_failed_unit_adoption") and (
-                implementation is None or implementation["status"] not in {"done", "paused"}
+            if (
+                not current.get("git_failed_unit_adoption")
+                and not partial
+                and (implementation is None or implementation["status"] not in {"done", "paused"})
             ):
                 raise ValueError("Git delivery requires a recorded closed implementation unit")
             if implementation is not None and implementation["status"] == "paused":
@@ -194,8 +208,7 @@ async def _guard(self, run_id: str, expected: Data, actor: Data | None, *, bound
         if boundary is not self and current["status"] not in {"paused", "failed"}:
             raise ValueError("Explicit checkpointing requires a known quiescent paused or failed run")
         if (
-            current.get("bridge_completion_hold")
-            or current.get("implementation_pending")
+            ((current.get("bridge_completion_hold") or current.get("implementation_pending")) and not partial)
             or current.get("cancel_request", {}).get("status") == "requested"
             or (current.get("provider_recovery") or {}).get("status")
             in {"unknown", "executing", "needs_reconciliation"}
@@ -219,7 +232,15 @@ async def _guard(self, run_id: str, expected: Data, actor: Data | None, *, bound
         return dict(final)
 
     async def _entries(self, run_id: str) -> list[Data]:
-        entries = await self.journal.entries(run_id)
+        run = await self.store.get_run(run_id)
+        if run.get("git_partial_source_checkpoint"):
+            from firefly_forge.git_partial_source_checkpoints import GitPartialSourceCheckpoints
+
+            # These rows stay internal; public methods project typed Git receipts, never model bodies.
+            # The source-frontier producer separately scans every decoded document with the real guard.
+            entries = await GitPartialSourceCheckpoints(self)._entries(run_id)
+        else:
+            entries = await self.journal.entries(run_id)
         verifier = getattr(self.journal, "verify_chain", None)
         if (
             not callable(verifier)
@@ -331,6 +352,9 @@ async def checkpoint(
             raise ValueError("Keep a canonical checkpoint request identity before admission")
         async with self.locks.setdefault(run_id, asyncio.Lock()):
             run = await self.store.get_run(run_id)
+            partial = run.get("git_partial_source_checkpoint")
+            if partial and (final or request_id != partial.get("request_id")):
+                raise ValueError("Partial source has one unverified checkpoint request and no final-ready authority")
             from firefly_forge.git_failed_unit_adoption import effective_policy
 
             policy = await effective_policy(self, run, actor=actor, writing=True)
@@ -459,7 +483,7 @@ async def checkpoint(
                     f"Checkpoint {run.get('active_task', 'implement')} unit {run.get('implementation_rounds', 1)} "
                     f"for forge run {run_id}"
                 )
-                if run.get("git_failed_unit_adoption"):
+                if run.get("git_failed_unit_adoption") or partial:
                     message = "[UNVERIFIED] " + message
                 await checked(
                     [
@@ -504,7 +528,7 @@ async def checkpoint(
                     "pr_url": None,
                     "created_at": datetime.now(UTC).isoformat(),
                 }
-                if run.get("git_failed_unit_adoption"):
+                if run.get("git_failed_unit_adoption") or partial:
                     receipt["verification_status"] = "partial_unverified"
                 receipt["checkpoint_digest"] = delivery_digest(receipt)
                 await self._append(run_id, "git_delivery_receipt", key, receipt)
@@ -830,7 +854,9 @@ async def _publish_checkpoint(
             source_tree=receipt["source_tree"],
             sequence=receipt["sequence"],
             changed_file_count=receipt["changed_file_count"],
-            implementation_disposition="failed" if run.get("git_failed_unit_adoption") else None,
+            implementation_disposition="failed"
+            if run.get("git_failed_unit_adoption") or run.get("git_partial_source_checkpoint")
+            else None,
         )
         await self.store.admit_credentials(context.model_dump())
         native = (
@@ -855,7 +881,7 @@ async def _publish_checkpoint(
             raise ValueError("Selected checkpoint communication is held; no branch or draft PR was published")
         body = drafted["body"] + f"\n\n<!-- forge-git-delivery:{run_id} -->"
         title = drafted["body"].splitlines()[0].removeprefix("### ")[:100]
-        if run.get("git_failed_unit_adoption"):
+        if run.get("git_failed_unit_adoption") or run.get("git_partial_source_checkpoint"):
             title = "[UNVERIFIED] " + title[:87]
         await self.store.admit_credentials({"title": title, "body": body, "communication": evidence})
         await self._guard(run_id, run, actor, boundary=boundary)

src/firefly_forge/git_delivery_controllers.py

@@ -55,6 +55,22 @@ async def request(self, run_id: PathVar[str], body: Body[GitDeliveryInputV1]) ->
     async def candidate(self, run_id: PathVar[str]) -> dict[str, Any]:
         return await self.deliveries.candidate(run_id)
 
+    @get_mapping("/partial-source-candidate")
+    async def partial_source_candidate(self, run_id: PathVar[str]) -> dict[str, Any]:
+        from firefly_forge.git_partial_source_checkpoints import GitPartialSourceCheckpoints
+        from firefly_forge.project_access import CURRENT_PRINCIPAL, AccessDenied, ProjectAccess
+
+        user = CURRENT_PRINCIPAL.get()
+        if user is None:
+            raise AccessDenied("Sign in before reviewing partial source")
+        return await GitPartialSourceCheckpoints(self.deliveries).candidate(run_id, ProjectAccess.actor_snapshot(user))
+
+    @post_mapping("/partial-source-requests", status_code=202)
+    async def partial_source_request(self, run_id: PathVar[str], body: Body[GitDeliveryInputV1]) -> dict[str, Any]:
+        if body.explicit_partial_source is None:
+            raise ValueError("Partial source requires explicit source-only consent")
+        return await self.tickets.request(run_id, body)
+
     @get_mapping("/requests/{request_id}")
     async def ticket(self, run_id: PathVar[str], request_id: PathVar[str]) -> dict[str, Any]:
         return await self.tickets.get(run_id, request_id)

src/firefly_forge/git_delivery_models.py

@@ -15,6 +15,7 @@
 )
 
 from firefly_forge.git_diff_models import GitDiffBudget, GitDiffRefV1
+from firefly_forge.git_partial_source_models import GitPartialSourceConsentV1
 from firefly_forge.scm_authority_models import Digest, NativeID, StrictContract
 
 GitRevision = Annotated[str, Field(pattern=r"^(?:[a-f0-9]{40}|[a-f0-9]{64})$")]
@@ -100,9 +101,17 @@ class GitDeliveryInputV1(GitDeliveryContract):
     expected_source_digest: Digest | None = None
     explicit_adoption: GitFailedUnitAdoptionV1 | None = None
     explicit_reconciliation: GitRemoteReconciliationV1 | None = None
+    explicit_partial_source: GitPartialSourceConsentV1 | None = None
 
     @model_validator(mode="after")
     def reviewed_source(self) -> GitDeliveryInputV1:
+        if self.explicit_partial_source is not None and (
+            self.explicit_adoption is not None
+            or self.explicit_reconciliation is not None
+            or self.expected_source_digest != self.explicit_partial_source.source_digest
+            or self.expected_source_tree is not None
+        ):
+            raise ValueError("Partial source requires its exclusive exact source-frontier consent")
         if self.explicit_reconciliation is not None and (
             self.explicit_adoption is not None
             or self.request_id == self.explicit_reconciliation.original_request_id

src/firefly_forge/git_delivery_requests.py

@@ -72,6 +72,26 @@ async def request(self, run_id: str, body: GitDeliveryInputV1) -> Data:
                 raise ValueError("The adopted source HEAD differs from the explicit review")
             policy = GitDeliveryPolicyV1.model_validate(candidate["policy"])
         else:
+            if body.explicit_partial_source is not None:
+                from firefly_forge.git_partial_source_checkpoints import GitPartialSourceCheckpoints
+
+                async with self.store._unit():
+                    retained = await self.requests.records.find_by_id(identifier(body.request_id))
+                if retained is not None:
+                    if (
+                        retained.kind != KIND
+                        or retained.run_id.hex != run_id
+                        or retained.data["request"]["input"] != body.model_dump(mode="json")
+                        or retained.data["request"]["actor"] != actor
+                    ):
+                        raise ValueError("The original partial-source request belongs to another input or actor")
+                    return await self.get(run_id, body.request_id)
+                candidate = await GitPartialSourceCheckpoints(self.deliveries).candidate(run_id, actor)
+                GitPartialSourceCheckpoints.compare(candidate, body.explicit_partial_source)
+                if body.expected_head != candidate["expected_head"]:
+                    raise ValueError("The reviewed partial-source HEAD changed")
+            elif run.get("git_partial_source_checkpoint") and reconciliation is None:
+                raise ValueError("Inspect the original partial-source request without replay")
             if run.get("git_failed_unit_adoption") and reconciliation is None:
                 raise ValueError(
                     "An adopted failed unit has one original checkpoint request; inspect it without replay"
@@ -100,6 +120,10 @@ async def request(self, run_id: str, body: GitDeliveryInputV1) -> Data:
             if body.explicit_adoption is not None:
                 current_candidate = await GitFailedUnitAdoptions(self.deliveries).candidate(run_id, actor)
                 GitFailedUnitAdoptions.compare(current_candidate, body.explicit_adoption)
+            if body.explicit_partial_source is not None:
+                await GitPartialSourceCheckpoints(self.deliveries).revalidate(
+                    run_id, actor, candidate, staged_tree=None
+                )
             if reconciliation is not None:
                 from firefly_forge.git_remote_reconciliation import GitRemoteReconciliations
 
@@ -238,6 +262,12 @@ async def execute(self, request_id: str) -> None:
             if binding(run) != request["binding"]:
                 raise ValueError("The admitted checkpoint run changed before worker dispatch")
             body = GitDeliveryInputV1.model_validate(request["input"])
+            if body.explicit_partial_source is not None:
+                from firefly_forge.git_partial_source_checkpoints import GitPartialSourceCheckpoints
+
+                await GitPartialSourceCheckpoints(self.deliveries).admit(
+                    claimed["run_id"], body.request_id, request["actor"], body.explicit_partial_source
+                )
             if body.explicit_adoption is not None:
                 from firefly_forge.git_failed_unit_adoption import GitFailedUnitAdoptions