remem-ai 0.6.93

Local-first coding agent memory for Claude Code and OpenAI Codex
Documentation
#!/usr/bin/env python3

from __future__ import annotations

import copy
import importlib.util
import json
import sys
import unittest
from pathlib import Path


MODULE_PATH = Path(__file__).with_name("memory_baseline.py")
SPEC = importlib.util.spec_from_file_location("memory_baseline", MODULE_PATH)
assert SPEC is not None and SPEC.loader is not None
baseline = importlib.util.module_from_spec(SPEC)
sys.modules[SPEC.name] = baseline
SPEC.loader.exec_module(baseline)


def sample_status() -> dict:
    return {
        "version": "0.6.59 (schema v80)",
        "totals": {
            "memories": 10,
            "observations": 2,
            "sessions": 3,
            "raw_messages": 40,
        },
        "review_queue": {
            "pending": 4,
            "median_age_secs": 50,
            "max_age_secs": 90,
            "inflow_7d": 3,
            "resolved_7d": 1,
            "projects": [{"project": "/secret/repo", "pending": 4}],
        },
        "candidate_promotion": [
            {
                "source_kind": "summary",
                "review_status": "pending_review",
                "block_reason": "source_trust_below_floor",
                "total": 4,
                "last_7_days": 3,
            }
        ],
        "raw_archive": {
            "messages": 40,
            "ingest_failures": 1,
            "parse_errors": 0,
            "insert_errors": 2,
            "latest_failure_path": "/secret/transcript.jsonl",
            "latest_failure_message": "secret contents",
        },
        "capture_pipeline": {
            "captured": 20,
            "dropped": 5,
            "unrecovered_spills": 0,
            "latest_drop_detail": "secret command",
            "extract_todo": 2,
            "extract_running": 0,
            "extract_expired": 0,
            "extract_failed": 1,
            "retryable_replay_ranges": 1,
            "active_replay_ranges": 0,
            "quarantined_replay_ranges": 0,
            "pending_candidates": 4,
            "pending_graph_candidates": 0,
        },
        "database": {"path": "/secret/remem.db"},
        "top_projects": [{"project": "/secret/repo"}],
    }


class MemoryBaselineTests(unittest.TestCase):
    def test_report_is_deterministic_for_same_metrics(self) -> None:
        first = baseline.build_report(sample_status())
        second_input = sample_status()
        second_input["candidate_promotion"].reverse()
        second = baseline.build_report(second_input)
        self.assertEqual(first, second)

    def test_report_excludes_paths_and_content(self) -> None:
        encoded = json.dumps(baseline.build_report(sample_status()))
        self.assertNotIn("/secret", encoded)
        self.assertNotIn("secret contents", encoded)
        self.assertNotIn("secret command", encoded)
        self.assertNotIn("projects", encoded)

    def test_digest_changes_when_allowlisted_metric_changes(self) -> None:
        original = baseline.build_report(sample_status())
        changed_input = copy.deepcopy(sample_status())
        changed_input["review_queue"]["pending"] += 1
        changed = baseline.build_report(changed_input)
        self.assertNotEqual(original["metrics_sha256"], changed["metrics_sha256"])

    def test_missing_required_field_fails_closed(self) -> None:
        value = sample_status()
        del value["capture_pipeline"]["extract_failed"]
        with self.assertRaisesRegex(baseline.BaselineError, "extract_failed"):
            baseline.build_report(value)


if __name__ == "__main__":
    unittest.main()