agent-team/agent_team/coordinator.py

@@ -197,7 +197,9 @@ def default_clarify_node_factory() -> Callable[[PipelineState], PipelineState]:
     )
 
 
-def default_plan_node_factory() -> Callable[[PipelineState], PipelineState]:
+def default_plan_node_factory(
+    context_provider: "Callable[[], str] | None" = None,
+) -> Callable[[PipelineState], PipelineState]:
     """Build the live planner node wrapped fail-safe (§3.3 P2).
 
     The canonical planner is :func:`agent_team.nodes.planner.plan_node` (it owns
@@ -209,6 +211,9 @@ def default_plan_node_factory() -> Callable[[PipelineState], PipelineState]:
     reply escalates to Adam rather than taking down the pipeline — the same
     fail-SAFE discipline the clarifier and review stages use.
 
+    ``context_provider`` is the optional WS5 injection seam (memory/handbook
+    context). Default None = byte-identical behavior.
+
     Imported lazily so the coordinator module stays import-clean and the SDK is
     only pulled when the live node is actually built.
     """
@@ -217,6 +222,8 @@ def default_plan_node_factory() -> Callable[[PipelineState], PipelineState]:
 
     def plan_node(state: PipelineState) -> PipelineState:
         try:
+            if context_provider is not None:
+                return planner.plan_node(state, context_provider=context_provider)
             return planner.plan_node(state)
         except planner.PlannerError:
             # Unparseable plan -> park + ALARM rather than crash the graph.

agent-team/agent_team/nodes/clarifier_llm.py

@@ -44,6 +44,11 @@ from agent_team.nodes.clarifier import (
 )
 from agent_team.task_model import PipelineState
 
+# Optional context-provider callable: () -> str. When injected, the returned
+# string (e.g. memory + handbook summary) is prepended to the clarifier prompt
+# before the system instruction. When None (default), behavior is unchanged.
+ContextProvider = Callable[[], str]
+
 __all__ = [
     "FALLBACK_QUESTION",
     "ClaudeClarifier",
@@ -130,12 +135,16 @@ class ClaudeClarifier:
         config: Any = None,
         system: str = _DEFAULT_SYSTEM,
         confidence_threshold: float = DEFAULT_CONFIDENCE_THRESHOLD,
+        context_provider: ContextProvider | None = None,
     ) -> None:
         self._invoke: ClaudeInvoke = invoke if invoke is not None else claude_invoke
         self._model = model
         self._config = config
         self._system = system
         self._confidence_threshold = confidence_threshold
+        # Optional seam (WS5): when set, called once per turn and its result
+        # prepended to the prompt. Default None = byte-identical behavior.
+        self._context_provider = context_provider
         # Memo of the single per-turn call. The key is task-scoped, NOT just the
         # history length: one ClaudeClarifier instance serves every task through
         # the long-lived graph node, so a key of len(qa_history) alone would let
@@ -212,7 +221,18 @@ class ClaudeClarifier:
         qa = _format_qa_history(qa_history)
         threshold_pct = int(round(self._confidence_threshold * 100))
 
