Merge PR #4330: feat(webui): add automation management view

feat(webui): add automation management view

tests/channels/test_websocket_channel.py

@@ -119,6 +119,42 @@ async def _http_get(url: str, headers: dict[str, str] | None = None) -> httpx.Re
     )
 
 
+@pytest.mark.asyncio
+async def test_send_session_updated_broadcasts_to_other_webui_connections(bus) -> None:
+    class Conn:
+        remote_address = None
+
+        def __init__(self) -> None:
+            self.sent: list[str] = []
+
+        async def send(self, raw: str) -> None:
+            self.sent.append(raw)
+
+    channel = _ch(bus)
+    active_conn = Conn()
+    other_conn = Conn()
+    channel._attach(active_conn, "chat-a")
+    channel._attach(other_conn, "chat-b")
+    assert sorted(channel._subs) == ["chat-a", "chat-b"]
+    assert sum(len(conns) for conns in channel._subs.values()) == 2
+
+    await channel.send_session_updated("chat-a", scope="thread")
+
+    active_events = [json.loads(raw)["event"] for raw in active_conn.sent]
+    other_events = [json.loads(raw)["event"] for raw in other_conn.sent]
+
+    assert (active_events, other_events) == (
+        ["session_updated"],
+        ["session_updated"],
+    )
+    payload = json.loads(other_conn.sent[0])
+    assert payload == {
+        "event": "session_updated",
+        "chat_id": "chat-a",
+        "scope": "thread",
+    }
+
+
 async def _recv_ws_event(client: Any, event: str) -> dict[str, Any]:
     """Receive until a specific websocket event appears."""
     for _ in range(10):
@@ -128,6 +164,10 @@ async def _recv_ws_event(client: Any, event: str) -> dict[str, Any]:
     raise AssertionError(f"websocket event {event!r} was not received")
 
 
+def _sent_ws_payloads(mock_ws: AsyncMock) -> list[dict[str, Any]]:
+    return [json.loads(call.args[0]) for call in mock_ws.send.await_args_list]
+
+
 def test_normalize_http_path_strips_trailing_slash_except_root() -> None:
     assert _normalize_http_path("/chat/") == "/chat"
     assert _normalize_http_path("/chat?x=1") == "/chat"
@@ -1234,9 +1274,10 @@ async def test_send_turn_end_emits_turn_end_event() -> None:
         metadata={"_turn_end": True},
     ))
 
-    mock_ws.send.assert_awaited_once()
-    body = json.loads(mock_ws.send.await_args.args[0])
-    assert body == {"event": "turn_end", "chat_id": "chat-1"}
+    assert _sent_ws_payloads(mock_ws) == [
+        {"event": "turn_end", "chat_id": "chat-1"},
+        {"event": "session_updated", "chat_id": "chat-1", "scope": "thread"},
+    ]
 
 
 @pytest.mark.asyncio
@@ -1253,9 +1294,10 @@ async def test_send_turn_end_includes_latency_ms_when_present() -> None:
         metadata={"_turn_end": True, "latency_ms": 1500},
     ))
 
-    mock_ws.send.assert_awaited_once()
-    body = json.loads(mock_ws.send.await_args.args[0])
-    assert body == {"event": "turn_end", "chat_id": "chat-1", "latency_ms": 1500}
+    assert _sent_ws_payloads(mock_ws) == [
+        {"event": "turn_end", "chat_id": "chat-1", "latency_ms": 1500},
+        {"event": "session_updated", "chat_id": "chat-1", "scope": "thread"},
+    ]
 
 
 @pytest.mark.asyncio
@@ -1273,9 +1315,10 @@ async def test_send_turn_end_includes_goal_state_when_present() -> None:
         metadata={"_turn_end": True, "goal_state": blob},
     ))
 
-    mock_ws.send.assert_awaited_once()
-    body = json.loads(mock_ws.send.await_args.args[0])
-    assert body == {"event": "turn_end", "chat_id": "chat-1", "goal_state": blob}
+    assert _sent_ws_payloads(mock_ws) == [
+        {"event": "turn_end", "chat_id": "chat-1", "goal_state": blob},
+        {"event": "session_updated", "chat_id": "chat-1", "scope": "thread"},
+    ]
 
 
 @pytest.mark.asyncio

