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/routes.py b/backend/druks/build/routes.py
index 3aca69e8..02072c4c 100644
--- a/backend/druks/build/routes.py
+++ b/backend/druks/build/routes.py
@@ -178,9 +178,7 @@ async def add_project_repo(
         )
         .values(project_id=project.id)
     )
-    await Profile.start(
-        subject={"type": "project_repo", "id": repo.id}, extension="build", repo_id=repo.id
-    )
+    await Profile.start(subject={"type": "project_repo", "id": repo.id}, repo_id=repo.id)
     return ProjectRepoSummary.from_repo(repo)
 
 
@@ -216,9 +214,7 @@ async def profile_project_repo(project_id: int, repo_id: int) -> ProjectRepoSumm
         raise HTTPException(status.HTTP_404_NOT_FOUND, "repo not found")
     # Profile is subject-unique: start() returns the live run when one is already
     # active for this repo, so the route just dispatches and lets the lock dedup.
-    await Profile.start(
-        subject={"type": "project_repo", "id": repo_id}, extension="build", repo_id=repo_id
-    )
+    await Profile.start(subject={"type": "project_repo", "id": repo_id}, repo_id=repo_id)
     return ProjectRepoSummary.from_repo(row)
 
 
diff --git a/backend/druks/build/scoping/workflows.py b/backend/druks/build/scoping/workflows.py
index 55944dc4..26d8b600 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__)
 
@@ -50,19 +46,14 @@ async def dispatch(cls, *, ticket: Ticket) -> str | None:
             )
         return await cls.start(
             subject=WorkItem.subject_for(item.id),
-            extension="build",
             remote_key=ticket.key,
             source=ticket.provider,
         )
 
     @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/build/subscribers.py b/backend/druks/build/subscribers.py
