ENG-760 - [build] Board and History read one stored fact — the PR's resolution, not druks's opinion of it

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.

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"

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 import Run, RunState
 from druks.extensions import Extension
 from druks.workflows import Subject, SubjectActivity
 
@@ -134,9 +135,15 @@ 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, 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
+            and states[str(item.id)] not in (RunState.CANCELLED, RunState.ORPHANED)
+        ]
 
     @classmethod
     async def get_subject_activity(cls, subject: Subject) -> SubjectActivity | None:

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):
@@ -377,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:
@@ -389,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()

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)

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,18 @@ 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.
+        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)
+        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,
@@ -161,9 +171,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),
         )
 

backend/druks/contrib/ship/subscribers.py

@@ -1,20 +1,15 @@
+from datetime import datetime
+
 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.db import db_session
 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 +31,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 +59,18 @@ 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 = 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()
-    elif item.status != HandoffStatus.CANCELLED:
+    else:
+        Ship.record_event(type="cancelled", subject=item)
         await item.close_external()
 
 

backend/druks/contrib/ship/workflows.py

@@ -125,12 +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)
-        # (Re)dispatched → back in flight; clear the handoff lane.
-        item.set_status(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(
@@ -313,7 +317,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

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

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, limit=limit)
+        return [
+            cls(id=subject_id, subject_type=subject_type)
+            for subject_id, state in states.items()
+            if state in OPEN_STATES
+        ][:limit]

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, *, 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")
         driving = (
@@ -171,14 +169,16 @@ 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())
+            .limit(limit)
         )
+        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