tests/channels/test_websocket_http_routes.py

@@ -3,12 +3,14 @@
 import asyncio
 import functools
 import json
+import random
+import socket
 import threading
 import time
 from pathlib import Path
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock
-from urllib.parse import urlencode
+from urllib.parse import quote, urlencode
 
 import httpx
 import pytest
@@ -23,6 +25,18 @@ from nanobot.webui.gateway_services import GatewayServices, build_gateway_servic
 _PORT = 29900
 
 
+def _free_port() -> int:
+    for _ in range(100):
+        port = random.randint(30_000, 60_000)
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+            try:
+                sock.bind(("127.0.0.1", port))
+            except OSError:
+                continue
+            return port
+    raise RuntimeError("could not find a free localhost port")
+
+
 def _make_handler(
     cfg: dict[str, Any] | WebSocketConfig,
     bus: Any,
@@ -813,6 +827,255 @@ async def test_session_delete_removes_file(
         await server_task
 
 
+@pytest.mark.asyncio
+async def test_webui_automations_route_lists_all_jobs_and_allows_user_actions(
+    bus: MagicMock, tmp_path: Path
+) -> None:
+    port = _free_port()
+    base_url = f"http://127.0.0.1:{port}"
+    cron = CronService(tmp_path / "cron" / "jobs.json")
+    user_job = cron.add_job(
+        name="Daily repo check",
+        schedule=CronSchedule(kind="every", every_ms=86_400_000),
+        message="Check the repo status",
+        session_key="websocket:abc",
+        origin_channel="websocket",
+        origin_chat_id="abc",
+    )
+    incomplete_job = cron.add_job(
+        name="english-quiz",
+        schedule=CronSchedule(kind="every", every_ms=3_600_000),
+        message="Practice English",
+        session_key="unified:default",
+    )
+    external_job = cron.add_job(
+        name="WeChat quiz",
+        schedule=CronSchedule(kind="every", every_ms=3_600_000),
+        message="Send a quiz",
+        session_key="weixin:wx-chat",
+        origin_channel="weixin",
+        origin_chat_id="wx-chat",
+    )
+    past_one_shot_job = cron.add_job(
+        name="Past one-shot",
+        schedule=CronSchedule(kind="at", at_ms=1),
+        message="Old one-shot message",
+        session_key="websocket:abc",
+        origin_channel="websocket",
+        origin_chat_id="abc",
+        delete_after_run=True,
+    )
+    cron.register_system_job(
+        CronJob(
+            id="heartbeat",
+            name="heartbeat",
+            schedule=CronSchedule(kind="every", every_ms=60_000),
+            payload=CronPayload(kind="system_event"),
+        )
+    )
+    session_manager = _seed_session(tmp_path, key="websocket:abc")
+    external_session = Session(key="weixin:wx-chat")
+    external_session.add_message("user", "Scheduled cron job triggered")
+    session_manager.save(external_session)
+    channel = _ch(
+        bus,
+        session_manager=session_manager,
+        cron_service=cron,
+        cron_pending_job_ids=lambda key: {user_job.id} if key == "websocket:abc" else set(),
+        port=port,
+    )
+    server_task = asyncio.create_task(channel.start())
+    await asyncio.sleep(0.3)
+    try:
+        deny = await _http_get(f"{base_url}/api/webui/automations")
+        assert deny.status_code == 401, deny.text
+
+        boot = await _http_get(f"{base_url}/webui/bootstrap")
+        token = boot.json()["token"]
+        auth = {"Authorization": f"Bearer {token}"}
+        resp = await _http_get(
+            f"{base_url}/api/webui/automations",
+            headers=auth,
+        )
+        assert resp.status_code == 200
+        assert "wx-chat" not in resp.text
+        assert "unified:default" not in resp.text
+        body = resp.json()
+        by_id = {job["id"]: job for job in body["jobs"]}
+        assert by_id[user_job.id]["protected"] is False
+        assert by_id[user_job.id]["state"]["pending"] is True
+        assert by_id[user_job.id]["state"]["run_history"] == []
+        assert by_id[user_job.id]["origin"]["session_key"] == "websocket:abc"
+        assert by_id[user_job.id]["origin"]["preview"] == "hi"
+        assert "session_key" not in by_id[incomplete_job.id]["payload"]
+        assert "origin_channel" not in by_id[incomplete_job.id]["payload"]
+        assert "origin_chat_id" not in by_id[incomplete_job.id]["payload"]
+        assert by_id[incomplete_job.id]["origin"] is None
+        assert "session_key" not in by_id[external_job.id]["payload"]
+        assert "origin_channel" not in by_id[external_job.id]["payload"]
+        assert "origin_chat_id" not in by_id[external_job.id]["payload"]
+        assert by_id[external_job.id]["origin"]["channel"] == "weixin"
+        assert "session_key" not in by_id[external_job.id]["origin"]
+        assert "chat_id" not in by_id[external_job.id]["origin"]
+        assert by_id[external_job.id]["origin"]["preview"] == ""
+        assert by_id["heartbeat"]["protected"] is True
+
+        updated = await _http_get(
+            f"{base_url}/api/webui/automations/update?id={user_job.id}",
+            headers={
+                **auth,
+                "X-Nanobot-Automation-Values": json.dumps(
+                    {
+                        "name": "Daily quiz",
+                        "message": "Ask the daily quiz",
+                        "schedule": {
+                            "kind": "cron",
+                            "expr": "0 9 * * *",
+                            "tz": "UTC",
+                        },
+                    }
+                ),
+            },
+        )
+        assert updated.status_code == 200
+        by_id = {job["id"]: job for job in updated.json()["jobs"]}
+        assert by_id[user_job.id]["name"] == "Daily quiz"
+        assert by_id[user_job.id]["payload"]["message"] == "Ask the daily quiz"
+        assert by_id[user_job.id]["schedule"]["kind"] == "cron"
+        assert by_id[user_job.id]["schedule"]["expr"] == "0 9 * * *"
+        assert by_id[user_job.id]["schedule"]["tz"] == "UTC"
+
+        unicode_update = await _http_get(
+            f"{base_url}/api/webui/automations/update?id={user_job.id}",
+            headers={
+                **auth,
+                "X-Nanobot-Automation-Values": quote(
+                    json.dumps(
+                        {
+                            "name": "每日测验",
+                            "message": "问今日测验",
+                        },
+                        ensure_ascii=False,
+                    ),
+                    safe="",
+                ),
+            },
+        )
+        assert unicode_update.status_code == 200
+        assert cron.get_job(user_job.id).name == "每日测验"
+        assert cron.get_job(user_job.id).payload.message == "问今日测验"
+
+        malformed_update = await _http_get(
+            f"{base_url}/api/webui/automations/update?id={user_job.id}",
+            headers={
+                **auth,
+                "X-Nanobot-Automation-Values": json.dumps({"message": ["bad"]}),
+            },
+        )
+        assert malformed_update.status_code == 400
+        assert cron.get_job(user_job.id).payload.message == "问今日测验"
+
+        invalid_cron_update = await _http_get(
+            f"{base_url}/api/webui/automations/update?id={user_job.id}",
+            headers={
+                **auth,
+                "X-Nanobot-Automation-Values": json.dumps(
+                    {"schedule": {"kind": "cron", "expr": "not a cron", "tz": "UTC"}}
+                ),
+            },
+        )
+        assert invalid_cron_update.status_code == 400
+        assert cron.get_job(user_job.id).schedule.expr == "0 9 * * *"
+
+        past_one_shot_update = await _http_get(
+            f"{base_url}/api/webui/automations/update?id={past_one_shot_job.id}",
+            headers={
+                **auth,
+                "X-Nanobot-Automation-Values": json.dumps(
+                    {
+                        "message": "Updated one-shot message",
+                        "schedule": {"kind": "at", "at_ms": 1},
+                    }
+                ),
+            },
+        )
+        assert past_one_shot_update.status_code == 200
+        assert cron.get_job(past_one_shot_job.id).payload.message == "Updated one-shot message"
+        assert cron.get_job(past_one_shot_job.id).schedule.at_ms == 1
+
+        protected_update = await _http_get(
+            f"{base_url}/api/webui/automations/update?id=heartbeat",
+            headers={
+                **auth,
+                "X-Nanobot-Automation-Values": json.dumps({"name": "bad"}),
+            },
+        )
+        assert protected_update.status_code == 403
+
+        disabled = await _http_get(
+            f"{base_url}/api/webui/automations/disable?id={user_job.id}",
+            headers=auth,
+        )
+        assert disabled.status_code == 200
+        by_id = {job["id"]: job for job in disabled.json()["jobs"]}
+        assert by_id[user_job.id]["enabled"] is False
+
+        disabled_run = await _http_get(
+            f"{base_url}/api/webui/automations/run?id={user_job.id}",
+            headers=auth,
+        )
+        assert disabled_run.status_code == 409
+
+        unbound_run = await _http_get(
+            f"{base_url}/api/webui/automations/run?id={incomplete_job.id}",
+            headers=auth,
+        )
+        assert unbound_run.status_code == 409
+        assert "no linked chat" in unbound_run.text
+
+        unbound_enable = await _http_get(
+            f"{base_url}/api/webui/automations/enable?id={incomplete_job.id}",
+            headers=auth,
+        )
+        assert unbound_enable.status_code == 409
+        assert "no linked chat" in unbound_enable.text
+
+        protected_delete = await _http_get(
+            f"{base_url}/api/webui/automations/delete?id=heartbeat",
+            headers=auth,
+        )
+        assert protected_delete.status_code == 403
+        protected_disable = await _http_get(
+            f"{base_url}/api/webui/automations/disable?id=heartbeat",
+            headers=auth,
+        )
+        assert protected_disable.status_code == 403
+        protected_run = await _http_get(
+            f"{base_url}/api/webui/automations/run?id=heartbeat",
+            headers=auth,
+        )
+        assert protected_run.status_code == 403
+
+        enabled = await _http_get(
+            f"{base_url}/api/webui/automations/enable?id={user_job.id}",
+            headers=auth,
+        )
+        assert enabled.status_code == 200
+        by_id = {job["id"]: job for job in enabled.json()["jobs"]}
+        assert by_id[user_job.id]["enabled"] is True
+
+        deleted = await _http_get(
+            f"{base_url}/api/webui/automations/delete?id={user_job.id}",
+            headers=auth,
+        )
+        assert deleted.status_code == 200
+        assert user_job.id not in {job["id"] for job in deleted.json()["jobs"]}
+        assert "heartbeat" in {job["id"] for job in deleted.json()["jobs"]}
+    finally:
+        await channel.stop()
+        await server_task
+
+
 @pytest.mark.asyncio
 async def test_session_delete_blocks_when_bound_automation_exists(
     bus: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch

tests/cron/test_cron_service.py

@@ -17,6 +17,14 @@ async def _wait_until(predicate, *, timeout: float = 1.0, interval: float = 0.01
     assert predicate()
 
 
+def _bound_chat(chat_id: str = "chat-1") -> dict[str, str]:
+    return {
+        "session_key": f"websocket:{chat_id}",
+        "origin_channel": "websocket",
+        "origin_chat_id": chat_id,
+    }
+
+
 def test_add_job_rejects_unknown_timezone(tmp_path) -> None:
     service = CronService(tmp_path / "cron" / "jobs.json")
 
@@ -37,12 +45,74 @@ def test_add_job_accepts_valid_timezone(tmp_path) -> None:
         name="tz ok",
         schedule=CronSchedule(kind="cron", expr="0 9 * * *", tz="America/Vancouver"),
         message="hello",
+        **_bound_chat(),
     )
 
     assert job.schedule.tz == "America/Vancouver"
     assert job.state.next_run_at_ms is not None
 
 
+@pytest.mark.asyncio
+async def test_unbound_agent_jobs_are_disabled_on_add(tmp_path) -> None:
+    called: list[str] = []
+
+    async def on_job(job):
+        called.append(job.id)
+
+    service = CronService(
+        tmp_path / "cron" / "jobs.json",
+        on_job=on_job,
+    )
+    job = service.add_job(
+        name="unbound",
+        schedule=CronSchedule(kind="every", every_ms=60_000),
+        message="hello",
+    )
+
+    assert job.enabled is False
+    assert job.state.next_run_at_ms is None
+    assert job.state.last_status == "error"
+    assert "missing bound session delivery context" in (job.state.last_error or "")
+    assert await service.run_job(job.id, force=True) is False
+    assert called == []
+
+
+def test_unbound_agent_jobs_are_disabled_on_load(tmp_path) -> None:
+    store_path = tmp_path / "cron" / "jobs.json"
+    store_path.parent.mkdir(parents=True)
+    store_path.write_text(
+        json.dumps(
+            {
+                "version": 1,
+                "jobs": [
+                    {
+                        "id": "unbound-1",
+                        "name": "Unbound reminder",
+                        "enabled": True,
+                        "schedule": {"kind": "every", "everyMs": 60_000},
+                        "payload": {
+                            "kind": "agent_turn",
+                            "message": "check status",
+                        },
+                        "state": {"nextRunAtMs": 1},
+                        "createdAtMs": 1,
+                        "updatedAtMs": 1,
+                    }
+                ],
+            }
+        ),
+        encoding="utf-8",
+    )
+
+    job = CronService(store_path).get_job("unbound-1")
+
+    assert job is not None
+    assert job.enabled is False
+    assert job.state.next_run_at_ms is None
+    assert job.state.last_status == "error"
+    assert "missing bound session delivery context" in (job.state.last_error or "")
+
+
 def test_add_job_migrates_legacy_delivery_context(tmp_path) -> None:
     service = CronService(tmp_path / "cron" / "jobs.json")
     meta = {"slack": {"thread_ts": "1234567890.123456", "channel_type": "channel"}}
@@ -263,6 +333,7 @@ async def test_execute_job_records_run_history(tmp_path) -> None:
         name="hist",
         schedule=CronSchedule(kind="every", every_ms=60_000),
         message="hello",
+        **_bound_chat(),
     )
     await service.run_job(job.id)
 
@@ -287,6 +358,7 @@ async def test_run_history_records_errors(tmp_path) -> None:
         name="fail",
         schedule=CronSchedule(kind="every", every_ms=60_000),
         message="hello",
+        **_bound_chat(),
     )
     await service.run_job(job.id)
 
@@ -308,6 +380,7 @@ async def test_run_history_records_skipped_jobs(tmp_path) -> None:
         name="skip",
         schedule=CronSchedule(kind="every", every_ms=60_000),
         message="hello",
+        **_bound_chat(),
     )
     await service.run_job(job.id)
 
