diff --git a/app/repository/convert_repository.py b/app/repository/convert_repository.py
new file mode 100644
index 0000000..f4916d0
--- /dev/null
+++ b/app/repository/convert_repository.py
@@ -0,0 +1,181 @@
+"""基金转换（convert）代理侧仓储 · T-4（agent 库 risk_convert_detail）。
+
+阶段零占位 / 阶段二回写 / 重试与幂等查询 / 超时清理。仅服务于 convert 模块，
+**不与 risk_repository 混职责**（后者不动；开发计划 §5.2 DoD）。
+
+写侧全部走 agent 库可写引擎（D20：`xh_agent_rw`，`risk_convert_detail` 在授权表内）；
+读侧也走同一引擎（本仓储只碰 risk_convert_detail 一张表，无 Core 依赖）。
+
+S2（评审）：清理**标记不硬删**——`mark_expired` 置 `status='expired'`，绝不 DELETE，
+留痕供对账（与 cleanup_pending_convert.py 的口径一致）。
+"""
+
+from __future__ import annotations
+
+from datetime import date, datetime, timedelta
+from decimal import Decimal
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.engine import Engine
+
+from app.config.settings import settings
+from app.utils.db import get_engine
+
+
+def _to_bind(value: Any) -> Any:
+    """Decimal 转 float 再绑定（sqlite 不支持直接绑定 Decimal；MySQL DECIMAL 列自动收口）。
+
+    与 T-3 测试教训一致：插入/更新 Decimal 一律转 float，读回侧用 `Decimal(str(...))` 还原。
+    """
+    return float(value) if isinstance(value, Decimal) else value
+
+
+_STATUS_PENDING = "pending"
+_STATUS_COMPLETED = "completed"
+_STATUS_FAILED = "failed"
+_STATUS_EXPIRED = "expired"
+
+
+class ConvertRepository:
+    """risk_convert_detail 读写（占位 / 回写 / 查询 / 清理）。"""
+
+    def __init__(self, engine: Engine | None = None) -> None:
+        self._engine = engine or get_engine(settings.mysql_database, "rw")
+
+    # ---------- 阶段零：占位（uk_idem 兜底） ----------
+
+    def insert_placeholder(self, group_id: str, client_request_id: str | None) -> None:
+        """阶段零占位：插一行 `status='pending'`。
+
+        `client_request_id` 为 None 时绑 NULL——MySQL / sqlite 的 UNIQUE 约束均允许多个 NULL，
+        故「无幂等键的请求」可重复占位、互不冲突（uk_idem 仅对非空键兜底）。
+        `estimated=0`：convert 占位是真实请求，非风控预估单（与 risk_alert 语义区分）。
+        """
+        sql = text(
+            """
+            INSERT INTO risk_convert_detail
+                (convert_group_id, client_request_id, status, estimated)
+            VALUES (:gid, :cid_req, :status, 0)
+            """
+        )
+        with self._engine.begin() as conn:
+            conn.execute(
+                sql,
+                {"gid": group_id, "cid_req": client_request_id, "status": _STATUS_PENDING},
+            )
+
+    # ---------- 阶段二：回写 completed + 详情 ----------
+
+    def complete_convert(
+        self,
+        group_id: str,
+        *,
+        out_trade_id: str,
+        in_trade_id: str,
+        related_trade_id: str | None = None,
+        nav: Decimal | None = None,
+        nav_date: date | None = None,
+        fee_amount: Decimal | None = None,
+        hold_days_min: int | None = None,
+        hold_days_max: int | None = None,
+        nav_stale: bool = False,
+    ) -> None:
+        """阶段二回写：置 `status='completed'` + 折算详情（架构 §3 步骤⑧）。
+
+        nav / fee_amount 为 Decimal → 转 float 绑定（见 `_to_bind`）。
+        本方法只改 risk_convert_detail；主审计写入由 convert_service（T-7）负责，不在此处。
+        """
+        sql = text(
+            """
+            UPDATE risk_convert_detail
+            SET status = :status,
+                out_trade_id = :out,
+                in_trade_id = :in,
+                related_trade_id = :rel,
+                nav = :nav,
+                nav_date = :nav_date,
+                fee_amount = :fee,
+                hold_days_min = :hmin,
+                hold_days_max = :hmax,
+                nav_stale = :stale
+            WHERE convert_group_id = :gid
+            """
+        )
+        with self._engine.begin() as conn:
+            conn.execute(
+                sql,
+                {
+                    "status": _STATUS_COMPLETED,
+                    "out": out_trade_id,
+                    "in": in_trade_id,
+                    "rel": related_trade_id,
+                    "nav": _to_bind(nav),
+                    "nav_date": nav_date,
+                    "fee": _to_bind(fee_amount),
+                    "hmin": hold_days_min,
+                    "hmax": hold_days_max,
+                    "stale": 1 if nav_stale else 0,
+                    "gid": group_id,
+                },
+            )
+
+    def mark_failed(self, group_id: str) -> None:
+        """阶段一失败 → 占位置 `failed`（供巡检/人工补偿，SLA 24h 内）。"""
+        with self._engine.begin() as conn:
+            conn.execute(
+                text("UPDATE risk_convert_detail SET status = :s WHERE convert_group_id = :gid"),
+                {"s": _STATUS_FAILED, "gid": group_id},
+            )
+
+    # ---------- 查询 ----------
+
+    def get_by_group_id(self, group_id: str) -> dict[str, Any] | None:
+        """按 convert_group_id 读单行（重试判定 / 阶段二补跑读取）。"""
+        with self._engine.connect() as conn:
+            row = conn.execute(
+                text("SELECT * FROM risk_convert_detail WHERE convert_group_id = :gid"),
+                {"gid": group_id},
+            ).mappings().first()
+            return dict(row) if row else None
+
+    def get_by_client_request_id(self, client_request_id: str | None) -> dict[str, Any] | None:
+        """按 client_request_id 读单行（幂等命中读取）；None 直接返回 None（WHERE = NULL 永不命中）。"""
+        if client_request_id is None:
+            return None
+        with self._engine.connect() as conn:
+            row = conn.execute(
+                text("SELECT * FROM risk_convert_detail WHERE client_request_id = :cid_req"),
+                {"cid_req": client_request_id},
+            ).mappings().first()
+            return dict(row) if row else None
+
+    # ---------- 清理（S2：标记不硬删） ----------
+
+    def list_expired_candidates(self, hours: int) -> list[dict[str, Any]]:
+        """取超 `hours` 小时的 `pending` 孤儿（供 cleanup_pending_convert.py 巡检）。
+
+        cutoff = 当前本地时间 - hours；sqlite 默认以 localtime 落 created_at，
+        与 Python datetime.now()（本地）口径一致（B5 评审 P3-4 同口径）。
+        """
+        cutoff = datetime.now() - timedelta(hours=hours)
+        with self._engine.connect() as conn:
+            return [
+                dict(r)
+                for r in conn.execute(
+                    text(
+                        "SELECT * FROM risk_convert_detail "
+                        "WHERE status = :s AND created_at < :cutoff "
+                        "ORDER BY created_at ASC"
+                    ),
+                    {"s": _STATUS_PENDING, "cutoff": cutoff},
+                ).mappings()
+            ]
+
+    def mark_expired(self, group_id: str) -> None:
+        """超时 pending 孤儿 → 置 `status='expired'`（标记不硬删，留痕供对账，S2）。"""
+        with self._engine.begin() as conn:
+            conn.execute(
+                text("UPDATE risk_convert_detail SET status = :s WHERE convert_group_id = :gid"),
+                {"s": _STATUS_EXPIRED, "gid": group_id},
+            )
diff --git a/app/service/risk/locks.py b/app/service/risk/locks.py
index 6cf748a..4a76db3 100644
--- a/app/service/risk/locks.py
+++ b/app/service/risk/locks.py
@@ -90,3 +90,63 @@ def run_locked(key: str, fn: Callable[[bool], Any]) -> Any:
         return fn(locked=True)
     finally:
         lock.release()
