fix(core): re-read models.dev catalog after subscribing to Refreshed (#50012)

packages/core/src/plugin/models-dev.ts

@@ -57,16 +57,19 @@ export const ModelsDevPlugin = define({
         })
       }
     })
+    const apply = (data: readonly ModelsDev.Snapshot[]) => {
+      loaded.data = snapshots(data)
+      return ctx.integration.reload().pipe(Effect.andThen(ctx.provider.reload()))
+    }
     yield* bus.subscribe(ModelsDev.Event.Refreshed).pipe(
-      Stream.runForEach(() =>
-        modelsDev.get().pipe(
-          Effect.tap((data) => Effect.sync(() => (loaded.data = snapshots(data)))),
-          Effect.andThen(ctx.integration.reload()),
-          Effect.andThen(ctx.provider.reload()),
-        ),
-      ),
+      Stream.runForEach(() => modelsDev.get().pipe(Effect.flatMap(apply))),
       Effect.forkScoped({ startImmediately: true }),
     )
+    // A refresh that landed between the initial read and the subscription above published
+    // Refreshed to nobody here. On a cold cache that read served the bundled snapshot, so
+    // re-read now instead of waiting for the next TTL refresh.
+    const latest = yield* modelsDev.get()
+    if (snapshots(latest) !== loaded.data) yield* apply(latest)
   }),
 })
 

packages/core/test/plugin/models-dev.test.ts

@@ -290,6 +290,37 @@ describe("ModelsDevPlugin", () => {
     }),
   )
 
+  isolated.effect("adopts a refresh that completes between its initial read and its subscription", () =>
+    Effect.gen(function* () {
+      const bundled = richSnapshot("Acme Bundled")
+      const fresh = richSnapshot("Acme Fresh")
+      const current = { snapshot: bundled.snapshot }
+      const location = yield* owner
+      // Cold cache: the first read serves the bundled snapshot, and the boot-time
+      // ModelsDev.refresh() lands right after it, before the plugin subscribes.
+      const source = ModelsDev.Service.of({
+        get: () =>
+          Effect.gen(function* () {
+            const data = current.snapshot
+            if (data !== bundled.snapshot) return data
+            current.snapshot = fresh.snapshot
+            yield* location.bus.publish(ModelsDev.Event.Refreshed, {})
+            return data
+          }),
+        refresh: () => Effect.void,
+      })
+      yield* ModelsDevPlugin.effect(location.host).pipe(
+        Effect.provideService(ModelsDev.Service, source),
+        Effect.provideContext(location.context),
+      )
+      yield* TestClock.adjust("500 millis")
+      yield* TestClock.adjust("500 millis")
+      yield* TestClock.adjust("500 millis")
+
+      expect(required(yield* location.providers.get(bundled.providerID)).name).toBe("Acme Fresh")
+    }),
+  )
+
   real.effect("keeps the retained definition unchanged across model replay", () =>
     Effect.gen(function* () {
       const providers = yield* Provider.Service