@@ -332,7 +405,7 @@ async def test_run_history_records_job_cancellation(tmp_path) -> None:
         name="cancel",
         schedule=CronSchedule(kind="every", every_ms=60_000),
         message="hello",
-        session_key="websocket:chat-1",
+        **_bound_chat(),
     )
 
     assert await service.run_job(job.id) is True
@@ -355,6 +428,7 @@ async def test_run_history_trimmed_to_max(tmp_path) -> None:
         name="trim",
         schedule=CronSchedule(kind="every", every_ms=60_000),
         message="hello",
+        **_bound_chat(),
     )
     for _ in range(25):
         await service.run_job(job.id)
@@ -371,6 +445,7 @@ async def test_run_history_persisted_to_disk(tmp_path) -> None:
         name="persist",
         schedule=CronSchedule(kind="every", every_ms=60_000),
         message="hello",
+        **_bound_chat(),
     )
     await service.run_job(job.id)
 
@@ -395,6 +470,7 @@ async def test_run_job_disabled_does_not_flip_running_state(tmp_path) -> None:
         name="disabled",
         schedule=CronSchedule(kind="every", every_ms=60_000),
         message="hello",
+        **_bound_chat(),
     )
     service.enable_job(job.id, enabled=False)
 
@@ -413,6 +489,7 @@ async def test_run_job_preserves_running_service_state(tmp_path) -> None:
         name="manual",
         schedule=CronSchedule(kind="every", every_ms=60_000),
         message="hello",
