diff --git a/backend/src/alphapulse/crawler/spiders/wscn.py b/backend/src/alphapulse/crawler/spiders/wscn.py
index 4c514aa..0368263 100644
--- a/backend/src/alphapulse/crawler/spiders/wscn.py
+++ b/backend/src/alphapulse/crawler/spiders/wscn.py
@@ -1,6 +1,7 @@
 import base64
 import json
 from datetime import datetime
+from html import escape
 
 from scrapy import Request
 from scrapy.http import Response
@@ -10,6 +11,12 @@ from alphapulse.crawler.spiders.base import NewsSpider, ts
 API = "https://api-one-wscn.awtmt.com/apiv1/content"
 
 
+def _api_data(body: dict) -> dict:
+    if body["code"] != 20000:
+        raise ValueError(f"华尔街见闻接口业务错误（code={body['code']}）：{body['message']}")
+    return body["data"]
+
+
 class WscnNewsSpider(NewsSpider):
     """华尔街见闻·文章：列表只有摘要，新条目或标题/摘要变化时再拉详情"""
 
@@ -27,14 +34,21 @@ class WscnNewsSpider(NewsSpider):
         return self.list_request(base64.b64encode(cursor.encode()).decode(), 1, history=True)
 
     async def parse_list(self, response: Response):
-        data = response.json()["data"]
+        data = _api_data(response.json())
         # 信息流里混有快讯，快讯由 wscn_live 负责
         rows = [x["resource"] for x in data["items"] if x["resource_type"] == "article"]
         ids = [str(r["id"]) for r in rows]
         known = await self.known_items(ids)
         for r in rows:
             if known.get(str(r["id"])) != (r["title"], r["content_short"]):
-                yield self.request(f"{API}/articles/{r['id']}?extract=0", self.parse_detail, summary=r["content_short"])
+                listed = {key: r[key] for key in ("id", "uri", "title", "display_time", "is_priced")}
+                listed["author"] = (r.get("author") or {}).get("display_name")
+                listed["source_name"] = r.get("source_name")
+                yield self.request(
+                    f"{API}/articles/{r['id']}?extract=0", self.parse_detail,
+                    summary=r["content_short"], listed_article=listed,
+                    list_raw_response_id=response.meta["raw_response_id"], history=response.meta.get("history", False),
+                )
         page = response.meta["page"]
         oldest = ts(min(r["display_time"] for r in rows)) if rows else None
         if response.meta.get("history"):
@@ -44,11 +58,30 @@ class WscnNewsSpider(NewsSpider):
             yield self.list_request(data["next_cursor"], page + 1)
 
     async def parse_detail(self, response: Response):
