business-assistant/scripts/deploy.py

"""Linux deployment controller. Only manages the API belonging to this checkout."""
import argparse
from contextlib import closing, contextmanager
import errno
import ipaddress
import json
import os
from pathlib import Path
import signal
import socket
import sqlite3
import subprocess
import sys
import time
from urllib.request import build_opener, ProxyHandler

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from scripts.memory_service import unit_name
MODULE = 'src.step4_api'


class Deployment:
    def __init__(self, root=ROOT):
        self.root = Path(root).resolve()
        self.output = self.root / 'output/deploy'
        self.output.mkdir(parents=True, exist_ok=True)
        self.environments = self.output / 'environments'
        self.runtime_file = self.output / 'python-path'
        self.python = self.root / '.venv/bin/python'
        if self.runtime_file.exists() and os.environ.get('DEPLOY_OFFLINE') != '1':
            selected = Path(self.runtime_file.read_text(encoding='utf-8').strip())
            if not self.owned_python(selected):
                raise RuntimeError('invalid_deployment_runtime_path')
            self.python = selected
        self.pidfile = self.output / 'api.json'
        self.log = self.output / 'api.log'
        self.prepared = self.root / 'knowledge/prepared'

    def run(self, *args, env=None):
        subprocess.run([str(a) for a in args], cwd=self.root, check=True, env=env)

    def owned_python(self, path):
        if not path.is_absolute():
            return False
        if path == self.root / '.venv/bin/python':
            return True
        return (path.name == 'python' and path.parent.name == 'bin'
                and path.parent.parent.name.startswith('env-')
                and path.parent.parent.resolve().parent == self.environments.resolve())

    def activate_runtime(self):
        temporary = self.runtime_file.with_suffix('.tmp')
        temporary.write_text(str(self.python) + '\n', encoding='utf-8')
        temporary.replace(self.runtime_file)

    def module(self, name, *args):
        self.run(self.python, '-m', name, *args)

    @contextmanager
    def lock(self):
        import fcntl
        with (self.output / 'deploy.lock').open('a') as handle:
            try:
                fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError:
                raise RuntimeError('another_deployment_is_running') from None
            yield

    def identity(self, pid):
        """PID + start ticks + boot ID guard against PID reuse and server reboots."""
        try:
            proc = Path('/proc') / str(int(pid))
            stat = (proc / 'stat').read_text().rsplit(')', 1)[1].split()
            if stat[0] == 'Z' or (proc / 'cwd').resolve() != self.root:
                return None
            cmd = (proc / 'cmdline').read_bytes().split(b'\0')
            if len(cmd) < 4 or cmd[1:3] != [b'-m', MODULE.encode()]:
                return None
            if not self.owned_python(Path(os.fsdecode(cmd[0]))):
                return None
            return {'pid': int(pid), 'start': stat[19],
                    'boot': Path('/proc/sys/kernel/random/boot_id').read_text().strip()}
        except (FileNotFoundError, ProcessLookupError):
            return None

    def processes(self):
        # Also find a manually launched API using this exact venv and working directory.
        found = []
        for proc in Path('/proc').iterdir():
            if proc.name.isdigit():
                try:
                    identity = self.identity(int(proc.name))
                except PermissionError:
                    continue
                if identity:
                    found.append(identity)
        return found

    def send(self, identity, sig):
        pid = identity['pid']
        if type(pid) is not int or pid <= 1:
            raise RuntimeError('invalid_api_process_identity')
        if self.identity(pid) != identity:
            return
        opened = getattr(os, 'pidfd_open', None)
        send_fd = getattr(signal, 'pidfd_send_signal', None)
        # Prefer a kernel-pinned identity. Some older Python/Linux builds expose
        # neither function, and some expose them on a kernel without support.
        if callable(opened) and callable(send_fd):
            try:
                fd = opened(pid)
            except ProcessLookupError:
                return
            except OSError as exc:
                if exc.errno not in (errno.ENOSYS, errno.ENOTSUP):
                    raise
            else:
                try:
                    if self.identity(pid) != identity:
                        return
                    try:
                        send_fd(fd, sig)
                        return
                    except ProcessLookupError:
                        return
                    except OSError as exc:
                        if exc.errno not in (errno.ENOSYS, errno.ENOTSUP):
                            raise
                finally:
                    os.close(fd)
        if not getattr(self, '_pid_signal_notice', False):
            print('API stop mode: checked PID signals (pidfd unavailable).', flush=True)
            self._pid_signal_notice = True
        # Recheck cwd, command, owned interpreter, start ticks and boot ID as
        # close to kill() as possible. Unlike pidfd this cannot atomically pin
        # the process; do not claim to eliminate the final PID-reuse race.
        if self.identity(pid) != identity:
            return
        try:
            os.kill(pid, sig)
        except ProcessLookupError:
            return

    def stop(self):
        if self.systemd_managed():
            self.run('systemctl', 'stop', unit_name(self.root))
        processes = self.processes()
        for proc in processes:
            self.send(proc, signal.SIGTERM)
        deadline = time.monotonic() + 35
        while any(self.identity(p['pid']) == p for p in processes) and time.monotonic() < deadline:
            time.sleep(.2)
        for proc in processes:
            self.send(proc, signal.SIGKILL)
        deadline = time.monotonic() + 5
        while any(self.identity(p['pid']) == p for p in processes) and time.monotonic() < deadline:
            time.sleep(.1)
        if any(self.identity(p['pid']) == p for p in processes):
            raise RuntimeError('api_did_not_stop')
        self.pidfile.unlink(missing_ok=True)
        print('API stopped (only this checkout).', flush=True)

    def settings(self):
        from src.step4_api.config import load_settings
        return load_settings(self.root / '.env')

    def healthy(self, host, port):
        host = '::1' if host == '::' else '127.0.0.1' if host == '0.0.0.0' else host
        authority = f'[{host}]' if ':' in host else host
        try:
            opener = build_opener(ProxyHandler({}))
            with opener.open(f'http://{authority}:{port}/health', timeout=2) as response:
                return response.status == 200 and json.load(response).get('status') == 'ready'
        except (OSError, ValueError):
            return False

    def start(self):
        settings = self.settings()
        if self.systemd_managed():
            self.run('systemctl', 'start', unit_name(self.root))
            deadline = time.monotonic() + int(os.environ.get('DEPLOY_START_TIMEOUT', '180'))
            while time.monotonic() < deadline:
                pid = self.systemd_pid()
                if (any(proc['pid'] == pid for proc in self.processes()) and
                        self.healthy(settings.host, settings.port)):
                    print('Memory-isolated API ready (systemd).', flush=True)
                    return
                time.sleep(1)
            raise RuntimeError('systemd_api_health_timeout')
        if self.processes():
            if self.healthy(settings.host, settings.port):
                print('API already running and healthy.', flush=True)
                return
            raise RuntimeError('existing_api_unhealthy_use_restart')
        if not self.manifest_ready():
            raise RuntimeError('knowledge_missing_run_update_knowledge')
        # Do not mistake another application's /health for successful startup.
        family = socket.AF_INET6 if ipaddress.ip_address(settings.host).version == 6 else socket.AF_INET
        with socket.socket(family) as probe:
            probe.bind((settings.host, settings.port))
        with self.log.open('ab') as log:
            child = subprocess.Popen([str(self.python), '-m', MODULE], cwd=self.root,
                                     stdin=subprocess.DEVNULL, stdout=log, stderr=log,
                                     start_new_session=True)
        identity = None
        try:
            for _ in range(50):
                if child.poll() is not None:
                    raise RuntimeError('api_exited_before_ready')
                identity = self.identity(child.pid)
                if identity:
                    break
                time.sleep(.1)
            if not identity:
                raise RuntimeError('api_identity_unavailable')
            self.pidfile.write_text(json.dumps(identity), encoding='utf-8')
            deadline = time.monotonic() + int(os.environ.get('DEPLOY_START_TIMEOUT', '180'))
            while time.monotonic() < deadline:
                if child.poll() is not None:
                    raise RuntimeError('api_exited_before_ready')
                if self.healthy(settings.host, settings.port) and self.identity(child.pid) == identity:
                    print(f'API ready. PID={child.pid}; log={self.log}', flush=True)
                    return
                time.sleep(1)
            raise RuntimeError('api_health_timeout')
        except BaseException:
            if identity:
                self.send(identity, signal.SIGTERM)
            elif child.poll() is None:
                child.terminate()
            try:
                child.wait(timeout=35)
            except subprocess.TimeoutExpired:
                child.kill()
                child.wait(timeout=5)
            self.pidfile.unlink(missing_ok=True)
            raise

    def status(self):
        processes = self.processes()
        if not processes:
            print('API stopped.', flush=True)
            return 1
        settings = self.settings()
        ready = self.healthy(settings.host, settings.port)
        print(json.dumps({'pids': [p['pid'] for p in processes], 'healthy': ready,
                          'log': str(self.log)}), flush=True)
        return 0 if ready else 1

    def systemd_managed(self):
        return (Path('/etc/systemd/system') / unit_name(self.root)).is_file()

    def systemd_pid(self):
        result = subprocess.run(['systemctl', 'show', '--property=MainPID', '--value', unit_name(self.root)],
                                check=True, capture_output=True, text=True, timeout=10)
        return int(result.stdout.strip())

    def serve(self):
        # systemd owns the foreground process and its children. Resolve the
        # current deployment runtime on every restart, including after sync.
        if os.environ.get('API_REQUIRE_MEMORY_CGROUP') != '1':
            raise RuntimeError('serve_requires_memory_service')
        if self.processes():
            raise RuntimeError('existing_api_stop_before_memory_service')
        os.execv(str(self.python), [str(self.python), '-m', MODULE])

    def sync(self):
        if not (self.root / 'uv.lock').is_file():
            raise RuntimeError('uv_lock_required')
        version = (self.root / '.python-version').read_text().strip()
        if os.environ.get('DEPLOY_OFFLINE') == '1':
            self.python = self.root / '.venv/bin/python'
            if not self.python.is_file():
                raise RuntimeError('offline_project_environment_missing')
            self.run(self.python, self.root / 'scripts/check-runtime.py', '--environment', self.root / '.venv')
            self.run(self.python, '-c',
                     'import torch, transformers, pymilvus, redis, langgraph, starlette, uvicorn')
            return
        # Never mutate packages used by the running API or relocate a venv.
        environment = self.environments / f'env-{time.time_ns()}'
        environment.mkdir(parents=True)
        env = dict(os.environ, UV_PROJECT_ENVIRONMENT=str(environment))
        command = ['uv', 'sync', '--locked', '--python', version]
        self.run(*command, env=env)
        self.python = environment / 'bin/python'
        self.run(self.python, self.root / 'scripts/check-runtime.py', '--environment', environment)

    def models(self, *, check_only=False):
        # Keep download, ingestion and API on exactly the same local BGE path.
        from src.step2_knowledge_ingest.ingest import config
        actual = config(self.root / '.env')['EMBEDDING_MODEL_PATH'].resolve()
        if actual != (self.root / 'models/bge-large-zh-v1.5').resolve():
            raise RuntimeError('set_EMBEDDING_MODEL_PATH_to_models_bge_large_zh_v1_5')
        local_only = check_only or os.environ.get('DEPLOY_OFFLINE') == '1'
        self.run(self.python, self.root / 'scripts/deploy_models.py', *(['--check-only'] if local_only else []))

    def manifest_ready(self):
        try:
            return json.loads((self.prepared / 'ingestion/manifest.json').read_text())['complete'] is True
        except (OSError, ValueError, KeyError):
            return False

    def update_knowledge(self):
        stage = self.output / 'knowledge-staging'
        stage.mkdir(exist_ok=True)
        # SQLite backup includes committed WAL entries; never copy an open DB blindly.
        for relative in ('.summary_cache.sqlite3', 'ingestion/embedding_cache.sqlite3'):
            source = self.prepared / relative
            if source.exists() and not (stage / relative).exists():
                destination = stage / relative
                destination.parent.mkdir(parents=True, exist_ok=True)
                with closing(sqlite3.connect(source)) as old, closing(sqlite3.connect(destination)) as new:
                    old.backup(new)
        print(f'Knowledge staging: {stage}. Remote LLM summaries and Milvus writes follow.', flush=True)
        self.module('src.step1_knowledge_prepare', '--output', stage)
        self.module('src.step1_knowledge_prepare', '--output', stage, '--verify')
        self.module('src.step1_knowledge_prepare.summarize', '--prepared', stage, '--llm-workbook')
        self.module('src.step1_knowledge_prepare.summarize', '--prepared', stage, '--llm-workbook', '--verify-only')
        self.module('src.step2_knowledge_ingest', '--prepared', stage)
        self.module('src.step2_knowledge_ingest', '--prepared', stage, '--verify-only')
        self.module('src.step1_knowledge_prepare', '--output', stage, '--verify')
        manifest = json.loads((stage / 'ingestion/manifest.json').read_text())
        if manifest.get('complete') is not True:
            raise RuntimeError('incomplete_ingestion_manifest')
        return stage

    def publish_knowledge(self, stage):
        backup = self.output / f'knowledge-{time.time_ns()}-previous'
        self.prepared.parent.mkdir(parents=True, exist_ok=True)
        if self.prepared.exists():
            self.prepared.rename(backup)
        try:
            stage.rename(self.prepared)
        except BaseException:
            if backup.exists():
                backup.rename(self.prepared)
            raise
        print(f'Knowledge published; previous local snapshot (if any): {backup}', flush=True)

    def execute(self, action, company_name=None, archive=None, database=None, *, agent=None, db_prefix=None):
        libraries = ('policy', 'finance', 'qa', 'article', 'service_guide', 'service_catalog')
        if database is not None and (database not in libraries or archive is None):
            raise RuntimeError('invalid_knowledge_import_arguments')
        if action != 'import-knowledge' and (archive is not None or database is not None):
            raise RuntimeError('unexpected_knowledge_import_arguments')
        if agent is not None or db_prefix is not None:
            if action != 'import-knowledge':
                raise RuntimeError('knowledge_target_only_for_import')
            from src.knowledge_profiles import AGENTS, prepared_path
            if agent is not None and agent not in AGENTS:
                raise RuntimeError('unknown_knowledge_agent')
            self.prepared = prepared_path(self.root, agent or 'business')
        if action == 'serve':
            return self.serve()  # No deployment lock held for the service lifetime.
        with self.lock():
            if action == 'import-knowledge':
                if not (self.root / '.env').is_file():
                    raise RuntimeError('configure_dotenv_before_deployment')
                snapshot = Path(archive) if archive is not None else self.root / 'output/knowledge-snapshot.zip'
                if not snapshot.is_absolute():
                    snapshot = self.root / snapshot
                if not snapshot.is_file():
                    raise RuntimeError('offline_knowledge_snapshot_missing')
                options = ('--database', database) if database else ()
                if agent is not None:
                    options += ('--agent', agent)
                if db_prefix is not None:
                    options += ('--db-prefix', db_prefix)
                if agent is not None or db_prefix is not None:
                    options += ('--prepared', self.prepared)
                self.run(sys.executable, '-m', 'src.step2_knowledge_ingest.portable', 'import', snapshot, *options)
                if not self.manifest_ready():
                    raise RuntimeError('offline_knowledge_import_incomplete')
                return 0
            if action in ('check', 'check-milvus', 'check-redis', 'check-dms', 'check-deepseek', 'check-qcc'):
                if not (self.root / '.env').is_file():
                    raise RuntimeError('configure_dotenv_before_check')
                from scripts.deploy_check import run
                return run(self.root, company_name, action)
            if action == 'status':
                return self.status()
            if action == 'stop':
                self.stop()
                return 0
            if action == 'start':
                self.start()
                return 0
            if not (self.root / '.env').is_file():
                raise RuntimeError('configure_dotenv_before_deployment')
            self.settings()  # Reject invalid listener configuration before preparing.
            live = bool(self.processes())
            self.sync()
            stage = None
            if action in ('deploy', 'update-knowledge'):
                self.models(check_only=live)
                if os.environ.get('DEPLOY_OFFLINE') == '1' and action == 'update-knowledge':
                    raise RuntimeError('offline_knowledge_update_requires_online_model')
                if os.environ.get('DEPLOY_OFFLINE') == '1' and not self.manifest_ready():
                    snapshot = self.root / 'output/knowledge-snapshot.zip'
                    if not snapshot.is_file():
                        raise RuntimeError('offline_knowledge_snapshot_missing')
                    self.module('src.step2_knowledge_ingest.portable', 'import', snapshot)
                    if not self.manifest_ready():
                        raise RuntimeError('offline_knowledge_import_incomplete')
                elif action == 'update-knowledge' or not self.manifest_ready():
                    stage = self.update_knowledge()
            # All expensive work and validations precede the interruption.
            self.stop()
            if stage is not None:
                self.publish_knowledge(stage)
            self.activate_runtime()
            self.start()
            return 0


def apply_api_env_policy():
    """Load external API and DMS connection settings only from .env when requested."""
    if os.environ.get('DEPLOY_API_ENV_FILE_ONLY') == '1':
        for name in ('QCC_KEY', 'QCC_SECRET_KEY', 'QCC_TIMEOUT_SECONDS', 'QCC_HONOR_MAX_PAGES',
                     'QCC_SEARCH_URL', 'QCC_BASIC_URL', 'QCC_HONOR_URL',
                     'DEEPSEEK_API_KEY', 'DEEPSEEK_BASE_URL', 'DEEPSEEK_MODEL',
                     'DMS_BASE_URL', 'DMS_LOGIN_URL', 'AUTH_PORT', 'DMS_PORT',
                     'DMS_USERNAME', 'DMS_PASSWORD', 'DMS_TOKEN'):
            os.environ.pop(name, None)


def main():
    apply_api_env_policy()
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('action', choices=('deploy', 'import-knowledge', 'update-knowledge', 'restart', 'start', 'stop',
                                           'status', 'serve', 'check', 'check-milvus', 'check-redis', 'check-dms',
                                           'check-deepseek', 'check-qcc'))
    parser.add_argument('argument', nargs='?', help='Company name for checks, or ZIP path for import-knowledge')
    parser.add_argument('database', nargs='?', choices=('policy', 'finance', 'qa', 'article', 'service_guide', 'service_catalog'))
    parser.add_argument('--agent', choices=('business', 'talent'), help='Knowledge import destination (default business)')
    parser.add_argument('--db-prefix', help='Explicit knowledge import database prefix')
    args = parser.parse_args()
    if args.action != 'import-knowledge' and (args.agent is not None or args.db_prefix is not None):
        parser.error('knowledge target options require import-knowledge')
    if args.action in ('check', 'check-qcc'):
        if not args.argument or args.database:
            parser.error('check and check-qcc require one exact company name')
    elif args.action != 'import-knowledge' and (args.argument is not None or args.database is not None):
        parser.error('only import-knowledge accepts a ZIP path and optional library name')
    if not sys.platform.startswith('linux'):
        parser.error('Linux with /proc is required')
    try:
        if args.action == 'import-knowledge':
            if args.agent is not None or args.db_prefix is not None:
                return Deployment().execute(args.action, archive=args.argument, database=args.database,
                                            agent=args.agent, db_prefix=args.db_prefix)
            return Deployment().execute(args.action, archive=args.argument, database=args.database)
        return Deployment().execute(args.action, args.argument)
    except Exception as exc:
        # Never echo subprocess exception command lines, dotenv or service log bodies.
        message = str(exc) if isinstance(exc, RuntimeError) else type(exc).__name__
        print('Deployment failed: ' + message + '. Check output/deploy/api.log; API may be stopped.', file=sys.stderr)
        return 1


if __name__ == '__main__':
    raise SystemExit(main())