diff --git a/agent/task_history.py b/agent/task_history.py
new file mode 100644
index 0000000..7f5d7b5
--- /dev/null
+++ b/agent/task_history.py
@@ -0,0 +1,84 @@
+"""완료 작업 이력 DB.
+
+작업의 비용, 소요시간, 토큰 사용량을 SQLite에 기록한다.
+"""
+from __future__ import annotations
+
+import logging
+import os
+
+import aiosqlite
+
+logger = logging.getLogger(__name__)
+
+_CREATE_TABLE = """
+CREATE TABLE IF NOT EXISTS task_history (
+    id INTEGER PRIMARY KEY AUTOINCREMENT,
+    task_id TEXT UNIQUE NOT NULL,
+    thread_id TEXT NOT NULL,
+    issue_number INTEGER NOT NULL DEFAULT 0,
+    repo_name TEXT NOT NULL DEFAULT '',
+    source TEXT NOT NULL DEFAULT '',
+    status TEXT NOT NULL,
+    created_at TEXT NOT NULL,
+    completed_at TEXT NOT NULL,
+    duration_seconds REAL NOT NULL DEFAULT 0,
+    tokens_input INTEGER NOT NULL DEFAULT 0,
+    tokens_output INTEGER NOT NULL DEFAULT 0,
+    cost_usd REAL NOT NULL DEFAULT 0,
+    error_message TEXT NOT NULL DEFAULT ''
+)
+"""
+
+
+class TaskHistory:
+    def __init__(self, db_path: str = "/data/task_history.db"):
+        self._db_path = db_path
+        self._db: aiosqlite.Connection | None = None
+
+    async def initialize(self) -> None:
+        self._db = await aiosqlite.connect(self._db_path)
+        self._db.row_factory = aiosqlite.Row
+        await self._db.execute(_CREATE_TABLE)
+        await self._db.commit()
+
+    async def close(self) -> None:
+        if self._db:
+            await self._db.close()
+
+    async def record(
+        self, task_id: str, thread_id: str, issue_number: int, repo_name: str,
+        source: str, status: str, created_at: str, completed_at: str,
+        duration_seconds: float, tokens_input: int, tokens_output: int,
+        cost_usd: float, error_message: str = "",
+    ) -> None:
+        await self._db.execute(
+            "INSERT OR REPLACE INTO task_history "
+            "(task_id, thread_id, issue_number, repo_name, source, status, "
+            "created_at, completed_at, duration_seconds, tokens_input, tokens_output, "
+            "cost_usd, error_message) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
+            (task_id, thread_id, issue_number, repo_name, source, status,
+             created_at, completed_at, duration_seconds, tokens_input, tokens_output,
+             cost_usd, error_message),
+        )
+        await self._db.commit()
+        logger.info("Recorded history: task=%s status=%s cost=$%.4f", task_id, status, cost_usd)
+
+    async def get_recent(self, limit: int = 20) -> list[dict]:
+        cursor = await self._db.execute(
+            "SELECT * FROM task_history ORDER BY completed_at DESC LIMIT ?", (limit,),
+        )
+        rows = await cursor.fetchall()
+        return [dict(row) for row in rows]
+
+
+_history: TaskHistory | None = None
+
+
+async def get_task_history() -> TaskHistory:
+    global _history
+    if _history is None:
+        db_path = os.environ.get("TASK_HISTORY_DB", "/data/task_history.db")
+        _history = TaskHistory(db_path=db_path)
+        await _history.initialize()
+    return _history
diff --git a/tests/test_task_history.py b/tests/test_task_history.py
new file mode 100644
index 0000000..303aba2
--- /dev/null
+++ b/tests/test_task_history.py
@@ -0,0 +1,69 @@
+import pytest
+import os
+import tempfile
+
+from agent.task_history import TaskHistory
+
+
+@pytest.fixture
+async def history():
+    fd, db_path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    h = TaskHistory(db_path=db_path)
+    await h.initialize()
+    yield h
+    await h.close()
+    os.unlink(db_path)
+
+
+@pytest.mark.asyncio
+async def test_record_completed(history):
+    await history.record(
+        task_id="task-1", thread_id="thread-1", issue_number=42,
+        repo_name="galaxis-po", source="gitea", status="completed",
+        created_at="2026-03-20T10:00:00Z", completed_at="2026-03-20T10:05:00Z",
+        duration_seconds=300.0, tokens_input=5000, tokens_output=2000, cost_usd=0.045,
+    )
+    records = await history.get_recent(limit=10)
+    assert len(records) == 1
+    assert records[0]["task_id"] == "task-1"
+    assert records[0]["status"] == "completed"
+
+
+@pytest.mark.asyncio
+async def test_record_failed(history):
+    await history.record(
+        task_id="task-2", thread_id="thread-2", issue_number=10,
+        repo_name="galaxis-po", source="discord", status="failed",
+        created_at="2026-03-20T11:00:00Z", completed_at="2026-03-20T11:01:00Z",
+        duration_seconds=60.0, tokens_input=1000, tokens_output=500, cost_usd=0.01,
+        error_message="Agent crashed",
+    )
+    records = await history.get_recent(limit=10)
+    assert len(records) == 1
+    assert records[0]["error_message"] == "Agent crashed"
+
+
+@pytest.mark.asyncio
+async def test_get_recent_ordered(history):
+    await history.record(
+        task_id="task-1", thread_id="t1", issue_number=1, repo_name="r",
+        source="gitea", status="completed", created_at="2026-03-20T10:00:00Z",
+        completed_at="2026-03-20T10:05:00Z", duration_seconds=300,
+        tokens_input=100, tokens_output=50, cost_usd=0.001,
+    )
+    await history.record(
+        task_id="task-2", thread_id="t2", issue_number=2, repo_name="r",
+        source="gitea", status="completed", created_at="2026-03-20T11:00:00Z",
+        completed_at="2026-03-20T11:05:00Z", duration_seconds=300,
+        tokens_input=200, tokens_output=100, cost_usd=0.002,
+    )
+    records = await history.get_recent(limit=10)
+    assert len(records) == 2
+    assert records[0]["task_id"] == "task-2"
+
+
+@pytest.mark.asyncio
+async def test_empty_history(history):
+    records = await history.get_recent(limit=10)
+    assert records == []