fix(cron): sanitize persisted origin metadata

This commit is contained in:
Oxygen56
2026-08-30 17:24:05 +08:00
committed by Xubin Ren
parent 919e3d341e
commit 679a07460e
3 changed files with 91 additions and 2 deletions
+20 -2
View File
@@ -25,6 +25,7 @@ from nanobot.cron.types import (
CronSchedule, CronSchedule,
CronStore, CronStore,
) )
from nanobot.runtime_context import RUNTIME_CONTEXT_INPUT_META
from nanobot.utils.run_records import ( from nanobot.utils.run_records import (
write_run_record as write_automation_run_record, write_run_record as write_automation_run_record,
) )
@@ -115,8 +116,21 @@ def _disable_malformed_legacy_job(job: CronJob) -> None:
logger.warning("Cron: disabled malformed legacy job '{}' ({}): {}", job.name, job.id, reason) logger.warning("Cron: disabled malformed legacy job '{}' ({}): {}", job.name, job.id, reason)
def _persistable_origin_metadata(metadata: dict[str, Any]) -> dict[str, Any]:
"""Return a detached JSON-safe routing snapshot for a cron payload."""
snapshot: dict[str, Any] = {}
for key, value in metadata.items():
if key == RUNTIME_CONTEXT_INPUT_META:
continue
try:
snapshot[key] = json.loads(json.dumps(value, ensure_ascii=False, allow_nan=False))
except (TypeError, ValueError, RecursionError):
continue
return snapshot
def _normalize_agent_turn_job(job: CronJob) -> bool: def _normalize_agent_turn_job(job: CronJob) -> bool:
"""Migrate legacy user cron payloads into session-bound payloads. """Make routing metadata persistable and migrate legacy user cron payloads.
Pre-bound user cron jobs stored their delivery target in ``channel``/``to``. Pre-bound user cron jobs stored their delivery target in ``channel``/``to``.
Normal user-created legacy jobs always have those fields; if they are Normal user-created legacy jobs always have those fields; if they are
@@ -124,8 +138,12 @@ def _normalize_agent_turn_job(job: CronJob) -> bool:
a runtime legacy execution path. a runtime legacy execution path.
""" """
payload = job.payload payload = job.payload
origin_metadata = _persistable_origin_metadata(payload.origin_metadata)
changed = origin_metadata != payload.origin_metadata
payload.origin_metadata = origin_metadata
if payload.kind != "agent_turn" or not _has_legacy_delivery_context(payload): if payload.kind != "agent_turn" or not _has_legacy_delivery_context(payload):
return False return changed
if not payload.channel or not payload.to: if not payload.channel or not payload.to:
_disable_malformed_legacy_job(job) _disable_malformed_legacy_job(job)
+34
View File
@@ -7,6 +7,7 @@ import pytest
from nanobot.cron.service import CronJobSkippedError, CronService from nanobot.cron.service import CronJobSkippedError, CronService
from nanobot.cron.types import CronJob, CronPayload, CronSchedule from nanobot.cron.types import CronJob, CronPayload, CronSchedule
from nanobot.runtime_context import RUNTIME_CONTEXT_INPUT_META
async def _wait_until(predicate, *, timeout: float = 1.0, interval: float = 0.01) -> None: async def _wait_until(predicate, *, timeout: float = 1.0, interval: float = 0.01) -> None:
@@ -411,6 +412,39 @@ def test_add_job_preserves_origin_delivery_context(tmp_path) -> None:
assert reloaded.payload.origin_metadata == metadata assert reloaded.payload.origin_metadata == metadata
@pytest.mark.asyncio
async def test_start_heals_runtime_context_from_pending_external_add(tmp_path) -> None:
"""Flattened runtime blocks from older action files must not be replayed."""
store_path = tmp_path / "cron" / "jobs.json"
external = CronService(store_path)
job = external.add_job(
name="quoted reminder",
schedule=CronSchedule(kind="every", every_ms=60_000),
message="remember this",
origin_metadata={"webui": True},
**_bound_chat("quoted"),
)
action_path = tmp_path / "cron" / "action.jsonl"
action = json.loads(action_path.read_text(encoding="utf-8"))
action["params"]["payload"]["origin_metadata"][RUNTIME_CONTEXT_INPUT_META] = [
{"source": "webui_quote", "content": "quoted reply"}
]
action_path.write_text(json.dumps(action), encoding="utf-8")
owner = CronService(store_path)
await owner.start()
try:
loaded = owner.get_job(job.id)
assert loaded is not None
assert loaded.payload.origin_metadata == {"webui": True}
raw = json.loads(store_path.read_text(encoding="utf-8"))
assert raw["jobs"][0]["payload"]["originMetadata"] == {"webui": True}
finally:
owner.stop()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_channel_meta_and_session_key_survive_store_reload(tmp_path) -> None: async def test_channel_meta_and_session_key_survive_store_reload(tmp_path) -> None:
store_path = tmp_path / "cron" / "jobs.json" store_path = tmp_path / "cron" / "jobs.json"
+37
View File
@@ -1,6 +1,7 @@
from __future__ import annotations from __future__ import annotations
import asyncio import asyncio
import json
from unittest.mock import MagicMock from unittest.mock import MagicMock
import pytest import pytest
@@ -11,6 +12,7 @@ from nanobot.agent.tools.message import MessageTool
from nanobot.agent.tools.spawn import SpawnTool from nanobot.agent.tools.spawn import SpawnTool
from nanobot.cron.service import CronService from nanobot.cron.service import CronService
from nanobot.providers.base import GenerationSettings, LLMProvider from nanobot.providers.base import GenerationSettings, LLMProvider
from nanobot.runtime_context import RUNTIME_CONTEXT_INPUT_META, RuntimeContextBlock
from nanobot.session.keys import UNIFIED_SESSION_KEY from nanobot.session.keys import UNIFIED_SESSION_KEY
from nanobot.utils.llm_runtime import LLMRuntime from nanobot.utils.llm_runtime import LLMRuntime
@@ -299,6 +301,41 @@ async def test_webui_cron_tool_uses_origin_session_when_unified_enabled(tmp_path
assert jobs[0].payload.origin_metadata == {"webui": True} assert jobs[0].payload.origin_metadata == {"webui": True}
@pytest.mark.asyncio
async def test_cron_tool_snapshots_only_persistable_request_metadata(tmp_path) -> None:
"""Live runtime context must not poison a persisted WebUI cron job."""
store_path = tmp_path / "jobs.json"
service = CronService(store_path)
tool = CronTool(service)
await service.start()
try:
with request_context(
RequestContext(
channel="websocket",
chat_id="chat-123",
metadata={
"webui": True,
RUNTIME_CONTEXT_INPUT_META: [
RuntimeContextBlock(source="webui_quote", content="quoted reply")
],
"opaque": object(),
},
session_key=UNIFIED_SESSION_KEY,
)
):
result = await tool.execute(action="add", message="standup", every_seconds=300)
assert result.startswith("Created job")
jobs = service.list_jobs()
assert len(jobs) == 1
assert jobs[0].payload.origin_metadata == {"webui": True}
raw = json.loads(store_path.read_text(encoding="utf-8"))
assert raw["jobs"][0]["payload"]["originMetadata"] == {"webui": True}
finally:
service.stop()
@pytest.mark.asyncio @pytest.mark.asyncio
async def test_cron_tool_preserves_thread_scoped_session_key(tmp_path) -> None: async def test_cron_tool_preserves_thread_scoped_session_key(tmp_path) -> None:
"""Channel-provided thread session keys should remain the cron owner.""" """Channel-provided thread session keys should remain the cron owner."""