"""M6 — the soul. Plan §4.2's test cases as an executable suite over
courts_of/visible_players/rep (the fleshout worked example, every
assertion derivable from the three functions), plus the API-level
projections of §4.3/§4.3b: rally collapse, comment scoping, board
assignment rep + holder_label + can_spike, viewer-scoped rosters,
mentionable, write-side target validation, and per-subscriber SSE
projection (recipient-only notifications included)."""
from __future__ import annotations

import datetime

import pytest

from app.models import Ball, Touch
from app.services import projection as projection_service
from app.services.perspective import MatchTopology, PerspectiveLens
from tests.conftest import make_fleshout_match, make_player, mint_client_for


@pytest.fixture()
def world(db_session):
    return make_fleshout_match(db_session)


def _topology(db, world) -> MatchTopology:
    return MatchTopology(db, world["match"].id)


def _ids(world, *usernames) -> set:
    return {world["players"][u].id for u in usernames}


# =====================================================================
# Part 1 — §4.2 verbatim: courts_of / visible_players / rep
# =====================================================================

# courts_of(V): [(court name, is_root, near usernames, far usernames,
# far_label)], in band order (member court first, then boundary courts).
COURTS_OF_CASES = {
    "rian": [("Front line", True, {"rian", "adi"}, {"erin", "tracy"}, "Brentwood")],
    "adi": [
        ("Front line", True, {"rian", "adi"}, {"erin", "tracy"}, "Brentwood"),
        ("Tingang", False, {"zeina", "rahman", "stevan"}, {"adi"}, "Bowden Works"),
    ],
    "zeina": [
        ("Tingang", False, {"zeina", "rahman", "stevan"}, {"adi"}, "Bowden Works")
    ],
    "erin": [
        ("Front line", True, {"erin", "tracy"}, {"rian", "adi"}, "Bowden Works"),
        ("Back office", False, {"rob"}, {"erin"}, "Brentwood"),
    ],
    "tracy": [("Front line", True, {"erin", "tracy"}, {"rian", "adi"}, "Bowden Works")],
    "rob": [("Back office", False, {"rob"}, {"erin"}, "Brentwood")],
}


@pytest.mark.parametrize("viewer", sorted(COURTS_OF_CASES))
def test_courts_of(db_session, world, viewer):
    topo = _topology(db_session, world)
    got = topo.courts_of(world["players"][viewer].id)
    expected = COURTS_OF_CASES[viewer]
    assert len(got) == len(expected)
    for court, (name, is_root, near, far, far_label) in zip(got, expected):
        assert court.name == name
        assert court.is_root == is_root
        assert set(court.near) == _ids(world, *near)
        assert set(court.far) == _ids(world, *far)
        assert court.far_label == far_label


VISIBLE_CASES = {
    "rian": {"rian", "adi", "erin", "tracy"},
    "adi": {"rian", "adi", "erin", "tracy", "zeina", "rahman", "stevan"},
    "zeina": {"zeina", "rahman", "stevan", "adi"},
    "erin": {"erin", "tracy", "rian", "adi", "rob"},
    "tracy": {"erin", "tracy", "rian", "adi"},
    "rob": {"rob", "erin"},
}


@pytest.mark.parametrize("viewer", sorted(VISIBLE_CASES))
def test_visible_players(db_session, world, viewer):
    topo = _topology(db_session, world)
    got = topo.visible_players(world["players"][viewer].id)
    assert got == _ids(world, *VISIBLE_CASES[viewer])


