宜博 / lemongraph

"""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 sqlite3
import subprocess
import sys
import uuid

from src import paths, schema


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."""


class ConcurrentDeploymentError(Exception):
    """CONCURRENT_DEPLOYMENT -- ux_deployment_active unique index detected an
    existing deployment row not yet in a terminal state (finalized/
    rolled_back/blocked_pending_human) when attempting to INSERT a new one.
    Unlike CheckoutFailedError, the INSERT never committed, so there is no
    half-finished deployment row to mark blocked_pending_human -- callers
    (the Deploy Supervisor polling loop) should treat this as 'skip this
    round, retry next poll', not as a failure needing human intervention.
    Diagnostic fields are real attributes, not just baked into the message:
      - git_commit_hash: the commit hash this rejected call attempted to stage.
      - blocking_deployment_id / blocking_status: the deployment row currently
        holding the slot (if it could be located via a best-effort read-only
        query); both None if it could not be located -- the exception is
        still raised in that case.
    """

    def __init__(self, git_commit_hash, blocking_deployment_id=None, blocking_status=None):
        self.git_commit_hash = git_commit_hash
        self.blocking_deployment_id = blocking_deployment_id
        self.blocking_status = blocking_status
        message = "cannot stage commit {}: a deployment is already in progress".format(git_commit_hash)
        if blocking_deployment_id is not None:
            message += " (blocking deployment_id={}, status={})".format(
                blocking_deployment_id, blocking_status
            )
        super().__init__(message)


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).

    **Bootstrap (2026-08-03, first real self-hosting run)**: the very first
    deployment ever has no `prod_db_path` to copy from yet -- nothing has
    run as a persistent service before, the same class of "no prior state"
    gap `stage_deployment()` already handles for `previous_commit_hash`/
    `previous_instance_dir`. Copying a nonexistent file raised
    `FileNotFoundError` uncaught the first time this actually ran for real.
    A fresh, correctly-schema'd db is the right substitute -- there's no
    production data to inherit yet, so an empty-but-valid db is exactly
    what a genuinely-first candidate should start from."""
    dest = os.path.join(_candidate_dbs_dir(), deployment_id + ".db")
    if os.path.exists(prod_db_path):
        shutil.copy2(prod_db_path, dest)
    else:
        conn = schema.connect(dest)
        schema.create_schema(conn)
        conn.close()
    return dest


def _launch_candidate(candidate_dir, port, db_path, python_executable=None):
    """`--disable-reconciliation-loop` (2026-08-06, task_reconciliation):
    `db_path` here is a COPY of the real production db (`_stage_candidate_db`
    above) -- it can genuinely contain resumable-status tasks. Without this
    flag, a candidate's own reconciliation loop would find them "stale" (its
    fresh process has no in-flight record for anything) and start
    dispatching real `claude -p` subprocess calls against the throwaway
    candidate db during what's supposed to be a passive health-check-only
    process, before Deploy Supervisor has even decided whether this
    candidate is good. Found for real: this exact scenario made candidate
    health checks fail/hang in `supervisor`'s own test suite the first time
    the reconciliation loop shipped with no guard here.

    `--disable-new-goal-dispatch-loop` (2026-08-13, new_goal_dispatch): same
    risk, same fix, for the sibling loop that auto-starts `status='queued'`
    tasks -- the copied db can contain those too, and a candidate must stay
    passive under health-check until Deploy Supervisor decides it's live."""
    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, "--disable-reconciliation-loop", "--disable-new-goal-dispatch-loop",
        ],
        cwd=candidate_dir, stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
    )


def _is_concurrent_deployment_conflict(error):
    """Distinguish a ux_deployment_active unique-index conflict from any
    other IntegrityError. sqlite reports named UNIQUE INDEX conflicts with
    the index name embedded in the message, e.g. "UNIQUE constraint
    failed: index 'ux_deployment_active'". Match only this exact index
    name string -- do not broadly catch all IntegrityErrors, or unrelated
    integrity errors would be misclassified and masked as concurrency
    conflicts.
    """
    return "ux_deployment_active" in str(error)


def _find_blocking_deployment(conn):
    """Read-only diagnostic query: find the non-terminal deployment row
    currently occupying the ux_deployment_active slot. Only called after a
    conflict has already been detected, purely to enrich the exception. If
    the query itself fails (e.g. a broken connection), swallow the
    sqlite3.Error and return None -- the caller still raises
    ConcurrentDeploymentError, just without blocking_deployment_id/status.
    """
    try:
        row = conn.execute(
            "SELECT deployment_id, status FROM deployment "
            "WHERE status NOT IN ('finalized', 'rolled_back', 'blocked_pending_human') "
            "ORDER BY rowid DESC LIMIT 1"
        ).fetchone()
        return tuple(row) if row else None
    except sqlite3.Error:
        return None


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

    try:
        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()
    except sqlite3.IntegrityError as e:
        conn.rollback()
        if not _is_concurrent_deployment_conflict(e):
            raise
        blocking = _find_blocking_deployment(conn)
        blocking_deployment_id, blocking_status = blocking if blocking else (None, None)
        raise ConcurrentDeploymentError(
            git_commit_hash=git_commit_hash,
            blocking_deployment_id=blocking_deployment_id,
            blocking_status=blocking_status,
        ) from e

    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,
    }