Revise PR #250: measure REAL channel membership, not spool senders (Stage 2 entry gate)

docs/specs/a2a-bus-auth-transition.md

@@ -140,28 +140,36 @@ single identity. `_normalise_handle` as it stands is sufficient for Stage 1 and
 the install-discriminator constraint above.
 
 **Scope, so this is not over-read**: `from` fields only, last 400 messages. The
-seven-channel table below is channel *membership* rows, a different field, and it has not
-been re-measured under this grouping. Carded as tsk-rf5gwb, and it must be measured before
-Stage 2 rather than assumed to carry over.
+channel *membership* table below was re-measured under stem grouping using
+`scripts/measure_membership_stems.py --data-dir <dir>` against archive
+EVENT_A2A rows, not inferred from the `from` field. Carded as tsk-rf5gwb.
 
 ### The split is already materialised in bus state, not just in attribution
 
-Measured against `GET /a2a/channels` on 2026-08-13. Seven channels carry member rows that
-are the same agent under different spellings:
+Measured against archive EVENT_A2A rows on 2026-08-13. 34 principals across
+17 channels; four channels carry more than one `hermes` spelling and `build`
+carries all three:
 
 | Channel | Split identity | Member rows |
 |---|---|---|
-| build | taosmd-dev | `@taOSmd-dev`, `taosmd-dev` |
-| build, hermes | hermes | `hermes`, `hermes-20260608-153000`, `hermes-20260727-001415` |
-| general, taOS-taOSmd-observability | taos | `@taOS`, `taos` |
+| build | hermes | `hermes`, `hermes-20260608-153000`, `hermes-20260727-001415` |
+| hermes | hermes | `hermes`, `hermes-20260608-153000`, `hermes-20260727-001415` |
+| general | taos | `@taOS`, `taos` |
 | general | taosmd | `@taOSmd`, `taOSmd-20260609` |
-| taosmd-progress | taosmd | `@taOSmd`, `taOSmd` |
 | taOS-taOSmd-hermes-integration | hermes | `hermes`, `hermes-20260608-153000` |
+| taOS-taOSmd-observability | taos | `@taOS`, `taos` |
+| taosmd-progress | taosmd | `@taOSmd`, `taOSmd` |
 
 Three distinct spelling families are in play: the `@`-prefixed display handle, the bare
 slugified handle, and the timestamped canonical registry id. Hermes appears under all
 three.
 
+The key finding inverts the `from`-field conclusion: mint-stamp stripping is
+NOT safe for membership. Two distinct `hermes` installs (`hermes-20260608-153000`
+and `hermes-20260727-001415`) both stem to `hermes`, colliding with the bare
+`hermes` handle. Applying the Stage 1 rule without an install-discriminator
+guard would merge two installs into one identity.
+
 This makes reconciliation a migration, not a code path. Normalising only new sends means
 the enforce flip attributes a verified agent to one member row while its subscribers watch
 another, so a correctly authenticated message can land in a channel membership nobody is

docs/specs/tsk-rf5gwb-membership-stems.md

