"""Korean maintenance entry point. Live mutations require explicit subcommands/sudo.""" import argparse import json import os from pathlib import Path import shlex import subprocess import sys from field_ui import print, input, text ROOT = Path(__file__).resolve().parent SERVICE = "contextra-runtime.service" def command(args, timeout=30, **kwargs): from maintenance_process import run return run(args, check=True, timeout=timeout, **kwargs) def admin(args): if os.geteuid() != 0: os.execvp("sudo", ["sudo", "--", "/usr/local/bin/ct", *args]) def runtime(action): admin([action]) if action == "start": import maintenance_guard maintenance_guard.check() processes = subprocess.run(["pgrep", "-x", "llama-server"], capture_output=True).returncode cli = subprocess.run(["pgrep", "-x", "llama-cli"], capture_output=True).returncode if processes == 0 or cli == 0: raise RuntimeError("LLM을 먼저 종료하세요. 자동 종료하지 않습니다.") command(["systemctl", "start", SERVICE], 330) import connection compose=connection.active_compose() command(["docker","compose","-p","contextra-icerti","-f",str(compose),"up","-d","--pull","never","--no-build"],240) elif action == "stop": forced=[] rows = command(["docker", "ps", "--filter", "label=com.docker.compose.project=contextra-icerti", "--format", "{{.Names}}"], capture_output=True, text=True).stdout.splitlines() gateways = [x for x in rows if "ai-gateway" in x] for item in gateways: command(["docker", "stop", "--timeout", "30", item], 45) state=json.loads(command(['docker','inspect','--format','{{json .State}}',item],capture_output=True,text=True).stdout) if state.get('OOMKilled') or state.get('ExitCode')==137: forced.append(item) analytics = [x for x in rows if "analytics" in x] for item in analytics: command(["docker", "exec", "-i", item, "python3", "-"], 240, input=(ROOT / "stop_channels.py").read_bytes()) command(["systemctl", "stop", SERVICE], 330) remaining = command(["docker", "ps", "--filter", "label=com.docker.compose.project=contextra-icerti", "--format", "{{.Names}}"], capture_output=True, text=True).stdout.strip() if remaining: import connection command(["docker","compose","-p","contextra-icerti","-f",str(connection.active_compose()),"stop","--timeout","60"],330) remaining=command(["docker","ps","--filter","label=com.docker.compose.project=contextra-icerti","--format","{{.ID}}"],capture_output=True,text=True).stdout.strip() if remaining:raise RuntimeError("일부 컨테이너가 실행 중입니다. 자동 강제 회수하지 않습니다.") if forced: import field_profiles field_profiles.atomic(field_profiles.STATE/'last-stop-warning.json', {'status':'STOPPED_WITH_FORCED_GATEWAY','containers':forced,'graceful_verified':False}) raise RuntimeError('Services stopped, but Gateway required force/OOM termination. Not graceful PASS; inspect last-stop-warning.json.') print("완료. 카메라는 자동 Start하지 않습니다.") def status(): admin(['status']) checks=[["docker", "ps", "-a", "--filter", "label=com.docker.compose.project=contextra-icerti", "--format", "table {{.Names}}\t{{.Status}}\t{{.Image}}"], ["df", "-h", "/", "/backup", "/training"], ["nvidia-smi", "--query-gpu=name,utilization.gpu,memory.used,memory.total", "--format=csv"]] for args in checks: try:command(args) except (OSError,subprocess.SubprocessError):print(args[0], '조회 실패. 다른 상태 확인은 계속합니다.') import field_cli field_cli.status() def user_command(args, timeout=3600): """Keep HOME/work records and LLM execution with the installed operator, not root.""" if os.geteuid() == 0: return command(['runuser','-u','si','--',*args],timeout) return command(args,timeout) def menu(): admin([]) dispatch={'1':['status'],'2':['profile','select'],'3':['profile','edit'],'4':['profile','credentials'], '5':['ca'],'6':['net','status'],'7':['net','apply'],'8':['config','prepare'], '9':['config','connect'],'10':['config','apply'],'11':['start'],'12':['stop'], '13':['emergency'],'14':['diagnose'],'15':['config','resume'],'16':['config','restore'], '17':['profile','paths'],'18':['profile','file'],'19':['config','export'],'20':['apply'], '23':['config','confirm'],'24':['config','note'],'25':['profile','web'], '26':['config','version'],'27':['resources'],'28':['lang'],'29':['net','arp'],'30':['vscode'], '31':['usb']} while True: print('\nContextra 현장 관리 / field maintenance (0 Exit)') print('1 상태/status 2 환경선택/profile 3 설정/edit 4 계정저장/credentials 5 CA 파일') print('6 NIC 상태 7 NIC 적용 8 앱준비/prepare 9 명시적 연결/connect 10 앱적용/apply') print('11 서비스 Start 12 정상 Stop 13 긴급/emergency 14 진단/diagnose') print('15 적용재개/resume 16 설정복구/restore 17 파일경로/paths 18 파일편집/nano') print('19 기록반출/export 20 Python APP 패치 21 LLM chat 22 LLM API') print('23 사용자 확인기록 24 수동메모 25 주 Web 선택') print('26 Connection version 27 CPU/RAM resources 28 Language 29 ARP status') print('30 VSCode safety status (settings only; no automatic restart)') print('31 USB status (ct usb mount / ct usb unmount)') choice=input('번호 [Enter=메뉴 유지]: ').strip() if choice=='0':return try: if choice in ('21','22'):user_command(['ct-llm','chat' if choice=='21' else 'serve'],43200) elif choice in dispatch: if choice in ('11','12') and input('실행하려면 YES 입력: ')!='YES':continue main(dispatch[choice]) except (ValueError,RuntimeError,OSError,subprocess.SubprocessError,KeyError) as error: print('해당 작업 중단:',type(error).__name__,str(error)[:250]) print('메뉴는 계속 사용할 수 있습니다. 상태1 / 경로17 / 복구16') except KeyboardInterrupt:print('\n현재 작업을 중단했습니다. 진행 중 적용은 상태/재개로 확인하세요.') def main(argv=None): parser = argparse.ArgumentParser(description=text("Contextra field maintenance. status/stop/start/profile/config/resources/usb/lang")) parser.add_argument("action", nargs="?", choices=("status","stop","start","observe","work","connection","apply","edit","logs","access","auth","profile","config","net","ca","emergency","diagnose","resources","lang","vscode","usb")) parser.add_argument("arguments", nargs=argparse.REMAINDER) args = parser.parse_args(argv) if args.action is None: menu() elif args.action=='resources': admin(['resources',*args.arguments]) import field_resources field_resources.main(args.arguments) elif args.action=='lang': admin(['lang',*args.arguments]) import field_ui field_ui.configure(args.arguments) elif args.action=='vscode': admin(['vscode',*args.arguments]) import vscode_safety vscode_safety.main(args.arguments) elif args.action=='usb': admin(['usb',*args.arguments]) import field_usb field_usb.main(args.arguments) elif args.action in ('profile','config','net','ca','emergency','diagnose'): admin([args.action,*args.arguments]) import field_cli field_cli.main(args.action,args.arguments) elif args.action == "status": status() elif args.action in {"stop", "start"}: runtime(args.action) elif args.action == "access": admin(["access"]) command(["nice", "-n", "10", "ionice", "-c", "2", "-n", "7", sys.executable, str(ROOT / "workspace_access.py"), "--apply"], 600) elif args.action == "auth": admin(["auth", *args.arguments]) print('ct auth는 현재 실행 설정의 인증용입니다. 선택 프로필 계정 저장은 ct profile credentials, 새 Web 연결은 ct config connect.') command([sys.executable, str(ROOT / "auth_control.py"), *(args.arguments or ["status"])], 240) elif args.action == "observe": admin(["observe", *args.arguments]) import observability observability.action(args.arguments[0] if args.arguments else "status") elif args.action == "connection": admin(["connection", *args.arguments]) import connection connection.main(args.arguments) elif args.action == "apply": admin(["apply", *args.arguments]) import apply_source active = Path("/workspace/contextra/runtime/state/active.json") source = args.arguments[0] if args.arguments else json.loads(active.read_text()).get("offline_source",{}).get("worktree","/workspace/contextra/source/crowd-0f172d94") import hashlib expected = hashlib.sha256(active.read_bytes()).hexdigest() print("APP source:", source, "\nCurrent active SHA256:", expected) if input("Python 변경을 오프라인 APP image로 적용하려면 APPLY 입력: ") != "APPLY": return apply_source.apply(source,expected,confirmed=True) elif args.action == "logs": service = args.arguments[0] if args.arguments else "analytics" if service not in {"analytics","ai-control-sidecar","ai-gateway","redis","postgresql"}: raise ValueError("Unknown service") print("화면에만 표시합니다. 외부 전달 전 비밀정보를 확인하세요.") command(["docker","logs","--tail","100","contextra-icerti-"+service+"-1"]) elif args.action == "edit": if os.geteuid() == 0: user_command(['/usr/local/bin/ct','edit',*args.arguments]);return if len(args.arguments) != 1: raise ValueError("si 계정: ct edit /workspace/contextra/source/.../app/file.py") path = Path(args.arguments[0]).resolve(strict=True) if not path.is_relative_to("/workspace/contextra/source") or not path.is_file(): raise ValueError("Source file only") busy = command(["docker","ps","--filter","label=com.docker.compose.project=contextra-icerti","--format","{{.ID}}"],capture_output=True,text=True).stdout.strip() if busy: raise RuntimeError("편집 전에 ct stop을 실행하세요.") command(["nano","--",str(path)],3600) elif args.action == "work": if os.geteuid() == 0: user_command(['/usr/local/bin/ct','work',*args.arguments]);return import work_record sub = args.arguments if not sub: print("ct work begin SOURCE | note TEXT | end | export RECORD --reviewed") elif sub[0] == "begin" and len(sub) == 2: work_record.begin(sub[1]) elif sub[0] == "note": work_record.note(" ".join(sub[1:])) elif sub[0] == "end": work_record.end() elif sub[0] == "export" and len(sub) == 3: work_record.export(sub[1], reviewed=sub[2] == "--reviewed") else: raise ValueError("Invalid work record command") if __name__ == "__main__": try: main() except (ValueError, RuntimeError, OSError, subprocess.SubprocessError, KeyError) as error: print("중단:", type(error).__name__, str(error)[:250], file=sys.stderr) raise SystemExit(2)