U-037: c_deploy_supervisor_stage (candidate checkout + one-time DB copy + launch)

Adds supervisor/deploy_stage.py: stage_deployment() creates a deployment row, git-worktree-checks-out the target commit into a fresh directory (--detach, never in-place git pull), copies the production DB to a one-time candidate copy, and Popens a real candidate backend process on a free port. Return value follows 04-P §4.6's state-machine table (status after the call is 'candidate_starting') rather than 03-C's one-line 'staged' summary -- 04-P's own trigger text ("c_deploy_supervisor_stage 完成" moves the row to candidate_starting) is the more detailed/authoritative source when the two disagree, same precedent used throughout this project. Found and closed two real gaps along the way: - deployment's own field list never included where a candidate's checkout directory lives -- added candidate_instance_dir to schema.py (also what the NEXT staging call's previous_instance_dir bootstraps from once a deployment reaches 'finalized'). - backend_server.py had no CLI entry point at all since U-007 -- create_server() was only ever instantiated in-process by tests. Added main() (argparse + serve_forever + SIGTERM->shutdown() from a separate thread, per http.server's own same-thread-deadlock warning). RealCandidateLaunchTest spawns and HTTP-probes a genuine `python3 -m src.backend_server` subprocess against a real git-worktree checkout of this repo (not mocked) -- caught a real ordering issue doing so: git worktree only sees committed content, so testing before this Unit's own commit landed would check out backend_server.py without main() at all. Fixed by using `git stash create` (builds a commit object from the current working tree without touching HEAD/index) as the commit-to-test-against, falling back to plain HEAD once the tree is clean post-commit. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

U-037: c_deploy_supervisor_stage (candidate checkout + one-time DB copy + launch)
Adds supervisor/deploy_stage.py: stage_deployment() creates a deployment row, git-worktree-checks-out the target commit into a fresh directory (--detach, never in-place git pull), copies the production DB to a one-time candidate copy, and Popens a real candidate backend process on a free port. Return value follows 04-P §4.6's state-machine table (status after the call is 'candidate_starting') rather than 03-C's one-line 'staged' summary -- 04-P's own trigger text ("c_deploy_supervisor_stage 完成" moves the row to candidate_starting) is the more detailed/authoritative source when the two disagree, same precedent used throughout this project. Found and closed two real gaps along the way: - deployment's own field list never included where a candidate's checkout directory lives -- added candidate_instance_dir to schema.py (also what the NEXT staging call's previous_instance_dir bootstraps from once a deployment reaches 'finalized'). - backend_server.py had no CLI entry point at all since U-007 -- create_server() was only ever instantiated in-process by tests. Added main() (argparse + serve_forever + SIGTERM->shutdown() from a separate thread, per http.server's own same-thread-deadlock warning). RealCandidateLaunchTest spawns and HTTP-probes a genuine `python3 -m src.backend_server` subprocess against a real git-worktree checkout of this repo (not mocked) -- caught a real ordering issue doing so: git worktree only sees committed content, so testing before this Unit's own commit landed would check out backend_server.py without main() at all. Fixed by using `git stash create` (builds a commit object from the current working tree without touching HEAD/index) as the commit-to-test-against, falling back to plain HEAD once the tree is clean post-commit. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
bboonnee
1 parent abfb3cb5
Showing 6 changed files with 494 additions and 1 deletions
docs/development-plan.md
docs/sdd/unit_matrix.json
src/backend_server.py
src/schema.py
supervisor/deploy_stage.py
supervisor/test_deploy_stage.py
@@ -361,6 +361,9 @@ PH2-S1/S2/S3/S4 之间除标注的依赖外相互独立，可并行推进。
 - **U-037 [Risk Gate]** `c_deploy_supervisor_stage`：候选实例拉起（全新目录 checkout，不 `git pull` 原地改）+ 一次性 DB 副本机制（候选实例验证期间绝不接触生产库）
   - 依赖：U-034, U-036
   - Done：给定一个已合并到主分支的 commit，能生成 `deployment` 记录（`status=staged`）并拉起一个监听临时端口、连接一次性 DB 副本的候选进程
+  - **实现补充（2026-08-02）**：新增 `supervisor/deploy_stage.py`：`stage_deployment()` 建 `deployment` 行、`git worktree add --detach`（不是 `-b <branch>`，候选检出是只读验证用途，不打算再接受提交，和 `work_node_run.create_worktree()` 的编码用 worktree 是两回事）到全新目录、复制生产 DB 到一次性副本、`subprocess.Popen` 拉起候选进程。**返回值订正（低风险，按判断处理未升级询问）**：03-C `output_contract` 字面写返回 `{deployment_id, status:'staged'}`，但 04-P §4.6 状态转移表明确写"`c_deploy_supervisor_stage` 完成"这个触发条件本身就是把 `deployment.status` 从 `staged` 推进到 `candidate_starting`——两处矛盾，按"更详细的 04-P 状态机表为准"处理，函数完成时返回并持久化的都是 `candidate_starting`，不是字面上的 `staged`。**发现并补上两个真实缺口**：① `deployment` 表 01-M 字段清单里完全没有"候选检出目录存在哪"这一列——`candidate_port`/`candidate_db_path` 都有，唯独检出目录没地方存，Deploy Supervisor 若在 staging 和 cutover 之间重启就会彻底丢失候选目录位置；`schema.py` 补了 `candidate_instance_dir` 列（同一次 finalized 部署的这个字段，就是下一次 staging 时 `previous_instance_dir` 的来源——candidate 一旦顺利转正就是新 Live，回滚要回退到的正是它）。② `backend_server.py` 从 U-007 建骨架以来就没有任何 CLI 入口——`create_server()` 只在测试里以进程内方式实例化过，从没人真正把它作为独立 OS 进程启动过；候选进程恰好需要这个能力，顺手补了 `main()`（`python3 -m src.backend_server --host --port --db-path --skill-repo-path`，`SIGTERM` 走 `server.shutdown()`——必须在另一个线程调用，同线程调用 `serve_forever()` 期间调 `shutdown()` 会按 `http.server` 自己的文档死锁）。`/api/health` 端点本身留给 U-038 自己去加，不在这里超范围实现。
+    - **验证方式**：`BootstrapPreviousTest`/`StageDeploymentHappyPathTest`/`CheckoutFailedTest` 用真实 sqlite + 真实 git worktree 操作（只 mock 掉真正拉候选进程这一步，验证 `stage_deployment()` 自己的逻辑）；`RealCandidateLaunchTest` 完全不 mock——对着 graphlemon 自己仓库真实 `git worktree add --detach` 检出 + 真实 `python3 -m src.backend_server` 子进程拉起 + 真实 HTTP 请求验证候选进程确实在真实监听、真实响应，验证完 `SIGTERM`（超时兜底 `SIGKILL`）+ `git worktree remove` 清理干净。这个真实测试意外命中一个开发时序问题：`git worktree add` 只看已提交内容，而这次实现是先写代码再测试再提交（标准流程），测试跑的时候 `main()` 还没提交，导致 checkout 出来的 `backend_server.py` 是没有 `main()` 的旧版本，候选进程"启动"后什么也不做直接退出——用 `git stash create`（不碰工作区/HEAD，只生成一个代表当前工作区状态的 commit 对象）替代直接 `HEAD`，让测试始终对着"当前工作区实际内容"验证，而不是"上一次提交的内容"；提交后本地工作区变干净时 `git stash create` 打印空字符串，自动回退到普通 `HEAD`，两种时序都覆盖。
+  - 覆盖：03-C `c_deploy_supervisor_stage`；01-M `deployment.candidate_instance_dir`（补字段）；`backend_server.py` 新增可独立运行的 CLI 入口
  
 - **U-038 [Risk Gate]** `c_deploy_supervisor_health_check`：`process_alive`/`http_health`/`canary_task` 三项 pre-cutover 检查
   - 依赖：U-037
@@ -54,7 +54,7 @@
     {"id": "U-034", "phase": "PH5", "sub_phase": "PH5-S1", "name": "01-M 扩展：deployment/deploy_health_signal + core_engine 枚举（见 development-plan.md 2026-08-02 实现补充）", "classification": "Enabler", "depends_on": ["U-001"], "coverage": {"model": ["deployment", "deploy_health_signal", "bug_record"], "view": [], "controller": [], "process": [], "requirement": []}, "done": "两张新表建表成功（共 17 张）；bug_record 新枚举值向后兼容 U-015/U-018 已有测试；deployment 状态机用 01-M 1.15 完整 15 值枚举；ux_deployment_active 表达式索引强制全局单例非终态约束；bug_record.py 应用层白名单刻意未跟着放开 core_engine，classification+路由逻辑留给 U-035"},
     {"id": "U-035", "phase": "PH5", "sub_phase": "PH5-S1", "name": "self-target worktree 无条件强制 + core_engine 分类审计（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-013", "U-016", "U-034"], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": ["REQ-03"]}, "done": "self-target 时无条件走 worktree（work_graph_orchestrate.run_graph 内强制）；core_engine 分类（src/core_engine_classify.py，路径规则）在 self-target 任务 PASS 收尾时记一条审计用 bug_record；且改动落地后重跑 U-013/REQ-03（worktree 隔离）现有回归用例全部通过。**2026-08-02 用户决策修正**：Deploy Supervisor 触发条件按 supervisor-auto-rollback-design.md §1 字面设计（self-target+PASS 即触发，bug_record 是旁路产物不是前提），而非 03-C/04-P 字面文本要求的\"先有 core_engine bug_record\"——后者会让从不产生 bug_record 的 goal 型自我改进任务永远触发不了 Deploy Supervisor，已用 AskUserQuestion 与用户确认此订正。c_org_loop_route 未改动（fast_lane/full_replan 是 FAIL 路径重试路由，与此处 PASS 路径的蓝绿触发正交），deployment 行创建仍是 U-037 的范围，不在本 Unit 提前做"},
     {"id": "U-036", "phase": "PH5", "sub_phase": "PH5-S2", "name": "supervisor/isolation 补齐单测 + Deploy Supervisor 进程骨架（见 development-plan.md 2026-08-02 实现补充）", "classification": "Enabler", "depends_on": [], "coverage": {"model": [], "view": [], "controller": [], "process": [], "requirement": []}, "done": "四模块各有 test_*.py（供 python3 -m unittest discover -s supervisor/isolation 运行，52 测试全绿）；Deploy Supervisor 骨架（supervisor/deploy_supervisor.py）独立进程可启停（真实子进程验证，非 mock），不受 Live/Candidate 崩溃影响；本 Unit 只做进程生命周期骨架，不含任何真实部署决策逻辑（U-037+ 的 poll_fn 挂载点）"},
