The ORM is async end to end

backend/druks/accounts/dependencies.py

@@ -20,11 +20,11 @@
 _bearer_scheme = HTTPBearer(auto_error=False, scheme_name="personalAccessToken")
 
 
-def resolve_pat_account(credentials: HTTPAuthorizationCredentials | None) -> Account:
+async def resolve_pat_account(credentials: HTTPAuthorizationCredentials | None) -> Account:
     """A present Authorization must authenticate — never a fall-through."""
     if credentials:
         try:
-            return PersonalAccessToken.authenticate(credentials.credentials).account
+            return (await PersonalAccessToken.authenticate(credentials.credentials)).account
         except InvalidPatError as error:
             raise HTTPException(
                 status_code=401,
@@ -38,10 +38,10 @@ def resolve_pat_account(credentials: HTTPAuthorizationCredentials | None) -> Acc
     )
 
 
-def resolve_single_operator() -> Account | None:
+async def resolve_single_operator() -> Account | None:
     """None while zero accounts exist (setup); more than one refuses rather
     than guesses."""
-    operators = Account.list_non_system()
+    operators = await Account.list_non_system()
     if len(operators) > 1:
         raise AuthConfigurationError(
             f"auth mode 'none' expects exactly one operator account, found "
@@ -56,16 +56,16 @@ async def _resolve_operator(connection: HTTPConnection) -> Account | None:
     connection, not a request, so a WebSocket upgrade resolves the same way."""
     settings = connection.app.state.settings
     if settings.identity.mode == "none":
-        return resolve_single_operator()
+        return await resolve_single_operator()
     values = connection.headers.getlist(settings.identity.header)
     if len(values) == 1 and (asserted := values[0].strip()):
         if settings.identity.mode == "header":
-            return Account.get_or_create(asserted)
+            return await Account.get_or_create(asserted)
         try:
             email = await verify_assertion(asserted, settings)
         except InvalidAssertionError as error:
             raise HTTPException(status_code=401, detail=str(error)) from error
-        return Account.get_or_create(email)
+        return await Account.get_or_create(email)
     raise HTTPException(
         status_code=401,
         detail=f"The edge must assert exactly one nonblank {settings.identity.header} identity.",
@@ -102,7 +102,7 @@ async def current_account(
     """The Bearer PAT when Authorization is present — present-but-empty still
     challenges — else the session identity."""
     if "Authorization" in request.headers:
-        account = resolve_pat_account(bearer)
+        account = await resolve_pat_account(bearer)
     else:
         account = await _resolve_operator(request)
         if not account:
@@ -154,7 +154,7 @@ async def current_account_or_setup(
     """PAT-first identity that reads none/zero setup as None instead of
     refusing — ``/api/auth/me`` only."""
     if "Authorization" in request.headers:
-        account = resolve_pat_account(bearer)
+        account = await resolve_pat_account(bearer)
     else:
         account = await _resolve_operator(request)
     token = current_account_id.set(account.id if account else None)

backend/druks/accounts/models.py

@@ -38,35 +38,35 @@ class Account(Base, Uuid7Pk):
     created_at: Mapped[datetime] = mapped_column(default=Base.utc_now)
 
     @classmethod
-    def get(cls, account_id: str, *, exclude_system: bool = False) -> "Account | None":
+    async def get(cls, account_id: str, *, exclude_system: bool = False) -> "Account | None":
         if exclude_system and account_id == SYSTEM_ACCOUNT_ID:
             return
-        return db_session().get(cls, account_id)
+        return await db_session().get(cls, account_id)
 
     @classmethod
-    def get_for_username(cls, username: str) -> "Account | None":
-        return db_session().scalar(select(cls).where(cls.username == username))
+    async def get_for_username(cls, username: str) -> "Account | None":
+        return await db_session().scalar(select(cls).where(cls.username == username))
 
     @classmethod
-    def get_or_create(cls, username: str) -> "Account":
+    async def get_or_create(cls, username: str) -> "Account":
         """Concurrency-safe lookup-or-create: racing requests both INSERT with
         ON CONFLICT DO NOTHING, then converge on the one row through the
         canonical CITEXT lookup."""
-        account = cls.get_for_username(username)
+        account = await cls.get_for_username(username)
         if account:
             return account
         session = db_session()
-        session.execute(
+        await session.execute(
             insert(cls)
             .values(username=username)
             .on_conflict_do_nothing(index_elements=["username"])
         )
-        return session.scalars(select(cls).where(cls.username == username)).one()
+        return (await session.scalars(select(cls).where(cls.username == username))).one()
 
     @classmethod
-    def list_non_system(cls) -> list["Account"]:
+    async def list_non_system(cls) -> list["Account"]:
         stmt = select(cls).where(cls.username != SYSTEM_ACCOUNT_ID).order_by(cls.created_at)
-        return list(db_session().scalars(stmt))
+        return list(await db_session().scalars(stmt))
 
 
 def _hash_token(token: str) -> bytes:
@@ -108,24 +108,24 @@ def status(self) -> str:
         return "active"
 
     @classmethod
-    def get(cls, pat_id: str) -> "PersonalAccessToken | None":
-        return db_session().get(cls, pat_id)
+    async def get(cls, pat_id: str) -> "PersonalAccessToken | None":
+        return await db_session().get(cls, pat_id)
 
     @classmethod
-    def get_for_prefix(cls, prefix: str) -> "PersonalAccessToken | None":
-        return db_session().scalar(select(cls).where(cls.token_prefix == prefix))
+    async def get_for_prefix(cls, prefix: str) -> "PersonalAccessToken | None":
+        return await db_session().scalar(select(cls).where(cls.token_prefix == prefix))
 
     @classmethod
-    def list_for_account(cls, account_id: str) -> list["PersonalAccessToken"]:
+    async def list_for_account(cls, account_id: str) -> list["PersonalAccessToken"]:
         stmt = select(cls).where(cls.account_id == account_id).order_by(cls.created_at.desc())
-        return list(db_session().scalars(stmt))
+        return list(await db_session().scalars(stmt))
 
     @classmethod
-    def create(cls, *, account_id: str, name: str) -> "tuple[PersonalAccessToken, str]":
+    async def create(cls, *, account_id: str, name: str) -> "tuple[PersonalAccessToken, str]":
         """Mint ``account_id`` a token; returns (row, plaintext). The plaintext
         is shown exactly once — only its hash lands in the row."""
         prefix = _new_prefix()
-        while cls.get_for_prefix(prefix):
+        while await cls.get_for_prefix(prefix):
             prefix = _new_prefix()
         secret = base64.urlsafe_b64encode(secrets.token_bytes(PAT_SECRET_BYTES))
         token = f"{PAT_TOKEN_TAG}_{prefix}_{secret.rstrip(b'=').decode()}"
@@ -141,16 +141,16 @@ def create(cls, *, account_id: str, name: str) -> "tuple[PersonalAccessToken, st
         )
         session = db_session()
         session.add(row)
-        session.flush()
+        await session.flush()
         return row, token
 
     @classmethod
-    def authenticate(cls, credential: str) -> "PersonalAccessToken":
+    async def authenticate(cls, credential: str) -> "PersonalAccessToken":
         """Resolve a presented bearer credential to its live row — the one
         authentication door for both HTTP and MCP — or raise InvalidPatError.
         Stamps last_used_at, at most hourly."""
         prefix, _, _ = credential.removeprefix(f"{PAT_TOKEN_TAG}_").partition("_")
-        row = cls.get_for_prefix(prefix)
+        row = await cls.get_for_prefix(prefix)
         if not row:
             raise InvalidPatError("Not a recognized personal access token.")
         if not hmac.compare_digest(_hash_token(credential), row.token_hash):
@@ -162,10 +162,10 @@ def authenticate(cls, credential: str) -> "PersonalAccessToken":
         now = Base.utc_now()
         if not row.last_used_at or now - row.last_used_at >= PAT_LAST_USED_RESOLUTION:
             row.last_used_at = now
-            db_session().flush()
+            await db_session().flush()
         return row
 
-    def revoke(self) -> None:
+    async def revoke(self) -> None:
         # Keep the first revocation instant — a repeat revoke changes nothing.
         self.revoked_at = self.revoked_at or Base.utc_now()
-        db_session().flush()
+        await db_session().flush()

backend/druks/accounts/routes.py

@@ -18,15 +18,15 @@ async def get_identity(
         account=AccountResponse.model_validate(account) if account else None,
         # An account needs onboarding exactly while it has no harness
         # connection; none/zero is onboarding before the account exists.
-        onboarding_required=not (account and HarnessConnection.list_for_account(account.id)),
+        onboarding_required=not (account and await HarnessConnection.list_for_account(account.id)),
     )
 
 
 @router.get("/personal-tokens", response_model=list[PatResponse], response_model_by_alias=True)
 async def list_pats(
     account: Account = Depends(current_session_account),
 ) -> list[PersonalAccessToken]:
-    return PersonalAccessToken.list_for_account(account.id)
+    return await PersonalAccessToken.list_for_account(account.id)
 
 
 @router.post("/personal-tokens")
@@ -38,7 +38,7 @@ async def create_pat(
     if name and len(name) <= PAT_NAME_LENGTH:
         # The plaintext, handed back exactly once — only its hash is stored,
         # and the new row surfaces through the list.
-        _, token = PersonalAccessToken.create(account_id=account.id, name=name)
+        _, token = await PersonalAccessToken.create(account_id=account.id, name=name)
         return {"token": token}
     raise HTTPException(
         status_code=422,
@@ -52,9 +52,9 @@ async def create_pat(
 async def revoke_pat(
     pat_id: str, account: Account = Depends(current_session_account)
 ) -> PersonalAccessToken:
-    pat = PersonalAccessToken.get(pat_id)
+    pat = await PersonalAccessToken.get(pat_id)
     if pat and pat.account_id == account.id:
-        pat.revoke()
+        await pat.revoke()
         return pat
     # One shape for missing and foreign — existence stays account-scoped.
     raise HTTPException(status_code=404, detail="No such token.")

backend/druks/agents.py

@@ -123,16 +123,16 @@ def __set_name__(self, owner: type, attr: str) -> None:
     # override → the agent's declared value → the operator's global default.
     # ``run`` uses these; callers that drive the harness themselves call them
     # directly.
-    def get_model_name(self) -> str:
-        return SettingsOverride.agent_model(self.id, self.model).value
+    async def get_model_name(self) -> str:
+        return (await SettingsOverride.agent_model(self.id, self.model)).value
 
-    def get_effort(self) -> str:
-        harness = get_harness_for_model(self.get_model_name()).name
-        return SettingsOverride.agent_effort(self.id, self.effort, harness).value
+    async def get_effort(self) -> str:
+        harness = (await get_harness_for_model(await self.get_model_name())).name
+        return (await SettingsOverride.agent_effort(self.id, self.effort, harness)).value
 
-    def get_timeout(self) -> int:
-        harness = get_harness_for_model(self.get_model_name()).name
-        resolved = SettingsOverride.agent_timeout(self.id, self.timeout, harness).value
+    async def get_timeout(self) -> int:
+        harness = (await get_harness_for_model(await self.get_model_name())).name
+        resolved = (await SettingsOverride.agent_timeout(self.id, self.timeout, harness)).value
         # Capped so a single call always fits inside a fresh sandbox lease.
         return min(resolved, MAX_AGENT_TIMEOUT_SECONDS)
 
@@ -187,11 +187,11 @@ async def _quota_retry_wait() -> float:
             # Runs as its own step: the body does no IO, and replay reuses the
             # recorded wait instead of re-reading the scrape.
             async with step_session():
-                harness = get_harness_for_model(self.get_model_name())
+                harness = await get_harness_for_model(await self.get_model_name())
                 # The scrape belongs to the charged connection — its account
                 # differs from the run's on fallback.
-                connection = HarnessConnection.lookup(harness.name, workflow.account_id)
-                scrape = UsageScrape.latest_for(harness.name, connection.account_id)
+                connection = await HarnessConnection.lookup(harness.name, workflow.account_id)
+                scrape = await UsageScrape.latest_for(harness.name, connection.account_id)
                 if scrape:
                     now = datetime.now(UTC)
                     reset = scrape.soonest_reset_after(now)
@@ -253,18 +253,18 @@ async def _run(self, *, workflow_id: str, **context: Any) -> Any:
         the harness. ``__call__`` handles the durable wrapping + nesting."""
         if not self.prompt:
             raise WorkflowError(f"agent {self.id!r} has no prompt template to render")
-        model = self.get_model_name()
-        harness = get_harness_for_model(model)
+        model = await self.get_model_name()
+        harness = await get_harness_for_model(model)
         workflow = current_workflow.get()
         # Refusing an unservable call here beats provisioning a VM and
         # 401ing mid-run.
-        connection = HarnessConnection.lookup(harness.name, workflow.account_id)
+        connection = await HarnessConnection.lookup(harness.name, workflow.account_id)
         # Plain snapshots: the commits below expire the ORM row mid-flight.
         connection_id, charged_account_id = connection.id, connection.account_id
         # An agent call is a durability boundary — its effects don't roll back —
         # so commit here rather than hold the step's connection idle through the
         # minutes of provisioning and the run.
-        db_session().commit()
+        await db_session().commit()
         settings = load_settings()
         artifact_dir = settings.artifacts_dir / f"run-{workflow_id}"
 
@@ -290,7 +290,7 @@ async def _run(self, *, workflow_id: str, **context: Any) -> Any:
                 prompt_context.setdefault("workspace", runner)
                 prompt = await render_prompt(self.prompt, **prompt_context)
                 await set_run_phase("agent_running")
-                AgentCall.start(
+                await AgentCall.start(
                     engine,
                     call_id=call_id,
                     run_id=workflow_id,
@@ -310,16 +310,16 @@ async def _run(self, *, workflow_id: str, **context: Any) -> Any:
                         account_id=workflow.account_id,
                     )
                 except BaseException as error:
-                    AgentCall.fail(engine, call_id=call_id, error=error)
+                    await AgentCall.fail(engine, call_id=call_id, error=error)
                     raise
-                AgentCall.finish(engine, call_id=call_id, result=result)
+                await AgentCall.finish(engine, call_id=call_id, result=result)
 
         if result.error:
             raise result.error
 
         output = self.contract.model_validate(result.output)
         if spec := output.get_artifact():
-            Artifact.record(call_dir=artifact_dir / call_id, call_id=call_id, **spec)
+            await Artifact.record(call_dir=artifact_dir / call_id, call_id=call_id, **spec)
         return output.to_result()
 
     async def _execute(
@@ -340,8 +340,8 @@ async def _execute(
             prompt=prompt,
             schema=schema,
             agent=self.id,
-            effort=self.get_effort(),
-            timeout=self.get_timeout(),
+            effort=await self.get_effort(),
+            timeout=await self.get_timeout(),
             artifact_dir=artifact_dir,
             call_id=call_id,
             include_plugins=self.include_plugins,

backend/druks/api/artifacts.py

@@ -11,10 +11,10 @@
 async def get_artifact(artifact_id: str) -> ArtifactContent:
     # A call's renderable output, reached through the call that produced it — the in-app
     # review fetches this to render the plan beside its controls.
-    artifact = db_session().get(Artifact, artifact_id)
+    artifact = await db_session().get(Artifact, artifact_id)
     if not artifact:
         raise HTTPException(status.HTTP_404_NOT_FOUND, "artifact not found")
-    call = AgentCall.get(artifact.agent_call_id)
+    call = await AgentCall.get(artifact.agent_call_id)
     path = call.get_file_path(artifact.path)
     if not path:
         raise HTTPException(status.HTTP_404_NOT_FOUND, "artifact content missing")

backend/druks/api/health_status.py

@@ -7,17 +7,19 @@
 from druks.webhooks.deliveries import last_delivery_at
 
 
-def _spend_for_local_today(*, timezone_name: str, now: datetime) -> tuple[float, int]:
+async def _spend_for_local_today(*, timezone_name: str, now: datetime) -> tuple[float, int]:
     _, local_start = operator_local_day(timezone_name, now)
-    return AgentCall.total_run_spend_between(
+    return await AgentCall.total_run_spend_between(
         start=local_start.astimezone(UTC),
         end=(local_start + timedelta(days=1)).astimezone(UTC),
     )
 
 
 async def build_health() -> api_schemas.DashboardHealth:
     now = datetime.now(UTC)
-    spend, tokens = _spend_for_local_today(timezone_name=UserSettings.get().timezone, now=now)
+    spend, tokens = await _spend_for_local_today(
+        timezone_name=(await UserSettings.get()).timezone, now=now
+    )
     # GitHub is the code host and is always present.
     sources = ["github"]
     return api_schemas.DashboardHealth(

backend/druks/api/runs.py

@@ -22,7 +22,7 @@
 async def resume_run(run_id: Annotated[str, Path(alias="run")], body: ResumeRequest) -> None:
     # The in-app half of a gate: the operator answers the parked run from Druks
     # (external gates resume through their own webhook).
-    run = Run.get(run_id)
+    run = await Run.get(run_id)
     if not run:
         raise HTTPException(status.HTTP_404_NOT_FOUND, "run not found")
     ask = run.input_request
@@ -65,7 +65,7 @@ async def cancel_run(
 ) -> CancelRunResponse:
     """Cancel an active run, recording the reason as its failure; a repeat
     cancel reports already_cancelled."""
-    run = Run.get(run_id)
+    run = await Run.get(run_id)
     if not run:
         raise RunNotFound(run_id)
     if run.state == RunState.CANCELLED.value:
@@ -94,15 +94,15 @@ async def retry_run(
 ) -> RetryRunResponse:
     """Rerun a failed run from the step that killed it, reusing every
     completed step."""
-    run = Run.get(run_id)
+    run = await Run.get(run_id)
     if not run:
         raise RunNotFound(run_id)
     if run.state != RunState.FAILED.value:
         raise RunNotFailed(run_id)
 
-    subject = run.subject
+    subject = await run.get_subject()
     if subject:
-        latest = Run.get_latest_for_subject(subject["type"], subject["id"])
+        latest = await Run.get_latest_for_subject(subject["type"], subject["id"])
         if latest and latest.is_active:
             raise SubjectBusy(latest.id)