"""Contextra Stream Manager — 다중 스트림 인프로세스 관리.

각 스트림 = det+pose+seg 3파이프라인 그룹 (multi_pipeline의 검증된 아키텍처 재사용).
FastAPI 프로세스 내에서 직접 제어: add / remove / start / stop / restart / status.

소스 타입 (source_type):
  rtsp   — uri: "rtsp://..."            (nvurisrcbin, 무한 루프) ✅
  file   — uri: "/path/video.mp4"        (file-loop, 무한 반복) ✅
  onvif  — host/port/username/password (+profile_token) → Media.GetStreamUri로 RTSP 발견 ✅
  hls    — 미지원 (nvurisrcbin이 http 스킴 불가 — 실측 확인, 커스텀 체인 필요) ❌
  webrtc — 미지원 (webrtcbin 존재하나 capture 경로 미구현) ❌

설정 모델:
  - 베이스   : STREAMS_CONFIG_PATH (server-local, read-only)
  - 오버레이 : STREAMS_RUNTIME_PATH (server-local Runtime state, writable)
               API로 추가(added)/제거(removed)한 스트림만 기록.
  - 유효 세트 = 베이스 − removed + added. 기동 시 복구만 하며 명시 Start를 기다린다.
  → read-only Base는 쓰지 않는다 (API 변경은 오버레이에 영속).

스레드 모델:
  - StreamGroup._persist_loop : 병합 프레임 PG/Redis 기록 (5s 주기, 기존 main과 동일)
  - run_pipeline()의 worker 스레드 x3 (det/pose/seg) — pyservicemaker Pipeline.start().wait()
"""
import os
import json
import threading
import time
import traceback
from copy import deepcopy
from functools import wraps
from pathlib import Path
from app.stream_config_store import ConfigPersistenceError, StreamConfigStore
from app.start_diagnostics import StartDiagnostics, setup_failure_summary

import yaml

# CHANNELPROC-01: API imports must not create SDK/GPU state before spawn.
def ResultCollector(*args, **kwargs):
    from app.multi_pipeline import ResultCollector as factory
    return factory(*args, **kwargs)


def StorageWriter(*args, **kwargs):
    from app.multi_pipeline import StorageWriter as factory
    return factory(*args, **kwargs)


def run_pipeline(*args, **kwargs):
    from app.multi_pipeline import run_pipeline as run
    return run(*args, **kwargs)

from app.inference_profiles import ProfileError, resolve_from_spec, render_cfg, profile_summary, inference_component_id
from app.runtime_defaults import ACTIVE_LEARNING_DEFAULT, STREAM_AUTOSTART_DEFAULT, STREAM_START_DEFAULT
from app.source_adapters.rtsp import RtspSourceAdapter, RtspSourceOptions
from app.stream_lifecycle import (
    StreamFailure,
    StreamLifecycle,
    make_stream_failure,
    stop_role_handles,
)

# server-local read-only Base 설정.
BASE_CONFIG_PATH = os.environ.get("STREAMS_CONFIG_PATH", "/etc/contextra/streams.yaml")
# API add/remove 영속화 오버레이 — 반드시 writable 디렉터리.
RUNTIME_OVERLAY_PATH = os.environ.get(
    "STREAMS_RUNTIME_PATH",
    "/var/lib/contextra/runtime/streams_runtime.yaml",
)


def _log(msg: str):
    print(f"[STREAM-MGR] {msg}", flush=True)


# ---------------------------------------------------------------------------
# 소스 리졸버 — source_type -> GStreamer URI (+ 메타)
# ---------------------------------------------------------------------------

class SourceError(Exception):
    pass


def resolve_source(spec: dict) -> dict:
    """streams.yaml 항목을 (uri, meta)로 해석. 실패 시 SourceError.

    반환: {"uri": str, "source_type": str, "resolved_from": str|None}
      - onvif: uri=발견된 RTSP URI, resolved_from="onvif:<host>"
      - file : 절대경로가 아니면 SourceError (컨테이너 내 경로)
    """
    st = (spec.get("source_type") or "rtsp").lower()

    if st == "rtsp":
        uri = spec.get("uri") or ""
        if not uri.startswith("rtsp://"):
            raise SourceError("rtsp source requires a valid RTSP URI")
        return {"uri": uri, "source_type": "rtsp", "resolved_from": None}

    if st == "hls":
        uri = spec.get("uri") or ""
        if not (uri.startswith("http://") or uri.startswith("https://")):
            raise SourceError(f"hls source requires http(s):// uri, got: {uri!r}")
        # 2026-08-17 실측: nvurisrcbin은 rtsp/file 스킴만 지원 (http m3u8 →
        # "Internal data stream error"). HLS는 uridecodebin 커스텀 체인 필요 — 미구현.
        raise SourceError(
            "HLS source not supported yet: container capture uses nvurisrcbin which "
            "only accepts rtsp:// and file:// (verified by probe). HLS requires a "
            "custom uridecodebin chain — tracked as future work.")

    if st == "webrtc":
        uri = spec.get("uri") or ""
        if not uri:
            raise SourceError("webrtc source requires 'uri' (webrtc://...)")
        # webrtcbin은 컨테이너에 존재하나 pyservicemaker capture 경로(nvurisrcbin)가
        # WebRTC를 지원하지 않음 — 커스텀 체인(webrtcbin+rtph264depay+decodebin) 필요.
        raise SourceError(
            "WebRTC source not supported yet: webrtcbin element exists but the "
            "capture path (nvurisrcbin) has no WebRTC support — requires a custom "
            "webrtcbin chain, tracked as future work.")

    if st == "file":
        path = spec.get("uri") or ""
        if not os.path.isabs(path):
            raise SourceError(f"file source requires absolute container path, got: {path!r}")
        if not os.path.exists(path):
            raise SourceError(f"file not found in container: {path}")
        return {"uri": f"file://{path}", "source_type": "file", "resolved_from": None}

    if st == "onvif":
        host = spec.get("host") or ""
        if not host:
            raise SourceError("onvif source requires 'host'")
        from app.onvif_discovery import resolve_onvif_stream, OnvifError
        try:
            res = resolve_onvif_stream(
                host=host,
                port=int(spec.get("port") or 80),
                username=spec.get("username"),
                password=spec.get("password"),
                profile_token=spec.get("profile_token"),
                timeout=float(spec.get("timeout") or 5.0),
            )
        except OnvifError as e:
            raise SourceError(f"ONVIF discovery failed for {host}: {e}") from e
        return {"uri": res["stream_uri"], "source_type": "onvif",
                "resolved_from": f"onvif:{host} profile={res['profile_token']}"}

    raise SourceError(f"unknown source_type: {st!r}")


# ---------------------------------------------------------------------------
# StreamGroup — 1 스트림의 3파이프라인 + 저장 + 상태
# ---------------------------------------------------------------------------

