## CPG-Radio

A fork of the CPG server that adds **non-blocking peer messaging** and a
**durable task-lease layer** on top of the Cognitive Planning Graph.

The base server is memory with no bus: it models a *completed* fan-out run
(WORKFLOW → WORKER → STRUCTURED_RESULT) but gives no agent a way to tell another
anything while work is in flight. AgentRadio-style message buses are the
converse — a bus with no memory, whose threads evaporate at the end of a run.

CPG-Radio adds the bus, and every message is also a typed, provenance-linked,
FTS + embedding-searchable graph node. Passive awareness that stays queryable
after the run and across sessions.

Section references below (DESIGN §x, SPEC §y) point at internal design
documents that are not published; they are kept as provenance markers on the
code comments that cite them.

---

### What's new

**Messaging** (DESIGN §6.1)

| Tool | What it does |
|---|---|
| `create_thread` | Open a THREAD node — a named channel for one topic or handoff. Validates any `workflow_id` / `task_ids` it links. |
| `send_message` | Post a MESSAGE to a thread as the calling agent. Creates the node, `posted_in` / `replies_to` / `refers_to` edges, and the ordering index row. Never waits for a reader. |
| `poll_messages` | Read messages newer than a cursor, oldest first, excluding your own. **Read-only — it does not advance any cursor.** |
| `ack_messages` | The only writer of the caller's cursor. Idempotent and forward-only; a replayed or stale ack cannot re-deliver processed messages. |
| `wait_for_mention` | Block up to 30 s for a message mentioning you (or broadcasting to `*`). Returns immediately if one is already pending; expiry is an empty success with `timed_out: true`, never an error. |

**Session identity** (§6.2)

| Tool | What it does |
|---|---|
| `join_session` | Bind this connection to an existing *active* session and declare your agent id. Returns a `binding_token` that preserves this logical identity when several agents share one MCP transport. Archives nothing. |

**Task leases** (§6.3)

| Tool | What it does |
|---|---|
| `claim_task` | Atomically take the lease on a TASK. Succeeds when it is unowned, expired, or already yours. Losing the race returns `claimed: false` with the real owner — losing is normal, not an exception. |
| `renew_lease` | Extend your lease. `renewed: false` means the task was reaped and reassigned, and you must stop working on it. |
| `release_task` | Give the lease up with a terminal status (or `planned`, a voluntary hand-back). Requires the `lease_generation` you hold. |
| `worker_heartbeat` | Stamp `heartbeat_at` on **your own** WORKER node so the reaper does not treat it as stale. Refused for another agent's worker. |
| `complete_worker` | Close your own WORKER as `completed` or `failed`; idempotent for an already-terminal worker and refused for another agent's worker. |

**Tips**

| Tool | What it does |
|---|---|
| `drop_tip` | Store a short project TIP (`command`, `check-first`, or `craft`) as the bound agent. Normalized-text duplicates return the existing id; `supersedes_id` retires an older tip. |
| `tips` | List tips by usefulness and recency, or search them through the existing hybrid FTS + embedding path with a usefulness boost. Superseded tips are hidden by default. |
| `tip_useful` | Give one idempotent usefulness vote per agent. |

`join_session` and `plan_session` surface up to five top project tips on arrival.
Pass the returned `binding_token` on later identity-bearing calls. A connection
that has carried more than one agent identity rejects tokenless identity-bearing
calls, preventing a sibling join from redirecting messages, acknowledgements,
leases, worker completion, or result authorship.
Ordinary `cpg_query` calls exclude TIP nodes unless `node_types` explicitly
includes `TIP`, keeping operational advice separate from normal graph recall.

**Swarm coordination plane**