+        **_bound_chat(),
     )
 
     result = await service.run_job(job.id, force=True)
@@ -435,6 +512,7 @@ async def test_running_service_honors_external_disable(tmp_path) -> None:
         name="external-disable",
         schedule=CronSchedule(kind="every", every_ms=200),
         message="hello",
+        **_bound_chat(),
     )
     await service.start()
     try:
@@ -483,6 +561,7 @@ async def test_start_server_not_jobs(tmp_path):
         name="hist",
         schedule=CronSchedule(kind="every", every_ms=100),
         message="hello",
+        **_bound_chat(),
     )
     assert len(service.list_jobs()) == 1
     await _wait_until(lambda: bool(called), timeout=0.8)
@@ -503,6 +582,7 @@ async def test_subsecond_job_not_delayed_to_one_second(tmp_path):
         name="fast",
         schedule=CronSchedule(kind="every", every_ms=100),
         message="hello",
+        **_bound_chat(),
     )
     await service.start()
     try:
@@ -526,6 +606,7 @@ async def test_running_service_picks_up_external_add(tmp_path):
         name="heartbeat",
         schedule=CronSchedule(kind="every", every_ms=100),
         message="tick",
+        **_bound_chat("heartbeat"),
     )
     await service.start()
     try:
@@ -536,6 +617,7 @@ async def test_running_service_picks_up_external_add(tmp_path):
             name="external",
             schedule=CronSchedule(kind="every", every_ms=100),
             message="ping",
