Stabilize DM threading and enforce mutation confirmation proof

Co-Authored-By: GPT-5 Codex <codex@openai.com>

Author: dcramer

packages/ash-sandbox-cli/src/ash_sandbox_cli/commands/capability.py

@@ -122,6 +122,20 @@ def invoke_capability(
         str | None,
         typer.Option("--account", help="Optional linked account alias"),
     ] = None,
+    mutation_plan_id: Annotated[
+        str | None,
+        typer.Option(
+            "--plan-id",
+            help="Optional mutation plan id for host confirmation proof",
+        ),
+    ] = None,
+    target_fingerprint: Annotated[
+        str | None,
+        typer.Option(
+            "--target-fingerprint",
+            help="Optional target fingerprint for host confirmation proof",
+        ),
+    ] = None,
 ) -> None:
     """Invoke one capability operation."""
     try:
@@ -139,6 +153,10 @@ def invoke_capability(
         params["idempotency_key"] = idempotency_key
     if account:
         params["account_ref"] = account
+    if mutation_plan_id:
+        params["mutation_plan_id"] = mutation_plan_id
+    if target_fingerprint:
+        params["target_fingerprint"] = target_fingerprint
 
     result = _call("capability.invoke", params)
     request_id = result.get("request_id", "?")

specs/sessions.md

@@ -11,7 +11,13 @@ Ash uses **per-user sessions scoped to thread** for group chats:
 - Standalone `@mention` messages create a new thread (thread_id = message external_id)
 - Replies follow the parent thread via `ThreadIndex`
 - Session key for groups: `telegram_{chat_id}_{user_id}_{thread_id}` (users in the same thread do not share session state)
-- DMs use a single session: `telegram_{chat_id}_{user_id}` (no thread_id)
+
+For DMs, Ash uses **hybrid active-thread routing**:
+
+- Replies follow parent thread via `ThreadIndex`.
+- Non-reply messages continue on the active DM thread when it is fresh.
+- A new thread is created when no active thread is available (or after explicit new-topic intent/timeout rollover).
+- Session key for DM turns remains thread-scoped when a thread_id exists: `telegram_{chat_id}_{user_id}_{thread_id}`.
 
 ## File Structure
 

src/ash/capabilities/manager.py

@@ -29,6 +29,7 @@
     CapabilityDefinition,
     CapabilityInvokeResult,
 )
+from ash.chats import ChatStateManager
 
 _NAMESPACED_CAPABILITY_ID = re.compile(r"^[a-z0-9][a-z0-9_-]*\.[a-z0-9][a-z0-9_-]*$")
 _NAMESPACE = re.compile(r"^[a-z0-9][a-z0-9_-]*$")