| Area | Tools and guarantees |
|---|---|
| Authenticated campaigns | `create_campaign`, `invite_member`, `join_campaign`, `revoke_member`. Tokens are generated once, stored only as SHA-256 digests, bound to the connection, role-scoped, and rechecked against active membership on every protected operation. |
| Information barriers | Campaign threads enforce `open`, `campaign`, or exact `participants` access. Workflows, workers, structured results, spawned-worker views, and tips carry server-stamped campaign provenance. Messages carry trust, non-executable authority, visibility, sharing policy, priority, deadline, acknowledgement, supersession, and resolution metadata. `cpg_query` applies access policy before FTS/vector ranking. |
| Synchronization | `publish_signal` is a versioned CAS register with TTL and evidence refs. `create_barrier` / `arrive_barrier` provide explicit participant, quorum, phase, deadline, and CAS semantics. |
| Resource coordination | `claim_resource`, `renew_resource`, and `release_resource` manage shared/exclusive leases for names such as `gpu:0`, `path:/repo`, or `server:8773`, with TTLs and generations. |
| Reliable delivery | `subscribe_thread`, `poll_thread`, and `ack_thread` give each thread an independent durable cursor. A `requires_ack` message atomically creates one `delivered` receipt per authorized recipient; `ack_delivery` transitions only that recipient's obligation. Each thread is capped at 1,000 outstanding obligations, and `protocol_health` separates pending, overdue, and legacy orphaned requirements. |
| Atomic lifecycle | `post_progress` and `complete_assignment` combine lease, heartbeat/result, worker, and message mutations in one idempotent transaction. Self-managed workers close there; live spawned workers return `worker_completion="supervisor_owned"` and remain under the supervisor's single-writer lifecycle. |
| Independent work | `commit_result` / `reveal_result` seal outputs before reveal. Bound workers cannot attribute structured results to another worker. |
| Governance and health | Campaign command tips are quarantined until `moderate_tip` approves them with evidence. `protocol_health` reports coordination failures without returning message bodies. |
| Staleness | `stale_report` (read-only) lists running workers past the heartbeat threshold, leased tasks whose claimant's worker is terminal or missing, a `gpu.lock` `owner.json` that disagrees with the lease claimant, and watchers of finished jobs. `register_watch` / `close_watch` let a watcher declare the job (worker id, label, or done-file) it waits on so the report can see it. Added 2026-09-05 after seven of nine local log watchers were found tailing jobs that had ended hours earlier. |

Legacy session threads remain open. Migrate a crew by creating a campaign,
issuing each stable agent a membership token through a trusted channel, having
each call `join_campaign`, then creating campaign or participant threads.
Planning or joining another session clears the connection's old campaign grant.
Message and TIP text are always data; no body invokes a tool or changes control
state, regardless of its authority label.

**Reaper** (§6.4) — a background task, not a tool. Every 60 s it returns expired
task leases to `planned` (bumping the generation and telling the linked thread),
CAS-expires overdue barriers and shared/exclusive resource leases with dedicated
audit events, fails WORKERs with no heartbeat for an hour, and marks a WORKFLOW
whose workers have all terminated as `blocked`. Expired coordination records are
retained, and resource fencing generations remain monotonic when reclaimed. It
never auto-`completed` anything: completion is a claim about results, and only an
agent may make it. A coordinator can explicitly reconcile `blocked` to `completed`
after it has collected the terminal results.

Everything else from the base server — `cpg_think`, `record_fact`,
`record_belief`, `plan_session`, `update_task`, `replan`, `start_workflow`,
`update_workflow`, `register_worker`, `record_structured_result`,
`import_fable_trace`, `record_execution`, `reflect`, `resolve_contradiction`,
and the three `cpg://session/*` resources — keeps its existing contract.
`cpg_query`, `get_nodes`, and `record_structured_result` now receive caller
context so access and authorship can be enforced.

---

### The agent protocol

A **crew** is not a new entity: it is one `session_id` that several agents joined.
Because nodes stay session-scoped, `cpg_query`'s existing
`session_scope: current | project | all` gives crew-scoped and cross-crew recall
for free.

```
join_session  →  claim_task  →  work  →  poll / ack  →  release_task
```

1. **`join_session(session_id, agent_id)`** — or `plan_session(..., agent_id=...,
   archive_previous=False)` if you are the one opening the session. Identity is
   bound here. Keep the returned `binding_token` and pass it to later
   identity-bearing tools. No later tool takes an `agent_id` argument.
2. **`claim_task(task_id)`** — take the lease before touching the work. Keep the
   returned `lease_generation`; every later mutation needs it. `claimed: false`
   just means someone else got there first — pick another task.
3. **Work.** Call `renew_lease` if you will exceed the lease (default 900 s,
   clamped to `[60, 7200]`), and `worker_heartbeat` if you registered a worker.
4. **`poll_messages` / `ack_messages`** between work steps. Poll is read-only, so
   ack only what you have actually processed; a crash between the two re-delivers
   rather than loses. Post your own findings with `send_message`, mentioning the
   agents who need them.
5. **`release_task(task_id, lease_generation, status)`** — `done`,
   `done_pending_verification`, `failed`, `blocked`, or `planned` to hand it back.

