Fix scheduled reply routing and DM chat-type propagation

- treat replies to bot messages as active even when thread index misses

- normalize thread index keys to strings for consistent lookup

- propagate schedule chat_type through CLI/RPC/store into scheduled session context

- add regression tests for schedule chat_type and reply-skip policy

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

Author: dcramer

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

@@ -22,6 +22,7 @@ def _get_context() -> dict[str, str]:
     return {
         "user_id": context.get("user_id") or "",
         "chat_id": context.get("chat_id") or "",
+        "chat_type": context.get("chat_type") or "",
         "chat_title": context.get("chat_title") or "",
         "provider": context.get("provider") or "",
         "username": context.get("username") or "",
@@ -257,6 +258,8 @@ def create(
         params["cron"] = cron
     if ctx["chat_title"]:
         params["chat_title"] = ctx["chat_title"]
+    if ctx["chat_type"]:
+        params["chat_type"] = ctx["chat_type"]
     if ctx["user_id"]:
         params["user_id"] = ctx["user_id"]
     if ctx["username"]:

src/ash/chats/thread_index.py

@@ -51,26 +51,28 @@ def resolve_thread_id(
             returns the same thread_id. Otherwise returns external_id as a new thread.
         """
         index = self._ensure_loaded()
+        external_key = str(external_id)
+        reply_key = str(reply_to_external_id) if reply_to_external_id else None
 
-        if reply_to_external_id:
-            parent_thread = index.get(reply_to_external_id)
+        if reply_key:
+            parent_thread = index.get(reply_key)
             if parent_thread:
                 logger.debug(
                     "Message %s joins thread %s (via reply to %s)",
-                    external_id,
+                    external_key,
                     parent_thread,
-                    reply_to_external_id,
+                    reply_key,
                 )
                 return parent_thread
             logger.debug(
                 "Message %s replied to unknown message %s, starting new thread",
-                external_id,
-                reply_to_external_id,
+                external_key,
+                reply_key,
             )
 
         # Start new thread using this message's ID as the thread_id
-        logger.debug("Message %s starts new thread", external_id)
-        return external_id
+        logger.debug("Message %s starts new thread", external_key)
+        return external_key
 
     def register_message(self, external_id: str, thread_id: str) -> None:
         """Register a message in a thread.
@@ -81,11 +83,13 @@ def register_message(self, external_id: str, thread_id: str) -> None:
         """
         with self._lock:
             index = self._ensure_loaded()
-            if external_id not in index:
-                index[external_id] = thread_id
+            external_key = str(external_id)
+            thread_key = str(thread_id)
+            if external_key not in index:
+                index[external_key] = thread_key
                 self._manager.save()
                 logger.debug(
-                    "Registered message %s in thread %s", external_id, thread_id
+                    "Registered message %s in thread %s", external_key, thread_key
                 )
 
     def get_thread_id(self, external_id: str) -> str | None:
@@ -98,4 +102,4 @@ def get_thread_id(self, external_id: str) -> str | None:
             The thread_id if the message is registered, None otherwise.
         """
         index = self._ensure_loaded()
-        return index.get(external_id)
+        return index.get(str(external_id))

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

@@ -372,6 +372,8 @@ async def should_skip_reply(self, message: IncomingMessage) -> bool:
             return False
         if message.metadata.get("was_mentioned", False):
             return False
+        if message.metadata.get("is_reply_to_bot", False):
+            return False
 
         # Check thread index first
         thread_index = self.get_thread_index(message.chat_id)

src/ash/providers/telegram/provider.py

@@ -471,12 +471,14 @@ def _to_incoming_message(
         images: list[ImageAttachment] | None = None,
         *,
         was_mentioned: bool = False,
+        is_reply_to_bot: bool = False,
     ) -> IncomingMessage:
         """Convert a Telegram message to an IncomingMessage."""
         metadata = {
             "chat_type": message.chat.type,
             "chat_title": message.chat.title,
             "was_mentioned": was_mentioned,
+            "is_reply_to_bot": is_reply_to_bot,
         }
         # Include thread_id for forum topics (supergroups with topics enabled)
         if message.message_thread_id is not None:
@@ -621,6 +623,7 @@ async def handle_photo(message: TelegramMessage) -> None:
             # Strip bot mention from caption if in group
             is_group = message.chat.type in ("group", "supergroup")
             was_mentioned = is_group and self._is_mentioned(message)
+            is_reply_to_bot = is_group and self._is_reply(message)
             caption = message.caption or ""
             if is_group and caption:
                 caption = self._strip_mention(caption)
@@ -632,6 +635,7 @@ async def handle_photo(message: TelegramMessage) -> None:
                 caption,
                 images=[image],
                 was_mentioned=was_mentioned,
+                is_reply_to_bot=is_reply_to_bot,
             )
 
             if self._handler:
@@ -655,10 +659,16 @@ async def handle_message(message: TelegramMessage) -> None:
             # Strip bot mention from text if in group
             is_group = message.chat.type in ("group", "supergroup")
             was_mentioned = is_group and self._is_mentioned(message)
+            is_reply_to_bot = is_group and self._is_reply(message)
             text = self._strip_mention(message.text) if is_group else message.text
 
             incoming = self._to_incoming_message(
-                message, user_id, username, text, was_mentioned=was_mentioned
+                message,
+                user_id,
+                username,
+                text,
+                was_mentioned=was_mentioned,
+                is_reply_to_bot=is_reply_to_bot,
             )
             # Add processing mode to metadata
             incoming.metadata["processing_mode"] = processing_mode

src/ash/rpc/methods/schedule.py

@@ -35,6 +35,7 @@ async def schedule_create(params: dict[str, Any]) -> dict[str, Any]:
             trigger_at: ISO datetime for one-shot (mutually exclusive with cron)
             cron: Cron expression for periodic (mutually exclusive with trigger_at)
             chat_id: Target chat ID (required)
+            chat_type: Chat type for policy checks at execution time (optional)
             provider: Provider name (required)
             user_id: User ID
             username: Username
@@ -67,6 +68,7 @@ async def schedule_create(params: dict[str, Any]) -> dict[str, Any]:
             trigger_at=trigger_at,
             cron=cron,
             chat_id=chat_id,
+            chat_type=params.get("chat_type"),
             chat_title=params.get("chat_title"),
             provider=provider,
             user_id=params.get("user_id"),

src/ash/scheduling/handler.py

@@ -90,6 +90,21 @@ def format_delay(seconds: float) -> str:
     return f"~{days:.1f} days"
 
 
+def _resolve_chat_type(entry: ScheduleEntry) -> str | None:
+    """Resolve effective chat_type for scheduled executions.
+
+    New entries should carry chat_type directly. For older Telegram entries
+    without this field, infer private vs group from Telegram chat ID shape.
+    """
+    if entry.chat_type:
+        return entry.chat_type
+
+    if entry.provider == "telegram" and entry.chat_id:
+        return "group" if entry.chat_id.startswith("-") else "private"
+
+    return None
+
+
 class MessageSender(Protocol):
     """Protocol for sending messages to a chat. Returns the sent message ID."""
 
@@ -217,6 +232,7 @@ async def handle(self, entry: ScheduleEntry) -> None:
         # Populate context so system prompt builder includes full context
         session.context.username = entry.username or ""
         session.context.is_scheduled_task = True
+        session.context.chat_type = _resolve_chat_type(entry)
         if entry.chat_title:
             session.context.chat_title = entry.chat_title
 

src/ash/scheduling/types.py

@@ -31,6 +31,7 @@ class ScheduleEntry:
     timezone: str | None = None
     # Context for routing response back
     chat_id: str | None = None
+    chat_type: str | None = None  # "private", "group", "supergroup", ...
     chat_title: str | None = None  # Friendly name for the chat
     user_id: str | None = None
     username: str | None = None  # For @mentions in response
@@ -234,6 +235,8 @@ def to_dict(self) -> dict[str, Any]:
         # Context fields
         if self.chat_id:
             data["chat_id"] = self.chat_id
+        if self.chat_type:
+            data["chat_type"] = self.chat_type
         if self.chat_title:
             data["chat_title"] = self.chat_title
         if self.user_id:
@@ -290,6 +293,7 @@ def parse_datetime(key: str) -> datetime | None:
             "last_run",
             "timezone",
             "chat_id",
+            "chat_type",
             "chat_title",
             "user_id",
             "username",
@@ -306,6 +310,7 @@ def parse_datetime(key: str) -> datetime | None:
             last_run=last_run,
             timezone=data.get("timezone"),
             chat_id=data.get("chat_id"),
+            chat_type=data.get("chat_type"),
             chat_title=data.get("chat_title"),
             user_id=data.get("user_id"),
             username=data.get("username"),

tests/test_sandbox_cli.py

@@ -15,13 +15,15 @@ def _context_token(
     *,
     effective_user_id: str = "user123",
     chat_id: str | None = "chat456",
+    chat_type: str | None = "private",
     provider: str | None = "telegram",
     source_username: str | None = "testuser",
     timezone: str | None = "UTC",
 ) -> str:
     return get_default_context_token_service().issue(
         effective_user_id=effective_user_id,
         chat_id=chat_id,
+        chat_type=chat_type,
         provider=provider,
         source_username=source_username,
         timezone=timezone,
@@ -82,6 +84,7 @@ def test_create_one_shot(self, cli_runner, mock_rpc):
         params = call_args[1]
         assert params["message"] == "Test reminder"
         assert params["chat_id"] == "chat456"
+        assert params["chat_type"] == "private"
         assert params["provider"] == "telegram"
 
     def test_create_periodic(self, cli_runner, mock_rpc):

tests/test_schedule.py

@@ -1627,6 +1627,41 @@ async def test_handler_uses_configured_timezone(self):
         # The previous 8 AM LA should show as 08:00, not UTC equivalent
         assert "08:00" in wrapped_message
 
+    @pytest.mark.asyncio
+    async def test_handler_sets_session_chat_type_for_dm_policy(self):
+        """Scheduled tasks in DMs preserve private chat_type for skill policy checks."""
+        from unittest.mock import AsyncMock, MagicMock
+
+        from ash.scheduling import ScheduledTaskHandler
+
+        mock_agent = MagicMock()
+        mock_response = MagicMock()
+        mock_response.text = "ok"
+        mock_agent.process_message = AsyncMock(return_value=mock_response)
+
+        mock_sender = AsyncMock(return_value="msg_123")
+        handler = ScheduledTaskHandler(
+            agent=mock_agent,
+            senders={"telegram": mock_sender},
+            timezone="UTC",
+        )
+
+        entry = ScheduleEntry(
+            id="dm_ctx_1",
+            message="Check my calendar",
+            trigger_at=datetime.now(UTC) - timedelta(minutes=1),
+            provider="telegram",
+            chat_id="123456789",
+            chat_type="private",
+            user_id="42",
+        )
+
+        await handler.handle(entry)
+
+        call_args = mock_agent.process_message.call_args
+        session = call_args.args[1]
+        assert session.context.chat_type == "private"
+
 
 class TestStalenessGuard:
     """Tests for the staleness guard in ScheduleWatcher."""

tests/test_sessions.py

@@ -1096,3 +1096,32 @@ def test_session_context_has_no_chat_history_field(self):
 
         ctx = SessionContext()
         assert not hasattr(ctx, "chat_history")
+
+
+class TestGroupReplySkipPolicy:
+    @pytest.mark.asyncio
+    async def test_reply_to_bot_is_not_skipped_when_thread_unknown(self):
+        from ash.config.models import ConversationConfig
+        from ash.providers.base import IncomingMessage
+        from ash.providers.telegram.handlers.session_handler import SessionHandler
+
+        handler = SessionHandler(
+            provider_name="telegram",
+            config=None,
+            conversation_config=ConversationConfig(),
+        )
+
+        message = IncomingMessage(
+            id="201",
+            chat_id="-1001",
+            user_id="u1",
+            text="follow up",
+            reply_to_message_id="200",
+            metadata={
+                "chat_type": "group",
+                "was_mentioned": False,
+                "is_reply_to_bot": True,
+            },
+        )
+
+        assert await handler.should_skip_reply(message) is False