fix: harden cron session automation flows

This commit is contained in:
chengyongru 2026-06-12 18:13:25 +08:00
parent b5f9d51b5b
commit a50b3ac0f2
30 changed files with 267 additions and 56 deletions

View File

@ -7,7 +7,11 @@ import dataclasses
from collections.abc import Awaitable, Callable, Iterable from collections.abc import Awaitable, Callable, Iterable
from nanobot.bus.events import InboundMessage, OutboundMessage from nanobot.bus.events import InboundMessage, OutboundMessage
from nanobot.cron.session_turns import cron_run_id, defer_cron_until_session_idle from nanobot.cron.session_turns import (
cron_run_id,
cron_trigger,
defer_cron_until_session_idle,
)
class CronTurnCoordinator: class CronTurnCoordinator:
@ -25,6 +29,7 @@ class CronTurnCoordinator:
self._is_running = is_running self._is_running = is_running
self.deferred_queues: dict[str, list[InboundMessage]] = {} self.deferred_queues: dict[str, list[InboundMessage]] = {}
self._waiters: dict[str, asyncio.Future[OutboundMessage | None]] = {} self._waiters: dict[str, asyncio.Future[OutboundMessage | None]] = {}
self._pending_messages_by_run_id: dict[str, InboundMessage] = {}
async def submit(self, msg: InboundMessage) -> OutboundMessage | None: async def submit(self, msg: InboundMessage) -> OutboundMessage | None:
"""Submit a scheduled cron turn and wait for its session response.""" """Submit a scheduled cron turn and wait for its session response."""
@ -37,6 +42,7 @@ class CronTurnCoordinator:
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
future: asyncio.Future[OutboundMessage | None] = loop.create_future() future: asyncio.Future[OutboundMessage | None] = loop.create_future()
self._waiters[run_id] = future self._waiters[run_id] = future
self._pending_messages_by_run_id[run_id] = msg
try: try:
if self._is_running(): if self._is_running():
await self._publish_inbound(msg) await self._publish_inbound(msg)
@ -45,6 +51,7 @@ class CronTurnCoordinator:
return await future return await future
finally: finally:
self._waiters.pop(run_id, None) self._waiters.pop(run_id, None)
self._pending_messages_by_run_id.pop(run_id, None)
def should_defer( def should_defer(
self, self,
@ -102,6 +109,21 @@ class CronTurnCoordinator:
def defer(self, session_key: str, msg: InboundMessage) -> None: def defer(self, session_key: str, msg: InboundMessage) -> None:
self.deferred_queues.setdefault(session_key, []).append(msg) self.deferred_queues.setdefault(session_key, []).append(msg)
def pending_job_ids_for_session(self, session_key: str) -> set[str]:
"""Return cron jobs that are waiting for or running in *session_key*."""
job_ids: set[str] = set()
for msg in self.deferred_queues.get(session_key, []):
job_id = _cron_job_id(msg)
if job_id:
job_ids.add(job_id)
for msg in self._pending_messages_by_run_id.values():
if msg.session_key != session_key:
continue
job_id = _cron_job_id(msg)
if job_id:
job_ids.add(job_id)
return job_ids
async def publish_next_deferred(self, session_key: str) -> None: async def publish_next_deferred(self, session_key: str) -> None:
queue = self.deferred_queues.get(session_key) queue = self.deferred_queues.get(session_key)
if not queue: if not queue:
@ -110,3 +132,11 @@ class CronTurnCoordinator:
if not queue: if not queue:
self.deferred_queues.pop(session_key, None) self.deferred_queues.pop(session_key, None)
await self._publish_inbound(msg) await self._publish_inbound(msg)
def _cron_job_id(msg: InboundMessage) -> str | None:
trigger = cron_trigger(msg.metadata)
if not trigger:
return None
value = trigger.get("job_id")
return value if isinstance(value, str) and value else None

View File

@ -574,6 +574,9 @@ class AgentLoop:
async def submit_cron_turn(self, msg: InboundMessage) -> OutboundMessage | None: async def submit_cron_turn(self, msg: InboundMessage) -> OutboundMessage | None:
return await self._cron_turns.submit(msg) return await self._cron_turns.submit(msg)
def pending_cron_job_ids_for_session(self, session_key: str) -> set[str]:
return self._cron_turns.pending_job_ids_for_session(session_key)
def _persist_user_message_early( def _persist_user_message_early(
self, self,
msg: InboundMessage, msg: InboundMessage,

View File

@ -58,6 +58,7 @@ class ChannelManager:
session_manager: "SessionManager | None" = None, session_manager: "SessionManager | None" = None,
cron_service: Any | None = None, cron_service: Any | None = None,
webui_runtime_model_name: Callable[[], str | None] | None = None, webui_runtime_model_name: Callable[[], str | None] | None = None,
webui_cron_pending_job_ids: Callable[[str], set[str]] | None = None,
webui_static_dist: bool = True, webui_static_dist: bool = True,
webui_runtime_surface: str = "browser", webui_runtime_surface: str = "browser",
webui_runtime_capabilities: dict[str, Any] | None = None, webui_runtime_capabilities: dict[str, Any] | None = None,
@ -67,6 +68,7 @@ class ChannelManager:
self._session_manager = session_manager self._session_manager = session_manager
self._cron_service = cron_service self._cron_service = cron_service
self._webui_runtime_model_name = webui_runtime_model_name self._webui_runtime_model_name = webui_runtime_model_name
self._webui_cron_pending_job_ids = webui_cron_pending_job_ids
self._webui_static_dist = webui_static_dist self._webui_static_dist = webui_static_dist
self._webui_runtime_surface = webui_runtime_surface self._webui_runtime_surface = webui_runtime_surface
self._webui_runtime_capabilities = dict(webui_runtime_capabilities or {}) self._webui_runtime_capabilities = dict(webui_runtime_capabilities or {})
@ -126,6 +128,7 @@ class ChannelManager:
runtime_surface=self._webui_runtime_surface, runtime_surface=self._webui_runtime_surface,
runtime_capabilities_overrides=self._webui_runtime_capabilities, runtime_capabilities_overrides=self._webui_runtime_capabilities,
cron_service=self._cron_service, cron_service=self._cron_service,
cron_pending_job_ids=self._webui_cron_pending_job_ids,
logger=logger, logger=logger,
) )
kwargs["gateway"] = gateway kwargs["gateway"] = gateway

View File

@ -947,7 +947,7 @@ def _run_gateway(
from nanobot.bus.runtime_events import RuntimeEventBus from nanobot.bus.runtime_events import RuntimeEventBus
from nanobot.channels.manager import ChannelManager from nanobot.channels.manager import ChannelManager
from nanobot.cron.bound_runner import run_bound_cron_job from nanobot.cron.bound_runner import run_bound_cron_job
from nanobot.cron.service import CronService from nanobot.cron.service import CronJobSkippedError, CronService
from nanobot.cron.session_turns import is_bound_cron_job from nanobot.cron.session_turns import is_bound_cron_job
from nanobot.cron.types import CronJob from nanobot.cron.types import CronJob
from nanobot.providers.factory import build_provider_snapshot, load_provider_snapshot from nanobot.providers.factory import build_provider_snapshot, load_provider_snapshot
@ -1167,12 +1167,14 @@ def _run_gateway(
if is_bound_cron_job(job): if is_bound_cron_job(job):
return await run_bound_cron_job(job, agent=agent, cron=cron) return await run_bound_cron_job(job, agent=agent, cron=cron)
reason = "unbound agent cron job must be recreated from a chat session"
logger.warning( logger.warning(
"Cron: skipped unbound agent job '{}' ({}); recreate it from a chat session", "Cron: skipped unbound agent job '{}' ({}): {}",
job.name, job.name,
job.id, job.id,
reason,
) )
return None raise CronJobSkippedError(reason)
cron.on_job = on_cron_job cron.on_job = on_cron_job
@ -1191,6 +1193,7 @@ def _run_gateway(
session_manager=session_manager, session_manager=session_manager,
cron_service=cron, cron_service=cron,
webui_runtime_model_name=_webui_runtime_model_name, webui_runtime_model_name=_webui_runtime_model_name,
webui_cron_pending_job_ids=getattr(agent, "pending_cron_job_ids_for_session", None),
webui_static_dist=webui_static_dist, webui_static_dist=webui_static_dist,
webui_runtime_surface=webui_runtime_surface, webui_runtime_surface=webui_runtime_surface,
webui_runtime_capabilities=webui_runtime_capabilities, webui_runtime_capabilities=webui_runtime_capabilities,

View File

@ -25,6 +25,10 @@ from nanobot.cron.types import (
) )
class CronJobSkippedError(Exception):
"""Raised by cron callbacks when a job was intentionally skipped."""
def _now_ms() -> int: def _now_ms() -> int:
return int(time.time() * 1000) return int(time.time() * 1000)
@ -524,6 +528,10 @@ class CronService:
job.state.last_error = None job.state.last_error = None
logger.info("Cron: job '{}' completed", job.name) logger.info("Cron: job '{}' completed", job.name)
except CronJobSkippedError as e:
job.state.last_status = "skipped"
job.state.last_error = str(e) or None
logger.warning("Cron: job '{}' skipped: {}", job.name, job.state.last_error or "")
except asyncio.CancelledError as e: except asyncio.CancelledError as e:
current = asyncio.current_task() current = asyncio.current_task()
if current is not None and current.cancelling(): if current is not None and current.cancelling():

View File

@ -1,6 +1,6 @@
"""Post-run evaluation for background tasks (heartbeat & cron). """Post-run notification evaluation for heartbeat checks.
After the agent executes a background task, this module makes a lightweight After heartbeat executes an internal check, this module makes a lightweight
LLM call to decide whether the result warrants notifying the user. LLM call to decide whether the result warrants notifying the user.
""" """
@ -46,10 +46,10 @@ async def evaluate_response(
model: str, model: str,
default_notify: bool = True, default_notify: bool = True,
) -> bool: ) -> bool:
"""Decide whether a background-task result should be delivered to the user. """Decide whether a heartbeat result should be delivered to the user.
On any failure, falls back to ``default_notify`` (cron reminders fail open; On any failure, falls back to ``default_notify``. Heartbeat passes
heartbeat passes ``False`` to fail closed). ``False`` to fail closed.
""" """
try: try:
llm_response = await provider.chat_with_retry( llm_response = await provider.chat_with_retry(

View File

@ -4,7 +4,7 @@ from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any, Callable
from loguru import logger as default_logger from loguru import logger as default_logger
@ -26,6 +26,7 @@ class GatewayServices:
workspaces: WebUIWorkspaceController workspaces: WebUIWorkspaceController
session_manager: Any | None session_manager: Any | None
cron_service: Any | None cron_service: Any | None
cron_pending_job_ids: Callable[[str], set[str]] | None
def build_gateway_services( def build_gateway_services(
@ -41,6 +42,7 @@ def build_gateway_services(
runtime_capabilities_overrides: dict[str, Any] | None, runtime_capabilities_overrides: dict[str, Any] | None,
disabled_skills: set[str] | None = None, disabled_skills: set[str] | None = None,
cron_service: Any | None = None, cron_service: Any | None = None,
cron_pending_job_ids: Callable[[str], set[str]] | None = None,
logger: Any = default_logger, logger: Any = default_logger,
) -> GatewayServices: ) -> GatewayServices:
tokens = GatewayTokenStore() tokens = GatewayTokenStore()
@ -68,6 +70,7 @@ def build_gateway_services(
skills_workspace_path=workspace_path, skills_workspace_path=workspace_path,
disabled_skills=disabled_skills, disabled_skills=disabled_skills,
cron_service=cron_service, cron_service=cron_service,
cron_pending_job_ids=cron_pending_job_ids,
log=logger, log=logger,
) )
return GatewayServices( return GatewayServices(
@ -78,4 +81,5 @@ def build_gateway_services(
workspaces=workspaces, workspaces=workspaces,
session_manager=session_manager, session_manager=session_manager,
cron_service=cron_service, cron_service=cron_service,
cron_pending_job_ids=cron_pending_job_ids,
) )

View File

@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
from collections.abc import Collection
from typing import Any, Protocol from typing import Any, Protocol
from nanobot.cron.types import CronJob from nanobot.cron.types import CronJob
@ -32,18 +33,27 @@ def session_automation_jobs(
def session_automations_payload( def session_automations_payload(
cron_service: _CronServiceLike | None, cron_service: _CronServiceLike | None,
session_key: str, session_key: str,
*,
pending_job_ids: Collection[str] | None = None,
) -> dict[str, Any]: ) -> dict[str, Any]:
"""Return user-created automation jobs attached to a WebUI session.""" """Return user-created automation jobs attached to a WebUI session."""
return { return {
"jobs": serialize_automation_jobs(session_automation_jobs(cron_service, session_key)) "jobs": serialize_automation_jobs(
session_automation_jobs(cron_service, session_key),
pending_job_ids=pending_job_ids,
)
} }
def serialize_automation_jobs(jobs: list[CronJob]) -> list[dict[str, Any]]: def serialize_automation_jobs(
return [_serialize_job(job) for job in jobs] jobs: list[CronJob],
*,
pending_job_ids: Collection[str] | None = None,
) -> list[dict[str, Any]]:
return [_serialize_job(job, pending=job.id in (pending_job_ids or ())) for job in jobs]
def _serialize_job(job: CronJob) -> dict[str, Any]: def _serialize_job(job: CronJob, *, pending: bool = False) -> dict[str, Any]:
return { return {
"id": job.id, "id": job.id,
"name": job.name, "name": job.name,
@ -61,5 +71,6 @@ def _serialize_job(job: CronJob) -> dict[str, Any]:
"state": { "state": {
"next_run_at_ms": job.state.next_run_at_ms, "next_run_at_ms": job.state.next_run_at_ms,
"last_status": job.state.last_status, "last_status": job.state.last_status,
"pending": pending,
}, },
} }

View File

@ -146,6 +146,7 @@ class GatewayHTTPHandler:
skills_workspace_path: Path, skills_workspace_path: Path,
disabled_skills: set[str] | None = None, disabled_skills: set[str] | None = None,
cron_service: CronService | None = None, cron_service: CronService | None = None,
cron_pending_job_ids: Callable[[str], set[str]] | None = None,
log: Any = logger, log: Any = logger,
) -> None: ) -> None:
self.config = config self.config = config
@ -159,6 +160,7 @@ class GatewayHTTPHandler:
self.skills_workspace_path = skills_workspace_path self.skills_workspace_path = skills_workspace_path
self.disabled_skills = disabled_skills or set() self.disabled_skills = disabled_skills or set()
self.cron_service = cron_service self.cron_service = cron_service
self.cron_pending_job_ids = cron_pending_job_ids
self._log = log self._log = log
self._runtime_surface = runtime_surface self._runtime_surface = runtime_surface
@ -436,8 +438,15 @@ class GatewayHTTPHandler:
return _http_error(400, "invalid session key") return _http_error(400, "invalid session key")
if not _is_websocket_channel_session_key(decoded_key): if not _is_websocket_channel_session_key(decoded_key):
return _http_error(404, "session not found") return _http_error(404, "session not found")
pending_job_ids: set[str] = set()
if self.cron_pending_job_ids is not None:
pending_job_ids = self.cron_pending_job_ids(decoded_key)
return _http_json_response( return _http_json_response(
session_automations_payload(self.cron_service, decoded_key) session_automations_payload(
self.cron_service,
decoded_key,
pending_job_ids=pending_job_ids,
)
) )
def _handle_session_delete(self, request: WsRequest, key: str) -> Response: def _handle_session_delete(self, request: WsRequest, key: str) -> Response:

View File

@ -1,7 +1,7 @@
import pytest import pytest
from nanobot.utils.evaluator import evaluate_response
from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest from nanobot.providers.base import LLMProvider, LLMResponse, ToolCallRequest
from nanobot.utils.evaluator import evaluate_response
class DummyProvider(LLMProvider): class DummyProvider(LLMProvider):

View File

@ -639,7 +639,7 @@ async def test_cron_turn_deferred_while_session_active(tmp_path):
chat_id="chat-1", chat_id="chat-1",
content="scheduled work", content="scheduled work",
metadata={ metadata={
CRON_TRIGGER_META: {"run_id": "run-1"}, CRON_TRIGGER_META: {"job_id": "job-1", "run_id": "run-1"},
CRON_DEFER_UNTIL_IDLE_META: True, CRON_DEFER_UNTIL_IDLE_META: True,
}, },
session_key_override=session_key, session_key_override=session_key,
@ -657,11 +657,49 @@ async def test_cron_turn_deferred_while_session_active(tmp_path):
assert pending.empty() assert pending.empty()
assert loop._dispatch.await_count == 0 assert loop._dispatch.await_count == 0
assert loop._cron_turns.deferred_queues[session_key] == [msg] assert loop._cron_turns.deferred_queues[session_key] == [msg]
assert loop.pending_cron_job_ids_for_session(session_key) == {"job-1"}
await loop._cron_turns.publish_next_deferred(session_key) await loop._cron_turns.publish_next_deferred(session_key)
queued = await asyncio.wait_for(loop.bus.consume_inbound(), timeout=0.5) queued = await asyncio.wait_for(loop.bus.consume_inbound(), timeout=0.5)
assert queued is msg assert queued is msg
assert session_key not in loop._cron_turns.deferred_queues assert session_key not in loop._cron_turns.deferred_queues
assert loop.pending_cron_job_ids_for_session(session_key) == set()
@pytest.mark.asyncio
async def test_submitted_cron_turn_reports_pending_until_completed(tmp_path):
"""Bound cron jobs remain marked pending while their session turn is in flight."""
from nanobot.bus.events import InboundMessage, OutboundMessage
from nanobot.cron.session_turns import CRON_TRIGGER_META
loop = _make_loop(tmp_path)
loop._running = True
session_key = "websocket:chat-1"
msg = InboundMessage(
channel="websocket",
sender_id="cron",
chat_id="chat-1",
content="scheduled work",
metadata={CRON_TRIGGER_META: {"job_id": "job-1", "run_id": "run-1"}},
session_key_override=session_key,
)
submit_task = asyncio.create_task(loop.submit_cron_turn(msg))
queued = await asyncio.wait_for(loop.bus.consume_inbound(), timeout=0.5)
assert queued is msg
assert loop.pending_cron_job_ids_for_session(session_key) == {"job-1"}
response = OutboundMessage(
channel="websocket",
chat_id="chat-1",
content="done",
)
loop._cron_turns.complete(msg, response=response)
assert await asyncio.wait_for(submit_task, timeout=0.5) is response
assert loop.pending_cron_job_ids_for_session(session_key) == set()
@pytest.mark.asyncio @pytest.mark.asyncio

View File

@ -30,6 +30,7 @@ def _make_handler(
workspace_path: Path | None = None, workspace_path: Path | None = None,
runtime_model_name: Any | None = None, runtime_model_name: Any | None = None,
cron_service: CronService | None = None, cron_service: CronService | None = None,
cron_pending_job_ids: Any | None = None,
) -> GatewayServices: ) -> GatewayServices:
config = WebSocketConfig.model_validate(cfg) if isinstance(cfg, dict) else cfg config = WebSocketConfig.model_validate(cfg) if isinstance(cfg, dict) else cfg
workspace = workspace_path or Path.cwd() workspace = workspace_path or Path.cwd()
@ -44,6 +45,7 @@ def _make_handler(
runtime_surface="browser", runtime_surface="browser",
runtime_capabilities_overrides=None, runtime_capabilities_overrides=None,
cron_service=cron_service, cron_service=cron_service,
cron_pending_job_ids=cron_pending_job_ids,
) )
@ -56,6 +58,7 @@ def _ch(
port: int = _PORT, port: int = _PORT,
runtime_model_name: Any | None = None, runtime_model_name: Any | None = None,
cron_service: CronService | None = None, cron_service: CronService | None = None,
cron_pending_job_ids: Any | None = None,
**extra: Any, **extra: Any,
) -> WebSocketChannel: ) -> WebSocketChannel:
cfg: dict[str, Any] = { cfg: dict[str, Any] = {
@ -74,6 +77,7 @@ def _ch(
workspace_path=workspace_path, workspace_path=workspace_path,
runtime_model_name=runtime_model_name, runtime_model_name=runtime_model_name,
cron_service=cron_service, cron_service=cron_service,
cron_pending_job_ids=cron_pending_job_ids,
) )
return WebSocketChannel(cfg, bus, gateway=gateway) return WebSocketChannel(cfg, bus, gateway=gateway)
@ -177,11 +181,12 @@ async def test_session_automations_route_filters_by_webui_session(
) -> None: ) -> None:
cron = CronService(tmp_path / "cron" / "jobs.json") cron = CronService(tmp_path / "cron" / "jobs.json")
hourly = CronSchedule(kind="every", every_ms=3_600_000) hourly = CronSchedule(kind="every", every_ms=3_600_000)
pending_job_id = ""
for name, message, to in ( for name, message, to in (
("Morning check", "Check the project status", "abc"), ("Morning check", "Check the project status", "abc"),
("Other session", "Do not show", "other"), ("Other session", "Do not show", "other"),
): ):
cron.add_job( job = cron.add_job(
name=name, name=name,
schedule=hourly, schedule=hourly,
message=message, message=message,
@ -189,6 +194,8 @@ async def test_session_automations_route_filters_by_webui_session(
origin_channel="websocket", origin_channel="websocket",
origin_chat_id=to, origin_chat_id=to,
) )
if name == "Morning check":
pending_job_id = job.id
cron.add_job( cron.add_job(
name="Legacy same target", name="Legacy same target",
schedule=hourly, schedule=hourly,
@ -210,6 +217,7 @@ async def test_session_automations_route_filters_by_webui_session(
bus, bus,
session_manager=_seed_session(tmp_path, key="websocket:abc"), session_manager=_seed_session(tmp_path, key="websocket:abc"),
cron_service=cron, cron_service=cron,
cron_pending_job_ids=lambda key: {pending_job_id} if key == "websocket:abc" else set(),
port=29914, port=29914,
) )
server_task = asyncio.create_task(channel.start()) server_task = asyncio.create_task(channel.start())
@ -235,6 +243,8 @@ async def test_session_automations_route_filters_by_webui_session(
assert job["schedule"]["kind"] == "every" assert job["schedule"]["kind"] == "every"
assert job["schedule"]["every_ms"] == 3_600_000 assert job["schedule"]["every_ms"] == 3_600_000
assert job["payload"]["message"] == "Check the project status" assert job["payload"]["message"] == "Check the project status"
assert job["state"]["pending"] is True
assert body["jobs"][1]["state"]["pending"] is False
finally: finally:
await channel.stop() await channel.stop()
await server_task await server_task

View File

@ -11,6 +11,7 @@ from typer.testing import CliRunner
from nanobot.bus.events import InboundMessage, OutboundMessage from nanobot.bus.events import InboundMessage, OutboundMessage
from nanobot.cli.commands import app from nanobot.cli.commands import app
from nanobot.config.schema import Config from nanobot.config.schema import Config
from nanobot.cron.service import CronJobSkippedError
from nanobot.cron.session_turns import CRON_DEFER_UNTIL_IDLE_META, CRON_TRIGGER_META from nanobot.cron.session_turns import CRON_DEFER_UNTIL_IDLE_META, CRON_TRIGGER_META
from nanobot.cron.types import CronJob, CronPayload from nanobot.cron.types import CronJob, CronPayload
from nanobot.cron.webui_metadata import cron_proactive_delivery_metadata from nanobot.cron.webui_metadata import cron_proactive_delivery_metadata
@ -1305,9 +1306,9 @@ def test_gateway_unbound_agent_cron_is_skipped(
), ),
) )
response = asyncio.run(cron.on_job(job)) with pytest.raises(CronJobSkippedError, match="unbound agent cron job"):
asyncio.run(cron.on_job(job))
assert response is None
bus.publish_outbound.assert_not_awaited() bus.publish_outbound.assert_not_awaited()

View File

@ -4,7 +4,7 @@ import time
import pytest import pytest
from nanobot.cron.service import CronService from nanobot.cron.service import CronJobSkippedError, CronService
from nanobot.cron.types import CronJob, CronPayload, CronSchedule from nanobot.cron.types import CronJob, CronPayload, CronSchedule
@ -296,6 +296,30 @@ async def test_run_history_records_errors(tmp_path) -> None:
assert loaded.state.run_history[0].error == "boom" assert loaded.state.run_history[0].error == "boom"
@pytest.mark.asyncio
async def test_run_history_records_skipped_jobs(tmp_path) -> None:
store_path = tmp_path / "cron" / "jobs.json"
async def skip(_):
raise CronJobSkippedError("missing session binding")
service = CronService(store_path, on_job=skip)
job = service.add_job(
name="skip",
schedule=CronSchedule(kind="every", every_ms=60_000),
message="hello",
)
await service.run_job(job.id)
loaded = service.get_job(job.id)
assert loaded is not None
assert loaded.state.last_status == "skipped"
assert loaded.state.last_error == "missing session binding"
assert len(loaded.state.run_history) == 1
assert loaded.state.run_history[0].status == "skipped"
assert loaded.state.run_history[0].error == "missing session binding"
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_run_history_records_job_cancellation(tmp_path) -> None: async def test_run_history_records_job_cancellation(tmp_path) -> None:
store_path = tmp_path / "cron" / "jobs.json" store_path = tmp_path / "cron" / "jobs.json"

View File

@ -43,7 +43,7 @@ import type {
} from "@/lib/types"; } from "@/lib/types";
import { Button } from "@/components/ui/button"; import { Button } from "@/components/ui/button";
import { Input } from "@/components/ui/input"; import { Input } from "@/components/ui/input";
import { fetchSessionAutomations, fetchSettings, fetchWorkspaces } from "@/lib/api"; import { fetchSettings, fetchWorkspaces } from "@/lib/api";
import { import {
createRuntimeHost, createRuntimeHost,
getHostApi, getHostApi,
@ -528,7 +528,15 @@ function Shell({
const { t, i18n } = useTranslation(); const { t, i18n } = useTranslation();
const { client, token } = useClient(); const { client, token } = useClient();
const { theme, toggle } = useTheme(); const { theme, toggle } = useTheme();
const { sessions, loading, refresh, createChat, forkChat, deleteChat } = useSessions(); const {
sessions,
loading,
refresh,
createChat,
forkChat,
deleteChat,
getSessionAutomations,
} = useSessions();
const { state: sidebarState, update: updateSidebarState } = const { state: sidebarState, update: updateSidebarState } =
useSidebarState(sessions, !loading); useSidebarState(sessions, !loading);
const initialRouteRef = useRef<ShellRoute | null>(null); const initialRouteRef = useRef<ShellRoute | null>(null);
@ -1306,12 +1314,12 @@ function Shell({
const onRequestDelete = useCallback(async (key: string, label: string) => { const onRequestDelete = useCallback(async (key: string, label: string) => {
let automations: SessionAutomationJob[] = []; let automations: SessionAutomationJob[] = [];
try { try {
automations = (await fetchSessionAutomations(token, key)).jobs; automations = await getSessionAutomations(key);
} catch { } catch {
// Delete remains protected by the backend block; prefetch only improves the first prompt. // Delete remains protected by the backend block; prefetch only improves the first prompt.
} }
setPendingDelete({ key, label, automations }); setPendingDelete({ key, label, automations });
}, [token]); }, [getSessionAutomations]);
const headerTitle = activeSession const headerTitle = activeSession
? sidebarState.title_overrides[activeSession.key] || ? sidebarState.title_overrides[activeSession.key] ||

View File

@ -51,9 +51,7 @@ export function DeleteConfirm({
</AlertDialogTitle> </AlertDialogTitle>
<AlertDialogDescription className="mt-3 max-w-[17rem] text-center text-[14px] leading-6 text-muted-foreground"> <AlertDialogDescription className="mt-3 max-w-[17rem] text-center text-[14px] leading-6 text-muted-foreground">
{hasAutomations {hasAutomations
? t("deleteConfirm.automationsDescription", { ? t("deleteConfirm.automationsDescription")
count: automations.length,
})
: t("deleteConfirm.description")} : t("deleteConfirm.description")}
</AlertDialogDescription> </AlertDialogDescription>
{hasAutomations ? ( {hasAutomations ? (
@ -94,9 +92,7 @@ export function DeleteConfirm({
className="h-11 rounded-full bg-destructive px-5 text-[15px] font-semibold text-destructive-foreground shadow-[0_10px_25px_rgba(239,68,68,0.28)] hover:bg-destructive/90" className="h-11 rounded-full bg-destructive px-5 text-[15px] font-semibold text-destructive-foreground shadow-[0_10px_25px_rgba(239,68,68,0.28)] hover:bg-destructive/90"
> >
{hasAutomations {hasAutomations
? t("deleteConfirm.confirmWithAutomations", { ? t("deleteConfirm.confirmWithAutomations")
count: automations.length,
})
: t("deleteConfirm.confirm")} : t("deleteConfirm.confirm")}
</AlertDialogAction> </AlertDialogAction>
</AlertDialogFooter> </AlertDialogFooter>

View File

@ -176,6 +176,9 @@ function formatNextRun(job: SessionAutomationJob, t: TFunction, now: number) {
if (!job.enabled) { if (!job.enabled) {
return { label: t("thread.sessionInfo.next.disabled"), title: "" }; return { label: t("thread.sessionInfo.next.disabled"), title: "" };
} }
if (job.state.pending) {
return { label: t("thread.sessionInfo.next.pending"), title: "" };
}
const next = job.state.next_run_at_ms; const next = job.state.next_run_at_ms;
if (!next) { if (!next) {
return { label: t("thread.sessionInfo.next.none"), title: "" }; return { label: t("thread.sessionInfo.next.none"), title: "" };

View File

@ -5,6 +5,7 @@ import i18n from "@/i18n";
import { import {
ApiError, ApiError,
deleteSession as apiDeleteSession, deleteSession as apiDeleteSession,
fetchSessionAutomations,
fetchWebuiThread, fetchWebuiThread,
listSessions, listSessions,
} from "@/lib/api"; } from "@/lib/api";
@ -12,6 +13,7 @@ import { hasPendingAgentActivity } from "@/lib/activity-timeline";
import { deriveTitle } from "@/lib/format"; import { deriveTitle } from "@/lib/format";
import type { import type {
ChatSummary, ChatSummary,
SessionAutomationJob,
SessionDeleteResult, SessionDeleteResult,
UIMessage, UIMessage,
WorkspaceScopePayload, WorkspaceScopePayload,
@ -52,6 +54,7 @@ export function useSessions(): {
key: string, key: string,
options?: { deleteAutomations?: boolean }, options?: { deleteAutomations?: boolean },
) => Promise<SessionDeleteResult>; ) => Promise<SessionDeleteResult>;
getSessionAutomations: (key: string) => Promise<SessionAutomationJob[]>;
} { } {
const { client, token } = useClient(); const { client, token } = useClient();
const [sessions, setSessions] = useState<ChatSummary[]>([]); const [sessions, setSessions] = useState<ChatSummary[]>([]);
@ -159,7 +162,21 @@ export function useSessions(): {
[], [],
); );
return { sessions, loading, error, refresh, createChat, forkChat, deleteChat }; const getSessionAutomations = useCallback(async (key: string) => {
const result = await fetchSessionAutomations(tokenRef.current, key);
return result.jobs;
}, []);
return {
sessions,
loading,
error,
refresh,
createChat,
forkChat,
deleteChat,
getSessionAutomations,
};
} }
/** Lazy-load a session's on-disk messages the first time the UI displays it. */ /** Lazy-load a session's on-disk messages the first time the UI displays it. */

View File

@ -663,6 +663,7 @@
}, },
"next": { "next": {
"label": "{{time}}", "label": "{{time}}",
"pending": "Runs shortly",
"disabled": "Paused", "disabled": "Paused",
"none": "No next run" "none": "No next run"
} }

View File

@ -663,6 +663,7 @@
}, },
"next": { "next": {
"label": "Siguiente {{time}}", "label": "Siguiente {{time}}",
"pending": "Se ejecutará pronto",
"disabled": "En pausa", "disabled": "En pausa",
"none": "Sin próxima ejecución" "none": "Sin próxima ejecución"
} }

View File

@ -663,6 +663,7 @@
}, },
"next": { "next": {
"label": "Prochaine {{time}}", "label": "Prochaine {{time}}",
"pending": "Exécution imminente",
"disabled": "En pause", "disabled": "En pause",
"none": "Aucune prochaine exécution" "none": "Aucune prochaine exécution"
} }

View File

@ -663,6 +663,7 @@
}, },
"next": { "next": {
"label": "Berikutnya {{time}}", "label": "Berikutnya {{time}}",
"pending": "Segera berjalan",
"disabled": "Dijeda", "disabled": "Dijeda",
"none": "Tidak ada jadwal berikutnya" "none": "Tidak ada jadwal berikutnya"
} }

View File

@ -663,6 +663,7 @@
}, },
"next": { "next": {
"label": "次回 {{time}}", "label": "次回 {{time}}",
"pending": "まもなく実行",
"disabled": "一時停止", "disabled": "一時停止",
"none": "次回実行なし" "none": "次回実行なし"
} }

View File

@ -663,6 +663,7 @@
}, },
"next": { "next": {
"label": "다음 {{time}}", "label": "다음 {{time}}",
"pending": "곧 실행됨",
"disabled": "일시 중지됨", "disabled": "일시 중지됨",
"none": "다음 실행 없음" "none": "다음 실행 없음"
} }

View File

@ -663,6 +663,7 @@
}, },
"next": { "next": {
"label": "Tiếp theo {{time}}", "label": "Tiếp theo {{time}}",
"pending": "Sắp chạy",
"disabled": "Đã tạm dừng", "disabled": "Đã tạm dừng",
"none": "Không có lần chạy tiếp theo" "none": "Không có lần chạy tiếp theo"
} }

View File

@ -663,6 +663,7 @@
}, },
"next": { "next": {
"label": "下次 {{time}}", "label": "下次 {{time}}",
"pending": "即将执行",
"disabled": "已暂停", "disabled": "已暂停",
"none": "没有下次运行" "none": "没有下次运行"
} }