# rep(P, V): (player, viewer) -> a username (collapse target) or a
# ("side", label) bottom-out. Every fleshout collapse assertion.
REP_CASES = [
    # Rian never sees Tingang or Rob — they collapse into their leads.
    ("zeina", "rian", "adi"),
    ("rahman", "rian", "adi"),
    ("stevan", "rian", "adi"),
    ("rob", "rian", "erin"),
    # Erin sees Adi's blob the same way from across the big net.
    ("zeina", "erin", "adi"),
    ("rahman", "tracy", "adi"),
    ("rob", "tracy", "erin"),
    # Visible players rep as themselves.
    ("adi", "rian", "adi"),
    ("zeina", "adi", "zeina"),
    ("rob", "erin", "rob"),
    ("rian", "erin", "rian"),
    # From inside a sub-court, EVERYTHING beyond your net is the far-side
    # label of YOUR court (naming Brentwood to Zeina would leak).
    ("erin", "zeina", ("side", "Bowden Works")),
    ("rian", "zeina", ("side", "Bowden Works")),
    ("tracy", "zeina", ("side", "Bowden Works")),
    ("rob", "zeina", ("side", "Bowden Works")),
    # Rob's world ends at Erin: everyone else bottoms out at his label.
    ("rian", "rob", ("side", "Brentwood")),
    ("tracy", "rob", ("side", "Brentwood")),
    ("zeina", "rob", ("side", "Brentwood")),
]


@pytest.mark.parametrize("player,viewer,expected", REP_CASES)
def test_rep(db_session, world, player, viewer, expected):
    topo = _topology(db_session, world)
    rep = topo.rep(world["players"][player].id, world["players"][viewer].id)
    if isinstance(expected, tuple):
        assert rep.kind == "side"
        assert rep.label == expected[1]
    else:
        assert rep.kind == "player"
        assert rep.player_id == world["players"][expected].id


def test_lens_rep_matches_topology_rep(db_session, world):
    """The memoized PerspectiveLens must agree with the spec function for
    every (player, viewer) pair — it is an optimization, never a fork."""
    topo = _topology(db_session, world)
    for viewer in world["players"].values():
        lens = PerspectiveLens(topo, viewer.id)
        for player in world["players"].values():
            assert lens.rep(player.id) == topo.rep(player.id, viewer.id)


# ---------------------------------------------------------------------
# Observer default-deny: no membership in the match = the most-collapsed
# view (front-line-level data of both sides), never more.
# ---------------------------------------------------------------------


def test_outside_observer_gets_front_lines_only(db_session, world):
    topo = _topology(db_session, world)
    sam = make_player(db_session, "sam", display_name="Sam")  # no membership

    for viewer_id in (None, sam.id):
        assert not topo.is_participant(viewer_id)
        courts = topo.effective_courts(viewer_id)
        assert len(courts) == 1
        assert courts[0].is_root
        assert set(courts[0].near) == _ids(world, "rian", "adi")  # side 1
        assert set(courts[0].far) == _ids(world, "erin", "tracy")
        assert courts[0].far_label == "Brentwood"
        visible = topo.effective_visible(viewer_id)
        assert visible == _ids(world, "rian", "adi", "erin", "tracy")
        # Sub-team members collapse into their leads, exactly as for Rian.
        lens = PerspectiveLens(topo, viewer_id)
        assert lens.rep(world["players"]["zeina"].id).player_id == world["players"]["adi"].id
        assert lens.rep(world["players"]["rob"].id).player_id == world["players"]["erin"].id


def test_node_visibility_rule(db_session, world):
    """The documented comment-scoping rule: your courts' nodes + your far
    players' nodes; NULL always visible."""
    topo = _topology(db_session, world)
    p, n = world["players"], world["nodes"]
    # Rian: both front lines, neither sub-team.
    assert topo.node_visible(p["rian"].id, n["bowden_front"].id)
    assert topo.node_visible(p["rian"].id, n["brentwood_front"].id)
    assert not topo.node_visible(p["rian"].id, n["tingang"].id)
    assert not topo.node_visible(p["rian"].id, n["back_office"].id)
    # Adi adds Tingang; Erin adds Back office; Zeina sees Tingang + the
    # far boundary's node (Adi's front line) but NOT Brentwood's contexts.
    assert topo.node_visible(p["adi"].id, n["tingang"].id)
    assert topo.node_visible(p["erin"].id, n["back_office"].id)
    assert not topo.node_visible(p["erin"].id, n["tingang"].id)
    assert topo.node_visible(p["zeina"].id, n["tingang"].id)
    assert topo.node_visible(p["zeina"].id, n["bowden_front"].id)
    assert not topo.node_visible(p["zeina"].id, n["brentwood_front"].id)
    # NULL node (legacy/outside) is visible to everyone.
    assert topo.node_visible(p["rob"].id, None)


