waitbus

https://github.com/astrogilda/waitbus
Admin
waitbus is a workstation-local, cross-harness status bus that enables agents, CI jobs, tests, and...
Tokens:56,349
Snippets:757
Trust Score:9.2
License:MIT
Update:4 months ago
Show doc for...
Tokens:
Raw
### Install Waitbus CLI and Initialize

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

Install the Waitbus CLI tool using `uv` or `pipx`. Then, run `waitbus init` for a one-time setup of directories, SQLite schema, and scaffold files.

```bash
uv tool install waitbus              # or: pipx install waitbus
waitbus init                         # one-time setup: state dirs, SQLite schema, scaffold files
```

--------------------------------

### Install and Initialize Waitbus

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

Install the waitbus package and perform the initial setup, including creating state directories, setting up the SQLite schema, and scaffolding necessary files. The `install-credentials` command is optional and specifically for GitHub CI integration.

```bash
uv tool install waitbus
waitbus init
waitbus install-credentials github-webhook-secret
```

--------------------------------

### Install waitbus CLI

Source: https://github.com/astrogilda/waitbus/blob/main/benchmarks/BENCHMARKING.md

Installs the waitbus CLI tool. Ensure your project is set up for development.

```bash
pip install -e .
```

--------------------------------

### Install waitbus CLI and Daemon

Source: https://github.com/astrogilda/waitbus/blob/main/docs/demo/README.md

Installs the waitbus daemon and CLI tools using uv. This is the first step to running the demo.

```bash
uv tool install waitbus
```

--------------------------------

### Start Waitbus Services and Daemons

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

Commands to start the core daemons, background timers, and enable units for automatic startup on login. Ensure systemd user services are configured correctly.

```bash
# Start the core daemons
systemctl --user start waitbus-listener.service
systemctl --user start waitbus-broadcast.socket

# Start the background timers
systemctl --user start waitbus-etag-poll.timer
systemctl --user start waitbus-watchdog.timer

# Health check — exits 1 on any missing required config or stopped unit
waitbus doctor

# Watch webhook deliveries arrive in real time
waitbus read-events watch
```

```bash
systemctl --user enable waitbus-broadcast.socket \
    waitbus-etag-poll.timer waitbus-watchdog.timer
# The webhook listener is opt-in; `install-credentials github-webhook-secret` enables it.
```

--------------------------------

### Install waitbus with plugin verification toolchain

Source: https://github.com/astrogilda/waitbus/blob/main/docs/CUSTOM_SOURCES.md

Install the optional extra for the verification toolchain. This enables in-process attestation verification.

```bash
pip install 'waitbus[plugin-verify]'
```

--------------------------------

### Install Agent Recipes

Source: https://github.com/astrogilda/waitbus/blob/main/examples/agent_pydantic_ai/README.md

Install the necessary dependencies for the agent recipes, including `pydantic-ai-slim` and `langchain-core`. This command uses `uv` for package management.

```sh
uv sync --group agent-recipes
# (pip does not support PEP 735 dependency-groups; with pip, install the deps directly:
#   pip install pydantic-ai-slim langchain-core)
```

--------------------------------

### Install waitbus-langgraph

Source: https://github.com/astrogilda/waitbus/blob/main/adapters/waitbus-langgraph/README.md

Install the waitbus-langgraph package using pip. Requires a running waitbus broadcast daemon.

```bash
pip install waitbus-langgraph
```

--------------------------------

### Clone, Install, and Run Full 24-Hour Soak Test

Source: https://github.com/astrogilda/waitbus/blob/main/docs/SOAK_TEST.md

Set up the waitbus project on a soak host by cloning the repository, installing dependencies, and running a 24-hour soak test with standard suspend cycles. Output is redirected to 'soak.log'.

```bash
git clone https://github.com/astrogilda/waitbus.git
cd waitbus
uv sync --all-extras

# Two-mode with standard suspend cycles:
nohup python -m scripts.soak \
    --duration 24h \
    --rate 5 \
    --inject-suspend-cycles standard \
    --output soak-verdict.json > soak.log 2>&1 &

# Watch progress:
tail -F soak.log

# When done:
jq '.overall_passed, .verdicts' soak-verdict.json
```

--------------------------------

### Install and Verify waitbus Plugin

Source: https://github.com/astrogilda/waitbus/blob/main/docs/CUSTOM_SOURCES.md

Install a waitbus plugin using pip and restart the waitbus daemons. Use `waitbus source list`, `waitbus source show`, and `waitbus source verify` to confirm registration and publisher binding.

```sh
pip install waitbus-circleci
# Restart waitbus daemons so the entry-point walk picks up the plugin.
systemctl --user restart waitbus-listener waitbus-broadcast

# Verify the plugin registered and is bound to its publisher:
waitbus source list
waitbus source show circleci
waitbus source verify circleci
```

--------------------------------

### Install waitbus-pydantic-ai

Source: https://github.com/astrogilda/waitbus/blob/main/adapters/waitbus-pydantic-ai/README.md

Install the package using pip. Requires a running waitbus broadcast daemon.

```bash
pip install waitbus-pydantic-ai
```

--------------------------------

### Install waitbus Package

Source: https://github.com/astrogilda/waitbus/blob/main/SKILL.md

Install the waitbus package using pip, uv tool, or pipx. Ensure your Python environment is set up correctly.

```bash
pip install waitbus
```

```bash
uv tool install waitbus
```

```bash
pipx install waitbus
```

--------------------------------

### Install waitbus Custom Source Plugin

Source: https://github.com/astrogilda/waitbus/blob/main/examples/custom_source_plugin/README.md

Install the custom plugin into your Python environment. This can be done directly from a local path or from a published wheel.

```sh
pip install /path/to/examples/custom_source_plugin
```

```sh
pip install waitbus-circleci
```

--------------------------------

### Install System Services

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

Install system services for Waitbus. On Linux, this involves copying and enabling 8 systemd-user units. On macOS, it involves copying and bootstrapping 4 LaunchAgent plists.

```bash
waitbus install-systemd              # Linux: copy + enable the 8 systemd-user units
waitbus install-launchd              # macOS: copy + bootstrap the 4 LaunchAgent plists
```

--------------------------------

### Initialize waitbus and Systemd Units

Source: https://github.com/astrogilda/waitbus/blob/main/SKILL.md

Run these commands once to initialize waitbus, install GitHub webhook credentials if needed, and set up systemd user units for background services.

```bash
waitbus init
```

```bash
waitbus install-credentials github-webhook-secret
```