class StreamGroup:
    """한 스트림 = det/pose/seg 파이프라인 그룹.

    요청 상태(desired)와 실제 Pipeline 상태(observed)를 독립적으로 관리한다.
    """

    FIRST_FRAME_TIMEOUT_S = 60.0
    CLEANUP_TIMEOUT_S = 60.0

    def __init__(self, spec: dict, *, metadata_publisher=None):
        self.stream_id = str(spec["stream_id"])
        self.site_id = spec.get("site_id", "SITE01")
        self.spec = dict(spec)
        self.source_type = (spec.get("source_type") or "rtsp").lower()

        self.lifecycle = StreamLifecycle()
        self.started_at: float | None = None
        self.frames_merged = 0
        self.profile = None             # Legacy single-provider StreamProfile
        self.provider_plan = None        # PR13 multi-provider plan
        self.scenario_runner = None
        self._lock = threading.Lock()
        self._teardown_lock = threading.RLock()
        self._pending_restart = None
        self._cleanup_deadline = None
        self._last_cleanup = None
        self.CLEANUP_TIMEOUT_S = float(os.environ.get("CONTEXTRA_CLEANUP_TIMEOUT_S", "60"))
        if not 0 < self.CLEANUP_TIMEOUT_S < float("inf"):
            raise ValueError("Invalid cleanup timeout.")
        self._cleanup_started = False
        self._cleanup_done = threading.Event()
        self._cleanup_done.set()
        self._cleanup_failure = None
        self._cleanup_thread = None
        self._run_stop_event = threading.Event()
        self._bootstrap_thread: threading.Thread | None = None
        self._supervisor_thread: threading.Thread | None = None

        self.collector: ResultCollector | None = None
        self.writer: StorageWriter | None = None
        self.al_writer = None
        self.gate = None   # EventGate (persist_mode=event_gated 시)
        self.tracker = None
        self.result_path = None
        self.metadata_publisher = metadata_publisher  # process owner가 주입; Stream cleanup은 close하지 않음.
        self.threads: list[threading.Thread] = []
        self.pipes: dict[str, object] = {}   # task -> Pipeline (stop용)
        self.role_handles: dict[str, object] = {}
        self._validate_det_config(self.spec)

    @staticmethod
    def _validate_det_config(spec):
        from app.pose_config import execution_plan
        execution_plan(spec)
        from app.intrusion import IntrusionConfig
        intrusion = IntrusionConfig.parse(spec.get('intrusion', {}))
        if str(spec.get('scenarios') or 'off').lower() not in ('all', 'off'):
            raise ValueError('Legacy scenario selection is retired; use intrusion settings.')
        if intrusion.enabled and (not spec.get('server_id') or not spec.get('source_id')):
            raise ValueError('Canonical identity is required for INTRUSION.')
        from app.analysis_metrics import analysis_fps
        analysis_fps(spec.get('analysis_fps_target', 10))
        from app.stream_settings import display_score, tracking_settings
        display_score(spec, .5)
        tracking_settings(spec)
        if spec.get('frame_time_anchor') is not None:
            from app.frame_time import FrameAnchor
            FrameAnchor.model_validate(spec['frame_time_anchor'])
            if not spec.get('server_id') or not spec.get('source_id'):
                raise ValueError('Canonical Frame identity is required for time settings.')
        if 'server_id' not in spec:
            return
        from app.frame_ingress import FrameIngress
        from app.byte_tracking import TrackingSettings
        FrameIngress(spec.get('server_id'), spec.get('source_id'), str(spec['stream_id']),
                     'validation-only', 1, spec.get('web_config_version'),
                     frame_time_anchor=spec.get('frame_time_anchor'))
        TrackingSettings(**(spec.get('tracking') or {}))

    @property
    def state(self) -> str:
        return self.lifecycle.snapshot()["state"]

    @property
    def desired_state(self) -> str:
        return self.lifecycle.snapshot()["desired_state"]

    @property
    def observed_state(self) -> str:
        return self.lifecycle.snapshot()["observed_state"]

    @property
    def run_generation(self) -> int:
        return int(self.lifecycle.snapshot()["run_generation"])

    @property
    def last_error(self) -> dict | None:
        return self.lifecycle.snapshot()["last_error"]

    # -- lifecycle ----------------------------------------------------------

    def start(self, *, assigned_run_id=None, assigned_generation=None):
        # 이전 generation cleanup과 새 generation resource 할당이 겹치지 않게 직렬화한다.
        with self._teardown_lock:
            with self._lock:
                status = self.lifecycle.snapshot()
                if self._pending_restart is not None:
                    return status["run_generation"]
                if (
                    status["desired_state"] == "running"
                    and status["observed_state"] in ("starting", "running")
                ):
                    return status["run_generation"]
                if not self._cleanup_done.is_set() or self._cleanup_failure is not None:
                    raise RuntimeError("Stream cleanup is incomplete.")
                if self.role_handles or (self._bootstrap_thread and self._bootstrap_thread.is_alive()):
                    raise RuntimeError("Previous Stream resources are still owned.")
                self._cleanup_started = False
                generation = self.lifecycle.begin(assigned_run_id=assigned_run_id,
                                                   assigned_generation=assigned_generation)
                stop_event = threading.Event()
                self._run_stop_event = stop_event
                self.started_at = time.time()
                bootstrap = threading.Thread(
                    target=self._begin,
                    args=(generation, stop_event),
                    name=f"bootstrap-{self.stream_id}-{generation}",
                    daemon=True,
                )
                self._bootstrap_thread = bootstrap
                try:
                    bootstrap.start()
                except Exception:
                    self.lifecycle.signal_error(generation, make_stream_failure(
                        "PIPELINE_INIT_FAILED", "bootstrap_thread"))
                return generation

    def _single_decode_enabled(self):
        value = os.environ.get('CONTEXTRA_SINGLE_DECODE_MULTIPROVIDER', 'true')
        if value not in ('true', 'false'):
            raise ValueError('Invalid CONTEXTRA_SINGLE_DECODE_MULTIPROVIDER.')
        return value == 'true'

    def _begin(self, generation: int, stop_event: threading.Event):
        diagnostic = StartDiagnostics("bootstrap")
        setup_stage = "source"
        try:
            with diagnostic.span("source"):
                resolved = resolve_source(self.spec)
            uri, stype = resolved["uri"], resolved["source_type"]
            if not self._generation_active(generation, stop_event):
                return

            rtsp_options = None
            if stype == "rtsp":
                try:
                    rtsp_options = RtspSourceOptions.model_validate(
                        self.spec.get("source_options") or {}
                    )
                except Exception as error:
                    self._fail_run(
                        generation,
                        stop_event,
                        make_stream_failure(
                            "CONFIG_INVALID", "source_options_validation", None, error
                        ),
                    )
                    return
            first_frame_timeout_s = (
                float(rtsp_options.first_frame_timeout_s)
                if rtsp_options is not None
                else self.FIRST_FRAME_TIMEOUT_S
            )

            single_decode = self._single_decode_enabled()
            setup_stage = "profile"
            if single_decode:
                from app.provider_plan import resolve as resolve_providers
                from app.provider_delivery import runtime_plan as make_provider_plan
                with diagnostic.span("profile"):
                    provider_plan = make_provider_plan(self.spec)
                    unavailable = set(resolve_providers(provider_plan))
                self.provider_plan = provider_plan
                self.profile = None
                active_providers = tuple(
                    p for p in provider_plan.providers if p.key not in unavailable)
                dependencies = {
                    scenario: provider.key
                    for provider in provider_plan.providers
                    for scenario in provider.scenarios
                }
                if (not active_providers or
                        (provider_plan.requested_scenarios
                         and all(dependencies[s] in unavailable
                                 for s in provider_plan.requested_scenarios))):
                    raise ProfileError(
                        self.stream_id + ': no READY provider for requested scenarios')
                setup_stage = "config"
                with diagnostic.span("config"):
                    role_cfgs = {
                        p.key: render_cfg(
                            self.stream_id, p.role, p.model, component_key=p.key)
                        for p in active_providers
                    }
                    component_ids = {
                        p.key: inference_component_id(self.stream_id, p.key)
                        for p in active_providers
                    }
                # Provider health/degradation is coordinator-owned. Lifecycle
                # only supervises the one shared source/decode pipeline.
                if not self.lifecycle.configure_roles(
                    generation, required_roles={'shared'}, optional_roles=set()):
                    return
                person_task = ('pose' if provider_plan.person_owner == 'pose'
                               else 'det' if provider_plan.person_owner == 'medium'
                               else 'fire')
                collector = ResultCollector(required_roles=[person_task])
                result_path = self._prepare_multi_results(
                    provider_plan, generation, stop_event, collector)
                for key in unavailable:
                    result_path.mark_provider_unavailable(key, 'provider_unavailable')
                tracker = result_path.tracker
            else:
                # Stable rollback path: one provider/source pipeline per historical contract.
                with diagnostic.span("profile"):
                    prof = resolve_from_spec(self.stream_id, self.spec)
                setup_stage = "config"
                with diagnostic.span("config"):
                    role_cfgs = {
                        _role: render_cfg(
                            self.stream_id, _role, prof.resolved_models[_role])
                        for _role in prof.roles
                    }
                if not role_cfgs:
                    raise ValueError("stream requires at least one inference role")
                optional_roles = {str(role) for role in (self.spec.get("optional_roles") or [])}
                unknown_optional = optional_roles - set(prof.roles)
                required_roles = set(prof.roles) - optional_roles
                if unknown_optional or not required_roles:
                    raise ValueError("optional_roles must be a proper subset of active roles")
                if not self.lifecycle.configure_roles(
                    generation,
                    required_roles=required_roles,
                    optional_roles=optional_roles,
                ):
                    return
                self.profile = prof
                self.provider_plan = None
                setup_stage = "result_path"
                collector = ResultCollector(required_roles=list(required_roles))
                result_path = self._prepare_det_results(
                    prof, generation, stop_event, collector)
                if result_path is not None:
                    tracker = result_path.tracker
                else:
                    from app.tracker import IoUTracker
                    tracker = IoUTracker()
            self.tracker = tracker

            scenario_runner = None
            # INTRUSION-01: decisions run on canonical Frames, never storage/legacy engines.

            al_writer = None
            if self.spec.get("active_learning", ACTIVE_LEARNING_DEFAULT):
                try:
                    from app.active_learning import ActiveLearningWriter
                    al_writer = ActiveLearningWriter(site_id=self.site_id,
                                                     stream_id=self.stream_id, enabled=True)
                except Exception as ale:
                    _log(f"{self.stream_id}: active_learning disabled (non-fatal): {ale}")

            # 이벤트 게이트 영속화 — persist_mode=event_gated 시 프레임은 메모리만,
            # 이벤트 발생 시 ±윈도우(전 10s/후 5s)만 PG에 플러시.
            gate = None
            if str(self.spec.get("persist_mode") or "all").lower() == "event_gated":
                try:
                    from app.event_gate import EventGate
                    from app.metadata_storage import storage as _storage
                    gate = EventGate(_storage, stream_id=self.stream_id, site_id=self.site_id)
                    gate.start()
                except Exception as ge:
                    _log(f"{self.stream_id}: event_gated disabled (non-fatal): {ge}")

            setup_stage = "storage_writer"
            writer = StorageWriter(
                site_id=self.site_id, stream_id=self.stream_id,
                store=bool(self.spec.get("store", True)),
                persist_every=int(self.spec.get("persist_every") or 1),
                scenario_runner=scenario_runner, al_writer=al_writer, gate=gate,
                required_roles=(['pose'] if single_decode and self.provider_plan.person_owner == 'pose'
                                else ['det'] if single_decode and self.provider_plan.person_owner == 'medium'
                                else ['fire'] if single_decode else list(required_roles)),
                tracker=tracker)

            self.collector = collector
            self.writer = writer
            self.al_writer = al_writer
            self.gate = gate
            # Phase 3 — 시나리오 가시성: status()에서 활성/비활성 엔진 노출.
            self.scenario_runner = scenario_runner
            if single_decode:
                setup_stage = "single_decode_worker"
                from app.single_decode_pipeline import run_single_decode_pipeline
                runner = run_single_decode_pipeline
                if getattr(result_path, 'linear_trial', False):
                    from app.provider_linear_pipeline import run_provider_linear_pipeline
                    runner = run_provider_linear_pipeline
                handle = runner(
                    self.stream_id, active_providers, role_cfgs, component_ids,
                    uri, result_path, pipes=self.pipes,
                    run_generation=generation, on_role_update=self._on_role_update,
                    is_current=lambda: self._generation_active(generation, stop_event),
                    source_adapter=(
                        RtspSourceAdapter(
                            uri, rtsp_options,
                            gpu_id=int(os.environ.get("GPU_ID", "0")))
                        if rtsp_options is not None else None))
                with self._lock:
                    self.role_handles['shared'] = handle
            else:
                # Per-stream cfg legacy rollback path.
                for role in ("det", "pose", "seg"):
                    if role not in role_cfgs or not self._generation_active(generation, stop_event):
                        continue
                    setup_stage = "role_worker_" + role
                    handle = run_pipeline(
                        role, role_cfgs[role], uri, collector, 0,
                        al_writer if role == "det" else None,
                        pipes=self.pipes, score_threshold=prof.score_threshold,
                        tracker=tracker if role == "det" else None,
                        result_path=result_path if role in ('det','pose') else None,
                        run_generation=generation,
                        on_role_update=self._on_role_update,
                        component_id=inference_component_id(self.stream_id, role),
                        is_current=lambda: self._generation_active(generation, stop_event),
                        drop_on_stop=(role == "det" and len(prof.roles) == 1
                                      and not self.spec.get("store", True)
                                      and not self.spec.get("active_learning", True)
                                      and self.spec.get("scenarios") == "off"),
                        source_adapter=(
                            RtspSourceAdapter(
                                uri, rtsp_options,
                                gpu_id=int(os.environ.get("GPU_ID", "0")))
                            if rtsp_options is not None else None))
                    with self._lock:
                        self.role_handles[role] = handle

            if not self._generation_active(generation, stop_event):
                self._teardown_pipelines(generation, wait=False)
                return

            setup_stage = "persist_thread"
            t = threading.Thread(
                target=self._persist_loop,
                args=(generation, stop_event, collector, writer),
                name=f"persist-{self.stream_id}-{generation}",
                daemon=True,
            )
            self.threads.append(t)
            t.start()

            setup_stage = "supervisor_thread"
            supervisor = threading.Thread(
                target=self._supervise,
                args=(generation, stop_event, first_frame_timeout_s),
                name=f"supervisor-{self.stream_id}-{generation}",
                daemon=True,
            )
            self._supervisor_thread = supervisor
            supervisor.start()
            _log(f"{self.stream_id}: STARTING type={stype} generation={generation}")
        except SourceError as error:
            self._fail_run(
                generation,
                stop_event,
                make_stream_failure("SOURCE_CONNECT_FAILED", "source_resolve", None, error),
            )
        except ProfileError as error:
            message = str(error).lower()
            dependency_failure = ("no ready model" in message
                                  or "no ready provider" in message)
            self._fail_run(
                generation,
                stop_event,
                make_stream_failure(
                    "DEPENDENCY_NOT_READY" if dependency_failure else "CONFIG_INVALID",
                    "model_readiness" if dependency_failure else "profile_validation",
                    None,
                    error,
                ),
            )
        except Exception as error:
            # One bounded record per failed bootstrap; public error stays allowlisted.
            try:
                _log("SETUP_DIAGNOSTIC " + json.dumps({
                    "stream_id": self.stream_id, "generation": generation,
                    "setup_stage": setup_stage, "exceptions": setup_failure_summary(error),
                }, ensure_ascii=True))
            finally:
                self._fail_run(
                    generation,
                    stop_event,
                    make_stream_failure("CONFIG_INVALID", "stream_setup", None, error),
                )
        finally:
            diagnostic.close()

    def _prepare_multi_results(self, provider_plan, generation, stop_event, collector):
        from app.frame_ingress import FrameIngress, require_identity_binding
        from app.multi_provider_results import MultiProviderCoordinator
        from app.provider_linear_results import linear_trial_enabled, LinearProviderResults
        ingress = FrameIngress(
            self.spec.get('server_id'), self.spec.get('source_id'), self.stream_id,
            self.lifecycle.snapshot()['run_id'], generation,
            self.spec.get('web_config_version'),
            frame_time_anchor=self.spec.get('frame_time_anchor'),
            analysis_fps_target=self.spec.get('analysis_fps_target', 10))
        require_identity_binding()
        ingress.track_result_gaps = True
        coordinator_type = LinearProviderResults if linear_trial_enabled() else MultiProviderCoordinator
        coordinator = coordinator_type(
            self.spec, ingress, provider_plan, collector,
            lambda: self._generation_active(generation, stop_event),
            lambda: self._fail_run(
                generation, stop_event,
                make_stream_failure(
                    'ROLE_RUNTIME_FAILED', 'multi_provider_results', 'shared')),
            publisher=self.metadata_publisher,
            on_provider_frame=lambda role, at: self._on_role_update(
                generation, role, 'frame', at))
        self.result_path = coordinator
        try:
            coordinator.start()
        except Exception:
            coordinator.close(1)
            raise
        return coordinator

    def _prepare_det_results(self, profile, generation, stop_event, collector):
        # IDIMPL/PIPECORE: 기존 API에는 가짜 설치 identity를 보충하지 않는다.
        configured = 'server_id' in self.spec
        if not configured:
            return None
        if set(profile.roles) not in ({'det'}, {'pose'}):
            raise ValueError('Canonical Det/Pose join is not implemented yet.')
        from app.frame_ingress import FrameIngress, require_identity_binding
        from app.byte_tracking import ByteTracker, TrackingSettings
        from app.det_result_path import DetResultPath
        from app.stream_settings import display_score
        from app.region_scenario import compose as compose_region_scenario
        from app.stream_settings import revision
        from app.stream_settings import visual_options
        from app.fall_runtime import FallSettings, FallScenario
        from app.fight_runtime import FightSettings, FightScenario
        from app.crowd_config import CrowdConfig
        from app.crowd_scenario import Crowd
        from app.crowd_result import CrowdResultProcessor
        region = compose_region_scenario(self.spec)
        fall_config = FallSettings.parse(self.spec.get('fall') or {})
        fall = FallScenario(fall_config) if fall_config.enabled else None
        fight_config = FightSettings.parse(self.spec.get('fight') or {})
        fight = FightScenario(fight_config) if fight_config.enabled else None
        crowd_config = CrowdConfig.parse(self.spec.get('crowd') or {})
        crowd = CrowdResultProcessor(Crowd(crowd_config), camera_id=self.spec.get('camera_id'),
                    scenario_revision=revision(self.spec)) if crowd_config.enabled else None
        ingress = FrameIngress(self.spec.get('server_id'), self.spec.get('source_id'),
                               self.stream_id, self.lifecycle.snapshot()['run_id'], generation,
                               self.spec.get('web_config_version'),
                               frame_time_anchor=self.spec.get('frame_time_anchor'),
                               analysis_fps_target=self.spec.get('analysis_fps_target', 10))
        require_identity_binding()
        ingress.track_result_gaps = fall is not None or fight is not None or crowd is not None
        tracking = dict(self.spec.get('tracking') or {})
        tracking['frame_rate'] = ingress.sampler.fps
        tracking_config = TrackingSettings.parse(tracking)
        tracker = ByteTracker(tracking_config, metrics=ingress.metrics) if tracking_config.enabled else None
        from app.pose_config import PoseConfig
        from app.fire_config import build_engine as build_fire_engine
        role = profile.roles[0]
        pose_config = PoseConfig.parse(self.spec.get('pose',{})) if role=='pose' else None
        path = DetResultPath(ingress, tracker, profile.resolved_models[role], profile.score_threshold,
                             collector, lambda: self._generation_active(generation, stop_event),
                             lambda: self._fail_run(generation, stop_event, make_stream_failure(
                                 'ROLE_RUNTIME_FAILED', 'canonical_results', role)),
                             publisher=self.metadata_publisher,
                             display_threshold=display_score(self.spec, profile.score_threshold),
                             intrusion=region, fire=build_fire_engine(self.spec), fall=fall, fight=fight, crowd=crowd,
                             scenario_revision=revision(self.spec),
                             display_categories=visual_options(self.spec)[0],
                             debug_all_candidates=visual_options(self.spec)[1], pose_config=pose_config,
                             camera_id=self.spec.get('camera_id'))
        from app.event_cooldown import EventCooldown
        path.event_cooldown = EventCooldown(overrides=self.spec.get('event_cooldowns'))
        engines = getattr(region, 'engines', (region,))
        for engine in (*engines, fall.engine if fall else None, fight.engine if fight else None,
                       path.fire, crowd.engine if crowd else None):
            if engine is not None:
                engine.event_admission = path.event_cooldown.allow
        from app.metadata_visual import MetadataVisual
        path.visual = MetadataVisual.from_environment(display_threshold=path.display_threshold)
        if path.visual is not None and pose_config is not None:
            path.visual.pts_interval_s = pose_config.capture_interval_s
        self.result_path = path
        try:
            from app.event_capture import EventCapture
            path.artifacts = EventCapture.from_environment(self.spec)
            path.start()
        except Exception:
            path.close(1)
            raise
        return path

    def stop(self):
        with self._teardown_lock:
            self._pending_restart = None
            generation = self.lifecycle.request_stop()
            self._teardown_pipelines(generation, wait=False)

    def check_restart(self, candidate):
        self._validate_det_config(candidate)
        if self._cleanup_failure is not None:
            raise RuntimeError("Stream cleanup failed.")
        if self._pending_restart is not None and self._pending_restart != candidate:
            raise RuntimeError("Another configuration is pending.")

    def restart(self, candidate=None):
        with self._teardown_lock:
            target = deepcopy(self.spec if candidate is None else candidate)
            self.check_restart(target)
            if self._pending_restart is not None:
                return
            if (self._cleanup_done.is_set() and not self.role_handles
                    and not (self._bootstrap_thread and self._bootstrap_thread.is_alive())):
                self._apply_spec(target)
                self.start()
                return
            self._pending_restart = target
            generation = self.lifecycle.request_stop()
            self._teardown_pipelines(generation, wait=False)

    def _apply_spec(self, candidate):
        self.spec = candidate
        self.site_id = candidate.get("site_id")
        self.source_type = candidate.get("source_type", "rtsp")

    def _cleanup_expired(self, generation):
        # Native call은 강제 중단하지 않는다. 기한 초과는 공개하고 소유 자원을 보존한다.
        with self._lock:
            if generation != self.run_generation or self._cleanup_done.is_set():
                return
            self._cleanup_failure = self._cleanup_failure or make_stream_failure(
                "STOP_CLEANUP_FAILED", "cleanup_timeout")
            self.lifecycle.record_cleanup(generation, self._cleanup_failure)

    def _teardown_pipelines(self, generation: int, *, wait=True) -> StreamFailure | None:
        # 단일 owner가 SDK 종료를 맡고 bootstrap/supervisor는 요청 후 반환한다.
        with self._lock:
            if generation != self.lifecycle.run_generation:
                return None
            self._run_stop_event.set()
            if not self._cleanup_started:
                self._cleanup_started = True
                self._cleanup_done.clear()
                self._cleanup_deadline = time.monotonic() + self.CLEANUP_TIMEOUT_S
                self._cleanup_thread = threading.Thread(
                    target=self._cleanup_resources, args=(generation,),
                    name="stream-cleanup", daemon=True)
                try:
                    self._cleanup_thread.start()
                except Exception:
                    self._cleanup_failure = make_stream_failure('STOP_CLEANUP_FAILED', 'cleanup_start')
                    self.lifecycle.record_cleanup(generation, self._cleanup_failure)
                    self._cleanup_done.set()
        if wait:
            self._cleanup_done.wait(max(0, self._cleanup_deadline - time.monotonic()))
        with self._lock:
            return self._cleanup_failure

    def _retain_cleanup_failure(self, generation, failure):
        if failure is None:
            return
        with self._lock:
            self._cleanup_failure = self._cleanup_failure or failure
        self.lifecycle.record_cleanup(generation, failure)

    def _cleanup_resources(self, generation):
        deadline = self._cleanup_deadline
        timer = threading.Timer(max(0, deadline - time.monotonic()), self._cleanup_expired, args=(generation,))
        timer.daemon = True
        timer.start()
        try:
            bootstrap = self._bootstrap_thread
            if bootstrap is not None:
                bootstrap.join(timeout=max(0, deadline - time.monotonic()))
                if bootstrap.is_alive():
                    self._retain_cleanup_failure(generation, make_stream_failure(
                        "STOP_CLEANUP_FAILED", "bootstrap_join"))
                    return
            with self._lock:
                handles = dict(self.role_handles)
                threads = list(self.threads)
                supervisor = self._supervisor_thread
            if self.result_path is not None and hasattr(self.result_path, 'stop_admission'):
                self.result_path.stop_admission()
            self._retain_cleanup_failure(generation, stop_role_handles(handles, deadline=deadline))
            if self.result_path is not None:
                try:
                    self.result_path.close(max(0, deadline - time.monotonic()))
                except Exception as error:
                    self._retain_cleanup_failure(generation, make_stream_failure(
                        'STOP_CLEANUP_FAILED', 'det_results_join', 'det', error))
            for thread in threads + ([supervisor] if supervisor else []):
                thread.join(timeout=max(0, deadline - time.monotonic()))
                if thread.is_alive():
                    self._retain_cleanup_failure(generation, make_stream_failure(
                        "STOP_CLEANUP_FAILED", "control_thread_join"))
            if time.monotonic() >= deadline:
                self._cleanup_expired(generation)
            if self._cleanup_failure is not None:
                return
            for resource, method, stage in (
                (self.writer, "_flush", "writer_flush"),
                (self.gate, "close", "event_gate_close"),
                (self.al_writer, "close", "active_learning_close"),
                (self.tracker, "reset", "tracker_reset"),
            ):
                if resource is None or not hasattr(resource, method):
                    continue
                try:
                    getattr(resource, method)()
                except Exception as error:
                    self._retain_cleanup_failure(generation, make_stream_failure(
                        "STOP_CLEANUP_FAILED", stage, None, error))
            if time.monotonic() >= deadline:
                self._cleanup_expired(generation)
            if self._cleanup_failure is not None:
                return
            if self.collector is not None:
                with self.collector.lock:
                    self.collector.frames.clear()
                    self.collector.completed_tasks.clear()
                    self.collector.per_task.clear()

            with self._lock:
                if (generation == self.lifecycle.snapshot()["run_generation"]
                        and self._cleanup_failure is None):
                    self.pipes.clear()
                    self.role_handles = {}
                    self.threads = []
                    self._bootstrap_thread = None
                    self._supervisor_thread = None
                    self.collector = None
                    self.writer = None
                    self.al_writer = None
                    self.gate = None
                    self.tracker = None
                    self.result_path = None
                    self.provider_plan = None
            # 새 bootstrap 전에 cleanup의 임시 사본도 SDK Pipeline 참조를 놓는다.
            handles.clear()
        except Exception:
            self._retain_cleanup_failure(generation, make_stream_failure(
                "STOP_CLEANUP_FAILED", "cleanup_worker"))
        finally:
            timer.cancel()
            with self._teardown_lock:
                with self._lock:
                    if time.monotonic() >= deadline:
                        self._cleanup_failure = self._cleanup_failure or make_stream_failure(
                            "STOP_CLEANUP_FAILED", "cleanup_timeout")
                    if self.lifecycle.snapshot()["desired_state"] == "stopped":
                        self.lifecycle.finish_stop(generation, self._cleanup_failure)
                    self._last_cleanup = {
                        "generation": generation,
                        "completed": self._cleanup_failure is None,
                        "finished_monotonic_ns": time.monotonic_ns(),
                        "elapsed_s": max(0, time.monotonic() - (deadline - self.CLEANUP_TIMEOUT_S)),
                    }
                    self.started_at = None
                    self._cleanup_done.set()
                    pending, self._pending_restart = self._pending_restart, None
                if pending is not None and self._cleanup_failure is None:
                    self._apply_spec(pending)
                    self.start()

    def _on_role_update(
        self, generation: int, role: str, event: str, payload: object | None
    ) -> bool:
        if event == "initialized":
            return self.lifecycle.signal_initialized(generation, role)
        if event == "frame":
            frame_time = float(payload) if payload is not None else None
            return self.lifecycle.signal_frame(generation, role, frame_time)
        if event == "source_stage" and isinstance(payload, tuple) and len(payload) == 2:
            stage, stage_time = payload
            return self.lifecycle.signal_source_stage(
                generation, role, str(stage), float(stage_time)
            )
        if event == "error" and isinstance(payload, StreamFailure):
            return self.lifecycle.signal_error(generation, payload)
        if event == "stopped":
            return self.lifecycle.signal_stopped(generation, role)
        return False

    def _supervise(
        self,
        generation: int,
        stop_event: threading.Event,
        first_frame_timeout_s: float | None = None,
    ):
        timeout_s = (
            self.FIRST_FRAME_TIMEOUT_S
            if first_frame_timeout_s is None
            else float(first_frame_timeout_s)
        )
        deadline = time.monotonic() + timeout_s
        while not stop_event.wait(0.05):
            status = self.lifecycle.snapshot()
            if generation != status["run_generation"] or status["desired_state"] != "running":
                return
            if status["observed_state"] == "error":
                self._teardown_pipelines(generation, wait=False)
                return
            if status["observed_state"] == "running":
                continue
            if time.monotonic() >= deadline:
                missing = next(
                    (
                        role
                        for role, role_status in status["roles"].items()
                        if role_status["required"]
                        and role_status["state"] != "first_frame"
                    ),
                    None,
                )
                self._fail_run(
                    generation,
                    stop_event,
                    make_stream_failure(
                        "FIRST_FRAME_TIMEOUT", "first_frame_wait", missing
                    ),
                )
                return

    def _fail_run(
        self,
        generation: int,
        stop_event: threading.Event,
        failure: StreamFailure,
    ):
        if not self.lifecycle.signal_error(generation, failure):
            return
        stop_event.set()
        _log(
            f"{self.stream_id}: FAILED generation={generation} "
            f"code={failure.error_code} stage={failure.failure_stage} "
            f"role={failure.role or '-'}"
        )
        self._teardown_pipelines(generation, wait=False)

    def _generation_active(self, generation: int, stop_event: threading.Event) -> bool:
        return not stop_event.is_set() and self.lifecycle.is_active(generation)

    # -- persist loop -------------------------------------------------------

    def _persist_loop(self, generation, stop_event, collector, writer):
        while not stop_event.wait(5):
            if not self._generation_active(generation, stop_event):
                break
            try:
                n = writer.persist_new(collector)
                if n:
                    self.frames_merged += n
            except Exception as we:
                _log(f"{self.stream_id}: persist failed (non-fatal): {type(we).__name__}")

    # -- status -------------------------------------------------------------

    def status(self, detail: bool = False) -> dict:
        lifecycle_status = self.lifecycle.snapshot()
        # Group lock과 SDK/handle lock을 중첩하지 않는다. 진단은 상태를 변경하지 않는다.
        with self._lock:
            diagnostic_handles = list(self.role_handles.values())
        observations = [value for handle in diagnostic_handles
                        if (value := handle.diagnostic_snapshot()) is not None]
        out = {
            "stream_id": self.stream_id,
            "site_id": self.site_id,
            "source_type": self.source_type,
            "state": lifecycle_status["state"],
            "desired_state": lifecycle_status["desired_state"],
            "observed_state": lifecycle_status["observed_state"],
            "run_generation": lifecycle_status["run_generation"],
            "run_id": lifecycle_status["run_id"],
            "roles": lifecycle_status["roles"],
            "last_error": lifecycle_status["last_error"],
            "cleanup_error": lifecycle_status["cleanup_error"],
            "cleanup_pending": not self._cleanup_done.is_set(),
            "restart_pending": self._pending_restart is not None,
            "last_cleanup": dict(self._last_cleanup) if self._last_cleanup else None,
            "started_at": self.started_at,
            "uptime_s": round(time.time() - self.started_at, 1) if self.started_at else None,
            # Phase 1 — 활성 추론 프로파일 (모델/문턱값/강등 상태 가시성).
            "inference_profile": (
                profile_summary(self.profile) if self.profile is not None
                else {'mode': 'single_decode_multi_provider',
                      **self.provider_plan.payload()} if self.provider_plan is not None else None),
            "metadata_path": (self.result_path.status() if self.result_path is not None
                              else {'state': 'NOT_CONNECTED'}),
        }
        # v2 — 시나리오 가시성: 활성화 계획 + 비활성화 사유 포함 (zone_not_defined / missing_capability / disabled_by_profile).
        if observations:
            out["runtime_observation"] = observations
        sr = getattr(self, "scenario_runner", None)
        plan = getattr(sr, "activation_plan", None) if sr else None
        from app.region_scenario import name as region_scenario_name, members
        if self.result_path is not None and hasattr(self.result_path, 'scenario_status'):
            out["scenarios"] = self.result_path.scenario_status()
            active_scenarios = None
        else:
            active_scenarios = []
        if self.result_path is not None and active_scenarios is not None:
            if self.result_path.intrusion is not None:
                active_scenarios.extend(region_scenario_name(c) for c in members(self.result_path.intrusion.config))
            if getattr(self.result_path, 'fire', None) is not None:
                active_scenarios.append('FIRE')
            if getattr(self.result_path, 'fall', None) is not None:
                active_scenarios.append('FALL')
            if getattr(self.result_path, 'fight', None) is not None:
                active_scenarios.append('FIGHT')
            if getattr(self.result_path, 'crowd', None) is not None:
                active_scenarios.append('CROWD')
        if active_scenarios is not None:
            out["scenarios"] = {
                "active": active_scenarios,
                # v2 — 비활성화 시 사유 포함.
                "disabled": [
                    {"name": d.name, "reason": d.reason}
                    for d in plan.decisions if not d.active
                ] if plan is not None else [],
            }
        if detail and self.collector is not None:
            # Phase 3 — 활성 zone 가시성 (드로잉 UX ↔ 이벤트 연결 확인용).
            try:
                from app.zone_store import get_zone_store as _gzs
                out["zones"] = [
                    {"name": z["name"], "version": z["version"]}
                    for z in _gzs().list_zones(stream_id=self.stream_id)
                ]
            except Exception:
                out["zones"] = []
            with self.collector.lock:
                per_task = dict(self.collector.per_task)
                n_frames = len(self.collector.frames)
                latest = None
                for k, f in sorted(self.collector.frames.items(),
                                   key=lambda x: x[1].get("_fn", 0)):
                    if self.collector.required <= set(f.keys()):
                        latest = f
            out["frames_in_window"] = n_frames
            out["per_task_buffers"] = per_task
            if self.gate is not None:
                with self.gate._lock:
                    out["event_gate"] = {
                        "mode": "event_gated",
                        **dict(self.gate.stats),
                        "pending_events": len(self.gate._pending),
                        "buffer_frames": len(self.gate.buffer),
                    }
            if latest is not None:
                # roles 제한 시 존재하는 역할만 요약 (KeyError 방지).
                d = latest.get("det") or {}
                po = latest.get("pose") or {}
                s = latest.get("seg") or {}
                persons = sum(1 for x in d.get("items", []) if x["class_id"] == 0)
                out["latest"] = {
                    "fn": latest.get("_fn"),
                    "det": d.get("count", 0), "persons": persons,
                    "pose_persons": po.get("count", 0), "seg": s.get("count", 0),
                }
        return out