# =====================================================================
# Part 2 — reading through the projection (§4.3) at the API
# =====================================================================


def _add_touch(db, world, ball, seq, kind, frm, to, minutes_ago=0):
    p = world["players"]
    member_node = {
        "rian": "bowden_front",
        "adi": "bowden_front",
        "zeina": "tingang",
        "rahman": "tingang",
        "stevan": "tingang",
        "erin": "brentwood_front",
        "tracy": "brentwood_front",
        "rob": "back_office",
    }
    db.add(
        Touch(
            ball_id=ball.id,
            seq=seq,
            kind=kind,
            from_player=p[frm].id if frm else None,
            to_players=[p[u].id for u in to],
            acting_node_id=world["nodes"][member_node[frm]].id if frm else None,
            created_at=datetime.datetime.now(datetime.timezone.utc)
            - datetime.timedelta(minutes=minutes_ago),
        )
    )
    db.commit()


def _fleshout_rally_ball(db, world) -> Ball:
    """The fleshout chain: erin serve -> rian pass adi -> adi pass zeina
    -> zeina pass rahman -> rahman back to adi -> adi spike rian."""
    p = world["players"]
    ball = Ball(
        court_id=world["court"].id,
        title="Hero block",
        status="Ready for Review",
        category="Revision",
        action_to=[p["rian"].id],
        prev_action_to=[p["adi"].id],
        author_id=p["erin"].id,
    )
    db.add(ball)
    db.commit()
    chain = [
        (1, "serve", "erin", ["rian"]),
        (2, "pass", "rian", ["adi"]),
        (3, "pass", "adi", ["zeina"]),
        (4, "pass", "zeina", ["rahman"]),
        (5, "pass", "rahman", ["adi"]),
        (6, "spike", "adi", ["rian"]),
    ]
    for seq, kind, frm, to in chain:
        _add_touch(db, world, ball, seq, kind, frm, to, minutes_ago=60 - seq * 5)
    return ball


def test_rally_collapses_tingang_stretch_for_rian(client, db_session, world):
    ball = _fleshout_rally_ball(db_session, world)
    c = mint_client_for(client, "rian")
    entries = c.get(f"/api/balls/{ball.id}/rally").json()["touches"]
    assert [e["kind"] for e in entries] == ["serve", "pass", "pass", "collapsed", "spike"]
    collapsed = entries[3]
    assert collapsed["count"] == 2  # zeina -> rahman, rahman -> adi
    assert collapsed["label"] == "the Bowden Works side"
    # Adi's inward pass keeps its actor; its target reps to Adi himself.
    adi_id = str(world["players"]["adi"].id)
    assert entries[2]["from_player"] == adi_id
    assert entries[2]["to_players"] == [adi_id]


def test_rally_full_detail_for_adi(client, db_session, world):
    ball = _fleshout_rally_ball(db_session, world)
    c = mint_client_for(client, "adi")
    entries = c.get(f"/api/balls/{ball.id}/rally").json()["touches"]
    assert [e["kind"] for e in entries] == ["serve", "pass", "pass", "pass", "pass", "spike"]
    assert all(e["kind"] != "collapsed" for e in entries)