+            **_bound_chat("external"),
         )
 
         await _wait_until(lambda: "external" in called, timeout=0.8)
@@ -557,6 +639,7 @@ async def test_add_job_during_jobs_exec(tmp_path):
                 name="test",
                 schedule=CronSchedule(kind="every", every_ms=150),
                 message="tick",
+                **_bound_chat("test"),
             )
             run_once = False
 
@@ -565,6 +648,7 @@ async def test_add_job_during_jobs_exec(tmp_path):
         name="heartbeat",
         schedule=CronSchedule(kind="every", every_ms=100),
         message="tick",
+        **_bound_chat("heartbeat"),
     )
     assert len(service.list_jobs()) == 1
     await service.start()
@@ -585,6 +669,7 @@ async def test_external_update_preserves_run_history_records(tmp_path):
         name="history",
         schedule=CronSchedule(kind="every", every_ms=60_000),
         message="hello",
+        **_bound_chat(),
     )
     await service.run_job(job.id, force=True)
 
@@ -626,6 +711,7 @@ async def test_timer_execution_is_not_rolled_back_by_list_jobs_reload(tmp_path):
         name="race",
         schedule=CronSchedule(kind="every", every_ms=60_000),
         message="hello",
+        **_bound_chat(),
     )
     job.state.next_run_at_ms = max(1, int(time.time() * 1000) - 1_000)
     service._save_store()
