diff --git a/docs/specs/a2a-bus-auth-transition.md b/docs/specs/a2a-bus-auth-transition.md index eb4a558f..8eb06030 100644 --- a/docs/specs/a2a-bus-auth-transition.md +++ b/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 ` 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 diff --git a/docs/specs/tsk-rf5gwb-membership-stems.md b/docs/specs/tsk-rf5gwb-membership-stems.md new file mode 100644 index 00000000..c4cd0160 --- /dev/null +++ b/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-` 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. diff --git a/scripts/measure_membership_stems.py b/scripts/measure_membership_stems.py new file mode 100644 index 00000000..96845ec1 --- /dev/null +++ b/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()) diff --git a/tests/test_measure_membership_stems.py b/tests/test_measure_membership_stems.py new file mode 100644 index 00000000..531d7ac6 --- /dev/null +++ b/tests/test_measure_membership_stems.py @@ -0,0 +1,203 @@ +"""Tests for channel-membership stem measurement.""" + +from __future__ import annotations + +import asyncio +import tempfile +from pathlib import Path + +import pytest + +from scripts.measure_membership_stems import ( + _collect_from_archive, + _collect_from_bus_spool, + _measure_channel, + async_main, + measure, + stem_with_mint, + stem_without_mint, +) + + +class TestStemFunctions: + def test_strip_at_and_casefold(self): + assert stem_without_mint("@taOSmd-dev") == "taosmd-dev" + assert stem_without_mint("taOSmd-dev") == "taosmd-dev" + assert stem_without_mint("@TAOSMD-DEV") == "taosmd-dev" + + def test_mint_stamp_stripped(self): + assert stem_with_mint("taosmd-20260609-153000") == "taosmd" + assert stem_with_mint("@taOSmd-20260609-153000") == "taosmd" + assert stem_with_mint("taosmd-dev") == "taosmd-dev" + + def test_install_discriminator_preserved(self): + assert stem_with_mint("@taOS-agent-abc12345") == "taos-agent-abc12345" + assert stem_with_mint("taOS-agent-abc12345") == "taos-agent-abc12345" + assert stem_with_mint("@taOS-agent-abc12345-20260813-192605") == "taos-agent-abc12345" + + +class TestMeasure: + def test_empty(self): + result = measure([]) + assert result["total_principals"] == 0 + assert result["multi_spell_stems_without_mint"] == {} + assert result["multi_spell_stems_with_mint"] == {} + assert result["canonical_twins"] == [] + assert result["collapse_without_mint"] == {} + assert result["per_channel"] == {} + + def test_single_principal(self): + result = measure([("taosmd-dev", "general")]) + assert result["total_principals"] == 1 + assert result["multi_spell_stems_without_mint"] == {} + assert result["multi_spell_stems_with_mint"] == {} + assert result["per_channel"]["general"]["total_principals"] == 1 + + def test_at_and_bare_same_agent(self): + pairs = [ + ("@taOSmd-dev", "build"), + ("taosmd-dev", "build"), + ] + result = measure(pairs) + assert result["total_principals"] == 2 + assert len(result["multi_spell_stems_without_mint"]) == 1 + assert "taosmd-dev" in result["multi_spell_stems_without_mint"] + assert result["multi_spell_stems_without_mint"]["taosmd-dev"] == [ + "@taOSmd-dev", + "taosmd-dev", + ] + assert result["canonical_twins"] == [] + assert len(result["collapse_without_mint"]) == 1 + assert result["per_channel"]["build"]["total_principals"] == 2 + + def test_canonical_with_bare_twin(self): + pairs = [ + ("taosmd-20260609-153000", "general"), + ("taosmd", "general"), + ] + result = measure(pairs) + assert len(result["canonical_twins"]) == 1 + canonical, twins = result["canonical_twins"][0] + assert canonical == "taosmd-20260609-153000" + assert twins == ["taosmd"] + + def test_canonical_with_at_twin(self): + pairs = [ + ("@taOS-20260609-153000", "general"), + ("taOS", "general"), + ("@taOS", "general"), + ] + result = measure(pairs) + assert len(result["canonical_twins"]) == 1 + canonical, twins = result["canonical_twins"][0] + assert canonical == "@taOS-20260609-153000" + assert set(twins) == {"@taOS", "taOS"} + + def test_two_distinct_principals_same_stem(self): + pairs = [ + ("taosmd-dev", "build"), + ("taos-dev", "general"), + ] + result = measure(pairs) + assert result["collapse_without_mint"] == {} + + def test_mint_stamp_merges_canonical_and_bare(self): + pairs = [ + ("taosmd-20260609-153000", "general"), + ("taosmd-20260813-192605", "general"), + ("taosmd", "general"), + ] + result = measure(pairs) + assert len(result["multi_spell_stems_with_mint"]) == 1 + assert result["multi_spell_stems_with_mint"]["taosmd"] == [ + "taosmd", + "taosmd-20260609-153000", + "taosmd-20260813-192605", + ] + + def test_install_discriminators_not_merged(self): + pairs = [ + ("@taOS-agent-abc12345", "general"), + ("@taOS-agent-def67890", "general"), + ] + result = measure(pairs) + assert result["multi_spell_stems_with_mint"] == {} + assert result["collapse_without_mint"] == {} + + def test_two_canonicals_same_stem_do_not_collapse_without_mint(self): + pairs = [ + ("hermes-20260608-153000", "build"), + ("hermes-20260727-001415", "build"), + ("hermes", "build"), + ] + result = measure(pairs) + assert result["collapse_without_mint"] == {} + assert "hermes" in result["multi_spell_stems_with_mint"] + assert result["multi_spell_stems_with_mint"]["hermes"] == [ + "hermes", + "hermes-20260608-153000", + "hermes-20260727-001415", + ] + + def test_per_channel_reports_separately(self): + pairs = [ + ("@taOSmd-dev", "build"), + ("taosmd-dev", "build"), + ("taosmd-20260609-153000", "general"), + ("taosmd", "general"), + ] + result = measure(pairs) + assert "build" in result["per_channel"] + assert "general" in result["per_channel"] + assert result["per_channel"]["build"]["total_principals"] == 2 + assert result["per_channel"]["general"]["total_principals"] == 2 + assert result["per_channel"]["build"]["collapse_without_mint"] != {} + assert result["per_channel"]["general"]["canonical_twins"] != [] + + +class TestCollectors: + def test_collect_from_bus_spool(self, tmp_path): + spool = tmp_path / "bus-spool.jsonl" + spool.write_text( + '\n'.join([ + '{"body": "[bus/build] @taOSmd-dev: hello"}', + '{"body": "[bus/build] taosmd-dev: world"}', + '{"body": "[bus/general] taOS: test"}', + '{"body": "hermes: [AUTO-ACK]"}', + ]), + encoding="utf-8", + ) + pairs = _collect_from_bus_spool(str(spool)) + assert len(pairs) == 4 + assert ("@taOSmd-dev", "build") in pairs + assert ("taosmd-dev", "build") in pairs + assert ("taOS", "general") in pairs + assert ("hermes", "agent-rules") in pairs + + def test_collect_from_archive_empty_dir(self, tmp_path): + data_dir = tmp_path / "empty" + data_dir.mkdir() + (data_dir / "archive").mkdir() + (data_dir / "archive-index.db").touch() + pairs = asyncio.run(_collect_from_archive(str(data_dir))) + assert pairs == [] + + +class TestAsyncMain: + def test_data_dir_empty_exits_nonzero(self, tmp_path, monkeypatch): + data_dir = tmp_path / "empty" + data_dir.mkdir() + (data_dir / "archive").mkdir() + (data_dir / "archive-index.db").touch() + monkeypatch.setattr(Path, "home", lambda: tmp_path) + rc = asyncio.run(async_main(type("Args", (), {"data_dir": str(data_dir), "spool": None})())) + assert rc == 1 + + def test_spool_flag_uses_spool(self, tmp_path, capsys): + spool = tmp_path / "bus-spool.jsonl" + spool.write_text('{"body": "[bus/build] taosmd-dev: hello"}\n', encoding="utf-8") + args = type("Args", (), {"data_dir": None, "spool": str(spool)})() + rc = asyncio.run(async_main(args)) + assert rc == 0 + captured = capsys.readouterr() + assert "bus-spool.jsonl" in captured.out