fix(tui): drop unavailable saved variants

packages/core/src/session/runner/publish-llm-event.ts

@@ -150,6 +150,10 @@ export const createLLMEventPublisher = (bus: Pick<Bus.Interface, "publish">, inp
       if (!current) return yield* Effect.die(new Error(`${name} delta before start: ${id}`))
       if (!current.pending) return undefined
       const now = yield* Clock.currentTimeMillis
+      if (!force && current.publishedAt === undefined) {
+        current.publishedAt = now
+        return undefined
+      }
       if (!force && current.publishedAt !== undefined && now - current.publishedAt < deltaBatchInterval)
         return undefined
       yield* delta(id, current.pending, current.ordinal)

packages/core/test/plugin/promise.test.ts

@@ -1,8 +1,7 @@
 import { describe, expect } from "bun:test"
 import { Message, SystemPart } from "@opencode-ai/ai"
-import { DateTime, Deferred, Effect, Schema, Stream } from "effect"
+import { DateTime, Effect, Schema } from "effect"
 import { Agent } from "@opencode-ai/core/agent"
-import { Bus } from "@opencode-ai/core/bus"
 import { Catalog } from "@opencode-ai/core/catalog"
 import { Model } from "@opencode-ai/core/model"
 import { Location } from "@opencode-ai/core/location"
@@ -19,7 +18,6 @@ import { Provider } from "@opencode-ai/core/provider"
 import { Project } from "@opencode-ai/core/project"
 import { AbsolutePath } from "@opencode-ai/core/schema"
 import { define } from "@opencode-ai/plugin/promise/plugin"
-import { Config as ConfigSchema } from "@opencode-ai/schema/config"
 import { Money } from "@opencode-ai/schema/money"
 import type { SessionHooks } from "@opencode-ai/plugin/effect/session"
 import { testEffect } from "../lib/effect"
@@ -439,170 +437,6 @@ describe("fromPromise", () => {
     }),
   )
 
