"""container compose 의 env 합성 + up argv 순수 함수.

이 task 는 docker 실행 없는 순수 함수만 담는다(서비스 목록/라벨은 인자로 받아
generator 가 순수하게 유지된다). 실제 compose 파일 파싱·override 파일 쓰기·docker
실행은 후속 task/e2e 로 미룬다.

env 우선순위(spec §6): compose 가 `--env-file <worktree .env>` 다음에
`--env-file <override>` 를 읽어 중복 키는 override 가 이긴다.

라벨 SSOT(spec §7): 모든 컨테이너는 `okstra.task-key`/`okstra.project-name`/
`okstra.run-trace` 라벨을 달아야 후속 status/down 이 라벨 질의로 그룹을 찾는다.
`docker compose up` 에는 `--label` 플래그가 없으므로(compose v5 `up --help` +
docker compose 문서로 확인), 라벨은 `-f <override.yml>` override 파일이 각 서비스
밑에 `labels:` 를 합쳐 주입한다.
"""
from __future__ import annotations

import argparse
import json
import re
import subprocess
import sys
import time
from pathlib import Path
from typing import Callable, Iterable

from .stage_targets import PrepareError  # 단일 PrepareError 타입 재노출(run.py 와 공유)
from .stage_map import StageMapError, parse_stage_map_file, stage_map_records
from .json_boundary import JsonBoundaryError, load_owned_object, write_owned_object_atomic
from .fixed_text import line

OKSTRA_LABEL_KEYS = ("okstra.task-key", "okstra.project-name", "okstra.run-trace")

COMPOSE_FILENAME = "docker-compose.yml"

def merge_env(base_env: dict, override: dict) -> dict:
    """base_env 위에 override 를 얹는다. 동일 키는 override 가 이긴다."""
    return {**base_env, **override}


def build_label_override_yaml(services: list[str], labels: dict) -> str:
    """각 서비스 밑에 `labels:` 를 주입하는 compose override yaml 본문을 낸다.

    docker compose 가 `-f` 로 합칠 때 이 본문이 컨테이너 라벨을 단다. 디스크 쓰기는
    호출자(후속 task) 몫이며 이 함수는 문자열만 반환한다.
    """
    lines = ["services:"]
    for service in services:
        lines.append(f"  {service}:")
        lines.append("    labels:")
        for key, value in labels.items():
            lines.append(f"      {key}: {_yaml_quote(value)}")
    return "\n".join(lines) + "\n"


def _yaml_quote(value: str) -> str:
    """라벨 값을 YAML double-quoted scalar 로 감싼다.

    task-key 는 raw 보존(slugify 안 함)이라 ` #`(주석), `{`/`[`(flow), `&`/`*`
    (anchor/alias), `:` 같은 indicator 문자가 들어올 수 있다. 무인용으로 쓰면
    override 파일이 깨져 `up` 이 중단되거나 라벨이 silently 잘린다 — double-quote
    안에서 `\\`·`"` 만 이스케이프하면 모든 indicator 가 평문으로 안전해진다."""
    escaped = str(value).replace("\\", "\\\\").replace('"', '\\"')
    return f'"{escaped}"'


