diff --git a/docs/specifications/post-start.md b/docs/specifications/post-start.md
index 685ea117..d7b9b7d4 100644
--- a/docs/specifications/post-start.md
+++ b/docs/specifications/post-start.md
@@ -143,8 +143,10 @@ Container startup failed: 2 of 3 instances did not become ready.
 
 ### 起動の後の処理の段
 
-`_run_post_start(project_name, scale, indices, dev_service_name, config, compose_file, run_deploy=True)` は、
+`_run_post_start(deployment, indices, config, run_deploy=True)` は、
 後処理の対象 `indices` へ次の順で処理を行う。対象が空なら何も行わない（token も発行しない）。
+`deployment` は起動した構成（`_ComposeDeployment`。`project_name`・`scale`・`dev_service_name`・`compose_file` の組）で、
+`up` / `scale` / `project post-start` が組んで渡す。
 
 | 順 | 処理 | 条件 | 失敗したとき |
 | --- | --- | --- | --- |
diff --git a/lib/devbase/commands/container.py b/lib/devbase/commands/container.py
index 42b4220c..94c04c70 100644
--- a/lib/devbase/commands/container.py
+++ b/lib/devbase/commands/container.py
@@ -8,6 +8,7 @@
 import sys
 import time
 from contextlib import contextmanager
+from dataclasses import dataclass
 from datetime import datetime, timezone
 from pathlib import Path
 from typing import List, Optional
@@ -1273,8 +1274,21 @@ def _run_pre_up_checks(config) -> bool:
     return True
 
 
