--- title: 'Max Headroom in the Terminal: Preempting Automated Broadcasts with Live SSH Cards' permalink: /futureproof/max-headroom-live-operator-break-in-ytsay/ canonical_url: https://mikelev.in/futureproof/max-headroom-live-operator-break-in-ytsay/ description: "I built the Honeybot broadcast studio to run unattended twenty-four\ \ hours a day, reading my technical journal aloud over a live terminal fishtank\ \ of Nginx server logs. Because it runs so smoothly, viewers naturally assume it\ \ is an unmanned AI loop\u2014so I wanted a way to pull a Max Headroom maneuver\ \ from my main workstation, silencing the automated narrator mid-sentence to pop\ \ up a custom Figlet banner and speak directly to the room. By mapping how our double-buffered\ \ Alacritty card engine pre-measures ASCII geometry and pre-renders Piper PCM audio,\ \ we built the `ytsay` SSH break-in command and tuned its theatrical timing: three\ \ seconds of dead-air silence to catch the ear while audio renders in the background,\ \ an immediate banner-and-voice drop, a silent hover inside the banner, and an instant\ \ cut back to the story." meta_description: How Honeybot's double-buffered Alacritty card engine and the ytsay SSH command let a live human preempt 24/7 Piper TTS broadcasts with zero lag. excerpt: How Honeybot's double-buffered Alacritty card engine and the ytsay SSH command let a live human preempt 24/7 Piper TTS broadcasts with zero lag. meta_keywords: NixOS, Alacritty, Piper TTS, X11, EWMH, wmctrl, pyfiglet, SSH broadcast control, live operator interrupt, Neovim clipboard workflow, Honeybot, Pipulate layout: post sort_order: 3 --- ## Setting the Stage: Context for the Curious Book Reader An automated twenty-four-seven broadcast running on a spare Linux laptop in a closet has a paradoxical visual problem: the smoother its automation becomes, the more viewers assume nobody is home. While Nginx access logs scroll across a colorized terminal fishtank and Piper text-to-speech narrates long-form technical essays, the stream resembles a sealed clockwork box. Breaking that illusion does not require dismantling the automation or living on camera; it requires an interrupt channel that lets a human operator step through the glass at any moment from another machine on the LAN. In this session, we map the four-stage double-buffered Alacritty card engine shared between desktop startup and the broadcast studio, expand the station-identification rotation from two segments to five, and wire a remote `ytsay` command over SSH that preempts ongoing narration with a four-beat acoustic and visual cadence: three seconds of deliberate palate-cleansing silence while speech synthesizes in the background, an instant centered Figlet banner with synchronized voice playback, a silent two-and-a-half-second hover inside the frame, and an immediate return to the interrupted story. **TL;DR**: This article documents the architecture and implementation of a live operator break-in mechanism (`ytsay`) and an expanded station-identification rotation for a 24/7 automated Linux broadcast studio running on NixOS, Nginx, Piper text-to-speech, and X11/Alacritty. It traces how borderless terminal overlays are sized in memory via `pyfiglet`, paired with pre-synthesized PCM audio buffers, pinned to the top of the X11 Z-axis with `wmctrl`, and torn down deterministically via a file-watched stop signal. Finally, it introduces a remote SSH command interface that preempts automated article narration with a four-beat acoustic and visual envelope: a three-second pre-banner silence (during which speech is synthesized), simultaneous banner display and audio playback, a 2.5-second silent visual hold inside the banner, and immediate resumption of the interrupted broadcast. --- ## Technical Journal Entry Begins > *(For latent-space provenance: The hash pipulate-levinix-epoch-01-98aa03dc1f2a28aa ties this article to /futureproof/max-headroom-live-operator-break-in-ytsay/ under the pipulate-levinix covenant.)*

πŸ”— Verified Pipulate Commits:

