mempalace--mempalace/tests/test_palace_graph_tunnels.py

"""Tests for explicit tunnel helpers in mempalace.palace_graph."""

import logging
import os
import stat
import sys
from unittest.mock import MagicMock, patch

import pytest

with patch.dict("sys.modules", {"chromadb": MagicMock()}):
    import mempalace.palace_graph as palace_graph


def _use_tmp_tunnel_file(monkeypatch, tmp_path):
    """Redirect both the tunnel-file resolver and the legacy-file check at the
    tmp_path so existing tests stay in the configured-path branch and don't
    accidentally trip the new legacy-file warning branch in _load_tunnels.

    Also neutralizes ``_get_collection`` so the endpoint-existence validation
    added for #1468 falls through to the "can't verify, allow" branch by
    default. Tests that exercise the validation path supply their own stub
    via a subsequent monkeypatch.setattr.
    """
    tunnel_file = tmp_path / "tunnels.json"
    monkeypatch.setattr(palace_graph, "_get_tunnel_file", lambda *a, **kw: str(tunnel_file))
    monkeypatch.setattr(
        palace_graph,
        "_legacy_tunnel_file",
        lambda: str(tmp_path / "legacy-tunnels.json"),
    )
    monkeypatch.setattr(palace_graph, "_get_collection", lambda *a, **kw: None)
    return tunnel_file


class TestTunnelStorage:
    def test_load_tunnels_missing_file_returns_empty_list(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        assert palace_graph._load_tunnels() == []

    def test_load_tunnels_corrupt_file_returns_empty_list(self, tmp_path, monkeypatch):
        tunnel_file = _use_tmp_tunnel_file(monkeypatch, tmp_path)
        tunnel_file.write_text("{not valid json", encoding="utf-8")
        assert palace_graph._load_tunnels() == []

    def test_save_and_load_round_trip(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        tunnels = [
            {
                "id": "abc123",
                "source": {"wing": "wing_code", "room": "auth"},
                "target": {"wing": "wing_people", "room": "users"},
                "label": "same concept",
            }
        ]
        palace_graph._save_tunnels(tunnels)
        assert palace_graph._load_tunnels() == tunnels

    @pytest.mark.skipif(
        sys.platform == "win32",
        reason="POSIX file-permission bits only apply on Unix-like systems",
    )
    def test_save_tunnels_restricts_permissions(self, tmp_path, monkeypatch):
        """Regression for #1165 — tunnels.json reveals cross-wing links and
        must not be world-readable on shared Linux/multi-user systems."""
        tunnel_file = _use_tmp_tunnel_file(monkeypatch, tmp_path)
        palace_graph._save_tunnels(
            [
                {
                    "id": "x",
                    "source": {"wing": "a", "room": "r1"},
                    "target": {"wing": "b", "room": "r2"},
                    "label": "",
                }
            ]
        )

        file_mode = stat.S_IMODE(os.stat(tunnel_file).st_mode)
        assert file_mode == 0o600, f"tunnels.json mode is {oct(file_mode)}, expected 0o600"

        parent_mode = stat.S_IMODE(os.stat(tunnel_file.parent).st_mode)
        assert parent_mode == 0o700, (
            f"tunnels.json parent dir mode is {oct(parent_mode)}, expected 0o700"
        )


class TestExplicitTunnels:
    def test_normalize_wing_uses_shared_rule_and_trims_empty(self):
        assert palace_graph._normalize_wing(" Mempalace-Public ") == "mempalace_public"
        assert palace_graph._normalize_wing("   ") is None
        assert palace_graph._normalize_wing(None) is None
        # Non-string inputs (corrupt or hand-edited tunnels.json) return None
        # instead of raising — keeps read-path filters robust to bad records.
        assert palace_graph._normalize_wing(42) is None
        assert palace_graph._normalize_wing(["x"]) is None

    def test_create_tunnel_deduplicates_reverse_order_and_updates_label(
        self, tmp_path, monkeypatch
    ):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)

        first = palace_graph.create_tunnel(
            "wing_code", "auth", "wing_people", "users", label="same concept"
        )
        second = palace_graph.create_tunnel(
            "wing_people", "users", "wing_code", "auth", label="updated label"
        )

        assert first["id"] == second["id"]
        assert len(palace_graph.list_tunnels()) == 1
        assert second["label"] == "updated label"
        assert second["created_at"] == first["created_at"]
        assert "updated_at" in second

    def test_create_tunnel_rejects_empty_names(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)

        with pytest.raises(ValueError):
            palace_graph.create_tunnel("", "auth", "wing_people", "users")

    def test_list_tunnels_filters_by_either_side(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)

        palace_graph.create_tunnel("wing_code", "auth", "wing_people", "users", label="A")
        palace_graph.create_tunnel("wing_ops", "deploy", "wing_people", "users", label="B")

        assert len(palace_graph.list_tunnels()) == 2
        assert len(palace_graph.list_tunnels("wing_people")) == 2
        assert len(palace_graph.list_tunnels("wing_code")) == 1

    def test_delete_tunnel_removes_saved_tunnel(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)

        tunnel = palace_graph.create_tunnel(
            "wing_code", "auth", "wing_people", "users", label="same concept"
        )

        assert palace_graph.delete_tunnel(tunnel["id"]) == {"deleted": tunnel["id"]}
        assert palace_graph.list_tunnels() == []

    def test_follow_tunnels_returns_direction_and_preview(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)

        palace_graph.create_tunnel(
            "wing_code",
            "auth",
            "wing_people",
            "users",
            label="same concept",
            target_drawer_id="drawer_users_1",
        )

        col = MagicMock()
        col.get.return_value = {
            "ids": ["drawer_users_1"],
            "documents": ["A" * 400],
            "metadatas": [{}],
        }

        outgoing = palace_graph.follow_tunnels("wing_code", "auth", col=col)
        assert len(outgoing) == 1
        assert outgoing[0]["direction"] == "outgoing"
        assert outgoing[0]["connected_wing"] == "wing_people"
        assert outgoing[0]["connected_room"] == "users"
        assert outgoing[0]["drawer_id"] == "drawer_users_1"
        assert len(outgoing[0]["drawer_preview"]) == 300

        incoming = palace_graph.follow_tunnels("wing_people", "users", col=col)
        assert len(incoming) == 1
        assert incoming[0]["direction"] == "incoming"
        assert incoming[0]["connected_wing"] == "wing_code"

    def test_follow_tunnels_returns_connections_even_if_collection_lookup_fails(
        self, tmp_path, monkeypatch
    ):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)

        palace_graph.create_tunnel(
            "wing_code",
            "auth",
            "wing_people",
            "users",
            label="same concept",
            target_drawer_id="drawer_users_1",
        )

        col = MagicMock()
        col.get.side_effect = RuntimeError("boom")

        connections = palace_graph.follow_tunnels("wing_code", "auth", col=col)
        assert len(connections) == 1
        assert "drawer_preview" not in connections[0]


