mortal/strix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(notes): drop disk persistence + shared-wiki prose

Browse Source 
The notes tool no longer touches disk. ``_notes_storage`` lives in
memory for the lifetime of one scan process, shared across every
agent in that process via the existing RLock. Process exit clears
the lot — no notes.jsonl event log, no wiki/<slug>.md Markdown
rendering, no replay-on-startup hydration.

Removed ~10 internal helpers (``_get_run_dir``,
``_get_notes_jsonl_path``, ``_append_note_event``,
``_load_notes_from_jsonl``, ``_ensure_notes_loaded``,
``_persist_wiki_note``, ``_remove_wiki_note``,
``_get_wiki_directory``, ``_get_wiki_note_path``,
``_sanitize_wiki_title``) plus the ``_loaded_notes_run_dir`` module
state, ``wiki_filename`` per-note field, and the ``OSError`` branches
that only existed for the wiki write path.

The ``wiki`` category is preserved as a free-form long-form bucket;
it just no longer has any special persistence behaviour.

Skill prompts scrubbed of every "shared wiki memory" / "repo wiki" /
"append a delta before agent_finish" instruction:
``coordination/source_aware_whitebox.md``,
``custom/source_aware_sast.md``,
``scan_modes/{quick,standard,deep}.md``, plus the WHITE-BOX TESTING
block in ``agents/prompts/system_prompt.jinja``.

HARNESS_WIKI.md updated to drop the wiki-as-shared-knowledge-base
description, the per-run output-tree references to ``notes/notes.jsonl``
and ``wiki/{note_id}-{slug}.md``, and the ``is_whitebox`` toggle prose.

