> ## Documentation Index
> Fetch the complete documentation index at: https://druks-codex-dru-85-codex-device-login.mintlify.site/llms.txt
> Use this file to discover all available pages before exploring further.

# Writing an app

> Build a separately packaged Druks app with workflows, agents, gates, routes, models, and migrations.

An app is a Python distribution installed in Druks. It owns domain behavior.
Druks supplies durable execution and shared operating services. Read
[the app boundary](concepts.md#the-app-boundary)
before you assign ownership of a capability.

## Publish useful activity

An agent output declares its saved result and its Activity topic:

```python theme={null}
from druks.agents import AgentOutput


class GistOutput(AgentOutput):
    gist: str

    def to_artifact(self) -> dict[str, str]:
        return {"kind": "markdown", "title": "Gist", "content": self.gist}

    def to_event(self) -> dict[str, str]:
        return {"topic": "gist.prepared", "summary": self.gist}
```

A workflow asks the operator to review the result. It announces the outcome that
it owns:

```python theme={null}
from druks.workflows import Workflow, step

from druks_field_notes.app import FieldNotes
from druks_field_notes.models import Note


class Summarize(Workflow):
    subject = Note

    async def run_multistep(self) -> None:
        note = await self.subject
        operator_note = ""
        while True:
            result = await FieldNotes.summarize(note_body=note.body, operator_note=operator_note)
            reply = await self.review()
            if reply.action == "approve":
                break
            operator_note = reply.note
        await self.save_gist(result.gist)
        await self.announce("note.gist_approved")

    @step
    async def save_gist(self, gist: str) -> None:
        note = await self.subject
        await note.save_gist(gist)
```

The independently installed Field Notes proof app uses these classes. The
operator answers the review from the dashboard. Druks records the accepted work,
each saved result, each request, and each validated reply. It also records a
terminal failure. Authors supply no routing IDs, timestamps, sessions, or event
rows. See [Activity facts](#activity-facts-and-signals) for the ownership rules.

## Scaffold and prove the package

```bash theme={null}
uvx --from druks druks create app night-watch
cd druks-night-watch
uv sync
uv run pytest
```

From a Druks checkout, `uv run druks create app night-watch` scaffolds with
that checkout's CLI instead.

The command writes a standalone `druks-night-watch` project in the current
directory. The folder and the distribution use hyphens. The app name and its
package use underscores. Its `pyproject.toml` contains:

```toml theme={null}
[project.entry-points."druks.apps"]
night_watch = "druks_night_watch.app:NightWatch"
```

The app name must match `[a-z][a-z0-9_]*`. The command also accepts the
hyphenated spelling and converts it. The name becomes the API namespace, table
prefix, migration version-table suffix, and settings namespace. Installing the
distribution is the registration:

```bash theme={null}
uv pip install -e /path/to/druks-night-watch
```

At boot Druks imports installed entry points and fails loudly on duplicate
names, a mismatched entry-point key, malformed target, import failure, or
unprefixed table.

The project root also carries an `AGENTS.md` holding the contracts a coding agent
cannot infer from the stubs, and a link back to this guide.

The scaffold depends on the published `druks`. To develop an app against a
local checkout instead, pin it:

```toml theme={null}
[tool.uv.sources]
druks = { path = "../druks", editable = true }
```

## Package layout

The scaffold separates self-registering capability modules from ordinary
package modules:

| Path | Contract |
| - | - |
| `app.py` | `App` subclass, agents, app settings |
| `workflows.py` | durable `Workflow` and `Gate` subclasses |
| `tasks.py` | Optional `@task` background functions |
| `models.py` | SQLAlchemy models with `<name>_` table names, `StoredSubject` among them |
| `contracts.py` | `AgentOutput` contracts |
| `schemas.py` | HTTP responses and subject summaries |
| `routes.py` | FastAPI routers |
| `pages.py` | `@page` declarations that return `Page` objects |
| `subscribers.py` | signal reactions |
| `webhooks.py` | Optional authenticated provider deliveries |
| `services.py` | Optional `Service` declarations for appliance credentials at external providers |
| `migrations/versions/` | this distribution's Alembic history |
| `dist/` | optional built frontend module, mounted inside the shell (served under `/app/<name>`) |

Druks recursively discovers leaf modules named `workflows`, `tasks`, `routes`,
`pages`, `subscribers`, `webhooks`, `services`, and `channel`. A capability
hidden in `workflow.py` is not discovered. Ordinary names such as `policy.py` and `workspace.py` have no import
side effect unless a discovered module imports them.

## Declare the app

```python theme={null}
from druks.apps import App


class NightWatch(App):
    name = "night_watch"
    icon = "telescope"
    description = "Checks repositories after hours."
```

The class is a stateless installation singleton. Do not instantiate it. Druks mounts
every router found in its `routes` modules under `/api/night_watch`, supplies
transcript routes, and serves `druks_night_watch/dist/` under
`/app/night_watch` when it contains `entry.js`.

## Choose the right workflow shape

The parameters of `run()` or `run_multistep()` are the workflow input. Druks
builds a Pydantic model from their annotations and validates the call to
`start()`.

If the whole body is one durable operation, use `run()`:

```python theme={null}
from druks.workflows import Workflow


class RecordHeartbeat(Workflow):
    async def run(self, source: str) -> None:
        Heartbeat.record(source)
```

If completed operations require independent recovery, use `run_multistep()`.
Also use it for a workflow that waits on a gate:

```python theme={null}
from druks.workflows import Workflow, step


class Sweep(Workflow):
    async def run_multistep(self, repo: str) -> None:
        findings = await self.scan(repo)
        await NightWatch.report(repo=repo, findings=findings)

    @step
    async def scan(self, repo: str) -> list[str]:
        return await scanner.scan(repo)
```

Druks treats `run()` as one step, so it must not carry `@step`.
DBOS replays `run_multistep()` orchestration, so it must not carry `@step`.
Decorate its side-effecting operations instead. An agent called directly from
the orchestration body gets its own step. An agent called inside `@step` or
`run()` shares that enclosing checkpoint.

### Declare the sandbox environment

A workflow can ship the environment its agents need as a plain shell file:

```python theme={null}
from druks.sandbox import Sandbox
from druks.workflows import Workflow


class BuildSite(Workflow):
    sandbox = Sandbox(setup="sandboxes/build.sh")
```

Place the file at `site_builder/sandboxes/build.sh`. The path is relative to the
app package.

Druks reads the raw bytes. It does not render
the file or run it during import. Drukbox builds a reusable template from the
platform base and the script.
A run waits with a visible sandbox-building phase when that template is still
building. A workflow with no declaration uses the platform base unchanged.

The content hash of the base and script identifies the template. App authors do
not name provider images.

Use provider idempotency keys for writes. An interrupted operation can run again.
DBOS reuses completed checkpoints on recovery. Keep decisions in replayable
control flow. Keep I/O inside steps. See
[durability and recovery](concepts.md#durability-and-recovery).

Start a workflow with an explicit subject — an instance of the class it declares:

```python theme={null}
run_id = await Sweep.start(
    subject=repository,
    repo=full_name,
)
```

A workflow without a subject declaration passes `subject=None`. A subject has
at most one active run for each workflow kind. A duplicate start returns the
active run ID. Attribution does not change this rule. Two accounts that start
the same subject share one run.

If the app requires prelaunch policy, wrap
`start()` in a domain `dispatch()` method. This method can own lookup, snapshot,
or routing policy. It binds its own database session, so a workflow body can
call it.

A browser start attributes itself. The request identity gate records the
resolved account, and `start()` inherits it. A route does not require more
attribution code. If the dispatcher has a better account, pass `account_id`.
For example, a webhook can resolve the ticket assignee.

`Run.account_id` is required. A browser start records the authenticated account.
An unattended start records the default account. Druks refuses to start a run
before an account is available. A parked run keeps its account after resume.

Each agent call uses the installation execution defaults. Agent overrides take
priority. Subscription billing uses the run account's subscription.
The call records exactly one billing reference: `subscription_id` or
`api_key_provider`. Druks uses that selected credential for execution. Missing
credentials refuse the call. A workflow can use different providers across its
agent calls. Disconnect clears the credential secret and retains its billing
identity for call history.
See [personal and installation settings](configuration.md#personal-and-installation-settings)
for execution defaults, personal preferences, and timezone rules.

### The journal

Druks keeps a journal of the typed values for each run. Each body-level agent
call and gate reply enters it in call order. Add your values with
`self.journal.add()`. Read them by contract type:

```python theme={null}
self.journal.filter(PlanData)                                # all entries, oldest first
self.journal.latest(PlanData)                                # newest, or None
self.journal.filter(ImplementationOutput, status="success")  # keyword filters: ANDed equality
self.journal.filter(ReviewWork)                              # gate replies, by their Gate class
```

Subclass `Journal` to name your projections, and declare it on the workflow:

```python theme={null}
class SweepJournal(Journal):
    @property
    def findings(self) -> list[FindingData]:
        return self.filter(FindingData)


class Sweep(Workflow):
    journal_class = SweepJournal
```

The journal survives crashes without separate storage. Recovery runs the body again
with every durable call memoized, so the same entries land in the same order.

Two rules:

* Druks journals only body-level calls. An agent call inside a `@step` — or in
  a `run()` body, which is one big step — never enters it. Keep that state
  in local variables.
* Never mutate body-held state inside a `@step`. DBOS skips a completed step
  on replay, so the write disappears.

### Announcing domain events

Announce a domain fact from the workflow body:

```python theme={null}
await self.announce(
    "pr.opened",
    repo=item.repo,
    pr_number=delivery.pr_number,
    branch=delivery.branch,
)
```

Druks records the event in one checkpoint. It notifies subscribers in a second
checkpoint. A subscriber retry cannot insert the completed event again. Recovery
reuses completed checkpoints. An interrupted operation can run again, so
subscribers must remain idempotent. Call this method outside a `@step`.

A domain method announces through its subject:

```python theme={null}
from druks.db import StoredSubject
from sqlalchemy.orm import Mapped


class Report(StoredSubject):
    published_url: Mapped[str | None]

    async def publish(self, url: str) -> None:
        if self.published_url != url:
            self.published_url = url
            await self.announce("report.published", url=url)
```

`Subject` and `StoredSubject` both supply `announce()`. Druks gets the owner from
the registered app package. The call records the subject identity, its current
key, its optional summary title, and the supplied facts. It then notifies subscribers in the same
transaction as the domain change. A rollback removes the change and its event.
The app must prevent duplicate domain changes on webhook redelivery.

Authors supply no app ID, run ID, timestamp, or session. Frontend code owns the
wording.

### Schedules and settings

Set `every` to declare a cron:

```python theme={null}
class Sweep(Workflow):
    every = "0 6 * * *"
```

The tick fires the workflow's body with no subject and no input, so every body
parameter needs a default. A workflow whose runs are *about* something (it
declares a `subject`) must not start that way. Give it a `dispatch()` classmethod
and the schedule fires that instead — it resolves the subject and starts the
real run:

```python theme={null}
class Engage(Workflow):
    subject = Account
    every = "0 */4 * * *"

    @classmethod
    async def dispatch(cls) -> str:
        return await cls.start(subject=Account.get())

    async def run(self) -> None:
        ...
```

A scheduled `dispatch()` fires with no arguments, so it must be nullary. Druks
evaluates cron expressions in the installation timezone. An operator can change
the cadence or pause a declared schedule on the Schedules page or in the app
settings. Druks cannot add a schedule to a workflow that declares none.

### Background tasks

A `Workflow` is the right home for work you want on a subject's timeline — a run
with agent calls, gates, and operator-tunable settings. Plumbing that wants none
of that — periodic maintenance, a fire-and-forget side effect — is a `task`:

```python theme={null}
from druks.workflows import task


@task(every="*/15 * * * *")
async def refresh_tokens() -> None:
    ...


@task(retries=4)
async def sync_labels(pull_request_id: int) -> None:
    ...
```

Call `await sync_labels.enqueue(pull_request_id=7)` from a route, subscriber, or
workflow body. Never call it inside a `@step`. Like a workflow, the signature is
the wire contract. Parameters are
annotated. `enqueue()` validates them and stores JSON. A task keeps no run row
and never reaches the timeline.

It has no subject, gate, or operator settings.
It cannot make agent calls. `every=` on a task is a fixed UTC cadence that the
code owns. An operator can change only a workflow schedule, on the Schedules
page or in the app settings.
`retries=` sets retries after the first attempt, both here and on `@step`.

A workflow can declare its own operator settings:

```python theme={null}
from pydantic import BaseModel, Field


class Sweep(Workflow):
    class Settings(BaseModel):
        batch_size: int = Field(default=20, ge=1, le=100)

    @step
    async def load_settings(self) -> "Sweep.Settings":
        return await self.settings()
```

Reading settings inside a step snapshots them for replay. Reading them directly
from replayed orchestration allows later edits to change an in-flight run.

## Add an agent

An agent belongs to the app class. Which CLI runs it, which model, which
login it bills, and at what effort are the operator's choices: defaults in
**Settings → Agents**, overridden per agent on the app's own page. `timeout`
is the one declarable knob, because how long a step may take is a fact about
the task.

```python theme={null}
from druks.agents import Agent, AgentOutput


class ReportOutput(AgentOutput):
    title: str
    body: str


class NightWatch(App):
    name = "night_watch"

    report = Agent(
        prompt="night_watch/report.md",
        contract=ReportOutput,
        description="Turns findings into an operator report.",
    )
```

The app name and the attribute name form the agent's id: `night_watch.report`.
Settings overrides, the timeline, and the step name use that id.

An agent that reads untrusted content, such as email or a web page, gets no
plugin state and no MCP server:

```python theme={null}
    triage = Agent(
        prompt="triage.md",
        contract=TriageOutput,
        include_plugins=False,
        include_mcp=False,
    )
```

A workflow that sets `steps_reuse_sandbox = True` keeps one sandbox for all its
agents, and the sandbox takes its entries from the agent call that creates it.
Give such a workflow agents that all include MCP servers, or none that do.

Call it only inside a workflow:

```python theme={null}
result = await NightWatch.report(repo=repo, findings=findings)
```

Druks renders the prompt with the current workflow, workspace, and supplied
context. It resolves the agent's harness, model, and login in one place,
refuses before any sandbox work when the login is missing, then provisions or
attaches a sandbox, executes the CLI, validates the structured output, and
records the call. Override
`AgentOutput.to_result()` to map the strict agent contract to a domain value.
Override `to_artifact()` to publish a reviewable artifact. Override `to_event()`
to record that saved result in Activity: return a `topic` and an optional
`summary`. An output that returns an event must also return an artifact.
Otherwise the agent call fails with `WorkflowError`.

Druks records one Activity row with the artifact and its producing agent call.
The row and the artifact share a transaction. Recovery reuses a completed call
and does not create a second row. Two new calls can produce two rows with the
same topic. The saved result has its own identity in each row.

Pass `contract=OutputType` on an agent call when its required output fields
depend on the input. Druks uses that type for the harness schema, validation,
artifact, and result conversion. The agent's declared contract stays unchanged;
`contract` is not prompt context. Build dynamic types with Pydantic `create_model`
and use `to_result()` to return a stable type for durable storage.

If an agent produces a file, use [`File` and `FileField`](files.md).
The contract declares the file, Druks transports and serves it, and the app can
persist its stable reference on an app row.

An app that runs a CLI of its own inside the sandbox reads how a declared agent
would run, and hands that to the CLI. Declare the agent and never call it; the
operator configures it in the app's **Settings → Agents**. Shared defaults
are in **Settings → Agents**:

```python theme={null}
config = await NightWatch.auditor.get_config()
config.harness   # "claude" | "codex" | "opencode" | "pi"
config.model_id  # the model as that CLI names it, provider prefix stripped
config.model     # "provider/model"
config.effort
config.billing   # "subscription" | "api_key"
config.secrets   # the Drukbox entries that put the key in the VM as a placeholder
```

`get_config()` returns an `AgentConfig` inside a workflow. This temporary value
contains shared execution settings and the run account's selected credential.
Druks resolves it at call time, as it does for an agent call. It stores no
personal preferences and has no database table. A missing login or key raises
before any sandbox work. Under subscription billing there is no key. The VM home holds the login of the calling agent's
subscription only, so a nested CLI on another provider needs `api_key` billing.
Under `api_key` billing, the VM holds the key as a placeholder in the variable
the entry names: `ANTHROPIC_API_KEY`, `OPENAI_API_KEY`, or `CODEX_API_KEY` for
Codex. A nested CLI reads it from the environment. The
[configuration guide](configuration.md#harnesses) lists the variable, host, and
header per harness.

### Give an agent a service's secret

An agent that calls a provider from its sandbox lists the secret it needs. The
[service](#declare-a-service) declares the one host that its secrets can go to:

```python theme={null}
class Acme(Service):
    host = "api.acme.example"

    class Settings(BaseModel):
        api_key: SecretStr = Field(title="API key")
        webhook_secret: SecretStr = Field(title="Webhook secret")


class NightWatch(App):
    name = "night_watch"

    report = Agent(
        prompt="night_watch/report.md",
        contract=ReportOutput,
        secrets=(Acme.fields.api_key,),
    )
```

`Acme.fields` holds the fields of `Settings` by name. The sandbox of each
`report` call holds a placeholder in `ACME_API_KEY`: the service slug and the
field name, in upper case. The secrets proxy puts the key in an
`Authorization: Bearer` header only on requests to `host`. The sandbox never
holds the key, and it never holds `webhook_secret`, which no agent lists.

Druks refuses to load an agent that lists a field that is not a secret, or a
field of a service with no `host`. If a service with `required = False` is not
connected, the sandbox gets no variable for it, so the prompt must handle that.

A workflow that sets `steps_reuse_sandbox = True` binds the secrets of the call
that creates the sandbox. Give the agents of such a workflow the same `secrets`.

For a credential that depends on the subject, such as one account's connection,
override [`get_secrets(subject)`](#customize-the-workspace) on the workspace.

A person's own sign-in needs no list. If a service has OAuth endpoints and a
`host`, the Chat sandbox of each operator holds that operator's sign-in under
the service slug, the way it holds a listed field. An operator with no sign-in
gets no variable.

Do not ask the framework to infer domain side effects from agent prose.
The prompt or a subsequent explicit step owns those actions.

## Answer WhatsApp with a Bot

A Bot answers the connections that an operator links to your app, such as a
WhatsApp number. It is not an agent: it has no contract, and no workflow awaits
it. Each person who writes gets their own [chat conversation](chat.md#whatsapp).
The Bot acts only through the app routes that you tag `bot`.

Declare the Bot as the `bot` attribute of the app class. An app has one Bot:

```python theme={null}
from druks.agents import Bot
from druks.apps import App


class Helpdesk(App):
    name = "helpdesk"

    bot = Bot(
        prompt="helpdesk/bot.md",
        user_tools=("get_ticket", "request_access"),
        admin_tools=("list_requests",),
    )
```

* `prompt` names a template, like an agent's prompt. Druks renders it with
  `source`, the conversation's source (`web` or `whatsapp`), and gives it to the
  agent as its system prompt. Druks adds one paragraph at the end. It explains
  the `[Internal: …]` messages that Druks writes to the agent.
* `user_tools` are the tools of each person who writes to the connection.
* `admin_tools` are the tools of the connection's admin. The admin also gets
  Druks's admin prompt, `answer_gate`, and `chat_resume_conversation`.

Each name is the `operation_id` of one of the app's routes tagged `bot`. Druks
refuses to start when an app declares a Bot under another attribute name, or
when a name matches no such route. The Bot's agent has no shell, file, or web
tools.

Tag a route with `bot` to make it a tool for the Bot. The rules for `agent`
routes apply: an explicit `operation_id`, a docstring, and the app name as the
tool prefix. A `bot` route is never in the Druks toolkit of an operator's
agent. Tag a route with both `agent` and `bot` to put it in both.

A bot tool takes the person who writes as `user: BotUser`:

```python theme={null}
from typing import Annotated

from fastapi import APIRouter, Body

from druks.agents import BotUser

from .workflows import GrantAccess

router = APIRouter(prefix="/access")


@router.post("", tags=["bot"], operation_id="request_access")
async def request_access(system: Annotated[str, Body(embed=True)], user: BotUser) -> str:
    """Ask an admin to approve access to a system for the person writing."""
    return await GrantAccess.dispatch(system=system, user_id=user.id)
```

| Field | Value |
| - | - |
| `id` | The person's id on the channel, such as a WhatsApp id or a caller's number. It is always set. |
| `name` | The person's name, or empty when the channel does not know it. |
| `phone` | The person's phone number, or empty when the channel does not know it. |
| `source` | The channel of the conversation, such as `whatsapp` or `call`. |

Druks fills `user` from the conversation that the tool call came from, so it
never appears in the tool's input schema. A route that takes `BotUser` refuses a
call from outside a conversation.

A run that a bot tool starts remembers its conversation. Ask for approval with
an in-app question, `self.review()`. Druks then asks the connection's admin in the
admin's own chat, and only that admin can answer. When a run that waited ends,
Druks tells the conversation the run's result or its failure, and the Bot tells
the person. A cancelled run tells nothing.

To withdraw a waiting request, cancel it in a bot tool:
`await GrantAccess.cancel(request)`. A subject has one active run per workflow, so
cancel a request before you start a changed one.

The Bot has a row in the app's **Settings → Bots** under the id
`<app>.bot`. Its harness, model, billing, effort, and timeout resolve
like an agent's. The timeout is the longest that one turn can run. `timeout=`
on the Bot declares its default. The Bot runs on Claude, like Chat.

## Customize the workspace

Every agent uses a `Workspace` around a Drukbox sandbox. `Workflow.workspace_class`
names the kind. A workflow about a GitHub repository declares `RepoWorkspace`
and nothing else:

```python theme={null}
from druks.workspaces import RepoWorkspace


class Sweep(Workflow):
    subject = Repository  # a StoredSubject with a ``repo`` column, "owner/name"
    workspace_class = RepoWorkspace
```

Before every agent call Druks clones the default branch into
`workspace.repo_path`. Prompts read `{{ workspace.repo_path }}`. The sandbox
holds a placeholder for its GitHub token in `GH_TOKEN`, and Drukbox points git
and `gh` at it. The Drukbox secrets proxy swaps the placeholder for a token
that Druks mints on demand, so a long run never outlives its token. The clone
is idempotent, so a warm host keeps its working tree and a host rotated in
bare gets one back.

Every workspace holds the run's `subject`. `RepoWorkspace.get_repo(subject)`
reads its `repo` column. Override it when the subject names the repository
differently:

```python theme={null}
class SweepWorkspace(RepoWorkspace):
    @classmethod
    def get_repo(cls, subject) -> str:
        return subject.full_name
```

The sandbox's GitHub token comes from a connected service of the appliance.
`RepoWorkspace` names the operator App, `Github`. An app that acts as another
identity declares its own service in its `services` module, a subclass with
its own connect card, and names it on its workspace:

```python theme={null}
from pydantic import BaseModel, Field, SecretStr

from druks.core.services import Github
from druks.workspaces import RepoWorkspace


class GithubReviewer(Github):
    required = False

    class Settings(BaseModel):
        app_id: str = Field(title="App ID")
        private_key: SecretStr = Field(title="Private key (PEM)")


class ReviewWorkspace(RepoWorkspace):
    github = GithubReviewer
```

The operator connects the service in **Settings → Connections → Services**. The
sandbox's identity stores the service and the repo. The issuer reads only
those two, so a request cannot select another repo or identity.

Override `Workflow.get_workspace_kwargs()` to pass `branch` or the fields a
subclass adds. Extend `RepoWorkspace` by adding fields, not by cloning again.
Override `run_agent()` to prepare the VM before the call,
`get_agent_run_kwargs()` to grant directories or skills, and `get_env()` to give
every agent call environment variables:

```python theme={null}
@dataclass(frozen=True, kw_only=True)
class DeployWorkspace(RepoWorkspace):
    deploy_token: str

    def get_env(self) -> dict[str, str]:
        return {"DEPLOY_TOKEN": self.deploy_token}
```

Override `get_secrets(subject)` to give the sandbox a secret of its own:

```python theme={null}
from druks.sandbox import SandboxSecret
from druks.workspaces import Workspace

from .services import BillingApi


class InvoiceWorkspace(Workspace):
    @classmethod
    async def get_secrets(cls, subject) -> list[SandboxSecret]:
        return [
            SandboxSecret(
                name="billing_token",
                secret_id=(await BillingApi.get()).id,
                host="api.billing.example",
            )
        ]
```

The secret names the vault row the issuer answers from. With a `host`, it is a
custom entry. The sandbox holds a placeholder in `BILLING_TOKEN`, the name in
upper case. The secrets proxy puts the value in a request header only for that
host. A header row supplies its own header. Any other row goes out as
`Authorization: Bearer <token>`. An entry for `github.com` is Drukbox's GitHub
service under any name: `GH_TOKEN`, with git and `gh` set up. Without a `host`,
the name is a Drukbox catalog entry, and Drukbox sets its variable and hosts.
`resource` tells the issuer what the token is for, such as a repo. Druks reads
the secrets before the sandbox exists, so read them from the subject alone.

A vault row with one secret issues it under any name. For the secret of a
service that does not depend on the subject,
[list it on the agent](#give-an-agent-a-services-secret) instead.

Override `get_mcp_servers(subject)` to give the sandbox an MCP server with its
own vault row:

```python theme={null}
from druks.sandbox import SandboxMcpServer


class BuildWorkspace(RepoWorkspace):
    @classmethod
    async def get_mcp_servers(cls, subject) -> tuple[SandboxMcpServer, ...]:
        actor = await get_review_actor()
        return (
            SandboxMcpServer(
                name="github",
                url="https://api.githubcopilot.com/mcp/",
                secret_id=(await actor.service.get()).id,
                resource=cls.get_repo(subject),
            ),
        )
```

The server names the vault row the issuer answers from and what the token is
for: here a connected GitHub service and its repo. Druks binds the server's
host and the variable `MCP_GITHUB_TOKEN` to the entry when it creates the
sandbox. The harness configuration names the variable, and the sandbox never
holds the token. A workspace server owns its name, so a same-named registry
server is not delivered. `Workspace.get_all_mcp_servers(subject, account_id)`
returns the harness shapes and the secret refs for every MCP server of a
sandbox: the workspace's servers and the enabled registry servers. An agent
declared with `include_mcp=False` gets none of them.

Keep durable state outside the VM. A workflow can set
`steps_reuse_sandbox = True` to retain one host across a segment. Druks releases
the host at a gate and at workflow exit. It rotates the host near lease expiry,
and when the next agent call needs other secret entries.

### Borrow a browser session

Declare required logins on the app class. The attribute name and app name form
the session identity. The sessions pane asks the operator to sign in:

```python theme={null}
from druks.browser import BrowserSession
from druks.apps import App


class NightWatch(App):
    name = "night_watch"
    acme = BrowserSession(site="acme.example", persist=True)
```

A workflow borrows the logged-in browser as a Playwright handle. The app
declares Playwright as its dependency. Druks owns the browser container and its
lifecycle. The browser starts in a container on the Druks host. It stops with
the block. Druks exports and stores a `persist` session before the stop:

```python theme={null}
async with NightWatch.acme.playwright() as browser:
    page = await browser.new_page()  # opened on the logged-in context
    await page.goto("https://acme.example/home")
```

`playwright()` yields the logged-in browser context. Pages that you open in
this context use the session. `NightWatch.acme.cdp()` borrows the same browser
and yields the raw CDP URL. Use this URL with a test suite, raw CDP client, or
custom wrapper.

`persist=True` writes rotated state after each borrow. Use it for sites that
expire an unused login. `headless=True` is an optional optimization for sites
that do not fingerprint headless browsers.

`anonymous=True` declares a session that needs no login. A borrow opens a
browser with an empty profile. The operator does not sign in. Use this option
for a public target or app-owned credentials. These credentials can be an
identity header or a token in the URL. An anonymous session stores no state.
Thus, `persist=True` with it fails during class definition.

If a borrowed browser returns to the login page, raise
`BrowserSessionSignedOutError` from `druks.browser`. Druks marks the session
as stale. The sessions pane shows this state and stops new borrows. The run
fails with the same reason. After the operator signs in again, the next
scheduled run can proceed.

Provider selection is an operator concern. App workspace code targets the
Druks sandbox contract, not `exe`, AWS, or Docker directly.

## Wait for input

The gate fields form the reply schema. `name` fixes the durable gate identity.
This identity selects the receive channel and the `gate` value of the parked
run. You must declare it because the identity must survive a class rename:

```python theme={null}
from typing import Literal

from druks.workflows import Gate, Workflow


class ApproveReport(Gate):
    name = "approve_report"
    action: Literal["approve", "revise", "cancel"]
    note: str | None = None

    @classmethod
    async def on_wait(cls, workflow: Workflow) -> None:
        await notifier.report_ready(workflow.workflow_id)
```

Wait from `run_multistep()`:

```python theme={null}
reply = await ApproveReport.wait(
    input_request={
        "presentation": "external",
        "label": "Review the night-watch report",
        "url": review_url,
    }
)
```

`on_wait()` is a checkpointed notification step. The workflow then parks
durably and releases its warm sandbox. The owning external system resumes the
workflow through the gate and its subject:

```python theme={null}
await ApproveReport.answer(
    repository,
    action="approve",
    note="Ship it.",
)
```

`answer()` resolves the subject run that waits on the gate. It raises if no such
run exists. This includes a gate that has an answer or timeout. A subject can
have runs from several workflows. The gate identifies the applicable run.

For a subject-backed decision inside the Druks dashboard, use:

```python theme={null}
reply = await self.review(questions=report.questions, context=review_context)
```

It offers `approve` and `request_changes`. Druks shows optional nonblank
`context` next to the review. With this context, `request_changes` does not
require answers or a note. Authors must treat that response as another pass and
include the context. A subjectless workflow cannot use in-app review. A
subjectless custom gate must override `on_wait()` to show the wait.

Without this
override, Druks raises an error instead of a silent park.

For a yes or no decision, park the built-in `YesNo` gate:

```python theme={null}
from druks.workflows import YesNo

reply = await YesNo.wait(input_request={"presentation": "in_app", "label": "Keep this quote?"})
if reply.action == "yes":
    ...
```

Raise `FatalError` for a deliberate domain stop. If readers need a stable
machine failure code, subclass it. Set `code` on the subclass. Unexpected exceptions fail
the run and are re-raised to DBOS.

Stop this workflow's active execution for a subject through the workflow class:

```python theme={null}
await Sweep.cancel(repository)
```

The workflow class supplies its kind. The caller does not find or handle the
internal timeline row. A cancel request with no active run has no effect. Thus,
a redelivered webhook stays idempotent.

## Give runs a subject read-side

A subject is what your runs are about — a repository, a work item, a pull
request. It is always a class, and the workflow names it:

```python theme={null}
class Sweep(Workflow):
    subject = Repository
```

This declaration lets Druks show subject history and available actions before a
run exists. `start()`, `cancel()`, and `Gate.answer()` enforce the declaration.
A workflow with a subject starts with an instance of that class. A workflow
without a subject passes `subject=None`.

When the subject is a row you keep — one you list, edit, and show fields from —
subclass `StoredSubject` instead of `Model`. The class name is the subject type:
`Repository` becomes `repository`.

```python theme={null}
from druks.db import StoredSubject
from sqlalchemy.orm import Mapped


class Repository(StoredSubject):
    full_name: Mapped[str]

    def __str__(self) -> str:
        return self.full_name
```

A subject's `__str__` is its name on runs, the Activity feed, and its board. The
default is its type and id, such as `repository 7`. Its type and id identify it,
so the name need not be unique. Druks supplies the rest: the table
`night_watch_repository`, an `id`, `created_at` and `updated_at`, `create()`,
`save()`, `delete()`, and a board of the newest hundred rows by `updated_at`, or
in the class's declared ordering. To scope the board by caller, or to select
other rows, override `list_summaries()`.

If you keep no row for a subject, subclass `Subject`. The platform requires only
an identity. The ID is the full record and its name:

```python theme={null}
from druks.workflows import Subject, SubjectSummary


class PullRequest(Subject):
    @classmethod
    async def list_summaries(cls, account_id: str | None) -> list[SubjectSummary]:
        return [pull_request.get_summary() for pull_request in await cls.list_open()]
```

Each ID names one of these subjects, so a detail read always answers. Override
`get_or_none(id)` to reject an invalid shape. For example,
`owner/repo#7` is a pull request and `nonsense` returns a 404:
`await PullRequest.get(id=subject_id)` raises `ObjectNotFound` for it, the same
as `Model.get`. Code that already knows the parts builds the subject directly:
`PullRequest(id=f"{repo}#{number}")`.

An identity-only `Subject` a workflow declares must implement
`list_summaries()`; a `StoredSubject` has a board by default. The board reads
the method and passes the caller. `account_id` is the signed-in account, or None
outside a request. If each operator has a separate board, use it to scope the
rows. If all operators share one board, ignore it.

A model method never reads request context. Druks validates the method at load.
If a `Subject` lacks it, the app does not load. The error names the app, the
subject, and the method.

Druks serves the same `/api/night_watch/repository` surface for both subject
types. This surface contains a board, detail pages, and a live stream. Druks
mounts it for each declared subject. Each response contains your summary, run
status, timeline, agent calls, artifacts, active question, and the sandbox
phase while a run starts.

Pass the subject instance to each component that requires one. This includes a
workflow start, gate answer, or event:

```python theme={null}
await Sweep.start(subject=repository, repo=repository.full_name)
```

Inside the workflow, `self.subject` resolves through the declared class. It is
live, not a snapshot from dispatch. A run can park on a gate for three days.
After resume, it reads the current row. If the row no longer exists, the subject
does not resolve.

Your app names domain outcomes. For example, a work item ships or an operator cancels it.
Druks owns the active run state. Read this state from the status:

```python theme={null}
status = await repository.get_status()
if status.is_parked:
    ...  # a run stopped to ask a human something
```

`status.kind` names the workflow currently driving the row and `status.gate` the
question it stopped on. While a run is active, `await repository.get_phase()`
returns the step it is on.

A subject that Druks did not run has no state: `status.state` is None, and so
is `status.run`. `await repository.get_status(workflow=Sweep)` narrows the read to one
workflow's runs. It answers the same way when the subject has no run of that
kind.

A page that lists subjects reads them all at once instead. `get_statuses()`
takes the ids and returns one status per id, in a single read:

```python theme={null}
summaries = await Repository.list_summaries(account_id)
statuses = await Repository.get_statuses([summary.id for summary in summaries])
```

This is the read the platform's own board uses, so a declared page listing
fifty rows costs one query rather than fifty.

A page needs neither read to show where the work stands.
`ui.SubjectStatus(repository)` takes the subject, and Druks reads every status
on the page when it serves the page. See
[Where the work stands](druks-ui.md#where-the-work-stands).

## Activity facts and signals

Use [announcements](#announcing-domain-events) for facts the app owns.
Use an agent output's `to_artifact()` and `to_event()` for a saved result.
Do not announce that same result again from a completion subscriber. Field Notes
records `gist.prepared` for each gist that its agent prepares. It announces
`note.gist_approved` when the operator approves a gist.

Druks records these workflow facts without app calls:

* Accepted work: a new subject run entered the queue. A deduplicated start adds
  no second row.
* Input requested: the workflow reached a gate. The record holds the gate and
  the request time for that round.
* Response received: the concrete gate validated a reply. This does not mean
  that the domain accepted the reply's proposal. The workflow decides that.
* Run failed: the run ended with a terminal failure. Routine running and
  finished signals remain available to subscribers but do not enter Activity.
* Run cancelled: an operator cancelled the run through the cancel route. Domain
  cleanup that cancels a run records no row and announces no owner outcome.

An external owner can announce an outcome after the run stops. Record that
outcome when the owner reports it. Do not infer it from the run state.

Each Activity row keeps the subject's name as it was recorded. `start()` stores
`str(subject)` in the workflow attributes. Admission, transitions, workflow
announcements, output artifacts, and operator cancellation use that run's
recorded name. A rename during the run applies to the next run. A subject
announcement records the subject's name when it records the event.

Druks records `payload.run` and `payload.kind`. An announcement that names one of
them raises `WorkflowError`.
A later rename or deletion does not change history. Search matches a literal,
case-insensitive part of the recorded key or title. It does not search current
subjects, failure text, or artifacts.

The feed exposes one `payload` dictionary with the stored fact names, including
app-owned facts. The envelope supplies `id`, `seq`, `at`, `topic`, `app`,
`subjectType`, `subjectId`, and `subjectKey`. The dashboard supplies readable
activity labels. For example, it shows `gist.prepared` as "Gist prepared". Keep
UI wording out of the event identity.

A gate request and its reply retain the same run, gate, and request-time
identity. A result retains its artifact identity. These references describe the
recorded occurrence even when the current subject, run, or file is unavailable.

React with filters rather than body guards:

```python theme={null}
from druks.signals import subscribe
from druks.workflows import WorkflowEvent


@subscribe(WorkflowEvent.FINISHED, subject=Repository)
async def on_sweep_finished(*, subject: Repository, **_: object) -> None:
    await notify(subject.full_name)
```

`subject=Repository` selects each workflow for a repository. `workflow=Sweep`
selects one workflow and its declared subject. Do not use both filters together.
The subscriber body receives its subject with either filter.

Signals deliver at least one time. A subscriber exception propagates. Then the
webhook provider or DBOS retries the publication. Make each reaction idempotent.

## Receive webhooks

A webhook authenticates and normalizes provider input. It must publish a
domain-neutral signal rather than contain workflow policy:

```python theme={null}
from fastapi.responses import JSONResponse

from druks.signals import publish
from druks.webhooks import Webhook, verify_hmac_sha256


class NightWatchWebhook(Webhook):
    provider = "night_watch"
    category = "events"

    async def request_is_authentic(self) -> bool:
        verify_hmac_sha256(
            self.raw_body,
            self.request.headers.get("x-signature"),
            secret,
        )
        return True

    def get_action(self) -> str:
        return self.data["type"].replace(".", "_")

    async def on_report_approved(self) -> JSONResponse:
        await publish("report.approved", payload=self.data)
        return JSONResponse({"accepted": True})
```

The public path is `/_external/night_watch/events/`. Druks deduplicates a
delivery when the class supplies a delivery key. A failing handler releases the
claim so the provider can retry.

## Models and migrations

Models subclass `druks.db.Model`. Druks names the table for the app and the
class: `Report` in `night_watch` is the table `night_watch_report`. A class that
sets `__tablename__` keeps it, and every app table starts with `<name>_`:

```python theme={null}
from sqlalchemy.orm import Mapped, mapped_column

from druks.db import Model


class Report(Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    repo: Mapped[str] = mapped_column(unique=True)
    status: Mapped[str]
```

A model reads and writes by field:

```python theme={null}
report = await Report.create(repo="acme/widgets", status="open")
report = await Report.get(id=report_id)
report = await Report.get_or_none(repo="acme/widgets")
reports = await Report.all()
open_reports = await Report.filter(status="open")
report.status = "closed"
await report.save()
await report.delete()
```

`get` raises `ObjectNotFound` on a miss; the API answers it with 404 and a
page with an empty state, so a route or page that names a row by id never
spells either. `get_or_none` answers None instead. Both expect one row: two
raise SQLAlchemy's `MultipleResultsFound`, so back the fields they read with a
unique constraint. `all` returns every row and `filter` the rows that match at
least one field, both in primary key order, or in the order the class declares
on its class line, in Django's form:

```python theme={null}
class Report(Model, ordering=("-created_at",)):
    ...
```

A text value is read as its column's type, so an id straight off a URL finds its
row. A read that needs a limit, a different order, or anything but equality
writes `select()`:

```python theme={null}
reports = await db_session().scalars(
    select(Report).where(Report.status != "closed").order_by(Report.created_at).limit(20)
)
```

A page that names a missing row by id answers an empty state that links back to
its parent page, or to the app's landing page.

A `Mapped[datetime]` column stores UTC. A `Mapped[SomeStrEnum]` or
`Mapped[SomeLiteral]` column stores its value as text under a CHECK constraint of
the allowed values. A `Mapped[list]` or `Mapped[dict]` column is JSONB; a typed
one such as `Mapped[list[dict[str, Any]]]` names the type,
`mapped_column(JSONB)`. Encrypted columns come from
`druks.db.fields`: `EncryptedTextField` and `EncryptedJsonField`, with their
value types `Secret` and `SecretsMapping`.

Generate the app's revision after the model is importable:

```bash theme={null}
uv run druks makemigrations night_watch -m "add reports"
uv run druks init-db
```

Druks scopes autogeneration to the table prefix, names the revisions
`night_watch_0001`, `night_watch_0002`, and so on, and writes the version to
`alembic_version_night_watch`. Never write a revision by hand. The one change
Alembic does not detect is a new member of an enum: that is one
`drop_constraint` and one `create_check_constraint`.

Druks supports Postgres 16 and newer, so a model or revision uses only what 16
has. For example, do not use the SQL function `uuidv7()`, virtual generated
columns, or `RETURNING OLD/NEW`. Make a UUIDv7 in Python, as Druks does.

Query through `druks.db.db_session()` inside an
HTTP request, durable step, or other platform-bound session. Outside those,
`db_session()` raises. A workflow body holds no session: read inside a `@step`.
`await self.subject`, `start()`, and `dispatch()` bring their own. A row a step
returns is detached: load what the body reads from it inside that step.

HTTP response models subclass `druks.schemas.Schema`, whose snake\_case fields
serialize as camelCase. Request models are ordinary Pydantic models.
Druks mounts each router from a discovered `routes.py` below the app namespace.
It tags the router with the app name. A router declares only the prefix of its
resource:

```python theme={null}
router = APIRouter(prefix="/reviews")
```

Your routes require authentication. The loader puts each router behind the
platform identity gate. The gate accepts a Bearer PAT or the signed-in session.
The gate blocks anonymous requests before your code. You do not implement authentication.
If a route uses account scope, read the caller:

```python theme={null}
from druks.accounts import current_account_id

@router.get("/reviews")
def list_reviews() -> list[ReviewResponse]:
    return Review.list_for_account(current_account_id.get())
```

Druks prefixes every operation id of an app with the app name, so write the
bare verb. For example, `operation_id="write_note"` in `field_notes` becomes
`field_notes_write_note` in the OpenAPI document. Startup refuses an id that
already starts with the app name. An `Action` names the bare id.

Tag a route with `agent` to create an MCP tool from it. Give the route an
explicit `operation_id`, and the tool takes the prefixed name. The docstring
supplies the description.

A `GET` route is read-only. If a write is non-destructive, declare
`x-destructive: false`. If a write is idempotent, declare `x-idempotent: true`.
Safe defaults are destructive
and non-idempotent. Startup refuses a missing `operation_id` or docstring.

Two spellings run through druks, and which one a segment wears says who owns it:

| | |
| - | - |
| `snake_case` | an identity the platform serves — your app name, a subject type |
| `kebab-case` | a resource you named — your route prefixes, your frontend paths |

Thus, `/api/software_factory/pull_request` is the subject board for pull request
review runs. `/api/software_factory/reviews` is the resource that your POST
creates. The platform matches `<subject_type>`, `transcripts`, and `pages` before
your routers. A custom router cannot take a platform read, including through a
catch-all.
`transcripts` and `pages` are reserved: a subject type or a router prefix that
takes one fails the load. Name the router for its resource to prevent a
conflict.

## Declare a service

A service identity is the appliance registration at an external provider. A
deployment has one identity for each service string. The platform GitHub App is
the first service identity. OAuth grants are not service identities. The
platform stores them after an operator connection.

See [Connect provider accounts](#connect-provider-accounts-oauth).

An app serves a service's endpoints under `/api/<app>/services/<slug>/...`, where
the app mounts those routers like every other route.

Put a credential that only your app uses in its app settings.

Declare one class in `services.py`. The platform creates the connection card in
Settings. It validates and stores the submitted values. It encrypts
`SecretStr` fields and stores plain fields as identity facts. A field with a
default can stay empty on the card.

It also reports the state through `druks doctor`. The class name is the identity.
Druks derives the slug
from it (`Gmail` → `gmail`, `GoogleCalendar` → `google_calendar`) and derives
the card heading from the slug:

```python theme={null}
from pydantic import BaseModel, Field, SecretStr

from druks.services import Service, ServiceConnectError


class Gmail(Service):
    description = "The appliance's own OAuth client — every mailbox authenticates against it."

    class Settings(BaseModel):
        client_id: str = Field(title="Client ID")
        client_secret: SecretStr = Field(title="Client secret")
```

The slug names the service's vault row and the connect wire. A class
rename changes the slug, rekeys the card, and orphans the connected identity.
Set `slug = "gmail"` on the class to keep the old key.

Read it back through the same class:

```python theme={null}
(await Gmail.get()).secrets["client_secret"]   # raises ServiceNotConnectedError when unset
await Gmail.is_connected()
```

An optional `verify` classmethod proves the paste against the live provider
before anything replaces a working identity. It returns extra identity facts
to store, and raises `ServiceConnectError` with a message safe to show — the
platform never echoes what the operator pasted:

```python theme={null}
@classmethod
async def verify(cls, settings: Settings) -> dict:
    if not await probe(settings):
        raise ServiceConnectError("The provider did not accept these credentials.")
    return {}
```

If the appliance is healthy without the service, set `required = False` on the
class. Doctor then reports a note instead of pending setup.

The card renders `description` as Markdown, so it can link to a setup guide.

Key the service for the integration that your app consumes (`Gmail`), not the
provider (`Google`). A second integration on the same provider declares its own
service. The operator decides whether each card uses a shared or narrow
registration. This choice controls scope and the effect of a credential problem.

If the provider runs an MCP server that the service's identity also signs in to,
set `mcp_host` to the host of that server. An OAuth MCP server at that host then
uses the service's credential. If the service has OAuth endpoints, each account
uses its own sign-in at the service, with no second consent. If not, the
default account sends the service's pasted login, and every other account
connects its own grant. Override `get_authorization` to turn the connected row
into the `Authorization` value:

```python theme={null}
class Acme(Service):
    mcp_host = "mcp.acme.example"

    class Settings(BaseModel):
        api_key: SecretStr = Field(title="API key")

    @classmethod
    def get_authorization(cls, login) -> str:
        return f"Bearer {login.secrets['api_key']}"
```

## Connect provider accounts (OAuth)

Declare the OAuth endpoints on the service that holds the client
credentials. The `Settings` model must have `client_id` and `client_secret`
fields:

```python theme={null}
class Acme(Service):
    authorization_endpoint = "https://acme.example/oauth/authorize"
    token_endpoint = "https://acme.example/oauth/token"
    # True = HTTP Basic on the token endpoint. False = secret in the body.
    basic_auth = True
    identity_endpoint = "https://acme.example/oauth/userinfo"
    identity_scopes = ("openid", "email")
    # Query parameters the provider's consent URL must carry.
    extra_authorize_params = {"access_type": "offline", "prompt": "consent"}

    class Settings(BaseModel):
        client_id: str = Field(title="Client ID")
        client_secret: SecretStr = Field(title="Client secret")
```

`extra_authorize_params` declares special consent-query values for the provider. The
platform adds them to every sign-in it starts for the service. The example
shows Google's: it grants a refresh token only when the consent asks for
`access_type=offline` with `prompt=consent`.

`identity_endpoint` names the provider endpoint that returns the signed-in
account's facts (email, username, name). Druks calls it once at consent and
shows the facts as the connection label in Settings. `identity_scopes` are the
scopes that this call requires. Druks adds them to the consent request.

If one identity fact names the provider account, declare `identity_key`:
`"sub"` for Google, `"id"` for GitHub. A fresh sign-in that matches an
existing connection for the same owner updates that row instead of
creating a second one. A revoked row that matches becomes live again and
keeps its id. Without the declaration, each fresh sign-in creates a new
connection.

Some providers have no such endpoint, or return the facts in a different
shape. Override `get_identity` for them:

```python theme={null}
    @classmethod
    async def get_identity(cls, access_token: str) -> dict:
        payload = await fetch_profile(access_token)
        return payload["data"]
```

One provider can back several services — Google backs both Gmail and Google
Calendar, and each keeps its own card and its own key. Share the provider's
declarations through an abstract base. Set `abstract = True`. The base never
registers. Each subclass inherits everything it declares, `Settings`
included, and needs nothing beyond its class name:

```python theme={null}
class GoogleOauth(Service):
    abstract = True
    authorization_endpoint = "https://accounts.google.com/o/oauth2/v2/auth"
    token_endpoint = "https://oauth2.googleapis.com/token"
    extra_authorize_params = {"access_type": "offline", "prompt": "consent"}
    identity_endpoint = "https://openidconnect.googleapis.com/v1/userinfo"
    identity_scopes = ("openid", "email")
    identity_key = "sub"

    class Settings(BaseModel):
        client_id: str = Field(title="Client ID")
        client_secret: SecretStr = Field(title="Client secret")


class Gmail(GoogleOauth):
    pass


class GoogleCalendar(GoogleOauth):
    pass
```

Declare your app's use of the service, with the scopes your calls
need:

```python theme={null}
class NightWatch(App):
    name = "night_watch"
    acme = Acme.with_scopes("profile.read", "posts.write")
```

A *connection* is one signed-in provider account. A user can have several
connections for each provider. Each mailbox, handle, or workspace can have one.
The platform stores the refresh token, granted scopes, and owner for each
connection. Workflow code reads them through the declaration. It gets one token
for each connection:

```python theme={null}
for connection in await NightWatch.acme.list_for_account(account_id):
    token = await connection.get_access_token()
```

Read existing sign-in facts directly from the service when you do not need to
declare scopes:

```python theme={null}
from druks.core.services import Github

sign_ins = await Github.list_for_account(self.account_id)
```

Each connection's `identity["login"]` names the GitHub user. The direct read
does not add the app to the service's scope declarations. An empty list means
the account has no live sign-in. Both reads exclude revoked connections.

`account_id` is the caller: `self.account_id` in a run body,
`current_account_id.get()` in a route, the handler's argument in a
subscriber, the platform's argument in `list_summaries`. `await NightWatch.acme.get(connection_id)` returns one connection
when your own row stored its id. Each connection carries `id`, `scopes`, `identity` — the
provider's facts for the sign-in — `account_id` — the druks account that
signed it in — and `connected_at`. The `scopes` value is a list, or `None`
when the provider did not report scopes. The handle serves
live connections only. A revoked connection drops out of `get` and
`list_for_account`, but its platform row survives with its owner and
identity. Your rows never need tombstone copies of either.

Your UI starts a sign-in by opening `/api/oauth/acme/connect`. The platform
requests the combined scopes from each installed app. It stores the connection
for the signed-in user. A fresh
sign-in creates a new connection, unless the service's `identity_key`
matches it to an existing connection for the same owner.

To widen an existing connection's scopes, open
`/api/oauth/acme/connect?connection=<id>`. Reconsent replaces its tokens.
Only the user who owns the connection can reconsent or revoke it. For
another user's connection, the platform answers 404.
Reconsent names the row, so it also makes a revoked connection live again
under its old id. A fresh sign-in that matches the `identity_key` does the
same.

Add `?next=/night_watch/accounts` to land the user back on your
page after consent instead of the generic "connected" page. `next`
must be a bare path that starts with `/`. Druks rejects a URL with a scheme or
host. Thus, the connection flow cannot redirect away from the host. Register
`https://<host>/api/oauth/callback` as the provider redirect URI. It serves each
service.

React to sign-ins with the signal machinery. The platform publishes
`oauth.connected` when a consent completes. `reconsent` is true when the
consent replaced an existing connection's tokens. This happens on
reconsent by id and on an `identity_key` match, for a live or a revoked
row.

It publishes `oauth.disconnected` after a user revokes a connection. A
replacement of the service's client credentials also publishes this signal,
and so does a token refresh that the provider answers with `invalid_grant`:
Druks revokes that connection, because the grant is dead at the provider.
When the operator disconnects the service in Settings, Druks publishes the
signal for each of its connections.
Revocation is a state, not a deletion: your subscriber can still read the
connection it is told about. Subscribe in `subscribers.py`:

```python theme={null}
from druks.signals import subscribe


@subscribe("oauth.connected", provider="acme")
async def adopt_sign_in(
    provider: str, connection_id: str, account_id: str, reconsent: bool
) -> None:
    if not reconsent:
        WatchedAccount.adopt(connection_id, account_id)


@subscribe("oauth.disconnected", provider="acme")
async def drop_sign_in(provider: str, connection_id: str, account_id: str) -> None:
    WatchedAccount.drop(connection_id)
```

The user sees and revokes everything in Settings — every connection they
hold, across services, revoked ones shown as history. Replacing a service's
client credentials revokes its connections: a new client can never refresh
the old client's tokens, but the consents stay on record.

`get_access_token` serves a Redis-cached access token and lets only one
refresher run per connection and scope set. This is necessary: two
refreshes at the same time can make the provider revoke the whole
connection. It raises `OauthRefreshError` when the refresh fails. Then ask
the user to reconnect.

`get_access_token(scopes=("profile.read",))` asks the provider for a token with
fewer scopes than the grant. If the token goes to untrusted compute, use it.
Pass a subset of the connection scopes. `cached=False` bypasses the cache to get
a full-lifetime token.

## App settings and checks

An inner `AppSettings` class defines dashboard-editable knobs and owns their
cross-field coherence:

```python theme={null}
from typing import Literal

from pydantic import Field

from druks.apps import App, AppSettings, Secret


class NightWatch(App):
    name = "night_watch"

    class Settings(AppSettings):
        provider: Literal["none", "acme"] = "none"
        service_token: Secret = Field(
            json_schema_extra={"section": "Acme", "visible_when": {"provider": ["acme"]}},
        )
        webhook_secret: Secret = Field(
            title="Webhook secret",
            json_schema_extra={"section": "Acme", "visible_when": {"provider": ["acme"]}},
        )

        def clean(self) -> dict[str, str]:
            if self.service_token and not self.webhook_secret:
                return {"webhook_secret": "Required once the service token is set."}
            return {}
```

Supported display shapes are scalar values, `Literal` choices, and
`SecretStr`, including optional forms. Druks rejects nested Pydantic models.
It redacts secret values and submitted validation errors. Declare a secret
field as `Secret`.

A `Literal` field can give each value a label and description through
`json_schema_extra`. `choice_details` maps each value to its `label` and `help`.
Druks rejects a key outside the declared choices when it loads the declaration.
The form shows only the selected value's help and saves the original value:

```python theme={null}
review: Literal["human", "automatic"] = Field(
    default="human",
    title="Review policy",
    json_schema_extra={
        "choice_details": {
            "human": {"label": "Human review", "help": "Wait for your approval."},
            "automatic": {"label": "Automatic review", "help": "Continue after the automated checks pass."},
        },
    },
)
```

A `str` field can take its choices from a live source, such as a connected
service. Declare it as `Annotated[str, Choices(source)]`, with `Choices` from
`druks.apps`. The async source returns dicts with `value` and `label` strings.
An optional `group` string puts choices under a group heading. Druks calls each
source once per choices request and shows these fields as searchable selects.
The select starts with an empty choice. It marks a stored or edited value that
the source no longer lists under **Unavailable choices**. When the source returns
no choices, the field stays a text box.

```python theme={null}
async def list_board_choices() -> list[dict[str, str]]:
    return [
        {"value": "ops", "label": "Operations board", "group": "Boards"},
        {"value": "dev", "label": "Development board", "group": "Boards"},
    ]


class Settings(AppSettings):
    board: Annotated[str, Choices(list_board_choices)] = Field(default="ops", title="Board")
```

An unset field is an empty, false `SecretStr`. Thus,
`if self.service_token:` reads its state without a guard for
`.get_secret_value()`. A multiline secret, such as a PEM private key, can use
`json_schema_extra={"multiline": True}`. The settings form shows a textarea and
keeps newlines. Storage, redaction, and write-only behavior do not change.

`section` is a plain heading that Druks renders in first-declaration order, with
unsectioned fields first. `visible_when` takes one same-model `{field: [values]}`
condition. The field shows when its controller holds one of the values. The
controller must be non-secret and unconditional, and a `Literal` controller
requires declared members.

Hidden fields keep their stored values. Read the resolved model with
`await NightWatch.settings()`. The settings form runs `clean()` against the
resolved settings after the proposed edits and rejects an incoherent save. `druks doctor`
runs the same method over stored settings so rows from older releases or manual database
edits remain visible. Workflow settings stay plain Pydantic `BaseModel` declarations.

An app can add precondition checks through `checks`. These checks supplement
settings validation. Return `druks.doctor.CheckResult`. Druks namespaces the
result. It converts an exception or malformed result into an error without
hiding later checks.

## Test an app

The Druks installation registers its pytest plugin. An app can request the
fixtures directly without a `conftest.py` or `pytest_plugins` declaration:

| Fixture | Contract |
| - | - |
| `druks_db` | A SQLAlchemy `AsyncSession` bound to a per-test transaction. Commits become savepoints, and teardown rolls the outer transaction back. |
| `druks_client` | An authenticated `TestClient` with installed apps mounted, sharing `druks_db`'s connection. A request starts with no session bound, as in production. |
| `druks_redis` | The test Redis database, flushed before the test. |
| `druks_without_dispatch` | Workflow starts and run-phase writes become no-ops and a run-phase read finds no phase, for tests that stand up no durable engine. |
| `druks_without_remote_config` | Every `.druks` namespace lookup misses, so prompts resolve to bundled templates and config to its declared defaults. |

The fixtures are not autouse. A test that requests `druks_client` also gets
`druks_db`. A test accesses Redis only if it requests `druks_redis`.

`druks check-app` loads one installed app and checks its subjects, pages,
operations, and routers. It needs no database and no configured install, so an
app's CI can run it. It exits non-zero on the first contract the app breaks:

```bash theme={null}
druks check-app field_notes
```

Run a workflow's body against a subject with no durable engine — no checkpoints,
no lifecycle events, no retries:

```python theme={null}
from druks.testing import run_workflow

summary = await run_workflow(RecordHeartbeat, subject=None, source="night_watch")
```

`@step` calls inside a `run_multistep` body still need the real engine.

Seed platform-owned run and agent-call rows with plain functions:

```python theme={null}
from druks.testing import seed_call, seed_run

run = await seed_run(
    druks_db,
    kind=Sweep.kind,
    subject=repository,
    state="running",
)
call = await seed_call(
    druks_db,
    run,
    NightWatch.report.id,
    status="running",
)
```

`seed_run` writes both the run row and its DBOS workflow status. That status
determines `Run.state`. `seed_run` requires `kind`. If you seed
`state="parked"`, pass `input_gate`. `seed_call` accepts an agent's string ID.

`make_settings(tmp_path, **overrides)` builds isolated Druks settings.
`configure_app_for_test(settings=..., authenticated=False)` returns the mounted
app if a test needs its own client or an unauthenticated request path.
`druks_client` covers the normal authenticated case.

The fixtures never read the runtime's settings. They read
`DRUKS_TEST_DATABASE_URL` and `DRUKS_TEST_REDIS_URL`. The defaults are a local
`druks_test` database and Redis index 15. The fixtures point the code under test
to the same pair. Thus, a run cannot use the values in `DRUKS_DATABASE_URL` or
`DRUKS_REDIS_URL`.

Create the database one time with `createdb druks_test`. The
development Compose project already creates it.

On that database, the plugin creates `citext` and imports installed app models.
It runs SQLAlchemy `create_all`. It builds the DBOS system tables through DBOS
database migrations. It does not reset or drop a schema. It rolls back each test
write through `druks_db`. `druks_redis` runs `FLUSHDB` on the test index.

## Declare pages

An app declares its screens in Python and ships no JavaScript. Pages live in
`pages.py`:

```python theme={null}
from druks import ui


@ui.page("/")
async def reports():
    return ui.Page("Reports", description="Every sweep this install ran.")


@ui.page("/peers/{peer_id}")
async def peer(peer_id: int):
    return ui.Page(title=f"Peer {peer_id}")


@peer.child("/history")
async def peer_history(peer_id: int):
    return ui.Page(title=f"Peer {peer_id} history")
```

`@page` declares a top-level page. `@parent.child` declares a page under it,
one level deep. A page function takes one parameter for each parameter of its
route, so a child inherits its parent's. It needs no return annotation.

Exactly one page declares `/`. That page is the one the app opens on. A static
child renders as a tab on its parent. A parameterized child is a detail page a
`Link` reaches.

The page name is the function name. The page label is that name with its
underscores as spaces, and `label=` overrides it.

`App.pages()` enumerates the pages in route-match order: literal segments
before parameters, at every depth. Declaration order never decides a match.

`navigation` names the pages the appbar shows as subnav tabs, in order. Each
one must be a static top-level page. The tab wears the page's label, so an app
never spells a label twice:

```python theme={null}
class NightWatch(App):
    name = "night_watch"
    navigation = ["reports"]
```

Druks checks the whole table at boot. Two landing pages, a repeated page
name, a nested child, two routes a request could not tell apart, a signature
that does not match its route, or a navigation entry that is not a static
top-level page fails the load, with the app name and the exact cause.

### A page function is a pure read

Druks reruns a page function on initial load, on an event, on reconnect, on a
manual refresh, and on a retry. The call count and the call order are not
guaranteed. These are the rules.

A page function can:

* read Druks state,
* read the app's own data,
* read a projection,
* read a read-only external source.

A page function must not:

* write data,
* start or enqueue work,
* publish an event,
* answer a gate,
* cause an external effect,
* depend on mutable process state.

Write the function so that a repeat call is free. Put every write behind an
operation the operator triggers.

### Watch a subject

A page, or a named region inside it, declares the subject it `follows`. Druks
streams that subject through the read side every app already gets, and the
shell rereads the page on each snapshot:

```python theme={null}
@ui.page("/notes/{note_id}")
async def note(note_id: int):
    found = await Note.get(id=note_id)
    return ui.Page(
        title=f"Note {note_id}",
        blocks=[
            ui.Section(
                title="Your decision",
                name="decision",
                follows=found,
                blocks=[ui.GateControls(found)],
            )
        ],
    )
```

The shell replaces the named region and leaves the rest of the page alone, so
scroll position, focus, and half-filled inputs outside it survive. A region
that follows a subject must have a name. That is how the shell finds it.

`GateControls` takes the subject. The shell shows nothing while the subject
waits on nothing. Otherwise it shows the ask, its options, its context, and its
artifact, and sends the operator's answer. A `GateControls` block must sit
inside something that follows a subject, or an answered gate would stay on
screen.

### Let an operator act

A page acts through the app's own routes. Give the route an `operation_id`, and
name it from an `Action`:

```python theme={null}
# routes.py
@router.post("/notes", status_code=201, operation_id="write_note")
async def write_note(body: Annotated[str, Body(embed=True)]) -> dict[str, int]:
    note = await Note.create(body=body)
    await Summarize.dispatch(note=note)
    return {"id": note.id}
```

```python theme={null}
# pages.py
@ui.page("/notes/new")
async def new_note():
    return ui.Page(
        "Write a note",
        blocks=[
            ui.Form(
                title="New note",
                fields=[ui.TextAreaField(name="body", label="Note", is_required=True, rows=3)],
                action=ui.Action(
                    label="Save",
                    operation="write_note",
                    tone="primary",
                    link=ui.Link("Notes", page="notes"),
                ),
            )
        ],
    )
```

The shell resolves the operation to its method and URL, so the author writes no
URL. It sends the action's `arguments` and the field values as one object, fills
the route's path parameters from it, and posts the rest as the body. Your route
keeps the identity gate, and its Pydantic and FastAPI errors come back on the
fields they belong to.

Once the operation answers, the action does what it declared: `refresh` reads
the page or the region again, `link` navigates, and `confirm` asks the operator
first. A `tone` of `danger` says so on screen.

A form that needs a token takes a `ui.SecretField`:

```python theme={null}
ui.SecretField(name="token", label="Access token", help_text="From your account settings.")
```

Druks masks it, keeps it from the browser's password managers, and leaves it
empty after a successful submit. A refusal shows a fixed line, never the
server's words, because a validation message can carry the token back.

The masking protects the screen. Your operation receives the token in plain
text and owns it from there. Keep one secret per record in an
`EncryptedJsonField` column from `druks.db.fields`. A token the whole app shares
belongs in `AppSettings` as a `Secret`, where Druks encrypts it and never reads
it back.

Two routes of one app cannot share an `operation_id`. An action that names an
operation the app does not declare, a GET route, or a route with a query
parameter fails the page read: a GET is a read, and an action fills path
parameters and a JSON body.

### What V1 leaves out

V1 has no `Tabs` block, no accordion, no general modal block, no inline reveal
form, and no general client-state API. Static child pages give tabs, and the URL
holds the current one. Use `TableRow.detail` for expandable row text.

`MoneyValue`, `PercentValue`, `DurationValue`, and date and time input fields
are agreed and named. Druks adds each one when an app needs it. Ask instead of
working around it.

The [Druks UI contract](druks-ui.md) holds the block, value, and field catalog,
actions, and liveness.

## Frontends

An installed app is visible in the dashboard without a custom UI. The shell
reads the installed roster from `/api/apps`. It gives each app an entry in the
Apps sidebar and generic pages. Each subject type gets a board. Each subject
gets a page with its timeline, transcripts, and gate controls. The subject
summary fields form the board row.

No additional declaration is necessary.
The shell derives the sidebar label from `name` (underscores become spaces).

An app that needs full control of its interface ships a frontend instead — the
escape hatch, not the ordinary path. Its pages are its own JavaScript, so it
declares its own tabs there and leaves `App.navigation` empty. The scaffold
writes no JavaScript and no `dist/`. Add one only when the block catalog cannot
say what your app needs to say.

The frontend is an ES module that the shell mounts
inside its own document, below the chrome. The scaffold writes none, so an app
that takes this path creates `druks_night_watch/dist/` itself and points its
frontend build output there. The contract uses `shellApi: 1`:

* **Entry module:** `entry.js` exports `shellApi = 1` and `mount(el, ctx)`. The function renders into
  `el` and returns a dispose function. A missing `mount` or a version mismatch
  renders a visible error panel in the shell.
* **Context:** `ctx` carries `apiBase` (`/api/<name>`), `navigate(path)` for shell-side
  navigation, `theme.accent`, and `markdown(source)` — the shell's own
  markdown renderer, so an app does not bundle one. The app renders in the
  shell's document, so the shell's CSS variables cascade into it. The shell
  re-broadcasts every location change as a `popstate` event while the app is
  mounted.
* When `dist/style.css` is present, the shell loads it while it mounts the app.

Build the bundle in Vite library mode. Keep `react`, `react-dom`,
`react-dom/client`, and `react/jsx-runtime` external. The shell import map
resolves them to its copy. Thus, one React instance serves the document.
Bundle other dependencies as usual. Route by reading `location.pathname` under
`/<name>/`.

The bundled Druks SPA also has a shared React app registry. To join this shell,
compile the app UI module into the dashboard image. An installed Python wheel
cannot change an existing JavaScript bundle. See the
[frontend guide](https://github.com/czpython/druks/blob/main/frontend/README.md)
for that in-repository path.

## Stable author imports

Import from concern namespaces, not from `druks.durable` or internal modules:

| Namespace | Public names |
| - | - |
| `druks.accounts` | `current_account_id` |
| `druks.apps` | `App`, `AppSettings`, `Choices`, `Secret` |
| `druks.services` | `Service`, `Connection`, `ServiceConnectError`, `ServiceNotConnectedError`, `OauthClient`, `OauthExchangeError`, `OauthRefreshError` |
| `druks.browser` | `BrowserSession`, `BrowserSessionSignedOutError`, `BrowserSessionStatus` |
| `druks.agents` | `Agent`, `AgentOutput`, `Bot`, `BotUser` |
| `druks.workflows` | `Workflow`, `Gate`, `GateTimeout`, `step`, run/agent response types, lifecycle enums and workflow errors |
| `druks.sandbox` | `Sandbox`, `SandboxMcpServer`, `SandboxSecret` |
| `druks.workspaces` | `Workspace`, `RepoWorkspace` |
| `druks.db` | `Model`, `StoredSubject`, `db_session` |
| `druks.db.fields` | `EncryptedJsonField`, `EncryptedTextField`, `Secret`, `SecretsMapping` |
| `druks.exceptions` | `DruksError`, `ObjectNotFound` |
| `druks.schemas` | `Schema` |
| `druks.ui` | `Action`, `Block`, `Callout`, `Card`, `Cards`, `Chart`, `ChartSeries`, `CheckboxField`, `Columns`, `ControlsValue`, `Divider`, `EmptyState`, `Fact`, `Facts`, `Field`, `FileSummary`, `Files`, `Follows`, `Form`, `GateControls`, `Image`, `ImageGallery`, `Link`, `List`, `Markdown`, `Metric`, `Metrics`, `MultiSelectField`, `MultiUploadField`, `NumberField`, `NumberValue`, `Option`, `Page`, `Progress`, `ProgressStep`, `Quote`, `RadioField`, `Section`, `SecretField`, `SelectField`, `Stack`, `StatusValue`, `SubjectStatus`, `Table`, `TableColumn`, `TableRow`, `Text`, `TextAreaField`, `TextField`, `TextValue`, `TimeValue`, `Timeline`, `TimelineItem`, `UploadField`, `Value`, `page` |
| `druks.signals` | `subscribe` |
| `druks.events` | `Event` |
| `druks.files` | `File`, `FileField` |
| `druks.prompts` | `render_prompt` |
| `druks.webhooks` | `Webhook`, `verify_hmac_sha256` |
| `druks.testing` | `run_workflow`, `seed_run`, `seed_call`, `init_db` — plus the fixtures the plugin registers |

The root `druks` package deliberately exports only its version.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.