diff --git a/backend/druks/build/extension.py b/backend/druks/build/extension.py index 8149b5ab..407344d1 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 230198ea..81350332 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 12bc9c2d..17038e96 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 9b048dc0..f5f6f5a5 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 5a1679cd..ffae9f8c 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 5704207c..4793c35d 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 00000000..49c424c4 --- /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 46f50793..2396db52 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 9aa75b34..3af37411 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 b5d6ef83..a6dff10b 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 ? ( ) : (