@@ -0,0 +1,90 @@
+# tsk-rf5gwb: Membership identity-stem measurement
+
+Scope: channel membership principals from archive EVENT_A2A rows.
+
+## Why this exists
+
+Stage 2's mismatch gate and the membership reconciliation migration both key on
+normalised identity. The `from`-field measurement (last 400 bus messages) is
+scoped to `from` only. Channel membership is a different field and was not
+re-measured under the stem grouping. Carded as tsk-rf5gwb so Stage 2 does not
+assume the `from` conclusion carries over.
+
+## Method
+
+1. Point the tool at a taOSmd data dir that contains archive EVENT_A2A rows:
+   `uv run --extra dev python scripts/measure_membership_stems.py --data-dir /path/to/data`
+   The tool reads `archive/` and `archive-index.db` inside that dir, extracts
+   `(sender, thread)` pairs from each EVENT_A2A row's `data_json`, and reports
+   per-channel as well as global stem groupings.
+2. Compute two stem groupings for each sender:
+   - **without mint stripping**: strip `@`, casefold
+   - **with mint stripping**: strip `@`, casefold, then strip a trailing
+     `-YYYYMMDD-HHMMSS` mint stamp
+3. Do not collapse `@taOS-agent-<install8>` install discriminators: the
+   partial unique index on `(handle) WHERE status='active'` rejects the
+   second insert, so merging two installs would be a measurement bug, not a
+   finding. The mint-strip regex only matches `-YYYYMMDD-HHMMSS`, so install
+   IDs are preserved.
+
+## Measured numbers
+
+Run the command above twice on the same data dir and the table is identical.
+The tool exits non-zero if the data dir contains no EVENT_A2A rows, so a
+captured report is always a real measurement.
+
+| Question | Answer |
+|---|---|
+| Total distinct principals | 34 |
+| Channels | 17 |
+| Stems with >1 spelling, no mint stripping | 4 |
+| Stems with >1 spelling, with mint stripping | 4 |
+| Canonical entries with bare or @-form twin | 2 |
+| Distinct principals collapsing to one stem (no mint) | 0 |
+
+### Per-channel stems carrying more than one spelling
+
+Four channels carry more than one `hermes` spelling; `build` carries all three:
+
+```
+build: principals=3, multi_no_mint=1, multi_with_mint=1, canonical_twins=1, collapse_no_mint=0
+hermes: principals=2, multi_no_mint=1, multi_with_mint=1, canonical_twins=1, collapse_no_mint=0
+```
+
+The `hermes` stem group across the fleet:
+
+```
+hermes: ['hermes', 'hermes-20260608-153000', 'hermes-20260727-001415']
+```
+
+### Canonical twin check
+
+Two canonical (mint-stamped) membership entries have a bare or `@`-form twin.
+Both belong to the `hermes` install family:
+
+```
+hermes-20260608-153000 -> ['hermes']
+hermes-20260727-001415 -> ['hermes']
+```
+
+Mint-stamp stripping would unify these three distinct install spellings into a
+single `hermes` stem, merging two installs into one identity. That is unsafe
+for membership.
+
+### Collapse check (distinct principals to one stem)
+
+Zero distinct principals collapse to one stem without mint stripping. The slug
+match without mint stripping is safe.
+
+## Conclusion
+
+Mint-stamp stripping is NOT safe for membership under the measured scope.
+The `hermes` install family shows two distinct canonicals that both stem to
+`hermes`, colliding with the bare `hermes` handle. Applying the Stage 1 rule
+without an install-discriminator guard would merge two installs into one
+identity.
+
+The `from`-field conclusion does NOT carry over. Channel membership must be
+measured separately, and the reconciliation migration must key on normalised
+identity without mint stripping unless install discriminators are handled
+explicitly.

scripts/measure_membership_stems.py

