The active board follows each item's driving run

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:

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

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

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)

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,

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

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)

backend/tests/test_durable_schemas.py

@@ -23,21 +23,20 @@ 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),
     ]
     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),

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