|
1 | 1 | import { createLogger } from '@sim/logger' |
| 2 | +import { getErrorMessage } from '@sim/utils/errors' |
2 | 3 | import { sleep } from '@sim/utils/helpers' |
3 | 4 | import { DEFAULT_EXECUTION_TIMEOUT_MS } from '@/lib/core/execution-limits' |
4 | | -import type { |
5 | | - DatabricksGenieAskParams, |
6 | | - DatabricksGenieAskResponse, |
7 | | - DatabricksGenieGetMessageResponse, |
8 | | - DatabricksGenieQueryResultResponse, |
| 5 | +import { |
| 6 | + type DatabricksGenieAskParams, |
| 7 | + type DatabricksGenieAskResponse, |
| 8 | + type DatabricksGenieGetMessageResponse, |
| 9 | + type DatabricksGenieQueryResultResponse, |
| 10 | + GENIE_MESSAGE_OUTPUT_PROPERTIES, |
| 11 | + STATEMENT_RESULT_OUTPUT_PROPERTIES, |
9 | 12 | } from '@/tools/databricks/types' |
10 | 13 | import { |
11 | 14 | databricksErrorMessage, |
12 | 15 | databricksUrl, |
13 | | - GENIE_MESSAGE_OUTPUT_PROPERTIES, |
14 | 16 | GENIE_SPACE_PARAMS, |
15 | 17 | GENIE_TERMINAL_STATUSES, |
16 | 18 | genieConversationPath, |
17 | 19 | genieSpacePath, |
18 | 20 | mapGenieMessage, |
19 | | - STATEMENT_RESULT_OUTPUT_PROPERTIES, |
20 | 21 | } from '@/tools/databricks/utils' |
21 | 22 | import type { ToolConfig } from '@/tools/types' |
22 | 23 |
|
@@ -129,75 +130,93 @@ export const genieAskTool: ToolConfig<DatabricksGenieAskParams, DatabricksGenieA |
129 | 130 | } |
130 | 131 |
|
131 | 132 | let message = result.output |
132 | | - const startedAt = Date.now() |
133 | | - let attempt = 0 |
134 | 133 |
|
135 | | - while (!GENIE_TERMINAL_STATUSES.has(message.status)) { |
136 | | - if (Date.now() - startedAt >= MAX_POLL_TIME_MS) { |
137 | | - logger.warn(`Genie message ${messageId} did not complete within ${MAX_POLL_TIME_MS} ms`) |
| 134 | + /** |
| 135 | + * A nested call can throw instead of returning a failed result, and the executor falls back |
| 136 | + * to the pending first response when post-processing throws. Convert every failure here so |
| 137 | + * the ask fails with its conversation and message IDs intact. |
| 138 | + */ |
| 139 | + try { |
| 140 | + const startedAt = Date.now() |
| 141 | + let attempt = 0 |
| 142 | + |
| 143 | + while (!GENIE_TERMINAL_STATUSES.has(message.status)) { |
| 144 | + if (Date.now() - startedAt >= MAX_POLL_TIME_MS) { |
| 145 | + logger.warn(`Genie message ${messageId} did not complete within ${MAX_POLL_TIME_MS} ms`) |
| 146 | + return { |
| 147 | + ...result, |
| 148 | + output: message, |
| 149 | + success: false, |
| 150 | + error: `Genie did not answer within ${MAX_POLL_TIME_MS / 1000}s (last status: ${message.status}). Use Get Genie Message with this conversation and message ID to check on it.`, |
| 151 | + } |
| 152 | + } |
| 153 | + |
| 154 | + attempt += 1 |
| 155 | + const intervalMs = Math.min(1000 * attempt, MAX_POLL_INTERVAL_MS) |
| 156 | + await sleep(intervalMs) |
| 157 | + |
| 158 | + const polled = (await executeTool( |
| 159 | + 'databricks_genie_get_message', |
| 160 | + messageParams |
| 161 | + )) as DatabricksGenieGetMessageResponse |
| 162 | + if (!polled.success) { |
| 163 | + return { |
| 164 | + ...result, |
| 165 | + output: message, |
| 166 | + success: false, |
| 167 | + error: polled.error ?? 'Failed to poll Genie message', |
| 168 | + } |
| 169 | + } |
| 170 | + message = { ...message, ...polled.output } |
| 171 | + } |
| 172 | + |
| 173 | + if (message.status === 'FAILED' || message.status === 'CANCELLED') { |
138 | 174 | return { |
139 | 175 | ...result, |
140 | 176 | output: message, |
141 | 177 | success: false, |
142 | | - error: `Genie did not answer within ${MAX_POLL_TIME_MS / 1000}s (last status: ${message.status}). Use Get Genie Message with this conversation and message ID to check on it.`, |
| 178 | + error: message.error ?? `Genie message ${message.status.toLowerCase()}`, |
143 | 179 | } |
144 | 180 | } |
145 | 181 |
|
146 | | - attempt += 1 |
147 | | - const intervalMs = Math.min(1000 * attempt, MAX_POLL_INTERVAL_MS) |
148 | | - await sleep(intervalMs) |
| 182 | + if (message.status !== 'COMPLETED' || !message.queryAttachmentId) { |
| 183 | + return { ...result, output: message } |
| 184 | + } |
149 | 185 |
|
150 | | - const polled = (await executeTool( |
151 | | - 'databricks_genie_get_message', |
152 | | - messageParams |
153 | | - )) as DatabricksGenieGetMessageResponse |
154 | | - if (!polled.success) { |
| 186 | + const queryResult = (await executeTool('databricks_genie_get_query_result', { |
| 187 | + ...messageParams, |
| 188 | + attachmentId: message.queryAttachmentId, |
| 189 | + })) as DatabricksGenieQueryResultResponse |
| 190 | + if (!queryResult.success) { |
155 | 191 | return { |
156 | 192 | ...result, |
157 | 193 | output: message, |
158 | 194 | success: false, |
159 | | - error: polled.error ?? 'Failed to poll Genie message', |
| 195 | + error: queryResult.error ?? 'Failed to get Genie query result', |
160 | 196 | } |
161 | 197 | } |
162 | | - message = { ...message, ...polled.output } |
163 | | - } |
164 | 198 |
|
165 | | - if (message.status === 'FAILED' || message.status === 'CANCELLED') { |
166 | 199 | return { |
167 | 200 | ...result, |
168 | | - output: message, |
169 | | - success: false, |
170 | | - error: message.error ?? `Genie message ${message.status.toLowerCase()}`, |
| 201 | + output: { |
| 202 | + ...message, |
| 203 | + columns: queryResult.output.columns, |
| 204 | + data: queryResult.output.data, |
| 205 | + totalRows: queryResult.output.totalRows, |
| 206 | + truncated: queryResult.output.truncated, |
| 207 | + }, |
171 | 208 | } |
172 | | - } |
173 | | - |
174 | | - if (message.status !== 'COMPLETED' || !message.queryAttachmentId) { |
175 | | - return { ...result, output: message } |
176 | | - } |
177 | | - |
178 | | - const queryResult = (await executeTool('databricks_genie_get_query_result', { |
179 | | - ...messageParams, |
180 | | - attachmentId: message.queryAttachmentId, |
181 | | - })) as DatabricksGenieQueryResultResponse |
182 | | - if (!queryResult.success) { |
| 209 | + } catch (error) { |
| 210 | + logger.error(`Error waiting for Genie message ${messageId}`, { |
| 211 | + message: getErrorMessage(error, 'Unknown error'), |
| 212 | + }) |
183 | 213 | return { |
184 | 214 | ...result, |
185 | 215 | output: message, |
186 | 216 | success: false, |
187 | | - error: queryResult.error ?? 'Failed to get Genie query result', |
| 217 | + error: `Error waiting for Genie to answer: ${getErrorMessage(error, 'Unknown error')}`, |
188 | 218 | } |
189 | 219 | } |
190 | | - |
191 | | - return { |
192 | | - ...result, |
193 | | - output: { |
194 | | - ...message, |
195 | | - columns: queryResult.output.columns, |
196 | | - data: queryResult.output.data, |
197 | | - totalRows: queryResult.output.totalRows, |
198 | | - truncated: queryResult.output.truncated, |
199 | | - }, |
200 | | - } |
201 | 220 | }, |
202 | 221 |
|
203 | 222 | outputs: { |
|
0 commit comments