A workflow's subject is one word, read off the class or off the run

backend/druks/durable/models.py

@@ -344,13 +344,20 @@ class AgentCall(Base, Uuid7Pk):
     # and treat 404 as the host being gone.
     sandbox_host_id: Mapped[str]
 
-    def get_live_status(self) -> str:
-        # Unfinished: "running" while the run is live, "abandoned" once it's terminal.
+    @property
+    def live_status(self) -> AgentCallStatus:
+        # Unfinished: running while the run is live, abandoned once it's terminal.
         if self.finished_at:
-            return AgentCallStatus(self.status).value
+            return AgentCallStatus(self.status)
         if self.run.is_active:
-            return "running"
-        return "abandoned"
+            return AgentCallStatus.RUNNING
+        return AgentCallStatus.ABANDONED
+
+    @property
+    def token_usage(self) -> dict[str, int] | None:
+        # Each harness reports tokens its own way; this is the one shape, and None
+        # once a call reported nothing countable.
+        return normalize_token_usage(self.cost_metadata)
 
     @property
     def artifact_dir(self) -> str:

backend/druks/durable/reads.py

@@ -198,7 +198,7 @@ async def stream_transcript(
         with session_scope(engine):
             call = AgentCall.get(call_id)
             path = call.get_stream_path(stream) if call else None
-            summary = AgentCallResponse.from_call(call) if call else None
+            summary = AgentCallResponse.model_validate(call) if call else None
 
         if not summary:
             # Unknown (or deleted) call: nothing will ever arrive — close the

backend/druks/durable/schemas.py

@@ -2,12 +2,11 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, Annotated, Any, Literal
 
-from pydantic import BeforeValidator, ConfigDict, Field, SerializeAsAny
+from pydantic import AliasPath, BeforeValidator, ConfigDict, Field, SerializeAsAny, computed_field
 
-from druks.harnesses.artifacts import normalize_token_usage
 from druks.schemas import BaseResponse
 
-from .enums import RunState
+from .enums import AgentCallStatus, RunState
 
 if TYPE_CHECKING:
     from .models import AgentCall, Artifact, Run
@@ -22,46 +21,32 @@ class TokenUsage(BaseResponse):
     total_tokens: int
 
 
-def _token_usage(cost_metadata: dict | None) -> TokenUsage | None:
-    canonical = normalize_token_usage(cost_metadata)
-    return TokenUsage(**canonical) if canonical else None
-
-
 def get_display_label(kind: str) -> str:
     # "ship.build" → "Build"; "implement" → "Implement".
     return kind.rsplit(".", 1)[-1].replace("_", " ").capitalize()
 
 
 class AgentCallResponse(BaseResponse):
+    model_config = ConfigDict(from_attributes=True)
+
     id: str
     # Which agent made this call ("scope", "implement") — the timeline's row label.
     agent: str | None = None
-    label: str = ""
     # The account charged — differs from the run's on fallback.
-    account_username: str
-    status: Literal["running", "succeeded", "failed", "abandoned"]
+    account_username: str = Field(validation_alias=AliasPath("account", "username"))
+    status: AgentCallStatus = Field(validation_alias="live_status")
     # started_at + finished_at are the facts; the client derives elapsed (a live
     # tick off started_at while running), so nothing here churns between polls.
     started_at: datetime
     finished_at: datetime | None = None
     last_error: str | None = None
     cost_usd: float | None = None
-    tokens: TokenUsage | None = None
+    tokens: TokenUsage | None = Field(default=None, validation_alias="token_usage")
 
-    @classmethod
-    def from_call(cls, call: "AgentCall") -> "AgentCallResponse":
-        return cls(
-            id=call.id,
-            agent=call.agent,
-            label=get_display_label(call.agent) if call.agent else "Agent",
-            account_username=call.account.username,
-            status=call.get_live_status(),  # type: ignore[arg-type]
-            started_at=call.started_at,
-            finished_at=call.finished_at,
-            last_error=call.last_error,
-            cost_usd=call.cost_usd,
-            tokens=_token_usage(call.cost_metadata),
-        )
+    @computed_field
+    @property
+    def label(self) -> str:
+        return get_display_label(self.agent) if self.agent else "Agent"
 
 
 class ArtifactFile(BaseResponse):
@@ -159,7 +144,7 @@ def from_run(
             created_at=run.created_at,
             updated_at=run.updated_at,
             account_username=run.account.username,
-            agent_calls=[AgentCallResponse.from_call(c) for c in calls],
+            agent_calls=[AgentCallResponse.model_validate(call) for call in calls],
         )
 
 

backend/druks/events/builder.py

@@ -14,7 +14,7 @@ def build_feed(
     before: int | None = None,
     limit: int = _PAGE_LIMIT_DEFAULT,
 ) -> tuple[list[FeedItem], str | None]:
-    items = [FeedItem.from_event(event) for event in _events(extension, before)]
+    items = [FeedItem.model_validate(event) for event in _events(extension, 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

backend/druks/events/feed.py

@@ -1,38 +1,31 @@
 from datetime import datetime
 
-from druks.events.models import Event
+from pydantic import AliasPath, ConfigDict, Field, computed_field
+
 from druks.schemas import BaseResponse
 
 
 class FeedItem(BaseResponse):
-    id: str
+    model_config = ConfigDict(from_attributes=True)
+
     # The event's monotonic log position (its pk) — the feed's ordering and
     # pagination key. ``at`` is whole-second and ties constantly; this never does.
-    seq: int
-    at: datetime
+    seq: int = Field(validation_alias="id")
+    at: datetime = Field(validation_alias="created_at")
     # The event type verbatim: a lifecycle topic ("workflow.finished") or the
     # milestone an extension recorded ("shipped"). The words are the client's.
-    kind: str
+    kind: str = Field(validation_alias="type")
     extension: str | None = None
     # The durable kind of the workflow a lifecycle row is about ("ship.build").
-    workflow: str | None = None
+    workflow: str | None = Field(default=None, validation_alias=AliasPath("payload", "kind"))
     subject_type: str | None = None
     subject_id: str | None = None
     subject_label: str | None = None
 
-    @classmethod
-    def from_event(cls, event: Event) -> "FeedItem":
-        return cls(
-            id=f"event:{event.id}",
-            seq=event.id,
-            at=event.created_at,
-            kind=event.type,
-            extension=event.extension,
-            workflow=event.payload.get("kind"),
-            subject_type=event.subject_type,
-            subject_id=event.subject_id,
-            subject_label=event.subject_label,
-        )
+    @computed_field
+    @property
+    def id(self) -> str:
+        return f"event:{self.seq}"
 
 
 class FeedResponse(BaseResponse):

backend/druks/extensions/base.py

@@ -158,7 +158,7 @@ def subject_classes(cls) -> "list[type[Subject] | type[StoredSubject]]":
         """What this extension's runs are about, read off the workflows that declare
         them — each one gets a board and a page, ordered by subject type so the routes
         it mounts are stable."""
-        declared = {wf._subject_class for wf in cls.workflows() if wf._subject_class}
+        declared = {wf.subject for wf in cls.workflows() if wf.subject}
         for subject_class in declared:
             if subject_class.subject_type == "transcripts":
                 raise TypeError(
@@ -335,7 +335,7 @@ async def get_transcript(
             call = AgentCall.get(call_id)
             if not call:
                 raise HTTPException(status.HTTP_404_NOT_FOUND, "Run not found.")
-            if call.get_live_status() == AgentCallStatus.RUNNING:
+            if call.live_status == AgentCallStatus.RUNNING:
                 response.headers["Cache-Control"] = "no-store"
             else:
                 response.headers["Cache-Control"] = settled_cache

backend/druks/mcp/gateway/services.py

@@ -76,7 +76,7 @@ def get_agent_call(call_id: str) -> schemas.AgentCallDetailResponse:
     layout = call.artifact_layout
     return schemas.AgentCallDetailResponse(
         run_id=call.run_id,
-        call=AgentCallResponse.from_call(call),
+        call=AgentCallResponse.model_validate(call),
         transcript=read_slice(
             layout.transcript, offset=-_TRANSCRIPT_TAIL_BYTES, limit=_TRANSCRIPT_TAIL_BYTES
         ).text,

backend/druks/signals.py

@@ -46,7 +46,7 @@ def register(fn: Subscriber) -> Subscriber:
             )
         if workflow_class:
             filters["kind"] = workflow_class.kind
-            subject_class = workflow_class._subject_class
+            subject_class = workflow_class.subject
         if subject_class:
             if not (
                 isinstance(subject_class, type)

backend/druks/workflows.py

@@ -132,20 +132,37 @@ def _resolve_body_method(cls: type["Workflow"]) -> str:
     )
 
 
-def _resolve_subject_class(cls: type["Workflow"]) -> "type[Subject] | type[StoredSubject] | None":
-    # The declaration and the run's live subject are the same word, so the class
-    # attribute comes off and the property answers in its place.
+class _DeclaredSubject:
+    """``subject = WorkItem`` on a workflow: the kind of thing its runs are about, read
+    off the workflow, and the particular one a run is about, read off that run."""
+
+    def __init__(self, subject_class: "type[Subject] | type[StoredSubject] | None") -> None:
+        self.subject_class = subject_class
+
+    def __get__(self, run: "Workflow | None", owner: type) -> Any:
+        if run is None:
+            return self.subject_class
+        # Live, not a snapshot taken at dispatch: a long-parked run resumes against
+        # whatever the declared class says then, and finds nothing if it went away.
+        if not run._subject:
+            return
+        return self.subject_class.get_for_subject_id(str(run._subject["id"]))
+
+
+def _declare_subject(cls: type["Workflow"]) -> None:
+    # The author writes a plain class attribute; it becomes the descriptor above.
+    # Presence in ``cls.__dict__``, not the value: an explicit ``subject = None`` is a
+    # declaration of nothing, which the error below names rather than passing over.
     if "subject" not in cls.__dict__:
-        return cls._subject_class
+        return
     declared = cls.__dict__["subject"]
     if not (isinstance(declared, type) and issubclass(declared, Subject | StoredSubject)):
         raise WorkflowError(
             f"{cls.__name__}.subject must be a Subject or StoredSubject subclass — a run "
             f"is about a kind of thing, not {declared!r}. A workflow about nothing "
             "declares no subject at all."
         )
-    del cls.subject
-    return declared
+    cls.subject = _DeclaredSubject(declared)
 
 
 def _input_model_from_signature(cls: type["Workflow"]) -> type[BaseModel] | None:
@@ -483,10 +500,9 @@ class Workflow:
     # the loader's package registrations at definition time and namespacing
     # ``kind``. Never supplied or stored per run.
     extension: ClassVar[str | None] = None
-    # The subject class this workflow's runs are about, written ``subject = WorkItem``
-    # on the subclass and lifted off it in __init_subclass__ so the instance property
-    # below keeps the name. None for a workflow about nothing, which says so by silence.
-    _subject_class: ClassVar[type[Subject] | type[StoredSubject] | None] = None
+    # What this workflow's runs are about, written ``subject = WorkItem`` on the
+    # subclass. None for a workflow about nothing, which says so by silence.
+    subject = _DeclaredSubject(None)
     # When set to a cron string, the workflow also registers a schedule that
     # fires its run() on that cadence (no subject — a framework cron).
     every: ClassVar[str | None] = None
@@ -532,7 +548,7 @@ def __init_subclass__(cls, **kwargs: Any) -> None:
                 "the declaring extension supplies the namespace"
             )
         cls.kind = f"{cls.extension}.{local_kind}" if cls.extension else local_kind
-        cls._subject_class = _resolve_subject_class(cls)
+        _declare_subject(cls)
         validate_settings_declaration(cls.Settings)
         cls._body_method = _resolve_body_method(cls)
         # Before _wrap_steps: run()'s wrapper signature is (*args, **kwargs).
@@ -565,14 +581,6 @@ def __init__(self) -> None:
         # its lease expiry decides when it must rotate.
         self._host: Sandbox | None = None
 
-    @property
-    def subject(self) -> Any:
-        # Live, not a snapshot taken at dispatch: a long-parked run resumes against
-        # whatever the declared class says then, and finds nothing if it went away.
-        if not self._subject:
-            return
-        return self._subject_class.get_for_subject_id(str(self._subject["id"]))
-
     async def announce(self, topic: str, **facts: Any) -> None:
         # The workflow announcing a domain event in its extension's vocabulary
         # ("pr.opened", pr_number=12, branch="agent/eng-8"). The platform injects
@@ -703,13 +711,11 @@ def override_setting(cls, field: str, value: Any) -> None:
     def _validate_subject(cls, subject: "Subject | StoredSubject | None") -> None:
         # The declaration is the contract — subscribers derive their subject class
         # from ``workflow=``, so it has to hold for every run.
-        if cls._subject_class:
-            if isinstance(subject, cls._subject_class):
+        if cls.subject:
+            if isinstance(subject, cls.subject):
                 return
             given = "nothing" if subject is None else type(subject).__name__
-            raise WorkflowError(
-                f"{cls.__name__} is about {cls._subject_class.__name__}, not {given}"
-            )
+            raise WorkflowError(f"{cls.__name__} is about {cls.subject.__name__}, not {given}")
         if subject is None:
             return
         raise WorkflowError(

backend/tests/druks-field_notes/druks_field_notes/contracts.py

@@ -1,6 +1,6 @@
 from druks.agents import AgentOutput
 
 
-class NoteSummary(AgentOutput):
-    # What the summarizer agent returns: a one-line distillation of the note it read.
-    summary: str
+class GistOutput(AgentOutput):
+    # What the summarizer agent returns: the note it read, in one line.
+    gist: str

backend/tests/druks-field_notes/druks_field_notes/extension.py

@@ -6,7 +6,7 @@
 from druks.extensions import Extension
 from pydantic import BaseModel, Field, SecretStr, field_validator
 
-from druks_field_notes.contracts import NoteSummary
+from druks_field_notes.contracts import GistOutput
 
 if TYPE_CHECKING:
     from druks.settings import Settings
@@ -33,7 +33,7 @@ def check_summary_api_key(settings: "Settings") -> CheckResult:
 class FieldNotes(Extension):
     name = "field_notes"
     icon = "notebook"
-    description = "Turns a jotted observation into a one-line summary with an agent."
+    description = "Turns a jotted observation into a one-line gist with an agent."
 
     class Settings(BaseModel):
         # How many recent notes the board shows — an operator knob, so it lives here.
@@ -71,11 +71,11 @@ def _well_formed_token(cls, value: SecretStr | None) -> SecretStr | None:
                 raise ValueError(f"sync token {value.get_secret_value()!r} must start with 'sk-'")
             return value
 
-    # The one agent this extension runs: it reads a note and writes its summary.
+    # The one agent this extension runs: it reads a note and writes its gist.
     summarize = Agent(
-        description="reads a note and writes a one-line summary",
+        description="reads a note and writes its one-line gist",
         prompt="field_notes/summarize.md",
-        contract=NoteSummary,
+        contract=GistOutput,
         model="claude",
     )
 

backend/tests/druks-field_notes/druks_field_notes/migrations/versions/field_notes_0001_notes.py

@@ -22,7 +22,7 @@ def upgrade() -> None:
         "field_notes_notes",
         sa.Column("id", sa.Integer(), nullable=False),
         sa.Column("body", sa.String(), nullable=False),
-        sa.Column("summary", sa.String(), nullable=True),
+        sa.Column("gist", sa.String(), nullable=True),
         sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
         sa.PrimaryKeyConstraint("id"),
     )

backend/tests/druks-field_notes/druks_field_notes/models.py

@@ -11,11 +11,10 @@ class Note(StoredSubject):
     __tablename__ = "field_notes_notes"
 
     # What the note is about — the raw observation an operator jotted down. A run's
-    # agent reads this and writes back a one-line summary.
+    # agent reads this and writes back its gist.
     body: Mapped[str]
-    # The agent's summary of ``body``, written when a Summarize run finishes. None
-    # until then.
-    summary: Mapped[str | None]
+    # ``body`` in one line, written when a Summarize run finishes. None until then.
+    gist: Mapped[str | None]
     created_at: Mapped[datetime] = mapped_column(default=StoredSubject.utc_now)
 
     @classmethod
@@ -35,8 +34,8 @@ def list_recent(cls, *, limit: int = 100) -> list["Note"]:
         stmt = select(cls).order_by(cls.created_at.desc(), cls.id.desc()).limit(limit)
         return list(db_session().scalars(stmt))
 
-    def save_summary(self, summary: str) -> None:
-        self.summary = summary
+    def save_gist(self, gist: str) -> None:
+        self.gist = gist
         db_session().flush()
 
     def get_summary(self) -> NoteSummary:

backend/tests/druks-field_notes/druks_field_notes/schemas.py

@@ -7,5 +7,5 @@ class NoteSummary(SubjectSummary):
     # The note's domain header — what only field_notes knows. The platform's subject
     # read-side composes it with the generic status + timeline.
     body: str
-    summary: str | None = None
+    gist: str | None = None
     created_at: datetime