"""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 os
import re
import shlex
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"

# spec §5/§10 — 에러 트리거 화이트리스트. 이 정규식 중 하나라도 라인에 걸려야
# 2단계(LLM analyze)로 넘어간다. 건강한 컨테이너는 어느 것도 안 걸려 토큰 0.
ERROR_PATTERNS = (
    re.compile(r"\bERROR\b"),
    re.compile(r"\bFATAL\b"),
    re.compile(r"\bCRITICAL\b"),
    re.compile(r"\bPANIC\b"),
    re.compile(r"\b\w+(?:Error|Exception)\b"),
    re.compile(r"^Traceback \(most recent call last\):"),
    re.compile(r"\b(?:exit(?:ed)?|exit code|status)\s+(?:code\s+)?[1-9]\d*\b", re.I),
    re.compile(r"\bUnhandled\b", re.I),
)

# error_signature 정규화: 변동성 노이즈(타임스탬프/pid/라인참조/hex 주소)만 깎고,
# ERROR_PATTERNS 가 *탐지하는* 의미 있는 코드(HTTP status, exit/signal code)는 보존한다.
# 블랭킷 `\d+` 치환은 distinct 에러(500 vs 404, exit 1 vs 137)를 한 시그니처로
# collapse 시켜 디바운스가 영영 분석을 막으므로 쓰지 않는다(spec §10 signal vs noise).
_NOISE_SUBS = (
    (re.compile(
        r"\d{4}-\d{2}-\d{2}[ T]\d{2}:\d{2}:\d{2}(?:[.,]\d+)?(?:Z|[+-]\d{2}:?\d{2})?"
    ), "<ts>"),
    (re.compile(r"\b0x[0-9a-fA-F]+\b"), "<hex>"),
    (re.compile(r"\b(pid|tid|ppid)=\d+", re.I), r"\1=<n>"),
    (re.compile(r":\d+:"), ":<n>:"),
)
_WS_RE = re.compile(r"\s+")


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


def scan_log_chunk(text: str, patterns: Iterable[re.Pattern]) -> list[str]:
    """ERROR_PATTERNS 중 하나라도 걸리는 라인만 순서대로 반환(순수, docker 무의존)."""
    return [
        line
        for line in text.splitlines()
        if line and any(p.search(line) for p in patterns)
    ]


def error_signature(line: str) -> str:
    """디바운스 키: 변동성 노이즈만 깎고 의미 있는 코드는 보존한 라인.

    타임스탬프/pid/라인참조/hex 주소는 placeholder 로 정규화하지만 HTTP status·
    exit/signal code 같은 판별자는 그대로 둔다. 같은 에러가 timestamp·pid 만
    달리해 반복되면 동일 키 → analyze 1회(spec §10 디바운스), 그러나 500 vs 404
    처럼 의미가 다르면 별도 키 → 각각 분석된다.
    """
    s = line
    for pattern, repl in _NOISE_SUBS:
        s = pattern.sub(repl, s)
    return _WS_RE.sub(" ", s).strip()


def watcher_step(
    *,
    log_chunk: str,
    seen_signatures: set,
    findings_path: Path,
    analyze: Callable[[str], object],
) -> int:
    """log_chunk 를 1단계 스캔 → 새 시그니처마다 analyze 호출 + findings append.

    이미 본 시그니처는 디바운스(skip). 새 분석 건수를 반환한다. analyze 는 주입이라
    유닛 테스트는 스텁으로 LLM 없이 행위를 검증한다.
    """
    new_count = 0
    for line in scan_log_chunk(log_chunk, ERROR_PATTERNS):
        sig = error_signature(line)
        if sig in seen_signatures:
            continue
        seen_signatures.add(sig)
        # 탐지 자체를 analyze 보다 먼저 영속화한다 — analyze(=pane spawn) 가 실패해도
        # 에러 라인은 findings 에 남는다. analyze 예외가 장기 watcher 루프를 죽이지
        # 않도록 격리하고, 실패 사실도 findings 에 기록한다(silent 중단 방지).
        _append_finding(Path(findings_path), sig, line)
        try:
            analyze(line)
        except Exception as exc:  # noqa: BLE001 — best-effort 외부 spawn, 루프 보존이 우선
            _append_finding(Path(findings_path), sig, f"[analyze 실패] {exc}")
        new_count += 1
    return new_count


def _append_finding(findings_path: Path, signature: str, line: str) -> None:
    findings_path.parent.mkdir(parents=True, exist_ok=True)
    block = f"## {signature}\n\n```\n{line}\n```\n\n"
    with findings_path.open("a", encoding="utf-8") as fh:
        fh.write(block)


# spec §9 — watcher AI 는 "탐지·보고만, 자동 치유 제외". 기본 analyze 가 기동하는
# 에이전트는 findings 파일 append 외 어떤 코드/설정 Edit/Write 도 금지한다. 이
# 계약 문구를 에이전트 프롬프트에 그대로 넣어 enforcement 를 명시한다.
WATCHER_READONLY_CONTRACT = (
    "You are a READ-ONLY container log watcher. You MUST NOT Edit or Write any "
    "code or config file. Your ONLY permitted write is appending an analysis "
    "note to the findings file. Detect and report; never auto-fix."
)


def default_analyze(findings_path: Path, *, session_pane: str | None = None):
    """프로덕션 기본 analyze 팩토리: read-only 계약을 건 에이전트를 pane 으로 기동.

    codex/antigravity-worker 와 동일한 split_container_pane 메커니즘을 쓰되
    WATCHER_READONLY_CONTRACT 를 프롬프트에 박아 코드/설정 수정 권한을 차단한다.
    라이브 spawn 세부(에이전트 CLI·모델·권한 스코프)는 아직 미구현이라 e2e 영역으로
    둔다 — session_pane 이 없으면 split_container_pane 이 guard 로 None 을 돌려주어
    실제 spawn 없이 no-op 이 된다(에러 탐지·findings 기록은 watcher_step 이 담당).
    프롬프트를 셸 명령으로 그대로 실행하지 않도록, 자동 `$TMUX_PANE` fallback 은
    두지 않는다 — 자연어 프롬프트가 pane 명령으로 새지 않게 하는 안전장치다.
    """
    from okstra_ctl import tmux

    def analyze(window: str) -> str | None:
        prompt = f"{WATCHER_READONLY_CONTRACT}\n\nError window:\n{window}\n"
        return tmux.split_container_pane(
            session_pane=session_pane,
            cwd=str(Path(findings_path).parent),
            command=prompt,
            title="okstra-watcher",
            scope_value=str(findings_path),
            kind="watcher",
        )

    return analyze


def build_watcher_pane_command(
    *, python_exe: str, module_root: str, project_name: str,
    service: str, findings_path: str, scan_interval_seconds: float,
) -> str:
    """watcher pane 이 실행할 셸 명령 문자열.

    `python -m okstra_ctl.container watch` 를 PYTHONPATH 와 함께 박는다 — tmux pane
    의 환경 상속에 의존하면 PYTHONPATH 가 비어 import 가 깨지므로, 현재 프로세스의
    module-root 를 명시적으로 넘긴다. 모든 인자는 shlex 로 인용해 경로에 공백/메타
    문자가 있어도 안전하다."""
    return " ".join([
        f"PYTHONPATH={shlex.quote(module_root)}",
        shlex.quote(python_exe), "-m", "okstra_ctl.container", "watch",
        "--project-name", shlex.quote(project_name),
        "--service", shlex.quote(service),
        "--findings", shlex.quote(findings_path),
        "--scan-interval", str(scan_interval_seconds),
    ])


def run_service_watcher(
    *, project_name: str, service: str, findings_path: str,
    scan_interval_seconds: float,
) -> None:
    """`container watch` 서브커맨드 본체 — 한 서비스의 감시 루프를 돈다.

    stage ① 경량 스캔(ERROR 패턴) → 매칭 시 findings append + stage ② analyze.
    사용자가 pane/세션을 종료(stop-watcher/down)할 때까지 지속한다."""
    analyze = default_analyze(Path(findings_path))
    run_watcher_loop(
        service=service, project_name=project_name,
        findings_path=Path(findings_path),
        scan_interval_seconds=scan_interval_seconds, analyze=analyze,
    )


def run_watcher_loop(
    *,
    service: str,
    project_name: str,
    findings_path: Path,
    scan_interval_seconds: float,
    analyze: Callable[[str], object],
) -> None:
    """`docker compose -p <project> logs --since <last>` 증분 fetch + sleep 루프.

    사용자 종료(stop-watcher/down) 까지 지속. 토큰은 ERROR_PATTERNS 가 걸릴 때만
    쓴다. 루프는 얇게 — 매 tick 의 판단은 watcher_step 에 위임한다.
    """
    seen: set = set()
    # 첫 tick 은 --since 없이 컨테이너 기동 이후 backlog 전체를 본다. `--since 0s`
    # 는 'now 이후'를 의미해 기동 중 찍힌 startup-failure 에러를 영구히 놓친다.
    since: str | None = None
    while True:
        tick_start = time.monotonic()
        chunk = _fetch_log_chunk(service=service, project_name=project_name, since=since)
        watcher_step(
            log_chunk=chunk, seen_signatures=seen,
            findings_path=Path(findings_path), analyze=analyze,
        )
        time.sleep(scan_interval_seconds)
        # 다음 윈도우는 '직전 fetch 이후 실제 경과시간'(sleep + analyze 포함)을
        # 덮는다. 고정 interval 윈도우는 analyze 가 interval 보다 오래 걸리면 그
        # 사이 찍힌 에러 라인을 영구히 놓친다. +1s 오버랩의 중복은 디바운스가 흡수.
        since = f"{int(time.monotonic() - tick_start) + 1}s"


def _fetch_log_chunk(*, service: str, project_name: str, since: str | None) -> str:
    argv = ["docker", "compose", "-p", project_name, "logs", "--no-color"]
    if since:  # 첫 tick(None)은 --since 생략 → 기동 backlog 전체
        argv += ["--since", since]
    argv.append(service)
    result = subprocess.run(argv, capture_output=True, text=True, check=False)
    return result.stdout or ""


# --------------------------------------------------------------------------- #
# 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,
    scan_interval_seconds: float = 5.0,
) -> dict:
    """skill / bin okstra / okstra.sh 가 수렴하는 container up 단일 진입점."""
    from . import paths
    from .ids import compose_project_name, build_container_session_name

    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"])
    session_name = build_container_session_name(
        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, session_name=session_name, 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, watch={"enabled": False, "note": "기동 컨테이너 없음"})
        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, watch={"enabled": False, "note": "healthcheck 미통과"},
            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}` 로 정리하세요.")
    # watcher/deploy-state 는 실제 기동된 started 만 대상으로 한다 — full services
    # 를 쓰면 profiles 미기동 서비스에 컨테이너 없는 tail/watcher pane 과 registry
    # 항목이 phantom 으로 생겨 배포 수명 동안 누수된다.
    watch = _spawn_watchers(cp=cp, worktree=worktree, project_name=project_name,
                            services=started, session_name=session_name,
                            project_root=project_root, inputs=inputs,
                            scan_interval_seconds=scan_interval_seconds)
    _write_deploy_state(cp, project_name=project_name, task_key=task_key,
                        services=started, warnings=warnings, watch=watch)
    return {"projectName": project_name, "services": started,
            "warnings": warnings, "watch": watch}


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,
    session_name: 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 식별자(container session name)로 채워
        # project-name 과 distinct 하게 둔다(M1).
        "okstra.run-trace": session_name,
    }
    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 _spawn_watchers(
    *, cp: dict, worktree: Path, project_name: str, services: list[str],
    session_name: str, project_root: Path, inputs: dict, scan_interval_seconds: float,
) -> dict:
    """tmux 가 있으면 detached 세션 + 서비스별 tail/watcher pane. 없으면 skip.

    pane split 실패(pane 한도/세션 외부 종료 등)는 watcher 만 degrade 시키고 raise
    하지 않는다 — 컨테이너는 이미 떠 있으므로 호출자가 deploy-state 를 반드시 써서
    `down --all` 스윕에서 보이게 해야 한다."""
    from . import tmux

    if not tmux.tmux_available():
        return {"enabled": False, "note": "감시 비활성(tmux 없음)"}
    cp["watchers_dir"].mkdir(parents=True, exist_ok=True)
    started: list[str] = []
    try:
        # holder pane 은 기본 셸로 띄워 오래 살린다(`true` 는 split 전에 세션을
        # 무너뜨림). container 태그를 붙여 down/stop-watcher 의 스코프 reap 로
        # tail/watcher pane 과 함께 회수되게 한다(마지막 pane 회수 시 세션 종료).
        pane = tmux.new_detached_session(session_name, str(worktree))
        tmux.tag_container_pane(pane, str(cp["container_dir"]))
        for svc in services:
            _spawn_service_panes(
                pane=pane, cp=cp, worktree=worktree, project_name=project_name,
                svc=svc, session_name=session_name, project_root=project_root,
                inputs=inputs, scan_interval_seconds=scan_interval_seconds)
            started.append(svc)
    except (RuntimeError, OSError, subprocess.SubprocessError) as exc:
        # watcher spawn 실패(tmux split/RuntimeError, registry flock/디스크 OSError,
        # tmux timeout/SubprocessError)는 watcher 만 degrade 시키고 raise 하지
        # 않는다 — 호출자가 deploy-state 를 반드시 써 컨테이너 leak 을 막는다.
        return {"enabled": True, "session": session_name, "services": started,
                "partial": True, "note": f"watcher 부분 실패: {exc}"}
    return {"enabled": True, "session": session_name, "services": started}


def _spawn_service_panes(
    *, pane: str, cp: dict, worktree: Path, project_name: str, svc: str,
    session_name: str, project_root: Path, inputs: dict, scan_interval_seconds: float,
) -> None:
    """한 서비스의 tail pane + watcher loop pane 을 split 하고 registry 에 등록."""
    from . import tmux, container_registry

    findings = cp["watchers_dir"] / f"{svc}.findings.md"
    tmux.split_container_pane(
        session_pane=pane, cwd=str(worktree),
        command=f"docker compose -p {project_name} logs -f {svc}",
        title=f"tail:{svc}", scope_value=str(cp["container_dir"]), kind="tail")
    wpane = tmux.split_container_pane(
        session_pane=pane, cwd=str(worktree),
        command=build_watcher_pane_command(
            python_exe=sys.executable,
            module_root=str(Path(__file__).resolve().parent.parent),
            project_name=project_name, service=svc, findings_path=str(findings),
            scan_interval_seconds=scan_interval_seconds),
        title=f"watch:{svc}", scope_value=str(cp["container_dir"]), kind="watcher")
    container_registry.reserve(
        project_root, inputs["task_group"], inputs["task_id"], svc,
        session_name=session_name, pane_id=wpane or "", findings_path=str(findings))


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 / logs / stop-watcher / down
# --------------------------------------------------------------------------- #

def status_container_group(*, project_root, task_key: str) -> dict:
    """라벨 쿼리(컨테이너 존재 SSOT) + registry(보조 watcher 메타) 합본."""
    from . import container_registry
    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)
    containers = query_containers_by_label(project_name)
    registry = container_registry.lookup(Path(project_root), group, task_id)
    return {"projectName": project_name, "containers": containers,
            "watchers": registry.get("services", {})}


def logs_container_group(*, project_root, task_key: str, service: str = "") -> dict:
    """deploy-state + watcher findings 경로를 안내(로그 자체는 tmux pane 에 있음)."""
    from . import paths, container_registry

    project_id, group, task_id = _split_task_key(task_key)
    cp = paths.container_paths(Path(project_root), group, task_id)
    services = container_registry.lookup(Path(project_root), group, task_id).get(
        "services", {})
    if service:
        services = {service: services[service]} if service in services else {}
    return {"watchersDir": str(cp["watchers_dir"]), "watchers": services}


def stop_watcher_group(*, project_root, task_key: str) -> dict:
    """watcher/tail pane(@okstra_container_run) 만 회수하고 컨테이너는 살려둔다."""
    project_id, group, task_id = _split_task_key(task_key)
    reaped = _reap_container_panes(Path(project_root), group, task_id)
    return {"reapedPanes": reaped, "note": "컨테이너는 유지됨(감시만 종료)"}


def down_container_group(*, project_root, task_key: str = "", all_groups: bool = False) -> dict:
    """라벨 쿼리로 컨테이너 down + tmux 실측으로 orphan watcher 회수.

    registry 는 보조 인덱스라 누락돼도 'watcher 없음'을 의미하지 않는다 — tmux 를
    직접 스캔(@okstra_container_run)해 orphan pane 까지 reap 한다(spec §11).
    pane 회수는 항상 현재 project-root 스코프로 한정한다 — 동시 세션이 흔한
    레포라 다른 task/프로젝트의 watcher pane 을 절대 죽이지 않는다(I2)."""
    from . import paths
    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"])
        reaped = _reap_container_panes(project_root, group, task_id)
        results.append({"projectName": project_name, "reapedPanes": reaped})
    # registry drift 로 등록 안 된 orphan pane 회수는 `--all`(이 프로젝트 전체)
    # 때만, 그것도 **이 project-root 의 .okstra/ prefix** 로만 스코프한다. 단일
    # down 은 위 per-task 회수로 끝낸다(글로벌 sweep 금지 — 동시 세션 간섭 차단).
    orphans = _reap_project_orphan_panes(project_root) if all_groups else []
    return {"downed": results, "orphanPanesReaped": orphans}


def _reap_project_orphan_panes(project_root: Path) -> list[str]:
    prefix = str((project_root / ".okstra").resolve())
    return _kill_panes_by_tag(
        lambda tag: tag == prefix or tag.startswith(prefix + os.sep))


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


def _reap_container_panes(project_root: Path, group: str, task_id: str) -> list[str]:
    """이 task 의 container_dir 태그를 가진 pane 만 회수한다(task 스코프)."""
    from . import paths, container_registry
    cp = paths.container_paths(project_root, group, task_id)
    container_dir = str(cp["container_dir"].resolve())
    container_registry.clear(project_root, group, task_id)
    # tail/watcher pane 은 container_dir 로, analyze pane 은 그 하위 findings 경로로
    # 태그된다 — 둘 다 회수하되 prefix 가 task 의 container_dir 라 스코프는 유지된다.
    return _kill_panes_by_tag(
        lambda tag: tag == container_dir or tag.startswith(container_dir + os.sep))


def _kill_panes_by_tag(match: Callable[[str], bool]) -> list[str]:
    """`@okstra_container_run` 태그값이 ``match`` 를 통과하는 pane 만 tmux 로 kill.

    태그값은 해당 task 의 container_dir 절대경로다. 무조건 글로벌 sweep 은 동시
    세션 간섭을 일으키므로 호출자가 항상 스코프 predicate 를 준다(I2)."""
    from . import tmux
    if not tmux.tmux_available():
        return []
    result = tmux.run_tmux(
        ["list-panes", "-a", "-F", "#{pane_id}\t#{@okstra_container_run}"])
    if result.returncode != 0:
        return []
    killed: list[str] = []
    for line in result.stdout.splitlines():
        parts = line.split("\t")
        if len(parts) == 2 and parts[0] and parts[1] and match(parts[1]):
            tmux.kill_pane(parts[0])
            killed.append(parts[0])
    return killed


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

def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(prog="okstra container")
    sub = parser.add_subparsers(dest="command", required=True)
    for name in ("up", "status", "logs", "stop-watcher", "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")
        if name == "logs":
            sp.add_argument("--service", default="")
    watch = sub.add_parser("watch")  # watcher pane 내부에서 도는 장기 루프
    watch.add_argument("--project-name", required=True)
    watch.add_argument("--service", required=True)
    watch.add_argument("--findings", required=True)
    watch.add_argument("--scan-interval", type=float, default=5.0)
    args = parser.parse_args(argv)
    if args.command == "watch":
        run_service_watcher(
            project_name=args.project_name, service=args.service,
            findings_path=args.findings, scan_interval_seconds=args.scan_interval)
        return 0
    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", "logs", "stop-watcher", "down"}:
        raise ValueError("unknown container text purpose")
    rows = [f"Okstra container {command}\n", line("Status", "ready")]
    for label, key in (("Project name", "projectName"), ("Watchers dir", "watchersDir"),
                       ("Note", "note")):
        if key in payload:
            rows.append(line(label, payload.get(key)))
    if command == "up":
        rows.extend(_render_container_up(payload))
    elif command in {"status", "logs"}:
        rows.extend(_render_container_watchers(payload))
        if command == "status":
            rows.extend(_render_container_status(payload))
    elif command == "stop-watcher":
        rows.extend(_render_reaped_panes("Reaped pane", payload.get("reapedPanes")))
    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))
    watch = payload.get("watch") if isinstance(payload.get("watch"), dict) else {}
    for label, key in (("Watcher enabled", "enabled"), ("Watcher session", "session"),
                       ("Watcher note", "note")):
        if key in watch:
            rows.append(line(label, watch.get(key)))
    for index, service in enumerate(watch.get("services") or [], 1):
        rows.append(line(f"Watcher service {index} name", service))
    return rows


def _render_container_watchers(payload: dict) -> list[str]:
    watchers = payload.get("watchers") if isinstance(payload.get("watchers"), dict) else {}
    rows: list[str] = []
    for index, (service, metadata) in enumerate(sorted(watchers.items()), 1):
        rows.append(line(f"Watcher {index} service", service))
        values = metadata if isinstance(metadata, dict) else {}
        for label, key in (("pane ID", "paneId"), ("session", "sessionName"),
                           ("findings path", "findingsPath")):
            if key in values:
                rows.append(line(f"Watcher {index} {label}", values.get(key)))
    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_reaped_panes(prefix: str, raw_panes: object) -> list[str]:
    panes = raw_panes if isinstance(raw_panes, list) else []
    return [line(f"{prefix} {index}", pane) for index, pane in enumerate(panes, 1)]


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")))
        rows.extend(_render_reaped_panes(
            f"Downed {index} reaped pane", values.get("reapedPanes")
        ))
    rows.extend(_render_reaped_panes("Orphan pane", payload.get("orphanPanesReaped")))
    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)
    if args.command == "logs":
        return logs_container_group(
            project_root=pr, task_key=args.task_key, service=args.service)
    if args.command == "stop-watcher":
        return stop_watcher_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:]))