diff --git a/app/api/chat.py b/app/api/chat.py
index 52407fd..a3932a3 100644
--- a/app/api/chat.py
+++ b/app/api/chat.py
@@ -11,6 +11,9 @@ has_disclaimer, trace_id}。
 落库：agent_session/agent_message（同 trace_id）；agent_tool_call 由 Tool
 节点落（T-04，success/blocked/error 全留痕）。审计：鉴权失败/越权经
 deps.deny 双写留痕；输入防护拒绝经 T-03 落 input_guard_log。
+方案 B（前端拉侧）：GET /sessions（本人会话分页列表）、
+GET /sessions/{id}/messages（历史消息升序分页）、POST /sessions/{id}/close
+（active→closed，重复关闭 409）；与 POST "" 共用入口守卫 + SessionGuard。
 """
 
 from __future__ import annotations
@@ -18,7 +21,7 @@ from __future__ import annotations
 import logging
 from uuid import uuid4
 
-from fastapi import APIRouter, Depends, Request
+from fastapi import APIRouter, Depends, Query, Request
 from pydantic import BaseModel, Field
 
 from app.api.deps import (
@@ -93,20 +96,51 @@ def _resolve_customer_id(
     return None
 
 
-@router.post("")
-def chat_api(req: ChatRequest, request: Request, auth: AuthContext = Depends(get_auth_context)) -> dict:
+def _resolve_agent_type(request: Request) -> str:
+    """X-Agent-Type 解析（POST 与查询/关闭端点共用；debug 通道在此强制）。"""
     agent_type = request.headers.get("X-Agent-Type", "").strip()
     if not agent_type:
         raise ApiError(401, "AUTH_401_MISSING_AGENT_TYPE", "missing X-Agent-Type header")
     if agent_type not in AGENT_TYPES:
         raise ApiError(400, "BAD_REQUEST", f"invalid X-Agent-Type: {agent_type}")
+    return agent_type
+
+
+def _assert_chat_entry(auth: AuthContext, agent_type: str) -> None:
+    """对话线入口守卫（四端点共用）：矩阵准入 + risk_manager 显式拒绝。
+
+    C5 前置（PRD 4A.1）：对话线不放行 risk_manager——HTTP 台账才放行，保住
+    FR-6 冻结口径。矩阵放行解决 HTTP 通道，chat 层显式拒绝兜底（manager 根本
+    进不了对话线，Tool 层 assert_tool_access 天然 fail-closed）。会话查询/
+    关闭端点同属对话线数据面，沿用同一口径（方案 B 拍板）。
+    """
     assert_agent_access(auth, agent_type, risk_repo=_repo())
-    # C5 前置（PRD 4A.1）：对话线不放行 risk_manager——HTTP 台账才放行，保住
-    # FR-6 冻结口径。矩阵放行解决 HTTP 通道，chat 层显式拒绝兜底（manager 根本
-    # 进不了对话线，Tool 层 assert_tool_access 天然 fail-closed）。
     if agent_type == "risk" and "risk_manager" in auth.roles:
         deny(auth, "AUTH_403_ROLE", _repo(), message="对话线仅限 risk_officer，请走 HTTP 台账")
 
+
+def _guard_session(auth: AuthContext, agent_type: str, session_id: str) -> dict:
+    """SessionGuard（手册 §9）：会话存在 + actor/agent_type 一致，他人会话 fail-closed。"""
+    session = _session_repo().get_session(session_id)
+    if session is None:
+        raise ApiError(404, "NOT_FOUND", f"session not found: {session_id}")
+    if session["actor_id"] != auth.actor_id or session["agent_type"] != agent_type:
+        deny(
+            auth,
+            "AUTH_403_SESSION_AGENT",
+            _repo(),
+            customer_id=session.get("customer_id"),
+            message="session belongs to another actor or agent",
+            agent_type=agent_type,
+        )
+    return session
+
+
+@router.post("")
+def chat_api(req: ChatRequest, request: Request, auth: AuthContext = Depends(get_auth_context)) -> dict:
+    agent_type = _resolve_agent_type(request)
+    _assert_chat_entry(auth, agent_type)
+
     message = req.message.strip()
     if not message:
         raise ApiError(400, "BAD_REQUEST", "message is blank")
@@ -164,19 +198,7 @@ def chat_api(req: ChatRequest, request: Request, auth: AuthContext = Depends(get
     session_repo = _session_repo()
 
     if req.session_id:
-        session = session_repo.get_session(req.session_id)
-        if session is None:
-            raise ApiError(404, "NOT_FOUND", f"session not found: {req.session_id}")
-        # SessionGuard（手册 §9）：actor/agent_type 一致；他人会话 fail-closed
-        if session["actor_id"] != auth.actor_id or session["agent_type"] != agent_type:
-            deny(
-                auth,
-                "AUTH_403_SESSION_AGENT",
-                _repo(),
-                customer_id=session.get("customer_id"),
-                message="session belongs to another actor or agent",
-                agent_type=agent_type,
-            )
+        session = _guard_session(auth, agent_type, req.session_id)
         if session["status"] != "active":
             raise ApiError(409, "STATE_CONFLICT", f"session is {session['status']}")
         sid = session["session_id"]
@@ -235,3 +257,68 @@ def chat_api(req: ChatRequest, request: Request, auth: AuthContext = Depends(get
         "has_disclaimer": result["has_disclaimer"],
         "trace_id": trace_id,
     }
+
+
+# ---------- 方案 B：前端「拉」侧只读接口（会话列表 / 历史消息 / 关闭会话） ----------
+#
+# 前端对话页三件套，与 POST "" 共用同一套入口守卫（_assert_chat_entry：
+# 矩阵准入 + risk_manager 显式拒绝）与 SessionGuard（_guard_session：
+# 仅本人会话 + agent_type 一致，越权 403 留痕）。纯读/状态流转，不改表、
+# 不碰 Tool 契约；manager 与对话线保持同口径 deny（见 _assert_chat_entry）。
+
+
+@router.get("/sessions")
+def list_sessions_api(
+    request: Request,
+    limit: int = Query(20, ge=1, le=100),
+    offset: int = Query(0, ge=0),
+    auth: AuthContext = Depends(get_auth_context),
+) -> dict:
+    """当前登录人的会话列表（created_at 倒序分页；agent_type 经 X-Agent-Type 头指定）。"""
+    agent_type = _resolve_agent_type(request)
+    _assert_chat_entry(auth, agent_type)
+    items, total = _session_repo().list_sessions(
+        actor_id=auth.actor_id, agent_type=agent_type, limit=limit, offset=offset
+    )
+    return {"items": items, "total": total, "limit": limit, "offset": offset}
+
+
+@router.get("/sessions/{session_id}/messages")
+def list_messages_api(
+    session_id: str,
+    request: Request,
+    limit: int = Query(50, ge=1, le=200),
+    offset: int = Query(0, ge=0),
+    auth: AuthContext = Depends(get_auth_context),
+) -> dict:
+    """指定会话的历史消息（seq_no 升序分页；closed 会话历史仍可读）。"""
+    agent_type = _resolve_agent_type(request)
+    _assert_chat_entry(auth, agent_type)
+    _guard_session(auth, agent_type, session_id)
+    items, total = _session_repo().list_messages_page(session_id, limit=limit, offset=offset)
+    return {
+        "session_id": session_id,
+        "items": items,
+        "total": total,
+        "limit": limit,
+        "offset": offset,
+    }
+
+
+@router.post("/sessions/{session_id}/close")
+def close_session_api(
+    session_id: str,
+    request: Request,
+    auth: AuthContext = Depends(get_auth_context),
+) -> dict:
+    """关闭会话（active → closed + closed_at）；重复关闭/非 active 409。"""
+    agent_type = _resolve_agent_type(request)
+    _assert_chat_entry(auth, agent_type)
+    session = _guard_session(auth, agent_type, session_id)
+    if session["status"] != "active":
+        raise ApiError(409, "STATE_CONFLICT", f"session is {session['status']}")
+    # 条件更新（WHERE status='active'）：并发双击时后到者 rowcount=0 —— 不静默
+    # 返回 200，转 409 与“重复关闭”同语义（评审 P1）。
+    if not _session_repo().close_session(session_id):
+        raise ApiError(409, "STATE_CONFLICT", "session is closed")
+    return {"session_id": session_id, "status": "closed"}
diff --git a/app/repository/session_repository.py b/app/repository/session_repository.py
index 6537443..3b6218d 100644
--- a/app/repository/session_repository.py
+++ b/app/repository/session_repository.py
@@ -33,6 +33,66 @@ class SessionRepository:
             ).mappings().first()
             return dict(row) if row else None
 
+    def list_sessions(
+        self, *, actor_id: str, agent_type: str, limit: int = 20, offset: int = 0
+    ) -> tuple[list[dict[str, Any]], int]:
+        """前端会话列表（方案 B）：仅本人 + 本 Agent 线，created_at 倒序分页。
+
+        返回 (items, total)；total 供前端分页器。id 倒序兜底同秒并发建的
+        会话排序稳定（created_at 精度秒级时并列）。datetime 统一转 str——
+        sqlite 返 str、MySQL 返 datetime，响应体跨库同构。
+        """
+        where = "WHERE actor_id = :actor AND agent_type = :atype"
+        with self._engine.connect() as conn:
+            total = int(
+                conn.execute(
+                    text(f"SELECT COUNT(*) FROM agent_session {where}"),
+                    {"actor": actor_id, "atype": agent_type},
+                ).scalar_one()
+            )
+            rows = conn.execute(
+                text(
+                    f"""
+                    SELECT session_id, agent_type, actor_id, actor_role, customer_id,
+                           advisor_id, title, status, created_at, closed_at
+                    FROM agent_session {where}
+                    ORDER BY created_at DESC, id DESC
+                    LIMIT :lim OFFSET :off
+                    """
+                ),
+                {"actor": actor_id, "atype": agent_type, "lim": limit, "off": offset},
+            ).mappings().all()
+        items = [
+            {
+                **{k: r[k] for k in (
+                    "session_id", "agent_type", "actor_id", "actor_role",
+                    "customer_id", "advisor_id", "title", "status",
+                )},
+                "created_at": str(r["created_at"]) if r["created_at"] is not None else None,
+                "closed_at": str(r["closed_at"]) if r["closed_at"] is not None else None,
+            }
+            for r in rows
+        ]
+        return items, total
+
+    def close_session(self, session_id: str) -> bool:
+        """关闭会话（方案 B）：active → closed + closed_at 落时间。
+
+        条件更新（WHERE status='active'）保证幂等语义由路由层判定——重复
+        关闭返回 False 由路由转 409，不会出现并发双击把 closed_at 刷新的
+        静默写。审计表只 INSERT 铁律不适用于本表（agent_session 是对话域
+        业务表，status 流转是既有设计，见 _ddl/01-mysql-共用底座）。
+        """
+        with self._engine.begin() as conn:
+            result = conn.execute(
+                text(
+                    "UPDATE agent_session SET status = 'closed', closed_at = CURRENT_TIMESTAMP"
+                    " WHERE session_id = :sid AND status = 'active'"
+                ),
+                {"sid": session_id},
+            )
+            return result.rowcount > 0
+
     def create_session(
         self,
         *,
@@ -180,3 +240,41 @@ class SessionRepository:
                 }
                 for r in rows
             ]
+
+    def list_messages_page(
+        self, session_id: str, *, limit: int = 50, offset: int = 0
+    ) -> tuple[list[dict[str, Any]], int]:
+        """前端历史消息分页（方案 B）：seq_no 升序 + offset，返回 (items, total)。
+
+        与 list_messages（LLM 窗口「最近 N 条」）语义不同：前端要全量可翻页，
+        从第一条开始升序拉取。created_at 一并返回供前端展示时间。
+        """
+        with self._engine.connect() as conn:
+            total = int(
+                conn.execute(
+                    text("SELECT COUNT(*) FROM agent_message WHERE session_id = :sid"),
+                    {"sid": session_id},
+                ).scalar_one()
+            )
+            rows = conn.execute(
+                text(
+                    """
+                    SELECT seq_no, role, content, has_disclaimer, created_at
+                    FROM agent_message WHERE session_id = :sid
+                    ORDER BY seq_no ASC
+                    LIMIT :lim OFFSET :off
+                    """
+                ),
+                {"sid": session_id, "lim": limit, "off": offset},
+            ).mappings().all()
+        items = [
+            {
+                "seq_no": r["seq_no"],
+                "role": r["role"],
+                "content": r["content"],
+                "has_disclaimer": bool(r["has_disclaimer"]),
+                "created_at": str(r["created_at"]) if r["created_at"] is not None else None,
+            }
+            for r in rows
+        ]
+        return items, total
diff --git a/tests/test_chat.py b/tests/test_chat.py
index 95a275a..ebd4a72 100644
--- a/tests/test_chat.py
+++ b/tests/test_chat.py
@@ -335,3 +335,156 @@ def test_chat_jwt_agent_mismatch_denied(env):
         headers={"Authorization": f"Bearer {tok}", "X-Agent-Type": "customer"},
     )
     assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_AGENT_MISMATCH"
+
+
+# ---------- 方案 B：前端拉侧（会话列表 / 历史消息 / 关闭会话） ----------
+
+RISK_OFFICER = {"X-Debug-Role": "risk_officer", "X-Debug-Actor": "STAFF-30001", "X-Agent-Type": "risk"}
+RISK_MANAGER = {"X-Debug-Role": "risk_manager", "X-Debug-Actor": "STAFF-31001", "X-Agent-Type": "risk"}
+
+
+def test_sessions_list_only_own_and_paged(env):
+    """会话列表：仅本人 + 本 Agent 线；created_at 倒序；limit/offset 分页 + total。"""
+    sid1 = env["client"].post("/api/chat", json={"message": "第一条"}, headers=CUSTOMER).json()["session_id"]
+    sid2 = env["client"].post("/api/chat", json={"message": "第二条"}, headers=CUSTOMER).json()["session_id"]
+    # 另一 actor 的会话不应出现在我的列表里
+    env["client"].post(
+        "/api/chat", json={"message": "别人的"},
+        headers={"X-Debug-Role": "customer", "X-Debug-Actor": "CUST-1001", "X-Agent-Type": "customer"},
+    )
+
+    r = env["client"].get("/api/chat/sessions", headers=CUSTOMER)
+    assert r.status_code == 200
+    body = r.json()
+    assert body["total"] == 2
+    assert [s["session_id"] for s in body["items"]] == [sid2, sid1]  # 倒序
+    assert all(s["agent_type"] == "customer" and s["status"] == "active" for s in body["items"])
+
+    r2 = env["client"].get("/api/chat/sessions?limit=1&offset=1", headers=CUSTOMER)
+    body2 = r2.json()
+    assert body2["total"] == 2 and len(body2["items"]) == 1
+    assert body2["items"][0]["session_id"] == sid1
+
+
+def test_sessions_list_risk_line_isolated(env):
+    """agent_type 隔离：risk 线会话不出现在 customer 线列表（反之亦然）。"""
+    env["client"].post("/api/chat", json={"message": "客户线"}, headers=CUSTOMER)
+    r = env["client"].get("/api/chat/sessions", headers=RISK_OFFICER)
+    assert r.status_code == 200 and r.json()["total"] == 0
+
+
+def test_messages_page_ascending_and_paged(env):
+    """历史消息：seq_no 升序全量分页；has_disclaimer 透出；closed 后仍可读。"""
+    sid = env["client"].post("/api/chat", json={"message": "第一句"}, headers=CUSTOMER).json()["session_id"]
+    env["client"].post("/api/chat", json={"message": "第二句", "session_id": sid}, headers=CUSTOMER)
+
+    r = env["client"].get(f"/api/chat/sessions/{sid}/messages", headers=CUSTOMER)
+    assert r.status_code == 200
+    body = r.json()
+    assert body["total"] == 4
+    assert [m["role"] for m in body["items"]] == ["user", "assistant", "user", "assistant"]
+    assert [m["seq_no"] for m in body["items"]] == [1, 2, 3, 4]
+    assert body["items"][0]["content"] == "第一句"
+    assert body["items"][1]["has_disclaimer"] is True  # customer 线 assistant 带免责声明
+
+    r2 = env["client"].get(f"/api/chat/sessions/{sid}/messages?limit=2&offset=2", headers=CUSTOMER)
+    body2 = r2.json()
+    assert body2["total"] == 4 and [m["seq_no"] for m in body2["items"]] == [3, 4]
+
+
+def test_messages_of_other_actor_denied(env):
+    """SessionGuard：他人会话历史 403 + 审计留痕（与 POST 同口径）。"""
+    sid = env["client"].post("/api/chat", json={"message": "hi"}, headers=CUSTOMER).json()["session_id"]
+    other = {**CUSTOMER, "X-Debug-Actor": "CUST-1001"}
+    r = env["client"].get(f"/api/chat/sessions/{sid}/messages", headers=other)
+    assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_SESSION_AGENT"
+    assert _rows(env["engine"], "SELECT 1 FROM audit_log WHERE decision = 'forbidden'")
+
+
+def test_messages_not_found_and_agent_type_mismatch(env):
+    r = env["client"].get("/api/chat/sessions/sess-nope/messages", headers=CUSTOMER)
+    assert r.status_code == 404
+    sid = env["client"].post("/api/chat", json={"message": "hi"}, headers=CUSTOMER).json()["session_id"]
+    # 会话在 customer 线，advisor 头查 → agent_type 不一致 403
+    r2 = env["client"].get(f"/api/chat/sessions/{sid}/messages", headers=ADVISOR)
+    assert r2.status_code == 403 and r2.json()["error_code"] == "AUTH_403_SESSION_AGENT"
+
+
+def test_close_session_then_read_only(env):
+    """关闭会话：200 + closed_at 落库；续聊 409；重复关闭 409；历史仍可读。"""
+    sid = env["client"].post("/api/chat", json={"message": "hi"}, headers=CUSTOMER).json()["session_id"]
+
+    r = env["client"].post(f"/api/chat/sessions/{sid}/close", headers=CUSTOMER)
+    assert r.status_code == 200 and r.json() == {"session_id": sid, "status": "closed"}
+    row = _rows(env["engine"], "SELECT status, closed_at FROM agent_session WHERE session_id = :s", s=sid)[0]
+    assert row["status"] == "closed" and row["closed_at"] is not None
+
+    r2 = env["client"].post("/api/chat", json={"message": "续聊", "session_id": sid}, headers=CUSTOMER)
+    assert r2.status_code == 409 and r2.json()["error_code"] == "STATE_CONFLICT"
+
+    r3 = env["client"].post(f"/api/chat/sessions/{sid}/close", headers=CUSTOMER)
+    assert r3.status_code == 409 and r3.json()["error_code"] == "STATE_CONFLICT"
+
+    r4 = env["client"].get(f"/api/chat/sessions/{sid}/messages", headers=CUSTOMER)
+    assert r4.status_code == 200 and r4.json()["total"] == 2
+
+
+def test_close_other_actor_denied(env):
+    sid = env["client"].post("/api/chat", json={"message": "hi"}, headers=CUSTOMER).json()["session_id"]
+    other = {**CUSTOMER, "X-Debug-Actor": "CUST-1001"}
+    r = env["client"].post(f"/api/chat/sessions/{sid}/close", headers=other)
+    assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_SESSION_AGENT"
+    # 未关闭成功
+    row = _rows(env["engine"], "SELECT status FROM agent_session WHERE session_id = :s", s=sid)[0]
+    assert row["status"] == "active"
+
+
+def test_close_not_found_and_agent_type_mismatch(env):
+    """close 端点：会话不存在 404；会话在 customer 线、advisor 头关 → 403（评审 P2）。"""
+    r = env["client"].post("/api/chat/sessions/sess-nope/close", headers=CUSTOMER)
+    assert r.status_code == 404
+
+    sid = env["client"].post("/api/chat", json={"message": "hi"}, headers=CUSTOMER).json()["session_id"]
+    r2 = env["client"].post(f"/api/chat/sessions/{sid}/close", headers=ADVISOR)
+    assert r2.status_code == 403 and r2.json()["error_code"] == "AUTH_403_SESSION_AGENT"
+    row = _rows(env["engine"], "SELECT status FROM agent_session WHERE session_id = :s", s=sid)[0]
+    assert row["status"] == "active"
+
+
+def test_query_endpoints_limit_clamped(env):
+    """分页参数钳制：limit 超上限（sessions 100 / messages 200）由 Query 拦 422（评审 P2）。"""
+    assert env["client"].get("/api/chat/sessions?limit=101", headers=CUSTOMER).status_code == 422
+    assert env["client"].get("/api/chat/sessions?offset=-1", headers=CUSTOMER).status_code == 422
+    r = env["client"].get("/api/chat/sessions?limit=100", headers=CUSTOMER)
+    assert r.status_code == 200 and r.json()["limit"] == 100
+
+
+def test_query_endpoints_risk_manager_denied(env):
+    """方案 B 拍板：查询/关闭端点与对话线同口径——risk_manager 一律 403 AUTH_403_ROLE。"""
+    for method, url in (
+        ("get", "/api/chat/sessions"),
+        ("get", "/api/chat/sessions/sess-x/messages"),
+        ("post", "/api/chat/sessions/sess-x/close"),
+    ):
+        r = getattr(env["client"], method)(url, headers=RISK_MANAGER)
+        assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_ROLE"
+
+
+def test_query_endpoints_missing_agent_type(env):
+    r = env["client"].get(
+        "/api/chat/sessions",
+        headers={"X-Debug-Role": "customer", "X-Debug-Actor": "CUST-9527"},
+    )
+    assert r.status_code == 401 and r.json()["error_code"] == "AUTH_401_MISSING_AGENT_TYPE"
+
+
+def test_sessions_list_jwt_channel(env):
+    """JWT 通道：risk_officer 拉自己的 risk 线会话列表。"""
+    tok = auth_service.issue_dev_token(sub="STAFF-30001", roles=["risk_officer"])
+    headers = {"Authorization": f"Bearer {tok}", "X-Agent-Type": "risk"}
+    env["client"].post("/api/chat", json={"message": "看下预警"}, headers=headers)
+    r = env["client"].get("/api/chat/sessions", headers=headers)
+    assert r.status_code == 200
+    body = r.json()
+    assert body["total"] == 1 and body["items"][0]["agent_type"] == "risk"
+    assert body["items"][0]["actor_id"] == "STAFF-30001"
diff --git a/tests/test_main.py b/tests/test_main.py
index 4982392..0529247 100644
--- a/tests/test_main.py
+++ b/tests/test_main.py
@@ -65,6 +65,10 @@ def test_all_routers_mounted(client):
         "/api/risk/aml/scan",
         "/api/simulate/trade",
         "/api/chat",
+        # 方案 B：前端拉侧三端点
+        "/api/chat/sessions",
+        "/api/chat/sessions/{session_id}/messages",
+        "/api/chat/sessions/{session_id}/close",
     }