-  it.effect("closes a pending event iterator with the plugin scope", () =>
-    Effect.gen(function* () {
-      const started = yield* Deferred.make<void>()
-      let finalized = 0
-      let iterator: AsyncIterator<unknown> | undefined
-      let pending: Promise<IteratorResult<unknown>> | undefined
-      const host = testHost({
-        event: {
-          subscribe: () =>
-            Stream.fromEffect(Deferred.succeed(started, undefined)).pipe(
-              Stream.flatMap(() => Stream.never),
-              Stream.ensuring(Effect.sync(() => finalized++)),
-            ),
-        },
-      })
-
-      yield* Effect.scoped(
-        PluginPromise.fromPromise(
-          define({
-            id: "promise-event-pending",
-            setup: async (ctx) => {
-              iterator = ctx.event.subscribe()[Symbol.asyncIterator]()
-              pending = iterator.next()
-              await Effect.runPromise(Deferred.await(started))
-            },
-          }),
-        ).effect(host),
-      )
-
-      expect(finalized).toBe(1)
-      if (!pending || !iterator) yield* Effect.die("event iterator was not initialized")
-      expect(yield* Effect.promise(() => pending)).toEqual({ done: true, value: undefined })
-      expect(yield* Effect.promise(() => iterator.next())).toEqual({ done: true, value: undefined })
-    }),
-  )
-
-  it.live("closes event iterators on break, completion, and failure", () =>
-    Effect.gen(function* () {
-      const bus = yield* Bus.Service
-      const plugins = yield* Plugin.Service
-      const closed: string[] = []
-      const broke = yield* Deferred.make<void>()
-      const promisePlugin = define({
-        id: "promise-event-terminal",
-        setup: async (ctx) => {
-          void (async () => {
-            for await (const _event of ctx.event.subscribe()) break
-            await Effect.runPromise(Deferred.succeed(broke, undefined))
-          })()
-        },
-      })
-
-      yield* plugins.activate([{ ...PluginPromise.fromPromise(promisePlugin), version: "1" }])
-      yield* Effect.sleep("10 millis")
-      yield* bus.publish(ConfigSchema.Event.Updated, {})
-      yield* Deferred.await(broke)
-
-      yield* Effect.scoped(
-        PluginPromise.fromPromise(
-          define({
-            id: "promise-event-complete",
-            setup: async (ctx) => {
-              const iterator = ctx.event.subscribe()[Symbol.asyncIterator]()
-              expect(await iterator.next()).toEqual({ done: true, value: undefined })
-              expect(await iterator.next()).toEqual({ done: true, value: undefined })
-            },
-          }),
-        ).effect(
-          testHost({
-            event: {
-              subscribe: () => Stream.empty.pipe(Stream.ensuring(Effect.sync(() => closed.push("complete")))),
-            },
-          }),
-        ),
-      )
-
-      yield* Effect.scoped(
-        PluginPromise.fromPromise(
-          define({
-            id: "promise-event-failure",
-            setup: async (ctx) => {
-              const iterator = ctx.event.subscribe()[Symbol.asyncIterator]()
-              await expect(iterator.next()).rejects.toThrow("event failure")
-              expect(await iterator.next()).toEqual({ done: true, value: undefined })
-            },
-          }),
-        ).effect(
-          testHost({
-            event: {
-              subscribe: () =>
-                Stream.fail(new Error("event failure")).pipe(
-                  Stream.ensuring(Effect.sync(() => closed.push("failure"))),
-                ),
-            },
-          }),
-        ),
-      )
-
-      expect(closed).toEqual(["complete", "failure"])
-    }),
-  )
-
-  it.effect("closes every event iterator when the plugin scope closes", () =>
-    Effect.gen(function* () {
-      const ready = yield* Deferred.make<void>()
-      let started = 0
-      let finalized = 0
-      const host = testHost({
-        event: {
-          subscribe: () =>
-            Stream.fromEffect(
-              Effect.sync(() => ++started).pipe(
-                Effect.tap((count) => (count === 3 ? Deferred.succeed(ready, undefined) : Effect.void)),
-              ),
-            ).pipe(
-              Stream.flatMap(() => Stream.never),
-              Stream.ensuring(Effect.sync(() => finalized++)),
-            ),
-        },
-      })
-
-      yield* Effect.scoped(
-        PluginPromise.fromPromise(
-          define({
-            id: "promise-event-multiple",
-            setup: async (ctx) => {
-              const events = ctx.event.subscribe()
-              void events[Symbol.asyncIterator]().next()
-              void events[Symbol.asyncIterator]().next()
-              void ctx.event.subscribe()[Symbol.asyncIterator]().next()
-              await Effect.runPromise(Deferred.await(ready))
-            },
-          }),
-        ).effect(host),
-      )
-
-      expect(finalized).toBe(3)
-    }),
-  )
-
-  it.effect("closes a Promise event iterator when the plugin is replaced", () =>
-    Effect.gen(function* () {
-      const plugins = yield* Plugin.Service
-      let iterator: AsyncIterator<unknown> | undefined
-      let pending: Promise<IteratorResult<unknown>> | undefined
-      const previous = PluginPromise.fromPromise(
-        define({
-          id: "promise-event-replacement",
-          setup: async (ctx) => {
-            iterator = ctx.event.subscribe()[Symbol.asyncIterator]()
-            pending = iterator.next()
-          },
-        }),
-      )
-
-      yield* plugins.activate([{ ...previous, version: "1" }])
-      yield* plugins.activate([{ id: previous.id, version: "2", effect: () => Effect.void }])
-
-      if (!pending || !iterator) yield* Effect.die("event iterator was not initialized")
-      expect(yield* Effect.promise(() => pending)).toEqual({ done: true, value: undefined })
-      expect(yield* Effect.promise(() => iterator.next())).toEqual({ done: true, value: undefined })
-    }),
-  )
-
   it.effect("constructs plain Promise tool definitions in the host", () =>
     Effect.gen(function* () {
       const plugins = yield* Plugin.Service

packages/core/test/session-runner-tool-events.test.ts

@@ -217,16 +217,13 @@ it.effect("batches text deltas and flushes pending text before the terminal even
       { discard: true },
     )
 
-    expect(published.filter((event) => event.type === "session.text.delta").map((event) => event.data)).toMatchObject([
-      { delta: "one" },
-    ])
+    expect(published.filter((event) => event.type === "session.text.delta")).toHaveLength(0)
     yield* TestClock.adjust("99 millis")
-    expect(published.filter((event) => event.type === "session.text.delta")).toHaveLength(1)
+    expect(published.filter((event) => event.type === "session.text.delta")).toHaveLength(0)
     yield* TestClock.adjust("1 millis")
     yield* publisher.publish(LLMEvent.textDelta({ id: "text", text: " four" }))
     expect(published.filter((event) => event.type === "session.text.delta").map((event) => event.data)).toMatchObject([
-      { delta: "one" },
-      { delta: " two three four" },
+      { delta: "one two three four" },
     ])
 
     yield* publisher.publish(LLMEvent.textDelta({ id: "text", text: " five" }))
@@ -253,7 +250,7 @@ it.effect("batches reasoning deltas and flushes pending reasoning before the ter
 
     expect(
       published.filter((event) => event.type === "session.reasoning.delta").map((event) => event.data),
-    ).toMatchObject([{ delta: "one" }, { delta: " two three" }])
+    ).toMatchObject([{ delta: "one two three" }])
     expect(published.slice(-2).map((event) => event.type)).toEqual([
       "session.reasoning.delta",
       "session.reasoning.ended.1",

packages/core/test/session-runner.test.ts

@@ -767,7 +767,7 @@ const verifyEphemeralDeltas = (kind: FragmentKind) =>
     yield* admit(session, prompt)
     const bus = yield* Bus.Service
     const live = fixture.delta
-      ? yield* bus.subscribe(fixture.delta).pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped)
+      ? yield* bus.subscribe(fixture.delta).pipe(Stream.take(1), Stream.runCollect, Effect.forkScoped)
       : undefined
     yield* Effect.yieldNow
     yield* TestLLM.push(fixture.completeEvents)
@@ -785,7 +785,7 @@ const verifyEphemeralDeltas = (kind: FragmentKind) =>
       : []
     if (live) {
       const streamed = Array.from(yield* Fiber.join(live))
-      expect(streamed).toHaveLength(2)
+      expect(streamed).toHaveLength(1)
       expect(
         streamed
           .map((event) => {

packages/plugin/src/promise/adapter.ts

@@ -1,5 +1,5 @@
 import { Tool } from "@opencode-ai/schema/tool"
-import { Cause, Effect, Exit, Queue, Schema, SchemaAST, Scope, Stream } from "effect"
+import { Effect, Schema, SchemaAST, Scope, Stream } from "effect"
 import { HttpApiEndpoint, HttpApiSchema } from "effect/unstable/httpapi"
 import { define } from "../effect/plugin.js"
 import type { Context, Plugin } from "./plugin.js"
@@ -149,55 +149,13 @@ export function fromPromise(plugin: Plugin) {
             reload: () => run(host.command.reload()),
           },
           event: {
-            subscribe: () => {
-              const events = host.event.subscribe().pipe(
-                Stream.mapEffect((event) => Schema.encodeUnknownEffect(OpenCodeEvent)(event)),
-                Stream.map((event) => event as unknown as PromiseEvent),
-              )
-              return {
-                [Symbol.asyncIterator]() {
-                  const child = Scope.forkUnsafe(scope)
-                  const done = { done: true, value: undefined } as const
-                  let terminal = false
-                  let closing: Promise<IteratorResult<PromiseEvent>> | undefined
-                  const queue = Effect.gen(function* () {
-                    const queue = yield* Stream.toQueue(events, { capacity: "unbounded" })
-                    // Finalizers are LIFO: mark terminal before queue shutdown wakes a pending next().
-                    yield* Scope.addFinalizer(
-                      child,
-                      Effect.sync(() => (terminal = true)),
-                    )
-                    return queue
-                  }).pipe(Scope.provide(child), Effect.runPromiseWith(context))
-                  const iterator = {
-                    next: () => {
-                      if (terminal) return closing ?? Promise.resolve(done)
-                      return queue
-                        .then((queue) => Effect.runPromiseWith(context)(Queue.take(queue)))
-                        .then(
-                          (value) => (terminal ? (closing ?? done) : { done: false as const, value }),
-                          async (error) => {
-                            if (terminal) return closing ?? done
-                            await iterator.return()
-                            if (Cause.isDone(error)) return done
-                            throw error
-                          },
-                        )
-                    },
-                    return: () => {
-                      if (closing) return closing
-                      terminal = true
-                      closing = Effect.runPromiseWith(context)(Scope.close(child, Exit.void)).then(
-                        () => done,
-                        () => done,
-                      )
-                      return closing
-                    },
-                  }
-                  return iterator
-                },
-              }
-            },
+            subscribe: () =>
+              Stream.toAsyncIterable(
+                host.event.subscribe().pipe(
+                  Stream.mapEffect((event) => Schema.encodeUnknownEffect(OpenCodeEvent)(event)),
+                  Stream.map((event) => event as unknown as PromiseEvent),
+                ),
+              ),
           },
           integration: {
             list: adaptApiMethod(IntegrationEndpoints["integration.list"], host.integration.list),

packages/tui/src/context/local.tsx

@@ -10,6 +10,7 @@ import { useClient } from "./client"
 import { RGBA } from "@opentui/core"
 import { readJson, writeJsonAtomic } from "../util/persistence"
 import {
+  availableModelVariant,
   createModelPreferenceRepository,
   cycleModelVariant,
   modelPreferenceKey,
@@ -221,7 +222,14 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
         if (route.data.type === "session") return sessionSelection(route.data.sessionID)
         const model = newSessionModel()
         if (!model) return
-        return { ...model, variant: normalizeModelVariant(preferences.variant[modelPreferenceKey(model)]) }
+        const info = models()?.find((item) => item.providerID === model.providerID && item.id === model.modelID)
+        return {
+          ...model,
+          variant: availableModelVariant(
+            preferences.variant[modelPreferenceKey(model)],
+            info?.variants?.map((item) => item.id) ?? [],
+          ),
+        }
       })
 
       const currentModel = createMemo(() => {
@@ -262,13 +270,12 @@ export const { use: useLocal, provider: LocalProvider } = createSimpleContext({
         if (route.data.type === "session") {
           const sessionID = route.data.sessionID
           const current = sessionSelection(sessionID)
-          const preferred = normalizeModelVariant(
+          const preferred =
             current?.providerID === model.providerID && current.modelID === model.modelID
               ? current.variant
-              : preferences.variant[modelPreferenceKey(model)],
-          )
+              : preferences.variant[modelPreferenceKey(model)]
           const info = models()?.find((item) => item.providerID === model.providerID && item.id === model.modelID)
-          const variant = preferred && info?.variants?.some((item) => item.id === preferred) ? preferred : undefined
+          const variant = availableModelVariant(preferred, info?.variants?.map((item) => item.id) ?? [])
           setSessionDraft(sessionID, { ...model, variant })
           return true
         }

packages/tui/src/model-preference.ts

@@ -39,6 +39,11 @@ export function normalizeModelVariant(value: string | undefined) {
   return value === "default" ? undefined : value
 }
 
+export function availableModelVariant(value: string | undefined, variants: string[]) {
+  const variant = normalizeModelVariant(value)
+  return variant && variants.includes(variant) ? variant : undefined
+}
+
 export function modelPreferenceKey(model: ModelPreferenceModel) {
   return `${model.providerID}/${model.modelID}`
 }

packages/tui/test/model-preference.test.ts

@@ -1,6 +1,6 @@
 import { expect, test } from "bun:test"
 import path from "node:path"
-import { createModelPreferenceRepository, decodeModelPreference } from "../src/model-preference"
+import { availableModelVariant, createModelPreferenceRepository, decodeModelPreference } from "../src/model-preference"
 import { tmpdir } from "./fixture/fixture"
 
 test("repairs known model preferences and preserves unrelated fields", () => {
@@ -19,6 +19,12 @@ test("repairs known model preferences and preserves unrelated fields", () => {
   })
 })
 
+test("drops a saved variant that is no longer available", () => {
+  expect(availableModelVariant("medium", [])).toBeUndefined()
+  expect(availableModelVariant("medium", ["low", "high"])).toBeUndefined()
+  expect(availableModelVariant("medium", ["low", "medium", "high"])).toBe("medium")
+})
+
 test("atomically serializes patches and variant updates", async () => {
   await using tmp = await tmpdir()
   const file = path.join(tmp.path, "model.json")