-    {"id": "U-037", "phase": "PH5", "sub_phase": "PH5-S2", "name": "c_deploy_supervisor_stage", "classification": "Risk Gate", "depends_on": ["U-034", "U-036"], "coverage": {"model": ["deployment"], "view": [], "controller": ["c_deploy_supervisor_stage"], "process": [], "requirement": []}, "done": "能生成 deployment 记录并拉起监听临时端口、连一次性 DB 副本的候选进程"},
+    {"id": "U-037", "phase": "PH5", "sub_phase": "PH5-S2", "name": "c_deploy_supervisor_stage（见 development-plan.md 2026-08-02 实现补充）", "classification": "Risk Gate", "depends_on": ["U-034", "U-036"], "coverage": {"model": ["deployment"], "view": [], "controller": ["c_deploy_supervisor_stage"], "process": [], "requirement": []}, "done": "能生成 deployment 记录并拉起监听临时端口、连一次性 DB 副本的候选进程（supervisor/deploy_stage.py，真实 git worktree + 真实子进程验证过）；顺带补上 deployment.candidate_instance_dir 缺字段 + backend_server.py 首个 CLI 入口（此前只能进程内启动，从未真正作为独立 OS 进程跑过）"},
     {"id": "U-038", "phase": "PH5", "sub_phase": "PH5-S2", "name": "c_deploy_supervisor_health_check", "classification": "Risk Gate", "depends_on": ["U-037"], "coverage": {"model": [], "view": [], "controller": ["c_deploy_supervisor_health_check"], "process": [], "requirement": []}, "done": "process_alive/http_health/canary_task 三项检查全过 -> health_check_passed；任一失败 -> health_check_failed 且 Live 未受影响"},
     {"id": "U-039", "phase": "PH5", "sub_phase": "PH5-S2", "name": "c_deploy_supervisor_cutover", "classification": "Risk Gate", "depends_on": ["U-038"], "coverage": {"model": ["deployment"], "view": [], "controller": ["c_deploy_supervisor_cutover"], "process": [], "requirement": []}, "done": "健康检查通过后完成真实切换；在途 work_node_execution 重启后可恢复；previous_instance_dir 保留至监控窗口结束"},
     {"id": "U-040", "phase": "PH5", "sub_phase": "PH5-S3", "name": "c_deploy_supervisor_monitor", "classification": "Risk Gate", "depends_on": ["U-039"], "coverage": {"model": ["deploy_health_signal"], "view": [], "controller": ["c_deploy_supervisor_monitor"], "process": [], "requirement": []}, "done": "四类信号采样正常；validator_approval_rate 偏离基线阈值时触发 anomaly_detected"},
