From 6800bcc33949c0647f9630b58b47d3d37d4c6ee9 Mon Sep 17 00:00:00 2001 From: murashit Date: Sat, 18 Jul 2026 06:04:06 +0900 Subject: [PATCH] refactor(runtime): simplify thread settings serialization --- .../runtime-settings-commit-coordinator.ts | 211 ------------------ .../application/runtime/settings-actions.ts | 163 ++++++++++---- ...untime-settings-commit-coordinator.test.ts | 40 ---- .../runtime/settings-actions.test.ts | 92 +++++++- 4 files changed, 216 insertions(+), 290 deletions(-) delete mode 100644 src/features/chat/application/runtime/runtime-settings-commit-coordinator.ts delete mode 100644 tests/features/chat/application/runtime/runtime-settings-commit-coordinator.test.ts diff --git a/src/features/chat/application/runtime/runtime-settings-commit-coordinator.ts b/src/features/chat/application/runtime/runtime-settings-commit-coordinator.ts deleted file mode 100644 index 23219272..00000000 --- a/src/features/chat/application/runtime/runtime-settings-commit-coordinator.ts +++ /dev/null @@ -1,211 +0,0 @@ -import type { RuntimeSettingsPatch } from "../../../../domain/runtime/thread-settings"; -import { createKeyedOperationQueue, type KeyedOperationQueue } from "../../../../shared/runtime/keyed-operation-queue"; - -export type RuntimeSettingsCommitTarget = { readonly kind: "settle" } | { readonly kind: "fields"; readonly update: RuntimeSettingsPatch }; - -interface RuntimeSettingsCommitScope { - readonly threadId: string; - readonly panelTargetRevision: number; -} - -export interface RuntimeSettingsCommitCoordinatorHost { - scopeIsCurrent: (scope: RuntimeSettingsCommitScope) => boolean; - pendingPatch: () => RuntimeSettingsPatch; - updateThreadSettings: (threadId: string, update: RuntimeSettingsPatch) => Promise; - commitPatch: (update: RuntimeSettingsPatch) => void; - reportError: (error: unknown) => void; -} - -export interface RuntimeSettingsCommitCoordinator { - commit: (scope: RuntimeSettingsCommitScope, target: RuntimeSettingsCommitTarget) => Promise; -} - -interface PendingCommit { - readonly kind: RuntimeSettingsCommitTarget["kind"]; - remaining: RuntimeSettingsPatch; - resolve: (ok: boolean) => void; -} - -interface ThreadCommitDrain { - revision: number; - readonly pending: PendingCommit[]; -} - -export function createRuntimeSettingsCommitCoordinator( - host: RuntimeSettingsCommitCoordinatorHost, - threadCommits: KeyedOperationQueue = createKeyedOperationQueue(), -): RuntimeSettingsCommitCoordinator { - const drainsByThread = new Map>(); - - return { - commit: (scope, target) => { - let threadDrains = drainsByThread.get(scope.threadId); - if (!threadDrains) { - threadDrains = new Map(); - drainsByThread.set(scope.threadId, threadDrains); - } - let drain = threadDrains.get(scope.panelTargetRevision); - if (!drain) { - drain = { revision: 0, pending: [] }; - threadDrains.set(scope.panelTargetRevision, drain); - } - drain.revision += 1; - const result = new Promise((resolve) => { - drain.pending.push({ - kind: target.kind, - remaining: target.kind === "fields" ? { ...target.update } : {}, - resolve, - }); - }); - if (drain.pending.length === 1) { - const scheduledDrain = drain; - void threadCommits - .run(scope.threadId, () => runThreadCommitDrain(host, scope, scheduledDrain)) - .finally(() => { - if (threadDrains.get(scope.panelTargetRevision) === scheduledDrain) { - threadDrains.delete(scope.panelTargetRevision); - if (threadDrains.size === 0) drainsByThread.delete(scope.threadId); - } - }); - } - return result; - }, - }; -} - -async function runThreadCommitDrain( - host: RuntimeSettingsCommitCoordinatorHost, - scope: RuntimeSettingsCommitScope, - drain: ThreadCommitDrain, -): Promise { - while (drain.pending.length > 0) { - if (!host.scopeIsCurrent(scope)) { - resolveAll(drain, false); - return; - } - - const update = host.pendingPatch(); - if (patchEmpty(update)) { - resolveSettledCommits(drain); - return; - } - - const sentRevision = drain.revision; - let updated: boolean; - try { - updated = await host.updateThreadSettings(scope.threadId, update); - } catch (error) { - if (host.scopeIsCurrent(scope)) host.reportError(error); - resolveAll(drain, false); - return; - } - - if (!host.scopeIsCurrent(scope)) { - resolveAll(drain, false); - return; - } - if (!updated) { - resolveAll(drain, false); - return; - } - - const committed = matchingPendingPatch(host.pendingPatch(), update); - if (!patchEmpty(committed)) host.commitPatch(committed); - resolveFieldCommits(drain, committed, host.pendingPatch()); - - if (drain.pending.length === 0) return; - if (patchEmpty(host.pendingPatch())) { - resolveSettledCommits(drain); - return; - } - if (drain.revision !== sentRevision || drain.pending.some((pending) => pending.kind === "settle")) { - continue; - } - - resolveAll(drain, false); - return; - } -} - -function matchingPendingPatch(current: RuntimeSettingsPatch, sent: RuntimeSettingsPatch): RuntimeSettingsPatch { - const matching: RuntimeSettingsPatch = {}; - for (const key of patchKeys(sent)) { - if (key in current && threadSettingsValueEqual(current[key], sent[key])) { - Object.assign(matching, { [key]: sent[key] }); - } - } - return matching; -} - -function resolveFieldCommits(drain: ThreadCommitDrain, committed: RuntimeSettingsPatch, current: RuntimeSettingsPatch): void { - for (const pending of [...drain.pending]) { - if (pending.kind === "settle") continue; - for (const key of patchKeys(pending.remaining)) { - if (key in committed && threadSettingsValueEqual(pending.remaining[key], committed[key])) { - pending.remaining = patchWithoutField(pending.remaining, key); - } - } - if (patchEmpty(pending.remaining)) { - resolveCommit(drain, pending, true); - continue; - } - if (patchKeys(pending.remaining).some((key) => !(key in current) || !threadSettingsValueEqual(pending.remaining[key], current[key]))) { - resolveCommit(drain, pending, false); - } - } -} - -function resolveSettledCommits(drain: ThreadCommitDrain): void { - for (const pending of [...drain.pending]) { - resolveCommit(drain, pending, pending.kind === "settle" || patchEmpty(pending.remaining)); - } -} - -function resolveAll(drain: ThreadCommitDrain, ok: boolean): void { - for (const pending of [...drain.pending]) resolveCommit(drain, pending, ok); -} - -function resolveCommit(drain: ThreadCommitDrain, pending: PendingCommit, ok: boolean): void { - const index = drain.pending.indexOf(pending); - if (index >= 0) drain.pending.splice(index, 1); - pending.resolve(ok); -} - -function patchKeys(patch: RuntimeSettingsPatch): (keyof RuntimeSettingsPatch)[] { - return Object.keys(patch) as (keyof RuntimeSettingsPatch)[]; -} - -function patchEmpty(patch: RuntimeSettingsPatch): boolean { - return patchKeys(patch).length === 0; -} - -function patchWithoutField(patch: RuntimeSettingsPatch, omittedKey: keyof RuntimeSettingsPatch): RuntimeSettingsPatch { - const remaining: RuntimeSettingsPatch = {}; - for (const key of patchKeys(patch)) { - if (key !== omittedKey) Object.assign(remaining, { [key]: patch[key] }); - } - return remaining; -} - -function threadSettingsValueEqual(left: unknown, right: unknown): boolean { - if (Object.is(left, right)) return true; - if (Array.isArray(left) || Array.isArray(right)) { - return ( - Array.isArray(left) && - Array.isArray(right) && - left.length === right.length && - left.every((value, index) => threadSettingsValueEqual(value, right[index])) - ); - } - if (!isPlainRecord(left) || !isPlainRecord(right)) return false; - const leftKeys = Object.keys(left); - const rightKeys = Object.keys(right); - return ( - leftKeys.length === rightKeys.length && - leftKeys.every((key) => Object.hasOwn(right, key) && threadSettingsValueEqual(left[key], right[key])) - ); -} - -function isPlainRecord(value: unknown): value is Record { - return value !== null && typeof value === "object"; -} diff --git a/src/features/chat/application/runtime/settings-actions.ts b/src/features/chat/application/runtime/settings-actions.ts index 6cab11be..45c17fa3 100644 --- a/src/features/chat/application/runtime/settings-actions.ts +++ b/src/features/chat/application/runtime/settings-actions.ts @@ -14,11 +14,6 @@ import { type ActivePanelOperation, activePanelOperationDecision } from "../pane import { capturePanelTargetLease, panelTargetLeaseIsCurrent } from "../state/panel-target"; import { activeThreadId, type ChatAction, type ChatState } from "../state/root-reducer"; import type { ChatStateStore } from "../state/store"; -import { - createRuntimeSettingsCommitCoordinator, - type RuntimeSettingsCommitCoordinator, - type RuntimeSettingsCommitTarget, -} from "./runtime-settings-commit-coordinator"; import type { RuntimeSettingsTransport } from "./settings-transport"; interface RuntimeSettingsCommitResult { @@ -38,7 +33,12 @@ export interface RuntimeSettingsActionsHost { } interface RuntimeSettingsActionsContext extends RuntimeSettingsActionsHost { - runtimeSettingsCommits: RuntimeSettingsCommitCoordinator; + threadCommits: KeyedOperationQueue; +} + +interface RuntimeSettingsCommandScope { + readonly threadId: string; + readonly panelTargetRevision: number; } export interface ChatRuntimeSettingsActions { @@ -67,30 +67,7 @@ export function createChatRuntimeSettingsActions( host: RuntimeSettingsActionsHost, threadCommits: KeyedOperationQueue = createKeyedOperationQueue(), ): ChatRuntimeSettingsActions { - const runtimeSettingsCommits = createRuntimeSettingsCommitCoordinator( - { - scopeIsCurrent: (scope) => { - const currentState = state(host); - return ( - activeThreadId(currentState) === scope.threadId && - panelTargetLeaseIsCurrent(currentState, { - revision: scope.panelTargetRevision, - target: { kind: "thread", threadId: scope.threadId }, - }) - ); - }, - pendingPatch: () => currentPendingRuntimeSettingsPatch(host), - updateThreadSettings: (threadId, update) => host.runtimeTransport.updateThreadSettings(threadId, update), - commitPatch: (update) => { - dispatch(host, { type: "runtime/pending-thread-settings-committed", update }); - }, - reportError: (error) => { - host.addSystemMessage(error instanceof Error ? error.message : String(error)); - }, - }, - threadCommits, - ); - const context: RuntimeSettingsActionsContext = { ...host, runtimeSettingsCommits }; + const context: RuntimeSettingsActionsContext = { ...host, threadCommits }; return { applyPendingThreadSettings: () => applyPendingThreadSettings(context), requestModel: (model) => requestModel(context, model), @@ -137,11 +114,78 @@ async function commitPendingThreadSettings( if (Object.keys(update).length === 0) return { ok: true, collaborationModeApplied }; if (activePanelOperationBlocked(host, "thread-settings")) return { ok: false, collaborationModeApplied: false }; - const target = runtimeSettingsCommitTarget(update, fields); - const ok = await host.runtimeSettingsCommits.commit({ threadId, panelTargetRevision: panelTarget.revision }, target); + const scope = { threadId, panelTargetRevision: panelTarget.revision }; + const ok = fields + ? await commitRuntimeSettingsFields(host, scope, pickRuntimeSettingsFields(update, fields)) + : await settleRuntimeSettings(host, scope); return { ok, collaborationModeApplied: ok && collaborationModeApplied }; } +function commitRuntimeSettingsFields( + host: RuntimeSettingsActionsContext, + scope: RuntimeSettingsCommandScope, + update: RuntimeSettingsPatch, +): Promise { + if (patchEmpty(update)) return Promise.resolve(true); + const command = { ...update }; + return host.threadCommits.run(scope.threadId, async () => { + if (!runtimeSettingsScopeIsCurrent(host, scope)) return false; + if (!patchEqual(matchingPendingPatch(currentPendingRuntimeSettingsPatch(host), command), command)) return false; + + const updated = await updateRuntimeSettings(host, scope, command); + if (!updated || !runtimeSettingsScopeIsCurrent(host, scope)) return false; + + const committed = matchingPendingPatch(currentPendingRuntimeSettingsPatch(host), command); + if (!patchEmpty(committed)) commitRuntimeSettingsPatch(host, committed); + return patchEqual(committed, command); + }); +} + +function settleRuntimeSettings(host: RuntimeSettingsActionsContext, scope: RuntimeSettingsCommandScope): Promise { + return host.threadCommits.run(scope.threadId, async () => { + while (runtimeSettingsScopeIsCurrent(host, scope)) { + const update = currentPendingRuntimeSettingsPatch(host); + if (patchEmpty(update)) return true; + + if (!(await updateRuntimeSettings(host, scope, update)) || !runtimeSettingsScopeIsCurrent(host, scope)) return false; + + const committed = matchingPendingPatch(currentPendingRuntimeSettingsPatch(host), update); + if (!patchEmpty(committed)) commitRuntimeSettingsPatch(host, committed); + } + return false; + }); +} + +async function updateRuntimeSettings( + host: RuntimeSettingsActionsContext, + scope: RuntimeSettingsCommandScope, + update: RuntimeSettingsPatch, +): Promise { + try { + return await host.runtimeTransport.updateThreadSettings(scope.threadId, update); + } catch (error) { + if (runtimeSettingsScopeIsCurrent(host, scope)) { + host.addSystemMessage(error instanceof Error ? error.message : String(error)); + } + return false; + } +} + +function runtimeSettingsScopeIsCurrent(host: RuntimeSettingsActionsHost, scope: RuntimeSettingsCommandScope): boolean { + const currentState = state(host); + return ( + activeThreadId(currentState) === scope.threadId && + panelTargetLeaseIsCurrent(currentState, { + revision: scope.panelTargetRevision, + target: { kind: "thread", threadId: scope.threadId }, + }) + ); +} + +function commitRuntimeSettingsPatch(host: RuntimeSettingsActionsHost, update: RuntimeSettingsPatch): void { + dispatch(host, { type: "runtime/pending-thread-settings-committed", update }); +} + async function requestModel(host: RuntimeSettingsActionsContext, model: string): Promise { if (activePanelOperationBlocked(host, "thread-settings")) return false; dispatch(host, { type: "runtime/model-requested", model }); @@ -289,16 +333,59 @@ function currentPendingRuntimeSettingsPatch(host: RuntimeSettingsActionsHost): R return buildPendingRuntimeSettingsPatch(snapshot, config).update; } -function runtimeSettingsCommitTarget( - update: RuntimeSettingsPatch, - fields: readonly (keyof RuntimeSettingsPatch)[] | undefined, -): RuntimeSettingsCommitTarget { - if (!fields) return { kind: "settle" }; +function pickRuntimeSettingsFields(update: RuntimeSettingsPatch, fields: readonly (keyof RuntimeSettingsPatch)[]): RuntimeSettingsPatch { const targetedUpdate: RuntimeSettingsPatch = {}; for (const key of fields) { if (key in update) Object.assign(targetedUpdate, { [key]: update[key] }); } - return Object.keys(targetedUpdate).length === 0 ? { kind: "settle" } : { kind: "fields", update: targetedUpdate }; + return targetedUpdate; +} + +function matchingPendingPatch(current: RuntimeSettingsPatch, sent: RuntimeSettingsPatch): RuntimeSettingsPatch { + const matching: RuntimeSettingsPatch = {}; + for (const key of patchKeys(sent)) { + if (key in current && threadSettingsValueEqual(current[key], sent[key])) { + Object.assign(matching, { [key]: sent[key] }); + } + } + return matching; +} + +function patchEqual(left: RuntimeSettingsPatch, right: RuntimeSettingsPatch): boolean { + const leftKeys = patchKeys(left); + const rightKeys = patchKeys(right); + return leftKeys.length === rightKeys.length && leftKeys.every((key) => key in right && threadSettingsValueEqual(left[key], right[key])); +} + +function patchEmpty(patch: RuntimeSettingsPatch): boolean { + return patchKeys(patch).length === 0; +} + +function patchKeys(patch: RuntimeSettingsPatch): (keyof RuntimeSettingsPatch)[] { + return Object.keys(patch) as (keyof RuntimeSettingsPatch)[]; +} + +function threadSettingsValueEqual(left: unknown, right: unknown): boolean { + if (Object.is(left, right)) return true; + if (Array.isArray(left) || Array.isArray(right)) { + return ( + Array.isArray(left) && + Array.isArray(right) && + left.length === right.length && + left.every((value, index) => threadSettingsValueEqual(value, right[index])) + ); + } + if (!isPlainRecord(left) || !isPlainRecord(right)) return false; + const leftKeys = Object.keys(left); + const rightKeys = Object.keys(right); + return ( + leftKeys.length === rightKeys.length && + leftKeys.every((key) => Object.hasOwn(right, key) && threadSettingsValueEqual(left[key], right[key])) + ); +} + +function isPlainRecord(value: unknown): value is Record { + return value !== null && typeof value === "object"; } function runtimeProjection(host: RuntimeSettingsActionsHost): { diff --git a/tests/features/chat/application/runtime/runtime-settings-commit-coordinator.test.ts b/tests/features/chat/application/runtime/runtime-settings-commit-coordinator.test.ts deleted file mode 100644 index afdcb929..00000000 --- a/tests/features/chat/application/runtime/runtime-settings-commit-coordinator.test.ts +++ /dev/null @@ -1,40 +0,0 @@ -import { describe, expect, it, vi } from "vitest"; - -import type { RuntimeSettingsPatch } from "../../../../../src/domain/runtime/thread-settings"; -import { createRuntimeSettingsCommitCoordinator } from "../../../../../src/features/chat/application/runtime/runtime-settings-commit-coordinator"; -import { deferred } from "../../../../support/async"; - -describe("createRuntimeSettingsCommitCoordinator", () => { - it("serializes one thread's commits across an A to B to A panel revision", async () => { - let currentRevision = 1; - let pending: RuntimeSettingsPatch = { model: "old" }; - const firstUpdate = deferred(); - const updateThreadSettings = vi.fn().mockReturnValueOnce(firstUpdate.promise).mockResolvedValueOnce(true); - const coordinator = createRuntimeSettingsCommitCoordinator({ - scopeIsCurrent: (scope) => scope.panelTargetRevision === currentRevision, - pendingPatch: () => pending, - updateThreadSettings, - commitPatch: () => { - pending = {}; - }, - reportError: vi.fn(), - }); - - const oldCommit = coordinator.commit({ threadId: "thread-a", panelTargetRevision: 1 }, { kind: "fields", update: { model: "old" } }); - await vi.waitFor(() => expect(updateThreadSettings).toHaveBeenCalledOnce()); - currentRevision = 3; - pending = { model: "latest" }; - const latestCommit = coordinator.commit( - { threadId: "thread-a", panelTargetRevision: 3 }, - { kind: "fields", update: { model: "latest" } }, - ); - await Promise.resolve(); - expect(updateThreadSettings).toHaveBeenCalledOnce(); - - firstUpdate.resolve(true); - await expect(oldCommit).resolves.toBe(false); - await vi.waitFor(() => expect(updateThreadSettings).toHaveBeenCalledTimes(2)); - await expect(latestCommit).resolves.toBe(true); - expect(updateThreadSettings).toHaveBeenNthCalledWith(2, "thread-a", { model: "latest" }); - }); -}); diff --git a/tests/features/chat/application/runtime/settings-actions.test.ts b/tests/features/chat/application/runtime/settings-actions.test.ts index 93b8622e..92434f15 100644 --- a/tests/features/chat/application/runtime/settings-actions.test.ts +++ b/tests/features/chat/application/runtime/settings-actions.test.ts @@ -440,6 +440,28 @@ describe("createChatRuntimeSettingsActions", () => { expect(messages).toEqual([]); }); + it("does not report stale turn-submission settings failures", async () => { + let state = chatStateFixture(); + state = chatStateWith(state, { activeThread: { id: "thread" } }); + const store = createChatStateStore(state); + const transport = settingsTransportFixture({ + updateThreadSettings: vi.fn().mockImplementation(async () => { + store.dispatch({ type: "active-thread/cleared" }); + throw new Error("nope"); + }), + }); + const messages: string[] = []; + const actions = runtimeActionsFixture(store, transport, messages); + + store.dispatch({ type: "runtime/model-requested", model: "gpt-5.5" }); + + await expect(actions.applyPendingThreadSettings()).resolves.toBe(false); + + expect(transport.updateThreadSettings).toHaveBeenCalledWith("thread", { model: "gpt-5.5" }); + expect(activeThreadId(store.getState())).toBeNull(); + expect(messages).toEqual([]); + }); + it("does not commit stale runtime updates after a newer pending intent replaces them", async () => { let state = chatStateFixture(); state = chatStateWith(state, { activeThread: { id: "thread" } }); @@ -527,7 +549,75 @@ describe("createChatRuntimeSettingsActions", () => { await expect(secondMutation).resolves.toBe(true); }); - it("coalesces a different-field intent behind the active settings update", async () => { + it("runs A1, B, then A2 in shared thread FIFO order and leaves A2 on the server", async () => { + const panelState = chatStateWith(chatStateFixture(), { activeThread: { id: "thread" } }); + const firstStore = createChatStateStore(panelState); + const secondStore = createChatStateStore(panelState); + const firstUpdate = deferred(true); + const order: string[] = []; + let serverModel: string | null = null; + const firstTransport = settingsTransportFixture({ + updateThreadSettings: vi.fn((_threadId, update) => { + order.push(`A:${update.model}`); + serverModel = update.model ?? null; + if (update.model === "a1") return firstUpdate.promise; + return Promise.resolve(true); + }), + }); + const secondTransport = settingsTransportFixture({ + updateThreadSettings: vi.fn((_threadId, update) => { + order.push(`B:${update.model}`); + serverModel = update.model ?? null; + return Promise.resolve(true); + }), + }); + const threadCommits = createKeyedOperationQueue(); + const firstActions = runtimeActionsFixture(firstStore, firstTransport, [], threadCommits); + const secondActions = runtimeActionsFixture(secondStore, secondTransport, [], threadCommits); + + const a1 = firstActions.requestModel("a1"); + await vi.waitFor(() => expect(firstTransport.updateThreadSettings).toHaveBeenCalledWith("thread", { model: "a1" })); + const b = secondActions.requestModel("b"); + const a2 = firstActions.requestModel("a2"); + expect(order).toEqual(["A:a1"]); + + firstUpdate.resolve(); + + await expect(a1).resolves.toBe(false); + await expect(b).resolves.toBe(true); + await expect(a2).resolves.toBe(true); + expect(order).toEqual(["A:a1", "B:b", "A:a2"]); + expect(serverModel).toBe("a2"); + expect(firstStore.getState().runtime.active.model).toBe("a2"); + }); + + it("settles newly requested settings inside its queued turn-submission command", async () => { + let state = chatStateFixture(); + state = chatStateWith(state, { activeThread: { id: "thread" } }); + const store = createChatStateStore(state); + const firstUpdate = deferred(true); + const transport = settingsTransportFixture({ + updateThreadSettings: vi + .fn() + .mockImplementationOnce(() => firstUpdate.promise) + .mockResolvedValueOnce(true), + }); + const actions = runtimeActionsFixture(store, transport, []); + + store.dispatch({ type: "runtime/model-requested", model: "gpt-old" }); + const settingsSettled = actions.applyPendingThreadSettings(); + await vi.waitFor(() => expect(transport.updateThreadSettings).toHaveBeenNthCalledWith(1, "thread", { model: "gpt-old" })); + + store.dispatch({ type: "runtime/model-requested", model: "gpt-new" }); + firstUpdate.resolve(); + + await vi.waitFor(() => expect(transport.updateThreadSettings).toHaveBeenNthCalledWith(2, "thread", { model: "gpt-new" })); + await expect(settingsSettled).resolves.toBe(true); + expect(store.getState().runtime.active.model).toBe("gpt-new"); + expect(store.getState().runtime.pending.model).toEqual({ kind: "unchanged" }); + }); + + it("serializes a different-field intent behind the active settings update", async () => { let state = chatStateFixture(); state = chatStateWith(state, { activeThread: { id: "thread" } }); const store = createChatStateStore(state);