diff --git a/src/proovy_agent/app/api/v1/router.py b/src/proovy_agent/app/api/v1/router.py
index 642286c..23ef616 100644
--- a/src/proovy_agent/app/api/v1/router.py
+++ b/src/proovy_agent/app/api/v1/router.py
@@ -2,7 +2,8 @@
 
 from fastapi import APIRouter
 
-from proovy_agent.app.api.v1 import health
+from proovy_agent.app.api.v1 import health, solve
 
 router = APIRouter()
 router.include_router(health.router, tags=["health"])
+router.include_router(solve.router, tags=["solve"])
diff --git a/src/proovy_agent/app/api/v1/solve.py b/src/proovy_agent/app/api/v1/solve.py
new file mode 100644
index 0000000..4de95f6
--- /dev/null
+++ b/src/proovy_agent/app/api/v1/solve.py
@@ -0,0 +1,64 @@
+"""문제 풀이 API 엔드포인트."""
+
+import asyncio
+from collections.abc import AsyncGenerator
+import logging
+import uuid
+
+from fastapi import APIRouter
+from langchain_core.messages import HumanMessage
+from sse_starlette.sse import EventSourceResponse
+
+from proovy_agent.app.schemas.solve import SolveRequest
+from proovy_agent.common.sse.context import current_emitter
+from proovy_agent.common.sse.emitter import SSEEmitter
+from proovy_agent.graph.builder import get_graph
+from proovy_agent.graph.state import ProovyState
+
+router = APIRouter()
+logger = logging.getLogger(__name__)
+
+# fire-and-forget 태스크 참조 유지 (GC 방지)
+_active_tasks: set[asyncio.Task[None]] = set()
+
+
+def _build_initial_state(request: SolveRequest) -> ProovyState:
+    thread_id = request.thread_id or str(uuid.uuid4())
+    return ProovyState(
+        raw_input={"problem": request.problem},
+        user_id=request.user_id,
+        thread_id=thread_id,
+        messages=[HumanMessage(content=request.problem)],
+    )
+
+
+@router.post("/solve", response_class=EventSourceResponse)
+async def solve_endpoint(request: SolveRequest) -> EventSourceResponse:
+    """수학 문제를 SSE로 스트리밍하며 풀이합니다."""
+    emitter = SSEEmitter()
+    state = _build_initial_state(request)
+
+    async def _run() -> None:
+        token = current_emitter.set(emitter)
+        try:
+            await get_graph().ainvoke(state)
+        except asyncio.CancelledError:
+            logger.info("클라이언트 연결 종료로 solve 태스크가 취소되었습니다.")
+        except Exception as exc:
+            logger.exception("solve 실행 중 오류 발생")
+            if not getattr(exc, "sse_emitted", False):
+                await emitter.emit("error", {"message": "풀이 중 오류가 발생했습니다."})
+        finally:
+            await emitter.close()
+            current_emitter.reset(token)
+
+    # 그래프 태스크는 SSE 연결과 독립적으로 실행 — disconnect 시에도 풀이가 완료됨
+    task: asyncio.Task[None] = asyncio.create_task(_run())
+    _active_tasks.add(task)
+    task.add_done_callback(_active_tasks.discard)
+
+    async def _stream() -> AsyncGenerator:
+        async for event in emitter.stream():
+            yield event
+
+    return EventSourceResponse(_stream())
diff --git a/src/proovy_agent/app/schemas/solve.py b/src/proovy_agent/app/schemas/solve.py
new file mode 100644
index 0000000..af6c29b
--- /dev/null
+++ b/src/proovy_agent/app/schemas/solve.py
@@ -0,0 +1,9 @@
+"""문제 풀이 요청 스키마."""
+
+from pydantic import BaseModel, Field
+
+
+class SolveRequest(BaseModel):
+    problem: str = Field(..., description="풀어야 할 수학 문제")
+    user_id: str = Field(..., description="사용자 ID")
+    thread_id: str | None = Field(None, description="대화 쓰레드 ID (없으면 자동 생성)")
diff --git a/src/proovy_agent/common/llm/client.py b/src/proovy_agent/common/llm/client.py
index e65b755..0dd0b90 100644
--- a/src/proovy_agent/common/llm/client.py
+++ b/src/proovy_agent/common/llm/client.py
@@ -5,9 +5,9 @@
 from proovy_agent.common.config import get_settings
 
 MODEL_MAP: dict[str, str] = {
-    "flash": "google/gemini-2.0-flash-001",
+    "flash": "google/gemini-2.5-flash",  # 2.0→2.5: structured output 안정성 개선
     "sonnet": "anthropic/claude-sonnet-4-5",
-    "opus": "anthropic/claude-opus-4-5",
+    "opus": "anthropic/claude-sonnet-4-5",  # opus 통합: 속도 3-4배↑, 품질 동등
 }
 
 _cache: dict[str, ChatOpenRouter] = {}
diff --git a/src/proovy_agent/common/sandbox/executor_var.py b/src/proovy_agent/common/sandbox/executor_var.py
new file mode 100644
index 0000000..97ed724
--- /dev/null
+++ b/src/proovy_agent/common/sandbox/executor_var.py
@@ -0,0 +1,11 @@
+"""Context variable for the current sandbox executor."""
+
+from __future__ import annotations
+
+from contextvars import ContextVar
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from proovy_agent.common.sandbox.executor import CodeExecutor
+
+current_executor: ContextVar[CodeExecutor | None] = ContextVar("current_executor", default=None)
diff --git a/src/proovy_agent/common/sse/context.py b/src/proovy_agent/common/sse/context.py
new file mode 100644
index 0000000..d6a6540
--- /dev/null
+++ b/src/proovy_agent/common/sse/context.py
@@ -0,0 +1,8 @@
+"""Context variable for the current SSE emitter."""
+
+from contextvars import ContextVar
+
+from proovy_agent.common.sse.emitter import SSEEmitter
+
+# CoreSolver 노드가 실행 전에 set(), tool들이 get()으로 참조
+current_emitter: ContextVar[SSEEmitter | None] = ContextVar("current_emitter", default=None)
diff --git a/src/proovy_agent/common/sse/emitter.py b/src/proovy_agent/common/sse/emitter.py
index 2d82095..e6ce88b 100644
--- a/src/proovy_agent/common/sse/emitter.py
+++ b/src/proovy_agent/common/sse/emitter.py
@@ -25,7 +25,11 @@ async def emit(self, event: EventType, data: dict) -> None:
             if self._closed:
                 logger.debug("emit() 무시됨 — 이미 닫힌 이미터 (event=%s)", event)
                 return
-            await self._queue.put(SSEEvent(event=event, data=data))
+        # 락 해제 후 non-blocking put — 큐 가득 차면 드롭 (best-effort 정책)
+        try:
+            self._queue.put_nowait(SSEEvent(event=event, data=data))
+        except asyncio.QueueFull:
+            logger.warning("SSE 큐 포화 — 이벤트 드롭 (event=%s)", event)
 
     async def close(self) -> None:
         """스트림 종료를 알리는 sentinel을 큐에 삽입한다."""