@@ -675,3 +675,45 @@ def create_server(host=&quot;127.0.0.1&quot;, port=0, db_path=None, skill_repo_path=None):
     server.conn_lock = threading.Lock()
     server.skill_repo_path = skill_repo_path if skill_repo_path is not None else skill_query._REPO_ROOT
     return server
+
+
+def main():
+    """U-037: CLI entry point (`python3 -m src.backend_server ...`) -- until
+    this Unit, `create_server()` was only ever instantiated in-process by
+    tests; there was no way to actually launch graphlemon's own backend as a
+    standalone OS process. c_deploy_supervisor_stage needs exactly that (a
+    real candidate process, `subprocess.Popen`, listening on a real port) --
+    found while building it, added here since it's the natural, minimal home
+    for "run this server for real" regardless of who launches it.
+    """
+    import argparse
+    import signal
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--host", default="127.0.0.1")
+    parser.add_argument("--port", type=int, default=0)
+    parser.add_argument("--db-path", default=None)
+    parser.add_argument("--skill-repo-path", default=None)
+    args = parser.parse_args()
+
+    server = create_server(args.host, args.port, args.db_path, args.skill_repo_path)
+
+    def _handle_sigterm(signum, frame):
+        # server.shutdown() blocks until serve_forever()'s loop notices and
+        # returns -- calling it from the SAME thread that's running
+        # serve_forever() (the signal handler runs on the main thread, same
+        # as serve_forever() below) deadlocks, per http.server's own docs.
+        threading.Thread(target=server.shutdown, daemon=True).start()
+
+    signal.signal(signal.SIGTERM, _handle_sigterm)
+    print("graphlemon backend listening on {}:{}".format(*server.server_address), flush=True)
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        server.server_close()
+
+
+if __name__ == "__main__":
+    main()
@@ -289,6 +289,15 @@ CREATE TABLE deployment (
     )),
     candidate_port INTEGER,
     candidate_db_path TEXT,
