From 282b0c1c9fe3179029f21983ca1e09dc8071a451 Mon Sep 17 00:00:00 2001 From: Kamil Krzywda Date: Fri, 25 Sep 2026 10:20:15 +0000 Subject: [PATCH] Resolve opencode workspace + session dynamically per request in notify forwarder --- AGENTS.md | 2 +- CHANGELOG.md | 8 ++++++ README.md | 7 +++-- docker-compose.yml | 5 ++-- ops/forwarder/forwarder.py | 54 ++++++++++++++++++++++++++++++----- package.json | 2 +- src/mesh/mesh.service.spec.ts | 31 ++++++++++++++++++++ src/mesh/mesh.service.ts | 3 +- 8 files changed, 97 insertions(+), 15 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index ce31846..e244503 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"?: }` — 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. 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). +2. **Endpoints** — all under the `/api` global prefix: `GET /api/status` (`{ connected, myNodeNum, host, port }`), `POST /api/send` (body `{ "text": "...", "to"?: }` — 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. 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). 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?)` (same resolution as `POST /api/send`; 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 7c97025..b5c4a02 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,13 @@ # Changelog +## v0.12.4 — 2026-09-25 10:17 + +- Notify forwarder (`ops/forwarder`) now resolves the opencode workspace + session dynamically per request instead of hardcoding them as env defaults: `GET {FORWARDER_API}/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. Fixes lost user messages after an opencode host restart regenerated the workspace UUID (the old hardcoded base URL 404'd on every notify POST). +- New env config replaces `FORWARDER_BASE`: `FORWARDER_API` (default `http://127.0.0.1:9898`) and `FORWARDER_WORKSPACE_PATH` (default `/home/krzyw/projects/sandbox/mesh-bot`); `FORWARDER_SESSION` is now an optional pin — when set, session discovery is skipped; empty/unset discovers dynamically. `FORWARDER_USER` / `FORWARDER_PASS` stay required. Each failure mode gets a clear 502 + log line: API unreachable, workspace not found, no sessions in workspace. The message POST itself (endpoint under `{proxyPath}/session/{sid}`, body and auth), timeout=15 semantics, 202-on-timeout and the `_reply` BrokenPipe guard are unchanged. +- `docker-compose.yml`: forwarder environment block now passes `FORWARDER_API` / `FORWARDER_WORKSPACE_PATH` / `FORWARDER_SESSION` (empty default) — old `FORWARDER_BASE` line removed; bot service untouched. +- `MeshService.postNotify()`: after `await fetch(...)`, checks `res.ok` and throws `Error('HTTP ')` when not ok, so a forwarder 5xx during an opencode restart is retried with the same backoff as network failures (the `res &&` guard keeps specs that stub fetch to resolve `undefined` green). One new spec: fetch resolves status 502 twice then 200 → exactly 3 attempts, success path. All 61 pre-existing tests stay green (62 total). +- Docs: README "Notify forwarder" section rewritten for dynamic resolution + the new env vars; AGENTS.md webhook note gains a clause that the forwarder resolves the mesh-bot workspace/session from the opencode API per request (`FORWARDER_SESSION` pins a session when set). + ## v0.12.3 — 2026-09-25 09:40 - Notify webhook POSTs are now retried instead of dropped on the first failure: `notifyRx`'s fire-and-forget POST runs through a retry loop with up to 4 attempts total (initial + 3 retries) and backoff delays `[2 s, 10 s, 30 s]` (~45 s window). Each failed attempt logs WARN with the attempt number and error; after the final failure it logs WARN `notify dropped after N attempts: ...` and stops. Success on any attempt ends the loop (unchanged success behavior/logging). Fire-and-forget semantics preserved — the RX handler never awaits the loop, which runs as an un-awaited async IIFE exactly like today's single POST. Self-echo filter and channel-match logic unchanged; backoff is hardcoded (no new env vars). diff --git a/README.md b/README.md index 4362b2a..bdf60a4 100644 --- a/README.md +++ b/README.md @@ -99,12 +99,13 @@ Hourly cron job (`src/weather`): fetches current conditions for Lublin from [Ope ## Notify forwarder -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 an opencode API session (Basic auth). 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). +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). Config lives in `.env` (gitignored) and is passed via compose environment — credentials are never committed to the repo: -- `FORWARDER_BASE` (default `http://127.0.0.1:9898/workspaces/679fd6f6-b237-4a46-bf69-64cb0fc865b0/instance`) — opencode instance base URL -- `FORWARDER_SESSION` (default `ses_f31037caeffeioRwixqGV7VS6m`) — target session ID +- `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 ## Env vars diff --git a/docker-compose.yml b/docker-compose.yml index 4c86080..50339e8 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -23,8 +23,9 @@ services: build: ./ops/forwarder network_mode: host environment: - FORWARDER_BASE: ${FORWARDER_BASE:-http://127.0.0.1:9898/workspaces/679fd6f6-b237-4a46-bf69-64cb0fc865b0/instance} - FORWARDER_SESSION: ${FORWARDER_SESSION:-ses_f31037caeffeioRwixqGV7VS6m} + 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 diff --git a/ops/forwarder/forwarder.py b/ops/forwarder/forwarder.py index 69461f2..346e10d 100644 --- a/ops/forwarder/forwarder.py +++ b/ops/forwarder/forwarder.py @@ -1,6 +1,10 @@ #!/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. """ @@ -8,11 +12,43 @@ import json, base64, urllib.request from http.server import BaseHTTPRequestHandler, HTTPServer import os -BASE = os.environ.get("FORWARDER_BASE", "http://127.0.0.1:9898/workspaces/679fd6f6-b237-4a46-bf69-64cb0fc865b0/instance") -SESSION = os.environ.get("FORWARDER_SESSION", "ses_f31037caeffeioRwixqGV7VS6m") +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": @@ -22,23 +58,27 @@ class H(BaseHTTPRequestHandler): 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/{SESSION}/message", + f"{base}/session/{sid}/message", data=json.dumps({"parts": [{"type": "text", "text": msg}]}).encode(), - headers={"Content-Type": "application/json", - "Authorization": "Basic " + base64.b64encode(f"{USER}:{PASS}".encode()).decode()}, + 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 {SESSION}: HTTP {code} {out[:150]!r}", flush=True) + 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 {SESSION}: accepted (response still pending after 15 s)", flush=True) + 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()) diff --git a/package.json b/package.json index 15fd07c..dac8f4f 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "mesh-bot", - "version": "0.12.3", + "version": "0.12.4", "description": "", "author": "", "private": true, diff --git a/src/mesh/mesh.service.spec.ts b/src/mesh/mesh.service.spec.ts index be7ed20..ae8dbcd 100644 --- a/src/mesh/mesh.service.spec.ts +++ b/src/mesh/mesh.service.spec.ts @@ -824,6 +824,37 @@ describe('MeshService notify webhook', () => { } }); + 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'; diff --git a/src/mesh/mesh.service.ts b/src/mesh/mesh.service.ts index a8c1de1..36a4de8 100644 --- a/src/mesh/mesh.service.ts +++ b/src/mesh/mesh.service.ts @@ -452,11 +452,12 @@ export class MeshService implements OnModuleInit, OnModuleDestroy { const maxAttempts = NOTIFY_RETRY_DELAYS_MS.length + 1; for (let attempt = 1; ; attempt++) { try { - await fetch(this.notifyUrl, { + 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;