Expose searchable Activity history with exact destinations (DRU-511)

backend/druks/events/builder.py

@@ -1,32 +1,75 @@
-from sqlalchemy import or_, select
+from datetime import datetime
 
+from sqlalchemy import select
+
+from druks.apps.loader import iter_apps
 from druks.database import db_session
+from druks.durable.models import AgentCall, Artifact, Run
 from druks.events.feed import FeedItem
 from druks.events.models import Event
 
-_PAGE_LIMIT_DEFAULT = 200
-_FETCH_LIMIT = 500
-
 
 async def build_feed(
     *,
     app: str | None = None,
+    q: str | None = None,
+    kind: str | None = None,
+    from_at: datetime | None = None,
+    until: datetime | None = None,
     before: int | None = None,
-    limit: int = _PAGE_LIMIT_DEFAULT,
+    after: int | None = None,
+    limit: int = 200,
 ) -> tuple[list[FeedItem], str | None]:
-    items = [FeedItem.model_validate(event) for event in await _events(app, before)]
-    items.sort(key=lambda item: item.seq, reverse=True)
-    page = items[:limit]
-    next_cursor = str(page[-1].seq) if len(page) == limit and page else None
+    """Read one page of recorded Activity and its current destination availability."""
+    statement = Event.get_history(app=app).order_by(Event.id.desc())
+    if q and q.strip():
+        pattern = q.strip().replace("/", "//").replace("%", "/%").replace("_", "/_")
+        statement = statement.where(Event.subject_label.ilike(f"%{pattern}%", escape="/"))
+    if kind is not None:
+        statement = statement.where(Event.type == kind)
+    if from_at:
+        statement = statement.where(Event.created_at >= from_at)
+    if until:
+        statement = statement.where(Event.created_at < until)
+    if before is not None:
+        statement = statement.where(Event.id < before)
+    if after is not None:
+        statement = statement.where(Event.id > after)
+    events = list(await db_session().scalars(statement.limit(limit + 1)))
+    page = [FeedItem.model_validate(event) for event in events[:limit]]
+    next_cursor = str(page[-1].seq) if len(events) > limit else None
+    run_ids = {item.run for item in page if item.run}
+    runs = (
+        set(await db_session().scalars(select(Run.id).where(Run.id.in_(run_ids))))
+        if run_ids
+        else set()
+    )
+    artifact_ids = {item.artifact_id for item in page if item.artifact_id}
+    artifacts = {}
+    if artifact_ids:
+        rows = await db_session().execute(
+            select(Artifact, AgentCall)
+            .join(AgentCall, AgentCall.id == Artifact.agent_call_id)
+            .where(Artifact.id.in_(artifact_ids))
+        )
+        artifacts = {
+            artifact.id: bool(call.get_file_path(artifact.path)) for artifact, call in rows
+        }
+    subjects = {
+        (owner.name, subject.subject_type): subject
+        for owner in iter_apps()
+        if not owner.builtin
+        for subject in owner.subjects()
+    }
+    available_subjects = {}
+    for item in page:
+        identity = (item.app, item.subject_type, item.subject_id)
+        if identity not in available_subjects:
+            subject = subjects.get((item.app, item.subject_type))
+            available_subjects[identity] = bool(
+                subject and item.subject_id and await subject.get_for_subject_id(item.subject_id)
+            )
+        item.is_subject_available = available_subjects[identity]
+        item.is_run_available = item.run in runs
+        item.is_artifact_available = artifacts.get(item.artifact_id, False)
     return page, next_cursor
-
-
-async def _events(app: str | None, before: int | None) -> list[Event]:
-    # This app's events plus any unscoped (core) ones. The log stores the app;
-    # the core never derives it from the subject.
-    stmt = select(Event).order_by(Event.id.desc())
-    if before:
-        stmt = stmt.where(Event.id < before)
-    if app:
-        stmt = stmt.where(or_(Event.app == app, Event.app.is_(None)))
-    return list((await db_session().scalars(stmt.limit(_FETCH_LIMIT))).all())

backend/druks/events/feed.py

@@ -1,6 +1,7 @@
 from datetime import datetime
+from typing import Any
 
-from pydantic import AliasPath, ConfigDict, Field, computed_field
+from pydantic import AliasChoices, AliasPath, ConfigDict, Field, computed_field
 
 from druks.schemas import Schema
 