```bash
waitbus install-systemd
```

--------------------------------

### Install waitbus Claude Code Plugin

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

Use this command to install the waitbus plugin for Claude Code. Alternatively, use the marketplace entry once published.

```bash
claude --plugin-dir ~/.local/share/waitbus/plugin
```

--------------------------------

### Reproduce waitbus demo GIF

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

This command reproduces the waitbus cross-harness demo GIF shown in the documentation. Ensure you have make installed.

```bash
make hero
```

--------------------------------

### Run waitbus demo

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

This command runs a gentler, single-agent demo of waitbus, suitable for a quick start.

```bash
waitbus demo
```

--------------------------------

### Run Hero Swarm Demo

Source: https://github.com/astrogilda/waitbus/blob/main/examples/hero_swarm/README.md

Execute the hero swarm demo from the repository root. Ensure the 'agent-recipes' group is installed, as it provides the necessary Pydantic AI and LangGraph frameworks.

```bash
uv run --group agent-recipes python -m examples.hero_swarm
```

--------------------------------

### Basic StateGraph Usage with waitbus-langgraph

Source: https://github.com/astrogilda/waitbus/blob/main/adapters/waitbus-langgraph/README.md

Demonstrates how to build a StateGraph using waitbus-langgraph nodes. This example sets up a node to wait for a specific bus event and then routes based on event delivery or timeout.

```python
from langgraph.graph import END, START, StateGraph
from waitbus_langgraph import event_router, wait_node

builder = StateGraph(MyState)
builder.add_node("wait_on_bus", wait_node('fields.conclusion="failure"', source="github", timeout=600.0))
builder.add_node("triage", my_triage_node)
builder.add_edge(START, "wait_on_bus")
builder.add_conditional_edges("wait_on_bus", event_router(on_event="triage", on_timeout=END))
builder.add_edge("triage", END)
graph = builder.compile()
```

--------------------------------

### Install Optional Credentials

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

Optionally install HMAC secrets for securing the GitHub webhook listener and Alertmanager. Re-running these commands rotates the existing credential.

```bash
waitbus install-credentials github-webhook-secret   # optional: HMAC for the GitHub webhook listener (opt-in)
waitbus install-credentials alertmanager-hmac       # optional: alertmanager / watchdog HMAC
```

--------------------------------

### Check CLI versions

Source: https://github.com/astrogilda/waitbus/blob/main/benchmarks/BENCHMARKING.md

Verifies that the Claude and Gemini CLIs are installed and accessible. Sign-in may be required.

```bash
claude --version
```

```bash
gemini --version
```

--------------------------------

### Custom Source Plugin Directory Structure

Source: https://github.com/astrogilda/waitbus/blob/main/examples/custom_source_plugin/README.md

Example directory structure for a custom Waitbus source plugin written in Python.

```text
my-waitbus-plugin/
    pyproject.toml
    src/
        my_waitbus_plugin/
            __init__.py
            plugin.py
```

--------------------------------

### Pre-deploy Gate: Verify Attestation and Pin

Source: https://github.com/astrogilda/waitbus/blob/main/docs/EXIT_CODES.md

Implement this script as a pre-deployment gate to fail closed if the live attestation disagrees with the recorded pin, or if the necessary tooling is not installed. It checks for successful verification, missing attestation/tooling, missing pin, or publisher mismatch.

```bash
# Pre-deploy gate: fail closed if the live attestation disagrees
# with the recorded pin, OR if the live extra is not installed.
waitbus allowlist verify circleci
case $? in
    0)  echo "circleci publisher binding is intact" ;; 
    65|78) echo "FAIL: cannot verify (no attestation OR no tooling)" && exit 1 ;; 
    66) echo "FAIL: pin missing for circleci" && exit 1 ;; 
    76) echo "FAIL: publisher mismatch -- inspect via `waitbus allowlist verify` output" && exit 1 ;; 
esac
```

--------------------------------

### Install Hint for Match-CEL Evaluator

Source: https://github.com/astrogilda/waitbus/blob/main/docs/CONSUMER_API.md

This error message indicates that the 'cel' extra is required to use the --match-cel flag. Install it using pip.

```bash
to use --match-cel, install waitbus[cel]
```

--------------------------------

### CPU Pinning and Execution

Source: https://github.com/astrogilda/waitbus/blob/main/benchmarks/BENCHMARKING.md

Example of pinning a process to specific CPU cores and running a Python module. Ensure the specified cores are isolated for accurate results.

```bash
taskset -c 2,3 uv run python -m benchmarks.bench_ttfae_pytest ...
```

--------------------------------

### Start waitbus Broadcast Daemon

Source: https://github.com/astrogilda/waitbus/blob/main/docs/demo/README.md

Starts the waitbus broadcast daemon in the background. This daemon is responsible for broadcasting events.

```bash
waitbus broadcast serve &
```

--------------------------------

### Configure Kernel for CPU Isolation

Source: https://github.com/astrogilda/waitbus/blob/main/benchmarks/BENCHMARKING.md

Example of kernel command-line parameters for advanced CPU isolation. These settings help dedicate cores to specific tasks.

```bash
Add `isolcpus=2,3 nohz_full=2,3 rcu_nocbs=2,3` to GRUB_CMDLINE_LINUX.
```

--------------------------------

### Install GitHub Webhook Secret Credentials

Source: https://github.com/astrogilda/waitbus/blob/main/SKILL.md

Rotate or install GitHub webhook secret credentials. This is often required after a signature rejection.

```bash
waitbus install-credentials github-webhook-secret
```

--------------------------------

### Platform-Specific Installations for Waitbus

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

Install waitbus systemd units on Linux or LaunchAgents on macOS. These commands are platform-specific and will error if run on the incorrect operating system.

```bash
waitbus install-systemd
```

```bash
waitbus install-launchd
```

--------------------------------

### Configure waitbus Plugin Autoloading

Source: https://github.com/astrogilda/waitbus/blob/main/examples/custom_source_plugin/README.md

Configure waitbus to discover and load installed plugins. The `autoload` setting in `config.toml` controls this behavior.

```toml
[plugins]
autoload = true   # default — all installed waitbus.sources.v1 plugins are loaded
# allow = ["circleci"]   # if autoload=false, explicitly allow specific plugins
# deny = []              # always-applied blocklist
```

--------------------------------

### Verify Release Artifacts with GH Attestation

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

Use this command to verify the integrity of release artifacts. It requires the `gh` CLI and Python to be installed.

