diff --git a/src/agent_harness/api.py b/src/agent_harness/api.py index 370af31..0e9af09 100644 --- a/src/agent_harness/api.py +++ b/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( diff --git a/src/agent_harness/executor.py b/src/agent_harness/executor.py index 3562c0d..213d42e 100644 --- a/src/agent_harness/executor.py +++ b/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. diff --git a/src/agent_harness/fleet.py b/src/agent_harness/fleet.py index 705152e..b2ba064 100644 --- a/src/agent_harness/fleet.py +++ b/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. diff --git a/src/agent_harness/work.py b/src/agent_harness/work.py index 04cc6df..02f53ad 100644 --- a/src/agent_harness/work.py +++ b/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: diff --git a/tests/test_queue_lifecycle.py b/tests/test_queue_lifecycle.py new file mode 100644 index 0000000..b2cec03 --- /dev/null +++ b/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") == []