Fix P0: llm log redaction (remove full prompt/response from logs) and session.py thread-safe atomic writes

agent/nodes.py

@@ -1266,9 +1266,9 @@ def _check_ocr_fidelity(jrxml: str, state: dict) -> dict:
 
     issues = []
 
-    # 1. 元素数量对比
-    text_fields = len(re.findall(r"<textField", jrxml))
-    static_texts = len(re.findall(r"<staticText", jrxml))
+    # 1. 元素数量对比（支持 namespace 前缀，如 <jrxml:textField>）
+    text_fields = len(re.findall(r"<[a-zA-Z0-9_-]+:textField|<textField", jrxml))
+    static_texts = len(re.findall(r"<[a-zA-Z0-9_-]+:staticText|<staticText", jrxml))
     total_jrxml_elements = text_fields + static_texts
 
     ocr_text_count = 0
@@ -1288,7 +1288,9 @@ def _check_ocr_fidelity(jrxml: str, state: dict) -> dict:
         element_coverage = 1.0
 
     # 2. 字段名覆盖（英文字段名 vs OCR 中文字段名天然不匹配，权重降低）
-    jrxml_fields = set(re.findall(r'<field name="([^"]+)"', jrxml))
+    # 支持 namespace 前缀的 field 声明（如 <jrxml:field>）
+    raw_fields = re.findall(r'(?:<[a-zA-Z0-9_-]+:)?field\s+name="([^"]+)"', jrxml)
+    jrxml_fields = set(raw_fields)
     ocr_field_names = set()
     ocr_fields = ocr_result.get("fields", []) if isinstance(ocr_result, dict) else []
     for f in ocr_fields:

backend/llm.py

@@ -35,7 +35,6 @@ class _LLMLoggingWrapper(_BaseLLM):
     def invoke(self, prompt: str) -> Any:
         t0 = time.time()
         prompt_len = len(prompt)
-        prompt_preview = prompt[:500]
         _llm_log.debug(
             "LLM invoke 请求",
             extra={
@@ -44,8 +43,7 @@ class _LLMLoggingWrapper(_BaseLLM):
                 "backend": self._backend,
                 "caller": self._caller,
                 "prompt_length": prompt_len,
-                "prompt_preview": prompt_preview,
-                "prompt": prompt[:10000],
+                "prompt_preview": prompt[:500],
             },
         )
         try:
@@ -64,7 +62,6 @@ class _LLMLoggingWrapper(_BaseLLM):
                     "duration_ms": elapsed,
                     "response_length": resp_len,
                     "response_preview": resp_preview,
-                    "response": content[:10000],
                 },
             )
             return result
@@ -79,7 +76,7 @@ class _LLMLoggingWrapper(_BaseLLM):
                     "caller": self._caller,
                     "duration_ms": elapsed,
                     "error": str(e),
-                    "prompt": prompt[:10000],
+                    "prompt_preview": prompt[:500],
                 },
             )
             raise
@@ -96,8 +93,7 @@ class _LLMLoggingWrapper(_BaseLLM):
                 "backend": self._backend,
                 "caller": self._caller,
                 "prompt_length": prompt_len,
-                "prompt_preview": prompt_preview,
-                "prompt": prompt[:10000],
+                "prompt_preview": prompt[:500],
             },
         )
         full = []
@@ -135,7 +131,6 @@ class _LLMLoggingWrapper(_BaseLLM):
                         "duration_ms": elapsed,
                         "response_length": resp_len,
                         "response_preview": resp_preview,
-                        "response": resp_text[:10000],
                         "stop_reason": stop_reason,
                     },
                 )
@@ -150,7 +145,7 @@ class _LLMLoggingWrapper(_BaseLLM):
                     "caller": self._caller,
                     "duration_ms": elapsed,
                     "error": str(e),
-                    "prompt": prompt[:10000],
+                    "prompt_preview": prompt[:500],
                 },
             )
             raise

backend/session.py

@@ -6,11 +6,50 @@
 import json
 import os
 import re
+import threading
 import uuid
 import tempfile
 from datetime import datetime, timezone
 from pathlib import Path
