Settle retained worker feedback without replaying execution

docs/architecture.md

@@ -520,6 +520,8 @@ The restore ticket binds the original checkpoint, intent and selected result rec
 
 Boundary snapshots retain lossless canonical native history within an explicit 8 MiB raw bound. New records above 64 KiB use the version-3 shared-history zlib/base64 codec within the existing 192 KiB snapshot, 256 KiB WAL and 4 MiB per-run retention limits. Equal native history arrays are stored once and reconstructed losslessly; the expanded canonical record keeps its full size/hash binding. This reduces duplication without promising unbounded history or chunked overflow storage. Canonical envelope/base64, raw SHA-256, size, end-of-stream and trailing-data checks precede restoration. Current credential admission scans decoded records at capture, consume, restore and generic journal read without exposing raw history in returned WAL or logs. Generic journal reads suspend ambient unit bindings and inspect one bounded record at a time under current reader/configuration guards. Historical source admission V11 added the decoder to its explicit closure; the checkpoint runtime contract binds custody producer and consumer. The changed version-3 boundary needs fresh source qualification before adoption.
 
+Known-worker output settlement is a separate operator-owned native transaction. A complete current-contract HTTP/checkpoint/consume closure must match its declared final output and exact stored typed worker feedback. A hash-only reviewed candidate binds the current operator, task, source/configuration and journal cut. Its immutable receipt retains the original observed hold; the receipt and only its resolved hold projection commit together. No model replay, source adoption, verification, resume or new execution grant is performed. Unknown custody and normal reports without worker feedback remain ineligible for this first path. Whole-journal reads retain inherited capacity boundaries; this is not an elastic-storage claim. See [known-output settlement](checkpoint-verification.md#settle-a-known-worker-output-without-replay).
+
 `BridgeProvider.execute()` creates a separate pending-request set for each invocation. Agentic task-graph children inherit that shared object through a context variable. Each completion tracks its own attempt before transport and clears it after a validated response, an exhausted stream or a known quota/pause response. Completing one request cannot clear another pending request. If the task graph unwraps a cancellation exception, the outer provider checks its own set before returning cancellation to the Engine. Both context tokens are reset when their scopes end. The Engine then records an unresolved submission against the original journal intent before finishing local cancellation. Regression controls cover the real `FireflyAgent` path, concurrent invocations on a shared provider, successive completions and restart through the native Store.
 
 Explicit manual continuation preserves the run's existing retry counter, including an absent field. Automatic retries increment it. Source-control consent can therefore compare the same frozen run fields before and after an authorized manual resume. The run identity, plan and execution actor remain subject to their existing checks.

docs/checkpoint-verification.md

@@ -129,3 +129,37 @@ Read `client.git_delivery_request(failed_run_id, request["request_id"])` to foll
 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.
+
+## Settle a known worker output without replay
+
+A retained HTTP result and consume record do not by themselves establish that
+an implementation finished. The native known-worker settlement API accepts
+only a complete current-contract HTTP/checkpoint/consume closure whose declared
+final JSON or output tool matches the exact stored typed worker feedback.
+Unknown, incomplete, ambiguous or changed custody refuses. This first version
+supports direct HTTP consume-v1; restored consume-v2 requires its separate
+restore-ticket proof. A normal report without worker feedback is also outside
+this first settlement path; it must not be replaced with invented feedback.
+
+Review `GET /api/v1/runs/{run_id}/worker-completion-settlements/candidate`, then
+submit its exact hashes and explicit acknowledgments to
+`POST /api/v1/runs/{run_id}/worker-completion-settlements/{request_id}`. The
+operator's current project/repository execution access, original execution
+identity, task attempt, provider/model scope and immutable journal cut are
+rechecked. `GET` on that request path returns its retained receipt under current
+authority and credential admission. Responses contain identities and hashes,
+not completion bodies, feedback text, prompts or configuration secrets.
+
+The transaction appends a typed immutable WAL receipt containing the original
+observed hold before projecting that exact hold as resolved and linking its
+receipt. Both changes commit or roll back together. It does not change run or
+task status, attempts, pending questions, implementation state, source, tool or
+provider allowances, usage or budgets. It does not resume a graph, run checks,
+publish Git changes or call a model. Those actions keep their existing consent
+and safety gates. Reusing the same request identity reads the known receipt;
+it never replays a consumed response.
+
+This path inherits the existing whole-journal and completion retention guards,
+including the legacy 10000-row/16MiB aggregate journal read boundary. It is not
+elastic storage or a scalable pinned-head reducer. It introduces no new
+project-sized count or aggregate ceiling and never admits a truncated closure.

src/firefly_forge/api.py

@@ -128,6 +128,8 @@
 from firefly_forge.knowledge_first import KnowledgeFirst
 from firefly_forge.knowledge_persistence import ForgeKnowledgeUploadRepository
 from firefly_forge.knowledge_uploads import KnowledgeUploads, KnowledgeUploadsController
+from firefly_forge.known_worker_completion_controllers import KnownWorkerCompletionController
+from firefly_forge.known_worker_completions import KnownWorkerCompletions
 from firefly_forge.learning import LearningLedger
 from firefly_forge.learning_controllers import LearningController
 from firefly_forge.learning_observation_attempts import LearningObservationAttempts
@@ -865,6 +867,10 @@ def execution_journal(
     def bridge_completions(self, store: Store, journal: Journal) -> BridgeCompletions:
         return BridgeCompletions(store, journal)
 
+    @bean
+    def known_worker_completions(self, store: Store, journal: Journal) -> KnownWorkerCompletions:
+        return KnownWorkerCompletions(store, journal)
+
     @bean
     def runner_backends(self) -> RunnerBackends:
         return RunnerBackends()
@@ -1232,6 +1238,7 @@ class Application:
         AgentCheckpointCandidateController,
         GitDeliveryPolicyController,
         GitDeliveryController,
+        KnownWorkerCompletionController,
         GitDiffArchive,
         GitDiffController,
         JournalController,

src/firefly_forge/known_worker_completion_controllers.py

@@ -0,0 +1,31 @@
+# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0
+"""Hash-only known-worker completion candidates and local settlement receipts."""
+
+from typing import Any
+
+from pyfly.container import rest_controller
+from pyfly.web import Body, PathVar, get_mapping, post_mapping, request_mapping
+
+from firefly_forge.known_worker_completion_models import WorkerCompletionInputV1
+from firefly_forge.known_worker_completions import KnownWorkerCompletions
+
+
+@rest_controller
+@request_mapping("/api/v1/runs/{run_id}/worker-completion-settlements")
+class KnownWorkerCompletionController:
+    def __init__(self, completions: KnownWorkerCompletions) -> None:
+        self.completions = completions
+
+    @get_mapping("/candidate")
+    async def candidate(self, run_id: PathVar[str]) -> dict[str, Any]:
+        return await self.completions.candidate(run_id)
+
+    @post_mapping("/{request_id}", status_code=201)
+    async def request(
+        self, run_id: PathVar[str], request_id: PathVar[str], body: Body[WorkerCompletionInputV1]
+    ) -> dict[str, Any]:
+        return await self.completions.request(run_id, request_id, body)
+
+    @get_mapping("/{request_id}")
+    async def get(self, run_id: PathVar[str], request_id: PathVar[str]) -> dict[str, Any]:
+        return await self.completions.get(run_id, request_id)

src/firefly_forge/known_worker_completion_models.py

@@ -0,0 +1,121 @@
+# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0
+"""Explicit consent to known worker output, never a new execution grant."""
+
+from typing import Any, Literal
+
+from pydantic import ConfigDict, Field, field_validator, model_validator
+
+from firefly_forge.journal import digest
+from firefly_forge.scm_authority_models import Digest, NativeID, StrictContract
+
+
+class WorkerCompletionInputV1(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
+    proof_digest: Digest
+    feedback_record_digest: Digest
+    acknowledge_known_worker_output: Literal[True]
+    acknowledge_no_execution_grant: Literal[True]
+
+    @field_validator("acknowledge_known_worker_output", "acknowledge_no_execution_grant", mode="before")
+    @classmethod
+    def exact_acknowledgment(cls, value: Any) -> Any:
+        if value is not True:
+            raise ValueError("Acknowledge known output without a new execution grant")
+        return value
+
+
+class CompletionCustodyV1(StrictContract):
+    request_id: str = Field(pattern=r"^fbr1_[a-f0-9]{32}$")
+    intent_digest: Digest
+    checkpoint_record_digest: Digest
+    response_record_digest: Digest
+    consume_record_digest: Digest
+
+
+class CompletionHoldV1(StrictContract):
+    version: Literal[1]
+    reason: Literal["graph_checkpoint_unavailable"]
+    request_id: str = Field(pattern=r"^fbr1_[a-f0-9]{32}$")
+    intent_digest: Digest
+
+
+class WorkerCompletionProofV1(StrictContract):
+    version: Literal[1]
+    run_id: NativeID
+    task_id: Literal["implement"]
+    task_attempt: int = Field(ge=1)
+    task_key: str = Field(min_length=1, max_length=64)
+    runtime_contract: Digest
+    completion_count: int = Field(ge=1)
+    completions: list[CompletionCustodyV1]
+    feedback_record_digest: Digest
+    feedback_sha256: Digest
+    final_response_sha256: Digest
+    observed_hold_digest: Digest
+    verification: Literal["worker_output_known_only"]
+    digest: Digest
+
+    @model_validator(mode="after")
+    def exact_proof(self) -> "WorkerCompletionProofV1":
+        if (
+            self.completion_count != len(self.completions)
+            or len({item.request_id for item in self.completions}) != self.completion_count
+            or digest(self.model_dump(exclude={"digest"})) != self.digest
+        ):
+            raise ValueError("Known worker completion proof differs from complete custody")
+        return self
+
+
+class WorkerCompletionCandidateV1(StrictContract):
+    model_config = ConfigDict(extra="forbid", strict=True, frozen=True, hide_input_in_errors=True)
+    version: Literal[1]
+    run_id: NativeID
+    actor_digest: Digest
+    policy_digest: Digest
+    run_digest: Digest
+    journal_digest: Digest
+    proof: WorkerCompletionProofV1
+    digest: Digest
+
+    @model_validator(mode="after")
+    def exact_digest(self) -> "WorkerCompletionCandidateV1":
+        if digest(self.model_dump(exclude={"digest"})) != self.digest:
+            raise ValueError("Worker completion candidate digest differs")
+        return self
+
+
+class WorkerCompletionReceiptV1(StrictContract):
+    model_config = ConfigDict(extra="forbid", strict=True, frozen=True, hide_input_in_errors=True)
+    version: Literal[1]
+    run_id: NativeID
+    request_id: NativeID
+    actor_digest: Digest
+    candidate_digest: Digest
+    policy_digest: Digest
+    run_digest: Digest
+    journal_digest: Digest
+    request: WorkerCompletionInputV1
+    proof: WorkerCompletionProofV1
+    observed_hold: CompletionHoldV1
+
+    @model_validator(mode="after")
+    def exact_receipt(self) -> "WorkerCompletionReceiptV1":
+        proof = self.proof.model_dump(mode="json")
+        if (
+            self.request.candidate_digest != self.candidate_digest
+            or self.request.policy_digest != self.policy_digest
+            or self.request.expected_run_digest != self.run_digest
+            or self.request.journal_digest != self.journal_digest
+            or self.request.proof_digest != proof.get("digest")
+            or self.request.feedback_record_digest != proof.get("feedback_record_digest")
+            or proof.get("run_id") != self.run_id
+            or digest(self.observed_hold.model_dump(mode="json")) != proof.get("observed_hold_digest")
+            or digest({k: v for k, v in proof.items() if k != "digest"}) != proof.get("digest")
+        ):
+            raise ValueError("Known worker receipt differs from its reviewed proof")
+        return self