fix: pi-worker systemd integration + lock file bug

- Fix FileLock: track 'held' state, only release if acquired, suppress ENOENT
- Upgrade systemd service: Restart=always (was on-failure, caused 2-day outage),
  StartLimitBurst, WatchdogSec=120, MemoryMax=512M, KillMode=mixed
- Add systemd watchdog: health server self-checks at :8787, notifies sd_notify
- Kill orphaned nohup process, restart cleanly via systemd

pi-worker/src/health/server.ts

@@ -1,14 +1,22 @@
 import { logger } from "../logger.js";
 
+/** Shape of the JSON body returned by `GET /health`. */
 interface HealthStatus {
+  /** Overall status of the process. Set to `"degraded"` on graceful shutdown. */
   status: "ok" | "degraded" | "error";
+  /** Seconds elapsed since the process started (computed at request time). */
   uptime: number;
+  /** ISO-8601 timestamp of when the process started. */
   startedAt: string;
+  /** `true` once the {@link TaskLoop} has been started successfully. */
   taskLoop: boolean;
+  /** `true` once the {@link ImprovementLoop} has been started successfully. */
   improvementLoop: boolean;
+  /** ISO-8601 timestamp of the most recent Asana poll, if any. */
   lastPoll?: string;
 }
 
+/** In-memory snapshot of the current health state. */
 let healthStatus: HealthStatus = {
   status: "ok",
   uptime: 0,
@@ -17,10 +25,99 @@ let healthStatus: HealthStatus = {
   improvementLoop: false,
 };
 
+/**
+ * Merges `updates` into the in-memory health snapshot.
+ * Call this whenever a subsystem changes state (e.g. after starting a loop
+ * or during graceful shutdown).
+ *
+ * @param updates - Partial fields to overwrite in the current health status.
+ */
 export function updateHealth(updates: Partial<HealthStatus>) {
   healthStatus = { ...healthStatus, ...updates };
 }
 
+/**
+ * Security headers applied to every response from the health server.
+ */
+const SECURITY_HEADERS: Record<string, string> = {
+  "X-Content-Type-Options": "nosniff",
+  "X-Frame-Options": "DENY",
+  "Cache-Control": "no-store",
+};
+
+/**
+ * Starts a systemd watchdog heartbeat if WatchdogSec is configured.
+ *
+ * sd_notify is done by writing "WATCHDOG=1" to the NOTIFY_SOCKET.
+ * Bun doesn't have native sd_notify, so we use a simple UDP/unix datagram.
+ * Fallback: we just curl the health endpoint and if it responds, we're alive.
+ */
+let watchdogTimer: ReturnType<typeof setInterval> | null = null;
+
+function startWatchdog(healthPort: number) {
+  const watchdogUsec = process.env.WATCHDOG_USEC;
+  if (!watchdogUsec) {
+    logger.debug("health", "No WATCHDOG_USEC set, skipping systemd watchdog");
+    return;
+  }
+
+  const intervalMs = Math.floor(parseInt(watchdogUsec) / 1000 / 2); // ping at half the deadline
+  logger.info("health", `Systemd watchdog enabled, pinging every ${intervalMs}ms`);
+
+  watchdogTimer = setInterval(async () => {
+    try {
+      // Self-check: hit our own health endpoint
+      const res = await fetch(`http://127.0.0.1:${healthPort}/health`);
+      if (!res.ok) {
+        logger.error("health", `Watchdog self-check failed: ${res.status}`);
+        return; // Don't notify systemd — let it kill us
+      }
+
+      const data = (await res.json()) as HealthStatus;
+      if (data.status !== "ok") {
+        logger.warn("health", `Watchdog: status is ${data.status}, not notifying systemd`);
+        return;
+      }
+
+      // Notify systemd via NOTIFY_SOCKET
+      const notifySocket = process.env.NOTIFY_SOCKET;
+      if (notifySocket) {
+        const { createSocket } = await import("dgram");
+        const socketPath = notifySocket.startsWith("@")
+          ? `\0${notifySocket.slice(1)}`
+          : notifySocket;
+        const client = createSocket("unix_dgram" as any);
+        const msg = Buffer.from("WATCHDOG=1");
+        client.send(msg, 0, msg.length, socketPath, () => client.close());
+      }
+    } catch (e) {
+      logger.error("health", `Watchdog ping failed: ${e}`);
+      // Don't notify — let systemd restart us
+    }
+  }, intervalMs);
+}
+
+export function stopWatchdog() {
+  if (watchdogTimer) {
+    clearInterval(watchdogTimer);
+    watchdogTimer = null;
+  }
+}
+
+/**
+ * Starts the HTTP health server on the given `port`.
+ *
+ * ### Endpoints
+ *
+ * | Method | Path      | Response |
+ * |--------|-----------|----------|
+ * | GET    | `/health` | `200 application/json` — {@link HealthStatus} with live `uptime` |
+ * | GET    | `/`       | `200 text/plain` — simple alive message |
+ * | *      | *         | `404 Not Found` |
+ *
+ * @param port - TCP port to listen on.
+ * @returns The `Bun.Server` instance (rarely needed by callers).
+ */
 export function startHealthServer(port: number) {
   const startTime = Date.now();
 
@@ -29,23 +126,38 @@ export function startHealthServer(port: number) {
     fetch(req) {
       const url = new URL(req.url);
 
-      if (url.pathname === "/health") {
-        return Response.json({
-          ...healthStatus,
-          uptime: Math.floor((Date.now() - startTime) / 1000),
+      // Only allow safe read-only methods on this server.
+      if (req.method !== "GET" && req.method !== "HEAD") {
+        return new Response("Method Not Allowed", {
+          status: 405,
+          headers: { ...SECURITY_HEADERS, Allow: "GET, HEAD" },
         });
       }
 
+      if (url.pathname === "/health") {
+        return Response.json(
+          {
+            ...healthStatus,
+            uptime: Math.floor((Date.now() - startTime) / 1000),
+          },
+          { headers: SECURITY_HEADERS }
+        );
+      }
+
       if (url.pathname === "/") {
         return new Response("Pi Worker is running. GET /health for status.", {
-          headers: { "Content-Type": "text/plain" },
+          headers: { ...SECURITY_HEADERS, "Content-Type": "text/plain" },
         });
       }
 
-      return new Response("Not Found", { status: 404 });
+      return new Response("Not Found", { status: 404, headers: SECURITY_HEADERS });
     },
   });
 
   logger.info("health", `Health server listening on :${port}`);
+
+  // Start systemd watchdog after health server is up
+  startWatchdog(port);
+
   return server;
 }

pi-worker/src/scheduler/lock.ts

@@ -1,46 +1,112 @@
 import { writeFileSync, unlinkSync, readFileSync, existsSync } from "fs";
 import { join } from "path";
 
+/**
+ * A simple file-based mutex used to prevent concurrent Pi CLI executions.
+ *
+ * The lock file stores the acquisition timestamp (milliseconds since epoch)
+ * as plain text. Any lock older than {@link STALE_LOCK_MS} is automatically
+ * treated as stale and removed, ensuring a crashed process can never
+ * permanently block future runs.
+ *
+ * Lock files are stored in `dir` (default `/tmp`) as
+ * `pi-worker-<name>.lock`.
+ *
+ * @example
+ * ```ts
+ * const lock = new FileLock("task-loop");
+ * if (!lock.acquire()) return; // another execution in progress
+ * try {
+ *   await doWork();
+ * } finally {
+ *   lock.release();
+ * }
+ * ```
+ */
 export class FileLock {
   private lockPath: string;
+  private held = false;
 
+  /** Locks older than this (ms) are considered stale and are removed on the next acquire(). */
+  private static readonly STALE_LOCK_MS = 15 * 60 * 1000; // 15 minutes
+
+  /**
+   * @param name - Logical name for this lock (e.g. `"task-loop"`).
+   *   Combined with the prefix to form the lock filename.
+   * @param dir  - Directory in which to create the lock file. Defaults to `/tmp`.
+   */
   constructor(name: string, dir: string = "/tmp") {
     this.lockPath = join(dir, `pi-worker-${name}.lock`);
   }
 
+  /**
+   * Attempts to acquire the lock.
+   *
+   * If a lock file already exists and is **not** stale, returns `false`
+   * immediately (non-blocking). Stale lock files are removed and the
+   * acquisition is retried once.
+   *
+   * Uses the `wx` (exclusive create) flag so that two processes racing to
+   * acquire the lock cannot both succeed.
+   *
+   * @returns `true` if the lock was acquired, `false` if it is already held.
+   */
   acquire(): boolean {
     // Check for stale lock first
     if (existsSync(this.lockPath)) {
       try {
         const content = readFileSync(this.lockPath, "utf-8");
         const lockTime = parseInt(content);
-        if (Date.now() - lockTime > 15 * 60 * 1000) {
-          // Stale lock, remove it
-          try { unlinkSync(this.lockPath); } catch {}
+        if (Date.now() - lockTime > FileLock.STALE_LOCK_MS) {
+          // Stale lock — remove it
+          try { unlinkSync(this.lockPath); } catch (e) {
+            process.stderr.write(`[lock] Failed to remove stale lock ${this.lockPath}: ${e}\n`);
+          }
         } else {
           return false;
         }
-      } catch {
-        // Corrupted lock file, remove it
+      } catch (e) {
+        // Corrupted lock file — remove it so we don't get permanently stuck
+        process.stderr.write(`[lock] Corrupted lock file ${this.lockPath}, removing: ${e}\n`);
         try { unlinkSync(this.lockPath); } catch {}
       }
     }
 
-    // Atomic create
+    // Atomic create — fails with EEXIST if another process won the race
     try {
       writeFileSync(this.lockPath, Date.now().toString(), { flag: 'wx' });
+      this.held = true;
       return true;
     } catch {
-      return false; // Another process got it first
+      return false;
     }
   }
 
+  /**
+   * Releases the lock by deleting the lock file.
+   *
+   * Only deletes if this instance actually holds the lock. Safe to call
+   * even if the lock is not currently held (no-op in that case).
+   */
   release() {
+    if (!this.held) return;
     try {
       unlinkSync(this.lockPath);
-    } catch {}
+    } catch (e) {
+      // Suppress ENOENT — lock was already cleaned up (e.g. by stale detection)
+      if ((e as NodeJS.ErrnoException).code !== "ENOENT") {
+        process.stderr.write(`[lock] Failed to release lock ${this.lockPath}: ${e}\n`);
+      }
+    } finally {
+      this.held = false;
+    }
   }
 
+  /**
+   * Returns `true` if a lock file exists on disk (regardless of whether it
+   * is stale). Useful for diagnostics; prefer {@link acquire} for actual
+   * mutual-exclusion logic.
+   */
   isLocked(): boolean {
     return existsSync(this.lockPath);
   }