diff --git a/src/cache.ts b/src/cache.ts
index 87a7902bc84..aabec5db15b 100644
--- a/src/cache.ts
+++ b/src/cache.ts
@@ -49,6 +49,8 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
     private readonly callbackCache: { [key: string]: (ObjectCallback<T> | ErrorCallback)[] } = {};
     private request: AbortController | undefined;
     private stopped: boolean = false;
+    // Stop/start invalidates pending work and callbacks from the previous run.
+    private generation: number = 0;
     private reconnectDelayMs: number = 0;
     private lastBackoffAt: number | undefined;
     private hasConnected: boolean = false;
@@ -92,6 +94,7 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
     }
 
     public async start(): Promise<void> {
+        this.generation++;
         this.stopped = false;
         this.reconnectDelayMs = 0;
         this.lastBackoffAt = undefined;
@@ -100,6 +103,7 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
     }
 
     public async stop(): Promise<void> {
+        this.generation++;
         this.stopped = true;
         this._stop();
     }
@@ -207,19 +211,26 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
         return statusCode === undefined || statusCode === 408 || statusCode === 429 || statusCode >= 500;
     }
 
-    private async waitForRetry(): Promise<boolean> {
+    private isCurrentGeneration(generation: number): boolean {
+        return !this.stopped && generation === this.generation;
+    }
+
+    private async waitForRetry(generation: number): Promise<boolean> {
         if (this.reconnectDelayMs === 0) {
             this.reconnectDelayMs = this.nextBackoffLevelMs();
         }
         await this.delayFn(this.withJitter(this.reconnectDelayMs));
-        if (this.stopped) {
+        if (!this.isCurrentGeneration(generation)) {
             return false;
         }
         this.reconnectDelayMs = this.nextBackoffLevelMs();
         return true;
     }
 
-    private async doneHandler(err: any): Promise<void> {
+    private async doneHandler(err: any, generation: number = this.generation): Promise<void> {
+        if (!this.isCurrentGeneration(generation)) {
+            return;
+        }
         this._stop();
         let hasBackedOff = false;
         if (
@@ -235,16 +246,23 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
             this.reconnectDelayMs = 0;
         } else if (err) {
             this.callbackCache[ERROR].forEach((elt: ErrorCallback) => elt(err));
-            if (!this.isRetryableError(err) || this.stopped || !(await this.waitForRetry())) {
+            if (
+                !this.isRetryableError(err) ||
+                !this.isCurrentGeneration(generation) ||
+                !(await this.waitForRetry(generation))
+            ) {
                 return;
             }
             hasBackedOff = true;
         }
-        if (this.stopped) {
+        if (!this.isCurrentGeneration(generation)) {
             // do not auto-restart
             return;
         }
         this.callbackCache[CONNECT].forEach((elt: ErrorCallback) => elt(undefined));
+        if (!this.isCurrentGeneration(generation)) {
+            return;
+        }
         if (!this.resourceVersion) {
             let listed = false;
             while (!listed) {
@@ -253,13 +271,23 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
                     const promise = this.listFn();
                     list = await promise;
                 } catch (err) {
+                    if (!this.isCurrentGeneration(generation)) {
+                        return;
+                    }
                     this.callbackCache[ERROR].forEach((elt: ErrorCallback) => elt(err));
-                    if (!this.isRetryableError(err) || this.stopped || !(await this.waitForRetry())) {
+                    if (
+                        !this.isRetryableError(err) ||
+                        !this.isCurrentGeneration(generation) ||
+                        !(await this.waitForRetry(generation))
+                    ) {
                         return;
                     }
                     hasBackedOff = true;
                     continue;
                 }
+                if (!this.isCurrentGeneration(generation)) {
+                    return;
+                }
                 this.objects = deleteItems(this.objects, list.items, this.callbackCache[DELETE].slice());
                 this.addOrUpdateItems(list.items);
                 this.resourceVersion = list.metadata ? list.metadata!.resourceVersion || '' : '';
@@ -283,18 +311,30 @@ export class ListWatch<T extends KubernetesObject> implements ObjectCache<T>, In
             if (this.reconnectDelayMs > 0) {
                 await this.delayFn(this.withJitter(this.reconnectDelayMs));
             }
-            if (this.stopped) {
+            if (!this.isCurrentGeneration(generation)) {
                 return;
             }
             this.reconnectDelayMs = this.nextBackoffLevelMs();
         }
+        if (!this.isCurrentGeneration(generation)) {
+            return;
+        }
         this.hasConnected = true;
-        this.request = await this.watch.watch(
+        const request = await this.watch.watch(
             this.path,
             queryParams,
-            this.watchHandler.bind(this),
-            this.doneHandler.bind(this),
+            (phase, obj, watchObj) => {
+                if (this.isCurrentGeneration(generation)) {
+                    return this.watchHandler(phase, obj, watchObj);
+                }
+            },
+            (err) => this.doneHandler(err, generation),
         );
+        if (!this.isCurrentGeneration(generation)) {
+            request.abort();
+            return;
+        }
+        this.request = request;
     }
 
     private addOrUpdateItems(items: T[]): void {
diff --git a/src/cache_test.ts b/src/cache_test.ts
index da689afcc50..5b2f890e412 100644
--- a/src/cache_test.ts
+++ b/src/cache_test.ts
@@ -1,4 +1,4 @@
-import { describe, it } from 'node:test';
+import { describe, it, type TestContext } from 'node:test';
 import { deepStrictEqual, notStrictEqual, ok, strictEqual, throws } from 'node:assert';
 import mock from 'ts-mockito';
 
@@ -1076,21 +1076,17 @@ describe('ListWatchCache', () => {
                 resolve(listObj);
             });
         };
-        const promise = new Promise((resolve) => {
-            mock.when(
-                fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
-            ).thenCall(() => {
-                resolve(new AbortController());
-            });
-        });
+        mock.when(
+            fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
+        ).thenCall(async () => new AbortController());
 
-        const cache = new ListWatch('/some/path', mock.instance(fakeWatch), listFn);
-        await promise;
+        const cache = new ListWatch('/some/path', mock.instance(fakeWatch), listFn, false);
+        await cache.start();
 
         const [, , , doneHandler] = mock.capture(fakeWatch.watch).last();
 
         // stop the informer
-        cache.stop();
+        await cache.stop();
 
         await doneHandler(null);
 
@@ -1134,16 +1130,12 @@ describe('ListWatchCache', () => {
                 resolve(listObj);
             });
         };
-        const promise = new Promise((resolve) => {
-            mock.when(
-                fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
-            ).thenCall(() => {
-                resolve(new AbortController());
-            });
-        });
+        mock.when(
+            fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
+        ).thenCall(async () => new AbortController());
 
-        const cache = new ListWatch('/some/path', mock.instance(fakeWatch), listFn);
-        await promise;
+        const cache = new ListWatch('/some/path', mock.instance(fakeWatch), listFn, false);
+        await cache.start();
 
         let errorEmitted = false;
         cache.on('error', () => (errorEmitted = true));
@@ -1176,17 +1168,12 @@ describe('ListWatchCache', () => {
                 resolve(listObj);
             });
         };
-        let promise = new Promise((resolve) => {
-            mock.when(
-                fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
-            ).thenCall(() => {
-                resolve(new AbortController());
-            });
-        });
+        mock.when(
+            fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
+        ).thenCall(async () => new AbortController());
         const informer = new ListWatch('/some/path', mock.instance(fakeWatch), listFn, false);
 
-        informer.start();
-        await promise;
+        await informer.start();
 
         const [, , watchHandler] = mock.capture(fakeWatch.watch).last();
         watchHandler(
@@ -1201,15 +1188,7 @@ describe('ListWatchCache', () => {
         );
 
         await informer.stop();
-        promise = new Promise((resolve) => {
-            mock.when(
-                fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
-            ).thenCall(() => {
-                resolve(new AbortController());
-            });
-        });
-        informer.start();
-        await promise;
+        await informer.start();
         strictEqual(listCalls, 1);
     });
 
@@ -1230,18 +1209,13 @@ describe('ListWatchCache', () => {
                 resolve(listObj);
             });
         };
-        let promise = new Promise((resolve) => {
-            mock.when(
-                fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
-            ).thenCall(() => {
-                resolve(new AbortController());
-            });
-        });
+        mock.when(
+            fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
+        ).thenCall(async () => new AbortController());
 
         const informer = new ListWatch('/some/path', mock.instance(fakeWatch), listFn, false);
 
-        informer.start();
-        await promise;
+        await informer.start();
 
         const [, , watchHandler] = mock.capture(fakeWatch.watch).last();
         watchHandler(
@@ -1260,18 +1234,9 @@ describe('ListWatchCache', () => {
         let errorEmitted = false;
         informer.on('error', () => (errorEmitted = true));
 
-        promise = new Promise((resolve) => {
-            mock.when(
-                fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
-            ).thenCall(() => {
-                resolve({});
-            });
-        });
-
-        informer.start();
-        await promise;
+        await informer.start();
 
-        const [, , , doneHandler] = mock.capture(fakeWatch.watch).last();
+        const [, , watchHandler2, doneHandler] = mock.capture(fakeWatch.watch).last();
 
         const object = {
             kind: 'Status',
@@ -1282,7 +1247,7 @@ describe('ListWatchCache', () => {
             reason: 'Expired',
             code: 410,
         };
-        await watchHandler('ERROR', object, { type: 'ERROR', object });
+        await watchHandler2('ERROR', object, { type: 'ERROR', object });
         await doneHandler(null);
 
         mock.verify(
@@ -1307,18 +1272,13 @@ describe('ListWatchCache', () => {
                 } as V1NamespaceList);
             });
         };
-        let promise = new Promise((resolve) => {
-            mock.when(
-                fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
-            ).thenCall(() => {
-                resolve({});
-            });
-        });
+        mock.when(
+            fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
+        ).thenCall(async () => new AbortController());
 
         const informer = new ListWatch('/some/path', mock.instance(fakeWatch), listFn, false);
 
-        informer.start();
-        await promise;
+        await informer.start();
 
         const [, , watchHandler] = mock.capture(fakeWatch.watch).last();
         watchHandler(
@@ -1337,16 +1297,7 @@ describe('ListWatchCache', () => {
         let errorEmitted = false;
         informer.on('error', () => (errorEmitted = true));
 
-        promise = new Promise((resolve) => {
-            mock.when(
-                fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
-            ).thenCall(() => {
-                resolve(new AbortController());
-            });
-        });
-
-        informer.start();
-        await promise;
+        await informer.start();
 
         const [, , , doneHandler] = mock.capture(fakeWatch.watch).last();
 
@@ -1378,18 +1329,13 @@ describe('ListWatchCache', () => {
                 resolve(listObj);
             });
         };
-        let promise = new Promise((resolve) => {
-            mock.when(
-                fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
-            ).thenCall(() => {
-                resolve(new AbortController());
-            });
-        });
+        mock.when(
+            fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
+        ).thenCall(async () => new AbortController());
 
         const informer = new ListWatch('/some/path', mock.instance(fakeWatch), listFn, false);
 
-        informer.start();
-        await promise;
+        await informer.start();
 
         const [, , watchHandler] = mock.capture(fakeWatch.watch).last();
         watchHandler(
@@ -1409,25 +1355,16 @@ describe('ListWatchCache', () => {
         let errorEmitted = false;
         informer.on('error', () => (errorEmitted = true));
 
-        promise = new Promise((resolve) => {
-            mock.when(
-                fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
-            ).thenCall(() => {
-                resolve(new AbortController());
-            });
-        });
-
-        informer.start();
-        await promise;
+        await informer.start();
 
         const [, , watchHandler2, doneHandler] = mock.capture(fakeWatch.watch).last();
         watchHandler2('ERROR', {
             code: 410,
         });
-        doneHandler(undefined);
+        await doneHandler(undefined);
         mock.verify(
             fakeWatch.watch(mock.anything(), mock.anything(), mock.anything(), mock.anything()),
-        ).twice();
+        ).thrice();
         strictEqual(errorEmitted, false);
         strictEqual(listCalls, 2);
     });
@@ -1779,6 +1716,147 @@ describe('ListWatchCache', () => {
         deepStrictEqual(errors, [error]);
     });
 
+    function deferred<T = void>() {
+        let resolve!: (value: T | PromiseLike<T>) => void;
+        const promise = new Promise<T>((done) => (resolve = done));
+        return { promise, resolve };
+    }
+
+    function lifecycleCache(
+        t: TestContext,
+        options: {
+            list?: ListPromise<V1Namespace>;
+            delay?: () => Promise<void>;
+            connect?: (controller: AbortController) => Promise<AbortController>;
+        } = {},
+    ) {
+        const watches: {
+            controller: AbortController;
+            event: Parameters<Watch['watch']>[2];
+            done: Parameters<Watch['watch']>[3];
+        }[] = [];
+        const watch = new Watch(new KubeConfig());
+        t.mock.method(watch, 'watch', async (_path, _query, event, done) => {
+            const controller = new AbortController();
+            watches.push({ controller, event, done });
+            return options.connect ? options.connect(controller) : controller;
+        });
+        const informer = new ListWatch<V1Namespace>(
+            '/some/path',
+            watch,
+            options.list ?? (async () => ({ metadata: { resourceVersion: '12345' }, items: [] })),
+            false,
+            undefined,
+            undefined,
+            { delayFn: options.delay ?? (() => Promise.resolve()), randFn: () => 0 },
+        );
+        t.after(() => informer.stop());
+        return { informer, watches };
+    }
+
+    it('should discard a pending retry after stopping and starting again', async (t) => {
+        const retryGate = deferred();
+        const retryEntered = deferred();
+        const { informer, watches } = lifecycleCache(t, {
+            delay: () => {
+                retryEntered.resolve();
+                return retryGate.promise;
+            },
+        });
+        await informer.start();
+        const oldWatch = watches[0];
+        const retry = oldWatch.done({ statusCode: 503 });
+        await retryEntered.promise;
+
+        await informer.stop();
+        await informer.start();
+
+        // The old retry must not see the new run's stopped=false as permission to reconnect.
+        retryGate.resolve();
+        await retry;
+        strictEqual(watches.length, 2);
+        strictEqual(oldWatch.controller.signal.aborted, true);
+        strictEqual(watches[1].controller.signal.aborted, false);
+
+        // Late callbacks must not mutate or stop the new run either.
+        oldWatch.event('ADDED', { metadata: { name: 'stale-event', resourceVersion: '99999' } });
+        await oldWatch.done(null);
+        strictEqual(informer.get('stale-event'), undefined);
+        strictEqual(watches[1].controller.signal.aborted, false);
+        await informer.stop();
+        ok(watches.every(({ controller }) => controller.signal.aborted));
+    });
+
+    it('should discard a retried relist that completes after stopping', async (t) => {
+        const pendingList = deferred<V1NamespaceList>();
+        const listEntered = deferred();
+        let listCalls = 0;
+        const { informer, watches } = lifecycleCache(t, {
+            list: async () => {
+                listCalls++;
+                if (listCalls === 1) {
+                    return { metadata: { resourceVersion: '12345' }, items: [] };
+                }
+                if (listCalls === 2) {
+                    throw Object.assign(new Error('Service Unavailable'), { statusCode: 503 });
+                }
+                listEntered.resolve();
+                return pendingList.promise;
+            },
+        });
+        const changes: V1Namespace[] = [];
+        informer.on('change', (obj) => changes.push(obj));
+        await informer.start();
+        const reconnecting = watches[0].done({ statusCode: 410 });
+        await listEntered.promise;
+        await informer.stop();
+
+        // This successful LIST follows a retry, which used to bypass the final stopped check.
+        pendingList.resolve({
+            metadata: { resourceVersion: '23456' },
+            items: [{ metadata: { name: 'late-result', resourceVersion: '23456' } }],
+        });
+        await reconnecting;
+        strictEqual(listCalls, 3);
+        strictEqual(watches.length, 1);
+        strictEqual(watches[0].controller.signal.aborted, true);
+        strictEqual(informer.get('late-result'), undefined);
+        deepStrictEqual(changes, []);
+    });
+
+    for (const restart of [false, true]) {
+        it(`should abort a pending watch that connects after ${restart ? 'restarting' : 'stopping'}`, async (t) => {
+            const pendingWatch = deferred<AbortController>();
+            const watchEntered = deferred();
+            const { informer, watches } = lifecycleCache(t, {
+                connect: async (controller) => {
+                    if (watches.length === 1) {
+                        watchEntered.resolve();
+                        return pendingWatch.promise;
+                    }
+                    return controller;
+                },
+            });
+            const starting = informer.start();
+            await watchEntered.promise;
+            await informer.stop();
+            if (restart) {
+                await informer.start();
+            }
+
+            // stop() could not abort this controller because watch() had not returned it yet.
+            pendingWatch.resolve(watches[0].controller);
+            await starting;
+            strictEqual(watches[0].controller.signal.aborted, true);
+            strictEqual(watches.length, restart ? 2 : 1);
+            if (restart) {
+                strictEqual(watches[1].controller.signal.aborted, false);
+                await informer.stop();
+                strictEqual(watches[1].controller.signal.aborted, true);
+            }
+        });
+    }
+
     it('should not back off between repeated TimeoutErrors', async () => {
         const fakeWatch = mock.mock(Watch);
         const listObj = {