From 5e258cbb500a61bbba3c7ab1e9589178cfce18ff Mon Sep 17 00:00:00 2001 From: Paulo Date: Sun, 26 Jul 2026 00:40:55 +0200 Subject: [PATCH] The active board follows each item's driving run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Membership read work_items.status, a column only the shipped/closed paths wrote, so a build that crashed or was cancelled left the item on the board forever — counted by the header but matching neither lane. The board now derives from the same run state the lanes bucket by, and the header counts the lanes it renders. The newest run speaks for its subject: dedup gives a subject one active run per kind and the next starts only once the last is terminal, so preferring an active run over a newer one decided nothing once scope folded into build. created_at holds whole seconds, so the uuid7 id settles same-second ties. Subject.list_open() composes Run.open_subject_ids(), so an extension asks its own subject rather than the platform read side. Run.cancel announces run.cancelled — the body re-raises without emitting, so nothing had observed an operator cancel before. --- backend/druks/build/extension.py | 10 +-- backend/druks/build/subscribers.py | 9 ++ backend/druks/durable/enums.py | 3 + backend/druks/durable/models.py | 85 +++++++++++++++++-- backend/druks/durable/reads.py | 32 +++---- backend/druks/models.py | 20 ++++- backend/tests/test_build_board_membership.py | 83 ++++++++++++++++++ backend/tests/test_durable_schemas.py | 11 ++- backend/tests/test_generic_subjects.py | 4 +- .../src/extensions/build/WorkItemsPage.tsx | 8 +- 10 files changed, 216 insertions(+), 49 deletions(-) create mode 100644 backend/tests/test_build_board_membership.py diff --git a/backend/druks/build/extension.py b/backend/druks/build/extension.py index 8149b5a..407344d 100644 --- a/backend/druks/build/extension.py +++ b/backend/druks/build/extension.py @@ -175,13 +175,9 @@ def subject_summary(cls, subject: WorkItem) -> WorkItemSummary: @classmethod def list_subjects(cls) -> list[WorkItemSummary]: - # The active board: in-flight items only — handed-off ones live in History. - # The 500 most-recent cover it; paginate if a board outgrows it. - return [ - WorkItemSummary.from_work_item(item) - for item in WorkItem.list_recent(limit=500) - if not item.status - ] + # The active board: whatever hasn't handed off yet. The 500 most-recent + # cover it; paginate if a board outgrows it. + return [WorkItemSummary.from_work_item(item) for item in WorkItem.list_open(limit=500)] @classmethod async def subject_activity(cls, subject: WorkItem) -> SubjectActivity | None: diff --git a/backend/druks/build/subscribers.py b/backend/druks/build/subscribers.py index 230198e..8135033 100644 --- a/backend/druks/build/subscribers.py +++ b/backend/druks/build/subscribers.py @@ -38,6 +38,15 @@ async def review_park_marks_ticket_in_review(*, subject: WorkItem, **_: object) await subject.set_remote_status(TicketStatus.IN_REVIEW) +@subscribe("run.failed", kind=BuildWorkflow.kind, subject=WorkItem) +@subscribe("run.cancelled", kind=BuildWorkflow.kind, subject=WorkItem) +async def build_end_settles_the_item(*, subject: WorkItem, **_: object) -> None: + # Nothing merged, so the attempt was abandoned — unless the PR already spoke: + # ship() cancels the run it just shipped, and that cancel arrives here. + if not subject.status: + subject.set_status(HandoffStatus.CANCELLED) + + @subscribe("repo.pushed", to_default_branch=True) async def policy_push_reprofiles_the_repo(*, repo: str, paths: list, **_: object) -> None: # The operator edited the repo's build policy — re-apply it over the diff --git a/backend/druks/durable/enums.py b/backend/druks/durable/enums.py index 12bc9c2..17038e9 100644 --- a/backend/druks/durable/enums.py +++ b/backend/druks/durable/enums.py @@ -15,6 +15,9 @@ class RunState(StrEnum): ACTIVE_STATES = (RunState.SCHEDULED, RunState.RUNNING, RunState.PENDING_INPUT) TERMINAL_STATES = (RunState.FINISHED, RunState.FAILED, RunState.CANCELLED, RunState.ORPHANED) +# A subject whose newest run is in one of these is still open: going, or +# failed and wanting the operator. +OPEN_STATES = (*ACTIVE_STATES, RunState.FAILED) class AgentCallStatus(StrEnum): diff --git a/backend/druks/durable/models.py b/backend/druks/durable/models.py index 9b048dc..f5f6f5a 100644 --- a/backend/druks/durable/models.py +++ b/backend/druks/durable/models.py @@ -4,7 +4,7 @@ from typing import TYPE_CHECKING, Any, Literal from dbos import DBOS -from sqlalchemy import ForeignKey, Index, String, func, select, update +from sqlalchemy import ForeignKey, Index, Integer, Select, String, cast, func, select, update from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.dialects.postgresql import insert as pg_insert from sqlalchemy.orm import Mapped, column_property, mapped_column, relationship @@ -13,12 +13,18 @@ from druks.accounts.models import Account from druks.core.models import Uuid7Pk from druks.database import db_session, get_session -from druks.durable.dbos_state import state_expression, subject_filter, updated_at_expression -from druks.durable.enums import ACTIVE_STATES, AgentCallStatus, RunState +from druks.durable.dbos_state import ( + state_expression, + subject_filter, + updated_at_expression, + workflow_status, +) +from druks.durable.enums import ACTIVE_STATES, OPEN_STATES, AgentCallStatus, RunState from druks.harnesses.artifacts import normalize_token_usage from druks.models import Base from druks.notifications.models import Notification from druks.settings import load_settings +from druks.signals import publish if TYPE_CHECKING: from druks.sandbox.datastructures import AgentResult @@ -102,18 +108,63 @@ def list_for_subject( ) -> list["Run"]: # Every run about this subject (stamped at start), newest first — a # subject's lifecycle spans many runs, so its status is theirs - # aggregated. ``kind`` narrows to one workflow's runs; per (kind, - # subject) the queue dedup makes runs strictly sequential, so newest - # first holds within a kind too. + # aggregated. Ordered by start, not last touch: which run began most + # recently is fixed, where updated_at moves and could lift an older run + # over a newer one. created_at holds whole seconds, so the uuid7 id + # settles runs that started within the same one. ``kind`` narrows to one + # workflow's runs. stmt = ( select(cls) .where(subject_filter(cls.id, subject_type, subject_id)) - .order_by(cls.updated_at.desc()) + .order_by(cls.created_at.desc(), cls.id.desc()) ) if kind: stmt = stmt.where(cls.kind == kind) return list(db_session().scalars(stmt)) + @classmethod + def get_latest_for_subject( + cls, subject_type: str, subject_id: str, kind: str | None = None + ) -> "Run | None": + """The run that speaks for the subject: a subject holds at most one active + run per kind (queue dedup) and the next starts only once the last is + terminal, so the newest is the live one whenever anything is live.""" + stmt = ( + select(cls) + .where(subject_filter(cls.id, subject_type, subject_id)) + .order_by(cls.created_at.desc(), cls.id.desc()) + .limit(1) + ) + if kind: + stmt = stmt.where(cls.kind == kind) + return db_session().scalars(stmt).first() + + @classmethod + def open_subject_ids(cls, subject_type: str) -> Select: + """The subjects of ``subject_type`` whose newest run hasn't handed off — + a subquery their own read composes.""" + state = state_expression(cls.id, cls.input_gate, cls.created_at).label("state") + subject_id = workflow_status.c.attributes["subject_id"].as_string().label("subject_id") + driving = ( + select( + subject_id, + state, + func.row_number() + .over( + partition_by=subject_id, + order_by=(cls.created_at.desc(), cls.id.desc()), + ) + .label("rank"), + ) + .join_from(cls, workflow_status, workflow_status.c.workflow_uuid == cls.id) + .where(workflow_status.c.attributes["subject_type"].as_string() == subject_type) + .subquery() + ) + return select(cast(driving.c.subject_id, Integer)).where( + driving.c.rank == 1, + driving.c.state.in_([run_state.value for run_state in OPEN_STATES]), + ) + def get_ask(self) -> dict[str, Any]: # The parked ask, ready to serve. An in-app review's ask names neither # label nor artifact — a parked run can't produce new artifacts, so the @@ -191,6 +242,16 @@ async def resume(self, **fields: Any) -> None: idempotency_key=f"{self.input_gate}:{self.input_requested_at}", ) + @property + def subject(self) -> dict[str, str] | None: + # Stamped at start; a subjectless cron has none. + attributes = db_session().scalar( + select(workflow_status.c.attributes).where(workflow_status.c.workflow_uuid == self.id) + ) + if attributes: + return {"type": attributes["subject_type"], "id": attributes["subject_id"]} + return + async def cancel(self, *, failure: str | None = None) -> None: # Clear the ask (so nothing tries to answer it) and keep the operator's # reason, then cancel the DBOS workflow — that writes the CANCELLED @@ -203,6 +264,16 @@ async def cancel(self, *, failure: str | None = None) -> None: self.failure = failure db_session().flush() await DBOS.cancel_workflow_async(self.id) + # The body raises DBOSWorkflowCancelledError and re-raises without + # emitting, so the canceller announces the terminal state itself. + subject = self.subject + if subject: + await publish( + "run.cancelled", + subject=subject, + kind=self.kind, + failure=failure, + ) @dataclass(frozen=True) diff --git a/backend/druks/durable/reads.py b/backend/druks/durable/reads.py index 5a1679c..ffae9f8 100644 --- a/backend/druks/durable/reads.py +++ b/backend/druks/durable/reads.py @@ -47,9 +47,8 @@ def list_subject_timeline(subject_type: str, subject_id: str) -> list[RunRespons def get_subject_status( subject_type: str, subject_id: str, *, kind: str | None = None ) -> SubjectStatus: - runs = Run.list_for_subject(subject_type, subject_id, kind=kind) - active_run = next((run for run in runs if run.is_active), None) - return _status(runs, active_run, _running_calls(active_run)) + latest = Run.get_latest_for_subject(subject_type, subject_id, kind=kind) + return _status(latest, _running_calls(latest)) async def get_subject_phase(subject_type: str, subject_id: str) -> str | None: @@ -67,16 +66,12 @@ def get_subject_response( summary: SubjectSummary, activity: SubjectActivity | None = None, ) -> SubjectResponse: - # One fetch feeds both the status derivation and the timeline. runs = Run.list_for_subject(subject_type, subject_id) calls_by_run = AgentCall.by_run([run.id for run in runs]) - active_run = next((run for run in runs if run.is_active), None) - active_calls: list[AgentCall] = [] - if active_run: - active_calls = calls_by_run[active_run.id] + latest = Run.get_latest_for_subject(subject_type, subject_id) return SubjectResponse( summary=summary, - status=_status(runs, active_run, active_calls), + status=_status(latest, calls_by_run.get(latest.id, []) if latest else []), timeline=_timeline(runs, calls_by_run), activity=activity, ) @@ -95,22 +90,17 @@ def _timeline(runs: list[Run], calls_by_run: dict[str, list[AgentCall]]) -> list ] -def _running_calls(active_run: Run | None) -> list[AgentCall]: +def _running_calls(run: Run | None) -> list[AgentCall]: # A parked run's status carries its gate ask; only a running run surfaces its # latest agent call. So a parked board row never queries agent_calls — the # board runs this per subject. - if active_run and active_run.state != RunState.PENDING_INPUT.value: - return AgentCall.list_for_run(active_run.id) + if run and not run.is_parked: + return AgentCall.list_for_run(run.id) return [] -def _status( - runs: list[Run], active_run: Run | None, active_calls: list[AgentCall] -) -> SubjectStatus: - # The newest active run drives the subject — a stale parked run a fresh dispatch - # superseded must not outrank it; once all are terminal, the latest one's outcome - # stands. Facts only: the extension's UI renders its copy from them. - driving_run = active_run or (runs[0] if runs else None) +def _status(driving_run: Run | None, calls: list[AgentCall]) -> SubjectStatus: + # Facts only: the extension's UI renders its copy from them. if not driving_run: return SubjectStatus(state=RunState.SCHEDULED) # A pending_input run is always the active one (ACTIVE_STATES), so the @@ -121,8 +111,8 @@ def _status( # is the inverse: only a parked run's input_gate is a live ask (a timed-out # run keeps the stale column). agent = None - if active_calls and not parked: - agent = active_calls[-1].agent + if calls and not parked: + agent = calls[-1].agent return SubjectStatus( state=RunState(driving_run.state), kind=driving_run.kind, diff --git a/backend/druks/models.py b/backend/druks/models.py index 5704207..4793c35 100644 --- a/backend/druks/models.py +++ b/backend/druks/models.py @@ -1,8 +1,8 @@ import re from datetime import UTC, datetime -from typing import Any, ClassVar +from typing import Any, ClassVar, Self -from sqlalchemy import DateTime +from sqlalchemy import DateTime, select from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy.types import TypeDecorator @@ -57,3 +57,19 @@ def identity(self) -> dict[str, Any]: if self.id: return {"type": self.subject_type, "id": self.id} raise ValueError(f"unsaved {type(self).__name__} has no identity — flush it first") + + @classmethod + def list_open(cls, *, limit: int = 50) -> list[Self]: + """The rows whose newest run hasn't handed off — still going, or failed + and wanting the operator. What an extension's active view lists.""" + # Cycle: the durable read side is built on this module's Base. + from druks.database import db_session + from druks.durable.models import Run + + stmt = ( + select(cls) + .where(cls.id.in_(Run.open_subject_ids(cls.subject_type))) + .order_by(cls.id.desc()) + .limit(limit) + ) + return list(db_session().scalars(stmt)) diff --git a/backend/tests/test_build_board_membership.py b/backend/tests/test_build_board_membership.py new file mode 100644 index 0000000..49c424c --- /dev/null +++ b/backend/tests/test_build_board_membership.py @@ -0,0 +1,83 @@ +import druks.build.workflows # noqa: F401 # registers build.build_workflow, the seeded kind +import pytest +from conftest import make_test_work_item, seed_build_run +from druks.build.enums import HandoffStatus +from druks.build.extension import Build +from druks.build.models import WorkItem + + +def _board_ids(db_session): + db_session.expire_all() + return {row.id for row in Build.list_subjects()} + + +@pytest.mark.parametrize("state", ["scheduled", "running", "pending_input", "failed"]) +def test_a_live_or_failed_build_holds_the_board(db_session, state): + # A failed run still wants the operator, so it stays alongside the live ones. + item = make_test_work_item(repo="ClawHaven/acme-app", title=f"build {state}") + seed_build_run(db_session, work_item_id=item.id, state=state) + assert str(item.id) in _board_ids(db_session) + + +@pytest.mark.parametrize("state", ["finished", "cancelled"]) +def test_an_ended_build_leaves_the_board(db_session, state): + # The strand this fixes: membership used to read work_items.status, which a + # cancelled or errored run never wrote, so the item lingered forever. + item = make_test_work_item(repo="ClawHaven/acme-app", title=f"build {state}") + seed_build_run(db_session, work_item_id=item.id, state=state) + db_session.expire_all() + assert item.status is None + assert str(item.id) not in _board_ids(db_session) + + +def test_a_redispatched_item_returns_to_the_board(db_session): + # Its newest run drives it, so a fresh build outranks the handed-off one. + item = make_test_work_item(repo="ClawHaven/acme-app", title="rebuilt") + seed_build_run(db_session, work_item_id=item.id, state="finished") + seed_build_run(db_session, work_item_id=item.id, state="running") + assert str(item.id) in _board_ids(db_session) + + +def test_an_item_without_runs_stays_off_the_board(db_session): + item = make_test_work_item(repo="ClawHaven/acme-app", title="never dispatched") + assert str(item.id) not in _board_ids(db_session) + + +async def test_a_cancelled_build_settles_as_cancelled(db_session): + # Nothing merged, so History records the abandonment. + from druks.build.subscribers import build_end_settles_the_item + + item = make_test_work_item(repo="ClawHaven/acme-app", title="cancelled build") + await build_end_settles_the_item(subject=item) + assert item.status == HandoffStatus.CANCELLED + + +async def test_a_shipped_item_keeps_its_outcome(db_session): + # ship() lands first and owns the verdict; the run's own cancel must not + # overwrite it on the way out. + from druks.build.subscribers import build_end_settles_the_item + + item = make_test_work_item(repo="ClawHaven/acme-app", title="merged build") + item.set_status(HandoffStatus.SHIPPED) + await build_end_settles_the_item(subject=item) + assert item.status == HandoffStatus.SHIPPED + + +def test_history_holds_the_handed_off(db_session): + shipped = make_test_work_item(repo="ClawHaven/acme-app", title="shipped") + shipped.set_status(HandoffStatus.SHIPPED) + make_test_work_item(repo="ClawHaven/acme-app", title="still open") + assert [item.id for item in WorkItem.list_handoff()] == [shipped.id] + + +def test_the_newest_run_speaks_for_the_item(db_session): + # One rule, two implementations — the bulk board query and the per-subject + # status must name the same driving run or the board and its lanes disagree. + from druks.workflows import get_subject_status + + item = make_test_work_item(repo="ClawHaven/acme-app", title="two runs") + seed_build_run(db_session, work_item_id=item.id, state="cancelled") + seed_build_run(db_session, work_item_id=item.id, state="pending_input", input_gate="review") + db_session.expire_all() + assert get_subject_status(item.subject_type, str(item.id)).state == "pending_input" + assert str(item.id) in _board_ids(db_session) diff --git a/backend/tests/test_durable_schemas.py b/backend/tests/test_durable_schemas.py index 46f5079..2396db5 100644 --- a/backend/tests/test_durable_schemas.py +++ b/backend/tests/test_durable_schemas.py @@ -23,13 +23,12 @@ def _run( ) -def _status_of(runs, active_calls=None): - active_run = next((run for run in runs if run.is_active), None) - return _status(runs, active_run, active_calls or []) +def _status_of(runs, calls=None): + # runs arrives newest-first, mirroring Run.list_for_subject. + return _status(runs[0], calls or []) -def test_subject_state_prefers_the_newer_active_run_over_a_stale_parked_one(): - # runs arrives newest-first, mirroring Run.list_for_subject. +def test_subject_state_takes_the_newest_run(): runs = [ _run("new", "build.build_workflow", RunState.RUNNING), _run("old", "build.build_workflow", RunState.PENDING_INPUT), @@ -37,7 +36,7 @@ def test_subject_state_prefers_the_newer_active_run_over_a_stale_parked_one(): assert _status_of(runs).state == RunState.RUNNING -def test_subject_state_prefers_a_newer_parked_run_over_an_older_running_one(): +def test_subject_state_takes_a_newer_parked_run_over_an_older_running_one(): # Recency decides, not a hardcoded state preference. runs = [ _run("new", "build.build_workflow", RunState.PENDING_INPUT), diff --git a/backend/tests/test_generic_subjects.py b/backend/tests/test_generic_subjects.py index 9aa75b3..3af3741 100644 --- a/backend/tests/test_generic_subjects.py +++ b/backend/tests/test_generic_subjects.py @@ -99,8 +99,8 @@ def client(tmp_path: Path, db_session, monkeypatch): def test_status_aggregates_across_runs_and_timeline_spans_them(client: TestClient, db_session): # Subject "1" lived across two runs: an earlier finished one and a current - # running one. Status is the active run's (running wins), and the timeline is - # every run, oldest first, each carrying its own agent calls. + # running one. Status is the newest run's, and the timeline is every run, + # oldest first, each carrying its own agent calls. done = _seed_run(db_session, subject_id="1", kind="faketest.prepare", state="finished") _seed_call(db_session, done, agent="prepare") live = _seed_run(db_session, subject_id="1", state="running") diff --git a/frontend/src/extensions/build/WorkItemsPage.tsx b/frontend/src/extensions/build/WorkItemsPage.tsx index b5d6ef8..a6dff10 100644 --- a/frontend/src/extensions/build/WorkItemsPage.tsx +++ b/frontend/src/extensions/build/WorkItemsPage.tsx @@ -140,12 +140,12 @@ export function WorkItemsPage() { (row.status.state === 'running' || row.status.state === 'scheduled') && matchesQuery(row, query), ) - const shown = needsYou.length + inFlight.length + const total = needsYou.length + inFlight.length const head = ( {needsYou.length} needs you @@ -169,10 +169,10 @@ export function WorkItemsPage() { return ( - {shown === 0 ? ( + {total === 0 ? ( ) : (