diff --git a/nanobot/triggers/local_store.py b/nanobot/triggers/local_store.py index 05250d53..977143fa 100644 --- a/nanobot/triggers/local_store.py +++ b/nanobot/triggers/local_store.py @@ -16,11 +16,13 @@ from filelock import FileLock from loguru import logger from nanobot.triggers.local_types import LocalTrigger, TriggerDelivery, TriggerRunRecord +from nanobot.utils.helpers import truncate_text from nanobot.utils.run_records import write_run_record as write_automation_run_record _TRIGGER_ID_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" _MAX_RUN_HISTORY = 20 _MAX_DELIVERY_ATTEMPTS = 10 +_RUN_RECORD_TEXT_MAX_CHARS = 4000 _PROCESSING_RECOVERY_ERROR = "delivery was recovered from interrupted processing" @@ -292,9 +294,9 @@ class LocalTriggerStore: record = _delivery_run_record(delivery, trigger) record["status"] = status if error: - record["error"] = error + record["error"] = _run_record_text(error) if response is not None: - record["response"] = response + record["response"] = _run_record_text(response) return self.write_run_record(delivery.id, record) def _ensure_dirs(self) -> None: @@ -448,12 +450,12 @@ def _delivery_run_record( "kind": "local_trigger", "trigger_id": delivery.trigger_id, "delivery_id": delivery.id, - "content": delivery.content, + "content": _run_record_text(delivery.content), "created_at_ms": delivery.created_at_ms, "attempts": delivery.attempts, } if delivery.last_error: - record["last_error"] = delivery.last_error + record["last_error"] = _run_record_text(delivery.last_error) if trigger is not None: record.update( { @@ -466,3 +468,7 @@ def _delivery_run_record( } ) return record + + +def _run_record_text(value: str) -> str: + return truncate_text(value, _RUN_RECORD_TEXT_MAX_CHARS) diff --git a/nanobot/utils/run_records.py b/nanobot/utils/run_records.py index 6f463288..e952c02d 100644 --- a/nanobot/utils/run_records.py +++ b/nanobot/utils/run_records.py @@ -2,6 +2,7 @@ from __future__ import annotations +import errno import json import os import time @@ -45,7 +46,11 @@ def _atomic_write(path: Path, content: str) -> None: with suppress(PermissionError): fd = os.open(str(path.parent), os.O_RDONLY) try: - os.fsync(fd) + try: + os.fsync(fd) + except OSError as exc: + if exc.errno != errno.EINVAL: + raise finally: os.close(fd) except BaseException: diff --git a/tests/triggers/test_local_triggers.py b/tests/triggers/test_local_triggers.py index 8c818cf3..ca5f4d1b 100644 --- a/tests/triggers/test_local_triggers.py +++ b/tests/triggers/test_local_triggers.py @@ -62,19 +62,20 @@ def test_trigger_store_allows_multiple_triggers_per_session(tmp_path: Path) -> N assert first.id != second.id -def test_trigger_store_atomic_write_ignores_unsupported_directory_fsync( +def test_trigger_store_atomic_writes_ignore_unsupported_directory_fsync( tmp_path: Path, monkeypatch: pytest.MonkeyPatch, ) -> None: """Shared folders may allow opening directories but reject directory fsync.""" store = LocalTriggerStore(tmp_path) real_open = os.open + real_close = os.close real_fsync = os.fsync directory_fds: set[int] = set() def fake_open(path: str, flags: int, *args: object, **kwargs: object) -> int: fd = real_open(path, flags, *args, **kwargs) - if Path(path).name == "triggers": + if Path(path).name in {"triggers", "runs"}: directory_fds.add(fd) return fd @@ -83,7 +84,12 @@ def test_trigger_store_atomic_write_ignores_unsupported_directory_fsync( raise OSError(errno.EINVAL, "Invalid argument") real_fsync(fd) + def fake_close(fd: int) -> None: + directory_fds.discard(fd) + real_close(fd) + monkeypatch.setattr(os, "open", fake_open) + monkeypatch.setattr(os, "close", fake_close) monkeypatch.setattr(os, "fsync", fake_fsync) trigger = store.create( @@ -94,6 +100,9 @@ def test_trigger_store_atomic_write_ignores_unsupported_directory_fsync( ) assert store.get(trigger.id) is not None + delivery = store.enqueue(trigger.id, "queued from shared folder") + record = _read_run_record(store, delivery.id) + assert record["content"] == "queued from shared folder" def test_enqueue_rejects_disabled_trigger(tmp_path: Path) -> None: @@ -138,6 +147,37 @@ def test_enqueue_writes_trigger_run_record(tmp_path: Path) -> None: assert record["updated_at_ms"] > 0 +def test_delivery_run_record_truncates_large_content_and_response(tmp_path: Path) -> None: + store = LocalTriggerStore(tmp_path) + trigger = store.create( + name="Large audit", + channel="websocket", + chat_id="chat-1", + session_key="websocket:chat-1", + ) + large_content = "content-" * 1000 + large_response = "response-" * 1000 + + delivery = store.enqueue(trigger.id, large_content) + queued_record = _read_run_record(store, delivery.id) + assert queued_record["content"].startswith("content-") + assert queued_record["content"].endswith("\n... (truncated)") + assert len(queued_record["content"]) < len(large_content) + + store.write_delivery_run_record( + delivery, + trigger=trigger, + status="ok", + response=large_response, + ) + + final_record = _read_run_record(store, delivery.id) + assert final_record["content"].endswith("\n... (truncated)") + assert final_record["response"].startswith("response-") + assert final_record["response"].endswith("\n... (truncated)") + assert len(final_record["response"]) < len(large_response) + + def test_delete_removes_delivery_files_for_trigger(tmp_path: Path) -> None: store = LocalTriggerStore(tmp_path) trigger = store.create(