Python: feat(python): cross-channel hosting improvements (endpoint paths, Activity push, Telegram/Teams fixes) (#6307)

* Update hosting channel endpoint paths

Treat channel paths as concrete endpoint paths so built-in channels can be mounted at their defaults or at the app root without sample-specific subclasses. Update docs, tests, and the Foundry Telegram Invocations sample accordingly.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Add push support to ActivityProtocolChannel

Implement the ChannelPush protocol so the Activity Protocol channel can
receive cross-channel fan-out (ResponseTarget.all_linked) and echo_input
replay as a non-originating destination:

- Add push() that reconstructs a proactive Bot Framework activity (bot/user
  swap) from the stored conversation reference and POSTs it to
  /v3/conversations/{id}/activities.
- Record a ChannelIdentity (service_url, conversation, bot, user, channel_id,
  locale) on ChannelRequest.identity so the host registers the channel under
  its isolation key for fan-out resolution.
- Route the streaming path through deliver_response so Activity-originated
  turns broadcast like Telegram/Discord.
- Add tests for push delivery, service_url validation, ChannelPush instance
  check, and inbound identity recording.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Don't delete Telegram webhook on shutdown by default

The TelegramChannel deleted its webhook on shutdown in webhook mode. During
a rolling redeploy the new revision registers the webhook on startup, then
the old revision's shutdown deletes it, silently breaking inbound delivery
until the next boot. setWebhook is overwriting/idempotent, so startup
re-asserts the webhook every boot and no teardown is needed.

Add a delete_webhook_on_shutdown flag (default False) so teardown is opt-in
for ephemeral deployments, and leave the webhook in place otherwise.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* Fix Activity channel streaming on non-Teams channels (405 on updateActivity)

The Activity Protocol channel streamed replies the Teams way: POST a
placeholder, then PUT-edit it as tokens arrive. Only Teams supports the
updateActivity REST op; Web Chat, Direct Line and the Emulator return
405 Method Not Allowed on the PUT, so the user saw only the placeholder.

Gate the placeholder+edit flow on edit-capable channels (msteams). Other
channels now buffer the stream and POST a single final message, mirroring
the non-streaming path's fan-out and response-hook semantics. Also add a
defensive 405 fallback inside the Teams edit loop so an unexpected 405
can never strand the user on the placeholder.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix(hosting-activity-protocol): don't parse Teams inline attachment content as a URI

Teams message activities include a text/html attachment whose inline
`content` is raw HTML (not a URL). _parse_activity fell back to
`attachment["content"]` and passed it to Content.from_uri, raising
ContentError ("URI must contain a scheme") and failing the whole turn,
so Teams users got no response.

Only treat `contentUrl` as a URI, require an absolute scheme, and skip
unparseable attachments defensively instead of failing the message.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* feat(hosting-activity-protocol): native slash-command dispatch for Teams/Activity

Add a commands= parameter to ActivityProtocolChannel that intercepts a
leading /command (after stripping the bot's own @mention) and dispatches
to ChannelCommand handlers, mirroring the Telegram channel. Unknown
commands fall through to the agent. The channel run_hook is applied to
command requests so handlers observe the same resolved isolation key as
ordinary messages, and handler errors are swallowed (200, no Bot Service
retry of non-idempotent commands).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* feat(hosting): silent attributed Telegram echoes + Teams markdown rendering

- hosting-telegram: send cross-channel input echoes with disable_notification
  (silent) and detect echo payloads so they aren't re-broadcast.
- hosting-activity-protocol: render outbound + push activities as textFormat
  'markdown' so Teams shows formatted replies (enables per-channel variants).

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

* fix(hosting-activity-protocol): address PR #6307 review feedback

Consult the host delivery pipeline even for empty streamed replies so
ResponseTarget.none is honoured and non-originating fan-out is consulted
instead of always emitting an originating "(no response)" message. Applies
to both the progressive-edit (Teams) and buffered (Web Chat/Direct Line)
streaming paths.

Re-validate service_url against the allow-list in push(): the identity is
read from a persisted store and push runs out-of-band, so the captured
service_url must be re-checked before a bearer token is sent.

Adds tests for empty-stream host consultation/suppression on both streaming
paths and for push rejecting a disallowed service_url.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

---------

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

python/packages/hosting-telegram/agent_framework_hosting_telegram/_channel.py

@@ -88,6 +88,21 @@ def _text_result(text: str) -> HostedRunResult[AgentResponse]:
     return HostedRunResult(AgentResponse(messages=[Message(role="assistant", contents=[Content.from_text(text=text)])]))
 
 
+def _is_echo_payload(payload: HostedRunResult[AgentResponse]) -> bool:
+    """Return ``True`` when a push payload is an echoed user turn.
+
+    Per the :class:`~agent_framework_hosting.ChannelPush` contract the host
+    mirrors the originating user's input as a one-or-more message
+    :class:`~agent_framework.AgentResponse` with every ``role == "user"``,
+    delivered *before* the agent's (``role == "assistant"``) reply. Treating a
+    payload whose messages are all user-role as an echo lets the channel pick
+    echo-only delivery options (e.g. silent notifications) without the host
+    having to thread an explicit ``is_echo`` flag through ``push``.
+    """
+    messages = getattr(payload.result, "messages", None) or []
+    return bool(messages) and all(getattr(m, "role", None) == "user" for m in messages)
+
+
 def _telegram_media_file_id(message: Mapping[str, Any]) -> tuple[str, str] | None:
     """Return ``(file_id, fallback_media_type)`` for any media on the message."""
     photo = message.get("photo")
@@ -151,7 +166,7 @@ class TelegramChannel:
         self,
         *,
         bot_token: str,
-        path: str = "/telegram",
+        path: str = "/telegram/webhook",
         commands: Sequence[ChannelCommand] = (),
         register_native_commands: bool = True,
         run_hook: ChannelRunHook | None = None,
@@ -159,6 +174,7 @@ class TelegramChannel:
         api_base: str = "https://api.telegram.org",
         webhook_url: str | None = None,
         secret_token: str | None = None,
+        delete_webhook_on_shutdown: bool = False,
         parse_mode: str | None = None,
         send_typing_action: bool = True,
         transport: Literal["auto", "polling", "webhook"] = "auto",
@@ -179,6 +195,7 @@ class TelegramChannel:
         self._api = f"{api_base}/bot{bot_token}"
         self._webhook_url = webhook_url
         self._secret_token = secret_token
+        self._delete_webhook_on_shutdown = delete_webhook_on_shutdown
         self._parse_mode = parse_mode
         self._send_typing_action = send_typing_action
         if transport == "auto":
@@ -210,7 +227,7 @@ class TelegramChannel:
         self._ctx = context
         routes: list[BaseRoute] = []
         if self._transport == "webhook":
-            routes.append(Route("/webhook", self._handle, methods=["POST"]))
+            routes.append(Route("/", self._handle, methods=["POST"]))
         return ChannelContribution(
             routes=routes,
             commands=self._commands,
@@ -258,7 +275,7 @@ class TelegramChannel:
             logger.info("Telegram polling started (long-poll timeout=%ss)", self._polling_timeout)
 
     async def _on_shutdown(self) -> None:
-        """Stop the polling task, drain in-flight workers, drop the webhook, close HTTP.
+        """Stop the polling task, drain in-flight workers, close HTTP.
 
         Drain order:
         1. Cancel the poll task so no new updates are admitted.
@@ -269,10 +286,17 @@ class TelegramChannel:
            ``_update_tasks`` (the webhook handler returns 200 immediately
            and runs the agent in a background task, which the previous
            shutdown ignored entirely).
-        4. Best-effort `deleteWebhook` and HTTP client close.
+        4. Close the HTTP client.
 
-        Webhook teardown is best-effort — failures (e.g. revoked token at
-        shutdown) are logged but never raised so app shutdown can complete.
+        The webhook registration is intentionally **left in place** on
+        shutdown. A Telegram webhook is a single global resource, so
+        deleting it here races rolling redeploys: the new revision calls
+        ``setWebhook`` on startup, then the old revision's shutdown would
+        delete it, silently breaking inbound delivery until the next boot.
+        ``setWebhook`` is overwriting/idempotent, so the next startup
+        re-asserts it anyway. Set ``delete_webhook_on_shutdown=True`` to opt
+        into best-effort teardown (e.g. for a one-off/ephemeral deployment);
+        failures are logged but never raised so app shutdown can complete.
         """
         if self._poll_task is not None:
             self._poll_task.cancel()
@@ -296,7 +320,7 @@ class TelegramChannel:
                 await task
         self._update_tasks.clear()
         if self._http is not None:
-            if self._transport == "webhook":
+            if self._transport == "webhook" and self._delete_webhook_on_shutdown:
                 try:
                     await self._http.post(f"{self._api}/deleteWebhook")
                 except Exception:  # pragma: no cover - best-effort cleanup
@@ -845,7 +869,17 @@ class TelegramChannel:
             raise ValueError(f"Telegram push requires an int chat_id, got {identity.native_id!r}") from exc
         if self._http is None:
             raise RuntimeError("TelegramChannel.push called before startup")
-        await self._send(chat_id, payload.result.text)
+        # The Bot API can only ever send AS the bot, so there is no way to
+        # impersonate the user for an echo (the MTProto ``send_as`` field is
+        # not exposed to bots). The next best UX is to deliver echoes
+        # *silently* (``disable_notification``) so a mirrored input doesn't
+        # buzz the user's device the way a genuine reply does. Echo phases are
+        # identified per the ChannelPush contract: a payload whose messages are
+        # all ``role == "user"`` is the originating turn mirrored here.
+        extra: dict[str, Any] = {}
+        if _is_echo_payload(payload):
+            extra["disable_notification"] = True
+        await self._send(chat_id, payload.result.text, **extra)
 
     async def _send_photo(self, chat_id: int, photo_url: str, caption: str | None = None) -> None:
         """POST a ``sendPhoto`` to Telegram with an optional caption."""