index 59208cd6..60ba6dc5 100644
--- a/backend/druks/build/subscribers.py
+++ b/backend/druks/build/subscribers.py
@@ -47,7 +47,6 @@ async def policy_push_reprofiles_the_repo(*, repo: str, paths: list, **_: object
         return
     await Profile.start(
         subject={"type": "project_repo", "id": project_repo.id},
-        extension="build",
         repo_id=project_repo.id,
         refresh_only=True,
     )
diff --git a/backend/druks/build/workflows.py b/backend/druks/build/workflows.py
index 85bbad66..dde4fb7b 100644
--- a/backend/druks/build/workflows.py
+++ b/backend/druks/build/workflows.py
@@ -146,7 +146,6 @@ async def dispatch(
             raise ValueError(f"dispatching a build for unknown work item {work_item_id}")
         run_id = await cls.start(
             subject=WorkItem.subject_for(item.id),
-            extension="build",
             repo=item.repo,
             source=item.source,
             ticket_ref=item.remote_key,
diff --git a/backend/druks/durable/dbos_state.py b/backend/druks/durable/dbos_state.py
index ad0e2504..f91a59cd 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,39 @@
 # 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 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..88eb344b 100644
--- a/backend/druks/durable/models.py
+++ b/backend/druks/durable/models.py
@@ -11,7 +11,7 @@
 
 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 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,13 +44,6 @@ 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
@@ -68,27 +61,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 +77,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 +134,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 +347,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/durable/schemas.py b/backend/druks/durable/schemas.py
index 4873f772..3e2406e8 100644
--- a/backend/druks/durable/schemas.py
+++ b/backend/druks/durable/schemas.py
@@ -1,7 +1,6 @@
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any, Literal
-from urllib.parse import quote
 
 from pydantic import Field, SerializeAsAny
 
@@ -73,27 +72,21 @@ class ArtifactFile(BaseResponse):
     name: str
     size_bytes: int
     updated_at: datetime
-    url: str
 
 
 class ArtifactDescriptor(BaseResponse):
     # A call's renderable output (a plan's markdown), rendered by kind — distinct
-    # from the raw files. Content lives in the call dir, served by the same route.
+    # from the raw files. ``name`` is its file in the call dir, downloadable from
+    # the transcript files route like any other.
     kind: str
     title: str
-    url: str
-
-    @classmethod
-    def from_artifact(cls, artifact: Artifact, *, base: str) -> "ArtifactDescriptor":
-        return cls(
-            kind=artifact.kind, title=artifact.title, url=base + quote(artifact.path, safe="")
-        )
+    name: str
 
 
 class AgentCallFiles(BaseResponse):
-    # A call's on-disk artifacts, each with a download URL back to the transcript router.
-    # The slot the file occupies is its role (prompt / response / stdout / stderr /
-    # metadata / manifest).
+    # A call's on-disk artifacts by role (prompt / response / stdout / stderr /
+    # metadata / manifest). Each carries its file name; the client composes the
+    # download URL from the transcript route it fetched this listing from.
     prompt: ArtifactFile | None = None
     stdout: ArtifactFile | None = None
     stderr: ArtifactFile | None = None
@@ -108,7 +101,6 @@ class AgentCallFiles(BaseResponse):
     @classmethod
     def from_call(cls, call: AgentCall, artifact: Artifact | None) -> "AgentCallFiles":
         layout = call.artifact_layout
-        base = f"/api/{call.run.extension}/transcripts/{call.id}/files/"
 
         def named(path: Path) -> ArtifactFile | None:
             if not path.is_file():
@@ -118,10 +110,13 @@ def named(path: Path) -> ArtifactFile | None:
                 name=path.name,
                 size_bytes=stat.st_size,
                 updated_at=datetime.fromtimestamp(stat.st_mtime, tz=UTC),
-                url=base + quote(path.name, safe=""),
             )
 
-        descriptor = ArtifactDescriptor.from_artifact(artifact, base=base) if artifact else None
+        descriptor = None
+        if artifact:
+            descriptor = ArtifactDescriptor(
+                kind=artifact.kind, title=artifact.title, name=artifact.path
+            )
         return cls(
             prompt=named(layout.prompt),
             response=named(layout.output),
diff --git a/backend/druks/extensions/loader.py b/backend/druks/extensions/loader.py
index 328be676..123b3f1a 100644
--- a/backend/druks/extensions/loader.py
+++ b/backend/druks/extensions/loader.py
@@ -12,6 +12,43 @@
 
 _GROUP = "druks.extensions"
 
+# Which extension owns each workflow-declaring package — the loader-validated
+# installation claims, stamped once an entry point checks out, so a Workflow
+# class resolves its identity at definition time. None marks a package whose
+# workflows belong to no extension (how test modules register themselves).
+_workflow_packages: dict[str, str | None] = {}
+
+
+def register_workflow_package(package: str, extension: str | None) -> None:
+    # Conflicting or overlapping claims are a packaging mistake — two installs
+    # can't share a workflow package.
+    if package in _workflow_packages:
+        if _workflow_packages[package] != extension:
+            raise MalformedExtension(
+                f"package {package!r} already belongs to "
+                f"{_workflow_packages[package]!r} — {extension!r} can't claim it"
+            )
+        return
+    for registered, owner in _workflow_packages.items():
+        nested = registered.startswith(f"{package}.") or package.startswith(f"{registered}.")
+        if nested and owner != extension:
+            raise MalformedExtension(
+                f"package {package!r} overlaps {registered!r} (owned by {owner!r}) — "
+                "workflow ownership must be unambiguous"
+            )
+    _workflow_packages[package] = extension
+
+
+def resolve_workflow_extension(module: str) -> str | None:
+    """The extension owning ``module``'s nearest registered ancestor package.
+    Raises ``LookupError`` when no registered package contains the module."""
+    prefix = module
+    while prefix:
+        if prefix in _workflow_packages:
+            return _workflow_packages[prefix]
+        prefix = prefix.rpartition(".")[0]
+    raise LookupError(module)
+
 
 def iter_extensions() -> list[type[Extension]]:
     """Every installed extension, resolved from the ``druks.extensions`` entry points.
@@ -25,9 +62,17 @@ def iter_extensions() -> list[type[Extension]]:
         extension = entry.load()
         if not (isinstance(extension, type) and issubclass(extension, Extension)):
             raise TypeError(f"extension entry point {entry.name!r} is not an Extension")
+        if extension.name != entry.name:
+            raise MalformedExtension(
+                f"extension {entry.name!r} entry point resolves to an Extension named "
+                f"{extension.name!r} — the entry-point name must match Extension.name"
+            )
         if extension.name in seen:
             raise ValueError(f"duplicate extension name {extension.name!r}")
         seen.add(extension.name)
+        # Ownership registers before load(app)/discover() imports the capability
+        # modules, whose Workflow classes resolve their extension at definition.
+        register_workflow_package(extension.package, extension.name)
         extensions.append(extension)
     return extensions
 
@@ -79,18 +124,16 @@ def _resolve(name: str) -> type[Extension]:
             f"({', '.join(e.value for e in matches)}) — uninstall all but one"
         )
     extension = _load_entry(matches[0])
-    # The entry-point key must equal the class's ``name``. That key is what scopes
-    # the /api, settings, and migration namespaces here, so this is the invariant
-    # that lets the duplicate-key check above stand in for a duplicate-name check —
-    # without importing sibling extensions and defeating the point of an app-less,
-    # single-extension load. A same-name collision hidden behind a mismatched key is
-    # that sibling's own malformed state, caught when it is loaded (or at full boot,
-    # where iter_extensions() imports everything).
+    # The entry-point key must equal the class's ``name`` — the key scopes the
+    # /api, settings, and migration namespaces, which is what lets the
+    # duplicate-key check above stand in for a duplicate-name check without
+    # importing sibling extensions.
     if extension.name != name:
         raise MalformedExtension(
             f"extension {name!r} entry point resolves to an Extension named "
             f"{extension.name!r} — the entry-point name must match Extension.name"
         )
+    register_workflow_package(extension.package, extension.name)
     return extension
 
 
diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py
index b48df923..0214ed05 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,
@@ -26,6 +26,7 @@
 from druks.durable.models import AgentCall, Run
 from druks.durable.schemas import AgentCallResponse, SubjectActivity, SubjectSummary
 from druks.events.models import Event
+from druks.extensions.loader import resolve_workflow_extension
 from druks.extensions.registry import workflows
 from druks.extensions.settings import (
     coerce_setting_value,
@@ -157,12 +158,6 @@ def _kind_from_class_name(name: str) -> str:
     return _CAMEL_BOUNDARY.sub("_", name).lower()
 
 
-def _namespace_from_module(module: str) -> str | None:
-    # druks.<extension>.… → <extension>; None for out-of-tree (test) modules.
-    parts = module.split(".")
-    return parts[1] if len(parts) > 1 and parts[0] == "druks" else None
-
-
 class Gate(BaseModel):
     """A typed human-in-the-loop gate. Subclass per park point; the class name is
     the durable recv topic, the fields are the reply's schema. `wait()` parks the
@@ -222,6 +217,7 @@ async def _park(
     await _emit_run_event(
         workflow.workflow_id,
         RunState.PENDING_INPUT,
+        subject=workflow.subject,
         facts={
             "input_gate": topic,
             "input_request": input_request,
@@ -230,15 +226,20 @@ 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,
+        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 +248,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 +284,7 @@ async def _emit_run_event(
     workflow_id: str,
     event: RunState,
     *,
+    subject: dict[str, Any] | None,
     facts: dict[str, Any] | None = None,
     result: Any = None,
 ) -> None:
@@ -290,7 +292,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 comes 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 +301,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, result),
             }
 
     transition = await DBOS.run_step_async(
@@ -327,7 +330,12 @@ 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],
+    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 +351,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=workflows.get(run.kind).extension,
     )
     return payload
 
@@ -358,9 +366,7 @@ 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,
 ) -> Any:
     # Ensure the row (idempotent, so a scheduled run with no start() makes it
@@ -368,15 +374,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)
     try:
         result = await body()
     except (DBOSAwaitedWorkflowCancelledError, DBOSWorkflowCancelledError):
@@ -385,6 +384,7 @@ async def _execute_run(
         await _emit_run_event(
             workflow_id,
             RunState.FAILED,
+            subject=subject,
             facts={
                 **_GATE_CLEARED,
                 "failure": str(exc),
@@ -392,12 +392,16 @@ 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, result=result)
     return result
 
 
 class Workflow:
     kind: ClassVar[str] = ""
+    # The extension that declares this workflow — class identity, resolved from
+    # the loader's package registrations at definition time and namespacing
+    # ``kind``. Never supplied or stored per run.
+    extension: ClassVar[str | None] = 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
@@ -425,10 +429,22 @@ class Settings(BaseModel):
 
     def __init_subclass__(cls, **kwargs: Any) -> None:
         super().__init_subclass__(**kwargs)
-        if not cls.__dict__.get("kind"):
-            namespace = _namespace_from_module(cls.__module__)
-            local_kind = _kind_from_class_name(cls.__name__)
-            cls.kind = f"{namespace}.{local_kind}" if namespace else local_kind
+        try:
+            cls.extension = resolve_workflow_extension(cls.__module__)
+        except LookupError:
+            raise WorkflowError(
+                f"{cls.__module__} declares workflow {cls.__name__} outside every "
+                "registered extension package — workflow modules load through "
+                "druks.extensions.loader; a module the loader doesn't own must "
+                "register_workflow_package() before importing"
+            ) from None
+        local_kind = cls.__dict__.get("kind") or _kind_from_class_name(cls.__name__)
+        if "." in local_kind:
+            raise WorkflowError(
+                f"{cls.__name__}.kind {local_kind!r} must be a local name — "
+                "the declaring extension supplies the namespace"
+            )
+        cls.kind = f"{cls.extension}.{local_kind}" if cls.extension else local_kind
         validate_settings_declaration(cls.Settings)
         cls._body_method = _resolve_body_method(cls)
         # Before _wrap_steps: run()'s wrapper signature is (*args, **kwargs).
@@ -439,8 +455,9 @@ 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
         # run()'s validated input bundle (the model synthesized from its signature),
         # set before run() — for templates and derived properties. None = no input.
@@ -586,13 +603,7 @@ def override_setting(cls, field: str, value: Any) -> None:
         SettingsOverride.set_workflow_setting(cls.kind, field, value)
 
     @classmethod
-    async def start(
-        cls,
-        *,
-        subject: dict[str, Any] | None,
-        extension: str | None = None,
-        **input: Any,
-    ) -> str:
+    async def start(cls, *, subject: dict[str, Any] | None, **input: Any) -> str:
         # Mint the id, write the projection row, enqueue the body. Returns the
         # workflow id; an extension that wants one-active-run-per-subject enforces
         # that on its own side before calling this. Enqueuing (not start_workflow)
@@ -620,12 +631,22 @@ 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 = None
+        if subject:
+            attributes = {"subject_type": subject["type"], "subject_id": str(subject["id"])}
         try:
-            with SetWorkflowID(workflow_id), enqueue_options:
-                await run_queue.enqueue_async(cls._entry, wire, subject, extension)
+            with (
+                SetWorkflowID(workflow_id),
+                SetWorkflowAttributes(attributes),
+                enqueue_options,
+            ):
+                await run_queue.enqueue_async(cls._entry, subject, wire)
         except DBOSQueueDeduplicatedError as duplicate:
             holder = _get_dbos_instance()._sys_db.get_deduplicated_workflow(
                 run_queue.name, duplicate.deduplication_id
@@ -634,17 +655,10 @@ async def start(
                 return holder
             # The holder reached terminal between the rejection and the lookup —
             # the slot is free now, so this start goes through.
-            return await cls.start(subject=subject, extension=extension, **input)
+            return await cls.start(subject=subject, **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
 
 
@@ -676,9 +690,8 @@ async def _do() -> Any:
 
 async def _run_instance(
     cls: type[Workflow],
-    input: dict[str, Any] | None = None,
     subject: dict[str, Any] | None = None,
-    extension: str | None = None,
+    input: dict[str, Any] | None = None,
 ) -> Any:
     instance = cls()
     instance._workflow_id = DBOS.workflow_id  # type: ignore[assignment]
@@ -696,9 +709,7 @@ 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),
         )
     finally:
@@ -707,13 +718,12 @@ async def _run_instance(
 
 
 def _register_entry(cls: type[Workflow]) -> None:
+    # The closure binds cls outside the durable arguments: the DBOS workflow
+    # NAME (the kind) is what says which class this is, so recovery rebinds by
+    # name and no class object ever rides a checkpoint.
     @DBOS.workflow(name=cls.kind)
-    async def _entry(
-        input: dict[str, Any],
-        subject: dict[str, Any] | None = None,
-        extension: str | None = None,
-    ) -> None:
-        await _run_instance(cls, input, subject, extension)
+    async def _entry(subject: dict[str, Any] | None, input: dict[str, Any]) -> None:
+        await _run_instance(cls, subject, input)
 
     cls._entry = staticmethod(_entry)  # type: ignore[assignment]
 
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..cc77e56d 100644
--- a/backend/tests/conftest.py
+++ b/backend/tests/conftest.py
@@ -13,6 +13,7 @@
     create_engine_from_url,
     init_db,
 )
+from druks.extensions.loader import iter_extensions, register_workflow_package
 from druks.settings import Settings
 from sqlalchemy.orm import Session
 
@@ -33,6 +34,14 @@
 # integration concerns), so its waits don't reach for a client that isn't there.
 _gate._redis = lambda: None
 
+# A Workflow class resolves its declaring extension at definition time, from
+# packages the loader registers before importing. Tests import workflow modules
+# directly and some declare their own workflows, so both register here — before
+# collection imports any test module.
+iter_extensions()
+for test_module in ("test_durable_sdk", "test_notifications_durable"):
+    register_workflow_package(test_module, None)
+
 
 class FakeRedis:
     # The subset the run lock and the MCP OAuth cache use; one instance per
@@ -344,9 +353,10 @@ 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) -> None:
+    """Write the ``dbos.workflow_status`` row a Run's derived ``state`` reads,
+    carrying the subject 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 +369,17 @@ def seed_dbos_status(session, workflow_id: str, state: str) -> None:
         "cancelled": "CANCELLED",
     }[state]
     now_ms = int(Base.utc_now().timestamp() * 1000)
+    attributes = None
+    if subject:
+        attributes = {"subject_type": subject["type"], "subject_id": str(subject["id"])}
     # 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,
         )
     )
     session.flush()
@@ -379,7 +393,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 +410,10 @@ 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})
     item = WorkItem.get(work_item_id)
     item.build_run_id = run.id
     session.flush()
@@ -472,11 +482,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/druks-field_notes/druks_field_notes/workflows.py b/backend/tests/druks-field_notes/druks_field_notes/workflows.py
index cb9b5407..c599cc0a 100644
--- a/backend/tests/druks-field_notes/druks_field_notes/workflows.py
+++ b/backend/tests/druks-field_notes/druks_field_notes/workflows.py
@@ -21,6 +21,5 @@ async def dispatch(cls, *, note_id: int) -> str:
         # Launch policy for a note: one run per note, keyed by its subject.
         return await cls.start(
             subject={"type": FieldNotes.subject_type, "id": note_id},
-            extension=FieldNotes.name,
             note_id=note_id,
         )
diff --git a/backend/tests/test_api_runs.py b/backend/tests/test_api_runs.py
index 1be3c9de..5669bb8e 100644
--- a/backend/tests/test_api_runs.py
+++ b/backend/tests/test_api_runs.py
@@ -49,10 +49,11 @@ def test_list_files_inventories_call_artifacts(
 
     assert response.status_code == 200
     files = response.json()
-    # The slot the file occupies is its role; each carries a name + download url.
+    # The slot the file occupies is its role; each carries the file name the
+    # client composes into the download URL.
     assert files["stdout"]["name"] == "stdout.jsonl"
     assert files["stderr"]["name"] == "stderr.log"
-    assert files["response"]["url"].endswith("/files/output.json")
+    assert files["response"]["name"] == "output.json"
     assert files["metadata"] is not None
 
 
@@ -148,9 +149,11 @@ def test_get_file_serves_inventory_paths(
     run_id = _seed_run(tmp_path=tmp_path)
 
     files = client.get(f"/api/build/transcripts/{run_id}/files").json()
-    response_url = files["response"]["url"]
+    # Compose the download URL the way the client does: the listing's own route
+    # plus the file's name — the wire carries names only.
+    name = files["response"]["name"]
 
-    response = client.get(response_url)
+    response = client.get(f"/api/build/transcripts/{run_id}/files/{name}")
     assert response.status_code == 200
     assert response.json() == {"ok": True}
 
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_extension_appless_load.py b/backend/tests/test_extension_appless_load.py
index 8a651688..ce39e44b 100644
--- a/backend/tests/test_extension_appless_load.py
+++ b/backend/tests/test_extension_appless_load.py
@@ -99,6 +99,7 @@ def external_extension(tmp_path_factory):
     entry point, and restore every global its load mutates (registries, table
     metadata, signal receivers) so the suite stays clean."""
     from blinker import signal
+    from druks.extensions import loader as extensions_loader
     from druks.extensions.registry import agents, webhooks, workflows
     from druks.models import Base
 
@@ -108,6 +109,7 @@ def external_extension(tmp_path_factory):
 
     tables = set(Base.metadata.tables)
     registries = {r: dict(r._items) for r in (agents, webhooks, workflows)}
+    packages = dict(extensions_loader._workflow_packages)
     finished = signal("run.finished")
     receivers = dict(finished.receivers)
     try:
@@ -118,6 +120,8 @@ def external_extension(tmp_path_factory):
             Base.metadata.remove(Base.metadata.tables[name])
         for registry, snapshot in registries.items():
             registry._items = snapshot
+        extensions_loader._workflow_packages.clear()
+        extensions_loader._workflow_packages.update(packages)
         finished.receivers = receivers
         for name in [m for m in sys.modules if m == _PACKAGE or m.startswith(f"{_PACKAGE}.")]:
             del sys.modules[name]
@@ -252,6 +256,18 @@ def test_entry_point_key_mismatch_raises_malformed_extension(installed, monkeypa
         load_extension("not_probe")
 
 
+def test_boot_rejects_an_entry_point_key_mismatch(installed, monkeypatch):
+    """Full boot applies the same key/name validation as the single load — an
+    aliased entry the app-less path rejects must not slip through iter_extensions()."""
+    aliased = EntryPoint(
+        name="not_probe", value=f"{_PACKAGE}.extension:Probe", group="druks.extensions"
+    )
+    monkeypatch.setattr(loader, "entry_points", lambda *, group: [aliased])
+
+    with pytest.raises(MalformedExtension, match="must match Extension.name"):
+        loader.iter_extensions()
+
+
 def test_import_error_in_entry_module_raises_extension_import_error(tmp_path, monkeypatch):
     """The extension's entry module raising on import (e.g. a missing dependency it
     imports) surfaces as ExtensionImportError — the extension's code failed, distinct
diff --git a/backend/tests/test_extension_doctor_checks.py b/backend/tests/test_extension_doctor_checks.py
index b036ed4a..49daceeb 100644
--- a/backend/tests/test_extension_doctor_checks.py
+++ b/backend/tests/test_extension_doctor_checks.py
@@ -31,6 +31,7 @@ def external_package():
     load mutates so the rest of the suite sees the in-tree extensions untouched.
     Mirrors the proof-extension suite's fixture."""
     from blinker import signal
+    from druks.extensions import loader as extensions_loader
     from druks.extensions.registry import agents, webhooks, workflows
     from druks.models import Base
 
@@ -38,6 +39,7 @@ def external_package():
 
     tables = set(Base.metadata.tables)
     registries = {registry: dict(registry._items) for registry in (agents, webhooks, workflows)}
+    packages = dict(extensions_loader._workflow_packages)
     finished = signal("run.finished")
     receivers = dict(finished.receivers)
     try:
@@ -48,6 +50,8 @@ def external_package():
             Base.metadata.remove(Base.metadata.tables[name])
         for registry, snapshot in registries.items():
             registry._items = snapshot
+        extensions_loader._workflow_packages.clear()
+        extensions_loader._workflow_packages.update(packages)
         finished.receivers = receivers
         for name in [m for m in sys.modules if m == _PACKAGE or m.startswith(f"{_PACKAGE}.")]:
             del sys.modules[name]
diff --git a/backend/tests/test_extensions.py b/backend/tests/test_extensions.py
index 796fb440..7f401dd1 100644
--- a/backend/tests/test_extensions.py
+++ b/backend/tests/test_extensions.py
@@ -40,7 +40,7 @@ class DupB(Extension):
     monkeypatch.setattr(
         loader,
         "entry_points",
-        lambda *, group: [_fake_entry("a", DupA), _fake_entry("b", DupB)],
+        lambda *, group: [_fake_entry("dup", DupA), _fake_entry("dup", DupB)],
     )
     with pytest.raises(ValueError, match="duplicate extension name"):
         iter_extensions()
diff --git a/backend/tests/test_generic_subjects.py b/backend/tests/test_generic_subjects.py
index 98d6d724..b490d785 100644
--- a/backend/tests/test_generic_subjects.py
+++ b/backend/tests/test_generic_subjects.py
@@ -47,14 +47,11 @@ 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})
     return run
 
 
diff --git a/backend/tests/test_lane_reactions.py b/backend/tests/test_lane_reactions.py
index 80ae4bb8..decddc1b 100644
--- a/backend/tests/test_lane_reactions.py
+++ b/backend/tests/test_lane_reactions.py
@@ -87,19 +87,18 @@ 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}
+    )
     return run
 
 