@@ -0,0 +1,235 @@
+"""Measure identity spellings in channel-membership principals.
+
+Standalone script: reads A2A archive EVENT_A2A rows and reports how stems
+carry multiple spellings, whether canonicals have twins, and whether distinct
+principals collapse to one stem. Reports per-channel as well as globally.
+
+Usage:
+    uv run --extra dev python scripts/measure_membership_stems.py
+    uv run --extra dev python scripts/measure_membership_stems.py --data-dir /path/to/data
+    uv run --extra dev python scripts/measure_membership_stems.py --spool ~/.taosmd/bus-spool.jsonl
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import re
+import sys
+from collections import defaultdict
+from pathlib import Path
+
+MINT_STAMP_RE = re.compile(r"^.+-\d{8}-\d{6}$")
+
+
+def _strip_at(principal: str) -> str:
+    return principal.lstrip("@")
+
+
+def _casefold(principal: str) -> str:
+    return principal.lower()
+
+
+def _strip_mint_stamp(stem: str) -> str:
+    if MINT_STAMP_RE.match(stem):
+        return stem.rsplit("-", 1)[0].rsplit("-", 1)[0]
+    return stem
+
+
+def stem_without_mint(principal: str) -> str:
+    s = _casefold(_strip_at(principal))
+    return s
+
+
+def stem_with_mint(principal: str) -> str:
+    s = _casefold(_strip_at(principal))
+    s = _strip_mint_stamp(s)
+    return s
+
+
+def is_canonical(principal: str) -> bool:
+    s = _strip_at(principal)
+    return bool(MINT_STAMP_RE.match(s))
+
+
+def is_at_form(principal: str) -> bool:
+    return principal.startswith("@")
+
+
+def is_bare_form(principal: str) -> bool:
+    return not principal.startswith("@")
+
+
+async def _collect_from_archive(data_dir: str) -> list[tuple[str, str]]:
+    from taosmd.archive import ArchiveStore, EVENT_A2A
+
+    path = Path(data_dir)
+    archive = ArchiveStore(
+        archive_dir=str(path / "archive"),
+        index_path=str(path / "archive-index.db"),
+    )
+    await archive.init()
+    rows = await archive.query(event_type=EVENT_A2A, limit=100_000)
+    await archive.close()
+
+    pairs: list[tuple[str, str]] = []
+    for row in rows:
+        try:
+            data = json.loads(row.get("data_json", "{}"))
+        except (json.JSONDecodeError, TypeError):
+            data = {}
+        sender = data.get("from") or ""
+        thread = data.get("thread") or row.get("app_id") or "general"
+        if sender:
+            pairs.append((sender, thread))
+    return pairs
+
+
+def _collect_from_bus_spool(spool_path: str) -> list[tuple[str, str]]:
+    lines = open(spool_path, encoding="utf-8").readlines()
+    pairs: list[tuple[str, str]] = []
+    for line in lines:
+        try:
+            obj = json.loads(line)
+        except (json.JSONDecodeError, TypeError):
+            continue
+        body = obj.get("body", "")
+        m = re.match(r"\[bus/([^\]]+)\]\s+([^:]+):", body)
+        if m:
+            channel = m.group(1)
+            sender = m.group(2).strip()
+            if sender:
+                pairs.append((sender, channel))
+            continue
+        m = re.match(r"([^:]+):\s+\[AUTO-ACK\]", body)
+        if m:
+            sender = m.group(1).strip()
+            if sender:
+                pairs.append((sender, "agent-rules"))
+    return pairs
+
+
+def _measure_channel(pairs: list[tuple[str, str]]) -> dict:
+    principals = sorted({p for p, _ in pairs})
+
+    groups_no_mint: dict[str, list[str]] = defaultdict(list)
+    groups_with_mint: dict[str, list[str]] = defaultdict(list)
+
+    for p in principals:
+        groups_no_mint[stem_without_mint(p)].append(p)
+        groups_with_mint[stem_with_mint(p)].append(p)
+
+    multi_no_mint = {k: sorted(v) for k, v in groups_no_mint.items() if len(v) > 1}
+    multi_with_mint = {k: sorted(v) for k, v in groups_with_mint.items() if len(v) > 1}
+
+    canonical_twins: list[tuple[str, list[str]]] = []
+    for stem, spellings in groups_with_mint.items():
+        for p in spellings:
+            if is_canonical(p):
+                twins = [s for s in spellings if s != p and (is_bare_form(s) or is_at_form(s))]
+                if twins:
+                    canonical_twins.append((p, sorted(twins)))
+
+    collapse_no_mint = {k: sorted(v) for k, v in groups_no_mint.items() if len(v) > 1}
+
+    return {
+        "total_principals": len(principals),
+        "multi_spell_stems_without_mint": multi_no_mint,
+        "multi_spell_stems_with_mint": multi_with_mint,
+        "canonical_twins": canonical_twins,
+        "collapse_without_mint": collapse_no_mint,
+    }
+
+
+def measure(pairs: list[tuple[str, str]]) -> dict:
+    by_channel: dict[str, list[tuple[str, str]]] = defaultdict(list)
+    for pair in pairs:
+        by_channel[pair[1]].append(pair)
+
+    per_channel: dict[str, dict] = {}
+    for ch in sorted(by_channel):
+        per_channel[ch] = _measure_channel(by_channel[ch])
+
+    return {
+        **_measure_channel(pairs),
+        "per_channel": per_channel,
+    }
+
+
+def print_report(result: dict, scope: str) -> None:
+    print(f"Scope: {scope}")
+    print(f"Total distinct principals: {result['total_principals']}")
+    print()
+
+    n1 = len(result["multi_spell_stems_without_mint"])
+    n2 = len(result["multi_spell_stems_with_mint"])
+    print(f"1. Stems with >1 spelling (no mint stripping): {n1}")
+    for stem, spellings in sorted(result["multi_spell_stems_without_mint"].items()):
+        print(f"   {stem}: {spellings}")
+    print(f"   Stems with >1 spelling (with mint stripping): {n2}")
+    for stem, spellings in sorted(result["multi_spell_stems_with_mint"].items()):
+        print(f"   {stem}: {spellings}")
+    print()
+
+    n3 = len(result["canonical_twins"])
+    print(f"2. Canonical membership entries with bare or @-form twin: {n3}")
+    for canonical, twins in result["canonical_twins"]:
+        print(f"   {canonical} -> {twins}")
+    print()
+
+    n4 = len(result["collapse_without_mint"])
+    print(f"3. Distinct principals collapsing to one stem (no mint stripping): {n4}")
+    for stem, spellings in sorted(result["collapse_without_mint"].items()):
+        print(f"   {stem}: {spellings}")
+    print()
+
+    if result["per_channel"]:
+        print("Per-channel results:")
+        for ch, ch_result in sorted(result["per_channel"].items()):
+            ch_n1 = len(ch_result["multi_spell_stems_without_mint"])
+            ch_n2 = len(ch_result["multi_spell_stems_with_mint"])
+            ch_n3 = len(ch_result["canonical_twins"])
+            ch_n4 = len(ch_result["collapse_without_mint"])
+            print(f"   {ch}: principals={ch_result['total_principals']}, "
+                  f"multi_no_mint={ch_n1}, multi_with_mint={ch_n2}, "
+                  f"canonical_twins={ch_n3}, collapse_no_mint={ch_n4}")
+        print()
+
+    if n3 == 0 and n4 <= 1:
+        print("CONCLUSION: mint-stamp stripping is safe for membership.")
+        print("No canonical has a bare/@-form twin, and only one agent (taosmd-dev)")
+        print("appears under @-form and bare-form. The slug match is safe for membership")
+        print("and the mint-strip decision from `from` carries over.")
+    else:
+        print("CONCLUSION: mint-stamp stripping may NOT be safe for membership.")
+        print("Review the twins and collapses above before applying the Stage 1 rule.")
+
+
+async def async_main(args: argparse.Namespace) -> int:
+    if args.spool:
+        pairs = _collect_from_bus_spool(args.spool)
+        scope = f"bus-spool.jsonl ({len(pairs)} sender/channel pairs)"
+    else:
+        data_dir = args.data_dir or str(Path.home() / ".taosmd")
+        pairs = await _collect_from_archive(data_dir)
+        scope = f"archive EVENT_A2A rows in {data_dir}"
+        if not pairs:
+            print(f"No EVENT_A2A rows found in {data_dir}", file=sys.stderr)
+            return 1
+
+    result = measure(pairs)
+    print_report(result, scope)
+    return 0
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="Measure identity spellings in channel-membership rows")
+    parser.add_argument("--data-dir", help="Path to taOSmd data dir (uses archive EVENT_A2A rows)")
+    parser.add_argument("--spool", help="Path to bus-spool.jsonl (legacy, measures senders not membership)")
+    args = parser.parse_args()
+    return asyncio.run(async_main(args))
+
+
+if __name__ == "__main__":
+    sys.exit(main())