-from typing import Optional
+from typing import Optional, Any
+
+# Per-session-file locks to prevent concurrent writes from corrupting JSON
+_session_locks: dict[str, threading.Lock] = {}
+_locks_lock = threading.Lock()
+
+def _get_lock(session_id: str) -> threading.Lock:
+    with _locks_lock:
+        if session_id not in _session_locks:
+            _session_locks[session_id] = threading.Lock()
+        return _session_locks[session_id]
+
+
+class _SafeEncoder(json.JSONEncoder):
+    """处理 numpy / lxml / 等非标准类型的 JSON 序列化"""
+
+    def default(self, o: Any) -> Any:
+        try:
+            # numpy 标量
+            import numpy as np
+            if isinstance(o, np.integer):
+                return int(o)
+            if isinstance(o, np.floating):
+                return float(o)
+            if isinstance(o, np.ndarray):
+                return o.tolist()
+            if isinstance(o, np.bool_):
+                return bool(o)
+        except ImportError:
+            pass
+        # lxml intc / 其他 C 类型
+        try:
+            return int(o)
+        except Exception:
+            pass
+        # bytes
+        if isinstance(o, bytes):
+            return o.decode("utf-8", errors="replace")
+        return super().default(o)
 
 from dotenv import load_dotenv
 
@@ -59,8 +98,21 @@ def create_session(name: str = "", agent_state: Optional[dict] = None,
         "kb_id": agent_state.get("kb_id", "") if agent_state else "",
         "agent_state": agent_state,
     }
-    with open(_session_path(sid), "w", encoding="utf-8") as f:
-        json.dump(data, f, ensure_ascii=False, indent=2)
+    fp = _session_path(sid)
+    tmp = tempfile.NamedTemporaryFile(
+        mode="w", suffix=".json", delete=False,
+        dir=SESSIONS_DIR, encoding="utf-8",
+    )
+    try:
+        json.dump(data, tmp, ensure_ascii=False, indent=2, cls=_SafeEncoder)
+        tmp.flush()
+        os.fsync(tmp.fileno())
+        tmp.close()
+        os.replace(tmp.name, str(fp))
+    except Exception:
+        tmp.close()
+        Path(tmp.name).unlink(missing_ok=True)
+        raise
     _session_log.info("创建会话", extra={"session_id": sid, "session_name": data["session_name"]})
     return data
 
@@ -79,39 +131,41 @@ def load_session(session_id: str) -> Optional[dict]:
 
 
 def save_session(session_id: str, agent_state: dict, session_name: str = ""):
-    """将会话状态原子保存至磁盘（temp file + rename，避免崩溃时截断）。"""
+    """线程安全地原子保存会话状态到磁盘。"""
     _ensure_dir()
     fp = _session_path(session_id)
-    data = {}
-    if fp.exists():
-        with open(fp, "r", encoding="utf-8") as f:
-            data = json.load(f)
+    lock = _get_lock(session_id)
+    with lock:
+        data = {}
+        if fp.exists():
+            with open(fp, "r", encoding="utf-8") as f:
+                data = json.load(f)
 
-    data["session_id"] = session_id
-    if session_name:
-        data["session_name"] = session_name
-    if not data.get("session_name"):
-        data["session_name"] = f"报表 {data.get('created_at', _now_iso())[:10]}"
-    data["updated_at"] = _now_iso()
-    if not data.get("created_at"):
-        data["created_at"] = data["updated_at"]
-    data["agent_state"] = agent_state
+        data["session_id"] = session_id
+        if session_name:
+            data["session_name"] = session_name
+        if not data.get("session_name"):
+            data["session_name"] = f"报表 {data.get('created_at', _now_iso())[:10]}"
+        data["updated_at"] = _now_iso()
+        if not data.get("created_at"):
+            data["created_at"] = data["updated_at"]
+        data["agent_state"] = agent_state
 
-    # 原子写入：先写临时文件，再 replace，避免崩溃时截断 JSON
-    tmp = tempfile.NamedTemporaryFile(
-        mode="w", suffix=".json", delete=False,
-        dir=SESSIONS_DIR, encoding="utf-8",
-    )
-    try:
-        json.dump(data, tmp, ensure_ascii=False, indent=2)
-        tmp.flush()
-        os.fsync(tmp.fileno())
-        tmp.close()
-        os.replace(tmp.name, str(fp))
-    except Exception:
-        tmp.close()
-        Path(tmp.name).unlink(missing_ok=True)
-        raise
+        # 原子写入：先写临时文件，再 replace，避免崩溃时截断 JSON
+        tmp = tempfile.NamedTemporaryFile(
+            mode="w", suffix=".json", delete=False,
+            dir=SESSIONS_DIR, encoding="utf-8",
+        )
+        try:
+            json.dump(data, tmp, ensure_ascii=False, indent=2, cls=_SafeEncoder)
+            tmp.flush()
+            os.fsync(tmp.fileno())
+            tmp.close()
+            os.replace(tmp.name, str(fp))
+        except Exception:
+            tmp.close()
+            Path(tmp.name).unlink(missing_ok=True)
+            raise
 
 
 def get_session_state(session_id: str) -> Optional[dict]:

prompts/correction.md

@@ -8,6 +8,8 @@
 - 如果当前 JRXML 内容为空或过短（<200 字符），请根据下方提供的 OCR 识别数据和布局 schema 重新生成完整的 JRXML，而非输出一个占位桩。
 - 如果错误是"字段 'field_N' 未在 <field> 部分声明"，**必须**为每个缺失的 field_N 添加 `<field name="field_N" class="java.lang.String"/>` 声明。这些是占位字段，不可删除。同时确保所有 $F{{field_N}} 引用都有对应的 <field> 声明。
 - 如果错误是"字段 'field_N' 未在 <field> 部分声明"且有 OCR 字段数据，尝试将 $F{{field_N}} 替换为 OCR 中对应的真实字段名（如 $F{{invoice_code}}），同时更新 <field> 声明和所有引用。
+- 【强制】修正后的 JRXML 必须保证所有 $F{...} 引用都有对应的 <field name="..."> 声明。禁止出现 $F{field_name} 却没有对应 field 声明的情况。
+- 【强制】font 标签必须符合 JasperReports XSD：<font fontName="..." size="..." isBold="..." isItalic="..." isUnderline="..."/>。禁止在 <font> 标签上写 fontName= 属性（错误写法），必须使用嵌套属性格式（正确写法）。
 - **始终检查并修复命名空间**：正确的根元素格式必须为：`<jasperReport xmlns="http://jasperreports.sourceforge.net/jasperreports" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://jasperreports.sourceforge.net/jasperreports http://jasperreports.sourceforge.net/xsd/jasperreport.xsd">`。删除所有 ns0: 前缀，删除所有 `xmlns:ns0` 声明，删除所有元素标签上的 `ns0:` 前缀。
 
 当前 JRXML（带错误）：

prompts/initial_generation.md

@@ -4,6 +4,8 @@ JRXML 必须兼容 JasperReports 7.0.6 schema。
 关键规则：
 - 只输出 JRXML 代码，不要解释，不要 markdown 标记。
 - 报表正文中使用的每个字段必须在 <field name="..."> 部分中声明。