```bash
gh attestation verify $(python -c "import waitbus; print(waitbus.__file__)") \
  --repo astrogilda/waitbus
```

--------------------------------

### Start Waitbus Broadcast Socket Unit

Source: https://github.com/astrogilda/waitbus/blob/main/SKILL.md

Start the systemd user socket unit for the Waitbus broadcast daemon. The service unit is pulled in on first subscriber connect.

```bash
systemctl --user start waitbus-broadcast.socket
```

--------------------------------

### Agent Messaging: Request/Reply Example

Source: https://github.com/astrogilda/waitbus/blob/main/docs/AGENT_MESSAGING.md

Demonstrates a request/reply interaction between two agents. Agent B acts as a responder, waiting for messages and sending replies. Agent A acts as a requester, sending a message and waiting for a correlated response. Use `timeout` to prevent indefinite blocking.

```python
from waitbus import request, respond, wait_for

# --- agent_b (responder), e.g. in its own process / thread ---
msg = wait_for(to="agent_b", source="agent", timeout=5.0)
if msg is not None:
    respond(msg, '{"answer": 42}')          # sender defaults to msg's recipient

# --- agent_a (requester) ---
reply = request("agent_b", '{"ask": "meaning"}', sender="agent_a", timeout=5.0)
if reply is None:
    ...                                      # timed out; nobody answered
else:
    assert reply.fields["msg_from"] == "agent_b"
    assert reply.fields["msg_to"] == "agent_a"
```

--------------------------------

### macOS LaunchAgent Plists

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

List of LaunchAgent plists installed by `waitbus install-launchd` on macOS systems, located in `~/Library/LaunchAgents/`.

```text
dev.waitbus.listener.plist     dev.waitbus.broadcast.plist
dev.waitbus.etag-poll.plist    dev.waitbus.watchdog.plist
```

--------------------------------

### Waitbus Operator-Facing CLI Commands

Source: https://github.com/astrogilda/waitbus/blob/main/docs/ARCHITECTURE.md

Overview of the umbrella CLI commands for managing and operating the waitbus daemon, including initialization, installation, credential management, and status checks.

```text
  waitbus                 typer dispatch:
                              init             bootstrap state dirs + schema
                              install-systemd  copy + enable units
                              install-launchd  copy launchd plists (macOS)
                              install-credentials  stage a secret into the 0600 secrets.json
                              doctor           validate the live install end-to-end
                              status           operational dashboard
                              verify-plugin    validate .claude-plugin/plugin.json
```

--------------------------------

### Forward GitHub Webhooks with gh CLI

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

Use this command to set up an ephemeral GitHub webhook and tunnel deliveries to a local listener. Ensure the secret matches the one used for installing credentials.

```bash
gh webhook forward \
    --repo=<owner>/<repo> \
    --events=workflow_run,workflow_job \
    --url=http://127.0.0.1:9000/webhook \
    --secret="<the same plaintext value you piped to install-credentials>"
```

--------------------------------

### Tap into waitbus Events

Source: https://github.com/astrogilda/waitbus/blob/main/docs/demo/README.md

Starts a listener to observe all events broadcast by the waitbus daemon. Useful for debugging and understanding event flow.

```bash
waitbus broadcast tap &
```

--------------------------------

### Pin install hashes for dependencies

Source: https://github.com/astrogilda/waitbus/blob/main/docs/CUSTOM_SOURCES.md

Use pip with `--require-hashes` to ensure that a published-but-tampered wheel cannot replace an attested one between operator audits. This is crucial for supply-chain security.

```bash
pip install --require-hashes -r requirements.txt
```

--------------------------------

### Manually pin a waitbus plugin publisher identity

Source: https://github.com/astrogilda/waitbus/blob/main/docs/CUSTOM_SOURCES.md

Workaround for the trust-model gap when installing plugins directly from PyPI. Manually add a publisher to the allowlist after a one-shot verification.

```bash
waitbus allowlist add <name> <publisher-identity>
```

--------------------------------

### Restart waitbus Daemon

Source: https://github.com/astrogilda/waitbus/blob/main/examples/custom_source_plugin/README.md

After installing or configuring plugins, restart the waitbus daemon for changes to take effect. This can be done via systemd or by running the broadcaster directly.

```sh
systemctl --user restart waitbus-broadcast.service
```

```sh
waitbus serve --all
```

--------------------------------

### Linux Systemd Units

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

List of systemd-user units installed by `waitbus install-systemd` on Linux systems, located in `~/.config/systemd/user/`.

```text
waitbus-listener.service    waitbus-broadcast.socket
waitbus-broadcast.service   waitbus-etag-poll.service
waitbus-etag-poll.timer     waitbus-watchdog.service
waitbus-watchdog.timer      waitbus-forward@.service
```

--------------------------------

### Run Hero Swarm E2E Test

Source: https://github.com/astrogilda/waitbus/blob/main/examples/hero_swarm/README.md

Execute the end-to-end proof test for the hero swarm demo. This test runs the full real-process topology and asserts observable framework reactions to failure events. It is marked 'slow' and skips on non-Linux systems or if Pydantic AI or LangGraph are not installed.

```bash
uv run python -m pytest tests/test_hero_swarm_e2e.py -q
```

--------------------------------

### Restart Waitbus Daemons to Pick Up New Plugins

Source: https://github.com/astrogilda/waitbus/blob/main/docs/CUSTOM_SOURCES.md

Use this command sequence to restart the Waitbus broadcast and listener daemons after installing or updating a plugin. This ensures the daemons re-discover and load the new plugin.

```bash
pip install --upgrade <new-or-changed-plugin>
systemctl --user restart waitbus-broadcast waitbus-listener
# or, on macOS:
launchctl kickstart -k gui/$UID/dev.waitbus.broadcast
launchctl kickstart -k gui/$UID/dev.waitbus.listener
```

--------------------------------

### Install Credentials with Waitbus CLI

Source: https://github.com/astrogilda/waitbus/blob/main/SECURITY.md

Use this command to stage HMAC secrets. Secrets are read from a file or stdin to prevent shell history leakage. The file is written atomically to ensure data integrity.

```bash
waitbus install-credentials <name> [--file PATH]   # or value on stdin
```

--------------------------------

### Trigger Docker Event

Source: https://github.com/astrogilda/waitbus/blob/main/docs/demo/README.md

Starts Docker containers using docker compose, which triggers a docker event. The event is generated when containers exit.

```bash
docker compose up
```

--------------------------------

### Broadcast Daemon Subscribe Frame

Source: https://github.com/astrogilda/waitbus/blob/main/SKILL.md

