"""Authored multi-facility UI adapter; the runner owns durable input receipts.

No planning, paid CLI actions, inferred slot capacity, or main-product import.
Movement/selection use the existing reviewed routes. Queue, free slots and
materials are read at batch boundaries, never between normal slot inputs.
"""
from __future__ import annotations
from .fullscreen_popups import recover_fullscreen_popup

from collections import Counter
from copy import deepcopy
from dataclasses import dataclass
import re
import time

from . import catalog
from .errors import Cancelled, GameClientClosed, CoordinatePreparationRequired, InputInterrupted, MiniError, require
from .focused_ocr import FocusedOCR, normalized
from .half_window_data import half_data
from .adaptive_ui import needs_adaptive, image_observation
from .coordinate_store import CoordinateRef
from .adaptive_life import AdaptiveLife, observed_slots, _green_caption
from .execution_rules import number, value, route, geometry
from .ui_layout import LayoutError
from .processing_recovery import (
    ProcessingViewUncertain, ProcessingReceiptUncertain, view_require, receipt_require,
    heading_visible, slot_detail_open, travel_confirmation,
    prepared_travel_button_visible, prepared_travel_popup_visible,
    EscapeLatch, REOBSERVE_INTERVAL, UNCERTAINTY_LIMIT, TRAVEL_CONFIRM_INTERVAL,
)


GROUPS = {"metal": "금속 가공", "wood": "목재 가공", "leather": "가죽 가공", "cloth": "옷감 가공"}
SINGLE_GROUPS = {**GROUPS, "medicine": "약품 가공", "food": "식재료 가공"}


def parse_facility_level(text, group):
    """Read one exact facility/Lv heading, not a recipe or arbitrary number."""
    if group not in SINGLE_GROUPS or not isinstance(text, str):
        return None
    matches = []
    for identifier, title in SINGLE_GROUPS.items():
        name = r"\s*".join(re.escape(char) for char in normalized(title))
        pattern = (r"(?<![가-힣A-Za-z])" + name +
                   r"(?:\s*시\s*설)?\s*L\s*v\s*\.?\s*([0-9]+)(?![0-9]|\.[0-9])")
        matches.extend((identifier, int(match.group(1))) for match in re.finditer(pattern, text, re.I))
    return matches[0][1] if len(matches) == 1 and matches[0][0] == group else None
RECIPE_IDS = frozenset(("metal-iron-specific", "metal-steel", "wood-basic", "wood-plus",
                       "leather-basic", "leather-plus", "cloth-basic", "cloth-silk", "cloth-plus"))
QUEUE_OBSERVATION_AGE = 15.0


@dataclass(frozen=True)
class _QueueObservation:
    group: str
    visible_count: int
    context: tuple
    observed_at: float
    upgrade_visible: bool = False


def work_counts(rows):
    require(isinstance(rows, list), "가공 작업 목록 형식 확인 필요")
    require(all(isinstance(row, dict) and isinstance(row.get("DisplayName"), str)
                and row["DisplayName"] and isinstance(row.get("FacilityName"), str)
                and row["FacilityName"] and type(row.get("IsCompleted")) is bool for row in rows),
            "가공 작업 이름·시설·완료 여부 확인 필요")
    return Counter((row["FacilityName"], row["DisplayName"]) for row in rows)


def slot_shapes(image, *, half=False):
    """Detect measured ring interiors; never turn level/7-minus-jobs into space.

    Uses the main product's Hough/flat-inner-disk detector on profile-specific
    measured ROIs. Dark textured/locked icons remain nonempty. All outputs are
    observations, not actuator locations. Unmatched/duplicate circles reject.
    """
    import cv2
    import numpy as np
    layout = geometry("processing.slots." + ("half" if half else "wide"), image=image)
    rgb = np.asarray(image.convert("RGB").resize(tuple(layout["size"])))
    x0, y0, x1, y1 = layout["bounds"]
    gray = cv2.cvtColor(rgb[y0:y1, x0:x1], cv2.COLOR_RGB2GRAY)
    enhanced = cv2.normalize(gray, None, 0, 255, cv2.NORM_MINMAX)
    circles = cv2.HoughCircles(cv2.GaussianBlur(enhanced, (5, 5), 0), cv2.HOUGH_GRADIENT,
                              dp=1, minDist=layout["min_distance"], param1=number("processing.slots.edge_threshold",70),
                              param2=number("processing.slots.fixed_accumulator",24),
                              minRadius=layout["min_radius"], maxRadius=layout["max_radius"])
    if circles is None:
        return []
    points = layout["centres"]
    found = {}
    yy, xx = np.ogrid[:gray.shape[0], :gray.shape[1]]
    for x, y, radius in circles[0]:
        x, y, radius = map(float, (x, y, radius))
        centre = (x0 + x, y0 + y)
        index = min(range(len(points)), key=lambda i: sum((a-b)**2 for a, b in zip(centre, points[i])))
        if index in found or max(abs(a-b) for a, b in zip(centre, points[index])) > layout["match_tolerance"]:
            return []
        inner = gray[(xx-x)**2 + (yy-y)**2 < (radius*layout["inner_ratio"])**2]
        if not inner.size:
            return []
        contrast = float(inner.std())
        found[index] = {"position": index, "state": "empty" if contrast < number("processing.slots.empty_contrast",2.5) else "nonempty",
                        "contrast": contrast}
    return [found[index] for index in sorted(found)]


class ChildGatherStore:
    """Separate child progress plus durable, non-renewable supply-operation quotas."""
    KEY = "multi_processing_child_run"
    OPERATIONS = "multi_processing_gather_operations"

    def __init__(self, parent, *, operation_id=None, node_id=None, item=None, quantity=None):
        self.parent = parent
        self.path, self.lock = parent.path, parent.lock
        self.operation_id = operation_id
        if operation_id is not None:
            require(isinstance(operation_id, str) and bool(operation_id) and
                    isinstance(node_id, str) and isinstance(item, str) and
                    type(quantity) is int and quantity > 0, "수급 작업 식별 정보가 올바르지 않습니다.")
            with self.lock:
                checkpoint = parent.value.get("multi_processing_checkpoint", {})
                identity = dict(operation_id=operation_id, node_id=node_id, item=item, quantity=quantity,
                                parent_signature=checkpoint.get("signature") or parent.value.get("run", {}).get("signature"),
                                parent_id=checkpoint.get("run_id"))
                record = parent.value.get(self.OPERATIONS, {}).get(operation_id)
                if record is not None:
                    require(all(record.get(key) == value for key, value in identity.items()),
                            "이전 수급 작업의 원래 목표/소유자가 다릅니다. 기존 기록을 보존하며 새 채집을 보내지 않습니다.")
                else:
                    self._commit(dict(identity, confirmed=0, state="prepared", child={}))

    def _record(self):
        return deepcopy(self.parent.value[self.OPERATIONS][self.operation_id])

    def _commit(self, record):
        previous = self.parent.value
        operations = previous.get(self.OPERATIONS, {})
        if operations.get(self.operation_id) == record and previous.get(self.KEY) == record["child"]:
            return
        # Parent settings/queues and other operations are not mutated here.
        # Replace only our operation and compatibility child branch; preserve
        # the complete old root until its single atomic save succeeds.
        updated = dict(previous)
        updated[self.OPERATIONS] = dict(operations)
        updated[self.OPERATIONS][self.operation_id] = deepcopy(record)
        updated[self.KEY] = deepcopy(record["child"])
        self.parent.value = updated
        try:
            self.parent.save()
        except BaseException:
            self.parent.value = previous
            raise

    def _commit_child(self, child):
        previous = self.parent.value
        if previous.get(self.KEY) == child:
            return
        self.parent.value = dict(previous, **{self.KEY: deepcopy(child)})
        try:
            self.parent.save()
        except BaseException:
            self.parent.value = previous
            raise

    def run_snapshot(self):
        """Independent child-only read, without copying parent settings/queues."""
        with self.lock:
            child = (self.parent.value[self.OPERATIONS][self.operation_id]["child"]
                     if self.operation_id is not None else self.parent.value.get(self.KEY, {}))
            return deepcopy(child)

    @property
    def value(self):
        with self.lock:
            value = deepcopy(self.parent.value)
            value["run"] = self.run_snapshot()
            return value

    def begin_run(self, kind, signature, **defaults):
        with self.lock:
            owner = self.parent.value.get("run", {}).get("signature")
            if self.operation_id is not None:
                record = self._record()
                current = record["child"]
                if current:
                    require(current.get("kind") == kind and current.get("signature") == signature,
                            "수급 재개의 원래 채집 목표가 다릅니다. 새 기준으로 다시 채집하지 않습니다.")
                else:
                    current = dict(kind=kind, signature=signature, parent_signature=owner,
                                   operation_id=self.operation_id, **defaults)
                    record["child"] = current
                    record["state"] = current.get("state", "running")
                    self._commit(record)
                return deepcopy(current)
            current = self.parent.value.get(self.KEY)
            if not (isinstance(current, dict) and current.get("kind") == kind and
                    current.get("signature") == signature and current.get("parent_signature") == owner and
                    current.get("state") not in ("complete", "complete_currency_exhausted")):
                current = dict(kind=kind, signature=signature, parent_signature=owner, **defaults)
                self._commit_child(current)
            return deepcopy(current)

    def set_run(self, **values):
        with self.lock:
            if self.operation_id is not None:
                record = self._record()
                child = record["child"]
                completed = values.get("completed", record["confirmed"])
                require(type(completed) is int and completed >= 0, "수급 진행 수량 형식이 올바르지 않습니다.")
                completed = max(record["confirmed"], completed)
                values["completed"] = completed
                baseline = child.get("baseline_quantity")
                require(baseline is None or values.get("baseline_quantity", baseline) == baseline,
                        "이전 수급의 기준 수량을 변경할 수 없습니다.")
                if (all(key in child and child[key] == value for key, value in values.items())
                        and record["confirmed"] == completed and
                        record["state"] == child.get("state", record["state"])):
                    return
                child.update(deepcopy(values))
                record["confirmed"] = completed
                record["state"] = child.get("state", record["state"])
                self._commit(record)
                return
            current = self.parent.value.get(self.KEY, {})
            if all(key in current and current[key] == value for key, value in values.items()):
                return
            child = dict(current)
            child.update(deepcopy(values))
            self._commit_child(child)

    def log(self, message, **data):
        self.parent.log(message, child="multi_processing_gather", **data)

    def preference(self, keys, default=None):
        # Material-supply gathering reads the SAME user settings container.
        return self.parent.preference(keys, default)