-def _run_post_start(project_name: str, scale: int, indices, dev_service_name: str,
-                    config, compose_file, run_deploy: bool = True) -> None:
+@dataclass(frozen=True)
+class _ComposeDeployment:
+    """起動した構成。``up`` / ``scale`` / ``project post-start`` が起動の後の処理の段へ渡す組。
+
+    ``compose_file`` は ``-f`` で渡すファイル (スケールのファイルが無ければ ``None``)。
+    """
+
+    project_name: str
+    scale: int
+    dev_service_name: str
+    compose_file: Optional[Path]
+
+
+def _run_post_start(deployment: _ComposeDeployment, indices, config,
+                    run_deploy: bool = True) -> None:
     """起動の後の処理の段。後処理の対象の番号へ、インスタンスごとの処理を決まった順に行う。
 
     順は、不足リポジトリの報告 → ``./deploy`` → token の配布 → 窓のタイトルの設定で、
@@ -1287,6 +1301,10 @@ def _run_post_start(project_name: str, scale: int, indices, dev_service_name: st
     indices = list(indices)
     if not indices:
         return
+    project_name = deployment.project_name
+    scale = deployment.scale
+    dev_service_name = deployment.dev_service_name
+    compose_file = deployment.compose_file
 
     # clone できなかった repo があれば伝える (揃っていれば何も出さない)。
     _report_missing_repos(config, scale, dev_service_name, project_name,
@@ -1450,8 +1468,9 @@ def cmd_up(project_name: str = None, scale: int = None,
         if startup_error is not None:
             _warn_partial_start(startup_error, started, dev_service_name)
 
-        _run_post_start(project_name, scale, started, dev_service_name, config,
-                        override_file)
+        _run_post_start(
+            _ComposeDeployment(project_name, scale, dev_service_name, override_file),
+            started, config)
 
         _maybe_open_editor(project_name, open_editor, open_index, scale,
                            config, compose_file=override_file,
@@ -1565,8 +1584,8 @@ def cmd_post_start(project_name: Optional[str] = None, context: Optional[str] =
         return 1
 
     compose_file = _SCALE_COMPOSE_FILE if _SCALE_COMPOSE_FILE.exists() else None
-    _run_post_start(project_name, scale, targets, dev_service_name, config, compose_file,
-                    run_deploy=False)
+    _run_post_start(_ComposeDeployment(project_name, scale, dev_service_name, compose_file),
+                    targets, config, run_deploy=False)
     logger.info("=== Post-start completed ===")
     return 0
 
@@ -1882,8 +1901,9 @@ def cmd_scale(new_scale: int, project_name: str = None,
         if startup_error is not None:
             _warn_partial_start(startup_error, targets, dev_service_name)
 
-        _run_post_start(project_name, new_scale, targets, dev_service_name, config,
-                        override_file)
+        _run_post_start(
+            _ComposeDeployment(project_name, new_scale, dev_service_name, override_file),
+            targets, config)
 
         if _finish_partial('Scale', startup_error, project_name):
             return 1
diff --git a/lib/devbase/commands/env.py b/lib/devbase/commands/env.py
index 7edf1c0b..560f4e6e 100644
--- a/lib/devbase/commands/env.py
+++ b/lib/devbase/commands/env.py
@@ -64,6 +64,26 @@ def __init__(self, message: str, exit_code: int = EXIT_USAGE):
         self.exit_code = exit_code
 
 
+def check_group_option(settings, group: str, *, account_rule: bool = False) -> str:
+    """``--group`` の名前を置き場のグループ名の検査 (``storage_group``) に通し、通った名前を返す。
+
+    通らなければ「``--group に使えない名前です: …``」の :class:`GroupOptionError` (終了コード 2)。
+    ``account_rule`` が真なら先に ``DEVBASE_ACCOUNT_GROUP`` の規則 (``validate_account_group``) を
+    通し、その結果 (前後の空白を外した名前) を検査して返す。CLI は ``storage_group`` の中で同じ
+    規則を通すため前段を持たず、名前の規則の誤りの文に「グループ名: 」が付く。TUI の選択は前段を
+    通し、接頭の無い文を出す。
+    """
+    from devbase.volume.manager import validate_account_group
+
+    try:
+        if account_rule:
+            group = validate_account_group(group)
+        settings.storage_group(group)
+    except DevbaseError as e:
+        raise GroupOptionError(f"--group に使えない名前です: {e}") from None
+    return group
+
+
 def _target_group(devbase_root: Path, store, group: Optional[str]) -> Optional[str]:
     """コマンドが相手にするグループ (PLAN56「対象のグループ」・#315)。
 
@@ -97,13 +117,7 @@ def _target_group(devbase_root: Path, store, group: Optional[str]) -> Optional[s
         raise GroupOptionError(
             "--group はグループ別の置き場 (backend: openbao、version: 2) を選んだ設定で"
             "だけ使えます")
-    from devbase.env.backend_config import BackendConfigError
-
-    try:
-        store.config.openbao.storage_group(group)
-    except BackendConfigError as e:
-        raise GroupOptionError(f"--group に使えない名前です: {e}") from None
-    return group
+    return check_group_option(store.config.openbao, group)
 
 
 def _project_group_mismatch(devbase_root: Path, store, group: Optional[str],
@@ -750,7 +764,7 @@ def cmd_env_sync(devbase_root: Path, user: bool = False, group: Optional[str] =
 def _sync_credential_sources(ctx: _SyncContext) -> None:
     """AWS と Git の認証情報をソースから同期する (``ctx.counts`` に集計する)"""
     # AWS: 取り込みの選択の範囲だけで入れ直す (#314 I7)
-    plan = _aws_sync_plan(ctx.sources, ctx.targets, ctx.store)
+    plan = _aws_sync_plan(ctx)
     if plan is None:
         # 書き込みを見送ったら控えの登録も見送る (丸ごとで登録すると次の変更で選択外まで送るため)
         ctx.counts.aws_selection = _SKIP_SELECTION
@@ -835,7 +849,24 @@ def _same_aws_payload(stored: Optional[str], encoded: str) -> bool:
     return current is not None and current == _aws_payload(stored)
 
 
-def _aws_sync_plan(sources, targets, store):
+_AWS_LABEL = 'AWS認証'
+
+
+def _report_unknown_profiles(payload) -> None:
+    """選んだのに ``~/.aws/config`` に無いプロファイルを 1 行ずつ知らせる"""
+    for name in payload.unknown:
+        logger.info("%s: 選んだプロファイル %s が ~/.aws/config にありません", _AWS_LABEL, name)
+
+
+def _report_profile_conflicts(payload) -> bool:
+    """空白だけが違う見出しの節を警告する。あれば真 (書かない)"""
+    for name in payload.conflicts:
+        logger.warning("%s: [%s] と空白だけが違う見出しの節が ~/.aws にあるため書きません",
+                       _AWS_LABEL, name)
+    return bool(payload.conflicts)
+
+
+def _aws_sync_plan(ctx: _SyncContext):
     """AWS の入れ直しの作り方と取り込みの選択 ``(encode_fn, selection)``。書かないなら ``None``。
 
     - 控えの項目が ``aws_profiles`` なら、控えの ``profiles`` とその連なりだけを切り出す (I7)
@@ -844,49 +875,54 @@ def _aws_sync_plan(sources, targets, store):
       プロファイルの集合が値のプロファイルの集合と一致すれば丸ごと。値が読めない、または値にあって
       今のファイルに無いプロファイルがあれば書かない (決定 10)
     """
-    from devbase.env import aws_profiles
     from devbase.env.collectors.aws import _encode_aws_config_files
     from devbase.env.host_import import ALL
 
-    key = keys.AWS_CONFIG_BASE64
-    label = 'AWS認証'
+    source = ctx.sources.get_source('aws')
+    if source and source.get('type') == 'aws_profiles':
+        return _aws_plan_for_selection(ctx, list(source.get('profiles') or []))
+    if source:
+        return _encode_aws_config_files, ALL
+    return _aws_plan_for_unregistered(ctx)
 
-    def report_unknown(payload) -> None:
-        for name in payload.unknown:
-            logger.info("%s: 選んだプロファイル %s が ~/.aws/config にありません", label, name)
 
-    def report_conflicts(payload) -> bool:
-        for name in payload.conflicts:
-            logger.warning("%s: [%s] と空白だけが違う見出しの節が ~/.aws にあるため書きません",
-                           label, name)
-        return bool(payload.conflicts)
+def _aws_plan_for_selection(ctx: _SyncContext, profiles):
+    """控えが ``aws_profiles`` のとき、選んだプロファイルとその連なりだけを切り出す計画 (I7)"""
+    from devbase.env import aws_profiles
 
-    source = sources.get_source('aws')
-    if source and source.get('type') == 'aws_profiles':
-        profiles = list(source.get('profiles') or [])
-        if targets.holder(key) is None:
-            return (lambda: aws_profiles.build_from_home(profiles).encode()), profiles
-        payload = aws_profiles.build_from_home(profiles)
-        report_unknown(payload)
-        if report_conflicts(payload):
-            return None
-        # 選んだプロファイルが消えていれば控えと比べられず書かない (_sync_source が知らせる)
-        if not payload.unknown and not _aws_selection_covers(targets, key, payload):
-            dest = targets.target_for(key)
-            logger.info("%s: %sの値に選択外のプロファイルがあるため書きません"
-                        "（選択はグループの控えのもの。入れ直すなら devbase env init --reset）",
-                        label, store.display_label(dest.ref))
-            return None
+    key = keys.AWS_CONFIG_BASE64
+    targets = ctx.targets
+    if targets.holder(key) is None:
         return (lambda: aws_profiles.build_from_home(profiles).encode()), profiles
-    if source:
-        return _encode_aws_config_files, ALL
+    payload = aws_profiles.build_from_home(profiles)
+    _report_unknown_profiles(payload)
+    if _report_profile_conflicts(payload):
+        return None
+    # 選んだプロファイルが消えていれば控えと比べられず書かない (_sync_source が知らせる)
+    if not payload.unknown and not _aws_selection_covers(targets, key, payload):
+        dest = targets.target_for(key)
+        logger.info("%s: %sの値に選択外のプロファイルがあるため書きません"
+                    "（選択はグループの控えのもの。入れ直すなら devbase env init --reset）",
+                    _AWS_LABEL, ctx.store.display_label(dest.ref))
+        return None
+    return (lambda: aws_profiles.build_from_home(profiles).encode()), profiles
+
+
+def _aws_plan_for_unregistered(ctx: _SyncContext):
+    """控えに項目が無いとき、参照の値のプロファイルを選択とみなす計画 (決定 10)"""
+    from devbase.env import aws_profiles
+    from devbase.env.collectors.aws import _encode_aws_config_files
+    from devbase.env.host_import import ALL
+
+    key = keys.AWS_CONFIG_BASE64
+    targets = ctx.targets
     holder = targets.holder(key)
     if holder is None:
         return _encode_aws_config_files, _KEEP_SELECTION
     in_value = aws_profiles.profiles_in_value(targets.get(key))
     if in_value is None:
         logger.info("%s: ソース未登録（%sにキーがあります）。値を読めないため書きません",
-                    label, store.display_label(holder.ref))
+                    _AWS_LABEL, ctx.store.display_label(holder.ref))
         return None
     config_text, credentials_text = aws_profiles.read_home()
     # 丸ごとの判定は credentials にだけあるプロファイルも含める (値の側の名前と同じ数え方)。
@@ -898,9 +934,9 @@ def report_conflicts(payload) -> bool:
         return _encode_aws_config_files, ALL
     payload = aws_profiles.build(config_text, credentials_text, in_value)
     if payload.unknown:
-        report_unknown(payload)
+        _report_unknown_profiles(payload)
         return None
-    if report_conflicts(payload):
+    if _report_profile_conflicts(payload):
         return None
     # 控えに残すのは選んだプロファイルだけで、連なりで入った節は含めない (sync の時点で求め直す)
     return payload.encode, aws_profiles.chosen_in_value(targets.get(key)) or in_value
@@ -1072,31 +1108,43 @@ def cmd_env_list(devbase_root: Path, global_only: bool = False,
         return e.exit_code
 
     if not project_only:
-        for as_user in owners:
-            env_file = _global_env(devbase_root, user=as_user, store=store, group=target)
-            if as_user and not env_file.file_exists():
-                continue
-            all_vars = env_file.get_all()
-            label = store.display_label(env_file.ref)
+        _list_global_sections(devbase_root, store, owners, target, keys_only, reveal)
+    if include_project:
+        _list_project_sections(devbase_root, store, owners, target, keys_only, reveal)
+    return 0
 
-            print(f"\n=== {label} ({env_file.path}{_mode_suffix(env_file)}) ===")
-            _print_env_vars(all_vars, keys_only, reveal)
-            print(f"\n{label}: {len(all_vars)}変数")
 
-    if include_project:
-        for as_user in owners:
-            proj_env = _project_env(devbase_root, user=as_user, store=store, group=target)
-            if proj_env is not None and proj_env.file_exists():
-                proj_vars = proj_env.get_all()
-                label = '個人のプロジェクト' if as_user else 'プロジェクト'
-                suffix = _group_suffix(store, proj_env.ref)
-
-                print(f"\n=== {label}: {proj_env.ref.name}{suffix} "
-                      f"({proj_env.path}{_mode_suffix(proj_env)}) ===")
-                _print_env_vars(proj_vars, keys_only, reveal)
-                print(f"\n{label}{suffix}: {len(proj_vars)}変数")
+def _print_env_section(title: str, count_label: str, env_vars, keys_only, reveal) -> None:
+    """``env list`` の節の 1 つ分 (見出し → 変数 → 件数の行)"""
+    print(f"\n=== {title} ===")
+    _print_env_vars(env_vars, keys_only, reveal)
+    print(f"\n{count_label}: {len(env_vars)}変数")
 
-    return 0
+
+def _list_global_sections(devbase_root: Path, store, owners, target, keys_only, reveal) -> None:
+    """共通の節を持ち主ごとに出す。チーム共通は 0 件でも出し、個人共通は在るときだけ"""
+    for as_user in owners:
+        env_file = _global_env(devbase_root, user=as_user, store=store, group=target)
+        if as_user and not env_file.file_exists():
+            continue
+        all_vars = env_file.get_all()
+        label = store.display_label(env_file.ref)
+        _print_env_section(f"{label} ({env_file.path}{_mode_suffix(env_file)})", label,
+                           all_vars, keys_only, reveal)
+
+
+def _list_project_sections(devbase_root: Path, store, owners, target, keys_only, reveal) -> None:
+    """プロジェクトの節を持ち主ごとに、参照が在るときだけ出す"""
+    for as_user in owners:
+        proj_env = _project_env(devbase_root, user=as_user, store=store, group=target)
+        if proj_env is None or not proj_env.file_exists():
+            continue
+        proj_vars = proj_env.get_all()
+        label = '個人のプロジェクト' if as_user else 'プロジェクト'
+        suffix = _group_suffix(store, proj_env.ref)
+        _print_env_section(
+            f"{label}: {proj_env.ref.name}{suffix} ({proj_env.path}{_mode_suffix(proj_env)})",
+            f"{label}{suffix}", proj_vars, keys_only, reveal)
 
 
 def _group_suffix(store, ref) -> str:
@@ -1368,13 +1416,37 @@ def _edit_via_tempfile(env_file, editor: str) -> int:
         shutil.rmtree(workdir, ignore_errors=True)
 
 
+def _generated_secret(spec: str) -> str:
+    """``generate`` の指定 (``token`` か ``token:<長さ>``) から自動生成の値を作る (既定 64 文字)"""
+    import secrets
+    length = 64
+    if ':' in spec:
+        _, length_str = spec.split(':', 1)
+        length = int(length_str)
+    return secrets.token_hex(length // 2)
+
+
+def _prompt_env_var(env_file, var: dict) -> bool:
+    """1 つの変数の値を入力で求めて書く。必須なのに空なら誤りを出して False を返す。"""
+    name = var.get('name', '')
+    prompt = var.get('prompt', name)
+    default = var.get('default', '')
+    required = var.get('required', False)
+    suffix = f" (デフォルト: {default})" if default else ""
+    suffix += " (必須)" if required else " (空でスキップ)"
+    value = safe_input(f"{prompt}{suffix}: ", default)
+    if value:
+        env_file.set(name, value)
+    elif required:
+        logger.error("必須変数 '%s' が設定されていません", name)
+        return False
+    return True
+
+
 def _collect_from_env_yml(env_file, variables: list) -> bool:
     """env.yml の変数定義に従って設定値を収集する。必須値未入力なら False を返す。"""
     for var in variables:
         name = var.get('name', '')
-        prompt = var.get('prompt', name)
-        default = var.get('default', '')
-        required = var.get('required', False)
         generate = var.get('generate', '')
 
         existing = env_file.get(name)
@@ -1383,23 +1455,10 @@ def _collect_from_env_yml(env_file, variables: list) -> bool:
             continue
 
         if generate:
-            import secrets
-            length = 64
-            if ':' in generate:
-                _, length_str = generate.split(':', 1)
-                length = int(length_str)
-            value = secrets.token_hex(length // 2)
-            env_file.set(name, value)
+            env_file.set(name, _generated_secret(generate))
             print(f"{name}: (自動生成)")
-        else:
-            suffix = f" (デフォルト: {default})" if default else ""
-            suffix += " (必須)" if required else " (空でスキップ)"
-            value = safe_input(f"{prompt}{suffix}: ", default)
-            if value:
-                env_file.set(name, value)
-            elif required:
-                logger.error("必須変数 '%s' が設定されていません", name)
-                return False
+        elif not _prompt_env_var(env_file, var):
+            return False
     return True
 
 
@@ -1541,41 +1600,23 @@ def _print_key_backup_notice(path, public: str) -> None:
     print("=" * 60)
 
 
-def cmd_env_keygen(devbase_root: Path, force: bool = False,
-                   assume_yes: bool = False) -> int:
-    """devbase 専用の age 鍵を生成する
-
-    生成先は必ず ``agekeys.key_file_path()`` (= ``DEVBASE_AGE_KEY_FILE`` があれば
-    それ、無ければ ``~/.config/devbase/age/keys.txt``) にする。生成先を CLI 引数で
-    自由に選べるようにすると、復号側の ``agekeys.resolve_identities()`` はそのパスを
-    探索しないため「生成した鍵で保存した機密を復号できない」状態を作れてしまう。
-    場所を変えたい場合は ``DEVBASE_AGE_KEY_FILE`` を設定してから実行してもらい、
-    生成先と探索先が構造的に一致する契約を保つ。
-    """
+def _show_existing_key(path: Path) -> int:
+    """既に在る鍵の公開鍵と作り直し方を出す (``--force`` が無いとき)。読めなければ 1"""
     from devbase.env import agekeys
-    from devbase.errors import DevbaseError
 
-    path = agekeys.key_file_path()
+    try:
+        public = agekeys.read_public_key(path)
+    except DevbaseError as e:
+        logger.error("%s", e)
+        return 1
+    print(f"鍵は既に存在します: {path}")
+    print(f"  公開鍵: {public}")
+    print("  作り直す場合: devbase env keygen --force")
+    return 0
 
-    if path.exists() and not force:
-        try:
-            public = agekeys.read_public_key(path)
-        except DevbaseError as e:
-            logger.error("%s", e)
-            return 1
-        print(f"鍵は既に存在します: {path}")
-        print(f"  公開鍵: {public}")
-        print("  作り直す場合: devbase env keygen --force")
-        return 0
 
-    # keygen はワークスペース固有の受信者リスト (secrets/recipients.txt) を触らない。
-    # 鍵はグローバル (~/.config/devbase/age/keys.txt) なのに受信者リストは
-    # ワークスペースごとに存在するため、ここで書き込むと別ワークスペースには旧公開鍵が
-    # 取り残され、既に失われた秘密鍵に対応する公開鍵で暗号化してしまう。
-    # agekeys.resolve_recipients() は recipients.txt が無ければ鍵ファイルの公開鍵へ
-    # フォールバックするので、単独利用ではリストを作る必要がない。チーム運用で明示的に
-    # 受信者を足す経路 (rekey) だけが recipients.txt を作る。
-    #
+def _refuse_unreadable_key(path: Path) -> bool:
+    """既存の鍵が在るのに読めなければ、上書きを中止する旨を出して真を返す。"""
     # 書き込みの原子性は agekeys.generate_key_file →
     # io_common.write_secure_bytes_atomic (一時ファイル + fsync + os.replace) が
     # 担保しており、生成が途中で失敗しても既存の鍵ファイルは元のまま残る。
@@ -1594,29 +1635,65 @@ def cmd_env_keygen(devbase_root: Path, force: bool = False,
         logger.error(
             "権限を確認するか、不要と判断できる場合は手動で退避してから"
             "再実行してください")
-        return 1
+        return True
+    return False
 
-    # ここへ来るのは「鍵が無い」か「--force で作り直す」場合だけ。後者は既存鍵を
-    # 捨てる操作なので、常に明示的な同意を取る。
+
+def _confirm_key_regeneration(devbase_root: Path, path: Path) -> bool:
+    """既存の鍵を作り直す同意を求める。``yes`` と入力されなければ中止を出して偽を返す。"""
+    # ここへ来るのは「--force で作り直す」場合だけ。既存鍵を捨てる操作なので、常に
+    # 明示的な同意を取る。
     #
     # 鍵は ~/.config/devbase/age/keys.txt = 全ワークスペース共通のグローバル資産
     # なのに対し、暗号化された機密はワークスペースごとに散らばっている。同意の要否を
     # カレントの DEVBASE_ROOT に機密があるか (_has_encrypted_secrets) で決めると、
     # まだ機密の無い別プロジェクトで --force した瞬間に無警告で鍵が消え、他プロジェクトの
     # 機密が復旧不能になる。カレントの状況は「文言をどれだけ強くするか」にだけ使う。
-    if path.exists() and not assume_yes:
-        print("鍵ファイルを作り直します。この鍵は全プロジェクト共通です。")
-        print(f"  鍵ファイル: {path}")
-        if _has_encrypted_secrets(devbase_root):
-            print("  このワークスペースには暗号化済みの機密があり、"
-                  "旧鍵でしか復号できないものは失われます。")
-        print("  他のワークスペースで暗号化した機密も、"
-              "旧鍵を失うと復号できなくなります。")
-        print("  続行前に旧鍵のバックアップがあるか確認してください。")
-        answer = safe_input("続行しますか? (yes と入力): ")
-        if answer != 'yes':
-            print("中止しました")
-            return 1
+    print("鍵ファイルを作り直します。この鍵は全プロジェクト共通です。")
+    print(f"  鍵ファイル: {path}")
+    if _has_encrypted_secrets(devbase_root):
+        print("  このワークスペースには暗号化済みの機密があり、"
+              "旧鍵でしか復号できないものは失われます。")
+    print("  他のワークスペースで暗号化した機密も、"
+          "旧鍵を失うと復号できなくなります。")
+    print("  続行前に旧鍵のバックアップがあるか確認してください。")
+    answer = safe_input("続行しますか? (yes と入力): ")
+    if answer != 'yes':
+        print("中止しました")
+        return False
+    return True
+
+
+def cmd_env_keygen(devbase_root: Path, force: bool = False,
+                   assume_yes: bool = False) -> int:
+    """devbase 専用の age 鍵を生成する
+
+    生成先は必ず ``agekeys.key_file_path()`` (= ``DEVBASE_AGE_KEY_FILE`` があれば
+    それ、無ければ ``~/.config/devbase/age/keys.txt``) にする。生成先を CLI 引数で
+    自由に選べるようにすると、復号側の ``agekeys.resolve_identities()`` はそのパスを
+    探索しないため「生成した鍵で保存した機密を復号できない」状態を作れてしまう。
+    場所を変えたい場合は ``DEVBASE_AGE_KEY_FILE`` を設定してから実行してもらい、
+    生成先と探索先が構造的に一致する契約を保つ。
+    """
+    from devbase.env import agekeys
+    from devbase.errors import DevbaseError
+
+    path = agekeys.key_file_path()
+
+    if path.exists() and not force:
+        return _show_existing_key(path)
+
+    # keygen はワークスペース固有の受信者リスト (secrets/recipients.txt) を触らない。
+    # 鍵はグローバル (~/.config/devbase/age/keys.txt) なのに受信者リストは
+    # ワークスペースごとに存在するため、ここで書き込むと別ワークスペースには旧公開鍵が
+    # 取り残され、既に失われた秘密鍵に対応する公開鍵で暗号化してしまう。
+    # agekeys.resolve_recipients() は recipients.txt が無ければ鍵ファイルの公開鍵へ
+    # フォールバックするので、単独利用ではリストを作る必要がない。チーム運用で明示的に
+    # 受信者を足す経路 (rekey) だけが recipients.txt を作る。
+    if _refuse_unreadable_key(path):
+        return 1
+    if path.exists() and not assume_yes and not _confirm_key_regeneration(devbase_root, path):
+        return 1
 
     # force はコマンドの --force をそのまま渡す。ここで無条件に force=True に
     # すると、上の path.exists() 判定から実際の書き込みまでの隙間に他プロセスが
diff --git a/lib/devbase/snapshot/manager.py b/lib/devbase/snapshot/manager.py
index 380b4fe6..ce550937 100644
--- a/lib/devbase/snapshot/manager.py
+++ b/lib/devbase/snapshot/manager.py
@@ -413,15 +413,7 @@ def restore(self, name: str, point: int | None = None) -> None:
             point: 差分の適用上限（例: 3なら incr-003 まで適用）。
                    Noneなら全差分を適用。
         """
-        if point is not None and point <= 0:
-            raise SnapshotError(f"--point は正の整数である必要があります: {point}")
-        snap_dir = self._safe_snap_dir(name)
-        if not snap_dir.exists():
-            raise SnapshotError(f"スナップショット '{name}' が見つかりません")
-
-        full_archive = snap_dir / FULL_ARCHIVE
-        if not full_archive.exists():
-            raise SnapshotError(f"フルバックアップが見つかりません: {full_archive}")
+        snap_dir = self._restore_source(name, point)
 
         # 控える組を決めるため、組は控えより先に読む。検証を通らない世代では
         # 控えも復元も行わない。
@@ -443,8 +435,31 @@ def restore(self, name: str, point: int | None = None) -> None:
             self.clear_command(volumes) + self.restore_command(FULL_ARCHIVE),
             volumes, pre_restore_name, skipped_renames,
         )
+        self._apply_incrementals(snap_dir, point, volumes, pre_restore_name, skipped_renames)
+
+        self._warn_about_lost_renames(snap_dir, volumes, skipped_renames)
+
+        if point is not None:
+            logger.info("復元完了: %s (incr-%03d まで)", name, point)
+        else:
+            logger.info("復元完了: %s", name)
+
+    def _restore_source(self, name: str, point: int | None) -> Path:
+        """復元の前の検査 (``point``・世代・full)。通れば世代のディレクトリを返す。"""
+        if point is not None and point <= 0:
+            raise SnapshotError(f"--point は正の整数である必要があります: {point}")
+        snap_dir = self._safe_snap_dir(name)
+        if not snap_dir.exists():
+            raise SnapshotError(f"スナップショット '{name}' が見つかりません")
 
-        # 差分バックアップを順番に適用（pointが指定されていればそこまで）
+        full_archive = snap_dir / FULL_ARCHIVE
+        if not full_archive.exists():
+            raise SnapshotError(f"フルバックアップが見つかりません: {full_archive}")
+        return snap_dir
+
+    def _apply_incrementals(self, snap_dir: Path, point: int | None, volumes: dict,
+                            pre_restore_name: Optional[str], skipped_renames: list) -> None:
+        """差分バックアップを順番に適用する（``point`` が指定されていればそこまで）"""
         incr_files = sorted(snap_dir.glob(INCR_ARCHIVE_GLOB))
         for incr in incr_files:
             if point is not None:
@@ -459,13 +474,6 @@ def restore(self, name: str, point: int | None = None) -> None:
                 volumes, pre_restore_name, skipped_renames,
             )
 
-        self._warn_about_lost_renames(snap_dir, volumes, skipped_renames)
-
-        if point is not None:
-            logger.info("復元完了: %s (incr-%03d まで)", name, point)
-        else:
-            logger.info("復元完了: %s", name)
-
     def _backup_before_restore(self, volumes: dict) -> Optional[str]:
         """復元前バックアップ ``pre-restore-<時刻>`` を、渡された組で作る。
 
diff --git a/lib/devbase/tui/actions_env_keys.py b/lib/devbase/tui/actions_env_keys.py
index 4f742cea..e26693cd 100644
--- a/lib/devbase/tui/actions_env_keys.py
+++ b/lib/devbase/tui/actions_env_keys.py
@@ -189,8 +189,8 @@ def group_attrs(devbase_root: Path) -> dict:
     置き場のグループ名の検査 (``storage_group``: 予約語 ``global`` / ``projects`` と
     ``group_aliases`` の読み替え先) も通し、通らない名前は委譲先へ渡さない。
     """
+    from devbase.commands.env import GroupOptionError, check_group_option
     from devbase.env.secret_store import SecretStore
-    from devbase.volume.manager import validate_account_group
 
     try:
         grouped = _grouped(devbase_root)
@@ -202,11 +202,9 @@ def group_attrs(devbase_root: Path) -> dict:
     while True:
         name = _select_group(devbase_root)
         try:
-            name = validate_account_group(name)
-            settings.storage_group(name)
-            return {"group": name}
-        except DevbaseError as e:
-            logger.error("--group に使えない名前です: %s", e)
+            return {"group": check_group_option(settings, name, account_rule=True)}
+        except GroupOptionError as e:
+            logger.error("%s", e)
 
 
 def project_title(name: str, count, width: int) -> str:
diff --git a/lib/devbase/volume/compose.py b/lib/devbase/volume/compose.py
index 9ddb1fbe..c16dcdee 100644
--- a/lib/devbase/volume/compose.py
+++ b/lib/devbase/volume/compose.py
@@ -226,6 +226,58 @@ def _iter_env_names(existing: Any) -> Iterator[str]:
                 yield name
 
 
+def _mask_env_none(existing: Any, secrets: List[str], secret_set: Set[str]) -> Optional[list]:
+    """元から environment が無ければ、機密が無い限り作らない (``None`` = 作らない)。"""
+    return list(secrets) if secrets else None
+
+
+def _mask_env_dict(existing: dict, secrets: List[str], secret_set: Set[str]) -> dict:
+    """map 形式: 機密キーの値を ``None`` にし、無い機密キーを足す。"""
+    masked = {
+        key: (None if key in secret_set else value)
+        for key, value in existing.items()
+    }
+    for name in secrets:
+        masked.setdefault(name, None)
+    return masked
+
+
+def _mask_env_list(existing: list, secrets: List[str], secret_set: Set[str]) -> list:
+    """list 形式: 機密キーを裸のキー名にし、無い機密キーを末尾へ足す。"""
+    masked_list = []
+    listed = set()
+    for item in existing:
+        name = _env_item_name(item)
+        if name is None:
+            masked_list.append(item)
+            continue
+        listed.add(name)
+        # 機密キーは `KEY=value` でも `KEY` でも、値なし参照に揃える
+        masked_list.append(name if name in secret_set else item)
+    masked_list.extend(name for name in secrets if name not in listed)
+    return masked_list
+
+
+def _mask_env_other(existing: Any, secrets: List[str], secret_set: Set[str]) -> list:
+    """map / list 以外は Compose が受け付けない書き方。
+
+    手掛かりを残しつつ、機密が渡らない事故を避けるため名前の列挙で置き換える。
+    """
+    logger.warning(
+        "environment の形式 (%s) を解釈できないため、機密の変数名の列挙で"
+        "置き換えます", type(existing).__name__)
+    return list(secrets)
+
+
+#: ``_env_shape`` の形 → 伏せた environment を返す関数
+_ENV_MASKERS = {
+    'none': _mask_env_none,
+    'dict': _mask_env_dict,
+    'list': _mask_env_list,
+    'other': _mask_env_other,
+}
+
+
 def _mask_secret_environment(
     service: dict, secret_env_names: Sequence[str],
 ) -> None:
@@ -239,51 +291,14 @@ def _mask_secret_environment(
 
     元の記法は尊重する。map 形式なら値を ``None`` にした map (Compose は ``KEY:``
     を「実行プロセスの環境変数から解決」と解釈する)、list 形式なら裸のキー名を
-    並べた list として出力する。
+    並べた list として出力する。形ごとの処理は ``_ENV_MASKERS`` にある。
     """
     # 重複を除きつつ、指定された順序は保つ
     secrets = list(dict.fromkeys(secret_env_names))
-    secret_set = set(secrets)
     existing = service.get('environment')
-    shape = _env_shape(existing)
-
-    if shape == 'none':
-        # 元から environment が無ければ、機密が無い限り作らない
-        if secrets:
-            service['environment'] = list(secrets)
-        return
-
-    if shape == 'dict':
-        masked = {
-            key: (None if key in secret_set else value)
-            for key, value in existing.items()
-        }
-        for name in secrets:
-            masked.setdefault(name, None)
+    masked = _ENV_MASKERS[_env_shape(existing)](existing, secrets, set(secrets))
+    if masked is not None:
         service['environment'] = masked
-        return
-
-    if shape == 'list':
-        masked_list = []
-        listed = set()
-        for item in existing:
-            name = _env_item_name(item)
-            if name is None:
-                masked_list.append(item)
-                continue
-            listed.add(name)
-            # 機密キーは `KEY=value` でも `KEY` でも、値なし参照に揃える
-            masked_list.append(name if name in secret_set else item)
-        masked_list.extend(name for name in secrets if name not in listed)
-        service['environment'] = masked_list
-        return
-
-    # map / list 以外は Compose が受け付けない書き方。手掛かりを残しつつ、
-    # 機密が渡らない事故を避けるため名前の列挙で置き換える。
-    logger.warning(
-        "environment の形式 (%s) を解釈できないため、機密の変数名の列挙で"
-        "置き換えます", type(existing).__name__)
-    service['environment'] = list(secrets)
 
 
 def _service_env_names(service: dict) -> List[str]:
diff --git a/tests/cli/tui/test_actions_env_keys.py b/tests/cli/tui/test_actions_env_keys.py
index f3820f60..e6688178 100644
--- a/tests/cli/tui/test_actions_env_keys.py
+++ b/tests/cli/tui/test_actions_env_keys.py
@@ -466,3 +466,23 @@ def test_an_age_backend_writes_the_encrypted_global(file_root, monkeypatch):
     assert (file_root / 'secrets' / 'global.env.age').is_file()
     assert not (file_root / '.env').exists()
     assert SecretStore(file_root).load(SecretRef.for_global()) == {'A': '1', 'K': 'v'}
+
+
+_TYPED_GROUP_ERRORS = {
+    '1234': ("--group に使えない名前です: DEVBASE_ACCOUNT_GROUP に数字だけの名前は使えません: '1234'。"
+             "インスタンス番号のボリューム devbase_home_<index> と同じ名前になります"),
+    'projects': ("--group に使えない名前です: グループ名 'projects' は置き場のグループ名に使えません"
+                 " (global / projects は version: 1 のパスと重なります)"),
+}
+
+
+@pytest.mark.parametrize('name', sorted(_TYPED_GROUP_ERRORS))
+def test_an_unusable_typed_group_prints_the_whole_error_line(grouped, monkeypatch, caplog, name):
+    """#444 R2: TUI で入れた名前の誤りの行を丸ごと固定する (CLI と違い「グループ名: 」が付かない)"""
+    Script(monkeypatch, select=[keys_ui.TYPE_GROUP, menu.MENU_BACK], text=[name])
+
+    with pytest.raises(flow.BackOut):
+        keys_ui.group_attrs(grouped)
+
+    errors = [r.getMessage() for r in caplog.records if r.levelname == 'ERROR']
+    assert errors == [_TYPED_GROUP_ERRORS[name]]
diff --git a/tests/commands/test_env_sync_group.py b/tests/commands/test_env_sync_group.py
index 2fa5c1aa..98f0a410 100644
--- a/tests/commands/test_env_sync_group.py
+++ b/tests/commands/test_env_sync_group.py
@@ -123,3 +123,20 @@ def test_project_writes_the_projects_group(grouped, openbao, monkeypatch):
 
     assert set(openbao.get('team/initech/projects/web')) == {'GENERATED'}
     assert kv_paths(openbao) == {'team/initech/projects/web'}
+
+
+_GROUP_OPTION_ERRORS = {
+    'global': ("--group に使えない名前です: グループ名 'global' は置き場のグループ名に使えません"
+               " (global / projects は version: 1 のパスと重なります)"),
+    '1234': ("--group に使えない名前です: グループ名: DEVBASE_ACCOUNT_GROUP に数字だけの名前は"
+             "使えません: '1234'。インスタンス番号のボリューム devbase_home_<index> と同じ名前に"
+             "なります"),
+}
+
+
+@pytest.mark.parametrize('name', sorted(_GROUP_OPTION_ERRORS))
+def test_an_unusable_group_option_prints_the_whole_message(grouped, openbao, caplog, name):
+    """#444 R2: --group の誤りの文を丸ごと固定する (予約語と数字だけ)"""
+    assert env_cmd.cmd_env_sync(grouped, group=name) == 2
+    errors = [r.getMessage() for r in caplog.records if r.levelname == 'ERROR']
+    assert errors == [_GROUP_OPTION_ERRORS[name]]
diff --git a/tests/commands/test_env_user_axis.py b/tests/commands/test_env_user_axis.py
index 963277d8..4f08ec0b 100644
--- a/tests/commands/test_env_user_axis.py
+++ b/tests/commands/test_env_user_axis.py
@@ -690,3 +690,36 @@ def test_file_backend_outside_projects_is_unchanged(file_root, monkeypatch, caps
 
     assert run_env(file_root, 'get', 'KEY') == 0
     assert capsys.readouterr().out == 'x\n'
+
+
+def _list_lines(keys_only: bool, key: str, value: str) -> str:
+    return f"  {key}" if keys_only else f"  {key:<35} {value}"
+
+
+@pytest.mark.parametrize('keys_only', [False, True])
+def test_list_prints_the_whole_sections(in_web, openbao, capsys, keys_only):
+    """#444 R8・R9: 節の見出し・変数・件数の行を丸ごと固定する (変数の無い節を含む)"""
+    openbao.put(USER_GLOBAL_PATH, {'B': '2'})
+    openbao.put(TEAM_WEB, {'C': '3', 'A': 'x'})
+
+    assert env_cmd.cmd_env_list(in_web, keys_only=keys_only) == 0
+
+    assert capsys.readouterr().out == '\n'.join([
+        '',
+        '=== グローバル (devbase/team/global) ===',
+        '  (変数なし)',
+        '',
+        'グローバル: 0変数',
+        '',
+        '=== 個人のグローバル (devbase/users/member01/global [openbao]) ===',
+        _list_lines(keys_only, 'B', '2'),
+        '',
+        '個人のグローバル: 1変数',
+        '',
+        '=== プロジェクト: web (devbase/team/projects/web [openbao]) ===',
+        _list_lines(keys_only, 'A', 'x'),
+        _list_lines(keys_only, 'C', '3'),
+        '',
+        'プロジェクト: 2変数',
+        '',
+    ])
diff --git a/tests/conftest.py b/tests/conftest.py
index 9c4f2522..e9e247a5 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -221,50 +221,79 @@ def _mount_ok(self, rec: Received) -> bool:
 
     def do_POST(self):
         rec = self._record('POST')
-        state = self.server_state
         if rec.path == '/v1/auth/approle/login':
-            state.logins += 1
-            if state.disable_entity:
-                return self._send(403, {'errors': ['permission denied']})
-            if (state.reject_login or rec.body.get('role_id') != state.role_id
-                    or rec.body.get('secret_id') != state.secret_id):
-                return self._send(400, {'errors': ['invalid role or secret']})
-            state.token = f'token-{state.logins}'
-            return self._send(200, {'auth': {'client_token': state.token,
-                                             'lease_duration': 3600,
-                                             'renewable': True}})
+            return self._login(rec)
+        return self._write_secret(rec)
+
+    def _login(self, rec: Received) -> None:
+        """AppRole のログイン。通れば新しい token を出す"""
+        state = self.server_state
+        state.logins += 1
+        if state.disable_entity:
+            return self._send(403, {'errors': ['permission denied']})
+        if (state.reject_login or rec.body.get('role_id') != state.role_id
+                or rec.body.get('secret_id') != state.secret_id):
+            return self._send(400, {'errors': ['invalid role or secret']})
+        state.token = f'token-{state.logins}'
+        return self._send(200, {'auth': {'client_token': state.token,
+                                         'lease_duration': 3600,
+                                         'renewable': True}})
+
+    def _write_secret(self, rec: Received) -> None:
+        """KV v2 の書き込み (拒否 → 失敗の注入 → ``data`` の検査 → CAS の更新 → 応答の注入)"""
+        state = self.server_state
         path = rec.kv_path
+        if self._refuse_write(rec, path):
+            return None
+        data = rec.body.get('data')
+        if not isinstance(data, dict):
+            return self._send(400, {'errors': ['data must be a map']})
+        new_version = self._apply_cas(rec, path, data)
+        if new_version is None:
+            return None
+        if state.drop_write_response:
+            return self._drop()
+        if state.garble_write_response:
+            return self._send(200, raw=b'<html>gateway</html>')
+        return self._send(200, {'data': {'version': new_version, 'destroyed': False,
+                                         'created_time': '2026-09-14T00:00:00Z'}})
+
+    def _refuse_write(self, rec: Received, path: Optional[str]) -> bool:
+        """書き込みの前の拒否 (mount・認可・禁止) と失敗の注入。応答を返したら真"""
+        state = self.server_state
         if path is None or not self._mount_ok(rec):
-            return self._send(404, {'errors': []})
+            self._send(404, {'errors': []})
+            return True
         if not self._authorized(rec):
-            return self._send(403, {'errors': ['permission denied']})
+            self._send(403, {'errors': ['permission denied']})
+            return True
         if self._forbidden(path) or (not state.team_writable and path.startswith('team/')):
-            return self._send(403, {'errors': ['1 error occurred:\n\t* permission denied\n\n']})
+            self._send(403, {'errors': ['1 error occurred:\n\t* permission denied\n\n']})
+            return True
         state.write_attempts += 1
         if state.write_attempts in state.fail_write_attempts:
-            return self._send(500, {'errors': ['boom']}, truncate=state.truncate_write_error_body)
+            self._send(500, {'errors': ['boom']}, truncate=state.truncate_write_error_body)
+            return True
         if state.write_status is not None:
-            return self._send(state.write_status, {'errors': ['refused']})
-        data = rec.body.get('data')
-        if not isinstance(data, dict):
-            return self._send(400, {'errors': ['data must be a map']})
+            self._send(state.write_status, {'errors': ['refused']})
+            return True
+        return False
+
+    def _apply_cas(self, rec: Received, path: str, data: dict) -> Optional[int]:
+        """CAS が合えば版を上げて書き、新しい版を返す。合わなければ 400 を返して ``None``"""
+        state = self.server_state
         with state._lock:
             current = state.versions.get(path, 0)
             cas = rec.cas
             if cas is None or cas != current:
-                return self._send(400, {'errors': [
+                self._send(400, {'errors': [
                     'check-and-set parameter did not match the current version']})
+                return None
             state.versions[path] = current + 1
             state.secrets[path] = {str(k): v for k, v in data.items()}
             state.soft_deleted.discard(path)
             state.writes += 1
-            new_version = state.versions[path]
-        if state.drop_write_response:
-            return self._drop()
-        if state.garble_write_response:
-            return self._send(200, raw=b'<html>gateway</html>')
-        return self._send(200, {'data': {'version': new_version, 'destroyed': False,
-                                         'created_time': '2026-09-14T00:00:00Z'}})
+            return state.versions[path]
 
     def do_DELETE(self):
         rec = self._record('DELETE')
diff --git a/tests/volume/conftest.py b/tests/volume/conftest.py
index 69bce0a8..1705fa8e 100644
--- a/tests/volume/conftest.py
+++ b/tests/volume/conftest.py
@@ -3,11 +3,43 @@
 ``devbase up`` / ``scale`` は起動の先頭でプロジェクトの宣言を読み、``DEVBASE_ACCOUNT_GROUP``
 へ置いてから構成を作る (#315)。構成の生成とボリュームの作成を単独で呼ぶテストは、その後の
 状態 (グループが決まっている) を前提にする。未設定の振る舞いを見るテストは自分で消す。
+
+compose の生成を見るテストが共有する補助 (``write_compose``・``load_scaled``・``mount_source``)
+もここに置き、``from tests.volume.conftest import ...`` で取り込む。
 """
 
+from __future__ import annotations
+
 import pytest
+import yaml
 
 
 @pytest.fixture(autouse=True)
 def _declared_account_group(_isolate_env, monkeypatch):
     monkeypatch.setenv('DEVBASE_ACCOUNT_GROUP', 'acme')
+
+
+def write_compose(tmp_path, services: dict, volumes: dict | None = None) -> None:
+    """``tmp_path`` に ``compose.yml`` を書く (``volumes`` を渡したときだけその節を持つ)"""
+    document = {"services": services}
+    if volumes is not None:
+        document["volumes"] = volumes
+    (tmp_path / "compose.yml").write_text(
+        yaml.safe_dump(document, sort_keys=False), encoding="utf-8")
+
+
+def load_scaled(tmp_path) -> dict:
+    """スケールで作ったファイル (``.docker-compose.scale.yml``) を読む"""
+    return yaml.safe_load((tmp_path / ".docker-compose.scale.yml").read_text())
+
+
+def mount_source(service: dict, target: str) -> str | None:
+    """サービスの ``volumes`` で ``target`` にマウントしている元 (短い記法と長い記法)"""
+    for vol in service.get("volumes", []):
+        if isinstance(vol, str):
+            parts = vol.split(":")
+            if len(parts) >= 2 and parts[1] == target:
+                return parts[0]
+        elif isinstance(vol, dict) and vol.get("target") == target:
+            return vol.get("source")
+    return None
diff --git a/tests/volume/test_compose_group.py b/tests/volume/test_compose_group.py
index 1e91d26b..897f8aef 100644
--- a/tests/volume/test_compose_group.py
+++ b/tests/volume/test_compose_group.py
@@ -11,10 +11,10 @@
 from __future__ import annotations
 
 import pytest
-import yaml
 
 from devbase.errors import DevbaseError
 from devbase.volume import compose
+from tests.volume.conftest import load_scaled, mount_source, write_compose
 
 
 @pytest.fixture
@@ -25,29 +25,6 @@ def in_tmp_cwd(tmp_path, monkeypatch):
     return tmp_path
 
 
-def _write_compose(tmp_path, services: dict, volumes: dict | None = None) -> None:
-    document = {"services": services}
-    if volumes is not None:
-        document["volumes"] = volumes
-    (tmp_path / "compose.yml").write_text(
-        yaml.safe_dump(document, sort_keys=False), encoding="utf-8")
-
-
-def _load_scaled(tmp_path) -> dict:
-    return yaml.safe_load((tmp_path / ".docker-compose.scale.yml").read_text())
-
-
-def _mount_source(service: dict, target: str) -> str | None:
-    for vol in service.get("volumes", []):
-        if isinstance(vol, str):
-            parts = vol.split(":")
-            if len(parts) >= 2 and parts[1] == target:
-                return parts[0]
-        elif isinstance(vol, dict) and vol.get("target") == target:
-            return vol.get("source")
-    return None
-
-
 def _env_value(service: dict, name: str):
     env = service.get("environment")
     if isinstance(env, dict):
@@ -65,28 +42,28 @@ def _env_value(service: dict, name: str):
 
 def test_group_mount_is_added_when_absent(in_tmp_cwd):
     """プロジェクト compose が宣言していなくても自動で足される (前提 4)。"""
-    _write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
+    write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
 
     compose.generate_scaled_compose(scale=1)
-    dev = _load_scaled(in_tmp_cwd)["services"]["dev-1"]
+    dev = load_scaled(in_tmp_cwd)["services"]["dev-1"]
 
-    assert _mount_source(dev, "/persistent/group") == "devbase_home_acme"
+    assert mount_source(dev, "/persistent/group") == "devbase_home_acme"
 
 
 def test_group_mount_follows_account_group(in_tmp_cwd, monkeypatch):
     monkeypatch.setenv("DEVBASE_ACCOUNT_GROUP", "globex")
-    _write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
+    write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
 
     compose.generate_scaled_compose(scale=1)
-    dev = _load_scaled(in_tmp_cwd)["services"]["dev-1"]
+    dev = load_scaled(in_tmp_cwd)["services"]["dev-1"]
 
-    assert _mount_source(dev, "/persistent/group") == "devbase_home_globex"
+    assert mount_source(dev, "/persistent/group") == "devbase_home_globex"
 
 
 def test_declared_group_mount_is_rewritten(in_tmp_cwd, monkeypatch):
     """プロジェクトが別のソースで宣言していても devbase 側の名前へ差し替える。"""
     monkeypatch.setenv("DEVBASE_ACCOUNT_GROUP", "globex")
-    _write_compose(
+    write_compose(
         in_tmp_cwd,
         {"dev": {"image": "dev:latest",
                  "volumes": ["someone_elses_volume:/persistent/group:rw"]}},
@@ -94,7 +71,7 @@ def test_declared_group_mount_is_rewritten(in_tmp_cwd, monkeypatch):
     )
 
     compose.generate_scaled_compose(scale=1)
-    dev = _load_scaled(in_tmp_cwd)["services"]["dev-1"]
+    dev = load_scaled(in_tmp_cwd)["services"]["dev-1"]
 
     assert "devbase_home_globex:/persistent/group:rw" in dev["volumes"]
 
@@ -102,24 +79,24 @@ def test_declared_group_mount_is_rewritten(in_tmp_cwd, monkeypatch):
 def test_every_instance_gets_the_same_group_volume(in_tmp_cwd, monkeypatch):
     """グループはインスタンス番号に依存しない (同グループ内で共有する)。"""
     monkeypatch.setenv("DEVBASE_ACCOUNT_GROUP", "globex")
-    _write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
+    write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
 
     compose.generate_scaled_compose(scale=3)
-    services = _load_scaled(in_tmp_cwd)["services"]
+    services = load_scaled(in_tmp_cwd)["services"]
 
     for i in (1, 2, 3):
-        assert _mount_source(services[f"dev-{i}"], "/persistent/group") == "devbase_home_globex"
+        assert mount_source(services[f"dev-{i}"], "/persistent/group") == "devbase_home_globex"
 
 
 def test_shared_ai_mount_is_unchanged(in_tmp_cwd, monkeypatch):
     """共通ボリュームは分離の影響を受けない (分類 A は全グループ同一実体 / AC4)。"""
     monkeypatch.setenv("DEVBASE_ACCOUNT_GROUP", "globex")
-    _write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
+    write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
 
     compose.generate_scaled_compose(scale=1)
-    dev = _load_scaled(in_tmp_cwd)["services"]["dev-1"]
+    dev = load_scaled(in_tmp_cwd)["services"]["dev-1"]
 
-    assert _mount_source(dev, "/persistent/ai") == "devbase_home_ubuntu"
+    assert mount_source(dev, "/persistent/ai") == "devbase_home_ubuntu"
 
 
 # ---------------------------------------------------------------------------
@@ -128,10 +105,10 @@ def test_shared_ai_mount_is_unchanged(in_tmp_cwd, monkeypatch):
 
 def test_group_volume_is_declared_external(in_tmp_cwd, monkeypatch):
     monkeypatch.setenv("DEVBASE_ACCOUNT_GROUP", "globex")
-    _write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
+    write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
 
     compose.generate_scaled_compose(scale=1)
-    volumes = _load_scaled(in_tmp_cwd)["volumes"]
+    volumes = load_scaled(in_tmp_cwd)["volumes"]
 
     assert volumes["devbase_home_globex"] == {"external": True}
     assert volumes["devbase_home_ubuntu"] == {"external": True}
@@ -143,10 +120,10 @@ def test_group_volume_is_declared_external(in_tmp_cwd, monkeypatch):
 
 def test_account_group_is_exposed_to_dev_service(in_tmp_cwd, monkeypatch):
     monkeypatch.setenv("DEVBASE_ACCOUNT_GROUP", "globex")
-    _write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
+    write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
 
     compose.generate_scaled_compose(scale=1)
-    dev = _load_scaled(in_tmp_cwd)["services"]["dev-1"]
+    dev = load_scaled(in_tmp_cwd)["services"]["dev-1"]
 
     assert _env_value(dev, "DEVBASE_ACCOUNT_GROUP") == "globex"
 
@@ -156,7 +133,7 @@ def test_unset_group_stops_generation(in_tmp_cwd, monkeypatch):
     from devbase.env.groups import GroupDeclarationError
 
     monkeypatch.delenv("DEVBASE_ACCOUNT_GROUP", raising=False)
-    _write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
+    write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
 
     with pytest.raises(GroupDeclarationError):
         compose.generate_scaled_compose(scale=1)
@@ -165,10 +142,10 @@ def test_unset_group_stops_generation(in_tmp_cwd, monkeypatch):
 
 def test_group_is_exposed_to_the_container(in_tmp_cwd):
     """解決結果を明示的に渡す (コンテナ側で再解決させない)。"""
-    _write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
+    write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
 
     compose.generate_scaled_compose(scale=1)
-    dev = _load_scaled(in_tmp_cwd)["services"]["dev-1"]
+    dev = load_scaled(in_tmp_cwd)["services"]["dev-1"]
 
     assert _env_value(dev, "DEVBASE_ACCOUNT_GROUP") == "acme"
 
@@ -176,12 +153,12 @@ def test_group_is_exposed_to_the_container(in_tmp_cwd):
 def test_account_group_is_added_to_list_form_environment(in_tmp_cwd, monkeypatch):
     """既存の list 形式 environment を壊さない。"""
     monkeypatch.setenv("DEVBASE_ACCOUNT_GROUP", "globex")
-    _write_compose(in_tmp_cwd, {
+    write_compose(in_tmp_cwd, {
         "dev": {"image": "dev:latest", "environment": ["FEATURE_FLAG=enabled"]},
     })
 
     compose.generate_scaled_compose(scale=1)
-    dev = _load_scaled(in_tmp_cwd)["services"]["dev-1"]
+    dev = load_scaled(in_tmp_cwd)["services"]["dev-1"]
 
     assert _env_value(dev, "FEATURE_FLAG") == "enabled"
     assert _env_value(dev, "DEVBASE_ACCOUNT_GROUP") == "globex"
@@ -190,11 +167,11 @@ def test_account_group_is_added_to_list_form_environment(in_tmp_cwd, monkeypatch
 def test_caller_supplied_dev_environment_is_preserved(in_tmp_cwd, monkeypatch):
     """clone プラン等 (PLAN32) と共存する。"""
     monkeypatch.setenv("DEVBASE_ACCOUNT_GROUP", "globex")
-    _write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
+    write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
 
     compose.generate_scaled_compose(
         scale=1, dev_environment={"DEVBASE_PRIMARY_DIR": "app"})
-    dev = _load_scaled(in_tmp_cwd)["services"]["dev-1"]
+    dev = load_scaled(in_tmp_cwd)["services"]["dev-1"]
 
     assert _env_value(dev, "DEVBASE_PRIMARY_DIR") == "app"
     assert _env_value(dev, "DEVBASE_ACCOUNT_GROUP") == "globex"
@@ -207,7 +184,7 @@ def test_caller_supplied_dev_environment_is_preserved(in_tmp_cwd, monkeypatch):
 def test_invalid_group_stops_compose_generation(in_tmp_cwd, monkeypatch):
     """不正なグループ名は構成生成の時点で弾く (コンテナを起動させない)。"""
     monkeypatch.setenv("DEVBASE_ACCOUNT_GROUP", "ubuntu")
-    _write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
+    write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
 
     with pytest.raises(DevbaseError):
         compose.generate_scaled_compose(scale=1)
diff --git a/tests/volume/test_compose_vscode.py b/tests/volume/test_compose_vscode.py
index 45208177..1b1d63ad 100644
--- a/tests/volume/test_compose_vscode.py
+++ b/tests/volume/test_compose_vscode.py
@@ -12,9 +12,9 @@
 from __future__ import annotations
 
 import pytest
-import yaml
 
 from devbase.volume import compose
+from tests.volume.conftest import load_scaled, mount_source, write_compose
 
 VSCODE_TARGET = "/home/ubuntu/.vscode-server"
 
@@ -28,91 +28,68 @@ def in_tmp_cwd(tmp_path, monkeypatch):
     return tmp_path
 
 
-def _write_compose(tmp_path, services: dict, volumes: dict | None = None) -> None:
-    document = {"services": services}
-    if volumes is not None:
-        document["volumes"] = volumes
-    (tmp_path / "compose.yml").write_text(
-        yaml.safe_dump(document, sort_keys=False), encoding="utf-8")
-
-
-def _load_scaled(tmp_path) -> dict:
-    return yaml.safe_load((tmp_path / ".docker-compose.scale.yml").read_text())
-
-
-def _mount_source(service: dict, target: str) -> str | None:
-    for vol in service.get("volumes", []):
-        if isinstance(vol, str):
-            parts = vol.split(":")
-            if len(parts) >= 2 and parts[1] == target:
-                return parts[0]
-        elif isinstance(vol, dict) and vol.get("target") == target:
-            return vol.get("source")
-    return None
-
-
 # ---------------------------------------------------------------------------
 # マウント (AC1 / AC2 / AC3)
 # ---------------------------------------------------------------------------
 
 def test_vscode_mount_is_added_when_absent(in_tmp_cwd):
     """プロジェクト compose が書いていなくても自動で足される。"""
-    _write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
+    write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
 
     compose.generate_scaled_compose(scale=1)
 
-    dev = _load_scaled(in_tmp_cwd)["services"]["dev-1"]
-    assert _mount_source(dev, VSCODE_TARGET) == "devbase_vscode_myapp-ai_1"
+    dev = load_scaled(in_tmp_cwd)["services"]["dev-1"]
+    assert mount_source(dev, VSCODE_TARGET) == "devbase_vscode_myapp-ai_1"
 
 
 def test_each_instance_mounts_its_own_volume(in_tmp_cwd):
     """scale > 1 でインスタンスごとに別のボリュームを掴む (AC3)。"""
-    _write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
+    write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
 
     compose.generate_scaled_compose(scale=2)
 
-    services = _load_scaled(in_tmp_cwd)["services"]
-    assert _mount_source(services["dev-1"], VSCODE_TARGET) == \
+    services = load_scaled(in_tmp_cwd)["services"]
+    assert mount_source(services["dev-1"], VSCODE_TARGET) == \
         "devbase_vscode_myapp-ai_1"
-    assert _mount_source(services["dev-2"], VSCODE_TARGET) == \
+    assert mount_source(services["dev-2"], VSCODE_TARGET) == \
         "devbase_vscode_myapp-ai_2"
 
 
 def test_mount_follows_project_name(in_tmp_cwd, monkeypatch):
     """別プロジェクトでは別のボリュームになる (AC4)。"""
     monkeypatch.setenv("COMPOSE_PROJECT_NAME", "bi-tools")
-    _write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
+    write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
 
     compose.generate_scaled_compose(scale=1)
 
-    dev = _load_scaled(in_tmp_cwd)["services"]["dev-1"]
-    assert _mount_source(dev, VSCODE_TARGET) == "devbase_vscode_bi-tools_1"
+    dev = load_scaled(in_tmp_cwd)["services"]["dev-1"]
+    assert mount_source(dev, VSCODE_TARGET) == "devbase_vscode_bi-tools_1"
 
 
 def test_existing_mounts_are_kept(in_tmp_cwd):
     """既存のマウント (/work 等) を壊さない。"""
-    _write_compose(in_tmp_cwd, {
+    write_compose(in_tmp_cwd, {
         "dev": {"image": "dev:latest", "volumes": ["./src:/src"]},
     })
 
     compose.generate_scaled_compose(scale=1)
 
-    dev = _load_scaled(in_tmp_cwd)["services"]["dev-1"]
+    dev = load_scaled(in_tmp_cwd)["services"]["dev-1"]
     assert "./src:/src" in dev["volumes"]
-    assert _mount_source(dev, "/work") == "devbase_work_1"
+    assert mount_source(dev, "/work") == "devbase_work_1"
 
 
 def test_non_dev_services_do_not_get_the_mount(in_tmp_cwd):
     """VS Code が入るのは dev だけ。db 等へは足さない。"""
-    _write_compose(in_tmp_cwd, {
+    write_compose(in_tmp_cwd, {
         "dev": {"image": "dev:latest"},
         "db": {"image": "mysql:8"},
     })
 
     compose.generate_scaled_compose(scale=1)
 
-    db = _load_scaled(in_tmp_cwd)["services"]["db"]
-    assert _mount_source(db, VSCODE_TARGET) is None
+    db = load_scaled(in_tmp_cwd)["services"]["db"]
+    assert mount_source(db, VSCODE_TARGET) is None
 
 
 # ---------------------------------------------------------------------------
@@ -121,23 +98,23 @@ def test_non_dev_services_do_not_get_the_mount(in_tmp_cwd):
 
 def test_vscode_volumes_are_declared_external(in_tmp_cwd):
     """devbase が作るボリュームなので external: true で宣言する。"""
-    _write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
+    write_compose(in_tmp_cwd, {"dev": {"image": "dev:latest"}})
 
     compose.generate_scaled_compose(scale=2)
 
-    volumes = _load_scaled(in_tmp_cwd)["volumes"]
+    volumes = load_scaled(in_tmp_cwd)["volumes"]
     assert volumes["devbase_vscode_myapp-ai_1"] == {"external": True}
     assert volumes["devbase_vscode_myapp-ai_2"] == {"external": True}
 
 
 def test_project_volumes_are_kept(in_tmp_cwd):
     """プロジェクトが宣言したボリュームは残る。"""
-    _write_compose(
+    write_compose(
         in_tmp_cwd, {"dev": {"image": "dev:latest"}}, volumes={"mysql": None})
 
     compose.generate_scaled_compose(scale=1)
 
-    assert "mysql" in _load_scaled(in_tmp_cwd)["volumes"]
+    assert "mysql" in load_scaled(in_tmp_cwd)["volumes"]
 
 
 # ---------------------------------------------------------------------------
@@ -149,15 +126,15 @@ def test_declared_vscode_mount_is_not_overridden(in_tmp_cwd):
 
     ホストのディレクトリを bind したい構成を devbase が奪わないため。
     """
-    _write_compose(in_tmp_cwd, {
+    write_compose(in_tmp_cwd, {
         "dev": {"image": "dev:latest",
                 "volumes": [f"./vscode-server:{VSCODE_TARGET}"]},
     })
 
     compose.generate_scaled_compose(scale=1)
 
-    scaled = _load_scaled(in_tmp_cwd)
+    scaled = load_scaled(in_tmp_cwd)
     dev = scaled["services"]["dev-1"]
-    assert _mount_source(dev, VSCODE_TARGET) == "./vscode-server"
+    assert mount_source(dev, VSCODE_TARGET) == "./vscode-server"
     # 使わないボリュームを宣言しない (external は実体を要求する)
     assert "devbase_vscode_myapp-ai_1" not in scaled["volumes"]