View File

@ -663,6 +663,7 @@
}, },
"next": { "next": {
"label": "下次 {{time}}", "label": "下次 {{time}}",
"pending": "即將執行",
"disabled": "已暫停", "disabled": "已暫停",
"none": "沒有下次執行" "none": "沒有下次執行"
} }

View File

@ -113,6 +113,7 @@ export interface SessionAutomationJob {
state: { state: {
next_run_at_ms?: number | null; next_run_at_ms?: number | null;
last_status?: "ok" | "error" | "skipped" | string | null; last_status?: "ok" | "error" | "skipped" | string | null;
pending?: boolean;
}; };
} }

View File

@ -1,12 +1,14 @@
import { act, fireEvent, render, screen, waitFor, within } from "@testing-library/react"; import { act, fireEvent, render, screen, waitFor, within } from "@testing-library/react";
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import type { ChatSummary } from "@/lib/types"; import i18n from "@/i18n";
import type { ChatSummary, SessionAutomationJob } from "@/lib/types";
const connectSpy = vi.fn(); const connectSpy = vi.fn();
const refreshSpy = vi.fn(); const refreshSpy = vi.fn();
const createChatSpy = vi.fn().mockResolvedValue("chat-1"); const createChatSpy = vi.fn().mockResolvedValue("chat-1");
const deleteChatSpy = vi.fn(); const deleteChatSpy = vi.fn();
const getSessionAutomationsSpy = vi.fn<(key: string) => Promise<SessionAutomationJob[]>>();
const toggleThemeSpy = vi.fn(); const toggleThemeSpy = vi.fn();
const updateUrlSpy = vi.fn(); const updateUrlSpy = vi.fn();
const attachSpy = vi.fn(); const attachSpy = vi.fn();
@ -146,6 +148,7 @@ vi.mock("@/hooks/useSessions", async (importOriginal) => {
refresh: refreshSpy, refresh: refreshSpy,
createChat: createChatSpy, createChat: createChatSpy,
forkChat: async () => "fork-chat", forkChat: async () => "fork-chat",
getSessionAutomations: getSessionAutomationsSpy,
deleteChat: async (key: string, options?: { deleteAutomations?: boolean }) => { deleteChat: async (key: string, options?: { deleteAutomations?: boolean }) => {
if (options === undefined) await deleteChatSpy(key); if (options === undefined) await deleteChatSpy(key);
else await deleteChatSpy(key, options); else await deleteChatSpy(key, options);
@ -212,13 +215,15 @@ import { deriveWsUrl, fetchBootstrap } from "@/lib/bootstrap";
import App from "@/App"; import App from "@/App";
describe("App layout", () => { describe("App layout", () => {
beforeEach(() => { beforeEach(async () => {
await i18n.changeLanguage("en");
mockSessions = []; mockSessions = [];
connectSpy.mockClear(); connectSpy.mockClear();
updateUrlSpy.mockClear(); updateUrlSpy.mockClear();
refreshSpy.mockReset(); refreshSpy.mockReset();
createChatSpy.mockClear(); createChatSpy.mockClear();
deleteChatSpy.mockReset(); deleteChatSpy.mockReset();
getSessionAutomationsSpy.mockReset().mockResolvedValue([]);
toggleThemeSpy.mockReset(); toggleThemeSpy.mockReset();
attachSpy.mockReset(); attachSpy.mockReset();
runStatusHandlers.clear(); runStatusHandlers.clear();
@ -435,7 +440,7 @@ describe("App layout", () => {
expect(document.body.style.pointerEvents).not.toBe("none"); expect(document.body.style.pointerEvents).not.toBe("none");
}, 15_000); }, 15_000);
it("shows bound automations in the first delete confirmation", async () => { it("shows localized bound automations in the first delete confirmation", async () => {
mockSessions = [ mockSessions = [
{ {
key: "websocket:chat-a", key: "websocket:chat-a",
@ -454,9 +459,7 @@ describe("App layout", () => {
preview: "Second chat", preview: "Second chat",
}, },
]; ];
mockFetchRoutes({ getSessionAutomationsSpy.mockResolvedValue([
"/api/sessions/websocket%3Achat-a/automations": {
jobs: [
{ {
id: "job-1", id: "job-1",
name: "Daily repo check", name: "Daily repo check",
@ -465,33 +468,36 @@ describe("App layout", () => {
payload: { message: "Check the repo" }, payload: { message: "Check the repo" },
state: { next_run_at_ms: Date.UTC(2026, 3, 17, 10, 0, 0) }, state: { next_run_at_ms: Date.UTC(2026, 3, 17, 10, 0, 0) },
}, },
], ]);
}, await i18n.changeLanguage("zh-CN");
});
render(<App />); render(<App />);
await waitFor(() => expect(connectSpy).toHaveBeenCalled()); await waitFor(() => expect(connectSpy).toHaveBeenCalled());
const sidebar = screen.getByRole("navigation", { name: "Sidebar navigation" }); const sidebar = screen.getByRole("navigation", { name: "侧边栏导航" });
await waitFor(() => await waitFor(() =>
expect( expect(
within(sidebar).getByRole("button", { name: /^First chat$/ }), within(sidebar).getByRole("button", { name: /^First chat$/ }),
).toBeInTheDocument(), ).toBeInTheDocument(),
); );
fireEvent.pointerDown(screen.getByLabelText("Chat actions for First chat"), { fireEvent.pointerDown(screen.getByLabelText(/First chat.*会话操作/), {
button: 0, button: 0,
}); });
fireEvent.click(await screen.findByRole("menuitem", { name: "Delete" })); fireEvent.click(await screen.findByRole("menuitem", { name: "删除" }));
await waitFor(() => await waitFor(() =>
expect(screen.getByText("Daily repo check")).toBeInTheDocument(), expect(screen.getByText("Daily repo check")).toBeInTheDocument(),
); );
expect(getSessionAutomationsSpy).toHaveBeenCalledWith("websocket:chat-a");
expect( expect(
screen.getByText("This chat has scheduled automations. Deleting it will also delete them."), screen.getByText("这个对话有关联的自动任务。删除对话也会删除这些自动任务。"),
).toBeInTheDocument(); ).toBeInTheDocument();
expect(
screen.queryByText("This chat has scheduled automations. Deleting it will also delete them."),
).not.toBeInTheDocument();
fireEvent.click(screen.getByRole("button", { name: "Delete chat and automations" })); fireEvent.click(screen.getByRole("button", { name: "删除对话和自动任务" }));
await waitFor(() => await waitFor(() =>
expect(deleteChatSpy).toHaveBeenCalledWith("websocket:chat-a", { expect(deleteChatSpy).toHaveBeenCalledWith("websocket:chat-a", {

View File

@ -5,14 +5,17 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { SessionInfoPopover } from "@/components/thread/SessionInfoPopover"; import { SessionInfoPopover } from "@/components/thread/SessionInfoPopover";
import { setAppLanguage } from "@/i18n"; import { setAppLanguage } from "@/i18n";
function automationJob(nextRunAt = Date.now() + 3_600_000) { function automationJob(
nextRunAt = Date.now() + 3_600_000,
state: Record<string, unknown> = {},
) {
return { return {
id: "job-1", id: "job-1",
name: "Morning check", name: "Morning check",
enabled: true, enabled: true,
schedule: { kind: "every", every_ms: 3_600_000 }, schedule: { kind: "every", every_ms: 3_600_000 },
payload: { message: "Check the project status" }, payload: { message: "Check the project status" },
state: { next_run_at_ms: nextRunAt }, state: { next_run_at_ms: nextRunAt, ...state },
}; };
} }
@ -27,7 +30,8 @@ function automationsResponse(jobs: unknown[]) {
} }
describe("SessionInfoPopover", () => { describe("SessionInfoPopover", () => {
beforeEach(() => { beforeEach(async () => {
await setAppLanguage("en");
vi.stubGlobal( vi.stubGlobal(
"fetch", "fetch",
vi.fn().mockResolvedValue(automationsResponse([automationJob()])), vi.fn().mockResolvedValue(automationsResponse([automationJob()])),
@ -86,6 +90,29 @@ describe("SessionInfoPopover", () => {
expect(screen.queryByText("Automations")).not.toBeInTheDocument(); expect(screen.queryByText("Automations")).not.toBeInTheDocument();
}); });
it("shows a short pending label for deferred automations", async () => {
vi.stubGlobal(
"fetch",
vi.fn().mockResolvedValue(
automationsResponse([automationJob(Date.now() - 1000, { pending: true })]),
),
);
const user = userEvent.setup();
render(
<SessionInfoPopover
sessionKey="websocket:chat-1"
token="tok"
title="Release work"
/>,
);
await user.click(screen.getByRole("button", { name: "Session details" }));
expect(await screen.findByText("Runs shortly")).toBeInTheDocument();
expect(screen.queryByText(/ago/i)).not.toBeInTheDocument();
});
it("refreshes while open so completed one-shot automations disappear", async () => { it("refreshes while open so completed one-shot automations disappear", async () => {
vi.stubGlobal( vi.stubGlobal(
"fetch", "fetch",