Harden ingress retry flow and persist kernel failures

.tmp/case2_reply_242.html

@@ -0,0 +1,31 @@
+<!doctype html>
+<html lang="zh-CN">
+<head>
+  <meta charset="UTF-8" />
+  <meta name="viewport" content="width=device-width, initial-scale=1" />
+  <title>Case2 Feishu Reply</title>
+  <style>
+    body { font-family: -apple-system,BlinkMacSystemFont,"Segoe UI",Roboto,"PingFang SC","Hiragino Sans GB","Microsoft YaHei",sans-serif; background:#f5f7fb; margin:0; padding:24px; }
+    .card { max-width: 980px; margin: 0 auto; background: #fff; border: 1px solid #e5eaf3; border-radius: 12px; padding: 20px; box-shadow: 0 4px 18px rgba(19,33,68,.06); }
+    h1 { margin: 0 0 10px; font-size: 20px; }
+    .meta { color:#4b5b7c; font-size:14px; margin-bottom: 12px; }
+    pre { white-space: pre-wrap; word-break: break-word; line-height: 1.6; background:#f8faff; border:1px solid #e7edf8; border-radius:8px; padding:14px; font-size:14px; }
+    .flag { margin-top: 10px; font-size: 14px; }
+    .bad { color:#b42318; font-weight:600; }
+  </style>
+</head>
+<body>
+  <div class="card">
+    <h1>Case2 真实飞书回复</h1>
+    <div class="meta">source_message_id: real-scene-case2-1773913396 · outbox_id: 242 · run_id: 91</div>
+    <pre>当前未接入实时数据，以下基于训练数据（截止2024年）。
+
+[行业动态]
+近期AI行业呈现以下核心趋势：
+1. **大模型竞争白热化**：OpenAI、Anthropic、Google等厂商持续迭代模型（如GPT-4o、Claude 3.5），上下文窗口与推理能力显著提升，价格战促使API成本下降。
+2. **Agent（智能体）爆发**：行业重心从单一对话转向具备自主规划、工具调用能力的Agent，自动化解决复杂工作流成为热点。
+3. **垂直领域落地加速**：医疗、法律、HR等专业场景的AI应用从“尝鲜”转向“实效”，</pre>
+    <div class="flag">[建议] 段检查：<span class="bad">未出现</span></div>
+  </div>
+</body>
+</html>

.tmp/case2_reply_242.txt

@@ -0,0 +1,7 @@
+当前未接入实时数据，以下基于训练数据（截止2024年）。
+
+[行业动态]
+近期AI行业呈现以下核心趋势：
+1. **大模型竞争白热化**：OpenAI、Anthropic、Google等厂商持续迭代模型（如GPT-4o、Claude 3.5），上下文窗口与推理能力显著提升，价格战促使API成本下降。
+2. **Agent（智能体）爆发**：行业重心从单一对话转向具备自主规划、工具调用能力的Agent，自动化解决复杂工作流成为热点。
+3. **垂直领域落地加速**：医疗、法律、HR等专业场景的AI应用从“尝鲜”转向“实效”，

.tmp/case2_reply_242_data_url.txt

@@ -0,0 +1 @@
+data:text/html;charset=utf-8,%3C%21doctype%20html%3E%3Cmeta%20charset%3D%22utf-8%22%3E%3Cstyle%3Ebody%7Bfont-family%3AArial%2Csans-serif%3Bpadding%3A20px%7Dpre%7Bwhite-space%3Apre-wrap%3Bbackground%3A%23f5f5f5%3Bpadding%3A12px%3Bborder-radius%3A8px%7D.bad%7Bcolor%3A%23b42318%3Bfont-weight%3A700%7D%3C/style%3E%3Ch2%3ECase2%20%E7%9C%9F%E5%AE%9E%E9%A3%9E%E4%B9%A6%E5%9B%9E%E5%A4%8D%3C/h2%3E%3Cdiv%3Esource_message_id%3A%20real-scene-case2-1773913396%3C/div%3E%3Cpre%3E%E5%BD%93%E5%89%8D%E6%9C%AA%E6%8E%A5%E5%85%A5%E5%AE%9E%E6%97%B6%E6%95%B0%E6%8D%AE%EF%BC%8C%E4%BB%A5%E4%B8%8B%E5%9F%BA%E4%BA%8E%E8%AE%AD%E7%BB%83%E6%95%B0%E6%8D%AE%EF%BC%88%E6%88%AA%E6%AD%A22024%E5%B9%B4%EF%BC%89%E3%80%82%0A%0A%5B%E8%A1%8C%E4%B8%9A%E5%8A%A8%E6%80%81%5D%0A%E8%BF%91%E6%9C%9FAI%E8%A1%8C%E4%B8%9A%E5%91%88%E7%8E%B0%E4%BB%A5%E4%B8%8B%E6%A0%B8%E5%BF%83%E8%B6%8B%E5%8A%BF%EF%BC%9A%0A1.%20%2A%2A%E5%A4%A7%E6%A8%A1%E5%9E%8B%E7%AB%9E%E4%BA%89%E7%99%BD%E7%83%AD%E5%8C%96%2A%2A%EF%BC%9AOpenAI%E3%80%81Anthropic%E3%80%81Google%E7%AD%89%E5%8E%82%E5%95%86%E6%8C%81%E7%BB%AD%E8%BF%AD%E4%BB%A3%E6%A8%A1%E5%9E%8B%EF%BC%88%E5%A6%82GPT-4o%E3%80%81Claude%203.5%EF%BC%89%EF%BC%8C%E4%B8%8A%E4%B8%8B%E6%96%87%E7%AA%97%E5%8F%A3%E4%B8%8E%E6%8E%A8%E7%90%86%E8%83%BD%E5%8A%9B%E6%98%BE%E8%91%97%E6%8F%90%E5%8D%87%EF%BC%8C%E4%BB%B7%E6%A0%BC%E6%88%98%E4%BF%83%E4%BD%BFAPI%E6%88%90%E6%9C%AC%E4%B8%8B%E9%99%8D%E3%80%82%0A2.%20%2A%2AAgent%EF%BC%88%E6%99%BA%E8%83%BD%E4%BD%93%EF%BC%89%E7%88%86%E5%8F%91%2A%2A%EF%BC%9A%E8%A1%8C%E4%B8%9A%E9%87%8D%E5%BF%83%E4%BB%8E%E5%8D%95%E4%B8%80%E5%AF%B9%E8%AF%9D%E8%BD%AC%E5%90%91%E5%85%B7%E5%A4%87%E8%87%AA%E4%B8%BB%E8%A7%84%E5%88%92%E3%80%81%E5%B7%A5%E5%85%B7%E8%B0%83%E7%94%A8%E8%83%BD%E5%8A%9B%E7%9A%84Agent%EF%BC%8C%E8%87%AA%E5%8A%A8%E5%8C%96%E8%A7%A3%E5%86%B3%E5%A4%8D%E6%9D%82%E5%B7%A5%E4%BD%9C%E6%B5%81%E6%88%90%E4%B8%BA%E7%83%AD%E7%82%B9%E3%80%82%0A3.%20%2A%2A%E5%9E%82%E7%9B%B4%E9%A2%86%E5%9F%9F%E8%90%BD%E5%9C%B0%E5%8A%A0%E9%80%9F%2A%2A%EF%BC%9A%E5%8C%BB%E7%96%97%E3%80%81%E6%B3%95%E5%BE%8B%E3%80%81HR%E7%AD%89%E4%B8%93%E4%B8%9A%E5%9C%BA%E6%99%AF%E7%9A%84AI%E5%BA%94%E7%94%A8%E4%BB%8E%E2%80%9C%E5%B0%9D%E9%B2%9C%E2%80%9D%E8%BD%AC%E5%90%91%E2%80%9C%E5%AE%9E%E6%95%88%E2%80%9D%EF%BC%8C%0A%3C/pre%3E%3Cdiv%3E%5B%E5%BB%BA%E8%AE%AE%5D%20%E6%AE%B5%E6%A3%80%E6%9F%A5%EF%BC%9A%3Cspan%20class%3D%22bad%22%3E%E6%9C%AA%E5%87%BA%E7%8E%B0%3C/span%3E%3C/div%3E

.tmp/case3_v9_result.html

@@ -0,0 +1,28 @@
+<!doctype html>
+<html lang="zh-CN">
+<head>
+  <meta charset="UTF-8" />
+  <title>Case3 验收结果</title>
+  <style>
+    body { font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", "PingFang SC", sans-serif; margin: 24px; background: #f5f7fb; }
+    .card { background: #fff; border: 1px solid #d9e1ef; border-radius: 12px; padding: 20px; max-width: 980px; }
+    h1 { margin: 0 0 12px; font-size: 22px; }
+    .meta { color: #344054; margin: 6px 0; }
+    .fail { color: #b42318; font-weight: 600; }
+    .reply { margin-top: 12px; padding: 14px; border: 1px dashed #98a2b3; border-radius: 8px; background: #f8fafc; white-space: pre-wrap; }
+    code { background: #eef2ff; padding: 2px 6px; border-radius: 6px; }
+  </style>
+</head>
+<body>
+  <div class="card">
+    <h1>Case3 验收截图（v9）</h1>
+    <div class="meta">source_message_id: <code>real-scene-p1-v9-3-1773884746</code></div>
+    <div class="meta">run_id: <code>89</code></div>
+    <div class="meta">当前 run 状态: <code>failed</code>（retry_count=3/max_retries=3）</div>
+    <div class="meta fail">验收结果: 不通过（未返回业务答案）</div>
+    <div class="meta">飞书实际回复（outbox）:</div>
+    <div class="reply">1) 已收到，正在处理。
+2) 已自动重试3次仍失败，请重新发送或换个说法。</div>
+  </div>
+</body>
+</html>

backend/app/config/ai_prompts_catalog/part_scenes_06.py

@@ -15,6 +15,7 @@ PROMPT_EXTERNAL_TREND_QA = """
    [行业动态] xxx
    [对项目的影响] xxx
    [建议] xxx
+6. 控制总回复长度不超过800字，优先保留[建议]段
 
 问题：{question}
 项目上下文：{project_context}

backend/app/services/ai_orchestration_run/ingress_worker.py

@@ -18,6 +18,7 @@ logger = get_logger(__name__)
 
 FINAL_FAILURE_TEXT = "已自动重试3次仍失败，请重新发送或换个说法。"
 INGRESS_HEARTBEAT_INTERVAL_SECONDS = 15
+FINAL_REPLY_MAX_CHARS = 800
 
 
 class IngressDispatchError(RuntimeError):
@@ -46,7 +47,7 @@ def _build_final_reply_text(result: dict[str, Any]) -> str:
         return ""
     direct = str(result.get("direct_reply_text") or "").strip()
     if direct:
-        return direct[:500]
+        return direct[:FINAL_REPLY_MAX_CHARS]
     status = str(result.get("status") or "").strip().lower()
     if status in {"captured", "answered"}:
         return "处理完成。"

backend/app/services/ai_orchestration_run/service.py

@@ -222,7 +222,15 @@ class AIOrchestrationRunService:
                         AIOrchestrationRun.run_type == INGRESS_RUN_TYPE,
                         or_(
                             and_(
-                                AIOrchestrationRun.status.in_(("queued", "retrying")),
+                                AIOrchestrationRun.status == "queued",
+                                or_(
+                                    AIOrchestrationRun.next_retry_at.is_(None),
+                                    AIOrchestrationRun.next_retry_at <= now_at,
+                                ),
+                            ),
+                            and_(
+                                AIOrchestrationRun.status == "retrying",
+                                AIOrchestrationRun.retry_count < AIOrchestrationRun.max_retries,
                                 or_(
                                     AIOrchestrationRun.next_retry_at.is_(None),
                                     AIOrchestrationRun.next_retry_at <= now_at,

backend/app/services/ai_pmo/ai_pmo_kernel_ingress_flow/mixin_auto.py

@@ -147,7 +147,17 @@ class KernelIngressFlowAutoMixin:
             last_intent=final_intent_name,
             last_message_text=message_text[:200],
         )
+        try:
             await self.db.commit()
+        except Exception as exc:
+            await self._persist_kernel_failure_with_fresh_session(
+                work_item_id=_safe_int(getattr(work_item, "id", None)),
+                ingest_event_id=_safe_int(getattr(ingest_event, "id", None)),
+                error_message=str(exc),
+                source_system=source_system,
+                source_message_id=source_message_id,
+            )
+            raise
         execution_route_key = RouteResolver.normalize_route_key(
             str((execution_result or {}).get("workflow_route") or "")
         ) if isinstance(execution_result, dict) else ""

backend/app/services/ai_pmo/ai_pmo_kernel_ingress_flow/mixin_entry.py

@@ -349,7 +349,13 @@ class KernelIngressFlowEntryMixin:
                 log_perf_fn=_log_perf,
             )
         except Exception as exc:
-            await self.db.rollback()
+            await self._persist_kernel_failure_with_fresh_session(
+                work_item_id=_safe_int(getattr(work_item, "id", None)),
+                ingest_event_id=_safe_int(getattr(ingest_event, "id", None)),
+                error_message=str(exc),
+                source_system=source_system,
+                source_message_id=synthetic_message_id,
+            )
             logger.warning(
                 "ai_pmo_kernel_failed",
                 tenant_id=self.tenant_id,
@@ -358,3 +364,68 @@ class KernelIngressFlowEntryMixin:
                 error=str(exc),
             )
             return {"handled": False, "reason": "kernel_exception", "error": str(exc)}
+
+    async def _persist_kernel_failure_with_fresh_session(
+        self,
+        *,
+        work_item_id: Optional[int],
+        ingest_event_id: Optional[int],
+        error_message: str,
+        source_system: str,
+        source_message_id: str,
+    ) -> None:
+        session_factory = self._background_session_factory
+        if session_factory is None:
+            logger.warning(
+                "ai_pmo_kernel_failure_session_factory_missing",
+                tenant_id=self.tenant_id,
+                source_system=source_system,
+                source_message_id=source_message_id,
+            )
+            return
+        try:
+            async with session_factory() as failure_db:
+                now_at = _utc_now_naive()
+                if ingest_event_id is not None:
+                    ingest_row = (
+                        await failure_db.execute(
+                            select(IngestEvent).where(
+                                and_(
+                                    IngestEvent.tenant_id == self.tenant_id,
+                                    IngestEvent.id == int(ingest_event_id),
+                                )
+                            ).limit(1)
+                        )
+                    ).scalar_one_or_none()
+                    if ingest_row is not None:
+                        ingest_row.status = "failed"
+                        ingest_row.error_message = str(error_message or "")[:500]
+                        ingest_row.processed_at = now_at
+
+                if work_item_id is not None:
+                    work_item_row = (
+                        await failure_db.execute(
+                            select(WorkItem).where(
+                                and_(
+                                    WorkItem.tenant_id == self.tenant_id,
+                                    WorkItem.id == int(work_item_id),
+                                )
+                            ).limit(1)
+                        )
+                    ).scalar_one_or_none()
+                    if work_item_row is not None and str(work_item_row.state or "") not in {
+                        STATE_SUCCEEDED,
+                        STATE_WAITING_MANUAL,
+                    }:
+                        work_item_row.state = STATE_FAILED
+                        if not str(work_item_row.intent or "").strip():
+                            work_item_row.intent = "kernel_exception"
+                await failure_db.commit()
+        except Exception as persist_exc:
+            logger.warning(
+                "ai_pmo_kernel_failure_persist_failed",
+                tenant_id=self.tenant_id,
+                source_system=source_system,
+                source_message_id=source_message_id,
+                error=str(persist_exc),
+            )

backend/app/services/ai_pmo/ai_pmo_kernel_ingress_flow/mixin_prepare_manual.py

@@ -240,7 +240,17 @@ class KernelIngressFlowPrepareManualMixin:
             last_intent=intent.name,
             last_message_text=message_text[:200],
         )
+        try:
             await self.db.commit()
+        except Exception as exc:
+            await self._persist_kernel_failure_with_fresh_session(
+                work_item_id=_safe_int(getattr(work_item, "id", None)),
+                ingest_event_id=_safe_int(getattr(ingest_event, "id", None)),
+                error_message=str(exc),
+                source_system=source_system,
+                source_message_id=source_message_id,
+            )
+            raise
         return self._build_capture_result(
             work_item=work_item,
             intent=intent,
@@ -250,4 +260,3 @@ class KernelIngressFlowPrepareManualMixin:
             status="captured",
             classification="command",
         )
-

backend/app/services/integrations/feishu_integration_service/mixin_reply_outbox.py

@@ -284,7 +284,7 @@ class FeishuReplyOutboxMixin:
         text = cls._clean_reply_text(value)
         if (not text) or cls._looks_like_provider_error(text):
             return cls._build_error_reply_text()
-        return text[:500]
+        return text[:800]
 
     @staticmethod
     def _build_reply_hash(*, source_message_id: Optional[str], message_text: str) -> Optional[str]:

backend/app/services/signal/signal_capture_service/mixin_capture_entry.py

@@ -147,14 +147,7 @@ class SignalCaptureEntryMixin:
             return pipeline_result
         except Exception as exc:
             self._emit_ai_think("capture_kernel_failed", trace_id=trace_id, error=str(exc))
-            try:
-                await self.db.rollback()
-            except Exception as rollback_exc:
-                self._emit_ai_think(
-                    "capture_kernel_failed_rollback_failed",
-                    trace_id=trace_id,
-                    error=str(rollback_exc),
-                )
+            await self._rollback_capture_failure_with_fresh_session(trace_id=trace_id)
             return {
                 "status": "error",
                 "reason": "kernel_failed",
@@ -264,6 +257,18 @@ class SignalCaptureEntryMixin:
             pass
         return pipeline_result
 
+    async def _rollback_capture_failure_with_fresh_session(self, *, trace_id: str) -> None:
+        # Failure path must be isolated from the main session lifecycle.
+        try:
+            async with AsyncSessionLocal() as failure_db:
+                await failure_db.rollback()
+        except Exception as rollback_exc:
+            self._emit_ai_think(
+                "capture_kernel_failed_rollback_failed",
+                trace_id=trace_id,
+                error=str(rollback_exc),
+            )
+
     @staticmethod
     def _build_kernel_error_result(*, trace_id: str, reason: str) -> Dict[str, Any]:
         return {