fix: make book locking atomic and scheduler non-reentrant

packages/core/src/__tests__/scheduler.test.ts

@@ -0,0 +1,68 @@
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { Scheduler, type SchedulerConfig } from "../pipeline/scheduler.js";
+
+function createConfig(): SchedulerConfig {
+  return {
+    client: {
+      provider: "openai",
+      apiFormat: "chat",
+      stream: false,
+      defaults: {
+        temperature: 0.7,
+        maxTokens: 1024,
+        thinkingBudget: 0,
+      },
+    } as SchedulerConfig["client"],
+    model: "test-model",
+    projectRoot: process.cwd(),
+    radarCron: "*/1 * * * *",
+    writeCron: "*/1 * * * *",
+    maxConcurrentBooks: 1,
+    chaptersPerCycle: 1,
+    retryDelayMs: 0,
+    cooldownAfterChapterMs: 0,
+    maxChaptersPerDay: 10,
+  };
+}
+
+describe("Scheduler", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+  });
+
+  afterEach(() => {
+    vi.restoreAllMocks();
+    vi.useRealTimers();
+  });
+
+  it("does not start a second write cycle while one is still running", async () => {
+    const scheduler = new Scheduler(createConfig());
+    let releaseCycle: (() => void) | undefined;
+    const blockedCycle = new Promise<void>((resolve) => {
+      releaseCycle = resolve;
+    });
+
+    const runWriteCycle = vi
+      .spyOn(scheduler as unknown as { runWriteCycle: () => Promise<void> }, "runWriteCycle")
+      .mockImplementation(async () => {
+        if (runWriteCycle.mock.calls.length === 1) {
+          return;
+        }
+        await blockedCycle;
+      });
+    vi.spyOn(scheduler as unknown as { runRadarScan: () => Promise<void> }, "runRadarScan")
+      .mockResolvedValue(undefined);
+
+    await scheduler.start();
+
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(runWriteCycle).toHaveBeenCalledTimes(2);
+
+    await vi.advanceTimersByTimeAsync(60_000);
+    expect(runWriteCycle).toHaveBeenCalledTimes(2);
+
+    releaseCycle?.();
+    await blockedCycle;
+    scheduler.stop();
+  });
+});

packages/core/src/__tests__/state-manager.test.ts

@@ -346,6 +346,26 @@ describe("StateManager", () => {
       expect(typeof release2).toBe("function");
       await release2();
     });
+
+    it("allows only one concurrent lock claimant", async () => {
+      await mkdir(manager.bookDir("lock-book-4"), { recursive: true });
+
+      const results = await Promise.allSettled([
+        manager.acquireBookLock("lock-book-4"),
+        manager.acquireBookLock("lock-book-4"),
+      ]);
+
+      const fulfilled = results.filter((result) => result.status === "fulfilled");
+      const rejected = results.filter((result) => result.status === "rejected");
+
+      for (const result of fulfilled) {
+        await result.value();
+      }
+
+      expect(fulfilled).toHaveLength(1);
+      expect(rejected).toHaveLength(1);
+      expect(String(rejected[0]?.reason)).toMatch(/is locked/);
+    });
   });
 
   // -------------------------------------------------------------------------

packages/core/src/pipeline/scheduler.ts