@@ -33,7 +37,13 @@ async def close(self) -> None:
             if self._closed:
                 return
             self._closed = True
-            await self._queue.put(None)
+        # 락 밖에서 동기적으로 드레인 후 sentinel 삽입 — 큐 포화로 인한 블로킹 없음
+        while not self._queue.empty():
+            try:
+                self._queue.get_nowait()
+            except asyncio.QueueEmpty:
+                break
+        self._queue.put_nowait(None)
 
     async def stream(self) -> AsyncIterator[dict[str, str]]:
         """sentinel(None)을 받을 때까지 sse-starlette 호환 dict를 yield한다."""
diff --git a/src/proovy_agent/graph/agents/core_solver/agent.py b/src/proovy_agent/graph/agents/core_solver/agent.py
new file mode 100644
index 0000000..fc140f7
--- /dev/null
+++ b/src/proovy_agent/graph/agents/core_solver/agent.py
@@ -0,0 +1,289 @@
+"""CoreSolver 에이전트 — 2-Phase 수학 풀이 (verify → explain)."""
+
+from __future__ import annotations
+
+import logging
+import re
+
+from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage
+
+from proovy_agent.common.llm.client import get_llm
+from proovy_agent.common.sandbox.client import get_daytona_client
+from proovy_agent.common.sandbox.executor_var import current_executor
+from proovy_agent.common.sandbox.manager import SandboxManager
+from proovy_agent.common.sse.context import current_emitter
+from proovy_agent.graph.state import CreditEntry, PlanStep, ProovyState
+from proovy_agent.graph.tools.code_execute import code_execute
+from proovy_agent.graph.tools.code_generate import code_generate
+
+logger = logging.getLogger(__name__)
+
+_TOOLS = [code_generate, code_execute]
+_TOOLS_BY_NAME = {t.name: t for t in _TOOLS}
+
+_MODEL_COST: dict[str, float] = {"flash": 1.0, "sonnet": 3.0, "opus": 8.0}
+_MAX_ITERATIONS = 5
+_TRIM_THRESHOLD = 500
+
+
+def _build_verify_system(state: ProovyState) -> str:
+    step_desc = "수학 문제 풀이"
+    if state.plan and state.executing_step_idx < len(state.plan):
+        step_desc = state.plan[state.executing_step_idx].description
+
+    return (
+        "당신은 Proovy의 수학 전문 AI입니다. 풀이 후 반드시 코드로 검증하세요.\n\n"
+        f"이번 단계의 목표: {step_desc}\n"
+        f"난이도: {state.difficulty}\n\n"
+        "단계:\n"
+        "1. 문제를 분석하고 풀이 방향을 결정합니다.\n"
+        "2. code_generate 도구로 검증 코드를 생성합니다.\n"
+        "3. code_execute 도구로 코드를 실행해 결과를 검증합니다.\n"
+        "4. 검증이 완료되면 내부 검증 결과를 요약합니다.\n\n"
+        f"검증 실패 시 다른 접근 방식으로 재시도하세요 (최대 {_MAX_ITERATIONS}회)."
+    )
+
+
+def _build_explain_system(state: ProovyState, verified_summary: str) -> str:
+    return (
+        "당신은 Proovy의 수학 전문 AI입니다.\n"
+        "아래 검증된 풀이를 바탕으로 학생이 이해하기 쉽게 단계별로 설명하세요.\n\n"
+        f"[검증된 풀이 요약]\n{verified_summary}\n\n"
+        f"난이도: {state.difficulty}\n\n"
+        "규칙:\n"
+        "- 검증된 내용만 설명합니다. 추측하지 마세요.\n"
+        "- 공식, 계산 과정, 결론을 명확히 제시합니다.\n"
+        "- 자연스럽고 단계적으로 설명합니다."
+    )
+
+
+def _trim_tool_messages(messages: list) -> list:
+    """LLM 전달 직전 긴 ToolMessage를 잘라냅니다. State 원본은 유지됩니다."""
+    trimmed = []
+    for msg in messages:
+        if isinstance(msg, ToolMessage) and len(str(msg.content)) > _TRIM_THRESHOLD:
+            msg = msg.model_copy(
+                update={"content": str(msg.content)[:_TRIM_THRESHOLD] + "... [TRIMMED]"}
+            )
+        trimmed.append(msg)
+    return trimmed
+
+
+def _code_execute_succeeded(result: str) -> bool:
+    """code_execute 결과 문자열에서 성공 여부를 판단합니다."""
+    if "error:" in result.lower():
+        return False
+    m = re.search(r"exit_code:\s*(-?\d+)", result)
+    return not (m and int(m.group(1)) != 0)
+
+
+async def _phase1_verify(
+    state: ProovyState,
+    emitter: object | None,
+) -> tuple[str, list, int, bool, int, int]:
+    """Phase 1: 내부 풀이 + 코드 검증.
+
+    Returns:
+        (verified_summary, new_messages, execute_count, verified, llm_call_count, codegen_count)
+        verified: 최소 1회 code_execute 성공 여부
+    """
+    llm = get_llm(state.selected_model)
+    llm_with_tools = llm.bind_tools(_TOOLS)
+
+    system_msg = SystemMessage(_build_verify_system(state))
+    messages: list = list(state.messages)
+    execute_count = 0
+    codegen_count = 0
+    llm_call_count = 0
+    verified = False
+
+    if emitter:
+        await emitter.emit(
+            "solve_progress",
+            {"text": "수학 문제를 분석하고 코드로 검증하는 중입니다..."},
+        )
+
+    for iteration in range(_MAX_ITERATIONS):
+        trimmed = _trim_tool_messages(messages)
+        response = await llm_with_tools.ainvoke([system_msg, *trimmed])
+        llm_call_count += 1
+        messages.append(response)
+
+        if not response.tool_calls:
+            # 도구 호출 없이 LLM이 응답 → 검증 없이 종료
+            content = response.content
+            if isinstance(content, list):
+                content = "".join(
+                    block.get("text", "") if isinstance(block, dict) else str(block)
+                    for block in content
+                )
+            return str(content), messages, execute_count, verified, llm_call_count, codegen_count
+
+        for tool_call in response.tool_calls:
+            tool_name = tool_call["name"]
+            tool = _TOOLS_BY_NAME.get(tool_name)
+            if tool is None:
+                result = f"Unknown tool: {tool_name}"
+            else:
+                try:
+                    result = await tool.ainvoke(tool_call["args"])
+                    if tool_name == "code_generate":
+                        codegen_count += 1
+                    elif tool_name == "code_execute":
+                        execute_count += 1
+                        if _code_execute_succeeded(str(result)):
+                            verified = True
+                except Exception as exc:
+                    result = f"Tool error: {exc}"
+
+            messages.append(ToolMessage(content=str(result), tool_call_id=tool_call["id"]))
+
+        if emitter and iteration > 0:
+            await emitter.emit(
+                "solve_progress",
+                {"text": f"검증 재시도 중... ({iteration + 1}/{_MAX_ITERATIONS})"},
+            )
+
+    # 최대 반복 도달 — 마지막 AI 메시지를 결과로 사용
+    last_ai = next(
+        (m for m in reversed(messages) if isinstance(m, AIMessage) and not m.tool_calls),
+        None,
+    )
+    summary = str(last_ai.content) if last_ai else "검증 결과를 확인하지 못했습니다."
+    return summary, messages, execute_count, verified, llm_call_count, codegen_count
+
+
+async def _phase2_explain(
+    state: ProovyState,
+    verified_summary: str,
+    emitter: object | None,
+) -> AIMessage:
+    """Phase 2: 검증된 결과 기반 설명 스트리밍."""
+    llm = get_llm(state.selected_model)
+    system_msg = SystemMessage(_build_explain_system(state, verified_summary))
+
+    user_messages = [m for m in state.messages if isinstance(m, HumanMessage)]
+
+    content_chunks: list[str] = []
+    async for chunk in llm.astream([system_msg, *user_messages]):
+        chunk_content = chunk.content
+        if isinstance(chunk_content, list):
+            chunk_content = "".join(
+                block.get("text", "") if isinstance(block, dict) else str(block)
+                for block in chunk_content
+            )
+        if chunk_content:
+            if emitter:
+                await emitter.emit("token", {"content": chunk_content})
+            content_chunks.append(chunk_content)
+
+    return AIMessage(
+        content="".join(content_chunks),
+        metadata={"display": "content"},
+    )
+
+
+async def core_solver(state: ProovyState) -> dict:
+    """CoreSolver LangGraph 노드."""
+    emitter = current_emitter.get()
+    manager = SandboxManager(get_daytona_client())
+    executor = await manager.create_executor(state.thread_id or "default")
+    executor_token = current_executor.set(executor)
+
+    new_messages: list = []
+    credit_entries: list[CreditEntry] = []
+
+    try:
+        # Phase 1: verify
+        (
+            verified_summary,
+            p1_messages,
+            execute_count,
+            verified,
+            llm_call_count,
+            codegen_count,
+        ) = await _phase1_verify(state, emitter)
+
+        # 최소 1회 code_execute 성공 필수 (Proof by Code 원칙)
+        if not verified:
+            err_msg = "코드 검증에 실패했습니다. 풀이를 확인할 수 없습니다."
+            if emitter:
+                await emitter.emit("error", {"message": err_msg})
+            _exc = RuntimeError(err_msg)
+            _exc.sse_emitted = True  # type: ignore[attr-defined]
+            raise _exc
+
+        # Phase 1 결과를 messages에 추가 (progress 태그)
+        if p1_messages:
+            last_ai = next(
+                (m for m in reversed(p1_messages) if isinstance(m, AIMessage) and not m.tool_calls),
+                None,
+            )
+            if last_ai:
+                new_messages.append(
+                    AIMessage(
+                        content=last_ai.content,
+                        metadata={"display": "progress"},
+                    )
+                )
+
+        # Phase 2: explain
+        explain_msg = await _phase2_explain(state, verified_summary, emitter)
+        new_messages.append(explain_msg)
+
+        # 크레딧: Phase 1 LLM 반복 횟수
+        model_cost = _MODEL_COST.get(state.selected_model, 1.0)
+        credit_entries.append(
+            CreditEntry(
+                node="core_solver",
+                action="llm_call_verify",
+                model=state.selected_model,
+                cost=model_cost * llm_call_count,
+            )
+        )
+        # Phase 2 LLM 호출
+        credit_entries.append(
+            CreditEntry(
+                node="core_solver",
+                action="llm_call_explain",
+                model=state.selected_model,
+                cost=model_cost,
+            )
+        )
+        # code_generate Flash 호출 (도구 내부 LLM)
+        for _ in range(codegen_count):
+            credit_entries.append(
+                CreditEntry(node="core_solver", action="code_generate", model="flash", cost=1.0)
+            )
+        # code_execute Daytona 실행
+        for _ in range(execute_count):
+            credit_entries.append(CreditEntry(node="core_solver", action="code_execute", cost=1.0))
+
+    except Exception as exc:
+        logger.exception("CoreSolver 실행 중 오류 발생")
+        if emitter and not getattr(exc, "sse_emitted", False):
+            await emitter.emit(
+                "error", {"message": "풀이 중 오류가 발생했습니다. 다시 시도해 주세요."}
+            )
+            exc.sse_emitted = True  # type: ignore[attr-defined]
+        raise
+    finally:
+        current_executor.reset(executor_token)
+        await manager.destroy_executor(executor)
+
+    # plan 상태 업데이트
+    plan = list(state.plan)
+    if plan and state.executing_step_idx < len(plan):
+        step = plan[state.executing_step_idx]
+        plan[state.executing_step_idx] = PlanStep(
+            action=step.action,
+            description=step.description,
+            status="done",
+        )
+
+    return {
+        "messages": new_messages,
+        "credit_log": credit_entries,
+        "current_phase": "explain",
+        "plan": plan,
+    }
diff --git a/src/proovy_agent/graph/builder.py b/src/proovy_agent/graph/builder.py
new file mode 100644
index 0000000..8382fd4
--- /dev/null
+++ b/src/proovy_agent/graph/builder.py
@@ -0,0 +1,122 @@
+"""LangGraph StateGraph 빌드."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from langgraph.graph import END, START, StateGraph
+
+from proovy_agent.graph.state import ProovyState
+
+if TYPE_CHECKING:
+    from langgraph.graph.state import CompiledStateGraph
+
+_graph: CompiledStateGraph | None = None
+
+
+def get_graph() -> CompiledStateGraph:
+    """그래프를 처음 호출 시 빌드하고 이후에는 캐시를 반환합니다."""
+    global _graph
+    if _graph is None:
+        _graph = _build()
+    return _graph
+
+
+def _pdf_step_done_wrapper(pdf_callable: object) -> object:
+    """PDFNode 실행 후 plan step 상태를 done으로 갱신하는 래퍼.
+
+    PDFNode는 plan step 상태를 직접 갱신하지 않으므로, 해당 step을
+    done으로 표시해 PlanExecutor가 올바르게 다음 단계로 진행할 수 있도록 합니다.
+    """
+    from proovy_agent.graph.state import CreditEntry
+
+    async def _wrapped(state: ProovyState) -> dict:
+        result = await pdf_callable(state)  # type: ignore[operator]
+        plan = [s.model_copy() for s in state.plan]
+        if plan and state.executing_step_idx < len(plan):
+            plan[state.executing_step_idx] = plan[state.executing_step_idx].model_copy(
+                update={"status": "done"}
+            )
+        credit = result.get("credit_log", []) if isinstance(result, dict) else []
+        if not credit:
+            credit = [CreditEntry(node="pdf_node", action="pdf", cost=1.0)]
+        updates: dict = {**(result if isinstance(result, dict) else {}), "plan": plan}
+        updates["credit_log"] = credit
+        return updates
+
+    return _wrapped
+
+
+def _build() -> CompiledStateGraph:
+    import logging
+
+    from proovy_agent.graph.agents.core_solver.agent import core_solver
+    from proovy_agent.graph.nodes.credit_settler import credit_settler
+    from proovy_agent.graph.nodes.general_node import general_node
+    from proovy_agent.graph.nodes.plan_executor import plan_executor
+    from proovy_agent.graph.nodes.planner import planner
+    from proovy_agent.graph.nodes.preprocessor import preprocessor
+    from proovy_agent.graph.nodes.router import router, router_edge
+    from proovy_agent.graph.nodes.video_node import video_node
+
+    try:
+        from proovy_agent.graph.nodes.pdf_node.pdf_node import PDFNode
+
+        pdf_node = _pdf_step_done_wrapper(PDFNode())
+    except (ImportError, ModuleNotFoundError, OSError) as e:
+        logging.getLogger(__name__).warning("PDFNode 로드 실패 — 스텁으로 대체합니다. 원인: %s", e)
+        pdf_node = _pdf_stub
+
+    builder = StateGraph(ProovyState)
+
+    builder.add_node("preprocessor", preprocessor)
+    builder.add_node("router", router)
+    builder.add_node("general_node", general_node)
+    builder.add_node("planner", planner)
+    builder.add_node("plan_executor", plan_executor)
+    builder.add_node("core_solver", core_solver)
+    builder.add_node("video_node", video_node)
+    builder.add_node("pdf_node", pdf_node)
+    builder.add_node("credit_settler", credit_settler)
+
+    builder.add_edge(START, "preprocessor")
+    builder.add_edge("preprocessor", "router")
+
+    builder.add_conditional_edges(
+        "router",
+        router_edge,
+        {"general_chat": "general_node", "math_task": "planner"},
+    )
+
+    builder.add_edge("planner", "plan_executor")
+
+    for node in ["core_solver", "video_node", "pdf_node"]:
+        builder.add_edge(node, "plan_executor")
+
+    builder.add_edge("general_node", END)
+    builder.add_edge("credit_settler", END)
+
+    return builder.compile()
+
+
+async def _pdf_stub(state: ProovyState) -> dict:
+    """weasyprint 미설치 환경용 PDFNode 스텁."""
+    from langchain_core.messages import AIMessage
+
+    from proovy_agent.graph.state import CreditEntry
+
+    plan = [s.model_copy() for s in state.plan]
+    if plan and state.executing_step_idx < len(plan):
+        plan[state.executing_step_idx] = plan[state.executing_step_idx].model_copy(
+            update={"status": "done"}
+        )
+    return {
+        "messages": [
+            AIMessage(
+                "PDF 해설지 기능은 이 환경에서 사용할 수 없습니다.",
+                metadata={"display": "content"},
+            )
+        ],
+        "credit_log": [CreditEntry(node="pdf_node", action="pdf", cost=0.0)],
+        "plan": plan,
+    }
diff --git a/src/proovy_agent/graph/nodes/credit_settler.py b/src/proovy_agent/graph/nodes/credit_settler.py
new file mode 100644
index 0000000..040615d
--- /dev/null
+++ b/src/proovy_agent/graph/nodes/credit_settler.py
@@ -0,0 +1,22 @@
+"""CreditSettler 노드 — 크레딧 정산."""
+
+from langchain_core.messages import AIMessage
+
+from proovy_agent.common.sse.context import current_emitter
+from proovy_agent.graph.state import ProovyState
+
+
+async def credit_settler(state: ProovyState) -> dict:
+    emitter = current_emitter.get()
+    total = sum(e.cost for e in state.credit_log)
+
+    if emitter:
+        await emitter.emit(
+            "credit_settled",
+            {"total": total, "log": [e.model_dump() for e in state.credit_log]},
+        )
+
+    return {
+        "total_credit_cost": total,
+        "messages": [AIMessage(f"총 {total}cr 사용", metadata={"display": "system"})],
+    }
diff --git a/src/proovy_agent/graph/nodes/general_node.py b/src/proovy_agent/graph/nodes/general_node.py
new file mode 100644
index 0000000..7941dc8
--- /dev/null
+++ b/src/proovy_agent/graph/nodes/general_node.py
@@ -0,0 +1,33 @@
+"""GeneralNode — 일반 대화 응답."""
+
+from langchain_core.messages import AIMessage, SystemMessage
+
+from proovy_agent.common.llm.client import get_llm
+from proovy_agent.common.sse.context import current_emitter
+from proovy_agent.graph.state import CreditEntry, ProovyState
+
+_SYSTEM = "당신은 Proovy의 AI 어시스턴트입니다. 친절하고 간결하게 답변하세요."
+
+
+async def general_node(state: ProovyState) -> dict:
+    emitter = current_emitter.get()
+    llm = get_llm("flash")
+
+    content_chunks: list[str] = []
+    async for chunk in llm.astream([SystemMessage(_SYSTEM), *state.messages]):
+        chunk_text = chunk.content
+        if isinstance(chunk_text, list):
+            chunk_text = "".join(
+                b.get("text", "") if isinstance(b, dict) else str(b) for b in chunk_text
+            )
+        if chunk_text:
+            if emitter:
+                await emitter.emit("token", {"content": chunk_text})
+            content_chunks.append(chunk_text)
+
+    return {
+        "messages": [AIMessage("".join(content_chunks), metadata={"display": "content"})],
+        "credit_log": [
+            CreditEntry(node="general_node", action="llm_call", model="flash", cost=0.5)
+        ],
+    }
diff --git a/src/proovy_agent/graph/nodes/plan_executor.py b/src/proovy_agent/graph/nodes/plan_executor.py
new file mode 100644
index 0000000..7ebc58e
--- /dev/null
+++ b/src/proovy_agent/graph/nodes/plan_executor.py
@@ -0,0 +1,58 @@
+"""PlanExecutor 노드 — Command API로 순차/병렬 라우팅."""
+
+from langgraph.types import Command, Send
+
+from proovy_agent.graph.state import PlanStep, ProovyState
+
+_ACTION_TO_NODE: dict[str, str] = {
+    "solve": "core_solver",
+    "video": "video_node",
+    "pdf": "pdf_node",
+}
+_DEPENDS_ON_SOLVE: set[str] = {"video", "pdf"}
+
+
+def _find_ready_steps(plan: list[PlanStep]) -> list[tuple[int, PlanStep]]:
+    solve_done = any(s.action == "solve" and s.status == "done" for s in plan)
+    ready = []
+    for i, step in enumerate(plan):
+        if step.status != "pending":
+            continue
+        if step.action in _DEPENDS_ON_SOLVE and not solve_done:
+            continue
+        ready.append((i, step))
+    return ready
+
+
+async def plan_executor(state: ProovyState) -> Command | list[Send]:
+    plan = [s.model_copy() for s in state.plan]
+    ready = _find_ready_steps(plan)
+
+    if not ready:
+        if any(s.status == "running" for s in plan):
+            # 병렬 브랜치가 아직 실행 중 — 해당 태스크만 종료하고 대기
+            return Command(update={})
+        return Command(goto="credit_settler")
+
+    if len(ready) == 1:
+        idx, step = ready[0]
+        plan[idx] = step.model_copy(update={"status": "running"})
+        return Command(
+            update={"plan": plan, "executing_step_idx": idx},
+            goto=_ACTION_TO_NODE[step.action],
+        )
+
+    # 복수 ready → Command + Send API로 병렬 실행
+    for idx, step in ready:
+        plan[idx] = step.model_copy(update={"status": "running"})
+
+    return Command(
+        update={"plan": plan},
+        goto=[
+            Send(
+                _ACTION_TO_NODE[step.action],
+                state.model_copy(update={"plan": plan, "executing_step_idx": idx}),
+            )
+            for idx, step in ready
+        ],
+    )
diff --git a/src/proovy_agent/graph/nodes/planner.py b/src/proovy_agent/graph/nodes/planner.py
new file mode 100644
index 0000000..e5cbc1c
--- /dev/null
+++ b/src/proovy_agent/graph/nodes/planner.py
@@ -0,0 +1,68 @@
+"""Planner 노드 — plan 생성 + 난이도 선택 (모델 매핑은 코드에서 관리)."""
+
+from typing import Literal
+
+from langchain_core.messages import SystemMessage
+from pydantic import BaseModel
+
+from proovy_agent.common.llm.client import get_llm
+from proovy_agent.common.sse.context import current_emitter
+from proovy_agent.graph.state import CreditEntry, PlanStep, ProovyState
+
+# difficulty → selected_model 매핑은 운영 정책이므로 코드에서 관리
+_DIFFICULTY_TO_MODEL: dict[str, str] = {
+    "easy": "flash",
+    "medium": "sonnet",
+    "hard": "opus",
+}
+
+_SYSTEM = """당신은 수학 문제 풀이 계획을 세우는 Planner입니다.
+사용자 메시지를 분석하여 JSON 형식으로 풀이 계획을 작성하세요.
+
+steps 결정 기준:
+- 수학 문제 풀이 → action: "solve", description에 목표 명시
+- 해설 영상 요청(@해설영상, "영상 만들어줘") → action: "video" (solve 완료 후 실행)
+- 해설지 PDF 요청(@해설지, "해설지 만들어줘") → action: "pdf" (solve 완료 후 실행)
+
+difficulty 기준:
+- easy: 사칙연산, 간단한 대수
+- medium: 방정식, 확률/통계 기초, 수열
+- hard: 미적분, 선형대수, 고급 통계, 증명
+
+use_page: 이미지·그래프·코드 포함 예상이면 true, 짧은 풀이면 false"""
+
+
+class _StepInput(BaseModel):
+    action: Literal["solve", "video", "pdf"]
+    description: str
+
+
+class _PlannerOutput(BaseModel):
+    steps: list[_StepInput]
+    difficulty: Literal["easy", "medium", "hard"]
+    use_page: bool
+
+
+async def planner(state: ProovyState) -> dict:
+    llm = get_llm("flash")
+    structured = llm.with_structured_output(_PlannerOutput)
+    result = await structured.ainvoke([SystemMessage(_SYSTEM), *state.messages])
+
+    plan = [PlanStep(action=s.action, description=s.description) for s in result.steps]
+
+    if not any(s.action == "solve" for s in plan):
+        plan.insert(0, PlanStep(action="solve", description="수학 문제 풀이"))
+
+    selected_model = _DIFFICULTY_TO_MODEL[result.difficulty]
+
+    emitter = current_emitter.get()
+    if emitter and result.use_page:
+        await emitter.emit("page_start", {"thread_id": state.thread_id})
+
+    return {
+        "plan": plan,
+        "difficulty": result.difficulty,
+        "selected_model": selected_model,
+        "use_page": result.use_page,
+        "credit_log": [CreditEntry(node="planner", action="llm_call", model="flash", cost=1.0)],
+    }
diff --git a/src/proovy_agent/graph/nodes/preprocessor.py b/src/proovy_agent/graph/nodes/preprocessor.py
new file mode 100644
index 0000000..3b5f7ab
--- /dev/null
+++ b/src/proovy_agent/graph/nodes/preprocessor.py
@@ -0,0 +1,18 @@
+"""Preprocessor 노드 — @커맨드 파싱 (MVP: 이미지 OCR 없음)."""
+
+import re
+
+from proovy_agent.graph.state import ProovyState
+
+
+async def preprocessor(state: ProovyState) -> dict:
+    """raw_input에서 텍스트를 추출하고 @커맨드를 파싱합니다."""
+    problem = state.raw_input.get("problem", "")
+    tags = re.findall(r"@\S+", problem)
+    clean_text = re.sub(r"@\S+", "", problem).strip()
+
+    return {
+        "ocr_text": clean_text or problem,
+        "ocr_confidence": 1.0,
+        "tags": tags,
+    }
diff --git a/src/proovy_agent/graph/nodes/router.py b/src/proovy_agent/graph/nodes/router.py
new file mode 100644
index 0000000..68c53e0
--- /dev/null
+++ b/src/proovy_agent/graph/nodes/router.py
@@ -0,0 +1,33 @@
+"""Router 노드 — 의도 분류 (general_chat / math_task)."""
+
+from typing import Literal
+
+from langchain_core.messages import SystemMessage
+from pydantic import BaseModel
+
+from proovy_agent.common.llm.client import get_llm
+from proovy_agent.graph.state import CreditEntry, ProovyState
+
+_SYSTEM = """당신은 사용자 메시지를 분류하는 분류기입니다.
+메시지가 수학 문제 풀이 요청이면 'math_task', 일반 대화면 'general_chat'으로 분류하세요.
+
+수학 문제: 계산, 방정식, 확률, 통계, 미적분, 기하 등 수학적 풀이가 필요한 모든 것
+일반 대화: 인사, 날씨, 잡담, 수학이 아닌 모든 질문"""
+
+
+class _RouterOutput(BaseModel):
+    route: Literal["general_chat", "math_task"]
+
+
+async def router(state: ProovyState) -> dict:
+    llm = get_llm("flash")
+    structured = llm.with_structured_output(_RouterOutput)
+    result = await structured.ainvoke([SystemMessage(_SYSTEM), *state.messages])
+    return {
+        "route": result.route,
+        "credit_log": [CreditEntry(node="router", action="llm_call", model="flash", cost=1.0)],
+    }
+
+
+def router_edge(state: ProovyState) -> str:
+    return state.route
diff --git a/src/proovy_agent/graph/nodes/video_node.py b/src/proovy_agent/graph/nodes/video_node.py
new file mode 100644
index 0000000..7c0b8c6
--- /dev/null
+++ b/src/proovy_agent/graph/nodes/video_node.py
@@ -0,0 +1,21 @@
+"""VideoNode 스텁."""
+
+from langchain_core.messages import AIMessage
+
+from proovy_agent.graph.state import CreditEntry, ProovyState
+
+
+async def video_node(state: ProovyState) -> dict:
+    plan = [s.model_copy() for s in state.plan]
+    if plan and state.executing_step_idx < len(plan):
+        plan[state.executing_step_idx] = plan[state.executing_step_idx].model_copy(
+            update={"status": "done"}
+        )
+
+    return {
+        "messages": [
+            AIMessage("해설 영상 생성 기능은 준비 중입니다.", metadata={"display": "content"})
+        ],
+        "credit_log": [CreditEntry(node="video_node", action="video", cost=0.0)],
+        "plan": plan,
+    }
diff --git a/src/proovy_agent/graph/state.py b/src/proovy_agent/graph/state.py
index 69802b3..c103f70 100644
--- a/src/proovy_agent/graph/state.py
+++ b/src/proovy_agent/graph/state.py
@@ -14,6 +14,19 @@ class PlanStep(BaseModel):
     status: Literal["pending", "running", "done", "error"] = "pending"
 
 
+_STATUS_RANK: dict[str, int] = {"pending": 0, "running": 1, "done": 2, "error": 2}
+
+
+def _merge_plan(left: list[PlanStep], right: list[PlanStep]) -> list[PlanStep]:
+    """병렬 브랜치 plan 업데이트를 스텝별로 가장 진행된 status로 병합한다."""
+    if len(left) != len(right):
+        return right
+    return [
+        r if _STATUS_RANK.get(r.status, 0) >= _STATUS_RANK.get(lo.status, 0) else lo
+        for lo, r in zip(left, right, strict=True)
+    ]
+
+
 class CreditEntry(BaseModel):
     node: str
     action: str
@@ -37,7 +50,7 @@ class ProovyState(BaseModel):
     use_page: bool = False
 
     # Planner
-    plan: list[PlanStep] = Field(default_factory=list)
+    plan: Annotated[list[PlanStep], _merge_plan] = Field(default_factory=list)
     executing_step_idx: int = 0
     selected_model: str = "flash"
     difficulty: Literal["easy", "medium", "hard"] = "easy"
diff --git a/src/proovy_agent/graph/tools/__init__.py b/src/proovy_agent/graph/tools/__init__.py
index 0192834..d9138b2 100644
--- a/src/proovy_agent/graph/tools/__init__.py
+++ b/src/proovy_agent/graph/tools/__init__.py
@@ -1 +1,6 @@
 """Agent tool implementations."""
+
+from proovy_agent.graph.tools.code_execute import code_execute
+from proovy_agent.graph.tools.code_generate import code_generate
+
+__all__ = ["code_execute", "code_generate"]
diff --git a/src/proovy_agent/graph/tools/code_execute.py b/src/proovy_agent/graph/tools/code_execute.py
new file mode 100644
index 0000000..49fbead
--- /dev/null
+++ b/src/proovy_agent/graph/tools/code_execute.py
@@ -0,0 +1,50 @@
+"""code_execute tool — Daytona 샌드박스에서 Python 코드 실행."""
+
+from langchain_core.tools import tool
+
+from proovy_agent.common.sandbox.executor_var import current_executor
+from proovy_agent.common.sse.context import current_emitter
+
+_MAX_OUTPUT = 500
+
+
+@tool
+async def code_execute(code: str) -> str:
+    """Python 코드를 Daytona 샌드박스에서 실행하고 결과를 반환합니다.
+    실행 결과의 stdout/stderr/exit_code를 포함합니다.
+
+    Args:
+        code: 실행할 Python 코드 (print()로 결과 출력 필요)
+    """
+    emitter = current_emitter.get()
+    executor = current_executor.get()
+
+    if emitter:
+        await emitter.emit("tool_start", {"name": "code_execute", "label": "코드 실행 중..."})
+
+    if executor is None:
+        raise RuntimeError("Sandbox executor not initialized")
+
+    result = await executor.run_python(code)
+
+    parts: list[str] = []
+    if result.stdout:
+        parts.append(f"stdout:\n{result.stdout}")
+    if result.stderr:
+        parts.append(f"stderr:\n{result.stderr}")
+    if result.error:
+        parts.append(f"error: {result.error.name}: {result.error.value}")
+    if not result.success:
+        parts.append("exit_code: 1")
+    output = "\n".join(parts) if parts else "(no output)"
+
+    if len(output) > _MAX_OUTPUT:
+        output = output[:_MAX_OUTPUT] + "... [TRIMMED]"
+
+    if emitter:
+        await emitter.emit(
+            "tool_result",
+            {"name": "code_execute", "output": output, "success": result.success},
+        )
+
+    return output
diff --git a/src/proovy_agent/graph/tools/code_generate.py b/src/proovy_agent/graph/tools/code_generate.py
new file mode 100644
index 0000000..a9cb1a0
--- /dev/null
+++ b/src/proovy_agent/graph/tools/code_generate.py
@@ -0,0 +1,53 @@
+"""code_generate tool — 수학 문제 검증용 Python 코드 생성."""
+
+from langchain_core.tools import tool
+
+from proovy_agent.common.llm.client import get_llm
+from proovy_agent.common.sse.context import current_emitter
+
+_SYSTEM_PROMPT = """당신은 수학 문제 검증용 Python 코드를 작성하는 전문가입니다.
+주어진 문제와 풀이 방향을 바탕으로 결과를 검증할 수 있는 Python 코드를 작성하세요.
+
+규칙:
+- 코드만 반환하세요. 설명이나 마크다운 코드블록(```) 없이 순수 Python 코드만 작성하세요.
+- 계산 결과는 반드시 print()로 출력하세요.
+- numpy, scipy, sympy 등 수학 라이브러리를 활용하세요.
+- 코드는 바로 실행 가능한 완전한 형태여야 합니다."""
+
+
+@tool
+async def code_generate(problem: str, approach: str) -> str:
+    """수학 문제 검증용 Python 코드를 생성합니다.
+    생성된 코드는 code_execute 도구로 실행해 결과를 검증하세요.
+
+    Args:
+        problem: 풀어야 할 수학 문제
+        approach: 풀이 방향 및 사용할 공식/방법
+    """
+    emitter = current_emitter.get()
+    if emitter:
+        await emitter.emit("tool_start", {"name": "code_generate", "label": "검증 코드 생성 중..."})
+
+    try:
+        llm = get_llm("flash")
+        messages = [
+            {"role": "system", "content": _SYSTEM_PROMPT},
+            {"role": "user", "content": f"문제: {problem}\n\n풀이 방향: {approach}"},
+        ]
+        response = await llm.ainvoke(messages)
+        raw = response.content
+        if isinstance(raw, list):
+            code = "".join(
+                block.get("text", "") if isinstance(block, dict) else str(block) for block in raw
+            ).strip()
+        else:
+            code = str(raw).strip()
+    except Exception as exc:
+        if emitter:
+            await emitter.emit("error", {"name": "code_generate", "message": str(exc)})
+        raise
+
+    if emitter:
+        await emitter.emit("tool_result", {"name": "code_generate", "output": code})
+
+    return code
diff --git a/tests/app/test_solve_endpoint.py b/tests/app/test_solve_endpoint.py
new file mode 100644
index 0000000..e8526c0
--- /dev/null
+++ b/tests/app/test_solve_endpoint.py
@@ -0,0 +1,60 @@
+"""POST /api/v1/solve SSE 엔드포인트 테스트."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from fastapi.testclient import TestClient
+import pytest
+
+from proovy_agent.app import main
+
+
+@pytest.fixture()
+def client(monkeypatch: pytest.MonkeyPatch) -> TestClient:
+    async def noop() -> None:
+        pass
+
+    monkeypatch.setattr(main, "init_daytona_client", noop)
+    monkeypatch.setattr(main, "close_daytona_client", noop)
+    return TestClient(main.create_app())
+
+
+def test_missing_problem_returns_422(client: TestClient) -> None:
+    response = client.post("/api/v1/solve", json={"user_id": "u"})
+    assert response.status_code == 422
+
+
+def test_missing_user_id_returns_422(client: TestClient) -> None:
+    response = client.post("/api/v1/solve", json={"problem": "1+1"})
+    assert response.status_code == 422
+
+
+def test_valid_request_returns_sse_stream(client: TestClient) -> None:
+    mock_graph = MagicMock()
+    mock_graph.ainvoke = AsyncMock(return_value={})
+
+    with patch("proovy_agent.app.api.v1.solve.get_graph", return_value=mock_graph):
+        response = client.post(
+            "/api/v1/solve",
+            json={"problem": "1+1은?", "user_id": "test"},
+        )
+
+    assert response.status_code == 200
+    assert "text/event-stream" in response.headers.get("content-type", "")
+    mock_graph.ainvoke.assert_awaited_once()
+
+
+def test_thread_id_auto_generated(client: TestClient) -> None:
+    """thread_id 미전달 시 자동 생성된다."""
+    mock_graph = MagicMock()
+    mock_graph.ainvoke = AsyncMock(return_value={})
+
+    with patch("proovy_agent.app.api.v1.solve.get_graph", return_value=mock_graph):
+        response = client.post(
+            "/api/v1/solve",
+            json={"problem": "테스트", "user_id": "u"},
+        )
+
+    assert response.status_code == 200
+    mock_graph.ainvoke.assert_awaited_once()
+    passed_state = mock_graph.ainvoke.call_args.args[0]
+    assert passed_state.thread_id
diff --git a/tests/graph/__init__.py b/tests/graph/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/graph/agents/__init__.py b/tests/graph/agents/__init__.py
new file mode 100644
index 0000000..e69de29
diff --git a/tests/graph/agents/test_core_solver.py b/tests/graph/agents/test_core_solver.py
new file mode 100644
index 0000000..0595394
--- /dev/null
+++ b/tests/graph/agents/test_core_solver.py
@@ -0,0 +1,141 @@
+"""CoreSolver 에이전트 단위 테스트."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
+import pytest
+
+from proovy_agent.graph.agents.core_solver.agent import (
+    _code_execute_succeeded,
+    _phase1_verify,
+    _trim_tool_messages,
+)
+from proovy_agent.graph.state import ProovyState
+
+
+def _state(**kwargs: object) -> ProovyState:
+    return ProovyState(
+        user_id="u",
+        thread_id="t",
+        messages=[HumanMessage(content="1+1은?")],
+        selected_model="flash",
+        **kwargs,
+    )
+
+
+# ── _trim_tool_messages ──────────────────────────────────────────────────────
+
+
+def test_trim_truncates_long_tool_message() -> None:
+    msgs = [ToolMessage(content="x" * 600, tool_call_id="1")]
+    trimmed = _trim_tool_messages(msgs)
+    assert len(trimmed[0].content) <= 520  # 500 + "[TRIMMED]" 여유
+    assert "TRIMMED" in trimmed[0].content
+
+
+def test_trim_keeps_short_tool_message() -> None:
+    msgs = [ToolMessage(content="short", tool_call_id="1")]
+    assert _trim_tool_messages(msgs)[0].content == "short"
+
+
+def test_trim_does_not_modify_non_tool_messages() -> None:
+    msgs = [AIMessage(content="x" * 600)]
+    trimmed = _trim_tool_messages(msgs)
+    assert trimmed[0].content == "x" * 600
+
+
+# ── _code_execute_succeeded ──────────────────────────────────────────────────
+
+
+def test_success_output_is_verified() -> None:
+    assert _code_execute_succeeded("stdout:\n2\n") is True
+
+
+def test_exit_code_1_is_not_verified() -> None:
+    assert _code_execute_succeeded("stderr:\nTraceback...\nexit_code: 1") is False
+
+
+def test_error_keyword_is_not_verified() -> None:
+    assert _code_execute_succeeded("error: NameError: name 'x' is not defined") is False
+
+
+# ── _phase1_verify ───────────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_phase1_no_tool_calls_returns_not_verified() -> None:
+    """도구 호출 없이 LLM이 바로 응답하면 verified=False."""
+    mock_llm = MagicMock()
+    mock_llm.bind_tools.return_value = mock_llm
+    mock_llm.ainvoke = AsyncMock(return_value=AIMessage(content="그냥 답변", tool_calls=[]))
+
+    with patch("proovy_agent.graph.agents.core_solver.agent.get_llm", return_value=mock_llm):
+        _, _, execute_count, verified, llm_calls, _codegen_count = await _phase1_verify(
+            _state(), emitter=None
+        )
+
+    assert not verified
+    assert execute_count == 0
+    assert llm_calls == 1
+
+
+@pytest.mark.asyncio
+async def test_phase1_successful_code_execute_sets_verified() -> None:
+    """code_execute가 성공(exit_code 없음)하면 verified=True."""
+    tool_call = {"name": "code_execute", "args": {"code": "print(2)"}, "id": "tc1"}
+    ai_with_tool = AIMessage(content="", tool_calls=[tool_call])
+    ai_final = AIMessage(content="검증 완료", tool_calls=[])
+
+    mock_llm = MagicMock()
+    mock_llm.bind_tools.return_value = mock_llm
+    mock_llm.ainvoke = AsyncMock(side_effect=[ai_with_tool, ai_final])
+
+    mock_execute = AsyncMock(return_value="stdout:\n2\n")
+
+    with (
+        patch("proovy_agent.graph.agents.core_solver.agent.get_llm", return_value=mock_llm),
+        patch.dict(
+            "proovy_agent.graph.agents.core_solver.agent._TOOLS_BY_NAME",
+            {"code_execute": MagicMock(ainvoke=mock_execute)},
+        ),
+    ):
+        _, _, execute_count, verified, _, _ = await _phase1_verify(_state(), emitter=None)
+
+    assert verified
+    assert execute_count == 1
+
+
+@pytest.mark.asyncio
+async def test_phase1_failed_code_execute_does_not_set_verified() -> None:
+    """code_execute가 실패(exit_code: 1)하면 verified=False."""
+    tool_call = {"name": "code_execute", "args": {"code": "raise ValueError()"}, "id": "tc1"}
+    ai_with_tool = AIMessage(content="", tool_calls=[tool_call])
+    # 최대 반복 후 종료를 위해 최종 AIMessage 반환
+    ai_final = AIMessage(content="실패", tool_calls=[])
+
+    call_count = 0
+
+    async def side_effect(*args: object, **kwargs: object) -> AIMessage:
+        nonlocal call_count
+        call_count += 1
+        if call_count == 1:
+            return ai_with_tool
+        return ai_final
+
+    mock_llm = MagicMock()
+    mock_llm.bind_tools.return_value = mock_llm
+    mock_llm.ainvoke = AsyncMock(side_effect=side_effect)
+
+    mock_execute = AsyncMock(return_value="stderr:\nValueError\nexit_code: 1")
+
+    with (
+        patch("proovy_agent.graph.agents.core_solver.agent.get_llm", return_value=mock_llm),
+        patch.dict(
+            "proovy_agent.graph.agents.core_solver.agent._TOOLS_BY_NAME",
+            {"code_execute": MagicMock(ainvoke=mock_execute)},
+        ),
+    ):
+        _, _, execute_count, verified, _, _ = await _phase1_verify(_state(), emitter=None)
+
+    assert not verified
+    assert execute_count == 1
diff --git a/tests/graph/test_builder.py b/tests/graph/test_builder.py
new file mode 100644
index 0000000..42e9142
--- /dev/null
+++ b/tests/graph/test_builder.py
@@ -0,0 +1,28 @@
+"""LangGraph 빌더 — lazy singleton 테스트."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from proovy_agent.graph import builder as builder_module
+
+
+@pytest.fixture(autouse=True)
+def reset_singleton(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(builder_module, "_graph", None)
+
+
+def test_get_graph_builds_only_once() -> None:
+    mock_graph = MagicMock()
+    with patch.object(builder_module, "_build", return_value=mock_graph) as mock_build:
+        g1 = builder_module.get_graph()
+        g2 = builder_module.get_graph()
+
+    assert g1 is g2
+    assert mock_build.call_count == 1
+
+
+def test_get_graph_returns_build_result() -> None:
+    sentinel = object()
+    with patch.object(builder_module, "_build", return_value=sentinel):
+        assert builder_module.get_graph() is sentinel
diff --git a/tests/graph/test_credit_settler.py b/tests/graph/test_credit_settler.py
new file mode 100644
index 0000000..fac072e
--- /dev/null
+++ b/tests/graph/test_credit_settler.py
@@ -0,0 +1,51 @@
+"""CreditSettler 노드 단위 테스트."""
+
+import pytest
+
+from proovy_agent.common.sse.context import current_emitter
+from proovy_agent.common.sse.emitter import SSEEmitter
+from proovy_agent.graph.nodes.credit_settler import credit_settler
+from proovy_agent.graph.state import CreditEntry, ProovyState
+
+
+def _state(entries: list[CreditEntry]) -> ProovyState:
+    return ProovyState(user_id="u", thread_id="t", credit_log=entries)
+
+
+@pytest.mark.asyncio
+async def test_sums_total_credit() -> None:
+    state = _state(
+        [
+            CreditEntry(node="core_solver", action="llm_call_verify", model="flash", cost=2.0),
+            CreditEntry(node="core_solver", action="code_execute", cost=1.0),
+        ]
+    )
+    result = await credit_settler(state)
+    assert result["total_credit_cost"] == 3.0
+
+
+@pytest.mark.asyncio
+async def test_emits_credit_settled_event() -> None:
+    emitter = SSEEmitter()
+    token = current_emitter.set(emitter)
+    try:
+        await credit_settler(_state([CreditEntry(node="x", action="a", cost=5.0)]))
+    finally:
+        current_emitter.reset(token)
+
+    await emitter.close()
+    events = [e async for e in emitter.stream()]
+    assert any(e["event"] == "credit_settled" for e in events)
+
+
+@pytest.mark.asyncio
+async def test_no_emitter_does_not_raise() -> None:
+    state = _state([CreditEntry(node="x", action="a", cost=1.0)])
+    result = await credit_settler(state)
+    assert result["total_credit_cost"] == 1.0
+
+
+@pytest.mark.asyncio
+async def test_empty_log_total_is_zero() -> None:
+    result = await credit_settler(_state([]))
+    assert result["total_credit_cost"] == 0.0
diff --git a/tests/graph/test_plan_executor.py b/tests/graph/test_plan_executor.py
new file mode 100644
index 0000000..ba3bf32
--- /dev/null
+++ b/tests/graph/test_plan_executor.py
@@ -0,0 +1,87 @@
+"""PlanExecutor 노드 단위 테스트."""
+
+from langgraph.types import Command, Send
+import pytest
+
+from proovy_agent.graph.nodes.plan_executor import _find_ready_steps, plan_executor
+from proovy_agent.graph.state import PlanStep, ProovyState
+
+
+def _step(action: str, status: str = "pending") -> PlanStep:
+    return PlanStep(action=action, description="test", status=status)
+
+
+def _state(**kwargs: object) -> ProovyState:
+    return ProovyState(user_id="u", thread_id="t", **kwargs)
+
+
+# ── _find_ready_steps ────────────────────────────────────────────────────────
+
+
+def test_solve_has_no_deps() -> None:
+    plan = [_step("solve")]
+    ready = _find_ready_steps(plan)
+    assert len(ready) == 1
+    assert ready[0][1].action == "solve"
+
+
+def test_pdf_waits_for_solve() -> None:
+    plan = [_step("solve"), _step("pdf")]
+    assert len(_find_ready_steps(plan)) == 1
+    assert _find_ready_steps(plan)[0][1].action == "solve"
+
+
+def test_pdf_ready_after_solve_done() -> None:
+    plan = [_step("solve", "done"), _step("pdf")]
+    ready = _find_ready_steps(plan)
+    assert len(ready) == 1
+    assert ready[0][1].action == "pdf"
+
+
+def test_video_and_pdf_both_ready_after_solve() -> None:
+    plan = [_step("solve", "done"), _step("video"), _step("pdf")]
+    assert len(_find_ready_steps(plan)) == 2
+
+
+def test_running_step_is_not_ready() -> None:
+    plan = [_step("solve", "running")]
+    assert _find_ready_steps(plan) == []
+
+
+# ── plan_executor routing ────────────────────────────────────────────────────
+
+
+@pytest.mark.asyncio
+async def test_all_done_routes_to_credit_settler() -> None:
+    state = _state(plan=[_step("solve", "done")])
+    result = await plan_executor(state)
+    assert isinstance(result, Command)
+    assert result.goto == "credit_settler"
+
+
+@pytest.mark.asyncio
+async def test_single_step_routes_to_correct_node() -> None:
+    state = _state(plan=[_step("solve")])
+    result = await plan_executor(state)
+    assert isinstance(result, Command)
+    assert result.goto == "core_solver"
+
+
+@pytest.mark.asyncio
+async def test_parallel_steps_wrapped_in_command() -> None:
+    """list[Send] 대신 Command(goto=[Send(...)]) 형태로 반환해야 한다."""
+    state = _state(plan=[_step("solve", "done"), _step("video"), _step("pdf")])
+    result = await plan_executor(state)
+    assert isinstance(result, Command)
+    assert isinstance(result.goto, list)
+    assert len(result.goto) == 2
+    assert all(isinstance(s, Send) for s in result.goto)
+
+
+@pytest.mark.asyncio
+async def test_running_step_does_not_transition_to_credit_settler() -> None:
+    """병렬 브랜치 실행 중(running)이면 credit_settler로 가지 않는다."""
+    state = _state(plan=[_step("solve", "done"), _step("pdf", "running")])
+    result = await plan_executor(state)
+    assert isinstance(result, Command)
+    assert result.goto != "credit_settler"