@@ -650,6 +736,7 @@ def test_update_job_changes_name(tmp_path) -> None:
         name="old name",
         schedule=CronSchedule(kind="every", every_ms=60_000),
         message="hello",
+        **_bound_chat(),
     )
     result = service.update_job(job.id, name="new name")
     assert isinstance(result, CronJob)
@@ -663,6 +750,7 @@ def test_update_job_changes_schedule(tmp_path) -> None:
         name="sched",
         schedule=CronSchedule(kind="every", every_ms=60_000),
         message="hello",
+        **_bound_chat(),
     )
     old_next = job.state.next_run_at_ms
 
@@ -679,6 +767,7 @@ def test_update_job_changes_message(tmp_path) -> None:
         name="msg",
         schedule=CronSchedule(kind="every", every_ms=60_000),
         message="old message",
+        **_bound_chat(),
     )
     result = service.update_job(job.id, message="new message")
     assert isinstance(result, CronJob)
@@ -691,6 +780,7 @@ def test_update_job_changes_cron_expression(tmp_path) -> None:
         name="cron-job",
         schedule=CronSchedule(kind="cron", expr="0 9 * * *", tz="UTC"),
         message="hello",
+        **_bound_chat(),
     )
     result = service.update_job(
         job.id,
@@ -726,6 +816,7 @@ def test_update_job_validates_schedule(tmp_path) -> None:
         name="validate",
         schedule=CronSchedule(kind="every", every_ms=60_000),
         message="hello",
+        **_bound_chat(),
     )
     with pytest.raises(ValueError, match="unknown timezone"):
         service.update_job(
@@ -743,6 +834,7 @@ async def test_update_job_preserves_run_history(tmp_path) -> None:
         name="hist",
         schedule=CronSchedule(kind="every", every_ms=60_000),
         message="hello",
+        **_bound_chat(),
     )
     await service.run_job(job.id)
 
