From b48646bc0cee2e23c598e33a5958d36c1bf2e989 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Wed, 29 Apr 2026 16:07:24 -0400 Subject: [PATCH] mesh: sync default-holds records younger than 1 hour (kindergarten window) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fox 2026-04-29: "part of the gossip protocol should be some default delay that allows a user to catch and burn kindergarteners before they are synced." Currently `mesh sync` enumerates the most-recent N items with no age filter — a doc ingested 30 seconds ago goes out on the next sync, & if a peer ingests it before the operator notices a problem, burn no longer suffices (peer has its own copy). Adds a sender-side kindergarten window: --kindergarten-seconds N default 3600 (1 hour) Records younger than `now - N` are held back from broadcast. Both ANNOUNCE_ROOT (filtered on documents.ingest_ts) & ANNOUNCE_FALSIFICATION (filtered on falsifications.at). N=0 = broadcast everything (cron- friendly opt-out for operators preferring immediate propagation). Result JSON now reports the held counts so the operator can see what the window protected: kindergarten_seconds: 3600 announced_roots: 12 kindergarten_held_roots: 3 announced_falsifications: 0 kindergarten_held_falsifications: 1 Sender-side discipline only — the receiver has no view into when the sender created the record, so it can't enforce. Adding a created_at on the envelope would let receivers reject too-fresh gossip, but that's a future protocol bump (envelopes today don't carry sender wall-clock; ts is the send time). Tests: - existing tests now pass `--kindergarten-seconds 0` so freshly- ingested fixtures broadcast immediately for the test - new test_sync_kindergarten_holds_fresh_records: a 30-second-old doc is held; an artificially-aged doc broadcasts. announced=1, held>=1. - new test_sync_kindergarten_zero_broadcasts_everything: explicit opt-out works. 342 passed, 1 skipped. --- aborist/cli.py | 97 +++++++++++++++++++++++++++++++++--------- tests/test_mesh_cli.py | 81 +++++++++++++++++++++++++++++++++++ 2 files changed, 157 insertions(+), 21 deletions(-) diff --git a/aborist/cli.py b/aborist/cli.py index 15e52ab..42cfe0f 100644 --- a/aborist/cli.py +++ b/aborist/cli.py @@ -533,6 +533,12 @@ def _render_inspect_human(result: dict) -> str: if diag == "trailing_artifact": lines.append(f" matched_prefix_chars: {d.get('matched_prefix_chars')}") lines.append(f" trailing_artifact: {_short(d.get('trailing_artifact', ''), 100)}") + elif diag == "interior_elision": + lines.append( + f" matched: {d.get('matched_prefix_chars')} prefix + " + f"{d.get('matched_suffix_chars')} suffix chars (parenthetical aside dropped)" + ) + lines.append(f" dropped_aside: {_short(d.get('dropped_aside', ''), 120)}") elif diag in ("paraphrase", "partial_paraphrase"): lines.append(f" token_coverage: {d.get('token_coverage')}") counts = d.get("token_counts", {}) @@ -2127,22 +2133,37 @@ def _cmd_mesh_sync(args: argparse.Namespace) -> int: Two pushes happen by default (unless ``--no-roots`` / ``--no-falsifications`` opts one out): - 1. **ANNOUNCE_ROOT** for the most-recent ``--limit`` documents. - Receivers that already have the root no-op (idempotent on the - receiver via dedup against ``documents.document_root``). + 1. **ANNOUNCE_ROOT** for the most-recent ``--limit`` documents + older than the kindergarten window. Receivers dedup by + ``documents.document_root``. 2. **ANNOUNCE_FALSIFICATION** for the most-recent ``--limit`` - falsifications. Per the protocol, this is the audit-preserving - channel for "this answer is wrong" — burns are deliberately - NOT propagated (kindergarten-only local cleanup). + falsifications older than the kindergarten window. Burns + deliberately NOT propagated — local kindergarten cleanup. + + **Kindergarten window.** Records younger than + ``--kindergarten-seconds`` (default 3600 = 1 hour) are NOT + broadcast. Gives the operator time to inspect a fresh ingest or + falsification & burn it before the network sees it. Override per + invocation; ``--kindergarten-seconds 0`` broadcasts everything + (cron-friendly opt-out for operators who prefer immediate + propagation). The window is sender-side discipline; receivers + don't enforce it because they have no view into when the sender + created the record. Receivers verify the Ed25519 signature, run per-peer chain-of- claims fork detection, then write one ``mesh_received`` audit event per accepted envelope. Duplicate broadcasts produce - duplicate audit-log entries on the receiver but no state corruption. + duplicate audit-log entries on the receiver but no state + corruption. """ + import time as _time + from aborist.mesh import is_enabled, load_identity from aborist.mesh.wire import MeshWireClient + now_ts = int(_time.time()) + cutoff_ts = now_ts - max(0, args.kindergarten_seconds) + conn = connect(args.db) try: if load_identity(conn) is None: @@ -2151,28 +2172,47 @@ def _cmd_mesh_sync(args: argparse.Namespace) -> int: if not is_enabled(conn): print("error: mesh.enabled is off", file=sys.stderr) return 2 - root_rows = ( - [] - if args.no_roots - else conn.execute( + # Total counts inform skipped-by-kindergarten reporting. + total_roots = 0 + total_falsifications = 0 + root_rows: list = [] + falsification_rows: list = [] + if not args.no_roots: + total_roots = conn.execute( + "SELECT COUNT(*) FROM documents" + ).fetchone()[0] + root_rows = conn.execute( "SELECT document_root, document_uri, chunking_version, " " canonicalization_version, schema_version " - "FROM documents ORDER BY rowid DESC LIMIT ?", - (args.limit,), + "FROM documents WHERE ingest_ts <= ? " + "ORDER BY rowid DESC LIMIT ?", + (cutoff_ts, args.limit), ).fetchall() - ) - falsification_rows = ( - [] - if args.no_falsifications - else conn.execute( + if not args.no_falsifications: + total_falsifications = conn.execute( + "SELECT COUNT(*) FROM falsifications" + ).fetchone()[0] + falsification_rows = conn.execute( "SELECT cache_key, reason FROM falsifications " + "WHERE at <= ? " "ORDER BY at DESC LIMIT ?", - (args.limit,), + (cutoff_ts, args.limit), ).fetchall() - ) finally: conn.close() + # Skipped-by-kindergarten = (rows younger than cutoff that would have + # been in the most-recent --limit) — approximated by total minus what + # we pulled, capped at limit. + fresh_roots_held = max( + 0, + min(total_roots, args.limit) - len(root_rows), + ) if not args.no_roots else 0 + fresh_falsifications_held = max( + 0, + min(total_falsifications, args.limit) - len(falsification_rows), + ) if not args.no_falsifications else 0 + sent_roots: list[dict] = [] sent_falsifications: list[dict] = [] errors: list[dict] = [] @@ -2209,8 +2249,11 @@ def _cmd_mesh_sync(args: argparse.Namespace) -> int: "peer": args.peer, "peer_member_id": peer_info.get("member_id"), "peer_epoch": peer_info.get("current_epoch"), + "kindergarten_seconds": args.kindergarten_seconds, "announced_roots": len(sent_roots), "announced_falsifications": len(sent_falsifications), + "kindergarten_held_roots": fresh_roots_held, + "kindergarten_held_falsifications": fresh_falsifications_held, "errors": len(errors), "sent_roots": sent_roots[: args.verbose], "sent_falsifications": sent_falsifications[: args.verbose], @@ -2675,7 +2718,8 @@ def build_parser() -> argparse.ArgumentParser: help=( "sidecar diagnostic for a providence_cache record — pulls " "source chunks and classifies each unverified span " - "(paraphrase / trailing_artifact / no_overlap). Read-only, " + "(paraphrase / trailing_artifact / interior_elision / " + "no_overlap). Read-only, " "no audit events, no v9.8 field changes." ), ) @@ -3002,6 +3046,17 @@ def build_parser() -> argparse.ArgumentParser: action="store_true", help="skip ANNOUNCE_FALSIFICATION broadcast (only push roots)", ) + mesh_sync.add_argument( + "--kindergarten-seconds", + dest="kindergarten_seconds", + type=int, + default=3600, + help=( + "hold records younger than this many seconds back from the " + "broadcast (default: 3600 = 1 hour). Gives operators time to " + "burn or falsify before peers see it. 0 = broadcast everything." + ), + ) mesh_sync.set_defaults(func=_cmd_mesh_sync) mesh_pull = mesh_sub.add_parser( diff --git a/tests/test_mesh_cli.py b/tests/test_mesh_cli.py index aaee742..67d8dd2 100644 --- a/tests/test_mesh_cli.py +++ b/tests/test_mesh_cli.py @@ -136,6 +136,7 @@ def test_sync_announces_local_roots(tmp_path, capsys): "--db", str(bob_db), "mesh", "sync", "--peer", srv.url, "--limit", "100", + "--kindergarten-seconds", "0", # immediate broadcast for the test ]) rc = args.func(args) assert rc == 0 @@ -198,6 +199,7 @@ def test_sync_announces_falsifications(tmp_path, capsys): "--peer", srv.url, "--limit", "10", "--no-roots", + "--kindergarten-seconds", "0", ]) rc = args.func(args) assert rc == 0 @@ -259,6 +261,7 @@ def test_sync_no_falsifications_flag_skips_them(tmp_path, capsys): "--peer", srv.url, "--limit", "10", "--no-falsifications", + "--kindergarten-seconds", "0", ]) rc = args.func(args) assert rc == 0 @@ -311,3 +314,81 @@ def test_serve_responds_on_info_when_started_in_thread(tmp_path): srv.stop() assert info["member_id"] == "alice" assert info["group_name"] == "t" + + +def test_sync_kindergarten_holds_fresh_records(tmp_path, capsys): + """Fox 2026-04-29: gossip should default-delay so an operator can + burn or falsify within a window before peers see fresh records. + A document ingested 30 seconds ago must NOT broadcast under the + default 1-hour kindergarten window.""" + import time as _time + from aborist.store import transaction + + alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path) + _ingest(bob_db, uri="https://example.com/old", content="Old prose.", title="Old") + _ingest(bob_db, uri="https://example.com/fresh", content="Fresh prose.", title="Fresh") + + # Backdate the "old" doc so it's >>> kindergarten window. Leave the + # "fresh" one at its just-set ingest_ts (within the last second). + long_ago = int(_time.time()) - 7200 # 2 hours ago + b_conn = connect(bob_db) + try: + with transaction(b_conn): + b_conn.execute( + "UPDATE documents SET ingest_ts = ? " + "WHERE document_uri = 'https://example.com/old'", + (long_ago,), + ) + finally: + b_conn.close() + + srv = MeshWireServer(alice_db, host="127.0.0.1", port=0) + srv.start_in_thread() + try: + parser = build_parser() + args = parser.parse_args([ + "--db", str(bob_db), "mesh", "sync", + "--peer", srv.url, + "--limit", "10", + # Default kindergarten = 3600s. Fresh doc held back; old doc + # broadcast. + ]) + rc = args.func(args) + assert rc == 0 + finally: + srv.stop() + + payload = json.loads(capsys.readouterr().out) + assert payload["kindergarten_seconds"] == 3600 + assert payload["announced_roots"] == 1 # only "old" broadcast + assert payload["kindergarten_held_roots"] >= 1 # "fresh" held + # Confirm exactly which root was sent. + sent_uris = [r["ack"]["wire_type"] for r in payload["sent_roots"]] + assert sent_uris == ["ANNOUNCE_ROOT"] + + +def test_sync_kindergarten_zero_broadcasts_everything(tmp_path, capsys): + """--kindergarten-seconds 0 is the explicit opt-out: cron-friendly + immediate broadcast. Verifies the existing "no kindergarten" tests + in this file aren't accidentally protected by the default window.""" + alice_db, bob_db, _alice, _bob = _bootstrap_two_peers(tmp_path) + _ingest(bob_db, uri="https://example.com/x", content="Just-ingested.", title="X") + + srv = MeshWireServer(alice_db, host="127.0.0.1", port=0) + srv.start_in_thread() + try: + parser = build_parser() + args = parser.parse_args([ + "--db", str(bob_db), "mesh", "sync", + "--peer", srv.url, + "--limit", "10", + "--kindergarten-seconds", "0", + ]) + rc = args.func(args) + assert rc == 0 + finally: + srv.stop() + + payload = json.loads(capsys.readouterr().out) + assert payload["announced_roots"] == 1 + assert payload["kindergarten_held_roots"] == 0