From 7d7c569d0832fc3c1df0dcd55f8bba5d4fd16a80 Mon Sep 17 00:00:00 2001 From: Kamil Krzywda Date: Fri, 25 Sep 2026 14:06:25 +0000 Subject: [PATCH 1/3] opencode: add in-process OpencodeService notifier (port of ops/forwarder) --- src/opencode/opencode.service.spec.ts | 324 ++++++++++++++++++++++++++ src/opencode/opencode.service.ts | 170 ++++++++++++++ 2 files changed, 494 insertions(+) create mode 100644 src/opencode/opencode.service.spec.ts create mode 100644 src/opencode/opencode.service.ts 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; + } + } +} -- 2.54.0 From 5eb1a79e70cb388f541970430fc2c01196edfa17 Mon Sep 17 00:00:00 2001 From: Kamil Krzywda Date: Fri, 25 Sep 2026 14:10:14 +0000 Subject: [PATCH 2/3] mesh: replace notify webhook with in-process opencode notifier; drop forwarder service --- MESHTASTIC.md | 2 +- README.md | 15 +- docker-compose.yml | 17 +-- ops/forwarder/Dockerfile | 4 - ops/forwarder/forwarder.py | 95 ------------- src/mesh/mesh.module.ts | 6 +- src/mesh/mesh.service.spec.ts | 253 ++++++---------------------------- src/mesh/mesh.service.ts | 62 ++------- 8 files changed, 77 insertions(+), 377 deletions(-) delete mode 100644 ops/forwarder/Dockerfile delete mode 100644 ops/forwarder/forwarder.py diff --git a/MESHTASTIC.md b/MESHTASTIC.md index 45396d7..f376496 100644 --- a/MESHTASTIC.md +++ b/MESHTASTIC.md @@ -25,4 +25,4 @@ Rules for how mesh-bot communicates over Meshtastic and how the agent should rea ## Notifications (mesh → agent) -- RX text on the notify channel (`MESH_NOTIFY_CHANNEL`, default 3) triggers an outbound webhook POST `{text, from}` to `MESH_NOTIFY_URL` (host forwarder :4097 → opencode session). Messages from our own node are never notified (self-echo filter, v0.11.2). +- RX text on the notify channel (`MESH_NOTIFY_CHANNEL`, default 3) is pushed in-process to the live opencode session by `OpencodeService` (workspace/session resolved per request from the opencode API; disabled without `OPENCODE_USER`/`OPENCODE_PASS`). Messages from our own node are never notified (self-echo filter, v0.11.2). diff --git a/README.md b/README.md index 1c6e064..69ed200 100644 --- a/README.md +++ b/README.md @@ -97,16 +97,16 @@ Hourly cron job (`src/weather`): fetches current conditions for Lublin from [Ope - `WEATHER_LAT` (default `51.2465`) / `WEATHER_LON` (default `22.5684`) — coordinates queried from Open-Meteo - `WEATHER_CITY` (default `"Lublin"`) — city name in the report line -## Notify forwarder +## Opencode notifier -Second compose service (`ops/forwarder`, `python:3.12-slim`, stdlib only): receives the bot's outbound notify webhook POSTs (`{text, from?}` on `MESH_NOTIFY_URL`) and forwards each one as a message to the mesh-bot folder's live opencode session (Basic auth). The opencode workspace UUID is regenerated on host restarts, so the forwarder resolves it dynamically per request via the opencode API: `GET /api/workspaces` → match by `path == FORWARDER_WORKSPACE_PATH` (fallback name `mesh-bot`, status `ready` preferred when several match) → `GET {proxyPath}/session` → most recently updated session (`time.updated`; first entry if absent). No caching — messages are rare. Runs with `network_mode: host` — it listens on the host's `0.0.0.0:4097`, which is why `MESH_NOTIFY_URL=http://172.24.0.1:4097/notify` works from the bot container, and reaches opencode at `127.0.0.1:9898`. `restart: unless-stopped`; no Traefik labels (host networking). +In-process service (`src/opencode`, no separate compose service): when an rx text message arrives on `MESH_NOTIFY_CHANNEL`, it pushes it to the mesh-bot folder's live opencode session (Basic auth). The opencode workspace UUID is regenerated on host restarts, so the notifier resolves it dynamically per request via the opencode API: `GET {OPENCODE_API}/api/workspaces` → match by `path == OPENCODE_WORKSPACE_PATH` (fallback name `mesh-bot`, status `ready` preferred when several match) → `GET {proxyPath}/session` → most recently updated session (`time.updated`; first entry if absent). No caching — messages are rare. The message POST body is exactly `{"parts":[{"type":"text","text":msg}]}` with msg = `New mesh message[ from node ]: "" — please react to it.` (the `from node` part omitted when there is no sender). opencode accepts the message immediately but holds the HTTP response until the agent turn finishes, so a 15 s read timeout on that POST means **delivered** — treated as success, never retried; workspace/session GET failures are retried up to 3 times with backoff (2 s / 10 s / 30 s) before being dropped. Config lives in `.env` (gitignored) and is passed via compose environment — credentials are never committed to the repo: -- `FORWARDER_API` (default `http://127.0.0.1:9898`) — opencode API base URL -- `FORWARDER_WORKSPACE_PATH` (default `/home/krzyw/projects/sandbox/mesh-bot`) — workspace matched by its `path` field (fallback name `mesh-bot`; status `ready` preferred when several match) -- `FORWARDER_SESSION` (optional, default empty) — pins a session ID; when set, session discovery is skipped. Empty/unset → the most recently updated session of the resolved workspace is used per request -- `FORWARDER_USER` / `FORWARDER_PASS` — Basic auth credentials; **required**, the container exits on startup if either is missing +- `OPENCODE_API` (default `http://172.24.0.1:9898`) — opencode API base URL as seen from the bot container (dev_net bridge → host; opencode binds 0.0.0.0 on the host) +- `OPENCODE_WORKSPACE_PATH` (default `/home/krzyw/projects/sandbox/mesh-bot`) — workspace matched by its `path` field (fallback name `mesh-bot`; status `ready` preferred when several match) +- `OPENCODE_SESSION` (optional, default empty) — pins a session ID; when set, session discovery is skipped. Empty/unset → the most recently updated session of the resolved workspace is used per request +- `OPENCODE_USER` / `OPENCODE_PASS` — Basic auth credentials; missing/empty **disables** the notifier (one startup log line, no-op) instead of failing ## Env vars @@ -114,8 +114,7 @@ Config lives in `.env` (gitignored) and is passed via compose environment — cr - `MESH_PORT` (default `4403`) — bridge port - `MESH_RECONNECT_MS` (default `5000`) — reconnect interval in ms - `MESH_SEND_CHANNEL` (default `3`) — channel index all sends target (`test`; never the primary channel) -- `MESH_NOTIFY_URL` — outbound notification webhook URL; when set (non-empty), an inbound rx text message arriving on channel `MESH_NOTIFY_CHANNEL` fires a fire-and-forget HTTP POST with JSON body `{text, from}` (`from` omitted when absent). Failed POSTs are retried up to 3 times with backoff (2 s / 10 s / 30 s) before being dropped. Synthetic injected messages (`POST /api/messages`, channel 0) trigger it too when the notify channel is set to match. Messages from our own node are never notified (self-echo filter). Unset or empty = disabled (no fetch at all) -- `MESH_NOTIFY_CHANNEL` (default `3`) — channel index that triggers the notify webhook +- `MESH_NOTIFY_CHANNEL` (default `3`) — channel index whose rx texts trigger the opencode notifier (see "Opencode notifier"; synthetic injected messages, channel 0, trigger it too when set to match; messages from our own node are never notified) - `MESH_DB_PATH` (default `./data/mesh-bot.db`) — SQLite file backing the message history (parent dir created if missing; in Docker, mounted at `/app/data` via a named volume so it survives container recreation) - `MESH_POLL_MS` (default `1000`) — how often the poller re-upserts the message history into SQLite and prunes rows beyond the newest 1000 - `PORT` (default `3000`) — HTTP listen port diff --git a/docker-compose.yml b/docker-compose.yml index 50339e8..5e7d5c9 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -6,8 +6,12 @@ services: MESH_PORT: '4403' MESH_DB_PATH: ./data/mesh-bot.db MESH_POLL_MS: '1000' - MESH_NOTIFY_URL: ${MESH_NOTIFY_URL:-} MESH_NOTIFY_CHANNEL: '${MESH_NOTIFY_CHANNEL:-3}' + OPENCODE_API: ${OPENCODE_API:-http://172.24.0.1:9898} + OPENCODE_WORKSPACE_PATH: ${OPENCODE_WORKSPACE_PATH:-/home/krzyw/projects/sandbox/mesh-bot} + OPENCODE_SESSION: ${OPENCODE_SESSION:-} + OPENCODE_USER: ${OPENCODE_USER:-} + OPENCODE_PASS: ${OPENCODE_PASS:-} WEATHER_CRON: ${WEATHER_CRON:-} volumes: - mesh_data:/app/data @@ -19,17 +23,6 @@ services: - 'traefik.http.routers.mesh-bot.rule=Host(`bot.dev.kkhost.pl`)' - 'traefik.http.routers.mesh-bot.entrypoints=web' - 'traefik.http.services.mesh-bot.loadbalancer.server.port=3000' - forwarder: - build: ./ops/forwarder - network_mode: host - environment: - FORWARDER_API: ${FORWARDER_API:-http://127.0.0.1:9898} - FORWARDER_WORKSPACE_PATH: ${FORWARDER_WORKSPACE_PATH:-/home/krzyw/projects/sandbox/mesh-bot} - FORWARDER_SESSION: ${FORWARDER_SESSION:-} - FORWARDER_USER: ${FORWARDER_USER:-} - FORWARDER_PASS: ${FORWARDER_PASS:-} - restart: unless-stopped - volumes: mesh_data: diff --git a/ops/forwarder/Dockerfile b/ops/forwarder/Dockerfile deleted file mode 100644 index 995d690..0000000 --- a/ops/forwarder/Dockerfile +++ /dev/null @@ -1,4 +0,0 @@ -FROM python:3.12-slim -WORKDIR /app -COPY forwarder.py . -CMD ["python3", "forwarder.py"] diff --git a/ops/forwarder/forwarder.py b/ops/forwarder/forwarder.py deleted file mode 100644 index 346e10d..0000000 --- a/ops/forwarder/forwarder.py +++ /dev/null @@ -1,95 +0,0 @@ -#!/usr/bin/env python3 -"""mesh-bot notify forwarder: POST /notify {text, from?} -> opencode session message (basic auth). - -The opencode workspace UUID is regenerated on host restarts, so the target workspace + session are -resolved dynamically per request via the opencode API (no caching — messages are rare): list -workspaces, pick the mesh-bot one by path (fallback: name), then its most recently updated session. - -Note on semantics: the opencode API accepts a session message immediately but holds the HTTP -response until the agent turn completes — so a read timeout here means DELIVERED, not failed. -""" -import json, base64, urllib.request -from http.server import BaseHTTPRequestHandler, HTTPServer -import os - -API = os.environ.get("FORWARDER_API", "http://127.0.0.1:9898") -WORKSPACE_PATH = os.environ.get("FORWARDER_WORKSPACE_PATH", "/home/krzyw/projects/sandbox/mesh-bot") -SESSION_OVERRIDE = os.environ.get("FORWARDER_SESSION", "") # when set, skips session discovery -USER = os.environ["FORWARDER_USER"] # required; fail fast if missing -PASS = os.environ["FORWARDER_PASS"] # required; fail fast if missing - -def _auth_headers(): - return {"Authorization": "Basic " + base64.b64encode(f"{USER}:{PASS}".encode()).decode()} - -def _get_json(url): - req = urllib.request.Request(url, headers=_auth_headers()) - try: - with urllib.request.urlopen(req, timeout=15) as r: - return json.loads(r.read().decode()) - except Exception as e: - raise RuntimeError(f"opencode API unreachable at {url}: {e!r}") from None - -def resolve_target(): - """Resolve (base_url, session_id) fresh per request; raises RuntimeError with a clear reason.""" - workspaces = _get_json(API + "/api/workspaces") - matches = [w for w in workspaces if w.get("path") == WORKSPACE_PATH] or \ - [w for w in workspaces if w.get("name") == "mesh-bot"] - pool = [w for w in matches if w.get("status") == "ready"] or matches - if not pool: - raise RuntimeError(f"workspace not found: path={WORKSPACE_PATH} name=mesh-bot") - base = API + pool[0]["proxyPath"] - sid = SESSION_OVERRIDE - if not sid: - sessions = _get_json(base + "/session") - if not sessions: - raise RuntimeError(f"no sessions in workspace {pool[0]['id']} ({WORKSPACE_PATH})") - - def updated(s): - return (s.get("time") or {}).get("updated") or 0 - sid = max(sessions, key=updated)["id"] if any(updated(s) for s in sessions) else sessions[0]["id"] - return base, sid - -class H(BaseHTTPRequestHandler): - def do_POST(self): - if self.path != "/notify": - self.send_response(404); self.end_headers(); return - n = int(self.headers.get("Content-Length", 0)) - try: - body = json.loads(self.rfile.read(n) or b"{}") - text, frm = body.get("text"), body.get("from") - msg = 'New mesh message' + (f' from node {frm}' if frm is not None else '') + f': "{text}" — please react to it.' - base, sid = resolve_target() # fresh per request; RuntimeError on API/workspace/session failure - req = urllib.request.Request( - f"{base}/session/{sid}/message", - data=json.dumps({"parts": [{"type": "text", "text": msg}]}).encode(), - headers={"Content-Type": "application/json", **_auth_headers()}, - method="POST") - try: - r = urllib.request.urlopen(req, timeout=15) - code = r.status # headers received => opencode accepted the message - try: out = r.read(200).decode(errors='replace') - except Exception as e2: out = f'(body read failed after delivery: {e2!r})' - print(f"forwarded to opencode session {sid}: HTTP {code} {out[:150]!r}", flush=True) - self._reply(code if 200 <= code < 300 else 502, b'{"ok":true}' if 200 <= code < 300 else str(code).encode()) - except TimeoutError: - # request was sent; opencode holds the response until the agent turn finishes => delivered - print(f"forwarded to opencode session {sid}: accepted (response still pending after 15 s)", flush=True) - self._reply(202, b'{"ok":true,"note":"accepted; opencode response pending"}') - except RuntimeError as e: - # resolve_target() failure modes: API unreachable / workspace not found / no sessions in workspace - print(f"target resolution failed: {e!r}", flush=True) - self._reply(502, repr(e).encode()) - except Exception as e: - print(f"opencode call failed: {e!r}", flush=True) - self._reply(502, repr(e).encode()) - - def _reply(self, code, payload): - try: - self.send_response(code); self.end_headers(); self.wfile.write(payload) - except (BrokenPipeError, ConnectionResetError): - pass # client went away; delivery state already logged above - - def log_message(self, fmt, *args): - print("forwarder:", fmt % args, flush=True) - -HTTPServer(("0.0.0.0", 4097), H).serve_forever() diff --git a/src/mesh/mesh.module.ts b/src/mesh/mesh.module.ts index bc945c3..433db53 100644 --- a/src/mesh/mesh.module.ts +++ b/src/mesh/mesh.module.ts @@ -2,6 +2,7 @@ import { Module } from '@nestjs/common'; import { mkdirSync } from 'node:fs'; import { dirname } from 'node:path'; import { DatabaseSync } from 'node:sqlite'; +import { OpencodeService } from '../opencode/opencode.service.js'; import { McpController } from './mcp.controller.js'; import { MeshController } from './mesh.controller.js'; import { initMessagesTable, MeshService } from './mesh.service.js'; @@ -18,9 +19,12 @@ function openMessageDb(): DatabaseSync { @Module({ controllers: [MeshController, McpController], providers: [ + OpencodeService, { provide: MeshService, - useFactory: () => new MeshService(openMessageDb()), + useFactory: (opencode: OpencodeService) => + new MeshService(openMessageDb(), opencode), + inject: [OpencodeService], }, ], exports: [MeshService], diff --git a/src/mesh/mesh.service.spec.ts b/src/mesh/mesh.service.spec.ts index fc10c7c..40f08bc 100644 --- a/src/mesh/mesh.service.spec.ts +++ b/src/mesh/mesh.service.spec.ts @@ -1,5 +1,4 @@ import { create, fromBinary, toBinary } from '@bufbuild/protobuf'; -import { Logger } from '@nestjs/common'; import * as P from '@meshtastic/protobufs'; import { DatabaseSync } from 'node:sqlite'; import { beforeEach, describe, expect, it, vi } from 'vitest'; @@ -15,6 +14,7 @@ import { telemetryPacketFrame, waitFor, } from './fake-bridge.js'; +import { OpencodeService } from '../opencode/opencode.service.js'; import { initMessagesTable, MeshService } from './mesh.service.js'; // live position update: POSITION_APP packet with a serialized Position in Data.payload @@ -692,12 +692,14 @@ describe('MeshService message history', () => { }); }); -describe('MeshService notify webhook', () => { - const fetchMock = vi.fn(async (..._args: unknown[]) => undefined); +describe('MeshService opencode notifier', () => { // our own node number for the self-echo filter specs (distinct from KAMIL_NUM, an external user) const MY_NODE = 987654321; + let notifyMock: ReturnType; - beforeEach(() => fetchMock.mockClear()); + beforeEach(() => { + notifyMock = vi.fn(async (_text: string, _from?: number) => {}); + }); // myInfo frame as streamed during the initial sync — sets svc.myNodeNum function myInfoFrame(num: number): Uint8Array { @@ -711,13 +713,20 @@ describe('MeshService notify webhook', () => { ); } - it('POSTs {text, from} to MESH_NOTIFY_URL when an rx text lands on the notify channel (default 3)', async () => { + // MeshService with a mocked OpencodeService wired in (structural cast — only notify() is used) + function makeSvc(): MeshService { + return new MeshService( + undefined, + { notify: notifyMock } as unknown as OpencodeService, + ); + } + + it('notifies via the opencode service when an rx text lands on the notify channel (default 3)', async () => { const bridge = await startBridge(); process.env.MESH_HOST = '127.0.0.1'; process.env.MESH_PORT = String(bridge.port); - process.env.MESH_NOTIFY_URL = 'http://notify.example/hook'; - vi.stubGlobal('fetch', fetchMock); - const svc = new MeshService(); + delete process.env.MESH_NOTIFY_CHANNEL; // default 3 + const svc = makeSvc(); try { svc.onModuleInit(); await waitFor( @@ -726,36 +735,22 @@ describe('MeshService notify webhook', () => { bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 3)); - await waitFor(() => fetchMock.mock.calls.length === 1); - expect(fetchMock).toHaveBeenCalledTimes(1); - const [url, init] = fetchMock.mock.calls[0]; - expect(url).toBe('http://notify.example/hook'); - const opts = init as { - method?: string; - headers?: Record; - body?: string; - }; - expect(opts.method).toBe('POST'); - expect(opts.headers?.['content-type']).toBe('application/json'); - expect(JSON.parse(String(opts.body))).toEqual({ - text: 'ping', - from: KAMIL_NUM, - }); + await waitFor(() => notifyMock.mock.calls.length === 1); + expect(notifyMock).toHaveBeenCalledTimes(1); + expect(notifyMock).toHaveBeenCalledWith('ping', KAMIL_NUM); } finally { - vi.unstubAllGlobals(); - delete process.env.MESH_NOTIFY_URL; + delete process.env.MESH_NOTIFY_CHANNEL; svc.onModuleDestroy(); bridge.close(); } }); - it('does not call fetch for rx texts on other channels', async () => { + it('does not notify for rx texts on other channels', async () => { const bridge = await startBridge(); process.env.MESH_HOST = '127.0.0.1'; process.env.MESH_PORT = String(bridge.port); - process.env.MESH_NOTIFY_URL = 'http://notify.example/hook'; - vi.stubGlobal('fetch', fetchMock); - const svc = new MeshService(); + delete process.env.MESH_NOTIFY_CHANNEL; // default 3 + const svc = makeSvc(); try { svc.onModuleInit(); await waitFor( @@ -764,109 +759,48 @@ describe('MeshService notify webhook', () => { bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 2)); - // the message is captured in history — but no webhook fires off-channel + // the message is captured in history — but no notification fires off-channel await waitFor(() => svc.messages().length === 1); - expect(fetchMock).not.toHaveBeenCalled(); + expect(notifyMock).not.toHaveBeenCalled(); } finally { - vi.unstubAllGlobals(); - delete process.env.MESH_NOTIFY_URL; + delete process.env.MESH_NOTIFY_CHANNEL; svc.onModuleDestroy(); bridge.close(); } }); - it('never calls fetch when MESH_NOTIFY_URL is unset or empty', async () => { - const bridge = await startBridge(); - process.env.MESH_HOST = '127.0.0.1'; - process.env.MESH_PORT = String(bridge.port); - delete process.env.MESH_NOTIFY_URL; // unset → disabled - vi.stubGlobal('fetch', fetchMock); - const svc = new MeshService(); - try { - svc.onModuleInit(); - await waitFor( - () => bridge.sockets.length === 1 && svc.status().connected, - ); - - bridge.sockets[0].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 3)); - await waitFor(() => svc.messages().length === 1); - expect(fetchMock).not.toHaveBeenCalled(); - - // empty string → also disabled (a fresh service re-reads the env) - process.env.MESH_NOTIFY_URL = ''; - const svc2 = new MeshService(); - try { - svc2.onModuleInit(); - await waitFor( - () => bridge.sockets.length === 2 && svc2.status().connected, - ); - - bridge.sockets[1].write(textPacketFrame(KAMIL_NUM, 42, 'ping', 3)); - await waitFor(() => svc2.messages().length === 1); - } finally { - delete process.env.MESH_NOTIFY_URL; - svc2.onModuleDestroy(); - } - expect(fetchMock).not.toHaveBeenCalled(); - } finally { - vi.unstubAllGlobals(); - svc.onModuleDestroy(); - bridge.close(); - } - }); - - it('fires the webhook for an injected message when MESH_NOTIFY_CHANNEL matches its channel (0)', async () => { - process.env.MESH_NOTIFY_URL = 'http://notify.example/hook'; + it('fires the notifier for an injected message when MESH_NOTIFY_CHANNEL matches its channel (0)', async () => { process.env.MESH_NOTIFY_CHANNEL = '0'; - vi.stubGlobal('fetch', fetchMock); - const svc = new MeshService(); + const svc = makeSvc(); try { svc.injectMessage('hi', 42); - expect(fetchMock).toHaveBeenCalledTimes(1); - const [url, init] = fetchMock.mock.calls[0]; - expect(url).toBe('http://notify.example/hook'); - const opts = init as { - method?: string; - headers?: Record; - body?: string; - }; - expect(opts.method).toBe('POST'); - expect(opts.headers?.['content-type']).toBe('application/json'); - expect(JSON.parse(String(opts.body))).toEqual({ text: 'hi', from: 42 }); + expect(notifyMock).toHaveBeenCalledTimes(1); + expect(notifyMock).toHaveBeenCalledWith('hi', 42); } finally { - vi.unstubAllGlobals(); - delete process.env.MESH_NOTIFY_URL; delete process.env.MESH_NOTIFY_CHANNEL; } }); - it('omits `from` in the webhook body when an injected message has no sender', async () => { - process.env.MESH_NOTIFY_URL = 'http://notify.example/hook'; + it('passes `from` through as undefined when an injected message has no sender', async () => { process.env.MESH_NOTIFY_CHANNEL = '0'; - vi.stubGlobal('fetch', fetchMock); - const svc = new MeshService(); + const svc = makeSvc(); try { svc.injectMessage('hi'); - expect(fetchMock).toHaveBeenCalledTimes(1); - const [, init] = fetchMock.mock.calls[0]; - const opts = init as { body?: string }; - expect(JSON.parse(String(opts.body))).toEqual({ text: 'hi' }); + expect(notifyMock).toHaveBeenCalledTimes(1); + expect(notifyMock).toHaveBeenCalledWith('hi', undefined); } finally { - vi.unstubAllGlobals(); - delete process.env.MESH_NOTIFY_URL; delete process.env.MESH_NOTIFY_CHANNEL; } }); - it('does not call fetch for rx texts from our own node on the notify channel', async () => { + it('does not notify for rx texts from our own node on the notify channel', async () => { const bridge = await startBridge(); process.env.MESH_HOST = '127.0.0.1'; process.env.MESH_PORT = String(bridge.port); - process.env.MESH_NOTIFY_URL = 'http://notify.example/hook'; - vi.stubGlobal('fetch', fetchMock); - const svc = new MeshService(); + delete process.env.MESH_NOTIFY_CHANNEL; // default 3 + const svc = makeSvc(); try { svc.onModuleInit(); await waitFor( @@ -877,28 +811,25 @@ describe('MeshService notify webhook', () => { bridge.sockets[0].write(myInfoFrame(MY_NODE)); await waitFor(() => svc.status().myNodeNum === MY_NODE); - // an rx text from ourselves on the notify channel must not fire the webhook + // an rx text from ourselves on the notify channel must not fire the notifier bridge.sockets[0].write(textPacketFrame(MY_NODE, 42, 'ping', 3)); // history capture proves the packet was processed (notifyRx runs right after it) await waitFor(() => svc.messages().length === 1); - expect(fetchMock).not.toHaveBeenCalled(); + expect(notifyMock).not.toHaveBeenCalled(); } finally { - vi.unstubAllGlobals(); - delete process.env.MESH_NOTIFY_URL; + delete process.env.MESH_NOTIFY_CHANNEL; svc.onModuleDestroy(); bridge.close(); } }); - it('does not call fetch for an injected message from our own node', async () => { + it('does not notify for an injected message from our own node', async () => { const bridge = await startBridge(); process.env.MESH_HOST = '127.0.0.1'; process.env.MESH_PORT = String(bridge.port); - process.env.MESH_NOTIFY_URL = 'http://notify.example/hook'; process.env.MESH_NOTIFY_CHANNEL = '0'; - vi.stubGlobal('fetch', fetchMock); - const svc = new MeshService(); + const svc = makeSvc(); try { svc.onModuleInit(); await waitFor( @@ -910,109 +841,13 @@ describe('MeshService notify webhook', () => { svc.injectMessage('hi', MY_NODE); - expect(fetchMock).not.toHaveBeenCalled(); + expect(notifyMock).not.toHaveBeenCalled(); } finally { - vi.unstubAllGlobals(); - delete process.env.MESH_NOTIFY_URL; delete process.env.MESH_NOTIFY_CHANNEL; svc.onModuleDestroy(); bridge.close(); } }); - - it('retries a failed webhook POST with backoff and stops on success', async () => { - process.env.MESH_NOTIFY_URL = 'http://notify.example/hook'; - process.env.MESH_NOTIFY_CHANNEL = '0'; - const fetchMock = vi - .fn() - .mockRejectedValueOnce(new Error('boom 1')) - .mockRejectedValueOnce(new Error('boom 2')) - .mockResolvedValueOnce(undefined); - vi.stubGlobal('fetch', fetchMock); - const warnSpy = vi.spyOn(Logger.prototype, 'warn'); - const sleep = vi.fn(async (_ms: number) => {}); - const svc = new MeshService(); - (svc as unknown as { sleep(ms: number): Promise }).sleep = sleep; - try { - expect(() => svc.injectMessage('hi', 42)).not.toThrow(); - - await waitFor(() => fetchMock.mock.calls.length === 3); - // success on the third attempt: exactly 3 POSTs, no dropped log - expect(fetchMock).toHaveBeenCalledTimes(3); - expect(sleep).toHaveBeenNthCalledWith(1, 2_000); - expect(sleep).toHaveBeenNthCalledWith(2, 10_000); - const warns = warnSpy.mock.calls.map((c) => String(c[0])); - expect(warns.some((w) => w.includes('attempt 1/4'))).toBe(true); - expect(warns.some((w) => w.includes('attempt 2/4'))).toBe(true); - expect(warns.some((w) => w.includes('notify dropped'))).toBe(false); - } finally { - warnSpy.mockRestore(); - vi.unstubAllGlobals(); - delete process.env.MESH_NOTIFY_URL; - delete process.env.MESH_NOTIFY_CHANNEL; - } - }); - - it('retries a webhook POST that answers with an HTTP error status and stops on success', async () => { - process.env.MESH_NOTIFY_URL = 'http://notify.example/hook'; - process.env.MESH_NOTIFY_CHANNEL = '0'; - const fetchMock = vi - .fn() - .mockResolvedValueOnce({ ok: false, status: 502 }) - .mockResolvedValueOnce({ ok: false, status: 502 }) - .mockResolvedValueOnce({ ok: true, status: 200 }); - vi.stubGlobal('fetch', fetchMock); - const warnSpy = vi.spyOn(Logger.prototype, 'warn'); - const sleep = vi.fn(async (_ms: number) => {}); - const svc = new MeshService(); - (svc as unknown as { sleep(ms: number): Promise }).sleep = sleep; - try { - expect(() => svc.injectMessage('hi', 42)).not.toThrow(); - - await waitFor(() => fetchMock.mock.calls.length === 3); - // HTTP 502 twice, then 200: exactly 3 POSTs, no dropped log - expect(fetchMock).toHaveBeenCalledTimes(3); - const warns = warnSpy.mock.calls.map((c) => String(c[0])); - expect(warns.some((w) => w.includes('attempt 1/4'))).toBe(true); - expect(warns.some((w) => w.includes('HTTP 502'))).toBe(true); - expect(warns.some((w) => w.includes('notify dropped'))).toBe(false); - } finally { - warnSpy.mockRestore(); - vi.unstubAllGlobals(); - delete process.env.MESH_NOTIFY_URL; - delete process.env.MESH_NOTIFY_CHANNEL; - } - }); - - it('gives up after 4 attempts and logs notify dropped', async () => { - process.env.MESH_NOTIFY_URL = 'http://notify.example/hook'; - process.env.MESH_NOTIFY_CHANNEL = '0'; - const fetchMock = vi.fn().mockRejectedValue(new Error('down')); - vi.stubGlobal('fetch', fetchMock); - const warnSpy = vi.spyOn(Logger.prototype, 'warn'); - const sleep = vi.fn(async (_ms: number) => {}); - const svc = new MeshService(); - (svc as unknown as { sleep(ms: number): Promise }).sleep = sleep; - try { - expect(() => svc.injectMessage('hi', 42)).not.toThrow(); - - await waitFor(() => - warnSpy.mock.calls.some((c) => - String(c[0]).includes('notify dropped after 4 attempts'), - ), - ); - // exactly 4 attempts, final WARN with the exact message, nothing thrown - expect(fetchMock).toHaveBeenCalledTimes(4); - const warns = warnSpy.mock.calls.map((c) => String(c[0])); - expect(warns).toContain('notify dropped after 4 attempts: down'); - expect(sleep).toHaveBeenNthCalledWith(3, 30_000); - } finally { - warnSpy.mockRestore(); - vi.unstubAllGlobals(); - delete process.env.MESH_NOTIFY_URL; - delete process.env.MESH_NOTIFY_CHANNEL; - } - }); }); describe('MeshService injectMessage', () => { diff --git a/src/mesh/mesh.service.ts b/src/mesh/mesh.service.ts index da51ac2..17f533e 100644 --- a/src/mesh/mesh.service.ts +++ b/src/mesh/mesh.service.ts @@ -9,6 +9,7 @@ import * as net from 'node:net'; // type-only import: node:sqlite is only required at runtime by whoever opens a database (mesh.module.ts / tests) import type { DatabaseSync } from 'node:sqlite'; import type * as Proto from '@meshtastic/protobufs'; +import { OpencodeService } from '../opencode/opencode.service.js'; import { BROADCAST_ADDR, PORT_NODEINFO_APP, @@ -132,8 +133,6 @@ function jsonSafe(value: T): T { // key material must never leave the process — replaced with a marker so callers can see the field exists const REDACTED = 'REDACTED'; -// ponytail: hardcoded notify-webhook backoff [2s, 10s, 30s] (~45 s window) — no env knob; tune here if outages run longer -const NOTIFY_RETRY_DELAYS_MS = [2_000, 10_000, 30_000]; /** Config sections that carry secrets: security (PKI keys) and network (wifi psk). */ function redactSection( name: string, @@ -161,8 +160,8 @@ export class MeshService implements OnModuleInit, OnModuleDestroy { process.env.MESH_SEND_CHANNEL ?? '3', 10, ); - // outbound notification webhook: fire-and-forget POST when an rx text lands on the notify channel; empty URL = disabled - private readonly notifyUrl = process.env.MESH_NOTIFY_URL ?? ''; + // outbound notification: fire-and-forget push to the opencode notifier when an rx text lands on the notify channel; + // disabled state (missing credentials) lives in OpencodeService itself private readonly notifyChannel = parseInt( process.env.MESH_NOTIFY_CHANNEL ?? '3', 10, @@ -172,10 +171,13 @@ export class MeshService implements OnModuleInit, OnModuleDestroy { /** SQLite database backing the message history; null = persistence disabled (tests construct without one). */ private db: DatabaseSync | null = null; + /** In-process opencode notifier for rx texts on the notify channel; null in tests that don't wire one. */ + private readonly opencode: OpencodeService | null; private pollTimer: NodeJS.Timeout | null = null; - constructor(db?: DatabaseSync) { + constructor(db?: DatabaseSync, opencode?: OpencodeService) { this.db = db ?? null; + this.opencode = opencode ?? null; } private socket: net.Socket | null = null; @@ -430,54 +432,20 @@ export class MeshService implements OnModuleInit, OnModuleDestroy { at: Date.now(), }; this.pushHistory(rec); - // injected records ride channel 0 — the webhook fires only when MESH_NOTIFY_CHANNEL is set to match + // injected records ride channel 0 — the notifier fires only when MESH_NOTIFY_CHANNEL is set to match this.notifyRx(0, text, from); return { id: rec.id }; } - /** Fire-and-forget notify-webhook POST for an rx text on `channel` (no-op unless URL is set and channels match). */ + /** Fire-and-forget opencode notifier push for an rx text on `channel` (no-op unless channels match and a notifier is wired). */ private notifyRx(channel: number, text: string, from?: number): void { - if (this.notifyUrl && channel === this.notifyChannel) { - // self-echo filter: our own node must never trigger a notification of itself - if (from !== undefined && from === this.myNodeNum) { - this.logger.debug(`notify skipped: rx from our own node ${from}`); - return; - } - const body: Record = { text }; - if (from !== undefined) body.from = from; - void this.postNotify(body); + if (!this.opencode || channel !== this.notifyChannel) return; + // self-echo filter: our own node must never trigger a notification of itself + if (from !== undefined && from === this.myNodeNum) { + this.logger.debug(`notify skipped: rx from our own node ${from}`); + return; } - } - - /** POST the webhook with retries: up to NOTIFY_RETRY_DELAYS_MS.length + 1 attempts total. */ - private async postNotify(body: Record): Promise { - const maxAttempts = NOTIFY_RETRY_DELAYS_MS.length + 1; - for (let attempt = 1; ; attempt++) { - try { - const res = await fetch(this.notifyUrl, { - method: 'POST', - headers: { 'content-type': 'application/json' }, - body: JSON.stringify(body), - }); - if (res && !res.ok) throw new Error('HTTP ' + res.status); // HTTP errors retry like network failures - return; // success on any attempt — same as today's single POST - } catch (err: unknown) { - const msg = (err as Error).message; - this.logger.warn( - `notify webhook failed (attempt ${attempt}/${maxAttempts}): ${msg}`, - ); - if (attempt === maxAttempts) { - this.logger.warn(`notify dropped after ${attempt} attempts: ${msg}`); - return; - } - await this.sleep(NOTIFY_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)); + void this.opencode.notify(text, from); } /** -- 2.54.0 From 83e47c96091ddbffa6c011197a0f29ab09a0692c Mon Sep 17 00:00:00 2001 From: Kamil Krzywda Date: Fri, 25 Sep 2026 14:14:04 +0000 Subject: [PATCH 3/3] =?UTF-8?q?docs+changelog:=20v0.14.0=20=E2=80=94=20in-?= =?UTF-8?q?process=20opencode=20notifier=20replaces=20the=20forwarder?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 2 +- CHANGELOG.md | 11 +++++++++++ package.json | 2 +- 3 files changed, 13 insertions(+), 2 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 0f1a3a1..c045f89 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -17,7 +17,7 @@ ## API 1. **Live base URL** — `http://bot.dev.kkhost.pl` (Traefik → docker-compose stack on this host; container port 3000 via external network dev_net, no published ports). -2. **Endpoints** — all under the `/api` global prefix: `GET /api/status` (`{ connected, myNodeNum, host, port }`), `POST /api/send` (body `{ "text": "...", "to"?: , "replyId"?: }` — numbers pass through as-is (integer 1..0xffffffff), names resolve like `GET /api/nodes/:id`, omitted/""/"!b"/"broadcast" → broadcast; unknown or ambiguous targets are a 400. Optional `replyId` (integer in 1..0xffffffff, else 400) references an earlier message's packet id from `GET /api/messages` — when set, the recipient app renders the text as a threaded reply/quote of that packet (`Data.reply_id`, field 7). Returns `{ id }`), `GET /api/channels` (the device's Meshtastic channels as `{ index, name, uplinkEnabled, downlinkEnabled }[]`; psk/keys redacted), `GET /api/device` (device info captured during the initial sync: `{ myNodeNum, firmwareVersion?, metadata?, localConfig: {
: ...}, moduleConfig: {: ...} }` — DeviceMetadata as-is plus per-section Config/ModuleConfig messages; security publicKey/privateKey/adminKey and network wifiPsk redacted to `"REDACTED"`; in-memory, reset on reconnect), `GET /api/messages` (message history newest first: rx text with `from`, tx sends with `to` + `status` pending→delivered/failed via routing ACKs — broadcasts stay pending because firmware strips want_ack; last 100 in memory (seeded from SQLite on startup) — backed by SQLite: re-upserted every MESH_POLL_MS and pruned to the newest 1000 rows, so it survives process restarts and bridge reconnects), `POST /api/messages` (body `{ "text": "...", "from"?: }`, returns `{ id }`; synthetic inbox injection for external triggers — records an rx entry in the history as if received on the mesh (channel 0) so external systems can push messages to trigger communication later; NOT transmitted over RF, nothing goes to the bridge or air; appears in `GET /api/messages` like any other rx message), `GET /api/telemetry` (telemetry history newest first as `{ from, at /* epoch ms */, kind, data }[]`: live TELEMETRY_APP packets — device metrics and environment sensor readings where nodes have them; kind is the meshtastic.Telemetry oneof case name (deviceMetrics/environmentMetrics/airQualityMetrics/powerMetrics/localStats/healthMetrics/hostMetrics/trafficManagementStats), data that variant's payload with proto uint64 fields as decimal strings; last 100, in-memory, reset on reconnect), `GET /api/nodes` (known nodes sorted by num as `{ num, longName?, shortName?, hwModel?, batteryLevel?, lastHeard: number|null, position?: { lat, lon, alt? }, snr?, hopsAway?, viaMqtt? }[]`, from the initial sync plus live POSITION_APP updates; reset on reconnect) and `GET /api/nodes/:id` (lookup by exact node number/numeric string, then case-insensitive exact long/short name, else unique substring; 404 when ambiguous or missing). Scalar API docs are served at `/` (OpenAPI JSON at `/docs-json`). Keep this list in sync with the controller. All `POST /api/send` traffic targets channel index 3 (`test`) on purpose — never the primary channel; override via `MESH_SEND_CHANNEL` (default `3`). Message history is persisted to SQLite: file path from `MESH_DB_PATH` (default `./data/mesh-bot.db`, parent dir auto-created), poller re-upserts every `MESH_POLL_MS` ms (default 1000) and prunes the table to the newest 1000 rows. Outbound notification webhook: when an RX text arrives on channel `MESH_NOTIFY_CHANNEL` (default `3`), a fire-and-forget HTTP POST of `{text, from}` (`from` omitted when absent) goes to `MESH_NOTIFY_URL` — unset/empty disables it entirely; failed POSTs are retried up to 3 times with backoff (2 s / 10 s / 30 s) before being dropped; no response handling. Injected messages (`POST /api/messages`, channel 0) trigger it too when the notify channel is set to match. Messages from our own node are never notified (self-echo filter). The forwarder resolves the mesh-bot workspace/session dynamically from the opencode API per request (`FORWARDER_API`/`FORWARDER_WORKSPACE_PATH`; `FORWARDER_SESSION` pins a session when set). +2. **Endpoints** — all under the `/api` global prefix: `GET /api/status` (`{ connected, myNodeNum, host, port }`), `POST /api/send` (body `{ "text": "...", "to"?: , "replyId"?: }` — numbers pass through as-is (integer 1..0xffffffff), names resolve like `GET /api/nodes/:id`, omitted/""/"!b"/"broadcast" → broadcast; unknown or ambiguous targets are a 400. Optional `replyId` (integer in 1..0xffffffff, else 400) references an earlier message's packet id from `GET /api/messages` — when set, the recipient app renders the text as a threaded reply/quote of that packet (`Data.reply_id`, field 7). Returns `{ id }`), `GET /api/channels` (the device's Meshtastic channels as `{ index, name, uplinkEnabled, downlinkEnabled }[]`; psk/keys redacted), `GET /api/device` (device info captured during the initial sync: `{ myNodeNum, firmwareVersion?, metadata?, localConfig: {
: ...}, moduleConfig: {: ...} }` — DeviceMetadata as-is plus per-section Config/ModuleConfig messages; security publicKey/privateKey/adminKey and network wifiPsk redacted to `"REDACTED"`; in-memory, reset on reconnect), `GET /api/messages` (message history newest first: rx text with `from`, tx sends with `to` + `status` pending→delivered/failed via routing ACKs — broadcasts stay pending because firmware strips want_ack; last 100 in memory (seeded from SQLite on startup) — backed by SQLite: re-upserted every MESH_POLL_MS and pruned to the newest 1000 rows, so it survives process restarts and bridge reconnects), `POST /api/messages` (body `{ "text": "...", "from"?: }`, returns `{ id }`; synthetic inbox injection for external triggers — records an rx entry in the history as if received on the mesh (channel 0) so external systems can push messages to trigger communication later; NOT transmitted over RF, nothing goes to the bridge or air; appears in `GET /api/messages` like any other rx message), `GET /api/telemetry` (telemetry history newest first as `{ from, at /* epoch ms */, kind, data }[]`: live TELEMETRY_APP packets — device metrics and environment sensor readings where nodes have them; kind is the meshtastic.Telemetry oneof case name (deviceMetrics/environmentMetrics/airQualityMetrics/powerMetrics/localStats/healthMetrics/hostMetrics/trafficManagementStats), data that variant's payload with proto uint64 fields as decimal strings; last 100, in-memory, reset on reconnect), `GET /api/nodes` (known nodes sorted by num as `{ num, longName?, shortName?, hwModel?, batteryLevel?, lastHeard: number|null, position?: { lat, lon, alt? }, snr?, hopsAway?, viaMqtt? }[]`, from the initial sync plus live POSITION_APP updates; reset on reconnect) and `GET /api/nodes/:id` (lookup by exact node number/numeric string, then case-insensitive exact long/short name, else unique substring; 404 when ambiguous or missing). Scalar API docs are served at `/` (OpenAPI JSON at `/docs-json`). Keep this list in sync with the controller. All `POST /api/send` traffic targets channel index 3 (`test`) on purpose — never the primary channel; override via `MESH_SEND_CHANNEL` (default `3`). Message history is persisted to SQLite: file path from `MESH_DB_PATH` (default `./data/mesh-bot.db`, parent dir auto-created), poller re-upserts every `MESH_POLL_MS` ms (default 1000) and prunes the table to the newest 1000 rows. Outbound notification: when an RX text arrives on channel `MESH_NOTIFY_CHANNEL` (default `3`), the in-process `OpencodeService` notifier (`src/opencode`, registered in `MeshModule`) pushes it to opencode — fire-and-forget, no response handling. It resolves the mesh-bot workspace dynamically per request from the opencode API (`OPENCODE_API`, default `http://172.24.0.1:9898`; matched by `OPENCODE_WORKSPACE_PATH` (default `/home/krzyw/projects/sandbox/mesh-bot`), fallback name `mesh-bot`, status `ready` preferred) and posts to its most recently updated session (`time.updated`) — or the pinned `OPENCODE_SESSION` when set, which skips discovery. The message POST body is exactly `{"parts":[{"type":"text","text":msg}]}` with msg = `New mesh message[ from node ]: "" — please react to it.` (the `from node` part omitted when absent), HTTP Basic auth via `OPENCODE_USER`/`OPENCODE_PASS`; empty credentials disable the notifier entirely (one startup log line, no-op). Every call has a 15 s timeout; a read timeout on the final message POST means DELIVERED (opencode accepts immediately but holds the response until the agent turn finishes) and is treated as success without retry, while workspace/session GET failures are retried up to 3 times with backoff (2 s / 10 s / 30 s) before being dropped. Injected messages (`POST /api/messages`, channel 0) trigger it too when the notify channel is set to match. Messages from our own node are never notified (self-echo filter). 3. **MCP endpoint** — `POST /mcp` (excluded from the `/api` prefix in main.ts, like Scalar at `/`): an in-process MCP server over the official SDK's Streamable HTTP transport in stateless mode (`sessionIdGenerator: undefined`, fresh `McpServer` + transport per request). Tools map 1:1 to MeshService methods: `mesh_status`, `send_message(text, to?, replyId?)` (`to` resolves like `POST /api/send`; optional `replyId` integer in 1..0xffffffff makes the message render as a threaded reply/quote of that packet id from `GET /api/messages` or `get_messages`; errors surface as tool errors with the same messages), `list_channels`, `list_nodes`, `get_node(id)` (tool error when missing/ambiguous), `get_messages`, `inject_message(text, from?)` (synthetic inbox injection for external triggers — records an rx entry in the history as if received on the mesh; NOT sent over RF; same validation as `POST /api/messages`), `get_telemetry` (same bounded recent window as `GET /api/telemetry`), `device_info` (same redaction as `GET /api/device`). Results are JSON in a text content block; sends ride channel index 3 like everything else. Note: the SDK's root entry is broken upstream (#2273) — import via subpaths (`@modelcontextprotocol/sdk/server/mcp.js` etc.). ## Other diff --git a/CHANGELOG.md b/CHANGELOG.md index dc02195..b6316c8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,16 @@ # Changelog +## v0.14.0 — 2026-09-25 14:11 + +- The standalone Python notify forwarder (`ops/forwarder`, compose service `forwarder`) is gone: its logic now lives in-process as a new `src/opencode` module — an `@Injectable()` `OpencodeService` registered in `MeshModule` and injected into `MeshService`. When an RX text lands on channel `MESH_NOTIFY_CHANNEL` (default 3), `MeshService.notifyRx()` calls `void this.opencode.notify(text, from)` instead of the old webhook POST; the channel gate and self-echo filter stay in `MeshService`, everything else (workspace/session resolution, message POST, retries) moved into the service. +- Ported forwarder behavior 1:1 with new env names: `OPENCODE_API` (default `http://172.24.0.1:9898` — opencode binds 0.0.0.0 on the host and is reachable from the bot container over dev_net; the old forwarder used 127.0.0.1 only because it ran with `network_mode: host`), `OPENCODE_WORKSPACE_PATH` (default `/home/krzyw/projects/sandbox/mesh-bot`), `OPENCODE_SESSION` (empty = discover the most recently updated session per request; non-empty = pin that session and skip discovery), `OPENCODE_USER` / `OPENCODE_PASS` (HTTP Basic auth on every call). Missing/empty credentials disable the notifier instead of failing: one warn at construction, `notify()` is a no-op — the app keeps running. +- Workspace/session resolution per request (no caching — messages are rare): `GET {api}/api/workspaces` → match by `path == OPENCODE_WORKSPACE_PATH`, fallback name `mesh-bot`, status `ready` preferred when several match; then, unless pinned, `GET {proxyPath}/session` → most recently updated session (`time.updated`; first entry when none carry it). The message POST body is exactly `{"parts":[{"type":"text","text":msg}]}` with msg = `New mesh message[ from node ]: "" — please react to it.` (em-dash, not a hyphen; the `from node` part omitted when there is no sender). +- Timeout semantics preserved: every call has a 15 s timeout (`AbortSignal.timeout(15000)`); a read timeout on the final message POST means DELIVERED — opencode accepts immediately but holds the response until the agent turn finishes — so it is treated as success and never retried. Workspace/session GET failures are ordinary retryable failures: backoff [2 s, 10 s, 30 s] (the constant moved here from `mesh.service.ts`), up to 4 attempts total, final failure logs `opencode notify dropped after N attempts`. +- `MeshService` loses `notifyUrl`, `postNotify()` and `NOTIFY_RETRY_DELAYS_MS`; the constructor gains an optional second parameter (`db?, opencode?`) so all existing direct constructions in specs keep working — without a notifier wired, `notifyRx()` is a no-op. +- `docker-compose.yml`: forwarder service block deleted; bot environment drops `MESH_NOTIFY_URL`, keeps `MESH_NOTIFY_CHANNEL`, and passes the five `OPENCODE_*` vars via `${VAR:-default}` interpolation from `.env`. Local `.env` (gitignored) renamed `FORWARDER_USER/PASS/SESSION` → `OPENCODE_USER/PASS/SESSION` (values unchanged, session pin kept), added `OPENCODE_API=http://172.24.0.1:9898`, dropped the `MESH_NOTIFY_URL=` line. +- Tests: new `src/opencode/opencode.service.spec.ts` (12 specs — exact wrapper text with/without `from` incl. em-dash and Basic auth header, workspace path match / name fallback / ready preference, session pin vs discovery-by-max-time.updated vs first-entry fallback, POST timeout = delivered success with no retry, GET failure retried then succeeds, give up after 4 attempts with the dropped log, disabled-when-no-creds no-op + construction warn). `mesh.service.spec.ts` notify block rewritten around a mocked `OpencodeService`: channel gate (default 3), off-channel skip, injected channel-0 case incl. `from` passthrough as undefined, self-echo skip for rx and injected; the ten webhook-specific tests (URL/body/retry/HTTP-error/give-up) are gone — retry semantics now live in the opencode spec. +- Docs: README "Notify forwarder" section replaced by "Opencode notifier", env-var list updated (`MESH_NOTIFY_URL` bullet removed, `OPENCODE_*` documented); AGENTS.md webhook paragraph rewritten for the built-in notifier; MESHTASTIC.md notification rule updated. All FORWARDER_* mentions removed from tracked files (CHANGELOG history keeps them as-is). + ## v0.13.1 — 2026-09-25 13:02 - Battery-low alerts now ride channel "testowy" (index 2) instead of the pinned send channel ("test", index 3): `BatteryAlertService` reads new env `BATTERY_ALERT_CHANNEL` at construction (default `2`, same pattern as its other env config) and passes it through the existing `MeshService.sendText(to, text, replyId?, channel?)` path — which gained an optional trailing `channel` argument defaulting to the pinned `MESH_SEND_CHANNEL`, so every other sender (API, MCP, weather) is untouched. Threshold, cooldown, merged one-message-per-cycle format and @mentions are unchanged. diff --git a/package.json b/package.json index 0d6ae26..0ff71bb 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "mesh-bot", - "version": "0.13.1", + "version": "0.14.0", "description": "", "author": "", "private": true, -- 2.54.0