Add files via upload

2026-08-02 00:48:55 +02:00 · 2026-07-28 18:25:11 +08:00
parent 4a19620137
commit 0324b41a01
11 changed files with 677 additions and 99 deletions
@@ -10,6 +10,7 @@ import (
 	"sync"
 	"time"

+	"cyberstrike-ai/internal/agentfinalizer"
 	"cyberstrike-ai/internal/config"
 	"cyberstrike-ai/internal/mcp"
 	"cyberstrike-ai/internal/multiagent"
@@ -197,6 +198,13 @@ func (h *AgentHandler) MultiAgentLoopStream(c *gin.Context) {
 	// 同一请求内分段续跑时，主代理 iteration 事件按偏移累计，避免 UI 出现「第3轮 → 第1轮」回跳。
 	var mainIterationOffset int
 	var emptyResponseContinueAttempt int
+	var finalizationAutoContinueAttempt int
+	effectiveOrch := config.NormalizeMultiAgentOrchestration(h.config.MultiAgent.Orchestration)
+	if o := strings.TrimSpace(req.Orchestration); o != "" {
+		effectiveOrch = config.NormalizeMultiAgentOrchestration(o)
+	}
+	agentMode := "eino_" + effectiveOrch
+	var decision agentfinalizer.Decision

 	for {
 		segmentMainIterationMax := 0
@@ -267,6 +275,13 @@ func (h *AgentHandler) MultiAgentLoopStream(c *gin.Context) {
 				baseCtx, cancelWithCause, taskCtx, timeoutCancel = h.rebindEinoRunningTask(taskCtx, conversationID, timeoutCancel)
 				continue
 			}
+			decision = h.decideAgentRunForDeliveryWithPolicy(conversationID, assistantMessageID, agentMode, result, cumulativeMCPExecutionIDs, requestRequiresExecutionEvidence(&req))
+			if h.tryAutoContinueAfterFinalization(taskCtx, conversationID, result, decision, &finalizationAutoContinueAttempt, &curHistory, &curFinalMessage, progressCallback) {
+				mainIterationOffset += segmentMainIterationMax
+				timeoutCancel()
+				baseCtx, cancelWithCause, taskCtx, timeoutCancel = h.rebindEinoRunningTask(taskCtx, conversationID, timeoutCancel)
+				continue
+			}
 			timeoutCancel()
 			break
 		}
@@ -367,9 +382,10 @@ func (h *AgentHandler) MultiAgentLoopStream(c *gin.Context) {

 	timeoutCancel()

-	if assistantMessageID != "" {
-		_ = h.db.UpdateAssistantMessageFinalize(assistantMessageID, result.Response, cumulativeMCPExecutionIDs, multiagent.AggregatedReasoningFromTraceJSON(result.LastAgentTraceInput))
+	if decision.CompletionReason == "" {
+		decision = h.decideAgentRunForDeliveryWithPolicy(conversationID, assistantMessageID, agentMode, result, cumulativeMCPExecutionIDs, requestRequiresExecutionEvidence(&req))
 	}
+	h.persistFinalizationDecision(conversationID, assistantMessageID, agentMode, cumulativeMCPExecutionIDs, multiagent.AggregatedReasoningFromTraceJSON(result.LastAgentTraceInput), decision)

 	if result.LastAgentTraceInput != "" || result.LastAgentTraceOutput != "" {
 		if err := h.db.SaveAgentTrace(conversationID, result.LastAgentTraceInput, result.LastAgentTraceOutput); err != nil {
@@ -377,16 +393,19 @@ func (h *AgentHandler) MultiAgentLoopStream(c *gin.Context) {
 		}
 	}

-	effectiveOrch := config.NormalizeMultiAgentOrchestration(h.config.MultiAgent.Orchestration)
-	if o := strings.TrimSpace(req.Orchestration); o != "" {
-		effectiveOrch = config.NormalizeMultiAgentOrchestration(o)
+	responseText := decision.FinalText
+	if !decision.Finalizable {
+		responseText = finalizationBlockedMessage(decision)
+		sendEvent("finalization_check", responseText, decision)
+		taskStatus = decision.Status
+		h.tasks.UpdateTaskStatus(conversationID, taskStatus)
 	}
-	sendEvent("response", result.Response, map[string]interface{}{
+	sendEvent("response", responseText, finalizationResponsePayload(decision, map[string]interface{}{
 		"mcpExecutionIds": cumulativeMCPExecutionIDs,
 		"conversationId":  conversationID,
 		"messageId":       assistantMessageID,
-		"agentMode":       "eino_" + effectiveOrch,
-	})
+		"agentMode":       agentMode,
+	}))
 	sendEvent("done", "", map[string]interface{}{"conversationId": conversationID})
 }

@@ -437,6 +456,14 @@ func (h *AgentHandler) MultiAgentLoop(c *gin.Context) {
 	curMsg := prep.FinalMessage
 	var result *multiagent.RunResult
 	var runErr error
+	var emptyResponseContinueAttempt int
+	var finalizationAutoContinueAttempt int
+	effectiveOrch := config.NormalizeMultiAgentOrchestration(h.config.MultiAgent.Orchestration)
+	if o := strings.TrimSpace(req.Orchestration); o != "" {
+		effectiveOrch = config.NormalizeMultiAgentOrchestration(o)
+	}
+	agentMode := "eino_" + effectiveOrch
+	var decision agentfinalizer.Decision
 	for {
 		result, runErr = multiagent.RunDeepAgent(
 			taskCtx,
@@ -456,24 +483,30 @@ func (h *AgentHandler) MultiAgentLoop(c *gin.Context) {
 			chatReasoningToClientIntent(req.Reasoning),
 			h.agentSessionContextBlock(prep.ConversationID),
 		)
-		if runErr == nil {
-			break
+		if runErr != nil {
+			if shouldPersistEinoAgentTraceAfterRunError(baseCtx) {
+				h.persistEinoAgentTraceForResume(prep.ConversationID, result)
+			}
+			h.logger.Error("Eino DeepAgent 执行失败", zap.Error(runErr))
+			errMsg := "执行失败: " + runErr.Error()
+			if prep.AssistantMessageID != "" {
+				_, _ = h.db.Exec("UPDATE messages SET content = ?, updated_at = ? WHERE id = ?", errMsg, time.Now(), prep.AssistantMessageID)
+			}
+			c.JSON(http.StatusInternalServerError, gin.H{"error": errMsg})
+			return
 		}
-		if shouldPersistEinoAgentTraceAfterRunError(baseCtx) {
-			h.persistEinoAgentTraceForResume(prep.ConversationID, result)
+		mw := &h.config.MultiAgent.EinoMiddleware
+		if h.tryContinueOnEinoEmptyResponse(taskCtx, mw, prep.ConversationID, result, &emptyResponseContinueAttempt, &curHist, &curMsg, progressCallback) {
+			continue
 		}
-		h.logger.Error("Eino DeepAgent 执行失败", zap.Error(runErr))
-		errMsg := "执行失败: " + runErr.Error()
-		if prep.AssistantMessageID != "" {
-			_, _ = h.db.Exec("UPDATE messages SET content = ?, updated_at = ? WHERE id = ?", errMsg, time.Now(), prep.AssistantMessageID)
+		decision = h.decideAgentRunForDeliveryWithPolicy(prep.ConversationID, prep.AssistantMessageID, agentMode, result, result.MCPExecutionIDs, requestRequiresExecutionEvidence(&req))
+		if h.tryAutoContinueAfterFinalization(taskCtx, prep.ConversationID, result, decision, &finalizationAutoContinueAttempt, &curHist, &curMsg, progressCallback) {
+			continue
 		}
-		c.JSON(http.StatusInternalServerError, gin.H{"error": errMsg})
-		return
+		break
 	}

-	if prep.AssistantMessageID != "" {
-		_ = h.db.UpdateAssistantMessageFinalize(prep.AssistantMessageID, result.Response, result.MCPExecutionIDs, multiagent.AggregatedReasoningFromTraceJSON(result.LastAgentTraceInput))
-	}
+	h.persistFinalizationDecision(prep.ConversationID, prep.AssistantMessageID, agentMode, result.MCPExecutionIDs, multiagent.AggregatedReasoningFromTraceJSON(result.LastAgentTraceInput), decision)

 	if result.LastAgentTraceInput != "" || result.LastAgentTraceOutput != "" {
 		if err := h.db.SaveAgentTrace(prep.ConversationID, result.LastAgentTraceInput, result.LastAgentTraceOutput); err != nil {
@@ -481,11 +514,23 @@ func (h *AgentHandler) MultiAgentLoop(c *gin.Context) {
 		}
 	}

+	responseText := decision.FinalText
+	if !decision.Finalizable {
+		responseText = finalizationBlockedMessage(decision)
+	}
 	c.JSON(http.StatusOK, ChatResponse{
-		Response:        result.Response,
-		MCPExecutionIDs: result.MCPExecutionIDs,
-		ConversationID:  prep.ConversationID,
-		Time:            time.Now(),
+		Response:            responseText,
+		MCPExecutionIDs:     result.MCPExecutionIDs,
+		ConversationID:      prep.ConversationID,
+		Time:                time.Now(),
+		Finalizable:         decision.Finalizable,
+		Finalized:           decision.Finalized,
+		Status:              decision.Status,
+		CompletionReason:    decision.CompletionReason,
+		EvidenceVerified:    decision.EvidenceVerified,
+		EvidenceRefs:        decision.EvidenceRefs,
+		PendingExecutionIDs: decision.PendingExecutionIDs,
+		MissingChecks:       decision.MissingChecks,
 	})
 }