DRU-376 - Operator chat

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_<secret>. 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 "_".

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

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)

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

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

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

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"

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