From 1c872974e27fce40c6df9731a37fd1afeaaba904 Mon Sep 17 00:00:00 2001 From: Paulo Date: Mon, 13 Jul 2026 11:04:02 +0200 Subject: [PATCH 01/12] Subject keying rides DBOS workflow attributes; durable_runs shrinks start() stamps {subject_type, subject_id, extension} as DBOS custom workflow attributes at enqueue (subject_id always a string), so every runs-for-a-subject query reads workflow_status itself through one EXISTS predicate (subject_filter). The subject, extension, and input columns leave durable_runs; the row keeps only the facts DBOS has no slot for: the gate ask, failure, and timestamps. Events and signals take subject/extension from the workflow's own arguments (replay-deterministic), the park notification receives the subject from the parking workflow, and the transcript-files route names its own mount instead of reading the run's stamp. The migration runs DBOS's system migrations first so the attributes backfill from durable_runs always has its column, whatever the host booted last. Co-Authored-By: Claude Fable 5 --- backend/druks/build/extension.py | 3 +- backend/druks/build/scoping/workflows.py | 3 +- backend/druks/durable/dbos_state.py | 31 +++++- backend/druks/durable/models.py | 46 ++------- backend/druks/durable/reads.py | 4 +- backend/druks/durable/schemas.py | 8 +- backend/druks/extensions/base.py | 2 +- backend/druks/workflows.py | 94 ++++++++++++------- ..._subject_keying_via_workflow_attributes.py | 79 ++++++++++++++++ backend/tests/conftest.py | 33 ++++--- backend/tests/test_api_work_items.py | 9 +- backend/tests/test_durable_sdk.py | 18 +++- backend/tests/test_generic_subjects.py | 11 ++- backend/tests/test_lane_reactions.py | 15 +-- backend/tests/test_manifest.py | 2 +- backend/tests/test_profiling.py | 9 +- backend/tests/test_run_state.py | 18 +++- backend/tests/test_scope_reply_rescope.py | 9 +- docs/concepts.md | 8 +- 19 files changed, 271 insertions(+), 131 deletions(-) create mode 100644 backend/migrations/versions/d7e8f9a0b1c2_subject_keying_via_workflow_attributes.py diff --git a/backend/druks/build/extension.py b/backend/druks/build/extension.py index e57c4364..219407d9 100644 --- a/backend/druks/build/extension.py +++ b/backend/druks/build/extension.py @@ -17,6 +17,7 @@ from druks.build.models import WorkItem from druks.build.scoping.contracts import ScopeBriefOutput from druks.db import db_session +from druks.durable.dbos_state import subject_filter from druks.durable.enums import RunState from druks.events import Event, FeedItem from druks.extensions import Extension @@ -221,7 +222,7 @@ async def subject_activity(cls, subject_id: str) -> SubjectActivity | None: if not run: newest_scope = ( select(Run) - .where(Run.kind == Scope.kind, Run.subject["id"].as_integer() == item.id) + .where(Run.kind == Scope.kind, subject_filter(Run.id, "work_item", str(item.id))) .order_by(Run.created_at.desc()) .limit(1) ) diff --git a/backend/druks/build/scoping/workflows.py b/backend/druks/build/scoping/workflows.py index 55944dc4..901ccf1f 100644 --- a/backend/druks/build/scoping/workflows.py +++ b/backend/druks/build/scoping/workflows.py @@ -6,6 +6,7 @@ from druks.build.models import Project, ProjectRepo, WorkItem from druks.build.scoping.contracts import ScopeBriefOutput from druks.db import db_session +from druks.durable.dbos_state import subject_filter from druks.durable.enums import RunState from druks.ticketing.datastructures import Ticket from druks.workflows import Gate, Run, Workflow @@ -60,7 +61,7 @@ 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, + subject_filter(Run.id, "work_item", str(work_item_id)), ) return db_session().scalars(stmt).first() diff --git a/backend/druks/durable/dbos_state.py b/backend/druks/durable/dbos_state.py index ad0e2504..25d17cce 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,41 @@ # 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), + # The custom workflow attributes start() stamps: subject_type, subject_id + # (always stored as a string), extension. + 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..0bb39a64 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() @@ -102,10 +82,7 @@ def list_for_subject(cls, subject_type: str, subject_id: str) -> list["Run"]: # subject's lifecycle spans many runs, so its status is theirs aggregated. 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()) ) return list(db_session().scalars(stmt)) @@ -150,16 +127,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 +340,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/reads.py b/backend/druks/durable/reads.py index 5f482687..fcf7172b 100644 --- a/backend/druks/durable/reads.py +++ b/backend/druks/durable/reads.py @@ -29,11 +29,11 @@ _TERMINAL_CALL_STATES = {"succeeded", "failed", "abandoned"} -def get_agent_call_files(call_id: str) -> AgentCallFiles | None: +def get_agent_call_files(call_id: str, extension: str) -> AgentCallFiles | None: call = AgentCall.get(call_id) if not call: return None - return AgentCallFiles.from_call(call, Artifact.get_for_call(call.id)) + return AgentCallFiles.from_call(call, Artifact.get_for_call(call.id), extension) def list_subject_timeline(subject_type: str, subject_id: str) -> list[RunResponse]: diff --git a/backend/druks/durable/schemas.py b/backend/druks/durable/schemas.py index 4873f772..aa034cc8 100644 --- a/backend/druks/durable/schemas.py +++ b/backend/druks/durable/schemas.py @@ -106,9 +106,13 @@ class AgentCallFiles(BaseResponse): artifact: ArtifactDescriptor | None = None @classmethod - def from_call(cls, call: AgentCall, artifact: Artifact | None) -> "AgentCallFiles": + def from_call( + cls, call: AgentCall, artifact: Artifact | None, extension: str + ) -> "AgentCallFiles": + # ``extension`` is the transcript router serving these URLs — the caller + # is that router, so it names its own mount. layout = call.artifact_layout - base = f"/api/{call.run.extension}/transcripts/{call.id}/files/" + base = f"/api/{extension}/transcripts/{call.id}/files/" def named(path: Path) -> ArtifactFile | None: if not path.is_file(): diff --git a/backend/druks/extensions/base.py b/backend/druks/extensions/base.py index 8ef83d43..ca784377 100644 --- a/backend/druks/extensions/base.py +++ b/backend/druks/extensions/base.py @@ -317,7 +317,7 @@ async def stream_transcript( @router.get("/files", response_model=AgentCallFiles, response_model_by_alias=True) async def list_files(call_id: str) -> AgentCallFiles: - files = reads.get_agent_call_files(call_id) + files = reads.get_agent_call_files(call_id, cls.name) if not files: raise HTTPException(status.HTTP_404_NOT_FOUND, "Agent call not found.") return files 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..bcf06dc1 --- /dev/null +++ b/backend/migrations/versions/d7e8f9a0b1c2_subject_keying_via_workflow_attributes.py @@ -0,0 +1,79 @@ +"""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 dbos import run_dbos_database_migrations +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: + # DBOS owns workflow_status and adds its attributes column in its own + # migrations, which normally run at app launch — after this history on an + # upgrading host. Run them first so the backfill always has the column. + url = op.get_bind().engine.url.render_as_string(hide_password=False) + run_dbos_database_migrations( + url.replace("postgresql+psycopg://", "postgresql://", 1), schema="dbos" + ) + # Every run's subject/extension moves onto its workflow as custom attributes, + # the shape start() stamps for new runs: subject_id always a string. + op.execute( + """ + UPDATE dbos.workflow_status ws + SET attributes = coalesce(ws.attributes, '{}'::jsonb) + || CASE WHEN r.subject IS NULL THEN '{}'::jsonb + ELSE jsonb_build_object( + 'subject_type', r.subject->>'type', + 'subject_id', r.subject->>'id') END + || CASE WHEN r.extension IS NULL THEN '{}'::jsonb + ELSE jsonb_build_object('extension', r.extension) END + FROM durable_runs r + WHERE ws.workflow_uuid = r.id + AND (r.subject IS NOT NULL OR r.extension IS NOT NULL) + """ + ) + 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=True), + ) + # Rebuild subject/extension from the attributes; the subject id comes back as + # a string (attributes never kept the original type). Workflow inputs are not + # recoverable from attributes, so input restores as {}. + op.execute( + """ + UPDATE durable_runs r + SET subject = CASE WHEN ws.attributes ? 'subject_type' + THEN jsonb_build_object( + 'type', ws.attributes->>'subject_type', + 'id', ws.attributes->>'subject_id') END, + extension = ws.attributes->>'extension' + FROM dbos.workflow_status ws + WHERE ws.workflow_uuid = r.id AND ws.attributes IS NOT NULL + """ + ) + op.execute("UPDATE durable_runs SET input = '{}'::jsonb") + op.alter_column("durable_runs", "input", nullable=False) 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_manifest.py b/backend/tests/test_manifest.py index c54caffe..7f3e0eca 100644 --- a/backend/tests/test_manifest.py +++ b/backend/tests/test_manifest.py @@ -207,7 +207,7 @@ def test_manifest_surfaces_in_agent_call_files(tmp_path, db_session): manifest = _build() with mock.patch("druks.durable.models.load_settings", return_value=make_settings(tmp_path)): persist_manifest(call.call_dir.parent, call_id=call.call_dir.name, manifest=manifest) - files = AgentCallFiles.from_call(call, None) + files = AgentCallFiles.from_call(call, None, "build") assert files.manifest assert files.manifest.name == "manifest.json" 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 From b578e5a142d1a2266554dca47c255bd5964c35eb Mon Sep 17 00:00:00 2001 From: Paulo Date: Mon, 13 Jul 2026 11:20:52 +0200 Subject: [PATCH 02/12] Migration drops the moved columns outright; no backfill Every host already runs dbos 2.26, so the attributes column exists before this history does, and runs predating the deploy stay unkeyed. Co-Authored-By: Claude Fable 5 --- ..._subject_keying_via_workflow_attributes.py | 51 +++---------------- 1 file changed, 8 insertions(+), 43 deletions(-) diff --git a/backend/migrations/versions/d7e8f9a0b1c2_subject_keying_via_workflow_attributes.py b/backend/migrations/versions/d7e8f9a0b1c2_subject_keying_via_workflow_attributes.py index bcf06dc1..1060b2f9 100644 --- a/backend/migrations/versions/d7e8f9a0b1c2_subject_keying_via_workflow_attributes.py +++ b/backend/migrations/versions/d7e8f9a0b1c2_subject_keying_via_workflow_attributes.py @@ -10,7 +10,6 @@ import sqlalchemy as sa from alembic import op -from dbos import run_dbos_database_migrations from sqlalchemy.dialects import postgresql # revision identifiers, used by Alembic. @@ -21,30 +20,8 @@ def upgrade() -> None: - # DBOS owns workflow_status and adds its attributes column in its own - # migrations, which normally run at app launch — after this history on an - # upgrading host. Run them first so the backfill always has the column. - url = op.get_bind().engine.url.render_as_string(hide_password=False) - run_dbos_database_migrations( - url.replace("postgresql+psycopg://", "postgresql://", 1), schema="dbos" - ) - # Every run's subject/extension moves onto its workflow as custom attributes, - # the shape start() stamps for new runs: subject_id always a string. - op.execute( - """ - UPDATE dbos.workflow_status ws - SET attributes = coalesce(ws.attributes, '{}'::jsonb) - || CASE WHEN r.subject IS NULL THEN '{}'::jsonb - ELSE jsonb_build_object( - 'subject_type', r.subject->>'type', - 'subject_id', r.subject->>'id') END - || CASE WHEN r.extension IS NULL THEN '{}'::jsonb - ELSE jsonb_build_object('extension', r.extension) END - FROM durable_runs r - WHERE ws.workflow_uuid = r.id - AND (r.subject IS NOT NULL OR r.extension IS NOT NULL) - """ - ) + # 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") @@ -58,22 +35,10 @@ def downgrade() -> None: ) op.add_column( "durable_runs", - sa.Column("input", postgresql.JSONB(astext_type=sa.Text()), nullable=True), - ) - # Rebuild subject/extension from the attributes; the subject id comes back as - # a string (attributes never kept the original type). Workflow inputs are not - # recoverable from attributes, so input restores as {}. - op.execute( - """ - UPDATE durable_runs r - SET subject = CASE WHEN ws.attributes ? 'subject_type' - THEN jsonb_build_object( - 'type', ws.attributes->>'subject_type', - 'id', ws.attributes->>'subject_id') END, - extension = ws.attributes->>'extension' - FROM dbos.workflow_status ws - WHERE ws.workflow_uuid = r.id AND ws.attributes IS NOT NULL - """ + sa.Column( + "input", + postgresql.JSONB(astext_type=sa.Text()), + nullable=False, + server_default=sa.text("'{}'::jsonb"), + ), ) - op.execute("UPDATE durable_runs SET input = '{}'::jsonb") - op.alter_column("durable_runs", "input", nullable=False) From 232fdb6034842e1981b0bd58297b4cf092017f64 Mon Sep 17 00:00:00 2001 From: Paulo Date: Mon, 13 Jul 2026 11:24:13 +0200 Subject: [PATCH 03/12] Extensions read runs through the author door, not engine internals Run.list_for_subject grows a kind filter; build's newest-scope lookup and Scope.parked_for compose it instead of importing the raw subject_filter predicate from the durable engine. Per (kind, subject) the queue dedup makes runs strictly sequential, so newest-first holds within a kind and at most one run is parked. Co-Authored-By: Claude Fable 5 --- backend/druks/build/extension.py | 14 +++----------- backend/druks/build/scoping/workflows.py | 15 +++------------ backend/druks/durable/models.py | 11 +++++++++-- 3 files changed, 15 insertions(+), 25 deletions(-) diff --git a/backend/druks/build/extension.py b/backend/druks/build/extension.py index 219407d9..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,11 +16,9 @@ from druks.build.models import WorkItem from druks.build.scoping.contracts import ScopeBriefOutput from druks.db import db_session -from druks.durable.dbos_state import subject_filter -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 @@ -220,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, subject_filter(Run.id, "work_item", str(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 901ccf1f..e117e75d 100644 --- a/backend/druks/build/scoping/workflows.py +++ b/backend/druks/build/scoping/workflows.py @@ -1,15 +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.dbos_state import subject_filter -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__) @@ -58,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, - subject_filter(Run.id, "work_item", str(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/models.py b/backend/druks/durable/models.py index 0bb39a64..88eb344b 100644 --- a/backend/druks/durable/models.py +++ b/backend/druks/durable/models.py @@ -77,14 +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(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]: From 647bc7f41a5edccfd45e93281146573b9e14db6f Mon Sep 17 00:00:00 2001 From: Paulo Date: Mon, 13 Jul 2026 14:21:28 +0200 Subject: [PATCH 04/12] Run.extension reads back from the stamped attributes; nothing threads it Co-Authored-By: Claude Fable 5 --- backend/druks/durable/dbos_state.py | 11 +++++++++-- backend/druks/durable/models.py | 10 +++++++++- backend/druks/durable/reads.py | 4 ++-- backend/druks/durable/schemas.py | 8 ++------ backend/druks/extensions/base.py | 2 +- backend/tests/test_manifest.py | 2 +- 6 files changed, 24 insertions(+), 13 deletions(-) diff --git a/backend/druks/durable/dbos_state.py b/backend/druks/durable/dbos_state.py index 25d17cce..b3246f93 100644 --- a/backend/druks/durable/dbos_state.py +++ b/backend/druks/durable/dbos_state.py @@ -25,12 +25,19 @@ sa.Column("workflow_uuid", sa.String, primary_key=True), sa.Column("status", sa.String), sa.Column("updated_at", sa.BigInteger), - # The custom workflow attributes start() stamps: subject_type, subject_id - # (always stored as a string), extension. 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: diff --git a/backend/druks/durable/models.py b/backend/druks/durable/models.py index 88eb344b..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, subject_filter, 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 @@ -50,6 +55,9 @@ class Run(Base): # 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( diff --git a/backend/druks/durable/reads.py b/backend/druks/durable/reads.py index fcf7172b..5f482687 100644 --- a/backend/druks/durable/reads.py +++ b/backend/druks/durable/reads.py @@ -29,11 +29,11 @@ _TERMINAL_CALL_STATES = {"succeeded", "failed", "abandoned"} -def get_agent_call_files(call_id: str, extension: str) -> AgentCallFiles | None: +def get_agent_call_files(call_id: str) -> AgentCallFiles | None: call = AgentCall.get(call_id) if not call: return None - return AgentCallFiles.from_call(call, Artifact.get_for_call(call.id), extension) + return AgentCallFiles.from_call(call, Artifact.get_for_call(call.id)) def list_subject_timeline(subject_type: str, subject_id: str) -> list[RunResponse]: diff --git a/backend/druks/durable/schemas.py b/backend/druks/durable/schemas.py index aa034cc8..4873f772 100644 --- a/backend/druks/durable/schemas.py +++ b/backend/druks/durable/schemas.py @@ -106,13 +106,9 @@ class AgentCallFiles(BaseResponse): artifact: ArtifactDescriptor | None = None @classmethod - def from_call( - cls, call: AgentCall, artifact: Artifact | None, extension: str - ) -> "AgentCallFiles": - # ``extension`` is the transcript router serving these URLs — the caller - # is that router, so it names its own mount. + def from_call(cls, call: AgentCall, artifact: Artifact | None) -> "AgentCallFiles": layout = call.artifact_layout - base = f"/api/{extension}/transcripts/{call.id}/files/" + base = f"/api/{call.run.extension}/transcripts/{call.id}/files/" def named(path: Path) -> ArtifactFile | None: if not path.is_file(): diff --git a/backend/druks/extensions/base.py b/backend/druks/extensions/base.py index ca784377..8ef83d43 100644 --- a/backend/druks/extensions/base.py +++ b/backend/druks/extensions/base.py @@ -317,7 +317,7 @@ async def stream_transcript( @router.get("/files", response_model=AgentCallFiles, response_model_by_alias=True) async def list_files(call_id: str) -> AgentCallFiles: - files = reads.get_agent_call_files(call_id, cls.name) + files = reads.get_agent_call_files(call_id) if not files: raise HTTPException(status.HTTP_404_NOT_FOUND, "Agent call not found.") return files diff --git a/backend/tests/test_manifest.py b/backend/tests/test_manifest.py index 7f3e0eca..c54caffe 100644 --- a/backend/tests/test_manifest.py +++ b/backend/tests/test_manifest.py @@ -207,7 +207,7 @@ def test_manifest_surfaces_in_agent_call_files(tmp_path, db_session): manifest = _build() with mock.patch("druks.durable.models.load_settings", return_value=make_settings(tmp_path)): persist_manifest(call.call_dir.parent, call_id=call.call_dir.name, manifest=manifest) - files = AgentCallFiles.from_call(call, None, "build") + files = AgentCallFiles.from_call(call, None) assert files.manifest assert files.manifest.name == "manifest.json" From 22e91a2e72c2ec4f2ce1a12d193bb25a850b3f44 Mon Sep 17 00:00:00 2001 From: Paulo Date: Mon, 13 Jul 2026 17:20:57 +0200 Subject: [PATCH 05/12] Extension is workflow-class identity; nothing carries it per run MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The declaring extension namespaces every workflow's kind (registered by the loader before extension modules import), so extension resolves from the class: start() loses its extension argument, the DBOS attributes stamp subject only, lifecycle events derive their extension from the run's kind, and the transcript file listing returns names the client composes into URLs it already knows. Deploy note: _entry dropped its third argument — drain or cancel in-flight runs before deploying; DBOS recovery re-invokes recorded arguments. Co-Authored-By: Claude Fable 5 --- AGENTS.md | 3 + backend/druks/build/routes.py | 8 +- backend/druks/build/scoping/workflows.py | 1 - backend/druks/build/subscribers.py | 1 - backend/druks/build/workflows.py | 1 - backend/druks/durable/dbos_state.py | 9 -- backend/druks/durable/models.py | 10 +- backend/druks/durable/schemas.py | 27 ++-- backend/druks/extensions/loader.py | 12 +- backend/druks/extensions/registry.py | 36 +++++ backend/druks/workflows.py | 84 ++++------ backend/tests/conftest.py | 33 ++-- .../druks_field_notes/workflows.py | 1 - backend/tests/test_extension_appless_load.py | 4 + backend/tests/test_extension_doctor_checks.py | 4 + backend/tests/test_lane_reactions.py | 6 +- backend/tests/test_manifest.py | 4 +- backend/tests/test_project_repo_routes.py | 6 +- backend/tests/test_proof_extension.py | 15 +- backend/tests/test_run_state.py | 9 +- backend/tests/test_scaffolding.py | 4 + backend/tests/test_scope_durable.py | 1 - backend/tests/test_webhooks_push.py | 2 +- backend/tests/test_workflow_identity.py | 148 ++++++++++++++++++ docs/concepts.md | 10 +- docs/writing-an-extension.md | 12 +- frontend/src/api/client.ts | 2 + frontend/src/api/types.ts | 6 +- .../src/extensions/build/AgentCallPage.tsx | 9 +- frontend/src/extensions/build/api.ts | 1 + 30 files changed, 321 insertions(+), 148 deletions(-) create mode 100644 backend/tests/test_workflow_identity.py diff --git a/AGENTS.md b/AGENTS.md index 5554f48c..952dcd5b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -32,6 +32,9 @@ For extension-surface changes, inspect the proof extension at arbitrary-line resume or exactly-once external side effects. - `Run.state` is derived from DBOS workflow status. Do not add a second writable state mirror. +- Workflow ownership is class identity: the declaring extension namespaces the + workflow's `kind` and is resolved from the registry. Never accept, thread, or + store an extension per run. - Extension authors import the public concern namespaces documented in `docs/writing-an-extension.md`, not Druks internals. - Backend extension discovery is runtime packaging. Shared-dashboard extension UI 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 e117e75d..26d8b600 100644 --- a/backend/druks/build/scoping/workflows.py +++ b/backend/druks/build/scoping/workflows.py @@ -46,7 +46,6 @@ 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, ) 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 b3246f93..f91a59cd 100644 --- a/backend/druks/durable/dbos_state.py +++ b/backend/druks/durable/dbos_state.py @@ -29,15 +29,6 @@ ) -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: diff --git a/backend/druks/durable/models.py b/backend/druks/durable/models.py index 645d1d78..88eb344b 100644 --- a/backend/druks/durable/models.py +++ b/backend/druks/durable/models.py @@ -11,12 +11,7 @@ from druks.core.models import Uuid7Pk from druks.database import db_session, get_session -from druks.durable.dbos_state import ( - extension_expression, - state_expression, - subject_filter, - 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 @@ -55,9 +50,6 @@ class Run(Base): # 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( 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..5f67719b 100644 --- a/backend/druks/extensions/loader.py +++ b/backend/druks/extensions/loader.py @@ -4,6 +4,7 @@ from druks.extensions import Extension from .exceptions import ExtensionImportError, ExtensionNotFound, MalformedExtension +from .registry import register_workflow_package if TYPE_CHECKING: from importlib.metadata import EntryPoint @@ -19,15 +20,22 @@ def iter_extensions() -> list[type[Extension]]: import, resolves to a non-``Extension``, or collides on ``name`` raises — there is no per-extension fault tolerance yet (deferred until a real external extension exists).""" + entries = list(entry_points(group=_GROUP)) + # Ownership registers before any extension module imports, so every Workflow + # class — including one the first entry's import pulls in — resolves its + # declaring extension at definition time. + for entry in entries: + register_workflow_package(entry.module.rsplit(".", 1)[0], entry.name) extensions: list[type[Extension]] = [] seen: set[str] = set() - for entry in entry_points(group=_GROUP): + for entry in entries: 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 in seen: raise ValueError(f"duplicate extension name {extension.name!r}") seen.add(extension.name) + register_workflow_package(extension.package, extension.name) extensions.append(extension) return extensions @@ -78,6 +86,7 @@ def _resolve(name: str) -> type[Extension]: f"extension {name!r} is declared by {len(matches)} installed packages " f"({', '.join(e.value for e in matches)}) — uninstall all but one" ) + register_workflow_package(matches[0].module.rsplit(".", 1)[0], name) 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 @@ -91,6 +100,7 @@ def _resolve(name: str) -> type[Extension]: 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/extensions/registry.py b/backend/druks/extensions/registry.py index 9bef1087..476f163b 100644 --- a/backend/druks/extensions/registry.py +++ b/backend/druks/extensions/registry.py @@ -61,6 +61,42 @@ def autodiscover(package: str) -> list[ModuleType]: return modules +# Which extension owns each workflow-declaring package, stamped by the loader +# before it imports any extension module — 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: + if package in _workflow_packages: + if _workflow_packages[package] != extension: + raise ValueError( + 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 ValueError( + 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) + + webhooks = Registry("webhooks", key=lambda cls: f"{cls.__module__}.{cls.__qualname__}") workflows = Registry("workflows", key=lambda cls: cls.kind) agents = Registry("agents", key=lambda agent: agent.id) diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py index 48a8b138..c76f35ae 100644 --- a/backend/druks/workflows.py +++ b/backend/druks/workflows.py @@ -26,7 +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.registry import workflows +from druks.extensions.registry import resolve_workflow_extension, workflows from druks.extensions.settings import ( coerce_setting_value, validate_setting_override, @@ -157,12 +157,6 @@ def _kind_from_class_name(name: str) -> str: return _CAMEL_BOUNDARY.sub("_", name).lower() -def _namespace_from_module(module: str) -> str | None: - # druks..… → ; 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 @@ -223,7 +217,6 @@ async def _park( workflow.workflow_id, RunState.PENDING_INPUT, subject=workflow.subject, - extension=workflow.extension, facts={ "input_gate": topic, "input_request": input_request, @@ -240,7 +233,6 @@ async def _park( workflow.workflow_id, RunState.RUNNING, subject=workflow.subject, - extension=workflow.extension, facts=_GATE_CLEARED, ) return payload @@ -292,7 +284,6 @@ async def _emit_run_event( event: RunState, *, subject: dict[str, Any] | None, - extension: str | None, facts: dict[str, Any] | None = None, result: Any = None, ) -> None: @@ -300,8 +291,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. subject/extension come from the - # workflow's own arguments, so a replay stamps the same routing every time. + # 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) @@ -315,7 +306,7 @@ async def _transition() -> dict[str, Any] | None: return { "kind": run.kind, "subject": subject, - "payload": _log_run_event(run, event, subject, extension, result), + "payload": _log_run_event(run, event, subject, result), } transition = await DBOS.run_step_async( @@ -342,7 +333,6 @@ 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 @@ -362,7 +352,7 @@ def _log_run_event( type=f"run.{event.value}", subject=subject, payload=payload, - extension=extension, + extension=workflows.get(run.kind).extension, ) return payload @@ -376,7 +366,6 @@ async def _execute_run( workflow_id: str, kind: str, subject: dict[str, Any] | None, - extension: str | None, body: Callable, ) -> Any: # Ensure the row (idempotent, so a scheduled run with no start() makes it @@ -385,7 +374,7 @@ async def _execute_run( # 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) - await _emit_run_event(workflow_id, RunState.RUNNING, subject=subject, extension=extension) + await _emit_run_event(workflow_id, RunState.RUNNING, subject=subject) try: result = await body() except (DBOSAwaitedWorkflowCancelledError, DBOSWorkflowCancelledError): @@ -395,7 +384,6 @@ async def _execute_run( workflow_id, RunState.FAILED, subject=subject, - extension=extension, facts={ **_GATE_CLEARED, "failure": str(exc), @@ -403,14 +391,16 @@ async def _execute_run( }, ) raise - await _emit_run_event( - workflow_id, RunState.FINISHED, subject=subject, extension=extension, 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 @@ -438,10 +428,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). @@ -456,9 +458,6 @@ def __init__(self) -> None: # 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 @@ -603,13 +602,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) @@ -643,18 +636,16 @@ async def start( # 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] = {} + attributes = None if subject: attributes = {"subject_type": subject["type"], "subject_id": str(subject["id"])} - if extension: - attributes["extension"] = extension try: with ( SetWorkflowID(workflow_id), - SetWorkflowAttributes(attributes or None), + SetWorkflowAttributes(attributes), enqueue_options, ): - await run_queue.enqueue_async(cls._entry, wire, subject, extension) + await run_queue.enqueue_async(cls._entry, wire, subject) except DBOSQueueDeduplicatedError as duplicate: holder = _get_dbos_instance()._sys_db.get_deduplicated_workflow( run_queue.name, duplicate.deduplication_id @@ -663,7 +654,7 @@ 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) @@ -700,12 +691,10 @@ async def _run_instance( cls: type[Workflow], input: dict[str, Any] | None = None, subject: dict[str, Any] | None = None, - extension: str | None = None, ) -> Any: 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. @@ -720,7 +709,6 @@ async def _run_instance( instance._workflow_id, cls.kind, subject, - extension, lambda: getattr(instance, cls._body_method)(**run_kwargs), ) finally: @@ -730,12 +718,8 @@ async def _run_instance( def _register_entry(cls: type[Workflow]) -> None: @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(input: dict[str, Any], subject: dict[str, Any] | None = None) -> None: + await _run_instance(cls, input, subject) cls._entry = staticmethod(_entry) # type: ignore[assignment] diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 9120ee7a..c15ad7da 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -33,6 +33,17 @@ # 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. +from druks.extensions.loader import iter_extensions # noqa: E402 +from druks.extensions.registry import register_workflow_package # noqa: E402 + +iter_extensions() +for _module in ("test_durable_sdk", "test_notifications_durable"): + register_workflow_package(_module, None) + class FakeRedis: # The subset the run lock and the MCP OAuth cache use; one instance per @@ -344,12 +355,10 @@ def bind_ambient_session(session) -> None: db_session.registry.set(session) -def seed_dbos_status( - session, workflow_id: str, state: str, *, subject=None, extension=None -) -> None: +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/extension attributes ``start()`` stamps — the paired - half of every persisted run seed (there is no state column).""" + 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 @@ -362,11 +371,9 @@ def seed_dbos_status( "cancelled": "CANCELLED", }[state] now_ms = int(Base.utc_now().timestamp() * 1000) - attributes = {} + attributes = None 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 and subject keying read only these. session.execute( @@ -374,7 +381,7 @@ def seed_dbos_status( workflow_uuid=workflow_id, status=status, updated_at=now_ms, - attributes=attributes or None, + attributes=attributes, ) ) session.flush() @@ -408,13 +415,7 @@ def seed_build_run( ) session.add(run) session.flush() - seed_dbos_status( - session, - run.id, - state, - subject={"type": "work_item", "id": work_item_id}, - extension="build", - ) + 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() 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_extension_appless_load.py b/backend/tests/test_extension_appless_load.py index 8a651688..417c83cc 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 registry as extensions_registry 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_registry._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_registry._workflow_packages.clear() + extensions_registry._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_extension_doctor_checks.py b/backend/tests/test_extension_doctor_checks.py index b036ed4a..58974601 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 registry as extensions_registry 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_registry._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_registry._workflow_packages.clear() + extensions_registry._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_lane_reactions.py b/backend/tests/test_lane_reactions.py index 8ed7905c..decddc1b 100644 --- a/backend/tests/test_lane_reactions.py +++ b/backend/tests/test_lane_reactions.py @@ -97,11 +97,7 @@ def _parked_scope_run(session, *, work_item_id): session.add(run) session.flush() seed_dbos_status( - session, - run.id, - "pending_input", - subject={"type": "work_item", "id": work_item_id}, - extension="build", + session, run.id, "pending_input", subject={"type": "work_item", "id": work_item_id} ) return run 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_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..54438c0b 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 registry as extensions_registry 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_registry._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_registry._workflow_packages.clear() + extensions_registry._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 794f99b8..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 @@ -143,7 +144,6 @@ async def _raises(**_: object) -> None: run.id, RunState.PENDING_INPUT, subject={"type": "work_item", "id": item.id}, - extension="build", facts={"input_gate": "review_work", "input_request": {"label": "Review"}}, ) @@ -176,7 +176,6 @@ async def _reads_both(*, run: str, **kwargs: object) -> None: run.id, RunState.FINISHED, subject={"type": "work_item", "id": item.id}, - extension="build", result={"status": "ok"}, ) @@ -197,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, {"type": "work_item", "id": item.id}, "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 @@ -226,7 +225,7 @@ async def body() -> None: raise FatalError("closed at review") with pytest.raises(FatalError): - await _execute_run(run.id, run.kind, {"type": "work_item", "id": item.id}, "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) @@ -253,7 +252,7 @@ async def body() -> None: raise GateTimeout("review_work") with pytest.raises(GateTimeout): - await _execute_run(run.id, run.kind, {"type": "work_item", "id": item.id}, "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_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..32248a60 --- /dev/null +++ b/backend/tests/test_workflow_identity.py @@ -0,0 +1,148 @@ +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 registry as extensions_registry +from druks.extensions.registry import ( + register_workflow_package, + resolve_workflow_extension, + 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_registry._workflow_packages) + items = dict(workflows._items) + yield + extensions_registry._workflow_packages.clear() + extensions_registry._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(ValueError, 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(ValueError, 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 5f22e6aa..af33114d 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -92,10 +92,12 @@ extension still owns domain policy and side-effect idempotency. ### State has one lifecycle owner 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: +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..a82d0e73 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, ) ``` @@ -136,6 +135,17 @@ active run of a workflow kind; a duplicate start returns the active run id. Wrap `start()` in a domain `dispatch()` method when the extension needs lookup, snapshot, or routing policy before launch. +A workflow's identity comes from the extension that declares it — never from +the caller. The loader registers each extension's package before importing its +modules, so `Sweep.extension` is `"night_watch"` and its durable kind is +`night_watch.sweep`: the class name, lower-snake, namespaced by the extension. +Declare `kind = "..."` to override the local name only — it is still prefixed, +and a dotted value is rejected. Because the namespace is the extension name, +two extensions can ship a `Sweep` without colliding on settings keys, dedup +slots, or DBOS workflow names. Importing a workflow module outside the loader +(a bare script, a REPL) fails at class definition until the package is +registered — load through `druks.extensions.loader` instead. + ### Schedules and settings Set `every` to declare a cron: 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(`/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//transcripts//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(initialLeft) + const activeFile = leftTab === 'prompt' ? files.prompt : files.response return ( @@ -143,13 +144,7 @@ function RunView({ response - +
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(`/api/${BUILD}/work-items/history${qs}`) From 9951558ff6563f2610423d487f49be5743e23597 Mon Sep 17 00:00:00 2001 From: Paulo Date: Mon, 13 Jul 2026 17:49:04 +0200 Subject: [PATCH 06/12] Review fixes: loader claim releases on failed load; files tests read names MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit An aliased entry point (key != Extension.name) no longer poisons the package ownership map: the pre-import claim is provisional — released when the load fails, and an already-held claim is never touched, so the name-mismatch check raises the promised MalformedExtension in both orders. The two transcript-files tests missed in the url removal now assert names and compose the download URL the way the client does. Co-Authored-By: Claude Fable 5 --- backend/druks/extensions/loader.py | 39 +++++++++++++++---------- backend/druks/extensions/registry.py | 19 +++++++++++- backend/tests/test_api_runs.py | 11 ++++--- backend/tests/test_workflow_identity.py | 26 +++++++++++++++++ 4 files changed, 74 insertions(+), 21 deletions(-) diff --git a/backend/druks/extensions/loader.py b/backend/druks/extensions/loader.py index 5f67719b..2bf918e0 100644 --- a/backend/druks/extensions/loader.py +++ b/backend/druks/extensions/loader.py @@ -4,7 +4,7 @@ from druks.extensions import Extension from .exceptions import ExtensionImportError, ExtensionNotFound, MalformedExtension -from .registry import register_workflow_package +from .registry import claimed_workflow_package, register_workflow_package if TYPE_CHECKING: from importlib.metadata import EntryPoint @@ -86,21 +86,28 @@ def _resolve(name: str) -> type[Extension]: f"extension {name!r} is declared by {len(matches)} installed packages " f"({', '.join(e.value for e in matches)}) — uninstall all but one" ) - register_workflow_package(matches[0].module.rsplit(".", 1)[0], name) - 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). - 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) + try: + with claimed_workflow_package(matches[0].module.rsplit(".", 1)[0], name): + 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). + 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) + except ValueError as error: + # A claim the ownership map rejects (a nested-package overlap, one package + # split across entry names) is a broken install — same taxonomy as a key + # mismatch, caught here so the loader's error contract holds. + raise MalformedExtension(str(error)) from error return extension diff --git a/backend/druks/extensions/registry.py b/backend/druks/extensions/registry.py index 476f163b..a78d9c34 100644 --- a/backend/druks/extensions/registry.py +++ b/backend/druks/extensions/registry.py @@ -1,6 +1,7 @@ import importlib import pkgutil -from collections.abc import Callable +from collections.abc import Callable, Iterator +from contextlib import contextmanager from types import ModuleType from typing import Any @@ -97,6 +98,22 @@ def resolve_workflow_extension(module: str) -> str | None: raise LookupError(module) +@contextmanager +def claimed_workflow_package(package: str, extension: str | None) -> Iterator[None]: + """Provisionally claim ``package`` for ``extension`` while its entry point + loads: a fresh claim is released if the load fails, and an existing claim — + whoever holds it — is left untouched for the post-load checks to judge.""" + fresh = package not in _workflow_packages + if fresh: + register_workflow_package(package, extension) + try: + yield + except BaseException: + if fresh: + del _workflow_packages[package] + raise + + webhooks = Registry("webhooks", key=lambda cls: f"{cls.__module__}.{cls.__qualname__}") workflows = Registry("workflows", key=lambda cls: cls.kind) agents = Registry("agents", key=lambda agent: agent.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_workflow_identity.py b/backend/tests/test_workflow_identity.py index 32248a60..3b696e69 100644 --- a/backend/tests/test_workflow_identity.py +++ b/backend/tests/test_workflow_identity.py @@ -9,6 +9,7 @@ from druks.events.models import Event from druks.extensions import registry as extensions_registry from druks.extensions.registry import ( + claimed_workflow_package, register_workflow_package, resolve_workflow_extension, workflows, @@ -55,6 +56,31 @@ def test_resolution_matches_package_boundaries(): resolve_workflow_extension("alpha_pkg_sibling.workflows") +def test_failed_claim_releases_a_fresh_registration(): + # A claim made for a load that then fails must not survive it — otherwise a + # malformed first attempt pins the package under the wrong name and a + # corrected retry can never register it. + with pytest.raises(RuntimeError), claimed_workflow_package("gamma_pkg", "gamma"): + raise RuntimeError("load failed") + with pytest.raises(LookupError): + resolve_workflow_extension("gamma_pkg.workflows") + + +def test_failed_claim_leaves_an_existing_registration_untouched(): + # An aliased entry claiming an already-owned package must not evict the real + # owner — the post-load name check judges the alias, the map stays truthful. + register_workflow_package("alpha_pkg", "alpha") + with pytest.raises(RuntimeError), claimed_workflow_package("alpha_pkg", "imposter"): + raise RuntimeError("load failed") + assert resolve_workflow_extension("alpha_pkg.workflows") == "alpha" + + +def test_successful_claim_keeps_the_registration(): + with claimed_workflow_package("delta_pkg", "delta"): + pass + assert resolve_workflow_extension("delta_pkg.workflows") == "delta" + + 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"): From 53bd7db571f1bd0747c78d4fd3267905c0f7616f Mon Sep 17 00:00:00 2001 From: Paulo Date: Mon, 13 Jul 2026 18:05:30 +0200 Subject: [PATCH 07/12] Entry takes subject before input; trim identity prose from docs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Subject is routing metadata and rides before the payload everywhere else (start, _execute_run, the event emitters) — the entry chain was the one deviator, flipped inside the same compatibility window the arity change already opened. Co-Authored-By: Claude Fable 5 --- AGENTS.md | 3 --- backend/druks/workflows.py | 11 +++++++---- docs/writing-an-extension.md | 11 ----------- 3 files changed, 7 insertions(+), 18 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 952dcd5b..5554f48c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -32,9 +32,6 @@ For extension-surface changes, inspect the proof extension at arbitrary-line resume or exactly-once external side effects. - `Run.state` is derived from DBOS workflow status. Do not add a second writable state mirror. -- Workflow ownership is class identity: the declaring extension namespaces the - workflow's `kind` and is resolved from the registry. Never accept, thread, or - store an extension per run. - Extension authors import the public concern namespaces documented in `docs/writing-an-extension.md`, not Druks internals. - Backend extension discovery is runtime packaging. Shared-dashboard extension UI diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py index c76f35ae..5d6c3727 100644 --- a/backend/druks/workflows.py +++ b/backend/druks/workflows.py @@ -645,7 +645,7 @@ async def start(cls, *, subject: dict[str, Any] | None, **input: Any) -> str: SetWorkflowAttributes(attributes), enqueue_options, ): - await run_queue.enqueue_async(cls._entry, wire, subject) + 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 @@ -689,8 +689,8 @@ async def _do() -> Any: async def _run_instance( cls: type[Workflow], - input: dict[str, Any] | None = None, subject: dict[str, Any] | None = None, + input: dict[str, Any] | None = None, ) -> Any: instance = cls() instance._workflow_id = DBOS.workflow_id # type: ignore[assignment] @@ -717,9 +717,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) -> None: - await _run_instance(cls, input, subject) + 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/docs/writing-an-extension.md b/docs/writing-an-extension.md index a82d0e73..2dd85c4f 100644 --- a/docs/writing-an-extension.md +++ b/docs/writing-an-extension.md @@ -135,17 +135,6 @@ active run of a workflow kind; a duplicate start returns the active run id. Wrap `start()` in a domain `dispatch()` method when the extension needs lookup, snapshot, or routing policy before launch. -A workflow's identity comes from the extension that declares it — never from -the caller. The loader registers each extension's package before importing its -modules, so `Sweep.extension` is `"night_watch"` and its durable kind is -`night_watch.sweep`: the class name, lower-snake, namespaced by the extension. -Declare `kind = "..."` to override the local name only — it is still prefixed, -and a dotted value is rejected. Because the namespace is the extension name, -two extensions can ship a `Sweep` without colliding on settings keys, dedup -slots, or DBOS workflow names. Importing a workflow module outside the loader -(a bare script, a REPL) fails at class definition until the package is -registered — load through `druks.extensions.loader` instead. - ### Schedules and settings Set `every` to declare a cron: From 0597f8a49ceb865536d6c8ded29e58f19525ddfb Mon Sep 17 00:00:00 2001 From: Paulo Date: Mon, 13 Jul 2026 18:34:39 +0200 Subject: [PATCH 08/12] Ownership conflicts raise MalformedExtension at the source The registry raising ValueError forced the loader to catch-and-rebrand at a distance; a conflicting claim IS a packaging mistake, so the registry raises the loader's taxonomy itself and _resolve flattens to a bare with-block. Co-Authored-By: Claude Fable 5 --- backend/druks/extensions/loader.py | 34 +++++++++---------------- backend/druks/extensions/registry.py | 8 ++++-- backend/tests/test_workflow_identity.py | 5 ++-- 3 files changed, 21 insertions(+), 26 deletions(-) diff --git a/backend/druks/extensions/loader.py b/backend/druks/extensions/loader.py index 2bf918e0..941ea08b 100644 --- a/backend/druks/extensions/loader.py +++ b/backend/druks/extensions/loader.py @@ -86,28 +86,18 @@ def _resolve(name: str) -> type[Extension]: f"extension {name!r} is declared by {len(matches)} installed packages " f"({', '.join(e.value for e in matches)}) — uninstall all but one" ) - try: - with claimed_workflow_package(matches[0].module.rsplit(".", 1)[0], name): - 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). - 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) - except ValueError as error: - # A claim the ownership map rejects (a nested-package overlap, one package - # split across entry names) is a broken install — same taxonomy as a key - # mismatch, caught here so the loader's error contract holds. - raise MalformedExtension(str(error)) from error + with claimed_workflow_package(matches[0].module.rsplit(".", 1)[0], name): + extension = _load_entry(matches[0]) + # 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/extensions/registry.py b/backend/druks/extensions/registry.py index a78d9c34..7e7462ab 100644 --- a/backend/druks/extensions/registry.py +++ b/backend/druks/extensions/registry.py @@ -5,6 +5,8 @@ from types import ModuleType from typing import Any +from .exceptions import MalformedExtension + # Leaf-module names that carry self-registering capabilities. ``autodiscover`` # imports exactly these (``routes`` defines routers the loader mounts; the rest # fire registration as an import side effect). The set is the single source of @@ -70,9 +72,11 @@ def autodiscover(package: str) -> list[ModuleType]: 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, so this raises the loader's own taxonomy. if package in _workflow_packages: if _workflow_packages[package] != extension: - raise ValueError( + raise MalformedExtension( f"package {package!r} already belongs to " f"{_workflow_packages[package]!r} — {extension!r} can't claim it" ) @@ -80,7 +84,7 @@ def register_workflow_package(package: str, extension: str | None) -> None: for registered, owner in _workflow_packages.items(): nested = registered.startswith(f"{package}.") or package.startswith(f"{registered}.") if nested and owner != extension: - raise ValueError( + raise MalformedExtension( f"package {package!r} overlaps {registered!r} (owned by {owner!r}) — " "workflow ownership must be unambiguous" ) diff --git a/backend/tests/test_workflow_identity.py b/backend/tests/test_workflow_identity.py index 3b696e69..49f68524 100644 --- a/backend/tests/test_workflow_identity.py +++ b/backend/tests/test_workflow_identity.py @@ -8,6 +8,7 @@ from druks.durable.schemas import get_display_label from druks.events.models import Event from druks.extensions import registry as extensions_registry +from druks.extensions.exceptions import MalformedExtension from druks.extensions.registry import ( claimed_workflow_package, register_workflow_package, @@ -39,13 +40,13 @@ async def run(self) -> None: ... def test_registration_is_idempotent_and_conflicts_loudly(): register_workflow_package("alpha_pkg", "alpha") register_workflow_package("alpha_pkg", "alpha") - with pytest.raises(ValueError, match="already belongs"): + 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(ValueError, match="overlaps"): + with pytest.raises(MalformedExtension, match="overlaps"): register_workflow_package("alpha_pkg.nested", "beta") From ddc2581143b7fcc366e0895042b823e56a4e5a10 Mon Sep 17 00:00:00 2001 From: Paulo Date: Mon, 13 Jul 2026 18:41:56 +0200 Subject: [PATCH 09/12] Name the entry-package convention; conftest imports live at the top Co-Authored-By: Claude Fable 5 --- backend/druks/extensions/loader.py | 13 ++++++++++--- backend/tests/conftest.py | 9 ++++----- 2 files changed, 14 insertions(+), 8 deletions(-) diff --git a/backend/druks/extensions/loader.py b/backend/druks/extensions/loader.py index 941ea08b..746c62c9 100644 --- a/backend/druks/extensions/loader.py +++ b/backend/druks/extensions/loader.py @@ -14,6 +14,12 @@ _GROUP = "druks.extensions" +def _entry_package(entry: "EntryPoint") -> str: + # The ``/extension.py`` convention: the entry module's parent is + # the extension's package — the same default ``Extension.package`` takes. + return entry.module.rsplit(".", 1)[0] + + def iter_extensions() -> list[type[Extension]]: """Every installed extension, resolved from the ``druks.extensions`` entry points. Loading an entry point imports the extension's class. An extension that fails to @@ -25,7 +31,7 @@ def iter_extensions() -> list[type[Extension]]: # class — including one the first entry's import pulls in — resolves its # declaring extension at definition time. for entry in entries: - register_workflow_package(entry.module.rsplit(".", 1)[0], entry.name) + register_workflow_package(_entry_package(entry), entry.name) extensions: list[type[Extension]] = [] seen: set[str] = set() for entry in entries: @@ -86,8 +92,9 @@ def _resolve(name: str) -> type[Extension]: f"extension {name!r} is declared by {len(matches)} installed packages " f"({', '.join(e.value for e in matches)}) — uninstall all but one" ) - with claimed_workflow_package(matches[0].module.rsplit(".", 1)[0], name): - extension = _load_entry(matches[0]) + entry = matches[0] + with claimed_workflow_package(_entry_package(entry), name): + extension = _load_entry(entry) # 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 diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index c15ad7da..116defd1 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -13,6 +13,8 @@ create_engine_from_url, init_db, ) +from druks.extensions.loader import iter_extensions +from druks.extensions.registry import register_workflow_package from druks.settings import Settings from sqlalchemy.orm import Session @@ -37,12 +39,9 @@ # 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. -from druks.extensions.loader import iter_extensions # noqa: E402 -from druks.extensions.registry import register_workflow_package # noqa: E402 - iter_extensions() -for _module in ("test_durable_sdk", "test_notifications_durable"): - register_workflow_package(_module, None) +for test_module in ("test_durable_sdk", "test_notifications_durable"): + register_workflow_package(test_module, None) class FakeRedis: From de902b5397437e42dede82416c4728959f88d7ca Mon Sep 17 00:00:00 2001 From: Paulo Date: Mon, 13 Jul 2026 18:51:08 +0200 Subject: [PATCH 10/12] Ownership registers after the name check; the provisional claim dies MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Entry modules never declare workflows — capability modules import at discover()/load(app), after the loader returns — so nothing needs the package registered before the entry loads. Registering the validated name with the authoritative Extension.package deletes the claim context manager, the derived-package guess, and the two-pass boot loop, and a failed load registers nothing by construction. Co-Authored-By: Claude Fable 5 --- backend/druks/extensions/loader.py | 42 +++++++++---------------- backend/druks/extensions/registry.py | 19 +---------- backend/tests/test_workflow_identity.py | 26 --------------- 3 files changed, 16 insertions(+), 71 deletions(-) diff --git a/backend/druks/extensions/loader.py b/backend/druks/extensions/loader.py index 746c62c9..1b9ea9f9 100644 --- a/backend/druks/extensions/loader.py +++ b/backend/druks/extensions/loader.py @@ -4,7 +4,7 @@ from druks.extensions import Extension from .exceptions import ExtensionImportError, ExtensionNotFound, MalformedExtension -from .registry import claimed_workflow_package, register_workflow_package +from .registry import register_workflow_package if TYPE_CHECKING: from importlib.metadata import EntryPoint @@ -14,33 +14,23 @@ _GROUP = "druks.extensions" -def _entry_package(entry: "EntryPoint") -> str: - # The ``/extension.py`` convention: the entry module's parent is - # the extension's package — the same default ``Extension.package`` takes. - return entry.module.rsplit(".", 1)[0] - - def iter_extensions() -> list[type[Extension]]: """Every installed extension, resolved from the ``druks.extensions`` entry points. Loading an entry point imports the extension's class. An extension that fails to import, resolves to a non-``Extension``, or collides on ``name`` raises — there is no per-extension fault tolerance yet (deferred until a real external extension exists).""" - entries = list(entry_points(group=_GROUP)) - # Ownership registers before any extension module imports, so every Workflow - # class — including one the first entry's import pulls in — resolves its - # declaring extension at definition time. - for entry in entries: - register_workflow_package(_entry_package(entry), entry.name) extensions: list[type[Extension]] = [] seen: set[str] = set() - for entry in entries: + for entry in entry_points(group=_GROUP): 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 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 @@ -92,19 +82,17 @@ def _resolve(name: str) -> type[Extension]: f"extension {name!r} is declared by {len(matches)} installed packages " f"({', '.join(e.value for e in matches)}) — uninstall all but one" ) - entry = matches[0] - with claimed_workflow_package(_entry_package(entry), name): - extension = _load_entry(entry) - # 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) + extension = _load_entry(matches[0]) + # 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/extensions/registry.py b/backend/druks/extensions/registry.py index 7e7462ab..5ca5b015 100644 --- a/backend/druks/extensions/registry.py +++ b/backend/druks/extensions/registry.py @@ -1,7 +1,6 @@ import importlib import pkgutil -from collections.abc import Callable, Iterator -from contextlib import contextmanager +from collections.abc import Callable from types import ModuleType from typing import Any @@ -102,22 +101,6 @@ def resolve_workflow_extension(module: str) -> str | None: raise LookupError(module) -@contextmanager -def claimed_workflow_package(package: str, extension: str | None) -> Iterator[None]: - """Provisionally claim ``package`` for ``extension`` while its entry point - loads: a fresh claim is released if the load fails, and an existing claim — - whoever holds it — is left untouched for the post-load checks to judge.""" - fresh = package not in _workflow_packages - if fresh: - register_workflow_package(package, extension) - try: - yield - except BaseException: - if fresh: - del _workflow_packages[package] - raise - - webhooks = Registry("webhooks", key=lambda cls: f"{cls.__module__}.{cls.__qualname__}") workflows = Registry("workflows", key=lambda cls: cls.kind) agents = Registry("agents", key=lambda agent: agent.id) diff --git a/backend/tests/test_workflow_identity.py b/backend/tests/test_workflow_identity.py index 49f68524..946aa94b 100644 --- a/backend/tests/test_workflow_identity.py +++ b/backend/tests/test_workflow_identity.py @@ -10,7 +10,6 @@ from druks.extensions import registry as extensions_registry from druks.extensions.exceptions import MalformedExtension from druks.extensions.registry import ( - claimed_workflow_package, register_workflow_package, resolve_workflow_extension, workflows, @@ -57,31 +56,6 @@ def test_resolution_matches_package_boundaries(): resolve_workflow_extension("alpha_pkg_sibling.workflows") -def test_failed_claim_releases_a_fresh_registration(): - # A claim made for a load that then fails must not survive it — otherwise a - # malformed first attempt pins the package under the wrong name and a - # corrected retry can never register it. - with pytest.raises(RuntimeError), claimed_workflow_package("gamma_pkg", "gamma"): - raise RuntimeError("load failed") - with pytest.raises(LookupError): - resolve_workflow_extension("gamma_pkg.workflows") - - -def test_failed_claim_leaves_an_existing_registration_untouched(): - # An aliased entry claiming an already-owned package must not evict the real - # owner — the post-load name check judges the alias, the map stays truthful. - register_workflow_package("alpha_pkg", "alpha") - with pytest.raises(RuntimeError), claimed_workflow_package("alpha_pkg", "imposter"): - raise RuntimeError("load failed") - assert resolve_workflow_extension("alpha_pkg.workflows") == "alpha" - - -def test_successful_claim_keeps_the_registration(): - with claimed_workflow_package("delta_pkg", "delta"): - pass - assert resolve_workflow_extension("delta_pkg.workflows") == "delta" - - 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"): From 9b237a795a3c094d95cee1b10b1a56fd6e5072f5 Mon Sep 17 00:00:00 2001 From: Paulo Date: Mon, 13 Jul 2026 20:33:36 +0200 Subject: [PATCH 11/12] Ownership lives with its validator; boot rejects aliased entries MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The package map is the loader-validated install claim, so it moves into the loader — registry.py returns to the capability catalogs it always was, and workflows.py asks the loader who owns a module. iter_extensions() gains the entry-key/name check _resolve() already had: production boot no longer accepts an aliased entry the app-less path rejects. Co-Authored-By: Claude Fable 5 --- backend/druks/extensions/loader.py | 43 ++++++++++++++++++- backend/druks/extensions/registry.py | 40 ----------------- backend/druks/workflows.py | 3 +- backend/tests/conftest.py | 3 +- backend/tests/test_extension_appless_load.py | 20 +++++++-- backend/tests/test_extension_doctor_checks.py | 8 ++-- backend/tests/test_proof_extension.py | 8 ++-- backend/tests/test_workflow_identity.py | 15 +++---- 8 files changed, 75 insertions(+), 65 deletions(-) diff --git a/backend/druks/extensions/loader.py b/backend/druks/extensions/loader.py index 1b9ea9f9..123b3f1a 100644 --- a/backend/druks/extensions/loader.py +++ b/backend/druks/extensions/loader.py @@ -4,7 +4,6 @@ from druks.extensions import Extension from .exceptions import ExtensionImportError, ExtensionNotFound, MalformedExtension -from .registry import register_workflow_package if TYPE_CHECKING: from importlib.metadata import EntryPoint @@ -13,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. @@ -26,6 +62,11 @@ 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) diff --git a/backend/druks/extensions/registry.py b/backend/druks/extensions/registry.py index 5ca5b015..9bef1087 100644 --- a/backend/druks/extensions/registry.py +++ b/backend/druks/extensions/registry.py @@ -4,8 +4,6 @@ from types import ModuleType from typing import Any -from .exceptions import MalformedExtension - # Leaf-module names that carry self-registering capabilities. ``autodiscover`` # imports exactly these (``routes`` defines routers the loader mounts; the rest # fire registration as an import side effect). The set is the single source of @@ -63,44 +61,6 @@ def autodiscover(package: str) -> list[ModuleType]: return modules -# Which extension owns each workflow-declaring package, stamped by the loader -# before it imports any extension module — 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, so this raises the loader's own taxonomy. - 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) - - webhooks = Registry("webhooks", key=lambda cls: f"{cls.__module__}.{cls.__qualname__}") workflows = Registry("workflows", key=lambda cls: cls.kind) agents = Registry("agents", key=lambda agent: agent.id) diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py index 5d6c3727..0214ed05 100644 --- a/backend/druks/workflows.py +++ b/backend/druks/workflows.py @@ -26,7 +26,8 @@ from druks.durable.models import AgentCall, Run from druks.durable.schemas import AgentCallResponse, SubjectActivity, SubjectSummary from druks.events.models import Event -from druks.extensions.registry import resolve_workflow_extension, workflows +from druks.extensions.loader import resolve_workflow_extension +from druks.extensions.registry import workflows from druks.extensions.settings import ( coerce_setting_value, validate_setting_override, diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 116defd1..cc77e56d 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -13,8 +13,7 @@ create_engine_from_url, init_db, ) -from druks.extensions.loader import iter_extensions -from druks.extensions.registry import register_workflow_package +from druks.extensions.loader import iter_extensions, register_workflow_package from druks.settings import Settings from sqlalchemy.orm import Session diff --git a/backend/tests/test_extension_appless_load.py b/backend/tests/test_extension_appless_load.py index 417c83cc..ce39e44b 100644 --- a/backend/tests/test_extension_appless_load.py +++ b/backend/tests/test_extension_appless_load.py @@ -99,7 +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 registry as extensions_registry + from druks.extensions import loader as extensions_loader from druks.extensions.registry import agents, webhooks, workflows from druks.models import Base @@ -109,7 +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_registry._workflow_packages) + packages = dict(extensions_loader._workflow_packages) finished = signal("run.finished") receivers = dict(finished.receivers) try: @@ -120,8 +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_registry._workflow_packages.clear() - extensions_registry._workflow_packages.update(packages) + 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] @@ -256,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 58974601..49daceeb 100644 --- a/backend/tests/test_extension_doctor_checks.py +++ b/backend/tests/test_extension_doctor_checks.py @@ -31,7 +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 registry as extensions_registry + from druks.extensions import loader as extensions_loader from druks.extensions.registry import agents, webhooks, workflows from druks.models import Base @@ -39,7 +39,7 @@ def external_package(): tables = set(Base.metadata.tables) registries = {registry: dict(registry._items) for registry in (agents, webhooks, workflows)} - packages = dict(extensions_registry._workflow_packages) + packages = dict(extensions_loader._workflow_packages) finished = signal("run.finished") receivers = dict(finished.receivers) try: @@ -50,8 +50,8 @@ def external_package(): Base.metadata.remove(Base.metadata.tables[name]) for registry, snapshot in registries.items(): registry._items = snapshot - extensions_registry._workflow_packages.clear() - extensions_registry._workflow_packages.update(packages) + 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_proof_extension.py b/backend/tests/test_proof_extension.py index 54438c0b..1d857671 100644 --- a/backend/tests/test_proof_extension.py +++ b/backend/tests/test_proof_extension.py @@ -49,7 +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 registry as extensions_registry + from druks.extensions import loader as extensions_loader from druks.extensions.registry import agents, webhooks, workflows from druks.models import Base @@ -57,7 +57,7 @@ def external_package(): tables = set(Base.metadata.tables) registries = {registry: dict(registry._items) for registry in (agents, webhooks, workflows)} - packages = dict(extensions_registry._workflow_packages) + packages = dict(extensions_loader._workflow_packages) finished = signal("run.finished") receivers = dict(finished.receivers) try: @@ -68,8 +68,8 @@ def external_package(): Base.metadata.remove(Base.metadata.tables[name]) for registry, snapshot in registries.items(): registry._items = snapshot - extensions_registry._workflow_packages.clear() - extensions_registry._workflow_packages.update(packages) + 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_workflow_identity.py b/backend/tests/test_workflow_identity.py index 946aa94b..4d7c5dd6 100644 --- a/backend/tests/test_workflow_identity.py +++ b/backend/tests/test_workflow_identity.py @@ -7,13 +7,10 @@ from druks.durable.models import Run from druks.durable.schemas import get_display_label from druks.events.models import Event -from druks.extensions import registry as extensions_registry +from druks.extensions import loader as extensions_loader from druks.extensions.exceptions import MalformedExtension -from druks.extensions.registry import ( - register_workflow_package, - resolve_workflow_extension, - workflows, -) +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 @@ -22,11 +19,11 @@ 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_registry._workflow_packages) + packages = dict(extensions_loader._workflow_packages) items = dict(workflows._items) yield - extensions_registry._workflow_packages.clear() - extensions_registry._workflow_packages.update(packages) + extensions_loader._workflow_packages.clear() + extensions_loader._workflow_packages.update(packages) workflows._items = items From d35b499c5f9cb5a4455bbc07801a225192e3901c Mon Sep 17 00:00:00 2001 From: Paulo Date: Mon, 13 Jul 2026 20:39:59 +0200 Subject: [PATCH 12/12] CI fixes: the last extension= stragglers and the dup-name test premise The generic-subjects seed and the scope-dispatch assertion still carried the per-run extension; the duplicate-name boot test now aliases both entries to one key, since per-entry name validation fires before duplication is observable. Co-Authored-By: Claude Fable 5 --- backend/tests/test_extensions.py | 2 +- backend/tests/test_generic_subjects.py | 8 +------- backend/tests/test_scope_reply_rescope.py | 1 - 3 files changed, 2 insertions(+), 9 deletions(-) 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 1b366bf5..b490d785 100644 --- a/backend/tests/test_generic_subjects.py +++ b/backend/tests/test_generic_subjects.py @@ -51,13 +51,7 @@ def _seed_run( ) session.add(run) session.flush() - seed_dbos_status( - session, - run.id, - state, - subject={"type": "thing", "id": subject_id}, - extension="faketest", - ) + seed_dbos_status(session, run.id, state, subject={"type": "thing", "id": subject_id}) return run diff --git a/backend/tests/test_scope_reply_rescope.py b/backend/tests/test_scope_reply_rescope.py index 1e460583..2ac68ba6 100644 --- a/backend/tests/test_scope_reply_rescope.py +++ b/backend/tests/test_scope_reply_rescope.py @@ -101,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", }