mirror of
https://github.com/murashit/codex-panel.git
synced 2026-07-22 06:57:10 +00:00
refactor(chat): replace rollback RPC with marker fork
This commit is contained in:
murashit
parent
01a20dcf70
commit
bb673b2a75
9 changed files with 437 additions and 171 deletions
|
|
@ -30,7 +30,6 @@ import type { ThreadItemsListResponse } from "../../generated/app-server/v2/Thre
|
|||
import type { ThreadListResponse } from "../../generated/app-server/v2/ThreadListResponse";
|
||||
import type { ThreadReadResponse } from "../../generated/app-server/v2/ThreadReadResponse";
|
||||
import type { ThreadResumeResponse } from "../../generated/app-server/v2/ThreadResumeResponse";
|
||||
import type { ThreadRollbackResponse } from "../../generated/app-server/v2/ThreadRollbackResponse";
|
||||
import type { ThreadSetNameResponse } from "../../generated/app-server/v2/ThreadSetNameResponse";
|
||||
import type { ThreadSettingsUpdateResponse } from "../../generated/app-server/v2/ThreadSettingsUpdateResponse";
|
||||
import type { ThreadStartResponse } from "../../generated/app-server/v2/ThreadStartResponse";
|
||||
|
|
@ -87,7 +86,6 @@ export interface ClientResponseByMethod {
|
|||
"thread/delete": ThreadDeleteResponse;
|
||||
"thread/unsubscribe": ThreadUnsubscribeResponse;
|
||||
"thread/unarchive": ThreadUnarchiveResponse;
|
||||
"thread/rollback": ThreadRollbackResponse;
|
||||
"thread/name/set": ThreadSetNameResponse;
|
||||
"thread/settings/update": ThreadSettingsUpdateResponse;
|
||||
"thread/turns/list": ThreadTurnsListResponse;
|
||||
|
|
|
|||
|
|
@ -6,7 +6,6 @@ import type { RuntimeServiceTierRequest, RuntimeSettingsPatch } from "../../doma
|
|||
import type { ThreadActivationSnapshot } from "../../domain/threads/activation";
|
||||
import type { ArchiveThreadInput } from "../../domain/threads/archive-markdown";
|
||||
import type { ThreadGoal, ThreadGoalUpdate } from "../../domain/threads/goal";
|
||||
import type { HistoricalTurn } from "../../domain/threads/history";
|
||||
import type { Thread } from "../../domain/threads/model";
|
||||
import { REFERENCED_THREAD_TURN_LIMIT } from "../../domain/threads/reference";
|
||||
import type { TurnTranscriptSummary } from "../../domain/threads/transcript";
|
||||
|
|
@ -187,44 +186,61 @@ export async function readReferencedThreadTurnTranscriptSummaries(
|
|||
return chronologicalTurnTranscriptSummariesFromTurnRecords(response.data);
|
||||
}
|
||||
|
||||
export interface ThreadRollbackSnapshot {
|
||||
thread: Thread;
|
||||
turns: readonly HistoricalTurn[];
|
||||
}
|
||||
type ThreadForkPosition =
|
||||
| { readonly kind: "through-turn"; readonly turnId: string }
|
||||
| { readonly kind: "before-turn"; readonly turnId: string };
|
||||
|
||||
interface ThreadRollbackResult {
|
||||
readonly thread: ThreadRecord & {
|
||||
readonly turns: readonly HistoricalTurn[];
|
||||
interface AppServerForkThreadOptions {
|
||||
readonly position?: ThreadForkPosition;
|
||||
readonly deferGoalContinuation?: boolean;
|
||||
readonly runtime?: {
|
||||
readonly model?: string;
|
||||
readonly reasoningEffort?: string | null;
|
||||
readonly serviceTier?: ServiceTier | null;
|
||||
readonly approvalPolicy?: RuntimePermissionState["approvalPolicy"];
|
||||
readonly approvalsReviewer?: ApprovalsReviewer;
|
||||
readonly permissions?: string;
|
||||
readonly sandboxPolicy?: RuntimePermissionState["sandboxPolicy"];
|
||||
};
|
||||
}
|
||||
|
||||
function threadRollbackSnapshotFromAppServerResponse(response: ThreadRollbackResult): ThreadRollbackSnapshot {
|
||||
return {
|
||||
thread: threadFromThreadRecord(response.thread),
|
||||
turns: response.thread.turns,
|
||||
};
|
||||
}
|
||||
|
||||
export async function rollbackThread(client: AppServerRequestClient, threadId: string, numTurns?: number): Promise<ThreadRollbackSnapshot> {
|
||||
const response = await client.request("thread/rollback", { threadId, numTurns: numTurns ?? 1 });
|
||||
return threadRollbackSnapshotFromAppServerResponse(response);
|
||||
}
|
||||
|
||||
export async function forkThread(
|
||||
client: AppServerRequestClient,
|
||||
threadId: string,
|
||||
cwd: string,
|
||||
lastTurnId: string | null = null,
|
||||
options: AppServerForkThreadOptions = {},
|
||||
): Promise<Thread> {
|
||||
const position = options.position;
|
||||
const runtime = options.runtime;
|
||||
const permissions = runtime?.permissions;
|
||||
const sandbox = permissions === undefined ? appServerSandboxMode(runtime?.sandboxPolicy) : undefined;
|
||||
const response = await client.request("thread/fork", {
|
||||
threadId,
|
||||
cwd,
|
||||
excludeTurns: true,
|
||||
...(lastTurnId ? { lastTurnId } : {}),
|
||||
...(position?.kind === "through-turn" ? { lastTurnId: position.turnId } : {}),
|
||||
...(position?.kind === "before-turn" ? { beforeTurnId: position.turnId } : {}),
|
||||
...(options.deferGoalContinuation ? { deferGoalContinuation: true } : {}),
|
||||
...(runtime?.model !== undefined ? { model: runtime.model } : {}),
|
||||
...(runtime?.serviceTier !== undefined ? { serviceTier: runtime.serviceTier } : {}),
|
||||
...(runtime?.approvalPolicy !== undefined && runtime.approvalPolicy !== null ? { approvalPolicy: runtime.approvalPolicy } : {}),
|
||||
...(runtime?.approvalsReviewer !== undefined ? { approvalsReviewer: runtime.approvalsReviewer } : {}),
|
||||
...(permissions !== undefined ? { permissions } : {}),
|
||||
...(sandbox !== undefined ? { sandbox } : {}),
|
||||
...(runtime?.reasoningEffort !== undefined ? { config: { model_reasoning_effort: runtime.reasoningEffort } } : {}),
|
||||
});
|
||||
return threadFromThreadRecord(response.thread);
|
||||
}
|
||||
|
||||
function appServerSandboxMode(
|
||||
policy: RuntimePermissionState["sandboxPolicy"] | undefined,
|
||||
): "read-only" | "workspace-write" | "danger-full-access" | undefined {
|
||||
if (!policy || policy.type === "externalSandbox") return undefined;
|
||||
if (policy.type === "dangerFullAccess") return "danger-full-access";
|
||||
if (policy.type === "readOnly") return "read-only";
|
||||
return "workspace-write";
|
||||
}
|
||||
|
||||
export interface EphemeralThreadForkSnapshot {
|
||||
readonly activation: ThreadActivationSnapshot;
|
||||
readonly sourceThreadId: string;
|
||||
|
|
|
|||
|
|
@ -10,7 +10,6 @@ import {
|
|||
readThreadGoal,
|
||||
recordThreadGoalUserMessage,
|
||||
resumeThread,
|
||||
rollbackThread,
|
||||
setThreadGoal,
|
||||
startThread,
|
||||
threadActivationSnapshotFromAppServerResponse,
|
||||
|
|
@ -30,7 +29,7 @@ import type {
|
|||
ThreadResumeSnapshot,
|
||||
ThreadResumeTransport,
|
||||
} from "../../application/threads/thread-loading-transport";
|
||||
import type { ThreadMutationTransport, ThreadRollbackSnapshot } from "../../application/threads/thread-mutation-transport";
|
||||
import type { ThreadMutationTransport } from "../../application/threads/thread-mutation-transport";
|
||||
import type { ThreadStartTransport } from "../../application/threads/thread-start-transport";
|
||||
import type { ThreadSubscriptionTransport } from "../../application/threads/thread-subscription-transport";
|
||||
import type { ChatTurnTransport } from "../../application/turns/turn-transport";
|
||||
|
|
@ -166,16 +165,8 @@ function createChatThreadMutationTransport(host: ChatAppServerTransportHost): Th
|
|||
runCurrentChatAppServerEffect(host, async (client) => {
|
||||
await compactThread(client, threadId);
|
||||
}),
|
||||
forkThread: (threadId, lastTurnId = null) =>
|
||||
runCurrentChatAppServerEffect(host, (client) => forkThread(client, threadId, host.vaultPath, lastTurnId)),
|
||||
rollbackThread: (threadId) =>
|
||||
runCurrentChatAppServerEffect(host, async (client): Promise<ThreadRollbackSnapshot> => {
|
||||
const snapshot = await rollbackThread(client, threadId);
|
||||
return {
|
||||
thread: snapshot.thread,
|
||||
items: threadStreamItemsFromTurns(snapshot.turns),
|
||||
};
|
||||
}),
|
||||
forkThread: (threadId, options) =>
|
||||
runCurrentChatAppServerEffect(host, (client) => forkThread(client, threadId, host.vaultPath, options)),
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -2,9 +2,8 @@ import { inheritedForkThreadName, type Thread } from "../../../../domain/threads
|
|||
import { activeThreadRuntimeState } from "../../domain/runtime/state";
|
||||
import { effectCompleted, effectCompletedInCurrentContext } from "../effect-outcome";
|
||||
import { type ActivePanelOperation, activePanelOperationDecision } from "../panel-operation-policy";
|
||||
import { resumedThreadActionFromActiveRuntime } from "../state/actions";
|
||||
import { capturePanelTargetLease, type PanelTargetLease, panelTargetLeaseIsCurrent } from "../state/panel-target";
|
||||
import { activeThreadId, type ChatAction, type ChatState } from "../state/root-reducer";
|
||||
import { activeThreadId, type ChatState } from "../state/root-reducer";
|
||||
import type { ChatStateStore } from "../state/store";
|
||||
import { threadStreamRollbackCandidate, threadStreamTurnsAfterTurnId } from "../state/thread-stream";
|
||||
import { chatTurnBusy } from "../turns/turn-state";
|
||||
|
|
@ -23,8 +22,7 @@ export interface ThreadManagementActionsHost {
|
|||
setStatus: (status: string) => void;
|
||||
setComposerText: (text: string) => void;
|
||||
openThreadInNewView: (threadId: string) => Promise<void>;
|
||||
openThreadInCurrentPanel: (threadId: string) => Promise<void>;
|
||||
notifyActiveThreadIdentityChanged: () => void;
|
||||
openThreadInCurrentPanel: (threadId: string, onAdopted: () => void) => Promise<{ adopted: boolean }>;
|
||||
recordThread: (thread: Thread) => void;
|
||||
threadHasPendingOrRunningPanel: (threadId: string) => boolean;
|
||||
}
|
||||
|
|
@ -138,7 +136,9 @@ async function forkThreadFromTurn(
|
|||
const sourceName = inheritedForkThreadName(threadId, threadManagementState(host).threadList.listedThreads);
|
||||
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);
|
||||
const effect = turnId
|
||||
? await host.threadTransport.forkThread(threadId, { position: { kind: "through-turn", turnId } })
|
||||
: await host.threadTransport.forkThread(threadId);
|
||||
if (!effectCompleted(effect)) return;
|
||||
const forkedThread = effect.value;
|
||||
const forkedThreadId = forkedThread.id;
|
||||
|
|
@ -156,15 +156,24 @@ async function forkThreadFromTurn(
|
|||
if (!threadManagementScopeStillTargetsOriginalPanel(host, scope)) return;
|
||||
}
|
||||
if (archiveSource) {
|
||||
if (!(await archiveThreadFromPanel(host, threadId))) return;
|
||||
if (!threadManagementScopeStillTargetsOriginalPanel(host, scope)) return;
|
||||
let adoption: { adopted: boolean };
|
||||
try {
|
||||
await host.openThreadInCurrentPanel(forkedThreadId);
|
||||
adoption = await host.openThreadInCurrentPanel(forkedThreadId, () => undefined);
|
||||
} 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}`);
|
||||
host.addSystemMessage(`Forked thread ${forkedThreadId}, but could not open it in the current panel: ${message}`);
|
||||
return;
|
||||
}
|
||||
if (!adoption.adopted) {
|
||||
if (threadManagementScopeStillTargetsOriginalPanel(host, scope)) {
|
||||
host.addSystemMessage(`Forked thread ${forkedThreadId}, but could not open it in the current panel.`);
|
||||
}
|
||||
return;
|
||||
}
|
||||
await archiveReplacedSource(host, threadId, forkedThreadId, {
|
||||
failureMessage: "Forked the thread, but could not archive the previous version",
|
||||
});
|
||||
return;
|
||||
}
|
||||
try {
|
||||
|
|
@ -204,36 +213,52 @@ async function rollbackThread(host: ThreadManagementActionsHost, threadId: strin
|
|||
host.addSystemMessage("No completed turn to roll back.");
|
||||
return;
|
||||
}
|
||||
const runtime = activeThreadRuntimeState(threadManagementState(host).runtime);
|
||||
const runtimeOverrides = {
|
||||
...(runtime.model ? { model: runtime.model } : {}),
|
||||
reasoningEffort: runtime.reasoningEffort,
|
||||
...(runtime.serviceTierKnown ? { serviceTier: runtime.serviceTier } : {}),
|
||||
...(runtime.approvalPolicyKnown && runtime.approvalPolicy ? { approvalPolicy: runtime.approvalPolicy } : {}),
|
||||
...(runtime.approvalsReviewer ? { approvalsReviewer: runtime.approvalsReviewer } : {}),
|
||||
...(runtime.permissionProfileKnown && runtime.activePermissionProfile
|
||||
? { permissions: runtime.activePermissionProfile.id }
|
||||
: runtime.sandboxPolicyKnown && runtime.sandboxPolicy
|
||||
? { sandboxPolicy: runtime.sandboxPolicy }
|
||||
: {}),
|
||||
};
|
||||
|
||||
try {
|
||||
host.setStatus(STATUS_ROLLBACK_STARTING);
|
||||
if (!(await host.threadTransport.ensureConnected())) return;
|
||||
if (!threadManagementScopeStillTargetsPanel(host, scope)) return;
|
||||
const effect = await host.threadTransport.rollbackThread(threadId);
|
||||
if (!effectCompleted(effect)) return;
|
||||
if (!effectCompletedInCurrentContext(effect)) return;
|
||||
const snapshot = effect.value;
|
||||
if (!threadManagementScopeStillTargetsPanel(host, scope)) return;
|
||||
threadManagementDispatch(
|
||||
host,
|
||||
resumedThreadActionFromActiveRuntime({
|
||||
thread: snapshot.thread,
|
||||
runtime: activeThreadRuntimeState(threadManagementState(host).runtime),
|
||||
listedThreads: threadManagementState(host).threadList.listedThreads,
|
||||
expectedPanelTargetRevision: scope.panelTarget.revision,
|
||||
}),
|
||||
);
|
||||
threadManagementDispatch(host, {
|
||||
type: "thread-stream/items-replaced",
|
||||
items: snapshot.items,
|
||||
historyCursor: null,
|
||||
loadingHistory: false,
|
||||
const effect = await host.threadTransport.forkThread(threadId, {
|
||||
position: { kind: "before-turn", turnId: candidate.turnId },
|
||||
deferGoalContinuation: true,
|
||||
runtime: runtimeOverrides,
|
||||
});
|
||||
if (!effectCompleted(effect)) return;
|
||||
const forkedThread = effect.value;
|
||||
if (effectCompletedInCurrentContext(effect)) host.recordThread(forkedThread);
|
||||
else return;
|
||||
if (!threadManagementScopeStillTargetsPanel(host, scope)) return;
|
||||
const adoption = await host.openThreadInCurrentPanel(forkedThread.id, () => {
|
||||
host.setComposerText(candidate.text);
|
||||
});
|
||||
if (!adoption.adopted) {
|
||||
if (threadManagementScopeStillTargetsPanel(host, scope)) {
|
||||
host.addSystemMessage("The rolled-back version was created but could not be opened in this panel. Open it from thread history.");
|
||||
host.setStatus(STATUS_ROLLBACK_FAILED);
|
||||
}
|
||||
return;
|
||||
}
|
||||
if (activeThreadId(threadManagementState(host)) === forkedThread.id) {
|
||||
host.addSystemMessage("Rolled back the latest turn. Local file changes were not reverted.");
|
||||
host.setStatus(STATUS_ROLLBACK_COMPLETE);
|
||||
}
|
||||
await archiveReplacedSource(host, threadId, forkedThread.id, {
|
||||
saveMarkdown: false,
|
||||
failureMessage: "Rolled back the latest turn, but could not archive the previous version",
|
||||
});
|
||||
host.setComposerText(candidate.text);
|
||||
host.addSystemMessage("Rolled back the latest turn. Local file changes were not reverted.");
|
||||
host.setStatus(STATUS_ROLLBACK_COMPLETE);
|
||||
host.notifyActiveThreadIdentityChanged();
|
||||
host.recordThread(snapshot.thread);
|
||||
} catch (error) {
|
||||
if (!threadManagementScopeStillTargetsPanel(host, scope)) return;
|
||||
host.addSystemMessage(error instanceof Error ? error.message : String(error));
|
||||
|
|
@ -241,6 +266,32 @@ async function rollbackThread(host: ThreadManagementActionsHost, threadId: strin
|
|||
}
|
||||
}
|
||||
|
||||
async function archiveReplacedSource(
|
||||
host: ThreadManagementActionsHost,
|
||||
sourceThreadId: string,
|
||||
replacementThreadId: string,
|
||||
options: { readonly saveMarkdown?: boolean; readonly failureMessage: string },
|
||||
): Promise<void> {
|
||||
try {
|
||||
const archiveOptions = options.saveMarkdown === undefined ? {} : { saveMarkdown: options.saveMarkdown };
|
||||
if (await host.operations.archiveThread(sourceThreadId, archiveOptions)) return;
|
||||
reportReplacementArchiveFailure(host, replacementThreadId, options.failureMessage, "archive was not completed");
|
||||
} catch (error) {
|
||||
const message = error instanceof Error ? error.message : String(error);
|
||||
reportReplacementArchiveFailure(host, replacementThreadId, options.failureMessage, message);
|
||||
}
|
||||
}
|
||||
|
||||
function reportReplacementArchiveFailure(
|
||||
host: ThreadManagementActionsHost,
|
||||
replacementThreadId: string,
|
||||
failureMessage: string,
|
||||
detail: string,
|
||||
): void {
|
||||
if (activeThreadId(threadManagementState(host)) !== replacementThreadId) return;
|
||||
host.addSystemMessage(`${failureMessage}: ${detail}`);
|
||||
}
|
||||
|
||||
function activePanelOperationBlocked(host: ThreadManagementActionsHost, threadId: string, operation: ActivePanelOperation): boolean {
|
||||
const state = threadManagementState(host);
|
||||
if (activeThreadId(state) !== threadId) return false;
|
||||
|
|
@ -254,10 +305,6 @@ function threadManagementState(host: ThreadManagementActionsHost): ChatState {
|
|||
return host.stateStore.getState();
|
||||
}
|
||||
|
||||
function threadManagementDispatch(host: ThreadManagementActionsHost, action: ChatAction): void {
|
||||
host.stateStore.dispatch(action);
|
||||
}
|
||||
|
||||
function captureThreadManagementPanelScope(host: ThreadManagementActionsHost, targetThreadId: string): ThreadManagementPanelScope {
|
||||
return {
|
||||
targetThreadId,
|
||||
|
|
|
|||
|
|
@ -1,15 +1,31 @@
|
|||
import type { ReasoningEffort } from "../../../../domain/catalog/metadata";
|
||||
import type { RuntimeApprovalPolicy, RuntimeSandboxPolicy } from "../../../../domain/runtime/permissions";
|
||||
import type { ApprovalsReviewer, ServiceTier } from "../../../../domain/runtime/policy";
|
||||
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;
|
||||
items: ThreadStreamItem[];
|
||||
type ThreadForkPosition =
|
||||
| { readonly kind: "through-turn"; readonly turnId: string }
|
||||
| { readonly kind: "before-turn"; readonly turnId: string };
|
||||
|
||||
interface ThreadForkOptions {
|
||||
readonly position?: ThreadForkPosition;
|
||||
readonly deferGoalContinuation?: boolean;
|
||||
readonly runtime?: ThreadForkRuntimeOverrides;
|
||||
}
|
||||
|
||||
interface ThreadForkRuntimeOverrides {
|
||||
readonly model?: string;
|
||||
readonly reasoningEffort?: ReasoningEffort | null;
|
||||
readonly serviceTier?: ServiceTier | null;
|
||||
readonly approvalPolicy?: RuntimeApprovalPolicy;
|
||||
readonly approvalsReviewer?: ApprovalsReviewer;
|
||||
readonly permissions?: string;
|
||||
readonly sandboxPolicy?: RuntimeSandboxPolicy;
|
||||
}
|
||||
|
||||
export interface ThreadMutationTransport {
|
||||
ensureConnected(): Promise<boolean>;
|
||||
compactThread(threadId: string): Promise<EffectOutcome<void>>;
|
||||
forkThread(threadId: string, lastTurnId?: string | null): Promise<EffectOutcome<Thread>>;
|
||||
rollbackThread(threadId: string): Promise<EffectOutcome<ThreadRollbackSnapshot>>;
|
||||
forkThread(threadId: string, options?: ThreadForkOptions): Promise<EffectOutcome<Thread>>;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -253,7 +253,7 @@ export function createThreadLifecycleBundle(
|
|||
}
|
||||
|
||||
export function createThreadActionBundle(host: ChatPanelThreadHost, input: ChatPanelThreadActionInput): ChatPanelThreadActionBundle {
|
||||
const { appServer, status, composerController, foundation, lifecycle, notifyActiveThreadIdentityChanged } = input;
|
||||
const { appServer, status, composerController, foundation, lifecycle } = input;
|
||||
const { environment, stateStore } = host;
|
||||
const threadManagementHost: ThreadManagementActionsHost = {
|
||||
stateStore,
|
||||
|
|
@ -271,10 +271,16 @@ export function createThreadActionBundle(host: ChatPanelThreadHost, input: ChatP
|
|||
composerController.setDraft(text, { focus: true });
|
||||
},
|
||||
openThreadInNewView: (threadId) => environment.plugin.workspace.openThreadInNewView(threadId),
|
||||
openThreadInCurrentPanel: async (threadId) => {
|
||||
await lifecycle.resume.resumeThread(threadId);
|
||||
openThreadInCurrentPanel: async (threadId, onAdopted) => {
|
||||
let adopted = false;
|
||||
await lifecycle.resume.resumeThread(threadId, undefined, {
|
||||
onAdopted: () => {
|
||||
adopted = true;
|
||||
onAdopted();
|
||||
},
|
||||
});
|
||||
return { adopted };
|
||||
},
|
||||
notifyActiveThreadIdentityChanged,
|
||||
recordThread: (thread) => {
|
||||
environment.plugin.threadCatalog.apply({ type: "thread-upserted", thread });
|
||||
},
|
||||
|
|
|
|||
|
|
@ -241,6 +241,24 @@ describe("chat app-server transports", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("forks through a selected turn with an inclusive marker", async () => {
|
||||
const request = vi.fn().mockResolvedValue({ thread: threadRecord("forked") });
|
||||
const client = { request } as unknown as AppServerClient;
|
||||
const transport = createTestGateway({
|
||||
currentClient: () => client,
|
||||
connectedClient: vi.fn().mockResolvedValue(client),
|
||||
}).threadMutation;
|
||||
|
||||
await transport.forkThread("source", { position: { kind: "through-turn", turnId: "turn-2" } });
|
||||
|
||||
expect(request).toHaveBeenCalledWith("thread/fork", {
|
||||
threadId: "source",
|
||||
cwd: "/vault",
|
||||
excludeTurns: true,
|
||||
lastTurnId: "turn-2",
|
||||
});
|
||||
});
|
||||
|
||||
it("drops stale fork transport responses after the current client changes", async () => {
|
||||
const fork = deferred<{ thread: ThreadRecord }>();
|
||||
const firstClient = { request: vi.fn().mockReturnValue(fork.promise) } as unknown as AppServerClient;
|
||||
|
|
@ -261,24 +279,89 @@ describe("chat app-server transports", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("projects rollback turns into thread stream items", async () => {
|
||||
const request = vi.fn().mockResolvedValue({ thread: threadRecord("thread", [turn([userMessage("u1", "prompt")])]) });
|
||||
it("forks before a marker turn while deferring goal continuation", async () => {
|
||||
const request = vi.fn().mockResolvedValue({ thread: threadRecord("forked") });
|
||||
const client = { request } as unknown as AppServerClient;
|
||||
const transport = createTestGateway({
|
||||
currentClient: () => client,
|
||||
connectedClient: vi.fn().mockResolvedValue(client),
|
||||
}).threadMutation;
|
||||
|
||||
const snapshot = await transport.rollbackThread("thread");
|
||||
const outcome = await transport.forkThread("source", {
|
||||
position: { kind: "before-turn", turnId: "turn-3" },
|
||||
deferGoalContinuation: true,
|
||||
});
|
||||
|
||||
expect(request).toHaveBeenCalledWith("thread/rollback", { threadId: "thread", numTurns: 1 });
|
||||
expect(snapshot).toMatchObject({
|
||||
expect(request).toHaveBeenCalledWith("thread/fork", {
|
||||
threadId: "source",
|
||||
cwd: "/vault",
|
||||
excludeTurns: true,
|
||||
beforeTurnId: "turn-3",
|
||||
deferGoalContinuation: true,
|
||||
});
|
||||
expect(outcome).toMatchObject({
|
||||
kind: "completed-current",
|
||||
value: {
|
||||
thread: { id: "thread" },
|
||||
items: [expect.objectContaining({ kind: "dialogue", role: "user", text: "prompt" })],
|
||||
value: { id: "forked" },
|
||||
});
|
||||
});
|
||||
|
||||
it("applies active runtime overrides to a rollback fork", async () => {
|
||||
const request = vi.fn().mockResolvedValue({ thread: threadRecord("forked") });
|
||||
const client = { request } as unknown as AppServerClient;
|
||||
const transport = createTestGateway({
|
||||
currentClient: () => client,
|
||||
connectedClient: vi.fn().mockResolvedValue(client),
|
||||
}).threadMutation;
|
||||
|
||||
await transport.forkThread("source", {
|
||||
position: { kind: "before-turn", turnId: "turn-3" },
|
||||
deferGoalContinuation: true,
|
||||
runtime: {
|
||||
model: "gpt-5.6",
|
||||
reasoningEffort: "high",
|
||||
serviceTier: "priority",
|
||||
approvalPolicy: "never",
|
||||
approvalsReviewer: "auto_review",
|
||||
permissions: ":workspace",
|
||||
},
|
||||
});
|
||||
|
||||
expect(request).toHaveBeenCalledWith("thread/fork", {
|
||||
threadId: "source",
|
||||
cwd: "/vault",
|
||||
excludeTurns: true,
|
||||
beforeTurnId: "turn-3",
|
||||
deferGoalContinuation: true,
|
||||
model: "gpt-5.6",
|
||||
serviceTier: "priority",
|
||||
approvalPolicy: "never",
|
||||
approvalsReviewer: "auto_review",
|
||||
permissions: ":workspace",
|
||||
config: { model_reasoning_effort: "high" },
|
||||
});
|
||||
});
|
||||
|
||||
it("preserves explicit default reasoning and service tier values on a rollback fork", async () => {
|
||||
const request = vi.fn().mockResolvedValue({ thread: threadRecord("forked") });
|
||||
const client = { request } as unknown as AppServerClient;
|
||||
const transport = createTestGateway({
|
||||
currentClient: () => client,
|
||||
connectedClient: vi.fn().mockResolvedValue(client),
|
||||
}).threadMutation;
|
||||
|
||||
await transport.forkThread("source", {
|
||||
position: { kind: "before-turn", turnId: "turn-3" },
|
||||
runtime: { reasoningEffort: null, serviceTier: null },
|
||||
});
|
||||
|
||||
expect(request).toHaveBeenCalledWith("thread/fork", {
|
||||
threadId: "source",
|
||||
cwd: "/vault",
|
||||
excludeTurns: true,
|
||||
beforeTurnId: "turn-3",
|
||||
serviceTier: null,
|
||||
config: { model_reasoning_effort: null },
|
||||
});
|
||||
});
|
||||
|
||||
it("reads thread history pages as thread stream items", async () => {
|
||||
|
|
|
|||
|
|
@ -9,20 +9,16 @@ import {
|
|||
type ThreadManagementActions,
|
||||
type ThreadManagementActionsHost,
|
||||
} from "../../../../../src/features/chat/application/threads/thread-management-actions";
|
||||
import type {
|
||||
ThreadMutationTransport,
|
||||
ThreadRollbackSnapshot,
|
||||
} from "../../../../../src/features/chat/application/threads/thread-mutation-transport";
|
||||
import type { ThreadMutationTransport } from "../../../../../src/features/chat/application/threads/thread-mutation-transport";
|
||||
import type { ThreadStreamItem } from "../../../../../src/features/chat/domain/thread-stream/items";
|
||||
import { deferred, waitForAsyncWork } from "../../../../support/async";
|
||||
import { chatStateFixture, chatStateWith } from "../../support/state";
|
||||
import { chatStateThreadStreamItems, withChatStateThreadStreamItems } from "../../support/thread-stream";
|
||||
import { withChatStateThreadStreamItems } from "../../support/thread-stream";
|
||||
|
||||
interface ThreadMutationTransportMock {
|
||||
ensureConnected: Mock<ThreadMutationTransport["ensureConnected"]>;
|
||||
compactThread: Mock<ThreadMutationTransport["compactThread"]>;
|
||||
forkThread: Mock<ThreadMutationTransport["forkThread"]>;
|
||||
rollbackThread: Mock<ThreadMutationTransport["rollbackThread"]>;
|
||||
}
|
||||
|
||||
interface ThreadOperationsMock {
|
||||
|
|
@ -33,7 +29,6 @@ interface ThreadOperationsMock {
|
|||
type ThreadManagementActionsHostMock = Omit<
|
||||
ThreadManagementActionsHost,
|
||||
| "addSystemMessage"
|
||||
| "notifyActiveThreadIdentityChanged"
|
||||
| "openThreadInCurrentPanel"
|
||||
| "openThreadInNewView"
|
||||
| "operations"
|
||||
|
|
@ -49,7 +44,6 @@ type ThreadManagementActionsHostMock = Omit<
|
|||
setComposerText: Mock<ThreadManagementActionsHost["setComposerText"]>;
|
||||
openThreadInNewView: Mock<ThreadManagementActionsHost["openThreadInNewView"]>;
|
||||
openThreadInCurrentPanel: Mock<ThreadManagementActionsHost["openThreadInCurrentPanel"]>;
|
||||
notifyActiveThreadIdentityChanged: Mock<ThreadManagementActionsHost["notifyActiveThreadIdentityChanged"]>;
|
||||
recordThread: Mock<ThreadManagementActionsHost["recordThread"]>;
|
||||
};
|
||||
|
||||
|
|
@ -125,7 +119,7 @@ describe("thread management actions", () => {
|
|||
{
|
||||
name: "rollback",
|
||||
invoke: (actions: ThreadManagementActions) => actions.rollbackThread("agent-thread"),
|
||||
transport: "rollbackThread" as const,
|
||||
transport: "forkThread" as const,
|
||||
message: "Agent threads cannot be rolled back.",
|
||||
},
|
||||
])("rejects $name mutations for subagent threads", async ({ invoke, transport, message }) => {
|
||||
|
|
@ -167,7 +161,7 @@ describe("thread management actions", () => {
|
|||
{
|
||||
name: "rollback",
|
||||
invoke: (actions: ThreadManagementActions) => actions.rollbackThread("source"),
|
||||
operation: "rollbackThread" as const,
|
||||
operation: "forkThread" as const,
|
||||
message: "Interrupt the current turn before rolling back.",
|
||||
},
|
||||
])("rejects $name while a turn is busy", async ({ invoke, operation, message }) => {
|
||||
|
|
@ -304,7 +298,9 @@ describe("thread management actions", () => {
|
|||
|
||||
await controller.forkThreadFromTurn("source", "turn-1", false);
|
||||
|
||||
expect(host.threadTransport.forkThread).toHaveBeenCalledWith("source", "turn-1");
|
||||
expect(host.threadTransport.forkThread).toHaveBeenCalledWith("source", {
|
||||
position: { kind: "through-turn", turnId: "turn-1" },
|
||||
});
|
||||
expect(host.recordThread).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
id: "forked",
|
||||
|
|
@ -315,19 +311,21 @@ describe("thread management actions", () => {
|
|||
expect(host.openThreadInCurrentPanel).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("archives the source before replacing the panel during fork and archive", async () => {
|
||||
it("replaces the panel before archiving the source during fork and archive", async () => {
|
||||
const host = hostMock({ items: turnItems() });
|
||||
const controller = threadManagementActions(host);
|
||||
|
||||
await controller.forkThreadFromTurn("source", "turn-3", true);
|
||||
|
||||
expect(host.threadTransport.forkThread).toHaveBeenCalledWith("source", "turn-3");
|
||||
expect(host.threadTransport.forkThread).toHaveBeenCalledWith("source", {
|
||||
position: { kind: "through-turn", turnId: "turn-3" },
|
||||
});
|
||||
expect(host.operations.archiveThread).toHaveBeenCalledWith("source", {});
|
||||
expect(host.openThreadInCurrentPanel).toHaveBeenCalledWith("forked");
|
||||
expect(callOrder(host.operations.archiveThread)).toBeLessThan(callOrder(host.openThreadInCurrentPanel));
|
||||
expect(host.openThreadInCurrentPanel).toHaveBeenCalledWith("forked", expect.any(Function));
|
||||
expect(callOrder(host.openThreadInCurrentPanel)).toBeLessThan(callOrder(host.operations.archiveThread));
|
||||
});
|
||||
|
||||
it("keeps the source panel when fork and archive fails to archive", async () => {
|
||||
it("keeps the replacement panel when source archiving fails", async () => {
|
||||
const host = hostMock({
|
||||
items: turnItems(),
|
||||
operations: {
|
||||
|
|
@ -339,11 +337,12 @@ describe("thread management actions", () => {
|
|||
await controller.forkThreadFromTurn("source", "turn-3", true);
|
||||
|
||||
expect(host.operations.archiveThread).toHaveBeenCalledWith("source", {});
|
||||
expect(host.openThreadInCurrentPanel).not.toHaveBeenCalled();
|
||||
expect(host.addSystemMessage).toHaveBeenCalledWith("archive failed");
|
||||
expect(host.openThreadInCurrentPanel).toHaveBeenCalledWith("forked", expect.any(Function));
|
||||
expect(activeThreadId(host.stateStore.getState())).toBe("forked");
|
||||
expect(host.addSystemMessage).toHaveBeenCalledWith("Forked the thread, but could not archive the previous version: archive failed");
|
||||
});
|
||||
|
||||
it("keeps the source panel when fork and archive is declined by thread operations", async () => {
|
||||
it("reports when source archiving is not completed after replacement", async () => {
|
||||
const host = hostMock({
|
||||
items: turnItems(),
|
||||
operations: {
|
||||
|
|
@ -355,19 +354,21 @@ describe("thread management actions", () => {
|
|||
await controller.forkThreadFromTurn("source", "turn-3", true);
|
||||
|
||||
expect(host.operations.archiveThread).toHaveBeenCalledWith("source", {});
|
||||
expect(host.openThreadInCurrentPanel).not.toHaveBeenCalled();
|
||||
expect(host.addSystemMessage).not.toHaveBeenCalledWith("archive failed");
|
||||
expect(activeThreadId(host.stateStore.getState())).toBe("forked");
|
||||
expect(host.addSystemMessage).toHaveBeenCalledWith(
|
||||
"Forked the thread, but could not archive the previous version: archive was not completed",
|
||||
);
|
||||
});
|
||||
|
||||
it("reports when fork and archive succeeds but the fork cannot replace the source panel", async () => {
|
||||
it("keeps the source when a fork cannot replace the current panel", async () => {
|
||||
const host = hostMock({ items: turnItems() });
|
||||
host.openThreadInCurrentPanel.mockRejectedValue(new Error("resume failed"));
|
||||
const controller = threadManagementActions(host);
|
||||
|
||||
await controller.forkThreadFromTurn("source", "turn-3", true);
|
||||
|
||||
expect(host.operations.archiveThread).toHaveBeenCalledWith("source", {});
|
||||
expect(host.addSystemMessage).toHaveBeenCalledWith("Archived thread source, but could not open forked thread forked: resume failed");
|
||||
expect(host.operations.archiveThread).not.toHaveBeenCalled();
|
||||
expect(host.addSystemMessage).toHaveBeenCalledWith("Forked thread forked, but could not open it in the current panel: resume failed");
|
||||
});
|
||||
|
||||
it("does not archive or replace the panel from stale fork responses", async () => {
|
||||
|
|
@ -455,7 +456,7 @@ describe("thread management actions", () => {
|
|||
expect(host.openThreadInCurrentPanel).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("does not replace a newer panel after fork source archiving completes", async () => {
|
||||
it("does not overwrite a newer panel while committed source cleanup completes", async () => {
|
||||
const archive = deferred<boolean>();
|
||||
const host = hostMock({
|
||||
items: turnItems(),
|
||||
|
|
@ -482,7 +483,7 @@ describe("thread management actions", () => {
|
|||
archive.resolve(true);
|
||||
await pendingFork;
|
||||
|
||||
expect(host.openThreadInCurrentPanel).not.toHaveBeenCalled();
|
||||
expect(host.openThreadInCurrentPanel).toHaveBeenCalledWith("forked", expect.any(Function));
|
||||
expect(activeThreadId(host.stateStore.getState())).toBe("other");
|
||||
});
|
||||
|
||||
|
|
@ -545,7 +546,7 @@ describe("thread management actions", () => {
|
|||
expect(host.operations.renameThread).toHaveBeenCalledWith("thread", " ");
|
||||
});
|
||||
|
||||
it("applies rollback response items before refreshing shared thread state", async () => {
|
||||
it("forks before the latest turn, adopts it, restores the prompt, and archives the source", async () => {
|
||||
const host = hostMock({ items: turnItems() });
|
||||
host.stateStore.dispatch({
|
||||
type: "active-thread/resumed",
|
||||
|
|
@ -571,13 +572,119 @@ describe("thread management actions", () => {
|
|||
|
||||
await controller.rollbackThread("source");
|
||||
|
||||
expect(host.threadTransport.rollbackThread).toHaveBeenCalledWith("source");
|
||||
expect(chatStateThreadStreamItems(host.stateStore.getState()).slice(0, 2)).toMatchObject([
|
||||
{ kind: "dialogue", role: "user", text: "kept prompt", turnId: "kept-turn" },
|
||||
{ kind: "dialogue", role: "assistant", text: "kept answer", turnId: "kept-turn" },
|
||||
]);
|
||||
expect(host.threadTransport.forkThread).toHaveBeenCalledWith("source", {
|
||||
position: { kind: "before-turn", turnId: "turn-3" },
|
||||
deferGoalContinuation: true,
|
||||
runtime: { reasoningEffort: null, serviceTier: null },
|
||||
});
|
||||
expect(host.openThreadInCurrentPanel).toHaveBeenCalledWith("forked", expect.any(Function));
|
||||
expect(host.setComposerText).toHaveBeenCalledWith("three");
|
||||
expect(host.operations.archiveThread).toHaveBeenCalledWith("source", { saveMarkdown: false });
|
||||
expect(callOrder(host.openThreadInCurrentPanel)).toBeLessThan(callOrder(host.operations.archiveThread));
|
||||
expect(callOrder(host.setComposerText)).toBeLessThan(callOrder(host.addSystemMessage));
|
||||
expect(host.recordThread).toHaveBeenCalledWith(rollbackSnapshot().thread);
|
||||
expect(host.recordThread).toHaveBeenCalledWith(panelThread("forked"));
|
||||
expect(activeThreadId(host.stateStore.getState())).toBe("forked");
|
||||
});
|
||||
|
||||
it("uses the first turn itself as the marker when rolling back a one-turn thread", async () => {
|
||||
const host = hostMock({ items: turnItems().slice(0, 2), activeThread: { id: "source" } });
|
||||
|
||||
await threadManagementActions(host).rollbackThread("source");
|
||||
|
||||
expect(host.threadTransport.forkThread).toHaveBeenCalledWith("source", {
|
||||
position: { kind: "before-turn", turnId: "turn-1" },
|
||||
deferGoalContinuation: true,
|
||||
runtime: { reasoningEffort: null },
|
||||
});
|
||||
expect(host.setComposerText).toHaveBeenCalledWith("one");
|
||||
expect(host.operations.archiveThread).toHaveBeenCalledWith("source", { saveMarkdown: false });
|
||||
});
|
||||
|
||||
it("preserves the active thread runtime when creating a rollback fork", async () => {
|
||||
const host = hostMock({ items: turnItems(), activeThread: { id: "source" } });
|
||||
host.stateStore.dispatch({
|
||||
type: "active-thread/resumed",
|
||||
approvalPolicyKnown: true,
|
||||
sandboxPolicyKnown: true,
|
||||
permissionProfileKnown: true,
|
||||
approvalPolicy: "never",
|
||||
sandboxPolicy: null,
|
||||
activePermissionProfile: { id: ":workspace", extends: null },
|
||||
thread: panelThread("source"),
|
||||
model: "gpt-5.6",
|
||||
reasoningEffort: "high",
|
||||
serviceTier: "priority",
|
||||
approvalsReviewer: "auto_review",
|
||||
});
|
||||
host.stateStore.dispatch({
|
||||
type: "thread-stream/items-replaced",
|
||||
items: turnItems(),
|
||||
historyCursor: null,
|
||||
loadingHistory: false,
|
||||
});
|
||||
|
||||
await threadManagementActions(host).rollbackThread("source");
|
||||
|
||||
expect(host.threadTransport.forkThread).toHaveBeenCalledWith("source", {
|
||||
position: { kind: "before-turn", turnId: "turn-3" },
|
||||
deferGoalContinuation: true,
|
||||
runtime: {
|
||||
model: "gpt-5.6",
|
||||
reasoningEffort: "high",
|
||||
serviceTier: "priority",
|
||||
approvalPolicy: "never",
|
||||
approvalsReviewer: "auto_review",
|
||||
permissions: ":workspace",
|
||||
},
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps the source when the rollback fork is not adopted", async () => {
|
||||
const host = hostMock({ items: turnItems(), activeThread: { id: "source" } });
|
||||
host.openThreadInCurrentPanel.mockResolvedValue({ adopted: false });
|
||||
|
||||
await threadManagementActions(host).rollbackThread("source");
|
||||
|
||||
expect(host.recordThread).toHaveBeenCalledWith(panelThread("forked"));
|
||||
expect(activeThreadId(host.stateStore.getState())).toBe("source");
|
||||
expect(host.setComposerText).not.toHaveBeenCalled();
|
||||
expect(host.operations.archiveThread).not.toHaveBeenCalled();
|
||||
expect(host.addSystemMessage).toHaveBeenCalledWith(
|
||||
"The rolled-back version was created but could not be opened in this panel. Open it from thread history.",
|
||||
);
|
||||
});
|
||||
|
||||
it("commits rollback cleanup from the adoption result", async () => {
|
||||
const host = hostMock({ items: turnItems(), activeThread: { id: "source" } });
|
||||
host.openThreadInCurrentPanel.mockImplementation(async (threadId, onAdopted) => {
|
||||
adoptThread(host, threadId);
|
||||
onAdopted();
|
||||
return { adopted: true };
|
||||
});
|
||||
|
||||
await threadManagementActions(host).rollbackThread("source");
|
||||
|
||||
expect(host.setComposerText).toHaveBeenCalledWith("three");
|
||||
expect(host.operations.archiveThread).toHaveBeenCalledWith("source", { saveMarkdown: false });
|
||||
expect(activeThreadId(host.stateStore.getState())).toBe("forked");
|
||||
});
|
||||
|
||||
it("keeps the adopted rollback when archiving its source fails", async () => {
|
||||
const host = hostMock({
|
||||
items: turnItems(),
|
||||
activeThread: { id: "source" },
|
||||
operations: {
|
||||
archiveThread: vi.fn<ThreadManagementActionsHost["operations"]["archiveThread"]>().mockRejectedValue(new Error("archive failed")),
|
||||
},
|
||||
});
|
||||
|
||||
await threadManagementActions(host).rollbackThread("source");
|
||||
|
||||
expect(activeThreadId(host.stateStore.getState())).toBe("forked");
|
||||
expect(host.setComposerText).toHaveBeenCalledWith("three");
|
||||
expect(host.addSystemMessage).toHaveBeenCalledWith(
|
||||
"Rolled back the latest turn, but could not archive the previous version: archive failed",
|
||||
);
|
||||
});
|
||||
|
||||
it("does not roll back an ephemeral side chat", async () => {
|
||||
|
|
@ -592,14 +699,14 @@ describe("thread management actions", () => {
|
|||
|
||||
await controller.rollbackThread("side-thread");
|
||||
|
||||
expect(host.threadTransport.rollbackThread).not.toHaveBeenCalled();
|
||||
expect(host.threadTransport.forkThread).not.toHaveBeenCalled();
|
||||
expect(host.addSystemMessage).toHaveBeenCalledWith("Side chats cannot be rolled back.");
|
||||
});
|
||||
|
||||
it("ignores stale rollback responses after the panel switches threads", async () => {
|
||||
const rollback = deferred<EffectOutcome<ThreadRollbackSnapshot>>();
|
||||
it("does not adopt or archive a rollback fork after the panel switches threads", async () => {
|
||||
const rollback = deferred<EffectOutcome<Thread>>();
|
||||
const host = hostMock({ items: turnItems() });
|
||||
host.threadTransport.rollbackThread.mockReturnValue(rollback.promise);
|
||||
host.threadTransport.forkThread.mockReturnValue(rollback.promise);
|
||||
host.stateStore.dispatch({
|
||||
type: "active-thread/resumed",
|
||||
approvalPolicyKnown: true,
|
||||
|
|
@ -624,7 +731,11 @@ describe("thread management actions", () => {
|
|||
|
||||
const pendingRollback = controller.rollbackThread("source");
|
||||
await waitForAsyncWork(() => {
|
||||
expect(host.threadTransport.rollbackThread).toHaveBeenCalledWith("source");
|
||||
expect(host.threadTransport.forkThread).toHaveBeenCalledWith("source", {
|
||||
position: { kind: "before-turn", turnId: "turn-3" },
|
||||
deferGoalContinuation: true,
|
||||
runtime: { reasoningEffort: null, serviceTier: null },
|
||||
});
|
||||
});
|
||||
host.stateStore.dispatch({
|
||||
type: "active-thread/resumed",
|
||||
|
|
@ -640,19 +751,19 @@ describe("thread management actions", () => {
|
|||
serviceTier: null,
|
||||
approvalsReviewer: null,
|
||||
});
|
||||
rollback.resolve(completedCurrent(rollbackSnapshot()));
|
||||
rollback.resolve(completedCurrent(panelThread("forked")));
|
||||
await pendingRollback;
|
||||
|
||||
expect(activeThreadId(host.stateStore.getState())).toBe("other");
|
||||
expect(host.setComposerText).not.toHaveBeenCalled();
|
||||
expect(host.notifyActiveThreadIdentityChanged).not.toHaveBeenCalled();
|
||||
expect(host.recordThread).not.toHaveBeenCalled();
|
||||
expect(host.recordThread).toHaveBeenCalledWith(panelThread("forked"));
|
||||
expect(host.operations.archiveThread).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("ignores rollback responses after a new turn starts in the same thread", async () => {
|
||||
const rollback = deferred<EffectOutcome<ThreadRollbackSnapshot>>();
|
||||
it("does not adopt or archive a rollback fork after a new turn starts in the source", async () => {
|
||||
const rollback = deferred<EffectOutcome<Thread>>();
|
||||
const host = hostMock({ items: turnItems() });
|
||||
host.threadTransport.rollbackThread.mockReturnValue(rollback.promise);
|
||||
host.threadTransport.forkThread.mockReturnValue(rollback.promise);
|
||||
host.stateStore.dispatch({
|
||||
type: "active-thread/resumed",
|
||||
approvalPolicyKnown: true,
|
||||
|
|
@ -671,24 +782,25 @@ describe("thread management actions", () => {
|
|||
const controller = threadManagementActions(host);
|
||||
|
||||
const pendingRollback = controller.rollbackThread("source");
|
||||
await waitForAsyncWork(() => expect(host.threadTransport.rollbackThread).toHaveBeenCalledOnce());
|
||||
await waitForAsyncWork(() => expect(host.threadTransport.forkThread).toHaveBeenCalledOnce());
|
||||
host.stateStore.dispatch({ type: "turn/started", threadId: "source", turnId: "new-turn" });
|
||||
rollback.resolve(completedCurrent(rollbackSnapshot()));
|
||||
rollback.resolve(completedCurrent(panelThread("forked")));
|
||||
await pendingRollback;
|
||||
|
||||
expect(host.stateStore.getState().turn.lifecycle).toEqual({ kind: "running", turnId: "new-turn" });
|
||||
expect(host.setComposerText).not.toHaveBeenCalled();
|
||||
expect(host.recordThread).not.toHaveBeenCalled();
|
||||
expect(host.recordThread).toHaveBeenCalledWith(panelThread("forked"));
|
||||
expect(host.operations.archiveThread).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("does not publish rollback facts from a stale app-server context", async () => {
|
||||
it("does not publish or adopt a rollback fork from a stale app-server context", async () => {
|
||||
const host = hostMock({
|
||||
items: turnItems(),
|
||||
activeThread: { id: "source" },
|
||||
threadTransport: {
|
||||
rollbackThread: vi
|
||||
.fn<ThreadMutationTransport["rollbackThread"]>()
|
||||
.mockResolvedValue({ kind: "completed-stale", value: rollbackSnapshot() }),
|
||||
forkThread: vi
|
||||
.fn<ThreadMutationTransport["forkThread"]>()
|
||||
.mockResolvedValue({ kind: "completed-stale", value: panelThread("forked") }),
|
||||
},
|
||||
});
|
||||
const controller = threadManagementActions(host);
|
||||
|
|
@ -697,14 +809,14 @@ describe("thread management actions", () => {
|
|||
|
||||
expect(host.recordThread).not.toHaveBeenCalled();
|
||||
expect(host.setComposerText).not.toHaveBeenCalled();
|
||||
expect(host.notifyActiveThreadIdentityChanged).not.toHaveBeenCalled();
|
||||
expect(host.operations.archiveThread).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("ignores rollback when the transport has no result", async () => {
|
||||
it("keeps the source when marker fork has no result", async () => {
|
||||
const host = hostMock({
|
||||
items: turnItems(),
|
||||
threadTransport: {
|
||||
rollbackThread: vi.fn<ThreadMutationTransport["rollbackThread"]>().mockResolvedValue({ kind: "not-started" }),
|
||||
forkThread: vi.fn<ThreadMutationTransport["forkThread"]>().mockResolvedValue({ kind: "not-started" }),
|
||||
},
|
||||
});
|
||||
host.stateStore.dispatch({
|
||||
|
|
@ -732,8 +844,8 @@ describe("thread management actions", () => {
|
|||
await controller.rollbackThread("source");
|
||||
|
||||
expect(host.setComposerText).not.toHaveBeenCalled();
|
||||
expect(host.notifyActiveThreadIdentityChanged).not.toHaveBeenCalled();
|
||||
expect(host.recordThread).not.toHaveBeenCalled();
|
||||
expect(host.operations.archiveThread).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -772,21 +884,6 @@ function turnItems(): ThreadStreamItem[] {
|
|||
];
|
||||
}
|
||||
|
||||
function rollbackItems(): ThreadStreamItem[] {
|
||||
return [
|
||||
{ id: "kept-user", kind: "dialogue", dialogueKind: "user", role: "user", text: "kept prompt", turnId: "kept-turn" },
|
||||
{
|
||||
id: "kept-agent",
|
||||
kind: "dialogue",
|
||||
role: "assistant",
|
||||
text: "kept answer",
|
||||
turnId: "kept-turn",
|
||||
dialogueKind: "assistantResponse",
|
||||
dialogueState: "completed",
|
||||
},
|
||||
];
|
||||
}
|
||||
|
||||
function threadManagementActions(host: ThreadManagementActionsHost): ThreadManagementActions {
|
||||
return createThreadManagementActions(host);
|
||||
}
|
||||
|
|
@ -809,7 +906,6 @@ function hostMock({
|
|||
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 = {
|
||||
|
|
@ -825,19 +921,33 @@ function hostMock({
|
|||
setStatus: vi.fn<ThreadManagementActionsHost["setStatus"]>(),
|
||||
setComposerText: vi.fn<ThreadManagementActionsHost["setComposerText"]>(),
|
||||
openThreadInNewView: vi.fn<ThreadManagementActionsHost["openThreadInNewView"]>().mockResolvedValue(undefined),
|
||||
openThreadInCurrentPanel: vi.fn<ThreadManagementActionsHost["openThreadInCurrentPanel"]>().mockResolvedValue(undefined),
|
||||
notifyActiveThreadIdentityChanged: vi.fn<ThreadManagementActionsHost["notifyActiveThreadIdentityChanged"]>(),
|
||||
openThreadInCurrentPanel: vi
|
||||
.fn<ThreadManagementActionsHost["openThreadInCurrentPanel"]>()
|
||||
.mockImplementation(async (threadId, onAdopted) => {
|
||||
adoptThread({ stateStore }, threadId);
|
||||
onAdopted();
|
||||
return { adopted: true };
|
||||
}),
|
||||
recordThread: vi.fn<ThreadManagementActionsHost["recordThread"]>(),
|
||||
threadHasPendingOrRunningPanel: vi.fn(() => false),
|
||||
};
|
||||
}
|
||||
|
||||
function rollbackSnapshot(overrides: Partial<ThreadRollbackSnapshot> = {}): ThreadRollbackSnapshot {
|
||||
return {
|
||||
thread: panelThread("source", { name: "Rolled Back Thread" }),
|
||||
items: rollbackItems(),
|
||||
...overrides,
|
||||
};
|
||||
function adoptThread(host: Pick<ThreadManagementActionsHost, "stateStore">, threadId: string): void {
|
||||
host.stateStore.dispatch({
|
||||
type: "active-thread/resumed",
|
||||
approvalPolicyKnown: true,
|
||||
sandboxPolicyKnown: true,
|
||||
permissionProfileKnown: true,
|
||||
approvalPolicy: null,
|
||||
sandboxPolicy: null,
|
||||
activePermissionProfile: null,
|
||||
thread: panelThread(threadId),
|
||||
model: null,
|
||||
reasoningEffort: null,
|
||||
serviceTier: null,
|
||||
approvalsReviewer: null,
|
||||
});
|
||||
}
|
||||
|
||||
function panelThread(id: string, overrides: Partial<Thread> = {}): Thread {
|
||||
|
|
|
|||
|
|
@ -187,7 +187,6 @@ function baseClientHandlers(): RequestHandlers {
|
|||
"thread/turns/list": vi.fn().mockResolvedValue({ data: [], nextCursor: null }),
|
||||
"turn/start": vi.fn().mockResolvedValue({ turn: { id: "turn-1" } }),
|
||||
"thread/fork": vi.fn().mockResolvedValue({ thread: threadFixture("thread-forked") }),
|
||||
"thread/rollback": vi.fn().mockResolvedValue({ thread: threadFixture("thread-forked") }),
|
||||
"thread/name/set": vi.fn().mockResolvedValue({}),
|
||||
"thread/goal/get": vi.fn().mockResolvedValue({ goal: null }),
|
||||
"thread/goal/set": vi.fn().mockResolvedValue({ goal: goalFixture("thread-1") }),
|
||||
|
|
|
|||
Loading…
Reference in a new issue