diff --git a/app/api/conversations.py b/app/api/conversations.py
new file mode 100644
index 0000000..e55328e
--- /dev/null
+++ b/app/api/conversations.py
@@ -0,0 +1,30 @@
+"""会话创建与历史查询接口"""
+
+from fastapi import APIRouter, Request
+
+from app.common.response import ApiResponse, success_response
+from app.core.exceptions import BizException
+from app.schemas.conversation import ConversationCreated, ConversationDetail
+
+router = APIRouter(prefix="/conversations", tags=["Conversation"])
+
+
+@router.post("")
+async def create_conversation(
+    request: Request,
+) -> ApiResponse[ConversationCreated]:
+    store = request.app.state.conversation_store
+    return success_response(ConversationCreated(conversation_id=await store.create()))
+
+
+@router.get("/{conversation_id}")
+async def get_conversation(
+    conversation_id: str, request: Request
+) -> ApiResponse[ConversationDetail]:
+    store = request.app.state.conversation_store
+    if not await store.exists(conversation_id):
+        raise BizException(404, "会话不存在或已过期")
+    turns = await store.load_turns(conversation_id, None)
+    return success_response(
+        ConversationDetail(conversation_id=conversation_id, turns=turns)
+    )
diff --git a/app/schemas/conversation.py b/app/schemas/conversation.py
new file mode 100644
index 0000000..d0ae87c
--- /dev/null
+++ b/app/schemas/conversation.py
@@ -0,0 +1,27 @@
+"""会话及对话轮次，用于多轮上下文和后续持久化"""
+
+from datetime import datetime
+from typing import Literal
+
+from pydantic import BaseModel
+
+
+class Turn(BaseModel):
+    """一次任务对应一轮对话，同一个 thread_id 多次写入时覆盖同一轮"""
+
+    conversation_id: str
+    thread_id: str
+    message_id: str
+    prompt: str  # 用户原始输入
+    outcome: Literal["completed", "waiting_for_input", "failed", "cancelled"]
+    summary: str  # 助手侧摘要，只描述后端生成了什么，不代表前端已执行
+    created_at: datetime
+
+
+class ConversationCreated(BaseModel):
+    conversation_id: str
+
+
+class ConversationDetail(BaseModel):
+    conversation_id: str
+    turns: list[Turn]
diff --git a/app/services/conversation.py b/app/services/conversation.py
new file mode 100644
index 0000000..8a659f1
--- /dev/null
+++ b/app/services/conversation.py
@@ -0,0 +1,71 @@
+"""会话存储：创建时即占位，用于校验 ID 由后端签发；当前为进程内存实现"""
+
+from datetime import UTC, datetime, timedelta
+from typing import Protocol
+from uuid import uuid4
+
+from app.config import settings
+from app.schemas.conversation import Turn
+
+
+class ConversationStore(Protocol):
+    """后续接入 MongoDB 时实现同一组方法，业务代码不需要改动"""
+
+    async def create(self) -> str: ...
+
+    async def exists(self, conversation_id: str) -> bool: ...
+
+    async def load_turns(
+        self, conversation_id: str, limit: int | None
+    ) -> list[Turn]: ...
+
+    async def save_turn(self, turn: Turn) -> None: ...
+
+
+def _now() -> datetime:
+    return datetime.now(UTC)
+
+
+class InMemoryConversationStore:
+    def __init__(self, empty_ttl: int | None = None):
+        ttl = empty_ttl if empty_ttl is not None else settings.conversation_empty_ttl
+        self._empty_ttl = timedelta(seconds=ttl)
+        # 会话 ID -> 创建时间；轮次按 thread_id 去重并保持首次写入顺序
+        self._created: dict[str, datetime] = {}
+        self._turns: dict[str, dict[str, Turn]] = {}
+
+    def _purge_expired(self) -> None:
+        # 只清理没有任何轮次且超过保留时长的空会话，有对话的会话不受影响
+        deadline = _now() - self._empty_ttl
+        for conversation_id in [
+            key
+            for key, created in self._created.items()
+            if created < deadline and not self._turns.get(key)
+        ]:
+            del self._created[conversation_id]
+            self._turns.pop(conversation_id, None)
+
+    async def create(self) -> str:
+        self._purge_expired()
+        conversation_id = str(uuid4())
+        self._created[conversation_id] = _now()
+        return conversation_id
+
+    async def exists(self, conversation_id: str) -> bool:
+        self._purge_expired()
+        return conversation_id in self._created
+
+    async def load_turns(self, conversation_id: str, limit: int | None) -> list[Turn]:
+        turns = list(self._turns.get(conversation_id, {}).values())
+        turns.sort(key=lambda turn: turn.created_at)
+        if limit is None:
+            return turns
+        # 只取最近 limit 轮，limit 为 0 时不带历史
+        return turns[-limit:] if limit else []
+
+    async def save_turn(self, turn: Turn) -> None:
+        # 同一任务再次写入（澄清恢复后）保留首次创建时间，只更新内容
+        existing = self._turns.setdefault(turn.conversation_id, {}).get(turn.thread_id)
+        if existing is not None:
+            turn = turn.model_copy(update={"created_at": existing.created_at})
+        self._turns[turn.conversation_id][turn.thread_id] = turn