@@ -130,7 +129,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_manifest.py b/backend/tests/test_manifest.py
index c54caffe..12cf6637 100644
--- a/backend/tests/test_manifest.py
+++ b/backend/tests/test_manifest.py
@@ -202,7 +202,7 @@ def test_persist_writes_manifest_into_the_call_dir(tmp_path, db_session):
 
 def test_manifest_surfaces_in_agent_call_files(tmp_path, db_session):
     """A written manifest.json is inventoried on the call's transcript files, in
-    the manifest slot with a download URL."""
+    the manifest slot under its downloadable file name."""
     call = seed_agent_run(agent="implement")
     manifest = _build()
     with mock.patch("druks.durable.models.load_settings", return_value=make_settings(tmp_path)):
@@ -211,4 +211,4 @@ def test_manifest_surfaces_in_agent_call_files(tmp_path, db_session):
 
     assert files.manifest
     assert files.manifest.name == "manifest.json"
-    assert files.manifest.url.endswith("/manifest.json")
+    assert files.manifest.size_bytes > 0
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_project_repo_routes.py b/backend/tests/test_project_repo_routes.py
index 263f31cc..4f97f617 100644
--- a/backend/tests/test_project_repo_routes.py
+++ b/backend/tests/test_project_repo_routes.py
@@ -21,8 +21,8 @@ def _stub_profile_start(monkeypatch):
 
     calls: list[dict] = []
 
-    async def _start(cls, *, subject, extension=None, **input):
-        calls.append({"subject": subject, "extension": extension, **input})
+    async def _start(cls, *, subject, **input):
+        calls.append({"subject": subject, **input})
         return "fake-run-id"
 
     monkeypatch.setattr(Profile, "start", classmethod(_start))
@@ -41,7 +41,6 @@ def test_adding_a_repo_dispatches_a_profile_run(client: TestClient, monkeypatch)
     assert calls == [
         {
             "subject": {"type": "project_repo", "id": repo["id"]},
-            "extension": "build",
             "repo_id": repo["id"],
         }
     ]
@@ -63,7 +62,6 @@ def test_profile_endpoint_dispatches(client: TestClient, monkeypatch):
     assert calls == [
         {
             "subject": {"type": "project_repo", "id": repo.id},
-            "extension": "build",
             "repo_id": repo.id,
         }
     ]
diff --git a/backend/tests/test_proof_extension.py b/backend/tests/test_proof_extension.py
index 82764d59..1d857671 100644
--- a/backend/tests/test_proof_extension.py
+++ b/backend/tests/test_proof_extension.py
@@ -49,6 +49,7 @@ def external_package():
     ``Note`` class is declared once — a re-declaration into the shared metadata
     collides."""
     from blinker import signal
