fix: hand work back properly, and let go of what is gone

src/agent_harness/api.py

@@ -343,7 +343,7 @@ def retry(
                 detail=f"{item_id} is claimed by {record.owner} and its lease is live; "
                 "wait for the lease to expire rather than racing it",
             )
-        queue.release(item_id, PENDING, error=None, project_id=project_id)
+        queue.requeue(item_id, project_id=project_id)
         return RetryResult(ok=True, item_id=item_id, state="pending")
 
     @app.post(

src/agent_harness/executor.py

@@ -783,6 +783,21 @@ def _execute(self, record: WorkRecord) -> Outcome:
         self._emit(record, "checks_passed")
         self._keepalive(record)
 
+        # The graph is re-checked here, at the last cheap point before review
+        # spends money and commit makes anything durable. `claim` checked it
+        # once, minutes ago; correcting a plan while work is in flight is a
+        # normal thing for an operator to do, and an item that is no longer
+        # eligible must not land on the strength of a stale check.
+        unmet = self.queue.unmet_dependencies(record.item_id, project_id=self.project_id)
+        if unmet:
+            outcome.reason = (
+                f"{record.item_id} now depends on {', '.join(unmet)}, which "
+                "is not done; the work is kept on its branch and the item goes back to pending"
+            )
+            self._emit(record, "dependency_invalidated", detail=outcome.reason)
+            outcome.state = PENDING
+            return outcome
+
         # 6. Review, by a different role -- and ideally a different vendor,
         #    which `ModelClient.reviewer_independence()` now reports on rather
         #    than leaving to a comment nobody reads.

src/agent_harness/fleet.py

@@ -117,6 +117,13 @@ def start(self, project_id: str) -> int:
                     return existing.size
 
             pool = ProjectPool(project_id=project_id)
+            # Before anything new is dispatched. A worker killed with its
+            # lease still running leaves an item `claimed` by a pid that no
+            # longer exists -- unavailable to healthy workers, with no session
+            # and nothing saying why, until the lease times out. Reclaiming it
+            # here is safe precisely because the process is provably gone.
+            for item_id in self.queue.reclaim_dead_workers(project_id=project_id):
+                log.info("reclaimed %s before starting project %s", item_id, project_id)
             # Set control BEFORE the threads exist. A worker that starts while
             # the project still reads `stopped` claims nothing and sleeps a
             # full poll for no reason.

src/agent_harness/work.py

@@ -259,6 +259,26 @@ def _run(self) -> None:
                 return
 
 
+def _process_alive(pid: int) -> bool:
+    """Whether a pid on this host is still running.
+
+    Signal 0 performs the permission and existence checks without delivering
+    anything. A pid we are not allowed to signal still exists, which is why
+    `PermissionError` is a yes.
+    """
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True
+    except OSError:
+        # Anything unexpected is treated as alive: releasing a claim from a
+        # live worker is far worse than leaving one for its lease to expire.
+        return True
+    return True
+
+
 def worker_identity() -> str:
     """Who holds a claim. Host and pid, so a stale claim can be traced to a
     specific process rather than to an anonymous 'someone'."""
@@ -835,6 +855,33 @@ def _dependencies_met(self, conn: sqlite3.Connection, record: WorkRecord) -> boo
                 return False
         return True
 
+    def unmet_dependencies(self, item_id: str, *, project_id: str = DEFAULT_PROJECT) -> list[str]:
+        """Dependencies of an item that are not done, right now.
+
+        `claim` checks this once, at the moment of claiming. The graph can be
+        corrected while an item is in flight -- that is what correcting a plan
+        looks like -- and an item that is no longer eligible must not go on to
+        pass a durable gate on the strength of a check made minutes earlier.
+        """
+        record = self.get(item_id, project_id=project_id)
+        if record is None:
+            return []
+        conn = self._connect()
+        try:
+            unmet = []
+            for dependency in record.depends_on:
+                row = conn.execute(
+                    "SELECT state FROM work WHERE project_id = ? AND item_id = ?",
+                    (project_id, dependency),
+                ).fetchone()
+                # Absent means tracked elsewhere, which is not a blocker --
+                # the same rule `claim` uses, for the same reason.
+                if row is not None and row["state"] != DONE:
+                    unmet.append(dependency)
+            return unmet
+        finally:
+            conn.close()
+
     def heartbeat(self, item_id: str, owner: str, project_id: str = DEFAULT_PROJECT) -> bool:
         """Extend the lease. Returns False if the claim was lost — which is
         the signal to stop working, because someone else now owns it."""
@@ -925,6 +972,75 @@ def release(
         finally:
             conn.close()
 
+    def requeue(self, item_id: str, *, project_id: str = DEFAULT_PROJECT) -> bool:
+        """Put an item back for another attempt, and mean it.
+
+        The attempt counter is reset, because an item at its ceiling is
+        retired by the very next claim scan: a "retry" that left the count
+        alone put the item back to `pending` and watched it return to
+        `exhausted` before any worker saw it, while reporting success.
+
+        `last_error` is kept. It is the only record of why the item failed,
+        it is what the retirement message appends to itself, and the operator
+        retrying an item is usually the person who needs to read it. Clearing
+        it turned a diagnosable failure into `gave up after N attempts` with
+        no cause.
+        """
+        conn = self._connect()
+        try:
+            cursor = conn.execute(
+                "UPDATE work SET state = ?, owner = NULL, lease_until = 0, attempts = 0, "
+                "updated_at = ? WHERE project_id = ? AND item_id = ?",
+                (PENDING, self.now(), project_id, item_id),
+            )
+            return cursor.rowcount > 0
+        finally:
+            conn.close()
+
+    def reclaim_dead_workers(self, *, project_id: str | None = None) -> list[str]:
+        """Release claims held by a process on this host that no longer exists.
+
+        A lease expiring is how a *crashed* worker's item comes back, and it
+        is deliberately slow so that a merely slow worker is not evicted. But
+        after a pool restart the old worker is provably gone -- its pid is not
+        running -- and waiting out its lease leaves an item stuck alongside
+        newly dispatched work, with no session and nothing saying why.
+
+        Only claims owned by *this host* are touched: a pid on another machine
+        says nothing about whether that process is alive.
+        """
+        here = socket.gethostname()
+        released: list[str] = []
+        for record in self.claimed(project_id=project_id):
+            owner = record.owner or ""
+            host, _, pid = owner.rpartition(":")
+            if host != here or not pid.isdigit():
+                continue
+            if _process_alive(int(pid)):
+                continue
+            self.release(
+                record.item_id,
+                PENDING,
+                error=f"worker {owner} is gone; the claim was released rather than waited out",
+                project_id=record.project_id,
+            )
+            released.append(record.item_id)
+            log.info("reclaimed %s from dead worker %s", record.item_id, owner)
+        return released
+
+    def claimed(self, project_id: str | None = None) -> list[WorkRecord]:
+        """Every item currently held by a worker, expired lease or not."""
+        conn = self._connect()
+        try:
+            sql = "SELECT * FROM work WHERE state = ?"
+            params: list[Any] = [CLAIMED]
+            if project_id is not None:
+                sql += " AND project_id = ?"
+                params.append(project_id)
+            return [WorkRecord.from_row(r) for r in conn.execute(sql, params)]
+        finally:
+            conn.close()
+
     def record_pr_url(
         self, item_id: str, pr_url: str, *, project_id: str = DEFAULT_PROJECT
     ) -> bool:

tests/test_queue_lifecycle.py

@@ -0,0 +1,159 @@
+"""Getting an item back, and letting go of one.
+
+Three ways the queue held on to work it should not have, or handed back work
+it had not really released:
+
+* retrying an exhausted item put it back and watched it be retired again,
+  while reporting success and erasing the reason;
+* a claim held by a process that no longer exists waited out its full lease
+  alongside newly dispatched work;
+* an item kept going through durable gates after the graph said it was no
+  longer eligible.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from agent_harness.work import (
+    CLAIMED,
+    DONE,
+    EXHAUSTED,
+    PENDING,
+    Project,
+    WorkQueue,
+    WorkRecord,
+)
+
+
+def rec(item_id: str, **kw: object) -> WorkRecord:
+    return WorkRecord(item_id=item_id, title=f"do {item_id}", brief="b", **kw)  # type: ignore[arg-type]
+
+
+def queue_at_the_ceiling(tmp_path: Path) -> WorkQueue:
+    """An item that has been given up on, the way the queue gives up."""
+    queue = WorkQueue(str(tmp_path / "w.sqlite"))
+    queue.add_project(Project(project_id="p", name="P", max_attempts=1))
+    queue.set_control("running", project_id="p")
+    queue.add([rec("T1")], project_id="p")
+    assert queue.claim("w", project_id="p") is not None
+    queue.release("T1", PENDING, error="checks failed: cargo clippy", project_id="p")
+    assert queue.claim("w", project_id="p") is None  # retires it
+    record = queue.get("T1", project_id="p")
+    assert record is not None and record.state == EXHAUSTED
+    return queue
+
+
+def test_retrying_an_exhausted_item_makes_it_claimable(tmp_path: Path) -> None:
+    """The regression for #126.
+
+    `release(..., PENDING)` left `attempts` at the ceiling, so the next claim
+    scan retired the item again before any worker saw it.
+    """
+    queue = queue_at_the_ceiling(tmp_path)
+
+    assert queue.requeue("T1", project_id="p")
+
+    claimed = queue.claim("worker", project_id="p")
+    assert claimed is not None, "a retried item must actually be claimable"
+    assert claimed.item_id == "T1"
+
+
+def test_retrying_keeps_the_reason_it_failed(tmp_path: Path) -> None:
+    """`error=None` erased the only record of why, which is what the operator
+    doing the retry most needs to read."""
+    queue = queue_at_the_ceiling(tmp_path)
+
+    queue.requeue("T1", project_id="p")
+
+    record = queue.get("T1", project_id="p")
+    assert record is not None
+    assert "cargo clippy" in (record.last_error or "")
+
+
+def test_a_claim_held_by_a_dead_process_is_reclaimed(tmp_path: Path) -> None:
+    """The regression for #104.
+
+    The lease is deliberately slow, so that a *slow* worker is not evicted.
+    A worker whose pid is gone is not slow, and waiting it out strands the
+    item alongside newly dispatched work with nothing saying why.
+    """
+    queue = WorkQueue(str(tmp_path / "w.sqlite"), lease_seconds=9999.0)
+    queue.add_project(Project(project_id="p", name="P"))
+    queue.set_control("running", project_id="p")
+    queue.add([rec("T1")], project_id="p")
+
+    import socket
+
+    dead = f"{socket.gethostname()}:999999"  # a pid that does not exist
+    assert queue.claim(dead, project_id="p") is not None
+    assert queue.get("T1", project_id="p").state == CLAIMED  # type: ignore[union-attr]
+
+    assert queue.reclaim_dead_workers(project_id="p") == ["T1"]
+
+    record = queue.get("T1", project_id="p")
+    assert record is not None
+    assert record.state == PENDING
+    assert "is gone" in (record.last_error or "")
+
+
+def test_a_live_worker_keeps_its_claim(tmp_path: Path) -> None:
+    """The other half, and the one that matters: reclaiming from a healthy
+    worker would hand its item to a second agent."""
+    queue = WorkQueue(str(tmp_path / "w.sqlite"), lease_seconds=9999.0)
+    queue.add_project(Project(project_id="p", name="P"))
+    queue.set_control("running", project_id="p")
+    queue.add([rec("T1")], project_id="p")
+
+    import socket
+
+    alive = f"{socket.gethostname()}:{os.getpid()}"
+    assert queue.claim(alive, project_id="p") is not None
+
+    assert queue.reclaim_dead_workers(project_id="p") == []
+    assert queue.get("T1", project_id="p").state == CLAIMED  # type: ignore[union-attr]
+
+
+def test_a_claim_on_another_host_is_never_reclaimed(tmp_path: Path) -> None:
+    """A pid on another machine says nothing about whether it is running."""
+    queue = WorkQueue(str(tmp_path / "w.sqlite"), lease_seconds=9999.0)
+    queue.add_project(Project(project_id="p", name="P"))
+    queue.set_control("running", project_id="p")
+    queue.add([rec("T1")], project_id="p")
+    assert queue.claim("some-other-host:999999", project_id="p") is not None
+
+    assert queue.reclaim_dead_workers(project_id="p") == []
+    assert queue.get("T1", project_id="p").state == CLAIMED  # type: ignore[union-attr]
+
+
+def test_unmet_dependencies_are_reported_for_work_in_flight(tmp_path: Path) -> None:
+    """The regression for #107.
+
+    `claim` checks the graph once. Correcting a plan while work is in flight
+    is normal, and the item must not pass a durable gate on a check made
+    minutes earlier.
+    """
+    queue = WorkQueue(str(tmp_path / "w.sqlite"))
+    queue.add_project(Project(project_id="p", name="P"))
+    queue.set_control("running", project_id="p")
+    queue.add([rec("A"), rec("B")], project_id="p")
+    assert queue.claim("w", project_id="p") is not None
+
+    # B is claimed and running when the operator corrects the graph.
+    queue.add([rec("B", depends_on=["A"])], project_id="p")
+
+    assert queue.unmet_dependencies("B", project_id="p") == ["A"]
+
+    queue.release("A", DONE, project_id="p")
+    assert queue.unmet_dependencies("B", project_id="p") == []
+
+
+def test_a_dependency_tracked_elsewhere_is_not_unmet(tmp_path: Path) -> None:
+    """Plans routinely reference work this queue does not hold; treating that
+    as a blocker would strand the item forever. Same rule `claim` uses."""
+    queue = WorkQueue(str(tmp_path / "w.sqlite"))
+    queue.add_project(Project(project_id="p", name="P"))
+    queue.add([rec("B", depends_on=["SOMEWHERE-ELSE"])], project_id="p")
+
+    assert queue.unmet_dependencies("B", project_id="p") == []