Example JSON payload for subscribing to the broadcast daemon. Filters and event types can be specified.

```json
{"proto": 1,
 "filters": ["owner/repo", "*", "owner/*"],
 "event_types": ["workflow_run", "workflow_job",
                 "prometheus_alert", "prometheus_watchdog"],
 "since": "01HZ...26chars",
 "envelope": "diffs"}
```

--------------------------------

### Wait for Filesystem Change

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

Block script execution until a change is detected in a watched filesystem path. This example uses the local source and has a timeout of 5 minutes. It matches events where the event type is 'fs_change'.

```bash
waitbus wait --source fs --match 'fields.event_type="fs_change"' --timeout 5m
```

--------------------------------

### Invoke PR Monitor Manually

Source: https://github.com/astrogilda/waitbus/blob/main/SKILL.md

Example of manually invoking the pr-monitor tick command for specific PRs, as it's designed for session-scoped, direct invocation.

```bash
waitbus pr-monitor tick --pr 7
```

--------------------------------

### Run Waitbus Demo

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

Execute the Waitbus demo command to allocate a temporary state directory, boot the broadcast daemon, and run a two-phase demonstration of its waiting capabilities.

```bash
uvx waitbus demo          # one command, no install needed
```

--------------------------------

### Subscribe Frame (Client to Server)

Source: https://github.com/astrogilda/waitbus/blob/main/docs/ARCHITECTURE.md

This JSON structure is sent by a client to subscribe to events. It specifies the protocol version, filtering criteria, event types, and an optional starting point for replay.

```json
{
  "proto": 1,
  "filters": ["owner/repo", "owner/*", "*"],
  "event_types": ["workflow_run", "workflow_job",
                  "prometheus_alert", "prometheus_watchdog"],
  "since": "01HZ...26chars"
}
```

--------------------------------

### Run All Adapter Tests

Source: https://github.com/astrogilda/waitbus/blob/main/adapters/README.md

A script to execute all adapter test suites from the repository root.

```bash
scripts/run_adapter_tests.sh
```

--------------------------------

### Wait for Docker Container Exit

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

Block script execution until a Docker container exits. This example uses the local source and has a timeout of 30 seconds. It matches events where the event type is 'docker_container'.

```bash
waitbus wait --source docker --match 'fields.event_type="docker_container"' --timeout 30s
```

--------------------------------

### Wait for Pytest Session Completion

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

Block script execution until a pytest session finishes. This example uses the local source and has a timeout of 10 minutes. It matches events where the event type is 'pytest_session'.

```bash
waitbus wait --source pytest --match 'fields.event_type="pytest_session"' --timeout 10m
```

--------------------------------

### Stream Claude Code Lifecycle Events (Python)

Source: https://github.com/astrogilda/waitbus/blob/main/docs/emitters/claude-code-hook.md

Subscribe to waitbus events from Claude Code sessions in Python. This example shows how to stream lifecycle events for a specific session and print the message body and delivery ID.

```python
from waitbus import subscribe

for frame in subscribe(source="agent", match='fields.msg_from="claude-code:my-session-id"'):
    print(frame.fields["msg_body"], frame.delivery_id)
```

--------------------------------

### Verify Linger Setting on Headless Server

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

Check if the systemd user manager is active for the waitbus listener after enabling linger. This test should be performed in a fresh SSH session without manually starting the units.

```bash
systemctl --user is-active waitbus-listener
```

--------------------------------

### Quick Liveness Overview

Source: https://github.com/astrogilda/waitbus/blob/main/SKILL.md

Provides a quick overview of the liveness status of Waitbus.

```bash
waitbus status
```

--------------------------------

### Declare Entry-Point in pyproject.toml

Source: https://github.com/astrogilda/waitbus/blob/main/docs/CUSTOM_SOURCES.md

Define the plugin's entry point for Waitbus to discover. The `waitbus.sources.v1` group is used for version 1 sources, and the key must match the canonical source name.

```toml
[project]
name = "waitbus-circleci"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = ["waitbus>=0.5,<0.6"]

[project.entry-points."waitbus.sources.v1"]
circleci = "waitbus_circleci:plugin"
```

--------------------------------

### Waitbus config.toml Plugin Policy

Source: https://github.com/astrogilda/waitbus/blob/main/docs/CUSTOM_SOURCES.md

Configure plugin autoloading, allowed, and denied plugins. Defaults to autoloading enabled with empty allow/deny lists.

```toml
[plugins]
autoload = true          # walk the entry-point group at startup
allow    = []            # if autoload=false, only these names load
deny     = []            # always-applied blocklist

```

--------------------------------

### Declare Plugin Entry-Point in pyproject.toml

Source: https://github.com/astrogilda/waitbus/blob/main/examples/custom_source_plugin/README.md

Configuration for a custom Waitbus source plugin in `pyproject.toml`, specifying dependencies and the entry-point for Waitbus to discover the plugin.

```toml
[project]
name = "my-waitbus-plugin"
dependencies = ["waitbus>=0.5,<0.6"]

[project.entry-points."waitbus.sources.v1"]
my_source = "my_waitbus_plugin:plugin"
```

--------------------------------

### Initialize Waitbus Database

Source: https://github.com/astrogilda/waitbus/blob/main/SKILL.md

Re-run waitbus init if the database is missing. This command is idempotent.

```bash
waitbus init
```

--------------------------------

### Touch a File to Trigger Event

Source: https://github.com/astrogilda/waitbus/blob/main/docs/demo/watched-dir/README.md

Use this command to create a new file in the watched directory, which will trigger a file system event.

```bash
touch watched-dir/$(date +%s)
```

--------------------------------

### Show Plugin Registration Details

Source: https://github.com/astrogilda/waitbus/blob/main/examples/custom_source_plugin/README.md

Inspect the full registration details of a specific custom source plugin.

```sh
waitbus source show circleci
```

--------------------------------

### Dry Run for Database Migrations

Source: https://github.com/astrogilda/waitbus/blob/main/SECURITY.md

Command to preview the SQL and Python code that will be executed during a schema migration. This allows operators to review changes before applying them.

```bash
waitbus db migrate --dry-run
```

--------------------------------

### Export Singleton Plugin Instance

Source: https://github.com/astrogilda/waitbus/blob/main/examples/custom_source_plugin/README.md

Python code for `__init__.py` that exports a singleton instance of the custom source plugin, making it discoverable by Waitbus.

