fix: make each step of a run idempotent across a relaunch

vero/src/vero/candidate_repository/git.py

@@ -137,6 +137,28 @@ async def create(
                 raise CandidateRepositoryError(
                     result.stderr or "failed to initialize candidate repository"
                 )
+
+        # A writer killed mid-transaction (the run that died in finalization was
+        # SIGKILLed while capturing a candidate) leaves a stale
+        # refs/vero/.../<name>.lock behind, and every later update-ref on that
+        # ref dies with "Unable to create ...: File exists" on a lock nobody
+        # holds, which is unrecoverable without a human deleting the file.
+        # Sweeping is only safe here and nowhere else: create() runs once while
+        # the session repository is being opened, before this process has any
+        # writer of its own, so a lock present at this instant is left over from
+        # a dead process. The same sweep on the capture path would be much worse
+        # than the bug, because it could delete the lock a concurrent writer is
+        # actively holding and let two ref updates clobber each other. For the
+        # same reason it stays inside refs/vero/candidates/, and not the whole of
+        # refs/vero/: the repository's own refs and packed-refs.lock are vero's to
+        # leave alone, and refs/vero/export/ and refs/vero/incoming/ are transient
+        # refs held by `vero session export` and its import counterpart, which run
+        # outside the session lock and so can genuinely be holding a live lock at
+        # the moment another process opens the repository.
+        candidate_refs = repository_path / "refs" / "vero" / "candidates"
+        for stale_lock in sorted(candidate_refs.rglob("*.lock")):
+            stale_lock.unlink(missing_ok=True)
+
         records_path.mkdir(parents=True, exist_ok=True)
         instance = cls(
             root=root,

vero/src/vero/evaluation/engine.py

@@ -439,6 +439,41 @@ async def _execute_record(
                     # original, as BudgetStore.refund already does.
                     raise cancellation from refund_error
             raise
+        except Exception as unexpected:
+            # The same last line of defence, for the failure side. The evaluator
+            # converts everything the *backend* raises into the two typed errors
+            # above, but not everything it does itself: an OSError from the
+            # result-directory mkdir or the running-manifest write it performs
+            # before its own try block, or from a shielded _persist_failure
+            # inside its handlers, unwinds as a bare exception none of the
+            # handlers above match. The reservation then stays charged with no
+            # evaluation to show for it, so the budget is short by that amount
+            # for the rest of the run and every later evaluation is silently
+            # starved -- over a long search that is the difference between
+            # finishing and running out of budget mid-way.
+            #
+            # Only the refund is recoverable here: a bare exception carries no
+            # evaluation id, so there is no record to load, exactly as with the
+            # raw cancellation above. Refunding a cost the run may never have
+            # spent is bounded by the clamp in BudgetLedger.refund, which never
+            # restores more than the budget's own total.
+            if charged:
+                try:
+                    await asyncio.shield(
+                        self.budget_ledger.refund(
+                            backend_id,
+                            request.evaluation_set,
+                            cost,
+                            principal,
+                        )
+                    )
+                except Exception as refund_error:
+                    # Same guard as the handlers above, for the same reason:
+                    # chain the refund's own failure and re-raise the original so
+                    # the caller still sees what actually stopped the evaluation,
+                    # as BudgetStore.refund already does.
+                    raise unexpected from refund_error
+            raise
         # Shielded for the same reason as the two handlers above, and one more:
         # the budget was charged for an evaluation that has now actually run, so
         # a cancellation landing inside _record would leave the ledger counting
@@ -485,7 +520,7 @@ async def _execute_record(
                         )
                     )
                 except Exception as refund_error:
-                    # Same guard as the three handlers above, for the same
+                    # Same guard as the four handlers above, for the same
                     # reason: chain the refund's own failure and re-raise the
                     # original, as BudgetStore.refund already does.
                     raise failure from refund_error

vero/src/vero/evaluation/store/persistence.py

@@ -44,17 +44,43 @@ def _atomic_write_json(path: Path, value: Any) -> None:
     )
     temporary_path = Path(temporary_name)
     try:
-        with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
+        handle = os.fdopen(descriptor, "w", encoding="utf-8")
+    except BaseException:
+        # This is the only window where we still own the raw descriptor: once
+        # os.fdopen succeeds the file object owns it and closes it exactly once
+        # when the block below exits, so closing it again from a shared failure
+        # path was a double close. Three writers reach this helper concurrently
+        # through asyncio.to_thread, and on CPython that second close can land
+        # on an unrelated descriptor that has since been handed the same number.
+        os.close(descriptor)
+        temporary_path.unlink(missing_ok=True)
+        raise
+    try:
+        with handle:
             json.dump(value, handle, ensure_ascii=False, indent=2)
             handle.write("\n")
             handle.flush()
             os.fsync(handle.fileno())
         os.replace(temporary_path, path)
-    except BaseException:
+        # Fsyncing the file only makes its bytes durable; the rename that
+        # publishes them is a change to the parent directory, so a hard power
+        # loss or container kill right here can leave the old contents behind or
+        # no file at all. Every durable artifact vero has commits through this
+        # one helper (the evaluation manifest, the per-case checkpoints,
+        # database.json, budgets.json, the session manifest, and the disclosure
+        # ledger), so fsync the directory as well. Some platforms refuse to open
+        # a directory for fsync, and this is a durability upgrade rather than
+        # part of the write itself, so an OSError from either step is ignored
+        # instead of failing a write that already landed.
         try:
-            os.close(descriptor)
+            directory_descriptor = os.open(path.parent, os.O_RDONLY)
+            try:
+                os.fsync(directory_descriptor)
+            finally:
+                os.close(directory_descriptor)
         except OSError:
             pass
+    except BaseException:
         temporary_path.unlink(missing_ok=True)
         raise
 
@@ -447,8 +473,33 @@ def load_reconciled(
         """Load the index and repair it from canonical completed evaluations."""
 
         existed = database_path.exists()
-        database = cls.load_from_file(database_path) if existed else cls(id=database_id)
+        database: EvaluationDatabase | None = None
+        if existed:
+            try:
+                database = cls.load_from_file(database_path)
+            except Exception as error:
+                # A crash in the middle of a write leaves a truncated index, and
+                # raising here bricked the session for good: every later run of
+                # it died on load even though this file is only a cache of the
+                # per-evaluation manifests that get reconciled in below. Rebuild
+                # from those instead of refusing to start, and log it, since the
+                # operator otherwise has no way to tell that an index they will
+                # find rewritten on disk was ever damaged.
+                logger.warning(
+                    "Rebuilding unreadable evaluation database %s from the "
+                    "per-evaluation manifests: %s",
+                    database_path,
+                    error,
+                )
+                # Treat the damaged file as absent so the repaired index is
+                # written back even when no new evaluation appeared on disk.
+                existed = False
+        if database is None:
+            database = cls(id=database_id)
         if database.id != database_id:
+            # A readable index naming a different session is a real "wrong
+            # session" signal (a copied or crossed session directory), never a
+            # torn write, so it stays a hard error rather than a rebuild.
             raise ValueError(
                 f"evaluation database belongs to {database.id!r}, not {database_id!r}"
             )

vero/src/vero/gateway/inference.py

@@ -1066,8 +1066,56 @@ async def finish() -> None:
                 finally:
                     await asyncio.shield(finish())
 
+            # Step the generator once before handing it over, because a generator
+            # that has never run is a reservation nobody owns. StreamingResponse
+            # races stream_response against listen_for_disconnect inside one anyio
+            # task group, so a client that has already gone away cancels the scope
+            # while stream_response is still awaiting its first send(), and the
+            # body iterator is never stepped at all: the generator body never
+            # runs, its finally never runs, and store.complete() is never called.
+            # A never-started async generator is invisible to asyncio's asyncgen
+            # finalization too, so nothing else cleans up behind it. Each such
+            # disconnect silently burned one concurrency permit until reserve()
+            # waited on the semaphore forever, with no timeout and no traceback:
+            # just a gateway that stopped answering, which is how two runs died.
+            # One anext() leaves the generator suspended inside its try, so the
+            # permit belongs to a started generator whose finally does run when it
+            # is closed. Cost: the response headers now wait on the first SSE
+            # event, so time to first byte grows by one event.
+            stream = chunks()
+            head: list[bytes] = []
+            replay: BaseException | None = None
+            try:
+                head.append(await anext(stream))
+            except StopAsyncIteration:
+                # An upstream that ended without a single event: the generator has
+                # already run its finally, so the reservation is already settled
+                # and the body below is simply empty.
+                pass
+            except asyncio.CancelledError:
+                # The request itself is being torn down, so there is no client
+                # left to hand a partial stream to. finish() has already run.
+                raise
+            except BaseException as error:
+                # A first chunk that fails has to keep failing *inside* the
+                # response rather than out of this handler: raising here would
+                # turn the truncated 200 that client SDKs retry as a connection
+                # error into a 500 that none of them retry, which would quietly
+                # change which trials fail. finish() has already run in the
+                # generator's finally, so all this still owes the client is the
+                # same broken stream it saw before.
+                replay = error
+
+            async def primed() -> AsyncIterator[bytes]:
+                for chunk in head:
+                    yield chunk
+                if replay is not None:
+                    raise replay
+                async for chunk in stream:
+                    yield chunk
+
             return StreamingResponse(
-                chunks(),
+                primed(),
                 status_code=upstream.status_code,
                 headers=response_headers,
                 media_type="text/event-stream",

vero/src/vero/harbor/backend.py

@@ -432,6 +432,14 @@ async def export_case_resources(
         cache = Path(configured_cache)
         if not (cache / "index.json").is_file():
             cache.parent.mkdir(parents=True, exist_ok=True)
+            # Deliberately still an mkdtemp and not a deterministic sibling. A
+            # stable staging path would let a retry reclaim the tree a killed
+            # attempt abandoned, which is the idempotent thing to want, but it
+            # also means two processes over one configured cache path would
+            # rmtree each other's staging tree mid-materialization. mkdtemp is
+            # what keeps them apart, and losing a case-resource tree under
+            # another process's feet is a worse failure than leaving a stale
+            # `.<partition>.XXXXXXXX` directory behind to be swept later.
             temporary = Path(
                 tempfile.mkdtemp(
                     dir=cache.parent,
@@ -923,7 +931,24 @@ def _trial_artifacts(
                         text = payload.decode("utf-8")
                         sanitized = sanitize_text(text, self._secrets())
                         if sanitized != text:
-                            resolved.write_text(sanitized, encoding="utf-8")
+                            # Write a sibling and rename over the trial file. The
+                            # in-place write truncated the record first, so a crash
+                            # (or a full disk) part way through left a trial that
+                            # was neither the original nor the redacted version,
+                            # and the original was already unrecoverable. The
+                            # rename is atomic, so whoever reads the record next
+                            # sees exactly one of the two.
+                            redacting = resolved.with_name(
+                                f".{resolved.name}.vero-redacting"
+                            )
+                            redacting.write_text(sanitized, encoding="utf-8")
+                            # Carry the record's own mode over, which the in-place
+                            # write kept for free. The agent context copies these
+                            # files in and only ever strips write bits from them,
+                            # so a redacted trial left at this process's umask
+                            # could become one the optimizer cannot read.
+                            shutil.copymode(resolved, redacting)
+                            os.replace(redacting, resolved)
                 except (OSError, UnicodeDecodeError):
                     pass
                 media_type = mimetypes.guess_type(resolved.name)[0]
@@ -1532,16 +1557,51 @@ async def evaluate(
                             "provisioning; check that every ancestor directory is "
                             "traversable by the dropped user"
                         )
-                result = await context.workspace.sandbox.run(
-                    command,
-                    cwd=context.workspace.project_path,
-                    timeout=request.limits.timeout_seconds,
-                    env=self._environment(
-                        context.evaluation_id, finalization=context.finalization
-                    ),
-                    run_as=self.config.harness_user,
-                )
-                await staging.download("jobs", attempt_jobs_dir)
+                sub_run_completed = False
+                try:
+                    result = await context.workspace.sandbox.run(
+                        command,
+                        cwd=context.workspace.project_path,
+                        timeout=request.limits.timeout_seconds,
+                        env=self._environment(
+                            context.evaluation_id, finalization=context.finalization
+                        ),
+                        run_as=self.config.harness_user,
+                    )
+                    sub_run_completed = True
+                finally:
+                    # Salvage whatever trials the sub-run did finish before it
+                    # died. This download used to sit after the run, so anything
+                    # that raised (a sandbox that went away, or the cancellation
+                    # quiesce_agent_evaluations delivers at finalization) skipped
+                    # it and the staging area's teardown a few lines below deleted
+                    # the only copy of the completed trials: exactly the evidence
+                    # needed to diagnose the failure, and exactly the work a
+                    # resume would want to reuse. Shielded so a cancelled task
+                    # still gets them off the sandbox. The cost is that a failed
+                    # sub-run now pays one extra copy off the sandbox.
+                    try:
+                        await asyncio.shield(staging.download("jobs", attempt_jobs_dir))
+                    except BaseException:
+                        if sub_run_completed:
+                            # Nothing is unwinding, so this is the same download
+                            # failure that always propagated from here.
+                            raise
+                        # A salvage that fails on its own must never replace the
+                        # sub-run's own failure: a dead sandbox fails this copy
+                        # too, and that error is the symptom, not the cause the
+                        # caller needs to see. BaseException and not Exception
+                        # because the failure being unwound is very often a
+                        # cancellation, and a second cancellation delivered while
+                        # this await is parked would otherwise raise CancelledError
+                        # out of the `finally` and take the sub-run's own error
+                        # with it. Swallowing it here does not lose the cancel:
+                        # the task stays in its cancelling state and the next
+                        # await in the caller sees it again.
+                        logger.warning(
+                            "unable to salvage Harbor trials from a failed sub-run",
+                            exc_info=True,
+                        )
             stdout = self.sanitize_error(result.stdout)
             stderr = self.sanitize_error(result.stderr)
             attempts.append((result, stdout, stderr))