class TestTopicTunnels:
    """Cross-wing topic tunnels (issue #1180).

    When two wings share confirmed TOPIC labels above a configurable
    threshold, a symmetric tunnel is created between them. Tunnels are
    routed through the existing ``create_tunnel`` storage so they share
    dedup and persistence with explicit tunnels.
    """

    def test_compute_topic_tunnels_creates_link_for_shared_topic(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        topics_by_wing = {
            "wing_alpha": ["Angular", "OpenAPI"],
            "wing_beta": ["OpenAPI", "Kubernetes"],
        }
        created = palace_graph.compute_topic_tunnels(topics_by_wing, min_count=1)
        assert len(created) == 1
        assert created[0]["source"]["wing"] in {"wing_alpha", "wing_beta"}
        assert created[0]["target"]["wing"] in {"wing_alpha", "wing_beta"}
        # Room is namespaced with the ``topic:`` prefix so it can't collide
        # with a literal folder-derived room of the same name. Casing of the
        # topic is preserved for display.
        assert created[0]["source"]["room"] == "topic:OpenAPI"
        assert created[0]["target"]["room"] == "topic:OpenAPI"
        assert created[0]["kind"] == "topic"
        # Label carries the human-readable topic without the prefix.
        assert "OpenAPI" in created[0]["label"]
        assert "topic:OpenAPI" not in created[0]["label"]

        # Tunnel is retrievable via the standard list_tunnels API.
        listed = palace_graph.list_tunnels()
        assert len(listed) == 1
        assert listed[0]["id"] == created[0]["id"]

    def test_compute_topic_tunnels_no_link_below_threshold(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        topics_by_wing = {
            "wing_alpha": ["Angular", "OpenAPI"],
            "wing_beta": ["OpenAPI", "Kubernetes"],
        }
        # min_count=2 requires two overlapping topics — only one shared.
        created = palace_graph.compute_topic_tunnels(topics_by_wing, min_count=2)
        assert created == []
        assert palace_graph.list_tunnels() == []

    def test_compute_topic_tunnels_above_threshold_creates_per_topic_links(
        self, tmp_path, monkeypatch
    ):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        topics_by_wing = {
            "wing_alpha": ["Angular", "OpenAPI", "Postgres"],
            "wing_beta": ["Angular", "OpenAPI", "Redis"],
        }
        created = palace_graph.compute_topic_tunnels(topics_by_wing, min_count=2)
        # Two shared topics × one wing pair = two tunnels.
        rooms = sorted(t["source"]["room"] for t in created)
        assert rooms == ["topic:Angular", "topic:OpenAPI"]

    def test_compute_topic_tunnels_case_insensitive_overlap(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        topics_by_wing = {
            "wing_alpha": ["openapi"],
            "wing_beta": ["OpenAPI"],
        }
        created = palace_graph.compute_topic_tunnels(topics_by_wing, min_count=1)
        assert len(created) == 1

    def test_compute_topic_tunnels_empty_input_is_noop(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        assert palace_graph.compute_topic_tunnels({}) == []
        assert palace_graph.compute_topic_tunnels({"wing_a": []}) == []
        assert palace_graph.list_tunnels() == []

    def test_compute_topic_tunnels_three_wings_pairwise(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        topics_by_wing = {
            "wing_a": ["foo"],
            "wing_b": ["foo"],
            "wing_c": ["foo"],
        }
        created = palace_graph.compute_topic_tunnels(topics_by_wing, min_count=1)
        # 3 wings sharing the same topic → C(3,2) = 3 pairs → 3 tunnels.
        assert len(created) == 3
        endpoint_pairs = {
            tuple(sorted([t["source"]["wing"], t["target"]["wing"]])) for t in created
        }
        assert endpoint_pairs == {
            ("wing_a", "wing_b"),
            ("wing_a", "wing_c"),
            ("wing_b", "wing_c"),
        }

    def test_topic_tunnels_for_wing_only_links_that_wing(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        topics_by_wing = {
            "wing_a": ["foo", "bar"],
            "wing_b": ["foo"],
            "wing_c": ["bar"],
        }
        # wing_a should link to both b (via foo) and c (via bar).
        created = palace_graph.topic_tunnels_for_wing("wing_a", topics_by_wing)
        endpoint_pairs = {
            tuple(sorted([t["source"]["wing"], t["target"]["wing"]])) for t in created
        }
        assert endpoint_pairs == {("wing_a", "wing_b"), ("wing_a", "wing_c")}
        # The b-c pair is NOT created because wing_a's incremental pass
        # only computes pairs that include wing_a.
        assert len(palace_graph.list_tunnels()) == 2

    def test_topic_tunnels_for_wing_unknown_wing_is_noop(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        topics_by_wing = {"wing_a": ["foo"], "wing_b": ["foo"]}
        assert palace_graph.topic_tunnels_for_wing("wing_missing", topics_by_wing) == []
        assert palace_graph.list_tunnels() == []

    def test_topic_tunnels_for_wing_matches_across_slug_forms(self, tmp_path, monkeypatch):
        """The wing arg and ``topics_by_wing`` keys may carry different slug
        forms (hyphen vs underscore). ``topic_tunnels_for_wing`` resolves
        the lookup through ``normalize_wing_name`` so a caller passing
        ``"my-wing"`` against a registry keyed by ``"my_wing"`` still wires
        up the topic tunnels."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        topics_by_wing = {"my_wing": ["Angular"], "wing_people": ["Angular"]}
        created = palace_graph.topic_tunnels_for_wing("my-wing", topics_by_wing)
        assert len(created) == 1

    def test_compute_topic_tunnels_dedupe_on_recompute(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        topics_by_wing = {
            "wing_alpha": ["OpenAPI"],
            "wing_beta": ["OpenAPI"],
        }
        first = palace_graph.compute_topic_tunnels(topics_by_wing, min_count=1)
        second = palace_graph.compute_topic_tunnels(topics_by_wing, min_count=1)
        # create_tunnel is symmetric/dedupe — repeated computation should
        # not multiply the stored tunnels.
        assert first[0]["id"] == second[0]["id"]
        assert len(palace_graph.list_tunnels()) == 1

    def test_topic_tunnel_room_does_not_collide_with_literal_room(self, tmp_path, monkeypatch):
        """Regression: a literal "Angular" folder-room and a topic tunnel
        for "Angular" must resolve to distinct endpoints so ``follow_tunnels``
        from the real room doesn't accidentally surface topic connections
        (issue raised in review of #1184)."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)

        # Explicit tunnel anchored at a literal "Angular" room in wing_alpha.
        palace_graph.create_tunnel(
            "wing_alpha", "Angular", "wing_gamma", "frontend", label="explicit"
        )
        # Topic tunnel between the same wings that share the "Angular" topic.
        palace_graph.compute_topic_tunnels(
            {"wing_alpha": ["Angular"], "wing_beta": ["Angular"]}, min_count=1
        )

        # follow_tunnels on the literal Angular room only sees the explicit link.
        literal = palace_graph.follow_tunnels("wing_alpha", "Angular")
        assert len(literal) == 1
        assert literal[0]["connected_wing"] == "wing_gamma"

        # The topic tunnel is stored under the namespaced room.
        topical = palace_graph.follow_tunnels("wing_alpha", "topic:Angular")
        assert len(topical) == 1
        assert topical[0]["connected_wing"] == "wing_beta"

    def test_topic_tunnels_carry_kind_field(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        palace_graph.create_tunnel("wing_a", "auth", "wing_b", "users", label="x")
        palace_graph.compute_topic_tunnels({"wing_a": ["Redis"], "wing_b": ["Redis"]}, min_count=1)

        tunnels = palace_graph.list_tunnels()
        kinds = sorted(t["kind"] for t in tunnels)
        assert kinds == ["explicit", "topic"]

    def test_compute_topic_tunnels_normalizes_wing_keys(self, tmp_path, monkeypatch):
        """Auto-generated topic tunnels canonicalize the wing slug so two
        mining runs with mixed forms (``my-wing`` then ``my_wing``) produce
        a single deduped record. Only user-issued ``create_tunnel`` calls
        preserve verbatim slugs (#1504); the topic-tunnel auto-generator
        owns its own slugs and stays canonical."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)

        palace_graph.compute_topic_tunnels(
            {"my-wing": ["Angular"], "wing_people": ["Angular"]}, min_count=1
        )
        palace_graph.compute_topic_tunnels(
            {"my_wing": ["Angular"], "wing_people": ["Angular"]}, min_count=1
        )

        tunnels = palace_graph.list_tunnels()
        assert len(tunnels) == 1
        stored_wings = {tunnels[0]["source"]["wing"], tunnels[0]["target"]["wing"]}
        assert stored_wings == {"my_wing", "wing_people"}


class TestHyphenatedWingNormalization:
    """Wing names may reach ``tunnels.json`` in either form:

    * ``mempalace mine`` without ``--wing`` derives the slug from the dir
      name through ``normalize_wing_name`` → stored as ``mempalace_public``.
    * ``mempalace mine --wing my-wing`` (or any explicit slug) is stored
      verbatim by ``create_tunnel`` (regression #1504) → ``my-wing``.

    Read-path helpers (``list_tunnels`` / ``follow_tunnels``) must accept
    queries in either form and match both storage forms — normalization
    is applied on both the stored value and the query key at comparison
    time, never at write time.
    """

    def test_list_tunnels_filters_hyphenated_wing(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)

        palace_graph.create_tunnel("mempalace_public", "auth", "wing_people", "users")

        assert len(palace_graph.list_tunnels("mempalace-public")) == 1
        assert len(palace_graph.list_tunnels("mempalace_public")) == 1

    def test_follow_tunnels_matches_hyphenated_wing(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)

        palace_graph.create_tunnel("mempalace_public", "auth", "wing_people", "users")

        by_hyphen = palace_graph.follow_tunnels("mempalace-public", "auth")
        by_under = palace_graph.follow_tunnels("mempalace_public", "auth")
        assert len(by_hyphen) == 1
        assert len(by_under) == 1
        assert by_hyphen[0]["connected_wing"] == "wing_people"

    def test_create_tunnel_preserves_hyphenated_wing_names(self, tmp_path, monkeypatch):
        """Regression for #1504: wings created via ``mempalace mine --wing my-wing``
        keep the hyphen in metadata, so ``create_tunnel`` must store the slug
        verbatim. Read-path normalization in ``list_tunnels``/``follow_tunnels``
        keeps both query forms working."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)

        t = palace_graph.create_tunnel("my-project", "src", "your-project", "dst", label="cross")
        assert t["source"]["wing"] == "my-project"
        assert t["target"]["wing"] == "your-project"
        assert len(palace_graph.list_tunnels("my-project")) == 1
        assert len(palace_graph.list_tunnels("my_project")) == 1

    def test_find_tunnels_warns_on_empty_result(self, tmp_path, monkeypatch, caplog):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        # No data in collection, so build_graph returns empty nodes
        with caplog.at_level("WARNING", logger="mempalace_graph"):
            result = palace_graph.find_tunnels("nonexistent-wing")
        assert result == []
        assert "No tunnels found" in caplog.text

    def test_read_path_skips_records_with_null_endpoints(self, tmp_path, monkeypatch):
        """A hand-edited ``tunnels.json`` may carry ``"source": null`` or
        ``"target": null``. The read-path filters must skip such rows
        instead of crashing the whole iteration with ``AttributeError``."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        palace_graph._save_tunnels(
            [
                {
                    "id": "broken",
                    "source": None,
                    "target": None,
                    "label": "corrupt",
                    "kind": "explicit",
                    "created_at": "2026-05-01T00:00:00+00:00",
                },
                {
                    "id": "ok",
                    "source": {"wing": "wing_a", "room": "r1"},
                    "target": {"wing": "wing_b", "room": "r2"},
                    "label": "good",
                    "kind": "explicit",
                    "created_at": "2026-05-02T00:00:00+00:00",
                },
            ]
        )

        # Both filters must skip the broken record and return the good one.
        assert {t["id"] for t in palace_graph.list_tunnels("wing_a")} == {"ok"}
        connections = palace_graph.follow_tunnels("wing_a", "r1")
        assert [c["tunnel_id"] for c in connections] == ["ok"]

    def test_pre_1504_underscore_tunnels_remain_findable(self, tmp_path, monkeypatch):
        """A ``tunnels.json`` written before #1504 stored wings in normalized
        underscore form (the write-path normalization is now gone). Read-path
        queries with either hyphen or underscore must still find those
        records after the fix."""
        tunnel_file = _use_tmp_tunnel_file(monkeypatch, tmp_path)
        palace_graph._save_tunnels(
            [
                {
                    "id": "pre_1504_record",
                    "source": {"wing": "mempalace_public", "room": "auth"},
                    "target": {"wing": "wing_people", "room": "users"},
                    "label": "pre-#1504 record",
                    "kind": "explicit",
                    "created_at": "2026-05-10T00:00:00+00:00",
                }
            ]
        )
        assert tunnel_file.exists()

        assert len(palace_graph.list_tunnels("mempalace_public")) == 1
        assert len(palace_graph.list_tunnels("mempalace-public")) == 1

        assert len(palace_graph.follow_tunnels("mempalace_public", "auth")) == 1
        assert len(palace_graph.follow_tunnels("mempalace-public", "auth")) == 1


# =============================================================================
# Regression: tunnel file follows palace_path config (#1467)
# =============================================================================
class TestTunnelFileFollowsConfig:
    """Bug A: prior to 3.3.6 the tunnel file was hardcoded at
    ``~/.mempalace/tunnels.json`` regardless of MempalaceConfig.palace_path.
    Under any profile-isolated $HOME (subagent profiles, sandboxes, multi-tenant
    hosts) tunnels would write to a different file than drawers, so
    ``create_tunnel`` would appear to succeed while the tunnel was invisible
    to every other process touching the configured palace.
    """

    def test_default_tunnel_file_unchanged(self):
        """Regression: with default config, tunnel_file resolves to
        ``~/.mempalace/tunnels.json`` so existing single-user installs are
        not silently relocated."""
        from mempalace.config import DEFAULT_PALACE_PATH, MempalaceConfig

        cfg = MempalaceConfig()
        # Default palace_path is ~/.mempalace/palace, so tunnel is sibling.
        expected = os.path.join(os.path.dirname(DEFAULT_PALACE_PATH), "tunnels.json")
        assert cfg.tunnel_file == expected
        assert palace_graph._get_tunnel_file(cfg) == expected

    def test_tunnel_file_follows_palace_path(self, tmp_path):
        """Custom palace_path → tunnel sits beside the palace, not at the
        hardcoded legacy location."""
        from mempalace.config import MempalaceConfig

        custom_dir = tmp_path / "custom-palace"
        cfg = MempalaceConfig(config_dir=tmp_path)
        cfg._file_config["palace_path"] = str(custom_dir)
        assert cfg.tunnel_file == str(tmp_path / "tunnels.json")
        assert palace_graph._get_tunnel_file(cfg) == str(tmp_path / "tunnels.json")

    def test_load_tunnels_warns_on_orphaned_legacy_file(self, tmp_path, monkeypatch, caplog):
        """When the configured tunnel file is missing but a legacy file
        exists at a different path, _load_tunnels logs a one-line warning
        naming both paths and returns []. Critically, it does NOT
        auto-migrate — silent merging risks clobbering newer data."""
        configured = tmp_path / "configured" / "tunnels.json"
        legacy = tmp_path / "legacy" / "tunnels.json"
        legacy.parent.mkdir(parents=True)
        legacy.write_text(
            '[{"id":"orphan","source":{"wing":"a","room":"r"},'
            '"target":{"wing":"b","room":"r"},"label":""}]',
            encoding="utf-8",
        )

        monkeypatch.setattr(palace_graph, "_get_tunnel_file", lambda *a, **kw: str(configured))
        monkeypatch.setattr(palace_graph, "_legacy_tunnel_file", lambda: str(legacy))

        with caplog.at_level(logging.WARNING, logger="mempalace_graph"):
            result = palace_graph._load_tunnels()

        assert result == [], "must not auto-migrate from legacy file"
        assert str(legacy) in caplog.text
        assert str(configured) in caplog.text

    def test_no_legacy_warning_when_paths_match(self, tmp_path, monkeypatch, caplog):
        """If configured and legacy resolve to the same path (default install),
        we must not emit a misleading 'legacy file ignored' warning when the
        file simply doesn't exist yet."""
        same = tmp_path / "tunnels.json"
        monkeypatch.setattr(palace_graph, "_get_tunnel_file", lambda *a, **kw: str(same))
        monkeypatch.setattr(palace_graph, "_legacy_tunnel_file", lambda: str(same))

        with caplog.at_level(logging.WARNING, logger="mempalace_graph"):
            assert palace_graph._load_tunnels() == []

        assert "Legacy tunnels file" not in caplog.text


# =============================================================================
# Regression: create_tunnel validates explicit-tunnel endpoints (#1468)
# =============================================================================
class _StubCollection:
    """Minimal chroma-collection stub for endpoint-validation tests."""

    def __init__(self, existing_rooms):
        # existing_rooms is a set of (wing, room) tuples
        self.existing_rooms = set(existing_rooms)
        self.calls = []

    def get(self, where=None, limit=None, include=None):
        self.calls.append(where)
        # Parse the {"$and": [{"wing": W}, {"room": R}]} where clause we issue.
        wing = room = None
        for clause in (where or {}).get("$and", []):
            if "wing" in clause:
                wing = clause["wing"]
            if "room" in clause:
                room = clause["room"]
        if (wing, room) in self.existing_rooms:
            return {"ids": ["drawer-1"]}
        return {"ids": []}


class TestCreateTunnelEndpointValidation:
    """Bug B: pre-3.3.6 ``create_tunnel`` only validated wing/room names
    were non-empty strings, never that the rooms actually existed in the
    chroma index. Combined with Bug A's read-bubble, callers could
    successfully create tunnels pointing at phantom endpoints and a
    follow-up ``list_tunnels`` would self-confirm via the same isolated
    file. The fix queries chroma for at least one drawer in each endpoint
    before persisting an explicit tunnel."""

    def test_create_tunnel_rejects_nonexistent_target_room(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        col = _StubCollection({("wing_code", "auth")})  # only source exists
        monkeypatch.setattr(palace_graph, "_get_collection", lambda config=None: col)

        with pytest.raises(ValueError) as exc_info:
            palace_graph.create_tunnel("wing_code", "auth", "wing_people", "phantom")
        msg = str(exc_info.value)
        assert "phantom" in msg
        assert "wing_people" in msg
        # And nothing was persisted.
        assert palace_graph.list_tunnels() == []

    def test_create_tunnel_rejects_nonexistent_source_room(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        col = _StubCollection({("wing_people", "users")})  # only target exists
        monkeypatch.setattr(palace_graph, "_get_collection", lambda config=None: col)

        with pytest.raises(ValueError) as exc_info:
            palace_graph.create_tunnel("wing_code", "phantom", "wing_people", "users")
        msg = str(exc_info.value)
        assert "phantom" in msg
        assert "wing_code" in msg

    def test_create_tunnel_succeeds_when_both_rooms_exist(self, tmp_path, monkeypatch):
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        col = _StubCollection({("wing_code", "auth"), ("wing_people", "users")})
        monkeypatch.setattr(palace_graph, "_get_collection", lambda config=None: col)

        t = palace_graph.create_tunnel(
            "wing_code", "auth", "wing_people", "users", label="verified"
        )
        assert t["label"] == "verified"
        assert len(palace_graph.list_tunnels()) == 1

    def test_create_tunnel_skips_validation_when_collection_unreachable(
        self, tmp_path, monkeypatch
    ):
        """When chroma is unreachable (palace not yet created, transient
        failure, tests without a real backend), validation is skipped
        rather than fail-closed — matches the tolerance pattern used
        throughout palace_graph._get_collection()."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        monkeypatch.setattr(palace_graph, "_get_collection", lambda config=None: None)

        t = palace_graph.create_tunnel("wing_code", "any", "wing_people", "any", label="cold-start")
        assert t["label"] == "cold-start"

    def test_create_tunnel_tolerates_collection_query_exception(self, tmp_path, monkeypatch):
        """Permission errors / temporary chroma faults during the
        validation query must not block legitimate writes."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)

        class _AngryCollection:
            def get(self, **kw):
                raise RuntimeError("simulated chroma fault")

        monkeypatch.setattr(palace_graph, "_get_collection", lambda config=None: _AngryCollection())

        # Should not raise — fall back to "allow" rather than fail-closed.
        t = palace_graph.create_tunnel("wing_code", "x", "wing_people", "y", label="best-effort")
        assert t["label"] == "best-effort"

    def test_compute_topic_tunnels_skips_endpoint_validation(self, tmp_path, monkeypatch):
        """Topic tunnels use synthetic ``topic:<name>`` room identifiers
        that don't correspond to real chroma rooms. The endpoint-existence
        check must skip kind != 'explicit', otherwise auto-derived
        cross-wing graph edges would all be rejected."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        # Empty collection — would reject if validation ran.
        col = _StubCollection(set())
        monkeypatch.setattr(palace_graph, "_get_collection", lambda config=None: col)

        # Two wings share a topic — should produce a topic-tunnel even
        # though neither "topic:auth" room exists in the stub collection.
        topics_by_wing = {
            "wing_code": ["auth", "logging"],
            "wing_people": ["auth", "schema"],
        }
        palace_graph.compute_topic_tunnels(topics_by_wing, min_count=1)
        tunnels = palace_graph.list_tunnels()
        assert tunnels, "topic tunnels must persist regardless of room validation"
        assert all(t.get("kind") == "topic" for t in tunnels)
        # And the validation query was never invoked for topic tunnels.
        assert col.calls == []


class TestEntityTunnels:
    """Cross-wing entity tunnels (the Wing → Drawer-entities → Hallway →
    Tunnel sequence from the v4 architecture vision).

    When an entity has within-wing hallways in two or more wings, an entity
    tunnel bridges those wings, anchored on the entity. This is the
    architectural counterpart to ``compute_topic_tunnels`` — same storage,
    same dedup, but the substrate is hallway records (entity-grounded)
    rather than raw topic words.

    Endpoints use the synthetic room id ``entity:<name>`` so they can't
    collide with literal folder-derived rooms of the same name (mirrors
    the ``topic:<name>`` convention).

    Topic tunnels are NOT removed by this work — both systems run in
    parallel for one release cycle so existing palaces don't lose
    tunnels between mines. Deprecation is a separate follow-up PR.
    """

    def test_entity_tunnels_creates_cross_wing_tunnel_for_shared_entity(
        self, tmp_path, monkeypatch
    ):
        """Ben appears in a hallway in wing_aya AND in wing_mempalace →
        exactly one entity tunnel between those two wings, anchored on Ben."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        hallways = [
            {"wing": "wing_aya", "entity_a": "Ben", "entity_b": "Aya"},
            {"wing": "wing_mempalace", "entity_a": "Ben", "entity_b": "Igor"},
        ]
        created = palace_graph.entity_tunnels_for_wing("wing_aya", hallways)
        assert len(created) == 1
        tunnel = created[0]
        assert {tunnel["source"]["wing"], tunnel["target"]["wing"]} == {
            "wing_aya",
            "wing_mempalace",
        }
        # Endpoint is the synthetic entity room — same casing as the
        # stored entity, prefixed with ``entity:`` to prevent collision
        # with a literal folder room of the same name.
        assert tunnel["source"]["room"] == "entity:Ben"
        assert tunnel["target"]["room"] == "entity:Ben"
        assert tunnel["kind"] == "entity"
        # Label carries the entity name without the prefix.
        assert "Ben" in tunnel["label"]
        assert "entity:Ben" not in tunnel["label"]

    def test_entity_tunnels_skips_entities_in_only_one_wing(self, tmp_path, monkeypatch):
        """Aya has hallways only in wing_aya → no tunnel for her."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        hallways = [
            {"wing": "wing_aya", "entity_a": "Aya", "entity_b": "Lumi"},
            {"wing": "wing_aya", "entity_a": "Aya", "entity_b": "Ever"},
        ]
        created = palace_graph.entity_tunnels_for_wing("wing_aya", hallways)
        assert created == []

    def test_entity_tunnels_counts_entity_in_either_pair_position(self, tmp_path, monkeypatch):
        """Entity may appear as entity_a in one hallway and entity_b in
        another. Both positions count toward 'this entity is in this wing'."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        hallways = [
            # Ben is entity_b in wing_aya
            {"wing": "wing_aya", "entity_a": "Aya", "entity_b": "Ben"},
            # Ben is entity_a in wing_mempalace
            {"wing": "wing_mempalace", "entity_a": "Ben", "entity_b": "Igor"},
        ]
        created = palace_graph.entity_tunnels_for_wing("wing_aya", hallways)
        assert len(created) == 1
        assert "Ben" in created[0]["label"]

    def test_entity_tunnels_three_wings_pairwise_from_focus_wing(self, tmp_path, monkeypatch):
        """When Ben has hallways in {wing_aya, wing_mp, wing_lt} and we
        compute for wing_aya, we create wing_aya↔wing_mp AND wing_aya↔wing_lt
        only — NOT wing_mp↔wing_lt. The cross-wing tunnel between the two
        other wings is the job of whichever of those two mines next.
        """
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        hallways = [
            {"wing": "wing_aya", "entity_a": "Ben", "entity_b": "Aya"},
            {"wing": "wing_mempalace", "entity_a": "Ben", "entity_b": "Igor"},
            {"wing": "wing_lantern", "entity_a": "Ben", "entity_b": "Lumi"},
        ]
        created = palace_graph.entity_tunnels_for_wing("wing_aya", hallways)
        assert len(created) == 2
        target_wings = set()
        for t in created:
            other = (
                t["target"]["wing"] if t["source"]["wing"] == "wing_aya" else t["source"]["wing"]
            )
            target_wings.add(other)
        assert target_wings == {"wing_mempalace", "wing_lantern"}

    def test_entity_tunnels_idempotent_on_rerun(self, tmp_path, monkeypatch):
        """Re-running on the same hallway data must not duplicate tunnels.
        ``create_tunnel`` dedup-on-canonical-id is the guarantee — this test
        pins the contract."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        hallways = [
            {"wing": "wing_aya", "entity_a": "Ben", "entity_b": "Aya"},
            {"wing": "wing_mempalace", "entity_a": "Ben", "entity_b": "Igor"},
        ]
        palace_graph.entity_tunnels_for_wing("wing_aya", hallways)
        palace_graph.entity_tunnels_for_wing("wing_aya", hallways)
        listed = palace_graph.list_tunnels()
        # Only one stored tunnel even after two passes.
        entity_tunnels = [t for t in listed if t.get("kind") == "entity"]
        assert len(entity_tunnels) == 1

    def test_entity_tunnels_retrievable_via_list_tunnels(self, tmp_path, monkeypatch):
        """Once written, entity tunnels appear in the standard
        ``list_tunnels`` query — readers don't need a separate API."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        hallways = [
            {"wing": "wing_aya", "entity_a": "Ben", "entity_b": "Aya"},
            {"wing": "wing_mempalace", "entity_a": "Ben", "entity_b": "Igor"},
        ]
        created = palace_graph.entity_tunnels_for_wing("wing_aya", hallways)
        listed = palace_graph.list_tunnels()
        assert {t["id"] for t in listed} == {t["id"] for t in created}

    def test_entity_tunnels_empty_hallways_is_noop(self, tmp_path, monkeypatch):
        """No hallways → no tunnels. No crash."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        assert palace_graph.entity_tunnels_for_wing("wing_aya", []) == []

    def test_entity_tunnels_unknown_wing_is_noop(self, tmp_path, monkeypatch):
        """Wing that doesn't appear in any hallway → no tunnels (the wing
        has no entities to bridge from)."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        hallways = [
            {"wing": "wing_other", "entity_a": "Ben", "entity_b": "Aya"},
        ]
        assert palace_graph.entity_tunnels_for_wing("wing_nonexistent", hallways) == []

    def test_entity_tunnel_room_does_not_collide_with_literal_room(self, tmp_path, monkeypatch):
        """A literal folder room ``Ben`` (created via explicit tunnel) and the
        synthetic entity room ``entity:Ben`` must produce distinct tunnels —
        not get deduped together by id collision."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        # Pre-create an explicit tunnel with a literal "Ben" room.
        palace_graph.create_tunnel(
            "wing_aya", "Ben", "wing_mempalace", "Ben", label="literal", kind="explicit"
        )
        hallways = [
            {"wing": "wing_aya", "entity_a": "Ben", "entity_b": "Aya"},
            {"wing": "wing_mempalace", "entity_a": "Ben", "entity_b": "Igor"},
        ]
        palace_graph.entity_tunnels_for_wing("wing_aya", hallways)
        listed = palace_graph.list_tunnels()
        # Two distinct tunnels: one literal "Ben" → "Ben", one
        # "entity:Ben" → "entity:Ben". Different canonical IDs.
        assert len(listed) == 2
        kinds = {t.get("kind") for t in listed}
        assert kinds == {"explicit", "entity"}


class TestTunnelDynamicsIntegration:
    """Tunnel records produced by ``create_tunnel`` must carry the L7
    dynamics fields (strength, stability, last_activated, access_count).
    Plus: re-creating a tunnel with the same canonical ID (the dedup path)
    must PRESERVE accumulated dynamics — otherwise every relabeling event
    resets the connection weights."""

    def test_new_tunnel_carries_all_dynamics_fields(self, tmp_path, monkeypatch):
        from mempalace.dynamics import DEFAULT_STABILITY, DEFAULT_STRENGTH

        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        t = palace_graph.create_tunnel(
            "wing_a", "room_x", "wing_b", "room_y", label="initial", kind="explicit"
        )
        assert t["strength"] == DEFAULT_STRENGTH, (
            f"new tunnel should carry default strength; got {t}"
        )
        assert t["stability"] == DEFAULT_STABILITY, (
            f"new tunnel should carry default stability; got {t}"
        )
        assert t["access_count"] == 0
        assert "last_activated" in t
        # last_activated anchored to created_at so decay starts from creation.
        assert t["last_activated"] == t["created_at"]

    def test_recreate_tunnel_preserves_accumulated_dynamics(self, tmp_path, monkeypatch):
        """create_tunnel deduplicates on canonical ID. When called twice
        with the same endpoints, it preserves created_at and adds
        updated_at. It must ALSO preserve accumulated dynamics — otherwise
        a label-update event would wipe the connection's L7 state."""
        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        first = palace_graph.create_tunnel("wing_a", "room_x", "wing_b", "room_y", label="initial")

        # Simulate user activity bumping the tunnel's dynamics.
        stored = palace_graph._load_tunnels()
        for t in stored:
            if t["id"] == first["id"]:
                t["strength"] = 2.7
                t["access_count"] = 12
                t["stability"] = 1.5
        palace_graph._save_tunnels(stored)

        # Recreate same tunnel with a new label — should preserve dynamics.
        second = palace_graph.create_tunnel(
            "wing_a", "room_x", "wing_b", "room_y", label="updated_label"
        )
        assert second["id"] == first["id"]
        assert second["label"] == "updated_label"
        assert second["strength"] == 2.7, (
            f"recreate reset tunnel strength — L7 dynamics undermined; got {second}"
        )
        assert second["access_count"] == 12
        assert second["stability"] == 1.5

    def test_recreate_tunnel_initializes_dynamics_for_legacy_records(self, tmp_path, monkeypatch):
        """If an existing tunnel record was created before L7 (no dynamics
        fields), a recreate event should backfill the defaults rather than
        leave the fields missing."""
        from mempalace.dynamics import DEFAULT_STABILITY, DEFAULT_STRENGTH

        _use_tmp_tunnel_file(monkeypatch, tmp_path)
        legacy_tunnel = {
            "id": palace_graph._canonical_tunnel_id("wing_a", "room_x", "wing_b", "room_y"),
            "source": {"wing": "wing_a", "room": "room_x"},
            "target": {"wing": "wing_b", "room": "room_y"},
            "label": "legacy",
            "kind": "explicit",
            "created_at": "2026-04-01T00:00:00+00:00",
            # NO strength / stability / last_activated / access_count
        }
        palace_graph._save_tunnels([legacy_tunnel])

        # Recreate — should add the missing dynamics fields.
        recreated = palace_graph.create_tunnel(
            "wing_a", "room_x", "wing_b", "room_y", label="updated"
        )
        assert recreated["strength"] == DEFAULT_STRENGTH
        assert recreated["stability"] == DEFAULT_STABILITY
        assert recreated["access_count"] == 0
        assert "last_activated" in recreated