opencode: add in-process OpencodeService notifier (port of ops/forwarder)

2026-09-25 14:06:25 +00:00
parent c0c069ef08
commit 7d7c569d08
2 changed files with 494 additions and 0 deletions
@@ -0,0 +1,324 @@
+import { Logger } from '@nestjs/common';
+import { afterEach, describe, expect, it, vi } from 'vitest';
+import { OpencodeService } from './opencode.service.js';
+
+const API = 'http://172.24.0.1:9898';
+const WS_PATH = '/home/krzyw/projects/sandbox/mesh-bot';
+// the workspace UUID is regenerated on host restarts — only path/name/proxyPath matter
+const WORKSPACE = {
+  id: 'ws-uuid',
+  path: WS_PATH,
+  name: 'mesh-bot',
+  status: 'ready',
+  proxyPath: '/proxy/abc',
+};
+
+function json(body: unknown): Response {
+  return { ok: true, status: 200, json: async () => body } as Response;
+}
+
+function timeoutError(): Error {
+  return new DOMException('signal timed out', 'TimeoutError');
+}
+
+/** Fresh service with a clean env (creds unset by default — pass them explicitly). */
+function makeService(env: Record<string, string> = {}): OpencodeService {
+  process.env.OPENCODE_API = API;
+  process.env.OPENCODE_WORKSPACE_PATH = WS_PATH;
+  for (const k of ['OPENCODE_SESSION', 'OPENCODE_USER', 'OPENCODE_PASS'])
+    delete process.env[k];
+  Object.assign(process.env, env);
+  return new OpencodeService();
+}
+
+function stubSleep(svc: OpencodeService): ReturnType<typeof vi.fn> {
+  const sleep = vi.fn(async (_ms: number) => {});
+  (svc as unknown as { sleep(ms: number): Promise<void> }).sleep = sleep;
+  return sleep;
+}
+
+describe('OpencodeService.notify', () => {
+  let fetchMock: ReturnType<typeof vi.fn>;
+
+  beforeEach(() => {
+    fetchMock = vi.fn();
+    vi.stubGlobal('fetch', fetchMock);
+  });
+
+  afterEach(() => {
+    vi.unstubAllGlobals();
+    vi.restoreAllMocks();
+    for (const k of [
+      'OPENCODE_API',
+      'OPENCODE_WORKSPACE_PATH',
+      'OPENCODE_SESSION',
+      'OPENCODE_USER',
+      'OPENCODE_PASS',
+    ])
+      delete process.env[k];
+  });
+
+  it('posts the exact wrapper text with `from` (em-dash) to the resolved session', async () => {
+    const svc = makeService({
+      OPENCODE_USER: 'u',
+      OPENCODE_PASS: 'p',
+      OPENCODE_SESSION: 'ses_pinned',
+    });
+    fetchMock
+      .mockResolvedValueOnce(json([WORKSPACE]))
+      .mockResolvedValueOnce(json({}));
+
+    await svc.notify('hi', 42);
+
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    const [url, init] = fetchMock.mock.calls[1];
+    expect(url).toBe(`${API}/proxy/abc/session/ses_pinned/message`);
+    const opts = init as {
+      method?: string;
+      headers?: Record<string, string>;
+      body?: string;
+    };
+    expect(opts.method).toBe('POST');
+    expect(opts.headers?.['content-type']).toBe('application/json');
+    expect(opts.headers?.authorization).toBe(
+      'Basic ' + Buffer.from('u:p').toString('base64'),
+    );
+    // body is EXACTLY {"parts":[{"type":"text","text":msg}]} — em-dash, not a hyphen
+    expect(String(opts.body)).toBe(
+      JSON.stringify({
+        parts: [
+          {
+            type: 'text',
+            text: 'New mesh message from node 42: "hi" — please react to it.',
+          },
+        ],
+      }),
+    );
+  });
+
+  it('omits the `from node` part when there is no sender', async () => {
+    const svc = makeService({
+      OPENCODE_USER: 'u',
+      OPENCODE_PASS: 'p',
+      OPENCODE_SESSION: 'ses_pinned',
+    });
+    fetchMock
+      .mockResolvedValueOnce(json([WORKSPACE]))
+      .mockResolvedValueOnce(json({}));
+
+    await svc.notify('hi');
+
+    const [, init] = fetchMock.mock.calls[1];
+    expect(String((init as { body?: string }).body)).toBe(
+      JSON.stringify({
+        parts: [
+          {
+            type: 'text',
+            text: 'New mesh message: "hi" — please react to it.',
+          },
+        ],
+      }),
+    );
+  });
+
+  it('matches the workspace by path, preferring ready entries when several match', async () => {
+    const svc = makeService({
+      OPENCODE_USER: 'u',
+      OPENCODE_PASS: 'p',
+      OPENCODE_SESSION: 'ses_pinned',
+    });
+    fetchMock.mockResolvedValueOnce(
+      json([
+        {
+          ...WORKSPACE,
+          id: 'other',
+          status: 'starting',
+          proxyPath: '/proxy/not-ready',
+        }, // same path but not ready — listed first
+        WORKSPACE,
+        {
+          id: 'x',
+          name: 'mesh-bot',
+          status: 'ready',
+          proxyPath: '/proxy/other',
+        }, // different path — must lose to the path match
+      ]),
+    );
+    fetchMock.mockResolvedValueOnce(json({}));
+
+    await svc.notify('hi');
+
+    expect(fetchMock.mock.calls[1][0]).toBe(
+      `${API}/proxy/abc/session/ses_pinned/message`,
+    );
+  });
+
+  it('falls back to the workspace name when no path matches', async () => {
+    const svc = makeService({
+      OPENCODE_USER: 'u',
+      OPENCODE_PASS: 'p',
+      OPENCODE_SESSION: 'ses_pinned',
+    });
+    fetchMock.mockResolvedValueOnce(
+      json([
+        {
+          id: 'a',
+          path: '/elsewhere',
+          name: 'other',
+          status: 'ready',
+          proxyPath: '/proxy/a',
+        },
+        {
+          id: 'b',
+          path: '/renamed',
+          name: 'mesh-bot',
+          status: 'ready',
+          proxyPath: '/proxy/b',
+        },
+      ]),
+    );
+    fetchMock.mockResolvedValueOnce(json({}));
+
+    await svc.notify('hi');
+
+    expect(fetchMock.mock.calls[1][0]).toBe(
+      `${API}/proxy/b/session/ses_pinned/message`,
+    );
+  });
+
+  it('discovers the most recently updated session when none is pinned', async () => {
+    const svc = makeService({ OPENCODE_USER: 'u', OPENCODE_PASS: 'p' });
+    fetchMock
+      .mockResolvedValueOnce(json([WORKSPACE]))
+      .mockResolvedValueOnce(
+        json([
+          { id: 'ses_old', time: { updated: 100 } },
+          { id: 'ses_new', time: { updated: 200 } },
+        ]),
+      )
+      .mockResolvedValueOnce(json({}));
+
+    await svc.notify('hi');
+
+    expect(fetchMock.mock.calls[1][0]).toBe(`${API}/proxy/abc/session`);
+    expect(fetchMock.mock.calls[2][0]).toBe(
+      `${API}/proxy/abc/session/ses_new/message`,
+    );
+  });
+
+  it('uses the first session when none carry time.updated', async () => {
+    const svc = makeService({ OPENCODE_USER: 'u', OPENCODE_PASS: 'p' });
+    fetchMock
+      .mockResolvedValueOnce(json([WORKSPACE]))
+      .mockResolvedValueOnce(json([{ id: 'ses_a' }, { id: 'ses_b', time: {} }]))
+      .mockResolvedValueOnce(json({}));
+
+    await svc.notify('hi');
+
+    expect(fetchMock.mock.calls[2][0]).toBe(
+      `${API}/proxy/abc/session/ses_a/message`,
+    );
+  });
+
+  it('skips session discovery entirely when a session is pinned', async () => {
+    const svc = makeService({
+      OPENCODE_USER: 'u',
+      OPENCODE_PASS: 'p',
+      OPENCODE_SESSION: 'ses_pinned',
+    });
+    fetchMock
+      .mockResolvedValueOnce(json([WORKSPACE]))
+      .mockResolvedValueOnce(json({}));
+
+    await svc.notify('hi');
+
+    // workspaces GET + message POST only — no /session discovery GET
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(fetchMock.mock.calls[0][0]).toBe(`${API}/api/workspaces`);
+    expect(fetchMock.mock.calls[1][0]).toBe(
+      `${API}/proxy/abc/session/ses_pinned/message`,
+    );
+  });
+
+  it('treats a final-POST read timeout as delivered success and does not retry', async () => {
+    const svc = makeService({
+      OPENCODE_USER: 'u',
+      OPENCODE_PASS: 'p',
+      OPENCODE_SESSION: 'ses_pinned',
+    });
+    fetchMock
+      .mockResolvedValueOnce(json([WORKSPACE]))
+      .mockRejectedValueOnce(timeoutError()); // the message POST times out — opencode holds the response until the turn finishes
+
+    const logSpy = vi.spyOn(Logger.prototype, 'log');
+    await expect(svc.notify('hi')).resolves.toBeUndefined();
+
+    // exactly 2 calls: no retry after a timeout-as-delivered
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(logSpy.mock.calls.map((c) => String(c[0]))).toContain(
+      'opencode notify accepted (response still pending after 15 s)',
+    );
+  });
+
+  it('retries a failed workspace GET and succeeds on the next attempt', async () => {
+    const svc = makeService({
+      OPENCODE_USER: 'u',
+      OPENCODE_PASS: 'p',
+      OPENCODE_SESSION: 'ses_pinned',
+    });
+    const sleep = stubSleep(svc);
+    fetchMock
+      .mockRejectedValueOnce(new Error('boom')) // attempt 1: workspaces GET fails
+      .mockResolvedValueOnce(json([WORKSPACE])) // attempt 2: workspaces ok
+      .mockResolvedValueOnce(json({})); // message POST ok
+
+    await expect(svc.notify('hi')).resolves.toBeUndefined();
+
+    expect(fetchMock).toHaveBeenCalledTimes(3);
+    expect(sleep).toHaveBeenNthCalledWith(1, 2_000);
+  });
+
+  it('gives up after 4 attempts and logs opencode notify dropped', async () => {
+    const svc = makeService({ OPENCODE_USER: 'u', OPENCODE_PASS: 'p' });
+    const sleep = stubSleep(svc);
+    fetchMock.mockRejectedValue(new Error('down'));
+    const warnSpy = vi.spyOn(Logger.prototype, 'warn');
+
+    await expect(svc.notify('hi')).resolves.toBeUndefined(); // never throws — fire-and-forget safe
+
+    // one workspaces GET per attempt: exactly 4 attempts total
+    expect(fetchMock).toHaveBeenCalledTimes(4);
+    expect(sleep).toHaveBeenNthCalledWith(1, 2_000);
+    expect(sleep).toHaveBeenNthCalledWith(2, 10_000);
+    expect(sleep).toHaveBeenNthCalledWith(3, 30_000);
+    const warns = warnSpy.mock.calls.map((c) => String(c[0]));
+    expect(warns.some((w) => w.includes('attempt 1/4'))).toBe(true);
+    // the GET failure is wrapped with the URL for a clear reason — match prefix + cause
+    expect(
+      warns.some(
+        (w) =>
+          w.startsWith('opencode notify dropped after 4 attempts:') &&
+          w.includes(': down'),
+      ),
+    ).toBe(true);
+  });
+
+  it('is a no-op without credentials (empty or unset)', async () => {
+    for (const env of [{}, { OPENCODE_USER: '', OPENCODE_PASS: 'p' }] as Record<
+      string,
+      string
+    >[]) {
+      const svc = makeService(env);
+      await expect(svc.notify('hi', 42)).resolves.toBeUndefined();
+      expect(fetchMock).not.toHaveBeenCalled();
+    }
+  });
+
+  it('logs once at construction when disabled', () => {
+    const warnSpy = vi.spyOn(Logger.prototype, 'warn');
+    makeService({}); // no creds
+    expect(warnSpy.mock.calls.map((c) => String(c[0]))).toContain(
+      'opencode notifier disabled: OPENCODE_USER/OPENCODE_PASS not set',
+    );
+  });
+});
@@ -0,0 +1,170 @@
+import { Injectable, Logger } from '@nestjs/common';
+
+// ponytail: hardcoded backoff [2s, 10s, 30s] (~45 s window) — no env knob; tune here if outages run longer
+const RETRY_DELAYS_MS = [2_000, 10_000, 30_000];
+// opencode accepts a session message immediately but holds the HTTP response until the agent turn finishes
+const TIMEOUT_MS = 15_000;
+
+/** Entry of GET {api}/api/workspaces (only the fields we use). */
+interface WorkspaceEntry {
+  id?: string;
+  path?: string;
+  name?: string;
+  status?: string;
+  proxyPath?: string;
+}
+
+/** Entry of GET {base}/session (only the fields we use). */
+interface SessionEntry {
+  id: string;
+  time?: { updated?: number };
+}
+
+@Injectable()
+export class OpencodeService {
+  private readonly logger = new Logger(OpencodeService.name);
+  // read at construction, like MeshService's env config
+  private readonly api = process.env.OPENCODE_API ?? 'http://172.0.0.1:9898';
+  private readonly workspacePath =
+    process.env.OPENCODE_WORKSPACE_PATH ??
+    '/home/krzyw/projects/sandbox/mesh-bot';
+  // when set, skips session discovery
+  private readonly sessionOverride = process.env.OPENCODE_SESSION ?? '';
+  private readonly user = process.env.OPENCODE_USER ?? '';
+  private readonly pass = process.env.OPENCODE_PASS ?? '';
+  /** Notifier disabled without credentials — the app must keep running (and notify nothing) when they are missing. */
+  private readonly enabled = this.user !== '' && this.pass !== '';
+
+  constructor() {
+    if (!this.enabled)
+      this.logger.warn(
+        'opencode notifier disabled: OPENCODE_USER/OPENCODE_PASS not set',
+      );
+  }
+
+  /**
+   * Push an rx mesh message into the opencode session as a text part. Resolves when delivered —
+   * including on a final-POST read timeout, which means DELIVERED (opencode holds the response until
+   * the agent turn finishes) and is never retried. Workspace/session resolution failures are retried
+   * up to RETRY_DELAYS_MS.length + 1 attempts total; the last failure logs `opencode notify dropped
+   * after N attempts`. No-op when disabled (no credentials).
+   */
+  async notify(text: string, from?: number): Promise<void> {
+    if (!this.enabled) return;
+    const msg =
+      'New mesh message' +
+      (from != null ? ` from node ${from}` : '') +
+      `: "${text}" — please react to it.`;
+    const maxAttempts = RETRY_DELAYS_MS.length + 1;
+    for (let attempt = 1; ; attempt++) {
+      try {
+        const { base, sid } = await this.resolveTarget();
+        await this.postMessage(base, sid, msg);
+        return; // success on any attempt — a POST timeout already resolved as delivered inside postMessage
+      } catch (err: unknown) {
+        const message = err instanceof Error ? err.message : String(err);
+        if (attempt === maxAttempts) {
+          this.logger.warn(
+            `opencode notify dropped after ${attempt} attempts: ${message}`,
+          );
+          return;
+        }
+        this.logger.warn(
+          `opencode notify failed (attempt ${attempt}/${maxAttempts}): ${message}`,
+        );
+        await this.sleep(RETRY_DELAYS_MS[attempt - 1]);
+      }
+    }
+  }
+
+  /** Injectable so specs can skip the backoff delays. */
+  private sleep(ms: number): Promise<void> {
+    return new Promise((resolve) => setTimeout(resolve, ms));
+  }
+
+  private authHeaders(): Record<string, string> {
+    return {
+      authorization:
+        'Basic ' + Buffer.from(`${this.user}:${this.pass}`).toString('base64'),
+    };
+  }
+
+  /** Resolve (base_url, session_id) fresh per request; throws with a clear reason on API/workspace/session failure. */
+  private async resolveTarget(): Promise<{ base: string; sid: string }> {
+    const workspaces = await this.getJson<WorkspaceEntry[]>(
+      `${this.api}/api/workspaces`,
+    );
+    // match by path first, fall back to name — the workspace UUID is regenerated on host restarts
+    const matches =
+      workspaces.filter((w) => w.path === this.workspacePath).length > 0
+        ? workspaces.filter((w) => w.path === this.workspacePath)
+        : workspaces.filter((w) => w.name === 'mesh-bot');
+    // prefer ready entries when several match
+    const pool =
+      matches.filter((w) => w.status === 'ready').length > 0
+        ? matches.filter((w) => w.status === 'ready')
+        : matches;
+    if (pool.length === 0)
+      throw new Error(
+        `workspace not found: path=${this.workspacePath} name=mesh-bot`,
+      );
+    const ws = pool[0];
+    const base = this.api + (ws.proxyPath ?? '');
+    let sid = this.sessionOverride;
+    if (!sid) {
+      const sessions = await this.getJson<SessionEntry[]>(`${base}/session`);
+      if (sessions.length === 0)
+        throw new Error(
+          `no sessions in workspace ${ws.id} (${this.workspacePath})`,
+        );
+      // most recently updated session; first entry when none carry time.updated
+      const updated = (s: SessionEntry): number => s.time?.updated ?? 0;
+      sid = sessions.some((s) => updated(s))
+        ? sessions.reduce((a, b) => (updated(b) > updated(a) ? b : a)).id
+        : sessions[0].id;
+    }
+    return { base, sid };
+  }
+
+  private async getJson<T>(url: string): Promise<T> {
+    let res: Response;
+    try {
+      res = await fetch(url, {
+        headers: this.authHeaders(),
+        signal: AbortSignal.timeout(TIMEOUT_MS),
+      });
+    } catch (err) {
+      throw new Error(
+        `opencode API unreachable at ${url}: ${(err as Error).message}`,
+      );
+    }
+    if (!res.ok) throw new Error(`opencode API HTTP ${res.status} at ${url}`);
+    return (await res.json()) as T;
+  }
+
+  private async postMessage(
+    base: string,
+    sid: string,
+    msg: string,
+  ): Promise<void> {
+    try {
+      const res = await fetch(`${base}/session/${sid}/message`, {
+        method: 'POST',
+        headers: { ...this.authHeaders(), 'content-type': 'application/json' },
+        body: JSON.stringify({ parts: [{ type: 'text', text: msg }] }),
+        signal: AbortSignal.timeout(TIMEOUT_MS),
+      });
+      if (!res.ok) throw new Error(`HTTP ${res.status}`); // HTTP errors retry like network failures
+    } catch (err) {
+      // a read timeout here means DELIVERED, not failed — opencode accepted the message and is holding
+      // the response until the agent turn finishes; never retried
+      if ((err as Error).name === 'TimeoutError') {
+        this.logger.log(
+          `opencode notify accepted (response still pending after ${TIMEOUT_MS / 1000} s)`,
+        );
+        return;
+      }
+      throw err;
+    }
+  }
+}