def test_rally_for_zeina_collapses_everything_beyond_her_net(client, db_session, world):
    ball = _fleshout_rally_ball(db_session, world)
    c = mint_client_for(client, "zeina")
    entries = c.get(f"/api/balls/{ball.id}/rally").json()["touches"]
    # erin + rian touches collapse; the label is HER far-side label, never
    # "Brentwood" (§4.2's deliberate structure-leak guard).
    assert entries[0]["kind"] == "collapsed"
    assert entries[0]["count"] == 2
    assert entries[0]["label"] == "the Bowden Works side"
    # adi -> zeina, zeina -> rahman, rahman -> adi in full; the spike's
    # target (rian) is beyond her world, so its to_players project empty.
    assert [e["kind"] for e in entries[1:]] == ["pass", "pass", "pass", "spike"]
    assert entries[4]["to_players"] == []


def test_comment_scoping(client, db_session, world):
    """§4.3: Rian gets the collapsed line for Tingang's internal chatter;
    Adi reads it. A visible-node comment by an invisible author (Tracy,
    viewed by Rob) keeps its body but reps its author to the side label."""
    from app.models import Comment

    p, n = world["players"], world["nodes"]
    ball = Ball(court_id=world["court"].id, title="Hero block", status="Open",
                category="Revision", author_id=p["erin"].id)
    db_session.add(ball)
    db_session.commit()
    base = datetime.datetime.now(datetime.timezone.utc)
    db_session.add_all(
        [
            Comment(parent_type="ball", parent_id=ball.id, author_id=p["erin"].id,
                    acting_node_id=n["brentwood_front"].id,
                    body="Can we tighten the hero copy?",
                    created_at=base - datetime.timedelta(minutes=30)),
            Comment(parent_type="ball", parent_id=ball.id, author_id=p["zeina"].id,
                    acting_node_id=n["tingang"].id,
                    body="internal: refactoring the hero block",
                    created_at=base - datetime.timedelta(minutes=20)),
            Comment(parent_type="ball", parent_id=ball.id, author_id=p["rahman"].id,
                    acting_node_id=n["tingang"].id,
                    body="internal: srcset cache is the culprit",
                    created_at=base - datetime.timedelta(minutes=10)),
        ]
    )
    db_session.commit()

    rian = mint_client_for(client, "rian")
    entries = rian.get(f"/api/balls/{ball.id}/comments").json()
    assert [e["kind"] for e in entries] == ["comment", "collapsed_comments"]
    assert entries[0]["body"] == "Can we tighten the hero copy?"
    assert entries[1]["count"] == 2
    assert "body" not in entries[1]  # no bodies, no authors

    adi = mint_client_for(client, "adi")
    entries = adi.get(f"/api/balls/{ball.id}/comments").json()
    assert [e["kind"] for e in entries] == ["comment", "comment", "comment"]

    # Tracy comments in front-line context; Rob sees the node (his far
    # boundary's) but not Tracy — author anonymizes to the side label.
    db_session.add(
        Comment(parent_type="ball", parent_id=ball.id, author_id=p["tracy"].id,
                acting_node_id=n["brentwood_front"].id,
                body="Client asked for a Thursday check-in.",
                created_at=base)
    )
    db_session.commit()
    rob = mint_client_for(client, "rob")
    entries = rob.get(f"/api/balls/{ball.id}/comments").json()
    tracy_comment = entries[-1]
    assert tracy_comment["kind"] == "comment"
    assert tracy_comment["author_id"] is None
    assert tracy_comment["author_label"] == "Brentwood"


def test_board_action_to_reps_and_roster_scopes(client, db_session, world):
    p = world["players"]
    ball = Ball(court_id=world["court"].id, title="Held by Zeina", status="Open",
                category="Revision", action_to=[p["zeina"].id], author_id=p["erin"].id)
    db_session.add(ball)
    db_session.commit()

    for viewer in ("rian", "erin"):
        c = mint_client_for(client, viewer)
        body = c.get("/api/matches/fleshout/board").json()
        got = body["courts"][0]["balls"][0]
        assert got["action_to"] == [str(p["adi"].id)]  # zeina reps to adi
        assert got["holder_label"] is None
        usernames = {pl["username"] for pl in body["players"]}
        assert "zeina" not in usernames  # roster is the visible set
        assert set(body["visible_player_ids"]) == {
            str(i) for i in _ids(world, *VISIBLE_CASES[viewer])
        }


