diff --git a/src/agent_env/explorer/app.py b/src/agent_env/explorer/app.py
index d23c801c..658c897a 100644
--- a/src/agent_env/explorer/app.py
+++ b/src/agent_env/explorer/app.py
@@ -24,6 +24,7 @@
 from agent_env.explorer.routers import conversations as conversations_router
 from agent_env.explorer.routers import objects as objects_router
 from agent_env.explorer.routers import runs as runs_router
+from agent_env.explorer.routers import triggers as triggers_router
 from agent_env.explorer.routers.common import versioned_router
 from agent_env.store import NotFoundError
 from agent_env.store.routing import configured_store
@@ -207,6 +208,7 @@ async def _not_found(request, exc: NotFoundError):
     app.include_router(versioned_router(prefix=f"{API}/agents", tag="agents", collection="a2a_agents", noun="agent"))
     app.include_router(versioned_router(prefix=f"{API}/evals", tag="evals", collection="evals", noun="eval"))
     app.include_router(runs_router.router)
+    app.include_router(triggers_router.router)
     app.include_router(objects_router.router)
     app.include_router(conversations_router.router)
 
diff --git a/src/agent_env/explorer/routers/triggers.py b/src/agent_env/explorer/routers/triggers.py
new file mode 100644
index 00000000..a1c44148
--- /dev/null
+++ b/src/agent_env/explorer/routers/triggers.py
@@ -0,0 +1,168 @@
+"""A run's trigger evidence, the feed the run viewer's Triggers tab polls."""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import logging
+from datetime import datetime, timezone
+
+import httpx
+from fastapi import APIRouter, HTTPException
+
+from agent_env.config import get_config
+from agent_env.env.store import ENV_INSTANCES_COLLECTION
+from agent_env.explorer.entity_ids import EntityId
+from agent_env.explorer.routers.common import docs
+from agent_env.store import Filter, In
+from agent_env.task.store import TASK_INSTANCES_COLLECTION
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter(prefix="/api/v1/tasks", tags=["runs"])
+
+TRIGGER_STATE_TIMEOUT_SECONDS = 2.0
+# Both fan-outs are driven by the run context, which a run request can supply.
+MAX_TRIGGER_ENVS = 20
+TRIGGER_METADATA_KEYS = (
+    "env_trigger_registrations",
+    "agent_trigger_registrations",
+    "agent_trigger_firings",
+    "agent_trigger_state",
+    "env_trigger_snapshots",
+    "usersim_turn_outputs",
+    "env_trigger_state",
+    "env_trajectory",
+)
+# Envelope keys the metadata ledger has no copy of.
+_ENVELOPE_META_KEYS = ("clock", "clock_read_at_utc")
+
+
+@router.get("/{task_id}/instances/{instance_id}/triggers")
+async def instance_triggers(task_id: EntityId, instance_id: EntityId) -> dict:
+    """The run's trigger metadata, plus each env's verbatim ``/triggers/state`` under ``state``:
+    from the live gateway while the run is running, else from the envelope ``prompt_agent`` saved.
+    An unreachable source degrades to the metadata; ``state_meta`` carries the gateway clock the
+    events' ``virtual_time`` is measured on."""
+    doc = await asyncio.to_thread(
+        docs().find_one, TASK_INSTANCES_COLLECTION, Filter.of(task_id=task_id, instance_id=instance_id)
+    )
+    if doc is None:
+        raise HTTPException(status_code=404, detail=f"instance {instance_id} not found for task {task_id}")
+
+    context = doc.get("context") or {}
+    metadata = context.get("metadata") or {}
+    ledger = {key: metadata.get(key) for key in TRIGGER_METADATA_KEYS}
+    captures = ledger["env_trigger_state"] if isinstance(ledger["env_trigger_state"], dict) else {}
+    ledger["env_trigger_state"] = captures
+    payload = {
+        "status": doc.get("status"),
+        "source": "metadata",
+        **ledger,
+        "state": {},
+        "state_sources": {},
+        "state_meta": {},
+    }
+
+    # Per env: one env's live gateway must not suppress another's envelope.
+    if payload["status"] == "running":
+        live, live_meta = await _live_trigger_state(context.get("deployed_envs") or [])
+        for env_id, state in live.items():
+            payload["state_sources"][env_id] = "live" if state is not None else "unavailable"
+            if state is not None:
+                payload["state"][env_id] = state
+        payload["state_meta"].update(live_meta)
+
+    pending = {
+        env_id: entry
+        for env_id, entry in list(captures.items())[:MAX_TRIGGER_ENVS]
+        if isinstance(entry, dict) and env_id not in payload["state"]
+    }
+    if pending:
+        saved, saved_meta = await asyncio.to_thread(_saved_trigger_state, pending, instance_id)
+        payload["state"].update(saved)
+        payload["state_meta"].update(saved_meta)
+        payload["state_sources"].update({env_id: "artifact" for env_id in saved})
+
+    sources = set(payload["state_sources"].values())
+    if sources & {"live", "artifact"}:
+        payload["source"] = "live" if "live" in sources else "artifact"
+    return payload
+
+
+def _live_gateways(deployed_envs: list) -> list[tuple[str, str | None]]:
+    """``(env_id, gateway_url or None)`` per deployed env, from the env-instance records the deploy
+    writes rather than the run context's own copy, which a run request can supply."""
+    pairs = [
+        (env["env_id"], env["instance_id"])
+        for env in deployed_envs
+        if isinstance(env, dict) and isinstance(env.get("env_id"), str) and isinstance(env.get("instance_id"), str)
+    ][:MAX_TRIGGER_ENVS]
+    if not pairs:
+        return []
+    records = docs().query(ENV_INSTANCES_COLLECTION, Filter().where("instance_id", In([iid for _, iid in pairs])))
+    urls = {record["instance_id"]: (record.get("gateway_url") or "").rstrip("/") or None for record in records}
+    return [(env_id, urls.get(instance_id)) for env_id, instance_id in pairs]
+
+
+async def _live_trigger_state(deployed_envs: list) -> tuple[dict, dict]:
+    """Each env's ``/triggers/state`` and gateway clock, read concurrently. An env with no reachable
+    gateway maps to None, so "unreachable" never reads as "no triggers"."""
+    targets = await asyncio.to_thread(_live_gateways, deployed_envs)
+    state: dict = {env_id: None for env_id, _ in targets}
+    meta: dict = {}
+    reachable = [(env_id, url) for env_id, url in targets if url]
+    if not reachable:
+        return state, meta
+
+    async with httpx.AsyncClient(timeout=TRIGGER_STATE_TIMEOUT_SECONDS) as client:
+
+        async def _get(url: str, env_id: str, what: str):
+            try:
+                response = await client.get(url)
+                response.raise_for_status()
+                return response.json()
+            except Exception:
+                logger.warning("live %s unavailable for %s", what, env_id, exc_info=True)
+                return None
+
+        async def _one(env_id: str, gateway_url: str):
+            triggers, clock = await asyncio.gather(
+                _get(f"{gateway_url}/triggers/state", env_id, "trigger state"),
+                _get(f"{gateway_url}/clock/state", env_id, "clock state"),
+            )
+            # Stamped at the read: at a high clock rate a second of drift is a virtual day.
+            return env_id, triggers, {"clock": clock, "clock_read_at_utc": datetime.now(timezone.utc).isoformat()}
+
+        for env_id, triggers, env_meta in await asyncio.gather(*(_one(e, u) for e, u in reachable)):
+            state[env_id] = triggers
+            if triggers is not None:
+                meta[env_id] = env_meta
+    return state, meta
+
+
+def _saved_trigger_state(entries: dict, instance_id: str) -> tuple[dict, dict]:
+    """Each env's ``/triggers/state`` from the envelope ``prompt_agent`` saved, with its capture-time
+    clock. Skips a url outside the configured object store, an unreadable envelope, and one another
+    instance captured (a resumed run's context carries the earlier attempt's pointers)."""
+    state: dict = {}
+    meta: dict = {}
+    for env_id, entry in entries.items():
+        object_url = entry.get("object_url")
+        if not isinstance(object_url, str):
+            continue
+        store = get_config().get_object_store_at(object_url)
+        if not store.owns(object_url):
+            logger.warning("refusing a trigger-state url outside the object store for %s", env_id)
+            continue
+        try:
+            envelope = json.loads(store.get(object_url))
+            captured_by = envelope.get("instance_id")
+            if captured_by and captured_by != instance_id:
+                logger.warning("skipping the trigger state for %s captured by %s", env_id, captured_by)
+                continue
+            state[env_id] = envelope.get("state") or {}
+            meta[env_id] = {key: envelope.get(key) for key in _ENVELOPE_META_KEYS}
+        except Exception:
+            logger.warning("trigger-state envelope unreadable for %s", env_id, exc_info=True)
+    return state, meta
diff --git a/src/agent_env/explorer/ui/src/components/task-instance-viewer.tsx b/src/agent_env/explorer/ui/src/components/task-instance-viewer.tsx
index 5b3aa993..19dc2624 100644
--- a/src/agent_env/explorer/ui/src/components/task-instance-viewer.tsx
+++ b/src/agent_env/explorer/ui/src/components/task-instance-viewer.tsx
@@ -31,7 +31,7 @@ import {
   type ServerConfigChange,
   type ServerConfigSkip,
 } from './server-config-panel';
