Compare commits

...
26 Commits
Author SHA1 Message Date
kamilkrzywda b54f909fbc Merge pull request 'Ignore 0% battery readings in low-battery alerting' (#55) from fix/battery-zero-ignore into main
CI / build + unit (push) Successful in 1m10s
2026-10-06 12:53:41 +02:00
kamilkrzywda 7ec97a06d2 Ignore 0% battery readings in low-battery alerting
CI / build + unit (pull_request) Successful in 1m18s
A 0% batteryLevel is an invalid/no-data value (no battery wired or
misconfigured node), not a genuinely critical battery, so it must not
produce a "⚠️ battery low" alert. Add a non-positive guard to
shouldAlert after the undefined check: 0 (and any non-positive) now
skips, while 1-20 still alert and >20 still doesn't. Threshold/cooldown/
batching unchanged.

Tests: 3 new specs in the check-threshold block (0% excluded from the
batch, lone 0% sends nothing, 1% still alerts). 180 -> 183.
2026-10-06 08:02:37 +00:00
kamilkrzywda 84915fe314 Merge pull request 'fix(notifier): migrate opencode notifier to v2 routes (v0.21.11)' (#54) from fix/opencode-v2-notifier into main
CI / build + unit (push) Successful in 22s
2026-10-05 15:25:38 +02:00
kamilkrzywda 5506fb9330 fix(notifier): migrate opencode notifier to v2 routes (v0.21.11)
CI / build + unit (pull_request) Successful in 46s
CodeNomad's per-workspace instance proxy was upgraded and now only
serves opencode v2 /api/* routes; the v1 /session/... routes 403, which
broke all notifications.

- Discovery: GET {api}{proxyPath}/api/session (v2 {data,cursor} envelope)
  instead of GET {base}/session; sessions filtered by location.directory
  and the max time.updated entry is picked.
- Send: POST {api}{proxyPath}/api/session/{sid}/prompt with a {text}
  body instead of POST {base}/session/{id}/message with {parts:[...]}.
- v2 returns 2xx immediately, so the 15 s read-timeout-as-delivered
  branch is kept only as a harmless safety net.
- Updated the spec to the v2 routes/body/envelope (incl. a
  directory-filter assertion) and kept AGENTS.md in sync.
2026-10-05 12:48:47 +00:00
kamilkrzywda beebcd5851 Merge pull request 'Container TZ: Europe/Warsaw so cron hours are local (v0.21.10)' (#53) from fix/tz-warsaw into main
CI / build + unit (push) Successful in 1m16s
2026-09-30 07:49:50 +02:00
kamilkrzywda 5a1493fb4a Run container in Europe/Warsaw so cron hours and logs use local Polish time (v0.21.10)
CI / build + unit (pull_request) Successful in 50s
2026-09-30 05:47:10 +00:00
kamilkrzywda 8fb16217f7 Merge pull request 'Add #loser command — worst packet delivery record (v0.21.9)' (#52) from fix/loser-command into main
CI / build + unit (push) Successful in 1m36s
2026-09-29 23:01:29 +02:00
kamilkrzywda 3b1f5de101 Add #loser command — worst packet delivery record over directed sends (v0.21.9)
CI / build + unit (pull_request) Successful in 48s
2026-09-29 20:58:43 +00:00
kamilkrzywda 46d32b3553 Merge pull request 'Fix unhandled ECONNRESET in fake-bridge teardown — npm test exits 0 again (v0.21.8)' (#51) from fix/spec-teardown-econnreset into main
CI / build + unit (push) Successful in 1m11s
2026-09-27 09:18:09 +02:00
kamilkrzywda e20e11e2d5 Fix npm test exit 1 — uncaught ECONNRESET from fake-bridge sockets without error handlers (v0.21.8)
CI / build + unit (pull_request) Successful in 46s
2026-09-27 07:13:56 +00:00
kamilkrzywda 0b20df242c Merge pull request 'Fix double ScheduleModule.forRoot() — all cron jobs fired twice (v0.21.7)' (#50) from fix/schedule-double-registration into main
CI / build + unit (push) Failing after 34s
2026-09-27 08:52:35 +02:00
kamilkrzywda a16b6c19d4 Fix double ScheduleModule.forRoot() registration doubling every cron/interval job (v0.21.7)
CI / build + unit (pull_request) Failing after 22s
2026-09-27 06:51:50 +00:00
kamilkrzywda 9e558fd95e Merge pull request 'Add #help command with short list of all commands (v0.21.6)' (#49) from fix/help-command into main
CI / build + unit (push) Failing after 34s
2026-09-26 12:54:06 +02:00
kamilkrzywda 70be8bdaa7 Add #help command — short single-message list of all commands (v0.21.6)
CI / build + unit (pull_request) Failing after 21s
Owner request: 'now add #help with short list of all command'. Registered in
the CommandService constructor next to #ping; returns a SINGLE multi-line
string (one packet, ~203 chars) — not an array, which would drip-feed one
spaced packet per line through the v0.21.5 MESH_MULTI_SEND_DELAY_MS loop.

Also hardened the v0.21.5 multi-send-spacing spec: its wall-clock assertion
('first send < 8 ms after t0') flaked under full-suite load — replaced with
fake timers, which assert deterministically that line 1 is out before any time
elapses and lines 2-3 land exactly one delay apart.
2026-09-26 10:52:18 +00:00
kamilkrzywda 847c3d47d0 Merge pull request 'Space multi-line command replies 5 s apart (v0.21.5)' (#48) from fix/multi-send-spacing into main
CI / build + unit (push) Failing after 46s
2026-09-26 12:31:22 +02:00
kamilkrzywda 1b768b14a7 Space multi-line command replies MESH_MULTI_SEND_DELAY_MS apart (v0.21.5)
CI / build + unit (pull_request) Failing after 21s
Production incident: '#rss get lublin' returned an array of lines and
CommandService sent each one back-to-back within the same second — on the real
mesh 3 of 4 items were dropped in RF (the owner saw only the last one).

CommandService.onRx now waits MESH_MULTI_SEND_DELAY_MS (new env, default 5000 =
5 s, read at construction like the other timing knobs) between consecutive
lines of an array reply — no delay before the first line. The delay sits in the
shared per-line loop, so both the respond() path and the legacy sendText
fallback are covered by one await.
2026-09-26 10:29:21 +00:00
kamilkrzywda 3f6c02d71e Merge pull request 'LLM timeout default 60 s → 180 s (v0.21.4)' (#47) from fix/llm-timeout into main
CI / build + unit (push) Failing after 38s
2026-09-26 11:55:54 +02:00
kamilkrzywda 98b9ec5185 LLM_TIMEOUT_MS default 60 s -> 3 min (v0.21.4)
CI / build + unit (pull_request) Failing after 21s
Production incident: '#rss get lublin' at 9:41 AM failed with 'The operation
was aborted due to timeout' — the LlmService call hit its 60 s cap while the
local OpenAI-compatible server (qwen3.8-27b, uncontrollable reasoning tokens)
was slow/busy. The same endpoint answers a trivial prompt in ~1 s when idle,
so this is latency variance on the server side, not a dead endpoint.

Bump the LLM_TIMEOUT_MS fallback from 60000 to 180000 (3 min headroom for
slow reasoning periods); still env-overridable, only the default changed.
2026-09-26 09:52:18 +00:00
kamilkrzywda 0722c82359 Merge pull request 'Reconnect tuning: 1 min idle timeout + 5 min periodic autoreconnect (v0.21.3)' (#46) from fix/reconnect-tuning into main
CI / build + unit (push) Failing after 42s
2026-09-26 11:37:41 +02:00
kamilkrzywda 1ae8d7a926 Idle timeout default 60 s; lazy periodic autoreconnect every 5 min (v0.21.3)
CI / build + unit (pull_request) Failing after 22s
The v0.21.2 frame-based watchdog works in production (it fired today: 'no
decoded bridge frames for 628474 ms — forcing reconnect', and the reconnect
delivered queued packets), but the bridge/radio side at 192.168.1.4 keeps
killing inbound RX within ~20 min of a fresh sync, repeatedly. Two tuning
changes requested by the owner:

- MESH_IDLE_TIMEOUT_MS default lowered from 600000 (10 min) to 60000 (1 min).
- New lazy periodic autoreconnect: MESH_RECONNECT_INTERVAL_MS (new env,
  default 300000 = 5 min, read at construction like the others). An interval
  started in onModuleInit unconditionally logs and destroys the socket so the
  existing close -> scheduleReconnect loop re-establishes with a fresh sync on
  schedule; cleared in onModuleDestroy. When not connected at tick time,
  destroy is a no-op and the retry loop owns recovery — no special-casing.

Tests: 171 -> 172 — periodic reconnect observed end-to-end with an overridden
short interval (fake bridge disconnect + re-establish); existing watchdog tests
pass unchanged, comments updated to the new 60 s default.
2026-09-26 09:34:30 +00:00
kamilkrzywda 77459adc13 Merge pull request 'Watchdog: track decoded frames, not raw bytes (v0.21.2)' (#45) from fix/watchdog-frame-tracking into main
CI / build + unit (push) Failing after 21s
2026-09-26 10:53:28 +02:00
kamilkrzywda 33a88f3d83 Watchdog tracks decoded frames, not raw bytes; buffer-overflow backstop (v0.21.2)
CI / build + unit (pull_request) Failing after 39s
Second production zombie slipped past v0.21.1's watchdog: text+telemetry RX
dead for >1 h while TX worked and the watchdog never fired. Root cause proven
at runtime via /proc/net/snmp — InSegs delta = 2 segments per 30 s (OutSegs 0):
the bridge peer keeps dribbling small NON-frame bytes (~every 15 s, keepalive
noise or a desynced stream) and v0.21.1 updated lastPacketAt on EVERY socket
chunk, so 'no traffic for MESH_IDLE_TIMEOUT_MS' was never true even though zero
valid Meshtastic frames decoded.

- The 'data' handler no longer updates lastPacketAt per chunk; it is updated
  only when extractFrames yields >=1 valid packet (the reset in the 'connect'
  handler is unchanged), so a stream that decodes nothing trips the watchdog
  after MESH_IDLE_TIMEOUT_MS even while raw bytes keep arriving.
- Buffer-overflow backstop: if the undecoded frame buffer exceeds
  MESH_MAX_FRAME_BUFFER (new env, default 256000) — e.g. one data event of a
  desynced stream that never forms frames — log ERROR and destroy the socket;
  the existing close -> scheduleReconnect loop self-heals it. The undecoded
  buffer is also reset on every (re)connect so leftover garbage from a dead
  session can't re-trigger the backstop or corrupt the first frames of a
  healthy new stream.

Tests: 169 -> 171 — raw non-frame bytes do NOT keep the watchdog alive
(garbage-only stream forces disconnect + reconnect, today's incident shape),
and an undecoded chunk over MESH_MAX_FRAME_BUFFER triggers destroy + a fresh
session that decodes normally (cap env-overridden to 100 since Node delivers
socket data in ~64 KB events, far below the 256_000 default).
2026-09-26 08:50:27 +00:00
kamilkrzywda 306756c63b Merge pull request 'Bridge idle watchdog + forced reconnect (v0.21.1)' (#44) from fix/bridge-idle-reconnect into main
CI / build + unit (push) Successful in 45s
2026-09-26 08:53:06 +02:00
kamilkrzywda af1c77fd54 Force reconnect when the bridge connection goes silent (v0.21.1)
CI / build + unit (pull_request) Successful in 1m8s
Production incident: the bridge TCP connection went zombie — outbound sends
kept working for ~10 h while inbound packets stopped entirely, and
/api/status stayed connected:true because nothing detects inbound silence on
an established connection.

- Track lastPacketAt: updated on every chunk received from the bridge in the
  lowest-level socket 'data' path (any frame kind, decodable or not) and reset
  when a connection successfully starts, so a fresh connection can't
  self-trigger.
- Idle watchdog: a 30 s interval checks connected AND now - lastPacketAt >
  MESH_IDLE_TIMEOUT_MS (new env, default 600000 = 10 min) — logs WARN with the
  idle duration and destroys the socket so the EXISTING connect/retry loop
  re-establishes it with a fresh sync. No changes to connect-error retry or
  backoff; this only adds detection for established-but-silent connections.
- TCP keepalive: socket.setKeepAlive(true, 30_000) on the bridge socket —
  cheap insurance against dead peers at the TCP level.

Tests (fake-bridge based): silence past an overridden timeout forces a
disconnect + reconnect; inbound traffic keeps resetting the idle clock so no
false trigger; the default timeout does not fire in a test window; and
setKeepAlive(true, 30_000) is asserted on the socket. checkIdle() is public so
tests drive a tick without waiting for the real interval (like flushHistory()).
2026-09-26 06:50:01 +00:00
kamilkrzywda 73ce49c741 Merge pull request '#rss get: 3-5 separate messages with short headers (v0.21.0)' (#43) from feat/rss-multi-message into main
CI / build + unit (push) Successful in 41s
2026-09-25 22:12:05 +02:00
kamilkrzywda cdd52a3ddb Send #rss get summaries as 3-5 separate mesh messages (v0.21.0)
CI / build + unit (pull_request) Successful in 1m20s
The single-message summary was too small (owner request). The summarizer
prompt now asks for exactly 3-5 lines, one item per line, in the format
'<short header>: <one or two short sentences>' with a <=20-char header,
no intro/outro/numbering. maxTokens stays 2500 (reasoning-token headroom).

The reply is parsed into items: drop empty lines and lines without a ':'
header separator, cap at 5; fewer than 3 valid lines sends what we got
(no failure/retry), zero valid lines sends nothing. Each item goes out as
its own mesh message via respond(id, line) - all threaded to the trigger
packet id on its channel. Per-item backstop fitOneLine() hard-cuts any
line >185 chars at a word boundary near 180 so every item fits one packet;
the old whole-reply fitOnePacket() cut is gone.

CommandService sends array replies line by line through the same routing
as single replies (respond per line, or legacy sendText fallback per line
when there is no packet id). LLM failure path unchanged: single
'summary failed' reply.
2026-09-25 20:09:16 +00:00
18 changed files with 886 additions and 124 deletions
+4 -4
View File
File diff suppressed because one or more lines are too long
+68
View File
@@ -1,5 +1,73 @@
# Changelog # Changelog
## v0.21.12 — 2026-10-06 08:01
- Battery monitoring now ignores a 0% reading: a 0 (or any non-positive) `batteryLevel` is treated as an invalid/no-data value (node has no battery wired or is misconfigured) and no longer produces a "⚠️ battery low" alert, while genuinely-low readings (1–20%) still alert exactly as before. One-line guard added to `BatteryAlertService.shouldAlert` right after the `undefined` check. Tests: 3 new specs in the `check threshold` block — a 0% node is excluded from the batch (and a lone 0% sends nothing) while a 1% node still alerts.
## v0.21.11 — 2026-10-05 14:47
- Migrated the opencode notifier to the v2 API routes: discovery is now `GET {api}{proxyPath}/api/session` (a `{ data, cursor }` envelope; sessions filtered by `location.directory`, most-recent `time.updated` wins) and send is now `POST {api}{proxyPath}/api/session/{sid}/prompt` with a `{text}` body, instead of the v1 `GET /session` + `POST /session/{id}/message` (`{parts:[{type:"text",text}]}`) routes. CodeNomad's per-workspace instance proxy was upgraded and now only serves opencode v2 `/api/*` — every v1 `/session/…` call 403s, which was breaking all notifications. v2 returns 2xx immediately (no longer holds the response until the turn finishes), so the 15 s read-timeout-as-delivered branch is kept only as a harmless safety net.
## v0.21.10 — 2026-09-30 05:46
- Container now runs with `TZ=Europe/Warsaw` (new line in the bot service's `environment:` block, hardcoded like MESH_HOST/MESH_PORT): scheduled jobs and log timestamps use local Polish time instead of UTC — the weather cron (`WEATHER_CRON=0 6,12,18 * * *`) previously fired at 06/12/18 UTC = 08/14/20 CEST. DST-aware via the IANA zone (CEST +2 / CET +1). Deployment-config only — no application code changed; takes effect on the next `docker compose up -d --build`.
## v0.21.9 — 2026-09-29 20:56
- New `#loser` command (owner request): replies with which user/node has the worst packet delivery record, over directed sends only. Meshtastic shares no mesh-wide loss stats, so it groups our own tx history by destination and counts delivered vs failed per node; broadcasts are excluded (firmware strips want_ack pre-TX, they stay pending — and the broadcast addr is never a real destination even if a NACK lands on it). Loser = most failures, tie-break worse failure ratio. Node resolved to `longName`/`shortName`, falling back to the `@!xxxxxxxx` mention when unknown. Three reply shapes: worst delivery line (`worst delivery: <name> (@!hex) - N failed / M delivered`), a "no failed sends yet" line, and a "not enough data yet — only directed sends count (broadcasts have no acks)" line. Registered in the CommandService constructor next to `#ping`; listed in `#help`.
- Tests: 175 → 180 — new `#loser` specs cover all three reply cases, broadcast/pending exclusion, the unknown-node `@!hex` fallback, and the failure-ratio tie-break; the `#help` spec now also asserts `#loser` is listed.
## v0.21.8 — 2026-09-27 07:13
- Fixed `npm test` exiting 1 with "Errors 2" even though all tests pass (every CI run red since v0.21.2): the two watchdog specs that force a reconnect destroy the service's socket while data is still pending in its receive buffer (the overflow spec floods 300 KB and destroys on the first ~64 KB chunk; the garbage spec destroys right after its last write), so the kernel sends RST instead of FIN and the fake bridge's accepted server-side sockets emit `read ECONNRESET`. Those test-owned sockets had no `'error'` handler, so Node threw it as an uncaught exception that vitest counts as a run-level failure. `startBridge()` now attaches a no-op `'error'` handler to every accepted socket (single registration point, covers all four spec files using the fake bridge). Test-only change — `fake-bridge.ts` is imported by specs only; no runtime diff, no deploy needed.
## v0.21.7 — 2026-09-27 06:50
- Fixed double registration of scheduled jobs: `ScheduleModule.forRoot()` was called from both `WeatherModule` and `BatteryAlertModule`, so Nest created two ScheduleExplorers, each scanning all providers — every @Cron/@Interval job (hourly weather broadcast, per-minute battery check) fired twice. Removed the call from `battery.module.ts`; the single registration in `weather.module.ts` remains.
- Tests: 174 → 175 — new regression spec (`src/schedule-singleton.spec.ts`) asserts `ScheduleModule.forRoot(` appears exactly once across all `src/**/*.module.ts`.
## v0.21.6 — 2026-09-26 10:37
- New `#help` command (owner request): replies with a short list of all available commands as ONE multi-line message (single packet, not one spaced packet per line) — `#ping`, the `#rss help/add/remove/get/sources/tags` family. Registered in the CommandService constructor next to `#ping`.
- Tests: 173 → 174 — `#help` dispatches through respond() exactly once with text containing both `#ping` and `#rss`. (Also hardened the v0.21.5 multi-send-spacing spec, which flaked under full-suite load: it now uses fake timers to assert deterministically that line 1 goes out before any time elapses and lines 2-3 land exactly one delay apart.)
## v0.21.5 — 2026-09-26 10:26
- Multi-line command replies are now spaced apart: `CommandService.onRx` waits `MESH_MULTI_SEND_DELAY_MS` (new env, default **5000** = 5 s, read at construction) between consecutive lines of an array reply — no delay before the first line. Production incident: `#rss get lublin` sent its items back-to-back within the same second and 3 of 4 were dropped in RF (the owner saw only the last one); each packet now gets time to land.
- Tests: 172 → 173 — a 3-line reply with an overridden short delay asserts all lines are sent, the first goes out immediately (no pre-delay), and each subsequent line lands ≥ the configured gap after the previous one; existing specs pin `MESH_MULTI_SEND_DELAY_MS=0` in `beforeEach` so they stay fast.
## v0.21.4 — 2026-09-26 09:51
- `LLM_TIMEOUT_MS` default raised from 60000 (60 s) to **180000** (3 min): production incident — `#rss get lublin` at 9:41 AM failed with "The operation was aborted due to timeout" because the local OpenAI-compatible server (qwen3.8-27b, uncontrollable reasoning tokens) was slow/busy on that call; the same endpoint answers a trivial prompt in ~1 s when idle, so this is latency variance, not a dead endpoint. Still env-overridable (`LLM_TIMEOUT_MS`), only the fallback changed.
## v0.21.3 — 2026-09-26 09:31
- Idle timeout default lowered from 10 min to **1 min** (`MESH_IDLE_TIMEOUT_MS` fallback `600000` → `60000`) — the v0.21.2 frame-based watchdog works in production (it fired today and the reconnect delivered queued packets), but the bridge/radio side at 192.168.1.4 keeps killing inbound RX within ~20 min of a fresh sync, so a shorter detection window is wanted.
- New **lazy periodic autoreconnect**: `MESH_RECONNECT_INTERVAL_MS` (new env, default **300000** = 5 min, read at construction) — an interval that unconditionally logs ("periodic reconnect — MESH_RECONNECT_INTERVAL_MS elapsed") and destroys the socket so the existing close→scheduleReconnect loop re-establishes with a fresh sync on schedule. When not connected at tick time, destroy is a no-op and the retry loop owns recovery (no special-casing). Started in `onModuleInit`, cleared in `onModuleDestroy`.
- Tests: 171 → 172 — periodic reconnect observed end-to-end with an overridden short interval (fake bridge disconnect + re-establish); existing watchdog tests pass unchanged, comments updated to the new 60 s default.
## v0.21.2 — 2026-09-26 08:47
- Idle watchdog now tracks DECODED FRAMES, not raw bytes (fixes a second production zombie that slipped past v0.21.1): text+telemetry RX was dead for >1 h while TX worked and the watchdog never fired — /proc/net/snmp showed InSegs delta = 2 segments per 30 s, i.e. the bridge peer keeps dribbling small NON-frame bytes (~every 15 s) that kept resetting `lastPacketAt` on every socket chunk even though zero valid Meshtastic frames decoded. The `'data'` handler no longer updates `lastPacketAt` per chunk; it is updated only when `extractFrames` yields ≥1 valid packet (the reset in the `'connect'` handler is unchanged), so a stream that decodes nothing trips the watchdog after `MESH_IDLE_TIMEOUT_MS` even while raw bytes keep arriving.
- Buffer-overflow backstop: if the undecoded frame buffer exceeds `MESH_MAX_FRAME_BUFFER` (new env, default **256000** = 250 KB) the service logs ERROR ("frame buffer overflow — stream desync, forcing reconnect") and destroys the socket → existing close→scheduleReconnect self-heals a desynced stream. The undecoded buffer is also reset on every (re)connect so leftover garbage from a dead session can't re-trigger the backstop or corrupt the first frames of a healthy new stream.
- Tests: 169 → 171 — raw non-frame bytes do NOT keep the watchdog alive (garbage-only stream forces disconnect + reconnect, exactly today's incident shape), and an undecoded chunk over `MESH_MAX_FRAME_BUFFER` triggers destroy + a fresh session that decodes normally. The existing valid-frames-reset-the-clock test now exercises frame-based tracking unchanged.
## v0.21.1 — 2026-09-26 06:47
- Bridge zombie-connection fix (production incident: outbound sends kept working ~10 h while inbound packets stopped entirely and `/api/status` stayed `connected:true`). New **idle watchdog**: the service tracks `lastPacketAt`, updated on EVERY chunk received from the bridge in the lowest-level socket `'data'` path (any frame kind, decodable or not) and reset when a connection successfully starts. A 30 s interval checks: if connected AND no inbound traffic for `MESH_IDLE_TIMEOUT_MS` (new env, default **600000** = 10 min), it logs WARN ("no bridge traffic for X ms — forcing reconnect") and destroys the socket — the EXISTING connect/retry loop (`MESH_RECONNECT_MS`) then re-establishes with a fresh sync. No changes to connect-error retry/backoff behavior; this only adds detection for established-but-silent connections.
- TCP keepalive on the bridge socket: `socket.setKeepAlive(true, 30_000)` — cheap insurance against dead peers at the TCP level, complementing the application-level watchdog.
- Tests: 165 → 169 — silence past an overridden timeout forces disconnect + reconnect (fake bridge), inbound traffic keeps resetting the idle clock so no false trigger, default timeout does not fire in a test window, and `setKeepAlive(true, 30_000)` is asserted on the socket.
## v0.21.0 — 2026-09-25 19:59
- `#rss get <name|#tag>` now replies with **3–5 SEPARATE mesh messages** instead of one small summary (owner request: the single-message summary was too small). Each item is its own packet, so no fragmentation — and each rides the trigger's channel threaded as a quote of the command packet.
- New summarizer prompt asks for exactly 3-5 lines, one item per line, format `<short header>: <one or two short sentences>` with a ≤20-char header (e.g. `Culture: Museum opens new wing`), no intro/outro/numbering. maxTokens stays 2500 (reasoning-token headroom — see v0.20.1).
- The LLM reply is parsed into items: split on newlines, trim, drop empty lines and lines without a `:` header separator, cap at 5. Fewer than 3 valid lines → send what we got (no failure, no retry — the model is non-deterministic); zero valid lines → no message sent.
- Per-item backstop replaces the old whole-reply cut: any single line >185 chars is hard-cut at a word boundary near 180 (`fitOneLine`), so every item fits one packet even if the model ignores the prompt. `fitOnePacket()` (whole-reply ≤190) is gone — it no longer applies to joined text.
- CommandService now sends array replies line by line: each element goes through the same routing as a single reply (`respond(id, line)` threaded to the trigger packet id; legacy sendText fallback per line when there is no packet id). A failing send stops the remaining lines (warn log), like any failed command.
- LLM failure path unchanged: single `summary failed` reply via respond(). `#rss get #tag` gets the same multi-message treatment automatically (shared summary path — covered by tests). All other subcommands unchanged.
- Tests: 158 → 165 — prompt shape + maxTokens asserted on the new call, 5-item cap, invalid-line filtering, fewer-than-3 fallback, per-line over-length cut through `handle()`, direct `parseSummaryItems`/`fitOneLine` specs, and CommandService array-reply routing (threaded respond ×N to one packet id; legacy sendText fallback).
## v0.20.1 — 2026-09-25 19:20 ## v0.20.1 — 2026-09-25 19:20
- `#rss get` summaries now fit ONE mesh packet (≤190 chars). Mesh fragments longer replies into fire-and-forget packets that recipients frequently never receive, and the old prompt ("a few short one-line highlights") produced ~326-334 char replies. - `#rss get` summaries now fit ONE mesh packet (≤190 chars). Mesh fragments longer replies into fire-and-forget packets that recipients frequently never receive, and the old prompt ("a few short one-line highlights") produced ~326-334 char replies.
+1
View File
@@ -2,6 +2,7 @@ services:
bot: bot:
build: . build: .
environment: environment:
TZ: Europe/Warsaw
MESH_HOST: 192.168.1.4 MESH_HOST: 192.168.1.4
MESH_PORT: '4403' MESH_PORT: '4403'
MESH_DB_PATH: ./data/mesh-bot.db MESH_DB_PATH: ./data/mesh-bot.db
+1 -1
View File
@@ -1,6 +1,6 @@
{ {
"name": "mesh-bot", "name": "mesh-bot",
"version": "0.20.1", "version": "0.21.12",
"description": "", "description": "",
"author": "", "author": "",
"private": true, "private": true,
+1 -2
View File
@@ -1,10 +1,9 @@
import { Module } from '@nestjs/common'; import { Module } from '@nestjs/common';
import { ScheduleModule } from '@nestjs/schedule';
import { MeshModule } from '../mesh/mesh.module.js'; import { MeshModule } from '../mesh/mesh.module.js';
import { BatteryAlertService } from './battery.service.js'; import { BatteryAlertService } from './battery.service.js';
@Module({ @Module({
imports: [ScheduleModule.forRoot(), MeshModule], imports: [MeshModule],
providers: [BatteryAlertService], providers: [BatteryAlertService],
exports: [BatteryAlertService], exports: [BatteryAlertService],
}) })
+41
View File
@@ -119,6 +119,47 @@ describe('BatteryAlertService check threshold', () => {
expect(sendText).not.toHaveBeenCalled(); expect(sendText).not.toHaveBeenCalled();
}); });
it('ignores a 0% reading (invalid/no battery) even though 0 <= threshold', async () => {
const sendText = vi.fn(async (..._args: unknown[]) => 1);
// 0% must not alert, but a genuinely-low 5% node in the same batch must
const svc = makeService([node(1, 0), node(2, 5)], sendText);
await svc.check();
expect(sendText).toHaveBeenCalledTimes(1);
// only the 5% node — the 0% node is excluded entirely
expect(sendText.mock.calls[0]).toEqual([
BROADCAST_ADDR,
'⚠️ battery low: @!00000002 5%',
undefined,
2,
]);
});
it('ignores a lone 0% reading (no broadcast at all)', async () => {
const sendText = vi.fn(async (..._args: unknown[]) => 1);
const svc = makeService([node(1, 0)], sendText); // only an invalid 0% reading
await svc.check();
expect(sendText).not.toHaveBeenCalled();
});
it('still alerts a genuine 1% reading (lowest valid value)', async () => {
const sendText = vi.fn(async (..._args: unknown[]) => 1);
const svc = makeService([node(1, 1)], sendText);
await svc.check();
expect(sendText).toHaveBeenCalledTimes(1);
expect(sendText.mock.calls[0]).toEqual([
BROADCAST_ADDR,
'⚠️ battery low: @!00000001 1%',
undefined,
2,
]);
});
it('merges all qualifying nodes into exactly ONE broadcast with the broadcast target', async () => { it('merges all qualifying nodes into exactly ONE broadcast with the broadcast target', async () => {
const sendText = vi.fn(async (..._args: unknown[]) => 1); const sendText = vi.fn(async (..._args: unknown[]) => 1);
const svc = makeService( const svc = makeService(
+1
View File
@@ -60,6 +60,7 @@ export class BatteryAlertService {
private shouldAlert(node: NodeSummary, now: number): boolean { private shouldAlert(node: NodeSummary, now: number): boolean {
if (node.batteryLevel === undefined) return false; // no battery reported — skip if (node.batteryLevel === undefined) return false; // no battery reported — skip
if (node.batteryLevel <= 0) return false; // 0/invalid reading — no battery wired or misconfigured node, not a real low-battery alert
// <= on purpose: a node at exactly the threshold is already "low" (spec allowed either < or <=; tests pin this choice) // <= on purpose: a node at exactly the threshold is already "low" (spec allowed either < or <=; tests pin this choice)
if (!(node.batteryLevel <= this.threshold)) return false; if (!(node.batteryLevel <= this.threshold)) return false;
const last = this.lastAlertAt.get(node.num); const last = this.lastAlertAt.get(node.num);
+190 -1
View File
@@ -13,7 +13,11 @@ import {
KAMIL_NUM, KAMIL_NUM,
} from '../mesh/fake-bridge.js'; } from '../mesh/fake-bridge.js';
import { BROADCAST_ADDR } from '../mesh/codec.js'; import { BROADCAST_ADDR } from '../mesh/codec.js';
import type { MeshService } from '../mesh/mesh.service.js'; import type {
MessageRecord,
MeshService,
NodeSummary,
} from '../mesh/mesh.service.js';
import { MeshService as RealMeshService } from '../mesh/mesh.service.js'; import { MeshService as RealMeshService } from '../mesh/mesh.service.js';
import { CommandService } from './command.service.js'; import { CommandService } from './command.service.js';
@@ -30,6 +34,7 @@ function makeService(
beforeEach(() => { beforeEach(() => {
delete process.env.COMMAND_CHANNEL; // default channel 2 under test delete process.env.COMMAND_CHANNEL; // default channel 2 under test
process.env.MESH_MULTI_SEND_DELAY_MS = '0'; // no spacing under test unless a test opts in — the real default (5000) would slow the suite down
}); });
describe('CommandService.onRx', () => { describe('CommandService.onRx', () => {
@@ -40,6 +45,16 @@ describe('CommandService.onRx', () => {
expect(mesh.sendText).not.toHaveBeenCalled(); expect(mesh.sendText).not.toHaveBeenCalled();
}); });
it('replies #help with a short single-message list of all commands', async () => {
const [svc, mesh] = makeService();
await svc.onRx(2, '#help', 42, 7);
expect(mesh.respond).toHaveBeenCalledTimes(1); // one packet — not one per line
const text = mesh.respond.mock.calls[0][1];
expect(text).toContain('#ping');
expect(text).toContain('#loser');
expect(text).toContain('#rss');
});
it('falls back to a directed sendText on COMMAND_CHANNEL when the trigger has no packet id (DM)', async () => { it('falls back to a directed sendText on COMMAND_CHANNEL when the trigger has no packet id (DM)', async () => {
const [svc, mesh] = makeService(); const [svc, mesh] = makeService();
await svc.onRx(2, '#ping', 42); // from present but no id — defensive path await svc.onRx(2, '#ping', 42); // from present but no id — defensive path
@@ -114,6 +129,61 @@ describe('CommandService.onRx', () => {
expect(mesh.respond).toHaveBeenCalledWith(5, 'up from 99'); expect(mesh.respond).toHaveBeenCalledWith(5, 'up from 99');
}); });
it('sends each line of an array reply as its own threaded respond() to the same packet id', async () => {
const [svc, mesh] = makeService();
svc.register('multi', () => ['one: first item', 'two: second item']);
await svc.onRx(2, '#multi', 42, 9);
expect(mesh.respond).toHaveBeenCalledTimes(2);
expect(mesh.respond.mock.calls[0]).toEqual([9, 'one: first item']);
expect(mesh.respond.mock.calls[1]).toEqual([9, 'two: second item']);
expect(mesh.sendText).not.toHaveBeenCalled();
});
it('spaces consecutive multi-line sends by MESH_MULTI_SEND_DELAY_MS — no delay before the first line', async () => {
process.env.MESH_MULTI_SEND_DELAY_MS = '50'; // short so the test stays fast
let sends = 0;
const respond = vi.fn(async () => {
sends++;
return { id: 1 };
});
vi.useFakeTimers();
try {
const [svc, mesh] = makeService(undefined, respond);
svc.register('multi3', () => ['a: one', 'b: two', 'c: three']);
const p = svc.onRx(2, '#multi3', 42, 9);
// flush microtasks without advancing time — the first line must already be out (no delay before it)
await vi.advanceTimersByTimeAsync(0);
expect(sends).toBe(1);
// each subsequent line waits exactly MESH_MULTI_SEND_DELAY_MS after the previous one landed
await vi.advanceTimersByTimeAsync(50);
expect(sends).toBe(2);
await vi.advanceTimersByTimeAsync(50);
expect(sends).toBe(3);
await p;
expect(mesh.respond).toHaveBeenCalledTimes(3);
} finally {
vi.useRealTimers();
delete process.env.MESH_MULTI_SEND_DELAY_MS;
}
});
it('sends each line of an array reply via the legacy sendText fallback when there is no packet id', async () => {
const [svc, mesh] = makeService();
svc.register('multi', () => ['one: first item', 'two: second item']);
await svc.onRx(2, '#multi', 42); // no id → legacy path on COMMAND_CHANNEL, DM back to the sender
expect(mesh.sendText).toHaveBeenCalledTimes(2);
expect(mesh.sendText.mock.calls[0]).toEqual([
42,
'one: first item',
undefined,
2,
]);
expect(mesh.respond).not.toHaveBeenCalled();
});
it('swallows a failing handler (a bad command must not crash the app)', async () => { it('swallows a failing handler (a bad command must not crash the app)', async () => {
const [svc, mesh] = makeService(); const [svc, mesh] = makeService();
svc.register('boom', () => { svc.register('boom', () => {
@@ -140,6 +210,125 @@ describe('CommandService.onRx', () => {
}); });
}); });
describe('#loser', () => {
const tx = (
id: number,
to: number,
status: MessageRecord['status'],
): MessageRecord => ({
id,
dir: 'tx',
to,
text: 'hi',
channel: 3,
at: Date.now(),
status,
});
const makeLoserMesh = (
messages: MessageRecord[],
nodes: NodeSummary[],
): [CommandService, MeshService] => {
const mesh = {
sendText: vi.fn(),
respond: vi.fn().mockResolvedValue({ id: 1 }),
messages: () => messages,
nodes: async () => nodes,
} as unknown as MeshService;
return [new CommandService(mesh), mesh];
};
it('reports the node with the most failures — named via longName, broadcasts and pending sends excluded', async () => {
const messages = [
tx(10, BROADCAST_ADDR, 'failed'), // firmware quirk: a broadcast NACK is never a real destination
tx(9, 222, 'failed'), // Alice: 3 failed / 0 delivered → loser
tx(8, 111, 'delivered'), // Bob: 1 failed / 2 delivered
tx(7, 222, 'failed'),
tx(6, 111, 'failed'),
tx(5, 333, 'pending'), // no final status yet — ignored
tx(4, 222, 'failed'),
{
id: 3,
dir: 'rx',
from: 999,
text: 'hello',
channel: 0,
at: Date.now(),
},
];
const nodes = [
{ num: 111, shortName: 'Bob', lastHeard: null },
{ num: 222, longName: 'Alice', lastHeard: null },
];
const [svc, mesh] = makeLoserMesh(messages, nodes);
await svc.onRx(2, '#loser', 42, 7);
expect(mesh.respond).toHaveBeenCalledWith(
7,
'worst delivery: Alice (@!000000de) - 3 failed / 0 delivered',
);
expect(mesh.sendText).not.toHaveBeenCalled();
});
it('says no failed sends when the directed history is all delivered', async () => {
const messages = [tx(2, 5, 'delivered'), tx(1, 6, 'delivered')];
const [svc, mesh] = makeLoserMesh(messages, []);
await svc.onRx(2, '#loser', 42, 7);
expect(mesh.respond).toHaveBeenCalledWith(
7,
'no failed sends yet - everyone delivers :)',
);
});
it('explains the directed-only rule when there is no final-status data at all', async () => {
const messages = [tx(2, BROADCAST_ADDR, 'pending'), tx(1, 5, 'pending')];
const [svc, mesh] = makeLoserMesh(messages, []);
await svc.onRx(2, '#loser', 42, 7);
expect(mesh.respond).toHaveBeenCalledWith(
7,
'not enough data yet - only directed sends count (broadcasts have no acks)',
);
});
it('falls back to the @!hex mention when the worst node has no known name', async () => {
const messages = [tx(1, 600, 'failed')];
const nodes = [{ num: 5, longName: 'Someone Else', lastHeard: null }];
const [svc, mesh] = makeLoserMesh(messages, nodes);
await svc.onRx(2, '#loser', 42, 7);
expect(mesh.respond).toHaveBeenCalledWith(
7,
'worst delivery: @!00000258 - 1 failed / 0 delivered',
);
});
it('tie-breaks equal failure counts by the worse failure ratio', async () => {
const messages = [
// node 7: 1 failed / 9 delivered (ratio 0.1) — not the loser
tx(16, 8, 'failed'), // node 8: 1 failed / 1 delivered (ratio 0.5) → loser
tx(15, 8, 'delivered'),
tx(14, 7, 'failed'),
...Array.from({ length: 9 }, (_, i) =>
tx(i + 1, 7, 'delivered' as const),
),
];
const [svc, mesh] = makeLoserMesh(messages, []);
await svc.onRx(2, '#loser', 42, 7);
expect(mesh.respond).toHaveBeenCalledWith(
7,
'worst delivery: @!00000008 - 1 failed / 1 delivered',
);
});
});
describe('#rss prefix dispatch', () => { describe('#rss prefix dispatch', () => {
let dir: string | undefined; let dir: string | undefined;
+89 -13
View File
@@ -1,4 +1,6 @@
import { Injectable, Logger } from '@nestjs/common'; import { Injectable, Logger } from '@nestjs/common';
// pure formatting helper (no cycle: battery → mesh only)
import { mention } from '../battery/battery.service.js';
import { BROADCAST_ADDR } from '../mesh/codec.js'; import { BROADCAST_ADDR } from '../mesh/codec.js';
// type-only: the instances come from DI (command.module.ts factory) or direct construction in tests — no runtime import cycles with src/mesh / src/rss // type-only: the instances come from DI (command.module.ts factory) or direct construction in tests — no runtime import cycles with src/mesh / src/rss
import type { MeshService } from '../mesh/mesh.service.js'; import type { MeshService } from '../mesh/mesh.service.js';
@@ -14,10 +16,10 @@ export interface CommandContext {
id?: number; id?: number;
} }
/** A command handler: returns the reply text to send back, or void/undefined for no reply. */ /** A command handler: returns the reply text(s) to send back — an array sends one message per line — or void/undefined for no reply. */
export type CommandHandler = ( export type CommandHandler = (
ctx: CommandContext, ctx: CommandContext,
) => string | undefined | Promise<string | undefined>; ) => string | string[] | undefined | Promise<string | string[] | undefined>;
@Injectable() @Injectable()
export class CommandService { export class CommandService {
@@ -27,6 +29,11 @@ export class CommandService {
process.env.COMMAND_CHANNEL ?? '2', process.env.COMMAND_CHANNEL ?? '2',
10, 10,
); );
// spacing between consecutive multi-line reply messages — back-to-back packets get dropped in RF on a real mesh; each packet needs time to land. Read at construction like the other env config.
private readonly multiSendDelayMs = parseInt(
process.env.MESH_MULTI_SEND_DELAY_MS ?? '5000',
10,
);
// single registration point: the constructor registers every built-in command — add the next one there (or via register()) // single registration point: the constructor registers every built-in command — add the next one there (or via register())
private readonly commands = new Map<string, CommandHandler>(); private readonly commands = new Map<string, CommandHandler>();
// prefix handlers own a whole family (`#rss`, `#rss add ...`) instead of one exact name // prefix handlers own a whole family (`#rss`, `#rss add ...`) instead of one exact name
@@ -38,9 +45,68 @@ export class CommandService {
rss?: RssService, rss?: RssService,
) { ) {
this.register('ping', () => '#pong!'); this.register('ping', () => '#pong!');
this.register('loser', () => this.loserReply());
// single multi-line string = ONE packet — not an array, which would drip-feed one spaced packet per line
this.register('help', () =>
[
'#ping - alive check',
'#loser - worst delivery record',
'#rss help - rss subcommands',
'#rss add <url> <name> - add feed with AI tags',
'#rss remove <name> - drop feed',
'#rss get <name|#tag> - news summary',
'#rss sources / #rss tags - list feeds/tags',
].join('\n'),
);
if (rss) this.registerPrefix('rss', (ctx) => rss.handle(ctx)); if (rss) this.registerPrefix('rss', (ctx) => rss.handle(ctx));
} }
/**
* #loser — worst packet delivery record, over directed sends only. Meshtastic shares no mesh-wide
* loss stats, so the only real data is our own tx history: group by destination, count delivered vs
* failed per node; broadcasts are excluded (firmware strips want_ack pre-TX, they stay pending — and
* the broadcast addr is never a real destination even if a NACK ever lands on it). Loser = most
* failures, tie-break worse failure ratio. One reply line, ≤ ~180 chars.
*/
private async loserReply(): Promise<string> {
const mesh = this.mesh;
if (!mesh)
return 'not enough data yet - only directed sends count (broadcasts have no acks)';
const stats = new Map<number, { failed: number; delivered: number }>();
for (const m of mesh.messages()) {
if (m.dir !== 'tx' || m.to === undefined || m.to === BROADCAST_ADDR)
continue; // directed sends only
if (m.status !== 'delivered' && m.status !== 'failed') continue; // pending = no final status yet
const s = stats.get(m.to) ?? { failed: 0, delivered: 0 };
if (m.status === 'failed') s.failed += 1;
else s.delivered += 1;
stats.set(m.to, s);
}
if (stats.size === 0)
return 'not enough data yet - only directed sends count (broadcasts have no acks)';
let worst: { num: number; failed: number; total: number } | undefined;
for (const [num, s] of stats) {
if (s.failed === 0) continue; // zero-failure nodes can't be the loser
const total = s.failed + s.delivered;
// most failures first, then worse failure ratio on a tie
if (
!worst ||
s.failed > worst.failed ||
(s.failed === worst.failed &&
s.failed / total > worst.failed / worst.total)
) {
worst = { num, failed: s.failed, total };
}
}
if (!worst) return 'no failed sends yet - everyone delivers :)';
const names = new Map(
(await mesh.nodes()).map((n) => [n.num, n.longName ?? n.shortName]),
);
const name = names.get(worst.num);
const who = name ? `${name} (${mention(worst.num)})` : mention(worst.num);
return `worst delivery: ${who} - ${worst.failed} failed / ${worst.total - worst.failed} delivered`;
}
/** Register a command handler (names are case-sensitive; the leading `#` is stripped before lookup). */ /** Register a command handler (names are case-sensitive; the leading `#` is stripped before lookup). */
register(name: string, fn: CommandHandler): void { register(name: string, fn: CommandHandler): void {
this.commands.set(name, fn); this.commands.set(name, fn);
@@ -83,17 +149,27 @@ export class CommandService {
try { try {
const reply = await handler({ text: trimmed, from, id }); const reply = await handler({ text: trimmed, from, id });
if (reply === undefined || reply === null) return; // void = no reply if (reply === undefined || reply === null) return; // void = no reply
if (id !== undefined && this.mesh) { // an array reply sends one message per line, all threaded to the same trigger packet id — spaced so each packet has time to land in RF (no delay before the first line)
// respond() auto-routes the reply: rides the original record's channel with replyId = id (threaded quote) const lines = Array.isArray(reply) ? reply : [reply];
await this.mesh.respond(id, reply); for (let i = 0; i < lines.length; i++) {
} else { if (i > 0 && this.multiSendDelayMs > 0) {
// defensive fallback when the trigger has no packet id: legacy routing on COMMAND_CHANNEL await new Promise((resolve) =>
await this.mesh?.sendText( setTimeout(resolve, this.multiSendDelayMs),
from ?? BROADCAST_ADDR, );
reply, }
undefined, const line = lines[i];
this.commandChannel, if (id !== undefined && this.mesh) {
); // respond() auto-routes each line: rides the original record's channel with replyId = id (threaded quote)
await this.mesh.respond(id, line);
} else {
// defensive fallback when the trigger has no packet id: legacy routing on COMMAND_CHANNEL
await this.mesh?.sendText(
from ?? BROADCAST_ADDR,
line,
undefined,
this.commandChannel,
);
}
} }
} catch (err) { } catch (err) {
this.logger.warn( this.logger.warn(
+2 -2
View File
@@ -12,9 +12,9 @@ export class LlmService {
private readonly apiBase = private readonly apiBase =
process.env.LLM_API_BASE ?? 'http://192.168.1.70:1111/v1'; process.env.LLM_API_BASE ?? 'http://192.168.1.70:1111/v1';
private readonly model = process.env.LLM_MODEL ?? 'qwen3.8-27b'; private readonly model = process.env.LLM_MODEL ?? 'qwen3.8-27b';
// local 27B model can be slow — generous default // local 27B model can be slow (uncontrollable reasoning tokens) — generous default: 3 min headroom for slow reasoning periods
private readonly timeoutMs = parseInt( private readonly timeoutMs = parseInt(
process.env.LLM_TIMEOUT_MS ?? '60000', process.env.LLM_TIMEOUT_MS ?? '180000',
10, 10,
); );
+2
View File
@@ -10,6 +10,8 @@ export function startBridge(
): Promise<{ port: number; sockets: net.Socket[]; close: () => void }> { ): Promise<{ port: number; sockets: net.Socket[]; close: () => void }> {
const sockets: net.Socket[] = []; const sockets: net.Socket[] = [];
const server = net.createServer((socket) => { const server = net.createServer((socket) => {
// Test-owned socket: the service destroys its end mid-test with unread data pending (overflow/garbage tests), which makes the kernel send RST and this side emit ECONNRESET. Expected teardown noise — without a handler Node throws it as an uncaught exception and vitest fails the whole run.
socket.on('error', () => {});
onSocket?.(socket); // attached at accept time, before any data can arrive onSocket?.(socket); // attached at accept time, before any data can arrive
sockets.push(socket); sockets.push(socket);
}); });
+190
View File
@@ -1,6 +1,7 @@
import { create, fromBinary, toBinary } from '@bufbuild/protobuf'; import { create, fromBinary, toBinary } from '@bufbuild/protobuf';
import * as P from '@meshtastic/protobufs'; import * as P from '@meshtastic/protobufs';
import { BadRequestException, NotFoundException } from '@nestjs/common'; import { BadRequestException, NotFoundException } from '@nestjs/common';
import * as net from 'node:net';
import { DatabaseSync } from 'node:sqlite'; import { DatabaseSync } from 'node:sqlite';
import { beforeEach, describe, expect, it, vi } from 'vitest'; import { beforeEach, describe, expect, it, vi } from 'vitest';
import { BROADCAST_ADDR, extractFrames, frame } from './codec.js'; import { BROADCAST_ADDR, extractFrames, frame } from './codec.js';
@@ -249,6 +250,195 @@ describe('MeshService connect', () => {
}); });
}); });
const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms));
describe('MeshService idle watchdog', () => {
it('forces a reconnect when an established connection goes silent past MESH_IDLE_TIMEOUT_MS', async () => {
const bridge = await startBridge();
process.env.MESH_HOST = '127.0.0.1';
process.env.MESH_PORT = String(bridge.port);
process.env.MESH_RECONNECT_MS = '50'; // fast retry so the re-establishment is observable
process.env.MESH_IDLE_TIMEOUT_MS = '100'; // override — the default (60000) would never fire in a test
const svc = new MeshService();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
// zombie case: connected but no inbound traffic at all — wait past the timeout, then drive a watchdog tick
await sleep(250);
svc.checkIdle();
// socket torn down → existing connect/retry loop re-establishes with a fresh session
await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected);
} finally {
delete process.env.MESH_RECONNECT_MS;
delete process.env.MESH_IDLE_TIMEOUT_MS;
svc.onModuleDestroy();
bridge.close();
}
});
it('raw non-frame bytes do NOT keep the watchdog alive — a garbage-only stream forces a reconnect', async () => {
const bridge = await startBridge();
process.env.MESH_HOST = '127.0.0.1';
process.env.MESH_PORT = String(bridge.port);
process.env.MESH_RECONNECT_MS = '50'; // fast retry so the re-establishment is observable
process.env.MESH_IDLE_TIMEOUT_MS = '100'; // override — the default (60000) would never fire in a test
const svc = new MeshService();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
// today's incident: the peer keeps dribbling small non-frame bytes (keepalive noise / desynced stream) — raw chunks must not reset the idle clock
const garbage = Buffer.from('zz-no-frames-here'); // pure ASCII — never forms a 0x94 0xc3 frame header
for (let i = 0; i < 6; i++) {
await sleep(50); // ~300 ms wall time, well past the timeout
bridge.sockets[0].write(garbage);
}
svc.checkIdle();
await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected);
} finally {
delete process.env.MESH_RECONNECT_MS;
delete process.env.MESH_IDLE_TIMEOUT_MS;
svc.onModuleDestroy();
bridge.close();
}
});
it('forces a reconnect when the undecoded frame buffer exceeds MESH_MAX_FRAME_BUFFER', async () => {
const bridge = await startBridge();
process.env.MESH_HOST = '127.0.0.1';
process.env.MESH_PORT = String(bridge.port);
process.env.MESH_RECONNECT_MS = '50'; // fast retry so the re-establishment is observable
process.env.MESH_MAX_FRAME_BUFFER = '100'; // tiny cap — Node delivers socket data in ~64 KB events, far below the 256_000 default
const svc = new MeshService();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
// non-frame bytes (0x42 is not part of any frame header) — a desynced stream that never forms frames; the first ~64 KB data event alone exceeds the cap
bridge.sockets[0].write(Buffer.alloc(300_000, 0x42));
// overflow forced a reconnect — fresh session
await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected);
// the new socket starts with a clean buffer and decodes valid frames normally (the myInfo frame is well under the cap)
const fr = create(P.Mesh.FromRadioSchema, {
payloadVariant: { case: 'myInfo', value: { myNodeNum: 777 } },
});
bridge.sockets[1].write(frame(toBinary(P.Mesh.FromRadioSchema, fr)));
await waitFor(() => svc.status().myNodeNum === 777);
} finally {
delete process.env.MESH_RECONNECT_MS;
delete process.env.MESH_MAX_FRAME_BUFFER;
svc.onModuleDestroy();
bridge.close();
}
});
it('inbound traffic keeps resetting the idle clock — no false trigger while packets arrive', async () => {
const bridge = await startBridge();
process.env.MESH_HOST = '127.0.0.1';
process.env.MESH_PORT = String(bridge.port);
process.env.MESH_IDLE_TIMEOUT_MS = '100';
const svc = new MeshService();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
// steady drip of inbound frames for well over the timeout in wall time — each chunk resets lastPacketAt
for (let i = 0; i < 5; i++) {
await sleep(60);
bridge.sockets[0].write(nodeInfoFrame(42, kamilNodeInfo()));
}
svc.checkIdle(); // wall time is past the timeout, but traffic kept resetting it
await sleep(100); // enough for a destroy to surface as 'close' (connected would flip false)
expect(svc.status().connected).toBe(true);
expect(bridge.sockets.length).toBe(1);
} finally {
delete process.env.MESH_IDLE_TIMEOUT_MS;
svc.onModuleDestroy();
bridge.close();
}
});
it('does not trigger under the default timeout (MESH_IDLE_TIMEOUT_MS unset)', async () => {
const bridge = await startBridge();
process.env.MESH_HOST = '127.0.0.1';
process.env.MESH_PORT = String(bridge.port);
delete process.env.MESH_IDLE_TIMEOUT_MS; // default 60000 — a test can't wait that long, so silence must NOT fire
const svc = new MeshService();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
await sleep(250); // silent, but far under the default budget
svc.checkIdle();
expect(svc.status().connected).toBe(true);
expect(bridge.sockets.length).toBe(1);
} finally {
delete process.env.MESH_IDLE_TIMEOUT_MS;
svc.onModuleDestroy();
bridge.close();
}
});
it('enables TCP keepalive on the bridge socket', async () => {
const setKeepAlive = vi.spyOn(net.Socket.prototype, 'setKeepAlive');
const bridge = await startBridge();
process.env.MESH_HOST = '127.0.0.1';
process.env.MESH_PORT = String(bridge.port);
const svc = new MeshService();
try {
svc.onModuleInit();
await waitFor(() => svc.status().connected);
expect(setKeepAlive).toHaveBeenCalledWith(true, 30_000);
} finally {
setKeepAlive.mockRestore();
svc.onModuleDestroy();
bridge.close();
}
});
});
describe('MeshService periodic autoreconnect', () => {
it('unconditionally reconnects when MESH_RECONNECT_INTERVAL_MS elapses', async () => {
const bridge = await startBridge();
process.env.MESH_HOST = '127.0.0.1';
process.env.MESH_PORT = String(bridge.port);
process.env.MESH_RECONNECT_MS = '50'; // fast retry so the re-establishment is observable
process.env.MESH_RECONNECT_INTERVAL_MS = '100'; // override — the default (300000) would never elapse in a test
const svc = new MeshService();
try {
svc.onModuleInit();
await waitFor(
() => bridge.sockets.length === 1 && svc.status().connected,
);
// interval fires → destroy → existing close/retry loop re-establishes with a fresh session
await waitFor(() => bridge.sockets.length >= 2 && svc.status().connected);
} finally {
delete process.env.MESH_RECONNECT_MS;
delete process.env.MESH_RECONNECT_INTERVAL_MS;
svc.onModuleDestroy();
bridge.close();
}
});
});
describe('MeshService send', () => { describe('MeshService send', () => {
it('sends on the test channel (index 3) by default, never primary', async () => { it('sends on the test channel (index 3) by default, never primary', async () => {
const bridge = await startBridge(); const bridge = await startBridge();
+68 -1
View File
@@ -81,6 +81,8 @@ export interface TelemetryRecord {
const HISTORY_LIMIT = 100; const HISTORY_LIMIT = 100;
/** Rows kept in SQLite across restarts; older rows are pruned on every poller tick. */ /** Rows kept in SQLite across restarts; older rows are pruned on every poller tick. */
const DB_RETENTION = 1000; const DB_RETENTION = 1000;
// how often the idle watchdog checks for a silent-but-connected bridge (ms)
const IDLE_CHECK_MS = 30_000;
// one row per MessageRecord, keyed by packet id (tx ids and injected ids are random uint32, rx uses the wire id) // one row per MessageRecord, keyed by packet id (tx ids and injected ids are random uint32, rx uses the wire id)
export const MESSAGES_DDL = `CREATE TABLE IF NOT EXISTS messages( export const MESSAGES_DDL = `CREATE TABLE IF NOT EXISTS messages(
@@ -208,6 +210,21 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
); );
// how often the poller re-upserts the history buffer into SQLite (ms) // how often the poller re-upserts the history buffer into SQLite (ms)
private readonly pollMs = parseInt(process.env.MESH_POLL_MS ?? '1000', 10); private readonly pollMs = parseInt(process.env.MESH_POLL_MS ?? '1000', 10);
// idle watchdog: force a reconnect when an ESTABLISHED connection stops decoding frames — the zombie-bridge case where outbound sends keep working but inbound frame flow is dead (raw non-frame bytes do NOT reset the clock). Read at construction like the other env config.
private readonly idleTimeoutMs = parseInt(
process.env.MESH_IDLE_TIMEOUT_MS ?? '60000',
10,
);
// lazy periodic autoreconnect: unconditionally destroy + re-establish every N ms so a fresh sync happens on schedule — the bridge/radio side tends to kill inbound RX within ~20 min of a fresh sync. Read at construction like the other env config.
private readonly reconnectIntervalMs = parseInt(
process.env.MESH_RECONNECT_INTERVAL_MS ?? '300000',
10,
);
// cap on undecoded frame buffer: a desynced stream that never forms frames would grow it without bound — force a reconnect instead. Env-overridable so tests can trigger it (Node caps socket 'data' events at ~64 KB).
private readonly maxFrameBuffer = parseInt(
process.env.MESH_MAX_FRAME_BUFFER ?? '256000',
10,
);
/** SQLite database backing the message history; null = persistence disabled (tests construct without one). */ /** SQLite database backing the message history; null = persistence disabled (tests construct without one). */
private db: DatabaseSync | null = null; private db: DatabaseSync | null = null;
@@ -255,6 +272,10 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
// telemetry history (newest kept, capped at HISTORY_LIMIT) // telemetry history (newest kept, capped at HISTORY_LIMIT)
private telemetryHistory: TelemetryRecord[] = []; private telemetryHistory: TelemetryRecord[] = [];
private reconnectTimer: NodeJS.Timeout | null = null; private reconnectTimer: NodeJS.Timeout | null = null;
/** epoch ms when the last valid frame was decoded from the bridge; reset on connect — idle watchdog baseline */
private lastPacketAt = Date.now();
private idleTimer: NodeJS.Timeout | null = null;
private periodicReconnectTimer: NodeJS.Timeout | null = null;
private shuttingDown = false; private shuttingDown = false;
onModuleInit(): void { onModuleInit(): void {
@@ -264,6 +285,15 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
// ponytail: full re-upsert of the ≤100-row buffer every tick — trivial load, zero bookkeeping // ponytail: full re-upsert of the ≤100-row buffer every tick — trivial load, zero bookkeeping
this.pollTimer = setInterval(() => this.flushHistory(), this.pollMs); this.pollTimer = setInterval(() => this.flushHistory(), this.pollMs);
} }
// idle watchdog: catches established-but-silent connections (zombie bridge) that the connect/retry loop never sees
this.idleTimer = setInterval(() => this.checkIdle(), IDLE_CHECK_MS);
// lazy periodic autoreconnect: unconditional fresh sync on a fixed schedule; when not connected, destroy is a no-op and the retry loop owns recovery
this.periodicReconnectTimer = setInterval(() => {
this.logger.log(
'periodic reconnect — MESH_RECONNECT_INTERVAL_MS elapsed',
);
this.socket?.destroy();
}, this.reconnectIntervalMs);
this.connect(); this.connect();
} }
@@ -271,6 +301,8 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
this.shuttingDown = true; this.shuttingDown = true;
if (this.reconnectTimer) clearTimeout(this.reconnectTimer); if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
if (this.pollTimer) clearInterval(this.pollTimer); if (this.pollTimer) clearInterval(this.pollTimer);
if (this.idleTimer) clearInterval(this.idleTimer);
if (this.periodicReconnectTimer) clearInterval(this.periodicReconnectTimer);
this.socket?.destroy(); this.socket?.destroy();
this.db?.close(); this.db?.close();
} }
@@ -641,11 +673,17 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
private connect(): void { private connect(): void {
if (this.shuttingDown) return; if (this.shuttingDown) return;
const socket = net.createConnection({ host: this.host, port: this.port }); const socket = net.createConnection({ host: this.host, port: this.port });
// TCP-level keepalive: cheap insurance against a dead peer that the idle watchdog (application level) complements
socket.setKeepAlive(true, 30_000);
this.socket = socket; this.socket = socket;
socket.on('connect', () => { socket.on('connect', () => {
this.connected = true; this.connected = true;
// fresh connection — reset the idle clock so it can't self-trigger before any traffic arrives
this.lastPacketAt = Date.now();
// fresh sync on every (re)connect — drop stale channels/nodes/device info/telemetry from the old session; // fresh sync on every (re)connect — drop stale channels/nodes/device info/telemetry from the old session;
// message history is NOT reset: it's an append-only log backed by SQLite and survives reconnects // message history is NOT reset: it's an append-only log backed by SQLite and survives reconnects.
// The undecoded frame buffer goes too: leftover desync garbage must not re-trigger the overflow backstop on a healthy new stream.
this.buffer = new Uint8Array(0);
this.channelMap = new Map(); this.channelMap = new Map();
this.nodeMap = new Map(); this.nodeMap = new Map();
this.metadata = null; this.metadata = null;
@@ -666,9 +704,19 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
const merged = new Uint8Array(this.buffer.length + chunk.length); const merged = new Uint8Array(this.buffer.length + chunk.length);
merged.set(this.buffer, 0); merged.set(this.buffer, 0);
merged.set(chunk, this.buffer.length); merged.set(chunk, this.buffer.length);
// buffer-overflow backstop: a desynced stream that never forms frames would grow the buffer without bound — cap it and self-heal via reconnect
if (merged.length > this.maxFrameBuffer) {
this.logger.error(
'frame buffer overflow — stream desync, forcing reconnect',
);
socket.destroy();
return;
}
this.buffer = merged; this.buffer = merged;
const { packets, rest } = extractFrames(this.buffer); const { packets, rest } = extractFrames(this.buffer);
this.buffer = rest; this.buffer = rest;
// the watchdog tracks DECODED FRAMES, not raw bytes: non-frame noise (keepalives, desynced streams) must not keep a dead connection alive
if (packets.length) this.lastPacketAt = Date.now();
for (const p of packets) { for (const p of packets) {
decodeFromRadio(p) decodeFromRadio(p)
.then((fr) => this.handleFromRadio(fr)) .then((fr) => this.handleFromRadio(fr))
@@ -694,6 +742,25 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
this.reconnectTimer = setTimeout(() => this.connect(), this.reconnectMs); this.reconnectTimer = setTimeout(() => this.connect(), this.reconnectMs);
} }
/**
* Idle watchdog tick: force a reconnect when an ESTABLISHED connection has decoded no frames from
* the bridge for MESH_IDLE_TIMEOUT_MS — the zombie case where outbound sends keep working but inbound
* frame flow is dead (raw non-frame bytes arriving on the socket do not count). Destroying the socket
* routes through 'close' → onDown → scheduleReconnect, so the existing connect/retry loop re-establishes
* with a fresh sync; no new retry machinery here. Public so tests can drive a tick without waiting for
* the real interval (like flushHistory()).
*/
checkIdle(): void {
if (!this.connected) return; // disconnected state is owned by the connect/retry loop
const idle = Date.now() - this.lastPacketAt;
if (idle > this.idleTimeoutMs) {
this.logger.warn(
`no decoded bridge frames for ${idle} ms — forcing reconnect`,
);
this.socket?.destroy();
}
}
private async handleFromRadio( private async handleFromRadio(
fr: import('@meshtastic/protobufs').Mesh.FromRadio, fr: import('@meshtastic/protobufs').Mesh.FromRadio,
): Promise<void> { ): Promise<void> {
+45 -27
View File
@@ -72,7 +72,7 @@ describe('OpencodeService.notify', () => {
expect(fetchMock).toHaveBeenCalledTimes(2); expect(fetchMock).toHaveBeenCalledTimes(2);
const [url, init] = fetchMock.mock.calls[1]; const [url, init] = fetchMock.mock.calls[1];
expect(url).toBe(`${API}/proxy/abc/session/ses_pinned/message`); expect(url).toBe(`${API}/proxy/abc/api/session/ses_pinned/prompt`);
const opts = init as { const opts = init as {
method?: string; method?: string;
headers?: Record<string, string>; headers?: Record<string, string>;
@@ -83,15 +83,10 @@ describe('OpencodeService.notify', () => {
expect(opts.headers?.authorization).toBe( expect(opts.headers?.authorization).toBe(
'Basic ' + Buffer.from('u:p').toString('base64'), 'Basic ' + Buffer.from('u:p').toString('base64'),
); );
// body is EXACTLY {"parts":[{"type":"text","text":msg}]} — em-dash, not a hyphen // body is EXACTLY {"text": msg} — em-dash, not a hyphen — and a 2xx resolves notify as delivered
expect(String(opts.body)).toBe( expect(String(opts.body)).toBe(
JSON.stringify({ JSON.stringify({
parts: [ text: 'New mesh message (id 7) from node 42: "hi" — please react to it.',
{
type: 'text',
text: 'New mesh message (id 7) from node 42: "hi" — please react to it.',
},
],
}), }),
); );
}); });
@@ -111,12 +106,7 @@ describe('OpencodeService.notify', () => {
const [, init] = fetchMock.mock.calls[1]; const [, init] = fetchMock.mock.calls[1];
expect(String((init as { body?: string }).body)).toBe( expect(String((init as { body?: string }).body)).toBe(
JSON.stringify({ JSON.stringify({
parts: [ text: 'New mesh message (id 7): "hi" — please react to it.',
{
type: 'text',
text: 'New mesh message (id 7): "hi" — please react to it.',
},
],
}), }),
); );
}); });
@@ -149,7 +139,7 @@ describe('OpencodeService.notify', () => {
await svc.notify(7, 'hi'); await svc.notify(7, 'hi');
expect(fetchMock.mock.calls[1][0]).toBe( expect(fetchMock.mock.calls[1][0]).toBe(
`${API}/proxy/abc/session/ses_pinned/message`, `${API}/proxy/abc/api/session/ses_pinned/prompt`,
); );
}); });
@@ -182,7 +172,7 @@ describe('OpencodeService.notify', () => {
await svc.notify(7, 'hi'); await svc.notify(7, 'hi');
expect(fetchMock.mock.calls[1][0]).toBe( expect(fetchMock.mock.calls[1][0]).toBe(
`${API}/proxy/b/session/ses_pinned/message`, `${API}/proxy/b/api/session/ses_pinned/prompt`,
); );
}); });
@@ -191,18 +181,37 @@ describe('OpencodeService.notify', () => {
fetchMock fetchMock
.mockResolvedValueOnce(json([WORKSPACE])) .mockResolvedValueOnce(json([WORKSPACE]))
.mockResolvedValueOnce( .mockResolvedValueOnce(
json([ // v2 envelope {data, cursor}; the /elsewhere session is more recent but a different
{ id: 'ses_old', time: { updated: 100 } }, // directory, so the location.directory filter must exclude it
{ id: 'ses_new', time: { updated: 200 } }, json({
]), data: [
{
id: 'ses_old',
location: { directory: WS_PATH },
time: { updated: 100 },
},
{
id: 'ses_other_ws',
location: { directory: '/elsewhere' },
time: { updated: 300 },
},
{
id: 'ses_new',
location: { directory: WS_PATH },
time: { updated: 200 },
},
],
cursor: { previous: null, next: null },
}),
) )
.mockResolvedValueOnce(json({})); .mockResolvedValueOnce(json({}));
await svc.notify(7, 'hi'); await svc.notify(7, 'hi');
expect(fetchMock.mock.calls[1][0]).toBe(`${API}/proxy/abc/session`); // discovery GET hits the v2 route; the prompt POST goes to the most recently updated session of THIS directory
expect(fetchMock.mock.calls[1][0]).toBe(`${API}/proxy/abc/api/session`);
expect(fetchMock.mock.calls[2][0]).toBe( expect(fetchMock.mock.calls[2][0]).toBe(
`${API}/proxy/abc/session/ses_new/message`, `${API}/proxy/abc/api/session/ses_new/prompt`,
); );
}); });
@@ -210,13 +219,22 @@ describe('OpencodeService.notify', () => {
const svc = makeService({ OPENCODE_USER: 'u', OPENCODE_PASS: 'p' }); const svc = makeService({ OPENCODE_USER: 'u', OPENCODE_PASS: 'p' });
fetchMock fetchMock
.mockResolvedValueOnce(json([WORKSPACE])) .mockResolvedValueOnce(json([WORKSPACE]))
.mockResolvedValueOnce(json([{ id: 'ses_a' }, { id: 'ses_b', time: {} }])) .mockResolvedValueOnce(
json({
data: [
{ id: 'ses_a', location: { directory: WS_PATH } },
{ id: 'ses_b', location: { directory: WS_PATH }, time: {} },
],
cursor: { previous: null, next: null },
}),
)
.mockResolvedValueOnce(json({})); .mockResolvedValueOnce(json({}));
await svc.notify(7, 'hi'); await svc.notify(7, 'hi');
expect(fetchMock.mock.calls[1][0]).toBe(`${API}/proxy/abc/api/session`);
expect(fetchMock.mock.calls[2][0]).toBe( expect(fetchMock.mock.calls[2][0]).toBe(
`${API}/proxy/abc/session/ses_a/message`, `${API}/proxy/abc/api/session/ses_a/prompt`,
); );
}); });
@@ -232,11 +250,11 @@ describe('OpencodeService.notify', () => {
await svc.notify(7, 'hi'); await svc.notify(7, 'hi');
// workspaces GET + message POST only — no /session discovery GET // workspaces GET + prompt POST only — no /api/session discovery GET
expect(fetchMock).toHaveBeenCalledTimes(2); expect(fetchMock).toHaveBeenCalledTimes(2);
expect(fetchMock.mock.calls[0][0]).toBe(`${API}/api/workspaces`); expect(fetchMock.mock.calls[0][0]).toBe(`${API}/api/workspaces`);
expect(fetchMock.mock.calls[1][0]).toBe( expect(fetchMock.mock.calls[1][0]).toBe(
`${API}/proxy/abc/session/ses_pinned/message`, `${API}/proxy/abc/api/session/ses_pinned/prompt`,
); );
}); });
@@ -248,7 +266,7 @@ describe('OpencodeService.notify', () => {
}); });
fetchMock fetchMock
.mockResolvedValueOnce(json([WORKSPACE])) .mockResolvedValueOnce(json([WORKSPACE]))
.mockRejectedValueOnce(timeoutError()); // the message POST times out — opencode holds the response until the turn finishes .mockRejectedValueOnce(timeoutError()); // the prompt POST times out — a read timeout is still treated as delivered (safety net)
const logSpy = vi.spyOn(Logger.prototype, 'log'); const logSpy = vi.spyOn(Logger.prototype, 'log');
await expect(svc.notify('hi')).resolves.toBeUndefined(); await expect(svc.notify('hi')).resolves.toBeUndefined();
+24 -13
View File
@@ -2,7 +2,8 @@ import { Injectable, Logger } from '@nestjs/common';
// ponytail: hardcoded backoff [2s, 10s, 30s] (~45 s window) — no env knob; tune here if outages run longer // 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]; 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 // opencode v2 returns 2xx immediately (it no longer holds the response until the agent turn finishes);
// the 15 s timeout only guards a stuck connection — a read timeout is still treated as delivered
const TIMEOUT_MS = 15_000; const TIMEOUT_MS = 15_000;
/** Entry of GET {api}/api/workspaces (only the fields we use). */ /** Entry of GET {api}/api/workspaces (only the fields we use). */
@@ -14,9 +15,11 @@ interface WorkspaceEntry {
proxyPath?: string; proxyPath?: string;
} }
/** Entry of GET {base}/session (only the fields we use). */ /** Entry of GET {base}/api/session (v2; only the fields we use — directory lives under location, time is epoch ms). */
interface SessionEntry { interface SessionEntry {
id: string; id: string;
title?: string;
location?: { directory?: string };
time?: { updated?: number }; time?: { updated?: number };
} }
@@ -43,11 +46,11 @@ export class OpencodeService {
} }
/** /**
* Push an rx mesh message into the opencode session as a text part. Resolves when delivered — * Push an rx mesh message into the opencode session (v2: POST {base}/api/session/{id}/prompt with a
* including on a final-POST read timeout, which means DELIVERED (opencode holds the response until * `{text}` body). Resolves when delivered — including on a final-POST read timeout, which still means
* the agent turn finishes) and is never retried. Workspace/session resolution failures are retried * DELIVERED (kept as a harmless safety net; v2 normally returns 2xx immediately) and is never retried.
* up to RETRY_DELAYS_MS.length + 1 attempts total; the last failure logs `opencode notify dropped * Workspace/session resolution failures are retried up to RETRY_DELAYS_MS.length + 1 attempts total;
* after N attempts`. No-op when disabled (no credentials). * the last failure logs `opencode notify dropped after N attempts`. No-op when disabled (no credentials).
*/ */
async notify(id: number, text: string, from?: number): Promise<void> { async notify(id: number, text: string, from?: number): Promise<void> {
if (!this.enabled) return; if (!this.enabled) return;
@@ -112,12 +115,19 @@ export class OpencodeService {
const base = this.api + (ws.proxyPath ?? ''); const base = this.api + (ws.proxyPath ?? '');
let sid = this.sessionOverride; let sid = this.sessionOverride;
if (!sid) { if (!sid) {
const sessions = await this.getJson<SessionEntry[]>(`${base}/session`); // v2 discovery: GET {base}/api/session returns an envelope { data, cursor } — NOT a bare array
const envelope = await this.getJson<{ data?: SessionEntry[] }>(
`${base}/api/session`,
);
// only sessions of THIS workspace directory — the instance proxy may serve other sessions too
const sessions = (envelope.data ?? []).filter(
(s) => s.location?.directory === this.workspacePath,
);
if (sessions.length === 0) if (sessions.length === 0)
throw new Error( throw new Error(
`no sessions in workspace ${ws.id} (${this.workspacePath})`, `no sessions in workspace ${ws.id} (${this.workspacePath})`,
); );
// most recently updated session; first entry when none carry time.updated // most recently updated session (epoch ms); first entry when none carry time.updated
const updated = (s: SessionEntry): number => s.time?.updated ?? 0; const updated = (s: SessionEntry): number => s.time?.updated ?? 0;
sid = sessions.some((s) => updated(s)) sid = sessions.some((s) => updated(s))
? sessions.reduce((a, b) => (updated(b) > updated(a) ? b : a)).id ? sessions.reduce((a, b) => (updated(b) > updated(a) ? b : a)).id
@@ -148,16 +158,17 @@ export class OpencodeService {
msg: string, msg: string,
): Promise<void> { ): Promise<void> {
try { try {
const res = await fetch(`${base}/session/${sid}/message`, { const res = await fetch(`${base}/api/session/${sid}/prompt`, {
method: 'POST', method: 'POST',
headers: { ...this.authHeaders(), 'content-type': 'application/json' }, headers: { ...this.authHeaders(), 'content-type': 'application/json' },
body: JSON.stringify({ parts: [{ type: 'text', text: msg }] }), body: JSON.stringify({ text: msg }),
signal: AbortSignal.timeout(TIMEOUT_MS), signal: AbortSignal.timeout(TIMEOUT_MS),
}); });
if (!res.ok) throw new Error(`HTTP ${res.status}`); // HTTP errors retry like network failures if (!res.ok) throw new Error(`HTTP ${res.status}`); // HTTP errors retry like network failures
} catch (err) { } catch (err) {
// a read timeout here means DELIVERED, not failed — opencode accepted the message and is holding // a read timeout here means DELIVERED, not failed — kept as a harmless safety net: v2 normally
// the response until the agent turn finishes; never retried // returns 2xx immediately, but if the response is still pending past the timeout we treat the
// message as accepted and never retry
if ((err as Error).name === 'TimeoutError') { if ((err as Error).name === 'TimeoutError') {
this.logger.log( this.logger.log(
`opencode notify accepted (response still pending after ${TIMEOUT_MS / 1000} s)`, `opencode notify accepted (response still pending after ${TIMEOUT_MS / 1000} s)`,
+101 -35
View File
@@ -5,7 +5,7 @@ import { DatabaseSync } from 'node:sqlite';
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
import type { LlmService } from '../llm/llm.service.js'; import type { LlmService } from '../llm/llm.service.js';
import { RssStore } from './rss.store.js'; import { RssStore } from './rss.store.js';
import { fitOnePacket, RssService } from './rss.service.js'; import { fitOneLine, parseSummaryItems, RssService } from './rss.service.js';
const FEED_A = `<rss version="2.0"><channel> const FEED_A = `<rss version="2.0"><channel>
<item><title>Alpha headline</title></item> <item><title>Alpha headline</title></item>
@@ -149,40 +149,79 @@ describe('RssService.handle', () => {
expect(await svc.handle(ctx('#rss sources'))).toBe('Aye, Bee'); expect(await svc.handle(ctx('#rss sources'))).toBe('Aye, Bee');
}); });
it('get <name>: summarizes the latest titles via chat with a bounded maxTokens', async () => { it('get <name>: summarizes via chat with a bounded maxTokens and replies with headered lines', async () => {
stored().add({ name: 'myfeed', url: 'https://a.example/rss', tags: [] }); stored().add({ name: 'myfeed', url: 'https://a.example/rss', tags: [] });
fetchMock.mockResolvedValueOnce(feedResponse(FEED_A)); fetchMock.mockResolvedValueOnce(feedResponse(FEED_A));
llm.chat.mockResolvedValueOnce('Alpha and beta happened.'); llm.chat.mockResolvedValueOnce(
'Chips: EU funds foundries.\nPark: Riverfront expansion approved.',
);
const reply = await svc.handle(ctx('#rss get myfeed')); const reply = await svc.handle(ctx('#rss get myfeed'));
expect(reply).toBe('Alpha and beta happened.'); expect(reply).toEqual([
'Chips: EU funds foundries.',
'Park: Riverfront expansion approved.',
]);
const [messages, opts] = llm.chat.mock.calls[0]; const [messages, opts] = llm.chat.mock.calls[0];
// headroom for the server's reasoning tokens (they count against max_tokens) — see rss.service.ts // headroom for the server's reasoning tokens (they count against max_tokens) — see rss.service.ts
expect(opts).toEqual({ maxTokens: 2500 }); expect(opts).toEqual({ maxTokens: 2500 });
const prompt = messages[0].content; const prompt = messages[0].content;
expect(prompt).toContain('at most 3 short bullet lines'); expect(prompt).toContain('exactly 3-5 lines');
expect(prompt).toContain('each under 45 characters'); expect(prompt).toContain('<short header>: <one or two short sentences>');
expect(prompt).toContain('no intro, no other text'); expect(prompt).toContain('at most 20 characters');
expect(prompt).toContain('No intro, no outro, no numbering');
expect(prompt).toContain('Alpha headline'); expect(prompt).toContain('Alpha headline');
expect(prompt).toContain('Beta headline'); expect(prompt).toContain('Beta headline');
}); });
it('get <name>: cuts an over-length LLM reply at bullet boundaries so it fits one packet', async () => { it('get <name>: caps items at 5', async () => {
stored().add({ name: 'myfeed', url: 'https://a.example/rss', tags: [] }); stored().add({ name: 'myfeed', url: 'https://a.example/rss', tags: [] });
fetchMock.mockResolvedValueOnce(feedResponse(FEED_A)); fetchMock.mockResolvedValueOnce(feedResponse(FEED_A));
// 5 bullets × 46 chars = 234 joined — over the 190-char one-packet cap const seven = ['a', 'b', 'c', 'd', 'e', 'f', 'g']
const tooLong = ['a', 'b', 'c', 'd', 'e'] .map((c) => `${c}: item ${c}`)
.map((c) => `-${c}`.padEnd(46, c))
.join('\n'); .join('\n');
llm.chat.mockResolvedValueOnce(tooLong); llm.chat.mockResolvedValueOnce(seven);
const reply = await svc.handle(ctx('#rss get myfeed')); const reply = await svc.handle(ctx('#rss get myfeed'));
expect(reply.length).toBeLessThanOrEqual(190); expect(reply).toEqual([
// the first 4 bullets fit (47*4-1=187), the 5th does not — and only complete lines survive 'a: item a',
expect(reply.split('\n')).toHaveLength(4); 'b: item b',
for (const line of reply.split('\n')) expect(line.length).toBe(46); 'c: item c',
'd: item d',
'e: item e',
]);
});
it('get <name>: drops empty and separator-less lines, sends what remains when fewer than 3 are valid', async () => {
stored().add({ name: 'myfeed', url: 'https://a.example/rss', tags: [] });
fetchMock.mockResolvedValueOnce(feedResponse(FEED_A));
llm.chat.mockResolvedValueOnce(
'Chips: EU funds foundries.\n\nJust a sentence without a header\n \nPark: expansion approved.',
);
const reply = await svc.handle(ctx('#rss get myfeed'));
expect(reply).toEqual([
'Chips: EU funds foundries.',
'Park: expansion approved.',
]);
});
it('get <name>: hard-cuts an over-length line at a word boundary so each item fits one packet', async () => {
stored().add({ name: 'myfeed', url: 'https://a.example/rss', tags: [] });
fetchMock.mockResolvedValueOnce(feedResponse(FEED_A));
const longLine = `Chips: ${'word '.repeat(40)}`; // 207 chars — over the per-line cap
llm.chat.mockResolvedValueOnce(`${longLine}\nPark: short line.`);
const reply = await svc.handle(ctx('#rss get myfeed'));
expect(Array.isArray(reply)).toBe(true);
const [cut, ok] = reply as string[];
expect(cut.length).toBeLessThanOrEqual(180);
expect(longLine.startsWith(cut)).toBe(true); // prefix — cut at a word boundary, not mid-word
expect(cut.endsWith(' ')).toBe(false);
expect(ok).toBe('Park: short line.'); // other lines untouched
}); });
it('get #tag: collects items across all sources with the tag; one dead feed does not sink it', async () => { it('get #tag: collects items across all sources with the tag; one dead feed does not sink it', async () => {
@@ -193,11 +232,11 @@ describe('RssService.handle', () => {
.mockResolvedValueOnce(feedResponse(FEED_A)) // A answers (Alpha, Beta) .mockResolvedValueOnce(feedResponse(FEED_A)) // A answers (Alpha, Beta)
.mockRejectedValueOnce(new Error('down')); // B is dead .mockRejectedValueOnce(new Error('down')); // B is dead
llm.chat.mockResolvedValueOnce('Only alpha survived.'); llm.chat.mockResolvedValueOnce('Chips: only alpha survived.');
const reply = await svc.handle(ctx('#rss get #tech')); const reply = await svc.handle(ctx('#rss get #tech'));
// B's fetch rejected, yet the summary still went out from A's titles alone // B's fetch rejected, yet the summary still went out from A's titles alone
expect(reply).toBe('Only alpha survived.'); expect(reply).toEqual(['Chips: only alpha survived.']);
expect(llm.chat).toHaveBeenCalledTimes(1); expect(llm.chat).toHaveBeenCalledTimes(1);
const content = llm.chat.mock.calls[0][0][0].content; const content = llm.chat.mock.calls[0][0][0].content;
expect(content).toContain('Alpha headline'); expect(content).toContain('Alpha headline');
@@ -210,9 +249,15 @@ describe('RssService.handle', () => {
.mockResolvedValueOnce(feedResponse(FEED_A)) // A → Alpha, Beta .mockResolvedValueOnce(feedResponse(FEED_A)) // A → Alpha, Beta
.mockResolvedValueOnce(feedResponse(FEED_B)); // B → Gamma .mockResolvedValueOnce(feedResponse(FEED_B)); // B → Gamma
llm.chat.mockResolvedValueOnce('all three.'); llm.chat.mockResolvedValueOnce(
await svc.handle(ctx('#rss get #tech')); 'Alpha: first headline.\nGamma: second headline.',
);
const reply = await svc.handle(ctx('#rss get #tech'));
expect(reply).toEqual([
'Alpha: first headline.',
'Gamma: second headline.',
]);
const content = llm.chat.mock.calls[0][0][0].content; const content = llm.chat.mock.calls[0][0][0].content;
expect(content).toContain('Alpha headline'); expect(content).toContain('Alpha headline');
expect(content).toContain('Gamma headline'); // proves B's feed was collected too expect(content).toContain('Gamma headline'); // proves B's feed was collected too
@@ -227,25 +272,46 @@ describe('RssService.handle', () => {
}); });
}); });
describe('fitOnePacket (one mesh packet ≤ 190 chars)', () => { describe('parseSummaryItems (per-line items for multi-message summaries)', () => {
it('passes through replies that already fit, including exactly 190 chars', () => { it('keeps lines with a header separator, drops empty and separator-less ones', () => {
const short = '- a\n- b'; expect(
expect(fitOnePacket(short)).toBe(short); parseSummaryItems(
const exact = 'x'.repeat(190); 'Chips: EU funds foundries.\n\nNo header here\n \nPark: expansion approved.',
expect(fitOnePacket(exact)).toBe(exact); ),
).toEqual(['Chips: EU funds foundries.', 'Park: expansion approved.']);
}); });
it('keeps complete bullet lines that fit and drops the rest', () => { it('caps at 5 items', () => {
// 46-char bullets: 4 fit (47*4-1=187), 5 do not (234) const seven = ['a', 'b', 'c', 'd', 'e', 'f', 'g']
const five = ['a', 'b', 'c', 'd', 'e'] .map((c) => `${c}: item`)
.map((c) => `-${c}`.padEnd(46, c))
.join('\n'); .join('\n');
expect(fitOnePacket(five).split('\n')).toHaveLength(4); expect(parseSummaryItems(seven)).toHaveLength(5);
expect(fitOnePacket(five).length).toBeLessThanOrEqual(190); expect(parseSummaryItems(seven)[4]).toBe('e: item');
}); });
it('hard-cuts a non-bulleted over-length paragraph rather than replying empty', () => { it('returns [] when nothing is valid', () => {
const para = 'w'.repeat(250); expect(parseSummaryItems('no headers at all\n')).toEqual([]);
expect(fitOnePacket(para)).toBe('w'.repeat(190)); });
});
describe('fitOneLine (one mesh packet per line)', () => {
it('passes through lines that already fit, including exactly 185 chars', () => {
const short = 'Chips: EU funds foundries.';
expect(fitOneLine(short)).toBe(short);
const exact = 'x'.repeat(185);
expect(fitOneLine(exact)).toBe(exact);
});
it('cuts over-long lines at a word boundary near 180', () => {
const long = `Chips: ${'word '.repeat(40)}`; // 207 chars
const cut = fitOneLine(long);
expect(cut.length).toBeLessThanOrEqual(180);
expect(long.startsWith(cut)).toBe(true); // prefix — not mid-word
expect(cut.endsWith(' ')).toBe(false);
});
it('hard-cuts at 180 when there is no word boundary in reach', () => {
const long = 'w'.repeat(250);
expect(fitOneLine(long)).toBe('w'.repeat(180));
}); });
}); });
+30 -25
View File
@@ -18,14 +18,16 @@ const TAG_MAX_TOKENS = 50;
// 2500 covers every observed run (worst case ≈42 s at ~60 tok/s, under LLM_TIMEOUT_MS=60 s). // 2500 covers every observed run (worst case ≈42 s at ~60 tok/s, under LLM_TIMEOUT_MS=60 s).
const SUMMARY_MAX_TOKENS = 2500; const SUMMARY_MAX_TOKENS = 2500;
// Mesh fragments replies longer than ~190 chars into fire-and-forget packets that recipients // Mesh fragments replies longer than ~190 chars into fire-and-forget packets that recipients
// often never get — a summary must fit ONE packet. The prompt asks for ≤3 bullets × <45 chars; // often never get — each summary line must fit ONE packet on its own. The prompt asks for short
// this is the hard backstop if the model ignores it. // headered lines; this is the hard backstop if the model ignores it (cut at a word boundary).
const MAX_REPLY_CHARS = 190; const MAX_LINE_CHARS = 185; // per-line one-packet threshold
const LINE_CUT_AT = 180; // where over-long lines get cut, snapped back to a word boundary
const MAX_ITEMS = 5; // cap on summary items sent as separate messages
/** /**
* The `#rss` command family: stored RSS sources (SQLite, shared MESH_DB_PATH file) with * The `#rss` command family: stored RSS sources (SQLite, shared MESH_DB_PATH file) with
* LLM-generated tags and on-demand headline summaries. All replies are short strings — * LLM-generated tags and on-demand headline summaries. Replies are short strings — or an array
* CommandService routes them through MeshService.respond() like any other command reply. * of lines for multi-message summaries, which CommandService sends one respond() per line.
*/ */
@Injectable() @Injectable()
export class RssService { export class RssService {
@@ -39,8 +41,8 @@ export class RssService {
this.store = new RssStore(db); this.store = new RssStore(db);
} }
/** Entry point for the `#rss` prefix — always replies (errors included). */ /** Entry point for the `#rss` prefix — always replies (errors included); `get` may reply with several lines. */
async handle(ctx: CommandContext): Promise<string> { async handle(ctx: CommandContext): Promise<string | string[]> {
const words = ctx.text.trim().split(/\s+/); const words = ctx.text.trim().split(/\s+/);
switch (words[1]) { switch (words[1]) {
case undefined: // bare `#rss` case undefined: // bare `#rss`
@@ -141,8 +143,8 @@ export class RssService {
return names.length ? names.join(', ') : 'no sources yet'; return names.length ? names.join(', ') : 'no sources yet';
} }
/** `#rss get <name>` or `#rss get #<tag>`. */ /** `#rss get <name>` or `#rss get #<tag>` — replies with 3-5 separate one-packet lines. */
private async get(target: string): Promise<string> { private async get(target: string): Promise<string | string[]> {
if (!target) return 'usage: #rss get <name|#tag>'; if (!target) return 'usage: #rss get <name|#tag>';
let urls: string[]; let urls: string[];
@@ -177,12 +179,14 @@ export class RssService {
[ [
{ {
role: 'user', role: 'user',
content: `Summarize these RSS headlines into at most 3 short bullet lines, each under 45 characters. Reply with the bullets only - no intro, no other text:\n${titles.join('\n')}`, content: `Summarize these RSS headlines into exactly 3-5 lines, one item per line, each in the format "<short header>: <one or two short sentences>" where the header is at most 20 characters (e.g. "Culture: Museum opens new wing"). No intro, no outro, no numbering - items only:\n${titles.join('\n')}`,
}, },
], ],
{ maxTokens: SUMMARY_MAX_TOKENS }, { maxTokens: SUMMARY_MAX_TOKENS },
); );
return fitOnePacket(reply); // each item goes out as its own mesh message — CommandService sends the array line by line;
// fewer than 3 valid lines is fine (the model is non-deterministic), we send what we got
return parseSummaryItems(reply).map(fitOneLine);
} catch (err) { } catch (err) {
this.logger.warn(`[rss] summary failed: ${(err as Error).message}`); this.logger.warn(`[rss] summary failed: ${(err as Error).message}`);
return 'summary failed'; return 'summary failed';
@@ -201,20 +205,21 @@ export class RssService {
} }
} }
/** Keep only complete bullet lines that fit one mesh packet (≤ MAX_REPLY_CHARS). */ /** Parse an LLM summary into per-line items: drop empty lines and lines without a `:` header separator, cap at MAX_ITEMS. */
export function fitOnePacket(reply: string): string { export function parseSummaryItems(reply: string): string[] {
if (reply.length <= MAX_REPLY_CHARS) return reply; return reply
const kept: string[] = []; .split('\n')
for (const line of reply.split('\n')) { .map((l) => l.trim())
const bullet = line.trim(); .filter((l) => l && l.includes(':'))
if (!bullet) continue; .slice(0, MAX_ITEMS);
// Σ(len+1): exactly the joined-with-\n length this would have AFTER appending `bullet` }
const total = kept.reduce((n, b) => n + b.length + 1, 0);
if (total + bullet.length > MAX_REPLY_CHARS) break; /** Hard-cut a single summary line at a word boundary near LINE_CUT_AT when it exceeds MAX_LINE_CHARS. */
kept.push(bullet); export function fitOneLine(line: string): string {
} if (line.length <= MAX_LINE_CHARS) return line;
// non-bulleted paragraph: a hard cut still beats an empty reply const slice = line.slice(0, LINE_CUT_AT);
return kept.length ? kept.join('\n') : reply.slice(0, MAX_REPLY_CHARS); const at = slice.lastIndexOf(' ');
return (at > 0 ? slice.slice(0, at) : slice).trimEnd();
} }
/** Normalize LLM tag output: lowercase, no #, whitespace → '-', deduped. */ /** Normalize LLM tag output: lowercase, no #, whitespace → '-', deduped. */
+28
View File
@@ -0,0 +1,28 @@
import { readdirSync, readFileSync } from 'node:fs';
import { join } from 'node:path';
import { fileURLToPath } from 'node:url';
import { describe, expect, it } from 'vitest';
const srcDir = fileURLToPath(new URL('.', import.meta.url));
function moduleFiles(dir: string): string[] {
return readdirSync(dir, { withFileTypes: true }).flatMap((entry) =>
entry.isDirectory()
? moduleFiles(join(dir, entry.name))
: entry.name.endsWith('.module.ts')
? [join(dir, entry.name)]
: [],
);
}
describe('ScheduleModule.forRoot single registration', () => {
// Regression guard (v0.21.7): each forRoot() call creates its own ScheduleExplorer that scans all providers, so a second call made every @Cron/@Interval job fire twice.
it('appears exactly once across all src/**/*.module.ts', () => {
const hits = moduleFiles(srcDir).flatMap((file) =>
(
readFileSync(file, 'utf8').match(/ScheduleModule\.forRoot\(/g) ?? []
).map(() => file.slice(srcDir.length)),
);
expect(hits).toHaveLength(1);
});
});