From c91bd0b1e8183814f248b0751d88a8e422a905e8 Mon Sep 17 00:00:00 2001 From: Craig Jennings Date: Tue, 9 Jun 2026 17:16:08 -0500 Subject: 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. --- .ai/scripts/broadcast.py | 155 ++++++++++++++++++++++ .ai/scripts/cross-project-broadcast.py | 155 ---------------------- .ai/scripts/tests/test_broadcast.py | 116 ++++++++++++++++ .ai/scripts/tests/test_cross_project_broadcast.py | 116 ---------------- 4 files changed, 271 insertions(+), 271 deletions(-) create mode 100755 .ai/scripts/broadcast.py delete mode 100755 .ai/scripts/cross-project-broadcast.py create mode 100644 .ai/scripts/tests/test_broadcast.py delete mode 100644 .ai/scripts/tests/test_cross_project_broadcast.py (limited to '.ai/scripts') diff --git a/.ai/scripts/broadcast.py b/.ai/scripts/broadcast.py new file mode 100755 index 0000000..ba5c786 --- /dev/null +++ b/.ai/scripts/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: + broadcast.py --list + broadcast.py --file [--exclude ...] [--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("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"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()) 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 [--exclude ...] [--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()) diff --git a/.ai/scripts/tests/test_broadcast.py b/.ai/scripts/tests/test_broadcast.py new file mode 100644 index 0000000..a0decf5 --- /dev/null +++ b/.ai/scripts/tests/test_broadcast.py @@ -0,0 +1,116 @@ +"""Tests for broadcast.py: project fingerprinting + discovery. + +Plain python3 script. The pure-ish helpers are driven against tmp project +trees; discovery is exercised with SEARCH_ROOTS monkeypatched to the tree, and +the cwd-based helpers with monkeypatch.chdir. +""" +from __future__ import annotations + +import importlib.util +from pathlib import Path + +import pytest + +SCRIPT = Path(__file__).resolve().parents[1] / "broadcast.py" + + +@pytest.fixture(scope="module") +def bcast(): + spec = importlib.util.spec_from_file_location("broadcast", SCRIPT) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _make_project(root: Path, name: str, with_inbox: bool = True, + with_protocols: bool = True) -> Path: + p = root / name + (p / ".ai").mkdir(parents=True) + if with_protocols: + (p / ".ai" / "protocols.org").write_text("#+TITLE: protocols\n") + if with_inbox: + (p / "inbox").mkdir() + return p + + +# --- is_broadcastable --- + +def test_is_broadcastable_true_with_protocols_and_inbox(bcast, tmp_path): + assert bcast.is_broadcastable(_make_project(tmp_path, "proj")) is True + + +def test_is_broadcastable_false_without_inbox(bcast, tmp_path): + p = _make_project(tmp_path, "proj", with_inbox=False) + assert bcast.is_broadcastable(p) is False + + +def test_is_broadcastable_false_without_protocols(bcast, tmp_path): + p = _make_project(tmp_path, "proj", with_protocols=False) + assert bcast.is_broadcastable(p) is False + + +def test_is_broadcastable_false_on_plain_dir(bcast, tmp_path): + assert bcast.is_broadcastable(tmp_path) is False + + +# --- discover (SEARCH_ROOTS monkeypatched onto the tmp tree) --- + +def test_discover_finds_broadcastable_subprojects(bcast, tmp_path, monkeypatch): + root = tmp_path / "code" + root.mkdir() + _make_project(root, "alpha") + _make_project(root, "beta") + _make_project(root, "no-inbox", with_inbox=False) # not broadcastable + monkeypatch.setattr(bcast, "SEARCH_ROOTS", [root]) + assert [p.name for p in bcast.discover()] == ["alpha", "beta"] + + +def test_discover_handles_root_that_is_itself_a_project(bcast, tmp_path, monkeypatch): + root = _make_project(tmp_path, ".emacs.d") + monkeypatch.setattr(bcast, "SEARCH_ROOTS", [root]) + assert [p.name for p in bcast.discover()] == [".emacs.d"] + + +def test_discover_dedups_by_basename_across_roots(bcast, tmp_path, monkeypatch): + root1 = tmp_path / "code" + root1.mkdir() + root2 = tmp_path / "projects" + root2.mkdir() + _make_project(root1, "dup") + _make_project(root2, "dup") + monkeypatch.setattr(bcast, "SEARCH_ROOTS", [root1, root2]) + assert [p.name for p in bcast.discover()] == ["dup"] + + +def test_discover_skips_missing_roots(bcast, tmp_path, monkeypatch): + monkeypatch.setattr(bcast, "SEARCH_ROOTS", [tmp_path / "does-not-exist"]) + assert bcast.discover() == [] + + +# --- sender_project / inbox_send_path (cwd-based) --- + +def test_sender_project_returns_basename_inside_an_ai_project(bcast, tmp_path, monkeypatch): + p = _make_project(tmp_path, "myproj") + monkeypatch.chdir(p) + assert bcast.sender_project() == "myproj" + + +def test_sender_project_none_outside_an_ai_project(bcast, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + assert bcast.sender_project() is None + + +def test_inbox_send_path_found_in_project(bcast, tmp_path, monkeypatch): + p = _make_project(tmp_path, "myproj") + (p / ".ai" / "scripts").mkdir() + helper = p / ".ai" / "scripts" / "inbox-send.py" + helper.write_text("# stub\n") + monkeypatch.chdir(p) + assert bcast.inbox_send_path() == helper + + +def test_inbox_send_path_raises_when_missing(bcast, tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + with pytest.raises(SystemExit): + bcast.inbox_send_path() diff --git a/.ai/scripts/tests/test_cross_project_broadcast.py b/.ai/scripts/tests/test_cross_project_broadcast.py deleted file mode 100644 index 5919fbf..0000000 --- a/.ai/scripts/tests/test_cross_project_broadcast.py +++ /dev/null @@ -1,116 +0,0 @@ -"""Tests for cross-project-broadcast.py: project fingerprinting + discovery. - -Plain python3 script. The pure-ish helpers are driven against tmp project -trees; discovery is exercised with SEARCH_ROOTS monkeypatched to the tree, and -the cwd-based helpers with monkeypatch.chdir. -""" -from __future__ import annotations - -import importlib.util -from pathlib import Path - -import pytest - -SCRIPT = Path(__file__).resolve().parents[1] / "cross-project-broadcast.py" - - -@pytest.fixture(scope="module") -def bcast(): - spec = importlib.util.spec_from_file_location("cross_project_broadcast", SCRIPT) - assert spec and spec.loader - module = importlib.util.module_from_spec(spec) - spec.loader.exec_module(module) - return module - - -def _make_project(root: Path, name: str, with_inbox: bool = True, - with_protocols: bool = True) -> Path: - p = root / name - (p / ".ai").mkdir(parents=True) - if with_protocols: - (p / ".ai" / "protocols.org").write_text("#+TITLE: protocols\n") - if with_inbox: - (p / "inbox").mkdir() - return p - - -# --- is_broadcastable --- - -def test_is_broadcastable_true_with_protocols_and_inbox(bcast, tmp_path): - assert bcast.is_broadcastable(_make_project(tmp_path, "proj")) is True - - -def test_is_broadcastable_false_without_inbox(bcast, tmp_path): - p = _make_project(tmp_path, "proj", with_inbox=False) - assert bcast.is_broadcastable(p) is False - - -def test_is_broadcastable_false_without_protocols(bcast, tmp_path): - p = _make_project(tmp_path, "proj", with_protocols=False) - assert bcast.is_broadcastable(p) is False - - -def test_is_broadcastable_false_on_plain_dir(bcast, tmp_path): - assert bcast.is_broadcastable(tmp_path) is False - - -# --- discover (SEARCH_ROOTS monkeypatched onto the tmp tree) --- - -def test_discover_finds_broadcastable_subprojects(bcast, tmp_path, monkeypatch): - root = tmp_path / "code" - root.mkdir() - _make_project(root, "alpha") - _make_project(root, "beta") - _make_project(root, "no-inbox", with_inbox=False) # not broadcastable - monkeypatch.setattr(bcast, "SEARCH_ROOTS", [root]) - assert [p.name for p in bcast.discover()] == ["alpha", "beta"] - - -def test_discover_handles_root_that_is_itself_a_project(bcast, tmp_path, monkeypatch): - root = _make_project(tmp_path, ".emacs.d") - monkeypatch.setattr(bcast, "SEARCH_ROOTS", [root]) - assert [p.name for p in bcast.discover()] == [".emacs.d"] - - -def test_discover_dedups_by_basename_across_roots(bcast, tmp_path, monkeypatch): - root1 = tmp_path / "code" - root1.mkdir() - root2 = tmp_path / "projects" - root2.mkdir() - _make_project(root1, "dup") - _make_project(root2, "dup") - monkeypatch.setattr(bcast, "SEARCH_ROOTS", [root1, root2]) - assert [p.name for p in bcast.discover()] == ["dup"] - - -def test_discover_skips_missing_roots(bcast, tmp_path, monkeypatch): - monkeypatch.setattr(bcast, "SEARCH_ROOTS", [tmp_path / "does-not-exist"]) - assert bcast.discover() == [] - - -# --- sender_project / inbox_send_path (cwd-based) --- - -def test_sender_project_returns_basename_inside_an_ai_project(bcast, tmp_path, monkeypatch): - p = _make_project(tmp_path, "myproj") - monkeypatch.chdir(p) - assert bcast.sender_project() == "myproj" - - -def test_sender_project_none_outside_an_ai_project(bcast, tmp_path, monkeypatch): - monkeypatch.chdir(tmp_path) - assert bcast.sender_project() is None - - -def test_inbox_send_path_found_in_project(bcast, tmp_path, monkeypatch): - p = _make_project(tmp_path, "myproj") - (p / ".ai" / "scripts").mkdir() - helper = p / ".ai" / "scripts" / "inbox-send.py" - helper.write_text("# stub\n") - monkeypatch.chdir(p) - assert bcast.inbox_send_path() == helper - - -def test_inbox_send_path_raises_when_missing(bcast, tmp_path, monkeypatch): - monkeypatch.chdir(tmp_path) - with pytest.raises(SystemExit): - bcast.inbox_send_path() -- cgit v1.2.3