diff --git a/docs/chat-commands.md b/docs/chat-commands.md
index 31377340..96729be9 100644
--- a/docs/chat-commands.md
+++ b/docs/chat-commands.md
@@ -78,7 +78,8 @@ nanobot trigger trg_8K4P2Q9X "Review PR #4502"
 Replace `"Review PR #4502"` with the message you want nanobot to receive. The
 trigger is bound to the session where it was created, so the message goes back
 to that same chat. Keep `nanobot gateway` running so trigger messages can be
-delivered.
+delivered. The trigger message starts an automation turn; it is not shown in
+the chat as a normal user message.
 
 For longer or generated content, omit the message argument and pipe stdin:
 
diff --git a/nanobot/agent/loop.py b/nanobot/agent/loop.py
index 7e28d1e9..61ad0614 100644
--- a/nanobot/agent/loop.py
+++ b/nanobot/agent/loop.py
@@ -47,9 +47,6 @@ from nanobot.bus.runtime_events import (
 )
 from nanobot.command import CommandContext, CommandRouter, register_builtin_commands
 from nanobot.config.schema import AgentDefaults, ModelPresetConfig
-from nanobot.cron.session_turns import (
-    cron_history_overrides,
-)
 from nanobot.providers.base import LLMProvider
 from nanobot.providers.factory import ProviderSnapshot
 from nanobot.security.workspace_access import (
@@ -58,6 +55,7 @@ from nanobot.security.workspace_access import (
     reset_workspace_scope,
 )
 from nanobot.session import turn_continuation
+from nanobot.session.automation_turns import automation_history_overrides
 from nanobot.session.goal_state import (
     goal_state_runtime_lines,
     runner_wall_llm_timeout_s,
@@ -612,10 +610,10 @@ class AgentLoop:
             extra: dict[str, Any] = ({"media": list(media_paths)} if media_paths else {}) | agent_context.session_extra(msg.metadata)
             extra.update(kwargs)
             text = msg.content if isinstance(msg.content, str) else ""
-            text_override, cron_extra = cron_history_overrides(msg.metadata)
+            text_override, automation_extra = automation_history_overrides(msg.metadata)
             if text_override is not None:
                 text = text_override
-            extra.update(cron_extra)
+            extra.update(automation_extra)
             session.add_message("user", text, **extra)
             self._mark_pending_user_turn(session)
             self.sessions.save(session)
diff --git a/nanobot/cron/session_turns.py b/nanobot/cron/session_turns.py
index a85c4793..27622b83 100644
--- a/nanobot/cron/session_turns.py
+++ b/nanobot/cron/session_turns.py
@@ -5,16 +5,43 @@ from __future__ import annotations
 from typing import Any, Mapping
 
 from nanobot.cron.types import CronJob
+from nanobot.session.automation_turns import (
+    AutomationTurnSpec,
+    automation_history_overrides_for_spec,
+    automation_trigger,
+)
 
 CRON_TRIGGER_META = "_cron_trigger"
 CRON_DEFER_UNTIL_IDLE_META = "_cron_defer_until_session_idle"
 CRON_HISTORY_META = "_cron_turn"
 
 
+def _cron_history_text(trigger: Mapping[str, Any]) -> str | None:
+    persist_content = trigger.get("persist_content")
+    return (
+        persist_content
+        if isinstance(persist_content, str) and persist_content.strip()
+        else None
+    )
+
+
+CRON_AUTOMATION_SPEC = AutomationTurnSpec(
+    kind="cron",
+    trigger_meta_key=CRON_TRIGGER_META,
+    legacy_history_meta_key=CRON_HISTORY_META,
+    history_fields={
+        "cron_job_id": "job_id",
+        "cron_job_name": "job_name",
+        "cron_run_id": "run_id",
+        "cron_prompt_ref": "prompt_ref",
+    },
+    text_builder=_cron_history_text,
+)
+
+
 def cron_trigger(metadata: Mapping[str, Any] | None) -> dict[str, Any] | None:
     """Return structured cron trigger metadata when present."""
-    raw = (metadata or {}).get(CRON_TRIGGER_META)
-    return raw if isinstance(raw, dict) else None
+    return automation_trigger(metadata, CRON_AUTOMATION_SPEC)
 
 
 def is_cron_turn(metadata: Mapping[str, Any] | None) -> bool:
@@ -38,22 +65,7 @@ def cron_run_id(metadata: Mapping[str, Any] | None) -> str | None:
 
 def cron_history_overrides(metadata: Mapping[str, Any] | None) -> tuple[str | None, dict[str, Any]]:
     """Return session-history text/metadata overrides for a cron turn."""
-    trigger = cron_trigger(metadata)
-    if not trigger:
-        return None, {}
-    persist_content = trigger.get("persist_content")
-    text = (
-        persist_content
-        if isinstance(persist_content, str) and persist_content.strip()
-        else None
-    )
-    return text, {
-        CRON_HISTORY_META: True,
-        "cron_job_id": trigger.get("job_id"),
-        "cron_job_name": trigger.get("job_name"),
-        "cron_run_id": trigger.get("run_id"),
-        "cron_prompt_ref": trigger.get("prompt_ref"),
-    }
+    return automation_history_overrides_for_spec(metadata, CRON_AUTOMATION_SPEC)
 
 
 def is_bound_cron_job(job: CronJob) -> bool:
diff --git a/nanobot/session/automation_turns.py b/nanobot/session/automation_turns.py
new file mode 100644
index 00000000..116ac9c3
--- /dev/null
+++ b/nanobot/session/automation_turns.py
@@ -0,0 +1,90 @@
+"""Shared handling for session-bound automation turns."""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Mapping
+from dataclasses import dataclass, field
+from functools import lru_cache
+from typing import Any
+
+AUTOMATION_HISTORY_META = "_automation_turn"
+
+
+@dataclass(frozen=True)
+class AutomationTurnSpec:
+    """Source-specific wiring for one session-bound automation turn type."""
+
+    kind: str
+    trigger_meta_key: str
+    legacy_history_meta_key: str | None = None
+    history_fields: Mapping[str, str] = field(default_factory=dict)
+    text_builder: Callable[[Mapping[str, Any]], str | None] | None = None
+
+
+def automation_trigger(
+    metadata: Mapping[str, Any] | None,
+    spec: AutomationTurnSpec,
+) -> dict[str, Any] | None:
+    """Return source trigger metadata for *spec* when present."""
+    raw = (metadata or {}).get(spec.trigger_meta_key)
+    return raw if isinstance(raw, dict) else None
+
+
+def automation_history_overrides_for_spec(
+    metadata: Mapping[str, Any] | None,
+    spec: AutomationTurnSpec,
+) -> tuple[str | None, dict[str, Any]]:
+    """Return hidden session-history text/metadata overrides for *spec*."""
+    trigger = automation_trigger(metadata, spec)
+    if not trigger:
+        return None, {}
+
+    details: dict[str, Any] = {"kind": spec.kind}
+    extra: dict[str, Any] = {AUTOMATION_HISTORY_META: details}
+    if spec.legacy_history_meta_key:
+        extra[spec.legacy_history_meta_key] = True
+    for history_key, trigger_key in spec.history_fields.items():
+        value = trigger.get(trigger_key)
+        extra[history_key] = value
+        details[history_key] = value
+
+    text = spec.text_builder(trigger) if spec.text_builder else None
+    return text, extra
+
+
+@lru_cache(maxsize=1)
+def _automation_specs() -> tuple[AutomationTurnSpec, ...]:
+    # Source modules import the generic helpers above, so keep spec loading lazy.
+    from nanobot.cron.session_turns import CRON_AUTOMATION_SPEC
+    from nanobot.triggers.session_turns import EXTERNAL_TRIGGER_AUTOMATION_SPEC
+
+    return (CRON_AUTOMATION_SPEC, EXTERNAL_TRIGGER_AUTOMATION_SPEC)
+
+
+def automation_history_overrides(
+    metadata: Mapping[str, Any] | None,
+) -> tuple[str | None, dict[str, Any]]:
+    """Return session-history text/metadata overrides for supported automation turns."""
+    for spec in _automation_specs():
+        text, extra = automation_history_overrides_for_spec(metadata, spec)
+        if extra:
+            return text, extra
+    return None, {}
+
+
+def is_automation_history_message(message: Mapping[str, Any] | None) -> bool:
+    """True for hidden automation trigger records in session history."""
+    if not message:
+        return False
+    marker = message.get(AUTOMATION_HISTORY_META)
+    if marker is True or isinstance(marker, Mapping):
+        return True
+    return any(
+        spec.legacy_history_meta_key
+        and message.get(spec.legacy_history_meta_key) is True
+        for spec in _automation_specs()
+    )
+
+
+def is_automation_kind(value: Any) -> bool:
+    return isinstance(value, str) and any(spec.kind == value for spec in _automation_specs())
diff --git a/nanobot/session/webui_turns.py b/nanobot/session/webui_turns.py
index b97125e7..7a049b38 100644
--- a/nanobot/session/webui_turns.py
+++ b/nanobot/session/webui_turns.py
@@ -30,8 +30,8 @@ from nanobot.bus.runtime_events import (
     TurnCompleted,
     TurnRunStatusChanged,
 )
-from nanobot.cron.session_turns import CRON_HISTORY_META
 from nanobot.providers.base import LLMProvider
+from nanobot.session.automation_turns import is_automation_history_message
 from nanobot.session.goal_state import goal_state_ws_blob
 from nanobot.session.manager import Session, SessionManager
 from nanobot.utils.helpers import strip_think, truncate_text
@@ -77,7 +77,7 @@ def _title_inputs(session: Session) -> tuple[str, str]:
     for message in session.messages:
         if message.get("_command") is True:
             continue
-        if message.get(CRON_HISTORY_META) is True:
+        if is_automation_history_message(message):
             continue
         role = message.get("role")
         content = message.get("content")
diff --git a/nanobot/triggers/runner.py b/nanobot/triggers/runner.py
index b7dc8c99..ef75dccc 100644
--- a/nanobot/triggers/runner.py
+++ b/nanobot/triggers/runner.py
@@ -10,12 +10,11 @@ from loguru import logger
 
 from nanobot.bus.events import InboundMessage
 from nanobot.bus.queue import MessageBus
+from nanobot.triggers.session_turns import EXTERNAL_TRIGGER_META
 from nanobot.triggers.store import ExternalTriggerStore
 from nanobot.triggers.types import ExternalTrigger, TriggerDelivery
 from nanobot.webui.metadata import WEBUI_MESSAGE_SOURCE_METADATA_KEY, WEBUI_TURN_METADATA_KEY
 
-EXTERNAL_TRIGGER_META = "_external_trigger"
-
 
 async def run_external_trigger_queue(
     *,
diff --git a/nanobot/triggers/session_turns.py b/nanobot/triggers/session_turns.py
new file mode 100644
index 00000000..eeb36759
--- /dev/null
+++ b/nanobot/triggers/session_turns.py
@@ -0,0 +1,51 @@
+"""Shared metadata helpers for local external trigger session turns."""
+
+from __future__ import annotations
+
+from typing import Any, Mapping
+
+from nanobot.session.automation_turns import (
+    AutomationTurnSpec,
+    automation_history_overrides_for_spec,
+    automation_trigger,
+)
+
+EXTERNAL_TRIGGER_META = "_external_trigger"
+
+
+def _external_trigger_history_text(trigger: Mapping[str, Any]) -> str:
+    name = trigger.get("trigger_name")
+    trigger_id = trigger.get("trigger_id")
+    label = name if isinstance(name, str) and name.strip() else trigger_id
+    return (
+        f"External trigger received: {label}"
+        if isinstance(label, str) and label.strip()
+        else "External trigger received"
+    )
+
+
+EXTERNAL_TRIGGER_AUTOMATION_SPEC = AutomationTurnSpec(
+    kind="trigger",
+    trigger_meta_key=EXTERNAL_TRIGGER_META,
+    history_fields={
+        "trigger_id": "trigger_id",
+        "trigger_name": "trigger_name",
+        "trigger_delivery_id": "delivery_id",
+    },
+    text_builder=_external_trigger_history_text,
+)
+
+
+def external_trigger(metadata: Mapping[str, Any] | None) -> dict[str, Any] | None:
+    """Return structured external trigger metadata when present."""
+    return automation_trigger(metadata, EXTERNAL_TRIGGER_AUTOMATION_SPEC)
+
+
+def external_trigger_history_overrides(
+    metadata: Mapping[str, Any] | None,
+) -> tuple[str | None, dict[str, Any]]:
+    """Return session-history text/metadata overrides for an external trigger turn."""
+    return automation_history_overrides_for_spec(
+        metadata,
+        EXTERNAL_TRIGGER_AUTOMATION_SPEC,
+    )
diff --git a/nanobot/webui/session_automations.py b/nanobot/webui/session_automations.py
index e22be27e..51de158d 100644
--- a/nanobot/webui/session_automations.py
+++ b/nanobot/webui/session_automations.py
@@ -5,8 +5,8 @@ from __future__ import annotations
 from collections.abc import Collection
 from typing import Any, Protocol
 
-from nanobot.cron.session_turns import CRON_HISTORY_META
 from nanobot.cron.types import CronJob
+from nanobot.session.automation_turns import is_automation_history_message
 from nanobot.session.manager import _message_preview_text
 from nanobot.triggers.types import ExternalTrigger
 
@@ -326,7 +326,7 @@ def _session_preview(messages: Any) -> str:
     for message in messages:
         if not isinstance(message, dict):
             continue
-        if message.get(CRON_HISTORY_META) is True:
+        if is_automation_history_message(message):
             continue
         text = _message_preview_text(message)
         if not text:
diff --git a/nanobot/webui/session_list_index.py b/nanobot/webui/session_list_index.py
index ccadcd07..b8bf0478 100644
--- a/nanobot/webui/session_list_index.py
+++ b/nanobot/webui/session_list_index.py
@@ -16,7 +16,7 @@ from typing import Any
 from loguru import logger
 
 from nanobot.config.paths import get_webui_dir
-from nanobot.cron.session_turns import CRON_HISTORY_META
+from nanobot.session.automation_turns import is_automation_history_message
 from nanobot.session.manager import (
     _SESSION_LIST_PREVIEW_MAX_CHARS,
     _SESSION_LIST_PREVIEW_MAX_RECORDS,
@@ -154,7 +154,7 @@ def _preview_from_messages(messages: list[dict[str, Any]]) -> str:
             or scanned_chars > _SESSION_LIST_PREVIEW_MAX_CHARS
         ):
             break
-        if item.get(CRON_HISTORY_META) is True:
+        if is_automation_history_message(item):
             continue
         text = _message_preview_text(item)
         if not text:
@@ -216,7 +216,7 @@ def _latest_updated_at(stored: str | None, activity: str | None) -> str | None:
 
 
 def _visible_message_timestamp(item: dict[str, Any]) -> str | None:
-    if item.get(CRON_HISTORY_META) is True:
+    if is_automation_history_message(item):
         return None
     if item.get("role") not in _VISIBLE_TRANSCRIPT_ROLES:
         return None
@@ -296,7 +296,9 @@ def _scan_session_row(session_manager: SessionManager, path: Path) -> dict[str,
                     ):
                         preview_done = True
                         continue
-                    if item.get(CRON_HISTORY_META) is True:
+                    if item.get("_type") == "metadata":
+                        continue
+                    if is_automation_history_message(item):
                         continue
                     text = _message_preview_text(item)
                     if not text:
diff --git a/nanobot/webui/transcript.py b/nanobot/webui/transcript.py
index 67be9669..9c412bf7 100644
--- a/nanobot/webui/transcript.py
+++ b/nanobot/webui/transcript.py
@@ -17,7 +17,7 @@ from urllib.parse import unquote, urlparse
 from loguru import logger
 
 from nanobot.config.paths import get_webui_dir
-from nanobot.cron.session_turns import CRON_HISTORY_META
+from nanobot.session.automation_turns import is_automation_history_message, is_automation_kind
 from nanobot.session.manager import SessionManager
 from nanobot.webui.metadata import WEBUI_MESSAGE_SOURCE_METADATA_KEY, WEBUI_TURN_METADATA_KEY
 
@@ -598,9 +598,12 @@ def normalize_webui_turn_id(value: Any) -> str:
 
 def webui_message_source(metadata: dict[str, Any] | None) -> dict[str, str] | None:
     raw = (metadata or {}).get(WEBUI_MESSAGE_SOURCE_METADATA_KEY)
-    if not isinstance(raw, dict) or raw.get("kind") != "cron":
+    if not isinstance(raw, dict):
         return None
-    source: dict[str, str] = {"kind": "cron"}
+    kind = raw.get("kind")
+    if not is_automation_kind(kind):
+        return None
+    source: dict[str, str] = {"kind": kind}
     label = raw.get("label")
     if isinstance(label, str) and label.strip():
         source["label"] = label.strip()
@@ -779,6 +782,8 @@ def write_session_messages_as_transcript(
     target_chat_id = _chat_id_from_session_key(target_key)
     rows: list[dict[str, Any]] = []
     for msg in messages:
+        if is_automation_history_message(msg):
+            continue
         role = msg.get("role")
         content = msg.get("content")
         text = content if isinstance(content, str) else ""
@@ -855,7 +860,7 @@ def _session_user_event(
 ) -> dict[str, Any] | None:
     if message.get("role") != "user":
         return None
-    if message.get(CRON_HISTORY_META) is True:
+    if is_automation_history_message(message):
         return None
     content = message.get("content")
     text = content if isinstance(content, str) else ""
@@ -1271,9 +1276,12 @@ def replay_transcript_to_ui_messages(
 
     def _source_fields(rec: dict[str, Any]) -> dict[str, Any]:
         source = rec.get("source")
-        if not isinstance(source, dict) or source.get("kind") != "cron":
+        if not isinstance(source, dict):
             return {}
-        out: dict[str, Any] = {"source": {"kind": "cron"}}
+        kind = source.get("kind")
+        if not is_automation_kind(kind):
+            return {}
+        out: dict[str, Any] = {"source": {"kind": kind}}
         label = source.get("label")
         if isinstance(label, str) and label.strip():
             out["source"]["label"] = label.strip()
diff --git a/tests/agent/test_loop_save_turn.py b/tests/agent/test_loop_save_turn.py
index daac07a4..bfb7a38f 100644
--- a/tests/agent/test_loop_save_turn.py
+++ b/tests/agent/test_loop_save_turn.py
@@ -18,6 +18,7 @@ from nanobot.bus.outbound_events import (
 from nanobot.bus.queue import MessageBus
 from nanobot.cron.session_turns import CRON_HISTORY_META, CRON_TRIGGER_META
 from nanobot.providers.base import LLMResponse
+from nanobot.session.automation_turns import AUTOMATION_HISTORY_META
 from nanobot.session.goal_state import GOAL_STATE_KEY
 from nanobot.session.manager import Session, SessionManager
 from nanobot.session.turn_continuation import (
@@ -33,6 +34,7 @@ from nanobot.session.webui_turns import (
     clean_generated_title,
     maybe_generate_webui_title,
 )
+from nanobot.triggers.session_turns import EXTERNAL_TRIGGER_META
 from nanobot.utils.llm_runtime import LLMRuntime
 
 
@@ -101,6 +103,13 @@ def test_persist_cron_turn_uses_distinct_history_marker(tmp_path: Path) -> None:
     assert persisted is True
     message = session.messages[-1]
     assert message["content"] == "Scheduled cron job triggered: Daily check"
+    assert message[AUTOMATION_HISTORY_META] == {
+        "kind": "cron",
+        "cron_job_id": "job-1",
+        "cron_job_name": "Daily check",
+        "cron_run_id": "job-1:1",
+        "cron_prompt_ref": prompt_ref,
+    }
     assert message[CRON_HISTORY_META] is True
     assert CRON_TRIGGER_META not in message
     assert message["cron_job_id"] == "job-1"
@@ -109,6 +118,44 @@ def test_persist_cron_turn_uses_distinct_history_marker(tmp_path: Path) -> None:
     assert message["cron_prompt_ref"] == prompt_ref
 
 
+def test_persist_external_trigger_turn_uses_hidden_automation_marker(tmp_path: Path) -> None:
+    loop = _make_full_loop(tmp_path)
+    session = loop.sessions.get_or_create("websocket:auto")
+
+    persisted = loop._persist_user_message_early(
+        InboundMessage(
+            channel="websocket",
+            sender_id="trigger",
+            chat_id="auto",
+            content="Review PR #4502",
+            metadata={
+                EXTERNAL_TRIGGER_META: {
+                    "trigger_id": "trg_123",
+                    "trigger_name": "PR review",
+                    "delivery_id": "tdel_456",
+                    "created_at_ms": 1_700_000_000_000,
+                }
+            },
+        ),
+        session,
+    )
+
+    assert persisted is True
+    message = session.messages[-1]
+    assert message["content"] == "External trigger received: PR review"
+    assert "Review PR #4502" not in message["content"]
+    assert message[AUTOMATION_HISTORY_META] == {
+        "kind": "trigger",
+        "trigger_id": "trg_123",
+        "trigger_name": "PR review",
+        "trigger_delivery_id": "tdel_456",
+    }
+    assert EXTERNAL_TRIGGER_META not in message
+    assert message["trigger_id"] == "trg_123"
+    assert message["trigger_name"] == "PR review"
+    assert message["trigger_delivery_id"] == "tdel_456"
+
+
 def test_clean_generated_title_strips_reasoning_tags() -> None:
     assert clean_generated_title("<think>reasoning</think> WebUI polish") == "WebUI polish"
     assert clean_generated_title("Title: <think> The user said hello") == ""
diff --git a/tests/channels/test_websocket_channel.py b/tests/channels/test_websocket_channel.py
index 2a0562bc..0fe89ddc 100644
--- a/tests/channels/test_websocket_channel.py
+++ b/tests/channels/test_websocket_channel.py
@@ -2975,3 +2975,49 @@ def test_handle_webui_thread_get_does_not_backfill_cron_internal_prompt(
     body = json.loads(resp.body.decode())
     assert [message["role"] for message in body["messages"]] == ["assistant"]
     assert [message["content"] for message in body["messages"]] == ["提醒已经到期。"]
+
+
+def test_handle_webui_thread_get_does_not_backfill_trigger_internal_prompt(
+    tmp_path,
+    monkeypatch,
+) -> None:
+    from urllib.parse import quote
+
+    from websockets.datastructures import Headers
+    from websockets.http11 import Request
+
+    from nanobot.session.automation_turns import AUTOMATION_HISTORY_META
+    from nanobot.webui.transcript import append_transcript_object
+
+    monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path)
+    workspace = tmp_path / "workspace"
+    sessions = SessionManager(workspace)
+    key = "websocket:c-trigger"
+    session = sessions.get_or_create(key)
+    session.add_message(
+        "user",
+        "External trigger received: PR review",
+        **{AUTOMATION_HISTORY_META: {"kind": "trigger", "trigger_id": "trg_123"}},
+    )
+    session.add_message("assistant", "PR #4502 已经开始 review。")
+    sessions.save(session)
+    append_transcript_object(
+        key,
+        {"event": "message", "chat_id": "c-trigger", "text": "PR #4502 已经开始 review。"},
+    )
+
+    bus = MagicMock()
+    channel = WebSocketChannel(
+        {"enabled": True, "allowFrom": ["*"]},
+        bus,
+        gateway=_basic_handler(bus, session_manager=sessions, workspace_path=workspace),
+    )
+    channel.gateway.tokens.api_tokens["tok"] = time.monotonic() + 300.0
+    enc = quote(key, safe="")
+    req = Request(f"/api/sessions/{enc}/webui-thread", Headers([("Authorization", "Bearer tok")]))
+    resp = channel.gateway.http._handle_webui_thread_get(req, enc)
+
+    assert resp.status_code == 200
+    body = json.loads(resp.body.decode())
+    assert [message["role"] for message in body["messages"]] == ["assistant"]
+    assert [message["content"] for message in body["messages"]] == ["PR #4502 已经开始 review。"]
diff --git a/tests/utils/test_webui_transcript.py b/tests/utils/test_webui_transcript.py
index f79cb319..6482a0e8 100644
--- a/tests/utils/test_webui_transcript.py
+++ b/tests/utils/test_webui_transcript.py
@@ -473,6 +473,24 @@ def test_replay_reused_turn_id_after_turn_end_starts_new_turn(tmp_path, monkeypa
     assert msgs[2]["source"] == {"kind": "cron", "label": "drink water"}
 
 
+def test_replay_preserves_trigger_source_metadata(tmp_path, monkeypatch) -> None:
+    monkeypatch.setattr("nanobot.config.paths.get_data_dir", lambda: tmp_path)
+    key = "websocket:t-trigger-source"
+    append_transcript_object(
+        key,
+        {
+            "event": "message",
+            "chat_id": "t-trigger-source",
+            "text": "PR #4502 review started.",
+            "source": {"kind": "trigger", "label": "PR review"},
+        },
+    )
+
+    msgs = replay_transcript_to_ui_messages(read_transcript_lines(key))
+
+    assert msgs[0]["source"] == {"kind": "trigger", "label": "PR review"}
+
+
 def test_build_response_restores_session_users_for_legacy_transcript(
     tmp_path,
     monkeypatch,
diff --git a/tests/webui/test_session_list_index.py b/tests/webui/test_session_list_index.py
index e65ca8de..9fe72118 100644
--- a/tests/webui/test_session_list_index.py
+++ b/tests/webui/test_session_list_index.py
@@ -6,6 +6,7 @@ from pathlib import Path
 
 import nanobot.webui.session_list_index as session_list_index
 from nanobot.cron.session_turns import CRON_HISTORY_META
+from nanobot.session.automation_turns import AUTOMATION_HISTORY_META
 from nanobot.session.manager import SessionManager
 
 
@@ -88,6 +89,20 @@ def test_webui_session_list_skips_cron_internal_user_preview(tmp_path: Path) ->
     assert list_webui_sessions(manager)[0]["preview"] == "提醒已经到期。"
 
 
+def test_webui_session_list_skips_trigger_internal_user_preview(tmp_path: Path) -> None:
+    manager = SessionManager(tmp_path)
+    session = manager.get_or_create("websocket:trigger-preview")
+    session.add_message(
+        "user",
+        "External trigger received: PR review",
+        **{AUTOMATION_HISTORY_META: {"kind": "trigger", "trigger_id": "trg_123"}},
+    )
+    session.add_message("assistant", "PR #4502 已经开始 review。")
+    manager.save(session)
+
+    assert list_webui_sessions(manager)[0]["preview"] == "PR #4502 已经开始 review。"
+
+
 def test_webui_session_list_uses_webui_transcript_activity_for_sort(
     tmp_path: Path,
     monkeypatch,