Console

DeepSeek Harness

A dsh plugin that polls EventPort for buffered webhook events on a timer and wakes the agent with each one. It runs in-process — no external scheduler needed.

Prerequisites

  • Node >= 18
  • The dsh CLI (pnpm dsh --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 plugin directory

mkdir -p ~/.dsh/plugins/eventport/src

2. Save the three files below

Each file goes to its destination path under ~/.dsh/plugins/eventport/.

File 1 of 3: package.json

Save to ~/.dsh/plugins/eventport/package.json:

{
  "name": "dsh-eventport",
  "version": "0.1.0",
  "description": "DeepSeek Harness plugin: poll eventport events and wake the agent to process them",
  "type": "module",
  "main": "src/index.ts",
  "keywords": [
    "dsh-plugin",
    "eventport",
    "webhook"
  ],
  "license": "MIT"
}

File 2 of 3: src/index.ts

Save to ~/.dsh/plugins/eventport/src/index.ts:

/**
 * dsh-eventport — poll eventport for buffered webhook events and wake
 * the agent to process them, in-process (no external scheduler needed).
 *
 * 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
 *      and they are picked up by the next cycle.
 *   3. If the agent wake fails, the event is dead-lettered to a local NDJSON
 *      file (EG_DLQ) so nothing is silently lost.
 *
 * Configuration (environment variables, read once at load):
 *   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
 *
 * NOTE: dsh is a developer preview; the agent wake API surface
 * (followup / inject) is isolated in `wakeAgent()` so it is a one-function
 * fix if the service shape changes across versions.
 */
import { appendFileSync } from 'node:fs';
import type { Context } from '@deepseek-ai/cordis';

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

export const name = 'eventport-poller';

export function apply(ctx: Context) {
  const url = process.env.EG_URL;
  const token = process.env.EG_TOKEN;
  const interval = Number(process.env.EG_INTERVAL ?? 60_000);
  const dlq = process.env.EG_DLQ ?? 'eventport.dead.ndjson';
  const sources = (process.env.EG_SOURCES ?? '')
    .split(',')
    .map((s) => s.trim())
    .filter(Boolean);

  if (!url || !token) {
    ctx.logger.warn(
      '[eventport] EG_URL / EG_TOKEN not set — plugin loaded but idle',
    );
    return;
  }

  const log = ctx.logger;
  const base = url.replace(/\/$/, '');
  /** messageIds already handed to the agent in this process lifetime */
  const delivered = new Set<string>();
  let polling = false;

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

  async function pollEvents(): Promise<GatewayEvent[]> {
    const res = await fetch(`${base}/events`, {
      headers: { Authorization: `Bearer ${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(
        dlq,
        `${JSON.stringify({ ...event, reason, failedAt: Date.now() })}\n`,
      );
      log.warn(`[eventport] ${event.messageId} dead-lettered to ${dlq} (${reason})`);
    } catch (err) {
      log.error(`[eventport] cannot write dead-letter file ${dlq}: ${String(err)}`);
    }
  }

  // ------------------------------------------------------------------ agent

  /**
   * Wake the agent with a new turn. Isolates the preview-era agent service
   * API: `followup()` when idle (new turn), `inject()` when busy (queued
   * into the running turn). Returns false when the service is unavailable.
   */
  async function wakeAgent(task: string): Promise<boolean> {
    const agent = (ctx as Record<string, unknown>).agent as
      | { followup?: (t: string) => Promise<unknown>; inject?: (t: string) => Promise<unknown>; isBusy?: () => boolean }
      | undefined;
    if (!agent) {
      log.warn('[eventport] agent service not found — dead-lettering');
      return false;
    }
    const busy = agent.isBusy?.() ?? false;
    if (busy && agent.inject) await agent.inject(task);
    else if (agent.followup) await agent.followup(task);
    else return false;
    return true;
  }

  /** 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;
    // GitHub PR events
    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(' ');
    }
    // Fallback: generic instruction with the raw payload attached
    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();
    let ok = 0;
    for (const event of events) {
      if (delivered.has(event.messageId)) continue;
      const source = (event.payload as Record<string, unknown> | null)?.source;
      if (sources.length > 0 && typeof source === 'string' && !sources.includes(source)) {
        // Not in the allowlist: already consumed by the poll, just skip.
        log.info(`[eventport] ${event.messageId} skipped (source '${source}' not in EG_SOURCES)`);
        continue;
      }
      const woken = await wakeAgent(buildTask(event));
      if (woken) {
        delivered.add(event.messageId);
        ok += 1;
      } else {
        deadLetter(event, 'agent-wake-failed');
      }
    }
    if (events.length > 0) {
      log.info(`[eventport] ${events.length} event(s) consumed, ${ok} dispatched, ${events.length - ok} dead-lettered`);
    }
  }

  ctx.effect(() => {
    const timer = setInterval(() => {
      if (polling) return;
      polling = true;
      cycle()
        .catch((err) => log.warn(`[eventport] poll failed: ${String(err)}`))
        .finally(() => (polling = false));
    }, interval);
    log.info(`[eventport] polling ${base} every ${interval}ms (consume-on-read)`);
    return () => clearInterval(timer); // auto-cleanup on plugin unload
  });
}

File 3 of 3: cordis.patch.yml

Save to ~/.dsh/plugins/eventport/cordis.patch.yml:

# Dev overlay: load the plugin into a running dsh web profile without installing it.
# Replace the name path below with the ABSOLUTE path of src/index.ts on this
# machine, e.g. /Users/you/.dsh/plugins/eventport/src/index.ts.
- insert:
  - id: eventport
    name: '/absolute/path/to/.dsh/plugins/eventport/src/index.ts'

In cordis.patch.yml, replace the name: value with the absolute path of the src/index.ts file you saved in step 2.

3. Start dsh web with the overlay

EG_URL=https://gw.eventport.dev \
EG_TOKEN=egc_xxx \
pnpm dsh web --patch ~/.dsh/plugins/eventport/cordis.patch.yml

4. Verify

On startup the plugin 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.

  • If waking the agent fails, the event is dead-lettered to the local file EG_DLQ (default ./eventport.dead.ndjson) so nothing is silently lost.
  • Make agent actions idempotent by messageId — dead-letter replays can present the same event twice.
  • One plugin instance tracks one subscription (the one that issued EG_TOKEN). To consume another subscription, repeat this setup with its token.