+
+
+class _NoLock:
+    """try_lock 未抢到锁的哨兵：``with`` 进入返回 False，避免 ``with None`` 报错；退出无操作。"""
+
+    def __enter__(self) -> bool:
+        return False
+
+    def __exit__(self, *exc: object) -> None:
+        return None
+
+
+class _Token:
+    """try_lock 抢到锁的上下文令牌：``with`` 进入返回 True，退出时按来源释放。
+
+    redis_key/token 非空 → 释放 Redis 锁；local_lock 非空 → 释放进程内锁。
+    """
+
+    def __init__(
+        self,
+        *,
+        redis_key: str | None = None,
+        token: str | None = None,
+        local_lock: threading.Lock | None = None,
+    ) -> None:
+        self._redis_key = redis_key
+        self._token = token
+        self._local_lock = local_lock
+
+    def __enter__(self) -> bool:
+        return True
+
+    def __exit__(self, *exc: object) -> None:
+        if self._local_lock is not None:
+            self._local_lock.release()
+        elif self._redis_key is not None:
+            _release_redis(self._redis_key, self._token)  # type: ignore[arg-type]
+
+
+def try_lock(key: str, ttl_seconds: int = LOCK_TTL_SECONDS) -> "_Token | _NoLock":
+    """单次尝试抢锁：抢到返回 ``_Token``（``with`` 进入 True），抢不到立即 ``_NoLock``（进入 False）。
+
+    **不等、不降级**：Redis 可用但锁被他人持有 → 立即失败（不重试、不降级 fn(False)），
+    与 ``run_locked``（等 2s 后降级）语义相反（D4）。
+    Redis 不可用（任何异常）→ 退回进程内 ``Lock.acquire(blocking=False)``，语义一致。
+
+    返回对象均可安全用于 ``with try_lock(...) as acquired:``（抢到 acquired=True，否则 False）。
+    """
+    token = uuid4().hex
+    redis_key = _LOCK_KEY_PREFIX + key
+    try:
+        if redis_gateway.get_gateway().acquire_lock(redis_key, token, ttl_seconds):
+            return _Token(redis_key=redis_key, token=token)
+    except Exception:
+        logger.warning("Redis 锁不可用，退回进程内锁：%s", redis_key, exc_info=True)
+        local = lock_for(key)
+        if local.acquire(blocking=False):
+            return _Token(local_lock=local)
+        return _NoLock()
+    return _NoLock()
diff --git a/docs/项目框架设计/开发计划-基金转换交易.md b/docs/项目框架设计/开发计划-基金转换交易.md
index 944dde1..a7c9758 100644
--- a/docs/项目框架设计/开发计划-基金转换交易.md
+++ b/docs/项目框架设计/开发计划-基金转换交易.md
@@ -787,10 +787,21 @@ T-7 幂等窗口 · T-13 的 50 并发压测与性能补录 · PRD §5.3 实算
 | `list_expired_candidates(hours)` | 供 `cleanup_pending_convert.py` 取超 24h 的 `pending` |
 | `mark_expired(group_id)` | 置 `expired`（**标记不硬删**，S2） |
 
