fix(agents): keep facets out of root-owned host job sync (#2299)

.changeset/facet-host-jobs.md

@@ -0,0 +1,5 @@
+---
+"agents": patch
+---
+
+Sub-agents no longer sync root-owned host jobs. A facet has no alarm slot, so `_syncHostJobs()` on a facet only opened a `schedule_agent_alarm` span on every wake, and when a facet's startup fiber recovery left a row behind (for example a throwing `onFiberRecovered`), it pushed a housekeeping job whose `setAlarm()` threw and failed the facet's first call after every restart. The root's facet-run lease already drives that retry. A host job an earlier release left in a facet's queue is dropped on the next wake.

packages/agents/src/index.ts

@@ -4767,6 +4767,21 @@ export class Agent<
    */
   private async _syncHostJobs(): Promise<void> {
     if (this._destroyed) return;
+    if (this._isFacet) {
+      // A facet has no alarm slot (`setAlarm()` throws): the root holds its
+      // keepAlive refs and facet-run leases, and facets never write the
+      // destroy marker. Earlier releases could persist a host job here
+      // before the re-arm threw; drop it so it cannot re-arm again.
+      const work = this.lifecycle.jobs;
+      for (const id of [
+        HOST_JOB_DESTROY_ID,
+        HOST_JOB_KEEP_ALIVE_ID,
+        HOST_JOB_HOUSEKEEPING_ID
+      ]) {
+        if (work.get(id)) await work.cancel(id);
+      }
+      return;
+    }
     await this._withAgentSpan("schedule_agent_alarm", "alarm", {}, async () => {
       const work = this.lifecycle.jobs;
       const nowMs = Date.now();

packages/agents/src/tests/agents/sub-agent.ts

@@ -68,6 +68,9 @@ export class CounterSubAgent extends Agent {
       VALUES
         (${ctx.id}, ${ctx.name}, ${JSON.stringify(ctx.snapshot)}, ${ctx.createdAt})
     `;
+    if (ctx.name === "recovery-throws") {
+      throw new Error("recovery hook failed");
+    }
     if (ctx.name === "managed-recovery-complete") {
       return { status: "completed", snapshot: { recovered: true } };
     }
@@ -518,6 +521,18 @@ export class CounterSubAgent extends Agent {
     return rows[0]?.count ?? 0;
   }
 
+  getLocalJobIds(): string[] {
+    return this.lifecycle.jobs.list().map((job) => job.id);
+  }
+
+  /** Persist a host job row the way a failed pre-#2299 facet push left it. */
+  insertStaleHostJob(id: string, fn: string): void {
+    this.sql`
+      INSERT INTO cf_agents_jobs (id, capability, fn, time)
+      VALUES (${id}, 'host', ${fn}, ${Date.now()})
+    `;
+  }
+
   async inspectManagedFiber(fiberId: string): Promise<FiberInspection | null> {
     return this.inspectFiber(fiberId);
   }
@@ -2014,6 +2029,23 @@ export class TestSubAgentParent extends Agent {
     return child.tryCancelSchedule();
   }
 
+  async subAgentInsertStaleHostJob(
+    subAgentName: string,
+    id: string,
+    fn: string
+  ): Promise<void> {
+    const child = await this.subAgent(CounterSubAgent, subAgentName);
+    await child.insertStaleHostJob(id, fn);
+  }
+
+  async subAgentLocalJobIdsAfterRestart(
+    subAgentName: string
+  ): Promise<string[]> {
+    this.abortSubAgent(CounterSubAgent, subAgentName);
+    const child = await this.subAgent(CounterSubAgent, subAgentName);
+    return child.getLocalJobIds();
+  }
+
   async subAgentTryScheduleAfterAbort(subAgentName: string): Promise<string> {
     // Create the sub-agent and let it be marked as a facet
     await this.subAgent(CounterSubAgent, subAgentName);

packages/agents/src/tests/sub-agent.test.ts

@@ -1161,6 +1161,50 @@ describe("SubAgent", () => {
     ]);
   });
 
+  it("leaves host jobs to the root while a sub-agent's fiber recovery is pending (#2299)", async () => {
+    const name = uniqueName();
+    const agent = await getAgentByName(env.TestSubAgentParent, name);
+
+    await agent.insertSubAgentInterruptedFiber(
+      "orphan-wake-child",
+      "fiber-orphan-wake-1",
+      "recovery-throws"
+    );
+
+    // Startup recovery throws, so the orphaned row survives the wake. The
+    // root's facet-run lease owns the retry; the facet queues nothing.
+    await expect(
+      agent.subAgentLocalJobIdsAfterRestart("orphan-wake-child")
+    ).resolves.toEqual([]);
+    expect(await agent.subAgentRunningFiberCount("orphan-wake-child")).toBe(1);
+    expect((await agent.facetRunRows()).map((row) => row.runId)).toContain(
+      "fiber-orphan-wake-1"
+    );
+  });
+
+  it("drops host jobs an earlier release left in a sub-agent's queue (#2299)", async () => {
+    const name = uniqueName();
+    const agent = await getAgentByName(env.TestSubAgentParent, name);
+
+    await agent.subAgentInsertStaleHostJob(
+      "stale-jobs-child",
+      "cf:housekeeping",
+      "housekeeping"
+    );
+    await agent.subAgentInsertStaleHostJob(
+      "stale-jobs-child",
+      "cf:keep-alive",
+      "keepAlive"
+    );
+
+    await expect(
+      agent.subAgentLocalJobIdsAfterRestart("stale-jobs-child")
+    ).resolves.toEqual([]);
+    await expect(
+      agent.subAgentIncrement("stale-jobs-child", "c")
+    ).resolves.toBe(1);
+  });
+
   it("applies managed sub-agent fiber recovery outcomes from the child", async () => {
     const name = uniqueName();
     const agent = await getAgentByName(env.TestSubAgentParent, name);

packages/think/src/tests/agents/messengers.ts

@@ -157,8 +157,8 @@ export class ThinkMessengerDeliveryTestAgent extends Think {
     const recordEnd = (text: string) => this._record("stream-end", text);
     const mode = this._recoveryMode();
     const nextCall = () => ++this._streamCalls;
-    // An agent named `slow-…` takes 2.5s per model call.
-    const slowMs = this.name.startsWith("slow-") ? 2500 : 0;
+    // An agent named `slow-…` takes 4s per model call.
+    const slowMs = this.name.startsWith("slow-") ? 4000 : 0;
     return {
       specificationVersion: "v3",
       provider: "test",
@@ -592,14 +592,15 @@ export class ThinkMessengerDeliveryTestAgent extends Think {
         this._chat = chat;
         if (this.name.startsWith("slow-")) {
           // Stand-in for the Chat SDK's fixed 30s lock: short enough that a
-          // slow turn outlives it unless the lock is kept alive.
+          // slow turn outlives it unless the lock is kept alive, long enough
+          // that a loaded runner's stalls do not starve the heartbeat.
           const state = chat.getState();
           const acquireLock = state.acquireLock.bind(state);
           const extendLock = state.extendLock.bind(state);
           state.acquireLock = (threadId, ttlMs) =>
-            acquireLock(threadId, Math.min(ttlMs, 300));
+            acquireLock(threadId, Math.min(ttlMs, 1000));
           state.extendLock = (lock, ttlMs) =>
-            extendLock(lock, Math.min(ttlMs, 300));
+            extendLock(lock, Math.min(ttlMs, 1000));
         }
         return Promise.resolve();
       },

packages/think/src/tests/messengers.test.ts

@@ -1611,15 +1611,15 @@ describe("think messengers core", () => {
           .then((res) => res.text());
       const agent = await getAgentByName(
         env.ThinkMessengerDeliveryTestAgent,
-        "slow-dm"
+        `slow-dm-${crypto.randomUUID()}`
       );
       const first = send("l1", "first");
       for (let i = 0; i < 100; i++) {
         if ((await agent.getModelLog()).length > 0) break;
         await new Promise((resolve) => setTimeout(resolve, 50));
       }
-      // Past the 300ms lock TTL, well inside the first turn.
-      await new Promise((resolve) => setTimeout(resolve, 600));
+      // Past the 1s lock TTL, well inside the 4s first turn.
+      await new Promise((resolve) => setTimeout(resolve, 1500));
       await send("l2", "second");
       // Queued behind the running turn, not handled under a lock of its own.
       expect(await agent.queueDepthForTest(threadId)).toBe(1);