From 6badcc7f2b141473dcd3e6dd603ecedb4e2615c3 Mon Sep 17 00:00:00 2001 From: Andres Contreras Date: Sat, 24 Oct 2026 16:43:50 -0700 Subject: [PATCH 1/3] Release Forge 0.2.43 with durable checkpoint continuation --- CHANGELOG.md | 9 + docs/architecture.md | 8 +- docs/recovery.md | 13 +- docs/release-notes.md | 12 + docs/sandbox.md | 8 + docs/scm.md | 52 ++ pyproject.toml | 2 +- src/firefly_forge/__init__.py | 2 +- src/firefly_forge/agent_checkpoint_models.py | 2 +- src/firefly_forge/agent_checkpoints.py | 35 +- src/firefly_forge/agent_restores.py | 112 +++- .../agentic_checkpoint_boundary.py | 112 +++- .../agentic_checkpoint_invocation.py | 27 +- src/firefly_forge/agentic_runtime.py | 2 +- src/firefly_forge/api.py | 2 + src/firefly_forge/collaboration.py | 62 +- src/firefly_forge/collaboration_feedback.py | 9 +- src/firefly_forge/engine.py | 95 ++- src/firefly_forge/journal.py | 55 +- src/firefly_forge/runner_pool_worker.py | 131 ++-- src/firefly_forge/sandbox.py | 98 +-- src/firefly_forge/sandbox_command_slots.py | 294 ++++++++ src/firefly_forge/scm_connections.py | 12 +- src/firefly_forge/scm_conversations.py | 6 +- .../scm_maintain_ticket_controllers.py | 30 + .../scm_maintain_ticket_models.py | 108 +++ src/firefly_forge/scm_maintain_tickets.py | 376 +++++++++++ src/firefly_forge/scm_natural_effects.py | 4 +- src/firefly_forge/scm_natural_feedback.py | 9 +- src/firefly_forge/scm_natural_links.py | 117 ++++ src/firefly_forge/scm_natural_models.py | 6 +- src/firefly_forge/scm_natural_render.py | 60 +- src/firefly_forge/scm_natural_tasks.py | 8 + src/firefly_forge/source_budget_allowances.py | 2 + .../source_edit_qualification.py | 15 +- .../source_edit_supported_profiles.py | 328 +++++++++ tests/source_edit_qualification_fixtures.py | 2 + tests/test_agentic_checkpoint_codec.py | 503 ++++++++++++++ tests/test_agentic_checkpoint_codec_native.py | 113 ++++ ...t_agentic_checkpoint_frozen_design_host.py | 142 ++++ ...st_agentic_checkpoint_static_tools_host.py | 222 ++++++ ..._agentic_checkpoint_static_tools_native.py | 73 ++ ...agentic_completed_commands_restore_host.py | 265 ++++++++ ...entic_completed_commands_restore_native.py | 110 +++ tests/test_agentic_restore_receipt_native.py | 14 +- tests/test_product_context_projection_host.py | 148 ++++ tests/test_sandbox_command_ordinals.py | 114 ++++ tests/test_sandbox_command_ordinals_host.py | 631 ++++++++++++++++++ tests/test_scm_maintain_dispatch_host.py | 245 +++++++ tests/test_scm_maintain_question_host.py | 264 ++++++++ tests/test_scm_maintain_ticket_host.py | 329 +++++++++ tests/test_scm_maintain_ticket_native.py | 286 ++++++++ tests/test_scm_natural_trusted_links_host.py | 196 ++++++ ...test_source_budget_current_profile_host.py | 4 +- tests/test_source_budget_output_host.py | 6 +- ...test_source_edit_supported_profile_host.py | 13 +- tests/test_source_edit_v10_transition_host.py | 155 +++++ tests/test_source_edit_v11_transition_host.py | 54 ++ tests/test_source_edit_v9_transition_host.py | 4 +- tests/test_source_reader_contract_host.py | 4 +- uv.lock | 2 +- 61 files changed, 5957 insertions(+), 165 deletions(-) create mode 100644 src/firefly_forge/sandbox_command_slots.py create mode 100644 src/firefly_forge/scm_maintain_ticket_controllers.py create mode 100644 src/firefly_forge/scm_maintain_ticket_models.py create mode 100644 src/firefly_forge/scm_maintain_tickets.py create mode 100644 src/firefly_forge/scm_natural_links.py create mode 100644 tests/test_agentic_checkpoint_codec.py create mode 100644 tests/test_agentic_checkpoint_codec_native.py create mode 100644 tests/test_agentic_checkpoint_frozen_design_host.py create mode 100644 tests/test_agentic_checkpoint_static_tools_host.py create mode 100644 tests/test_agentic_checkpoint_static_tools_native.py create mode 100644 tests/test_agentic_completed_commands_restore_host.py create mode 100644 tests/test_agentic_completed_commands_restore_native.py create mode 100644 tests/test_product_context_projection_host.py create mode 100644 tests/test_sandbox_command_ordinals.py create mode 100644 tests/test_sandbox_command_ordinals_host.py create mode 100644 tests/test_scm_maintain_dispatch_host.py create mode 100644 tests/test_scm_maintain_question_host.py create mode 100644 tests/test_scm_maintain_ticket_host.py create mode 100644 tests/test_scm_maintain_ticket_native.py create mode 100644 tests/test_scm_natural_trusted_links_host.py create mode 100644 tests/test_source_edit_v10_transition_host.py create mode 100644 tests/test_source_edit_v11_transition_host.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 0f6e489f..7a1bf567 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,15 @@ +## 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. +- Qualify the V11 source profile with an explicit 34-module closure and fresh native/SQL evidence, retaining all six 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/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..7cdd668a 100644 --- a/docs/release-notes.md +++ b/docs/release-notes.md @@ -5,6 +5,18 @@ 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). + +Fresh controlled native tool-boundary and PostgreSQL measurements qualify the V11 34-module admission closure and its bounded source-read scope. All six 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..2d2b8834 100644 --- a/docs/sandbox.md +++ b/docs/sandbox.md @@ -25,6 +25,14 @@ 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. + ## 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..dd28a7a6 100644 --- a/src/firefly_forge/agentic_checkpoint_invocation.py +++ b/src/firefly_forge/agentic_checkpoint_invocation.py @@ -25,6 +25,7 @@ from firefly_forge.sandbox import pack_workspace 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 +55,21 @@ 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", "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", ) }, } @@ -127,8 +140,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..6b22e46d 100644 --- a/src/firefly_forge/agentic_runtime.py +++ b/src/firefly_forge/agentic_runtime.py @@ -432,7 +432,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( 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/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/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