+    from druks.extensions import loader as extensions_loader
     from druks.extensions.registry import agents, webhooks, workflows
     from druks.models import Base
 
@@ -56,6 +57,7 @@ def external_package():
 
     tables = set(Base.metadata.tables)
     registries = {registry: dict(registry._items) for registry in (agents, webhooks, workflows)}
+    packages = dict(extensions_loader._workflow_packages)
     finished = signal("run.finished")
     receivers = dict(finished.receivers)
     try:
@@ -66,6 +68,8 @@ def external_package():
             Base.metadata.remove(Base.metadata.tables[name])
         for registry, snapshot in registries.items():
             registry._items = snapshot
+        extensions_loader._workflow_packages.clear()
+        extensions_loader._workflow_packages.update(packages)
         finished.receivers = receivers
         for name in [m for m in sys.modules if m == _PACKAGE or m.startswith(f"{_PACKAGE}.")]:
             del sys.modules[name]
@@ -295,14 +299,18 @@ def test_feed_formats_the_extensions_event(installed):
 
 async def test_dispatch_starts_a_run_keyed_to_the_note(installed, monkeypatch):
     """The workflow's launch policy starts one run per note — the subject keys off
-    the note id, and the note id also rides the run as its typed input."""
+    the note id, and the note id also rides the run as its typed input. The
+    workflow's identity comes from its declaring extension, never the caller."""
     load_extension("field_notes")
     from druks_field_notes.workflows import Summarize
 