def build_compose_up_argv(
    *,
    project_name: str,
    compose_path: str,
    worktree_env_path: str,
    override_env_path: str,
    label_override_path: str,
) -> list[str]:
    """`docker compose up -d` argv 를 낸다.

    - `-f` 는 base compose → label override 순서다. `-f` 가 하나라도 명시되면
      docker compose 는 cwd 의 docker-compose.yml 자동탐색을 끄므로, 베이스 파일을
      반드시 첫 `-f` 로 직접 넘겨야 한다(누락 시 라벨 override 만 머지돼 'no image'
      로 실패). label override 가 뒤에 와서 base 위에 `labels:` 만 덧입힌다.
    - `--env-file` 은 worktree → override 순서라 override 가 나중(우선)에 온다.
      worktree `.env` 가 없으면 그 플래그를 통째로 뺀다 — docker compose v2 는
      존재하지 않는 명시적 `--env-file` 을 hard error 로 처리하므로(implicit `.env`
      와 달리) 빈 경로를 넣으면 `.env` 없는 평범한 프로젝트가 전부 up 실패한다.
    - 라벨은 `-f <label_override_path>` 로 배선한다(`up --label` 미지원).
    """
    argv = [
        "docker", "compose",
        "-p", project_name,
        "-f", compose_path,
        "-f", label_override_path,
    ]
    if worktree_env_path:
        argv += ["--env-file", worktree_env_path]
    argv += ["--env-file", override_env_path, "up", "-d"]
    return argv


# --------------------------------------------------------------------------- #
# compose 파싱 — docker 가 정본(canonical). 손수 정규식 파서는 라벨 SSOT 를
# silent 하게 깨므로 쓰지 않는다(inline comment / quoted name / 4-space 들여쓰기
# 오파싱 → 라벨 미부착 → status/down 라벨 쿼리가 빈 결과 → 컨테이너 leak).
# `docker compose config` 는 parse 일 뿐(run 아님)이라 up 전에 호출 가능하다.
# --------------------------------------------------------------------------- #

_SERVICES_BLOCK_RE = re.compile(r"^services:\s*$", re.M)


def compose_has_services_block(compose_text: str) -> bool:
    """top-level `services:` 아래에 들여쓰기된 내용이 한 줄이라도 있으면 True.

    docker 파싱 결과가 0개일 때 '빈 compose(정상 0개)' 인지 '오파싱(loud abort)'
    인지 가르는 ground-truth 다. YAML 의미 파싱이 아니라 블록 존재만 본다."""
    lines = compose_text.splitlines()
    for i, line in enumerate(lines):
        if not _SERVICES_BLOCK_RE.match(line):
            continue
        for nxt in lines[i + 1:]:
            if not nxt.strip() or nxt.lstrip().startswith("#"):
                continue
            return nxt.startswith((" ", "\t"))  # 들여쓰기된 첫 콘텐츠 줄
    return False


def _compose_config_argv(
    compose_path: Path, env_files: Iterable[str], tail: list[str],
) -> list[str]:
    """`docker compose [--env-file ...] -f <compose> config <tail>` argv.

    `--env-file` 과 cwd 는 up 과 반드시 동일해야 한다 — config 가 변수
    interpolation 을 다른 env 로 해소하면 bind-mount escape 탐지/서비스 파싱이
    실제 배포와 다른 값으로 계산된다."""
    argv = ["docker", "compose"]
    for env_file in env_files:
        argv += ["--env-file", env_file]
    argv += ["-f", str(compose_path), "config", *tail]
    return argv


def resolve_compose_services(
    compose_path: Path, env_files: Iterable[str] = (), cwd: str | None = None,
) -> list[str]:
    """`docker compose -f <compose> config --services` 로 서비스 목록을 얻는다.

    docker 가 정본 파서다. services 블록이 비어있지 않은데 0개가 나오면(docker
    부재/파싱 실패 포함) 라벨 없는 컨테이너를 띄우느니 PrepareError 로 loud 하게
    중단한다 — 라벨 SSOT 를 silent 하게 깨지 않는다. `--env-file`/cwd 는 up 과
    동일하게 넘겨 interpolation 결과를 일치시킨다."""
    out = _docker_run(
        _compose_config_argv(compose_path, env_files, ["--services"]), cwd=cwd)
    services = [s.strip() for s in out.splitlines() if s.strip()]
    if not services and compose_has_services_block(
            compose_path.read_text(encoding="utf-8")):
        raise PrepareError(
            f"container up: {compose_path.name} 에 services 가 정의돼 있으나 "
            "`docker compose config --services` 가 빈 목록을 반환했습니다 "
            "(docker 부재 또는 compose 파싱 실패). 라벨 미부착 컨테이너 leak 을 "
            "막기 위해 중단합니다 — docker 가용성과 compose 문법을 확인하세요.")
    return services


