From fcd5549ce231271d6dcb24a4d14b999e2719055d Mon Sep 17 00:00:00 2001 From: Matthew Robert Wesney <157447210+dovvnloading@users.noreply.github.com> Date: Tue, 29 Sep 2026 03:59:36 -0400 Subject: [PATCH 01/11] fix(execution): refuse job ids Windows cannot use as a directory, and report a failed mkdir or open as a repository error The job id names the job's artifact directory but was held only to the artifact name shape. "nul", "con", "aux" and the other device names, with or without an extension, are accepted by create_job and then fail at the first publish with a raw OSError that carries the absolute path; "a." shares a directory with "a" because Windows drops trailing dots. Use a stricter job id shape (no trailing dot, no device name) at create_job and publish_artifact. OSError from mkdir, resolve and the temporary file's open is now reported as ExecutionRepositoryError without the path, and a failing unlink in the failure handler no longer replaces the error being raised. The tests also pin the containment check that runs before mkdir: with link detection blind (older interpreters cannot see a junction) a linked job directory must still be refused before mkdir is pointed at its target. Co-Authored-By: Claude Sonnet 5.5 --- .../cortex_backend/execution/repository.py | 46 ++- tests/test_execution_path_safety.py | 267 ++++++++++++++++++ 2 files changed, 304 insertions(+), 9 deletions(-) create mode 100644 tests/test_execution_path_safety.py diff --git a/backend/cortex_backend/execution/repository.py b/backend/cortex_backend/execution/repository.py index 0069322..eeeb60c 100644 --- a/backend/cortex_backend/execution/repository.py +++ b/backend/cortex_backend/execution/repository.py @@ -70,8 +70,13 @@ _SWEEP_ENTRY_BUDGET = 2_000 _SWEEP_CHILD_LIMIT = 64 # A path-component-safe job id, and a printable owner. The job id becomes a -# directory name under the artifact root, so it is held to the same shape as an -# artifact name; the owner never reaches the filesystem. +# directory name under the artifact root, so it is held to a stricter shape +# than an artifact name: it may not end in a dot (Windows drops trailing dots +# and spaces, so ``a.`` would share a directory with ``a``) and it may not be a +# Windows device name, with or without an extension (``nul`` and ``nul.txt`` name +# the NUL device, not a directory). The owner never reaches the filesystem. +_SAFE_JOB_ID = re.compile(r"[A-Za-z0-9](?:[A-Za-z0-9._-]{0,198}[A-Za-z0-9_-])?") +_WINDOWS_DEVICE_NAME = re.compile(r"(?:con|prn|aux|nul|com[1-9]|lpt[1-9])(?:\..*)?", re.IGNORECASE) _SAFE_OWNER = re.compile(r"[^\x00-\x1f\x7f]{1,200}") _SAFE_INSTALLATION_PRINCIPAL = re.compile(r"^[0-9a-f]{64}$") _SAFE_MIME = re.compile(r"^[a-z0-9][a-z0-9.+-]{0,31}/[a-z0-9][a-z0-9.+-]{0,63}$") @@ -94,6 +99,16 @@ def _is_reparse_point(path: Path) -> bool: return bool(is_junction is not None and is_junction()) +def _is_safe_job_id(value: object) -> bool: + """Whether ``value`` can name a job directory on every platform Cortex runs on.""" + + return ( + isinstance(value, str) + and _SAFE_JOB_ID.fullmatch(value) is not None + and _WINDOWS_DEVICE_NAME.fullmatch(value) is None + ) + + def _has_reparse_parent(path: Path) -> bool: """Reject a configured root that would traverse a link in an ancestor.""" @@ -673,7 +688,7 @@ def create_job( # The job id names the job's artifact directory, so an id that could # lead out of the artifact root is refused when the job is created, not # only when its first artifact is published. - if not isinstance(job_id, str) or _SAFE_NAME.fullmatch(job_id) is None: + if not _is_safe_job_id(job_id): raise ValueError("job_id must be a bounded path-safe identifier") if not isinstance(owner, str) or _SAFE_OWNER.fullmatch(owner) is None: raise ValueError("owner must be a bounded printable identifier") @@ -1667,7 +1682,7 @@ def publish_artifact( raise ArtifactLimitError("Artifact exceeds the configured size limit.") if retention_seconds <= 0: raise ValueError("retention_seconds must be positive") - if not isinstance(job_id, str) or _SAFE_NAME.fullmatch(job_id) is None: + if not _is_safe_job_id(job_id): raise ExecutionRepositoryError("Execution job does not exist.") if self.get_job(job_id) is None: raise ExecutionRepositoryError("Execution job does not exist.") @@ -1691,13 +1706,21 @@ def publish_artifact( stream: Any = None try: with self._directory_lock: - job_root.mkdir(parents=True, exist_ok=True) - resolved_job_root = job_root.resolve(strict=True) + # An OSError from either of these carries the absolute path, + # so it is reported as this repository's own error instead. + try: + job_root.mkdir(parents=True, exist_ok=True) + resolved_job_root = job_root.resolve(strict=True) + except OSError: + raise ExecutionRepositoryError("Artifact directory is unavailable.") from None if not resolved_job_root.is_relative_to(root) or _is_reparse_point(resolved_job_root): raise ExecutionRepositoryError("Artifact path escaped the artifact root.") # The temporary file exists before the lock is released, so the # directory is never empty for a sweep to remove. - stream = temporary.open("xb") + try: + stream = temporary.open("xb") + except OSError: + raise ExecutionRepositoryError("Artifact file could not be created.") from None with stream: stream.write(content) stream.flush() @@ -1736,8 +1759,13 @@ def publish_artifact( except Exception: if stream is not None: stream.close() - target.unlink(missing_ok=True) - temporary.unlink(missing_ok=True) + for leftover in (target, temporary): + try: + leftover.unlink(missing_ok=True) + except OSError: + # Never replaces the error being raised. What stays is a + # file no row names, which the artifact-root sweep reclaims. + pass # The directory this call created (or found empty) is not left behind. self._remove_empty_artifact_directory(job_root) raise diff --git a/tests/test_execution_path_safety.py b/tests/test_execution_path_safety.py new file mode 100644 index 0000000..303bad2 --- /dev/null +++ b/tests/test_execution_path_safety.py @@ -0,0 +1,267 @@ +"""A job id names a directory, so it must be a name every platform can create. + +``create_job("nul")`` used to succeed and the first ``publish_artifact`` then +raised a raw ``OSError`` carrying the absolute path, because ``nul`` is the +Windows NUL device and not a directory name. ``a.`` shared a directory with +``a`` for the same reason: Windows drops trailing dots. +""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from cortex_backend.execution import repository as repository_module +from cortex_backend.execution.repository import ExecutionRepository, ExecutionRepositoryError + + +def _repository(tmp_path: Path) -> ExecutionRepository: + return ExecutionRepository(tmp_path / "execution.sqlite", tmp_path / "artifacts") + + +def _create(repository: ExecutionRepository, job_id: str): + return repository.create_job( + job_id=job_id, + owner=repository.installation_principal_id, + request_id=f"request-{job_id}", + profile="scratch.auto.v1", + payload={}, + ) + + +_REFUSED_JOB_IDS = ( + "nul", + "NUL", + "Nul", + "con", + "prn", + "aux", + "com1", + "COM9", + "lpt1", + "LPT9", + "nul.txt", + "con.tar.gz", + "aux.", # a device name and a trailing dot + "a.", + "a..", + "job-1.", + "x" * 199 + ".", + "x" * 201, + "", + ".hidden", + "-lead", + "a b", + "a/b", + "a\\b", + "..", + "a\n", +) + +_ACCEPTED_JOB_IDS = ( + "a", + "job-1", + "a.b", + "a..b", + "0123456789abcdef0123456789abcdef", + "null", + "console", + "com0", + "com10", + "lpt0", + "nul-1", + "nul_", + "auxiliary", + "x" * 200, +) + + +def _label(job_id: str) -> str: + return job_id if 0 < len(job_id) <= 24 else f"{len(job_id)}-chars-ending-{job_id[-1:]!r}" + + +@pytest.mark.parametrize("job_id", _REFUSED_JOB_IDS, ids=_label) +def test_a_job_id_that_is_not_a_portable_directory_name_is_refused_when_the_job_is_created( + tmp_path: Path, job_id: str +) -> None: + repository = _repository(tmp_path) + + with pytest.raises(ValueError, match="job_id"): + _create(repository, job_id) + + assert repository.list_jobs(owner=repository.installation_principal_id, include_terminal=True) == [] + + +@pytest.mark.parametrize("job_id", _ACCEPTED_JOB_IDS, ids=_label) +def test_ordinary_job_ids_are_still_accepted(tmp_path: Path, job_id: str) -> None: + repository = _repository(tmp_path) + + job, created = _create(repository, job_id) + artifact = repository.publish_artifact(job.job_id, name="out.txt", content=b"synthetic") + + assert created is True + assert repository.read_artifact(artifact.artifact_id) == b"synthetic" + + +@pytest.mark.parametrize("job_id", ("nul", "aux", "a.", "COM1")) +def test_publishing_for_a_job_that_was_stored_under_an_older_shape_touches_nothing( + tmp_path: Path, job_id: str +) -> None: + """A row an earlier build accepted is refused at publish, before any path is built.""" + + repository = _repository(tmp_path) + with repository.connect() as connection: + connection.execute( + """ + INSERT INTO execution_jobs + (job_id, owner, request_id, profile, status, sequence, payload_json, created_at, updated_at) + VALUES (?, ?, 'request-old', 'scratch.auto.v1', 'queued', 0, '{}', 'x', 'x') + """, + (job_id, repository.installation_principal_id), + ) + before = sorted(entry.name for entry in repository.artifact_root.iterdir()) + + with pytest.raises(ExecutionRepositoryError) as refused: + repository.publish_artifact(job_id, name="out.txt", content=b"synthetic") + + assert str(tmp_path) not in str(refused.value) + assert sorted(entry.name for entry in repository.artifact_root.iterdir()) == before + + +def test_a_directory_that_cannot_be_created_is_reported_without_its_path( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + repository = _repository(tmp_path) + job, _ = _create(repository, "job-1") + real_mkdir = Path.mkdir + + def refuses(self: Path, *args: object, **kwargs: object) -> None: + if self.name == "job-1": + raise PermissionError(13, "Access is denied", str(self)) + real_mkdir(self, *args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(Path, "mkdir", refuses) + + with pytest.raises(ExecutionRepositoryError) as failure: + repository.publish_artifact(job.job_id, name="out.txt", content=b"synthetic") + + assert str(tmp_path) not in str(failure.value) + assert failure.value.__cause__ is None + assert failure.value.__suppress_context__ is True + assert not (repository.artifact_root / "job-1").exists() + + +def test_a_file_that_cannot_be_opened_is_reported_without_its_path_and_leaves_nothing( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + repository = _repository(tmp_path) + job, _ = _create(repository, "job-1") + real_open = Path.open + + def refuses(self: Path, mode: str = "r", *args: object, **kwargs: object): + if self.name.startswith(".tmp-"): + raise PermissionError(13, "Access is denied", str(self)) + return real_open(self, mode, *args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(Path, "open", refuses) + + with pytest.raises(ExecutionRepositoryError) as failure: + repository.publish_artifact(job.job_id, name="out.txt", content=b"synthetic") + + assert str(tmp_path) not in str(failure.value) + assert failure.value.__cause__ is None + assert not (repository.artifact_root / "job-1").exists(), "the directory this call made was left behind" + + +def test_a_failing_cleanup_after_a_refused_write_does_not_replace_the_refusal( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """The handler unlinks partial files; an OSError there must not become the error raised.""" + + repository = _repository(tmp_path) + job, _ = _create(repository, "job-1") + real_open = Path.open + real_unlink = Path.unlink + + def refuses(self: Path, mode: str = "r", *args: object, **kwargs: object): + if self.name.startswith(".tmp-"): + raise PermissionError(13, "Access is denied", str(self)) + return real_open(self, mode, *args, **kwargs) # type: ignore[arg-type] + + def cannot_unlink(self: Path, *args: object, **kwargs: object) -> None: + if self.parent.name == "job-1": + raise PermissionError(13, "Access is denied", str(self)) + real_unlink(self, *args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(Path, "open", refuses) + monkeypatch.setattr(Path, "unlink", cannot_unlink) + + with pytest.raises(ExecutionRepositoryError) as failure: + repository.publish_artifact(job.job_id, name="out.txt", content=b"synthetic") + + assert str(tmp_path) not in str(failure.value) + + +def _directory_link(link: Path, target: Path) -> None: + """Make ``link`` a symbolic link, or on Windows a junction, to ``target``; skip if neither can be made.""" + + try: + link.symlink_to(target, target_is_directory=True) + return + except (OSError, NotImplementedError): + pass + try: + import _winapi # type: ignore[import-not-found] + + _winapi.CreateJunction(str(target), str(link)) + except (ImportError, AttributeError, OSError): + pytest.skip("neither symbolic links nor junctions can be created here") + + +def test_a_job_directory_that_is_a_link_is_refused_before_anything_is_created( + tmp_path: Path, +) -> None: + repository = _repository(tmp_path) + job, _ = _create(repository, "job-1") + outside = tmp_path / "outside" + outside.mkdir() + _directory_link(repository.artifact_root / "job-1", outside) + + with pytest.raises(ExecutionRepositoryError, match="escaped"): + repository.publish_artifact(job.job_id, name="out.txt", content=b"synthetic") + + assert list(outside.iterdir()) == [] + + +def test_the_containment_check_alone_refuses_a_link_before_the_directory_is_touched( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """``planned.parent != root`` is the check that does not depend on link detection. + + Older interpreters cannot see a junction (``Path.is_junction`` arrived in + 3.12), so ``_is_reparse_point`` says no; the resolved location is then the + only thing that says the job directory is somewhere else. It must run + before ``mkdir``, which would otherwise be pointed at the link's target. + """ + + repository = _repository(tmp_path) + job, _ = _create(repository, "job-1") + outside = tmp_path / "outside" + outside.mkdir() + _directory_link(repository.artifact_root / "job-1", outside) + monkeypatch.setattr(repository_module, "_is_reparse_point", lambda _path: False) + made: list[Path] = [] + real_mkdir = Path.mkdir + + def recording_mkdir(self: Path, *args: object, **kwargs: object) -> None: + made.append(self) + real_mkdir(self, *args, **kwargs) # type: ignore[arg-type] + + monkeypatch.setattr(Path, "mkdir", recording_mkdir) + + with pytest.raises(ExecutionRepositoryError, match="escaped"): + repository.publish_artifact(job.job_id, name="out.txt", content=b"synthetic") + + assert made == [], "a refused job directory was still handed to mkdir" + assert list(outside.iterdir()) == [] From f71e2ae773d4087ab4adab1dbcce1571c93ade6c Mon Sep 17 00:00:00 2001 From: Matthew Robert Wesney <157447210+dovvnloading@users.noreply.github.com> Date: Tue, 29 Sep 2026 04:00:16 -0400 Subject: [PATCH 02/11] fix(execution): finish_unsuccessful_job contains a failed job read and a raising cancel probe The docstring says nothing raises because this runs in a worker's error path, but repository.get_job and the cancel_requested callable sat outside the try, so a store failure while reading the job (or a raising probe) replaced the outcome the caller was about to report. Move both inside the guard; the failure is logged by exception type only and the job is left as it was. Co-Authored-By: Claude Sonnet 5.5 --- backend/cortex_backend/execution/finish.py | 19 +++---- tests/test_execution_finish.py | 60 +++++++++++++++++++++- 2 files changed, 69 insertions(+), 10 deletions(-) diff --git a/backend/cortex_backend/execution/finish.py b/backend/cortex_backend/execution/finish.py index 7014fda..e33931d 100644 --- a/backend/cortex_backend/execution/finish.py +++ b/backend/cortex_backend/execution/finish.py @@ -52,19 +52,20 @@ def finish_unsuccessful_job( against; if a concurrent actor moved the job first, the decision is made again from the new state. A job that is already terminal is left alone. Nothing here raises: this runs in a worker's error path, where the original - outcome is what matters. + outcome is what matters. That covers the read of the job and the + ``cancel_requested`` probe as well as the write. """ for _ in range(_ATTEMPTS): - current = repository.get_job(job_id) - if current is None or current.status in TerminalExecutionStatus: - return - cancelled = ( - failure_code == "cancelled" - or cancel_requested() - or current.status == "cancelling" - ) try: + current = repository.get_job(job_id) + if current is None or current.status in TerminalExecutionStatus: + return + cancelled = ( + failure_code == "cancelled" + or cancel_requested() + or current.status == "cancelling" + ) repository.transition( job_id, status="cancelled" if cancelled else "failed", diff --git a/tests/test_execution_finish.py b/tests/test_execution_finish.py index 6fc65d0..a5f3470 100644 --- a/tests/test_execution_finish.py +++ b/tests/test_execution_finish.py @@ -5,7 +5,11 @@ from threading import Event from cortex_backend.execution.finish import UnsuccessfulJobWording, finish_unsuccessful_job -from cortex_backend.execution.repository import ExecutionRepository, ExecutionTransitionConflict +from cortex_backend.execution.repository import ( + ExecutionRepository, + ExecutionRepositoryError, + ExecutionTransitionConflict, +) WORDING = UnsuccessfulJobWording( cancelled_event="cancelled", @@ -178,6 +182,60 @@ def broken(job_id, **kwargs): assert "simulated persistence outage" not in caplog.text +def test_a_job_that_cannot_be_read_is_contained_like_a_job_that_cannot_be_written( + execution_repository, monkeypatch, caplog +): + """``get_job`` sat outside the ``try``, so a store failure escaped a function that says nothing raises. + + This runs in a worker's error path; an exception here replaces the outcome + the caller was about to report. + """ + + job = _running_job(execution_repository) + + def unreadable(job_id, **kwargs): + raise ExecutionRepositoryError("SQLite execution operation failed.") + + monkeypatch.setattr(execution_repository, "get_job", unreadable) + + with caplog.at_level("WARNING", logger="cortex.execution.finish"): + finish_unsuccessful_job( + execution_repository, + job.job_id, + failure_code="worker_failed", + cancel_requested=Event().is_set, + wording=WORDING, + ) + + monkeypatch.undo() + assert execution_repository.get_job(job.job_id).status == "running" # nothing was written + assert any("ExecutionRepositoryError" in record.getMessage() for record in caplog.records) + assert "SQLite execution operation failed" not in caplog.text + assert job.job_id not in caplog.text + + +def test_a_cancel_probe_that_raises_is_contained_and_logged_by_type_only( + execution_repository, caplog +): + job = _running_job(execution_repository) + + def probe() -> bool: + raise RuntimeError("simulated probe failure") + + with caplog.at_level("WARNING", logger="cortex.execution.finish"): + finish_unsuccessful_job( + execution_repository, + job.job_id, + failure_code="worker_failed", + cancel_requested=probe, + wording=WORDING, + ) + + assert execution_repository.get_job(job.job_id).status == "running" + assert any("RuntimeError" in record.getMessage() for record in caplog.records) + assert "simulated probe failure" not in caplog.text + + def test_the_scratch_profile_does_not_overwrite_a_stop_with_a_failure( coordinator, monkeypatch ): From dd9a04f6d5a381f6a7373bebcdc817c37e5dd2f0 Mon Sep 17 00:00:00 2001 From: Matthew Robert Wesney <157447210+dovvnloading@users.noreply.github.com> Date: Tue, 29 Sep 2026 04:01:02 -0400 Subject: [PATCH 03/11] fix(execution): a failing artifact-root sweep no longer discards the retention pass that finished The supervisor ran the sweep inside the same try as the retention pass, so a failure in the sweep (connect() on a locked store, an unreadable directory) after the rows had been reclaimed recorded the pass as failed and dropped its artifact, job and event counters, as if the work had not happened. Give the sweep its own guard: the pass still counts as a success with its counters, the failure is logged by exception type only and kept in last_error, and the sweep is retried on the next pass. Co-Authored-By: Claude Sonnet 5.5 --- backend/cortex_backend/execution/cleanup.py | 20 +++++- tests/test_execution_cleanup.py | 69 +++++++++++++++++++++ 2 files changed, 88 insertions(+), 1 deletion(-) diff --git a/backend/cortex_backend/execution/cleanup.py b/backend/cortex_backend/execution/cleanup.py index 3c0ee20..22f0f8b 100644 --- a/backend/cortex_backend/execution/cleanup.py +++ b/backend/cortex_backend/execution/cleanup.py @@ -178,7 +178,7 @@ def run_once(self) -> bool: raise RuntimeError("cleanup returned an invalid result") # After the rows, so a directory the pass just emptied is # already gone by the time the sweep looks at it. - swept = self.repository.sweep_artifact_root(limit=self.batch_size) + swept = self._sweep() if self._lease_lost.is_set(): raise RuntimeError("cleanup lease lost") self._increment( @@ -215,6 +215,24 @@ def run_once(self) -> bool: finally: self._run_lock.release() + def _sweep(self) -> int: + """Sweep the artifact root, keeping a failure from undoing the retention pass. + + The retention pass has already committed by now, so the sweep failing + (a locked store, an unreadable directory) must not record the whole + pass as failed and drop its counters. The failure is logged by type + only, kept in ``last_error`` and simply retried on the next pass. + """ + + try: + return self.repository.sweep_artifact_root(limit=self.batch_size) + except Exception as exc: + self._increment(last_error=type(exc).__name__) + _LOGGER.warning( + "Cortex artifact-root sweep failed (%s); it will be retried.", type(exc).__name__ + ) + return 0 + def _run(self) -> None: while not self._stop_event.is_set(): self.run_once() diff --git a/tests/test_execution_cleanup.py b/tests/test_execution_cleanup.py index 31bc6b4..205e056 100644 --- a/tests/test_execution_cleanup.py +++ b/tests/test_execution_cleanup.py @@ -423,6 +423,75 @@ def test_cleanup_supervisor_reports_rows_it_could_not_reclaim_and_still_succeeds assert str(tmp_path) not in caplog.text +def test_a_failing_sweep_does_not_discard_the_retention_pass_that_already_finished( + tmp_path, monkeypatch, caplog +): + """The sweep shared one ``try`` with the retention pass. + + ``connect()`` failing inside the sweep, after the rows were already + reclaimed, recorded the whole pass as failed and dropped its counters, so + the work looked like it had not happened. + """ + + repository = _repository(tmp_path) + expired = _terminal_job(repository, "expired") + artifact = repository.publish_artifact( + expired.job_id, name="expired.txt", content=b"remove", mime_type="text/plain" + ) + with repository.connect() as connection: + connection.execute( + "UPDATE execution_artifacts SET expires_at = ? WHERE artifact_id = ?", + ("2000-01-01T00:00:00+00:00", artifact.artifact_id), + ) + _terminal_job(repository, "idle") + + def sweep_cannot_open_the_store(**_kwargs): + raise ExecutionRepositoryError(f"SQLite execution operation failed at {tmp_path}") + + monkeypatch.setattr(repository, "sweep_artifact_root", sweep_cannot_open_the_store) + supervisor = ExecutionCleanupSupervisor(repository, terminal_job_retention_seconds=0) + + with caplog.at_level(logging.WARNING, logger="cortex.execution.cleanup"): + assert supervisor.run_once() is True + + metrics = supervisor.metrics + assert (metrics.successes, metrics.failures) == (1, 0) + assert (metrics.artifacts, metrics.jobs, metrics.events) == (1, 2, 4) + assert metrics.swept == 0 + assert metrics.last_error == "ExecutionRepositoryError" + assert "sweep failed (ExecutionRepositoryError)" in caplog.text + assert str(tmp_path) not in caplog.text + # The pass really did its work, and the lease is free for the next one. + assert repository.get_job("expired") is None + assert repository.get_artifact(artifact.artifact_id) is None + repository.claim_cleanup_lease(lease_owner="next", ttl_seconds=1) + + +def test_a_sweep_that_fails_is_retried_by_the_next_pass(tmp_path, monkeypatch): + repository = _repository(tmp_path) + (repository.artifact_root / "job-empty").mkdir() + real_sweep = repository.sweep_artifact_root + outcomes: list[str] = [] + + def flaky_sweep(**kwargs): + if not outcomes: + outcomes.append("failed") + raise ExecutionRepositoryError("SQLite execution operation failed.") + outcomes.append("ran") + return real_sweep(**kwargs) + + monkeypatch.setattr(repository, "sweep_artifact_root", flaky_sweep) + supervisor = ExecutionCleanupSupervisor(repository) + + assert supervisor.run_once() is True + assert (repository.artifact_root / "job-empty").is_dir() + assert supervisor.run_once() is True + + assert outcomes == ["failed", "ran"] + assert supervisor.metrics.swept == 1 + assert not (repository.artifact_root / "job-empty").exists() + + def test_cleanup_supervisor_skips_live_peer_and_local_overlap(tmp_path): repository = _repository(tmp_path) first = ExecutionCleanupSupervisor(repository) From d8c082ac29fa538dbef73419e48a702b0bd13f7e Mon Sep 17 00:00:00 2001 From: Matthew Robert Wesney <157447210+dovvnloading@users.noreply.github.com> Date: Tue, 29 Sep 2026 04:03:34 -0400 Subject: [PATCH 04/11] fix(execution): make the artifact-root sweep's removal limit exact and test its entry budget The limit was checked only between top-level entries, so a single job directory could exceed it: with limit=1, one directory of 64 stale files lost all 64 files and then itself in one pass (65 removals). The entry-budget clause had no test that would fail without it. Pass each entry the removals it has left. A job directory stops at the limit, reports itself incomplete and is resumed by the next pass at that same directory, so nothing is skipped. The docstring now states exactly what is bounded: removals are exact, and the entry budget is checked between top-level entries so it can be overshot by one directory's children. Co-Authored-By: Claude Sonnet 5.5 --- .../cortex_backend/execution/repository.py | 80 +++++++++++----- tests/test_artifact_reclamation.py | 92 +++++++++++++++++++ 2 files changed, 150 insertions(+), 22 deletions(-) diff --git a/backend/cortex_backend/execution/repository.py b/backend/cortex_backend/execution/repository.py index eeeb60c..3456bef 100644 --- a/backend/cortex_backend/execution/repository.py +++ b/backend/cortex_backend/execution/repository.py @@ -2340,9 +2340,18 @@ def sweep_artifact_root(self, *, now: datetime | None = None, limit: int = 100) job; and * a link, a reparse point or anything of another name is never touched. - The pass is bounded: it removes at most ``limit`` things and looks at - at most ``_SWEEP_ENTRY_BUDGET`` entries, resuming where it stopped on - the next call. Best effort: a failure on one entry never stops the pass. + The pass is bounded, and the bounds are exact where they matter. It + removes at most ``limit`` things, counting every file, every job + directory and every staging directory as one; a job directory that + would need more than what is left is worked on until the limit is + reached and picked up again by the next call, at that same directory. + It stops taking on new top-level entries once it has looked at + ``_SWEEP_ENTRY_BUDGET`` entries (a directory counts as itself plus the + children it reads). That check runs between top-level entries, so a + pass can look at up to ``_SWEEP_CHILD_LIMIT`` + 1 entries more than the + budget; it never removes more than ``limit``. The next call resumes + where this one stopped. Best effort: a failure on one entry never + stops the pass. """ if isinstance(limit, bool) or not 1 <= limit <= 10_000: @@ -2365,55 +2374,76 @@ def sweep_artifact_root(self, *, now: datetime | None = None, limit: int = 100) if removed >= limit or examined >= _SWEEP_ENTRY_BUDGET: finished = False break - gone, looked_at = self._sweep_entry(connection, self.artifact_root / name, stale_before) + gone, looked_at, complete = self._sweep_entry( + connection, self.artifact_root / name, stale_before, limit - removed + ) removed += gone examined += looked_at + if not complete: + # The limit ran out inside this entry. What it removed is + # gone for good, so the next pass starts here and finishes it. + finished = False + break self._sweep_cursor = name if finished: self._sweep_cursor = "" return removed def _sweep_entry( - self, connection: sqlite3.Connection, entry: Path, stale_before: float - ) -> tuple[int, int]: - """Sweep one entry directly under the artifact root; return (removed, examined).""" + self, + connection: sqlite3.Connection, + entry: Path, + stale_before: float, + allowance: int, + ) -> tuple[int, int, bool]: + """Sweep one entry directly under the artifact root. + + Returns ``(removed, examined, complete)``. ``removed`` never exceeds + ``allowance``; ``complete`` is false only when the allowance ran out + with more of this entry left to remove, so the caller must come back + to it instead of moving on. + """ name = entry.name try: if _is_reparse_point(entry) or not stat.S_ISDIR(entry.lstat().st_mode): - return 0, 1 + return 0, 1, True except OSError: - return 0, 1 + return 0, 1, True if name == _LEGACY_QUARANTINE_NAME: - return int(self._remove_empty_directory(entry)), 1 + return int(self._remove_empty_directory(entry)), 1, True staging = _RECIPE_STAGING_NAME.fullmatch(name) if staging is not None: job_id = staging["job_id"] if _SAFE_NAME.fullmatch(job_id) is None: - return 0, 1 + return 0, 1, True job = connection.execute( "SELECT status FROM execution_jobs WHERE job_id = ?", (job_id,) ).fetchone() if job is not None and job["status"] not in TerminalExecutionStatus: - return 0, 1 # its worker may still be writing here + return 0, 1, True # its worker may still be writing here try: shutil.rmtree(entry) except OSError: - return 0, 1 - return 1, 1 + return 0, 1, True + return 1, 1, True if name.startswith(".") or _SAFE_NAME.fullmatch(name) is None: - return 0, 1 - return self._sweep_job_directory(connection, entry, stale_before) + return 0, 1, True + return self._sweep_job_directory(connection, entry, stale_before, allowance) def _sweep_job_directory( - self, connection: sqlite3.Connection, directory: Path, stale_before: float - ) -> tuple[int, int]: + self, + connection: sqlite3.Connection, + directory: Path, + stale_before: float, + allowance: int, + ) -> tuple[int, int, bool]: removed = 0 examined = 1 try: children = list(itertools.islice(directory.iterdir(), _SWEEP_CHILD_LIMIT)) except OSError: - return 0, examined + return 0, examined, True for child in children: examined += 1 try: @@ -2429,6 +2459,8 @@ def _sweep_job_directory( (stored["artifact_id"],), ).fetchone() is not None: continue + if removed >= allowance: + return removed, examined, False try: child.unlink() except OSError: @@ -2437,9 +2469,13 @@ def _sweep_job_directory( named = connection.execute( "SELECT 1 FROM execution_artifacts WHERE job_id = ? LIMIT 1", (directory.name,) ).fetchone() - if named is None and self._remove_empty_directory(directory): - removed += 1 - return removed, examined + if named is None: + if removed >= allowance: + # The directory may be empty now, and taking it is one more removal. + return removed, examined, False + if self._remove_empty_directory(directory): + removed += 1 + return removed, examined, True @staticmethod def _encode_event(data: Mapping[str, Any]) -> str: diff --git a/tests/test_artifact_reclamation.py b/tests/test_artifact_reclamation.py index 39668ea..2879ca8 100644 --- a/tests/test_artifact_reclamation.py +++ b/tests/test_artifact_reclamation.py @@ -340,6 +340,98 @@ def test_the_sweep_is_bounded_and_resumes_where_it_stopped(tmp_path, monkeypatch repository.sweep_artifact_root(limit=0) +def _stale_temporary_files(directory: Path, count: int) -> list[Path]: + """``count`` stranded ``.tmp-*`` files, old enough for the sweep to remove.""" + + made = [] + for index in range(count): + stale = directory / f".tmp-{index:032x}" + stale.write_bytes(b"half a write") + _age(stale) + made.append(stale) + return made + + +@pytest.mark.parametrize("limit", (1, 2, 7, 64, 65, 100)) +def test_a_pass_never_removes_more_than_its_limit_even_inside_one_directory(tmp_path, limit): + """The limit was checked only between entries. + + With ``limit=1`` one directory of 64 stale files lost all 64 of them and + then itself -- 65 removals -- in a single pass. + """ + + repository = _repository(tmp_path) + directory = repository.artifact_root / "job-big" + directory.mkdir() + _stale_temporary_files(directory, repository_module._SWEEP_CHILD_LIMIT) + + per_pass = [] + for _ in range(80): # bounded: 65 things to remove + per_pass.append(repository.sweep_artifact_root(limit=limit)) + if not directory.exists(): + break + + assert max(per_pass) <= limit + assert sum(per_pass) == repository_module._SWEEP_CHILD_LIMIT + 1 # every file, then the directory + assert not directory.exists() + + +def test_an_entry_cut_short_by_the_limit_is_resumed_not_skipped(tmp_path): + repository = _repository(tmp_path) + first = repository.artifact_root / "job-a" + second = repository.artifact_root / "job-b" + for directory, count in ((first, 3), (second, 1)): + directory.mkdir() + _stale_temporary_files(directory, count) + + per_pass = [] + for _ in range(12): # bounded: six things to remove + per_pass.append(repository.sweep_artifact_root(limit=2)) + if not first.exists() and not second.exists(): + break + + assert per_pass == [2, 2, 2] + assert not first.exists() and not second.exists() + + +def test_the_limit_bounds_staging_and_legacy_directories_as_well_as_job_directories(tmp_path): + repository = _repository(tmp_path) + _job(repository, "job-done", finished=True) + for index in range(3): + staging = repository.artifact_root / f".recipe-job-done-abcd123{index}" + staging.mkdir() + (staging / "output").write_bytes(b"synthetic") + (repository.artifact_root / ".artifact_quarantine").mkdir() + + per_pass = [] + for _ in range(8): # bounded: four things to remove + per_pass.append(repository.sweep_artifact_root(limit=1)) + if per_pass[-1] == 0: + break + + assert per_pass == [1, 1, 1, 1, 0] + assert [entry.name for entry in repository.artifact_root.iterdir()] == [".quarantine"] + + +def test_the_entry_budget_stops_a_pass_early_and_the_next_one_carries_on(tmp_path, monkeypatch): + """The clause was never exercised: the earlier test would pass without it.""" + + repository = _repository(tmp_path) + monkeypatch.setattr(repository_module, "_SWEEP_ENTRY_BUDGET", 3) + for index in range(6): + (repository.artifact_root / f"job-{index}").mkdir() + + per_pass = [] + for _ in range(10): # bounded: at most one pass per entry + per_pass.append(repository.sweep_artifact_root(limit=100)) + if not any((repository.artifact_root / f"job-{index}").exists() for index in range(6)): + break + + assert max(per_pass) <= 3, "one pass looked at more entries than its budget allows" + assert len([removed for removed in per_pass if removed]) >= 2 + assert sum(per_pass) == 6 + + def test_the_sweep_never_removes_an_unexpired_artifact_or_a_directory_a_publish_just_made(tmp_path): repository = _repository(tmp_path) job_id = _job(repository, "job-1") From 770d1bfef30aa5a6a18c4f69e31fcd55150aef82 Mon Sep 17 00:00:00 2001 From: Matthew Robert Wesney <157447210+dovvnloading@users.noreply.github.com> Date: Tue, 29 Sep 2026 04:04:55 -0400 Subject: [PATCH 05/11] test(execution): kill three mutants the review of the small-defects change found surviving Each of these was verified by removing the guard and watching the new test fail. - Dropping owner= from read_artifact in attachment staging's repeated-request path and in the recipe coordinator's input load left every test green, because get_artifact is already owner-scoped; a recording read now pins the owner that is passed. - Removing the reparse-point check from the artifact-root sweep survived because the link test used a symbolic link, which lstat reports as a link and the directory check rejects anyway. A junction reads as a plain directory to lstat, so the new test makes a real one (Windows only) whose target holds a stale temporary file the mutant deletes. The fourth survivor, the containment clause before mkdir in publish_artifact, is pinned in tests/test_execution_path_safety.py. Co-Authored-By: Claude Sonnet 5.5 --- tests/test_artifact_reclamation.py | 39 ++++++++++++++++++++++++++++++ tests/test_attachment_staging.py | 26 ++++++++++++++++++++ tests/test_recipe_coordinator.py | 21 ++++++++++++++++ 3 files changed, 86 insertions(+) diff --git a/tests/test_artifact_reclamation.py b/tests/test_artifact_reclamation.py index 2879ca8..3461631 100644 --- a/tests/test_artifact_reclamation.py +++ b/tests/test_artifact_reclamation.py @@ -315,6 +315,45 @@ def test_the_sweep_never_follows_or_removes_a_link(tmp_path): assert link.is_symlink() +def _make_junction(link: Path, target: Path) -> None: + """A real directory junction: a reparse point ``lstat`` reports as a plain directory.""" + + if os.name != "nt": + pytest.skip("directory junctions exist only on Windows") + import _winapi # type: ignore[import-not-found] + + try: + _winapi.CreateJunction(str(target), str(link)) + except OSError: + pytest.skip("directory junctions cannot be created here") + + +@pytest.mark.parametrize("name", ("job-junction", ".recipe-job-gone-abcd1234")) +def test_the_sweep_never_follows_or_removes_a_junction(tmp_path, name): + """A symbolic link reads as a link to ``lstat``; a junction reads as a directory. + + Only the reparse-point check stands between the sweep and the files the + junction leads to, so it needs a real junction to be tested at all. + """ + + repository = _repository(tmp_path) + outside = tmp_path / "outside" + outside.mkdir() + (outside / f".tmp-{'e' * 32}").write_bytes(b"not ours") + _age(outside / f".tmp-{'e' * 32}") + (outside / "output").write_bytes(b"not ours either") + junction = repository.artifact_root / name + _make_junction(junction, outside) + try: + assert repository.sweep_artifact_root() == 0 + + assert (outside / f".tmp-{'e' * 32}").read_bytes() == b"not ours" + assert (outside / "output").read_bytes() == b"not ours either" + assert junction.is_junction() + finally: + os.rmdir(junction) # removes the junction itself and nothing behind it + + def test_the_sweep_is_bounded_and_resumes_where_it_stopped(tmp_path, monkeypatch): repository = _repository(tmp_path) directories = [] diff --git a/tests/test_attachment_staging.py b/tests/test_attachment_staging.py index f1ba5db..75c672d 100644 --- a/tests/test_attachment_staging.py +++ b/tests/test_attachment_staging.py @@ -324,6 +324,32 @@ def test_a_retry_never_changes_the_payload_check_or_an_existing_success(executio assert conflict.value.code == "request_conflict" +def test_a_repeated_request_reads_the_staged_bytes_as_their_owner_never_unscoped( + execution_repository, monkeypatch +): + """``get_artifact`` is owner-scoped; the read that follows it must be too. + + The two calls are separate queries, so an unscoped read would trust + whatever the first one saw. + """ + + repository, _boundary, service = _service_over(execution_repository) + content = _image_bytes() + service.stage(owner=OWNER, request_id="attach-owner", content=content) + real_read = repository.read_artifact + owners: list[str | None] = [] + + def recording(artifact_id, *, owner=None): + owners.append(owner) + return real_read(artifact_id, owner=owner) + + monkeypatch.setattr(repository, "read_artifact", recording) + + service.stage(owner=OWNER, request_id="attach-owner", content=content) + + assert owners == [OWNER] + + def test_a_stager_finishing_after_its_job_was_retired_reports_failure_and_drops_the_artifact( execution_repository, monkeypatch ): diff --git a/tests/test_recipe_coordinator.py b/tests/test_recipe_coordinator.py index 64c77df..4cb8870 100644 --- a/tests/test_recipe_coordinator.py +++ b/tests/test_recipe_coordinator.py @@ -200,6 +200,27 @@ def test_coordinator_rejects_request_conflict_and_wrong_owner_artifact(tmp_path: assert missing.value.code == "input_artifact_unavailable" +def test_the_source_artifact_is_read_as_its_owner_never_unscoped(tmp_path: Path, monkeypatch): + repository, _source_job_id, source_artifact_id = _repository(tmp_path) + coordinator = RecipeExecutionCoordinator(repository, lambda _job: _FakeAttempt(_output())) + real_read = repository.read_artifact + owners: list[str | None] = [] + + def recording(artifact_id, *, owner=None): + owners.append(owner) + return real_read(artifact_id, owner=owner) + + monkeypatch.setattr(repository, "read_artifact", recording) + + assert coordinator._load_input(OWNER, source_artifact_id) == _image_bytes() + assert owners == [OWNER] + + with pytest.raises(RecipeExecutionError) as foreign: + coordinator._load_input("b" * 64, source_artifact_id) + assert foreign.value.code == "input_artifact_unavailable" + assert owners == [OWNER], "another owner's artifact was read before it was refused" + + def test_coordinator_failure_is_redacted_and_does_not_publish(tmp_path: Path): repository, _source_job_id, source_artifact_id = _repository(tmp_path) attempt = _FakeAttempt(error="provider_failed") From b900cd04dfd065fd3a12bbfedd9fd337e434e0be Mon Sep 17 00:00:00 2001 From: Matthew Robert Wesney <157447210+dovvnloading@users.noreply.github.com> Date: Tue, 29 Sep 2026 04:07:00 -0400 Subject: [PATCH 06/11] fix(chat): hold a lease while a chat attachment is staged and retire a job whose stager died Chat attachment staging creates a job, publishes the bytes and then completes the job, with nothing holding the job in between. A stager killed in that window left a queued job that recovery cannot see (it only looks at jobs with a lease) and that nothing retires, so it stayed in the task tray for good and its request id answered a conflict on every retry. The execution attachment staging closed the same window earlier. Take a short lease while the bytes are staged and release it on every exit that runs code, and let a retry of the same request retire a job nobody is working on (no live lease, no write for ABANDONED_STAGE_SECONDS). The answer to the retry is unchanged; only the job's state is. A request that conflicts with the job never retires it, and a live or recent stager is left alone. Not covered: the startup recovery in local_runtime does not yet fail a chat-attachment job whose lease it reclaims, so a stager killed and never retried is still left queued until that is added. Co-Authored-By: Claude Sonnet 5.5 --- .../cortex_backend/services/attachments.py | 111 ++++++++-- tests/test_chat_attachments.py | 205 +++++++++++++++++- 2 files changed, 301 insertions(+), 15 deletions(-) diff --git a/backend/cortex_backend/services/attachments.py b/backend/cortex_backend/services/attachments.py index 924acc3..d382269 100644 --- a/backend/cortex_backend/services/attachments.py +++ b/backend/cortex_backend/services/attachments.py @@ -26,6 +26,12 @@ from PIL import Image +from cortex_backend.execution.attachment_staging import ( + ABANDONED_STAGE_SECONDS, + ATTACHMENT_INTERRUPTED, + ATTACHMENT_INTERRUPTED_MESSAGE, + STAGING_LEASE_SECONDS, +) from cortex_backend.execution.repository import ExecutionRepository, ExecutionRepositoryError @@ -357,8 +363,50 @@ def stage( ) if not created: return self._existing(job, payload, content) + + # A stager that dies between creating the job and finishing it leaves a + # queued job that recovery cannot see unless it holds a lease, and that + # nothing else ever retires. Holding a short lease while the bytes are + # staged makes the job findable, and it is released on every exit that + # runs code. This is the same window the execution staging closes. + lease_owner = f"chat-attachment-stager-{uuid4().hex}" + try: + self.repository.claim_lease( + job.job_id, lease_owner=lease_owner, ttl_seconds=STAGING_LEASE_SECONDS + ) + except ExecutionRepositoryError as exc: + self._fail(job.job_id) + raise ChatAttachmentError("attachment_persist_failed") from exc + try: + return self._stage_created( + job, + safe_name=safe_name, + kind=kind, + mime_type=mime_type, + content=content, + digest=digest, + ) + finally: + try: + self.repository.release_lease(job.job_id, lease_owner=lease_owner) + except Exception: + pass # It expires by itself; nothing here may replace the real outcome. + + def _stage_created( + self, + job: Any, + *, + safe_name: str, + kind: str, + mime_type: str, + content: bytes, + digest: str, + ) -> ChatAttachment: + repository = self.repository + if repository is None: + raise ChatAttachmentError("attachment_unavailable") try: - artifact = self.repository.publish_artifact( + artifact = repository.publish_artifact( job.job_id, name=f"chat-{digest[:24]}.bin", content=content, @@ -374,7 +422,7 @@ def stage( kind=kind, expires_at=artifact.expires_at, ) - self.repository.transition( + repository.transition( job.job_id, status="succeeded", event="completed", @@ -384,19 +432,26 @@ def stage( ) return descriptor except (ExecutionRepositoryError, OSError) as exc: - try: - self.repository.transition( - job.job_id, - status="failed", - event="failed", - phase="failed", - data={"message": "Chat attachment staging failed."}, - error="attachment_persist_failed", - ) - except Exception: - pass + self._fail(job.job_id) raise ChatAttachmentError("attachment_persist_failed") from exc + def _fail(self, job_id: str) -> None: + """Record a failed staging job; best effort, because the caller is already raising.""" + + if self.repository is None: + return + try: + self.repository.transition( + job_id, + status="failed", + event="failed", + phase="failed", + data={"message": "Chat attachment staging failed."}, + error="attachment_persist_failed", + ) + except Exception: + pass + def resolve(self, *, owner: str, descriptor: Mapping[str, Any] | ChatAttachment) -> ResolvedChatAttachment: _safe_owner(owner) normalized = self._normalize_descriptor(descriptor) @@ -478,8 +533,36 @@ def _record_expired(record: _MemoryRecord, now: datetime) -> bool: return True return expiry <= now + def _retire_if_abandoned(self, job: Any) -> Any: + """Fail a job whose stager died, and return the job as it now reads. + + The store decides, atomically, that nothing is working on it: no live + lease and no write for ABANDONED_STAGE_SECONDS. A job that is merely + slow, or that finished in the meantime, comes back as it is. + """ + + if self.repository is None: + return job + try: + current = self.repository.retire_abandoned_job( + job.job_id, + idle_seconds=ABANDONED_STAGE_SECONDS, + error=ATTACHMENT_INTERRUPTED, + message=ATTACHMENT_INTERRUPTED_MESSAGE, + ) + except ExecutionRepositoryError: + return job + return job if current is None else current + def _existing(self, job: Any, payload: Mapping[str, Any], content: bytes) -> ChatAttachment: - if job.profile != CHAT_ATTACHMENT_PROFILE or dict(job.payload) != dict(payload) or job.status != "succeeded": + if job.profile != CHAT_ATTACHMENT_PROFILE or dict(job.payload) != dict(payload): + raise ChatAttachmentError("attachment_request_conflict") + if job.status in {"queued", "running"}: + # Only the request that matches the job may retire it, and only a + # job nobody is working on: a stager killed part-way otherwise + # left it queued for good. + job = self._retire_if_abandoned(job) + if job.status != "succeeded": raise ChatAttachmentError("attachment_request_conflict") result = job.result descriptor = self._normalize_descriptor(result) diff --git a/tests/test_chat_attachments.py b/tests/test_chat_attachments.py index 98115d8..216bca9 100644 --- a/tests/test_chat_attachments.py +++ b/tests/test_chat_attachments.py @@ -6,6 +6,7 @@ import binascii from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta, timezone +import hashlib from io import BytesIO from pathlib import Path import struct @@ -20,6 +21,7 @@ from cortex_backend.core.generation import GenerationAttachment import cortex_backend.services.attachments as attachments_module from cortex_backend.services.attachments import ( + CHAT_ATTACHMENT_PROFILE, ChatAttachmentError, ChatAttachmentService, MAX_CHAT_IMAGE_DECODED_BYTES, @@ -28,7 +30,12 @@ ) from cortex_backend.services.llm import PromptTemplate from cortex_backend.testing.fake_ollama import FakeOllamaState -from cortex_backend.execution.repository import ExecutionRepository +from cortex_backend.execution.attachment_staging import ( + ABANDONED_STAGE_SECONDS, + ATTACHMENT_INTERRUPTED, + STAGING_LEASE_SECONDS, +) +from cortex_backend.execution.repository import ExecutionRepository, ExecutionRepositoryError from support import session_headers as _session @@ -343,6 +350,202 @@ def test_durable_staging_is_idempotent_and_integrity_checked(tmp_path: Path): assert foreign.value.code == "attachment_unavailable" +# -- A stager that dies part-way must not leave its job queued for good --------------- + + +class _HardKill(BaseException): + """Stands in for a killed process: nothing in a ``finally`` runs when one dies.""" + + +_OWNER = "owner-a" +_STAGED = b"durable reference" + + +def _durable(tmp_path: Path) -> tuple[ExecutionRepository, ChatAttachmentService]: + repository = ExecutionRepository(tmp_path / "execution.sqlite", tmp_path / "artifacts") + return repository, ChatAttachmentService(repository) + + +def _stage(service: ChatAttachmentService, request_id: str, content: bytes = _STAGED): + return service.stage( + owner=_OWNER, request_id=request_id, filename="reference.txt", content=content + ) + + +def _queued_job(repository: ExecutionRepository, request_id: str, job_id: str): + """A chat attachment job whose stager is gone: created, never finished.""" + + payload = { + "filename": "reference.txt", + "mime_type": "text/plain", + "kind": "document", + "size": len(_STAGED), + "sha256": hashlib.sha256(_STAGED).hexdigest(), + } + job, _ = repository.create_job( + job_id=job_id, + owner=_OWNER, + request_id=request_id, + profile=CHAT_ATTACHMENT_PROFILE, + payload=payload, + ) + return job + + +def _stage_and_die_holding_the_lease(service, repository, monkeypatch, request_id: str) -> str: + """Run a stage that is killed mid-way; return the job it left behind.""" + + def killed(*_args, **_kwargs): + raise _HardKill + + with monkeypatch.context() as crash: + crash.setattr(repository, "publish_artifact", killed) + # A killed process never reaches its ``finally`` blocks. + crash.setattr(repository, "release_lease", lambda *_a, **_k: None) + with pytest.raises(_HardKill): + _stage(service, request_id) + (job,) = repository.list_jobs(owner=_OWNER) + return job.job_id + + +def test_the_staging_lease_is_held_while_bytes_are_staged_and_released_afterwards( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + repository, service = _durable(tmp_path) + real_publish = repository.publish_artifact + seen_during: list[str | None] = [] + + def observing(job_id, **kwargs): + seen_during.append(repository.lease_holder(job_id)) + return real_publish(job_id, **kwargs) + + monkeypatch.setattr(repository, "publish_artifact", observing) + _stage(service, "lease-held") + + (job,) = repository.list_jobs(owner=_OWNER, include_terminal=True) + assert len(seen_during) == 1 and seen_during[0] is not None + assert job.status == "succeeded" + assert repository.lease_holder(job.job_id) is None + + +def test_the_staging_lease_is_released_and_the_job_failed_when_staging_fails( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + repository, service = _durable(tmp_path) + + def refuses(*_args, **_kwargs): + raise OSError(28, "No space left on device") + + monkeypatch.setattr(repository, "publish_artifact", refuses) + with pytest.raises(ChatAttachmentError) as failure: + _stage(service, "lease-failed") + + assert failure.value.code == "attachment_persist_failed" + (job,) = repository.list_jobs(owner=_OWNER, include_terminal=True) + assert (job.status, job.error) == ("failed", "attachment_persist_failed") + assert repository.lease_holder(job.job_id) is None + + +def test_a_lease_that_cannot_be_taken_fails_the_job_instead_of_leaving_it_queued( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + repository, service = _durable(tmp_path) + + def refuses(*_args, **_kwargs): + raise ExecutionRepositoryError("SQLite execution operation failed.") + + monkeypatch.setattr(repository, "claim_lease", refuses) + with pytest.raises(ChatAttachmentError) as failure: + _stage(service, "lease-refused") + + assert failure.value.code == "attachment_persist_failed" + (job,) = repository.list_jobs(owner=_OWNER, include_terminal=True) + assert job.status == "failed" + with repository.connect() as connection: + assert connection.execute("SELECT COUNT(*) FROM execution_artifacts").fetchone()[0] == 0 + + +def test_a_stager_that_died_holding_its_lease_is_found_by_recovery( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, frozen_clock +): + repository, service = _durable(tmp_path) + job_id = _stage_and_die_holding_the_lease(service, repository, monkeypatch, "crash-recovery") + + assert repository.get_job(job_id).status == "queued" + assert repository.lease_holder(job_id) is not None + frozen_clock.advance(STAGING_LEASE_SECONDS + 1) + assert repository.recover_expired_leases() == [job_id] + + +def test_a_retry_after_a_crashed_stage_retires_the_job_instead_of_leaving_it_queued_forever( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, frozen_clock +): + """A kill between creating the job and finishing it left it queued for good. + + Nothing ever finished it, so it stayed in the task tray, and the request id + answered the same conflict on every retry. The retry now retires a job + nobody is working on; the answer is unchanged and stable. + """ + + repository, service = _durable(tmp_path) + job_id = _stage_and_die_holding_the_lease(service, repository, monkeypatch, "crash-retry") + frozen_clock.advance(ABANDONED_STAGE_SECONDS + 1) + + for _ in range(2): # the answer is stable, not a one-off + with pytest.raises(ChatAttachmentError) as retry: + _stage(service, "crash-retry") + assert retry.value.code == "attachment_request_conflict" + + job = repository.get_job(job_id) + assert job is not None + assert (job.status, job.error) == ("failed", ATTACHMENT_INTERRUPTED) + assert [event.event for event in repository.events(job_id)].count("failed") == 1 + assert repository.lease_holder(job_id) is None + assert repository.list_jobs(owner=_OWNER) == [] # no longer a task in the tray + + +def test_a_recent_or_leased_stage_is_left_alone_by_a_retry(tmp_path: Path, frozen_clock): + repository, service = _durable(tmp_path) + job = _queued_job(repository, "quiet", "job-quiet") + frozen_clock.advance(ABANDONED_STAGE_SECONDS - 5) + + with pytest.raises(ChatAttachmentError) as recent: + _stage(service, "quiet") + assert recent.value.code == "attachment_request_conflict" + assert repository.get_job(job.job_id).status == "queued" + + # Old enough to be abandoned, but another stager holds a live lease on it. + frozen_clock.advance(ABANDONED_STAGE_SECONDS) + repository.claim_lease(job.job_id, lease_owner="another-stager", ttl_seconds=STAGING_LEASE_SECONDS) + with pytest.raises(ChatAttachmentError): + _stage(service, "quiet") + assert repository.get_job(job.job_id).status == "queued" + + +def test_a_conflicting_retry_does_not_retire_the_job_it_conflicts_with(tmp_path: Path, frozen_clock): + repository, service = _durable(tmp_path) + job = _queued_job(repository, "conflict", "job-abandoned") + frozen_clock.advance(ABANDONED_STAGE_SECONDS * 2) + + with pytest.raises(ChatAttachmentError) as conflict: + _stage(service, "conflict", content=b"different bytes") + + assert conflict.value.code == "attachment_request_conflict" + assert repository.get_job(job.job_id).status == "queued" # only the matching request may retire it + + +def test_a_retry_never_changes_an_existing_success(tmp_path: Path, frozen_clock): + repository, service = _durable(tmp_path) + first = _stage(service, "done") + frozen_clock.advance(ABANDONED_STAGE_SECONDS * 10) + + again = _stage(service, "done") + + assert again == first + (job,) = repository.list_jobs(owner=_OWNER, include_terminal=True) + assert job.status == "succeeded" + + def test_api_reports_non_vision_models_and_returns_only_attachment_metadata(): state = FakeOllamaState( installed_models={"local-chat:7b"}, From beab8f2a4b8063d47bb5538f4da761b61a0bf3f9 Mon Sep 17 00:00:00 2001 From: Matthew Robert Wesney <157447210+dovvnloading@users.noreply.github.com> Date: Tue, 29 Sep 2026 04:12:36 -0400 Subject: [PATCH 07/11] feat(execution): record artifact and tombstone paths relative to the artifact root (schema version 4) A row recorded the absolute path it was written with, so moving the data directory left every existing row pointing at nothing: live attachments stopped reading, and expired ones were never reclaimed. Per-row containment already keeps one bad row from stopping retention; this is the rest. New artifact rows and new cleanup tombstones record their path relative to the artifact root (forward slashes). Both forms are read: a path column that is absolute is used as written, a relative one is joined to the current root, and the existing validators still confine either one, so a relative path that climbs out with .., names quarantine, or is drive-relative is refused exactly as an absolute one outside the root is. get_artifact still reports an absolute path. The change goes through the schema ladder as version 4, and nothing is rewritten on upgrade: version 3 rows keep their absolute paths and keep working, and no table changes. The version-3 step (installation principal and owner migration) now runs for stores below 3 only, as before; every older store is then stamped 4. Rollback: a build that only knows version 3 sets a version 4 store aside as "newer" (kept for inspection, then swept) and starts with an empty one, so job history and artifact rows are lost while the artifact files stay on disk for the sweep to reclaim once this build is back. Co-Authored-By: Claude Sonnet 5.5 --- .../cortex_backend/execution/repository.py | 58 ++- tests/test_allow_once_approval.py | 3 +- tests/test_execution_relative_paths.py | 394 ++++++++++++++++++ tests/test_execution_store.py | 6 +- tests/test_installation_principal.py | 10 +- 5 files changed, 455 insertions(+), 16 deletions(-) create mode 100644 tests/test_execution_relative_paths.py diff --git a/backend/cortex_backend/execution/repository.py b/backend/cortex_backend/execution/repository.py index 3456bef..efe2310 100644 --- a/backend/cortex_backend/execution/repository.py +++ b/backend/cortex_backend/execution/repository.py @@ -33,7 +33,14 @@ ) -SCHEMA_VERSION = 3 +# The schema ladder. 1-2: earlier builds. 3: every job is owned by the +# installation principal. 4: an artifact's path, and the paths of its cleanup +# tombstone, may be stored relative to the artifact root so a moved data +# directory keeps working; rows written before that keep their absolute paths +# and both forms are read (see ``ExecutionRepository._stored_path``). A build +# that only knows an earlier version sets a newer store aside instead of +# misreading it. +SCHEMA_VERSION = 4 MAX_EVENT_BYTES = 64 * 1024 MAX_APPROVAL_TTL_SECONDS = 300.0 # How long after the user's decision an approval can still be spent. The click @@ -614,7 +621,7 @@ def _ensure_schema_locked(self) -> None: # the file between that check and now. Refuse rather than # migrate it backwards. raise ExecutionRepositoryError("Execution schema is newer than this build.") - if current_version < SCHEMA_VERSION: + if current_version < 3: principal = self._ensure_installation_principal_connection(connection) ambiguous = connection.execute( """ @@ -633,6 +640,10 @@ def _ensure_schema_locked(self) -> None: "UPDATE execution_jobs SET owner = ? WHERE owner <> ?", (principal, principal), ) + # Version 4 changes what new rows may hold, not the tables and not + # the rows already there: nothing is rewritten, and a row that + # names its file by absolute path keeps working exactly as before. + if current_version < SCHEMA_VERSION: connection.execute( "UPDATE execution_schema SET version = ? WHERE id = 1", (SCHEMA_VERSION,), @@ -1751,7 +1762,7 @@ def publish_artifact( mime_type, len(content), digest, - str(target), + self._stored_text(target), now.isoformat(), expires.isoformat(), ), @@ -1814,7 +1825,7 @@ def get_artifact( mime_type=row["mime_type"], size=int(row["size"]), sha256=row["sha256"], - path=row["path"], + path=str(self._stored_path(row["path"])), created_at=row["created_at"], expires_at=row["expires_at"], ) @@ -1844,7 +1855,7 @@ def delete_artifact(self, artifact_id: str, *, owner: str | None = None) -> None ).fetchone() if row is None: return - path = self._validated_cleanup_path(Path(row["path"])) + path = self._validated_cleanup_path(self._stored_path(row["path"])) connection.execute( "DELETE FROM execution_artifacts WHERE artifact_id = ?", (artifact_id,), @@ -1879,7 +1890,7 @@ def read_artifact(self, artifact_id: str, *, owner: str | None = None) -> bytes: raise ExecutionRepositoryError("Artifact integrity check failed.") from None if not 0 <= expected_size <= self.max_artifact_bytes: raise ExecutionRepositoryError("Artifact integrity check failed.") - original_path = Path(row["path"]) + original_path = self._stored_path(row["path"]) if _is_reparse_point(original_path): raise ExecutionRepositoryError("Artifact path is unavailable.") try: @@ -1959,7 +1970,7 @@ def cleanup_expired( artifact_id = str(artifact_row["artifact_id"]) quarantine = self.quarantine_root / f"{artifact_id}-{uuid4().hex}.artifact" try: - path = self._validated_cleanup_path(Path(artifact_row["path"])) + path = self._validated_cleanup_path(self._stored_path(artifact_row["path"])) self._validated_quarantine_path(quarantine) self._record_artifact_cleanup(artifact_id, path, quarantine) except ArtifactCleanupRejected as exc: @@ -2021,6 +2032,31 @@ def cleanup_expired( skipped=skipped, ) + def _stored_path(self, text: str) -> Path: + """The location a row's path column names, under this build's artifact root. + + New rows record a path relative to the artifact root, so moving the + data directory does not orphan them; rows from before schema version 4 + hold an absolute path, and that is returned as written. Nothing here + decides whether the location is acceptable: an absolute path, a + relative one that climbs out with ``..`` and a drive-relative one all + come back as they resolve, and the validators that follow + (:meth:`_validated_cleanup_path`, :meth:`_validated_quarantine_path` + and the containment check in :meth:`read_artifact`) refuse any that + does not end up inside the root. + """ + + path = Path(text) + return path if path.is_absolute() else self.artifact_root / path + + def _stored_text(self, path: Path) -> str: + """How a row records ``path``: relative to the artifact root when it lies under it.""" + + try: + return path.relative_to(self.artifact_root).as_posix() + except ValueError: + return str(path) + def _validated_cleanup_path(self, path: Path) -> Path: """Validate a source path without following an untrusted reparse hop. @@ -2077,7 +2113,7 @@ def _record_artifact_cleanup( VALUES (?, ?, ?, 'pending', ?) ON CONFLICT(artifact_id) DO NOTHING """, - (artifact_id, str(path), str(quarantine), self._now()), + (artifact_id, self._stored_text(path), self._stored_text(quarantine), self._now()), ) def _resume_artifact_cleanup(self, *, limit: int) -> tuple[int, int]: @@ -2156,7 +2192,7 @@ def _reclaim_quarantined_file(self, artifact_id: str, quarantine_text: str) -> b """ try: - quarantine = self._validated_quarantine_path(Path(quarantine_text)) + quarantine = self._validated_quarantine_path(self._stored_path(quarantine_text)) except ArtifactCleanupBlocked: return False except ArtifactCleanupRejected: @@ -2221,8 +2257,8 @@ def _requeue_artifact_cleanup(self, artifact_id: str) -> None: def _advance_artifact_cleanup( self, artifact_id: str, path_text: str, quarantine_text: str, state: str ) -> int: - path = self._validated_cleanup_path(Path(path_text)) - quarantine = self._validated_quarantine_path(Path(quarantine_text)) + path = self._validated_cleanup_path(self._stored_path(path_text)) + quarantine = self._validated_quarantine_path(self._stored_path(quarantine_text)) if state not in {"pending", "quarantined", "finalized"}: raise ArtifactCleanupRejected("Artifact cleanup state is invalid.") removed = 0 diff --git a/tests/test_allow_once_approval.py b/tests/test_allow_once_approval.py index b63f6c9..d9b68a4 100644 --- a/tests/test_allow_once_approval.py +++ b/tests/test_allow_once_approval.py @@ -29,6 +29,7 @@ from cortex_backend.execution.models import ExecutionJob from cortex_backend.execution.repository import ( APPROVAL_GRANT_SECONDS, + SCHEMA_VERSION, ApprovalExpiredError, ApprovalTransitionError, ExecutionRepository, @@ -355,7 +356,7 @@ def test_the_use_record_is_additive_and_is_removed_with_its_job(tmp_path: Path) with reopened.connect() as connection: assert connection.execute( "SELECT version FROM execution_schema WHERE id = 1" - ).fetchone()[0] == 3 + ).fetchone()[0] == SCHEMA_VERSION job, _ = _approved_job(reopened) reopened.claim_approved_lease(job.job_id, lease_owner="worker", ttl_seconds=30) diff --git a/tests/test_execution_relative_paths.py b/tests/test_execution_relative_paths.py new file mode 100644 index 0000000..807019e --- /dev/null +++ b/tests/test_execution_relative_paths.py @@ -0,0 +1,394 @@ +"""Artifact paths are stored relative to the artifact root, additively. + +A row used to record the absolute path it was written with, so moving the data +directory made every existing row point at nothing: live attachments stopped +reading and expired ones were never reclaimed. Schema version 4 lets new rows +record their path relative to the artifact root. Nothing is rewritten on +upgrade -- rows written before that keep their absolute path -- and both forms +are read. +""" + +from __future__ import annotations + +import os +from pathlib import Path, PurePosixPath +import shutil + +import pytest + +from cortex_backend.execution.repository import ( + SCHEMA_VERSION, + ExecutionRepository, + ExecutionRepositoryError, +) + +_LONG_AGO = "2000-01-01T00:00:00+00:00" + + +def _open(data: Path) -> ExecutionRepository: + return ExecutionRepository(data / "execution.sqlite", data / "artifacts") + + +def _job(repository: ExecutionRepository, job_id: str) -> str: + repository.create_job( + job_id=job_id, + owner=repository.installation_principal_id, + request_id=f"request-{job_id}", + profile="scratch.auto.v1", + payload={}, + ) + repository.transition(job_id, status="succeeded", event="completed") + return job_id + + +def _publish(repository: ExecutionRepository, job_id: str, name: str = "out.txt", *, expired: bool = False): + artifact = repository.publish_artifact( + _job(repository, job_id), name=name, content=b"synthetic", mime_type="text/plain" + ) + if expired: + with repository.connect() as connection: + connection.execute( + "UPDATE execution_artifacts SET expires_at = ? WHERE artifact_id = ?", + (_LONG_AGO, artifact.artifact_id), + ) + return artifact + + +def _cleanup(repository: ExecutionRepository): + """One retention pass at the real time: only rows made expired on purpose are due.""" + + return repository.cleanup_expired() + + +def _rows(repository: ExecutionRepository, sql: str, *params: object): + with repository.connect() as connection: + return connection.execute(sql, params).fetchall() + + +def _artifact_row(repository: ExecutionRepository, artifact_id: str): + with repository.connect() as connection: + return connection.execute( + "SELECT path FROM execution_artifacts WHERE artifact_id = ?", (artifact_id,) + ).fetchone() + + +def _tombstones(repository: ExecutionRepository) -> list[tuple[str, str, str]]: + with repository.connect() as connection: + return [ + (row["path"], row["quarantine_path"], row["state"]) + for row in connection.execute( + "SELECT path, quarantine_path, state FROM execution_artifact_cleanup ORDER BY artifact_id" + ).fetchall() + ] + + +def _leave_a_pending_tombstone(repository: ExecutionRepository, monkeypatch: pytest.MonkeyPatch) -> None: + """Run a cleanup whose move into quarantine is refused, so the tombstone stays pending.""" + + def refused(*_args: object, **_kwargs: object) -> Path: + raise PermissionError(13, "held open by a scanner") + + with monkeypatch.context() as blocked: + blocked.setattr(Path, "replace", refused) + result = _cleanup(repository) + assert (result.artifacts, result.skipped) == (0, 1) + + +def _as_version_3_wrote_it(repository: ExecutionRepository) -> None: + """Rewrite every path the way the previous build recorded it: absolute, at version 3.""" + + root = repository.artifact_root + with repository.connect() as connection: + for row in connection.execute("SELECT artifact_id, path FROM execution_artifacts").fetchall(): + connection.execute( + "UPDATE execution_artifacts SET path = ? WHERE artifact_id = ?", + (str(root / row["path"]), row["artifact_id"]), + ) + for row in connection.execute( + "SELECT artifact_id, path, quarantine_path FROM execution_artifact_cleanup" + ).fetchall(): + connection.execute( + "UPDATE execution_artifact_cleanup SET path = ?, quarantine_path = ? WHERE artifact_id = ?", + (str(root / row["path"]), str(root / row["quarantine_path"]), row["artifact_id"]), + ) + connection.execute("UPDATE execution_schema SET version = 3 WHERE id = 1") + + +def _version(repository: ExecutionRepository) -> int: + with repository.connect() as connection: + return int(connection.execute("SELECT version FROM execution_schema WHERE id = 1").fetchone()[0]) + + +# -- A store written by this build ------------------------------------------------------------- + + +def test_new_rows_name_their_files_relative_to_the_artifact_root(tmp_path): + repository = _open(tmp_path / "data") + artifact = _publish(repository, "job-1", "out.txt") + + stored = _artifact_row(repository, artifact.artifact_id)["path"] + + assert stored == f"job-1/{Path(artifact.path).name}" + assert not os.path.isabs(stored) and ":" not in stored and "\\" not in stored + # What a caller is handed is still a real location, as it always was. + assert Path(artifact.path) == repository.artifact_root / "job-1" / Path(artifact.path).name + reread = repository.get_artifact(artifact.artifact_id) + assert reread is not None and reread.path == artifact.path + assert repository.read_artifact(artifact.artifact_id) == b"synthetic" + + +def test_a_cleanup_tombstone_records_relative_paths(tmp_path, monkeypatch): + repository = _open(tmp_path / "data") + artifact = _publish(repository, "job-1", expired=True) + + _leave_a_pending_tombstone(repository, monkeypatch) + + ((path, quarantine, state),) = _tombstones(repository) + assert state == "pending" + assert path == f"job-1/{Path(artifact.path).name}" + assert PurePosixPath(quarantine).parent.as_posix() == ".quarantine" + assert quarantine.endswith(".artifact") and not os.path.isabs(quarantine) + assert Path(artifact.path).is_file(), "the refused move left the file where it was" + + result = _cleanup(repository) # the move is allowed again + assert (result.artifacts, result.skipped) == (1, 0) + assert not Path(artifact.path).exists() + assert _tombstones(repository) == [] + + +def test_a_moved_data_directory_keeps_live_and_expiring_artifacts_working(tmp_path, monkeypatch): + first = tmp_path / "data" + repository = _open(first) + live = _publish(repository, "job-live", "live.txt") + stuck = _publish(repository, "job-stuck", "stuck.txt", expired=True) + _leave_a_pending_tombstone(repository, monkeypatch) + waiting = _publish(repository, "job-waiting", "waiting.txt", expired=True) + old_location = Path(stuck.path) + + moved = tmp_path / "moved" + shutil.move(str(first), str(moved)) + reopened = _open(moved) + + # The live artifact still reads, and it reports where it is now. + assert reopened.read_artifact(live.artifact_id) == b"synthetic" + found = reopened.get_artifact(live.artifact_id) + assert found is not None and Path(found.path) == moved / "artifacts" / "job-live" / Path(live.path).name + # The pending tombstone and the expired row are both reclaimed, with nothing skipped. + result = _cleanup(reopened) + assert (result.artifacts, result.skipped) == (2, 0) + assert reopened.get_artifact(stuck.artifact_id) is None + assert reopened.get_artifact(waiting.artifact_id) is None + assert not (moved / "artifacts" / "job-stuck").exists() + assert not (moved / "artifacts" / "job-waiting").exists() + assert not old_location.exists() + assert reopened.read_artifact(live.artifact_id) == b"synthetic" + assert _tombstones(reopened) == [] + + +# -- A store written by the previous build ------------------------------------------------------ + + +def test_a_store_written_before_version_4_is_read_as_it_was_and_is_not_rewritten(tmp_path, monkeypatch): + data = tmp_path / "data" + repository = _open(data) + live = _publish(repository, "job-live", "live.txt") + stuck = _publish(repository, "job-stuck", "stuck.txt", expired=True) + _leave_a_pending_tombstone(repository, monkeypatch) + _as_version_3_wrote_it(repository) + assert _version(repository) == 3 + before = { + "artifacts": sorted( + (row["artifact_id"], row["path"]) + for row in _rows(repository, "SELECT artifact_id, path FROM execution_artifacts") + ), + "tombstones": _tombstones(repository), + } + assert all(os.path.isabs(path) for _id, path in before["artifacts"]) + + upgraded = _open(data) + + assert _version(upgraded) == SCHEMA_VERSION == 4 + after = { + "artifacts": sorted( + (row["artifact_id"], row["path"]) + for row in _rows(upgraded, "SELECT artifact_id, path FROM execution_artifacts") + ), + "tombstones": _tombstones(upgraded), + } + assert after == before, "the upgrade rewrote rows it had no need to touch" + # Both forms read, and an absolute row reports the path it holds. + assert upgraded.read_artifact(live.artifact_id) == b"synthetic" + found = upgraded.get_artifact(live.artifact_id) + assert found is not None and found.path == dict(before["artifacts"])[live.artifact_id] + result = _cleanup(upgraded) + assert (result.artifacts, result.skipped) == (1, 0) + assert upgraded.get_artifact(stuck.artifact_id) is None + assert not Path(stuck.path).exists() + assert _tombstones(upgraded) == [] + # A row written after the upgrade is relative, next to the absolute one. + fresh = _publish(upgraded, "job-fresh", "fresh.txt") + assert not os.path.isabs(_artifact_row(upgraded, fresh.artifact_id)["path"]) + assert os.path.isabs(_artifact_row(upgraded, live.artifact_id)["path"]) + + +def test_reopening_a_version_4_store_changes_nothing(tmp_path): + data = tmp_path / "data" + repository = _open(data) + artifact = _publish(repository, "job-1") + before = _artifact_row(repository, artifact.artifact_id)["path"] + + again = _open(data) + + assert _version(again) == SCHEMA_VERSION + assert _artifact_row(again, artifact.artifact_id)["path"] == before + assert again.read_artifact(artifact.artifact_id) == b"synthetic" + + +def test_an_absolute_row_in_a_moved_data_directory_is_contained_as_it_always_was(tmp_path): + """The limitation is only for rows written before version 4: they are refused, not followed.""" + + first = tmp_path / "data" + repository = _open(first) + artifact = _publish(repository, "job-1", expired=True) + _as_version_3_wrote_it(repository) + moved = tmp_path / "moved" + shutil.move(str(first), str(moved)) + reopened = _open(moved) + survivor = moved / "artifacts" / "job-1" / Path(artifact.path).name + + with pytest.raises(ExecutionRepositoryError): + reopened.read_artifact(artifact.artifact_id) + result = _cleanup(reopened) + + assert (result.artifacts, result.skipped) == (0, 1) + assert reopened.get_artifact(artifact.artifact_id) is None # the unusable row is dropped + assert survivor.is_file() # and nothing outside a validated location was touched + + +# -- Failure paths: a relative row is confined exactly as an absolute one is ---------------------- + + +_ESCAPES = ( + "../outside.txt", + pytest.param( + "..\\outside.txt", + marks=pytest.mark.skipif(os.name != "nt", reason="a backslash is a separator only on Windows"), + ), + "job-1/../../outside.txt", + "./../outside.txt", +) + + +@pytest.mark.parametrize("stored", _ESCAPES) +def test_a_relative_artifact_row_that_climbs_out_of_the_root_is_refused_and_touches_nothing(tmp_path, stored): + repository = _open(tmp_path / "data") + outside = tmp_path / "data" / "outside.txt" + outside.write_bytes(b"not ours") + artifact = _publish(repository, "job-1", expired=True) + with repository.connect() as connection: + connection.execute( + "UPDATE execution_artifacts SET path = ? WHERE artifact_id = ?", (stored, artifact.artifact_id) + ) + + with pytest.raises(ExecutionRepositoryError): + repository.read_artifact(artifact.artifact_id) + with pytest.raises(ExecutionRepositoryError): + repository.delete_artifact(artifact.artifact_id) + assert outside.read_bytes() == b"not ours" + result = _cleanup(repository) + + assert (result.artifacts, result.skipped) == (0, 1) + assert outside.read_bytes() == b"not ours" + assert repository.get_artifact(artifact.artifact_id) is None + + +@pytest.mark.skipif(os.name != "nt", reason="drive-relative and rooted paths are a Windows form") +@pytest.mark.parametrize("form", ("rooted", "rooted-forward", "other-drive")) +def test_a_drive_relative_or_rooted_row_is_not_followed_out_of_the_root(tmp_path, form): + """Joining one of these to the root must not land anywhere but under it. + + A rooted path keeps only the root's drive, and a path on another drive + replaces the root altogether; either way it ends up outside and is refused. + """ + + other_drive = "Y:" if tmp_path.drive.upper() == "Z:" else "Z:" + stored = { + "rooted": "\\outside.txt", + "rooted-forward": "/outside.txt", + "other-drive": f"{other_drive}outside.txt", + }[form] + repository = _open(tmp_path / "data") + artifact = _publish(repository, "job-1", expired=True) + with repository.connect() as connection: + connection.execute( + "UPDATE execution_artifacts SET path = ? WHERE artifact_id = ?", (stored, artifact.artifact_id) + ) + + with pytest.raises(ExecutionRepositoryError): + repository.read_artifact(artifact.artifact_id) + result = _cleanup(repository) + + assert (result.artifacts, result.skipped) == (0, 1) + + +@pytest.mark.parametrize("stored", ("../outside.txt", ".quarantine/foreign.artifact", "job-1")) +def test_a_relative_tombstone_that_names_something_it_may_not_touch_is_discarded_untouched(tmp_path, stored): + """Outside the root, inside quarantine (which is not an artifact) or a directory.""" + + repository = _open(tmp_path / "data") + outside = tmp_path / "data" / "outside.txt" + outside.write_bytes(b"not ours") + foreign = repository.quarantine_root / "foreign.artifact" + foreign.write_bytes(b"belongs to another tombstone") + artifact = _publish(repository, "job-1", expired=True) + with repository.connect() as connection: + connection.execute( + """ + INSERT INTO execution_artifact_cleanup (artifact_id, path, quarantine_path, state, created_at) + VALUES (?, ?, '.quarantine/own.artifact', 'pending', ?) + """, + (artifact.artifact_id, stored, _LONG_AGO), + ) + + result = _cleanup(repository) + + assert result.skipped == 1 and result.artifacts == 0 + assert _tombstones(repository) == [] + assert outside.read_bytes() == b"not ours" + assert foreign.read_bytes() == b"belongs to another tombstone" + assert Path(artifact.path).parent.is_dir() + + +def test_a_relative_quarantine_path_outside_quarantine_is_never_unlinked(tmp_path): + repository = _open(tmp_path / "data") + victim = repository.artifact_root / "job-other" / "keep.txt" + victim.parent.mkdir() + victim.write_bytes(b"live artifact bytes") + artifact = _publish(repository, "job-1", expired=True) + with repository.connect() as connection: + connection.execute( + """ + INSERT INTO execution_artifact_cleanup (artifact_id, path, quarantine_path, state, created_at) + VALUES (?, ?, ?, 'finalized', ?) + """, + (artifact.artifact_id, f"job-1/{Path(artifact.path).name}", "job-other/keep.txt", _LONG_AGO), + ) + + _cleanup(repository) + + assert victim.read_bytes() == b"live artifact bytes" + + +# -- The two conversions ---------------------------------------------------------------------- + + +def test_the_path_conversions_round_trip_and_leave_foreign_absolute_paths_alone(tmp_path): + repository = _open(tmp_path / "data") + inside = repository.artifact_root / "job-1" / "file.bin" + foreign = tmp_path / "elsewhere" / "file.bin" + + assert repository._stored_text(inside) == "job-1/file.bin" + assert repository._stored_path("job-1/file.bin") == inside + assert repository._stored_text(foreign) == str(foreign) + assert repository._stored_path(str(foreign)) == foreign + assert repository._stored_path(repository._stored_text(inside)) == inside diff --git a/tests/test_execution_store.py b/tests/test_execution_store.py index 0735a70..0a79e22 100644 --- a/tests/test_execution_store.py +++ b/tests/test_execution_store.py @@ -14,6 +14,7 @@ from cortex_backend.execution.models import ExecutionJob from cortex_backend.execution.repository import ( + SCHEMA_VERSION, ArtifactLimitError, ExecutionIntegrityError, ExecutionRepository, @@ -37,7 +38,10 @@ def _repository(tmp_path): def test_durable_idempotency_event_replay_and_restart_recovery(tmp_path, frozen_clock): repository = _repository(tmp_path) with repository.connect() as connection: - assert connection.execute("SELECT version FROM execution_schema WHERE id = 1").fetchone()[0] == 3 + assert ( + connection.execute("SELECT version FROM execution_schema WHERE id = 1").fetchone()[0] + == SCHEMA_VERSION + ) tables = { row[0] for row in connection.execute( diff --git a/tests/test_installation_principal.py b/tests/test_installation_principal.py index e48722d..699eb06 100644 --- a/tests/test_installation_principal.py +++ b/tests/test_installation_principal.py @@ -11,7 +11,11 @@ from cortex_backend.testing import build_demo_dependencies from cortex_backend.api.security import SessionManager from cortex_backend.testing import DurableFakeCoordinator -from cortex_backend.execution.repository import ExecutionRepository, ExecutionRepositoryError +from cortex_backend.execution.repository import ( + SCHEMA_VERSION, + ExecutionRepository, + ExecutionRepositoryError, +) from support import session_headers as _session @@ -37,7 +41,7 @@ def test_installation_principal_is_atomic_persistent_and_migrates_additively(tmp with first.connect() as connection: assert connection.execute( "SELECT version FROM execution_schema WHERE id = 1" - ).fetchone()[0] == 3 + ).fetchone()[0] == SCHEMA_VERSION assert connection.execute( "SELECT principal_id FROM execution_installation_principal WHERE id = 1" ).fetchone()[0] == principal @@ -53,7 +57,7 @@ def test_installation_principal_is_atomic_persistent_and_migrates_additively(tmp with migrated.connect() as connection: assert connection.execute( "SELECT version FROM execution_schema WHERE id = 1" - ).fetchone()[0] == 3 + ).fetchone()[0] == SCHEMA_VERSION def test_installation_principal_creation_is_singleton_across_repository_instances(tmp_path): From e2a1c28882a3e5db603c8f0e2565ffe5aeac28e3 Mon Sep 17 00:00:00 2001 From: Matthew Robert Wesney <157447210+dovvnloading@users.noreply.github.com> Date: Tue, 29 Sep 2026 04:15:13 -0400 Subject: [PATCH 08/11] test(execution): run the junction sweep test only where the interpreter can recognise a junction Path.is_junction arrived in Python 3.12, so on 3.10 and 3.11 the reparse-point check cannot see a junction and the sweep relies on what lstat reports for one, which was only verified on 3.12 and 3.14. Skip the test there instead of asserting a behaviour that was not checked, and assert the junction is still present without calling is_junction. Co-Authored-By: Claude Sonnet 5.5 --- tests/test_artifact_reclamation.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/test_artifact_reclamation.py b/tests/test_artifact_reclamation.py index 3461631..a025285 100644 --- a/tests/test_artifact_reclamation.py +++ b/tests/test_artifact_reclamation.py @@ -320,6 +320,8 @@ def _make_junction(link: Path, target: Path) -> None: if os.name != "nt": pytest.skip("directory junctions exist only on Windows") + if not hasattr(Path, "is_junction"): + pytest.skip("this interpreter cannot recognise a junction (Path.is_junction arrived in 3.12)") import _winapi # type: ignore[import-not-found] try: @@ -349,7 +351,7 @@ def test_the_sweep_never_follows_or_removes_a_junction(tmp_path, name): assert (outside / f".tmp-{'e' * 32}").read_bytes() == b"not ours" assert (outside / "output").read_bytes() == b"not ours either" - assert junction.is_junction() + assert junction.exists(), "the junction itself was removed" finally: os.rmdir(junction) # removes the junction itself and nothing behind it From c09dd8f53ea4bf79fbdfeccd310eebf5e8b2322e Mon Sep 17 00:00:00 2001 From: Matthew Robert Wesney <157447210+dovvnloading@users.noreply.github.com> Date: Tue, 29 Sep 2026 05:11:29 -0400 Subject: [PATCH 09/11] fix(chat): a stager retired mid-stage no longer reports success and leaves its artifact behind _stage_created ignored the job that transition(..., "succeeded") returns. When a retry with the same request id retired the job as abandoned (the 60 s lease had lapsed and nothing had written for 120 s, as after a suspend/resume or a clock jump), the transition on the now-failed job wrote nothing and returned it, and the original call still handed back a descriptor. The published artifact row stayed behind with no result naming it, and resolve() then raised attachment_integrity_failed at send time. Capture the returned job. When it is not succeeded, remove the published artifact (best effort, never replacing the error and never carrying a path) and raise ChatAttachmentError("attachment_persist_failed"), the error the caller already maps. execution/attachment_staging.py has the same guard. Tests retire the job between publish and completion, assert the error, the failed / attachment_interrupted job with no result, and no artifact row or file; and that a refused removal neither replaces the error nor leaks the store location. Co-Authored-By: Claude Sonnet 5.5 --- .../cortex_backend/services/attachments.py | 24 +++++- tests/test_chat_attachments.py | 85 +++++++++++++++++++ 2 files changed, 107 insertions(+), 2 deletions(-) diff --git a/backend/cortex_backend/services/attachments.py b/backend/cortex_backend/services/attachments.py index d382269..90c2f42 100644 --- a/backend/cortex_backend/services/attachments.py +++ b/backend/cortex_backend/services/attachments.py @@ -422,7 +422,7 @@ def _stage_created( kind=kind, expires_at=artifact.expires_at, ) - repository.transition( + completed = repository.transition( job.job_id, status="succeeded", event="completed", @@ -430,10 +430,30 @@ def _stage_created( data={"message": "Chat attachment staged."}, result=descriptor.as_dict(), ) - return descriptor except (ExecutionRepositoryError, OSError) as exc: self._fail(job.job_id) raise ChatAttachmentError("attachment_persist_failed") from exc + if completed.status != "succeeded": + # A finished job is immutable: the write above changed nothing and + # only reported the job as it stands. That is what a retry that + # retired this job as abandoned leaves behind when the lease ran + # out while this call was still staging. Handing out a descriptor + # for a job the store says failed would leave the artifact behind + # with no result that names it, and resolve() would then refuse it + # at send time. + self._discard_artifact(artifact.artifact_id) + raise ChatAttachmentError("attachment_persist_failed") + return descriptor + + def _discard_artifact(self, artifact_id: str) -> None: + """Remove an artifact no job result names; best effort, because the caller is already raising.""" + + if self.repository is None: + return + try: + self.repository.delete_artifact(artifact_id) + except Exception: + pass # Retention reclaims the row and the file. def _fail(self, job_id: str) -> None: """Record a failed staging job; best effort, because the caller is already raising.""" diff --git a/tests/test_chat_attachments.py b/tests/test_chat_attachments.py index 216bca9..e1122d3 100644 --- a/tests/test_chat_attachments.py +++ b/tests/test_chat_attachments.py @@ -546,6 +546,91 @@ def test_a_retry_never_changes_an_existing_success(tmp_path: Path, frozen_clock) assert job.status == "succeeded" +def _retire_between_publish_and_completion( + repository: ExecutionRepository, + service: ChatAttachmentService, + frozen_clock, + monkeypatch: pytest.MonkeyPatch, + request_id: str, +) -> None: + """After the bytes are published, let the lease lapse and a retry retire the job. + + This is the suspend/resume (or clock jump) window: the stager holds its + bytes, its 60 s lease is gone and nothing has written for longer than the + abandonment threshold, so the matching retry fails the job as interrupted. + """ + + real_publish = repository.publish_artifact + + def publish_then_get_retired(job_id, **kwargs): + artifact = real_publish(job_id, **kwargs) + frozen_clock.advance(ABANDONED_STAGE_SECONDS + STAGING_LEASE_SECONDS) + with pytest.raises(ChatAttachmentError) as retry: + _stage(service, request_id) + assert retry.value.code == "attachment_request_conflict" + return artifact + + monkeypatch.setattr(repository, "publish_artifact", publish_then_get_retired) + + +def _artifact_rows_and_files(repository: ExecutionRepository) -> tuple[int, list[Path]]: + with repository.connect() as connection: + rows = connection.execute("SELECT COUNT(*) FROM execution_artifacts").fetchone()[0] + files = [path for path in repository.artifact_root.rglob("*") if path.is_file()] + return rows, files + + +def test_a_stager_retired_between_publish_and_completion_fails_and_leaves_no_artifact( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, frozen_clock +): + """The call that lost its job must not report success for it. + + Finishing a job the store already failed writes nothing and returns the + failed job. Ignoring that returned a descriptor whose artifact no job result + named, and ``resolve`` then refused it at send time. + """ + + repository, service = _durable(tmp_path) + _retire_between_publish_and_completion( + repository, service, frozen_clock, monkeypatch, "retired-mid-stage" + ) + + with pytest.raises(ChatAttachmentError) as lost: + _stage(service, "retired-mid-stage") + + assert lost.value.code == "attachment_persist_failed" + (job,) = repository.list_jobs(owner=_OWNER, include_terminal=True) + assert (job.status, job.error, job.result) == ("failed", ATTACHMENT_INTERRUPTED, None) + assert [event.event for event in repository.events(job.job_id)].count("completed") == 0 + assert repository.lease_holder(job.job_id) is None + rows, files = _artifact_rows_and_files(repository) + assert rows == 0, "the published artifact row was left behind" + assert files == [], "the published artifact file was left behind" + + +def test_a_failed_removal_of_the_orphaned_artifact_never_replaces_the_staging_error( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch, frozen_clock +): + repository, service = _durable(tmp_path) + _retire_between_publish_and_completion(repository, service, frozen_clock, monkeypatch, "retired-stuck") + secret_location = str(tmp_path / "artifacts") + + def refuses(*_args, **_kwargs): + raise ExecutionRepositoryError(f"Artifact cleanup failed at {secret_location}.") + + monkeypatch.setattr(repository, "delete_artifact", refuses) + with pytest.raises(ChatAttachmentError) as lost: + _stage(service, "retired-stuck") + + assert lost.value.code == "attachment_persist_failed" + assert secret_location not in str(lost.value) + assert lost.value.__cause__ is None, "the failure text would carry the store's location" + (job,) = repository.list_jobs(owner=_OWNER, include_terminal=True) + assert (job.status, job.error) == ("failed", ATTACHMENT_INTERRUPTED) + # Nothing removed it, so it waits for retention; no job result names it either. + assert _artifact_rows_and_files(repository)[0] == 1 + + def test_api_reports_non_vision_models_and_returns_only_attachment_metadata(): state = FakeOllamaState( installed_models={"local-chat:7b"}, From 09ac69f26950ae60ee0a8dcafa302d52478708ac Mon Sep 17 00:00:00 2001 From: Matthew Robert Wesney <157447210+dovvnloading@users.noreply.github.com> Date: Tue, 29 Sep 2026 05:11:40 -0400 Subject: [PATCH 10/11] fix(execution): keep the execution schema at version 3 so a rollback keeps the store Recording artifact paths relative to the artifact root adds no table and no column, so it is not a schema step. Bumping the version to 4 made the previous release treat the store as newer than itself and set the whole file aside: every job and the installation principal were lost, and its own artifact-root sweep, run later, then removed every artifact file because no row named them. Leave SCHEMA_VERSION at 3 and the ladder as it was. New rows still record relative paths and the read path still accepts both forms. The previous build keeps every job, fails closed on reading a new artifact ("Artifact path is unavailable"), and discards only expired rows it cannot resolve; its sweep matches files to rows by artifact id, so a live file that still has a row is kept. Comments and docstrings that described a version-4 step now say so truthfully. The three tests that pinned the version are back to the literal 3, and the relative-path tests pin it against a constant spelled out in the test. A new test opens a version-3 store in which two owners share one request id and asserts the owner migration does not run again (it would raise "ambiguous" or rewrite owners), which also pins the ladder's step condition. Another opens a store holding absolute and relative rows and tombstones together. Co-Authored-By: Claude Sonnet 5.5 --- .../cortex_backend/execution/repository.py | 23 ++-- tests/test_allow_once_approval.py | 3 +- tests/test_execution_relative_paths.py | 109 +++++++++++++++--- tests/test_execution_store.py | 6 +- tests/test_installation_principal.py | 49 ++++++-- 5 files changed, 145 insertions(+), 45 deletions(-) diff --git a/backend/cortex_backend/execution/repository.py b/backend/cortex_backend/execution/repository.py index efe2310..e32dd1b 100644 --- a/backend/cortex_backend/execution/repository.py +++ b/backend/cortex_backend/execution/repository.py @@ -34,13 +34,16 @@ # The schema ladder. 1-2: earlier builds. 3: every job is owned by the -# installation principal. 4: an artifact's path, and the paths of its cleanup +# installation principal. An artifact's path, and the paths of its cleanup # tombstone, may be stored relative to the artifact root so a moved data -# directory keeps working; rows written before that keep their absolute paths -# and both forms are read (see ``ExecutionRepository._stored_path``). A build -# that only knows an earlier version sets a newer store aside instead of -# misreading it. -SCHEMA_VERSION = 4 +# directory keeps working. That is a change to what new rows hold, not to a +# table or a column, so it is not a version step: the version stays 3, rows +# written by an earlier build keep their absolute paths, and both forms are +# read (see ``ExecutionRepository._stored_path``). Leaving the version alone is +# what lets an earlier build open a store this one wrote: a build that only +# knows an older version sets a newer store aside, jobs and all, instead of +# reading it. +SCHEMA_VERSION = 3 MAX_EVENT_BYTES = 64 * 1024 MAX_APPROVAL_TTL_SECONDS = 300.0 # How long after the user's decision an approval can still be spent. The click @@ -621,7 +624,7 @@ def _ensure_schema_locked(self) -> None: # the file between that check and now. Refuse rather than # migrate it backwards. raise ExecutionRepositoryError("Execution schema is newer than this build.") - if current_version < 3: + if current_version < SCHEMA_VERSION: principal = self._ensure_installation_principal_connection(connection) ambiguous = connection.execute( """ @@ -640,10 +643,6 @@ def _ensure_schema_locked(self) -> None: "UPDATE execution_jobs SET owner = ? WHERE owner <> ?", (principal, principal), ) - # Version 4 changes what new rows may hold, not the tables and not - # the rows already there: nothing is rewritten, and a row that - # names its file by absolute path keeps working exactly as before. - if current_version < SCHEMA_VERSION: connection.execute( "UPDATE execution_schema SET version = ? WHERE id = 1", (SCHEMA_VERSION,), @@ -2036,7 +2035,7 @@ def _stored_path(self, text: str) -> Path: """The location a row's path column names, under this build's artifact root. New rows record a path relative to the artifact root, so moving the - data directory does not orphan them; rows from before schema version 4 + data directory does not orphan them; rows written by an earlier build hold an absolute path, and that is returned as written. Nothing here decides whether the location is acceptable: an absolute path, a relative one that climbs out with ``..`` and a drive-relative one all diff --git a/tests/test_allow_once_approval.py b/tests/test_allow_once_approval.py index d9b68a4..b63f6c9 100644 --- a/tests/test_allow_once_approval.py +++ b/tests/test_allow_once_approval.py @@ -29,7 +29,6 @@ from cortex_backend.execution.models import ExecutionJob from cortex_backend.execution.repository import ( APPROVAL_GRANT_SECONDS, - SCHEMA_VERSION, ApprovalExpiredError, ApprovalTransitionError, ExecutionRepository, @@ -356,7 +355,7 @@ def test_the_use_record_is_additive_and_is_removed_with_its_job(tmp_path: Path) with reopened.connect() as connection: assert connection.execute( "SELECT version FROM execution_schema WHERE id = 1" - ).fetchone()[0] == SCHEMA_VERSION + ).fetchone()[0] == 3 job, _ = _approved_job(reopened) reopened.claim_approved_lease(job.job_id, lease_owner="worker", ttl_seconds=30) diff --git a/tests/test_execution_relative_paths.py b/tests/test_execution_relative_paths.py index 807019e..c7622a6 100644 --- a/tests/test_execution_relative_paths.py +++ b/tests/test_execution_relative_paths.py @@ -2,10 +2,15 @@ A row used to record the absolute path it was written with, so moving the data directory made every existing row point at nothing: live attachments stopped -reading and expired ones were never reclaimed. Schema version 4 lets new rows -record their path relative to the artifact root. Nothing is rewritten on -upgrade -- rows written before that keep their absolute path -- and both forms -are read. +reading and expired ones were never reclaimed. New rows record their path +relative to the artifact root. This adds no table and no column, so the schema +version does not move: it stays 3, nothing is rewritten on upgrade -- rows an +earlier build wrote keep their absolute path -- and both forms are read. + +Leaving the version alone is what makes going back safe. A build that only +knows version 3 opens the same store, keeps every job, fails closed on reading +an artifact whose path it cannot resolve, and discards only the expired rows +it cannot reclaim; a version bump would have made it set the whole store aside. """ from __future__ import annotations @@ -22,6 +27,10 @@ ExecutionRepositoryError, ) +# The version the previous release build reads. It is spelled out, not taken from +# SCHEMA_VERSION, so that moving the constant fails these tests instead of following it. +_PREVIOUS_BUILDS_VERSION = 3 + _LONG_AGO = "2000-01-01T00:00:00+00:00" @@ -82,8 +91,14 @@ def _tombstones(repository: ExecutionRepository) -> list[tuple[str, str, str]]: ] -def _leave_a_pending_tombstone(repository: ExecutionRepository, monkeypatch: pytest.MonkeyPatch) -> None: - """Run a cleanup whose move into quarantine is refused, so the tombstone stays pending.""" +def _leave_a_pending_tombstone( + repository: ExecutionRepository, monkeypatch: pytest.MonkeyPatch, *, stuck: int = 1 +) -> None: + """Run a cleanup whose move into quarantine is refused, so the tombstone stays pending. + + ``stuck`` is how many tombstones the pass is expected to be unable to move: the + new one and any that an earlier call left behind. + """ def refused(*_args: object, **_kwargs: object) -> Path: raise PermissionError(13, "held open by a scanner") @@ -91,11 +106,11 @@ def refused(*_args: object, **_kwargs: object) -> Path: with monkeypatch.context() as blocked: blocked.setattr(Path, "replace", refused) result = _cleanup(repository) - assert (result.artifacts, result.skipped) == (0, 1) + assert (result.artifacts, result.skipped) == (0, stuck) -def _as_version_3_wrote_it(repository: ExecutionRepository) -> None: - """Rewrite every path the way the previous build recorded it: absolute, at version 3.""" +def _as_the_previous_build_wrote_it(repository: ExecutionRepository) -> None: + """Rewrite every path that is still relative the way the previous build recorded it: absolute.""" root = repository.artifact_root with repository.connect() as connection: @@ -111,7 +126,6 @@ def _as_version_3_wrote_it(repository: ExecutionRepository) -> None: "UPDATE execution_artifact_cleanup SET path = ?, quarantine_path = ? WHERE artifact_id = ?", (str(root / row["path"]), str(root / row["quarantine_path"]), row["artifact_id"]), ) - connection.execute("UPDATE execution_schema SET version = 3 WHERE id = 1") def _version(repository: ExecutionRepository) -> int: @@ -188,14 +202,14 @@ def test_a_moved_data_directory_keeps_live_and_expiring_artifacts_working(tmp_pa # -- A store written by the previous build ------------------------------------------------------ -def test_a_store_written_before_version_4_is_read_as_it_was_and_is_not_rewritten(tmp_path, monkeypatch): +def test_a_store_the_previous_build_wrote_is_read_as_it_was_and_is_not_rewritten(tmp_path, monkeypatch): data = tmp_path / "data" repository = _open(data) live = _publish(repository, "job-live", "live.txt") stuck = _publish(repository, "job-stuck", "stuck.txt", expired=True) _leave_a_pending_tombstone(repository, monkeypatch) - _as_version_3_wrote_it(repository) - assert _version(repository) == 3 + _as_the_previous_build_wrote_it(repository) + assert _version(repository) == _PREVIOUS_BUILDS_VERSION before = { "artifacts": sorted( (row["artifact_id"], row["path"]) @@ -207,7 +221,7 @@ def test_a_store_written_before_version_4_is_read_as_it_was_and_is_not_rewritten upgraded = _open(data) - assert _version(upgraded) == SCHEMA_VERSION == 4 + assert _version(upgraded) == _PREVIOUS_BUILDS_VERSION after = { "artifacts": sorted( (row["artifact_id"], row["path"]) @@ -225,32 +239,89 @@ def test_a_store_written_before_version_4_is_read_as_it_was_and_is_not_rewritten assert upgraded.get_artifact(stuck.artifact_id) is None assert not Path(stuck.path).exists() assert _tombstones(upgraded) == [] - # A row written after the upgrade is relative, next to the absolute one. + # A row written by this build is relative, next to the absolute one. fresh = _publish(upgraded, "job-fresh", "fresh.txt") assert not os.path.isabs(_artifact_row(upgraded, fresh.artifact_id)["path"]) assert os.path.isabs(_artifact_row(upgraded, live.artifact_id)["path"]) -def test_reopening_a_version_4_store_changes_nothing(tmp_path): +def test_a_store_holding_absolute_and_relative_rows_opens_and_serves_and_reclaims_both(tmp_path, monkeypatch): + """The store a rollback and a second upgrade leave: rows of both forms, live and expired, + and tombstones of both forms waiting in the same queue.""" + + data = tmp_path / "data" + repository = _open(data) + old_live = _publish(repository, "job-old-live", "old-live.txt") + old_expired = _publish(repository, "job-old-expired", "old-expired.txt", expired=True) + _leave_a_pending_tombstone(repository, monkeypatch) + _as_the_previous_build_wrote_it(repository) # everything so far is absolute + new_live = _publish(repository, "job-new-live", "new-live.txt") + new_expired = _publish(repository, "job-new-expired", "new-expired.txt", expired=True) + _leave_a_pending_tombstone(repository, monkeypatch, stuck=2) # the old tombstone is retried too + old, new = (old_live, old_expired), (new_live, new_expired) + assert all(os.path.isabs(_artifact_row(repository, a.artifact_id)["path"]) for a in old) + assert not any(os.path.isabs(_artifact_row(repository, a.artifact_id)["path"]) for a in new) + forms = sorted(os.path.isabs(path) for path, _quarantine, _state in _tombstones(repository)) + assert forms == [False, True], "one tombstone of each form is waiting" + before = sorted( + (row["artifact_id"], row["path"]) + for row in _rows(repository, "SELECT artifact_id, path FROM execution_artifacts") + ) + + reopened = _open(data) + + assert _version(reopened) == _PREVIOUS_BUILDS_VERSION + assert ( + sorted( + (row["artifact_id"], row["path"]) + for row in _rows(reopened, "SELECT artifact_id, path FROM execution_artifacts") + ) + == before + ) + for live in (old_live, new_live): + assert reopened.read_artifact(live.artifact_id) == b"synthetic" + result = _cleanup(reopened) + assert (result.artifacts, result.skipped) == (2, 0) + for gone in (old_expired, new_expired): + assert reopened.get_artifact(gone.artifact_id) is None + assert not Path(gone.path).exists() + assert _tombstones(reopened) == [] + for live in (old_live, new_live): + assert reopened.read_artifact(live.artifact_id) == b"synthetic" + + +def test_new_rows_leave_the_schema_version_where_the_previous_build_reads_it(tmp_path): + """Going back must not cost the store: a build that knows only version 3 keeps a version-3 store. + + A store stamped with a version an older build does not know is set aside whole, + jobs and installation principal included, so writing relative paths may not + move the stamp. + """ + + assert SCHEMA_VERSION == _PREVIOUS_BUILDS_VERSION data = tmp_path / "data" repository = _open(data) artifact = _publish(repository, "job-1") before = _artifact_row(repository, artifact.artifact_id)["path"] + assert not os.path.isabs(before) + assert _version(repository) == _PREVIOUS_BUILDS_VERSION again = _open(data) - assert _version(again) == SCHEMA_VERSION + assert _version(again) == _PREVIOUS_BUILDS_VERSION assert _artifact_row(again, artifact.artifact_id)["path"] == before assert again.read_artifact(artifact.artifact_id) == b"synthetic" + assert again.get_job("job-1") is not None + assert again.installation_principal_id == repository.installation_principal_id def test_an_absolute_row_in_a_moved_data_directory_is_contained_as_it_always_was(tmp_path): - """The limitation is only for rows written before version 4: they are refused, not followed.""" + """The limitation is only for rows an earlier build wrote: they are refused, not followed.""" first = tmp_path / "data" repository = _open(first) artifact = _publish(repository, "job-1", expired=True) - _as_version_3_wrote_it(repository) + _as_the_previous_build_wrote_it(repository) moved = tmp_path / "moved" shutil.move(str(first), str(moved)) reopened = _open(moved) diff --git a/tests/test_execution_store.py b/tests/test_execution_store.py index 0a79e22..0735a70 100644 --- a/tests/test_execution_store.py +++ b/tests/test_execution_store.py @@ -14,7 +14,6 @@ from cortex_backend.execution.models import ExecutionJob from cortex_backend.execution.repository import ( - SCHEMA_VERSION, ArtifactLimitError, ExecutionIntegrityError, ExecutionRepository, @@ -38,10 +37,7 @@ def _repository(tmp_path): def test_durable_idempotency_event_replay_and_restart_recovery(tmp_path, frozen_clock): repository = _repository(tmp_path) with repository.connect() as connection: - assert ( - connection.execute("SELECT version FROM execution_schema WHERE id = 1").fetchone()[0] - == SCHEMA_VERSION - ) + assert connection.execute("SELECT version FROM execution_schema WHERE id = 1").fetchone()[0] == 3 tables = { row[0] for row in connection.execute( diff --git a/tests/test_installation_principal.py b/tests/test_installation_principal.py index 699eb06..84fd25e 100644 --- a/tests/test_installation_principal.py +++ b/tests/test_installation_principal.py @@ -11,11 +11,7 @@ from cortex_backend.testing import build_demo_dependencies from cortex_backend.api.security import SessionManager from cortex_backend.testing import DurableFakeCoordinator -from cortex_backend.execution.repository import ( - SCHEMA_VERSION, - ExecutionRepository, - ExecutionRepositoryError, -) +from cortex_backend.execution.repository import ExecutionRepository, ExecutionRepositoryError from support import session_headers as _session @@ -41,7 +37,7 @@ def test_installation_principal_is_atomic_persistent_and_migrates_additively(tmp with first.connect() as connection: assert connection.execute( "SELECT version FROM execution_schema WHERE id = 1" - ).fetchone()[0] == SCHEMA_VERSION + ).fetchone()[0] == 3 assert connection.execute( "SELECT principal_id FROM execution_installation_principal WHERE id = 1" ).fetchone()[0] == principal @@ -57,7 +53,7 @@ def test_installation_principal_is_atomic_persistent_and_migrates_additively(tmp with migrated.connect() as connection: assert connection.execute( "SELECT version FROM execution_schema WHERE id = 1" - ).fetchone()[0] == SCHEMA_VERSION + ).fetchone()[0] == 3 def test_installation_principal_creation_is_singleton_across_repository_instances(tmp_path): @@ -118,6 +114,45 @@ def test_ambiguous_legacy_owner_migration_fails_closed_without_rewriting_jobs(tm assert owners == {"session-a", "session-b"} +def test_a_current_store_is_not_put_through_the_owner_migration_again(tmp_path): + """Two owners sharing a request id is legal at version 3; only a version-2 store is ambiguous. + + The step that folds every owner into the installation principal runs for a + store older than the current version and for no other. If it ran on every + open, this store would be refused as ambiguous, or its owners rewritten. + """ + + database = tmp_path / "execution.sqlite" + artifacts = tmp_path / "artifacts" + repository = ExecutionRepository(database, artifacts) + principal = repository.installation_principal_id + for job_id, owner in (("current-a", "session-a"), ("current-b", "session-b")): + repository.create_job( + job_id=job_id, + owner=owner, + request_id="same-request", + profile="fake.v1", + payload={"provider": "fake-v1"}, + ) + with repository.connect() as connection: + assert connection.execute( + "SELECT version FROM execution_schema WHERE id = 1" + ).fetchone()[0] == 3 + + reopened = ExecutionRepository(database, artifacts) # must not raise "ambiguous" + + assert reopened.installation_principal_id == principal + with reopened.connect() as connection: + owners = { + row["job_id"]: row["owner"] + for row in connection.execute("SELECT job_id, owner FROM execution_jobs").fetchall() + } + assert connection.execute( + "SELECT version FROM execution_schema WHERE id = 1" + ).fetchone()[0] == 3 + assert owners == {"current-a": "session-a", "current-b": "session-b"} + + def test_restart_reattaches_execution_to_same_installation_principal(tmp_path): database = tmp_path / "execution.sqlite" artifacts = tmp_path / "artifacts" From 464eb137562c362e7b37689debaf122349f7d8b1 Mon Sep 17 00:00:00 2001 From: Matthew Robert Wesney <157447210+dovvnloading@users.noreply.github.com> Date: Tue, 29 Sep 2026 05:11:40 -0400 Subject: [PATCH 11/11] test(execution): kill the sweep-resume and unscoped-lookup mutants found surviving Ignoring the "complete" flag in sweep_artifact_root still passed test_an_entry_cut_short_by_the_limit_is_resumed_not_skipped, because the total per pass is the same when the rest of an entry is skipped and picked up on the wrap. The test now asserts, pass by pass, that job-a is finished before job-b is touched. Dropping owner= from the get_artifact call in AttachmentStagingService._existing survived because only read_artifact's owner was pinned. Both calls are pinned now. Co-Authored-By: Claude Sonnet 5.5 --- tests/test_artifact_reclamation.py | 21 +++++++++++++-------- tests/test_attachment_staging.py | 26 +++++++++++++++++--------- 2 files changed, 30 insertions(+), 17 deletions(-) diff --git a/tests/test_artifact_reclamation.py b/tests/test_artifact_reclamation.py index a025285..9eca170 100644 --- a/tests/test_artifact_reclamation.py +++ b/tests/test_artifact_reclamation.py @@ -425,14 +425,19 @@ def test_an_entry_cut_short_by_the_limit_is_resumed_not_skipped(tmp_path): directory.mkdir() _stale_temporary_files(directory, count) - per_pass = [] - for _ in range(12): # bounded: six things to remove - per_pass.append(repository.sweep_artifact_root(limit=2)) - if not first.exists() and not second.exists(): - break - - assert per_pass == [2, 2, 2] - assert not first.exists() and not second.exists() + # Six things to remove, two per pass. A pass that ran out of allowance inside + # job-a must come back to job-a, not move on and leave the rest of it behind. + assert repository.sweep_artifact_root(limit=2) == 2 + assert len(list(first.iterdir())) == 1, "the first pass should have stopped inside job-a" + assert len(list(second.iterdir())) == 1 + + assert repository.sweep_artifact_root(limit=2) == 2 # job-a's last file, then job-a itself + assert not first.exists(), "job-a was not finished before the sweep went on to job-b" + assert len(list(second.iterdir())) == 1, "job-b was touched before job-a was finished" + + assert repository.sweep_artifact_root(limit=2) == 2 # job-b's file, then job-b itself + assert not second.exists() + assert repository.sweep_artifact_root(limit=2) == 0 def test_the_limit_bounds_staging_and_legacy_directories_as_well_as_job_directories(tmp_path): diff --git a/tests/test_attachment_staging.py b/tests/test_attachment_staging.py index 75c672d..05051f2 100644 --- a/tests/test_attachment_staging.py +++ b/tests/test_attachment_staging.py @@ -324,30 +324,38 @@ def test_a_retry_never_changes_the_payload_check_or_an_existing_success(executio assert conflict.value.code == "request_conflict" -def test_a_repeated_request_reads_the_staged_bytes_as_their_owner_never_unscoped( +def test_a_repeated_request_looks_up_and_reads_the_staged_artifact_as_its_owner_never_unscoped( execution_repository, monkeypatch ): - """``get_artifact`` is owner-scoped; the read that follows it must be too. + """Both repository calls that follow a repeated request are owner-scoped. - The two calls are separate queries, so an unscoped read would trust - whatever the first one saw. + The lookup and the read are separate queries, so either one left unscoped + would trust whatever an artifact id happens to name. """ repository, _boundary, service = _service_over(execution_repository) content = _image_bytes() service.stage(owner=OWNER, request_id="attach-owner", content=content) + real_get = repository.get_artifact real_read = repository.read_artifact - owners: list[str | None] = [] + looked_up_as: list[str | None] = [] + read_as: list[str | None] = [] - def recording(artifact_id, *, owner=None): - owners.append(owner) + def recording_get(artifact_id, *, owner=None): + looked_up_as.append(owner) + return real_get(artifact_id, owner=owner) + + def recording_read(artifact_id, *, owner=None): + read_as.append(owner) return real_read(artifact_id, owner=owner) - monkeypatch.setattr(repository, "read_artifact", recording) + monkeypatch.setattr(repository, "get_artifact", recording_get) + monkeypatch.setattr(repository, "read_artifact", recording_read) service.stage(owner=OWNER, request_id="attach-owner", content=content) - assert owners == [OWNER] + assert looked_up_as == [OWNER] + assert read_as == [OWNER] def test_a_stager_finishing_after_its_job_was_retired_reports_failure_and_drops_the_artifact(