+- 【强制】在 <jasperReport> 下必须包含完整的 <fields> 节，列出所有用到的字段。每个字段格式：<field name="field_name" class="java.lang.String"/>。禁止出现 $F{field_name} 却没有对应 field 声明的情况。
+- 【强制】font 标签结构：使用 <font fontName="Serif" size="12"/> 而非 <fontName="Serif"/> 等属性写法。font 标签必须符合 JasperReports XSD：<font fontName="..." size="..." isBold="..." isItalic="..." isUnderline="..."/>
 - 根元素为 <jasperReport>，包含正确的 xmlns 属性。**禁止在元素标签上使用 ns0: 前缀**。正确的根元素格式：
   ```xml
   <jasperReport xmlns="http://jasperreports.sourceforge.net/jasperreports" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://jasperreports.sourceforge.net/jasperreports http://jasperreports.sourceforge.net/xsd/jasperreport.xsd">

prompts/skeleton_generation.md

@@ -3,6 +3,8 @@
 关键规则：
 - 只输出 JRXML 代码，不要解释，不要 markdown 标记。
 - 使用 $F{{field_1}}, $F{{field_2}}, ... 作为占位字段名，并在 <field> 部分声明它们。
+- 【强制】在 <jasperReport> 下必须包含完整的 <fields> 节，列出所有用到的字段。每个字段格式：<field name="field_name" class="java.lang.String"/>。禁止出现 $F{field_name} 却没有对应 field 声明的情况。
+- 【强制】font 标签结构：使用 <font fontName="Serif" size="12"/> 而非 <fontName="Serif"/> 等属性写法。font 标签必须符合 JasperReports XSD：<font fontName="..." size="..." isBold="..." isItalic="..." isUnderline="..."/>
 - 报表结构必须正确（title, pageHeader, columnHeader, detail, pageFooter 等 band）。
 - 元素位置使用近似值即可，后续会精确调整。
 - 根元素为 <jasperReport xmlns="http://jasperreports.sourceforge.net/jasperreports" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://jasperreports.sourceforge.net/jasperreports http://jasperreports.sourceforge.net/xsd/jasperreport.xsd">，命名空间和 schemaLocation 必须精确，不可使用其他 URL（如 jaspersoft.com）。**禁止在元素标签上使用 ns0: 前缀**。

scripts/run_e2e.py

@@ -0,0 +1,126 @@
+"""
+Jaspersoft E2E 测试脚本
+用法: python scripts/run_e2e.py [--user-text "请根据图片生成结算单模板"]
+
+输出:
+  - tmp/e2e_events_{HHMMSS}.json  完整事件流
+  - tmp/e2e_log_{HHMMSS}.txt       节点日志
+"""
+import sys, os
+sys.path.insert(0, os.path.dirname(os.path.dirname(__file__)))
+sys.stdout = os.fdopen(sys.stdout.fileno(), 'w', buffering=1)
+
+import requests, json, time, uuid
+from pathlib import Path
+
+BASE_URL = "http://localhost:8000"
+TEST_IMAGE = Path(__file__).parent.parent / "test_image.jpg"
+USER_TEXT = "请根据图片信息生成结算单模板"
+
+ts = time.strftime("%H%M%S")
+out_path = Path(__file__).parent.parent / "tmp" / f"e2e_events_{ts}.json"
+log_path = Path(__file__).parent.parent / "tmp" / f"e2e_log_{ts}.txt"
+out_path.parent.mkdir(parents=True, exist_ok=True)
+
+def log(msg):
+    print(msg, flush=True)
+    with open(log_path, "a", encoding="utf-8") as f:
+        f.write(msg + "\n")
+
+def run():
+    log("=" * 60)
+    log(f"E2E 测试开始 {time.strftime('%H:%M:%S')}")
+
+    # 1. 创建会话
+    sid_resp = requests.post(f"{BASE_URL}/api/sessions", json={"session_id": "test"}, timeout=10)
+    sid = sid_resp.json()["session_id"]
+    log(f"[会话] {sid}")
+
+    # 2. 上传图片
+    with open(TEST_IMAGE, "rb") as f:
+        up_resp = requests.post(
+            f"{BASE_URL}/api/upload",
+            files={"file": ("test_image.jpg", f, "image/jpeg")},
+            data={"session_id": sid}, timeout=30,
+        )
+    fid = up_resp.json()["file_id"]
+    log(f"[上传] file_id={fid}")
+
+    # 3. 发送对话
+    log(f"[对话] 开始 pipeline...")
+    start = time.time()
+
+    events = []
+    node_times = {}
+    error_events = []
+
+    r = requests.post(
+        f"{BASE_URL}/api/sessions/{sid}/chat",
+        json={"text": USER_TEXT, "file_ids": [fid]},
+        stream=True, timeout=600,
+    )
+    log(f"[状态] HTTP {r.status_code}")
+
+    for line in r.iter_lines():
+        if not line:
+            continue
+        line = line.decode("utf-8", errors="replace")
+        if line.startswith("data:"):
+            try:
+                data = json.loads(line[5:].strip())
+                events.append(data)
+                evt = data.get("event", "")
+                d = data.get("data", {})
+                node = d.get("node", "")
+
+                if evt == "node_start":
+                    node_times.setdefault(node, {"start": time.time() - start, "complete": None})
+                    log(f"  [开始] {node}")
+
+                if evt == "node_complete":
+                    if node in node_times and node_times[node]["complete"] is None:
+                        dur = time.time() - start - node_times[node]["start"]
+                        node_times[node]["complete"] = time.time() - start
+                        detail = d.get("detail", "")[:80]
+                        log(f"  [完成] {node} ({dur:.1f}s) — {detail}")
+
+                if evt == "error":
+                    msg = d.get("message", str(data))[:200]
+                    log(f"  [错误] {msg}")
+                    error_events.append(data)
+
+                if evt == "result":
+                    result = data.get("data", {})
+                    elapsed = time.time() - start
+                    jrxml = result.get("jrxml", "")
+                    log(f"\n{'='*50}")
+                    log(f"[完成] 耗时 {elapsed:.1f}s")
+                    log(f"  status: {result.get('status', 'N/A')}")
+                    log(f"  jrxml_length: {len(jrxml)}")
+                    log(f"  error: {result.get('error', 'None')[:200]}")
+
+                if evt == "done":
+                    log(f"\n[SSE Done]")
+
+            except json.JSONDecodeError:
+                pass
+
+    elapsed_total = time.time() - start
+    log(f"\n总耗时: {elapsed_total:.1f}s")
+    log(f"共 {len(events)} 个事件，{len(node_times)} 个节点，{len(error_events)} 个错误")
+
+    # 保存
+    with open(out_path, "w", encoding="utf-8") as f:
+        json.dump({
+            "session_id": sid,
+            "elapsed": elapsed_total,
+            "events": events,
+            "node_times": node_times,
+            "error_events": error_events,
+        }, f, ensure_ascii=False, indent=2)
+
+    log(f"事件已保存: {out_path}")
+    log(f"日志已保存: {log_path}")
+
+if __name__ == "__main__":
+    run()