aboutsummaryrefslogtreecommitdiff
path: root/.ai/scripts/cross-project-broadcast.py
diff options
context:
space:
mode:
authorCraig Jennings <c@cjennings.net>2026-06-09 17:16:08 -0500
committerCraig Jennings <c@cjennings.net>2026-06-09 17:16:08 -0500
commitc91bd0b1e8183814f248b0751d88a8e422a905e8 (patch)
treef70abe516c47162f8e6538c2d9fa639aa9349d13 /.ai/scripts/cross-project-broadcast.py
parent1f0900281b8262539137bc1aff3f01cc05745139 (diff)
downloadrulesets-c91bd0b1e8183814f248b0751d88a8e422a905e8.tar.gz
rulesets-c91bd0b1e8183814f248b0751d88a8e422a905e8.zip
feat(workflows): generalize broadcast into announcement + situational modes
cross-project-broadcast handled tooling and rule announcements but had no shape for the situational case: a life or work event I want every project's agent to know, said once so none is missing context when I next talk to them. I renamed it to broadcast (helper and test alongside) and split it into two modes over the same fan-out plumbing. Announcement keeps the rigid capability template. Situational carries a general-not-comprehensive summary plus a fixed receiving-agent contract: record it in notes.org, hold it time-boxed or standing, apply on the project's own judgment, ask follow-ups at startup. The broadcasting agent does no per-project relevance analysis. Each receiving agent decides what the event means for its own work.
Diffstat (limited to '.ai/scripts/cross-project-broadcast.py')
-rwxr-xr-x.ai/scripts/cross-project-broadcast.py155
1 files changed, 0 insertions, 155 deletions
diff --git a/.ai/scripts/cross-project-broadcast.py b/.ai/scripts/cross-project-broadcast.py
deleted file mode 100755
index 2c4c690..0000000
--- a/.ai/scripts/cross-project-broadcast.py
+++ /dev/null
@@ -1,155 +0,0 @@
-#!/usr/bin/env python3
-"""Fan out a message file to every AI project's inbox/.
-
-Discovers AI projects by fingerprint — any directory under SEARCH_ROOTS
-whose .ai/protocols.org exists. Uses the existing inbox-send.py helper to
-deliver per-target.
-
-Usage:
- cross-project-broadcast.py --list
- cross-project-broadcast.py --file <path> [--exclude <name> ...] [--dry-run]
-"""
-from __future__ import annotations
-
-import argparse
-import subprocess
-import sys
-from pathlib import Path
-
-SEARCH_ROOTS = [
- Path.home() / "code",
- Path.home() / "projects",
- Path.home() / ".emacs.d",
-]
-
-
-def is_broadcastable(path: Path) -> bool:
- """A project is broadcastable if it has both .ai/ and a top-level inbox/.
-
- Matches inbox-send.py's fingerprint so the broadcast only targets
- projects that can actually receive (inbox-send rejects targets without
- inbox/). A project that has .ai/protocols.org but no inbox/ is an AI
- project that hasn't been bootstrapped for inbox messaging yet.
- """
- return (path / ".ai" / "protocols.org").is_file() and (path / "inbox").is_dir()
-
-
-def discover() -> list[Path]:
- """Return every broadcastable AI project, deduplicated and sorted."""
- seen: dict[str, Path] = {}
- for root in SEARCH_ROOTS:
- if not root.is_dir():
- continue
- # The root itself may be a project (~/.emacs.d).
- if is_broadcastable(root):
- seen.setdefault(root.name, root)
- continue
- # Otherwise scan one level down.
- for sub in sorted(root.iterdir()):
- if sub.is_dir() and is_broadcastable(sub):
- seen.setdefault(sub.name, sub)
- return [seen[name] for name in sorted(seen)]
-
-
-def sender_project() -> str | None:
- """Return the AI-project basename of the current working dir, if any."""
- cwd = Path.cwd()
- for ancestor in [cwd, *cwd.parents]:
- if (ancestor / ".ai" / "protocols.org").is_file():
- return ancestor.name
- return None
-
-
-def inbox_send_path() -> Path:
- """Locate the inbox-send.py helper in the current project."""
- cwd = Path.cwd()
- for ancestor in [cwd, *cwd.parents]:
- candidate = ancestor / ".ai" / "scripts" / "inbox-send.py"
- if candidate.is_file():
- return candidate
- raise SystemExit("cross-project-broadcast: inbox-send.py not found in current project")
-
-
-def main() -> int:
- parser = argparse.ArgumentParser(
- description="Broadcast a message file to every AI project's inbox/.",
- )
- parser.add_argument(
- "--list", action="store_true",
- help="List discovered AI projects and exit (sender-excluded).",
- )
- parser.add_argument(
- "--file",
- help="Path to the broadcast message file.",
- )
- parser.add_argument(
- "--exclude", action="append", default=[],
- help="Project basename to skip. Repeatable.",
- )
- parser.add_argument(
- "--dry-run", action="store_true",
- help="Show what would be sent without invoking inbox-send.",
- )
- args = parser.parse_args()
-
- projects = discover()
- sender = sender_project()
- excluded = set(args.exclude)
- if sender:
- excluded.add(sender)
-
- targets = [p for p in projects if p.name not in excluded]
-
- if args.list:
- print(f"Discovered {len(projects)} AI projects "
- f"(sender '{sender or '?'}' excluded, "
- f"{len(args.exclude)} explicit excludes):")
- for p in projects:
- mark = " -" if p.name in excluded else " +"
- print(f"{mark} {p.name:30s} {p}")
- print(f"\nWould broadcast to {len(targets)} target(s).")
- return 0
-
- if not args.file:
- parser.error("--file is required unless --list is given")
-
- msg_path = Path(args.file).resolve()
- if not msg_path.is_file():
- print(f"cross-project-broadcast: file not found: {msg_path}", file=sys.stderr)
- return 2
-
- inbox_send = inbox_send_path()
-
- print(f"Broadcasting {msg_path.name} to {len(targets)} project(s):")
- if args.dry_run:
- for target in targets:
- print(f" dry {target.name}")
- return 0
-
- sent = 0
- failed = []
- for target in targets:
- cmd = ["python3", str(inbox_send), target.name, "--file", str(msg_path)]
- res = subprocess.run(cmd, capture_output=True, text=True)
- if res.returncode == 0:
- print(f" ok {target.name}")
- sent += 1
- else:
- err = (res.stderr or res.stdout).strip().splitlines()[-1][:120]
- print(f" FAIL {target.name}: {err}")
- failed.append((target.name, err))
-
- print(f"\nSummary: {sent} sent, {len(failed)} failed, "
- f"{len(projects) - len(targets)} excluded.")
-
- if failed:
- print("\nFailures (re-run --file targeting these individually if needed):")
- for name, err in failed:
- print(f" {name}: {err}")
- return 1
-
- return 0
-
-
-if __name__ == "__main__":
- sys.exit(main())