fix: wait for active dev rebuilds and watcher cleanup on shutdown

docs/guides/running-locally.md

@@ -46,6 +46,12 @@ buckets, and their data stay up, so the next `prisma-composer dev` is a warm
 start — same ports, same data. `--fresh` is what wipes this app's local
 instances and data before starting.
 
+Shutdown closes the file watchers and waits for any in-flight rebuild or
+deployment before stopping services. A rebuild that has not reached deployment
+is skipped once shutdown begins, so it cannot restart the app after it stops.
+Cleanup failures are reported, but do not prevent the remaining services from
+being stopped or the session from finishing.
+
 `--fresh` is also the fix when a framework upgrade leaves stale rows in this
 app's local dev state — the symptom is a plan-time error naming an
 unregistered resource type (for example

packages/0-framework/3-tooling/cli/src/dev/__tests__/watch.test.ts

@@ -1,7 +1,8 @@
-import { describe, expect, test } from 'bun:test';
+import { describe, expect, spyOn, test } from 'bun:test';
 import * as fs from 'node:fs';
 import * as os from 'node:os';
 import * as path from 'node:path';
+import chokidar, { FSWatcher } from 'chokidar';
 import { startWatch, watchTargetsFrom } from '../watch.ts';
 
 function tempDir(): string {
@@ -29,6 +30,29 @@ describe('watchTargetsFrom()', () => {
 });
 
 describe('startWatch()', () => {
+  test('stop settles synchronous watcher cleanup failures and is idempotent', async () => {
+    const watcher = new FSWatcher();
+    const createWatcher = spyOn(chokidar, 'watch').mockReturnValue(watcher);
+    const closeWatcher = spyOn(watcher, 'close').mockImplementation(() => {
+      throw new Error('watch close failed');
+    });
+    try {
+      const watch = startWatch(
+        [{ address: 'app', paths: [path.join(os.tmpdir(), 'output.js')] }],
+        () => {},
+      );
+      const closing = watch.stop();
+      expect(watch.stop()).toBe(closing);
+      await expect(closing).rejects.toBeInstanceOf(AggregateError);
+      await watch.ready;
+      expect(closeWatcher).toHaveBeenCalledTimes(1);
+    } finally {
+      closeWatcher.mockRestore();
+      createWatcher.mockRestore();
+      await watcher.close();
+    }
+  });
+
   test('debounces a burst of changes across several files into one callback, 300ms after the last change', async () => {
     const dir = tempDir();
     const fileA = path.join(dir, 'a.txt');
@@ -64,7 +88,7 @@ describe('startWatch()', () => {
       await until(() => calls === 1, 2000);
       expect(calls).toBe(1);
     } finally {
-      watch.stop();
+      await watch.stop();
       fs.rmSync(dir, { recursive: true, force: true });
     }
   }, 10_000);
@@ -78,7 +102,7 @@ describe('startWatch()', () => {
     const watch = startWatch([{ address: 'a', paths: [file] }], () => {
       calls += 1;
     });
-    watch.stop();
+    await watch.stop();
 
     fs.writeFileSync(file, 'a2');
     await sleep(500);
@@ -123,7 +147,7 @@ describe('startWatch()', () => {
       await until(() => calls === 2, 3000);
       expect(calls).toBe(2);
     } finally {
-      watch.stop();
+      await watch.stop();
       fs.rmSync(dir, { recursive: true, force: true });
     }
   }, 10_000);

packages/0-framework/3-tooling/cli/src/dev/watch.ts

@@ -49,7 +49,8 @@ export function watchTargetsFrom(bundles: Readonly<Record<string, Bundle>>): {
 export interface WatchHandle {
   /** Resolves once chokidar's OS-level watches are attached — a change made before this can be missed entirely. Also resolves on `stop()` so an awaiting caller can never hang. */
   readonly ready: Promise<void>;
-  stop(): void;
+  /** Awaits every watcher close; rejects with an AggregateError if any close fails. */
+  stop(): Promise<void>;
 }
 
 /**
@@ -72,8 +73,11 @@ export function startWatch(
   onError?: (error: unknown) => void,
 ): WatchHandle {
   let timer: ReturnType<typeof setTimeout> | undefined;
+  let stopped = false;
+  let closing: Promise<void> | undefined;
 
   const trigger = (): void => {
+    if (stopped) return;
     if (timer !== undefined) clearTimeout(timer);
     timer = setTimeout(() => {
       timer = undefined;
@@ -135,9 +139,19 @@ export function startWatch(
   return {
     ready,
     stop: () => {
+      if (closing !== undefined) return closing;
+      stopped = true;
       if (timer !== undefined) clearTimeout(timer);
       markReady();
-      for (const watcher of watchers) void watcher.close();
+      closing = Promise.allSettled(
+        watchers.map((watcher) => Promise.resolve().then(() => watcher.close())),
+      ).then((results) => {
+        const failures = results.flatMap((result) =>
+          result.status === 'rejected' ? [result.reason] : [],
+        );
+        if (failures.length > 0) throw new AggregateError(failures, 'Failed to close dev watchers');
+      });
+      return closing;
     },
   };
 }

packages/0-framework/3-tooling/cli/src/family/commands/dev.ts

@@ -152,7 +152,7 @@ function reportDevEvent(
         report({
           kind: 'message',
           severity: 'warn',
-          text: `A service refused to stop: ${event.message}`,
+          text: `Dev cleanup failed: ${event.message}`,
         });
         return;
 

packages/0-framework/3-tooling/cli/src/operations/__tests__/operations.test.ts

@@ -22,6 +22,7 @@ import type {
 import type { LocalTargetAttachment, LocalTargetDescriptor } from '@internal/core/local-target';
 import * as Layer from 'effect/Layer';
 import { DEPLOYMENT_RESULT_FILE_ENV, type DeploymentSummary } from '../../deployment-summary.ts';
+import * as Watch from '../../dev/watch.ts';
 import type { AppIdentity } from '../../pipeline.ts';
 import type { AlchemyInvocation } from '../../run-alchemy.ts';
 import { deployWithDeps } from '../deploy.ts';
@@ -909,75 +910,175 @@ describe.skipIf(process.platform === 'win32')('dev()', () => {
     expect(stops).toBe(1);
   }, 15_000);
 
-  test('a startServices that throws mid-start is rolled back: the partially-started attachment is stopped again', async () => {
-    const app = makeAppDir('hello-dev');
-    let stops = 0;
-    const attachment: LocalTargetAttachment = {
-      // Models a partial start: some services came up before the throw, so
-      // the rollback must stop this attachment even though startServices
-      // never returned.
-      startServices: () => Promise.reject(new Error('service two failed to bind its port')),
-      stopServices: () => {
-        stops += 1;
-        return Promise.resolve();
-      },
-      endpoints: () => Promise.resolve([]),
-      logs: async function* () {},
-    };
-
-    const result = await silently(() =>
-      devWithDeps(
-        {
-          entry: app.entryPath,
-          cwd: app.dir,
+  test.each(['success', 'sync failure', 'async failure'] as const)(
+    'a partial start is rolled back and preserves its failure after cleanup %s',
+    async (cleanup) => {
+      const app = makeAppDir('hello-dev');
+      let stops = 0;
+      const attachment: LocalTargetAttachment = {
+        // Models a partial start: some services came up before the throw, so
+        // the rollback must stop this attachment even though startServices
+        // never returned.
+        startServices: () => Promise.reject(new Error('service two failed to bind its port')),
+        stopServices: () => {
+          stops += 1;
+          if (cleanup === 'sync failure') throw new Error('cleanup failed');
+          if (cleanup === 'async failure') return Promise.reject(new Error('cleanup failed'));
+          return Promise.resolve();
         },
-        {
-          config: devConfigWith(attachment),
-          runAssembler: fakeAssembler,
-          alchemy: async () => ({ exitCode: 0, signal: null }),
+        endpoints: () => Promise.resolve([]),
+        logs: async function* () {},
+      };
+
+      const result = await silently(() =>
+        devWithDeps(
+          {
+            entry: app.entryPath,
+            cwd: app.dir,
+          },
+          {
+            config: devConfigWith(attachment),
+            runAssembler: fakeAssembler,
+            alchemy: async () => ({ exitCode: 0, signal: null }),
+          },
+        ),
+      );
+
+      expect(result.ok).toBe(false);
+      if (result.ok) throw new Error('unreachable');
+      expect(result.failure.code).toBe('DEV.SERVICE_START_FAILED');
+      expect(result.failure.message).toBe('service two failed to bind its port');
+      expect(stops).toBe(1);
+    },
+    15_000,
+  );
+
+  test.each(['synchronous', 'asynchronous'] as const)(
+    'stop() surfaces a %s cleanup failure and still finishes',
+    async (failure) => {
+      const app = makeAppDir('hello-dev');
+      const attachment: LocalTargetAttachment = {
+        startServices: () => Promise.resolve(),
+        stopServices: () => {
+          const error = new Error('service pid 123 will not die');
+          if (failure === 'synchronous') throw error;
+          return Promise.reject(error);
         },
-      ),
-    );
+        endpoints: () => Promise.resolve([]),
+        logs: async function* () {},
+      };
+      const events: string[] = [];
 
-    expect(result.ok).toBe(false);
-    if (result.ok) throw new Error('unreachable');
-    expect(result.failure.code).toBe('DEV.SERVICE_START_FAILED');
-    expect(result.failure.message).toBe('service two failed to bind its port');
-    expect(stops).toBe(1);
-  }, 15_000);
+      const result = await silently(async () => {
+        const start = await devWithDeps(
+          {
+            entry: app.entryPath,
+            cwd: app.dir,
+            onEvent: (event) => void events.push(event.kind),
+          },
+          {
+            config: devConfigWith(attachment),
+            runAssembler: fakeAssembler,
+            alchemy: async () => ({ exitCode: 0, signal: null }),
+          },
+        );
+        if (!start.ok) throw new Error('expected a started session');
+        await start.value.stop();
+        await start.value.closed;
+        return start;
+      });
 
-  test('stop() surfaces a service that refuses to stop as a stop-error event, and still finishes', async () => {
-    const app = makeAppDir('hello-dev');
-    const attachment: LocalTargetAttachment = {
-      startServices: () => Promise.resolve(),
-      stopServices: () => Promise.reject(new Error('service pid 123 will not die')),
-      endpoints: () => Promise.resolve([]),
-      logs: async function* () {},
-    };
-    const events: string[] = [];
+      expect(result.ok).toBe(true);
+      expect(events).toEqual(['ready', 'unwatchable', 'stopping', 'stop-error', 'stopped']);
+    },
+    15_000,
+  );
 
-    const result = await silently(async () => {
+  test.each(['assembly', 'converge'] as const)(
+    'stop() waits for an active %s and prevents a late restart',
+    async (phase) => {
+      const app = makeAppDir('shutdown-race');
+      const watched = path.join(app.dir, 'output.txt');
+      let triggerChange = () => {};
+      const watch = spyOn(Watch, 'startWatch').mockImplementation((_targets, onChange) => {
+        triggerChange = onChange;
+        return { ready: Promise.resolve(), stop: async () => {} };
+      });
+      let resume = () => {};
+      const blockedWork = new Promise<void>((resolve) => {
+        resume = resolve;
+      });
+      let entered = false;
+      let rebuildFailure: string | undefined;
+      let assemblies = 0;
+      let converges = 0;
+      let stops = 0;
+      const events: string[] = [];
+      const attachment: LocalTargetAttachment = {
+        startServices: async () => {},
+        stopServices: async () => {
+          stops += 1;
+        },
+        endpoints: async () => [],
+        logs: async function* () {},
+      };
       const start = await devWithDeps(
         {
           entry: app.entryPath,
           cwd: app.dir,
-          onEvent: (event) => void events.push(event.kind),
+          onEvent: (event) => {
+            events.push(event.kind);
+            if (event.kind === 'rebuild-failed') rebuildFailure = event.message;
+          },
         },
         {
           config: devConfigWith(attachment),
-          runAssembler: fakeAssembler,
-          alchemy: async () => ({ exitCode: 0, signal: null }),
+          runAssembler: async (node) => {
+            assemblies += 1;
+            if (assemblies === 2 && phase === 'assembly') {
+              entered = true;
+              await blockedWork;
+            }
+            return { ...(await fakeAssembler(node)), watch: [watched] };
+          },
+          alchemy: async () => {
+            converges += 1;
+            if (converges === 2 && phase === 'converge') {
+              entered = true;
+              await blockedWork;
+            }
+            return { exitCode: 0, signal: null };
+          },
         },
-      );
+      ).finally(() => watch.mockRestore());
       if (!start.ok) throw new Error('expected a started session');
-      await start.value.stop();
-      await start.value.closed;
-      return start;
-    });
-
-    expect(result.ok).toBe(true);
-    expect(events).toEqual(['ready', 'unwatchable', 'stopping', 'stop-error', 'stopped']);
-  }, 15_000);
+      try {
+        triggerChange();
+        const deadline = Date.now() + 5000;
+        while (!entered && rebuildFailure === undefined && Date.now() < deadline) {
+          await new Promise((resolve) => setTimeout(resolve, 25));
+        }
+        expect(rebuildFailure).toBeUndefined();
+        expect(entered).toBe(true);
+        let stopped = false;
+        const closing = start.value.stop().then(() => {
+          stopped = true;
+        });
+        await Promise.resolve();
+        expect(stopped).toBe(false);
+        expect(stops).toBe(0);
+        resume();
+        await closing;
+        expect(stops).toBe(1);
+        expect(converges).toBe(phase === 'assembly' ? 1 : 2);
+        expect(events).toEqual(['ready', 'stopping', 'stopped']);
+      } finally {
+        resume();
+        await start.value.stop();
+      }
+    },
+    15_000,
+  );
 
   test('the DevSession contract: closed settles only via stop(), stop() is idempotent, and no process signal handler is ever registered', async () => {
     const app = makeAppDir('hello-dev');

packages/0-framework/3-tooling/cli/src/operations/dev.ts

@@ -2,7 +2,7 @@
  * The programmatic `dev` operation (`@prisma/composer/control`): typed input,
  * events out through `onEvent`, lifetime owned by the returned DevSession —
  * no argv, no console, no process.exit, and NEVER any process signal
- * handling (the host owns signals; the CLI adapter dev/run-dev.ts shows the
+ * handling (the host owns signals; the CLI adapter family/commands/dev.ts shows the
  * pattern). The executor loads lazily, so importing this module executes
  * nothing; an executor that fails to load comes back as a structured
  * failure, never a throw out of the host.
@@ -26,7 +26,7 @@ export type DevEvent =
       readonly cwd: string;
     }
   | { readonly kind: 'stopping' }
-  /** One service refused to stop during stop(); teardown continues and `stopped` still follows. */
+  /** Watcher or service cleanup failed during stop(); teardown continues and `stopped` still follows. */
   | { readonly kind: 'stop-error'; readonly message: string }
   | { readonly kind: 'stopped' };
 
@@ -38,9 +38,7 @@ export interface DevInput {
   readonly onEvent?: ((event: DevEvent) => void) | undefined;
 }
 
-/** A running dev session. The operation NEVER touches process signal handlers —
- * the host owns signals (and must evict alchemy's import-time SIGINT/SIGTERM
- * listeners before installing its own; see run-dev.ts). */
+/** A running dev session. The host owns process signals and calls stop() to shut down. */
 export interface DevSession {
   /** The initial front door, already merged across attachments. */
   readonly endpoints: readonly ServiceEndpoint[];