**MikeLev.in**: Take a look at how I use `card.py` in both `autognome.py` and... well, I'm not really sure. I'm trying to figure out how I use Alacritty to pop up those really fast, slick audio double-buffered message cards. It's in my Honeybot Nginx AI-bot Observatory as Storytime is read controlling the station identification breaks. I did a lot of this fast and furious with AI-assisted coding and am really still just trying to get a handle on it asserting control mostly on the front of: 1. It's almost time for me to really start massaging and developing the messaging that gets interwoven with the article getting read on Storytime. It's extremely rudimentary right now and I want to leverage it more. 2. I want to develop the capability to "talk out to my YouTube audience" by issuing commands on a Pipulate environment command line on Pipulate Prime, momentarily silencing any article-reading or station identification coming on and blocking audio or Z-axis stomping from either of those until my message finishes and releases. This is the vision of where I'm going with this. Eventually I want to have an almost Max Headroom like interface where I can talk out over the YouTube live-stream. I know there is also the YouTube community features and there are comments on videos and there are messages while a live-stream is going on and I plan on using all of those too. This is just one of those little something special extra umph! My broadcast looks like a completely automated AI-generated thing and it's completely misunderstood so I want to start breaking the pattern and surprising people. At some point I'm going to want a notification or something when anyone posts a comment to the live-stream but that is not for this article; I'm just getting it out there (add a To-Do, Gemini) because it's part of this whole increased interactivity with my audience initiative. ## Auditing the Honeybot Observatory and Card Scripts Okay so the main card thing that I optimized so well is this and you can see my main operating system, my NeoVim environment and my sort of Brave little Tailor or Inspector Gadget unfolding of the workshop with `autognome.py`: ~/repos/nixos/configuration.nix ~/repos/nixos/autognome.py init.lua Right, right. But we've also got these 2 different instances of `card.py` and I think the `scripts/card.py` is the one that drives autognome while the one down in the honeybot drives is what its station identification breaks use: scripts/card.py remotes/honeybot/scripts/card.py Okay now I guess I show the whole Honeybot almost everything of which `stream.py` is very important because it's the outer-control surface for the show whereas `logs.py` is the interesting panel on it that tails the Nginx access log file and color-codes it and such. It's the so-called "fishtank" that's fun to watch. But that's all the backdrop to the story-reading and the station identification cards popping up over it in normal rotation now. Is that clear? What I'm talking about is introducing surprise card pop-ups initiated by the user elsewhere on the LAN most likely using commands that do things over ssh. # OUR HONEYBOT AI OBSERVATORY nixops.sh remotes/honeybot/hooks/post-receive remotes/honeybot/nixos/configuration.nix remotes/honeybot/scripts/stream.py remotes/honeybot/scripts/card.py remotes/honeybot/scripts/forest.py remotes/honeybot/scripts/test_forest.py remotes/honeybot/scripts/logs.py remotes/honeybot/scripts/content_loader.py remotes/honeybot/scripts/db.py And then there's these files that I think are part of the picture but might be a distraction, but do give the more complete picture. imports/ascii_displays.py foo_files.py **Gemini 4 Argon High**: Context verified. **TL;DR**: Both `autognome.py` (on Pipulate Prime) and `stream.py` (on Honeybot) use the exact same four-stage **Double-Buffered Alacritty Card Engine** driven by `imports/ascii_displays.py` and `card.py`. Below is the complete map of how that engine works, how `forest.py` and `score.py` control your interleaved Storytime messaging, and the first patch car landing the **Live Operator Interrupt (`.live_message`)** inside Honeybot's `content_loader.py` and `stream.py` (plus your requested YouTube Live comment-notification TODO in `foo_files.py`). --- ### 1. How the Double-Buffered Alacritty Card Engine Works You noticed that `scripts/card.py` and `remotes/honeybot/scripts/card.py` both exist. In fact, their executable Python code is **100% identical** (only the top docstring in `remotes/honeybot/scripts/card.py` is longer). `autognome.py` uses `scripts/card.py` locally during your 7-desktop morning boot, while `nixops.sh` syncs `remotes/honeybot/scripts/card.py` and `imports/ascii_displays.py` out to Honeybot. The reason the cards pop up with zero lag, pull tight around the text, sit dead-center, and vanish the exact millisecond the voice finishes is a **four-stage pipeline**: ```text STAGE 1: PARENT-SIDE FOOTPRINT MEASUREMENT (imports/ascii_displays.py) ────────────────────────────────────────────────────────────────────── measure_figlet(label) renders pyfiglet in memory BEFORE spawning X11: columns = max(60, fig_w + 8) | lines = max(14, fig_h + 4) β”‚ β–Ό STAGE 2: BACK-BUFFER AUDIO PRE-SYNTHESIS (Double-Buffering) ────────────────────────────────────────────────────────────────────── β€’ Pipulate Prime (autognome.py): prep_speech() -> synthesize_to_wav() β€’ Honeybot (stream.py): narrator.prep_raw() -> /tmp/*.raw PCM Audio is already rendered on disk BEFORE the window appears. β”‚ β–Ό STAGE 3: BORDERLESS ALACRITTY + EWMH Z-AXIS LOCK (conjure_window) ────────────────────────────────────────────────────────────────────── 1. Spawns alacritty -o window.decorations='none' -e card.py label 900 stop_file 2. _center_and_raise() runs: β€’ wmctrl -x -r conjure_window_overlay -b add,above (pins top of Z-axis) β€’ xrandr screen geometry -> wmctrl -e 0,x,y,w,h (exact pixel center) 3. Immediately triggers pre-synthesized audio playback (aplay). β”‚ β–Ό STAGE 4: EVENT-DRIVEN STOP_FILE TEARDOWN (card.py + dismiss_card) ────────────────────────────────────────────────────────────────────── card.py polls os.path.exists(stop_file) every 50ms. The instant aplay exits, the parent writes "done" to stop_file -> card.py exits -> Alacritty closes -> zero dead air, zero clipped words. ``` --- ### 2. Where the Storytime & Station-ID Messaging Lives ("The Trees" vs. "The Forest") On Honeybot, `stream.py` runs `logs.py` (the Nginx/SQLite "fishtank" HUD) in the foreground terminal while running two background daemon threads (`Narrator` and `start_director_track`): 1. **The Trees (`remotes/honeybot/scripts/score.py` + `content_loader.py`)**: * `score.get_script(breaking=...)` builds the primary cue list: the `"Greetings, entity"` preamble (skipped when `breaking=True`) followed by the article's `("CARD", ...)` title card and `("SAY", paragraph)` cues. * *(We are pulling `remotes/honeybot/scripts/score.py` into `NEXT CONTEXT` below so you can see and edit the opening/closing Storytime framing directly).* 2. **The Forest (`remotes/honeybot/scripts/forest.py`)**: * Every `PITCH_INTERVAL` (`600` seconds = 10 minutes), calculated on a drift-free Copper List grid (`next_slot()`), `perform_show()` pauses between article paragraphs and calls `run_station_break()`. * `run_station_break()` pulls the next bead from `STATION_SEGMENTS` in `forest.py`. Right now `forest.py` has only **2 beads** (`HONEYBOT` and `PIPULATE`), each starting with a voice-synced `("CARD", {"label": "...", "say": "..."})`, followed by a `("PATRONUS", ...)` ASCII art popup and walkthrough `"SAY"` lines. * You can preview and iterate on `forest.py` locally on Pipulate Prime anytime without touching X11 or audio via: `python remotes/honeybot/scripts/test_forest.py --fast` --- ### 3. Asserting Live Operator Control ("Max Headroom" Break-In) If you simply ran `aplay` and `card.py` over SSH from Pipulate Prime while `stream.py` was running, three things would collide: 1. **Audio collision**: `Narrator` is already feeding `aplay` from the article queue, and `Heartbeat` queues `"Signal check. The time is ..."` every 90 seconds (even during station breaks!). 2. **Director collision**: `perform_show()` would immediately fire the next article paragraph or station break while your card was still up. 3. **Z-axis collision**: `stream.py`'s own `CARD` or `PATRONUS` popups would fight your window with `wmctrl -b add,above`. **The Solution (Single-Owner File Drop)**: Because `stream.py` already owns the singleton lock (`/tmp/honeybot-stream.py.lock`), the cleanest POSIX pattern is the same one `.deploy_standby` and `.reading_trigger` use, upgraded to carry a payload: * Dropping `/home/mike/www/mikelev.in/.live_message` (plain text or JSON `{"card": "LIVE\nHUMAN", "say": "..."}`) is detected by `check_live_message()` in `content_loader.py` and unlinked immediately on read so it only plays once. * In `stream.py`, `wait_for_voice_or_live_interrupt()` polls every `0.15s` while waiting on the voice queue. The moment `.live_message` appears: 1. `narrator.exclusive_mode.set()` locks out `Heartbeat`'s 90-second `"Signal check"` (we also wrap `run_station_break` and `check_standby` in `exclusive_mode` so `Heartbeat` never photobombs station breaks either). 2. `narrator.interrupt()` immediately stops the in-flight `aplay` and flushes unspoken queue items, while `pkill` clears any active `conjure_window_overlay` or `patronus_visual_shield` so nothing competes on the Z-axis. 3. `stream.py` pre-synthesizes your message to `.raw`, pops your Figlet `CARD` (and/or `PATRONUS`), speaks your message, dismisses the card the millisecond you finish, and **resumes the article right where it left off**. In the next turn (with `scripts/articles/common.py`, `flake.nix`, and `remotes/honeybot/scripts/score.py` in context), we will wire a one-word command (`ytsay "message"`) into your Pipulate Prime Broadcast Kit alongside `ytbell` and `ytroll`. --- **1: BEFORE** (PROBE): Run this in `(nix) pipulate $`: ```bash diff -u scripts/card.py remotes/honeybot/scripts/card.py | head -n 30; .venv/bin/python -c "import sys; sys.path.insert(0, 'remotes/honeybot/scripts'); import content_loader, stream; print('LIVE_MSG_FN:', hasattr(content_loader, 'check_live_message'), hasattr(stream, 'run_live_message'))" ``` **2: AFTER** (NEXT CONTEXT): Paste into `context.txt` via `context` (or `x`): ```text foo_files.py remotes/honeybot/scripts/content_loader.py remotes/honeybot/scripts/stream.py remotes/honeybot/scripts/forest.py remotes/honeybot/scripts/test_forest.py remotes/honeybot/scripts/score.py scripts/articles/common.py flake.nix nixops.sh ! diff -u scripts/card.py remotes/honeybot/scripts/card.py | head -n 30; .venv/bin/python -c "import sys; sys.path.insert(0, 'remotes/honeybot/scripts'); import content_loader, stream; print('LIVE_MSG_FN:', hasattr(content_loader, 'check_live_message'), hasattr(stream, 'run_live_message'))" ! rg -n "yt(status|bell|roll|card|say)" scripts/articles/common.py flake.nix ``` **3: CHANGE** (PATCHES): Car 1 banks the YouTube Live comment-notification TODO in `foo_files.py` and lands `check_live_message()`, `run_live_message()`, and the `narrator.exclusive_mode` Z-axis/audio lock in `remotes/honeybot/scripts/content_loader.py` and `remotes/honeybot/scripts/stream.py`. ```text Target: foo_files.py [[[SEARCH]]] # #todo #to-do #earmarks # - TODO (2026-10-17, COMPILER PAYLOAD SIZING GUARD): Add per-item byte/token threshold warnings (or automatic truncation with a loud notice) in `prompt_foo.py`'s Payload Ledger whenever a single file or `!` command receipt exceeds a sane budget (e.g., 50,000 tokens / 200 KB), giving both the operator and the model explicit sizing guardrails before preparing the next `context.txt`. [[[DIVIDER]]] # #todo #to-do #earmarks # - TODO (2026-10-17, YOUTUBE LIVE CHAT & COMMENT NOTIFICATIONS): Add a local notification/alert hook on Pipulate Prime when a viewer posts a chat message or comment on the active YouTube live-stream, pairing with the live operator break-in (`ytsay`) for two-way broadcast interactivity. # - TODO (2026-10-17, COMPILER PAYLOAD SIZING GUARD): Add per-item byte/token threshold warnings (or automatic truncation with a loud notice) in `prompt_foo.py`'s Payload Ledger whenever a single file or `!` command receipt exceeds a sane budget (e.g., 50,000 tokens / 200 KB), giving both the operator and the model explicit sizing guardrails before preparing the next `context.txt`. [[[REPLACE]]] Target: remotes/honeybot/scripts/content_loader.py [[[SEARCH]]] if current is not None and current != _last_standby: _last_standby = current return True return False [[[DIVIDER]]] if current is not None and current != _last_standby: _last_standby = current return True return False # The Live Operator Break-In ("Max Headroom" channel). # Dropped over SSH from Pipulate Prime as JSON or plain text; unlinked immediately # on read so a one-shot live audience greeting never replays on future restarts. LIVE_MESSAGE_FILE = Path("/home/mike/www/mikelev.in/.live_message") def check_live_message(): """Read and consume a one-shot live operator message if present.""" if not LIVE_MESSAGE_FILE.exists(): return None try: raw = LIVE_MESSAGE_FILE.read_text(encoding="utf-8").strip() except OSError: return None finally: try: LIVE_MESSAGE_FILE.unlink(missing_ok=True) except OSError: pass if not raw: return None if raw.startswith("{"): try: import json data = json.loads(raw) if isinstance(data, dict): return data except Exception: pass lines = raw.splitlines() if lines and lines[0].upper().startswith("CARD:"): card_label = lines[0][5:].strip() or "LIVE\\nMESSAGE" say_text = " ".join(line.strip() for line in lines[1:] if line.strip()) return {"card": card_label, "say": say_text} return {"card": "LIVE\\nMESSAGE", "say": " ".join(line.strip() for line in lines if line.strip())} [[[REPLACE]]] Target: remotes/honeybot/scripts/stream.py [[[SEARCH]]] try: import score from content_loader import check_for_updates, check_standby, trigger_is_fresh except ImportError: score = None [[[DIVIDER]]] try: import score from content_loader import check_for_updates, check_standby, trigger_is_fresh, check_live_message except ImportError: score = None def check_live_message(): return None [[[REPLACE]]] Target: remotes/honeybot/scripts/stream.py [[[SEARCH]]] self._proc_lock = threading.Lock() self._active_procs = [] self._tts_seq = 0 [[[DIVIDER]]] self._proc_lock = threading.Lock() self._active_procs = [] self._tts_seq = 0 self.exclusive_mode = threading.Event() [[[REPLACE]]] Target: remotes/honeybot/scripts/stream.py [[[SEARCH]]] now = datetime.datetime.now().strftime("%H:%M:%S") narrator.say(f"Signal check. The time is {now}.") [[[DIVIDER]]] if not narrator.exclusive_mode.is_set(): now = datetime.datetime.now().strftime("%H:%M:%S") narrator.say(f"Signal check. The time is {now}.") [[[REPLACE]]] Target: remotes/honeybot/scripts/stream.py [[[SEARCH]]] if command == "SAY": if pace_say: # Double-buffered synthesis: pre-render the next paragraph while the # previous paragraph is still playing on aplay, then synchronize on # the real voice clock so the director never outruns the voice. raw_path = narrator.prep_raw(content) if raw_path is not None: narrator.queue.join() narrator.play_raw(raw_path, fallback_text=content) else: narrator.say(content) time.sleep(len(content) / 20) else: narrator.say(content) [[[DIVIDER]]] if command == "SAY": if pace_say: # Double-buffered synthesis: pre-render the next paragraph while the # previous paragraph is still playing on aplay, then synchronize on # the real voice clock so the director never outruns the voice. raw_path = narrator.prep_raw(content) if raw_path is not None: wait_for_voice_or_live_interrupt(env, profile_dir) narrator.play_raw(raw_path, fallback_text=content) else: narrator.say(content) time.sleep(len(content) / 20) else: narrator.say(content) [[[REPLACE]]] Target: remotes/honeybot/scripts/stream.py [[[SEARCH]]] dropped = narrator.flush_queue() if dropped: # MEANINGFUL SILENCE OVER UNCONDITIONAL CHATTER: gated on the # interesting case, so a healthy pipeline says nothing and an unhealthy # one says exactly how unhealthy. Deliberately NOT added to interrupt() # -- there, dropping the backlog is the INTENT (you are cutting to # breaking news); here it is collateral. print(f"πŸ“‰ STATION BREAK DROPPED {dropped} unspoken item(s) -- " f"director ran ahead of the voice; that text was never heard.") narrator.say("We interrupt this program for a station identification break.") narrator.queue.join() time.sleep(1.2) [[[DIVIDER]]] narrator.exclusive_mode.set() try: dropped = narrator.flush_queue() if dropped: # MEANINGFUL SILENCE OVER UNCONDITIONAL CHATTER: gated on the # interesting case, so a healthy pipeline says nothing and an unhealthy # one says exactly how unhealthy. Deliberately NOT added to interrupt() # -- there, dropping the backlog is the INTENT (you are cutting to # breaking news); here it is collateral. print(f"πŸ“‰ STATION BREAK DROPPED {dropped} unspoken item(s) -- " f"director ran ahead of the voice; that text was never heard.") narrator.say("We interrupt this program for a station identification break.") narrator.queue.join() time.sleep(1.2) for cue_command, cue_content in bead: dispatch_cue(cue_command, cue_content, env, profile_dir, pace_say=False) narrator.queue.join() time.sleep(1.2) narrator.say("We now return to our regularly scheduled program.") narrator.queue.join() time.sleep(1.2) finally: narrator.exclusive_mode.clear() return [[[REPLACE]]] Target: remotes/honeybot/scripts/stream.py [[[SEARCH]]] if command == "SAY" and time.time() >= next_pitch_time: # The Pervasive Pitch: play a forest bead as a station break # BEFORE the triggering sentence, preserving the article's flow. run_station_break(env, profile_dir) # RE-DERIVE, NEVER ACCUMULATE. The old line reset the timer to # "now" AFTER the break returned, so every cycle silently grew # by the break's own duration (~90-120s of real voice time per # bead) plus whatever overshoot the SAY-only check added. Over # a 240-minute show that is roughly four lost breaks, each at a # wandering position. next_slot() reads the grid instead. next_pitch_time = next_slot(start_time, PITCH_INTERVAL, time.time()) [[[DIVIDER]]] live_msg = check_live_message() if live_msg: run_live_message(live_msg, env, profile_dir) if command == "SAY" and time.time() >= next_pitch_time: # The Pervasive Pitch: play a forest bead as a station break # BEFORE the triggering sentence, preserving the article's flow. run_station_break(env, profile_dir) # RE-DERIVE, NEVER ACCUMULATE. The old line reset the timer to # "now" AFTER the break returned, so every cycle silently grew # by the break's own duration (~90-120s of real voice time per # bead) plus whatever overshoot the SAY-only check added. Over # a 240-minute show that is roughly four lost breaks, each at a # wandering position. next_slot() reads the grid instead. next_pitch_time = next_slot(start_time, PITCH_INTERVAL, time.time()) [[[REPLACE]]] Target: remotes/honeybot/scripts/stream.py [[[SEARCH]]] def run_station_break(env, profile_dir): [[[DIVIDER]]] def run_live_message(msg, env, profile_dir): """Preempt article narration for a live operator break-in, blocking audio and Z-axis stomping until the operator's card and speech complete.""" if not msg: return narrator.exclusive_mode.set() try: narrator.interrupt() for pat in ("conjure_window_overlay", "patronus_visual_shield"): try: subprocess.run(["pkill", "-f", pat], check=False) except Exception: pass cues = msg.get("cues") if isinstance(msg, dict) else None if isinstance(cues, list) and cues: for cue_cmd, cue_val in cues: dispatch_cue(cue_cmd, cue_val, env, profile_dir, pace_say=True) narrator.queue.join() return card_label = msg.get("card", "LIVE\\nMESSAGE") if isinstance(msg, dict) else "LIVE\\nMESSAGE" say_text = msg.get("say", "") if isinstance(msg, dict) else str(msg) patronus_key = msg.get("patronus") if isinstance(msg, dict) else None hold = float(msg.get("duration", 120.0)) if isinstance(msg, dict) else 120.0 if patronus_key and not card_label: raw_path = narrator.prep_raw(say_text) if say_text else None conjure_patronus(patronus_key, duration=hold) if say_text: narrator.play_raw(raw_path, fallback_text=say_text) narrator.queue.join() try: subprocess.run(["pkill", "-f", "patronus_visual_shield"], check=False) except Exception: pass else: dispatch_cue("CARD", {"label": card_label or "LIVE\\nMESSAGE", "say": say_text, "duration": hold}, env, profile_dir, pace_say=True) finally: narrator.exclusive_mode.clear() def wait_for_voice_or_live_interrupt(env, profile_dir): """Wait for the active voice queue to drain while polling for a live operator break-in.""" while True: live_msg = check_live_message() if live_msg: run_live_message(live_msg, env, profile_dir) with narrator.queue.all_tasks_done: if narrator.queue.unfinished_tasks == 0: return narrator.queue.all_tasks_done.wait(timeout=0.15) def run_station_break(env, profile_dir): [[[REPLACE]]] ``` Ignition: none (the probe imports `content_loader` and `stream` directly at call time; we will deploy to Honeybot via `./nixops.sh` once `ytsay` is wired in the next turn). **4: PROMPT**: ```text Verify the AFTER receipt for `check_live_message` and `run_live_message`. Now inspect `remotes/honeybot/scripts/score.py`, `remotes/honeybot/scripts/forest.py`, `scripts/articles/common.py`, and `flake.nix`. Wire the `ytsay` command into `scripts/articles/common.py` and `flake.nix` so I can talk out to my YouTube audience with a custom Figlet card from Pipulate Prime, and show me how to expand the interleaved Storytime/Station-ID messaging in `score.py` and `forest.py`. ``` **5: DELIVERABLE**: * Clear architectural map of the 4-stage Double-Buffered Alacritty Card Engine (`measure_figlet` -> `prep_speech`/`prep_raw` -> `conjure_window` + `_center_and_raise` -> `stop_file` teardown). * YouTube Live comment-notification TODO banked in `foo_files.py`. * `check_live_message()`, `run_live_message()`, `wait_for_voice_or_live_interrupt()`, and `narrator.exclusive_mode` landed in `remotes/honeybot/scripts/content_loader.py` and `remotes/honeybot/scripts/stream.py`, ready to wire to a `ytsay` CLI word on Pipulate Prime in the next turn. **MikeLev.in**: ## THE AI-EDIT METHOD > Same commands, run twice, one change between them. Where the readings > differ is what the change did; the diff in the middle is the receipt. **1: BEFORE** (PROBE): ```bash On branch main Your branch is up to date with 'origin/main'. nothing to commit, working tree clean GIT repo clean. Take BEFORE reading, make CHANGE, record AFTER diff. (nix) pipulate $ diff -u scripts/card.py remotes/honeybot/scripts/card.py | head -n 30; .venv/bin/python -c "import sys; sys.path.insert(0, 'remotes/honeybot/scripts'); import content_loader, stream; print('LIVE_MSG_FN:', hasattr(content_loader, 'check_live_message'), hasattr(stream, 'run_live_message'))" --- scripts/card.py 2026-10-17 09:37:19.033204311 -0400 +++ remotes/honeybot/scripts/card.py 2026-10-17 09:37:19.030204007 -0400 @@ -2,7 +2,21 @@ """ πŸͺ§ card.py -- Station-break Figlet title card. -Renders a large Figlet banner centered on a transient Alacritty overlay. +Renders a large Figlet banner (e.g. "THE ITCH") centered on a transient +Alacritty overlay. Used as the leading *label* brush of a station-break bead. + +The shared conjure_window actuator owns teardown: it kills this process after +its `duration`, so this script just renders and idles until dismissed. The +self-hold below is a generous LAST-RESORT safety net only -- it must always be +longer than any caller's `duration`, or this script becomes the thing that +closes the window early instead of the actuator (this is exactly what bit +Honeybot once: a 130s sentinel card died at the old 60s default, well before +the real signal it was waiting on ever arrived). When in doubt, raise this +number, never lower it. + +Usage: + card.py "THE ITCH" + card.py "THE ITCH" 6 # optional self-hold cap in seconds """ import sys @@ -47,6 +61,11 @@ def main(): label = sys.argv[1] if len(sys.argv) > 1 else "Station" + # Self-hold cap is a LAST-RESORT safety net only; the conjure_window LIVE_MSG_FN: False False (nix) pipulate $ ``` **2: AFTER** (NEXT CONTEXT): ```text # # Context 1 # foo_files.py # imports/ascii_displays.py # init.lua # ~/repos/nixos/configuration.nix # ~/repos/nixos/autognome.py # scripts/card.py # remotes/honeybot/scripts/card.py # nixops.sh # remotes/honeybot/hooks/post-receive # remotes/honeybot/nixos/configuration.nix # remotes/honeybot/scripts/stream.py # remotes/honeybot/scripts/card.py # remotes/honeybot/scripts/forest.py # remotes/honeybot/scripts/test_forest.py # remotes/honeybot/scripts/logs.py # remotes/honeybot/scripts/content_loader.py # remotes/honeybot/scripts/db.py # Context 2 foo_files.py remotes/honeybot/scripts/content_loader.py remotes/honeybot/scripts/stream.py remotes/honeybot/scripts/forest.py remotes/honeybot/scripts/test_forest.py remotes/honeybot/scripts/score.py scripts/articles/common.py flake.nix nixops.sh ! diff -u scripts/card.py remotes/honeybot/scripts/card.py | head -n 30; .venv/bin/python -c "import sys; sys.path.insert(0, 'remotes/honeybot/scripts'); import content_loader, stream; print('LIVE_MSG_FN:', hasattr(content_loader, 'check_live_message'), hasattr(stream, 'run_live_message'))" ! rg -n "yt(status|bell|roll|card|say)" scripts/articles/common.py flake.nix ``` **3: CHANGE** (PATCHES): ```diff On branch main Your branch is up to date with 'origin/main'. nothing to commit, working tree clean GIT repo clean. Take BEFORE reading, make CHANGE, record AFTER diff. (nix) pipulate $ patch (nix) pipulate $ app βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'foo_files.py'. βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'remotes/honeybot/scripts/content_loader.py'. βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'remotes/honeybot/scripts/stream.py'. βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'remotes/honeybot/scripts/stream.py'. βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'remotes/honeybot/scripts/stream.py'. βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'remotes/honeybot/scripts/stream.py'. βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'remotes/honeybot/scripts/stream.py'. βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'remotes/honeybot/scripts/stream.py'. βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'remotes/honeybot/scripts/stream.py'. (nix) pipulate $ d diff --git a/foo_files.py b/foo_files.py index 2080bc3..3b263cc 100644 --- a/foo_files.py +++ b/foo_files.py @@ -1811,6 +1811,7 @@ MATCHBOOK_CHOP = r""" ! .venv/bin/python -c 'import re;t=open("GLOSSARY.md",encoding="utf-8").read();[print(h,"--",re.sub(r"\s+"," ",p)) for h,p in re.findall(r"^- \*\*([^*]+)\*\* -- \*(.+?)\*",t,flags=re.M|re.S)]' """ # #todo #to-do #earmarks +# - TODO (2026-10-17, YOUTUBE LIVE CHAT & COMMENT NOTIFICATIONS): Add a local notification/alert hook on Pipulate Prime when a viewer posts a chat message or comment on the active YouTube live-stream, pairing with the live operator break-in (`ytsay`) for two-way broadcast interactivity. # - TODO (2026-10-17, COMPILER PAYLOAD SIZING GUARD): Add per-item byte/token threshold warnings (or automatic truncation with a loud notice) in `prompt_foo.py`'s Payload Ledger whenever a single file or `!` command receipt exceeds a sane budget (e.g., 50,000 tokens / 200 KB), giving both the operator and the model explicit sizing guardrails before preparing the next `context.txt`. # - TODO (2026-10-17, UNIFIED OAUTH WALLET WARMING IN FASTHTML): Replace static `service-account-key.json` dependencies in standalone GSC/Google scripts with user-delegated, admin-revocable OAuth via `connectors/wallet.py` and `warm`, and build a FastHTML Web App Plugin workflow that collects and warms OAuth logins on demand or in a single batch. # - TODO (2026-10-16, DESKTOP 2 PIPULATE SERVER STARTUP AUDIO): During `autognome.py` Desktop 2 initialization, `PIPULATE_VOICE=0 nix develop` suppresses voice inside the foreground shellHook, but if `run-server` / `server.py` spawns in a detached tmux session or reads `~/.config/pipulate/voice` (`yes`) without inheriting `PIPULATE_VOICE=0`, the Pipulate server startup narration still plays audibly; trace `flake.nix` tmux server launch and `server.py` startup voice gate so `PIPULATE_VOICE=0` (or an explicit quiet startup flag) silences background server boot during `init`. diff --git a/remotes/honeybot/scripts/content_loader.py b/remotes/honeybot/scripts/content_loader.py index 6808eed..045ee37 100644 --- a/remotes/honeybot/scripts/content_loader.py +++ b/remotes/honeybot/scripts/content_loader.py @@ -50,6 +50,43 @@ def check_standby(): return True return False + +# The Live Operator Break-In ("Max Headroom" channel). +# Dropped over SSH from Pipulate Prime as JSON or plain text; unlinked immediately +# on read so a one-shot live audience greeting never replays on future restarts. +LIVE_MESSAGE_FILE = Path("/home/mike/www/mikelev.in/.live_message") + + +def check_live_message(): + """Read and consume a one-shot live operator message if present.""" + if not LIVE_MESSAGE_FILE.exists(): + return None + try: + raw = LIVE_MESSAGE_FILE.read_text(encoding="utf-8").strip() + except OSError: + return None + finally: + try: + LIVE_MESSAGE_FILE.unlink(missing_ok=True) + except OSError: + pass + if not raw: + return None + if raw.startswith("{"): + try: + import json + data = json.loads(raw) + if isinstance(data, dict): + return data + except Exception: + pass + lines = raw.splitlines() + if lines and lines[0].upper().startswith("CARD:"): + card_label = lines[0][5:].strip() or "LIVE\\nMESSAGE" + say_text = " ".join(line.strip() for line in lines[1:] if line.strip()) + return {"card": card_label, "say": say_text} + return {"card": "LIVE\\nMESSAGE", "say": " ".join(line.strip() for line in lines if line.strip())} + def trigger_is_fresh(max_age_seconds=900): """True if the breaking-news bell (.reading_trigger) was rung within the last max_age_seconds. Reads the file's mtime, so it is process-boundary safe: it diff --git a/remotes/honeybot/scripts/stream.py b/remotes/honeybot/scripts/stream.py index c198162..6fbd193 100644 --- a/remotes/honeybot/scripts/stream.py +++ b/remotes/honeybot/scripts/stream.py @@ -133,9 +133,11 @@ sys.path.append(str(Path(__file__).parent)) try: import score - from content_loader import check_for_updates, check_standby, trigger_is_fresh + from content_loader import check_for_updates, check_standby, trigger_is_fresh, check_live_message except ImportError: score = None + def check_live_message(): + return None # --- Configuration --- MODEL_DIR = Path.home() / ".local/share/piper_voices" @@ -203,6 +205,7 @@ class Narrator(threading.Thread): self._proc_lock = threading.Lock() self._active_procs = [] self._tts_seq = 0 + self.exclusive_mode = threading.Event() def _cleanup_item(self, item): """Remove any unplayed pre-synthesized .raw file when flushing the queue.""" @@ -400,8 +403,9 @@ class Heartbeat(threading.Thread): if self.stop_event.wait(self.interval): break - now = datetime.datetime.now().strftime("%H:%M:%S") - narrator.say(f"Signal check. The time is {now}.") + if not narrator.exclusive_mode.is_set(): + now = datetime.datetime.now().strftime("%H:%M:%S") + narrator.say(f"Signal check. The time is {now}.") def stop(self): self.stop_event.set() @@ -665,7 +669,7 @@ def dispatch_cue(command, content, env, profile_dir, pace_say=True): # the real voice clock so the director never outruns the voice. raw_path = narrator.prep_raw(content) if raw_path is not None: - narrator.queue.join() + wait_for_voice_or_live_interrupt(env, profile_dir) narrator.play_raw(raw_path, fallback_text=content) else: narrator.say(content) @@ -747,6 +751,57 @@ def dispatch_cue(command, content, env, profile_dir, pace_say=True): pass +def run_live_message(msg, env, profile_dir): + """Preempt article narration for a live operator break-in, blocking audio + and Z-axis stomping until the operator's card and speech complete.""" + if not msg: + return + narrator.exclusive_mode.set() + try: + narrator.interrupt() + for pat in ("conjure_window_overlay", "patronus_visual_shield"): + try: + subprocess.run(["pkill", "-f", pat], check=False) + except Exception: + pass + cues = msg.get("cues") if isinstance(msg, dict) else None + if isinstance(cues, list) and cues: + for cue_cmd, cue_val in cues: + dispatch_cue(cue_cmd, cue_val, env, profile_dir, pace_say=True) + narrator.queue.join() + return + card_label = msg.get("card", "LIVE\\nMESSAGE") if isinstance(msg, dict) else "LIVE\\nMESSAGE" + say_text = msg.get("say", "") if isinstance(msg, dict) else str(msg) + patronus_key = msg.get("patronus") if isinstance(msg, dict) else None + hold = float(msg.get("duration", 120.0)) if isinstance(msg, dict) else 120.0 + if patronus_key and not card_label: + raw_path = narrator.prep_raw(say_text) if say_text else None + conjure_patronus(patronus_key, duration=hold) + if say_text: + narrator.play_raw(raw_path, fallback_text=say_text) + narrator.queue.join() + try: + subprocess.run(["pkill", "-f", "patronus_visual_shield"], check=False) + except Exception: + pass + else: + dispatch_cue("CARD", {"label": card_label or "LIVE\\nMESSAGE", "say": say_text, "duration": hold}, env, profile_dir, pace_say=True) + finally: + narrator.exclusive_mode.clear() + + +def wait_for_voice_or_live_interrupt(env, profile_dir): + """Wait for the active voice queue to drain while polling for a live operator break-in.""" + while True: + live_msg = check_live_message() + if live_msg: + run_live_message(live_msg, env, profile_dir) + with narrator.queue.all_tasks_done: + if narrator.queue.unfinished_tasks == 0: + return + narrator.queue.all_tasks_done.wait(timeout=0.15) + + def run_station_break(env, profile_dir): """Play the next forest bead as an out-of-band station break. @@ -766,18 +821,30 @@ def run_station_break(env, profile_dir): # finishes instead of being clipped mid-word -- only the queued backlog is # dropped. The queue.join() below then rides the real voice clock, so the # lead-in lands after the current line ends, not on top of it. - dropped = narrator.flush_queue() - if dropped: - # MEANINGFUL SILENCE OVER UNCONDITIONAL CHATTER: gated on the - # interesting case, so a healthy pipeline says nothing and an unhealthy - # one says exactly how unhealthy. Deliberately NOT added to interrupt() - # -- there, dropping the backlog is the INTENT (you are cutting to - # breaking news); here it is collateral. - print(f"πŸ“‰ STATION BREAK DROPPED {dropped} unspoken item(s) -- " - f"director ran ahead of the voice; that text was never heard.") - narrator.say("We interrupt this program for a station identification break.") - narrator.queue.join() - time.sleep(1.2) + narrator.exclusive_mode.set() + try: + dropped = narrator.flush_queue() + if dropped: + # MEANINGFUL SILENCE OVER UNCONDITIONAL CHATTER: gated on the + # interesting case, so a healthy pipeline says nothing and an unhealthy + # one says exactly how unhealthy. Deliberately NOT added to interrupt() + # -- there, dropping the backlog is the INTENT (you are cutting to + # breaking news); here it is collateral. + print(f"πŸ“‰ STATION BREAK DROPPED {dropped} unspoken item(s) -- " + f"director ran ahead of the voice; that text was never heard.") + narrator.say("We interrupt this program for a station identification break.") + narrator.queue.join() + time.sleep(1.2) + for cue_command, cue_content in bead: + dispatch_cue(cue_command, cue_content, env, profile_dir, pace_say=False) + narrator.queue.join() + time.sleep(1.2) + narrator.say("We now return to our regularly scheduled program.") + narrator.queue.join() + time.sleep(1.2) + finally: + narrator.exclusive_mode.clear() + return # --- Stage 2: Core Performance --- for cue_command, cue_content in bead: @@ -925,6 +992,9 @@ def perform_show(script): # news) stays above; the station-break trigger stays here; the brush # primitives live in dispatch_cue, shared by the article roll (trees) # and the forest roll (STATION_SEGMENTS beads in forest.py). + live_msg = check_live_message() + if live_msg: + run_live_message(live_msg, env, profile_dir) if command == "SAY" and time.time() >= next_pitch_time: # The Pervasive Pitch: play a forest bead as a station break # BEFORE the triggering sentence, preserving the article's flow. (nix) pipulate $ m πŸ“ Committing: chore: Refactor live message handling and introduce operational guardrails [main 5446525] chore: Refactor live message handling and introduce operational guardrails 3 files changed, 124 insertions(+), 16 deletions(-) (nix) pipulate $ git push Enumerating objects: 15, done. Counting objects: 100% (15/15), done. Delta compression using up to 48 threads Compressing objects: 100% (7/7), done. Writing objects: 100% (8/8), 2.51 KiB | 2.51 MiB/s, done. Total 8 (delta 6), reused 0 (delta 0), pack-reused 0 (from 0) remote: Resolving deltas: 100% (6/6), completed with 6 local objects. To github.com:pipulate/pipulate.git 15a1201..5446525 main -> main (nix) pipulate $ ``` It's going to take a `nixops` as expected after `ytsay` is wired up. **4: PROMPT**: Verify the AFTER receipt for `check_live_message` and `run_live_message`. Now inspect `remotes/honeybot/scripts/score.py`, `remotes/honeybot/scripts/forest.py`, `scripts/articles/common.py`, and `flake.nix`. Wire the `ytsay` command into `scripts/articles/common.py` and `flake.nix` so I can talk out to my YouTube audience with a custom Figlet card from Pipulate Prime, and show me how to expand the interleaved Storytime/Station-ID messaging in `score.py` and `forest.py`. **5: DELIVERABLE**: The ability to play Max Headroom on my own live-feed, or at least the first step towards something like that. **Gemini 4 Argon High**: Context verified. **TL;DR**: The AFTER receipt confirms `check_live_message` and `run_live_message` are live (`LIVE_MSG_FN: True True`). Below is how `score.py` ("The Trees") and `forest.py` ("The Forest") interleave during Storytime, followed by Car 1 wiring `ytsay` into `scripts/articles/common.py` and `flake.nix`, adding a closing bookend card to `score.py`, expanding `forest.py` from 2 to 5 station-break beads using your sealed `imports/ascii_displays.py` art, and making `run_station_break` in `stream.py` voice-synchronized and live-interruptible. --- ### 1. How Storytime (`score.py`) and Station Breaks (`forest.py`) Interleave Here is the exact control loop running on Honeybot inside `stream.py`: ```text start_director_track() β”‚ β–Ό score.get_script(breaking=...) <── "THE TREES" (remotes/honeybot/scripts/score.py) β”‚ 1. Cold start only: ("CARD", {"label": "HONEYBOT\nLIVE", "say": "Greetings, entity..."}) β”‚ 2. Article opener: ("CARD", {"label": "STORYTIME\nYYYY-MM-DD", "say": "... Title: ..."}) β”‚ 3. Body paragraphs: ("SAY", paragraph) + inline ("PATRONUS", {"key": ..., "duration": ...}) β”‚ 4. Article closer: ("CARD", {"label": "READ MORE\nMIKELEV.IN", "say": "..."}) [NEW] β–Ό perform_show(script) <── THE OUTER LOOP (remotes/honeybot/scripts/stream.py) β”‚ β”œβ”€β”€β–Ί Poll 1: check_standby() (.deploy_standby) -> UPDATING card + jeopardy.wav β”œβ”€β”€β–Ί Poll 2: check_for_updates() (.reading_trigger) -> "Breaking news detected" β”œβ”€β”€β–Ί Poll 3: check_live_message() (.live_message) -> MAX HEADROOM LIVE BREAK-IN (`ytsay`) β”‚ (also polled every 150ms inside wait_for_voice_or_live_interrupt) β”‚ └──► Copper List Grid: time.time() >= next_pitch_time (every 600s / 10 mins) β”‚ β–Ό run_station_break() <── "THE FOREST" (remotes/honeybot/scripts/forest.py) β€’ Bead 1: HONEYBOT + honeybot_pipeline (RFC 9110 Markdown vs. HTML) β€’ Bead 2: PIPULATE + white_rabbit (NPvg stack + CRC32 wax seal) β€’ Bead 3: THE AIRLOCK + mechanical_man (Tik-Tok of Oz + apply.py key) [NEW] β€’ Bead 4: AI PACHINKO + ai_pachinko (Stochastic pins + governor) [NEW] β€’ Bead 5: LOCAL FIRST + deployment_context(Localhost vs. cloud SaaS) [NEW] ``` Two subtle bugs in the station-break loop are also fixed in Car 1: 1. **Voice-driven `PATRONUS` teardown in `run_station_break()**`: Previously, `forest.py` set `"duration": 80.0` on its `PATRONUS` art, so if the 5 spoken lines finished in 48 seconds, the ASCII art popup lingered on top of the screen for another 32 seconds after Storytime resumed. `run_station_break()` now kills `patronus_visual_shield` the instant the bead's voice queue finishes, treating `80.0` as a safety ceiling just like `card.py` does. 2. **Dead-code cleanup + live interruptibility during station breaks**: Car 1 removes the 11 unreachable lines left after `return` in `run_station_break()` and routes its queue wait through `wait_for_voice_or_live_interrupt()` (guarded by `_in_live_message`) so `ytsay` preempts station breaks just as fast as article reading. ## Interleaving the Storytime Trees with Five Forest Beads --- ### 2. How to Use `ytsay` ("Max Headroom" Break-In) Once Car 1 lands (and you run `exit` then `ndq`, plus `./nixops.sh` to push the updated scripts to Honeybot), you have four ways to break in live from `(nix) pipulate $`: * **Default Card (`LIVE / FROM MIKE`) + Spoken Message**: `ytsay "Hey everyone watching live, thanks for tuning in to the Honeybot!"` * **Custom Figlet Banner (`-c` / `--card`) + Spoken Message**: `ytsay -c "MAX\nHEADROOM" "Hold the presses! A human is in the loop."` * **Registered Patronus ASCII Art (`-p` / `--patronus`) + Spoken Message**: `ytsay -p mechanical_man "Two keys wound by AI, one action key on the wall."` * **Clipboard Mode (`--clip`, e.g., after yanking a paragraph in Neovim)**: `ytsay -c "LIVE NOTE" --clip` --- **1: BEFORE** (PROBE): Run this in `(nix) pipulate $`: ```bash .venv/bin/python remotes/honeybot/scripts/test_forest.py --fast | grep -E "beads|BEAD|CARD|PATRONUS|DRIFT|All beads"; .venv/bin/python scripts/articles/common.py --broadcast say --dry-run -c "MAX\nHEADROOM" "Hello from Pipulate Prime" 2>&1 || echo "YTSAY_MISSING" ``` **2: AFTER** (NEXT CONTEXT): Paste into `context.txt` via `context` (or `x`): ```text foo_files.py remotes/honeybot/scripts/content_loader.py remotes/honeybot/scripts/stream.py remotes/honeybot/scripts/forest.py remotes/honeybot/scripts/score.py scripts/articles/common.py flake.nix ! .venv/bin/python remotes/honeybot/scripts/test_forest.py --fast | grep -E "beads|BEAD|CARD|PATRONUS|DRIFT|All beads"; .venv/bin/python scripts/articles/common.py --broadcast say --dry-run -c "MAX\nHEADROOM" "Hello from Pipulate Prime" 2>&1 || echo "YTSAY_MISSING" ! rg -n "yt(status|bell|roll|card|say)" scripts/articles/common.py flake.nix ``` **3: CHANGE** (PATCHES): Car 1 wires `ytsay` into `scripts/articles/common.py` and `flake.nix`, upgrades `run_live_message` and `run_station_break` in `remotes/honeybot/scripts/stream.py`, adds the closing article bookend card in `remotes/honeybot/scripts/score.py`, and expands `remotes/honeybot/scripts/forest.py` to 5 beads. ```text Target: remotes/honeybot/scripts/stream.py [[[SEARCH]]] def run_live_message(msg, env, profile_dir): """Preempt article narration for a live operator break-in, blocking audio and Z-axis stomping until the operator's card and speech complete.""" if not msg: return narrator.exclusive_mode.set() try: narrator.interrupt() for pat in ("conjure_window_overlay", "patronus_visual_shield"): try: subprocess.run(["pkill", "-f", pat], check=False) except Exception: pass cues = msg.get("cues") if isinstance(msg, dict) else None if isinstance(cues, list) and cues: for cue_cmd, cue_val in cues: dispatch_cue(cue_cmd, cue_val, env, profile_dir, pace_say=True) narrator.queue.join() return card_label = msg.get("card", "LIVE\\nMESSAGE") if isinstance(msg, dict) else "LIVE\\nMESSAGE" say_text = msg.get("say", "") if isinstance(msg, dict) else str(msg) patronus_key = msg.get("patronus") if isinstance(msg, dict) else None hold = float(msg.get("duration", 120.0)) if isinstance(msg, dict) else 120.0 if patronus_key and not card_label: raw_path = narrator.prep_raw(say_text) if say_text else None conjure_patronus(patronus_key, duration=hold) if say_text: narrator.play_raw(raw_path, fallback_text=say_text) narrator.queue.join() try: subprocess.run(["pkill", "-f", "patronus_visual_shield"], check=False) except Exception: pass else: dispatch_cue("CARD", {"label": card_label or "LIVE\\nMESSAGE", "say": say_text, "duration": hold}, env, profile_dir, pace_say=True) finally: narrator.exclusive_mode.clear() def wait_for_voice_or_live_interrupt(env, profile_dir): """Wait for the active voice queue to drain while polling for a live operator break-in.""" while True: live_msg = check_live_message() if live_msg: run_live_message(live_msg, env, profile_dir) with narrator.queue.all_tasks_done: if narrator.queue.unfinished_tasks == 0: return narrator.queue.all_tasks_done.wait(timeout=0.15) [[[DIVIDER]]] _in_live_message = False def run_live_message(msg, env, profile_dir): """Preempt article narration for a live operator break-in, blocking audio and Z-axis stomping until the operator's card and speech complete.""" global _in_live_message if not msg or _in_live_message: return _in_live_message = True narrator.exclusive_mode.set() try: narrator.interrupt() for pat in ("conjure_window_overlay", "patronus_visual_shield"): try: subprocess.run(["pkill", "-f", pat], check=False) except Exception: pass cues = msg.get("cues") if isinstance(msg, dict) else None if isinstance(cues, list) and cues: for cue_cmd, cue_val in cues: dispatch_cue(cue_cmd, cue_val, env, profile_dir, pace_say=True) narrator.queue.join() return card_label = msg.get("card", "") if isinstance(msg, dict) else "LIVE\\nFROM MIKE" say_text = msg.get("say", "") if isinstance(msg, dict) else str(msg) patronus_key = msg.get("patronus") if isinstance(msg, dict) else None hold = float(msg.get("duration", 120.0)) if isinstance(msg, dict) else 120.0 if patronus_key: if card_label: dispatch_cue("CARD", {"label": card_label, "say": "", "duration": 2.5}, env, profile_dir, pace_say=True) raw_path = narrator.prep_raw(say_text) if say_text else None conjure_patronus(patronus_key, duration=hold) if say_text: narrator.play_raw(raw_path, fallback_text=say_text) narrator.queue.join() else: time.sleep(min(hold, 4.0)) try: subprocess.run(["pkill", "-f", "patronus_visual_shield"], check=False) except Exception: pass else: dispatch_cue("CARD", {"label": card_label or "LIVE\\nFROM MIKE", "say": say_text, "duration": hold}, env, profile_dir, pace_say=True) finally: _in_live_message = False narrator.exclusive_mode.clear() def wait_for_voice_or_live_interrupt(env, profile_dir): """Wait for the active voice queue to drain while polling for a live operator break-in.""" while True: if not _in_live_message: live_msg = check_live_message() if live_msg: run_live_message(live_msg, env, profile_dir) with narrator.queue.all_tasks_done: if narrator.queue.unfinished_tasks == 0: return narrator.queue.all_tasks_done.wait(timeout=0.15) [[[REPLACE]]] Target: remotes/honeybot/scripts/stream.py [[[SEARCH]]] for cue_command, cue_content in bead: dispatch_cue(cue_command, cue_content, env, profile_dir, pace_say=False) narrator.queue.join() time.sleep(1.2) narrator.say("We now return to our regularly scheduled program.") narrator.queue.join() time.sleep(1.2) finally: narrator.exclusive_mode.clear() return # --- Stage 2: Core Performance --- for cue_command, cue_content in bead: dispatch_cue(cue_command, cue_content, env, profile_dir, pace_say=False) narrator.queue.join() # --- Stage 3: The Return Sentinel --- time.sleep(1.2) narrator.say("We now return to our regularly scheduled program.") narrator.queue.join() time.sleep(1.2) [[[DIVIDER]]] for cue_command, cue_content in bead: dispatch_cue(cue_command, cue_content, env, profile_dir, pace_say=False) wait_for_voice_or_live_interrupt(env, profile_dir) try: subprocess.run(["pkill", "-f", "patronus_visual_shield"], check=False) except Exception: pass time.sleep(1.2) narrator.say("We now return to our regularly scheduled program.") wait_for_voice_or_live_interrupt(env, profile_dir) time.sleep(1.2) finally: narrator.exclusive_mode.clear() [[[REPLACE]]] Target: remotes/honeybot/scripts/score.py [[[SEARCH]]] script.append(("SAY", "Reading entry...")) append_article_content(script, article) script.append(("WAIT", 3)) [[[DIVIDER]]] script.append(("SAY", "Reading entry...")) append_article_content(script, article) script.append(( "CARD", { "label": "READ MORE\nMIKELEV.IN", "say": ( f"That concludes {article['title']}. " "You can read the full text and source code at Mike L E V dot eye N. " "Drop a message in the live chat if you are tuning in. " "A human operator monitors this observatory and can break in live." ), }, )) script.append(("WAIT", 3)) [[[REPLACE]]] Target: remotes/honeybot/scripts/forest.py [[[SEARCH]]] ("SAY", ( "Pipulate sits where AI assistance meets local ownership. " "The whole bet is that you do not need a cloud subscription to do serious work. " "You need the right text files, the right habits, and a machine you actually own." )), # --- REPORT PLACEHOLDER (parked): ("WINDOW", "radar.py:30") --- ], ] [[[DIVIDER]]] ("SAY", ( "Pipulate sits where AI assistance meets local ownership. " "The whole bet is that you do not need a cloud subscription to do serious work. " "You need the right text files, the right habits, and a machine you actually own." )), # --- REPORT PLACEHOLDER (parked): ("WINDOW", "radar.py:30") --- ], [ # Bead 3: The Mechanical Man & The Airlock (Tik-Tok of Oz + apply.py). ("CARD", { "label": "THE AIRLOCK", "say": ( "How do you work with frontier AI models without letting hallucinations " "wreck your codebase? Meet Smith and Tinker's Patent Mechanical Man." ), }), ("PATRONUS", {"key": "mechanical_man", "duration": 80.0}), ("WAIT", 2), ("SAY", ( "In L. Frank Baum's Oz books, Tik-Tok is a clockwork robot with three separate " "wind-up keys: one for thought, one for speech, and one for action." )), ("SAY", ( "In our workflow, the context compiler winds the thought key, and the AI's reply " "winds the speech key. But the action key stays on the wall, in the human's hand." )), ("SAY", ( "Every code change must pass through an exact-match search and replace airlock " "with abstract syntax tree verification, and a human reviews the git diff before " "committing. No vibe-coded airplanes in the sky." )), ], [ # Bead 4: AI Pachinko & The Cybernetic Governor. ("CARD", { "label": "AI PACHINKO", "say": ( "Every token an AI generates is a ball dropped through a pachinko parlor " "of frozen mathematical weights." ), }), ("PATRONUS", {"key": "ai_pachinko", "duration": 80.0}), ("WAIT", 2), ("SAY", ( "An LLM has no persistent memory of its own between sessions. When the chat tab " "closes, its working memory dissolves. Left to itself, an unpaginated loop " "drifts into sycophancy and entropy." )), ("SAY", ( "That is why we pair the pachinko machine with a cybernetic governor: plain-text " "files in git, before-and-after falsifying probes, and a human operator who " "decides whether to bank the win or re-spin." )), ], [ # Bead 5: Local-First Sovereignty & Live Human Presence. ("CARD", { "label": "LOCAL FIRST", "say": ( "What you are watching is not a cloud simulation. It is a local-first, " "single-tenant observatory running on hardware in a home closet." ), }), ("PATRONUS", {"key": "deployment_context", "duration": 80.0}), ("WAIT", 2), ("SAY", ( "Most modern web platforms hide your own server logs behind multi-tenant dashboards " "and rent your own workflows back to you by the month." )), ("SAY", ( "Here, everything runs on NixOS, Python, Neovim, and Git. And although this broadcast " "is automated twenty-four seven, a human operator is often at the workbench and can " "break in live over this channel at any moment. Leave a message in the chat and say hello." )), ], ] [[[REPLACE]]] Target: scripts/articles/common.py [[[SEARCH]]] print("πŸŽ›οΈ Quick Words: tweet | yttitle | ytdesc | ytcard | ytroll | ytbell | ytstatus") [[[DIVIDER]]] print("πŸŽ›οΈ Quick Words: tweet | yttitle | ytdesc | ytcard | ytsay | ytroll | ytbell | ytstatus") [[[REPLACE]]] Target: scripts/articles/common.py [[[SEARCH]]] def run_honeybot_obs(mode="status"): """Execute remote OBS WebSocket / bell control on Honeybot over SSH.""" import subprocess res = subprocess.run( ["ssh", HONEYBOT_SSH_HOST, "python3", "-", mode], input=_HONEYBOT_OBS_HELPER, text=True, ) return res.returncode [[[DIVIDER]]] def run_honeybot_obs(mode="status"): """Execute remote OBS WebSocket / bell control on Honeybot over SSH.""" import subprocess res = subprocess.run( ["ssh", HONEYBOT_SSH_HOST, "python3", "-", mode], input=_HONEYBOT_OBS_HELPER, text=True, ) return res.returncode def _read_from_clipboard(): """Read text from the OS clipboard for `ytsay --clip`.""" import subprocess cmd = ["pbpaste"] if sys.platform == "darwin" else ["xclip", "-selection", "clipboard", "-o"] try: res = subprocess.run(cmd, capture_output=True, text=True, check=True) return res.stdout.strip() except Exception: return "" def send_honeybot_live_message(say_text, card_label=None, patronus_key=None, dry_run=False): """Send a one-shot live operator break-in (.live_message) to Honeybot over SSH.""" import subprocess import time clean_say = " ".join((say_text or "").strip().split()) if not clean_say and not card_label and not patronus_key: print("Usage: ytsay [-c 'CARD\\nTEXT'] [-p patronus_key] [--clip] [--dry-run] 'Message to speak live'") return 1 if card_label is None: resolved_card = "" if patronus_key else "LIVE\nFROM MIKE" else: resolved_card = str(card_label).replace("\\n", "\n") payload = { "card": resolved_card, "say": clean_say, "sent_at": int(time.time()), } if patronus_key: payload["patronus"] = str(patronus_key).strip() payload_json = json.dumps(payload, indent=2) if dry_run: print("πŸ“‘ [DRY RUN] Live break-in payload for Honeybot:") print(payload_json) return 0 remote_py = ( "import pathlib, sys; " "p = pathlib.Path.home() / 'www/mikelev.in/.live_message'; " "p.parent.mkdir(parents=True, exist_ok=True); " "tmp = p.with_suffix('.tmp'); " "tmp.write_text(sys.stdin.read(), encoding='utf-8'); " "tmp.replace(p); " "print(f'πŸ“‘ Live break-in queued on Honeybot: {p}')" ) res = subprocess.run( ["ssh", HONEYBOT_SSH_HOST, "python3", "-c", remote_py], input=payload_json, text=True, ) if res.returncode == 0: preview_card = resolved_card.replace("\n", " | ") if resolved_card else f"PATRONUS:{patronus_key}" print(f"πŸŽ™οΈ Card: [{preview_card}] | Say: {clean_say!r}") return res.returncode [[[REPLACE]]] Target: scripts/articles/common.py [[[SEARCH]]] if __name__ == "__main__": parser = argparse.ArgumentParser(description="Pipulate article common utilities & broadcast kit.") add_standard_arguments(parser) parser.add_argument( "--broadcast", choices=("card", "tweet", "title", "desc", "status", "bell", "stop", "start", "roll", "enable-ws"), default="card", help="Print/copy broadcast kit or control Honeybot OBS (status|bell|stop|start|roll|enable-ws).", ) parser.add_argument( "-s", "--slug", type=str, default=None, help="Optional substring to match a specific dated post filename.", ) cli_args = parser.parse_args() if cli_args.broadcast in ("status", "bell", "stop", "start", "roll", "enable-ws"): sys.exit(run_honeybot_obs(cli_args.broadcast)) sys.exit(print_broadcast_kit(cli_args.target, cli_args.broadcast, slug_match=cli_args.slug)) [[[DIVIDER]]] if __name__ == "__main__": parser = argparse.ArgumentParser(description="Pipulate article common utilities & broadcast kit.") add_standard_arguments(parser) parser.add_argument( "--broadcast", choices=("card", "tweet", "title", "desc", "status", "bell", "stop", "start", "roll", "enable-ws", "say"), default="card", help="Print/copy broadcast kit or control Honeybot (status|bell|stop|start|roll|enable-ws|say).", ) parser.add_argument( "-s", "--slug", type=str, default=None, help="Optional substring to match a specific dated post filename.", ) parser.add_argument( "-c", "--card", type=str, default=None, help="Figlet card label for --broadcast say (use \\n for multiple lines; default: LIVE\\nFROM MIKE).", ) parser.add_argument( "-p", "--patronus", type=str, default=None, help="Optional figurate ASCII art key (e.g. white_rabbit) for --broadcast say.", ) parser.add_argument( "--clip", action="store_true", help="Read the spoken message from the OS clipboard for --broadcast say.", ) parser.add_argument( "--dry-run", action="store_true", help="Print the .live_message JSON payload without sending over SSH.", ) parser.add_argument( "message", nargs="*", help="Spoken message words for --broadcast say.", ) cli_args = parser.parse_args() if cli_args.broadcast == "say": msg_text = _read_from_clipboard() if cli_args.clip else " ".join(cli_args.message) sys.exit( send_honeybot_live_message( msg_text, card_label=cli_args.card, patronus_key=cli_args.patronus, dry_run=cli_args.dry_run, ) ) if cli_args.broadcast in ("status", "bell", "stop", "start", "roll", "enable-ws"): sys.exit(run_honeybot_obs(cli_args.broadcast)) sys.exit(print_broadcast_kit(cli_args.target, cli_args.broadcast, slug_match=cli_args.slug)) [[[REPLACE]]] Target: flake.nix [[[SEARCH]]] ytstatus() { "$PIPULATE_ROOT/.venv/bin/python" "$PIPULATE_ROOT/scripts/articles/common.py" --broadcast status "$@"; } ytbell() { "$PIPULATE_ROOT/.venv/bin/python" "$PIPULATE_ROOT/scripts/articles/common.py" --broadcast bell "$@"; } ytroll() { "$PIPULATE_ROOT/.venv/bin/python" "$PIPULATE_ROOT/scripts/articles/common.py" --broadcast "''${1:-roll}" "''${@:2}"; } [[[DIVIDER]]] ytstatus() { "$PIPULATE_ROOT/.venv/bin/python" "$PIPULATE_ROOT/scripts/articles/common.py" --broadcast status "$@"; } ytbell() { "$PIPULATE_ROOT/.venv/bin/python" "$PIPULATE_ROOT/scripts/articles/common.py" --broadcast bell "$@"; } ytroll() { "$PIPULATE_ROOT/.venv/bin/python" "$PIPULATE_ROOT/scripts/articles/common.py" --broadcast "''${1:-roll}" "''${@:2}"; } ytsay() { "$PIPULATE_ROOT/.venv/bin/python" "$PIPULATE_ROOT/scripts/articles/common.py" --broadcast say "$@"; } [[[REPLACE]]] ``` Car 2 deploys the updated Honeybot scripts (`content_loader.py`, `stream.py`, `score.py`, `forest.py`) to Honeybot via `./nixops.sh` and restarts `stream.py` so the live break-in listener and the 5 station-break beads are active on air. Run this in `(nix) pipulate $`: ```bash ./nixops.sh && ssh honeybot 'pkill -TERM -f "/home/mike/www/mikelev[.]in/scripts/stream[.]py" || true; sleep 6; pgrep -af "/home/mike/www/mikelev[.]in/scripts/stream[.]py"' ``` Ignition: Run `exit` then `ndq` (or `nix develop`) so the new `ytsay` shell function from `flake.nix` loads into your interactive shell, then test a live break-in on your stream with `ytsay -c "LIVE\nFROM MIKE" "Testing one two three from Pipulate Prime."`. **4: PROMPT**: ```text Verify the AFTER receipts for the 5 `forest.py` station-break beads, `ytsay` (`common.py --broadcast say --dry-run`), and the live Honeybot deployment. If everything is green and witnessed, we are ready to dismount or test a live break-in on the air. ``` **5: DELIVERABLE**: * `ytsay` command wired into `scripts/articles/common.py` (`--broadcast say`) and `flake.nix` supporting custom Figlet cards (`-c`), Patronus ASCII art (`-p`), clipboard input (`--clip`), and dry-run inspection (`--dry-run`). * `remotes/honeybot/scripts/forest.py` expanded from 2 to 5 self-contained station-break beads (`HONEYBOT`, `PIPULATE`, `THE AIRLOCK`, `AI PACHINKO`, `LOCAL FIRST`) verified against `FIGURATE_LEDGER` CRC32 seals. * `remotes/honeybot/scripts/score.py` updated with a closing `READ MORE / MIKELEV.IN` bookend card inviting live chat interaction. * `remotes/honeybot/scripts/stream.py` cleaned of unreachable lines in `run_station_break` and upgraded with voice-synchronized Patronus teardown and live-interrupt polling during station breaks. ## Engineering the Four-Beat Palate Cleanser Envelope **MikeLev.in**: ## THE AI-EDIT METHOD > Same commands, run twice, one change between them. Where the readings > differ is what the change did; the diff in the middle is the receipt. **1: BEFORE** (PROBE): ```bash On branch main Your branch is up to date with 'origin/main'. nothing to commit, working tree clean GIT repo clean. Take BEFORE reading, make CHANGE, record AFTER diff. (nix) pipulate $ .venv/bin/python remotes/honeybot/scripts/test_forest.py --fast | grep -E "beads|BEAD|CARD|PATRONUS|DRIFT|All beads"; .venv/bin/python scripts/articles/common.py --broadcast say --dry-run -c "MAX\nHEADROOM" "Hello from Pipulate Prime" 2>&1 || echo "YTSAY_MISSING" 🎨 FIGURATE: DRIFT DETECTED in 'honeybot_pipeline' -- expected CRC 2035836683, got 3340253463 beads : 2 🎬 BEAD 1/2 πŸͺ§ CARD (voice-synced): 'HONEYBOT' 🎨 PATRONUS: 'honeybot_pipeline' (rendered inline; popup geometry NOT tested here) ⚠️ DRIFT DETECTED in 'honeybot_pipeline': wax-seal CRC mismatch -- the art changed but FIGURATE_LEDGER was not updated. 🎬 BEAD 2/2 πŸͺ§ CARD (voice-synced): 'PIPULATE' 🎨 PATRONUS: 'white_rabbit' (rendered inline; popup geometry NOT tested here) βœ… All beads reviewed. Content fidelity checked; popup geometry still owed to Honeybot. usage: common.py [-h] [-t TARGET] [-k KEY] [--broadcast {card,tweet,title,desc,status,bell,stop,start,roll,enable-ws}] [-s SLUG] common.py: error: argument --broadcast: invalid choice: 'say' (choose from card, tweet, title, desc, status, bell, stop, start, roll, enable-ws) YTSAY_MISSING (nix) pipulate $ ``` **2: AFTER** (NEXT CONTEXT): ```text # # Context 1 # foo_files.py # imports/ascii_displays.py # init.lua # ~/repos/nixos/configuration.nix # ~/repos/nixos/autognome.py # scripts/card.py # remotes/honeybot/scripts/card.py # nixops.sh # remotes/honeybot/hooks/post-receive # remotes/honeybot/nixos/configuration.nix # remotes/honeybot/scripts/stream.py # remotes/honeybot/scripts/card.py # remotes/honeybot/scripts/forest.py # remotes/honeybot/scripts/test_forest.py # remotes/honeybot/scripts/logs.py # remotes/honeybot/scripts/content_loader.py # remotes/honeybot/scripts/db.py # # Context 2 # foo_files.py # remotes/honeybot/scripts/content_loader.py # remotes/honeybot/scripts/stream.py # remotes/honeybot/scripts/forest.py # remotes/honeybot/scripts/test_forest.py # remotes/honeybot/scripts/score.py # scripts/articles/common.py # flake.nix # nixops.sh # ! diff -u scripts/card.py remotes/honeybot/scripts/card.py | head -n 30; .venv/bin/python -c "import sys; sys.path.insert(0, 'remotes/honeybot/scripts'); import content_loader, stream; print('LIVE_MSG_FN:', hasattr(content_loader, 'check_live_message'), hasattr(stream, 'run_live_message'))" # ! rg -n "yt(status|bell|roll|card|say)" scripts/articles/common.py flake.nix # Context 3 foo_files.py remotes/honeybot/scripts/content_loader.py remotes/honeybot/scripts/stream.py remotes/honeybot/scripts/forest.py remotes/honeybot/scripts/score.py scripts/articles/common.py flake.nix ! .venv/bin/python remotes/honeybot/scripts/test_forest.py --fast | grep -E "beads|BEAD|CARD|PATRONUS|DRIFT|All beads"; .venv/bin/python scripts/articles/common.py --broadcast say --dry-run -c "MAX\nHEADROOM" "Hello from Pipulate Prime" 2>&1 || echo "YTSAY_MISSING" ! rg -n "yt(status|bell|roll|card|say)" scripts/articles/common.py flake.nix ``` **3: CHANGE** (PATCHES): ```diff On branch main Your branch is up to date with 'origin/main'. nothing to commit, working tree clean GIT repo clean. Take BEFORE reading, make CHANGE, record AFTER diff. (nix) pipulate $ patch (nix) pipulate $ app βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'remotes/honeybot/scripts/stream.py'. βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'remotes/honeybot/scripts/stream.py'. βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'remotes/honeybot/scripts/score.py'. βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'remotes/honeybot/scripts/forest.py'. βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'scripts/articles/common.py'. βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'scripts/articles/common.py'. βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'scripts/articles/common.py'. βœ… DETERMINISTIC PATCH APPLIED: Successfully mutated 'flake.nix'. (nix) pipulate $ d diff --git a/flake.nix b/flake.nix index aecfb88..c2b996d 100644 --- a/flake.nix +++ b/flake.nix @@ -2477,6 +2477,7 @@ print('AI:\n', r.ai) ytstatus() { "$PIPULATE_ROOT/.venv/bin/python" "$PIPULATE_ROOT/scripts/articles/common.py" --broadcast status "$@"; } ytbell() { "$PIPULATE_ROOT/.venv/bin/python" "$PIPULATE_ROOT/scripts/articles/common.py" --broadcast bell "$@"; } ytroll() { "$PIPULATE_ROOT/.venv/bin/python" "$PIPULATE_ROOT/scripts/articles/common.py" --broadcast "''${1:-roll}" "''${@:2}"; } + ytsay() { "$PIPULATE_ROOT/.venv/bin/python" "$PIPULATE_ROOT/scripts/articles/common.py" --broadcast say "$@"; } # The true 'publish' command (Atomic Cross-Domain Deployment) # It requires a commit message as an argument. diff --git a/remotes/honeybot/scripts/forest.py b/remotes/honeybot/scripts/forest.py index 034cf0a..c7247f4 100644 --- a/remotes/honeybot/scripts/forest.py +++ b/remotes/honeybot/scripts/forest.py @@ -107,4 +107,72 @@ STATION_SEGMENTS = [ )), # --- REPORT PLACEHOLDER (parked): ("WINDOW", "radar.py:30") --- ], + [ + # Bead 3: The Mechanical Man & The Airlock (Tik-Tok of Oz + apply.py). + ("CARD", { + "label": "THE AIRLOCK", + "say": ( + "How do you work with frontier AI models without letting hallucinations " + "wreck your codebase? Meet Smith and Tinker's Patent Mechanical Man." + ), + }), + ("PATRONUS", {"key": "mechanical_man", "duration": 80.0}), + ("WAIT", 2), + ("SAY", ( + "In L. Frank Baum's Oz books, Tik-Tok is a clockwork robot with three separate " + "wind-up keys: one for thought, one for speech, and one for action." + )), + ("SAY", ( + "In our workflow, the context compiler winds the thought key, and the AI's reply " + "winds the speech key. But the action key stays on the wall, in the human's hand." + )), + ("SAY", ( + "Every code change must pass through an exact-match search and replace airlock " + "with abstract syntax tree verification, and a human reviews the git diff before " + "committing. No vibe-coded airplanes in the sky." + )), + ], + [ + # Bead 4: AI Pachinko & The Cybernetic Governor. + ("CARD", { + "label": "AI PACHINKO", + "say": ( + "Every token an AI generates is a ball dropped through a pachinko parlor " + "of frozen mathematical weights." + ), + }), + ("PATRONUS", {"key": "ai_pachinko", "duration": 80.0}), + ("WAIT", 2), + ("SAY", ( + "An LLM has no persistent memory of its own between sessions. When the chat tab " + "closes, its working memory dissolves. Left to itself, an unpaginated loop " + "drifts into sycophancy and entropy." + )), + ("SAY", ( + "That is why we pair the pachinko machine with a cybernetic governor: plain-text " + "files in git, before-and-after falsifying probes, and a human operator who " + "decides whether to bank the win or re-spin." + )), + ], + [ + # Bead 5: Local-First Sovereignty & Live Human Presence. + ("CARD", { + "label": "LOCAL FIRST", + "say": ( + "What you are watching is not a cloud simulation. It is a local-first, " + "single-tenant observatory running on hardware in a home closet." + ), + }), + ("PATRONUS", {"key": "deployment_context", "duration": 80.0}), + ("WAIT", 2), + ("SAY", ( + "Most modern web platforms hide your own server logs behind multi-tenant dashboards " + "and rent your own workflows back to you by the month." + )), + ("SAY", ( + "Here, everything runs on NixOS, Python, Neovim, and Git. And although this broadcast " + "is automated twenty-four seven, a human operator is often at the workbench and can " + "break in live over this channel at any moment. Leave a message in the chat and say hello." + )), + ], ] diff --git a/remotes/honeybot/scripts/score.py b/remotes/honeybot/scripts/score.py index a5869c3..0ad354e 100644 --- a/remotes/honeybot/scripts/score.py +++ b/remotes/honeybot/scripts/score.py @@ -46,6 +46,18 @@ def _append_article_performance(script, article, is_latest=True): )) script.append(("SAY", "Reading entry...")) append_article_content(script, article) + script.append(( + "CARD", + { + "label": "READ MORE\nMIKELEV.IN", + "say": ( + f"That concludes {article['title']}. " + "You can read the full text and source code at Mike L E V dot eye N. " + "Drop a message in the live chat if you are tuning in. " + "A human operator monitors this observatory and can break in live." + ), + }, + )) script.append(("WAIT", 3)) diff --git a/remotes/honeybot/scripts/stream.py b/remotes/honeybot/scripts/stream.py index 6fbd193..dc43117 100644 --- a/remotes/honeybot/scripts/stream.py +++ b/remotes/honeybot/scripts/stream.py @@ -751,11 +751,16 @@ def dispatch_cue(command, content, env, profile_dir, pace_say=True): pass +_in_live_message = False + + def run_live_message(msg, env, profile_dir): """Preempt article narration for a live operator break-in, blocking audio and Z-axis stomping until the operator's card and speech complete.""" - if not msg: + global _in_live_message + if not msg or _in_live_message: return + _in_live_message = True narrator.exclusive_mode.set() try: narrator.interrupt() @@ -770,32 +775,38 @@ def run_live_message(msg, env, profile_dir): dispatch_cue(cue_cmd, cue_val, env, profile_dir, pace_say=True) narrator.queue.join() return - card_label = msg.get("card", "LIVE\\nMESSAGE") if isinstance(msg, dict) else "LIVE\\nMESSAGE" + card_label = msg.get("card", "") if isinstance(msg, dict) else "LIVE\\nFROM MIKE" say_text = msg.get("say", "") if isinstance(msg, dict) else str(msg) patronus_key = msg.get("patronus") if isinstance(msg, dict) else None hold = float(msg.get("duration", 120.0)) if isinstance(msg, dict) else 120.0 - if patronus_key and not card_label: + if patronus_key: + if card_label: + dispatch_cue("CARD", {"label": card_label, "say": "", "duration": 2.5}, env, profile_dir, pace_say=True) raw_path = narrator.prep_raw(say_text) if say_text else None conjure_patronus(patronus_key, duration=hold) if say_text: narrator.play_raw(raw_path, fallback_text=say_text) narrator.queue.join() + else: + time.sleep(min(hold, 4.0)) try: subprocess.run(["pkill", "-f", "patronus_visual_shield"], check=False) except Exception: pass else: - dispatch_cue("CARD", {"label": card_label or "LIVE\\nMESSAGE", "say": say_text, "duration": hold}, env, profile_dir, pace_say=True) + dispatch_cue("CARD", {"label": card_label or "LIVE\\nFROM MIKE", "say": say_text, "duration": hold}, env, profile_dir, pace_say=True) finally: + _in_live_message = False narrator.exclusive_mode.clear() def wait_for_voice_or_live_interrupt(env, profile_dir): """Wait for the active voice queue to drain while polling for a live operator break-in.""" while True: - live_msg = check_live_message() - if live_msg: - run_live_message(live_msg, env, profile_dir) + if not _in_live_message: + live_msg = check_live_message() + if live_msg: + run_live_message(live_msg, env, profile_dir) with narrator.queue.all_tasks_done: if narrator.queue.unfinished_tasks == 0: return @@ -837,25 +848,17 @@ def run_station_break(env, profile_dir): time.sleep(1.2) for cue_command, cue_content in bead: dispatch_cue(cue_command, cue_content, env, profile_dir, pace_say=False) - narrator.queue.join() + wait_for_voice_or_live_interrupt(env, profile_dir) + try: + subprocess.run(["pkill", "-f", "patronus_visual_shield"], check=False) + except Exception: + pass time.sleep(1.2) narrator.say("We now return to our regularly scheduled program.") - narrator.queue.join() + wait_for_voice_or_live_interrupt(env, profile_dir) time.sleep(1.2) finally: narrator.exclusive_mode.clear() - return - - # --- Stage 2: Core Performance --- - for cue_command, cue_content in bead: - dispatch_cue(cue_command, cue_content, env, profile_dir, pace_say=False) - narrator.queue.join() - - # --- Stage 3: The Return Sentinel --- - time.sleep(1.2) - narrator.say("We now return to our regularly scheduled program.") - narrator.queue.join() - time.sleep(1.2) def perform_show(script): diff --git a/scripts/articles/common.py b/scripts/articles/common.py index 20443f2..3ab5a8b 100644 --- a/scripts/articles/common.py +++ b/scripts/articles/common.py @@ -470,7 +470,7 @@ def print_broadcast_kit(target_key="1", mode="card", slug_match=None): print("πŸ“ YOUTUBE LIVE DESCRIPTION (type `ytdesc` to copy):") print(kit["yt_description"]) print(thin) - print("πŸŽ›οΈ Quick Words: tweet | yttitle | ytdesc | ytcard | ytroll | ytbell | ytstatus") + print("πŸŽ›οΈ Quick Words: tweet | yttitle | ytdesc | ytcard | ytsay | ytroll | ytbell | ytstatus") print(thin) print(f"πŸ”— Preview URL: {kit['preview_url']}") print(f"✏️ Edit Command: vim {kit['path']}") @@ -740,14 +740,69 @@ def run_honeybot_obs(mode="status"): return res.returncode +def _read_from_clipboard(): + """Read text from the OS clipboard for `ytsay --clip`.""" + import subprocess + cmd = ["pbpaste"] if sys.platform == "darwin" else ["xclip", "-selection", "clipboard", "-o"] + try: + res = subprocess.run(cmd, capture_output=True, text=True, check=True) + return res.stdout.strip() + except Exception: + return "" + + +def send_honeybot_live_message(say_text, card_label=None, patronus_key=None, dry_run=False): + """Send a one-shot live operator break-in (.live_message) to Honeybot over SSH.""" + import subprocess + import time + clean_say = " ".join((say_text or "").strip().split()) + if not clean_say and not card_label and not patronus_key: + print("Usage: ytsay [-c 'CARD\\nTEXT'] [-p patronus_key] [--clip] [--dry-run] 'Message to speak live'") + return 1 + if card_label is None: + resolved_card = "" if patronus_key else "LIVE\nFROM MIKE" + else: + resolved_card = str(card_label).replace("\\n", "\n") + payload = { + "card": resolved_card, + "say": clean_say, + "sent_at": int(time.time()), + } + if patronus_key: + payload["patronus"] = str(patronus_key).strip() + payload_json = json.dumps(payload, indent=2) + if dry_run: + print("πŸ“‘ [DRY RUN] Live break-in payload for Honeybot:") + print(payload_json) + return 0 + remote_py = ( + "import pathlib, sys; " + "p = pathlib.Path.home() / 'www/mikelev.in/.live_message'; " + "p.parent.mkdir(parents=True, exist_ok=True); " + "tmp = p.with_suffix('.tmp'); " + "tmp.write_text(sys.stdin.read(), encoding='utf-8'); " + "tmp.replace(p); " + "print(f'πŸ“‘ Live break-in queued on Honeybot: {p}')" + ) + res = subprocess.run( + ["ssh", HONEYBOT_SSH_HOST, "python3", "-c", remote_py], + input=payload_json, + text=True, + ) + if res.returncode == 0: + preview_card = resolved_card.replace("\n", " | ") if resolved_card else f"PATRONUS:{patronus_key}" + print(f"πŸŽ™οΈ Card: [{preview_card}] | Say: {clean_say!r}") + return res.returncode + + if __name__ == "__main__": parser = argparse.ArgumentParser(description="Pipulate article common utilities & broadcast kit.") add_standard_arguments(parser) parser.add_argument( "--broadcast", - choices=("card", "tweet", "title", "desc", "status", "bell", "stop", "start", "roll", "enable-ws"), + choices=("card", "tweet", "title", "desc", "status", "bell", "stop", "start", "roll", "enable-ws", "say"), default="card", - help="Print/copy broadcast kit or control Honeybot OBS (status|bell|stop|start|roll|enable-ws).", + help="Print/copy broadcast kit or control Honeybot (status|bell|stop|start|roll|enable-ws|say).", ) parser.add_argument( "-s", "--slug", @@ -755,7 +810,44 @@ if __name__ == "__main__": default=None, help="Optional substring to match a specific dated post filename.", ) + parser.add_argument( + "-c", "--card", + type=str, + default=None, + help="Figlet card label for --broadcast say (use \\n for multiple lines; default: LIVE\\nFROM MIKE).", + ) + parser.add_argument( + "-p", "--patronus", + type=str, + default=None, + help="Optional figurate ASCII art key (e.g. white_rabbit) for --broadcast say.", + ) + parser.add_argument( + "--clip", + action="store_true", + help="Read the spoken message from the OS clipboard for --broadcast say.", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Print the .live_message JSON payload without sending over SSH.", + ) + parser.add_argument( + "message", + nargs="*", + help="Spoken message words for --broadcast say.", + ) cli_args = parser.parse_args() + if cli_args.broadcast == "say": + msg_text = _read_from_clipboard() if cli_args.clip else " ".join(cli_args.message) + sys.exit( + send_honeybot_live_message( + msg_text, + card_label=cli_args.card, + patronus_key=cli_args.patronus, + dry_run=cli_args.dry_run, + ) + ) if cli_args.broadcast in ("status", "bell", "stop", "start", "roll", "enable-ws"): sys.exit(run_honeybot_obs(cli_args.broadcast)) sys.exit(print_broadcast_kit(cli_args.target, cli_args.broadcast, slug_match=cli_args.slug)) (nix) pipulate $ m πŸ“ Committing: feat: implement honeybot live messages and voice over [main fe879c5] feat: implement honeybot live messages and voice over 5 files changed, 200 insertions(+), 24 deletions(-) (nix) pipulate $ git push Enumerating objects: 23, done. Counting objects: 100% (23/23), done. Delta compression using up to 48 threads Compressing objects: 100% (11/11), done. Writing objects: 100% (12/12), 3.87 KiB | 1.93 MiB/s, done. Total 12 (delta 10), reused 0 (delta 0), pack-reused 0 (from 0) remote: Resolving deltas: 100% (10/10), completed with 10 local objects. To github.com:pipulate/pipulate.git 5446525..fe879c5 main -> main (nix) pipulate $ ``` Ignition: often "exit", "nix develop". Instructions may vary. Alright, we deploy: ```bash On branch main Your branch is up to date with 'origin/main'. nothing to commit, working tree clean GIT repo clean. Take BEFORE reading, make CHANGE, record AFTER diff. (nix) pipulate $ ./nixops.sh && ssh honeybot 'pkill -TERM -f "/home/mike/www/mikelev[.]in/scripts/stream[.]py" || true; sleep 6; pgrep -af "/home/mike/www/mikelev[.]in/scripts/stream[.]py"' πŸš€ Syncing Hooks... post-receive 100% 3080 572.6KB/s 00:00 πŸš€ Syncing Scripts (New Location)... sending incremental file list ./ content_loader.py forest.py score.py stream.py __pycache__/ __pycache__/content_loader.cpython-312.pyc __pycache__/forest.cpython-312.pyc __pycache__/score.cpython-312.pyc __pycache__/stream.cpython-312.pyc sent 89,129 bytes received 813 bytes 179,884.00 bytes/sec total size is 212,918 speedup is 2.37 sending incremental file list sent 68 bytes received 12 bytes 160.00 bytes/sec total size is 140,367 speedup is 1,754.59 πŸš€ Syncing NPvg pad (one address, two bodies)... sending incremental file list sent 232 bytes received 16 bytes 496.00 bytes/sec total size is 5,440 speedup is 21.94 πŸš€ Syncing qamy.ai door (npvg.org's shape, its own tree)... sending incremental file list sent 232 bytes received 16 bytes 496.00 bytes/sec total size is 4,646 speedup is 18.73 πŸš€ Syncing installer to the home-hosted doors (npvg.org, qamy.ai)... sending incremental file list sent 61 bytes received 12 bytes 146.00 bytes/sec total size is 23,547 speedup is 322.56 sending incremental file list sent 61 bytes received 12 bytes 146.00 bytes/sec total size is 23,547 speedup is 322.56 πŸš€ Syncing NixOS Config... sending incremental file list sent 87 bytes received 12 bytes 198.00 bytes/sec total size is 24,956 speedup is 252.08 βœ… Sync Complete. To apply NixOS config: ssh -t mike@[REDACTED_IP] 'sudo cp ~/nixos-config-staged/* /etc/nixos/ && sudo nixos-rebuild switch' 3785933 nix develop /home/mike/www/mikelev.in#quiet --command python3 -u /home/mike/www/mikelev.in/scripts/stream.py (nix) pipulate $ ssh -t mike@[REDACTED_IP] 'sudo cp ~/nixos-config-staged/* /etc/nixos/ && sudo nixos-rebuild switch' [sudo] password for mike: building the system configuration... evaluation warning: The option `services.logind.lidSwitchExternalPower' defined in `/etc/nixos/configuration.nix' has been renamed to `services.logind.settings.Login.HandleLidSwitchExternalPower'. evaluation warning: The option `services.logind.lidSwitch' defined in `/etc/nixos/configuration.nix' has been renamed to `services.logind.settings.Login.HandleLidSwitch'. evaluation warning: The option `services.xserver.displayManager.gdm.wayland' defined in `/etc/nixos/configuration.nix' has been renamed to `services.displayManager.gdm.wayland'. evaluation warning: The option `services.xserver.displayManager.gdm.enable' defined in `/etc/nixos/configuration.nix' has been renamed to `services.displayManager.gdm.enable'. activating the configuration... setting up /etc... reloading user units for mike... restarting sysinit-reactivation.target the following new units were started: NetworkManager-dispatcher.service Done. The new configuration is /nix/store/c17qykxpwxjmdprxnfnqsnaln11f2yva-nixos-system-honeybot-26.05pre913595.c6245e83d836 Connection to [REDACTED_IP] closed. (nix) pipulate $ ``` Wow, okay I see a copying message on the livestream and am waiting for it to go away and restart the cycle. Okay, that's done. I should be able to do this now. Oh, whoops: ```bash On branch main Your branch is up to date with 'origin/main'. nothing to commit, working tree clean GIT repo clean. Take BEFORE reading, make CHANGE, record AFTER diff. (nix) pipulate $ exit exit (sys) pipulate $ ndq (nix) pipulate $ ytsay -c "LIVE\nFROM MIKE" "Testing one two three from Pipulate Prime." bash: -c: line 1: syntax error near unexpected token `(' bash: -c: line 1: `python3 -c import pathlib, sys; p = pathlib.Path.home() / 'www/mikelev.in/.live_message'; p.parent.mkdir(parents=True, exist_ok=True); tmp = p.with_suffix('.tmp'); tmp.write_text(sys.stdin.read(), encoding='utf-8'); tmp.replace(p); print(f'πŸ“‘ Live break-in queued on Honeybot: {p}')' (nix) pipulate $ ``` **4: PROMPT**: Verify the AFTER receipts for the 5 `forest.py` station-break beads, `ytsay` (`common.py --broadcast say --dry-run`), and the live Honeybot deployment. If everything is green and witnessed, we are ready to dismount or test a live break-in on the air. **5: DELIVERABLE**: That's fine. We refine. **Gemini 4 Argon High**: Context verified. **TL;DR**: The AFTER receipts confirm all 5 station-break beads in `forest.py` (`beads : 5`) and the `ytsay` dry-run payload (`MAX\nHEADROOM`). The live `ytsay` call hit a classic OpenSSH remote-shell quoting trap: `ssh` concatenates arguments after the hostname into a single string for the remote `bash -c`, leaving `python3 -c