diff --git a/backend/druks/accounts/constants.py b/backend/druks/accounts/constants.py index 4e0431bb..de2a95ed 100644 --- a/backend/druks/accounts/constants.py +++ b/backend/druks/accounts/constants.py @@ -5,6 +5,14 @@ # is the row's lookup key and the only part that may appear in errors, logs, or # lists; the secret exists only inside the copy-once plaintext. PAT_TOKEN_TAG = "druks_pat" +# A call-scoped operator credential serializes as druks_call_. It is +# Redis-only: the plaintext never sits in Settings, and revoke drops it when +# the agent call ends. +OPERATOR_TOKEN_TAG = "druks_call" +OPERATOR_TOKEN_PREFIX = "operator_token:" +OPERATOR_TOKEN_CALL_PREFIX = "operator_token_call:" +OPERATOR_DEFERRED_PREFIX = "operator_deferred:" +OPERATOR_WRITES = frozenset({"deny", "defer", "allow"}) PAT_NAME_LENGTH = 80 PAT_PREFIX_LENGTH = 12 # No separator characters, so the serialized token splits unambiguously on "_". diff --git a/backend/druks/accounts/dependencies.py b/backend/druks/accounts/dependencies.py index 1cd84872..65005d0b 100644 --- a/backend/druks/accounts/dependencies.py +++ b/backend/druks/accounts/dependencies.py @@ -11,7 +11,7 @@ InvalidPatError, ) from druks.accounts.jwt import verify_assertion -from druks.accounts.models import Account, PersonalAccessToken +from druks.accounts.models import Account, OperatorToken, PersonalAccessToken _BEARER_CHALLENGE = 'Bearer realm="druks"' # auto_error=False: absence and malformed both come back None — presence is @@ -24,7 +24,7 @@ async def resolve_pat_account(credentials: HTTPAuthorizationCredentials | None) """A present Authorization must authenticate — never a fall-through.""" if credentials: try: - return (await PersonalAccessToken.authenticate(credentials.credentials)).account + return await resolve_bearer_account(credentials.credentials) except InvalidPatError as error: raise HTTPException( status_code=401, @@ -38,6 +38,17 @@ async def resolve_pat_account(credentials: HTTPAuthorizationCredentials | None) ) +async def resolve_bearer_account(credential: str) -> Account: + """The one bearer door: a live call-scoped operator token, else a PAT.""" + operator = await OperatorToken.lookup(credential) + if operator: + account = await Account.get(operator.account_id) + if account: + return account + raise InvalidPatError("Not a recognized operator token.") + return (await PersonalAccessToken.authenticate(credential)).account + + async def resolve_single_operator() -> Account | None: """None while zero accounts exist (setup); more than one refuses rather than guesses.""" diff --git a/backend/druks/accounts/models.py b/backend/druks/accounts/models.py index b7c5536b..f8ec618b 100644 --- a/backend/druks/accounts/models.py +++ b/backend/druks/accounts/models.py @@ -1,14 +1,23 @@ import base64 import hashlib import hmac +import json import secrets +from dataclasses import dataclass from datetime import datetime +from typing import Any +import httpx2 from sqlalchemy import ForeignKey, Index, LargeBinary, String, select, text from sqlalchemy.dialects.postgresql import CITEXT, insert from sqlalchemy.orm import Mapped, mapped_column, relationship from druks.accounts.constants import ( + OPERATOR_DEFERRED_PREFIX, + OPERATOR_TOKEN_CALL_PREFIX, + OPERATOR_TOKEN_PREFIX, + OPERATOR_TOKEN_TAG, + OPERATOR_WRITES, PAT_LAST_USED_RESOLUTION, PAT_LIFETIME, PAT_NAME_LENGTH, @@ -18,9 +27,11 @@ PAT_TOKEN_TAG, ) from druks.accounts.exceptions import AuthConfigurationError, InvalidPatError -from druks.core.models import Uuid7Pk +from druks.core.models import Uuid7Pk, uuid7_str from druks.database import db_session from druks.models import Base +from druks.redis import get_client +from druks.sandbox.constants import MAX_AGENT_TIMEOUT_SECONDS from druks.settings import load_settings from druks.user_settings.models import InstallationSettings @@ -196,3 +207,127 @@ async def revoke(self) -> None: # Keep the first revocation instant — a repeat revoke changes nothing. self.revoked_at = self.revoked_at or Base.utc_now() await db_session().flush() + + +def _hash_operator_token(token: str) -> str: + return hashlib.sha256(token.encode()).hexdigest() + + +_operator_api = None + + +@dataclass(frozen=True) +class OperatorToken: + """A call-scoped bearer with PAT authority. Redis holds it for the agent + call; Settings never does. ``writes`` is deny (read tools only), defer + (stash mutating calls), or allow (execute as ``account_id``).""" + + account_id: str + agent_call_id: str + run_id: str + writes: str + + @classmethod + def bind_api(cls, api: Any) -> None: + global _operator_api + _operator_api = api + + @classmethod + async def mint( + cls, + *, + account_id: str, + agent_call_id: str, + run_id: str, + writes: str, + ) -> str: + if writes not in OPERATOR_WRITES: + raise ValueError( + f"operator token writes must be one of {sorted(OPERATOR_WRITES)}, not {writes!r}" + ) + token = f"{OPERATOR_TOKEN_TAG}_{secrets.token_urlsafe(32)}" + payload = json.dumps( + { + "account_id": account_id, + "agent_call_id": agent_call_id, + "run_id": run_id, + "writes": writes, + } + ) + digest = _hash_operator_token(token) + redis = get_client() + await redis.set(f"{OPERATOR_TOKEN_PREFIX}{digest}", payload, ex=MAX_AGENT_TIMEOUT_SECONDS) + await redis.set( + f"{OPERATOR_TOKEN_CALL_PREFIX}{agent_call_id}", digest, ex=MAX_AGENT_TIMEOUT_SECONDS + ) + return token + + @classmethod + async def lookup(cls, credential: str) -> "OperatorToken | None": + if not credential.startswith(f"{OPERATOR_TOKEN_TAG}_"): + return + raw = await get_client().get(f"{OPERATOR_TOKEN_PREFIX}{_hash_operator_token(credential)}") + if raw: + return cls(**json.loads(raw)) + return + + @classmethod + async def authenticate(cls, credential: str) -> "OperatorToken": + found = await cls.lookup(credential) + if found: + return found + raise InvalidPatError("Not a recognized operator token.") + + @classmethod + async def revoke(cls, agent_call_id: str) -> None: + redis = get_client() + call_key = f"{OPERATOR_TOKEN_CALL_PREFIX}{agent_call_id}" + digest = await redis.get(call_key) + if digest: + await redis.delete(f"{OPERATOR_TOKEN_PREFIX}{digest.decode()}", call_key) + + @classmethod + async def defer_write(cls, run_id: str, write: dict[str, str]) -> None: + redis = get_client() + key = f"{OPERATOR_DEFERRED_PREFIX}{run_id}" + await redis.rpush(key, json.dumps(write)) + await redis.expire(key, MAX_AGENT_TIMEOUT_SECONDS) + + @classmethod + async def take_deferred(cls, run_id: str) -> list[dict[str, str]]: + redis = get_client() + key = f"{OPERATOR_DEFERRED_PREFIX}{run_id}" + items = await redis.lrange(key, 0, -1) + await redis.delete(key) + return [json.loads(item) for item in items] + + @classmethod + async def play_deferred(cls, account_id: str, writes: list[dict[str, str]]) -> None: + if not _operator_api: + raise RuntimeError("operator token replay needs the API bound at MCP boot") + call_id = uuid7_str() + token = await cls.mint( + account_id=account_id, agent_call_id=call_id, run_id=call_id, writes="allow" + ) + try: + async with httpx2.AsyncClient( + transport=httpx2.ASGITransport(app=_operator_api, raise_app_exceptions=False), + base_url="http://druks", + ) as client: + for write in writes: + response = await client.request( + write["method"], + write["path"], + content=write["body"] or None, + headers={ + "Authorization": f"Bearer {token}", + "Content-Type": write["content_type"], + }, + ) + if response.status_code >= 400: + raise RuntimeError( + f"deferred {write['method']} {write['path']} failed: " + f"{response.status_code} {response.text}" + ) + finally: + await cls.revoke(call_id) diff --git a/backend/druks/api/server.py b/backend/druks/api/server.py index e7e5c019..197d7b2c 100644 --- a/backend/druks/api/server.py +++ b/backend/druks/api/server.py @@ -13,6 +13,7 @@ from druks.accounts.dependencies import current_account, resolve_single_operator from druks.accounts.exceptions import AuthConfigurationError +from druks.accounts.models import OperatorToken from druks.accounts.routes import router as auth_router from druks.api.artifacts import router as artifacts_router from druks.api.dashboard import router as dashboard_router @@ -317,6 +318,7 @@ async def _unhandled_exception_handler( # A bare Route at exactly /mcp (a Mount would 307 the no-slash path); PATs # authenticate it, so it sits outside the identity gate. mcp_app = create_mcp_app(app) +OperatorToken.bind_api(app) app.router.routes.append( Route("/mcp", mcp_app, methods=["POST", "DELETE"], include_in_schema=False) ) diff --git a/backend/druks/contrib/chat/__init__.py b/backend/druks/contrib/chat/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/druks/contrib/chat/app.py b/backend/druks/contrib/chat/app.py new file mode 100644 index 00000000..1507c8cd --- /dev/null +++ b/backend/druks/contrib/chat/app.py @@ -0,0 +1,61 @@ +import httpx + +from druks.agents import Agent +from druks.apps import App +from druks.contrib.chat.contracts import TurnOutput +from druks.doctor import CheckResult +from druks.settings import load_settings + + +async def check_appliance_mcp() -> CheckResult: + """Whether this appliance's /mcp answers, so a Talk sandbox can operate + as the signed-in operator. Skip when sandbox execution is off.""" + settings = load_settings() + if not settings.sandbox.service_url: + return CheckResult( + name="appliance_mcp", + ok=True, + detail="skipped — sandbox execution is off", + ) + endpoint = settings.urls.endpoint.rstrip("/") + if not endpoint: + return CheckResult( + name="appliance_mcp", + ok=False, + pending=True, + detail="urls.endpoint is unset — the sandbox needs it to reach /mcp.", + ) + url = f"{endpoint}/mcp" + try: + async with httpx.AsyncClient(timeout=5.0) as client: + response = await client.get(url) + except httpx.RequestError as error: + return CheckResult( + name="appliance_mcp", + ok=False, + detail=f"{url} is unreachable: {error}. Chat Talk needs it.", + ) + if response.status_code >= 500: + return CheckResult( + name="appliance_mcp", + ok=False, + detail=f"{url} returned {response.status_code}. Chat Talk needs it.", + ) + return CheckResult(name="appliance_mcp", ok=True, detail=url) + + +class Chat(App): + name = "chat" + icon = "message-square" + description = "Operator conversations this appliance owns — several threads, one account each." + # Every table this app owns carries the ``chat_`` prefix, so the thread's + # schema can never collide with core's or another app's. + prefix_tables = True + navigation = ["list"] + checks = [check_appliance_mcp] + + reply = Agent( + description="replies to the operator on one conversation turn", + prompt="chat/talk.md", + contract=TurnOutput, + ) diff --git a/backend/druks/contrib/chat/contracts.py b/backend/druks/contrib/chat/contracts.py new file mode 100644 index 00000000..c81e7414 --- /dev/null +++ b/backend/druks/contrib/chat/contracts.py @@ -0,0 +1,6 @@ +from druks.agents import AgentOutput + + +class TurnOutput(AgentOutput): + # What one chat turn returns: the assistant line to append. + text: str diff --git a/backend/druks/contrib/chat/enums.py b/backend/druks/contrib/chat/enums.py new file mode 100644 index 00000000..4b7898c8 --- /dev/null +++ b/backend/druks/contrib/chat/enums.py @@ -0,0 +1,19 @@ +from enum import StrEnum + + +class Autonomy(StrEnum): + """How far a conversation's next agent call may go. The setting is the + conversation's, not the turn's, so a mode change applies to the next call.""" + + PROPOSE = "propose" + CONFIRM = "confirm" + FULL = "full" + + +class Role(StrEnum): + """Who wrote a line on the thread. Closed: a column can never hold a speaker + no screen knows how to render.""" + + USER = "user" + ASSISTANT = "assistant" + SYSTEM = "system" diff --git a/backend/druks/contrib/chat/migrations/__init__.py b/backend/druks/contrib/chat/migrations/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/druks/contrib/chat/migrations/versions/__init__.py b/backend/druks/contrib/chat/migrations/versions/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/druks/contrib/chat/migrations/versions/chat_0001_conversations_messages.py b/backend/druks/contrib/chat/migrations/versions/chat_0001_conversations_messages.py new file mode 100644 index 00000000..562e1258 --- /dev/null +++ b/backend/druks/contrib/chat/migrations/versions/chat_0001_conversations_messages.py @@ -0,0 +1,46 @@ +"""chat: conversations and messages + +Revision ID: chat_0001 +Revises: +Create Date: 2026-09-04 00:00:00.000000 + +""" + +import sqlalchemy as sa +from alembic import op + +# This app owns an independent migration history — its own +# alembic_version_chat table, never linked to core's revisions. +revision = "chat_0001" +down_revision = None +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "chat_conversations", + # Integer subject key (StoredSubject.id) — serial, matching create_all. + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("account_id", sa.String(), nullable=False), + sa.Column("title", sa.String(), nullable=False), + sa.Column("autonomy", sa.String(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["account_id"], ["accounts.id"], ondelete="RESTRICT"), + sa.PrimaryKeyConstraint("id"), + ) + op.create_table( + "chat_messages", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("conversation_id", sa.Integer(), nullable=False), + sa.Column("role", sa.String(), nullable=False), + sa.Column("body", sa.String(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["conversation_id"], ["chat_conversations.id"]), + sa.PrimaryKeyConstraint("id"), + ) + + +def downgrade() -> None: + op.drop_table("chat_messages") + op.drop_table("chat_conversations") diff --git a/backend/druks/contrib/chat/models.py b/backend/druks/contrib/chat/models.py new file mode 100644 index 00000000..cd4f11bc --- /dev/null +++ b/backend/druks/contrib/chat/models.py @@ -0,0 +1,160 @@ +from collections.abc import Sequence +from datetime import datetime + +from sqlalchemy import ForeignKey, select +from sqlalchemy.orm import Mapped, mapped_column + +from druks.contrib.chat.enums import Autonomy, Role +from druks.contrib.chat.schemas import ConversationSummary +from druks.db import Base, StoredSubject, db_session + +_PROMPT_MESSAGES = 40 +_PROMPT_CHARS = 16_000 +_TITLE_CHARS = 80 + + +class Conversation(StoredSubject): + __tablename__ = "chat_conversations" + + # id: the integer subject key inherited from StoredSubject; the class name + # derives subject_type "conversation". + account_id: Mapped[str] = mapped_column(ForeignKey("accounts.id", ondelete="RESTRICT")) + title: Mapped[str] + # Autonomy is a String column driven by this app's closed StrEnum, not a + # native PG enum: the modes stay in code and a rename never needs ALTER TYPE. + autonomy: Mapped[str] = mapped_column(default=Autonomy.PROPOSE) + created_at: Mapped[datetime] = mapped_column(default=Base.utc_now) + + @classmethod + async def create( + cls, + *, + account_id: str, + title: str, + autonomy: Autonomy = Autonomy.PROPOSE, + ) -> "Conversation": + session = db_session() + conversation = cls(account_id=account_id, title=title, autonomy=autonomy) + session.add(conversation) + await session.flush() + return conversation + + @classmethod + async def get(cls, conversation_id: int) -> "Conversation | None": + return await db_session().get(cls, conversation_id) + + @classmethod + async def get_for_account( + cls, conversation_id: int, account_id: str | None + ) -> "Conversation | None": + """This account's thread, or nothing. Another operator's id is a miss, + not a leak.""" + conversation = await cls.get(conversation_id) + if conversation and conversation.account_id == account_id: + return conversation + return + + @classmethod + async def list_for_account(cls, account_id: str) -> list["Conversation"]: + """This account's threads, newest first. A conversation belongs to one + operator; another account's list never includes it.""" + statement = ( + select(cls) + .where(cls.account_id == account_id) + .order_by(cls.created_at.desc(), cls.id.desc()) + ) + return list(await db_session().scalars(statement)) + + async def add_message(self, *, role: Role, body: str) -> "Message": + return await Message.create(conversation_id=self.id, role=role, body=body) + + async def save_autonomy(self, autonomy: Autonomy) -> None: + self.autonomy = autonomy + await db_session().flush() + + async def name_from_first_line(self) -> None: + """Fill an empty title from the first user line. A title the operator + already set, or a prior call, is left alone.""" + if self.title: + return + first = next( + (message for message in await self.list_messages() if message.role == Role.USER), + None, + ) + if not first: + return + line = next( + (" ".join(raw.split()) for raw in first.body.splitlines() if raw.strip()), + "", + ) + if line: + self.title = line[:_TITLE_CHARS] + await db_session().flush() + + async def list_messages(self) -> list["Message"]: + return await Message.list_for_conversation(self.id) + + async def list_prompt_messages(self) -> list["Message"]: + return Message.bound_recent(await self.list_messages()) + + def get_summary(self) -> ConversationSummary: + return ConversationSummary.model_validate(self) + + @classmethod + async def list_summaries(cls, account_id: str | None) -> list[ConversationSummary]: + """This account's threads. A missing caller is not a shared board — + conversations are per-account, so the list is empty.""" + if account_id: + return [ + conversation.get_summary() + for conversation in await cls.list_for_account(account_id) + ] + return [] + + +class Message(Base): + __tablename__ = "chat_messages" + + # A row, not an event and not a StoredSubject: events stay facts about what + # happened, while a message is the thread the conversation reads back in + # order. Issues' ``Comment`` is the same shape. + id: Mapped[int] = mapped_column(primary_key=True) + conversation_id: Mapped[int] = mapped_column(ForeignKey("chat_conversations.id")) + role: Mapped[str] + body: Mapped[str] + created_at: Mapped[datetime] = mapped_column(default=Base.utc_now) + + @classmethod + async def create(cls, *, conversation_id: int, role: Role, body: str) -> "Message": + session = db_session() + message = cls(conversation_id=conversation_id, role=role, body=body) + session.add(message) + await session.flush() + return message + + @classmethod + async def list_for_conversation(cls, conversation_id: int) -> list["Message"]: + """The thread, oldest first — a conversation reads down. A conversation + nobody has spoken on is an empty list, never None.""" + statement = ( + select(cls) + .where(cls.conversation_id == conversation_id) + .order_by(cls.created_at, cls.id) + ) + return list(await db_session().scalars(statement)) + + @classmethod + def bound_recent(cls, messages: Sequence["Message"]) -> list["Message"]: + """Newest lines that fit the prompt. Older lines drop first. One + oversize last line still goes in — truncating it would hide the turn + the operator just sent.""" + chosen: list[Message] = [] + chars = 0 + for message in reversed(messages): + size = len(message.body) + if chosen and (len(chosen) >= _PROMPT_MESSAGES or chars + size > _PROMPT_CHARS): + break + chosen.append(message) + chars += size + chosen.reverse() + return chosen diff --git a/backend/druks/contrib/chat/pages.py b/backend/druks/contrib/chat/pages.py new file mode 100644 index 00000000..33e27628 --- /dev/null +++ b/backend/druks/contrib/chat/pages.py @@ -0,0 +1,143 @@ +from druks import ui +from druks.accounts import current_account_id +from druks.contrib.chat.enums import Autonomy, Role +from druks.contrib.chat.models import Conversation + + +@ui.page("/", label="Conversations") +async def list(): + threads = await Conversation.list_for_account(current_account_id.get()) + return ui.Page( + "Conversations", + controls=[ui.Link("New", page="new")], + blocks=[ + ui.Cards( + title="Threads", + cards=[ + ui.Card( + title=thread.title or thread.label, + controls=[ + ui.Link( + "Open", + page="thread", + arguments={"conversation_id": str(thread.id)}, + ) + ], + ) + for thread in threads + ], + empty=ui.EmptyState( + "No conversations yet", + description="Start one with a first message.", + controls=[ui.Link("New", page="new")], + ), + ) + ], + ) + + +@ui.page("/new") +async def new(): + return ui.Page( + "New conversation", + blocks=[ + ui.Form( + title="New conversation", + description="Title is optional. The first message starts Talk.", + fields=[ + ui.TextField(name="title", label="Title"), + ui.TextAreaField( + name="body", + label="Message", + is_required=True, + rows=4, + ), + ], + action=ui.Action( + label="Start", + operation="create_conversation", + tone="primary", + link=ui.Link("Conversations", page="list"), + ), + ) + ], + ) + + +@ui.page("/conversations/{conversation_id}") +async def thread(conversation_id: int): + conversation = await Conversation.get_for_account(conversation_id, current_account_id.get()) + if conversation: + status = await conversation.get_status() + if status.is_parked: + turn = [ui.GateControls(status.run)] + elif status.is_running: + turn = [ui.Text(status.agent or "The agent is running.")] + else: + turn = [] + cards = [] + for message in await conversation.list_messages(): + if message.role == Role.USER: + speaker = "You" + elif message.role == Role.ASSISTANT: + speaker = "Assistant" + else: + speaker = "System" + cards.append(ui.Card(title=speaker, blocks=[ui.Quote(message.body)])) + return ui.Page( + conversation.title or conversation.label, + controls=[ + ui.Link( + "Settings", + page="settings", + arguments={"conversation_id": str(conversation.id)}, + ), + ui.Link("This conversation", subject=conversation), + ], + blocks=[ + ui.Section( + name="thread", + follows=conversation, + blocks=[ + ui.Cards( + cards=cards, + empty=ui.EmptyState("No messages yet"), + ), + *turn, + ], + ) + ], + ) + return ui.Page( + f"Conversation {conversation_id}", + blocks=[ui.EmptyState("No such conversation")], + ) + + +@thread.child("/settings") +async def settings(conversation_id: int): + conversation = await Conversation.get_for_account(conversation_id, current_account_id.get()) + if conversation: + return ui.Page( + "Settings", + blocks=[ + ui.Form( + title="Autonomy", + fields=[ + ui.SelectField( + name="autonomy", + label="Autonomy", + value=conversation.autonomy, + options=[ui.Option(mode.capitalize(), value=mode) for mode in Autonomy], + is_required=True, + ) + ], + action=ui.Action( + label="Save", + operation="set_autonomy", + arguments={"conversation_id": conversation.id}, + ), + ) + ], + ) + return ui.Page("Settings", blocks=[ui.EmptyState("No such conversation")]) diff --git a/backend/druks/contrib/chat/routes.py b/backend/druks/contrib/chat/routes.py new file mode 100644 index 00000000..56df31c9 --- /dev/null +++ b/backend/druks/contrib/chat/routes.py @@ -0,0 +1,36 @@ +from typing import Annotated + +from fastapi import APIRouter, Body, Depends, HTTPException, status + +from druks.accounts.dependencies import current_account +from druks.accounts.models import Account +from druks.contrib.chat.enums import Autonomy, Role +from druks.contrib.chat.models import Conversation +from druks.contrib.chat.workflows import Talk + +router = APIRouter(prefix="/conversations") + + +@router.post("", status_code=status.HTTP_201_CREATED, operation_id="create_conversation") +async def create_conversation( + body: Annotated[str, Body(embed=True)], + account: Account = Depends(current_account), + title: Annotated[str, Body(embed=True)] = "", +) -> dict[str, int]: + conversation = await Conversation.create(account_id=account.id, title=title) + await conversation.add_message(role=Role.USER, body=body) + await Talk.dispatch(conversation=conversation) + return {"id": conversation.id} + + +@router.post("/{conversation_id}/autonomy", operation_id="set_autonomy") +async def set_autonomy( + conversation_id: int, + autonomy: Annotated[Autonomy, Body(embed=True)], + account: Account = Depends(current_account), +) -> dict[str, str]: + conversation = await Conversation.get_for_account(conversation_id, account.id) + if conversation: + await conversation.save_autonomy(autonomy) + return {"autonomy": conversation.autonomy} + raise HTTPException(status.HTTP_404_NOT_FOUND, f"No conversation {conversation_id}.") diff --git a/backend/druks/contrib/chat/schemas.py b/backend/druks/contrib/chat/schemas.py new file mode 100644 index 00000000..9b368d10 --- /dev/null +++ b/backend/druks/contrib/chat/schemas.py @@ -0,0 +1,12 @@ +from datetime import datetime + +from druks.contrib.chat.enums import Autonomy +from druks.workflows import SubjectSummary + + +class ConversationSummary(SubjectSummary): + # The conversation's domain header — title and autonomy are this app's; + # status and the timeline come from the platform's subject read-side. + title: str + autonomy: Autonomy + created_at: datetime diff --git a/backend/druks/contrib/chat/templates/talk.md b/backend/druks/contrib/chat/templates/talk.md new file mode 100644 index 00000000..137ac825 --- /dev/null +++ b/backend/druks/contrib/chat/templates/talk.md @@ -0,0 +1,9 @@ +Act on behalf of the signed-in operator. Use the attached MCP tools to inspect and change this appliance. + +Discover work with list_open_subjects. Echo parkedAt from get_gate unchanged when you answer a gate. Do not invent run ids. + +Autonomy: {{ autonomy }} + +{% for message in messages %} +{{ message.role }}: {{ message.body }} +{% endfor %} diff --git a/backend/druks/contrib/chat/workflows.py b/backend/druks/contrib/chat/workflows.py new file mode 100644 index 00000000..2e35fce3 --- /dev/null +++ b/backend/druks/contrib/chat/workflows.py @@ -0,0 +1,162 @@ +from dataclasses import dataclass +from datetime import timedelta +from typing import TYPE_CHECKING, Any, Literal + +from druks.accounts.models import OperatorToken +from druks.contrib.chat.app import Chat +from druks.contrib.chat.enums import Role +from druks.contrib.chat.models import Conversation +from druks.mcp.constants import THIS_APPLIANCE +from druks.mcp.helpers import get_bearer_token_env_var +from druks.sandbox.datastructures import McpServer +from druks.workflows import FatalError, Gate, Workflow, step +from druks.workspaces import Workspace, this_appliance_mcp_url + +if TYPE_CHECKING: + from druks.sandbox.host import Host + +_WRITES = { + "propose": "deny", + "confirm": "defer", + "full": "allow", +} + + +class ChatTurn(Gate): + """The operator's next line, or a stop. Not ``review()`` — those controls + would offer approve/request_changes on a chat turn.""" + + name = "chat_turn" + action: Literal["send", "stop"] + note: str = "" + + +class ConfirmTool(Gate): + """The operator approves or skips the mutating MCP call the agent proposed. + Parks after the agent step and before the next ChatTurn — a run holds one + gate at a time.""" + + name = "confirm_tool" + action: Literal["approve", "reject"] + + +@dataclass(frozen=True, kw_only=True) +class TalkWorkspace(Workspace): + account_id: str + run_id: str + writes: str + + async def with_mcp_servers(self, account_id: str | None, **kwargs: Any) -> dict[str, Any]: + # Call-scoped: minted here, not a vault row. The box cannot hold it — + # it dies with the agent call. + kwargs = await super().with_mcp_servers(account_id, **kwargs) + token = await OperatorToken.mint( + account_id=self.account_id, + agent_call_id=kwargs["call_id"], + run_id=self.run_id, + writes=self.writes, + ) + variable = get_bearer_token_env_var(THIS_APPLIANCE) + servers = [ + server for server in kwargs.get("mcp_servers") or () if server.name != THIS_APPLIANCE + ] + servers.append( + McpServer( + name=THIS_APPLIANCE, + url=this_appliance_mcp_url(self.host), + bearer_token_env_var=variable, + ) + ) + kwargs["mcp_servers"] = tuple(servers) + env = dict(kwargs.get("extra_env") or {}) + env[variable] = token + kwargs["extra_env"] = env + return kwargs + + async def run_agent(self, *, account_id: str | None, **kwargs: Any): + try: + return await super().run_agent(account_id=account_id, **kwargs) + finally: + await OperatorToken.revoke(kwargs["call_id"]) + + +class Talk(Workflow): + """One conversation: agent reply, park, operator line, repeat until stop.""" + + subject = Conversation + steps_reuse_sandbox = True + sandbox_hold = timedelta(minutes=15) + workspace_class = TalkWorkspace + + async def run_multistep(self) -> None: + while True: + conversation = await self.subject + messages = await conversation.list_prompt_messages() + result = await Chat.reply( + autonomy=conversation.autonomy, + messages=[{"role": message.role, "body": message.body} for message in messages], + ) + await self.record_message(Role.ASSISTANT, result.text) + await self.name_thread() + deferred = await self.deferred_writes() + if deferred: + decision = await ConfirmTool.wait( + input_request={ + "presentation": "in_app", + "label": "Confirm the proposed action", + "controls": ["approve", "reject"], + "questions": [], + }, + hold_sandbox=self.sandbox_hold, + ) + if decision.action == "approve": + await self.apply_deferred_writes(deferred) + reply = await ChatTurn.wait( + input_request={ + "presentation": "in_app", + "label": "Message", + "controls": ["send", "stop"], + "questions": [], + }, + hold_sandbox=self.sandbox_hold, + ) + if reply.action == "stop": + return + await self.record_message(Role.USER, reply.note) + + async def get_workspace_kwargs(self, host: "Host") -> dict[str, Any]: + conversation = await self.subject + if not self.account_id: + raise FatalError("Talk runs as the operator who started the conversation.") + return { + **await super().get_workspace_kwargs(host), + "account_id": self.account_id, + "run_id": self.workflow_id, + "writes": _WRITES[conversation.autonomy], + } + + @step + async def record_message(self, role: Role, body: str) -> None: + conversation = await self.subject + await conversation.add_message(role=role, body=body) + + @step + async def name_thread(self) -> None: + conversation = await self.subject + await conversation.name_from_first_line() + + @step + async def deferred_writes(self) -> list[dict[str, str]]: + return await OperatorToken.take_deferred(self.workflow_id) + + @step + async def apply_deferred_writes(self, writes: list[dict[str, str]]) -> None: + if not self.account_id: + raise FatalError("Talk runs as the operator who started the conversation.") + await OperatorToken.play_deferred(self.account_id, writes) + + @classmethod + async def dispatch(cls, *, conversation: Conversation) -> str: + # One Talk per conversation: start() already dedups live runs on the + # subject, so a later line answers ChatTurn instead of starting again. + return await cls.start(subject=conversation) diff --git a/backend/druks/doctor.py b/backend/druks/doctor.py index 64246a8f..40ff303b 100644 --- a/backend/druks/doctor.py +++ b/backend/druks/doctor.py @@ -529,6 +529,20 @@ def check_capability_modules(settings: Settings) -> CheckResult: ) +def check_chat(settings: Settings) -> CheckResult: + """Chat ships in this distribution. A missing roster entry is a + packaging fault, not an optional install.""" + if any(app.name == "chat" for app in iter_apps()): + return CheckResult(name="chat", ok=True, detail="bundled") + return CheckResult( + name="chat", + ok=False, + detail=( + "chat is missing from the app roster — it ships with Druks, not as an optional package." + ), + ) + + async def check_apps(settings: Settings) -> list[CheckResult]: """Each installed app's resolved settings and own checks, namespaced under it. Read off the class headlessly through the loader, so doctor never imports an @@ -598,6 +612,7 @@ async def _run_app_check(app_name: str, check) -> CheckResult: check_drukbox, check_secrets_exchange, check_capability_modules, + check_chat, check_apps, check_declared_sandboxes, ) diff --git a/backend/druks/mcp/constants.py b/backend/druks/mcp/constants.py index 30195666..f2d05be5 100644 --- a/backend/druks/mcp/constants.py +++ b/backend/druks/mcp/constants.py @@ -4,6 +4,9 @@ # the var name ever lands in emitted config, never the value. TOKEN_ENV_PREFIX = "MCP_" TOKEN_ENV_SUFFIX = "_TOKEN" +# The name a workspace uses when it injects this appliance's /mcp into a +# sandbox — one config key, one bearer env var. +THIS_APPLIANCE = "druks" # A server name is one identifier reused as the MCP config key (a bare TOML path # segment for codex, a JSON object key for claude) and the stem of the bearer env diff --git a/backend/druks/mcp/server.py b/backend/druks/mcp/server.py index 74dfddf3..8486fe9b 100644 --- a/backend/druks/mcp/server.py +++ b/backend/druks/mcp/server.py @@ -3,21 +3,24 @@ # operation's single declaration — schema, docstring, operation_id — and a # tagged app route joins the surface the same way. import inspect -from collections.abc import Generator +from collections.abc import Generator, Sequence import httpx2 from fastapi import FastAPI from fastapi.routing import APIRoute, iter_route_contexts from fastmcp import FastMCP from fastmcp.server.auth import AccessToken, TokenVerifier -from fastmcp.server.dependencies import get_http_request +from fastmcp.server.dependencies import get_access_token, get_http_request from fastmcp.server.http import StarletteWithLifespan from fastmcp.server.providers.openapi import MCPType, OpenAPIProvider, OpenAPITool, RouteMap +from fastmcp.server.transforms import GetToolNext, Transform +from fastmcp.tools.base import Tool from fastmcp.utilities.openapi import HTTPRoute +from fastmcp.utilities.versions import VersionSpec from mcp.types import ToolAnnotations from druks.accounts.exceptions import InvalidPatError -from druks.accounts.models import PersonalAccessToken +from druks.accounts.models import OperatorToken, PersonalAccessToken from druks.apps.loader import iter_apps from druks.database import db_session from druks.mcp.exceptions import InvalidAgentToolError @@ -41,14 +44,28 @@ class PatTokenVerifier(TokenVerifier): async def verify_token(self, token: str) -> AccessToken | None: # Auth middleware runs outside the request session boundary, so this - # owns one — authenticate stamps last_used_at. + # owns one — authenticate stamps last_used_at on a PAT; a call token + # is Redis-only and needs no commit. try: + operator = await OperatorToken.lookup(token) + if operator: + return AccessToken( + token=token, + client_id=operator.agent_call_id, + scopes=[], + claims={ + "account_id": operator.account_id, + "agent_call_id": operator.agent_call_id, + "run_id": operator.run_id, + "writes": operator.writes, + }, + ) pat = await PersonalAccessToken.authenticate(token) access = AccessToken( token=token, client_id=pat.token_prefix, scopes=[], - claims={"account_id": pat.account_id, "pat_id": pat.id}, + claims={"account_id": pat.account_id, "pat_id": pat.id, "writes": "allow"}, ) await db_session().commit() return access @@ -75,6 +92,87 @@ def auth_flow( yield request +def _writes_claim() -> str: + token = get_access_token() + if token: + return token.claims.get("writes", "allow") + return "allow" + + +def _is_read_tool(tool: Tool) -> bool: + return bool(tool.annotations and tool.annotations.read_only_hint) + + +class OperatorWritesFilter(Transform): + """Propose (writes=deny) lists only GET-derived tools. Confirm and full + keep the live catalog; mutating calls are intercepted on the HTTP hop.""" + + async def list_tools(self, tools: Sequence[Tool]) -> Sequence[Tool]: + if _writes_claim() == "deny": + return [tool for tool in tools if _is_read_tool(tool)] + return tools + + async def get_tool( + self, name: str, call_next: GetToolNext, *, version: VersionSpec | None = None + ) -> Tool | None: + tool = await call_next(name, version=version) + if tool and _writes_claim() == "deny" and not _is_read_tool(tool): + return + return tool + + +def _bearer_credential(header: str | None) -> str: + if header: + scheme, _, credential = header.partition(" ") + if scheme.lower() == "bearer" and credential: + return credential + return "" + + +class OperatorWritesTransport(httpx2.AsyncBaseTransport): + """Mutating OpenAPI hops: deny 403s, defer stashes, allow passes through.""" + + def __init__(self, inner: httpx2.AsyncBaseTransport): + self._inner = inner + + async def handle_async_request(self, request: httpx2.Request) -> httpx2.Response: + if request.method.upper() != "GET": + operator = await OperatorToken.lookup( + _bearer_credential(request.headers.get("authorization")) + ) + if operator and operator.writes == "deny": + return httpx2.Response( + 403, + json={ + "code": "AUTONOMY_READ_ONLY", + "message": ( + "This conversation's autonomy is propose; mutating tools do not run." + ), + "retryable": False, + }, + request=request, + ) + if operator and operator.writes == "defer": + await OperatorToken.defer_write( + operator.run_id, + { + "method": request.method, + "path": request.url.path, + "body": request.content.decode() if request.content else "", + "content_type": request.headers.get("content-type") or "application/json", + }, + ) + return httpx2.Response( + 200, + json={ + "result": "deferred", + "message": "Proposed. The operator must confirm before this runs.", + }, + request=request, + ) + return await self._inner.handle_async_request(request) + + def _validate_agent_tools(api: FastAPI) -> None: # The provider logs component-fn errors instead of raising, so derived tools # cannot refuse boot; validate the routes before derivation. Inclusion is @@ -160,6 +258,11 @@ def _annotate(route: HTTPRoute, component: object) -> None: destructive_hint=not is_read and route.extensions.get("x-destructive", True), idempotent_hint=route.extensions.get("x-idempotent", False), ) + # Confirm (writes=defer) intercepts mutating hops with a deferred stub, + # not the route's 201 model. Advertising that model as outputSchema + # makes MCP reject the stub (`identifier` required on create_ticket). + if not is_read: + component.output_schema = None def create_mcp_app(api: FastAPI) -> StarletteWithLifespan: @@ -169,7 +272,9 @@ def create_mcp_app(api: FastAPI) -> StarletteWithLifespan: # raise_app_exceptions=False makes an app crash reach the tool as the # app's sanitized 500, so no masking is needed and the taxonomy travels. client = httpx2.AsyncClient( - transport=httpx2.ASGITransport(app=api, raise_app_exceptions=False), + transport=OperatorWritesTransport( + httpx2.ASGITransport(app=api, raise_app_exceptions=False) + ), base_url="http://druks", auth=CallerPat(), ) @@ -185,6 +290,7 @@ def create_mcp_app(api: FastAPI) -> StarletteWithLifespan: server = FastMCP( name="druks", providers=[provider], + transforms=[OperatorWritesFilter()], instructions=_INSTRUCTIONS, auth=PatTokenVerifier(), ) diff --git a/backend/druks/sandbox/client.py b/backend/druks/sandbox/client.py index fa187b5e..3bbe0592 100644 --- a/backend/druks/sandbox/client.py +++ b/backend/druks/sandbox/client.py @@ -319,6 +319,23 @@ async def request_refreshes(self, secret_id: str, *, except_host_id: str = "") - "refresh request for box %s service %s failed: %s", host_id, service, answer ) + async def set_expiry(self, *, host_id: str, expires_at: datetime) -> None: + """Move a host's lease to ``expires_at`` without touching the VM. + + Clipping the expiry down is how an idle hold ends: the host keeps + running until the new expiry, and drukbox's janitor reaps it then, + so druks needs no reconciler of its own. ``release`` remains the + hard delete for callers that want the VM gone now. + + A host the control plane no longer knows about raises the SDK's + ``SandboxNotFoundError`` — it is not swallowed here. + """ + api = self._api() + try: + await api.renew_host(host_id, expires_at=expires_at) + finally: + await api.aclose() + async def release(self, *, host_id: str) -> None: """Terminate the VM. Idempotent and infallible — already-gone hosts no-op silently; any other failure is logged but not surfaced so diff --git a/backend/druks/workflows.py b/backend/druks/workflows.py index 35ad61a8..9725aab6 100644 --- a/backend/druks/workflows.py +++ b/backend/druks/workflows.py @@ -2,7 +2,7 @@ from collections.abc import Awaitable, Callable from contextlib import nullcontext, suppress from contextvars import ContextVar -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from functools import partial from typing import ( TYPE_CHECKING, @@ -290,13 +290,20 @@ async def answer(cls, subject: Subject | StoredSubject, **reply: Any) -> None: @classmethod async def wait( - cls, *, input_request: dict[str, Any] | None = None, ttl_seconds: float = GATE_TTL_SECONDS + cls, + *, + input_request: dict[str, Any] | None = None, + ttl_seconds: float = GATE_TTL_SECONDS, + hold_sandbox: bool | timedelta | None = False, ) -> Self: # Suspend the running workflow until its gate is answered. A gate is a # run-level state — the read surfaces "needs you" straight off the parked run. # ``input_request`` is the plain-dict ask (at least a ``label`` and # ``presentation``), stored on the run beside ``input_gate`` and cleared on # resume — so an app declares the ask here, beside on_wait, not at read time. + # ``hold_sandbox`` keeps the warm VM across the park (see ``_hold_host``): + # ``True`` holds it for as long as its lease could still cover one more + # worst-case agent call, a timedelta for at most that long. workflow = current_workflow.get() if not workflow._subject and cls.on_wait.__func__ is Gate.on_wait.__func__: # No subject means no feed surface; if on_wait wasn't overridden @@ -311,7 +318,9 @@ async def _on_wait() -> None: await cls.on_wait(workflow) await DBOS.run_step_async(StepOptions(name=f"{cls.name}._on_wait"), _on_wait) - payload = await _park(workflow, cls.name, input_request, ttl_seconds) + payload = await _park( + workflow, cls.name, input_request, ttl_seconds, hold_sandbox=hold_sandbox + ) reply = cls.model_validate(payload) workflow.journal.add(reply) return reply @@ -333,10 +342,15 @@ async def _park( gate: str, input_request: dict[str, Any] | None, ttl_seconds: float, + hold_sandbox: bool | timedelta | None = False, ) -> dict[str, Any]: # Shared park core: a park lasts days, so reap the warm VM, then suspend on the - # gate's channel until Run.resume answers it. - await workflow._reap_run() + # gate's channel until Run.resume answers it. A caller that expects a quick + # answer can hold the VM instead — the clipped lease is what ends the hold. + if hold_sandbox: + await workflow._hold_host(hold_sandbox) + else: + await workflow._reap_run() await _emit_run_event( workflow.workflow_id, RunState.PARKED, @@ -899,6 +913,25 @@ async def _reap_run(self) -> None: host, self._host = self._host, None await sandbox_client.release(host_id=host.id) + async def _hold_host(self, hold: bool | timedelta) -> None: + # Keep the warm VM across a park instead of reaping it, by clipping its lease + # down: drukbox reaps at the new expiry, so a hold nobody ever answers still + # frees the VM with no druks-side sweep. ``True`` holds it for as long as the + # lease could still cover one more worst-case call — past that the next call + # would rotate anyway. Never extends: the lease drukbox already granted is the + # ceiling. A run with no warm host has nothing to hold. + if not self._host: + return + span = ( + hold + if isinstance(hold, timedelta) + else timedelta(seconds=SANDBOX_HOST_ROTATE_BEFORE_SECONDS) + ) + expires_at = datetime.now(UTC) + span + if self._host.expires_at: + expires_at = min(self._host.expires_at, expires_at) + await sandbox_client.set_expiry(host_id=self._host.id, expires_at=expires_at) + @property def workflow_id(self) -> str: return self._workflow_id diff --git a/backend/druks/workspaces.py b/backend/druks/workspaces.py index c660b14b..dd394f14 100644 --- a/backend/druks/workspaces.py +++ b/backend/druks/workspaces.py @@ -5,7 +5,7 @@ from dataclasses import dataclass from pathlib import Path, PurePosixPath from typing import TYPE_CHECKING, Any, ClassVar -from urllib.parse import urlsplit +from urllib.parse import urlparse, urlsplit, urlunparse from druks.accounts.models import Account from druks.core.apis.github import get_github_client @@ -28,11 +28,30 @@ from druks.sandbox.exceptions import ExecFailed from druks.sandbox.layout import get_repo_root, get_work_root from druks.sandbox.models import SecretRef +from druks.settings import load_settings if TYPE_CHECKING: from druks.sandbox.host import Host +def this_appliance_mcp_url(host: "Host") -> str: + """The /mcp hop a sandbox uses to reach this process. + + Docker sibling containers cannot use the host loopback; the engine + publishes that address as host.docker.internal:8001. An exe VM is a + different machine and uses the dashboard URL (urls.endpoint), never + webhook_host. + """ + base = (load_settings().urls.endpoint or "http://127.0.0.1:8001").rstrip("/") + parsed = urlparse(base) + if host.record.provider == "docker" and parsed.hostname in {"127.0.0.1", "localhost", "::1"}: + port = parsed.port + if not port: + port = 8001 if parsed.scheme == "http" else 443 + base = urlunparse(parsed._replace(netloc=f"host.docker.internal:{port}")).rstrip("/") + return f"{base}/mcp" + + @dataclass(frozen=True) class Workspace: # What an agent runs in: the VM it abstracts. diff --git a/backend/tests/chat/__init__.py b/backend/tests/chat/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/backend/tests/chat/test_doctor.py b/backend/tests/chat/test_doctor.py new file mode 100644 index 00000000..8c0cb528 --- /dev/null +++ b/backend/tests/chat/test_doctor.py @@ -0,0 +1,95 @@ +from types import SimpleNamespace +from unittest.mock import MagicMock + +import httpx +from druks import doctor +from druks.contrib.chat.app import check_appliance_mcp +from druks.testing import make_settings + + +def test_chat_is_in_the_roster(tmp_path): + result = doctor.check_chat(make_settings(tmp_path)) + + assert result.ok + assert result.detail == "bundled" + + +def test_chat_missing_from_the_roster_is_a_fault(tmp_path, monkeypatch): + monkeypatch.setattr(doctor, "iter_apps", lambda: iter(())) + + result = doctor.check_chat(make_settings(tmp_path)) + + assert not result.ok + assert not result.pending + assert "optional" in result.detail + + +def test_check_chat_is_in_the_battery(): + assert doctor.check_chat in doctor.CHECKS + + +def _settings(*, service_url: str, endpoint: str): + return SimpleNamespace( + sandbox=SimpleNamespace(service_url=service_url), + urls=SimpleNamespace(endpoint=endpoint), + ) + + +async def test_appliance_mcp_skips_when_sandbox_execution_is_off(monkeypatch): + monkeypatch.setattr( + "druks.contrib.chat.app.load_settings", + lambda: _settings(service_url="", endpoint="http://127.0.0.1:8001"), + ) + + result = await check_appliance_mcp() + + assert result.ok + assert result.detail.startswith("skipped") + + +async def test_appliance_mcp_pends_without_an_endpoint(monkeypatch): + monkeypatch.setattr( + "druks.contrib.chat.app.load_settings", + lambda: _settings(service_url="http://127.0.0.1:8780", endpoint=""), + ) + + result = await check_appliance_mcp() + + assert not result.ok + assert result.pending + assert "/mcp" in result.detail + + +async def test_appliance_mcp_names_an_unreachable_url(monkeypatch): + monkeypatch.setattr( + "druks.contrib.chat.app.load_settings", + lambda: _settings(service_url="http://127.0.0.1:8780", endpoint="http://druks.test:8001"), + ) + + async def fake_get(self, url): + raise httpx.ConnectError("connection refused", request=httpx.Request("GET", url)) + + monkeypatch.setattr(httpx.AsyncClient, "get", fake_get) + + result = await check_appliance_mcp() + + assert not result.ok + assert not result.pending + assert "http://druks.test:8001/mcp" in result.detail + + +async def test_appliance_mcp_accepts_a_live_endpoint(monkeypatch): + monkeypatch.setattr( + "druks.contrib.chat.app.load_settings", + lambda: _settings(service_url="http://127.0.0.1:8780", endpoint="http://127.0.0.1:8001"), + ) + + async def fake_get(self, url): + return MagicMock(status_code=401) + + monkeypatch.setattr(httpx.AsyncClient, "get", fake_get) + + result = await check_appliance_mcp() + + assert result.ok + assert result.detail == "http://127.0.0.1:8001/mcp" diff --git a/backend/tests/chat/test_models.py b/backend/tests/chat/test_models.py new file mode 100644 index 00000000..d528763e --- /dev/null +++ b/backend/tests/chat/test_models.py @@ -0,0 +1,143 @@ +from druks.accounts.models import Account +from druks.apps.loader import get_app +from druks.contrib.chat.enums import Autonomy, Role +from druks.contrib.chat.models import Conversation, Message + + +def test_chat_app_is_bundled_with_prefixed_tables(): + app = get_app("chat") + assert app.name == "chat" + assert app.prefix_tables is True + assert app.table_prefix == "chat_" + + +async def test_list_for_account_excludes_another_operators_threads(): + owner = await Account.get_or_create("op@example.com") + other = await Account.get_or_create("dev@example.com") + mine = await Conversation.create(account_id=owner.id, title="mine") + await Conversation.create(account_id=other.id, title="theirs") + + listed = await Conversation.list_for_account(owner.id) + assert [conversation.id for conversation in listed] == [mine.id] + assert listed[0].title == "mine" + assert listed[0].autonomy == Autonomy.PROPOSE + assert listed[0].account_id == owner.id + + +async def test_conversations_list_newest_first(): + account = await Account.get_or_create("op@example.com") + first = await Conversation.create(account_id=account.id, title="first") + second = await Conversation.create( + account_id=account.id, title="second", autonomy=Autonomy.FULL + ) + + listed = await Conversation.list_for_account(account.id) + assert [conversation.id for conversation in listed] == [second.id, first.id] + assert listed[0].autonomy == Autonomy.FULL + + +async def test_messages_are_rows_and_empty_is_a_list(): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="quiet") + + assert await conversation.list_messages() == [] + + first = await conversation.add_message(role=Role.USER, body="hello") + second = await conversation.add_message(role=Role.ASSISTANT, body="hi") + listed = await conversation.list_messages() + assert [message.body for message in listed] == ["hello", "hi"] + assert [message.role for message in listed] == [Role.USER, Role.ASSISTANT] + assert listed[0].id == first.id + assert listed[1].id == second.id + assert all(isinstance(message, Message) for message in listed) + + +async def test_list_summaries_is_this_accounts_threads(): + owner = await Account.get_or_create("op@example.com") + other = await Account.get_or_create("dev@example.com") + mine = await Conversation.create(account_id=owner.id, title="mine") + await Conversation.create(account_id=other.id, title="theirs") + + listed = await Conversation.list_summaries(owner.id) + assert [summary.id for summary in listed] == [str(mine.id)] + assert listed[0].title == "mine" + assert listed[0].autonomy == Autonomy.PROPOSE + assert listed[0].label == mine.label + assert await Conversation.list_summaries(None) == [] + + +async def test_get_for_account_misses_another_operators_thread(): + owner = await Account.get_or_create("op@example.com") + other = await Account.get_or_create("dev@example.com") + mine = await Conversation.create(account_id=owner.id, title="mine") + theirs = await Conversation.create(account_id=other.id, title="theirs") + + assert (await Conversation.get_for_account(mine.id, owner.id)).id == mine.id + assert await Conversation.get_for_account(theirs.id, owner.id) is None + assert await Conversation.get_for_account(mine.id, None) is None + + +async def test_name_from_first_line_fills_an_empty_title(): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="") + await conversation.add_message(role=Role.USER, body=" Pump the tires \nmore detail") + await conversation.add_message(role=Role.ASSISTANT, body="ok") + + await conversation.name_from_first_line() + + assert conversation.title == "Pump the tires" + + +async def test_name_from_first_line_leaves_an_operator_title(): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="Pump") + await conversation.add_message(role=Role.USER, body="something else") + + await conversation.name_from_first_line() + + assert conversation.title == "Pump" + + +async def test_name_from_first_line_caps_a_long_first_line(): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="") + await conversation.add_message(role=Role.USER, body="x" * 100) + + await conversation.name_from_first_line() + + assert conversation.title == "x" * 80 + + +async def test_name_from_first_line_is_idempotent(): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="") + await conversation.add_message(role=Role.USER, body="first") + await conversation.name_from_first_line() + await conversation.add_message(role=Role.USER, body="later") + + await conversation.name_from_first_line() + + assert conversation.title == "first" + + +async def test_prompt_history_keeps_the_newest_lines(): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="t") + for index in range(41): + await conversation.add_message(role=Role.USER, body=f"m{index}") + + prompt = await conversation.list_prompt_messages() + + assert [message.body for message in prompt] == [f"m{index}" for index in range(1, 41)] + assert [message.body for message in await conversation.list_messages()][0] == "m0" + + +async def test_prompt_history_drops_older_lines_that_overflow_the_char_budget(): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="t") + await conversation.add_message(role=Role.USER, body="a" * 10_000) + await conversation.add_message(role=Role.ASSISTANT, body="b" * 10_000) + + prompt = await conversation.list_prompt_messages() + + assert [message.body[0] for message in prompt] == ["b"] diff --git a/backend/tests/chat/test_operator_mcp.py b/backend/tests/chat/test_operator_mcp.py new file mode 100644 index 00000000..06d25c8e --- /dev/null +++ b/backend/tests/chat/test_operator_mcp.py @@ -0,0 +1,92 @@ +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +from druks.accounts.models import Account, OperatorToken +from druks.contrib.chat.enums import Autonomy +from druks.contrib.chat.models import Conversation +from druks.contrib.chat.workflows import Talk, TalkWorkspace +from druks.mcp.constants import THIS_APPLIANCE +from druks.mcp.helpers import get_bearer_token_env_var +from druks.workspaces import this_appliance_mcp_url + + +class _FakeHost: + ssh_username = "exedev" + + def __init__(self, provider: str = "docker"): + self.record = SimpleNamespace(provider=provider) + self.run_agent = AsyncMock(return_value="ok") + + +def test_docker_sandbox_uses_host_docker_internal_for_loopback(monkeypatch): + settings = MagicMock() + settings.urls.endpoint = "http://127.0.0.1:8001" + monkeypatch.setattr("druks.workspaces.load_settings", lambda: settings) + + assert this_appliance_mcp_url(_FakeHost("docker")) == "http://host.docker.internal:8001/mcp" + assert this_appliance_mcp_url(_FakeHost("exe.dev")) == "http://127.0.0.1:8001/mcp" + + +def test_appliance_mcp_url_never_uses_webhook_host(monkeypatch): + settings = MagicMock() + settings.urls.endpoint = "https://druks.example.com" + settings.urls.webhook_host = "hooks.example.com" + monkeypatch.setattr("druks.workspaces.load_settings", lambda: settings) + + assert this_appliance_mcp_url(_FakeHost("docker")) == "https://druks.example.com/mcp" + + +async def test_talk_workspace_injects_this_appliance_mcp(druks_db, monkeypatch): + account = await Account.get_or_create("op@example.com") + settings = MagicMock() + settings.urls.endpoint = "http://127.0.0.1:8001" + monkeypatch.setattr("druks.workspaces.load_settings", lambda: settings) + workspace = TalkWorkspace( + host=_FakeHost("docker"), # type: ignore[arg-type] + account_id=account.id, + run_id="run-1", + writes="deny", + ) + kwargs = await workspace.with_mcp_servers(account.id, call_id="call-1") + server = next(s for s in kwargs["mcp_servers"] if s.name == THIS_APPLIANCE) + token = kwargs["extra_env"][get_bearer_token_env_var(THIS_APPLIANCE)] + + assert server.url == "http://host.docker.internal:8001/mcp" + found = await OperatorToken.lookup(token) + assert found.account_id == account.id + assert found.writes == "deny" + assert found.agent_call_id == "call-1" + + +async def test_talk_workspace_revokes_the_token_after_the_call(druks_db, monkeypatch): + account = await Account.get_or_create("op@example.com") + settings = MagicMock() + settings.urls.endpoint = "http://127.0.0.1:8001" + monkeypatch.setattr("druks.workspaces.load_settings", lambda: settings) + host = _FakeHost("exe.dev") + workspace = TalkWorkspace( + host=host, # type: ignore[arg-type] + account_id=account.id, + run_id="run-1", + writes="allow", + ) + + await workspace.run_agent(account_id=account.id, call_id="call-done") + token = host.run_agent.await_args.kwargs["extra_env"][get_bearer_token_env_var(THIS_APPLIANCE)] + assert await OperatorToken.lookup(token) is None + + +async def test_talk_reads_live_autonomy_for_the_next_call(druks_db): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="") + flow = Talk() + flow.subject = conversation + flow.account_id = account.id + flow._workflow_id = "run-1" + + kwargs = await flow.get_workspace_kwargs(_FakeHost()) # type: ignore[arg-type] + assert kwargs["writes"] == "deny" + + conversation.autonomy = Autonomy.FULL + kwargs = await flow.get_workspace_kwargs(_FakeHost()) # type: ignore[arg-type] + assert kwargs["writes"] == "allow" diff --git a/backend/tests/chat/test_pages.py b/backend/tests/chat/test_pages.py new file mode 100644 index 00000000..75c33c73 --- /dev/null +++ b/backend/tests/chat/test_pages.py @@ -0,0 +1,159 @@ +from datetime import UTC, datetime + +from druks.accounts.models import Account +from druks.contrib.chat.enums import Role +from druks.contrib.chat.models import Conversation +from druks.contrib.chat.workflows import ChatTurn, Talk +from druks.testing import seed_run + + +async def test_the_roster_names_chat_pages(druks_client): + roster = {entry["name"]: entry for entry in (await druks_client.get("/api/apps")).json()} + pages = roster["chat"]["pages"] + by_name = {page["name"]: page for page in pages} + assert set(by_name) == {"list", "new", "thread", "settings"} + assert by_name["list"]["path"] == "/chat" + assert by_name["new"]["path"] == "/chat/new" + assert by_name["thread"]["path"] == "/chat/conversations/{conversation_id}" + assert by_name["settings"]["path"] == "/chat/conversations/{conversation_id}/settings" + assert by_name["settings"]["parent"] == "thread" + assert by_name["list"]["parent"] == "" + assert by_name["list"]["label"] == "Conversations" + + +async def test_the_list_page_shows_this_accounts_threads(druks_client): + owner = await Account.get_or_create("op@example.com") + other = await Account.get_or_create("dev@example.com") + mine = await Conversation.create(account_id=owner.id, title="Pump") + await Conversation.create(account_id=other.id, title="theirs") + + page = (await druks_client.get("/api/chat/pages")).json() + + assert page["title"] == "Conversations" + assert page["controls"][0]["page"] == "new" + (cards,) = page["blocks"] + (card,) = cards["cards"] + assert card["title"] == "Pump" + assert card["controls"][0] == { + "block": "link", + "label": "Open", + "page": "thread", + "arguments": {"conversation_id": str(mine.id)}, + "url": "", + "subject": None, + } + + +async def test_the_list_page_empty_state_points_at_new(druks_client): + page = (await druks_client.get("/api/chat/pages")).json() + + (cards,) = page["blocks"] + assert cards["cards"] == [] + assert cards["empty"]["controls"][0]["page"] == "new" + + +async def test_the_new_page_collects_an_optional_title_and_a_required_message(druks_client): + page = (await druks_client.get("/api/chat/pages/new")).json() + + (form,) = page["blocks"] + assert form["block"] == "form" + assert [field["name"] for field in form["fields"]] == ["title", "body"] + assert form["fields"][0]["isRequired"] is False + assert form["fields"][1]["isRequired"] is True + assert form["action"]["operation"] == "create_conversation" + + +async def test_the_thread_shows_messages_and_follows_the_conversation(druks_client): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="Pump") + await conversation.add_message(role=Role.USER, body="hello") + + page = (await druks_client.get(f"/api/chat/pages/conversations/{conversation.id}")).json() + + assert page["title"] == "Pump" + region = page["blocks"][0] + assert region["follows"] == { + "subjectType": "conversation", + "subjectId": str(conversation.id), + } + (cards, *turn) = region["blocks"] + assert turn == [] + (card,) = cards["cards"] + assert card["title"] == "You" + assert card["blocks"][0] == {"block": "quote", "text": "hello"} + assert page["controls"][1]["subject"] == { + "subjectType": "conversation", + "subjectId": str(conversation.id), + } + + +async def test_a_parked_turn_puts_gate_controls_on_the_thread(druks_client, druks_db): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="") + run = await seed_run( + druks_db, + kind=Talk.kind, + subject=conversation, + state="parked", + input_gate=ChatTurn.name, + input_request={ + "presentation": "in_app", + "label": "Message", + "controls": ["send", "stop"], + "questions": [], + }, + ) + run.input_requested_at = datetime.now(UTC) + await druks_db.flush() + + page = (await druks_client.get(f"/api/chat/pages/conversations/{conversation.id}")).json() + + region = page["blocks"][0] + assert region["blocks"][-1] == {"block": "gate_controls", "run": run.id} + + +async def test_a_running_turn_shows_status_not_a_gate(druks_client, druks_db): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="") + await seed_run(druks_db, kind=Talk.kind, subject=conversation, state="running") + + page = (await druks_client.get(f"/api/chat/pages/conversations/{conversation.id}")).json() + + region = page["blocks"][0] + assert region["blocks"][-1]["block"] == "text" + assert all(block["block"] != "gate_controls" for block in region["blocks"]) + + +async def test_another_operators_thread_is_an_empty_state(druks_client): + other = await Account.get_or_create("dev@example.com") + conversation = await Conversation.create(account_id=other.id, title="secret") + await conversation.add_message(role=Role.USER, body="nope") + + page = (await druks_client.get(f"/api/chat/pages/conversations/{conversation.id}")).json() + + assert page["blocks"][0]["block"] == "empty_state" + assert "secret" not in str(page) + assert "nope" not in str(page) + + settings = ( + await druks_client.get(f"/api/chat/pages/conversations/{conversation.id}/settings") + ).json() + assert settings["blocks"][0]["block"] == "empty_state" + + +async def test_settings_offers_the_autonomy_modes(druks_client): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="Pump") + + page = ( + await druks_client.get(f"/api/chat/pages/conversations/{conversation.id}/settings") + ).json() + + (form,) = page["blocks"] + assert form["action"]["operation"] == "set_autonomy" + assert form["fields"][0]["name"] == "autonomy" + assert [option["value"] for option in form["fields"][0]["options"]] == [ + "propose", + "confirm", + "full", + ] diff --git a/backend/tests/chat/test_prompts.py b/backend/tests/chat/test_prompts.py new file mode 100644 index 00000000..9cbf7856 --- /dev/null +++ b/backend/tests/chat/test_prompts.py @@ -0,0 +1,20 @@ +from druks.prompts import render_prompt + + +async def test_talk_prompt_carries_autonomy_bounded_history_and_mcp_discovery(): + rendered = await render_prompt( + "chat/talk.md", + autonomy="propose", + messages=[ + {"role": "user", "body": "hello"}, + {"role": "assistant", "body": "hi"}, + ], + ) + + assert "Act on behalf of the signed-in operator" in rendered + assert "list_open_subjects" in rendered + assert "parkedAt" in rendered + assert "Do not invent run ids" in rendered + assert "Autonomy: propose" in rendered + assert "user: hello" in rendered + assert "assistant: hi" in rendered diff --git a/backend/tests/chat/test_routes.py b/backend/tests/chat/test_routes.py new file mode 100644 index 00000000..471497a5 --- /dev/null +++ b/backend/tests/chat/test_routes.py @@ -0,0 +1,92 @@ +from druks.accounts.models import Account +from druks.contrib.chat.enums import Autonomy, Role +from druks.contrib.chat.models import Conversation +from druks.contrib.chat.workflows import Talk + + +async def test_create_conversation_posts_body_and_starts_talk(druks_client, monkeypatch): + started = [] + + async def dispatch(*, conversation): + started.append(conversation.id) + return "run-1" + + monkeypatch.setattr(Talk, "dispatch", staticmethod(dispatch)) + + created = await druks_client.post( + "/api/chat/conversations", + json={"body": "hello"}, + ) + + assert created.status_code == 201 + conversation = await Conversation.get(created.json()["id"]) + assert conversation.title == "" + account = await Account.get_or_create("op@example.com") + assert conversation.account_id == account.id + assert started == [conversation.id] + messages = await conversation.list_messages() + assert [message.body for message in messages] == ["hello"] + assert [message.role for message in messages] == [Role.USER] + + +async def test_create_conversation_stores_an_optional_title(druks_client, monkeypatch): + async def dispatch(**kwargs): + return "run" + + monkeypatch.setattr(Talk, "dispatch", staticmethod(dispatch)) + + created = await druks_client.post( + "/api/chat/conversations", + json={"body": "hello", "title": "Pump"}, + ) + + assert created.status_code == 201 + conversation = await Conversation.get(created.json()["id"]) + assert conversation.title == "Pump" + + +async def test_set_autonomy_updates_this_accounts_thread(druks_client): + account = await Account.get_or_create("op@example.com") + conversation_id = (await Conversation.create(account_id=account.id, title="mine")).id + + response = await druks_client.post( + f"/api/chat/conversations/{conversation_id}/autonomy", + json={"autonomy": "full"}, + ) + + assert response.status_code == 200 + assert response.json()["autonomy"] == "full" + assert (await Conversation.get(conversation_id)).autonomy == Autonomy.FULL + + +async def test_set_autonomy_misses_another_operators_thread(druks_client): + other = await Account.get_or_create("dev@example.com") + conversation = await Conversation.create(account_id=other.id, title="theirs") + + response = await druks_client.post( + f"/api/chat/conversations/{conversation.id}/autonomy", + json={"autonomy": "full"}, + ) + + assert response.status_code == 404 + assert (await Conversation.get(conversation.id)).autonomy == "propose" + + +async def test_a_later_post_starts_another_conversation_not_a_new_talk_on_the_first( + druks_client, monkeypatch +): + started = [] + + async def dispatch(*, conversation): + started.append(conversation.id) + return "run" + + monkeypatch.setattr(Talk, "dispatch", staticmethod(dispatch)) + + first = await druks_client.post("/api/chat/conversations", json={"body": "hello"}) + second = await druks_client.post("/api/chat/conversations", json={"body": "again"}) + + assert first.status_code == 201 + assert second.status_code == 201 + assert first.json()["id"] != second.json()["id"] + assert started == [first.json()["id"], second.json()["id"]] diff --git a/backend/tests/chat/test_workflows.py b/backend/tests/chat/test_workflows.py new file mode 100644 index 00000000..98f05274 --- /dev/null +++ b/backend/tests/chat/test_workflows.py @@ -0,0 +1,244 @@ +from datetime import timedelta +from unittest import mock + +from druks.accounts.models import Account +from druks.contrib.chat.app import Chat +from druks.contrib.chat.contracts import TurnOutput +from druks.contrib.chat.enums import Role +from druks.contrib.chat.models import Conversation +from druks.contrib.chat.workflows import ChatTurn, ConfirmTool, Talk +from druks.notifications.services import validate_in_app_answer +from druks.workflows import current_workflow + +_CHAT_TURN_ASK = { + "presentation": "in_app", + "label": "Message", + "controls": ["send", "stop"], + "questions": [], +} + + +def test_chat_turn_accepts_the_in_app_send_and_stop_payload(): + send = validate_in_app_answer(_CHAT_TURN_ASK, "send", {}, "and then?") + assert ChatTurn.model_validate(send).action == "send" + assert ChatTurn.model_validate(send).note == "and then?" + stop = validate_in_app_answer(_CHAT_TURN_ASK, "stop", {}, "") + assert ChatTurn.model_validate(stop).action == "stop" + + +async def _run_talk(conversation: Conversation, monkeypatch) -> None: + monkeypatch.setattr(Talk, "record_message", Talk.record_message.__wrapped__) + monkeypatch.setattr(Talk, "name_thread", Talk.name_thread.__wrapped__) + flow = Talk() + flow.subject = conversation + flow.account_id = conversation.account_id + token = current_workflow.set(flow) + try: + await flow.run_multistep() + finally: + current_workflow.reset(token) + + +async def test_dispatch_starts_talk_for_the_conversation(monkeypatch): + conversation = Conversation(id=42) + start = mock.AsyncMock(return_value="run-1") + monkeypatch.setattr(Talk, "start", staticmethod(start)) + + run_id = await Talk.dispatch(conversation=conversation) + + assert run_id == "run-1" + start.assert_awaited_once_with(subject=conversation) + + +async def test_talk_appends_the_assistant_line_and_stops(druks_db, monkeypatch): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="") + await conversation.add_message(role=Role.USER, body="hello") + + reply = mock.AsyncMock(return_value=TurnOutput(text="hi")) + monkeypatch.setattr(Chat, "reply", staticmethod(reply)) + waits: list[dict] = [] + + async def wait(cls, **kwargs): + waits.append(kwargs) + return ChatTurn(action="stop") + + monkeypatch.setattr(ChatTurn, "wait", classmethod(wait)) + monkeypatch.setattr(Talk, "deferred_writes", mock.AsyncMock(return_value=[])) + + await _run_talk(conversation, monkeypatch) + + reply.assert_awaited_once() + assert waits[0]["hold_sandbox"] == timedelta(minutes=15) + assert waits[0]["input_request"] == _CHAT_TURN_ASK + messages = await conversation.list_messages() + assert [message.body for message in messages] == ["hello", "hi"] + assert [message.role for message in messages] == [Role.USER, Role.ASSISTANT] + assert conversation.title == "hello" + + +async def test_talk_appends_the_operator_line_and_loops(druks_db, monkeypatch): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="") + await conversation.add_message(role=Role.USER, body="hello") + + outputs = iter([TurnOutput(text="hi"), TurnOutput(text="ok")]) + turns: list[dict] = [] + + async def reply(**kwargs): + turns.append(kwargs) + return next(outputs) + + monkeypatch.setattr(Chat, "reply", staticmethod(reply)) + answers = iter( + [ChatTurn(action="send", note="and then?"), ChatTurn(action="stop", note="leave unused")] + ) + + async def wait(cls, **kwargs): + return next(answers) + + monkeypatch.setattr(ChatTurn, "wait", classmethod(wait)) + monkeypatch.setattr(Talk, "deferred_writes", mock.AsyncMock(return_value=[])) + + await _run_talk(conversation, monkeypatch) + + messages = await conversation.list_messages() + assert [message.body for message in messages] == ["hello", "hi", "and then?", "ok"] + assert [message.role for message in messages] == [ + Role.USER, + Role.ASSISTANT, + Role.USER, + Role.ASSISTANT, + ] + assert turns[0]["autonomy"] == conversation.autonomy + assert turns[1]["messages"][-1] == {"role": Role.USER, "body": "and then?"} + assert conversation.title == "hello" + + +async def test_stop_does_not_write_a_message(druks_db, monkeypatch): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="") + await conversation.add_message(role=Role.USER, body="hello") + monkeypatch.setattr(Chat, "reply", mock.AsyncMock(return_value=TurnOutput(text="hi"))) + monkeypatch.setattr(Talk, "deferred_writes", mock.AsyncMock(return_value=[])) + + async def wait(cls, **kwargs): + return ChatTurn(action="stop", note="goodnight") + + monkeypatch.setattr(ChatTurn, "wait", classmethod(wait)) + + await _run_talk(conversation, monkeypatch) + + assert [message.body for message in await conversation.list_messages()] == ["hello", "hi"] + + +async def test_talk_names_an_untitled_thread_once(druks_db, monkeypatch): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="Pump") + await conversation.add_message(role=Role.USER, body="hello") + monkeypatch.setattr(Chat, "reply", mock.AsyncMock(return_value=TurnOutput(text="hi"))) + monkeypatch.setattr(Talk, "deferred_writes", mock.AsyncMock(return_value=[])) + + async def wait(cls, **kwargs): + return ChatTurn(action="stop") + + monkeypatch.setattr(ChatTurn, "wait", classmethod(wait)) + + await _run_talk(conversation, monkeypatch) + + assert conversation.title == "Pump" + + +async def test_talk_prompts_with_bounded_history(druks_db, monkeypatch): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="t") + for index in range(41): + await conversation.add_message(role=Role.USER, body=f"m{index}") + turns: list[dict] = [] + + async def reply(**kwargs): + turns.append(kwargs) + return TurnOutput(text="ok") + + monkeypatch.setattr(Chat, "reply", staticmethod(reply)) + + async def wait(cls, **kwargs): + return ChatTurn(action="stop") + + monkeypatch.setattr(ChatTurn, "wait", classmethod(wait)) + monkeypatch.setattr(Talk, "deferred_writes", mock.AsyncMock(return_value=[])) + + await _run_talk(conversation, monkeypatch) + + assert [message["body"] for message in turns[0]["messages"]] == [ + f"m{index}" for index in range(1, 41) + ] + + +async def test_talk_confirms_deferred_writes_before_the_next_line(druks_db, monkeypatch): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="") + await conversation.add_message(role=Role.USER, body="hello") + + monkeypatch.setattr(Chat, "reply", mock.AsyncMock(return_value=TurnOutput(text="hi"))) + proposed = [ + { + "method": "POST", + "path": "/api/gates/run-1/answer", + "body": "{}", + "content_type": "application/json", + } + ] + monkeypatch.setattr(Talk, "deferred_writes", mock.AsyncMock(return_value=proposed)) + applied = [] + + async def apply(self, writes): + applied.extend(writes) + + monkeypatch.setattr(Talk, "apply_deferred_writes", apply) + parked = [] + + async def confirm_wait(cls, **kwargs): + parked.append(kwargs) + return ConfirmTool(action="approve") + + async def turn_wait(cls, **kwargs): + return ChatTurn(action="stop") + + monkeypatch.setattr(ConfirmTool, "wait", classmethod(confirm_wait)) + monkeypatch.setattr(ChatTurn, "wait", classmethod(turn_wait)) + + await _run_talk(conversation, monkeypatch) + + assert parked[0]["input_request"]["controls"] == ["approve", "reject"] + assert applied == proposed + + +async def test_talk_skips_deferred_writes_when_the_operator_rejects(druks_db, monkeypatch): + account = await Account.get_or_create("op@example.com") + conversation = await Conversation.create(account_id=account.id, title="") + await conversation.add_message(role=Role.USER, body="hello") + + monkeypatch.setattr(Chat, "reply", mock.AsyncMock(return_value=TurnOutput(text="hi"))) + monkeypatch.setattr( + Talk, + "deferred_writes", + mock.AsyncMock( + return_value=[{"method": "POST", "path": "/x", "body": "", "content_type": ""}] + ), + ) + apply = mock.AsyncMock() + monkeypatch.setattr(Talk, "apply_deferred_writes", apply) + + async def confirm_wait(cls, **kwargs): + return ConfirmTool(action="reject") + + async def turn_wait(cls, **kwargs): + return ChatTurn(action="stop") + + monkeypatch.setattr(ConfirmTool, "wait", classmethod(confirm_wait)) + monkeypatch.setattr(ChatTurn, "wait", classmethod(turn_wait)) + + await _run_talk(conversation, monkeypatch) + + apply.assert_not_awaited() diff --git a/backend/tests/conftest.py b/backend/tests/conftest.py index 5adc7309..58a34d30 100644 --- a/backend/tests/conftest.py +++ b/backend/tests/conftest.py @@ -163,6 +163,7 @@ def browser_session_declarations(): "test_provider_subscription_persistence", "test_app_migrations", "test_proof_app_migration", + "test_chat_migration", } diff --git a/backend/tests/test_app_doctor_checks.py b/backend/tests/test_app_doctor_checks.py index 95223426..2f6b114a 100644 --- a/backend/tests/test_app_doctor_checks.py +++ b/backend/tests/test_app_doctor_checks.py @@ -150,6 +150,7 @@ async def test_app_checks_are_wired_into_the_check_battery(installed, tmp_path: app_results = await doctor.check_apps(settings) assert isinstance(app_results, list) assert "field_notes:summary_api_key" in {result.name for result in app_results} + assert "chat:appliance_mcp" in {result.name for result in app_results} async def test_raising_app_check_is_isolated_and_does_not_stop_siblings( diff --git a/backend/tests/test_app_loader.py b/backend/tests/test_app_loader.py index d912b598..bde308c6 100644 --- a/backend/tests/test_app_loader.py +++ b/backend/tests/test_app_loader.py @@ -11,6 +11,7 @@ def test_import_app_models_registers_software_factory_via_generic_discovery(): from druks.models import Base assert get_app("software_factory").prefix_tables is False + assert get_app("chat").prefix_tables is True import_app_models() # idempotent; raises if the unprefixed tables aren't exempt assert {"projects", "work_items", "project_repos"} <= set(Base.metadata.tables) diff --git a/backend/tests/test_app_roster.py b/backend/tests/test_app_roster.py index e8229274..b21c9cdb 100644 --- a/backend/tests/test_app_roster.py +++ b/backend/tests/test_app_roster.py @@ -15,6 +15,13 @@ def test_roster_lists_installed_apps_with_subject_types(tmp_path: Path): # Software Factory's pages are React, so its tabs live in its frontend. assert software_factory["navigation"] == [] assert software_factory["icon"] + chat = roster["chat"] + assert chat["builtin"] is False + assert chat["hasFrontend"] is False + assert chat["icon"] == "message-square" + assert chat["subjectTypes"] == ["conversation"] + assert chat["navigation"] == [["/chat", "Conversations"]] + field_notes = roster["field_notes"] assert field_notes["subjectTypes"] == ["note", "repository"] # Derived from the landing page the app declares, labelled by that page. diff --git a/backend/tests/test_apps.py b/backend/tests/test_apps.py index 4281be1e..e452f108 100644 --- a/backend/tests/test_apps.py +++ b/backend/tests/test_apps.py @@ -24,7 +24,7 @@ def _subjects(cls) -> list[type[Subject]]: def test_iter_apps_discovers_the_bundled_apps(): """The bundled apps resolve from the ``druks.apps`` entry points.""" - assert {app.name for app in iter_apps()} >= {"core", "software_factory", "usage"} + assert {app.name for app in iter_apps()} >= {"core", "software_factory", "chat", "usage"} def test_platform_apps_are_builtin(): diff --git a/backend/tests/test_chat_migration.py b/backend/tests/test_chat_migration.py new file mode 100644 index 00000000..e312aa3c --- /dev/null +++ b/backend/tests/test_chat_migration.py @@ -0,0 +1,48 @@ +from pathlib import Path + +from alembic import command +from alembic.config import Config +from druks.testing import TEST_DATABASE_URL, init_db +from sqlalchemy import create_engine + +_ALEMBIC_INI = Path(__file__).resolve().parent.parent / "alembic.ini" +_VERSIONS = ( + Path(__file__).resolve().parent.parent + / "druks" + / "contrib" + / "chat" + / "migrations" + / "versions" +) +_TABLES = "chat_messages, chat_conversations, alembic_version_chat" + + +def _config() -> Config: + config = Config(str(_ALEMBIC_INI)) + config.set_main_option("version_locations", str(_VERSIONS)) + config.set_main_option("sqlalchemy.url", TEST_DATABASE_URL) + config.attributes["version_table"] = "alembic_version_chat" + return config + + +def _drop(conn) -> None: + conn.exec_driver_sql(f"DROP TABLE IF EXISTS {_TABLES}") + + +def test_chat_migration_applies_under_its_own_version_table(request): + engine = create_engine(TEST_DATABASE_URL, isolation_level="AUTOCOMMIT") + with engine.connect() as conn: + _drop(conn) + try: + command.upgrade(_config(), "head") + with engine.connect() as conn: + assert conn.exec_driver_sql("SELECT to_regclass('chat_conversations')").scalar() + assert conn.exec_driver_sql("SELECT to_regclass('chat_messages')").scalar() + head = conn.exec_driver_sql("SELECT version_num FROM alembic_version_chat").scalar() + assert head == "chat_0001" + finally: + with engine.connect() as conn: + _drop(conn) + init_db(engine) + engine.dispose() + request.getfixturevalue("_druks_engine").dispose() diff --git a/backend/tests/test_doctor.py b/backend/tests/test_doctor.py index 024cd7c6..ebe39d08 100644 --- a/backend/tests/test_doctor.py +++ b/backend/tests/test_doctor.py @@ -616,3 +616,4 @@ async def test_run_checks_includes_sandbox_e2e_only_when_flagged(tmp_path: Path) assert "sandbox_e2e" not in default assert "sandbox_e2e" in flagged + assert "chat" in default diff --git a/backend/tests/test_mcp_endpoint.py b/backend/tests/test_mcp_endpoint.py index 4f99a03f..b183b175 100644 --- a/backend/tests/test_mcp_endpoint.py +++ b/backend/tests/test_mcp_endpoint.py @@ -8,7 +8,7 @@ import httpx2 import pytest from conftest import finish_agent_run, make_test_note, seed_note_agent_run, seed_note_run -from druks.accounts.models import Account, PersonalAccessToken +from druks.accounts.models import Account, OperatorToken, PersonalAccessToken from druks.api.server import mcp_app from druks.contrib.software_factory.app import SoftwareFactory from druks.core.apis.exceptions import UnknownTicketError @@ -20,6 +20,7 @@ from fastapi import APIRouter, FastAPI from fastmcp.client import Client from fastmcp.client.transports import StreamableHttpTransport +from pydantic import BaseModel from starlette.routing import Route _IN_APP_ASK = { @@ -272,6 +273,43 @@ async def software_factory_scan(): create_mcp_app(api) +class _Issued(BaseModel): + identifier: str + + +def _create_shaped_app() -> FastAPI: + # A 201 whose model requires identifier — the confirm stub does not have one. + held: dict[str, object] = {} + + @asynccontextmanager + async def lifespan(scope_app): + async with held["mcp"].lifespan(scope_app): + yield + + api = FastAPI(lifespan=lifespan) + router = APIRouter() + + async def mint_ticket() -> _Issued: + """Write a ticket down.""" + return _Issued(identifier="BOX-1") + + router.add_api_route( + "/tickets", + mint_ticket, + methods=["POST"], + status_code=201, + operation_id="mint_ticket", + tags=["agent"], + ) + api.include_router(router, prefix="/api/software_factory", tags=["software_factory"]) + mcp = create_mcp_app(api) + held["mcp"] = mcp + api.router.routes.append( + Route("/mcp", mcp, methods=["POST", "DELETE"], include_in_schema=False) + ) + return api + + def _agent_route_app(operation_id: str) -> FastAPI: # A synthetic agent route owned by the installed software_factory app — the # loader-stamped app tag names the owner, exactly as a real router does. @@ -555,3 +593,158 @@ async def test_mcp_server_registry_routes_stay_untouched(tmp_path, druks_db, mon assert listed.status_code == 200 # The inbound endpoint never joins the outbound server registry. assert "druks" not in {server["name"] for server in listed.json()} + + +async def _operator_token(account, *, writes: str, run_id: str = "run-op"): + return await OperatorToken.mint( + account_id=account.id, + agent_call_id=f"call-{writes}", + run_id=run_id, + writes=writes, + ) + + +async def test_propose_cannot_hit_answer_gate(app, account, druks_db, resume_spy): + item = await make_test_note() + run = await seed_note_run( + druks_db, + note=item, + state="parked", + input_gate="review", + input_request=dict(_IN_APP_ASK), + ) + run.input_requested_at = datetime.now(UTC) + await druks_db.flush() + token = await _operator_token(account, writes="deny") + + async with live(app), _client(app, token) as client: + tools = {tool.name for tool in await client.list_tools()} + assert "list_open_subjects" in tools + assert "get_gate" in tools + assert "get_agent_call" in tools + assert "get_usage" in tools + assert "answer_gate" not in tools + assert "cancel_run" not in tools + result = await client.call_tool( + "answer_gate", + {"run": run.id, "parkedAt": run.input_requested_at.isoformat(), "control": "approve"}, + raise_on_error=False, + ) + + assert result.is_error + assert resume_spy == [] + + +async def test_confirm_defers_answer_gate_until_play(app, account, druks_db, resume_spy): + item = await make_test_note() + run = await seed_note_run( + druks_db, + note=item, + state="parked", + input_gate="review", + input_request=dict(_IN_APP_ASK), + ) + run.input_requested_at = datetime.now(UTC) + await druks_db.flush() + token = await _operator_token(account, writes="defer", run_id="run-confirm") + + async with live(app), _client(app, token) as client: + tools = {tool.name for tool in await client.list_tools()} + assert "answer_gate" in tools + parked_at = run.input_requested_at.isoformat() + await client.call_tool( + "answer_gate", + {"run": run.id, "parkedAt": parked_at, "control": "approve"}, + raise_on_error=False, + ) + + assert resume_spy == [] + writes = await OperatorToken.take_deferred("run-confirm") + assert writes[0]["method"] == "POST" + await OperatorToken.play_deferred(account.id, writes) + assert resume_spy == [{"id": run.id, "action": "approve", "answers": {}, "note": ""}] + + +async def test_mutating_agent_tools_omit_output_schema(app, pat_token): + async with live(app), _client(app, pat_token) as client: + tools = {tool.name: tool for tool in await client.list_tools()} + assert tools["get_gate"].output_schema + assert tools["answer_gate"].output_schema is None + assert tools["cancel_run"].output_schema is None + + +async def test_confirm_defers_a_create_shaped_write(account): + token = await _operator_token(account, writes="defer", run_id="run-confirm-create") + api = _create_shaped_app() + + async with live(api), _client(api, token) as client: + result = await client.call_tool("software_factory_mint_ticket", {}, raise_on_error=False) + + if result.is_error: + raise AssertionError(result.content[0].text) + assert result.structured_content["result"] == "deferred" + writes = await OperatorToken.take_deferred("run-confirm-create") + assert writes[0]["method"] == "POST" + assert writes[0]["path"] == "/api/software_factory/tickets" + + +async def test_full_answer_gate_runs_as_the_token_account(app, account, druks_db, resume_spy): + item = await make_test_note() + run = await seed_note_run( + druks_db, + note=item, + state="parked", + input_gate="review", + input_request=dict(_IN_APP_ASK), + ) + run.input_requested_at = datetime.now(UTC) + await druks_db.flush() + token = await _operator_token(account, writes="allow") + + async with live(app), _client(app, token) as client: + gate = (await client.call_tool("get_gate", {"run": run.id})).structured_content + answered = ( + await client.call_tool( + "answer_gate", + {"run": run.id, "parkedAt": gate["parkedAt"], "control": "approve"}, + ) + ).structured_content + + assert answered["result"] == "answered" + assert resume_spy == [{"id": run.id, "action": "approve", "answers": {}, "note": ""}] + + +async def test_operator_token_cannot_act_as_another_account(app, druks_db): + mine = await Account.get_or_create("op@example.com") + theirs = await Account.get_or_create("peer@example.com") + druks_db.add( + UsageScrape( + provider="openai", + account_id=theirs.id, + scraped_at=datetime.now(UTC), + five_hour_percent_left=42, + ) + ) + await druks_db.flush() + token = await _operator_token(mine, writes="allow") + + async with live(app), _client(app, token) as client: + usage = (await client.call_tool("get_usage", {})).structured_content + codex = next(h for h in usage["providers"] if h["id"] == "openai") + assert codex["fiveHourPercentLeft"] is None + + +async def test_operator_token_is_gone_after_the_call(app, account): + token = await _operator_token(account, writes="allow", run_id="run-gone") + assert (await OperatorToken.lookup(token)).account_id == account.id + await OperatorToken.revoke("call-allow") + assert await OperatorToken.lookup(token) is None + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=app), base_url="http://druks.test" + ) as wire: + gone = await wire.post( + "/mcp", + json=_INIT, + headers={**_WIRE_HEADERS, "Authorization": f"Bearer {token}"}, + ) + assert gone.status_code == 401 diff --git a/backend/tests/test_sandbox_lifecycle.py b/backend/tests/test_sandbox_lifecycle.py index a995dae2..b95d55c2 100644 --- a/backend/tests/test_sandbox_lifecycle.py +++ b/backend/tests/test_sandbox_lifecycle.py @@ -1,5 +1,5 @@ from dataclasses import dataclass, field -from datetime import UTC, datetime +from datetime import UTC, datetime, timedelta from pathlib import Path from typing import Any @@ -94,10 +94,12 @@ class _FakeAPI: created_secrets: list[dict[str, Secret] | None] = field(default_factory=list) created_expires_at: list[datetime | None] = field(default_factory=list) deleted_ids: list[str] = field(default_factory=list) + renewed: list[tuple[str, datetime | None]] = field(default_factory=list) get_host_responses: list[SandboxHostRecord] = field(default_factory=list) create_record: SandboxHostRecord | None = None create_raises: Exception | None = None delete_raises: Exception | None = None + renew_raises: Exception | None = None # When set, every get_host call raises this exception instead of # returning from get_host_responses. Used by the attach() tests # to simulate the provider 404'ing a host we still have in our @@ -137,6 +139,17 @@ async def delete_host(self, host_id: str) -> None: if self.delete_raises is not None: raise self.delete_raises + async def renew_host( + self, + host_id: str, + *, + expires_at: datetime | None = None, + ) -> SandboxHostRecord: + self.renewed.append((host_id, expires_at)) + if self.renew_raises is not None: + raise self.renew_raises + return _record(host_id=host_id) + def _record( status: str = "active", @@ -923,3 +936,35 @@ async def test_a_gone_box_loses_its_identity_and_the_retry_provisions_anew( await db_session().refresh(identity) assert not identity.is_live assert await SandboxIdentity.lookup("run-1", "workflow", identity.secret_refs) is None + + +async def test_set_expiry_forwards_expires_at_to_sdk_renew( + patched_sandbox_api: list[_FakeAPI], +): + """Clipping a lease is a renew, not a delete: the caller's expiry goes + through to the SDK verbatim — no druks-side clamp — and the VM stays up.""" + api = _FakeAPI(create_record=None) + patched_sandbox_api.append(api) + expires_at = datetime.now(UTC) + timedelta(seconds=90) + + await sandbox_client.set_expiry(host_id="host-xyz", expires_at=expires_at) + + assert api.renewed == [("host-xyz", expires_at)] + assert api.deleted_ids == [] + + +async def test_set_expiry_surfaces_missing_host(patched_sandbox_api: list[_FakeAPI]): + """A host drukbox no longer knows about surfaces as the SDK's + ``SandboxNotFoundError`` — the idle-hold caller decides what "gone" + means, so it is neither swallowed nor remapped to ``HostGone``.""" + missing = SandboxNotFoundError("host-xyz not found") + api = _FakeAPI(create_record=None, renew_raises=missing) + patched_sandbox_api.append(api) + + with pytest.raises(SandboxNotFoundError) as excinfo: + await sandbox_client.set_expiry( + host_id="host-xyz", + expires_at=datetime.now(UTC) + timedelta(seconds=90), + ) + + assert excinfo.value is missing diff --git a/backend/tests/test_warm_host_rotation.py b/backend/tests/test_warm_host_rotation.py index fa333ca7..038ee417 100644 --- a/backend/tests/test_warm_host_rotation.py +++ b/backend/tests/test_warm_host_rotation.py @@ -37,6 +37,7 @@ def __init__(self, *, lease: timedelta) -> None: self.secrets: list[dict[str, Secret]] = [] self.released: list[str] = [] self.reattached: list[str] = [] + self.expiry_sets: list[tuple[str, datetime]] = [] async def provision( self, @@ -60,6 +61,9 @@ async def reattach(self, *, host_id: str) -> _FakeSandbox: self.reattached.append(host_id) return _FakeSandbox(id=host_id, expires_at=datetime.now(UTC) + self.lease) + async def set_expiry(self, *, host_id: str, expires_at: datetime) -> None: + self.expiry_sets.append((host_id, expires_at)) + def _warm_workflow(*, reuse: bool = True) -> Workflow: # __new__ skips __init__/__init_subclass__ so the host logic can be exercised @@ -73,6 +77,19 @@ def _warm_workflow(*, reuse: bool = True) -> Workflow: return flow +def _park_without_dbos(monkeypatch) -> None: + # The park's durable surroundings — the run event it emits and the channel it + # suspends on — say nothing about the hold, so the gate answers immediately. + async def _emit(*args, **kwargs) -> None: + return + + async def _answer(gate, timeout_seconds=None) -> dict[str, str]: + return {"action": "approve"} + + monkeypatch.setattr(sdk, "_emit_run_event", _emit) + monkeypatch.setattr(sdk.DBOS, "recv_async", _answer) + + @pytest.mark.asyncio async def test_warm_host_reused_while_lease_covers_another_call(monkeypatch): """A warm host with lease to spare is reused across calls, never re-provisioned.""" @@ -198,3 +215,134 @@ async def test_a_replay_finds_the_warm_box_through_its_identity( assert client.reattached == ["host-crashed"] assert client.provisions == [] + + +@pytest.mark.asyncio +async def test_park_without_hold_releases_the_warm_host(monkeypatch): + """A park with no hold is today's park: the VM goes, nothing is clipped.""" + fake = _FakeSandboxClient(lease=timedelta(hours=2)) + monkeypatch.setattr(sdk, "sandbox_client", fake) + _park_without_dbos(monkeypatch) + flow = _warm_workflow() + await flow._lease_host(_NONE) + + await sdk._park(flow, "review", None, 60.0) + + assert fake.released == ["host-1"] + assert fake.expiry_sets == [] + assert flow._host is None + + +@pytest.mark.asyncio +async def test_park_with_hold_clips_the_lease_and_keeps_the_host(monkeypatch): + """A held park clips the lease instead of deleting the VM, and the run keeps + the handle so a same-worker resume reattaches warm.""" + fake = _FakeSandboxClient(lease=timedelta(hours=2)) + monkeypatch.setattr(sdk, "sandbox_client", fake) + _park_without_dbos(monkeypatch) + flow = _warm_workflow() + await flow._lease_host(_NONE) + + await sdk._park(flow, "review", None, 60.0, hold_sandbox=True) + + assert [host_id for host_id, _ in fake.expiry_sets] == ["host-1"] + assert fake.released == [] + assert flow._host is not None + assert flow._host.id == "host-1" + + +@pytest.mark.asyncio +async def test_hold_true_clips_to_one_more_worst_case_call(monkeypatch): + """``True`` holds the VM for as long as its lease could still cover one more + worst-case call — past that the next call would rotate anyway.""" + fake = _FakeSandboxClient(lease=timedelta(hours=2)) + monkeypatch.setattr(sdk, "sandbox_client", fake) + flow = _warm_workflow() + await flow._lease_host(_NONE) + + await flow._hold_host(True) + + ((_, expires_at),) = fake.expiry_sets + clip = datetime.now(UTC) + timedelta(seconds=SANDBOX_HOST_ROTATE_BEFORE_SECONDS) + assert clip - timedelta(seconds=5) <= expires_at <= clip + assert expires_at <= flow._host.expires_at + + +@pytest.mark.asyncio +async def test_hold_never_outlasts_the_lease_drukbox_granted(monkeypatch): + """The clip is a floor, never an extension: a lease shorter than the hold + stands as it is.""" + fake = _FakeSandboxClient(lease=timedelta(minutes=20)) + monkeypatch.setattr(sdk, "sandbox_client", fake) + flow = _warm_workflow() + await flow._lease_host(_NONE) + + await flow._hold_host(timedelta(hours=1)) + + assert fake.expiry_sets == [("host-1", flow._host.expires_at)] + + +@pytest.mark.asyncio +async def test_hold_timedelta_clips_to_the_requested_span(monkeypatch): + """A timedelta hold ends at ``now + hold`` when the lease outlasts it.""" + fake = _FakeSandboxClient(lease=timedelta(hours=2)) + monkeypatch.setattr(sdk, "sandbox_client", fake) + flow = _warm_workflow() + await flow._lease_host(_NONE) + + await flow._hold_host(timedelta(minutes=30)) + + ((_, expires_at),) = fake.expiry_sets + clip = datetime.now(UTC) + timedelta(minutes=30) + assert clip - timedelta(seconds=5) <= expires_at <= clip + + +@pytest.mark.asyncio +async def test_hold_without_a_warm_host_touches_nothing(monkeypatch): + """Without steps_reuse_sandbox there is no warm host to hold, so a held park + neither clips nor deletes.""" + fake = _FakeSandboxClient(lease=timedelta(hours=2)) + monkeypatch.setattr(sdk, "sandbox_client", fake) + _park_without_dbos(monkeypatch) + flow = _warm_workflow(reuse=False) + await flow._lease_host(_NONE) + + await sdk._park(flow, "review", None, 60.0, hold_sandbox=True) + + assert fake.expiry_sets == [] + assert fake.released == [] + assert fake.provisions == [] + + +@pytest.mark.asyncio +async def test_resume_after_a_hold_reuses_the_held_host(monkeypatch): + """The worker that survived the recv still holds the handle, so the first + call after the resume lands on the same VM.""" + fake = _FakeSandboxClient(lease=timedelta(hours=2)) + monkeypatch.setattr(sdk, "sandbox_client", fake) + _park_without_dbos(monkeypatch) + flow = _warm_workflow() + await flow._lease_host(_NONE) + await sdk._park(flow, "review", None, 60.0, hold_sandbox=True) + + assert await flow._lease_host(_NONE) == "host-1" + assert fake.provisions == ["wf-1:workflow"] + + +@pytest.mark.asyncio +async def test_resume_on_a_restarted_worker_re_leases_under_the_run_key(monkeypatch): + """A worker that died over the park has no handle: the resume goes back + through the run's idempotency key exactly once — warm if the clipped lease + still stands, cold if drukbox already reaped it.""" + fake = _FakeSandboxClient(lease=timedelta(hours=2)) + monkeypatch.setattr(sdk, "sandbox_client", fake) + _park_without_dbos(monkeypatch) + flow = _warm_workflow() + await flow._lease_host(_NONE) + await sdk._park(flow, "review", None, 60.0, hold_sandbox=timedelta(minutes=30)) + flow._host = None + + await flow._lease_host(_NONE) + + assert fake.provisions == ["wf-1:workflow", "wf-1:workflow"] + assert fake.released == [] diff --git a/docs/chat.md b/docs/chat.md new file mode 100644 index 00000000..250dd44c --- /dev/null +++ b/docs/chat.md @@ -0,0 +1,59 @@ +--- +title: "Chat" +description: "Start operator conversations, set autonomy, and reuse a warm sandbox across turns." +icon: "message-square" +--- + +Chat is a bundled app. It ships with Druks. It is not an optional package. + +Open **Chat** in the dashboard rail. Each conversation belongs to the signed-in +operator. Other accounts cannot read it. + +## Start a conversation + +1. Open **Chat → Conversations**. +2. Choose **New**. +3. Enter the first message. A title is optional. An empty title is filled from + the first user line after the first assistant reply. +4. Choose **Start**. + +That start creates the thread and one Talk run. Later lines answer the parked +turn. **Send** appends your line and continues. **Stop** ends Talk and reaps +the sandbox. Stop does not write a message. + +## Autonomy + +Each conversation has a mode. A change applies on the next agent call. + +| Mode | Tools | +| --- | --- | +| **propose** (default) | Read tools only. Mutating tools do not run. The agent proposes the action in the thread. You commit it in the real dashboard. | +| **confirm** | The live catalog is visible. Mutating tools stash the call. After the turn, a confirm gate asks you to approve or reject before the next user line. | +| **full** | Tools run immediately as your account. | + +Set the mode on the conversation. The next Talk call reads the live row. + +Chat uses the same `/mcp` catalog as an external agent. It is not a second +catalog and it does not replace [Connect your agent](connect-your-agent.md). +The sandbox talks inward to this appliance as you, through a call-scoped token +that dies with the agent call. + +## Idle window + +Talk parks each turn with `hold_sandbox` of 15 minutes. The park itself still +lasts days. The hold only clips the Drukbox lease so the warm VM stays up for +a short idle window. + +Send inside that window and Talk reuses the same host. It does not provision +a new one. After the clipped lease ends, the next line provisions a cold host +and still works. + +**Stop** reaps the host. A worker crash still frees the VM when the Drukbox +lease lapses. `review()` and any park without `hold_sandbox` still delete the +host at once. + +`druks doctor` reports that Chat is bundled. It also probes `/mcp` when sandbox +execution is on. Set `urls.endpoint` so a sandbox can reach this appliance. +A Docker sandbox rewrites a loopback dashboard URL to +`host.docker.internal`. See +[public URLs](configuration.md#public-urls-and-access-control). diff --git a/docs/concepts.md b/docs/concepts.md index 78bf025f..6bfad96c 100644 --- a/docs/concepts.md +++ b/docs/concepts.md @@ -54,8 +54,9 @@ name must match `App.name`. The same name scopes: - Optional static frontend assets in the app package. The bundled `software_factory` app owns projects, work items, ticket intake, -GitHub branches, pull requests, coding-agent policy, and dashboard pages. These -features are examples, not platform guarantees. +GitHub branches, pull requests, coding-agent policy, and dashboard pages. The +bundled `chat` app owns operator conversations. These features are examples, not +platform guarantees. ## Durability and recovery @@ -170,12 +171,17 @@ does not infer access health from configuration. A `Gate` defines a typed reply and a durable receive topic. When a workflow waits at a gate, Druks: -1. Releases each warm sandbox that the workflow holds. +1. Releases each warm sandbox that the workflow holds, unless the wait passes + `hold_sandbox`. A hold clips the Drukbox lease. It is shorter than the + remaining lease. The park itself still lasts up to 14 days. 2. Records `parked` and the request for the operator. 3. Sends an optional notification. 4. Suspends the workflow until a reply arrives or the 14-day timeout expires. 5. Clears the gate and returns the validated reply after the workflow resumes. +`review()` does not pass `hold_sandbox`, so it still reaps. See +[`Gate.wait`](writing-an-app.md#wait-for-input). + Each parked round accepts one answer through an idempotency key. In-app review requires a subject because the subject read-side is where the question appears. A subjectless custom gate must override `on_wait()` to send an external @@ -205,8 +211,10 @@ not write provider-specific execution code. By default, each agent call uses an ephemeral sandbox. A workflow can retain one warm sandbox across a segment. Druks releases it before a gate and at workflow -exit. Druks also rotates it before the lease becomes too short for another -call. Store durable state in an external system such as Git, not only on the VM. +exit, unless that gate wait passes `hold_sandbox`. A hold clips the lease. It +never extends it. Druks also rotates the host before the lease becomes too +short for another call. Store durable state in an external system such as Git, +not only on the VM. A sandbox never holds a subscription token. Druks gives each sandbox that fetches one an identity at its issuer, before Drukbox provisions it. The diff --git a/docs/configuration.md b/docs/configuration.md index efbf7cc9..3779f1c6 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -137,7 +137,7 @@ caches, and the sandbox provisioning gate. | TOML key | Purpose | | --- | --- | -| `urls.endpoint` | Browser-visible dashboard base URL used to build MCP OAuth callbacks | +| `urls.endpoint` | Browser-visible dashboard base URL. MCP OAuth callbacks and sandbox hops to this appliance's `/mcp` use it | | `urls.webhook_host` | Public webhook hostname used by `druks doctor` for its ingress probe | | `identity.mode` | `none` (default, no authentication, single operator), `header` (edge-asserted identity), or `jwt` (validated edge-signed assertion) | | `identity.header` | The trusted identity header. The shipped Caddy edge also uses it. Header and JWT modes have no default and require it | @@ -153,6 +153,10 @@ other addresses free, set `DRUKS_WEBHOOK_BIND_HOST` in `[env]` to the public address. Caddy then serves only that address. To keep IPv6, list the IPv4 and the IPv6 addresses. +A Docker sandbox cannot use the host loopback. Druks rewrites a loopback +`urls.endpoint` to `host.docker.internal` for that hop. An exe VM uses +`urls.endpoint` as given. See [Chat](chat.md). + `urls.endpoint` and `urls.webhook_host` are different. The first is where an operator's browser reaches Druks. The second is the public ingress host for webhook senders. They can share a hostname on exe.dev. diff --git a/docs/connect-your-agent.md b/docs/connect-your-agent.md index 3de2aed2..7a88d807 100644 --- a/docs/connect-your-agent.md +++ b/docs/connect-your-agent.md @@ -70,3 +70,12 @@ Three details matter when an agent uses the MCP endpoint: and `RUN_NOT_ACTIVE` are stable match values. App tools use the API shape `{"error", "detail"}` for refusals. Shape errors contain `VALIDATION_ERROR` detail. + +## Chat uses the same catalog inward + +Dashboard [Chat](chat.md) is not a replacement for this external MCP connection. +A Talk sandbox calls the same `/mcp` tools as the signed-in operator. Mint a +personal access token when Claude Code or Codex should talk *into* Druks from +your laptop. Chat does not use that token. It mints a call-scoped credential +that dies with the agent call. + diff --git a/docs/docs.json b/docs/docs.json index b130701d..7d1b3347 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -66,6 +66,7 @@ "deployment", "configuration", "connect-your-agent", + "chat", "troubleshooting" ] }, diff --git a/docs/full-local.md b/docs/full-local.md index 32d50091..e2167981 100644 --- a/docs/full-local.md +++ b/docs/full-local.md @@ -156,6 +156,9 @@ ticket or GitHub trigger. The run appears on the subject page and in the Events feed. Agent-call pages stream transcript and artifact data. +Open **Chat** to start an operator conversation on this appliance. See +[Chat](chat.md). + If you develop a different app, install that distribution into a development Druks environment and invoke its documented trigger or `Workflow.start()` path. See [writing an app](writing-an-app.md). diff --git a/docs/index.md b/docs/index.md index 0ed038f2..e6c1f515 100644 --- a/docs/index.md +++ b/docs/index.md @@ -47,8 +47,9 @@ isolated agent calls, or waits. Examples include software delivery, incident investigation, research review, approval flows, and periodic operational checks. The bundled **Software Factory** app coordinates coding agents from a work item -to a reviewed pull request. It demonstrates the framework. GitHub policy and -software-delivery behavior belong to the app, not to Druks. +to a reviewed pull request. The bundled **Chat** app keeps operator conversations +on this appliance. Both demonstrate the framework. Domain policy belongs to the +app, not to Druks. ## Choose a path @@ -58,6 +59,7 @@ software-delivery behavior belong to the app, not to Druks. - **Give it screens:** Read the [Druks UI contract](druks-ui.md). - **Run a production stack:** Follow the [deployment runbook](deployment.md). - **Diagnose a failure:** Use [troubleshooting](troubleshooting.md). +- **Chat on this appliance:** Read [Chat](chat.md). ## What Druks is not diff --git a/docs/troubleshooting.md b/docs/troubleshooting.md index 01a58694..23378112 100644 --- a/docs/troubleshooting.md +++ b/docs/troubleshooting.md @@ -15,8 +15,9 @@ docker compose logs --tail=200 web `druks doctor` examines the full platform. It covers settings, secrets, service credentials, data-directory writes, Postgres, Redis, Drukbox, harnesses, app -imports, capability modules, and app-owned checks. A failed check exits with a -nonzero status. +imports, capability modules, bundled Chat, and app-owned checks. A failed +check exits with a nonzero status. Chat is always installed. Its `/mcp` probe +skips with a reason when sandbox execution is off. If the normal Drukbox check passes but real execution fails, use the opt-in sandbox check: @@ -173,6 +174,14 @@ The exchange binds `127.0.0.1:8781` on the Druks host. Until it answers, a sandbox gets no value for its placeholders, and each agent call fails at the provider with an authentication error. +### Chat cannot reach `/mcp` + +Talk injects this appliance's `/mcp` into the sandbox. `druks doctor` reports +`chat:appliance_mcp`. A skip means sandbox execution is off. Pending means +`urls.endpoint` is unset. An unreachable URL means the VM cannot operate as the +operator. Set `urls.endpoint`. A Docker sandbox rewrites loopback to +`host.docker.internal`. See [Chat](chat.md). + ### A sandbox process appears stuck Druks copies the dashboard transcript from files that a detached VM process writes. @@ -185,7 +194,8 @@ agent process. Recovery follows the durable operation boundary. `parked` means that DBOS suspended the workflow on a gate. The workflow did not stall. Open the subject detail page to see its current ask. In-app review offers -approve, request changes, or cancel. The owner system answers an external gate. +approve, request changes, or cancel. Chat parks send and stop. Confirm parks +approve and reject. The owner system answers an external gate. If no notification arrived: diff --git a/docs/writing-an-app.md b/docs/writing-an-app.md index bc6d764a..3bb46066 100644 --- a/docs/writing-an-app.md +++ b/docs/writing-an-app.md @@ -10,6 +10,9 @@ Druks supplies durable execution and shared operating services. Read [the app boundary](concepts.md#the-app-boundary) before you assign ownership of a capability. +The bundled `chat` and `software_factory` apps register through the same +`druks.apps` entry point. They ship with Druks. They are not optional packages. + ## Scaffold and prove the package ```bash @@ -544,7 +547,8 @@ Override it to deliver none. 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, +the host at a gate and at workflow exit, unless `Gate.wait` passes +`hold_sandbox`. It rotates the host near lease expiry, and when the next agent call needs other secret entries. ### Borrow a browser session @@ -632,7 +636,31 @@ reply = await ApproveReport.wait( ``` `on_wait()` is a checkpointed notification step. The workflow then parks -durably and releases its warm sandbox. The owning external system resumes the +durably. By default it releases its warm sandbox. Pass `hold_sandbox` to keep +the VM across a short idle window instead: + +```python +from datetime import timedelta + +reply = await ApproveReport.wait( + input_request={ + "presentation": "external", + "label": "Review the night-watch report", + "url": review_url, + }, + hold_sandbox=timedelta(minutes=15), +) +``` + +Default `False` reaps. `True` holds for as long as the remaining +lease could still cover one more worst-case agent call. A `timedelta` holds +for at most that span. A hold never extends the lease Drukbox already granted. +The park itself still lasts up to 14 days. The clipped lease is what ends the +hold if nobody answers. + +`review()` calls the park path without `hold_sandbox`, so it still reaps. + +The owning external system resumes the workflow through the gate and its subject: ```python diff --git a/frontend/src/components/RunControls.test.tsx b/frontend/src/components/RunControls.test.tsx index d6f0c811..d96e3daf 100644 --- a/frontend/src/components/RunControls.test.tsx +++ b/frontend/src/components/RunControls.test.tsx @@ -97,6 +97,14 @@ describe('InAppReview', () => { expect(screen.getByText('A note is sent to the agent as feedback.')).toBeTruthy() }) + + it('labels send and stop', () => { + stubFetch() + renderReview({ presentation: 'in_app', controls: ['send', 'stop'], questions: [] }) + + expect(screen.getByText('Send')).toBeTruthy() + expect(screen.getByText('Stop')).toBeTruthy() + }) }) describe('the lent run controls', () => { diff --git a/frontend/src/components/RunControls.tsx b/frontend/src/components/RunControls.tsx index 1ebdcd61..1fa07ea0 100644 --- a/frontend/src/components/RunControls.tsx +++ b/frontend/src/components/RunControls.tsx @@ -84,6 +84,8 @@ const CONTROL_LABEL: Record = { approve: 'Approve', request_changes: 'Request changes', revise_contract: 'Revise contract', + send: 'Send', + stop: 'Stop', } export function InAppReview({ diff --git a/pyproject.toml b/pyproject.toml index 0978e5d4..84820392 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -71,6 +71,7 @@ druks = "druks.testing" [project.entry-points."druks.apps"] core = "druks.core.app:Core" software_factory = "druks.contrib.software_factory.app:SoftwareFactory" +chat = "druks.contrib.chat.app:Chat" usage = "druks.usage.app:Usage" [build-system]