class MultiProcessingPort:
    def __init__(self, service, *, single=False):
        self.service = service
        self.group = None
        self.recipe = None
        self.single = single
        self._life = None
        self._owned_view = False
        self._travel_group = None
        self._travel_need_t = False
        self._travel_confirm_at = -float("inf")
        self._escape_latch = EscapeLatch()
        self._recovery_started = {}
        self._queue_observation = None
        self._selection_owner = None
        self._local_owner = None
        self._route_owner = None
        self._surface_owner = None
        self._transition_last_popup = time.monotonic()
        self._move_upgrade_visible = False
        self._single_start_pending = False
        self._travel_started = False
        self._arrival_detail_closed = False
        self._collection_tail = None
        self._facility_levels = {}

    def collection_cleanup(self):
        store = getattr(self.service, "store", None)
        return (store.value.get("processing_cleanup") if store is not None else self._collection_tail)

    def _save_collection_cleanup(self, receipt):
        store = getattr(self.service, "store", None)
        if store is not None:
            store.set_processing_cleanup(receipt)
        self._collection_tail = deepcopy(receipt)

    def remember_collected_result(self, recipe):
        """Persist UI ownership before the caller clears its consumption receipt."""
        receipt = dict(kind="processing" if self.single else "multi_processing",
                       group=recipe["group"], recipe=recipe["id"],
                       task_identity=getattr(self.service, "_task_identity", None))
        current = self.collection_cleanup()
        require(current is None or current == receipt,
                "다른 가공 작업의 수령 완료창 기록이 남아 있습니다. 같은 작업으로 재개해 주세요.")
        if current is None:
            self._save_collection_cleanup(receipt)

    def _cleanup_recipe(self):
        receipt = self.collection_cleanup()
        if receipt is None:
            return None
        require(receipt.get("kind") == ("processing" if self.single else "multi_processing") and
                receipt.get("task_identity") == getattr(self.service, "_task_identity", None),
                "다른 작업의 가공 완료창 기록이 남아 있습니다. 같은 작업으로 재개해 주세요.")
        recipe = next((row for row in catalog.processing_catalog()["recipes"]
                       if row["id"] == receipt.get("recipe") and row["group"] == receipt.get("group")), None)
        require(recipe is not None, "가공 완료창의 품목/시설 기록을 확인해 주세요.")
        return recipe

    def _clear_collection_cleanup(self, group):
        recipe = self._cleanup_recipe()
        if recipe is not None and recipe["group"] == group:
            self._save_collection_cleanup(None)

    def settle_collection_tail(self, *, observed_state=None):
        recipe = self._cleanup_recipe()
        if recipe is None:
            return None
        return self.finish_collection(recipe, observed_state=observed_state)

    def recovery_progress(self, group):
        """Called only after the runner has confirmed actual work progress."""
        self._recovery_started.pop(group, None)
        self._escape_latch.progress()

    def recovery_opened(self, group, state):
        """An authored input opened this known surface, not merely a reread."""
        self._escape_latch.opened(group, state)
        self._remember_surface(group, state)
        if state == "local":
            self._local_owner = (group, self._queue_context())
            self._route_owner = self._local_owner
            self._selection_owner = None
        elif state in ("field", "remote", "remote_detail"):
            self._route_owner = self._local_owner = self._selection_owner = None

    def _remember_surface(self, group, state):
        self._surface_owner = (None if state == "field" else group, state, self._surface_context())

    def _surface_context(self):
        from .windows import last_input_stamp
        return self._queue_context()[:-1], last_input_stamp()

    def _surface_valid(self, context):
        from .automation import _input_is_owned
        current, stamp = self._surface_context()
        return current == context[0] and (stamp == context[1] or _input_is_owned(self.service, stamp))

    def owned_surface(self, group=None):
        surface = self._surface_owner
        if surface is None or not self._surface_valid(surface[2]):
            return None
        return surface[1] if group is None or surface[0] == group or surface[1] == "field" else None

    def _context_guard(self, context):
        self.service.check()
        return view_require(self._surface_valid(context),
                            "가공 경로 중 사용자 조작/화면 변경 · 현재 단계부터 재확인")

    def local_owned(self, group):
        return self._route_owner == (group, self._queue_context())

    def adopt_local(self, recipe):
        """Compatibility entry: observe the actual field/remote/local branch."""
        return self.visit(recipe)

    def _leave_other_observed_facility(self, group, image):
        """An initial different facility needs positive identity, not blind ESC."""
        found = [(other, self._state(other, image)) for other in SINGLE_GROUPS if other != group]
        found = [(other, state) for other, state in found
                 if state in ("local", "remote", "local_detail", "remote_detail")]
        if len(found) != 1:
            return False
        self.group, state = found[0]
        self.recipe, self._owned_view = None, True
        self._remember_surface(self.group, state)
        if self._life is not None:
            self._life.group, self._life.recipe = self.group, None
        self._leave()
        return True

    def ensure_start_local(self, group, image):
        """A user-opened remote overview is not a usable local work queue."""
        if not self._single_start_pending:
            return False
        state = self._transition_state(group, image)
        rechecked = state not in ("local", "remote")
        if rechecked:
            # The user's opened facility owns its identity, not proximity.
            # Keep this one start boundary until the actual branch is known;
            # an unread Move caption cannot grant local registration rights.
            self._route_owner = self._local_owner = self._surface_owner = None
            state = self._wait(group, ("local", "remote"))
        self._single_start_pending = False
        if state == "local":
            self.recovery_opened(group, "local")
            return rechecked
        self._route_owner = self._local_owner = None
        self.recovery_opened(group, "remote")
        self._travel(group, self._facility_travel_ref())
        return True

    def _facility_travel_ref(self):
        if not self._adaptive() and not self._move_upgrade_visible:
            return self._fixed_target("processing.facility.travel")
        runtime, _ = self._source_runtime()
        return runtime.prepared_ref("processing.facility.travel", "processing.facility.travel",
            facility_id=self.group, remote=True, move_visible=True,
            upgrade_visible=self._move_upgrade_visible)

    def _tape_guard(self, group, context):
        # Keep input/PID/UI/manual-pause protection, not an OCR approval per
        # click. Consumption receipts are armed at the physical press boundary.
        self.service.check()
        return view_require(self.local_owned(group) and self._queue_context() == context,
                            "가공 입력 중 사용자 조작/화면 변경 · 현재 작업 목록부터 재확인")

    def _fixed_target(self, role, recipe=None):
        runtime, _ = self._source_runtime()
        recipe = self.recipe if recipe is None else recipe
        view_require(runtime is not None and runtime.require_prepared and recipe is not None,
                     "고정 가공 좌표 컨테이너 준비가 필요합니다.")
        return runtime.prepared_ref("processing.fixed", role, recipe_id=recipe["id"])

    def remember_selection(self, recipe):
        """A sent authored selection owns identity, never quantity or success."""
        self._selection_owner = (recipe["group"], recipe["id"], self._queue_context())
        self._remember_surface(recipe["group"], "local_detail")

    def selection_owned(self, recipe):
        return self._selection_owner == (recipe["group"], recipe["id"], self._queue_context())

    def _owned_detail_ready(self, recipe, image):
        if not self.selection_owned(recipe):
            return False
        if bool(getattr(self.service, "_source_window", lambda: False)()):
            life = self._adaptive_life()
            register = life._ocr_bands("processing.register")[0]
            body = life._ocr_bands("processing.recipe.body")[0]
        elif self.service._half_window():
            controls = half_data()["processing_controls"]
            register, body = controls["register"], controls["materials_text"]
        elif self._adaptive():
            return False  # The observed-target adapter still needs current boxes.
        else:
            rois = geometry("processing.fixed.observation", image=image)
            register, body = rois["register"], rois["recipe_body"]
        return _green_caption(image, register) and heading_visible(image, body)

    def _owned_visual_state(self, group, image):
        """Current pixels plus route identity, with OCR only as fallback.

        Ownership is invalid after actual user input, PID/size/UI changes or a
        deliberate close. It never turns an empty/black frame into local state.
        """
        if not getattr(self.service.window, "is_foreground", lambda: True)():
            return None
        selected = self._selection_owner
        if selected and selected[0] == group and selected[2] == self._queue_context():
            recipe = next((row for row in catalog.processing_catalog()["recipes"]
                           if row["id"] == selected[1]), None)
            if recipe is not None and self._owned_detail_ready(recipe, image):
                return "local_detail"
        if self._local_owner != (group, self._queue_context()):
            return None
        if bool(getattr(self.service, "_source_window", lambda: False)()):
            life = self._adaptive_life()
            heading = life._ocr_bands("processing.facility.heading")[0]
            slots, _ = self._read_source_queue(image, group=group)
        elif self._adaptive():
            return None
        else:
            heading = geometry("processing.fixed.observation", image=image)["heading"]
            slots = slot_shapes(image, half=self.service._half_window())
        if heading_visible(image, heading) and slots:
            return "local"
        return None

    def _close_once(self, group, state, *, kind="key", value="ESC", recovery=False):
        s = self.service
        latch = self._escape_latch
        view_require(latch.available(group, state, time.monotonic(), recovery=recovery),
                     "이미 닫기 입력을 보낸 화면입니다. 반복 ESC 없이 상태를 다시 확인합니다.")
        context = self._surface_context()
        owned = not recovery and self.owned_surface(group) == state
        def current():
            same = ((self._context_guard(context) and self.owned_surface(group) == state if owned
                     else self._state(group) == state) and
                    latch.available(group, state, time.monotonic(), recovery=recovery))
            if recovery:
                return same
            return view_require(same, "닫기 직전 가공 화면 변경 · 상태 재확인 필요")
        def closing():
            latch.sent(group, state, time.monotonic())
            if state == "local_detail" and self._travel_group == group:
                self._arrival_detail_closed = True
            self._selection_owner = None
            self._surface_owner = None
            if state in ("local", "remote", "unknown"):
                self._local_owner = None
                self._route_owner = None
        sent = s.processing_input(kind, value, guard=current, skip_if_guard_fails=recovery,
                                  before_send=closing)
        if sent is not False and owned:
            following = {"local_detail": "local", "remote_detail": "remote", "slot_detail": "local",
                         "local": "field", "remote": "field"}[state]
            self.recovery_opened(group, following)
        return sent

    def _adaptive(self):
        reader = getattr(self.service, "_adaptive_window", None)
        if callable(reader):
            return reader()
        window=self.service.window
        width,height=getattr(window,"width",None),getattr(window,"height",None)
        return needs_adaptive((width,height)) if type(width) is int and type(height) is int else False

    def _adaptive_life(self):
        from .source_life_port import SourceLife
        adapter = SourceLife
        if self._life is None or type(self._life) is not adapter:
            previous = self._life
            self._life = adapter(self.service, processing_port=self)
            if previous is not None:
                self._life.group, self._life.recipe = previous.group, previous.recipe
        return self._life

    def _queue_context(self):
        """Physical identity/profile only; a desktop move does not resize UI."""
        window = self.service.window
        settings = getattr(window, "ui_settings", None)
        return (getattr(window, "hwnd", None), getattr(window, "pid", None),
                getattr(window, "width", None), getattr(window, "height", None),
                getattr(window, "layout_generation", None), getattr(settings, "signature", None),
                getattr(self.service, "input_resume_generation", 0))

    def _remember_facility_level(self, group, level):
        from .source_processing_capacity import slot_capacity
        count = slot_capacity(group, level)
        view_require(type(count) is int and 1 <= count <= 7,
                     "현재 가공 시설 레벨의 원본 슬롯 수를 확인하지 못했습니다.")
        self._facility_levels[group] = (self._queue_context(), level, count)
        return count

    def known_capacity(self, group):
        """Cached source-table value only; safe inside a prepared input tape."""
        entry = self._facility_levels.get(group)
        if entry is not None and entry[0] == self._queue_context():
            return entry[2]
        return None

    def facility_capacity(self, group, image=None):
        """Resolve current facility Lv once at a boundary, never from rings."""
        s = self.service
        s.check()
        require(group in SINGLE_GROUPS and self.group == group, "현재 방문한 가공 시설이 아닙니다.")
        if self._single_start_pending:
            image = s.window.capture() if image is None else image
            if self.ensure_start_local(group, image):
                image = None  # A remote start may have moved before reading Lv.
        view_require(self.local_owned(group), "현장 가공 시설의 레벨을 확인해 주세요.")
        cached = self.known_capacity(group)
        if cached is not None:
            return cached
        view_require(not getattr(s, "_coordinate_depth", 0),
                     "가공 입력 묶음 전에 시설 레벨을 다시 확인합니다.")
        context = self._queue_context()
        runtime, snapshot = self._source_runtime()
        image = s.window.capture() if image is None else image
        view_require(image.size == snapshot.client_size, "시설 레벨 관측 중 게임 크기가 변경되었습니다.")
        row = runtime.prepared_bundle("life.observation")["processing.facility.heading"]
        region = row["metadata"]["normalized_roi"]
        view_require(heading_visible(image, region), "현장 가공 시설 레벨 제목 재확인 필요")
        if s.ocr is None:
            s.ocr = FocusedOCR()
        level = parse_facility_level(s.ocr.text(image, region), group)
        view_require(level is not None and self._queue_context() == context and self.local_owned(group),
                     "현재 가공 시설 이름/레벨을 다시 확인합니다.")
        count = self._remember_facility_level(group, level)
        s.status(f"{SINGLE_GROUPS[group]} 시설 Lv.{level} · 원본 슬롯 {count}개 · 준비 배치 사용")
        return count

    def _remember_visible_queue(self, group, image, slots, *, upgrade_visible=False):
        """Retain real visible rings separately from usable work capacity."""
        self._queue_observation = None
        context = self._queue_context()
        frame_generation = image.info.get("game_layout_generation")
        view_require(image.size == context[2:4] and
                     (frame_generation is None or frame_generation == context[4]),
                     "슬롯 관측 중 게임 크기/UI 설정이 변경되어 다시 확인합니다.")
        view_require(1 <= len(slots) <= 7, "현재 보이는 가공 큐 개수를 확인하지 못했습니다.")
        captured_at = image_observation(image)[0]
        view_require(0 <= time.monotonic() - captured_at <= QUEUE_OBSERVATION_AGE,
                     "가공 큐 관측이 오래되어 현재 화면에서 다시 확인합니다.")
        self._queue_observation = _QueueObservation(group, len(slots), context, captured_at, upgrade_visible)
        if self.collection_cleanup() is not None:
            heading = (self._adaptive_life()._ocr_bands("processing.facility.heading")[0]
                       if self._adaptive() else geometry("processing.fixed.observation", image=image)["heading"])
            if heading_visible(image, heading):
                self._clear_collection_cleanup(group)

    def _queue_observation_valid(self, group):
        observed = self._queue_observation
        return bool(observed is not None and observed.group == group and
                    observed.context == self._queue_context() and
                    0 <= time.monotonic() - observed.observed_at <= QUEUE_OBSERVATION_AGE)

    def _source_queue_hint(self, image, *, upgrade_visible=False, group=None):
        """Read the prepared observation grid; pixels supply only slot state."""
        runtime, snapshot = self._source_runtime()
        view_require(image.size == snapshot.client_size,
                     "슬롯 관측 중 게임 크기가 변경되었습니다.")
        return runtime.prepared_bundle("processing.queue.observe", maximum=7,
            upgrade_visible=upgrade_visible, queue_count=self.known_capacity(self.group if group is None else group))[
                "processing.queue.hint"]["metadata"]["hint"]

    def _read_source_queue(self, image, *, group=None):
        # Select a precomputed local layout branch. This observes real slots;
        # it never changes input coordinates or infers seven from work counts.
        variants = []
        _, snapshot = self._source_runtime()
        for upgrade in ((False, True) if snapshot.orientation == "portrait" else (False,)):
            slots = observed_slots(image, source_grid=self._source_queue_hint(image, upgrade_visible=upgrade, group=group))
            if slots:
                variants.append((slots, upgrade))
        return variants[0] if len(variants) == 1 else ([], False)

    def _source_runtime(self):
        runtime = self.service._ui_route_runtime()
        view_require(runtime.require_prepared, "생활 좌표 컨테이너 준비가 필요합니다.")
        return runtime, runtime.refresh(self.service.window)

    def _focus_for_visit(self, group):
        """Retry only a rejected foreground request, without owning any UI.

        Before the first recognized screen, recover() intentionally cannot
        dismiss unknown UI. A focus interruption must not enter that recovery
        branch and turn into a permanent invalid-start failure. No click/key,
        recipe selection or receipt is sent/created while focus is pending.
        """
        from .windows import wait_input_idle
        from .automation import wait_route_input_idle
        s = self.service
        retry = False
        last_report = -float("inf")
        while True:
            s.check()
            if retry:
                wait_route_input_idle(s, wait_input_idle)
                s.check()
            try:
                s.window.focus()
                if self.owned_surface() is None:
                    recover_fullscreen_popup(s)
                return
            except InputInterrupted:
                s.check()  # Cancellation is never changed into a focus retry.
                now = time.monotonic()
                if now - last_report >= 10:
                    s.status("가공 시작 포커스 전환 보류 · 게임 입력 없이 안정 대기 후 다시 확인합니다.")
                    last_report = now
                retry = True
                s.sleep(number("processing.recovery.poll_interval", REOBSERVE_INTERVAL))

    def _recipe(self, recipe):
        allowed = {row["id"] for row in catalog.processing_catalog()["recipes"]} if self.single else RECIPE_IDS
        require(isinstance(recipe, dict) and recipe.get("id") in allowed, "가공 지원 품목이 아닙니다.")
        actual = next(row for row in catalog.processing_catalog()["recipes"] if row["id"] == recipe["id"])
        require(all(recipe.get(key) == actual[key] for key in ("name", "group", "materials")),
                "멀티 가공 품목·재료 도감 불일치")
        return actual

    def inventory(self):
        self.service.check()
        try:
            rows = self.service.connector.inventory()
            require(isinstance(rows, list), "공식 재고 목록 확인 필요")
        except (Cancelled, GameClientClosed, CoordinatePreparationRequired):
            raise
        except MiniError as exc:
            raise ProcessingViewUncertain("현재 가공 재고 재조회 필요: " + str(exc)) from exc
        return rows

    def works(self, group):
        require(group in (SINGLE_GROUPS if self.single else GROUPS), "가공 지원 시설이 아닙니다.")
        self.service.check()
        try:
            rows = self.service._works_for_group(group)
            work_counts(rows)
        except (Cancelled, GameClientClosed, CoordinatePreparationRequired):
            raise
        except MiniError as exc:
            raise ProcessingViewUncertain("현재 가공 작업 목록 재조회 필요: " + str(exc)) from exc
        return rows

    def _text(self, image, region):
        if self.service.ocr is None:
            self.service.ocr = FocusedOCR()
        return normalized(self.service.ocr.text(image, region))

    @staticmethod
    def _label(text, role, fallback):
        labels = value("life.labels."+role, fallback)
        labels = (labels,) if isinstance(labels, str) else labels
        require(isinstance(labels, (list, tuple)) and all(isinstance(label, str) and label for label in labels),
                "가공 판독 문구 규칙 형식 오류")
        return any(normalized(label) in text for label in labels)

    def _rows(self, image, region):
        import numpy as np
        if self.service.ocr is None:
            self.service.ocr = FocusedOCR()
        bounds = tuple(round(v*image.size[i % 2]) for i, v in enumerate(region))
        result = self.service.ocr.engine(np.asarray(image.crop(bounds).convert("RGB"))[:, :, ::-1].copy())
        return [(str(text), min(p[0] for p in box)+bounds[0], min(p[1] for p in box)+bounds[1])
                for text, box in zip(result.txts or (), result.boxes if result.boxes is not None else ())]

    def _state(self, group, image=None):
        s = self.service
        image = s.window.capture() if image is None else image
        owned = self._owned_visual_state(group, image)
        if owned is not None:
            return owned
        if self._adaptive():
            return self._adaptive_life().state(group,image)
        if s.ocr is None:
            s.ocr = FocusedOCR()
        rois = geometry("processing.fixed.observation", image=image)
        if slot_detail_open(image,s.ocr):
            return "slot_detail"
        if s._half_window():
            state=s._half_processing_state(group,image)
            bright_region=(half_data()["processing_controls"]["materials_text"]
                           if state in ("local_detail","remote_detail") else rois["heading"])
            if state in ("local","remote","local_detail","remote_detail","facilities") and not heading_visible(image,bright_region):
                return "unknown"
            return state
        if (self._label(self._text(image, rois["result_title"]), "processing.result.title", "가공 완료") and
                self._label(self._text(image, rois["result_confirm"]), "processing.result.confirm", "확인")):
            return "reward"
        heading = self._text(image, rois["heading"])
        facility_name=(SINGLE_GROUPS if self.single else GROUPS)[group]
        if normalized(facility_name) in heading:
            caption = self._text(image, rois["register"])
            # A normal recipe detail dims the facility behind it. Its own
            # foreground text, not that background heading, proves the view.
            caption_lit=heading_visible(image,rois["recipe_body"])
            if self._label(caption, "processing.travel", "가공하러 가기") and caption_lit:
                return "remote_detail"
            if (self._label(caption, "processing.register", "가공하기") and caption_lit and
                    self._label(self._text(image, rois["materials"]), "processing.materials", "필요한 재료")):
                return "local_detail"
            if not heading_visible(image):
                return "unknown"
            return self._transition_state(group, image)
        if s.ocr.field_ready(image):
            return "field"
        if self._adaptive_life().facility_list_ready(image):
            return "facilities"
        return "unknown"

    def _travelling(self, group):
        """Observe movement only for this port's already-sent travel request."""
        if group is None or self._travel_group != group:
            return False
        from .automation import _flag
        activity = self.service.connector.read("get_activity")
        travelling = _flag(activity, "IsAutoTraveling", "autoTravel")
        self._travel_started = self._travel_started or travelling is True
        return travelling

    def _close_arrival_detail(self, group):
        # Closing a remembered detail is not proof that its underlying list
        # has appeared. Keep the travel owner and wait for the actual list;
        # a fading/same detail frame must not rearm and resend Escape.
        arrival = self._travel_group == group
        if arrival and self._arrival_detail_closed:
            return
        if arrival or self._escape_latch.available(group, "local_detail", time.monotonic()):
            self.recovery_opened(group, "local_detail")
            self._close_detail(group)

    def _wait(self, group, expected, timeout=15, *, result_cleanup=False):
        s = self.service
        context = self._surface_context()
        last_report = time.monotonic()
        while True:
            s.check()
            view_require(self._surface_valid(context),
                         "가공 전환 중 사용자 입력/창 변경 · 현재 단계부터 다시 확인합니다.")
            foreground = getattr(s, "observation_foreground", None)
            if callable(foreground) and not foreground():
                s.sleep(number("processing.wait.poll_interval", .5))
                continue
            s.coordinate_step()
            image=s.window.capture()
            if self._confirm_travel(group,image):
                continue
            # A movement observation belongs ahead of a visual arrival
            # candidate. HUD colors/transition frames cannot dismiss travel.
            travelling = self._travelling(group)
            state = "unknown" if travelling else self._transition_state(group, image)
            if state == "unknown" and "field" in expected:
                # Receipt recovery may already have returned to the HUD.
                # Normal facility travel never asks for field and therefore
                # does not acquire an extra HUD/OCR approval here.
                field = (self._adaptive_life()._field_ready(image) if self._adaptive()
                         else s.ocr is not None and s.ocr.field_ready(image))
                if field:
                    state = "field"
            if result_cleanup and state == "unknown":
                # Only an owned result tail uses this exception reader. A
                # still-visible reward permits another Confirm, never Collect.
                state = self._state(group, image)
            if not travelling and state in expected and not (travelling is None and state == "field"):
                if state in ("local", "field"):
                    self._travel_group=None
                self.recovery_opened(group, state)
                return state
            if not travelling and state == "local_detail" and "local" in expected:
                # Native arrival may open a remembered recipe as well as the
                # facility. Reuse its owned close, then observe the real list.
                self._close_arrival_detail(group)
                s.sleep(number("processing.wait.poll_interval", .5))
                continue
            if (travelling and self._travel_need_t and getattr(s,"use_travel_wings",True)):
                with s.coordinate_sequence():
                    s.processing_input("key","T",guard=lambda:getattr(s,"use_travel_wings",True) and self._context_guard(context),
                        skip_if_guard_fails=True,
                        before_send=lambda:setattr(self,"_travel_need_t",False))
            now = time.monotonic()
            # Popup/OCR recovery is exceptional, not a gate in every normal
            # menu transition. It is never placed before a ready move button.
            if now-self._transition_last_popup >= number("processing.popup.poll_interval", 3):
                self._transition_last_popup = now
                if recover_fullscreen_popup(s, image):
                    continue
            if travelling:
                s.sleep(number("processing.wait.poll_interval", .5))
                continue
            if (state == "remote" and travelling is False and self._travel_group == group and
                    now-getattr(self, "_travel_request_at", now) >= number("processing.travel.confirm_interval", .5)):
                self._send_travel(group, self._travel_target)
                continue
            if time.monotonic()-last_report >= number("processing.wait.report_interval", timeout):
                # A positive old remote overview means a missed non-consuming
                # movement request, not permission to replay registration.
                recovered = self._state(group,image)
                if recovered == "facilities" and "remote" in expected and self._travel_group is None:
                    # A positively observed list proves the non-consuming
                    # facility selection did not finish. Retry only that suffix,
                    # never K/tab, Register, Collect or an unknown overlay.
                    self.recovery_opened(group, "facilities")
                    self._retry_facility_selection(group)
                elif recovered in ("local", "remote", "local_detail", "field"):
                    if recovered in expected and not (travelling is None and recovered == "field"):
                        self.recovery_opened(group, recovered)
                        if recovered in ("local", "field"):
                            self._travel_group = None
                        return recovered
                    if recovered == "local_detail" and "local" in expected:
                        self._close_arrival_detail(group)
                s.status("가공 이동 연결 재확인 · 설비/이동/도착 단계 유지 · 등록 재전송 없음")
                last_report = time.monotonic()
            s.sleep(number("processing.wait.poll_interval", .5))

    def _confirm_travel(self,group,image):
        """Only an actually sent facility-travel request owns this question."""
        if self._travel_group!=group:return False
        s=self.service
        context = self._surface_context()
        runtime, _ = self._source_runtime()
        observations = runtime.prepared_bundle("life.observation")
        confirm = observations["processing.travel.confirm"]["metadata"]["normalized_roi"]
        cancel = observations["processing.travel.cancel"]["metadata"]["normalized_roi"]
        if not prepared_travel_popup_visible(image, confirm, cancel):
            return False
        if time.monotonic()-self._travel_confirm_at < number("processing.travel.confirm_interval", .5):
            s.coordinate_step()
            return True
        reference = (runtime.prepared_ref("processing.travel.confirm", "processing.travel.confirm", facility_id=group)
                     if self._adaptive() else self._fixed_target("processing.travel.confirm"))
        with s.coordinate_sequence():
            s.processing_input("click", reference,
                guard=lambda: self._context_guard(context) and prepared_travel_popup_visible(s.window.capture(), confirm, cancel),
                before_send=lambda:setattr(self,"_travel_confirm_at",time.monotonic()))
        s.status(f"{SINGLE_GROUPS[group]} 시설이 있는 마을로 이동 확인 · 제작/개별 수령 입력 없음")
        return True

    def _travel(self,group,region):
        self._travel_target = region
        self._travel_started = False
        self._send_travel(group, region)
        self._wait(group,("local",))
        self._travel_group = None
        self.recovery_opened(group,"local")

    def _send_travel(self,group,region):
        view_require(isinstance(region, CoordinateRef), "설비 이동은 준비된 좌표 ID만 사용합니다.")
        def armed():
            self._owned_view=True
            if self._travel_group != group:
                self._arrival_detail_closed = False
            self._travel_group=group
            self._travel_need_t=True
            self._travel_confirm_at=-float("inf")
            self._travel_request_at=time.monotonic()
        context = self._surface_context()
        def current():
            return self._context_guard(context) and view_require(self.owned_surface(group) == "remote",
                "시설 이동 경로가 변경되어 현재 단계부터 재확인합니다.")
        with self.service.coordinate_sequence():
            self.service.processing_input("click",region,guard=current,before_send=armed)

    def _transition_state(self, group, image):
        """Only dynamic overview/arrival branches, not full-screen OCR approval."""
        # Upgrade and Move use the SAME inactive backing in the client. One
        # small role read is the actual dynamic branch, not a facility-title
        # or full-screen approval. All subsequent clicks use prepared IDs.
        life = self._adaptive_life()
        found = life.transition_state(group, image)
        if self._adaptive() or found != "unknown":
            return found
        lane = life._ocr_bands("processing.facility.travel")[0]
        if prepared_travel_button_visible(image, lane, fraction=.12):
            return "unknown"  # A failed Move label is not a local facility.
        slots = slot_shapes(image, half=self.service._half_window())
        if slots and heading_visible(image, geometry("processing.fixed.observation", image=image)["heading"]):
            self._remember_visible_queue(group,image,slots)
            return "local"
        # The prepared source observer owns detail evidence too. A green HUD
        # or terrain patch in this wide legacy crop is never a recipe detail.
        return "unknown"

    def _guard_state(self,group,expected):
        return view_require(self._state(group) in expected,"입력 직전 가공 화면 변경 · 상태 재확인 필요")

    def _guard_observed(self,target,*,expected_state,recipe=None):
        return view_require(self._adaptive_life().guard_target(target,expected_state=expected_state,recipe=recipe),
                            "입력 직전 관측한 가공 버튼/품목이 변경되었습니다.")

    def recover(self,group,*,timeout=UNCERTAINTY_LIMIT,owned_receipt=False):
        """Reobserve owned work UI; only Escape may dismiss unknown overlays.

        The runner must reconcile pending consumption before entering here.
        This method does not clear receipts, register, collect or cancel work.
        """
        s=self.service
        # Recipe is the operation's semantic target, not proof that its detail
        # is still selected. A same-facility recovery must retain it for the
        # shared collector, while all selection/route authority is reobserved.
        target_recipe = (self._recipe(self.recipe) if isinstance(self.recipe, dict)
                         and self.recipe.get("group") == group else None)
        self._queue_observation = None
        self._route_owner = None
        self._surface_owner = None
        self._selection_owner = None
        # An unreadable FIRST supply/start frame owns no UI yet. It still
        # needs fresh observations, but must not gain a blind Escape budget.
        may_close = self._owned_view or owned_receipt
        now=time.monotonic()
        self._recovery_started[group]=now
        deadline=float("inf") if timeout is None else now+timeout
        while time.monotonic()<deadline:
            s.check()
            if not getattr(s.window,"is_foreground",lambda:True)():
                from .windows import wait_input_idle
                from .automation import wait_route_input_idle
                wait_route_input_idle(s, wait_input_idle)
            s.window.focus()
            image=s.window.capture()
            if recover_fullscreen_popup(s, image):
                deadline=float("inf") if timeout is None else time.monotonic()+timeout
                continue
            if self._confirm_travel(group,image):
                s.sleep(number("processing.recovery.poll_interval", REOBSERVE_INTERVAL));continue
            travelling = self._travelling(group)
            if travelling is True:
                # Recovery must not erase travel ownership or open ESC/K just
                # because an intermediate field/detail candidate was observed.
                if self._travel_need_t and getattr(s, "use_travel_wings", True):
                    context = self._surface_context()
                    s.processing_input("key", "T",
                        guard=lambda: getattr(s, "use_travel_wings", True) and self._context_guard(context),
                        skip_if_guard_fails=True,
                        before_send=lambda: setattr(self, "_travel_need_t", False))
                s.sleep(number("processing.recovery.poll_interval", REOBSERVE_INTERVAL))
                continue
            state=self._state(group,image)
            if travelling is None and state in ("field", "unknown", "remote", "facilities"):
                # A missing movement value proves neither arrival nor a missed
                # request. Preserve the same route; no ESC/K/move replay.
                s.sleep(number("processing.recovery.poll_interval", REOBSERVE_INTERVAL))
                continue
            if self.collection_cleanup() is not None and state in ("reward", "local", "local_detail", "field"):
                result = self.settle_collection_tail(observed_state=state)
                return result["state"]
            self._escape_latch.observe(state,time.monotonic())
            if state in ("local","remote","facilities"):
                self._owned_view = True
                self.group=group;self.recipe=target_recipe;self._travel_group=None
                s._half_processing_viewport=None
                if self._life is not None:self._life.group,self._life.recipe=group,target_recipe
                self.recovery_opened(group, state)
                s.status(f"{SINGLE_GROUPS[group]} 기본 목록 복구 · 확인된 잔여 작업부터 재개")
                return state
            if state=="field":
                self._owned_view = True
                self.recovery_opened(group, "field")
                self.group=self.recipe=self._travel_group=None
                s._half_processing_viewport=None
                if self._life is not None:self._life.group=self._life.recipe=None
                s.status("기본 화면 복구 · 진행 수량 보존 후 해당 가공기부터 재진입")
                return state
            now=time.monotonic()
            if may_close and self._escape_latch.available(group,state,now,recovery=True):
                if self._close_once(group,state,recovery=True):
                    s.status("가공 화면 닫기 1회 · 반복 ESC 없이 상태 재확인")
            s.sleep(number("processing.recovery.poll_interval", REOBSERVE_INTERVAL))
        raise ProcessingViewUncertain("현재 가공 화면을 다시 확인합니다. 작업·입력 기록은 보존합니다.")

    def collection_hit_slot(self,group):
        # Reconciliation reaches this observation BEFORE recover(). A hidden
        # adaptive client otherwise produces a black frame forever, so restore
        # only its focus after user idle; do not dismiss/receive anything here.
        s=self.service
        s.check()
        if not getattr(s.window,"is_foreground",lambda:True)():
            from .windows import wait_input_idle
            from .automation import wait_route_input_idle
            wait_route_input_idle(s, wait_input_idle)
            s.check()
            s.window.focus()
        s.check()
        return self._state(group)=="slot_detail"

    def close_misdirected_slot(self,group):
        self._owned_view=True
        view_require(self.collection_hit_slot(group),"개별 가공 슬롯 상세가 변경되었습니다.")
        self._close_detail(group)

    def _close_detail(self, group):
        s = self.service
        state=self.owned_surface(group)
        if state is None:
            state=self._state(group)
            self._remember_surface(group, state)
        view_require(state in ("local_detail", "remote_detail", "slot_detail"), "가공 상세 창 확인 필요")
        with s.coordinate_sequence():
            self._close_once(group,state)

    def _leave(self, recipe=None):
        """Close only an identified owned facility before another known route."""
        self.settle_collection_tail()
        s = self.service
        self._queue_observation = None
        if self.group is None and recipe is not None:
            # A resumed registration can be proved by official work rows
            # before visit(). Its semantic recipe permits observing that
            # facility, not assuming its screen or sending a blind close.
            recipe = self._recipe(recipe)
            self.group, self.recipe = recipe["group"], recipe
        view_require(self._travelling(self.group) is False,
                     "설비 이동 중에는 화면을 닫지 않고 도착 상태를 다시 확인합니다.")
        state = self.owned_surface(self.group)
        if state is None:
            if self.group is not None:
                state = self._state(self.group)
            else:
                if s.ocr is None:
                    s.ocr = FocusedOCR()
                recover_fullscreen_popup(s)
                state = "field" if s.ocr.field_ready(s.window.capture()) else "unknown"
            self._remember_surface(self.group, state)
        view_require(state in ("local_detail", "remote_detail", "slot_detail", "local", "remote", "field"),
                     "소유한 가공 경로를 확인하지 못해 닫지 않았습니다.")
        with s.coordinate_sequence():
            if state in ("local_detail", "remote_detail", "slot_detail"):
                self._close_detail(self.group)
                state = self.owned_surface(self.group)
            if state in ("local", "remote"):
                self._close_once(self.group,state)
        self._route_owner = self._local_owner = None
        self.group = self.recipe = None
        if self._life is not None:
            self._life.group = self._life.recipe = None

    def visit(self, recipe):
        recipe = self._recipe(recipe)
        self._visiting_recipe = recipe
        self.settle_collection_tail()
        s, group = self.service, recipe["group"]
        s.check()
        if self.local_owned(group):
            self.group, self.recipe = group, recipe
            if self._adaptive():
                life = self._adaptive_life()
                life.group, life.recipe = group, recipe
            return {"group": group, "state": "local"}
        if self._adaptive():
            result=self._adaptive_life().visit(recipe)
            self.group,self.recipe=group,recipe
            return result
        # The start button leaves Mini in front. Match the existing half-window
        # visitor: bring the already-bound game forward before reading its UI.
        # This sends no game key or click and does not re-prepare the service.
        self._focus_for_visit(group)
        if self.group is not None and self.group != group:
            self._leave()
        state = self.owned_surface(group)
        if state is None:
            image = s.window.capture()
            state = self._state(group, image)
            if state == "unknown" and self._leave_other_observed_facility(group, image):
                state = "field"
            self._remember_surface(group, state)
        if state in ("local","remote","field","local_detail","remote_detail","slot_detail","facilities"):
            self._owned_view=True
        if state in ("local_detail", "remote_detail", "slot_detail"):
            self._close_detail(group)
            state = self.owned_surface(group)
        if state != "local":
            view_require(state in ("field", "remote", "facilities"), "기본 화면 또는 선택한 가공 시설 목록을 열어 주세요.")
            if state == "field":
                targets = {"menu": ("key", "K"),
                    "tab": ("click", "processing.tab"),
                    "facility": ("click", "processing.facility")}
                steps = route("processing.visit.fixed", ("menu", "tab", "facility"))
                require(isinstance(steps, (list, tuple)) and 1 <= len(steps) <= 12 and
                        all(isinstance(step, str) and step in targets for step in steps),
                        "가공 준비 선택 규칙의 단계가 잘못되었습니다.")
                with s.coordinate_sequence():
                    context = self._surface_context()
                    for step in steps:
                        kind, target = targets[step]
                        sent = s.processing_input(kind, self._fixed_target(target, recipe) if kind == "click" else target,
                                                  guard=lambda: self._context_guard(context))
                        view_require(sent is not False, "가공 준비 선택 입력 보류")
                    s.coordinate_step()
                state = self._wait(group, ("remote", "local"))
            elif state == "facilities":
                self._retry_facility_selection(group)
                state = self._wait(group, ("remote", "local"))
            if state == "remote":
                self.group, self.recipe = group, recipe
                self._travel(group, self._facility_travel_ref())
                state = "local"
        view_require(state == "local", "가공 현장 목록 미확인 · 등록/수령하지 않았습니다.")
        self.group, self.recipe = group, recipe
        self.recovery_opened(group, "local")
        return {"group": group, "state": "local"}

    def _retry_facility_selection(self, group):
        """Retry an already identified facility list, not a processing commit."""
        selected = getattr(self, "_visiting_recipe", None) or self.recipe
        view_require(selected is not None and selected["group"] == group,
                     "재선택할 시설의 기존 경로 소유권이 없습니다.")
        context = self._surface_context()
        def current():
            return self._context_guard(context) and self.owned_surface(group) == "facilities"
        with self.service.coordinate_sequence():
            if self._adaptive():
                self._adaptive_life().select_facility(group, guard=current)
            else:
                sent = self.service.processing_input("click", self._fixed_target("processing.facility", selected),
                                                     guard=current)
                view_require(sent is not False, "가공 시설 재선택 입력 보류")
            self.service.coordinate_step()
        self.service.status("가공 시설 목록 재확인 · 선택한 시설만 다시 열기 · 등록/수령 재전송 없음")

    def capacity(self, group, rows):
        require(group == self.group, "현재 방문한 가공 시설이 아닙니다.")
        work_counts(rows)
        if self.collection_cleanup() is not None:
            selected = self._cleanup_recipe()
            self.settle_collection_tail()
            if not self.local_owned(group):
                self.visit(selected)
        total = self.facility_capacity(group)
        if len(rows) > total:
            self._facility_levels.pop(group, None)
            raise ProcessingViewUncertain("가공 시설 레벨과 현재 작업 목록을 다시 확인합니다.")
        return total

    def _materials(self, recipe, count, rows, *, partial=False):
        """Validate current allowed stock and optionally return fillable units."""
        require(isinstance(rows, list), "공식 재고 목록 확인 필요")
        source = bool(getattr(self.service, "_source_window", lambda: False)())
        bag_only = (self._adaptive() or self.service._half_window()) and not source
        locations = ("inventory",) if bag_only else ("inventory", "character_storage", "account_storage")
        available = count
        for material in recipe["materials"]:
            name, required = material["name"], material["required"]
            require(type(required) is int and required > 0, "가공 재료 기준 수량이 없습니다.")
            matches = [row for row in rows if isinstance(row, dict) and row.get("DisplayName") == name
                       and row.get("Location") in locations]
            require(len({row.get("Category") for row in matches}) <= 1, "가공 재료 분류 중복: "+name)
            require(all(type(row.get("Count")) is int and row["Count"] >= 0 and type(row.get("IsLocked")) is bool
                        for row in matches), "가공 재료 수량/잠금 미확인: "+name)
            units = sum(row["Count"] for row in matches if not row["IsLocked"])//required
            available = min(available, units)
            if not partial:
                require(units >= count, ("창모드 가공은 가방 재료만 사용합니다. 보관함 재료 확인 팝업이 미확인입니다: "
                        if bag_only else "가방·보관함의 잠금 제외 가공 재료 부족: ")+name)
        return available

    def _bag_jobs(self, recipe, count, stock):
        """One batch-boundary snapshot; never infer a hidden popup is inert."""
        self._materials(recipe, count, stock, partial=True)
        return min([count] + [sum(row["Count"] for row in stock
            if isinstance(row, dict) and row.get("DisplayName") == material["name"]
            and row.get("Location") == "inventory" and row.get("IsLocked") is False)
            // material["required"] for material in recipe["materials"]])

    def settle_registration_tail(self, recipe, receipt, save, *, rows=None):
        """Resolve this one storage registration, never repeat Register.

        A skipped/delayed question remains pending until its exact popup can
        be confirmed or the official queue proves this registration appeared.
        """
        tail = receipt.get("storage_confirmation")
        if not tail:
            return rows
        recipe = self._recipe(recipe)
        require(tail.get("recipe") == recipe["id"] and tail.get("group") == recipe["group"],
                "보관함 재료 확인 기록의 품목/시설이 다릅니다.")
        rows = self.works(recipe["group"]) if rows is None else rows
        work_counts(rows)
        name = catalog.ALTERING_NAMES.get(recipe["id"], recipe["name"])
        if sum(row["DisplayName"] == name for row in rows) > tail["before_named"]:
            receipt.pop("storage_confirmation", None)
            save()
            self._storage_confirmation_owner = None
            return rows
        # 'confirm_attempted' records a sent click, not a proven dismissal.
        # A dropped click must not permanently strand the owned popup. The
        # observer proves the current same storage question before Confirm;
        # no Register is resent and an absent popup receives no extra click.
        sent = self._confirm_storage_registration(recipe, tail, save)
        raise ProcessingReceiptUncertain(
            "보관함 확인 전송 후 등록 결과 재확인 · 등록은 다시 보내지 않습니다." if sent else
            "보관함 재료 전송창 재확인 · 현재 확인창이 보일 때만 확인을 보냅니다.")

    def _confirm_storage_registration(self, recipe, tail, save, *, armed=None):
        from .source_processing_storage import SourceStorageConfirmation
        self.group, self.recipe = recipe["group"], recipe
        key = (recipe["id"], recipe["group"], tail["before_named"])
        previous = getattr(self, "_storage_confirmation_owner", None)
        if armed is None:
            if previous is not None and previous[0] == key:
                _, confirmation, owner = previous
            else:
                confirmation = SourceStorageConfirmation(self.service, processing_port=self)
                owner = confirmation.arm(recipe, recovered=True)
        else:
            confirmation, owner = armed
        self._storage_confirmation_owner = (key, confirmation, owner)
        def attempting():
            tail["confirm_attempted"] = True
            save()
        return confirmation.confirm_if_visible(owner, before_send=attempting, pause=False)

    def _select(self, recipe):
        s = self.service
        context = self._queue_context()
        if self._adaptive():
            return self._adaptive_life().select(recipe)
        sent = s.processing_input("click", self._fixed_target("processing.recipe", recipe),
                                  guard=lambda: self._tape_guard(recipe["group"], context))
        view_require(sent is not False, "가공 품목 선택 입력 보류")
        self.remember_selection(recipe)
        self.recovery_opened(recipe["group"], "local_detail")

    def register(self, recipe, count, *, before_send=None, rows=None, stock=None, capacity=None):
        recipe = self._recipe(recipe)
        require(type(count) is int and 1 <= count <= 7, "가공 등록 건수 오류")
        if self._single_start_pending and self.ensure_start_local(recipe["group"], self.service.window.capture()):
            rows = stock = capacity = None
        if not self.local_owned(recipe["group"]):
            self.visit(recipe)
            # A recovery visit can outlive the caller's boundary observations.
            rows = stock = capacity = None
        self.group, self.recipe = recipe["group"], recipe
        total = self.facility_capacity(recipe["group"])
        before = self.works(recipe["group"]) if rows is None else rows
        work_counts(before)
        if len(before) > total:
            self._facility_levels.pop(recipe["group"], None)
            raise ProcessingViewUncertain("가공 시설 레벨과 등록 전 작업 목록 재확인 필요")
        if self._adaptive():
            return self._register_adaptive(recipe, count, before, before_send, stock=stock, capacity=capacity)
        return self._register_current(recipe, count, before, before_send, adaptive=False,
                                      stock=stock, capacity=capacity)

    @staticmethod
    def _registration_result(recipe, before, after, attempted):
        old, new = work_counts(before), work_counts(after)
        expected = catalog.ALTERING_NAMES.get(recipe["id"], recipe["name"])
        registered = sum(amount for (_, name), amount in (new-old).items() if name == expected)
        return {"before": before, "after": after, "registered": registered, "attempted": attempted}

    def _register_adaptive(self, recipe, count, before, before_send, *, stock=None, capacity=None):
        return self._register_current(recipe, count, before, before_send, adaptive=True,
                                      stock=stock, capacity=capacity)

    def _register_current(self, recipe, count, before, before_send, *, adaptive, stock=None, capacity=None):
        """Bag-only authored tape, or one owned storage-question exception.

        Queue, actual free slots and stock are read at the batch boundary.
        No per-slot visit/OCR/CLI gate is inserted into the successful tape;
        a partial result is handed to existing missing-registration recovery.
        """
        s, group = self.service, recipe["group"]
        life = self._adaptive_life() if adaptive else None
        source = adaptive
        context = self._queue_context()
        stock = self.inventory() if stock is None else stock
        capacity = self.capacity(group, before) if capacity is None else capacity
        capacity = min(capacity, self.facility_capacity(group))
        free = max(0, capacity - len(before))
        jobs = min(count, free, self._materials(recipe, count, stock, partial=True))
        bag_jobs = self._bag_jobs(recipe, jobs, stock)
        storage_required = jobs > 0 and bag_jobs == 0
        # Finish the bag prefix as one normal tape. A storage question is a
        # separate, bounded exception; never click through an absent popup.
        jobs = 1 if storage_required else bag_jobs
        # One existing batch-boundary stock snapshot, not another CLI query or
        # an OCR gate inside the normal bag-material coordinate tape.
        material_notes = []
        for material in recipe["materials"]:
            amounts = []
            for location, label in (("inventory", "가방"), ("character_storage", "본인 보관함"),
                                    ("account_storage", "계정 보관함")):
                matches = [row for row in stock if isinstance(row, dict)
                           and row.get("DisplayName") == material["name"]
                           and row.get("Location") == location and row.get("IsLocked") is False]
                quantity = (str(sum(row["Count"] for row in matches))
                            if all(type(row.get("Count")) is int and row["Count"] >= 0 for row in matches)
                            else "미확인")
                amounts.append(label + " " + quantity)
            material_notes.append(material["name"] + " [" + ", ".join(amounts) + "]")
        s.status("가공 재료 분리 · " + " / ".join(material_notes)
                 + f" · 이번 {jobs}건 · " + ("보관함 전송 확인 필요" if storage_required else "가방 재료 묶음"))
        attempted = 0
        storage_tail = None
        storage_arm = None
        if storage_required:
            from .source_processing_storage import SourceStorageConfirmation
            confirmation = SourceStorageConfirmation(s, processing_port=self)
        with s.coordinate_sequence():
            for _ in range(jobs):
                s.check()
                self._tape_guard(group, context)
                self._select(recipe)
                target = (life.tape_target("register", recipe) if source else
                          self._fixed_target("processing.register", recipe))
                def registration_sent():
                    nonlocal storage_arm
                    self._selection_owner = None
                    if attempted == 0:
                        self.registration_before = deepcopy(before)
                        if before_send is not None:
                            before_send()
                    if storage_required:
                        storage_arm = (confirmation, confirmation.arm(recipe))
                        name = catalog.ALTERING_NAMES.get(recipe["id"], recipe["name"])
                        self.registration_storage_tail = dict(recipe=recipe["id"], group=group,
                            before_named=sum(row["DisplayName"] == name for row in before),
                            confirm_attempted=False)
                        getattr(self, "registration_tail_update", lambda _: None)(self.registration_storage_tail)
                sent = s.processing_input("click", target, guard=lambda: self._tape_guard(group, context),
                                          before_send=registration_sent)
                if sent is False:
                    break
                attempted += 1
                self.recovery_opened(group, "local")
            # One settled boundary after the complete tape, not a queue/OCR
            # approval after each of the seven registrations.
            s.coordinate_step()
        if storage_required and attempted:
            storage_tail = self.registration_storage_tail
            def save_tail():
                getattr(self, "registration_tail_update", lambda _: None)(storage_tail)
            self._confirm_storage_registration(recipe, storage_tail, save_tail, armed=storage_arm)
        after = self.works(group) if attempted else before
        if storage_tail is not None:
            temporary = {"storage_confirmation": storage_tail}
            self.settle_registration_tail(recipe, temporary,
                lambda: getattr(self, "registration_tail_update", lambda _: None)(temporary.get("storage_confirmation")),
                rows=after)
        return self._registration_result(recipe, before, after, attempted)

    def collect(self, group, rows, *, before_send=None, allow_partial=False):
        self.settle_collection_tail()
        require(group == self.group and self.recipe is not None, "수령할 가공 시설을 먼저 방문해 주세요.")
        work_counts(rows)
        # A single start may already have seven completed jobs and therefore
        # reach Collect without capacity(). Resolve proximity OUTSIDE its
        # two-click tape, never on the first consuming press.
        if self._single_start_pending and self.ensure_start_local(group, self.service.window.capture()):
            rows = self.works(group)
        before = rows
        def collectable(work):
            return bool(work) and (any(row["IsCompleted"] for row in work) if allow_partial
                                   else all(row["IsCompleted"] for row in work))
        if not collectable(before):
            return {"before": before, "after": before, "collected": 0, "attempted": 0}
        if not self.local_owned(group):
            self.visit(self.recipe)
            before = self.works(group)
            if not collectable(before):
                return {"before": before, "after": before, "collected": 0, "attempted": 0}
        s = self.service
        capacity = self.facility_capacity(group)
        if len(before) > capacity:
            self._facility_levels.pop(group, None)
            raise ProcessingViewUncertain("가공 시설 레벨과 수령 작업 목록 재확인 필요")
        context = self._queue_context()
        def collection_sent():
            self.collection_before = deepcopy(before)
            if before_send is not None:
                before_send()
            self.recovery_opened(group, "slot_detail")
        source = self._adaptive()
        if source:
            target = self._adaptive_life().tape_target("collect")
        elif capacity <= 4:
            runtime, _ = self._source_runtime()
            target = runtime.prepared_ref("processing.collect", "processing.collect_all", queue_count=capacity,
                facility_ready=True, full_safe_area=True, collect_enabled=True,
                move_visible=False, upgrade_visible=False)
        else:
            target = self._fixed_target("processing.collect_all")
        with s.coordinate_sequence():
            sent = s.processing_input("click", target,
                guard=lambda: self._tape_guard(group, context), before_send=collection_sent)
            if sent is not False:
                # The user-observed reward popup needs longer than the normal
                # 0.5s tape cadence. Wait from the actual Collect press, not an
                # extra full delay after its return, and keep cancellation live.
                interval = number("processing.collect.confirm_interval_seconds", 1.5)
                pressed_at = getattr(s, "_coordinate_last_event", None)
                elapsed = 0.0 if pressed_at is None else max(0.0, time.monotonic()-pressed_at)
                if interval > elapsed:
                    s.sleep(interval-elapsed)
                confirm = (self._adaptive_life().tape_target("result_confirm") if source else
                           self._fixed_target("processing.result.confirm"))
                confirmed = s.processing_input("click", confirm, guard=lambda: self._tape_guard(group, context))
                view_require(confirmed is not False, "수령 결과 확인 입력 보류 · 기존 수령 기록에서 복구")
                s.coordinate_step()
                self.recovery_opened(group, "local")
        after = self.works(group)
        removed = sum((work_counts(before)-work_counts(after)).values()) if sent is not False else 0
        if removed:
            self.remember_collected_result(self.recipe)
        return {"before": before, "after": after, "collected": removed,
                "attempted": int(sent is not False), "confirmed": sent is not False}

    def finish_collection(self, recipe, *, observed_state=None):
        """Dismiss only a runner-proven collected batch's remaining reward UI.

        Caller must first reconcile its durable collect receipt against fresh
        official jobs, including a partial collection. This never sends collect, registers, or navigates
        away from an unknown screen.
        """
        recipe = self._recipe(recipe)
        group, s = recipe["group"], self.service
        state = self._state(group) if observed_state is None else observed_state
        if state == "slot_detail":
            self.close_misdirected_slot(group)
            state = self._wait(group, ("local", "local_detail", "field"))
        while state == "reward":
            if self._adaptive():
                confirm = s._ui_route_runtime().prepared_ref("processing.result", "processing.result.confirm", settled=True)
            else:
                confirm = self._fixed_target("processing.result.confirm", recipe)
            context = self._surface_context()
            self._remember_surface(group, "reward")
            s.processing_input("click", confirm, guard=lambda: self._context_guard(context)
                and self.owned_surface(group) == "reward")
            state = self._wait(group, ("local", "local_detail", "field", "reward"), result_cleanup=True)
        view_require(state in ("local", "local_detail", "field"),
                "이전 수령의 완료 팝업/현장 복귀를 확인하지 못했습니다. 모두 받기를 다시 누르지 않습니다.")
        self.group = None if state == "field" else group
        self.recipe = None if state == "field" else recipe
        self._single_start_pending = False
        self.recovery_opened(group, state)
        if self._life is not None:
            self._life.group, self._life.recipe = self.group, self.recipe
        self._clear_collection_cleanup(group)
        return {"group": group, "state": state, "collect_replayed": False}

    def gather_progress(self, operation_id):
        with self.service.store.lock:
            return deepcopy(self.service.store.value.get(ChildGatherStore.OPERATIONS, {}).get(operation_id))

    def gather(self, node, item, quantity, *, operation_id=None):
        require(type(quantity) is int and quantity > 0, "부족분 채집 수량 오류")
        require(item in catalog.gather_output_options(node), "조사된 채집 산출물과 부족 품목이 다릅니다.")
        parent = self.service
        # Persist the original quota before any route/input. A completed child
        # remains this same operation even if inventory was consumed afterwards.
        child_store = ChildGatherStore(parent.store, operation_id=operation_id, node_id=node["id"],
                                       item=item, quantity=quantity)
        record = self.gather_progress(operation_id) if operation_id is not None else None
        done = bool(record and (record["confirmed"] >= quantity or record["state"] == "complete"))
        if record and record["child"] and not done:
            # A cancelled child may still have a native gather batch running.
            # Observe its end before any route/next-facility input, never restart it.
            from .automation import AutomationService
            observer = AutomationService(parent.connector, child_store,
                                         status=parent._ui_status, repeat_range=parent.repeat_range)
            observer.window, observer.ocr, observer.cancel = parent.window, parent.ocr, parent.cancel
            observer.before_game_input = getattr(parent, "before_game_input", None)
            observer._wait_idle()
        if record and not done:
            current = parent.connector.item_quantity(item)
            require(type(current) is int and current >= 0, "현재 수급 품목 수량 형식이 올바르지 않습니다.")
            #Stock can be consumed while the child is paused. Accumulated
            #positive gains and the original quota survive that decrease;
            #AutomationService owns last_quantity rebaselining on observation.
        if not done:
            if not self._owned_view:
                # Initial global supply precedes the first facility visit. Mini may
                # still be foreground: prepare observation without game input.
                parent.check()
                from .windows import wait_input_idle
                from .automation import wait_route_input_idle
                wait_route_input_idle(parent, wait_input_idle)
                parent.check()
                parent.window.focus()
                image = parent.window.capture()
                if parent.ocr is None:
                    parent.ocr = FocusedOCR()
                state = ("field" if parent.ocr.field_ready(image) else "unknown") if self.group is None else self._state(self.group, image)
                require(state in ("field", "local", "remote", "local_detail", "remote_detail", "slot_detail"),
                        "최초 부족분 수급은 메뉴 없는 기본 화면에서 시작해 주세요. 미확인 화면에 입력하지 않았습니다.")
                self._remember_surface(self.group, state)
                self._owned_view = True
            self._leave()
        else:
            parent.check()
            from .windows import wait_input_idle
            from .automation import wait_route_input_idle
            wait_route_input_idle(parent, wait_input_idle)
            parent.check()
            parent.window.focus()
        from .automation import AutomationService, _input_is_owned
        from .windows import last_input_stamp
        child = AutomationService(parent.connector, child_store,
                                  status=parent._ui_status, repeat_range=parent.repeat_range)
        child.window, child.ocr, child.cancel = parent.window, parent.ocr, parent.cancel
        child.before_game_input = getattr(parent, "before_game_input", None)
        # The authored facility close already established this field boundary.
        # Reuse it until foreign input invalidates it; no duplicate HUD OCR.
        child._processing_field_owner = lambda: self.owned_surface() == "field"
        if _input_is_owned(parent, last_input_stamp()):
            child._last_own_input_stamp = parent._last_own_input_stamp
            if getattr(parent, "_coordinate_last_event", None) is not None:
                child._coordinate_last_event = parent._coordinate_last_event
        child.prepare = child.check
        acquired = record["confirmed"] if done else child.gather(
            node, quantity, goal_kind="quantity", output_item=item)
        # A cancellation here must not cause the already-complete child to run
        # again. Idle/HUD confirmation is observation only, never another route.
        child._wait_idle()
        parent.ocr = child.ocr
        child_owns_input = _input_is_owned(child, last_input_stamp())
        if done or not child_owns_input:
            # A resumed old completion does not own the user's current UI.
            view_require(parent.ocr is not None and parent.ocr.field_ready(parent.window.capture()),
                    "부족분 채집 재개 후 기본 화면을 확인하지 못했습니다.")
        if child_owns_input:
            parent._last_own_input_stamp = child._last_own_input_stamp
            if getattr(child, "_coordinate_last_event", None) is not None:
                parent._coordinate_last_event = child._coordinate_last_event
        # A just-finished authored gather remains in the field after the
        # actual activity-idle read. The next facility route need not OCR it
        # once here and again before K.
        self._remember_surface(None, "field")
        self.group = self.recipe = None
        self._route_owner = self._local_owner = None
        if operation_id is not None:
            record = self.gather_progress(operation_id)
            require(record is not None and record["confirmed"] >= quantity,
                    "채집 확정 진행 기록이 없어 새 채집 없이 같은 작업을 보존합니다.")
            acquired = record["confirmed"]
        return {"acquired": acquired}