```python
# src/my_waitbus_plugin/__init__.py
from my_waitbus_plugin.plugin import MySourcePlugin

plugin: MySourcePlugin = MySourcePlugin()
__all__ = ["plugin"]
```

--------------------------------

### Build and Invoke LangGraph Agent

Source: https://github.com/astrogilda/waitbus/blob/main/examples/agent_langgraph/README.md

Use this code to build the LangGraph agent, specifying the socket path and timeout. Invoke the graph with initial state and assert the expected reaction.

```python
from examples.agent_langgraph import build_graph

graph = build_graph(socket_path="/path/to/broadcast.sock", timeout=5.0)
final_state = graph.invoke({"event_type": None, "summary": None, "reacted": False})
assert final_state["reacted"] is True
print(final_state["event_type"])  # "docker_container"
print(final_state["summary"])     # the fake model's deterministic reply
```

--------------------------------

### Verify a local wheel artifact's attestation

Source: https://github.com/astrogilda/waitbus/blob/main/docs/CUSTOM_SOURCES.md

Verify the attestation of a locally downloaded wheel artifact using the `pypi-attestations` CLI. This is part of the manual pinning workaround.

```bash
pypi-attestations verify <path-to-wheel-file>
```

--------------------------------

### Configure waitbus MCP Server

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

This JSON configuration is used to set up waitbus as an MCP server. Ensure 'uvx' is in your PATH.

```json
{
  "mcpServers": {
    "waitbus": {
      "command": "uvx",
      "args": ["--from", "waitbus", "waitbus", "mcp", "serve"]
    }
  }
}
```

--------------------------------

### Import Waitbus Request and Respond Functions

Source: https://github.com/astrogilda/waitbus/blob/main/docs/AGENT_MESSAGING.md

Import the necessary `request` and `respond` functions from the waitbus package for agent messaging.

```python
from waitbus import request, respond
```

--------------------------------

### Set OpenAI API Key

Source: https://github.com/astrogilda/waitbus/blob/main/benchmarks/BENCHMARKING.md

Optional: Sets the OpenAI API key environment variable to enable the gpt-4o-mini driver.

```bash
export OPENAI_API_KEY=...
```

--------------------------------

### Waitbus plugins.allowlist.toml Publisher Binding

Source: https://github.com/astrogilda/waitbus/blob/main/docs/CUSTOM_SOURCES.md

Define runtime-learned publisher bindings for plugins. This file is managed by `waitbus allowlist` commands and is designed to be hand-editable and version-controlled.

```toml
[[source]]
name = "circleci"
publisher_kind = "GitHub"
publisher_identity = "astrogilda/waitbus-circleci @ .github/workflows/release.yml"
first_pinned_at = "2026-05-20T09:42:11Z"

```

--------------------------------

### Sync and Run Adapter Tests

Source: https://github.com/astrogilda/waitbus/blob/main/adapters/README.md

Commands to synchronize dependencies and run the pytest suite for the waitbus-pydantic-ai adapter.

```bash
uv sync --directory adapters/waitbus-pydantic-ai --all-groups
uv run --directory adapters/waitbus-pydantic-ai python -m pytest -q
```

--------------------------------

### Build waitbus Docker Image

Source: https://github.com/astrogilda/waitbus/blob/main/packaging/docker/README.md

Builds the waitbus Docker image for multiple platforms using Docker Buildx. Specify the Dockerfile and the desired tag.

```bash
docker buildx build \
  --platform linux/amd64,linux/arm64 \
  -f packaging/docker/Dockerfile \
  -t ghcr.io/astrogilda/waitbus:dev .

```

--------------------------------

### Waitbus Runtime Topology Overview

Source: https://github.com/astrogilda/waitbus/blob/main/docs/ARCHITECTURE.md

Visual representation of the waitbus daemon's components and their interactions, including webhook ingress, database storage, and broadcast socket communication.

```text
GitHub / Alertmanager (HTTPS)
        |
        v
  waitbus listener serve  loopback :9000  (HMAC-SHA256 routes)
        |
        |
        v
  events table              SQLite, WAL mode, single-file DB
        |
        |
        v
  waitbus broadcast serve AF_UNIX SOCK_STREAM hub
        |
        +-- waitbus read-events watch   one line per event, stdout
        +-- waitbus pr-monitor tick     per-PR aggregate, prints transitions
        +-- waitbus mcp serve           MCP client bridge (Claude Code / Desktop)
        +-- future subscribers            same wire protocol
```

--------------------------------

### Verify PEP 740 PyPI Attestations

Source: https://github.com/astrogilda/waitbus/blob/main/SECURITY.md

Verify PEP 740 PyPI attestations for a waitbus wheel file using sigstore-keyless. This command requires the wheel filename, the certificate identity tied to the release workflow, and the OIDC issuer.

```bash
python -m sigstore verify identity waitbus-<version>-py3-none-any.whl \
  --cert-identity https://github.com/astrogilda/waitbus/.github/workflows/release.yml@refs/tags/v<version> \
  --cert-oidc-issuer https://token.actions.githubusercontent.com
```

--------------------------------

### Run All Robustness Tests

Source: https://github.com/astrogilda/waitbus/blob/main/docs/ROBUSTNESS_TESTS.md

Execute all tests in the broadcast robustness suite. Use this to ensure overall stability.

```bash
uv run pytest tests/test_broadcast_robustness.py -v
```

```bash
uv run pytest tests/test_broadcast.py -v
```

```bash
uv run pytest tests/test_broadcast_exception_mapping.py -v
```

```bash
uv run pytest tests/test_subscribe_envelope_hygiene.py -v
```

--------------------------------

### Download a wheel without dependencies

Source: https://github.com/astrogilda/waitbus/blob/main/docs/CUSTOM_SOURCES.md

Fetch a wheel artifact locally without downloading its dependencies. This is useful for manual verification before adding to an allowlist.

```bash
pip download --no-deps <plugin-name>
```

--------------------------------

### Record Hero Swarm Demo with VHS

Source: https://github.com/astrogilda/waitbus/blob/main/examples/hero_swarm/README.md

Record the hero swarm demo using VHS. This command checks for necessary tools like vhs, render font, and tmux before rendering the demo tape into GIF and MP4 formats. Rendering requires 'vhs', 'tmux', and 'ttyd' to be on the PATH.

```bash
make hero
```

```bash
vhs hero.tape
```

--------------------------------

### Go Minimal Subscriber

Source: https://github.com/astrogilda/waitbus/blob/main/docs/snippets/README.md

Connects to the waitbus daemon, subscribes to all events, and prints delivery ID, source, and event type. Requires Go.