+    assert Summarize.extension == "field_notes"
+    assert Summarize.kind == "field_notes.summarize"
+
     started: dict[str, object] = {}
 
-    async def _capture(*, subject, extension=None, **input):
-        started.update(subject=subject, extension=extension, input=input)
+    async def _capture(*, subject, **input):
+        started.update(subject=subject, input=input)
         return "run-1"
 
     monkeypatch.setattr(Summarize, "start", _capture)
@@ -311,7 +319,6 @@ async def _capture(*, subject, extension=None, **input):
 
     assert run_id == "run-1"
     assert started["subject"] == {"type": "note", "id": 42}
-    assert started["extension"] == "field_notes"
     assert started["input"] == {"note_id": 42}
 
 
diff --git a/backend/tests/test_run_state.py b/backend/tests/test_run_state.py
index 389bd8ad..c62d26fe 100644
--- a/backend/tests/test_run_state.py
+++ b/backend/tests/test_run_state.py
@@ -1,6 +1,7 @@
 from datetime import UTC, datetime, timedelta
 from unittest import mock
 
+import druks.build.workflows  # noqa: F401  # registers build.build_workflow, the seeded kind
 import pytest
 from conftest import make_test_work_item, seed_build_run, seed_run
 from dbos._error import DBOSWorkflowCancelledError