-**DoD**
-- [ ] sqlite 单测覆盖：占位 → completed / 占位 → failed / 超时 → expired
-- [ ] `client_request_id` 为 `None` 时不受 `uk_idem` 约束（MySQL 允许多个 NULL；sqlite 侧需断言行为一致）
-- [ ] 不与 `risk_repository` 混职责（后者不动）
+**DoD（全部达成，见下方执行记录）**
+- [x] sqlite 单测覆盖：占位 → completed / 占位 → failed / 超时 → expired
+- [x] `client_request_id` 为 `None` 时不受 `uk_idem` 约束（MySQL 允许多个 NULL；sqlite 侧断言一致）
+- [x] 不与 `risk_repository` 混职责（后者不动）
+
+**执行记录（2026-09-10）**
+
+| 项 | 内容 |
+| --- | --- |
+| 新增文件 | `app/repository/convert_repository.py`【agent 库 `risk_convert_detail` 读写】· `tests/test_convert_repository.py`【6 用例】 |
+| 方法 | `insert_placeholder`（占位 `pending`，`estimated=0` 真实请求）· `complete_convert`（回写 `completed`+折算详情）· `mark_failed` · `get_by_group_id` · `get_by_client_request_id`（None 直接返回 None）· `list_expired_candidates(hours)`（pending 且 `created_at < now-hours`）· `mark_expired`（S2 标记不硬删） |
+| status 枚举 | 5 值 `pending/completed/failed/cancelled/expired`（架构 §9，建表即全量、零 ALTER） |
+| 引擎 | `get_engine(settings.mysql_database, "rw")`（D20：`xh_agent_rw` 含 `risk_convert_detail`） |
+| 踩坑（同 T-3） | `complete_convert` 的 `nav`/`fee_amount` 为 Decimal → 经 `_to_bind` 转 `float` 绑定（sqlite 不支持直接绑 Decimal） |
+| 验证 | `pytest tests/test_convert_repository.py` 6 passed；全量 634 passed / 3 skipped（基线 624 + 10） |
 
 **依赖**：T-1
 