def scan_escaping_host_paths(
    compose_path: Path, worktree_root: Path,
    env_files: Iterable[str] = (), cwd: str | None = None,
) -> list[str]:
    """worktree 밖(절대경로 or `../`)을 가리키는 bind mount host 경로를 모은다.

    spec §6: worktree 를 벗어나는 마운트는 경고 후 진행(중단 아님). docker 가
    정규화한 config(`--format json`)의 volumes 를 보므로 short/long-form 모두
    커버한다. docker 실패 시 빈 목록(경고만 누락, 진행에는 영향 없음). `--env-file`/
    cwd 는 up 과 동일하게 넘겨 `${VAR}` 마운트 경로가 실제 배포와 일치하게 한다."""
    out = _docker_run(
        _compose_config_argv(compose_path, env_files, ["--format", "json"]), cwd=cwd)
    try:
        config = json.loads(out) if out.strip() else {}
    except json.JSONDecodeError:
        return []
    escaping: list[str] = []
    for svc in (config.get("services") or {}).values():
        for vol in (svc.get("volumes") or []):
            source = vol.get("source") if isinstance(vol, dict) else None
            if isinstance(source, str) and _is_escaping_host_path(source):
                escaping.append(source)
    return escaping


def _is_escaping_host_path(source: str) -> bool:
    """named volume(슬래시 없음)은 제외, 절대경로/`../` host 경로만 escape."""
    if source.startswith("/") or source.startswith("..") or "/../" in source:
        return True
    return False


# --------------------------------------------------------------------------- #
# docker 얇은 래퍼(daemon 필요 — 게이트 로직과 분리해 유닛 테스트 가능하게)
# --------------------------------------------------------------------------- #

def _compose_ps(project_name: str, cwd: str, include_stopped: bool = False) -> list[dict]:
    argv = ["docker", "compose", "-p", project_name, "ps"]
    if include_stopped:
        argv.append("-a")
    argv += ["--format", "json"]
    result = subprocess.run(argv, capture_output=True, text=True, check=False, cwd=cwd)
    rows: list[dict] = []
    for line in (result.stdout or "").splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            rows.append(json.loads(line))
        except json.JSONDecodeError:
            continue
    return rows


def _docker_run(argv: list[str], cwd: str | None = None) -> str:
    result = subprocess.run(argv, capture_output=True, text=True, check=False, cwd=cwd)
    return result.stdout or ""


def query_containers_by_label(project_name: str) -> list[dict]:
    """라벨 SSOT(spec §11): `okstra.project-name` 으로 컨테이너를 질의한다.

    registry 가 아니라 docker 라벨이 컨테이너 존재의 기준값이다."""
    out = _docker_run([
        "docker", "ps", "-a", "--filter",
        f"label=okstra.project-name={project_name}",
        "--format", (
            "{{.ID}}\t{{.Names}}\t{{.State}}\t"
            '{{.Label "com.docker.compose.service"}}\t{{.Ports}}'
        ),
    ])
    rows: list[dict] = []
    for line in out.splitlines():
        parts = line.split("\t")
        if len(parts) >= 5:
            rows.append({
                "id": parts[0], "name": parts[1], "state": parts[2],
                "service": parts[3], "ports": parts[4],
            })
    return rows


def _service_state(row: dict) -> str:
    return (row.get("State") or row.get("state") or "").lower()


def _service_health(row: dict) -> str:
    return (row.get("Health") or row.get("health") or "").lower()


