core: add shared canonical memlog.py in src/scripts

Single source-of-truth memlog: append-only, chronological working-memory
log for skills. Installs to _bmad/scripts/memlog.py via the existing
src/scripts sync (beside resolve_customization.py), so any skill can call
it at runtime — bmad-spec is the first consumer.

Merges the neutral API (--workspace, free-form --type/--by, generic set)
with crash-safe fsync atomic writes. No lifecycle status by design: a
memory log records completion as an event entry, never a frontmatter flag.
Also accepts --path for callers that hold the file path directly. 30 tests.
This commit is contained in:
Brian Madison 2026-06-09 08:39:05 -05:00
parent ed5827cbdc
commit 4eae47860c
2 changed files with 528 additions and 0 deletions

222
src/scripts/memlog.py Normal file
View File

@ -0,0 +1,222 @@
#!/usr/bin/env python3
# /// script
# requires-python = ">=3.10"
# ///
"""memlog — an append-only memory log: LLM-optimal working memory for a skill.
A memlog is the dense, chronological record of everything that mattered in a piece of
work every item the user generated or accepted kept minimal like human memory: only
what's important, never bloated. It persists ACROSS sessions, so a fresh session can
load it and continue. It is NOT a deliverable; downstream artifacts (a brief, a PRD, a
deck, a report) are *derived* from it on demand. The host skill supplies the vocabulary
by how it calls `append` the tool stays neutral.
It is a FLAT log: there are no sections or grouping. Every entry is one line, recorded
at the END in the order it happened. The chronology itself is the structure an event
like "started technique X" is just another entry, same as an idea or an insight.
Three invariants make it trustworthy:
1. Append-only, chronological. Entries land at the end, in the order they happen.
Nothing is ever inserted backward, reordered, edited, or removed. There is no
edit or delete subcommand by design; history is never rewritten.
2. Write-only / blind. Every command is an atomic, context-free write and echoes the
new state as one line of JSON, so the caller never re-reads the file mid-session.
The one time the file is read is on resume and the caller reads it itself, not
via this script.
3. No lifecycle status. A memory log has no "complete" flag. Whether the work is done,
blocked, or paused is itself a fact that happened, so it is recorded as an entry
(e.g. `append --type event --text "session complete"`), never as frontmatter the
log would have to mutate. The chronology stays the single source of truth, and a
resume learns the state by reading the last entries the same way it learns
everything else.
Atomicity: every write goes to a temp file, is flushed and fsync'd, then atomically
renamed over the target, so a crash never leaves a half-written entry.
The file shape (.memlog.md):
---
topic: Onboarding flow for a budgeting app
goal: lift week-1 retention
updated: 2026-06-07T14:22
---
- (note) user picked techniques: SCAMPER, then Six Thinking Hats
- (technique) started SCAMPER
- (idea) skip the signup wall: let people try with sample data first
- (idea) auto-import one bank account so the first screen shows real numbers
- (question) is open-banking consent too heavy for step one?
- (insight) the "scary numbers" risk and the "real numbers" idea are one lever: show real data, pre-categorized
- (direction) optimize for the anxious first-timer, not the power user
- (decision) lead with one pre-categorized account; defer multi-account import
- (event) session complete
Each entry may carry an optional `--type` what KIND it is (idea, insight, question,
decision, direction, assumption, gap, note, event, ) and an optional `--by` naming
who it came from (e.g. `user`, `coach`), for sessions where authorship matters. Both
render into one short inline tag: `(idea)`, `(idea by user)`, `(by coach)`. Omit them
for a plain note. The host skill names the vocabulary; the script does not enforce one.
Commands:
init (--workspace DIR | --path FILE) [--field k=v ...] create the memlog (errors if it exists)
append (--workspace DIR | --path FILE) --text STR [--type T] [--by W] append one entry at the end
set (--workspace DIR | --path FILE) --key K --value V set/replace a descriptive frontmatter field
Addressing: `--workspace` is the run folder, and the memlog is always {workspace}/.memlog.md.
`--path` points straight at the memlog file instead, for callers that already hold the path.
"""
import argparse
import json
import os
import sys
from datetime import datetime
from pathlib import Path
MEMLOG = ".memlog.md"
def now() -> str:
return datetime.now().strftime("%Y-%m-%dT%H:%M")
def resolve(args) -> Path:
"""The memlog file, from either addressing mode: {workspace}/.memlog.md or an explicit --path."""
return Path(args.path) if args.path else Path(args.workspace) / MEMLOG
def split(text: str) -> tuple[dict, str]:
"""Return (frontmatter dict in source order, body str). Frontmatter is plain key: value.
The closing fence is the first line that is *exactly* `---`, so a `---` inside a
field value (topic/goal are free user text) never truncates the frontmatter.
"""
lines = text.splitlines()
if not lines or lines[0] != "---":
raise ValueError(".memlog.md has no frontmatter")
end = next((i for i in range(1, len(lines)) if lines[i] == "---"), None)
if end is None:
raise ValueError(".memlog.md frontmatter is not terminated")
meta: dict[str, str] = {}
for line in lines[1:end]:
if ":" in line:
k, v = line.split(":", 1)
meta[k.strip()] = v.strip()
return meta, "\n".join(lines[end + 1:]).lstrip("\n")
def render(meta: dict, body: str) -> str:
# Neutralize newlines in values so a multi-line field can't break the fence on re-read.
fm = "\n".join(f"{k}: {' '.join(str(v).splitlines())}" for k, v in meta.items())
return "---\n" + fm + "\n---\n\n" + body.rstrip("\n") + "\n"
def touch(meta: dict) -> None:
"""Stamp `updated` and keep it last so the field order stays predictable."""
meta.pop("updated", None)
meta["updated"] = now()
def write_atomic(path: Path, text: str) -> None:
"""Temp + flush + fsync + atomic rename, so a crash never half-writes an entry."""
tmp = path.with_suffix(path.suffix + ".tmp")
with open(tmp, "w", encoding="utf-8") as f:
f.write(text)
f.flush()
os.fsync(f.fileno())
os.replace(tmp, path)
def entry_count(body: str) -> int:
return sum(1 for ln in body.splitlines() if ln.startswith("- "))
def ack(path: Path, body: str) -> None:
"""Echo new state so the caller never re-reads the file to know where it stands."""
print(json.dumps({
"ok": True,
"memlog": str(path),
"entries": entry_count(body),
}))
def cmd_init(args) -> int:
path = resolve(args)
if path.exists():
print(f"error: {path} already exists; use append/set to update it", file=sys.stderr)
return 2
path.parent.mkdir(parents=True, exist_ok=True)
meta: dict[str, str] = {}
for pair in args.field or []:
if "=" not in pair:
print(f"error: --field expects key=value, got {pair!r}", file=sys.stderr)
return 2
k, v = pair.split("=", 1)
meta[k.strip()] = v.strip()
touch(meta)
write_atomic(path, render(meta, ""))
ack(path, "")
return 0
def cmd_append(args) -> int:
path = resolve(args)
meta, body = split(path.read_text(encoding="utf-8"))
text = " ".join(args.text.split()) # collapse newlines/runs → one-line entry, no prose bloat
label = args.type or ""
if args.by:
label = f"{label} by {args.by}".strip() # attribution: "(idea by user)" / "(by coach)"
tag = f"({label}) " if label else ""
entry = f"- {tag}{text}"
body = (body.rstrip("\n") + "\n" + entry) if body.strip() else entry # always at the end
touch(meta)
write_atomic(path, render(meta, body))
ack(path, body)
return 0
def cmd_set(args) -> int:
path = resolve(args)
meta, body = split(path.read_text(encoding="utf-8"))
meta[args.key] = args.value
touch(meta)
write_atomic(path, render(meta, body))
ack(path, body)
return 0
def add_target(sp) -> None:
"""Every command addresses the memlog the same way: a run folder or an explicit path."""
g = sp.add_mutually_exclusive_group(required=True)
g.add_argument("--workspace", help="run folder; the memlog is {workspace}/.memlog.md")
g.add_argument("--path", help="explicit memlog file path (alternative to --workspace)")
def main(argv: list[str] | None = None) -> int:
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
sub = p.add_subparsers(dest="cmd", required=True)
pi = sub.add_parser("init", help="create the memlog")
add_target(pi)
pi.add_argument("--field", action="append", metavar="KEY=VALUE", help="frontmatter field (repeatable)")
pi.set_defaults(func=cmd_init)
pa = sub.add_parser("append", help="append one entry at the end")
add_target(pa)
pa.add_argument("--text", required=True)
pa.add_argument("--type", help="entry kind, rendered as an inline tag")
pa.add_argument("--by", help="who the entry came from (e.g. user, coach); rendered into the tag")
pa.set_defaults(func=cmd_append)
pset = sub.add_parser("set", help="set a descriptive frontmatter field")
add_target(pset)
pset.add_argument("--key", required=True)
pset.add_argument("--value", required=True)
pset.set_defaults(func=cmd_set)
args = p.parse_args(argv)
return args.func(args)
if __name__ == "__main__":
sys.exit(main())

