diff --git a/docs/architecture.md b/docs/architecture.md index d64ab44f..946f43fe 100644 --- a/docs/architecture.md +++ b/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. diff --git a/docs/checkpoint-verification.md b/docs/checkpoint-verification.md index 71eedb62..01ec7d38 100644 --- a/docs/checkpoint-verification.md +++ b/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. diff --git a/src/firefly_forge/api.py b/src/firefly_forge/api.py index 32f07900..9a00571d 100644 --- a/src/firefly_forge/api.py +++ b/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, diff --git a/src/firefly_forge/known_worker_completion_controllers.py b/src/firefly_forge/known_worker_completion_controllers.py new file mode 100644 index 00000000..5a149d13 --- /dev/null +++ b/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) diff --git a/src/firefly_forge/known_worker_completion_models.py b/src/firefly_forge/known_worker_completion_models.py new file mode 100644 index 00000000..f6a3aca8 --- /dev/null +++ b/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 diff --git a/src/firefly_forge/known_worker_completion_proof.py b/src/firefly_forge/known_worker_completion_proof.py new file mode 100644 index 00000000..b0ce9bfe --- /dev/null +++ b/src/firefly_forge/known_worker_completion_proof.py @@ -0,0 +1,395 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Known HTTP custody prepared outside a writer and checked atomically inside it.""" + +from __future__ import annotations + +import copy +import hashlib +import json +from dataclasses import dataclass +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.agent_roles import BINDINGS, worker_feedback_model +from firefly_forge.agentic_checkpoint_boundary import BoundaryCheckpoint +from firefly_forge.agentic_checkpoint_invocation import invocation_cut, runtime_contract +from firefly_forge.agentic_runtime import RuntimeReport +from firefly_forge.bridge_completion_models import CompletionIntentPayload, CompletionResponsePayload +from firefly_forge.bridge_completions import _cut, _outside_unit +from firefly_forge.configuration import validate_endpoint +from firefly_forge.journal import Journal, digest + +Data = dict[str, Any] +SETTLEMENT = "worker_completion_receipt" +CUSTODY = ("bridge_request_intent", "agent_checkpoint", "bridge_request_response", "agent_consume") + + +def _json(value: str) -> Any: + def pairs(items: list[tuple[str, Any]]) -> Data: + result: Data = {} + for key, item in items: + if key in result: + raise ValueError("Final worker output has duplicate JSON keys") + result[key] = item + return result + + def invalid(_: str) -> None: + raise ValueError("Final worker output is not finite JSON") + + return json.loads(value, object_pairs_hook=pairs, parse_constant=invalid) + + +def _schema(schema: Data, value: Any) -> None: + pending: list[Any] = [schema] + while pending: + current = pending.pop() + if isinstance(current, dict): + if any( + key in current and not str(current[key]).startswith("#/") + for key in ("$ref", "$dynamicRef", "$recursiveRef") + ): + raise ValueError("Final worker schema cannot resolve external references") + pending.extend(current.values()) + elif isinstance(current, list): + pending.extend(current) + Draft202012Validator(schema).validate(value) + + +def _final_feedback(completion: ChatCompletion, captured: BoundaryCheckpoint) -> Data: + parameters = captured.record.contract["parameters"] + choice = completion.choices[0] + if parameters.get("output_mode") == "prompted": + definition = parameters.get("output_object") + if choice.finish_reason != "stop" or choice.message.tool_calls or not isinstance(definition, dict): + raise ValueError("Known worker outcome requires its declared final JSON output") + value = _json(choice.message.content or "") + _schema(definition["json_schema"], value) + elif parameters.get("output_mode") == "tool": + calls = choice.message.tool_calls or [] + definitions = {item["name"]: item for item in parameters.get("output_tools", [])} + if choice.finish_reason != "tool_calls" or len(calls) != 1: + raise ValueError("Known worker outcome requires one declared final output tool") + call = calls[0] + if call.type != "function" or call.function.name not in definitions: + raise ValueError("Known worker outcome requires its declared final output tool") + value = _json(call.function.arguments) + _schema(definitions[call.function.name]["parameters_json_schema"], value) + else: + raise ValueError("Worker output has no supported declared final contract") + report = RuntimeReport.model_validate(value) + if report.worker_feedback is None: + raise ValueError("Known worker outcome has no typed final feedback") + raw = value.get("worker_feedback") + return worker_feedback_model(raw).model_validate(raw).model_dump(mode="json") + + +def _scope(run: Data, task: Data) -> str: + return digest( + { + "invocation": invocation_cut(run), + "hold": run.get("bridge_completion_hold"), + "active_task": run.get("active_task"), + "attempt": run.get("attempt"), + "task": {key: task.get(key) for key in ("id", "attempt", "idempotency_key")}, + } + ) + + +def _eligible(run: Data, task: Data, feedback: Data) -> None: + if ( + run.get("agent") != "bridge" + or run.get("active_task") != "implement" + or task.get("id") != "implement" + or task.get("attempt") != run.get("attempt") + or not task.get("idempotency_key") + or task.get("status") not in {"running", "paused", "needs_reconciliation"} + or run.get("status") in {"cancelled", "succeeded"} + or (run.get("cancel_request") or {}).get("status") == "requested" + or not run.get("bridge_completion_hold") + or feedback.get("task_id") != task["id"] + or feedback.get("attempt") != task["attempt"] + or feedback.get("task_id") not in BINDINGS[feedback["role_id"]][1] + ): + raise ValueError("Known worker outcome requires its original current held task attempt") + roles = (run.get("role_snapshot") or {}).get("roles", []) + if roles and not any(item["id"] == feedback["role_id"] and item["enabled"] for item in roles): + raise ValueError("Final worker role is not enabled in the frozen scope") + + +@dataclass(frozen=True) +class PreparedWorkerCompletion: + prefix_count: int + prefix_digest: str + scope_digest: str + feedback: Data + completions: tuple[Data, ...] + output_sha256: str + runtime_contract: str + existing_feedback_digest: str | None + seal: str + + +def _unsettled(entries: list[Data], run_id: str) -> dict[str, list[Data]]: + closed: set[str] = set() + for entry in entries: + if entry["kind"] != SETTLEMENT: + continue + proof = entry["payload"].get("proof") or {} + if ( + proof.get("version") != 1 + or proof.get("run_id") != run_id + or proof.get("verification") != "worker_output_known_only" + ): + raise ValueError("Prior completion receipt has invalid scope") + if digest({k: v for k, v in proof.items() if k != "digest"}) != proof.get("digest"): + raise ValueError("Prior completion receipt proof changed") + for item in proof.get("completions", []): + if item["request_id"] in closed: + raise ValueError("A completion cannot be settled twice") + for kind, field in zip( + CUSTODY, + ("intent_digest", "checkpoint_record_digest", "response_record_digest", "consume_record_digest"), + strict=True, + ): + matches = [ + record + for record in entries + if record["kind"] == kind and record["idempotency_key"] == item["request_id"] + ] + if ( + len(matches) != 1 + or matches[0]["digest"] != item[field] + or matches[0]["sequence"] >= entry["sequence"] + ): + raise ValueError("Prior completion receipt lost its immutable custody") + closed.add(item["request_id"]) + grouped = { + kind: [entry for entry in entries if entry["kind"] == kind and entry["idempotency_key"] not in closed] + for kind in CUSTODY + } + keys = [entry["idempotency_key"] for entry in grouped[CUSTODY[0]]] + if ( + not keys + or len(set(keys)) != len(keys) + or any([entry["idempotency_key"] for entry in grouped[kind]] != keys for kind in CUSTODY[1:]) + ): + raise ValueError("Original completion closure is incomplete, duplicated or unknown") + return grouped + + +def prepare_worker_completion( + run: Data, task: Data, entries: list[Data], *, feedback: Data | None = None +) -> PreparedWorkerCompletion: + _outside_unit() + if not entries or not Journal.verify_chain(entries): + raise ValueError("Known worker completion requires an intact native journal") + selected = ( + worker_feedback_model(feedback or run.get("worker_feedback") or {}) + .model_validate(feedback or run.get("worker_feedback") or {}) + .model_dump(mode="json") + ) + _eligible(run, task, selected) + grouped = _unsettled(entries, run["id"]) + keys = [entry["idempotency_key"] for entry in grouped[CUSTODY[0]]] + if run["bridge_completion_hold"] != { + "version": 1, + "reason": "graph_checkpoint_unavailable", + "request_id": keys[0], + "intent_digest": grouped[CUSTODY[0]][0]["digest"], + }: + raise ValueError("Observed hold does not identify the first unsettled completion") + contract = runtime_contract() + previous = None + previous_digest = invocation_id = None + closures = [] + final = captured = None + config = run.get("execution_config") or {} + endpoint = validate_endpoint(config.get("endpoint", "")) + endpoint = endpoint if endpoint.endswith("/v1") else endpoint + "/v1" + from firefly_forge.agentic_checkpoint_invocation import STATIC_TOOLS + + for intent, checkpoint, response, consume in zip(*(grouped[kind] for kind in CUSTODY), strict=True): + original = CompletionIntentPayload.model_validate(intent["payload"]) + payload = CheckpointPayloadV2.model_validate(checkpoint["payload"]) + binding = payload.invocation + if ( + any( + entry["run_id"] != run["id"] or entry["task_id"] != task["id"] + for entry in (intent, checkpoint, response, consume) + ) + or payload.request_id != intent["idempotency_key"] + or payload.intent_digest != intent["digest"] + or original.run_id != run["id"] + or original.task_id != task["id"] + or original.task_key != task["idempotency_key"] + or original.task_attempt != task["attempt"] + or original.project_id != run.get("project_id") + or original.repository_id != run.get("repository_id") + or original.profile_id != (run.get("model_selection") or {}).get("profile_id") + or original.config_field != "config_revision" + or original.config_revision != run.get("config_revision") + or original.scope_digest != _cut(run, "config_revision") + or original.descriptor.model != config.get("model", "bridge/default") + or original.descriptor.endpoint != endpoint + or original.descriptor.request_id != intent["idempotency_key"] + or binding.contract != contract + or binding.run_cut != invocation_cut(run) + or binding.round != task["attempt"] + or not binding.grants + or not set(binding.grants) <= STATIC_TOOLS + or (previous is None and (binding.ordinal != 1 or binding.predecessor is not None)) + or ( + previous is not None + and ( + binding.ordinal != previous.ordinal + 1 + or binding.predecessor != previous_digest + or binding.invocation_id != invocation_id + or binding.grants != previous.grants + ) + ) + or not intent["sequence"] < checkpoint["sequence"] < response["sequence"] < consume["sequence"] + ): + raise ValueError("Completion custody differs from its current runtime, provider, model or task scope") + AgentCheckpoints._validate_checkpoint(checkpoint, intent) + captured = BoundaryCheckpoint.decode(payload.snapshot.encode(), payload.checkpoint_digest) + body = CompletionResponsePayload.model_validate(response["payload"]) + if consume["payload"].get("version") != 1: + raise ValueError("Restored completion custody requires its original restore-ticket proof") + consumed = ConsumePayload.model_validate(consume["payload"]) + raw = body.response.encode() + if ( + body.request_id != intent["idempotency_key"] + or body.intent_digest != intent["digest"] + or body.response_sha256 != hashlib.sha256(raw).hexdigest() + or body.response_bytes != len(raw) + or consumed.request_id != body.request_id + 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("Known completion response and consume custody differ") + final = ChatCompletion.model_validate_json(raw, strict=True) + if ( + final.model != original.descriptor.model + or len(final.choices) != 1 + or final.choices[0].finish_reason not in {"stop", "tool_calls"} + ): + raise ValueError("Known completion does not identify one exact model outcome") + if intent is not grouped[CUSTODY[0]][-1]: + calls = final.choices[0].message.tool_calls or [] + schemas = { + item["name"]: item["parameters_json_schema"] + for item in captured.record.contract["parameters"]["function_tools"] + } + if final.choices[0].finish_reason != "tool_calls" or not calls: + raise ValueError("An earlier completion is not a declared intermediate tool result") + for call in calls: + if call.type != "function" or call.function.name not in schemas: + raise ValueError("An intermediate completion calls an undeclared function") + _schema(schemas[call.function.name], _json(call.function.arguments)) + closures.append( + { + "request_id": body.request_id, + "intent_digest": intent["digest"], + "checkpoint_record_digest": checkpoint["digest"], + "response_record_digest": response["digest"], + "consume_record_digest": consume["digest"], + } + ) + previous, previous_digest, invocation_id = binding, payload.checkpoint_digest, binding.invocation_id + assert final is not None and captured is not None + if _final_feedback(final, captured) != selected: + raise ValueError("Final model output differs from exact typed worker feedback") + active_feedbacks = [ + entry + for entry in entries + if entry["kind"] == "worker_feedback" and entry["sequence"] > grouped[CUSTODY[0]][0]["sequence"] + ] + feedbacks = [entry for entry in active_feedbacks if entry["payload"].get("id") == selected["id"]] + if ( + len(active_feedbacks) != len(feedbacks) + or len(feedbacks) > 1 + or ( + feedbacks + and ( + feedbacks[0]["payload"] != selected or feedbacks[0]["sequence"] <= grouped[CUSTODY[-1]][-1]["sequence"] + ) + ) + or (feedback is None and not feedbacks) + ): + raise ValueError("Known worker feedback has no unique consume-before-feedback custody") + prepared = PreparedWorkerCompletion( + len(entries), + digest({"entries": entries}), + _scope(run, task), + copy.deepcopy(selected), + tuple(closures), + hashlib.sha256(grouped[CUSTODY[2]][-1]["payload"]["response"].encode()).hexdigest(), + contract, + feedbacks[0]["digest"] if feedbacks else None, + "", + ) + from dataclasses import replace + + return replace(prepared, seal=_prepared_digest(prepared)) + + +def _prepared_digest(prepared: PreparedWorkerCompletion) -> str: + from dataclasses import asdict + + return digest({k: v for k, v in asdict(prepared).items() if k != "seal"}) + + +def validate_worker_completion(prepared: PreparedWorkerCompletion, run: Data, task: Data, entries: list[Data]) -> Data: + """Database-only final predicate; no filesystem, network, decode or result exposure.""" + if _prepared_digest(prepared) != prepared.seal: + raise ValueError("Prepared completion evidence changed") + _eligible(run, task, prepared.feedback) + if ( + not Journal.verify_chain(entries) + or _scope(run, task) != prepared.scope_digest + or digest({"entries": entries[: prepared.prefix_count]}) != prepared.prefix_digest + ): + raise ValueError("Known completion scope or immutable journal cut changed") + feedbacks = [ + entry + for entry in entries + if entry["kind"] == "worker_feedback" and entry["payload"].get("id") == prepared.feedback["id"] + ] + if ( + len(feedbacks) != 1 + or feedbacks[0]["payload"] != prepared.feedback + or run.get("worker_feedback") != prepared.feedback + or ( + prepared.existing_feedback_digest is not None + and feedbacks[0]["digest"] != prepared.existing_feedback_digest + ) + ): + raise ValueError("Known completion requires exact stored worker feedback projection") + if len(entries) != prepared.prefix_count + (prepared.existing_feedback_digest is None) or ( + prepared.existing_feedback_digest is None and entries[-1] != feedbacks[0] + ): + raise ValueError("Additional native effects changed the prepared completion cut") + proof = { + "version": 1, + "run_id": run["id"], + "task_id": task["id"], + "task_attempt": task["attempt"], + "task_key": task["idempotency_key"], + "runtime_contract": prepared.runtime_contract, + "completion_count": len(prepared.completions), + "completions": list(prepared.completions), + "feedback_record_digest": feedbacks[0]["digest"], + "feedback_sha256": digest(prepared.feedback), + "final_response_sha256": prepared.output_sha256, + "observed_hold_digest": digest(run["bridge_completion_hold"]), + "verification": "worker_output_known_only", + } + proof["digest"] = digest(proof) + return proof diff --git a/src/firefly_forge/known_worker_completions.py b/src/firefly_forge/known_worker_completions.py new file mode 100644 index 00000000..7d4989f3 --- /dev/null +++ b/src/firefly_forge/known_worker_completions.py @@ -0,0 +1,322 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Explicit native settlement of known worker output; no model or graph dispatch.""" + +from __future__ import annotations + +import copy +import hashlib +from pathlib import Path +from typing import Any + +from jsonschema.exceptions import ValidationError as SchemaError # type: ignore[import-untyped] +from pyfly.container import service + +from firefly_forge.bridge_completions import BridgeCompletions, _outside_unit +from firefly_forge.journal import Journal, digest +from firefly_forge.known_worker_completion_models import WorkerCompletionCandidateV1, WorkerCompletionInputV1 +from firefly_forge.known_worker_completion_proof import ( + SETTLEMENT, + prepare_worker_completion, + validate_worker_completion, +) +from firefly_forge.project_access import CURRENT_PRINCIPAL, AccessDenied, ProjectAccess +from firefly_forge.store import Store, identifier, prepared_credential_rejection + +Data = dict[str, Any] + + +def policy_digest() -> str: + _outside_unit() + root = Path(__file__).parent + return digest( + { + name: hashlib.sha256((root / name).read_bytes()).hexdigest() + for name in ( + "known_worker_completions.py", + "known_worker_completion_models.py", + "known_worker_completion_proof.py", + ) + } + ) + + +def _actor() -> Any: + actor = CURRENT_PRINCIPAL.get() + if actor is None or actor.authentication_method not in {"oidc_session", "oidc_bearer", "local_bootstrap"}: + raise AccessDenied("A current native operator must explicitly review worker completion") + return actor + + +def _entries(rows: Any, run_id: str) -> list[Data]: + result = [] + for row in rows: + value = copy.deepcopy(row.data) + if ( + any( + value.get(key) != item + for key, item in { + "run_id": row.run_id.hex, + "sequence": row.sequence, + "task_id": row.task_id, + "idempotency_key": row.idempotency_key, + "kind": row.kind, + "digest": row.digest, + }.items() + ) + or row.run_id.hex != run_id + ): + raise ValueError("Native completion row differs from its canonical columns") + result.append(value) + if not Journal.verify_chain(result): + raise ValueError("Native completion journal is not intact") + return result + + +def _task(row: Any) -> Data: + return {**copy.deepcopy(row.data), "id": row.task_id, "status": row.status, "attempt": row.attempt} + + +@service +class KnownWorkerCompletions: + def __init__(self, store: Store, journal: Journal) -> None: + self.store, self.journal = store, journal + self.completions = BridgeCompletions(store, journal) + + async def _access(self, run_id: str, actor: Any) -> Data: + run = await self.completions._reader(run_id, actor, execute=True) + if ( + run.get("status") in {"failed", "cancelled", "interrupted", "succeeded"} + or (run.get("cancel_request") or {}).get("status") == "requested" + ): + raise ValueError("Known worker settlement refuses an ended or cancelled execution") + await self.completions._execution(run, actor) + return run + + async def _read(self, run_id: str) -> tuple[Data, Data, list[Data]]: + actor = _actor() + run = await self._access(run_id, actor) + await self.journal.entries(run_id) + async with self.store._unit(): + current = await self._access(run_id, actor) + if current != run: + raise ValueError("Worker completion run changed during custody admission") + task = await self.journal._task(run_id, "implement") + rows = await self.journal.records.find_for_run(identifier(run_id)) + return current, _task(task), _entries(rows, run_id) + + async def _prepare(self, run_id: str) -> tuple[Data, Any]: + actor = _actor() + run, task, entries = await self._read(run_id) + try: + prepared = prepare_worker_completion(run, task, entries) + proof = validate_worker_completion(prepared, run, task, entries) + except (ValueError, KeyError, TypeError, SchemaError): + raise ValueError("Known worker completion is ineligible; inspect its retained native evidence") from None + value = { + "version": 1, + "run_id": run_id, + "actor_digest": digest(ProjectAccess.actor_snapshot(_actor())), + "policy_digest": policy_digest(), + "run_digest": digest(run), + "journal_digest": digest({"entries": entries}), + "proof": proof, + } + value["digest"] = digest(value) + candidate = WorkerCompletionCandidateV1.model_validate(value).model_dump(mode="json") + await self.store.admit_credentials(candidate) + async with self.store._unit(): + latest = await self._access(run_id, actor) + rows = await self.journal.records.find_for_run(identifier(run_id)) + if ( + digest(latest) != candidate["run_digest"] + or digest({"entries": _entries(rows, run_id)}) != candidate["journal_digest"] + ): + raise ValueError("Known worker candidate changed during credential admission") + self.store.reject_credentials(candidate) + if CURRENT_PRINCIPAL.get() != actor: + raise AccessDenied("Known worker completion operator changed during preparation") + return candidate, prepared + + async def candidate(self, run_id: str) -> Data: + _outside_unit() + with prepared_credential_rejection(self.store): + return (await self._prepare(identifier(run_id).hex))[0] + + @staticmethod + def _existing( + entries: list[Data], request_id: str, request: WorkerCompletionInputV1, actor_digest: str + ) -> Data | None: + from firefly_forge.known_worker_completion_models import WorkerCompletionReceiptV1 + + matches = [ + e for e in entries if e["kind"] == SETTLEMENT and e["idempotency_key"] == "worker-completion:" + request_id + ] + if not matches: + return None + if len(matches) != 1: + raise ValueError("Duplicate native completion receipt identity") + value = WorkerCompletionReceiptV1.model_validate(matches[0]["payload"]) + if ( + value.run_id != matches[0]["run_id"] + or value.request_id != request_id + or value.request != request + or value.actor_digest != actor_digest + ): + raise AccessDenied("Worker completion identity belongs to another input or operator") + return copy.deepcopy(matches[0]) + + async def request(self, run_id: str, request_id: str, request: WorkerCompletionInputV1) -> Data: + _outside_unit() + if identifier(request_id).hex != request_id: + raise ValueError("Use an exact lowercase native completion request identity") + run_id = identifier(run_id).hex + request = WorkerCompletionInputV1.model_validate(request.model_dump(mode="json")) + actor = _actor() + actor_digest = digest(ProjectAccess.actor_snapshot(actor)) + with prepared_credential_rejection(self.store): + await self.completions._reader(run_id, actor, execute=True) + async with self.store._unit(): + rows = await self.journal.records.find_for_run(identifier(run_id)) + existing = self._existing(_entries(rows, run_id), request_id, request, actor_digest) + if existing is not None: + return await self._admitted_receipt(run_id, existing, actor, execute=True) + try: + candidate, prepared = await self._prepare(run_id) + except ValueError: + async with self.store._unit(): + await self.completions._reader(run_id, actor, execute=True) + rows = await self.journal.records.find_for_run(identifier(run_id)) + existing = self._existing(_entries(rows, run_id), request_id, request, actor_digest) + if existing is None: + raise + return await self._admitted_receipt(run_id, existing, actor, execute=True) + expected = { + "candidate_digest": candidate["digest"], + "policy_digest": candidate["policy_digest"], + "expected_run_digest": candidate["run_digest"], + "journal_digest": candidate["journal_digest"], + "proof_digest": candidate["proof"]["digest"], + "feedback_record_digest": candidate["proof"]["feedback_record_digest"], + } + if ( + any(getattr(request, key) != value for key, value in expected.items()) + or candidate["actor_digest"] != actor_digest + ): + raise ValueError("The reviewed known-worker completion candidate changed") + await self.store.assert_ownership() + async with self.store._unit(): + frozen = await self._access(run_id, actor) + await self.completions._writer(frozen, actor) + row = await self.journal._run(run_id) + current = await self._access(run_id, actor) + rows = await self.journal.records.find_for_run(identifier(run_id)) + entries = _entries(rows, run_id) + existing = self._existing(entries, request_id, request, actor_digest) + if existing is not None: + self.store.reject_credentials(existing) + result = existing + else: + if digest(current) != candidate["run_digest"] or CURRENT_PRINCIPAL.get() != actor: + raise AccessDenied("Known worker run or operator changed before settlement") + task = await self.journal._task(run_id, "implement") + proof = validate_worker_completion(prepared, current, _task(task), entries) + if proof != candidate["proof"]: + raise ValueError("Known worker proof changed before settlement") + result = await self._commit(row, current, request_id, request, candidate, proof, actor) + return await self._admitted_receipt(run_id, result, actor, execute=True) + + async def _commit( + self, + row: Any, + current: Data, + request_id: str, + request: WorkerCompletionInputV1, + candidate: Data, + proof: Data, + actor: Any, + ) -> Data: + from firefly_forge.known_worker_completion_models import WorkerCompletionReceiptV1 + + payload = WorkerCompletionReceiptV1.model_validate( + { + "version": 1, + "run_id": current["id"], + "request_id": request_id, + "actor_digest": candidate["actor_digest"], + "candidate_digest": candidate["digest"], + "policy_digest": candidate["policy_digest"], + "run_digest": candidate["run_digest"], + "journal_digest": candidate["journal_digest"], + "request": request.model_dump(mode="json"), + "proof": proof, + "observed_hold": copy.deepcopy(current["bridge_completion_hold"]), + } + ).model_dump(mode="json") + self.store.reject_credentials(payload) + record = await self.journal._append( + current["id"], "implement", "worker-completion:" + request_id, SETTLEMENT, payload + ) + row.data = { + **row.data, + "bridge_completion_hold": {}, + "bridge_completion_settlement": { + "version": 1, + "request_id": request_id, + "record_digest": record["digest"], + "proof_digest": proof["digest"], + }, + } + await self.store.run_records.save(row) + await self.store.assert_ownership() + if CURRENT_PRINCIPAL.get() != actor: + raise AccessDenied("Known worker completion operator changed before commit") + self.store.reject_credentials(record) + return record + + async def get(self, run_id: str, request_id: str) -> Data: + actor = _actor() + run_id = identifier(run_id).hex + await self.completions._reader(run_id, actor) + async with self.store._unit(): + entries = _entries(await self.journal.records.find_for_run(identifier(run_id)), run_id) + matches = [ + e for e in entries if e["kind"] == SETTLEMENT and e["idempotency_key"] == "worker-completion:" + request_id + ] + if len(matches) != 1: + raise KeyError("Known worker completion receipt not found") + from firefly_forge.known_worker_completion_models import WorkerCompletionReceiptV1 + + value = WorkerCompletionReceiptV1.model_validate(matches[0]["payload"]) + if value.run_id != run_id or value.request_id != request_id: + raise ValueError("Known worker completion receipt scope differs") + return await self._admitted_receipt(run_id, matches[0], actor, execute=False) + + async def _admitted_receipt(self, run_id: str, record: Data, actor: Any, *, execute: bool) -> Data: + _outside_unit() + with prepared_credential_rejection(self.store): + await self.store.admit_credentials(record) + async with self.store._unit(): + await self.completions._reader(run_id, actor, execute=execute) + row = await self.journal.records.bridge_record( + identifier(run_id), record["idempotency_key"], SETTLEMENT + ) + if row is None: + raise ValueError("The native worker completion receipt disappeared") + current = self.completions._validate_record( + copy.deepcopy(row.data), + run_id, + record["idempotency_key"], + SETTLEMENT, + { + "run_id": row.run_id.hex, + "sequence": row.sequence, + "task_id": row.task_id, + "idempotency_key": row.idempotency_key, + "kind": row.kind, + "digest": row.digest, + }, + ) + if current != record or CURRENT_PRINCIPAL.get() != actor: + raise AccessDenied("Known worker receipt or reader changed during admission") + self.store.reject_credentials(record) + return copy.deepcopy(record) diff --git a/tests/test_known_worker_completion_native.py b/tests/test_known_worker_completion_native.py new file mode 100644 index 00000000..20b5b251 --- /dev/null +++ b/tests/test_known_worker_completion_native.py @@ -0,0 +1,205 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Prepared real PostgreSQL/ACL settlement cases with controlled SDK transport.""" + +import asyncio +import json +from contextlib import asynccontextmanager +from types import SimpleNamespace +from uuid import UUID, uuid4 + +import httpx2 +import pytest +from pydantic_ai import models +from test_agentic_checkpoint_activation_host import result +from test_agentic_restore_receipt_native import seed_restore_workspace +from test_bridge_completion_callbacks import CAPABILITY +from test_bridge_completion_native import mock_model_requests, prepared +from test_known_worker_completion_service_host import consent + +from firefly_forge.agent_roles import ProjectRoles, RoleRegistry +from firefly_forge.configuration import ProviderConfiguration +from firefly_forge.engine import Engine +from firefly_forge.git_workspace import metadata_digest +from firefly_forge.identity_persistence import ForgeIdentityUserRepository +from firefly_forge.journal import Journal +from firefly_forge.known_worker_completions import KnownWorkerCompletions +from firefly_forge.models import RunInput +from firefly_forge.project_access import AccessDenied +from firefly_forge.providers import BridgeProvider +from firefly_forge.runtime_roles import RuntimeRoles + + +@asynccontextmanager +async def known_native(store, tmp_path, monkeypatch, *, unknown=False): + async with prepared(store, tmp_path) as (_, actor, seeded, *_): + store.configuration.update("bridge", ProviderConfiguration(values={"checkpoints": "required"})) + run = await store.create_run( + RunInput( + repository_id=seeded["repository_id"], + project_id=seeded["project_id"], + component_id="api", + agent="bridge", + brief="Controlled known worker outcome", + skills=[], + ) + ) + run_id = run["id"] + await store.update( + run_id, + **store.configuration.snapshot("bridge"), + role_snapshot=RoleRegistry(tmp_path / "known-roles").snapshot( + ProjectRoles(disabled=["docs"], allowed_tools={"implementer": ["write_file"]}) + ), + resume_existing=True, + commands=[], + ) + journal = Journal(store) + registry = store.session_provider._registry() + assert registry is not None + for repository in (journal.records, journal.tasks): + repository._bind_transaction_managers(registry) + 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, tmp_path / "known-owned", lambda _: provider, journal=journal) + engine.enforce_ownership = True + workspace = engine.workspaces / run_id + revision = seed_restore_workspace(workspace) + await store.update(run_id, git_metadata_digest=metadata_digest(workspace)) + + async def inspect(*args, **kwargs): + return {"base_revision": revision, "head": revision, "source_tree": "c" * 40} + + monkeypatch.setattr(engine, "_inspect_workspace", inspect) + feedback = dict( + id="known-native-final", + role_id="implementer", + task_id="implement", + attempt=1, + disposition="blocked", + summary="A native decision is retained", + evidence_paths=[], + request=dict( + id="known-native-question", + kind="other", + question="Keep the existing value?", + context="The known response remains retained.", + choices=["Keep", "Review"], + ), + ) + posts = [] + + def transport(request): + if request.method == "GET": + return httpx2.Response(200, json={"completionRecovery": CAPABILITY}) + posts.append(request.content) + assert len(posts) <= 2 + if unknown and len(posts) == 2: + raise httpx2.ReadError("Controlled original outcome unknown", request=request) + return httpx2.Response( + 200, + json=result(tool=True) + if len(posts) == 1 + else result( + json.dumps({"summary": "Known worker output", "worker_feedback": feedback, "qa_report": None}) + ), + ) + + model.client._client._transport = httpx2.MockTransport(transport) + model.client._client._mounts = {} + runtime = RuntimeRoles(store) + await runtime.start() + try: + with models.override_allow_model_requests(False), mock_model_requests(model): + await engine.execute_run(run_id) + current = await store.get_run(run_id) + assert current["status"] == ("needs_reconciliation" if unknown else "paused"), current.get("error") + assert len(posts) == 2 and current["bridge_completion_hold"] + yield SimpleNamespace( + store=store, + journal=journal, + service=KnownWorkerCompletions(store, journal), + run_id=run_id, + actor=actor, + posts=posts, + workspace=workspace, + ) + finally: + await runtime.stop() + await model.client.close() + + +async def test_native_known_worker_receipt_projection_and_rollback_are_atomic(store, tmp_path, monkeypatch): + async with known_native(store, tmp_path, monkeypatch) as c: + candidate = await c.service.candidate(c.run_id) + request = consent(candidate) + before = await store.get_run(c.run_id) + prefix = await c.journal.entries(c.run_id) + original = store.run_records.save + + async def refused(row): + result = await original(row) + if row.data.get("bridge_completion_settlement"): + raise RuntimeError("Controlled native projection rollback") + return result + + with monkeypatch.context() as patch: + patch.setattr(store.run_records, "save", refused) + with pytest.raises(RuntimeError, match="projection rollback"): + await c.service.request(c.run_id, uuid4().hex, request) + assert await store.get_run(c.run_id) == before and await c.journal.entries(c.run_id) == prefix + request_id = uuid4().hex + record = await c.service.request(c.run_id, request_id, request) + after = await store.get_run(c.run_id) + assert ( + after["bridge_completion_hold"] == {} + and record["payload"]["observed_hold"] == before["bridge_completion_hold"] + ) + assert { + k: v for k, v in after.items() if k not in {"bridge_completion_hold", "bridge_completion_settlement"} + } == {k: v for k, v in before.items() if k != "bridge_completion_hold"} + assert await c.service.request(c.run_id, request_id, request) == record + assert len(c.posts) == 2 and (c.workspace / "service.py").read_text() == "VALUE = 1\n" + + +async def test_native_competing_settlements_have_one_receipt_without_extra_grant(store, tmp_path, monkeypatch): + async with known_native(store, tmp_path, monkeypatch) as c: + candidate = await c.service.candidate(c.run_id) + request = consent(candidate) + outcomes = await asyncio.gather( + *(c.service.request(c.run_id, uuid4().hex, request) for _ in range(2)), return_exceptions=True + ) + assert sum(isinstance(value, dict) for value in outcomes) == 1 + assert all(isinstance(value, (dict, ValueError, AccessDenied)) for value in outcomes) + entries = await c.journal.entries(c.run_id) + assert sum(e["kind"] == "worker_completion_receipt" for e in entries) == 1 + run = await store.get_run(c.run_id) + assert run["status"] == "paused" and run["worker_request"]["status"] == "pending" and len(c.posts) == 2 + assert not any(e["kind"] in {"resume", "agent_restore"} for e in entries) + + +async def test_native_current_actor_revocation_refuses_known_output_settlement(store, tmp_path, monkeypatch): + async with known_native(store, tmp_path, monkeypatch) as c: + candidate = await c.service.candidate(c.run_id) + before = await store.get_run(c.run_id) + async with store._unit(): + users = ForgeIdentityUserRepository() + user = await users.find_by_id(c.actor.id) + user.enabled = False + await users.save(user) + with pytest.raises(AccessDenied): + await c.service.request(c.run_id, uuid4().hex, consent(candidate)) + assert await store.get_run(c.run_id) == before and len(c.posts) == 2 + async with store._unit(): + rows = await c.journal.records.find_for_run(UUID(c.run_id)) + assert not any(row.kind == "worker_completion_receipt" for row in rows) + + +async def test_native_unknown_completion_never_becomes_a_known_worker_receipt(store, tmp_path, monkeypatch): + async with known_native(store, tmp_path, monkeypatch, unknown=True) as c: + before = await store.get_run(c.run_id) + with pytest.raises(ValueError, match="ineligible"): + await c.service.candidate(c.run_id) + assert await store.get_run(c.run_id) == before and len(c.posts) == 2 + entries = await c.journal.entries(c.run_id) + assert not any(e["kind"] == "worker_completion_receipt" for e in entries) diff --git a/tests/test_known_worker_completion_proof_host.py b/tests/test_known_worker_completion_proof_host.py new file mode 100644 index 00000000..94d53d0c --- /dev/null +++ b/tests/test_known_worker_completion_proof_host.py @@ -0,0 +1,191 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Known final output closes custody without model replay or a free graph reset.""" + +import copy +import importlib +import importlib.util +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 known_worker(tmp_path, monkeypatch): + env = await activated(tmp_path, monkeypatch) + posts = [] + feedback = dict( + id="known-final-worker", + role_id="implementer", + task_id="implement", + attempt=1, + disposition="blocked", + summary="A recorded decision is required", + evidence_paths=[], + request=dict( + id="known-final-question", + kind="other", + question="Keep the existing value?", + context="This local fixture retains its exact source.", + choices=["Keep", "Review"], + ), + ) + + def transport(request): + if request.method == "GET": + return httpx2.Response(200, json={"completionRecovery": CAPABILITY}) + posts.append(request.content) + assert len(posts) <= 2 + return httpx2.Response( + 200, + json=result(tool=True) + if len(posts) == 1 + else result( + json.dumps( + {"summary": "Known completed worker response", "worker_feedback": feedback, "qa_report": None} + ) + ), + ) + + 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) + run = await env.store.get_run(RUN) + assert run["status"] == "paused", run.get("error") + assert run["bridge_completion_hold"] and run["worker_feedback"] == feedback + 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") + env.known_run, env.known_task, env.known_entries, env.posts = run, task, entries, posts + try: + yield env + finally: + await env.model.client.close() + + +def proof_module(): + assert importlib.util.find_spec("firefly_forge.known_worker_completion_proof") is not None, ( + "Native known-worker completion proof is missing" + ) + return importlib.import_module("firefly_forge.known_worker_completion_proof") + + +async def test_real_sdk_known_worker_output_has_a_settleable_proof_without_any_response_replay(known_worker): + env = known_worker + before = copy.deepcopy((env.known_run, env.known_task, env.known_entries)) + prepared = proof_module().prepare_worker_completion(env.known_run, env.known_task, env.known_entries) + proof = proof_module().validate_worker_completion(prepared, env.known_run, env.known_task, env.known_entries) + assert proof["completion_count"] == 2 + assert proof["feedback_record_digest"] == next( + e["digest"] for e in env.known_entries if e["kind"] == "worker_feedback" + ) + assert proof["verification"] == "worker_output_known_only" + assert len(env.posts) == 2 + assert (env.known_run, env.known_task, env.known_entries) == before + assert "summary" not in json.dumps(proof) + + +def rechain(entries): + from firefly_forge.journal import ZERO_HASH, digest + + prior = ZERO_HASH + moved = {} + + def resolve(value): + if isinstance(value, dict): + return {key: resolve(item) for key, item in value.items()} + if isinstance(value, list): + return [resolve(item) for item in value] + return moved.get(value, value) if isinstance(value, str) else value + + for sequence, entry in enumerate(entries, 1): + old = entry["digest"] + entry["payload"] = resolve(entry["payload"]) + entry.update(sequence=sequence, previous_hash=prior) + entry["digest"] = digest({key: value for key, value in entry.items() if key != "digest"}) + moved[old] = prior = entry["digest"] + return entries + + +@pytest.mark.parametrize( + "change", + [ + "missing_consume", + "duplicate_intent", + "wrong_hold", + "wrong_profile", + "wrong_task", + "feedback_changed", + "old_contract", + "restored_consume", + ], +) +async def test_unknown_ambiguous_or_changed_custody_never_produces_a_settlement(known_worker, monkeypatch, change): + env = known_worker + run, task, entries = copy.deepcopy((env.known_run, env.known_task, env.known_entries)) + module = proof_module() + if change == "missing_consume": + entries.remove(next(e for e in entries if e["kind"] == "agent_consume")) + elif change == "duplicate_intent": + entries.append(copy.deepcopy(next(e for e in entries if e["kind"] == "bridge_request_intent"))) + elif change == "wrong_hold": + run["bridge_completion_hold"]["intent_digest"] = "f" * 64 + elif change == "wrong_profile": + run["model_selection"] = {**run.get("model_selection", {}), "profile_id": "another-profile"} + elif change == "wrong_task": + task["idempotency_key"] = "f" * 32 + elif change == "feedback_changed": + run["worker_feedback"]["summary"] = "An unsupported replacement" + elif change == "old_contract": + monkeypatch.setattr(module, "runtime_contract", lambda: "f" * 64) + else: + next(e for e in entries if e["kind"] == "agent_consume")["payload"]["version"] = 2 + rechain(entries) + with pytest.raises(ValueError): + module.prepare_worker_completion(run, task, entries) + assert env.store.run.data["bridge_completion_hold"] + assert len(env.posts) == 2 + + +async def test_final_predicate_needs_no_filesystem_or_sdk_decode_inside_writer(known_worker, monkeypatch): + env = known_worker + module = proof_module() + prepared = module.prepare_worker_completion(env.known_run, env.known_task, env.known_entries) + + def forbidden(*args, **kwargs): + raise AssertionError("No metadata or decode in the final writer predicate") + + monkeypatch.setattr(module, "runtime_contract", forbidden) + monkeypatch.setattr(module.BoundaryCheckpoint, "decode", forbidden) + async with env.store._unit(): + assert ( + module.validate_worker_completion(prepared, env.known_run, env.known_task, env.known_entries)[ + "completion_count" + ] + == 2 + ) + + +async def test_mutable_prepared_data_cannot_replace_the_proven_final_output(known_worker): + env = known_worker + module = proof_module() + prepared = module.prepare_worker_completion(env.known_run, env.known_task, env.known_entries) + prepared.feedback["summary"] = "Changed after proof" + with pytest.raises(ValueError, match="Prepared completion evidence changed"): + module.validate_worker_completion(prepared, env.known_run, env.known_task, env.known_entries) + + +def test_external_dynamic_schema_reference_is_refused_before_any_resolver(monkeypatch): + module = proof_module() + + def forbidden(*args, **kwargs): + raise AssertionError("External schema resolution must not begin") + + monkeypatch.setattr(module, "Draft202012Validator", forbidden) + with pytest.raises(ValueError, match="external references"): + module._schema({"$dynamicRef": "https://example.invalid/schema"}, {}) diff --git a/tests/test_known_worker_completion_service_host.py b/tests/test_known_worker_completion_service_host.py new file mode 100644 index 00000000..c0284e5f --- /dev/null +++ b/tests/test_known_worker_completion_service_host.py @@ -0,0 +1,199 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Native rollback semantics with explicit host ACL substitutes, not physical-lock proof.""" + +import copy +import importlib +import importlib.util +from uuid import uuid4 + +import pytest +from test_known_worker_completion_proof_host import known_worker as known_worker + +from firefly_forge.project_access import AccessDenied, ProjectAccess +from firefly_forge.store import Store + + +@pytest.fixture +async def settlement(known_worker, monkeypatch): + from types import MethodType + + env = known_worker + env.allowed = True + + async def allow(*args, **kwargs): + if not env.allowed: + raise AccessDenied("Controlled current project permission revoked") + + monkeypatch.setattr(ProjectAccess, "require_project", allow) + monkeypatch.setattr(ProjectAccess, "require_repository", allow) + env.store.reject_credentials = MethodType(Store.reject_credentials, env.store) + + async def admit(value): + env.store.reject_credentials(value) + + env.store.admit_credentials = admit + token = env.principal_context.set(env.principal) + try: + yield env + finally: + env.principal_context.reset(token) + + +def service(env): + assert importlib.util.find_spec("firefly_forge.known_worker_completions"), ( + "Native known-worker settlement is missing" + ) + return importlib.import_module("firefly_forge.known_worker_completions").KnownWorkerCompletions( + env.store, env.journal + ) + + +def consent(candidate): + from firefly_forge.known_worker_completion_models import WorkerCompletionInputV1 + + return WorkerCompletionInputV1( + version=1, + candidate_digest=candidate["digest"], + policy_digest=candidate["policy_digest"], + expected_run_digest=candidate["run_digest"], + journal_digest=candidate["journal_digest"], + proof_digest=candidate["proof"]["digest"], + feedback_record_digest=candidate["proof"]["feedback_record_digest"], + acknowledge_known_worker_output=True, + acknowledge_no_execution_grant=True, + ) + + +async def test_typed_receipt_and_observed_hold_projection_commit_once_without_replay(settlement): + env = settlement + api = service(env) + before = copy.deepcopy(env.store.run.data) + candidate = await api.candidate(env.known_run["id"]) + request = consent(candidate) + request_id = uuid4().hex + record = await api.request(env.known_run["id"], request_id, request) + assert record["kind"] == "worker_completion_receipt" + assert record["payload"]["proof"] == candidate["proof"] + assert record["payload"]["observed_hold"] == before["bridge_completion_hold"] + assert env.store.run.data["bridge_completion_hold"] == {} + assert { + k: v + for k, v in env.store.run.data.items() + if k not in {"bridge_completion_hold", "bridge_completion_settlement"} + } == {k: v for k, v in before.items() if k != "bridge_completion_hold"} + assert await api.request(env.known_run["id"], request_id, request) == record + assert len([r for r in env.journal.records.rows if r.kind == "worker_completion_receipt"]) == 1 + assert len(env.posts) == 2 + + +async def test_operator_change_during_candidate_admission_never_returns_a_candidate(settlement): + env = settlement + api = service(env) + original = env.store.admit_credentials + + async def changed(value): + await original(value) + if isinstance(value, dict) and "actor_digest" in value and "proof" in value: + env.principal_context.set(None) + + env.store.admit_credentials = changed + with pytest.raises(AccessDenied): + await api.candidate(env.known_run["id"]) + assert env.store.run.data["bridge_completion_hold"] + + +async def test_receipt_insert_and_projection_rollback_together_on_save_failure(settlement, monkeypatch): + env = settlement + api = service(env) + candidate = await api.candidate(env.known_run["id"]) + before = copy.deepcopy((env.store.run.data, [row.data for row in env.journal.records.rows])) + original = env.store.run_records.save + + async def failed(row): + await original(row) + raise RuntimeError("Controlled projection storage failure") + + monkeypatch.setattr(env.store.run_records, "save", failed) + with pytest.raises(RuntimeError, match="storage failure"): + await api.request(env.known_run["id"], uuid4().hex, consent(candidate)) + assert (env.store.run.data, [row.data for row in env.journal.records.rows]) == before + assert len(env.posts) == 2 + + +async def test_revoked_operator_and_stale_candidate_cannot_change_the_observed_hold(settlement): + env = settlement + api = service(env) + candidate = await api.candidate(env.known_run["id"]) + env.allowed = False + with pytest.raises(AccessDenied): + await api.request(env.known_run["id"], uuid4().hex, consent(candidate)) + env.allowed = True + env.store.run.data["verification_refusal"] = {"kind": "changed independently"} + with pytest.raises(ValueError, match="candidate changed"): + await api.request(env.known_run["id"], uuid4().hex, consent(candidate)) + assert env.store.run.data["bridge_completion_hold"] + assert not any(row.kind == "worker_completion_receipt" for row in env.journal.records.rows) + + +async def test_request_identity_cannot_replace_an_already_confirmed_input(settlement): + env = settlement + api = service(env) + candidate = await api.candidate(env.known_run["id"]) + request = consent(candidate) + request_id = uuid4().hex + receipt = await api.request(env.known_run["id"], request_id, request) + other = request.model_copy(update={"proof_digest": "f" * 64}) + with pytest.raises(AccessDenied, match="another input or operator"): + await api.request(env.known_run["id"], request_id, other) + assert await api.get(env.known_run["id"], request_id) == receipt + assert len(env.posts) == 2 + + +async def test_same_actor_permission_revocation_during_candidate_scan_refuses(settlement): + env = settlement + api = service(env) + original = env.store.admit_credentials + + async def revoke(value): + await original(value) + if isinstance(value, dict) and "actor_digest" in value and "proof" in value: + env.allowed = False + + env.store.admit_credentials = revoke + with pytest.raises(AccessDenied): + await api.candidate(env.known_run["id"]) + assert env.store.run.data["bridge_completion_hold"] + + +@pytest.mark.parametrize("route", ["get", "idempotent_post"]) +async def test_new_credential_in_retained_metadata_blocks_every_receipt_return(settlement, route): + env = settlement + api = service(env) + candidate = await api.candidate(env.known_run["id"]) + request = consent(candidate) + request_id = uuid4().hex + receipt = await api.request(env.known_run["id"], request_id, request) + secret = receipt["payload"]["proof"]["feedback_sha256"] + env.store.credential_registry = lambda: {"RETAINED_OUTPUT_TOKEN": secret} + with pytest.raises(ValueError, match="configured credentials"): + if route == "get": + await api.get(env.known_run["id"], request_id) + else: + await api.request(env.known_run["id"], request_id, request) + assert len([row for row in env.journal.records.rows if row.kind == "worker_completion_receipt"]) == 1 + assert len(env.posts) == 2 + + +async def test_terminal_run_refuses_before_the_live_execution_guard(settlement, monkeypatch): + env = settlement + api = service(env) + env.store.run.data["status"] = "failed" + env.store.run.status = "failed" + + async def unexpected(*args, **kwargs): + raise AssertionError("An ended run must not reach the live completion guard") + + monkeypatch.setattr(api.completions, "_execution", unexpected) + with pytest.raises(ValueError, match="ended or cancelled"): + await api.candidate(env.known_run["id"]) + assert env.store.run.data["bridge_completion_hold"]