# ---------------------------------------------------------------------------
# StreamManager — 레지스트리 + streams.yaml 영속화
# ---------------------------------------------------------------------------

def _serialized(method):
    @wraps(method)
    def call(self, *args, **kwargs):
        # 초기 A1 제어 명령은 직렬화한다. Frame worker는 이 lock을 사용하지 않는다.
        with self._commands:
            return method(self, *args, **kwargs)
    return call


class StreamManager:
    def __init__(self, base_path=None, overlay_path=None):
        self._groups: dict[str, StreamGroup] = {}
        self._lock = threading.RLock()
        self._commands = threading.RLock()
        # server-local Base는 읽기 전용이다.
        self.base_path = base_path or BASE_CONFIG_PATH
        # 오버레이(writable) — API add/remove만 기록.
        self.overlay_path = overlay_path or RUNTIME_OVERLAY_PATH
        self.config_store = StreamConfigStore(self.overlay_path)
        self.metadata_publisher = None
        self.execution_mode = os.environ.get('CONTEXTRA_STREAM_EXECUTION_MODE', 'thread')
        if self.execution_mode not in ('thread', 'process'):
            raise ValueError('Unknown channel execution mode.')
        self._process_slots = None

    def _slots(self):
        if self._process_slots is None:
            from app.channel_process import ProcessSlots
            raw_limit = os.environ.get('CONTEXTRA_CHANNEL_PROCESS_LIMIT', '0')
            try:
                limit = int(raw_limit)
            except (TypeError, ValueError):
                raise ValueError('Invalid CONTEXTRA_CHANNEL_PROCESS_LIMIT.') from None
            self._process_slots = ProcessSlots(
                limit=limit,
                lock_path=str(Path(self.overlay_path).parent / 'channel-parent.lock'))
        return self._process_slots

    def _new_group(self, spec):
        if self.execution_mode == 'process':
            from app.provider_process_group import ProviderProcessGroup
            return ProviderProcessGroup(spec, slots=self._slots(), metadata_publisher=self.metadata_publisher)
        return StreamGroup(spec, metadata_publisher=self.metadata_publisher)

    def shutdown_channels(self, timeout=70):
        if self.execution_mode != 'process':
            return True
        with self._lock:
            groups = list(self._groups.values())
        for group in groups:
            group.stop()
        deadline = time.monotonic() + timeout
        for group in groups:
            group._cleanup_done.wait(max(0, deadline-time.monotonic()))
        return (all(g._cleanup_done.is_set() and g._process is None for g in groups)
                and (self._process_slots is None or self._process_slots.close()))

    @_serialized
    def configure_metadata(self, publisher):
        if self.metadata_publisher is not None or any(g.state != 'stopped' for g in self._groups.values()):
            raise RuntimeError('Metadata composition must precede Stream execution.')
        if self.execution_mode == 'process':
            self._slots().guard()
        if publisher is not None:
            publisher.start()
        self.metadata_publisher = publisher
        for group in self._groups.values():
            group.metadata_publisher = publisher
            self._register_metadata(group)

    def _register_metadata(self, group):
        if self.metadata_publisher is not None and group.spec.get('server_id') and group.spec.get('source_id'):
            self.metadata_publisher.channel(group.spec['source_id'], None, 0, 'STOPPED')

    def close_metadata(self):
        publisher = self.metadata_publisher
        return publisher is None or publisher.close(1)

    def _saved_spec(self, group):
        return next((s for s in self._load_overlay()['added']
                     if str(s['stream_id']) == group.stream_id), group.spec)

    @_serialized
    def settings(self, sid):
        from app.stream_settings import settings_view
        group = self.get(sid)
        if group is None:
            raise KeyError(sid)
        with group._teardown_lock:
            return dict(configured=settings_view(self._saved_spec(group)),
                        effective=settings_view(group.spec),
                        restart_pending=group._pending_restart is not None)

    @_serialized
    def patch_settings(self, sid, expected_revision, changes):
        from app.stream_settings import patch_settings, revision
        group = self.get(sid)
        if group is None:
            raise KeyError(sid)
        if any(str(s['stream_id']) == sid for s in self._load_base()):
            raise ValueError('Read-only base configuration cannot be changed.')
        with group._teardown_lock:
            current = self._saved_spec(group)
            if revision(current) != expected_revision:
                raise ValueError('Configuration revision conflicts.')
            candidate = patch_settings(current, changes)
            group.check_restart(candidate)
            if current == candidate:
                return self.settings(sid)
            def change(overlay):
                overlay['added'] = [s for s in overlay['added'] if str(s['stream_id']) != sid]
                overlay['added'].append(candidate)
            if group.observed_state == 'stopping' and group._pending_restart is None:
                raise RuntimeError('Explicit Stop cleanup is pending.')
            self.config_store.update(change)
            if group.observed_state == 'stopped':
                group._apply_spec(candidate)
            else:
                group.restart(candidate)
            return self.settings(sid)

    # -- config io ----------------------------------------------------------

    def _load_base(self) -> list[dict]:
        """server-local read-only Base 설정 로드 (쓰지 않음)."""
        if not self.base_path or not os.path.isfile(self.base_path):
            return []
        try:
            with open(self.base_path) as f:
                data = yaml.safe_load(f) or {}
            streams = data.get("streams") or []
            for s in streams:
                s.setdefault("stream_id", None)
            return [s for s in streams if s.get("stream_id")]
        except Exception as e:
            _log(f"base config load failed ({self.base_path}): {e}")
            return []

    def _load_overlay(self) -> dict:
        return self.config_store.read()

    def effective_specs(self) -> list[dict]:
        """유효 스트림 세트 = 베이스 − removed + added."""
        ov = self._load_overlay()
        removed = set(ov["removed"])
        out: list[dict] = []
        for s in self._load_base():
            if str(s.get("stream_id")) not in removed:
                out.append(dict(s))
        # added는 베이스에 없어야 정상 (있으면 덮어쓰기).
        base_ids = {str(s.get("stream_id")) for s in self._load_base()}
        for a in ov["added"]:
            if str(a.get("stream_id")) not in base_ids:
                out.append(dict(a))
        return out

    # -- api ----------------------------------------------------------------

    @_serialized
    def add(self, spec: dict, start: bool = STREAM_START_DEFAULT) -> StreamGroup:
        sid = str(spec.get("stream_id") or "").strip()
        if not sid:
            raise ValueError("stream_id required")
        base = next((s for s in self._load_base() if str(s.get("stream_id")) == sid), None)
        if sid in self._groups:
            raise ValueError("Stream already exists.")
        gspec = deepcopy(base if base is not None else spec)
        if gspec.get("source_id") is not None and any(
                g.spec.get("source_id") == gspec["source_id"] for g in self._groups.values()):
            raise ValueError("Source identity is already registered to another stream.")
        gspec["stream_id"] = sid
        gspec.setdefault("site_id", "SITE01")
        gspec.setdefault("source_type", "rtsp")
        group = self._new_group(gspec)
        def change(overlay):
            overlay["removed"] = [r for r in overlay["removed"] if r != sid]
            if base is None:
                overlay["added"] = [s for s in overlay["added"] if str(s["stream_id"]) != sid]
                overlay["added"].append(gspec)
        self.config_store.update(change)
        with self._lock:
            self._groups[sid] = group
        self._register_metadata(group)
        if start:
            group.start()
        return group

    @_serialized
    def remove(self, sid: str, stop_first: bool = True):
        group = self.get(sid)
        if group is None:
            raise KeyError(sid)
        base_ids = {str(s["stream_id"]) for s in self._load_base()}
        def change(overlay):
            if sid in base_ids:
                if sid not in overlay["removed"]:
                    overlay["removed"].append(sid)
            else:
                overlay["added"] = [s for s in overlay["added"] if str(s["stream_id"]) != sid]
        if stop_first and group.state != "stopped":
            group.stop()
            if group.observed_state != "stopped":
                raise RuntimeError("Stream cleanup failed.")
        if not group._cleanup_done.is_set():
            raise RuntimeError('Stream cleanup is incomplete.')
        self.config_store.update(change)
        with self._lock:
            self._groups.pop(sid, None)

    def get(self, sid: str) -> StreamGroup | None:
        with self._lock:
            return self._groups.get(sid)

    def list(self, detail: bool = False) -> list[dict]:
        with self._lock:
            groups = list(self._groups.values())
        return [g.status(detail=detail) for g in groups]

    @_serialized
    def start(self, sid: str):
        group = self.get(sid)
        if group is None:
            raise KeyError(sid)
        configured = self._saved_spec(group)
        if configured != group.spec:
            group.restart(configured)
        else:
            group.start()
        return group

    @_serialized
    def stop(self, sid: str):
        group = self.get(sid)
        if group is None:
            raise KeyError(sid)
        group.stop()
        return group

    @_serialized
    def restart(self, sid: str):
        group = self.get(sid)
        if group is None:
            raise KeyError(sid)
        group.restart(self._saved_spec(group))
        return group

    def camera_stream_id(self, source_id):
        with self._lock:
            matches = [g.stream_id for g in self._groups.values()
                       if g.spec.get("source_id") == source_id]
        if len(matches) != 1:
            raise KeyError("Camera is not registered.")
        return matches[0]

    @_serialized
    def apply_and_start(self, spec: dict):
        sid = spec["stream_id"]
        group = self.get(sid)
        if group is None:
            return self.add(spec, start=True)
        if group.spec.get("source_id") != spec.get("source_id"):
            raise ValueError("Stream identity conflicts.")
        if any(str(s["stream_id"]) == sid for s in self._load_base()):
            raise ValueError("Read-only base configuration cannot be changed.")
        candidate = deepcopy(spec)
        def change(overlay):
            overlay["added"] = [s for s in overlay["added"] if str(s["stream_id"]) != sid]
            overlay["added"].append(candidate)
            overlay["removed"] = [r for r in overlay["removed"] if r != sid]
        with group._teardown_lock:
            group.check_restart(candidate)
            if getattr(group, "_pending_restart", None) == candidate:
                return group
            if group.spec == candidate:
                return self.start(sid)
            self.config_store.update(change)
            group.restart(candidate)
        return group

    # -- boot ---------------------------------------------------------------

    def load_and_autostart(self):
        """FastAPI 기동 시 설정을 로드하되 Pipeline은 명시 Start 전까지 중지한다."""
        specs = self.effective_specs()
        if not specs:
            _log("no streams configured (base empty / all removed)")
            return
        for spec in specs:
            sid = str(spec["stream_id"])
            with self._lock:
                g = self._new_group(spec)
                self._groups[sid] = g
            self._register_metadata(g)
            if spec.get("autostart", STREAM_AUTOSTART_DEFAULT):
                _log(f"{sid}: autostart ignored; explicit Start is required")
        _log(f"loaded {len(specs)} streams (base={self.base_path}, overlay={self.overlay_path})")


# 전역 싱글턴 (main.py가 import)
manager = StreamManager()