@@ -21,6 +22,35 @@ class FeedItem(Schema):
     subject_type: str | None = None
     subject_id: str | None = None
     subject_label: str | None = None
+    run: str | None = Field(default=None, validation_alias=AliasPath("payload", "run"))
+    gate: str | None = Field(default=None, validation_alias=AliasPath("payload", "gate"))
+    parked_at: datetime | None = Field(
+        default=None, validation_alias=AliasPath("payload", "input_requested_at")
+    )
+    input_request: dict[str, Any] | None = Field(
+        default=None, validation_alias=AliasPath("payload", "input_request")
+    )
+    result: Any = Field(default=None, validation_alias=AliasPath("payload", "result"))
+    summary: str | None = Field(default=None, validation_alias=AliasPath("payload", "summary"))
+    reason: str | None = Field(
+        default=None,
+        validation_alias=AliasChoices(
+            AliasPath("payload", "reason"), AliasPath("payload", "failure")
+        ),
+    )
+    artifact_id: str | None = Field(
+        default=None,
+        validation_alias=AliasChoices(
+            AliasPath("payload", "artifact_id"),
+            AliasPath("payload", "input_request", "artifact_id"),
+        ),
+    )
+    agent_call_id: str | None = Field(
+        default=None, validation_alias=AliasPath("payload", "agent_call_id")
+    )
+    is_subject_available: bool = False
+    is_run_available: bool = False
+    is_artifact_available: bool = False
 
     @computed_field
     @property
@@ -32,3 +62,4 @@ class FeedResponse(Schema):
     items: list[FeedItem]
     # Event sequence cursor for the next (older) page; None at the tail.
     next_cursor: str | None = None
+    kinds: list[str] | None = None

backend/druks/events/models.py

@@ -1,7 +1,7 @@
 from datetime import datetime
 from typing import Any
 
-from sqlalchemy import Index
+from sqlalchemy import Index, Select, and_, or_, select
 from sqlalchemy.dialects.postgresql import JSONB
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import Mapped, mapped_column
@@ -32,6 +32,36 @@ class Event(Base):
     created_at: Mapped[datetime] = mapped_column(default=Base.utc_now)
     payload: Mapped[dict[str, Any]] = mapped_column(JSONB, default=dict)
 
+    @classmethod
+    def get_history(cls, *, app: str | None = None) -> Select[tuple["Event"]]:
+        # App imports Event while the loader imports App.
+        from druks.apps.loader import iter_apps
+
+        owners = [owner.name for owner in iter_apps() if not owner.builtin]
+        statement = select(cls).where(
+            cls.app.in_(owners),
+            or_(
+                cls.type.not_like("workflow.%"),
+                cls.type.in_(
+                    [
+                        "workflow.scheduled",
+                        "workflow.parked",
+                        "workflow.failed",
+                        "workflow.cancelled",
+                    ]
+                ),
+                and_(
+                    cls.type == "workflow.running",
+                    cls.payload["gate"].astext != "",
+                    cls.payload["input_requested_at"].astext != "",
+                    cls.payload["result"].astext.is_not(None),
+                ),
+            ),
+        )
+        if app is not None:
+            statement = statement.where(cls.app == app)
+        return statement
+
     @classmethod
     async def emit(
         cls,

backend/druks/events/routes.py

@@ -1,75 +1,104 @@
 import asyncio
+from datetime import UTC
+from typing import Annotated
 
-from fastapi import APIRouter, HTTPException, Query, Request, status
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
 from fastapi.responses import StreamingResponse
+from pydantic import AwareDatetime
 
 from druks.api.dependencies import EngineDep
-from druks.database import session_scope
+from druks.database import db_session, session_scope
 from druks.durable.live import SSE_HEADERS
 from druks.events.builder import build_feed
 from druks.events.feed import FeedResponse
+from druks.events.models import Event
 
 router = APIRouter(prefix="/api/events", tags=["feed"])
 
-# Per-connection SSE poll cadence. Short enough that the operator's screen feels
-# live, long enough that we're not hammering the DB; the cost is one bounded
-# read per tick, so cadence is set by perceived latency rather than load.
 _SSE_POLL_INTERVAL_SECONDS = 2.0
 
 
 def _parse_cursor(raw: str | None) -> int | None:
-    # The cursor is a feed sequence (an event's monotonic pk), opaque to the client —
-    # it hands back whatever ``next_cursor`` returned.
     if raw is None:
-        return None
+        return
     try:
-        return int(raw)
-    except ValueError as exc:
+        cursor = int(raw)
+        if cursor < 0:
+            raise ValueError
+        return cursor
+    except ValueError as error:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
-            detail=f"Invalid ``before`` cursor: {raw!r}",
-        ) from exc
+            detail=f"Invalid event cursor: {raw!r}. Use a returned sequence.",
+        ) from error
 
 
-@router.get("", response_model=FeedResponse, response_model_by_alias=True)
+def get_filters(
+    app: str | None = Query(default=None),
+    q: str | None = Query(default=None),
+    kind: str | None = Query(default=None),
+    from_at: Annotated[AwareDatetime | None, Query(alias="from")] = None,
+    until: AwareDatetime | None = Query(default=None),
+) -> dict:
+    if from_at and until and from_at >= until:
+        raise HTTPException(status.HTTP_422_UNPROCESSABLE_CONTENT, "until must be after from.")
+    return {
+        "app": app,
+        "q": q,
+        "kind": kind,
+        "from_at": from_at.astimezone(UTC) if from_at else None,
+        "until": until.astimezone(UTC) if until else None,
+    }
+
+
+@router.get(
+    "", response_model=FeedResponse, response_model_by_alias=True, response_model_exclude_unset=True
+)
 async def list_feed(
+    filters: Annotated[dict, Depends(get_filters)],
     limit: int = Query(default=200, ge=1, le=500),
     before: str | None = Query(default=None),
-    app: str | None = Query(default=None),
 ) -> FeedResponse:
     cursor = _parse_cursor(before)
