Writing an app

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}

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)

uvx --from druks druks create app night-watch
cd druks-night-watch
uv sync
uv run pytest

[project.entry-points."druks.apps"]
night_watch = "druks_night_watch.app:NightWatch"

uv pip install -e /path/to/druks-night-watch

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

from druks.apps import App


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

from druks.workflows import Workflow


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

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)

from druks.sandbox import Sandbox
from druks.workflows import Workflow


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

run_id = await Sweep.start(
    subject=repository,
    repo=full_name,
)

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

class SweepJournal(Journal):
    @property
    def findings(self) -> list[FindingData]:
        return self.filter(FindingData)


class Sweep(Workflow):
    journal_class = SweepJournal

await self.announce(
    "pr.opened",
    repo=item.repo,
    pr_number=delivery.pr_number,
    branch=delivery.branch,
)

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)

class Sweep(Workflow):
    every = "0 6 * * *"

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:
        ...

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:
    ...

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()

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.",
    )

    triage = Agent(
        prompt="triage.md",
        contract=TriageOutput,
        include_plugins=False,
        include_mcp=False,
    )

result = await NightWatch.report(repo=repo, findings=findings)

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

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,),
    )

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",),
    )

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)

from druks.workspaces import RepoWorkspace


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

class SweepWorkspace(RepoWorkspace):
    @classmethod
    def get_repo(cls, subject) -> str:
        return subject.full_name

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

@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}

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",
            )
        ]

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),
            ),
        )

from druks.browser import BrowserSession
from druks.apps import App


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

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")

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)

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

await ApproveReport.answer(
    repository,
    action="approve",
    note="Ship it.",
)

reply = await self.review(questions=report.questions, context=review_context)

from druks.workflows import YesNo

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

await Sweep.cancel(repository)

class Sweep(Workflow):
    subject = Repository

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

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()]

await Sweep.start(subject=repository, repo=repository.full_name)

status = await repository.get_status()
if status.is_parked:
    ...  # a run stopped to ask a human something

summaries = await Repository.list_summaries(account_id)
statuses = await Repository.get_statuses([summary.id for summary in summaries])

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)

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})

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]

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()

class Report(Model, ordering=("-created_at",)):
    ...

reports = await db_session().scalars(
    select(Report).where(Report.status != "closed").order_by(Report.created_at).limit(20)
)

uv run druks makemigrations night_watch -m "add reports"
uv run druks init-db

router = APIRouter(prefix="/reviews")

from druks.accounts import current_account_id

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

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")

(await Gmail.get()).secrets["client_secret"]   # raises ServiceNotConnectedError when unset
await Gmail.is_connected()

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

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']}"

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")

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

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

class NightWatch(App):
    name = "night_watch"
    acme = Acme.with_scopes("profile.read", "posts.write")

for connection in await NightWatch.acme.list_for_account(account_id):
    token = await connection.get_access_token()

from druks.core.services import Github

sign_ins = await Github.list_for_account(self.account_id)

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)

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 {}

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."},
        },
    },
)

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")

druks check-app field_notes

from druks.testing import run_workflow

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

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",
)

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")

class NightWatch(App):
    name = "night_watch"
    navigation = ["reports"]

@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)],
            )
        ],
    )

# 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}

# 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"),
                ),
            )
        ],
    )

ui.SecretField(name="token", label="Access token", help_text="From your account settings.")