Pi

A hook for Pi that runs inside the pi process, polls EventPort on a timer, and injects each event into the session via pi.send() — a new agent loop starts when idle, the message is queued when busy.

Prerequisites

Node >= 18
The Pi coding agent (pi --version runs)
An EventPort subscription — you need its gateway URL and consumer token (starts with egc_, from the subscription's Agent tab)

Installation

1. Create the hook directory

Pick one scope — global (all projects) or the current project:

# global scope
mkdir -p ~/.pi/agent/hooks

# project scope (inside your project root)
mkdir -p .pi/hooks

2. Save the two files below

The examples use the global path; if you chose project scope, replace ~/.pi/agent/hooks with .pi/hooks.

File 1 of 2: `poller.ts`

Save to ~/.pi/agent/hooks/poller.ts:

/**
 * eventport poller hook for Pi (pi-mono coding agent).
 *
 * Pi hooks are TypeScript modules loaded inside the pi process. This hook
 * starts a timer that polls eventport for buffered webhook events and
 * injects each one into the session via `pi.send()` — when the agent is idle
 * a new agent loop starts immediately, when busy the message is queued.
 * That wake/queue semantics is documented in the Pi README ("Inject messages
 * from external sources to wake up the agent").
 *
 * Gateway semantics (as implemented in apps/gateway/src/index.ts):
 *   1. GET /events (Bearer egc_*) returns pending events and DELETES them
 *      server-side (consume-on-read, the default subscription mode).
 *   2. Delivery is at-most-once: there is no /ack endpoint and no
 *      redelivery. A poll that fails before the GET leaves events in place
 *      for the next cycle.
 *   3. If `pi.send()` throws, the event is dead-lettered to a local NDJSON
 *      file (EG_DLQ) so nothing is silently lost.
 *
 * Configuration (environment variables of the `pi` process):
 *   EG_URL        gateway base URL, e.g. https://gw.eventport.dev
 *   EG_TOKEN      consumer token (egc_*)
 *   EG_INTERVAL   poll interval in ms, default 60000 (limit is 60 req/min)
 *   EG_SOURCES    optional comma-separated allowlist, e.g. "github"
 *   EG_DLQ        dead-letter file, default ./eventport.dead.ndjson
 */
import { appendFileSync } from 'node:fs';
import type { PiSession } from './pi-types';

const EG_URL = (process.env.EG_URL ?? '').replace(/\/$/, '');
const EG_TOKEN = process.env.EG_TOKEN ?? '';
const EG_INTERVAL = Number(process.env.EG_INTERVAL ?? 60_000);
const EG_DLQ = process.env.EG_DLQ ?? 'eventport.dead.ndjson';
const EG_SOURCES = (process.env.EG_SOURCES ?? '')
  .split(',')
  .map((s) => s.trim())
  .filter(Boolean);

interface GatewayEvent {
  messageId: string;
  payload: unknown;
  timestamp: number;
}

export default function eventPortPoller(pi: PiSession) {
  if (!EG_URL || !EG_TOKEN) {
    console.warn('[eventport] EG_URL / EG_TOKEN not set — hook idle');
    return;
  }

  /** messageIds already injected in this process lifetime */
  const delivered = new Set<string>();
  let polling = false;
  let timer: ReturnType<typeof setInterval> | undefined;

  // ---------------------------------------------------------------- gateway

  async function pollEvents(): Promise<GatewayEvent[]> {
    const res = await fetch(`${EG_URL}/events`, {
      headers: { Authorization: `Bearer ${EG_TOKEN}` },
    });
    if (!res.ok) throw new Error(`GET /events → ${res.status}`);
    const body = (await res.json()) as { events?: GatewayEvent[] };
    // consume-on-read: returned rows are already deleted server-side
    return body.events ?? [];
  }

  function deadLetter(event: GatewayEvent, reason: string): void {
    try {
      appendFileSync(
        EG_DLQ,
        `${JSON.stringify({ ...event, reason, failedAt: Date.now() })}\n`,
      );
      console.warn(`[eventport] ${event.messageId} dead-lettered to ${EG_DLQ} (${reason})`);
    } catch (err) {
      console.error(`[eventport] cannot write dead-letter file ${EG_DLQ}:`, err);
    }
  }

  // ------------------------------------------------------------------ task

  /** Map a raw webhook payload to an instruction the agent can act on. */
  function buildTask(event: GatewayEvent): string {
    const p = event.payload as Record<string, unknown> | null;
    if (p && typeof p.action === 'string' && p.pull_request) {
      const pr = p.pull_request as Record<string, unknown>;
      const repo = (p.repository as Record<string, unknown> | undefined)?.full_name ?? 'unknown repo';
      return [
        `GitHub PR event (${p.action}) on ${repo} #${pr.number ?? '?'} — ${pr.title ?? ''}`,
        `Review the changes (gh pr diff), look for bugs and security issues,`,
        `and post your findings as a PR comment. Event messageId: ${event.messageId}.`,
      ].join(' ');
    }
    return [
      'An external webhook event arrived from eventport.',
      `messageId: ${event.messageId}`,
      'Inspect the payload below and handle it according to your skills:',
      JSON.stringify(event.payload).slice(0, 8_000),
    ].join('\n');
  }

  // ------------------------------------------------------------------- loop

  async function cycle(): Promise<void> {
    const events = await pollEvents();
    for (const event of events) {
      if (delivered.has(event.messageId)) continue;
      const source = (event.payload as Record<string, unknown> | null)?.source;
      if (EG_SOURCES.length > 0 && typeof source === 'string' && !EG_SOURCES.includes(source)) {
        // Not in the allowlist: already consumed by the poll, just skip.
        console.log(`[eventport] ${event.messageId} skipped (source '${source}' not in EG_SOURCES)`);
        continue;
      }
      try {
        pi.send(buildTask(event)); // queues when busy, new loop when idle
        delivered.add(event.messageId);
      } catch (err) {
        console.warn(`[eventport] send failed for ${event.messageId}:`, err);
        deadLetter(event, 'pi-send-failed');
      }
    }
    if (events.length > 0) {
      console.log(`[eventport] ${events.length} event(s) consumed`);
    }
  }

  pi.on('session', () => {
    if (timer) clearInterval(timer); // one timer across session restarts
    timer = setInterval(() => {
      if (polling) return;
      polling = true;
      cycle()
        .catch((err) => console.warn('[eventport] poll failed:', err))
        .finally(() => (polling = false));
    }, EG_INTERVAL);
    console.log(`[eventport] polling ${EG_URL} every ${EG_INTERVAL}ms (consume-on-read)`);
  });
}

File 2 of 2: `pi-types.ts`

Save to ~/.pi/agent/hooks/pi-types.ts:

/**
 * Minimal local typings for the Pi hook API surface used by poller.ts.
 * Pi (pi-mono) loads hooks as TypeScript modules and passes a session-ish
 * object; `send()` queues the message when the agent is streaming and starts
 * a new agent loop when idle. If the official API exports types later,
 * switch to those and delete this file.
 */
export interface PiSession {
  /** Inject a user message into the session (text + optional attachments). */
  send(text: string, attachments?: unknown): void;
  /** Subscribe to hook events, e.g. 'session' (fires on session start). */
  on(event: string, handler: () => void): void;
}

3. Start Pi with the gateway configuration

EG_URL=https://gw.eventport.dev \
EG_TOKEN=egc_xxx \
pi

4. Verify

On session start the hook logs one line:

[eventport] polling https://gw.eventport.dev every 60000ms (consume-on-read)

You can also check the queue directly (it should return {"events":[]} when empty):

curl -s -H 'Authorization: Bearer egc_xxx' https://gw.eventport.dev/events

Configuration

| Env | Meaning | Default | |-----|---------|---------| | EG_URL | gateway base URL | required | | EG_TOKEN | consumer token (egc_*) | required | | EG_INTERVAL | poll interval in ms | 60000 | | EG_SOURCES | comma allowlist, e.g. github | all | | EG_DLQ | dead-letter file for failed sends | ./eventport.dead.ndjson |

Delivery semantics

The gateway deletes events on read (consume-on-read): once GET /events returns an event it is removed server-side, and delivery is at-most-once with no redelivery.

pi.send() accepting the message (new loop when idle, queued when busy) counts as delivered. If pi.send() throws, the event is dead-lettered to the local file EG_DLQ so nothing is silently lost.
Make agent actions idempotent by messageId — the pi send queue and any manual dead-letter replay can present the same event twice.
One hook instance tracks one subscription (the one that issued EG_TOKEN). To consume another subscription, repeat this setup with its token.

Pi