authorgravatar for maria@kuuro.netmaria <maria@kuuro.net> 2026-09-15 23:41:40-03:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-15 23:41:40-03:00
logeed974c122e192de45b54b7dac7ca1a51448c08f
treecd7761f652ceba57865bfe9d4b4ef1f4545fa1ed
parent0ec2b08a9a1db6fa35fab8d505b5bdc2d3709a8a
signature Signed by PGP key B5690EEEBB952194

fix(server): keep Claude rewind when fork history length changes (#11954)

Co-authored-by: Cursor <cursoragent@cursor.com>

2 files changed, 734 insertions(+), 25 deletions(-)

apps/server/src/provider/Layers/ClaudeAdapter.test.ts+634
......@@ -315,6 +315,57 @@ async function readPromptMessages(
315315const THREAD_ID = ThreadId.make("thread-claude-1");
316316const RESUME_THREAD_ID = ThreadId.make("thread-claude-resume");
317317const SYNTHETIC_SUBAGENT_MODEL = "claude-synthetic-subagent[expanded]";
318const CLAUDE_ORIGINAL_SESSION_ID = "550e8400-e29b-41d4-a716-446655440010";
319const CLAUDE_FORK_SESSION_ID = "550e8400-e29b-41d4-a716-446655440020";
320
321function claudeHistoryMessage(input: {
322 readonly type: "user" | "assistant" | "system";
323 readonly uuid: string;
324 readonly sessionId?: string;
325 readonly content?: unknown;
326 readonly parentToolUseId?: string | null;
327}) {
328 const content =
329 input.content ??
330 (input.type === "user" ? "prompt" : input.type === "assistant" ? [] : { subtype: "init" });
331 return {
332 type: input.type,
333 uuid: input.uuid,
334 session_id: input.sessionId ?? CLAUDE_ORIGINAL_SESSION_ID,
335 parent_tool_use_id: input.parentToolUseId ?? null,
336 parent_agent_id: null,
337 message: input.type === "system" ? content : { content },
338 };
339}
340
341const sendCompletedClaudeTurn = (
342 adapter: ClaudeAdapterShape,
343 harness: ReturnType<typeof makeHarness>,
344 threadId: ThreadId,
345 input: string,
346) =>
347 Effect.gen(function* () {
348 const turn = yield* adapter.sendTurn({
349 threadId,
350 input,
351 attachments: [],
352 });
353 const completedFiber = yield* Stream.filter(
354 adapter.streamEvents,
355 (event) => event.type === "turn.completed",
356 ).pipe(Stream.runHead, Effect.forkChild);
357 harness.queries.at(-1)!.emit({
358 type: "result",
359 subtype: "success",
360 is_error: false,
361 errors: [],
362 session_id: CLAUDE_ORIGINAL_SESSION_ID,
363 uuid: `result-${turn.turnId}`,
364 } as unknown as SDKMessage);
365 const completed = yield* Fiber.join(completedFiber);
366 assert.equal(completed._tag, "Some");
367 return turn;
368 });
318369
319370describe("ClaudeAdapterLive", () => {
320371 it.effect("returns validation error for non-claude provider on startSession", () => {
......@@ -6484,6 +6535,589 @@ describe("ClaudeAdapterLive", () => {
64846535 );
64856536 });
64866537
6538 it.effect("rewinds Claude history when the fork omits retained system messages", () => {
6539 const forkCalls: Array<Parameters<NonNullable<ClaudeAdapterLiveOptions["forkSession"]>>> = [];
6540 let firstTurnId = "";
6541 let secondTurnId = "";
6542 const harness = makeHarness({
6543 forkSession: async (...args) => {
6544 forkCalls.push(args);
6545 return { sessionId: CLAUDE_FORK_SESSION_ID };
6546 },
6547 getSessionMessages: async (sessionId) => {
6548 if (sessionId === CLAUDE_FORK_SESSION_ID) {
6549 return [
6550 claudeHistoryMessage({
6551 type: "user",
6552 uuid: `fork-${firstTurnId}`,
6553 sessionId,
6554 content: "first",
6555 }),
6556 claudeHistoryMessage({
6557 type: "assistant",
6558 uuid: "fork-assistant-1",
6559 sessionId,
6560 }),
6561 ];
6562 }
6563 return [
6564 claudeHistoryMessage({ type: "system", uuid: "system-init" }),
6565 claudeHistoryMessage({ type: "user", uuid: firstTurnId, content: "first" }),
6566 claudeHistoryMessage({ type: "assistant", uuid: "assistant-1" }),
6567 claudeHistoryMessage({
6568 type: "system",
6569 uuid: "compact-boundary",
6570 content: { subtype: "compact_boundary" },
6571 }),
6572 claudeHistoryMessage({ type: "user", uuid: secondTurnId, content: "second" }),
6573 claudeHistoryMessage({ type: "assistant", uuid: "assistant-2" }),
6574 ];
6575 },
6576 });
6577 return Effect.gen(function* () {
6578 const adapter = yield* ClaudeAdapter;
6579 const session = yield* adapter.startSession({
6580 threadId: THREAD_ID,
6581 provider: ProviderDriverKind.make("claudeAgent"),
6582 runtimeMode: "full-access",
6583 });
6584 firstTurnId = (yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "first"))
6585 .turnId;
6586 secondTurnId = (yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "second"))
6587 .turnId;
6588
6589 const snapshot = yield* adapter.rollbackThread(session.threadId, 1);
6590 assert.equal(snapshot.turns.length, 1);
6591 assert.deepEqual(forkCalls, [
6592 [CLAUDE_ORIGINAL_SESSION_ID, { upToMessageId: "compact-boundary" }],
6593 ]);
6594 assert.deepEqual((yield* adapter.listSessions())[0]?.resumeCursor, {
6595 threadId: session.threadId,
6596 resume: CLAUDE_FORK_SESSION_ID,
6597 turnCount: 1,
6598 turnStartMessageIds: [`fork-${firstTurnId}`],
6599 });
6600 }).pipe(
6601 Effect.provideService(Random.Random, makeDeterministicRandomService()),
6602 Effect.provide(harness.layer),
6603 );
6604 });
6605
6606 it.effect("rewinds Claude history when the fork inserts extra system messages", () => {
6607 let firstTurnId = "";
6608 let secondTurnId = "";
6609 const harness = makeHarness({
6610 forkSession: async () => ({ sessionId: CLAUDE_FORK_SESSION_ID }),
6611 getSessionMessages: async (sessionId) => {
6612 if (sessionId === CLAUDE_FORK_SESSION_ID) {
6613 return [
6614 claudeHistoryMessage({
6615 type: "system",
6616 uuid: "fork-system-init",
6617 sessionId,
6618 }),
6619 claudeHistoryMessage({
6620 type: "user",
6621 uuid: `fork-${firstTurnId}`,
6622 sessionId,
6623 content: "first",
6624 }),
6625 claudeHistoryMessage({
6626 type: "assistant",
6627 uuid: "fork-assistant-1",
6628 sessionId,
6629 }),
6630 ];
6631 }
6632 return [
6633 claudeHistoryMessage({ type: "user", uuid: firstTurnId, content: "first" }),
6634 claudeHistoryMessage({ type: "assistant", uuid: "assistant-1" }),
6635 claudeHistoryMessage({ type: "user", uuid: secondTurnId, content: "second" }),
6636 claudeHistoryMessage({ type: "assistant", uuid: "assistant-2" }),
6637 ];
6638 },
6639 });
6640 return Effect.gen(function* () {
6641 const adapter = yield* ClaudeAdapter;
6642 const session = yield* adapter.startSession({
6643 threadId: THREAD_ID,
6644 provider: ProviderDriverKind.make("claudeAgent"),
6645 runtimeMode: "full-access",
6646 });
6647 firstTurnId = (yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "first"))
6648 .turnId;
6649 secondTurnId = (yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "second"))
6650 .turnId;
6651
6652 yield* adapter.rollbackThread(session.threadId, 1);
6653 assert.deepEqual((yield* adapter.listSessions())[0]?.resumeCursor, {
6654 threadId: session.threadId,
6655 resume: CLAUDE_FORK_SESSION_ID,
6656 turnCount: 1,
6657 turnStartMessageIds: [`fork-${firstTurnId}`],
6658 });
6659 }).pipe(
6660 Effect.provideService(Random.Random, makeDeterministicRandomService()),
6661 Effect.provide(harness.layer),
6662 );
6663 });
6664
6665 it.effect("rewinds Claude history when the fork keeps a compact conversation prefix", () => {
6666 let firstTurnId = "";
6667 let secondTurnId = "";
6668 const harness = makeHarness({
6669 forkSession: async () => ({ sessionId: CLAUDE_FORK_SESSION_ID }),
6670 getSessionMessages: async (sessionId) => {
6671 if (sessionId === CLAUDE_FORK_SESSION_ID) {
6672 return [
6673 claudeHistoryMessage({
6674 type: "user",
6675 uuid: "fork-compacted-user",
6676 sessionId,
6677 content: "earlier compacted turn",
6678 }),
6679 claudeHistoryMessage({
6680 type: "assistant",
6681 uuid: "fork-compacted-assistant",
6682 sessionId,
6683 }),
6684 claudeHistoryMessage({
6685 type: "user",
6686 uuid: `fork-${firstTurnId}`,
6687 sessionId,
6688 content: "first",
6689 }),
6690 claudeHistoryMessage({
6691 type: "assistant",
6692 uuid: "fork-assistant-1",
6693 sessionId,
6694 }),
6695 ];
6696 }
6697 return [
6698 claudeHistoryMessage({ type: "user", uuid: firstTurnId, content: "first" }),
6699 claudeHistoryMessage({ type: "assistant", uuid: "assistant-1" }),
6700 claudeHistoryMessage({ type: "user", uuid: secondTurnId, content: "second" }),
6701 claudeHistoryMessage({ type: "assistant", uuid: "assistant-2" }),
6702 ];
6703 },
6704 });
6705 return Effect.gen(function* () {
6706 const adapter = yield* ClaudeAdapter;
6707 const session = yield* adapter.startSession({
6708 threadId: THREAD_ID,
6709 provider: ProviderDriverKind.make("claudeAgent"),
6710 runtimeMode: "full-access",
6711 });
6712 firstTurnId = (yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "first"))
6713 .turnId;
6714 secondTurnId = (yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "second"))
6715 .turnId;
6716
6717 yield* adapter.rollbackThread(session.threadId, 1);
6718 assert.deepEqual((yield* adapter.listSessions())[0]?.resumeCursor, {
6719 threadId: session.threadId,
6720 resume: CLAUDE_FORK_SESSION_ID,
6721 turnCount: 1,
6722 turnStartMessageIds: [`fork-${firstTurnId}`],
6723 });
6724 }).pipe(
6725 Effect.provideService(Random.Random, makeDeterministicRandomService()),
6726 Effect.provide(harness.layer),
6727 );
6728 });
6729
6730 it.effect("rewinds a Claude turn that includes tool results and a later steer", () => {
6731 const forkCalls: Array<Parameters<NonNullable<ClaudeAdapterLiveOptions["forkSession"]>>> = [];
6732 let firstTurnId = "";
6733 let secondTurnId = "";
6734 const harness = makeHarness({
6735 forkSession: async (...args) => {
6736 forkCalls.push(args);
6737 return { sessionId: CLAUDE_FORK_SESSION_ID };
6738 },
6739 getSessionMessages: async (sessionId) => {
6740 const history = [
6741 claudeHistoryMessage({ type: "user", uuid: firstTurnId, content: "first" }),
6742 claudeHistoryMessage({ type: "assistant", uuid: "assistant-1-tool" }),
6743 claudeHistoryMessage({
6744 type: "user",
6745 uuid: "tool-result-1",
6746 content: [{ type: "tool_result" }],
6747 }),
6748 claudeHistoryMessage({ type: "assistant", uuid: "assistant-1-final" }),
6749 claudeHistoryMessage({ type: "user", uuid: secondTurnId, content: "second" }),
6750 claudeHistoryMessage({ type: "assistant", uuid: "assistant-2" }),
6751 claudeHistoryMessage({
6752 type: "user",
6753 uuid: "steer",
6754 sessionId,
6755 content: "steer the second turn",
6756 }),
6757 claudeHistoryMessage({ type: "assistant", uuid: "assistant-steer", sessionId }),
6758 ];
6759 return sessionId === CLAUDE_FORK_SESSION_ID
6760 ? history.slice(0, 4).map((message) => ({
6761 ...message,
6762 uuid: `fork-${message.uuid}`,
6763 session_id: sessionId,
6764 }))
6765 : history;
6766 },
6767 });
6768 return Effect.gen(function* () {
6769 const adapter = yield* ClaudeAdapter;
6770 const session = yield* adapter.startSession({
6771 threadId: THREAD_ID,
6772 provider: ProviderDriverKind.make("claudeAgent"),
6773 runtimeMode: "full-access",
6774 });
6775 firstTurnId = (yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "first"))
6776 .turnId;
6777 const secondTurn = yield* adapter.sendTurn({
6778 threadId: session.threadId,
6779 input: "second",
6780 attachments: [],
6781 });
6782 secondTurnId = secondTurn.turnId;
6783 const steer = yield* adapter.sendTurn({
6784 threadId: session.threadId,
6785 input: "steer the second turn",
6786 attachments: [],
6787 });
6788 assert.equal(steer.turnId, secondTurn.turnId);
6789 const secondCompletedFiber = yield* Stream.filter(
6790 adapter.streamEvents,
6791 (event) => event.type === "turn.completed",
6792 ).pipe(Stream.runHead, Effect.forkChild);
6793 harness.queries.at(-1)!.emit({
6794 type: "result",
6795 subtype: "success",
6796 is_error: false,
6797 errors: [],
6798 session_id: CLAUDE_ORIGINAL_SESSION_ID,
6799 uuid: "result-second",
6800 } as unknown as SDKMessage);
6801 yield* Fiber.join(secondCompletedFiber);
6802
6803 const snapshot = yield* adapter.rollbackThread(session.threadId, 1);
6804 assert.equal(snapshot.turns.length, 1);
6805 assert.deepEqual(forkCalls, [
6806 [CLAUDE_ORIGINAL_SESSION_ID, { upToMessageId: "assistant-1-final" }],
6807 ]);
6808 assert.deepEqual((yield* adapter.listSessions())[0]?.resumeCursor, {
6809 threadId: session.threadId,
6810 resume: CLAUDE_FORK_SESSION_ID,
6811 turnCount: 1,
6812 turnStartMessageIds: [`fork-${firstTurnId}`],
6813 });
6814 }).pipe(
6815 Effect.provideService(Random.Random, makeDeterministicRandomService()),
6816 Effect.provide(harness.layer),
6817 );
6818 });
6819
6820 it.effect("rewinds the last two turns of a three-turn Claude thread", () => {
6821 let firstTurnId = "";
6822 let secondTurnId = "";
6823 let thirdTurnId = "";
6824 const harness = makeHarness({
6825 forkSession: async () => ({ sessionId: CLAUDE_FORK_SESSION_ID }),
6826 getSessionMessages: async (sessionId) => {
6827 const history = [
6828 claudeHistoryMessage({ type: "system", uuid: "system-init" }),
6829 claudeHistoryMessage({ type: "user", uuid: firstTurnId, content: "first" }),
6830 claudeHistoryMessage({ type: "assistant", uuid: "assistant-1" }),
6831 claudeHistoryMessage({ type: "user", uuid: secondTurnId, content: "second" }),
6832 claudeHistoryMessage({ type: "assistant", uuid: "assistant-2" }),
6833 claudeHistoryMessage({ type: "user", uuid: thirdTurnId, content: "third" }),
6834 claudeHistoryMessage({ type: "assistant", uuid: "assistant-3" }),
6835 ];
6836 return sessionId === CLAUDE_FORK_SESSION_ID
6837 ? [
6838 claudeHistoryMessage({
6839 type: "user",
6840 uuid: `fork-${firstTurnId}`,
6841 sessionId,
6842 content: "first",
6843 }),
6844 claudeHistoryMessage({
6845 type: "assistant",
6846 uuid: "fork-assistant-1",
6847 sessionId,
6848 }),
6849 ]
6850 : history;
6851 },
6852 });
6853 return Effect.gen(function* () {
6854 const adapter = yield* ClaudeAdapter;
6855 const session = yield* adapter.startSession({
6856 threadId: THREAD_ID,
6857 provider: ProviderDriverKind.make("claudeAgent"),
6858 runtimeMode: "full-access",
6859 });
6860 firstTurnId = (yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "first"))
6861 .turnId;
6862 secondTurnId = (yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "second"))
6863 .turnId;
6864 thirdTurnId = (yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "third"))
6865 .turnId;
6866
6867 const snapshot = yield* adapter.rollbackThread(session.threadId, 2);
6868 assert.equal(snapshot.turns.length, 1);
6869 assert.deepEqual((yield* adapter.listSessions())[0]?.resumeCursor, {
6870 threadId: session.threadId,
6871 resume: CLAUDE_FORK_SESSION_ID,
6872 turnCount: 1,
6873 turnStartMessageIds: [`fork-${firstTurnId}`],
6874 });
6875 }).pipe(
6876 Effect.provideService(Random.Random, makeDeterministicRandomService()),
6877 Effect.provide(harness.layer),
6878 );
6879 });
6880
6881 it.effect("rewinds only the latest turn of a three-turn Claude thread", () => {
6882 let firstTurnId = "";
6883 let secondTurnId = "";
6884 let thirdTurnId = "";
6885 const harness = makeHarness({
6886 forkSession: async () => ({ sessionId: CLAUDE_FORK_SESSION_ID }),
6887 getSessionMessages: async (sessionId) => {
6888 const history = [
6889 claudeHistoryMessage({ type: "system", uuid: "system-init" }),
6890 claudeHistoryMessage({ type: "user", uuid: firstTurnId, content: "first" }),
6891 claudeHistoryMessage({ type: "assistant", uuid: "assistant-1" }),
6892 claudeHistoryMessage({ type: "user", uuid: secondTurnId, content: "second" }),
6893 claudeHistoryMessage({ type: "assistant", uuid: "assistant-2" }),
6894 claudeHistoryMessage({ type: "user", uuid: thirdTurnId, content: "third" }),
6895 claudeHistoryMessage({ type: "assistant", uuid: "assistant-3" }),
6896 ];
6897 return sessionId === CLAUDE_FORK_SESSION_ID
6898 ? [
6899 claudeHistoryMessage({
6900 type: "user",
6901 uuid: `fork-${firstTurnId}`,
6902 sessionId,
6903 content: "first",
6904 }),
6905 claudeHistoryMessage({
6906 type: "assistant",
6907 uuid: "fork-assistant-1",
6908 sessionId,
6909 }),
6910 claudeHistoryMessage({
6911 type: "system",
6912 uuid: "fork-notice",
6913 sessionId,
6914 }),
6915 claudeHistoryMessage({
6916 type: "user",
6917 uuid: `fork-${secondTurnId}`,
6918 sessionId,
6919 content: "second",
6920 }),
6921 claudeHistoryMessage({
6922 type: "assistant",
6923 uuid: "fork-assistant-2",
6924 sessionId,
6925 }),
6926 ]
6927 : history;
6928 },
6929 });
6930 return Effect.gen(function* () {
6931 const adapter = yield* ClaudeAdapter;
6932 const session = yield* adapter.startSession({
6933 threadId: THREAD_ID,
6934 provider: ProviderDriverKind.make("claudeAgent"),
6935 runtimeMode: "full-access",
6936 });
6937 firstTurnId = (yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "first"))
6938 .turnId;
6939 secondTurnId = (yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "second"))
6940 .turnId;
6941 thirdTurnId = (yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "third"))
6942 .turnId;
6943
6944 const snapshot = yield* adapter.rollbackThread(session.threadId, 1);
6945 assert.equal(snapshot.turns.length, 2);
6946 assert.deepEqual((yield* adapter.listSessions())[0]?.resumeCursor, {
6947 threadId: session.threadId,
6948 resume: CLAUDE_FORK_SESSION_ID,
6949 turnCount: 2,
6950 turnStartMessageIds: [`fork-${firstTurnId}`, `fork-${secondTurnId}`],
6951 });
6952 }).pipe(
6953 Effect.provideService(Random.Random, makeDeterministicRandomService()),
6954 Effect.provide(harness.layer),
6955 );
6956 });
6957
6958 it.effect("resets a Claude thread when rewind removes every recorded turn", () => {
6959 const forkCalls: Array<Parameters<NonNullable<ClaudeAdapterLiveOptions["forkSession"]>>> = [];
6960 const harness = makeHarness({
6961 forkSession: async (...args) => {
6962 forkCalls.push(args);
6963 return { sessionId: CLAUDE_FORK_SESSION_ID };
6964 },
6965 getSessionMessages: async () => [
6966 claudeHistoryMessage({ type: "user", uuid: "unused", content: "first" }),
6967 ],
6968 });
6969 return Effect.gen(function* () {
6970 const adapter = yield* ClaudeAdapter;
6971 const session = yield* adapter.startSession({
6972 threadId: THREAD_ID,
6973 provider: ProviderDriverKind.make("claudeAgent"),
6974 runtimeMode: "full-access",
6975 });
6976 yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "first");
6977 yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "second");
6978
6979 const snapshot = yield* adapter.rollbackThread(session.threadId, 2);
6980 assert.equal(snapshot.turns.length, 0);
6981 assert.equal(forkCalls.length, 0);
6982 const resetOptions = harness.getLastCreateQueryInput()?.options;
6983 assert.equal(resetOptions?.resume, undefined);
6984 assert.ok(resetOptions?.sessionId);
6985 }).pipe(
6986 Effect.provideService(Random.Random, makeDeterministicRandomService()),
6987 Effect.provide(harness.layer),
6988 );
6989 });
6990
6991 it.effect("rejects a Claude fork that drops a retained user turn", () => {
6992 let firstTurnId = "";
6993 let secondTurnId = "";
6994 const harness = makeHarness({
6995 forkSession: async () => ({ sessionId: CLAUDE_FORK_SESSION_ID }),
6996 getSessionMessages: async (sessionId) => {
6997 if (sessionId === CLAUDE_FORK_SESSION_ID) {
6998 return [
6999 claudeHistoryMessage({
7000 type: "assistant",
7001 uuid: "fork-assistant-1",
7002 sessionId,
7003 }),
7004 ];
7005 }
7006 return [
7007 claudeHistoryMessage({ type: "user", uuid: firstTurnId, content: "first" }),
7008 claudeHistoryMessage({ type: "assistant", uuid: "assistant-1" }),
7009 claudeHistoryMessage({ type: "user", uuid: secondTurnId, content: "second" }),
7010 claudeHistoryMessage({ type: "assistant", uuid: "assistant-2" }),
7011 ];
7012 },
7013 });
7014 return Effect.gen(function* () {
7015 const adapter = yield* ClaudeAdapter;
7016 const session = yield* adapter.startSession({
7017 threadId: THREAD_ID,
7018 provider: ProviderDriverKind.make("claudeAgent"),
7019 runtimeMode: "full-access",
7020 });
7021 firstTurnId = (yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "first"))
7022 .turnId;
7023 secondTurnId = (yield* sendCompletedClaudeTurn(adapter, harness, session.threadId, "second"))
7024 .turnId;
7025
7026 const error = yield* adapter.rollbackThread(session.threadId, 1).pipe(Effect.flip);
7027 assert.match(error.message, /did not preserve the retained turn boundaries/);
7028 assert.equal(harness.queries.at(-1)?.closeCalls, 0);
7029 }).pipe(
7030 Effect.provideService(Random.Random, makeDeterministicRandomService()),
7031 Effect.provide(harness.layer),
7032 );
7033 });
7034
7035 for (const scenario of [
7036 "restores conversation messages between retained turns",
7037 "changes retained assistant content without changing its role",
7038 ] as const) {
7039 it.effect(`rejects a Claude fork that ${scenario}`, () => {
7040 const turnIds: Array<string> = [];
7041 const harness = makeHarness({
7042 forkSession: async () => ({ sessionId: CLAUDE_FORK_SESSION_ID }),
7043 getSessionMessages: async (sessionId) => {
7044 const history = turnIds.flatMap((turnId, index) => [
7045 claudeHistoryMessage({
7046 type: "user",
7047 uuid: turnId,
7048 content: `prompt ${index + 1}`,
7049 }),
7050 claudeHistoryMessage({
7051 type: "assistant",
7052 uuid: `assistant-${index + 1}`,
7053 content: [{ type: "text", text: `reply ${index + 1}` }],
7054 }),
7055 ]);
7056 if (sessionId !== CLAUDE_FORK_SESSION_ID) return history;
7057 const forkHistory = history.slice(0, 4).map((message) => ({
7058 ...message,
7059 uuid: `fork-${message.uuid}`,
7060 session_id: sessionId,
7061 }));
7062 if (scenario === "restores conversation messages between retained turns") {
7063 forkHistory.splice(
7064 2,
7065 0,
7066 claudeHistoryMessage({
7067 type: "user",
7068 uuid: "fork-restored-steer",
7069 sessionId,
7070 content: "earlier steer omitted by compaction",
7071 }),
7072 claudeHistoryMessage({
7073 type: "assistant",
7074 uuid: "fork-restored-reply",
7075 sessionId,
7076 content: [{ type: "text", text: "earlier steering reply" }],
7077 }),
7078 );
7079 } else {
7080 forkHistory[1] = claudeHistoryMessage({
7081 type: "assistant",
7082 uuid: "fork-assistant-1",
7083 sessionId,
7084 content: [{ type: "text", text: "different retained reply" }],
7085 });
7086 }
7087 return forkHistory;
7088 },
7089 });
7090 return Effect.gen(function* () {
7091 const adapter = yield* ClaudeAdapter;
7092 const session = yield* adapter.startSession({
7093 threadId: THREAD_ID,
7094 provider: ProviderDriverKind.make("claudeAgent"),
7095 runtimeMode: "full-access",
7096 });
7097 for (let index = 0; index < 3; index++) {
7098 const turn = yield* sendCompletedClaudeTurn(
7099 adapter,
7100 harness,
7101 session.threadId,
7102 `prompt ${index + 1}`,
7103 );
7104 turnIds.push(turn.turnId);
7105 }
7106 const cursorBeforeRollback = (yield* adapter.listSessions())[0]?.resumeCursor;
7107 const queryCountBeforeRollback = harness.queries.length;
7108
7109 const error = yield* adapter.rollbackThread(session.threadId, 1).pipe(Effect.flip);
7110 assert.match(error.message, /did not preserve the retained turn boundaries/);
7111 assert.equal(harness.queries.length, queryCountBeforeRollback);
7112 assert.equal(harness.queries.at(-1)?.closeCalls, 0);
7113 assert.deepEqual((yield* adapter.listSessions())[0]?.resumeCursor, cursorBeforeRollback);
7114 }).pipe(
7115 Effect.provideService(Random.Random, makeDeterministicRandomService()),
7116 Effect.provide(harness.layer),
7117 );
7118 });
7119 }
7120
64877121 it.effect("updates model on sendTurn when model override is provided", () => {
64887122 const harness = makeHarness();
64897123 return Effect.gen(function* () {
apps/server/src/provider/Layers/ClaudeAdapter.ts+100-25
......@@ -8,6 +8,7 @@
88 * @module ClaudeAdapterLive
99 */
1010
11import * as NodeUtil from "node:util";
1112import {
1213 type CanUseTool,
1314 query,
......@@ -135,6 +136,94 @@ const decodeSessionMessages = Schema.decodeSync(
135136 ),
136137);
137138
139type ClaudeHistoryMessage = {
140 readonly type: string;
141 readonly uuid: string;
142 readonly parent_tool_use_id: string | null;
143 readonly message: unknown;
144};
145
146const isClaudeConversationMessage = (message: ClaudeHistoryMessage): boolean =>
147 message.type === "user" || message.type === "assistant";
148
149const isClaudeHumanTurnStart = (message: ClaudeHistoryMessage): boolean => {
150 if (message.type !== "user" || message.parent_tool_use_id !== null) return false;
151 const body = message.message;
152 if (typeof body !== "object" || body === null || !("content" in body)) return false;
153 const content = body.content;
154 return (
155 typeof content === "string" ||
156 (Array.isArray(content) &&
157 content.some(
158 (part: unknown) =>
159 typeof part === "object" &&
160 part !== null &&
161 "type" in part &&
162 part.type !== "tool_result",
163 ))
164 );
165};
166
167const conversationIndexForUuid = (
168 messages: ReadonlyArray<ClaudeHistoryMessage>,
169 uuid: string,
170): number => {
171 let index = -1;
172 for (const message of messages) {
173 if (!isClaudeConversationMessage(message)) continue;
174 index += 1;
175 if (message.uuid === uuid) return index;
176 }
177 return -1;
178};
179
180// Native forks rewrite every UUID. getSessionMessages then rebuilds the
181// parentUuid chain, so system notices and compact metadata can change the
182// raw length without dropping retained user/assistant turns. Align those
183// conversation messages from the truncated end, then remap T3 turn starts.
184const remapClaudeForkTurnBoundaries = (
185 messages: ReadonlyArray<ClaudeHistoryMessage>,
186 forkMessages: ReadonlyArray<ClaudeHistoryMessage>,
187 firstRemoved: number,
188 retainedBoundaries: ReadonlyArray<string | null>,
189): Array<string | null> | undefined => {
190 const retainedConversation = messages.slice(0, firstRemoved).filter(isClaudeConversationMessage);
191 const forkConversation = forkMessages.filter(isClaudeConversationMessage);
192 if (retainedConversation.length === 0) {
193 return retainedBoundaries.every((id) => id === null) ? [...retainedBoundaries] : undefined;
194 }
195 const offset = forkConversation.length - retainedConversation.length;
196 // Forks preserve message bodies. Matching roles alone can mistake a restored
197 // steering message for a retained turn when compaction changes the chain.
198 if (
199 offset < 0 ||
200 retainedConversation.some((message, index) => {
201 const forkMessage = forkConversation[index + offset];
202 return (
203 forkMessage === undefined ||
204 forkMessage.type !== message.type ||
205 !NodeUtil.isDeepStrictEqual(forkMessage.message, message.message)
206 );
207 })
208 ) {
209 return undefined;
210 }
211 const remapped = retainedBoundaries.map((originalId) => {
212 if (originalId === null) return null;
213 const originalIndex = conversationIndexForUuid(messages, originalId);
214 const forkIndex = originalIndex + offset;
215 const forkMessage =
216 originalIndex >= 0 && forkIndex >= 0 ? forkConversation[forkIndex] : undefined;
217 const originalMessage = messages.find((message) => message.uuid === originalId);
218 return forkMessage !== undefined &&
219 originalMessage !== undefined &&
220 forkMessage.type === originalMessage.type
221 ? forkMessage.uuid
222 : null;
223 });
224 return remapped.some((id) => id === null) ? undefined : remapped;
225};
226
138227const PROVIDER = ProviderDriverKind.make("claudeAgent");
139228type ClaudeTextStreamKind = Extract<RuntimeContentStreamKind, "assistant_text" | "reasoning_text">;
140229type ClaudeToolResultStreamKind = Extract<
......@@ -5190,23 +5279,9 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
51905279 });
51915280 const messages = yield* readHistory(sessionId);
51925281 // Tool results are user-role messages too. Only human prompts begin a turn.
5193 const turnStarts = messages.flatMap((message, index) => {
5194 if (message.type !== "user" || message.parent_tool_use_id !== null) return [];
5195 const body = message.message;
5196 if (typeof body !== "object" || body === null || !("content" in body)) return [];
5197 const content = body.content;
5198 return typeof content === "string" ||
5199 (Array.isArray(content) &&
5200 content.some(
5201 (part: unknown) =>
5202 typeof part === "object" &&
5203 part !== null &&
5204 "type" in part &&
5205 part.type !== "tool_result",
5206 ))
5207 ? [index]
5208 : [];
5209 });
5282 const turnStarts = messages.flatMap((message, index) =>
5283 isClaudeHumanTurnStart(message) ? [index] : [],
5284 );
52105285 if (messages.length === 0) {
52115286 return yield* new ProviderAdapterRequestError({
52125287 provider: PROVIDER,
......@@ -5263,20 +5338,20 @@ export const makeClaudeAdapter = Effect.fn("makeClaudeAdapter")(function* (
52635338 const retainedBoundaries = boundaries.slice(0, retainedCount);
52645339 if (fork) {
52655340 const forkMessages = yield* readHistory(fork.sessionId);
5266 if (forkMessages.length !== firstRemoved) {
5341 const remappedBoundaries = remapClaudeForkTurnBoundaries(
5342 messages,
5343 forkMessages,
5344 firstRemoved,
5345 retainedBoundaries,
5346 );
5347 if (!remappedBoundaries) {
52675348 return yield* new ProviderAdapterRequestError({
52685349 provider: PROVIDER,
52695350 method: "thread/rollback",
52705351 detail: "Claude fork history did not preserve the retained turn boundaries.",
52715352 });
52725353 }
5273 // Native forks replace every UUID while preserving transcript order.
5274 for (let index = 0; index < retainedBoundaries.length; index++) {
5275 const messageIndex = messages.findIndex(
5276 (message) => message.uuid === retainedBoundaries[index],
5277 );
5278 retainedBoundaries[index] = forkMessages[messageIndex]?.uuid ?? null;
5279 }
5354 retainedBoundaries.splice(0, retainedBoundaries.length, ...remappedBoundaries);
52805355 }
52815356 yield* stopSessionInternal(context, { emitExitEvent: false });
52825357 yield* startSession({