obsidian-plugin/murashit_codex-panel
1
0
Fork 0
mirror of https://github.com/murashit/codex-panel.git synced 2026-07-22 06:57:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

refactor(chat): model scoped async operations

Browse source
This commit is contained in:
murashit 2026-07-17 20:57:19 +09:00
parent 716d6a2db5
commit 1eb074b361
35 changed files with 747 additions and 218 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

56
src/features/chat/app-server/transports/session-transports.ts
View file

@ -21,6 +21,7 @@ import { interruptTurn, startTurn, steerTurn } from "../../../../app-server/serv
import type { RuntimeSettingsPatch } from "../../../../domain/runtime/thread-settings";
import type { ThreadTurnsPage } from "../../../../domain/threads/history";
import type { RuntimeSettingsTransport } from "../../application/runtime/settings-transport";
import type { EffectOutcome } from "../../application/effect-outcome";
import type { EphemeralThreadTransport } from "../../application/threads/ephemeral-thread-transport";
import type { ThreadGoalReadTransport, ThreadGoalTransport } from "../../application/threads/goal-transport";
import type {
@ -93,7 +94,7 @@ export function createChatConnectedSessionTransports(host: ChatAppServerTranspor
function createChatThreadStartTransport(host: ChatCurrentAppServerTransportHost): ThreadStartTransport {
return {
startThread: (request) =>
withCurrentChatAppServerClient(host, async (client) => {
runCurrentChatAppServerEffect(host, async (client) => {
const response = await startThread(client, {
cwd: host.vaultPath,
serviceTier: request.serviceTier,
@ -155,24 +156,23 @@ function createChatThreadHistoryTransport(host: CurrentChatAppServerClientHost):
function createChatThreadResumeTransport(host: ChatAppServerTransportHost): ThreadResumeTransport {
return {
resumeThread: (threadId): Promise<ThreadResumeSnapshot | null> =>
withConnectedChatAppServerClient(host, (client) => resumeChatThread(client, threadId, host.vaultPath)),
ensureConnected: async () => (await host.connectedClient()) !== null,
resumeThread: (threadId): Promise<EffectOutcome<ThreadResumeSnapshot>> =>
runCurrentChatAppServerEffect(host, (client) => resumeChatThread(client, threadId, host.vaultPath)),
};
}
function createChatThreadMutationTransport(host: ChatAppServerTransportHost): ThreadMutationTransport {
return {
compactThread: async (threadId) => {
const result = await withConnectedChatAppServerClient(host, async (client) => {
ensureConnected: async () => (await host.connectedClient()) !== null,
compactThread: (threadId) =>
runCurrentChatAppServerEffect(host, async (client) => {
await compactThread(client, threadId);
return true;
});
return result ?? false;
},
}),
forkThread: (threadId, lastTurnId = null) =>
withConnectedChatAppServerClient(host, (client) => forkThread(client, threadId, host.vaultPath, lastTurnId)),
runCurrentChatAppServerEffect(host, (client) => forkThread(client, threadId, host.vaultPath, lastTurnId)),
rollbackThread: (threadId) =>
withConnectedChatAppServerClient(host, async (client): Promise<ThreadRollbackSnapshot> => {
runCurrentChatAppServerEffect(host, async (client): Promise<ThreadRollbackSnapshot> => {
const snapshot = await rollbackThread(client, threadId);
return {
thread: snapshot.thread,
@ -186,7 +186,7 @@ function createChatThreadMutationTransport(host: ChatAppServerTransportHost): Th
function createChatEphemeralThreadTransport(host: ChatAppServerTransportHost): EphemeralThreadTransport {
return {
forkEphemeralThread: (sourceThreadId) =>
withConnectedChatAppServerClient(host, async (client) => {
runCurrentChatAppServerEffect(host, async (client) => {
try {
const snapshot = await forkEphemeralThread(client, sourceThreadId, host.vaultPath);
return { kind: "ready" as const, ...snapshot };
@ -230,18 +230,12 @@ function createChatThreadGoalTransport(host: ConnectedChatAppServerClientHost):
ensureConnected: async () => (await host.connectedClient()) !== null,
readThreadGoal: (threadId) => readThreadGoalFromCurrentClient(host, threadId),
setThreadGoal: async (threadId, params) => {
const client = await host.connectedClient();
if (!client) return undefined;
const goal = await setThreadGoal(client, threadId, params);
return chatAppServerClientIsStale(host, client) ? undefined : goal;
return runCurrentChatAppServerEffect(host, (client) => setThreadGoal(client, threadId, params));
},
clearThreadGoal: async (threadId) => {
const result = await withConnectedChatAppServerClient(host, async (client) => {
clearThreadGoal: (threadId) =>
runCurrentChatAppServerEffect(host, async (client) => {
await clearThreadGoal(client, threadId);
return true;
});
return result ?? false;
},
}),
recordThreadGoalUserMessage: async (threadId, objective) => {
const result = await withCurrentChatAppServerClient(host, async (client) => {
await recordThreadGoalUserMessage(client, threadId, objective);
@ -256,14 +250,18 @@ function chatAppServerClientIsStale(host: CurrentChatAppServerClientHost, client
return host.currentClient() !== client;
}
async function withConnectedChatAppServerClient<T>(
host: ConnectedChatAppServerClientHost,
function runCurrentChatAppServerEffect<T>(
host: CurrentChatAppServerClientHost,
operation: (client: AppServerClient) => Promise<T>,
): Promise<T | null> {
const client = await host.connectedClient();
if (!client) return null;
const result = await operation(client);
return chatAppServerClientIsStale(host, client) ? null : result;
): Promise<EffectOutcome<T>> {
const client = host.currentClient();
if (!client) return Promise.resolve({ kind: "not-started" });
const effect = operation(client);
return effect.then((value) =>
chatAppServerClientIsStale(host, client)
? { kind: "completed-stale", value }
: { kind: "completed-current", value },
);
}
async function withCurrentChatAppServerClient<T>(

12
src/features/chat/application/connection/reconnect-actions.ts
View file

@ -1,4 +1,5 @@
import { activeThreadId, activeThreadState, type ChatConnectionPhase, panelThreadId } from "../state/root-reducer";
import { capturePanelTargetLease, panelTargetLeaseIsCurrent } from "../state/panel-target";
import type { ChatStateStore } from "../state/store";
const STATUS_RECONNECTING = "Reconnecting...";
@ -12,7 +13,7 @@ export interface ChatReconnectActionsHost {
setStatus: (statusText: string, phase?: ChatConnectionPhase) => void;
ensureConnected: () => Promise<void>;
isConnected: () => boolean;
resumeThread: (threadId: string) => Promise<void>;
resumeThread: (threadId: string) => Promise<boolean>;
addSystemMessage: (text: string) => void;
}
@ -21,6 +22,7 @@ export async function reconnectPanel(
target: { resumeThreadId: string | null; isCurrent?: () => boolean } | null = null,
): Promise<boolean> {
const currentState = host.stateStore.getState();
const panelTarget = capturePanelTargetLease(currentState);
const threadId = target
? target.resumeThreadId
: activeThreadState(currentState)?.lifetime?.kind === "ephemeral"
@ -36,11 +38,13 @@ export async function reconnectPanel(
host.setStatus(STATUS_RECONNECTING, { kind: "connecting" });
await host.ensureConnected();
if (!isCurrent() || !host.isConnected()) return false;
if (!isCurrent() || !host.isConnected() || !panelTargetLeaseIsCurrent(host.stateStore.getState(), panelTarget)) return false;
if (!threadId) return true;
try {
await host.resumeThread(threadId);
if (!isCurrent()) return false;
if (!(await host.resumeThread(threadId))) return false;
if (!isCurrent() || !panelTargetLeaseIsCurrent(host.stateStore.getState(), panelTarget)) {
return false;
}
return activeThreadId(host.stateStore.getState()) === threadId;
} catch (error) {
host.addSystemMessage(error instanceof Error ? error.message : String(error));

14
src/features/chat/application/effect-outcome.ts Normal file
View file

@ -0,0 +1,14 @@
export type EffectOutcome<T> =
| { readonly kind: "not-started" }
| { readonly kind: "completed-current"; readonly value: T }
| { readonly kind: "completed-stale"; readonly value: T };
export function effectCompleted<T>(outcome: EffectOutcome<T>): outcome is Exclude<EffectOutcome<T>, { kind: "not-started" }> {
return outcome.kind !== "not-started";
}
export function effectCompletedInCurrentContext<T>(
outcome: EffectOutcome<T>,
): outcome is Extract<EffectOutcome<T>, { kind: "completed-current" }> {
return outcome.kind === "completed-current";
}

10
src/features/chat/application/state/actions.ts
View file

@ -20,6 +20,7 @@ interface ResumedThreadActionParams {
preserveRequestedRuntimeSettings?: boolean;
serviceTierKnown?: boolean;
preservePendingSubmissionId?: string;
expectedPanelTargetRevision?: number;
}
interface ResumedThreadFromActiveRuntimeParams {
@ -41,6 +42,7 @@ interface ResumedThreadFromActiveRuntimeParams {
>;
listedThreads?: readonly Thread[];
items?: readonly ThreadStreamItem[];
expectedPanelTargetRevision?: number;
}
export interface ActiveThreadResumedAction extends RuntimePermissionState, RuntimePermissionKnownState {
@ -57,6 +59,7 @@ export interface ActiveThreadResumedAction extends RuntimePermissionState, Runti
listedThreads?: readonly Thread[];
preserveRequestedRuntimeSettings?: boolean;
preservePendingSubmissionId?: string;
expectedPanelTargetRevision?: number;
lifetime?:
| { readonly kind: "persistent" }
| { readonly kind: "ephemeral"; readonly sourceThreadId: string; readonly sourceThreadTitle: string | null };
@ -100,6 +103,7 @@ export interface ClearLocalTurnAction {
export interface ClearActiveThreadAction {
type: "active-thread/cleared";
expectedPanelTargetRevision?: number;
}
export interface ThreadListAppliedAction {
@ -151,6 +155,9 @@ export function resumedThreadActionFromActiveRuntime(params: ResumedThreadFromAc
serviceTierKnown: params.runtime.serviceTierKnown,
...(params.listedThreads ? { listedThreads: params.listedThreads } : {}),
...(params.items ? { items: params.items } : {}),
...(params.expectedPanelTargetRevision === undefined
? {}
: { expectedPanelTargetRevision: params.expectedPanelTargetRevision }),
});
}
@ -177,6 +184,9 @@ export function resumedThreadAction(params: ResumedThreadActionParams): ActiveTh
: {}),
...(params.preserveRequestedRuntimeSettings ? { preserveRequestedRuntimeSettings: true } : {}),
...(params.preservePendingSubmissionId ? { preservePendingSubmissionId: params.preservePendingSubmissionId } : {}),
...(params.expectedPanelTargetRevision === undefined
? {}
: { expectedPanelTargetRevision: params.expectedPanelTargetRevision }),
};
}

22
src/features/chat/application/state/panel-target.ts Normal file
View file

@ -0,0 +1,22 @@
import { panelThreadId, type ChatState } from "./root-reducer";
export type PanelTarget = { readonly kind: "empty" } | { readonly kind: "thread"; readonly threadId: string };
export interface PanelTargetLease {
readonly revision: number;
readonly target: PanelTarget;
}
export function capturePanelTargetLease(state: ChatState): PanelTargetLease {
const threadId = panelThreadId(state);
return {
revision: state.panelTargetRevision,
target: threadId ? { kind: "thread", threadId } : { kind: "empty" },
};
}
export function panelTargetLeaseIsCurrent(state: ChatState, lease: PanelTargetLease): boolean {
if (state.panelTargetRevision !== lease.revision) return false;
const threadId = panelThreadId(state);
return lease.target.kind === "empty" ? threadId === null : threadId === lease.target.threadId;
}

48
src/features/chat/application/state/root-reducer.ts
View file

@ -140,6 +140,7 @@ interface ChatStateShape {
connection: ChatConnectionState;
threadList: ChatThreadListState;
panelThread: ChatPanelThreadState;
panelTargetRevision: number;
runtime: ChatRuntimeState;
turn: ChatTurnState;
threadStream: ChatThreadStreamState;
@ -271,6 +272,7 @@ export function createChatState(): ChatState {
connection: initialConnectionState(),
threadList: initialThreadListState(),
panelThread: initialPanelThreadState(),
panelTargetRevision: 0,
runtime: initialChatRuntimeState(),
turn: initialTurnState(),
threadStream: initialThreadStreamState(),
@ -339,7 +341,13 @@ function reduceChatTransition(state: ChatState, action: ChatTransitionAction): C
case "connection/context-replaced":
return clearConnectionContextState(state);
case "active-thread/cleared":
return clearThreadScopedState(state);
if (
action.expectedPanelTargetRevision !== undefined &&
action.expectedPanelTargetRevision !== state.panelTargetRevision
) {
return state;
}
return clearThreadScopedState(state, { invalidatePanelTarget: true });
case "active-thread/resumed":
return reduceActiveThreadResumedTransition(state, action);
case "active-thread/settings-applied":
@ -412,8 +420,16 @@ function adoptPendingSteerItem(state: ChatThreadStreamState, item: ThreadStreamD
}
function reduceActiveThreadResumedTransition(state: ChatState, action: ActiveThreadResumedAction): ChatState {
if (
action.expectedPanelTargetRevision !== undefined &&
action.expectedPanelTargetRevision !== state.panelTargetRevision
) {
return state;
}
const runtimeBase = action.preserveRequestedRuntimeSettings ? state.runtime : initialChatRuntimeState();
const turnScopedState = clearTurnScopedState(state);
const nextPanelTargetRevision =
panelThreadId(state) === action.thread.id ? state.panelTargetRevision : state.panelTargetRevision + 1;
return patchChatState(turnScopedState, {
connection: {
...state.connection,
@ -435,6 +451,7 @@ function reduceActiveThreadResumedTransition(state: ChatState, action: ActiveThr
provenance: action.thread.provenance,
},
},
panelTargetRevision: nextPanelTargetRevision,
runtime: {
...runtimeBase,
active: {
@ -506,7 +523,7 @@ function reduceActiveThreadGoalSetTransition(state: ChatState, goal: ThreadGoal
}
function reduceRestoredThreadAppliedTransition(state: ChatState, threadId: string, fallbackTitle: string | null): ChatState {
const cleared = clearThreadScopedState(state);
const cleared = clearThreadScopedState(state, { invalidatePanelTarget: true });
return patchChatState(cleared, {
connection: { ...cleared.connection, statusText: "Thread ready to resume." },
panelThread: createAwaitingResumeThreadState(threadId, fallbackTitle),
@ -519,7 +536,7 @@ function reduceRestoredThreadRenamedTransition(state: ChatState, threadId: strin
}
function reduceViewStateClearedTransition(state: ChatState): ChatState {
const cleared = clearThreadScopedState(state);
const cleared = clearThreadScopedState(state, { invalidatePanelTarget: true });
return patchChatState(cleared, { connection: { ...cleared.connection, statusText: "Idle" } });
}
@ -531,6 +548,8 @@ function reduceTurnStartedTransition(state: ChatState, action: TurnStartedAction
if (!activeThread || activeThread.id !== action.threadId) return state;
return patchChatState(state, {
panelThread: { kind: "active", thread: activeThread },
panelTargetRevision:
panelThreadId(state) === action.threadId ? state.panelTargetRevision : state.panelTargetRevision + 1,
turn: { lifecycle },
connection: { ...state.connection, statusText: STATUS_TURN_RUNNING },
threadStream: action.items
@ -623,10 +642,17 @@ function clearTurnScopedState(state: ChatState): ChatState {
});
}
function clearThreadScopedState(state: ChatState): ChatState {
function clearThreadScopedState(
state: ChatState,
options: { invalidatePanelTarget?: boolean } = {},
): ChatState {
return clearTurnScopedState(
patchChatState(state, {
panelThread: initialPanelThreadState(),
panelTargetRevision:
options.invalidatePanelTarget || panelThreadId(state) !== null
? state.panelTargetRevision + 1
: state.panelTargetRevision,
runtime: initialChatRuntimeState(),
threadStream: initialThreadStreamState(),
pendingSubmission: null,
@ -639,8 +665,13 @@ function clearThreadScopedState(state: ChatState): ChatState {
function clearConnectionScopedState(state: ChatState): ChatState {
const cleared = clearTurnScopedState(state);
const ephemeralExpired = state.panelThread.kind === "active" && state.panelThread.thread.lifetime?.kind === "ephemeral";
const nextPanelThread = panelThreadAfterConnectionExit(state.panelThread);
return patchChatState(cleared, {
panelThread: panelThreadAfterConnectionExit(state.panelThread),
panelThread: nextPanelThread,
panelTargetRevision:
panelThreadIdForState(nextPanelThread) === panelThreadId(state)
? state.panelTargetRevision
: state.panelTargetRevision + 1,
runtime: initialChatRuntimeState(),
connection: {
...state.connection,
@ -659,6 +690,7 @@ function clearConnectionContextState(state: ChatState): ChatState {
const cleared = clearConnectionScopedState(state);
return patchChatState(cleared, {
panelThread: initialPanelThreadState(),
panelTargetRevision: state.panelTargetRevision + 1,
connection: {
...cleared.connection,
runtimeConfig: null,
@ -680,11 +712,17 @@ function panelThreadAfterConnectionExit(panelThread: ChatPanelThreadState): Chat
return createAwaitingResumeThreadState(panelThread.thread.id, panelThread.thread.title ?? null, panelThread.thread.provenance);
}
function panelThreadIdForState(panelThread: ChatPanelThreadState): string | null {
if (panelThread.kind === "awaiting-resume") return panelThread.threadId;
return panelThread.kind === "active" ? panelThread.thread.id : null;
}
function reduceChatSlices(state: ChatState, action: ChatSliceAction): ChatState {
return patchChatState(state, {
connection: reduceConnectionSlice(state.connection, action),
threadList: reduceThreadListSlice(state.threadList, action),
panelThread: reducePanelThreadSlice(state.panelThread, action),
panelTargetRevision: state.panelTargetRevision,
runtime: reduceRuntimeSlice(state.runtime, action),
turn: state.turn,
pendingSubmission: state.pendingSubmission,

22
src/features/chat/application/threads/ephemeral-thread-lifecycle.ts
View file

@ -1,5 +1,7 @@
import { ephemeralThreadActivatedAction } from "../state/actions";
import { effectCompletedInCurrentContext } from "../effect-outcome";
import { activeThreadState } from "../state/root-reducer";
import { capturePanelTargetLease, panelTargetLeaseIsCurrent } from "../state/panel-target";
import type { ChatStateStore } from "../state/store";
import { activeTurnId, chatTurnBusy } from "../turns/turn-state";
import type { EphemeralThreadTransport } from "./ephemeral-thread-transport";
@ -41,16 +43,19 @@ export function createEphemeralThreadLifecycle(host: EphemeralThreadLifecycleHos
return {
async open(input): Promise<boolean> {
const generation = ++openGeneration;
const panelTarget = capturePanelTargetLease(host.stateStore.getState());
if (!(await host.ensureConnected())) return false;
const result = await host.transport.forkEphemeralThread(input.sourceThreadId);
if (!result) return false;
if (disposed || generation !== openGeneration || !panelTargetLeaseIsCurrent(host.stateStore.getState(), panelTarget)) return false;
const effect = await host.transport.forkEphemeralThread(input.sourceThreadId);
if (!effectCompletedInCurrentContext(effect)) return false;
const result = effect.value;
if (result.kind === "cleanup-required") {
cleanupRequiredThreadIds.add(result.threadId);
host.addSystemMessage("Could not prepare the side chat. Cleanup will be retried when this view closes.");
return false;
}
const snapshot = result;
if (disposed || generation !== openGeneration) {
if (disposed || generation !== openGeneration || !panelTargetLeaseIsCurrent(host.stateStore.getState(), panelTarget)) {
try {
await host.transport.unsubscribeEphemeralThread(snapshot.activation.thread.id);
} catch {
@ -63,29 +68,36 @@ export function createEphemeralThreadLifecycle(host: EphemeralThreadLifecycleHos
response: snapshot.activation,
sourceThreadId: input.sourceThreadId,
sourceThreadTitle: input.sourceThreadTitle,
expectedPanelTargetRevision: panelTarget.revision,
}),
);
if (activeThreadState(host.stateStore.getState())?.id !== snapshot.activation.thread.id) return false;
host.notifyActiveThreadIdentityChanged();
return true;
},
async prepareForPersistentNavigation(): Promise<boolean> {
const state = host.stateStore.getState();
if (activeThreadState(state)?.lifetime?.kind !== "ephemeral") return true;
const active = activeThreadState(state);
if (active?.lifetime?.kind !== "ephemeral") return true;
const panelTarget = capturePanelTargetLease(state);
if (chatTurnBusy(state)) {
host.addSystemMessage("Finish or interrupt the current turn before switching threads.");
return false;
}
try {
if (!(await unsubscribeActiveEphemeralThread())) {
if (!panelTargetLeaseIsCurrent(host.stateStore.getState(), panelTarget)) return false;
host.addSystemMessage("Could not discard the side chat. Try again before switching threads.");
return false;
}
} catch (error) {
if (!panelTargetLeaseIsCurrent(host.stateStore.getState(), panelTarget)) return false;
host.addSystemMessage(error instanceof Error ? error.message : String(error));
return false;
}
host.stateStore.dispatch({ type: "active-thread/cleared" });
if (!panelTargetLeaseIsCurrent(host.stateStore.getState(), panelTarget)) return false;
host.stateStore.dispatch({ type: "active-thread/cleared", expectedPanelTargetRevision: panelTarget.revision });
host.notifyActiveThreadIdentityChanged();
return true;
},

3
src/features/chat/application/threads/ephemeral-thread-transport.ts
View file

@ -1,10 +1,11 @@
import type { ThreadActivationSnapshot } from "../../../../domain/threads/activation";
import type { EffectOutcome } from "../effect-outcome";
type EphemeralThreadForkResult =
| { kind: "ready"; activation: ThreadActivationSnapshot; sourceThreadId: string }
| { kind: "cleanup-required"; threadId: string };
export interface EphemeralThreadTransport {
forkEphemeralThread(sourceThreadId: string): Promise<EphemeralThreadForkResult | null>;
forkEphemeralThread(sourceThreadId: string): Promise<EffectOutcome<EphemeralThreadForkResult>>;
unsubscribeEphemeralThread(threadId: string): Promise<boolean>;
}

48
src/features/chat/application/threads/goal-actions.ts
View file

@ -2,8 +2,14 @@ import type { ThreadGoal, ThreadGoalStatus, ThreadGoalUpdate } from "../../../..
import { goalChangeItem } from "../../domain/thread-stream/factories/goal-items";
import type { GoalThreadStreamItem } from "../../domain/thread-stream/items";
import type { LocalIdSource } from "../local-id-source";
import { effectCompletedInCurrentContext } from "../effect-outcome";
import { activePanelOperationDecision } from "../panel-operation-policy";
import { activeThreadId, activeThreadState } from "../state/root-reducer";
import {
capturePanelTargetLease,
type PanelTargetLease,
panelTargetLeaseIsCurrent,
} from "../state/panel-target";
import type { ChatStateStore } from "../state/store";
import type { ThreadGoalReadTransport, ThreadGoalTransport } from "./goal-transport";
@ -81,12 +87,13 @@ export function createGoalActions(host: GoalActionsHost): GoalActions {
}
async function syncThreadGoal(host: ThreadGoalSyncHost, threadId: string): Promise<void> {
const panelTarget = capturePanelTargetLease(host.stateStore.getState());
try {
const goal = await host.goalTransport.readThreadGoal(threadId);
if (goal === undefined) return;
applyGoalIfActive(host, threadId, goal, { reportChange: false });
applyGoalIfActive(host, threadId, goal, { reportChange: false, panelTarget });
} catch (error) {
addThreadScopedSystemMessage(host, threadId, `Could not load thread goal: ${errorMessage(error)}`);
addThreadScopedSystemMessage(host, threadId, `Could not load thread goal: ${errorMessage(error)}`, panelTarget);
}
}
@ -138,24 +145,28 @@ function setGoalStatus(host: GoalActionsHost, threadId: string, status: ThreadGo
async function clearGoal(host: GoalActionsHost, threadId: string): Promise<boolean> {
if (!(await prepareGoalMutation(host)) || !goalMutationTargetsActiveThread(host, threadId)) return false;
const panelTarget = capturePanelTargetLease(host.stateStore.getState());
try {
if (!(await host.goalTransport.clearThreadGoal(threadId))) return false;
applyGoalIfActive(host, threadId, null, { reportChange: true });
return true;
if (!(await host.goalTransport.ensureConnected()) || !goalMutationScopeIsCurrent(host, threadId, panelTarget)) return false;
const effect = await host.goalTransport.clearThreadGoal(threadId);
if (!effectCompletedInCurrentContext(effect)) return false;
return applyGoalIfActive(host, threadId, null, { reportChange: true, panelTarget });
} catch (error) {
addThreadScopedSystemMessage(host, threadId, errorMessage(error));
addThreadScopedSystemMessage(host, threadId, errorMessage(error), panelTarget);
return false;
}
}
async function setGoal(host: GoalActionsHost, threadId: string, params: ThreadGoalUpdate): Promise<boolean> {
if (!(await prepareGoalMutation(host)) || !goalMutationTargetsActiveThread(host, threadId)) return false;
const panelTarget = capturePanelTargetLease(host.stateStore.getState());
try {
const goal = await host.goalTransport.setThreadGoal(threadId, params);
if (goal === undefined) return false;
return applyGoalIfActive(host, threadId, goal, { reportChange: true });
if (!(await host.goalTransport.ensureConnected()) || !goalMutationScopeIsCurrent(host, threadId, panelTarget)) return false;
const effect = await host.goalTransport.setThreadGoal(threadId, params);
if (!effectCompletedInCurrentContext(effect)) return false;
return applyGoalIfActive(host, threadId, effect.value, { reportChange: true, panelTarget });
} catch (error) {
addThreadScopedSystemMessage(host, threadId, errorMessage(error));
addThreadScopedSystemMessage(host, threadId, errorMessage(error), panelTarget);
return false;
}
}
@ -164,9 +175,10 @@ function applyGoalIfActive(
host: ThreadGoalSyncHost,
threadId: string,
goal: ThreadGoal | null,
options: { reportChange: boolean },
options: { reportChange: boolean; panelTarget?: PanelTargetLease },
): boolean {
const state = host.stateStore.getState();
if (options.panelTarget && !panelTargetLeaseIsCurrent(state, options.panelTarget)) return false;
const activeThread = activeThreadState(state);
if (!activeThread || activeThread.id !== threadId) return false;
const item = options.reportChange ? goalChangeItem(host.localItemIds.next("goal"), activeThread.goal, goal) : null;
@ -254,11 +266,21 @@ async function recordGoalUserMessage(host: GoalActionsHost, threadId: string, ob
}
}
function addThreadScopedSystemMessage(host: ThreadGoalSyncHost, threadId: string, text: string): void {
if (activeThreadId(host.stateStore.getState()) !== threadId) return;
function addThreadScopedSystemMessage(
host: ThreadGoalSyncHost,
threadId: string,
text: string,
panelTarget?: PanelTargetLease,
): void {
const state = host.stateStore.getState();
if ((panelTarget && !panelTargetLeaseIsCurrent(state, panelTarget)) || activeThreadId(state) !== threadId) return;
host.addSystemMessage(text);
}
function goalMutationScopeIsCurrent(host: GoalActionsHost, threadId: string, panelTarget: PanelTargetLease): boolean {
return panelTargetLeaseIsCurrent(host.stateStore.getState(), panelTarget) && goalMutationTargetsActiveThread(host, threadId);
}
function errorMessage(error: unknown): string {
return error instanceof Error ? error.message : String(error);
}

5
src/features/chat/application/threads/goal-transport.ts
View file

@ -1,12 +1,13 @@
import type { ThreadGoal, ThreadGoalUpdate } from "../../../../domain/threads/goal";
import type { EffectOutcome } from "../effect-outcome";
export interface ThreadGoalReadTransport {
readThreadGoal(threadId: string): Promise<ThreadGoal | null | undefined>;
}
export interface ThreadGoalTransport extends ThreadGoalReadTransport {
setThreadGoal(threadId: string, params: ThreadGoalUpdate): Promise<ThreadGoal | null | undefined>;
clearThreadGoal(threadId: string): Promise<boolean>;
setThreadGoal(threadId: string, params: ThreadGoalUpdate): Promise<EffectOutcome<ThreadGoal | null>>;
clearThreadGoal(threadId: string): Promise<EffectOutcome<void>>;
recordThreadGoalUserMessage(threadId: string, objective: string): Promise<boolean>;
ensureConnected(): Promise<boolean>;
}

25
src/features/chat/application/threads/restoration-controller.ts
View file

@ -1,4 +1,9 @@
import { awaitingResumeThreadState } from "../state/root-reducer";
import { activeThreadId, awaitingResumeThreadState } from "../state/root-reducer";
import {
capturePanelTargetLease,
type PanelTargetLease,
panelTargetLeaseIsCurrent,
} from "../state/panel-target";
import type { ChatStateStore } from "../state/store";
export interface RestorationControllerHost {
@ -8,7 +13,7 @@ export interface RestorationControllerHost {
export type RestoredThreadLoader = (threadId: string) => Promise<void>;
export class RestorationController {
private loading: { threadId: string; promise: Promise<void> } | null = null;
private loading: { threadId: string; panelTarget: PanelTargetLease; promise: Promise<void> } | null = null;
constructor(private readonly host: RestorationControllerHost) {}
@ -19,27 +24,29 @@ export class RestorationController {
async ensureLoaded(loadThread: RestoredThreadLoader): Promise<boolean> {
const restoredThread = awaitingResumeThreadState(this.host.stateStore.getState());
if (!restoredThread) return true;
if (this.loading?.threadId === restoredThread.threadId) {
await this.loading.promise;
return this.restorationLoaded();
const activeLoading = this.loading;
if (activeLoading?.threadId === restoredThread.threadId) {
await activeLoading.promise;
return this.restorationLoaded(activeLoading.panelTarget, restoredThread.threadId);
}
const threadId = restoredThread.threadId;
const loading = { threadId, promise: loadThread(threadId) };
const loading = { threadId, panelTarget: capturePanelTargetLease(this.host.stateStore.getState()), promise: loadThread(threadId) };
this.loading = loading;
try {
await loading.promise;
} finally {
if (this.loading === loading) this.loading = null;
}
return this.restorationLoaded();
return this.restorationLoaded(loading.panelTarget, threadId);
}
isPending(threadId: string): boolean {
return awaitingResumeThreadState(this.host.stateStore.getState())?.threadId === threadId;
}
private restorationLoaded(): boolean {
return awaitingResumeThreadState(this.host.stateStore.getState()) === null;
private restorationLoaded(panelTarget: PanelTargetLease, threadId: string): boolean {
const state = this.host.stateStore.getState();
return panelTargetLeaseIsCurrent(state, panelTarget) && activeThreadId(state) === threadId;
}
}

78
src/features/chat/application/threads/resume-actions.ts
View file

@ -1,6 +1,12 @@
import type { ThreadTokenUsage } from "../../../../domain/runtime/metrics";
import { effectCompletedInCurrentContext } from "../effect-outcome";
import { resumedThreadAction } from "../state/actions";
import { activeThreadState } from "../state/root-reducer";
import {
capturePanelTargetLease,
type PanelTargetLease,
panelTargetLeaseIsCurrent,
} from "../state/panel-target";
import type { ChatStateStore } from "../state/store";
import { threadStreamIsEmpty } from "../state/thread-stream";
import type { HistoryController } from "./history-controller";
@ -23,65 +29,85 @@ export interface ResumeActionsHost {
}
export interface ResumeActions {
resumeThread(threadId: string): Promise<void>;
resumeThread(threadId: string, intent?: ActiveChatResume): Promise<boolean>;
}
export function createResumeActions(host: ResumeActionsHost): ResumeActions {
return {
resumeThread: (threadId) => resumeThread(host, threadId),
resumeThread: (threadId, intent) => resumeThread(host, threadId, intent),
};
}
async function resumeThread(host: ResumeActionsHost, threadId: string): Promise<void> {
async function resumeThread(host: ResumeActionsHost, threadId: string, intent?: ActiveChatResume): Promise<boolean> {
if (!canSwitchToThread(host.stateStore.getState(), threadId)) {
host.addSystemMessage("Finish or interrupt the current turn before switching threads.");
return;
return false;
}
const resume = host.resumeWork.begin(threadId);
const resume = intent ?? host.resumeWork.begin(threadId);
if (resume.threadId !== threadId || host.resumeWork.isStale(resume)) return false;
const initialPanelTarget = capturePanelTargetLease(host.stateStore.getState());
host.history.invalidate();
try {
const response = await host.resumeTransport.resumeThread(threadId);
if (!response) return;
if (isStaleResume(host, resume)) return;
applyResumedThread(host, response);
recoverResumedThreadTokenUsage(host, response.activation.thread.id, response.rolloutPath, resume);
if (response.initialHistoryPage) {
host.history.applyLatestPage(response.activation.thread.id, response.initialHistoryPage);
if (!(await host.resumeTransport.ensureConnected())) return false;
if (isStaleResume(host, resume, initialPanelTarget)) return false;
const effect = await host.resumeTransport.resumeThread(threadId);
if (!effectCompletedInCurrentContext(effect)) return false;
if (isStaleResume(host, resume, initialPanelTarget)) return false;
const adoptedPanelTarget = applyResumedThread(host, effect.value, initialPanelTarget.revision);
if (!adoptedPanelTarget) return false;
recoverResumedThreadTokenUsage(host, effect.value.activation.thread.id, effect.value.rolloutPath, resume, adoptedPanelTarget);
if (effect.value.initialHistoryPage) {
host.history.applyLatestPage(effect.value.activation.thread.id, effect.value.initialHistoryPage);
} else {
await host.history.loadLatest(response.activation.thread.id);
await host.history.loadLatest(effect.value.activation.thread.id);
}
if (isStaleResume(host, resume)) return;
await host.syncThreadGoal(response.activation.thread.id);
if (isStaleResume(host, resume)) return;
if (isStaleResume(host, resume, adoptedPanelTarget)) return false;
await host.syncThreadGoal(effect.value.activation.thread.id);
if (isStaleResume(host, resume, adoptedPanelTarget)) return false;
const renderFallbackMessage = threadStreamIsEmpty(host.stateStore.getState().threadStream);
if (renderFallbackMessage) {
host.addSystemMessage(`Resumed thread ${response.activation.thread.id}`);
host.addSystemMessage(`Resumed thread ${effect.value.activation.thread.id}`);
}
host.refreshLiveState();
return true;
} catch (error) {
if (isStaleResume(host, resume)) return;
if (isStaleResume(host, resume, initialPanelTarget)) return false;
host.addSystemMessage(error instanceof Error ? error.message : String(error));
return false;
}
}
function applyResumedThread(host: ResumeActionsHost, response: ThreadResumeSnapshot): void {
host.stateStore.dispatch(
function applyResumedThread(
host: ResumeActionsHost,
response: ThreadResumeSnapshot,
expectedPanelTargetRevision: number,
): PanelTargetLease | null {
const state = host.stateStore.dispatch(
resumedThreadAction({
response: response.activation,
listedThreads: host.stateStore.getState().threadList.listedThreads,
expectedPanelTargetRevision,
}),
);
if (activeThreadState(state)?.id !== response.activation.thread.id) return null;
host.resetThreadTurnPresence(false);
host.notifyActiveThreadIdentityChanged();
return capturePanelTargetLease(state);
}
function recoverResumedThreadTokenUsage(host: ResumeActionsHost, threadId: string, path: string | null, resume: ActiveChatResume): void {
function recoverResumedThreadTokenUsage(
host: ResumeActionsHost,
threadId: string,
path: string | null,
resume: ActiveChatResume,
panelTarget: PanelTargetLease,
): void {
if (!path || !host.recoverTokenUsageFromRollout) return;
void host
.recoverTokenUsageFromRollout(path)
.then((tokenUsage) => {
if (!tokenUsage || isStaleResume(host, resume)) return;
if (!tokenUsage || isStaleResume(host, resume, panelTarget)) return;
const state = host.stateStore.getState();
const activeThread = activeThreadState(state);
if (!activeThread || activeThread.id !== threadId || activeThread.tokenUsage !== null) return;
@ -91,6 +117,10 @@ function recoverResumedThreadTokenUsage(host: ResumeActionsHost, threadId: strin
.catch(() => undefined);
}
function isStaleResume(host: ResumeActionsHost, resume: ActiveChatResume): boolean {
return host.resumeWork.isStale(resume) || host.closing();
function isStaleResume(host: ResumeActionsHost, resume: ActiveChatResume, panelTarget?: PanelTargetLease): boolean {
return (
host.resumeWork.isStale(resume) ||
host.closing() ||
Boolean(panelTarget && !panelTargetLeaseIsCurrent(host.stateStore.getState(), panelTarget))
);
}

8
src/features/chat/application/threads/resume-work.ts
View file

@ -1,11 +1,11 @@
export type ChatResumeLifecycleState = { kind: "idle" } | { kind: "resuming"; threadId: string };
export type ChatResumeLifecycleState = { kind: "idle" } | { kind: "resuming"; threadId: string | null };
export type ActiveChatResume = Extract<ChatResumeLifecycleState, { kind: "resuming" }>;
type ChatResumeLifecycleEvent = { type: "started"; resume: ActiveChatResume } | { type: "invalidated" };
export class ChatResumeWorkTracker {
private state: ChatResumeLifecycleState = { kind: "idle" };
begin(threadId: string): ActiveChatResume {
begin(threadId: string | null): ActiveChatResume {
const resume: ActiveChatResume = { kind: "resuming", threadId };
this.state = transitionChatResumeLifecycle(this.state, { type: "started", resume });
return resume;
@ -18,6 +18,10 @@ export class ChatResumeWorkTracker {
isStale(resume: ActiveChatResume): boolean {
return this.state !== resume;
}
isCurrent(resume: ActiveChatResume): boolean {
return this.state === resume;
}
}
function transitionChatResumeLifecycle(state: ChatResumeLifecycleState, event: ChatResumeLifecycleEvent): ChatResumeLifecycleState {

4
src/features/chat/application/threads/thread-loading-transport.ts
View file

@ -1,5 +1,6 @@
import type { ThreadActivationSnapshot } from "../../../../domain/threads/activation";
import type { ThreadStreamItem } from "../../domain/thread-stream/items";
import type { EffectOutcome } from "../effect-outcome";
export interface ThreadHistoryPage {
items: ThreadStreamItem[];
@ -18,5 +19,6 @@ export interface ThreadResumeSnapshot {
}
export interface ThreadResumeTransport {
resumeThread(threadId: string): Promise<ThreadResumeSnapshot | null>;
ensureConnected(): Promise<boolean>;
resumeThread(threadId: string): Promise<EffectOutcome<ThreadResumeSnapshot>>;
}

44
src/features/chat/application/threads/thread-management-actions.ts
View file

@ -1,8 +1,14 @@
import { inheritedForkThreadName, type Thread } from "../../../../domain/threads/model";
import { activeThreadRuntimeState } from "../../domain/runtime/state";
import { effectCompletedInCurrentContext } from "../effect-outcome";
import { type ActivePanelOperation, activePanelOperationDecision } from "../panel-operation-policy";
import { resumedThreadActionFromActiveRuntime } from "../state/actions";
import { activeThreadId, type ChatAction, type ChatState } from "../state/root-reducer";
import {
capturePanelTargetLease,
type PanelTargetLease,
panelTargetLeaseIsCurrent,
} from "../state/panel-target";
import type { ChatStateStore } from "../state/store";
import { threadStreamRollbackCandidate, threadStreamTurnsAfterTurnId } from "../state/thread-stream";
import { chatTurnBusy } from "../turns/turn-state";
@ -46,6 +52,7 @@ interface ThreadManagementPanelScope {
targetThreadId: string;
initialActiveThreadId: string | null;
initialTurnLifecycle: ChatState["turn"]["lifecycle"];
panelTarget: PanelTargetLease;
}
export function createThreadManagementActions(host: ThreadManagementActionsHost): ThreadManagementActions {
@ -73,11 +80,15 @@ async function compactThread(host: ThreadManagementActionsHost, threadId: string
if (activePanelOperationBlocked(host, threadId, "compact")) return;
const scope = captureThreadManagementPanelScope(host, threadId);
try {
if (!(await host.threadTransport.compactThread(threadId))) return;
if (!(await host.threadTransport.ensureConnected())) return;
if (!threadManagementScopeStillTargetsOriginalPanel(host, scope)) return;
const effect = await host.threadTransport.compactThread(threadId);
if (!effectCompletedInCurrentContext(effect)) return;
if (!threadManagementScopeStillTargetsOriginalPanel(host, scope)) return;
host.addSystemMessage(STATUS_COMPACTION_REQUESTED);
host.setStatus(STATUS_COMPACTION_REQUESTED);
} catch (error) {
if (!threadManagementScopeStillTargetsOriginalPanel(host, scope)) return;
host.addSystemMessage(error instanceof Error ? error.message : String(error));
}
}
@ -125,8 +136,11 @@ async function forkThreadFromTurn(
try {
const sourceName = inheritedForkThreadName(threadId, threadManagementState(host).threadList.listedThreads);
const forkedThread = turnId ? await host.threadTransport.forkThread(threadId, turnId) : await host.threadTransport.forkThread(threadId);
if (!forkedThread) return;
if (!(await host.threadTransport.ensureConnected())) return;
if (!threadManagementScopeStillTargetsOriginalPanel(host, scope)) return;
const effect = turnId ? await host.threadTransport.forkThread(threadId, turnId) : await host.threadTransport.forkThread(threadId);
if (!effectCompletedInCurrentContext(effect)) return;
const forkedThread = effect.value;
const forkedThreadId = forkedThread.id;
host.recordForkedThread(forkedThread);
if (!threadManagementScopeStillTargetsOriginalPanel(host, scope)) return;
@ -134,6 +148,7 @@ async function forkThreadFromTurn(
try {
if (!(await host.operations.renameThread(forkedThreadId, sourceName))) return;
} catch (error) {
if (!threadManagementScopeStillTargetsOriginalPanel(host, scope)) return;
const message = error instanceof Error ? error.message : String(error);
host.addSystemMessage(`Forked thread ${forkedThreadId}, but could not copy the source thread name: ${message}`);
}
@ -143,6 +158,7 @@ async function forkThreadFromTurn(
try {
await host.openThreadInCurrentPanel(forkedThreadId);
} catch (error) {
if (!threadManagementScopeStillTargetsOriginalPanel(host, scope)) return;
const message = error instanceof Error ? error.message : String(error);
host.addSystemMessage(`Archived thread ${threadId}, but could not open forked thread ${forkedThreadId}: ${message}`);
}
@ -151,10 +167,12 @@ async function forkThreadFromTurn(
try {
await host.openThreadInNewView(forkedThreadId);
} catch (error) {
if (!threadManagementScopeStillTargetsOriginalPanel(host, scope)) return;
const message = error instanceof Error ? error.message : String(error);
host.addSystemMessage(`Forked thread ${forkedThreadId}, but could not open it in a new panel: ${message}`);
}
} catch (error) {
if (!threadManagementScopeStillTargetsOriginalPanel(host, scope)) return;
host.addSystemMessage(error instanceof Error ? error.message : String(error));
}
}
@ -186,8 +204,11 @@ async function rollbackThread(host: ThreadManagementActionsHost, threadId: strin
try {
host.setStatus(STATUS_ROLLBACK_STARTING);
const snapshot = await host.threadTransport.rollbackThread(threadId);
if (!snapshot) return;
if (!(await host.threadTransport.ensureConnected())) return;
if (!threadManagementScopeStillTargetsPanel(host, scope)) return;
const effect = await host.threadTransport.rollbackThread(threadId);
if (!effectCompletedInCurrentContext(effect)) return;
const snapshot = effect.value;
if (!threadManagementScopeStillTargetsPanel(host, scope)) return;
threadManagementDispatch(
host,
@ -196,6 +217,7 @@ async function rollbackThread(host: ThreadManagementActionsHost, threadId: strin
cwd: snapshot.cwd,
runtime: activeThreadRuntimeState(threadManagementState(host).runtime),
listedThreads: threadManagementState(host).threadList.listedThreads,
expectedPanelTargetRevision: scope.panelTarget.revision,
}),
);
threadManagementDispatch(host, {
@ -210,6 +232,7 @@ async function rollbackThread(host: ThreadManagementActionsHost, threadId: strin
host.notifyActiveThreadIdentityChanged();
await host.refreshAfterThreadMutation();
} catch (error) {
if (!threadManagementScopeStillTargetsPanel(host, scope)) return;
host.addSystemMessage(error instanceof Error ? error.message : String(error));
host.setStatus(STATUS_ROLLBACK_FAILED);
}
@ -237,15 +260,22 @@ function captureThreadManagementPanelScope(host: ThreadManagementActionsHost, ta
targetThreadId,
initialActiveThreadId: activeThreadId(threadManagementState(host)),
initialTurnLifecycle: threadManagementState(host).turn.lifecycle,
panelTarget: capturePanelTargetLease(threadManagementState(host)),
};
}
function threadManagementScopeStillTargetsPanel(host: ThreadManagementActionsHost, scope: ThreadManagementPanelScope): boolean {
const state = threadManagementState(host);
return activeThreadId(state) === scope.targetThreadId && state.turn.lifecycle === scope.initialTurnLifecycle;
return (
panelTargetLeaseIsCurrent(state, scope.panelTarget) &&
activeThreadId(state) === scope.targetThreadId &&
state.turn.lifecycle === scope.initialTurnLifecycle
);
}
function threadManagementScopeStillTargetsOriginalPanel(host: ThreadManagementActionsHost, scope: ThreadManagementPanelScope): boolean {
const state = threadManagementState(host);
if (!panelTargetLeaseIsCurrent(state, scope.panelTarget)) return false;
if (!scope.initialActiveThreadId) return true;
return scope.initialActiveThreadId === scope.targetThreadId && activeThreadId(threadManagementState(host)) === scope.targetThreadId;
return scope.initialActiveThreadId === scope.targetThreadId && activeThreadId(state) === scope.targetThreadId;
}

8
src/features/chat/application/threads/thread-mutation-transport.ts
View file

@ -1,5 +1,6 @@
import type { Thread } from "../../../../domain/threads/model";
import type { ThreadStreamItem } from "../../domain/thread-stream/items";
import type { EffectOutcome } from "../effect-outcome";
export interface ThreadRollbackSnapshot {
thread: Thread;
@ -8,7 +9,8 @@ export interface ThreadRollbackSnapshot {
}
export interface ThreadMutationTransport {
compactThread(threadId: string): Promise<boolean>;
forkThread(threadId: string, lastTurnId?: string | null): Promise<Thread | null>;
rollbackThread(threadId: string): Promise<ThreadRollbackSnapshot | null>;
ensureConnected(): Promise<boolean>;
compactThread(threadId: string): Promise<EffectOutcome<void>>;
forkThread(threadId: string, lastTurnId?: string | null): Promise<EffectOutcome<Thread>>;
rollbackThread(threadId: string): Promise<EffectOutcome<ThreadRollbackSnapshot>>;
}

13
src/features/chat/application/threads/thread-navigation-actions.ts
View file

@ -3,6 +3,7 @@ import type { ChatStateStore } from "../state/store";
import { chatTurnBusy } from "../turns/turn-state";
import type { ActiveThreadIdentitySync } from "./active-thread-identity-sync";
import type { PersistentNavigationLifecycle } from "./persistent-navigation-lifecycle";
import type { ActiveChatResume, ChatResumeWorkTracker } from "./resume-work";
import { canSwitchToThread } from "./thread-switching";
export interface ThreadNavigationActionsHost {
@ -10,7 +11,8 @@ export interface ThreadNavigationActionsHost {
identity: ActiveThreadIdentitySync;
closeForThreadSelection: () => void;
focusThreadInOpenView: (threadId: string) => Promise<boolean>;
resumeThread: (threadId: string) => Promise<void>;
resumeThread: (threadId: string, intent: ActiveChatResume) => Promise<boolean>;
resumeWork: ChatResumeWorkTracker;
addSystemMessage: (text: string) => void;
focusComposer: () => void;
navigation: PersistentNavigationLifecycle;
@ -28,12 +30,14 @@ export function createThreadNavigationActions(host: ThreadNavigationActionsHost)
host.addSystemMessage("Finish or interrupt the current turn before switching threads.");
return;
}
const intent = host.resumeWork.begin(threadId);
host.closeForThreadSelection();
if (await host.focusThreadInOpenView(threadId)) return;
if (!host.resumeWork.isCurrent(intent)) return;
const preparation = await host.navigation.prepareForPersistentNavigation(threadId);
if (!preparation) return;
await host.resumeThread(threadId);
if (!preparation || !host.resumeWork.isCurrent(intent)) return;
if (!(await host.resumeThread(threadId, intent)) || !host.resumeWork.isCurrent(intent)) return;
await host.navigation.completePersistentNavigation(preparation);
};
@ -41,7 +45,8 @@ export function createThreadNavigationActions(host: ThreadNavigationActionsHost)
async startNewThread(): Promise<void> {
const state = host.stateStore.getState();
if (chatTurnBusy(state) && activeThreadState(state)?.provenance?.kind !== "subagent") return;
if (!(await host.navigation.prepareForPersistentNavigation(null))) return;
const intent = host.resumeWork.begin(null);
if (!(await host.navigation.prepareForPersistentNavigation(null)) || !host.resumeWork.isCurrent(intent)) return;
host.identity.clearActiveThreadIdentity();
host.stateStore.dispatch({ type: "ui/panel-set", panel: null });

49
src/features/chat/application/threads/thread-start-actions.ts
View file

@ -2,15 +2,18 @@ import { runtimeConfigOrDefault } from "../../../../domain/runtime/config";
import type { Thread } from "../../../../domain/threads/model";
import type { RuntimeSnapshot } from "../../domain/runtime/snapshot";
import { permissionProfileRequestForThreadStart, serviceTierRequestForThreadStart } from "../../domain/runtime/thread-settings-patch";
import { effectCompleted, effectCompletedInCurrentContext } from "../effect-outcome";
import { resumedThreadAction } from "../state/actions";
import { capturePanelTargetLease, panelTargetLeaseIsCurrent } from "../state/panel-target";
import { pendingSubmissionMatches } from "../state/pending-submission";
import { activeThreadId, type ChatState } from "../state/root-reducer";
import type { ChatStateStore } from "../state/store";
import type { ThreadStartTransport } from "./thread-start-transport";
interface StartedThreadSummary {
threadId: string;
}
export type ThreadStartOutcome =
| { readonly kind: "not-started" }
| { readonly kind: "created-activated"; readonly threadId: string }
| { readonly kind: "created-not-activated"; readonly threadId: string };
export interface ThreadStartActionsHost {
stateStore: ChatStateStore;
@ -24,7 +27,7 @@ export interface ThreadStartActions {
startThread: (
preview?: string,
options?: { syncGoal?: boolean; preservePendingSubmissionId?: string },
) => Promise<StartedThreadSummary | null>;
) => Promise<ThreadStartOutcome>;
}
export function createThreadStartActions(host: ThreadStartActionsHost): ThreadStartActions {
@ -37,15 +40,27 @@ async function startThread(
host: ThreadStartActionsHost,
preview?: string,
options: { syncGoal?: boolean; preservePendingSubmissionId?: string } = {},
): Promise<StartedThreadSummary | null> {
): Promise<ThreadStartOutcome> {
const requestState = host.stateStore.getState();
const panelTarget = capturePanelTargetLease(requestState);
const runtimeSnapshot = host.runtimeSnapshotForState(requestState);
const runtimeConfig = runtimeConfigOrDefault(requestState.connection.runtimeConfig);
const activation = await host.threadStartTransport.startThread({
const effect = await host.threadStartTransport.startThread({
serviceTier: serviceTierRequestForThreadStart(runtimeSnapshot, runtimeConfig),
permissions: permissionProfileRequestForThreadStart(runtimeSnapshot, runtimeConfig),
});
if (!activation) return null;
if (!effectCompleted(effect)) return { kind: "not-started" };
const activation = effect.value;
if (!effectCompletedInCurrentContext(effect)) {
return { kind: "created-not-activated", threadId: activation.thread.id };
}
const fallbackPreview = preview?.trim();
const thread =
activation.thread.preview.trim().length > 0 || !fallbackPreview
? activation.thread
: { ...activation.thread, preview: fallbackPreview };
const patchedActivation = thread === activation.thread ? activation : { ...activation, thread };
host.recordStartedThread(thread);
const current = host.stateStore.getState();
if (
options.preservePendingSubmissionId &&
@ -54,24 +69,24 @@ async function startThread(
options.preservePendingSubmissionId,
)
) {
return null;
return { kind: "created-not-activated", threadId: activation.thread.id };
}
if (!panelTargetLeaseIsCurrent(current, panelTarget)) {
return { kind: "created-not-activated", threadId: activation.thread.id };
}
const state = host.stateStore.getState();
const fallbackPreview = preview?.trim();
const thread =
activation.thread.preview.trim().length > 0 || !fallbackPreview
? activation.thread
: { ...activation.thread, preview: fallbackPreview };
const patchedActivation = thread === activation.thread ? activation : { ...activation, thread };
const action = resumedThreadAction({
response: patchedActivation,
listedThreads: state.threadList.listedThreads,
preserveRequestedRuntimeSettings: activeThreadId(requestState) === null,
expectedPanelTargetRevision: panelTarget.revision,
...(options.preservePendingSubmissionId ? { preservePendingSubmissionId: options.preservePendingSubmissionId } : {}),
});
host.stateStore.dispatch(action);
host.recordStartedThread(action.thread);
const applied = host.stateStore.dispatch(action);
if (activeThreadId(applied) !== action.thread.id) {
return { kind: "created-not-activated", threadId: action.thread.id };
}
if (options.syncGoal ?? true) host.syncThreadGoal(action.thread.id);
return { threadId: action.thread.id };
return { kind: "created-activated", threadId: action.thread.id };
}

3
src/features/chat/application/threads/thread-start-transport.ts
View file

@ -1,5 +1,6 @@
import type { RuntimeServiceTierRequest, RuntimeSettingsPatch } from "../../../../domain/runtime/thread-settings";
import type { ThreadActivationSnapshot } from "../../../../domain/threads/activation";
import type { EffectOutcome } from "../effect-outcome";
interface ThreadStartRequest {
serviceTier?: RuntimeServiceTierRequest;
@ -7,5 +8,5 @@ interface ThreadStartRequest {
}
export interface ThreadStartTransport {
startThread(request: ThreadStartRequest): Promise<ThreadActivationSnapshot | null>;
startThread(request: ThreadStartRequest): Promise<EffectOutcome<ThreadActivationSnapshot>>;
}

9
src/features/chat/application/turns/composition.ts
View file

@ -7,6 +7,7 @@ import type { ChatRuntimeSettingsActions } from "../runtime/settings-actions";
import type { ChatStateStore } from "../state/store";
import type { GoalActions } from "../threads/goal-actions";
import type { ThreadManagementActions } from "../threads/thread-management-actions";
import type { ThreadStartOutcome } from "../threads/thread-start-actions";
import { type ComposerSubmitActions, type ComposerSubmitActionsHost, submitComposer } from "./composer-submit-actions";
import { implementPlan, type PlanImplementationHost } from "./plan-implementation";
import type { ThreadReferenceInput, WebUrlInput } from "./slash-command-execution";
@ -66,7 +67,7 @@ interface TurnWorkflowThreadStarter {
startThread: (
preview?: string,
options?: { syncGoal?: boolean; preservePendingSubmissionId?: string },
) => Promise<{ threadId: string } | null>;
) => Promise<ThreadStartOutcome>;
}
interface PlanImplementation {
@ -97,7 +98,7 @@ export function createTurnWorkflowActions(context: TurnWorkflowContext, refs: Tu
localItemIds,
turnTransport,
ensureRestoredThreadLoaded: thread.ensureRestoredThreadLoaded,
startThread: async (preview, options) => (await refs.threadStarter.startThread(preview, options)) !== null,
startThread: async (preview, options) => (await refs.threadStarter.startThread(preview, options)).kind === "created-activated",
notifyActiveThreadIdentityChanged: thread.notifyIdentityChanged,
resetThreadTurnPresence: thread.resetTurnPresence,
applyPendingThreadSettings: () => refs.runtimeSettings.applyPendingThreadSettings(),
@ -180,6 +181,6 @@ export function createTurnWorkflowActions(context: TurnWorkflowContext, refs: Tu
}
async function startThreadForGoal(starter: TurnWorkflowThreadStarter, objective: string): Promise<string | null> {
const response = await starter.startThread(objective, { syncGoal: false });
return response?.threadId ?? null;
const outcome = await starter.startThread(objective, { syncGoal: false });
return outcome.kind === "created-activated" ? outcome.threadId : null;
}

30
src/features/chat/application/turns/turn-submission-actions.ts
View file

@ -5,6 +5,11 @@ import type { LocalIdSource } from "../local-id-source";
import { activePanelOperationDecision } from "../panel-operation-policy";
import { pendingSubmissionMatches } from "../state/pending-submission";
import type { ChatStateStore } from "../state/store";
import {
capturePanelTargetLease,
type PanelTargetLease,
panelTargetLeaseIsCurrent,
} from "../state/panel-target";
import {
acknowledgeOptimisticTurnStart,
cleanupFailedTurnStart,
@ -76,16 +81,17 @@ async function sendTurnText(
request: TurnSubmissionRequest,
): Promise<boolean> {
const { text, inputSnapshot, codexInputOverride, referencedThread } = request;
let panelTarget = capturePanelTargetLease(host.stateStore.getState());
const prepared = codexInputOverride
? { text, input: codexInputOverride }
: inputSnapshot
? host.prepareInput(text, inputSnapshot)
: { text, input: codexTextInput(text) };
if (!pendingRequestIsCurrent(host, request)) return false;
if (!submissionScopeIsCurrent(host, request, panelTarget)) return false;
if (!(await host.turnTransport.ensureConnected())) return false;
if (!pendingRequestIsCurrent(host, request)) return false;
if (!submissionScopeIsCurrent(host, request, panelTarget)) return false;
if (!(await host.ensureRestoredThreadLoaded())) return false;
if (!pendingRequestIsCurrent(host, request)) return false;
if (!submissionScopeIsCurrent(host, request, panelTarget)) return false;
const operationDecision = activePanelOperationDecision(host.stateStore.getState(), "submit");
if (operationDecision.kind === "blocked") {
@ -111,7 +117,8 @@ async function sendTurnText(
if (failPendingRequest(host, request)) restoreSubmittedDraft(host, text, request);
return false;
}
if (!pendingRequestIsCurrent(host, request)) return false;
panelTarget = capturePanelTargetLease(host.stateStore.getState());
if (!submissionScopeIsCurrent(host, request, panelTarget)) return false;
break;
case "start-turn":
break;
@ -127,7 +134,10 @@ async function sendTurnText(
if (failPendingRequest(host, request)) restoreSubmittedDraft(host, text, request);
return false;
}
if (!pendingRequestIsCurrent(host, request) || submissionStateSnapshot(host.stateStore.getState()).activeThreadId !== activeThreadId) {
if (
!submissionScopeIsCurrent(host, request, panelTarget) ||
submissionStateSnapshot(host.stateStore.getState()).activeThreadId !== activeThreadId
) {
return false;
}
@ -190,7 +200,7 @@ async function sendTurnText(
return true;
} catch (error) {
const failedState = submissionStateSnapshot(host.stateStore.getState());
const currentBeforeAdoption = !optimisticItemId && pendingRequestIsCurrent(host, request);
const currentBeforeAdoption = !optimisticItemId && submissionScopeIsCurrent(host, request, panelTarget);
const currentAfterAdoption =
optimisticItemId !== null &&
failedState.activeThreadId === expectedThreadId &&
@ -338,6 +348,14 @@ function pendingRequestIsCurrent(host: TurnSubmissionActionsHost, request: TurnS
);
}
function submissionScopeIsCurrent(
host: TurnSubmissionActionsHost,
request: TurnSubmissionRequest,
panelTarget: PanelTargetLease,
): boolean {
return pendingRequestIsCurrent(host, request) && panelTargetLeaseIsCurrent(host.stateStore.getState(), panelTarget);
}
function commitPendingRequest(host: TurnSubmissionActionsHost, request: TurnSubmissionRequest): boolean {
if (!request.pendingSubmissionId) return true;
if (!pendingRequestIsCurrent(host, request)) return false;

16
src/features/chat/host/bundles/thread-bundle.ts
View file

@ -217,10 +217,15 @@ export function createThreadLifecycleBundle(
stateStore: host.stateStore,
goalTransport: appServer.threadGoal,
localItemIds,
startThread: (preview, options) => threadStart.startThread(preview, options),
startThread: async (preview, options) => {
const outcome = await threadStart.startThread(preview, options);
return outcome.kind === "created-activated" ? { threadId: outcome.threadId } : null;
},
ensureRestoredThreadLoaded: async () => {
if (!sessionLifecycle) return false;
return sessionLifecycle.restoration.ensureLoaded((threadId) => sessionLifecycle?.resume.resumeThread(threadId) ?? Promise.resolve());
return sessionLifecycle.restoration.ensureLoaded(async (threadId) => {
await sessionLifecycle?.resume.resumeThread(threadId);
});
},
addSystemMessage: (text) => {
status.addSystemMessage(text);
@ -281,7 +286,9 @@ export function createThreadActionBundle(host: ChatPanelThreadHost, input: ChatP
composerController.setDraft(text, { focus: true });
},
openThreadInNewView: (threadId) => environment.plugin.workspace.openThreadInNewView(threadId),
openThreadInCurrentPanel: (threadId) => lifecycle.resume.resumeThread(threadId),
openThreadInCurrentPanel: async (threadId) => {
await lifecycle.resume.resumeThread(threadId);
},
notifyActiveThreadIdentityChanged,
refreshAfterThreadMutation: async () => {
await refreshActiveThreads();
@ -302,7 +309,8 @@ export function createThreadActionBundle(host: ChatPanelThreadHost, input: ChatP
toolbarPanelActions.closeForThreadSelection();
},
focusThreadInOpenView: (threadId) => environment.plugin.workspace.focusThreadInOpenView(threadId),
resumeThread: (threadId) => lifecycle.resume.resumeThread(threadId),
resumeThread: (threadId, intent) => lifecycle.resume.resumeThread(threadId, intent),
resumeWork: host.resumeWork,
addSystemMessage: status.addSystemMessage,
focusComposer: () => {
composerController.focusComposer();

4
src/features/chat/host/bundles/turn-bundle.ts
View file

@ -122,7 +122,9 @@ export function createTurnBundle(host: ChatPanelTurnHost, input: ChatPanelTurnIn
},
thread: {
ensureRestoredThreadLoaded: () =>
threadLifecycle.restoration.ensureLoaded((threadId) => threadLifecycle.resume.resumeThread(threadId)),
threadLifecycle.restoration.ensureLoaded(async (threadId) => {
await threadLifecycle.resume.resumeThread(threadId);
}),
startNewThread: () => navigation.startNewThread(),
selectThread: (threadId) => navigation.selectThread(threadId),
notifyIdentityChanged: notifyActiveThreadIdentityChanged,

9
src/features/chat/host/session.ts
View file

@ -178,9 +178,10 @@ export class ChatPanelSession implements ChatPanelHandle {
}
async openThread(threadId: string): Promise<void> {
const intent = this.resumeWork.begin(threadId);
const preparation = await this.runtime.thread.navigation.prepareForPersistentNavigation(threadId);
if (!preparation) return;
await this.runtime.thread.resume.resumeThread(threadId);
if (!preparation || !this.resumeWork.isCurrent(intent)) return;
if (!(await this.runtime.thread.resume.resumeThread(threadId, intent)) || !this.resumeWork.isCurrent(intent)) return;
await this.runtime.thread.navigation.completePersistentNavigation(preparation);
this.focusComposer();
}
@ -309,7 +310,9 @@ export class ChatPanelSession implements ChatPanelHandle {
}
private ensureRestoredThreadLoaded(): Promise<boolean> {
return this.runtime.thread.restoration.ensureLoaded((threadId) => this.runtime.thread.resume.resumeThread(threadId));
return this.runtime.thread.restoration.ensureLoaded(async (threadId) => {
await this.runtime.thread.resume.resumeThread(threadId);
});
}
private createSessionRuntime(): ChatPanelSessionRuntime {

61
tests/features/chat/app-server/transports.test.ts
View file

@ -51,8 +51,10 @@ describe("chat app-server transports", () => {
serviceTier: "priority",
permissions: ":workspace",
});
expect(snapshot?.thread.id).toBe("thread");
expect(snapshot?.cwd).toBe("/vault");
expect(snapshot).toMatchObject({
kind: "completed-current",
value: { thread: { id: "thread" }, cwd: "/vault" },
});
});
it("drops stale thread start responses after the current client changes", async () => {
@ -69,7 +71,10 @@ describe("chat app-server transports", () => {
currentClient = secondClient;
start.resolve(threadStartResponse("thread"));
await expect(starting).resolves.toBeNull();
await expect(starting).resolves.toMatchObject({
kind: "completed-stale",
value: { thread: { id: "thread" } },
});
});
it("starts turns with the session vault path and returns chat-owned turn ids", async () => {
@ -165,7 +170,7 @@ describe("chat app-server transports", () => {
connectedClient: vi.fn().mockResolvedValue(client),
}).threadMutation;
await expect(transport.compactThread("thread")).resolves.toBe(true);
await expect(transport.compactThread("thread")).resolves.toEqual({ kind: "completed-current", value: undefined });
expect(request).toHaveBeenCalledWith("thread/compact/start", { threadId: "thread" });
});
@ -178,10 +183,13 @@ describe("chat app-server transports", () => {
connectedClient: vi.fn().mockResolvedValue(client),
}).threadMutation;
const thread = await transport.forkThread("source");
const outcome = await transport.forkThread("source");
expect(request).toHaveBeenCalledWith("thread/fork", { threadId: "source", cwd: "/vault", excludeTurns: true });
expect(thread).toMatchObject({ id: "forked", preview: "Preview", archived: false });
expect(outcome).toMatchObject({
kind: "completed-current",
value: { id: "forked", preview: "Preview", archived: false },
});
});
it("drops stale fork transport responses after the current client changes", async () => {
@ -198,7 +206,10 @@ describe("chat app-server transports", () => {
currentClient = secondClient;
fork.resolve({ thread: threadRecord("forked") });
await expect(forking).resolves.toBeNull();
await expect(forking).resolves.toMatchObject({
kind: "completed-stale",
value: { id: "forked" },
});
});
it("projects rollback turns into thread stream items", async () => {
@ -212,9 +223,14 @@ describe("chat app-server transports", () => {
const snapshot = await transport.rollbackThread("thread");
expect(request).toHaveBeenCalledWith("thread/rollback", { threadId: "thread", numTurns: 1 });
expect(snapshot?.thread.id).toBe("thread");
expect(snapshot?.cwd).toBe("/vault");
expect(snapshot?.items).toEqual([expect.objectContaining({ kind: "dialogue", role: "user", text: "prompt" })]);
expect(snapshot).toMatchObject({
kind: "completed-current",
value: {
thread: { id: "thread" },
cwd: "/vault",
items: [expect.objectContaining({ kind: "dialogue", role: "user", text: "prompt" })],
},
});
});
it("reads thread history pages as thread stream items", async () => {
@ -289,13 +305,17 @@ describe("chat app-server transports", () => {
excludeTurns: true,
initialTurnsPage: { limit: 20, sortDirection: "desc", itemsView: "full" },
});
expect(snapshot?.activation.thread.id).toBe("thread");
expect(snapshot?.activation.cwd).toBe("/vault");
expect(snapshot?.rolloutPath).toBe("/tmp/rollout.jsonl");
expect(snapshot?.initialHistoryPage).toMatchObject({
nextCursor: "older",
hadTurns: true,
items: [expect.objectContaining({ kind: "dialogue", role: "user", text: "prompt" })],
expect(snapshot).toMatchObject({
kind: "completed-current",
value: {
activation: { thread: { id: "thread" }, cwd: "/vault" },
rolloutPath: "/tmp/rollout.jsonl",
initialHistoryPage: {
nextCursor: "older",
hadTurns: true,
items: [expect.objectContaining({ kind: "dialogue", role: "user", text: "prompt" })],
},
},
});
});
@ -313,7 +333,10 @@ describe("chat app-server transports", () => {
currentClient = secondClient;
resume.resolve(threadResumeResponse("thread"));
await expect(resuming).resolves.toBeNull();
await expect(resuming).resolves.toMatchObject({
kind: "completed-stale",
value: { activation: { thread: { id: "thread" } } },
});
});
it("returns no resume snapshot when no connected client is available", async () => {
@ -322,7 +345,7 @@ describe("chat app-server transports", () => {
connectedClient: vi.fn().mockResolvedValue(null),
}).threadResume;
await expect(transport.resumeThread("thread")).resolves.toBeNull();
await expect(transport.resumeThread("thread")).resolves.toEqual({ kind: "not-started" });
});
it("unsubscribes a panel from a persistent thread without interrupting its turn", async () => {

1
tests/features/chat/application/connection/reconnect-actions.test.ts
View file

@ -60,6 +60,7 @@ function createHost(overrides: Partial<ChatReconnectActionsHost> = {}) {
serviceTier: null,
approvalsReviewer: null,
});
return true;
}),
addSystemMessage: vi.fn(),
...overrides,

46
tests/features/chat/application/state/root-reducer.test.ts
View file

@ -4,6 +4,10 @@ import { createServerDiagnostics, diagnosticProbeOk, diagnosticsWithProbe } from
import type { ThreadGoal } from "../../../../../src/domain/threads/goal";
import type { Thread } from "../../../../../src/domain/threads/model";
import { activeThreadState, type ChatState, chatReducer } from "../../../../../src/features/chat/application/state/root-reducer";
import {
capturePanelTargetLease,
panelTargetLeaseIsCurrent,
} from "../../../../../src/features/chat/application/state/panel-target";
import { createChatStateStore } from "../../../../../src/features/chat/application/state/store";
import { threadStreamItems } from "../../../../../src/features/chat/application/state/thread-stream";
import { activeTurnId, chatTurnBusy, pendingTurnStart } from "../../../../../src/features/chat/application/turns/turn-state";
@ -244,6 +248,30 @@ describe("chatReducer", () => {
expect(disconnected.panelThread).toEqual(restored.panelThread);
});
it("invalidates an old panel lease even when navigation returns to the same thread", () => {
let state = chatReducer(chatStateFixture(), resumedThreadAction("first"));
const oldLease = capturePanelTargetLease(state);
state = chatReducer(state, resumedThreadAction("second"));
state = chatReducer(state, resumedThreadAction("first"));
expect(panelTargetLeaseIsCurrent(state, oldLease)).toBe(false);
expect(state.panelTargetRevision).toBeGreaterThan(oldLease.revision);
});
it("preserves a panel lease when an awaited thread becomes active", () => {
let state = chatReducer(chatStateFixture(), {
type: "panel/restored-thread-applied",
threadId: "restored",
fallbackTitle: "Restored",
});
const lease = capturePanelTargetLease(state);
state = chatReducer(state, resumedThreadAction("restored"));
expect(panelTargetLeaseIsCurrent(state, lease)).toBe(true);
});
it("keeps active-only metadata out of the awaiting-resume phase", () => {
let state = chatReducer(chatStateFixture(), {
type: "panel/restored-thread-applied",
@ -814,3 +842,21 @@ function thread(id: string): Thread {
provenance: { kind: "interactive" },
};
}
function resumedThreadAction(threadId: string) {
return {
type: "active-thread/resumed" as const,
approvalPolicyKnown: true,
sandboxPolicyKnown: true,
permissionProfileKnown: true,
approvalPolicy: null,
sandboxPolicy: null,
activePermissionProfile: null,
thread: thread(threadId),
cwd: "/vault",
model: null,
reasoningEffort: null,
serviceTier: null,
approvalsReviewer: null,
};
}

56
tests/features/chat/application/threads/ephemeral-thread-lifecycle.test.ts
View file

@ -4,6 +4,7 @@ import { activeThreadId, activeThreadState } from "../../../../../src/features/c
import { createChatStateStore } from "../../../../../src/features/chat/application/state/store";
import { createEphemeralThreadLifecycle } from "../../../../../src/features/chat/application/threads/ephemeral-thread-lifecycle";
import type { EphemeralThreadTransport } from "../../../../../src/features/chat/application/threads/ephemeral-thread-transport";
import { deferred } from "../../../../support/async";
describe("ephemeral thread lifecycle", () => {
it("activates an ephemeral fork without adding it to the thread list", async () => {
@ -61,6 +62,47 @@ describe("ephemeral thread lifecycle", () => {
expect(activeThreadId(store.getState())).toBeNull();
});
it("does not clear a newer panel target after side-chat unsubscribe completes", async () => {
const store = createChatStateStore();
const unsubscribe = deferred<boolean>();
const transport = transportMock();
transport.unsubscribeEphemeralThread = vi.fn(() => unsubscribe.promise);
const notifyActiveThreadIdentityChanged = vi.fn();
const lifecycle = createEphemeralThreadLifecycle({
stateStore: store,
transport,
ensureConnected: vi.fn().mockResolvedValue(true),
addSystemMessage: vi.fn(),
notifyActiveThreadIdentityChanged,
interruptTurn: vi.fn().mockResolvedValue(true),
});
await lifecycle.open({ sourceThreadId: "source", sourceThreadTitle: null });
notifyActiveThreadIdentityChanged.mockClear();
const preparing = lifecycle.prepareForPersistentNavigation();
await vi.waitFor(() => expect(transport.unsubscribeEphemeralThread).toHaveBeenCalledWith("side"));
store.dispatch({
type: "active-thread/resumed",
approvalPolicyKnown: true,
sandboxPolicyKnown: true,
permissionProfileKnown: true,
approvalPolicy: null,
sandboxPolicy: null,
activePermissionProfile: null,
thread: { ...activationFixture().thread, id: "other" },
cwd: "/vault",
model: null,
reasoningEffort: null,
serviceTier: null,
approvalsReviewer: null,
});
unsubscribe.resolve(true);
await expect(preparing).resolves.toBe(false);
expect(activeThreadId(store.getState())).toBe("other");
expect(notifyActiveThreadIdentityChanged).not.toHaveBeenCalled();
});
it("interrupts a running side turn before close cleanup", async () => {
const store = createChatStateStore();
const transport = transportMock();
@ -104,7 +146,10 @@ describe("ephemeral thread lifecycle", () => {
const opening = lifecycle.open({ sourceThreadId: "source", sourceThreadTitle: "Source" });
await Promise.resolve();
await lifecycle.dispose();
resolveFork({ kind: "ready", sourceThreadId: "source", activation: activationFixture() });
resolveFork({
kind: "completed-current",
value: { kind: "ready", sourceThreadId: "source", activation: activationFixture() },
});
await expect(opening).resolves.toBe(false);
expect(transport.unsubscribeEphemeralThread).toHaveBeenCalledWith("side");
@ -180,7 +225,9 @@ describe("ephemeral thread lifecycle", () => {
it("retries cleanup-required forks when the side view is disposed", async () => {
const store = createChatStateStore();
const transport = transportMock();
transport.forkEphemeralThread = vi.fn().mockResolvedValue({ kind: "cleanup-required", threadId: "side" });
transport.forkEphemeralThread = vi
.fn()
.mockResolvedValue({ kind: "completed-current", value: { kind: "cleanup-required", threadId: "side" } });
const addSystemMessage = vi.fn();
const lifecycle = createEphemeralThreadLifecycle({
stateStore: store,
@ -201,7 +248,10 @@ describe("ephemeral thread lifecycle", () => {
function transportMock(): EphemeralThreadTransport {
return {
forkEphemeralThread: vi.fn().mockResolvedValue({ kind: "ready", sourceThreadId: "source", activation: activationFixture() }),
forkEphemeralThread: vi.fn().mockResolvedValue({
kind: "completed-current",
value: { kind: "ready", sourceThreadId: "source", activation: activationFixture() },
}),
unsubscribeEphemeralThread: vi.fn().mockResolvedValue(true),
};
}

60
tests/features/chat/application/threads/goal-actions.test.ts
View file

@ -1,5 +1,6 @@
import { describe, expect, it, vi } from "vitest";
import type { ThreadGoal } from "../../../../../src/domain/threads/goal";
import type { EffectOutcome } from "../../../../../src/features/chat/application/effect-outcome";
import { createLocalIdSource } from "../../../../../src/features/chat/application/local-id-source";
import { activeThreadId, activeThreadState } from "../../../../../src/features/chat/application/state/root-reducer";
import { createChatStateStore } from "../../../../../src/features/chat/application/state/store";
@ -63,8 +64,8 @@ describe("createGoalActions", () => {
const updated = goal({ objective: "Updated", tokenBudget: 250 });
const paused = goal({ objective: "Updated", status: "paused", tokenBudget: 250 });
const goalTransport = goalTransportFixture({
setThreadGoal: vi.fn().mockResolvedValueOnce(updated).mockResolvedValueOnce(paused),
clearThreadGoal: vi.fn().mockResolvedValue(true),
setThreadGoal: vi.fn().mockResolvedValueOnce(completedCurrent(updated)).mockResolvedValueOnce(completedCurrent(paused)),
clearThreadGoal: vi.fn().mockResolvedValue(completedCurrent(undefined)),
});
const { setThreadGoal, clearThreadGoal } = goalTransport;
const addSystemMessage = vi.fn();
@ -141,7 +142,7 @@ describe("createGoalActions", () => {
});
const pending = actions.setStatus("thread", "paused");
await Promise.resolve();
await vi.waitFor(() => expect(goalTransport.setThreadGoal).toHaveBeenCalledOnce());
stateStore.dispatch({ type: "active-thread/cleared" });
update.reject(new Error("offline"));
await pending;
@ -168,7 +169,7 @@ describe("createGoalActions", () => {
});
const pending = actions.clear("thread");
await Promise.resolve();
await vi.waitFor(() => expect(goalTransport.clearThreadGoal).toHaveBeenCalledOnce());
stateStore.dispatch({ type: "active-thread/cleared" });
clear.reject(new Error("offline"));
await pending;
@ -199,11 +200,38 @@ describe("createGoalActions", () => {
expect(goalTransport.clearThreadGoal).not.toHaveBeenCalled();
});
it("does not set a goal on an old panel target after connection completes", async () => {
let state = chatStateFixture();
state = chatStateWith(state, { activeThread: { id: "thread" } });
const stateStore = createChatStateStore(state);
const connection = deferred<boolean>();
const goalTransport = goalTransportFixture({ ensureConnected: vi.fn(() => connection.promise) });
const addSystemMessage = vi.fn();
const actions = createGoalActions({
stateStore,
goalTransport,
localItemIds: createLocalIdSource({ nowMs: () => 1, seed: "goal" }),
startThread: vi.fn().mockResolvedValue({ threadId: "thread" }),
addSystemMessage,
addGoalEvent: vi.fn(),
refreshLiveState: vi.fn(),
});
const pending = actions.setObjective("thread", "Finish", null);
await vi.waitFor(() => expect(goalTransport.ensureConnected).toHaveBeenCalledOnce());
stateStore.dispatch({ type: "active-thread/cleared" });
connection.resolve(true);
await expect(pending).resolves.toBe(false);
expect(goalTransport.setThreadGoal).not.toHaveBeenCalled();
expect(addSystemMessage).not.toHaveBeenCalled();
});
it("reports goal creation as a structured goal event", async () => {
let state = chatStateFixture();
state = chatStateWith(state, { activeThread: { id: "thread" } });
const stateStore = createChatStateStore(state);
const goalTransport = goalTransportFixture({ setThreadGoal: vi.fn().mockResolvedValueOnce(goal()) });
const goalTransport = goalTransportFixture({ setThreadGoal: vi.fn().mockResolvedValueOnce(completedCurrent(goal())) });
const addSystemMessage = vi.fn();
const addGoalEvent = vi.fn();
const actions = createGoalActions({
@ -234,7 +262,7 @@ describe("createGoalActions", () => {
it("starts a thread before saving a new goal objective when no thread is active", async () => {
const stateStore = createChatStateStore(chatStateFixture());
const savedGoal = goal({ threadId: "thread-new", objective: "Plan release" });
const goalTransport = goalTransportFixture({ setThreadGoal: vi.fn().mockResolvedValueOnce(savedGoal) });
const goalTransport = goalTransportFixture({ setThreadGoal: vi.fn().mockResolvedValueOnce(completedCurrent(savedGoal)) });
const { setThreadGoal } = goalTransport;
const startThread = vi.fn().mockImplementation(async () => {
stateStore.dispatch({
@ -308,7 +336,7 @@ describe("createGoalActions", () => {
const stateStore = createChatStateStore(chatStateFixture());
stateStore.dispatch({ type: "panel/restored-thread-applied", threadId: "restored", fallbackTitle: "Restored" });
const goalTransport = goalTransportFixture({
setThreadGoal: vi.fn().mockResolvedValue(goal({ threadId: "restored", objective: "Resume work" })),
setThreadGoal: vi.fn().mockResolvedValue(completedCurrent(goal({ threadId: "restored", objective: "Resume work" }))),
});
const startThread = vi.fn().mockResolvedValue({ threadId: "new-thread" });
const ensureRestoredThreadLoaded = vi.fn(async () => {
@ -386,7 +414,7 @@ describe("createGoalActions", () => {
state = chatStateWith(state, { activeThread: { id: "thread" } });
const stateStore = createChatStateStore(state);
const goalTransport = goalTransportFixture({
setThreadGoal: vi.fn().mockResolvedValueOnce(goal()),
setThreadGoal: vi.fn().mockResolvedValueOnce(completedCurrent(goal())),
recordThreadGoalUserMessage: vi.fn().mockRejectedValue(new Error("offline")),
});
const addSystemMessage = vi.fn();
@ -410,7 +438,7 @@ describe("createGoalActions", () => {
state = chatStateWith(state, { activeThread: { id: "thread" } });
const stateStore = createChatStateStore(state);
const goalTransport = goalTransportFixture({
setThreadGoal: vi.fn().mockResolvedValueOnce(goal()),
setThreadGoal: vi.fn().mockResolvedValueOnce(completedCurrent(goal())),
recordThreadGoalUserMessage: vi.fn().mockImplementation(async () => {
stateStore.dispatch({ type: "active-thread/cleared" });
throw new Error("offline");
@ -437,7 +465,9 @@ describe("createGoalActions", () => {
state = chatStateWith(state, { activeThread: { id: "thread" } });
state = chatStateWith(state, { activeThread: { goal: goal() } });
const stateStore = createChatStateStore(state);
const goalTransport = goalTransportFixture({ setThreadGoal: vi.fn().mockResolvedValueOnce(goal({ objective: "Updated" })) });
const goalTransport = goalTransportFixture({
setThreadGoal: vi.fn().mockResolvedValueOnce(completedCurrent(goal({ objective: "Updated" }))),
});
const addGoalEvent = vi.fn();
const actions = createGoalActions({
stateStore,
@ -460,7 +490,7 @@ describe("createGoalActions", () => {
state = chatStateWith(state, { activeThread: { id: "thread" } });
state = chatStateWith(state, { activeThread: { goal: goal({ status: "paused" }) } });
const stateStore = createChatStateStore(state);
const goalTransport = goalTransportFixture({ setThreadGoal: vi.fn().mockResolvedValueOnce(goal()) });
const goalTransport = goalTransportFixture({ setThreadGoal: vi.fn().mockResolvedValueOnce(completedCurrent(goal())) });
const addSystemMessage = vi.fn();
const addGoalEvent = vi.fn();
const actions = createGoalActions({
@ -520,10 +550,14 @@ function goal(overrides: Partial<ThreadGoal> = {}): ThreadGoal {
function goalTransportFixture(overrides: Partial<ThreadGoalTransport> = {}): ThreadGoalTransport {
return {
readThreadGoal: vi.fn().mockResolvedValue(null),
setThreadGoal: vi.fn().mockResolvedValue(goal()),
clearThreadGoal: vi.fn().mockResolvedValue(true),
setThreadGoal: vi.fn().mockResolvedValue(completedCurrent(goal())),
clearThreadGoal: vi.fn().mockResolvedValue(completedCurrent(undefined)),
recordThreadGoalUserMessage: vi.fn().mockResolvedValue(true),
ensureConnected: vi.fn().mockResolvedValue(true),
...overrides,
};
}
function completedCurrent<T>(value: T): EffectOutcome<T> {
return { kind: "completed-current", value };
}

32
tests/features/chat/application/threads/resume-actions.test.ts
View file

@ -38,15 +38,16 @@ function activation(threadId: string, overrides: Partial<ThreadResumeSnapshot> =
function createActions(response: ThreadResumeSnapshot | null = activation("thread"), overrides: Partial<ResumeActionsHost> = {}) {
const stateStore = createChatStateStore(createChatState());
const resumeThread = vi.fn<ThreadResumeTransport["resumeThread"]>().mockResolvedValue(response);
const resumeThread = vi
.fn<ThreadResumeTransport["resumeThread"]>()
.mockResolvedValue(response ? { kind: "completed-current", value: response } : { kind: "not-started" });
const loadLatest = vi.fn().mockResolvedValue(undefined);
const applyLatestPage = vi.fn();
const invalidateHistory = vi.fn();
const host = {
const host: ResumeActionsHost & { systemItem: (text: string) => ThreadStreamItem } = {
stateStore,
resumeWork: new ChatResumeWorkTracker(),
history: { loadLatest, applyLatestPage, invalidate: invalidateHistory } as unknown as HistoryController,
resumeTransport: { resumeThread },
closing: () => false,
systemItem: (text: string) => ({ id: "system", kind: "system" as const, role: "system" as const, text }),
resetThreadTurnPresence: vi.fn(),
@ -55,6 +56,7 @@ function createActions(response: ThreadResumeSnapshot | null = activation("threa
refreshLiveState: vi.fn(),
syncThreadGoal: vi.fn().mockResolvedValue(undefined),
...overrides,
resumeTransport: overrides.resumeTransport ?? { ensureConnected: vi.fn().mockResolvedValue(true), resumeThread },
};
return {
actions: createResumeActions(host),
@ -103,6 +105,30 @@ describe("ResumeActions", () => {
expect(host.syncThreadGoal).not.toHaveBeenCalled();
});
it("does not invoke an older resume after a newer intent wins during connection", async () => {
const firstConnection = deferred<boolean>();
const resumeThread = vi
.fn<ThreadResumeTransport["resumeThread"]>()
.mockImplementation(async (threadId) => ({
kind: "completed-current",
value: activation(threadId),
}));
const ensureConnected = vi.fn().mockReturnValueOnce(firstConnection.promise).mockResolvedValue(true);
const { actions, stateStore } = createActions(undefined, {
resumeTransport: { ensureConnected, resumeThread },
});
const firstResume = actions.resumeThread("first");
await vi.waitFor(() => expect(ensureConnected).toHaveBeenCalledOnce());
await expect(actions.resumeThread("second")).resolves.toBe(true);
await firstConnection.resolveAndFlush(true);
await expect(firstResume).resolves.toBe(false);
expect(resumeThread).toHaveBeenCalledOnce();
expect(resumeThread).toHaveBeenCalledWith("second");
expect(activeThreadId(stateStore.getState())).toBe("second");
});
it("refreshes live state after resumed history and goal sync finish", async () => {
const { actions, host } = createActions();

67
tests/features/chat/application/threads/thread-management-actions.test.ts
View file

@ -1,6 +1,7 @@
import type { Mock } from "vitest";
import { describe, expect, it, vi } from "vitest";
import type { Thread } from "../../../../../src/domain/threads/model";
import type { EffectOutcome } from "../../../../../src/features/chat/application/effect-outcome";
import { activeThreadId } from "../../../../../src/features/chat/application/state/root-reducer";
import { createChatStateStore } from "../../../../../src/features/chat/application/state/store";
import {
@ -18,6 +19,7 @@ import { chatStateFixture, chatStateWith } from "../../support/state";
import { chatStateThreadStreamItems, withChatStateThreadStreamItems } from "../../support/thread-stream";
interface ThreadMutationTransportMock {
ensureConnected: Mock<ThreadMutationTransport["ensureConnected"]>;
compactThread: Mock<ThreadMutationTransport["compactThread"]>;
forkThread: Mock<ThreadMutationTransport["forkThread"]>;
rollbackThread: Mock<ThreadMutationTransport["rollbackThread"]>;
@ -68,6 +70,38 @@ describe("thread management actions", () => {
expect(host.threadTransport.compactThread).toHaveBeenCalledWith("side-thread");
});
it("does not compact an old panel target after connection completes", async () => {
const connection = deferred<boolean>();
const host = hostMock({
items: [],
activeThread: { id: "source" },
threadTransport: { ensureConnected: vi.fn(() => connection.promise) },
});
const compacting = threadManagementActions(host).compactThread("source");
await waitForAsyncWork(() => expect(host.threadTransport.ensureConnected).toHaveBeenCalledOnce());
host.stateStore.dispatch({
type: "active-thread/resumed",
approvalPolicyKnown: true,
sandboxPolicyKnown: true,
permissionProfileKnown: true,
approvalPolicy: null,
sandboxPolicy: null,
activePermissionProfile: null,
thread: panelThread("other"),
cwd: "/vault",
model: null,
reasoningEffort: null,
serviceTier: null,
approvalsReviewer: null,
});
connection.resolve(true);
await compacting;
expect(host.threadTransport.compactThread).not.toHaveBeenCalled();
expect(activeThreadId(host.stateStore.getState())).toBe("other");
});
it("does not fork an ephemeral side chat", async () => {
const host = hostMock({
items: [],
@ -172,7 +206,7 @@ describe("thread management actions", () => {
});
it("does not report compaction completion after the panel switches threads", async () => {
const compact = deferred<boolean>();
const compact = deferred<EffectOutcome<void>>();
const host = hostMock({ items: [] });
host.threadTransport.compactThread.mockReturnValue(compact.promise);
host.stateStore.dispatch({
@ -211,7 +245,7 @@ describe("thread management actions", () => {
serviceTier: null,
approvalsReviewer: null,
});
compact.resolve(true);
compact.resolve(completedCurrent(undefined));
await pendingCompact;
expect(host.addSystemMessage).not.toHaveBeenCalledWith("Compaction requested.");
@ -222,7 +256,7 @@ describe("thread management actions", () => {
const host = hostMock({
items: [],
threadTransport: {
compactThread: vi.fn<ThreadMutationTransport["compactThread"]>().mockResolvedValue(false),
compactThread: vi.fn<ThreadMutationTransport["compactThread"]>().mockResolvedValue({ kind: "not-started" }),
},
});
const controller = threadManagementActions(host);
@ -332,7 +366,7 @@ describe("thread management actions", () => {
});
it("does not archive or replace the panel from stale fork responses", async () => {
const fork = deferred<Thread | null>();
const fork = deferred<EffectOutcome<Thread>>();
const host = hostMock({ items: turnItems() });
host.threadTransport.forkThread.mockReturnValue(fork.promise);
host.stateStore.dispatch({
@ -375,7 +409,7 @@ describe("thread management actions", () => {
serviceTier: null,
approvalsReviewer: null,
});
fork.resolve(panelThread("forked"));
fork.resolve(completedCurrent(panelThread("forked")));
await pendingFork;
expect(host.operations.archiveThread).not.toHaveBeenCalled();
@ -387,7 +421,7 @@ describe("thread management actions", () => {
const host = hostMock({
items: turnItems(),
threadTransport: {
forkThread: vi.fn<ThreadMutationTransport["forkThread"]>().mockResolvedValue(null),
forkThread: vi.fn<ThreadMutationTransport["forkThread"]>().mockResolvedValue({ kind: "not-started" }),
},
});
const controller = threadManagementActions(host);
@ -475,7 +509,7 @@ describe("thread management actions", () => {
});
it("ignores stale rollback responses after the panel switches threads", async () => {
const rollback = deferred<ThreadRollbackSnapshot | null>();
const rollback = deferred<EffectOutcome<ThreadRollbackSnapshot>>();
const host = hostMock({ items: turnItems() });
host.threadTransport.rollbackThread.mockReturnValue(rollback.promise);
host.stateStore.dispatch({
@ -520,7 +554,7 @@ describe("thread management actions", () => {
serviceTier: null,
approvalsReviewer: null,
});
rollback.resolve(rollbackSnapshot());
rollback.resolve(completedCurrent(rollbackSnapshot()));
await pendingRollback;
expect(activeThreadId(host.stateStore.getState())).toBe("other");
@ -530,7 +564,7 @@ describe("thread management actions", () => {
});
it("ignores rollback responses after a new turn starts in the same thread", async () => {
const rollback = deferred<ThreadRollbackSnapshot | null>();
const rollback = deferred<EffectOutcome<ThreadRollbackSnapshot>>();
const host = hostMock({ items: turnItems() });
host.threadTransport.rollbackThread.mockReturnValue(rollback.promise);
host.stateStore.dispatch({
@ -554,7 +588,7 @@ describe("thread management actions", () => {
const pendingRollback = controller.rollbackThread("source");
await waitForAsyncWork(() => expect(host.threadTransport.rollbackThread).toHaveBeenCalledOnce());
host.stateStore.dispatch({ type: "turn/started", threadId: "source", turnId: "new-turn" });
rollback.resolve(rollbackSnapshot());
rollback.resolve(completedCurrent(rollbackSnapshot()));
await pendingRollback;
expect(host.stateStore.getState().turn.lifecycle).toEqual({ kind: "running", turnId: "new-turn" });
@ -565,7 +599,7 @@ describe("thread management actions", () => {
const host = hostMock({
items: turnItems(),
threadTransport: {
rollbackThread: vi.fn<ThreadMutationTransport["rollbackThread"]>().mockResolvedValue(null),
rollbackThread: vi.fn<ThreadMutationTransport["rollbackThread"]>().mockResolvedValue({ kind: "not-started" }),
},
});
host.stateStore.dispatch({
@ -668,9 +702,10 @@ function hostMock({
if (activeThread) state = chatStateWith(state, { activeThread });
const stateStore = createChatStateStore(state);
const threadTransport: ThreadMutationTransportMock = {
compactThread: vi.fn<ThreadMutationTransport["compactThread"]>().mockResolvedValue(true),
forkThread: vi.fn<ThreadMutationTransport["forkThread"]>().mockResolvedValue(panelThread("forked")),
rollbackThread: vi.fn<ThreadMutationTransport["rollbackThread"]>().mockResolvedValue(rollbackSnapshot()),
ensureConnected: vi.fn<ThreadMutationTransport["ensureConnected"]>().mockResolvedValue(true),
compactThread: vi.fn<ThreadMutationTransport["compactThread"]>().mockResolvedValue(completedCurrent(undefined)),
forkThread: vi.fn<ThreadMutationTransport["forkThread"]>().mockResolvedValue(completedCurrent(panelThread("forked"))),
rollbackThread: vi.fn<ThreadMutationTransport["rollbackThread"]>().mockResolvedValue(completedCurrent(rollbackSnapshot())),
...transportOverrides,
};
const operations: ThreadOperationsMock = {
@ -720,3 +755,7 @@ function callOrder(fn: Mock): number {
if (order === undefined) throw new Error("Expected function to be called.");
return order;
}
function completedCurrent<T>(value: T): EffectOutcome<T> {
return { kind: "completed-current", value };
}

28
tests/features/chat/application/threads/thread-navigation-actions.test.ts
View file

@ -11,6 +11,8 @@ import {
createThreadNavigationActions,
type ThreadNavigationActionsHost,
} from "../../../../../src/features/chat/application/threads/thread-navigation-actions";
import { ChatResumeWorkTracker } from "../../../../../src/features/chat/application/threads/resume-work";
import { deferred } from "../../../../support/async";
function resumeThreadState(stateStore: ChatStateStore, threadId: string, subagent = false): void {
stateStore.dispatch({
@ -53,7 +55,8 @@ function createActionsHarness(overrides: Partial<ThreadNavigationActionsHost> =
} as unknown as ActiveThreadIdentitySync,
closeForThreadSelection: vi.fn(),
focusThreadInOpenView: vi.fn().mockResolvedValue(false),
resumeThread: vi.fn().mockResolvedValue(undefined),
resumeThread: vi.fn().mockResolvedValue(true),
resumeWork: new ChatResumeWorkTracker(),
addSystemMessage: vi.fn(),
focusComposer: vi.fn(),
navigation: navigationMock(),
@ -149,7 +152,24 @@ describe("ThreadNavigationActions", () => {
await actions.selectThread("thread");
expect(host.closeForThreadSelection).toHaveBeenCalledOnce();
expect(host.resumeThread).toHaveBeenCalledWith("thread");
expect(host.resumeThread).toHaveBeenCalledWith("thread", expect.objectContaining({ threadId: "thread" }));
});
it("does not resume an older selection after a newer selection wins during view lookup", async () => {
const firstLookup = deferred<boolean>();
const focusThreadInOpenView = vi.fn((threadId: string) =>
threadId === "first" ? firstLookup.promise : Promise.resolve(false),
);
const { actions, host } = createActionsHarness({ focusThreadInOpenView });
const firstSelection = actions.selectThread("first");
await vi.waitFor(() => expect(focusThreadInOpenView).toHaveBeenCalledWith("first"));
await actions.selectThread("second");
firstLookup.resolve(false);
await firstSelection;
expect(host.resumeThread).toHaveBeenCalledOnce();
expect(host.resumeThread).toHaveBeenCalledWith("second", expect.objectContaining({ threadId: "second" }));
});
it("allows switching away from a running subagent after unsubscribe preparation", async () => {
@ -163,7 +183,7 @@ describe("ThreadNavigationActions", () => {
expect(navigation.prepareForPersistentNavigation).toHaveBeenCalledWith("other");
expect(host.closeForThreadSelection).toHaveBeenCalledOnce();
expect(host.resumeThread).toHaveBeenCalledWith("other");
expect(host.resumeThread).toHaveBeenCalledWith("other", expect.objectContaining({ threadId: "other" }));
expectCallBefore(host.resumeThread as ReturnType<typeof vi.fn>, navigation.completePersistentNavigation);
});
@ -188,7 +208,7 @@ describe("ThreadNavigationActions", () => {
expect(stateStore.getState().ui.toolbarPanel).toBeNull();
expect(host.closeForThreadSelection).toHaveBeenCalledOnce();
expect(host.resumeThread).toHaveBeenCalledWith("thread");
expect(host.resumeThread).toHaveBeenCalledWith("thread", expect.objectContaining({ threadId: "thread" }));
});
it("ignores toolbar selection while another thread is busy", async () => {

45
tests/features/chat/application/threads/thread-start-actions.test.ts
View file

@ -2,6 +2,7 @@ import { describe, expect, it, vi } from "vitest";
import { emptyRuntimeConfigSnapshot } from "../../../../../src/domain/runtime/config";
import type { ThreadActivationSnapshot } from "../../../../../src/domain/threads/activation";
import type { Thread } from "../../../../../src/domain/threads/model";
import type { EffectOutcome } from "../../../../../src/features/chat/application/effect-outcome";
import { runtimeSnapshotForChatState } from "../../../../../src/features/chat/application/runtime/snapshot";
import { resumedThreadAction } from "../../../../../src/features/chat/application/state/actions";
import { activeThreadId } from "../../../../../src/features/chat/application/state/root-reducer";
@ -25,7 +26,7 @@ describe("thread start actions", () => {
const actions = createThreadStartActions({
stateStore,
threadStartTransport: { startThread: vi.fn().mockResolvedValue(activationFixture(started)) },
threadStartTransport: { startThread: vi.fn().mockResolvedValue(completedActivation(activationFixture(started))) },
runtimeSnapshotForState: runtimeSnapshotForChatState,
recordStartedThread,
syncThreadGoal,
@ -47,10 +48,12 @@ describe("thread start actions", () => {
stateStore.dispatch({ type: "runtime/approvals-reviewer-requested", approvalsReviewer: "auto_review" });
stateStore.dispatch({ type: "runtime/requested-collaboration-mode-set", collaborationMode: "plan" });
const startThread = vi.fn().mockResolvedValue(
activationFixture(threadFixture("started"), {
model: "gpt-5",
serviceTier: "fast",
}),
completedActivation(
activationFixture(threadFixture("started"), {
model: "gpt-5",
serviceTier: "fast",
}),
),
);
const actions = createThreadStartActions({
@ -81,7 +84,9 @@ describe("thread start actions", () => {
const syncThreadGoal = vi.fn();
const actions = createThreadStartActions({
stateStore,
threadStartTransport: { startThread: vi.fn().mockResolvedValue(activationFixture(threadFixture("started"))) },
threadStartTransport: {
startThread: vi.fn().mockResolvedValue(completedActivation(activationFixture(threadFixture("started")))),
},
runtimeSnapshotForState: runtimeSnapshotForChatState,
recordStartedThread: vi.fn(),
syncThreadGoal,
@ -108,7 +113,9 @@ describe("thread start actions", () => {
});
const actions = createThreadStartActions({
stateStore,
threadStartTransport: { startThread: vi.fn().mockResolvedValue(activationFixture(threadFixture("started"))) },
threadStartTransport: {
startThread: vi.fn().mockResolvedValue(completedActivation(activationFixture(threadFixture("started")))),
},
runtimeSnapshotForState: runtimeSnapshotForChatState,
recordStartedThread: vi.fn(),
syncThreadGoal: vi.fn(),
@ -133,7 +140,7 @@ describe("thread start actions", () => {
phase: "cancellable",
},
});
const started = deferred<ThreadActivationSnapshot | null>();
const started = deferred<EffectOutcome<ThreadActivationSnapshot>>();
const recordStartedThread = vi.fn();
const actions = createThreadStartActions({
stateStore,
@ -145,18 +152,20 @@ describe("thread start actions", () => {
const starting = actions.startThread(pending.text, { preservePendingSubmissionId: pending.id });
stateStore.dispatch(resumedThreadAction({ response: activationFixture(threadFixture("selected")) }));
started.resolve(activationFixture(threadFixture("delayed")));
started.resolve(completedActivation(activationFixture(threadFixture("delayed"))));
await expect(starting).resolves.toBeNull();
await expect(starting).resolves.toEqual({ kind: "created-not-activated", threadId: "delayed" });
expect(activeThreadId(stateStore.getState())).toBe("selected");
expect(recordStartedThread).not.toHaveBeenCalled();
expect(recordStartedThread).toHaveBeenCalledWith(threadFixture("delayed", { preview: pending.text }));
});
it("starts threads with service tier from explicit effective config", async () => {
let state = chatStateFixture();
state = chatStateWith(state, { connection: { runtimeConfig: { ...emptyRuntimeConfigSnapshot(), serviceTier: "flex" } } });
const stateStore = createChatStateStore(state);
const startThread = vi.fn().mockResolvedValue(activationFixture(threadFixture("started"), { serviceTier: "flex" }));
const startThread = vi
.fn()
.mockResolvedValue(completedActivation(activationFixture(threadFixture("started"), { serviceTier: "flex" })));
const actions = createThreadStartActions({
stateStore,
threadStartTransport: { startThread },
@ -184,7 +193,7 @@ describe("thread start actions", () => {
},
});
const stateStore = createChatStateStore(state);
const startThread = vi.fn().mockResolvedValue(activationFixture(threadFixture("started")));
const startThread = vi.fn().mockResolvedValue(completedActivation(activationFixture(threadFixture("started"))));
const actions = createThreadStartActions({
stateStore,
threadStartTransport: { startThread },
@ -204,7 +213,7 @@ describe("thread start actions", () => {
const recordStartedThread = vi.fn();
const actions = createThreadStartActions({
stateStore,
threadStartTransport: { startThread: vi.fn().mockResolvedValue(activationFixture(started)) },
threadStartTransport: { startThread: vi.fn().mockResolvedValue(completedActivation(activationFixture(started))) },
runtimeSnapshotForState: runtimeSnapshotForChatState,
recordStartedThread,
syncThreadGoal: vi.fn(),
@ -221,13 +230,13 @@ describe("thread start actions", () => {
const syncThreadGoal = vi.fn();
const actions = createThreadStartActions({
stateStore,
threadStartTransport: { startThread: vi.fn().mockResolvedValue(null) },
threadStartTransport: { startThread: vi.fn().mockResolvedValue({ kind: "not-started" }) },
runtimeSnapshotForState: runtimeSnapshotForChatState,
recordStartedThread,
syncThreadGoal,
});
await expect(actions.startThread("local preview")).resolves.toBeNull();
await expect(actions.startThread("local preview")).resolves.toEqual({ kind: "not-started" });
expect(activeThreadId(stateStore.getState())).toBeNull();
expect(stateStore.getState().threadList.listedThreads).toEqual([]);
expect(recordStartedThread).not.toHaveBeenCalled();
@ -266,3 +275,7 @@ function activationFixture(thread: Thread, overrides: Partial<ThreadActivationSn
...overrides,
};
}
function completedActivation(value: ThreadActivationSnapshot): EffectOutcome<ThreadActivationSnapshot> {
return { kind: "completed-current", value };
}

27
tests/features/chat/application/turns/turn-submission-actions.test.ts
View file

@ -151,6 +151,32 @@ describe("TurnSubmissionActions", () => {
expect(startTurn).not.toHaveBeenCalled();
});
it("does not submit to an old thread after the panel changes during connection", async () => {
const connection = deferred<boolean>();
const ensureConnected = vi.fn(() => connection.promise);
const { host, startTurn, stateStore } = createHost({
turnTransport: {
ensureConnected,
startTurn: vi.fn().mockResolvedValue({ turnId: "turn" }),
steerTurn: vi.fn().mockResolvedValue(true),
interruptTurn: vi.fn().mockResolvedValue(true),
},
});
resumeThread(stateStore, undefined, "first");
const actions = createTurnSubmissionActions(host);
const submitting = actions.sendTurnText({ text: "hello" });
await vi.waitFor(() => expect(ensureConnected).toHaveBeenCalledOnce());
resumeThread(stateStore, undefined, "second");
connection.resolve(true);
await expect(submitting).resolves.toBe(false);
expect(startTurn).not.toHaveBeenCalled();
expect(host.applyPendingThreadSettings).not.toHaveBeenCalled();
expect(host.setDraft).not.toHaveBeenCalled();
expect(activeThreadId(stateStore.getState())).toBe("second");
});
it("blocks direct turn submission after a restored thread resolves to a subagent", async () => {
const { host, startTurn, stateStore } = createHost({
ensureRestoredThreadLoaded: vi.fn().mockImplementation(async () => {
@ -158,6 +184,7 @@ describe("TurnSubmissionActions", () => {
return true;
}),
});
stateStore.dispatch({ type: "panel/restored-thread-applied", threadId: "child", fallbackTitle: "Agent" });
const actions = createTurnSubmissionActions(host);
await expect(actions.sendTurnText({ text: "hello" })).resolves.toBe(false);

2
tests/features/chat/host/session-runtime.test.ts
View file

@ -153,7 +153,7 @@ describe("ChatPanelSessionRuntime actions", () => {
});
vi.spyOn(runtime.connection.actions, "ensureConnected").mockResolvedValue(undefined);
vi.spyOn(runtime.connection.manager, "isConnected").mockReturnValue(true);
const resumeThread = vi.spyOn(runtime.thread.resume, "resumeThread").mockResolvedValue(undefined);
const resumeThread = vi.spyOn(runtime.thread.resume, "resumeThread").mockResolvedValue(true);
runtime.shell.parts.toolbar.actions.status.connect();