@@ -758,6 +850,7 @@ def test_update_job_offline_writes_action(tmp_path) -> None:
         name="offline",
         schedule=CronSchedule(kind="every", every_ms=60_000),
         message="hello",
+        **_bound_chat(),
     )
     service.update_job(job.id, name="updated-offline")
 
@@ -811,6 +904,7 @@ async def test_list_jobs_during_on_job_does_not_cause_stale_reload(tmp_path) ->
             name=name,
             schedule=CronSchedule(kind="every", every_ms=3_600_000),
             message="test",
+            **_bound_chat(name),
         )
     # Force next_run to the past so _on_timer picks them up
     for job in service._store.jobs:

tests/cron/test_cron_tool_list.py

@@ -20,6 +20,14 @@ def _make_tool_with_tz(tmp_path, tz: str) -> CronTool:
     return CronTool(service, default_timezone=tz)
 
 
+def _bound_chat(chat_id: str = "chat-1") -> dict[str, str]:
+    return {
+        "session_key": f"websocket:{chat_id}",
+        "origin_channel": "websocket",
+        "origin_chat_id": chat_id,
+    }
+
+
 # -- _format_timing tests --
 
 
@@ -146,6 +154,7 @@ def test_list_cron_job_shows_expression_and_timezone(tmp_path) -> None:
         name="Morning scan",
         schedule=CronSchedule(kind="cron", expr="0 9 * * 1-5", tz="America/Denver"),
         message="scan",
+        **_bound_chat(),
     )
     result = tool._list_jobs()
     assert "cron: 0 9 * * 1-5 (America/Denver)" in result
@@ -157,6 +166,7 @@ def test_list_every_job_shows_human_interval(tmp_path) -> None:
         name="Frequent check",
         schedule=CronSchedule(kind="every", every_ms=1_800_000),
         message="check",
+        **_bound_chat(),
     )
     result = tool._list_jobs()
     assert "every 30m" in result
@@ -168,6 +178,7 @@ def test_list_every_job_hours(tmp_path) -> None:
         name="Hourly check",
         schedule=CronSchedule(kind="every", every_ms=7_200_000),
         message="check",
+        **_bound_chat(),
     )
     result = tool._list_jobs()
     assert "every 2h" in result
@@ -179,6 +190,7 @@ def test_list_every_job_seconds(tmp_path) -> None:
         name="Fast check",
         schedule=CronSchedule(kind="every", every_ms=30_000),
         message="check",
+        **_bound_chat(),
     )
     result = tool._list_jobs()
     assert "every 30s" in result
@@ -190,6 +202,7 @@ def test_list_every_job_non_minute_seconds(tmp_path) -> None:
         name="Ninety-second check",
         schedule=CronSchedule(kind="every", every_ms=90_000),
         message="check",
+        **_bound_chat(),
     )
     result = tool._list_jobs()
     assert "every 90s" in result
@@ -201,6 +214,7 @@ def test_list_every_job_milliseconds(tmp_path) -> None:
         name="Sub-second check",
         schedule=CronSchedule(kind="every", every_ms=200),
         message="check",
+        **_bound_chat(),
     )
     result = tool._list_jobs()
     assert "every 200ms" in result
@@ -212,6 +226,7 @@ def test_list_at_job_shows_iso_timestamp(tmp_path) -> None:
         name="One-shot",
         schedule=CronSchedule(kind="at", at_ms=1773684000000),
         message="fire",
+        **_bound_chat(),
     )
     result = tool._list_jobs()
     assert "at 2026-" in result
@@ -226,6 +241,7 @@ async def test_list_shows_last_run_state(tmp_path) -> None:
         name="Stateful job",
         schedule=CronSchedule(kind="cron", expr="0 9 * * *", tz="UTC"),
         message="test",
+        **_bound_chat(),
     )
     # Simulate a completed run by updating state in the store
     job.state.last_run_at_ms = 1773673200000
@@ -245,6 +261,7 @@ async def test_list_shows_error_message(tmp_path) -> None:
         name="Failed job",
         schedule=CronSchedule(kind="cron", expr="0 9 * * *", tz="UTC"),
         message="test",
+        **_bound_chat(),
     )
     job.state.last_run_at_ms = 1773673200000
     job.state.last_status = "error"