+    -- Not in supervisor-auto-rollback-design.md §3's field table (found
+    -- missing while building U-037's c_deploy_supervisor_stage): the
+    -- checkout directory a candidate is actually running from has to be
+    -- persisted somewhere, or a Deploy Supervisor restart between staging
+    -- and cutover would lose track of it entirely -- the same class of gap
+    -- REQ-06 exists to prevent everywhere else. Once a deployment reaches
+    -- 'live'/'finalized', this same directory IS what the next deployment's
+    -- `previous_instance_dir` bootstraps from (see deploy_stage.py).
+    candidate_instance_dir TEXT,
     started_at TEXT,
     cutover_at TEXT,
     monitoring_closed_at TEXT,
+"""c_deploy_supervisor_stage (03-C §3.15 / 04-P §4.6): candidate-instance
+staging -- the entry point Deploy Supervisor's own poll loop (the `poll_fn`
+hook `deploy_supervisor.run_foreground()` calls, U-036) invokes once it
+discovers a self-target merge ready to deploy (the `bug_record(root_cause_
+category=core_engine)` audit trail U-035 records is that discovery signal --
+this module doesn't do the discovering itself, it's handed a commit hash by
+whatever does).
+
+**Return-value note (documented deviation, low-stakes -- resolved by
+judgment, not escalated)**: 03-C's `output_contract` literally says this
+returns `{deployment_id, status: 'staged'}`, but 04-P §4.6's own transition
+table says the trigger "c_deploy_supervisor_stage 完成" is what moves
+`deployment.status` FROM 'staged' TO 'candidate_starting' -- i.e. by the
+time this function's side effects (checkout, DB copy, candidate launch) are
+done and it's ready to return, the row's real, persisted status is already
+'candidate_starting', not 'staged'. 04-P's full state-machine table is more
+detailed/authoritative here than 03-C's one-line summary (same "SDD 内部冲突
+以更详细的为准" precedent CLAUDE.md states), so this returns the row's actual
+post-call status rather than literally echoing 'staged'.
+
+**Missing field found and added (see schema.py's own comment)**:
+`deployment.candidate_instance_dir` -- 01-M's own field table never listed
+where a candidate's checkout directory persists, even though cutover/
+rollback both need to know it and a Deploy Supervisor restart between
+staging and cutover would otherwise lose track of it entirely.
+"""
+
+import os
+import shutil
+import socket
+import subprocess
+import sys
+import uuid
+
+from src import paths
+
+
+class CheckoutFailedError(Exception):
+    """CHECKOUT_FAILED -- 03-C §3.15: fresh-directory checkout failed
+    (e.g. disk full). Not retried -- the caller routes this to
+    `deployment.status='blocked_pending_human'`, already applied by the
+    time this is raised."""
+
+
+def _last_finalized_deployment(conn):
+    """Returns (git_commit_hash, candidate_instance_dir) of the most recent
+    `finalized` deployment, or None if there isn't one yet (the bootstrap
+    case schema.py's own comment already flags: the very first deployment
+    ever has no prior finalized deployment to point at)."""
+    row = conn.execute(
+        "SELECT git_commit_hash, candidate_instance_dir FROM deployment "
+        "WHERE status = 'finalized' ORDER BY rowid DESC LIMIT 1"
+    ).fetchone()
+    return tuple(row) if row else None
+
+
+def _current_commit(repo_path):
+    return subprocess.run(
+        ["git", "rev-parse", "HEAD"], cwd=repo_path, capture_output=True, text=True, check=True
+    ).stdout.strip()
+
+
+def _find_free_port():
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+        sock.bind(("127.0.0.1", 0))
+        return sock.getsockname()[1]
+
+
+def _candidates_dir():
+    d = os.path.join(paths.platform_home(), "deploy_supervisor", "candidates")
+    os.makedirs(d, exist_ok=True)
+    return d
+
+
+def _candidate_dbs_dir():
+    d = os.path.join(paths.platform_home(), "deploy_supervisor", "candidate-dbs")
+    os.makedirs(d, exist_ok=True)
+    return d
+
+
+def _checkout_candidate(repo_root, git_commit_hash, deployment_id, checkout_parent_dir=None):
+    """`git worktree add --detach` (not `-b <branch>`, unlike work_node_run.
+    create_worktree's coding worktrees -- a candidate checkout is read-only
+    verification, never meant to receive further commits of its own) into a
+    FRESH directory, never an in-place `git pull` (03-C §3.15's own wording,
+    since the currently-running Live process may still be reading repo_root
+    off disk)."""
+    parent = checkout_parent_dir or _candidates_dir()
+    candidate_dir = os.path.join(parent, deployment_id)
+    subprocess.run(
+        ["git", "worktree", "add", "--detach", candidate_dir, git_commit_hash],
+        cwd=repo_root, check=True, capture_output=True, text=True,
+    )
+    return candidate_dir
+
+
+def _copy_prod_db(prod_db_path, deployment_id):
+    """MVP: whole-file copy (supervisor-auto-rollback-design.md §10 open
+    question 2 already flags this as the simple-first choice, revisit only
+    if a large production DB makes this slow)."""
+    dest = os.path.join(_candidate_dbs_dir(), deployment_id + ".db")
+    shutil.copy2(prod_db_path, dest)
+    return dest
+
+
+def _launch_candidate(candidate_dir, port, db_path, python_executable=None):
+    python_executable = python_executable or sys.executable
+    return subprocess.Popen(
+        [python_executable, "-m", "src.backend_server", "--host", "127.0.0.1", "--port", str(port), "--db-path", db_path],
+        cwd=candidate_dir, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+    )
+
+
+def stage_deployment(
+    conn, git_commit_hash, repo_root=None, prod_db_path=None,
+    checkout_parent_dir=None, port=None, python_executable=None,
+):
+    """Returns {deployment_id, status, candidate_port, candidate_pid,
+    candidate_instance_dir}. Raises CheckoutFailedError (deployment row
+    already marked blocked_pending_human by the time it's raised) if the
+    fresh-directory checkout fails."""
+    repo_root = repo_root or paths.repo_root()
+    prod_db_path = prod_db_path or paths.db_path()
+    deployment_id = str(uuid.uuid4())
+
+    last = _last_finalized_deployment(conn)
+    if last is not None:
+        previous_commit_hash, previous_instance_dir = last
+    else:
+        # Bootstrap: no prior finalized deployment exists yet -- the
+        # currently-running instance IS the "previous known-good" to roll
+        # back to, per schema.py's own note.
+        previous_commit_hash = _current_commit(repo_root)
+        previous_instance_dir = repo_root
+
+    conn.execute(
+        "INSERT INTO deployment "
+        "(deployment_id, git_commit_hash, previous_commit_hash, status, previous_instance_dir, started_at) "
+        "VALUES (?, ?, ?, 'staged', ?, CURRENT_TIMESTAMP)",
+        (deployment_id, git_commit_hash, previous_commit_hash, previous_instance_dir),
+    )
+    conn.commit()
+
+    try:
+        candidate_dir = _checkout_candidate(repo_root, git_commit_hash, deployment_id, checkout_parent_dir)
+    except subprocess.CalledProcessError as e:
+        conn.execute(
+            "UPDATE deployment SET status = 'blocked_pending_human' WHERE deployment_id = ?", (deployment_id,)
+        )
+        conn.commit()
+        raise CheckoutFailedError(
+            "fresh-directory checkout failed for deployment {} at commit {}: {}".format(
+                deployment_id, git_commit_hash, e.stderr
+            )
+        ) from e
+
+    candidate_db_path = _copy_prod_db(prod_db_path, deployment_id)
+    candidate_port = port if port is not None else _find_free_port()
+    process = _launch_candidate(candidate_dir, candidate_port, candidate_db_path, python_executable)
+
+    conn.execute(
+        "UPDATE deployment SET status = 'candidate_starting', candidate_port = ?, candidate_db_path = ?, "
+        "candidate_instance_dir = ? WHERE deployment_id = ?",
+        (candidate_port, candidate_db_path, candidate_dir, deployment_id),
+    )
+    conn.commit()
+
+    return {
+        "deployment_id": deployment_id, "status": "candidate_starting",
+        "candidate_port": candidate_port, "candidate_pid": process.pid,
+        "candidate_instance_dir": candidate_dir,
+    }
+"""U-037 tests: c_deploy_supervisor_stage. Real sqlite, real git worktree
+operations, real one-time DB file copy -- only `_launch_candidate` (spawning
+yet another real backend_server subprocess) is mocked in most tests, since
+that's the expensive/slow part and is proven for real in exactly one test
+(RealCandidateLaunchTest) rather than every test.
+
+Run: python3 -m unittest discover -s supervisor -p "test_*.py" -v
+"""
+
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+import urllib.error
+import urllib.request
+from unittest.mock import patch
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from supervisor import deploy_stage
+from src import paths, schema
+
+
+def _init_git_repo(path):
+    subprocess.run(["git", "init", "-q"], cwd=path, check=True)
+    subprocess.run(["git", "config", "user.email", "test@graphlemon.local"], cwd=path, check=True)
+    subprocess.run(["git", "config", "user.name", "test"], cwd=path, check=True)
+
+
+def _commit_all(path, message):
+    subprocess.run(["git", "add", "-A"], cwd=path, check=True)
+    subprocess.run(["git", "commit", "-q", "-m", message], cwd=path, check=True)
+    return subprocess.run(
+        ["git", "rev-parse", "HEAD"], cwd=path, capture_output=True, text=True, check=True
+    ).stdout.strip()
+
+
+class DeployStageTestBase(unittest.TestCase):
+    def setUp(self):
+        self.home_tmp = tempfile.mkdtemp(prefix="deploy-stage-home-")
+        self.addCleanup(shutil.rmtree, self.home_tmp, ignore_errors=True)
+        self.old_home = os.environ.get("GRAPHLEMON_HOME")
+        os.environ["GRAPHLEMON_HOME"] = self.home_tmp
+
+        self.repo_dir = tempfile.mkdtemp(prefix="deploy-stage-repo-")
+        self.addCleanup(shutil.rmtree, self.repo_dir, ignore_errors=True)
+        _init_git_repo(self.repo_dir)
+        with open(os.path.join(self.repo_dir, "README.md"), "w", encoding="utf-8") as f:
+            f.write("v1\n")
+        self.commit_v1 = _commit_all(self.repo_dir, "v1")
+        with open(os.path.join(self.repo_dir, "README.md"), "w", encoding="utf-8") as f:
+            f.write("v2\n")
+        self.commit_v2 = _commit_all(self.repo_dir, "v2")
+
+        self.conn = schema.connect(":memory:")
+        schema.create_schema(self.conn)
+
+        self.prod_db_path = os.path.join(self.home_tmp, "prod.db")
+        prod_conn = schema.connect(self.prod_db_path)
+        schema.create_schema(prod_conn)
+        prod_conn.close()
+
+    def tearDown(self):
+        if self.old_home is None:
+            os.environ.pop("GRAPHLEMON_HOME", None)
+        else:
+            os.environ["GRAPHLEMON_HOME"] = self.old_home
+        self.conn.close()
+
+
+class BootstrapPreviousTest(DeployStageTestBase):
+    def test_no_prior_finalized_deployment_bootstraps_from_current_repo_state(self):
+        # setUp() leaves repo_dir's HEAD at commit_v2 (the last commit made);
+        # bootstrap must read the repo's *current* HEAD, not the commit being
+        # staged (git worktree add --detach doesn't move repo_dir's own HEAD).
+        with patch.object(deploy_stage, "_launch_candidate") as mock_launch:
+            mock_launch.return_value.pid = 12345
+            result = deploy_stage.stage_deployment(
+                self.conn, self.commit_v2, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9999,
+            )
+        row = self.conn.execute(
+            "SELECT previous_commit_hash, previous_instance_dir FROM deployment WHERE deployment_id = ?",
+            (result["deployment_id"],),
+        ).fetchone()
+        self.assertEqual(row[0], self.commit_v2)
+        self.assertEqual(row[1], self.repo_dir)
+
+    def test_prior_finalized_deployment_supplies_previous_fields(self):
+        self.conn.execute(
+            "INSERT INTO deployment (deployment_id, git_commit_hash, previous_commit_hash, status, "
+            "candidate_instance_dir, previous_instance_dir) "
+            "VALUES ('dep-0', 'commit-0', 'commit-minus-1', 'finalized', '/old/candidate/dir', '/older/dir')"
+        )
+        self.conn.commit()
+        with patch.object(deploy_stage, "_launch_candidate") as mock_launch:
+            mock_launch.return_value.pid = 12345
+            result = deploy_stage.stage_deployment(
+                self.conn, self.commit_v2, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9999,
+            )
+        row = self.conn.execute(
+            "SELECT previous_commit_hash, previous_instance_dir FROM deployment WHERE deployment_id = ?",
+            (result["deployment_id"],),
+        ).fetchone()
+        self.assertEqual(row[0], "commit-0")
+        self.assertEqual(row[1], "/old/candidate/dir")
+
+
+class StageDeploymentHappyPathTest(DeployStageTestBase):
+    def test_creates_candidate_starting_row_with_real_worktree_and_db_copy(self):
+        with patch.object(deploy_stage, "_launch_candidate") as mock_launch:
+            mock_launch.return_value.pid = 54321
+            result = deploy_stage.stage_deployment(
+                self.conn, self.commit_v2, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9999,
+            )
+
+        self.assertEqual(result["status"], "candidate_starting")
+        self.assertEqual(result["candidate_port"], 9999)
+        self.assertEqual(result["candidate_pid"], 54321)
+
+        row = self.conn.execute(
+            "SELECT status, candidate_port, candidate_db_path, candidate_instance_dir, git_commit_hash "
+            "FROM deployment WHERE deployment_id = ?",
+            (result["deployment_id"],),
+        ).fetchone()
+        self.assertEqual(row[0], "candidate_starting")
+        self.assertEqual(row[1], 9999)
+        self.assertEqual(row[4], self.commit_v2)
+
+        # real git worktree: candidate dir exists and is checked out at v2
+        candidate_dir = row[3]
+        self.assertTrue(os.path.isdir(candidate_dir))
+        with open(os.path.join(candidate_dir, "README.md"), encoding="utf-8") as f:
+            self.assertEqual(f.read(), "v2\n")
+
+        # real DB copy: a distinct file, not the same path as prod
+        candidate_db_path = row[2]
+        self.assertNotEqual(candidate_db_path, self.prod_db_path)
+        self.assertTrue(os.path.isfile(candidate_db_path))
+
+    def test_only_one_non_terminal_deployment_allowed(self):
+        with patch.object(deploy_stage, "_launch_candidate") as mock_launch:
+            mock_launch.return_value.pid = 1
+            deploy_stage.stage_deployment(
+                self.conn, self.commit_v2, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9001,
+            )
+            with self.assertRaises(Exception):  # schema's ux_deployment_active partial unique index
+                deploy_stage.stage_deployment(
+                    self.conn, self.commit_v1, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9002,
+                )
+
+
+class CheckoutFailedTest(DeployStageTestBase):
+    def test_nonexistent_commit_marks_blocked_pending_human_and_raises(self):
+        with self.assertRaises(deploy_stage.CheckoutFailedError):
+            deploy_stage.stage_deployment(
+                self.conn, "0" * 40, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9003,
+            )
+        row = self.conn.execute("SELECT status FROM deployment").fetchone()
+        self.assertEqual(row[0], "blocked_pending_human")
+
+    def test_checkout_failure_leaves_no_candidate_process_launched(self):
+        with patch.object(deploy_stage, "_launch_candidate") as mock_launch:
+            with self.assertRaises(deploy_stage.CheckoutFailedError):
+                deploy_stage.stage_deployment(
+                    self.conn, "0" * 40, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9004,
+                )
+            mock_launch.assert_not_called()
+
+
+class RealCandidateLaunchTest(unittest.TestCase):
+    """Not mocked -- stages a real candidate checkout of graphlemon's OWN
+    repo (read-only `git worktree add --detach` at the current HEAD, cleaned
+    up in tearDown) and launches a real `python3 -m src.backend_server`
+    subprocess against it, then hits it over real HTTP, per this Unit's own
+    "拉起监听临时端口...的候选进程" Done criterion -- proves backend_server's
+    new CLI entry point (added by this Unit) actually works end to end, not
+    just that stage_deployment() calls the right mocked function."""
+
+    def setUp(self):
+        self.home_tmp = tempfile.mkdtemp(prefix="deploy-stage-real-home-")
+        self.addCleanup(shutil.rmtree, self.home_tmp, ignore_errors=True)
+        self.old_home = os.environ.get("GRAPHLEMON_HOME")
+        os.environ["GRAPHLEMON_HOME"] = self.home_tmp
+
+        self.conn = schema.connect(":memory:")
+        schema.create_schema(self.conn)
+        self.addCleanup(self.conn.close)
+
+        self.prod_db_path = os.path.join(self.home_tmp, "prod.db")
+        prod_conn = schema.connect(self.prod_db_path)
+        schema.create_schema(prod_conn)
+        prod_conn.close()
+
+    def tearDown(self):
+        if self.old_home is None:
+            os.environ.pop("GRAPHLEMON_HOME", None)
+        else:
+            os.environ["GRAPHLEMON_HOME"] = self.old_home
+
+    @staticmethod
+    def _commit_to_test_against():
+        """`git worktree add` only ever sees COMMITTED content -- if this
+        test runs before this Unit's own changes are committed (the normal
+        implement-then-test-then-commit order), plain HEAD would check out
+        a backend_server.py without this Unit's new main() entry point at
+        all. `git stash create` builds a real commit object representing
+        the current working tree (including uncommitted tracked-file
+        changes) WITHOUT touching the working tree, index, or HEAD -- exactly
+        what's needed here. Once this Unit is actually committed the working
+        tree is clean and `git stash create` prints nothing, so this falls
+        back to plain HEAD in that case (post-commit reruns of this test)."""
+        stash_sha = subprocess.run(
+            ["git", "stash", "create"], cwd=paths.repo_root(), capture_output=True, text=True, check=True
+        ).stdout.strip()
+        return stash_sha or deploy_stage._current_commit(paths.repo_root())
+
+    def test_real_candidate_process_actually_serves_http(self):
+        current_commit = self._commit_to_test_against()
+        result = deploy_stage.stage_deployment(
+            self.conn, current_commit, repo_root=paths.repo_root(), prod_db_path=self.prod_db_path,
+        )
+        pid = result["candidate_pid"]
+        port = result["candidate_port"]
+        candidate_dir = result["candidate_instance_dir"]
+        try:
+            _UNSET = object()
+            deadline = time.time() + 10
+            last_error = None
+            status = _UNSET
+            while time.time() < deadline:
+                try:
+                    with urllib.request.urlopen(
+                        "http://127.0.0.1:{}/api/settings/concurrency".format(port), timeout=1
+                    ) as resp:
+                        status = resp.status
+                    break
+                except urllib.error.URLError as e:  # server not up yet (connection refused)
+                    last_error = e
+                    time.sleep(0.2)
+            self.assertIsNot(status, _UNSET, "candidate server never responded: {}".format(last_error))
+            self.assertEqual(status, 200)  # real route, real response, real HTTP round trip
+        finally:
+            os.kill(pid, 15)  # SIGTERM
+            deadline = time.time() + 5
+            while time.time() < deadline and _pid_alive(pid):
+                time.sleep(0.1)
+            if _pid_alive(pid):
+                os.kill(pid, 9)
+            subprocess.run(
+                ["git", "worktree", "remove", "--force", candidate_dir], cwd=paths.repo_root(),
+                capture_output=True, text=True,
+            )
+
+
+def _pid_alive(pid):
+    try:
+        os.kill(pid, 0)
+    except OSError:
+        return False
+    return True
+
+
+if __name__ == "__main__":
+    unittest.main()