feat: renew workers from manager activity

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## Unreleased
 
+- Renew an owned worker's activity-bounded lifecycle only when the primary OpenCode manager receives a message from that exact worker, and expose manager acknowledgment for terminal `forget` operations.
 - Add ID-free `oldest`/`latest` selection for multiple pending asks from one sender, hide protocol IDs from pending output, and refuse a second unresolved ask to the same recipient.
 - Automatically reconnect the runtime with its stable Intercom identity after broker restarts and report reconnecting health state.
 - Clarify that assignments and progress/status checkpoints use `intercom_send`, reserving `intercom_ask` for blocking decisions.

README.md

@@ -172,7 +172,7 @@ uses `wl-copy`, `xclip`, or `xsel`; macOS uses `pbcopy`, and Windows uses
 Pending output never exposes protocol message IDs. Keep at most one unresolved `intercom_ask` to the same recipient; the broker rejects a second ask and recommends `intercom_send` for a non-blocking follow-up. Use `intercom_send`—not `intercom_ask`—for assignments and progress/status checkpoints.
 
 The OpenCode runtime automatically reconnects its stable Intercom identity after a broker restart and reports the temporary reconnecting state through peer health metadata.
-- `agent_fleet` *(opt-in manager only)*: create, inspect, adopt, renew, stop, and clean up owned coworkers; inspect coordinated adapter versions and preview or execute source-aware updates using the same implementation as Pi
+- `agent_fleet` *(opt-in manager only)*: create, inspect, adopt, renew, stop, and clean up owned coworkers; inspect coordinated adapter versions and preview or execute source-aware updates using the same implementation as Pi. Manager-received messages from an owned worker automatically renew that exact worker's activity-bounded lease. Deleting a stopped record with `forget` requires `acknowledge: true`.
 
 ## Inbound Delivery Model
 

dist/plugin.mjs

