diff options
Diffstat (limited to '.ai/scripts')
| -rwxr-xr-x | .ai/scripts/cross-project-broadcast.py | 155 |
1 files changed, 155 insertions, 0 deletions
diff --git a/.ai/scripts/cross-project-broadcast.py b/.ai/scripts/cross-project-broadcast.py new file mode 100755 index 0000000..2c4c690 --- /dev/null +++ b/.ai/scripts/cross-project-broadcast.py @@ -0,0 +1,155 @@ +#!/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()) |
