fix(persistence): snapshot resume-instruction + persist notes to disk
Two follow-ups from the post-fix audit:
**#1 critical**: ``orchestration/scan.py`` injects the user's new
``--instruction`` into the root's bus inbox via ``bus.send`` on resume,
but ``send`` is one of the deliberately-not-snapshotted high-frequency
mutations. A SIGKILL between that send and the model's first turn
would silently drop the user's new directive. Force a snapshot
immediately after the inject — that's the one specific message we
can't afford to lose, while leaving general ``send`` traffic
unsnapshotted as designed.
**Notes persistence**: ``strix/tools/notes/tools.py`` now mirrors the
todo pattern. ``_notes_storage`` writes through to
``{run_dir}/notes.json`` after every create/update/delete via the
same atomic-tempfile + ``Path.replace`` flow. New
``hydrate_notes_from_disk(run_dir)`` is wired in ``run_strix_scan``
alongside ``hydrate_todos_from_disk`` so a resumed scan recovers the
exact note set the prior process saw, including ``wiki``-category
notes.
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -260,10 +260,12 @@ async def run_strix_scan(
|
|||||||
|
|
||||||
# Wire the per-agent todo store to ``{run_dir}/todos.json`` (mirrored
|
# Wire the per-agent todo store to ``{run_dir}/todos.json`` (mirrored
|
||||||
# on every CRUD) and reload any prior todos so respawned subagents
|
# on every CRUD) and reload any prior todos so respawned subagents
|
||||||
# find their lists intact.
|
# find their lists intact. Same for the shared notes store.
|
||||||
|
from strix.tools.notes.tools import hydrate_notes_from_disk
|
||||||
from strix.tools.todo.tools import hydrate_todos_from_disk
|
from strix.tools.todo.tools import hydrate_todos_from_disk
|
||||||
|
|
||||||
hydrate_todos_from_disk(run_dir)
|
hydrate_todos_from_disk(run_dir)
|
||||||
|
hydrate_notes_from_disk(run_dir)
|
||||||
|
|
||||||
root_id: str | None = None
|
root_id: str | None = None
|
||||||
if is_resume:
|
if is_resume:
|
||||||
@@ -425,6 +427,12 @@ async def run_strix_scan(
|
|||||||
"content": resume_instruction,
|
"content": resume_instruction,
|
||||||
},
|
},
|
||||||
)
|
)
|
||||||
|
# ``bus.send`` is one of the high-frequency mutations that
|
||||||
|
# deliberately skips ``_maybe_snapshot``. The resume-instruction
|
||||||
|
# is the one specific message we can't lose: a SIGKILL between
|
||||||
|
# the send and root's first turn would silently drop the
|
||||||
|
# user's new ``--instruction``. Force a snapshot here.
|
||||||
|
await bus._maybe_snapshot()
|
||||||
logger.info(
|
logger.info(
|
||||||
"Resume: injected new instruction into root inbox (len=%d)",
|
"Resume: injected new instruction into root inbox (len=%d)",
|
||||||
len(resume_instruction),
|
len(resume_instruction),
|
||||||
|
|||||||
@@ -1,8 +1,9 @@
|
|||||||
"""Per-run notes (shared across agents).
|
"""Per-run notes (shared across agents).
|
||||||
|
|
||||||
Pure in-memory state for the lifetime of one scan: the module-level
|
Module-level dict shared across every agent in the same scan process.
|
||||||
``_notes_storage`` dict is visible to every agent in the same process.
|
Mirrored to ``{run_dir}/notes.json`` after every CRUD via :func:`_persist`
|
||||||
No disk I/O — process exit clears the lot. Concurrent writers are
|
so a process restart can :func:`hydrate_notes_from_disk` and the resumed
|
||||||
|
scan picks up exactly where it left off. Concurrent writers are
|
||||||
serialised by ``_notes_lock`` since each tool entry-point dispatches
|
serialised by ``_notes_lock`` since each tool entry-point dispatches
|
||||||
the impl onto a worker thread via ``asyncio.to_thread``.
|
the impl onto a worker thread via ``asyncio.to_thread``.
|
||||||
"""
|
"""
|
||||||
@@ -12,9 +13,11 @@ from __future__ import annotations
|
|||||||
import asyncio
|
import asyncio
|
||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
|
import tempfile
|
||||||
import threading
|
import threading
|
||||||
import uuid
|
import uuid
|
||||||
from datetime import UTC, datetime
|
from datetime import UTC, datetime
|
||||||
|
from pathlib import Path
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
from agents import RunContextWrapper, function_tool
|
from agents import RunContextWrapper, function_tool
|
||||||
@@ -28,6 +31,79 @@ _VALID_NOTE_CATEGORIES = ["general", "findings", "methodology", "questions", "pl
|
|||||||
_notes_lock = threading.RLock()
|
_notes_lock = threading.RLock()
|
||||||
_DEFAULT_CONTENT_PREVIEW_CHARS = 280
|
_DEFAULT_CONTENT_PREVIEW_CHARS = 280
|
||||||
|
|
||||||
|
# On-disk mirror path. Set by :func:`hydrate_notes_from_disk` once per
|
||||||
|
# scan; unset means "no persistence" (e.g. unit tests). All writes go
|
||||||
|
# through :func:`_persist`, which is a no-op until the path is set.
|
||||||
|
_notes_path: Path | None = None
|
||||||
|
|
||||||
|
|
||||||
|
def hydrate_notes_from_disk(run_dir: Path) -> None:
|
||||||
|
"""Wire the on-disk mirror at ``{run_dir}/notes.json`` and reload it.
|
||||||
|
|
||||||
|
Called by :func:`run_strix_scan` once at scan setup. Subsequent CRUD
|
||||||
|
calls auto-persist after every mutation. Idempotent on missing file.
|
||||||
|
Tolerant of corruption — logs and starts empty rather than failing
|
||||||
|
the scan over a broken sidecar artifact.
|
||||||
|
"""
|
||||||
|
global _notes_path # noqa: PLW0603
|
||||||
|
_notes_path = run_dir / "notes.json"
|
||||||
|
with _notes_lock:
|
||||||
|
_notes_storage.clear()
|
||||||
|
if not _notes_path.exists():
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
data = json.loads(_notes_path.read_text(encoding="utf-8"))
|
||||||
|
except (OSError, json.JSONDecodeError):
|
||||||
|
logger.exception(
|
||||||
|
"notes.json at %s is unreadable; starting with empty notes",
|
||||||
|
_notes_path,
|
||||||
|
)
|
||||||
|
return
|
||||||
|
if not isinstance(data, dict):
|
||||||
|
return
|
||||||
|
_notes_storage.update(
|
||||||
|
{
|
||||||
|
nid: note
|
||||||
|
for nid, note in data.items()
|
||||||
|
if isinstance(nid, str) and isinstance(note, dict)
|
||||||
|
}
|
||||||
|
)
|
||||||
|
logger.info(
|
||||||
|
"notes hydrated from %s (%d note(s))",
|
||||||
|
_notes_path,
|
||||||
|
len(_notes_storage),
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def _persist() -> None:
|
||||||
|
"""Atomic-rename mirror of ``_notes_storage`` → ``{run_dir}/notes.json``.
|
||||||
|
|
||||||
|
No-op when ``_notes_path`` isn't wired (tests). Errors are logged
|
||||||
|
and swallowed — a disk hiccup must never tear down the agent's call.
|
||||||
|
"""
|
||||||
|
path = _notes_path
|
||||||
|
if path is None:
|
||||||
|
return
|
||||||
|
try:
|
||||||
|
payload = json.dumps(_notes_storage, ensure_ascii=False, default=str)
|
||||||
|
path.parent.mkdir(parents=True, exist_ok=True)
|
||||||
|
with (
|
||||||
|
_notes_lock,
|
||||||
|
tempfile.NamedTemporaryFile(
|
||||||
|
mode="w",
|
||||||
|
encoding="utf-8",
|
||||||
|
dir=str(path.parent),
|
||||||
|
prefix=f".{path.name}.",
|
||||||
|
suffix=".tmp",
|
||||||
|
delete=False,
|
||||||
|
) as tmp,
|
||||||
|
):
|
||||||
|
tmp.write(payload)
|
||||||
|
tmp_path = Path(tmp.name)
|
||||||
|
tmp_path.replace(path)
|
||||||
|
except Exception:
|
||||||
|
logger.exception("notes persist to %s failed", path)
|
||||||
|
|
||||||
|
|
||||||
def _filter_notes(
|
def _filter_notes(
|
||||||
category: str | None = None,
|
category: str | None = None,
|
||||||
@@ -122,6 +198,7 @@ def _create_note_impl(
|
|||||||
except (ValueError, TypeError) as e:
|
except (ValueError, TypeError) as e:
|
||||||
return {"success": False, "error": f"Failed to create note: {e}", "note_id": None}
|
return {"success": False, "error": f"Failed to create note: {e}", "note_id": None}
|
||||||
else:
|
else:
|
||||||
|
_persist()
|
||||||
return {
|
return {
|
||||||
"success": True,
|
"success": True,
|
||||||
"note_id": note_id,
|
"note_id": note_id,
|
||||||
@@ -194,6 +271,7 @@ def _update_note_impl(
|
|||||||
except (ValueError, TypeError) as e:
|
except (ValueError, TypeError) as e:
|
||||||
return {"success": False, "error": f"Failed to update note: {e}"}
|
return {"success": False, "error": f"Failed to update note: {e}"}
|
||||||
else:
|
else:
|
||||||
|
_persist()
|
||||||
return {
|
return {
|
||||||
"success": True,
|
"success": True,
|
||||||
"message": f"Note '{note['title']}' updated successfully",
|
"message": f"Note '{note['title']}' updated successfully",
|
||||||
@@ -211,6 +289,7 @@ def _delete_note_impl(note_id: str) -> dict[str, Any]:
|
|||||||
except (ValueError, TypeError) as e:
|
except (ValueError, TypeError) as e:
|
||||||
return {"success": False, "error": f"Failed to delete note: {e}"}
|
return {"success": False, "error": f"Failed to delete note: {e}"}
|
||||||
else:
|
else:
|
||||||
|
_persist()
|
||||||
return {
|
return {
|
||||||
"success": True,
|
"success": True,
|
||||||
"message": f"Note '{note_title}' deleted successfully",
|
"message": f"Note '{note_title}' deleted successfully",
|
||||||
|
|||||||
Reference in New Issue
Block a user