diff --git a/sanguo_backtest/result_store.py b/sanguo_backtest/result_store.py
index eaaa06b..b1dcdc1 100644
--- a/sanguo_backtest/result_store.py
+++ b/sanguo_backtest/result_store.py
@@ -1,4 +1,5 @@
 """Backtest result storage using SQLite + parquet files."""
+import os
 import sqlite3
 import json
 import logging
@@ -65,6 +66,13 @@ _SCHEMA = """CREATE TABLE IF NOT EXISTS backtest_stats (
     trades_path TEXT,
     error_msg TEXT,
     created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
+);
+
+-- 提交时间表:任务提交即写一行(backtest_stats.created_at 是存结果时=完成时间,
+-- 不是真创建时间)。完成后两表相减即真实耗时。
+CREATE TABLE IF NOT EXISTS task_submissions (
+    task_id TEXT PRIMARY KEY,
+    submitted_at TIMESTAMP
 );"""
 
 
@@ -133,6 +141,72 @@ def save_result(result: BacktestResult, db_path: str, file_dir: Optional[str] =
         conn.close()
 
 
+def record_submission(task_id: str, db_path: str) -> str:
+    """Record task submission time (UTC, same format as CURRENT_TIMESTAMP).
+
+    提交即写:真创建时间。失败只 warning 不抛——时间戳缺失只影响展示,
+    不应影响任务提交本身。
+    """
+    from datetime import datetime, timezone
+    ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
+    try:
+        conn = _connect(db_path)
+        try:
+            conn.execute(
+                "INSERT OR REPLACE INTO task_submissions (task_id, submitted_at) VALUES (?, ?)",
+                (task_id, ts),
+            )
+            conn.commit()
+        finally:
+            conn.close()
+    except Exception as e:
+        logger.warning("record_submission %s failed: %s", task_id, e)
+    return ts
+
+
+def load_submissions(db_path: str) -> dict[str, str]:
+    """Return {task_id: submitted_at(UTC)} for all recorded submissions."""
+    conn = _connect(db_path)
+    try:
+        return {
+            row[0]: row[1]
+            for row in conn.execute("SELECT task_id, submitted_at FROM task_submissions")
+        }
+    finally:
+        conn.close()
+
+
+def delete_result_by_task_id(task_id: str, db_path: str) -> int:
+    """Delete all persisted traces of a task: backtest_stats rows +
+    equity/trades JSON files + task_submissions row.
+
+    Returns deleted backtest_stats row count. Missing files tolerated
+    (historical records may reference stale paths after host migration).
+    """
+    conn = _connect(db_path)
+    try:
+        rows = conn.execute(
+            "SELECT equity_path, trades_path FROM backtest_stats WHERE task_id=?",
+            (task_id,),
+        ).fetchall()
+        for equity_path, trades_path in rows:
+            for path in (equity_path, trades_path):
+                if not path:
+                    continue
+                try:
+                    os.remove(path)
+                except OSError:
+                    pass
+        n = conn.execute(
+            "DELETE FROM backtest_stats WHERE task_id=?", (task_id,)
+        ).rowcount
+        conn.execute("DELETE FROM task_submissions WHERE task_id=?", (task_id,))
+        conn.commit()
+        return n
+    finally:
+        conn.close()
+
+
 def _safe_read_json(path: Optional[str]) -> Optional[pd.DataFrame]:
     """Read a JSON equity/trades file; return None if missing or unreadable.