-        a = response.json()["data"]
+        body = response.json()
+        listed = response.meta["listed_article"]
+        if body["code"] == 50000 and response.meta["history"] and listed["is_priced"]:
+            # 旧付费课程的详情和官网页面均可能失效；保存列表公开信息，不伪装成已取得正文。
+            self.logger.warning("历史付费文章 %s 详情接口失败（code=%s）：%s；仅保存列表公开摘要",
+                                listed["id"], body["code"], body["message"])
+            summary = response.meta["summary"]
+            item = self.item(
+                response, external_id=str(listed["id"]), url=listed["uri"], title=listed["title"],
+                summary=summary, content_html=f"<p>{escape(summary)}</p>",
+                published_at=ts(listed["display_time"]), author=listed["author"], is_paywalled=True,
+                extra={"source_name": listed["source_name"], "is_priced": True,
+                       "content_scope": "public_summary", "historical_backfill": True,
+                       "detail_error": {"code": body["code"], "message": body["message"],
+                                        "raw_response_id": response.meta["raw_response_id"]}},
+            )
+            item["raw_response_id"] = response.meta["list_raw_response_id"]
+            yield item
+            return
+        a = _api_data(body)
         yield self.item(
             response,
             external_id=str(a["id"]),
-            url=f"https://wallstreetcn.com/articles/{a['id']}",
+            url=listed["uri"],
             title=a["title"],
             summary=response.meta["summary"],
             content_html=a["content"],
@@ -75,7 +108,7 @@ class WscnLiveSpider(NewsSpider):
         return self.list_request(str(int(before.timestamp())), 1, history=True)
 
     async def parse_list(self, response: Response):
-        data = response.json()["data"]
+        data = _api_data(response.json())
         rows = data["items"]
         # 先查已入库集合，再产出条目：否则管道可能先落库，导致误判为没有新条目而停止翻页
         ids = [str(x["id"]) for x in rows]
diff --git a/backend/tests/test_wscn.py b/backend/tests/test_wscn.py
new file mode 100644
index 0000000..d30f1eb
--- /dev/null
+++ b/backend/tests/test_wscn.py
@@ -0,0 +1,111 @@
+import asyncio
+import json
+import uuid
+from datetime import UTC, datetime, timedelta
+
+import pytest
+from scrapy.http import Request, TextResponse
+
+
+@pytest.fixture
+def wscn(trading):
+    from alphapulse.crawler.spiders.base import RunState
+    from alphapulse.crawler.spiders.wscn import WscnNewsSpider
+
+    spider = WscnNewsSpider()
+    now = datetime(2026, 10, 22, tzinfo=UTC)
+    spider.run = RunState(uuid.uuid4(), "schedule", now, now - timedelta(hours=72),
+                          history_target=datetime(2025, 1, 1, tzinfo=UTC),
+                          history_position=datetime(2025, 12, 10, tzinfo=UTC))
+    return spider
+
+
+def json_response(request, body, raw_id):
+    request.meta["raw_response_id"] = raw_id
+    return TextResponse(request.url, request=request, body=json.dumps(body).encode(), encoding="utf-8")
+
+
+async def detail_request(spider, monkeypatch, *, history=True, priced=True):
+    async def known_items(ids):
+        return {}
+    monkeypatch.setattr(spider, "known_items", known_items)
+    response = json_response(spider.list_request("", 1, history=history), {
+        "code": 20000, "data": {"items": [{"resource_type": "article", "resource": {
+            "id": 3759372, "uri": "https://wallstreetcn.com/premium/articles/3759372?layout=wscn-layout",
+            "title": "公开课程标题", "content_short": "公开摘要 <script> is text & data",
+            "display_time": 1765281395, "is_priced": priced,
+            "author": {"display_name": "公开作者"}, "source_name": "公开来源",
+        }}], "next_cursor": "next-page"},
+    }, 41)
+    requests = [request async for request in spider.parse_list(response)]
+    return next(request for request in requests if "/articles/3759372?" in request.url)
+
+
+def test_historical_paid_error_keeps_list_summary_and_both_raw_responses(wscn, monkeypatch, caplog):
+    async def check():
+        request = await detail_request(wscn, monkeypatch)
+        response = json_response(request, {"code": 50000, "message": "内部错误, 请稍后重试", "data": {}}, 42)
+        item, = [item async for item in wscn.parse_detail(response)]
+        assert item["external_id"] == "3759372"
+        assert item["url"] == request.meta["listed_article"]["uri"]
+        assert item["title"] == "公开课程标题" and item["author"] == "公开作者"
+        assert item["summary"] == item["content_text"] == request.meta["summary"]
+        assert "<script>" not in item["content_html"] and "&lt;script&gt;" in item["content_html"]
+        assert item["published_at"] == datetime.fromtimestamp(1765281395, UTC)
+        assert item["is_paywalled"] is True
+        assert item["extra"]["content_scope"] == "public_summary"
+        assert item["extra"]["historical_backfill"] is True
+        assert item["extra"]["detail_error"] == {"code": 50000, "message": "内部错误, 请稍后重试", "raw_response_id": 42}
+        assert item["raw_response_id"] == 41
+        assert not wscn.run.errors
+        assert wscn.run.history_position == item["published_at"]
+    asyncio.run(check())
+    assert "50000" in caplog.text and "内部错误" in caplog.text and "仅保存列表公开摘要" in caplog.text
+
+
+@pytest.mark.parametrize("handler", ["article_list", "article_detail", "live_list"])
+def test_http_200_business_error_reports_upstream_cause(wscn, monkeypatch, handler):
+    from alphapulse.crawler.spiders.wscn import WscnLiveSpider
+
+    async def check():
+        if handler == "article_detail":
+            request = await detail_request(wscn, monkeypatch, history=False)
+            callback = wscn.parse_detail
+        elif handler == "article_list":
+            request = wscn.list_request("", 1)
+            callback = wscn.parse_list
+        else:
+            request = Request("https://example.invalid/live")
+            callback = WscnLiveSpider().parse_list
+        response = json_response(request, {"code": 50000, "message": "内部错误, 请稍后重试", "data": {}}, 42)
+        with pytest.raises(ValueError, match="code=50000.*内部错误"):
+            _ = [item async for item in callback(response)]
+    asyncio.run(check())
+
+
+@pytest.mark.parametrize("priced,code", [(False, 50000), (True, 60327)])
+def test_public_article_or_other_business_error_does_not_become_summary(wscn, monkeypatch, priced, code):
+    async def check():
+        request = await detail_request(wscn, monkeypatch, priced=priced)
+        response = json_response(request, {"code": code, "message": "上游错误", "data": {}}, 42)
+        with pytest.raises(ValueError, match=f"code={code}.*上游错误"):
+            _ = [item async for item in wscn.parse_detail(response)]
+    asyncio.run(check())
+
+
+@pytest.mark.parametrize("priced", [False, True])
+def test_successful_detail_preserves_body_and_uses_official_list_link(wscn, monkeypatch, priced):
+    async def check():
+        request = await detail_request(wscn, monkeypatch, priced=priced)
+        response = json_response(request, {"code": 20000, "data": {
+            "id": 3759372, "title": "详情标题", "content": "<p>详情公开正文</p>",
+            "display_time": 1765281395, "author": {"display_name": "详情作者"},
+            "is_need_pay": priced, "is_priced": priced, "source_name": "详情来源",
+        }}, 42)
+        item, = [item async for item in wscn.parse_detail(response)]
+        assert item["content_html"] == "<p>详情公开正文</p>" and item["content_text"] == "详情公开正文"
+        assert item["title"] == "详情标题" and item["author"] == "详情作者"
+        assert item["is_paywalled"] is priced
+        assert item["url"] == request.meta["listed_article"]["uri"] and item["raw_response_id"] == 42
+        assert "content_scope" not in item["extra"] and "detail_error" not in item["extra"]
+    asyncio.run(check())