diff --git a/examples/claudecode-skills-memanto/README.md b/examples/claudecode-skills-memanto/README.md new file mode 100644 index 00000000..04ee6a72 --- /dev/null +++ b/examples/claudecode-skills-memanto/README.md @@ -0,0 +1,150 @@ +# Claude Code Skills + Memanto Memory Bridge + +This example shows how Memanto can act as a global memory companion across separate developer skill executions. + +The bridge has two lifecycle hooks: + +- `before_skill(...)`: recall relevant engineering memory and return a concise context block that can be appended to a skill prompt. +- `after_skill(...)`: extract durable project decisions, coding preferences, and codebase quirks from a completed skill transcript, then store them in Memanto. +- `run_with_memory(...)`: wrap any existing skill runner callable with both hooks, so teams can drop the bridge around their current `/tdd`, `/handoff`, or custom skill executor without rewriting those skills. + +![Cross-skill memory demo](assets/demo.gif) + +## Why This Matters + +Developer skills are intentionally small and focused. That is useful, but it means a decision captured during a review skill can disappear before a later testing or implementation skill starts. + +This example keeps those decisions outside the individual skill run: + +1. `/grill-with-docs` reviews an architecture plan and records durable decisions. +2. A fresh `/tdd` run asks for tests in the same project. +3. Memanto recalls the earlier decisions and injects them as a compact engineering-memory block. + +## Files + +```text +examples/claudecode-skills-memanto/ +|-- README.md +|-- assets/demo.gif +|-- make_demo_gif.py +|-- memory_backends.py +|-- requirements.txt +|-- run_cross_skill_demo.py +|-- skill_memory_bridge.py +`-- tests/test_skill_memory_bridge.py +``` + +## Quick Start + +```bash +cd examples/claudecode-skills-memanto +python -m venv .venv +source .venv/bin/activate +pip install -r requirements.txt + +# Runs without external keys by using the local JSON backend. +python run_cross_skill_demo.py --backend file --reset +``` + +## Run With Memanto + +Install and configure Memanto first: + +```bash +pip install memanto +memanto +``` + +Then run the same bridge against the real Memanto CLI backend: + +```bash +python run_cross_skill_demo.py --backend memanto --agent-id claudecode-skills-demo +``` + +## Integration Pattern + +Wrap skill execution with the bridge directly: + +```python +from skill_memory_bridge import SkillMemoryBridge, SkillRun + +bridge = SkillMemoryBridge(memory_backend) +run = SkillRun( + skill_name="/tdd", + task="Add tests for invoice webhook idempotency", + file_paths=["apps/billing/webhooks/stripe.ts"], +) + +memory_context = bridge.before_skill(run) + +skill_prompt = f"{memory_context}\n\n{original_skill_prompt}" + +result = run_skill(skill_prompt) + +bridge.after_skill(run, result.transcript) +``` + +Or use the executor-agnostic wrapper when you already have a function that runs a skill: + +```python +def run_skill(prompt: str) -> str: + # Call your existing Claude Code, shell, or local skill runner here. + return "Learning: Invoice export tests should preserve customer locale." + +result = bridge.run_with_memory( + run, + "Create a handoff note for the invoice export branch.", + run_skill, +) + +print(result.prompt) +print(result.stored_memories) +``` + +`SkillRun.metadata` can carry project, framework, branch, tenant, or issue identifiers into the recall query without changing the bridge internals: + +```python +run = SkillRun( + skill_name="/tdd", + task="Add route tests", + file_paths=["apps/mobile/app/(tabs)/index.tsx"], + metadata={"framework": "expo-router", "project": "mobile-app"}, +) +``` + +The bridge deliberately stores only durable engineering facts: + +- `Decision: Keep Stripe webhook handlers idempotent by event id.` +- `Preference: Tests should cover replayed webhook payloads.` +- `Quirk: Billing code stores timestamps as UTC ISO strings.` + +It avoids saving full prompts, private credentials, or large transient logs. + +## Verification + +Regenerate the GIF: + +```bash +python make_demo_gif.py +``` + +Run the demo: + +```bash +python run_cross_skill_demo.py --backend file --reset +``` + +Run the focused offline tests: + +```bash +python -m unittest discover -s tests -v +``` + +Expected output includes: + +```text +MEMANTO ENGINEERING MEMORY +- Decision: Keep billing writes idempotent by Stripe event id. +- Preference: Add replay tests before changing webhook behavior. +- Quirk: Billing timestamps are stored as UTC ISO strings. +``` diff --git a/examples/claudecode-skills-memanto/assets/demo.gif b/examples/claudecode-skills-memanto/assets/demo.gif new file mode 100644 index 00000000..ab170bd2 Binary files /dev/null and b/examples/claudecode-skills-memanto/assets/demo.gif differ diff --git a/examples/claudecode-skills-memanto/make_demo_gif.py b/examples/claudecode-skills-memanto/make_demo_gif.py new file mode 100644 index 00000000..ea281ec5 --- /dev/null +++ b/examples/claudecode-skills-memanto/make_demo_gif.py @@ -0,0 +1,123 @@ +"""Generate a compact GIF for the Claude Code skills memory demo.""" + +from __future__ import annotations + +from pathlib import Path + +from PIL import Image, ImageDraw, ImageFont + +WIDTH = 1040 +HEIGHT = 640 +BG = (15, 19, 27) +PANEL = (27, 34, 45) +TEXT = (238, 244, 250) +MUTED = (149, 166, 184) +GREEN = (91, 205, 149) +BLUE = (112, 177, 255) + + +STEPS = [ + ( + "Skill run 1: /grill-with-docs", + [ + "Decision: Keep billing writes idempotent by Stripe event id.", + "Preference: Add replay tests before webhook changes.", + "Quirk: Billing timestamps are UTC ISO strings.", + ], + ), + ( + "after_skill hook", + [ + "Extract durable engineering memory.", + "Store only decisions, preferences, quirks, constraints.", + "Skip full prompts, secrets, and noisy logs.", + ], + ), + ( + "Skill run 2: /tdd starts fresh", + [ + "Task: add tests for Stripe webhook replay.", + "Files: stripe.ts, stripe.test.ts", + "Graph state is new; Memanto memory persists.", + ], + ), + ( + "before_skill hook", + [ + "MEMANTO ENGINEERING MEMORY", + "- Keep billing writes idempotent by Stripe event id.", + "- Add replay tests before changing webhook behavior.", + "- Billing timestamps are stored as UTC ISO strings.", + ], + ), + ( + "Prompt injection", + [ + "Append compact memory block to the next skill prompt.", + "The /tdd skill now sees the prior review decisions.", + "No manual context shoving between sessions.", + ], + ), +] + + +def main() -> None: + """Render the animated walkthrough GIF into the local assets directory.""" + assets_dir = Path(__file__).parent / "assets" + assets_dir.mkdir(exist_ok=True) + output = assets_dir / "demo.gif" + frames: list[Image.Image] = [] + font = _font(30) + small = _font(23) + tiny = _font(20) + + for index, (title, lines) in enumerate(STEPS, start=1): + for _ in range(6): + image = Image.new("RGB", (WIDTH, HEIGHT), BG) + draw = ImageDraw.Draw(image) + draw.rounded_rectangle((44, 44, WIDTH - 44, HEIGHT - 44), radius=18, fill=PANEL) + draw.text((82, 82), "Claude Code Skills + Memanto", font=font, fill=GREEN) + draw.text((82, 126), f"Step {index}/5: {title}", font=small, fill=TEXT) + draw.line((82, 174, WIDTH - 82, 174), fill=(63, 77, 94), width=2) + + y = 218 + for line in lines: + color = BLUE if line.startswith("MEMANTO") else TEXT + draw.text((104, y), line, font=small, fill=color) + y += 50 + + draw.text( + (82, HEIGHT - 96), + "Global active memory across isolated developer skills", + font=tiny, + fill=MUTED, + ) + frames.append(image) + + frames[0].save( + output, + save_all=True, + append_images=frames[1:], + duration=1000, + loop=0, + optimize=True, + ) + print(output) + + +def _font(size: int) -> ImageFont.FreeTypeFont | ImageFont.ImageFont: + """Load a readable system font with a portable default fallback.""" + candidates = [ + "/System/Library/Fonts/SFNS.ttf", + "/System/Library/Fonts/Supplemental/Arial.ttf", + "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", + ] + for candidate in candidates: + path = Path(candidate) + if path.exists(): + return ImageFont.truetype(str(path), size=size) + return ImageFont.load_default() + + +if __name__ == "__main__": + main() diff --git a/examples/claudecode-skills-memanto/memory_backends.py b/examples/claudecode-skills-memanto/memory_backends.py new file mode 100644 index 00000000..4378136d --- /dev/null +++ b/examples/claudecode-skills-memanto/memory_backends.py @@ -0,0 +1,308 @@ +"""Memory backends for the Claude Code skills + Memanto demo.""" + +from __future__ import annotations + +import json +import os +import re +import subprocess +import tempfile +import warnings +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +CLI_TIMEOUT_SECONDS = 30 + + +@dataclass +class MemoryRecord: + """Serializable memory entry used by the offline JSON backend.""" + + content: str + memory_type: str = "learning" + source: str = "claudecode-skills-demo" + tags: str = "claudecode,skills,memanto" + created_at: str = "" + + def to_json(self) -> dict[str, str]: + """Return the JSON-ready record with a creation timestamp filled in.""" + payload = asdict(self) + payload["created_at"] = payload["created_at"] or datetime.now( + timezone.utc + ).isoformat() + return payload + + +class BaseMemoryBackend: + """Minimal storage protocol consumed by the skill memory bridge.""" + + def remember( + self, + content: str, + *, + memory_type: str = "learning", + tags: str = "claudecode,skills,memanto", + ) -> None: + """Persist a durable memory extracted from a completed skill run.""" + raise NotImplementedError + + def recall(self, query: str, *, limit: int = 6) -> list[str]: + """Return relevant stored memories for a new skill run query.""" + raise NotImplementedError + + +class FileMemoryBackend(BaseMemoryBackend): + """Local JSON backend for offline reviewer demos.""" + + def __init__(self, path: Path, *, source: str = "claudecode-skills-demo") -> None: + """Create a file-backed memory store at the supplied path.""" + self.path = path + self.source = source + + def remember( + self, + content: str, + *, + memory_type: str = "learning", + tags: str = "claudecode,skills,memanto", + ) -> None: + """Append one memory record to the demo JSON file.""" + self.path.parent.mkdir(parents=True, exist_ok=True) + records = self._load() + records.append( + MemoryRecord( + content=content, + memory_type=memory_type, + source=self.source, + tags=tags, + ).to_json() + ) + _write_records_atomic(self.path, records) + + def recall(self, query: str, *, limit: int = 6) -> list[str]: + """Rank stored memories by token overlap with the recall query.""" + query_terms = _terms(query) + ranked: list[tuple[int, str]] = [] + for record in self._load(): + content = str(record.get("content", "")) + score = len(query_terms.intersection(_terms(content))) + if score: + ranked.append((score, content)) + ranked.sort(key=lambda item: item[0], reverse=True) + return [content for _, content in ranked[:limit]] + + def _load(self) -> list[dict[str, str]]: + """Load stored records, tolerating missing or malformed demo files.""" + if not self.path.exists(): + return [] + try: + payload = json.loads(self.path.read_text(encoding="utf-8")) + if _is_record_list(payload): + return payload + warnings.warn( + f"Ignoring demo memory file with unexpected shape: {self.path}", + RuntimeWarning, + stacklevel=2, + ) + return [] + except json.JSONDecodeError: + warnings.warn( + f"Ignoring malformed demo memory file: {self.path}", + RuntimeWarning, + stacklevel=2, + ) + return [] + except OSError as exc: + warnings.warn( + f"Could not read demo memory file {self.path}: {exc}", + RuntimeWarning, + stacklevel=2, + ) + return [] + + +class MemantoCliBackend(BaseMemoryBackend): + """Backend that uses the installed Memanto package and CLI session.""" + + def __init__(self, agent_id: str) -> None: + """Bind the backend to a Memanto agent and activate its session.""" + self.agent_id = agent_id + self._ensure_agent() + + def remember( + self, + content: str, + *, + memory_type: str = "learning", + tags: str = "claudecode,skills,memanto", + ) -> None: + """Store a memory through the Memanto SDK with provenance metadata.""" + self._client().remember( + agent_id=self.agent_id, + memory_type=memory_type, + title=_memory_title(content), + content=content, + confidence=0.8, + tags=_split_tags(tags), + source=self.agent_id, + provenance="explicit_statement", + ) + + def recall(self, query: str, *, limit: int = 6) -> list[str]: + """Recall relevant Memanto memories using the SDK response payload.""" + result = self._client().recall( + agent_id=self.agent_id, + query=query, + limit=limit, + tags=_split_tags("claudecode,skills,memanto"), + ) + memories = result.get("memories", []) + recalled: list[str] = [] + for memory in memories: + content = str(memory.get("content") or memory.get("title") or "").strip() + if content: + recalled.append(content) + return recalled[:limit] + + def _ensure_agent(self) -> None: + """Create or activate the configured Memanto demo agent.""" + try: + create = subprocess.run( + ["memanto", "agent", "create", self.agent_id], + capture_output=True, + text=True, + check=False, + timeout=CLI_TIMEOUT_SECONDS, + ) + except OSError as exc: + raise _missing_memanto_error() from exc + except subprocess.TimeoutExpired as exc: + raise RuntimeError("Memanto CLI agent creation timed out") from exc + if create.returncode == 0: + return + detail = _command_detail(create.stdout, create.stderr) + if "already exists" not in detail.lower() and "exists" not in detail.lower(): + raise RuntimeError(f"Memanto CLI agent creation failed: {detail}") + _run(["memanto", "agent", "activate", self.agent_id]) + + def _client(self) -> Any: + """Build a configured SDK client for the active Memanto session.""" + try: + from memanto.cli.client.sdk_client import SdkClient + from memanto.cli.config.manager import ConfigManager + except ImportError as exc: + raise _missing_memanto_error() from exc + + config = ConfigManager() + api_key = config.get_api_key() + if not api_key: + raise RuntimeError( + "MEMANTO is not configured. Run `memanto` to set an API key." + ) + + active_agent_id, active_session_token = config.get_active_session() + if active_agent_id != self.agent_id or not active_session_token: + self._ensure_agent() + active_agent_id, active_session_token = config.get_active_session() + if active_agent_id != self.agent_id or not active_session_token: + raise RuntimeError(f"Memanto agent `{self.agent_id}` is not active.") + + client = SdkClient(api_key) + client.agent_id = self.agent_id + client.session_token = active_session_token + return client + + +def _terms(text: str) -> set[str]: + """Extract lowercase search terms from a free-form string.""" + return set(re.findall(r"[a-z0-9]{3,}", text.lower())) + + +def _write_records_atomic(path: Path, records: list[dict[str, str]]) -> None: + """Write memory records through a same-directory temp file replacement.""" + payload = json.dumps(records, indent=2) + "\n" + path.parent.mkdir(parents=True, exist_ok=True) + tmp_name = "" + try: + with tempfile.NamedTemporaryFile( + "w", + dir=path.parent, + encoding="utf-8", + prefix=f".{path.name}.", + suffix=".tmp", + delete=False, + ) as tmp: + tmp_name = tmp.name + tmp.write(payload) + tmp.flush() + os.fsync(tmp.fileno()) + Path(tmp_name).replace(path) + except Exception: + if tmp_name: + Path(tmp_name).unlink(missing_ok=True) + raise + + +def _is_record_list(payload: Any) -> bool: + """Return whether parsed JSON matches list[dict[str, str]].""" + return isinstance(payload, list) and all( + isinstance(item, dict) + and all( + isinstance(key, str) and isinstance(value, str) + for key, value in item.items() + ) + for item in payload + ) + + +def _split_tags(tags: str) -> list[str]: + """Split a comma-separated tag list into non-empty tag names.""" + return [tag.strip() for tag in tags.split(",") if tag.strip()] + + +def _memory_title(content: str) -> str: + """Create a compact Memanto title from memory content.""" + return content[:47] + "..." if len(content) > 50 else content + + +def _command_detail(stdout: str | bytes | None, stderr: str | bytes | None) -> str: + """Return the most helpful text from a completed or timed-out command.""" + for stream in (stderr, stdout): + if isinstance(stream, bytes): + stream = stream.decode(errors="replace") + detail = (stream or "").strip() + if detail: + return detail + return "unknown CLI error" + + +def _run(cmd: list[str]) -> str: + """Run a Memanto CLI command with bounded execution time.""" + try: + result = subprocess.run( + cmd, + capture_output=True, + text=True, + check=True, + timeout=CLI_TIMEOUT_SECONDS, + ) + except OSError as exc: + raise _missing_memanto_error() from exc + except subprocess.TimeoutExpired as exc: + detail = _command_detail(exc.stdout, exc.stderr) + raise RuntimeError(f"Memanto CLI command timed out: {detail}") from exc + except subprocess.CalledProcessError as exc: + detail = _command_detail(exc.stdout, exc.stderr) + raise RuntimeError(f"Memanto CLI command failed: {detail}") from exc + return result.stdout + + +def _missing_memanto_error() -> RuntimeError: + """Create a consistent error for missing or unconfigured Memanto tooling.""" + return RuntimeError( + "The `memanto` CLI was not found. Run `pip install memanto` and " + "`memanto` to configure your Moorcheh API key, or use " + "`--backend file` for the offline demo." + ) diff --git a/examples/claudecode-skills-memanto/requirements.txt b/examples/claudecode-skills-memanto/requirements.txt new file mode 100644 index 00000000..ee19be25 --- /dev/null +++ b/examples/claudecode-skills-memanto/requirements.txt @@ -0,0 +1 @@ +memanto>=0.1.0 diff --git a/examples/claudecode-skills-memanto/run_cross_skill_demo.py b/examples/claudecode-skills-memanto/run_cross_skill_demo.py new file mode 100644 index 00000000..9b6480ee --- /dev/null +++ b/examples/claudecode-skills-memanto/run_cross_skill_demo.py @@ -0,0 +1,73 @@ +"""Demonstrate Memanto memory moving between separate skill runs.""" + +from __future__ import annotations + +import argparse +from pathlib import Path + +from memory_backends import FileMemoryBackend, MemantoCliBackend +from skill_memory_bridge import SkillMemoryBridge, SkillRun + +REVIEW_TRANSCRIPT = """ +/grill-with-docs reviewed the billing webhook plan. + +Decision: Keep billing writes idempotent by Stripe event id. +Preference: Add replay tests before changing webhook behavior. +Quirk: Billing timestamps are stored as UTC ISO strings. +Constraint: Do not persist raw Stripe payloads after signature verification. +""" + + +def parse_args() -> argparse.Namespace: + """Parse backend and reset options for the demo runner.""" + parser = argparse.ArgumentParser( + description="Run the Claude Code skills + Memanto memory bridge demo.", + ) + parser.add_argument("--backend", choices=["file", "memanto"], default="file") + parser.add_argument("--agent-id", default="claudecode-skills-demo") + parser.add_argument("--memory-file", default=".demo_skill_memory.json") + parser.add_argument("--reset", action="store_true") + return parser.parse_args() + + +def main() -> None: + """Run two isolated skill sessions that share memory through the bridge.""" + args = parse_args() + if args.backend == "file" and args.reset: + Path(args.memory_file).unlink(missing_ok=True) + + memory = ( + MemantoCliBackend(args.agent_id) + if args.backend == "memanto" + else FileMemoryBackend(Path(args.memory_file), source=args.agent_id) + ) + bridge = SkillMemoryBridge(memory) + + review_run = SkillRun( + skill_name="/grill-with-docs", + task="Review billing webhook architecture", + file_paths=["apps/billing/webhooks/stripe.ts", "apps/billing/db/events.ts"], + ) + print("Session 1: /grill-with-docs finishes and stores durable memory") + stored = bridge.after_skill(review_run, REVIEW_TRANSCRIPT) + for item in stored: + print(f"- remembered: {item}") + + tdd_run = SkillRun( + skill_name="/tdd", + task="Add tests for Stripe webhook replay and invoice creation", + file_paths=["apps/billing/webhooks/stripe.ts", "apps/billing/webhooks/stripe.test.ts"], + ) + print("\nSession 2: /tdd starts fresh and asks Memanto for context") + context = bridge.before_skill(tdd_run) + print(context) + + print("\nPrompt fragment for the next skill:") + print( + "Use the recalled engineering memory above when choosing test cases, " + "fixtures, and persistence assertions." + ) + + +if __name__ == "__main__": + main() diff --git a/examples/claudecode-skills-memanto/skill_memory_bridge.py b/examples/claudecode-skills-memanto/skill_memory_bridge.py new file mode 100644 index 00000000..a25e0dbf --- /dev/null +++ b/examples/claudecode-skills-memanto/skill_memory_bridge.py @@ -0,0 +1,132 @@ +"""Lifecycle bridge between developer skills and Memanto.""" + +from __future__ import annotations + +import re +from collections.abc import Callable +from dataclasses import dataclass, field + +from memory_backends import BaseMemoryBackend + +MEMORY_LINE = re.compile( + r"^\s*(Decision|Preference|Quirk|Constraint|Learning):\s*(.+)$", + flags=re.IGNORECASE | re.MULTILINE, +) +TAG_SEPARATOR = re.compile(r"[^a-z0-9]+") + + +@dataclass +class SkillRun: + """Metadata about one isolated developer skill execution.""" + + skill_name: str + task: str + file_paths: list[str] + metadata: dict[str, str] = field(default_factory=dict) + + +@dataclass +class SkillExecution: + """Result returned when the bridge wraps an arbitrary skill runner.""" + + prompt: str + transcript: str + stored_memories: list[str] + + +class SkillMemoryBridge: + """Drop-in recall-before and remember-after hooks for skill execution.""" + + def __init__( + self, + memory: BaseMemoryBackend, + *, + header: str = "MEMANTO ENGINEERING MEMORY", + base_tags: tuple[str, ...] = ("claudecode", "skills"), + ) -> None: + """Create a bridge around any compatible memory backend.""" + self.memory = memory + self.header = header + self.base_tags = base_tags + + def before_skill(self, run: SkillRun, *, limit: int = 6) -> str: + """Format relevant recalled memories for injection into a skill prompt.""" + query = self._query_for(run) + memories = self.memory.recall(query, limit=limit) + if not memories: + return f"{self.header}\n- No relevant memories found." + lines = [self.header] + lines.extend(f"- {memory}" for memory in memories) + return "\n".join(lines) + + def prompt_with_memory( + self, + run: SkillRun, + original_prompt: str, + *, + limit: int = 6, + ) -> str: + """Return a skill prompt prefixed with recalled engineering memory.""" + memory_context = self.before_skill(run, limit=limit) + prompt = original_prompt.strip() + if not prompt: + return memory_context + return f"{memory_context}\n\n{prompt}" + + def run_with_memory( + self, + run: SkillRun, + original_prompt: str, + executor: Callable[[str], str], + *, + limit: int = 6, + ) -> SkillExecution: + """Wrap any skill runner callable with Memanto memory hooks.""" + prompt = self.prompt_with_memory(run, original_prompt, limit=limit) + transcript = executor(prompt) + stored_memories = self.after_skill(run, transcript) + return SkillExecution( + prompt=prompt, + transcript=transcript, + stored_memories=stored_memories, + ) + + def after_skill(self, run: SkillRun, transcript: str) -> list[str]: + """Extract labeled durable memories from a completed skill transcript.""" + stored: list[str] = [] + for label, value in MEMORY_LINE.findall(transcript): + memory = f"{label.title()}: {value.strip()}" + memory_type = self._memory_type(label) + tags = ",".join(self._tags_for(run)) + self.memory.remember(memory, memory_type=memory_type, tags=tags) + stored.append(memory) + return stored + + def _query_for(self, run: SkillRun) -> str: + """Build a compact recall query from skill metadata.""" + path_text = " ".join(run.file_paths) + metadata_text = " ".join(f"{key}:{value}" for key, value in run.metadata.items()) + return f"{run.skill_name} {run.task} {path_text} {metadata_text}" + + def _tags_for(self, run: SkillRun) -> tuple[str, ...]: + """Build stable tags for memories emitted by one skill run.""" + skill_tag = self._sanitize_tag(run.skill_name) + if skill_tag: + return (*self.base_tags, skill_tag) + return self.base_tags + + def _sanitize_tag(self, value: str) -> str: + """Normalize user-facing skill names into comma-safe memory tags.""" + normalized = TAG_SEPARATOR.sub("-", value.strip().lower()) + return normalized.strip("-") + + def _memory_type(self, label: str) -> str: + """Map a transcript label to Memanto's memory type vocabulary.""" + normalized = label.lower() + if normalized == "decision": + return "decision" + if normalized == "preference": + return "preference" + if normalized in {"quirk", "constraint"}: + return "context" + return "learning" diff --git a/examples/claudecode-skills-memanto/tests/test_skill_memory_bridge.py b/examples/claudecode-skills-memanto/tests/test_skill_memory_bridge.py new file mode 100644 index 00000000..347b86ea --- /dev/null +++ b/examples/claudecode-skills-memanto/tests/test_skill_memory_bridge.py @@ -0,0 +1,205 @@ +"""Tests for the Claude Code skills + Memanto bridge example.""" + +from __future__ import annotations + +import json +import sys +import tempfile +import unittest +import warnings +from pathlib import Path + +EXAMPLE_DIR = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(EXAMPLE_DIR)) + +from memory_backends import FileMemoryBackend # noqa: E402 +from skill_memory_bridge import ( # noqa: E402 + SkillExecution, + SkillMemoryBridge, + SkillRun, +) + + +class SkillMemoryBridgeTests(unittest.TestCase): + """Exercise the offline reviewer path used by the bounty PR.""" + + def test_bridge_stores_labeled_memories_and_recalls_relevant_context(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + memory = FileMemoryBackend(Path(tmp_dir) / "memory.json") + bridge = SkillMemoryBridge(memory) + review_run = SkillRun( + skill_name="/grill-with-docs", + task="Review billing webhook behavior", + file_paths=["apps/billing/webhooks/stripe.ts"], + ) + + stored = bridge.after_skill( + review_run, + """ + Decision: Keep writes idempotent by Stripe event id. + Preference: Add replay tests before webhook changes. + Quirk: Billing timestamps are UTC ISO strings. + Constraint: Do not persist raw Stripe payloads. + Note: This unlabeled line should not be stored. + """, + ) + + self.assertEqual( + stored, + [ + "Decision: Keep writes idempotent by Stripe event id.", + "Preference: Add replay tests before webhook changes.", + "Quirk: Billing timestamps are UTC ISO strings.", + "Constraint: Do not persist raw Stripe payloads.", + ], + ) + tdd_run = SkillRun( + skill_name="/tdd", + task="Add Stripe webhook replay tests", + file_paths=["apps/billing/webhooks/stripe.test.ts"], + ) + + context = bridge.before_skill(tdd_run) + + self.assertIn("MEMANTO ENGINEERING MEMORY", context) + self.assertIn("Stripe event id", context) + self.assertIn("Add replay tests", context) + self.assertNotIn("unlabeled line", context) + + def test_file_backend_ranks_recall_by_query_overlap(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + memory = FileMemoryBackend(Path(tmp_dir) / "memory.json") + memory.remember("Decision: Stripe webhook replay tests use event ids.") + memory.remember("Learning: Dashboard filters use URL search params.") + memory.remember("Constraint: Stripe payloads are discarded after signature checks.") + + recalled = memory.recall("Stripe webhook replay event tests", limit=2) + + self.assertEqual( + recalled, + [ + "Decision: Stripe webhook replay tests use event ids.", + "Constraint: Stripe payloads are discarded after signature checks.", + ], + ) + + def test_bridge_wraps_any_skill_executor_with_memory_hooks(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + memory = FileMemoryBackend(Path(tmp_dir) / "memory.json") + memory.remember( + "Decision: Invoice exports must preserve customer locale settings." + ) + bridge = SkillMemoryBridge(memory) + run = SkillRun( + skill_name="/handoff", + task="Summarize invoice export implementation", + file_paths=["apps/billing/invoices/export.ts"], + metadata={"project": "billing"}, + ) + prompts_seen: list[str] = [] + + def fake_executor(prompt: str) -> str: + prompts_seen.append(prompt) + return "Learning: Invoice exports need a locale regression test." + + result = bridge.run_with_memory( + run, + "Create a handoff note for the invoice export branch.", + fake_executor, + ) + + self.assertIsInstance(result, SkillExecution) + self.assertEqual(prompts_seen, [result.prompt]) + self.assertIn("MEMANTO ENGINEERING MEMORY", result.prompt) + self.assertIn("preserve customer locale", result.prompt) + self.assertIn("Create a handoff note", result.prompt) + self.assertEqual( + result.stored_memories, + ["Learning: Invoice exports need a locale regression test."], + ) + + records = json.loads((Path(tmp_dir) / "memory.json").read_text()) + self.assertEqual(records[-1]["tags"], "claudecode,skills,handoff") + + def test_metadata_contributes_to_recall_query(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + memory = FileMemoryBackend(Path(tmp_dir) / "memory.json") + memory.remember("Decision: Mobile builds use expo-router defaults.") + bridge = SkillMemoryBridge(memory) + run = SkillRun( + skill_name="/tdd", + task="Add route tests", + file_paths=[], + metadata={"framework": "expo-router"}, + ) + + context = bridge.before_skill(run) + + self.assertIn("expo-router defaults", context) + + def test_skill_name_is_sanitized_before_tag_storage(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + memory = FileMemoryBackend(Path(tmp_dir) / "memory.json") + bridge = SkillMemoryBridge(memory) + run = SkillRun( + skill_name="/Custom skill,runner/v2!!", + task="Summarize memory bridge tag handling", + file_paths=[], + ) + + bridge.after_skill(run, "Learning: Tags should stay comma-safe.") + bridge.after_skill( + SkillRun( + skill_name=" ///,,, ", + task="Summarize empty skill tag handling", + file_paths=[], + ), + "Decision: Empty skill tags fall back to base tags.", + ) + + records = json.loads((Path(tmp_dir) / "memory.json").read_text()) + self.assertEqual(records[0]["tags"], "claudecode,skills,custom-skill-runner-v2") + self.assertEqual(records[1]["tags"], "claudecode,skills") + + def test_malformed_offline_memory_file_recovers_on_write(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = Path(tmp_dir) / "memory.json" + path.write_text("{not-json", encoding="utf-8") + memory = FileMemoryBackend(path) + bridge = SkillMemoryBridge(memory) + run = SkillRun( + skill_name="/handoff", + task="Summarize webhook constraints", + file_paths=["apps/billing/webhooks/stripe.ts"], + ) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + self.assertEqual(memory.recall("webhook"), []) + bridge.after_skill(run, "Learning: Keep webhook fixtures minimal.") + + self.assertTrue(caught) + records = json.loads(path.read_text(encoding="utf-8")) + self.assertEqual( + [record["content"] for record in records], + ["Learning: Keep webhook fixtures minimal."], + ) + + def test_unexpected_offline_memory_shape_recovers_on_write(self) -> None: + with tempfile.TemporaryDirectory() as tmp_dir: + path = Path(tmp_dir) / "memory.json" + path.write_text('{"content": "not a list"}', encoding="utf-8") + memory = FileMemoryBackend(path) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + memory.remember("Decision: Use a same-directory temp file.") + + self.assertTrue(caught) + records = json.loads(path.read_text(encoding="utf-8")) + self.assertEqual(records[0]["content"], "Decision: Use a same-directory temp file.") + self.assertEqual(records[0]["memory_type"], "learning") + + +if __name__ == "__main__": + unittest.main()