fix: 修复 file_path 始终为 null 的问题

backend/alembic/versions/009_add_verdict_to_agent_findings.py

@@ -0,0 +1,24 @@
+"""Add verdict column to agent_findings
+
+Revision ID: 009_add_verdict_to_agent_findings
+Revises: 008_add_files_with_findings
+Create Date: 2026-05-04
+
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision = '009_add_verdict_to_agent_findings'
+down_revision = '008_add_files_with_findings'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column('agent_findings', sa.Column('verdict', sa.String(length=30), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column('agent_findings', 'verdict')

backend/app/api/v1/endpoints/agent_tasks.py

@@ -194,6 +194,7 @@ class AgentFindingResponse(BaseModel):
     status: str
 
     suggestion: Optional[str] = None
+    verdict: Optional[str] = None
     poc: Optional[dict] = None
 
     created_at: datetime
@@ -1264,9 +1265,17 @@ async def _save_findings(
             file_path = (
                 finding.get("file_path") or
                 finding.get("file") or
-                finding.get("location", "").split(":")[0] if ":" in finding.get("location", "") else finding.get("location")
+                (finding.get("location", "").split(":")[0] if ":" in finding.get("location", "") else finding.get("location", ""))
             )
 
+            # 🔥 v2.2: file_path 为空直接跳过
+            if not file_path or not file_path.strip():
+                logger.warning(
+                    f"[SaveFindings] 🚫 跳过无 file_path 的 finding: "
+                    f"title={finding.get('title', 'N/A')[:50]}, type={finding.get('vulnerability_type', '?')}"
+                )
+                continue
+
             # 🔥 v2.1: 文件路径验证 - 过滤幻觉发现
             if project_root and file_path:
                 # 清理路径（移除可能的行号）
@@ -1336,7 +1345,8 @@ async def _save_findings(
 
             # 🔥 Handle verification status
             is_verified = finding.get("is_verified", False)
-            if finding.get("verdict") == "confirmed":
+            verdict = finding.get("verdict")  # confirmed / likely / uncertain / false_positive
+            if verdict == "confirmed":
                 is_verified = True
 
             # 🔥 Handle PoC information
@@ -1381,6 +1391,7 @@ async def _save_findings(
                 code_snippet=code_snippet[:10000] if code_snippet else None,
                 suggestion=suggestion[:5000] if suggestion else None,
                 is_verified=is_verified,
+                verdict=verdict,  # 🔥 新增：保存 verdict 到数据库
                 ai_confidence=confidence,  # 🔥 FIX: Use ai_confidence, not confidence
                 status=FindingStatus.VERIFIED if is_verified else FindingStatus.NEW,
                 # 🔥 Additional fields

backend/app/models/agent_task.py

@@ -355,6 +355,7 @@ class AgentFinding(Base):
 
     # 验证信息
     status = Column(String(30), default=FindingStatus.NEW, index=True)
+    verdict = Column(String(30), nullable=True, index=True)  # confirmed / likely / uncertain / false_positive
     is_verified = Column(Boolean, default=False)
     verification_method = Column(Text, nullable=True)
     verification_result = Column(JSON, nullable=True)

backend/app/services/agent/agents/analysis.py

@@ -759,18 +759,33 @@ async def run(self, input_data: Dict[str, Any]) -> AgentResult:
             # 标准化发现
             logger.info(f"[{self.name}] Standardizing {len(all_findings)} findings")
             standardized_findings = []
+            skipped_no_filepath = 0
             for finding in all_findings:
                 # 确保 finding 是字典
                 if not isinstance(finding, dict):
                     logger.warning(f"Skipping invalid finding (not a dict): {finding}")
                     continue
+
+                # 🔥 v2.2: file_path 必填校验 - 没有 file_path 的 finding 直接拒绝
+                # 优先从 file_path 获取，fallback 到 file / location
+                file_path = finding.get("file_path") or finding.get("file") or ""
+                if not file_path.strip() and finding.get("location"):
+                    loc = finding.get("location", "")
+                    file_path = loc.split(":")[0] if ":" in loc else loc
+                if not file_path.strip():
+                    skipped_no_filepath += 1
+                    logger.warning(
+                        f"[Analysis] 🚫 跳过无 file_path 的 finding: "
+                        f"title={finding.get('title', '?')[:50]}, type={finding.get('vulnerability_type', '?')}"
+                    )
+                    continue
 
                 standardized = {
                     "vulnerability_type": finding.get("vulnerability_type", "other"),
                     "severity": finding.get("severity", "medium"),
                     "title": finding.get("title", "Unknown Finding"),
                     "description": finding.get("description", ""),
-                    "file_path": finding.get("file_path", ""),
+                    "file_path": file_path,
                     "line_start": finding.get("line_start") or finding.get("line", 0),
                     "code_snippet": finding.get("code_snippet", ""),
                     "source": finding.get("source", ""),
@@ -781,6 +796,12 @@ async def run(self, input_data: Dict[str, Any]) -> AgentResult:
                 }
                 standardized_findings.append(standardized)
 
+            if skipped_no_filepath > 0:
+                logger.warning(
+                    f"[Analysis] ⚠️ 跳过了 {skipped_no_filepath} 个无 file_path 的 findings，"
+                    f"保留 {len(standardized_findings)} 个有效 findings"
+                )
+
             await self.emit_event(
                 "info",
                 f"Analysis Agent 完成: {len(standardized_findings)} 个发现, {self._iteration} 轮迭代, {self._tool_calls} 次工具调用"

backend/app/services/agent/agents/orchestrator.py

@@ -878,10 +878,11 @@ async def run_with_cancel_check():
                                 potential_file = parts[0].strip()
                                 # 只有当 parts[0] 看起来像文件路径时才提取
                                 # 文件路径通常包含 . 且没有空格（或只在结尾有扩展名）
+                                # 🔥 FIX: 放宽文件路径校验 - 不再限制扩展名，只要像文件路径就提取
                                 if ("." in potential_file and
                                     " " not in potential_file and
                                     len(potential_file) < 100 and
-                                    any(potential_file.endswith(ext) for ext in ['.py', '.js', '.ts', '.java', '.go', '.php', '.rb', '.c', '.cpp', '.h'])):
+                                    not potential_file.endswith("/")):
                                     file_path = potential_file
                                     # 尝试提取行号
                                     if len(parts) > 1:
@@ -962,7 +963,28 @@ async def run_with_cancel_check():
                                 (new_type in existing_type) or (existing_type in new_type)
                             )
 
+                            # 🔥 Match criteria for same-file findings
                             if same_file and (same_line or similar_desc or same_type):
+                                match_found = True
+                            elif same_type and same_line and not same_file:
+                                # 🔥 FIX: Only allow cross-file matching when:
+                                # 1. new_file is garbage ("?"/empty) - verification returned bad path
+                                # 2. One path is a prefix of the other ("src/foo.py" vs "foo.py")
+                                # Do NOT merge when existing_file is garbage - that would lose the real path.
+                                new_is_garbage = not new_file or new_file == "?"
+                                prefix_match = (
+                                    new_file.endswith("/" + existing_file) or
+                                    existing_file.endswith("/" + new_file)
+                                )
+                                if new_is_garbage or prefix_match:
+                                    match_found = True
+                                    logger.info(f"[Orchestrator] Matched by type+line despite file mismatch: {new_file} vs {existing_file}")
+                                else:
+                                    match_found = False
+                            else:
+                                match_found = False
+
+                            if match_found:
                                 # Update existing with new info (e.g. verification results)
                                 # 🔥 FIX: Smart merge - don't overwrite good data with empty values
                                 merged = dict(existing_f)  # Start with existing data