If you crash instead, the reaper does step 5 for you within a minute.

**Parameter names are singular by design**: `after_seq`, `max_results`,
`lease_seconds`, `lease_generation`. There are no aliases.

---

### Running it

CPG-Radio runs on **port 8773** with its own DB and project id, deliberately
distinct from the live CPG server on 8766:

```sh
scripts/run-cpg-radio-http
```

That script pins `CPG_PORT=8773`, `CPG_PROJECT_ID=cpg-radio`, and
`CPG_DB_PATH=~/.cpg-radio/radio.db`. Health check:

```sh
curl -s http://127.0.0.1:8773/          # {"service": "cpg-radio", ...}
```

MCP requests go to `http://127.0.0.1:8773/mcp`.

**A development checkout refuses to start on the reserved legacy endpoint.**
Port 8766, or a `db_path` under `~/.cpg/` or `~/.local/state/codex-mcp/`, is
rejected at startup with `LiveDeploymentError` — the guard exists so a working
copy can never attach to an already-installed server's database. An installed
deployment that genuinely owns that endpoint may set
`CPG_ALLOW_LEGACY_ENDPOINT=1`; the spawn gate then requires the complete
canonical identity (loopback bind, the canonical project id, SQLite, and the
canonical DB path) and refuses mixtures of development and canonical settings.
Installed launchers should also set `CPG_SPAWN_TOKEN_PATH`,
`CPG_WORKER_LOG_DIR`, and `CPG_SUBAGENT_MCP_CONFIG` to their own state paths.

#### Configuration

| Variable | Meaning | Default |
|---|---|---|
| `CPG_DB_BACKEND` | `sqlite` or `postgres` | inferred |
| `CPG_DATABASE_URL` | PostgreSQL connection string | — |
| `CPG_DB_PATH` | SQLite path | `~/.cpg-radio/radio.db` |
| `CPG_PROJECT_ID` | project / session namespace | `cpg-radio` |
| `CPG_TRANSPORT` | `stdio` or `streamable_http` | `stdio` |
| `CPG_BIND_HOST` / `CPG_PORT` | HTTP bind | `127.0.0.1` / `8773` |
| `CPG_STREAMABLE_HTTP_PATH` | HTTP MCP path | `/mcp` |
| `CPG_REAPER_INTERVAL_S` | reaper tick; `0` disables it | `60` |
| `CPG_WORKER_STALE_S` | heartbeat age that fails a WORKER | `3600` |

Schema version is **3**. `create_schema` upgrades a v2 DB in place (the new
tables are `IF NOT EXISTS` and the new payload fields are all optional, so there
is no data migration) and **refuses to start** against a DB recording a *newer*
version than the binary.

---

### Development

```sh
.venv/bin/python -m pytest -q

# Optional: only against an isolated disposable PostgreSQL database
CPG_TEST_POSTGRES_URL=postgresql://... \
  .venv/bin/python -m pytest tests/integration/test_coordination_postgres.py -q
```

End-to-end smoke test — in-process, no network, no server, throwaway DB:

```sh
.venv/bin/python scripts/smoke-radio.py
```

It walks the full protocol (two agents join one session, one claims a task and
posts a finding mentioning the other, the other polls and acks, the lease
expires, the reaper hands the task back) and prints PASS/FAIL per step, exiting
nonzero on the first failure.

---

### Notes

- **Single process owns the DB.** The reaper assumes it. Multi-process
  deployment still needs leader election, TLS/service identity, and Postgres
  notification ownership. Coordination objects are backend-neutral; this does
  not pretend a multi-host control plane is safe yet.
- `wait_for_mention` is poll-backed at 500 ms on both backends. There is no
  `LISTEN/NOTIFY`: `get_db()` opens and closes a connection per call, so a
  `LISTEN` would never become active before the wait begins. Cursor polling is
  the primary path regardless; `wait_for_mention` is latency sugar.
- Message bodies are capped at 16 KiB and **rejected, never truncated**. Larger
  findings go through `record_structured_result` with the message carrying a
  `refs` pointer.
- `kind="status"` and `kind="system"` messages skip embedding, so a chatty bus
  does not run the model on every ping. `finding`, `question`, `answer`, `claim`,
  and `handoff` are embedded — those are the ones worth recalling later.
- `kind="system"` is server-only and rejected from `send_message`.
- Embedding search degrades gracefully when the model or vector backend is
  unavailable; SQLite uses `sqlite-vec`, PostgreSQL uses `pgvector`.