DRU-384 - Let Chat operate this appliance as the signed-in operator through a call-scoped MCP token.

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/templates/talk.md

@@ -1,4 +1,6 @@
-Reply to the operator.
+You are this operator. Act through the live Druks MCP catalog.
+
+Discover work with list_open_subjects. Echo parkedAt from get_gate unchanged when you answer a gate. Do not invent run ids.
 
 Autonomy: {{ autonomy }}
 

backend/druks/contrib/chat/workflows.py

@@ -1,9 +1,25 @@
+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.workflows import Gate, Workflow, step
+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):
@@ -15,12 +31,62 @@ class ChatTurn(Gate):
     stop: bool = False
 
 
+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:
@@ -31,6 +97,18 @@ async def run_multistep(self) -> None:
                 messages=[{"role": message.role, "body": message.body} for message in messages],
             )
             await self.record_message(Role.ASSISTANT, result.text)
+            deferred = await self.deferred_writes()
+            if deferred:
+                decision = await ConfirmTool.wait(
+                    input_request={
+                        "presentation": "in_app",
+                        "label": "Confirm the proposed action",
+                        "controls": ["approve", "reject"],
+                    },
+                    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": "Chat turn"},
                 hold_sandbox=self.sandbox_hold,
@@ -39,11 +117,32 @@ async def run_multistep(self) -> None:
                 return
             await self.record_message(Role.USER, reply.text)
 
+    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 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

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