兼容Debug EML SSE断流兜底

This commit is contained in:
andy
2026-07-12 19:26:52 +08:00
parent 9d095cfd53
commit 807d045526
17 changed files with 556 additions and 47 deletions

View File

@@ -13,6 +13,8 @@ import { isAuthSessionInvalidPayload, notifyUnauthorized } from '@/services/http
const apiBaseUrl = import.meta.env.VITE_API_BASE_URL ?? ''
const debugEmlEndpoint = '/api/system/debug/eml-superagent-runs'
const debugEmlStreamEndpoint = `${debugEmlEndpoint}/stream`
const streamFallbackPollIntervalMs = 2_000
const streamFallbackPollTimeoutMs = 30 * 60 * 1_000
const jsonSecretPattern =
/("(?:api[_-]?key|token|secret|password|cookie|authorization)"\s*:\s*")[^"]*(")/gi
const headerSecretPattern = /((?:authorization|cookie)\s*:\s*)[^\s,;]+/gi
@@ -99,42 +101,120 @@ export async function uploadDebugEmlSuperAgentRunStream(
let finalResult: DebugEmlSuperAgentRunResult | null = null
let streamError: DebugEmlUploadError | null = null
await readSseStream(response.body, (event) => {
const payload = parseSsePayload(event.data)
if (event.event === 'debug_stage') {
handlers.onStage?.(normalizeStageEvent(payload))
return
}
if (event.event === 'superagent_trace') {
handlers.onTrace?.(normalizeTraceEvent(payload))
return
}
if (event.event === 'debug_heartbeat') {
return
}
if (event.event === 'superagent_result') {
finalResult = normalizeDebugEmlResult(payload)
handlers.onResult?.(finalResult)
return
}
if (event.event === 'debug_error') {
const errorPayload = isDebugEmlErrorResponse(payload) ? payload : {}
streamError = new DebugEmlUploadError(
errorPayload.message || 'Debug EML 调试失败。',
response.status,
errorPayload.error_code || 'DEBUG_EML_RUN_FAILED',
payload,
)
}
})
let fallbackDebugRunId: string | null = null
let streamReadError: unknown = null
try {
await readSseStream(response.body, (event) => {
const payload = parseSsePayload(event.data)
if (event.event === 'debug_stage') {
const stage = normalizeStageEvent(payload)
fallbackDebugRunId = stage.debug_run_id ?? fallbackDebugRunId
handlers.onStage?.(stage)
return
}
if (event.event === 'superagent_trace') {
handlers.onTrace?.(normalizeTraceEvent(payload))
return
}
if (event.event === 'debug_heartbeat') {
const heartbeat = normalizeStageEvent(payload)
fallbackDebugRunId = heartbeat.debug_run_id ?? fallbackDebugRunId
return
}
if (event.event === 'superagent_result') {
finalResult = normalizeDebugEmlResult(payload)
fallbackDebugRunId = finalResult.debug_run_id || fallbackDebugRunId
handlers.onResult?.(finalResult)
return
}
if (event.event === 'debug_error') {
const errorPayload = isDebugEmlErrorResponse(payload) ? payload : {}
streamError = new DebugEmlUploadError(
errorPayload.message || 'Debug EML 调试失败。',
response.status,
errorPayload.error_code || 'DEBUG_EML_RUN_FAILED',
payload,
)
}
})
} catch (error) {
streamReadError = error
}
if (streamError) {
throw streamError
}
if (!finalResult) {
throw new DebugEmlUploadError('Debug EML SSE 未收到最终结果。', response.status, 'DEBUG_EML_RUN_FAILED', null)
if (finalResult) {
return finalResult
}
return finalResult
if (fallbackDebugRunId) {
return pollDebugEmlSuperAgentRun(input.debugUploadKey, fallbackDebugRunId)
}
if (streamReadError) {
throw new DebugEmlUploadError('Debug EML SSE 读取失败。', response.status, 'DEBUG_EML_RUN_FAILED', null)
}
throw new DebugEmlUploadError('Debug EML SSE 未收到最终结果。', response.status, 'DEBUG_EML_RUN_FAILED', null)
}
async function getDebugEmlSuperAgentRun(
debugUploadKey: string,
debugRunId: string,
): Promise<DebugEmlSuperAgentRunResult> {
const response = await fetch(`${apiBaseUrl}${debugEmlEndpoint}/${encodeURIComponent(debugRunId)}`, {
method: 'GET',
headers: buildUploadHeaders(debugUploadKey),
})
const payload = await readResponsePayload(response)
if (!response.ok) {
if (isAuthSessionInvalidPayload(payload)) {
notifyUnauthorized()
}
const errorPayload = isDebugEmlErrorResponse(payload) ? payload : {}
throw new DebugEmlUploadError(
errorPayload.message || `Request failed with status ${response.status}`,
response.status,
errorPayload.error_code || 'DEBUG_EML_RUN_FAILED',
payload,
)
}
return normalizeDebugEmlResult(payload)
}
async function pollDebugEmlSuperAgentRun(
debugUploadKey: string,
debugRunId: string,
): Promise<DebugEmlSuperAgentRunResult> {
const deadline = Date.now() + streamFallbackPollTimeoutMs
let latestResult: DebugEmlSuperAgentRunResult | null = null
while (Date.now() <= deadline) {
latestResult = await getDebugEmlSuperAgentRun(debugUploadKey, debugRunId)
if (latestResult.status === 'SUPERAGENT_SUCCEEDED') {
return latestResult
}
if (latestResult.status === 'SUPERAGENT_FAILED' || latestResult.status === 'FAILED') {
throw new DebugEmlUploadError(
latestResult.safe_error_summary || 'Debug EML 调试失败。',
200,
latestResult.status === 'SUPERAGENT_FAILED'
? 'SUPERAGENT_OPEN_API_FAILED'
: 'DEBUG_EML_RUN_FAILED',
latestResult,
)
}
await delay(streamFallbackPollIntervalMs)
}
throw new DebugEmlUploadError(
latestResult?.safe_error_summary || 'Debug EML SSE 未收到最终结果。',
200,
'DEBUG_EML_RUN_FAILED',
latestResult,
)
}
function delay(ms: number): Promise<void> {
return new Promise((resolve) => {
window.setTimeout(resolve, ms)
})
}
function buildUploadForm(input: DebugEmlUploadInput): FormData {
@@ -202,6 +282,7 @@ function normalizeDebugEmlResult(payload: unknown): DebugEmlSuperAgentRunResult
: [],
warnings: Array.isArray(result.warnings) ? result.warnings.filter(isString) : [],
status: nullableStringValue(result.status),
safe_error_summary: nullableStringValue(result.safe_error_summary),
}
}

View File

@@ -41,6 +41,27 @@ function mockSseResponse(chunks: string[], status = 200): Response {
})
}
function mockErroredSseResponse(chunksBeforeError: string[], error: Error, status = 200): Response {
const encoder = new TextEncoder()
let nextChunkIndex = 0
const stream = new ReadableStream<Uint8Array>({
pull(controller) {
if (nextChunkIndex < chunksBeforeError.length) {
controller.enqueue(encoder.encode(chunksBeforeError[nextChunkIndex]))
nextChunkIndex += 1
return
}
controller.error(error)
},
})
return new Response(stream, {
status,
headers: {
'Content-Type': 'text/event-stream',
},
})
}
describe('debugEmlService', () => {
beforeEach(() => {
vi.restoreAllMocks()
@@ -275,4 +296,86 @@ describe('debugEmlService', () => {
message: 'SuperAgent 调用失败。',
} satisfies Partial<DebugEmlUploadError>)
})
it('polls backend run status when SSE closes before final result', async () => {
const fetchMock = vi.spyOn(globalThis, 'fetch')
fetchMock.mockResolvedValueOnce(
mockSseResponse([
'event: debug_stage\n',
'data: {"debug_run_id":"90005","status":"CALLING_SUPERAGENT","safe_summary":"阶段:调用 SuperAgent Open API。"}\n\n',
]),
)
fetchMock.mockResolvedValueOnce(
mockJsonResponse({
debug_run_id: '90005',
source_message_id: '30005',
uploaded_media: [],
superagent_session_id: 'session-poll-1',
superagent_run_id: 'run-poll-1',
superagent_raw_answer: '{"route_code":"S10"}',
superagent_parsed_json: { route_code: 'S10' },
superagent_trace_events: [],
warnings: [],
status: 'SUPERAGENT_SUCCEEDED',
}),
)
const file = new File(['From: guest@example.test'], 'booking.eml', { type: 'message/rfc822' })
const result = await uploadDebugEmlSuperAgentRunStream({
hotelId: 'HOTEL-TEST',
debugUploadKey: 'manual-debug-key',
file,
})
expect(fetchMock).toHaveBeenCalledTimes(2)
expect(fetchMock.mock.calls[1]?.[0]).toBe('/api/system/debug/eml-superagent-runs/90005')
expect(fetchMock.mock.calls[1]?.[1]).toMatchObject({
method: 'GET',
headers: {
Accept: 'application/json',
'X-TH-Hotel-Debug-Upload-Key': 'manual-debug-key',
},
})
expect(result.superagent_run_id).toBe('run-poll-1')
expect(result.superagent_parsed_json).toEqual({ route_code: 'S10' })
})
it('polls backend run status when SSE stream errors after sending debug run id', async () => {
const fetchMock = vi.spyOn(globalThis, 'fetch')
fetchMock.mockResolvedValueOnce(
mockErroredSseResponse(
[
'event: debug_stage\n',
'data: {"debug_run_id":"90006","status":"CALLING_SUPERAGENT","safe_summary":"阶段:调用 SuperAgent Open API。"}\n\n',
],
new Error('stream reset by proxy'),
),
)
fetchMock.mockResolvedValueOnce(
mockJsonResponse({
debug_run_id: '90006',
source_message_id: '30006',
uploaded_media: [],
superagent_session_id: 'session-poll-error-1',
superagent_run_id: 'run-poll-error-1',
superagent_raw_answer: '{"route_code":"S10"}',
superagent_parsed_json: { route_code: 'S10' },
superagent_trace_events: [],
warnings: [],
status: 'SUPERAGENT_SUCCEEDED',
}),
)
const file = new File(['From: guest@example.test'], 'booking.eml', { type: 'message/rfc822' })
const result = await uploadDebugEmlSuperAgentRunStream({
hotelId: 'HOTEL-TEST',
debugUploadKey: 'manual-debug-key',
file,
})
expect(fetchMock).toHaveBeenCalledTimes(2)
expect(fetchMock.mock.calls[1]?.[0]).toBe('/api/system/debug/eml-superagent-runs/90006')
expect(result.superagent_run_id).toBe('run-poll-error-1')
expect(result.superagent_parsed_json).toEqual({ route_code: 'S10' })
})
})

