diff --git a/backend/druks/build/extension.py b/backend/druks/build/extension.py
index e57c4364..2e26e416 100644
--- a/backend/druks/build/extension.py
+++ b/backend/druks/build/extension.py
@@ -1,7 +1,6 @@
 from typing import TYPE_CHECKING
 
 from pydantic import BaseModel, Field
-from sqlalchemy import select
 
 from druks.agents import Agent
 from druks.build.contracts import (
@@ -17,10 +16,9 @@
 from druks.build.models import WorkItem
 from druks.build.scoping.contracts import ScopeBriefOutput
 from druks.db import db_session
-from druks.durable.enums import RunState
 from druks.events import Event, FeedItem
 from druks.extensions import Extension
-from druks.workflows import Run, SubjectActivity, get_run_phase
+from druks.workflows import Run, RunState, SubjectActivity, get_run_phase
 
 if TYPE_CHECKING:
     from druks.build.schemas import WorkItemSummary
@@ -219,13 +217,8 @@ async def subject_activity(cls, subject_id: str) -> SubjectActivity | None:
         assert item is not None  # the read-side resolves the summary before the activity
         run = item.get_build_run()
         if not run:
-            newest_scope = (
-                select(Run)
-                .where(Run.kind == Scope.kind, Run.subject["id"].as_integer() == item.id)
-                .order_by(Run.created_at.desc())
-                .limit(1)
-            )
-            run = db_session().scalars(newest_scope).first()
+            scope_runs = Run.list_for_subject("work_item", str(item.id), kind=Scope.kind)
+            run = scope_runs[0] if scope_runs else None
         # Only while a run is actually RUNNING — a run parked on a gate isn't working.
         if not run or run.state != RunState.RUNNING.value:
             return None
diff --git a/backend/druks/build/scoping/workflows.py b/backend/druks/build/scoping/workflows.py
index 55944dc4..e117e75d 100644
--- a/backend/druks/build/scoping/workflows.py
+++ b/backend/druks/build/scoping/workflows.py
@@ -1,14 +1,10 @@
 import logging
 
-from sqlalchemy import select
-
 from druks.build.extension import Build
 from druks.build.models import Project, ProjectRepo, WorkItem
 from druks.build.scoping.contracts import ScopeBriefOutput
-from druks.db import db_session
-from druks.durable.enums import RunState
 from druks.ticketing.datastructures import Ticket
-from druks.workflows import Gate, Run, Workflow
+from druks.workflows import Gate, Run, RunState, Workflow
 
 logger = logging.getLogger(__name__)
 
@@ -57,12 +53,8 @@ async def dispatch(cls, *, ticket: Ticket) -> str | None:
 
     @classmethod
     def parked_for(cls, work_item_id: int) -> Run | None:
-        stmt = select(Run).where(
-            Run.kind == cls.kind,
-            Run.state == RunState.PENDING_INPUT.value,
-            Run.subject["id"].as_integer() == work_item_id,
-        )
-        return db_session().scalars(stmt).first()
+        runs = Run.list_for_subject("work_item", str(work_item_id), kind=cls.kind)
+        return next((run for run in runs if run.state == RunState.PENDING_INPUT.value), None)
 
     async def get_prompt_context(self, **context: object) -> dict[str, object]:
         # Everything the agent needs beyond the ticket it fetches itself: where
diff --git a/backend/druks/durable/dbos_state.py b/backend/druks/durable/dbos_state.py
index ad0e2504..b3246f93 100644
--- a/backend/druks/durable/dbos_state.py
+++ b/backend/druks/durable/dbos_state.py
@@ -1,6 +1,7 @@
 from datetime import timedelta
 
 import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import JSONB
 
 from druks.durable.enums import RunState
 
@@ -14,19 +15,48 @@
 # its executor destroyed) and reads orphaned rather than scheduled forever.
 _MISSING_STATUS_GRACE = timedelta(minutes=5)
 
-# Read-only handle on the columns the derivation needs; DBOS owns and migrates
-# the real table. NOTE: DBOS workflow_status GC/retention must stay off while
-# durable_runs keeps history — a purged workflow row derives as orphaned once
-# past the grace window.
+# Read-only handle on the columns the derivation and subject keying need; DBOS
+# owns and migrates the real table. NOTE: DBOS workflow_status GC/retention must
+# stay off — a purged workflow row derives as orphaned once past the grace
+# window and falls out of its subject's timeline.
 workflow_status = sa.Table(
     "workflow_status",
     sa.MetaData(schema=DBOS_SYSTEM_SCHEMA),
     sa.Column("workflow_uuid", sa.String, primary_key=True),
     sa.Column("status", sa.String),
     sa.Column("updated_at", sa.BigInteger),
+    sa.Column("attributes", JSONB),
 )
 
 
+def extension_expression(run_id: sa.ColumnElement) -> sa.ColumnElement:
+    return (
+        sa.select(workflow_status.c.attributes["extension"].as_string())
+        .where(workflow_status.c.workflow_uuid == run_id)
+        .correlate_except(workflow_status)
+        .scalar_subquery()
+    )
+
+
+def subject_filter(
+    run_id: sa.ColumnElement, subject_type: str, subject_id: str
+) -> sa.ColumnElement:
+    # "This run is about that subject" — the predicate every runs-for-a-subject
+    # query composes, reading the attributes stamped at start(). A fresh alias
+    # per call keeps it independent of the state/updated_at subqueries, which
+    # claim the bare workflow_status table via correlate_except.
+    ws = workflow_status.alias()
+    return (
+        sa.select(ws.c.workflow_uuid)
+        .where(
+            ws.c.workflow_uuid == run_id,
+            ws.c.attributes["subject_type"].as_string() == subject_type,
+            ws.c.attributes["subject_id"].as_string() == subject_id,
+        )
+        .exists()
+    )
+
+
 def state_expression(
     run_id: sa.ColumnElement, input_gate: sa.ColumnElement, created_at: sa.ColumnElement
 ) -> sa.ColumnElement:
diff --git a/backend/druks/durable/models.py b/backend/druks/durable/models.py
index fc7b3272..645d1d78 100644
--- a/backend/druks/durable/models.py
+++ b/backend/druks/durable/models.py
@@ -11,7 +11,12 @@
 
 from druks.core.models import Uuid7Pk
 from druks.database import db_session, get_session
-from druks.durable.dbos_state import state_expression, updated_at_expression
+from druks.durable.dbos_state import (
+    extension_expression,
+    state_expression,
+    subject_filter,
+    updated_at_expression,
+)
 from druks.durable.enums import ACTIVE_STATES
 from druks.harnesses.artifacts import normalize_token_usage
 from druks.models import Base
@@ -44,19 +49,15 @@ class Run(Base):
     # error, so read-sides tell e.g. a gate timeout from a crash without parsing
     # `failure`. Empty/None for a crash.
     failure_code: Mapped[str | None] = mapped_column(default=None)
-    # run()'s validated input, dumped to JSON at start(); {} for an input-less run.
-    input: Mapped[dict[str, Any]] = mapped_column(JSONB, default=dict)
-    # What the run is about ({type, id}); None for a framework cron. Routing
-    # metadata, not body input — its own column so events and reads key off it.
-    subject: Mapped[dict[str, Any] | None] = mapped_column(JSONB, default=None)
-    # The extension that launched the run, stamped onto its events.
-    extension: Mapped[str | None] = mapped_column(default=None)
     created_at: Mapped[datetime] = mapped_column(default=Base.utc_now)
     # Derived, never stored: DBOS owns whether the workflow is live — every
     # SELECT (session.get() on a not-yet-loaded instance included) computes it
     # fresh; an already-loaded instance keeps what it read until expired.
     # Read-only; an operator ends a run through cancel().
     state: Mapped[str] = column_property(state_expression(id, input_gate, created_at))
+    # The extension that launched the run, read back from the attributes
+    # stamped at start(); None for a framework cron.
+    extension: Mapped[str | None] = column_property(extension_expression(id))
     # When the run last changed — the newest of creation, the parked ask, and
     # DBOS's status write.
     updated_at: Mapped[datetime] = column_property(
@@ -68,27 +69,14 @@ def is_active(self) -> bool:
         return self.state in {s.value for s in ACTIVE_STATES}
 
     @classmethod
-    def create_row(
-        cls,
-        engine,
-        *,
-        workflow_id: str,
-        kind: str,
-        input: dict[str, Any],
-        subject: dict[str, Any] | None = None,
-        extension: str | None = None,
-    ) -> None:
+    def create_row(cls, engine, *, workflow_id: str, kind: str) -> None:
         # Own committed transaction (not the caller's request txn) so the row
         # exists before the running workflow's first lifecycle event. Idempotent:
         # a scheduled run creates its row inside the (replayable) body, and a
         # start that races its own retry must not double-insert.
         with get_session(engine) as session:
             session.execute(
-                pg_insert(cls)
-                .values(
-                    id=workflow_id, kind=kind, input=input, subject=subject, extension=extension
-                )
-                .on_conflict_do_nothing()
+                pg_insert(cls).values(id=workflow_id, kind=kind).on_conflict_do_nothing()
             )
             session.commit()
 
@@ -97,17 +85,21 @@ def get(cls, workflow_id: str) -> "Run | None":
         return db_session().get(cls, workflow_id)
 
     @classmethod
-    def list_for_subject(cls, subject_type: str, subject_id: str) -> list["Run"]:
+    def list_for_subject(
+        cls, subject_type: str, subject_id: str, kind: str | None = None
+    ) -> list["Run"]:
         # Every run about this subject (stamped at start), newest first — a
-        # subject's lifecycle spans many runs, so its status is theirs aggregated.
+        # subject's lifecycle spans many runs, so its status is theirs
+        # aggregated. ``kind`` narrows to one workflow's runs; per (kind,
+        # subject) the queue dedup makes runs strictly sequential, so newest
+        # first holds within a kind too.
         stmt = (
             select(cls)
-            .where(
-                cls.subject["type"].as_string() == subject_type,
-                cls.subject["id"].as_string() == subject_id,
-            )
+            .where(subject_filter(cls.id, subject_type, subject_id))
             .order_by(cls.updated_at.desc())
         )
+        if kind:
+            stmt = stmt.where(cls.kind == kind)
         return list(db_session().scalars(stmt))
 
     def get_ask(self) -> dict[str, Any]:
@@ -150,16 +142,17 @@ def get_rendered_ask(self) -> dict[str, Any]:
         # actions; url is an optional gate-author-declared view-link.
         return {"body": ask["label"], "actions": None, "deep_link": ask.get("url")}
 
-    def create_park_notification(self, destination_id: str) -> str:
+    def create_park_notification(self, destination_id: str, subject: dict[str, Any]) -> str:
         # Create the notification for the round this run just parked on — the
-        # caller enqueues delivery. run_id + run_parked_at snapshot the round
-        # so a click on an old button can be refused once the run re-parks.
+        # caller supplies the run's subject and enqueues delivery. run_id +
+        # run_parked_at snapshot the round so a click on an old button can be
+        # refused once the run re-parks.
         rendered = self.get_rendered_ask()
         notification = Notification.create(
             destination_id=destination_id,
             reason="gate.parked",
             body=rendered["body"],
-            subject=self.subject,
+            subject=subject,
             actions=rendered["actions"],
             run_id=self.id,
             run_parked_at=self.input_requested_at,
@@ -362,11 +355,7 @@ def by_run(cls, run_ids: list[str]) -> dict[str, list["AgentCall"]]:
     def list_for_subject(cls, subject_type: str, subject_id: str) -> list["AgentCall"]:
         stmt = (
             select(cls)
-            .join(Run, cls.run_id == Run.id)
-            .where(
-                Run.subject["type"].as_string() == subject_type,
-                Run.subject["id"].as_string() == subject_id,
-            )
+            .where(subject_filter(cls.run_id, subject_type, subject_id))
             .order_by(cls.created_at, cls.id)
         )
         return list(db_session().scalars(stmt))
diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py
index b48df923..48a8b138 100644
--- a/backend/druks/workflows.py
+++ b/backend/druks/workflows.py
@@ -9,7 +9,7 @@
 from typing import TYPE_CHECKING, Any, ClassVar, Self, get_type_hints
 
 from croniter import croniter
-from dbos import DBOS, SetEnqueueOptions, SetWorkflowID, StepOptions
+from dbos import DBOS, SetEnqueueOptions, SetWorkflowAttributes, SetWorkflowID, StepOptions
 from dbos._dbos import _get_dbos_instance
 from dbos._error import (
     DBOSAwaitedWorkflowCancelledError,
@@ -222,6 +222,8 @@ async def _park(
     await _emit_run_event(
         workflow.workflow_id,
         RunState.PENDING_INPUT,
+        subject=workflow.subject,
+        extension=workflow.extension,
         facts={
             "input_gate": topic,
             "input_request": input_request,
@@ -230,15 +232,21 @@ async def _park(
     )
     if workflow.subject:
         # Every subjected park notifies the designated destination — no author opt-in.
-        await _notify_designated_destination(workflow.workflow_id)
+        await _notify_designated_destination(workflow.workflow_id, workflow.subject)
     payload = await DBOS.recv_async(topic, timeout_seconds=ttl_seconds)
     if payload is None:
         raise GateTimeout(topic)
-    await _emit_run_event(workflow.workflow_id, RunState.RUNNING, facts=_GATE_CLEARED)
+    await _emit_run_event(
+        workflow.workflow_id,
+        RunState.RUNNING,
+        subject=workflow.subject,
+        extension=workflow.extension,
+        facts=_GATE_CLEARED,
+    )
     return payload
 
 
-async def _notify_designated_destination(workflow_id: str) -> None:
+async def _notify_designated_destination(workflow_id: str, subject: dict[str, Any]) -> None:
     # Reads the ask off the run row the pending_input step just wrote (the
     # signal payload carries no ask — producer-side placement is the point);
     # the settings pointer is the operator's off-switch.
@@ -247,7 +255,7 @@ async def _create() -> str | None:
             destination_id = UserSettings.get().gate_park_destination_id
             if not destination_id:
                 return None
-            return Run.get(workflow_id).create_park_notification(destination_id)
+            return Run.get(workflow_id).create_park_notification(destination_id, subject)
 
     notification_id = await DBOS.run_step_async(
         StepOptions(name="notifications.gate_park", **_IO_RETRIES), _create
@@ -283,6 +291,8 @@ async def _emit_run_event(
     workflow_id: str,
     event: RunState,
     *,
+    subject: dict[str, Any] | None,
+    extension: str | None,
     facts: dict[str, Any] | None = None,
     result: Any = None,
 ) -> None:
@@ -290,7 +300,8 @@ async def _emit_run_event(
     # memoized step; the signal publishes in a second, so a raising subscriber
     # can't roll back the record. Publish is at-least-once and can land before
     # DBOS commits the terminal status — subscribers stay idempotent and read
-    # the payload, never derived Run.state.
+    # the payload, never derived Run.state. subject/extension come from the
+    # workflow's own arguments, so a replay stamps the same routing every time.
     async def _transition() -> dict[str, Any] | None:
         async with step_session() as session:
             run = Run.get(workflow_id)
@@ -298,13 +309,13 @@ async def _transition() -> dict[str, Any] | None:
                 for field, value in facts.items():
                     setattr(run, field, value)
                 session.flush()
-            if not run.subject:
+            if not subject:
                 # Subjectless framework crons are plumbing: no feed entry.
                 return None
             return {
                 "kind": run.kind,
-                "subject": run.subject,
-                "payload": _log_run_event(run, event, result),
+                "subject": subject,
+                "payload": _log_run_event(run, event, subject, extension, result),
             }
 
     transition = await DBOS.run_step_async(
@@ -327,7 +338,13 @@ async def _propagate() -> None:
     )
 
 
-def _log_run_event(run: Run, event: RunState, result: Any = None) -> dict[str, Any]:
+def _log_run_event(
+    run: Run,
+    event: RunState,
+    subject: dict[str, Any],
+    extension: str | None,
+    result: Any = None,
+) -> dict[str, Any]:
     # One event per transition — the feed's run-level granularity, read off the
     # just-written row so gate and failure ride the transition that set them.
     # The result rides the finished event so reactions read the outcome off the
@@ -343,9 +360,9 @@ def _log_run_event(run: Run, event: RunState, result: Any = None) -> dict[str, A
         payload["result"] = result
     Event.emit(
         type=f"run.{event.value}",
-        subject=run.subject,
+        subject=subject,
         payload=payload,
-        extension=run.extension,
+        extension=extension,
     )
     return payload
 
@@ -358,7 +375,6 @@ def _log_run_event(run: Run, event: RunState, result: Any = None) -> dict[str, A
 async def _execute_run(
     workflow_id: str,
     kind: str,
-    input: dict[str, Any],
     subject: dict[str, Any] | None,
     extension: str | None,
     body: Callable,
@@ -368,15 +384,8 @@ async def _execute_run(
     # Every failure re-raises so DBOS records the terminal ERROR derived state
     # reads; an operator cancel already carries its own reason and terminal
     # status, so it passes through untouched.
-    Run.create_row(
-        _step_engine(),
-        workflow_id=workflow_id,
-        kind=kind,
-        input=input,
-        subject=subject,
-        extension=extension,
-    )
-    await _emit_run_event(workflow_id, RunState.RUNNING)
+    Run.create_row(_step_engine(), workflow_id=workflow_id, kind=kind)
+    await _emit_run_event(workflow_id, RunState.RUNNING, subject=subject, extension=extension)
     try:
         result = await body()
     except (DBOSAwaitedWorkflowCancelledError, DBOSWorkflowCancelledError):
@@ -385,6 +394,8 @@ async def _execute_run(
         await _emit_run_event(
             workflow_id,
             RunState.FAILED,
+            subject=subject,
+            extension=extension,
             facts={
                 **_GATE_CLEARED,
                 "failure": str(exc),
@@ -392,7 +403,9 @@ async def _execute_run(
             },
         )
         raise
-    await _emit_run_event(workflow_id, RunState.FINISHED, result=result)
+    await _emit_run_event(
+        workflow_id, RunState.FINISHED, subject=subject, extension=extension, result=result
+    )
     return result
 
 
@@ -439,9 +452,13 @@ def __init_subclass__(cls, **kwargs: Any) -> None:
 
     def __init__(self) -> None:
         self._workflow_id: str = ""
-        # What the run is about ({"type", "id"}), set from config before run();
-        # None for a subjectless framework run. Dispatch-only — read, don't write.
+        # What the run is about ({"type", "id"}), set from the dispatch arguments
+        # before run(); None for a subjectless framework run. Dispatch-only —
+        # read, don't write.
         self.subject: dict[str, Any] | None = None
+        # The extension that launched the run, stamped onto its events beside
+        # the subject. Dispatch-only — read, don't write.
+        self.extension: str | None = None
         # run()'s validated input bundle (the model synthesized from its signature),
         # set before run() — for templates and derived properties. None = no input.
         self.input: BaseModel | None = None
@@ -620,11 +637,23 @@ async def start(
         # the live run's id. Subjectless runs are unbounded.
         enqueue_options = (
             SetEnqueueOptions(deduplication_id=f"{cls.kind}:{subject['type']}:{subject['id']}")
-            if subject is not None
+            if subject
             else nullcontext()
         )
+        # The workflow's routing metadata, stamped as DBOS custom attributes so
+        # "runs for this subject" is answered by workflow_status itself. The
+        # subject id is stamped as a string — the one shape every reader compares.
+        attributes: dict[str, str] = {}
+        if subject:
+            attributes = {"subject_type": subject["type"], "subject_id": str(subject["id"])}
+        if extension:
+            attributes["extension"] = extension
         try:
-            with SetWorkflowID(workflow_id), enqueue_options:
+            with (
+                SetWorkflowID(workflow_id),
+                SetWorkflowAttributes(attributes or None),
+                enqueue_options,
+            ):
                 await run_queue.enqueue_async(cls._entry, wire, subject, extension)
         except DBOSQueueDeduplicatedError as duplicate:
             holder = _get_dbos_instance()._sys_db.get_deduplicated_workflow(
@@ -637,14 +666,7 @@ async def start(
             return await cls.start(subject=subject, extension=extension, **input)
         # The body also creates its row (idempotently) — this one just makes it
         # visible before an executor picks the workflow up.
-        Run.create_row(
-            _step_engine(),
-            workflow_id=workflow_id,
-            kind=cls.kind,
-            input=wire,
-            subject=subject,
-            extension=extension,
-        )
+        Run.create_row(_step_engine(), workflow_id=workflow_id, kind=cls.kind)
         return workflow_id
 
 
@@ -683,6 +705,7 @@ async def _run_instance(
     instance = cls()
     instance._workflow_id = DBOS.workflow_id  # type: ignore[assignment]
     instance.subject = subject
+    instance.extension = extension
     # The body's input re-validates from its wire dict; a cron fires with no
     # input, so a scheduled workflow must default every parameter. The validated
     # bundle also lands on the instance for templates / derived properties.
@@ -696,7 +719,6 @@ async def _run_instance(
         return await _execute_run(
             instance._workflow_id,
             cls.kind,
-            input or {},
             subject,
             extension,
             lambda: getattr(instance, cls._body_method)(**run_kwargs),
diff --git a/backend/migrations/versions/d7e8f9a0b1c2_subject_keying_via_workflow_attributes.py b/backend/migrations/versions/d7e8f9a0b1c2_subject_keying_via_workflow_attributes.py
new file mode 100644
index 00000000..1060b2f9
--- /dev/null
+++ b/backend/migrations/versions/d7e8f9a0b1c2_subject_keying_via_workflow_attributes.py
@@ -0,0 +1,44 @@
+"""subject keying via workflow attributes
+
+Revision ID: d7e8f9a0b1c2
+Revises: c9d0e1f2a3b4
+Create Date: 2026-07-13 09:10:00.000000
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = "d7e8f9a0b1c2"
+down_revision: str | Sequence[str] | None = "c9d0e1f2a3b4"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    # Subject and extension live on the DBOS workflow as custom attributes,
+    # stamped at start(); workflow inputs are DBOS's own record.
+    op.drop_column("durable_runs", "input")
+    op.drop_column("durable_runs", "subject")
+    op.drop_column("durable_runs", "extension")
+
+
+def downgrade() -> None:
+    op.add_column("durable_runs", sa.Column("extension", sa.String(), nullable=True))
+    op.add_column(
+        "durable_runs",
+        sa.Column("subject", postgresql.JSONB(astext_type=sa.Text()), nullable=True),
+    )
+    op.add_column(
+        "durable_runs",
+        sa.Column(
+            "input",
+            postgresql.JSONB(astext_type=sa.Text()),
+            nullable=False,
+            server_default=sa.text("'{}'::jsonb"),
+        ),
+    )
diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py
index bd510398..9120ee7a 100644
--- a/backend/tests/conftest.py
+++ b/backend/tests/conftest.py
@@ -344,9 +344,12 @@ def bind_ambient_session(session) -> None:
     db_session.registry.set(session)
 
 
-def seed_dbos_status(session, workflow_id: str, state: str) -> None:
-    """Write the ``dbos.workflow_status`` row a Run's derived ``state`` reads —
-    the paired half of every persisted run seed (there is no state column)."""
+def seed_dbos_status(
+    session, workflow_id: str, state: str, *, subject=None, extension=None
+) -> None:
+    """Write the ``dbos.workflow_status`` row a Run's derived ``state`` reads,
+    carrying the subject/extension attributes ``start()`` stamps — the paired
+    half of every persisted run seed (there is no state column)."""
     from druks.durable.dbos_state import workflow_status
     from druks.models import Base
 
@@ -359,13 +362,19 @@ def seed_dbos_status(session, workflow_id: str, state: str) -> None:
         "cancelled": "CANCELLED",
     }[state]
     now_ms = int(Base.utc_now().timestamp() * 1000)
+    attributes = {}
+    if subject:
+        attributes = {"subject_type": subject["type"], "subject_id": str(subject["id"])}
+    if extension:
+        attributes["extension"] = extension
     # created_at / priority carry server defaults in the dbos schema; the
-    # derivation reads only status + updated_at, so seed just those.
+    # derivation and subject keying read only these.
     session.execute(
         workflow_status.insert().values(
             workflow_uuid=workflow_id,
             status=status,
             updated_at=now_ms,
+            attributes=attributes or None,
         )
     )
     session.flush()
@@ -379,7 +388,6 @@ def seed_build_run(
     input_gate: str | None = None,
     input_request: dict | None = None,
     failure: str | None = None,
-    config_extra: dict | None = None,
 ):
     """Seed a build Run row for a work item and bind it via
     ``item.build_run_id``. Returns the Run. Attach agent calls with
@@ -397,13 +405,16 @@ def seed_build_run(
         input_gate=input_gate,
         input_request=input_request,
         failure=failure,
-        input=config_extra or {},
-        subject={"type": "work_item", "id": work_item_id},
-        extension="build",
     )
     session.add(run)
     session.flush()
-    seed_dbos_status(session, run.id, state)
+    seed_dbos_status(
+        session,
+        run.id,
+        state,
+        subject={"type": "work_item", "id": work_item_id},
+        extension="build",
+    )
     item = WorkItem.get(work_item_id)
     item.build_run_id = run.id
     session.flush()
@@ -472,11 +483,11 @@ def seed_agent_run(
     return call
 
 
-def seed_run(session, run_id, *, kind="build.build_workflow", input=None):
+def seed_run(session, run_id, *, kind="build.build_workflow"):
     # A bare durable_runs row so an AgentCall / Artifact FK to it resolves.
     from druks.durable import Run
 
-    run = Run(id=run_id, kind=kind, input=input or {})
+    run = Run(id=run_id, kind=kind)
     session.add(run)
     session.flush()
     return run
diff --git a/backend/tests/test_api_work_items.py b/backend/tests/test_api_work_items.py
index 0e460287..c34c87d6 100644
--- a/backend/tests/test_api_work_items.py
+++ b/backend/tests/test_api_work_items.py
@@ -29,12 +29,15 @@ def _seed_scope_run(db_session, item, *, state="finished", status=None):
         kind="build.scope",
         input_gate=ScopeReply.topic if parked else None,
         input_request=parked.model_dump(mode="json") if parked else None,
-        input={"remote_key": item.remote_key},
-        subject={"type": "work_item", "id": item.id},
     )
     db_session.add(run)
     db_session.flush()
-    seed_dbos_status(db_session, run.id, "pending_input" if parked else state)
+    seed_dbos_status(
+        db_session,
+        run.id,
+        "pending_input" if parked else state,
+        subject={"type": "work_item", "id": item.id},
+    )
     seed_call(db_session, run, "scope", status="failed" if state == "failed" else "succeeded")
     if status is not None:
         from druks.events.models import Event
diff --git a/backend/tests/test_durable_sdk.py b/backend/tests/test_durable_sdk.py
index 921ad17b..12472470 100644
--- a/backend/tests/test_durable_sdk.py
+++ b/backend/tests/test_durable_sdk.py
@@ -8,11 +8,12 @@
 from druks.agents import Agent, AgentOutput
 from druks.database import configure_session, get_session
 from druks.durable import FatalError, Run, RunState
+from druks.durable.dbos_state import workflow_status
 from druks.durable.engine import configure_engine, init_dbos, launch, shutdown
 from druks.extensions.registry import agents, workflows
 from druks.workflows import Gate, Workflow, step
 from pydantic import BaseModel
-from sqlalchemy import create_engine
+from sqlalchemy import create_engine, select
 
 PG_BASE = os.environ.get("DRUKS_TEST_PG", "postgresql://druks:druks@localhost:5432")
 DB = "druks_durable_test"
@@ -314,6 +315,13 @@ async def test_subject_gate_parks_unchanged(rt):
     wfid = await rt.ConfirmFlow.start(subject={"type": "widget", "id": 636363})
     parked = await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.PENDING_INPUT)
     assert parked.input_gate == "confirm"
+    # start() stamped the subject as workflow attributes — the keying every
+    # runs-for-a-subject query reads; the id normalizes to a string.
+    with rt.engine.connect() as conn:
+        attributes = conn.execute(
+            select(workflow_status.c.attributes).where(workflow_status.c.workflow_uuid == wfid)
+        ).scalar_one()
+    assert attributes == {"subject_type": "widget", "subject_id": "636363"}
 
     await parked.resume(action="go")
     await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FINISHED)
@@ -555,9 +563,9 @@ async def test_subject_shape_validation(rt):
             await rt.SubjectFlow.start(subject=bad)
 
 
-async def test_config_is_validated_at_start_and_stored_flat(rt):
+async def test_input_is_validated_at_start(rt):
     # run()'s annotation is the wire contract: a bad input fails at start(), a
-    # good one lands on Run.input as the model's plain JSON — no envelope.
+    # good one reaches the body through DBOS's own checkpointed arguments.
     from druks.durable import WorkflowError
     from pydantic import ValidationError
 
@@ -567,8 +575,8 @@ async def test_config_is_validated_at_start_and_stored_flat(rt):
         await rt.SubjectFlow.start(subject=None, repo="x")  # takes no input
 
     wfid = await rt.RecordFeedback.start(subject=None, repo="owner/flat")
-    row = await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FINISHED)
-    assert row.input == {"repo": "owner/flat"}
+    await _wait_for(rt.engine, wfid, lambda r: r.state == RunState.FINISHED)
+    assert "owner/flat" in SINK
 
 
 async def test_run_signature_is_enforced(rt):
diff --git a/backend/tests/test_generic_subjects.py b/backend/tests/test_generic_subjects.py
index 98d6d724..1b366bf5 100644
--- a/backend/tests/test_generic_subjects.py
+++ b/backend/tests/test_generic_subjects.py
@@ -47,14 +47,17 @@ def _seed_run(
         kind=kind,
         input_gate=input_gate,
         input_request=input_request,
-        input={},
         failure=failure,
-        subject={"type": "thing", "id": subject_id},
-        extension="faketest",
     )
     session.add(run)
     session.flush()
-    seed_dbos_status(session, run.id, state)
+    seed_dbos_status(
+        session,
+        run.id,
+        state,
+        subject={"type": "thing", "id": subject_id},
+        extension="faketest",
+    )
     return run
 
 
diff --git a/backend/tests/test_lane_reactions.py b/backend/tests/test_lane_reactions.py
index 80ae4bb8..8ed7905c 100644
--- a/backend/tests/test_lane_reactions.py
+++ b/backend/tests/test_lane_reactions.py
@@ -87,19 +87,22 @@ async def _push(self, status):
     assert pushed == [SemanticStatus.IN_PROGRESS, SemanticStatus.IN_REVIEW]
 
 
-def _parked_scope_run(session, *, work_item_id, remote_key):
+def _parked_scope_run(session, *, work_item_id):
     run = Run(
         id=str(uuid7()),
         kind=Scope.kind,
         input_gate=ScopeReply.topic,
         input_request={"presentation": "external", "label": "Answer scope questions"},
-        input={"remote_key": remote_key},
-        subject={"type": "work_item", "id": work_item_id},
-        extension="build",
     )
     session.add(run)
     session.flush()
-    seed_dbos_status(session, run.id, "pending_input")
+    seed_dbos_status(
+        session,
+        run.id,
+        "pending_input",
+        subject={"type": "work_item", "id": work_item_id},
+        extension="build",
+    )
     return run
 
 
@@ -130,7 +133,7 @@ async def _dbos_cancel(workflow_id):
 
     monkeypatch.setattr("dbos.DBOS.cancel_workflow_async", _dbos_cancel)
     item = make_test_work_item(repo="acme/widget", title="t", source="linear", remote_key="ACME-5")
-    run = _parked_scope_run(db_session, work_item_id=item.id, remote_key="ACME-5")
+    run = _parked_scope_run(db_session, work_item_id=item.id)
 
     await publish(
         "ticket.transitioned",
diff --git a/backend/tests/test_profiling.py b/backend/tests/test_profiling.py
index 6c1b4896..0e7ff1ab 100644
--- a/backend/tests/test_profiling.py
+++ b/backend/tests/test_profiling.py
@@ -175,15 +175,10 @@ def _seed_run(self, db_session, repo, *, state, failure=None):
         from druks.durable import Run
         from uuid_utils import uuid7
 
-        run = Run(
-            id=str(uuid7()),
-            kind="build.profile",
-            failure=failure,
-            subject={"type": "project_repo", "id": repo.id},
-        )
+        run = Run(id=str(uuid7()), kind="build.profile", failure=failure)
         db_session.add(run)
         db_session.flush()
-        seed_dbos_status(db_session, run.id, state)
+        seed_dbos_status(db_session, run.id, state, subject={"type": "project_repo", "id": repo.id})
         return run
 
     def test_unprofiled_when_no_run_and_no_profile(self, db_session):
diff --git a/backend/tests/test_run_state.py b/backend/tests/test_run_state.py
index 389bd8ad..794f99b8 100644
--- a/backend/tests/test_run_state.py
+++ b/backend/tests/test_run_state.py
@@ -142,6 +142,8 @@ async def _raises(**_: object) -> None:
         await _emit_run_event(
             run.id,
             RunState.PENDING_INPUT,
+            subject={"type": "work_item", "id": item.id},
+            extension="build",
             facts={"input_gate": "review_work", "input_request": {"label": "Review"}},
         )
 
@@ -170,7 +172,13 @@ async def test_lifecycle_subscribers_get_the_payload_before_dbos_commits(db_sess
     async def _reads_both(*, run: str, **kwargs: object) -> None:
         seen.append((Run.get(run).state, kwargs))
 
-    await _emit_run_event(run.id, RunState.FINISHED, result={"status": "ok"})
+    await _emit_run_event(
+        run.id,
+        RunState.FINISHED,
+        subject={"type": "work_item", "id": item.id},
+        extension="build",
+        result={"status": "ok"},
+    )
 
     ((state_at_signal, payload),) = seen
     assert state_at_signal == RunState.RUNNING.value
@@ -189,7 +197,7 @@ async def body() -> None:
         raise DBOSWorkflowCancelledError(f"workflow {run.id} cancelled")
 
     with pytest.raises(DBOSWorkflowCancelledError):
-        await _execute_run(run.id, run.kind, {}, run.subject, "build", body)
+        await _execute_run(run.id, run.kind, {"type": "work_item", "id": item.id}, "build", body)
 
     ambient_session().expire_all()
     assert Run.get(run.id).failure is None
@@ -218,7 +226,7 @@ async def body() -> None:
         raise FatalError("closed at review")
 
     with pytest.raises(FatalError):
-        await _execute_run(run.id, run.kind, {}, run.subject, "build", body)
+        await _execute_run(run.id, run.kind, {"type": "work_item", "id": item.id}, "build", body)
 
     ambient_session().expire_all()
     row = Run.get(run.id)
@@ -239,13 +247,13 @@ async def test_gate_timeout_stamps_its_failure_code(db_session, _inline_steps):
     # unanswered gate from a crash without parsing the failure text.
     from druks.durable.exceptions import GateTimeout
 
-    _item, run = _item_and_run(db_session, "running")
+    item, run = _item_and_run(db_session, "running")
 
     async def body() -> None:
         raise GateTimeout("review_work")
 
     with pytest.raises(GateTimeout):
-        await _execute_run(run.id, run.kind, {}, run.subject, "build", body)
+        await _execute_run(run.id, run.kind, {"type": "work_item", "id": item.id}, "build", body)
 
     ambient_session().expire_all()
     assert Run.get(run.id).failure_code == "gate_timeout"
diff --git a/backend/tests/test_scope_reply_rescope.py b/backend/tests/test_scope_reply_rescope.py
index afe024de..1e460583 100644
--- a/backend/tests/test_scope_reply_rescope.py
+++ b/backend/tests/test_scope_reply_rescope.py
@@ -25,12 +25,15 @@ def _scope_run(db_session, *, work_item_id, parked=True):
         id=str(uuid7()),
         kind="build.scope",
         input_gate=ScopeReply.topic if parked else None,
-        input={"remote_key": "ACME-1"},
-        subject={"type": "work_item", "id": work_item_id},
     )
     db_session.add(run)
     db_session.flush()
-    seed_dbos_status(db_session, run.id, "pending_input" if parked else "finished")
+    seed_dbos_status(
+        db_session,
+        run.id,
+        "pending_input" if parked else "finished",
+        subject={"type": "work_item", "id": work_item_id},
+    )
     return run
 
 
diff --git a/docs/concepts.md b/docs/concepts.md
index d90279e9..5f22e6aa 100644
--- a/docs/concepts.md
+++ b/docs/concepts.md
@@ -91,9 +91,11 @@ extension still owns domain policy and side-effect idempotency.
 
 ### State has one lifecycle owner
 
-The `durable_runs` row stores Druks metadata such as input, subject, current
-gate, failure, and timestamps. Its lifecycle state is read-only and derived
-from DBOS's workflow status:
+The `durable_runs` row stores the Druks-owned facts DBOS has no slot for: the
+current gate ask, the failure text, and timestamps. The run's subject and
+extension live on the DBOS workflow itself as custom attributes, so
+"runs for this subject" is answered by `workflow_status` alone. The row's
+lifecycle state is read-only and derived from DBOS's workflow status:
 
 ```text
 scheduled -> running -> finished