@@ -687,6 +688,8 @@ async def invoke(
         source_display_name: str | None = None,
         idempotency_key: str | None = None,
         account_ref: str | None = None,
+        mutation_plan_id: str | None = None,
+        target_fingerprint: str | None = None,
     ) -> CapabilityInvokeResult:
         """Invoke one capability operation under caller scope."""
         normalized_user_id = _required_text(
@@ -709,6 +712,8 @@ async def invoke(
         normalized_source_display_name = _optional_text(source_display_name)
         normalized_idempotency_key = _optional_text(idempotency_key)
         normalized_account_ref = _optional_text(account_ref)
+        normalized_mutation_plan_id = _optional_text(mutation_plan_id)
+        normalized_target_fingerprint = _optional_text(target_fingerprint)
 
         account_ref: str | None = None
         provider_impl: CapabilityProvider | None = None
@@ -774,6 +779,22 @@ async def invoke(
             source_username=normalized_source_username,
             source_display_name=normalized_source_display_name,
         )
+        confirmed_plan_id: str | None = None
+        if _requires_mutation_confirmation(
+            capability_id=normalized_capability_id,
+            operation=normalized_operation,
+            provider=normalized_provider,
+            mutating=bool(op.mutating),
+        ):
+            confirmed_plan_id = _assert_mutation_confirmation_proof(
+                chat_id=normalized_chat_id,
+                thread_id=normalized_thread_id,
+                capability_id=normalized_capability_id,
+                operation=normalized_operation,
+                mutation_plan_id=normalized_mutation_plan_id,
+                target_fingerprint=normalized_target_fingerprint,
+            )
+
         raw_output = await self._provider_invoke(
             provider_impl,
             capability_id=normalized_capability_id,
@@ -791,6 +812,11 @@ async def invoke(
             )
 
         request_id = f"cap_{secrets.token_hex(8)}"
+        if confirmed_plan_id and normalized_chat_id:
+            _mark_mutation_plan_executed(
+                chat_id=normalized_chat_id,
+                plan_id=confirmed_plan_id,
+            )
 
         return CapabilityInvokeResult(
             request_id=request_id,
@@ -1135,6 +1161,72 @@ def _find_sensitive_key_path(value: Any, path: str = "output") -> str | None:
     return None
 
 
+def _requires_mutation_confirmation(
+    *,
+    capability_id: str,
+    operation: str,
+    provider: str | None,
+    mutating: bool,
+) -> bool:
+    if not mutating:
+        return False
+    if provider != "telegram":
+        return False
+    return capability_id == "gog.email" and operation in {
+        "archive_messages",
+        "update_labels",
+    }
+
+
+def _assert_mutation_confirmation_proof(
+    *,
+    chat_id: str | None,
+    thread_id: str | None,
+    capability_id: str,
+    operation: str,
+    mutation_plan_id: str | None,
+    target_fingerprint: str | None,
+) -> str:
+    normalized_chat_id = _optional_text(chat_id)
+    if normalized_chat_id is None:
+        raise CapabilityError(
+            "capability_mutation_not_confirmed",
+            "mutating operation requires chat-scoped confirmation proof",
+        )
+
+    manager = ChatStateManager(provider="telegram", chat_id=normalized_chat_id)
+    state = manager.load()
+    state.prune_expired_mutation_confirmations()
+    confirmed = state.find_confirmed_mutation(
+        capability_id=capability_id,
+        operation=operation,
+        target_fingerprint=target_fingerprint,
+        thread_id=thread_id,
+    )
+    if confirmed is None:
+        raise CapabilityError(
+            "capability_mutation_not_confirmed",
+            (
+                "mutation requires prior confirmation in chat history; "
+                "show targets and get explicit user confirm first"
+            ),
+        )
+    if mutation_plan_id and mutation_plan_id != confirmed.plan_id:
+        raise CapabilityError(
+            "capability_mutation_plan_mismatch",
+            "provided mutation_plan_id does not match confirmed chat plan",
+        )
+    manager.save()
+    return confirmed.plan_id
+
+
+def _mark_mutation_plan_executed(*, chat_id: str, plan_id: str) -> None:
+    manager = ChatStateManager(provider="telegram", chat_id=chat_id)
+    state = manager.load()
+    if state.mark_mutation_executed(plan_id=plan_id):
+        manager.save()
+
+
 async def create_capability_manager(
     *,
     providers: list[CapabilityProvider] | None = None,

src/ash/chats/models.py

@@ -1,6 +1,6 @@
 """Chat state models."""
 
-from datetime import UTC, datetime
+from datetime import UTC, datetime, timedelta
 
 from pydantic import BaseModel, Field
 
@@ -28,12 +28,32 @@ class ChatInfo(BaseModel):
     title: str | None = None
 
 
+class MutationConfirmation(BaseModel):
+    """Chat-scoped proof that a mutating operation was shown and confirmed."""
+
+    plan_id: str
+    capability_id: str
+    operation: str
+    status: str = "presented"  # presented | confirmed | executed
+    presented_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
+    expires_at: datetime
+    confirmed_at: datetime | None = None
+    executed_at: datetime | None = None
+    target_fingerprint: str | None = None
+    thread_id: str | None = None
+    summary: str | None = None
+
+
 class ChatState(BaseModel):
     """State for a chat, stored in state.json."""
 
     chat: ChatInfo
     participants: list[Participant] = Field(default_factory=list)
     thread_index: dict[str, str] = Field(default_factory=dict)
+    active_thread_id: str | None = None
+    active_thread_updated_at: datetime | None = None
+    active_thread_reason: str | None = None
+    mutation_confirmations: list[MutationConfirmation] = Field(default_factory=list)
     updated_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
     graph_chat_id: str | None = None  # Reference to graph ChatEntry.id
 
@@ -73,3 +93,144 @@ def update_participant(
 
         self.updated_at = now
         return participant
+
+    def set_active_thread(
+        self,
+        thread_id: str,
+        *,
+        reason: str,
+        now: datetime | None = None,
+    ) -> None:
+        """Record the active thread for chat-scoped DM routing."""
+        ts = now or datetime.now(UTC)
+        self.active_thread_id = str(thread_id)
+        self.active_thread_updated_at = ts
+        self.active_thread_reason = reason
+        self.updated_at = ts
+
+    def get_active_thread(
+        self,
+        *,
+        max_age_minutes: int,
+        now: datetime | None = None,
+    ) -> str | None:
+        """Return active thread_id if it is still within the freshness window."""
+        if not self.active_thread_id or not self.active_thread_updated_at:
+            return None
+        ts = now or datetime.now(UTC)
+        max_age = max(1, int(max_age_minutes))
+        if ts - self.active_thread_updated_at > timedelta(minutes=max_age):
+            return None
+        return self.active_thread_id
+
+    def add_mutation_confirmation(
+        self,
+        *,
+        plan_id: str,
+        capability_id: str,
+        operation: str,
+        target_fingerprint: str | None = None,
+        thread_id: str | None = None,
+        summary: str | None = None,
+        ttl_hours: int = 24,
+        now: datetime | None = None,
+    ) -> MutationConfirmation:
+        """Store a mutation confirmation prompt shown to the user."""
+        ts = now or datetime.now(UTC)
+        self.prune_expired_mutation_confirmations(now=ts)
+        confirmation = MutationConfirmation(
+            plan_id=plan_id,
+            capability_id=capability_id,
+            operation=operation,
+            expires_at=ts + timedelta(hours=max(1, int(ttl_hours))),
+            target_fingerprint=target_fingerprint,
+            thread_id=thread_id,
+            summary=summary,
+        )
+        self.mutation_confirmations.append(confirmation)
+        self.updated_at = ts
+        return confirmation
+
+    def confirm_latest_mutation(
+        self,
+        *,
+        now: datetime | None = None,
+        thread_id: str | None = None,
+    ) -> MutationConfirmation | None:
+        """Confirm the latest non-expired presented mutation plan."""
+        ts = now or datetime.now(UTC)
+        self.prune_expired_mutation_confirmations(now=ts)
+        for confirmation in reversed(self.mutation_confirmations):
+            if confirmation.status != "presented":
+                continue
+            if (
+                thread_id
+                and confirmation.thread_id
+                and confirmation.thread_id != thread_id
+            ):
+                continue
+            confirmation.status = "confirmed"
+            confirmation.confirmed_at = ts
+            self.updated_at = ts
+            return confirmation
+        return None
+
+    def find_confirmed_mutation(
+        self,
+        *,
+        capability_id: str,
+        operation: str,
+        target_fingerprint: str | None = None,
+        thread_id: str | None = None,
+        now: datetime | None = None,
+    ) -> MutationConfirmation | None:
+        """Find a non-expired confirmed mutation authorization."""
+        ts = now or datetime.now(UTC)
+        self.prune_expired_mutation_confirmations(now=ts)
+        for confirmation in reversed(self.mutation_confirmations):
+            if confirmation.status != "confirmed":
+                continue
+            if confirmation.capability_id != capability_id:
+                continue
+            if confirmation.operation != operation:
+                continue
+            if target_fingerprint and confirmation.target_fingerprint:
+                if confirmation.target_fingerprint != target_fingerprint:
+                    continue
+            if (
+                thread_id
+                and confirmation.thread_id
+                and confirmation.thread_id != thread_id
+            ):
+                continue
+            return confirmation
+        return None
+
+    def mark_mutation_executed(
+        self,
+        *,
+        plan_id: str,
+        now: datetime | None = None,
+    ) -> bool:
+        """Mark a confirmed mutation plan as executed."""
+        ts = now or datetime.now(UTC)
+        for confirmation in self.mutation_confirmations:
+            if confirmation.plan_id != plan_id:
+                continue
+            confirmation.status = "executed"
+            confirmation.executed_at = ts
+            self.updated_at = ts
+            return True
+        return False
+
+    def prune_expired_mutation_confirmations(
+        self,
+        *,
+        now: datetime | None = None,
+    ) -> None:
+        """Remove expired mutation confirmation entries."""
+        ts = now or datetime.now(UTC)
+        kept = [item for item in self.mutation_confirmations if item.expires_at > ts]
+        if len(kept) != len(self.mutation_confirmations):
+            self.mutation_confirmations = kept
+            self.updated_at = ts

src/ash/providers/telegram/handlers/message_handler.py

@@ -492,6 +492,8 @@ async def _process_single_message_inner(
             if isinstance(candidate, IncomingMessage):
                 message = candidate
 
+        self._session_handler.maybe_record_mutation_confirmation_from_user(message)
+
         if await self._try_handle_capability_oauth_callback(message):
             return