View File

@@ -83,6 +83,7 @@ function createResult() {
],
warnings: ['SuperAgent 返回非 JSON 时会展示原始回答'],
status: 'SUPERAGENT_SUCCEEDED',
safe_error_summary: null,
}
}

View File

@@ -38,6 +38,7 @@ export interface DebugEmlSuperAgentRunResult {
superagent_trace_events: DebugEmlSuperAgentTraceEvent[]
warnings: string[]
status: string | null
safe_error_summary: string | null
}
export interface DebugEmlSuperAgentTraceEvent {

View File

@@ -22,7 +22,8 @@ Debug EML 页面第一版只做一件事:
- 不调用 SuperAgent 任务结果通知接口。
- 不触发 AgentBus 实时链路。
- 不做批量上传。
- 不提供 Debug run 历史列表或详情查询
- 不提供 Debug run 历史列表。
- 仅提供按 `debug_run_id` 查询单次运行状态和安全结果,用于 SSE 断流后的前端兜底轮询。
- 不在生产普通业务页面开放。
## 3. 页面建议结构
@@ -40,6 +41,17 @@ Debug EML 页面第一版只做一件事:
## 4. 接口
实时 Trace 页面优先使用流式接口:
```text
POST /api/system/debug/eml-superagent-runs/stream
Content-Type: multipart/form-data
Accept: text/event-stream
Header: X-TH-Hotel-Debug-Upload-Key: <调试上传口令>
```
同步调试接口仍保留:
```text
POST /api/system/debug/eml-superagent-runs
Content-Type: multipart/form-data
@@ -47,6 +59,16 @@ Accept: application/json
Header: X-TH-Hotel-Debug-Upload-Key: <调试上传口令>
```
SSE 断流兜底查询接口:
```text
GET /api/system/debug/eml-superagent-runs/{debug_run_id}
Accept: application/json
Header: X-TH-Hotel-Debug-Upload-Key: <调试上传口令>
```
中文说明:前端不直接调用 SuperAgent。上传后如果 SSE 已收到 `debug_run_id`,但浏览器或反向代理提前关闭了流,前端可以用同一个 Debug 上传口令轮询该 GET 接口,直到状态变成 `SUPERAGENT_SUCCEEDED``SUPERAGENT_FAILED``FAILED`。该接口不是历史列表接口,也不允许绕过 Debug 上传口令。
注意:
- 该接口只有在后端 `debug.eml-upload.enabled=true` 时存在;如果后端未开启,前端可能收到 `404`
@@ -229,7 +251,8 @@ idle
- 文件未选择或 Debug 上传口令为空时禁用提交按钮;`hotel_id` 为空是允许的,表示使用后端系统酒店。
- 提交后禁用文件选择和提交按钮,避免重复上传。
- SuperAgent 调用可能耗时较长,页面 loading 文案不要只写“上传中”,建议写“正在解析邮件并等待 SuperAgent 返回”。
- 成功后保留本次响应在页面内存中;当前没有 Debug run 查询接口,刷新页面后需要重新上传。
- 成功后保留本次响应在页面内存中;当前没有 Debug run 历史列表,刷新页面后通常需要重新上传或手动使用已知 `debug_run_id` 排查
- 如果 SSE 正常返回 `superagent_result`,以前端收到的最终结果为准;如果 SSE 在最终结果前关闭但已经收到 `debug_run_id`,前端应轮询 GET 兜底接口。
- 再次上传同一封 `.eml` 会生成新的 `external_message_id` 和新的 SourceMessage前端不要按原始 `Message-ID` 去重。
## 8. 安全与日志
@@ -268,8 +291,8 @@ idle
## 11. 当前后置事项
- Debug run 历史列表 / 详情查询接口未做。
- Debug run 取消或超时轮询接口未做。
- Debug run 历史列表未做。
- Debug run 取消接口未做。
- 批量上传未做。
- 前端白名单元数据独立接口未做。
- 用户身份 / 权限体系未接入;当前依赖调试上传口令保护。

View File

@@ -158,6 +158,7 @@
| `DEBUG_EML_UPLOAD_PROD_ACCESS_KEY` | 是 | prod Debug EML 上传访问口令;生产通常不应启用该接口。 |
| `DEBUG_EML_UPLOAD_MAX_FILE_BYTES` | 否 | `.eml` 上传大小上限,默认 `10485760`。 |
| `DEBUG_EML_UPLOAD_SSE_HEARTBEAT_INTERVAL` | 否 | Debug EML 页面到后端的 SSE 心跳间隔,默认 `15s`;测试机如仍遇到空闲断流可调小到 `10s`。 |
| `DEBUG_EML_UPLOAD_SSE_REQUEST_TIMEOUT` | 否 | Spring MVC 异步请求总超时,默认 `1800s`;当前用于保障 Debug EML SSE 不先于 SuperAgent read timeout 关闭。注意 Spring MVC async timeout 是应用级全局设置,后续如增加其他 async/SSE 接口需一起评估。 |
| `DEERFLOW_DEV_BASE_URL` / `DEERFLOW_TEST_BASE_URL` / `DEERFLOW_PROD_BASE_URL` | 否 | SuperAgent / DeerFlow Open API 基础地址,未配置时可兜底 `DEERFLOW_BASE_URL`。 |
| `DEERFLOW_DEV_OPEN_API_KEY` / `DEERFLOW_TEST_OPEN_API_KEY` / `DEERFLOW_PROD_OPEN_API_KEY` | 是 | SuperAgent Open API Key未配置时可兜底 `DEERFLOW_OPEN_API_KEY`。 |
| `SUPERAGENT_DEV_OPEN_API_ENABLED` / `SUPERAGENT_TEST_OPEN_API_ENABLED` / `SUPERAGENT_PROD_OPEN_API_ENABLED` | 否 | 是否启用真实 SuperAgent Open API 调用prod 默认关闭。 |

View File

@@ -426,6 +426,7 @@ DEBUG_EML_UPLOAD_ENABLED=true
DEBUG_EML_UPLOAD_ACCESS_KEY=
DEBUG_EML_UPLOAD_MAX_FILE_BYTES=10485760
DEBUG_EML_UPLOAD_SSE_HEARTBEAT_INTERVAL=15s
DEBUG_EML_UPLOAD_SSE_REQUEST_TIMEOUT=1800s
ALIYUN_OSS_ENDPOINT=
ALIYUN_OSS_BUCKET=
@@ -441,6 +442,12 @@ SUPERAGENT_DEBUG_EML_CONNECT_TIMEOUT=15s
SUPERAGENT_DEBUG_EML_READ_TIMEOUT=180s
```
中文说明:
- `DEBUG_EML_UPLOAD_SSE_HEARTBEAT_INTERVAL` 控制 Debug EML SSE 在等待 SuperAgent Open API 返回期间的心跳事件间隔,用于避免中间链路按空闲连接断开。
- `DEBUG_EML_UPLOAD_SSE_REQUEST_TIMEOUT` 目前映射到 Spring MVC async request timeout。Spring MVC 对 `StreamingResponseBody` 使用全局异步超时,因此该变量虽然为 Debug EML 设置,但会影响同一 Spring MVC 应用内其他异步请求;如后续增加其他 SSE / async 接口,需要统一评估该全局值。
- 前端实时页面优先调用 `POST /api/system/debug/eml-superagent-runs/stream`。如果 SSE 在最终 `superagent_result` 前断开但已收到 `debug_run_id`,前端可通过 `GET /api/system/debug/eml-superagent-runs/{debug_run_id}` 轮询运行状态和安全结果;该 GET 接口不是历史列表接口,仍必须携带 `X-TH-Hotel-Debug-Upload-Key`
分环境建议:
- dev 可以默认关闭真实 SuperAgent 调用,但允许配置后开启。

View File

@@ -93,16 +93,16 @@ public class SuperAgentOpenApiSseParser {
* 汇总解析状态并生成结果对象。
*/
private SuperAgentOpenApiResult buildResult(String sessionId, Set<String> eventTypes, ParsedState state) {
if (!state.endSeen) {
throw new SuperAgentOpenApiException("SuperAgent SSE 未收到结束事件。");
}
if ((state.rawAnswer == null || state.rawAnswer.isBlank()) && state.fallbackRawAnswer != null) {
if ((state.rawAnswer == null || state.rawAnswer.isBlank()) && state.endSeen && state.fallbackRawAnswer != null) {
state.rawAnswer = state.fallbackRawAnswer;
state.modelName = state.fallbackModelName;
state.inputTokens = state.fallbackInputTokens;
state.outputTokens = state.fallbackOutputTokens;
state.totalTokens = state.fallbackTotalTokens;
}
if (!state.endSeen && (state.rawAnswer == null || state.rawAnswer.isBlank())) {
throw new SuperAgentOpenApiException("SuperAgent SSE 未收到结束事件。");
}
if (state.rawAnswer == null || state.rawAnswer.isBlank()) {
throw new SuperAgentOpenApiException("SuperAgent SSE 未找到最终 AI 回答。");
}

View File

@@ -9,13 +9,49 @@ import java.time.LocalDateTime;
* @param hotelId 酒店或业务上下文 ID
* @param runStatus 当前运行状态
* @param sourceMessageId 关联的内部 SourceMessage ID
* @param externalMessageId Debug 外部邮件 ID
* @param externalConversationId Debug 外部会话 ID
* @param originalFileName 原始 EML 安全文件名
* @param originalEmlOssUrl 原始 EML OSS URL
* @param originalEmlSha256 原始 EML SHA-256
* @param payloadJson 发送给 SuperAgent 的 AgentBus-like payload
* @param superagentSessionId SuperAgent session ID
* @param superagentRunId SuperAgent run ID
* @param superagentProfileId SuperAgent profile ID
* @param superagentProfileVersionId SuperAgent profile version ID
* @param superagentModelName SuperAgent 模型名称
* @param superagentRawAnswer SuperAgent 最终原始回答
* @param superagentParsedJson 后端解析出的 SuperAgent JSON 文本
* @param superagentInputTokens SuperAgent 输入 token 数
* @param superagentOutputTokens SuperAgent 输出 token 数
* @param superagentTotalTokens SuperAgent 总 token 数
* @param safeErrorSummary 安全错误摘要
* @param createdAt 创建 UTC 时间
* @param updatedAt 更新 UTC 时间
*/
public record DebugEmlSuperAgentRunSnapshot(
Long id,
String hotelId,
String runStatus,
Long sourceMessageId,
LocalDateTime createdAt
String externalMessageId,
String externalConversationId,
String originalFileName,
String originalEmlOssUrl,
String originalEmlSha256,
String payloadJson,
String superagentSessionId,
String superagentRunId,
String superagentProfileId,
String superagentProfileVersionId,
String superagentModelName,
String superagentRawAnswer,
String superagentParsedJson,
Integer superagentInputTokens,
Integer superagentOutputTokens,
Integer superagentTotalTokens,
String safeErrorSummary,
LocalDateTime createdAt,
LocalDateTime updatedAt
) {
}

View File

@@ -29,6 +29,7 @@ import java.util.Map;
* @param superagentTraceEvents SuperAgent 公开 Trace 事件列表
* @param warnings 可展示的安全警告
* @param status Debug 运行状态
* @param safeErrorSummary 安全错误摘要,成功时为空;不包含 Secret、邮件正文或附件签名 URL
*/
public record DebugEmlSuperAgentRunResult(
@JsonProperty("debug_run_id")
@@ -68,6 +69,8 @@ public record DebugEmlSuperAgentRunResult(
@JsonProperty("superagent_trace_events")
List<SuperAgentOpenApiTraceEvent> superagentTraceEvents,
List<String> warnings,
String status
String status,
@JsonProperty("safe_error_summary")
String safeErrorSummary
) {
}

View File

@@ -7,6 +7,8 @@ import org.springframework.http.HttpHeaders;
import org.springframework.http.HttpStatus;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RequestMapping;
@@ -44,6 +46,16 @@ public class DebugEmlSuperAgentController {
return ResponseEntity.status(HttpStatus.CREATED).body(runService.uploadAndRun(accessKey, file, hotelId, runLabel));
}
/**
* 查询 Debug EML 运行结果,用于流式响应提前结束后的前端兜底轮询。
*/
@GetMapping(path = "/{runId}", produces = MediaType.APPLICATION_JSON_VALUE)
public DebugEmlSuperAgentRunResult getRun(
@RequestHeader(name = "X-TH-Hotel-Debug-Upload-Key", required = false) String accessKey,
@PathVariable String runId) {
return runService.getRun(accessKey, runId);
}
/**
* 上传单封 .eml 邮件并以 SSE 实时返回本系统阶段、SuperAgent 公开 Trace 和最终回答。
*/

View File

@@ -100,6 +100,24 @@ public class MybatisDebugEmlSuperAgentRunRepository implements DebugEmlSuperAgen
entity.getHotelId(),
entity.getRunStatus(),
entity.getSourceMessageId(),
entity.getCreatedAt());
entity.getExternalMessageId(),
entity.getExternalConversationId(),
entity.getOriginalFileName(),
entity.getOriginalEmlOssUrl(),
entity.getOriginalEmlSha256(),
entity.getPayloadJson(),
entity.getSuperagentSessionId(),
entity.getSuperagentRunId(),
entity.getSuperagentProfileId(),
entity.getSuperagentProfileVersionId(),
entity.getSuperagentModelName(),
entity.getSuperagentRawAnswer(),
entity.getSuperagentParsedJson(),
entity.getSuperagentInputTokens(),
entity.getSuperagentOutputTokens(),
entity.getSuperagentTotalTokens(),
entity.getSafeErrorSummary(),
entity.getCreatedAt(),
entity.getUpdatedAt());
}
}

View File

@@ -23,6 +23,11 @@ public interface DebugEmlSuperAgentRunService {
String hotelId,
String runLabel);
/**
* 查询 Debug EML 运行安全结果,用于流式响应被中间链路提前关闭后的前端兜底轮询。
*/
DebugEmlSuperAgentRunResult getRun(String accessKey, String runId);
/**
* 上传并处理单封 EML按 text/event-stream 实时写出内部阶段、SuperAgent Trace 和最终结果。
*/

View File

@@ -12,6 +12,7 @@ import cn.nianxx.thhotel.integrations.storage.aliyunoss.service.impl.AliyunOssPr
import cn.nianxx.thhotel.integrations.storage.aliyunoss.service.impl.ObjectStorageException;
import cn.nianxx.thhotel.platform.common.enums.SourceMessageOnlyResultCode;
import cn.nianxx.thhotel.platform.debug.common.dto.DebugEmlSuperAgentRunDraft;
import cn.nianxx.thhotel.platform.debug.common.dto.DebugEmlSuperAgentRunSnapshot;
import cn.nianxx.thhotel.platform.debug.common.dto.DebugEmlSuperAgentRunStatusUpdate;
import cn.nianxx.thhotel.platform.debug.common.dto.DebugEmlSuperAgentRunUpdate;
import cn.nianxx.thhotel.platform.debug.common.enums.DebugEmlSuperAgentRunStatus;
@@ -23,15 +24,19 @@ import cn.nianxx.thhotel.platform.hotel.service.HotelContextService;
import cn.nianxx.thhotel.platform.hotel.service.HotelContextException;
import cn.nianxx.thhotel.platform.message.common.dto.ParsedEmlMediaItem;
import cn.nianxx.thhotel.platform.message.common.dto.ParsedEmlMessage;
import cn.nianxx.thhotel.platform.message.common.dto.SourceMessageOriginalContent;
import cn.nianxx.thhotel.platform.message.common.dto.SourceMessageOriginalMediaItem;
import cn.nianxx.thhotel.platform.message.common.enums.SourceMessageMediaType;
import cn.nianxx.thhotel.platform.message.common.request.CaptureSourceMessageCommand;
import cn.nianxx.thhotel.platform.message.common.request.CaptureSourceMessageMedia;
import cn.nianxx.thhotel.platform.message.common.result.SourceMessageCaptureResult;
import cn.nianxx.thhotel.platform.message.repository.SourceMessageInboxRepository;
import cn.nianxx.thhotel.platform.message.service.EmlMessageParseService;
import cn.nianxx.thhotel.platform.message.service.SourceMessageCaptureService;
import cn.nianxx.thhotel.platform.message.service.SourceMessageHtmlSanitizerService;
import cn.nianxx.thhotel.platform.message.service.impl.EmlMessageParseException;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -97,6 +102,7 @@ public class DebugEmlSuperAgentRunServiceImpl implements DebugEmlSuperAgentRunSe
private final EmlMessageParseService parseService;
private final ObjectStorageService objectStorageService;
private final SourceMessageCaptureService sourceMessageCaptureService;
private final SourceMessageInboxRepository sourceMessageInboxRepository;
private final SourceMessageHtmlSanitizerService htmlSanitizerService;
private final SuperAgentOpenApiClient superAgentOpenApiClient;
private final DebugEmlSuperAgentRunRepository runRepository;
@@ -112,6 +118,7 @@ public class DebugEmlSuperAgentRunServiceImpl implements DebugEmlSuperAgentRunSe
EmlMessageParseService parseService,
ObjectStorageService objectStorageService,
SourceMessageCaptureService sourceMessageCaptureService,
SourceMessageInboxRepository sourceMessageInboxRepository,
SourceMessageHtmlSanitizerService htmlSanitizerService,
SuperAgentOpenApiClient superAgentOpenApiClient,
DebugEmlSuperAgentRunRepository runRepository,
@@ -122,6 +129,7 @@ public class DebugEmlSuperAgentRunServiceImpl implements DebugEmlSuperAgentRunSe
this.parseService = parseService;
this.objectStorageService = objectStorageService;
this.sourceMessageCaptureService = sourceMessageCaptureService;
this.sourceMessageInboxRepository = sourceMessageInboxRepository;
this.htmlSanitizerService = htmlSanitizerService;
this.superAgentOpenApiClient = superAgentOpenApiClient;
this.runRepository = runRepository;
@@ -199,6 +207,21 @@ public class DebugEmlSuperAgentRunServiceImpl implements DebugEmlSuperAgentRunSe
}
}
/**
* 查询 Debug EML 运行安全结果。用于前端 SSE 被中间链路提前关闭后按 runId 兜底轮询。
*/
@Override
public DebugEmlSuperAgentRunResult getRun(String accessKey, String runId) {
validateAccessKey(accessKey);
Long parsedRunId = parseRunId(runId);
DebugEmlSuperAgentRunSnapshot snapshot = runRepository.findById(parsedRunId)
.orElseThrow(() -> new DebugEmlSuperAgentException(
HttpStatus.NOT_FOUND,
"DEBUG_EML_RUN_NOT_FOUND",
"Debug EML 运行记录不存在。"));
return toRunResult(snapshot);
}
/**
* 处理单封 Debug EML 上传,并实时输出安全的调试 SSE 事件。
*/
@@ -571,7 +594,131 @@ public class DebugEmlSuperAgentRunServiceImpl implements DebugEmlSuperAgentRunSe
parsedJson,
traceEvents,
List.copyOf(warnings),
status.name());
status.name(),
null);
}
/**
* 将持久化快照转换为 Debug 页面可安全展示的查询结果。
*/
private DebugEmlSuperAgentRunResult toRunResult(DebugEmlSuperAgentRunSnapshot snapshot) {
SourceMessageOriginalContent originalContent = loadOriginalContent(snapshot.sourceMessageId());
String htmlBodyWithOssUrls = originalContent == null ? null : originalContent.htmlBody();
String htmlBodySanitized = htmlBodyWithOssUrls == null ? null : htmlSanitizerService.sanitizeHtml(htmlBodyWithOssUrls);
return new DebugEmlSuperAgentRunResult(
snapshot.id().toString(),
snapshot.sourceMessageId() == null ? null : snapshot.sourceMessageId().toString(),
snapshot.sourceMessageId() == null ? null : SOURCE_PROVIDER,
snapshot.externalMessageId(),
snapshot.externalConversationId(),
snapshot.originalEmlOssUrl(),
snapshot.originalEmlSha256(),
originalContent == null ? List.of() : debugMediaResults(originalContent.mediaItems()),
htmlBodyWithOssUrls,
htmlBodySanitized,
originalContent == null ? null : true,
originalContent == null ? null : htmlSanitizerService.htmlRenderMode(htmlBodyWithOssUrls),
parsePayloadJson(snapshot.payloadJson()),
snapshot.superagentSessionId(),
snapshot.superagentRunId(),
snapshot.superagentRawAnswer(),
parseJsonNode(snapshot.superagentParsedJson()),
List.of(),
List.of(),
snapshot.runStatus(),
snapshot.safeErrorSummary());
}
/**
* 读取 Debug run 关联 SourceMessage 的受控原文快照;缺失时返回 null查询接口仍返回运行状态。
*/
private SourceMessageOriginalContent loadOriginalContent(Long sourceMessageId) {
if (sourceMessageId == null) {
return null;
}
return sourceMessageInboxRepository.findOriginalContent(sourceMessageId).orElse(null);
}
/**
* 将 SourceMessage 媒体快照转换为 Debug 响应媒体项。
*/
private List<DebugEmlUploadedMediaResult> debugMediaResults(List<SourceMessageOriginalMediaItem> mediaItems) {
if (mediaItems == null || mediaItems.isEmpty()) {
return List.of();
}
return mediaItems.stream()
.map(item -> new DebugEmlUploadedMediaResult(
item.mediaType(),
item.fileName(),
item.contentType(),
item.sizeBytes(),
item.externalUrl(),
item.externalMediaId(),
objectKeyFromPublicUrl(item.externalUrl())))
.toList();
}
/**
* 尽量从公开 URL 还原 OSS object key无法可靠还原时返回 null不影响前端调试展示。
*/
private String objectKeyFromPublicUrl(String externalUrl) {
String normalizedUrl = trimToNull(externalUrl);
String publicBaseUrl = trimTrailingSlash(trimToNull(ossProperties.getPublicBaseUrl()));
if (normalizedUrl == null || publicBaseUrl == null || !normalizedUrl.startsWith(publicBaseUrl + "/")) {
return null;
}
String encodedObjectKey = normalizedUrl.substring(publicBaseUrl.length() + 1);
return URLDecoder.decode(encodedObjectKey, StandardCharsets.UTF_8);
}
/**
* 解析查询路径中的 Debug runId。
*/
private Long parseRunId(String runId) {
String normalizedRunId = trimToNull(runId);
if (normalizedRunId == null) {
throw new DebugEmlSuperAgentException(HttpStatus.BAD_REQUEST, "REQUEST_FIELD_REQUIRED", "debug_run_id 不能为空。");
}
try {
return Long.valueOf(normalizedRunId);
} catch (NumberFormatException exception) {
throw new DebugEmlSuperAgentException(HttpStatus.BAD_REQUEST, "REQUEST_FIELD_INVALID", "debug_run_id 格式错误。");
}
}
/**
* 解析已入库的 payload JSON解析失败返回 null避免查询接口抛出内部异常。
*/
private Map<String, Object> parsePayloadJson(String payloadJson) {
String normalizedPayload = trimToNull(payloadJson);
if (normalizedPayload == null) {
return null;
}
try {
JsonNode jsonNode = objectMapper.readTree(normalizedPayload);
if (jsonNode == null || !jsonNode.isObject()) {
return null;
}
return objectMapper.convertValue(jsonNode, new TypeReference<>() {
});
} catch (Exception exception) {
return null;
}
}
/**
* 解析已入库的 SuperAgent JSON解析失败返回 null前端仍可查看 raw answer。
*/
private JsonNode parseJsonNode(String jsonText) {
String normalizedJson = trimToNull(jsonText);
if (normalizedJson == null) {
return null;
}
try {
return objectMapper.readTree(normalizedJson);
} catch (Exception exception) {
return null;
}
}
/**
@@ -1135,6 +1282,20 @@ public class DebugEmlSuperAgentRunServiceImpl implements DebugEmlSuperAgentRunSe
return trimmed.isEmpty() ? null : trimmed;
}
/**
* 去掉 URL 末尾斜杠,便于用公开基础 URL 还原对象路径。
*/
private String trimTrailingSlash(String value) {
if (value == null) {
return null;
}
String trimmed = value;
while (trimmed.endsWith("/")) {
trimmed = trimmed.substring(0, trimmed.length() - 1);
}
return trimmed;
}
/**
* 截断安全摘要,避免超过数据库限制。
*/

View File

@@ -11,6 +11,10 @@ spring:
# multipart 需要高于 Debug EML 业务文件上限,避免超限文件在进入 Controller 前被框架直接 413 拦截。
max-file-size: ${DEBUG_EML_UPLOAD_MULTIPART_MAX_FILE_BYTES:20971520}
max-request-size: ${DEBUG_EML_UPLOAD_MULTIPART_MAX_REQUEST_BYTES:${DEBUG_EML_UPLOAD_MAX_REQUEST_BYTES:25165824}}
mvc:
async:
# Spring MVC async timeout 是应用级全局值;当前主要用于避免 Debug EML SSE 先于 SuperAgent 调试调用关闭。
request-timeout: ${DEBUG_EML_UPLOAD_SSE_REQUEST_TIMEOUT:${SPRING_MVC_ASYNC_REQUEST_TIMEOUT:1800s}}
mybatis-plus:
configuration:

View File

@@ -89,16 +89,30 @@ class SuperAgentOpenApiSseParserTest {
}
@Test
void shouldFailWhenEndEventMissingEvenIfAiContentExists() {
void shouldFailWhenEndEventMissingAndOnlyPartialAiContentExists() {
String sse = """
event: messages
data: {"type":"ai","content":"partial answer"}
""";
assertThatThrownBy(() -> parser.parse("session-debug-partial", sse))
.isInstanceOf(SuperAgentOpenApiException.class)
.hasMessageContaining("SuperAgent SSE 未收到结束事件。");
}
@Test
void shouldAcceptMissingEndEventWhenFinalAiContentExists() {
String sse = """
event: messages
data: {"type":"ai","content":"{\\"ai_task_results\\":[]}","response_metadata":{"finish_reason":"stop"}}
""";
assertThatThrownBy(() -> parser.parse("session-debug-004", sse))
.isInstanceOf(SuperAgentOpenApiException.class)
.hasMessageContaining("SuperAgent SSE 未收到结束事件。");
SuperAgentOpenApiResult result = parser.parse("session-debug-004", sse);
assertThat(result.rawAnswer()).isEqualTo("{\"ai_task_results\":[]}");
assertThat(result.eventTypes()).containsExactly("messages");
}
@Test

View File

@@ -8,6 +8,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.when;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.asyncDispatch;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.multipart;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath;
@@ -186,6 +187,44 @@ class DebugEmlSuperAgentControllerTest {
org.assertj.core.api.Assertions.assertThat(debugRunCount).isEqualTo(1L);
}
@Test
void shouldQueryDebugRunResultByIdForStreamFallback() throws Exception {
mockStorageAndSuperAgentSuccess();
mockMvc.perform(multipart(ENDPOINT)
.file(emlFile())
.param("hotel_id", "HOTEL-TEST")
.param("run_label", "query-run-fallback")
.header("X-TH-Hotel-Debug-Upload-Key", "test-debug-upload-key"))
.andExpect(status().isCreated());
Long runId = jdbcTemplate.queryForObject("""
SELECT id
FROM platform_debug_eml_superagent_run
WHERE run_label = 'query-run-fallback'
""", Long.class);
mockMvc.perform(get(ENDPOINT + "/" + runId)
.header("X-TH-Hotel-Debug-Upload-Key", "test-debug-upload-key"))
.andExpect(status().isOk())
.andExpect(jsonPath("$.debug_run_id").value(runId.toString()))
.andExpect(jsonPath("$.source_message_id").isNotEmpty())
.andExpect(jsonPath("$.source_provider").value("DEBUG_EML_UPLOAD"))
.andExpect(jsonPath("$.uploaded_media", hasSize(greaterThanOrEqualTo(3))))
.andExpect(jsonPath("$.html_body_with_oss_urls", containsString("https://oss.example.test/")))
.andExpect(jsonPath("$.html_body_with_oss_urls", not(containsString("cid:inline-001"))))
.andExpect(jsonPath("$.html_body_sanitized", containsString("https://oss.example.test/")))
.andExpect(jsonPath("$.html_sanitize_required").value(true))
.andExpect(jsonPath("$.html_render_mode").value("SANITIZED_HTML"))
.andExpect(jsonPath("$.superagent_session_id").value("session-debug-001"))
.andExpect(jsonPath("$.superagent_run_id").value("run-debug-001"))
.andExpect(jsonPath("$.superagent_raw_answer", containsString("ai_task_results")))
.andExpect(jsonPath("$.superagent_parsed_json.ai_task_results[0].task_type").value("New Booking"))
.andExpect(jsonPath("$.status").value("SUPERAGENT_SUCCEEDED"))
.andExpect(jsonPath("$.safe_error_summary").doesNotExist())
.andExpect(content().string(not(containsString("test-debug-upload-key"))));
}
@Test
void shouldStreamDebugStagesSuperAgentTraceAndFinalResult() throws Exception {
when(objectStorageService.putObject(any())).thenAnswer(invocation -> {