-    items, next_cursor = await build_feed(app=app, before=cursor, limit=limit)
-    return FeedResponse(items=items, next_cursor=next_cursor)
+    items, next_cursor = await build_feed(**filters, before=cursor, limit=limit)
+    response = FeedResponse(items=items, next_cursor=next_cursor)
+    if before is None:
+        statement = (
+            Event.get_history(app=filters["app"])
+            .with_only_columns(Event.type)
+            .distinct()
+            .order_by(Event.type)
+        )
+        response.kinds = list(await db_session().scalars(statement))
+    return response
 
 
 @router.get("/stream")
 async def stream_feed(
     request: Request,
     engine: EngineDep,
-    app: str | None = Query(default=None),
+    filters: Annotated[dict, Depends(get_filters)],
+    after: str | None = Query(default=None),
 ) -> StreamingResponse:
+    last_seq = _parse_cursor(request.headers.get("last-event-id") or after)
+
     async def feed_stream():
-        last_seq: int | None = None
-        first = True
+        nonlocal last_seq
         while True:
             if await request.is_disconnected():
                 return
-            # New Session per tick so we don't hold a transaction open across the
-            # sleep; the open/close cost is irrelevant against the poll cadence.
             async with session_scope(engine):
-                items, _next_cursor = await build_feed(
-                    app=app,
-                    before=None,
-                    limit=100 if first else 50,
-                )
-            # Strictly past the last emitted sequence — the monotonic pk never ties,
-            # so this neither re-sends the boundary event nor drops a same-second one.
-            fresh = items if last_seq is None else [e for e in items if e.seq > last_seq]
-            for item in reversed(fresh):  # oldest-first within a tick
-                yield f"data: {item.model_dump_json(by_alias=True)}\n\n"
-            if fresh:
-                last_seq = fresh[0].seq  # newest just-emitted (page is seq-desc)
-            first = False
+                items, cursor = await build_feed(**filters, after=last_seq, limit=100)
+                # Catch-up can span several pages. Finish the interval before advancing its head.
+                while last_seq is not None and cursor:
+                    older, cursor = await build_feed(
+                        **filters, after=last_seq, before=int(cursor), limit=100
+                    )
+                    items.extend(older)
+            for item in reversed(items):
+                yield f"id: {item.seq}\ndata: {item.model_dump_json(by_alias=True)}\n\n"
+            if items:
+                last_seq = items[0].seq
             try:
                 await asyncio.sleep(_SSE_POLL_INTERVAL_SECONDS)
             except asyncio.CancelledError: