"""サーバ上で実行するシェルコマンド文字列と ssh 引数を組み立てる純関数群.
`ssh <user>@<address> '<command>'` で送る文字列を作る．リモートのログインシェル（Ubuntu では
bash）が解釈する前提で POSIX sh の範囲で書き，ジョブ本体だけは `${PIPESTATUS[0]}` を使うため
`bash -c` で包む．ユーザー由来の値（パス・コマンド引数）はすべて `shlex.quote` でクォートし，
ジョブ名・環境変数名・RUN_ID は書式を検証してから埋め込む．
Windows から ssh へ渡す引数に改行を含めないため，複数の文は `; ` で 1 行につなぐ．
リモート側が出すメッセージは ASCII に限る（ssh の stderr は Windows コンソールへ素通しされ，
コードページによっては日本語が化けるため．日本語の説明は呼び出し側の CLI が出す）．
"""
from __future__ import annotations
import re
import shlex
from collections.abc import Mapping, Sequence
from tias_judge_model.schema import DATA_ROOT_ENV_VAR
REMOTE_RUNS_DIR = "runs"
REMOTE_LOG_DIR = f"{REMOTE_RUNS_DIR}/_remote_logs"
# リモートスクリプトが返す終了コード．ssh 自身の 255・tar の 1/2 と衝突しない値にする
EXIT_REMOTE_DIR_MISSING = 90
EXIT_SESSION_EXISTS = 91
EXIT_NOT_FOUND = 92
EXIT_NO_SESSION = 93
EXIT_UNSAFE_PATH = 94
# probe コマンドの出力を区切る目印（ログインシェルが余計な出力をしても区切りで読み分ける）
PROBE_SESSIONS_MARKER = "@@tias-sessions"
PROBE_GPUS_MARKER = "@@tias-gpus"
PROBE_GPU_RC_PREFIX = "@@tias-gpu-rc "
PROBE_SAMPLES_MARKER = "@@tias-samples"
PROBE_PMON_MARKER = "@@tias-pmon"
PROBE_APPS_MARKER = "@@tias-apps"
PROBE_APPS_RC_PREFIX = "@@tias-apps-rc "
PROBE_OWNERS_MARKER = "@@tias-owners"
PROBE_END_MARKER = "@@tias-end"
PROBE_JOB_PREFIX = "job "
PROBE_OTHER_PREFIX = "other "
NVIDIA_SMI_QUERY = (
"nvidia-smi --query-gpu=index,uuid,name,memory.used,memory.total,utilization.gpu"
" --format=csv,noheader,nounits"
)
# 観測用のサンプリング（-lms <間隔> を付けて繰り返し出力させ，timeout で打ち切る）
NVIDIA_SMI_SAMPLE_QUERY = (
"nvidia-smi --query-gpu=index,utilization.gpu,memory.used --format=csv,noheader,nounits"
)
# プロセス別の SM 使用率（1 秒間隔．-c <回数> を付けて観測秒数ぶん取る）
NVIDIA_SMI_PMON_COMMAND = "nvidia-smi pmon -s u -d 1"
# 計算プロセスのみ（画面描画の Xorg 等は含まない）．gpu_uuid で上の GPU と対応付ける
NVIDIA_SMI_APPS_QUERY = (
"nvidia-smi --query-compute-apps=gpu_uuid,pid,used_memory --format=csv,noheader,nounits"
)
# tmux のセッション名に使えない "." ":" を避け，ログのファイル名にもそのまま使える文字に限る
_JOB_NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$")
_ENV_KEY_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
# ssh の接続先（ユーザー名・アドレス・踏み台）に許す文字．空白や `-` 始まりでオプションとして
# 解釈されるのを防ぐ
_SSH_TOKEN_RE = re.compile(r"^[A-Za-z0-9_.][A-Za-z0-9_.@:%-]*$")
_NOW = "\"$(date '+%Y-%m-%d %H:%M:%S')\""
def quote_remote_path(path: str) -> str:
"""リモートのパスをシェル用にクォートする.
先頭の `~` はリモートの `$HOME` に展開させる必要があるため，`~/...` は
`"$HOME"/'...'` の形に変換する（丸ごとクォートするとチルダ展開が起きない）．
Args:
path: リモートのパス（`~/...`・絶対パス・ホーム基準の相対パス）．
Returns:
シェルにそのまま埋め込める文字列．
Raises:
ValueError: 空文字，または `~ユーザー名/...` 形式の場合．
"""
if not path or not path.strip():
raise ValueError("リモートパスが空である")
if path == "~":
return '"$HOME"'
if path.startswith("~/"):
rest = path[2:]
return f'"$HOME"/{shlex.quote(rest)}' if rest else '"$HOME"'
if path.startswith("~"):
raise ValueError(
f"~ユーザー名 形式のリモートパスには対応していない: {path!r}（~/... か絶対パスを指定）"
)
return shlex.quote(path)
def validate_ssh_token(value: str, *, what: str) -> str:
"""ssh の接続先に使う文字列（ユーザー名・アドレス・踏み台）を検証して返す.
Args:
value: 検証する文字列．
what: エラーメッセージに出す項目名．
Returns:
検証済みの文字列．
Raises:
ValueError: 空・`-` 始まり・空白や記号を含む場合．
"""
if not _SSH_TOKEN_RE.match(value):
raise ValueError(f"{what} が不正である: {value!r}（英数字・. _ - @ : % のみ）")
return value
def validate_job_name(name: str) -> str:
"""ジョブ名（tmux セッション名・ログのファイル名）を検証して返す.
Args:
name: ジョブ名．
Returns:
検証済みのジョブ名．
Raises:
ValueError: 英数字で始まり英数字・`_`・`-` のみ（1〜64 文字）でない場合．
"""
if not _JOB_NAME_RE.match(name):
raise ValueError(
f"ジョブ名が不正である: {name!r}（英数字で始まり，英数字・_・- のみ，64 文字以内）"
)
return name
def validate_env_key(key: str) -> str:
"""環境変数名を検証して返す.
Args:
key: 環境変数名．
Returns:
検証済みの環境変数名．
Raises:
ValueError: 環境変数名として不正な場合．
"""
if not _ENV_KEY_RE.match(key):
raise ValueError(f"環境変数名が不正である: {key!r}")
return key
def parse_env_assignment(text: str) -> tuple[str, str]:
"""`KEY=VALUE` 形式の環境変数指定を分解する.
Args:
text: `KEY=VALUE` 形式の文字列（VALUE は空でもよい）．
Returns:
(KEY, VALUE)．
Raises:
ValueError: `=` が無い，または KEY が環境変数名として不正な場合．
"""
key, sep, value = text.partition("=")
if not sep or not _ENV_KEY_RE.match(key):
raise ValueError(f"環境変数の指定が不正である: {text!r}（KEY=VALUE 形式）")
return key, value
def build_job_env(gpu_index: int, extra: Sequence[str]) -> dict[str, str]:
"""ジョブに渡す環境変数（値をそのまま渡すもの）を組み立てる.
`CUDA_DEVICE_ORDER=PCI_BUS_ID` を付けるのは，CUDA の既定の番号付け（速い順）が
nvidia-smi の index（PCI バス順）と異なりうるため（異なると空きと判定した GPU とは別の
GPU を使ってしまう）．`TIAS_DATA_ROOT` はパスとして別に渡す（`build_job_script` の
`path_env`）．GPU の割り当てを迂回させないため，`--env` で予約済みの変数は上書きできない．
Args:
gpu_index: 選んだ GPU の番号（nvidia-smi の index）．
extra: `--env` で指定された `KEY=VALUE`（後勝ち）．
Returns:
環境変数の辞書（挿入順＝ログに表示する順）．
Raises:
ValueError: GPU 番号が負，`KEY=VALUE` が不正，または予約済みの変数を指定した場合．
"""
if gpu_index < 0:
raise ValueError(f"GPU 番号は 0 以上: {gpu_index}")
env = {
"CUDA_DEVICE_ORDER": "PCI_BUS_ID",
"CUDA_VISIBLE_DEVICES": str(gpu_index),
# パイプ越しでも Python の出力がすぐログに出るようにする（既定はブロックバッファ）
"PYTHONUNBUFFERED": "1",
}
reserved = {*env, DATA_ROOT_ENV_VAR}
for item in extra:
key, value = parse_env_assignment(item)
if key in reserved:
raise ValueError(f"{key} は tias-remote が設定するため --env では指定できない")
env[key] = value
return env
def build_ssh_argv(
ssh: str,
*,
user: str,
address: str,
jump_host: str | None,
connect_timeout_sec: int,
remote_command: str,
stdin: bool = False,
) -> list[str]:
"""ssh を起動する引数リストを返す.
常に `BatchMode=yes`（パスワード入力を求めず失敗させる）と `ConnectTimeout` を付ける．
標準入力を使わないコマンドには `-n` を付ける（標準入力が開いたままだと応答せず止まる
ことがあるため．ENV_02 トラブルシューティング）．
Args:
ssh: ssh 実行ファイル．
user: 接続ユーザー．
address: 接続先アドレス．
jump_host: 踏み台（`-J` に渡す）．None なら直接接続．
connect_timeout_sec: 接続タイムアウト（秒）．
remote_command: リモートで実行するシェルコマンド（1 引数として渡す）．
stdin: True なら標準入力をリモートへ渡す（`-n` を付けない）．
Returns:
subprocess に渡す引数リスト．
Raises:
ValueError: 接続先の書式が不正，またはタイムアウトが 1 未満の場合．
"""
if connect_timeout_sec < 1:
raise ValueError(f"接続タイムアウトは 1 以上でなければならない（{connect_timeout_sec}）")
destination = (
f"{validate_ssh_token(user, what='ユーザー名')}@"
f"{validate_ssh_token(address, what='アドレス')}"
)
argv = [ssh, "-o", "BatchMode=yes", "-o", f"ConnectTimeout={connect_timeout_sec}"]
if jump_host is not None:
argv += ["-J", validate_ssh_token(jump_host, what="踏み台")]
if not stdin:
argv.append("-n")
argv += [destination, remote_command]
return argv
def _cd_command(remote_dir: str) -> str:
"""リモートディレクトリへ移動し，無ければ専用の終了コードで抜ける文を返す."""
quoted = quote_remote_path(remote_dir)
return (
f"cd {quoted} || {{ echo 'tias-remote: remote directory not found:' {quoted} >&2; "
f"exit {EXIT_REMOTE_DIR_MISSING}; }}"
)
def job_log_path(name: str) -> str:
"""ジョブログのパス（remote_dir 基準．クォート前）を返す.
Args:
name: 検証済みのジョブ名（`validate_job_name`）．
Returns:
`runs/_remote_logs/<name>.log`．
"""
return f"{REMOTE_LOG_DIR}/{name}.log"
def _tmux_target(name: str) -> str:
"""tmux の `-t` に渡すセッション指定をクォートして返す．name は検証済みとする.
`=` を前置して完全一致にする（付けないと tmux は前方一致で別のジョブを選びうる）．
"""
return shlex.quote(f"={name}")
def build_probe_command(remote_dir: str, *, observe_sec: int, sample_interval_ms: int) -> str:
"""tmux セッション・GPU の観測・計算プロセスとその所有者を 1 回の ssh で取るコマンドを返す.
出力の形式（`parse_probe_output` が解釈する）:
- `@@tias-sessions` の後に，tmux セッション 1 件につき 1 行．このツールのジョブログ
（`remote_dir/runs/_remote_logs/<名前>.log`）があるものは `job <名前>`，無いものは
`other <名前>`（同じユーザーの別用途のセッション）
- `@@tias-gpus` の後に GPU の一覧（`nvidia-smi --query-gpu` の CSV），続けて
`@@tias-gpu-rc <終了コード>`（nvidia-smi が無ければ 127 で，以降の節は出ない）
- `observe_sec > 0` のとき，`@@tias-samples` の後に観測中のサンプル（`index, 使用率,
使用中メモリ` の CSV．`sample_interval_ms` 間隔），`@@tias-pmon` の後に
`nvidia-smi pmon -s u` の出力（1 秒間隔．失敗・未対応なら空）
- `@@tias-apps` の後に計算プロセス（`nvidia-smi --query-compute-apps` の CSV），続けて
`@@tias-apps-rc <終了コード>`
- `@@tias-owners` の後に，計算プロセス 1 件につき `<pid> <所有ユーザー> <経過秒>`
（ps で取れなければ pid だけ）
- 最後に `@@tias-end`（出力が途中で切れていないことの確認用）
remote_dir が無くても失敗しない（まだ push していないサーバも GPU 一覧に出すため）．
所要時間はおよそ observe_sec 秒（観測はサーバ側で行う）．
Args:
remote_dir: リモートのプロジェクトルート．
observe_sec: 観測秒数（0 なら観測せず瞬間値だけを取る）．
sample_interval_ms: 観測の間隔（ミリ秒）．
Returns:
リモートで実行するシェルコマンド．
Raises:
ValueError: 観測秒数が負，または間隔が 1 未満の場合．
"""
log_dir = f"{quote_remote_path(remote_dir)}/{REMOTE_LOG_DIR}"
sessions = (
"tmux ls -F '#{session_name}' 2>/dev/null | while IFS= read -r s; do "
f'if [ -f {log_dir}/"$s".log ]; then echo "{PROBE_JOB_PREFIX}$s"; '
f'else echo "{PROBE_OTHER_PREFIX}$s"; fi; done'
)
if observe_sec < 0 or sample_interval_ms < 1:
raise ValueError(
f"観測の設定が不正である（observe_sec={observe_sec}, "
f"sample_interval_ms={sample_interval_ms}）"
)
# 所有者は ps の user 列を広げて取る（既定幅では長いユーザー名が UID 表記・切り詰めになる）．
# etimes は起動からの経過秒数（他ユーザーのプロセスでも取れる）
owners = (
"printf '%s\\n' \"$apps\" | while IFS=, read -r _u p _m; do "
"p=$(printf '%s' \"$p\" | tr -d ' '); "
"case \"$p\" in ''|*[!0-9]*) continue;; esac; "
'echo "$p $(ps -o user:64=,etimes= -p "$p" 2>/dev/null | tr -s \' \')"; done'
)
observe = ""
if observe_sec > 0:
# GPU ごとのサンプリングと pmon（プロセス別 SM 使用率）を並行して走らせる．出力が
# 混ざらないよう一時ディレクトリのファイルに分けて書き，終わってから順に出す．
# timeout で打ち切った nvidia-smi -lms は 124 で終わるが，出力は有効なので終了コードは
# 見ない（サンプルが 0 件の GPU を状態不明にする）．pmon が止まらない場合に備えて
# pmon にも余裕を持たせた timeout を付ける
observe = (
't=$(mktemp -d 2>/dev/null); if [ -n "$t" ]; then '
f"timeout {observe_sec + 5} {NVIDIA_SMI_PMON_COMMAND} -c {observe_sec}"
' > "$t/pmon" 2>/dev/null & pp=$!; '
f"timeout {observe_sec} {NVIDIA_SMI_SAMPLE_QUERY} -lms {sample_interval_ms}"
' > "$t/samples" 2>/dev/null; wait "$pp"; '
f"echo '{PROBE_SAMPLES_MARKER}'; cat \"$t/samples\"; "
f"echo '{PROBE_PMON_MARKER}'; cat \"$t/pmon\"; "
'rm -rf "$t"; fi; '
)
gpus = (
"if command -v nvidia-smi >/dev/null 2>&1; then "
f'{NVIDIA_SMI_QUERY} 2>/dev/null; echo "{PROBE_GPU_RC_PREFIX}$?"; '
f"{observe}"
f"apps=$({NVIDIA_SMI_APPS_QUERY} 2>/dev/null); arc=$?; "
f"echo '{PROBE_APPS_MARKER}'; if [ -n \"$apps\" ]; then printf '%s\\n' \"$apps\"; fi; "
f'echo "{PROBE_APPS_RC_PREFIX}$arc"; '
f"echo '{PROBE_OWNERS_MARKER}'; {owners}; "
f"else echo '{PROBE_GPU_RC_PREFIX}127'; fi"
)
return "; ".join(
[
f"echo '{PROBE_SESSIONS_MARKER}'",
sessions,
f"echo '{PROBE_GPUS_MARKER}'",
gpus,
f"echo '{PROBE_END_MARKER}'",
]
)
def build_push_command(remote_dir: str) -> str:
"""標準入力の tar ストリームをリモートディレクトリへ展開するコマンドを返す.
Args:
remote_dir: 展開先（リモートのプロジェクトルート）．
Returns:
リモートで実行するシェルコマンド．
"""
quoted = quote_remote_path(remote_dir)
return f"mkdir -p {quoted} && tar -xf - -C {quoted}"
def build_job_script(
remote_dir: str,
name: str,
command: Sequence[str],
*,
uv_path: str,
server_name: str,
gpu_index: int,
env: Mapping[str, str],
path_env: Mapping[str, str],
) -> str:
"""tmux セッション内で bash が実行するジョブ本体のスクリプトを返す.
出力は `runs/_remote_logs/<name>.log` に追記し（前回の同名ジョブのログも残す），
開始行（サーバ名・GPU・日時）と終了コード行で挟む．標準入力は /dev/null にする
（入力待ちで止まらないように）．非対話 ssh では `~/.local/bin` が PATH に無いため，
uv のディレクトリと `$HOME/.local/bin` を PATH に足す．
Args:
remote_dir: リモートのプロジェクトルート．
name: ジョブ名．
command: `uv run` の後ろに付けて実行するコマンドと引数．
uv_path: リモートの uv 実行ファイル．
server_name: ログの開始行に書くサーバ名．
gpu_index: ログの開始行に書く GPU 番号．
env: ジョブに渡す環境変数（値はそのまま渡す）．
path_env: ジョブに渡す環境変数のうち，値がリモートのパス（`~` を `$HOME` に展開する）
であるもの．
Returns:
bash スクリプト（1 行）．
Raises:
ValueError: コマンドが空，またはジョブ名・環境変数名・パスが不正な場合．
"""
if not command:
raise ValueError("実行するコマンドが空である")
log = job_log_path(validate_job_name(name))
for key in [*env, *path_env]:
validate_env_key(key)
uv_quoted = quote_remote_path(uv_path)
run_argv = [uv_quoted, "run", *(shlex.quote(arg) for arg in command)]
display = shlex.join(
[f"{k}={v}" for k, v in [*env.items(), *path_env.items()]] + ["uv", "run", *command]
)
header = f"server={server_name} gpu={gpu_index}"
statements = [
_cd_command(remote_dir),
f"log={shlex.quote(log)}",
f"printf '=== [%s] start %s host=%s: %s\\n' {_NOW} {shlex.quote(header)} "
f'"$(hostname)" {shlex.quote(display)} >> "$log"',
f'export PATH="$(dirname -- {uv_quoted})":"$HOME/.local/bin:$PATH"',
*[f"export {k}={shlex.quote(v)}" for k, v in env.items()],
*[f"export {k}={quote_remote_path(v)}" for k, v in path_env.items()],
f'{" ".join(run_argv)} < /dev/null 2>&1 | tee -a "$log"',
"rc=${PIPESTATUS[0]}",
f'printf \'=== [%s] exit %s\\n\' {_NOW} "$rc" >> "$log"',
'exit "$rc"',
]
return "; ".join(statements)
def build_run_command(remote_dir: str, name: str, job_script: str) -> str:
"""ジョブ本体を tmux の新規セッション（デタッチ状態）で起動するコマンドを返す.
同名セッションが既にあれば起動せず `EXIT_SESSION_EXISTS` で終わる．
tmux のサーバはデーモン化して ssh の入出力を閉じるため，ssh はすぐ戻り，
接続が切れてもジョブは継続する．
Args:
remote_dir: リモートのプロジェクトルート．
name: ジョブ名（tmux セッション名）．
job_script: `build_job_script` が返したスクリプト．
Returns:
リモートで実行するシェルコマンド．
Raises:
ValueError: ジョブ名が不正な場合．
"""
name = validate_job_name(name)
inner = "bash -c " + shlex.quote(job_script)
statements = [
_cd_command(remote_dir),
f"mkdir -p {REMOTE_LOG_DIR} || exit 1",
f"if tmux has-session -t {_tmux_target(name)} 2>/dev/null; then "
f"echo 'tias-remote: tmux session already exists:' {shlex.quote(name)} >&2; "
f"exit {EXIT_SESSION_EXISTS}; fi",
f'tmux new-session -d -s {shlex.quote(name)} -c "$PWD" {shlex.quote(inner)}',
]
return "; ".join(statements)
def build_log_exists_command(remote_dir: str, name: str) -> str:
"""ジョブログがあれば 0，無ければ `EXIT_NOT_FOUND` で終わるコマンドを返す.
Args:
remote_dir: リモートのプロジェクトルート．
name: ジョブ名．
Returns:
リモートで実行するシェルコマンド．
Raises:
ValueError: ジョブ名が不正な場合．
"""
log = shlex.quote(job_log_path(validate_job_name(name)))
return (
f"cd {quote_remote_path(remote_dir)} 2>/dev/null && [ -f {log} ] || exit {EXIT_NOT_FOUND}"
)
def build_logs_command(remote_dir: str, name: str, *, lines: int, follow: bool) -> str:
"""ジョブログの末尾を表示する（follow なら追従する）コマンドを返す.
追従モードでは，ssh の標準入力が閉じたら（手元で Ctrl+C・終了したら）tail を止める．
疑似端末なしの ssh では切断しても tail に SIGHUP が届かず，リモートに残り続けるため．
Args:
remote_dir: リモートのプロジェクトルート．
name: ジョブ名．
lines: 表示する末尾の行数（0 以上）．
follow: True なら追記を追従する．
Returns:
リモートで実行するシェルコマンド．
Raises:
ValueError: ジョブ名が不正，または lines が負の場合．
"""
if lines < 0:
raise ValueError(f"表示行数は 0 以上でなければならない（{lines}）")
log = shlex.quote(job_log_path(validate_job_name(name)))
check = (
f"[ -f {log} ] || {{ echo 'tias-remote: log not found:' {log} >&2; "
f"exit {EXIT_NOT_FOUND}; }}"
)
if follow:
tail = f'tail -n {lines} -F {log} & t=$!; cat >/dev/null; kill "$t" 2>/dev/null'
else:
tail = f"tail -n {lines} {log}"
return "; ".join([_cd_command(remote_dir), check, tail])
def build_kill_command(remote_dir: str, name: str) -> str:
"""ジョブの tmux セッションを終了し，ログにその旨を追記するコマンドを返す.
tmux がセッションを閉じるとジョブの bash は SIGHUP で終わり，自分では終了行を
書けないため，終了前にここで追記する．
Args:
remote_dir: リモートのプロジェクトルート．
name: ジョブ名．
Returns:
リモートで実行するシェルコマンド．
Raises:
ValueError: ジョブ名が不正な場合．
"""
name = validate_job_name(name)
target = _tmux_target(name)
log = shlex.quote(job_log_path(name))
statements = [
f"tmux has-session -t {target} 2>/dev/null || "
f"{{ echo 'tias-remote: no such tmux session:' {shlex.quote(name)} >&2; "
f"exit {EXIT_NO_SESSION}; }}",
f"if cd {quote_remote_path(remote_dir)} 2>/dev/null && [ -f {log} ]; then "
f"printf '=== [%s] killed by tias-remote kill\\n' {_NOW} >> {log}; fi",
f"tmux kill-session -t {target}",
]
return "; ".join(statements)
def build_pull_list_command(remote_dir: str, run_id: str) -> str:
"""`runs/<RUN_ID>/` 配下のディレクトリ以外のエントリを種類付きで列挙するコマンドを返す.
出力は `<種類> <runs/ 基準のパス>` を NUL で区切ったもの（GNU find の `-printf '%y %p\\0'`．
種類は `f` 通常ファイル・`l` シンボリックリンク 等）．回収対象の絞り込みは手元の
`plan_pull` が行い，ここでは何も送らない（一覧はファイル名を含むため表示しない）．
RUN_ID のディレクトリが無ければ `EXIT_NOT_FOUND`，リンクなら `EXIT_UNSAFE_PATH` で終わる．
Args:
remote_dir: リモートのプロジェクトルート．
run_id: `validate_run_id` で検証済みの RUN_ID．
Returns:
リモートで実行するシェルコマンド．
"""
run = shlex.quote(run_id)
statements = [
f"cd {quote_remote_path(remote_dir)}/{REMOTE_RUNS_DIR} 2>/dev/null"
f" || exit {EXIT_NOT_FOUND}",
f"if [ -L {run} ]; then exit {EXIT_UNSAFE_PATH}; fi",
f"[ -d {run} ] || exit {EXIT_NOT_FOUND}",
f"find {run} -mindepth 1 ! -type d -printf '%y %p\\0'",
]
return "; ".join(statements)
def build_pull_tar_command(remote_dir: str) -> str:
"""標準入力の NUL 区切りのパス（`runs/` 基準）だけを tar ストリームで書き出すコマンドを返す.
`--no-recursion` で指定したパス以外を含めない（ディレクトリを渡されても中身を入れない）．
Args:
remote_dir: リモートのプロジェクトルート．
Returns:
リモートで実行するシェルコマンド．
"""
return (
f"cd {quote_remote_path(remote_dir)}/{REMOTE_RUNS_DIR} || exit {EXIT_NOT_FOUND}; "
"tar --null --no-recursion -T - -cf -"
)