@@ -142,6 +143,7 @@ async def _raises(**_: object) -> None:
         await _emit_run_event(
             run.id,
             RunState.PENDING_INPUT,
+            subject={"type": "work_item", "id": item.id},
             facts={"input_gate": "review_work", "input_request": {"label": "Review"}},
         )
 
@@ -170,7 +172,12 @@ 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},
+        result={"status": "ok"},
+    )
 
     ((state_at_signal, payload),) = seen
     assert state_at_signal == RunState.RUNNING.value
@@ -189,7 +196,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}, body)
 
     ambient_session().expire_all()
     assert Run.get(run.id).failure is None
@@ -218,7 +225,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}, body)
 
     ambient_session().expire_all()
     row = Run.get(run.id)
@@ -239,13 +246,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}, body)
 
     ambient_session().expire_all()
     assert Run.get(run.id).failure_code == "gate_timeout"
diff --git a/backend/tests/test_scaffolding.py b/backend/tests/test_scaffolding.py
index edef2a69..4097807e 100644
--- a/backend/tests/test_scaffolding.py
+++ b/backend/tests/test_scaffolding.py
@@ -45,6 +45,10 @@ def test_create_extension_scaffolds_a_loadable_package(tmp_path):
         for role in ("models", "schemas", "contracts", "workflows", "routes", "subscribers"):
             importlib.import_module(f"druks_night_watch.{role}")
 
+        # The workflow guidance must not teach a per-run extension= argument —
+        # a workflow's identity comes from its declaring extension.
+        assert "extension=" not in (target / "druks_night_watch" / "workflows.py").read_text()
+
         app = FastAPI()
         extension.load(app)
         client = TestClient(app)
diff --git a/backend/tests/test_scope_durable.py b/backend/tests/test_scope_durable.py
index 8680e5e5..0db3e006 100644
--- a/backend/tests/test_scope_durable.py
+++ b/backend/tests/test_scope_durable.py
@@ -124,7 +124,6 @@ async def _run_agent(self, **kwargs):
 
     wfid = await rt.flow.start(
         subject=WorkItem.subject_for(item_id),
-        extension="build",
         source="linear",
         remote_key="ACME-1",
     )
diff --git a/backend/tests/test_scope_reply_rescope.py b/backend/tests/test_scope_reply_rescope.py
index afe024de..2ac68ba6 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
 
 
@@ -98,7 +101,6 @@ async def test_transition_scopes_an_unlabeled_ticket(db_session, _stub_enqueue):
     # Identity only crosses the wire — the agent fetches the rest itself.
     assert _stub_enqueue[0] == {
         "subject": _stub_enqueue[0]["subject"],
-        "extension": "build",
         "remote_key": "ACME-9",
         "source": "linear",
     }
