From 907dc48777ffa19f0ecc32afd9cc1c565c817e95 Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Sat, 24 Oct 2026 23:56:12 -0700 Subject: [PATCH 1/5] Retain failed source through native stored-frontier checkpoints --- docs/checkpoint-verification.md | 52 ++ src/firefly_forge/client.py | 19 + src/firefly_forge/git_deliveries.py | 46 +- src/firefly_forge/git_delivery_controllers.py | 16 + src/firefly_forge/git_delivery_models.py | 9 + src/firefly_forge/git_delivery_requests.py | 30 + .../git_partial_source_checkpoints.py | 620 ++++++++++++++++++ .../git_partial_source_models.py | 105 +++ tests/test_git_partial_source_clients.py | 42 ++ tests/test_git_partial_source_effects_host.py | 54 ++ .../test_git_partial_source_frontier_host.py | 293 +++++++++ tests/test_git_partial_source_models.py | 47 ++ tests/test_git_partial_source_native.py | 319 +++++++++ .../test_git_partial_source_requests_host.py | 309 +++++++++ tests/test_partial_settlement_design_host.py | 76 +++ 15 files changed, 2027 insertions(+), 10 deletions(-) create mode 100644 src/firefly_forge/git_partial_source_checkpoints.py create mode 100644 src/firefly_forge/git_partial_source_models.py create mode 100644 tests/test_git_partial_source_clients.py create mode 100644 tests/test_git_partial_source_effects_host.py create mode 100644 tests/test_git_partial_source_frontier_host.py create mode 100644 tests/test_git_partial_source_models.py create mode 100644 tests/test_git_partial_source_native.py create mode 100644 tests/test_git_partial_source_requests_host.py create mode 100644 tests/test_partial_settlement_design_host.py diff --git a/docs/checkpoint-verification.md b/docs/checkpoint-verification.md index e9cbd7ff..06a902bd 100644 --- a/docs/checkpoint-verification.md +++ b/docs/checkpoint-verification.md @@ -65,3 +65,55 @@ 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. + +```mermaid +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"] +``` + +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. diff --git a/src/firefly_forge/client.py b/src/firefly_forge/client.py index e1a1ccac..958241ce 100644 --- a/src/firefly_forge/client.py +++ b/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") diff --git a/src/firefly_forge/git_deliveries.py b/src/firefly_forge/git_deliveries.py index 033de975..e4ec545e 100644 --- a/src/firefly_forge/git_deliveries.py +++ b/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" 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) diff --git a/src/firefly_forge/git_delivery_controllers.py b/src/firefly_forge/git_delivery_controllers.py index bb586ffd..bcd1f43a 100644 --- a/src/firefly_forge/git_delivery_controllers.py +++ b/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) diff --git a/src/firefly_forge/git_delivery_models.py b/src/firefly_forge/git_delivery_models.py index e925bce4..a74cc9e0 100644 --- a/src/firefly_forge/git_delivery_models.py +++ b/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 diff --git a/src/firefly_forge/git_delivery_requests.py b/src/firefly_forge/git_delivery_requests.py index d87d7146..368e7990 100644 --- a/src/firefly_forge/git_delivery_requests.py +++ b/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 diff --git a/src/firefly_forge/git_partial_source_checkpoints.py b/src/firefly_forge/git_partial_source_checkpoints.py new file mode 100644 index 00000000..54b26b9b --- /dev/null +++ b/src/firefly_forge/git_partial_source_checkpoints.py @@ -0,0 +1,620 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Stored source frontiers do not authorize restoration of their failed graphs.""" + +from __future__ import annotations + +import copy +import hashlib +import json +from typing import Any + +from jsonschema import Draft202012Validator # type: ignore[import-untyped] +from openai.types.chat import ChatCompletion + +from firefly_forge.agent_checkpoint_models import CheckpointPayloadV2, ConsumePayload +from firefly_forge.agent_checkpoints import AgentCheckpoints +from firefly_forge.agentic_checkpoint_boundary import BoundaryCheckpoint +from firefly_forge.agentic_checkpoint_invocation import STATIC_TOOLS, invocation_cut, runtime_contract +from firefly_forge.bridge_completion_models import CompletionIntentPayload, CompletionResponsePayload +from firefly_forge.bridge_completions import _cut +from firefly_forge.git_partial_source_models import ( + PartialSourceAdmissionV1, + PartialSourceCandidateV1, + PartialSourceFrontierV1, +) +from firefly_forge.journal import Journal, digest + +Data = dict[str, Any] +READ_ONLY = frozenset({"list_files", "read_file", "search_text", "git_diff"}) +KIND = "git_partial_source_checkpoint" + + +def stored_frontier_proof(run: Data, task: Data, entries: list[Data]) -> Data: + """Validate the complete original HTTP custody; never expose or consume its results.""" + if ( + run.get("agent") != "bridge" + or run.get("status") != "failed" + or run.get("active_task") != "implement" + or task.get("id") != "implement" + or task.get("status") not in {"running", "needs_reconciliation"} + or task.get("attempt") != run.get("attempt") + or not run.get("implementation_pending") + or not run.get("bridge_completion_hold") + or not entries + or not Journal.verify_chain(entries) + ): + raise ValueError("Partial source requires the exact failed, held native implementation") + kinds = ("bridge_request_intent", "agent_checkpoint", "bridge_request_response", "agent_consume") + grouped = {kind: [e for e in entries if e["kind"] == kind] for kind in kinds} + keys = [e["idempotency_key"] for e in grouped[kinds[0]]] + if ( + not keys + or len(keys) != len(set(keys)) + or any([e["idempotency_key"] for e in grouped[kind]] != keys for kind in kinds[1:]) + ): + raise ValueError("The original completion closure is incomplete, duplicated or unknown") + if run["bridge_completion_hold"] != { + "version": 1, + "reason": "graph_checkpoint_unavailable", + "request_id": keys[0], + "intent_digest": grouped["bridge_request_intent"][0]["digest"], + }: + raise ValueError("The held graph does not identify its original native completion") + current_contract = runtime_contract() + previous: CheckpointPayloadV2 | None = None + invocation_id = None + for parts in zip(*(grouped[kind] for kind in kinds), strict=True): + intent, checkpoint, response, consume = parts + if any(e["run_id"] != run["id"] or e["task_id"] != "implement" for e in parts): + raise ValueError("Completion closure belongs to another run or task") + original_intent = CompletionIntentPayload.model_validate(intent["payload"]) + payload = CheckpointPayloadV2.model_validate(checkpoint["payload"]) + bound = payload.invocation + if bound.contract != current_contract: + raise ValueError("The stored frontier has an unsupported runtime contract") + if ( + payload.request_id != intent["idempotency_key"] + or payload.intent_digest != intent["digest"] + or original_intent.run_id != run["id"] + or original_intent.descriptor.request_id != intent["idempotency_key"] + or original_intent.profile_id != (run.get("model_selection") or {}).get("profile_id") + or original_intent.project_id != run.get("project_id") + or original_intent.repository_id != run.get("repository_id") + or original_intent.config_field != "config_revision" + or original_intent.config_revision != run.get("config_revision") + or original_intent.scope_digest != _cut(run, "config_revision") + or (previous is None and (bound.ordinal != 1 or bound.predecessor is not None)) + or bound.run_cut != invocation_cut(run) + or bound.round != task["attempt"] + or not bound.grants + or not set(bound.grants) <= STATIC_TOOLS + or intent["payload"].get("task_key") != task.get("idempotency_key") + or intent["payload"].get("task_attempt") != task["attempt"] + or (invocation_id is not None and bound.invocation_id != invocation_id) + or ( + previous is not None + and ( + bound.ordinal != previous.invocation.ordinal + 1 or bound.predecessor != previous.checkpoint_digest + ) + ) + or not intent["sequence"] < checkpoint["sequence"] < response["sequence"] < consume["sequence"] + ): + raise ValueError("The completion closure has a different invocation, task or frozen scope") + AgentCheckpoints._validate_checkpoint(checkpoint, intent) + body = CompletionResponsePayload.model_validate(response["payload"]) + consumed = ConsumePayload.model_validate(consume["payload"]) + raw = body.response.encode() + if ( + body.intent_digest != intent["digest"] + or body.request_id != intent["idempotency_key"] + or body.response_sha256 != hashlib.sha256(raw).hexdigest() + or body.response_bytes != len(raw) + or consumed.request_id != intent["idempotency_key"] + or consumed.intent_digest != intent["digest"] + or consumed.checkpoint_record_digest != checkpoint["digest"] + or consumed.response_record_digest != response["digest"] + or consumed.response_sha256 != body.response_sha256 + or consumed.response_bytes != body.response_bytes + ): + raise ValueError("The original response and consumed record have different custody") + completion = ChatCompletion.model_validate_json(raw, strict=True) + if ( + completion.model != intent["payload"]["descriptor"]["model"] + or len(completion.choices) != 1 + or completion.choices[0].finish_reason not in {"stop", "tool_calls"} + ): + raise ValueError("The completion does not identify one exact original model response") + previous, invocation_id = payload, bound.invocation_id + frontier = grouped["agent_checkpoint"][-1] + captured = BoundaryCheckpoint.decode( + frontier["payload"]["snapshot"].encode(), frontier["payload"]["checkpoint_digest"] + ) + schemas = { + item["name"]: item["parameters_json_schema"] + for item in captured.record.contract["parameters"]["function_tools"] + } + names = [] + for response in grouped["bridge_request_response"]: + if response["sequence"] <= frontier["sequence"]: + continue + completion = ChatCompletion.model_validate_json(response["payload"]["response"], strict=True) + choice = completion.choices[0] + calls = choice.message.tool_calls + if choice.finish_reason != "tool_calls" or not calls: + raise ValueError("Partial source needs only read-only calls after the stored frontier") + for call in calls: + if call.type != "function" or call.function.name not in READ_ONLY or call.function.name not in schemas: + raise ValueError("Only supported read-only source calls may follow the stored frontier") + arguments = json.loads(call.function.arguments) + if not isinstance(arguments, dict) or not Draft202012Validator(schemas[call.function.name]).is_valid( + arguments + ): + raise ValueError("Read-only source arguments differ from the frozen schema") + names.append(call.function.name) + return PartialSourceFrontierV1.model_validate( + { + "version": 1, + "kind": KIND, + "frontier_record_digest": frontier["digest"], + "frontier_checkpoint_digest": frontier["payload"]["checkpoint_digest"], + "invocation": { + key: value + for key, value in previous.invocation.model_dump(mode="json").items() + if key not in {"brief", "backend", "frontend", "deadline_ms"} + } + if previous is not None + else None, + "invocation_digest": digest(previous.invocation.model_dump(mode="json")) if previous is not None else None, + "completion_count": len(keys), + "closure_digest": digest({"records": [e["digest"] for group in grouped.values() for e in group]}), + "journal_head": entries[-1]["digest"], + "later_tool_names": names, + "task": copy.deepcopy({k: task.get(k) for k in ("id", "status", "attempt", "idempotency_key")}), + "verification_status": "partial_unverified", + } + ).model_dump(mode="json") + + +class GitPartialSourceCheckpoints: + """A source-only owner path, separate from failed feedback and graph restoration.""" + + def __init__(self, deliveries: Any, *, tool_records: Any = None) -> None: + from firefly_forge.tool_persistence import ForgeToolInvocationRepository + + self.d, self.store, self.journal = deliveries, deliveries.store, deliveries.journal + self.tools = tool_records if tool_records is not None else ForgeToolInvocationRepository() + + async def _entries(self, run_id: str) -> list[Data]: + from firefly_forge.store import identifier + + async with self.store._unit(): + rows = await self.journal.records.find_for_run(identifier(run_id)) + entries = [copy.deepcopy(row.data) for row in rows] + for row, entry in zip(rows, entries, strict=True): + scalar = dict( + run_id=row.run_id.hex, + task_id=row.task_id, + kind=row.kind, + previous_hash=row.previous_hash, + sequence=row.sequence, + idempotency_key=row.idempotency_key, + digest=row.digest, + ) + if any(entry.get(key) != value for key, value in scalar.items()): + raise ValueError("Native partial-source scalar identity differs from its journal envelope") + if ( + len(entries) > 10000 + or len(json.dumps(entries).encode()) > 16 * 1024 * 1024 + or not self.journal.verify_chain(entries) + ): + raise ValueError("Partial source requires its complete intact native journal") + return entries + + async def _authority(self, run: Data, actor: Data) -> Data: + from firefly_forge.git_delivery_models import GitDeliveryPolicyV1 + from firefly_forge.git_delivery_policy import GitDeliveryPolicies + from firefly_forge.project_access import CURRENT_PRINCIPAL, AccessDenied, ProjectAccess + + user = CURRENT_PRINCIPAL.get() + if user is not None and ProjectAccess.actor_snapshot(user) != actor: + raise AccessDenied("Partial source identity changed") + await self.d._authorize(run, actor, "execute") + policy = GitDeliveryPolicyV1.model_validate(run.get("git_delivery_policy") or {}) + saved = await GitDeliveryPolicies(self.store).snapshot(run.get("project_id")) + if ( + not policy.enabled + or not saved + or saved["policy"] != run.get("git_delivery_policy") + or saved["digest"] != run.get("git_delivery_policy_digest") + ): + raise ValueError("The original enabled Git consent is no longer current") + if policy.remote == "draft_pr": + await self.d._authorize(run, actor, "release") + await self.store.assert_ownership() + return saved + + def _quiescent(self, run: Data) -> None: + active = self.d.engine.active.get(run["id"]) + if ( + self.d.engine.stopping + or (active is not None and not active.done()) + or run.get("cancel_request", {}).get("status") == "requested" + or (run.get("provider_recovery") or {}).get("status") + in {"reserved", "executing", "unknown", "needs_reconciliation"} + or (run.get("worker_request") or {}).get("status") == "pending" + ): + raise ValueError("A live, cancelled or unknown owner prevents partial source settlement") + + async def _effects(self, run: Data, entries: list[Data], frontier_sequence: int) -> str: + from firefly_forge.runner_factory_models import validate_factory_pool + from firefly_forge.sandbox_command_slots import COMMAND_CONTRACT, _canonical, _complete + from firefly_forge.sandbox_models import capture_sandbox, validate_sandbox_snapshot + from firefly_forge.store import identifier + + async with self.store._unit(): + commands = await self.journal.sandboxes.for_run(identifier(run["id"])) + tools = await self.tools.for_run(identifier(run["id"])) + if { + e["payload"].get("execution_id") + for e in entries + if e["kind"] in {"sandbox_intent", "sandbox_receipt", "sandbox_reconcile"} + } != {r.id.hex for r in commands}: + raise ValueError("Orphan command custody prevents partial source settlement") + snapshot = run.get("sandbox_snapshot") or capture_sandbox(None) + policy = ( + validate_factory_pool(snapshot) if snapshot.get("mode") == "pool" else validate_sandbox_snapshot(snapshot) + ) + values = [] + for row in commands: + value = _canonical(row, entries, run["id"]) + effects = [ + e + for e in entries + if e["kind"] in {"sandbox_intent", "sandbox_receipt", "sandbox_reconcile"} + and e["payload"].get("execution_id") == row.id.hex + ] + if ( + value["request"].get("executor_contract") != COMMAND_CONTRACT + or row.snapshot_digest != snapshot["digest"] + or row.ordinal > policy.command_limit + or not _complete(row, value) + or max(e["sequence"] for e in effects) >= frontier_sequence + ): + raise ValueError("Incomplete or later command effects prevent partial source settlement") + values.append(value) + if { + e["idempotency_key"] + for e in entries + if e["kind"] in {"tool_intent", "tool_receipt", "tool_unknown", "tool_reconciled"} + } != {r.correlation_id for r in tools}: + raise ValueError("Orphan tool custody prevents partial source settlement") + tool_values = [] + for row in tools: + intents = [e for e in entries if e["kind"] == "tool_intent" and e["idempotency_key"] == row.correlation_id] + outcomes = [ + e + for e in entries + if e["kind"] in {"tool_receipt", "tool_unknown", "tool_reconciled"} + and e["idempotency_key"] == row.correlation_id + ] + evidence = {k: copy.deepcopy(row.data[k]) for k in ("result", "error", "observation") if k in row.data} + if ( + row.run_id.hex != run["id"] + or row.status not in {"succeeded", "rejected"} + or len(intents) != 1 + or len(outcomes) != 1 + or intents[0]["task_id"] != row.task_id + or outcomes[0]["task_id"] != row.task_id + or row.data.get("admission") != intents[0]["payload"] + or any( + intents[0]["payload"].get(key) != run.get(key) + for key in ("repository_id", "project_id", "component_id") + ) + or intents[0]["payload"].get("run_id") != run["id"] + or intents[0]["payload"].get("tool") != row.tool + or intents[0]["payload"].get("task_id") != row.task_id + or intents[0]["payload"].get("correlation_id") != row.correlation_id + or row.data.get("request_digest") != digest(intents[0]["payload"]) + or outcomes[0]["kind"] != "tool_receipt" + or outcomes[0]["payload"] + != {"correlation_id": row.correlation_id, "tool": row.tool, "status": row.status, **evidence} + or not intents[0]["sequence"] < outcomes[0]["sequence"] < frontier_sequence + ): + raise ValueError("Unsettled or later tool effects prevent partial source settlement") + tool_values.append({"id": row.id.hex, "status": row.status, "data": row.data}) + return digest({"commands": values, "tools": tool_values}) + + @staticmethod + def _run_cut(run: Data) -> str: + # Only native source-delivery projections may advance; all failed graph data remains bound. + value = { + key: item + for key, item in run.items() + if key + not in { + "events", + "updated_at", + "git_delivery_head", + "git_delivery_summary", + "git_delivery_draft", + "git_partial_source_checkpoint", + } + } + value["checkpoint_revision"] = run.get("checkpoint_revision") + return digest(value) + + async def candidate(self, run_id: str, actor: Data) -> Data: + from firefly_forge.agentic_checkpoint_invocation import workspace_cut + from firefly_forge.source_snapshot import git_index_identity + from firefly_forge.verification_models import verification_source + + run = await self.store.get_run(run_id) + if ( + run.get("git_partial_source_checkpoint") + or run.get("git_failed_unit_adoption") + or run.get("git_delivery_head") + or run.get("git_delivery_summary") is not None + or run.get("git_delivery_draft") is not None + ): + raise ValueError("Inspect the retained partial source request; do not replace its custody") + self._quiescent(run) + saved = await self._authority(run, actor) + entries = await self._entries(run_id) + if any(e["kind"].startswith("git_delivery_") or e["kind"] == KIND for e in entries): + raise ValueError("An existing Git effect must be inspected rather than bypassed") + task = next(t for t in (await self.journal.plan(run_id))["tasks"] if t["id"] == "implement") + proof = stored_frontier_proof(run, task, entries) + frontier_sequence = next(e["sequence"] for e in entries if e["digest"] == proof["frontier_record_digest"]) + effects = await self._effects(run, entries, frontier_sequence) + # Scan decoded histories as well as encoded WAL; compression must never conceal credentials. + scanner = await self.store.source_patch_scanner() + for entry in entries: + documents = [json.dumps(entry, ensure_ascii=False).encode()] + if entry["kind"] == "agent_checkpoint": + documents.append( + BoundaryCheckpoint.decode( + entry["payload"]["snapshot"].encode(), entry["payload"]["checkpoint_digest"] + ).canonical_payload + ) + for document in documents: + for start in range(0, len(document), 65536): + await scanner(document[start : start + 65536], False) + scanner.finish_document() + await scanner(b"", True) + root = self.d.workspace(run) + cut = await workspace_cut(self.store, root) + if ( + cut != proof["invocation"]["workspace"] + or cut["head"] != (run.get("checkpoint_revision") or run.get("base_revision")) + or cut["metadata"] != run.get("git_metadata_digest") + ): + raise ValueError("Current source differs from the latest stored native frontier") + index = git_index_identity(root).sha256 + source = verification_source(root) + value = dict( + version=1, + run_id=run_id, + actor=copy.deepcopy(actor), + policy_digest=saved["digest"], + expected_run_digest=self._run_cut(run), + journal_digest=entries[-1]["digest"], + source_digest=source, + index_digest=index, + expected_head=cut["head"], + workspace=cut, + frontier_record_digest=proof["frontier_record_digest"], + proof=proof, + effects_digest=effects, + original_checkpoint_revision=run.get("checkpoint_revision"), + ) + await self.store.admit_credentials(value) + await scanner.revalidate() + await self.revalidate(run_id, actor, value, staged_tree=None) + return PartialSourceCandidateV1.model_validate({**value, "digest": digest(value)}).model_dump(mode="json") + + @staticmethod + def compare(candidate: Data, body: Any) -> None: + if any( + getattr(body, key) != candidate[field] + for key, field in { + "candidate_digest": "digest", + "policy_digest": "policy_digest", + "expected_run_digest": "expected_run_digest", + "journal_digest": "journal_digest", + "source_digest": "source_digest", + "frontier_record_digest": "frontier_record_digest", + }.items() + ): + raise ValueError("The reviewed partial source frontier changed") + + async def revalidate(self, run_id: str, actor: Data, candidate: Data, *, staged_tree: str | None) -> None: + from firefly_forge.git_workspace import metadata_digest + from firefly_forge.source_snapshot import git_index_identity + from firefly_forge.verification_models import verification_source + from firefly_forge.workspace_reconciliation import _git + + run = await self.store.get_run(run_id) + self._quiescent(run) + await self._authority(run, actor) + if self._run_cut(run) != candidate["expected_run_digest"]: + raise ValueError("Failed graph, task, grants or source scope changed") + entries = await self._entries(run_id) + if not entries or entries[-1]["digest"] != candidate["journal_digest"]: + raise ValueError("The native source-frontier journal changed") + task = next(t for t in (await self.journal.plan(run_id))["tasks"] if t["id"] == "implement") + if {k: task.get(k) for k in candidate["proof"]["task"]} != candidate["proof"]["task"]: + raise ValueError("The retained failed task changed") + sequence = next(e["sequence"] for e in entries if e["digest"] == candidate["frontier_record_digest"]) + if await self._effects(run, entries, sequence) != candidate["effects_digest"]: + raise ValueError("The retained command or tool inventory changed") + root = self.d.workspace(run) + if ( + str(root.resolve()) != candidate["workspace"]["path"] + or metadata_digest(root) != candidate["workspace"]["metadata"] + or _git(root, "rev-parse", "HEAD").decode().strip() != candidate["expected_head"] + or _git(root, "symbolic-ref", "--short", "HEAD").decode().strip() != run["branch"] + or verification_source(root) != candidate["source_digest"] + or (staged_tree is None and git_index_identity(root).sha256 != candidate["index_digest"]) + or (staged_tree is not None and _git(root, "write-tree").decode().strip() != staged_tree) + ): + raise ValueError("The owned Git/source frontier changed") + await self._authority(await self.store.get_run(run_id), actor) + + async def admit(self, run_id: str, request_id: str, actor: Data, body: Any) -> None: + from firefly_forge.git_diff_models import GitDiffBudget + from firefly_forge.source_snapshot import SourceDiffSnapshotV1 + + candidate = await self.candidate(run_id, actor) + self.compare(candidate, body) + run = await self.store.get_run(run_id) + settings = self.store.configuration.load_snapshot(run["config_revision"]) + budget = GitDiffBudget.model_validate(run["git_delivery_policy"].get("diff_budget") or {}) + _, source, tree = await self.d.engine._snapshot( + self.d.workspace(run), + candidate["expected_head"], + secrets=settings, + expected_head=candidate["expected_head"], + budget=budget, + ) + if not isinstance(source, SourceDiffSnapshotV1) or source.artifact.file_count == 0: + raise ValueError("Partial source requires exact nonempty native diff evidence") + async with self.d._owned(run_id, actor): + await self.revalidate(run_id, actor, candidate, staged_tree=tree) + event = await self.journal._append( + run_id, + "checkpoint", + request_id, + KIND, + PartialSourceAdmissionV1.model_validate( + dict( + version=1, + request_id=request_id, + actor=copy.deepcopy(actor), + request=body.model_dump(mode="json"), + candidate=candidate, + staged_tree=tree, + source_evidence=source.artifact.model_dump(mode="json"), + verification_status="partial_unverified", + ) + ).model_dump(mode="json"), + ) + await self.store.update( + run_id, git_partial_source_checkpoint={"request_id": request_id, "receipt_digest": event["digest"]} + ) + + async def guard(self, run: Data, actor: Data) -> None: + """Validate the owned staging/commit phase without reopening native graph history.""" + from firefly_forge.git_partial_source_models import GitPartialSourceConsentV1 + from firefly_forge.git_workspace import metadata_digest + from firefly_forge.verification_models import verification_source + from firefly_forge.workspace_reconciliation import _git + + marker = run.get("git_partial_source_checkpoint") or {} + entries = await self._entries(run["id"]) + matches = [ + e + for e in entries + if e["kind"] == KIND + and e["idempotency_key"] == marker.get("request_id") + and e["digest"] == marker.get("receipt_digest") + ] + if len(matches) != 1: + raise ValueError("Partial source lacks its exact server-owned settlement receipt") + admission = matches[0] + payload = PartialSourceAdmissionV1.model_validate(admission["payload"]).model_dump(mode="json") + candidate = payload["candidate"] + if payload["request_id"] != marker.get("request_id") or candidate["run_id"] != run["id"]: + raise ValueError("Partial source receipt belongs to another request or run") + self.compare(candidate, GitPartialSourceConsentV1.model_validate(payload["request"])) + if ( + candidate["digest"] != digest({k: v for k, v in candidate.items() if k != "digest"}) + or payload["actor"] != actor + or candidate["actor"] != actor + ): + raise ValueError("Partial source proof or original actor differs") + before = [e for e in entries if e["sequence"] < admission["sequence"]] + if ( + not before + or before[-1]["digest"] != candidate["journal_digest"] + or any( + e["kind"] + not in { + "git_delivery_intent", + "git_delivery_receipt", + "git_delivery_remote_intent", + "git_delivery_remote_receipt", + "git_delivery_observation", + "git_delivery_draft", + "git_delivery_hold", + } + or e["payload"].get("request_id") != payload["request_id"] + for e in entries + if e["sequence"] > admission["sequence"] + ) + ): + raise ValueError("A later effect changed the partial source custody") + receipts = self.d._records(entries) + expected_projection = receipts[-1]["revision"] if receipts else candidate["original_checkpoint_revision"] + if run.get("checkpoint_revision") != expected_projection or run.get("git_delivery_head") != ( + receipts[-1]["revision"] if receipts else None + ): + raise ValueError("Source delivery projections differ from their native checkpoint receipt") + # Projections are mutable only when their exact native delivery records account for them. + if len(receipts) > 1: + raise ValueError("Partial source has more than its one owned checkpoint receipt") + summaries: list[Any] = [None] + remote_intents = [e for e in entries if e["kind"] == "git_delivery_remote_intent"] + remote = self.d._remote_state(entries, allow_pending=True) + if receipts: + receipt = receipts[-1] + local = {key: receipt[key] for key in ("revision", "source_tree", "sequence", "remote_status", "pr_url")} + summaries.append(local) + if remote_intents: + if len(remote_intents) != 1 or any( + remote_intents[0]["payload"].get(key) != receipt[key] + for key in ("request_id", "revision", "checkpoint_digest") + ): + raise ValueError("Remote projection lacks its exact owned checkpoint intent") + summaries = [local, {**local, "remote_status": "publishing"}] + if remote: + if remote["remote_status"] != "draft" or remote["revision"] != receipt["revision"]: + raise ValueError("Partial source projection cannot claim final-ready publication") + summaries.append({**local, "remote_status": "draft", "pr_url": remote["reference"]["url"]}) + elif remote_intents or remote: + raise ValueError("Remote projection has no owned source checkpoint receipt") + if run.get("git_delivery_summary") not in summaries or run.get("git_delivery_draft") != ( + remote["reference"] if remote else None + ): + raise ValueError("Delivery projections differ from their exact native source receipts") + normalized = copy.deepcopy(run) + normalized["checkpoint_revision"] = candidate["original_checkpoint_revision"] + self._quiescent(run) + if self._run_cut(normalized) != candidate["expected_run_digest"]: + raise ValueError("Partial source must preserve its failed graph and original counters") + await self._authority(run, actor) + task = next(t for t in (await self.journal.plan(run["id"]))["tasks"] if t["id"] == "implement") + if {k: task.get(k) for k in candidate["proof"]["task"]} != candidate["proof"]["task"]: + raise ValueError("The failed task was changed by source settlement") + sequence = next(e["sequence"] for e in before if e["digest"] == candidate["frontier_record_digest"]) + if await self._effects(run, before, sequence) != candidate["effects_digest"]: + raise ValueError("Tool or command custody changed after partial-source admission") + root = self.d.workspace(run) + head = _git(root, "rev-parse", "HEAD").decode().strip() + if head != candidate["expected_head"] and ( + _git(root, "rev-parse", "HEAD^").decode().strip() != candidate["expected_head"] + or _git(root, "rev-parse", "HEAD^{tree}").decode().strip() != payload["staged_tree"] + ): + raise ValueError("Partial source committed a different parent or tree") + checks = { + "path": str(root.resolve()) == candidate["workspace"]["path"], + "metadata": metadata_digest(root) == candidate["workspace"]["metadata"], + "staged tree": _git(root, "write-tree").decode().strip() == payload["staged_tree"], + "branch": _git(root, "symbolic-ref", "--short", "HEAD").decode().strip() == run["branch"], + "source": verification_source(root) == candidate["source_digest"], + } + if not all(checks.values()): + raise ValueError( + "The admitted partial source changed during its Git effect: " + + ", ".join(k for k, v in checks.items() if not v) + ) + await self._authority(await self.store.get_run(run["id"]), actor) diff --git a/src/firefly_forge/git_partial_source_models.py b/src/firefly_forge/git_partial_source_models.py new file mode 100644 index 00000000..f6197d33 --- /dev/null +++ b/src/firefly_forge/git_partial_source_models.py @@ -0,0 +1,105 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Finite source-only consent preserves failed graph custody and original budgets.""" + +from typing import Annotated, Any, Literal + +from pydantic import ConfigDict, Field, field_validator + +from firefly_forge.git_diff_models import GitDiffRefV1 +from firefly_forge.scm_authority_models import Digest, NativeID, StrictContract + + +class GitPartialSourceConsentV1(StrictContract): + model_config = ConfigDict(extra="forbid", strict=True, frozen=True, hide_input_in_errors=True) + + version: Literal[1] + candidate_digest: Digest + policy_digest: Digest + expected_run_digest: Digest + journal_digest: Digest + source_digest: Digest + frontier_record_digest: Digest + acknowledge_partial_unverified: Literal[True] + acknowledge_preserve_failed_graph: Literal[True] + acknowledge_no_execution_grant: Literal[True] + + @field_validator( + "acknowledge_partial_unverified", + "acknowledge_preserve_failed_graph", + "acknowledge_no_execution_grant", + mode="before", + ) + @classmethod + def explicit_acknowledgment(cls, value: Any) -> bool: + if value is not True: + raise ValueError("Explicitly acknowledge partial source without graph execution") + return True + + +Revision = Annotated[str, Field(pattern=r"^(?:[a-f0-9]{40}|[a-f0-9]{64})$")] + + +class PartialSourceRecord(StrictContract): + model_config = ConfigDict(extra="forbid", strict=True, frozen=True, hide_input_in_errors=True) + + +class PartialSourceTaskV1(PartialSourceRecord): + id: Literal["implement"] + status: Literal["running", "needs_reconciliation"] + attempt: int = Field(ge=1, le=100) + idempotency_key: NativeID + + +class PartialSourceInvocationV1(PartialSourceRecord): + invocation_id: NativeID + ordinal: int = Field(ge=1, le=100) + predecessor: Digest | None + round: int = Field(ge=1, le=1000000) + grants: list[str] = Field(min_length=1, max_length=9) + contract: Digest + run_cut: Digest + workspace: dict[str, str] = Field(min_length=4, max_length=4) + + +class PartialSourceFrontierV1(PartialSourceRecord): + version: Literal[1] + kind: Literal["git_partial_source_checkpoint"] + frontier_record_digest: Digest + frontier_checkpoint_digest: Digest + invocation: PartialSourceInvocationV1 + invocation_digest: Digest + completion_count: int = Field(ge=1, le=100) + closure_digest: Digest + journal_head: Digest + later_tool_names: list[Literal["list_files", "read_file", "search_text", "git_diff"]] + task: PartialSourceTaskV1 + verification_status: Literal["partial_unverified"] + + +class PartialSourceCandidateV1(PartialSourceRecord): + version: Literal[1] + run_id: NativeID + actor: dict[str, Any] + policy_digest: Digest + expected_run_digest: Digest + journal_digest: Digest + source_digest: Digest + index_digest: Digest + expected_head: Revision + workspace: dict[str, str] = Field(min_length=4, max_length=4) + frontier_record_digest: Digest + proof: PartialSourceFrontierV1 + effects_digest: Digest + original_checkpoint_revision: Revision | None + digest: Digest + + +class PartialSourceAdmissionV1(PartialSourceRecord): + version: Literal[1] + request_id: NativeID + actor: dict[str, Any] + request: GitPartialSourceConsentV1 + candidate: PartialSourceCandidateV1 + staged_tree: Revision + source_evidence: GitDiffRefV1 + verification_status: Literal["partial_unverified"] diff --git a/tests/test_git_partial_source_clients.py b/tests/test_git_partial_source_clients.py new file mode 100644 index 00000000..1cb313aa --- /dev/null +++ b/tests/test_git_partial_source_clients.py @@ -0,0 +1,42 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""The SDK reviews source custody and submits one explicit request without retry.""" + +import json + +import httpx +import pytest +from test_git_partial_source_models import consent +from test_original_recovery_sdk import connection + +RUN = "a" * 32 +BODY = dict( + version=1, + request_id="b" * 32, + expected_head="c" * 40, + expected_source_digest="e" * 64, + explicit_partial_source=consent(), +) + + +def test_partial_source_sdk_uses_exact_typed_paths_and_body(): + calls = [] + with connection(lambda r: calls.append(r) or httpx.Response(200, json={"retained": True})) as client: + client.partial_source_candidate(RUN) + client.request_partial_source_checkpoint(RUN, BODY) + assert [(r.method, r.url.path) for r in calls] == [ + ("GET", f"/api/v1/runs/{RUN}/git-deliveries/partial-source-candidate"), + ("POST", f"/api/v1/runs/{RUN}/git-deliveries/partial-source-requests"), + ] + assert json.loads(calls[1].content) == BODY + + +def test_unknown_partial_source_sdk_outcome_is_not_automatically_retried(): + calls = [] + + def lost(request): + calls.append(request) + raise httpx.ReadTimeout("Controlled source-only fixture", request=request) + + with connection(lost) as client, pytest.raises(httpx.ReadTimeout): + client.request_partial_source_checkpoint(RUN, BODY) + assert len(calls) == 1 diff --git a/tests/test_git_partial_source_effects_host.py b/tests/test_git_partial_source_effects_host.py new file mode 100644 index 00000000..0ceb93fa --- /dev/null +++ b/tests/test_git_partial_source_effects_host.py @@ -0,0 +1,54 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""The real command slot/WAL validator never dispatches a prior command again.""" + +import copy +from types import SimpleNamespace + +import pytest +from test_sandbox_command_ordinals_host import command + +from firefly_forge.git_partial_source_checkpoints import GitPartialSourceCheckpoints +from firefly_forge.sandbox import SandboxReconciliationRequired + +pytest_plugins = ["test_sandbox_command_ordinals_host"] + + +async def effect_case(case): + await command(case, ["known-command"]) + + async def empty(_): + return [] + + service = GitPartialSourceCheckpoints( + SimpleNamespace(store=case.store, journal=case.journal), tool_records=SimpleNamespace(for_run=empty) + ) + run = await case.store.get_run(case.run_id) + entries = [copy.deepcopy(r.data) for r in case.store.records.rows] + return service, run, entries, max(e["sequence"] for e in entries) + 1 + + +async def test_completed_command_is_attested_without_dispatch_or_counter_reset(command_case): + c = command_case + service, run, entries, frontier = await effect_case(c) + observed = await service._effects(run, entries, frontier) + assert await service._effects(run, entries, frontier) == observed + assert c.calls == [["known-command"]] + assert [r.ordinal for r in c.store.sandboxes.rows] == [1] + + +@pytest.mark.parametrize("change", ["unknown", "interrupted", "orphan", "late"]) +async def test_unknown_or_later_command_custody_does_not_create_source_authority(command_case, change): + c = command_case + service, run, entries, frontier = await effect_case(c) + row = c.store.sandboxes.rows[0] + if change == "unknown": + row.data["effects"]["start"]["status"] = "intent" + elif change == "interrupted": + row.data["effects"]["start"]["evidence"]["outcome"] = "interrupted" + elif change == "orphan": + c.store.sandboxes.rows.clear() + else: + frontier = entries[-1]["sequence"] + with pytest.raises((ValueError, SandboxReconciliationRequired)): + await service._effects(run, entries, frontier) + assert c.calls == [["known-command"]] diff --git a/tests/test_git_partial_source_frontier_host.py b/tests/test_git_partial_source_frontier_host.py new file mode 100644 index 00000000..315c2af4 --- /dev/null +++ b/tests/test_git_partial_source_frontier_host.py @@ -0,0 +1,293 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Real native consumed turns distinguish a stored source frontier from graph reentry.""" + +import copy +import json + +import httpx2 +import pytest +from pydantic_ai import models +from test_agentic_checkpoint_activation_host import activated, result +from test_bridge_completion_callbacks import CAPABILITY +from test_bridge_completion_native import mock_model_requests +from test_bridge_inflight_cancel import RUN + + +@pytest.fixture +async def frontier_case(tmp_path, monkeypatch): + from firefly_forge.agent_roles import ProjectRoles, RoleRegistry + from firefly_forge.agentic_checkpoint_boundary import AgenticBoundary + + env = await activated(tmp_path, monkeypatch) + env.store.run.data["role_snapshot"] = RoleRegistry(tmp_path / "frontier-roles").snapshot( + ProjectRoles(disabled=["docs"], allowed_tools={"implementer": ["write_file", "read_file"]}) + ) + original = AgenticBoundary.capture + captures = 0 + posts = [] + + def capture(boundary, wire): + nonlocal captures + captures += 1 + if captures == 3: + raise ValueError("Fictional checkpoint storage refusal before dispatch") + return original(boundary, wire) + + monkeypatch.setattr(AgenticBoundary, "capture", capture) + + async def run(*, late_write=False): + def transport(request): + if request.method == "GET": + return httpx2.Response(200, json={"completionRecovery": CAPABILITY}) + posts.append(request.content) + assert len(posts) <= 2 + body = result(tool=True) + if len(posts) == 2: + call = body["choices"][0]["message"]["tool_calls"][0] + call["id"] = "second-call" + if late_write: + call["function"]["arguments"] = json.dumps({"path": "service.py", "content": "VALUE = 2\n"}) + else: + call["function"] = {"name": "read_file", "arguments": json.dumps({"path": "service.py"})} + return httpx2.Response(200, json=body) + + env.model.client._client._transport = httpx2.MockTransport(transport) + env.model.client._client._mounts = {} + with models.override_allow_model_requests(False), mock_model_requests(env.model): + await env.engine.execute_run(RUN) + current = await env.store.get_run(RUN) + assert current["status"] == "failed", current.get("error") + assert len(posts) == 2 + entries = [copy.deepcopy(row.data) for row in env.journal.records.rows] + task = next(t for t in (await env.journal.plan(RUN))["tasks"] if t["id"] == "implement") + return current, task, entries + + env.run_frontier = run + yield env + await env.model.client.close() + + +async def test_stored_frontier_proof_keeps_failed_graph_and_all_consumes(frontier_case): + from firefly_forge.git_partial_source_checkpoints import stored_frontier_proof + + run, task, entries = await frontier_case.run_frontier() + before = copy.deepcopy((run, task, entries)) + proof = stored_frontier_proof(run, task, entries) + assert proof["verification_status"] == "partial_unverified" + assert proof["completion_count"] == 2 + assert proof["frontier_record_digest"] == next( + e["digest"] for e in reversed(entries) if e["kind"] == "agent_checkpoint" + ) + assert proof["later_tool_names"] == ["read_file"] + assert "brief" not in proof["invocation"] + assert (run, task, entries) == before + assert run["implementation_pending"] is True and task["status"] == "running" + + +async def test_mutation_after_frontier_never_authorizes_partial_checkpoint(frontier_case): + from firefly_forge.git_partial_source_checkpoints import stored_frontier_proof + + run, task, entries = await frontier_case.run_frontier(late_write=True) + with pytest.raises(ValueError, match="read-only"): + stored_frontier_proof(run, task, entries) + + +async def test_known_response_without_consume_is_still_not_settled(frontier_case): + from firefly_forge.git_partial_source_checkpoints import stored_frontier_proof + + run, task, entries = await frontier_case.run_frontier() + assert entries[-1]["kind"] == "agent_consume" + with pytest.raises(ValueError, match="completion closure"): + stored_frontier_proof(run, task, entries[:-1]) + + +async def test_old_runtime_contract_is_not_accepted_by_a_current_reader(frontier_case, monkeypatch): + import firefly_forge.git_partial_source_checkpoints as partial + + run, task, entries = await frontier_case.run_frontier() + monkeypatch.setattr(partial, "runtime_contract", lambda: "0" * 64) + with pytest.raises(ValueError, match="runtime contract"): + partial.stored_frontier_proof(run, task, entries) + + +@pytest.fixture +async def partial_service(frontier_case, monkeypatch, request): + from types import SimpleNamespace + + from firefly_forge.git_deliveries import GitDeliveries + from firefly_forge.git_delivery_policy import GitDeliveryPolicies + from firefly_forge.git_partial_source_checkpoints import GitPartialSourceCheckpoints + + env = frontier_case + if getattr(request, "param", None) == "compressed": + env.store.run.data["brief"] = "Controlled source custody context. " * 1200 + policy = dict(version=1, enabled=True, every_units=1, on_pause=True, remote="local", max_checkpoints=50) + if getattr(request, "param", None) == "draft": + policy["remote"] = "draft_pr" + from test_git_delivery_local import git + + env.store.run.data.update( + git_delivery_policy=policy, git_delivery_policy_digest="b" * 64, branch="forge/" + RUN[:12] + ) + git(env.workspace, "checkout", "-qb", "forge/" + RUN[:12]) + if getattr(request, "param", None) == "draft": + from firefly_forge.git_workspace import metadata_digest + + env.partial_remote = env.workspace.parent / "partial-remote.git" + env.partial_remote.mkdir() + git(env.partial_remote, "init", "--bare", "-q") + git(env.workspace, "remote", "add", "origin", str(env.partial_remote)) + env.store.run.data["git_metadata_digest"] = metadata_digest(env.workspace) + await env.run_frontier() + state = {"allowed": True} + + async def authorize(run, actor, permission): + assert run["id"] == RUN and actor == env.store.run.data["execution_actor"] + if not state["allowed"]: + raise PermissionError("Controlled current authority revocation") + + async def policy_snapshot(_self, _project): + return {"policy": copy.deepcopy(policy), "digest": "b" * 64} + + async def empty(_): + return [] + + monkeypatch.setattr(GitDeliveryPolicies, "snapshot", policy_snapshot) + env.journal.sandboxes = SimpleNamespace(for_run=empty) + env.store.source_budget_admission_unit = env.store._unit + from firefly_forge.tool_persistence import ForgeToolInvocationRepository + + monkeypatch.setattr(ForgeToolInvocationRepository, "for_run", lambda _self, run_id: empty(run_id)) + deliveries = GitDeliveries( + env.store, env.engine, env.journal, workspace=lambda _: env.workspace, authorize=authorize + ) + service = GitPartialSourceCheckpoints(deliveries, tool_records=SimpleNamespace(for_run=empty)) + env.partial, env.partial_state = service, state + return env + + +async def test_native_candidate_compares_actual_source_with_its_stored_frontier(partial_service): + env = partial_service + old = copy.deepcopy(await env.store.get_run(RUN)) + candidate = await env.partial.candidate(RUN, old["execution_actor"]) + assert candidate["workspace"] == candidate["proof"]["invocation"]["workspace"] + assert candidate["proof"]["later_tool_names"] == ["read_file"] + assert await env.store.get_run(RUN) == old + + +async def test_candidate_source_drift_and_late_revocation_are_not_bypassed(partial_service): + env = partial_service + actor = env.store.run.data["execution_actor"] + source = env.workspace / "service.py" + original = source.read_bytes() + source.write_text("VALUE = 99\n") + with pytest.raises(ValueError, match="Current source differs"): + await env.partial.candidate(RUN, actor) + source.write_bytes(original) + env.partial_state["allowed"] = False + with pytest.raises(PermissionError, match="revocation"): + await env.partial.candidate(RUN, actor) + + +async def test_owner_commits_real_partial_git_without_clearing_failed_graph(partial_service): + from test_git_delivery_local import git + + from firefly_forge.git_partial_source_models import GitPartialSourceConsentV1 + + env = partial_service + parent = copy.deepcopy(await env.store.get_run(RUN)) + actor = parent["execution_actor"] + candidate = await env.partial.candidate(RUN, actor) + consent = GitPartialSourceConsentV1.model_validate( + { + "version": 1, + "candidate_digest": candidate["digest"], + **{ + key: candidate[key] + for key in [ + "policy_digest", + "expected_run_digest", + "journal_digest", + "source_digest", + "frontier_record_digest", + ] + }, + "acknowledge_partial_unverified": True, + "acknowledge_preserve_failed_graph": True, + "acknowledge_no_execution_grant": True, + } + ) + checkpoint = env.partial.d + await env.partial.admit(RUN, "a" * 32, actor, consent) + receipt = await checkpoint.checkpoint( + RUN, + request_id="a" * 32, + expected_head=candidate["expected_head"], + expected_source_digest=candidate["source_digest"], + actor=actor, + ) + assert receipt["verification_status"] == "partial_unverified" + assert git(env.workspace, "log", "-1", "--format=%s").startswith("[UNVERIFIED]") + assert receipt["parent_revision"] == candidate["expected_head"] + assert git(env.workspace, "show", "HEAD:service.py") == "VALUE = 1" + current = await env.store.get_run(RUN) + for key in ( + "status", + "active_task", + "implementation_pending", + "bridge_completion_hold", + "implementation_rounds", + "attempt", + ): + assert current.get(key) == parent.get(key) + assert current["checkpoint_revision"] == receipt["revision"] + count = len(env.journal.records.rows) + repeated = await checkpoint.checkpoint( + RUN, + request_id="a" * 32, + expected_head=candidate["expected_head"], + expected_source_digest=candidate["source_digest"], + actor=actor, + ) + assert repeated["checkpoint_digest"] == receipt["checkpoint_digest"] + assert repeated["revision"] == receipt["revision"] == git(env.workspace, "rev-parse", "HEAD") + assert len(env.journal.records.rows) == count + with pytest.raises(ValueError, match="one unverified checkpoint"): + await checkpoint.checkpoint(RUN, request_id="b" * 32, expected_head=receipt["revision"], actor=actor) + with pytest.raises(ValueError, match="no final-ready"): + await checkpoint.checkpoint( + RUN, request_id="a" * 32, expected_head=receipt["revision"], actor=actor, final=True + ) + + +async def test_initial_index_change_is_refused_instead_of_treated_as_owned_staging(partial_service): + from test_git_delivery_local import git + + env = partial_service + actor = env.store.run.data["execution_actor"] + candidate = await env.partial.candidate(RUN, actor) + git(env.workspace, "add", "service.py") + with pytest.raises(ValueError, match="frontier changed"): + await env.partial.revalidate(RUN, actor, candidate, staged_tree=None) + + +@pytest.mark.parametrize("partial_service", ["compressed"], indirect=True) +async def test_decoded_checkpoint_documents_use_real_current_credential_guard(partial_service): + env = partial_service + canary = "Wrote service.py (10 bytes)" + entries = await env.partial._entries(RUN) + checkpoints = [e for e in entries if e["kind"] == "agent_checkpoint"] + assert json.loads(checkpoints[-1]["payload"]["snapshot"])["version"] == 3 + assert canary not in json.dumps(entries) + env.store.credential_registry = lambda: {"TOKEN": canary} + with pytest.raises(ValueError, match="configured or historical credential"): + await env.partial.candidate(RUN, env.store.run.data["execution_actor"]) + + +async def test_hold_pointer_must_identify_the_original_native_completion(frontier_case): + from firefly_forge.git_partial_source_checkpoints import stored_frontier_proof + + run, task, entries = await frontier_case.run_frontier() + run["bridge_completion_hold"]["intent_digest"] = "0" * 64 + with pytest.raises(ValueError, match="original native completion"): + stored_frontier_proof(run, task, entries) diff --git a/tests/test_git_partial_source_models.py b/tests/test_git_partial_source_models.py new file mode 100644 index 00000000..3780cf85 --- /dev/null +++ b/tests/test_git_partial_source_models.py @@ -0,0 +1,47 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""A partial source checkpoint is explicit source custody, never execution consent.""" + +import pytest + + +def consent(): + return dict( + version=1, + candidate_digest="a" * 64, + policy_digest="b" * 64, + expected_run_digest="c" * 64, + journal_digest="d" * 64, + source_digest="e" * 64, + frontier_record_digest="f" * 64, + acknowledge_partial_unverified=True, + acknowledge_preserve_failed_graph=True, + acknowledge_no_execution_grant=True, + ) + + +def test_partial_consent_is_exclusive_and_legacy_requests_do_not_change(): + from firefly_forge.git_delivery_models import GitDeliveryInputV1 + + old = dict(version=1, request_id="1" * 32, expected_head="2" * 40, expected_source_tree="3" * 40) + assert GitDeliveryInputV1.model_validate(old).model_dump(mode="json") == old + new = dict( + version=1, + request_id="1" * 32, + expected_head="2" * 40, + expected_source_digest="e" * 64, + explicit_partial_source=consent(), + ) + assert GitDeliveryInputV1.model_validate(new).model_dump(mode="json") == new + with pytest.raises(ValueError): + GitDeliveryInputV1.model_validate({**new, "explicit_partial_source": {**consent(), "execute": True}}) + + +@pytest.mark.parametrize("value", [False, 1, "true", None]) +@pytest.mark.parametrize( + "field", ["acknowledge_partial_unverified", "acknowledge_preserve_failed_graph", "acknowledge_no_execution_grant"] +) +def test_partial_consent_requires_exact_acknowledgments(field, value): + from firefly_forge.git_partial_source_models import GitPartialSourceConsentV1 + + with pytest.raises(ValueError): + GitPartialSourceConsentV1.model_validate({**consent(), field: value}) diff --git a/tests/test_git_partial_source_native.py b/tests/test_git_partial_source_native.py new file mode 100644 index 00000000..a596c351 --- /dev/null +++ b/tests/test_git_partial_source_native.py @@ -0,0 +1,319 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Actual PostgreSQL/current authority/real Git; model HTTP stays in memory.""" + +import asyncio +import copy +import json +from contextlib import asynccontextmanager +from contextvars import Context +from types import SimpleNamespace +from uuid import uuid4 + +import httpx2 +import pytest +from pydantic_ai import models +from test_agentic_checkpoint_activation_host import result +from test_bridge_completion_callbacks import CAPABILITY +from test_bridge_completion_native import mock_model_requests +from test_git_delivery_native import source_project +from worker_reply_fixtures import authenticated_operator + +from firefly_forge.agent_roles import ProjectRoles, RoleRegistry +from firefly_forge.agentic_checkpoint_boundary import AgenticBoundary +from firefly_forge.configuration import ProviderConfiguration +from firefly_forge.engine import Engine +from firefly_forge.execution_requests import ExecutionRequests +from firefly_forge.git_deliveries import GitDeliveries +from firefly_forge.git_delivery_controllers import GitDeliveryController +from firefly_forge.git_delivery_models import GitDeliveryInputV1, GitDeliveryPolicyInputV1 +from firefly_forge.git_delivery_policy import GitDeliveryPolicies +from firefly_forge.git_delivery_requests import GitDeliveryRequests +from firefly_forge.identity_persistence import ForgeIdentityUserRepository +from firefly_forge.journal import Journal +from firefly_forge.models import RunInput +from firefly_forge.providers import BridgeProvider +from firefly_forge.runtime_roles import RuntimeRoles +from firefly_forge.scm import checked +from firefly_forge.store import identifier + + +@asynccontextmanager +async def native_partial(store, root, monkeypatch): + async with authenticated_operator(store): + runtime = RuntimeRoles(store, "combined") + store.runtime_roles = runtime + await runtime.start() + await store.acquire_ownership() + model = None + try: + repository, project = await source_project(store, root) + await GitDeliveryPolicies(store).configure( + project["id"], + GitDeliveryPolicyInputV1( + version=1, + request_id=uuid4().hex, + expected_previous_digest=None, + policy={"enabled": True, "remote": "local"}, + ), + ) + store.configuration.update( + "bridge", + ProviderConfiguration( + values={ + "endpoint": "http://127.0.0.1:19191", + "token": "fictional-partial-source-token", + "completion_recovery": "required", + "checkpoints": "required", + } + ), + ) + run = await store.create_run( + RunInput( + repository_id=repository["id"], + project_id=project["id"], + component_id="core", + agent="bridge", + frontend="none", + skills=[], + brief="Retain a public unverified source frontier.", + ) + ) + await store.update( + run["id"], + repair_budget=0, + role_snapshot=RoleRegistry(root / "roles").snapshot( + ProjectRoles(disabled=["docs"], allowed_tools={"implementer": ["write_file", "read_file"]}) + ), + ) + journal = Journal(store) + provider = BridgeProvider( + store.configuration.load_snapshot((await store.get_run(run["id"]))["config_revision"]) + ) + model = provider.model() + monkeypatch.setattr(provider, "model", lambda **kwargs: model) + engine = Engine(store, root / "workspaces", lambda _: provider, journal=journal) + engine.enforce_ownership = True + delivery = GitDeliveries(store, engine, journal) + engine.git_deliveries = delivery + captures, posts = 0, [] + original = AgenticBoundary.capture + + def capture(boundary, wire): + nonlocal captures + captures += 1 + if captures == 3: + raise ValueError("Controlled pre-dispatch checkpoint refusal") + return original(boundary, wire) + + monkeypatch.setattr(AgenticBoundary, "capture", capture) + + def transport(request): + if request.method == "GET": + return httpx2.Response(200, json={"completionRecovery": CAPABILITY}) + posts.append(request.content) + assert len(posts) <= 2 + body = result(tool=True) + call = body["choices"][0]["message"]["tool_calls"][0] + call["id"] = "partial-" + str(len(posts)) + call["function"] = ( + {"name": "write_file", "arguments": json.dumps({"path": "product.py", "content": "value = 2\n"})} + if len(posts) == 1 + else {"name": "read_file", "arguments": json.dumps({"path": "product.py"})} + ) + return httpx2.Response(200, json=body) + + model.client._client._transport = httpx2.MockTransport(transport) + model.client._client._mounts = {} + engine.active[run["id"]] = asyncio.current_task() + try: + with models.override_allow_model_requests(False), mock_model_requests(model): + await engine.execute_run(run["id"]) + finally: + engine.active.pop(run["id"], None) + failed = await store.get_run(run["id"]) + assert failed["status"] == "failed", failed.get("error") + assert failed["implementation_pending"] and failed["bridge_completion_hold"] + entries = await journal.entries(run["id"]) + assert sum(e["kind"] == "agent_consume" for e in entries) == len(posts) == 2 + requests = ExecutionRequests(store, runtime) + tickets = GitDeliveryRequests(delivery, requests) + controller = GitDeliveryController(delivery, tickets) + candidate = await controller.partial_source_candidate(run["id"]) + body = GitDeliveryInputV1.model_validate( + dict( + 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"], + **{ + key: candidate[key] + for key in [ + "policy_digest", + "expected_run_digest", + "journal_digest", + "source_digest", + "frontier_record_digest", + ] + }, + "acknowledge_partial_unverified": True, + "acknowledge_preserve_failed_graph": True, + "acknowledge_no_execution_grant": True, + }, + ) + ) + yield SimpleNamespace( + store=store, + engine=engine, + journal=journal, + delivery=delivery, + tickets=tickets, + controller=controller, + run_id=run["id"], + workspace=engine.workspaces / run["id"], + candidate=candidate, + body=body, + parent=copy.deepcopy(failed), + prefix=copy.deepcopy(entries), + posts=posts, + ) + finally: + if model is not None: + await model.client.close() + await runtime.stop() + await store.release_ownership() + + +async def test_native_partial_source_queue_commits_without_resuming_or_resetting_parent(store, tmp_path, monkeypatch): + async with native_partial(store, tmp_path, monkeypatch) as c: + queued = await c.controller.partial_source_request(c.run_id, c.body) + assert queued["status"] == "queued" + await asyncio.create_task(c.tickets.execute(c.body.request_id), context=Context()) + receipt = await c.controller.ticket(c.run_id, c.body.request_id) + assert receipt["status"] == "completed", receipt.get("result") + assert receipt["result"]["verification_status"] == "partial_unverified" + assert await checked(["git", "show", "HEAD:product.py"], c.workspace) == "value = 2" + assert (await checked(["git", "log", "-1", "--format=%s"], c.workspace)).startswith("[UNVERIFIED]") + after = await store.get_run(c.run_id) + for key in [ + "status", + "active_task", + "implementation_pending", + "bridge_completion_hold", + "implementation_rounds", + "attempt", + ]: + assert after.get(key) == c.parent.get(key) + entries = await c.journal.entries(c.run_id) + assert entries[: len(c.prefix)] == c.prefix and c.journal.verify_chain(entries) + assert sum(e["kind"] == "agent_consume" for e in entries) == len(c.posts) == 2 + assert await c.controller.partial_source_request(c.run_id, c.body) == receipt + await c.tickets.execute(c.body.request_id) + assert await c.controller.ticket(c.run_id, c.body.request_id) == receipt + + +@pytest.mark.parametrize("drift", ["source", "authority", "ownership"]) +async def test_native_partial_source_late_change_has_no_git_effect(store, tmp_path, monkeypatch, drift): + async with native_partial(store, tmp_path, monkeypatch) as c: + await c.controller.partial_source_request(c.run_id, c.body) + if drift == "source": + (c.workspace / "product.py").write_text("value = 99\n") + elif drift == "authority": + async with store._unit(): + repo = ForgeIdentityUserRepository() + user = await repo.find_by_id(identifier(c.candidate["actor"]["user_id"])) + user.enabled = False + await repo.save(user) + else: + await store.release_ownership() + work = asyncio.create_task(c.tickets.execute(c.body.request_id), context=Context()) + if drift == "ownership": + with pytest.raises(RuntimeError, match="ownership"): + await work + else: + await work + assert await checked(["git", "rev-parse", "HEAD"], c.workspace) == c.candidate["expected_head"] + async with store._unit(): + rows = await c.journal.records.find_for_run(identifier(c.run_id)) + assert not any(r.kind in {"git_partial_source_checkpoint", "git_delivery_intent"} for r in rows) + + +async def test_native_partial_admission_receipt_and_marker_roll_back_together(store, tmp_path, monkeypatch): + async with native_partial(store, tmp_path, monkeypatch) as c: + await c.controller.partial_source_request(c.run_id, c.body) + append = c.journal._append + + async def fail(run_id, task, key, kind, payload): + result = await append(run_id, task, key, kind, payload) + if kind == "git_partial_source_checkpoint": + raise RuntimeError("Controlled native refusal after partial receipt append") + return result + + monkeypatch.setattr(c.journal, "_append", fail) + await asyncio.create_task(c.tickets.execute(c.body.request_id), context=Context()) + after = await store.get_run(c.run_id) + assert not after.get("git_partial_source_checkpoint") + assert await c.journal.entries(c.run_id) == c.prefix + assert await checked(["git", "rev-parse", "HEAD"], c.workspace) == c.candidate["expected_head"] + assert (await c.controller.ticket(c.run_id, c.body.request_id))["status"] == "failed" + + +async def test_native_partial_unknown_commit_keeps_intent_and_does_not_reexecute(store, tmp_path, monkeypatch): + import firefly_forge.git_deliveries as module + + async with native_partial(store, tmp_path, monkeypatch) as c: + await c.controller.partial_source_request(c.run_id, c.body) + command = module.checked + + async def lost(argv, *args, **kwargs): + result = await command(argv, *args, **kwargs) + if "commit" in argv: + raise OSError("Controlled lost response after actual native Git commit") + return result + + monkeypatch.setattr(module, "checked", lost) + await asyncio.create_task(c.tickets.execute(c.body.request_id), context=Context()) + receipt = await c.controller.ticket(c.run_id, c.body.request_id) + assert receipt["status"] == "needs_reconciliation" + head = await checked(["git", "rev-parse", "HEAD"], c.workspace) + assert head != c.candidate["expected_head"] + assert await checked(["git", "rev-parse", "HEAD^"], c.workspace) == c.candidate["expected_head"] + assert await c.controller.partial_source_request(c.run_id, c.body) == receipt + await c.tickets.execute(c.body.request_id) + assert await checked(["git", "rev-parse", "HEAD"], c.workspace) == head + async with store._unit(): + rows = await c.journal.records.find_for_run(identifier(c.run_id)) + assert sum(r.kind == "git_delivery_intent" for r in rows) == 1 + assert not any(r.kind == "git_delivery_receipt" for r in rows) + + +async def test_native_partial_delivery_projection_requires_exact_retained_receipts(store, tmp_path, monkeypatch): + from firefly_forge.git_partial_source_checkpoints import GitPartialSourceCheckpoints + + async with native_partial(store, tmp_path, monkeypatch) as c: + await c.controller.partial_source_request(c.run_id, c.body) + await asyncio.create_task(c.tickets.execute(c.body.request_id), context=Context()) + receipt = await c.controller.ticket(c.run_id, c.body.request_id) + assert receipt["status"] == "completed", receipt.get("result") + current = await store.get_run(c.run_id) + original = copy.deepcopy(current["git_delivery_summary"]) + service = GitPartialSourceCheckpoints(c.delivery) + await service.guard(current, current["execution_actor"]) + await store.update(c.run_id, git_delivery_summary={**original, "revision": "f" * 40}) + with pytest.raises(ValueError, match="projection"): + await service.guard(await store.get_run(c.run_id), current["execution_actor"]) + await store.update( + c.run_id, git_delivery_summary=original, git_delivery_draft={"url": "https://example.test/forged"} + ) + with pytest.raises(ValueError, match="projection"): + await service.guard(await store.get_run(c.run_id), current["execution_actor"]) + await store.update(c.run_id, git_delivery_draft=None) + await service.guard(await store.get_run(c.run_id), current["execution_actor"]) + after = await store.get_run(c.run_id) + assert ( + after["status"] == c.parent["status"] + and after["bridge_completion_hold"] == c.parent["bridge_completion_hold"] + ) + assert sum(e["kind"] == "agent_consume" for e in await c.journal.entries(c.run_id)) == len(c.posts) == 2 diff --git a/tests/test_git_partial_source_requests_host.py b/tests/test_git_partial_source_requests_host.py new file mode 100644 index 00000000..cbbe965e --- /dev/null +++ b/tests/test_git_partial_source_requests_host.py @@ -0,0 +1,309 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Actual controller/owner queue and real Git; memory locks do not claim PostgreSQL proof.""" + +import copy +from uuid import UUID + +import pytest +from test_bridge_inflight_cancel import RUN +from test_git_delivery_local import git + +from firefly_forge.git_delivery_controllers import GitDeliveryController +from firefly_forge.git_delivery_models import GitDeliveryInputV1 +from firefly_forge.git_delivery_requests import GitDeliveryRequests +from firefly_forge.journal import Journal +from firefly_forge.project_access import CURRENT_PRINCIPAL + +pytest_plugins = ["test_git_partial_source_frontier_host"] + + +async def queue(env): + controller = GitDeliveryController(env.partial.d, GitDeliveryRequests(env.partial.d, env.requests)) + candidate = await controller.partial_source_candidate(RUN) + body = GitDeliveryInputV1.model_validate( + dict( + version=1, + request_id="a" * 32, + expected_head=candidate["expected_head"], + expected_source_digest=candidate["source_digest"], + explicit_partial_source={ + "version": 1, + "candidate_digest": candidate["digest"], + **{ + key: candidate[key] + for key in [ + "policy_digest", + "expected_run_digest", + "journal_digest", + "source_digest", + "frontier_record_digest", + ] + }, + "acknowledge_partial_unverified": True, + "acknowledge_preserve_failed_graph": True, + "acknowledge_no_execution_grant": True, + }, + ) + ) + receipt = await controller.partial_source_request(RUN, body) + assert receipt["status"] == "queued" and git(env.workspace, "rev-parse", "HEAD") == candidate["expected_head"] + return controller, body, receipt + + +async def test_partial_source_api_queues_once_and_worker_returns_exact_retained_receipt(partial_service): + env = partial_service + token = CURRENT_PRINCIPAL.set(env.principal) + try: + before = copy.deepcopy(await env.store.get_run(RUN)) + controller, body, admitted = await queue(env) + await controller.tickets.execute(body.request_id) + final = await controller.ticket(RUN, body.request_id) + assert final["status"] == "completed", final.get("result") + assert final["result"]["verification_status"] == "partial_unverified" + assert await controller.partial_source_request(RUN, body) == final + assert len(env.requests.records.rows) == 1 + count = len(env.journal.records.rows) + head = git(env.workspace, "rev-parse", "HEAD") + await controller.tickets.execute(body.request_id) + assert git(env.workspace, "rev-parse", "HEAD") == head + assert len(env.journal.records.rows) == count + current = await env.store.get_run(RUN) + for key in [ + "status", + "active_task", + "implementation_pending", + "bridge_completion_hold", + "implementation_rounds", + "attempt", + ]: + assert current.get(key) == before.get(key) + changed = body.model_copy(update={"expected_head": "f" * 40}) + with pytest.raises(ValueError, match="another input"): + await controller.partial_source_request(RUN, changed) + finally: + CURRENT_PRINCIPAL.reset(token) + + +async def test_partial_source_late_revocation_preserves_uncommitted_source(partial_service): + env = partial_service + token = CURRENT_PRINCIPAL.set(env.principal) + try: + controller, body, _ = await queue(env) + head = git(env.workspace, "rev-parse", "HEAD") + env.partial_state["allowed"] = False + await controller.tickets.execute(body.request_id) + row = await env.requests.records.find_by_id(UUID(body.request_id)) + assert row.status == "failed" + assert git(env.workspace, "rev-parse", "HEAD") == head + assert not any(r.kind == "git_delivery_intent" for r in env.journal.records.rows) + finally: + CURRENT_PRINCIPAL.reset(token) + + +async def test_crash_after_real_git_commit_keeps_unknown_ticket_and_never_replays(partial_service, monkeypatch): + env = partial_service + token = CURRENT_PRINCIPAL.set(env.principal) + append = Journal.git_delivery_append + + async def refuse(journal, run_id, kind, key, payload): + if kind == "git_delivery_receipt": + raise RuntimeError("Controlled receipt storage failure after actual Git commit") + return await append(journal, run_id, kind, key, payload) + + try: + controller, body, _ = await queue(env) + head = git(env.workspace, "rev-parse", "HEAD") + monkeypatch.setattr(Journal, "git_delivery_append", refuse) + await controller.tickets.execute(body.request_id) + final = await controller.ticket(RUN, body.request_id) + assert final["status"] == "needs_reconciliation" + after = git(env.workspace, "rev-parse", "HEAD") + assert after != head and git(env.workspace, "rev-parse", "HEAD^") == head + assert await controller.partial_source_request(RUN, body) == final + await controller.tickets.execute(body.request_id) + assert git(env.workspace, "rev-parse", "HEAD") == after + assert sum(r.kind == "git_delivery_intent" for r in env.journal.records.rows) == 1 + assert not any(r.kind == "git_delivery_receipt" for r in env.journal.records.rows) + finally: + CURRENT_PRINCIPAL.reset(token) + + +async def test_partial_owner_uses_native_custody_rows_not_aggregate_wal_projection(partial_service, monkeypatch): + env = partial_service + token = CURRENT_PRINCIPAL.set(env.principal) + try: + controller, body, _ = await queue(env) + + async def aggregate_refusal(*args, **kwargs): + raise AssertionError("Partial Git custody must not use aggregate protected-content projection") + + monkeypatch.setattr(env.journal, "entries", aggregate_refusal) + await controller.tickets.execute(body.request_id) + receipt = await controller.ticket(RUN, body.request_id) + assert receipt["status"] == "completed", receipt.get("result") + assert receipt["result"]["verification_status"] == "partial_unverified" + finally: + CURRENT_PRINCIPAL.reset(token) + + +async def test_owned_activity_projection_remains_valid_for_partial_publication(partial_service): + env = partial_service + token = CURRENT_PRINCIPAL.set(env.principal) + try: + controller, body, _ = await queue(env) + await controller.tickets.execute(body.request_id) + receipt = await controller.ticket(RUN, body.request_id) + assert receipt["status"] == "completed", receipt.get("result") + current = await env.store.get_run(RUN) + assert current["git_delivery_summary"]["revision"] == receipt["result"]["revision"] + await env.partial.guard(current, current["execution_actor"]) + finally: + CURRENT_PRINCIPAL.reset(token) + + +@pytest.mark.parametrize("projection", ["summary", "draft"]) +async def test_partial_source_rejects_forged_delivery_projection(partial_service, projection): + env = partial_service + token = CURRENT_PRINCIPAL.set(env.principal) + try: + controller, body, _ = await queue(env) + await controller.tickets.execute(body.request_id) + receipt = await controller.ticket(RUN, body.request_id) + assert receipt["status"] == "completed" + if projection == "summary": + env.store.run.data["git_delivery_summary"]["revision"] = "f" * 40 + else: + env.store.run.data["git_delivery_draft"] = {"url": "https://example.test/forged"} + with pytest.raises(ValueError, match="projection"): + await env.partial.guard(await env.store.get_run(RUN), env.store.run.data["execution_actor"]) + finally: + CURRENT_PRINCIPAL.reset(token) + + +@pytest.mark.parametrize("partial_service", ["draft"], indirect=True) +async def test_partial_source_publishes_only_its_exact_unverified_draft(partial_service, tmp_path, monkeypatch): + import json + from types import SimpleNamespace + + import httpx + + from firefly_forge import scm + + env = partial_service + remote = env.partial_remote + branch = env.store.run.data["branch"] + state = {"draft": None} + calls = [] + repository = { + "id": env.store.run.data["repository_id"], + "url": "https://example.test/team/product.git", + "provider": "github", + "connection_provider": "github", + "default_branch": "main", + } + + async def repo(_): + return repository + + async def authentication(_): + return "controlled-fixture", "https://example.test/api/v3" + + async def environment(_): + return {} + + async def wording(run, native, safe, **kwargs): + assert kwargs["git_delivery"].implementation_disposition == "failed" + return { + "body": "### Retained source checkpoint\n\nVerification remains pending.", + "communication": {"mode": "deterministic", "reason": "unconfigured"}, + } + + async def revision(repository, selected, environment): + value = git(remote, "ls-remote", str(remote), "refs/heads/" + selected) + return value.split("\t")[0] if value else None + + real_push = scm.push + + async def push(workspace, selected, **kwargs): + calls.append("push") + await real_push( + workspace, + selected, + repository_url=str(remote), + revision=kwargs["revision"], + expected_remote_revision=kwargs["expected_remote_revision"], + ) + + def record(): + draft = state["draft"] + return { + "number": 7, + "node_id": "PR_fixture", + "html_url": "https://example.test/team/product/pull/7", + "state": "open", + "draft": True, + "body": draft["body"], + "title": draft["title"], + "head": { + "ref": branch, + "sha": git(remote, "rev-parse", "refs/heads/" + branch), + "repo": {"full_name": "team/product"}, + }, + "base": {"ref": "main", "repo": {"full_name": "team/product"}}, + } + + async def provider(request): + calls.append(request.method) + if request.method == "GET" and request.url.path.endswith("/pulls"): + return httpx.Response(200, json=[] if state["draft"] is None else [record()]) + assert request.method == "POST" + state["draft"] = json.loads(request.content) + return httpx.Response(201, json=record()) + + deliveries = env.partial.d + publication_errors = [] + publish = deliveries._publish_checkpoint + + async def inspect_publication(*args, **kwargs): + try: + return await publish(*args, **kwargs) + except Exception as exc: + publication_errors.append(repr(exc)) + raise + + deliveries._publish_checkpoint = inspect_publication + monkeypatch.setattr(scm, "push", push) + env.store.repository = repo + deliveries._remote_revision = revision + deliveries.http_client = lambda **kwargs: httpx.AsyncClient(transport=httpx.MockTransport(provider), **kwargs) + env.engine.connections = SimpleNamespace( + authentication=authentication, git_environment=environment, assert_snapshot=lambda *args: None + ) + env.engine.communication_writer = SimpleNamespace(write=wording, assert_current=lambda evidence: None) + token = CURRENT_PRINCIPAL.set(env.principal) + try: + before = copy.deepcopy(await env.store.get_run(RUN)) + controller, body, _ = await queue(env) + await controller.tickets.execute(body.request_id) + receipt = await controller.ticket(RUN, body.request_id) + assert receipt["status"] == "completed", (receipt.get("result"), publication_errors) + result = receipt["result"] + assert result["remote_status"] == "draft" and result["verification_status"] == "partial_unverified" + assert calls.count("push") == calls.count("POST") == 1 + assert state["draft"]["title"].startswith("[UNVERIFIED]") and state["draft"]["draft"] is True + current = await env.store.get_run(RUN) + await env.partial.guard(current, current["execution_actor"]) + assert current["git_delivery_summary"]["remote_status"] == "draft" + assert current["git_delivery_draft"]["url"] == result["pr_url"] + for key in ("status", "implementation_pending", "bridge_completion_hold", "attempt", "implementation_rounds"): + assert current.get(key) == before.get(key) + with pytest.raises(ValueError, match="no final-ready"): + await deliveries.checkpoint( + RUN, + request_id=body.request_id, + expected_head=result["revision"], + actor=current["execution_actor"], + final=True, + ) + finally: + CURRENT_PRINCIPAL.reset(token) diff --git a/tests/test_partial_settlement_design_host.py b/tests/test_partial_settlement_design_host.py new file mode 100644 index 00000000..a7646bd3 --- /dev/null +++ b/tests/test_partial_settlement_design_host.py @@ -0,0 +1,76 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Characterize custody before proposing a partial-source settlement API.""" + +import httpx2 +import pytest +from pydantic_ai import models +from test_agentic_checkpoint_activation_host import activated, result +from test_bridge_completion_callbacks import CAPABILITY +from test_bridge_completion_native import mock_model_requests +from test_bridge_inflight_cancel import RUN +from test_git_delivery_local import delivery_case as delivery_case + + +async def test_capture_refusal_after_known_write_preserves_consume_without_next_intent(tmp_path, monkeypatch): + from firefly_forge.agentic_checkpoint_boundary import AgenticBoundary + + env = await activated(tmp_path, monkeypatch) + posts = [] + original = AgenticBoundary.capture + + def capture(boundary, wire): + if (env.workspace / "service.py").exists(): + raise ValueError("Checkpoint size or digest differs") + return original(boundary, wire) + + monkeypatch.setattr(AgenticBoundary, "capture", capture) + + def transport(request): + if request.method == "GET": + return httpx2.Response(200, json={"completionRecovery": CAPABILITY}) + posts.append(request.content) + assert len(posts) == 1, "A refused checkpoint must never send the next POST" + return httpx2.Response(200, json=result(tool=True)) + + env.model.client._client._transport = httpx2.MockTransport(transport) + env.model.client._client._mounts = {} + try: + with models.override_allow_model_requests(False), mock_model_requests(env.model): + await env.engine.execute_run(RUN) + current = await env.store.get_run(RUN) + assert current["status"] == "failed" + assert current["implementation_pending"] is True + assert current["error"] == "Checkpoint size or digest differs" + assert (env.workspace / "service.py").read_text() == "VALUE = 1\n" + rows = env.journal.records.rows + for kind in ("bridge_request_intent", "agent_checkpoint", "bridge_request_response", "agent_consume"): + assert len([row for row in rows if row.kind == kind]) == 1 + assert not any(row.kind == "worker_feedback" for row in rows) + assert not any("predispatch" in row.kind or "partial_source" in row.kind for row in rows) + assert len(posts) == 1 + finally: + await env.model.client.close() + + +@pytest.mark.parametrize("flag", ["implementation_pending", "bridge_completion_hold"]) +async def test_failed_status_does_not_settle_effects_for_existing_git_candidate(delivery_case, flag): + c = delivery_case() + c.state["run"]["status"] = "failed" + c.state["run"][flag] = True + (c.root / "product.py").write_text("value = 2\n") + with pytest.raises(ValueError, match="source owner to settle"): + await c.service.candidate("1" * 32, actor=c.state["run"]["execution_actor"]) + assert not c.state["entries"] + + +def test_failed_feedback_proof_cannot_be_invented_from_an_exception_string(): + from test_git_failed_unit_adoption import failed_case + + from firefly_forge.git_failed_unit_adoption import failed_unit_proof + + run, task, entries, observation = failed_case() + run["worker_feedback"] = None + run["error"] = "Checkpoint size or digest differs" + entries = [entry for entry in entries if entry["kind"] != "worker_feedback"] + with pytest.raises(ValueError, match="exact accepted feedback"): + failed_unit_proof(run, task, entries, observation) From 5ee4033c885f05b9a41efb15d619a830c3184565 Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Sun, 25 Oct 2026 00:33:04 -0700 Subject: [PATCH 2/5] Keep partial-source diagrams legible on small screens --- docs/checkpoint-verification.md | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/docs/checkpoint-verification.md b/docs/checkpoint-verification.md index 06a902bd..71eedb62 100644 --- a/docs/checkpoint-verification.md +++ b/docs/checkpoint-verification.md @@ -72,7 +72,15 @@ A failed model session may have already written useful source. A **partial-sourc 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. +
+ ```mermaid +--- +config: + flowchart: + wrappingWidth: 96 + rankSpacing: 24 +--- flowchart LR A["Failed implementation"] --> B["Review stored source and receipts"] B --> C["Unverified Git checkpoint"] @@ -81,6 +89,10 @@ flowchart LR B -->|"Unknown or changed"| H["Keep source held"] ``` +
+ +

Follow the recorded source into a checkpoint and a new run. Focus and scroll the diagram on a narrow screen.

+ 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: From 9fc383dc802a4cb4d705862e8274488836e8ccfd Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Sun, 25 Oct 2026 00:57:01 -0700 Subject: [PATCH 3/5] Resolve native maintenance lineage through its authenticated SCM parent --- docs/scm.md | 8 ++ src/firefly_forge/scm_conversations.py | 134 +++++++++++------- src/firefly_forge/scm_maintain_tickets.py | 9 +- .../test_scm_maintain_derived_parent_host.py | 119 ++++++++++++++++ ...test_scm_maintain_derived_parent_native.py | 81 +++++++++++ 5 files changed, 296 insertions(+), 55 deletions(-) create mode 100644 tests/test_scm_maintain_derived_parent_host.py create mode 100644 tests/test_scm_maintain_derived_parent_native.py diff --git a/docs/scm.md b/docs/scm.md index ffffae1f..a5e3581b 100644 --- a/docs/scm.md +++ b/docs/scm.md @@ -1071,6 +1071,14 @@ resume, inherited parent grant, extra coding budget, or retry of an uncertain remote write. Unbound children, checkpoint tests and other project readers keep read-only lineage. Revoked or expired consent holds further notifications. +A parent reached through native checkpoint lineage has two different identities: +its run-creation receipt proves the derived run, while its authenticated SCM +parent receipt proves the issue link. Configuration and every later notification +revalidation resolve both stores and compare the same typed lineage, frozen +source and current project/component/owner authority. The original creator is +provenance, not a new execution grant. Newer status comments do not replace the +retained parent, and no duplicate parent notification is needed. + This first setup path is native API configuration. The natural coordinator does not yet create a maintenance unit or its communication binding from an issue request. Controlled host/native tests and a real GitHub acceptance trial are diff --git a/src/firefly_forge/scm_conversations.py b/src/firefly_forge/scm_conversations.py index e6a5775e..69b2086e 100644 --- a/src/firefly_forge/scm_conversations.py +++ b/src/firefly_forge/scm_conversations.py @@ -98,68 +98,100 @@ async def references(self, delivery: Data, scopes: list[Data], *, durable: bool raise AccessDenied("Ticket context exceeds the bounded lookup; review its exact native receipts") candidates = [(copy.deepcopy(row.data), row.sequence) for row in rows] for previous, sequence in candidates: - run_id = previous.get("run_id") - if ( - not run_id - or previous["id"] == delivery["id"] - or previous.get("natural_parent") - or (previous.get("command") or {}).get("adapter") - ): - continue try: - run = await self.store.get_run(run_id) - scope = next( - ( - s - for s in scopes - if s["project_id"] == run.get("project_id") and run.get("component_id") in s["component_ids"] - ), - None, - ) - if scope is None or run.get("repository_id") != delivery["repository_id"]: - continue - if previous.get("maintain_ticket") is not None: - from firefly_forge.scm_maintain_tickets import MaintainTickets - - await MaintainTickets(self.collaboration).authorize_reference(previous, run, delivery, scopes) - elif previous.get("native_question") is not None: - from firefly_forge.native_question_publications import NativeQuestionPublications - - await NativeQuestionPublications(self.collaboration).authorize_reference(previous, run) - elif (previous.get("verification") or {}).get("transport") == "github_rest_poll_v1": - polling = self.collaboration.polling or getattr(self.store, "scm_polling", None) - if polling is None: - continue - await polling.ticket_reference(previous, run) - else: - await SCMActorAuthority(self.store, self.collaboration.scm).ticket_reference(previous, run) + reference, run = await self._receipt_reference(previous, sequence, delivery, scopes) except (AccessDenied, KeyError, ValueError, TypeError): continue + run_id = run["id"] if run_id in runs: continue runs[run_id] = run if len(runs) > 200: raise AccessDenied("The current ticket has too many distinct run references for one natural choice") - references.append( - TicketContextReferenceV1( - receipt_id=previous["id"], - sequence=sequence, - run_id=run_id, - project_id=run["project_id"], - component_id=run["component_id"], - context_digest=digest( - { - "observation": observation_cut(previous), - "actor": previous.get("scm_execution_actor"), - "run_id": run_id, - "action_status": previous.get("action_status"), - } - ), - ).model_dump(mode="json") - ) + references.append(reference) await self._checkpoint_tests(delivery, scopes, references, runs) return references, runs + async def _receipt_reference( + self, previous: Data, sequence: int, delivery: Data, scopes: list[Data] + ) -> tuple[Data, Data]: + run_id = previous.get("run_id") + if ( + not run_id + or previous["id"] == delivery["id"] + or previous.get("natural_parent") + or (previous.get("command") or {}).get("adapter") + or any(previous.get(key) != delivery.get(key) for key in ("repository_id", "provider")) + or any( + (previous.get("issue") or {}).get(key) != (delivery.get("issue") or {}).get(key) + for key in ("kind", "number") + ) + ): + raise AccessDenied("The retained ticket reference has no exact SCM parent") + run = await self.store.get_run(run_id) + scope = next( + ( + s + for s in scopes + if s["project_id"] == run.get("project_id") and run.get("component_id") in s["component_ids"] + ), + None, + ) + if scope is None or run.get("repository_id") != delivery["repository_id"]: + raise AccessDenied("The retained ticket polling authority is unavailable") + if previous.get("maintain_ticket") is not None: + from firefly_forge.scm_maintain_tickets import MaintainTickets + + await MaintainTickets(self.collaboration).authorize_reference(previous, run, delivery, scopes) + elif previous.get("native_question") is not None: + from firefly_forge.native_question_publications import NativeQuestionPublications + + await NativeQuestionPublications(self.collaboration).authorize_reference(previous, run) + elif (previous.get("verification") or {}).get("transport") == "github_rest_poll_v1": + polling = self.collaboration.polling or getattr(self.store, "scm_polling", None) + if polling is None: + raise AccessDenied("The retained ticket reference is outside current native scope") + await polling.ticket_reference(previous, run) + else: + await SCMActorAuthority(self.store, self.collaboration.scm).ticket_reference(previous, run) + reference = TicketContextReferenceV1( + receipt_id=previous["id"], + sequence=sequence, + run_id=run_id, + project_id=run["project_id"], + component_id=run["component_id"], + context_digest=digest( + { + "observation": observation_cut(previous), + "actor": previous.get("scm_execution_actor"), + "run_id": run_id, + "action_status": previous.get("action_status"), + } + ), + ).model_dump(mode="json") + return reference, run + + async def resolve_reference(self, reference: Data, delivery: Data, scopes: list[Data]) -> Data: + """Resolve stable typed lineage through its SCM receipt and native creation stores.""" + retained = TicketContextReferenceV1.model_validate(reference).model_dump(mode="json") + receipt_id = ( + retained["receipt_id"] + if retained.get("kind", "scm_receipt") == "scm_receipt" + else retained["parent_receipt_id"] + ) + async with self.store._unit(): + row = await self.collaboration.inbox.find_by_id(identifier(receipt_id)) + if row is None: + raise AccessDenied("The authenticated SCM parent receipt is unavailable") + previous, sequence = copy.deepcopy(row.data), row.sequence + parent, run = await self._receipt_reference(previous, sequence, delivery, scopes) + references = [parent] + if retained.get("kind", "scm_receipt") != "scm_receipt": + await self._checkpoint_tests(delivery, scopes, references, {run["id"]: run}) + if sum(item == retained for item in references) != 1: + raise AccessDenied("The native creation or authenticated ticket parent provenance changed") + return previous + async def _checkpoint_tests(self, delivery: Data, scopes: list[Data], references: list[Data], runs: Data) -> None: from firefly_forge.checkpoint_test_models import frozen_checkpoint_source from firefly_forge.run_creation_requests import RunCreationRequests diff --git a/src/firefly_forge/scm_maintain_tickets.py b/src/firefly_forge/scm_maintain_tickets.py index 4005f05a..147149eb 100644 --- a/src/firefly_forge/scm_maintain_tickets.py +++ b/src/firefly_forge/scm_maintain_tickets.py @@ -138,7 +138,7 @@ async def configure(self, project: str, repository: str, request: MaintainTicket parents = [value for value in references if value["run_id"] == parent_id] if len(parents) != 1 or run["component_id"] not in scope["component_ids"]: raise AccessDenied("The observed ticket has no exact authenticated checkpoint parent") - parent = await self.c.receipt(parents[0]["receipt_id"]) + parent = await SCMConversations(self.c).resolve_reference(parents[0], source, [scope]) parent_cut = observation_cut(parent) await self.store.admit_credentials({"request": incoming, "source": observation_cut(source)}) async with self.polling._locked(project, repository, [actor.id.hex, owner["user_id"]]): @@ -149,7 +149,8 @@ async def configure(self, project: str, repository: str, request: MaintainTicket current_receipt = await self._creation(request.creation_request_id) current_run = await self.store.get_run(request.run_id) creation_scope(current_run, current_receipt) - if current_receipt != receipt or observation_cut(await self.c.receipt(parent["id"])) != parent_cut: + current_parent = await SCMConversations(self.c).resolve_reference(parents[0], source, [current_scope]) + if current_receipt != receipt or observation_cut(current_parent) != parent_cut: raise AccessDenied("The retained maintain creation or ticket parent changed") row, _ = await self.polling._state(repository) state = state_of(row.data) @@ -329,10 +330,10 @@ async def current(self, data: Data, *, remote: bool = False, dispatch: bool = Fa raise AccessDenied("The maintain question changed from its native feedback") if creation["digest"] != value["creation_digest"]: raise AccessDenied("The maintain child creation changed") - parent = await self.c.receipt(value["parent_reference"]["receipt_id"]) + scope, _, owner = await self._target(source, project, repository, remote=remote) + parent = await SCMConversations(self.c).resolve_reference(value["parent_reference"], source, [scope]) if observation_cut(parent) != value["parent_observation"]: raise AccessDenied("The exact native parent ticket provenance changed") - scope, _, owner = await self._target(source, project, repository, remote=remote) if ( scope != value["polling_scope"] or owner != value["owner"] diff --git a/tests/test_scm_maintain_derived_parent_host.py b/tests/test_scm_maintain_derived_parent_host.py new file mode 100644 index 00000000..bf9b6905 --- /dev/null +++ b/tests/test_scm_maintain_derived_parent_host.py @@ -0,0 +1,119 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Native creation provenance must resolve to its distinct authenticated SCM parent.""" + +import copy +from types import SimpleNamespace +from uuid import UUID, uuid4 + +import pytest +from test_scm_maintain_ticket_host import body, prepared_binding +from test_scm_polling import poll_case as poll_case + +from firefly_forge.journal import digest +from firefly_forge.project_access import AccessDenied +from firefly_forge.run_creation_requests import RunCreationRequests +from firefly_forge.scm_maintain_ticket_models import MaintainTicketInput +from firefly_forge.scm_maintain_tickets import MaintainTickets + + +async def nested_binding(c, monkeypatch): + service, first = await prepared_binding(c, monkeypatch) + parent = c.db["runs"][first.run_id].data + parent_creation = await service._creation(first.creation_request_id) + parent_creation["original_actor"] = {"user_id": c.user_id} + parent_creation["digest"] = digest({k: v for k, v in parent_creation.items() if k != "digest"}) + child_id, creation_id = uuid4().hex, uuid4().hex + reference = {**parent["source_checkpoint"], "run_id": parent["id"], "request_id": uuid4().hex} + source = {**copy.deepcopy(parent["source_checkpoint_snapshot"]), "reference": reference} + source["digest"] = digest({k: v for k, v in source.items() if k != "digest"}) + child = { + **copy.deepcopy(parent), + "id": child_id, + "source_checkpoint": reference, + "source_checkpoint_snapshot": source, + } + c.db["runs"][child_id] = SimpleNamespace( + id=UUID(child_id), repository_id=UUID(c.repository), status="queued", data=child + ) + creation = { + **copy.deepcopy(parent_creation), + "request_id": creation_id, + "run_id": child_id, + "source_checkpoint": source, + "input": {"operation": "maintain", "publish": False, "source_checkpoint": reference}, + } + creation["digest"] = digest({k: v for k, v in creation.items() if k != "digest"}) + creations = {parent_creation["request_id"]: parent_creation, creation_id: creation} + + async def load(self, request_id): + return copy.deepcopy(creations.get(request_id)) + + async def children(parents, repository): + assert repository == UUID(c.repository) + return [UUID(parent_creation["request_id"])] if UUID(c.run_id) in parents else [] + + monkeypatch.setattr(RunCreationRequests, "_load", load) + monkeypatch.setattr(MaintainTickets, "_creation", load) + c.store.run_creation_records = SimpleNamespace(checkpoint_children=children) + request = MaintainTicketInput.model_validate( + { + **body(), + "request_id": uuid4().hex, + "run_id": child_id, + "creation_request_id": creation_id, + "expected_creation_digest": creation["digest"], + "observed_receipt_id": first.observed_receipt_id, + "expected_reference_digest": first.expected_reference_digest, + } + ) + c.state["derived_parent_posts"] = copy.deepcopy(c.state["posts"]) + return service, request, creations, parent + + +async def test_native_derived_parent_configures_and_revalidates_without_treating_creation_as_inbox( + poll_case, monkeypatch +): + async with poll_case() as c: + service, request, creations, parent = await nested_binding(c, monkeypatch) + before = copy.deepcopy(parent) + kept = await service.configure(c.project, c.repository, request) + ref = kept["parent_reference"] + assert ref["kind"] == "native_checkpoint_maintain" + assert ref["receipt_id"] in creations and ref["receipt_id"] not in c.db["inbox"] + assert ref["parent_receipt_id"] == kept["parent_observation"]["id"] + anchor = await c.collaboration.receipt(kept["outbox_id"]) + assert (await service.current(anchor))["run"]["id"] == request.run_id + assert c.db["runs"][parent["id"]].data == before + assert c.state["posts"] == c.state["derived_parent_posts"] + + +@pytest.mark.parametrize("change", ["digest", "creator", "source", "scope"]) +async def test_derived_parent_provenance_drift_refuses_current_notification(poll_case, monkeypatch, change): + async with poll_case() as c: + service, request, creations, parent = await nested_binding(c, monkeypatch) + kept = await service.configure(c.project, c.repository, request) + creation = creations[kept["parent_reference"]["receipt_id"]] + if change == "digest": + creation["digest"] = "f" * 64 + elif change == "creator": + creation["original_actor"]["user_id"] = "f" * 32 + elif change == "source": + c.db["runs"][parent["id"]].data["source_checkpoint_snapshot"]["revision"] = "f" * 40 + else: + creation["component_id"] = "other" + anchor = await c.collaboration.receipt(kept["outbox_id"]) + with pytest.raises((AccessDenied, ValueError)): + await service.current(anchor) + assert c.state["posts"] == c.state["derived_parent_posts"] + + +async def test_newer_ticket_reference_does_not_replace_the_bound_native_parent(poll_case, monkeypatch): + async with poll_case() as c: + service, request, _, _ = await nested_binding(c, monkeypatch) + kept = await service.configure(c.project, c.repository, request) + previous = copy.deepcopy(kept["parent_reference"]) + await c.new_command("status", number=102) + anchor = await c.collaboration.receipt(kept["outbox_id"]) + context = await service.current(anchor) + assert context["binding"]["parent_reference"] == previous + assert context["run"]["id"] == request.run_id diff --git a/tests/test_scm_maintain_derived_parent_native.py b/tests/test_scm_maintain_derived_parent_native.py new file mode 100644 index 00000000..8fa3840d --- /dev/null +++ b/tests/test_scm_maintain_derived_parent_native.py @@ -0,0 +1,81 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Real Store/creation lineage with controlled SCM and no model or provider dispatch.""" + +import copy +from uuid import UUID + +import pytest +from test_scm_maintain_ticket_native import maintain_ticket as maintain_ticket # noqa: F401 + +from firefly_forge.identity_persistence import ForgeIdentityUserRepository +from firefly_forge.project_access import AccessDenied +from firefly_forge.run_creation_persistence import ForgeRunCreationRequestRepository +from firefly_forge.scm_conversations import SCMConversations, observation_cut + + +async def retained_lineage(values): + c, child, creation, observed, _, _ = values + conversations = SCMConversations(c.case.collaboration) + scopes = await conversations.scopes(observed, remote=False) + references, _ = await conversations.references(observed, scopes, durable=True) + reference = next(value for value in references if value["run_id"] == child["id"]) + assert reference["kind"] == "native_checkpoint_maintain" + assert reference["receipt_id"] == creation["request_id"] + assert reference["parent_receipt_id"] != reference["receipt_id"] + return conversations, scopes, reference + + +def test_native_derived_parent_resolves_canonical_creation_and_distinct_scm_receipt(maintain_ticket): + c, child, creation, observed, _, _ = maintain_ticket + before = copy.deepcopy(c.case.state["posts"]) + + async def verify(): + conversations, scopes, reference = await retained_lineage(maintain_ticket) + parent = await conversations.resolve_reference(reference, observed, scopes) + assert parent["id"] == reference["parent_receipt_id"] + assert parent["run_id"] == child["source_checkpoint_snapshot"]["reference"]["run_id"] + assert reference["creator_user_id"] == creation["original_actor"]["user_id"] + expected = await c.case.collaboration.receipt(reference["parent_receipt_id"]) + assert observation_cut(parent) == observation_cut(expected) + assert await c.case.store.get_run(child["id"]) == child + + c.case.http.portal.call(verify) + assert c.case.state["posts"] == before + + +def test_native_derived_parent_rejects_changed_creation_envelope(maintain_ticket): + c, child, _, observed, _, _ = maintain_ticket + before = copy.deepcopy(c.case.state["posts"]) + + async def verify(): + conversations, scopes, reference = await retained_lineage(maintain_ticket) + async with c.case.store._unit(): + records = ForgeRunCreationRequestRepository() + row = await records.find_by_id(UUID(reference["receipt_id"])) + row.data = {**copy.deepcopy(row.data), "digest": "f" * 64} + await records.save(row) + with pytest.raises(AccessDenied, match="provenance changed"): + await conversations.resolve_reference(reference, observed, scopes) + assert await c.case.store.get_run(child["id"]) == child + + c.case.http.portal.call(verify) + assert c.case.state["posts"] == before + + +def test_native_derived_parent_rechecks_current_reader_before_resolving(maintain_ticket): + c, child, _, observed, _, _ = maintain_ticket + before = copy.deepcopy(c.case.state["posts"]) + + async def verify(): + conversations, scopes, reference = await retained_lineage(maintain_ticket) + async with c.case.store._unit(): + users = ForgeIdentityUserRepository() + user = await users.find_by_id(UUID(scopes[0]["native_user_id"])) + user.enabled = False + await users.save(user) + with pytest.raises(AccessDenied, match="provenance changed"): + await conversations.resolve_reference(reference, observed, scopes) + assert await c.case.store.get_run(child["id"]) == child + + c.case.http.portal.call(verify) + assert c.case.state["posts"] == before From 0765f34b9d5e85dcea1d3935e478e0e84ac81d04 Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Sun, 25 Oct 2026 01:14:18 -0700 Subject: [PATCH 4/5] Expose native lineage fixture dependencies during isolated SQL tests --- tests/test_scm_maintain_derived_parent_native.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/test_scm_maintain_derived_parent_native.py b/tests/test_scm_maintain_derived_parent_native.py index 8fa3840d..22e70224 100644 --- a/tests/test_scm_maintain_derived_parent_native.py +++ b/tests/test_scm_maintain_derived_parent_native.py @@ -5,6 +5,8 @@ from uuid import UUID import pytest +from checkpoint_test_native_fixtures import checkpoint_native as checkpoint_native # noqa: F401 +from scm_polling_native_fixtures import native_poll as native_poll # noqa: F401 from test_scm_maintain_ticket_native import maintain_ticket as maintain_ticket # noqa: F401 from firefly_forge.identity_persistence import ForgeIdentityUserRepository From b194b6fc5409e933c494e50e0141afca282b2a79 Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Sun, 25 Oct 2026 01:22:41 -0700 Subject: [PATCH 5/5] Reject forged lineage references without mutating immutable birth records --- tests/test_scm_maintain_derived_parent_native.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/tests/test_scm_maintain_derived_parent_native.py b/tests/test_scm_maintain_derived_parent_native.py index 22e70224..c8231b71 100644 --- a/tests/test_scm_maintain_derived_parent_native.py +++ b/tests/test_scm_maintain_derived_parent_native.py @@ -45,7 +45,7 @@ async def verify(): assert c.case.state["posts"] == before -def test_native_derived_parent_rejects_changed_creation_envelope(maintain_ticket): +def test_native_derived_parent_rejects_forged_creation_reference(maintain_ticket): c, child, _, observed, _, _ = maintain_ticket before = copy.deepcopy(c.case.state["posts"]) @@ -53,11 +53,14 @@ async def verify(): conversations, scopes, reference = await retained_lineage(maintain_ticket) async with c.case.store._unit(): records = ForgeRunCreationRequestRepository() - row = await records.find_by_id(UUID(reference["receipt_id"])) - row.data = {**copy.deepcopy(row.data), "digest": "f" * 64} - await records.save(row) - with pytest.raises(AccessDenied, match="provenance changed"): - await conversations.resolve_reference(reference, observed, scopes) + original = copy.deepcopy((await records.find_by_id(UUID(reference["receipt_id"]))).data) + for field, changed in (("context_digest", "f" * 64), ("creator_user_id", "f" * 32)): + assert reference[field] != changed + forged = {**reference, field: changed} + with pytest.raises(AccessDenied, match="provenance changed"): + await conversations.resolve_reference(forged, observed, scopes) + async with c.case.store._unit(): + assert (await records.find_by_id(UUID(reference["receipt_id"]))).data == original assert await c.case.store.get_run(child["id"]) == child c.case.http.portal.call(verify)