diff --git a/src/agents/prepared-model-runtime-lease.ts b/src/agents/prepared-model-runtime-lease.ts index 1a4939eda706..d6e899daecfa 100644 --- a/src/agents/prepared-model-runtime-lease.ts +++ b/src/agents/prepared-model-runtime-lease.ts @@ -83,6 +83,16 @@ export async function acquirePreparedModelRuntimeLeaseFromOwners( existing?.needsRefresh && !existing.pending && (existing.provenance === "run" || existing.provenance === "ephemeral"); + const pluginGenerationChanged = + options.pluginGeneration !== undefined && + (existing?.pending ? existing.pendingPluginGeneration : existing?.pluginGeneration) !== + options.pluginGeneration; + if (existing?.pending && pluginGenerationChanged) { + // Do not supersede active discovery. Wait for its owner to settle, then retry against + // the published identity so same-generation callers still coalesce. + await existing.pending.catch(() => undefined); + continue; + } if ( context.getGatewayLifecycleActive() && provenance === "run" && @@ -114,7 +124,17 @@ export async function acquirePreparedModelRuntimeLeaseFromOwners( } } try { - if (existing && !staleDynamicOwner) { + if (existing?.pending && !pluginGenerationChanged) { + // Matching callers lease the immutable generation they joined even if a queued + // mismatched caller publishes the next owner immediately after this one settles. + snapshot = await existing.pending; + if (existing.snapshot !== snapshot || existing.needsRefresh) { + continue; + } + owner = existing; + break; + } + if (existing && !staleDynamicOwner && !pluginGenerationChanged) { snapshot = await context.prepareSnapshot(input); } else { // Fresh keys publish a first generation; stale dynamic owners publish a distinct diff --git a/src/agents/prepared-model-runtime.owner-selection.test.ts b/src/agents/prepared-model-runtime.owner-selection.test.ts index 9268df33aeb2..2d31445fae06 100644 --- a/src/agents/prepared-model-runtime.owner-selection.test.ts +++ b/src/agents/prepared-model-runtime.owner-selection.test.ts @@ -226,6 +226,61 @@ describe("prepared model runtime owner selection", () => { expect(mocks.ensureOpenClawModelsJson).not.toHaveBeenCalled(); }); + it("sequences pending owners by their explicit plugin generation", async () => { + mocks.configuredAgentIds = ["default"]; + const config = {}; + await refreshPreparedModelRuntimeSnapshots(config, { gatewayLifecycle: true }); + const generationA = (await loadPublishedGatewayReplyDispatchRuntime({ agentId: "default" })) + ?.pluginGeneration; + expect(generationA).toBeDefined(); + const generationB = { + ...generationA!, + pluginMetadataSnapshot: { ...generationA!.pluginMetadataSnapshot }, + }; + let finishGenerationA!: () => void; + mocks.ensureOpenClawModelsJson.mockImplementationOnce( + async () => + await new Promise<{ agentDir: string; wrote: false }>((resolve) => { + finishGenerationA = () => + resolve({ agentDir: "/tmp/dynamic-generation-agent", wrote: false }); + }), + ); + const input = { + config, + agentId: "default", + agentDir: "/tmp/dynamic-generation-agent", + workspaceDir: "/tmp/dynamic-generation-workspace", + }; + + const pendingA = acquireAgentRunPreparedModelRuntime(input, { + pluginGeneration: generationA!, + }); + await vi.waitFor(() => expect(mocks.ensureOpenClawModelsJson).toHaveBeenCalledTimes(2)); + const matchingPendingA = acquireAgentRunPreparedModelRuntime(input, { + pluginGeneration: generationA!, + }); + const pendingB = acquireAgentRunPreparedModelRuntime(input, { + pluginGeneration: generationB, + }); + await Promise.resolve(); + expect(mocks.ensureOpenClawModelsJson).toHaveBeenCalledTimes(2); + finishGenerationA(); + const [leaseA, matchingLeaseA, leaseB] = await Promise.all([ + pendingA, + matchingPendingA, + pendingB, + ]); + + expect(matchingLeaseA.snapshot).toBe(leaseA.snapshot); + expect(leaseB.snapshot).not.toBe(leaseA.snapshot); + expect(leaseA.snapshot.metadataSnapshot).toBe(generationA!.pluginMetadataSnapshot); + expect(leaseB.snapshot.metadataSnapshot).toBe(generationB.pluginMetadataSnapshot); + leaseA.release(); + matchingLeaseA.release(); + await expect(prepareModelRuntimeSnapshot(input)).resolves.toBe(leaseB.snapshot); + leaseB.release(); + }); + it("bounds retained gateway run owners while reusing recent selections", async () => { mocks.configuredAgentIds = ["default"]; const config = { agents: { defaults: { model: "openai/gpt-5.5" } } }; diff --git a/src/agents/prepared-model-runtime.owner.ts b/src/agents/prepared-model-runtime.owner.ts index a74f3ad419c7..0dd3e943b679 100644 --- a/src/agents/prepared-model-runtime.owner.ts +++ b/src/agents/prepared-model-runtime.owner.ts @@ -414,6 +414,9 @@ export async function publishPreparedModelRuntimeOwnerBatch(params: { owner.generation += 1; owner.needsRefresh = true; owner.refreshError = undefined; + owner.pendingPluginGeneration = params.reusePluginGenerations + ? owner.pluginGeneration + : undefined; const generation = owner.generation; const key = ownerKey(input); let registered = params.owners.get(key) === owner; @@ -431,6 +434,7 @@ export async function publishPreparedModelRuntimeOwnerBatch(params: { owner.generation === generation && params.owners.get(key) === owner, key, + generation, markRegistered: () => { registered = true; }, @@ -525,6 +529,9 @@ export async function publishPreparedModelRuntimeOwnerBatch(params: { } } for (const candidate of candidates) { + if (candidate.owner.generation === candidate.generation) { + candidate.owner.pendingPluginGeneration = undefined; + } if (!candidate.isCurrent()) { continue; } @@ -542,6 +549,9 @@ export async function publishPreparedModelRuntimeOwnerBatch(params: { } catch (error) { const refreshError = toStringifiedError(error); for (const candidate of candidates) { + if (candidate.owner.generation === candidate.generation) { + candidate.owner.pendingPluginGeneration = undefined; + } if (!candidate.isCurrent()) { continue; } @@ -590,6 +600,7 @@ export async function publishModelRuntimeSnapshot( owner.needsRefresh = true; owner.refreshError = undefined; owner.pluginGeneration = undefined; + owner.pendingPluginGeneration = reusablePluginGeneration; const generation = owner.generation; const build = startSerializedSnapshotBuild( input, @@ -618,11 +629,15 @@ export async function publishModelRuntimeSnapshot( } owner.snapshot = result.snapshot; owner.pluginGeneration = result.pluginGeneration; + owner.pendingPluginGeneration = undefined; owner.pending = undefined; owner.needsRefresh = false; return result.snapshot; } catch (error) { const refreshError = toStringifiedError(error); + if (owner.generation === generation) { + owner.pendingPluginGeneration = undefined; + } if (owner.generation === generation && owners.get(key) === owner) { owner.pending = undefined; owner.needsRefresh = true; diff --git a/src/agents/prepared-model-runtime.types.ts b/src/agents/prepared-model-runtime.types.ts index bfc213dcf908..4082179b0666 100644 --- a/src/agents/prepared-model-runtime.types.ts +++ b/src/agents/prepared-model-runtime.types.ts @@ -150,6 +150,8 @@ export type PreparedModelRuntimeOwner = { refreshError?: Error; snapshot?: PreparedModelRuntimeSnapshot; pluginGeneration?: PreparedModelRuntimePluginGeneration; + /** Explicit generation admitted for the current publication, when known. */ + pendingPluginGeneration?: PreparedModelRuntimePluginGeneration; pending?: Promise; buildCompletion?: Promise; leaseCount?: number;