Skip to content

Commit 601bbf7

Browse files
author
Eric Wheeler
committed
refactor: make resumeTaskFromHistory message updates atomic
The resumeTaskFromHistory method was refactored to ensure that updates to both the cline message history and the API conversation history are fully atomic. Previously, the method would read the histories, modify them in-memory, and then call the respective modify functions with the already-modified data. This approach did not guarantee atomicity. The new implementation moves the modification logic directly inside the callbacks for `modifyClineMessages` and `modifyApiConversationHistory`. This ensures that the entire read-modify-write cycle for each history is performed as a single, uninterruptible transaction, preventing potential race conditions or partial state saves. This change also involved: - Adjusting variable scopes to support the new callback structure. - Removing the now-unused `getSavedClineMessages` helper method as part of the refactor. Signed-off-by: Eric Wheeler <[email protected]>
1 parent a4c663a commit 601bbf7

File tree

1 file changed

+135
-140
lines changed

1 file changed

+135
-140
lines changed

‎src/core/task/Task.ts‎

Lines changed: 135 additions & 140 deletions
Original file line numberDiff line numberDiff line change
@@ -360,11 +360,6 @@ export class Task extends EventEmitter<ClineEvents> {
360360
}
361361

362362
// Cline Messages
363-
364-
private async getSavedClineMessages(): Promise<ClineMessage[]> {
365-
return readTaskMessages({ taskId: this.taskId, globalStoragePath: this.globalStoragePath })
366-
}
367-
368363
private async addToClineMessages(message: ClineMessage) {
369364
await this.modifyClineMessages(async (messages) => {
370365
messages.push(message)
@@ -900,33 +895,31 @@ export class Task extends EventEmitter<ClineEvents> {
900895
}
901896

902897
private async resumeTaskFromHistory() {
903-
const modifiedClineMessages = await this.getSavedClineMessages()
904-
905-
// Remove any resume messages that may have been added before
906-
const lastRelevantMessageIndex = findLastIndex(
907-
modifiedClineMessages,
908-
(m) => !(m.ask === "resume_task" || m.ask === "resume_completed_task"),
909-
)
898+
await this.modifyClineMessages(async (modifiedClineMessages) => {
899+
// Remove any resume messages that may have been added before
900+
const lastRelevantMessageIndex = findLastIndex(
901+
modifiedClineMessages,
902+
(m) => !(m.ask === "resume_task" || m.ask === "resume_completed_task"),
903+
)
910904

911-
if (lastRelevantMessageIndex !== -1) {
912-
modifiedClineMessages.splice(lastRelevantMessageIndex + 1)
913-
}
905+
if (lastRelevantMessageIndex !== -1) {
906+
modifiedClineMessages.splice(lastRelevantMessageIndex + 1)
907+
}
914908

915-
// since we don't use api_req_finished anymore, we need to check if the last api_req_started has a cost value, if it doesn't and no cancellation reason to present, then we remove it since it indicates an api request without any partial content streamed
916-
const lastApiReqStartedIndex = findLastIndex(
917-
modifiedClineMessages,
918-
(m) => m.type === "say" && m.say === "api_req_started",
919-
)
909+
// since we don't use api_req_finished anymore, we need to check if the last api_req_started has a cost value, if it doesn't and no cancellation reason to present, then we remove it since it indicates an api request without any partial content streamed
910+
const lastApiReqStartedIndex = findLastIndex(
911+
modifiedClineMessages,
912+
(m) => m.type === "say" && m.say === "api_req_started",
913+
)
920914

921-
if (lastApiReqStartedIndex !== -1) {
922-
const lastApiReqStarted = modifiedClineMessages[lastApiReqStartedIndex]
923-
const { cost, cancelReason }: ClineApiReqInfo = JSON.parse(lastApiReqStarted.text || "{}")
924-
if (cost === undefined && cancelReason === undefined) {
925-
modifiedClineMessages.splice(lastApiReqStartedIndex, 1)
915+
if (lastApiReqStartedIndex !== -1) {
916+
const lastApiReqStarted = modifiedClineMessages[lastApiReqStartedIndex]
917+
const { cost, cancelReason }: ClineApiReqInfo = JSON.parse(lastApiReqStarted.text || "{}")
918+
if (cost === undefined && cancelReason === undefined) {
919+
modifiedClineMessages.splice(lastApiReqStartedIndex, 1)
920+
}
926921
}
927-
}
928922

929-
await this.modifyClineMessages(async () => {
930923
return modifiedClineMessages
931924
})
932925

@@ -963,125 +956,131 @@ export class Task extends EventEmitter<ClineEvents> {
963956

964957
// Make sure that the api conversation history can be resumed by the API,
965958
// even if it goes out of sync with cline messages.
966-
let existingApiConversationHistory: ApiMessage[] = await this.getSavedApiConversationHistory()
967-
968-
// v2.0 xml tags refactor caveat: since we don't use tools anymore, we need to replace all tool use blocks with a text block since the API disallows conversations with tool uses and no tool schema
969-
const conversationWithoutToolBlocks = existingApiConversationHistory.map((message) => {
970-
if (Array.isArray(message.content)) {
971-
const newContent = message.content.map((block) => {
972-
if (block.type === "tool_use") {
973-
// It's important we convert to the new tool schema
974-
// format so the model doesn't get confused about how to
975-
// invoke tools.
976-
const inputAsXml = Object.entries(block.input as Record<string, string>)
977-
.map(([key, value]) => `<${key}>\n${value}\n</${key}>`)
978-
.join("\n")
979-
return {
980-
type: "text",
981-
text: `<${block.name}>\n${inputAsXml}\n</${block.name}>`,
982-
} as Anthropic.Messages.TextBlockParam
983-
} else if (block.type === "tool_result") {
984-
// Convert block.content to text block array, removing images
985-
const contentAsTextBlocks = Array.isArray(block.content)
986-
? block.content.filter((item) => item.type === "text")
987-
: [{ type: "text", text: block.content }]
988-
const textContent = contentAsTextBlocks.map((item) => item.text).join("\n\n")
989-
const toolName = findToolName(block.tool_use_id, existingApiConversationHistory)
990-
return {
991-
type: "text",
992-
text: `[${toolName} Result]\n\n${textContent}`,
993-
} as Anthropic.Messages.TextBlockParam
994-
}
995-
return block
996-
})
997-
return { ...message, content: newContent }
998-
}
999-
return message
1000-
})
1001-
existingApiConversationHistory = conversationWithoutToolBlocks
1002-
1003-
// FIXME: remove tool use blocks altogether
1004-
1005-
// if the last message is an assistant message, we need to check if there's tool use since every tool use has to have a tool response
1006-
// if there's no tool use and only a text block, then we can just add a user message
1007-
// (note this isn't relevant anymore since we use custom tool prompts instead of tool use blocks, but this is here for legacy purposes in case users resume old tasks)
1008-
1009-
// if the last message is a user message, we can need to get the assistant message before it to see if it made tool calls, and if so, fill in the remaining tool responses with 'interrupted'
1010-
1011-
let modifiedOldUserContent: Anthropic.Messages.ContentBlockParam[] // either the last message if its user message, or the user message before the last (assistant) message
1012-
let modifiedApiConversationHistory: ApiMessage[] // need to remove the last user message to replace with new modified user message
1013-
if (existingApiConversationHistory.length > 0) {
1014-
const lastMessage = existingApiConversationHistory[existingApiConversationHistory.length - 1]
1015-
1016-
if (lastMessage.role === "assistant") {
1017-
const content = Array.isArray(lastMessage.content)
1018-
? lastMessage.content
1019-
: [{ type: "text", text: lastMessage.content }]
1020-
const hasToolUse = content.some((block) => block.type === "tool_use")
1021-
1022-
if (hasToolUse) {
1023-
const toolUseBlocks = content.filter(
1024-
(block) => block.type === "tool_use",
1025-
) as Anthropic.Messages.ToolUseBlock[]
1026-
const toolResponses: Anthropic.ToolResultBlockParam[] = toolUseBlocks.map((block) => ({
1027-
type: "tool_result",
1028-
tool_use_id: block.id,
1029-
content: "Task was interrupted before this tool call could be completed.",
1030-
}))
1031-
modifiedApiConversationHistory = [...existingApiConversationHistory] // no changes
1032-
modifiedOldUserContent = [...toolResponses]
1033-
} else {
1034-
modifiedApiConversationHistory = [...existingApiConversationHistory]
1035-
modifiedOldUserContent = []
959+
let modifiedOldUserContent: Anthropic.Messages.ContentBlockParam[] | undefined
960+
await this.modifyApiConversationHistory(async (existingApiConversationHistory) => {
961+
const conversationWithoutToolBlocks = existingApiConversationHistory.map((message) => {
962+
if (Array.isArray(message.content)) {
963+
const newContent = message.content.map((block) => {
964+
if (block.type === "tool_use") {
965+
// It's important we convert to the new tool schema
966+
// format so the model doesn't get confused about how to
967+
// invoke tools.
968+
const inputAsXml = Object.entries(block.input as Record<string, string>)
969+
.map(([key, value]) => `<${key}>\n${value}\n</${key}>`)
970+
.join("\n")
971+
return {
972+
type: "text",
973+
text: `<${block.name}>\n${inputAsXml}\n</${block.name}>`,
974+
} as Anthropic.Messages.TextBlockParam
975+
} else if (block.type === "tool_result") {
976+
// Convert block.content to text block array, removing images
977+
const contentAsTextBlocks = Array.isArray(block.content)
978+
? block.content.filter((item) => item.type === "text")
979+
: [{ type: "text", text: block.content }]
980+
const textContent = contentAsTextBlocks.map((item) => item.text).join("\n\n")
981+
const toolName = findToolName(block.tool_use_id, existingApiConversationHistory)
982+
return {
983+
type: "text",
984+
text: `[${toolName} Result]\n\n${textContent}`,
985+
} as Anthropic.Messages.TextBlockParam
986+
}
987+
return block
988+
})
989+
return { ...message, content: newContent }
1036990
}
1037-
} else if (lastMessage.role === "user") {
1038-
const previousAssistantMessage: ApiMessage | undefined =
1039-
existingApiConversationHistory[existingApiConversationHistory.length - 2]
1040-
1041-
const existingUserContent: Anthropic.Messages.ContentBlockParam[] = Array.isArray(lastMessage.content)
1042-
? lastMessage.content
1043-
: [{ type: "text", text: lastMessage.content }]
1044-
if (previousAssistantMessage && previousAssistantMessage.role === "assistant") {
1045-
const assistantContent = Array.isArray(previousAssistantMessage.content)
1046-
? previousAssistantMessage.content
1047-
: [{ type: "text", text: previousAssistantMessage.content }]
1048-
1049-
const toolUseBlocks = assistantContent.filter(
1050-
(block) => block.type === "tool_use",
1051-
) as Anthropic.Messages.ToolUseBlock[]
1052-
1053-
if (toolUseBlocks.length > 0) {
1054-
const existingToolResults = existingUserContent.filter(
1055-
(block) => block.type === "tool_result",
1056-
) as Anthropic.ToolResultBlockParam[]
1057-
1058-
const missingToolResponses: Anthropic.ToolResultBlockParam[] = toolUseBlocks
1059-
.filter(
1060-
(toolUse) => !existingToolResults.some((result) => result.tool_use_id === toolUse.id),
1061-
)
1062-
.map((toolUse) => ({
1063-
type: "tool_result",
1064-
tool_use_id: toolUse.id,
1065-
content: "Task was interrupted before this tool call could be completed.",
1066-
}))
1067-
1068-
modifiedApiConversationHistory = existingApiConversationHistory.slice(0, -1) // removes the last user message
1069-
modifiedOldUserContent = [...existingUserContent, ...missingToolResponses]
991+
return message
992+
})
993+
existingApiConversationHistory = conversationWithoutToolBlocks
994+
995+
// FIXME: remove tool use blocks altogether
996+
997+
// if the last message is an assistant message, we need to check if there's tool use since every tool use has to have a tool response
998+
// if there's no tool use and only a text block, then we can just add a user message
999+
// (note this isn't relevant anymore since we use custom tool prompts instead of tool use blocks, but this is here for legacy purposes in case users resume old tasks)
1000+
1001+
// if the last message is a user message, we can need to get the assistant message before it to see if it made tool calls, and if so, fill in the remaining tool responses with 'interrupted'
1002+
1003+
let modifiedApiConversationHistory: ApiMessage[] // need to remove the last user message to replace with new modified user message
1004+
if (existingApiConversationHistory.length > 0) {
1005+
const lastMessage = existingApiConversationHistory[existingApiConversationHistory.length - 1]
1006+
1007+
if (lastMessage.role === "assistant") {
1008+
const content = Array.isArray(lastMessage.content)
1009+
? lastMessage.content
1010+
: [{ type: "text", text: lastMessage.content }]
1011+
const hasToolUse = content.some((block) => block.type === "tool_use")
1012+
1013+
if (hasToolUse) {
1014+
const toolUseBlocks = content.filter(
1015+
(block) => block.type === "tool_use",
1016+
) as Anthropic.Messages.ToolUseBlock[]
1017+
const toolResponses: Anthropic.ToolResultBlockParam[] = toolUseBlocks.map((block) => ({
1018+
type: "tool_result",
1019+
tool_use_id: block.id,
1020+
content: "Task was interrupted before this tool call could be completed.",
1021+
}))
1022+
modifiedApiConversationHistory = [...existingApiConversationHistory] // no changes
1023+
modifiedOldUserContent = [...toolResponses]
1024+
} else {
1025+
modifiedApiConversationHistory = [...existingApiConversationHistory]
1026+
modifiedOldUserContent = []
1027+
}
1028+
} else if (lastMessage.role === "user") {
1029+
const previousAssistantMessage: ApiMessage | undefined =
1030+
existingApiConversationHistory[existingApiConversationHistory.length - 2]
1031+
1032+
const existingUserContent: Anthropic.Messages.ContentBlockParam[] = Array.isArray(
1033+
lastMessage.content,
1034+
)
1035+
? lastMessage.content
1036+
: [{ type: "text", text: lastMessage.content }]
1037+
if (previousAssistantMessage && previousAssistantMessage.role === "assistant") {
1038+
const assistantContent = Array.isArray(previousAssistantMessage.content)
1039+
? previousAssistantMessage.content
1040+
: [{ type: "text", text: previousAssistantMessage.content }]
1041+
1042+
const toolUseBlocks = assistantContent.filter(
1043+
(block) => block.type === "tool_use",
1044+
) as Anthropic.Messages.ToolUseBlock[]
1045+
1046+
if (toolUseBlocks.length > 0) {
1047+
const existingToolResults = existingUserContent.filter(
1048+
(block) => block.type === "tool_result",
1049+
) as Anthropic.ToolResultBlockParam[]
1050+
1051+
const missingToolResponses: Anthropic.ToolResultBlockParam[] = toolUseBlocks
1052+
.filter(
1053+
(toolUse) =>
1054+
!existingToolResults.some((result) => result.tool_use_id === toolUse.id),
1055+
)
1056+
.map((toolUse) => ({
1057+
type: "tool_result",
1058+
tool_use_id: toolUse.id,
1059+
content: "Task was interrupted before this tool call could be completed.",
1060+
}))
1061+
1062+
modifiedApiConversationHistory = existingApiConversationHistory.slice(0, -1) // removes the last user message
1063+
modifiedOldUserContent = [...existingUserContent, ...missingToolResponses]
1064+
} else {
1065+
modifiedApiConversationHistory = existingApiConversationHistory.slice(0, -1)
1066+
modifiedOldUserContent = [...existingUserContent]
1067+
}
10701068
} else {
10711069
modifiedApiConversationHistory = existingApiConversationHistory.slice(0, -1)
10721070
modifiedOldUserContent = [...existingUserContent]
10731071
}
10741072
} else {
1075-
modifiedApiConversationHistory = existingApiConversationHistory.slice(0, -1)
1076-
modifiedOldUserContent = [...existingUserContent]
1073+
throw new Error("Unexpected: Last message is not a user or assistant message")
10771074
}
10781075
} else {
1079-
throw new Error("Unexpected: Last message is not a user or assistant message")
1076+
throw new Error("Unexpected: No existing API conversation history")
10801077
}
1081-
} else {
1082-
throw new Error("Unexpected: No existing API conversation history")
1083-
}
1078+
return modifiedApiConversationHistory
1079+
})
10841080

1081+
if (!modifiedOldUserContent) {
1082+
throw new Error("modifiedOldUserContent was not set")
1083+
}
10851084
let newUserContent: Anthropic.Messages.ContentBlockParam[] = [...modifiedOldUserContent]
10861085

10871086
const agoText = ((): string => {
@@ -1130,10 +1129,6 @@ export class Task extends EventEmitter<ClineEvents> {
11301129
newUserContent.push(...formatResponse.imageBlocks(responseImages))
11311130
}
11321131

1133-
await this.modifyApiConversationHistory(async () => {
1134-
return modifiedApiConversationHistory
1135-
})
1136-
11371132
console.log(`[subtasks] task ${this.taskId}.${this.instanceId} resuming from history item`)
11381133

11391134
await this.initiateTaskLoop(newUserContent)

0 commit comments

Comments
 (0)