Merge pull request #321 from kevin9327/fix/strategy-build-stream-ping-newline

fix(strategy): 策略生成流的心跳行补换行, 不再吞掉首个 delta 与 result 事件

backend/app/api/strategy.py

@@ -919,11 +919,11 @@ async def build_strategy_stream(req: BuildRequest, request: Request):
     async def event_generator():
         gen = AIStrategyGenerator()
         chunks: list[str] = []
-        yield json.dumps({"type": "meta", "strategy_id": req.strategy_id, "step": req.step}, ensure_ascii=False) + "\n"
+        yield json.dumps({"type": "meta", "strategy_id": req.strategy_id, "step": req.step}, ensure_ascii=False)
         try:
             async for chunk in gen.stream(prompt):
                 chunks.append(chunk)
-                yield json.dumps({"type": "delta", "content": chunk}, ensure_ascii=False) + "\n"
+                yield json.dumps({"type": "delta", "content": chunk}, ensure_ascii=False)
             result = gen.validate_code("".join(chunks))
             if gen.needs_structural_repair(result):
                 result = await gen.repair_code(result["code"], result["error"])
@@ -931,14 +931,20 @@ async def build_strategy_stream(req: BuildRequest, request: Request):
                 result = _normalize_build_result(result, req.strategy_id, req.name, req.description)
             elif req.strategy_id:
                 result = _normalize_build_result(result, req.strategy_id)
-            yield json.dumps({"type": "result", **result}, ensure_ascii=False) + "\n"
+            yield json.dumps({"type": "result", **result}, ensure_ascii=False)
         except RuntimeError as e:
-            yield json.dumps({"type": "error", "message": str(e)}, ensure_ascii=False) + "\n"
+            yield json.dumps({"type": "error", "message": str(e)}, ensure_ascii=False)
         except Exception as e:
-            yield json.dumps({"type": "error", "message": f"AI生成失败: {e}"}, ensure_ascii=False) + "\n"
+            yield json.dumps({"type": "error", "message": f"AI生成失败: {e}"}, ensure_ascii=False)
+
+    # 与复盘/个股/财务/轮动的流式端点同口径: 事件行不带换行, 由外层统一补换行。
+    # with_heartbeat 插入的 ping 行本身不带换行, 事件行若自带换行, ping 会与下一行粘成一行。
+    async def stream_gen():
+        async for line in with_heartbeat(event_generator()):
+            yield line + "\n"
 
     return StreamingResponse(
-        with_heartbeat(event_generator()),
+        stream_gen(),
         media_type="application/x-ndjson",
         headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
     )

backend/tests/test_strategy_build_stream.py

@@ -1,10 +1,14 @@
 from __future__ import annotations
 
+import asyncio
+import functools
 import json
 
 import pytest
 
+from app.api import strategy as strategy_api
 from app.api.strategy import BuildRequest, build_strategy_stream
+from app.services.ndjson_heartbeat import with_heartbeat
 from app.strategy.ai_generator import AIStrategyGenerator
 
 STREAM_CODE = '''"""测试策略"""
@@ -100,3 +104,85 @@ async def test_build_strategy_stream_repairs_missing_meta_once(monkeypatch):
     assert result["valid"] is True
     assert result["meta"]["id"] == "ai_repaired"
     assert result["meta"]["name"] == "修复后策略"
+
+
+async def _read_body(response) -> str:
+    body = b""
+    async for chunk in response.body_iterator:
+        body += chunk.encode("utf-8") if isinstance(chunk, str) else chunk
+    return body.decode("utf-8")
+
+
+def _parse_like_frontend(text: str) -> tuple[list[dict], list[str]]:
+    """与前端 api.strategyBuildStream 同口径: 按换行切行, JSON 解析失败的行被静默跳过。"""
+    events: list[dict] = []
+    dropped: list[str] = []
+    for line in text.split("\n"):
+        s = line.strip()
+        if not s:
+            continue
+        try:
+            events.append(json.loads(s))
+        except json.JSONDecodeError:
+            dropped.append(s)
+    return events, dropped
+
+
+@pytest.mark.asyncio
+async def test_build_strategy_stream_heartbeat_does_not_swallow_events(monkeypatch):
+    """LLM 首包前与结构修复期间空闲超过心跳间隔: ping 必须独占一行, 不能把 delta/result 粘掉。"""
+
+    async def slow_stream(self, prompt):
+        await asyncio.sleep(0.2)  # 推理模型思考期, 流上无字节
+        yield "import polars as pl\n\ndef filter(df, params):\n    return pl.lit(True)\n"
+
+    async def slow_repair(self, code, error):
+        await asyncio.sleep(0.2)  # 修复是一次非流式 LLM 调用
+        return self.validate_code(STREAM_CODE)
+
+    monkeypatch.setattr(AIStrategyGenerator, "stream", slow_stream)
+    monkeypatch.setattr(AIStrategyGenerator, "repair_code", slow_repair)
+    monkeypatch.setattr(strategy_api, "with_heartbeat", functools.partial(with_heartbeat, interval=0.05))
+    req = BuildRequest(
+        step=1,
+        name="心跳策略",
+        description="心跳描述",
+        direction="long",
+        rules="1. 测试规则",
+        strategy_id="ai_heartbeat",
+    )
+
+    text = await _read_body(await build_strategy_stream(req, None))
+    events, dropped = _parse_like_frontend(text)
+    types = [event["type"] for event in events]
+
+    assert [t for t in types if t != "ping"] == ["meta", "delta", "result"]
+    assert "ping" in types
+    assert dropped == []
+    assert events[-1]["valid"] is True
+    assert events[-1]["meta"]["id"] == "ai_heartbeat"
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize(
+    ("exc", "message"),
+    [
+        (RuntimeError("AI API Key 未配置"), "AI API Key 未配置"),
+        (ValueError("boom"), "AI生成失败: boom"),
+    ],
+)
+async def test_build_strategy_stream_error_event_is_own_line(monkeypatch, exc, message):
+    async def failing_stream(self, prompt):
+        raise exc
+        yield  # pragma: no cover - 使其成为异步生成器
+
+    monkeypatch.setattr(AIStrategyGenerator, "stream", failing_stream)
+    req = BuildRequest(step=2, current_code="x = 1", instruction="改一下", strategy_id="")
+
+    text = await _read_body(await build_strategy_stream(req, None))
+    events, dropped = _parse_like_frontend(text)
+
+    assert dropped == []
+    assert [event["type"] for event in events] == ["meta", "error"]
+    assert events[-1]["message"] == message
+    assert text.endswith("\n")