@@ -34,6 +34,8 @@ export class Scheduler {
   private readonly config: SchedulerConfig;
   private tasks: ScheduledTask[] = [];
   private running = false;
+  private writeCycleInFlight: Promise<void> | null = null;
+  private radarScanInFlight: Promise<void> | null = null;
 
   // Quality gate tracking (per book)
   private consecutiveFailures = new Map<string, number>();
@@ -57,7 +59,7 @@ export class Scheduler {
     this.running = true;
 
     // Run write cycle immediately on start, then schedule
-    await this.runWriteCycle();
+    await this.triggerWriteCycle();
 
     // Schedule recurring write cycle
     const writeCycleMs = this.cronToMs(this.config.writeCron);
@@ -66,7 +68,7 @@ export class Scheduler {
       intervalMs: writeCycleMs,
     };
     writeTask.timer = setInterval(() => {
-      this.runWriteCycle().catch((e) => {
+      this.triggerWriteCycle().catch((e) => {
         this.config.onError?.("scheduler", e as Error);
       });
     }, writeCycleMs);
@@ -79,7 +81,7 @@ export class Scheduler {
       intervalMs: radarMs,
     };
     radarTask.timer = setInterval(() => {
-      this.runRadarScan().catch((e) => {
+      this.triggerRadarScan().catch((e) => {
         this.config.onError?.("radar", e as Error);
       });
     }, radarMs);
@@ -98,6 +100,36 @@ export class Scheduler {
     return this.running;
   }
 
+  private async triggerWriteCycle(): Promise<void> {
+    if (this.writeCycleInFlight) {
+      this.log?.warn("Write cycle still running, skipping overlapping tick");
+      return;
+    }
+
+    const cycle = this.runWriteCycle().finally(() => {
+      if (this.writeCycleInFlight === cycle) {
+        this.writeCycleInFlight = null;
+      }
+    });
+    this.writeCycleInFlight = cycle;
+    await cycle;
+  }
+
+  private async triggerRadarScan(): Promise<void> {
+    if (this.radarScanInFlight) {
+      this.log?.warn("Radar scan still running, skipping overlapping tick");
+      return;
+    }
+
+    const scan = this.runRadarScan().finally(() => {
+      if (this.radarScanInFlight === scan) {
+        this.radarScanInFlight = null;
+      }
+    });
+    this.radarScanInFlight = scan;
+    await scan;
+  }
+
   /** Resume a paused book. */
   resumeBook(bookId: string): void {
     this.pausedBooks.delete(bookId);

packages/core/src/state/manager.ts

@@ -1,4 +1,4 @@
-import { readFile, writeFile, mkdir, readdir, stat, unlink } from "node:fs/promises";
+import { readFile, writeFile, mkdir, readdir, stat, unlink, open } from "node:fs/promises";
 import { join } from "node:path";
 import type { BookConfig } from "../models/book.js";
 import type { ChapterMeta } from "../models/chapter.js";
@@ -7,18 +7,29 @@ export class StateManager {
   constructor(private readonly projectRoot: string) {}
 
   async acquireBookLock(bookId: string): Promise<() => Promise<void>> {
+    await mkdir(this.bookDir(bookId), { recursive: true });
     const lockPath = join(this.bookDir(bookId), ".write.lock");
     try {
-      await stat(lockPath);
-      const lockData = await readFile(lockPath, "utf-8");
-      throw new Error(
-        `Book "${bookId}" is locked by another process (${lockData}). ` +
-          `If this is stale, delete ${lockPath}`,
-      );
+      const handle = await open(lockPath, "wx");
+      try {
+        await handle.writeFile(`pid:${process.pid} ts:${Date.now()}`, "utf-8");
+      } catch (error) {
+        await handle.close().catch(() => undefined);
+        await unlink(lockPath).catch(() => undefined);
+        throw error;
+      }
+      await handle.close();
     } catch (e) {
-      if (e instanceof Error && e.message.includes("is locked")) throw e;
+      const code = (e as NodeJS.ErrnoException | undefined)?.code;
+      if (code === "EEXIST") {
+        const lockData = await readFile(lockPath, "utf-8").catch(() => "pid:unknown ts:unknown");
+        throw new Error(
+          `Book "${bookId}" is locked by another process (${lockData}). ` +
+            `If this is stale, delete ${lockPath}`,
+        );
+      }
+      throw e;
     }
-    await writeFile(lockPath, `pid:${process.pid} ts:${Date.now()}`, "utf-8");
     return async () => {
       try {
         await unlink(lockPath);