def test_board_holder_label_when_fully_collapsed(client, db_session, world):
    p = world["players"]
    ball = Ball(court_id=world["court"].id, title="Held by Erin", status="Open",
                category="Revision", action_to=[p["erin"].id], author_id=p["rian"].id)
    db_session.add(ball)
    db_session.commit()
    c = mint_client_for(client, "zeina")
    got = c.get("/api/matches/fleshout/board").json()["courts"][0]["balls"][0]
    assert got["action_to"] == []  # erin is beyond zeina's world
    assert got["holder_label"] == "Bowden Works"


def test_board_strip_bands_and_sides(client, db_session, world):
    p = world["players"]
    adi = mint_client_for(client, "adi")
    body = adi.get("/api/matches/fleshout/board").json()
    assert [b["name"] for b in body["strip"]] == ["Front line", "Tingang"]
    assert body["strip"][1]["far"] == [str(p["adi"].id)]
    assert body["strip"][1]["far_label"] == "Bowden Works"
    assert [s["name"] for s in body["sides"]] == ["Bowden Works", "Brentwood"]

    zeina = mint_client_for(client, "zeina")
    body = zeina.get("/api/matches/fleshout/board").json()
    assert [b["name"] for b in body["strip"]] == ["Tingang"]
    # A sub-team-only viewer's payload names ONLY her own root side.
    assert [s["name"] for s in body["sides"]] == ["Bowden Works"]


def test_board_can_spike(client, db_session, world):
    """can_spike is false exactly when the auto-spike target (the author)
    is beyond the viewer's net — Rahman on a big-net ball."""
    p = world["players"]
    ball = Ball(court_id=world["court"].id, title="Erin's ask", status="Addressing",
                category="Revision", action_to=[p["rahman"].id], author_id=p["erin"].id)
    db_session.add(ball)
    db_session.commit()

    rahman = mint_client_for(client, "rahman")
    got = rahman.get("/api/matches/fleshout/board").json()["courts"][0]["balls"][0]
    assert got["can_spike"] is False

    adi = mint_client_for(client, "adi")
    got = adi.get("/api/matches/fleshout/board").json()["courts"][0]["balls"][0]
    assert got["can_spike"] is True


def test_mentionable_offers_only_visible_players(client, db_session, world):
    erin = mint_client_for(client, "erin")
    names = {m["username"] for m in erin.get("/api/matches/fleshout/mentionable").json()}
    assert names == VISIBLE_CASES["erin"]

    zeina = mint_client_for(client, "zeina")
    names = {m["username"] for m in zeina.get("/api/matches/fleshout/mentionable").json()}
    assert names == VISIBLE_CASES["zeina"]