View File

@ -0,0 +1,306 @@
# /// script
# requires-python = ">=3.10"
# dependencies = ["pytest>=8.0"]
# ///
"""Tests for memlog.py. Run: uv run --with pytest pytest scripts/tests/test_memlog.py
The spine under test is the flat, append-only, chronological invariant: every entry is
one line recorded at the end in the order it happened no sections, no grouping, and no
lifecycle status the log would have to mutate.
"""
import json
import sys
from pathlib import Path
import pytest
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
import memlog # noqa: E402
MEMLOG = ".memlog.md"
@pytest.fixture
def ws(tmp_path):
return str(tmp_path)
def read(ws):
return (Path(ws) / MEMLOG).read_text(encoding="utf-8")
def body_of(ws):
return memlog.split(read(ws))[1]
def entries(ws):
return [ln for ln in body_of(ws).splitlines() if ln.startswith("- ")]
def init(ws, **fields):
fields = fields or {"topic": "Reinvent the lunchbox", "goal": "ideas for a pitch"}
argv = ["init", "--workspace", ws]
for k, v in fields.items():
argv += ["--field", f"{k}={v}"]
assert memlog.main(argv) == 0
def append(ws, text, entry_type=None, by=None):
argv = ["append", "--workspace", ws, "--text", text]
if entry_type:
argv += ["--type", entry_type]
if by:
argv += ["--by", by]
assert memlog.main(argv) == 0
# --- init ---------------------------------------------------------------
def test_init_writes_frontmatter_fields(ws):
init(ws)
meta, body = memlog.split(read(ws))
assert meta["topic"] == "Reinvent the lunchbox"
assert meta["goal"] == "ideas for a pitch"
assert "updated" in meta
assert body.strip() == ""
def test_init_has_no_lifecycle_status(ws):
# A memory log carries no "status" flag; completion is an appended entry, not frontmatter.
init(ws)
meta, _ = memlog.split(read(ws))
assert "status" not in meta
def test_init_arbitrary_fields(ws):
init(ws, topic="T", audience="board")
meta, _ = memlog.split(read(ws))
assert meta["audience"] == "board"
def test_init_refuses_overwrite(ws):
init(ws)
assert memlog.main(["init", "--workspace", ws, "--field", "topic=other"]) == 2
def test_init_creates_missing_workspace(tmp_path):
nested = str(tmp_path / "a" / "b")
assert memlog.main(["init", "--workspace", nested, "--field", "topic=T"]) == 0
assert (Path(nested) / MEMLOG).is_file()
def test_init_rejects_malformed_field(ws):
assert memlog.main(["init", "--workspace", ws, "--field", "noequals"]) == 2
# --- addressing: --workspace and --path are interchangeable --------------
def test_path_addressing_targets_the_file_directly(tmp_path):
target = tmp_path / "run" / ".memlog.md"
assert memlog.main(["init", "--path", str(target), "--field", "topic=T"]) == 0
assert target.is_file()
assert memlog.main(["append", "--path", str(target), "--text", "an idea", "--type", "idea"]) == 0
body = memlog.split(target.read_text(encoding="utf-8"))[1]
assert "- (idea) an idea" in body
def test_workspace_and_path_resolve_to_same_file(ws):
init(ws)
via_path = str(Path(ws) / MEMLOG)
assert memlog.main(["append", "--path", via_path, "--text", "from path"]) == 0
assert memlog.main(["append", "--workspace", ws, "--text", "from workspace"]) == 0
assert entries(ws) == ["- from path", "- from workspace"]
def test_target_is_required(ws):
with pytest.raises(SystemExit):
memlog.main(["append", "--text", "orphan"]) # neither --workspace nor --path
# --- append: flat chronological order is the whole point -----------------
def test_append_lands_at_end_in_order(ws):
init(ws)
append(ws, "first")
append(ws, "second")
append(ws, "third")
assert entries(ws) == ["- first", "- second", "- third"]
def test_no_sections_or_headings_ever(ws):
init(ws)
append(ws, "started foo", entry_type="technique")
append(ws, "an idea", entry_type="idea")
append(ws, "started bar", entry_type="technique")
assert "## " not in body_of(ws) # the flat log never grows headings
def test_type_renders_as_inline_tag(ws):
init(ws)
append(ws, "the earth revolves around the sun", entry_type="idea")
append(ws, "how do we handle stampede?", entry_type="question")
body = body_of(ws)
assert "- (idea) the earth revolves around the sun" in body
assert "- (question) how do we handle stampede?" in body
def test_append_without_type_is_plain_note(ws):
init(ws)
append(ws, "bare entry")
assert entries(ws) == ["- bare entry"]
def test_completion_is_an_entry_not_a_status(ws):
# The documented way to mark a session done: append it. Frontmatter never gains a status.
init(ws)
append(ws, "session complete", entry_type="event")
meta, _ = memlog.split(read(ws))
assert "status" not in meta
assert entries(ws)[-1] == "- (event) session complete"
def test_append_collapses_newlines_into_one_line(ws):
init(ws)
append(ws, "line one\nline two\n spaced out")
assert entries(ws) == ["- line one line two spaced out"]
def test_revisited_technique_is_just_a_later_entry(ws):
# the user's model: switching techniques is an entry, not a section to return to
init(ws)
append(ws, "started SCAMPER", entry_type="technique")
append(ws, "magnetic latch", entry_type="idea")
append(ws, "started Six Hats", entry_type="technique")
append(ws, "stale data risk", entry_type="idea")
append(ws, "started SCAMPER", entry_type="technique") # back to SCAMPER — just appended again
append(ws, "stackable tiers", entry_type="idea")
assert entries(ws) == [
"- (technique) started SCAMPER",
"- (idea) magnetic latch",
"- (technique) started Six Hats",
"- (idea) stale data risk",
"- (technique) started SCAMPER",
"- (idea) stackable tiers",
]
def test_by_renders_attribution_in_tag(ws):
# Creative Partner mode must record whose idea each one was
init(ws)
append(ws, "magnetic latch lid", entry_type="idea", by="user")
append(ws, "lid doubles as a plate", entry_type="idea", by="coach")
body = body_of(ws)
assert "- (idea by user) magnetic latch lid" in body
assert "- (idea by coach) lid doubles as a plate" in body
def test_by_without_type_renders_alone(ws):
init(ws)
append(ws, "off-the-cuff thought", by="coach")
assert entries(ws) == ["- (by coach) off-the-cuff thought"]
def test_heterogeneous_entry_types_coexist(ws):
init(ws)
append(ws, "an idea", entry_type="idea")
append(ws, "an open question", entry_type="question")
append(ws, "a decision we made", entry_type="decision")
append(ws, "user wants mobile-first", entry_type="direction")
body = body_of(ws)
for tag in ("(idea)", "(question)", "(decision)", "(direction)"):
assert tag in body
def test_free_vocabulary_is_not_enforced(ws):
# The tool is neutral: any --type the host skill names renders verbatim.
init(ws)
append(ws, "a custom kind", entry_type="crack")
append(ws, "another", entry_type="lock")
body = body_of(ws)
assert "- (crack) a custom kind" in body
assert "- (lock) another" in body
# --- set: generic descriptive frontmatter, no lifecycle semantics --------
def test_set_adds_field(ws):
init(ws)
memlog.main(["set", "--workspace", ws, "--key", "mode", "--value", "partner"])
assert memlog.split(read(ws))[0]["mode"] == "partner"
def test_set_replaces_field(ws):
init(ws, topic="T", mode="facilitator")
memlog.main(["set", "--workspace", ws, "--key", "mode", "--value", "partner"])
assert memlog.split(read(ws))[0]["mode"] == "partner"
def test_set_preserves_body(ws):
init(ws)
append(ws, "keep me", entry_type="idea")
memlog.main(["set", "--workspace", ws, "--key", "mode", "--value", "partner"])
meta, body = memlog.split(read(ws))
assert meta["mode"] == "partner"
assert "- (idea) keep me" in body
def test_updated_stays_last(ws):
init(ws)
memlog.main(["set", "--workspace", ws, "--key", "owner", "--value", "BMad"])
meta = memlog.split(read(ws))[0]
assert list(meta)[-1] == "updated"
# --- robustness ---------------------------------------------------------
def test_roundtrip_render_is_stable(ws):
init(ws)
append(ws, "one", entry_type="idea")
first = read(ws)
meta, body = memlog.split(first)
assert memlog.render(meta, body) == first
def test_commas_in_field_survive(ws):
init(ws, topic="cars, trains, and planes")
append(ws, "z", entry_type="idea")
meta, _ = memlog.split(read(ws))
assert meta["topic"] == "cars, trains, and planes"
def test_triple_dash_in_field_does_not_corrupt_frontmatter(ws):
# A `---` inside a value must NOT be read as the closing fence: topic stays intact
# and the body never leaks frontmatter text.
init(ws, topic="Pricing --- tiers --- and add-ons")
append(ws, "an idea", entry_type="idea")
meta, body = memlog.split(read(ws))
assert meta["topic"] == "Pricing --- tiers --- and add-ons"
assert entries(ws) == ["- (idea) an idea"]
assert "topic:" not in body # frontmatter never bled into the body
def test_newline_in_field_is_neutralized(ws):
# A value carrying a newline can't break the fence on the next round-trip.
memlog.main(["init", "--workspace", ws, "--field", "topic=line one\nline two"])
append(ws, "x", entry_type="idea")
meta, _ = memlog.split(read(ws))
assert "\n" not in meta["topic"]
def test_append_emits_json_ack(ws, capsys):
init(ws)
append(ws, "x", entry_type="idea")
out = json.loads(capsys.readouterr().out.strip().splitlines()[-1])
assert out["ok"] is True
assert out["entries"] == 1
assert out["memlog"].endswith(MEMLOG)
assert "status" not in out # no lifecycle status
assert "section" not in out # sections are gone
def test_ack_entry_count_climbs(ws, capsys):
init(ws)
append(ws, "a")
append(ws, "b")
out = json.loads(capsys.readouterr().out.strip().splitlines()[-1])
assert out["entries"] == 2