@@ -806,10 +817,20 @@ def try_lock(key: str, ttl_seconds: int = LOCK_TTL_SECONDS) -> _Token | None:
 - **不改 `run_locked`**（D4：`run_locked` 会等 2s 并降级 `fn(False)`，与「抢不到立即 202」语义相反；且它已有 3 处调用：`alert_service.py:180` 的 `agg:event:` / `:239` 的 `agg:suitability:` 等）
 - 返回 `_NoLock` 哨兵对象，`__enter__` 返回 `False`，避免 `with None` 报错
 
-**DoD**
-- [ ] 抢到 / 抢不到 / Redis 不可用回退 三条路径单测
-- [ ] key 形态覆盖：`convert:idem:{cid_req}` / `convert:rerun:{gid}`
-- [ ] `run_locked` 的既有用例零改动
+**DoD（全部达成，见下方执行记录）**
+- [x] 抢到 / 抢不到 / Redis 不可用回退 三条路径单测
+- [x] key 形态覆盖：`convert:idem:{cid_req}` / `convert:rerun:{gid}`
+- [x] `run_locked` 的既有用例零改动
+
+**执行记录（2026-09-10）**
+
+| 项 | 内容 |
+| --- | --- |
+| 改动文件 | `app/service/risk/locks.py`【改·+`_Token`/`_NoLock` + `try_lock`】· `tests/test_locks_redis.py`【+4 用例】 |
+| 语义 | `try_lock` **单次非阻塞**：抢到返回 `_Token`（`with` 进入 True），抢不到立即 `_NoLock`（进入 False，不等、不降级）；Redis 不可用 → 退回进程内 `Lock.acquire(blocking=False)`（语义一致） |
+| 不改 | `run_locked`（D4：其会等 2s 后降级 `fn(False)`，与「抢不到立即 202」相反；3 处调用点零改动） |
+| 哨兵 | `_NoLock.__enter__` 返回 False，避免 `with None` 报错；`_Token.__exit__` 按来源释放 Redis 或进程内锁 |
+| 验证 | `pytest tests/test_locks_redis.py` 10 passed（既有 6 + 新增 4）；全量 634 passed / 3 skipped |
 
 **依赖**：无
 