@@ -1796,6 +1796,7 @@ var OpenCodeIntercomRuntime = class {
   clientFactory;
   prepareConnection;
   reconnectDelays;
+  onInboundActivity;
   constructor(identity, cwd, onInboundMessage, inboundStore, options = {}) {
     this.identity = identity ?? buildOpenCodeRuntimeIdentity(process.env, cwd);
     this.onInboundMessage = onInboundMessage;
@@ -1806,6 +1807,7 @@ var OpenCodeIntercomRuntime = class {
       await spawnBrokerIfNeeded(config.brokerCommand, config.brokerArgs);
     });
     this.reconnectDelays = options.reconnectDelays?.length ? options.reconnectDelays : [250, 500, 1e3, 2e3, 5e3];
+    this.onInboundActivity = options.onInboundActivity;
     this.inboundStore = inboundStore ?? new DurableInboundStore(
       process.env.OPENCODE_INTERCOM_INBOUND_STATE?.trim() || getOpenCodeInboundStatePath(this.identity.sessionId)
     );
@@ -1908,6 +1910,7 @@ var OpenCodeIntercomRuntime = class {
       const senderTarget = from.name || from.id;
       const fromMatches = senderTarget.toLowerCase() === waiter.from.toLowerCase() || from.id === waiter.from;
       if (fromMatches) {
+        void Promise.resolve(this.onInboundActivity?.(from, message)).catch(() => void 0);
         this.replyWaiters.delete(waiter.replyTo);
         clearTimeout(waiter.timeout);
         waiter.cleanup?.();
@@ -1922,6 +1925,7 @@ var OpenCodeIntercomRuntime = class {
       this.client?.acknowledgeMessage(deliveryId);
       return;
     }
+    void Promise.resolve(this.onInboundActivity?.(from, message)).catch(() => void 0);
     this.unread.push(entry);
     if (message.expectsReply) {
       this.unresolvedAsks.set(message.id, entry);
@@ -2293,6 +2297,9 @@ var OpenCodeIntercomPlugin = async ({ client, directory, serverUrl }) => {
   let healthReporter;
   const canUseTuiInjection = Boolean(process.stdin.isTTY || process.stdout.isTTY);
   const debugInject = process.env.OPENCODE_INTERCOM_DEBUG === "1";
+  const fleetManagementEnabled = isFleetManagementEnabled();
+  let fleetHeartbeatRunning = false;
+  let fleetHeartbeat;
   function logInject(step, details) {
     if (!debugInject) {
       return;
@@ -2577,7 +2584,15 @@ var OpenCodeIntercomPlugin = async ({ client, directory, serverUrl }) => {
       enqueuePendingInject(entry, "prompt_async_throw");
     }
   }
-  runtime = new OpenCodeIntercomRuntime(void 0, directory, injectInbound);
+  runtime = new OpenCodeIntercomRuntime(void 0, directory, injectInbound, void 0, {
+    onInboundActivity(from) {
+      if (!fleetManagementEnabled) return;
+      void invokeAgentFleet({ action: "renew", id: from.id }, {
+        managerSessionId: runtime.getIdentity().sessionId,
+        cwd: directory
+      }, { ...process.env, AGENT_INTERCOM_DISABLE_CLEANUP_TIMER: "1" }).catch(() => void 0);
+    }
+  });
   const runtimeIdentity = runtime.getIdentity();
   healthReporter = new OpenCodePeerHealthReporter({
     path: process.env.AGENT_INTERCOM_OPENCODE_HEALTH_PATH,
@@ -2606,7 +2621,27 @@ var OpenCodeIntercomPlugin = async ({ client, directory, serverUrl }) => {
     }
   })();
   if (activeSessionID) rememberBounded(knownSessionIDs, activeSessionID);
-  const fleetManagementEnabled = isFleetManagementEnabled();
+  if (fleetManagementEnabled) {
+    fleetHeartbeat = setInterval(() => {
+      if (fleetHeartbeatRunning) return;
+      fleetHeartbeatRunning = true;
+      void invokeAgentFleet({ action: "_heartbeat" }, {
+        managerSessionId: runtimeIdentity.sessionId,
+        cwd: directory
+      }).then(async (result) => {
+        const requests = Array.isArray(result?.details?.checkpointRequests) ? result.details.checkpointRequests : [];
+        for (const request of requests) {
+          if (typeof request?.target !== "string" || typeof request?.message !== "string") continue;
+          await runtime.send(request.target, request.message);
+        }
+      }).catch((error) => {
+        logInject("fleet.heartbeat.error", { error: formatError(error) });
+      }).finally(() => {
+        fleetHeartbeatRunning = false;
+      });
+    }, 6e4);
+    fleetHeartbeat.unref?.();
+  }
   const stopControlServer = startOpenCodeControlServer({
     acceptsSession: (sessionID) => knownSessionIDs.has(sessionID),
     async handle(action) {
@@ -2629,6 +2664,8 @@ var OpenCodeIntercomPlugin = async ({ client, directory, serverUrl }) => {
   });
   return {
     dispose: async () => {
+      if (fleetHeartbeat) clearInterval(fleetHeartbeat);
+      fleetHeartbeat = void 0;
       stopControlServer();
       healthReporter.update({ connected: false, ready: false, status: "stopped" });
       await runtime.disconnect();
@@ -2651,6 +2688,7 @@ var OpenCodeIntercomPlugin = async ({ client, directory, serverUrl }) => {
             fresh: tool.schema.boolean().optional().describe("Start a fresh persistent session rather than resume this worker ID."),
             all: tool.schema.boolean().optional().describe("Include workers owned by other manager sessions for list/status diagnostics."),
             execute: tool.schema.boolean().optional().describe("Actually execute cleanup or updates; false previews."),
+            acknowledge: tool.schema.boolean().optional().describe("Manager acknowledgment required before deleting a stopped worker record."),
             lines: tool.schema.number().optional().describe("Journal lines for logs.")
           },
           async execute(args, context) {

opencode/plugin.ts

@@ -37,6 +37,9 @@ export const OpenCodeIntercomPlugin: Plugin = async ({ client, directory, server
   let healthReporter: OpenCodePeerHealthReporter;
   const canUseTuiInjection = Boolean(process.stdin.isTTY || process.stdout.isTTY);
   const debugInject = process.env.OPENCODE_INTERCOM_DEBUG === "1";
+  const fleetManagementEnabled = isFleetManagementEnabled();
+  let fleetHeartbeatRunning = false;
+  let fleetHeartbeat: NodeJS.Timeout | undefined;
 
   function logInject(step: string, details: Record<string, unknown>): void {
     if (!debugInject) {
@@ -357,7 +360,15 @@ export const OpenCodeIntercomPlugin: Plugin = async ({ client, directory, server
     }
   }
 
-  runtime = new OpenCodeIntercomRuntime(undefined, directory, injectInbound);
+  runtime = new OpenCodeIntercomRuntime(undefined, directory, injectInbound, undefined, {
+    onInboundActivity(from) {
+      if (!fleetManagementEnabled) return;
+      void invokeAgentFleet({ action: "renew", id: from.id }, {
+        managerSessionId: runtime.getIdentity().sessionId,
+        cwd: directory,
+      }, { ...process.env, AGENT_INTERCOM_DISABLE_CLEANUP_TIMER: "1" }).catch(() => undefined);
+    },
+  });
   const runtimeIdentity = runtime.getIdentity();
   healthReporter = new OpenCodePeerHealthReporter({
     path: process.env.AGENT_INTERCOM_OPENCODE_HEALTH_PATH,
@@ -386,7 +397,27 @@ export const OpenCodeIntercomPlugin: Plugin = async ({ client, directory, server
     }
   })();
   if (activeSessionID) rememberBounded(knownSessionIDs, activeSessionID);
-  const fleetManagementEnabled = isFleetManagementEnabled();
+  if (fleetManagementEnabled) {
+    fleetHeartbeat = setInterval(() => {
+      if (fleetHeartbeatRunning) return;
+      fleetHeartbeatRunning = true;
+      void invokeAgentFleet({ action: "_heartbeat" }, {
+        managerSessionId: runtimeIdentity.sessionId,
+        cwd: directory,
+      }).then(async (result) => {
+        const requests = Array.isArray(result?.details?.checkpointRequests) ? result.details.checkpointRequests : [];
+        for (const request of requests) {
+          if (typeof request?.target !== "string" || typeof request?.message !== "string") continue;
+          await runtime.send(request.target, request.message);
+        }
+      }).catch((error) => {
+        logInject("fleet.heartbeat.error", { error: formatError(error) });
+      }).finally(() => {
+        fleetHeartbeatRunning = false;
+      });
+    }, 60_000);
+    fleetHeartbeat.unref?.();
+  }
   const stopControlServer = startOpenCodeControlServer({
     acceptsSession: sessionID => knownSessionIDs.has(sessionID),
     async handle(action) {
@@ -410,6 +441,8 @@ export const OpenCodeIntercomPlugin: Plugin = async ({ client, directory, server
 
   return {
     dispose: async () => {
+      if (fleetHeartbeat) clearInterval(fleetHeartbeat);
+      fleetHeartbeat = undefined;
       stopControlServer();
       healthReporter.update({ connected: false, ready: false, status: "stopped" });
       await runtime.disconnect();
@@ -433,6 +466,7 @@ export const OpenCodeIntercomPlugin: Plugin = async ({ client, directory, server
             fresh: tool.schema.boolean().optional().describe("Start a fresh persistent session rather than resume this worker ID."),
             all: tool.schema.boolean().optional().describe("Include workers owned by other manager sessions for list/status diagnostics."),
             execute: tool.schema.boolean().optional().describe("Actually execute cleanup or updates; false previews."),
+            acknowledge: tool.schema.boolean().optional().describe("Manager acknowledgment required before deleting a stopped worker record."),
             lines: tool.schema.number().optional().describe("Journal lines for logs."),
           },
           async execute(args, context) {

opencode/runtime.test.ts

@@ -98,11 +98,13 @@ test("inbound delivery is durably queued and acknowledged before model injection
     let finishInjection!: () => void;
     const injection = new Promise<void>((resolve) => { finishInjection = resolve; });
     const store = new DurableInboundStore(join(dir, "inbound.json"));
+    const activity: string[] = [];
     const runtime = new OpenCodeIntercomRuntime(
       { sessionId: "receiver", name: "receiver", cwd: "/repo", model: "test", startedAt: 1 },
       "/repo",
       async () => injection,
       store,
+      { onInboundActivity: (from) => { activity.push(from.id); } },
     );
     const acknowledgements: string[] = [];
     (runtime as any).client = {
@@ -118,8 +120,17 @@ test("inbound delivery is durably queued and acknowledged before model injection
       "delivery-1",
     );
 
+    await new Promise((resolve) => setImmediate(resolve));
     assert.deepEqual(acknowledgements, ["delivery-1"]);
+    assert.deepEqual(activity, ["sender"]);
     assert.deepEqual(new DurableInboundStore(store.path).pendingInjection().map((entry) => entry.message.id), ["message-1"]);
+    (runtime as any).handleIncomingMessage(
+      { id: "sender", name: "sender", cwd: "/repo", model: "test", pid: 1, startedAt: 1, lastActivity: 1 },
+      { id: "message-1", content: { text: "hello" }, timestamp: 1 },
+      "delivery-retry",
+    );
+    await new Promise((resolve) => setImmediate(resolve));
+    assert.deepEqual(activity, ["sender"], "durable duplicate replay must not renew activity twice");
     finishInjection();
     await injection;
   } finally {

opencode/runtime.ts

@@ -181,6 +181,7 @@ export interface OpenCodeIntercomRuntimeOptions {
   clientFactory?: () => IntercomClient;
   prepareConnection?: () => Promise<void>;
   reconnectDelays?: number[];
+  onInboundActivity?: (from: SessionInfo, message: Message) => void | Promise<void>;
 }
 
 export class OpenCodeIntercomRuntime {
@@ -199,6 +200,7 @@ export class OpenCodeIntercomRuntime {
   private readonly clientFactory: () => IntercomClient;
   private readonly prepareConnection: () => Promise<void>;
   private readonly reconnectDelays: number[];
+  private readonly onInboundActivity?: (from: SessionInfo, message: Message) => void | Promise<void>;
 
   constructor(identity?: OpenCodeRuntimeIdentity, cwd?: string, onInboundMessage?: InboundMessageHandler, inboundStore?: InboundDeliveryStore, options: OpenCodeIntercomRuntimeOptions = {}) {
     this.identity = identity ?? buildOpenCodeRuntimeIdentity(process.env, cwd);
@@ -210,6 +212,7 @@ export class OpenCodeIntercomRuntime {
       await spawnBrokerIfNeeded(config.brokerCommand, config.brokerArgs);
     });
     this.reconnectDelays = options.reconnectDelays?.length ? options.reconnectDelays : [250, 500, 1000, 2000, 5000];
+    this.onInboundActivity = options.onInboundActivity;
     this.inboundStore = inboundStore ?? new DurableInboundStore(
       process.env.OPENCODE_INTERCOM_INBOUND_STATE?.trim() || getOpenCodeInboundStatePath(this.identity.sessionId),
     );
@@ -321,6 +324,7 @@ export class OpenCodeIntercomRuntime {
       const senderTarget = from.name || from.id;
       const fromMatches = senderTarget.toLowerCase() === waiter.from.toLowerCase() || from.id === waiter.from;
       if (fromMatches) {
+        void Promise.resolve(this.onInboundActivity?.(from, message)).catch(() => undefined);
         this.replyWaiters.delete(waiter.replyTo);
         clearTimeout(waiter.timeout);
         waiter.cleanup?.();
@@ -336,6 +340,7 @@ export class OpenCodeIntercomRuntime {
       this.client?.acknowledgeMessage(deliveryId);
       return;
     }
+    void Promise.resolve(this.onInboundActivity?.(from, message)).catch(() => undefined);
     this.unread.push(entry);
     if (message.expectsReply) {
       this.unresolvedAsks.set(message.id, entry);