diff --git a/app/harness/log.py b/app/harness/log.py
new file mode 100644
index 0000000..8d9f970
--- /dev/null
+++ b/app/harness/log.py
@@ -0,0 +1,347 @@
+"""HarnessLog：SQLite 薄包装 + RunLogImpl 只读查询端口。
+
+HarnessLog 提供统一的结构化日志接口，从 TRM4 直搬，保留全部线程安全与幂等语义。
+RunLogImpl 实现 core/evolution/protocols.py::RunLog Protocol，用独立连接做只读 SELECT，
+不经 HarnessLog 生命周期（不触发 _runs INSERT OR IGNORE），避免污染运行状态。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import sqlite3
+import subprocess
+import threading
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+
+def _get_git_sha() -> str | None:
+    """获取当前 git commit SHA。"""
+    try:
+        result = subprocess.run(
+            ["git", "rev-parse", "HEAD"],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+        return result.stdout.strip()
+    except (subprocess.CalledProcessError, FileNotFoundError):
+        return None
+
+
+def _now_iso() -> str:
+    """返回当前 UTC 时间的 ISO 格式字符串。"""
+    return datetime.now(UTC).isoformat()
+
+
+class HarnessLog:
+    """SQLite 薄包装，为科研项目提供统一的结构化日志接口。
+
+    关键设计：
+    - WAL 模式 + threading.Lock 保证共享连接下并发安全。
+    - INSERT OR IGNORE INTO _runs 保证幂等（同 run_id 多次创建不报错）。
+    - query 也持锁：共享连接（check_same_thread=False）下并发 SELECT + INSERT
+      在同一连接上 execute 会损坏游标状态，故读也须串行化。
+    - context manager 语义：正常退出 completed，异常退出 failed。
+
+    参数:
+        db_path: SQLite 数据库文件路径。
+        run_id: 本次运行的唯一标识。
+        git_sha: 代码版本，默认自动获取。
+        config_snapshot: 本次运行的配置快照。
+    """
+
+    def __init__(
+        self,
+        db_path: str,
+        run_id: str,
+        git_sha: str | None = None,
+        config_snapshot: dict[str, Any] | None = None,
+    ) -> None:
+        self._run_id = run_id
+        Path(db_path).parent.mkdir(parents=True, exist_ok=True)
+        self._conn = sqlite3.connect(db_path, check_same_thread=False)
+        self._lock = threading.Lock()
+        self._conn.row_factory = sqlite3.Row
+        self._conn.execute("PRAGMA journal_mode=WAL")
+        self._init_fixed_tables()
+        resolved_sha = git_sha or _get_git_sha()
+        config_json = (
+            json.dumps(config_snapshot, ensure_ascii=False) if config_snapshot else None
+        )
+        self._conn.execute(
+            "INSERT OR IGNORE INTO _runs"
+            " (run_id, git_sha, started_at, config, status)"
+            " VALUES (?, ?, ?, ?, ?)",
+            (run_id, resolved_sha, _now_iso(), config_json, "running"),
+        )
+        self._conn.commit()
+
+    def _init_fixed_tables(self) -> None:
+        """创建 _runs 和 _events 固定表。"""
+        self._conn.execute("""
+            CREATE TABLE IF NOT EXISTS _runs (
+                run_id TEXT PRIMARY KEY,
+                git_sha TEXT,
+                started_at TEXT,
+                finished_at TEXT,
+                config JSON,
+                status TEXT DEFAULT 'running',
+                skills_version TEXT,
+                prompts_version TEXT,
+                questions_ref TEXT
+            )
+        """)
+        self._conn.execute("""
+            CREATE TABLE IF NOT EXISTS _events (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                run_id TEXT,
+                timestamp TEXT,
+                event_type TEXT,
+                payload JSON
+            )
+        """)
+        self._conn.commit()
+
+    def create_table(
+        self,
+        name: str,
+        columns: dict[str, str],
+        primary_key: str | None = None,
+    ) -> None:
+        """创建自定义表，自动追加 run_id 和 timestamp 列。
+
+        参数:
+            name: 表名。
+            columns: 列定义，如 {"epoch": "INTEGER", "loss": "REAL"}。
+            primary_key: 主键列名。
+        """
+        all_columns = {"run_id": "TEXT", "timestamp": "TEXT"}
+        all_columns.update(columns)
+        col_defs = []
+        for col_name, col_type in all_columns.items():
+            pk_suffix = " PRIMARY KEY" if col_name == primary_key else ""
+            col_defs.append(f"{col_name} {col_type}{pk_suffix}")
+        sql = f"CREATE TABLE IF NOT EXISTS {name} ({', '.join(col_defs)})"
+        self._conn.execute(sql)
+        self._conn.commit()
+
+    def insert(self, table: str, record: dict[str, Any], mode: str = "append") -> None:
+        """插入一条记录，自动填充 run_id 和 timestamp。
+
+        参数:
+            table: 目标表名。
+            record: 要插入的数据。
+            mode: "append" 或 "upsert"。
+        """
+        enriched = {"run_id": self._run_id, "timestamp": _now_iso()}
+        enriched.update(record)
+        cols = list(enriched.keys())
+        placeholders = ", ".join(["?"] * len(cols))
+        col_names = ", ".join(cols)
+        values = [enriched[c] for c in cols]
+        if mode == "upsert":
+            sql = (
+                f"INSERT OR REPLACE INTO {table} ({col_names}) VALUES ({placeholders})"
+            )
+        else:
+            sql = f"INSERT INTO {table} ({col_names}) VALUES ({placeholders})"
+        with self._lock:
+            self._conn.execute(sql, values)
+            self._conn.commit()
+
+    def insert_many(
+        self, table: str, records: list[dict[str, Any]], mode: str = "append"
+    ) -> None:
+        """批量插入多条记录。
+
+        参数:
+            table: 目标表名。
+            records: 要插入的数据列表。
+            mode: "append" 或 "upsert"。
+        """
+        for record in records:
+            self.insert(table, record, mode=mode)
+
+    def execute(self, sql: str, params: tuple[Any, ...] = ()) -> None:
+        """执行原生 SQL 写操作。
+
+        参数:
+            sql: SQL 语句。
+            params: 参数元组。
+        """
+        with self._lock:
+            self._conn.execute(sql, params)
+            self._conn.commit()
+
+    def query(self, sql: str, params: tuple[Any, ...] = ()) -> list[dict[str, Any]]:
+        """执行原生 SQL 查询，返回 list[dict]。
+
+        与所有写方法同持 self._lock：共享连接（check_same_thread=False）下，
+        并发 SELECT 与 INSERT 在同一连接上 execute 会损坏游标状态，故读也须串行化。
+
+        参数:
+            sql: SQL 查询语句。
+            params: 参数元组。
+
+        返回:
+            查询结果列表，每行为一个字典。
+        """
+        with self._lock:
+            cursor = self._conn.execute(sql, params)
+            columns = [desc[0] for desc in cursor.description]
+            return [
+                dict(zip(columns, row, strict=True)) for row in cursor.fetchall()
+            ]
+
+    def log_event(self, event_type: str, payload: dict[str, Any]) -> None:
+        """向 _events 表写入一条事件。
+
+        参数:
+            event_type: 事件类型标识。
+            payload: 事件数据。
+        """
+        with self._lock:
+            self._conn.execute(
+                "INSERT INTO _events (run_id, timestamp, event_type, payload)"
+                " VALUES (?, ?, ?, ?)",
+                (
+                    self._run_id,
+                    _now_iso(),
+                    event_type,
+                    json.dumps(payload, ensure_ascii=False),
+                ),
+            )
+            self._conn.commit()
+
+    def close(self, status: str = "completed") -> None:
+        """更新运行状态并关闭连接。
+
+        参数:
+            status: 最终状态，"completed" 或 "failed"。
+        """
+        with self._lock:
+            self._conn.execute(
+                "UPDATE _runs SET finished_at = ?, status = ? WHERE run_id = ?",
+                (_now_iso(), status, self._run_id),
+            )
+            self._conn.commit()
+            self._conn.close()
+
+    def __enter__(self) -> HarnessLog:
+        return self
+
+    def __exit__(
+        self,
+        exc_type: type[BaseException] | None,
+        exc_val: BaseException | None,
+        exc_tb: Any,
+    ) -> None:
+        status = "failed" if exc_type is not None else "completed"
+        self.close(status=status)
+
+
+# ---------------------------------------------------------------------------
+# RunLogImpl — core/evolution/protocols.py::RunLog 的只读实现
+# ---------------------------------------------------------------------------
+
+
+def _read_table(
+    db_path: str,
+    table: str,
+    run_id: str,
+    *,
+    question_ids: list[str] | None = None,
+) -> list[dict[str, Any]]:
+    """纯读某表指定 run 的行——不经 HarnessLog 生命周期，避免回读污染 _runs 运行状态。
+
+    HarnessLog.__enter__/__exit__ 会对 run_id 做 INSERT OR IGNORE 并在退出时标 completed；
+    回读指标绝不应改运行状态，故走独立只读连接（仅 SELECT）。
+
+    参数:
+        db_path: SQLite 路径。
+        table: 表名（内部固定常量，非外部输入，无注入风险）。
+        run_id: 过滤的 run ID。
+        question_ids: 可选的 question_id 过滤列表。
+
+    返回:
+        行 dict 列表；表尚未建（没写过）视为无数据返 []。
+    """
+    conn = sqlite3.connect(db_path)
+    conn.row_factory = sqlite3.Row
+    try:
+        exists = conn.execute(
+            "SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table,)
+        ).fetchone()
+        if exists is None:
+            return []
+
+        if question_ids is not None:
+            placeholders = ", ".join(["?"] * len(question_ids))
+            sql = (
+                f"SELECT * FROM {table}"
+                f" WHERE run_id = ? AND question_id IN ({placeholders})"
+            )
+            rows = conn.execute(sql, (run_id, *question_ids)).fetchall()
+        else:
+            rows = conn.execute(
+                f"SELECT * FROM {table} WHERE run_id = ?", (run_id,)
+            ).fetchall()
+
+        return [dict(r) for r in rows]
+    finally:
+        conn.close()
+
+
+class RunLogImpl:
+    """RunLog Protocol 的只读实现。
+
+    用独立 sqlite3.connect 做 SELECT，不经 HarnessLog 生命周期（不触发 _runs INSERT），
+    asyncio.to_thread 包装同步 SQL 查询，避免引入 aiosqlite 新依赖。
+
+    参数:
+        db_path: SQLite 数据库文件路径。
+    """
+
+    def __init__(self, db_path: str) -> None:
+        self._db_path = db_path
+
+    async def get_predictions(
+        self,
+        run_id: str,
+        *,
+        question_ids: list[str] | None = None,
+    ) -> list[dict[str, Any]]:
+        """查询指定 run 的预测记录。
+
+        参数:
+            run_id: 运行标识。
+            question_ids: 可选的题目 ID 过滤列表。
+
+        返回:
+            预测记录字典列表。
+        """
+        return await asyncio.to_thread(
+            _read_table, self._db_path, "predictions", run_id, question_ids=question_ids
+        )
+
+    async def get_traces(
+        self,
+        run_id: str,
+        *,
+        question_ids: list[str] | None = None,
+    ) -> list[dict[str, Any]]:
+        """查询指定 run 的推理轨迹。
+
+        参数:
+            run_id: 运行标识。
+            question_ids: 可选的题目 ID 过滤列表。
+
+        返回:
+            轨迹记录字典列表。
+        """
+        return await asyncio.to_thread(
+            _read_table, self._db_path, "traces", run_id, question_ids=question_ids
+        )
diff --git a/tests/unit/test_harness_log.py b/tests/unit/test_harness_log.py
new file mode 100644
index 0000000..ed86871
--- /dev/null
+++ b/tests/unit/test_harness_log.py
@@ -0,0 +1,347 @@
+"""HarnessLog + RunLogImpl 单元测试。
+
+HarnessLog: SQLite 薄包装的线程安全、幂等、context manager 语义验证。
+RunLogImpl: 只读 RunLog Protocol 实现，独立连接不污染 _runs 运行状态。
+"""
+
+from __future__ import annotations
+
+import sqlite3
+import threading
+from typing import TYPE_CHECKING
+
+import pytest
+
+from app.harness.log import HarnessLog, RunLogImpl
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+# ---------------------------------------------------------------------------
+# Fixtures
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture()
+def db_path(tmp_path: Path) -> str:
+    """返回临时 SQLite 数据库路径。"""
+    return str(tmp_path / "test.db")
+
+
+@pytest.fixture()
+def run_id() -> str:
+    return "test-run-001"
+
+
+# ===========================================================================
+# HarnessLog 测试
+# ===========================================================================
+
+
+class TestHarnessLog:
+    """HarnessLog 核心功能测试。"""
+
+    def test_create_table_and_insert(self, db_path: str, run_id: str) -> None:
+        """create_table 建表 + insert 写入 + query 读回。"""
+        with HarnessLog(db_path, run_id) as log:
+            log.create_table("metrics", {"epoch": "INTEGER", "loss": "REAL"})
+            log.insert("metrics", {"epoch": 1, "loss": 0.5})
+            rows = log.query("SELECT * FROM metrics WHERE run_id = ?", (run_id,))
+
+        assert len(rows) == 1
+        assert rows[0]["epoch"] == 1
+        assert rows[0]["loss"] == 0.5
+        assert rows[0]["run_id"] == run_id
+        # insert 自动填充 timestamp
+        assert rows[0]["timestamp"] is not None
+
+    def test_query_thread_safety(self, db_path: str, run_id: str) -> None:
+        """并发读写不损坏游标状态。"""
+        errors: list[Exception] = []
+
+        with HarnessLog(db_path, run_id) as log:
+            log.create_table("counter", {"value": "INTEGER"})
+
+            def writer() -> None:
+                try:
+                    for i in range(50):
+                        log.insert("counter", {"value": i})
+                except Exception as exc:
+                    errors.append(exc)
+
+            def reader() -> None:
+                try:
+                    for _ in range(50):
+                        log.query("SELECT COUNT(*) as cnt FROM counter")
+                except Exception as exc:
+                    errors.append(exc)
+
+            threads = [
+                threading.Thread(target=writer),
+                threading.Thread(target=reader),
+                threading.Thread(target=reader),
+            ]
+            for t in threads:
+                t.start()
+            for t in threads:
+                t.join()
+
+        assert errors == [], f"并发读写产生错误: {errors}"
+
+    def test_context_manager_completed(self, db_path: str, run_id: str) -> None:
+        """正常退出时 status = completed。"""
+        with HarnessLog(db_path, run_id):
+            pass
+
+        conn = sqlite3.connect(db_path)
+        conn.row_factory = sqlite3.Row
+        row = conn.execute(
+            "SELECT status, finished_at FROM _runs WHERE run_id = ?", (run_id,)
+        ).fetchone()
+        conn.close()
+
+        assert row["status"] == "completed"
+        assert row["finished_at"] is not None
+
+    def test_context_manager_failed(self, db_path: str, run_id: str) -> None:
+        """异常退出时 status = failed。"""
+        with pytest.raises(ValueError, match="boom"), HarnessLog(db_path, run_id):
+            raise ValueError("boom")
+
+        conn = sqlite3.connect(db_path)
+        conn.row_factory = sqlite3.Row
+        row = conn.execute(
+            "SELECT status FROM _runs WHERE run_id = ?", (run_id,)
+        ).fetchone()
+        conn.close()
+
+        assert row["status"] == "failed"
+
+    def test_insert_or_ignore_idempotent(self, db_path: str, run_id: str) -> None:
+        """同一 run_id 多次创建 HarnessLog 不报错（INSERT OR IGNORE 幂等）。"""
+        with HarnessLog(db_path, run_id):
+            pass
+
+        # 再次用同一 run_id 打开——不应抛异常
+        with HarnessLog(db_path, run_id) as log:
+            log.create_table("t", {"x": "INTEGER"})
+            log.insert("t", {"x": 42})
+
+        conn = sqlite3.connect(db_path)
+        count = conn.execute(
+            "SELECT COUNT(*) FROM _runs WHERE run_id = ?", (run_id,)
+        ).fetchone()[0]
+        conn.close()
+
+        assert count == 1, "INSERT OR IGNORE 应保证 _runs 只有一行"
+
+    def test_wal_mode(self, db_path: str, run_id: str) -> None:
+        """连接初始化后 journal_mode 应为 WAL。"""
+        with HarnessLog(db_path, run_id) as log:
+            rows = log.query("PRAGMA journal_mode")
+
+        assert rows[0]["journal_mode"].lower() == "wal"
+
+    def test_insert_many(self, db_path: str, run_id: str) -> None:
+        """insert_many 批量插入多条记录。"""
+        records = [{"epoch": i, "loss": float(i) * 0.1} for i in range(5)]
+
+        with HarnessLog(db_path, run_id) as log:
+            log.create_table("batch", {"epoch": "INTEGER", "loss": "REAL"})
+            log.insert_many("batch", records)
+            rows = log.query(
+                "SELECT * FROM batch WHERE run_id = ? ORDER BY epoch", (run_id,)
+            )
+
+        assert len(rows) == 5
+        assert [r["epoch"] for r in rows] == [0, 1, 2, 3, 4]
+
+    def test_log_event(self, db_path: str, run_id: str) -> None:
+        """log_event 向 _events 表写入事件。"""
+        with HarnessLog(db_path, run_id) as log:
+            log.log_event("train_start", {"epoch": 1, "lr": 0.001})
+            log.log_event("train_end", {"epoch": 1, "loss": 0.42})
+            rows = log.query(
+                "SELECT * FROM _events WHERE run_id = ? ORDER BY id", (run_id,)
+            )
+
+        assert len(rows) == 2
+        assert rows[0]["event_type"] == "train_start"
+        assert rows[1]["event_type"] == "train_end"
+
+    def test_execute_raw_sql(self, db_path: str, run_id: str) -> None:
+        """execute 执行原生 SQL 写操作。"""
+        with HarnessLog(db_path, run_id) as log:
+            log.create_table("raw", {"val": "INTEGER"})
+            log.execute(
+                "INSERT INTO raw (run_id, timestamp, val) VALUES (?, ?, ?)",
+                (run_id, "2026-01-01T00:00:00", 99),
+            )
+            rows = log.query("SELECT val FROM raw WHERE run_id = ?", (run_id,))
+
+        assert len(rows) == 1
+        assert rows[0]["val"] == 99
+
+    def test_upsert_mode(self, db_path: str, run_id: str) -> None:
+        """insert mode='upsert' 使用 INSERT OR REPLACE。"""
+        with HarnessLog(db_path, run_id) as log:
+            log.create_table("kv", {"key": "TEXT", "val": "TEXT"}, primary_key="key")
+            log.insert("kv", {"key": "a", "val": "1"}, mode="upsert")
+            log.insert("kv", {"key": "a", "val": "2"}, mode="upsert")
+            rows = log.query("SELECT val FROM kv WHERE key = 'a'")
+
+        assert len(rows) == 1
+        assert rows[0]["val"] == "2"
+
+
+# ===========================================================================
+# RunLogImpl 测试
+# ===========================================================================
+
+
+def _setup_predictions_and_traces(db_path: str, run_id: str) -> None:
+    """向测试数据库写入 predictions 和 traces 数据，模拟 inference 阶段产物。"""
+    with HarnessLog(db_path, run_id) as log:
+        log.create_table(
+            "predictions",
+            {
+                "question_id": "TEXT",
+                "predicted_answer": "TEXT",
+                "correct": "INTEGER",
+            },
+        )
+        log.create_table(
+            "traces",
+            {
+                "question_id": "TEXT",
+                "step_idx": "INTEGER",
+                "action": "TEXT",
+                "observation": "TEXT",
+            },
+        )
+        log.insert_many(
+            "predictions",
+            [
+                {"question_id": "q1", "predicted_answer": "A", "correct": 1},
+                {"question_id": "q2", "predicted_answer": "B", "correct": 0},
+                {"question_id": "q3", "predicted_answer": "C", "correct": 1},
+            ],
+        )
+        log.insert_many(
+            "traces",
+            [
+                {
+                    "question_id": "q1",
+                    "step_idx": 0,
+                    "action": "search",
+                    "observation": "found",
+                },
+                {
+                    "question_id": "q1",
+                    "step_idx": 1,
+                    "action": "verify",
+                    "observation": "ok",
+                },
+                {
+                    "question_id": "q2",
+                    "step_idx": 0,
+                    "action": "search",
+                    "observation": "not found",
+                },
+            ],
+        )
+
+
+class TestRunLogImpl:
+    """RunLogImpl 只读查询端口测试。"""
+
+    @pytest.mark.asyncio
+    async def test_get_predictions(self, db_path: str, run_id: str) -> None:
+        """get_predictions 返回指定 run 的全部预测记录。"""
+        _setup_predictions_and_traces(db_path, run_id)
+        impl = RunLogImpl(db_path)
+
+        preds = await impl.get_predictions(run_id)
+
+        assert len(preds) == 3
+        q_ids = {p["question_id"] for p in preds}
+        assert q_ids == {"q1", "q2", "q3"}
+
+    @pytest.mark.asyncio
+    async def test_get_predictions_filtered(self, db_path: str, run_id: str) -> None:
+        """get_predictions 按 question_ids 过滤。"""
+        _setup_predictions_and_traces(db_path, run_id)
+        impl = RunLogImpl(db_path)
+
+        preds = await impl.get_predictions(run_id, question_ids=["q1", "q3"])
+
+        assert len(preds) == 2
+        q_ids = {p["question_id"] for p in preds}
+        assert q_ids == {"q1", "q3"}
+
+    @pytest.mark.asyncio
+    async def test_get_traces(self, db_path: str, run_id: str) -> None:
+        """get_traces 返回指定 run 的全部轨迹记录。"""
+        _setup_predictions_and_traces(db_path, run_id)
+        impl = RunLogImpl(db_path)
+
+        traces = await impl.get_traces(run_id)
+
+        assert len(traces) == 3
+        # q1 有 2 步，q2 有 1 步
+        q1_traces = [t for t in traces if t["question_id"] == "q1"]
+        assert len(q1_traces) == 2
+
+    @pytest.mark.asyncio
+    async def test_get_traces_filtered(self, db_path: str, run_id: str) -> None:
+        """get_traces 按 question_ids 过滤。"""
+        _setup_predictions_and_traces(db_path, run_id)
+        impl = RunLogImpl(db_path)
+
+        traces = await impl.get_traces(run_id, question_ids=["q2"])
+
+        assert len(traces) == 1
+        assert traces[0]["question_id"] == "q2"
+
+    @pytest.mark.asyncio
+    async def test_readonly_no_runs_insert(self, db_path: str, run_id: str) -> None:
+        """RunLogImpl 查询不触发 _runs INSERT（不污染运行状态）。"""
+        _setup_predictions_and_traces(db_path, run_id)
+
+        # 用不同 run_id 查询，确保不会创建新的 _runs 行
+        other_run = "nonexistent-run"
+        impl = RunLogImpl(db_path)
+        preds = await impl.get_predictions(other_run)
+
+        assert preds == []
+
+        # 验证 _runs 表没有 other_run 的记录
+        conn = sqlite3.connect(db_path)
+        count = conn.execute(
+            "SELECT COUNT(*) FROM _runs WHERE run_id = ?", (other_run,)
+        ).fetchone()[0]
+        conn.close()
+        assert count == 0, "RunLogImpl 不应向 _runs 插入记录"
+
+    @pytest.mark.asyncio
+    async def test_protocol_compliance(self, db_path: str, run_id: str) -> None:
+        """RunLogImpl 满足 RunLog Protocol（runtime_checkable isinstance 检查）。"""
+        from core.evolution.protocols import RunLog
+
+        impl = RunLogImpl(db_path)
+        assert isinstance(impl, RunLog)
+
+    @pytest.mark.asyncio
+    async def test_missing_table_returns_empty(self, db_path: str, run_id: str) -> None:
+        """查询不存在的表（predictions/traces 未建）返回空列表。"""
+        # 仅创建 _runs，不建 predictions/traces 表
+        with HarnessLog(db_path, run_id):
+            pass
+
+        impl = RunLogImpl(db_path)
+        preds = await impl.get_predictions(run_id)
+        traces = await impl.get_traces(run_id)
+
+        assert preds == []
+        assert traces == []