diff --git a/CHANGELOG.md b/CHANGELOG.md
index 0f6e489f..eb9faed1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -2,6 +2,17 @@
+## 0.2.43 — 2026-10-24
+
+- Allocate command ordinals durably across service instances and restored graphs. Preserve exact explicit replay, unknown-effect holds and run-lifetime command budgets.
+- Restore fully completed command history through canonical native WAL and consume receipts. Keep changed source, authority, runtime contracts and interrupted effects blocked.
+- Compress accumulated native checkpoint history losslessly within an explicit 8 MiB raw bound while preserving the 192 KiB snapshot and 256 KiB WAL limits. Scan decoded records against current credential custody, including generic journal reads.
+- Stage complete branding files with a compact manifest, send the composed brief once, and retain the nine granted static checkpoint tools and exact frozen wireframe descriptors.
+- Preserve verified GitHub delivery links and bind finite maintenance communication to the exact new native creation receipt. Lifecycle notices and owner answers grant no resume or new execution authority.
+- Stream checkpoint identity from tracked Git source, including ignored tracked assets, and nonignored untracked source. Generated ignored dependencies no longer turn a host checkpoint into a transport-archive failure; runner archive transfer remains separately guarded.
+- Page source listing and text search explicitly, classify binary source, and hash every eligible file completely for manifests without the old 2,000-path or 1 MiB manifest cutoff.
+- Bind the V12 source profile to an explicit 35-module admission closure and 22 loader inputs with fresh same-source native/SQL measurements, retaining all seven historical profiles. Studio remains on its unchanged 0.2.42 assets and captures; see the [release notes](docs/release-notes.md#0243) for verification boundaries.
+
## 0.2.42 — 2026-10-24
- Complete an uncertain remote Git checkpoint only through a new finite native request, preserving the original unknown request, retained review body, current permissions and exact branch checks.
diff --git a/docs/agent-tools.md b/docs/agent-tools.md
index 6fcdd945..bcf55b54 100644
--- a/docs/agent-tools.md
+++ b/docs/agent-tools.md
@@ -9,10 +9,10 @@ Firefly Agentic registers the tools allowed by the run's frozen role configurati
| Tool | Scope and evidence |
| --- | --- |
-| `list_files` | Up to 2,000 component source paths; skips dependency trees, credentials and symlinks. |
+| `list_files` | Versioned pages of component source paths, with a source-view digest and continuation cursor. A page is never the whole project unless its scope is complete. |
| `read_file` | Bounded UTF-8 source with its SHA256 hash. |
| `write_file` | Complete bounded source content in a relative component path. |
-| `search_text` | Literal text search with line numbers, file hashes and an explicit truncation flag. |
+| `search_text` | Complete selected text search with finite result pages, full-file hashes, explicit snippet truncation and binary/UTF-8 classification. |
| `edit_lines` | Inclusive line replacement requiring the current file hash, rechecked on the open write descriptor. |
| `git_diff` | A bounded component diff with external drivers, text conversion, filesystem monitors and hooks disabled. |
| `execute_command` | Bounded argv execution through the frozen host or Docker policy. |
@@ -36,6 +36,20 @@ The table is a conditional native registry, not a promise that every provider re
The separately qualified source-only Codex route has a different, measured contract: governed source list/read/search and source edits within its exact admitted workspace, plus only the utilities explicitly recorded in its supported profile. It does not inherit `execute_command`, browser tools, internet search or `read_run_evidence` from the ordinary native registry. The current binary, model metadata, package policy and runtime must match that qualified profile before dispatch. A configured provider, an available capability, a qualified source-only profile and a completed invocation are separate states; check the [tool prerequisites](tooling-prerequisites.md) and [source-only recovery requirements](recovery.md) for the applicable route. Owner guidance does not modify a call already in flight or become delivery authority for a sealed source-only invocation.
+## Continue through a large source tree
+
+The native `list_files` tool returns a version-one page with `files`, `view_sha256`, `next_cursor` and `complete`. Start without a cursor. If `complete` is false, send the returned cursor with the same prefix and page size. Keep going until the final page. The default page size is 100; up to 1,000 paths can be requested in one response. The admitted byte window can shorten a page further. These are response windows, not project file-count limits.
+
+Firefly Agentic uses `path_prefix`; the source-only CLI reader uses its existing `path` argument as the prefix. Both accept an optional `cursor`. The CLI also accepts `page_size` and requires a new unique `request_id` for each page. Reusing an ID reads its exact stored result; changing that ID's cursor or arguments is refused. Old requests without paging arguments remain valid and receive an explicit first page. Internal Python callers of `WorkspaceTools.list_files()` receive the complete eligible path list.
+
+A cursor binds the repository/component, prefix or search query, response options and complete filename/metadata view. A changed view is refused: start a new listing instead of combining pages from different source states. Cursor contents are navigation data, not execution or permission grants.
+
+Git workspaces select tracked files, including tracked files covered by ignore rules, plus nonignored untracked files. Ignored installed dependencies are not treated as product source. Selected private paths, unsafe links, aliases and conflicts refuse; they are not silently counted as inspected. Standalone non-Git callers keep confined source traversal. Present but malformed Git metadata never triggers a filesystem fallback.
+
+`search_text` applies the prefix before scanning and can find text in files larger than the ordinary whole-file read window. Its pages retain the full-file SHA256 and report `snippet_truncated` separately from result-page continuation. Binary and invalid-UTF8 files are counted explicitly; unreadable or changed source refuses the operation. `complete` means the selected text-search page sequence has reached its end, not that binary content was searched as text.
+
+Before and after a native command, the internal manifest streams all selected file bytes, including large and binary files, without using a reader page or a fixed aggregate byte cutoff. Source identity is rechecked before returning. This complete accounting does not remove the separate role, command, provider or host budgets. It also does not expand the existing whole-file write/edit window; large-file editing needs its own supported interface.
+
## Observe a Python workload
Grant both `profile_python` and `execute_command` to the implementation role before creating the run. The repository runner must use an immutable Docker image, archive workspace transfer and network `none`, without a pool binding. Its image must already contain `python3` and the workload's dependencies. Host, bind-mount, network-enabled and pool profiles do not receive this optional tool. Readiness checks runner availability; the invocation still determines whether the image has a working Python interpreter. The default six source grants are unchanged.
diff --git a/docs/architecture.md b/docs/architecture.md
index 06990f73..16dba101 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -516,7 +516,9 @@ flowchart LR
Save the original checkpoint and intent before submitting to Bridge. If the result is uncertain, retain its identity, read or check the original result, then review one continuation. Continuing commits the one-use consume receipt before native parsing, followed by implementation and verification. On a narrow screen, scroll horizontally or focus the diagram and use the arrow keys.
-The restore ticket binds the original checkpoint, intent and selected result record. The execution owner repeats authority, source, configuration and deadline checks before restoring native Agentic history with `prompt=None`. The selected response is consumed once before the SDK parser receives it. A failure after that commit cannot authorize replay. A later repair keeps its own task identity; its mutable task key cannot replace the original implementation receipt. Owner recovery retains a committed consume boundary and holds unfinished tickets for reconciliation. See the [operator recovery journey](recovery.md#review-a-held-result-in-studio) for supported file tools and current limits.
+The restore ticket binds the original checkpoint, intent and selected result record. The execution owner repeats authority, source, configuration and deadline checks before restoring native Agentic history with `prompt=None`. The selected response is consumed once before the SDK parser receives it. A failure after that commit cannot authorize replay. A later repair keeps its own task identity; its mutable task key cannot replace the original implementation receipt. Owner recovery retains a committed consume boundary and holds unfinished tickets for reconciliation. Static command, knowledge and evidence tools retain explicit grants. Command-enabled restoration verifies bounded native WAL against canonical, fully completed durable command records. Prior effects must precede the selected checkpoint; later command creation must follow the matching restore consume. Unknown, interrupted, orphaned or altered effects refuse. Fresh commands use the next durable ordinal and preserve lifetime budgets. The checkpoint runtime contract binds the restore implementation, preventing an older contract from silently gaining this behavior. See the [operator recovery journey](recovery.md#review-a-held-result-in-studio) for supported tools and current limits.
+
+Boundary snapshots retain lossless canonical native history within an explicit 8 MiB raw bound. New records above 64 KiB use zlib/base64 within the existing 192 KiB snapshot and 256 KiB WAL limits. 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. Source admission V11 adds the decoder to its explicit 34-module closure; the checkpoint runtime contract binds custody producer and consumer.
`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.
@@ -590,6 +592,10 @@ Source: [`api.py`](https://github.com/fireflyframework/firefly-forge/blob/main/s
## Agent roles, native tools and cited context
+Product branding enters implementation through a compact file manifest. Engine validates the frozen kit's digest and admits the complete assets against current credentials, then materializes them under `docs/brand//` inside the selected component. Each manifest entry carries the component-relative path, UTF-8 byte count and SHA-256. Complete tokens, logos, styles and font-license text stay available through the governed file tools. Existing conflicting content refuses materialization; the file writer also checks the empty-file hash before creating a missing asset. Cancellation, ownership and current project authority are checked around effects.
+
+The manifest passes through the existing brand-instruction template. The complete assembled development brief appears once in the native user prompt; backend/frontend instructions, tool rules and frozen QA descriptors remain separate instructions. Embedded SVG and font CSS therefore stay in source files, while requirements and knowledge remain complete in the brief. This changes context representation without increasing checkpoint limits or silently cutting content. Engine and runtime changes require a newly measured source capability profile before native admission.
+
Inspect instructions, tool authority and knowledge scope
diff --git a/docs/recovery.md b/docs/recovery.md
index 123bb5e4..b1cbe560 100644
--- a/docs/recovery.md
+++ b/docs/recovery.md
@@ -427,7 +427,13 @@ Without explicit checkpoint selection this storage service remains unused by exe
### Explicit closed-text implementation continuation
-New Bridge implementations may select `checkpoints: required` together with `completion_recovery: required` in saved provider configuration or an implementation model profile. These optional fields freeze into the new run; absent fields leave historical snapshots and defaults unchanged. The initial slice requires an explicit static grant containing only `list_files`, `read_file`, `write_file`, `search_text`, `edit_lines` and `git_diff`. Unsupported grants, streaming, vision, QA state, deferred tools, retry state, dynamic instructions or dynamic tool catalogs refuse before completion submission. Direct API providers and independent review do not support required checkpoints; use a separate review profile with checkpoints disabled.
+New Bridge implementations may select `checkpoints: required` together with `completion_recovery: required` in saved provider configuration or an implementation model profile. These optional fields freeze into the new run; absent fields leave historical snapshots and defaults unchanged. The supported static text tools are `list_files`, `read_file`, `write_file`, `search_text`, `edit_lines`, `git_diff`, `execute_command`, `knowledge_lookup` and `read_run_evidence`. Each needs an explicit frozen role grant and its configured native executor. A grant does not create runner, knowledge or evidence access by itself. Unsupported grants, streaming, vision, QA state, deferred tools, retry state, dynamic instructions or dynamic tool catalogs refuse before completion submission. Direct API providers and independent review do not support required checkpoints; use a separate review profile with checkpoints disabled.
+
+Reviewed wireframe references may accompany those tools as textual descriptors. They must exactly match the native descriptors derived from the frozen design bundle, product review and frontend choice, including hashes and ordering. Missing or changed references are refused. The generator's Engine source joins the runtime contract, and both snapshots remain bound to the invocation. Descriptor text grants no image, browser or other QA tool access; active QA and multimodal checkpoint history remain unsupported.
+
+Commands keep their sandbox intents, receipts and lifetime budget. Completed text results and changed source paths enter the next closed checkpoint history. A command-enabled restore requires every retained execution to have a canonical durable v2 identity, matching native WAL, a completed result and confirmed cleanup under the frozen runner policy. Every prior effect must finish before the selected checkpoint. Pending, interrupted, orphaned, altered or unreconciled effects refuse restoration; explicit stage reconciliation does not silently make an interrupted command restorable.
+
+After the original result is consumed, later completed commands must have creation intents after that exact restore ticket and generation's consume record. Forge allocates their next durable ordinal; restoring a graph does not replay an earlier command, reset command or browser budgets, or expose a consumed result again. Read-only restoration retains its existing behavior. Runtime contract hashing includes the restore implementation, so this correction does not make older checkpoint contracts eligible. These checks require their own native SQL qualification before a runtime adopts them.
In Studio, open **Connections → Coding AI → Bridge → Configure**. Under **Original model results**, select **Required: retain the original result**, then **Required: save implementation checkpoints**. Saving these settings does not test the connection or change an existing run. **Keep current default (unset)** omits the field from this save; it does not clear a saved value. Choose **Disabled** to stop requiring checkpoints. The other choices explicitly save `legacy`/`required` recovery or `disabled`/`required` checkpoints. The equivalent provider API is `PUT /api/v1/providers/bridge/configuration` with `values` containing the selected fields. Named implementation profiles use `PUT /api/v1/model-profiles/{id}`.
@@ -498,6 +504,11 @@ These controlled Studio fixtures show a saved result, an acknowledged review and
The real Agentic implementation captures a closed native message history at the exact HTTP request boundary. Version-2 checkpoint envelopes bind the original composed instructions, runtime and prompt contracts, cumulative native usage and limits, absolute deadline, task attempt, file-tool state and retained owned workspace. The original Bridge intent and checkpoint commit together before POST. Both ordinary and restored results commit a consume receipt before reaching the native parser. Version-1 checkpoints and HTTP-only consume receipts retain their original semantics and cannot authorize restoration.
+Accumulated native history uses a lossless boundary codec. New canonical records above 64 KiB use a version-2 canonical JSON envelope containing standard zlib/base64 data, the exact raw byte count and SHA-256. Small records and legacy raw version-1 decoding retain their original encoding. The raw canonical record has an explicit **8 MiB** limit. Bounded decompression rejects truncated streams, trailing or concatenated data, size/hash mismatches and noncanonical envelopes. It never truncates messages, tool results, usage counters or file-tool state. The encoded snapshot remains bounded at **192 KiB**; its invocation binding, JSON escaping, WAL record and consume reserve must still fit **256 KiB**. Incompressible history or other exhausted custody/credential budgets remain held before dispatch. This is bounded checkpoint capacity, not unbounded enterprise history.
+
+Capture, consumption, restoration and authenticated generic journal reads inspect the full decoded canonical record against current credential custody. Generic reads check one record at a time outside native database units and return the original encoded WAL. A restored request must first match native model state and original wire bytes; it then reuses the original encoded snapshot and digest. Valid zlib streams do not require byte-identical recompression across library versions. This codec does not grant restoration authority to an outer version-1 WAL checkpoint or re-arm consumed results. Source profile V11 explicitly binds the Journal/decoder dependency and requires fresh qualification; earlier profiles retain their exact provenance.
+
+
An authorized operator first selects an immutable original HTTP result or retains a successful original-result observation through the existing `POST /api/v1/runs/{run_id}/bridge-requests/{completion_request_id}/observations` route. Then POST `/api/v1/runs/{run_id}/agent-checkpoints/{completion_request_id}/restores` with a retained 32-character hexadecimal `request_id`, `expected_checkpoint_digest`, `expected_intent_digest` and a tagged `source`. An `http_response` source has its `record_digest`; a `bridge_observation` source also has `observation_id` and the original `terminal_digest`. The server does not accept an arbitrary response body. GET the same route with `/{request_id}` for exact readback; a missing or uncertain response never authorizes another admission. Each checkpoint permits one ticket, within a maximum of 50 restore tickets per run and the shared bounded worker queue.
Discover completion request IDs through `GET /api/v1/runs/{run_id}/bridge-requests` and read the selected intent at `/{completion_request_id}`. The authenticated run journal (`GET /api/v1/runs/{run_id}/journal`) contains the matching `agent_checkpoint`, any `bridge_request_response` and their immutable record digests. Select the version-2 checkpoint whose idempotency key matches that request; its record digest is `expected_checkpoint_digest` and its payload binds `expected_intent_digest`. For an observation source, retain the observation POST response and its successful original terminal digest. Keep the client-generated restore request ID before POST, then read that exact ticket after a lost response; do not generate a replacement ID.
diff --git a/docs/release-notes.md b/docs/release-notes.md
index f252c366..93b3624a 100644
--- a/docs/release-notes.md
+++ b/docs/release-notes.md
@@ -5,6 +5,22 @@ description: Review firefly forge release changes, runtime requirements and the
# firefly forge release notes
+## 0.2.43
+
+Command slots now come from a durable, atomic run/task/attempt cursor. Fresh service instances allocate a new invocation; an explicit ordinal retains exact replay semantics. Pending, unknown, interrupted or unowned effects hold further dispatch. Command-enabled graph restoration accepts canonical, fully completed durable v2 records only, checks their position before the selected checkpoint or after the matching consume, and preserves lifetime command and browser budgets.
+
+Accumulated native history uses a lossless, versioned canonical zlib/base64 envelope above 64 KiB. The raw record is limited to 8 MiB, the encoded snapshot to 192 KiB, and the complete WAL envelope and reserve to 256 KiB. Bounded decoding verifies the raw size and hash, canonical encoding, complete stream and absence of trailing data. Capture, consume, restore and generic journal reads scan the full decoded record against current credentials. A restored request retains its original encoded bytes and must match native history, usage, limits, state and wire identity. Unknown or consumed results are never re-armed.
+
+Complete branding assets are staged through ordinary workspace tools while the model receives their compact path/hash manifest and one composed brief. Checkpointed execution retains the nine explicitly granted static tools and exact frozen wireframe descriptors. GitHub status cards can preserve only verified native PR/commit links for the current repository; the existing path, credential, image, mention and length guards remain active.
+
+A native API configuration can bind finite maintenance communication to one exact server-owned creation receipt and checkpoint lineage. The configured issue can receive lifecycle and worker-question notices, and its mapped owner can answer through the existing native worker-response path. An answer does not resume execution, inherit a prior run's grants, reset a budget or launch another unit. Natural issue text does not create this maintenance binding or a new maintenance run. See [SCM setup](scm.md#bind-a-fresh-maintenance-unit-to-its-owners-issue).
+
+Checkpoint source identity streams tracked Git files, including ignored tracked assets, and nonignored untracked files through the current credential scanner. It binds source bytes, index, HEAD and the selected namespace without loading dependency trees into a TAR archive. Ignored generated output is outside this source fingerprint; it is not evidence of dependency integrity or successful installation. Docker and pool archive transport retain their independent byte limits, link rejection and publication checks.
+
+Source listing and search return explicit complete/continuation pages bound to the source view, scope and query. Internal manifests hash every eligible file completely, including binary and large files. Search scans complete UTF-8 source with bounded excerpts and explicit binary/invalid-UTF-8 classification. The exhaustive Python inventory remains available; response windows do not silently drop the rest of a repository.
+
+The V12 source profile binds an explicit 35-module admission closure and 22 loader inputs. It requires fresh same-source controlled native tool-boundary and PostgreSQL measurements, including the source-identity and paging cases. All seven historical profile literals and seals remain unchanged. Standard distribution payload checks, controlled transport tests and these qualifications remain separate from live model quality, completed generated-product delivery and GitHub acceptance. Backend 0.2.43 keeps the unchanged Studio 0.2.42 bundle and controlled captures; no screenshot version is relabeled.
+
## 0.2.42
An explicitly reviewed remote-completion request can confirm a saved local checkpoint after an uncertain push or draft-review outcome. It retains the original unknown request, exact AI-written review text and local checkpoint; a new finite consent grants remote completion only. The execution owner checks current authority, source, credentials and the exact expected branch before any write. A newly confirmed receipt records recovery provenance rather than claiming the original write succeeded. A later unmatched intent remains visible as uncertain even when an earlier draft was confirmed.
diff --git a/docs/sandbox.md b/docs/sandbox.md
index 2741585a..c6b3fbed 100644
--- a/docs/sandbox.md
+++ b/docs/sandbox.md
@@ -25,6 +25,18 @@ After acquiring its run lock and execution capacity, the runner rechecks coordin
Cleanup can still remove the invocation's owned containers and volumes after revocation or cancellation, with the existing ownership-label and coordinator checks. Removing execution permission must not prevent those resources from draining. Effects a command already performed remain part of its retained history and recovery process.
+## Preserve command identity across service restarts
+
+An omitted command ordinal represents a new invocation, even when its arguments match an earlier command. The factory allocates the next slot from canonical persisted run/task/attempt records and writes the create intent under the same database run lock. A fresh executor instance cannot restart that cursor or reset the total run budget. An explicit ordinal retains exact replay semantics: a matching completed result can be returned without executing the command again, while changed arguments are refused.
+
+Allocation checks the retained scalar identities, request payloads, effect receipts, and intact journal together. Missing or inconsistent projections, unknown creation/start/cleanup, interrupted effects, and uncertain workspace publication prevent fresh dispatch. An explicitly reconciled interruption or closed preparation can advance only after all command effects have known receipts and the original native task has a later explicit retry reconciliation. Merely increasing an attempt or restarting the service does not authorize another effect. Allocation and dispatch reject an ambient caller transaction; process and controller effects start after the factory's own intent transaction commits.
+
+Implementation checkpoint runtime contracts include the command allocator, executor, pool worker, and their policy/persistence modules. A checkpoint captured under different executor semantics cannot be restored silently. The conservative restore restriction and native source qualification gates still apply; this command identity contract does not by itself authorize command-enabled restoration.
+
+Checkpoint source identity is separate from command workspace transport. It streams every tracked Git file, including tracked assets matched by ignore rules, and every nonignored untracked file. A newly created, nonignored package lock is included. Ignored, untracked installed dependencies and generated outputs are outside this source identity; their presence does not prove a successful or reproducible installation. Selected private paths, source links, hard links, unreadable files, or concurrent source, index, HEAD, credential, or ignore-rule changes refuse the checkpoint. Complete file bytes are hashed and checked in bounded chunks without a total source-size or file-count truncation. Git metadata and the selected source are checked again before publishing the identity.
+
+The runner's archive transport retains its own explicit workspace byte budget and strict regular-file contract. Checkpoint source selection does not silently change transported files, import dependencies, or permit dependency links in an archive. A runner that needs installed dependencies must provide them through its configured workspace or image and retain the corresponding command receipts.
+
## Keep Python verification tied to current source
Factory commands receive a fresh execution-owned Python cache namespace and `PYTHONDONTWRITEBYTECODE=1`. The namespace replaces inherited or caller-supplied `PYTHONPYCACHEPREFIX` values. This prevents normal Python imports from accepting an earlier timestamp-and-size cache after a same-size source edit. Disabling bytecode writes alone would still allow an existing stale cache to be read.
diff --git a/docs/scm.md b/docs/scm.md
index ad9b626f..ffffae1f 100644
--- a/docs/scm.md
+++ b/docs/scm.md
@@ -1023,3 +1023,55 @@ pipeline without a mention or copied identifier. Saving that answer does not
renew source-only consent, resume execution, create a run or reset a budget.
The separate current native continuation gates still apply. Controlled host and
SQL fixtures are separate from real GitHub delivery acceptance.
+
+GitHub natural status replies can preserve PR and commit links from the current repository’s verified native delivery journal. The server binds these HTTPS links to the exact project, issue, actor, run, and checkpoint; model text cannot authorize another URL. This link rendering currently supports GitHub only. A local checkpoint alone does not establish a published PR or product acceptance. Credential, setting, private path, HTML, image, and mention guards still apply.
+
+### Bind a fresh maintenance unit to its owner's issue
+
+A maintenance unit created from a confirmed checkpoint has its own run, creation
+receipt, provider session and budgets. Its checkpoint parent supplies source
+provenance, not communication or execution authority. By default its issue
+lineage is read-only, just like a fresh checkpoint test. Neither a status request
+nor an ordinary project reader can upgrade it into a writable conversation.
+
+For a native administrator who can manage the project's members, the API offers
+explicit, finite setup at
+`POST /api/v1/projects/{project_id}/repositories/{repository_id}/maintain-ticket-bindings`.
+The strict version-1 request contains:
+
+- A new `request_id`, the maintain `run_id`, its `creation_request_id` and the
+ exact `expected_creation_digest` returned by run creation.
+- An actual API-observed owner message in `observed_receipt_id` and its
+ `expected_reference_digest`. The existing authenticated conversation must link
+ the checkpoint parent in the same repository, project and component.
+- `expected_previous_digest`, initially `null`, for an exact binding update.
+- A `lifetime_seconds` value from 60 through 3,600 and `max_notifications` from 1
+ through 32. These bound the new communication consent, not the run's coding
+ allowance.
+- Explicit `allowed_actions: ["status", "question", "answer"]` and
+ `acknowledge_native_communication: true`.
+
+Read the immutable result at the same base path plus
+`/requests/{request_id}`. Repeating the identical request recovers that result
+without extending its expiration. A new binding requires its own request and the
+current previous digest. An uncertain prior notification blocks replacement.
+
+After setup, the existing collaboration owner loop automatically posts lifecycle
+updates and supported pending worker questions to the confirmed issue. It checks
+current polling consent, repository scope, owner identity, project permissions,
+communication configuration and finite consent again before dispatch, including
+after waiting for the provider lock. Native configuration stores a clearly marked
+factory outbox; it does not invent an inbound GitHub event or modify
+`issue_context`, the parent's receipts, or the source-only question producer.
+
+A confirmed question comment lets its mapped owner reply naturally in the issue.
+The existing answer pipeline records the child's exact `worker_response` with
+current SCM attribution and leaves the child paused. There is no automatic
+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.
+
+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
+separate evidence; configuring the API alone does not establish live delivery.
diff --git a/pyproject.toml b/pyproject.toml
index 26b73971..3674953c 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -5,7 +5,7 @@ build-backend = "hatchling.build"
[project]
name = "firefly-forge"
-version = "0.2.42"
+version = "0.2.43"
description = "Firefly software factory: brief to verified product with interchangeable coding agents"
readme = "README.md"
requires-python = ">=3.13"
diff --git a/src/firefly_forge/__init__.py b/src/firefly_forge/__init__.py
index 056a1f1e..c0e8c016 100644
--- a/src/firefly_forge/__init__.py
+++ b/src/firefly_forge/__init__.py
@@ -1,4 +1,4 @@
# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0
"""firefly forge software factory."""
-__version__ = "0.2.42"
+__version__ = "0.2.43"
diff --git a/src/firefly_forge/agent_checkpoint_models.py b/src/firefly_forge/agent_checkpoint_models.py
index 5e03a730..d08af7f1 100644
--- a/src/firefly_forge/agent_checkpoint_models.py
+++ b/src/firefly_forge/agent_checkpoint_models.py
@@ -35,7 +35,7 @@ class InvocationBinding(StrictRecord):
frontend: str = Field(min_length=1, max_length=32)
round: int = Field(ge=1, le=1000000)
deadline_ms: int = Field(ge=1, le=2**53 - 1)
- grants: list[str] = Field(max_length=6)
+ grants: list[str] = Field(max_length=9)
contract: Digest
run_cut: Digest
workspace: dict[str, str] = Field(min_length=4, max_length=4)
diff --git a/src/firefly_forge/agent_checkpoints.py b/src/firefly_forge/agent_checkpoints.py
index 6f7ea25c..286767e4 100644
--- a/src/firefly_forge/agent_checkpoints.py
+++ b/src/firefly_forge/agent_checkpoints.py
@@ -46,6 +46,13 @@ def __init__(self, completions: BridgeCompletions) -> None:
self.completions = completions
self.store, self.journal = completions.store, completions.journal
+ @staticmethod
+ def credential_payload(checkpoint: Data) -> list[Any]:
+ _outside_unit()
+ payload = checkpoint["payload"]
+ captured = BoundaryCheckpoint.decode(payload["snapshot"].encode("utf-8"), payload["checkpoint_digest"])
+ return [checkpoint, captured.canonical_payload.decode("utf-8")]
+
async def _budget(self, run_id: str, *, extra: int = 0, reserve: bool = False, consume: bool = False) -> None:
budget = await self.journal.records.checkpoint_budget(identifier(run_id))
if (
@@ -131,7 +138,11 @@ async def checkpoint_admit(raw: Data, body: bytes) -> Data:
raise ValueError("Checkpoint differs from the original wire request")
snapshot = checkpoint.payload.decode("utf-8")
invocation = await capture_context() if capture_context is not None else None
- credential_payload = snapshot if invocation is None else {"snapshot": snapshot, "invocation": invocation}
+ credential_payload = {
+ "snapshot": snapshot,
+ "canonical_record": checkpoint.canonical_payload.decode("utf-8"),
+ "invocation": invocation,
+ }
await self.store.admit_credentials(credential_payload)
await current()
self.store.reject_credentials(credential_payload)
@@ -226,7 +237,11 @@ async def _exact(self, run_id: str, request_id: str) -> Data:
raise ValueError("Checkpoint consume receipt has different immutable bindings")
result = {"checkpoint": checkpoint, "consume": consumed}
self._validate_checkpoint(checkpoint, intent)
- return await self.completions._admitted_read(run_id, result, reader, scope)
+ credentials = self.credential_payload(checkpoint)
+ await self.store.admit_credentials(credentials)
+ value = await self.completions._admitted_read(run_id, result, reader, scope)
+ self.store.reject_credentials(credentials)
+ return value
@staticmethod
def _validate_checkpoint(checkpoint: Data, intent: Data) -> None:
@@ -276,7 +291,8 @@ async def _consume(
async with self.store._unit():
bundle = await self._bundle(run, intent, body)
self._validate_checkpoint(bundle[1], bundle[0])
- await self.store.admit_credentials(bundle)
+ credentials = (bundle, self.credential_payload(bundle[1]))
+ await self.store.admit_credentials(credentials)
await current()
async with self.store._unit():
await self.completions._writer(run, reader)
@@ -305,9 +321,9 @@ async def _consume(
if _size(consumed) > CONSUME_RESERVE:
raise ValueError("Checkpoint consumption exceeds its reserved envelope")
await current()
- self.store.reject_credentials(bundle)
+ self.store.reject_credentials(credentials)
await current()
- self.store.reject_credentials(bundle)
+ self.store.reject_credentials(credentials)
async def restore_bundle(
self, run_id: str, request_id: str, request: RestoreInput
@@ -364,7 +380,7 @@ async def restore_bundle(
or any(c.finish_reason not in {"stop", "tool_calls"} for c in completion.choices)
):
raise ValueError("Original result is not a supported successful completion")
- self.store.reject_credentials([intent, checkpoint, record])
+ self.store.reject_credentials([intent, self.credential_payload(checkpoint), record])
return intent, checkpoint, record, body
async def consume_restored(
@@ -381,7 +397,8 @@ async def consume_restored(
_outside_unit()
with prepared_credential_rejection(self.store):
bundle = await self.restore_bundle(run["id"], request_id, request)
- await self.store.admit_credentials(bundle[:3])
+ credentials = (bundle[:3], self.credential_payload(bundle[1]))
+ await self.store.admit_credentials(credentials)
await guard()
if outside_guard is not None:
await outside_guard()
@@ -419,7 +436,7 @@ async def consume_restored(
if _size(record) > CONSUME_RESERVE:
raise ValueError("Restoration consume exceeds its reserved envelope")
await guard()
- self.store.reject_credentials(bundle[:3])
+ self.store.reject_credentials(credentials)
await guard()
- self.store.reject_credentials(bundle[:3])
+ self.store.reject_credentials(credentials)
return body
diff --git a/src/firefly_forge/agent_restores.py b/src/firefly_forge/agent_restores.py
index a53022b7..0a486cc3 100644
--- a/src/firefly_forge/agent_restores.py
+++ b/src/firefly_forge/agent_restores.py
@@ -6,10 +6,11 @@
import asyncio
import copy
import hashlib
+import json
import time
import uuid
from pathlib import Path
-from typing import Any, cast
+from typing import Any, NoReturn, cast
from pyfly.data.relational.sqlalchemy import LockMode
from sqlalchemy import text
@@ -75,7 +76,9 @@ async def consume(self, checkpoint_digest: str, response_digest: str) -> None:
if body != self.body:
raise ValueError("Consumed result changed before exposure")
await self.guard_source()
- self.service.store.reject_credentials([self.record, self.body.decode("utf-8")])
+ self.service.store.reject_credentials(
+ [self.service.checkpoints.credential_payload(self.record), self.body.decode("utf-8")]
+ )
async def receipt(self, engine: Any, key: str, evidence: dict[str, Any], checkpoint: dict[str, Any]) -> None:
_outside_unit()
@@ -151,6 +154,103 @@ async def validate_db(
or task.status not in ({"running"} if active else {"needs_reconciliation"})
):
raise ValueError("Only the same unfinished implementation task attempt can restore")
+ await self._command_restore(run, context)
+
+ async def _command_restore(
+ self, run: dict[str, Any], context: InvocationBinding, *, ticket: dict[str, Any] | None = None
+ ) -> None:
+ if "execute_command" not in context.grants:
+ return
+ from firefly_forge.agent_checkpoint_models import ConsumePayloadV2
+ from firefly_forge.runner_factory_models import validate_factory_pool
+ from firefly_forge.sandbox import SandboxReconciliationRequired
+ from firefly_forge.sandbox_command_slots import COMMAND_CONTRACT, _canonical, _complete
+ from firefly_forge.sandbox_models import capture_sandbox, validate_sandbox_snapshot
+
+ def refuse() -> NoReturn:
+ raise ValueError("Prior runner custody or pending effects prevent command-enabled checkpoint restore")
+
+ async with self.store._unit():
+ entries = [
+ {**copy.deepcopy(row.data), "digest": row.digest}
+ for row in await self.journal.records.find_for_run(identifier(run["id"]))
+ ]
+ rows = await self.journal.sandboxes.for_run(identifier(run["id"]))
+ if len(entries) > 10000 or len(json.dumps(entries)) > 16 * 1024 * 1024 or len(rows) > 100:
+ refuse()
+ if not self.journal.verify_chain(entries):
+ refuse()
+ retained = {
+ entry["payload"].get("execution_id")
+ for entry in entries
+ if entry["kind"] in {"sandbox_intent", "sandbox_receipt", "sandbox_reconcile"}
+ }
+ if retained != {row.id.hex for row in rows} or len({row.id for row in rows}) != len(rows):
+ refuse()
+ if not rows:
+ return
+ checkpoints = [
+ entry
+ for entry in entries
+ if entry["kind"] == "agent_checkpoint"
+ and entry["payload"].get("invocation") == context.model_dump(mode="json")
+ ]
+ if len(checkpoints) != 1:
+ refuse()
+ checkpoint = checkpoints[0]
+ consumes = [
+ entry
+ for entry in entries
+ if entry["kind"] == "agent_consume" and entry["idempotency_key"] == checkpoint["idempotency_key"]
+ ]
+ consumed_at = None
+ if consumes:
+ if ticket is None or len(consumes) != 1:
+ refuse()
+ consumed = ConsumePayloadV2.model_validate(consumes[0]["payload"])
+ if (
+ consumed.restore_request_id != ticket["id"]
+ or consumed.generation != ticket["generation"]
+ or consumed.checkpoint_record_digest != checkpoint["digest"]
+ or consumed.intent_digest != checkpoint["payload"].get("intent_digest")
+ or consumes[0]["sequence"] <= checkpoint["sequence"]
+ ):
+ refuse()
+ consumed_at = consumes[0]["sequence"]
+ snapshot = run.get("sandbox_snapshot") or capture_sandbox(None)
+ policy = (
+ validate_factory_pool(snapshot)
+ if snapshot.get("mode") == "pool"
+ else validate_sandbox_snapshot(snapshot)
+ )
+ if len(rows) > policy.command_limit:
+ refuse()
+ browser_calls = 0
+ for row in rows:
+ try:
+ value = _canonical(row, entries, run["id"])
+ except (SandboxReconciliationRequired, KeyError, TypeError, ValueError):
+ refuse()
+ 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)
+ ):
+ refuse()
+ browser_calls += bool(value["request"]["request"].get("browser"))
+ effects = [
+ entry["sequence"]
+ for entry in entries
+ if entry["kind"] in {"sandbox_intent", "sandbox_receipt", "sandbox_reconcile"}
+ and entry["payload"].get("execution_id") == row.id.hex
+ ]
+ # Before exposure, every effect must already belong to the selected checkpoint cut.
+ if max(effects) >= checkpoint["sequence"] and (consumed_at is None or min(effects) <= consumed_at):
+ refuse()
+ browser = getattr(policy, "browser", None)
+ if browser_calls and (browser is None or browser_calls > browser.max_calls):
+ refuse()
async def request(self, run_id: str, completion_id: str, body: RestoreInput) -> dict[str, Any]:
with prepared_credential_rejection(self.store):
@@ -183,7 +283,8 @@ async def _request(self, run_id: str, completion_id: str, body: RestoreInput) ->
raise ValueError("Original checkpoint contract changed or deadline expired")
if await workspace_cut(self.store, self.workspace(run_id)) != context.workspace:
raise ValueError("Retained workspace drifted; no restore ticket admitted")
- await self.store.admit_credentials([incoming, bundle[:3]])
+ credentials = [incoming, bundle[:3], self.checkpoints.credential_payload(bundle[1])]
+ await self.store.admit_credentials(credentials)
if CURRENT_PRINCIPAL.get() != reader:
raise AccessDenied("Restore caller changed during admission")
async with self.store._unit() as session:
@@ -229,9 +330,9 @@ async def _request(self, run_id: str, completion_id: str, body: RestoreInput) ->
await self.requests.records.save(row)
await self.requests._event(row, "request_admitted", {"request_digest": data["request_digest"]})
await self.access(current, actor)
- self.store.reject_credentials([incoming, bundle[:3]])
+ self.store.reject_credentials(credentials)
await self.access(await self.store.get_run(run_id), actor)
- self.store.reject_credentials([incoming, bundle[:3]])
+ self.store.reject_credentials(credentials)
return await self.get(run_id, completion_id, body.request_id)
async def get(self, run_id: str, completion_id: str, request_id: str) -> dict[str, Any]:
@@ -328,6 +429,7 @@ async def guard(self, ticket: dict[str, Any]) -> None:
task = await self.journal._effect_task(run["id"], intent["payload"]["task_key"])
if task.status != "running" or task.attempt != intent["payload"]["task_attempt"]:
raise ValueError("Restore original task attempt changed")
+ await self._command_restore(run, context, ticket=ticket)
async def execute(self, request_id: str) -> None:
self.requests.runtime.require_execution()
diff --git a/src/firefly_forge/agentic_checkpoint_boundary.py b/src/firefly_forge/agentic_checkpoint_boundary.py
index c79b73f1..b6ae8300 100644
--- a/src/firefly_forge/agentic_checkpoint_boundary.py
+++ b/src/firefly_forge/agentic_checkpoint_boundary.py
@@ -3,10 +3,13 @@
from __future__ import annotations
+import base64
+import binascii
import dataclasses
import hashlib
import json
import math
+import zlib
from collections.abc import Awaitable, Callable, Iterator
from contextlib import contextmanager
from contextvars import ContextVar
@@ -16,7 +19,7 @@
from typing import Any, Literal, cast
from fireflyframework_agentic.model_utils import deserialize_model_messages, serialize_model_messages
-from pydantic import BaseModel, ConfigDict, TypeAdapter, field_validator
+from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError, field_validator
from pydantic_ai.capabilities import AbstractCapability
from pydantic_ai.messages import (
ModelMessage,
@@ -33,6 +36,8 @@
from pydantic_ai.usage import RunUsage, UsageLimits
MAX_PAYLOAD = 192 * 1024
+MAX_RAW_PAYLOAD = 8 * 1024 * 1024
+COMPRESSION_THRESHOLD = 64 * 1024
MAX_COUNTER = 2**53 - 1
USAGE = TypeAdapter(RunUsage)
LIMITS = TypeAdapter(UsageLimits)
@@ -47,7 +52,10 @@ def _compatibility() -> dict[str, str]:
def _native_round_trip(adapter: TypeAdapter[Any], value: dict[str, Any]) -> None:
- restored = adapter.validate_json(_encoded(value))
+ try:
+ restored = adapter.validate_json(_encoded(value))
+ except ValidationError:
+ raise ValueError("Checkpoint usage or limits contain unsupported fields") from None
if _encoded(adapter.dump_python(restored, mode="json")) != _encoded(value):
raise ValueError("Checkpoint usage or limits contain unsupported fields")
@@ -169,6 +177,79 @@ def exact_version(cls, value: Any) -> Any:
return value
+class _CompressedRecord(BaseModel):
+ model_config = ConfigDict(extra="forbid", strict=True)
+ version: Literal[2]
+ codec: Literal["zlib-base64"]
+ raw_bytes: int
+ raw_sha256: str
+ data: str
+
+ @field_validator("version", mode="before")
+ @classmethod
+ def exact_version(cls, value: Any) -> Any:
+ if type(value) is not int:
+ raise ValueError("Checkpoint envelope version must be an integer")
+ return value
+
+
+def _checkpoint_payload(raw: bytes) -> bytes:
+ if not 1 <= len(raw) <= MAX_RAW_PAYLOAD:
+ raise ValueError("Checkpoint raw record exceeds 8 MiB")
+ if len(raw) <= COMPRESSION_THRESHOLD:
+ return raw
+ return _encoded(
+ {
+ "version": 2,
+ "codec": "zlib-base64",
+ "raw_bytes": len(raw),
+ "raw_sha256": hashlib.sha256(raw).hexdigest(),
+ "data": base64.b64encode(zlib.compress(raw, level=9)).decode("ascii"),
+ }
+ )
+
+
+def _checkpoint_record(payload: bytes) -> bytes:
+ if not 1 <= len(payload) <= MAX_PAYLOAD:
+ raise ValueError("Checkpoint size differs")
+ try:
+ value = json.loads(payload)
+ except (ValueError, UnicodeDecodeError, RecursionError):
+ raise ValueError("Checkpoint JSON encoding differs") from None
+ if type(value) is not dict or value.get("version") != 2:
+ return payload
+ try:
+ envelope = _CompressedRecord.model_validate_json(payload)
+ except ValidationError:
+ raise ValueError("Checkpoint compression envelope schema differs") from None
+ if not 1 <= envelope.raw_bytes <= MAX_RAW_PAYLOAD:
+ raise ValueError("Checkpoint raw record exceeds 8 MiB")
+ try:
+ compressed = base64.b64decode(envelope.data, validate=True)
+ inflater = zlib.decompressobj()
+ raw = inflater.decompress(compressed, envelope.raw_bytes + 1)
+ except (binascii.Error, ValueError, zlib.error):
+ raise ValueError("Checkpoint compression encoding differs") from None
+ if (
+ len(raw) != envelope.raw_bytes
+ or not inflater.eof
+ or inflater.unused_data
+ or inflater.unconsumed_tail
+ or hashlib.sha256(raw).hexdigest() != envelope.raw_sha256
+ or base64.b64encode(compressed).decode("ascii") != envelope.data
+ or _encoded(envelope.model_dump(mode="json")) != payload
+ ):
+ raise ValueError("Checkpoint compression size, digest or canonical encoding differs")
+ return raw
+
+
+def _record(raw: bytes) -> _Record:
+ try:
+ return _Record.model_validate_json(raw)
+ except ValidationError:
+ raise ValueError("Checkpoint record schema differs") from None
+
+
@dataclasses.dataclass(frozen=True)
class BoundaryCheckpoint:
payload: bytes
@@ -184,10 +265,14 @@ def decode(cls, payload: bytes, digest: str) -> BoundaryCheckpoint:
raise ValueError("Checkpoint payload must be immutable bytes")
if not 1 <= len(payload) <= MAX_PAYLOAD or hashlib.sha256(payload).hexdigest() != digest:
raise ValueError("Checkpoint size or digest differs")
- record = _Record.model_validate_json(payload)
- if record.compatibility != _compatibility() or _encoded(record.model_dump(mode="json")) != payload:
+ raw = _checkpoint_record(payload)
+ record = _record(raw)
+ if record.compatibility != _compatibility() or _encoded(record.model_dump(mode="json")) != raw:
raise ValueError("Checkpoint runtime or canonical encoding differs")
- messages = deserialize_model_messages(record.history)
+ try:
+ messages = deserialize_model_messages(record.history)
+ except ValidationError:
+ raise ValueError("Checkpoint native history schema differs") from None
_history(messages)
if record.request_history != record.history:
raise ValueError("Request-only history cannot restore the agent graph")
@@ -205,7 +290,12 @@ def decode(cls, payload: bytes, digest: str) -> BoundaryCheckpoint:
@property
def record(self) -> _Record:
- return _Record.model_validate_json(self.payload)
+ return _record(_checkpoint_record(self.payload))
+
+ @property
+ def canonical_payload(self) -> bytes:
+ BoundaryCheckpoint.decode(self.payload, self.digest)
+ return _checkpoint_record(self.payload)
@property
def state(self) -> dict[str, Any]:
@@ -335,13 +425,19 @@ def capture(self, wire: bytes) -> BoundaryCheckpoint:
wire_sha256=hashlib.sha256(wire).hexdigest(),
wire_bytes=len(wire),
)
- if self.restore is not None and not invocation.restored:
+ retained = self.restore is not None and not invocation.restored
+ if retained:
+ assert self.restore is not None
expected = self.restore.record.model_dump(mode="json")
actual = record.model_dump(mode="json")
changed = [field for field in expected if _encoded(actual[field]) != _encoded(expected[field])]
if changed:
raise ValueError("Restored model state differs from the checkpoint: " + ", ".join(changed))
- payload = _encoded(record.model_dump(mode="json"))
+ payload = (
+ self.restore.payload
+ if retained and self.restore is not None
+ else _checkpoint_payload(_encoded(record.model_dump(mode="json")))
+ )
checkpoint = BoundaryCheckpoint.decode(payload, hashlib.sha256(payload).hexdigest())
contract_bytes = _encoded(record.contract)
if invocation.contract is not None and invocation.contract != contract_bytes:
diff --git a/src/firefly_forge/agentic_checkpoint_invocation.py b/src/firefly_forge/agentic_checkpoint_invocation.py
index 6ee084a2..f0b98924 100644
--- a/src/firefly_forge/agentic_checkpoint_invocation.py
+++ b/src/firefly_forge/agentic_checkpoint_invocation.py
@@ -18,13 +18,15 @@
from firefly_forge.agent_checkpoints import AgentCheckpoints
from firefly_forge.agentic_checkpoint_boundary import AgenticBoundary, BoundaryCheckpoint, OriginalCompletion
from firefly_forge.bridge_completions import BridgeCompletions, _cut, _outside_unit
+from firefly_forge.checkpoint_source_identity import git_head_identity, source_identity
from firefly_forge.git_workspace import git_environment, metadata_digest, trusted_git
from firefly_forge.journal import digest
from firefly_forge.process import checked
from firefly_forge.prompts import prompt_catalog
-from firefly_forge.sandbox import pack_workspace
+from firefly_forge.source_snapshot import git_index_identity
FILE_TOOLS = frozenset({"list_files", "read_file", "write_file", "search_text", "edit_lines", "git_diff"})
+STATIC_TOOLS = FILE_TOOLS | {"execute_command", "knowledge_lookup", "read_run_evidence"}
_current: ContextVar[CheckpointInvocation | None] = ContextVar("agentic_checkpoint_invocation", default=None)
@@ -54,9 +56,26 @@ def runtime_contract() -> str:
name: hashlib.sha256((root / name).read_bytes()).hexdigest()
for name in (
"agentic_runtime.py",
+ "agent_restores.py",
+ "agent_checkpoints.py",
"agentic_checkpoint_boundary.py",
"agentic_checkpoint_invocation.py",
+ "checkpoint_source_identity.py",
+ "source_snapshot.py",
+ "source_diff_admission.py",
+ "git_workspace.py",
+ "workspace_tools.py",
"providers.py",
+ "sandbox.py",
+ "sandbox_command_slots.py",
+ "journal.py",
+ "runner_pool_worker.py",
+ "sandbox_models.py",
+ "sandbox_persistence.py",
+ "runner_backends.py",
+ "runner_factory_models.py",
+ "runner_pool_models.py",
+ "engine.py",
)
},
}
@@ -93,9 +112,22 @@ async def workspace_cut(store: Any, workspace: Path) -> dict[str, str]:
raise ValueError("The retained owned workspace is unavailable")
metadata = metadata_digest(workspace)
head = await checked(trusted_git(["git", "rev-parse", "HEAD"]), workspace, env=git_environment())
- _, source = pack_workspace(workspace, 16_777_216, registry=store.credential_registry())
- if metadata_digest(workspace) != metadata:
- raise ValueError("Workspace Git metadata changed during inspection")
+ index = git_index_identity(workspace)
+ head_identity = git_head_identity(workspace)
+ scanner = await store.source_patch_scanner()
+
+ async def finalize() -> None:
+ current_head = await checked(trusted_git(["git", "rev-parse", "HEAD"]), workspace, env=git_environment())
+ await scanner.revalidate()
+ if (
+ current_head != head
+ or metadata_digest(workspace) != metadata
+ or git_index_identity(workspace) != index
+ or git_head_identity(workspace) != head_identity
+ ):
+ raise ValueError("Workspace Git metadata changed during inspection")
+
+ source = await source_identity(workspace, scanner, finalize=finalize)
return {"path": str(workspace.resolve()), "head": head, "metadata": metadata, "source": source}
@@ -127,8 +159,18 @@ def attach(
self, boundary: AgenticBoundary, brief: str, backend: str, frontend: str, context: Any, settings: Any
) -> Any:
grants = context.allowed_tools
- if grants is None or not grants or not set(grants) <= FILE_TOOLS or context.qa_references:
- raise ValueError("Required checkpoints support only explicit static file-tool grants without QA state")
+ if grants is None or not grants or not set(grants) <= STATIC_TOOLS:
+ raise ValueError("Required checkpoints support only explicit static text-tool grants without QA state")
+ from firefly_forge.engine import Engine
+
+ try:
+ references_match = digest({"references": context.qa_references}) == digest(
+ {"references": Engine._qa_reference_descriptors(self.run)}
+ )
+ except (TypeError, ValueError):
+ references_match = False
+ if not references_match:
+ raise ValueError("Required checkpoints need the exact frozen textual QA reference descriptors")
self.boundary = boundary
if self.restored is not None:
binding = self.restored.binding
diff --git a/src/firefly_forge/agentic_runtime.py b/src/firefly_forge/agentic_runtime.py
index dd54c924..f301145f 100644
--- a/src/firefly_forge/agentic_runtime.py
+++ b/src/firefly_forge/agentic_runtime.py
@@ -138,18 +138,7 @@ async def invoke(*args: Any, **kwargs: Any) -> Any:
return invoke
async def source_manifest() -> dict[str, str]:
- manifest = {}
- total = 0
- for path in await tools.list_files():
- try:
- content = await tools.read_file(path)
- except (ValueError, OSError, UnicodeDecodeError):
- continue
- total += len(content.encode("utf-8"))
- if total > 16_000_000:
- raise ValueError("Command source manifest exceeds 16 MB")
- manifest[path] = hashlib.sha256(content.encode("utf-8")).hexdigest()
- return manifest
+ return await tools.source_manifest()
async def execute_command(argv: list[str], timeout: int = 300) -> dict[str, Any]:
@@ -163,7 +152,7 @@ async def execute_command(argv: list[str], timeout: int = 300) -> dict[str, Any]
before = await source_manifest()
result = await context.command_executor(argv, timeout)
after = await source_manifest()
- tools.files_written.update(path for path, digest in after.items() if before.get(path) != digest)
+ tools.files_written.update(path for path in before.keys() | after.keys() if before.get(path) != after.get(path))
return {"exit_code": result.exit_code, "output": result.output}
async def browser_check(
@@ -335,7 +324,7 @@ async def read_run_evidence(query: dict[str, Any]) -> dict[str, Any]:
return await context.read_run_evidence(query)
handlers: dict[str, Callable[..., Any]] = {
- "list_files": tools.list_files,
+ "list_files": tools.list_files_page,
"read_file": read_source,
"write_file": write_source,
"search_text": tools.search_text,
@@ -432,7 +421,7 @@ async def read_run_evidence(query: dict[str, Any]) -> dict[str, Any]:
agent: FireflyAgent[None, RuntimeReport] = FireflyAgent(
name="forge-builder",
model=selected_model,
- instructions=build_instruction(brief, backend, frontend)
+ instructions=build_instruction("", backend, frontend)
+ ("\n" + qa_instructions() if "browser_journey" in handlers else "")
+ (
prompt_text(
@@ -452,7 +441,13 @@ async def read_run_evidence(query: dict[str, Any]) -> dict[str, Any]:
else None
),
tools=[
- firefly_tool(name, description=prompt_text("tools/" + name), auto_register=False)(guarded(handler))
+ firefly_tool(
+ name,
+ description=prompt_text(
+ "tools/" + name, version="1.0.1" if name in {"list_files", "search_text"} else "1.0.0"
+ ),
+ auto_register=False,
+ )(guarded(handler))
for name, handler in handlers.items()
],
capabilities=[ProcessHistory(image_history), *([boundary] if boundary is not None else [])],
diff --git a/src/firefly_forge/api.py b/src/firefly_forge/api.py
index 5a410803..32f07900 100644
--- a/src/firefly_forge/api.py
+++ b/src/firefly_forge/api.py
@@ -212,6 +212,7 @@
from firefly_forge.scm_actor_authority import SCMActorAuthority
from firefly_forge.scm_actor_authority_controllers import SCMActorAuthorityController
from firefly_forge.scm_connections import SCMConnections
+from firefly_forge.scm_maintain_ticket_controllers import MaintainTicketController
from firefly_forge.scm_natural_controllers import SCMNaturalController
from firefly_forge.scm_poll_controllers import SCMPollController
from firefly_forge.scm_poll_functional import SCMPollFunctionalAuthority
@@ -1254,6 +1255,7 @@ class Application:
SCMPollResumeController,
SCMPollFunctionalController,
SCMNaturalController,
+ MaintainTicketController,
SCMSourceEditController,
SCMPollController,
SCMProductChangeController,
diff --git a/src/firefly_forge/checkpoint_source_identity.py b/src/firefly_forge/checkpoint_source_identity.py
new file mode 100644
index 00000000..42b032ad
--- /dev/null
+++ b/src/firefly_forge/checkpoint_source_identity.py
@@ -0,0 +1,246 @@
+# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0
+"""Stream checkpoint source identity independently of runner transport archives."""
+
+from __future__ import annotations
+
+import asyncio
+import hashlib
+import json
+import os
+import stat
+import tempfile
+import time
+from collections.abc import Awaitable, Callable
+from dataclasses import dataclass
+from pathlib import Path
+from typing import BinaryIO
+
+from firefly_forge.sandbox import _directory_fd, _parent_fd, _regular_data
+from firefly_forge.source_diff_admission import SourceDiffAdmission
+from firefly_forge.source_snapshot import stream_git_bytes
+from firefly_forge.workspace_tools import _descriptor_path, _private_name
+
+
+def _stamp(value: os.stat_result) -> list[int]:
+ return [
+ getattr(value, "st_" + key) for key in ("dev", "ino", "uid", "mode", "nlink", "size", "mtime_ns", "ctime_ns")
+ ]
+
+
+def _parts(name: str) -> tuple[str, ...]:
+ parts = tuple(name.split("/"))
+ if not name or any(part in {"", ".", ".."} for part in parts) or "\x00" in name:
+ raise ValueError("Checkpoint source requires exact relative Git paths")
+ if any(_private_component(part) for part in parts):
+ raise ValueError("Checkpoint source cannot include selected private paths")
+ return parts
+
+
+def _private_component(name: str) -> bool:
+ return _private_name(name) or name.casefold() in {".aws", ".azure", ".docker", ".forge-sandbox-results"}
+
+
+def _remember(records: BinaryIO, path: str, info: os.stat_result) -> None:
+ records.write(json.dumps([path, _stamp(info)], separators=(",", ":")).encode() + b"\n")
+
+
+def _parents(root: int, parts: tuple[str, ...], records: BinaryIO) -> int:
+ current = os.dup(root)
+ try:
+ _remember(records, "", os.fstat(current))
+ for ordinal, name in enumerate(parts[:-1], 1):
+ child = _directory_fd(name, parent=current)
+ os.close(current)
+ current = child
+ if os.fstat(current).st_uid != os.geteuid():
+ raise ValueError("Checkpoint source directories require current ownership")
+ _remember(records, "/".join(parts[:ordinal]), os.fstat(current))
+ return current
+ except BaseException:
+ os.close(current)
+ raise
+
+
+def _revalidate(root: int, records: BinaryIO, deadline: float) -> None:
+ records.seek(0)
+ for record in records:
+ if time.monotonic() >= deadline:
+ raise TimeoutError("Checkpoint source inspection exceeded its finite time budget")
+ name, expected = json.loads(record)
+ parts = tuple(name.split("/")) if name else ()
+ parent = _parent_fd(root, parts) if parts else os.dup(root)
+ try:
+ actual = os.stat(parts[-1], dir_fd=parent, follow_symlinks=False) if parts else os.fstat(parent)
+ if _stamp(actual) != expected:
+ raise ValueError("Checkpoint source changed during final admission")
+ except OSError:
+ raise ValueError("Checkpoint source changed during final admission") from None
+ finally:
+ os.close(parent)
+
+
+async def _directory_witnesses(directory: int, prefix: tuple[str, ...], records: BinaryIO) -> None:
+ _remember(records, "/".join(prefix), os.fstat(directory))
+ with os.scandir(directory) as entries:
+ for entry in entries:
+ if _private_component(entry.name):
+ continue
+ if entry.is_dir(follow_symlinks=False):
+ child = _directory_fd(entry.name, parent=directory)
+ try:
+ await _directory_witnesses(child, (*prefix, entry.name), records)
+ finally:
+ os.close(child)
+ await asyncio.sleep(0)
+
+
+def git_head_identity(workspace: Path) -> str:
+ directory = _directory_fd(workspace / ".git")
+ try:
+ head, _ = _regular_data(directory, "HEAD", 1024)
+ result = hashlib.sha256(head)
+ if head.startswith(b"ref: "):
+ name = head[5:].rstrip(b"\n").decode("ascii")
+ parts = tuple(name.split("/"))
+ if not name.startswith("refs/") or any(part in {"", ".", ".."} for part in parts):
+ raise ValueError("Checkpoint source HEAD reference is invalid")
+ parent = None
+ try:
+ parent = _parent_fd(directory, parts)
+ os.stat(parts[-1], dir_fd=parent, follow_symlinks=False)
+ except FileNotFoundError:
+ result.update(b"packed-ref")
+ else:
+ data, _ = _regular_data(parent, parts[-1], 1024)
+ result.update(data)
+ finally:
+ if parent is not None:
+ os.close(parent)
+ return result.hexdigest()
+ finally:
+ os.close(directory)
+
+
+@dataclass(frozen=True)
+class SourceEntry:
+ path: str
+ mode: str | None
+ object_id: str | None
+
+
+async def stream_source_entries(
+ workspace: Path, consume: Callable[[SourceEntry], Awaitable[None]], *, timeout_seconds: int = 300
+) -> None:
+ """Enumerate tracked source, including ignored paths, then nonignored new files."""
+ for tracked in (True, False):
+ tail = b""
+
+ async def receive(chunk: bytes, *, tracked: bool = tracked) -> None:
+ nonlocal tail
+ values = (tail + chunk).split(b"\0")
+ tail = values.pop()
+ if len(tail) > 65536:
+ raise ValueError("Checkpoint source path exceeds the Git name window")
+ for value in values:
+ mode = object_id = None
+ if tracked:
+ prefix, tab, value = value.partition(b"\t")
+ fields = prefix.split(b" ")
+ if not tab or len(fields) != 3 or fields[0] not in {b"100644", b"100755"} or fields[2] != b"0":
+ raise ValueError("Checkpoint source index requires regular files without links or conflicts")
+ mode, object_id = fields[0].decode(), fields[1].decode("ascii")
+ try:
+ name = value.decode("utf-8", errors="strict")
+ except UnicodeDecodeError:
+ raise ValueError("Checkpoint source requires exact UTF-8 Git paths") from None
+ _parts(name)
+ await consume(SourceEntry(name, mode, object_id))
+
+ args = ["git", "ls-files", "--stage"] if tracked else ["git", "ls-files", "--others", "--exclude-standard"]
+ await stream_git_bytes(workspace, [*args, "-z", "--"], receive, timeout_seconds=timeout_seconds)
+ if tail:
+ raise ValueError("Checkpoint source inventory ended without an exact path boundary")
+
+
+async def source_identity(
+ workspace: Path,
+ scanner: SourceDiffAdmission,
+ *,
+ finalize: Callable[[], Awaitable[None]],
+ timeout_seconds: int = 300,
+) -> str:
+ """Hash complete selected source using bounded chunks and temporary metadata only."""
+ root = _directory_fd(workspace)
+ deadline = time.monotonic() + timeout_seconds
+ try:
+ if os.fstat(root).st_uid != os.geteuid():
+ raise ValueError("Checkpoint source requires current workspace ownership")
+ canonical = _descriptor_path(root)
+ with tempfile.TemporaryFile(mode="w+b") as records:
+ async with asyncio.timeout(timeout_seconds):
+ first = await _capture(canonical, root, scanner, records, timeout_seconds)
+ records.seek(0)
+ records.truncate()
+ await _directory_witnesses(root, (), records)
+ second = await _capture(canonical, root, scanner, records, timeout_seconds)
+ if first != second:
+ raise ValueError("Checkpoint source selection or contents changed during capture")
+ await scanner(b"", True)
+ await finalize()
+ _revalidate(root, records, deadline)
+ if _stamp(os.stat(workspace, follow_symlinks=False)) != _stamp(os.fstat(root)):
+ raise ValueError("Checkpoint source root changed during capture")
+ return "git-source-v1:" + second
+ finally:
+ os.close(root)
+
+
+async def _capture(
+ workspace: Path, root: int, scanner: SourceDiffAdmission, records: BinaryIO, timeout_seconds: int
+) -> str:
+ result = hashlib.sha256(b"forge-checkpoint-git-source-v1\0")
+ _remember(records, "", os.fstat(root))
+
+ async def consume(entry: SourceEntry) -> None:
+ parts = _parts(entry.path)
+ raw = json.dumps([entry.path, entry.mode, entry.object_id], ensure_ascii=False, separators=(",", ":")).encode()
+ await scanner(entry.path.encode(), False)
+ scanner.finish_document()
+ result.update(len(raw).to_bytes(8, "big") + raw)
+ parent = descriptor = None
+ try:
+ parent = _parents(root, parts, records)
+ descriptor = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent)
+ before = os.fstat(descriptor)
+ if not stat.S_ISREG(before.st_mode) or before.st_nlink != 1 or before.st_uid != os.geteuid():
+ raise ValueError("Checkpoint source requires owned unaliased regular files without links")
+ if _descriptor_path(descriptor).relative_to(workspace) != Path(entry.path):
+ raise ValueError("Checkpoint source path differs from its opened descriptor")
+ content, size = hashlib.sha256(), 0
+ while chunk := os.read(descriptor, 65536):
+ size += len(chunk)
+ content.update(chunk)
+ await scanner(chunk, False)
+ scanner.finish_document()
+ after = os.fstat(descriptor)
+ named = os.stat(parts[-1], dir_fd=parent, follow_symlinks=False)
+ if _stamp(before) != _stamp(after) or _stamp(before) != _stamp(named) or size != before.st_size:
+ raise ValueError("Checkpoint source changed while reading a regular file")
+ if _descriptor_path(descriptor).relative_to(workspace) != Path(entry.path):
+ raise ValueError("Checkpoint source path changed during its read")
+ _remember(records, entry.path, after)
+ result.update(b"file\0" + str(stat.S_IMODE(after.st_mode)).encode() + b"\0")
+ result.update(size.to_bytes(8, "big") + content.digest())
+ except FileNotFoundError:
+ if entry.mode is None:
+ raise ValueError("Checkpoint source changed during inventory") from None
+ result.update(b"deleted\0")
+ except OSError:
+ raise ValueError("Checkpoint source requires regular files and directories without links") from None
+ finally:
+ for fd in (descriptor, parent):
+ if fd is not None:
+ os.close(fd)
+
+ await stream_source_entries(workspace, consume, timeout_seconds=timeout_seconds)
+ return result.hexdigest()
diff --git a/src/firefly_forge/coding_check_transport.py b/src/firefly_forge/coding_check_transport.py
index 59dc13bf..033e3a50 100644
--- a/src/firefly_forge/coding_check_transport.py
+++ b/src/firefly_forge/coding_check_transport.py
@@ -205,9 +205,21 @@ async def invoke(request: ConfiguredCheckRequest | SourceReadRequest) -> dict[st
if self.source_callback is not None:
- @server.tool(description=prompt_text("coding-runtime/source-list", version="1.0.1"))
- async def list_files(request_id: Annotated[str, Field(pattern=r"^[a-f0-9]{32}$")]) -> dict[str, Any]:
- return await invoke(SourceReadRequest(tool="list_files", request_id=request_id))
+ @server.tool(description=prompt_text("coding-runtime/source-list", version="1.0.2"))
+ async def list_files(
+ request_id: Annotated[str, Field(pattern=r"^[a-f0-9]{32}$")],
+ path: Annotated[str, Field(max_length=1000)] = "",
+ cursor: Annotated[str | None, Field(max_length=2048)] = None,
+ page_size: Annotated[int, Field(ge=1, le=1000)] = 100,
+ ) -> dict[str, Any]:
+ options: dict[str, Any] = {}
+ if path:
+ options["path"] = path
+ if cursor is not None:
+ options["cursor"] = cursor
+ if page_size != 100:
+ options["page_size"] = page_size
+ return await invoke(SourceReadRequest(tool="list_files", request_id=request_id, **options))
@server.tool(description=prompt_text("coding-runtime/source-read", version="1.0.1"))
async def read_file(
@@ -216,16 +228,23 @@ async def read_file(
) -> dict[str, Any]:
return await invoke(SourceReadRequest(tool="read_file", path=path, request_id=request_id))
- @server.tool(description=prompt_text("coding-runtime/source-search", version="1.0.1"))
+ @server.tool(description=prompt_text("coding-runtime/source-search", version="1.0.2"))
async def search_text(
text: Annotated[str, Field(min_length=1, max_length=500)],
request_id: Annotated[str, Field(pattern=r"^[a-f0-9]{32}$")],
path: Annotated[str, Field(max_length=1000)] = "",
max_results: Annotated[int, Field(ge=1, le=100)] = 50,
+ cursor: Annotated[str | None, Field(max_length=2048)] = None,
) -> dict[str, Any]:
+ options: dict[str, Any] = {"cursor": cursor} if cursor is not None else {}
return await invoke(
SourceReadRequest(
- tool="search_text", text=text, path=path, max_results=max_results, request_id=request_id
+ tool="search_text",
+ text=text,
+ path=path,
+ max_results=max_results,
+ request_id=request_id,
+ **options,
)
)
else:
diff --git a/src/firefly_forge/collaboration.py b/src/firefly_forge/collaboration.py
index ccd72193..4685472e 100644
--- a/src/firefly_forge/collaboration.py
+++ b/src/firefly_forge/collaboration.py
@@ -396,6 +396,7 @@ async def settings(self, repository_id: str) -> Data:
data.pop("_scm_source_edit_used_v1", None)
data.pop("_scm_product_change_authority_v1", None)
data.pop("_scm_product_intake_v1", None)
+ data.pop("_maintain_ticket_v1", None)
return {
**data,
"repository_id": repository_id,
@@ -451,6 +452,7 @@ async def configure(self, repository_id: str, request: CollaborationSettings) ->
"_scm_product_change_authority_v1",
"_scm_product_intake_v1",
"_github_poll_v1",
+ "_maintain_ticket_v1",
):
if row is not None and key in row.data:
data[key] = copy.deepcopy(row.data[key])
@@ -1647,6 +1649,7 @@ async def _remote_write(
issue: Data | None = None,
*,
before_write: Callable[[], Awaitable[None]] | None = None,
+ before_dispatch: Callable[[], Awaitable[None]] | None = None,
) -> Data:
provider = str(repo["connection_provider"])
ownership_required = self.store.execution_ownership_acquired or (
@@ -1673,11 +1676,13 @@ async def _remote_write(
):
await self.store.assert_ownership()
self.store.reject_credentials(payload)
+ dispatch = {"before_dispatch": before_dispatch} if before_dispatch is not None else {}
response = await self.scm._request(
"POST",
api + self._issue_path(provider, repo, issue),
headers={"Authorization": "Bearer " + token, "Accept": "application/json"},
json=payload,
+ **dispatch,
)
result: Data = response.json()
return {
@@ -1700,9 +1705,16 @@ async def _feedback_head(self, data: Data, *, proactive: bool) -> bool:
for peer in peers
):
return False
+ if data.get("maintain_ticket") is not None:
+ peers = [peer for peer in peers if peer.get("maintain_ticket") is not None]
return not proactive or not peers or peers[0]["id"] == data["id"]
async def _feedback_authority(self, data: Data, *, local: bool = False) -> None:
+ if data.get("maintain_ticket") is not None:
+ from firefly_forge.scm_maintain_tickets import MaintainTickets
+
+ await MaintainTickets(self).authorize_feedback(data, remote=not local)
+ return
from firefly_forge.project_access import CURRENT_PRINCIPAL, AccessDenied, ProjectAccess
from firefly_forge.scm_actor_authority import SCMActorAuthority
@@ -1781,6 +1793,11 @@ async def _feedback_authority(self, data: Data, *, local: bool = False) -> None:
raise AccessDenied("Current repository collaboration permission is required for feedback")
async def _feedback_context(self, data: Data, run: Data) -> Data:
+ if data.get("maintain_ticket") is not None:
+ from firefly_forge.scm_maintain_tickets import MaintainTickets
+
+ await MaintainTickets(self).authorize_feedback(data)
+ return {"_feedback_origin": self.browser_origin, "_feedback_actions": []}
from firefly_forge.project_access import ProjectAccess
from firefly_forge.scm_actor_authority import SCMActorAuthority
@@ -1821,6 +1838,9 @@ async def _feedback_context(self, data: Data, run: Data) -> Data:
return {"_feedback_origin": self.browser_origin, "_feedback_actions": actions}
async def feedback(self, receipt_id: str) -> None:
+ from firefly_forge.scm_connections import ProviderDispatchRefused
+ from firefly_forge.scm_maintain_tickets import MaintainNotificationNotDispatched, MaintainTickets
+
initial = await self.receipt(receipt_id)
if initial.get("native_question") is not None:
from firefly_forge.native_question_publications import NativeQuestionPublications
@@ -1918,7 +1938,7 @@ def native_card(current_run: Data) -> str:
if data.get("natural_turn") is not None:
from firefly_forge.scm_natural_feedback import natural_card
- return natural_card(data, public_text)
+ return natural_card(data, public_text, repository=repo, host=host)
message = feedback_card(data, current_run, repo, host, public_text)
if not conversation or (current_run and data["command"]["language"] != "es"):
return message
@@ -2135,12 +2155,52 @@ async def retained_before_write() -> None:
raise DeferredFeedbackChanged("The deferred feedback authority or route changed") from None
write_started = True
+ dispatch_check: Callable[[], Awaitable[None]] | None = None
+ if data.get("maintain_ticket") is not None:
+
+ async def maintain_dispatch_check() -> None:
+ from pyfly.data.transaction import current_unit_of_work
+
+ from firefly_forge.project_access import AccessDenied
+
+ if current_unit_of_work() is not None:
+ raise ValueError("Admit maintain notifications outside a writer transaction")
+ await self.store.assert_ownership()
+ tickets = MaintainTickets(self)
+ try:
+ context = await tickets.authorize_feedback(data)
+ if (
+ milestone_key(context["run"]) != intent["key"]
+ or content_digest(communication_facts(context["run"]))
+ != intent["communication"]["facts_sha256"]
+ ):
+ raise MaintainNotificationNotDispatched("The exact maintain notification facts changed")
+ self.store.reject_credentials(public_values)
+ self.communication_writer.assert_current(intent["communication"])
+ tickets.confirm_local_cut(context)
+ except (AccessDenied, KeyError, ValueError) as error:
+ raise MaintainNotificationNotDispatched(
+ "Current native maintain publication was refused before HTTP"
+ ) from error
+
+ dispatch_check = maintain_dispatch_check
+
remote = await self._remote_write(
repo,
payload,
data["issue"],
before_write=retained_before_write if retained is not None else before_write,
+ before_dispatch=dispatch_check,
)
+ if data.get("maintain_ticket") is not None:
+ from firefly_forge.scm_maintain_ticket_models import confirmed_remote
+
+ confirmed_remote(remote, self._ticket_url(repo, data["issue"]))
+ except (MaintainNotificationNotDispatched, ProviderDispatchRefused):
+ async with self.store._unit():
+ row = await self.inbox.find_by_id(identifier(receipt_id), lock=LockMode.PESSIMISTIC_WRITE)
+ assert row is not None
+ await self._write_row(row, feedback_status="held", reason="native_maintain_authority_changed")
except DeferredFeedbackChanged:
async with self.store._unit():
row = await self.inbox.find_by_id(identifier(receipt_id), lock=LockMode.PESSIMISTIC_WRITE)
diff --git a/src/firefly_forge/collaboration_feedback.py b/src/firefly_forge/collaboration_feedback.py
index 06229dde..3b22362b 100644
--- a/src/firefly_forge/collaboration_feedback.py
+++ b/src/firefly_forge/collaboration_feedback.py
@@ -34,7 +34,14 @@ def _text(value: object, safe: Callable[[str], str], limit: int = 1200, *, prese
if preserve_word_slashes:
path_pattern = r"(? str:
+ return re.sub(r"\b[A-Z][A-Z0-9_]*=\S+", _copy()["redactions"]["setting"], value)
+
+
+def _escaped_text(value: str, limit: int) -> str:
result = []
length = 0
for character in " ".join(value.split()):
diff --git a/src/firefly_forge/engine.py b/src/firefly_forge/engine.py
index 0c3ce6fd..53f6379d 100644
--- a/src/firefly_forge/engine.py
+++ b/src/firefly_forge/engine.py
@@ -858,9 +858,8 @@ async def read_run_evidence(query: dict[str, Any]) -> dict[str, Any]:
source_brief += customer_instruction(run["product_definition"])
if run.get("branding_snapshot"):
- branding = run["branding_snapshot"]
source_brief += prompt_text(
- "implementation/brand", files=json.dumps(branding["files"], ensure_ascii=False)
+ "implementation/brand", files=json.dumps(self._branding_manifest(run["branding_snapshot"]))
)
if run.get("design_snapshot"):
if run["design_snapshot"]["method"] == "process_map" or run["frontend"] == "none":
@@ -973,6 +972,7 @@ async def read_run_evidence(query: dict[str, Any]) -> dict[str, Any]:
await self.store.update(identifier, implementation_pending=True)
if restoration is None:
await self._materialize_design(identifier, workspace)
+ await self._materialize_branding(identifier, workspace, settings)
if contains_known_credential(brief, {**settings, **self.credential_registry()}):
raise ValueError(
"The composed worker context contains a configured credential; execution blocked"
@@ -1791,6 +1791,97 @@ async def _resolve_qa_reference(
)
return approved, evidence
+ @staticmethod
+ def _branding_manifest(branding: dict[str, Any]) -> dict[str, Any]:
+ files = branding.get("files")
+ if (
+ not isinstance(files, dict)
+ or not files
+ or any(
+ type(name) is not str
+ or not name
+ or name.startswith(".")
+ or any(character in name for character in ("/", "\\", "\x00"))
+ or type(content) is not str
+ for name, content in files.items()
+ )
+ ):
+ raise ValueError("Frozen product branding needs complete named text files")
+ expected = hashlib.sha256(json.dumps(files, sort_keys=True).encode()).hexdigest()
+ if branding.get("digest") != expected:
+ raise ValueError("Frozen product branding failed its content digest")
+ return {
+ "revision": branding["revision"],
+ "digest": expected,
+ "scope": "component",
+ "files": [
+ {
+ "path": "docs/brand/" + expected + "/" + name,
+ "bytes": len(files[name].encode("utf-8")),
+ "sha256": hashlib.sha256(files[name].encode("utf-8")).hexdigest(),
+ }
+ for name in sorted(files)
+ ],
+ }
+
+ async def _materialize_branding(self, identifier: str, workspace: Path, settings: dict[str, str]) -> None:
+ run = await self.store.get_run(identifier)
+ branding = run.get("branding_snapshot")
+ if not branding:
+ return
+ from firefly_forge.agentic_runtime import WorkspaceTools
+ from firefly_forge.project_access import ProjectAccess
+
+ manifest = self._branding_manifest(branding)
+ await self.store.admit_credentials(branding)
+ if contains_known_credential(branding, {**settings, **self.credential_registry()}):
+ raise ValueError("Frozen product branding contains a configured credential")
+
+ async def guard() -> dict[str, Any]:
+ await self.assert_ownership()
+ current = await self.store.get_run(identifier)
+ if current.get("cancel_request", {}).get("status") == "requested":
+ raise asyncio.CancelledError()
+ if current.get("branding_snapshot") != branding:
+ raise ValueError("Frozen product branding changed before materialization")
+ await ProjectAccess(self.store).authorize_execution(current)
+ self.store.reject_credentials(branding)
+ return current
+
+ await guard()
+ tools = WorkspaceTools(self.component_directory(run, workspace, create=True))
+ missing = []
+ for item in manifest["files"]:
+ content = branding["files"][item["path"].rsplit("/", 1)[1]]
+ try:
+ existing = await tools.read_file(item["path"])
+ except FileNotFoundError:
+ existing = None
+ except ValueError as error:
+ if not isinstance(error.__cause__, FileNotFoundError):
+ raise
+ existing = None
+ if existing is not None and existing != content:
+ raise ValueError("Materialized product branding conflicts with existing source")
+ if existing is None:
+ missing.append((item["path"], content))
+ for path, content in missing:
+ await guard()
+ # A late nonempty file must not be overwritten after the absence check.
+ await tools.write_file(path, content, expected_sha256=hashlib.sha256(b"").hexdigest())
+ if await tools.read_file(path) != content:
+ raise ValueError("Materialized product branding differs from its manifest")
+ prefix = run.get("component_snapshot", {}).get("path", ".") if run.get("component_snapshot") else "."
+ paths = [item["path"] if prefix == "." else prefix + "/" + item["path"] for item in manifest["files"]]
+ current = await guard()
+ if (
+ current.get("branding_materialized_digest") != manifest["digest"]
+ or current.get("branding_materialized_paths") != paths
+ ):
+ await self.store.update(
+ identifier, branding_materialized_digest=manifest["digest"], branding_materialized_paths=paths
+ )
+
async def _materialize_design(self, identifier: str, workspace: Path) -> None:
run = await self.store.get_run(identifier)
bundle = run.get("design_snapshot")
diff --git a/src/firefly_forge/journal.py b/src/firefly_forge/journal.py
index c45621eb..9099ea7f 100644
--- a/src/firefly_forge/journal.py
+++ b/src/firefly_forge/journal.py
@@ -27,7 +27,7 @@
from firefly_forge.profiles import build_instruction
from firefly_forge.prompts import prompt_text
from firefly_forge.sandbox_persistence import ForgeSandboxRepository
-from firefly_forge.store import TERMINAL, Store, identifier, now
+from firefly_forge.store import TERMINAL, Store, identifier, now, prepared_credential_rejection
if TYPE_CHECKING:
from firefly_forge.scm_source_edit_models import SealedSourceEditAdmission
@@ -364,14 +364,31 @@ async def entries(self, run_id: str) -> list[Data]:
if not protected:
return entries
- async def admitted() -> None:
- await self.store.admit_credentials(protected)
+ checkpoint_payloads = [entry["payload"] for entry in entries if entry["kind"] == "agent_checkpoint"]
+ if len(checkpoint_payloads) > 100:
+ raise ValueError("Journal checkpoint history exceeds its retained bound")
+
+ async def checked(value: Any) -> None:
async with self.store._unit():
current = await self.store.get_run(run_id)
if self._worker_scope(current) != scope:
raise AccessDenied("Journal run scope changed during credential admission")
await self._diagnostic_reader(current, reader)
- self.store.reject_credentials(protected)
+ self.store.reject_credentials(value)
+
+ async def admitted() -> None:
+ from firefly_forge.agentic_checkpoint_boundary import BoundaryCheckpoint
+
+ with prepared_credential_rejection(self.store):
+ await self.store.admit_credentials(protected)
+ await checked(protected)
+ for payload in checkpoint_payloads:
+ checkpoint = BoundaryCheckpoint.decode(
+ payload["snapshot"].encode("utf-8"), payload["checkpoint_digest"]
+ )
+ canonical = checkpoint.canonical_payload.decode("utf-8")
+ await self.store.admit_credentials(canonical)
+ await checked(canonical)
if caller is None:
await admitted()
@@ -2284,6 +2301,36 @@ async def ready_retries(self) -> list[str]:
async with self.store._unit():
return [row.id.hex for row in await self.store.run_records.find_ready_retries(time.time())]
+ async def sandbox_command_intent(
+ self,
+ run_id: str,
+ task_id: str,
+ attempt: int,
+ snapshot: Data,
+ payload: Callable[[int, str, Data | None], Data],
+ *,
+ ordinal: int | None = None,
+ ) -> Data:
+ from firefly_forge.sandbox_command_slots import command_intent
+
+ return await command_intent(self, run_id, task_id, attempt, snapshot, payload, ordinal=ordinal)
+
+ async def sandbox_command_replay(
+ self,
+ run_id: str,
+ task_id: str,
+ attempt: int,
+ snapshot: Data,
+ payload: Callable[[int, str, Data | None], Data],
+ *,
+ ordinal: int,
+ ) -> Data:
+ from firefly_forge.sandbox_command_slots import command_intent
+
+ return await command_intent(
+ self, run_id, task_id, attempt, snapshot, payload, ordinal=ordinal, replay_only=True
+ )
+
async def sandbox_effect_intent(
self, run_id: str, task_id: str, execution_id: str, *, effect: str, payload: Data
) -> Data:
diff --git a/src/firefly_forge/resources/prompts/v1/coding-runtime/source-list-1.0.2.yaml b/src/firefly_forge/resources/prompts/v1/coding-runtime/source-list-1.0.2.yaml
new file mode 100644
index 00000000..4dce432a
--- /dev/null
+++ b/src/firefly_forge/resources/prompts/v1/coding-runtime/source-list-1.0.2.yaml
@@ -0,0 +1,12 @@
+# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0
+{
+ "name": "coding-runtime/source-list",
+ "version": "1.0.2",
+ "description": "Versioned complete source traversal with bounded reader pages.",
+ "system_template": "",
+ "user_template": "List one version 1 page of confined component source paths. Optional path is a component-relative prefix of 0 to 1000 characters; page_size is an integer from 1 to 1000, default 100. Supply next_cursor unchanged on the next page using the same prefix and page size until complete=true. A page is not the entire project. A changed source view or cursor scope is refused; start a fresh listing. Response windows are bounded independently of project size. No credentials, Git metadata, commands or new authority. Supply a unique request_id of exactly 32 lowercase hexadecimal characters (0-9 and a-f); repeated IDs read exact source-bound receipts. Only declared fields are accepted.",
+ "required_variables": [],
+ "metadata": {
+ "defaults": {}
+ }
+}
diff --git a/src/firefly_forge/resources/prompts/v1/coding-runtime/source-search-1.0.2.yaml b/src/firefly_forge/resources/prompts/v1/coding-runtime/source-search-1.0.2.yaml
new file mode 100644
index 00000000..c0165fdf
--- /dev/null
+++ b/src/firefly_forge/resources/prompts/v1/coding-runtime/source-search-1.0.2.yaml
@@ -0,0 +1,12 @@
+# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0
+{
+ "name": "coding-runtime/source-search",
+ "version": "1.0.2",
+ "description": "Versioned complete source traversal with bounded reader pages.",
+ "system_template": "",
+ "user_template": "Search all selected source within a component-relative path prefix, returning one finite version 1 result page. text has 1 to 500 characters on one line, path has 0 to 1000 characters, max_results is an integer from 1 to 100, default 50. Supply next_cursor unchanged with the same query, prefix and max_results to continue until complete=true. Each snippet has an explicit snippet_truncated flag; full-file SHA256 binds its source. Binary and invalid-UTF8 files are classified explicitly; unreadable or changed source is refused. No commands or new authority. Supply a unique request_id of exactly 32 lowercase hexadecimal characters (0-9 and a-f); repeated IDs only read exact source-bound receipts. Only declared fields are accepted.",
+ "required_variables": [],
+ "metadata": {
+ "defaults": {}
+ }
+}
diff --git a/src/firefly_forge/resources/prompts/v1/manifest.json b/src/firefly_forge/resources/prompts/v1/manifest.json
index fc52278f..0f4779a8 100644
--- a/src/firefly_forge/resources/prompts/v1/manifest.json
+++ b/src/firefly_forge/resources/prompts/v1/manifest.json
@@ -77,6 +77,14 @@
"field": "user",
"variables": {}
},
+ {
+ "id": "coding-runtime/source-list",
+ "version": "1.0.2",
+ "path": "coding-runtime/source-list-1.0.2.yaml",
+ "sha256": "85376d131679dd2220b1556fac1caba1d45fead764c6f87319b185f33a99e107",
+ "field": "user",
+ "variables": {}
+ },
{
"id": "coding-runtime/source-read",
"version": "1.0.0",
@@ -109,6 +117,14 @@
"field": "user",
"variables": {}
},
+ {
+ "id": "coding-runtime/source-search",
+ "version": "1.0.2",
+ "path": "coding-runtime/source-search-1.0.2.yaml",
+ "sha256": "83bbd8df12bfd4414803601e0c97005c97216d97a8f9468f7518b738a2dc9544",
+ "field": "user",
+ "variables": {}
+ },
{
"id": "collaboration/action",
"version": "1.0.0",
@@ -2616,6 +2632,14 @@
"field": "user",
"variables": {}
},
+ {
+ "id": "tools/list_files",
+ "version": "1.0.1",
+ "path": "tools/list_files-1.0.1.yaml",
+ "sha256": "d795c6b4b4126571dade971a40708ba5099a0c570a0e4bacb393b8f5aa2be843",
+ "field": "user",
+ "variables": {}
+ },
{
"id": "tools/profile_php",
"version": "1.0.0",
@@ -2664,6 +2688,14 @@
"field": "user",
"variables": {}
},
+ {
+ "id": "tools/search_text",
+ "version": "1.0.1",
+ "path": "tools/search_text-1.0.1.yaml",
+ "sha256": "e4b5b619960bd8737791b9775620d6856e830678857bbdcd258d1c6b670089c0",
+ "field": "user",
+ "variables": {}
+ },
{
"id": "tools/trace_python",
"version": "1.0.0",
diff --git a/src/firefly_forge/resources/prompts/v1/tools/list_files-1.0.1.yaml b/src/firefly_forge/resources/prompts/v1/tools/list_files-1.0.1.yaml
new file mode 100644
index 00000000..c33d7566
--- /dev/null
+++ b/src/firefly_forge/resources/prompts/v1/tools/list_files-1.0.1.yaml
@@ -0,0 +1,12 @@
+# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0
+{
+ "name": "tools/list_files",
+ "version": "1.0.1",
+ "description": "Versioned complete source traversal with bounded reader pages.",
+ "system_template": "",
+ "user_template": "List a bounded version 1 page of project source paths. Use path_prefix to select a component-relative subtree and page_size from 1 to 1000 (default 100). The response includes files, view_sha256, next_cursor and complete. Continue next_cursor with the same options until complete=true; never call a partial page the whole project. A changed source view or cursor scope is refused. Source is selected by the native Git inventory, including tracked ignored files; credentials, unsafe links and aliases remain unavailable. No new tool or execution authority.",
+ "required_variables": [],
+ "metadata": {
+ "defaults": {}
+ }
+}
diff --git a/src/firefly_forge/resources/prompts/v1/tools/search_text-1.0.1.yaml b/src/firefly_forge/resources/prompts/v1/tools/search_text-1.0.1.yaml
new file mode 100644
index 00000000..86890f0f
--- /dev/null
+++ b/src/firefly_forge/resources/prompts/v1/tools/search_text-1.0.1.yaml
@@ -0,0 +1,12 @@
+# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0
+{
+ "name": "tools/search_text",
+ "version": "1.0.1",
+ "description": "Versioned complete source traversal with bounded reader pages.",
+ "system_template": "",
+ "user_template": "Search the complete selected source within path_prefix, including large valid UTF-8 files. max_results bounds one page from 1 to 100 (default 50), not total project matches. Continue next_cursor using the same query/options until complete=true. Snippets disclose snippet_truncated and full-file SHA256; binary and invalid-UTF8 classifications are explicit. Unreadable, changed or unsafe source refuses. No commands or new authority.",
+ "required_variables": [],
+ "metadata": {
+ "defaults": {}
+ }
+}
diff --git a/src/firefly_forge/runner_pool_worker.py b/src/firefly_forge/runner_pool_worker.py
index 17421947..3fea53fc 100644
--- a/src/firefly_forge/runner_pool_worker.py
+++ b/src/firefly_forge/runner_pool_worker.py
@@ -9,7 +9,7 @@
from contextlib import suppress
from pathlib import Path
from typing import Any
-from uuid import NAMESPACE_URL, UUID, uuid5
+from uuid import UUID
import httpx
from pyfly.container import service
@@ -224,8 +224,11 @@ async def execute(
timeout: int,
task_id: str,
*,
- ordinal: int,
+ ordinal: int | None = None,
) -> ProcessResult:
+ from firefly_forge.sandbox_command_slots import assert_command_boundary
+
+ assert_command_boundary()
bounded_argv(argv)
if type(timeout) is not int or timeout < 1:
raise ValueError("Command timeout must be a positive integer")
@@ -236,10 +239,9 @@ async def execute(
run, attempt = scope["run"], scope["attempt"]
snapshot = run.get("sandbox_snapshot")
policy = validate_factory_pool(snapshot or {})
- if type(ordinal) is not int or not 1 <= ordinal <= policy.command_limit:
+ if ordinal is not None and (type(ordinal) is not int or not 1 <= ordinal <= policy.command_limit):
raise ValueError("Frozen command budget is exhausted")
relative = self._workspace(run, workspace, cwd)
- execution_id = uuid5(NAMESPACE_URL, f"forge:sandbox:{run_id}:{task_id}:{attempt}:{ordinal}").hex
command = {
"argv": argv,
"cwd": relative,
@@ -247,18 +249,25 @@ async def execute(
"workspace": str(workspace),
"snapshot_digest": snapshot["digest"],
}
- records = await self.journal.sandbox_executions(run_id)
- previous = next((row for row in records if row["execution_id"] == execution_id), None)
- if previous is not None:
- if previous["request"]["command"] != command:
+
+ def retained_command(number: int, identity: str, previous: dict[str, Any] | None) -> dict[str, Any]:
+ assert previous is not None
+ retained = previous["request"]
+ if retained["command"] != command:
raise ValueError("Frozen Factory invocation command changed")
- evidence = previous["effects"].get("start", {}).get("evidence", {})
- if previous["status"] == "removed" and evidence.get("workspace_import") == "complete":
- await self._current(run_id, task_id, scope)
- return self._result(argv, evidence["exit_code"], evidence["output"])
- raise RuntimeError("Existing Factory invocation requires reconciliation; it cannot be resubmitted")
- if len(records) >= policy.command_limit:
- raise ValueError("Frozen total command budget is exhausted")
+ return dict(retained)
+
+ if ordinal is not None:
+ replay = await self.journal.sandbox_command_replay(
+ run_id, task_id, attempt, snapshot, retained_command, ordinal=ordinal
+ )
+ if replay["disposition"] != "absent":
+ previous = replay["execution"]
+ evidence = previous["effects"].get("start", {}).get("evidence", {})
+ if previous["status"] == "removed" and evidence.get("workspace_import") == "complete":
+ await self._current(run_id, task_id, scope)
+ return self._result(argv, evidence["exit_code"], evidence["output"])
+ raise RuntimeError("Existing Factory invocation requires reconciliation; it cannot be resubmitted")
await self.probe(snapshot)
await self._current(run_id, task_id, scope)
source, _ = pack_workspace(
@@ -266,43 +275,65 @@ async def execute(
)
metadata = await self.git_snapshot.build(source, self.artifacts.directory, policy.workspace_max_bytes)
archive, bundle = frame_project_archives(source, metadata, policy.workspace_max_bytes)
- invocation = RunnerProjectInvocation(
- execution_id=execution_id,
- profile_id=policy.profile.profile_id,
- profile_revision=policy.profile.revision,
- project_id=run.get("project_id"),
- repository_id=run["repository_id"],
- run_id=UUID(run_id),
- image=policy.image,
- argv=tuple(argv),
- cwd=relative,
- source_digest=bundle.source_digest,
- archive_digest=bundle.archive_digest,
- source_bytes=len(archive),
- bundle=bundle,
- cpus=policy.cpus,
- memory_mb=policy.memory_mb,
- pids_limit=policy.pids_limit,
- timeout=min(timeout, policy.timeout),
- max_output_bytes=policy.max_output_bytes,
- )
- self.artifacts.retain(policy.profile, invocation, "source", archive)
- payload = {
- "parent_attempt": attempt,
- "factory_fence": scope["fence"].model_dump(mode="json"),
- "ordinal": ordinal,
- "container_name": "pool-" + execution_id,
- "snapshot_digest": snapshot["digest"],
- "command": command,
- "request_digest": pool_digest(invocation.model_dump(mode="json")),
- "request": invocation.model_dump(mode="json"),
- }
+
+ def prepare(number: int, identity: str, previous: dict[str, Any] | None) -> dict[str, Any]:
+ if previous is not None:
+ retained = previous["request"]
+ if retained["command"] != command:
+ raise ValueError("Frozen Factory invocation command changed")
+ return dict(retained)
+ invocation = RunnerProjectInvocation(
+ execution_id=identity,
+ profile_id=policy.profile.profile_id,
+ profile_revision=policy.profile.revision,
+ project_id=run.get("project_id"),
+ repository_id=run["repository_id"],
+ run_id=UUID(run_id),
+ image=policy.image,
+ argv=tuple(argv),
+ cwd=relative,
+ source_digest=bundle.source_digest,
+ archive_digest=bundle.archive_digest,
+ source_bytes=len(archive),
+ bundle=bundle,
+ cpus=policy.cpus,
+ memory_mb=policy.memory_mb,
+ pids_limit=policy.pids_limit,
+ timeout=min(timeout, policy.timeout),
+ max_output_bytes=policy.max_output_bytes,
+ )
+ payload = {
+ "parent_attempt": attempt,
+ "factory_fence": scope["fence"].model_dump(mode="json"),
+ "ordinal": number,
+ "container_name": "pool-" + identity,
+ "snapshot_digest": snapshot["digest"],
+ "command": command,
+ "request_digest": pool_digest(invocation.model_dump(mode="json")),
+ "request": invocation.model_dump(mode="json"),
+ }
+ return payload
+
await self._current(run_id, task_id, scope)
- created = await self.journal.sandbox_effect_intent(
- run_id, task_id, execution_id, effect="create", payload=payload
+ current_source, _ = pack_workspace(
+ workspace, policy.workspace_max_bytes, registry=self.artifacts.credential_registry()
+ )
+ if hashlib.sha256(current_source).hexdigest() != bundle.source_archive_digest:
+ raise ValueError("Command source changed during pool preparation")
+ created = await self.journal.sandbox_command_intent(
+ run_id, task_id, attempt, snapshot, prepare, ordinal=ordinal
)
- if created["disposition"] != "execute":
- raise RuntimeError("Factory preparation already exists; reconciliation required")
+ execution_id = created["execution"]["execution_id"]
+ payload = created["execution"]["request"]
+ invocation = RunnerProjectInvocation.model_validate(payload["request"])
+ if not created["fresh"]:
+ previous = created["execution"]
+ evidence = previous["effects"].get("start", {}).get("evidence", {})
+ if previous["status"] == "removed" and evidence.get("workspace_import") == "complete":
+ await self._current(run_id, task_id, scope)
+ return self._result(argv, evidence["exit_code"], evidence["output"])
+ raise RuntimeError("Existing Factory invocation requires reconciliation; it cannot be resubmitted")
+ self.artifacts.retain(policy.profile, invocation, "source", archive)
await self.journal.sandbox_effect_receipt(
run_id,
task_id,
diff --git a/src/firefly_forge/sandbox.py b/src/firefly_forge/sandbox.py
index 00582ee9..6ad7ca4c 100644
--- a/src/firefly_forge/sandbox.py
+++ b/src/firefly_forge/sandbox.py
@@ -471,7 +471,6 @@ def __init__(self, store: Store, journal: Journal, backends: RunnerBackends | No
self.transfer_image = pinned_image(transfer_image) if transfer_image else None
self._capacity = asyncio.Semaphore(2)
self._run_locks: dict[str, asyncio.Lock] = defaultdict(asyncio.Lock)
- self._ordinals: dict[tuple[str, str, int], int] = defaultdict(int)
def _safe_result(self, result: ProcessResult, *, browser: bool = False) -> ProcessResult:
if browser and contains_known_credential(result.output, self.credential_registry()):
@@ -835,6 +834,9 @@ async def execute(
_receipt_sink: Callable[[dict[str, Any]], None] | None = None,
_invocation_guard: Callable[[], Awaitable[None]] | None = None,
) -> ProcessResult:
+ from firefly_forge.sandbox_command_slots import assert_command_boundary
+
+ assert_command_boundary()
if task_id not in {"implement", "verify"}:
raise ValueError("Runner commands must belong to an implementation or verification stage")
if not argv or len(argv) > 64 or type(timeout) is not int or timeout < 1:
@@ -880,11 +882,6 @@ async def admit() -> dict[str, Any]:
raise RuntimeError("Factory pool worker is not configured; execution cannot fall back")
if _browser or _browser_input is not None:
raise ValueError("Pool browser execution requires a separately qualified browser profile")
- stage = next(item for item in (await self.journal.plan(run_id))["tasks"] if item["id"] == task_id)
- if ordinal is None:
- slot = (run_id, task_id, stage["attempt"])
- self._ordinals[slot] += 1
- ordinal = self._ordinals[slot]
return await self.pool_worker.execute(
run_id, workspace, directory, argv, timeout, task_id, ordinal=ordinal
)
@@ -897,33 +894,15 @@ async def admit() -> dict[str, Any]:
output_limit = 8_300_000 if _browser else config.max_output_bytes
planned = next(item for item in (await self.journal.plan(run_id))["tasks"] if item["id"] == task_id)
attempt = planned["attempt"]
- slot = (run_id, task_id, attempt)
- if ordinal is None:
- self._ordinals[slot] += 1
- ordinal = self._ordinals[slot]
- if type(ordinal) is not int or not 1 <= ordinal <= config.command_limit:
- raise ValueError("Frozen command budget is exhausted")
- execution_id = uuid.uuid5(uuid.NAMESPACE_URL, f"forge:sandbox:{run_id}:{task_id}:{attempt}:{ordinal}").hex
- execution_identity = execution_id
- records = await self.journal.sandbox_executions(run_id)
- existing = next((record for record in records if record["execution_id"] == execution_id), None)
- if run.get("source_edit_continuation"):
- from firefly_forge.source_edit_continuations import assert_source_final_budget
-
- await assert_source_final_budget(self.store, self.journal, run, task_id, execution_id)
- if existing is None and len(records) >= config.command_limit:
- raise ValueError("Frozen total command budget is exhausted; resume and handoff do not reset it")
- if _browser and existing is None:
- assert config.browser is not None
- browser_calls = sum(bool(row["request"]["request"].get("browser", False)) for row in records)
- if browser_calls >= config.browser.max_calls:
- raise ValueError("Frozen browser call budget is exhausted; resume and handoff do not reset it")
- labels = {
- "org.firefly.forge.owner": self.owner,
- "org.firefly.forge.execution": execution_id,
- "org.firefly.forge.run": run_id,
- "org.firefly.forge.snapshot": snapshot["digest"],
- }
+ explicit_id = (
+ uuid.uuid5(uuid.NAMESPACE_URL, f"forge:sandbox:{run_id}:{task_id}:{attempt}:{ordinal}").hex
+ if ordinal is not None
+ else None
+ )
+ retained = next(
+ (row for row in await self.journal.sandbox_executions(run_id) if row["execution_id"] == explicit_id),
+ None,
+ )
request: dict[str, Any] = {
"browser": _browser,
"output_limit": output_limit,
@@ -940,21 +919,47 @@ async def admit() -> dict[str, Any]:
raise ValueError("Archive execution requires an administrator-configured pinned transfer image")
request["transfer_image"] = self.transfer_image
request["workspace_digest"] = (
- existing["request"]["request"]["workspace_digest"]
- if existing
+ retained["request"]["request"]["workspace_digest"]
+ if retained
else pack_workspace(workspace, config.workspace_max_bytes, registry=self.credential_registry())[1]
)
- payload = {
- "parent_attempt": attempt,
- "ordinal": ordinal,
- "container_name": "forge-sandbox-" + execution_id,
- "snapshot_digest": snapshot["digest"],
- "request_digest": digest(request),
- "request": request,
+
+ def prepare(number: int, identity: str, previous: dict[str, Any] | None) -> dict[str, Any]:
+ frozen_request = dict(request)
+ if previous and SandboxInput.model_validate(previous["request"]["request"]["configuration"]) == config:
+ frozen_request["configuration"] = previous["request"]["request"]["configuration"]
+ value = {
+ "parent_attempt": attempt,
+ "ordinal": number,
+ "container_name": "forge-sandbox-" + identity,
+ "snapshot_digest": snapshot["digest"],
+ "request_digest": digest(frozen_request),
+ "request": frozen_request,
+ }
+ if previous and "executor_contract" in previous["request"]:
+ value["executor_contract"] = previous["request"]["executor_contract"]
+ return value
+
+ await admit()
+ if config.workspace_transport == "archive" and retained is None:
+ current_digest = pack_workspace(
+ workspace, config.workspace_max_bytes, registry=self.credential_registry()
+ )[1]
+ if current_digest != request["workspace_digest"]:
+ raise ValueError("Command source changed during archive preparation")
+ created = await self.journal.sandbox_command_intent(
+ run_id, task_id, attempt, snapshot, prepare, ordinal=ordinal
+ )
+ payload = created["execution"]["request"]
+ execution_id = created["execution"]["execution_id"]
+ execution_identity = execution_id
+ labels = {
+ "org.firefly.forge.owner": self.owner,
+ "org.firefly.forge.execution": execution_id,
+ "org.firefly.forge.run": run_id,
+ "org.firefly.forge.snapshot": snapshot["digest"],
}
- if existing and SandboxInput.model_validate(existing["request"]["request"]["configuration"]) == config:
- request["configuration"] = existing["request"]["request"]["configuration"]
- payload["request_digest"] = digest(request)
+ existing = None if created["fresh"] else created["execution"]
if existing is not None:
if existing["request"] != payload:
raise SandboxReconciliationRequired("Durable runner invocation differs from the requested command")
@@ -983,9 +988,6 @@ async def admit() -> dict[str, Any]:
"Interrupted runner invocation requires explicit stage recovery"
)
await admit()
- created = await self.journal.sandbox_effect_intent(
- run_id, task_id, execution_id, effect="create", payload=payload
- )
if config.mode == "host":
if created["disposition"] == "needs_reconciliation":
raise SandboxReconciliationRequired("Unknown host command outcome cannot be replayed")
diff --git a/src/firefly_forge/sandbox_command_slots.py b/src/firefly_forge/sandbox_command_slots.py
new file mode 100644
index 00000000..aadd5230
--- /dev/null
+++ b/src/firefly_forge/sandbox_command_slots.py
@@ -0,0 +1,294 @@
+# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0
+"""Allocate a new command identity and its create WAL atomically under the run writer."""
+
+from __future__ import annotations
+
+import copy
+import hashlib
+import json
+from collections.abc import Callable
+from typing import TYPE_CHECKING, Any
+from uuid import NAMESPACE_URL, uuid5
+
+from firefly_forge.journal import digest
+from firefly_forge.project_access import ProjectAccess
+from firefly_forge.runner_factory_models import validate_factory_pool
+from firefly_forge.runner_pool_models import pool_digest
+from firefly_forge.sandbox_models import capture_sandbox, validate_sandbox_snapshot
+
+if TYPE_CHECKING:
+ from firefly_forge.journal import Journal
+
+Data = dict[str, Any]
+COMMAND_CONTRACT = "forge-command-ordinal/v2"
+
+
+def assert_command_boundary() -> None:
+ from pyfly.data.transaction import current_unit_of_work
+
+ if current_unit_of_work() is not None:
+ raise ValueError("Command allocation and dispatch require no ambient transaction")
+
+
+def _refuse() -> None:
+ from firefly_forge.sandbox import SandboxReconciliationRequired
+
+ raise SandboxReconciliationRequired("Durable command identity or effect requires reconciliation")
+
+
+def _canonical(row: Any, entries: list[Data], run_id: str) -> Data:
+ identity = uuid5(NAMESPACE_URL, f"forge:sandbox:{run_id}:{row.task_id}:{row.parent_attempt}:{row.ordinal}").hex
+ if (
+ row.run_id.hex != run_id
+ or row.task_id not in {"implement", "verify"}
+ or type(row.parent_attempt) is not int
+ or not 1 <= row.parent_attempt <= 100
+ or type(row.ordinal) is not int
+ or not 1 <= row.ordinal <= 100
+ or row.id.hex != identity
+ or row.container_name not in {"forge-sandbox-" + identity, "pool-" + identity}
+ or set(row.data) != {"request", "effects"}
+ ):
+ _refuse()
+ request, effects = row.data["request"], row.data["effects"]
+ if (
+ not isinstance(request, dict)
+ or not isinstance(effects, dict)
+ or "create" not in effects
+ or set(effects) - {"create", "start", "cleanup"}
+ or type(request.get("parent_attempt")) is not int
+ or type(request.get("ordinal")) is not int
+ or request.get("executor_contract", COMMAND_CONTRACT) != COMMAND_CONTRACT
+ or request.get("parent_attempt") != row.parent_attempt
+ or request.get("ordinal") != row.ordinal
+ or request.get("container_name") != row.container_name
+ or request.get("snapshot_digest") != row.snapshot_digest
+ or not isinstance(request.get("request"), dict)
+ or request.get("request_digest")
+ != (pool_digest(request["request"]) if row.container_name.startswith("pool-") else digest(request["request"]))
+ ):
+ _refuse()
+ for effect, state in effects.items():
+ intents = [
+ entry
+ for entry in entries
+ if entry["kind"] == "sandbox_intent"
+ and entry["payload"].get("execution_id") == identity
+ and entry["payload"].get("effect") == effect
+ ]
+ outcomes = [
+ entry
+ for entry in entries
+ if entry["kind"] in {"sandbox_receipt", "sandbox_reconcile"}
+ and entry["payload"].get("execution_id") == identity
+ and entry["payload"].get("effect") == effect
+ ]
+ if (
+ not isinstance(state, dict)
+ or state.get("status") not in {"intent", "done"}
+ or len(intents) != 1
+ or intents[0]["task_id"] != row.task_id
+ or intents[0]["payload"].get("request") != request
+ ):
+ _refuse()
+ if state["status"] == "intent":
+ if outcomes:
+ _refuse()
+ continue
+ evidence = copy.deepcopy(state.get("evidence"))
+ if not isinstance(evidence, dict) or len(outcomes) != 1 or outcomes[0]["sequence"] <= intents[0]["sequence"]:
+ _refuse()
+ if isinstance(evidence.get("output"), str):
+ output = evidence.pop("output")
+ evidence.update(
+ output_sha256=hashlib.sha256(output.encode()).hexdigest(),
+ output_bytes=len(output.encode()),
+ output_excerpt=output[:2000],
+ )
+ if outcomes[0]["payload"].get("evidence") != evidence or outcomes[0]["task_id"] != row.task_id:
+ _refuse()
+ related = [
+ entry
+ for entry in entries
+ if entry["kind"] in {"sandbox_intent", "sandbox_receipt", "sandbox_reconcile"}
+ and entry["payload"].get("execution_id") == identity
+ ]
+ if {entry["payload"].get("effect") for entry in related if entry["kind"] == "sandbox_intent"} != set(effects):
+ _refuse()
+ last = related[-1]
+ expected_status = (
+ {"create": "creating", "start": "running", "cleanup": "cleaning"}
+ if last["kind"] == "sandbox_intent"
+ else {"create": "created", "start": "finished", "cleanup": "removed"}
+ )[last["payload"]["effect"]]
+ if last["payload"].get("evidence", {}).get("outcome") == "interrupted":
+ expected_status = "interrupted"
+ if row.status != expected_status:
+ _refuse()
+ if row.container_name.startswith("pool-"):
+ from firefly_forge.runner_pool_models import RunnerProjectInvocation
+
+ invocation = RunnerProjectInvocation.model_validate(request["request"])
+ if invocation.execution_id != identity or invocation.run_id.hex != run_id:
+ _refuse()
+ return {"execution_id": identity, "request": copy.deepcopy(request), "effects": copy.deepcopy(effects)}
+
+
+def _complete(row: Any, value: Data) -> bool:
+ effects = value["effects"]
+ start = effects.get("start", {})
+ evidence = start.get("evidence", {})
+ return (
+ row.status == "removed"
+ and set(effects) == {"create", "start", "cleanup"}
+ and all(
+ state.get("status") == "done" and state.get("evidence", {}).get("outcome") != "interrupted"
+ for state in effects.values()
+ )
+ and type(evidence.get("exit_code")) is int
+ and isinstance(evidence.get("output"), str)
+ and evidence.get("workspace_import") != "requires_reconciliation"
+ and (not row.container_name.startswith("pool-") or evidence.get("workspace_import") == "complete")
+ )
+
+
+def _recovered(row: Any, value: Data, entries: list[Data], tasks: list[Any]) -> bool:
+ effects = value["effects"]
+ if (
+ not {"create", "cleanup"} <= set(effects)
+ or any(state.get("status") != "done" for state in effects.values())
+ or any(
+ state.get("evidence", {}).get("workspace_import") == "requires_reconciliation" for state in effects.values()
+ )
+ or (
+ "start" in effects
+ and not any(state.get("evidence", {}).get("outcome") == "interrupted" for state in effects.values())
+ )
+ ):
+ return False
+ for effect, state in effects.items():
+ if state.get("evidence", {}).get("outcome") == "interrupted" and not any(
+ entry["kind"] == "sandbox_reconcile"
+ and entry["payload"].get("execution_id") == row.id.hex
+ and entry["payload"].get("effect") == effect
+ for entry in entries
+ ):
+ return False
+ task = next((item for item in tasks if item.task_id == row.task_id), None)
+ if task is None or task.attempt <= row.parent_attempt:
+ return False
+ key = uuid5(NAMESPACE_URL, f"forge:{row.run_id.hex}:{row.task_id}:{row.parent_attempt}").hex
+ last_effect = max(
+ entry["sequence"]
+ for entry in entries
+ if entry["kind"] in {"sandbox_intent", "sandbox_receipt", "sandbox_reconcile"}
+ and entry["payload"].get("execution_id") == row.id.hex
+ )
+ return any(
+ entry["kind"] == "reconciliation"
+ and entry["task_id"] == row.task_id
+ and entry["idempotency_key"] == key
+ and entry["payload"].get("outcome") == "retry"
+ and entry["payload"].get("evidence")
+ and entry["sequence"] > last_effect
+ for entry in entries
+ )
+
+
+async def command_intent(
+ journal: Journal,
+ run_id: str,
+ task_id: str,
+ attempt: int,
+ snapshot: Data,
+ payload: Callable[[int, str, Data | None], Data],
+ *,
+ ordinal: int | None = None,
+ replay_only: bool = False,
+) -> Data:
+ assert_command_boundary()
+ if task_id not in {"implement", "verify"} or type(attempt) is not int or not 1 <= attempt <= 100:
+ raise ValueError("Command allocation requires an exact native task attempt")
+ policy = validate_factory_pool(snapshot) if snapshot.get("mode") == "pool" else validate_sandbox_snapshot(snapshot)
+ if ordinal is not None and (type(ordinal) is not int or not 1 <= ordinal <= policy.command_limit):
+ raise ValueError("Frozen command budget is exhausted")
+ await journal.store.assert_ownership()
+ async with journal.store._unit():
+ run = await journal._run(run_id)
+ task = await journal._task(run_id, task_id)
+ current = {**copy.deepcopy(run.data), "id": run.id.hex}
+ if (run.data.get("sandbox_snapshot") or capture_sandbox(None)) != snapshot or task.attempt != attempt:
+ raise ValueError("Frozen command configuration or task attempt changed")
+ if task.status != "running":
+ raise ValueError("Command allocation requires the active native stage")
+ if run.data.get("cancel_request", {}).get("status") == "requested":
+ import asyncio
+
+ raise asyncio.CancelledError()
+ await ProjectAccess(journal.store).authorize_execution(current)
+ entries = [
+ {**copy.deepcopy(row.data), "digest": row.digest} for row in await journal.records.find_for_run(run.id)
+ ]
+ if len(entries) > 10000 or len(json.dumps(entries)) > 16 * 1024 * 1024 or not journal.verify_chain(entries):
+ _refuse()
+ rows = await journal.sandboxes.for_run(run.id)
+ if len(rows) > 100:
+ _refuse()
+ retained_ids = {
+ entry["payload"].get("execution_id")
+ for entry in entries
+ if entry["kind"] in {"sandbox_intent", "sandbox_receipt", "sandbox_reconcile"}
+ }
+ if retained_ids != {row.id.hex for row in rows}:
+ _refuse()
+ values = {row.id.hex: _canonical(row, entries, run_id) for row in rows}
+ if ordinal is None:
+ ordinal = 1 + max(
+ (row.ordinal for row in rows if row.task_id == task_id and row.parent_attempt == attempt), default=0
+ )
+ if ordinal > policy.command_limit:
+ raise ValueError("Frozen command budget is exhausted")
+ execution_id = uuid5(NAMESPACE_URL, f"forge:sandbox:{run_id}:{task_id}:{attempt}:{ordinal}").hex
+ existing = values.get(execution_id)
+ if existing is None:
+ tasks = await journal.tasks.find_for_run(run.id)
+ if any(
+ not _complete(row, values[row.id.hex]) and not _recovered(row, values[row.id.hex], entries, tasks)
+ for row in rows
+ ):
+ _refuse()
+ if len(rows) >= policy.command_limit:
+ raise ValueError("Frozen total command budget is exhausted; attempts do not reset it")
+ if replay_only and existing is None:
+ return {"disposition": "absent"}
+ request = payload(ordinal, execution_id, existing)
+ if existing is None and request.get("request", {}).get("browser"):
+ browser = getattr(policy, "browser", None)
+ if (
+ browser is None
+ or sum(bool(value["request"]["request"].get("browser")) for value in values.values())
+ >= browser.max_calls
+ ):
+ raise ValueError("Frozen browser call budget is exhausted")
+ if existing is None:
+ request = {**request, "executor_contract": COMMAND_CONTRACT}
+ inner = request.get("request")
+ prefix = "pool-" if snapshot.get("mode") == "pool" else "forge-sandbox-"
+ if (
+ type(request.get("parent_attempt")) is not int
+ or type(request.get("ordinal")) is not int
+ or request.get("parent_attempt") != attempt
+ or request.get("ordinal") != ordinal
+ or request.get("snapshot_digest") != snapshot["digest"]
+ or request.get("container_name") != prefix + execution_id
+ or not isinstance(inner, dict)
+ or request.get("request_digest") != (pool_digest(inner) if prefix == "pool-" else digest(inner))
+ ):
+ raise ValueError("Command payload differs from its canonical allocated slot")
+ if current.get("source_edit_continuation"):
+ from firefly_forge.source_edit_continuations import assert_source_final_budget
+
+ await assert_source_final_budget(journal.store, journal, current, task_id, execution_id)
+ await journal.store.assert_ownership()
+ result = await journal.sandbox_effect_intent(run_id, task_id, execution_id, effect="create", payload=request)
+ return {**result, "fresh": existing is None}
diff --git a/src/firefly_forge/scm_connections.py b/src/firefly_forge/scm_connections.py
index 7651a55c..63032b2f 100644
--- a/src/firefly_forge/scm_connections.py
+++ b/src/firefly_forge/scm_connections.py
@@ -60,6 +60,10 @@
}
+class ProviderDispatchRefused(ValueError):
+ """An opt-in local admission failed before provider HTTP dispatch."""
+
+
@service
class SCMConnections:
def __init__(self, store: Store, directory: Path | None = None, *, client: httpx.AsyncClient | None = None) -> None:
@@ -320,6 +324,7 @@ async def configure(self, provider: str, body: Data) -> Data:
async def _request(self, method: str, url: str, **kwargs: Any) -> httpx.Response:
allowed_statuses = kwargs.pop("allowed_statuses", (200, 201))
response_limit = kwargs.pop("response_limit", None)
+ before_dispatch = kwargs.pop("before_dispatch", None)
provider = next(
(
name
@@ -358,7 +363,12 @@ async def perform(client: httpx.AsyncClient) -> httpx.Response:
async with guard:
if self._record(provider) != record:
- raise ValueError("Application settings changed before provider dispatch; repeat with the current grant")
+ refusal = ProviderDispatchRefused if before_dispatch is not None else ValueError
+ raise refusal("Application settings changed before provider dispatch; repeat with the current grant")
+ if before_dispatch is not None:
+ await before_dispatch()
+ if self._record(provider) != record:
+ raise ProviderDispatchRefused("Application settings changed during provider admission")
try:
async with asyncio.timeout(20):
if self._client is not None:
diff --git a/src/firefly_forge/scm_conversations.py b/src/firefly_forge/scm_conversations.py
index 82d72350..e6a5775e 100644
--- a/src/firefly_forge/scm_conversations.py
+++ b/src/firefly_forge/scm_conversations.py
@@ -118,7 +118,11 @@ async def references(self, delivery: Data, scopes: list[Data], *, durable: bool
)
if scope is None or run.get("repository_id") != delivery["repository_id"]:
continue
- if previous.get("native_question") is not None:
+ 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)
diff --git a/src/firefly_forge/scm_maintain_ticket_controllers.py b/src/firefly_forge/scm_maintain_ticket_controllers.py
new file mode 100644
index 00000000..e2b5bf35
--- /dev/null
+++ b/src/firefly_forge/scm_maintain_ticket_controllers.py
@@ -0,0 +1,30 @@
+# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0
+"""Finite native setup for one maintain child's automatic GitHub communication."""
+
+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.collaboration import Collaboration
+from firefly_forge.scm_maintain_ticket_models import MaintainTicketInput
+from firefly_forge.scm_maintain_tickets import MaintainTickets
+
+
+@rest_controller
+@request_mapping("/api/v1/projects/{project_id}/repositories/{repository_id}/maintain-ticket-bindings")
+class MaintainTicketController:
+ def __init__(self, collaboration: Collaboration) -> None:
+ self.tickets = MaintainTickets(collaboration)
+
+ @post_mapping("", status_code=201)
+ async def configure(
+ self, project_id: PathVar[str], repository_id: PathVar[str], body: Body[MaintainTicketInput]
+ ) -> dict[str, Any]:
+ return await self.tickets.configure(project_id, repository_id, body)
+
+ @get_mapping("/requests/{request_id}")
+ async def read(
+ self, project_id: PathVar[str], repository_id: PathVar[str], request_id: PathVar[str]
+ ) -> dict[str, Any]:
+ return await self.tickets.read(project_id, repository_id, request_id)
diff --git a/src/firefly_forge/scm_maintain_ticket_models.py b/src/firefly_forge/scm_maintain_ticket_models.py
new file mode 100644
index 00000000..f117bcce
--- /dev/null
+++ b/src/firefly_forge/scm_maintain_ticket_models.py
@@ -0,0 +1,108 @@
+# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0
+"""Explicit finite communication consent binds one native maintain creation."""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Literal
+
+from pydantic import Field, field_validator, model_validator
+
+from firefly_forge.collaboration_feedback import MAX_FEEDBACK
+from firefly_forge.journal import digest
+from firefly_forge.scm_authority_models import Digest, NativeID, StrictContract
+
+
+class MaintainTicketInput(StrictContract):
+ version: Literal[1]
+ request_id: NativeID
+ run_id: NativeID
+ creation_request_id: NativeID
+ expected_creation_digest: Digest
+ observed_receipt_id: NativeID
+ expected_reference_digest: Digest
+ expected_previous_digest: Digest | None
+ lifetime_seconds: int = Field(ge=60, le=3600)
+ max_notifications: int = Field(ge=1, le=MAX_FEEDBACK)
+ allowed_actions: list[Literal["status", "question", "answer"]] = Field(min_length=1, max_length=3)
+ acknowledge_native_communication: Literal[True]
+
+ @field_validator("acknowledge_native_communication", mode="before")
+ @classmethod
+ def explicit_boolean(cls, value: object) -> object:
+ if type(value) is not bool:
+ raise ValueError("Native communication consent must be an explicit boolean")
+ return value
+
+ @model_validator(mode="after")
+ def exact_actions(self) -> MaintainTicketInput:
+ if len(set(self.allowed_actions)) != len(self.allowed_actions) or set(self.allowed_actions) != {
+ "status",
+ "question",
+ "answer",
+ }:
+ raise ValueError("This native binding requires explicit status, question and answer consent")
+ return self
+
+
+class MaintainTicketMarker(StrictContract):
+ version: Literal[1]
+ kind: Literal["native_maintain_ticket_v1"]
+ project_id: NativeID
+ request_id: NativeID
+ run_id: NativeID
+ binding_digest: Digest
+
+
+class MaintainTicketReceipt(StrictContract):
+ version: Literal[1]
+ event_id: NativeID
+ project_id: NativeID
+ repository_id: NativeID
+ run_id: NativeID
+ binding_key: NativeID
+ revision: int = Field(ge=1, le=100)
+ previous_digest: Digest | None
+ request: MaintainTicketInput
+ request_digest: Digest
+ admin_snapshot: dict[str, Any]
+ creation_digest: Digest
+ parent_reference: dict[str, Any]
+ parent_observation: dict[str, Any]
+ observation: dict[str, Any]
+ polling_scope: dict[str, Any]
+ owner: dict[str, Any]
+ outbox_id: NativeID
+ prepared_at_ms: int = Field(ge=1)
+ expires_at_ms: int = Field(ge=1)
+ policy_digest: Digest
+ digest: Digest
+
+ @model_validator(mode="after")
+ def exact_binding(self) -> MaintainTicketReceipt:
+ value = self.model_dump(mode="json")
+ if (
+ self.request_digest != request_digest(self.request)
+ or self.run_id != self.request.run_id
+ or self.binding_key != self.request.run_id
+ or self.creation_digest != self.request.expected_creation_digest
+ or self.previous_digest != self.request.expected_previous_digest
+ or self.expires_at_ms != self.prepared_at_ms + self.request.lifetime_seconds * 1000
+ or digest({k: v for k, v in value.items() if k != "digest"}) != self.digest
+ ):
+ raise ValueError("The immutable maintain ticket receipt changed")
+ return self
+
+
+def request_digest(request: MaintainTicketInput) -> str:
+ return digest(request.model_dump(mode="json"))
+
+
+def confirmed_remote(remote: dict[str, Any], ticket_url: str) -> dict[str, str]:
+ value = remote.get("id")
+ if type(value) not in {int, str} or re.fullmatch(r"[1-9][0-9]{0,39}", str(value)) is None:
+ raise ValueError("The maintain notification has no exact remote comment identity")
+ url = ticket_url + "#issuecomment-" + str(value)
+ if remote.get("url") != url:
+ raise ValueError("The maintain notification belongs to another issue or comment")
+ return {"id": str(value), "url": url}
diff --git a/src/firefly_forge/scm_maintain_tickets.py b/src/firefly_forge/scm_maintain_tickets.py
new file mode 100644
index 00000000..4005f05a
--- /dev/null
+++ b/src/firefly_forge/scm_maintain_tickets.py
@@ -0,0 +1,376 @@
+# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0
+"""Native communication bindings for independently admitted checkpoint maintenance."""
+
+from __future__ import annotations
+
+import copy
+import json
+from time import time
+from typing import Any
+from uuid import uuid4
+
+from pyfly.data.transaction import current_unit_of_work
+
+from firefly_forge.checkpoint_test_models import frozen_checkpoint_source
+from firefly_forge.collaboration_persistence import ForgeWebhookInboxEntity
+from firefly_forge.journal import digest
+from firefly_forge.project_access import CURRENT_PRINCIPAL, AccessDenied, ProjectAccess
+from firefly_forge.run_creation_requests import RunCreationRequests
+from firefly_forge.scm_conversations import SCMConversations, observation_cut
+from firefly_forge.scm_maintain_ticket_models import (
+ MaintainTicketInput,
+ MaintainTicketMarker,
+ MaintainTicketReceipt,
+ request_digest,
+)
+from firefly_forge.scm_poll_models import PollPointer
+from firefly_forge.scm_polling import MAX_BYTES, inbox_size
+from firefly_forge.store import identifier, now
+
+Data = dict[str, Any]
+STATE_KEY = "_maintain_ticket_v1"
+KIND = "native_maintain_ticket_v1"
+
+
+def clock_ms() -> int:
+ return int(time() * 1000)
+
+
+class MaintainNotificationNotDispatched(AccessDenied):
+ """Current local authority refused this notification before HTTP dispatch."""
+
+
+def state_of(data: Data) -> Data:
+ value = copy.deepcopy(data.get(STATE_KEY, {"version": 1, "requests": {}, "heads": {}}))
+ if (
+ set(value) != {"version", "requests", "heads"}
+ or type(value["version"]) is not int
+ or value["version"] != 1
+ or type(value["requests"]) is not dict
+ or type(value["heads"]) is not dict
+ or len(value["requests"]) > 100
+ or len(value["heads"]) > 100
+ or len(json.dumps(value).encode()) > MAX_BYTES
+ ):
+ raise ValueError("The native maintain communication state changed")
+ for pointer in [*value["requests"].values(), *value["heads"].values()]:
+ PollPointer.model_validate(pointer)
+ return dict(value)
+
+
+def creation_scope(run: Data, receipt: Data) -> None:
+ if (
+ receipt.get("run_id") != run["id"]
+ or receipt.get("source_checkpoint") != run.get("source_checkpoint_snapshot")
+ or run.get("operation") != "maintain"
+ or run.get("publish") is not False
+ or receipt.get("input", {}).get("operation") != "maintain"
+ or receipt.get("input", {}).get("publish", False) is not False
+ or any(receipt.get(k) != run.get(k) for k in ("repository_id", "project_id", "component_id"))
+ ):
+ raise AccessDenied("The ticket requires one exact native maintain creation")
+ frozen_checkpoint_source(run, {"id": run["repository_id"]})
+
+
+class MaintainTickets:
+ def __init__(self, collaboration: Any) -> None:
+ self.c, self.store, self.polling = collaboration, collaboration.store, collaboration.polling
+
+ async def _creation(self, request_id: str) -> Data:
+ async with self.store._unit():
+ receipt = await RunCreationRequests(self.store)._load(request_id)
+ if receipt is None:
+ raise AccessDenied("The exact native creation receipt is unavailable")
+ return receipt
+
+ async def _load(self, pointer: Data, project: str, repository: str) -> Data:
+ value = await self.polling._event(pointer, KIND, project, repository)
+ return MaintainTicketReceipt.model_validate(value).model_dump(mode="json")
+
+ async def _target(self, source: Data, project: str, repository: str, *, remote: bool) -> tuple[Data, Data, Data]:
+ scope = await self.polling.conversation_scope(source, remote=remote)
+ if scope["project_id"] != project or scope["repository_id"] != repository:
+ raise AccessDenied("The current observed owner belongs to another project or repository")
+ head = await self.polling._head(project, repository, scope["binding_key"])
+ if not {"status", "respond"}.issubset(head["request"]["allowed_actions"]):
+ raise AccessDenied("Current status and owner-answer consent are required")
+ async with self.store._unit():
+ user = await self.polling.users.find_by_id(identifier(scope["native_user_id"]))
+ if user is None or not user.enabled:
+ raise AccessDenied("The mapped maintain owner is unavailable")
+ identity = digest({"id": user.id.hex, "issuer": user.issuer, "subject": user.subject})
+ await ProjectAccess(self.store).authorize_actor(
+ {"user_id": scope["native_user_id"], "authentication_method": "oidc_bearer"}, project, "execute"
+ )
+ return scope, head, {"user_id": scope["native_user_id"], "identity_digest": identity}
+
+ async def configure(self, project: str, repository: str, request: MaintainTicketInput) -> Data:
+ if current_unit_of_work() is not None:
+ raise ValueError("Configure maintain communication outside a caller transaction")
+ actor = CURRENT_PRINCIPAL.get()
+ await self.polling.authority._reader(actor, project, repository, admin=True)
+ assert actor is not None
+ incoming = request.model_dump(mode="json")
+ await self.store.admit_credentials(incoming)
+ async with self.store._unit():
+ row, _ = await self.polling._state(repository)
+ state = state_of(row.data)
+ request_key = project + ":" + request.request_id
+ if pointer := state["requests"].get(request_key):
+ kept = await self._load(pointer, project, repository)
+ if kept["request"] != incoming or kept["admin_snapshot"]["user_id"] != actor.id.hex:
+ raise AccessDenied("The native communication request identity belongs to another input or actor")
+ return kept
+ run = await self.store.get_run(request.run_id)
+ receipt = await self._creation(request.creation_request_id)
+ creation_scope(run, receipt)
+ if receipt["digest"] != request.expected_creation_digest or (run["project_id"], run["repository_id"]) != (
+ project,
+ repository,
+ ):
+ raise AccessDenied("The reviewed maintain creation or scope changed")
+ source = await self.polling._observation(project, repository, request.observed_receipt_id)
+ if source.get("source_conflict") or digest(source["verification"]) != request.expected_reference_digest:
+ raise AccessDenied("The real GitHub owner observation changed")
+ scope, head, owner = await self._target(source, project, repository, remote=True)
+ references, _ = await SCMConversations(self.c).references(source, [scope], durable=True)
+ parent_id = run["source_checkpoint_snapshot"]["reference"]["run_id"]
+ 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_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"]]):
+ await self.polling.authority._reader(actor, project, repository, admin=True)
+ current_scope, current_head, current_owner = await self._target(source, project, repository, remote=False)
+ if (current_scope, current_head, current_owner) != (scope, head, owner):
+ raise AccessDenied("The maintain communication owner or current consent changed")
+ 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:
+ raise AccessDenied("The retained maintain creation or ticket parent changed")
+ row, _ = await self.polling._state(repository)
+ state = state_of(row.data)
+ if pointer := state["requests"].get(request_key):
+ kept = await self._load(pointer, project, repository)
+ if kept["request"] != incoming or kept["admin_snapshot"]["user_id"] != actor.id.hex:
+ raise AccessDenied("The concurrent maintain request identity changed")
+ return kept
+ old_pointer = state["heads"].get(request.run_id)
+ old = await self._load(old_pointer, project, repository) if old_pointer else None
+ if (old["digest"] if old else None) != request.expected_previous_digest:
+ raise ValueError("Maintain communication conflict; retain this native request identity")
+ if old is not None:
+ prior = await self.c.inbox.find_by_id(identifier(old["outbox_id"]))
+ if prior is not None and prior.data.get("feedback_status") in {"writing", "needs_reconciliation"}:
+ raise AccessDenied("The prior notification outcome requires reconciliation")
+ if len(state["requests"]) >= 100:
+ raise ValueError("Native maintain communication retention is full")
+ prepared = clock_ms()
+ outbox_id = digest({"kind": KIND, "project": project, "request": incoming})[:32]
+ value = dict(
+ version=1,
+ event_id=uuid4().hex,
+ project_id=project,
+ repository_id=repository,
+ run_id=request.run_id,
+ binding_key=request.run_id,
+ revision=old["revision"] + 1 if old else 1,
+ previous_digest=request.expected_previous_digest,
+ request=incoming,
+ request_digest=request_digest(request),
+ admin_snapshot=ProjectAccess.actor_snapshot(actor),
+ creation_digest=receipt["digest"],
+ parent_reference=parents[0],
+ parent_observation=parent_cut,
+ observation=observation_cut(source),
+ polling_scope=scope,
+ owner=owner,
+ outbox_id=outbox_id,
+ prepared_at_ms=prepared,
+ expires_at_ms=prepared + request.lifetime_seconds * 1000,
+ )
+ from firefly_forge.scm_natural_tasks import natural_policy_digest
+
+ value["policy_digest"] = natural_policy_digest()
+ value["digest"] = digest(value)
+ value = MaintainTicketReceipt.model_validate(value).model_dump(mode="json")
+ pointer = await self.polling._save_event(KIND, value)
+ state["requests"][request_key] = pointer
+ state["heads"][request.run_id] = pointer
+ state_of({STATE_KEY: state})
+ marker = {
+ "version": 1,
+ "kind": KIND,
+ "project_id": project,
+ "request_id": request.request_id,
+ "run_id": request.run_id,
+ "binding_digest": value["digest"],
+ }
+ data = {
+ key: copy.deepcopy(source[key])
+ for key in (
+ "repository_id",
+ "provider",
+ "remote_id",
+ "actor",
+ "issue",
+ "verification",
+ "source_snapshot",
+ )
+ }
+ data.update(
+ id=outbox_id,
+ delivery_id="maintain-ticket:" + outbox_id,
+ digest=digest(marker),
+ status="processed",
+ action_status="completed",
+ reason=None,
+ run_id=request.run_id,
+ feedback_status="pending",
+ maintain_ticket=marker,
+ global_ingress=False,
+ command={"action": "status", "run_id": request.run_id, "language": "en"},
+ created_at=now(),
+ updated_at=now(),
+ )
+ count, retained = await inbox_size(self.store, repository)
+ if count >= 5000 or retained + len(json.dumps(data).encode()) > MAX_BYTES:
+ raise ValueError("Maintain notification retention is full")
+ self.store.reject_credentials(data)
+ await self.c.inbox.save(
+ ForgeWebhookInboxEntity(
+ id=identifier(outbox_id),
+ repository_id=identifier(repository),
+ provider="github",
+ delivery_id=data["delivery_id"],
+ source_key=data["delivery_id"],
+ status="processed",
+ data=data,
+ )
+ )
+ row.data = {**row.data, STATE_KEY: state}
+ await self.polling.settings.save(row)
+ self.confirm_local_cut({"binding": value})
+ return copy.deepcopy(value)
+
+ async def read(self, project: str, repository: str, request_id: str) -> Data:
+ await self.polling.authority._reader(CURRENT_PRINCIPAL.get(), project, repository, admin=False)
+ async with self.store._unit():
+ row, _ = await self.polling._state(repository)
+ pointer = state_of(row.data)["requests"].get(project + ":" + request_id)
+ if pointer is None:
+ raise KeyError("The native maintain ticket request is unavailable")
+ return await self._load(pointer, project, repository)
+
+ async def current(self, data: Data, *, remote: bool = False, dispatch: bool = False) -> Data:
+ from firefly_forge.scm_natural_tasks import natural_policy_digest
+
+ marker = MaintainTicketMarker.model_validate(data.get("maintain_ticket")).model_dump(mode="json")
+ project, repository = marker["project_id"], data["repository_id"]
+ async with self.store._unit():
+ row, _ = await self.polling._state(repository)
+ state = state_of(row.data)
+ pointer = state["heads"].get(marker["run_id"])
+ if pointer is None or pointer.get("digest") != marker["binding_digest"]:
+ raise AccessDenied("The maintain ticket binding is no longer current")
+ value = await self._load(pointer, project, repository)
+ request = MaintainTicketInput.model_validate(value["request"])
+ async with self.store._unit():
+ administrator = await self.polling.users.find_by_id(identifier(value["admin_snapshot"]["user_id"]))
+ if administrator is None or not administrator.enabled:
+ raise AccessDenied("The maintain communication administrator is unavailable")
+ if not await ProjectAccess(self.store).authorize_actor(value["admin_snapshot"], project, "members"):
+ raise AccessDenied("Current factory administrator authority is required for maintain communication")
+ if (
+ value["request"]["request_id"] != marker["request_id"]
+ or await self.polling._latest(project, repository, KIND, request.run_id) != value["event_id"]
+ or value["policy_digest"] != natural_policy_digest()
+ or data["id"] != value["outbox_id"]
+ or data["digest"] != digest(marker)
+ or data["run_id"] != request.run_id
+ ):
+ raise AccessDenied("The native maintain notification lost its exact current receipt")
+ source = await self.polling._observation(project, repository, request.observed_receipt_id)
+ if observation_cut(source) != value["observation"] or source.get("source_conflict"):
+ raise AccessDenied("The native maintain observation changed")
+ for key in ("repository_id", "provider", "remote_id", "actor", "issue", "verification", "source_snapshot"):
+ if data.get(key) != source.get(key):
+ raise AccessDenied("The maintain notification destination changed")
+ if data.get("command") != {"action": "status", "run_id": request.run_id, "language": "en"}:
+ raise AccessDenied("A maintain notification cannot become an inbound command")
+ creation = await self._creation(request.creation_request_id)
+ run = await self.store.get_run(request.run_id)
+ creation_scope(run, creation)
+ if worker_request := run.get("worker_request"):
+ from firefly_forge.agent_roles import worker_feedback_model
+ from firefly_forge.native_question_publications import NativeQuestionPublications
+
+ entries = await NativeQuestionPublications(self.c)._entries(run["id"])
+ questions = [
+ entry
+ for entry in entries
+ if entry["kind"] == "worker_feedback"
+ and (entry["payload"].get("request") or {}).get("id") == worker_request.get("id")
+ ]
+ if len(questions) != 1:
+ raise AccessDenied("The maintain question requires one exact immutable worker feedback")
+ payload = questions[0]["payload"]
+ feedback = worker_feedback_model(payload).model_validate(payload)
+ if feedback.request is None or worker_request != {
+ **feedback.request.model_dump(),
+ "status": worker_request.get("status"),
+ "role_id": feedback.role_id,
+ "task_id": feedback.task_id,
+ "attempt": feedback.attempt,
+ }:
+ 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"])
+ 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"]
+ or run["component_id"] not in scope["component_ids"]
+ ):
+ raise AccessDenied("The current maintain owner or native scope changed")
+ if dispatch:
+ async with self.store._unit():
+ row = await self.c.inbox.find_by_id(identifier(data["id"]))
+ if row is None or row.data.get("feedback_status") not in {"pending", "writing", "sent"}:
+ raise AccessDenied("An unknown maintain notification cannot be retried")
+ if len(row.data.get("feedback_history") or []) >= request.max_notifications:
+ raise AccessDenied("The explicit maintain notification allowance is exhausted")
+ context = {"binding": value, "run": run, "source": source, "scope": scope}
+ self.confirm_local_cut(context)
+ return context
+
+ @staticmethod
+ def confirm_local_cut(context: Data) -> None:
+ from firefly_forge.scm_natural_tasks import natural_policy_digest
+
+ value = context["binding"]
+ if value["policy_digest"] != natural_policy_digest() or clock_ms() >= value["expires_at_ms"]:
+ raise MaintainNotificationNotDispatched("The maintain communication policy or finite consent expired")
+
+ async def authorize_feedback(self, data: Data, *, remote: bool = False) -> Data:
+ try:
+ return await self.current(data, remote=remote, dispatch=True)
+ except (AccessDenied, KeyError, ValueError) as error:
+ raise MaintainNotificationNotDispatched(
+ "Current native maintain communication was refused before HTTP"
+ ) from error
+
+ async def authorize_reference(self, data: Data, run: Data, delivery: Data, scopes: list[Data]) -> None:
+ context = await self.current(data)
+ if (
+ context["run"]["id"] != run["id"]
+ or delivery["actor"] != data["actor"]
+ or not any(scope["native_user_id"] == context["binding"]["owner"]["user_id"] for scope in scopes)
+ ):
+ raise AccessDenied("Another project reader retains only checkpoint status access")
diff --git a/src/firefly_forge/scm_natural_effects.py b/src/firefly_forge/scm_natural_effects.py
index 2b36addb..b49df10e 100644
--- a/src/firefly_forge/scm_natural_effects.py
+++ b/src/firefly_forge/scm_natural_effects.py
@@ -309,9 +309,11 @@ async def review(self, receipt_id: str, *, remote: bool = True) -> Data:
resume = self.collaboration.poll_resume_authority
grant = (
await resume._grant(context["scope"]["project_id"], parent["repository_id"], key=scopes[0]["binding_key"])
- if resume
+ if resume and current.get("answer_only") is not True
else None
)
+ if current.get("answer_only") is True and resolution["state"] != "answer":
+ raise AccessDenied("Native maintain communication authorizes answers without continuation")
native_cut = None
if grant and grant["request"]["active"] and current["question_status"] == "answered":
binding = await resume._binding(
diff --git a/src/firefly_forge/scm_natural_feedback.py b/src/firefly_forge/scm_natural_feedback.py
index 140c556f..bf789f08 100644
--- a/src/firefly_forge/scm_natural_feedback.py
+++ b/src/firefly_forge/scm_natural_feedback.py
@@ -8,11 +8,14 @@
from typing import Any
from firefly_forge.collaboration_feedback import _text
+from firefly_forge.scm_natural_links import delivery_links
from firefly_forge.scm_natural_render import reply_quote
from firefly_forge.scm_natural_runtime import catalog
-def natural_card(data: dict[str, Any], safe: Callable[[str], str]) -> str:
+def natural_card(
+ data: dict[str, Any], safe: Callable[[str], str], *, repository: dict[str, Any] | None = None, host: str = ""
+) -> str:
turn = data["natural_turn"]
language = (turn.get("decision") or {}).get("language") or data["command"].get("language", "en")
copy = json.loads(catalog().text("scm-natural/copy", version="1.0.1"))[language]
@@ -22,7 +25,9 @@ def natural_card(data: dict[str, Any], safe: Callable[[str], str]) -> str:
outcome = copy.get(state, copy["waiting"])
message = str(turn.get("reply") or "")
questions = (turn.get("decision") or {}).get("questions", [])
- quote = reply_quote(message, questions, safe) if message else ""
+ quote = (
+ reply_quote(message, questions, safe, trusted_links=delivery_links(data, repository, host)) if message else ""
+ )
resolution = turn.get("resolution") or {}
if resolution.get("state") == "waiting" and len(resolution.get("candidates", [])) > 1:
options = [
diff --git a/src/firefly_forge/scm_natural_links.py b/src/firefly_forge/scm_natural_links.py
new file mode 100644
index 00000000..9237b4f7
--- /dev/null
+++ b/src/firefly_forge/scm_natural_links.py
@@ -0,0 +1,117 @@
+# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0
+"""Internal delivery hrefs from already verified native status context, never model authority."""
+
+from __future__ import annotations
+
+import re
+from typing import Any, Literal
+from urllib.parse import urlsplit, urlunsplit
+
+from pydantic import Field, model_validator
+
+from firefly_forge.collaboration_feedback import _link
+from firefly_forge.scm_authority_models import Digest, NativeID, StrictContract
+from firefly_forge.scm_natural_models import SCOPE_FIELDS
+
+Data = dict[str, Any]
+
+
+class NativeDeliveryLinkV1(StrictContract):
+ version: Literal[1] = 1
+ kind: Literal["pull_request", "commit"]
+ href: str = Field(min_length=1, max_length=600)
+ run_id: NativeID
+ repository_id: NativeID
+ project_id: NativeID
+ checkpoint_revision: str = Field(pattern=r"^(?:[a-f0-9]{40}|[a-f0-9]{64})$")
+ checkpoint_digest: Digest
+
+ @model_validator(mode="after")
+ def canonical_delivery_path(self) -> NativeDeliveryLinkV1:
+ parsed = urlsplit(self.href)
+ suffix = r"pull/[1-9][0-9]{0,19}" if self.kind == "pull_request" else "commit/" + self.checkpoint_revision
+ if (
+ not self.href.isascii()
+ or parsed.scheme != "https"
+ or not parsed.netloc
+ or parsed.username
+ or parsed.password
+ or "?" in self.href
+ or "#" in self.href
+ or not re.fullmatch(r"/[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+/" + suffix, parsed.path)
+ or re.search(r"[\s<>\\\[\](){}]", self.href)
+ ):
+ raise ValueError("A native delivery href must have its exact canonical HTTPS SCM path")
+ return self
+
+
+def delivery_links(data: Data, repository: Data | None, host: str) -> tuple[NativeDeliveryLinkV1, ...]:
+ # This producer consumes server-owned context after native feedback authority checks.
+ # DTO construction alone does not verify a journal or confer a project read grant.
+ if (
+ repository is None
+ or data.get("provider") != "github"
+ or (repository.get("connection_provider") or repository.get("provider")) != "github"
+ or urlsplit(host).scheme != "https"
+ ):
+ return ()
+ turn = data.get("natural_turn") or {}
+ context = turn.get("context") or {}
+ scope = context.get("scope") or {}
+ candidates = context.get("candidates") or []
+ if (
+ (turn.get("decision") or {}).get("intent") != "status"
+ or set(scope) != set(SCOPE_FIELDS)
+ or any(scope.get(key) != data.get(key) for key in ("repository_id", "actor", "issue"))
+ or scope.get("repository_id") != repository.get("id")
+ or not isinstance(candidates, list)
+ or len(candidates) > 200
+ ):
+ return ()
+ links: dict[str, NativeDeliveryLinkV1] = {}
+ for row in candidates:
+ if not isinstance(row, dict) or any(row.get(key) != scope.get(key) for key in SCOPE_FIELDS):
+ continue
+ delivery = row.get("delivery") or {}
+ checkpoint, remote = delivery.get("checkpoint") or {}, delivery.get("remote") or {}
+ if (
+ delivery.get("evidence") != "verified_native_journal"
+ or delivery.get("product_acceptance") != "not_established_by_checkpoint"
+ or remote.get("status") not in {"draft", "ready"}
+ ):
+ continue
+ review = _link(remote.get("pr_url"), host, repository)
+ if not review:
+ continue
+ parsed = urlsplit(review)
+ if not re.fullmatch(r"/[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+/pull/[1-9][0-9]{0,19}", parsed.path):
+ continue
+ try:
+ shared = {
+ "run_id": row.get("run_id"),
+ "repository_id": scope.get("repository_id"),
+ "project_id": scope.get("project_id"),
+ "checkpoint_revision": checkpoint.get("revision"),
+ "checkpoint_digest": checkpoint.get("checkpoint_digest"),
+ }
+ pull = NativeDeliveryLinkV1.model_validate({"kind": "pull_request", "href": review, **shared})
+ commit = NativeDeliveryLinkV1.model_validate(
+ {
+ "kind": "commit",
+ "href": urlunsplit(
+ (
+ "https",
+ parsed.netloc,
+ parsed.path.rsplit("/pull/", 1)[0] + "/commit/" + pull.checkpoint_revision,
+ "",
+ "",
+ )
+ ),
+ **shared,
+ }
+ )
+ except (TypeError, ValueError):
+ continue
+ for link in (pull, commit):
+ links.setdefault(link.href, link)
+ return tuple(links.values())
diff --git a/src/firefly_forge/scm_natural_models.py b/src/firefly_forge/scm_natural_models.py
index 266621e9..b9ebdd85 100644
--- a/src/firefly_forge/scm_natural_models.py
+++ b/src/firefly_forge/scm_natural_models.py
@@ -33,13 +33,13 @@ class NaturalDecision(BaseModel):
@model_validator(mode="after")
def unreviewed_dialogue(self) -> NaturalDecision:
- from firefly_forge.scm_natural_render import reply_quote
+ from firefly_forge.scm_natural_render import validate_reply_quote
if self.intent == "product" and self.questions:
raise ValueError("A product handoff leaves architectural follow-up questions to the native product service")
if any(not question.strip() for question in self.questions):
raise ValueError("Typed questions cannot be empty")
- reply_quote(self.message, self.questions, lambda text: text)
+ validate_reply_quote(self.message, self.questions)
if self.proposal is not None:
from firefly_forge.collaboration import ProductDefinition
@@ -201,6 +201,8 @@ def resolve_task(
row = rows[choice - 1]
if row.get("read_only") is True and decision.intent != "status":
return result.model_copy(update={"reason": "read_only_reference"})
+ if row.get("answer_only") is True and decision.intent == "resume":
+ return result.model_copy(update={"reason": "answer_only_reference"})
if row.get("origin") in {"product_assistant_turn", "natural_turn"}:
if decision.intent == "resume":
return result.model_copy(update={"reason": "product_answer_has_no_execution_authority"})
diff --git a/src/firefly_forge/scm_natural_render.py b/src/firefly_forge/scm_natural_render.py
index d3b930b9..8b4c3aae 100644
--- a/src/firefly_forge/scm_natural_render.py
+++ b/src/firefly_forge/scm_natural_render.py
@@ -1,18 +1,68 @@
# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0
-"""Bound the complete escaped AI quote before accepting its typed response."""
+"""Bound complete AI quotations and preserve only server-verified delivery hrefs."""
+import re
from collections.abc import Callable
-from firefly_forge.collaboration_feedback import _text
+from firefly_forge.collaboration_feedback import _escaped_text, _redact_settings, _text
+from firefly_forge.scm_natural_links import NativeDeliveryLinkV1
+_URL = r"https://[^\s<>\"\\()\[\]{}]+"
+_LINK = re.compile(
+ r"(?P!?)\[(?P