def test_notification_actor_reps_for_recipient(client, db_session, world):
    """Zeina comments on Rian's ball with @rian-style participation: the
    thread_reply Rian receives renders its actor as Adi (rep), never
    Zeina. Driven through the real comment endpoint."""
    p = world["players"]
    ball = Ball(court_id=world["court"].id, title="Rian's serve", status="Open",
                category="Revision", author_id=p["rian"].id,
                action_to=[p["zeina"].id])
    db_session.add(ball)
    db_session.commit()
    # Zeina's comment is Tingang-scoped -> rian is NOT a permitted
    # thread_reply recipient (write-side scoping). Use rahman -> adi
    # instead: adi CAN see Tingang, so the notification exists, and for
    # ADI the actor rahman is visible. For the rep case use a pass:
    # zeina passes to adi (visible write) -> adi auto... no rule here, so
    # adi simply receives action_to with actor zeina — visible to him.
    # The side-label case: rob passes to erin; tracy's bell shows...
    # tracy gets nothing (not a participant). The cleanest rep case that
    # can ARISE honestly: zeina passes the ball to adi -> RIAN is not
    # notified; but a pass to a ball AUTHORED by rian notifies nobody
    # else. So drive it via action_to: zeina passes to adi while adi's
    # bell renders actor zeina (visible). The boundary-collapse case
    # needs a recipient who cannot see the actor: erin assigned by
    # zeina's pass — zeina cannot target erin (invisible). By §4.3b no
    # honest write produces it on the big court; the projection is still
    # exercised directly below.
    zeina = mint_client_for(client, "zeina")
    resp = zeina.post(f"/api/balls/{ball.id}/pass", json={"to_players": [str(p["adi"].id)]})
    assert resp.status_code == 200

    adi = mint_client_for(client, "adi")
    items = adi.get("/api/notifications").json()["items"]
    assert len(items) == 1
    assert items[0]["actor_id"] == str(p["zeina"].id)  # visible actor: raw

    # Direct projection check for the collapse + side-label paths (the
    # stored row keeps raw truth; the read projects).
    topo = _topology(db_session, world)
    lens_rian = PerspectiveLens(topo, p["rian"].id)
    projected = projection_service.project_notification_actor(
        lens_rian, {"actor_id": p["zeina"].id}
    )
    assert projected["actor_id"] == p["adi"].id
    lens_zeina = PerspectiveLens(topo, p["zeina"].id)
    projected = projection_service.project_notification_actor(
        lens_zeina, {"actor_id": p["erin"].id}
    )
    assert projected["actor_id"] is None
    assert projected["actor_label"] == "Bowden Works"


def test_write_side_pass_and_serve_target_validation(client, db_session, world):
    """§4.3b: passing or serving to a player you cannot see is 422
    TARGET_NOT_VISIBLE; passing to the boundary (the rep target) works."""
    p = world["players"]
    ball = Ball(court_id=world["court"].id, title="Big-net ball", status="Open",
                category="Revision", action_to=[p["erin"].id], author_id=p["erin"].id)
    db_session.add(ball)
    db_session.commit()

    erin = mint_client_for(client, "erin")
    resp = erin.post(f"/api/balls/{ball.id}/pass", json={"to_players": [str(p["zeina"].id)]})
    assert resp.status_code == 422
    assert resp.json()["error_code"] == "TARGET_NOT_VISIBLE"
    resp = erin.post(f"/api/balls/{ball.id}/pass", json={"to_players": [str(p["adi"].id)]})
    assert resp.status_code == 200

    # The serve sheet: same rule at creation.
    resp = erin.post(
        "/api/matches/fleshout/balls",
        json={"court_id": str(world["court"].id), "title": "Sneaky serve",
              "action_to": [str(p["rahman"].id)]},
    )
    assert resp.status_code == 422
    assert resp.json()["error_code"] == "TARGET_NOT_VISIBLE"

    # And the explicit action_to patch (a pass by another spelling).
    resp = erin.patch(f"/api/balls/{ball.id}", json={"action_to": [str(p["zeina"].id)]})
    assert resp.status_code == 422
    assert resp.json()["error_code"] == "TARGET_NOT_VISIBLE"


def test_write_side_spike_blocked_when_author_invisible(client, db_session, world):
    p = world["players"]
    ball = Ball(court_id=world["court"].id, title="Erin's ask", status="Addressing",
                category="Revision", action_to=[p["rahman"].id], author_id=p["erin"].id)
    db_session.add(ball)
    db_session.commit()

    rahman = mint_client_for(client, "rahman")
    resp = rahman.patch(f"/api/balls/{ball.id}", json={"status": "Ready for Review"})
    assert resp.status_code == 422
    assert resp.json()["error_code"] == "AUTOMATION_TARGET_NOT_VISIBLE"

    # Adi CAN see the author — the same patch from him spikes normally.
    adi = mint_client_for(client, "adi")
    resp = adi.patch(f"/api/balls/{ball.id}", json={"status": "Ready for Review"})
    assert resp.status_code == 200
    assert resp.json()["action_to"] == [str(p["erin"].id)]


