fix(eventing): decide group-member suppression on the store, not ce_groupid

[mrsabath]

What

Group-member notification suppression now resolves membership from the store instead of
trusting the ce_groupid attribute on the event.

Fixes #891.

Why

groupid = event.get("groupid")
store = self._store_for(event)
if not groupid or store is None:
    return False

The decision was made on an attribute the forger supplies, so a frame that omits it
was not a group member as far as the check was concerned:

• a forged non-terminal frame was pushed, stored in the member's transcript and
  shown on the group page;
• a forged terminal frame was pushed at priority 5 even with
  NTFY_GROUP_NOTIFY_ERRORS off, and did not count the member as failed — so the
  batch's own arithmetic disagreed with the notification just received.

Reachable with signing fully enforced: this is a routing decision, not a signature
check. DESIGN_PHASE2.md §8.5.4 already states the rule — prefer deciding on state the
attacker does not supply.

How

Membership comes from group_of(correlationid) on the event's own routed store. The
method already exists over group_members and is backed by the group_members_by_corr
index, so there is no new query and no migration. Phase 3's per-user _store_for(event)
routing is preserved — the group row is in its owner's store, same as the body lookup.

Redeliveries are unaffected, which is the property worth checking in review, since
the ~47-notifications-per-batch incident is what the current shape was written for. A
genuine member is always recorded before suppression runs:

• both submission paths (group_service.submit_members, handlers) call
  add_group_member before publishing the request, explicitly so a fast agent's
  terminal event can be attributed;
• on_member_event records membership again on arrival, and kafka_in.run() invokes it
  before the ntfy fan-out (_on_member_event, then _on_event).

Behaviour change, called out

test_a_redelivered_member_event_after_completion_still_does_not_notify also asserted
that late-agent-0009 — a correlation id that is not a recorded member but carries
the group's id — stays silent. Under this fix it notifies, because that is exactly the
forgery being closed: an event must not be able to silence itself by asserting a group
it does not belong to. The test is split accordingly — the redelivery still stays
silent, and the unrecorded claimant now notifies, with a dedicated test naming why.

Verification

• tests/test_groups.py — 72 passed, including all five pre-existing suppression tests
  (the regression surface for a membership-source change)
• Full eventing/ suite on this base — 900 passed, 7 skipped (3 new tests)
• ruff check clean
• Both new omission tests fail against the pre-change code, and the failure output
  reproduces this issue's claim verbatim: priority: 5 with errors off
• Checked against the real ordering (membership recorded first): a legitimate member
  carrying ce_groupid is suppressed, the same frame with the attribute deleted is
  suppressed, and a genuinely ungrouped event still notifies
• Nothing run on a cluster (no cluster access)

[aslom]

Summary

The fix is right and the reasoning in the description is the kind worth copying: it
names the invariant (DESIGN_PHASE2.md §8.5.4), shows the omission case rather than
just the forged-value case, and proves the new tests fail against the pre-change code.
group_members_by_corr already exists for precisely this query — store.py's own
schema comment calls it "what lets the responses consumer answer does this event
belong to a group? with a primary-key hit instead of a scan, on the hot path" — so the
new lookup lands where the schema anticipated it. Splitting
test_a_redelivered_member_event_after_completion_still_does_not_notify instead of
quietly deleting the assertion that now inverts is the correct call, and the
redelivery-ordering argument checks out against the source: group_service.submit_members
and handlers both add_group_member before publish_request, and kafka_in.run()
calls _on_member_event before _on_event.

One substantive gap, raised as a suggestion rather than a blocker because it is
pre-existing and not a regression: in multi-tenant mode the store is still selected
from ce_userkey, which is omissible in exactly the same way ce_groupid was, so the
forgery this PR closes remains reachable there. Details inline. OwnerIndex.owner_of()
is the existing non-attacker-supplied answer, and owners is already constructed in
__main__.py and passed to handlers the same way.

One correction to the description, not the code: "there is no new query" holds for the
method and the index, but not for the call count. Previously an event with no
ce_groupid returned before touching the store at all; now every event whose phase is
in NTFY_PHASES does one indexed group_of SELECT, and a grouped event does two store
reads instead of one. That is cheap and clearly intended by the index, so it is a wording
nit — but a future reader debugging consumer latency should not be told there is no new
query on the path.

Author: mrsabath (MEMBER — maintainer)
Areas reviewed: Python (eventing/eventbridge/ntfy.py), Tests (eventing/tests/test_groups.py)
Agent/IDE config (.claude/.vscode): none
Commits: 1 commit, all signed-off: yes (DCO passing)
CI status: passing — all 12 checks green (eventing-test, lint, codeql, trivy-scan, dependency-review, hadolint, test, test-startup)
Verified against: head 13285619, source read at that SHA (ntfy.py, store.py, store_registry.py, group_service.py, handlers.py, kafka_in.py, owner_index.py, shared/signing.py, shared/ce.py, __main__.py)