Merge pull request 'v0.14.0 — in-process opencode notifier replaces the forwarder service' (#35) from feat/opencode-notifier into main

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"?: <node number | node name | "!b" | "broadcast">, "replyId"?: <packet id> }` — 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: {<section>: ...}, moduleConfig: {<module>: ...} }` — 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"?: <node number> }`, 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"?: <node number | node name | "!b" | "broadcast">, "replyId"?: <packet id> }` — 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: {<section>: ...}, moduleConfig: {<module>: ...} }` — 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"?: <node number> }`, 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 <n>]: "<text>" — 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

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 <n>]: "<text>" — 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.

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).

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 <n>]: "<text>" — 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

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:
 

ops/forwarder/Dockerfile

@@ -1,4 +0,0 @@
-FROM python:3.12-slim
-WORKDIR /app
-COPY forwarder.py .
-CMD ["python3", "forwarder.py"]

ops/forwarder/forwarder.py

@@ -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()

package.json

@@ -1,6 +1,6 @@
 {
   "name": "mesh-bot",
-  "version": "0.13.1",
+  "version": "0.14.0",
   "description": "",
   "author": "",
   "private": true,

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],

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<typeof vi.fn>;
 
-  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<string, string>;
-        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<string, string>;
-        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<void> }).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<void> }).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<void> }).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', () => {

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<T>(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<string, unknown> = { 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<string, unknown>): Promise<void> {
-    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<void> {
-    return new Promise((resolve) => setTimeout(resolve, ms));
+    void this.opencode.notify(text, from);
   }
 
   /**

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<string, string> = {}): OpencodeService {
+  process.env.OPENCODE_API = API;
+  process.env.OPENCODE_WORKSPACE_PATH = WS_PATH;
+  for (const k of ['OPENCODE_SESSION', 'OPENCODE_USER', 'OPENCODE_PASS'])
+    delete process.env[k];
+  Object.assign(process.env, env);
+  return new OpencodeService();
+}
+
+function stubSleep(svc: OpencodeService): ReturnType<typeof vi.fn> {
+  const sleep = vi.fn(async (_ms: number) => {});
+  (svc as unknown as { sleep(ms: number): Promise<void> }).sleep = sleep;
+  return sleep;
+}
+
+describe('OpencodeService.notify', () => {
+  let fetchMock: ReturnType<typeof vi.fn>;
+
+  beforeEach(() => {
+    fetchMock = vi.fn();
+    vi.stubGlobal('fetch', fetchMock);
+  });
+
+  afterEach(() => {
+    vi.unstubAllGlobals();
+    vi.restoreAllMocks();
+    for (const k of [
+      'OPENCODE_API',
+      'OPENCODE_WORKSPACE_PATH',
+      'OPENCODE_SESSION',
+      'OPENCODE_USER',
+      'OPENCODE_PASS',
+    ])
+      delete process.env[k];
+  });
+
+  it('posts the exact wrapper text with `from` (em-dash) to the resolved session', async () => {
+    const svc = makeService({
+      OPENCODE_USER: 'u',
+      OPENCODE_PASS: 'p',
+      OPENCODE_SESSION: 'ses_pinned',
+    });
+    fetchMock
+      .mockResolvedValueOnce(json([WORKSPACE]))
+      .mockResolvedValueOnce(json({}));
+
+    await svc.notify('hi', 42);
+
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    const [url, init] = fetchMock.mock.calls[1];
+    expect(url).toBe(`${API}/proxy/abc/session/ses_pinned/message`);
+    const opts = init as {
+      method?: string;
+      headers?: Record<string, string>;
+      body?: string;
+    };
+    expect(opts.method).toBe('POST');
+    expect(opts.headers?.['content-type']).toBe('application/json');
+    expect(opts.headers?.authorization).toBe(
+      'Basic ' + Buffer.from('u:p').toString('base64'),
+    );
+    // body is EXACTLY {"parts":[{"type":"text","text":msg}]} — em-dash, not a hyphen
+    expect(String(opts.body)).toBe(
+      JSON.stringify({
+        parts: [
+          {
+            type: 'text',
+            text: 'New mesh message from node 42: "hi" — please react to it.',
+          },
+        ],
+      }),
+    );
+  });
+
+  it('omits the `from node` part when there is no sender', async () => {
+    const svc = makeService({
+      OPENCODE_USER: 'u',
+      OPENCODE_PASS: 'p',
+      OPENCODE_SESSION: 'ses_pinned',
+    });
+    fetchMock
+      .mockResolvedValueOnce(json([WORKSPACE]))
+      .mockResolvedValueOnce(json({}));
+
+    await svc.notify('hi');
+
+    const [, init] = fetchMock.mock.calls[1];
+    expect(String((init as { body?: string }).body)).toBe(
+      JSON.stringify({
+        parts: [
+          {
+            type: 'text',
+            text: 'New mesh message: "hi" — please react to it.',
+          },
+        ],
+      }),
+    );
+  });
+
+  it('matches the workspace by path, preferring ready entries when several match', async () => {
+    const svc = makeService({
+      OPENCODE_USER: 'u',
+      OPENCODE_PASS: 'p',
+      OPENCODE_SESSION: 'ses_pinned',
+    });
+    fetchMock.mockResolvedValueOnce(
+      json([
+        {
+          ...WORKSPACE,
+          id: 'other',
+          status: 'starting',
+          proxyPath: '/proxy/not-ready',
+        }, // same path but not ready — listed first
+        WORKSPACE,
+        {
+          id: 'x',
+          name: 'mesh-bot',
+          status: 'ready',
+          proxyPath: '/proxy/other',
+        }, // different path — must lose to the path match
+      ]),
+    );
+    fetchMock.mockResolvedValueOnce(json({}));
+
+    await svc.notify('hi');
+
+    expect(fetchMock.mock.calls[1][0]).toBe(
+      `${API}/proxy/abc/session/ses_pinned/message`,
+    );
+  });
+
+  it('falls back to the workspace name when no path matches', async () => {
+    const svc = makeService({
+      OPENCODE_USER: 'u',
+      OPENCODE_PASS: 'p',
+      OPENCODE_SESSION: 'ses_pinned',
+    });
+    fetchMock.mockResolvedValueOnce(
+      json([
+        {
+          id: 'a',
+          path: '/elsewhere',
+          name: 'other',
+          status: 'ready',
+          proxyPath: '/proxy/a',
+        },
+        {
+          id: 'b',
+          path: '/renamed',
+          name: 'mesh-bot',
+          status: 'ready',
+          proxyPath: '/proxy/b',
+        },
+      ]),
+    );
+    fetchMock.mockResolvedValueOnce(json({}));
+
+    await svc.notify('hi');
+
+    expect(fetchMock.mock.calls[1][0]).toBe(
+      `${API}/proxy/b/session/ses_pinned/message`,
+    );
+  });
+
+  it('discovers the most recently updated session when none is pinned', async () => {
+    const svc = makeService({ OPENCODE_USER: 'u', OPENCODE_PASS: 'p' });
+    fetchMock
+      .mockResolvedValueOnce(json([WORKSPACE]))
+      .mockResolvedValueOnce(
+        json([
+          { id: 'ses_old', time: { updated: 100 } },
+          { id: 'ses_new', time: { updated: 200 } },
+        ]),
+      )
+      .mockResolvedValueOnce(json({}));
+
+    await svc.notify('hi');
+
+    expect(fetchMock.mock.calls[1][0]).toBe(`${API}/proxy/abc/session`);
+    expect(fetchMock.mock.calls[2][0]).toBe(
+      `${API}/proxy/abc/session/ses_new/message`,
+    );
+  });
+
+  it('uses the first session when none carry time.updated', async () => {
+    const svc = makeService({ OPENCODE_USER: 'u', OPENCODE_PASS: 'p' });
+    fetchMock
+      .mockResolvedValueOnce(json([WORKSPACE]))
+      .mockResolvedValueOnce(json([{ id: 'ses_a' }, { id: 'ses_b', time: {} }]))
+      .mockResolvedValueOnce(json({}));
+
+    await svc.notify('hi');
+
+    expect(fetchMock.mock.calls[2][0]).toBe(
+      `${API}/proxy/abc/session/ses_a/message`,
+    );
+  });
+
+  it('skips session discovery entirely when a session is pinned', async () => {
+    const svc = makeService({
+      OPENCODE_USER: 'u',
+      OPENCODE_PASS: 'p',
+      OPENCODE_SESSION: 'ses_pinned',
+    });
+    fetchMock
+      .mockResolvedValueOnce(json([WORKSPACE]))
+      .mockResolvedValueOnce(json({}));
+
+    await svc.notify('hi');
+
+    // workspaces GET + message POST only — no /session discovery GET
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(fetchMock.mock.calls[0][0]).toBe(`${API}/api/workspaces`);
+    expect(fetchMock.mock.calls[1][0]).toBe(
+      `${API}/proxy/abc/session/ses_pinned/message`,
+    );
+  });
+
+  it('treats a final-POST read timeout as delivered success and does not retry', async () => {
+    const svc = makeService({
+      OPENCODE_USER: 'u',
+      OPENCODE_PASS: 'p',
+      OPENCODE_SESSION: 'ses_pinned',
+    });
+    fetchMock
+      .mockResolvedValueOnce(json([WORKSPACE]))
+      .mockRejectedValueOnce(timeoutError()); // the message POST times out — opencode holds the response until the turn finishes
+
+    const logSpy = vi.spyOn(Logger.prototype, 'log');
+    await expect(svc.notify('hi')).resolves.toBeUndefined();
+
+    // exactly 2 calls: no retry after a timeout-as-delivered
+    expect(fetchMock).toHaveBeenCalledTimes(2);
+    expect(logSpy.mock.calls.map((c) => String(c[0]))).toContain(
+      'opencode notify accepted (response still pending after 15 s)',
+    );
+  });
+
+  it('retries a failed workspace GET and succeeds on the next attempt', async () => {
+    const svc = makeService({
+      OPENCODE_USER: 'u',
+      OPENCODE_PASS: 'p',
+      OPENCODE_SESSION: 'ses_pinned',
+    });
+    const sleep = stubSleep(svc);
+    fetchMock
+      .mockRejectedValueOnce(new Error('boom')) // attempt 1: workspaces GET fails
+      .mockResolvedValueOnce(json([WORKSPACE])) // attempt 2: workspaces ok
+      .mockResolvedValueOnce(json({})); // message POST ok
+
+    await expect(svc.notify('hi')).resolves.toBeUndefined();
+
+    expect(fetchMock).toHaveBeenCalledTimes(3);
+    expect(sleep).toHaveBeenNthCalledWith(1, 2_000);
+  });
+
+  it('gives up after 4 attempts and logs opencode notify dropped', async () => {
+    const svc = makeService({ OPENCODE_USER: 'u', OPENCODE_PASS: 'p' });
+    const sleep = stubSleep(svc);
+    fetchMock.mockRejectedValue(new Error('down'));
+    const warnSpy = vi.spyOn(Logger.prototype, 'warn');
+
+    await expect(svc.notify('hi')).resolves.toBeUndefined(); // never throws — fire-and-forget safe
+
+    // one workspaces GET per attempt: exactly 4 attempts total
+    expect(fetchMock).toHaveBeenCalledTimes(4);
+    expect(sleep).toHaveBeenNthCalledWith(1, 2_000);
+    expect(sleep).toHaveBeenNthCalledWith(2, 10_000);
+    expect(sleep).toHaveBeenNthCalledWith(3, 30_000);
+    const warns = warnSpy.mock.calls.map((c) => String(c[0]));
+    expect(warns.some((w) => w.includes('attempt 1/4'))).toBe(true);
+    // the GET failure is wrapped with the URL for a clear reason — match prefix + cause
+    expect(
+      warns.some(
+        (w) =>
+          w.startsWith('opencode notify dropped after 4 attempts:') &&
+          w.includes(': down'),
+      ),
+    ).toBe(true);
+  });
+
+  it('is a no-op without credentials (empty or unset)', async () => {
+    for (const env of [{}, { OPENCODE_USER: '', OPENCODE_PASS: 'p' }] as Record<
+      string,
+      string
+    >[]) {
+      const svc = makeService(env);
+      await expect(svc.notify('hi', 42)).resolves.toBeUndefined();
+      expect(fetchMock).not.toHaveBeenCalled();
+    }
+  });
+
+  it('logs once at construction when disabled', () => {
+    const warnSpy = vi.spyOn(Logger.prototype, 'warn');
+    makeService({}); // no creds
+    expect(warnSpy.mock.calls.map((c) => String(c[0]))).toContain(
+      'opencode notifier disabled: OPENCODE_USER/OPENCODE_PASS not set',
+    );
+  });
+});

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<void> {
+    if (!this.enabled) return;
+    const msg =
+      'New mesh message' +
+      (from != null ? ` from node ${from}` : '') +
+      `: "${text}" — please react to it.`;
+    const maxAttempts = RETRY_DELAYS_MS.length + 1;
+    for (let attempt = 1; ; attempt++) {
+      try {
+        const { base, sid } = await this.resolveTarget();
+        await this.postMessage(base, sid, msg);
+        return; // success on any attempt — a POST timeout already resolved as delivered inside postMessage
+      } catch (err: unknown) {
+        const message = err instanceof Error ? err.message : String(err);
+        if (attempt === maxAttempts) {
+          this.logger.warn(
+            `opencode notify dropped after ${attempt} attempts: ${message}`,
+          );
+          return;
+        }
+        this.logger.warn(
+          `opencode notify failed (attempt ${attempt}/${maxAttempts}): ${message}`,
+        );
+        await this.sleep(RETRY_DELAYS_MS[attempt - 1]);
+      }
+    }
+  }
+
+  /** Injectable so specs can skip the backoff delays. */
+  private sleep(ms: number): Promise<void> {
+    return new Promise((resolve) => setTimeout(resolve, ms));
+  }
+
+  private authHeaders(): Record<string, string> {
+    return {
+      authorization:
+        'Basic ' + Buffer.from(`${this.user}:${this.pass}`).toString('base64'),
+    };
+  }
+
+  /** Resolve (base_url, session_id) fresh per request; throws with a clear reason on API/workspace/session failure. */
+  private async resolveTarget(): Promise<{ base: string; sid: string }> {
+    const workspaces = await this.getJson<WorkspaceEntry[]>(
+      `${this.api}/api/workspaces`,
+    );
+    // match by path first, fall back to name — the workspace UUID is regenerated on host restarts
+    const matches =
+      workspaces.filter((w) => w.path === this.workspacePath).length > 0
+        ? workspaces.filter((w) => w.path === this.workspacePath)
+        : workspaces.filter((w) => w.name === 'mesh-bot');
+    // prefer ready entries when several match
+    const pool =
+      matches.filter((w) => w.status === 'ready').length > 0
+        ? matches.filter((w) => w.status === 'ready')
+        : matches;
+    if (pool.length === 0)
+      throw new Error(
+        `workspace not found: path=${this.workspacePath} name=mesh-bot`,
+      );
+    const ws = pool[0];
+    const base = this.api + (ws.proxyPath ?? '');
+    let sid = this.sessionOverride;
+    if (!sid) {
+      const sessions = await this.getJson<SessionEntry[]>(`${base}/session`);
+      if (sessions.length === 0)
+        throw new Error(
+          `no sessions in workspace ${ws.id} (${this.workspacePath})`,
+        );
+      // most recently updated session; first entry when none carry time.updated
+      const updated = (s: SessionEntry): number => s.time?.updated ?? 0;
+      sid = sessions.some((s) => updated(s))
+        ? sessions.reduce((a, b) => (updated(b) > updated(a) ? b : a)).id
+        : sessions[0].id;
+    }
+    return { base, sid };
+  }
+
+  private async getJson<T>(url: string): Promise<T> {
+    let res: Response;
+    try {
+      res = await fetch(url, {
+        headers: this.authHeaders(),
+        signal: AbortSignal.timeout(TIMEOUT_MS),
+      });
+    } catch (err) {
+      throw new Error(
+        `opencode API unreachable at ${url}: ${(err as Error).message}`,
+      );
+    }
+    if (!res.ok) throw new Error(`opencode API HTTP ${res.status} at ${url}`);
+    return (await res.json()) as T;
+  }
+
+  private async postMessage(
+    base: string,
+    sid: string,
+    msg: string,
+  ): Promise<void> {
+    try {
+      const res = await fetch(`${base}/session/${sid}/message`, {
+        method: 'POST',
+        headers: { ...this.authHeaders(), 'content-type': 'application/json' },
+        body: JSON.stringify({ parts: [{ type: 'text', text: msg }] }),
+        signal: AbortSignal.timeout(TIMEOUT_MS),
+      });
+      if (!res.ok) throw new Error(`HTTP ${res.status}`); // HTTP errors retry like network failures
+    } catch (err) {
+      // a read timeout here means DELIVERED, not failed — opencode accepted the message and is holding
+      // the response until the agent turn finishes; never retried
+      if ((err as Error).name === 'TimeoutError') {
+        this.logger.log(
+          `opencode notify accepted (response still pending after ${TIMEOUT_MS / 1000} s)`,
+        );
+        return;
+      }
+      throw err;
+    }
+  }
+}