def poll_healthcheck(
    *, project_name: str, cwd: str, services: list[str],
    healthcheck_timeout_seconds: int, healthcheck_interval_seconds: int,
    sleep: Callable[[float], None] = time.sleep,
) -> list[str]:
    """healthy(또는 healthcheck 미정의 시 running)까지 폴링. 미기동 서비스명 반환.

    healthcheck 가 정의된 서비스는 Health=healthy 를, 없는 서비스는 State=running
    을 성공 조건으로 본다(spec §4 step6). `deploy.replicas`>1 이면 `ps` 가 한
    서비스에 여러 컨테이너 행을 내므로, 서비스명별로 묶어 **모든** 레플리카가
    ready 일 때만 통과시킨다(한 행으로 collapse 하면 건강한 레플리카가 비정상
    형제를 가린다). 빈 리스트면 전부 기동.

    `-a`(include_stopped)로 폴링해 크래시-종료한 컨테이너 행까지 본다 —
    `_started_services` 와 같은 가시성. 안 그러면 멀티-레플리카에서 죽은 레플리카가
    running-only ps 에 행이 없어, 살아있는 형제만으로 `all(ready)` 가 통과해 죽은
    배포를 healthy 로 오인한다(exited 행은 _container_ready 가 False 를 돌려준다)."""
    deadline = time.monotonic() + healthcheck_timeout_seconds
    pending = set(services)
    while pending and time.monotonic() < deadline:
        by_service = _group_ps_rows_by_service(
            _compose_ps(project_name, cwd, include_stopped=True))
        for svc in list(pending):
            rows = by_service.get(svc)
            if rows and all(_container_ready(r) for r in rows):
                pending.discard(svc)
        if pending:
            sleep(healthcheck_interval_seconds)
    return sorted(pending)


def _group_ps_rows_by_service(rows: list[dict]) -> dict[str, list[dict]]:
    out: dict[str, list[dict]] = {}
    for row in rows:
        key = row.get("Service") or row.get("Name")
        if key:
            out.setdefault(key, []).append(row)
    return out


def _container_ready(row: dict) -> bool:
    health = _service_health(row)
    if health:
        return health == "healthy"
    return _service_state(row) == "running"


# --------------------------------------------------------------------------- #
# up 흐름 (provision_container_group) — 단계별 헬퍼로 분해
# --------------------------------------------------------------------------- #

def _split_task_key(task_key: str) -> tuple[str, str, str]:
    parts = task_key.split(":")
    if len(parts) != 3 or not all(parts):
        raise PrepareError(
            f"container: task-key 형식은 '<project-id>:<task-group>:<task-id>' 이어야 "
            f"합니다(받음: {task_key!r})")
    return parts[0], parts[1], parts[2]


def _resolve_up_inputs(project_root: Path, task_key: str) -> dict:
    """task-key 로 up 에 필요한 입력(worktree/stage_map/done_rows/anchor)을 모은다."""
    from . import worktree_registry as _reg
    from .consumers import read_stage_consumer_state
    from .plan_run_root import resolve_plan_run_root_by_task_key

    project_id, group, task_id = _split_task_key(task_key)
    entry = _reg.lookup(project_id, group, task_id)
    if entry is None or not entry.worktree_path:
        raise PrepareError(
            f"container up: task-key {task_key} 의 task worktree 가 registry 에 "
            "없습니다. 먼저 해당 task 의 phase 를 한 번 실행해 worktree 를 만드세요.")
    plan = resolve_plan_run_root_by_task_key(
        project_root=project_root, task_group=group, task_id=task_id)
    done_rows = read_stage_consumer_state(
        plan.run_root, recover_from_carry=True,
    ).done_rows
    stage_map = _plan_stage_map(plan.approved_plan_path)
    anchor = _reg.get_implementation_base(project_id, group, task_id) or ""
    return {
        "project_id": project_id, "task_group": group, "task_id": task_id,
        "worktree_path": entry.worktree_path, "stage_map": stage_map,
        "done_rows": done_rows, "anchor_base": anchor,
    }


def _plan_stage_map(approved_plan_path: str) -> list[dict]:
    """Parse the approved plan's complete Stage Map for container gating."""
    try:
        return stage_map_records(parse_stage_map_file(Path(approved_plan_path)))
    except StageMapError as exc:
        raise PrepareError(
            "approved-plan 의 Stage Map 을 신뢰할 수 없어 거부합니다 "
            f"({approved_plan_path}): {exc.reason}. plan 의 Stage Map 을 점검하세요."
        ) from exc


def _verify_compose_present(worktree_root: Path) -> Path:
    compose = worktree_root / COMPOSE_FILENAME
    if not compose.is_file():
        raise PrepareError(
            f"container up: worktree 루트에 {COMPOSE_FILENAME} 가 없습니다 "
            f"({compose}). okstra 는 compose 파일을 생성하지 않습니다 — "
            "프로젝트에 직접 추가한 뒤 다시 시도하세요.")
    return compose


def _synthesize_env_override(worktree_root: Path, cp: dict) -> tuple[str, str]:
    """worktree `.env`(okstra-owned 입력) → env.override 합성. (worktree_env, override) 반환.

    spec §6 artifact-home: 입력 `.env` 는 raw project root 가 아니라 worktree
    트리에서 해소한다. override 는 container_dir 밑에 쓴다."""
    worktree_env = worktree_root / ".env"
    base = _read_env_file(worktree_env)
    override = merge_env(base, {})
    cp["container_dir"].mkdir(parents=True, exist_ok=True)
    cp["env_override"].write_text(
        "".join(f"{k}={v}\n" for k, v in override.items()), encoding="utf-8")
    # 파일이 없으면 빈 경로를 반환해 up argv 가 `--env-file <missing>` 을 넣지
    # 않게 한다(존재하는 override 는 항상 위에서 써 두므로 그 쪽은 안전).
    worktree_env_arg = str(worktree_env) if worktree_env.is_file() else ""
    return worktree_env_arg, str(cp["env_override"])


def _read_env_file(path: Path) -> dict:
    out: dict[str, str] = {}
    if not path.is_file():
        return out
    for line in path.read_text(encoding="utf-8").splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, v = line.split("=", 1)
        out[k.strip()] = v.strip()
    return out


def _write_label_override(cp: dict, services: list[str], labels: dict) -> str:
    cp["container_dir"].mkdir(parents=True, exist_ok=True)
    path = cp["container_dir"] / "labels.override.yml"
    path.write_text(build_label_override_yaml(services, labels), encoding="utf-8")
    return str(path)


