Expose immutable source revisions and bounded authorized file reads

docs/api.md

@@ -15,6 +15,30 @@ Request and response bodies use JSON unless a route downloads an artifact or
 file. Project permissions apply to the authenticated identity. The actual
 running application's OpenAPI document is the authoritative schema.
 
+## Read recorded source
+
+These read-only endpoints require the current native identity and project read
+permission. Responses use `Cache-Control: private, no-store`. Snapshot IDs,
+catalog manifests and file indexes identify evidence; they do not grant access.
+
+| GET endpoint under `/api/v1/runs/{run_id}/source` | Result |
+| --- | --- |
+| `/revisions?after={snapshot_id}` | Up to 128 recorded snapshots and the next cursor; pending/failed publications remain explicit. |
+| `/revisions/{snapshot_id}/catalog?expected_manifest={sha256}&page=0` | An immutable source reference and a `SourceCatalogPageV1`. |
+| `/revisions/{snapshot_id}/files/{entry_index}?expected_manifest={sha256}&page=0&offset=0&limit=32768` | A verified byte window for an entry in that exact catalog page. |
+
+Catalog/file requests may include `request_id` for an existing Git delivery
+checkpoint; its recorded source tree must match the selected snapshot. File
+windows are at most 1 MiB and include exact byte offsets, a window SHA-256,
+Base64 content and optional UTF-8 text. `binary` describes the returned window,
+including a window that bisects a UTF-8 sequence; it does not guess a file type.
+
+The Python SDK exposes `run_source_revisions`, `run_source_catalog` and
+`run_source_file`. It validates response scope, indexes, extents and content
+hashes without retries. Unpublished snapshots have no content endpoint, and
+the server never substitutes a live workspace read. See
+[source storage and publication](architecture.md#read-recorded-source).
+
 ## Automatic verification handoffs
 
 New runs freeze `settings.automatic_verification` from `GET /tools/policy`.

docs/architecture.md

@@ -166,12 +166,50 @@ the retained central byte-bundle path. Existing local installations continue
 with that explicit binding when `object_storage` is absent; enabling a backend
 does not migrate or delete their files.
 
-The existing `SourceCatalogRefV1`, `SourceCatalogEntryV1` and `SourceCatalogPageV1`
-already provide paged immutable source metadata. A future file explorer should
-bind that catalog's source digest, paths, modes and hashes to retained object
-content, with the current project/run permission checked for every read. This
-artifact slice does not yet expose per-file content or grant access from an
-absolute local path or an object reference alone.
+#### Read recorded source
+
+The source explorer backend reuses `SourceCatalogRefV1`, `SourceCatalogEntryV1`
+and `SourceCatalogPageV1`. At the Engine snapshot boundary it captures the public
+catalog, stores each file through the native object publication ledger, and
+copies catalog pages into PostgreSQL. Roots have constant-size lineage and
+digest-chain fields; pages contain at most 128 entries. The source digest is
+recomputed from the ordered catalog before a page is returned. Source and owner
+fences cover publication, including the final compare-and-set. An incomplete or
+ambiguous archive stays unavailable and is never retried automatically.
+
+A factory with native object storage configured records fresh snapshots by
+default. Its private configuration requires `FORGE_OBJECT_STORAGE_TENANT_ID`
+alongside the existing `FORGE_OBJECT_STORAGE_*` provider settings. Administrators
+can set `FORGE_SOURCE_ARCHIVE_POLICY` to a JSON object with `enabled`, an
+`excluded_projects` array of exact project UUIDs, and
+`retention: "retain_until_explicit_admin_gc"`. This policy does not grant read
+access or delete retained data. Production GC and a portal storage setup wizard
+remain separate work; the implemented retention policy keeps recorded bytes.
+
+Authenticated GET routes list recorded snapshots and return catalog/file pages.
+Each response rechecks the native reader, current project access, run scope,
+tenant and immutable reference. An optional Git delivery request selects only a
+checkpoint with the same recorded source tree. Reads perform no Git or SCM
+network operation. `branch` and `head` describe the captured run baseline;
+`tree` describes the admitted snapshot, without claiming a remote branch update.
+
+File windows come only from verified object bytes, never the current workspace.
+Publication scans the complete file for configured credentials and verifies its
+SHA-256. Reads verify the bounded metadata chain, then fetch only chunks that
+intersect the requested window and its credential-scan overlap. They never
+restore the complete file or archive per window. The configured credential
+fingerprint must still match; a changed registry requires a fresh publication.
+Request-specific credential taints are checked against the returned byte range
+without changing that persistent fingerprint. Application memory scales with
+chunk/page width and the bounded response window. Publication scratch scales
+with the file being captured. Base64 carries binary or partial UTF-8 windows
+without inventing text. Private paths, aliases and symbolic links cannot become
+file capabilities.
+
+Existing installations without storage configuration are unchanged. Old runs
+and checkpoints without a complete publication explicitly have no archived
+content; there is no mutable fallback or retroactive migration. This backend
+does not provide an editor, a terminal or a complete live SCM branch browser.
 
 <figure class="forge-shot forge-architecture-diagram" markdown="1">
 <picture>

src/firefly_forge/api.py

@@ -247,8 +247,10 @@
 from firefly_forge.scm_source_edit_controllers import SCMSourceEditController
 from firefly_forge.search_tools import SearchTools
 from firefly_forge.secrets import redact_environment
+from firefly_forge.source_archive_service import SourceArchiveManager
 from firefly_forge.source_budget_allowances import SourceBudgetPolicies
 from firefly_forge.source_budget_controllers import SourceBudgetPolicyController
+from firefly_forge.source_explorer import SourceExplorer, SourceExplorerController
 from firefly_forge.store import Store
 from firefly_forge.task_routing import TaskRouting
 from firefly_forge.tool_broker import ToolInvocationBroker
@@ -686,6 +688,7 @@ def coordinator(
         routing: TaskRouting,
         runtime_roles: RuntimeRoles,
         worker_evidence: WorkerEvidence,
+        source_archives: SourceArchiveManager,
     ) -> Engine:
         engine = Engine(
             store,
@@ -696,6 +699,7 @@ def coordinator(
         )
         engine.assurance = assurance
         engine.worker_evidence = worker_evidence
+        engine.source_archives = source_archives
         engine.sandbox = sandbox
         engine.browser, engine.search = browser, search
         engine.tool_broker = tool_broker
@@ -1304,6 +1308,9 @@ class Application:
         OriginalSelectionReadmissionController,
         GitDiffArchive,
         GitDiffController,
+        SourceArchiveManager,
+        SourceExplorer,
+        SourceExplorerController,
         JournalController,
         VerificationHandoffController,
         ProviderRecoveryController,

src/firefly_forge/client.py

@@ -68,6 +68,7 @@
         SourceBudgetProposal,
     )
     from firefly_forge.source_edit_models import SourceEditContinuationInput
+    from firefly_forge.source_explorer import SourceCatalogObservationV1, SourceFilePageV1
     from firefly_forge.task_routing import TaskRoutingInput
     from firefly_forge.task_routing_mcp import TaskRoutingInput as MCPTaskRoutingInput
     from firefly_forge.unknown_command_models import UnknownCommandAcknowledgementV1
@@ -1418,6 +1419,139 @@ def run_diff_file(
             raise ValueError("The returned diff content differs from the requested verified window")
         return result
 
+    def run_source_revisions(self, run_id: str, *, after: str | None = None) -> dict[str, Any]:
+        from firefly_forge.source_archive_index import SourceArchiveRefV1
+
+        run_id = self._id(run_id)
+        if after is not None:
+            after = self._id(after)
+        result = self.request("GET", f"runs/{run_id}/source/revisions", params={"after": after} if after else {})
+        if (
+            not isinstance(result, dict)
+            or set(result) != {"run_id", "items", "next_after", "recording_enabled", "availability"}
+            or result["run_id"] != run_id
+            or type(result["recording_enabled"]) is not bool
+            or result["availability"] not in {"recorded", "unavailable"}
+            or not isinstance(result["items"], list)
+            or len(result["items"]) > 128
+        ):
+            raise ValueError("The returned source revisions differ from the requested run")
+        previous = after
+        for item in result["items"]:
+            if (
+                not isinstance(item, dict)
+                or set(item) != {"reference", "status", "reason"}
+                or item["status"] not in {"pending", "published", "failed"}
+            ):
+                raise ValueError("The returned source revision status is invalid")
+            ref = SourceArchiveRefV1.model_validate(item["reference"])
+            if ref.run_id != run_id or previous is not None and ref.snapshot_id <= previous:
+                raise ValueError("The returned source revision cursor or scope changed")
+            previous = ref.snapshot_id
+        if (
+            result["next_after"] not in (None, previous)
+            or result["next_after"] is not None
+            and len(result["items"]) != 128
+        ):
+            raise ValueError("The returned source revision cursor changed")
+        return dict(result)
+
+    def _source_path(self, run_id: str, snapshot_id: str, expected_manifest: str) -> str:
+        if not isinstance(expected_manifest, str) or re.fullmatch(r"[a-f0-9]{64}", expected_manifest) is None:
+            raise ValueError("An exact source catalog manifest is required")
+        return f"runs/{self._id(run_id)}/source/revisions/{self._id(snapshot_id)}"
+
+    def run_source_catalog(
+        self, run_id: str, snapshot_id: str, expected_manifest: str, *, page: int = 0, request_id: str | None = None
+    ) -> SourceCatalogObservationV1:
+        from firefly_forge.source_archive import public_entry
+        from firefly_forge.source_explorer import SourceCatalogObservationV1
+
+        if type(page) is not int or page < 0:
+            raise ValueError("A nonnegative source catalog page is required")
+        path = self._source_path(run_id, snapshot_id, expected_manifest) + "/catalog"
+        params = {"expected_manifest": expected_manifest, "page": page}
+        if request_id is not None:
+            params["request_id"] = self._id(request_id)
+        result = SourceCatalogObservationV1.model_validate(self.request("GET", path, params=params))
+        ref = result.reference
+        if (
+            ref.run_id != self._id(run_id)
+            or ref.snapshot_id != self._id(snapshot_id)
+            or ref.catalog.manifest_sha256 != expected_manifest
+            or result.catalog.manifest_sha256 != expected_manifest
+            or result.catalog.artifact_id != ref.catalog.artifact_id
+            or result.catalog.page != page
+            or len(result.catalog.entries) > 128
+            or result.catalog.next_page != (page + 1 if page + 1 < ref.catalog.page_count else None)
+        ):
+            raise ValueError("The returned source catalog differs from the requested immutable page")
+        for entry in result.catalog.entries:
+            public_entry(entry)
+        return result
+
+    def run_source_file(
+        self,
+        run_id: str,
+        snapshot_id: str,
+        expected_manifest: str,
+        *,
+        page: int,
+        entry_index: int,
+        offset: int = 0,
+        limit: int = 32768,
+        request_id: str | None = None,
+    ) -> SourceFilePageV1:
+        from firefly_forge.source_archive import public_entry
+        from firefly_forge.source_explorer import SourceFilePageV1
+
+        if (
+            type(page) is not int
+            or page < 0
+            or type(entry_index) is not int
+            or entry_index < 0
+            or type(offset) is not int
+            or offset < 0
+            or type(limit) is not int
+            or not 1 <= limit <= 1048576
+        ):
+            raise ValueError("Use a bounded source file window and exact catalog indexes")
+        path = self._source_path(run_id, snapshot_id, expected_manifest) + f"/files/{entry_index}"
+        params = {"expected_manifest": expected_manifest, "page": page, "offset": offset, "limit": limit}
+        if request_id is not None:
+            params["request_id"] = self._id(request_id)
+        result = SourceFilePageV1.model_validate(self.request("GET", path, params=params))
+        public_entry(result.entry)
+        content = base64.b64decode(result.content_base64, validate=True)
+        text_value: str | None
+        try:
+            text_value = content.decode("utf-8", errors="strict")
+            if "\0" in text_value:
+                text_value = None
+        except UnicodeDecodeError:
+            text_value = None
+        if (
+            result.reference.run_id != self._id(run_id)
+            or result.reference.snapshot_id != self._id(snapshot_id)
+            or result.reference.catalog.manifest_sha256 != expected_manifest
+            or result.catalog_page != page
+            or result.entry_index != entry_index
+            or result.entry.kind != "file"
+            or result.offset != offset
+            or result.byte_count > limit
+            or len(content) != result.byte_count
+            or offset + len(content) > result.entry.byte_count
+            or not content
+            and offset < result.entry.byte_count
+            or result.next_offset
+            != (offset + len(content) if offset + len(content) < result.entry.byte_count else None)
+            or hashlib.sha256(content).hexdigest() != result.sha256
+            or result.text != text_value
+            or result.binary is not (text_value is None)
+        ):
+            raise ValueError("The returned source content differs from its verified byte window")
+        return result
+
     def analytics_configuration(self, project_id: str) -> dict[str, Any]:
         return dict(self.request("GET", f"projects/{self._id(project_id)}/analytics/configuration"))
 

src/firefly_forge/engine.py

@@ -53,6 +53,7 @@
 if TYPE_CHECKING:
     from firefly_forge.consumed_terminal_completions import ConsumedTerminalExecution
     from firefly_forge.git_deliveries import GitDeliveries
+    from firefly_forge.source_archive_service import SourceArchiveManager
 
 
 async def checked(argv: list[str], cwd: Path, timeout: int = 300, *, env: dict[str, str] | None = None) -> str:
@@ -191,6 +192,7 @@ def __init__(
         self.knowledge_first: Any = None
         self.git_metadata: dict[Path, str] = {}
         self.worker_evidence: Any = None
+        self.source_archives: SourceArchiveManager | None = None
 
     async def start(self) -> None:
         if self.store.runtime_roles and not self.store.runtime_roles.executes_jobs:
@@ -2615,6 +2617,8 @@ async def admit(chunk: bytes, final: bool) -> None:
                 preview_bytes=preview_bytes,
                 preview_truncated=preview_bytes < artifact.byte_count,
             )
+            if archives := getattr(self, "source_archives", None):
+                await archives.publish(run, workspace, expected_head or base, tree, assert_current=current)
             return files, source, tree
 
     def _assert_git_metadata(self, workspace: Path) -> None: