diff --git a/docs/api.md b/docs/api.md index 6b96ae290..d0138338e 100644 --- a/docs/api.md +++ b/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`. diff --git a/docs/architecture.md b/docs/architecture.md index 1e74fc120..cb33f647d 100644 --- a/docs/architecture.md +++ b/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.
diff --git a/src/firefly_forge/api.py b/src/firefly_forge/api.py index 31f004b23..cfcddc6d6 100644 --- a/src/firefly_forge/api.py +++ b/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, diff --git a/src/firefly_forge/client.py b/src/firefly_forge/client.py index a75d13fb5..126d80bbe 100644 --- a/src/firefly_forge/client.py +++ b/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")) diff --git a/src/firefly_forge/engine.py b/src/firefly_forge/engine.py index c9e2c1506..9ae55b893 100644 --- a/src/firefly_forge/engine.py +++ b/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: diff --git a/src/firefly_forge/source_archive.py b/src/firefly_forge/source_archive.py new file mode 100644 index 000000000..7ae1588d3 --- /dev/null +++ b/src/firefly_forge/source_archive.py @@ -0,0 +1,266 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Immutable catalog-linked file bytes using the native paged object ledger.""" + +from __future__ import annotations + +import codecs +import hashlib +import json +import os +import re +import stat +from collections.abc import AsyncGenerator +from contextlib import aclosing +from contextvars import Context +from pathlib import Path +from typing import Any +from uuid import NAMESPACE_URL, UUID, uuid5 + +from firefly_forge.object_storage import ArtifactObjectStorage, Authorize, ObjectChunk, ObjectScope +from firefly_forge.runner_object_artifacts import ( + ArtifactLedger, + RunnerObjectArtifacts, + RunnerObjectChunk, + RunnerObjectManifest, +) +from firefly_forge.runner_pool_models import FactoryFence, pool_digest +from firefly_forge.secrets import KnownCredentialScan, credential_values +from firefly_forge.source_catalog import ( + SourceCatalogEntryV1, + _assert_record, + _directory, + _expected, + _private, + _read, + _stamp, +) +from firefly_forge.verification_source_preparation import PreparedVerificationSource + + +def catalog_fence(prepared: PreparedVerificationSource) -> None: + """Keep the complete captured file witnesses across the final SQL awaits.""" + prepared.fence() + source = _directory(prepared.workspace) + root = _directory(prepared.catalog_root, private=True) + folder = data = None + try: + folder = _directory(prepared.reference.artifact_id, root, private=True) + raw = _read(folder, "manifest.json", 1048576) + if hashlib.sha256(raw).hexdigest() != prepared.reference.manifest_sha256: + raise ValueError("Source catalog changed before publication") + manifest = json.loads(raw) + name = manifest["data_run"] + if not isinstance(name, str) or re.fullmatch(r"run-[a-f0-9]{32}", name) is None: + raise ValueError("Source catalog page namespace changed") + data = _directory(name, folder, private=True) + width = manifest["budget"]["chunk_bytes"] + for page in range(prepared.reference.page_count): + encoded = _read(data, f"page-{page:016x}.jsonl", width) + if hashlib.sha256(encoded).digest() != _expected(folder, manifest["proof"], width, page): + raise ValueError("Source catalog page changed before publication") + for line in encoded.splitlines(): + _assert_record(source, json.loads(line)) + prepared.fence() + finally: + for fd in (data, folder, root, source): + if fd is not None: + os.close(fd) + + +def public_entry(entry: SourceCatalogEntryV1) -> None: + parts = entry.path.rstrip("/").split("/") + if ( + not entry.path + or entry.path.startswith("/") + or "\\" in entry.path + or any(part in {"", ".", ".."} or any(ord(c) < 32 for c in part) for part in parts) + or _private(entry.path) + or entry.kind == "file" + and (not stat.S_ISREG(entry.mode) or entry.sha256 is None) + or entry.kind == "directory" + and (not stat.S_ISDIR(entry.mode) or entry.sha256 is not None) + ): + raise ValueError("Source content requires an exact public catalog entry") + + +class SourceArchiveContent: + def __init__(self, tenant: UUID, objects: ArtifactObjectStorage, ledger: ArtifactLedger, scratch: Path) -> None: + self.helper = RunnerObjectArtifacts(tenant, objects, ledger, scratch) + + def credential_fingerprint(self) -> str: + # Request-specific bearer taints must not make another authorized reader stale. + values = Context().run(credential_values, self.helper.credential_registry()) + return pool_digest({"configured_credentials": values}) + + def identity(self, binding: dict[str, Any], entry: SourceCatalogEntryV1) -> dict[str, Any]: + public_entry(entry) + if entry.kind != "file": + raise ValueError("Directories have no source content") + scope = ObjectScope(tenant_id=self.helper.tenant_id, project_id=UUID(binding["project_id"])) + receipt = {"kind": "source", "bytes": entry.byte_count, "sha256": entry.sha256} + width = self.helper.objects.settings.chunk_bytes + value = { + "protocol": "forge-source-content-v1", + "scope": scope.model_dump(mode="json"), + "snapshot_id": binding["snapshot_id"], + "source_digest": binding["source_digest"], + "entry": entry.model_dump(mode="json"), + "storage_binding": self.helper.objects.settings.binding, + "chunk_bytes": width, + "expected_chunks": (entry.byte_count + width - 1) // width, + "receipt": receipt, + } + return {"key": uuid5(NAMESPACE_URL, pool_digest(value)).hex, **value} + + async def _source(self, workspace: Path, entry: SourceCatalogEntryV1, current: Authorize) -> AsyncGenerator[bytes]: + parent = _directory(workspace) + fd = None + try: + parts = entry.path.split("/") + for part in parts[:-1]: + child = _directory(part, parent) + os.close(parent) + parent = child + fd = os.open(parts[-1], os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=parent) + before = os.fstat(fd) + if ( + not stat.S_ISREG(before.st_mode) + or before.st_uid != os.getuid() + or before.st_nlink != 1 + or before.st_mode != entry.mode + or before.st_size != entry.byte_count + ): + raise ValueError("Source file differs from its immutable catalog") + while data := os.read(fd, self.helper.objects.settings.chunk_bytes): + await current() + yield data + if _stamp(before) != _stamp(os.fstat(fd)) or _stamp(before) != _stamp( + os.stat(parts[-1], dir_fd=parent, follow_symlinks=False) + ): + raise ValueError("Source changed during content capture") + finally: + if fd is not None: + os.close(fd) + os.close(parent) + + async def publish( + self, + binding: dict[str, Any], + prepared: PreparedVerificationSource, + entry: SourceCatalogEntryV1, + fence: FactoryFence, + *, + authorize: Authorize, + ) -> dict[str, Any]: + identity = self.identity(binding, entry) + authorize = self.helper._current(authorize) + receipt = identity["receipt"] + async with ( + aclosing(self._source(prepared.workspace, entry, authorize)) as source, + self.helper._spooled(source, receipt, authorize) as spool, + ): + owner, prior = await self.helper.ledger.begin(identity, fence) + if prior is None: + + async def progress(chunk: ObjectChunk, confirmed: bool) -> None: + await authorize() + await self.helper.ledger.progress(identity, owner, chunk, confirmed, fence) + + count, head = 0, "0" * 64 + async for chunk in self.helper.objects.write_chunks( + ObjectScope.model_validate(identity["scope"]), + self.helper._parts(spool.file, authorize), + authorize=authorize, + progress=progress, + ): + count += 1 + head = RunnerObjectChunk.capture(count, head, chunk).hash + manifest = RunnerObjectManifest( + scope=ObjectScope.model_validate(identity["scope"]), + binding=identity["storage_binding"], + sha256=receipt["sha256"], + bytes=receipt["bytes"], + chunk_count=count, + chain_sha256=head, + ) + await self.helper._manifest(identity, manifest.model_dump(mode="json")) + await authorize() + await self.helper.ledger.publish(identity, owner, manifest, fence) + else: + await self.helper._manifest(identity, prior) + await authorize() + return { + "identity_sha256": pool_digest(identity), + "receipt": receipt, + "credential_fingerprint": self.credential_fingerprint(), + } + + async def window( + self, + binding: dict[str, Any], + entry: SourceCatalogEntryV1, + descriptor: dict[str, Any], + offset: int, + limit: int, + *, + authorize: Authorize, + ) -> bytes: + if type(offset) is not int or offset < 0 or type(limit) is not int or not 1 <= limit <= 1048576: + raise ValueError("Source content requires a bounded nonnegative byte window") + identity = self.identity(binding, entry) + if descriptor.get("credential_fingerprint") != self.credential_fingerprint(): + raise ValueError("Configured credentials changed; a fresh source publication is required") + if descriptor != { + "identity_sha256": pool_digest(identity), + "receipt": identity["receipt"], + "credential_fingerprint": self.credential_fingerprint(), + }: + raise ValueError("Source content differs from its catalog or immutable object binding") + if offset > entry.byte_count: + raise ValueError("Source window is outside the recorded file") + authorize = self.helper._current(authorize) + await authorize() + manifest = await self.helper._manifest(identity, await self.helper.ledger.load(identity)) + end = min(manifest.bytes, offset + limit) + scanner = KnownCredentialScan(credential_values(self.helper.credential_registry())) + guard = (scanner.overlap + 1) * 4 + scan_start, scan_end = max(0, offset - guard), min(manifest.bytes, end + guard) + decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") + result = bytearray() + count, size, head = 0, 0, "0" * 64 + async for envelope in self.helper.ledger.chunks(identity): + await authorize() + count += 1 + envelope.validate_chain(count, head) + chunk = envelope.chunk + expected = min(identity["chunk_bytes"], manifest.bytes - size) + if ( + chunk.scope != manifest.scope + or chunk.binding != manifest.binding + or count > manifest.chunk_count + or expected <= 0 + or chunk.bytes != expected + ): + raise ValueError("Source chunk scope or extent changed") + chunk_end = size + chunk.bytes + if end > offset and chunk_end > scan_start and size < scan_end: + data = await self.helper.objects.read_chunk(manifest.scope, chunk, authorize=authorize) + guarded = data[max(0, scan_start - size) : min(len(data), scan_end - size)] + if scanner.feed(decoder.decode(guarded)): + raise ValueError("Source window contains configured credentials") + first, last = max(offset, size), min(end, chunk_end) + if first < last: + result.extend(data[first - size : last - size]) + size = chunk_end + head = envelope.hash + if ( + count != manifest.chunk_count + or size != manifest.bytes + or head != manifest.chain_sha256 + or len(result) != end - offset + ): + raise ValueError("Source chunk ledger differs from its immutable file manifest") + if scanner.feed(decoder.decode(b"", final=True)) or scanner.feed(None): + raise ValueError("Source window contains configured credentials") + await authorize() + return bytes(result) diff --git a/src/firefly_forge/source_archive_index.py b/src/firefly_forge/source_archive_index.py new file mode 100644 index 000000000..f681943b5 --- /dev/null +++ b/src/firefly_forge/source_archive_index.py @@ -0,0 +1,293 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Native publication of bounded catalog pages and immutable source lineage.""" + +from __future__ import annotations + +import copy +import hashlib +import re +from collections.abc import Callable +from contextlib import AbstractAsyncContextManager +from typing import Any, Literal +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field +from pyfly.container import repository +from pyfly.data.relational.sqlalchemy import BaseEntity, Repository +from sqlalchemy import BigInteger, ForeignKey, UniqueConstraint, select, text +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.ext.asyncio import AsyncSession +from sqlalchemy.orm import Mapped, mapped_column + +from firefly_forge.runner_pool_models import FactoryFence, pool_digest +from firefly_forge.runner_pools import RunnerPools +from firefly_forge.source_archive import public_entry +from firefly_forge.source_catalog import SourceCatalogPageV1, SourceCatalogRefV1, _json + + +class SourceArchiveRefV1(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, strict=True, hide_input_in_errors=True) + protocol: Literal["forge-source-archive-v1"] = "forge-source-archive-v1" + snapshot_id: str = Field(pattern=r"^[a-f0-9]{32}$") + tenant_id: str = Field(pattern=r"^[a-f0-9]{32}$") + project_id: str = Field(pattern=r"^[a-f0-9]{32}$") + repository_id: str = Field(pattern=r"^[a-f0-9]{32}$") + run_id: str = Field(pattern=r"^[a-f0-9]{32}$") + attempt: int = Field(ge=0) + head: str = Field(pattern=r"^[a-f0-9]{40,64}$") + tree: str = Field(pattern=r"^[a-f0-9]{40,64}$") + branch: str | None = Field(default=None, max_length=1024) + catalog: SourceCatalogRefV1 + storage_binding: str = Field(pattern=r"^[a-f0-9]{64}$") + + def content_binding(self) -> dict[str, Any]: + return { + "snapshot_id": self.snapshot_id, + "project_id": self.project_id, + "source_digest": self.catalog.source_digest, + } + + def identity(self) -> dict[str, Any]: + return {**self.model_dump(mode="json", exclude={"catalog"}), "source_digest": self.catalog.source_digest} + + +class ForgeSourceArchiveEntity(BaseEntity): + __tablename__ = "forge_source_archives" + data: Mapped[dict[str, Any]] = mapped_column(JSONB) + + +class ForgeSourceArchivePageEntity(BaseEntity): + __tablename__ = "forge_source_archive_pages" + __table_args__ = (UniqueConstraint("snapshot_id", "page"),) + snapshot_id: Mapped[UUID] = mapped_column(ForeignKey("forge_source_archives.id")) + page: Mapped[int] = mapped_column(BigInteger) + data: Mapped[dict[str, Any]] = mapped_column(JSONB) + + +@repository +class ForgeSourceArchiveRepository(Repository[ForgeSourceArchiveEntity, UUID]): + async def locked(self, value: str) -> ForgeSourceArchiveEntity | None: + session = self._require_session() + await session.execute( + text("SELECT pg_advisory_xact_lock(hashtextextended(:key,0))"), {"key": "forge:source-archive:" + value} + ) + return await session.scalar( + select(ForgeSourceArchiveEntity).where(ForgeSourceArchiveEntity.id == UUID(value)).with_for_update() + ) + + async def for_run(self, run_id: str, after: str | None) -> list[ForgeSourceArchiveEntity]: + statement = select(ForgeSourceArchiveEntity).where( + ForgeSourceArchiveEntity.data["reference"]["run_id"].astext == run_id + ) + if after is not None: + statement = statement.where(ForgeSourceArchiveEntity.id > UUID(after)) + return list(await self._require_session().scalars(statement.order_by(ForgeSourceArchiveEntity.id).limit(129))) + + +@repository +class ForgeSourceArchivePageRepository(Repository[ForgeSourceArchivePageEntity, UUID]): + async def at(self, value: UUID, page: int) -> ForgeSourceArchivePageEntity | None: + return await self._require_session().scalar( + select(ForgeSourceArchivePageEntity).where( + ForgeSourceArchivePageEntity.snapshot_id == value, ForgeSourceArchivePageEntity.page == page + ) + ) + + +class SourceArchiveIndex: + def __init__( + self, pools: RunnerPools, roots: ForgeSourceArchiveRepository, pages: ForgeSourceArchivePageRepository + ) -> None: + self.pools, self.roots, self.pages = pools, roots, pages + + def _unit(self) -> AbstractAsyncContextManager[AsyncSession]: + if self.pools.sessions.current() is not None: + raise RuntimeError("Source publication requires an independent committed native unit") + return self.pools._unit() + + async def begin(self, reference: SourceArchiveRefV1, fence: FactoryFence) -> SourceArchiveRefV1 | None: + async with self._unit(): + await self.pools._owned(fence) + row = await self.roots.locked(reference.snapshot_id) + if row is not None: + prior = SourceArchiveRefV1.model_validate(row.data["reference"]) + if prior.identity() != reference.identity(): + raise ValueError("Source publication identity changed") + if row.data["status"] != "published": + raise RuntimeError("Prior source publication needs reconciliation; no automatic retry") + return prior + await self.roots.save( + ForgeSourceArchiveEntity( + id=UUID(reference.snapshot_id), + data={ + "reference": reference.model_dump(mode="json"), + "status": "pending", + "page_count": 0, + "page_head": "0" * 64, + "fence": fence.model_dump(mode="json"), + "reason": None, + }, + ) + ) + return None + + async def put( + self, + reference: SourceArchiveRefV1, + page: SourceCatalogPageV1, + content: list[dict[str, Any] | None], + fence: FactoryFence, + ) -> None: + if ( + page.artifact_id != reference.catalog.artifact_id + or page.manifest_sha256 != reference.catalog.manifest_sha256 + or len(page.entries) > 128 + or len(page.entries) != len(content) + or page.page >= reference.catalog.page_count + ): + raise ValueError("Source catalog page differs from its bounded publication") + async with self._unit(): + await self.pools._owned(fence) + row = await self.roots.locked(reference.snapshot_id) + if ( + row is None + or row.data["status"] != "pending" + or row.data["reference"] != reference.model_dump(mode="json") + or row.data["fence"] != fence.model_dump(mode="json") + or row.data["page_count"] != page.page + ): + raise ValueError("Source publication owner, catalog or sequence changed") + value = {"catalog": page.model_dump(mode="json"), "content": content, "previous": row.data["page_head"]} + digest = pool_digest(value) + await self.pages.save( + ForgeSourceArchivePageEntity( + snapshot_id=UUID(reference.snapshot_id), page=page.page, data={**value, "hash": digest} + ) + ) + row.data = {**row.data, "page_count": page.page + 1, "page_head": digest} + await self.roots.save(row) + + async def seal(self, reference: SourceArchiveRefV1, fence: FactoryFence, source_fence: Callable[[], None]) -> None: + async with self._unit(): + await self.pools._owned(fence) + row = await self.roots.locked(reference.snapshot_id) + if ( + row is None + or row.data["status"] != "pending" + or row.data["reference"] != reference.model_dump(mode="json") + or row.data["fence"] != fence.model_dump(mode="json") + or row.data["page_count"] != reference.catalog.page_count + ): + raise ValueError("Incomplete or changed source publication cannot be sealed") + source_fence() + row.data = {**row.data, "status": "published"} + await self.roots.save(row) + await self.roots._require_session().flush() + source_fence() + source_fence() + + async def fail(self, snapshot_id: str, fence: FactoryFence) -> None: + async with self._unit(): + await self.pools._owned(fence) + row = await self.roots.locked(snapshot_id) + if row is not None and row.data["fence"] == fence.model_dump(mode="json"): + row.data = { + **row.data, + "status": "failed", + "reason": "source_publication_failed_reconciliation_required", + } + await self.roots.save(row) + + async def get(self, snapshot_id: str) -> SourceArchiveRefV1: + async with self._unit(): + row = await self.roots.locked(snapshot_id) + if row is None or row.data["status"] != "published": + raise ValueError("Immutable source content is unavailable") + return SourceArchiveRefV1.model_validate(copy.deepcopy(row.data["reference"])) + + async def page( + self, reference: SourceArchiveRefV1, page: int + ) -> tuple[SourceCatalogPageV1, list[dict[str, Any] | None]]: + if type(page) is not int or page < 0 or page >= reference.catalog.page_count: + if page == 0 and reference.catalog.entry_count == 0: + return SourceCatalogPageV1( + artifact_id=reference.catalog.artifact_id, + manifest_sha256=reference.catalog.manifest_sha256, + page=0, + next_page=None, + entries=[], + ), [] + raise ValueError("Source catalog page is outside its immutable extent") + async with self._unit(): + root = await self.roots.locked(reference.snapshot_id) + if ( + root is None + or root.data["status"] != "published" + or root.data["reference"] != reference.model_dump(mode="json") + ): + raise ValueError("Published source reference changed") + head, result = "0" * 64, None + files_hash, source_hash = hashlib.sha256(b"{"), hashlib.sha256(b'{"files":{') + previous, entries, files, total = None, 0, 0, 0 + for number in range(reference.catalog.page_count): + row = await self.pages.at(UUID(reference.snapshot_id), number) + if row is None: + raise ValueError("Published source catalog page is unavailable") + value = row.data + if ( + value["previous"] != head + or pool_digest({k: v for k, v in value.items() if k != "hash"}) != value["hash"] + ): + raise ValueError("Published source catalog chain changed") + catalog = SourceCatalogPageV1.model_validate(value["catalog"]) + if ( + catalog.page != number + or catalog.artifact_id != reference.catalog.artifact_id + or catalog.manifest_sha256 != reference.catalog.manifest_sha256 + or len(catalog.entries) > 128 + or len(value["content"]) != len(catalog.entries) + or catalog.next_page != (number + 1 if number + 1 < reference.catalog.page_count else None) + ): + raise ValueError("Published source catalog scope changed") + head = value["hash"] + for entry in catalog.entries: + public_entry(entry) + if previous is not None and entry.path <= previous: + raise ValueError("Source catalog digest ordering changed") + payload_hash = entry.sha256 if entry.kind == "file" else "" + if payload_hash is None: + raise ValueError("Source catalog digest is unavailable") + encoded = ( + (b"," if entries else b"") + _json(entry.path) + b":" + _json(payload_hash + str(entry.mode)) + ) + files_hash.update(encoded) + source_hash.update(encoded) + entries += 1 + if entry.kind == "file": + files += 1 + total += entry.byte_count + previous = entry.path + if number == page: + result = catalog, copy.deepcopy(value["content"]) + files_hash.update(b"}") + source_hash.update(b'},"protocol":"forge-verification-source-v1"}') + if ( + files_hash.hexdigest() != reference.catalog.manifest_digest + or source_hash.hexdigest() != reference.catalog.source_digest + or (entries, files, total) + != (reference.catalog.entry_count, reference.catalog.file_count, reference.catalog.byte_count) + ): + raise ValueError("Published catalog differs from its immutable source digest") + if ( + head != root.data["page_head"] + or root.data["page_count"] != reference.catalog.page_count + or result is None + ): + raise ValueError("Published source catalog root changed") + return result + + async def revisions(self, run_id: str, after: str | None) -> list[dict[str, Any]]: + if after is not None and re.fullmatch(r"[a-f0-9]{32}", after) is None: + raise ValueError("Use an exact recorded revision cursor") + async with self._unit(): + return [copy.deepcopy(row.data) for row in await self.roots.for_run(run_id, after)] diff --git a/src/firefly_forge/source_archive_service.py b/src/firefly_forge/source_archive_service.py new file mode 100644 index 000000000..998131f8b --- /dev/null +++ b/src/firefly_forge/source_archive_service.py @@ -0,0 +1,232 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Native factory publication of immutable source, separate from browser reads.""" + +from __future__ import annotations + +import copy +import hashlib +import os +from collections.abc import Mapping +from pathlib import Path +from typing import Any +from uuid import NAMESPACE_URL, UUID, uuid5 + +from pydantic import BaseModel, ConfigDict, Field +from pyfly.container import service +from sqlalchemy import text + +from firefly_forge.object_storage import ( + ArtifactObjectStorage, + Authorize, + ObjectStorageSettings, + native_document_storage, +) +from firefly_forge.project_access import AccessDenied, ProjectAccess +from firefly_forge.runner_artifact_publications import ForgeRunnerArtifactRepository, RunnerArtifactPublications +from firefly_forge.runner_pool_models import FactoryFence, pool_digest +from firefly_forge.runner_pools import RunnerPools +from firefly_forge.secrets import contains_known_credential +from firefly_forge.source_archive import SourceArchiveContent, catalog_fence, public_entry +from firefly_forge.source_archive_index import ( + ForgeSourceArchiveEntity, + ForgeSourceArchivePageEntity, + ForgeSourceArchivePageRepository, + ForgeSourceArchiveRepository, + SourceArchiveIndex, + SourceArchiveRefV1, +) +from firefly_forge.source_catalog import SourceCatalogBudget, _directory, read_source_catalog_page +from firefly_forge.store import Store +from firefly_forge.verification_source_preparation import prepare_verification_source + + +class SourceArchivePolicy(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + enabled: bool = True + excluded_projects: frozenset[UUID] = frozenset() + retention: str = Field(default="retain_until_explicit_admin_gc", pattern=r"^retain_until_explicit_admin_gc$") + + @classmethod + def from_environment(cls, values: Mapping[str, str]) -> SourceArchivePolicy: + raw = values.get("FORGE_SOURCE_ARCHIVE_POLICY", "{}") + return cls.model_validate_json(raw) + + +class SourceArchivePublisher: + def __init__( + self, content: SourceArchiveContent, index: SourceArchiveIndex, catalogs: Path, budget: SourceCatalogBudget + ) -> None: + self.content, self.index, self.catalogs = content, index, catalogs + self.budget = budget.model_copy(update={"page_entries": min(128, budget.page_entries)}) + + async def publish( + self, run: dict[str, Any], workspace: Path, head: str, tree: str, fence: FactoryFence, *, authorize: Authorize + ) -> SourceArchiveRefV1: + authorize = self.content.helper._current(authorize) + await authorize() + root = self.catalogs / UUID(run["id"]).hex + root.mkdir(mode=0o700, exist_ok=True) + os.close(_directory(root, private=True)) + prepared = await prepare_verification_source(workspace, root, budget=self.budget, assert_current=authorize) + value = { + "tenant_id": self.content.helper.tenant_id.hex, + "project_id": UUID(run["project_id"]).hex, + "repository_id": UUID(run["repository_id"]).hex, + "run_id": UUID(run["id"]).hex, + "attempt": run["attempt"], + "head": head, + "tree": tree, + "branch": run.get("branch"), + "source_digest": prepared.source_digest, + "storage_binding": self.content.helper.objects.settings.binding, + "credential_scope": self.content.credential_fingerprint(), + } + snapshot_id = uuid5(NAMESPACE_URL, "forge-source-archive-v1:" + pool_digest(value)).hex + ref = SourceArchiveRefV1.model_validate( + {k: v for k, v in value.items() if k not in {"source_digest", "credential_scope"}} + | {"snapshot_id": snapshot_id, "catalog": prepared.reference.model_dump(mode="json")} + ) + await prepared.revalidate(assert_current=authorize) + prior = await self.index.begin(ref, fence) + if prior is not None: + catalog_fence(prepared) + await authorize() + return prior + try: + registry = self.content.helper.credential_registry() + if contains_known_credential(ref.model_dump(mode="json"), registry): + raise ValueError("Source publication metadata contains configured credentials") + for number in range(ref.catalog.page_count): + await authorize() + metadata = read_source_catalog_page(root, prepared.reference, number) + if contains_known_credential(metadata.model_dump(mode="json"), registry): + raise ValueError("Source catalog metadata contains configured credentials") + for page_number in range(ref.catalog.page_count): + await authorize() + page = read_source_catalog_page(root, prepared.reference, page_number) + content: list[dict[str, Any] | None] = [] + for entry in page.entries: + public_entry(entry) + content.append( + await self.content.publish(ref.content_binding(), prepared, entry, fence, authorize=authorize) + if entry.kind == "file" + else None + ) + catalog_fence(prepared) + await self.index.put(ref, page, content, fence) + await prepared.revalidate(assert_current=authorize) + await authorize() + await self.index.seal(ref, fence, lambda: catalog_fence(prepared)) + await authorize() + catalog_fence(prepared) + return ref + except BaseException: + # A failed or ambiguous publication stays visible as unavailable. No retry is dispatched. + await self.index.fail(snapshot_id, fence) + raise + + +@service +class SourceArchiveManager: + phase = 10 + + def __init__(self, store: Store) -> None: + self.store = store + self.settings = ObjectStorageSettings.from_environment(os.environ) + self.policy = SourceArchivePolicy.from_environment(os.environ) + self.tenant = ( + UUID(os.environ["FORGE_OBJECT_STORAGE_TENANT_ID"]) if self.settings.provider != "disabled" else None + ) + self.publisher: SourceArchivePublisher | None = None + + async def start(self) -> None: + if self.settings.provider == "disabled": + return + assert self.tenant is not None + root = self.store.path.parent / "source-archives" + for path in (root, root / "scratch", root / "catalogs"): + path.mkdir(mode=0o700, exist_ok=True) + os.close(_directory(path, private=True)) + schema = "forge_source_" + hashlib.sha256(self.store.schema.encode()).hexdigest()[:32] + if self.store.session_provider is None: + raise RuntimeError("Source storage requires initialized native PostgreSQL persistence") + pools = RunnerPools( + self.store.session_provider, schema, factory_schema=self.store.schema, object_artifacts=True + ) + await pools.initialize() + async with pools._unit() as session: + await session.execute( + text("SELECT pg_advisory_xact_lock(hashtextextended(:key,0))"), + {"key": "forge:source-index-migration:" + schema}, + ) + connection = await session.connection() + tables = [ + ForgeSourceArchiveEntity.metadata.tables[entity.__tablename__] + for entity in (ForgeSourceArchiveEntity, ForgeSourceArchivePageEntity) + ] + await connection.run_sync(lambda sync: ForgeSourceArchiveEntity.metadata.create_all(sync, tables=tables)) + ledger = RunnerArtifactPublications(pools, ForgeRunnerArtifactRepository()) + content = SourceArchiveContent( + self.tenant, + ArtifactObjectStorage(self.settings, native_document_storage(self.settings)), + ledger, + root / "scratch", + ) + content.helper.credential_registry = lambda: { + **self.store.credential_registry(), + **self.settings.credential_registry(), + } + self.publisher = SourceArchivePublisher( + content, + SourceArchiveIndex(pools, ForgeSourceArchiveRepository(), ForgeSourceArchivePageRepository()), + root / "catalogs", + SourceCatalogBudget(), + ) + + async def stop(self) -> None: + pass + + async def publish( + self, run: dict[str, Any], workspace: Path, head: str, tree: str, *, assert_current: Authorize + ) -> SourceArchiveRefV1 | None: + if ( + self.settings.provider == "disabled" + or not self.policy.enabled + or UUID(run["project_id"]) in self.policy.excluded_projects + ): + return None + if self.publisher is None: + raise RuntimeError("Configured immutable source storage is not initialized") + owner = self.store.execution_ownership_token + scope = { + key: copy.deepcopy(run.get(key)) + for key in ("id", "repository_id", "project_id", "execution_actor", "attempt", "branch") + } + + async def current() -> None: + await assert_current() + await self.store.assert_ownership() + present = await self.store.get_run(run["id"]) + if ( + owner is None + or self.store.execution_ownership_token is not owner + or any(present.get(k) != v for k, v in scope.items()) + ): + raise AccessDenied("Source publication owner or native scope changed") + if not present.get("execution_actor"): + raise AccessDenied("Source publication requires a recorded canonical actor") + await ProjectAccess(self.store).authorize_execution(present) + await self.store.assert_campaign_current(present) + if self.store.execution_ownership_token is not owner: + raise AccessDenied("Source publication native owner changed") + + await current() + async with self.store._unit() as session: + started = await session.scalar( + text("SELECT backend_start FROM pg_stat_activity WHERE pid=:pid"), {"pid": self.store._ownership_pid} + ) + pid = self.store._ownership_pid + if pid is None: + raise AccessDenied("Source publication requires the current native database owner") + fence = FactoryFence(factory_schema=self.store.schema, pid=pid, backend_started_at=started) + return await self.publisher.publish(run, workspace, head, tree, fence, authorize=current) diff --git a/src/firefly_forge/source_explorer.py b/src/firefly_forge/source_explorer.py new file mode 100644 index 000000000..717df7d5e --- /dev/null +++ b/src/firefly_forge/source_explorer.py @@ -0,0 +1,265 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Current-authorized observations of published source; never a live filesystem API.""" + +from __future__ import annotations + +import base64 +import copy +import hashlib +import re +from typing import Any +from uuid import UUID + +from pydantic import BaseModel, ConfigDict, Field +from pyfly.container import rest_controller, service +from pyfly.web import PathVar, QueryParam, get_mapping, request_mapping +from starlette.responses import JSONResponse, Response + +from firefly_forge.git_deliveries import GitDeliveries +from firefly_forge.identity import Principal +from firefly_forge.project_access import CURRENT_PRINCIPAL, AccessDenied, ProjectAccess +from firefly_forge.secrets import contains_known_credential +from firefly_forge.source_archive import public_entry +from firefly_forge.source_archive_index import SourceArchiveRefV1 +from firefly_forge.source_archive_service import SourceArchiveManager +from firefly_forge.source_catalog import SourceCatalogEntryV1, SourceCatalogPageV1 +from firefly_forge.store import identifier + + +class SourceCatalogObservationV1(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) + reference: SourceArchiveRefV1 + catalog: SourceCatalogPageV1 + + +class SourceFilePageV1(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True, strict=True) + reference: SourceArchiveRefV1 + entry: SourceCatalogEntryV1 + catalog_page: int = Field(ge=0) + entry_index: int = Field(ge=0) + offset: int = Field(ge=0) + byte_count: int = Field(ge=0, le=1048576) + next_offset: int | None + sha256: str = Field(pattern=r"^[a-f0-9]{64}$") + content_base64: str = Field(max_length=1398104) + binary: bool + text: str | None = Field(max_length=1048576) + + +@service +class SourceExplorer: + def __init__(self, manager: SourceArchiveManager, deliveries: GitDeliveries) -> None: + self.manager, self.deliveries = manager, deliveries + + async def _run(self, run_id: str, reader: Principal) -> dict[str, Any]: + if CURRENT_PRINCIPAL.get() != reader: + raise AccessDenied("Source observations require the same current native reader") + run = await self.deliveries.store.get_run(identifier(run_id).hex) + await self.deliveries._authorize(run, ProjectAccess.actor_snapshot(reader), "read") + if CURRENT_PRINCIPAL.get() != reader: + raise AccessDenied("The current source reader changed") + return { + k: copy.deepcopy(run.get(k)) + for k in ("id", "project_id", "repository_id", "component_id", "execution_actor") + } + + async def _capture( + self, run_id: str, snapshot_id: str, manifest: str, reader: Principal, request_id: str | None + ) -> tuple[SourceArchiveRefV1, dict[str, Any], dict[str, Any] | None]: + if re.fullmatch(r"[a-f0-9]{64}", manifest) is None: + raise ValueError("An exact source catalog manifest is required") + run = await self._run(run_id, reader) + publisher = self.manager.publisher + if publisher is None: + raise ValueError("Immutable source content is unavailable; no workspace fallback") + ref = await publisher.index.get(identifier(snapshot_id).hex) + if ( + ref.run_id != run_id + or ref.project_id != run["project_id"] + or ref.repository_id != run["repository_id"] + or self.manager.tenant is None + or ref.tenant_id != self.manager.tenant.hex + or ref.storage_binding != self.manager.settings.binding + or ref.catalog.manifest_sha256 != manifest + ): + raise ValueError("Source snapshot differs from its authorized run, tenant or manifest") + checkpoint = None + if request_id is not None: + request_id = identifier(request_id).hex + view = await self.deliveries.view(run_id, actor=ProjectAccess.actor_snapshot(reader)) + matches = [item for item in view["items"] if item["request_id"] == request_id] + if len(matches) != 1 or matches[0]["source_tree"] != ref.tree: + raise ValueError("Published source does not match the recorded checkpoint") + checkpoint = copy.deepcopy(matches[0]) + return ref, run, checkpoint + + def _private(self, result: dict[str, Any]) -> dict[str, Any]: + publisher = self.manager.publisher + if publisher and contains_known_credential(result, publisher.content.helper.credential_registry()): + raise ValueError("Source observation contains configured credentials") + return result + + async def revisions(self, run_id: str, *, after: str | None = None) -> dict[str, Any]: + run_id = identifier(run_id).hex + reader = CURRENT_PRINCIPAL.get() + if reader is None: + raise AccessDenied("Source observations require a current native reader") + before = await self._run(run_id, reader) + items = [] + rows = await self.manager.publisher.index.revisions(run_id, after) if self.manager.publisher else [] + for row in rows[:128]: + ref = SourceArchiveRefV1.model_validate(row["reference"]) + if ( + ref.run_id != run_id + or ref.project_id != before["project_id"] + or ref.repository_id != before["repository_id"] + or self.manager.tenant is None + or ref.tenant_id != self.manager.tenant.hex + ): + raise ValueError("Recorded source lineage differs from its current authorized scope") + items.append( + {"reference": ref.model_dump(mode="json"), "status": row["status"], "reason": row.get("reason")} + ) + if await self._run(run_id, reader) != before: + raise ValueError("Source observation scope changed") + return self._private( + { + "run_id": run_id, + "items": items, + "next_after": items[-1]["reference"]["snapshot_id"] if len(rows) > 128 else None, + "recording_enabled": self.manager.settings.provider != "disabled" + and self.manager.policy.enabled + and bool(before["project_id"]) + and UUID(before["project_id"]) not in self.manager.policy.excluded_projects, + "availability": "recorded" if items else "unavailable", + } + ) + + async def catalog( + self, run_id: str, snapshot_id: str, expected_manifest: str, *, page: int = 0, request_id: str | None = None + ) -> dict[str, Any]: + reader = CURRENT_PRINCIPAL.get() + if reader is None: + raise AccessDenied("Source observations require a current native reader") + run_id = identifier(run_id).hex + before = await self._capture(run_id, snapshot_id, expected_manifest, reader, request_id) + publisher = self.manager.publisher + assert publisher is not None + result, _ = await publisher.index.page(before[0], page) + if await self._capture(run_id, snapshot_id, expected_manifest, reader, request_id) != before: + raise ValueError("Source observation scope or checkpoint changed") + return self._private(SourceCatalogObservationV1(reference=before[0], catalog=result).model_dump(mode="json")) + + async def 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, + ) -> dict[str, Any]: + if type(entry_index) is not int or entry_index < 0: + raise ValueError("An exact catalog entry index is required") + reader = CURRENT_PRINCIPAL.get() + if reader is None: + raise AccessDenied("Source observations require a current native reader") + run_id = identifier(run_id).hex + before = await self._capture(run_id, snapshot_id, expected_manifest, reader, request_id) + publisher = self.manager.publisher + assert publisher is not None + catalog, descriptors = await publisher.index.page(before[0], page) + if entry_index >= len(catalog.entries): + raise ValueError("Source entry is outside its recorded catalog page") + entry = catalog.entries[entry_index] + public_entry(entry) + descriptor = descriptors[entry_index] + if entry.kind != "file" or descriptor is None: + raise ValueError("The recorded entry has no file content") + + async def current() -> None: + if await self._capture(run_id, snapshot_id, expected_manifest, reader, request_id) != before: + raise ValueError("Source observation scope or checkpoint changed") + + data = await publisher.content.window( + before[0].content_binding(), entry, descriptor, offset, limit, authorize=current + ) + text_value: str | None + try: + text_value = data.decode("utf-8", errors="strict") + if "\0" in text_value: + text_value = None + except UnicodeDecodeError: + text_value = None + result = SourceFilePageV1( + reference=before[0], + entry=entry, + catalog_page=page, + entry_index=entry_index, + offset=offset, + byte_count=len(data), + next_offset=offset + len(data) if offset + len(data) < entry.byte_count else None, + sha256=hashlib.sha256(data).hexdigest(), + content_base64=base64.b64encode(data).decode("ascii"), + binary=text_value is None, + text=text_value, + ) + await current() + return self._private(result.model_dump(mode="json")) + + +@rest_controller +@request_mapping("/api/v1/runs/{run_id}/source") +class SourceExplorerController: + def __init__(self, explorer: SourceExplorer) -> None: + self.explorer = explorer + + @get_mapping("/revisions") + async def revisions(self, run_id: PathVar[str], after: QueryParam[str | None] = None) -> Response: + return JSONResponse( + await self.explorer.revisions(run_id, after=after), headers={"Cache-Control": "private, no-store"} + ) + + @get_mapping("/revisions/{snapshot_id}/catalog") + async def catalog( + self, + run_id: PathVar[str], + snapshot_id: PathVar[str], + expected_manifest: QueryParam[str], + page: QueryParam[int] = 0, + request_id: QueryParam[str | None] = None, + ) -> Response: + return JSONResponse( + await self.explorer.catalog(run_id, snapshot_id, expected_manifest, page=page, request_id=request_id), + headers={"Cache-Control": "private, no-store"}, + ) + + @get_mapping("/revisions/{snapshot_id}/files/{entry_index}") + async def file( + self, + run_id: PathVar[str], + snapshot_id: PathVar[str], + entry_index: PathVar[int], + expected_manifest: QueryParam[str], + page: QueryParam[int] = 0, + offset: QueryParam[int] = 0, + limit: QueryParam[int] = 32768, + request_id: QueryParam[str | None] = None, + ) -> Response: + return JSONResponse( + await self.explorer.file( + run_id, + snapshot_id, + expected_manifest, + page=page, + entry_index=entry_index, + offset=offset, + limit=limit, + request_id=request_id, + ), + headers={"Cache-Control": "private, no-store"}, + ) diff --git a/studio/playwright.source-explorer.config.ts b/studio/playwright.source-explorer.config.ts new file mode 100644 index 000000000..1dc98412a --- /dev/null +++ b/studio/playwright.source-explorer.config.ts @@ -0,0 +1,21 @@ +/* Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 */ +import { defineConfig } from "@playwright/test"; +export default defineConfig({ + testDir: "./tests", + testMatch: "source-explorer.spec.ts", + workers: 1, + timeout: 30000, + globalTimeout: 150000, + use: { + baseURL: "http://127.0.0.1:5197", + trace: "off", + screenshot: "only-on-failure", + }, + webServer: { + command: "node tests/source-explorer-server.mjs", + url: "http://127.0.0.1:5197/__source_explorer__", + reuseExistingServer: false, + timeout: 20000, + gracefulShutdown: { signal: "SIGTERM", timeout: 2000 }, + }, +}); diff --git a/studio/src/SourceExplorer.css b/studio/src/SourceExplorer.css new file mode 100644 index 000000000..969a7a427 --- /dev/null +++ b/studio/src/SourceExplorer.css @@ -0,0 +1,335 @@ +/* Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 */ +.source-explorer { + min-width: 0; + color: var(--text); + margin: 20px 0; +} +.source-heading { + display: flex; + align-items: start; + justify-content: space-between; + gap: 20px; + margin-bottom: 20px; +} +.source-heading h3 { + display: flex; + align-items: center; + gap: 10px; + margin: 0 0 8px; + font-size: 18px; +} +.source-heading p { + margin: 0; + color: var(--muted); + max-width: 65ch; + font-size: 14px; + line-height: 1.6; +} +.source-explorer button, +.source-explorer select { + min-height: 44px; + white-space: normal; +} +.source-explorer button:focus-visible, +.source-explorer select:focus-visible, +.source-code:focus-visible, +.source-provenance summary:focus-visible { + outline: 2px solid var(--accent-ink); + outline-offset: 3px; +} +.source-selectors { + display: grid; + grid-template-columns: minmax(0, 1fr) minmax(0, 1.3fr); + gap: 16px; + margin-bottom: 16px; +} +.source-selectors label { + display: grid; + gap: 8px; + min-width: 0; + font-size: 13px; +} +.source-selectors label > span { + display: flex; + align-items: center; + gap: 8px; +} +.source-selectors select { + width: 100%; + min-width: 0; + padding: 10px 12px; + border: 1px solid var(--line); + border-radius: 6px; + background: var(--input); + color: var(--text); +} +.source-provenance { + margin: 0 0 20px; + font-size: 13px; +} +.source-provenance summary { + color: var(--accent-ink); + cursor: pointer; + min-height: 44px; + padding: 4px 0; +} +.source-provenance .rich-markdown { + color: var(--muted); + max-width: 70ch; +} +.source-provenance dl { + display: grid; + gap: 12px; + margin: 16px 0; +} +.source-provenance dl > div { + display: grid; + grid-template-columns: minmax(140px, 20%) minmax(0, 1fr); + gap: 12px; +} +.source-provenance dt { + color: var(--muted); +} +.source-provenance dd { + margin: 0; + overflow-wrap: anywhere; + font-variant-numeric: tabular-nums; +} +.source-browser { + display: grid; + grid-template-columns: minmax(220px, 28%) minmax(0, 1fr); + border: 1px solid var(--line); + border-radius: 10px; + overflow: hidden; + background: var(--surface); +} +.source-browser > aside { + min-width: 0; + padding: 16px; + border-right: 1px solid var(--line); + background: var(--surface-subtle); +} +.source-index-heading { + display: flex; + justify-content: space-between; + align-items: center; + gap: 12px; + margin-bottom: 16px; +} +.source-index-heading h4 { + margin: 0; + font-size: 14px; +} +.source-picker-toggle { + display: none; +} +.source-tree { + list-style: none; + padding: 0; + margin: 0; + max-height: 32rem; + overflow: auto; + scrollbar-gutter: stable; + scrollbar-color: var(--muted) var(--surface-subtle); +} +.source-tree li { + margin: 0 0 6px; +} +.source-tree button { + display: flex; + align-items: start; + gap: 10px; + width: 100%; + padding: 12px; + padding-inline-start: calc(12px + var(--source-depth, 0) * 8px); + text-align: left; + border: 1px solid transparent; + border-radius: 6px; + background: transparent; + color: var(--text); + font-size: 13px; +} +.source-tree button svg { + flex-shrink: 0; + margin-top: 2px; + color: var(--muted); +} +.source-tree button > span { + min-width: 0; + overflow-wrap: anywhere; +} +.source-tree button small { + display: block; + color: var(--muted); + font-size: 12px; + margin-top: 5px; +} +.source-tree button:hover, +.source-tree button[aria-current="true"] { + border-color: var(--accent-line); + background: var(--surface); +} +.source-file-content { + min-width: 0; + padding: 20px; + overflow: hidden; +} +.source-file-content:focus-visible { + outline: 2px solid var(--accent-ink); + outline-offset: -3px; +} +.source-file-count { + margin: 0 0 12px; + color: var(--muted); + font-size: 12px; +} +.source-file-heading h4 { + display: flex; + gap: 10px; + align-items: start; + margin: 0 0 8px; + overflow-wrap: anywhere; + font-size: 14px; +} +.source-file-heading h4 svg { + flex-shrink: 0; + margin-top: 2px; +} +.source-file-heading p { + color: var(--muted); + font-size: 12px; + margin: 0 0 16px; +} +.source-window-meta { + display: flex; + justify-content: space-between; + align-items: center; + flex-wrap: wrap; + gap: 8px 16px; + margin-bottom: 12px; + font-size: 12px; + font-variant-numeric: tabular-nums; +} +.source-window-meta p { + margin: 0; + color: var(--muted); +} +.source-window-meta span { + display: flex; + align-items: center; + gap: 6px; + color: var(--success-ink); +} +.source-code { + max-width: 100%; + max-height: 32rem; + overflow: auto; + scrollbar-color: var(--muted) var(--surface-subtle); + padding: 16px; + margin: 0; + border: 1px solid var(--line); + border-radius: 6px; + background: var(--surface-subtle); + color: var(--text); + font-size: 12px; + line-height: 1.75; + tab-size: 4; +} +.source-code code { + font: inherit; +} +.source-hex { + white-space: pre-wrap; + overflow-wrap: anywhere; +} +.source-byte-note { + font-size: 13px; + color: var(--muted); + line-height: 1.6; + max-width: 65ch; + margin: 0 0 12px; +} +.source-pagination { + display: flex; + flex-wrap: wrap; + align-items: center; + justify-content: space-between; + gap: 10px; + margin-top: 16px; + font-size: 12px; + color: var(--muted); +} +.source-file-pages { + display: grid; + grid-template-columns: minmax(0, 1fr); + text-align: center; +} +.source-notice { + border: 1px solid var(--line); + border-radius: 8px; + padding: 20px; + background: var(--surface-subtle); +} +.source-notice h4 { + margin: 0 0 8px; + font-size: 15px; + overflow-wrap: anywhere; +} +.source-notice .rich-markdown { + font-size: 14px; + color: var(--muted); + line-height: 1.6; + max-width: 65ch; +} +.source-notice .rich-markdown > p { + margin: 0; +} +@media (max-width: 800px) { + .source-browser { + grid-template-columns: minmax(0, 1fr); + } + .source-browser > aside { + border-right: 0; + border-bottom: 1px solid var(--line); + } + .source-file-content { + padding: 16px; + } + .source-tree { + max-height: 15rem; + } + .source-picker-toggle { + display: inline-flex; + } + .source-tree-list.collapsed { + display: none; + } + .source-index-heading { + margin-bottom: 0; + } + .source-tree-list.expanded { + margin-top: 16px; + } + .source-file-pages { + display: flex; + } +} +@media (max-width: 540px) { + .source-heading { + flex-direction: column; + gap: 16px; + } + .source-heading > button { + width: 100%; + } + .source-selectors { + grid-template-columns: minmax(0, 1fr); + } + .source-provenance dl > div { + grid-template-columns: minmax(0, 1fr); + gap: 4px; + } + .source-pagination button { + flex: 1 1 120px; + } +} diff --git a/studio/src/SourceExplorer.test.tsx b/studio/src/SourceExplorer.test.tsx new file mode 100644 index 000000000..bb233acd8 --- /dev/null +++ b/studio/src/SourceExplorer.test.tsx @@ -0,0 +1,28 @@ +/* Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 */ +import { renderToStaticMarkup } from "react-dom/server"; +import { expect, it } from "vitest"; +import { SourceExplorer } from "./SourceExplorer"; + +it("starts with a read-only verified-source loading state and never exposes connection credentials", () => { + const html = renderToStaticMarkup( + , + ); + expect(html).toContain("Loading recorded snapshots"); + expect(html).toContain("Recorded source snapshot"); + expect(html).not.toContain("private-ui-token-never-display"); + expect(html).not.toMatch(/>Save|>Run|>Commit| { + const html = renderToStaticMarkup( + , + ); + expect(html).toContain("Select a recorded run"); + expect(html).not.toContain("../private"); + expect(html).not.toContain("Loading recorded snapshots"); +}); diff --git a/studio/src/SourceExplorer.tsx b/studio/src/SourceExplorer.tsx new file mode 100644 index 000000000..33784aa61 --- /dev/null +++ b/studio/src/SourceExplorer.tsx @@ -0,0 +1,705 @@ +/* Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 */ +import { + Fragment, + useCallback, + useEffect, + useMemo, + useRef, + useState, + type CSSProperties, +} from "react"; +import { + Check, + ChevronLeft, + ChevronRight, + FileCode2, + Files, + Folder, + GitBranch, + RefreshCw, +} from "lucide-react"; +import { request, type Connection } from "./api"; +import { diffBytes } from "./GitDiffBrowser"; +import { tokenizeCode } from "./ExecutionFormat"; +import { RichMarkdown } from "./RichMarkdown"; +import { + sourceRevisions, + sourceCatalog, + sourceFile, + verifySourceContent, + alignedSourceText, + sourceProblem, + type SourceRevisions, + type SourceRevision, + type SourceReference, + type SourceCatalog, + type SourceEntry, + type SourceFile, +} from "./SourceExplorerData"; +import "./SourceExplorer.css"; + +type Props = { connection: Connection; runId: string; requestId?: string }; +type Problem = ReturnType; +const number = (value: number) => value.toLocaleString("en-US"); +function Notice({ + title, + message, + alert = false, +}: { + title: string; + message: string; + alert?: boolean; +}) { + return ( +
+

{title}

+ +
+ ); +} +export function SourceExplorer(props: Props) { + if ( + !/^[a-f0-9]{32}$/.test(props.runId) || + (props.requestId !== undefined && !/^[a-f0-9]{32}$/.test(props.requestId)) + ) + return ( + + ); + return ( + + ); +} +function SourceSession({ connection, runId, requestId }: Props) { + const { base, token, authenticated } = connection; + const scoped = useMemo( + () => ({ base, token, authenticated }), + [base, token, authenticated], + ); + const [records, setRecords] = useState(null), + [problem, setProblem] = useState(null); + const [after, setAfter] = useState(null), + [reload, setReload] = useState(0); + const [branch, setBranch] = useState(null), + [selectedId, setSelectedId] = useState(""); + const path = `/runs/${encodeURIComponent(runId)}/source/revisions`; + const hold = useCallback((error: unknown) => { + setRecords(null); + setProblem(sourceProblem(error)); + }, []); + useEffect(() => { + const controller = new AbortController(); + void request(scoped, path + (after ? `?after=${after}` : ""), { + signal: controller.signal, + }) + .then((raw) => { + const value = sourceRevisions(raw, runId, after); + if (controller.signal.aborted) return; + const first = + value.items.find((item) => item.status === "published") ?? + value.items[0]; + setRecords(value); + setProblem(null); + setBranch(first?.reference.branch ?? null); + setSelectedId(first?.reference.snapshot_id ?? ""); + }) + .catch((error: unknown) => { + if (!controller.signal.aborted) hold(error); + }); + return () => controller.abort(); + }, [scoped, path, after, reload, runId, hold]); + function refresh(cursor = after) { + setRecords(null); + setProblem(null); + setSelectedId(""); + setAfter(cursor); + setReload((value) => value + 1); + } + const branches = records + ? [...new Set(records.items.map((item) => item.reference.branch))] + : []; + const options = + records?.items.filter((item) => item.reference.branch === branch) ?? []; + const selected = records?.items.find( + (item) => item.reference.snapshot_id === selectedId, + ); + function chooseBranch(value: string) { + const next = + branches.find((candidate) => (candidate ?? "") === value) ?? null; + setBranch(next); + const items = + records?.items.filter((item) => item.reference.branch === next) ?? []; + setSelectedId( + (items.find((item) => item.status === "published") ?? items[0])?.reference + .snapshot_id ?? "", + ); + } + return ( +
+
+
+

+

+

+ Read archived files and their provenance. This view is read only. +

+
+ +
+ {problem ? ( + + ) : !records ? ( +

Loading recorded snapshots…

+ ) : !records.items.length ? ( + + ) : ( + <> +
+ + +
+ {selected && ( + + )} + + )} + {records && (after !== null || records.next_after !== null) && ( + + )} +
+ ); +} +function Snapshot({ + selected, + connection, + requestId, + onProblem, +}: { + selected: SourceRevision; + connection: Connection; + requestId?: string; + onProblem: (error: unknown) => void; +}) { + const reference = selected.reference; + return ( + <> +
+ View snapshot provenance + +
+ {[ + ["Recorded branch", reference.branch ?? "Not recorded"], + ["Baseline commit (context)", reference.head], + ["Recorded tree (context)", reference.tree], + ["Source catalog digest", reference.catalog.source_digest], + ["Catalog manifest", reference.catalog.manifest_sha256], + ["Snapshot ID", reference.snapshot_id], + ].map(([label, value]) => ( +
+
{label}
+
{value}
+
+ ))} +
+
+ {selected.status !== "published" ? ( + + ) : ( + + )} + + ); +} +function CatalogBrowser({ + connection, + reference, + requestId, + onProblem, +}: { + connection: Connection; + reference: SourceReference; + requestId?: string; + onProblem: (error: unknown) => void; +}) { + const [page, setPage] = useState(0); + return ( + + ); +} +function CatalogPage({ + connection, + reference, + requestId, + page, + setPage, + onProblem, +}: { + connection: Connection; + reference: SourceReference; + requestId?: string; + page: number; + setPage: (page: number) => void; + onProblem: (error: unknown) => void; +}) { + const contentTarget = useRef(null); + const [catalog, setCatalog] = useState(null), + [selected, setSelected] = useState(null), + [showFiles, setShowFiles] = useState(false); + const path = `/runs/${reference.run_id}/source/revisions/${reference.snapshot_id}`; + const manifest = `expected_manifest=${reference.catalog.manifest_sha256}${requestId ? `&request_id=${requestId}` : ""}`; + useEffect(() => { + const controller = new AbortController(); + void request( + connection, + `${path}/catalog?${manifest}&page=${page}`, + { signal: controller.signal }, + ) + .then((raw) => { + const value = sourceCatalog(raw, reference, page); + if (controller.signal.aborted) return; + setCatalog(value); + const file = value.catalog.entries.findIndex( + (item) => item.kind === "file", + ); + setSelected(file >= 0 ? file : value.catalog.entries.length ? 0 : null); + }) + .catch((error: unknown) => { + if (!controller.signal.aborted) onProblem(error); + }); + return () => controller.abort(); + }, [connection, path, manifest, page, reference, onProblem]); + const entry = selected === null ? null : catalog?.catalog.entries[selected]; + const listId = `source-files-${reference.snapshot_id}-${page}`; + return ( +
+ +
+ {!catalog ? ( +

Loading source catalog…

+ ) : !entry ? ( + + ) : entry.kind === "directory" ? ( + + ) : ( + + )} +
+
+ ); +} +function SourceFileBrowser(props: { + connection: Connection; + reference: SourceReference; + entry: SourceEntry; + page: number; + entryIndex: number; + path: string; + manifest: string; + onProblem: (error: unknown) => void; +}) { + const [offset, setOffset] = useState(0); + return ( + <> +
+

+

+

+ {diffBytes(props.entry.byte_count)} · Mode{" "} + {(props.entry.mode & 0o777).toString(8).padStart(4, "0")} +

+
+ + + ); +} +function FileWindow({ + connection, + reference, + entry, + page, + entryIndex, + path, + manifest, + onProblem, + offset, + setOffset, +}: { + connection: Connection; + reference: SourceReference; + entry: SourceEntry; + page: number; + entryIndex: number; + path: string; + manifest: string; + onProblem: (error: unknown) => void; + offset: number; + setOffset: (offset: number) => void; +}) { + const [content, setContent] = useState<{ + page: SourceFile; + bytes: Uint8Array; + text: string | null; + textStart: number; + textEnd: number; + } | null>(null), + [expanded, setExpanded] = useState(false); + useEffect(() => { + const controller = new AbortController(); + void (async () => { + async function read(start: number, limit: number) { + if (controller.signal.aborted) + throw new DOMException("Read cancelled", "AbortError"); + const raw = await request( + connection, + `${path}/files/${entryIndex}?${manifest}&page=${page}&offset=${start}&limit=${limit}`, + { signal: controller.signal }, + ); + const value = sourceFile( + raw, + reference, + page, + entryIndex, + entry, + start, + limit, + ); + return { page: value, ...(await verifySourceContent(value)) }; + } + const window = await read(offset, 32768); + if (controller.signal.aborted) return; + const preview = await alignedSourceText(window, read); + if (!controller.signal.aborted) + setContent({ + ...window, + text: preview.text, + textStart: preview.start, + textEnd: preview.end, + }); + })().catch((error: unknown) => { + if (!controller.signal.aborted) onProblem(error); + }); + return () => controller.abort(); + }, [ + connection, + path, + manifest, + page, + entryIndex, + entry, + reference, + offset, + onProblem, + ]); + if (!content) return

Verifying source window…

; + const bytes = expanded ? content.bytes : content.bytes.subarray(0, 256); + return ( + <> +
+

+ {entry.byte_count + ? `Bytes ${number(offset + 1)}–${number(offset + content.page.byte_count)} of ${number(entry.byte_count)}` + : "Empty file"} +

+ + +
+ {(content.textStart !== offset || + content.textEnd !== offset + content.page.byte_count) && ( +

+ Text preview aligned to UTF-8: bytes {number(content.textStart + 1)}– + {number(content.textEnd)}. Adjacent windows can overlap at a character + boundary. +

+ )} + {content.text === null ? ( + <> +

+ Raw byte window. It may contain binary data or cut a UTF-8 + character; hexadecimal preserves the exact bytes. +

+ 
+            {Array.from(bytes, (byte) =>
+              byte.toString(16).padStart(2, "0"),
+            ).join(" ")}
+
+ {content.bytes.length > 256 && ( + + )} + + ) : ( + 
+          
+            {content.text.split("\n").map((line, index, lines) => (
+              
+                {tokenizeCode(line, entry.path).map((token, i) => (
+                  
+                    {token.text}
+                  
+                ))}
+                {index < lines.length - 1 ? "\n" : ""}
+              
+            ))}
+          
+
+ )} + + + ); +} diff --git a/studio/src/SourceExplorerData.test.ts b/studio/src/SourceExplorerData.test.ts new file mode 100644 index 000000000..63e744038 --- /dev/null +++ b/studio/src/SourceExplorerData.test.ts @@ -0,0 +1,219 @@ +/* Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 */ +import { expect, it } from "vitest"; +import { + sourceRevisions, + sourceCatalog, + sourceFile, + verifySourceContent, + alignedSourceText, +} from "./SourceExplorerData"; + +const sha = "ba7816bf8f01cfea414140de5dae2223b00361a396177a9cb410ff61f20015ad"; +const reference = { + protocol: "forge-source-archive-v1", + snapshot_id: "b".repeat(32), + tenant_id: "1".repeat(32), + project_id: "2".repeat(32), + repository_id: "3".repeat(32), + run_id: "a".repeat(32), + attempt: 3, + head: "4".repeat(40), + tree: "5".repeat(40), + branch: "forge/retained-source", + storage_binding: "9".repeat(64), + catalog: { + version: 1, + artifact_id: "c".repeat(32), + manifest_sha256: "d".repeat(64), + source_digest: "e".repeat(64), + manifest_digest: "f".repeat(64), + byte_count: 3, + file_count: 1, + entry_count: 2, + page_count: 1, + }, +}; +const entry = { + path: "src/app.ts", + kind: "file" as const, + mode: 33188, + byte_count: 3, + sha256: sha, +}; +function revisions() { + return { + run_id: reference.run_id, + items: [{ reference, status: "published", reason: null }], + next_after: null, + recording_enabled: true, + availability: "recorded", + }; +} +function catalog() { + return { + reference, + catalog: { + version: 1, + artifact_id: reference.catalog.artifact_id, + manifest_sha256: reference.catalog.manifest_sha256, + page: 0, + next_page: null, + entries: [entry], + }, + }; +} +function file() { + return { + reference, + entry, + catalog_page: 0, + entry_index: 0, + offset: 0, + byte_count: 3, + next_offset: null, + sha256: sha, + content_base64: "YWJj", + binary: false, + text: "abc", + }; +} +it("keeps recorded revisions contextual and binds catalog/file bytes to their exact snapshot", async () => { + const ref = sourceRevisions(revisions(), reference.run_id).items[0].reference; + const page = sourceCatalog(catalog(), ref, 0); + const result = sourceFile(file(), ref, 0, 0, page.catalog.entries[0], 0); + expect((await verifySourceContent(result)).text).toBe("abc"); + expect(result.reference.head).toBe(reference.head); +}); +it.each([ + "run", + "snapshot", + "manifest", + "project", + "storage", + "page", + "entry", + "offset", + "extent", + "path", +])("refuses mismatched %s before displaying any source", (mutation) => { + const ref = sourceRevisions(revisions(), reference.run_id).items[0].reference; + const value = structuredClone(file()); + if (mutation === "run") value.reference.run_id = "f".repeat(32); + if (mutation === "snapshot") value.reference.snapshot_id = "f".repeat(32); + if (mutation === "manifest") + value.reference.catalog.manifest_sha256 = "f".repeat(64); + if (mutation === "project") value.reference.project_id = "f".repeat(32); + if (mutation === "storage") value.reference.storage_binding = "f".repeat(64); + if (mutation === "page") value.catalog_page = 1; + if (mutation === "entry") value.entry_index = 1; + if (mutation === "offset") value.offset = 1; + if (mutation === "extent") value.byte_count = 4; + if (mutation === "path") value.entry.path = "../private"; + expect(() => sourceFile(value, ref, 0, 0, entry, 0)).toThrow(); +}); +it.each([ + ".env", + ".git/config", + "a//b", + "a/../b", + "/absolute", + "keys/server.key", + "a\\b", +])("does not admit private or aliased catalog path %s", (path) => { + const ref = sourceRevisions(revisions(), reference.run_id).items[0].reference; + const value = catalog(); + value.catalog.entries[0] = { ...entry, path }; + expect(() => sourceCatalog(value, ref, 0)).toThrow(); +}); +it("rejects unverifiable bytes, invented text and a false binary flag", async () => { + const ref = sourceRevisions(revisions(), reference.run_id).items[0].reference; + for (const changes of [ + { content_base64: "ZmFr" }, + { text: "invented" }, + { binary: true }, + ]) { + await expect( + (async () => { + const value = sourceFile( + { ...file(), ...changes }, + ref, + 0, + 0, + entry, + 0, + ); + return verifySourceContent(value); + })(), + ).rejects.toThrow(); + } +}); +it("keeps pending/failed metadata unavailable and refuses unknown states or a foreign run", () => { + for (const status of ["pending", "failed"]) { + const value = revisions(); + value.items[0].status = status; + expect(sourceRevisions(value, reference.run_id).items[0].status).toBe( + status, + ); + } + const value = revisions(); + value.items[0].status = "succeeded"; + expect(() => sourceRevisions(value, reference.run_id)).toThrow(); + expect(() => sourceRevisions(revisions(), "f".repeat(32))).toThrow(); +}); + +it("aligns multibyte text with at most three verified boundary bytes and preserves exact offsets", async () => { + const ref = sourceRevisions(revisions(), reference.run_id).items[0].reference; + const data = new TextEncoder().encode("a€b"); + const fullHash = Array.from( + new Uint8Array(await crypto.subtle.digest("SHA-256", data)), + (value) => value.toString(16).padStart(2, "0"), + ).join(""); + const selected = { ...entry, byte_count: data.length, sha256: fullHash }; + async function window(offset: number, limit: number) { + const bytes = data.slice(offset, offset + limit); + let text: string | null; + try { + text = new TextDecoder("utf-8", { fatal: true, ignoreBOM: true }).decode( + bytes, + ); + } catch { + text = null; + } + const digest = Array.from( + new Uint8Array(await crypto.subtle.digest("SHA-256", bytes)), + (value) => value.toString(16).padStart(2, "0"), + ).join(""); + const page = sourceFile( + { + ...file(), + entry: selected, + offset, + byte_count: bytes.length, + next_offset: + offset + bytes.length < data.length ? offset + bytes.length : null, + sha256: digest, + content_base64: btoa(String.fromCharCode(...bytes)), + binary: text === null, + text, + }, + ref, + 0, + 0, + selected, + offset, + limit, + ); + return { page, ...(await verifySourceContent(page)) }; + } + const calls: number[][] = []; + const read = async (offset: number, limit: number) => { + calls.push([offset, limit]); + return window(offset, limit); + }; + const first = await alignedSourceText(await window(0, 2), read); + expect(first).toEqual({ text: "a€", start: 0, end: 4 }); + const second = await alignedSourceText(await window(2, 3), read); + expect(second).toEqual({ text: "€b", start: 1, end: 5 }); + expect(calls.every(([, length]) => length <= 3)).toBe(true); + expect((first.text ?? "") + (second.text ?? "")).not.toContain("�"); +}); diff --git a/studio/src/SourceExplorerData.ts b/studio/src/SourceExplorerData.ts new file mode 100644 index 000000000..c401875ac --- /dev/null +++ b/studio/src/SourceExplorerData.ts @@ -0,0 +1,525 @@ +/* Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 */ +import { ApiError } from "./api"; + +export type CatalogReference = { + version: 1; + artifact_id: string; + manifest_sha256: string; + source_digest: string; + manifest_digest: string; + byte_count: number; + file_count: number; + entry_count: number; + page_count: number; +}; +export type SourceReference = { + protocol: "forge-source-archive-v1"; + snapshot_id: string; + tenant_id: string; + project_id: string; + repository_id: string; + run_id: string; + attempt: number; + head: string; + tree: string; + branch: string | null; + catalog: CatalogReference; + storage_binding: string; +}; +export type SourceRevision = { + reference: SourceReference; + status: "published" | "pending" | "failed"; + reason: string | null; +}; +export type SourceRevisions = { + run_id: string; + items: SourceRevision[]; + next_after: string | null; + recording_enabled: boolean; + availability: "recorded" | "unavailable"; +}; +export type SourceEntry = { + path: string; + kind: "file" | "directory"; + mode: number; + byte_count: number; + sha256: string | null; +}; +export type SourceCatalog = { + reference: SourceReference; + catalog: { + version: 1; + artifact_id: string; + manifest_sha256: string; + page: number; + next_page: number | null; + entries: SourceEntry[]; + }; +}; +export type SourceFile = { + reference: SourceReference; + entry: SourceEntry; + catalog_page: number; + entry_index: number; + offset: number; + byte_count: number; + next_offset: number | null; + sha256: string; + content_base64: string; + binary: boolean; + text: string | null; +}; + +class SourceIntegrityError extends Error {} +type Row = Record; +function check(value: unknown): asserts value { + if (!value) + throw new SourceIntegrityError( + "The recorded source response could not be verified.", + ); +} +function row(value: unknown): Row { + check(value && typeof value === "object" && !Array.isArray(value)); + return value as Row; +} +function fields(value: Row, names: string) { + check( + Object.keys(value).sort().join(" ") === names.split(" ").sort().join(" "), + ); +} +function integer(value: unknown): number { + check(typeof value === "number" && Number.isSafeInteger(value) && value >= 0); + return value; +} +function hash(value: unknown): string { + check(typeof value === "string" && /^[a-f0-9]{64}$/.test(value)); + return value; +} +function id(value: unknown): string { + check(typeof value === "string" && /^[a-f0-9]{32}$/.test(value)); + return value; +} +function revision(value: unknown): string { + check( + typeof value === "string" && /^(?:[a-f0-9]{40}|[a-f0-9]{64})$/.test(value), + ); + return value; +} +function hasControls(value: string) { + return [...value].some( + (char) => char.charCodeAt(0) < 32 || char.charCodeAt(0) === 127, + ); +} +function catalogReference(value: unknown): CatalogReference { + const r = row(value); + fields( + r, + "version artifact_id manifest_sha256 source_digest manifest_digest byte_count file_count entry_count page_count", + ); + check(r.version === 1); + const count = integer(r.entry_count), + pages = integer(r.page_count), + files = integer(r.file_count); + check( + files <= count && (count === 0 ? pages === 0 : pages > 0 && pages <= count), + ); + return { + version: 1, + artifact_id: id(r.artifact_id), + manifest_sha256: hash(r.manifest_sha256), + source_digest: hash(r.source_digest), + manifest_digest: hash(r.manifest_digest), + byte_count: integer(r.byte_count), + file_count: files, + entry_count: count, + page_count: pages, + }; +} +export function sourceReference( + value: unknown, + runId: string, +): SourceReference { + const r = row(value); + fields( + r, + "protocol snapshot_id tenant_id project_id repository_id run_id attempt head tree branch catalog storage_binding", + ); + check(r.protocol === "forge-source-archive-v1" && r.run_id === id(runId)); + check( + r.branch === null || + (typeof r.branch === "string" && + r.branch.length <= 1024 && + !hasControls(r.branch)), + ); + return { + protocol: "forge-source-archive-v1", + snapshot_id: id(r.snapshot_id), + tenant_id: id(r.tenant_id), + project_id: id(r.project_id), + repository_id: id(r.repository_id), + run_id: id(r.run_id), + attempt: integer(r.attempt), + head: revision(r.head), + tree: revision(r.tree), + branch: r.branch as string | null, + catalog: catalogReference(r.catalog), + storage_binding: hash(r.storage_binding), + }; +} +function same(value: unknown, expected: SourceReference): SourceReference { + const ref = sourceReference(value, expected.run_id); + check(JSON.stringify(ref) === JSON.stringify(expected)); + return ref; +} +export function sourceRevisions( + value: unknown, + runId: string, + after: string | null = null, +): SourceRevisions { + const r = row(value); + fields(r, "run_id items next_after recording_enabled availability"); + check( + r.run_id === id(runId) && + typeof r.recording_enabled === "boolean" && + Array.isArray(r.items) && + r.items.length <= 128, + ); + let previous = after === null ? null : id(after); + const items = r.items.map((item): SourceRevision => { + const itemRow = row(item); + fields(itemRow, "reference status reason"); + check( + itemRow.status === "published" || + itemRow.status === "pending" || + itemRow.status === "failed", + ); + check( + itemRow.reason === null || + itemRow.reason === "source_publication_failed_reconciliation_required", + ); + const ref = sourceReference(itemRow.reference, runId); + check(previous === null || ref.snapshot_id > previous); + previous = ref.snapshot_id; + return { + reference: ref, + status: itemRow.status, + reason: itemRow.reason as string | null, + }; + }); + check(r.availability === (items.length ? "recorded" : "unavailable")); + const next = r.next_after === null ? null : id(r.next_after); + check(next === null || (items.length === 128 && next === previous)); + return { + run_id: runId, + items, + next_after: next, + recording_enabled: r.recording_enabled, + availability: r.availability as SourceRevisions["availability"], + }; +} +function entry(value: unknown): SourceEntry { + const r = row(value); + fields(r, "path kind mode byte_count sha256"); + check( + typeof r.path === "string" && + r.path.length > 0 && + !r.path.startsWith("/") && + !r.path.includes("\\") && + !hasControls(r.path), + ); + const parts = r.path.replace(/\/$/, "").split("/"); + const privateNames = new Set([ + ".git", + ".codex", + ".claude", + ".forge", + "node_modules", + ".venv", + "vendor", + ".env", + ".npmrc", + ".pypirc", + ".ssh", + "credentials.json", + "id_rsa", + "id_ed25519", + ]); + check( + parts.every( + (part) => + part && + part !== "." && + part !== ".." && + !privateNames.has(part.toLowerCase()) && + !/\.(?:pem|key)$/i.test(part) && + !( + part.toLowerCase().startsWith(".env.") && + ![".env.example", ".env.sample", ".env.template"].includes( + part.toLowerCase(), + ) + ), + ), + ); + const mode = integer(r.mode), + bytes = integer(r.byte_count); + check(mode <= 0o177777); + check(r.kind === "file" || r.kind === "directory"); + if (r.kind === "file") + check( + !r.path.endsWith("/") && + (mode & 0o170000) === 0o100000 && + typeof r.sha256 === "string", + ); + else + check( + r.path.endsWith("/") && + (mode & 0o170000) === 0o040000 && + r.sha256 === null && + bytes === 0, + ); + return { + path: r.path, + kind: r.kind, + mode, + byte_count: bytes, + sha256: r.sha256 === null ? null : hash(r.sha256), + }; +} +export function sourceCatalog( + value: unknown, + reference: SourceReference, + page: number, +): SourceCatalog { + const r = row(value); + fields(r, "reference catalog"); + const ref = same(r.reference, reference), + c = row(r.catalog); + fields(c, "version artifact_id manifest_sha256 page next_page entries"); + check( + c.version === 1 && + c.artifact_id === ref.catalog.artifact_id && + c.manifest_sha256 === ref.catalog.manifest_sha256 && + c.page === integer(page), + ); + check( + page < ref.catalog.page_count || + (page === 0 && ref.catalog.entry_count === 0), + ); + check( + Array.isArray(c.entries) && + c.entries.length <= 128 && + c.entries.length <= ref.catalog.entry_count, + ); + const entries = c.entries.map(entry); + check(entries.every((item, i) => i === 0 || entries[i - 1].path < item.path)); + const next = c.next_page === null ? null : integer(c.next_page); + check(next === (page + 1 < ref.catalog.page_count ? page + 1 : null)); + return { + reference: ref, + catalog: { + version: 1, + artifact_id: ref.catalog.artifact_id, + manifest_sha256: ref.catalog.manifest_sha256, + page, + next_page: next, + entries, + }, + }; +} +export function sourceFile( + value: unknown, + reference: SourceReference, + page: number, + entryIndex: number, + selected: SourceEntry, + offset: number, + limit = 32768, +): SourceFile { + const r = row(value); + fields( + r, + "reference entry catalog_page entry_index offset byte_count next_offset sha256 content_base64 binary text", + ); + const ref = same(r.reference, reference), + file = entry(r.entry); + check( + file.kind === "file" && JSON.stringify(file) === JSON.stringify(selected), + ); + check( + r.catalog_page === integer(page) && + r.entry_index === integer(entryIndex) && + r.offset === integer(offset), + ); + const count = integer(r.byte_count), + total = file.byte_count; + check( + count <= limit && + offset + count <= total && + (count > 0 || offset === total), + ); + const next = r.next_offset === null ? null : integer(r.next_offset); + check(next === (offset + count < total ? offset + count : null)); + check( + typeof r.content_base64 === "string" && + r.content_base64.length === 4 * Math.ceil(count / 3) && + /^[A-Za-z0-9+/]*={0,2}$/.test(r.content_base64), + ); + check( + typeof r.binary === "boolean" && + (r.text === null || typeof r.text === "string") && + r.binary === (r.text === null), + ); + return { + reference: ref, + entry: file, + catalog_page: page, + entry_index: entryIndex, + offset, + byte_count: count, + next_offset: next, + sha256: hash(r.sha256), + content_base64: r.content_base64, + binary: r.binary, + text: r.text as string | null, + }; +} +export async function verifySourceContent( + page: SourceFile, +): Promise<{ bytes: Uint8Array; text: string | null }> { + const bytes = Uint8Array.from(atob(page.content_base64), (char) => + char.charCodeAt(0), + ); + const digest = Array.from( + new Uint8Array(await crypto.subtle.digest("SHA-256", bytes)), + (value) => value.toString(16).padStart(2, "0"), + ).join(""); + check(bytes.length === page.byte_count && digest === page.sha256); + let text: string | null; + try { + text = new TextDecoder("utf-8", { fatal: true, ignoreBOM: true }).decode( + bytes, + ); + if (text.includes("\0")) text = null; + } catch { + text = null; + } + check(page.text === text && page.binary === (text === null)); + return { bytes, text }; +} +export type VerifiedSourceWindow = { + page: SourceFile; + bytes: Uint8Array; + text: string | null; +}; +export async function alignedSourceText( + window: VerifiedSourceWindow, + read: (offset: number, limit: number) => Promise, +): Promise<{ text: string | null; start: number; end: number }> { + const end = window.page.offset + window.bytes.length; + const raw = { text: window.text, start: window.page.offset, end }; + if (window.text !== null || !window.bytes.length || window.bytes.includes(0)) + return raw; + const length = (byte: number) => + byte >= 0xc2 && byte <= 0xdf + ? 2 + : byte >= 0xe0 && byte <= 0xef + ? 3 + : byte >= 0xf0 && byte <= 0xf4 + ? 4 + : 0; + async function boundary(offset: number, count: number) { + const value = await read(offset, count); + check( + value.page.offset === offset && + value.bytes.length === count && + JSON.stringify(value.page.reference) === + JSON.stringify(window.page.reference) && + JSON.stringify(value.page.entry) === + JSON.stringify(window.page.entry) && + value.page.catalog_page === window.page.catalog_page && + value.page.entry_index === window.page.entry_index, + ); + return value.bytes; + } + let prefix: Uint8Array = new Uint8Array(), + suffix: Uint8Array = new Uint8Array(); + if ((window.bytes[0] & 0xc0) === 0x80 && window.page.offset > 0) { + const count = Math.min(3, window.page.offset), + before = await boundary(window.page.offset - count, count); + for (let i = before.length - 1; i >= 0; i--) { + if (length(before[i]) > before.length - i) { + prefix = before.slice(i); + break; + } + } + if (!prefix.length) return raw; + } + let last = window.bytes.length - 1; + while (last >= 0 && (window.bytes[last] & 0xc0) === 0x80) last--; + const needed = + last >= 0 + ? Math.max(0, length(window.bytes[last]) - (window.bytes.length - last)) + : 0; + if (needed && end < window.page.entry.byte_count) + suffix = await boundary( + end, + Math.min(needed, window.page.entry.byte_count - end), + ); + if (!prefix.length && !suffix.length) return raw; + const bytes = new Uint8Array( + prefix.length + window.bytes.length + suffix.length, + ); + bytes.set(prefix); + bytes.set(window.bytes, prefix.length); + bytes.set(suffix, prefix.length + window.bytes.length); + try { + const text = new TextDecoder("utf-8", { + fatal: true, + ignoreBOM: true, + }).decode(bytes); + if (!text.includes("\0")) + return { + text, + start: window.page.offset - prefix.length, + end: end + suffix.length, + }; + } catch { + /* Preserve undecodable data as exact bytes. */ + } + return raw; +} +export function sourceProblem(error: unknown): { + kind: string; + title: string; + message: string; +} { + if (error instanceof ApiError && [401, 403].includes(error.status)) + return { + kind: "revoked", + title: "Source access is unavailable", + message: + "Sign in or check your current project access, then refresh. Previous source content has been cleared.", + }; + if ( + error instanceof SourceIntegrityError || + (error instanceof ApiError && error.status === 409) + ) + return { + kind: "review_required", + title: "Snapshot needs review", + message: + "The recorded source could not be verified with current access and configuration. Refresh its records before continuing.", + }; + if (error instanceof ApiError && error.status === 404) + return { + kind: "unavailable", + title: "Recorded source is unavailable", + message: "Check the selected run and server, then refresh its records.", + }; + return { + kind: "error", + title: "Source records could not be loaded", + message: + "Check the connection and refresh. No unverified source is displayed.", + }; +} diff --git a/studio/tests/source-explorer-server.mjs b/studio/tests/source-explorer-server.mjs new file mode 100644 index 000000000..d65907806 --- /dev/null +++ b/studio/tests/source-explorer-server.mjs @@ -0,0 +1,65 @@ +/* Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 */ +import { mkdirSync, writeFileSync } from "node:fs"; +import { resolve } from "node:path"; +import process from "node:process"; +import { createServer } from "vite"; + +const studio = process.cwd(); +const output = resolve( + process.env.FORGE_SOURCE_FIXTURE_OUTPUT || + "test-results/source-explorer-fixture", +); +mkdirSync(output, { recursive: true }); +const entry = resolve(output, "source-fixture.tsx"); +writeFileSync( + entry, + `/* Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 */ +import { useEffect, useState } from "react"; +import { createRoot } from "react-dom/client"; +import { SourceExplorer } from ${JSON.stringify(resolve(studio, "src/SourceExplorer.tsx"))}; +import ${JSON.stringify(resolve(studio, "src/styles.css"))}; +document.documentElement.dataset.theme=new URLSearchParams(location.search).get("theme")||"light"; +function Fixture(){ + const [run,setRun]=useState("a".repeat(32)); + const [token,setToken]=useState("controlled-source-viewer-token"); + useEffect(()=>{const change=(event)=>{if(typeof event.detail?.runId==="string")setRun(event.detail.runId);if(typeof event.detail?.token==="string")setToken(event.detail.token);}; + window.addEventListener("source-fixture",change);return()=>window.removeEventListener("source-fixture",change);},[]); + return

Source archive

Controlled browser fixture · no live project

+ +
Fixture-only readback · no source or execution changes
; +} +createRoot(document.getElementById("root")).render(); +`, +); +const html = ` +Controlled source explorer fixture +
`; +const server = await createServer({ + root: studio, + configFile: resolve(studio, "vite.config.ts"), + resolve: { dedupe: ["react", "react-dom"] }, + server: { + host: "127.0.0.1", + port: 5197, + strictPort: true, + fs: { allow: [resolve(studio, ".."), output] }, + }, + plugins: [ + { + name: "controlled-source-fixture", + configureServer(current) { + current.middlewares.use(async (req, res, next) => { + if (!req.url?.startsWith("/__source_explorer__")) return next(); + res.setHeader("Content-Type", "text/html"); + res.end(await current.transformIndexHtml(req.url, html)); + }); + }, + }, + ], +}); +await server.listen(); +for (const signal of ["SIGINT", "SIGTERM"]) + process.once(signal, async () => { + await server.close(); + process.exit(0); + }); diff --git a/studio/tests/source-explorer.spec.ts b/studio/tests/source-explorer.spec.ts new file mode 100644 index 000000000..196875b3b --- /dev/null +++ b/studio/tests/source-explorer.spec.ts @@ -0,0 +1,359 @@ +/* Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 */ +import { createHash } from "node:crypto"; +import { expect, test, type Page } from "@playwright/test"; + +const hash = (data: Buffer) => createHash("sha256").update(data).digest("hex"); +const longPath = + "backend/inventory/reconciliation/domain/operations/retained_reservations.ts"; +const opening = `// Controlled archived source fixture\nexport type Reservation = { sku: string; quantity: number };\nexport function retainStock(input: Reservation) {\n const html = "";\n // \`\`\` This is source text, never a Markdown fence.\n return { ...input, retained: true, html };\n}\n`; +let prefix = Buffer.from(opening); +const comment = Buffer.from( + "// Controlled pagination marker for recorded source.\n", +); +while (prefix.length + comment.length + 3 < 32767) + prefix = Buffer.concat([prefix, comment]); +prefix = Buffer.concat([ + prefix, + Buffer.from("// " + " ".repeat(32767 - prefix.length - 3)), +]); +const source = Buffer.concat([ + prefix, + Buffer.from("€\nexport const retainedAfterBoundary = true;\n".repeat(900)), +]); +const binary = Buffer.from([255, 0, 128, 1]); +const frontend = Buffer.from('export const columns = ["sku", "available"];\n'); +function reference(runId = "a".repeat(32), second = false) { + return { + protocol: "forge-source-archive-v1", + snapshot_id: (second ? "c" : "b").repeat(32), + tenant_id: "1".repeat(32), + project_id: "2".repeat(32), + repository_id: "3".repeat(32), + run_id: runId, + attempt: 3, + head: "4".repeat(40), + tree: "5".repeat(40), + branch: second ? "release/retained-ui" : "forge/source-review", + storage_binding: "9".repeat(64), + catalog: { + version: 1, + artifact_id: (second ? "6" : "7").repeat(32), + manifest_sha256: (second ? "6" : "8").repeat(64), + source_digest: "e".repeat(64), + manifest_digest: "f".repeat(64), + byte_count: source.length + binary.length + frontend.length, + file_count: 3, + entry_count: 5, + page_count: 2, + }, + }; +} +const file = (path: string, data: Buffer) => ({ + path, + kind: "file", + mode: 33188, + byte_count: data.length, + sha256: hash(data), +}); +const directory = (path: string) => ({ + path, + kind: "directory", + mode: 16877, + byte_count: 0, + sha256: null, +}); +type Control = { + mode: string; + calls: string[]; + writes: number; + external: number; + delay: boolean; +}; +async function intercept(page: Page): Promise { + const control: Control = { + mode: "published", + calls: [], + writes: 0, + external: 0, + delay: false, + }; + await page.route("**/*", async (route) => { + const req = route.request(), + url = new URL(req.url()); + if (url.hostname !== "127.0.0.1" || req.method() !== "GET") { + if (req.method() !== "GET") control.writes++; + else control.external++; + return route.abort(); + } + if (!url.pathname.startsWith("/api/v1/")) return route.continue(); + control.calls.push(url.pathname + url.search); + if (control.mode === "revoked") + return route.fulfill({ + status: 403, + json: { detail: "Controlled current scope revoked" }, + }); + const runId = url.pathname.split("/")[4]; + const second = url.pathname.includes("/" + "c".repeat(32) + "/"); + const ref = reference(runId, second); + if (url.pathname.endsWith("/source/revisions")) { + const items = + control.mode === "empty" + ? [] + : [ + { + reference: reference(runId), + status: ["pending", "failed"].includes(control.mode) + ? control.mode + : "published", + reason: + control.mode === "failed" + ? "source_publication_failed_reconciliation_required" + : null, + }, + ...(["pending", "failed"].includes(control.mode) + ? [] + : [ + { + reference: reference(runId, true), + status: "published", + reason: null, + }, + ]), + ]; + return route.fulfill({ + json: { + run_id: runId, + items, + next_after: null, + recording_enabled: true, + availability: items.length ? "recorded" : "unavailable", + }, + }); + } + expect(url.searchParams.get("expected_manifest")).toBe( + ref.catalog.manifest_sha256, + ); + const n = Number(url.searchParams.get("page")); + const entries = + n === 0 + ? [ + directory("backend/"), + file(longPath, source), + file("backend/review/retained-byte-evidence.bin", binary), + ] + : [ + directory("frontend/"), + file("frontend/views/StockLedger.tsx", frontend), + ]; + if (url.pathname.endsWith("/catalog")) + return route.fulfill({ + json: { + reference: ref, + catalog: { + version: 1, + artifact_id: ref.catalog.artifact_id, + manifest_sha256: ref.catalog.manifest_sha256, + page: n, + next_page: n === 0 ? 1 : null, + entries, + }, + }, + }); + const index = Number(url.pathname.split("/").at(-1)), + offset = Number(url.searchParams.get("offset")), + limit = Number(url.searchParams.get("limit")); + expect(limit === 32768 || (limit >= 1 && limit <= 3)).toBe(true); + if (control.delay && runId === "a".repeat(32)) + await new Promise((resolve) => setTimeout(resolve, 500)); + const data = n === 1 ? frontend : index === 2 ? binary : source; + const content = data.subarray(offset, offset + limit); + let text: string | null; + try { + text = new TextDecoder("utf-8", { fatal: true, ignoreBOM: true }).decode( + content, + ); + if (text.includes("\0")) text = null; + } catch { + text = null; + } + const json = { + reference: ref, + entry: entries[index], + catalog_page: n, + entry_index: index, + offset, + byte_count: content.length, + next_offset: + offset + content.length < data.length ? offset + content.length : null, + sha256: control.mode === "corrupt" ? "0".repeat(64) : hash(content), + content_base64: content.toString("base64"), + binary: text === null, + text, + }; + try { + await route.fulfill({ json }); + } catch { + /* A scoped view change can abort this controlled response. */ + } + }); + return control; +} +for (const [width, theme] of [ + [1440, "light"], + [1440, "dark"], + [390, "light"], + [390, "dark"], +] as const) { + test(`controlled source explorer ${width} ${theme}`, async ({ + page, + }, info) => { + await page.setViewportSize({ width, height: 900 }); + await page.emulateMedia({ colorScheme: theme, reducedMotion: "reduce" }); + const errors: string[] = []; + page.on("pageerror", (error) => errors.push(error.message)); + const control = await intercept(page); + await page.goto(`/__source_explorer__?theme=${theme}`); + const panel = page.getByRole("region", { + name: "Recorded source snapshot", + exact: true, + }); + await expect( + panel.getByText("Window checksum verified", { exact: true }), + ).toBeVisible(); + await expect( + panel.getByText(/Text preview aligned to UTF-8/), + ).toBeVisible(); + await expect(panel.getByLabel("Recorded source code")).toContainText( + "retainStock", + ); + await expect(panel.getByLabel("Recorded source code")).toContainText("€"); + expect(await page.locator("img").count()).toBe(0); + expect( + await page.evaluate(() => Reflect.get(window, "__SOURCE_EXECUTED__")), + ).toBeUndefined(); + await page.evaluate(() => document.fonts.ready); + const geometry = await panel.evaluate((element) => ({ + overflow: document.documentElement.scrollWidth > innerWidth + 1, + font: getComputedStyle(element).fontFamily, + targets: [...element.querySelectorAll("button,select,summary")] + .filter((x) => x.getClientRects().length > 0) + .map((x) => x.getBoundingClientRect().height), + footer: + document + .querySelector(".source-fixture-footer")! + .getBoundingClientRect().top >= + element.getBoundingClientRect().bottom, + })); + expect(geometry.overflow).toBe(false); + expect(geometry.font).toContain("Manrope"); + expect(geometry.targets.every((n) => n >= 44)).toBe(true); + expect(geometry.footer).toBe(true); + await page.screenshot({ + path: info.outputPath(`source-${width}-${theme}.png`), + fullPage: false, + }); + await panel + .getByRole("button", { name: "Next window", exact: true }) + .click(); + await expect(panel).toContainText("Bytes 32,769"); + await expect(panel.getByLabel("Recorded source code")).not.toContainText( + "�", + ); + if (width === 390) + await panel + .getByRole("button", { name: "Browse files", exact: true }) + .click(); + await panel + .getByRole("button", { name: /backend\/review\/retained-byte-evidence/ }) + .focus(); + await page.keyboard.press("Enter"); + await expect(panel.getByLabel("Exact source bytes")).toHaveText( + "ff 00 80 01", + ); + await expect( + panel.getByRole("region", { name: "Recorded file content", exact: true }), + ).toBeFocused(); + if (width === 390) + await panel + .getByRole("button", { name: "Browse files", exact: true }) + .click(); + await panel + .getByRole("button", { name: "Next files", exact: true }) + .click(); + await expect(panel.getByLabel("Recorded source code")).toContainText( + "columns", + ); + await panel + .getByRole("combobox", { name: "Recorded branch", exact: true }) + .selectOption("release/retained-ui"); + await expect( + panel.getByText("Window checksum verified", { exact: true }), + ).toBeVisible(); + control.mode = "corrupt"; + await panel + .getByRole("button", { name: "Next window", exact: true }) + .click(); + await expect( + panel.getByText("Snapshot needs review", { exact: true }), + ).toBeVisible(); + await expect(panel.getByLabel("Recorded source code")).toHaveCount(0); + control.mode = "pending"; + await panel + .getByRole("button", { name: "Refresh records", exact: true }) + .click(); + await expect( + panel.getByText("Publication is incomplete", { exact: true }), + ).toBeVisible(); + control.mode = "empty"; + await panel + .getByRole("button", { name: "Refresh records", exact: true }) + .click(); + await expect( + panel.getByText("No recorded source", { exact: true }), + ).toBeVisible(); + control.mode = "published"; + await panel + .getByRole("button", { name: "Refresh records", exact: true }) + .click(); + await expect( + panel.getByText("Window checksum verified", { exact: true }), + ).toBeVisible(); + control.mode = "revoked"; + await panel + .getByRole("button", { name: "Next window", exact: true }) + .click(); + await expect( + panel.getByText("Source access is unavailable", { exact: true }), + ).toBeVisible(); + await expect(panel.getByLabel("Recorded source code")).toHaveCount(0); + expect(control.writes).toBe(0); + expect(control.external).toBe(0); + expect(errors).toEqual([]); + }); +} +test("a late response from the previous run cannot cross a scoped view switch", async ({ + page, +}) => { + const control = await intercept(page); + control.delay = true; + await page.goto("/__source_explorer__?theme=light"); + await expect( + page.getByText("Verifying source window…", { exact: true }), + ).toBeVisible(); + await page.evaluate(() => + window.dispatchEvent( + new CustomEvent("source-fixture", { + detail: { runId: "f".repeat(32), token: "other-scoped-fixture-token" }, + }), + ), + ); + await expect( + page.getByText("Window checksum verified", { exact: true }), + ).toBeVisible(); + await page.getByText("View snapshot provenance", { exact: true }).click(); + await expect( + page.getByRole("region", { name: "Recorded source snapshot" }), + ).not.toContainText("a".repeat(32)); + expect(control.calls.some((url) => url.includes("f".repeat(32)))).toBe(true); + expect(control.writes).toBe(0); +}); diff --git a/tests/test_source_archive_authority_host.py b/tests/test_source_archive_authority_host.py new file mode 100644 index 000000000..b5ebaf55d --- /dev/null +++ b/tests/test_source_archive_authority_host.py @@ -0,0 +1,141 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""The last asynchronous workspace check cannot carry a stale actor or owner.""" + +import copy +from contextlib import asynccontextmanager +from datetime import UTC, datetime +from types import SimpleNamespace +from uuid import UUID + +import pytest + +from firefly_forge.object_storage import ObjectStorageSettings +from firefly_forge.project_access import CURRENT_PRINCIPAL, AccessDenied +from firefly_forge.source_archive_service import SourceArchiveManager, SourceArchivePolicy + + +@pytest.mark.parametrize("drift", ["actor", "scope", "owner"]) +async def test_last_async_workspace_check_cannot_admit_revoked_authority(tmp_path, monkeypatch, drift): + import firefly_forge.source_archive_service as module + + state = SimpleNamespace(valid=True, checks=0, effects=[]) + owner = object() + run = dict( + id=UUID(int=1).hex, + project_id=UUID(int=2).hex, + repository_id=UUID(int=3).hex, + attempt=1, + execution_actor={"user_id": UUID(int=4).hex}, + branch="recorded/topic", + ) + + async def check(): + return None + + async def get(_): + return copy.deepcopy(run) + + async def campaign(_): + return None + + async def scalar(*args, **kwargs): + return datetime(2026, 10, 30, tzinfo=UTC) + + @asynccontextmanager + async def unit(): + yield SimpleNamespace(scalar=scalar) + + store = SimpleNamespace( + assert_ownership=check, + get_run=get, + assert_campaign_current=campaign, + _unit=unit, + execution_ownership_token=owner, + _ownership_pid=123, + schema="controlled_source", + ) + + class Access: + def __init__(self, store): + pass + + async def authorize_execution(self, value): + if not state.valid: + raise AccessDenied("canonical actor revoked") + + monkeypatch.setattr(module, "ProjectAccess", Access) + + async def workspace_current(): + state.checks += 1 + if state.checks == 2: + if drift == "actor": + state.valid = False + elif drift == "scope": + run["project_id"] = UUID(int=9).hex + else: + store.execution_ownership_token = object() + + async def publish(*args, authorize, **kwargs): + await authorize() + state.effects.append("would write object") + + manager = object.__new__(SourceArchiveManager) + manager.store = store + manager.settings = SimpleNamespace(provider="local") + manager.policy = SourceArchivePolicy() + manager.publisher = SimpleNamespace(publish=publish) + assert CURRENT_PRINCIPAL.get() is None + with pytest.raises(AccessDenied): + await manager.publish(copy.deepcopy(run), tmp_path, "a" * 40, "b" * 40, assert_current=workspace_current) + assert state.effects == [] + + +def test_native_configuration_defaults_to_recording_and_requires_explicit_safe_retention(): + assert SourceArchivePolicy.from_environment({}).enabled is True + policy = SourceArchivePolicy.from_environment( + {"FORGE_SOURCE_ARCHIVE_POLICY": '{"excluded_projects":["' + UUID(int=2).hex + '"]}'} + ) + assert policy.excluded_projects == frozenset({UUID(int=2)}) + with pytest.raises(ValueError): + SourceArchivePolicy(retention="delete_after_seven_days") + + +async def test_fresh_configured_source_index_serializes_ddl_before_native_creation(tmp_path, monkeypatch): + import firefly_forge.source_archive_service as module + + calls = [] + + async def execute(statement, parameters): + calls.append(("sql", str(statement), parameters)) + + async def sync(function): + return function(object()) + + async def connection(): + return SimpleNamespace(run_sync=sync) + + @asynccontextmanager + async def unit(): + yield SimpleNamespace(execute=execute, connection=connection) + + class Pools: + def __init__(self, *args, **kwargs): + self._unit = unit + + async def initialize(self): + return None + + monkeypatch.setattr(module, "RunnerPools", Pools) + monkeypatch.setattr( + module.ForgeSourceArchiveEntity.metadata, "create_all", lambda *args, **kwargs: calls.append(("ddl",)) + ) + manager = object.__new__(SourceArchiveManager) + manager.settings = ObjectStorageSettings( + provider="local", mode="test", local_directory=tmp_path / "objects", storage_id=UUID(int=8) + ) + manager.tenant = UUID(int=1) + manager.store = SimpleNamespace( + path=tmp_path / "factory.db", schema="controlled_source", session_provider=object(), credential_registry=dict + ) + await manager.start() + assert calls[0][0] == "sql" and "pg_advisory_xact_lock" in calls[0][1] and calls[1] == ("ddl",) diff --git a/tests/test_source_archive_host.py b/tests/test_source_archive_host.py new file mode 100644 index 000000000..078f73123 --- /dev/null +++ b/tests/test_source_archive_host.py @@ -0,0 +1,187 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Published source content survives workspace loss without granting a live read.""" + +import hashlib +import os +from uuid import UUID + +import pytest +from test_runner_object_artifacts_host import Ledger, allowed, authority_fence + +from firefly_forge.object_storage import ArtifactObjectStorage, ObjectStorageSettings, native_document_storage +from firefly_forge.source_archive import SourceArchiveContent +from firefly_forge.source_catalog import SourceCatalogBudget, read_source_catalog_page +from firefly_forge.verification_source_preparation import prepare_verification_source + + +async def test_catalog_file_is_uploaded_and_read_from_scoped_objects_after_workspace_loss(tmp_path): + workspace = tmp_path / "source" + workspace.mkdir(mode=0o700) + source = workspace / "example.py" + source.write_bytes(b"print('retained')\n" * 10000) + source.chmod(0o644) + catalog = tmp_path / "catalog" + catalog.mkdir(mode=0o700) + prepared = await prepare_verification_source( + workspace, catalog, budget=SourceCatalogBudget(min_free_bytes=0), assert_current=allowed + ) + entry = read_source_catalog_page(catalog, prepared.reference, 0).entries[0] + settings = ObjectStorageSettings( + provider="local", mode="test", local_directory=tmp_path / "objects", storage_id=UUID(int=8), chunk_bytes=4096 + ) + scratch = tmp_path / "scratch" + scratch.mkdir(mode=0o700) + content = SourceArchiveContent( + UUID(int=1), ArtifactObjectStorage(settings, native_document_storage(settings)), Ledger(), scratch + ) + binding = dict(snapshot_id="a" * 32, project_id=UUID(int=2).hex, source_digest=prepared.source_digest) + descriptor = await content.publish(binding, prepared, entry, authority_fence(), authorize=allowed) + source.unlink() + restored = await content.window(binding, entry, descriptor, 2, 31, authorize=allowed) + assert restored == (b"print('retained')\n" * 10000)[2:33] + assert descriptor["receipt"]["sha256"] == hashlib.sha256(b"print('retained')\n" * 10000).hexdigest() + assert not list(scratch.iterdir()) + + +@pytest.mark.parametrize( + "path", ["../source", "/source", "a//b", "a/./b", "a\\b", ".env", "a/settings.key", ".ssh/id_rsa"] +) +async def test_public_source_entry_never_becomes_a_path_capability(tmp_path, path): + import stat + + from firefly_forge.source_archive import public_entry + from firefly_forge.source_catalog import SourceCatalogEntryV1 + + entry = SourceCatalogEntryV1(path=path, kind="file", mode=stat.S_IFREG | 0o644, byte_count=1, sha256="a" * 64) + with pytest.raises(ValueError): + public_entry(entry) + + +@pytest.mark.parametrize("kind", ["symlink", "hardlink", "secret"]) +async def test_capture_refuses_aliases_and_known_secret_without_leaking_source_descriptors(tmp_path, kind, monkeypatch): + workspace = tmp_path / "source" + workspace.mkdir(mode=0o700) + source = workspace / "file.txt" + source.write_bytes(b"controlled-private-source-credential-938420") + catalogs = tmp_path / "catalogs" + catalogs.mkdir(mode=0o700) + prepared = await prepare_verification_source( + workspace, catalogs, budget=SourceCatalogBudget(min_free_bytes=0), assert_current=allowed + ) + entry = read_source_catalog_page(catalogs, prepared.reference, 0).entries[0] + settings = ObjectStorageSettings( + provider="local", mode="test", local_directory=tmp_path / "objects", storage_id=UUID(int=8), chunk_bytes=4096 + ) + scratch = tmp_path / "scratch" + scratch.mkdir(mode=0o700) + ledger = Ledger() + content = SourceArchiveContent( + UUID(int=1), ArtifactObjectStorage(settings, native_document_storage(settings)), ledger, scratch + ) + if kind == "secret": + content.helper.credential_registry = lambda: {"CONTROLLED_TOKEN": "controlled-private-source-credential-938420"} + else: + target = tmp_path / "target" + source.rename(target) + if kind == "symlink": + source.symlink_to(target) + else: + os.link(target, source) + opened, closed = [], [] + native_open, native_close = os.open, os.close + + def opening(*args, **kwargs): + result = native_open(*args, **kwargs) + if args[0] == "file.txt": + opened.append(result) + return result + + def closing(fd): + closed.append(fd) + return native_close(fd) + + monkeypatch.setattr(os, "open", opening) + monkeypatch.setattr(os, "close", closing) + with pytest.raises((ValueError, OSError)): + await content.publish( + dict(snapshot_id="a" * 32, project_id=UUID(int=2).hex, source_digest=prepared.source_digest), + prepared, + entry, + authority_fence(), + authorize=allowed, + ) + assert ledger.rows == {} and not list(scratch.iterdir()) + assert all(fd in closed for fd in opened), "Source descriptors must close before a refused read returns" + + +async def test_window_fetches_only_intersecting_chunk_bytes_and_never_restores_the_file(tmp_path, monkeypatch): + workspace = tmp_path / "source" + workspace.mkdir(mode=0o700) + data = b"0123456789abcdef" * 50000 + (workspace / "large.txt").write_bytes(data) + catalogs = tmp_path / "catalogs" + catalogs.mkdir(mode=0o700) + prepared = await prepare_verification_source( + workspace, catalogs, budget=SourceCatalogBudget(min_free_bytes=0), assert_current=allowed + ) + entry = read_source_catalog_page(catalogs, prepared.reference, 0).entries[0] + settings = ObjectStorageSettings( + provider="local", mode="test", local_directory=tmp_path / "objects", storage_id=UUID(int=8), chunk_bytes=4096 + ) + scratch = tmp_path / "scratch" + scratch.mkdir(mode=0o700) + content = SourceArchiveContent( + UUID(int=1), ArtifactObjectStorage(settings, native_document_storage(settings)), Ledger(), scratch + ) + binding = dict(snapshot_id="a" * 32, project_id=UUID(int=2).hex, source_digest=prepared.source_digest) + descriptor = await content.publish(binding, prepared, entry, authority_fence(), authorize=allowed) + reads = [] + native_read = content.helper.objects.read_chunk + + async def read(scope, chunk, **kwargs): + reads.append(chunk) + return await native_read(scope, chunk, **kwargs) + + def no_spool(*args, **kwargs): + raise AssertionError("Window reads must not restore the full file") + + monkeypatch.setattr(content.helper.objects, "read_chunk", read) + monkeypatch.setattr(content.helper, "_spooled", no_spool) + assert await content.window(binding, entry, descriptor, 180000, 31, authorize=allowed) == data[180000:180031] + assert len(reads) <= 3 and not list(scratch.iterdir()) + + +async def test_changed_configured_credentials_hold_published_content_for_fresh_capture(tmp_path): + from test_source_archive_publisher_host import fixture + + publisher, workspace, run, _ = fixture(tmp_path) + ref = await publisher.publish(run, workspace, "a" * 40, "b" * 40, authority_fence(), authorize=allowed) + page, descriptors = await publisher.index.page(ref, 0) + publisher.content.helper.credential_registry = lambda: {"NEW_TOKEN": "new-configured-source-credential-72145"} + with pytest.raises(ValueError, match="credentials"): + await publisher.content.window(ref.content_binding(), page.entries[0], descriptors[0], 0, 3, authorize=allowed) + + +async def test_reader_specific_taints_are_scanned_without_invalidating_other_readers(tmp_path): + from test_source_archive_publisher_host import fixture + + from firefly_forge.secrets import ephemeral_credentials + + publisher, workspace, run, _ = fixture(tmp_path) + credential = "new-reader-session-private-credential-99591" + (workspace / "app.py").write_text(credential) + with ephemeral_credentials({"ACCESS_TOKEN": "first-reader-private-credential-81231"}): + ref = await publisher.publish(run, workspace, "a" * 40, "b" * 40, authority_fence(), authorize=allowed) + page0, descriptors0 = await publisher.index.page(ref, 0) + page1, descriptors1 = await publisher.index.page(ref, 1) + with ephemeral_credentials({"ACCESS_TOKEN": credential}): + assert ( + await publisher.content.window( + ref.content_binding(), page1.entries[0], descriptors1[0], 0, 3, authorize=allowed + ) + == b"\x00\xffr" + ) + with pytest.raises(ValueError, match="credentials"): + await publisher.content.window( + ref.content_binding(), page0.entries[0], descriptors0[0], 0, 3, authorize=allowed + ) diff --git a/tests/test_source_archive_index_host.py b/tests/test_source_archive_index_host.py new file mode 100644 index 000000000..0d4ecffd2 --- /dev/null +++ b/tests/test_source_archive_index_host.py @@ -0,0 +1,90 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Publication roots never contain an unbounded project file array.""" + +import copy +from types import SimpleNamespace +from uuid import UUID + +import pytest +from test_runner_object_artifacts_host import Pools, authority_fence + +from firefly_forge.source_archive_index import SourceArchiveIndex, SourceArchiveRefV1 +from firefly_forge.source_catalog import SourceCatalogRefV1 + + +class Records: + def __init__(self): + self.rows, self.parts = {}, {} + + async def locked(self, value): + return self.rows.get(str(value)) + + async def at(self, value, page): + return self.parts.get((value.hex, page)) + + async def save(self, row): + if hasattr(row, "snapshot_id"): + self.parts[(row.snapshot_id.hex, row.page)] = row + else: + self.rows[row.id.hex] = row + + def _require_session(self): + async def flush(): + return None + + return SimpleNamespace(flush=flush) + + +def reference(): + catalog = SourceCatalogRefV1( + artifact_id="b" * 32, + manifest_sha256="c" * 64, + source_digest="d" * 64, + manifest_digest="e" * 64, + byte_count=0, + file_count=0, + entry_count=0, + page_count=0, + ) + return SourceArchiveRefV1( + snapshot_id="a" * 32, + tenant_id=UUID(int=1).hex, + project_id=UUID(int=2).hex, + repository_id=UUID(int=3).hex, + run_id=UUID(int=4).hex, + attempt=1, + head="f" * 40, + tree="1" * 40, + branch="recorded/topic", + catalog=catalog, + storage_binding="2" * 64, + ) + + +async def test_empty_snapshot_publishes_with_scalar_root_and_refuses_ambient_transaction(): + records = Records() + index = SourceArchiveIndex(Pools(records), records, records) + ref = reference() + assert await index.begin(ref, authority_fence()) is None + await index.seal(ref, authority_fence(), lambda: None) + assert await index.get(ref.snapshot_id) == ref + assert "pages" not in records.rows[ref.snapshot_id].data + index.pools.sessions.current = lambda: object() + with pytest.raises(RuntimeError, match="independent"): + await index.get(ref.snapshot_id) + + +async def test_pending_or_failed_snapshot_never_looks_published_or_restarts_write(): + records = Records() + index = SourceArchiveIndex(Pools(records), records, records) + ref = reference() + await index.begin(ref, authority_fence()) + before = copy.deepcopy(records.rows) + with pytest.raises(ValueError, match="unavailable"): + await index.get(ref.snapshot_id) + with pytest.raises(RuntimeError, match="reconciliation"): + await index.begin(ref, authority_fence()) + assert records.rows[ref.snapshot_id].data == before[ref.snapshot_id].data + await index.fail(ref.snapshot_id, authority_fence()) + with pytest.raises(ValueError, match="unavailable"): + await index.get(ref.snapshot_id) diff --git a/tests/test_source_archive_publisher_host.py b/tests/test_source_archive_publisher_host.py new file mode 100644 index 000000000..3e3fb09ae --- /dev/null +++ b/tests/test_source_archive_publisher_host.py @@ -0,0 +1,98 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Snapshot publication is complete, immutable and explicitly fails closed.""" + +from uuid import UUID + +import pytest +from test_runner_object_artifacts_host import Ledger, Pools, allowed, authority_fence +from test_source_archive_index_host import Records + +from firefly_forge.object_storage import ArtifactObjectStorage, ObjectStorageSettings, native_document_storage +from firefly_forge.runner_pool_models import pool_digest +from firefly_forge.source_archive import SourceArchiveContent +from firefly_forge.source_archive_index import SourceArchiveIndex +from firefly_forge.source_archive_service import SourceArchivePublisher +from firefly_forge.source_catalog import SourceCatalogBudget + + +def fixture(tmp_path): + workspace = tmp_path / "workspace" + workspace.mkdir(mode=0o700) + (workspace / "app.py").write_text("print(1)\n") + (workspace / "data.bin").write_bytes(b"\x00\xffrecorded") + (workspace / ".env").write_text("private") + scratch = tmp_path / "scratch" + scratch.mkdir(mode=0o700) + catalogs = tmp_path / "catalogs" + catalogs.mkdir(mode=0o700) + settings = ObjectStorageSettings( + provider="local", mode="test", local_directory=tmp_path / "objects", storage_id=UUID(int=8), chunk_bytes=4096 + ) + content = SourceArchiveContent( + UUID(int=1), ArtifactObjectStorage(settings, native_document_storage(settings)), Ledger(), scratch + ) + records = Records() + index = SourceArchiveIndex(Pools(records), records, records) + publisher = SourceArchivePublisher(content, index, catalogs, SourceCatalogBudget(min_free_bytes=0, page_entries=1)) + run = dict( + id=UUID(int=4).hex, + project_id=UUID(int=2).hex, + repository_id=UUID(int=3).hex, + attempt=1, + branch="recorded/topic", + ) + return publisher, workspace, run, records + + +async def test_publish_copies_bounded_catalog_pages_and_file_bytes_with_no_private_entries(tmp_path): + publisher, workspace, run, records = fixture(tmp_path) + ref = await publisher.publish(run, workspace, "a" * 40, "b" * 40, authority_fence(), authorize=allowed) + assert ref.catalog.file_count == 2 and ref.catalog.page_count == 2 + pages = [await publisher.index.page(ref, n) for n in range(2)] + assert [page.entries[0].path for page, _ in pages] == ["app.py", "data.bin"] + assert len(records.parts) == 2 and all(len(page.entries) == 1 for page, _ in pages) + assert records.rows[ref.snapshot_id].data["status"] == "published" + assert all("chunks" not in value.data for value in records.rows.values()) + + +async def test_changed_selected_file_during_final_publication_is_recorded_failed(tmp_path): + publisher, workspace, run, records = fixture(tmp_path) + save = records.save + + async def changed(row): + await save(row) + if getattr(row, "data", {}).get("status") == "published": + (workspace / "app.py").write_text("changed\n") + + records.save = changed + with pytest.raises(ValueError, match="changed"): + await publisher.publish(run, workspace, "a" * 40, "b" * 40, authority_fence(), authorize=allowed) + assert len(records.rows) == 1 + assert next(iter(records.rows.values())).data["status"] == "failed" + + +async def test_recomputed_page_chain_cannot_relabel_the_catalog_source_digest(tmp_path): + publisher, workspace, run, records = fixture(tmp_path) + ref = await publisher.publish(run, workspace, "a" * 40, "b" * 40, authority_fence(), authorize=allowed) + head = "0" * 64 + for number in range(ref.catalog.page_count): + row = records.parts[(ref.snapshot_id, number)] + if number == 0: + row.data["catalog"]["entries"][0]["path"] = "else.py" + row.data["previous"] = head + row.data["hash"] = pool_digest({k: v for k, v in row.data.items() if k != "hash"}) + head = row.data["hash"] + records.rows[ref.snapshot_id].data["page_head"] = head + with pytest.raises(ValueError, match="digest"): + await publisher.index.page(ref, 0) + + +async def test_secret_filename_is_refused_before_any_object_write_and_recorded_unavailable(tmp_path): + publisher, workspace, run, records = fixture(tmp_path) + secret = "controlled-private-source-credential-938420" + (workspace / (secret + ".txt")).write_text("not a credential payload") + publisher.content.helper.credential_registry = lambda: {"CONTROLLED_TOKEN": secret} + with pytest.raises(ValueError, match="credentials"): + await publisher.publish(run, workspace, "a" * 40, "b" * 40, authority_fence(), authorize=allowed) + assert next(iter(records.rows.values())).data["status"] == "failed" + assert publisher.content.helper.ledger.rows == {} diff --git a/tests/test_source_archive_wiring_host.py b/tests/test_source_archive_wiring_host.py new file mode 100644 index 000000000..59325a4e5 --- /dev/null +++ b/tests/test_source_archive_wiring_host.py @@ -0,0 +1,108 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Real native registration and the Engine snapshot publication boundary.""" + +import base64 +import hashlib +from types import SimpleNamespace + +from firefly_forge.engine import Engine +from firefly_forge.git_diff_models import GitDiffPageV1, GitDiffRefV1 +from firefly_forge.source_archive_service import SourceArchiveManager +from firefly_forge.source_explorer import SourceExplorer, SourceExplorerController + + +def test_native_application_registers_source_readers_and_publication_lifecycle_without_startup(tmp_path): + from firefly_forge.api import create_runtime + + runtime = create_runtime(tmp_path, "controlled-source-archive-bootstrap-token", "api") + for component in (SourceArchiveManager, SourceExplorer, SourceExplorerController): + assert runtime.context.container.contains_type(component) + + +async def test_engine_snapshot_awaits_publication_and_propagates_failure_before_return(tmp_path, monkeypatch): + import firefly_forge.engine as subject + import firefly_forge.git_diff_artifacts as artifacts + import firefly_forge.source_index_sanitation as sanitation + import firefly_forge.source_snapshot as snapshots + + run = {"id": "a" * 32} + calls = [] + + async def current(): + return None + + class Scanner: + async def __call__(self, *args): + return None + + async def revalidate(self): + return None + + async def scanner(**kwargs): + return Scanner() + + async def get(_): + return run + + async def checked(argv, *args): + if argv[1:] == ["write-tree"]: + return "b" * 40 + if argv[1:] == ["rev-parse", "HEAD"]: + return "c" * 40 + return "" + + async def scan(*args, **kwargs): + return ["app.py"], 1 + + raw = b"+recorded\n" + artifact = GitDiffRefV1( + artifact_id="d" * 32, + manifest_sha256="e" * 64, + sha256=hashlib.sha256(raw).hexdigest(), + byte_count=len(raw), + file_count=1, + index_page_count=1, + ) + + async def retain(*args, scan_chunk, **kwargs): + await scan_chunk(raw, True) + return artifact + + def read(*args): + return GitDiffPageV1( + artifact_id=artifact.artifact_id, + manifest_sha256=artifact.manifest_sha256, + offset=0, + byte_count=len(raw), + total_bytes=len(raw), + next_offset=None, + content_base64=base64.b64encode(raw).decode(), + content_sha256=artifact.sha256, + starts_mid_line=False, + ends_mid_line=False, + utf8_complete=True, + ) + + async def publish(*args, **kwargs): + calls.append(args) + await kwargs["assert_current"]() + raise ValueError("controlled archive publication refusal") + + engine = object.__new__(Engine) + engine.store = SimpleNamespace(get_run=get, source_patch_scanner=scanner) + engine.workspaces = tmp_path + engine.component_roots = {} + engine.assert_ownership = current + engine._assert_git_metadata = lambda _: None + engine.source_archives = SimpleNamespace(publish=publish) + monkeypatch.setattr(subject, "checked", checked) + monkeypatch.setattr(sanitation, "staging_exclusions", lambda *args: []) + monkeypatch.setattr(snapshots, "git_index_identity", lambda _: "stable-index") + monkeypatch.setattr(snapshots, "scan_changed_source", scan) + monkeypatch.setattr(artifacts, "retain_git_diff", retain) + monkeypatch.setattr(artifacts, "read_diff_page", read) + import pytest + + with pytest.raises(ValueError, match="archive publication refusal"): + await engine._snapshot(tmp_path / run["id"], "c" * 40) + assert len(calls) == 1 and calls[0] == (run, tmp_path / run["id"], "c" * 40, "b" * 40) diff --git a/tests/test_source_explorer_clients_host.py b/tests/test_source_explorer_clients_host.py new file mode 100644 index 000000000..1976a8a00 --- /dev/null +++ b/tests/test_source_explorer_clients_host.py @@ -0,0 +1,95 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Typed SDK observations reject foreign or altered source windows.""" + +import copy +from uuid import UUID + +import httpx +import pytest +from test_source_explorer_host import prepared + +from firefly_forge.client import ForgeClient +from firefly_forge.identity import Principal +from firefly_forge.project_access import CURRENT_PRINCIPAL +from firefly_forge.source_explorer import SourceExplorerController + + +async def payload(tmp_path): + explorer, ref, _, _, _ = await prepared(tmp_path) + token = CURRENT_PRINCIPAL.set(Principal(UUID(int=7), "issuer", "user")) + try: + body = await explorer.file( + ref.run_id, ref.snapshot_id, ref.catalog.manifest_sha256, page=0, entry_index=0, offset=0, limit=5 + ) + catalog = await explorer.catalog(ref.run_id, ref.snapshot_id, ref.catalog.manifest_sha256, page=0) + response = await SourceExplorerController(explorer).file( + ref.run_id, ref.snapshot_id, 0, ref.catalog.manifest_sha256, page=0, limit=5 + ) + assert response.headers["Cache-Control"] == "private, no-store" + return ref, body, catalog + finally: + CURRENT_PRINCIPAL.reset(token) + + +async def test_sdk_carries_exact_snapshot_manifest_and_validates_file_window(tmp_path): + ref, body, catalog = await payload(tmp_path) + requests = [] + + def handle(request): + requests.append(request) + assert request.method == "GET" and request.url.params["expected_manifest"] == ref.catalog.manifest_sha256 + return httpx.Response(200, json=catalog if request.url.path.endswith("/catalog") else body) + + with ForgeClient("https://factory.invalid", "controlled-token") as client: + client.http.close() + client.http = httpx.Client(base_url="https://factory.invalid/api/v1/", transport=httpx.MockTransport(handle)) + page = client.run_source_catalog(ref.run_id, ref.snapshot_id, ref.catalog.manifest_sha256, page=0) + data = client.run_source_file( + ref.run_id, ref.snapshot_id, ref.catalog.manifest_sha256, page=0, entry_index=0, limit=5 + ) + assert page.catalog.entries[0].path == "app.py" and data.text == "print" + assert len(requests) == 2 and all("/source/revisions/" + ref.snapshot_id in r.url.path for r in requests) + + +@pytest.mark.parametrize( + "mutation", + ["run", "snapshot", "manifest", "offset", "page", "index", "content", "digest", "text", "binary", "extent"], +) +async def test_sdk_refuses_foreign_or_forged_source_response_without_retry(tmp_path, mutation): + ref, body, _ = await payload(tmp_path) + body = copy.deepcopy(body) + if mutation in {"run", "snapshot", "manifest"}: + if mutation == "manifest": + body["reference"]["catalog"]["manifest_sha256"] = "f" * 64 + else: + body["reference"]["run_id" if mutation == "run" else "snapshot_id"] = "f" * 32 + elif mutation == "offset": + body["offset"] = 1 + elif mutation == "page": + body["catalog_page"] = 1 + elif mutation == "index": + body["entry_index"] = 1 + elif mutation == "content": + body["content_base64"] = "ZmFrZQ==" + elif mutation == "digest": + body["sha256"] = "f" * 64 + elif mutation == "text": + body["text"] = "invented" + elif mutation == "binary": + body["binary"] = True + elif mutation == "extent": + body["entry"]["byte_count"] = 1 + calls = [] + + def handle(request): + calls.append(request) + return httpx.Response(200, json=body) + + with ForgeClient("https://factory.invalid", "controlled-token") as client: + client.http.close() + client.http = httpx.Client(base_url="https://factory.invalid/api/v1/", transport=httpx.MockTransport(handle)) + with pytest.raises(ValueError): + client.run_source_file( + ref.run_id, ref.snapshot_id, ref.catalog.manifest_sha256, page=0, entry_index=0, limit=5 + ) + assert len(calls) == 1 diff --git a/tests/test_source_explorer_host.py b/tests/test_source_explorer_host.py new file mode 100644 index 000000000..1ab8199f4 --- /dev/null +++ b/tests/test_source_explorer_host.py @@ -0,0 +1,108 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Read-only native source observations recheck the current reader and lineage.""" + +import copy +from types import SimpleNamespace +from uuid import UUID + +import pytest +from test_runner_object_artifacts_host import allowed, authority_fence +from test_source_archive_publisher_host import fixture + +from firefly_forge.identity import Principal +from firefly_forge.project_access import CURRENT_PRINCIPAL, AccessDenied +from firefly_forge.source_explorer import SourceExplorer + + +async def prepared(tmp_path): + publisher, workspace, run, records = fixture(tmp_path) + run["execution_actor"] = {"user_id": UUID(int=7).hex} + ref = await publisher.publish(run, workspace, "a" * 40, "b" * 40, authority_fence(), authorize=allowed) + state = SimpleNamespace(run=copy.deepcopy(run), valid=True, checks=0, receipts=[]) + + async def get(value): + assert value == run["id"] + return copy.deepcopy(state.run) + + async def authorize(current, actor, permission): + state.checks += 1 + assert permission == "read" and actor["user_id"] == UUID(int=7).hex + if not state.valid: + raise AccessDenied("revoked") + + async def view(*args, **kwargs): + return {"items": copy.deepcopy(state.receipts)} + + manager = SimpleNamespace( + publisher=publisher, + tenant=UUID(int=1), + settings=publisher.content.helper.objects.settings, + policy=SimpleNamespace(enabled=True, excluded_projects=frozenset()), + ) + deliveries = SimpleNamespace(store=SimpleNamespace(get_run=get), _authorize=authorize, view=view) + return SourceExplorer(manager, deliveries), ref, state, publisher, workspace + + +async def test_catalog_and_binary_windows_have_recorded_lineage_and_no_mutable_workspace_read(tmp_path): + explorer, ref, state, publisher, workspace = await prepared(tmp_path) + reader = Principal(UUID(int=7), "issuer", "user") + token = CURRENT_PRINCIPAL.set(reader) + try: + (workspace / "data.bin").unlink() + observation = await explorer.catalog(ref.run_id, ref.snapshot_id, ref.catalog.manifest_sha256, page=1) + assert observation["catalog"]["entries"][0]["path"] == "data.bin" + data = await explorer.file( + ref.run_id, ref.snapshot_id, ref.catalog.manifest_sha256, page=1, entry_index=0, offset=1, limit=3 + ) + assert data["binary"] is True and data["text"] is None and data["byte_count"] == 3 + assert data["reference"]["tree"] == "b" * 40 and state.checks > 3 + finally: + CURRENT_PRINCIPAL.reset(token) + + +@pytest.mark.parametrize("mutation", ["tenant", "project", "manifest", "reader", "revoked"]) +async def test_foreign_or_revoked_snapshot_never_returns_content(tmp_path, mutation): + explorer, ref, state, publisher, _ = await prepared(tmp_path) + token = CURRENT_PRINCIPAL.set(Principal(UUID(int=7), "issuer", "user")) + expected = ref.catalog.manifest_sha256 + if mutation == "tenant": + explorer.manager.tenant = UUID(int=9) + elif mutation == "project": + state.run["project_id"] = UUID(int=9).hex + elif mutation == "manifest": + expected = "f" * 64 + elif mutation == "reader": + CURRENT_PRINCIPAL.set(None) + elif mutation == "revoked": + state.valid = False + try: + with pytest.raises((AccessDenied, ValueError)): + await explorer.file(ref.run_id, ref.snapshot_id, expected, page=0, entry_index=0, offset=0, limit=3) + finally: + CURRENT_PRINCIPAL.reset(token) + + +async def test_revocation_during_object_read_and_foreign_checkpoint_tree_hold(tmp_path): + explorer, ref, state, publisher, _ = await prepared(tmp_path) + token = CURRENT_PRINCIPAL.set(Principal(UUID(int=7), "issuer", "user")) + try: + state.receipts = [{"request_id": "d" * 32, "source_tree": "f" * 40}] + with pytest.raises(ValueError, match="checkpoint"): + await explorer.catalog( + ref.run_id, ref.snapshot_id, ref.catalog.manifest_sha256, page=0, request_id="d" * 32 + ) + read = publisher.content.helper.objects.read_chunk + + async def revoked(*args, **kwargs): + result = await read(*args, **kwargs) + state.valid = False + return result + + publisher.content.helper.objects.read_chunk = revoked + with pytest.raises(AccessDenied): + await explorer.file( + ref.run_id, ref.snapshot_id, ref.catalog.manifest_sha256, page=0, entry_index=0, offset=0, limit=3 + ) + assert not list(publisher.content.helper.scratch.iterdir()) + finally: + CURRENT_PRINCIPAL.reset(token) diff --git a/tests/test_source_explorer_native.py b/tests/test_source_explorer_native.py new file mode 100644 index 000000000..985a12bb2 --- /dev/null +++ b/tests/test_source_explorer_native.py @@ -0,0 +1,158 @@ +# Copyright 2026 Firefly Software Foundation. SPDX-License-Identifier: Apache-2.0 +"""Dedicated non-superuser PG and controlled OIDC/ASGI; no worker or cloud.""" + +from uuid import UUID, uuid4 + +import httpx +import pytest +from sqlalchemy import text +from test_run_creation_scopes_native import nonadministrator + +from firefly_forge.api import create_application +from firefly_forge.identity import IdentityService +from firefly_forge.object_storage import ObjectWriteUncertain +from firefly_forge.persistence import ForgeRunEntity +from firefly_forge.project_access import CURRENT_PRINCIPAL, ProjectAccess +from firefly_forge.source_archive_service import SourceArchiveManager +from firefly_forge.source_explorer import SourceExplorer + + +async def native(store, tmp_path, monkeypatch, actor, repository, project): + for name, value in { + "FORGE_OBJECT_STORAGE_PROVIDER": "local", + "FORGE_OBJECT_STORAGE_MODE": "test", + "FORGE_OBJECT_STORAGE_STORAGE_ID": UUID(int=8).hex, + "FORGE_OBJECT_STORAGE_TENANT_ID": UUID(int=1).hex, + "FORGE_OBJECT_STORAGE_LOCAL_DIRECTORY": str(tmp_path / "objects"), + }.items(): + monkeypatch.setenv(name, value) + async with store._unit() as session: + assert await session.scalar(text("SELECT rolsuper FROM pg_roles WHERE rolname=current_user")) is False + value = dict( + id=uuid4().hex, + project_id=project["id"], + repository_id=repository["id"], + attempt=1, + status="paused", + branch="recorded/source-test", + agent="api", + sandbox_snapshot={"mode": "docker"}, + execution_actor=ProjectAccess.actor_snapshot(actor), + ) + await store.run_records.save( + ForgeRunEntity(id=UUID(value["id"]), repository_id=UUID(repository["id"]), status="paused", data=value) + ) + await store.acquire_ownership() + manager = SourceArchiveManager(store) + await manager.start() + workspace = tmp_path / "workspace" + workspace.mkdir(mode=0o700) + (workspace / "app.py").write_text('print("native archived source")\n') + (workspace / "binary.dat").write_bytes(b"\xff\x00retained") + return manager, workspace, await store.get_run(value["id"]) + + +async def test_native_lifecycle_asgi_paged_cold_reads_current_project_scope_and_revocation( + store, tmp_path, monkeypatch +): + async with nonadministrator(store, tmp_path) as c: + identity, _, credential, actor, access, repositories, projects, _ = c + manager, workspace, run = await native(store, tmp_path, monkeypatch, actor, repositories[0], projects[0]) + for number in range(129): + (workspace / f"z-{number:03d}.py").write_bytes(b"") + assert CURRENT_PRINCIPAL.get() is None + + async def current(): + await store.assert_ownership() + + await store.release_ownership() + for key, value in { + "FORGE_AUTH_MODE": "oidc", + "FORGE_API_TOKEN": "", + "FORGE_OIDC_ISSUER": identity.config.issuer, + "FORGE_OIDC_CLIENT_ID": identity.config.client_id, + "FORGE_OIDC_CLIENT_SECRET": identity.config.client_secret.get_secret_value(), + "FORGE_OIDC_API_AUDIENCE": identity.config.api_audience, + "FORGE_OIDC_REDIRECT_URI": identity.config.redirect_uri, + "FORGE_IDENTITY_ENCRYPTION_KEY": identity.config.encryption_key.get_secret_value(), + }.items(): + monkeypatch.setenv(key, value) + # Capture the configured native registry after infrastructure initialization, + # then reopen an independent API for cold reads with the same retained keys. + boot = create_application(store.path.parent, role="api") + async with boot.router.lifespan_context(boot): + store.credential_registry = boot.state.store.credential_registry + await store.acquire_ownership() + ref = await manager.publish(run, workspace, "a" * 40, "b" * 40, assert_current=current) + await store.release_ownership() + assert ref.catalog.file_count == 131 and ref.catalog.page_count == 2 + (workspace / "app.py").unlink() + app = create_application(store.path.parent, role="api") + async with app.router.lifespan_context(app): + native_identity = app.state.runtime.context.get_bean(IdentityService) + await native_identity.client.aclose() + native_identity.client = identity.client + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app), + base_url="https://testserver", + headers={"Authorization": "Bearer " + credential}, + ) as client: + prefix = f"/api/v1/runs/{run['id']}/source/revisions" + revisions = await client.get(prefix) + assert revisions.status_code == 200 and revisions.json()["recording_enabled"] is True + assert revisions.json()["items"][0]["reference"]["snapshot_id"] == ref.snapshot_id + params = {"expected_manifest": ref.catalog.manifest_sha256} + page = await client.get(prefix + f"/{ref.snapshot_id}/catalog", params={**params, "page": 1}) + assert page.status_code == 200 and len(page.json()["catalog"]["entries"]) == 3 + content = await client.get( + prefix + f"/{ref.snapshot_id}/files/0", params={**params, "page": 0, "limit": 8} + ) + assert content.status_code == 200 and content.json()["text"] == 'print("n' + assert content.headers["Cache-Control"] == "private, no-store" + explorer = app.state.runtime.context.get_bean(SourceExplorer) + assert explorer.manager.publisher is not manager.publisher + saved = explorer.manager.tenant + explorer.manager.tenant = UUID(int=9) + assert (await client.get(prefix + f"/{ref.snapshot_id}/catalog", params=params)).status_code == 409 + explorer.manager.tenant = saved + await access.set_membership(projects[0]["id"], actor, None) + assert (await client.get(prefix + f"/{ref.snapshot_id}/catalog", params=params)).status_code == 403 + async with store._unit() as session: + assert await session.scalar(text("SELECT count(*) FROM forge_execution_requests")) == 0 + + +async def test_native_lost_source_write_is_retained_unavailable_after_restart_without_retry( + store, tmp_path, monkeypatch +): + async with nonadministrator(store, tmp_path) as c: + _, _, _, actor, _, repositories, projects, _ = c + manager, workspace, run = await native(store, tmp_path, monkeypatch, actor, repositories[0], projects[0]) + adapter = manager.publisher.content.helper.objects.adapter + writes = 0 + + class LostReply: + async def upload(self, *args): + nonlocal writes + writes += 1 + await adapter.upload(*args) + raise TimeoutError("controlled reply loss") + + async def download(self, *args): + return await adapter.download(*args) + + manager.publisher.content.helper.objects.adapter = LostReply() + + async def current(): + await store.assert_ownership() + + with pytest.raises(ObjectWriteUncertain): + await manager.publish(run, workspace, "a" * 40, "b" * 40, assert_current=current) + reopened = SourceArchiveManager(store) + await reopened.start() + rows = await reopened.publisher.index.revisions(run["id"], None) + assert len(rows) == 1 and rows[0]["status"] == "failed" + with pytest.raises(ValueError, match="unavailable"): + await reopened.publisher.index.get(rows[0]["reference"]["snapshot_id"]) + with pytest.raises(RuntimeError, match="reconciliation"): + await reopened.publish(run, workspace, "a" * 40, "b" * 40, assert_current=current) + assert writes == 1