diff --git a/tests/test_convert_repository.py b/tests/test_convert_repository.py
new file mode 100644
index 0000000..8201374
--- /dev/null
+++ b/tests/test_convert_repository.py
@@ -0,0 +1,141 @@
+"""T-4 `convert_repository`（agent 侧 risk_convert_detail）单测（开发计划 §5.2 DoD）。
+
+sqlite 内存库（conftest.sqlite_engine 含 agent 库表 + risk_convert_detail）；
+数据由本文件经仓储自建，不依赖真 MySQL。
+
+覆盖：
+1. 占位 → completed（阶段二回写详情，含 Decimal 折算字段）
+2. 占位 → failed
+3. 超时 → expired（list_expired_candidates + mark_expired，S2 标记不硬删）
+4. client_request_id 为 None 不受 uk_idem 约束（多次占位互不冲突）
+5. get_by_client_request_id 幂等命中 / None 直接返回 None
+6. 不与 risk_repository 混职责（convert 写操作不动 risk_alert）
+"""
+
+from __future__ import annotations
+
+from datetime import date, datetime, timedelta
+from decimal import Decimal
+
+import pytest
+from sqlalchemy import text
+
+from app.repository.convert_repository import ConvertRepository
+from app.repository.risk_repository import RiskRepository
+
+
+def _backdate_created_at(engine, group_id: str, hours_ago: int) -> None:
+    """把占位行的 created_at 回拨，模拟超时（测试不依赖真实时间漂移）。"""
+    old = datetime.now() - timedelta(hours=hours_ago)
+    with engine.begin() as conn:
+        conn.execute(
+            text("UPDATE risk_convert_detail SET created_at = :ts WHERE convert_group_id = :gid"),
+            {"ts": old, "gid": group_id},
+        )
+
+
+# ── 1. 占位 → completed ────────────────────────────────────────────────
+
+def test_placeholder_then_complete(sqlite_engine):
+    repo = ConvertRepository(engine=sqlite_engine)
+    repo.insert_placeholder("G-1", "REQ-1")
+
+    pending = repo.get_by_group_id("G-1")
+    assert pending is not None
+    assert pending["status"] == "pending"
+    assert pending["estimated"] == 0  # 真实请求，非预估单
+
+    repo.complete_convert(
+        "G-1",
+        out_trade_id="TRD-OUT-1",
+        in_trade_id="TRD-IN-1",
+        related_trade_id="TRD-REL-1",
+        nav=Decimal("1.0234"),
+        nav_date=date(2026, 9, 4),
+        fee_amount=Decimal("252.40"),
+        hold_days_min=3,
+        hold_days_max=180,
+        nav_stale=True,
+    )
+
+    done = repo.get_by_group_id("G-1")
+    assert done["status"] == "completed"
+    assert done["out_trade_id"] == "TRD-OUT-1"
+    assert done["in_trade_id"] == "TRD-IN-1"
+    assert done["related_trade_id"] == "TRD-REL-1"
+    assert done["nav"] == pytest.approx(1.0234)          # Decimal→float 绑定，读回一致
+    assert done["fee_amount"] == pytest.approx(252.40)
+    assert date.fromisoformat(str(done["nav_date"])) == date(2026, 9, 4)
+    assert done["hold_days_min"] == 3
+    assert done["hold_days_max"] == 180
+    assert done["nav_stale"] == 1
+
+
+# ── 2. 占位 → failed ───────────────────────────────────────────────────
+
+def test_placeholder_then_failed(sqlite_engine):
+    repo = ConvertRepository(engine=sqlite_engine)
+    repo.insert_placeholder("G-2", "REQ-2")
+    repo.mark_failed("G-2")
+
+    assert repo.get_by_group_id("G-2")["status"] == "failed"
+
+
+# ── 3. 超时 → expired（S2 标记不硬删） ───────────────────────────────────
+
+def test_expired_lifecycle(sqlite_engine):
+    repo = ConvertRepository(engine=sqlite_engine)
+    repo.insert_placeholder("G-3", "REQ-3")
+    _backdate_created_at(sqlite_engine, "G-3", hours_ago=48)
+
+    # 超 24h 的 pending 应被捞出
+    candidates = repo.list_expired_candidates(24)
+    assert any(c["convert_group_id"] == "G-3" for c in candidates)
+
+    repo.mark_expired("G-3")
+    row = repo.get_by_group_id("G-3")
+    assert row["status"] == "expired"          # 标记，不硬删
+    # 标记后不再出现在 pending 候选
+    assert not any(c["convert_group_id"] == "G-3" for c in repo.list_expired_candidates(24))
+
+
+# ── 4. client_request_id 为 None 不受 uk_idem 约束 ───────────────────────
+
+def test_client_request_id_none_not_unique(sqlite_engine):
+    repo = ConvertRepository(engine=sqlite_engine)
+    # 两次无幂等键占位都应成功（UNIQUE 允许多个 NULL）
+    repo.insert_placeholder("G-4", None)
+    repo.insert_placeholder("G-5", None)
+
+    assert repo.get_by_group_id("G-4")["status"] == "pending"
+    assert repo.get_by_group_id("G-5")["status"] == "pending"
+    # None 作为查询键永不命中
+    assert repo.get_by_client_request_id(None) is None
+
+
+# ── 5. 幂等命中查询 ─────────────────────────────────────────────────────
+
+def test_get_by_client_request_id_hit(sqlite_engine):
+    repo = ConvertRepository(engine=sqlite_engine)
+    repo.insert_placeholder("G-6", "REQ-6")
+
+    row = repo.get_by_client_request_id("REQ-6")
+    assert row is not None
+    assert row["convert_group_id"] == "G-6"
+
+    assert repo.get_by_client_request_id("REQ-NOPE") is None
+
+
+# ── 6. 不与 risk_repository 混职责 ──────────────────────────────────────
+
+def test_no_cross_write_to_risk_tables(sqlite_engine):
+    """convert 占位只写 risk_convert_detail，绝不污染 risk_alert（职责隔离）。"""
+    repo = ConvertRepository(engine=sqlite_engine)
+    repo.insert_placeholder("G-7", "REQ-7")
+
+    with sqlite_engine.connect() as conn:
+        alert_count = conn.execute(text("SELECT COUNT(*) FROM risk_alert")).scalar_one()
+    assert alert_count == 0
+    # 类型隔离：两个仓储是不同类，各自只碰自己的表
+    assert isinstance(repo, ConvertRepository)
+    assert not isinstance(repo, RiskRepository)
diff --git a/tests/test_locks_redis.py b/tests/test_locks_redis.py
index 4e1e6e7..0cefcca 100644
--- a/tests/test_locks_redis.py
+++ b/tests/test_locks_redis.py
@@ -136,3 +136,40 @@ def test_three_call_site_keys_have_lock_prefix(gateway):
     assert "lock:agg:event:CUST-X:2026-09-09" in redis_keys
     assert "lock:agg:suitability:CUST-X:P-Y:2026-09-09" in redis_keys
     assert "lock:l3:CUST-X" in redis_keys