```Go
package main

import (
	"bytes"
	"encoding/binary"
	"encoding/json"
	"fmt"
	"io"
	"log"
	"net"
	"os"
)

func main() {
	conn, err := net.Dial("unix", "unix_socket")
	if err != nil {
		log.Fatal("net.Dial: ", err)
	}
	defer conn.Close()

	// Subscribe to all events
	envelope := map[string]interface{}{"proto": 1}
	payload, err := json.Marshal(envelope)
	if err != nil {
		log.Fatal("json.Marshal: ", err)
	}

	// Send length-prefixed payload
	lengthBuf := new(bytes.Buffer)
	err = binary.Write(lengthBuf, binary.BigEndian, uint32(len(payload)))
	if err != nil {
		log.Fatal("binary.Write: ", err)
	}
	_, err = conn.Write(append(lengthBuf.Bytes(), payload...))
	if err != nil {
		log.Fatal("conn.Write: ", err)
	}

	for {
		// Read frame length
		var length uint32
		err := binary.Read(conn, binary.BigEndian, &length)
		if err != nil {
			if err == io.EOF {
				break // Connection closed
			}
			log.Fatal("binary.Read length: ", err)
		}

		// Read frame payload
		payload := make([]byte, length)
		_, err = io.ReadFull(conn, payload)
		if err != nil {
			if err == io.EOF {
				break // Connection closed
			}
			log.Fatal("io.ReadFull payload: ", err)
		}

		var frame map[string]interface{}
		err = json.Unmarshal(payload, &frame)
		if err != nil {
			log.Printf("json.Unmarshal: %v", err)
			continue
		}

		kind, ok := frame["kind"].(string)
		if !ok {
			continue
		}

		switch kind {
		case "event":
			deliveryID, _ := frame["delivery_id"].(string)
			fields, ok := frame["fields"].(map[string]interface{})
			source, _ := fields["source"].(string)
			eventType, _ := frame["event_type"].(string)
			fmt.Printf("%s %s %s\n", deliveryID, source, eventType)
		case "subscribe_rejected":
			reason, _ := frame["reason"].(string)
			fmt.Fprintf(os.Stderr, "Subscription rejected: %s\n", reason)
			return
		case "subscribe_ack":
			fmt.Println("Subscription acknowledged.")
		// Ignore other frame kinds
		}
	}
}

```

--------------------------------

### Bash Wait for Any Source

Source: https://github.com/astrogilda/waitbus/blob/main/docs/snippets/README.md

Wraps `waitbus wait --match` three times in parallel to demonstrate cross-source predicate matching. Exits on the first match.

```Bash
#!/bin/bash

set -euo pipefail

# Default timeout in seconds
TIMEOUT=${1:-5}

# Function to wait for a specific source with a timeout
wait_for_source() {
    local source=$1
    echo "Waiting for source: $source"
    waitbus wait --match "source=$source" --timeout $TIMEOUT
    echo "Source $source matched."
}

# Run waits in parallel and exit on the first match
( wait_for_source "github.com" & )
( wait_for_source "gitlab.com" & )
( wait_for_source "bitbucket.org" & )

# Wait for any of the background jobs to finish
wait -n

# Kill any remaining background jobs
pkill -f "waitbus wait --match"

echo "First match received."

```

--------------------------------

### Waitbus Wait CLI Invocation Shapes

Source: https://github.com/astrogilda/waitbus/blob/main/docs/CONSUMER_API.md

Demonstrates various ways to invoke the waitbus wait command with different predicate matching options. Use --sha for GitHub CI, --match for key-value pairs, --cond for named conditions, and --match-cel/--match-jmespath for advanced expressions.

```bash
waitbus wait --sha <SHA> [--repo owner/repo] [--timeout 30s]   # GitHub CI (sugar)
waitbus wait --source <s> --match <dotted.key>=<json_literal>... [--timeout 30s]
waitbus wait --cond <named-condition>           [--timeout 30s]
waitbus wait --match-cel '<expr>'               # extras-gated; see below
waitbus wait --match-jmespath '<expr>'          # extras-gated; see below
```

--------------------------------

### Run Local Soak Test

Source: https://github.com/astrogilda/waitbus/blob/main/docs/SOAK_TEST.md

Execute a local soak test for 60 seconds with a rate of 5, saving output to /tmp/soak.json. Useful for developing the harness.

```bash
python -m scripts.soak --duration 60s --rate 5 --output /tmp/soak.json
```

--------------------------------

### Publish Plugin with PEP 740 Attestations

Source: https://github.com/astrogilda/waitbus/blob/main/docs/CUSTOM_SOURCES.md

Configure a GitHub Actions workflow to publish attested wheels using PyPI Trusted Publishing. The `pypa/gh-action-pypi-publish` action automatically generates attestations.

```yaml
# .github/workflows/release.yml
permissions:
  id-token: write
  attestations: write
  contents: read

jobs:
  publish:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.13" }
      - run: python -m pip install build && python -m build
      - uses: pypa/gh-action-pypi-publish@release/v1
```

--------------------------------

### Add Repo for Polling Coverage

Source: https://github.com/astrogilda/waitbus/blob/main/SKILL.md

Add a repository to watched_repos.txt for polling-only coverage if webhook issues arise.

```bash
add the repo to watched_repos.txt under the state directory
```

--------------------------------

### Run Hetzner Pre-24h Soak Test with Fault Injection

Source: https://github.com/astrogilda/waitbus/blob/main/docs/SOAK_TEST.md

Perform a 30-minute soak test on Hetzner CCX23 hardware with standard fault injection. This validates daemon control flow under real-world conditions before a full 24h run.

```bash
scripts/run_soak_on_hetzner.sh \
    --duration 30m \
    --inject-fault-scenarios standard \
    --server-type ccx23
```

--------------------------------

### Query CI Events with waitbus

Source: https://github.com/astrogilda/waitbus/blob/main/SKILL.md

Use these commands to query CI event history. The `--text` flag provides human-readable output, while `--json` is for programmatic parsing. Omit `--owner` and `--repo` to query the auto-detected repository.

```bash
waitbus read-events list --text
```

```bash
waitbus read-events list --owner OWNER --repo REPO --last-n 5 --text
```

```bash
waitbus read-events list --json
```

--------------------------------

### Provision Hetzner Server for Soak Test

Source: https://github.com/astrogilda/waitbus/blob/main/docs/SOAK_TEST.md

Create a Hetzner CCX23 server instance named 'waitbus-soak' using the hcloud CLI. This server will be used for the full 24-hour soak test.