Net: -178 LoC in notes/tools.py, -45 LoC across skills/system_prompt
and the wiki doc. The notes tool surface (5 ``@function_tool``s) is
unchanged for the agent.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
0xallam
2026-04-25 23:56:22 -07:00
parent f8213452ea
commit 81703e286f
8 changed files with 20 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files
strix/tools/notes/tools.py
+12 -190
View File
@@ -1,10 +1,10 @@
"""Per-run notes (shared across agents).
Persisted to ``run_dir/notes/notes.jsonl`` (replayable event log) and,
for the ``wiki`` category, also rendered as Markdown to
``run_dir/wiki/<slug>.md``. Concurrent appends are serialised by a
threading.RLock so two agents writing simultaneously can't corrupt
the JSONL.
Pure in-memory state for the lifetime of one scan: the module-level
``_notes_storage`` dict is visible to every agent in the same process.
No disk I/O — process exit clears the lot. Concurrent writers are
serialised by ``_notes_lock`` since each tool entry-point dispatches
the impl onto a worker thread via ``asyncio.to_thread``.
"""
from __future__ import annotations
@@ -15,11 +15,7 @@ import logging
import threading
import uuid
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from pathlib import Path
from typing import Any
from agents import RunContextWrapper, function_tool
@@ -30,162 +26,14 @@ logger = logging.getLogger(__name__)
_notes_storage: dict[str, dict[str, Any]] = {}
_VALID_NOTE_CATEGORIES = ["general", "findings", "methodology", "questions", "plan", "wiki"]
_notes_lock = threading.RLock()
_loaded_notes_run_dir: str | None = None
_DEFAULT_CONTENT_PREVIEW_CHARS = 280
def _get_run_dir() -> Path | None:
try:
from strix.telemetry.tracer import get_global_tracer
tracer = get_global_tracer()
if not tracer:
return None
return tracer.get_run_dir()
except (ImportError, OSError, RuntimeError):
return None
def _get_notes_jsonl_path() -> Path | None:
run_dir = _get_run_dir()
if not run_dir:
return None
notes_dir = run_dir / "notes"
notes_dir.mkdir(parents=True, exist_ok=True)
return notes_dir / "notes.jsonl"
def _append_note_event(op: str, note_id: str, note: dict[str, Any] | None = None) -> None:
"""Append one note operation to the run's ``notes/notes.jsonl``.
C6: hold ``_notes_lock`` across the file open + write so two
concurrent agents can't interleave bytes mid-line.
"""
notes_path = _get_notes_jsonl_path()
if not notes_path:
return
event: dict[str, Any] = {
"timestamp": datetime.now(UTC).isoformat(),
"op": op,
"note_id": note_id,
}
if note is not None:
event["note"] = note
with _notes_lock, notes_path.open("a", encoding="utf-8") as f:
f.write(f"{json.dumps(event, ensure_ascii=True)}\n")
def _load_notes_from_jsonl(notes_path: Path) -> dict[str, dict[str, Any]]:
hydrated: dict[str, dict[str, Any]] = {}
if not notes_path.exists():
return hydrated
with notes_path.open(encoding="utf-8") as f:
for raw_line in f:
line = raw_line.strip()
if not line:
continue
try:
event = json.loads(line)
except json.JSONDecodeError:
continue
op = str(event.get("op", "")).strip().lower()
note_id = str(event.get("note_id", "")).strip()
if not note_id or op not in {"create", "update", "delete"}:
continue
if op == "delete":
hydrated.pop(note_id, None)
continue
note = event.get("note")
if not isinstance(note, dict):
continue
existing = hydrated.get(note_id, {})
existing.update(note)
hydrated[note_id] = existing
return hydrated
def _ensure_notes_loaded() -> None:
global _loaded_notes_run_dir # noqa: PLW0603
run_dir = _get_run_dir()
run_dir_key = str(run_dir.resolve()) if run_dir else "__no_run_dir__"
if _loaded_notes_run_dir == run_dir_key:
return
_notes_storage.clear()
notes_path = _get_notes_jsonl_path()
if notes_path:
_notes_storage.update(_load_notes_from_jsonl(notes_path))
try:
for note_id, note in _notes_storage.items():
if note.get("category") == "wiki":
_persist_wiki_note(note_id, note)
except OSError:
pass
_loaded_notes_run_dir = run_dir_key
def _sanitize_wiki_title(title: str) -> str:
cleaned = "".join(ch.lower() if ch.isalnum() else "-" for ch in title.strip())
slug = "-".join(part for part in cleaned.split("-") if part)
return slug or "wiki-note"
def _get_wiki_directory() -> Path | None:
try:
run_dir = _get_run_dir()
if not run_dir:
return None
wiki_dir = run_dir / "wiki"
wiki_dir.mkdir(parents=True, exist_ok=True)
except OSError:
return None
else:
return wiki_dir
def _get_wiki_note_path(note_id: str, note: dict[str, Any]) -> Path | None:
wiki_dir = _get_wiki_directory()
if not wiki_dir:
return None
wiki_filename = note.get("wiki_filename")
if not isinstance(wiki_filename, str) or not wiki_filename.strip():
title = note.get("title", "wiki-note")
wiki_filename = f"{note_id}-{_sanitize_wiki_title(str(title))}.md"
note["wiki_filename"] = wiki_filename
return wiki_dir / wiki_filename
def _persist_wiki_note(note_id: str, note: dict[str, Any]) -> None:
wiki_path = _get_wiki_note_path(note_id, note)
if not wiki_path:
return
tags = note.get("tags", [])
tags_line = ", ".join(str(tag) for tag in tags) if isinstance(tags, list) and tags else "none"
content = (
f"# {note.get('title', 'Wiki Note')}\n\n"
f"**Note ID:** {note_id}\n"
f"**Created:** {note.get('created_at', '')}\n"
f"**Updated:** {note.get('updated_at', '')}\n"
f"**Tags:** {tags_line}\n\n"
"## Content\n\n"
f"{note.get('content', '')}\n"
)
wiki_path.write_text(content, encoding="utf-8")
def _remove_wiki_note(note_id: str, note: dict[str, Any]) -> None:
wiki_path = _get_wiki_note_path(note_id, note)
if not wiki_path:
return
if wiki_path.exists():
wiki_path.unlink()
def _filter_notes(
category: str | None = None,
tags: list[str] | None = None,
search_query: str | None = None,
) -> list[dict[str, Any]]:
_ensure_notes_loaded()
filtered: list[dict[str, Any]] = []
for note_id, note in _notes_storage.items():
if category and note.get("category") != category:
@@ -220,9 +68,6 @@ def _to_note_listing_entry(
"created_at": note.get("created_at", ""),
"updated_at": note.get("updated_at", ""),
}
wiki_filename = note.get("wiki_filename")
if isinstance(wiki_filename, str) and wiki_filename:
entry["wiki_filename"] = wiki_filename
content = str(note.get("content", ""))
if include_content:
entry["content"] = content
@@ -234,16 +79,14 @@ def _to_note_listing_entry(
return entry
def _create_note_impl( # noqa: PLR0911
def _create_note_impl(
title: str,
content: str,
category: str = "general",
tags: list[str] | None = None,
) -> dict[str, Any]:
"""Create one note. Public — used by tests."""
with _notes_lock:
try:
_ensure_notes_loaded()
if not title or not title.strip():
return {"success": False, "error": "Title cannot be empty", "note_id": None}
if not content or not content.strip():
@@ -276,13 +119,8 @@ def _create_note_impl( # noqa: PLR0911
"updated_at": timestamp,
}
_notes_storage[note_id] = note
_append_note_event("create", note_id, note)
if category == "wiki":
_persist_wiki_note(note_id, note)
except (ValueError, TypeError) as e:
return {"success": False, "error": f"Failed to create note: {e}", "note_id": None}
except OSError as e:
return {"success": False, "error": f"Failed to persist wiki note: {e}", "note_id": None}
else:
return {
"success": True,
@@ -314,7 +152,6 @@ def _list_notes_impl(
def _get_note_impl(note_id: str) -> dict[str, Any]:
with _notes_lock:
try:
_ensure_notes_loaded()
if not note_id or not note_id.strip():
return {"success": False, "error": "Note ID cannot be empty", "note": None}
note = _notes_storage.get(note_id)
@@ -340,7 +177,6 @@ def _update_note_impl(
) -> dict[str, Any]:
with _notes_lock:
try:
_ensure_notes_loaded()
if note_id not in _notes_storage:
return {"success": False, "error": f"Note with ID '{note_id}' not found"}
note = _notes_storage[note_id]
@@ -355,13 +191,8 @@ def _update_note_impl(
if tags is not None:
note["tags"] = tags
note["updated_at"] = datetime.now(UTC).isoformat()
_append_note_event("update", note_id, note)
if note.get("category") == "wiki":
_persist_wiki_note(note_id, note)
except (ValueError, TypeError) as e:
return {"success": False, "error": f"Failed to update note: {e}"}
except OSError as e:
return {"success": False, "error": f"Failed to persist wiki note: {e}"}
else:
return {
"success": True,
@@ -372,19 +203,13 @@ def _update_note_impl(
def _delete_note_impl(note_id: str) -> dict[str, Any]:
with _notes_lock:
try:
_ensure_notes_loaded()
if note_id not in _notes_storage:
return {"success": False, "error": f"Note with ID '{note_id}' not found"}
note = _notes_storage[note_id]
note_title = note["title"]
if note.get("category") == "wiki":
_remove_wiki_note(note_id, note)
del _notes_storage[note_id]
_append_note_event("delete", note_id)
except (ValueError, TypeError) as e:
return {"success": False, "error": f"Failed to delete note: {e}"}
except OSError as e:
return {"success": False, "error": f"Failed to delete wiki note: {e}"}
else:
return {
"success": True,
@@ -405,10 +230,9 @@ async def create_note(
) -> str:
"""Document an observation, finding, methodology step, or research note.
Notes are your **shared run memory** — they're visible to every
agent in the same scan and persist to ``run_dir/notes/notes.jsonl``
(replayable event log). Wiki-category notes are additionally
rendered as Markdown under ``run_dir/wiki/<slug>.md``.
Notes are visible to every agent in the same scan for the lifetime
of the run; they live in-memory only and are cleared when the
process exits.
For actionable tasks, use ``todo`` instead — notes are for capturing
information, todos are for tracking work.
@@ -422,9 +246,7 @@ async def create_note(
useful for the final scan report.
- ``questions`` — open questions / things to come back to.
- ``plan`` — multi-step plans you want to track.
- ``wiki`` — repository or target source maps shared across agents
in the same run. Use this for codebase architecture notes the
whole agent tree should see.
- ``wiki`` — long-form repository or target maps.
Tags are free-form (e.g. ``["sqli", "auth", "critical"]``) — useful
for later ``list_notes(tags=...)`` filtering.
@@ -528,7 +350,7 @@ async def update_note(
@function_tool(timeout=30)
async def delete_note(ctx: RunContextWrapper, note_id: str) -> str:
"""Delete a note. For wiki notes, also removes the rendered Markdown file.
"""Delete a note.
Args:
note_id: Note id to delete.