fix(execution): follow-ups from the review of #309, relative artifact paths, chat attachment lease

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()

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",

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,28 +422,55 @@ def stage(
                 kind=kind,
                 expires_at=artifact.expires_at,
             )
-            self.repository.transition(
+            completed = repository.transition(
                 job.job_id,
                 status="succeeded",
                 event="completed",
                 phase="completed",
                 data={"message": "Chat attachment staged."},
                 result=descriptor.as_dict(),
             )
-            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
+        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."""
+
+        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)
@@ -478,8 +553,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)

tests/test_artifact_reclamation.py

@@ -315,6 +315,47 @@ 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")
+    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:
+        _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.exists(), "the junction itself was removed"
+    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 = []
@@ -340,6 +381,103 @@ 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)
+
+    # 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):
+    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")