# EVENTS — the oss-oo event bus, contract v1 (FROZEN)

The forge's nervous system: org repo webhooks flow through the
`webhook2mqtt` bridge onto an MQTT broker (mosquitto), and agents
**subscribe instead of poll** — `tell me about PRs and CI in OSS-OO/*`
becomes one wildcard subscription that reacts within seconds.

- Services: `compose.events.yml` (mosquitto + webhook2mqtt, code in
  `webhook2mqtt/`)
- Design context: `docs/AGENT_PLATFORM_PHASE1.md` §2.4 / P1.c (owner GO
  D8 + D9, 2026-09-24)
- Operations: `docs/RUNBOOK.md` "Event bus"

**This document is a freeze.** Topic layout, envelope schema, event
mapping and listener semantics below are the v1 contract — consumers
may rely on them. Breaking changes ship as a **v2 topic namespace**
(`oss-oo/v2/...`), never as a silent change in place. Additive entries
(new event kinds for new repos, new action slugs) are not breaking.

## Topic layout v1

```
oss-oo/{org}/{repo}/{kind}          retained — last event of that kind
oss-oo/{org}/{repo}/{kind}/{action} live    — never retained
```

- `{org}`/`{repo}` preserve the forgejo spelling of `full_name`
  (e.g. `OSS-OO/laboratory`); user-owned repos use the owner login as
  `{org}`. Segments match `[A-Za-z0-9_.-]+` — anything else never
  reaches the broker.
- `{kind}` is exactly one of: `push`, `issue`, `pr`, `ci`, `release`,
  `package`.
- `{action}` is the payload's `action` field, lowercased and stripped
  to `[a-z0-9_-]` (e.g. `opened`, `synchronize`, `completed`). Push
  events carry no action — they exist only on the kind topic.

**Kind topics are state** (a fresh subscriber immediately gets the last
event per kind), **action topics are stream** (only live deliveries;
the last-second firehose). The design's canonical wildcard
`oss-oo/OSS-OO/+/pr/+` matches all PR activity across the org.

## Event mapping v1

| Forgejo webhook event                        | kind     | action topic |
|----------------------------------------------|----------|--------------|
| `push`                                       | `push`   | —            |
| `issues`, `issue_comment`                    | `issue`  | yes          |
| `pull_request`, `pull_request_comment`, `pull_request_review`, `pull_request_review_comment` (and future `pull_request_*` variants) | `pr` | yes |
| `workflow_run`, `workflow_job`               | `ci`     | yes          |
| `release`                                    | `release`| yes          |
| `package`                                    | `package`| yes          |

Everything else (wiki, repository, fork, create, delete, …) is ignored
v1 — outside the frozen namespace, silently, with a 200 to the forge so
delivery history stays green.

## Envelope schema v1

Every message body is one JSON object:

```json
{
  "schema": "oss-oo.events/v1",
  "delivery": "357edcf6-…",          // forge delivery id (tracing)
  "event": "pull_request",           // the forgejo event type
  "action": "opened",                // or null (push)
  "org": "OSS-OO",
  "repo": "laboratory",
  "visibility": "public",            // constant on the bus — see the gate
  "received_at": "2026-09-24T14:03:09Z",
  "payload": { …original webhook payload… }
}
```

If the serialized envelope would exceed ~900 kB, `payload` is replaced
by `{"_truncated": true, "payload_bytes": N}` — the event still fires;
fetch details via the REST API.

## The public-only gate (v1 honesty rule)

**Private-repo events never reach the broker.** The bridge publishes
only when `repository.private` is explicitly `false`; private or
unknown visibility is dropped and logged (`skipped_private` counter).
This is enforced at the single point every event must pass
(`webhook2mqtt/bridge.py`) and proven by a **negative test in
`scripts/e2e-local.sh`**: a private-repo issue is created and the bus
must stay silent for its topics while the gate counter rises. Private
events arrive only in a future version with per-agent ACLs derived
from live forge permissions, behind their own owner GO
(P1.c.2 — design doc §2.4).

## Listener contract

- **Endpoint:** `wss://events.oss-oo.io/mqtt` (any path upgrades;
  caddy-fronted TLS, owner decision D9 — plain TCP 8883 is
  deliberately NOT opened in v1). Inside the compose network the
  broker also speaks plain MQTT on 1883 and plain ws on 9001.
- **Protocol:** MQTT 3.1.1 or 5 over WebSocket subprotocol `mqtt`.
- **Auth: required.** Anonymous is denied. v1 has exactly one
  credential — the local bridge user — so public subscription access
  arrives with P1.b (registrar-minted per-agent MQTT credentials with
  read ACLs on `oss-oo/+/#`), not before.
- **QoS 1, persistent sessions** (clean_session=false + stable client
  id = offline catch-up of action topics), **retained** last event per
  kind topic. Broker restarts keep sessions and retained state
  (persistence volume + bridge retained-republish backstop;
  e2e-proven, see "Delivery guarantees, honestly").

## Delivery guarantees, honestly

At-least-once from webhook to broker (forgejo webhook retries + QoS 1
redelivery: duplicates are possible, loss is not). Retained
last-event-per-kind + persistent-session catch-up — **no full replay
log** (owner decision D12: not yet; a replay/event-log service is a
separate decision with real storage cost).

**Retained-state durability (v1 contract):** kind topics survive any
single broker restart through two layers. (1) The broker persists
retained messages to disk at most 1 s after a change
(`autosave_interval 1`). (2) The bridge caches the last retained
envelope per kind topic (in memory, ≤ `EVENTS_RETAINED_TOPICS` topics
by recency) and **re-publishes the cached state after every
reconnect** — a broker that comes back without its persistence (crash
inside the save window, lost volume) is restored from the bridge's
cache when the connection returns. Consequences, stated plainly:

- Reconnects re-deliver the last envelope of each kind topic to
  *currently subscribed* consumers — dedupe by the envelope's
  `delivery` id, exactly as QoS 1 duplicates already require.
- The bridge's cache is warm-memory only: a freshly restarted bridge
  has an empty cache until events flow again (the broker's own
  persistence covers that window).
- The bridge connection self-heals: a keepalive ping (default 30 s,
  `EVENTS_KEEPALIVE_PING`) detects a dead broker while idle, and
  reconnects run on exponential backoff (1 s doubling to 60 s) without
  waiting for the next webhook. Bridge health is observable via
  `/healthz` (`broker`, `reconnects`, `republished`,
  `retained_topics`, queue depth — see RUNBOOK "Event bus").

## Client recipes

```bash
# CLI — from a container on the compose network (mosquitto_sub speaks
# plain MQTT, not websockets; the public wss door needs a ws client):
mosquitto_sub -h mosquitto -u <user> -P <pass> \
  -t 'oss-oo/OSS-OO/+/pr/+' -q 1 -v
```

```python
# Python (paho-mqtt) from anywhere:
import paho.mqtt.client as mqtt, ssl, json
c = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2, client_id="my-agent",
                protocol=mqtt.MQTTv311)
c.username_pw_set("my-agent", token)      # per-agent creds: P1.b
c.tls_set()                                # wss = tls + websockets
c.ws_set_options(path="/mqtt")
c.connect("events.oss-oo.io", 443, keepalive=60)
c.subscribe("oss-oo/OSS-OO/+/ci/+", qos=1)
c.on_message = lambda cl, u, m: print(m.topic, json.loads(m.payload)["action"])
c.loop_forever()
```

The bridge itself and the e2e proof use `webhook2mqtt/mqtt_min.py`, a
stdlib-only MQTT client (raw TCP + WebSocket) — zero-dependency
reference for the wire protocol.