import io
import json
import unittest
from contextlib import redirect_stdout
from datetime import datetime, timedelta, timezone

from openclaw_mem.cli import _connect, build_parser, cmd_optimize_consolidation_review


def _insert_lifecycle_row(conn, *, selected_refs: list[str], ts_iso: str | None = None) -> None:
    ts_value = ts_iso or datetime.now(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
    receipt = {
        "kind": "openclaw-mem.pack.lifecycle-shadow.v1",
        "ts": ts_value,
        "selection": {
            "pack_selected_refs": list(selected_refs),
            "citation_record_refs": list(selected_refs),
            "trace_refreshed_record_refs": list(selected_refs),
            "selection_signature": "sha256:test",
        },
        "mutation": {
            "memory_mutation": "none",
            "auto_archive_applied": 0,
            "auto_mutation_applied": 0,
        },
    }
    conn.execute(
        """
        INSERT INTO pack_lifecycle_shadow_log (
            ts, query_hash, selection_signature, selected_count, citation_count, candidate_count, receipt_json
        ) VALUES (?, ?, ?, ?, ?, ?, ?)
        """,
        (
            ts_value,
            "sha256:q",
            "sha256:test",
            len(selected_refs),
            len(selected_refs),
            len(selected_refs),
            json.dumps(receipt, ensure_ascii=False, sort_keys=True),
        ),
    )
    conn.commit()


class TestOptimizeConsolidationReview(unittest.TestCase):
    def _run(self, conn, argv, *, expect_exit=None):
        args = build_parser().parse_args(argv)
        buf = io.StringIO()
        with redirect_stdout(buf):
            if expect_exit is None:
                args.func(conn, args)
            else:
                with self.assertRaises(SystemExit) as cm:
                    args.func(conn, args)
                self.assertEqual(cm.exception.code, expect_exit)
        text = buf.getvalue().strip()
        return json.loads(text) if text else None

    @staticmethod
    def _ts_days_ago(days: int) -> int:
        dt = datetime.now(timezone.utc).replace(microsecond=0) - timedelta(days=days)
        return int(dt.timestamp() * 1000)

    @staticmethod
    def _iso_days_ago(days: int) -> str:
        dt = datetime.now(timezone.utc).replace(microsecond=0) - timedelta(days=days)
        return dt.isoformat().replace("+00:00", "Z")

    @staticmethod
    def _insert_episode(
        conn,
        *,
        event_id: str,
        ts_ms: int,
        scope: str,
        session_id: str,
        event_type: str,
        summary: str,
        payload=None,
        refs=None,
        redacted: int = 0,
    ):
        conn.execute(
            """
            INSERT INTO episodic_events (
                event_id, ts_ms, scope, session_id, agent_id, type, summary,
                payload_json, refs_json, redacted, schema_version, created_at
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                event_id,
                ts_ms,
                scope,
                session_id,
                "tester",
                event_type,
                summary,
                json.dumps(payload, ensure_ascii=False) if payload is not None else None,
                json.dumps(refs, ensure_ascii=False) if refs is not None else None,
                redacted,
                "v0",
                datetime.now(timezone.utc).isoformat(),
            ),
        )

    def test_optimize_parser_parses_consolidation_review_flags(self):
        args = build_parser().parse_args(
            [
                "optimize",
                "consolidation-review",
                "--limit",
                "250",
                "--scope",
                "alpha/team",
                "--session-id",
                "s-42",
                "--summary-min-group-size",
                "3",
                "--summary-min-shared-tokens",
                "2",
                "--archive-lookahead-days",
                "9",
                "--archive-min-signal-reasons",
                "3",
                "--link-min-shared-tokens",
                "4",
                "--link-lexical-backfill-max",
                "2",
                "--lifecycle-limit",
                "33",
                "--top",
                "6",
            ]
        )

        self.assertEqual(args.cmd, "optimize")
        self.assertEqual(args.optimize_cmd, "consolidation-review")
        self.assertEqual(args.limit, 250)
        self.assertEqual(args.scope, "alpha/team")
        self.assertEqual(args.session_id, "s-42")
        self.assertEqual(args.summary_min_group_size, 3)
        self.assertEqual(args.summary_min_shared_tokens, 2)
        self.assertEqual(args.archive_lookahead_days, 9)
        self.assertEqual(args.archive_min_signal_reasons, 3)
        self.assertEqual(args.link_min_shared_tokens, 4)
        self.assertEqual(args.link_lexical_backfill_max, 2)
        self.assertEqual(args.lifecycle_limit, 33)
        self.assertEqual(args.top, 6)

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

        # Summary cluster within one session.
        self._insert_episode(
            conn,
            event_id="ev-1",
            ts_ms=self._ts_days_ago(3),
            scope="alpha",
            session_id="s-1",
            event_type="conversation.user",
            summary="Investigate latency spike in memory recall lane",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:1"},
        )
        self._insert_episode(
            conn,
            event_id="ev-2",
            ts_ms=self._ts_days_ago(2),
            scope="alpha",
            session_id="s-1",
            event_type="conversation.assistant",
            summary="Memory recall latency spike still affects the debug lane",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:2"},
        )
        self._insert_episode(
            conn,
            event_id="ev-3",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-1",
            event_type="tool.result",
            summary="Recall lane latency metrics confirm spike in debug memory path",
            payload={"kind": "tool"},
            refs={"recordRef": "obs:3"},
        )

        # Cross-session lexical link candidate.
        self._insert_episode(
            conn,
            event_id="ev-4",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-2",
            event_type="conversation.user",
            summary="Need a rollback plan for memory recall latency",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:4"},
        )

        # Archive candidate: nearing TTL with low-signal traits.
        self._insert_episode(
            conn,
            event_id="ev-5",
            ts_ms=self._ts_days_ago(28),
            scope="alpha",
            session_id="s-old",
            event_type="tool.result",
            summary="ok temp note",
            payload=None,
            refs=None,
            redacted=1,
        )
        conn.commit()

        args = type(
            "Args",
            (),
            {
                "limit": 500,
                "scope": None,
                "session_id": None,
                "summary_min_group_size": 2,
                "summary_min_shared_tokens": 2,
                "archive_lookahead_days": 7,
                "archive_min_signal_reasons": 2,
                "link_min_shared_tokens": 2,
                "lifecycle_limit": 200,
                "top": 10,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_optimize_consolidation_review(conn, args)
        out = json.loads(buf.getvalue())

        self.assertEqual(out["kind"], "openclaw-mem.optimize.consolidation-review.v0")
        self.assertEqual(out["policy"]["writes_performed"], 0)
        self.assertEqual(out["policy"]["memory_mutation"], "none")
        self.assertEqual(out["policy"]["canonical_rewrite"], "forbidden")

        self.assertGreaterEqual(out["candidates"]["summary"]["groups"], 1)
        self.assertGreaterEqual(out["candidates"]["archive"]["count"], 1)
        self.assertGreaterEqual(out["candidates"]["links"]["pairs"], 1)

        summary_item = out["candidates"]["summary"]["items"][0]
        self.assertIn("draft_summary", summary_item)
        self.assertGreaterEqual(len(summary_item["source_event_refs"]), 2)
        self.assertIn("memory", summary_item["shared_tokens"])

        archive_item = out["candidates"]["archive"]["items"][0]
        self.assertEqual(archive_item["event_id"], "ev-5")
        self.assertIn("redacted", archive_item["low_signal_reasons"])

        link_item = out["candidates"]["links"]["items"][0]
        self.assertEqual(link_item["left"]["scope"], "alpha")
        self.assertNotEqual(link_item["left"]["session_id"], link_item["right"]["session_id"])

        rec_types = {r["type"] for r in out.get("recommendations", [])}
        self.assertIn("review_summary_candidates", rec_types)
        self.assertIn("stage_archive_candidates", rec_types)
        self.assertIn("review_link_candidates", rec_types)
        conn.close()

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

        # Receipt-supported pair (no lexical overlap required).
        self._insert_episode(
            conn,
            event_id="ev-r1",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-a",
            event_type="conversation.user",
            summary="orion pipeline coldpath handoff",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:1"},
        )
        self._insert_episode(
            conn,
            event_id="ev-r2",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-b",
            event_type="conversation.assistant",
            summary="nebula ledger warmstart guard",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:2"},
        )

        # Lexical-only pair should still be allowed in bounded low-confidence backfill.
        self._insert_episode(
            conn,
            event_id="ev-l1",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-c",
            event_type="conversation.user",
            summary="memory recall latency triage",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:30"},
        )
        self._insert_episode(
            conn,
            event_id="ev-l2",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-d",
            event_type="conversation.assistant",
            summary="recall latency rollback plan",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:31"},
        )

        _insert_lifecycle_row(conn, selected_refs=["obs:1", "obs:2"])

        args = type(
            "Args",
            (),
            {
                "limit": 500,
                "scope": None,
                "session_id": None,
                "summary_min_group_size": 2,
                "summary_min_shared_tokens": 2,
                "archive_lookahead_days": 7,
                "archive_min_signal_reasons": 2,
                "link_min_shared_tokens": 2,
                "link_lexical_backfill_max": 1,
                "lifecycle_limit": 50,
                "top": 10,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_optimize_consolidation_review(conn, args)
        out = json.loads(buf.getvalue())

        links = out["candidates"]["links"]
        self.assertEqual(links["pairs"], 2)
        first = links["items"][0]
        self.assertEqual(first["evidence_mode"], "receipt_co_selection")
        self.assertEqual(first["left"]["event_id"], "ev-r1")
        self.assertEqual(first["right"]["event_id"], "ev-r2")
        self.assertEqual(first["shared_token_count"], 0)
        self.assertGreaterEqual(first["receipt_evidence"]["co_selection_events"], 1)
        self.assertIn("sha256:test", first["receipt_evidence"]["selection_signatures"])

        backfill = [it for it in links["items"] if it["evidence_mode"] == "lexical_backfill_low_confidence"]
        self.assertEqual(len(backfill), 1)
        self.assertGreater(backfill[0]["confidence"], 0.35)
        self.assertLess(backfill[0]["confidence"], 0.56)
        self.assertEqual(backfill[0]["confidence_components"]["model"], "evidence_weighted_v0")

        self.assertGreater(first["confidence"], backfill[0]["confidence"])
        self.assertGreater(first["confidence_components"]["contributions"]["co_selection_events"], 0.0)

        link_evidence = out["signals"]["link_evidence"]
        self.assertEqual(link_evidence["receipt_supported_pairs"], 1)
        self.assertEqual(link_evidence["lexical_fallback_pairs"], 0)
        self.assertEqual(link_evidence["lexical_backfill_pairs"], 1)
        self.assertEqual(link_evidence["lexical_backfill_candidate_pairs"], 1)
        self.assertEqual(link_evidence["lexical_backfill_suppressed_pairs"], 0)
        self.assertTrue(link_evidence["hybrid_gate"]["active"])
        self.assertEqual(link_evidence["hybrid_gate"]["lexical_backfill_max"], 1)
        self.assertEqual(link_evidence["confidence_model"]["name"], "evidence_weighted_v0")
        conn.close()

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

        self._insert_episode(
            conn,
            event_id="ev-r1",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-a",
            event_type="conversation.user",
            summary="orion pipeline coldpath handoff",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:1"},
        )
        self._insert_episode(
            conn,
            event_id="ev-r2",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-b",
            event_type="conversation.assistant",
            summary="nebula ledger warmstart guard",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:2"},
        )

        self._insert_episode(
            conn,
            event_id="ev-l1",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-c",
            event_type="conversation.user",
            summary="atlas drift reconciliation memo",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:30"},
        )
        self._insert_episode(
            conn,
            event_id="ev-l2",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-d",
            event_type="conversation.assistant",
            summary="atlas drift rollback checklist",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:31"},
        )
        self._insert_episode(
            conn,
            event_id="ev-l3",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-e",
            event_type="conversation.user",
            summary="phoenix cache warmup note",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:32"},
        )
        self._insert_episode(
            conn,
            event_id="ev-l4",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-f",
            event_type="conversation.assistant",
            summary="phoenix cache incident review",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:33"},
        )

        _insert_lifecycle_row(conn, selected_refs=["obs:1", "obs:2"])

        args = type(
            "Args",
            (),
            {
                "limit": 500,
                "scope": None,
                "session_id": None,
                "summary_min_group_size": 2,
                "summary_min_shared_tokens": 2,
                "archive_lookahead_days": 7,
                "archive_min_signal_reasons": 2,
                "link_min_shared_tokens": 2,
                "link_lexical_backfill_max": 1,
                "lifecycle_limit": 50,
                "top": 10,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_optimize_consolidation_review(conn, args)
        out = json.loads(buf.getvalue())

        links = out["candidates"]["links"]
        self.assertEqual(links["pairs"], 2)
        self.assertEqual(sum(1 for it in links["items"] if it["evidence_mode"] == "receipt_co_selection"), 1)
        self.assertEqual(sum(1 for it in links["items"] if it["evidence_mode"] == "lexical_backfill_low_confidence"), 1)

        link_evidence = out["signals"]["link_evidence"]
        self.assertEqual(link_evidence["lexical_backfill_candidate_pairs"], 2)
        self.assertEqual(link_evidence["lexical_backfill_pairs"], 1)
        self.assertEqual(link_evidence["lexical_backfill_suppressed_pairs"], 1)
        conn.close()

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

        self._insert_episode(
            conn,
            event_id="ev-weak-1",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-a",
            event_type="conversation.user",
            summary="orion controlplane handoff",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:1"},
        )
        self._insert_episode(
            conn,
            event_id="ev-weak-2",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-b",
            event_type="conversation.assistant",
            summary="nebula pager warmstart",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:2"},
        )
        self._insert_episode(
            conn,
            event_id="ev-strong-1",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-c",
            event_type="conversation.user",
            summary="atlas checkpoint freeze",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:3"},
        )
        self._insert_episode(
            conn,
            event_id="ev-strong-2",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-d",
            event_type="conversation.assistant",
            summary="phoenix ledger rotate",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:4"},
        )

        _insert_lifecycle_row(conn, selected_refs=["obs:1", "obs:2"])
        _insert_lifecycle_row(conn, selected_refs=["obs:3", "obs:4"])
        _insert_lifecycle_row(conn, selected_refs=["obs:3", "obs:4"])
        _insert_lifecycle_row(conn, selected_refs=["obs:3", "obs:4"])

        args = type(
            "Args",
            (),
            {
                "limit": 500,
                "scope": None,
                "session_id": None,
                "summary_min_group_size": 2,
                "summary_min_shared_tokens": 2,
                "archive_lookahead_days": 7,
                "archive_min_signal_reasons": 2,
                "link_min_shared_tokens": 2,
                "link_lexical_backfill_max": 1,
                "lifecycle_limit": 50,
                "top": 10,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_optimize_consolidation_review(conn, args)
        out = json.loads(buf.getvalue())

        receipt_links = [it for it in out["candidates"]["links"]["items"] if it["evidence_mode"] == "receipt_co_selection"]
        self.assertEqual(len(receipt_links), 2)

        by_pair = {
            tuple(sorted([it["left"]["event_id"], it["right"]["event_id"]])): it
            for it in receipt_links
        }
        weak = by_pair[("ev-weak-1", "ev-weak-2")]
        strong = by_pair[("ev-strong-1", "ev-strong-2")]

        self.assertEqual(weak["receipt_evidence"]["co_selection_events"], 1)
        self.assertEqual(strong["receipt_evidence"]["co_selection_events"], 3)
        self.assertGreater(strong["confidence"], weak["confidence"])
        self.assertGreater(
            strong["confidence_components"]["contributions"]["co_selection_events"],
            weak["confidence_components"]["contributions"]["co_selection_events"],
        )
        conn.close()

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

        self._insert_episode(
            conn,
            event_id="ev-old-1",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-old-a",
            event_type="conversation.user",
            summary="orion glacier spindle",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:10"},
        )
        self._insert_episode(
            conn,
            event_id="ev-old-2",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-old-b",
            event_type="conversation.assistant",
            summary="nebula kiln tracer",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:11"},
        )
        self._insert_episode(
            conn,
            event_id="ev-new-1",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-new-a",
            event_type="conversation.user",
            summary="atlas vector lattice",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:12"},
        )
        self._insert_episode(
            conn,
            event_id="ev-new-2",
            ts_ms=self._ts_days_ago(1),
            scope="alpha",
            session_id="s-new-b",
            event_type="conversation.assistant",
            summary="phoenix relay anchor",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:13"},
        )

        _insert_lifecycle_row(conn, selected_refs=["obs:10", "obs:11"], ts_iso=self._iso_days_ago(45))
        _insert_lifecycle_row(conn, selected_refs=["obs:12", "obs:13"], ts_iso=self._iso_days_ago(1))

        args = type(
            "Args",
            (),
            {
                "limit": 500,
                "scope": None,
                "session_id": None,
                "summary_min_group_size": 2,
                "summary_min_shared_tokens": 2,
                "archive_lookahead_days": 7,
                "archive_min_signal_reasons": 2,
                "link_min_shared_tokens": 2,
                "link_lexical_backfill_max": 1,
                "lifecycle_limit": 50,
                "top": 10,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_optimize_consolidation_review(conn, args)
        out = json.loads(buf.getvalue())

        receipt_links = [it for it in out["candidates"]["links"]["items"] if it["evidence_mode"] == "receipt_co_selection"]
        self.assertEqual(len(receipt_links), 2)

        by_pair = {
            tuple(sorted([it["left"]["event_id"], it["right"]["event_id"]])): it
            for it in receipt_links
        }
        old_pair = by_pair[("ev-old-1", "ev-old-2")]
        new_pair = by_pair[("ev-new-1", "ev-new-2")]

        self.assertEqual(old_pair["receipt_evidence"]["co_selection_events"], 1)
        self.assertEqual(new_pair["receipt_evidence"]["co_selection_events"], 1)
        self.assertGreater(
            new_pair["receipt_evidence"]["co_selection_recency_score"],
            old_pair["receipt_evidence"]["co_selection_recency_score"],
        )
        self.assertGreater(new_pair["confidence"], old_pair["confidence"])
        self.assertGreater(
            new_pair["confidence_components"]["contributions"]["co_selection_recency"],
            old_pair["confidence_components"]["contributions"]["co_selection_recency"],
        )
        self.assertEqual(
            out["signals"]["link_evidence"]["confidence_model"]["normalization"]["co_selection_recency_half_life_days"],
            30.0,
        )
        conn.close()

    def test_consolidation_review_protects_archive_candidates_with_recent_use_refs(self):
        conn = _connect(":memory:")
        self._insert_episode(
            conn,
            event_id="ev-protected",
            ts_ms=self._ts_days_ago(28),
            scope="alpha",
            session_id="s-old",
            event_type="tool.result",
            summary="temp low signal note",
            payload=None,
            refs={"recordRef": "obs:1"},
            redacted=1,
        )
        self._insert_episode(
            conn,
            event_id="ev-plain",
            ts_ms=self._ts_days_ago(28),
            scope="alpha",
            session_id="s-old",
            event_type="tool.result",
            summary="plain low signal note",
            payload=None,
            refs=None,
            redacted=1,
        )
        _insert_lifecycle_row(conn, selected_refs=["obs:1"])

        args = type(
            "Args",
            (),
            {
                "limit": 500,
                "scope": None,
                "session_id": None,
                "summary_min_group_size": 2,
                "summary_min_shared_tokens": 2,
                "archive_lookahead_days": 7,
                "archive_min_signal_reasons": 2,
                "link_min_shared_tokens": 2,
                "lifecycle_limit": 50,
                "top": 10,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_optimize_consolidation_review(conn, args)
        out = json.loads(buf.getvalue())

        archive = out["candidates"]["archive"]
        recent_use = out["signals"]["recent_use"]
        self.assertEqual(archive["count"], 1)
        self.assertEqual(archive["protected_by_recent_use"], 1)
        self.assertEqual(archive["items"][0]["event_id"], "ev-plain")
        self.assertEqual(recent_use["protected_archive_candidates"], 1)
        self.assertEqual(recent_use["linked_observation_rows"], 1)
        self.assertEqual(recent_use["items"][0]["id"], 1)
        conn.close()

    def test_consolidation_review_scope_and_session_filters(self):
        conn = _connect(":memory:")
        self._insert_episode(
            conn,
            event_id="alpha-1",
            ts_ms=self._ts_days_ago(1),
            scope="alpha-team",
            session_id="s-focus",
            event_type="conversation.user",
            summary="Cluster alpha memory drift issue",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:10"},
        )
        self._insert_episode(
            conn,
            event_id="alpha-2",
            ts_ms=self._ts_days_ago(1),
            scope="alpha-team",
            session_id="s-focus",
            event_type="conversation.assistant",
            summary="Alpha memory drift issue needs triage",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:11"},
        )
        self._insert_episode(
            conn,
            event_id="beta-1",
            ts_ms=self._ts_days_ago(1),
            scope="beta-team",
            session_id="s-other",
            event_type="conversation.user",
            summary="Unrelated beta scope event",
            payload={"kind": "msg"},
            refs={"recordRef": "obs:12"},
        )
        conn.commit()

        args = type(
            "Args",
            (),
            {
                "limit": 500,
                "scope": "alpha team",
                "session_id": "s-focus",
                "summary_min_group_size": 2,
                "summary_min_shared_tokens": 2,
                "archive_lookahead_days": 7,
                "archive_min_signal_reasons": 2,
                "link_min_shared_tokens": 2,
                "lifecycle_limit": 200,
                "top": 10,
                "json": True,
            },
        )()

        buf = io.StringIO()
        with redirect_stdout(buf):
            cmd_optimize_consolidation_review(conn, args)
        out = json.loads(buf.getvalue())
        self.assertEqual(out["source"]["scope"], "alpha-team")
        self.assertEqual(out["source"]["session_id"], "s-focus")
        self.assertEqual(out["source"]["rows_scanned"], 2)
        self.assertEqual(out["candidates"]["summary"]["groups"], 1)
        conn.close()
