fix(models): retain active-session selections across admission and refresh races (#771)

Persist explicit model selection through native switchModel during active and compacting execution. Previously those choices were local-only and could be replaced by a later authoritative session refresh.

Treat historical assistant model/agent metadata as fallback for missing selection fields, rather than overwriting the current selection on transcript hydration. Preserve choices made before or during the history request and avoid unnecessary default-model lookup for an existing choice.

Order explicit model and agent selections with the existing per-session admission queue. The gatekeeper reproduced an older prompt waiting on instruction synchronization rewriting its captured model after a successful newer selection. Queueing the complete selection, including its optimistic snapshot and rollback, ensures earlier prompt setup finishes first and later prompts use the selected context.

Treat model-selection SSE events as invalidations because their FIFO echoes can lag successful HTTP writes. Reconcile current native state through coalesced, bounded reads on the admission queue, with client/generation/selection fences and retained display on read failure. No sticky local override blocks later TUI or other-window choices.

Preserve pending optimistic choices in catalog merges and publish a fresh model reference on native confirmation. This fences native catalog snapshots captured during the write, including those arriving before authoritative event reconciliation.

Apply the same pre-request snapshot and publication merge to search, incremental catalog pages and restored-chain hydration, preserving selections introduced or confirmed while those reads were pending.

Validated 155 targeted store tests, the real Chromium session-search regression, UI typecheck, production UI build and isolated native OpenCode 2.0.16 active selection behavior. Regression coverage includes working/compacting persistence and rollback, transcript-load race orders, pending instruction/model/prompt/agent requests, delayed native selection echoes, external choices, rapid changes, rejected choices, reconciliation ownership and root/search/page/restoration catalog races. CI failure triage and clean-base reproduction evidence are recorded in the PR.
This commit is contained in:
Pascal André 2026-09-24 21:16:40 +02:00 • committed by GitHub
parent 89db55d340
commit 66bd8a79ae
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 470 additions and 33 deletions

View file

@ -325,6 +325,7 @@ jobs:
packages/ui/src/stores/permission-lifecycle.test.ts
packages/ui/src/stores/shell-store-reactivity.test.ts
packages/ui/src/stores/session-actions.test.ts
packages/ui/src/stores/session-model-reconciliation.test.ts
packages/ui/src/stores/session-pruning-events.test.ts
packages/ui/src/stores/session-pruning-projection.test.ts
packages/ui/src/stores/session-pruning-pagination.test.ts

View file

@ -26,6 +26,7 @@ where their authority, lifetime or runtime differs.
| Session outline | Structural indexes independent of transcript and excerpts; up to 16 memory snapshots / 200k entries (one larger active index allowed). Restoration persists up to 16 indexes / 200k entries / 16 MiB in optional content-addressed chunks. Saved geometry displays before full checkpoint revalidation; unchanged ranges do not resend entries. The last range grows to 512 rows before another is added. Unchanged session returns make no read; native message changes revalidate checkpoints. Destructive mutation, generation, location/project and undo boundary fence reuse. Paused scans retain cursors. | `packages/ui/src/stores/session-outline.ts`, `session-outline-persistence.ts`, `client-state-outline-partitions.ts` |
| Timeline previews | At most 512 bounded Markdown excerpts. Visible-nearby batches of 12, hovered ID first; obsolete demand cancels. Thirty-second age prompts on-demand revalidation only. Generation, mutation and undo fence reuse. No transcript-window loads or geometry changes. | `packages/ui/src/stores/timeline-previews.ts` |
| Providers/models | Retained signals; shared in-flight catalogue load; dirty-bit trailing refresh; instance, location and request-generation checks. No completed-result TTL inside `fetchProviders` itself. | `packages/ui/src/stores/session-api.ts` |
| Session model selection | Native selection events invalidate rather than overwrite the displayed selection, because SSE echoes can lag successful HTTP writes. Coalesced authoritative reads share the session admission queue with prompts and explicit selections; client/generation and captured-selection checks fence late responses. Catalog merges preserve pending optimistic choices; successful writes publish a fresh model object so pre-confirmation snapshots lose authority. Read failures retain display state until the next event or reconnect refresh. | `stores/session-model-reconciliation.ts`, `stores/session-action-queue.ts`, `stores/session-generation-recovery.ts` |
| OpenCode setup | Shared in-flight status read plus coalesced trailing refresh for focus/visibility/reconnect demand. Executable changes and explicit actions fence old responses. Failed reads retain version/path diagnostics but revoke the connected claim and stale reload/restart controls; Continue checks again before retrying the pending workspace. Activation requires a successful pre-action status read. | UI `stores/opencode-setup.ts`, `components/settings/opencode-setup-panel.tsx` |
| Git changes | Filesystem events debounce for 100 ms; one passive refresh plus a pending follow-up; hidden project/panel marked stale. Queued reads cancel on deactivation/disposal and request versions fence late status/diff results. Reads share the secondary HTTP budget. Server shares concurrent status requests, not completed results, and runs status/diff process creation in the Git worker. | `useGitChanges.ts`, `filesystem-events.ts`, server `workspaces/git-status.ts` |
| Worktree display | Last successful server snapshot; demand-driven refresh after 10 s or invalidation; one scan per workspace; obsolete scans discarded and followed by validation; UI requests coalesced. | server `workspaces/worktree-inventory.ts`, UI `stores/worktrees.ts` |

View file

@ -0,0 +1,13 @@
const admissions = new Map<string, Promise<unknown>>()
// Native context writes, prompt admission and context reconciliation share an
// order. This waits for HTTP admission, never for the model's execution to end.
export function serializeSessionAction<T>(instanceId: string, sessionId: string, action: () => Promise<T>): Promise<T> {
const key = `${instanceId}:${sessionId}`
const run = (admissions.get(key) ?? Promise.resolve()).catch(() => undefined).then(action)
const settled = run.finally(() => {
if (admissions.get(key) === settled) admissions.delete(key)
})
admissions.set(key, settled)
return settled
}

View file

@ -21,7 +21,9 @@ import {
} from "./session-actions.ts"
import { setConversationModeEnabled } from "./conversation-speech.ts"
import { getModelThinkingSelection, setModelThinkingSelection } from "./preferences"
import { sessions, setProviders, setSessions } from "./session-state.ts"
import { sessions, setAgents, setProviders, setSessions, withSession } from "./session-state.ts"
import { loadMessages } from "./session-api.ts"
import { handleNativeSessionEvent } from "./session-events.ts"
import { messageStoreBus } from "./message-v2/bus.ts"
import { contentRevision } from "../../../server/src/opencode/session-pruning/revision.ts"
import { normalizeSessionMessage } from "./message-v2/normalizers.ts"
@ -90,6 +92,9 @@ function seed(client: any): void {
;(sdkManager as any).clients.set(`${instanceId}:/workspaces/${instanceId}/instance`, client)
addInstance({ id: instanceId, folder: "/work", port: 0, pid: 0, proxyPath: "", status: "ready", client })
setSessions(new Map([[instanceId, new Map([[sessionId, session]])]]))
// Give agent-switch tests a deterministic default independent of recents.
setAgents(new Map([[instanceId, [{ id: "plan", name: "Plan", description: "", mode: "primary",
model: { providerId: "provider", modelId: "old" } }]]]))
setProviders(new Map([[instanceId, [{ id: "provider", name: "Provider", models: [
{ id: "old", name: "Old", providerId: "provider", variantKeys: ["high"] },
{ id: "new", name: "New", providerId: "provider", variantKeys: ["high"] },
@ -108,6 +113,7 @@ afterEach(() => {
serverApi.pruneSessionMessage = storageMethods.pruneSessionMessage
setSessions(new Map())
setProviders(new Map())
setAgents(new Map())
removeInstance(instanceId, { authoritative: false })
sdkManager.destroyClientsForInstance(instanceId)
setConversationModeEnabled(instanceId, false)
@ -678,6 +684,116 @@ describe("plugin RPC message pruning", () => {
})
describe("native session selection persistence", () => {
for (const blockedAt of ["instructions", "model", "prompt", "agent"] as const) {
it(`orders a model choice after pending ${blockedAt} and before the next prompt`, async () => {
let release!: () => void
const pending = new Promise<void>(resolve => { release = resolve })
let entered!: () => void
const started = new Promise<void>(resolve => { entered = resolve })
let held = false, sequence = 0, nativeModel = "old"
const calls: string[] = []
const hold = async (stage: string) => {
if (stage !== blockedAt || held) return
held = true
entered()
await pending
}
seed({ session: {
get: async () => ({ model: { providerID: "provider", id: nativeModel } }),
instructions: { entry: { put: () => hold("instructions"), remove: async () => {} } },
switchAgent: () => hold("agent"),
switchModel: async ({ model }: any) => {
await hold("model")
nativeModel = model.id
calls.push(`model:${model.id}`)
handleNativeSessionEvent(instanceId, {
id: `selection-${++sequence}`, type: "session.model.selected", created: sequence,
durable: { aggregateID: sessionId, seq: sequence, version: 1 },
data: { sessionID: sessionId, model },
} as any)
},
prompt: async (input: any) => { await hold("prompt"); calls.push(`prompt:${nativeModel}`); return { id: input.id } },
} })
if (blockedAt !== "agent") withSession(instanceId, sessionId, current => { current.status = "working" })
const first = blockedAt === "agent" ? updateSessionAgent(instanceId, sessionId, "plan")
: sendMessage(instanceId, sessionId, "first", undefined, { delivery: "steer" })
await started
const selection = updateSessionModel(instanceId, sessionId, { providerId: "provider", modelId: "new" })
const second = sendMessage(instanceId, sessionId, "second")
release()
await Promise.all([first, selection, second])
assert.deepEqual(calls, blockedAt === "agent"
? ["model:old", "model:new", "model:new", "prompt:new"]
: ["model:old", "prompt:old", "model:new", "model:new", "prompt:new"])
assert.equal(nativeModel, "new")
assert.equal(sessions().get(instanceId)?.get(sessionId)?.model.modelId, "new")
})
}
for (const rejectFirst of [false, true]) {
it(`orders rapid choices and takes rollback snapshots at admission (first rejected: ${rejectFirst})`, async () => {
let release!: () => void
const pending = new Promise<void>(resolve => { release = resolve })
const calls: string[] = []
seed({ session: { switchModel: async ({ model }: any) => {
calls.push(model.id)
if (calls.length === 1) { await pending; if (rejectFirst) throw new Error("first rejected") }
} } })
const first = updateSessionModel(instanceId, sessionId, { providerId: "provider", modelId: "new" })
const outcome = first.catch(error => { assert.equal(error.message, "first rejected") })
const second = updateSessionModel(instanceId, sessionId, { providerId: "provider", modelId: "old" })
release()
await Promise.all([outcome, second])
assert.deepEqual(calls, ["new", "old"])
assert.equal(sessions().get(instanceId)?.get(sessionId)?.model.modelId, "old")
})
}
for (const status of ["working", "compacting"] as const) {
it(`persists one model selection while ${status}`, async () => {
const inputs: unknown[] = []
seed({ session: { switchModel: async (input: unknown) => { inputs.push(input) } } })
withSession(instanceId, sessionId, current => { current.status = status })
await updateSessionModel(instanceId, sessionId, { providerId: "provider", modelId: "new" })
assert.deepEqual(inputs, [{ sessionID: sessionId, model: { providerID: "provider", id: "new" } }])
})
it(`rolls back a rejected model switch while ${status}`, async () => {
const error = new Error("model switch rejected")
seed({ session: { switchModel: async () => { throw error } } })
withSession(instanceId, sessionId, current => { current.status = status })
await assert.rejects(updateSessionModel(instanceId, sessionId, { providerId: "provider", modelId: "new" }), error)
assert.deepEqual(sessions().get(instanceId)?.get(sessionId)?.model, { providerId: "provider", modelId: "old" })
assert.equal(sessions().get(instanceId)?.get(sessionId)?.status, status)
assert.equal(sessions().get(instanceId)?.get(sessionId)?.modelSelectionPending, undefined)
})
}
for (const duringLoad of [false, true]) {
it(`retains the selected model when old transcript history reloads (${duringLoad ? "in-flight selection" : "already selected"})`, async () => {
let release!: (value: any) => void
const pending = new Promise(resolve => { release = resolve })
seed({ session: { switchModel: async () => {} }, message: { list: async () => pending } })
const select = () => updateSessionModel(instanceId, sessionId, { providerId: "provider", modelId: "new" })
if (!duringLoad) await select()
const loading = loadMessages(instanceId, sessionId, { force: true })
if (duringLoad) await select()
release({ data: [{ id: "historical-assistant", type: "assistant", agent: "plan",
model: { providerID: "provider", id: "old" }, time: { created: 1, completed: 2 }, content: [] }], cursor: {} })
await loading
assert.deepEqual(sessions().get(instanceId)?.get(sessionId)?.model, { providerId: "provider", modelId: "new" })
assert.equal(sessions().get(instanceId)?.get(sessionId)?.agent, "build")
})
}
it("still hydrates missing selections from the latest transcript", async () => {
seed({ message: { list: async () => ({ data: [{ id: "historical-assistant", type: "assistant", agent: "plan",
model: { providerID: "provider", id: "old" }, time: { created: 1, completed: 2 }, content: [] }], cursor: {} }) } })
withSession(instanceId, sessionId, current => { current.agent = ""; current.model = { providerId: "", modelId: "" } })
await loadMessages(instanceId, sessionId, { force: true })
assert.deepEqual(sessions().get(instanceId)?.get(sessionId)?.model, { providerId: "provider", modelId: "old" })
assert.equal(sessions().get(instanceId)?.get(sessionId)?.agent, "plan")
})
it("switches the native agent", async () => {
const inputs: unknown[] = []
seed({ session: {

View file

@ -21,20 +21,10 @@ import { normalizeSessionMessage } from "./message-v2/normalizers"
import { getLogger } from "../lib/logger"
import { clearConversationPlaybackForSession } from "./conversation-speech"
import { syncSessionInstructions } from "./session-instructions"
import { serializeSessionAction } from "./session-action-queue"
const log = getLogger("actions")
const technicalPartUpdates = new Map<string, Promise<void>>()
const sessionAdmissions = new Map<string, Promise<unknown>>()
function serializeSessionAction<T>(instanceId: string, sessionId: string, action: () => Promise<T>): Promise<T> {
const key = `${instanceId}:${sessionId}`
const run = (sessionAdmissions.get(key) ?? Promise.resolve()).catch(() => undefined).then(action)
const settled = run.finally(() => {
if (sessionAdmissions.get(key) === settled) sessionAdmissions.delete(key)
})
sessionAdmissions.set(key, settled)
return settled
}
function admitSessionAction<T>(
instanceId: string,
@ -425,7 +415,14 @@ async function abortSession(instanceId: string, sessionId: string): Promise<void
}
}
async function updateSessionAgent(instanceId: string, sessionId: string, agent: string): Promise<void> {
// Selection and prompt setup both write native session context. Order the
// entire operation (including its local snapshot/rollback) with admissions so
// a prompt waiting on instructions cannot later overwrite a successful choice.
function updateSessionAgent(instanceId: string, sessionId: string, agent: string): Promise<void> {
return serializeSessionAction(instanceId, sessionId, () => applySessionAgent(instanceId, sessionId, agent))
}
async function applySessionAgent(instanceId: string, sessionId: string, agent: string): Promise<void> {
const instanceSessions = sessions().get(instanceId)
const session = instanceSessions?.get(sessionId)
if (!session) {
@ -475,7 +472,15 @@ async function updateSessionAgent(instanceId: string, sessionId: string, agent:
}
}
async function updateSessionModel(
function updateSessionModel(
instanceId: string,
sessionId: string,
model: { providerId: string; modelId: string },
): Promise<void> {
return serializeSessionAction(instanceId, sessionId, () => applySessionModel(instanceId, sessionId, model))
}
async function applySessionModel(
instanceId: string,
sessionId: string,
model: { providerId: string; modelId: string },
@ -493,16 +498,25 @@ async function updateSessionModel(
withSession(instanceId, sessionId, (current) => {
current.model = model
current.modelSelectionPending = true
})
const nativeModel = getNativeModel(instanceId, model)
try {
if (!isSessionBusy(instanceId, sessionId)) {
await getRootClient(instanceId).session.switchModel({ sessionID: sessionId, model: nativeModel })
}
// Native model selection is supported during execution too. Keeping this
// local until the next prompt lets a session refresh restore the old model.
await getRootClient(instanceId).session.switchModel({ sessionID: sessionId, model: nativeModel })
withSession(instanceId, sessionId, (current) => {
// Confirmation is a new selection revision even if its values match the
// optimistic choice. Catalog reads started during the write captured that
// earlier object and must not publish their pre-write native model later.
current.model = { ...model }
delete current.modelSelectionPending
})
} catch (error) {
withSession(instanceId, sessionId, (current) => {
if (current.model.providerId !== model.providerId || current.model.modelId !== model.modelId) return false
delete current.modelSelectionPending
if (current.model.providerId !== model.providerId || current.model.modelId !== model.modelId) return
current.model = session.model
})
throw error

View file

@ -508,7 +508,11 @@ async function hydrateRestoredSessionChainAttempt(
if (!isRequestCurrent() || getAuthoritativelyDeletedSessionIdsForInstance(instanceId).has(sessionId) || signal?.aborted) return prev
const next = new Map(prev)
const instanceSessions = new Map(next.get(instanceId) ?? new Map())
instanceSessions.set(sessionId, toClientSessionV2(instanceId, apiSession, instanceSessions.get(sessionId)))
const latest = instanceSessions.get(sessionId)
// This row was absent when the read began. A concurrent event or
// selection that introduced it owns its current fields.
const merged = mergeFetchedSessionRuntimeState(toClientSessionV2(instanceId, apiSession), undefined, latest)
if (merged) instanceSessions.set(sessionId, merged)
next.set(instanceId, instanceSessions)
return next
})
@ -868,6 +872,7 @@ async function loadNextSessionPage(instanceId: string): Promise<void> {
const isCurrent = () => instances().get(instanceId)?.client === client
&& sessionListRequestIds.get(instanceId) === listRequestId
&& generationCurrent()
const capturedSessions = new Map(sessions().get(instanceId) ?? [])
const [response, activeSessions] = await Promise.all([
fetchV2Sessions(instanceId, { cursor }),
getRootClient(instanceId).session.active().catch((error) => {
@ -891,9 +896,9 @@ async function loadNextSessionPage(instanceId: string): Promise<void> {
const current = new Map(next.get(instanceId) ?? [])
for (const item of pageSessions) {
if (!deleted.has(item.id)) {
const existing = current.get(item.id)
const fetched = withActiveSessionState(instanceId, item, existing, activeSessions)
const merged = mergeFetchedSessionRuntimeState(fetched, existing, existing, false)
const captured = capturedSessions.get(item.id)
const fetched = withActiveSessionState(instanceId, item, captured, activeSessions)
const merged = mergeFetchedSessionRuntimeState(fetched, captured, current.get(item.id), false)
if (merged) current.set(item.id, merged)
}
}
@ -926,6 +931,7 @@ async function searchSessions(instanceId: string, query: string): Promise<void>
const isCurrent = () => isLatestSessionSearch(instanceId, trimmedQuery, requestId)
&& instances().get(instanceId)?.client === client
&& generationCurrent()
const capturedSessions = new Map(sessions().get(instanceId) ?? [])
try {
log.info("v2.session.search", { instanceId, query: trimmedQuery, directory: instance.folder })
@ -963,8 +969,10 @@ async function searchSessions(instanceId: string, query: string): Promise<void>
for (const apiSession of searchResults) {
if (deletedSessionIds.has(apiSession.id)) continue
const existingSession = instanceSessions.get(apiSession.id)
instanceSessions.set(apiSession.id, toClientSessionV2(instanceId, apiSession, existingSession))
const captured = capturedSessions.get(apiSession.id)
const fetched = toClientSessionV2(instanceId, apiSession, captured)
const merged = mergeFetchedSessionRuntimeState(fetched, captured, instanceSessions.get(apiSession.id))
if (merged) instanceSessions.set(apiSession.id, merged)
}
next.set(instanceId, instanceSessions)
@ -1697,7 +1705,7 @@ async function loadMessages(
if (agentName && providerID && modelID) break
}
if (!agentName && !providerID && !modelID) {
if (!agentName && !providerID && !modelID && (!session.model.providerId || !session.model.modelId)) {
const defaultModel = await getDefaultModel(instanceId, session.agent)
if (!isCurrent()) return
agentName = session.agent
@ -1714,8 +1722,13 @@ async function loadMessages(
if (!existingSession) return next
nextInstanceSessions.set(sessionId, {
...existingSession,
agent: agentName || existingSession.agent,
model: providerID && modelID ? { providerId: providerID, modelId: modelID } : existingSession.model,
// Transcript metadata describes past execution, not the current
// selection. Only fill missing values, including when a user or native
// selection event changed them while this history request was pending.
agent: existingSession.agent || agentName,
model: existingSession.model.providerId && existingSession.model.modelId
? existingSession.model
: providerID && modelID ? { providerId: providerID, modelId: modelID } : existingSession.model,
})
next.set(instanceId, nextInstanceSessions)
return next

View file

@ -13,6 +13,7 @@ import type {
TuiToastShow,
} from "@opencode/client"
import { getLogger } from "../lib/logger"
import { reconcileSessionModel } from "./session-model-reconciliation"
import { handlePruningEvent } from "./session-pruning-events"
import type { EventSessionDeleted, NativeSessionEvent } from "../lib/sse-manager"
import {
@ -104,10 +105,12 @@ function handleNativeSessionEvent(instanceId: string, event: NativeSessionEvent)
withSession(instanceId, event.data.sessionID, (session) => { session.agent = event.data.agent })
return
case "session.model.selected":
if (!sessions().get(instanceId)?.has(event.data.sessionID)) void fetchSessionInfo(instanceId, event.data.sessionID, event.location?.directory)
withSession(instanceId, event.data.sessionID, (session) => {
session.model = { providerId: event.data.model.providerID, modelId: event.data.model.id }
})
if (!sessions().get(instanceId)?.has(event.data.sessionID)) {
void fetchSessionInfo(instanceId, event.data.sessionID, event.location?.directory)
.then(() => reconcileSessionModel(instanceId, event.data.sessionID))
} else {
void reconcileSessionModel(instanceId, event.data.sessionID)
}
return
case "session.usage.updated":
withSession(instanceId, event.data.sessionID, (session) => {

View file

@ -39,7 +39,7 @@ export function mergeFetchedSessionRuntimeState(
if (captured && !latest) return null
if (!latest) return fetched
if (latest === captured) {
return latest.generationAdmissionToken === undefined ? fetched : { ...fetched, ...latest }
return latest.generationAdmissionToken === undefined ? preservePendingModel(fetched, latest) : { ...fetched, ...latest }
}
const merged = { ...fetched }
const keys = new Set<keyof Session>([
@ -59,5 +59,11 @@ export function mergeFetchedSessionRuntimeState(
(merged as any)[key] = fetched[key]
}
}
return merged
return preservePendingModel(merged, latest)
}
function preservePendingModel(fetched: Session, latest: Session): Session {
return latest.modelSelectionPending
? { ...fetched, model: latest.model, modelSelectionPending: true }
: fetched
}

View file

@ -0,0 +1,221 @@
import assert from "node:assert/strict"
import { after, afterEach, before, it } from "node:test"
import { serverApi } from "../lib/api-client.ts"
import { sdkManager } from "../lib/sdk-manager.ts"
import { addInstance, removeInstance, updateInstance } from "./instances.ts"
import { sendMessage, updateSessionModel } from "./session-actions.ts"
import { handleNativeSessionEvent } from "./session-events.ts"
import { fetchSessions, hydrateRestoredSessionChain, loadMoreSessions, searchSessions } from "./session-api.ts"
import { serializeSessionAction } from "./session-action-queue.ts"
import { reconcileSessionModel } from "./session-model-reconciliation.ts"
import { sessions, setSessions, setProviders, setSessionPage, withSession } from "./session-state.ts"
import type { Session } from "../types/session.ts"
const instanceId = "model-reconciliation", sessionId = "session"
const storage = { fetchConfigOwner: serverApi.fetchConfigOwner, fetchStateOwner: serverApi.fetchStateOwner,
patchStateOwner: serverApi.patchStateOwner, fetchWorktrees: serverApi.fetchWorktrees }
before(() => {
serverApi.fetchConfigOwner = async () => ({}) as any
serverApi.fetchStateOwner = async () => ({}) as any
serverApi.patchStateOwner = async (_owner, patch) => patch as any
serverApi.fetchWorktrees = async () => ({ isGitRepo: true, worktrees: [] }) as any
})
after(() => Object.assign(serverApi, storage))
afterEach(() => {
setSessions(new Map())
setProviders(new Map())
removeInstance(instanceId, { authoritative: false })
sdkManager.destroyClientsForInstance(instanceId)
})
function deferred<T>() {
let resolve!: (value: T) => void
const promise = new Promise<T>(done => { resolve = done })
return { promise, resolve }
}
function seed() {
let nativeModel = "base", sequence = 0
const events: any[] = [], calls: string[] = []
const event = (model: string) => ({ id: `event-${++sequence}`, type: "session.model.selected", created: sequence,
durable: { aggregateID: sessionId, seq: sequence, version: 1 },
data: { sessionID: sessionId, model: { providerID: "provider", id: model } } })
const info = () => ({ id: sessionId, title: sessionId, agent: "build", parentID: null,
model: { providerID: "provider", id: nativeModel }, location: { directory: "/work" }, time: { created: 1, updated: 1 } })
const client = { session: {
list: async (): Promise<any> => ({ data: [info()], cursor: {} }),
active: async () => ({ [sessionId]: {} }),
get: async (_input?: unknown, _options?: unknown): Promise<any> => info(),
instructions: { entry: { put: async () => {}, remove: async () => {} } },
switchAgent: async () => {},
switchModel: async ({ model }: any) => {
calls.push(`model:${model.id}`)
if (nativeModel !== model.id) { nativeModel = model.id; events.push(event(model.id)) }
},
prompt: async (input: any) => { calls.push(`prompt:${nativeModel}`); return { id: input.id } },
} }
;(sdkManager as any).clients.set(`${instanceId}:/workspaces/${instanceId}/instance`, client)
addInstance({ id: instanceId, folder: "/work", port: 0, pid: 0, proxyPath: "", status: "ready", client: client as any })
setSessions(new Map([[instanceId, new Map([[sessionId, {
id: sessionId, instanceId, parentId: null, title: sessionId, agent: "build",
model: { providerId: "provider", modelId: "base" }, status: "working",
location: { directory: "/work" }, time: { created: 1, updated: 1 },
} as Session]])]]))
setProviders(new Map([[instanceId, [{ id: "provider", name: "Provider", models:
["base", "old", "new"].map(id => ({ id, name: id, providerId: "provider" })) }]]]))
return { client, calls, events, info, native: () => nativeModel,
external: (model: string) => { nativeModel = model; handleNativeSessionEvent(instanceId, event(model) as any) },
deliver: () => handleNativeSessionEvent(instanceId, events.shift()),
}
}
const displayed = () => sessions().get(instanceId)?.get(sessionId)?.model.modelId
const select = (modelId: string) => updateSessionModel(instanceId, sessionId, { providerId: "provider", modelId })
for (const path of ["root", "search", "page"] as const) for (const listFirst of [true, false]) {
it(`protects selection from ${path} catalog captured during its write (list settles first: ${listFirst})`, async () => {
const fixture = seed()
const mutation = deferred<void>(), writeStarted = deferred<void>()
const list = deferred<void>(), listStarted = deferred<void>()
const get = deferred<void>(), getStarted = deferred<void>()
const switchModel = fixture.client.session.switchModel
fixture.client.session.switchModel = async input => {
writeStarted.resolve()
await mutation.promise
await switchModel(input)
if (fixture.events.length) fixture.deliver()
}
fixture.client.session.list = async () => {
const snapshot = fixture.info()
listStarted.resolve()
await list.promise
return { data: [snapshot], cursor: {} }
}
fixture.client.session.get = async () => {
const snapshot = fixture.info()
getStarted.resolve()
await get.promise
return snapshot
}
const selection = select("new")
await writeStarted.promise
if (path === "page") setSessionPage(instanceId, [], true, true, "next-page")
const refreshing = path === "search" ? searchSessions(instanceId, "session")
: path === "page" ? loadMoreSessions(instanceId) : fetchSessions(instanceId)
await listStarted.promise
if (listFirst) {
list.resolve()
await refreshing
assert.equal(displayed(), "new", "Keep the optimistic choice while its write is pending")
}
mutation.resolve()
await selection
await getStarted.promise
list.resolve()
await refreshing
assert.equal(displayed(), "new", "Stale catalog cannot overwrite the confirmed choice")
get.resolve()
await serializeSessionAction(instanceId, sessionId, async () => {})
await sendMessage(instanceId, sessionId, "next")
assert.equal(displayed(), "new")
assert.equal(fixture.native(), "new")
assert.equal(fixture.calls.at(-1), "prompt:new")
assert.equal(sessions().get(instanceId)?.get(sessionId)?.modelSelectionPending, undefined)
})
}
it("preserves a concurrently introduced selection during restored-chain hydration", async () => {
const fixture = seed(), held = deferred<void>(), started = deferred<void>()
const saved = sessions()
setSessions(new Map())
fixture.client.session.get = async () => {
const snapshot = fixture.info()
started.resolve()
await held.promise
return snapshot
}
const restoring = hydrateRestoredSessionChain(instanceId, [sessionId])
await started.promise
setSessions(saved)
await select("new")
held.resolve()
await restoring
assert.equal(displayed(), "new")
await sendMessage(instanceId, sessionId, "next")
assert.equal(fixture.calls.at(-1), "prompt:new")
})
it("does not let delayed FIFO echoes overwrite a successful choice or the next prompt", async () => {
const fixture = seed()
await select("old")
await select("new")
fixture.deliver()
assert.equal(displayed(), "new")
await sendMessage(instanceId, sessionId, "next")
while (fixture.events.length) fixture.deliver()
await reconcileSessionModel(instanceId, sessionId)
assert.deepEqual(fixture.calls, ["model:old", "model:new", "model:new", "prompt:new"])
assert.equal(fixture.native(), "new")
assert.equal(displayed(), "new")
// A no-op switch has no echo. It must not leave a sticky local override that
// ignores a later selection made in the TUI or another window.
await select("new")
fixture.external("old")
await sendMessage(instanceId, sessionId, "external choice")
assert.equal(fixture.calls.at(-1), "prompt:old")
assert.equal(displayed(), "old")
})
it("coalesces event bursts into one trailing read without publishing stale snapshots", async () => {
const fixture = seed(), held = deferred<any>(), started = deferred<void>()
let reads = 0
fixture.client.session.get = async () => {
if (++reads === 1) { started.resolve(); return held.promise }
return fixture.info()
}
const reading = reconcileSessionModel(instanceId, sessionId)
await started.promise
for (let i = 0; i < 20; i++) fixture.external(i % 2 ? "new" : "old")
held.resolve({ model: { providerID: "provider", id: "base" } })
await reading
assert.equal(reads, 2)
assert.equal(displayed(), "new")
})
it("keeps a later local selection ordered after a pending read", async () => {
const fixture = seed(), held = deferred<any>(), started = deferred<void>()
fixture.client.session.get = async () => { started.resolve(); return held.promise }
const reading = reconcileSessionModel(instanceId, sessionId)
await started.promise
const selection = select("new")
held.resolve({ model: { providerID: "provider", id: "old" } })
await Promise.all([reading, selection])
assert.equal(displayed(), "new")
assert.equal(fixture.native(), "new")
})
it("retains selection on read failure and permits the next event to recover", async () => {
const fixture = seed()
await select("new")
fixture.client.session.get = async () => { throw new Error("offline") }
await reconcileSessionModel(instanceId, sessionId)
assert.equal(displayed(), "new")
fixture.client.session.get = async () => fixture.info()
fixture.external("old")
await reconcileSessionModel(instanceId, sessionId)
assert.equal(displayed(), "old")
})
for (const supersede of ["client", "selection", "deletion"] as const) {
it(`fences a pending response after ${supersede} changes`, async () => {
const fixture = seed(), held = deferred<any>(), started = deferred<void>()
fixture.client.session.get = async () => { started.resolve(); return held.promise }
const reading = reconcileSessionModel(instanceId, sessionId)
await started.promise
if (supersede === "client") updateInstance(instanceId, { client: {} as any })
if (supersede === "selection") withSession(instanceId, sessionId, session => { session.model = { providerId: "provider", modelId: "new" } })
if (supersede === "deletion") setSessions(new Map())
held.resolve({ model: { providerID: "provider", id: "old" } })
await reading
assert.equal(displayed(), supersede === "deletion" ? undefined : supersede === "selection" ? "new" : "base")
})
}

View file

@ -0,0 +1,48 @@
import { getLogger } from "../lib/logger"
import { instances } from "./instances"
import { getRootClient } from "./opencode-client"
import { getOpenCodeInstanceGeneration } from "./opencode-data"
import { serializeSessionAction } from "./session-action-queue"
import { sessions, withSession } from "./session-state"
const log = getLogger("actions")
const pending = new Map<string, { dirty: boolean; current: () => boolean; promise: Promise<void> }>()
// SSE echoes can arrive after a newer HTTP selection has already succeeded.
// Treat them as invalidations, not current selection values. Reading on the
// admission queue also prevents a read of pre-mutation state during a switch.
export function reconcileSessionModel(instanceId: string, sessionId: string): Promise<void> {
const key = `${instanceId}\0${sessionId}`
const previous = pending.get(key)
if (previous?.current()) {
previous.dirty = true
return previous.promise
}
const owner = instances().get(instanceId)?.client
const generation = getOpenCodeInstanceGeneration(instanceId)
const current = () => Boolean(owner && instances().get(instanceId)?.client === owner &&
getOpenCodeInstanceGeneration(instanceId) === generation)
const state = { dirty: false, current, promise: Promise.resolve() }
state.promise = serializeSessionAction(instanceId, sessionId, async () => {
do {
state.dirty = false
const captured = sessions().get(instanceId)?.get(sessionId)
if (!current() || !captured) return
const info = await getRootClient(instanceId).session.get({ sessionID: sessionId }, { signal: AbortSignal.timeout(8000) })
if (!current()) return
if (state.dirty) continue
withSession(instanceId, sessionId, session => {
// Session replacement or another authoritative refresh owns its newer
// selection; never resurrect a removed session or overwrite that value.
if (session.model !== captured.model || !info.model) return false
session.model = { providerId: info.model.providerID, modelId: info.model.id }
})
} while (state.dirty)
}).catch(error => {
log.warn("Failed to reconcile session model", { instanceId, sessionId, error })
}).finally(() => {
if (pending.get(key) === state) pending.delete(key)
})
pending.set(key, state)
return state.promise
}

View file

@ -81,6 +81,7 @@ export interface Session extends Omit<SDKSession, "parentID" | "model"> {
generationRecovery?: GenerationRecoveryState | null // Local recovery state for work interrupted across restarts
runtimeStatusKnown?: boolean // Whether idle/working came from an authoritative runtime response
generationAdmissionToken?: number // Guards recovery state while a new input is being admitted
modelSelectionPending?: boolean // Protects optimistic model choice from catalog snapshots during its native write
}
// Adapter function to convert SDK Session to client Session