-        sections: list[str] = [
+        sections: list[str] = []
+        # Prepend optional memory/handbook context when a provider is injected
+        # (WS5). Absent provider = identical output so existing callers are
+        # unaffected. The provider must never raise; a failure returns "" safely.
+        if self._context_provider is not None:
+            try:
+                ctx = self._context_provider()
+            except Exception:  # noqa: BLE001
+                ctx = ""
+            if ctx:
+                sections += [ctx, ""]
+        sections += [
             self._system,
             "",
             "## Task",
@@ -290,6 +310,7 @@ def build_claude_clarifier_callables(
     config: Any = None,
     system: str = _DEFAULT_SYSTEM,
     confidence_threshold: float = DEFAULT_CONFIDENCE_THRESHOLD,
+    context_provider: ContextProvider | None = None,
 ) -> tuple[ConfidenceAssessor, QuestionGenerator]:
     """Build the ``(assess_confidence, generate_questions)`` pair for wiring.
 
@@ -305,6 +326,7 @@ def build_claude_clarifier_callables(
         config=config,
         system=system,
         confidence_threshold=confidence_threshold,
+        context_provider=context_provider,
     )
     return clarifier.assess_confidence, clarifier.generate_questions
 

agent-team/agent_team/nodes/handbook.py

@@ -0,0 +1,87 @@
+"""Sea Haven engineering-handbook conventions loader (WS5 handbook seam).
+
+Reads a handbook directory (``SEA_HAVEN_HANDBOOK_DIR`` env or
+``~/.sea-haven/engineering-handbook``) and returns a formatted summary of
+its ``*.md`` convention files for optional injection into prompt contexts.
+
+Safe no-op contract: if the directory is absent, empty, or any read fails,
+:func:`load_handbook_conventions` returns ``""`` and never raises. Callers
+treat an empty return as "no handbook available" and omit the context block
+rather than failing.
+
+The rsync that delivers the handbook from the Mac to the R720 box, and the
+nightly ``.github`` pull (systemd timers), are ATTENDED box steps and are
+NOT implemented here.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+from typing import Optional
+
+__all__ = [
+    "SEA_HAVEN_HANDBOOK_ENV",
+    "load_handbook_conventions",
+]
+
+SEA_HAVEN_HANDBOOK_ENV = "SEA_HAVEN_HANDBOOK_DIR"
+_DEFAULT_HANDBOOK_DIR = Path.home() / ".sea-haven" / "engineering-handbook"
+
+# Maximum number of convention files to load (defense-in-depth: avoid
+# accidentally summarising a huge handbook dir on first setup).
+_MAX_FILES = 20
+# Max bytes per file to include in the formatted summary (truncated if larger).
+_MAX_FILE_BYTES = 4096
+
+
+def _handbook_dir(path: Optional[Path | str]) -> Path:
+    """Resolve the handbook directory from an explicit path, env, or default."""
+    if path is not None:
+        return Path(path)
+    env = os.environ.get(SEA_HAVEN_HANDBOOK_ENV, "").strip()
+    if env:
+        return Path(env)
+    return _DEFAULT_HANDBOOK_DIR
+
+
+def load_handbook_conventions(path: Optional[Path | str] = None) -> str:
+    """Load engineering-handbook conventions and return a formatted summary.
+
+    Reads all ``*.md`` files in the resolved handbook directory (up to
+    :data:`_MAX_FILES`) and returns a single formatted string suitable for
+    prepending to a model prompt. Returns ``""`` — never raises — when:
+
+    * the directory does not exist or is not a directory;
+    * no ``*.md`` files are found;
+    * any read error occurs (the file is skipped silently).
+
+    The returned string includes a header and one section per convention
+    file; callers should include it only when it is non-empty.
+    """
+    try:
+        hdir = _handbook_dir(path)
+        if not hdir.is_dir():
+            return ""
+        files = sorted(hdir.glob("*.md"))[:_MAX_FILES]
+        if not files:
+            return ""
+        sections: list[str] = []
+        for fpath in files:
+            try:
+                raw = fpath.read_bytes()[:_MAX_FILE_BYTES].decode(
+                    "utf-8", errors="replace"
+                )
+                sections.append(f"### {fpath.stem}\n{raw.strip()}")
+            except Exception:  # noqa: BLE001 - safe no-op
+                continue
+        if not sections:
+            return ""
+        header = (
+            "## Sea Haven engineering-handbook conventions\n"
+            "These conventions are from the engineering handbook. Apply them "
+            "when planning or clarifying. They may not cover every scenario."
+        )
+        return header + "\n\n" + "\n\n---\n\n".join(sections)
+    except Exception:  # noqa: BLE001 - safe no-op in all failure modes
+        return ""

agent-team/agent_team/nodes/planner.py

@@ -34,11 +34,16 @@ from __future__ import annotations
 
 import json
 import re
+from collections.abc import Callable
 from typing import Any
 
 from agent_team.billing import ClaudeResult, claude_invoke
 from agent_team.task_model import Phase, PipelineState, TaskStatus
 
+# Optional context-provider callable (WS5): () -> str. When injected, its
+# result is prepended to the planner prompt. Default None = unchanged behavior.
+ContextProvider = Callable[[], str]
+
 __all__ = [
     "MAX_PLAN_REVISIONS",
     "PlannerError",
@@ -125,20 +130,36 @@ def _format_review_feedback(review_verdicts: list[Any]) -> str:
     return "\n".join(lines)
 
 
-def build_plan_prompt(state: PipelineState) -> str:
+def build_plan_prompt(
+    state: PipelineState, *, context_provider: ContextProvider | None = None
+) -> str:
     """Build the Claude prompt that drafts (or re-drafts) the phased plan.
 
     Pure string assembly over the graph state — no I/O — so the prompt shape is
     directly unit-testable. On a loop-back (``review_verdicts`` present) the
     prompt instructs Claude to revise the prior plan against the feedback rather
     than start from scratch.
+
+    ``context_provider`` is an optional WS5 injection seam. When set, it is
+    called once and its result prepended to the prompt. Default None = identical
+    behavior so existing callers are unaffected.
     """
     description = _task_description(state)
     qa = _format_qa_history(list(state.get("qa_history", [])))
     feedback = _format_review_feedback(list(state.get("review_verdicts", [])))
     prior_plan = state.get("plan")
 
-    sections = [
+    sections: list[str] = []
+    # Optional memory/handbook context prepended when provider is injected (WS5).
+    if context_provider is not None:
+        try:
+            ctx = context_provider()
+        except Exception:  # noqa: BLE001
+            ctx = ""
+        if ctx:
+            sections += [ctx, ""]
+
+    sections += [
         "You are the PLANNER stage of an agentic SDLC pipeline. Produce a "
         "phased implementation plan for the task below. The clarifier has "
         "already reached confidence with the human, so do not ask questions — "
@@ -252,6 +273,8 @@ def _revision_count(state: PipelineState) -> int:
 def plan_node(
     state: PipelineState,
     config: dict[str, Any] | None = None,
+    *,
+    context_provider: ContextProvider | None = None,
 ) -> PipelineState:
     """LangGraph node: draft/refine the phased plan, then advance to REVIEW.
 
@@ -280,7 +303,7 @@ def plan_node(
             current_phase=Phase.PARKED.value,
         )
 
-    prompt = build_plan_prompt(state)
+    prompt = build_plan_prompt(state, context_provider=context_provider)
     result: ClaudeResult = claude_invoke(prompt, config=config)
     plan = parse_plan(result.text)
     # Record how many times we have planned so review/observability can see it.

agent-team/tests/test_ws5_memory_handbook.py

@@ -0,0 +1,296 @@
+"""WS5 — memory + handbook injection seams tests.
+
+Covers:
+* retriever.save_memory() writes to _box-drafts/ subdir
+* retriever.retrieve() memory_dir param routes to correct dir
+* handbook.load_handbook_conventions() safe no-op when dir absent
+* handbook.load_handbook_conventions() reads files when dir exists
+* clarifier_llm: context_provider absent -> identical prompt; present -> prepended
+* planner: context_provider absent -> identical prompt; present -> prepended
+* coordinator.default_plan_node_factory: threads context_provider into planner
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+# retriever.py lives at the orchestrator root (parents[2] of this file
+# which is at agent-team/tests/test_ws5_memory_handbook.py)
+_ORCHESTRATOR_ROOT = Path(__file__).resolve().parents[2]
+if str(_ORCHESTRATOR_ROOT) not in sys.path:
+    sys.path.insert(0, str(_ORCHESTRATOR_ROOT))
+
+
+# ---------------------------------------------------------------------------
+# retriever: save_memory
+# ---------------------------------------------------------------------------
+
+
+def test_save_memory_writes_to_box_drafts(tmp_path: Path) -> None:
+    from retriever import save_memory
+
+    dest = save_memory("my-note", "hello world", memory_dir=tmp_path)
+    assert dest == tmp_path / "_box-drafts" / "my-note.md"
+    assert dest.exists()
+    assert dest.read_text() == "hello world"
+
+
+def test_save_memory_creates_subdir(tmp_path: Path) -> None:
+    from retriever import save_memory
+
+    save_memory("test", "content", memory_dir=tmp_path)
+    assert (tmp_path / "_box-drafts").is_dir()
+
+
+def test_save_memory_rejects_path_traversal(tmp_path: Path) -> None:
+    from retriever import save_memory
+
+    with pytest.raises(ValueError):
+        save_memory("../evil", "bad", memory_dir=tmp_path)
+    with pytest.raises(ValueError):
+        save_memory("sub/dir", "bad", memory_dir=tmp_path)
+    with pytest.raises(ValueError):
+        save_memory("", "bad", memory_dir=tmp_path)
+
+
+def test_save_memory_rejects_unsafe_allowlist_names(tmp_path: Path) -> None:
+    from retriever import save_memory
+
+    # Allowlist rejects dot-only / hidden / backslash / NUL / over-long names
+    # that the old blocklist let through as malformed-but-contained files.
+    for bad in (".", "..", ".hidden", "a\\b", "a\x00b", "x" * 200, "-leading"):
+        with pytest.raises(ValueError):
+            save_memory(bad, "bad", memory_dir=tmp_path)
+
+
+def test_save_memory_does_not_follow_symlink_out_of_drafts(tmp_path: Path) -> None:
+    import os
+
+    from retriever import save_memory
+
+    drafts = tmp_path / "_box-drafts"
+    drafts.mkdir()
+    outside = tmp_path / "outside.txt"
+    outside.write_text("original")
+    # Pre-plant a symlink in the drafts dir pointing outside it.
+    (drafts / "evil.md").symlink_to(outside)
+
+    # O_NOFOLLOW must refuse to write through the symlink (ELOOP).
+    with pytest.raises(OSError):
+        save_memory("evil", "overwrite attempt", memory_dir=tmp_path)
+    # The outside target is untouched.
+    assert outside.read_text() == "original"
+    assert os.path.islink(drafts / "evil.md")
+
+
+def test_save_memory_does_not_appear_in_live_dir(tmp_path: Path) -> None:
+    from retriever import load_memories, save_memory
+
+    save_memory("draft", "draft content", memory_dir=tmp_path)
+    # load_memories reads the memory_dir top level, not _box-drafts/
+    memories = load_memories(tmp_path)
+    names = [m.name for m in memories]
+    assert "draft" not in names
+
+
+def test_retrieve_uses_memory_dir_param(tmp_path: Path) -> None:
+    """retrieve(memory_dir=...) routes to the given dir (empty -> [])."""
+    from retriever import retrieve
+
+    # Empty dir -> no memories -> returns []
+    result = retrieve("some task", memory_dir=tmp_path)
+    assert result == []
+
+
+# ---------------------------------------------------------------------------
+# handbook: load_handbook_conventions
+# ---------------------------------------------------------------------------
+
+
+def test_handbook_no_op_when_dir_absent(tmp_path: Path) -> None:
+    from agent_team.nodes.handbook import load_handbook_conventions
+
+    missing = tmp_path / "nonexistent"
+    result = load_handbook_conventions(missing)
+    assert result == ""
+
+
+def test_handbook_no_op_when_empty_dir(tmp_path: Path) -> None:
+    from agent_team.nodes.handbook import load_handbook_conventions
+
+    result = load_handbook_conventions(tmp_path)
+    assert result == ""
+
+
+def test_handbook_loads_md_files(tmp_path: Path) -> None:
+    from agent_team.nodes.handbook import load_handbook_conventions
+
+    (tmp_path / "conventions.md").write_text("# Conventions\nUse typed Python.")
+    (tmp_path / "ci.md").write_text("# CI\nRun ruff + pytest.")
+    result = load_handbook_conventions(tmp_path)
+    assert "Use typed Python" in result
+    assert "Run ruff + pytest" in result
+    assert "## Sea Haven engineering-handbook conventions" in result
+
+
+def test_handbook_never_raises_on_unreadable_file(tmp_path: Path) -> None:
+    from agent_team.nodes.handbook import load_handbook_conventions
+
+    (tmp_path / "ok.md").write_text("content")
+    # Simulate an IOError by pointing to a non-directory path as the dir
+    result = load_handbook_conventions(tmp_path / "ok.md")
+    assert result == ""
+
+
+# ---------------------------------------------------------------------------
+# clarifier_llm: context_provider injection seam
+# ---------------------------------------------------------------------------
+
+
+def _make_state(**kwargs: Any) -> dict[str, Any]:
+    return {"task": "add a login form", "thread_id": "t1", **kwargs}
+
+
+def _fake_invoke(prompt: str, **kw: Any) -> Any:
+    from agent_team.billing import ClaudeResult
+
+    return ClaudeResult(
+        text='{"confidence": 0.9, "questions": [], "rationale": "clear"}',
+        mode="api",
+        usage={},
+        raw=None,
+    )
+
+
+def test_clarifier_no_context_provider_produces_base_prompt() -> None:
+    from agent_team.nodes.clarifier_llm import ClaudeClarifier
+
+    c = ClaudeClarifier(invoke=_fake_invoke)
+    state = _make_state()
+    prompt = c._build_prompt([], state)
+    assert "You are the CLARIFIER" in prompt
+    # No context header injected
+    assert "Sea Haven" not in prompt
+
+
+def test_clarifier_context_provider_prepends_context() -> None:
+    from agent_team.nodes.clarifier_llm import ClaudeClarifier
+
+    def _provider() -> str:
+        return "## Context\nsome memory"
+
+    c = ClaudeClarifier(invoke=_fake_invoke, context_provider=_provider)
+    state = _make_state()
+    prompt = c._build_prompt([], state)
+    assert prompt.startswith("## Context\nsome memory")
+    assert "You are the CLARIFIER" in prompt
+
+
+def test_clarifier_no_provider_prompt_identical_to_baseline() -> None:
+    """Absent context_provider -> exact same output as no-kwarg construction."""
+    from agent_team.nodes.clarifier_llm import ClaudeClarifier
+
+    state = _make_state()
+    base = ClaudeClarifier(invoke=_fake_invoke)._build_prompt([], state)
+    explicit_none = ClaudeClarifier(
+        invoke=_fake_invoke, context_provider=None
+    )._build_prompt([], state)
+    assert base == explicit_none
+
+
+def test_clarifier_provider_failure_is_silent() -> None:
+    """A context_provider that raises must not propagate to the caller."""
+
+    def _bad_provider() -> str:
+        raise RuntimeError("network down")
+
+    from agent_team.nodes.clarifier_llm import ClaudeClarifier
+
+    c = ClaudeClarifier(invoke=_fake_invoke, context_provider=_bad_provider)
+    state = _make_state()
+    prompt = c._build_prompt([], state)
+    assert "You are the CLARIFIER" in prompt
+
+
+def test_build_claude_clarifier_callables_accepts_context_provider() -> None:
+    from agent_team.nodes.clarifier_llm import build_claude_clarifier_callables
+
+    called: list[int] = []
+
+    def _provider() -> str:
+        called.append(1)
+        return "## ctx\nhi"
+
+    assess, generate = build_claude_clarifier_callables(
+        invoke=_fake_invoke, context_provider=_provider
+    )
+    state = _make_state()
+    # Calling assess_confidence triggers a prompt build -> provider call
+    assess([], state)
+    assert called, "context_provider was not called"
+
+
+# ---------------------------------------------------------------------------
+# planner: build_plan_prompt context_provider
+# ---------------------------------------------------------------------------
+
+
+def _make_plan_state(**kw: Any) -> Any:
+    from agent_team.task_model import PipelineState
+
+    return PipelineState(task="add a login form", **kw)
+
+
+def test_planner_no_context_provider_prompt_unchanged() -> None:
+    from agent_team.nodes.planner import build_plan_prompt
+
+    state = _make_plan_state()
+    base = build_plan_prompt(state)
+    explicit_none = build_plan_prompt(state, context_provider=None)
+    assert base == explicit_none
+
+
+def test_planner_context_provider_prepends_context() -> None:
+    from agent_team.nodes.planner import build_plan_prompt
+
+    def _provider() -> str:
+        return "## Memory\nremember this"
+
+    state = _make_plan_state()
+    prompt = build_plan_prompt(state, context_provider=_provider)
+    assert prompt.startswith("## Memory\nremember this")
+    assert "You are the PLANNER" in prompt
+
+
+def test_planner_provider_failure_silent() -> None:
+    from agent_team.nodes.planner import build_plan_prompt
+
+    def _bad() -> str:
+        raise ValueError("oops")
+
+    state = _make_plan_state()
+    prompt = build_plan_prompt(state, context_provider=_bad)
+    assert "You are the PLANNER" in prompt
+
+
+# ---------------------------------------------------------------------------
+# coordinator.default_plan_node_factory threads context_provider
+# ---------------------------------------------------------------------------
+
+
+def test_default_plan_node_factory_no_provider_returns_callable() -> None:
+    from agent_team.coordinator import default_plan_node_factory
+
+    node = default_plan_node_factory()
+    assert callable(node)
+
+
+def test_default_plan_node_factory_with_provider_returns_callable() -> None:
+    from agent_team.coordinator import default_plan_node_factory
+
+    node = default_plan_node_factory(context_provider=lambda: "ctx")
+    assert callable(node)

retriever.py

@@ -14,6 +14,7 @@ from __future__ import annotations
 import json
 import math
 import os
+import re
 from dataclasses import dataclass
 from pathlib import Path
 
@@ -126,13 +127,19 @@ def _cosine(a: list[float], b: list[float]) -> float:
     return dot / (math.sqrt(na) * math.sqrt(nb))
 
 
-def retrieve(task: str, k: int = TOP_K_DEFAULT) -> list[dict]:
+def retrieve(
+    task: str, k: int = TOP_K_DEFAULT, *, memory_dir: Path | None = None
+) -> list[dict]:
     """Return the top-k most relevant memories for `task`.
 
     Result: [{"name", "score", "content"}], sorted by descending score.
     Returns [] if the memory dir is missing or contains no memories.
+
+    ``memory_dir`` defaults to :data:`MEMORY_DIR` (the production path).
+    Pass an explicit path for tests or alternate memory stores.
     """
-    memories = load_memories()
+    effective_dir = memory_dir if memory_dir is not None else MEMORY_DIR
+    memories = load_memories(effective_dir)
     if not memories:
         return []
     embeddings = get_or_build_embeddings(memories)
@@ -149,6 +156,56 @@ def retrieve(task: str, k: int = TOP_K_DEFAULT) -> list[dict]:
     ]
 
 
+# Subdirectory under a memory dir where save_memory() writes drafts. Using a
+# separate queue directory keeps review-queue files out of the live memory dir
+# so the retriever never auto-indexes a not-yet-reviewed draft.
+_DRAFTS_SUBDIR = "_box-drafts"
+
+# Allowlist for save_memory() names (defense-in-depth over the old blocklist).
+# Must start alphanumeric, then alphanumerics / dot / underscore / hyphen. This
+# rejects path separators, NUL, leading-dot hidden files, and dot-only names
+# (".", "..") outright — a name cannot escape the drafts dir or be malformed.
+_SAFE_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$")
+_MAX_NAME_LEN = 128
+
+
+def save_memory(name: str, content: str, *, memory_dir: Path | None = None) -> Path:
+    """Write a memory draft to the review queue (``_box-drafts/`` subdir).
+
+    Writes ``{memory_dir}/_box-drafts/{name}.md`` and returns the path. The
+    ``_box-drafts/`` subdir is a REVIEW QUEUE — files land there for human
+    inspection before being promoted to the live memory dir. The retriever
+    never indexes ``_box-drafts/`` entries, so writing here never
+    auto-activates a draft.
+
+    ``memory_dir`` defaults to :data:`MEMORY_DIR`. ``name`` must match the safe
+    allowlist (alphanumeric start; then ``A-Z a-z 0-9 . _ -``; max 128 chars) so
+    it cannot contain path separators, NUL, ``..`` traversal, or be a dot-only /
+    hidden name. Raises ``ValueError`` on unsafe names. Creates the subdir if
+    absent.
+    """
+    if (
+        not name
+        or len(name) > _MAX_NAME_LEN
+        or ".." in name
+        or not _SAFE_NAME_RE.match(name)
+    ):
+        raise ValueError(f"unsafe memory name: {name!r}")
+    effective_dir = memory_dir if memory_dir is not None else MEMORY_DIR
+    drafts_dir = effective_dir / _DRAFTS_SUBDIR
+    drafts_dir.mkdir(parents=True, exist_ok=True)
+    dest = drafts_dir / f"{name}.md"
+    # Symlink-safe write: O_NOFOLLOW makes the open fail (ELOOP) if the final
+    # path component is a pre-planted symlink, closing the TOCTOU where a symlink
+    # in _box-drafts/ could redirect the write outside the dir. O_CREAT|O_TRUNC
+    # preserves the overwrite-on-resave behavior for a regular file.
+    flags = os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW
+    fd = os.open(dest, flags, 0o600)
+    with os.fdopen(fd, "w", encoding="utf-8") as fh:
+        fh.write(content)
+    return dest
+
+
 def format_memories_for_prompt(retrieved: list[dict]) -> str:
     """Render retrieved memories as a system-prompt-friendly block."""
     if not retrieved: