#!/usr/bin/env python3
"""
HARDWICK TRANSLATE - Lean Argos Translate Socket Server

Direct replacement for LibreTranslate Docker container (AGPL → MIT).
Uses argostranslate directly over Unix socket, same pattern as frankenstein-embeddings.py.

Protocol (newline-delimited JSON):
  Request:  {"q": "word1\nword2", "source": "en", "target": "zh"}
  Response: {"translatedText": "翻译1\n翻译2"}
  Health:   {"q": "__health_check__", "source": "en", "target": "zh"}
           → {"translatedText": "ok", "status": "healthy"}

@author hardwicksoftwareservices
"""

import signal
import sys
import os
import socket
import json
import time
import re
import unicodedata
import argparse
import threading
import subprocess
from functools import lru_cache
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass


def _ensure_deps():
    """Auto-install dependencies if missing."""
    needed = []
    for mod, pkg in [('argostranslate', 'argostranslate'), ('ctranslate2', 'ctranslate2'),
                     ('sentencepiece', 'sentencepiece'), ('emoji', 'emoji'),
                     ('opencc', 'opencc-python-reimplemented')]:
        try:
            __import__(mod)
        except ImportError:
            needed.append(pkg)
    if needed:
        print(f"⏳ Installing missing deps: {needed}", file=sys.stderr)
        _pip_cmd = [sys.executable, '-m', 'pip', 'install', '--break-system-packages', '-q']
        if os.getuid() != 0:
            _pip_cmd.append('--user')
        try:
            subprocess.check_call(_pip_cmd + needed)
            print(f"✅ Deps installed", file=sys.stderr)
        except Exception as e:
            # opencc may fail in air-gapped container — non-fatal
            print(f"⚠️ Some deps failed to install (ok if air-gapped): {e}", file=sys.stderr)

_ensure_deps()

# ============================================================================
# NETWORK ISOLATION — prevent argostranslate from phoning home
# Argos internally tries to check for updates/download packages which fails
# with DNS errors in containerized/air-gapped environments. We ship packed
# models, so disable ALL network access at the library level.
# ============================================================================
os.environ['ARGOS_DEVICE_TYPE'] = 'cpu'
# Set ARGOS_PACKAGES_DIR early — MUST be set before argostranslate is imported
# because argostranslate.settings computes package_dirs at module import time.
# Use the writable XDG data dir so argostranslate can register package metadata.
if not os.environ.get('ARGOS_PACKAGES_DIR'):
    _xdg_data = os.environ.get('XDG_DATA_HOME', os.path.expanduser('~/.local/share'))
    _early_argos_dir = os.path.join(_xdg_data, 'argos-translate', 'packages')
    os.makedirs(_early_argos_dir, exist_ok=True)
    os.environ['ARGOS_PACKAGES_DIR'] = _early_argos_dir
# Monkey-patch urllib to prevent argostranslate from phoning home for updates
# BUT: allow local/internal operations (sentencepiece model loading uses urllib internally)
import urllib.request
_original_urlopen = urllib.request.urlopen
_urlopen_blocked = True  # Start blocked, lifted during translation

def _guarded_urlopen(*args, **kwargs):
    """Block external network calls but allow local file/internal operations."""
    if _urlopen_blocked:
        url_str = str(args[0]) if args else ''
        # Allow file:// URLs and localhost — block everything else
        if url_str.startswith('file://') or '127.0.0.1' in url_str or 'localhost' in url_str:
            return _original_urlopen(*args, **kwargs)
        raise ConnectionError("Network access disabled — using packed models only")
    return _original_urlopen(*args, **kwargs)

urllib.request.urlopen = _guarded_urlopen

# Also patch urlretrieve (used by stanza/other libs for model downloads)
_original_urlretrieve = urllib.request.urlretrieve
def _guarded_urlretrieve(url, *args, **kwargs):
    """Block external downloads but allow local file operations."""
    if _urlopen_blocked:
        url_str = str(url)
        if url_str.startswith('file://') or '127.0.0.1' in url_str or 'localhost' in url_str:
            return _original_urlretrieve(url, *args, **kwargs)
        raise ConnectionError(f"Network download disabled (air-gapped): {url_str[:80]}")
    return _original_urlretrieve(url, *args, **kwargs)
urllib.request.urlretrieve = _guarded_urlretrieve

# Also patch requests library (used by stanza for downloading resources/models)
try:
    import requests as _requests
    _original_requests_get = _requests.get
    def _guarded_requests_get(url, *args, **kwargs):
        if _urlopen_blocked:
            url_str = str(url)
            if url_str.startswith('file://') or '127.0.0.1' in url_str or 'localhost' in url_str:
                return _original_requests_get(url, *args, **kwargs)
            raise ConnectionError(f"Network download disabled (air-gapped): {url_str[:80]}")
        return _original_requests_get(url, *args, **kwargs)
    _requests.get = _guarded_requests_get
except ImportError:
    pass

# Patch stanza to skip network downloads — use pre-cached resources + regex SBD
try:
    import stanza.resources.common as _stanza_res
    _original_download_resources = _stanza_res.download_resources_json
    def _offline_download_resources(model_dir=None, **kwargs):
        """Skip download if resources.json already exists."""
        import os
        if model_dir is None:
            model_dir = _stanza_res.DEFAULT_MODEL_DIR
        fpath = os.path.join(model_dir, 'resources.json')
        if os.path.exists(fpath):
            return  # Already cached, skip download
        # Try original (will fail in air-gapped)
        return _original_download_resources(model_dir=model_dir, **kwargs)
    _stanza_res.download_resources_json = _offline_download_resources
    # Also patch the import binding in stanza.pipeline.core (already imported)
    try:
        import stanza.pipeline.core as _stanza_core
        _stanza_core.download_resources_json = _offline_download_resources
    except (ImportError, AttributeError):
        pass
except ImportError:
    pass

# IMPORTANT: Register fake minisbd BEFORE any argostranslate imports
# (argostranslate.sbd does "from minisbd import SBDetect" at import time)
import types as _types_early

class _RegexSBDetect:
    """Regex-based sentence boundary detection — replaces AGPL minisbd."""
    def __init__(self, lang='en', use_gpu=False):
        self.lang = lang
    def sentences(self, text):
        if not text or not text.strip():
            return [text] if text else []
        parts = re.split(r'(?<=[.!?。！？\u2026])\s+', text)
        result = []
        for part in parts:
            result.extend(part.split('\n'))
        return [s for s in result if s.strip()]

class _FakeModels:
    cache_dir = '/tmp/hardwick-sbd'
    @staticmethod
    def list_models():
        return ['en', 'zh', 'zh-hans', 'zh-hant', 'es', 'fr', 'de', 'ja', 'ko', 'pt']

_fake_minisbd = _types_early.ModuleType('minisbd')
_fake_minisbd.SBDetect = _RegexSBDetect
_fake_models = _types_early.ModuleType('minisbd.models')
_fake_models.cache_dir = _FakeModels.cache_dir
_fake_models.list_models = _FakeModels.list_models
_fake_minisbd.models = _fake_models
sys.modules['minisbd'] = _fake_minisbd
sys.modules['minisbd.models'] = _fake_models
print(f"[FAKE_MINISBD] Registered: SBDetect={_fake_minisbd.SBDetect.__name__}", file=sys.stderr, flush=True)

# Patch StanzaSentencizer to skip network downloads (air-gapped container)
print("[STANZA_NUKE] Starting patch...", file=sys.stderr, flush=True)
try:
    import argostranslate.sbd as _asbd
    # Replace StanzaSentencizer entirely with MiniSBD — avoids stanza network downloads
    # in air-gapped containers. MiniSBD regex is sufficient for our use case.
    # Monkey-patch the CLASS methods directly — this works regardless of import binding
    # (patching module attributes doesn't affect `from X import Y` local bindings)
    def _mini_sbd_split(self, text):
        """Use regex SBD instead of stanza — no network needed."""
        if not text or not text.strip():
            return [text] if text else []
        parts = re.split(r'(?<=[.!?。！？\u2026])\s+', text)
        result = []
        for part in parts:
            result.extend(part.split('\n'))
        return [s for s in result if s.strip()]
    _asbd.StanzaSentencizer.split_sentences = _mini_sbd_split
    _asbd.StanzaSentencizer.lazy_pipeline = lambda self: None  # Never create stanza pipeline
    print("[PATCH] StanzaSentencizer.split_sentences patched to regex SBD", file=sys.stderr, flush=True)
except Exception as e:
    print(f"[PATCH] Failed to patch sbd module: {type(e).__name__}: {e}", file=sys.stderr)
    import traceback; traceback.print_exc(file=sys.stderr)

# (minisbd fake module already registered above - before argos imports)

# ============================================================================
# CPU Thread Governance — matches frankenstein-embeddings.py pattern
# Priority: SPECMEM_CPU_THREADS env → user-config.json resources.cpuCoreMax → default 1
# ============================================================================

PROJECT_PATH = os.environ.get('SPECMEM_PROJECT_PATH', os.getcwd())

def _get_cpu_thread_limit():
    """Get CPU thread limit from env or user-config.json"""
    if os.environ.get('SPECMEM_CPU_THREADS'):
        return int(os.environ['SPECMEM_CPU_THREADS'])
    try:
        config_path = os.path.join(PROJECT_PATH, 'specmem', 'user-config.json')
        if os.path.exists(config_path):
            with open(config_path, 'r') as f:
                config = json.load(f)
                core_max = config.get('resources', {}).get('cpuCoreMax')
                if core_max is not None:
                    return int(core_max)
    except Exception as e:
        print(f"⚠️ Could not read CPU core limit from config: {e}", file=sys.stderr)
    return 1

_CPU_THREAD_LIMIT = _get_cpu_thread_limit()
_CPU_THREAD_MIN = 1  # Crawl/idle mode always 1 thread

# Set thread limits at module level
os.environ.setdefault('OMP_NUM_THREADS', str(_CPU_THREAD_MIN))
os.environ.setdefault('MKL_NUM_THREADS', str(_CPU_THREAD_MIN))
os.environ.setdefault('OPENBLAS_NUM_THREADS', str(_CPU_THREAD_MIN))
os.environ.setdefault('CT2_COMPUTE_TYPE', 'int8')
print(f"🔒 Hardwick Translate CPU threads: {_CPU_THREAD_MIN}-{_CPU_THREAD_LIMIT} (crawl/active)", file=sys.stderr)


@dataclass
class TranslateResourceConfig:
    """Resource governance for translation service — matches frankenstein pattern."""
    cpu_min: float = float(os.environ.get('SPECMEM_CPU_MIN', '20'))
    cpu_max: float = float(os.environ.get('SPECMEM_CPU_MAX', '40'))
    ram_min_mb: float = float(os.environ.get('SPECMEM_RAM_MIN_MB', '4000'))
    ram_max_mb: float = float(os.environ.get('SPECMEM_RAM_MAX_MB', '6000'))

    # Crawl = idle, Active = during translation request
    crawl_threads: int = _CPU_THREAD_MIN
    active_threads: int = _CPU_THREAD_LIMIT

    # Idle unload: release model after N seconds idle, reload on next request
    idle_unload_seconds: int = int(os.environ.get('SPECMEM_TRANSLATE_IDLE_UNLOAD', '600'))

    def get_crawl_threads(self) -> int:
        return self.crawl_threads

    def get_active_threads(self) -> int:
        return min(self.active_threads, os.cpu_count() or 2)

# Ignore SIGPIPE
signal.signal(signal.SIGPIPE, signal.SIG_IGN)

# Language aliases (Argos uses full codes internally)
# 'zt' = Traditional Chinese — falls back to Simplified Chinese model
# (Claude reads both equally well, and zh-Hans model is what we ship)
LANG_ALIASES = {
    'zh': 'zh-Hans',
    'zt': 'zh-Hant',
}

# Fallback chain: if primary alias not found, try these
LANG_FALLBACKS = {
    'zh-Hant': ['zh-Hans', 'zh'],  # Traditional → Simplified → bare zh
    'zh-Hans': ['zh'],
}

# Model directory search order — packed models first, never download
_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
_SPECMEM_MODEL_CACHE = os.environ.get('SPECMEM_MODEL_CACHE', '/tmp/specmem-hardwick-models')
_MODEL_SEARCH_PATHS = [
    os.path.join(_SCRIPT_DIR, 'models', 'argos-translate'),  # Packed with npm package
    f'{_SPECMEM_MODEL_CACHE}/share/argos-translate/packages',
    os.path.expanduser('~/.local/share/argos-translate/packages'),
]

def _find_model_dir():
    """Find first existing model directory with actual model folders."""
    for p in _MODEL_SEARCH_PATHS:
        if os.path.isdir(p) and any(d.startswith('translate-') for d in os.listdir(p)):
            return p
    return _MODEL_SEARCH_PATHS[0]

DEFAULT_MODEL_DIR = _find_model_dir()

# Emoji detection — use Python's emoji category check
def _is_emoji_only(text):
    """Check if text is only emoji/whitespace/punctuation (nothing translatable)."""
    import emoji as emoji_lib
    stripped = text.strip()
    if not stripped:
        return True
    # Remove all emoji characters
    demojized = emoji_lib.replace_emoji(stripped, replace='')
    # Check if anything meaningful remains
    remaining = demojized.strip()
    if not remaining:
        return True
    for ch in remaining:
        cat = unicodedata.category(ch)
        if cat[0] not in ('P', 'Z', 'S', 'C'):
            return False
    return True


def detect_translatable(text):
    """Check if text contains translatable content (not just emoji/whitespace)."""
    return not _is_emoji_only(text)


def improve_translation_formatting(source, translation):
    """
    Adapted from LibreTranslate's language.py — preserve formatting from source.
    Handles: punctuation preservation, case fixing, salad bug.
    """
    if not source or not translation:
        return translation

    # Salad bug: model repeats a single word for short inputs
    # e.g. "hello" → "你好 你好 你好"
    if len(source.split()) <= 2:
        words = translation.split()
        if len(words) > 1 and len(set(words)) == 1:
            translation = words[0]

    # Preserve trailing punctuation from source
    src_trailing = ''
    for ch in reversed(source):
        if unicodedata.category(ch).startswith('P'):
            src_trailing = ch + src_trailing
        else:
            break

    if src_trailing:
        # Strip existing trailing punct from translation, add source's
        trans_stripped = translation.rstrip()
        while trans_stripped and unicodedata.category(trans_stripped[-1]).startswith('P'):
            trans_stripped = trans_stripped[:-1]
        if trans_stripped:
            translation = trans_stripped + src_trailing

    # Case preservation
    if source and source[0].islower() and translation and translation[0].isupper():
        # Source starts lowercase but translation starts uppercase — fix
        # Only if source language uses case (Latin scripts)
        if source[0].isascii():
            translation = translation[0].lower() + translation[1:]
    elif source and source[0].isupper() and translation and translation[0].islower():
        if translation[0].isascii():
            translation = translation[0].upper() + translation[1:]

    return translation


class HardwickTranslate:
    """Unix socket translation server using Argos Translate directly."""
    # TODO: This class is 400+ lines - consider splitting into separate modules (model_loader.py, socket_handler.py, etc.)

    def __init__(self, socket_path, model_dir=None):
        self.socket_path = socket_path
        self.model_dir = model_dir or DEFAULT_MODEL_DIR
        self.shutdown_requested = False
        self.warm_restart = False
        self.last_request_time = time.time()

        # Lazy-loaded translation models
        self._models_loaded = False
        self._load_lock = threading.Lock()
        self._installed_languages = None
        self.resource_config = TranslateResourceConfig()
        self._idle_monitor_thread = None
        self.idle_timeout = self.resource_config.idle_unload_seconds

        # Set up model directory
        # CRITICAL: ARGOS_PACKAGES_DIR must point to a WRITABLE directory where
        # argostranslate can register packages. We symlink packed models there.
        # Using the read-only bind mount directly causes get_installed_languages()
        # to return [] because argostranslate can't write its internal registry.
        xdg_data = os.environ.get('XDG_DATA_HOME',
                                   os.path.expanduser('~/.local/share'))
        argos_packages_dir = os.path.join(xdg_data, 'argos-translate', 'packages')
        os.makedirs(argos_packages_dir, exist_ok=True)
        os.environ['ARGOS_PACKAGES_DIR'] = argos_packages_dir
        # Also set XDG for argostranslate's internal paths
        os.environ.setdefault('XDG_DATA_HOME', xdg_data)

        # Signal handlers
        signal.signal(signal.SIGHUP, self._handle_sighup)
        signal.signal(signal.SIGTERM, self._handle_sigterm)
        signal.signal(signal.SIGINT, self._handle_sigterm)

    def _start_idle_monitor(self):
        """Monitor idle time and unload models when not in use."""
        def _monitor():
            while not self.shutdown_requested:
                time.sleep(30)  # Check every 30s
                if self.idle_timeout <= 0:
                    continue
                idle_time = time.time() - self.last_request_time
                if idle_time > self.idle_timeout and self._models_loaded:
                    print(f"💤 Idle for {idle_time:.0f}s — unloading models to save RAM", file=sys.stderr)
                    self._unload_models()

        self._idle_monitor_thread = threading.Thread(target=_monitor, daemon=True, name="translate-idle")
        self._idle_monitor_thread.start()

    def _unload_models(self):
        """Unload translation models to free RAM."""
        with self._load_lock:
            self._models_loaded = False
            self._installed_languages = None
            self._cached_translate.cache_clear()
            import gc
            gc.collect()
            print(f"✅ Models unloaded — will reload on next request", file=sys.stderr)

    def _handle_sighup(self, signum, frame):
        """Warm restart — clear cache, keep models loaded."""
        print(f"♻️  SIGHUP received — warm restart", file=sys.stderr)
        self.warm_restart = True
        self._clear_cache()

    def _handle_sigterm(self, signum, frame):
        """Graceful shutdown."""
        print(f"🛑 SIGTERM received — shutting down", file=sys.stderr)
        self.shutdown_requested = True

    def _clear_cache(self):
        """Clear translation cache."""
        self._cached_translate.cache_clear()
        print(f"   Cache cleared", file=sys.stderr)

    def _ensure_models(self):
        """Lazy-load Argos Translate models on first real request."""
        if self._models_loaded:
            return

        with self._load_lock:
            if self._models_loaded:
                return

            global _urlopen_blocked
            print(f"⏳ Loading Argos Translate models...", file=sys.stderr)
            start = time.time()

            try:
                # Temporarily lift network block for model discovery
                # (argostranslate uses urllib internally for package metadata)
                _urlopen_blocked = False

                import argostranslate.translate
                import argostranslate.package

                # Check if models are already installed
                self._installed_languages = argostranslate.translate.get_installed_languages()
                lang_codes = [l.code for l in self._installed_languages]
                print(f"   Installed languages: {lang_codes}", file=sys.stderr)

                # Check we have en, zh, and zt
                has_en = any(l.code == 'en' for l in self._installed_languages)
                has_zh = any(l.code in ('zh', 'zh-Hans') for l in self._installed_languages)
                has_zt = any(l.code in ('zt', 'zh-Hant') for l in self._installed_languages)

                if not has_en or not has_zh or not has_zt:
                    # Register packed models — NO downloads, internalized only
                    self._register_packed_models()
                    # MUST clear lru_cache — get_installed_languages() uses @lru_cache
                    # and the first call (above) cached []. Without clearing, the second
                    # call returns the stale cached [] even after symlinks are created.
                    argostranslate.translate.get_installed_languages.cache_clear()
                    self._installed_languages = argostranslate.translate.get_installed_languages()
                    lang_codes = [l.code for l in self._installed_languages]
                    has_en = any(l.code == 'en' for l in self._installed_languages)
                    has_zh = any(l.code in ('zh', 'zh-Hans') for l in self._installed_languages)
                    if not has_en or not has_zh:
                        raise RuntimeError(f"Packed models not found in {self.model_dir}. Models must be shipped with the package — no downloads.")

                elapsed = time.time() - start
                print(f"✅ Models loaded in {elapsed:.1f}s", file=sys.stderr)
                self._models_loaded = True

            except ConnectionError as e:
                # DNS/network failure — models not available offline
                print(f"⚠️ Network error during model load (expected in container): {e}", file=sys.stderr)
                print(f"   Translation will use simple fallback (no semantic splitting)", file=sys.stderr)
                self._models_loaded = False  # Will use fallback path
            except Exception as e:
                print(f"❌ Model loading failed: {e}", file=sys.stderr)
                raise
            finally:
                # Re-enable network block
                _urlopen_blocked = True

    def _register_packed_models(self):
        """Register packed model directories so argostranslate can find them."""
        import argostranslate.package
        model_dir = self.model_dir
        if not os.path.isdir(model_dir):
            return
        # argostranslate.package.install_from_path() is the correct API — it
        # copies/installs the package into ARGOS_PACKAGES_DIR and updates the
        # internal registry. Symlinks alone are not enough because argostranslate
        # maintains a separate metadata index that must be updated via install.
        for d in os.listdir(model_dir):
            if not d.startswith('translate-'):
                continue
            src = os.path.join(model_dir, d)
            # Fix double-nesting: if dir contains only a subdir with same name
            # (e.g. translate-en_zt-1_9/translate-en_zt-1_9/metadata.json)
            inner = os.path.join(src, d)
            if (not os.path.isfile(os.path.join(src, 'metadata.json'))
                    and os.path.isdir(inner)
                    and os.path.isfile(os.path.join(inner, 'metadata.json'))):
                print(f"   Fixing double-nested model: {d}/{d} → {d}", file=sys.stderr)
                src = inner
            # Check if already installed (avoid reinstalling on every restart)
            argos_dir = os.environ.get('ARGOS_PACKAGES_DIR', '')
            dst = os.path.join(argos_dir, d) if argos_dir else ''
            if dst and os.path.exists(dst):
                print(f"   Already installed: {d}", file=sys.stderr)
                continue
            print(f"   Registering packed model: {d}", file=sys.stderr)
            try:
                argostranslate.package.install_from_path(src)
            except Exception as e:
                print(f"   install_from_path failed for {d}: {e}, falling back to symlink", file=sys.stderr)
                if dst and not os.path.exists(dst):
                    os.makedirs(os.path.dirname(dst), exist_ok=True)
                    os.symlink(src, dst)

    def _find_lang(self, code):
        """Find an installed language by code, with alias + fallback chain."""
        # Build search order: exact → alias → fallbacks → base code
        candidates = [code]
        alias = LANG_ALIASES.get(code)
        if alias:
            candidates.append(alias)
            # Add fallbacks for the alias (e.g. zh-Hant → zh-Hans → zh)
            candidates.extend(LANG_FALLBACKS.get(alias, []))
        # Add base code (e.g. 'zh-Hans' → 'zh')
        base = code.split('-')[0]
        if base not in candidates:
            candidates.append(base)

        for candidate in candidates:
            for lang in self._installed_languages:
                if lang.code == candidate:
                    return lang
        return None

    def _get_translation(self, source_code, target_code):
        """Get a translation object for the given language pair with fallback."""
        src_lang = self._find_lang(source_code)
        tgt_lang = self._find_lang(target_code)

        if not src_lang:
            raise ValueError(f"Source language not found: {source_code}")
        if not tgt_lang:
            raise ValueError(f"Target language not found: {target_code}")

        translation = src_lang.get_translation(tgt_lang)
        if not translation:
            raise ValueError(f"No translation available: {source_code} → {target_code}")

        return translation

    def _log_failure(self, text, source, target, error):
        """Log translation failures to learn list for codebookLearner."""
        try:
            # In container: SPECMEM_PROJECT_PATH=/data, sockets are at /data/run/
            base = os.environ.get('SPECMEM_PROJECT_PATH', '/data')
            learn_file = os.path.join(base, 'run', 'translate-failures.jsonl')
            os.makedirs(os.path.dirname(learn_file), exist_ok=True)
            entry = json.dumps({
                'text': text[:200],  # Truncate long texts
                'source': source,
                'target': target,
                'error': str(error),
                'ts': time.time()
            })
            with open(learn_file, 'a') as f:
                f.write(entry + '\n')
        except Exception:
            pass  # Never let logging break translation

    def translate(self, text, source, target):
        """Translate text, handling newline-separated batches."""
        self._ensure_models()

        # If models failed to load (DNS/network error), return text as-is
        if not self._models_loaded:
            self._log_failure(text[:100], source, target, 'models_not_loaded')
            return text

        # Switch to active threads during translation
        try:
            import ctranslate2
            ctranslate2.set_num_threads(self.resource_config.get_active_threads())
        except Exception:
            pass

        lines = text.split('\n')
        results = []

        for line in lines:
            stripped = line.strip()
            if not stripped:
                results.append('')
                continue

            if not detect_translatable(stripped):
                results.append(stripped)
                continue

            # Use cached translation — on failure, let English through + log
            try:
                translated = self._cached_translate(stripped, source, target)
                results.append(translated)
            except Exception as e:
                # Let English slide through, add to learn list
                results.append(stripped)
                self._log_failure(stripped, source, target, e)

        # Return to crawl threads
        try:
            import ctranslate2
            ctranslate2.set_num_threads(self.resource_config.get_crawl_threads())
        except Exception:
            pass

        return '\n'.join(results)

    @lru_cache(maxsize=10000)
    def _cached_translate(self, text, source, target):
        """Cached translation — avoids re-translating the same words."""
        return self._translate_single(text, source, target)

    def _translate_single(self, text, source, target):
        """Translate a single string using Argos, with S→T Chinese conversion."""
        translation = self._get_translation(source, target)
        # Network stays blocked — all models baked in, no downloads needed.
        # Lifting guard causes stanza to attempt github downloads → timeout → fail.
        result = translation.translate(text)
        result = improve_translation_formatting(text, result)
        # Convert Simplified → Traditional if target was zt/zh-Hant
        if target in ('zt', 'zh-Hant'):
            result = self._to_traditional(result)
        return result

    def _to_traditional(self, text):
        """Convert Simplified Chinese to Traditional Chinese."""
        if not hasattr(self, '_s2t_converter'):
            try:
                import opencc
                self._s2t_converter = opencc.OpenCC('s2t')
            except ImportError:
                # opencc not available — return as-is (Simplified still saves tokens)
                self._s2t_converter = None
        if self._s2t_converter:
            return self._s2t_converter.convert(text)
        return text

    def _handle_connection(self, conn):
        """Handle a single client connection."""
        try:
            # Read request (newline-delimited JSON)
            data = b''
            while True:
                chunk = conn.recv(4096)
                if not chunk:
                    break
                data += chunk
                if b'\n' in chunk:
                    break

            if not data:
                return

            request = json.loads(data.decode('utf-8'))

            # Health check - don't update last_request_time (allows idle unload)
            if request.get('q') == '__health_check__':
                response = {'translatedText': 'ok', 'status': 'healthy', 'models_loaded': self._models_loaded}
                conn.sendall(json.dumps(response).encode('utf-8') + b'\n')
                return

            # Update keepalive only for real translation requests
            self.last_request_time = time.time()

            # Translate
            q = request.get('q', '')
            source = request.get('source', 'en')
            target = request.get('target', 'zh')

            translated = self.translate(q, source, target)
            response = {'translatedText': translated}
            conn.sendall(json.dumps(response, ensure_ascii=False).encode('utf-8') + b'\n')

        except BrokenPipeError:
            pass
        except ConnectionResetError:
            pass
        except socket.timeout:
            pass
        except Exception as e:
            import traceback
            print(f"❌ Connection error: {e}", file=sys.stderr)
            traceback.print_exc(file=sys.stderr)
            try:
                conn.sendall(json.dumps({'error': str(e)}).encode('utf-8') + b'\n')
            except:
                pass
        finally:
            try:
                conn.close()
            except:
                pass

    def start(self):
        """Start the Unix socket server."""
        # Resolve socket path (env override for container mode)
        if not self.socket_path:
            explicit = os.environ.get('SPECMEM_TRANSLATE_SOCKET')
            if explicit:
                self.socket_path = explicit
            else:
                project = os.environ.get('SPECMEM_PROJECT_PATH', os.getcwd())
                self.socket_path = os.path.join(project, 'specmem', 'sockets', 'translate.sock')

        # Remove old socket
        if os.path.exists(self.socket_path):
            os.remove(self.socket_path)

        # Create directory
        os.makedirs(os.path.dirname(self.socket_path), exist_ok=True)

        # Create Unix socket
        server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        old_umask = os.umask(0o077)
        try:
            server.bind(self.socket_path)
            sock_perms = 0o777 if os.environ.get('SPECMEM_CONTAINER_MODE') == 'true' else 0o660
            os.chmod(self.socket_path, sock_perms)
        finally:
            os.umask(old_umask)

        server.listen(32)
        server.settimeout(60)

        executor = ThreadPoolExecutor(max_workers=2, thread_name_prefix='translate-worker')

        print(f"", file=sys.stderr)
        print(f"HARDWICK TRANSLATE - Argos Translate Socket Server", file=sys.stderr)
        print(f"   Socket: {self.socket_path}", file=sys.stderr)
        print(f"   Models: {self.model_dir}", file=sys.stderr)
        print(f"   Workers: 2", file=sys.stderr)
        print(f"   Threads: {self.resource_config.crawl_threads} crawl / {self.resource_config.active_threads} active", file=sys.stderr)
        print(f"   Cache: LRU (10000 entries)", file=sys.stderr)
        print(f"   Compute: int8 quantization", file=sys.stderr)
        print(f"   Model loading: lazy (on first request)", file=sys.stderr)
        if self.idle_timeout > 0:
            print(f"   Idle unload: {self.idle_timeout}s", file=sys.stderr)
            self._start_idle_monitor()
        else:
            print(f"   Idle unload: disabled (always warm)", file=sys.stderr)
        print(f"", file=sys.stderr)

        try:
            while not self.shutdown_requested:
                # Handle warm restart
                if self.warm_restart:
                    self.warm_restart = False
                    print(f"♻️  Warm restart complete", file=sys.stderr)

                try:
                    conn, _ = server.accept()
                    conn.settimeout(120)
                    executor.submit(self._handle_connection, conn)
                except TimeoutError:
                    continue
                except Exception as e:
                    if self.shutdown_requested:
                        break
                    print(f"❌ Accept error: {e}", file=sys.stderr)
        finally:
            print(f"🛑 Hardwick Translate shutting down...", file=sys.stderr)
            executor.shutdown(wait=True, cancel_futures=True)
            server.close()
            if os.path.exists(self.socket_path):
                os.remove(self.socket_path)
            print(f"✅ Shutdown complete.", file=sys.stderr)


def main():
    parser = argparse.ArgumentParser(description='Hardwick Translate - Argos Translate Socket Server')
    parser.add_argument(
        '--socket',
        default=None,
        help='Unix socket path (default: {project}/specmem/sockets/translate.sock)'
    )
    parser.add_argument(
        '--model-dir',
        default=DEFAULT_MODEL_DIR,
        help='Argos model package directory (packed models source, NOT the argostranslate packages dir)'
    )
    parser.add_argument(
        '--service',
        action='store_true',
        help='Run in service mode (no idle shutdown)'
    )
    args = parser.parse_args()

    service = HardwickTranslate(
        socket_path=args.socket,
        model_dir=args.model_dir,
    )
    service.start()


if __name__ == '__main__':
    main()