diff --git a/src/opencode/opencode.service.spec.ts b/src/opencode/opencode.service.spec.ts new file mode 100644 index 0000000..0574654 --- /dev/null +++ b/src/opencode/opencode.service.spec.ts @@ -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 = {}): 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 { + const sleep = vi.fn(async (_ms: number) => {}); + (svc as unknown as { sleep(ms: number): Promise }).sleep = sleep; + return sleep; +} + +describe('OpencodeService.notify', () => { + let fetchMock: ReturnType; + + 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; + 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', + ); + }); +}); diff --git a/src/opencode/opencode.service.ts b/src/opencode/opencode.service.ts new file mode 100644 index 0000000..cd5dc69 --- /dev/null +++ b/src/opencode/opencode.service.ts @@ -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 { + 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 { + return new Promise((resolve) => setTimeout(resolve, ms)); + } + + private authHeaders(): Record { + 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( + `${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(`${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(url: string): Promise { + 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 { + 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; + } + } +}