Merge branch 'main' into fix/secondary-agent-oauth-fallback

src/agents/session-write-lock.test.ts

@@ -3,7 +3,7 @@ import os from "node:os";
 import path from "node:path";
 import { describe, expect, it } from "vitest";
 
-import { acquireSessionWriteLock } from "./session-write-lock.js";
+import { __testing, acquireSessionWriteLock } from "./session-write-lock.js";
 
 describe("acquireSessionWriteLock", () => {
   it("reuses locks across symlinked session paths", async () => {
@@ -31,4 +31,132 @@ describe("acquireSessionWriteLock", () => {
       await fs.rm(root, { recursive: true, force: true });
     }
   });
+
+  it("keeps the lock file until the last release", async () => {
+    const root = await fs.mkdtemp(path.join(os.tmpdir(), "clawdbot-lock-"));
+    try {
+      const sessionFile = path.join(root, "sessions.json");
+      const lockPath = `${sessionFile}.lock`;
+
+      const lockA = await acquireSessionWriteLock({ sessionFile, timeoutMs: 500 });
+      const lockB = await acquireSessionWriteLock({ sessionFile, timeoutMs: 500 });
+
+      await expect(fs.access(lockPath)).resolves.toBeUndefined();
+      await lockA.release();
+      await expect(fs.access(lockPath)).resolves.toBeUndefined();
+      await lockB.release();
+      await expect(fs.access(lockPath)).rejects.toThrow();
+    } finally {
+      await fs.rm(root, { recursive: true, force: true });
+    }
+  });
+
+  it("reclaims stale lock files", async () => {
+    const root = await fs.mkdtemp(path.join(os.tmpdir(), "clawdbot-lock-"));
+    try {
+      const sessionFile = path.join(root, "sessions.json");
+      const lockPath = `${sessionFile}.lock`;
+      await fs.writeFile(
+        lockPath,
+        JSON.stringify({ pid: 123456, createdAt: new Date(Date.now() - 60_000).toISOString() }),
+        "utf8",
+      );
+
+      const lock = await acquireSessionWriteLock({ sessionFile, timeoutMs: 500, staleMs: 10 });
+      const raw = await fs.readFile(lockPath, "utf8");
+      const payload = JSON.parse(raw) as { pid: number };
+
+      expect(payload.pid).toBe(process.pid);
+      await lock.release();
+    } finally {
+      await fs.rm(root, { recursive: true, force: true });
+    }
+  });
+
+  it("removes held locks on termination signals", async () => {
+    const signals = ["SIGINT", "SIGTERM", "SIGQUIT", "SIGABRT"] as const;
+    for (const signal of signals) {
+      const root = await fs.mkdtemp(path.join(os.tmpdir(), "clawdbot-lock-cleanup-"));
+      try {
+        const sessionFile = path.join(root, "sessions.json");
+        const lockPath = `${sessionFile}.lock`;
+        await acquireSessionWriteLock({ sessionFile, timeoutMs: 500 });
+        const keepAlive = () => {};
+        if (signal === "SIGINT") {
+          process.on(signal, keepAlive);
+        }
+
+        __testing.handleTerminationSignal(signal);
+
+        await expect(fs.stat(lockPath)).rejects.toThrow();
+        if (signal === "SIGINT") {
+          process.off(signal, keepAlive);
+        }
+      } finally {
+        await fs.rm(root, { recursive: true, force: true });
+      }
+    }
+  });
+
+  it("registers cleanup for SIGQUIT and SIGABRT", () => {
+    expect(__testing.cleanupSignals).toContain("SIGQUIT");
+    expect(__testing.cleanupSignals).toContain("SIGABRT");
+  });
+  it("cleans up locks on SIGINT without removing other handlers", async () => {
+    const root = await fs.mkdtemp(path.join(os.tmpdir(), "clawdbot-lock-"));
+    const originalKill = process.kill.bind(process);
+    const killCalls: Array<NodeJS.Signals | undefined> = [];
+    let otherHandlerCalled = false;
+
+    process.kill = ((pid: number, signal?: NodeJS.Signals) => {
+      killCalls.push(signal);
+      return true;
+    }) as typeof process.kill;
+
+    const otherHandler = () => {
+      otherHandlerCalled = true;
+    };
+
+    process.on("SIGINT", otherHandler);
+
+    try {
+      const sessionFile = path.join(root, "sessions.json");
+      const lockPath = `${sessionFile}.lock`;
+      await acquireSessionWriteLock({ sessionFile, timeoutMs: 500 });
+
+      process.emit("SIGINT");
+
+      await expect(fs.access(lockPath)).rejects.toThrow();
+      expect(otherHandlerCalled).toBe(true);
+      expect(killCalls).toEqual([]);
+    } finally {
+      process.off("SIGINT", otherHandler);
+      process.kill = originalKill;
+      await fs.rm(root, { recursive: true, force: true });
+    }
+  });
+
+  it("cleans up locks on exit", async () => {
+    const root = await fs.mkdtemp(path.join(os.tmpdir(), "clawdbot-lock-"));
+    try {
+      const sessionFile = path.join(root, "sessions.json");
+      const lockPath = `${sessionFile}.lock`;
+      await acquireSessionWriteLock({ sessionFile, timeoutMs: 500 });
+
+      process.emit("exit", 0);
+
+      await expect(fs.access(lockPath)).rejects.toThrow();
+    } finally {
+      await fs.rm(root, { recursive: true, force: true });
+    }
+  });
+  it("keeps other signal listeners registered", () => {
+    const keepAlive = () => {};
+    process.on("SIGINT", keepAlive);
+
+    __testing.handleTerminationSignal("SIGINT");
+
+    expect(process.listeners("SIGINT")).toContain(keepAlive);
+    process.off("SIGINT", keepAlive);
+  });
 });

src/agents/session-write-lock.ts

@@ -1,3 +1,4 @@
+import fsSync from "node:fs";
 import fs from "node:fs/promises";
 import path from "node:path";
 
@@ -13,6 +14,9 @@ type HeldLock = {
 };
 
 const HELD_LOCKS = new Map<string, HeldLock>();
+const CLEANUP_SIGNALS = ["SIGINT", "SIGTERM", "SIGQUIT", "SIGABRT"] as const;
+type CleanupSignal = (typeof CLEANUP_SIGNALS)[number];
+const cleanupHandlers = new Map<CleanupSignal, () => void>();
 
 function isAlive(pid: number): boolean {
   if (!Number.isFinite(pid) || pid <= 0) return false;
@@ -24,6 +28,65 @@ function isAlive(pid: number): boolean {
   }
 }
 
+/**
+ * Synchronously release all held locks.
+ * Used during process exit when async operations aren't reliable.
+ */
+function releaseAllLocksSync(): void {
+  for (const [sessionFile, held] of HELD_LOCKS) {
+    try {
+      if (typeof held.handle.fd === "number") {
+        fsSync.closeSync(held.handle.fd);
+      }
+    } catch {
+      // Ignore errors during cleanup - best effort
+    }
+    try {
+      fsSync.rmSync(held.lockPath, { force: true });
+    } catch {
+      // Ignore errors during cleanup - best effort
+    }
+    HELD_LOCKS.delete(sessionFile);
+  }
+}
+
+let cleanupRegistered = false;
+
+function handleTerminationSignal(signal: CleanupSignal): void {
+  releaseAllLocksSync();
+  const shouldReraise = process.listenerCount(signal) === 1;
+  if (shouldReraise) {
+    const handler = cleanupHandlers.get(signal);
+    if (handler) process.off(signal, handler);
+    try {
+      process.kill(process.pid, signal);
+    } catch {
+      // Ignore errors during shutdown
+    }
+  }
+}
+
+function registerCleanupHandlers(): void {
+  if (cleanupRegistered) return;
+  cleanupRegistered = true;
+
+  // Cleanup on normal exit and process.exit() calls
+  process.on("exit", () => {
+    releaseAllLocksSync();
+  });
+
+  // Handle termination signals
+  for (const signal of CLEANUP_SIGNALS) {
+    try {
+      const handler = () => handleTerminationSignal(signal);
+      cleanupHandlers.set(signal, handler);
+      process.on(signal, handler);
+    } catch {
+      // Ignore unsupported signals on this platform.
+    }
+  }
+}
+
 async function readLockPayload(lockPath: string): Promise<LockFilePayload | null> {
   try {
     const raw = await fs.readFile(lockPath, "utf8");
@@ -43,6 +106,7 @@ export async function acquireSessionWriteLock(params: {
 }): Promise<{
   release: () => Promise<void>;
 }> {
+  registerCleanupHandlers();
   const timeoutMs = params.timeoutMs ?? 10_000;
   const staleMs = params.staleMs ?? 30 * 60 * 1000;
   const sessionFile = path.resolve(params.sessionFile);
@@ -116,3 +180,9 @@ export async function acquireSessionWriteLock(params: {
   const owner = payload?.pid ? `pid=${payload.pid}` : "unknown";
   throw new Error(`session file locked (timeout ${timeoutMs}ms): ${owner} ${lockPath}`);
 }
+
+export const __testing = {
+  cleanupSignals: [...CLEANUP_SIGNALS],
+  handleTerminationSignal,
+  releaseAllLocksSync,
+};

src/agents/tools/telegram-actions.ts

@@ -3,6 +3,7 @@ import type { ClawdbotConfig } from "../../config/config.js";
 import { resolveTelegramReactionLevel } from "../../telegram/reaction-level.js";
 import {
   deleteMessageTelegram,
+  editMessageTelegram,
   reactMessageTelegram,
   sendMessageTelegram,
 } from "../../telegram/send.js";
@@ -209,5 +210,50 @@ export async function handleTelegramAction(
     return jsonResult({ ok: true, deleted: true });
   }
 
+  if (action === "editMessage") {
+    if (!isActionEnabled("editMessage")) {
+      throw new Error("Telegram editMessage is disabled.");
+    }
+    const chatId = readStringOrNumberParam(params, "chatId", {
+      required: true,
+    });
+    const messageId = readNumberParam(params, "messageId", {
+      required: true,
+      integer: true,
+    });
+    const content = readStringParam(params, "content", {
+      required: true,
+      allowEmpty: false,
+    });
+    const buttons = readTelegramButtons(params);
+    if (buttons) {
+      const inlineButtonsScope = resolveTelegramInlineButtonsScope({
+        cfg,
+        accountId: accountId ?? undefined,
+      });
+      if (inlineButtonsScope === "off") {
+        throw new Error(
+          'Telegram inline buttons are disabled. Set channels.telegram.capabilities.inlineButtons to "dm", "group", "all", or "allowlist".',
+        );
+      }
+    }
+    const token = resolveTelegramToken(cfg, { accountId }).token;
+    if (!token) {
+      throw new Error(
+        "Telegram bot token missing. Set TELEGRAM_BOT_TOKEN or channels.telegram.botToken.",
+      );
+    }
+    const result = await editMessageTelegram(chatId ?? "", messageId ?? 0, content, {
+      token,
+      accountId: accountId ?? undefined,
+      buttons,
+    });
+    return jsonResult({
+      ok: true,
+      messageId: result.messageId,
+      chatId: result.chatId,
+    });
+  }
+
   throw new Error(`Unsupported Telegram action: ${action}`);
 }