wangxuming/makelore
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix: isolate concurrent OpenCode chat runs

Browse Source
This commit is contained in:
brother7
2026-08-17 22:03:37 +08:00
parent 7e8d9e3811
commit 65040730ec
29 changed files with 5207 additions and 427 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/stores/opencode-session-run-machine.ts
View File

@@ -143,6 +143,7 @@ export function transitionSessionRunState<TPrompt extends SessionRunQueuedPrompt
switch (event.type) {
case 'post_accepted':
return state;
case 'remote_busy':
return {
...state,
@@ -151,6 +152,7 @@ export function transitionSessionRunState<TPrompt extends SessionRunQueuedPrompt
error: null,
};
case 'remote_idle':
if (state.phase === 'posting') return state;
return {
...state,
phase: 'idle',
@@ -178,6 +180,7 @@ export function transitionSessionRunState<TPrompt extends SessionRunQueuedPrompt
return {
...state,
phase: 'idle',
queue: [],
terminalReason: 'failed',
error: event.error,
suppressNextAbortError: false,

234
src/stores/opencode.ts
View File

@@ -254,6 +254,8 @@ interface OpencodeState {
const initialStatus: OpencodeStatus = { state: 'stopped', port: 4096 };
const POLL_INTERVAL_MS = 750;
const PROMPT_START_CONFIRMATION_TIMEOUT_MS = 10_000;
const PROMPT_START_UNCONFIRMED_ERROR = 'SESSION_START_UNCONFIRMED: 运行时未确认该会话已开始，本次消息不会自动重发，请检查运行时后手动重试。';
const IDLE_SESSION_STATUS: OpencodeSessionStatus = { type: 'idle' };
const activeSessionEventSources = new Map<string, EventSource>();
const activeProjectEventSources = new Map<string, EventSource>();
@@ -277,6 +279,11 @@ function removeEventListenerIfSupported(
let projectEventReconnectTimer: ReturnType<typeof setTimeout> | null = null;
let projectEventReconnectAttempt = 0;
const activeSessionRunTokens = new Map<string, number>();
const activePromptStartWatchdogs = new Map<string, {
runToken: number;
timer: ReturnType<typeof setTimeout>;
reject: (reason: Error) => void;
}>();
const pendingSessionRunStreamBatches = new Map<string, PendingSessionRunStreamBatch>();
const drainingAbortedSessionRunIds = new Map<string, number>();
const ignoredSuppressedPolledAbortKeys = new Map<string, { runId: number; keys: Set<string> }>();
@@ -307,6 +314,58 @@ type PendingSessionRunStreamBatch = {
timer: ReturnType<typeof setTimeout> | null;
};
class PromptStartWatchdogCancelledError extends Error {}
function rejectPromptStartWatchdog(
sessionId: string,
runToken: number,
reason: Error,
): void {
const watchdog = activePromptStartWatchdogs.get(sessionId);
if (!watchdog || watchdog.runToken !== runToken) return;
activePromptStartWatchdogs.delete(sessionId);
clearTimeout(watchdog.timer);
watchdog.reject(reason);
}
function cancelPromptStartWatchdog(sessionId: string, runToken: number): void {
rejectPromptStartWatchdog(
sessionId,
runToken,
new PromptStartWatchdogCancelledError(),
);
}
function startPromptStartWatchdog(sessionId: string, runToken: number): Promise<never> {
const existing = activePromptStartWatchdogs.get(sessionId);
if (existing) cancelPromptStartWatchdog(sessionId, existing.runToken);
let rejectWatchdog!: (reason: Error) => void;
const promise = new Promise<never>((_resolve, reject) => {
rejectWatchdog = reject;
});
void promise.catch(() => undefined);
const timer = setTimeout(() => {
rejectPromptStartWatchdog(
sessionId,
runToken,
new Error(PROMPT_START_UNCONFIRMED_ERROR),
);
}, PROMPT_START_CONFIRMATION_TIMEOUT_MS);
activePromptStartWatchdogs.set(sessionId, {
runToken,
timer,
reject: rejectWatchdog,
});
return promise;
}
function cancelAllPromptStartWatchdogs(): void {
for (const [sessionId, watchdog] of activePromptStartWatchdogs) {
cancelPromptStartWatchdog(sessionId, watchdog.runToken);
}
}
function beginSessionRun(sessionId: string): number {
sessionRunNonce += 1;
const nextToken = sessionRunNonce;
@@ -316,6 +375,10 @@ function beginSessionRun(sessionId: string): number {
function invalidateSessionRun(sessionId: string): void {
ignoredSuppressedPolledAbortKeys.delete(sessionId);
const activeRunToken = activeSessionRunTokens.get(sessionId);
if (activeRunToken !== undefined) {
cancelPromptStartWatchdog(sessionId, activeRunToken);
}
beginSessionRun(sessionId);
}
@@ -324,6 +387,7 @@ function isSessionRunCurrent(sessionId: string, token: number): boolean {
}
function closeActiveSessionEventSource(): void {
cancelAllPromptStartWatchdogs();
activeSessionRunTokens.clear();
for (const sessionId of pendingSessionRunStreamBatches.keys()) {
cancelSessionRunStreamBatch(sessionId);
@@ -2426,7 +2490,6 @@ function finishSessionRunSuccessfully(
sendingSessionIds,
sendingSessionId: getSendingSessionIdForSelection(nextState),
loading: false,
...errorState(null),
...(completedTranscript && completedTranscript !== currentTranscript
? {
sessionTranscriptBySessionId: {
@@ -2552,7 +2615,6 @@ async function runSessionSubmission(
...clearSessionStreamingPatch(state, sessionId),
...runStatePatch,
selectedSessionId,
...errorState(null),
sendingSessionIds,
sendingSessionId: selectedSessionId === sessionId ? sessionId : state.sendingSessionId,
sessionStatuses: {
@@ -2571,6 +2633,42 @@ async function runSessionSubmission(
});
let liveSessionError: Error | null = null;
let promptStartAcknowledged = submission.kind !== 'prompt';
let promptStartWatchdog: Promise<never> | null = null;
const acknowledgePromptStart = () => {
if (promptStartAcknowledged) return;
promptStartAcknowledged = true;
cancelPromptStartWatchdog(sessionId, runToken);
};
const awaitDuringPromptStart = async <T>(operation: Promise<T>): Promise<T> => {
if (promptStartAcknowledged || !promptStartWatchdog) return await operation;
try {
return await Promise.race([operation, promptStartWatchdog]);
} catch (error) {
if (error instanceof PromptStartWatchdogCancelledError && promptStartAcknowledged) {
return await operation;
}
throw error;
}
};
const dispatchObservedIdle = (state: OpencodeState) => {
const runState = getSessionRunStateForSession(state, sessionId);
if (submission.kind === 'prompt' || runState.phase !== 'posting') {
return dispatchSessionRunEvent(state, sessionId, {
type: 'remote_idle',
runId: runToken,
});
}
const startedPatch = dispatchSessionRunEvent(state, sessionId, {
type: 'remote_busy',
runId: runToken,
});
return dispatchSessionRunEvent(
{ ...state, ...startedPatch },
sessionId,
{ type: 'remote_idle', runId: runToken },
);
};
try {
try {
@@ -2586,20 +2684,32 @@ async function runSessionSubmission(
if (!isSessionRunCurrent(sessionId, runToken)) return;
if (getOpencodeEventSessionId(payload) !== sessionId) return;
const status = normalizeSessionStatus(payload.status);
const acknowledgesStart = status.type === 'busy' || status.type === 'retry';
if (acknowledgesStart) acknowledgePromptStart();
if (status.type === 'idle') {
flushSessionRunStreamBatch(set, sessionId, runToken);
}
set((state) => ({
...getSessionTranscriptPatch(state, 'session.status', payload),
...dispatchSessionRunEvent(state, sessionId, {
type: status.type === 'idle' ? 'remote_idle' : 'remote_busy',
runId: runToken,
}),
sessionStatuses: {
...state.sessionStatuses,
[sessionId]: status,
},
}));
set((state) => {
const runState = getSessionRunStateForSession(state, sessionId);
const ignoreUnconfirmedIdle = submission.kind === 'prompt'
&& status.type === 'idle'
&& runState.phase === 'posting';
return {
...getSessionTranscriptPatch(state, 'session.status', payload),
...(status.type === 'idle'
? dispatchObservedIdle(state)
: dispatchSessionRunEvent(state, sessionId, {
type: 'remote_busy',
runId: runToken,
})),
sessionStatuses: {
...state.sessionStatuses,
[sessionId]: ignoreUnconfirmedIdle
? state.sessionStatuses[sessionId] ?? { type: 'busy' }
: status,
},
};
});
});
registerSessionRunEventListener(sessionId, source, 'session.idle', (event) => {
@@ -2607,17 +2717,21 @@ async function runSessionSubmission(
if (!isSessionRunCurrent(sessionId, runToken)) return;
if (getOpencodeEventSessionId(payload) !== sessionId) return;
flushSessionRunStreamBatch(set, sessionId, runToken);
set((state) => ({
...getSessionTranscriptPatch(state, 'session.idle', payload),
...dispatchSessionRunEvent(state, sessionId, {
type: 'remote_idle',
runId: runToken,
}),
sessionStatuses: {
...state.sessionStatuses,
[sessionId]: IDLE_SESSION_STATUS,
},
}));
set((state) => {
const runState = getSessionRunStateForSession(state, sessionId);
const ignoreUnconfirmedIdle = submission.kind === 'prompt'
&& runState.phase === 'posting';
return {
...getSessionTranscriptPatch(state, 'session.idle', payload),
...dispatchObservedIdle(state),
sessionStatuses: {
...state.sessionStatuses,
[sessionId]: ignoreUnconfirmedIdle
? state.sessionStatuses[sessionId] ?? { type: 'busy' }
: IDLE_SESSION_STATUS,
},
};
});
});
registerSessionRunEventListener(sessionId, source, 'session.compacted', (event) => {
@@ -2674,7 +2788,6 @@ async function runSessionSubmission(
return {
...transcriptPatch,
...runStatePatch,
...errorState(null),
};
}
const sendingSessionIds = withoutKey(state.sendingSessionIds, sessionId);
@@ -2703,7 +2816,6 @@ async function runSessionSubmission(
sendingSessionIds,
sendingSessionId: getSendingSessionIdForSelection(nextState),
loading: false,
...errorState(null),
};
});
if (shouldSuppress) return;
@@ -2713,6 +2825,7 @@ async function runSessionSubmission(
}
liveSessionError = new Error(message);
rejectPromptStartWatchdog(sessionId, runToken, liveSessionError);
closeSessionEventSource(sessionId);
set((state) => {
const transcriptPatch = getSessionTranscriptPatch(state, 'session.error', payload);
@@ -2744,7 +2857,6 @@ async function runSessionSubmission(
sendingSessionIds,
sendingSessionId: getSendingSessionIdForSelection(nextState),
loading: false,
...errorState(message),
};
});
});
@@ -2754,7 +2866,12 @@ async function runSessionSubmission(
if (!isSessionRunCurrent(sessionId, runToken)) return;
const question = normalizeQuestionRequest(payload);
if (!question || question.sessionID !== sessionId) return;
acknowledgePromptStart();
set((state) => ({
...dispatchSessionRunEvent(state, sessionId, {
type: 'remote_busy',
runId: runToken,
}),
pendingQuestions: mergePendingQuestion(state.pendingQuestions, question),
}));
});
@@ -2794,7 +2911,12 @@ async function runSessionSubmission(
if (!isSessionRunCurrent(sessionId, runToken)) return;
const permission = normalizePermissionRequest(payload);
if (!permission || permission.sessionID !== sessionId) return;
acknowledgePromptStart();
set((state) => ({
...dispatchSessionRunEvent(state, sessionId, {
type: 'remote_busy',
runId: runToken,
}),
pendingPermissions: mergePendingPermission(state.pendingPermissions, permission),
}));
});
@@ -2803,6 +2925,13 @@ async function runSessionSubmission(
const payload = JSON.parse((event as MessageEvent<string>).data) as Record<string, unknown>;
if (!isSessionRunCurrent(sessionId, runToken)) return;
if (getOpencodeEventSessionId(payload) !== sessionId) return;
if (createStreamingMessageFromEvent(payload)?.role === 'assistant') {
acknowledgePromptStart();
set((state) => dispatchSessionRunEvent(state, sessionId, {
type: 'remote_busy',
runId: runToken,
}));
}
queueSessionRunStreamEvent(set, sessionId, runToken, 'message.updated', payload);
});
@@ -2849,6 +2978,9 @@ async function runSessionSubmission(
if (!response.success) {
throw new Error(response.error || `Failed to start ${submission.kind}`);
}
if (submission.kind === 'prompt' && !promptStartAcknowledged && !liveSessionError) {
promptStartWatchdog = startPromptStartWatchdog(sessionId, runToken);
}
try {
options?.onHostAccepted?.();
} catch {
@@ -2857,6 +2989,10 @@ async function runSessionSubmission(
if (liveSessionError) {
throw liveSessionError;
}
set((state) => dispatchSessionRunEvent(state, sessionId, {
type: submission.kind === 'prompt' ? 'post_accepted' : 'remote_busy',
runId: runToken,
}));
if (submission.completion === 'post-success') {
if (!isSessionRunCurrent(sessionId, runToken)) {
return getSessionMessagesForState(get(), sessionId);
@@ -2880,11 +3016,6 @@ async function runSessionSubmission(
if (!isSessionRunCurrent(sessionId, runToken)) {
return getSessionMessagesForState(get(), sessionId);
}
set((state) => dispatchSessionRunEvent(state, sessionId, {
type: 'post_accepted',
runId: runToken,
}));
let messages: RawMessage[] = getSessionMessagesForState(get(), sessionId);
let status: OpencodeSessionStatus = { type: 'busy' };
const useBaselineInference = submission.kind !== 'prompt';
@@ -2914,7 +3045,9 @@ async function runSessionSubmission(
|| eventStatus?.type === 'idle';
let statuses: OpencodeSessionStatusMap = {};
if (shouldPollStatus) {
const statusResponse = await hostApiFetch<OpencodeSessionStatusActionResponse>('/api/opencode/sessions/status');
const statusResponse = await awaitDuringPromptStart(
hostApiFetch<OpencodeSessionStatusActionResponse>('/api/opencode/sessions/status'),
);
statuses = normalizeSessionStatusMap(statusResponse.statuses);
nextStatusFallbackPollAt = Date.now() + SESSION_STATUS_FALLBACK_POLL_MS;
}
@@ -2926,7 +3059,7 @@ async function runSessionSubmission(
|| observedStatus?.type === 'idle';
let refreshedMessages = messages;
if (shouldRefreshMessages) {
refreshedMessages = await fetchSessionMessages(sessionId);
refreshedMessages = await awaitDuringPromptStart(fetchSessionMessages(sessionId));
nextMessageFallbackPollAt = Date.now() + SESSION_MESSAGE_FALLBACK_POLL_MS;
}
if (liveSessionError) {
@@ -2952,6 +3085,13 @@ async function runSessionSubmission(
: sendingUserMessage
? getAssistantErrorAfterPrompt(refreshedMessages, sendingUserMessage)
: null;
const assistantResponded = submission.kind === 'prompt'
&& sendingUserMessage
&& hasAssistantResponseAfterPrompt(refreshedMessages, sendingUserMessage);
const explicitBusy = httpStatus?.type === 'busy' || httpStatus?.type === 'retry';
if (explicitBusy || assistantResponded) {
acknowledgePromptStart();
}
const inferredStatus = useBaselineInference
? inferMissingPolledStatusAfterBaseline(refreshedMessages, baseline)
: sendingUserMessage
@@ -2960,13 +3100,16 @@ async function runSessionSubmission(
// A missing HTTP status is common while the runtime is transitioning.
// Prefer a newly observed assistant response over the optimistic/event
// busy marker, otherwise a completed run can remain busy forever.
const polledStatus = submission.kind === 'compact'
const observedPolledStatus = submission.kind === 'compact'
? shouldPollStatus
? httpStatus ?? eventStatus ?? status
: eventStatus ?? status
: shouldPollStatus
? httpStatus ?? inferredStatus
: eventStatus ?? status ?? inferredStatus;
const polledStatus = submission.kind === 'prompt' && !promptStartAcknowledged
? { type: 'busy' as const }
: observedPolledStatus;
status = assistantError || assistantAborted ? IDLE_SESSION_STATUS : polledStatus;
const currentMessages = getSessionMessagesForState(get(), sessionId);
messages = assistantAborted || status.type === 'idle'
@@ -2996,7 +3139,6 @@ async function runSessionSubmission(
...statuses,
[sessionId]: polledStatus,
},
...errorState(null),
};
}
@@ -3026,13 +3168,12 @@ async function runSessionSubmission(
sendingSessionIds,
sendingSessionId: getSendingSessionIdForSelection(nextState),
loading: false,
...errorState(null),
};
});
if (shouldSuppressAbort) {
rememberIgnoredSuppressedPolledAbortKey(sessionId, runToken, assistantAbortKey);
await sleep(POLL_INTERVAL_MS);
await awaitDuringPromptStart(sleep(POLL_INTERVAL_MS));
continue;
}
@@ -3045,6 +3186,12 @@ async function runSessionSubmission(
if (shouldRefreshMessages || shouldPollStatus || assistantError) {
set((state) => ({
...getSessionMessagePatch(state, sessionId, messages),
...(promptStartAcknowledged
? dispatchSessionRunEvent(state, sessionId, {
type: 'remote_busy',
runId: runToken,
})
: {}),
...(assistantError
? dispatchSessionRunEvent(state, sessionId, {
type: 'remote_failed',
@@ -3057,7 +3204,6 @@ async function runSessionSubmission(
...statuses,
[sessionId]: status,
},
...errorState(assistantError ?? null),
}));
}
@@ -3087,7 +3233,7 @@ async function runSessionSubmission(
return messages;
}
await sleep(POLL_INTERVAL_MS);
await awaitDuringPromptStart(sleep(POLL_INTERVAL_MS));
}
} catch (error) {
if (!isSessionRunCurrent(sessionId, runToken)) {
@@ -3132,12 +3278,13 @@ async function runSessionSubmission(
sendingSessionIds,
sendingSessionId: getSendingSessionIdForSelection(nextState),
loading: false,
...errorState(errorMessage),
};
});
invalidateSessionRun(sessionId);
clearDrainingAbortedRunIfForPreviousRun(sessionId, runToken);
throw error;
} finally {
cancelPromptStartWatchdog(sessionId, runToken);
}
}
@@ -3653,7 +3800,6 @@ export const useOpencodeStore = create<OpencodeState>((set, get) => ({
sendingSessionIds,
sendingSessionId: getSendingSessionIdForSelection(nextState),
loading: true,
...errorState(null),
};
});
try {
@@ -3692,7 +3838,6 @@ export const useOpencodeStore = create<OpencodeState>((set, get) => ({
return {
...clearSessionRunAbortSuppression(state, sessionId),
loading: false,
...errorState(null),
};
}
@@ -3715,7 +3860,6 @@ export const useOpencodeStore = create<OpencodeState>((set, get) => ({
sendingSessionIds,
sendingSessionId: getSendingSessionIdForSelection(nextState),
loading: false,
...errorState(null),
};
});
} catch (error) {
@@ -3731,7 +3875,6 @@ export const useOpencodeStore = create<OpencodeState>((set, get) => ({
})
: clearSessionRunAbortSuppression(state, sessionId)),
loading: false,
...errorState(errorMessage),
}));
throw error;
}
@@ -4249,7 +4392,6 @@ export const useOpencodeStore = create<OpencodeState>((set, get) => ({
return {
...getSessionMessagePatch(currentState, sessionId, messages),
...runStatePatch,
...errorState(null),
};
});
try {

15
src/stores/providers.ts
View File

@@ -38,9 +38,13 @@ interface ProviderState {
refreshProviderSnapshot: () => Promise<void>;
createAccount: (account: ProviderAccount, apiKey?: string) => Promise<void>;
removeAccount: (accountId: string) => Promise<void>;
importUserModelConfig: (accessToken: string) => Promise<{
importUserModelConfig: (
accessToken: string,
options?: { runtimeRefresh?: 'apply' | 'defer' },
) => Promise<{
account: ProviderAccount;
importedModels: string[];
runtimeRefreshRequired: boolean;
}>;
validateAccountApiKey: (
accountId: string,
@@ -218,16 +222,20 @@ export const useProviderStore = create<ProviderState>((set, get) => ({
deleteAccount: async (accountId) => get().removeAccount(accountId),
importUserModelConfig: async (accessToken) => {
importUserModelConfig: async (accessToken, options) => {
try {
const result = await hostApiFetch<{
success: boolean;
error?: string;
account?: ProviderAccount;
importedModels?: string[];
runtimeRefreshRequired?: boolean;
}>('/api/provider-accounts/import-user-model-config', {
method: 'POST',
body: JSON.stringify({ accessToken }),
body: JSON.stringify({
accessToken,
runtimeRefresh: options?.runtimeRefresh ?? 'apply',
}),
});
if (!result.success || !result.account) {
@@ -238,6 +246,7 @@ export const useProviderStore = create<ProviderState>((set, get) => ({
return {
account: result.account,
importedModels: result.importedModels ?? [],
runtimeRefreshRequired: result.runtimeRefreshRequired === true,
};
} catch (error) {
console.error('Failed to import current user model config:', error);