Coder
25d4e0e113
- Add lib/ (plain-ESM plugin code, no build step) so the published tarball actually contains the plugin; un-ignore lib/ for this package - Sync README: voice/audio transcription via home-lab Whishper (default large-v2, device cuda, language auto), /chats pagination and subagent hiding, whisper* config keys, updated security note
328 lines
11 KiB
JavaScript
328 lines
11 KiB
JavaScript
/**
|
|
* PluginState for dsh-telegram-remote: per-chat state, logging, push queue,
|
|
* prompt-echo suppression.
|
|
*/
|
|
|
|
import { appendFile, rename, stat } from "node:fs/promises";
|
|
import { readFileSync, writeFileSync, renameSync } from "node:fs";
|
|
import { join } from "node:path";
|
|
import { dshHome, nowIso, sleep, shortId } from "./util.js";
|
|
|
|
const LOG_MAX_BYTES = 2 * 1024 * 1024;
|
|
const LOG_ROTATE_EVERY = 64; // check size only every N writes
|
|
const STATE_DEBOUNCE_MS = 250;
|
|
|
|
export class PluginState {
|
|
ctx;
|
|
config;
|
|
bot;
|
|
logPath;
|
|
statePath;
|
|
chats = new Map();
|
|
recentPrompts = new Map();
|
|
jobStatuses = new Map();
|
|
pendingSends = new Map();
|
|
pendingOpts = new Map();
|
|
sendingChats = new Set();
|
|
restartScriptPath;
|
|
|
|
constructor(ctx, config, bot) {
|
|
this.ctx = ctx;
|
|
// Apply code-side defaults (the plugin deliberately ships no Config
|
|
// schema, so loader entries may omit any key).
|
|
this.config = {
|
|
botToken: "",
|
|
tokenEnv: "TELEGRAM_BOT_TOKEN",
|
|
tokenFile: "",
|
|
allowedUserIds: [],
|
|
ownerChatId: undefined,
|
|
workspaceRoot: process.cwd(),
|
|
defaultSessionId: "",
|
|
allowEval: true,
|
|
notifyOnStartup: true,
|
|
stateFile: "",
|
|
logFile: "",
|
|
pollTimeoutSec: 50,
|
|
maxOutputBytes: 120000,
|
|
whisperBaseUrl: "http://192.168.31.159:8082",
|
|
whisperModel: "large-v2",
|
|
whisperDevice: "cuda",
|
|
whisperLanguage: "auto",
|
|
whisperTimeoutMs: 120000,
|
|
whisperMaxBytes: 20971520,
|
|
...(config ?? {}),
|
|
};
|
|
this.bot = bot;
|
|
this.logPath = config.logFile || join(dshHome(), "telegram-remote.log");
|
|
this.statePath = config.stateFile || join(dshHome(), "telegram-remote-state.json");
|
|
this.restartScriptPath = join(dshHome(), "telegram-remote-restart.ps1");
|
|
// Batched async log writer: log() is fire-and-forget and never blocks
|
|
// the event loop (the old appendFileSync froze polling on every line).
|
|
// Lines accumulate and flush in one appendFile per microtask drain.
|
|
this._logBuffer = [];
|
|
this._logFlushing = false;
|
|
this._logWrites = 0;
|
|
this._stateTimer = null;
|
|
this._stateDirty = false;
|
|
this.loadState();
|
|
}
|
|
|
|
log(message) {
|
|
this._logBuffer.push("[" + nowIso() + "] " + message);
|
|
try { this.ctx.logger.info("[telegram-remote] " + message); } catch {}
|
|
if (!this._logFlushing) {
|
|
this._logFlushing = true;
|
|
queueMicrotask(() => {
|
|
const lines = this._logBuffer.splice(0);
|
|
this._logFlushing = false;
|
|
this._writeLog(lines);
|
|
});
|
|
}
|
|
}
|
|
|
|
/** Append a batch of lines; rotate the file lazily every N batches. */
|
|
async _writeLog(lines) {
|
|
if (lines.length === 0) return;
|
|
const path = this.logPath;
|
|
try {
|
|
await appendFile(path, lines.join("\n") + "\n", "utf8");
|
|
if (++this._logWrites % LOG_ROTATE_EVERY === 0) {
|
|
try {
|
|
const size = (await stat(path)).size;
|
|
if (size > LOG_MAX_BYTES) await rename(path, path + ".1");
|
|
} catch {}
|
|
}
|
|
} catch {}
|
|
}
|
|
|
|
loadState() {
|
|
try {
|
|
const raw = JSON.parse(readFileSync(this.statePath, "utf8"));
|
|
for (const [chatId, value] of Object.entries(raw.chats ?? {})) {
|
|
this.chats.set(Number(chatId), {
|
|
sessionId: value.sessionId ?? "",
|
|
notify: value.notify ?? "off",
|
|
});
|
|
}
|
|
if (typeof raw.botOffset === "number" && raw.botOffset > this.bot.offset) {
|
|
this.bot.offset = raw.botOffset;
|
|
}
|
|
} catch {}
|
|
}
|
|
|
|
/** Schedule a state write; coalesces bursts (one write per 250ms window). */
|
|
saveState() {
|
|
this._stateDirty = true;
|
|
if (this._stateTimer) return;
|
|
this._stateTimer = setTimeout(() => {
|
|
this._stateTimer = null;
|
|
if (this._stateDirty) {
|
|
this._stateDirty = false;
|
|
this.writeState();
|
|
}
|
|
}, STATE_DEBOUNCE_MS);
|
|
}
|
|
|
|
/** Write the state file immediately (reboot/shutdown/dispose paths). */
|
|
flushState() {
|
|
if (this._stateTimer) {
|
|
clearTimeout(this._stateTimer);
|
|
this._stateTimer = null;
|
|
}
|
|
this._stateDirty = false;
|
|
this.writeState();
|
|
}
|
|
|
|
writeState() {
|
|
const payload = {
|
|
chats: Object.fromEntries([...this.chats.entries()].map(([id, v]) => [String(id), v])),
|
|
botOffset: this.bot?.offset ?? 0,
|
|
};
|
|
try {
|
|
// Atomic: write a sibling temp file then rename, so a crash mid-write
|
|
// never leaves a truncated state file behind.
|
|
const tmp = this.statePath + ".tmp";
|
|
writeFileSync(tmp, JSON.stringify(payload, null, 2), "utf8");
|
|
try { renameSync(tmp, this.statePath); } catch { writeFileSync(this.statePath, JSON.stringify(payload, null, 2), "utf8"); }
|
|
} catch (error) {
|
|
this.log("state save failed: " + error.message);
|
|
}
|
|
}
|
|
|
|
chatState(chatId) {
|
|
let state = this.chats.get(chatId);
|
|
if (!state) {
|
|
state = { sessionId: this.config.defaultSessionId || "", notify: "session", sessionIds: [], lastListAt: 0 };
|
|
this.chats.set(chatId, state);
|
|
}
|
|
return state;
|
|
}
|
|
|
|
isAuthorized(userId) {
|
|
if (this.config.ownerChatId != null && userId === this.config.ownerChatId) return true;
|
|
return Array.isArray(this.config.allowedUserIds) && this.config.allowedUserIds.includes(userId);
|
|
}
|
|
|
|
async listSessions() {
|
|
const { callApi } = await import("./util.js");
|
|
try {
|
|
const result = await callApi(this.ctx, "sessions", "list", {});
|
|
const items = Array.isArray(result) ? result : (result?.items ?? []);
|
|
if (items.length > 0) return items;
|
|
} catch {}
|
|
// Fallback for profiles without the web api-proxy: read the live/corpus
|
|
// registries directly.
|
|
const items = [];
|
|
const sessions = this.ctx.get("sessions");
|
|
const sessionQuery = this.ctx.get("sessionQuery");
|
|
const seen = new Set();
|
|
if (sessions) {
|
|
for (const session of sessions.list()) {
|
|
seen.add(session.id);
|
|
items.push({
|
|
sessionId: session.id,
|
|
updatedAt: session.header?.updatedAt ?? session.header?.createdAt ?? 0,
|
|
running: this.ctx.get("agents")?.get(session.id)?.status === "running",
|
|
blank: !session.events.some((event) => event.type === "turn/start"),
|
|
cwd: session.header?.cwd,
|
|
origin: session.header?.origin,
|
|
});
|
|
}
|
|
}
|
|
if (sessionQuery) {
|
|
try {
|
|
for (const record of await sessionQuery.listSessions()) {
|
|
if (seen.has(record.header.id)) continue;
|
|
items.push({
|
|
sessionId: record.header.id,
|
|
updatedAt: record.header.createdAt ?? 0,
|
|
running: false,
|
|
blank: true,
|
|
cwd: record.header.cwd,
|
|
origin: record.header.origin,
|
|
});
|
|
}
|
|
} catch {}
|
|
}
|
|
items.sort((a, b) => (b.updatedAt ?? 0) - (a.updatedAt ?? 0));
|
|
return items;
|
|
}
|
|
|
|
/** Resolve a session id argument: full id, prefix, "last", or the chat's active session. */
|
|
async resolveSessionId(chatId, arg) {
|
|
const wanted = String(arg ?? "").trim();
|
|
let sessionId = wanted;
|
|
if (!sessionId) {
|
|
sessionId = this.chatState(chatId).sessionId;
|
|
if (!sessionId) return null;
|
|
return sessionId;
|
|
}
|
|
if (sessionId === "last") {
|
|
const summaries = await this.listSessions();
|
|
if (!summaries || summaries.length === 0) return null;
|
|
return summaries[0].sessionId;
|
|
}
|
|
const summaries = await this.listSessions();
|
|
if (summaries) {
|
|
const match = summaries.find((item) => item.sessionId === sessionId);
|
|
const prefix = summaries.find((item) => item.sessionId.startsWith(sessionId));
|
|
return (match ?? prefix ?? null)?.sessionId ?? sessionId;
|
|
}
|
|
return sessionId;
|
|
}
|
|
|
|
notePrompt(sessionId, text) {
|
|
const list = this.recentPrompts.get(sessionId) ?? [];
|
|
list.push({ text: String(text).trim().slice(0, 200), at: Date.now() });
|
|
while (list.length > 20) list.shift();
|
|
this.recentPrompts.set(sessionId, list);
|
|
// Bounded memory: drop sessions whose prompts all expired long ago.
|
|
if (this.recentPrompts.size > 64) {
|
|
const now = Date.now();
|
|
for (const [sid, entries] of this.recentPrompts) {
|
|
const fresh = entries.filter((entry) => now - entry.at < 600_000);
|
|
if (fresh.length === 0) this.recentPrompts.delete(sid);
|
|
else this.recentPrompts.set(sid, fresh);
|
|
}
|
|
}
|
|
}
|
|
|
|
isRecentPrompt(sessionId, text) {
|
|
const list = this.recentPrompts.get(sessionId) ?? [];
|
|
const needle = String(text).trim().slice(0, 200);
|
|
const now = Date.now();
|
|
return list.some((entry) => now - entry.at < 90_000 && entry.text === needle);
|
|
}
|
|
|
|
/** Record a job status transition; keeps only the most recent entries. */
|
|
noteJobStatus(jobId, status) {
|
|
this.jobStatuses.set(jobId, status);
|
|
if (this.jobStatuses.size > 256) {
|
|
const oldest = this.jobStatuses.keys().next().value;
|
|
this.jobStatuses.delete(oldest);
|
|
}
|
|
}
|
|
|
|
async #drain(chatId) {
|
|
if (this.sendingChats.has(chatId)) return;
|
|
this.sendingChats.add(chatId);
|
|
try {
|
|
for (;;) {
|
|
const text = this.pendingSends.get(chatId);
|
|
if (text === undefined) break;
|
|
this.pendingSends.delete(chatId);
|
|
const opts = this.pendingOpts.get(chatId) ?? {};
|
|
this.pendingOpts.delete(chatId);
|
|
try {
|
|
await this.bot.send(chatId, text, opts.replyTo ? { replyToMessageId: opts.replyTo } : {});
|
|
} catch (error) {
|
|
this.log("push to chat " + chatId + " failed: " + error.message);
|
|
const again = this.pendingSends.get(chatId);
|
|
this.pendingSends.set(chatId, (again ? again + "\n\n" : "") + text);
|
|
this.pendingOpts.set(chatId, opts);
|
|
await sleep(5000);
|
|
break;
|
|
}
|
|
}
|
|
} finally {
|
|
this.sendingChats.delete(chatId);
|
|
}
|
|
}
|
|
|
|
/**
|
|
* Push a session event to interested chats.
|
|
* @param category - "reply" (assistant replies, turn ends: delivered to
|
|
* every chat with notify session/all) or "detail" (tool calls, web-side
|
|
* user messages, errors: only notify=all chats — too noisy otherwise).
|
|
*/
|
|
notifySession(sessionId, text, category = "reply") {
|
|
for (const [chatId, state] of this.chats) {
|
|
if (state.notify === "off") continue;
|
|
if (state.notify === "session" && sessionId !== state.sessionId) continue;
|
|
if (category === "detail" && state.notify !== "all") continue;
|
|
// The active session reads as a plain conversation — no session id
|
|
// prefix. Other sessions (notify=all) get a small label to distinguish.
|
|
const body = sessionId === state.sessionId
|
|
? text
|
|
: "<b>" + shortId(sessionId) + "</b>\n" + text;
|
|
this.push(chatId, body, { replyTo: state.lastUserMessageId });
|
|
}
|
|
}
|
|
|
|
/** Coalescing push sender: one in-flight send per chat, pending text merged. */
|
|
push(chatId, text, opts = {}) {
|
|
const existing = this.pendingSends.get(chatId);
|
|
if (existing !== undefined) {
|
|
this.pendingSends.set(chatId, existing + "\n\n" + text);
|
|
return;
|
|
}
|
|
this.pendingSends.set(chatId, text);
|
|
this.pendingOpts.set(chatId, opts);
|
|
void this.#drain(chatId);
|
|
}
|
|
|
|
notifyAll(text) {
|
|
for (const chatId of this.chats.keys()) this.push(chatId, text);
|
|
}
|
|
}
|