From 3dee77e0a44d44e8cbd67a126f3504d58b37d28d Mon Sep 17 00:00:00 2001 From: "exe.dev user" Date: Mon, 27 Jul 2026 14:52:35 +0000 Subject: [PATCH 1/3] Store pull request resolution for board history --- AGENTS.md | 4 +- backend/druks/contrib/ship/enums.py | 5 - backend/druks/contrib/ship/extension.py | 11 +- backend/druks/contrib/ship/models.py | 29 +---- backend/druks/contrib/ship/routes.py | 3 +- backend/druks/contrib/ship/schemas.py | 23 +++- backend/druks/contrib/ship/subscribers.py | 30 ++--- backend/druks/contrib/ship/workflows.py | 4 +- backend/druks/durable/datastructures.py | 10 +- backend/druks/durable/models.py | 21 ++- backend/druks/models.py | 14 +- .../a4c9e2f7b1d6_store_pr_resolution.py | 60 +++++++++ backend/tests/ship/test_api_work_items.py | 26 ++-- .../tests/ship/test_build_board_membership.py | 80 +++++------- backend/tests/ship/test_build_dispatch.py | 13 +- backend/tests/ship/test_build_durable.py | 5 +- backend/tests/ship/test_lane_reactions.py | 10 -- .../tests/ship/test_webhooks_pull_request.py | 122 ++++++++---------- .../src/extensions/ship/HistoryPage.test.tsx | 74 +++++++++++ frontend/src/extensions/ship/HistoryPage.tsx | 48 +++---- frontend/src/extensions/ship/StatusTag.tsx | 16 +-- .../extensions/ship/WorkItemsPage.test.tsx | 87 +++++++++++++ .../src/extensions/ship/WorkItemsPage.tsx | 16 ++- frontend/src/extensions/ship/api.ts | 6 +- frontend/src/styles.css | 3 +- 25 files changed, 444 insertions(+), 276 deletions(-) create mode 100644 backend/migrations/versions/a4c9e2f7b1d6_store_pr_resolution.py create mode 100644 frontend/src/extensions/ship/HistoryPage.test.tsx create mode 100644 frontend/src/extensions/ship/WorkItemsPage.test.tsx diff --git a/AGENTS.md b/AGENTS.md index 7d32ec48..0bc828a0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -49,8 +49,8 @@ For extension-surface changes, inspect the proof extension at `druks.durable`, `druks.extensions`, and `druks.webhooks` are what an author imports; a reference to `druks.build` or any other extension inside them inverts the platform. - Liveness — is this subject still being worked — derives from run state; never mirror - it in a column. An extension's own outcome, such as shipped or cancelled, is a domain - fact it does own, and it records that by reacting to run lifecycle. + it in a column. An externally owned PR outcome is stored from the provider event, + never inferred from run lifecycle. - `@step` marks a replay checkpoint, not an expensive call. On replay the body re-executes from the top and completed steps return cached results, so code outside a step runs again. Moving the boundary changes correctness, not performance. diff --git a/backend/druks/contrib/ship/enums.py b/backend/druks/contrib/ship/enums.py index ad0c890c..1893265e 100644 --- a/backend/druks/contrib/ship/enums.py +++ b/backend/druks/contrib/ship/enums.py @@ -19,8 +19,3 @@ class HumanFeedbackAction(StrEnum): CONTRACT_CHANGE_REQUIRED = "contract_change_required" QUESTION = "question" CLOSE = "close" - - -class HandoffStatus(StrEnum): - SHIPPED = "shipped" - CANCELLED = "cancelled" diff --git a/backend/druks/contrib/ship/extension.py b/backend/druks/contrib/ship/extension.py index f0ed57e4..9754cb1f 100644 --- a/backend/druks/contrib/ship/extension.py +++ b/backend/druks/contrib/ship/extension.py @@ -13,6 +13,7 @@ ) from druks.contrib.ship.models import WorkItem from druks.contrib.ship.schemas import WorkItemSummary +from druks.durable.models import Run from druks.extensions import Extension from druks.workflows import Subject, SubjectActivity @@ -134,9 +135,13 @@ def get_subject_summary(cls, subject: Subject) -> WorkItemSummary | None: @classmethod def list_subjects(cls) -> list[WorkItemSummary]: - # 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)] + states = Run.subject_states(cls.subject_type) + items = WorkItem.list_recent(limit=500) + return [ + WorkItemSummary.from_work_item(item) + for item in items + if str(item.id) in states and item.pr_resolved_at is None + ] @classmethod async def get_subject_activity(cls, subject: Subject) -> SubjectActivity | None: diff --git a/backend/druks/contrib/ship/models.py b/backend/druks/contrib/ship/models.py index 1c3aa4e7..9d2f5629 100644 --- a/backend/druks/contrib/ship/models.py +++ b/backend/druks/contrib/ship/models.py @@ -6,7 +6,6 @@ from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import Mapped, mapped_column, relationship -from druks.contrib.ship.enums import HandoffStatus from druks.contrib.ship.policy import RepoPolicy from druks.core.apis.github import get_github_client from druks.db import Base, StoredSubject, db_session @@ -181,7 +180,6 @@ class WorkItem(StoredSubject): # don't collide once we support multiple providers. Index("work_items_ticket_unique", "source", "ticket_key", unique=True), Index("work_items_project_idx", "project_id"), - Index("work_items_status_idx", "status"), ) project_id: Mapped[int] = mapped_column( @@ -209,7 +207,8 @@ class WorkItem(StoredSubject): build_run_id: Mapped[str | None] = mapped_column( ForeignKey("durable_runs.id", ondelete="SET NULL"), default=None ) - status: Mapped[str | None] = mapped_column(default=None) + pr_merged: Mapped[bool | None] = mapped_column(default=None) + pr_resolved_at: Mapped[datetime | None] = mapped_column(default=None) created_at: Mapped[datetime] = mapped_column(default=Base.utc_now) updated_at: Mapped[datetime] = mapped_column(default=Base.utc_now) @@ -277,7 +276,6 @@ async def ship(self) -> None: # one converges on its own, its merge step finding the PR already closed. from druks.contrib.ship.workflows import Build - self.set_status(HandoffStatus.SHIPPED) build = self.get_status(workflow=Build) if build.is_parked: await Build.cancel(self, failure="pr merged while parked") @@ -289,7 +287,6 @@ async def close_external(self) -> None: # must not strand it there. from druks.contrib.ship.workflows import Build - self.set_status(HandoffStatus.CANCELLED, event_payload={"external": True}) await Build.cancel(self, failure="pr closed without merge") db_session().flush() try: @@ -320,27 +317,15 @@ def list_recent(cls, *, limit: int = 50, offset: int = 0) -> list["WorkItem"]: @classmethod def list_handoff(cls, *, limit: int = 10) -> list["WorkItem"]: - # The history list: items at rest in a handoff lane, newest first. + # The history list: resolved pull requests, newest first. stmt = ( - select(cls).where(cls.status.is_not(None)).order_by(cls.updated_at.desc()).limit(limit) + select(cls) + .where(cls.pr_resolved_at.is_not(None)) + .order_by(cls.pr_resolved_at.desc()) + .limit(limit) ) return list(db_session().scalars(stmt)) - def set_status( - self, status: HandoffStatus | None, *, event_payload: dict[str, Any] | None = None - ) -> None: - """The handoff-lane write. A non-None status is a milestone, so the - matching build event records first — the pairing is structural, not a - call-site convention. None clears the lane on (re)dispatch: no event.""" - if status: - # cycle: the extension imports this module at file scope. - import druks.contrib.ship.extension as ship_extension - - ship_extension.Ship.record_event(type=status, subject=self, payload=event_payload) - self.status = status - self.updated_at = Base.utc_now() - db_session().flush() - async def set_ticket_status(self, status: TicketStatus) -> None: # No-op for sources without a configured tracker (github, absent creds). if not is_tracker_source(self.source): diff --git a/backend/druks/contrib/ship/routes.py b/backend/druks/contrib/ship/routes.py index a4e70cbd..878516fb 100644 --- a/backend/druks/contrib/ship/routes.py +++ b/backend/druks/contrib/ship/routes.py @@ -250,7 +250,6 @@ async def list_work_items_history( ) -> WorkItemsHistoryResponse: response.headers["Cache-Control"] = "no-store" clamped = max(1, min(limit, _HISTORY_MAX_LIMIT)) - # Recent-history aggregation. History is "handoff" — druks finished a - # unit and handed it off as shipped or cancelled. + # Recent PR outcomes, ordered by the provider's resolution timestamp. items = [DashboardItem.from_work_item(wi) for wi in WorkItem.list_handoff(limit=clamped)] return WorkItemsHistoryResponse(items=items) diff --git a/backend/druks/contrib/ship/schemas.py b/backend/druks/contrib/ship/schemas.py index de57890c..1aea2a17 100644 --- a/backend/druks/contrib/ship/schemas.py +++ b/backend/druks/contrib/ship/schemas.py @@ -7,9 +7,14 @@ from druks.schemas import BaseResponse from druks.workflows import SubjectSummary -from .enums import HandoffStatus - ProfileState = Literal["unprofiled", "running", "ready", "failed"] +PRResolution = Literal["merged", "closed"] + + +def _pr_resolution(merged: bool | None) -> PRResolution | None: + if merged is None: + return + return "merged" if merged else "closed" class ProjectRepoSummary(BaseResponse): @@ -112,6 +117,7 @@ class WorkItemSummary(SubjectSummary): ticket_url: str | None = None pr_number: int | None = None branch: str | None = None + resolution: PRResolution | None = None created_at: datetime updated_at: datetime links: Links @@ -128,6 +134,7 @@ def from_work_item(cls, item: WorkItem) -> "WorkItemSummary": ticket_url=item.ticket_url, pr_number=item.pr_number, branch=item.branch, + resolution=_pr_resolution(item.pr_merged), created_at=item.created_at, updated_at=item.updated_at, links=Links.from_work_item(item), @@ -142,15 +149,17 @@ class DashboardItem(BaseResponse): repo: str | None = None pr_number: int | None = None project_name: str | None = None - # The stored handoff lane, verbatim — the FE words and colors it. - status: HandoffStatus + resolution: PRResolution created_at: datetime updated_at: datetime links: Links @classmethod def from_work_item(cls, item: WorkItem) -> "DashboardItem": - # History is terminal-only, so the stored lane is always set. + assert item.pr_merged is not None + assert item.pr_resolved_at is not None + resolution = _pr_resolution(item.pr_merged) + assert resolution is not None return cls( key=f"code:{item.id}", source_id=item.id, @@ -161,9 +170,9 @@ def from_work_item(cls, item: WorkItem) -> "DashboardItem": # Druks Project is required on WorkItem, so the dashboard # always has a curated project name to render. project_name=item.project.name, - status=HandoffStatus(item.status), + resolution=resolution, created_at=item.created_at, - updated_at=item.updated_at, + updated_at=item.pr_resolved_at, links=Links.from_work_item(item), ) diff --git a/backend/druks/contrib/ship/subscribers.py b/backend/druks/contrib/ship/subscribers.py index 528eadf8..c13e563a 100644 --- a/backend/druks/contrib/ship/subscribers.py +++ b/backend/druks/contrib/ship/subscribers.py @@ -1,20 +1,14 @@ from druks.contrib.ship.contracts import ReviewWork -from druks.contrib.ship.enums import HandoffStatus from druks.contrib.ship.extension import Ship from druks.contrib.ship.models import ProjectRepo, WorkItem from druks.contrib.ship.workflows import Build, Profile +from druks.database import db_session +from druks.models import Base from druks.signals import subscribe from druks.ticketing.enums import TicketStatus from druks.workflows import WorkflowEvent -@subscribe(WorkflowEvent.RUNNING, subject=WorkItem) -async def any_workflow_start_returns_item_to_board(*, subject: WorkItem, **_: object) -> None: - # Any workflow starting for a work item puts it back on the active board — - # a new build or resume means druks has it in court again. - subject.set_status(None) - - @subscribe("pr.opened", workflow=Build, subject=WorkItem) async def pr_open_mirrors_onto_item( *, subject: WorkItem, pr_number: int, branch: str, **_: object @@ -36,15 +30,6 @@ async def review_park_marks_ticket_in_review(*, subject: WorkItem, **_: object) await subject.set_ticket_status(TicketStatus.IN_REVIEW) -@subscribe(WorkflowEvent.FAILED, workflow=Build, subject=WorkItem) -@subscribe(WorkflowEvent.CANCELLED, workflow=Build, 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 @@ -73,15 +58,16 @@ async def pr_review_answers_the_gate(*, repo: str, pr_number: int, payload: dict @subscribe("pr.closed") async def pr_close_settles_the_item(*, repo: str, pr_number: int, payload: dict) -> None: - """A PR druks owns closed on GitHub — the owner announcing the outcome. - One path for every merge, druks's own included: GitHub says merged, druks - ships the item. The status guards are redelivery idempotency.""" + """Store GitHub's outcome once, then apply its operational side effects.""" item = WorkItem.get_for_pr(repo=repo, pr_number=pr_number, branch=payload["branch"]) - if not item or item.status == HandoffStatus.SHIPPED: + if not item or item.pr_merged is not None: return + item.pr_merged = payload["merged"] + item.pr_resolved_at = Base.utc_now() + db_session().flush() if payload["merged"]: await item.ship() - elif item.status != HandoffStatus.CANCELLED: + else: await item.close_external() diff --git a/backend/druks/contrib/ship/workflows.py b/backend/druks/contrib/ship/workflows.py index 9b563ea3..6223bb9f 100644 --- a/backend/druks/contrib/ship/workflows.py +++ b/backend/druks/contrib/ship/workflows.py @@ -129,8 +129,6 @@ async def dispatch(cls, *, ticket: dict) -> str | None: # can't resolve this item onto the new run and cancel it. if item.build_run_id != run_id: item.update(build_run_id=run_id, branch=None, pr_number=None) - # (Re)dispatched → back in flight; clear the handoff lane. - item.set_status(None) return run_id async def run_multistep( @@ -313,7 +311,7 @@ async def _work_gate(self) -> bool: return False async def _approved_work(self) -> bool: - # GitHub announces the merge; the pr.closed reaction settles shipped. + # GitHub announces the merge; the pr.closed reaction stores its outcome. if self._policy.on_approval == "merge": if await self.declare_merge_intent(): return True diff --git a/backend/druks/durable/datastructures.py b/backend/druks/durable/datastructures.py index 62134c33..db243d85 100644 --- a/backend/druks/durable/datastructures.py +++ b/backend/druks/durable/datastructures.py @@ -53,8 +53,12 @@ def list_open(cls, subject_type: str, *, limit: int = 50) -> list["Subject"]: the subject is identity alone; a subject with rows of its own lists them with ``StoredSubject.list_open``.""" # Cycle: the durable read side is built on this package's models. - from druks.database import db_session + from druks.durable.enums import OPEN_STATES from druks.durable.models import Run - open_ids = db_session().scalars(Run.open_subject_ids(subject_type).limit(limit)) - return [cls(id=subject_id, subject_type=subject_type) for subject_id in open_ids] + states = Run.subject_states(subject_type) + return [ + cls(id=subject_id, subject_type=subject_type) + for subject_id, state in states.items() + if state in OPEN_STATES + ][:limit] diff --git a/backend/druks/durable/models.py b/backend/druks/durable/models.py index ea6cf1de..3c61f8d8 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, Select, String, func, select, update +from sqlalchemy import ForeignKey, Index, String, 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 @@ -22,7 +22,6 @@ ) from druks.durable.enums import ( ACTIVE_STATES, - OPEN_STATES, AgentCallStatus, RunState, WorkflowEvent, @@ -150,9 +149,8 @@ def get_latest_for_subject( 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, that - run's most recent first — a subquery their own read composes.""" + def subject_states(cls, subject_type: str) -> dict[str, RunState]: + """Every subject's newest driving run state, newest run first.""" 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 = ( @@ -171,14 +169,15 @@ def open_subject_ids(cls, subject_type: str) -> Select: .where(workflow_status.c.attributes["subject_type"].as_string() == subject_type) .subquery() ) - return ( - select(driving.c.subject_id) - .where( - driving.c.rank == 1, - driving.c.state.in_([run_state.value for run_state in OPEN_STATES]), - ) + stmt = ( + select(driving.c.subject_id, driving.c.state) + .where(driving.c.rank == 1) .order_by(driving.c.created_at.desc()) ) + return { + subject_id: RunState(state) + for subject_id, state in db_session().execute(stmt) + } def get_ask(self) -> dict[str, Any]: # The parked ask, ready to serve. An in-app review's ask names neither diff --git a/backend/druks/models.py b/backend/druks/models.py index d4c7812a..6964435d 100644 --- a/backend/druks/models.py +++ b/backend/druks/models.py @@ -2,7 +2,7 @@ from datetime import UTC, datetime from typing import TYPE_CHECKING, Any, ClassVar, Self -from sqlalchemy import DateTime, Integer, cast, select +from sqlalchemy import DateTime, select from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy.types import TypeDecorator @@ -108,14 +108,18 @@ def list_open(cls, *, limit: int = 50) -> list[Self]: 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.enums import OPEN_STATES from druks.durable.models import Run - # The durable layer keys subjects by string, so the open ids come back as - # text and cast to this table's integer key. - open_ids = Run.open_subject_ids(cls.subject_type).subquery() + states = Run.subject_states(cls.subject_type) + open_ids = [ + int(subject_id) + for subject_id, state in states.items() + if state in OPEN_STATES + ] stmt = ( select(cls) - .where(cls.id.in_(select(cast(open_ids.c.subject_id, Integer)))) + .where(cls.id.in_(open_ids)) .order_by(cls.id.desc()) .limit(limit) ) diff --git a/backend/migrations/versions/a4c9e2f7b1d6_store_pr_resolution.py b/backend/migrations/versions/a4c9e2f7b1d6_store_pr_resolution.py new file mode 100644 index 00000000..228c1b33 --- /dev/null +++ b/backend/migrations/versions/a4c9e2f7b1d6_store_pr_resolution.py @@ -0,0 +1,60 @@ +"""store pull request resolution + +Revision ID: a4c9e2f7b1d6 +Revises: f2c8b81a9d4e +Create Date: 2026-07-27 16:00:00.000000 + +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "a4c9e2f7b1d6" +down_revision: str | Sequence[str] | None = "f2c8b81a9d4e" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column("work_items", sa.Column("pr_merged", sa.Boolean(), nullable=True)) + op.add_column( + "work_items", + sa.Column("pr_resolved_at", sa.DateTime(timezone=True), nullable=True), + ) + op.execute( + sa.text( + """ + UPDATE work_items + SET pr_merged = CASE status + WHEN 'shipped' THEN true + WHEN 'cancelled' THEN false + END, + pr_resolved_at = updated_at + WHERE status IN ('shipped', 'cancelled') + """ + ) + ) + op.drop_index("work_items_status_idx", table_name="work_items") + op.drop_column("work_items", "status") + + +def downgrade() -> None: + op.add_column("work_items", sa.Column("status", sa.String(), nullable=True)) + op.execute( + sa.text( + """ + UPDATE work_items + SET status = CASE + WHEN pr_merged = true THEN 'shipped' + WHEN pr_merged = false THEN 'cancelled' + END + WHERE pr_merged IS NOT NULL + """ + ) + ) + op.create_index("work_items_status_idx", "work_items", ["status"], unique=False) + op.drop_column("work_items", "pr_resolved_at") + op.drop_column("work_items", "pr_merged") diff --git a/backend/tests/ship/test_api_work_items.py b/backend/tests/ship/test_api_work_items.py index f84cea30..108ce94c 100644 --- a/backend/tests/ship/test_api_work_items.py +++ b/backend/tests/ship/test_api_work_items.py @@ -2,6 +2,8 @@ import pytest from druks.contrib.ship.models import WorkItem +from druks.database import db_session +from druks.models import Base from druks.testing import seed_call from druks.workflows import Subject from fastapi.testclient import TestClient @@ -54,12 +56,12 @@ def _seed_op(druks_db, work_item_id, *, kind="implement", state, input_gate=None seed_call(druks_db, run, kind) -def _ship(repo, pr_number): - """Merge a work item's PR — the 'shipped' log event that lands it in - History, mirroring the merge handler.""" +def _resolve_pr(repo, pr_number, *, merged=True): item = WorkItem.get_for_pr(repo=repo, pr_number=pr_number) if item: - item.set_status("shipped") + item.pr_merged = merged + item.pr_resolved_at = Base.utc_now() + db_session().flush() # The generic subject read-side — Ship declares subject=WorkItem, so the @@ -76,12 +78,13 @@ def test_subject_list_shows_active_and_excludes_handed_off(client: TestClient, d done = make_test_work_item(title="shipped one", repo=repo).id WorkItem.get(done).update(pr_number=1) _seed_op(druks_db, done, state="finished") - _ship(repo, 1) + _resolve_pr(repo, 1) rows = {r["summary"]["title"]: r for r in client.get("/api/ship/work_item").json()["rows"]} assert "building" in rows assert "shipped one" not in rows assert rows["building"]["status"]["state"] == "running" + assert rows["building"]["summary"]["resolution"] is None def test_subject_detail_composes_summary_status_and_timeline(client: TestClient, druks_db): @@ -162,7 +165,7 @@ def test_history_returns_only_done_work_items(client: TestClient, druks_db): done_id = make_test_work_item(title="shipped one", repo=repo).id WorkItem.get(done_id).update(pr_number=1) _seed_op(druks_db, done_id, state="finished") - _ship(repo, 1) + _resolve_pr(repo, 1) # Running → active. running_id = make_test_work_item(title="still running", repo=repo).id _seed_op(druks_db, running_id, state="running") @@ -176,19 +179,22 @@ def test_history_returns_only_done_work_items(client: TestClient, druks_db): assert "shipped one" in titles assert "still running" not in titles assert "broke" not in titles # failed items stay active, not history + resolved = next(item for item in items if item["title"] == "shipped one") + assert resolved["resolution"] == "merged" + assert "status" not in resolved -def test_pr_closed_without_merge_is_cancelled_in_history(client: TestClient, druks_db): +def test_pr_closed_without_merge_is_closed_in_history(client: TestClient, druks_db): repo = "ClawHaven/acme-app" # A build parked on the operator, whose PR was then closed without merging. wid = make_test_work_item(title="abandoned", repo=repo).id WorkItem.get(wid).update(pr_number=7) _seed_op(druks_db, wid, state="finished") - WorkItem.get(wid).set_status("cancelled") + _resolve_pr(repo, 7, merged=False) items = client.get("/api/ship/work-items/history").json()["items"] row = next(it for it in items if it["title"] == "abandoned") - assert row["status"] == "cancelled" + assert row["resolution"] == "closed" def test_history_clamps_limit(client: TestClient, druks_db): @@ -196,7 +202,7 @@ def test_history_clamps_limit(client: TestClient, druks_db): wid = make_test_work_item(title=f"shipped {i}", repo="ClawHaven/acme-app").id WorkItem.get(wid).update(pr_number=i + 1) _seed_op(druks_db, wid, state="finished") - _ship("ClawHaven/acme-app", i + 1) + _resolve_pr("ClawHaven/acme-app", i + 1) # limit > cap → clamps down, doesn't 400. response = client.get("/api/ship/work-items/history?limit=10000") diff --git a/backend/tests/ship/test_build_board_membership.py b/backend/tests/ship/test_build_board_membership.py index c999a5a1..7cd4ee37 100644 --- a/backend/tests/ship/test_build_board_membership.py +++ b/backend/tests/ship/test_build_board_membership.py @@ -1,8 +1,11 @@ +from datetime import UTC, datetime, timedelta + import druks.contrib.ship.workflows # noqa: F401 # registers ship.build, the seeded kind import pytest -from druks.contrib.ship.enums import HandoffStatus from druks.contrib.ship.extension import Ship from druks.contrib.ship.models import WorkItem +from druks.durable.enums import RunState +from druks.durable.models import Run from ship.factories import make_test_work_item, seed_build_run @@ -12,71 +15,52 @@ def _board_ids(druks_db): return {row.id for row in Ship.list_subjects()} -@pytest.mark.parametrize("state", ["scheduled", "running", "parked", "failed"]) -def test_a_live_or_failed_build_holds_the_board(druks_db, state): - # A failed run still wants the operator, so it stays alongside the live ones. +@pytest.mark.parametrize("state", ["scheduled", "running", "parked", "failed", "finished"]) +def test_an_unresolved_build_holds_the_board(druks_db, state): item = make_test_work_item(repo="ClawHaven/acme-app", title=f"build {state}") seed_build_run(druks_db, work_item_id=item.id, state=state) + assert str(item.id) in _board_ids(druks_db) -@pytest.mark.parametrize("state", ["finished", "cancelled"]) -def test_an_ended_build_leaves_the_board(druks_db, 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}") +@pytest.mark.parametrize("state", ["running", "failed", "finished"]) +def test_a_resolved_pr_leaves_the_board_before_run_state_is_considered(druks_db, state): + item = make_test_work_item(repo="ClawHaven/acme-app", title=f"resolved {state}") seed_build_run(druks_db, work_item_id=item.id, state=state) - druks_db.expire_all() - assert item.status is None - assert str(item.id) not in _board_ids(druks_db) + item.pr_merged = True + item.pr_resolved_at = datetime.now(UTC) + druks_db.flush() - -def test_a_redispatched_item_returns_to_the_board(druks_db): - # 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(druks_db, work_item_id=item.id, state="finished") - seed_build_run(druks_db, work_item_id=item.id, state="running") - assert str(item.id) in _board_ids(druks_db) + assert str(item.id) not in _board_ids(druks_db) def test_an_item_without_runs_stays_off_the_board(druks_db): item = make_test_work_item(repo="ClawHaven/acme-app", title="never dispatched") - assert str(item.id) not in _board_ids(druks_db) - -async def test_a_cancelled_build_settles_as_cancelled(druks_db): - # Nothing merged, so History records the abandonment. - from druks.contrib.ship.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(druks_db): - # ship() lands first and owns the verdict; the run's own cancel must not - # overwrite it on the way out. - from druks.contrib.ship.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 + assert str(item.id) not in _board_ids(druks_db) -def test_history_holds_the_handed_off(druks_db): - shipped = make_test_work_item(repo="ClawHaven/acme-app", title="shipped") - shipped.set_status(HandoffStatus.SHIPPED) +def test_history_holds_resolved_items_in_resolution_order(druks_db): + older = make_test_work_item(repo="ClawHaven/acme-app", title="older") + older.pr_merged = False + older.pr_resolved_at = datetime.now(UTC) - timedelta(minutes=1) + newer = make_test_work_item(repo="ClawHaven/acme-app", title="newer") + newer.pr_merged = True + newer.pr_resolved_at = datetime.now(UTC) make_test_work_item(repo="ClawHaven/acme-app", title="still open") - assert [item.id for item in WorkItem.list_handoff()] == [shipped.id] + druks_db.flush() + + assert [item.id for item in WorkItem.list_handoff()] == [newer.id, older.id] -def test_the_newest_run_speaks_for_the_item(druks_db): - # 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. +def test_the_newest_run_state_speaks_for_the_item(druks_db): item = make_test_work_item(repo="ClawHaven/acme-app", title="two runs") - seed_build_run(druks_db, work_item_id=item.id, state="cancelled") + seed_build_run(druks_db, work_item_id=item.id, state="finished") seed_build_run(druks_db, work_item_id=item.id, state="parked", input_gate="review") druks_db.expire_all() - assert item.get_status().state == "parked" + + states = Run.subject_states(WorkItem.subject_type) + + assert states[str(item.id)] == RunState.PARKED + assert item.get_status().state == RunState.PARKED assert str(item.id) in _board_ids(druks_db) diff --git a/backend/tests/ship/test_build_dispatch.py b/backend/tests/ship/test_build_dispatch.py index 396f6627..9569fa88 100644 --- a/backend/tests/ship/test_build_dispatch.py +++ b/backend/tests/ship/test_build_dispatch.py @@ -1,16 +1,14 @@ -from druks.contrib.ship.enums import HandoffStatus from druks.contrib.ship.workflows import Build +from druks.models import Base from druks.testing import seed_run from ship.factories import make_test_work_item -async def test_dispatch_pulls_cancelled_item_back_onto_the_board(druks_db, monkeypatch) -> None: - """A cancelled item rests in History; dispatching its build must clear the - handoff status so the active board shows the run (and its gates) instead - of a stale "Cancelled" row.""" +async def test_dispatch_does_not_clear_a_stored_pr_resolution(druks_db, monkeypatch) -> None: item = make_test_work_item(repo="o/r", title="t", ticket_key="ACME-1") - item.set_status(HandoffStatus.CANCELLED) + item.pr_merged = False + item.pr_resolved_at = Base.utc_now() seed_run(druks_db, kind=Build.kind, run_id="run-1") started = {} @@ -35,7 +33,8 @@ async def fake_start(cls, **kwargs): assert run_id == "run-1" assert item.build_run_id == "run-1" - assert item.status is None + assert item.pr_merged is False + assert item.pr_resolved_at is not None assert started["ticket_ref"] == "ACME-1" assert started["ticket_title"] == "t" assert started["ticket_url"] == "https://tracker.test/ACME-1" diff --git a/backend/tests/ship/test_build_durable.py b/backend/tests/ship/test_build_durable.py index fb51edee..4875495e 100644 --- a/backend/tests/ship/test_build_durable.py +++ b/backend/tests/ship/test_build_durable.py @@ -250,9 +250,8 @@ async def test_happy_path_declares_merge_intent(rt, monkeypatch): refreshed = session.get(WorkItem, item.id) assert (refreshed.pr_number, refreshed.branch) == (42, "agent/acme-1") - # Shipped settles via GitHub's pr.closed webhook (test_webhooks_pull_request), - # not the run — the run's job ends when GitHub accepts the merge intent. The durable - # residue is the event log of its state transitions. + # GitHub's pr.closed webhook stores the PR outcome (test_webhooks_pull_request), + # not the run — the run's job ends when GitHub accepts the merge intent. from druks.events.models import Event session = get_session(rt.engine) diff --git a/backend/tests/ship/test_lane_reactions.py b/backend/tests/ship/test_lane_reactions.py index 65445d80..5222ced6 100644 --- a/backend/tests/ship/test_lane_reactions.py +++ b/backend/tests/ship/test_lane_reactions.py @@ -1,7 +1,6 @@ import druks.contrib.ship.subscribers # noqa: F401 — registers the lane reactions import pytest from druks.contrib.ship.contracts import ReviewWork -from druks.contrib.ship.enums import HandoffStatus from druks.contrib.ship.models import WorkItem from druks.contrib.ship.workflows import Build from druks.durable import Run @@ -16,15 +15,6 @@ pytestmark = pytest.mark.asyncio -async def test_run_running_puts_item_back_on_board(druks_db): - item = make_test_work_item(repo="acme/widget", title="t", source="linear", ticket_key="ACME-1") - item.set_status(HandoffStatus.CANCELLED) - - await publish(WorkflowEvent.RUNNING, subject=item.identity, kind=Build.kind) - - assert WorkItem.get(item.id).status is None - - async def test_build_lifecycle_reaches_the_tracker(druks_db, monkeypatch): pushed = [] diff --git a/backend/tests/ship/test_webhooks_pull_request.py b/backend/tests/ship/test_webhooks_pull_request.py index 015cca25..9408bf94 100644 --- a/backend/tests/ship/test_webhooks_pull_request.py +++ b/backend/tests/ship/test_webhooks_pull_request.py @@ -1,3 +1,4 @@ +from datetime import datetime from types import SimpleNamespace import druks.contrib.ship.subscribers # noqa: F401 — registers the pr.closed subscriber @@ -5,9 +6,8 @@ from druks.contrib.ship.models import WorkItem from druks.core.webhooks.github import GitHubEvents from druks.durable import Run -from druks.events.models import Event -from druks.testing import make_settings -from sqlalchemy import func, select +from druks.testing import configure_app_for_test, make_settings +from fastapi.testclient import TestClient from ship.factories import make_test_work_item, seed_build_run @@ -22,20 +22,6 @@ async def _fetch(*, repo, path): monkeypatch.setattr("druks.extensions.config.fetch_file", _fetch) -def _milestone_count(work_item_id, milestone): - from druks.database import db_session - - return db_session().scalar( - select(func.count()) - .select_from(Event) - .where( - Event.subject_type == "work_item", - Event.subject_id == str(work_item_id), - Event.type == milestone, - ) - ) - - async def _fire_closed(*, repo, pr_number, branch, tmp_path, merged=True): payload = { "repository": {"full_name": repo}, @@ -79,15 +65,15 @@ def _fresh_run(run_id): @pytest.mark.asyncio -async def test_external_merge_records_event_and_ends_involvement(druks_db, tmp_path): +async def test_external_merge_stores_resolution_and_ends_involvement(druks_db, tmp_path): repo, pr_number, branch = "ClawHaven/acme-app", 42, "agent/eng-1" work_item_id, run_id = _park_work_item(repo=repo, pr_number=pr_number, branch=branch) await _fire_closed(repo=repo, pr_number=pr_number, branch=branch, tmp_path=tmp_path) - # Ship-ness is recorded as the 'shipped' milestone... - assert _milestone_count(work_item_id, "shipped") == 1 - # ...and involvement ended: the parked build run is cancelled. + resolved = WorkItem.get(work_item_id) + assert resolved.pr_merged is True + assert resolved.pr_resolved_at is not None assert not _fresh_run(run_id).is_active @@ -107,26 +93,58 @@ async def test_merge_ships_but_leaves_a_running_build_to_converge(druks_db, tmp_ await _fire_closed(repo=repo, pr_number=pr_number, branch=branch, tmp_path=tmp_path) assert Run.get(run_id).state == "running" # not cancelled from under druks - assert _milestone_count(work_item_id, "shipped") == 1 + assert WorkItem.get(work_item_id).pr_merged is True @pytest.mark.asyncio -async def test_redelivered_merge_webhook_does_not_double_record(druks_db, tmp_path): - """GitHub redelivers webhooks; a shipped item stays shipped once.""" +async def test_redelivered_webhook_does_not_rewrite_resolution(druks_db, tmp_path): repo, pr_number, branch = "ClawHaven/acme-app", 44, "agent/eng-3" work_item_id, _ = _park_work_item(repo=repo, pr_number=pr_number, branch=branch) await _fire_closed(repo=repo, pr_number=pr_number, branch=branch, tmp_path=tmp_path) + resolved_at = WorkItem.get(work_item_id).pr_resolved_at + + await _fire_closed( + repo=repo, + pr_number=pr_number, + branch=branch, + tmp_path=tmp_path, + merged=False, + ) + + item = WorkItem.get(work_item_id) + assert item.pr_merged is True + assert item.pr_resolved_at == resolved_at + + +@pytest.mark.asyncio +async def test_failed_build_later_merged_moves_from_board_to_history(druks_db, tmp_path): + repo, pr_number, branch = "ClawHaven/acme-app", 126, "agent/eng-760" + work_item_id, _ = _park_work_item( + repo=repo, + pr_number=pr_number, + branch=branch, + state="failed", + ) await _fire_closed(repo=repo, pr_number=pr_number, branch=branch, tmp_path=tmp_path) - assert _milestone_count(work_item_id, "shipped") == 1 + item = WorkItem.get(work_item_id) + assert item.pr_merged is True + settings = make_settings(tmp_path) + with TestClient(configure_app_for_test(settings=settings)) as client: + active = client.get("/api/ship/work_item").json()["rows"] + history = client.get("/api/ship/work-items/history").json()["items"] + + assert str(work_item_id) not in {row["summary"]["id"] for row in active} + resolved = next(row for row in history if row["sourceId"] == work_item_id) + assert resolved["resolution"] == "merged" + resolved_at = datetime.fromisoformat(resolved["updatedAt"].replace("Z", "+00:00")) + assert resolved_at == item.pr_resolved_at @pytest.mark.asyncio async def test_closed_unmerged_records_close_and_ends_involvement(druks_db, tmp_path): - """A PR closed *without* merging — the operator abandoned it (e.g. deleted - the branch). Emit 'cancelled' and un-park so the item derives as cancelled - and leaves the active board, rather than being ignored.""" + """A PR closed without merging stores GitHub's outcome and cancels the run.""" repo, pr_number, branch = "ClawHaven/acme-app", 45, "agent/eng-4" work_item_id, run_id = _park_work_item(repo=repo, pr_number=pr_number, branch=branch) @@ -138,8 +156,9 @@ async def test_closed_unmerged_records_close_and_ends_involvement(druks_db, tmp_ merged=False, ) - assert _milestone_count(work_item_id, "cancelled") == 1 - assert _milestone_count(work_item_id, "shipped") == 0 + resolved = WorkItem.get(work_item_id) + assert resolved.pr_merged is False + assert resolved.pr_resolved_at is not None assert not _fresh_run(run_id).is_active @@ -158,43 +177,7 @@ async def test_closed_unmerged_cancels_in_flight_run(druks_db, tmp_path): ) assert _fresh_run(run_id).state == "cancelled" - assert _milestone_count(work_item_id, "cancelled") == 1 - - -@pytest.mark.asyncio -async def test_remerge_after_retrigger_records_a_fresh_shipped(druks_db, tmp_path): - """A prior round's 'shipped' must not swallow a second merge after the item - was re-triggered — the lane is active again, so the webhook ships it again.""" - from datetime import UTC, datetime, timedelta - - from druks.database import db_session as ds - - repo, pr_number, branch = "ClawHaven/acme-app", 77, "agent/eng-9" - work_item_id, _ = _park_work_item(repo=repo, pr_number=pr_number, branch=branch) - # Round 1: shipped. - WorkItem.get(work_item_id).set_status("shipped") - # Re-trigger: a newer RUNNING build run; dispatch cleared the handoff status. - new_run = seed_build_run(ds(), work_item_id=work_item_id, state="running") - new_run.created_at = datetime.now(UTC) + timedelta(seconds=5) - WorkItem.get(work_item_id).set_status(None) - ds().flush() - - await _fire_closed(repo=repo, pr_number=pr_number, branch=branch, tmp_path=tmp_path) - - assert _milestone_count(work_item_id, "shipped") == 2 - - -@pytest.mark.asyncio -async def test_merge_echo_with_no_newer_activity_still_dedups(druks_db, tmp_path): - """The echo case: druks's own merge emits 'shipped' and GitHub's closed - webhook arrives with nothing newer — still dropped.""" - repo, pr_number, branch = "ClawHaven/acme-app", 78, "agent/eng-10" - work_item_id, _ = _park_work_item(repo=repo, pr_number=pr_number, branch=branch) - WorkItem.get(work_item_id).set_status("shipped") - - await _fire_closed(repo=repo, pr_number=pr_number, branch=branch, tmp_path=tmp_path) - - assert _milestone_count(work_item_id, "shipped") == 1 # deduped + assert WorkItem.get(work_item_id).pr_merged is False @pytest.mark.asyncio @@ -336,7 +319,7 @@ async def _record(self, status): assert deleted == [] # cleanup skipped when policy can't be resolved assert pushed == [TicketStatus.READY_FOR_AGENT] # ticket still reset - assert _milestone_count(work_item_id, "cancelled") == 1 + assert WorkItem.get(work_item_id).pr_merged is False @pytest.mark.asyncio @@ -356,4 +339,5 @@ async def test_stale_close_after_redispatch_spares_the_new_run(druks_db, tmp_pat await _fire_closed(repo=repo, pr_number=pr_a, branch=branch_a, tmp_path=tmp_path, merged=False) assert _fresh_run(new_run.id).state == "running" # the live attempt is untouched - assert _milestone_count(item.id, "cancelled") == 0 + assert item.pr_merged is None + assert item.pr_resolved_at is None diff --git a/frontend/src/extensions/ship/HistoryPage.test.tsx b/frontend/src/extensions/ship/HistoryPage.test.tsx new file mode 100644 index 00000000..280807f3 --- /dev/null +++ b/frontend/src/extensions/ship/HistoryPage.test.tsx @@ -0,0 +1,74 @@ +import { QueryClient, QueryClientProvider } from '@tanstack/react-query' +import { cleanup, fireEvent, render, screen } from '@testing-library/react' +import { afterEach, describe, expect, it, vi } from 'vitest' + +import { HistoryPage } from './HistoryPage' + +afterEach(() => { + cleanup() + vi.unstubAllGlobals() +}) + +describe('HistoryPage', () => { + it('counts, labels, and filters canonical PR resolutions', async () => { + vi.stubGlobal( + 'fetch', + vi.fn( + async () => + new Response( + JSON.stringify({ + items: [ + { + key: 'code:1', + sourceId: 1, + ticketKey: 'ENG-1', + title: 'Merged title', + repo: 'czpython/druks', + prNumber: 1, + projectName: 'druks', + resolution: 'merged', + createdAt: '2026-07-27T12:00:00Z', + updatedAt: '2026-07-27T13:00:00Z', + links: { repo: 'https://github.com/czpython/druks' }, + }, + { + key: 'code:2', + sourceId: 2, + ticketKey: 'ENG-2', + title: 'Closed title', + repo: 'czpython/druks', + prNumber: 2, + projectName: 'druks', + resolution: 'closed', + createdAt: '2026-07-27T12:00:00Z', + updatedAt: '2026-07-27T12:30:00Z', + links: { repo: 'https://github.com/czpython/druks' }, + }, + ], + }), + { status: 200 }, + ), + ), + ) + const queryClient = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }) + render( + + + , + ) + + expect(await screen.findByText('Merged title')).toBeTruthy() + expect(screen.getByText('Closed title')).toBeTruthy() + expect(screen.getByText('merged (1)')).toBeTruthy() + expect(screen.getByText('closed (1)')).toBeTruthy() + expect(screen.getByTitle('merged').textContent).toBe('✓') + expect(screen.getByTitle('closed').textContent).toBe('◯') + + fireEvent.click(screen.getByText('closed (1)')) + + expect(screen.queryByText('Merged title')).toBeNull() + expect(screen.getByText('Closed title')).toBeTruthy() + }) +}) diff --git a/frontend/src/extensions/ship/HistoryPage.tsx b/frontend/src/extensions/ship/HistoryPage.tsx index ee693630..a205e33b 100644 --- a/frontend/src/extensions/ship/HistoryPage.tsx +++ b/frontend/src/extensions/ship/HistoryPage.tsx @@ -3,7 +3,7 @@ import { useState } from 'react' import { useLocation } from 'wouter' import { buildApi } from './api' -import type { HandoffStatus } from './api' +import type { PRResolution } from './api' import { EmptyState } from '../../components/EmptyState' import { FilterChip } from '../../components/FilterChip' import { StatusTag } from './StatusTag' @@ -15,7 +15,7 @@ import { RepoCell } from '../../components/RepoCell' import { relTime, secondsSince, updatedAtSortKey } from '../../lib/format' import { dashboardItemPath } from './slug' -type StatusFilter = 'all' | HandoffStatus +type ResolutionFilter = 'all' | PRResolution function repoBareName(repo: string | null | undefined): string | null { if (!repo) return null @@ -27,7 +27,7 @@ export function HistoryPage() { queryKey: ['work-items-history'], queryFn: () => buildApi.history(), }) - const [statusFilter, setStatusFilter] = useState('all') + const [resolutionFilter, setResolutionFilter] = useState('all') const [query, setQuery] = useState('') const [, navigate] = useLocation() @@ -35,14 +35,14 @@ export function HistoryPage() { if (gate) return {gate} const items = historyQuery.data!.items - const counts: Record = { - shipped: items.filter((i) => i.status === 'shipped').length, - cancelled: items.filter((i) => i.status === 'cancelled').length, + const counts: Record = { + merged: items.filter((item) => item.resolution === 'merged').length, + closed: items.filter((item) => item.resolution === 'closed').length, } const filtered = items .filter((item) => { - if (statusFilter !== 'all' && item.status !== statusFilter) return false + if (resolutionFilter !== 'all' && item.resolution !== resolutionFilter) return false if (query.trim()) { const q = query.toLowerCase() if (!`${item.title} ${item.ticketKey}`.toLowerCase().includes(q)) return false @@ -58,11 +58,11 @@ export function HistoryPage() { meta={ <> - ✓ {counts.shipped} shipped + ✓ {counts.merged} merged · - ◯ {counts.cancelled} cancelled + ◯ {counts.closed} closed } @@ -76,23 +76,23 @@ export function HistoryPage() { onChange={(e) => setQuery(e.target.value)} /> · - + value="all" - current={statusFilter} - onSelect={setStatusFilter} + current={resolutionFilter} + onSelect={setResolutionFilter} label="all" /> - - value="shipped" - current={statusFilter} - onSelect={setStatusFilter} - label={`shipped (${counts.shipped})`} + + value="merged" + current={resolutionFilter} + onSelect={setResolutionFilter} + label={`merged (${counts.merged})`} /> - - value="cancelled" - current={statusFilter} - onSelect={setStatusFilter} - label={`cancelled (${counts.cancelled})`} + + value="closed" + current={resolutionFilter} + onSelect={setResolutionFilter} + label={`closed (${counts.closed})`} /> } @@ -124,14 +124,14 @@ export function HistoryPage() { className="row row-history" onClick={() => navigate(dashboardItemPath(item))} > - + {item.ticketKey} {item.title} - {item.status} + {item.resolution} {relTime(secondsSince(item.updatedAt))} diff --git a/frontend/src/extensions/ship/StatusTag.tsx b/frontend/src/extensions/ship/StatusTag.tsx index 8d73169b..895e2782 100644 --- a/frontend/src/extensions/ship/StatusTag.tsx +++ b/frontend/src/extensions/ship/StatusTag.tsx @@ -1,18 +1,18 @@ -import type { HandoffStatus } from './api' +import type { PRResolution } from './api' -const GLYPH: Record = { - shipped: '✓', - cancelled: '◯', +const GLYPH: Record = { + merged: '✓', + closed: '◯', } interface Props { - status: HandoffStatus + resolution: PRResolution } -export function StatusTag({ status }: Props) { +export function StatusTag({ resolution }: Props) { return ( - - {GLYPH[status]} + + {GLYPH[resolution]} ) } diff --git a/frontend/src/extensions/ship/WorkItemsPage.test.tsx b/frontend/src/extensions/ship/WorkItemsPage.test.tsx new file mode 100644 index 00000000..cc224f77 --- /dev/null +++ b/frontend/src/extensions/ship/WorkItemsPage.test.tsx @@ -0,0 +1,87 @@ +import { act, cleanup, render, screen } from '@testing-library/react' +import { afterEach, describe, expect, it, vi } from 'vitest' + +import type { RunState, SubjectStatus } from '../../api/types' +import type { PRResolution, WorkItemRow } from './api' +import { WorkItemsPage } from './WorkItemsPage' + +const sse = vi.hoisted(() => ({ + handlers: null as null | Record void>, +})) + +vi.mock('../../api/sse', () => ({ + useSSE: ( + _url: string, + options: { handlers: Record void> }, + ) => { + sse.handlers = options.handlers + }, +})) + +function row(id: string, state: RunState, resolution: PRResolution | null): WorkItemRow { + const status: SubjectStatus = { + state, + kind: 'ship.build', + agent: null, + gate: state === 'parked' ? 'review_work' : null, + failure: state === 'failed' ? 'tests failed' : null, + reason: null, + triggeredAt: '2026-07-27T12:01:00Z', + accountUsername: 'system', + } + return { + summary: { + id, + source: 'linear', + repo: 'czpython/druks', + projectName: 'druks', + title: `${state}-${id}`, + ticketKey: `ENG-${id}`, + ticketUrl: `https://linear.app/fellaworks/issue/ENG-${id}`, + prNumber: Number(id), + branch: `agent/eng-${id}`, + resolution, + createdAt: '2026-07-27T12:00:00Z', + updatedAt: `2026-07-27T12:0${id}:00Z`, + links: { + repo: 'https://github.com/czpython/druks', + pr: `https://github.com/czpython/druks/pull/${id}`, + ticket: `https://linear.app/fellaworks/issue/ENG-${id}`, + }, + }, + status, + } +} + +afterEach(() => { + cleanup() + sse.handlers = null +}) + +describe('WorkItemsPage', () => { + it('filters resolution first and renders every active bucket in its count', () => { + const { container } = render() + + act(() => { + sse.handlers?.snapshot({ + rows: [ + row('1', 'parked', null), + row('2', 'failed', null), + row('3', 'finished', null), + row('4', 'scheduled', null), + row('5', 'running', null), + row('6', 'running', 'merged'), + ], + }) + }) + + expect(screen.getByText('finished-3')).toBeTruthy() + expect(screen.queryByText('running-6')).toBeNull() + expect(container.querySelector('.dash-h1-count')?.textContent).toBe('(5)') + expect( + [...container.querySelectorAll('.wi-group-count')].map((node) => node.textContent), + ).toEqual(['(3)', '(2)']) + expect(screen.getByText('3 needs you')).toBeTruthy() + expect(screen.getByText('2 in flight')).toBeTruthy() + }) +}) diff --git a/frontend/src/extensions/ship/WorkItemsPage.tsx b/frontend/src/extensions/ship/WorkItemsPage.tsx index 6f02902a..8dddb67e 100644 --- a/frontend/src/extensions/ship/WorkItemsPage.tsx +++ b/frontend/src/extensions/ship/WorkItemsPage.tsx @@ -38,8 +38,8 @@ function WorkItemRowView({ row: WorkItemRow onOpen: (row: WorkItemRow) => void }) { - // Active lanes: parked → parked on you, failed → needs you to retry or - // cancel, running/scheduled → a step is live. + // Active lanes: parked/failed/finished need the operator; running/scheduled + // mean a step is live. const { summary: wi, status } = row const failed = status.state === 'failed' const parked = status.state === 'parked' @@ -124,15 +124,17 @@ export function WorkItemsPage() { ) } - // Two lanes: needs-you — parked on the operator OR failed (a failure still - // needs you) — and in-flight (a step running or about to). Newest - // movement first. - const active = [...rows].sort( + // Resolution wins before liveness: a provider-resolved PR is History even if + // its driving run still looks active or failed. + const unresolved = rows.filter((row) => row.summary.resolution === null) + const active = [...unresolved].sort( (a, b) => updatedAtSortKey(b.summary) - updatedAtSortKey(a.summary), ) const needsYou = active.filter( (row) => - (row.status.state === 'parked' || row.status.state === 'failed') && + (row.status.state === 'parked' || + row.status.state === 'failed' || + row.status.state === 'finished') && matchesQuery(row, query), ) const inFlight = active.filter( diff --git a/frontend/src/extensions/ship/api.ts b/frontend/src/extensions/ship/api.ts index dd2a6705..2538dd10 100644 --- a/frontend/src/extensions/ship/api.ts +++ b/frontend/src/extensions/ship/api.ts @@ -28,8 +28,7 @@ export const buildApi = { }, } -// The stored handoff lane, verbatim from the backend's HandoffStatus. -export type HandoffStatus = 'shipped' | 'cancelled' +export type PRResolution = 'merged' | 'closed' export interface Links { repo: string @@ -46,6 +45,7 @@ export interface WorkItemSummary extends SubjectSummary { ticketUrl?: string | null prNumber?: number | null branch?: string | null + resolution: PRResolution | null createdAt: string updatedAt: string links: Links @@ -60,7 +60,7 @@ export interface DashboardItem { repo?: string | null prNumber?: number | null projectName?: string | null - status: HandoffStatus + resolution: PRResolution createdAt: string updatedAt: string links: Links diff --git a/frontend/src/styles.css b/frontend/src/styles.css index efd31ac7..e0e2da4f 100644 --- a/frontend/src/styles.css +++ b/frontend/src/styles.css @@ -396,10 +396,9 @@ button { background: none; border: none; color: inherit; cursor: pointer; font: font-size: 11px; } .outcome-merged { color: var(--outcome-merged); } -.outcome-shipped { color: var(--outcome-merged); } .outcome-failed { color: var(--outcome-failed); } .outcome-abandoned { color: var(--outcome-abandoned); } -.outcome-cancelled { color: var(--outcome-abandoned); } +.outcome-closed { color: var(--outcome-abandoned); } /* ============================================================ Empty states From d9bfb17f4ea503c035920c2241dff2872afdce1c Mon Sep 17 00:00:00 2001 From: "exe.dev user" Date: Mon, 27 Jul 2026 15:11:09 +0000 Subject: [PATCH 2/3] Fix work items component test type safety --- frontend/src/extensions/ship/WorkItemsPage.test.tsx | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/frontend/src/extensions/ship/WorkItemsPage.test.tsx b/frontend/src/extensions/ship/WorkItemsPage.test.tsx index cc224f77..144f99d4 100644 --- a/frontend/src/extensions/ship/WorkItemsPage.test.tsx +++ b/frontend/src/extensions/ship/WorkItemsPage.test.tsx @@ -61,9 +61,13 @@ afterEach(() => { describe('WorkItemsPage', () => { it('filters resolution first and renders every active bucket in its count', () => { const { container } = render() + const snapshot = sse.handlers?.snapshot + if (!snapshot) { + throw new Error('snapshot handler was not registered') + } act(() => { - sse.handlers?.snapshot({ + snapshot({ rows: [ row('1', 'parked', null), row('2', 'failed', null), @@ -79,7 +83,9 @@ describe('WorkItemsPage', () => { expect(screen.queryByText('running-6')).toBeNull() expect(container.querySelector('.dash-h1-count')?.textContent).toBe('(5)') expect( - [...container.querySelectorAll('.wi-group-count')].map((node) => node.textContent), + Array.from(container.querySelectorAll('.wi-group-count')).map( + (node) => node.textContent, + ), ).toEqual(['(3)', '(2)']) expect(screen.getByText('3 needs you')).toBeTruthy() expect(screen.getByText('2 in flight')).toBeTruthy() From 49b39ed108ba85c341ed14343e152230cd84a291 Mon Sep 17 00:00:00 2001 From: "exe.dev user" Date: Mon, 27 Jul 2026 15:43:34 +0000 Subject: [PATCH 3/3] Correct PR resolution lifecycle and board filtering --- backend/druks/contrib/ship/extension.py | 8 +++-- backend/druks/contrib/ship/models.py | 6 ++++ backend/druks/contrib/ship/schemas.py | 7 +++-- backend/druks/contrib/ship/subscribers.py | 9 ++++-- backend/druks/contrib/ship/workflows.py | 12 ++++++-- backend/druks/core/webhooks/github.py | 5 ++++ backend/druks/durable/datastructures.py | 2 +- backend/druks/durable/models.py | 3 +- backend/druks/models.py | 25 +--------------- .../tests/ship/test_build_board_membership.py | 21 ++++++++++++++ backend/tests/ship/test_build_dispatch.py | 17 +++++++---- .../tests/ship/test_webhooks_pull_request.py | 29 +++++++++++++++++-- 12 files changed, 98 insertions(+), 46 deletions(-) diff --git a/backend/druks/contrib/ship/extension.py b/backend/druks/contrib/ship/extension.py index 9754cb1f..d5041401 100644 --- a/backend/druks/contrib/ship/extension.py +++ b/backend/druks/contrib/ship/extension.py @@ -13,7 +13,7 @@ ) from druks.contrib.ship.models import WorkItem from druks.contrib.ship.schemas import WorkItemSummary -from druks.durable.models import Run +from druks.durable import Run, RunState from druks.extensions import Extension from druks.workflows import Subject, SubjectActivity @@ -135,12 +135,14 @@ def get_subject_summary(cls, subject: Subject) -> WorkItemSummary | None: @classmethod def list_subjects(cls) -> list[WorkItemSummary]: - states = Run.subject_states(cls.subject_type) + states = Run.subject_states(cls.subject_type, limit=500) items = WorkItem.list_recent(limit=500) return [ WorkItemSummary.from_work_item(item) for item in items - if str(item.id) in states and item.pr_resolved_at is None + if str(item.id) in states + and item.pr_resolved_at is None + and states[str(item.id)] not in (RunState.CANCELLED, RunState.ORPHANED) ] @classmethod diff --git a/backend/druks/contrib/ship/models.py b/backend/druks/contrib/ship/models.py index 9d2f5629..a766cdd2 100644 --- a/backend/druks/contrib/ship/models.py +++ b/backend/druks/contrib/ship/models.py @@ -362,6 +362,8 @@ def update( pr_number: int | None = _KEEP, branch: str | None = _KEEP, build_run_id: str | None = _KEEP, + pr_merged: bool | None = _KEEP, + pr_resolved_at: datetime | None = _KEEP, project_id: int = _KEEP, ) -> None: if title is not _KEEP: @@ -374,6 +376,10 @@ def update( self.branch = branch if build_run_id is not _KEEP: self.build_run_id = build_run_id + if pr_merged is not _KEEP: + self.pr_merged = pr_merged + if pr_resolved_at is not _KEEP: + self.pr_resolved_at = pr_resolved_at if project_id is not _KEEP: self.project_id = project_id self.updated_at = Base.utc_now() diff --git a/backend/druks/contrib/ship/schemas.py b/backend/druks/contrib/ship/schemas.py index 1aea2a17..ffc39e76 100644 --- a/backend/druks/contrib/ship/schemas.py +++ b/backend/druks/contrib/ship/schemas.py @@ -156,10 +156,11 @@ class DashboardItem(BaseResponse): @classmethod def from_work_item(cls, item: WorkItem) -> "DashboardItem": - assert item.pr_merged is not None - assert item.pr_resolved_at is not None + if item.pr_merged is None or item.pr_resolved_at is None: + raise ValueError(f"work item {item.id} has no stored PR resolution") resolution = _pr_resolution(item.pr_merged) - assert resolution is not None + if resolution is None: + raise ValueError(f"work item {item.id} resolution could not be derived") return cls( key=f"code:{item.id}", source_id=item.id, diff --git a/backend/druks/contrib/ship/subscribers.py b/backend/druks/contrib/ship/subscribers.py index c13e563a..b59058af 100644 --- a/backend/druks/contrib/ship/subscribers.py +++ b/backend/druks/contrib/ship/subscribers.py @@ -1,9 +1,10 @@ +from datetime import datetime + from druks.contrib.ship.contracts import ReviewWork from druks.contrib.ship.extension import Ship from druks.contrib.ship.models import ProjectRepo, WorkItem from druks.contrib.ship.workflows import Build, Profile -from druks.database import db_session -from druks.models import Base +from druks.db import db_session from druks.signals import subscribe from druks.ticketing.enums import TicketStatus from druks.workflows import WorkflowEvent @@ -63,11 +64,13 @@ async def pr_close_settles_the_item(*, repo: str, pr_number: int, payload: dict) if not item or item.pr_merged is not None: return item.pr_merged = payload["merged"] - item.pr_resolved_at = Base.utc_now() + item.pr_resolved_at = datetime.fromisoformat(payload["resolved_at"].replace("Z", "+00:00")) db_session().flush() if payload["merged"]: + Ship.record_event(type="shipped", subject=item) await item.ship() else: + Ship.record_event(type="cancelled", subject=item) await item.close_external() diff --git a/backend/druks/contrib/ship/workflows.py b/backend/druks/contrib/ship/workflows.py index 6223bb9f..59c780c8 100644 --- a/backend/druks/contrib/ship/workflows.py +++ b/backend/druks/contrib/ship/workflows.py @@ -125,10 +125,16 @@ async def dispatch(cls, *, ticket: dict) -> str | None: task_owner_name=ticket["assignee_name"], ) # A duplicate dispatch gets the live run back; only a genuinely new run - # resets routing — drop the old branch/PR so a late close for the prior PR - # can't resolve this item onto the new run and cancel it. + # resets the attempt — drop the old branch/PR so a late close for the prior PR + # can't resolve this item onto the new run, and clear that prior PR's outcome. if item.build_run_id != run_id: - item.update(build_run_id=run_id, branch=None, pr_number=None) + item.update( + build_run_id=run_id, + branch=None, + pr_number=None, + pr_merged=None, + pr_resolved_at=None, + ) return run_id async def run_multistep( diff --git a/backend/druks/core/webhooks/github.py b/backend/druks/core/webhooks/github.py index 8886bea6..700f680f 100644 --- a/backend/druks/core/webhooks/github.py +++ b/backend/druks/core/webhooks/github.py @@ -79,6 +79,11 @@ async def on_pull_request_closed(self) -> Response: payload={ "branch": pull_request["head"]["ref"], "merged": pull_request["merged"], + "resolved_at": ( + pull_request["merged_at"] + if pull_request["merged"] + else pull_request["closed_at"] + ), }, ) return _accepted() diff --git a/backend/druks/durable/datastructures.py b/backend/druks/durable/datastructures.py index db243d85..cbdac479 100644 --- a/backend/druks/durable/datastructures.py +++ b/backend/druks/durable/datastructures.py @@ -56,7 +56,7 @@ def list_open(cls, subject_type: str, *, limit: int = 50) -> list["Subject"]: from druks.durable.enums import OPEN_STATES from druks.durable.models import Run - states = Run.subject_states(subject_type) + states = Run.subject_states(subject_type, limit=limit) return [ cls(id=subject_id, subject_type=subject_type) for subject_id, state in states.items() diff --git a/backend/druks/durable/models.py b/backend/druks/durable/models.py index 3c61f8d8..cae9a027 100644 --- a/backend/druks/durable/models.py +++ b/backend/druks/durable/models.py @@ -149,7 +149,7 @@ def get_latest_for_subject( return db_session().scalars(stmt).first() @classmethod - def subject_states(cls, subject_type: str) -> dict[str, RunState]: + def subject_states(cls, subject_type: str, *, limit: int = 500) -> dict[str, RunState]: """Every subject's newest driving run state, newest run first.""" 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") @@ -173,6 +173,7 @@ def subject_states(cls, subject_type: str) -> dict[str, RunState]: select(driving.c.subject_id, driving.c.state) .where(driving.c.rank == 1) .order_by(driving.c.created_at.desc()) + .limit(limit) ) return { subject_id: RunState(state) diff --git a/backend/druks/models.py b/backend/druks/models.py index 6964435d..8d779df4 100644 --- a/backend/druks/models.py +++ b/backend/druks/models.py @@ -2,7 +2,7 @@ from datetime import UTC, datetime from typing import TYPE_CHECKING, Any, ClassVar, Self -from sqlalchemy import DateTime, select +from sqlalchemy import DateTime from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy.types import TypeDecorator @@ -101,26 +101,3 @@ def get_timeline(self) -> "list[RunResponse]": async def get_phase(self) -> str | None: return await self.subject.get_phase() - - @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.enums import OPEN_STATES - from druks.durable.models import Run - - states = Run.subject_states(cls.subject_type) - open_ids = [ - int(subject_id) - for subject_id, state in states.items() - if state in OPEN_STATES - ] - stmt = ( - select(cls) - .where(cls.id.in_(open_ids)) - .order_by(cls.id.desc()) - .limit(limit) - ) - return list(db_session().scalars(stmt)) diff --git a/backend/tests/ship/test_build_board_membership.py b/backend/tests/ship/test_build_board_membership.py index 7cd4ee37..3873fa30 100644 --- a/backend/tests/ship/test_build_board_membership.py +++ b/backend/tests/ship/test_build_board_membership.py @@ -4,6 +4,7 @@ import pytest from druks.contrib.ship.extension import Ship from druks.contrib.ship.models import WorkItem +from druks.durable.dbos_state import workflow_status from druks.durable.enums import RunState from druks.durable.models import Run @@ -40,6 +41,26 @@ def test_an_item_without_runs_stays_off_the_board(druks_db): assert str(item.id) not in _board_ids(druks_db) +@pytest.mark.parametrize("state", ["cancelled", "orphaned"]) +def test_a_cancelled_or_orphaned_build_with_no_resolution_leaves_the_board( + druks_db, state +): + item = make_test_work_item(repo="ClawHaven/acme-app", title=f"{state} build") + if state == "cancelled": + seed_build_run(druks_db, work_item_id=item.id, state=state) + else: + run = seed_build_run(druks_db, work_item_id=item.id) + druks_db.execute( + workflow_status.delete().where(workflow_status.c.workflow_uuid == run.id) + ) + run.created_at = datetime.now(UTC) - timedelta(minutes=10) + druks_db.flush() + druks_db.expire(run, ["state"]) + assert Run.get(run.id).state == RunState.ORPHANED + + assert str(item.id) not in _board_ids(druks_db) + + def test_history_holds_resolved_items_in_resolution_order(druks_db): older = make_test_work_item(repo="ClawHaven/acme-app", title="older") older.pr_merged = False diff --git a/backend/tests/ship/test_build_dispatch.py b/backend/tests/ship/test_build_dispatch.py index 9569fa88..8b671ed7 100644 --- a/backend/tests/ship/test_build_dispatch.py +++ b/backend/tests/ship/test_build_dispatch.py @@ -1,3 +1,4 @@ +from druks.contrib.ship.extension import Ship from druks.contrib.ship.workflows import Build from druks.models import Base from druks.testing import seed_run @@ -5,11 +6,11 @@ from ship.factories import make_test_work_item -async def test_dispatch_does_not_clear_a_stored_pr_resolution(druks_db, monkeypatch) -> None: +async def test_dispatch_pulls_cancelled_item_back_onto_the_board(druks_db, monkeypatch) -> None: item = make_test_work_item(repo="o/r", title="t", ticket_key="ACME-1") - item.pr_merged = False + item.pr_merged = True item.pr_resolved_at = Base.utc_now() - seed_run(druks_db, kind=Build.kind, run_id="run-1") + seed_run(druks_db, kind=Build.kind, subject=item, run_id="run-1") started = {} async def fake_start(cls, **kwargs): @@ -33,8 +34,9 @@ async def fake_start(cls, **kwargs): assert run_id == "run-1" assert item.build_run_id == "run-1" - assert item.pr_merged is False - assert item.pr_resolved_at is not None + assert item.pr_merged is None + assert item.pr_resolved_at is None + assert str(item.id) in {summary.id for summary in Ship.list_subjects()} assert started["ticket_ref"] == "ACME-1" assert started["ticket_title"] == "t" assert started["ticket_url"] == "https://tracker.test/ACME-1" @@ -80,6 +82,9 @@ async def test_duplicate_dispatch_keeps_the_live_attempt_routing(druks_db, monke seed_run(druks_db, kind=Build.kind, run_id="run-live") item = make_test_work_item(repo="o/r", title="t", ticket_key="ACME-3") item.update(build_run_id="run-live", pr_number=7, branch="agent/live") + resolved_at = Base.utc_now() + item.pr_merged = False + item.pr_resolved_at = resolved_at async def dedup_start(cls, **kwargs): return "run-live" @@ -102,6 +107,8 @@ async def dedup_start(cls, **kwargs): assert item.build_run_id == "run-live" assert item.pr_number == 7 assert item.branch == "agent/live" + assert item.pr_merged is False + assert item.pr_resolved_at == resolved_at def test_update_clears_nullable_with_none_and_skips_omitted(druks_db) -> None: diff --git a/backend/tests/ship/test_webhooks_pull_request.py b/backend/tests/ship/test_webhooks_pull_request.py index 9408bf94..3974e13a 100644 --- a/backend/tests/ship/test_webhooks_pull_request.py +++ b/backend/tests/ship/test_webhooks_pull_request.py @@ -6,11 +6,15 @@ from druks.contrib.ship.models import WorkItem from druks.core.webhooks.github import GitHubEvents from druks.durable import Run +from druks.events.models import Event from druks.testing import configure_app_for_test, make_settings from fastapi.testclient import TestClient from ship.factories import make_test_work_item, seed_build_run +_MERGED_AT = "2026-07-25T21:59:09Z" +_CLOSED_AT = "2026-07-26T13:30:30Z" + @pytest.fixture(autouse=True) def _stub_config_fetch(monkeypatch): @@ -22,12 +26,23 @@ async def _fetch(*, repo, path): monkeypatch.setattr("druks.extensions.config.fetch_file", _fetch) -async def _fire_closed(*, repo, pr_number, branch, tmp_path, merged=True): +async def _fire_closed( + *, + repo, + pr_number, + branch, + tmp_path, + merged=True, + merged_at=_MERGED_AT, + closed_at=_CLOSED_AT, +): payload = { "repository": {"full_name": repo}, "pull_request": { "number": pr_number, "merged": merged, + "merged_at": merged_at, + "closed_at": closed_at, "head": {"ref": branch}, }, } @@ -73,7 +88,11 @@ async def test_external_merge_stores_resolution_and_ends_involvement(druks_db, t resolved = WorkItem.get(work_item_id) assert resolved.pr_merged is True - assert resolved.pr_resolved_at is not None + assert resolved.pr_resolved_at == datetime.fromisoformat( + _MERGED_AT.replace("Z", "+00:00") + ) + events = druks_db.query(Event).filter_by(subject_id=str(work_item_id)).all() + assert "shipped" in {event.type for event in events} assert not _fresh_run(run_id).is_active @@ -158,7 +177,11 @@ async def test_closed_unmerged_records_close_and_ends_involvement(druks_db, tmp_ resolved = WorkItem.get(work_item_id) assert resolved.pr_merged is False - assert resolved.pr_resolved_at is not None + assert resolved.pr_resolved_at == datetime.fromisoformat( + _CLOSED_AT.replace("Z", "+00:00") + ) + events = druks_db.query(Event).filter_by(subject_id=str(work_item_id)).all() + assert "cancelled" in {event.type for event in events} assert not _fresh_run(run_id).is_active