diff --git a/src/vs/sessions/contrib/automations/browser/providerAutomationService.ts b/src/vs/sessions/contrib/automations/browser/providerAutomationService.ts index 92ae3ecaaa751..9a83474304b42 100644 --- a/src/vs/sessions/contrib/automations/browser/providerAutomationService.ts +++ b/src/vs/sessions/contrib/automations/browser/providerAutomationService.ts @@ -7,10 +7,11 @@ import { disposableTimeout, Sequencer } from '../../../../base/common/async.js'; import { CancellationError, isCancellationError } from '../../../../base/common/errors.js'; import { Disposable, MutableDisposable } from '../../../../base/common/lifecycle.js'; import { derived, IObservable, observableSignalFromEvent } from '../../../../base/common/observable.js'; +import { localize } from '../../../../nls.js'; import { IInstantiationService } from '../../../../platform/instantiation/common/instantiation.js'; import { ILogService } from '../../../../platform/log/common/log.js'; import { IAutomationDescriptor, IAutomationRun, AutomationRunTrigger } from '../../../../workbench/contrib/chat/common/automations/automation.js'; -import { AutomationMutationGuard, IAutomationRunClaim, IAutomationService, ICreateAutomationOptions, IGuardedAutomationUpdateResult, IUpdateAutomationOptions, IUpdateAutomationRunOptions } from '../../../../workbench/contrib/chat/common/automations/automationService.js'; +import { AutomationMutationGuard, IAutomationRunClaim, IAutomationService, ICreateAutomationOptions, IGuardedAutomationUpdateResult, isAutomationActiveRunError, serializeAutomationEditableState, IUpdateAutomationOptions, IUpdateAutomationRunOptions } from '../../../../workbench/contrib/chat/common/automations/automationService.js'; import { ISessionsProvidersService } from '../../../services/sessions/browser/sessionsProvidersService.js'; import { IAutomation, ISessionsProviderAutomations } from '../../../services/sessions/common/sessionsProvider.js'; import { AutomationService } from './automationService.js'; @@ -92,16 +93,35 @@ export class ProviderAutomationService extends Disposable implements IAutomation async updateAutomation(id: string, patch: IUpdateAutomationOptions): Promise { const source = this.requireAutomationStore(id); - const updated = await source.updateAutomation(id, patch); - await this.retargetAutomationStorageIfNeeded(source, updated); - return updated; + let previous = source.getAutomation(id); + while (previous) { + const targetChanged = this.hasTargetChanged(previous, patch.target); + this.assertCanTransferStorage(source, id, patch.target, targetChanged); + const result = await source.updateAutomationIfUnchanged(id, patch, previous); + if (result.kind === 'conflict') { + previous = result.current; + continue; + } + await this.retargetAutomationStorageIfNeeded(source, result.automation, previous, targetChanged); + return result.automation; + } + throw new Error(`Automation '${id}' does not exist.`); } async updateAutomationIfUnchanged(id: string, patch: IUpdateAutomationOptions, expected: IAutomationDescriptor, mutationGuard?: AutomationMutationGuard): Promise { const source = this.requireAutomationStore(id); + const previous = source.getAutomation(id); + if (!previous) { + return { kind: 'conflict', current: undefined }; + } + if (serializeAutomationEditableState(previous) !== serializeAutomationEditableState(expected)) { + return { kind: 'conflict', current: previous }; + } + const targetChanged = this.hasTargetChanged(previous, patch.target); + this.assertCanTransferStorage(source, id, patch.target, targetChanged); const result = await source.updateAutomationIfUnchanged(id, patch, expected, mutationGuard); if (result.kind === 'updated') { - await this.retargetAutomationStorageIfNeeded(source, result.automation); + await this.retargetAutomationStorageIfNeeded(source, result.automation, previous, targetChanged); } return result; } @@ -204,7 +224,22 @@ export class ProviderAutomationService extends Disposable implements IAutomation return this.legacyStore; } - private async retargetAutomationStorageIfNeeded(sourceStore: ISessionsProviderAutomations, initialAutomation: IAutomationDescriptor): Promise { + private hasTargetChanged(current: IAutomationDescriptor, target: IUpdateAutomationOptions['target']): boolean { + return !!target && serializeAutomationEditableState(current) !== serializeAutomationEditableState({ ...current, target }); + } + + private assertCanTransferStorage(source: ISessionsProviderAutomations, automationId: string, target: IUpdateAutomationOptions['target'], targetChanged: boolean): void { + if (!targetChanged || !target || source === this.getTargetStore(target.providerId) || !source.getActiveRunFor(automationId)) { + return; + } + throw this.createActiveRunTransferError(); + } + + private createActiveRunTransferError(): Error { + return new Error(localize('automationActiveRunPreventsTransfer', "Wait for the active run to finish before changing this automation's agent.")); + } + + private async retargetAutomationStorageIfNeeded(sourceStore: ISessionsProviderAutomations, initialAutomation: IAutomationDescriptor, previousAutomation: IAutomationDescriptor, targetChanged: boolean): Promise { let snapshot: IAutomation = { automation: initialAutomation, runs: sourceStore.runsFor(initialAutomation.id).get(), @@ -215,7 +250,25 @@ export class ProviderAutomationService extends Disposable implements IAutomation return; } - await destinationStore.upsertAutomationSnapshot(snapshot); + if (sourceStore.getActiveRunFor(snapshot.automation.id)) { + if (!targetChanged) { + return; + } + await this.restoreSourceAfterBlockedTransfer(sourceStore, initialAutomation, previousAutomation); + throw this.createActiveRunTransferError(); + } + try { + await destinationStore.upsertAutomationSnapshot(snapshot); + } catch (error) { + if (!isAutomationActiveRunError(error)) { + throw error; + } + if (!targetChanged) { + return; + } + await this.restoreSourceAfterBlockedTransfer(sourceStore, initialAutomation, previousAutomation); + throw this.createActiveRunTransferError(); + } if (destinationStore.preservesImportedRunHistory === false) { return; } @@ -238,6 +291,26 @@ export class ProviderAutomationService extends Disposable implements IAutomation this.logService.warn(`[ProviderAutomationService] Automation '${snapshot.automation.id}' kept changing while transferring storage ownership; leaving the source copy in place.`); } + private async restoreSourceAfterBlockedTransfer(sourceStore: ISessionsProviderAutomations, expected: IAutomationDescriptor, previous: IAutomationDescriptor): Promise { + try { + const result = await sourceStore.updateAutomationIfUnchanged(expected.id, { + name: previous.name, + prompt: previous.prompt, + schedule: previous.schedule, + target: previous.target, + modelId: previous.modelId ?? null, + mode: previous.mode ?? null, + permissionLevel: previous.permissionLevel ?? null, + enabled: previous.enabled, + }, expected); + if (result.kind === 'conflict') { + this.logService.warn(`[ProviderAutomationService] Automation '${expected.id}' changed while its active run blocked storage transfer; the latest source state was preserved.`); + } + } catch (error) { + this.logService.error(`[ProviderAutomationService] Failed to restore the source state after an active run blocked storage transfer for '${expected.id}'.`, error); + } + } + private findAutomationStore(id: string): IAutomationStoreEntry | undefined { return this.getStores().find(entry => !!entry.store.getAutomation(id)); } @@ -257,11 +330,15 @@ export class ProviderAutomationService extends Disposable implements IAutomation private queueMigration(): void { this.migrationRetry.clear(); const migration = this.migrationSequencer.queue(async () => { + const initialRecoveryReason = this.staleRunRecoveryReason; + if (initialRecoveryReason) { + await this.recoverStores(this.getStores(), initialRecoveryReason, this.staleRunRecoveryGeneration); + } await this.migrateLegacyAutomations(); await this.completeProviderMigrations(); - const reason = this.staleRunRecoveryReason; - if (reason) { - await this.recoverStores(this.getStores(), reason, this.staleRunRecoveryGeneration); + const finalRecoveryReason = this.staleRunRecoveryReason; + if (finalRecoveryReason) { + await this.recoverStores(this.getStores(), finalRecoveryReason, this.staleRunRecoveryGeneration); } }); this.migrationPromise = migration; @@ -271,7 +348,11 @@ export class ProviderAutomationService extends Disposable implements IAutomation if (this._store.isDisposed || isCancellationError(error)) { return; } - this.logService.error(`[ProviderAutomationService] Failed to migrate legacy Automations; retrying in ${AUTOMATION_MIGRATION_RETRY_DELAY_MS}ms.`, error); + if (isAutomationActiveRunError(error)) { + this.logService.info(`[ProviderAutomationService] Automation migration deferred while a run is active; retrying in ${AUTOMATION_MIGRATION_RETRY_DELAY_MS}ms.`); + } else { + this.logService.error(`[ProviderAutomationService] Failed to migrate legacy Automations; retrying in ${AUTOMATION_MIGRATION_RETRY_DELAY_MS}ms.`, error); + } this.migrationRetry.value = disposableTimeout(() => this.queueMigration(), AUTOMATION_MIGRATION_RETRY_DELAY_MS); }, ); @@ -309,7 +390,11 @@ export class ProviderAutomationService extends Disposable implements IAutomation if (isCancellationError(error) || this._store.isDisposed) { throw new CancellationError(); } - this.logService.error(`[ProviderAutomationService] Failed to migrate Automation '${automation.id}'.`, error); + if (isAutomationActiveRunError(error)) { + this.logService.info(`[ProviderAutomationService] Deferred migration for Automation '${automation.id}' while a run is active.`); + } else { + this.logService.error(`[ProviderAutomationService] Failed to migrate Automation '${automation.id}'.`, error); + } failures.push(error instanceof Error ? error : new Error(String(error))); } } diff --git a/src/vs/sessions/contrib/automations/test/browser/automationService.test.ts b/src/vs/sessions/contrib/automations/test/browser/automationService.test.ts index a56d86a396c41..34aae1239c3bc 100644 --- a/src/vs/sessions/contrib/automations/test/browser/automationService.test.ts +++ b/src/vs/sessions/contrib/automations/test/browser/automationService.test.ts @@ -12,6 +12,7 @@ import { InMemoryStorageService, StorageScope, StorageTarget } from '../../../.. import { NullTelemetryService } from '../../../../../platform/telemetry/common/telemetryUtils.js'; import { AutomationService, AutomationStore } from '../../browser/automationService.js'; import { AutomationRunTrigger, AutomationTarget, AutomationWorkspaceIsolation, IAutomationRun, IAutomationSchedule } from '../../../../../workbench/contrib/chat/common/automations/automation.js'; +import { AutomationActiveRunError, isAutomationActiveRunError } from '../../../../../workbench/contrib/chat/common/automations/automationService.js'; import { createAutomationService, TestAutomationStorageService } from './automationTestUtils.js'; const FOLDER = URI.parse('file:///workspace'); @@ -41,6 +42,21 @@ suite('AutomationService', () => { const teardown = ensureNoDisposablesAreLeakedInTestSuite(); + test('classifies only homogeneous active-run aggregates as deferrals', () => { + const activeRunError = new AutomationActiveRunError('automation', 'run'); + assert.deepStrictEqual({ + direct: isAutomationActiveRunError(activeRunError), + nested: isAutomationActiveRunError(new AggregateError([new AggregateError([activeRunError])])), + mixed: isAutomationActiveRunError(new AggregateError([activeRunError, new Error('storage failed')])), + empty: isAutomationActiveRunError(new AggregateError([])), + }, { + direct: true, + nested: true, + mixed: false, + empty: false, + }); + }); + /** Records a run, asserting the automation's active-run slot was free. */ async function claimRun(service: AutomationService, automationId: string, trigger: AutomationRunTrigger, leaderWindowId = 1): Promise { const claim = await service.recordRunStart(automationId, trigger, leaderWindowId); diff --git a/src/vs/sessions/contrib/automations/test/browser/providerAutomationService.test.ts b/src/vs/sessions/contrib/automations/test/browser/providerAutomationService.test.ts index f7f50aace9440..7d39b2550ff97 100644 --- a/src/vs/sessions/contrib/automations/test/browser/providerAutomationService.test.ts +++ b/src/vs/sessions/contrib/automations/test/browser/providerAutomationService.test.ts @@ -16,6 +16,7 @@ import { ITelemetryService } from '../../../../../platform/telemetry/common/tele import { NullTelemetryService } from '../../../../../platform/telemetry/common/telemetryUtils.js'; import { ISessionsProvidersChangeEvent, ISessionsProvidersService } from '../../../../services/sessions/browser/sessionsProvidersService.js'; import { IAutomation, IAutomationSnapshotImportResult, ISessionsProvider } from '../../../../services/sessions/common/sessionsProvider.js'; +import { AutomationActiveRunError } from '../../../../../workbench/contrib/chat/common/automations/automationService.js'; import { AutomationStore } from '../../browser/automationService.js'; import { ProviderAutomationService } from '../../browser/providerAutomationService.js'; import { AUTOMATION_STORAGE_KEY, IAutomationStorageService, providerAutomationStorageKey } from '../../common/automationStorageService.js'; @@ -31,6 +32,24 @@ class FailingStaleRunRecoveryAutomationStore extends AutomationStore { } } +class MigrationDeferringAutomationStore extends AutomationStore { + recoveryCalls = 0; + migrationCalls = 0; + + override async markStaleRunsFailed(reason: string): Promise { + this.recoveryCalls++; + await super.markStaleRunsFailed(reason); + } + + async completeMigration(): Promise { + this.migrationCalls++; + const activeRun = this.runs.get().find(run => run.status === 'pending' || run.status === 'running'); + if (activeRun) { + throw new AutomationActiveRunError(activeRun.automationId, activeRun.id); + } + } +} + class PartiallyFailingMigrationAutomationStore extends AutomationStore { override async importAutomationSnapshot(snapshot: IAutomation): Promise { if (snapshot.automation.id === 'automation-1') { @@ -238,6 +257,7 @@ suite('ProviderAutomationService', () => { target: legacyTarget, }); const claim = await service.recordRunStart(created.id, 'manual', 1); + await service.updateRun(claim.run.id, { status: 'completed', completedAt: '2026-01-01T00:01:00.000Z' }); const transferToProvider = await service.updateAutomationIfUnchanged(created.id, { target: providerTarget }, created); const afterProviderTransfer = { @@ -282,6 +302,62 @@ suite('ProviderAutomationService', () => { }); }); + test('does not change storage ownership while an Automation run is active', async () => { + const { service, providerStore, storage } = createService(); + const legacyTarget = { kind: 'workspace', folderUri: FOLDER, providerId: 'provider-without-storage', sessionTypeId: 'other', isolation: { kind: 'default' } } as const; + const providerTarget = { kind: 'workspace', folderUri: FOLDER, providerId: PROVIDER_ID, sessionTypeId: SESSION_TYPE_ID, isolation: { kind: 'default' } } as const; + const created = await service.createAutomation({ + name: 'Active', + prompt: 'prompt', + schedule: { interval: 'manual', scheduleHour: 0, scheduleMinute: 0, scheduleDay: 0 }, + target: legacyTarget, + }); + await service.recordRunStart(created.id, 'manual', 1); + + await assert.rejects(service.updateAutomation(created.id, { target: providerTarget }), /Wait for the active run to finish/); + const legacyLedger = JSON.parse(storage.get(AUTOMATION_STORAGE_KEY, StorageScope.APPLICATION)!); + + assert.deepStrictEqual({ + providerAutomation: providerStore.getAutomation(created.id), + legacyTarget: URI.revive(legacyLedger.automations[0].target.folderUri).toString(), + legacyProviderId: legacyLedger.automations[0].target.providerId, + legacyRunStatus: legacyLedger.runs[0].status, + }, { + providerAutomation: undefined, + legacyTarget: FOLDER.toString(), + legacyProviderId: 'provider-without-storage', + legacyRunStatus: 'pending', + }); + }); + + test('allows unrelated edits while an active run defers storage migration', async () => { + const { service, providerStore, storage, automationStorage } = createService(); + await service.waitForMigrationForTesting(); + const legacy = teardown.add(new AutomationStore(AUTOMATION_STORAGE_KEY, storage, new NullLogService(), NullTelemetryService, automationStorage)); + const target = { kind: 'workspace', folderUri: FOLDER, providerId: PROVIDER_ID, sessionTypeId: SESSION_TYPE_ID, isolation: { kind: 'default' } } as const; + const created = await legacy.createAutomation({ + name: 'Active', + prompt: 'prompt', + schedule: { interval: 'manual', scheduleHour: 0, scheduleMinute: 0, scheduleDay: 0 }, + target, + }); + await legacy.recordRunStart(created.id, 'manual', 1); + + const updated = await service.updateAutomation(created.id, { name: 'Renamed', target }); + + assert.deepStrictEqual({ + updatedName: updated.name, + providerAutomation: providerStore.getAutomation(created.id), + legacyName: service.getAutomation(created.id)?.name, + activeRunStatus: service.getActiveRunFor(created.id)?.status, + }, { + updatedName: 'Renamed', + providerAutomation: undefined, + legacyName: 'Renamed', + activeRunStatus: 'pending', + }); + }); + test('does not transfer an Automation when a guarded update conflicts', async () => { const { service, providerStore, storage } = createService(); const created = await service.createAutomation({ @@ -290,6 +366,7 @@ suite('ProviderAutomationService', () => { schedule: { interval: 'manual', scheduleHour: 0, scheduleMinute: 0, scheduleDay: 0 }, target: { kind: 'workspace', folderUri: FOLDER, providerId: PROVIDER_ID, sessionTypeId: SESSION_TYPE_ID, isolation: { kind: 'default' } }, }); + await service.recordRunStart(created.id, 'manual', 1); const result = await service.updateAutomationIfUnchanged(created.id, { target: { kind: 'workspace', folderUri: FOLDER, providerId: 'provider-without-storage', sessionTypeId: 'other', isolation: { kind: 'default' } }, @@ -329,25 +406,51 @@ suite('ProviderAutomationService', () => { }); }); - test('retries ownership transfer when a run is added concurrently', async () => { + test('restores the source target when a run starts during ownership transfer', async () => { const { service, providerStore, storage } = createService(undefined, undefined, 'concurrentTransferRun'); + const legacyTarget = { kind: 'workspace', folderUri: FOLDER, providerId: 'provider-without-storage', sessionTypeId: 'other', isolation: { kind: 'default' } } as const; const created = await service.createAutomation({ name: 'Legacy', prompt: 'prompt', schedule: { interval: 'manual', scheduleHour: 0, scheduleMinute: 0, scheduleDay: 0 }, - target: { kind: 'workspace', folderUri: FOLDER, providerId: 'provider-without-storage', sessionTypeId: 'other', isolation: { kind: 'default' } }, + target: legacyTarget, }); - await service.updateAutomation(created.id, { + await assert.rejects(service.updateAutomation(created.id, { + name: 'Updated', + prompt: 'Updated prompt', + schedule: { interval: 'daily', scheduleHour: 8, scheduleMinute: 30, scheduleDay: 0 }, target: { kind: 'workspace', folderUri: FOLDER, providerId: PROVIDER_ID, sessionTypeId: SESSION_TYPE_ID, isolation: { kind: 'default' } }, - }); + modelId: 'new-model', + mode: 'plan', + permissionLevel: 'autoApprove', + enabled: false, + }), /Wait for the active run to finish/); + const legacyLedger = JSON.parse(storage.get(AUTOMATION_STORAGE_KEY, StorageScope.APPLICATION)!); + const restoredTarget = legacyLedger.automations[0].target; assert.deepStrictEqual({ - providerRunAutomationIds: providerStore.runs.get().map(run => run.automationId), - legacyAutomationIds: JSON.parse(storage.get(AUTOMATION_STORAGE_KEY, StorageScope.APPLICATION)!).automations.map((automation: { id: string }) => automation.id), + providerAutomation: providerStore.getAutomation(created.id), + legacyName: legacyLedger.automations[0].name, + legacyPrompt: legacyLedger.automations[0].prompt, + legacySchedule: legacyLedger.automations[0].schedule, + legacyTarget: { ...restoredTarget, folderUri: URI.revive(restoredTarget.folderUri).toString() }, + legacyModelId: legacyLedger.automations[0].modelId, + legacyMode: legacyLedger.automations[0].mode, + legacyPermissionLevel: legacyLedger.automations[0].permissionLevel, + legacyEnabled: legacyLedger.automations[0].enabled, + legacyRunStatuses: legacyLedger.runs.map((run: { status: string }) => run.status), }, { - providerRunAutomationIds: [created.id], - legacyAutomationIds: [], + providerAutomation: undefined, + legacyName: 'Legacy', + legacyPrompt: 'prompt', + legacySchedule: { interval: 'manual', scheduleHour: 0, scheduleMinute: 0, scheduleDay: 0 }, + legacyTarget: { ...legacyTarget, folderUri: FOLDER.toString() }, + legacyModelId: undefined, + legacyMode: undefined, + legacyPermissionLevel: undefined, + legacyEnabled: true, + legacyRunStatuses: ['pending'], }); }); @@ -819,6 +922,33 @@ suite('ProviderAutomationService', () => { }); }); + test('recovers a late provider before completing its migration', async () => { + const { service, storage, automationStorage, addProvider } = createService(); + await service.startStaleRunRecovery('Recovered after restart.'); + const providerId = 'late-migrating-provider'; + const store = teardown.add(new MigrationDeferringAutomationStore(providerAutomationStorageKey(providerId), storage, new NullLogService(), NullTelemetryService, automationStorage)); + const automation = await store.createAutomation({ + name: 'Late migration', + prompt: 'prompt', + schedule: { interval: 'manual', scheduleHour: 0, scheduleMinute: 0, scheduleDay: 0 }, + target: { kind: 'workspace', folderUri: FOLDER, providerId, sessionTypeId: 'late', isolation: { kind: 'default' } }, + }); + await store.recordRunStart(automation.id, 'manual', 1); + + addProvider(upcastPartial({ id: providerId, order: 1, automations: store })); + await service.waitForMigrationForTesting(); + + assert.deepStrictEqual({ + runStatuses: store.runs.get().map(run => run.status), + recoveryCalls: store.recoveryCalls, + migrationCalls: store.migrationCalls, + }, { + runStatuses: ['failed'], + recoveryCalls: 1, + migrationCalls: 1, + }); + }); + test('migrates before recovering a provider added while initial recovery is queued', async () => { const lateProviderId = 'late-provider'; const legacy = JSON.stringify({ diff --git a/src/vs/sessions/contrib/providers/agentHost/browser/agentHostAutomationStore.ts b/src/vs/sessions/contrib/providers/agentHost/browser/agentHostAutomationStore.ts index 69c79ef99c660..39644389e9d17 100644 --- a/src/vs/sessions/contrib/providers/agentHost/browser/agentHostAutomationStore.ts +++ b/src/vs/sessions/contrib/providers/agentHost/browser/agentHostAutomationStore.ts @@ -23,7 +23,7 @@ import { ILogService } from '../../../../../platform/log/common/log.js'; import { IStorageService, StorageScope } from '../../../../../platform/storage/common/storage.js'; import { ITelemetryService } from '../../../../../platform/telemetry/common/telemetry.js'; import type { AutomationRunTrigger, AutomationTarget, IAutomationDescriptor, IAutomationRun, IAutomationSchedule } from '../../../../../workbench/contrib/chat/common/automations/automation.js'; -import { type AutomationMutationGuard, type IAutomationRunClaim, type ICreateAutomationOptions, type IGuardedAutomationUpdateResult, serializeAutomationEditableState, type IUpdateAutomationOptions, type IUpdateAutomationRunOptions } from '../../../../../workbench/contrib/chat/common/automations/automationService.js'; +import { AutomationActiveRunError, type AutomationMutationGuard, type IAutomationRunClaim, type ICreateAutomationOptions, type IGuardedAutomationUpdateResult, isAutomationActiveRunError, serializeAutomationEditableState, type IUpdateAutomationOptions, type IUpdateAutomationRunOptions } from '../../../../../workbench/contrib/chat/common/automations/automationService.js'; import { publishAutomationMigration } from '../../../../../workbench/contrib/chat/common/automations/automationTelemetry.js'; import type { IAutomation, IAutomationSnapshotImportResult, IGuardedAutomationSnapshotRemovalResult, ISessionsProviderAutomations } from '../../../../services/sessions/common/sessionsProvider.js'; import { IAutomationStorageService } from '../../../automations/common/automationStorageService.js'; @@ -56,6 +56,11 @@ interface ILegacyRunArchive { readonly runs: readonly ISerializedArchivedRun[]; } +interface ILoadedLegacyRunArchive { + readonly runs: readonly IAutomationRun[]; + readonly repairedRuns: number; +} + export interface IAgentHostAutomationBoundaryMapper { toHost(resource: URI): URI; fromHost(resource: URI): URI; @@ -78,6 +83,7 @@ export class AgentHostAutomationStore extends Disposable implements ISessionsPro private readonly _archiveKey: string; private readonly _archivedRuns; private _migrationPromise: Promise | undefined; + private _lastPreflightDeferralKey: string | undefined; readonly automations: IObservable; readonly runs: IObservable; @@ -94,9 +100,13 @@ export class AgentHostAutomationStore extends Disposable implements ISessionsPro ) { super(); this._archiveKey = `agentHostAutomation.legacyRunArchive.${_providerId}`; - this._archivedRuns = observableValue(this, this._loadArchivedRuns()); + const archive = this._loadArchivedRuns(); + this._archivedRuns = observableValue(this, archive.runs); + this._persistRepairedArchivedRuns(archive); this._register(this._storageService.onDidChangeValue(StorageScope.APPLICATION, this._archiveKey, this._store)(() => { - this._archivedRuns.set(this._loadArchivedRuns(), undefined); + const archive = this._loadArchivedRuns(); + this._archivedRuns.set(archive.runs, undefined); + this._persistRepairedArchivedRuns(archive); })); this._catalogReference = this._register(_connection.getSubscription( StateComponents.AutomationCatalog, @@ -253,6 +263,7 @@ export class AgentHostAutomationStore extends Disposable implements ISessionsPro } private async _importAutomationSnapshot(snapshot: IAutomation, importPending: boolean): Promise { + assertTerminalRunHistory(snapshot.runs); const existing = this._findAutomationState(snapshot.automation.id); if (existing) { const current = this._requireProjectedAutomation(existing); @@ -282,6 +293,7 @@ export class AgentHostAutomationStore extends Disposable implements ISessionsPro } async upsertAutomationSnapshot(snapshot: IAutomation): Promise { + assertTerminalRunHistory(snapshot.runs); if (this._findAutomationState(snapshot.automation.id)) { await this._replaceDescriptor(snapshot.automation, true, true); } else { @@ -409,6 +421,25 @@ export class AgentHostAutomationStore extends Disposable implements ISessionsPro const startedAt = Date.now(); const source = this._legacySource; const discovered = source ? [...source.automations.get()] : []; + const activeRuns = source + ? discovered.flatMap(automation => source.runsFor(automation.id).get().filter(isNonTerminalRun)) + : []; + if (activeRuns.length > 0) { + const deferralKey = activeRuns.map(run => run.id).sort().join(','); + if (this._lastPreflightDeferralKey !== deferralKey) { + this._lastPreflightDeferralKey = deferralKey; + publishAutomationMigration(this._telemetryService, { + outcome: 'deferred', + discoveredCount: discovered.length, + migratedCount: 0, + failedCount: 0, + durationMs: Date.now() - startedAt, + }); + } + this._logService.info(`[AgentHostAutomationStore] Automation migration deferred: activeRuns=${activeRuns.length}.`); + throw new AutomationActiveRunError(activeRuns[0].automationId, activeRuns[0].id); + } + this._lastPreflightDeferralKey = undefined; this._logService.info(`[AgentHostAutomationStore] Automation migration started: discovered=${discovered.length}.`); publishAutomationMigration(this._telemetryService, { outcome: 'started', @@ -435,7 +466,11 @@ export class AgentHostAutomationStore extends Disposable implements ISessionsPro const failure = error instanceof Error ? error : new Error(String(error)); failures.push(failure); failedCount++; - this._logService.error(`[AgentHostAutomationStore] Automation migration item failed: resource=${automationResource(automation.id)}, error=${failure.message}`); + if (isAutomationActiveRunError(error)) { + this._logService.info(`[AgentHostAutomationStore] Automation migration item deferred while a run is active: resource=${automationResource(automation.id)}.`); + } else { + this._logService.error(`[AgentHostAutomationStore] Automation migration item failed: resource=${automationResource(automation.id)}, error=${failure.message}`); + } } } if (failures.length > 0) { @@ -475,10 +510,21 @@ export class AgentHostAutomationStore extends Disposable implements ISessionsPro if (isCancellationError(error)) { throw error; } + const durationMs = Date.now() - startedAt; + if (isAutomationActiveRunError(error)) { + this._logService.info(`[AgentHostAutomationStore] Automation migration deferred after ${migratedCount} item(s) while a run became active.`); + publishAutomationMigration(this._telemetryService, { + outcome: 'deferred', + discoveredCount: discovered.length, + migratedCount, + failedCount: 0, + durationMs, + }); + throw error; + } if (error instanceof AggregateError) { failedCount = Math.max(failedCount, error.errors.length); } - const durationMs = Date.now() - startedAt; this._logService.error(`[AgentHostAutomationStore] Automation migration failed: discovered=${discovered.length}, migrated=${migratedCount}, failed=${failedCount}, durationMs=${durationMs}, error=${error instanceof Error ? error.message : String(error)}.`); publishAutomationMigration(this._telemetryService, { outcome: 'failed', @@ -980,24 +1026,77 @@ export class AgentHostAutomationStore extends Disposable implements ISessionsPro throw lastError ?? new Error('Timed out waiting for Agent Host Automation migration completion.'); } - private _loadArchivedRuns(): readonly IAutomationRun[] { + private _loadArchivedRuns(): ILoadedLegacyRunArchive { const raw = this._storageService.get(this._archiveKey, StorageScope.APPLICATION); if (!raw) { - return []; + return { runs: [], repairedRuns: 0 }; } const parsed = parseArchivedRuns(raw); if (parsed.kind === 'unsupported') { this._logService.error(`[AgentHostAutomationStore] Ignoring legacy run archive with unsupported version: key=${this._archiveKey}, version=${parsed.version}.`); - return []; + return { runs: [], repairedRuns: 0 }; } if (parsed.kind === 'invalid') { this._logService.error(`[AgentHostAutomationStore] Ignoring invalid legacy run archive: key=${this._archiveKey}, error=${parsed.error}.`); - return []; + return { runs: [], repairedRuns: 0 }; } if (parsed.droppedRuns > 0) { this._logService.warn(`[AgentHostAutomationStore] Dropped ${parsed.droppedRuns} malformed run(s) from legacy run archive: key=${this._archiveKey}.`); } - return parsed.runs; + let repairedRuns = 0; + const runs = parsed.runs.map(run => { + const terminalRun = terminalizeArchivedRun(run); + if (terminalRun !== run) { + repairedRuns++; + } + return terminalRun; + }); + return { runs, repairedRuns }; + } + + private _persistRepairedArchivedRuns(archive: ILoadedLegacyRunArchive): void { + if (archive.repairedRuns === 0) { + return; + } + this._logService.warn(`[AgentHostAutomationStore] Repairing ${archive.repairedRuns} non-terminal legacy Automation run(s): key=${this._archiveKey}.`); + void this._repairArchivedRuns().catch(error => { + this._logService.error(`[AgentHostAutomationStore] Failed to persist repaired legacy Automation runs: key=${this._archiveKey}, error=${error instanceof Error ? error.message : String(error)}.`); + }); + } + + private async _repairArchivedRuns(): Promise { + let raw = await this._automationStorageService.read(this._archiveKey); + for (let attempt = 0; attempt < LEGACY_RUN_ARCHIVE_WRITE_ATTEMPTS; attempt++) { + if (raw === undefined) { + return; + } + const parsed = parseArchivedRuns(raw); + if (parsed.kind === 'unsupported') { + throw new Error(`Cannot repair legacy Automation run archive with unsupported version: key=${this._archiveKey}, version=${parsed.version}.`); + } + if (parsed.kind === 'invalid') { + throw new Error(`Cannot repair invalid legacy Automation run archive: key=${this._archiveKey}, error=${parsed.error}.`); + } + const runs = parsed.runs.map(terminalizeArchivedRun); + if (runs.every((run, index) => run === parsed.runs[index])) { + this._archivedRuns.set(runs, undefined); + return; + } + const archive: ILegacyRunArchive = { + version: LEGACY_RUN_ARCHIVE_VERSION, + runs: runs.map(run => ({ + ...run, + sessionResource: run.sessionResource?.toString(), + })), + }; + const result = await this._automationStorageService.compareAndSwap(this._archiveKey, raw, JSON.stringify(archive)); + if (result.swapped) { + this._archivedRuns.set(runs, undefined); + return; + } + raw = result.currentValue; + } + throw new Error(`Legacy Automation run archive kept changing while it was being repaired: ${this._archiveKey}`); } private async _archiveRuns(runs: readonly IAutomationRun[]): Promise { @@ -1021,7 +1120,7 @@ export class AgentHostAutomationStore extends Disposable implements ISessionsPro } } } - const merged = distinctById([...runs, ...current]); + const merged = distinctById([...runs, ...current]).map(terminalizeArchivedRun); const archive: ILegacyRunArchive = { version: LEGACY_RUN_ARCHIVE_VERSION, runs: merged.map(run => ({ @@ -1146,6 +1245,32 @@ function distinctById(items: readonly T[]): T return result; } +function assertTerminalRunHistory(runs: readonly IAutomationRun[]): void { + const activeRun = runs.find(isNonTerminalRun); + if (activeRun) { + throw new AutomationActiveRunError(activeRun.automationId, activeRun.id); + } +} + +function isNonTerminalRun(run: IAutomationRun): boolean { + return run.status === 'pending' || run.status === 'running'; +} + +/** + * Reuses the run's own timestamp because the interruption instant is unknowable and deterministic repair must be idempotent. + */ +function terminalizeArchivedRun(run: IAutomationRun): IAutomationRun { + if (!isNonTerminalRun(run)) { + return run; + } + return Object.freeze({ + ...run, + status: 'failed', + completedAt: run.completedAt ?? run.startedAt, + errorMessage: run.errorMessage ?? localize('agentHostAutomation.interruptedLegacyRun', "Interrupted while migrating Automation history"), + }); +} + function isSerializedArchivedRun(value: unknown): value is ISerializedArchivedRun { if (!value || typeof value !== 'object' || Array.isArray(value)) { return false; diff --git a/src/vs/sessions/contrib/providers/agentHost/browser/reconnectableAgentHostAutomationStore.ts b/src/vs/sessions/contrib/providers/agentHost/browser/reconnectableAgentHostAutomationStore.ts index 1c83ffb14cff8..819eb4a17b454 100644 --- a/src/vs/sessions/contrib/providers/agentHost/browser/reconnectableAgentHostAutomationStore.ts +++ b/src/vs/sessions/contrib/providers/agentHost/browser/reconnectableAgentHostAutomationStore.ts @@ -12,7 +12,7 @@ import { IInstantiationService } from '../../../../../platform/instantiation/com import { IConfigurationService } from '../../../../../platform/configuration/common/configuration.js'; import { ILogService } from '../../../../../platform/log/common/log.js'; import type { AutomationRunTrigger, IAutomationDescriptor, IAutomationRun } from '../../../../../workbench/contrib/chat/common/automations/automation.js'; -import type { AutomationMutationGuard, IAutomationRunClaim, ICreateAutomationOptions, IGuardedAutomationUpdateResult, IUpdateAutomationOptions, IUpdateAutomationRunOptions } from '../../../../../workbench/contrib/chat/common/automations/automationService.js'; +import { isAutomationActiveRunError, type AutomationMutationGuard, type IAutomationRunClaim, type ICreateAutomationOptions, type IGuardedAutomationUpdateResult, type IUpdateAutomationOptions, type IUpdateAutomationRunOptions } from '../../../../../workbench/contrib/chat/common/automations/automationService.js'; import type { IAutomation, IAutomationSnapshotImportResult, IGuardedAutomationSnapshotRemovalResult, ISessionsProviderAutomations } from '../../../../services/sessions/common/sessionsProvider.js'; import { AgentHostAutomationStore, type IAgentHostAutomationBoundaryMapper, type IAgentHostAutomationConnection } from './agentHostAutomationStore.js'; import { CHAT_AUTOMATIONS_ENABLED_SETTING } from '../../../../../workbench/contrib/chat/common/automations/automationsEnabled.js'; @@ -241,7 +241,11 @@ export class ReconnectableAgentHostAutomationStore extends Disposable implements if (this._store.isDisposed || isCancellationError(error) || this._currentStore.get() !== store) { return; } - this._logService.error(`[ReconnectableAgentHostAutomationStore] Failed to initialize remote Automation authority; retrying in ${MIGRATION_RETRY_DELAY_MS}ms.`, error); + if (isAutomationActiveRunError(error)) { + this._logService.info(`[ReconnectableAgentHostAutomationStore] Automation migration deferred while a legacy run is active; retrying in ${MIGRATION_RETRY_DELAY_MS}ms.`); + } else { + this._logService.error(`[ReconnectableAgentHostAutomationStore] Failed to initialize remote Automation authority; retrying in ${MIGRATION_RETRY_DELAY_MS}ms.`, error); + } this._migrationRetry.value = disposableTimeout(() => this._completeMigration(store), MIGRATION_RETRY_DELAY_MS); }); } diff --git a/src/vs/sessions/contrib/providers/agentHost/test/browser/agentHostAutomationStore.test.ts b/src/vs/sessions/contrib/providers/agentHost/test/browser/agentHostAutomationStore.test.ts index 247d4d7cc39ed..f7765b2500d15 100644 --- a/src/vs/sessions/contrib/providers/agentHost/test/browser/agentHostAutomationStore.test.ts +++ b/src/vs/sessions/contrib/providers/agentHost/test/browser/agentHostAutomationStore.test.ts @@ -282,6 +282,21 @@ class FailingArchiveStorageService extends TestAutomationStorageService { } } +class PausedArchiveRepairStorageService extends TestAutomationStorageService { + readonly repairWriteStarted = new DeferredPromise(); + readonly resumeRepairWrite = new DeferredPromise(); + pauseNextArchiveWrite = false; + + override async compareAndSwap(key: string, expectedValue: string | undefined, newValue: string) { + if (this.pauseNextArchiveWrite && key.startsWith('agentHostAutomation.legacyRunArchive.')) { + this.pauseNextArchiveWrite = false; + await this.repairWriteStarted.complete(); + await this.resumeRepairWrite.p; + } + return super.compareAndSwap(key, expectedValue, newValue); + } +} + class ToggleMigrationAutomationStore extends AutomationStore { migrationAllowed = false; @@ -305,14 +320,33 @@ class PausedRemovalAutomationStore extends AutomationStore { } } +class RunStartingDuringMigrationAutomationStore extends AutomationStore { + automationToStart: string | undefined; + + override async removeAutomationSnapshotIfUnchanged(expected: IAutomation) { + const result = await super.removeAutomationSnapshotIfUnchanged(expected); + if (this.automationToStart) { + const automationId = this.automationToStart; + this.automationToStart = undefined; + await this.recordRunStart(automationId, 'manual', 1); + } + return result; + } +} + class RecordingLogService extends NullLogService { readonly errors: string[] = []; + readonly infos: string[] = []; readonly warnings: string[] = []; override error(message: string, ..._args: unknown[]): void { this.errors.push(message); } + override info(message: string, ..._args: unknown[]): void { + this.infos.push(message); + } + override warn(message: string, ..._args: unknown[]): void { this.warnings.push(message); } @@ -769,6 +803,265 @@ suite('AgentHostAutomationStore', () => { ); }); + test('defers migration without partial imports until a live legacy run is terminal', async () => { + const connection = new TestAutomationConnection(false); + disposables.add(connection); + const storage = disposables.add(new InMemoryStorageService()); + const automationStorage = new TestAutomationStorageService(storage); + const legacy = disposables.add(new AutomationStore(providerAutomationStorageKey('local-agent-host'), storage, new NullLogService(), NullTelemetryService, automationStorage)); + const automation = await legacy.createAutomation({ + name: 'Active legacy run', + prompt: 'Review.', + schedule: { interval: 'manual', scheduleHour: 0, scheduleMinute: 0, scheduleDay: 0 }, + target: { kind: 'quickChat', providerId: 'local-agent-host', sessionTypeId: 'mock' }, + }); + const activeRun = await legacy.recordRunStart(automation.id, 'manual', 0); + const telemetryService = new RecordingTelemetryService(); + const store = disposables.add(new AgentHostAutomationStore('local-agent-host', connection, legacy, undefined, new NullLogService(), storage, telemetryService, automationStorage)); + + await assert.rejects(store.completeMigration(), /has active run/); + await assert.rejects(store.completeMigration(), /has active run/); + assert.deepStrictEqual({ + activeRunId: legacy.getActiveRunFor(automation.id)?.id, + hostCreateRequests: connection.dispatched.filter(entry => entry.action.type === ActionType.AutomationCreateRequested).length, + migrationOutcomes: telemetryService.events + .filter(event => event.name === 'automation.migration') + .map(event => event.data['outcome']), + }, { + activeRunId: activeRun.run.id, + hostCreateRequests: 0, + migrationOutcomes: ['deferred'], + }); + + await legacy.updateRun(activeRun.run.id, { + status: 'completed', + completedAt: '2026-01-01T00:01:00.000Z', + }); + await store.completeMigration(); + + assert.deepStrictEqual({ + legacyAutomations: legacy.automations.get(), + runs: store.runs.get().map(run => ({ + id: run.id, + status: run.status, + errorMessage: run.errorMessage, + })), + migrationOutcomes: telemetryService.events + .filter(event => event.name === 'automation.migration') + .map(event => event.data['outcome']), + }, { + legacyAutomations: [], + runs: [{ + id: activeRun.run.id, + status: 'completed', + errorMessage: undefined, + }], + migrationOutcomes: ['deferred', 'started', 'completed'], + }); + }); + + test('recovers a stale legacy run before retrying migration', async () => { + const connection = new TestAutomationConnection(false); + disposables.add(connection); + const storage = disposables.add(new InMemoryStorageService()); + const automationStorage = new TestAutomationStorageService(storage); + const legacy = disposables.add(new AutomationStore(providerAutomationStorageKey('local-agent-host'), storage, new NullLogService(), NullTelemetryService, automationStorage)); + const automation = await legacy.createAutomation({ + name: 'Stale legacy run', + prompt: 'Review.', + schedule: { interval: 'manual', scheduleHour: 0, scheduleMinute: 0, scheduleDay: 0 }, + target: { kind: 'quickChat', providerId: 'local-agent-host', sessionTypeId: 'mock' }, + }); + const staleRun = await legacy.recordRunStart(automation.id, 'manual', 0); + const store = disposables.add(new AgentHostAutomationStore('local-agent-host', connection, legacy, undefined, new NullLogService(), storage, NullTelemetryService, automationStorage)); + + await assert.rejects(store.completeMigration(), /has active run/); + await store.markStaleRunsFailed('Interrupted by app shutdown'); + await store.completeMigration(); + + assert.deepStrictEqual({ + legacyAutomations: legacy.automations.get(), + runs: store.runs.get().map(run => ({ + id: run.id, + status: run.status, + errorMessage: run.errorMessage, + })), + }, { + legacyAutomations: [], + runs: [{ + id: staleRun.run.id, + status: 'failed', + errorMessage: 'Interrupted by app shutdown', + }], + }); + }); + + test('reports a run starting between migration items as deferred', async () => { + const connection = new TestAutomationConnection(false); + disposables.add(connection); + const storage = disposables.add(new InMemoryStorageService()); + const automationStorage = new TestAutomationStorageService(storage); + const legacy = disposables.add(new RunStartingDuringMigrationAutomationStore(providerAutomationStorageKey('local-agent-host'), storage, new NullLogService(), NullTelemetryService, automationStorage)); + const blocked = await legacy.createAutomation({ + name: 'Blocked', + prompt: 'Review.', + schedule: { interval: 'manual', scheduleHour: 0, scheduleMinute: 0, scheduleDay: 0 }, + target: { kind: 'quickChat', providerId: 'local-agent-host', sessionTypeId: 'mock' }, + }); + await legacy.createAutomation({ + name: 'First', + prompt: 'Review.', + schedule: { interval: 'manual', scheduleHour: 0, scheduleMinute: 0, scheduleDay: 0 }, + target: { kind: 'quickChat', providerId: 'local-agent-host', sessionTypeId: 'mock' }, + }); + legacy.automationToStart = blocked.id; + const logService = new RecordingLogService(); + const telemetryService = new RecordingTelemetryService(); + const store = disposables.add(new AgentHostAutomationStore('local-agent-host', connection, legacy, undefined, logService, storage, telemetryService, automationStorage)); + + await assert.rejects(store.completeMigration(), /Failed to migrate 1 Agent Host Automation definition/); + + assert.deepStrictEqual({ + errorLogs: logService.errors, + deferredLogs: logService.infos.filter(message => message.includes('deferred')).length, + migrationOutcomes: telemetryService.events + .filter(event => event.name === 'automation.migration') + .map(event => event.data['outcome']), + }, { + errorLogs: [], + deferredLogs: 2, + migrationOutcomes: ['started', 'deferred'], + }); + }); + + test('repairs active archived legacy runs without changing authoritative host runs', async () => { + const connection = new TestAutomationConnection(true); + disposables.add(connection); + const storage = disposables.add(new InMemoryStorageService()); + const automationStorage = new TestAutomationStorageService(storage); + const archiveKey = 'agentHostAutomation.legacyRunArchive.local-agent-host'; + const snapshot = archivedSnapshot('archived', 'legacy-run'); + await automationStorage.compareAndSwap(archiveKey, undefined, JSON.stringify({ + version: 1, + runs: [{ + ...snapshot.runs[0], + status: 'running', + sessionResource: undefined, + completedAt: undefined, + errorMessage: 'Existing interruption reason', + }], + })); + const store = disposables.add(new AgentHostAutomationStore('local-agent-host', connection, undefined, undefined, new NullLogService(), storage, NullTelemetryService, automationStorage)); + const initiallyRepairedRun = store.runs.get().find(run => run.id === 'legacy-run'); + const automation = await store.createAutomation({ + name: 'Host run', + prompt: 'Review.', + schedule: { interval: 'manual', scheduleHour: 0, scheduleMinute: 0, scheduleDay: 0 }, + target: { kind: 'quickChat', providerId: 'local-agent-host', sessionTypeId: 'mock' }, + }); + const hostRun = await store.recordRunStart(automation.id, 'manual', 0); + void hostRun.externalDispatch?.whenCompleted.catch(() => { }); + await timeout(0); + const persistedArchive = JSON.parse((await automationStorage.read(archiveKey))!); + const persistedRun = persistedArchive.runs.find((run: { id: string }) => run.id === 'legacy-run'); + const restored = disposables.add(new AgentHostAutomationStore('local-agent-host', connection, undefined, undefined, new NullLogService(), storage, NullTelemetryService, automationStorage)); + + assert.deepStrictEqual({ + runs: restored.runs.get().map(run => ({ + id: run.id, + status: run.status, + errorMessage: run.errorMessage, + completedAt: run.completedAt, + })), + persistedRun: { + status: persistedRun.status, + errorMessage: persistedRun.errorMessage, + completedAt: persistedRun.completedAt, + }, + initialCompletedAt: initiallyRepairedRun?.completedAt, + restoredCompletedAt: restored.runs.get().find(run => run.id === 'legacy-run')?.completedAt, + }, { + runs: [{ + id: hostRun.run.id, + status: 'running', + errorMessage: undefined, + completedAt: undefined, + }, { + id: 'legacy-run', + status: 'failed', + errorMessage: 'Existing interruption reason', + completedAt: snapshot.runs[0].startedAt, + }], + persistedRun: { + status: 'failed', + errorMessage: 'Existing interruption reason', + completedAt: snapshot.runs[0].startedAt, + }, + initialCompletedAt: snapshot.runs[0].startedAt, + restoredCompletedAt: snapshot.runs[0].startedAt, + }); + }); + + test('archive repair preserves a terminal run written after its initial read', async () => { + const connection = new TestAutomationConnection(true); + disposables.add(connection); + const storage = disposables.add(new InMemoryStorageService()); + const automationStorage = new PausedArchiveRepairStorageService(storage); + const concurrentStorage = new TestAutomationStorageService(storage); + const archiveKey = 'agentHostAutomation.legacyRunArchive.local-agent-host'; + const snapshot = archivedSnapshot('archived', 'legacy-run'); + const runningArchive = JSON.stringify({ + version: 1, + runs: [{ + ...snapshot.runs[0], + status: 'running', + completedAt: undefined, + sessionResource: snapshot.runs[0].sessionResource?.toString(), + }], + }); + await automationStorage.compareAndSwap(archiveKey, undefined, runningArchive); + automationStorage.pauseNextArchiveWrite = true; + const store = disposables.add(new AgentHostAutomationStore('local-agent-host', connection, undefined, undefined, new NullLogService(), storage, NullTelemetryService, automationStorage)); + await automationStorage.repairWriteStarted.p; + const completedAt = '2026-01-02T00:02:00.000Z'; + const completedArchive = JSON.stringify({ + version: 1, + runs: [{ + ...snapshot.runs[0], + status: 'completed', + completedAt, + sessionResource: snapshot.runs[0].sessionResource?.toString(), + }], + }); + const concurrentWrite = await concurrentStorage.compareAndSwap(archiveKey, runningArchive, completedArchive); + await automationStorage.resumeRepairWrite.complete(); + await timeout(0); + const persistedArchive = JSON.parse((await automationStorage.read(archiveKey))!); + + assert.deepStrictEqual({ + concurrentWrite: concurrentWrite.swapped, + persistedRun: persistedArchive.runs[0], + observedRun: { + ...store.runs.get()[0], + sessionResource: store.runs.get()[0].sessionResource?.toString(), + }, + }, { + concurrentWrite: true, + persistedRun: { + ...snapshot.runs[0], + status: 'completed', + completedAt, + sessionResource: snapshot.runs[0].sessionResource?.toString(), + }, + observedRun: { + ...snapshot.runs[0], + status: 'completed', + completedAt, + sessionResource: snapshot.runs[0].sessionResource?.toString(), + }, + }); + }); + test('repairs malformed legacy run archive rows while preserving valid history', async () => { const connection = new TestAutomationConnection(true); disposables.add(connection); diff --git a/src/vs/workbench/contrib/chat/common/automations/automationService.ts b/src/vs/workbench/contrib/chat/common/automations/automationService.ts index 0a7b38c38abb8..f92f21f536e57 100644 --- a/src/vs/workbench/contrib/chat/common/automations/automationService.ts +++ b/src/vs/workbench/contrib/chat/common/automations/automationService.ts @@ -15,6 +15,21 @@ export const ConfigureAutomationToolReferenceName = 'configureAutomation'; /** Invoked immediately before each storage CAS attempt; throwing aborts before that attempt. */ export type AutomationMutationGuard = () => void; +/** Signals that Automation ownership cannot move while one of its runs is active. */ +export class AutomationActiveRunError extends Error { + constructor( + readonly automationId: string, + readonly runId: string, + ) { + super(`Automation '${automationId}' has active run '${runId}'.`); + } +} + +export function isAutomationActiveRunError(error: unknown): boolean { + return error instanceof AutomationActiveRunError + || (error instanceof AggregateError && error.errors.length > 0 && error.errors.every(isAutomationActiveRunError)); +} + /** * Input for `createAutomation`. The service fills in `id`, timestamps, and * `nextRunAt`. @@ -101,7 +116,7 @@ export interface IUpdateAutomationRunOptions { /** Outcome of an attempt to claim an automation's single active-run slot. */ export interface IAutomationRunClaim { - /** `false` when another run already held the slot, in which case nothing was recorded. */ + /** `false` when another run held the slot or an external authority recorded and dispatched the returned run. */ readonly claimed: boolean; /** The run occupying the slot: the newly recorded one, or the pre-existing one. */ readonly run: IAutomationRun; diff --git a/src/vs/workbench/contrib/chat/common/automations/automationTelemetry.ts b/src/vs/workbench/contrib/chat/common/automations/automationTelemetry.ts index d9898474b71d1..60e2508221d68 100644 --- a/src/vs/workbench/contrib/chat/common/automations/automationTelemetry.ts +++ b/src/vs/workbench/contrib/chat/common/automations/automationTelemetry.ts @@ -153,7 +153,7 @@ export function publishAutomationRunError(telemetryService: ITelemetryService, a } type AutomationMigrationEvent = { - outcome: 'started' | 'completed' | 'failed'; + outcome: 'started' | 'completed' | 'deferred' | 'failed'; discoveredCount: number; migratedCount: number; failedCount: number; @@ -161,7 +161,7 @@ type AutomationMigrationEvent = { }; type AutomationMigrationClassification = { - outcome: { classification: 'SystemMetaData'; purpose: 'PerformanceAndHealth'; comment: 'Whether the migration started, completed, or failed.' }; + outcome: { classification: 'SystemMetaData'; purpose: 'PerformanceAndHealth'; comment: 'Whether the migration started, completed, deferred for an active run, or failed.' }; discoveredCount: { classification: 'SystemMetaData'; purpose: 'PerformanceAndHealth'; isMeasurement: true; comment: 'Number of legacy Automation definitions discovered.' }; migratedCount: { classification: 'SystemMetaData'; purpose: 'PerformanceAndHealth'; isMeasurement: true; comment: 'Number of Automation definitions durably present in the Agent Host catalogue.' }; failedCount: { classification: 'SystemMetaData'; purpose: 'PerformanceAndHealth'; isMeasurement: true; comment: 'Number of Automation definitions that failed migration.' };