import io
import json
import os
import sqlite3
import sys
import tempfile
import unittest
from unittest.mock import patch
from contextlib import redirect_stdout

from openclaw_mem.cli import _connect, _enable_wal_best_effort, _insert_observation, _summary_has_task_marker, _normalize_importance_scorer_value, _pack_graph_resolve_scope, build_parser, cmd_ingest, cmd_search, cmd_get, cmd_timeline, cmd_triage, cmd_store, cmd_hybrid, cmd_pack, cmd_status, cmd_doctor, cmd_profile, cmd_backend, cmd_graph_index, cmd_graph_pack, cmd_graph_preflight, cmd_graph_auto_status, cmd_graph_capture_git, cmd_graph_capture_md, cmd_graph_export, cmd_graph_synth, cmd_graph_lint, cmd_vsearch




class TestSqliteConnectionPosture(unittest.TestCase):
    def test_wal_setup_is_best_effort_for_readonly_database(self):
        class FakeConn:
            def __init__(self):
                self.calls = []

            def execute(self, sql):
                self.calls.append(sql)
                if sql == "PRAGMA journal_mode=WAL;":
                    raise sqlite3.OperationalError("attempt to write a readonly database")
                return None

        conn = FakeConn()
        _enable_wal_best_effort(conn)
        self.assertEqual(
            conn.calls,
            ["PRAGMA journal_mode=WAL;", "PRAGMA busy_timeout=5000;"],
        )

    def test_wal_setup_still_raises_unexpected_operational_errors(self):
        class FakeConn:
            def execute(self, sql):
                if sql == "PRAGMA journal_mode=WAL;":
                    raise sqlite3.OperationalError("database disk image is malformed")
                return None

        with self.assertRaises(sqlite3.OperationalError):
            _enable_wal_best_effort(FakeConn())


class TestParserContracts(unittest.TestCase):
    def test_nested_episodes_db_flag_preserves_parent_value(self):
        parser = build_parser()

        parent = parser.parse_args(["episodes", "--db", "/tmp/parent.sqlite", "embed"])
        child = parser.parse_args(["episodes", "embed", "--db", "/tmp/child.sqlite"])
        global_arg = parser.parse_args(["--db", "/tmp/global.sqlite", "episodes", "embed"])

        self.assertEqual(parent.db, "/tmp/parent.sqlite")
        self.assertEqual(child.db, "/tmp/child.sqlite")
        self.assertEqual(global_arg.db_global, "/tmp/global.sqlite")
        self.assertIsNone(global_arg.db)

    def test_nested_episodes_json_flag_preserves_parent_value(self):
        parser = build_parser()

        parent = parser.parse_args(["episodes", "--json", "embed"])
        child = parser.parse_args(["episodes", "embed", "--json"])

        self.assertTrue(parent.json)
        self.assertTrue(child.json)