diff --git a/backend/tests/test_webhooks_push.py b/backend/tests/test_webhooks_push.py
index d1f2772f..0903f632 100644
--- a/backend/tests/test_webhooks_push.py
+++ b/backend/tests/test_webhooks_push.py
@@ -26,7 +26,7 @@ def _stub_profile_start(monkeypatch):
 
     calls: list[dict] = []
 
-    async def _start(cls, *, subject, extension=None, **input):
+    async def _start(cls, *, subject, **input):
         calls.append({"subject": subject, **input})
         return "fake-run-id"
 
diff --git a/backend/tests/test_workflow_identity.py b/backend/tests/test_workflow_identity.py
new file mode 100644
index 00000000..4d7c5dd6
--- /dev/null
+++ b/backend/tests/test_workflow_identity.py
@@ -0,0 +1,146 @@
+import pytest
+from druks.build.scoping.workflows import Scope
+from druks.build.workflows import BuildWorkflow, Profile
+from druks.core.workflows import RefreshTokens
+from druks.durable.enums import RunState
+from druks.durable.exceptions import WorkflowError
+from druks.durable.models import Run
+from druks.durable.schemas import get_display_label
+from druks.events.models import Event
+from druks.extensions import loader as extensions_loader
+from druks.extensions.exceptions import MalformedExtension
+from druks.extensions.loader import register_workflow_package, resolve_workflow_extension
+from druks.extensions.registry import workflows
+from druks.usage.workflows import PollUsage
+from druks.workflows import Workflow, _log_run_event, step
+
+
+@pytest.fixture(autouse=True)
+def _isolated_registrations():
+    # Every test here mutates the package-owner map and the workflows registry;
+    # restore both so the suite keeps seeing only the installed extensions.
+    packages = dict(extensions_loader._workflow_packages)
+    items = dict(workflows._items)
+    yield
+    extensions_loader._workflow_packages.clear()
+    extensions_loader._workflow_packages.update(packages)
+    workflows._items = items
+
+
+def _workflow(name: str, module: str, **attrs):
+    async def run(self) -> None: ...
+
+    return type(name, (Workflow,), {"__module__": module, "run": run, **attrs})
+
+
+def test_registration_is_idempotent_and_conflicts_loudly():
+    register_workflow_package("alpha_pkg", "alpha")
+    register_workflow_package("alpha_pkg", "alpha")
+    with pytest.raises(MalformedExtension, match="already belongs"):
+        register_workflow_package("alpha_pkg", "beta")
+
+
+def test_overlapping_ownership_across_owners_is_rejected():
+    register_workflow_package("alpha_pkg", "alpha")
+    with pytest.raises(MalformedExtension, match="overlaps"):
+        register_workflow_package("alpha_pkg.nested", "beta")
+
+
+def test_resolution_matches_package_boundaries():
+    register_workflow_package("alpha_pkg", "alpha")
+    assert resolve_workflow_extension("alpha_pkg.workflows") == "alpha"
+    with pytest.raises(LookupError):
+        resolve_workflow_extension("alpha_pkg_sibling.workflows")
+
+
+def test_unregistered_module_fails_at_class_definition():
+    # The error carries the invariant: load through the loader, or register first.
+    with pytest.raises(WorkflowError, match="druks.extensions.loader"):
+        _workflow("Orphan", "nowhere.workflows")
+
+
+def test_declaring_extension_namespaces_the_kind():
+    register_workflow_package("alpha_pkg", "alpha")
+    register_workflow_package("beta_pkg", "beta")
+
+    alpha = _workflow("Summarize", "alpha_pkg.workflows")
+    beta = _workflow("Summarize", "beta_pkg.workflows")
+
+    # Two extensions can share a local workflow name without colliding on kind.
+    assert (alpha.extension, alpha.kind) == ("alpha", "alpha.summarize")
+    assert (beta.extension, beta.kind) == ("beta", "beta.summarize")
+
+
+def test_explicit_kind_is_a_local_suffix():
+    register_workflow_package("alpha_pkg", "alpha")
+    digest = _workflow("Summarize", "alpha_pkg.workflows", kind="digest")
+    assert digest.kind == "alpha.digest"
+
+
+def test_dotted_explicit_kind_is_rejected():
+    register_workflow_package("alpha_pkg", "alpha")
+    with pytest.raises(WorkflowError, match="local name"):
+        _workflow("Summarize", "alpha_pkg.workflows", kind="alpha.digest")
+
+
+def test_none_owned_package_keeps_bare_kinds():
+    register_workflow_package("plain_pkg", None)
+    flow = _workflow("Sweep", "plain_pkg.workflows")
+    assert flow.extension is None
+    assert flow.kind == "sweep"
+
+
+def test_in_tree_identities_are_stable():
+    # These kinds are durable identities (DBOS workflow names, settings keys,
+    # dedup prefixes, step-name prefixes) — byte-for-byte pins.
+    assert BuildWorkflow.kind == "build.build_workflow"
+    assert Scope.kind == "build.scope"
+    assert Profile.kind == "build.profile"
+    assert RefreshTokens.kind == "core.refresh_tokens"
+    assert PollUsage.kind == "usage.poll_usage"
+    assert (BuildWorkflow.extension, RefreshTokens.extension, PollUsage.extension) == (
+        "build",
+        "core",
+        "usage",
+    )
+
+
+def test_steps_capture_the_namespaced_kind():
+    # _wrap_steps closes over cls.kind after the namespace lands, so durable
+    # step names carry the final kind — the replay identity.
+    register_workflow_package("alpha_pkg", "alpha")
+
+    async def run_multistep(self) -> None:
+        await self.ping()
+
+    @step
+    async def ping(self) -> None: ...
+
+    flow = type(
+        "Pinger",
+        (Workflow,),
+        {"__module__": "alpha_pkg.workflows", "run_multistep": run_multistep, "ping": ping},
+    )
+
+    captured = [cell.cell_contents for cell in flow.ping.__closure__]
+    assert "alpha.pinger" in captured
+
+
+def test_lifecycle_event_stamps_the_declaring_extension(db_session):
+    # The event's extension derives from the run's kind through the registry —
+    # never an argument, never a stored copy on the run.
+    register_workflow_package("alpha_pkg", "alpha")
+    flow = _workflow("Beacon", "alpha_pkg.workflows")
+    run = Run(id="wf-identity-1", kind=flow.kind)
+    db_session.add(run)
+    db_session.flush()
+
+    payload = _log_run_event(run, RunState.FINISHED, {"type": "note", "id": 1})
+
+    event = db_session.query(Event).filter_by(type="run.finished").one()
+    assert payload["run"] == run.id
+    assert event.extension == "alpha"
+
+
+def test_display_label_reads_the_local_kind():
+    assert get_display_label("field_notes.summarize") == "Summarize"
diff --git a/docs/concepts.md b/docs/concepts.md
index d90279e9..af33114d 100644
--- a/docs/concepts.md
+++ b/docs/concepts.md
@@ -91,9 +91,13 @@ 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 lives on
+the DBOS workflow itself as custom attributes, so "runs for this subject" is
+answered by `workflow_status` alone. The run's extension is not stored at all —
+it is workflow-class metadata, derivable from the run's `kind` through the
+extension registry. The row's lifecycle state is read-only and derived from
+DBOS's workflow status:
 
 ```text
 scheduled -> running -> finished
diff --git a/docs/writing-an-extension.md b/docs/writing-an-extension.md
index 9a9b94a9..2dd85c4f 100644
--- a/docs/writing-an-extension.md
+++ b/docs/writing-an-extension.md
@@ -126,7 +126,6 @@ Start a workflow with an explicit subject:
 ```python
 run_id = await Sweep.start(
     subject={"type": "repository", "id": repo_id},
-    extension=NightWatch.name,
     repo=full_name,
 )
 ```
diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts
index b55bd814..ff5b2408 100644
--- a/frontend/src/api/client.ts
+++ b/frontend/src/api/client.ts
@@ -114,6 +114,8 @@ export const subjectApi = {
     `/api/${extension}/transcripts/${callId}`,
   transcriptFiles: (extension: string, callId: string) =>
     getJSON<AgentCallFiles>(`/api/${extension}/transcripts/${callId}/files`),
+  transcriptFile: (extension: string, callId: string, name: string) =>
+    `/api/${extension}/transcripts/${callId}/files/${encodeURIComponent(name)}`,
 }
 
 export const api = {
diff --git a/frontend/src/api/types.ts b/frontend/src/api/types.ts
index 684ded5b..dc47ab17 100644
--- a/frontend/src/api/types.ts
+++ b/frontend/src/api/types.ts
@@ -132,11 +132,11 @@ export interface ArtifactFile {
   name: string
   sizeBytes: number
   updatedAt: string
-  url: string
 }
 
-// A call's on-disk artifacts, each with a download URL — served by the platform's
-// per-extension transcript router (/api/<ext>/transcripts/<callId>/files).
+// A call's on-disk artifacts by role. Each carries its file name; the client
+// composes the download URL from the transcript route it fetched this listing
+// from (subjectApi.transcriptFile).
 export interface AgentCallFiles {
   prompt?: ArtifactFile | null
   stdout?: ArtifactFile | null
diff --git a/frontend/src/extensions/build/AgentCallPage.tsx b/frontend/src/extensions/build/AgentCallPage.tsx
index f4446c55..423539a8 100644
--- a/frontend/src/extensions/build/AgentCallPage.tsx
+++ b/frontend/src/extensions/build/AgentCallPage.tsx
@@ -84,6 +84,7 @@ function RunView({
 
   const initialLeft: LeftTab = files.prompt ? 'prompt' : 'response'
   const [leftTab, setLeftTab] = useState<LeftTab>(initialLeft)
+  const activeFile = leftTab === 'prompt' ? files.prompt : files.response
 
   return (
     <Page scroll="internal" className="page-run">
@@ -143,13 +144,7 @@ function RunView({
               response
             </button>
           </div>
-          <FilePane
-            url={
-              leftTab === 'prompt'
-                ? (files.prompt?.url ?? null)
-                : (files.response?.url ?? null)
-            }
-          />
+          <FilePane url={activeFile ? buildApi.transcriptFile(call.id, activeFile.name) : null} />
         </section>
 
         <section className="run-col run-col-right">
diff --git a/frontend/src/extensions/build/api.ts b/frontend/src/extensions/build/api.ts
index c44c76b0..51763200 100644
--- a/frontend/src/extensions/build/api.ts
+++ b/frontend/src/extensions/build/api.ts
@@ -21,6 +21,7 @@ export const buildApi = {
   subjectStreamUrl: (id: number) => subjectApi.stream(BUILD, WORK_ITEM, id),
   transcriptBase: (callId: string) => subjectApi.transcriptBase(BUILD, callId),
   transcriptFiles: (callId: string) => subjectApi.transcriptFiles(BUILD, callId),
+  transcriptFile: (callId: string, name: string) => subjectApi.transcriptFile(BUILD, callId, name),
   history: (limit?: number) => {
     const qs = limit !== undefined ? `?limit=${limit}` : ''
     return getJSON<WorkItemsHistoryResponse>(`/api/${BUILD}/work-items/history${qs}`)