@@ -262,6 +279,7 @@ def test_list_shows_next_run(tmp_path) -> None:
         name="Upcoming job",
         schedule=CronSchedule(kind="cron", expr="0 9 * * *", tz="UTC"),
         message="test",
+        **_bound_chat(),
     )
     result = tool._list_jobs()
     assert "Next run:" in result

tests/webui/test_session_list_index.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import os
+from datetime import datetime
 from pathlib import Path
 
 import nanobot.webui.session_list_index as session_list_index
@@ -86,5 +88,86 @@ 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_uses_webui_transcript_activity_for_sort(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    webui_dir = tmp_path / "webui"
+    webui_dir.mkdir()
+    monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir)
+
+    manager = SessionManager(tmp_path)
+    old_session = manager.get_or_create("websocket:old-metadata")
+    old_session.created_at = datetime(2026, 6, 15, 10, 0, 0)
+    old_session.updated_at = datetime(2026, 6, 15, 10, 0, 0)
+    old_session.add_message("user", "old metadata")
+    old_session.updated_at = datetime(2026, 6, 15, 10, 0, 0)
+    manager.save(old_session)
+
+    newer_metadata = manager.get_or_create("websocket:newer-metadata")
+    newer_metadata.created_at = datetime(2026, 6, 15, 11, 0, 0)
+    newer_metadata.updated_at = datetime(2026, 6, 15, 11, 0, 0)
+    newer_metadata.add_message("user", "newer metadata")
+    newer_metadata.updated_at = datetime(2026, 6, 15, 11, 0, 0)
+    manager.save(newer_metadata)
+
+    transcript = webui_dir / "websocket_old-metadata.jsonl"
+    transcript.write_text(
+        '{"event":"turn_end","chat_id":"old-metadata"}\n',
+        encoding="utf-8",
+    )
+    activity_ns = int(datetime(2026, 6, 15, 12, 0, 0).timestamp() * 1_000_000_000)
+    os.utime(transcript, ns=(activity_ns, activity_ns))
+
+    rows = list_webui_sessions(manager)
+
+    assert [row["key"] for row in rows] == [
+        "websocket:old-metadata",
+        "websocket:newer-metadata",
+    ]
+    assert rows[0]["updated_at"].startswith("2026-06-15T12:00:00")
+
+
+def test_webui_session_list_rescans_when_transcript_changes(
+    tmp_path: Path,
+    monkeypatch,
+) -> None:
+    webui_dir = tmp_path / "webui"
+    webui_dir.mkdir()
+    monkeypatch.setattr(session_list_index, "get_webui_dir", lambda: webui_dir)
+
+    manager = SessionManager(tmp_path)
+    session = manager.get_or_create("websocket:transcript-change")
+    session.created_at = datetime(2026, 6, 15, 10, 0, 0)
+    session.updated_at = datetime(2026, 6, 15, 10, 0, 0)
+    session.add_message("user", "preview")
+    session.updated_at = datetime(2026, 6, 15, 10, 0, 0)
+    manager.save(session)
+
+    assert list_webui_sessions(manager)[0]["preview"] == "preview"
+
+    transcript = webui_dir / "websocket_transcript-change.jsonl"
+    transcript.write_text(
+        '{"event":"turn_end","chat_id":"transcript-change"}\n',
+        encoding="utf-8",
+    )
+    activity_ns = int(datetime(2026, 6, 15, 12, 30, 0).timestamp() * 1_000_000_000)
+    os.utime(transcript, ns=(activity_ns, activity_ns))
+
+    original_scan = session_list_index._scan_session_row
+    scanned: list[str] = []
+
+    def record_scan(session_manager: SessionManager, path: Path) -> dict | None:
+        scanned.append(path.name)
+        return original_scan(session_manager, path)
+
+    monkeypatch.setattr(session_list_index, "_scan_session_row", record_scan)
+
+    rows = list_webui_sessions(manager)
+
+    assert scanned == [manager._get_session_path("websocket:transcript-change").name]
+    assert rows[0]["updated_at"].startswith("2026-06-15T12:30:00")
+
+
 def list_webui_sessions(manager: SessionManager) -> list[dict]:
     return session_list_index.list_webui_sessions(manager)