```bash
hcloud server create --type ccx23 --image ubuntu-24.04 --name waitbus-soak
```

--------------------------------

### Verify Plugin Registration

Source: https://github.com/astrogilda/waitbus/blob/main/examples/custom_source_plugin/README.md

Confirm that a custom source plugin has been registered correctly by listing available sources.

```sh
waitbus source list
```

--------------------------------

### Python Wrapper for SQL Migrations

Source: https://github.com/astrogilda/waitbus/blob/main/SECURITY.md

A minimal Python wrapper used to execute SQL migration scripts. This pattern is preferred for schema changes that do not require procedural logic.

```python
conn.executescript(Path(__file__).with_suffix(".sql").read_text())
```

--------------------------------

### Verify a waitbus source plugin's attestation

Source: https://github.com/astrogilda/waitbus/blob/main/docs/CUSTOM_SOURCES.md

Verify the attestation of a third-party plugin using the waitbus CLI. This command wraps `pypi_attestations.Attestation.verify`.

```bash
waitbus source verify <name>
```

--------------------------------

### Run a command on waitbus event

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

This command executes a specified command when a waitbus event matching the given predicate occurs. The event data is passed as environment variables and a file.

```bash
waitbus on <predicate> -- <command>
```

--------------------------------

### Implement SourcePlugin Protocol

Source: https://github.com/astrogilda/waitbus/blob/main/docs/CUSTOM_SOURCES.md

Implement the `SourcePlugin` protocol by defining a class that returns a `SourceSpec`. The `SourceSpec` includes the source name, allowed event types, an optional payload schema, and the API version.

```python
# waitbus_circleci/__init__.py
from waitbus.sources import SourceSpec

class CircleCISource:
    def spec(self) -> SourceSpec:
        return SourceSpec(
            name="circleci",
            event_types=("pipeline_finished",),
            payload_schema=None,
            api_version=1,
        )

plugin = CircleCISource()
```

--------------------------------

### CI Gate: Verify Source with Exit Code Handling

Source: https://github.com/astrogilda/waitbus/blob/main/docs/EXIT_CODES.md

Use this script in CI to ensure a clean source verification. It treats specific non-fatal exit codes (missing attestation, unknown source) as alerts while failing on critical errors.

```bash
# CI gate: succeed only on a clean verify; treat missing-attestation
# (65) and unknown-source (66) as alerts but not hard failures.
if ! waitbus source verify circleci; then
    rc=$?
    case "$rc" in
        65) echo "WARN: circleci wheel has no attestation; pre-vet manually" ;; 
        66) echo "ERROR: circleci plugin not installed; reinstall" && exit 1 ;; 
        76) echo "ALERT: circleci attestation FAILED verification; suspect supply-chain compromise" && exit 1 ;; 
        78) echo "ERROR: waitbus[plugin-verify] missing; install the extra" && exit 1 ;; 
        *)  echo "ERROR: unexpected exit code $rc" && exit 1 ;; 
    esac
fi
```

--------------------------------

### Reproduce Subscribe-Rejected Exception Class Drift

Source: https://github.com/astrogilda/waitbus/blob/main/docs/ROBUSTNESS_TESTS.md

Run this command to reproduce the subscribe-rejected exception class drift issue. This test verifies correct exception mapping for broadcast rejections.

```bash
uv run pytest tests/test_broadcast_exception_mapping.py -v
```

--------------------------------

### Reproduce Replay Wrong-Key-Pop Subscriber Leak

Source: https://github.com/astrogilda/waitbus/blob/main/docs/ROBUSTNESS_TESTS.md

Execute this command to reproduce issues related to replay wrong-key-pop subscriber leaks and SUBSCRIBER_COUNT double-decrement. This covers lag eviction and fan-out close scenarios.

```bash
uv run pytest tests/test_broadcast.py::test_replay_lag_eviction_closes_subscriber_with_real_fd tests/test_broadcast_robustness.py -v
```

--------------------------------

### View live waitbus events

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

This command displays a live view of events occurring on the waitbus. It's useful for monitoring bus activity.

```bash
waitbus top
```

--------------------------------

### Agent Request/Reply Messaging in Python

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

Implement request/reply patterns between agents using `request` and `respond`. The responder uses `wait_for` to receive messages and `respond` to send a reply. The requester uses `request` to send a message and wait for a response.

```python
from waitbus import request, respond, wait_for

# responder (agent_b), in its own process
msg = wait_for(to="agent_b", source="agent", timeout=5.0)
if msg is not None:
    respond(msg, '{"answer": 42}')             # sender defaults to msg's recipient

# requester (agent_a)
reply = request("agent_b", '{"ask": "meaning"}', sender="agent_a", timeout=5.0)
# reply is the answer EventFrame, or None on timeout.
```

--------------------------------

### Verify SLSA Build Provenance

Source: https://github.com/astrogilda/waitbus/blob/main/SECURITY.md

Use this command to verify the SLSA build provenance for a release artifact. It requires the artifact name and specifies the repository and the signer workflow for enhanced security.

```bash
gh attestation verify <artifact> \
  --repo astrogilda/waitbus \
  --signer-workflow slsa-framework/slsa-github-generator/.github/workflows/generator_generic_slsa3.yml@v2.1.0
```

--------------------------------

### Implement SourcePlugin Protocol in Python

Source: https://github.com/astrogilda/waitbus/blob/main/examples/custom_source_plugin/README.md

Python code implementing the `SourcePlugin` protocol for a custom Waitbus source. It defines the plugin's specification and the fetch method for retrieving events.

```python
# src/my_waitbus_plugin/plugin.py
from __future__ import annotations
from collections.abc import Iterator
from typing import TYPE_CHECKING
from waitbus.sources._protocol import SOURCE_PLUGIN_API_VERSION, SourceSpec

if TYPE_CHECKING:
    from waitbus._types import EventInsert

class MySourcePlugin:
    def spec(self) -> SourceSpec:
        return SourceSpec(
            name="my_source",
            event_types=("build_finished",),
            payload_schema=None,
            api_version=SOURCE_PLUGIN_API_VERSION,
        )

    def fetch(self, *args: object, **kwargs: object) -> Iterator[EventInsert] | None:
        # Poll your CI system here and yield EventInsert objects,
        # or emit via waitbus's public emit API and return None.
        ...
```

--------------------------------

### Rust Minimal Subscriber

Source: https://github.com/astrogilda/waitbus/blob/main/docs/snippets/README.md