# =====================================================================
# Part 3 — SSE per-subscriber projection (pure project_event)
# =====================================================================


def _pass_event(world, ball_id, frm, to, kind="pass"):
    p = world["players"]
    return {
        "type": "ball.updated",
        "ball_id": str(ball_id),
        "match_id": str(world["match"].id),
        "action_to": [str(p[u].id) for u in to],
        "touch": {
            "kind": kind,
            "from_player": str(p[frm].id),
            "to_players": [str(p[u].id) for u in to],
            "auto": False,
        },
    }


def test_project_event_drops_invisible_inner_pass(db_session, world):
    topo = _topology(db_session, world)
    p = world["players"]
    evt = _pass_event(world, "00000000-0000-0000-0000-000000000001", "zeina", ["rahman"])
    # A Tingang inner pass never reaches Rian's stream...
    assert projection_service.project_event(evt, topo, p["rian"].id) is None
    # ...nor Zeina's over-the-net pass to Adi (Rian's holder view is
    # unchanged: the ball was "with Adi" before and after).
    evt = _pass_event(world, "00000000-0000-0000-0000-000000000001", "zeina", ["adi"])
    assert projection_service.project_event(evt, topo, p["rian"].id) is None
    # Adi sees the inner pass in full.
    evt = _pass_event(world, "00000000-0000-0000-0000-000000000001", "zeina", ["rahman"])
    got = projection_service.project_event(evt, topo, p["adi"].id)
    assert got is not None
    assert got["touch"]["from_player"] == str(p["zeina"].id)


def test_project_event_keeps_visible_changes(db_session, world):
    topo = _topology(db_session, world)
    p = world["players"]
    # Adi passes to Rian, watched by Zeina: actor visible, target beyond
    # her net — delivered, with the target projected away.
    evt = _pass_event(world, "00000000-0000-0000-0000-000000000002", "adi", ["rian"])
    got = projection_service.project_event(evt, topo, p["zeina"].id)
    assert got is not None
    assert got["action_to"] == []
    assert got["holder_label"] == "Bowden Works"
    assert got["touch"]["to_players"] == []
    # A resolve by an invisible actor still lands (the status flip is
    # board-visible) but sheds its actor.
    evt = _pass_event(world, "00000000-0000-0000-0000-000000000003", "zeina", [], kind="resolve")
    evt["action_to"] = []
    evt["touch"]["to_players"] = []
    got = projection_service.project_event(evt, topo, p["rian"].id)
    assert got is not None
    assert got["touch"]["from_player"] is None
    assert got["touch"]["from_label"]


def test_project_event_notifications_are_recipient_only(db_session, world):
    topo = _topology(db_session, world)
    p = world["players"]
    evt = {
        "type": "notification.created",
        "recipient_id": str(p["adi"].id),
        "match_id": str(world["match"].id),
        "notification": {"actor_id": str(p["zeina"].id), "kind": "action_to"},
    }
    # Only the recipient's stream carries it — the M5 broadcast leak fix.
    assert projection_service.project_event(evt, topo, p["rian"].id) is None
    assert projection_service.project_event(evt, topo, None) is None
    got = projection_service.project_event(evt, topo, p["adi"].id)
    assert got is not None
    assert got["notification"]["actor_id"] == str(p["zeina"].id)


def test_project_event_reps_notification_actor_for_recipient(db_session, world):
    """A notification whose actor is beyond the recipient's world arrives
    with the actor repped (id swapped or side-labelled) — the bell can
    never learn an invisible id from the live stream."""
    topo = _topology(db_session, world)
    p = world["players"]
    evt = {
        "type": "notification.created",
        "recipient_id": str(p["rian"].id),
        "match_id": str(world["match"].id),
        "notification": {"actor_id": str(p["zeina"].id), "kind": "thread_reply"},
    }
    got = projection_service.project_event(evt, topo, p["rian"].id)
    assert got["notification"]["actor_id"] == str(p["adi"].id)