Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
b54f909fbc | ||
|
|
7ec97a06d2 | ||
|
|
84915fe314 | ||
|
|
5506fb9330 | ||
|
|
beebcd5851 | ||
|
|
5a1493fb4a | ||
|
|
8fb16217f7 | ||
|
|
3b1f5de101 | ||
|
|
46d32b3553 | ||
|
|
e20e11e2d5 | ||
|
|
0b20df242c | ||
|
|
a16b6c19d4 | ||
|
|
9e558fd95e | ||
|
|
70be8bdaa7 | ||
|
|
847c3d47d0 | ||
|
|
1b768b14a7 | ||
|
|
3f6c02d71e | ||
|
|
98b9ec5185 | ||
|
|
0722c82359 | ||
|
|
1ae8d7a926 | ||
|
|
77459adc13 | ||
|
|
33a88f3d83 | ||
|
|
306756c63b | ||
|
|
af1c77fd54 | ||
|
|
73ce49c741 | ||
|
|
cdd52a3ddb |
@@ -1,5 +1,73 @@
|
||||
# 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
|
||||
|
||||
- `#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.
|
||||
|
||||
@@ -2,6 +2,7 @@ services:
|
||||
bot:
|
||||
build: .
|
||||
environment:
|
||||
TZ: Europe/Warsaw
|
||||
MESH_HOST: 192.168.1.4
|
||||
MESH_PORT: '4403'
|
||||
MESH_DB_PATH: ./data/mesh-bot.db
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "mesh-bot",
|
||||
"version": "0.20.1",
|
||||
"version": "0.21.12",
|
||||
"description": "",
|
||||
"author": "",
|
||||
"private": true,
|
||||
|
||||
@@ -1,10 +1,9 @@
|
||||
import { Module } from '@nestjs/common';
|
||||
import { ScheduleModule } from '@nestjs/schedule';
|
||||
import { MeshModule } from '../mesh/mesh.module.js';
|
||||
import { BatteryAlertService } from './battery.service.js';
|
||||
|
||||
@Module({
|
||||
imports: [ScheduleModule.forRoot(), MeshModule],
|
||||
imports: [MeshModule],
|
||||
providers: [BatteryAlertService],
|
||||
exports: [BatteryAlertService],
|
||||
})
|
||||
|
||||
@@ -119,6 +119,47 @@ describe('BatteryAlertService check threshold', () => {
|
||||
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 () => {
|
||||
const sendText = vi.fn(async (..._args: unknown[]) => 1);
|
||||
const svc = makeService(
|
||||
|
||||
@@ -60,6 +60,7 @@ export class BatteryAlertService {
|
||||
|
||||
private shouldAlert(node: NodeSummary, now: number): boolean {
|
||||
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)
|
||||
if (!(node.batteryLevel <= this.threshold)) return false;
|
||||
const last = this.lastAlertAt.get(node.num);
|
||||
|
||||
@@ -13,7 +13,11 @@ import {
|
||||
KAMIL_NUM,
|
||||
} from '../mesh/fake-bridge.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 { CommandService } from './command.service.js';
|
||||
|
||||
@@ -30,6 +34,7 @@ function makeService(
|
||||
|
||||
beforeEach(() => {
|
||||
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', () => {
|
||||
@@ -40,6 +45,16 @@ describe('CommandService.onRx', () => {
|
||||
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 () => {
|
||||
const [svc, mesh] = makeService();
|
||||
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');
|
||||
});
|
||||
|
||||
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 () => {
|
||||
const [svc, mesh] = makeService();
|
||||
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', () => {
|
||||
let dir: string | undefined;
|
||||
|
||||
|
||||
@@ -1,4 +1,6 @@
|
||||
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';
|
||||
// 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';
|
||||
@@ -14,10 +16,10 @@ export interface CommandContext {
|
||||
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 = (
|
||||
ctx: CommandContext,
|
||||
) => string | undefined | Promise<string | undefined>;
|
||||
) => string | string[] | undefined | Promise<string | string[] | undefined>;
|
||||
|
||||
@Injectable()
|
||||
export class CommandService {
|
||||
@@ -27,6 +29,11 @@ export class CommandService {
|
||||
process.env.COMMAND_CHANNEL ?? '2',
|
||||
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())
|
||||
private readonly commands = new Map<string, CommandHandler>();
|
||||
// prefix handlers own a whole family (`#rss`, `#rss add ...`) instead of one exact name
|
||||
@@ -38,9 +45,68 @@ export class CommandService {
|
||||
rss?: RssService,
|
||||
) {
|
||||
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));
|
||||
}
|
||||
|
||||
/**
|
||||
* #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(name: string, fn: CommandHandler): void {
|
||||
this.commands.set(name, fn);
|
||||
@@ -83,18 +149,28 @@ export class CommandService {
|
||||
try {
|
||||
const reply = await handler({ text: trimmed, from, id });
|
||||
if (reply === undefined || reply === null) return; // void = no reply
|
||||
// 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)
|
||||
const lines = Array.isArray(reply) ? reply : [reply];
|
||||
for (let i = 0; i < lines.length; i++) {
|
||||
if (i > 0 && this.multiSendDelayMs > 0) {
|
||||
await new Promise((resolve) =>
|
||||
setTimeout(resolve, this.multiSendDelayMs),
|
||||
);
|
||||
}
|
||||
const line = lines[i];
|
||||
if (id !== undefined && this.mesh) {
|
||||
// respond() auto-routes the reply: rides the original record's channel with replyId = id (threaded quote)
|
||||
await this.mesh.respond(id, reply);
|
||||
// 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,
|
||||
reply,
|
||||
line,
|
||||
undefined,
|
||||
this.commandChannel,
|
||||
);
|
||||
}
|
||||
}
|
||||
} catch (err) {
|
||||
this.logger.warn(
|
||||
`[commands] '${trimmed}' failed: ${(err as Error).message}`,
|
||||
|
||||
@@ -12,9 +12,9 @@ export class LlmService {
|
||||
private readonly apiBase =
|
||||
process.env.LLM_API_BASE ?? 'http://192.168.1.70:1111/v1';
|
||||
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(
|
||||
process.env.LLM_TIMEOUT_MS ?? '60000',
|
||||
process.env.LLM_TIMEOUT_MS ?? '180000',
|
||||
10,
|
||||
);
|
||||
|
||||
|
||||
@@ -10,6 +10,8 @@ export function startBridge(
|
||||
): Promise<{ port: number; sockets: net.Socket[]; close: () => void }> {
|
||||
const sockets: net.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
|
||||
sockets.push(socket);
|
||||
});
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
import { create, fromBinary, toBinary } from '@bufbuild/protobuf';
|
||||
import * as P from '@meshtastic/protobufs';
|
||||
import { BadRequestException, NotFoundException } from '@nestjs/common';
|
||||
import * as net from 'node:net';
|
||||
import { DatabaseSync } from 'node:sqlite';
|
||||
import { beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
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', () => {
|
||||
it('sends on the test channel (index 3) by default, never primary', async () => {
|
||||
const bridge = await startBridge();
|
||||
|
||||
@@ -81,6 +81,8 @@ export interface TelemetryRecord {
|
||||
const HISTORY_LIMIT = 100;
|
||||
/** Rows kept in SQLite across restarts; older rows are pruned on every poller tick. */
|
||||
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)
|
||||
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)
|
||||
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). */
|
||||
private db: DatabaseSync | null = null;
|
||||
@@ -255,6 +272,10 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
|
||||
// telemetry history (newest kept, capped at HISTORY_LIMIT)
|
||||
private telemetryHistory: TelemetryRecord[] = [];
|
||||
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;
|
||||
|
||||
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
|
||||
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();
|
||||
}
|
||||
|
||||
@@ -271,6 +301,8 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
|
||||
this.shuttingDown = true;
|
||||
if (this.reconnectTimer) clearTimeout(this.reconnectTimer);
|
||||
if (this.pollTimer) clearInterval(this.pollTimer);
|
||||
if (this.idleTimer) clearInterval(this.idleTimer);
|
||||
if (this.periodicReconnectTimer) clearInterval(this.periodicReconnectTimer);
|
||||
this.socket?.destroy();
|
||||
this.db?.close();
|
||||
}
|
||||
@@ -641,11 +673,17 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
|
||||
private connect(): void {
|
||||
if (this.shuttingDown) return;
|
||||
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;
|
||||
socket.on('connect', () => {
|
||||
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;
|
||||
// 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.nodeMap = new Map();
|
||||
this.metadata = null;
|
||||
@@ -666,9 +704,19 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
|
||||
const merged = new Uint8Array(this.buffer.length + chunk.length);
|
||||
merged.set(this.buffer, 0);
|
||||
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;
|
||||
const { packets, rest } = extractFrames(this.buffer);
|
||||
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) {
|
||||
decodeFromRadio(p)
|
||||
.then((fr) => this.handleFromRadio(fr))
|
||||
@@ -694,6 +742,25 @@ export class MeshService implements OnModuleInit, OnModuleDestroy {
|
||||
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(
|
||||
fr: import('@meshtastic/protobufs').Mesh.FromRadio,
|
||||
): Promise<void> {
|
||||
|
||||
@@ -72,7 +72,7 @@ describe('OpencodeService.notify', () => {
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(2);
|
||||
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 {
|
||||
method?: string;
|
||||
headers?: Record<string, string>;
|
||||
@@ -83,15 +83,10 @@ describe('OpencodeService.notify', () => {
|
||||
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
|
||||
// body is EXACTLY {"text": msg} — em-dash, not a hyphen — and a 2xx resolves notify as delivered
|
||||
expect(String(opts.body)).toBe(
|
||||
JSON.stringify({
|
||||
parts: [
|
||||
{
|
||||
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];
|
||||
expect(String((init as { body?: string }).body)).toBe(
|
||||
JSON.stringify({
|
||||
parts: [
|
||||
{
|
||||
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');
|
||||
|
||||
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');
|
||||
|
||||
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
|
||||
.mockResolvedValueOnce(json([WORKSPACE]))
|
||||
.mockResolvedValueOnce(
|
||||
json([
|
||||
{ id: 'ses_old', time: { updated: 100 } },
|
||||
{ id: 'ses_new', time: { updated: 200 } },
|
||||
]),
|
||||
// v2 envelope {data, cursor}; the /elsewhere session is more recent but a different
|
||||
// directory, so the location.directory filter must exclude it
|
||||
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({}));
|
||||
|
||||
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(
|
||||
`${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' });
|
||||
fetchMock
|
||||
.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({}));
|
||||
|
||||
await svc.notify(7, 'hi');
|
||||
|
||||
expect(fetchMock.mock.calls[1][0]).toBe(`${API}/proxy/abc/api/session`);
|
||||
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');
|
||||
|
||||
// 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.mock.calls[0][0]).toBe(`${API}/api/workspaces`);
|
||||
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
|
||||
.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');
|
||||
await expect(svc.notify('hi')).resolves.toBeUndefined();
|
||||
|
||||
@@ -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
|
||||
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;
|
||||
|
||||
/** Entry of GET {api}/api/workspaces (only the fields we use). */
|
||||
@@ -14,9 +15,11 @@ interface WorkspaceEntry {
|
||||
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 {
|
||||
id: string;
|
||||
title?: string;
|
||||
location?: { directory?: string };
|
||||
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 —
|
||||
* 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).
|
||||
* Push an rx mesh message into the opencode session (v2: POST {base}/api/session/{id}/prompt with a
|
||||
* `{text}` body). Resolves when delivered — including on a final-POST read timeout, which still means
|
||||
* DELIVERED (kept as a harmless safety net; v2 normally returns 2xx immediately) 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(id: number, text: string, from?: number): Promise<void> {
|
||||
if (!this.enabled) return;
|
||||
@@ -112,12 +115,19 @@ export class OpencodeService {
|
||||
const base = this.api + (ws.proxyPath ?? '');
|
||||
let sid = this.sessionOverride;
|
||||
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)
|
||||
throw new Error(
|
||||
`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;
|
||||
sid = sessions.some((s) => updated(s))
|
||||
? sessions.reduce((a, b) => (updated(b) > updated(a) ? b : a)).id
|
||||
@@ -148,16 +158,17 @@ export class OpencodeService {
|
||||
msg: string,
|
||||
): Promise<void> {
|
||||
try {
|
||||
const res = await fetch(`${base}/session/${sid}/message`, {
|
||||
const res = await fetch(`${base}/api/session/${sid}/prompt`, {
|
||||
method: 'POST',
|
||||
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),
|
||||
});
|
||||
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
|
||||
// a read timeout here means DELIVERED, not failed — kept as a harmless safety net: v2 normally
|
||||
// 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') {
|
||||
this.logger.log(
|
||||
`opencode notify accepted (response still pending after ${TIMEOUT_MS / 1000} s)`,
|
||||
|
||||
+101
-35
@@ -5,7 +5,7 @@ import { DatabaseSync } from 'node:sqlite';
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
|
||||
import type { LlmService } from '../llm/llm.service.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>
|
||||
<item><title>Alpha headline</title></item>
|
||||
@@ -149,40 +149,79 @@ describe('RssService.handle', () => {
|
||||
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: [] });
|
||||
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'));
|
||||
|
||||
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];
|
||||
// headroom for the server's reasoning tokens (they count against max_tokens) — see rss.service.ts
|
||||
expect(opts).toEqual({ maxTokens: 2500 });
|
||||
const prompt = messages[0].content;
|
||||
expect(prompt).toContain('at most 3 short bullet lines');
|
||||
expect(prompt).toContain('each under 45 characters');
|
||||
expect(prompt).toContain('no intro, no other text');
|
||||
expect(prompt).toContain('exactly 3-5 lines');
|
||||
expect(prompt).toContain('<short header>: <one or two short sentences>');
|
||||
expect(prompt).toContain('at most 20 characters');
|
||||
expect(prompt).toContain('No intro, no outro, no numbering');
|
||||
expect(prompt).toContain('Alpha 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: [] });
|
||||
fetchMock.mockResolvedValueOnce(feedResponse(FEED_A));
|
||||
// 5 bullets × 46 chars = 234 joined — over the 190-char one-packet cap
|
||||
const tooLong = ['a', 'b', 'c', 'd', 'e']
|
||||
.map((c) => `-${c}`.padEnd(46, c))
|
||||
const seven = ['a', 'b', 'c', 'd', 'e', 'f', 'g']
|
||||
.map((c) => `${c}: item ${c}`)
|
||||
.join('\n');
|
||||
llm.chat.mockResolvedValueOnce(tooLong);
|
||||
llm.chat.mockResolvedValueOnce(seven);
|
||||
|
||||
const reply = await svc.handle(ctx('#rss get myfeed'));
|
||||
|
||||
expect(reply.length).toBeLessThanOrEqual(190);
|
||||
// the first 4 bullets fit (47*4-1=187), the 5th does not — and only complete lines survive
|
||||
expect(reply.split('\n')).toHaveLength(4);
|
||||
for (const line of reply.split('\n')) expect(line.length).toBe(46);
|
||||
expect(reply).toEqual([
|
||||
'a: item a',
|
||||
'b: item b',
|
||||
'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 () => {
|
||||
@@ -193,11 +232,11 @@ describe('RssService.handle', () => {
|
||||
.mockResolvedValueOnce(feedResponse(FEED_A)) // A answers (Alpha, Beta)
|
||||
.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'));
|
||||
|
||||
// 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);
|
||||
const content = llm.chat.mock.calls[0][0][0].content;
|
||||
expect(content).toContain('Alpha headline');
|
||||
@@ -210,9 +249,15 @@ describe('RssService.handle', () => {
|
||||
.mockResolvedValueOnce(feedResponse(FEED_A)) // A → Alpha, Beta
|
||||
.mockResolvedValueOnce(feedResponse(FEED_B)); // B → Gamma
|
||||
|
||||
llm.chat.mockResolvedValueOnce('all three.');
|
||||
await svc.handle(ctx('#rss get #tech'));
|
||||
llm.chat.mockResolvedValueOnce(
|
||||
'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;
|
||||
expect(content).toContain('Alpha headline');
|
||||
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)', () => {
|
||||
it('passes through replies that already fit, including exactly 190 chars', () => {
|
||||
const short = '- a\n- b';
|
||||
expect(fitOnePacket(short)).toBe(short);
|
||||
const exact = 'x'.repeat(190);
|
||||
expect(fitOnePacket(exact)).toBe(exact);
|
||||
describe('parseSummaryItems (per-line items for multi-message summaries)', () => {
|
||||
it('keeps lines with a header separator, drops empty and separator-less ones', () => {
|
||||
expect(
|
||||
parseSummaryItems(
|
||||
'Chips: EU funds foundries.\n\nNo header here\n \nPark: expansion approved.',
|
||||
),
|
||||
).toEqual(['Chips: EU funds foundries.', 'Park: expansion approved.']);
|
||||
});
|
||||
|
||||
it('keeps complete bullet lines that fit and drops the rest', () => {
|
||||
// 46-char bullets: 4 fit (47*4-1=187), 5 do not (234)
|
||||
const five = ['a', 'b', 'c', 'd', 'e']
|
||||
.map((c) => `-${c}`.padEnd(46, c))
|
||||
it('caps at 5 items', () => {
|
||||
const seven = ['a', 'b', 'c', 'd', 'e', 'f', 'g']
|
||||
.map((c) => `${c}: item`)
|
||||
.join('\n');
|
||||
expect(fitOnePacket(five).split('\n')).toHaveLength(4);
|
||||
expect(fitOnePacket(five).length).toBeLessThanOrEqual(190);
|
||||
expect(parseSummaryItems(seven)).toHaveLength(5);
|
||||
expect(parseSummaryItems(seven)[4]).toBe('e: item');
|
||||
});
|
||||
|
||||
it('hard-cuts a non-bulleted over-length paragraph rather than replying empty', () => {
|
||||
const para = 'w'.repeat(250);
|
||||
expect(fitOnePacket(para)).toBe('w'.repeat(190));
|
||||
it('returns [] when nothing is valid', () => {
|
||||
expect(parseSummaryItems('no headers at all\n')).toEqual([]);
|
||||
});
|
||||
});
|
||||
|
||||
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));
|
||||
});
|
||||
});
|
||||
|
||||
+29
-24
@@ -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).
|
||||
const SUMMARY_MAX_TOKENS = 2500;
|
||||
// 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;
|
||||
// this is the hard backstop if the model ignores it.
|
||||
const MAX_REPLY_CHARS = 190;
|
||||
// often never get — each summary line must fit ONE packet on its own. The prompt asks for short
|
||||
// headered lines; this is the hard backstop if the model ignores it (cut at a word boundary).
|
||||
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
|
||||
* LLM-generated tags and on-demand headline summaries. All replies are short strings —
|
||||
* CommandService routes them through MeshService.respond() like any other command reply.
|
||||
* LLM-generated tags and on-demand headline summaries. Replies are short strings — or an array
|
||||
* of lines for multi-message summaries, which CommandService sends one respond() per line.
|
||||
*/
|
||||
@Injectable()
|
||||
export class RssService {
|
||||
@@ -39,8 +41,8 @@ export class RssService {
|
||||
this.store = new RssStore(db);
|
||||
}
|
||||
|
||||
/** Entry point for the `#rss` prefix — always replies (errors included). */
|
||||
async handle(ctx: CommandContext): Promise<string> {
|
||||
/** Entry point for the `#rss` prefix — always replies (errors included); `get` may reply with several lines. */
|
||||
async handle(ctx: CommandContext): Promise<string | string[]> {
|
||||
const words = ctx.text.trim().split(/\s+/);
|
||||
switch (words[1]) {
|
||||
case undefined: // bare `#rss`
|
||||
@@ -141,8 +143,8 @@ export class RssService {
|
||||
return names.length ? names.join(', ') : 'no sources yet';
|
||||
}
|
||||
|
||||
/** `#rss get <name>` or `#rss get #<tag>`. */
|
||||
private async get(target: string): Promise<string> {
|
||||
/** `#rss get <name>` or `#rss get #<tag>` — replies with 3-5 separate one-packet lines. */
|
||||
private async get(target: string): Promise<string | string[]> {
|
||||
if (!target) return 'usage: #rss get <name|#tag>';
|
||||
|
||||
let urls: string[];
|
||||
@@ -177,12 +179,14 @@ export class RssService {
|
||||
[
|
||||
{
|
||||
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 },
|
||||
);
|
||||
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) {
|
||||
this.logger.warn(`[rss] summary failed: ${(err as Error).message}`);
|
||||
return 'summary failed';
|
||||
@@ -201,20 +205,21 @@ export class RssService {
|
||||
}
|
||||
}
|
||||
|
||||
/** Keep only complete bullet lines that fit one mesh packet (≤ MAX_REPLY_CHARS). */
|
||||
export function fitOnePacket(reply: string): string {
|
||||
if (reply.length <= MAX_REPLY_CHARS) return reply;
|
||||
const kept: string[] = [];
|
||||
for (const line of reply.split('\n')) {
|
||||
const bullet = line.trim();
|
||||
if (!bullet) continue;
|
||||
// Σ(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;
|
||||
kept.push(bullet);
|
||||
/** Parse an LLM summary into per-line items: drop empty lines and lines without a `:` header separator, cap at MAX_ITEMS. */
|
||||
export function parseSummaryItems(reply: string): string[] {
|
||||
return reply
|
||||
.split('\n')
|
||||
.map((l) => l.trim())
|
||||
.filter((l) => l && l.includes(':'))
|
||||
.slice(0, MAX_ITEMS);
|
||||
}
|
||||
// non-bulleted paragraph: a hard cut still beats an empty reply
|
||||
return kept.length ? kept.join('\n') : reply.slice(0, MAX_REPLY_CHARS);
|
||||
|
||||
/** Hard-cut a single summary line at a word boundary near LINE_CUT_AT when it exceeds MAX_LINE_CHARS. */
|
||||
export function fitOneLine(line: string): string {
|
||||
if (line.length <= MAX_LINE_CHARS) return line;
|
||||
const slice = line.slice(0, LINE_CUT_AT);
|
||||
const at = slice.lastIndexOf(' ');
|
||||
return (at > 0 ? slice.slice(0, at) : slice).trimEnd();
|
||||
}
|
||||
|
||||
/** Normalize LLM tag output: lowercase, no #, whitespace → '-', deduped. */
|
||||
|
||||
@@ -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);
|
||||
});
|
||||
});
|
||||
Reference in New Issue
Block a user