"""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): 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 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 () 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')) args = parser.parse_args() 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': 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())