diff --git a/backend/druks/contrib/review/datastructures.py b/backend/druks/contrib/review/datastructures.py index 9ea0d7c2..81fae076 100644 --- a/backend/druks/contrib/review/datastructures.py +++ b/backend/druks/contrib/review/datastructures.py @@ -45,7 +45,7 @@ def get_summary(self) -> ReviewSummary: ) @classmethod - def list_summaries(cls) -> list[ReviewSummary]: + def list_summaries(cls, account_id: str | None) -> list[ReviewSummary]: """The reviews still going or stopped on a failure. A finished one lives on its pull request, so it leaves the board as soon as it has something to show there.""" return [pull_request.get_summary() for pull_request in cls.list_open()] diff --git a/backend/druks/contrib/ship/models.py b/backend/druks/contrib/ship/models.py index 20e1727f..9be838e0 100644 --- a/backend/druks/contrib/ship/models.py +++ b/backend/druks/contrib/ship/models.py @@ -110,7 +110,7 @@ def get_summary(self) -> "ProjectRepoSummary": return ProjectRepoSummary.model_validate(self) @classmethod - def list_summaries(cls) -> list["ProjectRepoSummary"]: + def list_summaries(cls, account_id: str | None) -> list["ProjectRepoSummary"]: # A repo is registered, not transient, so the board is all of them by name. stmt = select(cls).order_by(cls.full_name) return [repo.get_summary() for repo in db_session().scalars(stmt)] @@ -228,7 +228,7 @@ def get_summary(self) -> WorkItemSummary: return WorkItemSummary.model_validate(self) @classmethod - def list_summaries(cls) -> list[WorkItemSummary]: + def list_summaries(cls, account_id: str | None) -> list[WorkItemSummary]: # Where a run stands colours the row; it never decides whether the row is # here. The 500 most-recent cover it; paginate if a board outgrows it. stmt = ( diff --git a/backend/druks/durable/datastructures.py b/backend/druks/durable/datastructures.py index 1a8c8970..e58ca941 100644 --- a/backend/druks/durable/datastructures.py +++ b/backend/druks/durable/datastructures.py @@ -52,10 +52,12 @@ def get_summary(self) -> "SubjectSummary": ) @classmethod - def list_summaries(cls) -> "Sequence[SubjectSummary]": + def list_summaries(cls, account_id: str | None) -> "Sequence[SubjectSummary]": """The subjects on this class's board, newest-movement first, each as its domain - summary. Returns a covariant ``Sequence`` so an extension can return a ``list`` - of its own ``SubjectSummary`` subclass. Required once a workflow declares it.""" + summary. ``account_id`` is the caller, or None outside a request. A shared + board ignores it. Returns a covariant ``Sequence`` so an extension can return + a ``list`` of its own ``SubjectSummary`` subclass. Required once a workflow + declares it.""" raise NotImplementedError( f"a workflow declares {cls.__name__}, so it needs a list_summaries()" ) diff --git a/backend/druks/extensions/base.py b/backend/druks/extensions/base.py index a45b85bf..24b6f1b3 100644 --- a/backend/druks/extensions/base.py +++ b/backend/druks/extensions/base.py @@ -398,10 +398,12 @@ def _get_subject_routes( point-in-time read and a ``/stream`` that pushes the whole snapshot on change. Mounted at ``/api//`` for every subject the extension's workflows declare. Every read here is keyed by identity, so an extension that - keeps no row for its subject gets the same surface as one that does.""" + keeps no row for its subject gets the same surface as one that does. The board + reads pass the caller to ``list_summaries``.""" from fastapi import APIRouter, HTTPException, status from fastapi.responses import StreamingResponse + from druks.accounts.context import current_account_id from druks.api.dependencies import EngineDep from druks.database import session_scope from druks.durable import reads @@ -411,14 +413,14 @@ def _get_subject_routes( subject_type = subject_class.subject_type router = APIRouter(prefix=f"/{subject_type}", tags=[f"{cls.name}:{subject_type}"]) - def board() -> SubjectList: + def board(account_id: str | None) -> SubjectList: return SubjectList( rows=[ SubjectRow( summary=summary, status=reads.get_subject_status(subject_type, summary.id), ) - for summary in subject_class.list_summaries() + for summary in subject_class.list_summaries(account_id) ] ) @@ -435,14 +437,17 @@ async def subject_response(subject_id: str) -> SubjectResponse | None: @router.get("", response_model=SubjectList, response_model_by_alias=True) async def list_subjects() -> SubjectList: - return board() + return board(current_account_id.get()) # ``/stream`` before ``/{subject_id}`` so the literal path wins over the id matcher. @router.get("/stream", response_class=StreamingResponse) async def stream_board(engine: EngineDep) -> StreamingResponse: + # The generator polls after the request context ends. Capture the caller here. + account_id = current_account_id.get() + async def snapshot() -> SubjectList: with session_scope(engine): - return board() + return board(account_id) return StreamingResponse( stream(snapshot), media_type="text/event-stream", headers=SSE_HEADERS diff --git a/backend/druks/models.py b/backend/druks/models.py index a345e3b2..8b098a86 100644 --- a/backend/druks/models.py +++ b/backend/druks/models.py @@ -94,10 +94,12 @@ def get_summary(self) -> "SubjectSummary": ) @classmethod - def list_summaries(cls) -> "Sequence[SubjectSummary]": + def list_summaries(cls, account_id: str | None) -> "Sequence[SubjectSummary]": """The rows on this class's board, newest-movement first, each as its domain - summary. Returns a covariant ``Sequence`` so an extension can return a ``list`` - of its own ``SubjectSummary`` subclass. Required once a workflow declares it.""" + summary. ``account_id`` is the caller, or None outside a request. A shared + board ignores it. Returns a covariant ``Sequence`` so an extension can return + a ``list`` of its own ``SubjectSummary`` subclass. Required once a workflow + declares it.""" raise NotImplementedError( f"a workflow declares {cls.__name__}, so it needs a list_summaries()" ) diff --git a/backend/tests/druks-field_notes/druks_field_notes/models.py b/backend/tests/druks-field_notes/druks_field_notes/models.py index 4c2fadea..2e4cdd90 100644 --- a/backend/tests/druks-field_notes/druks_field_notes/models.py +++ b/backend/tests/druks-field_notes/druks_field_notes/models.py @@ -42,7 +42,7 @@ def get_summary(self) -> NoteSummary: return NoteSummary.model_validate(self) @classmethod - def list_summaries(cls) -> list[NoteSummary]: + def list_summaries(cls, account_id: str | None) -> list[NoteSummary]: # How many the board shows is an operator knob, so it lives on the extension. from druks_field_notes.extension import FieldNotes diff --git a/backend/tests/druks-field_notes/tests/test_extension.py b/backend/tests/druks-field_notes/tests/test_extension.py index a6721ae2..708c8a14 100644 --- a/backend/tests/druks-field_notes/tests/test_extension.py +++ b/backend/tests/druks-field_notes/tests/test_extension.py @@ -9,7 +9,7 @@ def test_the_board_honors_the_board_size(druks_db): newest = Note.create(body="second") FieldNotes.override_setting("board_size", 1) - summaries = Note.list_summaries() + summaries = Note.list_summaries(None) assert [summary.id for summary in summaries] == [str(newest.id)] assert summaries[0].body == "second" diff --git a/backend/tests/ship/test_build_board_membership.py b/backend/tests/ship/test_build_board_membership.py index 3e3a2749..f5736458 100644 --- a/backend/tests/ship/test_build_board_membership.py +++ b/backend/tests/ship/test_build_board_membership.py @@ -9,7 +9,7 @@ def _board_ids(druks_db): druks_db.expire_all() - return {row.id for row in WorkItem.list_summaries()} + return {row.id for row in WorkItem.list_summaries(None)} def _resolve(item, *, merged=True, at=None): diff --git a/backend/tests/ship/test_lane_reactions.py b/backend/tests/ship/test_lane_reactions.py index 7cf75936..9124f192 100644 --- a/backend/tests/ship/test_lane_reactions.py +++ b/backend/tests/ship/test_lane_reactions.py @@ -45,7 +45,7 @@ async def test_cancelled_build_settles_the_item(druks_db): assert refreshed.resolution == "closed" assert refreshed.resolved_at druks_db.expire_all() - assert str(item.id) not in {summary.id for summary in WorkItem.list_summaries()} + assert str(item.id) not in {summary.id for summary in WorkItem.list_summaries(None)} @pytest.mark.parametrize(("merged", "resolution"), [(True, "merged"), (False, "closed")]) @@ -79,7 +79,7 @@ async def test_failed_build_remains_unresolved_on_the_board(druks_db): refreshed = WorkItem.get(item.id) assert (refreshed.resolution, refreshed.resolved_at) == (None, None) druks_db.expire_all() - assert str(item.id) in {summary.id for summary in WorkItem.list_summaries()} + assert str(item.id) in {summary.id for summary in WorkItem.list_summaries(None)} async def test_build_lifecycle_reaches_the_tracker(druks_db, monkeypatch): diff --git a/backend/tests/ship/test_webhooks_pull_request.py b/backend/tests/ship/test_webhooks_pull_request.py index d6a8604c..ff506022 100644 --- a/backend/tests/ship/test_webhooks_pull_request.py +++ b/backend/tests/ship/test_webhooks_pull_request.py @@ -202,7 +202,7 @@ async def test_a_merge_after_a_failed_build_still_settles_the_item(druks_db, tmp item = WorkItem.get(work_item_id) assert item.resolution == "merged" - assert item.id not in {summary.id for summary in WorkItem.list_summaries()} + assert item.id not in {summary.id for summary in WorkItem.list_summaries(None)} assert [row.id for row in WorkItem.list_handoff()] == [item.id] diff --git a/backend/tests/test_extension_appless_load.py b/backend/tests/test_extension_appless_load.py index 93396215..e9e9d090 100644 --- a/backend/tests/test_extension_appless_load.py +++ b/backend/tests/test_extension_appless_load.py @@ -43,7 +43,7 @@ class ProbeItem(StoredSubject): __tablename__ = "probe_items" @classmethod - def list_summaries(cls) -> list[SubjectSummary]: + def list_summaries(cls, account_id: str | None) -> list[SubjectSummary]: return [] """, "routes.py": """ @@ -215,11 +215,11 @@ def test_surfaces_are_enumerable_from_the_loaded_extension(installed): assert f"{_PACKAGE}.subscribers" in capability_modules settings_model = extension.settings_model - assert settings_model is not None + assert settings_model assert list(settings_model.model_fields) == ["budget"] package_dir = extension.package_dir() - assert package_dir is not None + assert package_dir assert extension.migrations_dir() == package_dir / "migrations" diff --git a/backend/tests/test_extensions.py b/backend/tests/test_extensions.py index 63f67fae..8e0875c6 100644 --- a/backend/tests/test_extensions.py +++ b/backend/tests/test_extensions.py @@ -14,7 +14,7 @@ class Widget(Subject): in a real extension; these stand in for that.""" @classmethod - def list_summaries(cls) -> list: + def list_summaries(cls, account_id: str | None) -> list: return [] @@ -246,7 +246,7 @@ def workflows(cls): def test_a_concrete_inherited_list_summaries_satisfies_the_contract_without_calling_it(): class Concrete(Subject): @classmethod - def list_summaries(cls) -> list: + def list_summaries(cls, account_id: str | None) -> list: raise AssertionError("validation must not call list_summaries()") class Inheritor(Concrete): diff --git a/backend/tests/test_generic_subjects.py b/backend/tests/test_generic_subjects.py index 7757223b..3b1005fd 100644 --- a/backend/tests/test_generic_subjects.py +++ b/backend/tests/test_generic_subjects.py @@ -1,6 +1,7 @@ from pathlib import Path import pytest +from druks.accounts.context import current_account_id from druks.database import db_session from druks.durable import AgentCall, Run from druks.durable.datastructures import Subject @@ -29,7 +30,7 @@ def get_summary(self) -> _ThingSummary: return _ThingSummary(id=self.id, label=self.label, title=TITLES[self.id]) @classmethod - def list_summaries(cls) -> list[_ThingSummary]: + def list_summaries(cls, account_id: str | None) -> list[_ThingSummary]: return [thing.get_summary() for thing in db_session().scalars(select(cls))] @@ -47,10 +48,22 @@ def get_summary(self) -> _ThingSummary: return _ThingSummary(id=self.id, label=self.label, title=self.id.rpartition("#")[2]) @classmethod - def list_summaries(cls) -> list[_ThingSummary]: + def list_summaries(cls, account_id: str | None) -> list[_ThingSummary]: return [ticket.get_summary() for ticket in cls.list_open()] +CALLERS: list[str | None] = [] + + +class Inbox(Subject): + """A board scoped by who is asking.""" + + @classmethod + def list_summaries(cls, account_id: str | None) -> list[_ThingSummary]: + CALLERS.append(account_id) + return [] + + class _ThingExtension(Extension): name = "faketest" @@ -66,7 +79,7 @@ def _seed_run( input_gate=None, failure=None, ): - if state == "parked" and input_gate is None: + if state == "parked" and not input_gate: input_gate = "review" run = Run( id=str(uuid7()), @@ -229,6 +242,27 @@ def test_list_returns_every_subject_with_status(client: TestClient, druks_db): assert rows["2"]["status"]["state"] == "scheduled" +async def test_the_board_and_its_stream_hand_the_caller_to_list_summaries(druks_db): + # The route reads the caller at handler entry and passes it down as data. + # The stream keeps that caller after the request context ends. + endpoints = { + route.path: route.endpoint for route in _ThingExtension._get_subject_routes(Inbox).routes + } + + CALLERS.clear() + token = current_account_id.set("acct-7") + try: + await endpoints["/inbox"]() + response = await endpoints["/inbox/stream"](engine=druks_db.get_bind()) + finally: + current_account_id.reset(token) + assert CALLERS == ["acct-7"] + + await anext(response.body_iterator) + await response.body_iterator.aclose() + assert CALLERS == ["acct-7", "acct-7"] + + def test_unknown_subject_is_404(client: TestClient, druks_db): assert client.get("/api/faketest/thing/nope").status_code == 404 # An id the subject could never wear misses the same way, row or no row. diff --git a/docs/writing-an-extension.md b/docs/writing-an-extension.md index 058be4c3..d74a41c9 100644 --- a/docs/writing-an-extension.md +++ b/docs/writing-an-extension.md @@ -459,7 +459,7 @@ class Repository(StoredSubject): return self.full_name @classmethod - def list_summaries(cls) -> list[SubjectSummary]: + def list_summaries(cls, account_id: str | None) -> list[SubjectSummary]: return [repository.get_summary() for repository in cls.list_open()] ``` @@ -489,7 +489,7 @@ from druks.workflows import Subject class PullRequest(Subject): @classmethod - def list_summaries(cls) -> list[SubjectSummary]: + def list_summaries(cls, account_id: str | None) -> list[SubjectSummary]: return [pull_request.get_summary() for pull_request in cls.list_open()] ``` @@ -498,8 +498,12 @@ Any id names one of these, so a detail read always answers. Override `owner/repo#7` is a pull request, `nonsense` is a 404. Each subject a workflow declares must implement `list_summaries()`. The board -reads it. Druks checks this at load. If the method is missing, the extension -does not load. The error names the extension, the subject, and the method. +reads it and passes the caller. `account_id` is the signed-in account, or None +outside a request. Use it to scope the rows when each operator has their own +board. Ignore it when every operator shares one board. A model method never +reads request context. Druks checks the method at load. If it is missing, the +extension does not load. The error names the extension, the subject, and the +method. Druks serves the same `/api/night_watch/repository` surface either way: a board, a page for one, and a live stream of either, mounted for every subject your @@ -848,7 +852,7 @@ for connection in NightWatch.acme.list_for_account(account_id): `account_id` is the caller: `self.account_id` in a run body, `current_account_id.get()` in a route, the handler's argument in a -subscriber. `NightWatch.acme.get(connection_id)` returns one connection +subscriber, the platform's argument in `list_summaries`. `NightWatch.acme.get(connection_id)` returns one connection when your own row stored its id. Each connection carries `id`, `scopes`, `identity` — the provider's facts for the sign-in — `account_id` — the druks account that signed it in — and `connected_at`. The handle serves