From ee895987cb80740f0387e72fec9026f6da0d39c8 Mon Sep 17 00:00:00 2001 From: Ben U Date: Sun, 19 Jul 2026 08:55:34 -0600 Subject: [PATCH] feat: renew workers from manager activity --- CHANGELOG.md | 1 + README.md | 2 +- dist/plugin.mjs | 42 ++++++++++++++++++++++++++++++++++++++-- opencode/plugin.ts | 38 ++++++++++++++++++++++++++++++++++-- opencode/runtime.test.ts | 11 +++++++++++ opencode/runtime.ts | 5 +++++ 6 files changed, 94 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 534cd50..31be0da 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/README.md b/README.md index b5ea3f3..a57e569 100644 --- a/README.md +++ b/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 diff --git a/dist/plugin.mjs b/dist/plugin.mjs index e2d9886..187f78b 100644 --- a/dist/plugin.mjs +++ b/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) { diff --git a/opencode/plugin.ts b/opencode/plugin.ts index 7e3370c..7e628b0 100644 --- a/opencode/plugin.ts +++ b/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): 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) { diff --git a/opencode/runtime.test.ts b/opencode/runtime.test.ts index bd8ea8a..554111e 100644 --- a/opencode/runtime.test.ts +++ b/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((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 { diff --git a/opencode/runtime.ts b/opencode/runtime.ts index ae3b9b1..b11823b 100644 --- a/opencode/runtime.ts +++ b/opencode/runtime.ts @@ -181,6 +181,7 @@ export interface OpenCodeIntercomRuntimeOptions { clientFactory?: () => IntercomClient; prepareConnection?: () => Promise; reconnectDelays?: number[]; + onInboundActivity?: (from: SessionInfo, message: Message) => void | Promise; } export class OpenCodeIntercomRuntime { @@ -199,6 +200,7 @@ export class OpenCodeIntercomRuntime { private readonly clientFactory: () => IntercomClient; private readonly prepareConnection: () => Promise; private readonly reconnectDelays: number[]; + private readonly onInboundActivity?: (from: SessionInfo, message: Message) => void | Promise; 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);