Connects to the waitbus daemon, subscribes to all events, and prints delivery ID, source, and event type. Requires Rust.

```Rust
use std::io::{self, Read, Write};
use std::net::Shutdown;
use std::os::unix::net::UnixStream;

fn main() -> io::Result<()>
{
    let stream = UnixStream::connect("unix_socket")?;

    // Subscribe to all events
    let envelope = serde_json::json!({{"proto": 1}});
    let payload = serde_json::to_string(&envelope)?.as_bytes().to_vec();
    let length = payload.len() as u32;
    let mut buffer = Vec::new();
    buffer.extend_from_slice(&length.to_be_bytes());
    buffer.extend_from_slice(&payload);
    stream.write_all(&buffer)?;

    let mut reader = io::BufReader::new(stream);
    let mut length_buffer = [0u8; 4];

    loop {
        // Read frame length
        match reader.read_exact(&mut length_buffer) {
            Ok(_) => {},
            Err(ref e) if e.kind() == io::ErrorKind::UnexpectedEof => break, // Connection closed
            Err(e) => return Err(e),
        }
        let length = u32::from_be_bytes(length_buffer);

        // Read frame payload
        let mut payload_buffer = vec![0u8; length as usize];
        match reader.read_exact(&mut payload_buffer) {
            Ok(_) => {},
            Err(ref e) if e.kind() == io::ErrorKind::UnexpectedEof => break, // Connection closed
            Err(e) => return Err(e),
        }

        let frame: serde_json::Value = serde_json::from_slice(&payload_buffer)?;

        if let Some(kind) = frame.get("kind").and_then(|k| k.as_str()) {
            if kind == "event" {
                let delivery_id = frame.get("delivery_id").and_then(|id| id.as_str()).unwrap_or("N/A");
                let source = frame.get("fields").and_then(|f| f.get("source")).and_then(|s| s.as_str()).unwrap_or("N/A");
                let event_type = frame.get("event_type").and_then(|et| et.as_str()).unwrap_or("N/A");
                println!("{delivery_id} {source} {event_type}");
            } else if kind == "subscribe_rejected" {
                println!("Subscription rejected: {}", frame.get("reason").unwrap_or(&serde_json::Value::Null));
                break;
            } else if kind == "subscribe_ack" {
                println!("Subscription acknowledged.");
            }
            // Ignore other frame kinds
        }
    }

    reader.into_inner().shutdown(Shutdown::Both)?; // Close the stream
    Ok(())
}

```

--------------------------------

### Read Events with Waitbus

Source: https://github.com/astrogilda/waitbus/blob/main/README.md

Use 'waitbus read-events watch' for live event tailing. For historical data, use 'waitbus read-events list' with time filters and JSON output.

```bash
waitbus read-events watch
```

```bash
waitbus read-events list --since 1h --json
```

--------------------------------

### Python Minimal Subscriber

Source: https://github.com/astrogilda/waitbus/blob/main/docs/snippets/README.md

Connects to the waitbus daemon, subscribes to all events, and prints delivery ID, source, and event type. Requires Python.

```Python
import asyncio
import json
import socket
import struct

from typing import Any


def parse_frame(sock: socket.socket) -> Any:
    """Read a length-prefixed frame from the socket."""
    try:
        length_bytes = sock.recv(4)
        if not length_bytes:
            return None
        length = struct.unpack(">I", length_bytes)[0]
        payload = sock.recv(length)
        return json.loads(payload)
    except ConnectionResetError:
        return None


def main():
    """Connect, subscribe, and print events."""
    sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    sock.connect("unix_socket")

    # Subscribe to all events
    envelope = {"proto": 1}
    payload = json.dumps(envelope).encode("utf-8")
    length = len(payload)
    sock.sendall(struct.pack(">I", length) + payload)

    while True:
        frame = parse_frame(sock)
        if frame is None:
            break

        kind = frame.get("kind")
        if kind == "event":
            delivery_id = frame.get("delivery_id")
            source = frame.get("fields", {}).get("source")
            event_type = frame.get("event_type")
            print(f"{delivery_id} {source} {event_type}")
        elif kind == "subscribe_rejected":
            print(f"Subscription rejected: {frame.get('reason')}")
            break
        elif kind == "subscribe_ack":
            print("Subscription acknowledged.")
        # Ignore other frame kinds (e.g., daemon_heartbeat, truncated)

    sock.close()


if __name__ == "__main__":
    # Replace "unix_socket" with the actual path to the waitbus socket
    # On Linux: os.environ.get("XDG_RUNTIME_DIR", "/tmp") + "/waitbus/broadcast.sock"
    # On macOS: os.path.expanduser("~/Library/Application Support/waitbus/broadcast.sock")
    main()

```

--------------------------------

### Read Events with Filters

Source: https://github.com/astrogilda/waitbus/blob/main/SKILL.md

Prints one stdout line per matching event, suitable for background monitoring loops. Filters default to the current git checkout's owner/repo.

```bash
waitbus read-events watch
```

```bash
waitbus read-events watch --all-events
```

--------------------------------

### Check Webhook Registration

Source: https://github.com/astrogilda/waitbus/blob/main/SKILL.md

Verify if a webhook is registered for a specific repository and its target URL.

```bash
gh api repos/<owner>/<repo>/hooks
```

--------------------------------

### Webhook Received Rate by Path

Source: https://github.com/astrogilda/waitbus/blob/main/docs/monitoring/waitbus-grafana.md

Monitors the rate of deliveries arriving at various webhook paths per second. A drop to zero during expected activity indicates GitHub is not reaching the listener.

```PromQL
rate(waitbus_webhook_received_total[5m])
```

--------------------------------

### MCP Info Output

Source: https://github.com/astrogilda/waitbus/blob/main/docs/ARCHITECTURE.md

This JSON output is generated by the `waitbus mcp info` command. It displays the service name, version, current protocol version, and a list of supported protocol versions, useful for verifying alignment with the MCP SDK.

```json
{
  "name": "waitbus",
  "version": "0.1.0",
  "protocolVersion": "2025-11-25",
  "supportedProtocolVersions": ["2024-11-05", "2025-03-26", "2025-06-18", "2025-11-25"]
}
```

--------------------------------

### Secret Staging Conventions

Source: https://github.com/astrogilda/waitbus/blob/main/docs/ARCHITECTURE.md

Defines the conventions for staging secrets used by the listener. Secrets are loaded at daemon startup and are not refreshed at runtime.

```text
github-webhook-secret   listener  load-bearing
alertmanager-hmac       listener  optional
```
Tokens:
Raw