-import { TriggerTurnStrip } from './triggers-panel';
+import { TriggersPanel, TriggerTurnStrip } from './triggers-panel';
 import { parseTriggerRuntime } from '../lib/parse-trigger-runtime';
 import { indexAuthoredTriggers } from '../lib/parse-triggers';
 import {
@@ -1042,6 +1042,11 @@ export function TaskInstanceViewer({
                 Server Config
               </Tabs.Trigger>
             )}
+            {triggerRuntime && (
+              <Tabs.Trigger value="triggers" className="my-1">
+                Triggers
+              </Tabs.Trigger>
+            )}
             {hasPeerAgents && (
               <Tabs.Trigger value="peer-qna" className="my-1">
                 Peer Q&amp;A
@@ -1806,6 +1811,17 @@ export function TaskInstanceViewer({
           </Tabs.Content>
         )}
 
+        {triggerRuntime && (
+          <Tabs.Content value="triggers" className="p-4">
+            <TriggersPanel
+              runtime={triggerRuntime}
+              authored={authoredTriggers}
+              taskId={taskId}
+              instanceId={instanceId}
+            />
+          </Tabs.Content>
+        )}
+
         {hasPeerAgents && (
           <Tabs.Content value="peer-qna" className="p-0">
             <PeerQnAPanel
diff --git a/tst/unit/explorer/explorer_app_test.py b/tst/unit/explorer/explorer_app_test.py
index d77809ad..7fae5b1b 100644
--- a/tst/unit/explorer/explorer_app_test.py
+++ b/tst/unit/explorer/explorer_app_test.py
@@ -2,10 +2,12 @@
 
 import dataclasses
 import gzip
+import json
 import tempfile
 import time
 from urllib.parse import quote
 
+import httpx
 import pytest
 
 pytest.importorskip("fastapi")  # the explorer is the optional [explorer] extra (fastapi, uvicorn)
@@ -20,6 +22,7 @@
 from agent_env.store.document_store.sqlite_document_store import LocalSqliteDocumentStore
 from agent_env.task import Task
 from agent_env.explorer.routers import objects as objects_router
+from agent_env.explorer.routers import triggers as triggers_router
 from tst.unit.store.fakes import FakeObjectStore
 
 
@@ -388,6 +391,102 @@ def test_run_groups_tally_step_progress_for_the_batch_funnel(client):
     assert listed[gid]["step_counts"] == expected
 
 
+# --- Triggers ---------------------------------------------------------------
+
+
+def _seed_trigger_run(status="completed", metadata=None, deployed_envs=None, task_id="t1", instance_id="i-trig"):
+    get_config().get_document_store().insert("task_instances", {
+        "instance_id": instance_id, "task_id": task_id, "task_version": 1, "status": status,
+        "context": {"metadata": metadata or {}, "deployed_envs": deployed_envs or []},
+    })
+
+
+def _envelope(instance_id="i-trig"):
+    return {
+        "instance_id": instance_id, "env_id": "crm", "capture_source": "prompt_agent",
+        "clock": {"rate": 60.0, "virtual_now": "2026-01-04T09:00:00+00:00"},
+        "clock_read_at_utc": "2026-01-04T00:00:05+00:00",
+        "state": {"triggers": [{"id": "follow-up", "status": "fired", "fire_count": 1}], "events": [{"kind": "fired"}]},
+    }
+
+
+def test_triggers_report_the_metadata_ledger(client):
+    registrations = [{"env_id": "crm", "trigger_ids": ["follow-up"]}]
+    _seed_trigger_run(metadata={"env_trigger_registrations": registrations, "agent_trigger_firings": [{"turn": 1}]})
+
+    body = client.get("/api/v1/tasks/t1/instances/i-trig/triggers").json()
+
+    assert set(triggers_router.TRIGGER_METADATA_KEYS) <= set(body)
+    assert body["status"] == "completed" and body["source"] == "metadata"
+    assert body["env_trigger_registrations"] == registrations
+    assert body["agent_trigger_firings"] == [{"turn": 1}]
+    assert body["env_trigger_state"] == {} and body["usersim_turn_outputs"] is None
+    assert (body["state"], body["state_sources"], body["state_meta"]) == ({}, {}, {})
+
+
+def test_triggers_read_a_finished_runs_saved_envelope(client, tmp_path):
+    store = LocalFilesystemObjectStore(str(tmp_path / "obj"))
+    set_object_store(store)
+    url = store.put("env_trigger_state/instance_id=i-trig/crm-1.json", json.dumps(_envelope()).encode())
+    _seed_trigger_run(metadata={"env_trigger_state": {"crm": {"object_url": url, "event_count": 1}}})
+
+    body = client.get("/api/v1/tasks/t1/instances/i-trig/triggers").json()
+
+    assert body["source"] == "artifact" and body["state_sources"] == {"crm": "artifact"}
+    assert body["state"] == {"crm": _envelope()["state"]}
+    assert body["state_meta"] == {"crm": {"clock": _envelope()["clock"], "clock_read_at_utc": _envelope()["clock_read_at_utc"]}}
+
+
+def test_triggers_skip_an_envelope_outside_the_store_or_from_another_instance(client, tmp_path):
+    store = LocalFilesystemObjectStore(str(tmp_path / "obj"))
+    set_object_store(store)
+    foreign = store.put("env_trigger_state/instance_id=other/db-1.json", json.dumps(_envelope("other")).encode())
+    _seed_trigger_run(metadata={"env_trigger_state": {
+        "crm": {"object_url": "s3://some-bucket/env_trigger_state/crm.json"},
+        "db": {"object_url": foreign},
+        "web": {"error": "capture budget of 60s exceeded"},
+    }})
+
+    body = client.get("/api/v1/tasks/t1/instances/i-trig/triggers").json()
+
+    assert body["source"] == "metadata"
+    assert (body["state"], body["state_sources"], body["state_meta"]) == ({}, {}, {})
+    assert set(body["env_trigger_state"]) == {"crm", "db", "web"}
+
+
+def test_triggers_read_the_live_gateway_while_running(client, monkeypatch):
+    get_config().get_document_store().insert("env_instances", {"instance_id": "crm-abc", "env_id": "crm", "gateway_url": "http://gateway.test/"})
+    _seed_trigger_run(status="running", deployed_envs=[
+        {"env_id": "crm", "instance_id": "crm-abc", "gateway_url": "http://ignored.test"},
+        {"env_id": "db", "instance_id": "db-gone"},
+    ])
+    live = {"http://gateway.test/triggers/state": {"triggers": [{"id": "follow-up", "status": "armed"}]},
+            "http://gateway.test/clock/state": {"rate": 60.0}}
+    transport = httpx.MockTransport(lambda request: httpx.Response(200, json=live[str(request.url)]))
+    real_client = httpx.AsyncClient
+    monkeypatch.setattr(httpx, "AsyncClient", lambda **kw: real_client(transport=transport, **kw))
+
+    body = client.get("/api/v1/tasks/t1/instances/i-trig/triggers").json()
+
+    assert body["source"] == "live" and body["state_sources"] == {"crm": "live", "db": "unavailable"}
+    assert body["state"] == {"crm": live["http://gateway.test/triggers/state"]}
+    assert body["state_meta"]["crm"]["clock"] == {"rate": 60.0}
+    assert "db" not in body["state_meta"]
+
+
+def test_triggers_for_an_unknown_instance_are_404(client):
+    assert client.get("/api/v1/tasks/t1/instances/ghost/triggers").status_code == 404
+
+
+def test_triggers_address_a_namespaced_task_and_instance(client):
+    task_id, instance_id = "@local/~/triage/runs", "@local/~/triage/runs-7f3a9c2e"
+    _seed_trigger_run(task_id=task_id, instance_id=instance_id, metadata={"env_trigger_registrations": []})
+
+    got = client.get(f"/api/v1/tasks/{quote(task_id, safe='')}/instances/{quote(instance_id, safe='')}/triggers")
+
+    assert got.status_code == 200 and got.json()["env_trigger_registrations"] == []
+
+
 # --- Docs surface -----------------------------------------------------------