class TestCliM0(unittest.TestCase):
    def test_schema_contains_english_embeddings_table(self):
        conn = _connect(":memory:")
        tables = {
            r[0]
            for r in conn.execute(
                "SELECT name FROM sqlite_master WHERE type = 'table'"
            ).fetchall()
        }
        self.assertIn("observation_embeddings_en", tables)

        args = type("Args", (), {"db": ":memory:", "json": True})()
        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_status(conn, args)
        out = json.loads(buf.getvalue())
        self.assertIn("embeddings_en", out)
        self.assertEqual(out["embeddings_en"]["count"], 0)
        self.assertEqual(out["kind"], "openclaw-mem.status.v1")
        self.assertIn("backend", out)
        self.assertIn("runtime", out)
        conn.close()

    def test_doctor_emits_compact_health_contract(self):
        conn = _connect(":memory:")
        args = type("Args", (), {"db": ":memory:", "json": True, "db_preexisted": True})()
        fake_cfg = {
            "plugins": {
                "slots": {"memory": "memory-core"},
                "entries": {
                    "memory-core": {"enabled": True},
                    "memory-lancedb": {"enabled": False, "config": {}},
                    "openclaw-mem": {"enabled": True},
                },
            }
        }
        buf = io.StringIO()
        with patch("openclaw_mem.cli._read_openclaw_config", return_value=fake_cfg), redirect_stdout(buf):
            cmd_doctor(conn, args)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["kind"], "openclaw-mem.doctor.v0")
        self.assertIn("summary", out)
        self.assertIn("checks", out)
        self.assertTrue(any(item["name"] == "db.connection" for item in out["checks"]))
        conn.close()

    def test_pack_falls_back_to_fts_when_no_api_key(self):
        conn = _connect(":memory:")
        _insert_observation(
            conn,
            {
                "kind": "preference",
                "summary": "Prefers using Asia/Taipei (UTC+8) timezone for time displays.",
                "tool_name": "memory_store",
                "detail": {"importance": {"score": 0.8, "label": "must_remember"}},
            },
        )

        args = type(
            "Args",
            (),
            {
                "query": "timezone",
                "query_en": None,
                "limit": 12,
                "budget_tokens": 400,
                "trace": False,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_pack(conn, args)
        payload = json.loads(buf.getvalue())
        self.assertEqual(payload["context_pack"]["schema"], "openclaw-mem.context-pack.v1")
        self.assertEqual(payload["context_pack"]["bundle_text"], payload["bundle_text"])
        texts = [it.get("summary") for it in payload.get("items", [])]
        self.assertTrue(any("UTC+8" in (t or "") for t in texts))
        conn.close()

    def test_pack_prefers_compaction_sideband_text_and_exposes_raw_rehydrate_hint(self):
        conn = _connect(":memory:")
        _insert_observation(
            conn,
            {
                "kind": "tool.result",
                "summary": "git status compact sideband sample",
                "tool_name": "openclaw-mem.artifact.compact-receipt",
                "detail": {
                    "schema": "openclaw-mem.artifact.compaction-receipt.v1",
                    "tool": "rtk",
                    "command": "git status",
                    "rewrittenCommand": "rtk git status",
                    "rawArtifact": {
                        "handle": "ocm_artifact:v1:sha256:" + ("a" * 64),
                        "sha256": "a" * 64,
                        "bytes": 791,
                        "kind": "tool_output",
                    },
                    "compact": {
                        "text": "ok main",
                        "bytes": 7,
                    },
                },
            },
        )

        args = type(
            "Args",
            (),
            {
                "query": "git status compact",
                "query_en": None,
                "limit": 12,
                "budget_tokens": 400,
                "trace": True,
                "json": True,
                "trust_policy": "off",
                "use_graph": "off",
                "lifecycle_shadow": "on",
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_pack(conn, args)
        payload = json.loads(buf.getvalue())
        self.assertEqual(payload["items"][0]["summary"], "ok main")
        self.assertIn("compaction_sideband", payload)
        self.assertEqual(payload["compaction_sideband"]["mode"], "prefer_compact_fail_open")
        self.assertEqual(payload["compaction_sideband"]["selected"][0]["family"], "generic")
        self.assertEqual(payload["compaction_sideband"]["selected"][0]["rawArtifactHandle"], "ocm_artifact:v1:sha256:" + ("a" * 64))
        self.assertIn("raw artifact handle", payload["compaction_sideband"]["raw_rehydrate_hint"].lower())
        self.assertEqual(payload["compaction_policy_hints"]["mode"], "advisory_only")
        self.assertEqual(payload["compaction_policy_hints"]["family_counts"]["generic"], 1)
        self.assertIn("generic", payload["compaction_policy_hints"]["preferred_families"])
        self.assertEqual(payload["trace"]["extensions"]["compaction_sideband"]["selected_count"], 1)
        self.assertEqual(payload["trace"]["extensions"]["compaction_policy_hints"], payload["compaction_policy_hints"])
        self.assertEqual(payload["trace"]["extensions"]["compaction_sideband"]["selected"][0]["rewrittenCommand"], "rtk git status")
        self.assertTrue(any("compaction sideband" in note.lower() for note in payload["context_pack"]["notes"]["how_to_use"]))
        conn.close()

    def test_pack_compaction_sideband_classifies_families(self):
        conn = _connect(":memory:")
        for summary, command, expected, hex_char in [
            ("git diff compact", "git diff --stat", "git_diff", "a"),
            ("pytest compact", "pytest -q", "test_failures", "b"),
            ("logs compact", "docker logs api", "long_logs", "c"),
        ]:
            _insert_observation(
                conn,
                {
                    "kind": "tool.result",
                    "summary": summary,
                    "tool_name": "openclaw-mem.artifact.compact-receipt",
                    "detail": {
                        "schema": "openclaw-mem.artifact.compaction-receipt.v1",
                        "tool": "rtk",
                        "command": command,
                        "rawArtifact": {
                            "handle": f"ocm_artifact:v1:sha256:{hex_char * 64}",
                            "sha256": hex_char * 64,
                            "bytes": 120,
                            "kind": "tool_output",
                        },
                        "compact": {"text": f"compact {expected}", "bytes": len(f'compact {expected}'.encode('utf-8'))},
                    },
                },
            )

            args = type(
                "Args",
                (),
                {
                    "query": summary,
                    "query_en": None,
                    "limit": 4,
                    "budget_tokens": 200,
                    "trace": True,
                    "json": True,
                    "trust_policy": "off",
                    "use_graph": "off",
                    "lifecycle_shadow": "on",
                },
            )()
            buf = io.StringIO()
            with redirect_stdout(buf):
                cmd_pack(conn, args)
            payload = json.loads(buf.getvalue())
            self.assertEqual(payload["compaction_sideband"]["selected"][0]["family"], expected)
            self.assertEqual(payload["compaction_policy_hints"]["family_counts"][expected], 1)
            self.assertIn(expected, payload["compaction_policy_hints"]["preferred_families"])
        conn.close()

    def test_normalize_importance_scorer_value_accepts_common_aliases(self):
        self.assertEqual(_normalize_importance_scorer_value("heuristic_v1"), "heuristic-v1")
        self.assertEqual(_normalize_importance_scorer_value("heuristic_v2"), "heuristic-v2")
        self.assertEqual(_normalize_importance_scorer_value("Heuristic v1"), "heuristic-v1")
        self.assertEqual(_normalize_importance_scorer_value(" heuristic-v1 "), "heuristic-v1")
        self.assertEqual(_normalize_importance_scorer_value("heuristic-v2"), "heuristic-v2")
        self.assertEqual(_normalize_importance_scorer_value("Heuristic v2"), "heuristic-v2")
        self.assertEqual(_normalize_importance_scorer_value("off"), "off")
        self.assertEqual(_normalize_importance_scorer_value(""), "")

    def test_summary_has_task_marker_accepts_full_width_colon_and_unicode_dashes(self):
        self.assertTrue(_summary_has_task_marker("reminder：pay rent"))
        self.assertTrue(_summary_has_task_marker("ＴＯＤＯ：繳房租"))
        self.assertTrue(_summary_has_task_marker("Task－follow up on release checklist"))
        self.assertTrue(_summary_has_task_marker("Task–follow up on release checklist"))
        self.assertTrue(_summary_has_task_marker("Task—follow up on release checklist"))
        self.assertTrue(_summary_has_task_marker("Task−follow up on release checklist"))
        self.assertTrue(_summary_has_task_marker("TODO。rotate docs update"))
        self.assertTrue(_summary_has_task_marker("TODO; rotate runbook"))
        self.assertTrue(_summary_has_task_marker("TASK；follow up on release checklist"))

    def test_summary_has_task_marker_accepts_lowercase_marker_with_separator(self):
        self.assertTrue(_summary_has_task_marker("todo buy milk"))
        self.assertTrue(_summary_has_task_marker("task - clean desk"))
        self.assertTrue(_summary_has_task_marker("task\tclean desk"))

    def test_summary_has_task_marker_accepts_example_formats(self):
        self.assertTrue(_summary_has_task_marker("TODO: rotate runbook"))
        self.assertTrue(_summary_has_task_marker("task- check alerts"))
        self.assertTrue(_summary_has_task_marker("TODO. rotate runbook"))
        self.assertTrue(_summary_has_task_marker("(TASK): review PR"))
        self.assertTrue(_summary_has_task_marker("- [ ] TODO file patch"))

    def test_summary_has_task_marker_accepts_marker_only_forms(self):
        self.assertTrue(_summary_has_task_marker("TODO"))
        self.assertTrue(_summary_has_task_marker("[TASK]"))
        self.assertTrue(_summary_has_task_marker("(REMINDER)"))

    def test_summary_has_task_marker_accepts_bracket_wrapped_marker(self):
        self.assertTrue(_summary_has_task_marker("[TODO] buy milk"))
        self.assertTrue(_summary_has_task_marker("(task): clean desk"))
        self.assertTrue(_summary_has_task_marker("（ＲＥＭＩＮＤＥＲ） 續約網域"))
        self.assertTrue(_summary_has_task_marker("（TASK）rotate runbook"))
        self.assertTrue(_summary_has_task_marker("【ＴＡＳＫ】 續約網域"))
        self.assertTrue(_summary_has_task_marker("【task】 rotate runbook"))
        self.assertTrue(_summary_has_task_marker("【TODO】"))
        self.assertTrue(_summary_has_task_marker("【TODO】: clean desk"))
        self.assertTrue(_summary_has_task_marker("〔TODO〕: plan rollout"))
        self.assertTrue(_summary_has_task_marker("〔TASK〕 rotate notes"))
        self.assertTrue(_summary_has_task_marker("〔REMINDER〕: sync notes"))
        self.assertTrue(_summary_has_task_marker("{TODO} rotate runbook"))
        self.assertTrue(_summary_has_task_marker("{TASK}: renew reminders"))
        self.assertTrue(_summary_has_task_marker("《task》 clean notes"))
        self.assertTrue(_summary_has_task_marker("＜task＞ clean notes"))
        self.assertTrue(_summary_has_task_marker("〖task〗 clean notes"))
        self.assertTrue(_summary_has_task_marker("〘task〙 clean notes"))
        self.assertTrue(_summary_has_task_marker("「TODO」 rotate runbook"))
        self.assertTrue(_summary_has_task_marker("《TASK》 renew reminders"))
        self.assertTrue(_summary_has_task_marker("『task』renew reminders"))

    def test_summary_has_task_marker_accepts_list_and_checkbox_prefixes(self):
        self.assertTrue(_summary_has_task_marker("- TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("* [ ] TASK: clean desk"))
        self.assertTrue(_summary_has_task_marker("+ TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("> TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("> > [ ] TASK: clean desk"))
        self.assertTrue(_summary_has_task_marker(">> TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("• [x] [REMINDER] renew domain"))
        self.assertTrue(_summary_has_task_marker("‣ TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("▪ TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("◦ TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("● TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("○ TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("∙ [ ] TASK: clean desk"))
        self.assertTrue(_summary_has_task_marker("· [x] [REMINDER] renew domain"))
        self.assertTrue(_summary_has_task_marker("- [✓] TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("- [☑] TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("- [☐] TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("- [☒] TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("- [✅] TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("1. [✔] TASK: clean desk"))
        self.assertTrue(_summary_has_task_marker("1. TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("2) [ ] TASK: clean desk"))
        self.assertTrue(_summary_has_task_marker("(3) TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("a) TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("(a) TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("（ａ） TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("（1） TODO clean desk"))
        self.assertTrue(_summary_has_task_marker("B. [ ] TASK: clean desk"))
        self.assertTrue(_summary_has_task_marker("iv) TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("IX. [ ] TASK: clean desk"))
        self.assertTrue(_summary_has_task_marker("(iv) TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("– TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("— TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("− TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("・ TODO buy milk"))
        self.assertTrue(_summary_has_task_marker("1- TODO rotate runbook"))
        self.assertTrue(_summary_has_task_marker("1-TODO rotate runbook"))
        self.assertTrue(_summary_has_task_marker("iv- TODO rotate runbook"))
        self.assertTrue(_summary_has_task_marker("iv- TASK: clean desk"))

    def test_summary_has_task_marker_accepts_compact_wrapper_chaining(self):
        self.assertTrue(_summary_has_task_marker("•[x]TODO fix pipeline"))
        self.assertTrue(_summary_has_task_marker("‣[x]TODO clean pipeline"))
        self.assertTrue(_summary_has_task_marker("▪[x]TODO clean pipeline"))
        self.assertTrue(_summary_has_task_marker("∙[x]TODO clean pipeline"))
        self.assertTrue(_summary_has_task_marker("∙TODO compact pipeline"))
        self.assertTrue(_summary_has_task_marker("·[ ]TASK sync branch"))
        self.assertTrue(_summary_has_task_marker("◦[x]TODO compact pipeline"))
        self.assertTrue(_summary_has_task_marker("●[x]TODO compact pipeline"))
        self.assertTrue(_summary_has_task_marker("○[x]TODO compact pipeline"))
        self.assertTrue(_summary_has_task_marker("- (I)[ ] TODO reorder docs"))
        self.assertTrue(_summary_has_task_marker(">>‣TODO audit logs"))
        self.assertTrue(_summary_has_task_marker("—TODO audit logs"))
        self.assertTrue(_summary_has_task_marker("–TODO audit logs"))
        self.assertTrue(_summary_has_task_marker("・TODO audit logs"))

    def test_summary_has_task_marker_accepts_nested_prefix_combinations(self):
        self.assertTrue(_summary_has_task_marker("* (1) [ ] TODO: clean desk"))
        self.assertTrue(_summary_has_task_marker("• （２） [x] [TASK] rotate notes"))
        self.assertTrue(_summary_has_task_marker("> - a) [ ] TODO: clean desk"))
        self.assertTrue(_summary_has_task_marker("- > (iv) [ ] TODO: clean desk"))
        self.assertTrue(_summary_has_task_marker("- a) [ ] TODO: clean desk"))
        self.assertTrue(_summary_has_task_marker("- (iv) [ ] TODO: clean desk"))
        self.assertTrue(_summary_has_task_marker(">> [x]TODO: compact wrappers"))
        self.assertTrue(_summary_has_task_marker(">>[x]TODO: compact wrappers"))
        self.assertTrue(_summary_has_task_marker("-1)TODO compact ordered marker"))

    def test_summary_has_task_marker_rejects_non_marker_prefixes_and_requires_plain_marker_boundaries(self):
        self.assertFalse(_summary_has_task_marker("TODOLIST clean old notes"))
        self.assertFalse(_summary_has_task_marker("taskforce sync tomorrow"))
        self.assertFalse(_summary_has_task_marker("[TODOLIST] clean old notes"))
        self.assertFalse(_summary_has_task_marker("「TODOLIST」 clean old notes"))
        self.assertFalse(_summary_has_task_marker("《TODOLIST》 clean old notes"))
        self.assertFalse(_summary_has_task_marker("«TODOLIST» clean old notes"))
        self.assertFalse(_summary_has_task_marker("〈TODOLIST〉 clean old notes"))
        self.assertFalse(_summary_has_task_marker("‹TODOLIST› clean old notes"))
        self.assertFalse(_summary_has_task_marker("＜TODOLIST＞ clean old notes"))
        self.assertFalse(_summary_has_task_marker("<TODOLIST> clean old notes"))
        self.assertFalse(_summary_has_task_marker("〘TODOLIST〙 clean old notes"))
        self.assertTrue(_summary_has_task_marker("[TODO]clean old notes"))
        self.assertTrue(_summary_has_task_marker("「TODO」clean old notes"))
        self.assertTrue(_summary_has_task_marker("『TODO』clean old notes"))
        self.assertTrue(_summary_has_task_marker("《TODO》clean old notes"))
        self.assertTrue(_summary_has_task_marker("«TODO»clean old notes"))
        self.assertTrue(_summary_has_task_marker("〈TODO〉clean old notes"))
        self.assertTrue(_summary_has_task_marker("‹TODO›clean old notes"))
        self.assertTrue(_summary_has_task_marker("＜TODO＞clean old notes"))
        self.assertTrue(_summary_has_task_marker("<TODO>clean old notes"))
        self.assertTrue(_summary_has_task_marker("〖TODO〗clean old notes"))
        self.assertTrue(_summary_has_task_marker("〘TODO〙clean old notes"))
        self.assertTrue(_summary_has_task_marker("【TODO】clean old notes"))
        self.assertTrue(_summary_has_task_marker("〔TODO〕clean old notes"))
        self.assertTrue(_summary_has_task_marker("{TODO}clean old notes"))
        self.assertTrue(_summary_has_task_marker("｛TODO｝clean old notes"))
        self.assertTrue(_summary_has_task_marker("［TODO］clean old notes"))
        self.assertTrue(_summary_has_task_marker("-TODO clean old notes"))
        self.assertTrue(_summary_has_task_marker("+TODO clean old notes"))
        self.assertTrue(_summary_has_task_marker(">TODO clean old notes"))
        self.assertTrue(_summary_has_task_marker(">>TODO clean old notes"))
        self.assertTrue(_summary_has_task_marker(">>>TODO clean old notes"))
        self.assertTrue(_summary_has_task_marker("‣TODO clean old notes"))
        self.assertTrue(_summary_has_task_marker("▪TODO clean old notes"))
        self.assertTrue(_summary_has_task_marker("·TODO clean old notes"))
        self.assertTrue(_summary_has_task_marker("[x]TODO clean old notes"))
        self.assertTrue(_summary_has_task_marker("[✓]TODO clean old notes"))
        self.assertTrue(_summary_has_task_marker("[☒]TODO clean old notes"))
        self.assertTrue(_summary_has_task_marker("[✅]TODO clean old notes"))
        self.assertTrue(_summary_has_task_marker("(TASK)clean old notes"))
        self.assertTrue(_summary_has_task_marker("（TASK）clean old notes"))
        self.assertTrue(_summary_has_task_marker("〔REMINDER〕clean old notes"))
        self.assertTrue(_summary_has_task_marker("1.TODO clean old notes"))
        self.assertTrue(_summary_has_task_marker("1)TODO clean old notes"))
        self.assertTrue(_summary_has_task_marker("(1)TODO clean old notes"))
        self.assertTrue(_summary_has_task_marker("a)TODO clean old notes"))
        self.assertFalse(_summary_has_task_marker("ab) TODO clean old notes"))
        self.assertFalse(_summary_has_task_marker("(ab) TODO clean old notes"))
        self.assertFalse(_summary_has_task_marker("中) TODO clean old notes"))
        self.assertFalse(_summary_has_task_marker("(中) TODO clean old notes"))
        self.assertFalse(_summary_has_task_marker("é) TODO clean old notes"))
        self.assertFalse(_summary_has_task_marker("(é) TODO clean old notes"))
        self.assertFalse(_summary_has_task_marker("in) TODO clean old notes"))
        self.assertFalse(_summary_has_task_marker("ic) TODO clean old notes"))
        self.assertFalse(_summary_has_task_marker("(iiv) TODO clean old notes"))
        self.assertTrue(_summary_has_task_marker("* (1)TODO clean old notes"))

    def test_summary_has_task_marker_rejects_inline_markers_and_mismatched_wrappers(self):
        self.assertFalse(_summary_has_task_marker("We should TODO: fix this later"))
        self.assertFalse(_summary_has_task_marker("Meeting notes - TODO: fix this later"))
        self.assertFalse(_summary_has_task_marker("I think [TODO] fix this later"))
        self.assertFalse(_summary_has_task_marker("TODO! fix this later"))
        self.assertFalse(_summary_has_task_marker("[TODO) fix this later"))
        self.assertFalse(_summary_has_task_marker("TODOs: plural shouldn't match"))

    def test_parser_accepts_doctor_command(self):
        args = build_parser().parse_args(["doctor", "--json"])
        self.assertEqual(args.cmd, "doctor")
        self.assertTrue(args.json)

    def test_parser_merges_global_and_command_json_flags(self):
        before_args = build_parser().parse_args(["--json", "status"])
        after_args = build_parser().parse_args(["status", "--json"])

        self.assertTrue(before_args.json_global)
        self.assertFalse(before_args.json)

        self.assertTrue(after_args.json)
        self.assertFalse(after_args.json_global)

        merged_before = bool(before_args.json or before_args.json_global)
        merged_after = bool(after_args.json or after_args.json_global)

        self.assertTrue(merged_before)
        self.assertTrue(merged_after)

    def test_graph_parser_can_parse_subcommands(self):
        a = build_parser().parse_args(["graph", "index", "hello"])
        self.assertEqual(a.cmd, "graph")
        self.assertEqual(a.graph_cmd, "index")
        self.assertEqual(a.query, "hello")

        a = build_parser().parse_args(["graph", "pack", "obs:1"])
        self.assertEqual(a.cmd, "graph")
        self.assertEqual(a.graph_cmd, "pack")
        self.assertEqual(a.ids, ["obs:1"])

        a = build_parser().parse_args(["graph", "preflight", "hello"])
        self.assertEqual(a.cmd, "graph")
        self.assertEqual(a.graph_cmd, "preflight")
        self.assertEqual(a.query, "hello")

        a = build_parser().parse_args(["graph", "auto-status"])
        self.assertEqual(a.cmd, "graph")
        self.assertEqual(a.graph_cmd, "auto-status")

        a = build_parser().parse_args(["graph", "topology-refresh", "--file", "/tmp/topology.json"])
        self.assertEqual(a.cmd, "graph")
        self.assertEqual(a.graph_cmd, "topology-refresh")
        self.assertEqual(a.file, "/tmp/topology.json")

        a = build_parser().parse_args(["graph", "capture-git", "--repo", "/tmp/repo"])
        self.assertEqual(a.cmd, "graph")
        self.assertEqual(a.graph_cmd, "capture-git")
        self.assertEqual(a.repo, ["/tmp/repo"])

        a = build_parser().parse_args(["graph", "capture-git", "--repo", "/tmp/repo", "--importance-scorer", "heuristic-v1"])
        self.assertEqual(a.graph_cmd, "capture-git")
        self.assertEqual(a.importance_scorer, "heuristic-v1")

        a = build_parser().parse_args(["graph", "capture-md", "--path", "/tmp/notes"])
        self.assertEqual(a.cmd, "graph")
        self.assertEqual(a.graph_cmd, "capture-md")
        self.assertEqual(a.path, ["/tmp/notes"])

        a = build_parser().parse_args(["graph", "capture-md", "--path", "/tmp/notes", "--importance-scorer", "heuristic-v1"])
        self.assertEqual(a.graph_cmd, "capture-md")
        self.assertEqual(a.importance_scorer, "heuristic-v1")

        a = build_parser().parse_args(["graph", "export", "--query", "hello", "--to", "/tmp/graph.json"])
        self.assertEqual(a.cmd, "graph")
        self.assertEqual(a.graph_cmd, "export")
        self.assertEqual(a.query, "hello")

        a = build_parser().parse_args(["graph", "query", "upstream", "artifact.daily-mission"])
        self.assertEqual(a.cmd, "graph")
        self.assertEqual(a.graph_cmd, "query")
        self.assertEqual(a.graph_query_cmd, "upstream")
        self.assertEqual(a.node_id, "artifact.daily-mission")

        a = build_parser().parse_args([
            "graph",
            "query",
            "subgraph",
            "artifact.daily-mission",
            "--require-structured-provenance",
        ])
        self.assertEqual(a.graph_query_cmd, "subgraph")
        self.assertTrue(a.require_structured_provenance)

        a = build_parser().parse_args(
            [
                "pack",
                "--query",
                "provenance gated pack",
                "--use-graph",
                "on",
                "--graph-query-db",
                "/tmp/graph.db",
                "--graph-provenance-policy",
                "structured_only_fail_open",
                "--no-graph-require-structured-provenance",
            ]
        )
        self.assertEqual(a.graph_provenance_policy, "structured_only_fail_open")
        self.assertEqual(a.graph_query_db, "/tmp/graph.db")
        self.assertFalse(a.graph_require_structured_provenance)
        self.assertEqual(a.pack_trust_policy, "off")
        self.assertEqual(a.pack_lifecycle_shadow, "on")

        a = build_parser().parse_args(["pack", "--query", "trust", "--pack-trust-policy", "exclude_quarantined_fail_open"])
        self.assertEqual(a.pack_trust_policy, "exclude_quarantined_fail_open")

        a = build_parser().parse_args(["pack", "--query", "trust", "--pack-lifecycle-shadow", "off", "--pack-lifecycle-log-max-rows", "9"])
        self.assertEqual(a.pack_lifecycle_shadow, "off")
        self.assertEqual(a.pack_lifecycle_log_max_rows, 9)


    def test_writeback_lancedb_parser_accepts_flags(self):
        a = build_parser().parse_args([
            "writeback-lancedb",
            "--db",
            "/tmp/sidecar.sqlite",
            "--lancedb",
            "/tmp/lancedb",
            "--table",
            "memories",
            "--limit",
            "11",
            "--batch",
            "7",
            "--dry-run",
            "--force",
            "--force-fields",
            "importance,trust_tier,category",
        ])

        self.assertEqual(a.cmd, "writeback-lancedb")
        self.assertEqual(a.db, "/tmp/sidecar.sqlite")
        self.assertEqual(a.lancedb, "/tmp/lancedb")
        self.assertEqual(a.table, "memories")
        self.assertEqual(a.limit, 11)
        self.assertEqual(a.batch, 7)
        self.assertTrue(a.dry_run)
        self.assertTrue(a.force)
        self.assertEqual(a.force_fields, "importance,trust_tier,category")

    def test_graph_index_and_pack_smoke_budgeted(self):
        conn = _connect(":memory:")

        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": "2026-02-04T13:00:00Z",
                        "kind": "tool",
                        "tool_name": "exec",
                        "summary": "alpha one",
                        "detail": {},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:01:00Z",
                        "kind": "tool",
                        "tool_name": "read",
                        "summary": "beta two",
                        "detail": {},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:02:00Z",
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "alpha three",
                        "detail": {},
                    }
                ),
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        # graph index
        args = type(
            "Args",
            (),
            {
                "query": "alpha",
                "scope": None,
                "limit": 8,
                "window": 1,
                "suggest_limit": 3,
                "budget_tokens": 20,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_graph_index(conn, args)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["kind"], "openclaw-mem.graph.index.v0")
        self.assertIn("index_text", out)
        self.assertLessEqual(out["budget"]["estimatedTokens"], out["budget"]["budgetTokens"])

        # graph pack
        args = type(
            "Args",
            (),
            {
                "ids": ["obs:1", "2"],
                "max_items": 20,
                "budget_tokens": 30,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_graph_pack(conn, args)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["kind"], "openclaw-mem.graph.pack.v0")
        self.assertIn("selection", out)
        self.assertIn("bundle_text", out)
        self.assertIn("obs:1", out["bundle_text"])

        conn.close()

    def test_graph_preflight_smoke_selects_refs_and_respects_budget(self):
        conn = _connect(":memory:")

        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": "2026-02-04T13:00:00Z",
                        "kind": "tool",
                        "tool_name": "exec",
                        "summary": "alpha one",
                        "detail": {},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:01:00Z",
                        "kind": "tool",
                        "tool_name": "read",
                        "summary": "alpha two",
                        "detail": {},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:02:00Z",
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "beta three",
                        "detail": {},
                    }
                ),
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        args = type(
            "Args",
            (),
            {
                "query": "alpha",
                "scope": None,
                "limit": 8,
                "window": 1,
                "suggest_limit": 3,
                "budget_tokens": 25,
                "take": 12,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_graph_preflight(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["kind"], "openclaw-mem.graph.preflight.v0")
        self.assertIn("bundle_text", out)
        self.assertIn("obs:", out["bundle_text"])
        self.assertLessEqual(out["budget"]["estimatedTokens"], out["budget"]["budgetTokens"])
        self.assertGreaterEqual(len(out["selection"]["recordRefs"]), 1)

        conn.close()

    def test_graph_scope_filter_isolation_for_index_and_preflight(self):
        conn = _connect(":memory:")

        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": "2026-02-04T13:00:00Z",
                        "kind": "tool",
                        "tool_name": "exec",
                        "summary": "alpha one",
                        "detail": {"scope": "proj-a"},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:01:00Z",
                        "kind": "tool",
                        "tool_name": "read",
                        "summary": "alpha two",
                        "detail": {"scope": "proj-b"},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:02:00Z",
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "alpha three",
                        "detail": {"scope": "proj-a"},
                    }
                ),
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        index_args = type(
            "Args",
            (),
            {
                "query": "alpha",
                "scope": "proj-a",
                "limit": 10,
                "window": 1,
                "suggest_limit": 10,
                "budget_tokens": 120,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_graph_index(conn, index_args)
        index_out = json.loads(buf.getvalue())

        top_refs = [item["recordRef"] for item in index_out["top_candidates"]]
        self.assertIn("obs:1", top_refs)
        self.assertIn("obs:3", top_refs)
        self.assertNotIn("obs:2", top_refs)
        self.assertNotIn("obs:2", index_out["index_text"])

        preflight_args = type(
            "Args",
            (),
            {
                "query": "alpha",
                "scope": "proj-a",
                "limit": 10,
                "window": 1,
                "suggest_limit": 10,
                "budget_tokens": 120,
                "take": 10,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_graph_preflight(conn, preflight_args)
        preflight_out = json.loads(buf.getvalue())

        selected_refs = preflight_out["selection"]["recordRefs"]
        self.assertIn("obs:1", selected_refs)
        self.assertIn("obs:3", selected_refs)
        self.assertNotIn("obs:2", selected_refs)
        self.assertNotIn("obs:2", preflight_out["bundle_text"])

        conn.close()

    def test_pack_graph_resolve_scope_prefers_explicit_scope(self):
        conn = _connect(":memory:")
        try:
            out = _pack_graph_resolve_scope(conn, query="docs/specs/", explicit_scope="proj-a", use_graph="auto")
        finally:
            conn.close()
        self.assertEqual(out["resolved_scope"], "proj-a")
        self.assertEqual(out["scope_source"], "explicit")
        self.assertEqual(out["scope_decision"], "allow")

    def test_pack_graph_resolve_scope_infers_local_scope_from_query_token(self):
        conn = _connect(":memory:")

        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": "2026-02-04T13:00:00Z",
                        "kind": "tool",
                        "tool_name": "exec",
                        "summary": "alpha docs spec one",
                        "detail": {"scope": "proj-a"},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:01:00Z",
                        "kind": "tool",
                        "tool_name": "read",
                        "summary": "alpha docs spec two",
                        "detail": {"scope": "proj-b"},
                    }
                ),
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        try:
            out = _pack_graph_resolve_scope(conn, query="show proj-a docs spec status", explicit_scope=None, use_graph="auto")
        finally:
            conn.close()

        self.assertEqual(out["resolved_scope"], "proj-a")
        self.assertEqual(out["scope_source"], "inferred")
        self.assertEqual(out["scope_decision"], "allow")

    def test_pack_graph_auto_unresolved_scope_skips_preflight(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {
                1: {
                    "summary": "plain summary",
                    "summary_en": "plain summary",
                    "kind": "fact",
                    "lang": "en",
                }
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args(["pack", "--query", "latest dependency status", "--json", "--trace", "--use-graph", "auto"])

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._pack_graph_known_scopes", return_value=[]
        ):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertFalse(out["graph"]["triggered"])
        self.assertEqual(out["graph"]["trigger_reason"], "scope_unresolved")
        self.assertEqual(out["trace"]["extensions"]["graph"]["scope_decision"], "skip")
        conn.close()

    def test_graph_pack_prefers_fresh_synthesis_cards_over_raw_refs(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {"kind": "tool", "summary": "alpha source one", "tool_name": "read", "detail": {}})
        _insert_observation(conn, {"kind": "tool", "summary": "alpha source two", "tool_name": "exec", "detail": {}})

        compile_args = build_parser().parse_args([
            "graph", "--json", "synth", "compile",
            "--record-ref", "obs:1",
            "--record-ref", "obs:2",
            "--title", "Merged insight",
            "--summary", "Merged insight",
            "--why-it-matters", "Prefer the synthesis card",
        ])
        with redirect_stdout(io.StringIO()):
            compile_args.func(conn, compile_args)

        args = type(
            "Args",
            (),
            {
                "ids": ["obs:1", "obs:2"],
                "max_items": 20,
                "budget_tokens": 120,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_graph_pack(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["selection"]["recordRefs"], ["obs:3"])
        self.assertEqual(out["selection"]["preferredCardRefs"], ["obs:3"])
        self.assertIn("obs:1", out["selection"]["coveredRawRefs"])
        self.assertIn("obs:2", out["selection"]["coveredRawRefs"])
        self.assertIn("sources=2", out["bundle_text"])
        self.assertIn("why=Prefer the synthesis card", out["bundle_text"])
        conn.close()

    def test_graph_preflight_prefers_fresh_synthesis_cards_over_raw_refs(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {"kind": "tool", "summary": "alpha source one", "tool_name": "read", "detail": {}})
        _insert_observation(conn, {"kind": "tool", "summary": "alpha source two", "tool_name": "exec", "detail": {}})

        compile_args = build_parser().parse_args([
            "graph", "--json", "synth", "compile",
            "--record-ref", "obs:1",
            "--record-ref", "obs:2",
            "--title", "Merged insight",
            "--summary", "Merged insight",
            "--why-it-matters", "Prefer the synthesis card",
        ])
        with redirect_stdout(io.StringIO()):
            compile_args.func(conn, compile_args)

        args = type(
            "Args",
            (),
            {
                "query": "alpha",
                "scope": None,
                "limit": 8,
                "window": 1,
                "suggest_limit": 3,
                "budget_tokens": 120,
                "take": 12,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_graph_preflight(conn, args)

        out = json.loads(buf.getvalue())
        self.assertIn("obs:3", out["selection"]["recordRefs"])
        self.assertIn("obs:3", out["selection"]["preferredCardRefs"])
        self.assertIn("obs:1", out["selection"]["coveredRawRefs"])
        self.assertIn("obs:2", out["selection"]["coveredRawRefs"])
        self.assertIn("sources=2", out["bundle_text"])
        self.assertIn("why=Prefer the synthesis card", out["bundle_text"])
        conn.close()

    def test_graph_pack_prefers_narrower_synthesis_card_then_record_ref_for_ties(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {"kind": "tool", "summary": "alpha source one", "tool_name": "read", "detail": {}})
        _insert_observation(conn, {"kind": "tool", "summary": "alpha source two", "tool_name": "exec", "detail": {}})
        _insert_observation(conn, {"kind": "tool", "summary": "alpha source three", "tool_name": "write", "detail": {}})

        compile_three = build_parser().parse_args([
            "graph", "--json", "synth", "compile",
            "--record-ref", "obs:1",
            "--record-ref", "obs:2",
            "--record-ref", "obs:3",
            "--title", "Broader card",
            "--summary", "Broader card",
        ])
        with redirect_stdout(io.StringIO()):
            compile_three.func(conn, compile_three)

        compile_narrow_a = build_parser().parse_args([
            "graph", "--json", "synth", "compile",
            "--record-ref", "obs:1",
            "--record-ref", "obs:2",
            "--title", "Narrow card A",
            "--summary", "Narrow card A",
        ])
        with redirect_stdout(io.StringIO()):
            compile_narrow_a.func(conn, compile_narrow_a)

        compile_narrow_b = build_parser().parse_args([
            "graph", "--json", "synth", "compile",
            "--record-ref", "obs:1",
            "--record-ref", "obs:2",
            "--title", "Narrow card B",
            "--summary", "Narrow card B",
        ])
        with redirect_stdout(io.StringIO()):
            compile_narrow_b.func(conn, compile_narrow_b)

        args = type(
            "Args",
            (),
            {
                "ids": ["obs:1", "obs:2"],
                "max_items": 20,
                "budget_tokens": 120,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_graph_pack(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["selection"]["recordRefs"], ["obs:5"])
        self.assertEqual(out["selection"]["preferredCardRefs"], ["obs:5"])
        self.assertEqual(out["selection"]["coveredRawRefs"], ["obs:1", "obs:2"])
        self.assertIn("Narrow card A", out["bundle_text"])
        self.assertNotIn("Broader card", out["bundle_text"])
        self.assertNotIn("Narrow card B", out["bundle_text"])
        conn.close()

    def test_graph_auto_status_reports_flags_and_invalid_values(self):
        from unittest.mock import patch

        conn = _connect(":memory:")

        args = type("Args", (), {"json": True})()
        with patch.dict(
            "os.environ",
            {
                "OPENCLAW_MEM_GRAPH_AUTO_RECALL": "1",
                "OPENCLAW_MEM_GRAPH_AUTO_CAPTURE": "off",
                "OPENCLAW_MEM_GRAPH_AUTO_CAPTURE_MD": "maybe",
            },
            clear=False,
        ):
            buf = io.StringIO()
            with redirect_stdout(buf):
                cmd_graph_auto_status(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["kind"], "openclaw-mem.graph.auto-status.v0")

        recall = out["flags"]["OPENCLAW_MEM_GRAPH_AUTO_RECALL"]
        self.assertTrue(recall["enabled"])
        self.assertTrue(recall["valid"])
        self.assertEqual(recall["reason"], "parsed_truthy")

        capture = out["flags"]["OPENCLAW_MEM_GRAPH_AUTO_CAPTURE"]
        self.assertFalse(capture["enabled"])
        self.assertTrue(capture["valid"])
        self.assertEqual(capture["reason"], "parsed_falsy")

        capture_md = out["flags"]["OPENCLAW_MEM_GRAPH_AUTO_CAPTURE_MD"]
        self.assertFalse(capture_md["enabled"])
        self.assertFalse(capture_md["valid"])
        self.assertEqual(capture_md["reason"], "invalid_fallback_default")

        conn.close()

    def test_graph_auto_status_reports_unset_default_reason(self):
        from unittest.mock import patch

        conn = _connect(":memory:")

        args = type("Args", (), {"json": True})()
        with patch.dict("os.environ", {}, clear=True):
            buf = io.StringIO()
            with redirect_stdout(buf):
                cmd_graph_auto_status(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["kind"], "openclaw-mem.graph.auto-status.v0")

        for st in out["flags"].values():
            self.assertFalse(st["present"])
            self.assertFalse(st["enabled"])
            self.assertTrue(st["valid"])
            self.assertEqual(st["reason"], "unset_default")

        conn.close()

    def test_graph_capture_git_errors_cleanly_when_repo_missing(self):
        import tempfile

        conn = _connect(":memory:")
        missing_repo = "/tmp/openclaw-mem-not-a-repo"

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "repo": [missing_repo],
                "since": 24,
                "state": state_path,
                "max_commits": 10,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_graph_capture_git(conn, args)

        self.assertEqual(cm.exception.code, 1)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["totals"]["errors"], 1)
        self.assertEqual(out["repos"][0]["inserted"], 0)

        conn.close()

    def test_graph_capture_md_smoke_is_index_only_and_idempotent(self):
        import os
        import tempfile
        import time

        conn = _connect(":memory:")

        with tempfile.TemporaryDirectory() as td:
            md_path = os.path.join(td, "notes.md")
            state_path = os.path.join(td, "graph-capture-md-state.json")

            with open(md_path, "w", encoding="utf-8") as f:
                f.write(
                    "# Title\n\n"
                    "## Alpha\n"
                    "first line\n"
                    "```\n"
                    "## hidden\n"
                    "```\n\n"
                    "### Beta\n"
                    "second line\n"
                )

            args = type(
                "Args",
                (),
                {
                    "path": [td],
                    "include": [".md"],
                    "exclude_glob": ["**/node_modules/**", "**/.venv/**", "**/.git/**", "**/dist/**"],
                    "max_files": 200,
                    "max_sections_per_file": 50,
                    "min_heading_level": 2,
                    "state": state_path,
                    "since_hours": 24,
                    "importance_scorer": None,
                    "json": True,
                },
            )()

            buf1 = io.StringIO()
            with redirect_stdout(buf1):
                cmd_graph_capture_md(conn, args)

            out1 = json.loads(buf1.getvalue())
            self.assertEqual(out1["inserted"], 2)
            self.assertEqual(out1["skipped_existing"], 0)
            self.assertEqual(out1["errors"], 0)

            now = time.time() + 2.0
            os.utime(md_path, (now, now))

            buf2 = io.StringIO()
            with redirect_stdout(buf2):
                cmd_graph_capture_md(conn, args)

            out2 = json.loads(buf2.getvalue())
            self.assertEqual(out2["changed_files"], 1)
            self.assertEqual(out2["inserted"], 0)
            self.assertEqual(out2["skipped_existing"], 2)

            rows = conn.execute(
                "SELECT summary, detail_json FROM observations WHERE tool_name = 'graph.capture-md' ORDER BY id"
            ).fetchall()
            self.assertEqual(len(rows), 2)
            self.assertTrue(all(r["summary"].startswith("[MD] notes.md#") for r in rows))

            detail = json.loads(rows[0]["detail_json"])
            self.assertIn("source_path", detail)
            self.assertIn("heading", detail)
            self.assertIn("section_fingerprint", detail)
            self.assertNotIn("excerpt", detail)
            self.assertNotIn("content", detail)

        conn.close()

    def test_graph_capture_md_importance_scorer_override_controls_autograde(self):
        import os
        import tempfile
        from unittest.mock import patch

        conn = _connect(":memory:")

        with tempfile.TemporaryDirectory() as td:
            state_path = os.path.join(td, "graph-capture-md-state.json")
            md_on = os.path.join(td, "on.md")
            md_off = os.path.join(td, "off.md")

            with open(md_on, "w", encoding="utf-8") as f:
                f.write("# T\n\n## A\nline\n")
            with open(md_off, "w", encoding="utf-8") as f:
                f.write("# T\n\n## B\nline\n")

            with patch.dict("os.environ", {}, clear=False):
                os.environ.pop("OPENCLAW_MEM_IMPORTANCE_SCORER", None)
                args_on = type(
                    "Args",
                    (),
                    {
                        "path": [md_on],
                        "include": [".md"],
                        "exclude_glob": ["**/node_modules/**", "**/.venv/**", "**/.git/**", "**/dist/**"],
                        "max_files": 200,
                        "max_sections_per_file": 50,
                        "min_heading_level": 2,
                        "state": state_path,
                        "since_hours": 24,
                        "importance_scorer": "heuristic-v1",
                        "json": True,
                    },
                )()
                with redirect_stdout(io.StringIO()):
                    cmd_graph_capture_md(conn, args_on)

            row_on = conn.execute(
                "SELECT detail_json FROM observations WHERE tool_name='graph.capture-md' AND summary LIKE '%on.md%' LIMIT 1"
            ).fetchone()
            self.assertIsNotNone(row_on)
            detail_on = json.loads(row_on[0])
            self.assertIn("importance", detail_on)

            with patch.dict("os.environ", {"OPENCLAW_MEM_IMPORTANCE_SCORER": "heuristic-v1"}, clear=False):
                args_off = type(
                    "Args",
                    (),
                    {
                        "path": [md_off],
                        "include": [".md"],
                        "exclude_glob": ["**/node_modules/**", "**/.venv/**", "**/.git/**", "**/dist/**"],
                        "max_files": 200,
                        "max_sections_per_file": 50,
                        "min_heading_level": 2,
                        "state": state_path,
                        "since_hours": 24,
                        "importance_scorer": "off",
                        "json": True,
                    },
                )()
                with redirect_stdout(io.StringIO()):
                    cmd_graph_capture_md(conn, args_off)

            row_off = conn.execute(
                "SELECT detail_json FROM observations WHERE tool_name='graph.capture-md' AND summary LIKE '%off.md%' LIMIT 1"
            ).fetchone()
            self.assertIsNotNone(row_off)
            detail_off = json.loads(row_off[0])
            self.assertNotIn("importance", detail_off)

        conn.close()

    def test_profile_reports_counts_labels_and_recent(self):
        conn = _connect(":memory:")

        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": "2026-02-04T13:00:00Z",
                        "kind": "decision",
                        "tool_name": "memory_store",
                        "summary": "Pin model for dev lane",
                        "detail": {"importance": {"score": 0.92, "label": "must_remember"}},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:01:00Z",
                        "kind": "task",
                        "tool_name": "memory_store",
                        "summary": "TODO: run benchmark",
                        "detail": {"importance": 0.65},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:02:00Z",
                        "kind": "tool",
                        "tool_name": "exec",
                        "summary": "lint passed",
                        "detail": {},
                    }
                ),
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        from openclaw_mem.vector import pack_f32, l2_norm

        conn.execute(
            "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (1, "text-embedding-3-small", 2, pack_f32([1.0, 0.0]), l2_norm([1.0, 0.0]), "2026-02-05T00:00:00Z"),
        )
        conn.execute(
            "INSERT INTO observation_embeddings_en (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (2, "text-embedding-3-small", 2, pack_f32([0.0, 1.0]), l2_norm([0.0, 1.0]), "2026-02-05T00:00:00Z"),
        )
        conn.commit()

        args = type(
            "Args",
            (),
            {
                "db": ":memory:",
                "json": True,
                "recent_limit": 2,
                "tool_limit": 5,
                "kind_limit": 5,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_profile(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["observations"]["count"], 3)
        self.assertEqual(out["importance"]["present"], 2)
        self.assertEqual(out["importance"]["missing"], 1)
        self.assertEqual(out["importance"]["label_counts"]["must_remember"], 1)
        self.assertEqual(out["importance"]["label_counts"]["nice_to_have"], 1)
        self.assertEqual(out["importance"]["label_counts"]["unknown"], 1)
        self.assertEqual(out["embeddings"]["original"]["count"], 1)
        self.assertEqual(out["embeddings"]["english"]["count"], 1)
        self.assertLessEqual(len(out["recent"]), 2)
        self.assertEqual(out["recent"][0]["id"], 3)

        conn.close()

    def test_profile_counts_malformed_importance_as_unknown(self):
        conn = _connect(":memory:")

        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": "2026-02-04T13:00:00Z",
                        "kind": "decision",
                        "tool_name": "memory_store",
                        "summary": "Persist critical decision",
                        "detail": {"importance": {"label": "must remember"}},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:01:00Z",
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "Malformed importance payload",
                        "detail": {"importance": {"score": "high"}},
                    }
                ),
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        args = type(
            "Args",
            (),
            {
                "db": ":memory:",
                "json": True,
                "recent_limit": 5,
                "tool_limit": 5,
                "kind_limit": 5,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_profile(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["importance"]["present"], 2)
        self.assertEqual(out["importance"]["missing"], 0)
        self.assertEqual(out["importance"]["label_counts"]["must_remember"], 1)
        self.assertEqual(out["importance"]["label_counts"]["ignore"], 0)
        self.assertEqual(out["importance"]["label_counts"]["unknown"], 1)
        self.assertEqual(out["importance"]["avg_score"], 0.8)

        conn.close()

    def test_backend_reports_slot_and_readiness(self):
        conn = _connect(":memory:")

        fake_cfg = {
            "plugins": {
                "slots": {"memory": "memory-lancedb"},
                "entries": {
                    "memory-core": {"enabled": True},
                    "memory-lancedb": {
                        "enabled": True,
                        "config": {"embedding": {"apiKey": "${OPENAI_API_KEY}"}},
                    },
                    "openclaw-mem": {"enabled": True},
                },
            }
        }

        from unittest.mock import patch

        args = type("Args", (), {"json": True})()
        buf = io.StringIO()
        with patch("openclaw_mem.cli._read_openclaw_config", return_value=fake_cfg), patch.dict("os.environ", {"OPENAI_API_KEY": "test-key"}, clear=False):
            with redirect_stdout(buf):
                cmd_backend(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["memory_slot"], "memory-lancedb")
        self.assertTrue(out["entries"]["memory-lancedb"]["embedding_api_key_ready"])
        conn.close()

    def test_ingest_preserves_extra_fields_into_detail_json(self):
        conn = _connect(":memory:")

        sample = json.dumps(
            {
                "ts": "2026-02-04T13:00:00Z",
                "kind": "tool",
                "tool_name": "memory_recall",
                "summary": "Found memories",
                "detail": {"base": 1},
                "memory_backend": "memory-lancedb",
                "memory_backend_ready": True,
                "memory_operation": "recall",
            }
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        row = conn.execute("SELECT detail_json FROM observations WHERE id = 1").fetchone()
        detail = json.loads(row["detail_json"])
        self.assertEqual(detail["base"], 1)
        self.assertEqual(detail["memory_backend"], "memory-lancedb")
        self.assertEqual(detail["memory_operation"], "recall")
        conn.close()

    def test_ingest_search_timeline_get_json(self):
        conn = _connect(":memory:")

        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": "2026-02-04T13:00:00Z",
                        "kind": "tool",
                        "tool_name": "cron.list",
                        "summary": "cron list called",
                        "detail": {"ok": True},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:01:00Z",
                        "kind": "tool",
                        "tool_name": "gateway.config.get",
                        "summary": "read gateway config",
                        "detail": {"ok": True},
                    }
                ),
            ]
        )

        # Ingest via stdin
        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            buf = io.StringIO()
            with redirect_stdout(buf):
                cmd_ingest(conn, args)
            out = json.loads(buf.getvalue())
            self.assertEqual(out["inserted"], 2)
            self.assertTrue(out["ids"])
        finally:
            sys.stdin = old_stdin

        # Search
        args = type("Args", (), {"query": "cron", "limit": 20, "json": True})()
        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_search(conn, args)
        hits = json.loads(buf.getvalue())
        self.assertGreaterEqual(len(hits), 1)
        self.assertEqual(hits[0]["tool_name"], "cron.list")

        # Timeline around ID 1
        args = type("Args", (), {"ids": [1], "window": 1, "json": True})()
        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_timeline(conn, args)
        tl = json.loads(buf.getvalue())
        self.assertGreaterEqual(len(tl), 1)

        # Get
        args = type("Args", (), {"ids": [1, 2], "json": True})()
        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_get(conn, args)
        rows = json.loads(buf.getvalue())
        self.assertEqual([r["id"] for r in rows], [1, 2])

        conn.close()

    def test_search_prefers_fresh_synthesis_cards_in_results(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {"kind": "note", "summary": "alpha rollout note", "tool_name": "memory_store", "detail": {}})
        _insert_observation(conn, {"kind": "note", "summary": "alpha rollback note", "tool_name": "memory_store", "detail": {}})

        compile_args = build_parser().parse_args([
            "graph", "--json", "synth", "compile",
            "--record-ref", "obs:1",
            "--record-ref", "obs:2",
            "--title", "Merged insight",
            "--summary", "Merged insight",
            "--why-it-matters", "Prefer the synthesis card"
        ])
        with redirect_stdout(io.StringIO()):
            compile_args.func(conn, compile_args)

        args = type("Args", (), {"query": "alpha", "limit": 10, "json": True})()
        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_search(conn, args)

        hits = json.loads(buf.getvalue())
        self.assertGreaterEqual(len(hits), 1)
        self.assertEqual(hits[0]["tool_name"], "graph.synth-compile")
        self.assertEqual(hits[0]["graph_consumption"]["coveredRawRefs"], ["obs:1", "obs:2"])
        self.assertEqual(hits[0]["graph_consumption"]["coveredRanks"], [1, 2])
        self.assertEqual(hits[0]["match"], ["graph_synthesis"])
        conn.close()

    def test_search_suppresses_superseded_proposal_rows(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {
            "kind": "memory.proposal",
            "summary": "yijin-loop-engine alias dirty v3 contains question mark mojibake ???",
            "tool_name": "gateway.store.propose",
            "detail": {"provenance": {"seed_id": "dirty-v3"}},
        })
        _insert_observation(conn, {
            "kind": "memory.proposal",
            "summary": "yijin-loop-engine alias clean canonical memory v4",
            "tool_name": "gateway.store.propose",
            "detail": {"provenance": {"seed_id": "clean-v4", "supersedes": ["obs:1"]}},
        })

        args = type("Args", (), {"query": "yijin-loop-engine", "limit": 10, "json": True})()
        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_search(conn, args)

        hits = json.loads(buf.getvalue())
        self.assertEqual([h["id"] for h in hits], [2])
        self.assertIn("clean canonical memory v4", hits[0]["summary"])
        self.assertNotIn("???", hits[0]["summary"])
        conn.close()

    def test_search_supersede_filter_ignores_malformed_provenance(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {
            "kind": "memory.proposal",
            "summary": "yijin-loop-engine unrelated row remains",
            "tool_name": "gateway.store.propose",
            "detail": {"provenance": {"supersedes": [{"bad": "shape"}]}},
        })

        args = type("Args", (), {"query": "yijin-loop-engine", "limit": 10, "json": True})()
        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_search(conn, args)

        hits = json.loads(buf.getvalue())
        self.assertEqual(len(hits), 1)
        self.assertEqual(hits[0]["id"], 1)
        conn.close()

    def test_search_normalizes_hyphenated_terms_instead_of_raising(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {"kind": "note", "summary": "openclaw-mem gateway search cli_failed diagnosis", "tool_name": "gateway.store.propose", "detail": {}})

        args = type("Args", (), {"query": "openclaw-mem gateway search cli_failed", "limit": 10, "json": True})()
        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_search(conn, args)

        hits = json.loads(buf.getvalue())
        self.assertGreaterEqual(len(hits), 1)
        self.assertIn("openclaw-mem", hits[0]["summary"])
        conn.close()

    def test_cjk_search_fallback_when_fts_misses(self):
        conn = _connect(":memory:")

        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": "2026-02-04T13:00:00Z",
                        "kind": "tool",
                        "tool_name": "memorybench",
                        "summary": "我今天在台北開產品會議，晚上再整理筆記。",
                        "detail": {"session_id": "s-zh-1"},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:01:00Z",
                        "kind": "tool",
                        "tool_name": "memorybench",
                        "summary": "I booked a train ticket to Taichung for next week.",
                        "detail": {"session_id": "s-en-1"},
                    }
                ),
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        # This query is semantically related but not an exact phrase; CJK fallback should recover it.
        args = type("Args", (), {"query": "今天會議在什麼城市", "limit": 10, "json": True})()
        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_search(conn, args)

        hits = json.loads(buf.getvalue())
        self.assertGreaterEqual(len(hits), 1)
        self.assertIn("台北", hits[0]["summary"])
        conn.close()

    def test_triage_exit_code_and_json(self):
        conn = _connect(":memory:")

        # Ingest one normal and one error-ish observation
        from datetime import datetime, timezone

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps({"ts": now, "kind": "tool", "tool_name": "web_fetch", "summary": "ok", "detail": {}}),
                json.dumps({"ts": now, "kind": "tool", "tool_name": "exec", "summary": "Error: command failed", "detail": {}}),
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        import tempfile

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "observations",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        # Should signal attention
        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertTrue(out["needs_attention"])
        self.assertGreaterEqual(out["observations"]["found_new"], 1)

        conn.close()

    def test_triage_tasks_mode_dedupes_by_state(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "task",
                        "tool_name": "memory_store",
                        "summary": "TODO: buy coffee this afternoon",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        # First run: should alert
        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)
        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)

        # Second run: should NOT alert again (state dedupe)
        buf2 = io.StringIO()
        with redirect_stdout(buf2):
            with self.assertRaises(SystemExit) as cm2:
                cmd_triage(conn, args)
        self.assertEqual(cm2.exception.code, 0)
        out2 = json.loads(buf2.getvalue())
        self.assertEqual(out2["tasks"]["found_new"], 0)

        conn.close()

    def test_triage_tasks_mode_no_dedupe_does_not_write_state(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "task",
                        "tool_name": "memory_store",
                        "summary": "TODO: buy coffee this afternoon",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type("Args", (), {
            "mode": "tasks",
            "since_minutes": 60,
            "limit": 10,
            "keywords": None,
            "cron_jobs_path": None,
            "tasks_since_minutes": 1440,
            "importance_min": 0.7,
            "state_path": state_path,
            "dedupe": False,
            "json": True,
        })()

        buf1 = io.StringIO()
        with redirect_stdout(buf1):
            with self.assertRaises(SystemExit) as cm1:
                cmd_triage(conn, args)
        self.assertEqual(cm1.exception.code, 10)
        out1 = json.loads(buf1.getvalue())
        self.assertEqual(out1["dedupe"], False)
        self.assertEqual(out1["tasks"]["found_total"], 1)
        self.assertEqual(out1["tasks"]["found_new"], 1)

        # no-dedupe should not write state
        with open(state_path, "r", encoding="utf-8") as fp:
            self.assertEqual(fp.read().strip(), "")

        buf2 = io.StringIO()
        with redirect_stdout(buf2):
            with self.assertRaises(SystemExit) as cm2:
                cmd_triage(conn, args)
        self.assertEqual(cm2.exception.code, 10)
        out2 = json.loads(buf2.getvalue())
        self.assertEqual(out2["tasks"]["found_new"], 1)

        conn.close()

    def test_triage_tasks_accepts_task_marker_without_colon(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "TODO buy coffee this afternoon",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "TODO buy coffee this afternoon")

        conn.close()

    def test_triage_tasks_accepts_marker_only_summary(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "TODO",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "TODO")

        conn.close()

    def test_triage_tasks_accepts_bracket_wrapped_task_marker_prefix(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "[TODO]buy coffee this afternoon",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "[TODO]buy coffee this afternoon")

        conn.close()

    def test_triage_tasks_accepts_curly_brace_wrapped_task_marker_prefix(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "{TODO}buy coffee this afternoon",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "{TODO}buy coffee this afternoon")

        conn.close()

    def test_triage_tasks_accepts_cjk_bracket_wrapped_task_marker_prefix(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "【TODO】buy coffee this afternoon",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "【TODO】buy coffee this afternoon")

        conn.close()

    def test_triage_tasks_accepts_markdown_checkbox_prefixed_task_marker(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "- [ ] TODO: rotate on-call notes",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "- [ ] TODO: rotate on-call notes")

        conn.close()

    def test_triage_tasks_accepts_emoji_checkbox_prefixed_task_marker(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "- [✅] TODO: rotate on-call notes",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "- [✅] TODO: rotate on-call notes")

        conn.close()

    def test_triage_tasks_accepts_plus_bullet_prefixed_task_marker(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "+ TODO: rotate on-call notes",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "+ TODO: rotate on-call notes")

        conn.close()

    def test_triage_tasks_accepts_blockquote_prefixed_task_marker(self):
        import tempfile
        from datetime import datetime, timezone

        summaries = (
            "> TODO: rotate on-call notes",
            "> > [ ] TASK: rotate on-call notes",
            ">> TODO: rotate on-call notes",
            "- > (iv) [ ] TODO: rotate on-call notes",
        )

        for summary in summaries:
            with self.subTest(summary=summary):
                conn = _connect(":memory:")

                now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
                sample = "\n".join(
                    [
                        json.dumps(
                            {
                                "ts": now,
                                "kind": "note",
                                "tool_name": "memory_store",
                                "summary": summary,
                                "detail": {"importance": 0.9},
                            }
                        )
                    ]
                )

                old_stdin = sys.stdin
                try:
                    sys.stdin = io.StringIO(sample)
                    args = type("Args", (), {"file": None, "json": True})()
                    with redirect_stdout(io.StringIO()):
                        cmd_ingest(conn, args)
                finally:
                    sys.stdin = old_stdin

                with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
                    state_path = st.name

                args = type(
                    "Args",
                    (),
                    {
                        "mode": "tasks",
                        "since_minutes": 60,
                        "limit": 10,
                        "keywords": None,
                        "cron_jobs_path": None,
                        "tasks_since_minutes": 1440,
                        "importance_min": 0.7,
                        "state_path": state_path,
                        "json": True,
                    },
                )()

                buf = io.StringIO()
                with redirect_stdout(buf):
                    with self.assertRaises(SystemExit) as cm:
                        cmd_triage(conn, args)

                self.assertEqual(cm.exception.code, 10)
                out = json.loads(buf.getvalue())
                self.assertEqual(out["tasks"]["found_new"], 1)
                self.assertEqual(out["tasks"]["matches"][0]["summary"], summary)

                conn.close()

    def test_triage_tasks_accepts_compact_wrappers_without_space(self):
        import tempfile
        from datetime import datetime, timezone

        summaries = (
            ">>[x]TODO: rotate on-call notes",
            "-TODO: rotate on-call notes",
            "(1)TODO: rotate on-call notes",
            "a)TODO: rotate on-call notes",
            "「TODO」rotate on-call notes",
            "『TODO』rotate on-call notes",
            "『TASK』rotate on-call notes",
            "《TODO》rotate on-call notes",
            "<TODO>rotate on-call notes",
            "＜TODO＞rotate on-call notes",
            "〔TODO〕rotate on-call notes",
            "〖TODO〗rotate on-call notes",
            "〘TODO〙rotate on-call notes",
            "[☑]TODO: rotate on-call notes",
        )

        for summary in summaries:
            with self.subTest(summary=summary):
                conn = _connect(":memory:")

                now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
                sample = "\n".join(
                    [
                        json.dumps(
                            {
                                "ts": now,
                                "kind": "note",
                                "tool_name": "memory_store",
                                "summary": summary,
                                "detail": {"importance": 0.9},
                            }
                        )
                    ]
                )

                old_stdin = sys.stdin
                try:
                    sys.stdin = io.StringIO(sample)
                    args = type("Args", (), {"file": None, "json": True})()
                    with redirect_stdout(io.StringIO()):
                        cmd_ingest(conn, args)
                finally:
                    sys.stdin = old_stdin

                with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
                    state_path = st.name

                args = type(
                    "Args",
                    (),
                    {
                        "mode": "tasks",
                        "since_minutes": 60,
                        "limit": 10,
                        "keywords": None,
                        "cron_jobs_path": None,
                        "tasks_since_minutes": 1440,
                        "importance_min": 0.7,
                        "state_path": state_path,
                        "json": True,
                    },
                )()

                buf = io.StringIO()
                with redirect_stdout(buf):
                    with self.assertRaises(SystemExit) as cm:
                        cmd_triage(conn, args)

                self.assertEqual(cm.exception.code, 10)
                out = json.loads(buf.getvalue())
                self.assertEqual(out["tasks"]["found_new"], 1)
                self.assertEqual(out["tasks"]["matches"][0]["summary"], summary)

                conn.close()

    def test_triage_tasks_accepts_unicode_bullet_prefixed_task_marker(self):
        import tempfile
        from datetime import datetime, timezone

        summaries = (
            "‣ TODO: rotate on-call notes",
            "∙ [ ] TASK: rotate on-call notes",
            "· [x] [REMINDER] rotate on-call notes",
            "• [☐] TASK: rotate on-call notes",
        )

        for summary in summaries:
            with self.subTest(summary=summary):
                conn = _connect(":memory:")

                now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
                sample = "\n".join(
                    [
                        json.dumps(
                            {
                                "ts": now,
                                "kind": "note",
                                "tool_name": "memory_store",
                                "summary": summary,
                                "detail": {"importance": 0.9},
                            }
                        )
                    ]
                )

                old_stdin = sys.stdin
                try:
                    sys.stdin = io.StringIO(sample)
                    args = type("Args", (), {"file": None, "json": True})()
                    with redirect_stdout(io.StringIO()):
                        cmd_ingest(conn, args)
                finally:
                    sys.stdin = old_stdin

                with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
                    state_path = st.name

                args = type(
                    "Args",
                    (),
                    {
                        "mode": "tasks",
                        "since_minutes": 60,
                        "limit": 10,
                        "keywords": None,
                        "cron_jobs_path": None,
                        "tasks_since_minutes": 1440,
                        "importance_min": 0.7,
                        "state_path": state_path,
                        "json": True,
                    },
                )()

                buf = io.StringIO()
                with redirect_stdout(buf):
                    with self.assertRaises(SystemExit) as cm:
                        cmd_triage(conn, args)

                self.assertEqual(cm.exception.code, 10)
                out = json.loads(buf.getvalue())
                self.assertEqual(out["tasks"]["found_new"], 1)
                self.assertEqual(out["tasks"]["matches"][0]["summary"], summary)

                conn.close()

    def test_triage_tasks_accepts_ordered_list_prefixed_task_marker(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "1. [ ] TODO: rotate on-call notes",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "1. [ ] TODO: rotate on-call notes")

        conn.close()

    def test_triage_tasks_accepts_alpha_ordered_prefix_task_marker(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "a) [ ] TODO: rotate on-call notes",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "a) [ ] TODO: rotate on-call notes")

        conn.close()

    def test_triage_tasks_accepts_parenthesized_alpha_ordered_prefix_task_marker(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "(a) [ ] TODO: rotate on-call notes",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "(a) [ ] TODO: rotate on-call notes")

        conn.close()

    def test_triage_tasks_accepts_roman_ordered_prefix_task_marker(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "iv) [ ] TODO: rotate on-call notes",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "iv) [ ] TODO: rotate on-call notes")

        conn.close()

    def test_triage_tasks_rejects_invalid_roman_ordered_prefix(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "ic) TODO: should not match",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 0)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 0)

        conn.close()

    def test_triage_tasks_accepts_parenthesized_ordered_prefix_task_marker(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "(1) [ ] TODO: rotate on-call notes",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "(1) [ ] TODO: rotate on-call notes")

        conn.close()

    def test_triage_tasks_accepts_full_width_parenthesized_ordered_prefix_task_marker(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "（１） [ ] TODO: rotate on-call notes",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "（１） [ ] TODO: rotate on-call notes")

        conn.close()

    def test_triage_tasks_accepts_full_width_task_marker_prefix(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "ＴＡＳＫ：整理慢煮 lane 清單",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "ＴＡＳＫ：整理慢煮 lane 清單")

        conn.close()

    def test_triage_tasks_accepts_task_marker_with_en_dash_separator(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "REMINDER–renew domain this week",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "REMINDER–renew domain this week")

        conn.close()

    def test_triage_tasks_accepts_task_marker_with_em_dash_separator(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "REMINDER—renew domain this week",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "REMINDER—renew domain this week")

        conn.close()

    def test_triage_tasks_accepts_task_marker_with_unicode_minus_separator(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "TASK−follow up on release checklist",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "TASK−follow up on release checklist")

        conn.close()

    def test_triage_tasks_accepts_task_marker_with_full_width_hyphen_separator(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "TASK－follow up on release checklist",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "TASK－follow up on release checklist")

        conn.close()

    def test_triage_tasks_accepts_task_marker_with_newline_separator(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "note",
                        "tool_name": "memory_store",
                        "summary": "TASK\nfollow up on release checklist",
                        "detail": {"importance": 0.9},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertEqual(out["tasks"]["matches"][0]["summary"], "TASK\nfollow up on release checklist")

        conn.close()

    def test_triage_tasks_parses_importance_object(self):
        import tempfile
        from datetime import datetime, timezone

        conn = _connect(":memory:")

        now = datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": now,
                        "kind": "task",
                        "tool_name": "memory_store",
                        "summary": "TODO: send invoice to vendor",
                        "detail": {"importance": {"score": 0.9, "label": "must_remember"}},
                    }
                )
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, args)
        finally:
            sys.stdin = old_stdin

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "tasks",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": None,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)

        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["tasks"]["found_new"], 1)
        self.assertAlmostEqual(out["tasks"]["matches"][0]["importance"], 0.9, places=5)

        conn.close()

    def test_triage_cron_errors_reads_jobs_json(self):
        import tempfile

        # Fake cron jobs store
        jobs = {
            "jobs": [
                {
                    "id": "job1",
                    "name": "Job 1",
                    "enabled": True,
                    "state": {"lastStatus": "ok", "lastRunAtMs": 9999999999999},
                },
                {
                    "id": "job2",
                    "name": "Job 2",
                    "enabled": True,
                    "state": {"lastStatus": "error", "lastRunAtMs": 9999999999999, "lastDurationMs": 1234},
                },
            ]
        }
        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as tmp:
            json.dump(jobs, tmp)
            tmp_path = tmp.name

        conn = _connect(":memory:")
        import tempfile

        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "cron-errors",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": tmp_path,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)
        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["cron"]["found_new"], 1)
        self.assertEqual(out["cron"]["matches"][0]["id"], "job2")

        conn.close()

    def test_store_persists_dual_language_fields(self):
        conn = _connect(":memory:")

        args = type(
            "Args",
            (),
            {
                "text": "나는 커피를 좋아해",
                "text_en": "I like coffee",
                "lang": "ko",
                "category": "preference",
                "importance": 0.8,
                "model": "test-model",
                "base_url": "https://example.com/v1",
                "workspace": None,
                "json": True,
            },
        )()

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._get_api_key", return_value=None):
            with redirect_stdout(buf):
                cmd_store(conn, args)

        out = json.loads(buf.getvalue())
        self.assertTrue(out["ok"])
        row = conn.execute("SELECT summary, summary_en, lang FROM observations WHERE id = ?", (out["id"],)).fetchone()
        self.assertEqual(row["summary"], "나는 커피를 좋아해")
        self.assertEqual(row["summary_en"], "I like coffee")
        self.assertEqual(row["lang"], "ko")
        conn.close()

    def test_store_emits_canonical_importance_object(self):
        conn = _connect(":memory:")

        args = type(
            "Args",
            (),
            {
                "text": "Prefer tabs over spaces",
                "text_en": None,
                "lang": "en",
                "category": "preference",
                "importance": 0.9,
                "model": "test-model",
                "base_url": "https://example.com/v1",
                "workspace": None,
                "json": True,
            },
        )()

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._get_api_key", return_value=None):
            with redirect_stdout(buf):
                cmd_store(conn, args)

        out = json.loads(buf.getvalue())
        row = conn.execute("SELECT detail_json FROM observations WHERE id = ?", (out["id"],)).fetchone()
        detail = json.loads(row["detail_json"])
        self.assertIsInstance(detail.get("importance"), dict)
        self.assertAlmostEqual(float(detail["importance"]["score"]), 0.9, places=5)
        self.assertEqual(detail["importance"]["label"], "must_remember")
        self.assertEqual(detail["importance"]["method"], "manual-via-cli")
        self.assertIn("graded_at", detail["importance"])

        conn.close()

    def test_vsearch_filters_to_query_dimension_and_skips_zero_dim_rows(self):
        conn = _connect(":memory:")
        for idx, summary in [(1, "good vector"), (2, "wrong dimension"), (3, "legacy zero dim")]:
            _insert_observation(
                conn,
                {
                    "ts": "2026-02-05T00:00:00Z",
                    "kind": "fact",
                    "tool_name": "memory_store",
                    "summary": summary,
                    "detail": {},
                },
            )

        from openclaw_mem.vector import pack_f32, l2_norm

        for obs_id, dim, vec, norm in [
            (1, 2, [1.0, 0.0], l2_norm([1.0, 0.0])),
            (2, 3, [1.0, 0.0, 0.0], l2_norm([1.0, 0.0, 0.0])),
            (3, 0, [], 0.0),
        ]:
            conn.execute(
                "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
                (obs_id, "test-model", dim, pack_f32(vec), norm, "2026-02-05T00:00:00Z"),
            )
        conn.commit()

        args = type(
            "Args",
            (),
            {
                "query": "semantic",
                "query_vector_json": "[1.0, 0.0]",
                "query_vector_file": None,
                "model": "test-model",
                "limit": 10,
                "base_url": "https://example.com/v1",
                "json": True,
            },
        )()

        model_count = conn.execute("SELECT COUNT(*) FROM observation_embeddings WHERE model = ?", ("test-model",)).fetchone()[0]
        dim_count = conn.execute("SELECT COUNT(*) FROM observation_embeddings WHERE model = ? AND dim = ?", ("test-model", 2)).fetchone()[0]
        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_vsearch(conn, args)
        out = json.loads(buf.getvalue())

        self.assertEqual(model_count, 3)
        self.assertEqual(dim_count, 1)
        self.assertEqual([r["id"] for r in out], [1])
        conn.close()

    def test_store_persists_english_embedding_when_text_en_present(self):
        conn = _connect(":memory:")

        class _FakeEmbedClient:
            def __init__(self, api_key: str, base_url: str = ""):
                pass

            def embed(self, texts, model):
                return [[1.0, 0.0] for _ in texts]

        args = type(
            "Args",
            (),
            {
                "text": "원문",
                "text_en": "english",
                "lang": "ko",
                "category": "fact",
                "importance": 0.7,
                "model": "test-model",
                "base_url": "https://example.com/v1",
                "workspace": None,
                "json": True,
            },
        )()

        from unittest.mock import patch

        with patch("openclaw_mem.cli._get_api_key", return_value="test-key"), patch("openclaw_mem.cli.OpenAIEmbeddingsClient", _FakeEmbedClient):
            with redirect_stdout(io.StringIO()):
                cmd_store(conn, args)

        orig = conn.execute("SELECT COUNT(*) FROM observation_embeddings").fetchone()[0]
        en = conn.execute("SELECT COUNT(*) FROM observation_embeddings_en").fetchone()[0]
        self.assertEqual(orig, 1)
        self.assertEqual(en, 1)
        conn.close()

    def test_hybrid_supports_query_en_route(self):
        conn = _connect(":memory:")

        sample = "\n".join(
            [
                json.dumps({"ts": "2026-02-04T13:00:00Z", "kind": "fact", "summary": "사과", "summary_en": "apple", "tool_name": "memory_store", "detail": {}}),
                json.dumps({"ts": "2026-02-04T13:01:00Z", "kind": "fact", "summary": "바나나", "summary_en": "banana", "tool_name": "memory_store", "detail": {}}),
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            ingest_args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, ingest_args)
        finally:
            sys.stdin = old_stdin

        from openclaw_mem.vector import pack_f32, l2_norm

        # Original table: id=1 aligns with [1,0], id=2 aligns with [0,1]
        conn.execute(
            "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (1, "test-model", 2, pack_f32([1.0, 0.0]), l2_norm([1.0, 0.0]), "2026-02-05T00:00:00Z"),
        )
        conn.execute(
            "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (2, "test-model", 2, pack_f32([0.0, 1.0]), l2_norm([0.0, 1.0]), "2026-02-05T00:00:00Z"),
        )
        # EN table intentionally reversed so we can detect table preference.
        conn.execute(
            "INSERT INTO observation_embeddings_en (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (1, "test-model", 2, pack_f32([0.0, 1.0]), l2_norm([0.0, 1.0]), "2026-02-05T00:00:00Z"),
        )
        conn.execute(
            "INSERT INTO observation_embeddings_en (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (2, "test-model", 2, pack_f32([1.0, 0.0]), l2_norm([1.0, 0.0]), "2026-02-05T00:00:00Z"),
        )
        conn.commit()

        class _FakeEmbedClient:
            def __init__(self, api_key: str, base_url: str = ""):
                pass

            def embed(self, texts, model):
                out = []
                for t in texts:
                    if t == "zzzz":
                        out.append([0.0, 0.0])
                    else:
                        out.append([1.0, 0.0])
                return out

        args = type(
            "Args",
            (),
            {
                "query": "zzzz",
                "query_en": "banana",
                "model": "test-model",
                "limit": 5,
                "k": 60,
                "base_url": "https://example.com/v1",
                "json": True,
            },
        )()

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._get_api_key", return_value="test-key"), patch("openclaw_mem.cli.OpenAIEmbeddingsClient", _FakeEmbedClient):
            with redirect_stdout(buf):
                cmd_hybrid(conn, args)

        out = json.loads(buf.getvalue())
        self.assertGreaterEqual(len(out), 1)
        self.assertEqual(out[0]["id"], 2)
        self.assertIn("vector_en", out[0]["match"])
        conn.close()

    def test_hybrid_rerank_reorders_results(self):
        conn = _connect(":memory:")

        sample = "\n".join(
            [
                json.dumps({"ts": "2026-02-04T13:00:00Z", "kind": "fact", "summary": "first", "summary_en": "first", "tool_name": "memory_store", "detail": {}}),
                json.dumps({"ts": "2026-02-04T13:01:00Z", "kind": "fact", "summary": "second", "summary_en": "second", "tool_name": "memory_store", "detail": {}}),
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            ingest_args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, ingest_args)
        finally:
            sys.stdin = old_stdin

        from openclaw_mem.vector import pack_f32, l2_norm

        conn.execute(
            "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (1, "test-model", 2, pack_f32([1.0, 0.0]), l2_norm([1.0, 0.0]), "2026-02-05T00:00:00Z"),
        )
        conn.execute(
            "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (2, "test-model", 2, pack_f32([0.0, 1.0]), l2_norm([0.0, 1.0]), "2026-02-05T00:00:00Z"),
        )
        conn.commit()

        class _FakeEmbedClient:
            def __init__(self, api_key: str, base_url: str = ""):
                pass

            def embed(self, texts, model):
                return [[1.0, 0.0] for _ in texts]

        args = type(
            "Args",
            (),
            {
                "query": "zzzz",
                "query_en": None,
                "model": "test-model",
                "limit": 5,
                "k": 60,
                "base_url": "https://example.com/v1",
                "rerank_provider": "jina",
                "rerank_model": "jina-reranker-v2-base-multilingual",
                "rerank_topn": 2,
                "rerank_api_key": "test-rerank-key",
                "rerank_base_url": None,
                "rerank_timeout_sec": 5,
                "json": True,
            },
        )()

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._get_api_key", return_value="test-key"), patch("openclaw_mem.cli.OpenAIEmbeddingsClient", _FakeEmbedClient), patch(
            "openclaw_mem.cli._call_rerank_provider", return_value=[(1, 0.9), (0, 0.1)]
        ):
            with redirect_stdout(buf):
                cmd_hybrid(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out[0]["id"], 2)
        self.assertEqual(out[0].get("rerank_provider"), "jina")
        self.assertEqual(out[0].get("rank_stage"), "rerank")
        conn.close()

    def test_hybrid_prefers_fresh_synthesis_cards_in_results(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {"kind": "fact", "summary": "alpha source one", "summary_en": "alpha source one", "tool_name": "memory_store", "detail": {}})
        _insert_observation(conn, {"kind": "fact", "summary": "alpha source two", "summary_en": "alpha source two", "tool_name": "memory_store", "detail": {}})

        compile_args = build_parser().parse_args([
            "graph", "--json", "synth", "compile",
            "--record-ref", "obs:1",
            "--record-ref", "obs:2",
            "--title", "Merged insight",
            "--summary", "Merged insight",
            "--why-it-matters", "Prefer the synthesis card",
        ])
        with redirect_stdout(io.StringIO()):
            compile_args.func(conn, compile_args)

        args = type(
            "Args",
            (),
            {
                "query": "alpha",
                "query_en": None,
                "model": "test-model",
                "limit": 5,
                "k": 60,
                "base_url": "https://example.com/v1",
                "rerank_provider": "none",
                "rerank_model": None,
                "rerank_topn": 5,
                "rerank_api_key": None,
                "rerank_base_url": None,
                "rerank_timeout_sec": 5,
                "json": True,
            },
        )()

        state = {
            "ordered_ids": [1, 2],
            "fts_ids": {1, 2},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.9, 2: 0.8},
            "obs_map": {
                1: {"id": 1, "summary": "alpha source one", "summary_en": "alpha source one", "kind": "fact", "lang": "en", "tool_name": "memory_store"},
                2: {"id": 2, "summary": "alpha source two", "summary_en": "alpha source two", "kind": "fact", "lang": "en", "tool_name": "memory_store"},
            },
            "rerank_scores": {},
            "rerank_applied": False,
            "rerank_provider": "none",
            "rerank_enabled": False,
            "candidate_limit": 12,
        }

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=state):
            with redirect_stdout(buf):
                cmd_hybrid(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out[0]["tool_name"], "graph.synth-compile")
        self.assertIn("graph_synthesis", out[0]["match"])
        self.assertEqual(out[0]["graph_consumption"]["coveredRawRefs"], ["obs:1", "obs:2"])
        conn.close()

    def test_hybrid_rerank_fail_open_keeps_rrf_order(self):
        conn = _connect(":memory:")

        sample = "\n".join(
            [
                json.dumps({"ts": "2026-02-04T13:00:00Z", "kind": "fact", "summary": "first", "summary_en": "first", "tool_name": "memory_store", "detail": {}}),
                json.dumps({"ts": "2026-02-04T13:01:00Z", "kind": "fact", "summary": "second", "summary_en": "second", "tool_name": "memory_store", "detail": {}}),
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            ingest_args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, ingest_args)
        finally:
            sys.stdin = old_stdin

        from openclaw_mem.vector import pack_f32, l2_norm

        conn.execute(
            "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (1, "test-model", 2, pack_f32([1.0, 0.0]), l2_norm([1.0, 0.0]), "2026-02-05T00:00:00Z"),
        )
        conn.execute(
            "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (2, "test-model", 2, pack_f32([0.0, 1.0]), l2_norm([0.0, 1.0]), "2026-02-05T00:00:00Z"),
        )
        conn.commit()

        class _FakeEmbedClient:
            def __init__(self, api_key: str, base_url: str = ""):
                pass

            def embed(self, texts, model):
                return [[1.0, 0.0] for _ in texts]

        args = type(
            "Args",
            (),
            {
                "query": "zzzz",
                "query_en": None,
                "model": "test-model",
                "limit": 5,
                "k": 60,
                "base_url": "https://example.com/v1",
                "rerank_provider": "jina",
                "rerank_model": "jina-reranker-v2-base-multilingual",
                "rerank_topn": 2,
                "rerank_api_key": "test-rerank-key",
                "rerank_base_url": None,
                "rerank_timeout_sec": 5,
                "json": True,
            },
        )()

        from unittest.mock import patch
        from contextlib import redirect_stderr

        buf = io.StringIO()
        err = io.StringIO()
        with patch("openclaw_mem.cli._get_api_key", return_value="test-key"), patch("openclaw_mem.cli.OpenAIEmbeddingsClient", _FakeEmbedClient), patch(
            "openclaw_mem.cli._call_rerank_provider", side_effect=RuntimeError("boom")
        ):
            with redirect_stdout(buf), redirect_stderr(err):
                cmd_hybrid(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out[0]["id"], 1)
        self.assertEqual(out[0].get("rerank_provider"), "jina")
        self.assertNotIn("rank_stage", out[0])
        self.assertIn("rerank failed", err.getvalue())
        conn.close()

    def test_pack_suppresses_superseded_proposal_rows(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1, 2],
            "fts_ids": {1, 2},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.9, 2: 0.8},
            "obs_map": {
                1: {
                    "summary": "yijin-loop-engine alias dirty v3 contains question mark mojibake ???",
                    "summary_en": None,
                    "kind": "memory.proposal",
                    "lang": "en",
                    "tool_name": "gateway.store.propose",
                },
                2: {
                    "summary": "yijin-loop-engine alias clean canonical memory v4",
                    "summary_en": None,
                    "kind": "memory.proposal",
                    "lang": "en",
                    "tool_name": "gateway.store.propose",
                },
            },
            "candidate_limit": 12,
        }
        _insert_observation(conn, {
            "kind": "memory.proposal",
            "summary": pack_state["obs_map"][1]["summary"],
            "tool_name": "gateway.store.propose",
            "detail": {"provenance": {"seed_id": "dirty-v3"}},
        })
        _insert_observation(conn, {
            "kind": "memory.proposal",
            "summary": pack_state["obs_map"][2]["summary"],
            "tool_name": "gateway.store.propose",
            "detail": {"provenance": {"seed_id": "clean-v4", "supersedes": ["obs:1"]}},
        })

        args = build_parser().parse_args(["pack", "--query", "yijin-loop-engine", "--json", "--limit", "5"] )

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual([item["recordRef"] for item in out["items"]], ["obs:2"])
        self.assertIn("canonical memory v4", out["bundle_text"])
        self.assertNotIn("???", out["bundle_text"])
        conn.close()

    def test_pack_rejects_blank_query(self):
        conn = _connect(":memory:")

        args = build_parser().parse_args(["pack", "--query", "   ", "--json"])

        buf = io.StringIO()
        with self.assertRaises(SystemExit) as cm:
            with redirect_stdout(buf):
                args.func(conn, args)

        self.assertEqual(cm.exception.code, 2)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["error"], "empty query")
        conn.close()

    def test_pack_accepts_query_en_for_hybrid_retrieval(self):
        conn = _connect(":memory:")

        captured = {}

        def fake_retrieve(conn_, args_, candidate_limit_override=None):
            captured["query_en"] = args_.query_en
            captured["candidate_limit_override"] = candidate_limit_override
            return {
                "ordered_ids": [1],
                "fts_ids": {1},
                "vec_ids": set(),
                "vec_en_ids": set(),
                "rrf_scores": {1: 0.77},
                "obs_map": {
                    1: {
                        "summary": "plain summary",
                        "summary_en": "plain summary",
                        "kind": "fact",
                        "lang": "en",
                    }
                },
                "candidate_limit": 12,
            }

        args = build_parser().parse_args(["pack", "--query", "테스트", "--query-en", "test", "--json", "--limit", "4"])

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", side_effect=fake_retrieve):
            with redirect_stdout(buf):
                args.func(conn, args)

        self.assertEqual(captured.get("query_en"), "test")
        self.assertEqual(captured.get("candidate_limit_override"), 12)

        out = json.loads(buf.getvalue())
        self.assertEqual(len(out["items"]), 1)
        self.assertEqual(out["items"][0]["recordRef"], "obs:1")
        conn.close()

    def test_pack_no_json_outputs_plain_bundle_text(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {
                1: {
                    "summary": "plain summary",
                    "summary_en": "plain summary",
                    "kind": "fact",
                    "lang": "en",
                }
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args(["pack", "--query", "something", "--no-json", "--limit", "3"])

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state):
            with redirect_stdout(buf):
                args.func(conn, args)

        self.assertEqual(buf.getvalue(), "- [obs:1] plain summary\n")
        conn.close()


    def test_pack_use_graph_on_adds_graph_payload_and_trace_extensions(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {
                1: {
                    "summary": "plain summary",
                    "summary_en": "plain summary",
                    "kind": "fact",
                    "lang": "en",
                }
            },
            "candidate_limit": 12,
        }

        # Force graph on; we patch graph internals to keep the test deterministic.
        args = build_parser().parse_args(["pack", "--query", "something", "--json", "--trace", "--use-graph", "on"])

        fake_index_payload = {
            "kind": "openclaw-mem.graph.index.v0",
            "budget": {"budgetTokens": 900, "estimatedTokens": 10},
            "top_candidates": [
                {
                    "recordRef": "obs:1",
                    "id": 1,
                    "ts": "2026-02-04T13:00:00Z",
                    "kind": "fact",
                    "tool_name": "memory_store",
                    "score": -9.0,
                    "title": "stub",
                    "why_relevant": "fts_match",
                }
            ],
            "suggested_next_expansions": [],
            "index_text": "[GRAPH_INDEX v0]\\n",
        }

        fake_graph_pack_payload = {
            "kind": "openclaw-mem.graph.pack.v0",
            "ts": "2026-02-04T13:00:00Z",
            "budget": {"budgetTokens": 1200, "estimatedTokens": 20},
            "items": [
                {
                    "recordRef": "obs:1",
                    "id": 1,
                    "ts": "2026-02-04T13:00:00Z",
                    "kind": "fact",
                    "tool_name": "memory_store",
                    "summary": "plain summary",
                }
            ],
            "bundle_text": "[GRAPH_CONTEXT v0]\\nItems: 1\\n\\n1) obs:1 :: plain summary\\n",
        }

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._graph_index_payload", return_value=fake_index_payload
        ), patch("openclaw_mem.cli._graph_pack_payload", return_value=fake_graph_pack_payload):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertIn("graph", out)
        self.assertTrue(out["graph"]["triggered"])
        self.assertIn("bundle_text_with_graph", out)
        self.assertIn("[GRAPH_CONTEXT v0]", out["bundle_text_with_graph"])

        self.assertIn("trace", out)
        self.assertIn("extensions", out["trace"])
        self.assertIn("graph", out["trace"]["extensions"])
        self.assertTrue(out["trace"]["extensions"]["graph"]["triggered"])

        conn.close()

    def test_pack_use_graph_auto_keyword_trigger_is_traceable(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {
                1: {
                    "summary": "roadmap summary",
                    "summary_en": "roadmap summary",
                    "kind": "fact",
                    "lang": "en",
                }
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args([
            "pack",
            "--query",
            "where is roadmap",
            "--json",
            "--trace",
            "--use-graph",
            "auto",
            "--graph-scope",
            "proj-a",
        ])

        fake_index_payload = {
            "kind": "openclaw-mem.graph.index.v0",
            "budget": {"budgetTokens": 900, "estimatedTokens": 10},
            "top_candidates": [
                {
                    "recordRef": "obs:1",
                    "id": 1,
                    "ts": "2026-02-04T13:00:00Z",
                    "kind": "fact",
                    "tool_name": "memory_store",
                    "score": -9.0,
                    "title": "stub",
                    "why_relevant": "fts_match",
                }
            ],
            "suggested_next_expansions": [],
            "index_text": "[GRAPH_INDEX v0]\n",
        }

        fake_graph_pack_payload = {
            "kind": "openclaw-mem.graph.pack.v0",
            "ts": "2026-02-04T13:00:00Z",
            "budget": {"budgetTokens": 1200, "estimatedTokens": 20},
            "items": [
                {
                    "recordRef": "obs:1",
                    "id": 1,
                    "ts": "2026-02-04T13:00:00Z",
                    "kind": "fact",
                    "tool_name": "memory_store",
                    "summary": "roadmap summary",
                }
            ],
            "bundle_text": "[GRAPH_CONTEXT v0]\nItems: 1\n\n1) obs:1 :: roadmap summary\n",
        }

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._graph_index_payload", return_value=fake_index_payload
        ), patch("openclaw_mem.cli._graph_pack_payload", return_value=fake_graph_pack_payload):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertTrue(out["graph"]["triggered"])
        self.assertEqual(out["graph"]["trigger_reason"], "keyword:A+B")
        self.assertEqual(out["graph"]["stage1"]["categories"], ["A", "B"])
        self.assertEqual(out["graph"]["probe"], {"ran": False})
        self.assertIn("[GRAPH_CONTEXT v0]", out["bundle_text_with_graph"])

        ext = out["trace"]["extensions"]["graph"]
        self.assertTrue(ext["triggered"])
        self.assertEqual(ext["trigger_reason"], "keyword:A+B")
        self.assertEqual(ext["probe"], {"ran": False})
        conn.close()

    def test_pack_use_graph_auto_probe_strong_triggers_preflight(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {
                1: {
                    "summary": "latency spike summary",
                    "summary_en": "latency spike summary",
                    "kind": "fact",
                    "lang": "en",
                }
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args([
            "pack",
            "--query",
            "latency spike recall lane",
            "--json",
            "--trace",
            "--use-graph",
            "auto",
            "--graph-scope",
            "proj-a",
        ])

        fake_probe = {
            "ran": True,
            "latency_ms": 7,
            "hit_count": 1,
            "best_score": -9.5,
            "scores": [-9.5],
        }

        fake_index_payload = {
            "kind": "openclaw-mem.graph.index.v0",
            "budget": {"budgetTokens": 900, "estimatedTokens": 10},
            "top_candidates": [
                {
                    "recordRef": "obs:1",
                    "id": 1,
                    "ts": "2026-02-04T13:00:00Z",
                    "kind": "fact",
                    "tool_name": "memory_store",
                    "score": -9.5,
                    "title": "stub",
                    "why_relevant": "fts_match",
                }
            ],
            "suggested_next_expansions": [],
            "index_text": "[GRAPH_INDEX v0]\n",
        }

        fake_graph_pack_payload = {
            "kind": "openclaw-mem.graph.pack.v0",
            "ts": "2026-02-04T13:00:00Z",
            "budget": {"budgetTokens": 1200, "estimatedTokens": 20},
            "items": [
                {
                    "recordRef": "obs:1",
                    "id": 1,
                    "ts": "2026-02-04T13:00:00Z",
                    "kind": "fact",
                    "tool_name": "memory_store",
                    "summary": "latency spike summary",
                }
            ],
            "bundle_text": "[GRAPH_CONTEXT v0]\nItems: 1\n\n1) obs:1 :: latency spike summary\n",
        }

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._pack_graph_probe_observations", return_value=fake_probe
        ), patch("openclaw_mem.cli._graph_index_payload", return_value=fake_index_payload), patch(
            "openclaw_mem.cli._graph_pack_payload", return_value=fake_graph_pack_payload
        ):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertTrue(out["graph"]["triggered"])
        self.assertEqual(out["graph"]["trigger_reason"], "probe_strong")
        self.assertEqual(out["graph"]["probe_decision"], "fire_probe_strong")
        self.assertEqual(out["graph"]["stage1"]["categories"], [])
        self.assertEqual(out["graph"]["probe"]["hit_count"], 1)
        self.assertEqual(out["trace"]["extensions"]["graph"]["probe_decision"], "fire_probe_strong")
        conn.close()

    def test_pack_use_graph_on_provenance_policy_filters_unstructured_graph_candidates(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {
                1: {
                    "summary": "plain summary",
                    "summary_en": "plain summary",
                    "kind": "fact",
                    "lang": "en",
                }
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args(
            [
                "pack",
                "--query",
                "something",
                "--json",
                "--trace",
                "--use-graph",
                "on",
                "--graph-query-db",
                "/tmp/graph.db",
            ]
        )

        fake_index_payload = {
            "kind": "openclaw-mem.graph.index.v0",
            "budget": {"budgetTokens": 900, "estimatedTokens": 10},
            "top_candidates": [
                {"recordRef": "obs:1", "id": 1, "kind": "fact", "tool_name": "memory_store", "title": "structured"},
                {"recordRef": "obs:2", "id": 2, "kind": "fact", "tool_name": "memory_store", "title": "opaque"},
            ],
            "suggested_next_expansions": [],
            "index_text": "[GRAPH_INDEX v0]\\n",
        }

        captured: dict[str, list[str]] = {}

        def fake_graph_pack_payload(conn_, *, raw_ids, budget_tokens, max_items, allow_empty=False):
            captured["raw_ids"] = list(raw_ids)
            return {
                "kind": "openclaw-mem.graph.pack.v0",
                "ts": "2026-02-04T13:00:00Z",
                "budget": {"budgetTokens": budget_tokens, "estimatedTokens": 5},
                "items": [{"recordRef": ref, "id": int(ref.split(":", 1)[1]), "summary": "g"} for ref in raw_ids],
                "bundle_text": "\\n".join(raw_ids) + ("\\n" if raw_ids else ""),
            }

        def fake_query_subgraph(*, db_path, node_id, hops, direction, max_nodes, max_edges, require_structured_provenance):
            if node_id == "obs:1":
                return {
                    "edge_count": 1,
                    "skipped_unstructured_provenance": 0,
                    "provenance": {
                        "coverage": {
                            "edge_count": 1,
                            "structured_edge_count": 1,
                        },
                        "kind_counts": {"file_line": 1},
                    },
                }
            return {
                "edge_count": 0,
                "skipped_unstructured_provenance": 1,
                "provenance": {
                    "coverage": {
                        "edge_count": 0,
                        "structured_edge_count": 0,
                    },
                    "kind_counts": {"opaque": 1},
                },
            }

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._graph_index_payload", return_value=fake_index_payload
        ), patch("openclaw_mem.cli._graph_pack_payload", side_effect=fake_graph_pack_payload), patch(
            "openclaw_mem.cli.query_subgraph", side_effect=fake_query_subgraph
        ):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(captured.get("raw_ids"), ["obs:1"])
        self.assertEqual(out["graph"]["selection_count_pre_policy"], 2)
        self.assertEqual(out["graph"]["selection_count"], 1)

        decisions = out["graph"]["provenance_policy"]["decisions"]
        self.assertEqual(len(decisions), 2)
        self.assertEqual(decisions[0]["recordRef"], "obs:1")
        self.assertTrue(decisions[0]["included"])
        self.assertEqual(decisions[0]["reason"], "graph_provenance_structured")
        self.assertEqual(decisions[1]["recordRef"], "obs:2")
        self.assertFalse(decisions[1]["included"])
        self.assertEqual(decisions[1]["reason"], "graph_provenance_unstructured_only")

        ext = out["trace"]["extensions"]["graph"]
        self.assertEqual(ext["selection_count_pre_policy"], 2)
        self.assertEqual(ext["selected_refs_count"], 1)
        self.assertEqual(ext["provenance_policy"]["included_count"], 1)
        self.assertEqual(ext["provenance_policy"]["excluded_count"], 1)
        conn.close()

    def test_pack_use_graph_on_provenance_policy_fail_open_on_query_error(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {
                1: {
                    "summary": "plain summary",
                    "summary_en": "plain summary",
                    "kind": "fact",
                    "lang": "en",
                }
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args(
            [
                "pack",
                "--query",
                "something",
                "--json",
                "--trace",
                "--use-graph",
                "on",
                "--graph-query-db",
                "/tmp/graph.db",
            ]
        )

        fake_index_payload = {
            "kind": "openclaw-mem.graph.index.v0",
            "budget": {"budgetTokens": 900, "estimatedTokens": 10},
            "top_candidates": [
                {"recordRef": "obs:1", "id": 1, "kind": "fact", "tool_name": "memory_store", "title": "one"},
                {"recordRef": "obs:2", "id": 2, "kind": "fact", "tool_name": "memory_store", "title": "two"},
            ],
            "suggested_next_expansions": [],
            "index_text": "[GRAPH_INDEX v0]\\n",
        }

        captured: dict[str, list[str]] = {}

        def fake_graph_pack_payload(conn_, *, raw_ids, budget_tokens, max_items, allow_empty=False):
            captured["raw_ids"] = list(raw_ids)
            return {
                "kind": "openclaw-mem.graph.pack.v0",
                "ts": "2026-02-04T13:00:00Z",
                "budget": {"budgetTokens": budget_tokens, "estimatedTokens": 5},
                "items": [{"recordRef": ref, "id": int(ref.split(":", 1)[1]), "summary": "g"} for ref in raw_ids],
                "bundle_text": "\\n".join(raw_ids) + ("\\n" if raw_ids else ""),
            }

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._graph_index_payload", return_value=fake_index_payload
        ), patch("openclaw_mem.cli._graph_pack_payload", side_effect=fake_graph_pack_payload), patch(
            "openclaw_mem.cli.query_subgraph", side_effect=ValueError("graph db not found: /tmp/private/graph.db")
        ):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(captured.get("raw_ids"), ["obs:1", "obs:2"])
        policy = out["graph"]["provenance_policy"]
        self.assertEqual(policy["fail_open_count"], 2)
        reasons = [d["reason"] for d in policy["decisions"]]
        self.assertEqual(reasons, ["graph_provenance_fail_open_query_error", "graph_provenance_fail_open_query_error"])
        for d in policy["decisions"]:
            self.assertTrue(d["included"])
            self.assertEqual(d["error_code"], "graph_db_not_found")

        conn.close()

    def test_pack_use_graph_on_provenance_policy_contract_shape_is_stable(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {
                1: {
                    "summary": "plain summary",
                    "summary_en": "plain summary",
                    "kind": "fact",
                    "lang": "en",
                }
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args(
            [
                "pack",
                "--query",
                "something",
                "--json",
                "--trace",
                "--use-graph",
                "on",
                "--graph-query-db",
                "/tmp/graph.db",
            ]
        )

        fake_index_payload = {
            "kind": "openclaw-mem.graph.index.v0",
            "budget": {"budgetTokens": 900, "estimatedTokens": 10},
            "top_candidates": [
                {"recordRef": "obs:1", "id": 1, "kind": "fact", "tool_name": "memory_store", "title": "structured"},
                {"recordRef": "obs:2", "id": 2, "kind": "fact", "tool_name": "memory_store", "title": "opaque"},
                {"recordRef": "obs:3", "id": 3, "kind": "fact", "tool_name": "memory_store", "title": "query_error"},
            ],
            "suggested_next_expansions": [],
            "index_text": "[GRAPH_INDEX v0]\\n",
        }

        def fake_graph_pack_payload(conn_, *, raw_ids, budget_tokens, max_items, allow_empty=False):
            return {
                "kind": "openclaw-mem.graph.pack.v0",
                "ts": "2026-02-04T13:00:00Z",
                "budget": {"budgetTokens": budget_tokens, "estimatedTokens": 5},
                "items": [{"recordRef": ref, "id": int(ref.split(":", 1)[1]), "summary": "g"} for ref in raw_ids],
                "bundle_text": "\\n".join(raw_ids) + ("\\n" if raw_ids else ""),
            }

        def fake_query_subgraph(*, db_path, node_id, hops, direction, max_nodes, max_edges, require_structured_provenance):
            if node_id == "obs:1":
                return {
                    "edge_count": 1,
                    "skipped_unstructured_provenance": 0,
                    "provenance": {
                        "coverage": {
                            "edge_count": 1,
                            "structured_edge_count": 1,
                        },
                        "kind_counts": {"file_line": 1},
                    },
                }
            if node_id == "obs:2":
                return {
                    "edge_count": 0,
                    "skipped_unstructured_provenance": 1,
                    "provenance": {
                        "coverage": {
                            "edge_count": 0,
                            "structured_edge_count": 0,
                        },
                        "kind_counts": {"opaque": 1},
                    },
                }
            raise ValueError("unknown node_id: obs:3")

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._graph_index_payload", return_value=fake_index_payload
        ), patch("openclaw_mem.cli._graph_pack_payload", side_effect=fake_graph_pack_payload), patch(
            "openclaw_mem.cli.query_subgraph", side_effect=fake_query_subgraph
        ):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        policy = out["graph"]["provenance_policy"]
        self.assertEqual(
            set(policy.keys()),
            {
                "kind",
                "mode",
                "require_structured_provenance",
                "graph_query_db_configured",
                "bounds",
                "checked_count",
                "included_count",
                "excluded_count",
                "fail_open_count",
                "decision_reason_counts",
                "decisions",
                "selected_refs",
            },
        )
        self.assertEqual(policy["kind"], "openclaw-mem.pack.graph.provenance-policy.v1")
        self.assertEqual(policy["decision_reason_counts"], {
            "graph_provenance_fail_open_query_error": 1,
            "graph_provenance_structured": 1,
            "graph_provenance_unstructured_only": 1,
        })
        self.assertEqual(policy["selected_refs"], ["obs:1", "obs:3"])
        self.assertEqual(policy["included_count"], 2)
        self.assertEqual(policy["excluded_count"], 1)

        decisions = policy["decisions"]
        self.assertEqual(len(decisions), 3)
        for decision in decisions:
            self.assertEqual(
                set(decision.keys()),
                {"recordRef", "included", "reason", "fail_open", "error_code", "provenance_quality"},
            )
            prov_quality = decision["provenance_quality"]
            if prov_quality is not None:
                self.assertEqual(
                    set(prov_quality.keys()),
                    {"edge_count", "structured_edge_count", "skipped_unstructured_provenance", "kind_counts"},
                )

        trace_policy = out["trace"]["extensions"]["graph"]["provenance_policy"]
        self.assertEqual(trace_policy, policy)
        conn.close()

    def test_pack_graph_provenance_and_trust_policy_contracts_are_compatible(self):
        conn = _connect(":memory:")

        _insert_observation(
            conn,
            {
                "kind": "fact",
                "tool_name": "memory_store",
                "summary": "trusted row",
                "detail": {"trust_tier": "trusted"},
            },
        )
        _insert_observation(
            conn,
            {
                "kind": "fact",
                "tool_name": "memory_store",
                "summary": "quarantine row",
                "detail": {"trust_tier": "quarantine"},
            },
        )
        _insert_observation(
            conn,
            {
                "kind": "fact",
                "tool_name": "memory_store",
                "summary": "unknown trust row",
                "detail": {},
            },
        )

        pack_state = {
            "ordered_ids": [1, 2, 3],
            "fts_ids": {1, 2, 3},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.91, 2: 0.82, 3: 0.73},
            "obs_map": {
                1: {"summary": "trusted row", "summary_en": "trusted row", "kind": "fact", "lang": "en"},
                2: {"summary": "quarantine row", "summary_en": "quarantine row", "kind": "fact", "lang": "en"},
                3: {"summary": "unknown trust row", "summary_en": "unknown trust row", "kind": "fact", "lang": "en"},
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args(
            [
                "pack",
                "--query",
                "compat",
                "--json",
                "--trace",
                "--use-graph",
                "on",
                "--graph-query-db",
                "/tmp/graph.db",
                "--pack-trust-policy",
                "exclude_quarantined_fail_open",
            ]
        )

        fake_index_payload = {
            "kind": "openclaw-mem.graph.index.v0",
            "budget": {"budgetTokens": 900, "estimatedTokens": 10},
            "top_candidates": [
                {"recordRef": "obs:1", "id": 1, "kind": "fact", "tool_name": "memory_store", "title": "structured"},
                {"recordRef": "obs:2", "id": 2, "kind": "fact", "tool_name": "memory_store", "title": "opaque"},
                {"recordRef": "obs:3", "id": 3, "kind": "fact", "tool_name": "memory_store", "title": "query_error"},
            ],
            "suggested_next_expansions": [],
            "index_text": "[GRAPH_INDEX v0]\\n",
        }

        captured: dict[str, list[str]] = {}

        def fake_graph_pack_payload(conn_, *, raw_ids, budget_tokens, max_items, allow_empty=False):
            captured["raw_ids"] = list(raw_ids)
            return {
                "kind": "openclaw-mem.graph.pack.v0",
                "ts": "2026-02-04T13:00:00Z",
                "budget": {"budgetTokens": budget_tokens, "estimatedTokens": 5},
                "items": [{"recordRef": ref, "id": int(ref.split(":", 1)[1]), "summary": "g"} for ref in raw_ids],
                "bundle_text": "\\n".join(raw_ids) + ("\\n" if raw_ids else ""),
            }

        def fake_query_subgraph(*, db_path, node_id, hops, direction, max_nodes, max_edges, require_structured_provenance):
            if node_id == "obs:1":
                return {
                    "edge_count": 1,
                    "skipped_unstructured_provenance": 0,
                    "provenance": {
                        "coverage": {
                            "edge_count": 1,
                            "structured_edge_count": 1,
                        },
                        "kind_counts": {"file_line": 1},
                    },
                }
            if node_id == "obs:2":
                return {
                    "edge_count": 0,
                    "skipped_unstructured_provenance": 1,
                    "provenance": {
                        "coverage": {
                            "edge_count": 0,
                            "structured_edge_count": 0,
                        },
                        "kind_counts": {"manual": 1},
                    },
                }
            raise ValueError("unknown node_id: obs:3")

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._graph_index_payload", return_value=fake_index_payload
        ), patch("openclaw_mem.cli._graph_pack_payload", side_effect=fake_graph_pack_payload), patch(
            "openclaw_mem.cli.query_subgraph", side_effect=fake_query_subgraph
        ):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(captured.get("raw_ids"), ["obs:1", "obs:3"])

        graph_policy = out["graph"]["provenance_policy"]
        self.assertEqual(graph_policy["selected_refs"], ["obs:1", "obs:3"])
        self.assertEqual(graph_policy["decision_reason_counts"], {
            "graph_provenance_fail_open_query_error": 1,
            "graph_provenance_structured": 1,
            "graph_provenance_unstructured_only": 1,
        })

        trust_policy = out["trust_policy"]
        self.assertEqual(
            set(trust_policy.keys()),
            {
                "kind",
                "mode",
                "checked_count",
                "included_count",
                "excluded_count",
                "fail_open_count",
                "decision_reason_counts",
                "decisions",
                "selected_refs",
            },
        )
        self.assertEqual(trust_policy["kind"], "openclaw-mem.pack.trust-policy.v1")
        self.assertEqual(trust_policy["mode"], "exclude_quarantined_fail_open")
        self.assertEqual(trust_policy["selected_refs"], ["obs:1", "obs:3"])
        self.assertEqual(trust_policy["decision_reason_counts"], {
            "trust_allowed": 1,
            "trust_quarantined_excluded": 1,
            "trust_unknown_fail_open": 1,
        })

        selected_refs = [item["recordRef"] for item in out["items"]]
        citation_refs = [item["recordRef"] for item in out["citations"]]
        self.assertEqual(selected_refs, ["obs:1", "obs:3"])
        self.assertEqual(citation_refs, ["obs:1", "obs:3"])
        self.assertEqual(out["trace"]["extensions"]["graph"]["provenance_policy"], graph_policy)
        self.assertEqual(out["trace"]["extensions"]["trust_policy"], trust_policy)

        policy_surface = out["policy_surface"]
        self.assertEqual(policy_surface["kind"], "openclaw-mem.pack.policy-surface.v1")
        self.assertEqual(policy_surface["selection"]["pack_selected_refs"], selected_refs)
        self.assertEqual(policy_surface["selection"]["citation_record_refs"], citation_refs)
        self.assertEqual(policy_surface["selection"]["trust_selected_refs"], trust_policy["selected_refs"])
        self.assertEqual(policy_surface["selection"]["graph_selected_refs"], graph_policy["selected_refs"])
        self.assertEqual(policy_surface["selection"]["shared_pack_and_graph_refs"], selected_refs)
        self.assertEqual(policy_surface["reasons"]["trust_policy_reason_counts"], trust_policy["decision_reason_counts"])
        self.assertEqual(
            policy_surface["reasons"]["graph_provenance_reason_counts"],
            graph_policy["decision_reason_counts"],
        )
        self.assertEqual(policy_surface["policies"]["trust_policy"]["selected_refs"], trust_policy["selected_refs"])
        self.assertEqual(
            policy_surface["policies"]["graph_provenance_policy"]["selected_refs"],
            graph_policy["selected_refs"],
        )
        self.assertEqual(policy_surface["consistency"]["pack_items_match_citations"], True)
        self.assertEqual(policy_surface["consistency"]["pack_items_subset_of_trust_selected_refs"], True)
        self.assertEqual(policy_surface["consistency"]["pack_items_missing_from_trust_selected_refs"], [])
        self.assertEqual(policy_surface["selection"]["pack_selected_refs"], out["trace"]["output"]["refreshedRecordRefs"])
        self.assertEqual(out["trace"]["extensions"]["policy_surface"], policy_surface)

        conn.close()

    def test_pack_lifecycle_shadow_logging_tracks_real_selection_and_stays_non_mutating(self):
        conn = _connect(":memory:")

        _insert_observation(
            conn,
            {
                "kind": "fact",
                "tool_name": "memory_store",
                "summary": "seed observation",
                "detail": {"importance": {"score": 0.8, "label": "must_remember"}},
            },
        )

        obs_before = int(conn.execute("SELECT COUNT(*) FROM observations").fetchone()[0])

        pack_state = {
            "ordered_ids": [1, 2],
            "fts_ids": {1, 2},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.9, 2: 0.8},
            "obs_map": {
                1: {"summary": "short", "summary_en": "short", "kind": "fact", "lang": "en"},
                2: {
                    "summary": "this summary is intentionally long to exceed token budget",
                    "summary_en": "this summary is intentionally long to exceed token budget",
                    "kind": "fact",
                    "lang": "en",
                },
            },
            "candidate_limit": 10,
        }

        args = build_parser().parse_args(
            [
                "pack",
                "--query",
                "lifecycle demo",
                "--json",
                "--trace",
                "--limit",
                "2",
                "--budget-tokens",
                "2",
                "--pack-lifecycle-shadow",
                "on",
            ]
        )

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        lifecycle = out["trace"]["extensions"]["lifecycle_shadow"]

        obs_after = int(conn.execute("SELECT COUNT(*) FROM observations").fetchone()[0])
        self.assertEqual(obs_before, obs_after)

        log_rows = conn.execute(
            "SELECT receipt_json, selected_count, citation_count, candidate_count FROM pack_lifecycle_shadow_log ORDER BY id DESC LIMIT 1"
        ).fetchall()
        self.assertEqual(len(log_rows), 1)

        receipt = json.loads(log_rows[0]["receipt_json"])
        self.assertEqual(receipt, lifecycle)
        self.assertEqual(receipt["selection"]["pack_selected_refs"], ["obs:1"])
        self.assertEqual(receipt["selection"]["citation_record_refs"], ["obs:1"])
        self.assertEqual(receipt["selection"]["trace_refreshed_record_refs"], ["obs:1"])
        self.assertEqual(receipt["counts"]["selected_total"], 1)
        self.assertEqual(receipt["counts"]["citation_total"], 1)
        self.assertEqual(receipt["counts"]["candidate_total"], 2)
        self.assertEqual(receipt["counts"]["excluded_total"], 1)
        self.assertEqual(receipt["reasons"]["pack_excluded_reason_counts"], {"budget_tokens_exceeded": 1})
        self.assertEqual(receipt["mutation"]["memory_mutation"], "none")
        self.assertEqual(receipt["mutation"]["auto_archive_applied"], 0)
        self.assertEqual(receipt["mutation"]["auto_mutation_applied"], 0)

        self.assertEqual(log_rows[0]["selected_count"], 1)
        self.assertEqual(log_rows[0]["citation_count"], 1)
        self.assertEqual(log_rows[0]["candidate_count"], 2)
        conn.close()

    def test_pack_lifecycle_shadow_off_skips_log_and_trace_extension(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.7},
            "obs_map": {
                1: {"summary": "short", "summary_en": "short", "kind": "fact", "lang": "en"},
            },
            "candidate_limit": 8,
        }

        args = build_parser().parse_args(
            [
                "pack",
                "--query",
                "lifecycle disabled",
                "--json",
                "--trace",
                "--pack-lifecycle-shadow",
                "off",
            ]
        )

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertNotIn("lifecycle_shadow", out["trace"]["extensions"])

        row_count = int(conn.execute("SELECT COUNT(*) FROM pack_lifecycle_shadow_log").fetchone()[0])
        self.assertEqual(row_count, 0)
        conn.close()

    def test_pack_lifecycle_write_refreshes_selected_records_only_when_opted_in(self):
        conn = _connect(":memory:")

        _insert_observation(
            conn,
            {
                "kind": "fact",
                "tool_name": "memory_store",
                "summary": "selected lifecycle record",
                "detail": {"lifecycle": {"used_count": 2, "archived_at": "2026-01-01T00:00:00Z"}},
            },
        )
        _insert_observation(
            conn,
            {
                "kind": "fact",
                "tool_name": "memory_store",
                "summary": "budget excluded lifecycle record with a much longer summary",
                "detail": {"lifecycle": {"used_count": 7}},
            },
        )

        pack_state = {
            "ordered_ids": [1, 2],
            "fts_ids": {1, 2},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.9, 2: 0.8},
            "obs_map": {
                1: {"summary": "short", "summary_en": "short", "kind": "fact", "lang": "en"},
                2: {
                    "summary": "this summary is intentionally long to exceed token budget",
                    "summary_en": "this summary is intentionally long to exceed token budget",
                    "kind": "fact",
                    "lang": "en",
                },
            },
            "candidate_limit": 10,
        }

        args = build_parser().parse_args(
            [
                "pack",
                "--query",
                "lifecycle write demo",
                "--json",
                "--trace",
                "--limit",
                "2",
                "--budget-tokens",
                "2",
                "--pack-lifecycle-write",
                "on",
            ]
        )

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        lifecycle_write = out["trace"]["extensions"]["lifecycle_write"]
        self.assertEqual(lifecycle_write["selection"]["refreshed_record_refs"], ["obs:1"])
        self.assertEqual(lifecycle_write["mutation"]["writes_observations"], 1)
        self.assertEqual(lifecycle_write["mutation"]["hard_delete_applied"], 0)

        row1 = conn.execute("SELECT detail_json FROM observations WHERE id = 1").fetchone()
        row2 = conn.execute("SELECT detail_json FROM observations WHERE id = 2").fetchone()
        detail1 = json.loads(row1["detail_json"])
        detail2 = json.loads(row2["detail_json"])

        self.assertEqual(detail1["lifecycle"]["used_count"], 3)
        self.assertEqual(detail1["lifecycle"]["last_used_at"], lifecycle_write["ts"])
        self.assertEqual(detail1["lifecycle"]["archived_at"], "2026-01-01T00:00:00Z")
        self.assertEqual(detail2["lifecycle"]["used_count"], 7)
        self.assertNotIn("last_used_at", detail2["lifecycle"])
        conn.close()

    def test_pack_lifecycle_write_defaults_off(self):
        args = build_parser().parse_args(["pack", "--query", "x", "--json"])
        self.assertEqual(args.pack_lifecycle_write, "off")

    def test_pack_lifecycle_write_failure_rolls_back_before_shadow_log_commit(self):
        conn = _connect(":memory:")

        _insert_observation(
            conn,
            {
                "kind": "fact",
                "tool_name": "memory_store",
                "summary": "first selected lifecycle record",
                "detail": {"lifecycle": {"used_count": 2}},
            },
        )
        _insert_observation(
            conn,
            {
                "kind": "fact",
                "tool_name": "memory_store",
                "summary": "second selected lifecycle record",
                "detail": {"lifecycle": {"used_count": 7}},
            },
        )
        conn.execute(
            """
            CREATE TRIGGER lifecycle_write_fail_second
            BEFORE UPDATE OF detail_json ON observations
            WHEN old.id = 2
            BEGIN
                SELECT RAISE(ABORT, 'forced lifecycle write failure');
            END;
            """
        )
        conn.commit()

        pack_state = {
            "ordered_ids": [1, 2],
            "fts_ids": {1, 2},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.9, 2: 0.8},
            "obs_map": {
                1: {"summary": "short one", "summary_en": "short one", "kind": "fact", "lang": "en"},
                2: {"summary": "short two", "summary_en": "short two", "kind": "fact", "lang": "en"},
            },
            "candidate_limit": 10,
        }

        args = build_parser().parse_args(
            [
                "pack",
                "--query",
                "lifecycle rollback demo",
                "--json",
                "--trace",
                "--limit",
                "2",
                "--budget-tokens",
                "100",
                "--pack-lifecycle-write",
                "on",
            ]
        )

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        lifecycle_write = out["trace"]["extensions"]["lifecycle_write"]
        self.assertEqual(lifecycle_write["mutation"]["memory_mutation"], "failed_open")
        self.assertEqual(lifecycle_write["mutation"]["writes_observations"], 0)
        self.assertEqual(lifecycle_write["storage"]["error_code"], "IntegrityError")

        row1 = conn.execute("SELECT detail_json FROM observations WHERE id = 1").fetchone()
        row2 = conn.execute("SELECT detail_json FROM observations WHERE id = 2").fetchone()
        detail1 = json.loads(row1["detail_json"])
        detail2 = json.loads(row2["detail_json"])

        self.assertEqual(detail1["lifecycle"]["used_count"], 2)
        self.assertNotIn("last_used_at", detail1["lifecycle"])
        self.assertEqual(detail2["lifecycle"]["used_count"], 7)
        self.assertNotIn("last_used_at", detail2["lifecycle"])

        shadow_rows = conn.execute("SELECT COUNT(*) FROM pack_lifecycle_shadow_log").fetchone()[0]
        self.assertEqual(shadow_rows, 1)
        conn.close()

    def test_pack_use_graph_on_elides_l1_items_covered_by_preferred_cards(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {
                1: {
                    "summary": "plain summary",
                    "summary_en": "plain summary",
                    "kind": "fact",
                    "lang": "en",
                }
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args(["pack", "--query", "something", "--json", "--trace", "--use-graph", "on"])

        fake_index_payload = {
            "kind": "openclaw-mem.graph.index.v0",
            "budget": {"budgetTokens": 900, "estimatedTokens": 10},
            "top_candidates": [{"recordRef": "obs:1", "id": 1, "ts": "2026-02-04T13:00:00Z", "kind": "fact", "tool_name": "memory_store", "score": -9.0, "title": "stub", "why_relevant": "fts_match"}],
            "suggested_next_expansions": [],
            "index_text": "[GRAPH_INDEX v0]\n",
        }

        fake_graph_pack_payload = {
            "kind": "openclaw-mem.graph.pack.v0",
            "ts": "2026-02-04T13:00:00Z",
            "budget": {"budgetTokens": 1200, "estimatedTokens": 20},
            "selection": {
                "inputRecordRefs": ["obs:1", "obs:2"],
                "recordRefs": ["obs:99"],
                "selectedCount": 1,
                "preferredCardRefs": ["obs:99"],
                "coveredRawRefs": ["obs:1"],
            },
            "items": [{"recordRef": "obs:99", "id": 99, "ts": "2026-02-04T13:00:00Z", "kind": "note", "tool_name": "graph.synth-compile", "summary": "graph synthesis"}],
            "bundle_text": "[GRAPH_CONTEXT v0]\nItems: 1\n\n1) obs:99 ts=2026-02-04T13:00:00Z [note] graph.synth-compile :: graph synthesis\n",
        }

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._graph_index_payload", return_value=fake_index_payload
        ), patch("openclaw_mem.cli._graph_pack_payload", return_value=fake_graph_pack_payload):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertIn("graph", out)
        self.assertEqual(out["graph"]["consumption"]["preferredCardRefs"], ["obs:99"])
        self.assertEqual(out["graph"]["consumption"]["coveredRawRefs"], ["obs:1"])
        self.assertEqual(out["graph"]["consumption"]["elidedL1Refs"], ["obs:1"])
        self.assertEqual(out["graph"]["consumption"]["elidedL1Count"], 1)
        self.assertIn("obs:99", out["bundle_text_with_graph"])
        self.assertNotIn("- [obs:1] plain summary", out["bundle_text_with_graph"])
        self.assertEqual(out["trace"]["extensions"]["graph"]["consumption"]["elided_l1_count"], 1)

        conn.close()

    def test_pack_use_graph_auto_short_artifact_ref_can_bypass_too_short_antitrigger(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {
                1: {
                    "summary": "plain summary",
                    "summary_en": "plain summary",
                    "kind": "fact",
                    "lang": "en",
                }
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args(["pack", "--query", "docs/specs/", "--json", "--trace", "--use-graph", "auto", "--graph-scope", "proj-a"])

        fake_index_payload = {
            "kind": "openclaw-mem.graph.index.v0",
            "budget": {"budgetTokens": 900, "estimatedTokens": 10},
            "top_candidates": [{"recordRef": "obs:1", "id": 1, "ts": "2026-02-04T13:00:00Z", "kind": "fact", "tool_name": "memory_store", "score": -9.0, "title": "stub", "why_relevant": "fts_match"}],
            "suggested_next_expansions": [],
            "index_text": "[GRAPH_INDEX v0]\n",
        }

        fake_graph_pack_payload = {
            "kind": "openclaw-mem.graph.pack.v0",
            "ts": "2026-02-04T13:00:00Z",
            "budget": {"budgetTokens": 1200, "estimatedTokens": 20},
            "items": [{"recordRef": "obs:1", "id": 1, "ts": "2026-02-04T13:00:00Z", "kind": "fact", "tool_name": "memory_store", "summary": "plain summary"}],
            "bundle_text": "[GRAPH_CONTEXT v0]\nItems: 1\n\n1) obs:1 :: plain summary\n",
        }

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._graph_index_payload", return_value=fake_index_payload
        ), patch("openclaw_mem.cli._graph_pack_payload", return_value=fake_graph_pack_payload):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertTrue(out["graph"]["triggered"])
        self.assertEqual(out["graph"]["trigger_reason"], "keyword:A+E")
        self.assertTrue(out["trace"]["extensions"]["graph"]["stage1_hit"])
        self.assertEqual(out["trace"]["extensions"]["graph"]["stage1_categories"], ["A", "E"])
        conn.close()

    def test_pack_use_graph_auto_probe_receipts_include_thresholds_and_marginal_count(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {
                1: {
                    "summary": "plain summary",
                    "summary_en": "plain summary",
                    "kind": "fact",
                    "lang": "en",
                }
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args([
            "pack", "--query", "show me the latest dependency status", "--json", "--trace", "--use-graph", "auto", "--graph-scope", "proj-a", "--graph-probe-t-high", "-5", "--graph-probe-t-marginal", "-2", "--graph-probe-n-min", "2"
        ])

        fake_index_payload = {
            "kind": "openclaw-mem.graph.index.v0",
            "budget": {"budgetTokens": 900, "estimatedTokens": 10},
            "top_candidates": [{"recordRef": "obs:1", "id": 1, "ts": "2026-02-04T13:00:00Z", "kind": "fact", "tool_name": "memory_store", "score": -9.0, "title": "stub", "why_relevant": "fts_match"}],
            "suggested_next_expansions": [],
            "index_text": "[GRAPH_INDEX v0]\n",
        }

        fake_graph_pack_payload = {
            "kind": "openclaw-mem.graph.pack.v0",
            "ts": "2026-02-04T13:00:00Z",
            "budget": {"budgetTokens": 1200, "estimatedTokens": 20},
            "items": [{"recordRef": "obs:1", "id": 1, "ts": "2026-02-04T13:00:00Z", "kind": "fact", "tool_name": "memory_store", "summary": "plain summary"}],
            "bundle_text": "[GRAPH_CONTEXT v0]\nItems: 1\n\n1) obs:1 :: plain summary\n",
        }

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._pack_graph_probe_observations",
            return_value={"ran": True, "latency_ms": 7, "hit_count": 3, "best_score": -2.4, "scores": [-2.4, -2.1, -1.9]},
        ), patch("openclaw_mem.cli._graph_index_payload", return_value=fake_index_payload), patch(
            "openclaw_mem.cli._graph_pack_payload", return_value=fake_graph_pack_payload
        ), patch("openclaw_mem.cli._pack_graph_stage1_keywords", return_value={"hit": False, "categories": [], "matched_keywords": []}):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertTrue(out["graph"]["triggered"])
        self.assertEqual(out["graph"]["trigger_reason"], "probe_breadth")
        self.assertEqual(out["trace"]["extensions"]["graph"]["probe"]["threshold_high"], -5.0)
        self.assertEqual(out["trace"]["extensions"]["graph"]["probe"]["threshold_marginal"], -2.0)
        self.assertEqual(out["trace"]["extensions"]["graph"]["probe"]["breadth_min"], 2)
        self.assertEqual(out["trace"]["extensions"]["graph"]["probe"]["marginal_count"], 2)
        conn.close()

    def test_pack_use_graph_auto_probe_receipts_respect_explicit_zero_threshold(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {
                1: {
                    "summary": "plain summary",
                    "summary_en": "plain summary",
                    "kind": "fact",
                    "lang": "en",
                }
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args([
            "pack", "--query", "show me dependency state", "--json", "--trace", "--use-graph", "auto", "--graph-scope", "proj-a", "--graph-probe-t-marginal", "0"
        ])

        fake_index_payload = {
            "kind": "openclaw-mem.graph.index.v0",
            "budget": {"budgetTokens": 900, "estimatedTokens": 10},
            "top_candidates": [{"recordRef": "obs:1", "id": 1, "ts": "2026-02-04T13:00:00Z", "kind": "fact", "tool_name": "memory_store", "score": -9.0, "title": "stub", "why_relevant": "fts_match"}],
            "suggested_next_expansions": [],
            "index_text": "[GRAPH_INDEX v0]\n",
        }

        fake_graph_pack_payload = {
            "kind": "openclaw-mem.graph.pack.v0",
            "ts": "2026-02-04T13:00:00Z",
            "budget": {"budgetTokens": 1200, "estimatedTokens": 20},
            "items": [{"recordRef": "obs:1", "id": 1, "ts": "2026-02-04T13:00:00Z", "kind": "fact", "tool_name": "memory_store", "summary": "plain summary"}],
            "bundle_text": "[GRAPH_CONTEXT v0]\nItems: 1\n\n1) obs:1 :: plain summary\n",
        }

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._pack_graph_probe_observations",
            return_value={"ran": True, "latency_ms": 7, "hit_count": 3, "best_score": -1.0, "scores": [-1.0, 0.0, 1.0]},
        ), patch("openclaw_mem.cli._graph_index_payload", return_value=fake_index_payload), patch(
            "openclaw_mem.cli._graph_pack_payload", return_value=fake_graph_pack_payload
        ), patch("openclaw_mem.cli._pack_graph_stage1_keywords", return_value={"hit": False, "categories": [], "matched_keywords": []}):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["trace"]["extensions"]["graph"]["probe"]["threshold_marginal"], 0.0)
        self.assertEqual(out["trace"]["extensions"]["graph"]["probe"]["marginal_count"], 2)
        conn.close()

    def test_pack_use_graph_auto_latency_gate_degrades_combined_bundle(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {1: {"summary": "plain summary", "summary_en": "plain summary", "kind": "fact", "lang": "en"}},
            "candidate_limit": 12,
        }

        args = build_parser().parse_args([
            "pack", "--query", "latest dependency status", "--json", "--trace", "--use-graph", "auto", "--graph-scope", "proj-a", "--graph-latency-soft-ms", "5", "--graph-latency-hard-ms", "50"
        ])

        fake_index_payload = {
            "kind": "openclaw-mem.graph.index.v0",
            "budget": {"budgetTokens": 900, "estimatedTokens": 10},
            "top_candidates": [{"recordRef": "obs:1", "id": 1, "ts": "2026-02-04T13:00:00Z", "kind": "fact", "tool_name": "memory_store", "score": -9.0, "title": "stub", "why_relevant": "fts_match"}],
            "suggested_next_expansions": [],
            "index_text": "[GRAPH_INDEX v0]\n",
        }

        fake_graph_pack_payload = {
            "kind": "openclaw-mem.graph.pack.v0",
            "ts": "2026-02-04T13:00:00Z",
            "budget": {"budgetTokens": 1200, "estimatedTokens": 20},
            "items": [{"recordRef": "obs:1", "id": 1, "ts": "2026-02-04T13:00:00Z", "kind": "fact", "tool_name": "memory_store", "summary": "plain summary"}],
            "bundle_text": "[GRAPH_CONTEXT v0]\nItems: 1\n\n1) obs:1 :: plain summary\n",
        }

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._graph_index_payload", return_value=fake_index_payload
        ), patch("openclaw_mem.cli._graph_pack_payload", return_value=fake_graph_pack_payload), patch(
            "openclaw_mem.cli._pack_graph_stage1_keywords", return_value={"hit": True, "categories": ["D"], "matched_keywords": ["latest"]}
        ), patch("openclaw_mem.cli.time.perf_counter", side_effect=[100.0, 100.010, 100.030, 100.050]):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["graph"]["latency"]["decision"], "degrade")
        self.assertFalse(out["graph"]["latency"]["compose_graph_bundle"])
        self.assertNotIn("bundle_text_with_graph", out)
        conn.close()

    def test_pack_use_graph_auto_latency_gate_skips_combined_bundle(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {1: {"summary": "plain summary", "summary_en": "plain summary", "kind": "fact", "lang": "en"}},
            "candidate_limit": 12,
        }

        args = build_parser().parse_args([
            "pack", "--query", "latest dependency status", "--json", "--trace", "--use-graph", "auto", "--graph-scope", "proj-a", "--graph-latency-soft-ms", "5", "--graph-latency-hard-ms", "10"
        ])

        fake_index_payload = {
            "kind": "openclaw-mem.graph.index.v0",
            "budget": {"budgetTokens": 900, "estimatedTokens": 10},
            "top_candidates": [{"recordRef": "obs:1", "id": 1, "ts": "2026-02-04T13:00:00Z", "kind": "fact", "tool_name": "memory_store", "score": -9.0, "title": "stub", "why_relevant": "fts_match"}],
            "suggested_next_expansions": [],
            "index_text": "[GRAPH_INDEX v0]\n",
        }

        fake_graph_pack_payload = {
            "kind": "openclaw-mem.graph.pack.v0",
            "ts": "2026-02-04T13:00:00Z",
            "budget": {"budgetTokens": 1200, "estimatedTokens": 20},
            "items": [{"recordRef": "obs:1", "id": 1, "ts": "2026-02-04T13:00:00Z", "kind": "fact", "tool_name": "memory_store", "summary": "plain summary"}],
            "bundle_text": "[GRAPH_CONTEXT v0]\nItems: 1\n\n1) obs:1 :: plain summary\n",
        }

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._graph_index_payload", return_value=fake_index_payload
        ), patch("openclaw_mem.cli._graph_pack_payload", return_value=fake_graph_pack_payload), patch(
            "openclaw_mem.cli._pack_graph_stage1_keywords", return_value={"hit": True, "categories": ["D"], "matched_keywords": ["latest"]}
        ), patch("openclaw_mem.cli.time.perf_counter", side_effect=[200.0, 200.010, 200.040, 200.080]):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["graph"]["latency"]["decision"], "skip")
        self.assertFalse(out["graph"]["latency"]["compose_graph_bundle"])
        self.assertNotIn("bundle_text_with_graph", out)
        conn.close()

    def test_pack_use_graph_auto_latency_gate_respects_explicit_zero_thresholds(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {1: {"summary": "plain summary", "summary_en": "plain summary", "kind": "fact", "lang": "en"}},
            "candidate_limit": 12,
        }

        args = build_parser().parse_args([
            "pack", "--query", "latest dependency status", "--json", "--trace", "--use-graph", "auto", "--graph-scope", "proj-a", "--graph-latency-soft-ms", "0", "--graph-latency-hard-ms", "0"
        ])

        fake_index_payload = {
            "kind": "openclaw-mem.graph.index.v0",
            "budget": {"budgetTokens": 900, "estimatedTokens": 10},
            "top_candidates": [{"recordRef": "obs:1", "id": 1, "ts": "2026-02-04T13:00:00Z", "kind": "fact", "tool_name": "memory_store", "score": -9.0, "title": "stub", "why_relevant": "fts_match"}],
            "suggested_next_expansions": [],
            "index_text": "[GRAPH_INDEX v0]\n",
        }

        fake_graph_pack_payload = {
            "kind": "openclaw-mem.graph.pack.v0",
            "ts": "2026-02-04T13:00:00Z",
            "budget": {"budgetTokens": 1200, "estimatedTokens": 20},
            "items": [{"recordRef": "obs:1", "id": 1, "ts": "2026-02-04T13:00:00Z", "kind": "fact", "tool_name": "memory_store", "summary": "plain summary"}],
            "bundle_text": "[GRAPH_CONTEXT v0]\nItems: 1\n\n1) obs:1 :: plain summary\n",
        }

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._graph_index_payload", return_value=fake_index_payload
        ), patch("openclaw_mem.cli._graph_pack_payload", return_value=fake_graph_pack_payload), patch(
            "openclaw_mem.cli._pack_graph_stage1_keywords", return_value={"hit": True, "categories": ["D"], "matched_keywords": ["latest"]}
        ), patch("openclaw_mem.cli.time.perf_counter", side_effect=[300.0, 300.010, 300.040, 300.080]):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["graph"]["latency"]["threshold_soft_ms"], 0)
        self.assertEqual(out["graph"]["latency"]["threshold_hard_ms"], 0)
        self.assertEqual(out["graph"]["latency"]["decision"], "skip")
        conn.close()

    def test_pack_budget_tokens_clamped_to_minimum_one(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {
                1: {
                    "summary": "a",
                    "summary_en": "a",
                    "kind": "fact",
                    "lang": "en",
                }
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args(["pack", "--query", "x", "--no-json", "--limit", "3", "--budget-tokens", "-5"])

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state):
            with redirect_stdout(buf):
                args.func(conn, args)

        self.assertEqual(buf.getvalue(), "- [obs:1] a\n")
        conn.close()

    def test_pack_policy_prefers_graph_synthesis_ordering(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1, 2],
            "fts_ids": {1, 2},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.91, 2: 0.55},
            "obs_map": {
                1: {"summary": "raw note", "summary_en": "raw note", "kind": "fact", "lang": "en", "tool_name": "memory_store", "ts": "2026-02-04T13:00:00Z"},
                2: {"summary": "synthesis card", "summary_en": "synthesis card", "kind": "note", "lang": "en", "tool_name": "graph.synth-compile", "ts": "2026-02-05T13:00:00Z"},
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args(["pack", "--query", "something", "--json", "--trace", "--limit", "1"])

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._hybrid_prefer_synthesis_cards",
            return_value=(
                [2, 1],
                {"preferredCardRefs": ["obs:2"], "coveredRawRefs": ["obs:1"], "coverageMap": {"obs:2": ["obs:1"]}},
            ),
        ):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["items"][0]["recordRef"], "obs:2")
        self.assertEqual(out["trace"]["extensions"]["policy"]["graph_preferred_card_refs"], ["obs:2"])
        conn.close()

    def test_pack_protected_tail_reserves_budget_and_surfaces_tail_payload(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1],
            "fts_ids": {1},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.77},
            "obs_map": {
                1: {"summary": "M" * 180, "summary_en": "M" * 180, "kind": "fact", "lang": "en", "tool_name": "memory_store", "ts": "2026-02-04T13:00:00Z"}
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args([
            "pack",
            "--query",
            "something",
            "--json",
            "--trace",
            "--limit",
            "2",
            "--budget-tokens",
            "60",
            "--tail-budget-tokens",
            "20",
            "--tail-text",
            "recent user turn",
        ])

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._hybrid_prefer_synthesis_cards",
            return_value=([1], {"preferredCardRefs": [], "coveredRawRefs": [], "coverageMap": {}}),
        ):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["items"][0]["recordRef"], "tail:1")
        self.assertEqual(out["tail"]["includedCount"], 1)
        self.assertEqual(out["trace"]["extensions"]["policy"]["tail_reserved_tokens"], 20)
        self.assertEqual(out["trace"]["extensions"]["policy"]["tail_included_count"], 1)
        conn.close()

    def test_pack_protected_tail_reserves_item_slot_when_budget_reserved(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1, 2],
            "fts_ids": {1, 2},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.9, 2: 0.8},
            "obs_map": {
                1: {"summary": "first memory", "summary_en": "first memory", "kind": "fact", "lang": "en", "tool_name": "memory_store", "ts": "2026-02-04T13:00:00Z"},
                2: {"summary": "second memory", "summary_en": "second memory", "kind": "fact", "lang": "en", "tool_name": "memory_store", "ts": "2026-02-04T13:01:00Z"},
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args([
            "pack",
            "--query",
            "something",
            "--json",
            "--trace",
            "--limit",
            "2",
            "--budget-tokens",
            "120",
            "--tail-budget-tokens",
            "20",
            "--tail-text",
            "recent turn",
        ])

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state), patch(
            "openclaw_mem.cli._hybrid_prefer_synthesis_cards",
            return_value=([1, 2], {"preferredCardRefs": [], "coveredRawRefs": [], "coverageMap": {}}),
        ):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual([item["recordRef"] for item in out["items"]], ["obs:1", "tail:1"])
        self.assertEqual(out["trace"]["extensions"]["policy"]["primary_item_limit"], 1)
        self.assertEqual(out["trace"]["extensions"]["policy"]["tail_reserved_slots"], 1)
        conn.close()

    def test_pack_tail_file_dash_requires_piped_stdin(self):
        conn = _connect(":memory:")
        args = build_parser().parse_args([
            "pack",
            "--query",
            "something",
            "--json",
            "--tail-file",
            "-",
        ])

        class _TtyStdin:
            def isatty(self):
                return True

            def read(self):
                return ""

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("sys.stdin", _TtyStdin()):
            with redirect_stdout(buf):
                with self.assertRaises(SystemExit) as exc:
                    args.func(conn, args)

        self.assertEqual(exc.exception.code, 2)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["error"], "--tail-file - requires piped stdin")
        conn.close()

    def test_pack_trace_empty_candidates_returns_zero_counts(self):
        conn = _connect(":memory:")

        class _FakeEmbedClient:
            def __init__(self, api_key: str, base_url: str = ""):
                pass

            def embed(self, texts, model):
                return [[1.0, 0.0] for _ in texts]

        args = build_parser().parse_args(["pack", "--query", "nothing-matches", "--trace", "--json", "--limit", "5", "--budget-tokens", "1200"])

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._get_api_key", return_value="test-key"), patch("openclaw_mem.cli.OpenAIEmbeddingsClient", _FakeEmbedClient):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertIn("bundle_text", out)
        self.assertEqual(out["bundle_text"], "")
        self.assertEqual(out["items"], [])
        self.assertEqual(out["citations"], [])

        trace = out["trace"]
        self.assertEqual(trace["query"]["text"], "nothing-matches")
        self.assertEqual(trace["output"]["includedCount"], 0)
        self.assertEqual(trace["output"]["excludedCount"], 0)
        self.assertEqual(trace["output"]["citationsCount"], 0)
        self.assertEqual(trace["output"]["refreshedRecordRefs"], [])
        self.assertEqual(trace["candidates"], [])

        conn.close()

    def test_pack_trace_json_shape_and_redaction(self):
        conn = _connect(":memory:")

        sample = json.dumps(
            {
                "ts": "2026-02-04T13:00:00Z",
                "kind": "fact",
                "summary": "test pack summary",
                "summary_en": "test pack summary en",
                "tool_name": "memory_store",
                "detail": {},
            }
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            ingest_args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, ingest_args)
        finally:
            sys.stdin = old_stdin

        from openclaw_mem.vector import pack_f32, l2_norm

        conn.execute(
            "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (1, "text-embedding-3-small", 2, pack_f32([1.0, 0.0]), l2_norm([1.0, 0.0]), "2026-02-05T00:00:00Z"),
        )
        conn.commit()

        class _FakeEmbedClient:
            def __init__(self, api_key: str, base_url: str = ""):
                pass

            def embed(self, texts, model):
                return [[1.0, 0.0] for _ in texts]

        args = build_parser().parse_args(["pack", "--query", "test", "--trace", "--json"])

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._get_api_key", return_value="test-key"), patch("openclaw_mem.cli.OpenAIEmbeddingsClient", _FakeEmbedClient):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertIn("bundle_text", out)
        self.assertIn("items", out)
        self.assertIn("citations", out)
        self.assertIn("trace", out)
        self.assertEqual(out["trace"]["kind"], "openclaw-mem.pack.trace.v1")
        self.assertIn("ts", out["trace"])
        self.assertIn("T", out["trace"]["ts"])
        self.assertTrue(out["trace"]["ts"].endswith("+00:00"))
        self.assertEqual(out["trace"]["version"]["schema"], "v1")
        self.assertRegex(out["trace"]["version"]["openclaw_mem"], r"^\d+\.\d+\.\d+")
        self.assertEqual(out["trace"]["query"]["text"], "test")
        self.assertIn("output", out["trace"])
        self.assertEqual(out["trace"]["budgets"]["maxL2Items"], 0)
        self.assertEqual(out["trace"]["budgets"]["niceCap"], 100)
        self.assertEqual(out["trace"]["output"]["includedCount"], 1)
        self.assertEqual(out["trace"]["output"]["excludedCount"], 0)
        self.assertEqual(out["trace"]["output"]["citationsCount"], 1)
        self.assertEqual(out["trace"]["output"]["refreshedRecordRefs"], ["obs:1"])
        self.assertIn("coverage", out["trace"]["output"])
        self.assertEqual(out["trace"]["output"]["coverage"]["rationaleMissingCount"], 0)
        self.assertEqual(out["trace"]["output"]["coverage"]["citationMissingCount"], 0)

        candidate = out["trace"]["candidates"][0]
        self.assertIn("caps", candidate["decision"])
        self.assertFalse(candidate["decision"]["caps"]["niceCapHit"])
        self.assertFalse(candidate["decision"]["caps"]["l2CapHit"])
        self.assertEqual(candidate["decision"]["rationale"], candidate["decision"]["reason"])
        self.assertIsNone(candidate["citations"]["url"])

        self.assertEqual(
            set(out["trace"].keys()),
            {
                "kind",
                "ts",
                "version",
                "query",
                "budgets",
                "lanes",
                "candidates",
                "output",
                "timing",
                "extensions",
            },
        )

        for lane in out["trace"]["lanes"]:
            self.assertIn("retrievers", lane)
            self.assertIsInstance(lane["retrievers"], list)

        self.assertIn("reason", candidate["decision"])
        self.assertIsInstance(candidate["decision"]["reason"], list)
        self.assertGreater(len(candidate["decision"]["reason"]), 0)

        trace_dump = json.dumps(out["trace"], ensure_ascii=False)
        # Trace should be redaction-safe (no raw memory text or local absolute paths).
        self.assertNotIn("test pack summary", trace_dump)
        self.assertNotIn("/root/", trace_dump)
        self.assertNotIn("/home/", trace_dump)
        self.assertNotIn("/Users/", trace_dump)
        self.assertNotRegex(trace_dump, r"[A-Za-z]:\\\\")
        conn.close()

    def test_pack_trace_stable_schema_contract(self):
        conn = _connect(":memory:")

        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": "2026-02-04T13:00:00Z",
                        "kind": "fact",
                        "summary": "stability sample one",
                        "summary_en": "stability sample one",
                        "tool_name": "memory_store",
                        "detail": {},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:01:00Z",
                        "kind": "fact",
                        "summary": "stability sample two",
                        "summary_en": "stability sample two",
                        "tool_name": "memory_store",
                        "detail": {},
                    }
                ),
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            ingest_args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, ingest_args)
        finally:
            sys.stdin = old_stdin

        from openclaw_mem.vector import pack_f32, l2_norm

        for obs_id in (1, 2):
            conn.execute(
                "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
                (obs_id, "text-embedding-3-small", 2, pack_f32([1.0, 0.0]), l2_norm([1.0, 0.0]), "2026-02-05T00:00:00Z"),
            )
        conn.commit()

        class _FakeEmbedClient:
            def __init__(self, api_key: str, base_url: str = ""):
                pass

            def embed(self, texts, model):
                return [[1.0, 0.0] for _ in texts]

        args = build_parser().parse_args(["pack", "--query", "stability", "--query-en", "stability", "--trace", "--json", "--limit", "2", "--budget-tokens", "150"])

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._get_api_key", return_value="test-key"), patch("openclaw_mem.cli.OpenAIEmbeddingsClient", _FakeEmbedClient):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        trace = out["trace"]

        self.assertEqual(trace["kind"], "openclaw-mem.pack.trace.v1")
        self.assertIn("ts", trace)
        self.assertIn("version", trace)
        self.assertEqual(trace["version"]["schema"], "v1")
        self.assertIn("query", trace)
        self.assertIn("text", trace["query"])
        self.assertIn("scope", trace["query"])
        self.assertIn("intent", trace["query"])
        self.assertIn("budgets", trace)
        self.assertEqual(trace["budgets"]["maxItems"], 2)
        self.assertEqual(trace["budgets"]["maxL2Items"], 0)
        self.assertEqual(trace["budgets"]["niceCap"], 100)

        lane_names = [lane["name"] for lane in trace["lanes"]]
        self.assertEqual(lane_names, ["hot", "warm", "cold"])
        warm = next(lane for lane in trace["lanes"] if lane["name"] == "warm")
        self.assertEqual(warm["source"], "sqlite-observations")
        self.assertTrue(warm["searched"])
        self.assertEqual([r["kind"] for r in warm["retrievers"]], ["fts5", "vector", "rrf"])

        self.assertIn("candidates", trace)
        self.assertGreater(len(trace["candidates"]), 0)
        candidate = trace["candidates"][0]
        for key in ["id", "layer", "importance", "trust", "scores", "decision", "citations"]:
            self.assertIn(key, candidate)
        self.assertIn("niceCapHit", candidate["decision"]["caps"])
        self.assertIn("l2CapHit", candidate["decision"]["caps"])
        self.assertIn("rationale", candidate["decision"])
        self.assertEqual(candidate["decision"]["rationale"], candidate["decision"]["reason"])
        self.assertIn("output", trace)
        self.assertIn("includedCount", trace["output"])
        self.assertIn("refreshedRecordRefs", trace["output"])
        self.assertIn("coverage", trace["output"])
        self.assertIn("allIncludedHaveRationale", trace["output"]["coverage"])
        self.assertIn("allIncludedHaveCitations", trace["output"]["coverage"])
        self.assertIsInstance(trace["output"]["refreshedRecordRefs"], list)

        conn.close()

    def test_pack_trace_candidate_uses_importance_and_trust_from_detail_json(self):
        conn = _connect(":memory:")

        sample = json.dumps(
            {
                "ts": "2026-02-04T13:00:00Z",
                "kind": "fact",
                "summary": "trust aware memory item",
                "summary_en": "trust aware memory item",
                "tool_name": "memory_store",
                "detail": {
                    "importance": {"label": "must_remember", "score": 0.91},
                    "trust_tier": "quarantine",
                },
            }
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            ingest_args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, ingest_args)
        finally:
            sys.stdin = old_stdin

        from openclaw_mem.vector import pack_f32, l2_norm

        conn.execute(
            "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (1, "text-embedding-3-small", 2, pack_f32([1.0, 0.0]), l2_norm([1.0, 0.0]), "2026-02-05T00:00:00Z"),
        )
        conn.commit()

        class _FakeEmbedClient:
            def __init__(self, api_key: str, base_url: str = ""):
                pass

            def embed(self, texts, model):
                return [[1.0, 0.0] for _ in texts]

        args = build_parser().parse_args(["pack", "--query", "trust", "--trace", "--json"])

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._get_api_key", return_value="test-key"), patch("openclaw_mem.cli.OpenAIEmbeddingsClient", _FakeEmbedClient):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        candidate = out["trace"]["candidates"][0]
        self.assertEqual(candidate["importance"], "must_remember")
        self.assertEqual(candidate["trust"], "quarantined")
        conn.close()

    def test_pack_trace_candidate_uses_unknown_when_detail_labels_are_invalid(self):
        conn = _connect(":memory:")

        sample = json.dumps(
            {
                "ts": "2026-02-04T13:00:00Z",
                "kind": "fact",
                "summary": "unknown label sample",
                "summary_en": "unknown label sample",
                "tool_name": "memory_store",
                "detail": {
                    "importance": {"label": "someday_maybe"},
                    "provenance": {"trust": "semi_trusted"},
                },
            }
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            ingest_args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, ingest_args)
        finally:
            sys.stdin = old_stdin

        from openclaw_mem.vector import pack_f32, l2_norm

        conn.execute(
            "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (1, "text-embedding-3-small", 2, pack_f32([1.0, 0.0]), l2_norm([1.0, 0.0]), "2026-02-05T00:00:00Z"),
        )
        conn.commit()

        class _FakeEmbedClient:
            def __init__(self, api_key: str, base_url: str = ""):
                pass

            def embed(self, texts, model):
                return [[1.0, 0.0] for _ in texts]

        args = build_parser().parse_args(["pack", "--query", "unknown", "--trace", "--json"])

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._get_api_key", return_value="test-key"), patch("openclaw_mem.cli.OpenAIEmbeddingsClient", _FakeEmbedClient):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        candidate = out["trace"]["candidates"][0]
        self.assertEqual(candidate["importance"], "unknown")
        self.assertEqual(candidate["trust"], "unknown")
        conn.close()

    def test_pack_trace_output_counts_with_exclusions(self):
        conn = _connect(":memory:")

        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": "2026-02-04T13:00:00Z",
                        "kind": "fact",
                        "summary": "test pack alpha",
                        "summary_en": "test pack alpha",
                        "tool_name": "memory_store",
                        "detail": {},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:01:00Z",
                        "kind": "fact",
                        "summary": "test pack beta",
                        "summary_en": "test pack beta",
                        "tool_name": "memory_store",
                        "detail": {},
                    }
                ),
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            ingest_args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, ingest_args)
        finally:
            sys.stdin = old_stdin

        from openclaw_mem.vector import pack_f32, l2_norm

        conn.execute(
            "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (1, "text-embedding-3-small", 2, pack_f32([1.0, 0.0]), l2_norm([1.0, 0.0]), "2026-02-05T00:00:00Z"),
        )
        conn.execute(
            "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (2, "text-embedding-3-small", 2, pack_f32([1.0, 0.0]), l2_norm([1.0, 0.0]), "2026-02-05T00:00:00Z"),
        )
        conn.commit()

        class _FakeEmbedClient:
            def __init__(self, api_key: str, base_url: str = ""):
                pass

            def embed(self, texts, model):
                return [[1.0, 0.0] for _ in texts]

        args = build_parser().parse_args(["pack", "--query", "test", "--trace", "--json", "--limit", "1"])

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._get_api_key", return_value="test-key"), patch("openclaw_mem.cli.OpenAIEmbeddingsClient", _FakeEmbedClient):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        trace = out["trace"]
        self.assertEqual(trace["output"]["includedCount"], 1)
        self.assertGreaterEqual(trace["output"]["excludedCount"], 1)
        self.assertEqual(trace["output"]["includedCount"] + trace["output"]["excludedCount"], len(trace["candidates"]))
        self.assertEqual(trace["output"]["refreshedRecordRefs"], [out["items"][0]["recordRef"]])
        conn.close()

    def test_pack_trace_reason_keys_are_known(self):
        conn = _connect(":memory:")

        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": "2026-02-04T13:00:00Z",
                        "kind": "fact",
                        "summary": "trace reason sample one",
                        "summary_en": "trace reason sample one",
                        "tool_name": "memory_store",
                        "detail": {},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:01:00Z",
                        "kind": "fact",
                        "summary": "",
                        "summary_en": "",
                        "tool_name": "memory_store",
                        "detail": {},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:02:00Z",
                        "kind": "fact",
                        "summary": "trace reason sample three",
                        "summary_en": "trace reason sample three",
                        "tool_name": "memory_store",
                        "detail": {},
                    }
                ),
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            ingest_args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, ingest_args)
        finally:
            sys.stdin = old_stdin

        from openclaw_mem.vector import pack_f32, l2_norm

        for obs_id in (1, 2, 3):
            conn.execute(
                "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
                (obs_id, "text-embedding-3-small", 2, pack_f32([1.0, 0.0]), l2_norm([1.0, 0.0]), "2026-02-05T00:00:00Z"),
            )
        conn.commit()

        class _FakeEmbedClient:
            def __init__(self, api_key: str, base_url: str = ""):
                pass

            def embed(self, texts, model):
                return [[1.0, 0.0] for _ in texts]

        args = build_parser().parse_args(["pack", "--query", "trace", "--trace", "--json", "--limit", "1"])

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._get_api_key", return_value="test-key"), patch("openclaw_mem.cli.OpenAIEmbeddingsClient", _FakeEmbedClient):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        allowed = {
            "missing_row",
            "missing_summary",
            "max_items_reached",
            "budget_tokens_exceeded",
            "within_item_limit",
            "within_budget",
            "matched_fts",
            "matched_vector",
        }
        observed_reasons = set()
        for candidate in out["trace"]["candidates"]:
            reasons = candidate["decision"].get("reason", [])
            reason_set = set(reasons)
            self.assertTrue(reasons, f"candidate {candidate.get('id')} has no decision reasons")
            self.assertLessEqual(reason_set - allowed, set())
            observed_reasons |= reason_set

        self.assertIn("missing_summary", observed_reasons)
        self.assertIn("within_budget", observed_reasons)
        self.assertIn("within_item_limit", observed_reasons)
        conn.close()

    def test_pack_respects_budget_tokens(self):
        conn = _connect(":memory:")

        sample = "\n".join(
            [
                json.dumps(
                    {
                        "ts": "2026-02-04T13:00:00Z",
                        "kind": "fact",
                        "summary": "ping",
                        "summary_en": "ping",
                        "tool_name": "memory_store",
                        "detail": {},
                    }
                ),
                json.dumps(
                    {
                        "ts": "2026-02-04T13:01:00Z",
                        "kind": "fact",
                        "summary": "ping this is a much longer summary that should never fit into the tiny budget",
                        "summary_en": "ping this is a much longer summary that should never fit into the tiny budget",
                        "tool_name": "memory_store",
                        "detail": {},
                    }
                ),
            ]
        )

        old_stdin = sys.stdin
        try:
            sys.stdin = io.StringIO(sample)
            ingest_args = type("Args", (), {"file": None, "json": True})()
            with redirect_stdout(io.StringIO()):
                cmd_ingest(conn, ingest_args)
        finally:
            sys.stdin = old_stdin

        from openclaw_mem.vector import pack_f32, l2_norm

        conn.execute(
            "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (1, "text-embedding-3-small", 2, pack_f32([1.0, 0.0]), l2_norm([1.0, 0.0]), "2026-02-05T00:00:00Z"),
        )
        conn.execute(
            "INSERT INTO observation_embeddings (observation_id, model, dim, vector, norm, created_at) VALUES (?, ?, ?, ?, ?, ?)",
            (2, "text-embedding-3-small", 2, pack_f32([0.0, 1.0]), l2_norm([0.0, 1.0]), "2026-02-05T00:00:00Z"),
        )
        conn.commit()

        class _FakeEmbedClient:
            def __init__(self, api_key: str, base_url: str = ""):
                pass

            def embed(self, texts, model):
                return [[1.0, 0.0] for _ in texts]

        args = build_parser().parse_args(["pack", "--query", "ping", "--trace", "--json", "--limit", "2", "--budget-tokens", "1"])

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._get_api_key", return_value="test-key"), patch("openclaw_mem.cli.OpenAIEmbeddingsClient", _FakeEmbedClient):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["items"][0]["id"], 1)
        self.assertLessEqual(len(out["items"]), 1)
        self.assertEqual(out["trace"]["output"]["includedCount"], 1)
        self.assertEqual(out["trace"]["output"]["excludedCount"], 1)
        self.assertEqual(out["trace"]["candidates"][1]["id"], "obs:2")
        self.assertEqual(out["trace"]["candidates"][1]["decision"]["reason"], ["budget_tokens_exceeded"])
        self.assertIn("within_item_limit", out["trace"]["candidates"][0]["decision"]["reason"])
        self.assertIn("within_budget", out["trace"]["candidates"][0]["decision"]["reason"])
        conn.close()

    def test_pack_trace_records_max_items_reached_reason(self):
        conn = _connect(":memory:")

        pack_state = {
            "ordered_ids": [1, 2],
            "fts_ids": {1, 2},
            "vec_ids": set(),
            "vec_en_ids": set(),
            "rrf_scores": {1: 0.9, 2: 0.8},
            "obs_map": {
                1: {
                    "summary": "first short summary",
                    "summary_en": "first short summary",
                    "kind": "fact",
                    "lang": "en",
                },
                2: {
                    "summary": "second short summary",
                    "summary_en": "second short summary",
                    "kind": "fact",
                    "lang": "en",
                },
            },
            "candidate_limit": 12,
        }

        args = build_parser().parse_args(["pack", "--query", "short", "--trace", "--json", "--limit", "1"])

        from unittest.mock import patch

        buf = io.StringIO()
        with patch("openclaw_mem.cli._hybrid_retrieve", return_value=pack_state):
            with redirect_stdout(buf):
                args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(len(out["items"]), 1)
        self.assertEqual(out["items"][0]["recordRef"], "obs:1")

        candidates = out["trace"]["candidates"]
        self.assertEqual(len(candidates), 2)
        self.assertTrue(candidates[0]["decision"]["included"])
        self.assertIn("within_item_limit", candidates[0]["decision"]["reason"])
        self.assertFalse(candidates[1]["decision"]["included"])
        self.assertEqual(candidates[1]["decision"]["reason"], ["max_items_reached"])
        self.assertEqual(out["trace"]["output"]["includedCount"], 1)
        self.assertEqual(out["trace"]["output"]["excludedCount"], 1)
        conn.close()

    def test_triage_cron_errors_coerces_numeric_strings(self):
        import tempfile

        # Fake cron jobs store: timestamps/durations may be stored as strings.
        jobs = {
            "jobs": [
                {
                    "id": "job1",
                    "name": "Job 1",
                    "enabled": True,
                    "state": {"lastStatus": "ok", "lastRunAtMs": "9999999999999"},
                },
                {
                    "id": "job2",
                    "name": "Job 2",
                    "enabled": True,
                    "state": {
                        "lastStatus": "error",
                        "lastRunAtMs": "9999999999999",
                        "lastDurationMs": "1234",
                        "nextRunAtMs": "9999999999999",
                        "lastError": "boom first line\nboom second line",
                    },
                },
            ]
        }
        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as tmp:
            json.dump(jobs, tmp)
            tmp_path = tmp.name

        conn = _connect(":memory:")
        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", delete=False) as st:
            state_path = st.name

        args = type(
            "Args",
            (),
            {
                "mode": "cron-errors",
                "since_minutes": 60,
                "limit": 10,
                "keywords": None,
                "cron_jobs_path": tmp_path,
                "tasks_since_minutes": 1440,
                "importance_min": 0.7,
                "state_path": state_path,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            with self.assertRaises(SystemExit) as cm:
                cmd_triage(conn, args)
        self.assertEqual(cm.exception.code, 10)
        out = json.loads(buf.getvalue())

        self.assertEqual(out["cron"]["found_new"], 1)
        match = out["cron"]["matches"][0]
        self.assertEqual(match["id"], "job2")
        self.assertEqual(match["lastRunAtMs"], 9999999999999)
        self.assertEqual(match["lastDurationMs"], 1234)
        self.assertEqual(match["nextRunAtMs"], 9999999999999)
        self.assertEqual(match["lastErrorLine"], "boom first line")

        conn.close()


class TestGraphSynthesisCli(unittest.TestCase):
    def test_graph_synth_compile_explicit_refs_stores_card(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {"kind": "note", "summary": "alpha source", "tool_name": "memory_store", "detail": {"scope": "proj.a"}})
        _insert_observation(conn, {"kind": "note", "summary": "beta source", "tool_name": "memory_store", "detail": {"scope": "proj.a"}})

        args = build_parser().parse_args([
            "graph", "--json", "synth", "compile",
            "--record-ref", "obs:1",
            "--record-ref", "obs:2",
            "--title", "Merged finding",
            "--summary", "Merged finding summary",
            "--why-it-matters", "Use this later"
        ])

        buf = io.StringIO()
        with redirect_stdout(buf):
            args.func(conn, args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["kind"], "openclaw-mem.graph.synth.compile.v0")
        self.assertEqual(out["result"]["sourceRefs"], ["obs:1", "obs:2"])
        self.assertEqual(out["result"]["title"], "Merged finding")

        card_id = int(out["cardRef"].split(":", 1)[1])
        row = conn.execute("SELECT summary, tool_name, detail_json FROM observations WHERE id = ?", (card_id,)).fetchone()
        self.assertEqual(row["summary"], "Merged finding summary")
        self.assertEqual(row["tool_name"], "graph.synth-compile")
        detail = json.loads(row["detail_json"] or "{}")
        synth = detail["graph_synthesis"]
        self.assertEqual(synth["source_refs"], ["obs:1", "obs:2"])
        self.assertEqual(synth["status"], "fresh")
        conn.close()

    def test_graph_synth_compile_persists_card_on_file_backed_db(self):
        with tempfile.TemporaryDirectory() as td:
            db_path = os.path.join(td, "mem.sqlite")
            conn = _connect(db_path)
            _insert_observation(conn, {"kind": "note", "summary": "alpha source", "tool_name": "memory_store", "detail": {"scope": "proj.a"}})
            _insert_observation(conn, {"kind": "note", "summary": "beta source", "tool_name": "memory_store", "detail": {"scope": "proj.a"}})

            args = build_parser().parse_args([
                "--db", db_path,
                "graph", "--json", "synth", "compile",
                "--record-ref", "obs:1",
                "--record-ref", "obs:2",
                "--title", "Persisted finding",
                "--summary", "Persisted finding summary",
            ])
            buf = io.StringIO()
            with redirect_stdout(buf):
                args.func(conn, args)
            card_ref = json.loads(buf.getvalue())["cardRef"]
            conn.close()

            check = sqlite3.connect(db_path)
            check.row_factory = sqlite3.Row
            try:
                row = check.execute(
                    "SELECT summary, tool_name FROM observations WHERE id = ?",
                    (int(card_ref.split(":", 1)[1]),),
                ).fetchone()
                self.assertIsNotNone(row)
                self.assertEqual(row["summary"], "Persisted finding summary")
                self.assertEqual(row["tool_name"], "graph.synth-compile")
            finally:
                check.close()

    def test_graph_synth_refresh_persists_new_card_and_superseded_status_on_file_backed_db(self):
        with tempfile.TemporaryDirectory() as td:
            db_path = os.path.join(td, "mem.sqlite")
            conn = _connect(db_path)
            _insert_observation(conn, {"kind": "note", "summary": "alpha source", "tool_name": "memory_store", "detail": {}})

            compile_args = build_parser().parse_args([
                "--db", db_path,
                "graph", "--json", "synth", "compile",
                "--query", "alpha",
                "--title", "Alpha synthesis",
                "--summary", "Alpha synthesis",
            ])
            buf = io.StringIO()
            with redirect_stdout(buf):
                compile_args.func(conn, compile_args)
            card_ref = json.loads(buf.getvalue())["cardRef"]
            _insert_observation(conn, {"kind": "note", "summary": "alpha newer source", "tool_name": "memory_store", "detail": {}})

            refresh_args = build_parser().parse_args(["--db", db_path, "graph", "--json", "synth", "refresh", card_ref])
            refresh_buf = io.StringIO()
            with redirect_stdout(refresh_buf):
                refresh_args.func(conn, refresh_args)
            new_ref = json.loads(refresh_buf.getvalue())["result"]["recordRef"]
            conn.close()

            check = sqlite3.connect(db_path)
            check.row_factory = sqlite3.Row
            try:
                new_row = check.execute(
                    "SELECT tool_name FROM observations WHERE id = ?",
                    (int(new_ref.split(":", 1)[1]),),
                ).fetchone()
                self.assertIsNotNone(new_row)
                self.assertEqual(new_row["tool_name"], "graph.synth-compile")
                old_row = check.execute(
                    "SELECT detail_json FROM observations WHERE id = ?",
                    (int(card_ref.split(":", 1)[1]),),
                ).fetchone()
                self.assertIsNotNone(old_row)
                old_detail = json.loads(old_row["detail_json"] or "{}")
                self.assertEqual(old_detail["graph_synthesis"]["status"], "superseded")
                self.assertEqual(old_detail["graph_synthesis"]["superseded_by"], new_ref)
            finally:
                check.close()

    def test_graph_synth_stale_detects_query_selection_drift(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {"kind": "note", "summary": "alpha source", "tool_name": "memory_store", "detail": {}})

        args = build_parser().parse_args([
            "graph", "--json", "synth", "compile",
            "--query", "alpha",
            "--title", "Alpha synthesis",
            "--summary", "Alpha synthesis"
        ])
        buf = io.StringIO()
        with redirect_stdout(buf):
            args.func(conn, args)
        compiled = json.loads(buf.getvalue())
        card_ref = compiled["cardRef"]

        _insert_observation(conn, {"kind": "note", "summary": "alpha newer source", "tool_name": "memory_store", "detail": {}})

        stale_args = build_parser().parse_args(["graph", "--json", "synth", "stale", card_ref])
        stale_buf = io.StringIO()
        with redirect_stdout(stale_buf):
            stale_args.func(conn, stale_args)

        out = json.loads(stale_buf.getvalue())
        self.assertEqual(out["kind"], "openclaw-mem.graph.synth.stale.v0")
        self.assertEqual(out["count"], 1)
        self.assertEqual(out["items"][0]["status"], "stale")
        self.assertIn("selection_drift", out["items"][0]["reasons"])
        conn.close()

    def test_graph_synth_refresh_supersedes_stale_card(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {"kind": "note", "summary": "alpha source", "tool_name": "memory_store", "detail": {}})

        args = build_parser().parse_args([
            "graph", "--json", "synth", "compile",
            "--query", "alpha",
            "--title", "Alpha synthesis",
            "--summary", "Alpha synthesis",
        ])
        buf = io.StringIO()
        with redirect_stdout(buf):
            args.func(conn, args)
        card_ref = json.loads(buf.getvalue())["cardRef"]

        _insert_observation(conn, {"kind": "note", "summary": "alpha newer source", "tool_name": "memory_store", "detail": {}})

        refresh_args = build_parser().parse_args(["graph", "--json", "synth", "refresh", card_ref])
        refresh_buf = io.StringIO()
        with redirect_stdout(refresh_buf):
            refresh_args.func(conn, refresh_args)

        out = json.loads(refresh_buf.getvalue())
        self.assertEqual(out["kind"], "openclaw-mem.graph.synth.refresh.v0")
        self.assertTrue(out["result"]["refreshed"])
        self.assertEqual(out["result"]["supersededCardRef"], card_ref)
        self.assertNotEqual(out["result"]["recordRef"], card_ref)
        self.assertIn("obs:3", out["result"]["sourceRefs"])

        old_id = int(card_ref.split(":", 1)[1])
        old_row = conn.execute("SELECT detail_json FROM observations WHERE id = ?", (old_id,)).fetchone()
        old_detail = json.loads(old_row["detail_json"] or "{}")
        old_synth = old_detail["graph_synthesis"]
        self.assertEqual(old_synth["status"], "superseded")
        self.assertEqual(old_synth["superseded_by"], out["result"]["recordRef"])

        stale_args = build_parser().parse_args(["graph", "--json", "synth", "stale", card_ref])
        stale_buf = io.StringIO()
        with redirect_stdout(stale_buf):
            stale_args.func(conn, stale_args)
        stale_out = json.loads(stale_buf.getvalue())
        self.assertEqual(stale_out["items"][0]["status"], "superseded")
        self.assertEqual(stale_out["items"][0]["supersededBy"], out["result"]["recordRef"])
        conn.close()

    def test_graph_synth_refresh_fresh_card_is_noop_without_force(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {"kind": "note", "summary": "alpha source", "tool_name": "memory_store", "detail": {}})

        args = build_parser().parse_args([
            "graph", "--json", "synth", "compile",
            "--query", "alpha",
            "--title", "Alpha synthesis",
            "--summary", "Alpha synthesis",
        ])
        buf = io.StringIO()
        with redirect_stdout(buf):
            args.func(conn, args)
        card_ref = json.loads(buf.getvalue())["cardRef"]

        refresh_args = build_parser().parse_args(["graph", "--json", "synth", "refresh", card_ref])
        refresh_buf = io.StringIO()
        with redirect_stdout(refresh_buf):
            refresh_args.func(conn, refresh_args)

        out = json.loads(refresh_buf.getvalue())
        self.assertEqual(out["kind"], "openclaw-mem.graph.synth.refresh.v0")
        self.assertFalse(out["result"]["refreshed"])
        self.assertEqual(out["result"]["reason"], "already_fresh")

        count = conn.execute("SELECT COUNT(*) AS n FROM observations WHERE tool_name = 'graph.synth-compile'").fetchone()["n"]
        self.assertEqual(count, 1)
        conn.close()

    def test_graph_synth_stale_surfaces_review_and_contradiction_signals(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {"kind": "note", "summary": "alpha source", "tool_name": "memory_store", "detail": {}})

        args = build_parser().parse_args([
            "graph", "--json", "synth", "compile",
            "--query", "alpha",
            "--title", "Alpha synthesis",
            "--summary", "Alpha synthesis",
        ])
        with redirect_stdout(io.StringIO()) as _:
            args.func(conn, args)
        card_ref = json.loads(_.getvalue())["cardRef"]

        _insert_observation(conn, {"kind": "note", "summary": "alpha contradiction: replaced old flow", "tool_name": "memory_store", "detail": {}})

        stale_args = build_parser().parse_args(["graph", "--json", "synth", "stale", card_ref])
        stale_buf = io.StringIO()
        with redirect_stdout(stale_buf):
            stale_args.func(conn, stale_args)

        out = json.loads(stale_buf.getvalue())
        item = out["items"][0]
        self.assertEqual(item["status"], "stale")
        self.assertIn("selection_drift", item["reasons"])
        self.assertGreaterEqual(item["reviewSignalCount"], 1)
        self.assertGreaterEqual(item["contradictionSignalCount"], 1)
        self.assertTrue(any(sig["kind"] == "contradiction_keyword" for sig in item["reviewSignals"]))
        conn.close()

    def test_graph_lint_reports_stale_and_unreferenced_capture_rows(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {"kind": "note", "summary": "alpha captured", "tool_name": "graph.capture-md", "detail": {}})

        args = build_parser().parse_args([
            "graph", "--json", "synth", "compile",
            "--query", "alpha",
            "--title", "Alpha synthesis",
            "--summary", "Alpha synthesis"
        ])
        with redirect_stdout(io.StringIO()):
            args.func(conn, args)

        _insert_observation(conn, {"kind": "note", "summary": "alpha new capture", "tool_name": "graph.capture-git", "detail": {}})

        lint_args = build_parser().parse_args(["graph", "--json", "lint"])
        buf = io.StringIO()
        with redirect_stdout(buf):
            lint_args.func(conn, lint_args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["kind"], "openclaw-mem.graph.lint.v0")
        self.assertEqual(out["counts"]["synthesisCards"], 1)
        self.assertEqual(out["counts"]["staleCards"], 1)
        self.assertGreaterEqual(out["counts"]["reviewCards"], 0)
        self.assertGreaterEqual(out["counts"]["contradictionSignalCards"], 0)
        self.assertGreaterEqual(out["counts"]["unreferencedCaptureRows"], 1)
        self.assertTrue(any(item["recordRef"] == "obs:3" for item in out["samples"]["unreferencedCaptureRows"]))
        conn.close()

    def test_graph_lint_suggests_candidate_cards_for_uncovered_scope_clusters(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {"kind": "note", "summary": "rollout checkpoint ready", "tool_name": "memory_store", "detail": {"scope": "proj.alpha"}})
        _insert_observation(conn, {"kind": "note", "summary": "rollout checkpoint blocked", "tool_name": "memory_store", "detail": {"scope": "proj.alpha"}})
        _insert_observation(conn, {"kind": "note", "summary": "rollback guard pending", "tool_name": "memory_store", "detail": {"scope": "proj.alpha"}})
        _insert_observation(conn, {"kind": "note", "summary": "rollback guard approved", "tool_name": "memory_store", "detail": {"scope": "proj.alpha"}})

        lint_args = build_parser().parse_args(["graph", "--json", "lint"])
        buf = io.StringIO()
        with redirect_stdout(buf):
            lint_args.func(conn, lint_args)

        out = json.loads(buf.getvalue())
        self.assertEqual(out["counts"]["candidateCardSuggestions"], 2)
        self.assertEqual(out["counts"]["scopedSourceRows"], 4)
        self.assertEqual(out["counts"]["uncoveredScopedSourceRows"], 4)
        suggestions = out["samples"]["candidateCardSuggestions"]
        self.assertEqual(suggestions[0]["scope"], "proj.alpha")
        self.assertEqual(suggestions[0]["reason"], "uncovered_scope_term_cluster")
        self.assertEqual(suggestions[0]["sharedTerms"], ["checkpoint"])
        self.assertEqual(suggestions[0]["recordRefs"], ["obs:1", "obs:2"])
        self.assertEqual(suggestions[1]["sharedTerms"], ["rollback"])
        self.assertEqual(suggestions[1]["recordRefs"], ["obs:3", "obs:4"])
        conn.close()

    def test_graph_lint_prefers_keyword_clusters_before_scope_fallback(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {"kind": "note", "summary": "pricing migration ready", "tool_name": "memory_store", "detail": {"scope": "proj.gamma"}})
        _insert_observation(conn, {"kind": "note", "summary": "pricing migration blocked", "tool_name": "memory_store", "detail": {"scope": "proj.gamma"}})
        _insert_observation(conn, {"kind": "note", "summary": "audit followup", "tool_name": "memory_store", "detail": {"scope": "proj.gamma"}})

        lint_args = build_parser().parse_args(["graph", "--json", "lint"])
        buf = io.StringIO()
        with redirect_stdout(buf):
            lint_args.func(conn, lint_args)

        out = json.loads(buf.getvalue())
        suggestion = out["samples"]["candidateCardSuggestions"][0]
        self.assertEqual(suggestion["reason"], "uncovered_scope_term_cluster")
        self.assertEqual(suggestion["sharedTerms"], ["migration"])
        self.assertEqual(suggestion["recordRefs"], ["obs:1", "obs:2"])
        conn.close()

    def test_graph_synth_recommend_recommends_refresh_for_stale_card(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {"kind": "note", "summary": "alpha captured", "tool_name": "graph.capture-md", "detail": {}})

        compile_args = build_parser().parse_args([
            "graph", "synth", "compile",
            "--query", "alpha",
            "--title", "Alpha synthesis",
            "--summary", "Alpha synthesis",
        ])
        with redirect_stdout(io.StringIO()):
            compile_args.func(conn, compile_args)

        _insert_observation(conn, {"kind": "note", "summary": "alpha new capture", "tool_name": "graph.capture-git", "detail": {}})
        before_count = conn.execute("SELECT COUNT(*) AS n FROM observations").fetchone()["n"]

        args = build_parser().parse_args(["graph", "synth", "recommend"])
        buf = io.StringIO()
        with redirect_stdout(buf):
            args.func(conn, args)

        after_count = conn.execute("SELECT COUNT(*) AS n FROM observations").fetchone()["n"]
        out = json.loads(buf.getvalue())
        self.assertEqual(out["kind"], "openclaw-mem.graph.synth.recommend.v0")
        self.assertGreaterEqual(out["counts"]["refreshSynthesis"], 1)
        refresh_items = [item for item in out["items"] if item["action"] == "refresh_card"]
        self.assertTrue(refresh_items)
        self.assertEqual(refresh_items[0]["target"]["recordRef"], "obs:2")
        self.assertIn("graph synth refresh obs:2", refresh_items[0]["suggestion"]["command"])
        self.assertEqual(before_count, after_count)
        conn.close()

    def test_graph_synth_recommend_recommends_compile_for_uncovered_scope_cluster(self):
        conn = _connect(":memory:")
        _insert_observation(conn, {"kind": "note", "summary": "rollout checkpoint ready", "tool_name": "memory_store", "detail": {"scope": "proj.alpha"}})
        _insert_observation(conn, {"kind": "note", "summary": "rollout checkpoint blocked", "tool_name": "memory_store", "detail": {"scope": "proj.alpha"}})
        before_count = conn.execute("SELECT COUNT(*) AS n FROM observations").fetchone()["n"]

        args = build_parser().parse_args(["graph", "synth", "recommend"])
        buf = io.StringIO()
        with redirect_stdout(buf):
            args.func(conn, args)

        after_count = conn.execute("SELECT COUNT(*) AS n FROM observations").fetchone()["n"]
        out = json.loads(buf.getvalue())
        self.assertEqual(out["kind"], "openclaw-mem.graph.synth.recommend.v0")
        self.assertGreaterEqual(out["counts"]["compileSynthesis"], 1)
        compile_items = [item for item in out["items"] if item["action"] == "compile_new_card"]
        self.assertTrue(compile_items)
        self.assertEqual(compile_items[0]["target"]["scope"], "proj.alpha")
        self.assertEqual(compile_items[0]["target"]["clusterKey"], "checkpoint")
        self.assertEqual(compile_items[0]["target"]["recordRefs"], ["obs:1", "obs:2"])
        self.assertIn("--record-ref obs:1", compile_items[0]["suggestion"]["command"])
        self.assertEqual(before_count, after_count)
        conn.close()

    def test_graph_synth_recommend_returns_no_action_on_empty_state_without_writes(self):
        conn = _connect(":memory:")
        before_count = conn.execute("SELECT COUNT(*) AS n FROM observations").fetchone()["n"]

        args = build_parser().parse_args(["graph", "synth", "recommend"])
        buf = io.StringIO()
        with redirect_stdout(buf):
            args.func(conn, args)

        after_count = conn.execute("SELECT COUNT(*) AS n FROM observations").fetchone()["n"]
        out = json.loads(buf.getvalue())
        self.assertEqual(out["kind"], "openclaw-mem.graph.synth.recommend.v0")
        self.assertEqual(out["counts"]["refreshSynthesis"], 0)
        self.assertEqual(out["counts"]["compileSynthesis"], 0)
        self.assertEqual(out["counts"]["noAction"], 1)
        self.assertEqual(out["items"], [{
            "action": "no_action",
            "reasons": ["no_recommendations"],
            "target": {},
            "suggestion": {"args": [], "command": None},
        }])
        self.assertEqual(before_count, after_count)
        conn.close()


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