"""Linux release operations. No SSH dependency; invoked by production48.py. Only the ai-chatbot PM2 application is changed. The original checkout, public files, credentials, dependencies and historical builds remain in place. """ import argparse import datetime import hashlib import json import os from pathlib import Path, PurePosixPath import re import shutil import socket import subprocess import tarfile import time import urllib.request APP = Path("/var/www/ai-chatbot") RELEASES = Path("/var/www/ai-chatbot-releases") NAME = "ai-chatbot" RESERVE = 1024**3 def require(value, message): if not value: raise RuntimeError(message) def run(args, cwd=None, stdin=None, timeout=60): result = subprocess.run(args, cwd=cwd, input=stdin, text=True, capture_output=True, timeout=timeout) require(result.returncode == 0, f"{args[0]} failed (exit {result.returncode})") return result.stdout def sha(path): digest = hashlib.sha256() with open(path, "rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def json_write(path, value): temporary = path.with_name(path.name + ".tmp") with open(temporary, "x", encoding="utf8") as stream: os.chmod(temporary, 0o600) json.dump(value, stream, indent=2) stream.write("\n") os.replace(temporary, path) def validate_members(members, destination): for member in members: path = PurePosixPath(member.name) require(not path.is_absolute() and ".." not in path.parts, "Unsafe archive member") require(member.isdir() or member.isfile() or member.issym(), "Special files and hard links are not accepted") if member.issym(): link = PurePosixPath(member.linkname) require(not link.is_absolute(), "Absolute archive symlink") target = (destination / path.parent / member.linkname).resolve() require(target.is_relative_to(destination.resolve()), "Archive symlink escapes destination") def safe_extract(archive, destination): """Reject traversal, special files and links escaping a new directory.""" require(not destination.exists(), "Extraction destination already exists") with tarfile.open(archive, "r:gz") as bundle: members = bundle.getmembers() validate_members(members, destination) needed = sum(m.size for m in members if m.isfile()) require(shutil.disk_usage(destination.parent).free > needed + RESERVE, "Insufficient space: at least 1 GiB must remain after extraction") destination.mkdir(mode=0o700) # Extract normal entries first; never follow an archive-created link. for member in members: if member.issym(): continue path = destination / member.name path.parent.mkdir(parents=True, exist_ok=True) if member.isdir(): path.mkdir(exist_ok=True) else: with bundle.extractfile(member) as source, open(path, "xb") as target: shutil.copyfileobj(source, target) path.chmod(member.mode & 0o777) for member in members: if member.issym(): path = destination / member.name path.parent.mkdir(parents=True, exist_ok=True) path.symlink_to(member.linkname) def release_path(release_id): require(re.fullmatch(r"\d{8}T\d{6}Z-[0-9a-f]{12}", release_id), "Invalid release ID") return RELEASES / release_id def app_state(): require(socket.gethostname() == "SfSTE0ttmsycO", "Not the verified production host") apps = json.loads(run(["pm2", "jlist"])) matching = [app for app in apps if app["name"] == NAME] require(len(matching) == 1, "Expected exactly one ai-chatbot PM2 application") app = matching[0] env = app["pm2_env"] cwd = Path(env["pm_cwd"]) require(cwd == APP or cwd.parent == RELEASES, "Unexpected production cwd") require(env["pm_exec_path"] == str(cwd / "node_modules/next/dist/bin/next"), "Unexpected Next.js executable") require(env.get("watch") is False, "PM2 watch must be disabled") require(env.get("CANVAS_E2E_LOCAL") != "true", "Production E2E mode is enabled") require(env.get("args") == ["start", "-p", "3125", "-H", "127.0.0.1"], "Unexpected production listen arguments") dist = env.get("NEXT_DIST_DIR", ".next") require(re.fullmatch(r"\.next[\w.-]*", dist), "Unsafe active build directory") return app, cwd, dist def status(): app, cwd, dist = app_state() return {"hostname": socket.gethostname(), "pid": app["pid"], "status": app["pm2_env"]["status"], "cwd": str(cwd), "distDir": dist, "buildId": (cwd / dist / "BUILD_ID").read_text().strip(), "freeBytes": shutil.disk_usage(APP).free, "freeGiB": round(shutil.disk_usage(APP).free / 1024**3, 2), "node": run(["node", "-v"]).strip()} ENV_LOADER = """ const {createRequire}=require('node:module'); const nextRequire=createRequire(require.resolve('next/package.json')); nextRequire('@next/env').loadEnvConfig(process.cwd(),false,{info(){},error(){}}); """ def build_environment(): # Only public build-time configuration leaves production. source = ENV_LOADER + """ const extra=['AUTH_URL','NEXTAUTH_URL','PASCAL_EDITOR_URL','CANVAS_DOWNLOAD_CDN_ORIGIN']; console.log(JSON.stringify(Object.fromEntries(Object.entries(process.env) .filter(([key])=>key.startsWith('NEXT_PUBLIC_')||extra.includes(key))))); """ app_state() return json.loads(run(["node", "-"], cwd=APP, stdin=source)) def job_status(): source = ENV_LOADER + r""" const sql=require('postgres')(process.env.POSTGRES_URL,{max:1,connect_timeout:5,onnotice(){}}); (async()=>{try{ const tables=['CanvasImageTask','CanvasCapabilityTask','CanvasAgentRun', 'EnhanceProTask','AgentPsdRenderTask','VideoStoryboardTask'];const result={}; await sql.begin('read only',async tx=>{ await tx.unsafe("set local statement_timeout='5s'"); for(const table of tables){ const exists=await tx`select to_regclass(${JSON.stringify(table)}) as name`; if(!exists[0].name)throw new Error('Missing task table'); result[table]=await tx.unsafe('select status,count(*)::int as count,'+ 'count(*) filter (where "updatedAt">now()-interval \'1 hour\')::int as recent '+ 'from "'+table+'" where status not in '+ '(\'succeeded\',\'failed\',\'completed\',\'cancelled\',\'interrupted\',\'needs_review\') group by status'); } });console.log(JSON.stringify(result)); }catch(e){console.error('Task preflight failed');process.exitCode=1;} finally{await sql.end();}})(); """ return json.loads(run(["node", "-"], cwd=APP, stdin=source, timeout=50)) def build_manifest(workspace): source = workspace / "source" metadata = json.loads((workspace / "input.json").read_text()) runtime_files, runtime_links = {}, {} for name in ["public", "node_modules"]: for path in sorted((source / name).rglob("*")): relative = path.relative_to(source).as_posix() if path.is_symlink(): runtime_links[relative] = os.readlink(path) elif path.is_file(): runtime_files[relative] = {"sha256": sha(path), "bytes": path.stat().st_size} result = {"schema": 2, "sourceCommit": metadata["sourceCommit"], "sourceTree": metadata["sourceTree"], "releaseId": workspace.name, "nodeMajor": 20, "nodeAbi": "115", "pnpm": "9.12.3", "buildId": (source / ".next/BUILD_ID").read_text().strip(), "lockSha256": sha(source / "pnpm-lock.yaml"), "runtimeFiles": runtime_files, "runtimeLinks": runtime_links} require(runtime_files, "No runtime inventory") json_write(source / "release.json", result) # Preserve relative package links, but materialize hard links as ordinary # archive entries so the extractor never has to trust archive hard links. run(["nice", "-n", "10", "tar", "--hard-dereference", "--exclude=./.next/cache", "--exclude=./.env.production.local", "-czf", str(workspace / "artifact.tar.gz"), "-C", str(source), "."], timeout=600) return {"releaseId": workspace.name, "sourceCommit": result["sourceCommit"], "buildId": result["buildId"], "artifactSha256": sha(workspace / "artifact.tar.gz"), "artifactBytes": (workspace / "artifact.tar.gz").stat().st_size} def smoke(workspace): source = workspace / "source" expected = (source / ".next/BUILD_ID").read_text().strip() with socket.socket() as listener: listener.bind(("127.0.0.1", 0)) port = listener.getsockname()[1] environment = os.environ.copy() environment.update({"NODE_ENV": "production", "NEXT_DIST_DIR": ".next", "PORT": str(port), "POSTGRES_CONNECT_TIMEOUT_SECONDS": "1"}) with open(workspace / "smoke.log", "w") as log: process = subprocess.Popen(["node", "node_modules/next/dist/bin/next", "start", "-H", "127.0.0.1", "-p", str(port)], cwd=source, env=environment, stdout=log, stderr=log) try: ready = False for _ in range(45): require(process.poll() is None, "Isolated candidate exited; inspect smoke.log") try: with urllib.request.urlopen(f"http://127.0.0.1:{port}/", timeout=3) as response: ready = response.status == 200 and expected in response.read().decode("utf8") if ready: break except Exception: pass time.sleep(1) require(ready, "Isolated candidate readiness failed; inspect smoke.log") output = run(["env", f"SMOKE_BASE_URL=http://127.0.0.1:{port}", "node", "scripts/smoke-routes.mjs"], cwd=source, timeout=100) return {"candidateSmoke": "passed", "buildId": expected, "checks": output.strip().splitlines()} finally: process.terminate() try: process.wait(timeout=20) except subprocess.TimeoutExpired: process.kill() process.wait() def verify_runtime(manifest, root): require(manifest["schema"] == 2, "Unsupported artifact schema") require(sha(root / "pnpm-lock.yaml") == manifest["lockSha256"], "Release lock mismatch") for name, metadata in manifest["runtimeFiles"].items(): require(sha(root / name) == metadata["sha256"], f"Runtime file mismatch: {name}") for name, target in manifest["runtimeLinks"].items(): require((root / name).is_symlink() and os.readlink(root / name) == target, f"Runtime link mismatch: {name}") require(run(["node", "-p", "process.versions.modules"]).strip() == manifest["nodeAbi"], "Node ABI mismatch") def clone_file(source, destination): """Independent copy-on-write inode; never hard-link writable dependencies.""" import fcntl with open(source, "rb") as origin, open(destination, "xb") as target: # Linux FICLONE. Stop on unsupported filesystems, with no full-copy fallback. fcntl.ioctl(target.fileno(), 0x40049409, origin.fileno()) def extract_runtime(archive, destination, current): require(not destination.exists(), "Release destination already exists") with tarfile.open(archive, "r:gz") as bundle: members = bundle.getmembers() validate_members(members, destination) names = {m.name.removeprefix("./"): m for m in members} manifest = json.load(bundle.extractfile(names["release.json"])) require(manifest["schema"] == 2, "Unsupported artifact schema") reusable, needed = set(), 0 for member in members: if not member.isfile(): continue name = member.name.removeprefix("./") expected = manifest["runtimeFiles"].get(name) source = current / name if expected and source.is_file() and source.stat().st_size == member.size and sha(source) == expected["sha256"]: reusable.add(name) else: needed += member.size # Allow for inode/directory allocation as well as newly written file data. require(shutil.disk_usage(destination.parent).free > needed + RESERVE + 64 * 1024**2, "Insufficient space for release delta and 1 GiB reserve") destination.mkdir(mode=0o700) run(["cp", "-a", "--reflink=always", "--", str(current / "public"), str(destination / "public")], timeout=120) for member in members: if member.issym(): continue name = member.name.removeprefix("./") target = destination / name require(target.parent.resolve().is_relative_to(destination.resolve()) or target == destination, "Extraction parent escapes release") if member.isdir(): target.mkdir(parents=True, exist_ok=True) continue target.parent.mkdir(parents=True, exist_ok=True) if target.is_symlink(): target.unlink() if name in reusable: if not target.exists(): clone_file(current / name, target) else: with bundle.extractfile(member) as origin, open(target, "wb") as output: shutil.copyfileobj(origin, output) target.chmod(member.mode & 0o777) for member in members: if member.issym(): target = destination / member.name.removeprefix("./") require(target.parent.resolve().is_relative_to(destination.resolve()), "Link parent escapes release") target.parent.mkdir(parents=True, exist_ok=True) if target.is_symlink() or target.exists(): require(not target.is_dir(), "Unexpected directory at link path") target.unlink() target.symlink_to(member.linkname) require(shutil.disk_usage(destination).free > RESERVE, "Post-extraction disk reserve exhausted") return manifest def stage(release_id, digest): _, current, dist = app_state() archive = RELEASES / (release_id + ".tar.gz") require(sha(archive) == digest, "Artifact checksum mismatch") destination = release_path(release_id) manifest = extract_runtime(archive, destination, current) require(manifest["releaseId"] == release_id, "Artifact release ID mismatch") require(manifest["sourceCommit"][:12] == release_id[-12:], "Artifact source mismatch") verify_runtime(manifest, destination) run(["node", "-e", "const s=require('sharp');s({create:{width:1,height:1,channels:3,background:'white'}}).png().toBuffer().then(b=>{if(!b.length)process.exit(1)})"], cwd=destination) for name in [".env.local", "tmp", ".model-config-backups"]: require(not (destination / name).exists(), f"Unexpected bundled shared path: {name}") if (APP / name).exists(): (destination / name).symlink_to(APP / name) # Keep already open browser tabs' content-hashed assets available. for path in (current / dist / "static").rglob("*"): if not path.is_file(): continue target = destination / ".next/static" / path.relative_to(current / dist / "static") if target.exists(): require(sha(target) == sha(path), "Static asset name collision") else: target.parent.mkdir(parents=True, exist_ok=True) os.link(path, target) json_write(destination / "staged.json", {"artifactSha256": digest, "previousCwd": str(current), "previousDistDir": dist}) # The verified unpacked release is retained; the transport copy is redundant. archive.unlink() return {"staged": release_id, "buildId": manifest["buildId"], "freeGiB": round(shutil.disk_usage(APP).free / 1024**3, 2)} def ecosystem(app, cwd, dist): old = app["pm2_env"] env = {k: v for k, v in old.items() if k.isupper() and isinstance(v, (str, int, bool))} env.update({"NODE_ENV": "production", "PORT": "3125", "PWD": str(cwd), "NEXT_DIST_DIR": dist}) return {"apps": [{"name": NAME, "cwd": str(cwd), "script": str(cwd / "node_modules/next/dist/bin/next"), "args": old["args"], "interpreter": old.get("exec_interpreter", "/usr/bin/node"), "exec_mode": "fork", "instances": 1, "watch": False, "env": env, "autorestart": old.get("autorestart", True), "min_uptime": old.get("min_uptime", "30s"), "max_restarts": old.get("max_restarts", 10), "restart_delay": old.get("restart_delay", 3000), "kill_timeout": old.get("kill_timeout", 10000), "max_memory_restart": old.get("max_memory_restart", "3G"), "error_file": old["pm_err_log_path"], "out_file": old["pm_out_log_path"], "merge_logs": True, "log_date_format": "YYYY-MM-DD HH:mm:ss Z"}]} def healthy(cwd, build_id): for _ in range(30): try: app, active_cwd, _ = app_state() require(active_cwd == cwd and app["pm2_env"]["status"] == "online", "Wrong process") request = urllib.request.Request("http://127.0.0.1:3125/", headers={"Host": "ai.yydd.ai"}) with urllib.request.urlopen(request, timeout=5) as response: body = response.read().decode("utf8") if response.status == 200 and build_id in body: return True except Exception: pass time.sleep(2) return False def restart(config): run(["pm2", "startOrRestart", str(config), "--only", NAME, "--update-env"], timeout=150) def activate(release_id): destination = release_path(release_id) manifest = json.loads((destination / "release.json").read_text()) staged = json.loads((destination / "staged.json").read_text()) app, current, dist = app_state() require(str(current) == staged["previousCwd"] and dist == staged["previousDistDir"], "Production changed after staging") require(not (destination / "activation.json").exists(), "Already activated; inspect status") verify_runtime(manifest, destination) jobs = job_status() require(not any(row["recent"] for rows in jobs.values() for row in rows), "Recent active generation jobs: defer activation") before = ecosystem(app, current, dist) previous_build = (current / dist / "BUILD_ID").read_text().strip() json_write(destination / "previous.pm2.json", before) json_write(destination / "candidate.pm2.json", ecosystem(app, destination, ".next")) # Backup contains environment metadata and stays root-readable only. shutil.copy2(Path.home() / ".pm2/dump.pm2", destination / "pm2-dump.before.json") (destination / "pm2-dump.before.json").chmod(0o600) receipt = {"sourceCommit": manifest["sourceCommit"], "buildId": manifest["buildId"], "previousBuildId": previous_build, "previousCwd": str(current), "activatedAtUtc": datetime.datetime.now(datetime.timezone.utc).isoformat(), "historicalTaskStates": jobs, "artifactSha256": staged["artifactSha256"], "phase": "activating"} json_write(destination / "activation.json", receipt) try: restart(destination / "candidate.pm2.json") require(healthy(destination, manifest["buildId"]), "Candidate health check failed") run(["env", "SMOKE_BASE_URL=http://127.0.0.1:3125", "node", str(destination / "scripts/smoke-routes.mjs")], cwd=destination, timeout=100) run(["pm2", "save"]) except BaseException: restart(destination / "previous.pm2.json") require(healthy(current, previous_build), "Rollback health check failed; operator required") run(["pm2", "save"]) receipt["phase"] = "rolled-back" json_write(destination / "activation.json", receipt) raise receipt["phase"] = "active" json_write(destination / "activation.json", receipt) return receipt def rollback(release_id): destination = release_path(release_id) receipt = json.loads((destination / "activation.json").read_text()) _, current, _ = app_state() require(current == destination, "Rollback target is not currently active") jobs = job_status() require(not any(row["recent"] for rows in jobs.values() for row in rows), "Recent active generation jobs: defer rollback") restart(destination / "previous.pm2.json") require(healthy(Path(receipt["previousCwd"]), receipt["previousBuildId"]), "Rollback health check failed") run(["pm2", "save"]) receipt["phase"] = "rolled-back" json_write(destination / "activation.json", receipt) return status() def main(): parser = argparse.ArgumentParser() parser.add_argument("action", choices=["status", "build-env", "jobs", "smoke", "pack", "stage", "activate", "rollback"]) parser.add_argument("--release") parser.add_argument("--digest") parser.add_argument("--workspace", type=Path) args = parser.parse_args() os.umask(0o077) if args.action in {"pack", "smoke"}: require(socket.gethostname() == "dao02-Default-string", "Not the verified build host") require(args.workspace.parent == Path("/opt/ai-chatbot-builds"), "Invalid build workspace") result = build_manifest(args.workspace) if args.action == "pack" else smoke(args.workspace) elif args.action in {"status", "build-env", "jobs"}: result = {"status": status, "build-env": build_environment, "jobs": job_status}[args.action]() else: import fcntl require(RELEASES.is_dir() and not RELEASES.is_symlink(), "Invalid release root") with open(RELEASES / ".deploy.lock", "a") as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) if args.action == "stage": result = stage(args.release, args.digest) elif args.action == "activate": result = activate(args.release) else: result = rollback(args.release) print(json.dumps(result)) if __name__ == "__main__": main()