Subagent tool call persistence

This commit is contained in:
Siddharth Ganesan
2026-03-10 20:34:58 -07:00
parent 7b53137d22
commit f448134618
9 changed files with 170 additions and 11 deletions
@@ -206,6 +206,7 @@ export async function POST(req: NextRequest) {
: 'error'
: block.toolCall.status,
result: block.toolCall.result,
...(block.calledBy ? { calledBy: block.calledBy } : {}),
}
}
return stored
@@ -70,6 +70,7 @@ function mapStoredBlock(block: TaskStoredContentBlock): ContentBlock {
name: block.toolCall.name ?? 'unknown',
status: STATE_TO_STATUS[block.toolCall.state ?? ''] ?? 'success',
displayTitle: block.toolCall.display?.text,
calledBy: block.toolCall.calledBy,
result: block.toolCall.result,
}
}
@@ -383,11 +384,22 @@ export function useChat(workspaceId: string, initialChatId?: string): UseChatRet
let resource: MothershipResource | null = null
if (toolName === 'user_table') {
resource = extractTableResource(parsed, storedArgs, lastTableId)
if (resource) {
lastTableId = resource.id
queryClient.invalidateQueries({ queryKey: tableKeys.detail(resource.id) })
queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(resource.id) })
const redirected = extractFunctionExecuteResource(parsed, storedArgs)
if (redirected?.type === 'file') {
resource = redirected
queryClient.invalidateQueries({
queryKey: workspaceFilesKeys.list(workspaceId),
})
queryClient.invalidateQueries({
queryKey: workspaceFilesKeys.content(workspaceId, resource.id),
})
} else {
resource = extractTableResource(parsed, storedArgs, lastTableId)
if (resource) {
lastTableId = resource.id
queryClient.invalidateQueries({ queryKey: tableKeys.detail(resource.id) })
queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(resource.id) })
}
}
} else if (toolName === 'workspace_file') {
resource = extractFileResource(parsed, storedArgs)
@@ -415,6 +427,13 @@ export function useChat(workspaceId: string, initialChatId?: string): UseChatRet
})
}
}
} else if (toolName === 'read') {
resource = extractFunctionExecuteResource(parsed, storedArgs)
if (resource?.type === 'table') {
lastTableId = resource.id
queryClient.invalidateQueries({ queryKey: tableKeys.detail(resource.id) })
queryClient.invalidateQueries({ queryKey: tableKeys.rowsRoot(resource.id) })
}
} else if (toolName === 'create_workflow' || toolName === 'edit_workflow') {
resource = extractWorkflowResource(parsed, lastWorkflowId)
if (resource) {
@@ -7,6 +7,7 @@ export const RESOURCE_TOOL_NAMES = new Set([
'create_workflow',
'edit_workflow',
'function_execute',
'read',
])
function getResultData(parsed: SSEPayload): Record<string, unknown> | undefined {
@@ -138,13 +139,21 @@ export function extractResourcesFromHistory(messages: TaskStoredMessage[]): Moth
let resource: MothershipResource | null = null
if (tc.name === 'user_table') {
resource = extractTableResource(payload, args, lastTableId)
if (resource) lastTableId = resource.id
const redirected = extractFunctionExecuteResource(payload, args)
if (redirected?.type === 'file') {
resource = redirected
} else {
resource = extractTableResource(payload, args, lastTableId)
if (resource) lastTableId = resource.id
}
} else if (tc.name === 'workspace_file') {
resource = extractFileResource(payload, args)
} else if (tc.name === 'function_execute') {
resource = extractFunctionExecuteResource(payload, args)
if (resource?.type === 'table') lastTableId = resource.id
} else if (tc.name === 'read') {
resource = extractFunctionExecuteResource(payload, args)
if (resource?.type === 'table') lastTableId = resource.id
} else if (tc.name === 'create_workflow' || tc.name === 'edit_workflow') {
resource = extractWorkflowResource(payload, lastWorkflowId)
if (resource) lastWorkflowId = resource.id
+1
View File
@@ -41,6 +41,7 @@ export interface TaskStoredContentBlock {
params?: Record<string, unknown>
result?: { success: boolean; output?: unknown; error?: string }
display?: { text?: string }
calledBy?: string
} | null
}
@@ -484,6 +484,12 @@ export const subAgentHandlers: Record<string, SSEHandler> = {
}
if (!context.toolCalls.has(toolCallId)) {
context.toolCalls.set(toolCallId, toolCall)
const parentToolCall = context.toolCalls.get(parentToolCallId)
addContentBlock(context, {
type: 'tool_call',
toolCall,
calledBy: parentToolCall?.name,
})
}
if (isPartial) return
@@ -206,6 +206,119 @@ async function maybeWriteOutputToTable(
}
}
async function maybeWriteReadCsvToTable(
toolName: string,
params: Record<string, unknown> | undefined,
result: ToolCallResult,
context: ExecutionContext
): Promise<ToolCallResult> {
if (toolName !== 'read') return result
if (!result.success || !result.output) return result
if (!context.workspaceId || !context.userId) return result
const outputTable = params?.outputTable as string | undefined
if (!outputTable) return result
try {
const table = await getTableById(outputTable)
if (!table) {
return { success: false, error: `Table "${outputTable}" not found` }
}
const output = result.output as Record<string, unknown>
const content = (output.content as string) || ''
if (!content.trim()) {
return { success: false, error: 'File has no content to import into table' }
}
const filePath = (params?.path as string) || ''
const ext = filePath.split('.').pop()?.toLowerCase()
let rows: Record<string, unknown>[]
if (ext === 'json') {
const parsed = JSON.parse(content)
if (!Array.isArray(parsed)) {
return {
success: false,
error: 'JSON file must contain an array of objects for table import',
}
}
rows = parsed
} else {
const { parse } = await import('csv-parse/sync')
rows = parse(content, {
columns: true,
skip_empty_lines: true,
trim: true,
relax_column_count: true,
relax_quotes: true,
skip_records_with_error: true,
cast: false,
}) as Record<string, unknown>[]
}
if (rows.length === 0) {
return { success: false, error: 'File has no data rows to import' }
}
if (rows.length > MAX_OUTPUT_TABLE_ROWS) {
return {
success: false,
error: `Row limit exceeded: got ${rows.length}, max is ${MAX_OUTPUT_TABLE_ROWS}`,
}
}
await db.transaction(async (tx) => {
await tx.delete(userTableRows).where(eq(userTableRows.tableId, outputTable))
const now = new Date()
for (let i = 0; i < rows.length; i += BATCH_CHUNK_SIZE) {
const chunk = rows.slice(i, i + BATCH_CHUNK_SIZE)
const values = chunk.map((rowData, j) => ({
id: `row_${crypto.randomUUID().replace(/-/g, '')}`,
tableId: outputTable,
workspaceId: context.workspaceId!,
data: rowData,
position: i + j,
createdAt: now,
updatedAt: now,
createdBy: context.userId,
}))
await tx.insert(userTableRows).values(values)
}
})
logger.info('Read output written to table', {
toolName,
tableId: outputTable,
tableName: table.name,
rowCount: rows.length,
filePath,
})
return {
success: true,
output: {
message: `Imported ${rows.length} rows from "${filePath}" into table "${table.name}"`,
tableId: outputTable,
tableName: table.name,
rowCount: rows.length,
},
}
} catch (err) {
logger.warn('Failed to write read output to table', {
toolName,
outputTable,
error: err instanceof Error ? err.message : String(err),
})
return {
success: false,
error: `Failed to import into table: ${err instanceof Error ? err.message : String(err)}`,
}
}
}
export async function executeToolAndReport(
toolCallId: string,
context: StreamingContext,
@@ -230,6 +343,7 @@ export async function executeToolAndReport(
let result = await executeToolServerSide(toolCall, execContext)
result = await maybeWriteOutputToFile(toolCall.name, toolCall.params, result, execContext)
result = await maybeWriteOutputToTable(toolCall.name, toolCall.params, result, execContext)
result = await maybeWriteReadCsvToTable(toolCall.name, toolCall.params, result, execContext)
toolCall.status = result.success ? 'success' : 'error'
toolCall.result = result
toolCall.error = result.error
@@ -176,12 +176,20 @@ export async function runStreamLoop(
if (normalizedEvent.type === 'subagent_start') {
const eventData = normalizedEvent.data as Record<string, unknown> | undefined
const toolCallId = eventData?.tool_call_id as string | undefined
const subagentName = normalizedEvent.subagent || (eventData?.agent as string | undefined)
if (toolCallId) {
context.subAgentParentStack.push(toolCallId)
context.subAgentParentToolCallId = toolCallId
context.subAgentContent[toolCallId] = ''
context.subAgentToolCalls[toolCallId] = []
}
if (subagentName) {
context.contentBlocks.push({
type: 'subagent',
content: subagentName,
timestamp: Date.now(),
})
}
continue
}
+2 -1
View File
@@ -59,12 +59,13 @@ export interface ToolCallResult<T = unknown> {
error?: string
}
export type ContentBlockType = 'text' | 'thinking' | 'tool_call' | 'subagent_text'
export type ContentBlockType = 'text' | 'thinking' | 'tool_call' | 'subagent_text' | 'subagent'
export interface ContentBlock {
type: ContentBlockType
content?: string
toolCall?: ToolCallState
calledBy?: string
timestamp: number
}
+3 -3
View File
@@ -37,9 +37,9 @@ export class CsvParser implements FileParser {
`Parsing CSV buffer, size: ${bufferSize} bytes (${(bufferSize / 1024 / 1024).toFixed(2)} MB)`
)
const stream = Readable.from(buffer, {
highWaterMark: CONFIG.STREAM_CHUNK_SIZE,
})
const stream = new Readable({ read() {} })
stream.push(buffer)
stream.push(null)
return this.parseStream(stream)
}