+
+
+# ── T-5 · try_lock（单次非阻塞抢锁，按需用于 convert 幂等/重跑） ────────────
+
+
+def test_try_lock_acquired_idem_key(gateway):
+    """Redis 抢到 → with 进入 acquired=True，key 带 lock: 前缀（convert:idem:{cid_req} 形态）。"""
+    with locks.try_lock("convert:idem:REQ-1") as acquired:
+        assert acquired is True
+    key, _token, ttl = gateway.captured[0]
+    assert key == "lock:convert:idem:REQ-1"
+    assert ttl == locks.LOCK_TTL_SECONDS
+
+
+def test_try_lock_not_acquired_is_nonblocking(gateway):
+    """锁被占用 → 立即 _NoLock（with 进入 False）；非阻塞：只尝试一次、不重试。"""
+    gateway.acquire_result = False
+    with locks.try_lock("convert:rerun:GID-1") as acquired:
+        assert acquired is False
+    # 单次尝试，未进入 run_locked 的轮询循环
+    assert len(gateway.captured) == 1
+
+
+def test_try_lock_redis_unavailable_falls_back_local(gateway):
+    """Redis 抛 ConnectionError → 归类 unavailable → 退回进程内锁，acquired=True。"""
+    gateway.raise_on_acquire = ConnectionError("redis down")
+    with locks.try_lock("convert:idem:REQ-2") as acquired:
+        assert acquired is True
+
+
+def test_try_lock_releases_redis_lock_on_exit(gateway):
+    """抢到 Redis 锁后，``with`` 退出即释放（_Token.__exit__ 调 release_lock）。"""
+    assert not gateway._store  # 初始无锁
+    with locks.try_lock("convert:rerun:GID-3") as acquired:
+        assert acquired is True
+        assert gateway.captured[0][0] in gateway._store  # 持锁期间在册
+    assert gateway.captured[0][0] not in gateway._store  # 退出后已释放