def provision_container_group(
    *, project_root, task_key: str,
    healthcheck_timeout_seconds: int = 120,
    healthcheck_interval_seconds: int = 3,
) -> dict:
    """skill / bin okstra / okstra.sh 가 수렴하는 container up 단일 진입점."""
    from . import paths
    from .ids import compose_project_name, build_container_run_trace

    project_root = Path(project_root)
    inputs = _resolve_up_inputs(project_root, task_key)
    worktree = Path(_integrate_source_tree(inputs))
    compose = _verify_compose_present(worktree)
    cp = paths.container_paths(project_root, inputs["task_group"], inputs["task_id"])
    project_name = compose_project_name(
        inputs["project_id"], inputs["task_group"], inputs["task_id"])
    run_trace = build_container_run_trace(
        inputs["project_id"], inputs["task_group"], inputs["task_id"])
    worktree_env, override_env = _synthesize_env_override(worktree, cp)
    env_files = [p for p in (worktree_env, override_env) if p]
    warnings = scan_escaping_host_paths(compose, worktree, env_files, str(worktree))
    services = resolve_compose_services(compose, env_files, str(worktree))
    started = _compose_up(
        cp=cp, worktree=worktree, compose=compose, project_name=project_name,
        task_key=task_key, run_trace=run_trace, services=services,
        worktree_env=worktree_env, override_env=override_env)
    if not started:
        # ps 가 빈 출력(데몬 일시 오류)이거나 모든 서비스가 미생성이면 started 가
        # 비는데, 빈 리스트는 poll_healthcheck 를 vacuous 통과시켜 검증 없이 성공으로
        # 보고된다. fail-open 대신 raise 하되, `up -d` 후 컨테이너가 떠 있을 수도
        # 있으므로(transient ps 실패) deploy-state 를 먼저 써 `down --all` 이 회수
        # 가능하게 한다 — 실제 기동분을 모르니 full services 를 best-effort 로 기록.
        _write_deploy_state(
            cp, project_name=project_name, task_key=task_key, services=services,
            warnings=warnings)
        raise PrepareError(
            f"container up: `up -d` 후 기동된 컨테이너가 없습니다(`docker compose -p "
            f"{project_name} ps -a` 로 확인). compose 정의·활성 profile 을 점검한 뒤 "
            f"`okstra container down --task-key {task_key}` 로 정리하고 다시 시도하세요.")
    failed = poll_healthcheck(
        project_name=project_name, cwd=str(worktree), services=started,
        healthcheck_timeout_seconds=healthcheck_timeout_seconds,
        healthcheck_interval_seconds=healthcheck_interval_seconds)
    if failed:
        # 컨테이너는 이미 떠 있다(unhealthy/느림) — deploy-state 를 먼저 써 둬야
        # `down --all` 스윕(deploy-state.json 있는 task 만 발견)이 회수할 수 있다.
        _write_deploy_state(
            cp, project_name=project_name, task_key=task_key, services=started,
            warnings=warnings, healthcheckFailed=failed)
        raise PrepareError(
            f"container up: 다음 서비스가 시간 내 기동하지 못했습니다: {', '.join(failed)} "
            f"(timeout={healthcheck_timeout_seconds}s). `docker compose -p "
            f"{project_name} logs` 로 원인을 확인한 뒤 `okstra container down "
            f"--task-key {task_key}` 로 정리하세요.")
    # deploy-state 는 실제 기동된 started 만 기록한다 — full services 를 쓰면
    # profiles 로 안 뜬 서비스가 `down --all` 스윕의 대상으로 남는다.
    _write_deploy_state(cp, project_name=project_name, task_key=task_key,
                        services=started, warnings=warnings)
    return {"projectName": project_name, "services": started,
            "warnings": warnings}


def _integrate_source_tree(inputs: dict) -> str:
    """stage 통합(teardown=False)으로 소스트리를 확정하고 worktree 경로를 반환."""
    from .stage_targets import resolve_and_integrate_whole_task

    resolve_and_integrate_whole_task(
        teardown=False, project_id=inputs["project_id"],
        task_group=inputs["task_group"], task_id=inputs["task_id"],
        task_worktree_path=inputs["worktree_path"], stage_map=inputs["stage_map"],
        done_rows=inputs["done_rows"], anchor_base=inputs["anchor_base"])
    return inputs["worktree_path"]


def _compose_up(
    *, cp: dict, worktree: Path, compose: Path, project_name: str, task_key: str,
    run_trace: str, services: list[str], worktree_env: str, override_env: str,
) -> list[str]:
    """label override 주입 → compose up → 실제 기동된 서비스명 반환.

    반환은 `up -d` 가 실제로 생성한 서비스만(=ps 행 존재) 추린 목록이다 — compose
    `profiles:` 로 비활성화돼 기동되지 않은 서비스를 healthcheck 대기 대상에서
    빼, 영영 pending 으로 남아 오탐 timeout 을 내지 않게 한다."""
    labels = {
        "okstra.task-key": task_key,
        "okstra.project-name": project_name,
        # run-trace 는 이 기능의 trace 식별자로 채워 project-name 과
        # distinct 하게 둔다(M1).
        "okstra.run-trace": run_trace,
    }
    label_path = _write_label_override(cp, services, labels)
    argv = build_compose_up_argv(
        project_name=project_name, compose_path=str(compose),
        worktree_env_path=worktree_env, override_env_path=override_env,
        label_override_path=label_path)
    result = subprocess.run(argv, cwd=str(worktree), check=False)
    if result.returncode != 0:
        raise PrepareError(
            f"container up: `docker compose up` 가 실패했습니다(exit "
            f"{result.returncode}). `docker compose -p {project_name} logs` 로 "
            "원인을 확인하세요.")
    return _started_services(project_name, str(worktree), services)


def _started_services(project_name: str, cwd: str, services: list[str]) -> list[str]:
    """`up -d` 로 실제 생성된(running 또는 exited) 서비스만 입력 순서대로 추린다.

    `-a` 로 종료된 컨테이너까지 본다 — 부팅 직후 크래시-종료한 서비스를 profiles
    로 아예 안 뜬 서비스와 구분하려는 것. 전자는 ps 행이 남아 healthcheck 대기
    대상에 포함되고 running 이 아니라 timeout 으로 실패한다(죽은 배포를 성공으로
    오인 차단). 후자는 `-a` 에도 행이 없어 제외된다."""
    created = set(_group_ps_rows_by_service(
        _compose_ps(project_name, cwd, include_stopped=True)))
    return [s for s in services if s in created]


def _write_deploy_state(cp: dict, **state) -> None:
    cp["container_dir"].mkdir(parents=True, exist_ok=True)
    write_owned_object_atomic(cp["deploy_state"], state, artifact="container deploy state")


# --------------------------------------------------------------------------- #
# status / down
# --------------------------------------------------------------------------- #

def status_container_group(*, project_root, task_key: str) -> dict:
    """라벨 쿼리 — 컨테이너 존재 여부의 기준값(SSOT)."""
    from .ids import compose_project_name

    project_id, group, task_id = _split_task_key(task_key)
    project_name = compose_project_name(project_id, group, task_id)
    return {"projectName": project_name,
            "containers": query_containers_by_label(project_name)}


def down_container_group(*, project_root, task_key: str = "", all_groups: bool = False) -> dict:
    """라벨 쿼리로 컨테이너를 down 한다."""
    from .ids import compose_project_name

    project_root = Path(project_root)
    results = []
    for project_id, group, task_id in _down_targets(project_root, task_key, all_groups):
        project_name = compose_project_name(project_id, group, task_id)
        # `docker rm -f` 는 컨테이너만 지우고 compose 가 만든 네트워크를 누수시킨다.
        # compose 정본 teardown 으로 프로젝트(라벨 com.docker.compose.project=<name>)
        # 단위 컨테이너+네트워크를 회수한다. `-v` 는 의도적으로 빼서 compose 에
        # 선언된 named volume(예: DB 데이터)을 보존한다 — down 한 번에 사용자
        # 영속 데이터를 파괴하지 않는다.
        _docker_run(["docker", "compose", "-p", project_name, "down",
                     "--remove-orphans"])
        results.append({"projectName": project_name})
    return {"downed": results}


def _down_targets(project_root, task_key: str, all_groups: bool) -> list[tuple[str, str, str]]:
    if all_groups:
        return _discover_container_task_keys(Path(project_root))
    if not task_key:
        raise PrepareError("container down: task-key 또는 --all 중 하나가 필요합니다.")
    return [_split_task_key(task_key)]


def _discover_container_task_keys(project_root: Path) -> list[tuple[str, str, str]]:
    """프로젝트 안에서 container deploy-state 가 있는 task-key 들을 찾는다."""
    from okstra_project.dirs import project_json_path, tasks_root as _tasks_root

    project_id = _read_project_id(project_json_path(project_root))
    tasks_root = _tasks_root(project_root)
    out: list[tuple[str, str, str]] = []
    if not tasks_root.is_dir():
        return out
    for group_dir in tasks_root.iterdir():
        for task_dir in (group_dir.iterdir() if group_dir.is_dir() else []):
            if (task_dir / "container" / "deploy-state.json").is_file():
                out.append((project_id, group_dir.name, task_dir.name))
    return out


def _read_project_id(project_json: Path) -> str:
    if not project_json.is_file():
        return ""
    try:
        return load_owned_object(project_json, artifact="project configuration").get(
            "projectId", ""
        )
    except JsonBoundaryError:
        return ""


# --------------------------------------------------------------------------- #
# argparse 디스패치(recap.py 패턴)
# --------------------------------------------------------------------------- #

_CLI_EPILOG = r"""Usage:
  okstra container up [options]
  okstra container status [options]
  okstra container down [options]
"""
_CLI_DESCRIPTION = "Manage the okstra container runtime."


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description=_CLI_DESCRIPTION,
        epilog=_CLI_EPILOG,
        formatter_class=argparse.RawDescriptionHelpFormatter,
        prog="okstra container")
    sub = parser.add_subparsers(dest="command", required=True)
    for name in ("up", "status", "down"):
        sp = sub.add_parser(name)
        sp.add_argument("--project-root", required=True)
        sp.add_argument("--task-key", default="")
        sp.add_argument("--text", action="store_true")
        if name == "down":
            sp.add_argument("--all", action="store_true")
    args = parser.parse_args(argv)
    result = _dispatch(args)
    print(
        render_container_text(args.command, result)
        if args.text
        else json.dumps(result, ensure_ascii=False, indent=2)
    )
    return 0


def render_container_text(command: str, payload: dict) -> str:
    """container 명령별 승인된 상태만 고정 순서로 투영한다."""
    if command not in {"up", "status", "down"}:
        raise ValueError("unknown container text purpose")
    rows = [f"Okstra container {command}\n", line("Status", "ready")]
    for label, key in (("Project name", "projectName"), ("Note", "note")):
        if key in payload:
            rows.append(line(label, payload.get(key)))
    if command == "up":
        rows.extend(_render_container_up(payload))
    elif command == "status":
        rows.extend(_render_container_status(payload))
    else:
        rows.extend(_render_container_down(payload))
    return "".join(rows)


def _render_container_up(payload: dict) -> list[str]:
    rows: list[str] = []
    for index, service in enumerate(payload.get("services") or [], 1):
        rows.append(line(f"Service {index} name", service))
    for index, warning in enumerate(payload.get("warnings") or [], 1):
        rows.append(line(f"Warning {index} text", warning))
    return rows


def _render_container_status(payload: dict) -> list[str]:
    containers = payload.get("containers") if isinstance(payload.get("containers"), list) else []
    rows: list[str] = []
    for index, item in enumerate(containers, 1):
        values = item if isinstance(item, dict) else {}
        for label, key in (("name", "name"), ("service", "service"), ("state", "state"),
                           ("status", "status"), ("ports", "ports")):
            if key in values:
                rows.append(line(f"Container {index} {label}", values.get(key)))
    return rows


def _render_container_down(payload: dict) -> list[str]:
    downed = payload.get("downed") if isinstance(payload.get("downed"), list) else []
    rows: list[str] = []
    for index, item in enumerate(downed, 1):
        values = item if isinstance(item, dict) else {}
        rows.append(line(f"Downed {index} project name", values.get("projectName")))
    return rows


def _dispatch(args) -> dict:
    pr = args.project_root
    if args.command == "up":
        return provision_container_group(project_root=pr, task_key=args.task_key)
    if args.command == "status":
        return status_container_group(project_root=pr, task_key=args.task_key)
    return down_container_group(
        project_root=pr, task_key=args.task_key, all_groups=args.all)


if __name__ == "__main__":
    raise SystemExit(main(sys.argv[1:]))