feat(agents): synchronous Lifecycle job push and cancel

.changeset/lifecycle-jobs-sync.md

@@ -0,0 +1,5 @@
+---
+"agents": patch
+---
+
+Lifecycle capabilities can push and cancel jobs synchronously with `jobs.pushSync()` and `jobs.cancelSync()`, so a job commits or rolls back with the capability's own writes in one `storage.transactionSync`. Call `jobs.rearm()` once the transaction commits.

docs/agents/lifecycle.md

@@ -191,7 +191,21 @@ class Cleanup extends LifecycleCapability {
 ```
 
 The queue is ordered by timestamp, and every queue mutation re-arms the
-physical alarm automatically — there is no explicit rearm call. When the
+physical alarm automatically. The one exception is a job that must commit with
+the capability's own rows. `jobs.pushSync()` and `jobs.cancelSync()` write
+synchronously inside a `storage.transactionSync`, so the job and the rows
+commit or roll back together. They skip the re-arm, so call `jobs.rearm()`
+once the transaction commits:
+
+```ts
+this.lifecycle.storage.transactionSync(() => {
+  this.#insertRow(id, input);
+  this.lifecycle.jobs.pushSync({ id, fn: "step", time: Date.now() });
+});
+await this.lifecycle.jobs.rearm();
+```
+
+When the
 alarm fires, Lifecycle drives due jobs in due order as an event loop, then
 runs host `onAlarm()`, then re-arms from queue state. Before driving any job
 it arms a deadman pre-alarm so an isolate death mid-drive still wakes the

packages/agents/src/lifecycle/durable-object-lifecycle.ts

@@ -779,8 +779,11 @@ export class Lifecycle<
     return Object.freeze({
       push: (options: LifecycleJobPushOptions) =>
         rearmAfter(() => this.#jobQueue.push(owner, options)),
+      pushSync: (options: LifecycleJobPushOptions) =>
+        this.#jobQueue.push(owner, options),
       cancel: (id: string) =>
         rearmAfter(() => this.#jobQueue.cancel(owner, id)),
+      cancelSync: (id: string) => this.#jobQueue.cancel(owner, id),
       reschedule: (id: string, time: number) =>
         rearmAfter(() => this.#jobQueue.reschedule(owner, id, time)),
       get: (id: string) => this.#jobQueue.get(owner, id),

packages/agents/src/lifecycle/job-queue.ts

@@ -112,8 +112,21 @@ export type LifecycleJobs = {
    * owner's job is impossible; a cross-owner id collision throws instead.
    */
   readonly push: (options: LifecycleJobPushOptions) => Promise<LifecycleJob>;
+  /**
+   * {@link LifecycleJobs.push}, synchronously and without re-arming the
+   * alarm, so it can share a `storage.transactionSync` with the caller's own
+   * writes: the job and those writes commit or roll back together. Call
+   * {@link LifecycleJobs.rearm} after the transaction commits, or the alarm
+   * may not fire for the new job until another mutation re-arms it.
+   */
+  readonly pushSync: (options: LifecycleJobPushOptions) => LifecycleJob;
   /** Cancel one owned job. Returns false when no job matched. */
   readonly cancel: (id: string) => Promise<boolean>;
+  /**
+   * {@link LifecycleJobs.cancel}, synchronously and without re-arming, for
+   * use inside a `storage.transactionSync`. Call `rearm()` after it commits.
+   */
+  readonly cancelSync: (id: string) => boolean;
   /** Re-time one owned job. Returns false when no job matched. */
   readonly reschedule: (id: string, time: number) => Promise<boolean>;
   /** Read one owned job. */
@@ -195,6 +208,17 @@ export class JobQueue {
     try {
       return [...this.#storage.sql.exec(query, ...params)] as T[];
     } catch (cause) {
+      // A transactionSync that rolled back after the first synchronous push
+      // also rolled back the CREATE TABLE, so the cached flag is stale.
+      if (String(cause).includes("no such table: cf_agents_jobs")) {
+        this.#tableEnsured = false;
+        this.#ensureTable();
+        try {
+          return [...this.#storage.sql.exec(query, ...params)] as T[];
+        } catch (retryCause) {
+          throw new SqlError(query, retryCause);
+        }
+      }
       throw new SqlError(query, cause);
     }
   }

packages/agents/src/tests/lifecycle/job-sync.test.ts

@@ -0,0 +1,91 @@
+import { describe, expect, it } from "vitest";
+import {
+  LifecycleCapability,
+  type LifecycleJobPushOptions
+} from "../../lifecycle";
+import { withCapabilityHarness } from "../shared/capability-harness";
+
+class SyncJobCapability extends LifecycleCapability {
+  constructor() {
+    super("sync-job-test");
+  }
+
+  push(options: LifecycleJobPushOptions) {
+    return this.lifecycle.jobs.pushSync(options);
+  }
+
+  cancel(id: string) {
+    return this.lifecycle.jobs.cancelSync(id);
+  }
+
+  list() {
+    return this.lifecycle.jobs.list();
+  }
+
+  rearm() {
+    return this.lifecycle.jobs.rearm();
+  }
+}
+
+describe("Lifecycle synchronous jobs", () => {
+  it("rolls a synchronous job back with its surrounding transaction", async () => {
+    await withCapabilityHarness(async ({ storage, install }) => {
+      const capability = new SyncJobCapability();
+      const { lifecycle } = install(capability);
+      await lifecycle.start();
+
+      expect(() =>
+        storage.transactionSync(() => {
+          capability.push({ id: "job", fn: "run", time: Date.now() });
+          throw new Error("rollback");
+        })
+      ).toThrow("rollback");
+      expect(capability.list()).toEqual([]);
+    });
+  });
+
+  it("rearms after a committed synchronous mutation", async () => {
+    await withCapabilityHarness(async ({ storage, install }) => {
+      const capability = new SyncJobCapability();
+      const { lifecycle } = install(capability);
+      await lifecycle.start();
+
+      const time = Date.now() + 60_000;
+      storage.transactionSync(() => {
+        capability.push({ id: "job", fn: "run", time });
+      });
+      expect(await storage.getAlarm()).toBeNull();
+
+      await capability.rearm();
+      expect(await storage.getAlarm()).toBe(time);
+
+      storage.transactionSync(() => {
+        expect(capability.cancel("job")).toBe(true);
+      });
+      await capability.rearm();
+      expect(await storage.getAlarm()).toBeNull();
+    });
+  });
+
+  it("recreates the job table when the transaction that created it rolled back", async () => {
+    await withCapabilityHarness(async ({ storage, install }) => {
+      const capability = new SyncJobCapability();
+      const { lifecycle } = install(capability);
+      await lifecycle.start();
+      storage.sql.exec("DROP TABLE IF EXISTS cf_agents_jobs");
+
+      expect(() =>
+        storage.transactionSync(() => {
+          capability.push({ id: "first", fn: "run", time: Date.now() });
+          throw new Error("rollback");
+        })
+      ).toThrow("rollback");
+
+      const time = Date.now() + 60_000;
+      storage.transactionSync(() => {
+        capability.push({ id: "second", fn: "run", time });
+      });
+      expect(capability.list().map((job) => job.id)).toEqual(["second"]);
+    });
+  });
+});