feat: add reusable jump-host node onboarding runtime

AGENTS.md

@@ -12,6 +12,8 @@
 
 ## 批量节点接入：默认启动一个子助手
 
+后续新批次默认使用 [正式适配器入口](docs/maintenance/NODE_BATCH_RUNTIME.md)，由 `configure-site.py` 固定计划，再由 `batch.py` 的 `ssh-site-v1` 运行；动态读取验收账号应得节点集合，不固定节点数量。没有有效验收账号时自动创建本任务受限测试用户并负责禁用收尾；SSH始终经已登记白色机。具体实现、证据续跑和原服务调查边界见该专题；旧冻结计划及其处理脚本不改摘要迁就新实现。仅维护脚本/文档不触发真实接入或线上创建测试用户。
+
 用户要求执行批量节点接入（明确说批量，或同批接入两个及以上节点）时，主协调者必须默认启动一个子助手，无需再次询问是否使用助手：核查验收助手负责候选服务器、端口、中转与已有资源冲突核查，并独立审查脚本、测试并发/查重/恢复及执行接入验收。主协调者负责统一参数、展示编号、端口、唯一账本及生产写入。完整分工、执行关卡与消耗控制见 [批量接入协作流程](docs/maintenance/NODE_BATCH.md#多智能体协作流程)。
 
 这项授权针对当前任务的子助手，不自动新建或分叉普通聊天，也不授权向已有聊天发消息。仅讨论方案、维护脚本/文档或单节点任务不自动触发。若子助手工具不可用、并发名额不足或助手失败，说明具体缺口并恢复/等待；未经用户明确允许，不静默省略角色或跳过验收。当前批次已有对应助手时复用，不重复启动。具体线上变更仍以用户本次接入授权为限；项目规则不启动无人值守接入。

docs/maintenance/NODE_BATCH.md

@@ -6,6 +6,8 @@
 
 ## 文件与运行方式
 
+2026-10-17新增[正式站点适配器](NODE_BATCH_RUNTIME.md)：后续新批次默认用`configure-site.py`生成`ssh-site-v1`计划，具备动态完整节点验收、受限测试账号、白色机SSH会话复用、证据续跑及阶段事件。已通过本地隔离测试，尚未部署或线上演练。本页下述外部命令适配器协议保留兼容，旧任务脚本和既有冻结摘要不修改；旧命令适配器不透传缓存证据，继续逐阶段现场probe。
+
 | 文件 | 用途 |
 | --- | --- |
 | `scripts/node-onboarding/batch.py` | 参数校验、资源预留、并发、进程间锁、逐阶段账本与恢复 |

docs/maintenance/NODE_BATCH_RUNTIME.md

@@ -0,0 +1,111 @@
+# 批量接入正式适配器
+
+本专题是后续批量接入的默认技术入口。适配器实现、隔离测试及文档于2026-10-17完成；**本轮没有新增线上节点、部署新运维版本或运行真实SSH/客户端验收**。上一轮三节点实际验收使用的是已冻结旧任务适配器，不能套用为新实现的生产验收。
+
+`batch.py`保留原命令适配器兼容，新计划默认用`configure-site.py`生成`runtime.kind=ssh-site-v1`。旧计划、账本、任务handler一律保留，不修改源码摘要后直接续跑旧任务。生产操作仍须具体接入授权、独立助手审核、Git运维部署及现场验收。
+
+## 文件与职责
+
+| 文件 | 职责 |
+| --- | --- |
+| `configure-site.py` | 本地合并明确站点参数，固定解释器、Paramiko版本、源码与非秘密配置摘要；不联网、不写账本、不覆盖旧计划 |
+| `site_adapter.py` | 复用正式stage/provision/promote流程；按国家、组、批次和路径参数执行，按需连接服务器，进行动态正式订阅验收 |
+| `ssh_pool.py` | 线程内复用白色机及下游SSH会话；下游强制通过白色机direct-tcpip，不提供直连回退 |
+| `panel-acceptance.php` | 开工前选择有效正式组账号；缺少时创建本批受限非管理员测试用户，记录归属并收尾禁用 |
+| `acceptance.py` | 动态集合、逐节点结果、原始证据摘要、QA身份/禁用/拒绝/下发排除和时效的完整自动审计 |
+
+## 准备与Git部署
+
+先按CoDocs确认VPS、白色机访问路线、正式组、展示编号及资源冲突。Python执行环境使用`requirements.txt`的Paramiko 5.0.0；配置与执行必须使用同一解释器。安装来源限官方包源/已审核环境，不临时导入任务目录中的未知transport模块。
+
+生产前先提交并推送源码，再生成、推送和部署独立运维Git版本：保留原安装脚本，加入`panel-acceptance.php`及其`operations.php`，还有stage/promote/rollback/readiness/verify-node入口；面板、网关和目标节点的对应运维目录必须为计划中完整`provision_commit`且无工作区差异。配置文件和凭据不入Git。独立快照由正式`bundle-ops.py`生成，保持基础运维树、工作目录和真实Git索引不变，仅复制10个明确的已提交运维文件，拒绝覆盖已有分支或bundle，不联网：
+
+```sh
+python3 scripts/node-onboarding/bundle-ops.py --source FULL_SOURCE_COMMIT --base FULL_OPS_BASE_COMMIT --branch codex/node-tools-BATCH --bundle /private/task/provision.bundle
+```
+
+随后按已授权任务推送返回的运维分支、审核commit/bundle摘要，再经白色机部署到明确版本。当前源码/运维快照的提交推送与远端部署分别报告；本轮不自动切换任何生产应用或旧任务的运维目录。
+
+适配器在stage前检查本地依赖、远端Git版本及工具、正式组和账号、节点既有配置、独立端口、原服务、证书归属与钩子。既有证书无本任务owner、全局证书钩子或目标证书的额外pre/post钩子均停止。缺依赖须在准备阶段解决，不先创建一批面板对象后再临时安装工具。
+
+## 本地参数文件
+
+全部放在权限700的任务目录，文件600。基础计划沿用[NODE_BATCH](NODE_BATCH.md#参数格式)，包括本批任意数量、任意已支持国家的候选节点；不固定五节点、日本、组1或任务日期。新增站点配置JSON如下，示例路径必须换为实际已核实值：
+
+```json
+{
+  "ops_root": "/opt/dpl-batch/provision",
+  "node_root": "/opt/dpl-node/provision",
+  "batch_backup_root": "/root/codex-backups/dpl/BATCH",
+  "node_backup_root": "/root/codex-backups/dpl/BATCH-nodes",
+  "qa_root": "/var/lib/dpl-sgw/BATCH",
+  "provider_snapshot": "/private/task/providers.json",
+  "exit_registry": "/private/task/exits.json",
+  "ssh_config": "/private/task/ssh.json",
+  "preserved_service": {
+    "unit": "XrayR", "binary": "/usr/local/XrayR/XrayR",
+    "config": "/etc/XrayR/config.yml", "port": 443
+  },
+  "formal_socks_port": 19203,
+  "client_sha256": "填已核验官方客户端文件的64位SHA256",
+  "gateway_containers": ["dpl-subscribe-gateway", "dpl-subscribe-admin", "subscribe-gateway", "subscribe-admin"]
+}
+```
+
+批次备份目录必须独立且固定，不跨实际批次复用旧资源基线。新节点端口、正式客户端端口均查重预留；共享网关客户端还使用同一账本内的进程锁。当前适配器用于独立AnyTLS节点及明确列出的既有服务；不自动把白色机、面板或网关当节点宿主。
+
+`ssh.json`含`known_hosts`绝对路径及`hosts`映射，key为准确CoDocs VPS ID。每台登记`host`、`port`、`username`、用户已批准的`host_key_sha256`（OpenSSH SHA256格式）、可选`credential_ref`；白色机额外登记`role=white_jump`，ID须与`context.jump_vps`一致。候选IP必须与冻结计划相同。`credential_ref`只引用既定安全存储中的600文件。默认JSON支持password/key_filename/passphrase；既有CoDocs Markdown可设`credential_format=codocs-password`，仅按原文件中唯一的“密码：”字段在内存读取，不能为了适配而复制或改写原凭据正文。内容不得进入计划、日志、argv或Git。配置和出口登记摘要固定，凭据只按需读取，不默认自动轮换。
+
+只从known-hosts文件选择与冻结配置中已批准指纹完全相符的公钥；找不到匹配时在读取/发送凭据前停止。全局known-hosts增加无关记录无需改变本批受信任指纹。不关闭主机密钥检查，不自动TOFU；首次信任先按访问规则核验/批准精确指纹，保存公开known-host条目。会话丢失只标记未知，不自动重发安装/开放命令；下次仍从现场probe恢复。
+
+`exits.json`按SNI登记独立已知预期，例如：
+
+```json
+{"node.example.invalid":{"expected_exit":"192.0.2.11","country":"JP"}}
+```
+
+覆盖该账号当前应获得的全部旧节点与新候选，不能从本次实测结果反填预期来让检查通过。增加或改变节点时，按CoDocs/授权现场证据更新新任务的登记；遇到未知预期或不支持协议停止，不能静默省略。当前站点适配器针对IPv4 A记录、固定端口AnyTLS；端口范围或其他协议停止而非静默省略，扩展需另补验证能力。
+
+`providers.json`是实际已登记后台UI或授权官方API的近期观察，最多900秒有效，不使用语言相关DOM字符串：
+
+```json
+{
+  "captured_at": 0,
+  "dns": [{"type":"A","name":"node.example.invalid","address":"192.0.2.11","proxy":false}],
+  "relays": [{"name":"dpl-new-anytls","host":"relay.example.invalid","port":47011,"target":"192.0.2.11","listen_port":24443,"proxy_protocol_in":false,"proxy_protocol_out":false}]
+}
+```
+
+`captured_at`填实际采集Unix时间。观察文件可以刷新，不能把静态勾选当现场证据。资源存在但参数/归属不符停止；尚未创建的DNS/中转仍按注册后台批量操作、更新观察后续跑。本实现没有新增Cloudflare/中转官方API写入权限，也不默认直连其源站。
+
+## 执行与等待
+
+```sh
+/path/to/reviewed/python scripts/node-onboarding/configure-site.py --plan /private/task/base.json --site /private/task/site.json --output /private/task/frozen.json
+/path/to/reviewed/python scripts/node-onboarding/batch.py plan --plan /private/task/frozen.json --state-root /private/task/ledger
+/path/to/reviewed/python scripts/node-onboarding/batch.py run --plan /private/task/frozen.json --state-root /private/task/ledger --jobs 3 --timeout 900 --execute --events
+```
+
+配置/plan只做本地工作，不创建线上测试用户。`run --execute`进入准备前检时才允许准备验收账号，随后全批前检屏障通过才stage。`--events`输出无秘密的阶段状态JSON行，最后一行为批次结果；不加此参数保持原单JSON输出协议。账本`events.jsonl`仅追加状态变化，等待时使用进程输出/事件，不每几秒重复查全部状态。只有有新依赖证据、外部DNS/中转变化或用户恢复指令时续跑，不能无限重试。
+
+SSH会话仅在本次进程、同一工作线程内复用，线程不共用可变SSH客户端；进程退出关闭连接，不建立后台守护进程。每个阶段通过lazy连接只访问必要主机。Paramiko channel有远端命令超时；进程内模式不承诺`--timeout`能杀死已发送远端命令，停止后仍须核实实际结果。
+
+## 动态验收和测试用户
+
+准备阶段先按正式组、banned、额度和有效期选既有账号，不把管理员身份自动排除；检查真实`ServerService::getAvailableServers`及`getAvailableUsers`，包括插件过滤。没有可用账号时自动创建本任务测试用户：非管理员、100MiB、1小时、1设备、无订单/套餐改写及通知消息。私有journal在DB提交前记录身份，丢失回包时恢复同一用户，不重复创建或接管同名用户。
+
+新节点仍沿用各自隔离QA→严格TLS/出口/上报→独立助手按精确报告审核→append开放并禁用QA。全部本组候选开放后，正式验收按账号**当前实际应获得的集合**生成完整快照，与客户端完整订阅逐条校验，并逐条实际联网。集合绑定节点ID、标签、端点、SNI、独立预期出口/国家；验收后再取一次集合，增加、删除或改变任一项均阻止旧报告通过，不固定节点数。旧节点同样必须覆盖。
+
+每组正式客户端串行运行，保留完整订阅DNS/路由，只给诊断目的地固定出站。新建测试用户可能等待节点拉取用户列表；只对通用客户端运行失败最多做3轮诊断读测试，间隔15秒且共用900秒预算。结构、TLS或出口断言错误不重试；不重放生产写入、不另建用户来绕过失败。证据绑定完整计划摘要、节点集合摘要、原始报告及QA拒绝文件SHA256、QA实际ID、403、banned、实际下发排除及900秒时效。任一缺项或摘要/集合变化失败。旧节点关键配置、全部套餐、原用户关联和订阅地址按批次原始基线核对，网关容器ID/启动时间/重启数再次核对。
+
+正式验收成功或失败都禁用本任务新建的正式测试用户；既有账号不被禁用或改密码/权限。进程正常退出在finally收尾，即使等待或部分前检失败也尝试清理；遇到断网/SIGKILL无法保证立即清理，记录的归属、1小时有效期与额度是兜底，恢复后先核对旧journal。已关闭用户绝不重新启用；下一次需要用户时创建新受限租约，保留历史并在最终审计核对其403/下发排除。完成要求由服务端实际集合/订阅拒绝证明，不承诺立即踢断已经建立的客户端会话。
+
+## 续跑与原服务调查
+
+preflight/publish/audit始终现场执行。只有先前状态为done、收到真实前检依赖摘要、请求/代码/config未变、证据未过期，才可复用stage/dns/provision/relay/verify。前驱依赖变化使后续缓存失效；running/unknown/conflict绝不跳过。旧适配器不提供这些证据时仍按原逐阶段probe逻辑执行。生产写入仍先probe、持久化意图、apply后probe，不为速度跳过归属检查。
+
+原服务保存最初unit/程序/配置SHA256、监听、active及PID/启动时间/重启数。配置/程序/监听改变或不可用停止；仅进程变化分类为“需调查”，不宣称是DPL导致，也不自动刷新基线。记录观察后由操作者及助手核查生命周期、OOM、操作时间线和稳定采样；认可时在本节点私有备份登记`preserved-service-review.json`，绑定最初/本次观察摘要、原因及证据SHA256。原基线保持；再次重启不匹配旧审核，须重新调查。
+
+## 验证范围
+
+新增Python测试覆盖动态1/2/7/23节点、集合变化/漏旧节点、出口/TLS/QA身份/摘要/时效、白色机跳转及严格密钥策略、线程会话复用/断线、依赖续跑及未知/过期状态、原服务精确调查回执、远端脚本与PHP构造。PHP隔离测试使用真实Eloquent/SQLite和ServerService，覆盖自动受限用户、既有管理员复用、真实下发、无效账号、journal恢复、关闭后新租约、归属漂移、插件排除及失败收尾；不加载.env/应用provider/Redis/HTTP。生产停机、失败回退、远端真实会话复用性能及十节点耗时尚未演练，不承诺固定提速倍数。

docs/maintenance/NODE_RUNBOOK.md

@@ -84,6 +84,8 @@ esac
 
 ## 脚本能力与边界
 
+后续新批次按[正式适配器](NODE_BATCH_RUNTIME.md)准备和执行：动态账号授权集合验收、缺账号时自动创建受限用户、白色机SSH复用及证据有效期续跑。旧命令适配器继续兼容。下文运维提交为上一轮三节点已验收版本，新适配器与PHP验收用户入口尚未部署，不能把旧任务验收视为新版本线上通过。
+
 2026-10-17：下列脚本已完成 PHP 语法检查、面板隔离测试（9 tests / 65 assertions）及批量进程测试（11 tests），源码提交并推送 `6af6764`；独立运维快照 `b8d10fac963d63825c97d36d41ff7f525116bbfd` 已部署到本次面板/网关的 `/opt/dpl-batch/provision` 与三台新节点 `/opt/dpl-node/provision`。生产应用代码未随之更新。各阶段现场结果以 [STATUS](STATUS.md) 和 [接入记录](NODE_ONBOARDING.md) 为准；隔离测试中的通知替身不证明生产通知成功。
 
 | 工具 | 新行为 | 边界 |

docs/maintenance/README.md

@@ -24,6 +24,7 @@
 | NODE_ONBOARDING.md | 五节点链路、命名规则、共享资源与验收 |
 | NODE_RUNBOOK.md | initialize/append参数、脚本入口、前检/证书/验收与面板回退；隔离测试/运维版本与现场边界，含验证清单 |
 | NODE_BATCH.md | 批量接入默认一个子助手的分工与关卡；append并发锁/查重/断点恢复、命令协议与测试边界 |
+| NODE_BATCH_RUNTIME.md | 新批次默认正式适配器、动态集合验收、自动受限测试用户、白色机SSH复用、证据续跑及部署准备 |
 | GATEWAY_ADMIN_PROXY.md | 两网关管理域名、白色机新旧区域反代、独立证书续期、恢复和验收边界 |
 
 这些说明用于辅助 AI 接手，不代表 AI 会在聊天结束后自动持续运行。应用定时任务在服务器执行，AI 维护任务需要会话或明确配置的自动化来启动。

docs/maintenance/STATUS.md

@@ -2,6 +2,14 @@
 
 文档核对日期：2026-10-11。以下区分本会话检查结果与用户验收确认，不承诺服务器一直保持此状态。
 
+## 2026-10-17 批量接入流程优化（本地验证）
+
+已实现正式`ssh-site-v1`适配器及`configure-site.py`入口，去除临时handler中的固定日期、日本、组1和节点数量。正式验收从真实账号授权集合生成快照，完整校验新旧节点、预期出口、严格TLS、原始证据摘要、QA身份/403/禁用/下发排除及时效；集合变化停止。没有有效正式组账号时自动创建100MiB/1小时/1设备非管理员测试用户，journal恢复同一身份，正常成功/失败/等待均清理本任务用户，关闭后不复活；异常中断依靠日志恢复和时限兜底。已批准指纹匹配后才读取凭据，SSH始终经白色机，按线程复用会话、按阶段延迟连接，不自动重放断线命令。
+
+续跑保留全批前检及publish/audit现场关口，以请求/代码/当前依赖摘要和有效期复用已完成步骤，前驱变化向下游失效；事件输出替代重复状态轮询。原服务以unit/程序/配置/监听和进程信息区分实际冲突与待调查重启，原基线不覆盖。独立运维打包工具从已提交源码生成可追踪快照，保留工作区/真实索引及旧运维基础树；旧命令适配器及旧冻结任务文件不修改。
+
+本轮Python 30项测试通过；PHP新增9项/35断言及既有9项/65断言隔离测试通过，均未加载真实.env或联网。CoDocs全库离线检查通过，前次.obsidian权限问题已由用户处理，本轮未读写其配置正文。新版本尚未部署至远端或进行真实SSH/客户端性能演练，不能把上一轮三节点验收复用为新实现线上通过，也不承诺十节点耗时。准备/部署/执行与边界见[正式适配器专题](NODE_BATCH_RUNTIME.md)；后续新批次默认按该入口执行。
+
 ## 2026-10-17 三节点批量接入完成
 
 新增日本JP - 02/03/04（node-jp03/04/05），均已append正式组1、QA禁用。五节点正式完整订阅逐条严格TLS联网、预期IPv4出口/国家及三个QA订阅403通过；旧节点、套餐/用户关联、订阅设置与四个网关容器基线保持。续期演练逐台通过。源码6af6764、独立运维b8d10fac已提交推送并部署运维目录，生产应用代码未随之更新；9项/65断言面板隔离测试及11项批量进程测试通过。

scripts/node-onboarding/acceptance.py

@@ -0,0 +1,93 @@
+"""Token-free dynamic evidence validation shared by the reusable site adapter."""
+import hashlib
+import ipaddress
+import re
+import json
+import time
+
+
+def digest(value):
+    return hashlib.sha256(json.dumps(value, sort_keys=True, ensure_ascii=False, separators=(',', ':')).encode()).hexdigest()
+
+
+def snapshot(nodes):
+    if not isinstance(nodes, list) or not nodes:
+        raise ValueError('Empty authorized node set')
+    fields = ('id', 'tag', 'relay', 'relay_port', 'sni', 'expected_exit', 'country')
+    result = []
+    for node in nodes:
+        if node.get('type') != 'anytls' or any(node.get(k) in (None, '') for k in fields):
+            raise ValueError('Unsupported protocol or incomplete independently registered expectations')
+        if type(node['id']) is not int or node['id']<1 or type(node['relay_port']) is not int or not 1<=node['relay_port']<=65535 or not isinstance(node['country'],str) or not re.fullmatch('[A-Z]{2}',node['country']):
+            raise ValueError('Invalid authorized node identity or expectation')
+        address=ipaddress.ip_address(node['expected_exit'])
+        if getattr(address,'scope_id',None):raise ValueError('Scoped exit expectation unsupported')
+        result.append({k: node[k] for k in fields})
+    for key in ('id', 'tag'):
+        if len({n[key] for n in result}) != len(result):
+            raise ValueError('Duplicate authorized node identity')
+    if len({(n['relay'], n['relay_port'], n['sni']) for n in result}) != len(result):
+        raise ValueError('Ambiguous authorized endpoint')
+    return sorted(result, key=lambda n: n['id'])
+
+
+def audit(before, after, report_bytes, qa_bytes, binding, plan_sha, expected_qas, now=None, ttl=900):
+    now = time.time() if now is None else now
+    before, after = snapshot(before), snapshot(after)
+    if before != after or binding.get('snapshot_sha256') != digest(before):
+        raise ValueError('Authorized node set changed; regenerate and verify the complete set')
+    if binding.get('plan_sha256') != plan_sha:
+        raise ValueError('Wrong plan binding')
+    for label, raw in [('report', report_bytes), ('qa_rejections', qa_bytes)]:
+        if binding.get(label + '_sha256') != hashlib.sha256(raw).hexdigest():
+            raise ValueError('Evidence bytes changed')
+    report, qas = json.loads(report_bytes), json.loads(qa_bytes)
+    for stamp in (binding.get('verified_at'), report.get('verified_at'), qas.get('captured_at')):
+        if type(stamp) not in (int, float) or not 0 <= now-stamp <= ttl:
+            raise ValueError('Evidence expired or future dated')
+    if report.get('state') != 'live_verified' or len(report.get('results', [])) != len(before):
+        raise ValueError('Incomplete live verification')
+    expected = {(n['tag'], n['relay'], n['relay_port'], n['sni'], n['expected_exit'], n['country']) for n in before}
+    actual = set()
+    for n in report['results']:
+        if n.get('live') is not True or n.get('strict_tls') is not True:
+            raise ValueError('Live strict TLS required for every node')
+        actual.add(tuple(n.get(k) for k in ('tag', 'relay', 'relay_port', 'sni', 'exit_ip', 'country')))
+    if actual != expected:
+        raise ValueError('Node/endpoint/exit/country mismatch')
+    if qas.get('plan_sha256') != plan_sha or set(qas.get('users', {})) != set(expected_qas):
+        raise ValueError('QA evidence identity mismatch')
+    for key, identity in expected_qas.items():
+        q = qas['users'][key]
+        if q.get('id') != identity or q.get('http_status') != 403 or q.get('banned') is not True or q.get('excluded') is not True:
+            raise ValueError('QA account not fully revoked')
+    return {'nodes': len(before), 'snapshot_sha256': digest(before), 'verified_at': now}
+
+
+def service_change(before, current):
+    """Process movement is investigatory; never silently accept/rebase it."""
+    required = ('unit_sha256', 'binary_sha256', 'config_sha256', 'listeners')
+    if any(not before.get(k) or not current.get(k) for k in required):
+        return 'unknown'
+    if any(before[k] != current[k] for k in required) or current.get('active') is not True:
+        return 'conflict'
+    if any(before.get(k) != current.get(k) for k in ('pid', 'started', 'restarts')):
+        return 'investigate_restart'
+    return 'ready'
+
+
+def provider_state(observed, node, now=None):
+    """Canonical observations from the registered provider UI/API, never HTML substrings."""
+    now=time.time() if now is None else now
+    stamp=observed.get('captured_at')
+    if type(stamp) not in (int,float) or not 0<=now-stamp<=900:
+        raise ValueError('Provider observation expired')
+    if not isinstance(observed.get('dns'),list) or not isinstance(observed.get('relays'),list):
+        raise ValueError('Canonical DNS and relay observations required')
+    dns=[d for d in observed['dns'] if d.get('name')==node['tls_domain']]
+    relays=[r for r in observed['relays'] if r.get('host')==node['relay_host'] and r.get('port')==node['relay_port']]
+    if dns and (len(dns)!=1 or dns[0].get('type')!='A' or dns[0].get('address')!=node['node_address'] or dns[0].get('proxy') is not False):
+        raise ValueError('DNS provider resource conflict')
+    if relays and (len(relays)!=1 or relays[0].get('target')!=node['node_address'] or relays[0].get('listen_port')!=node['listen_port'] or relays[0].get('name')!='dpl-'+node['key']+'-anytls' or relays[0].get('proxy_protocol_in') is not False or relays[0].get('proxy_protocol_out') is not False):
+        raise ValueError('Relay provider resource conflict')
+    return {'dns':dns,'relays':relays}

scripts/node-onboarding/batch.py

@@ -6,6 +6,7 @@ import contextlib
 import fcntl
 import hashlib
 import ipaddress
+import importlib.util
 import json
 import os
 from pathlib import Path
@@ -20,6 +21,7 @@ import time
 PHASES = ('stage', 'dns', 'provision', 'relay', 'verify', 'publish', 'audit')
 SHARED = {'stage': 'panel', 'dns': 'dns', 'relay': 'relay', 'publish': 'panel'}
 STOP = threading.Event()
+EVENTS = False
 
 
 def digest(value):
@@ -85,7 +87,8 @@ def validate(plan):
         if type(n['formal_group_id']) is not int or n['formal_group_id'] < 1:
             raise ValueError('An existing formal group is required')
         ipaddress.ip_address(n['node_address'])
-        ipaddress.ip_address(n['expected_exit'])
+        exit_address=ipaddress.ip_address(n['expected_exit'])
+        if getattr(exit_address,'scope_id',None):raise ValueError('Scoped/link-local exit addresses are not supported')
         for k in ('relay_host', 'tls_domain'):
             if not re.fullmatch(r'[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?\.[a-z]{2,}', n[k]):
                 raise ValueError('Invalid DNS name: ' + k)
@@ -143,16 +146,21 @@ class Ledger:
                 CREATE TABLE IF NOT EXISTS batches(id TEXT PRIMARY KEY, digest TEXT NOT NULL);
                 CREATE TABLE IF NOT EXISTS claims(namespace TEXT, kind TEXT, value TEXT,
                     owner TEXT NOT NULL, PRIMARY KEY(namespace,kind,value));
+                CREATE TABLE IF NOT EXISTS evidence(batch TEXT,node TEXT,phase TEXT,
+                    dependency TEXT,expires REAL,request TEXT,PRIMARY KEY(batch,node,phase));
                 CREATE TABLE IF NOT EXISTS phases(batch TEXT, node TEXT, phase TEXT,
                     status TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0,
                     updated REAL NOT NULL, PRIMARY KEY(batch,node,phase));
             ''')
         os.chmod(self.path, 0o600)
 
+    @contextlib.contextmanager
     def db(self):
         c = sqlite3.connect(self.path, timeout=30)
+        try:
             c.execute('PRAGMA synchronous=FULL')
-        return c
+            with c: yield c
+        finally: c.close()
 
     def reserve(self, plan):
         namespace = plan['context']['panel_vps'] + ':' + plan['context']['app_root']
@@ -185,6 +193,15 @@ class Ledger:
                 if old and old[0] != owner:
                     raise ValueError('Another batch reserved the gateway SOCKS port')
                 c.execute('INSERT OR IGNORE INTO claims VALUES(?,?,?,?)', (scope, 'socks_port', value, owner))
+            formal_port=plan['context'].get('formal_socks_port')
+            if formal_port is not None:
+                if type(formal_port) is not int or not 1024<=formal_port<=65535 or any(n['socks_port']==formal_port for n in plan['nodes']):
+                    raise ValueError('A distinct formal SOCKS port is required')
+                scope,value='gateway:'+plan['context']['gateway_vps'],str(formal_port)
+                owner=digest({'context':plan['context'],'formal_port':formal_port})
+                old=c.execute('SELECT owner FROM claims WHERE namespace=? AND kind=? AND value=?',(scope,'socks_port',value)).fetchone()
+                if old and old[0]!=owner:raise ValueError('Formal SOCKS port already reserved')
+                c.execute('INSERT OR IGNORE INTO claims VALUES(?,?,?,?)',(scope,'socks_port',value,owner))
             c.execute('INSERT OR IGNORE INTO batches VALUES(?,?)', (plan['batch_id'], digest(plan)))
 
     def state(self, batch, key, phase):
@@ -193,11 +210,31 @@ class Ledger:
                             (batch, key, phase)).fetchone()
         return row[0] if row else 'pending'
 
+    def reusable(self, batch, key, phase, dependency, request):
+        if not dependency or self.state(batch, key, phase) != 'done':
+            return False
+        with self.db() as c:
+            row = c.execute('SELECT dependency,expires,request FROM evidence WHERE batch=? AND node=? AND phase=?', (batch,key,phase)).fetchone()
+        return bool(row and row[0] == dependency and time.time() <= row[1] <= time.time()+900 and row[2] == request)
+
+    def proof(self, batch, key, phase, dependency, expires, request):
+        with self.db() as c:
+            c.execute('INSERT OR REPLACE INTO evidence VALUES(?,?,?,?,?,?)',(batch,key,phase,dependency,expires,request))
+
     def mark(self, batch, key, phase, status, attempt=False):
+        previous=self.state(batch,key,phase)
         with self.db() as c:
             c.execute('''INSERT INTO phases VALUES(?,?,?,?,?,?) ON CONFLICT(batch,node,phase)
                 DO UPDATE SET status=excluded.status,attempts=phases.attempts+?,updated=excluded.updated''',
                       (batch, key, phase, status, int(attempt), time.time(), int(attempt)))
+        if previous!=status or attempt:
+            fd=os.open(self.root/'events.jsonl',os.O_CREAT|os.O_WRONLY|os.O_APPEND,0o600)
+            try:
+                os.fchmod(fd,0o600)
+                event=json.dumps({'batch':batch,'node':key,'phase':phase,'status':status,'time':time.time()})+'\n'
+                os.write(fd,event.encode())
+                if EVENTS:sys.stdout.write(event);sys.stdout.flush()
+            finally:os.close(fd)
 
 
 class Coordinator:
@@ -206,11 +243,35 @@ class Coordinator:
         self.ledger = Ledger(root)
         self.jobs = jobs
         self.timeout = timeout
+        self.dependencies = {}
+        self.runtime = None
         self.ledger.reserve(plan)
         self.dir = self.ledger.root / plan['batch_id']
         self.dir.mkdir(mode=0o700, exist_ok=True)
         self.plan_file = self.dir / 'plan.json'
         atomic_json(self.plan_file, plan)
+        if plan.get('runtime'):
+            self.check_runtime()
+            directory = str(Path(__file__).resolve().parent)
+            if directory not in sys.path: sys.path.insert(0, directory)
+            from site_adapter import SiteAdapter
+            self.runtime = SiteAdapter(plan, self.dir, atomic_writer=atomic_json, resource_lock=lock)
+
+    def check_runtime(self):
+        runtime = self.plan['runtime']
+        if runtime.get('kind') != 'ssh-site-v1': raise ValueError('Unknown runtime')
+        if runtime.get('python') != str(Path(sys.executable).resolve()):raise ValueError('Use the frozen runtime interpreter')
+        import paramiko
+        if paramiko.__version__ != runtime.get('paramiko_version'):raise ValueError('Runtime dependency version changed')
+        base = Path(__file__).resolve().parent
+        required = [base/n for n in ('batch.py','site_adapter.py','ssh_pool.py','acceptance.py')]
+        required += [Path(self.plan['context'][k]) for k in ('ssh_config','exit_registry')]
+        files = runtime.get('files', {})
+        if any(str(f) not in files for f in required): raise ValueError('Runtime source pin missing')
+        for f, sha in files.items():
+            if not Path(f).is_absolute() or hashlib.sha256(Path(f).read_bytes()).hexdigest() != sha:
+                raise ValueError('Runtime code/config changed')
+
 
     def call(self, n, phase, action):
         fingerprint = digest({'context': self.plan['context'], 'node': n})
@@ -218,6 +279,14 @@ class Coordinator:
         atomic_json(path, dict(n, request_sha256=fingerprint))
         argv = self.plan['adapter'] + ['--plan', str(self.plan_file), '--node', str(path),
                                        '--phase', phase, '--action', action, '--state-dir', str(self.dir)]
+        if self.runtime:
+            self.check_runtime()
+            r = self.runtime.call(dict(n, request_sha256=fingerprint), phase, action)
+            if (r.get('key'),r.get('phase'),r.get('request_sha256')) != (n['key'],phase,fingerprint):
+                raise RuntimeError('Runtime identity mismatch')
+            if r.get('status') not in ('ready','absent','waiting','unknown','conflict'):
+                raise RuntimeError('Runtime status invalid')
+            return r
         with subprocess.Popen(argv, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
                               start_new_session=True) as p:
             deadline = time.monotonic() + self.timeout
@@ -258,8 +327,20 @@ class Coordinator:
         if resource in ('dns', 'relay'):
             namespace = 'dpl-external'
         with lock(self.ledger.root, namespace + ':' + resource):
+            dependency = self.dependencies.get(key, {}).get(phase)
+            request = digest({'plan': self.plan, 'node': n})
+            # Publish/audit/preflight are always live; unknown or running phases cannot use a cache.
+            if phase in ('stage','dns','provision','relay','verify') and self.ledger.reusable(batch,key,phase,dependency,request):
+                if self.runtime: self.check_runtime()
+                return True
             r = self.call(n, phase, 'probe')
+            if phase == 'preflight' and r['status'] == 'ready':
+                deps = r.get('resume_dependencies', {})
+                clean = {k:v for k,v in deps.items() if k in PHASES and isinstance(v,str) and re.fullmatch('[0-9a-f]{64}',v)}
+                self.dependencies[key] = {k:digest({p:clean.get(p) for p in PHASES[:PHASES.index(k)+1]}) for k in clean}
             if r['status'] == 'ready':
+                if dependency and type(r.get('valid_until')) in (int,float) and time.time() < r['valid_until'] <= time.time()+900:
+                    self.ledger.proof(batch,key,phase,dependency,r['valid_until'],request)
                 self.ledger.mark(batch, key, phase, 'done')
                 return True
             if r['status'] == 'conflict':
@@ -278,6 +359,8 @@ class Coordinator:
                 return False
             r = self.call(n, phase, 'probe')
             self.ledger.mark(batch, key, phase, 'done' if r['status'] == 'ready' else r['status'])
+            if r['status'] == 'ready' and dependency and type(r.get('valid_until')) in (int,float) and time.time()<r['valid_until']<=time.time()+900:
+                self.ledger.proof(batch,key,phase,dependency,r['valid_until'],request)
             if r['status'] == 'conflict':
                 STOP.set()
             return r['status'] == 'ready'
@@ -324,8 +407,11 @@ def main():
     p.add_argument('--state-root', required=True)
     p.add_argument('--jobs', type=int, default=4)
     p.add_argument('--timeout', type=int, default=900)
+    p.add_argument('--events', action='store_true', help='Emit token-free phase transition JSON lines; final result is last line')
     p.add_argument('--execute', action='store_true', help='Required for a run that may write remotely')
     a = p.parse_args()
+    global EVENTS
+    EVENTS = a.events
     os.umask(0o077)
     if not 1 <= a.jobs <= 16 or not 1 <= a.timeout <= 3600:
         p.error('jobs must be 1..16; timeout 1..3600')
@@ -354,7 +440,14 @@ def main():
         runner = Coordinator(plan, a.state_root, a.jobs, a.timeout)
         for sig in (signal.SIGINT, signal.SIGTERM):
             signal.signal(sig, lambda *_: STOP.set())
+        try:
             success = runner.run()
+        finally:
+            if runner.runtime:
+                try:runner.runtime.close()
+                except Exception:
+                    for node in plan['nodes']:runner.ledger.mark(plan['batch_id'],node['key'],'coordinator','unknown')
+                    raise
         print(json.dumps({'batch_id': plan['batch_id'], 'complete': success,
                           'next': None if success else 'Inspect status, resolve waiting/conflict, rerun the exact plan'}))
         return 0 if success else 2

scripts/node-onboarding/bundle-ops.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+"""Build a versioned flat operations bundle from committed files, without touching the worktree/index."""
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+import tempfile
+
+FILES=('stage.php','promote.php','operations.php','rollback.php','readiness.php',
+       'verify-node.py','panel-acceptance.php','install-node.sh','renew-node.sh','dpl-node.service')
+
+
+def build(repo,source,base,branch,bundle):
+    if not all(re.fullmatch('[0-9a-f]{40}',r) for r in (source,base)) or not re.fullmatch('codex/[A-Za-z0-9_/-]+',branch) or '..' in branch:
+        raise ValueError('Explicit full commits and a new codex branch required')
+    destination=Path(bundle)
+    if not destination.is_absolute() or destination.exists() or destination.is_symlink():raise ValueError('New absolute bundle path required')
+    def git(*args,input=None,env=None):
+        r=subprocess.run(['git','-C',str(repo),*args],input=input,stdout=subprocess.PIPE,stderr=subprocess.DEVNULL,env=env)
+        if r.returncode:raise ValueError('Git source/base/ref check failed')
+        return r.stdout
+    git('cat-file','-e',source+'^{commit}');git('cat-file','-e',base+'^{commit}');git('check-ref-format','refs/heads/'+branch)
+    exists=subprocess.run(['git','-C',str(repo),'show-ref','--verify','--quiet','refs/heads/'+branch]).returncode
+    if exists!=1:raise ValueError('Branch exists or outcome unknown; inspect it, do not overwrite')
+    with tempfile.TemporaryDirectory(prefix='dpl-ops-index-',dir=destination.parent) as temp:
+        env=dict(os.environ,GIT_INDEX_FILE=str(Path(temp)/'index'))
+        git('read-tree',base,env=env)
+        for name in FILES:
+            entry=git('ls-tree',source,'--','scripts/node-onboarding/'+name).decode().strip()
+            if not entry:raise ValueError('Required committed operations file missing')
+            mode,kind,rest=entry.split(' ',2);sha,_=rest.split('\t',1)
+            if kind!='blob' or mode not in ('100644','100755'):raise ValueError('Regular committed operations file required')
+            git('update-index','--add','--cacheinfo',mode+','+sha+','+name,env=env)
+        tree=git('write-tree',env=env).decode().strip()
+        commit=git('commit-tree',tree,'-p',base,input=('DPL operations snapshot\n\nSource commit: '+source+'\n').encode()).decode().strip()
+        git('update-ref','refs/heads/'+branch,commit,'0'*40)
+        git('bundle','create',str(destination),'refs/heads/'+branch)
+    destination.chmod(0o600)
+    return {'commit':commit,'source_commit':source,'branch':branch,'bundle_sha256':hashlib.sha256(destination.read_bytes()).hexdigest(),'files':len(FILES),'network':False}
+
+
+def main():
+    p=argparse.ArgumentParser(description=__doc__)
+    p.add_argument('--repo',default=str(Path(__file__).resolve().parents[2]))
+    for k in ('source','base','branch','bundle'):p.add_argument('--'+k,required=True)
+    a=p.parse_args()
+    try:
+        print(json.dumps(build(a.repo,a.source,a.base,a.branch,a.bundle)));return 0
+    except Exception:
+        print(json.dumps({'state':'blocked','reason':'Inspect source/base, branch and bundle outcome before retry; no network performed'}));return 2
+
+if __name__=='__main__':raise SystemExit(main())

scripts/node-onboarding/configure-site.py

@@ -0,0 +1,44 @@
+#!/usr/bin/env python3
+"""Freeze a reusable SSH-site plan locally. No network, credentials or ledger writes."""
+import argparse
+import hashlib
+import json
+import importlib
+import sys
+from pathlib import Path
+from batch import validate,atomic_json
+from site_adapter import SiteAdapter
+
+
+def configure(plan,site):
+    allowed={'ops_root','node_root','batch_backup_root','node_backup_root','qa_root','provider_snapshot','exit_registry','ssh_config','preserved_service','formal_socks_port','gateway_containers','client_sha256'}
+    if set(site)-allowed:raise ValueError('Unexpected site configuration fields')
+    library=importlib.import_module('paramiko')
+    if library.__version__!='5.0.0':raise ValueError('Use the reviewed Paramiko 5.0.0 environment')
+    plan=dict(plan,context={**plan['context'],**site})
+    validate(plan)
+    adapter=SiteAdapter(plan,Path('/private/tmp/dpl-validation-only')) # Creates no directory or connections.
+    adapter.pool.close()
+    base=Path(__file__).resolve().parent
+    plan.pop('command_profile',None)
+    plan['adapter']=[str(Path(sys.executable).resolve()),str(base/'command-adapter.py')]
+    paths=[base/x for x in ('batch.py','site_adapter.py','ssh_pool.py','acceptance.py')]+[Path(site[k]) for k in ('ssh_config','exit_registry')]
+    plan['runtime']={'kind':'ssh-site-v1','python':str(Path(sys.executable).resolve()),'paramiko_version':library.__version__,'files':{str(p):hashlib.sha256(p.read_bytes()).hexdigest() for p in paths}}
+    return plan
+
+
+def main():
+    p=argparse.ArgumentParser(description=__doc__)
+    for k in ('plan','site','output'):p.add_argument('--'+k,required=True)
+    a=p.parse_args()
+    try:
+        output=Path(a.output)
+        if not output.is_absolute() or output.exists() or output.is_symlink():raise ValueError('New absolute output file required; never mutate a frozen plan')
+        plan=configure(json.loads(Path(a.plan).read_text()),json.loads(Path(a.site).read_text()))
+        atomic_json(output,plan)
+        print(json.dumps({'state':'configured','nodes':len(plan['nodes']),'network':False,'ledger_writes':False}))
+        return 0
+    except Exception:
+        print(json.dumps({'state':'blocked','reason':'Inspect explicit site paths, registry and plan; no network performed'}));return 2
+
+if __name__=='__main__':raise SystemExit(main())

scripts/node-onboarding/panel-acceptance.php

@@ -0,0 +1,94 @@
+<?php
+/** On-demand batch acceptance lease. No bootstrap occurs when included by tests. */
+namespace DplOnboarding;
+require_once __DIR__.'/operations.php';
+use App\Models\{User,Server,ServerGroup};
+use App\Services\ServerService;
+use Illuminate\Support\Facades\DB;
+use Illuminate\Support\Str;
+use RuntimeException;
+
+final class Acceptance {
+    public static function eligible(User $u,int $group): bool {
+        return !$u->banned && (int)$u->group_id===$group && (int)$u->u+(int)$u->d<(int)$u->transfer_enable && ($u->expired_at===null || (int)$u->expired_at>=time());
+    }
+    public static function prepare(string $dir,int $group,bool $apply): array {
+        return locked($dir,function()use($dir,$group,$apply){
+            if(!ServerGroup::find($group))throw new RuntimeException('Formal group missing');
+            $path=$dir.'/acceptance-user.json';
+            if(is_file($path) && readJson($path)['state']==='closed' && $apply){
+                $old=readJson($path);$prior=isset($old['id'])?User::find($old['id']):null;
+                if($old['owned'] && $prior && (!$prior->banned || $prior->email!==$old['email'] || $prior->uuid!==$old['uuid'] || $prior->token!==$old['token'] || $prior->is_admin || (int)$prior->group_id!==$old['group']))throw new RuntimeException('Prior lease not revoked');
+                writeJson($dir.'/acceptance-closed-'.($old['id']??'absent').'.json',$old);unlink($path);
+            }
+            if(is_file($path)){
+                $j=readJson($path);
+                if($j['group']!==$group || $j['state']==='closed')throw new RuntimeException('Lease closed or parameters changed; never reactivate');
+                $u=User::where('email',$j['email'])->first();
+                if(!$u && $j['owned'] && $j['state']==='prepared' && $apply){
+                    $u=DB::transaction(fn()=>self::create($j),1);
+                }
+                if(!$u || $u->uuid!==$j['uuid'] || $u->token!==$j['token'] || !self::eligible($u,$group) || ($j['owned'] && $u->is_admin))throw new RuntimeException('Acceptance lease drifted');
+                if($j['owned'])self::limits($u,$j);
+                $j['id']=$u->id;$j['state']='active';if($apply)writeJson($path,$j);
+                return ['state'=>'ready','id'=>$u->id,'owned'=>$j['owned']];
+            }
+            $u=User::where('group_id',$group)->where('banned',0)->whereRaw('u+d<transfer_enable')->where(fn($q)=>$q->whereNull('expired_at')->orWhere('expired_at','>=',time()))->where('email','not like','node-qa-%')->where('email','not like','batch-acceptance-%')->orderBy('id')->first();
+            if(!$apply)return ['state'=>$u?'existing':'would_create'];
+            $j=['group'=>$group,'state'=>'prepared','owned'=>!$u,'email'=>$u?->email??'batch-acceptance-'.substr(hash('sha256',$dir),0,12).'-'.Str::lower(Str::random(8)).'@example.invalid','uuid'=>$u?->uuid??(string)Str::uuid(),'token'=>$u?->token??Str::random(32),'quota'=>104857600,'expires_at'=>time()+3600];
+            // Persist identity before commit. A lost response resumes the same row, never another user.
+            if(!$u && User::where('email',$j['email'])->exists())throw new RuntimeException('Unowned acceptance email conflict');
+            writeJson($path,$j);
+            if(!$u)$u=DB::transaction(fn()=>self::create($j),1);
+            $j['id']=$u->id;$j['state']='active';writeJson($path,$j);
+            return ['state'=>'ready','id'=>$u->id,'owned'=>$j['owned']];
+        });
+    }
+    private static function limits(User $u,array $j):void {
+        if((int)$u->transfer_enable!==$j['quota'] || (int)$u->expired_at!==$j['expires_at'] || (int)$u->device_limit!==1)throw new RuntimeException('Owned lease limits changed; inspect before mutation');
+    }
+    private static function create(array $j): User {
+        $u=new User(['email'=>$j['email'],'uuid'=>$j['uuid'],'token'=>$j['token'],'password'=>password_hash(Str::random(40),PASSWORD_BCRYPT),'group_id'=>$j['group'],'transfer_enable'=>$j['quota'],'expired_at'=>$j['expires_at'],'banned'=>0,'is_admin'=>0,'remind_expire'=>0,'remind_traffic'=>0,'device_limit'=>1]);$u->saveQuietly();return $u;
+    }
+    public static function source(string $dir,int $group,bool $inspect=false): array {
+        $j=readJson($dir.'/acceptance-user.json');$u=User::findOrFail($j['id']);
+        if($j['group']!==$group || (!$inspect && $j['state']!=='active') || $u->uuid!==$j['uuid'] || $u->token!==$j['token'] || (!$inspect && !self::eligible($u,$group)) || ($inspect && !$j['owned'] && !self::eligible($u,$group)) || (int)$u->group_id!==$group)throw new RuntimeException('Acceptance user unavailable');
+        if($j['owned'])self::limits($u,$j);
+        $nodes=[];
+        foreach(ServerService::getAvailableServers($u) as $s){
+            if(isset($s['ports']))throw new RuntimeException('Port-range nodes require an explicit verifier');
+            $model=Server::findOrFail($s['id']);
+            if(!$inspect && !ServerService::getAvailableUsers($model)->contains('id',$u->id))throw new RuntimeException('User excluded by effective node filter');
+            $nodes[]=['id'=>$s['id'],'type'=>$s['type'],'tag'=>$s['name'],'relay'=>$s['host'],'relay_port'=>(int)$s['port'],'sni'=>$s['protocol_settings']['tls']['server_name']??null,'uuid'=>$s['password']];
+        }
+        if($inspect){foreach($nodes as &$n)unset($n['uuid']);unset($n);return ['nodes'=>$nodes];}
+        return ['user_id'=>$u->id,'owned'=>$j['owned'],'token'=>$u->token,'nodes'=>$nodes];
+    }
+    public static function cleanup(string $dir,bool $apply): array {
+        return locked($dir,function()use($dir,$apply){
+            $p=$dir.'/acceptance-user.json';if(!is_file($p))return ['state'=>'absent'];$j=readJson($p);
+            $u=isset($j['id'])?User::find($j['id']):User::where('email',$j['email'])->first();
+            if($j['owned'] && $u){
+                if($u->email!==$j['email'] || $u->uuid!==$j['uuid'] || $u->token!==$j['token'] || $u->is_admin || (int)$u->group_id!==$j['group'])throw new RuntimeException('Cleanup ownership mismatch');
+                self::limits($u,$j);
+                if($apply){$u->banned=true;$u->saveQuietly();}
+                if($apply && !$u->fresh()->banned)throw new RuntimeException('Cleanup failed');
+            }
+            if($apply){$j['state']='closed';$j['id']=$u?->id;writeJson($p,$j);}
+            return ['state'=>$apply?'closed':'preview','owned'=>$j['owned'],'id'=>$u?->id];
+        });
+    }
+}
+if(realpath($_SERVER['SCRIPT_FILENAME']??'')===__FILE__){
+    $o=getopt('',['action:','dir:','group:','apply','help']);
+    if(isset($o['help'])){echo "--action=prepare|source|snapshot|cleanup --dir=PRIVATE_DIR --group=EXISTING_ID [--apply]\n";exit;}
+    try{
+        $dir=$o['dir']??'';$group=(int)($o['group']??0);
+        if(!preg_match('~^/[A-Za-z0-9_./-]+$~D',$dir)||str_contains($dir,'..')||$group<1)throw new RuntimeException('Explicit private directory and group required');
+        umask(0077);if(!is_dir($dir))mkdir($dir,0700,true);
+        if(is_link($dir)||(fileperms($dir)&0077))throw new RuntimeException('Private directory required');
+        boot();$apply=isset($o['apply']);
+        $r=match($o['action']??''){'prepare'=>Acceptance::prepare($dir,$group,$apply),'source'=>Acceptance::source($dir,$group),'snapshot'=>Acceptance::source($dir,$group,true),'cleanup'=>Acceptance::cleanup($dir,$apply),default=>throw new RuntimeException('Unknown action')};
+        echo json_encode($r,JSON_THROW_ON_ERROR).PHP_EOL;
+    }catch(\Throwable $e){fwrite(STDERR,json_encode(['state'=>'blocked','reason'=>'Inspect protected lease and effective authorization; no automatic overwrite']).PHP_EOL);exit(2);}
+}

scripts/node-onboarding/requirements.txt

@@ -0,0 +1 @@
+paramiko==5.0.0

scripts/node-onboarding/site_adapter.py

@@ -0,0 +1,431 @@
+"""Reusable AnyTLS site adapter; invoked in-process by batch.py, never a daemon."""
+import base64,hashlib,json,os,shlex,threading,time
+from pathlib import Path
+from types import SimpleNamespace
+from ssh_pool import Pool,private_json
+from acceptance import audit,digest,snapshot,service_change,provider_state
+
+class SiteAdapter:
+    def __init__(self,plan,directory,pool=None,atomic_writer=None,resource_lock=None):
+        if atomic_writer is None or resource_lock is None:
+            from batch import atomic_json,lock
+            atomic_writer,resource_lock=atomic_writer or atomic_json,resource_lock or lock
+        self.atomic_json,self.resource_lock=atomic_writer,resource_lock
+        self.plan,self.directory=plan,Path(directory)
+        self.ctx=plan['context'];self.lock=threading.RLock();self.leases={};self.formal_done={}
+        for field in ('ops_root','node_root','batch_backup_root','node_backup_root','qa_root','provider_snapshot','exit_registry','ssh_config','client_binary'):
+            value=self.ctx.get(field,'')
+            if not isinstance(value,str) or not value.startswith('/') or '..' in value.split('/') or any(c not in 'abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_./-' for c in value):
+                raise ValueError('Explicit safe absolute site path required: '+field)
+        config=private_json(self.ctx['ssh_config'])
+        for candidate in plan['nodes']:
+            if config['hosts'].get(candidate['vps_id'],{}).get('host')!=candidate['node_address']:raise ValueError('SSH target differs from frozen candidate address')
+        for role in ('jump_vps','panel_vps','gateway_vps'):
+            if self.ctx[role] not in config['hosts']:raise ValueError('Missing registered SSH role')
+        self.pool=pool or Pool(config,self.ctx['jump_vps'])
+        if not __import__('re').fullmatch('[0-9a-f]{64}',self.ctx.get('client_sha256','')):raise ValueError('Reviewed official client SHA256 required')
+        self.exit_registry=private_json(self.ctx['exit_registry'])
+        for candidate in plan['nodes']:
+            if self.exit_registry.get(candidate['tls_domain'])!={'expected_exit':candidate['expected_exit'],'country':candidate['country']}:raise ValueError('Candidate exit registry missing or changed')
+        if any(n['vps_id'] in [self.ctx['jump_vps'],self.ctx['panel_vps'],self.ctx['gateway_vps']] for n in plan['nodes']):raise ValueError('Dedicated node host required')
+        for key in ('app_root',):
+            if not __import__('re').fullmatch('/[A-Za-z0-9_./-]+',self.ctx[key]) or '..' in self.ctx[key].split('/'):raise ValueError('Unsafe application root')
+        if type(self.ctx.get('formal_socks_port')) is not int or not 1024<=self.ctx['formal_socks_port']<=65535 or self.ctx['formal_socks_port'] in [n['socks_port'] for n in plan['nodes']]:raise ValueError('Reserve a distinct formal SOCKS port')
+        if not self.ctx.get('gateway_containers') or any(not __import__('re').fullmatch('[A-Za-z0-9_.-]+',v) for v in self.ctx['gateway_containers']):raise ValueError('Explicit gateway container inventory required')
+
+    def close(self):
+        # On waiting, failure and success, close owned formal leases before closing SSH.
+        failed=False
+        for group in list(self.leases):
+            try:self.account_command(group,'cleanup',True)
+            except Exception:failed=True
+        self.pool.close()
+        if failed:
+            self.atomic_json(self.directory/'cleanup-error.json',{'groups':list(self.leases),'reason':'cleanup_outcome_unknown','needs_probe':True})
+            raise RuntimeError('Acceptance cleanup incomplete; inspect protected leases before retry')
+
+    def account_command(self,group,action,apply=False,client=None):
+        panel=client or self.pool.get(self.ctx['panel_vps'])
+        directory=self.ctx['batch_backup_root']+'/acceptance-group-'+str(group)
+        cmd='cd '+self.ctx['app_root']+' && php '+self.ctx['ops_root']+'/panel-acceptance.php --action='+action+' --dir='+directory+' --group='+str(group)+(' --apply' if apply else '')
+        _,out,err=panel.exec_command(cmd,timeout=90);data=out.read();err.read()
+        if out.channel.recv_exit_status():raise RuntimeError('acceptance_account_blocked')
+        try:return json.loads(data)
+        except ValueError:raise RuntimeError('acceptance_account_response_not_json') from None
+
+    def prepare_account(self,n,panel,execute):
+        with self.lock:
+            group=n['formal_group_id']
+            for spec in self.plan['nodes']:
+                registered=self.exit_registry.get(spec['tls_domain'],{})
+                if registered != {'expected_exit':spec['expected_exit'],'country':spec['country']}:
+                    raise ValueError('Exit registry must independently cover the candidate plan')
+            if group not in self.leases:
+                # Register cleanup obligation before any possibly successful remote create.
+                self.leases[group]=True
+                self.account_command(group,'prepare',True,panel)
+            source=self.account_command(group,'source',client=panel)
+            for spec in source['nodes']:
+                if spec['sni'] not in self.exit_registry:raise ValueError('Missing old-node exit expectation')
+
+    def certificate_gate(self,n,node,read,execute):
+        owner=read(node,'/etc/dpl-node/batch-owner.json')
+        owned=bool(owner and json.loads(owner).get('request_sha256')==n['request_sha256'])
+        script=r"""from pathlib import Path
+import re
+for name in ['/etc/letsencrypt/cli.ini','/root/.config/letsencrypt/cli.ini']:
+ p=Path(name)
+ if p.exists():
+  for line in p.read_text().splitlines():
+   if re.match(r'^\s*(pre[-_]hook|post[-_]hook|deploy[-_]hook)\s*=\s*\S',line):raise SystemExit(2)
+domain=DOMAIN
+paths=[Path('/etc/letsencrypt')/kind/domain for kind in ['live','archive']]+[Path('/etc/letsencrypt/renewal')/(domain+'.conf')]
+if not OWNED and any(p.exists() or p.is_symlink() for p in paths):raise SystemExit(3)
+p=paths[-1]
+if p.exists():
+ for line in p.read_text().splitlines():
+  if re.match(r'^\s*(pre_hook|post_hook)\s*=\s*\S',line):raise SystemExit(4)
+  if re.match(r'^\s*deploy_hook\s*=',line) and line.split('=',1)[1].strip()!=HOOK:raise SystemExit(5)
+"""
+        prefix='DOMAIN='+repr(n['tls_domain'])+';OWNED='+repr(owned)+';HOOK='+repr('/bin/sh '+self.ctx['node_root']+'/renew-node.sh')+'\n'
+        execute(node,"python3 - <<'DPLCERTGATE'\n"+prefix+script+"\nDPLCERTGATE")
+        execute(node,'command -v git >/dev/null && command -v curl >/dev/null && command -v certbot >/dev/null && command -v ss >/dev/null && command -v sha256sum >/dev/null && command -v openssl >/dev/null')
+
+    def guard_service(self,node,backup,read,write,execute):
+        spec=self.ctx['preserved_service']
+        for field in ('unit','binary','config','port'):
+            if field not in spec:raise ValueError('Explicit preserved service specification required')
+        if not all(isinstance(spec[k],str) and __import__('re').fullmatch('[A-Za-z0-9_./@-]+',spec[k]) for k in ('unit','binary','config')) or type(spec['port']) is not int:
+            raise ValueError('Unsafe preserved service specification')
+        script="""import hashlib,json,pathlib,subprocess
+s=SPEC
+v=dict(x.split('=',1) for x in subprocess.check_output(['systemctl','show',s['unit'],'-p','MainPID','-p','ActiveEnterTimestamp','-p','NRestarts','-p','ActiveState','-p','FragmentPath'],text=True).splitlines())
+def h(p):return hashlib.sha256(pathlib.Path(p).read_bytes()).hexdigest()
+listeners=subprocess.check_output(['ss','-lntH','( sport = :'+str(s['port'])+' )'],text=True)
+print(json.dumps({'unit_sha256':h(v['FragmentPath']),'binary_sha256':h(s['binary']),'config_sha256':h(s['config']),'listeners':sorted(x.split()[3] for x in listeners.splitlines()),'pid':v['MainPID'],'started':v['ActiveEnterTimestamp'],'restarts':v['NRestarts'],'active':v['ActiveState']=='active'}))
+""".replace('SPEC',repr(spec))
+        current=json.loads(execute(node,"python3 - <<'DPLSERVICE'\n"+script+"\nDPLSERVICE"))
+        if not current['active'] or not current['listeners']:raise RuntimeError('preserved_service_unavailable')
+        execute(node,'install -d -m 700 '+backup)
+        p=backup+'/preserved-service-original.json';raw=read(node,p)
+        if raw is None:write(node,p,json.dumps(current));return current
+        before=json.loads(raw);status=service_change(before,current)
+        if status=='investigate_restart':
+            write(node,backup+'/preserved-service-observed.json',json.dumps(current))
+            review=read(node,backup+'/preserved-service-review.json')
+            review=json.loads(review) if review else {}
+            if review.get('before_sha256')!=digest(before) or review.get('after_sha256')!=digest(current) or not review.get('reason') or not review.get('evidence_sha256'):
+                raise RuntimeError('investigate_preserved_service_restart')
+        elif status!='ready':raise RuntimeError('preserved_service_'+status)
+        return current
+
+    def dependencies(self,n,panel,gateway,node,stage_probe,read,execute,gateway_py):
+        """Current dependency reads; a missing observation cannot authorize a skip."""
+        state=stage_probe(panel);deps={'stage':digest(state)}
+        ui=provider_state(private_json(self.ctx['provider_snapshot']),n)
+        try:
+            dns=json.loads(gateway_py(gateway,'import socket,json\nprint(json.dumps(sorted(set(x[4][0] for x in socket.getaddrinfo('+repr(n['tls_domain'])+',443,socket.AF_INET)))))'))
+            if dns==[n['node_address']]:deps['dns']=digest(dns)
+        except Exception:pass
+        try:
+            live=execute(node,'systemctl is-active dpl-node; git -C '+self.ctx['node_root']+' rev-parse HEAD; sha256sum /etc/dpl-node/config.yml /etc/dpl-node/secret.env /etc/systemd/system/dpl-node.service /etc/letsencrypt/live/'+n['tls_domain']+'/fullchain.pem')
+            deps['provision']=digest(live)
+            # End-to-end strict TLS at preflight is necessary to reuse a relay observation.
+            tls=gateway_py(gateway,'import ssl,socket\nc=ssl.create_default_context()\nwith socket.create_connection(('+repr(n['relay_host'])+','+str(n['relay_port'])+'),10) as s:\n with c.wrap_socket(s,server_hostname='+repr(n['tls_domain'])+') as t:print(t.version())')
+            rows=ui['relays']
+            deps['relay']=digest([rows,tls])
+            report=read(panel,self.ctx['node_backup_root']+'/'+n['key']+'/qa-report.json')
+            if report:deps['verify']=digest([state,live,hashlib.sha256(report).hexdigest()])
+        except Exception:pass
+        return deps
+
+    def formal(self,n,panel,gateway,php,execute,read,write):
+        group=n['formal_group_id'];ctx=self.ctx
+        with self.lock, self.resource_lock(self.directory.parent, ctx['gateway_vps']+':formal:'+str(ctx['formal_socks_port'])):
+            source=self.account_command(group,'snapshot' if group in self.formal_done else 'source',client=panel)
+            def public(source):
+                values=[]
+                for item in source['nodes']:
+                    if item['sni'] not in self.exit_registry:raise ValueError('Unregistered current node exit')
+                    values.append(dict(item,**self.exit_registry[item['sni']]))
+                return snapshot(values)
+            before=public(source)
+            # Do not run final verification while any candidate in this group is still isolated.
+            cohort=[x for x in self.plan['nodes'] if x['formal_group_id']==group]
+            wanted={(x['relay_host'],x['relay_port'],x['tls_domain']) for x in cohort}
+            if not wanted.issubset({(x['relay'],x['relay_port'],x['sni']) for x in before}):raise RuntimeError('formal_group_not_ready')
+            if group in self.formal_done:
+                prior=self.formal_done[group]
+                return audit(prior['before'],before,prior['report'],prior['qas'],prior['binding'],digest(self.plan),prior['expected_qas'])
+            root=ctx['qa_root']+'/formal-group-'+str(group)
+            execute(gateway,'install -d -m 700 '+root)
+            expectations=[]
+            for node in before:
+                original=next(x for x in source['nodes'] if x['id']==node['id'])
+                expectations.append({'tag':node['tag'],'host':node['relay'],'port':node['relay_port'],'sni':node['sni'],'uuid':original['uuid'],'expected_exit':node['expected_exit'],'expected_country':node['country']})
+            # Tokens/UUIDs never leave memory except the gateway's private task files.
+            write(gateway,root+'/source-private.json',json.dumps({'token':source['token'],'expectations':expectations,'owned':source['owned']}))
+            script="""import gzip,json,urllib.request,subprocess,os,time
+from pathlib import Path
+os.umask(0o077);root=Path(ROOT);source=json.loads((root/'source-private.json').read_text())
+r=urllib.request.Request(SUB+'/api/v1/client/subscribe?token='+source['token'],headers={'User-Agent':'sing-box/1.14.3'})
+with urllib.request.urlopen(r,timeout=30) as response:
+ data=response.read();headers={k.lower():v for k,v in response.headers.items()}
+ if headers.get('content-encoding','').lower()=='gzip':data=gzip.decompress(data)
+assert headers.get('cf-cache-status','').upper() not in ['HIT','STALE','UPDATING'] and not headers.get('age') and not headers.get('cf-mitigated') and 'no-store' in headers.get('cache-control','').lower()
+(root/'subscription.json').write_bytes(data);(root/'expectations.json').write_text(json.dumps(source['expectations']))
+deadline=time.monotonic()+900
+for attempt in range(3 if source['owned'] else 1):
+ with (root/'verify.stdout').open('wb') as out,(root/'verify.stderr').open('wb') as err:
+  result=subprocess.run(['python3',OPS+'/verify-node.py','--subscription',str(root/'subscription.json'),'--expectations',str(root/'expectations.json'),'--binary',BINARY,'--port',str(PORT),'--report',str(root/'report.json'),'--run'],timeout=max(1,deadline-time.monotonic()),stdout=out,stderr=err)
+ if result.returncode==0:break
+ try:reason=json.loads((root/'verify.stderr').read_text()).get('reason')
+ except Exception:reason=None
+ if not source['owned'] or attempt==2 or reason!='Verification failed; inspect protected diagnostics' or deadline-time.monotonic()<15:raise RuntimeError('Formal verification failed')
+ # A just-created user's UUID may need one node pull interval to propagate.
+ # Retry only diagnostic reads, keep the same lease; never replay production writes.
+ time.sleep(15)
+"""
+            prefix='ROOT='+repr(root)+';SUB='+repr(ctx['subscription_url'])+';OPS='+repr(ctx['ops_root'])+';BINARY='+repr(ctx['client_binary'])+';PORT='+repr(ctx['formal_socks_port'])+'\n'
+            try:
+                execute(gateway,"python3 - <<'DPLFORMAL'\n"+prefix+script+"\nDPLFORMAL",1000)
+                after=public(self.account_command(group,'source',client=panel))
+                report=read(gateway,root+'/report.json')
+            finally:
+                # Never leave an owned formal test user active, including failed client runs.
+                self.account_command(group,'cleanup',True,panel)
+            identities={};tokens={}
+            for candidate in cohort:
+                b=ctx['node_backup_root']+'/'+candidate['key']
+                result=php(panel,'$m=json_decode(file_get_contents('+self.php_value(b+'/manifest.json')+'),true);$s=App\\Models\\Server::findOrFail($m["node_id"]);$u=App\\Models\\User::findOrFail($m["test_user_id"]);echo json_encode(["id"=>$u->id,"token"=>$u->token,"banned"=>(bool)$u->banned,"excluded"=>!App\\Services\\ServerService::getAvailableUsers($s)->contains("id",$u->id)]);')
+                identities[candidate['key']]=result['id'];tokens[candidate['key']]=result
+            # Include every owned formal lease from earlier resumptions, not just the current user.
+            lease_dir=ctx['batch_backup_root']+'/acceptance-group-'+str(group)
+            leases=php(panel,'$out=[];foreach(glob('+self.php_value(lease_dir+'/acceptance-*.json')+') as $f){$j=json_decode(file_get_contents($f),true);if(!($j["owned"]??false)||empty($j["id"]))continue;$u=App\\Models\\User::findOrFail($j["id"]);$excluded=true;foreach(App\\Models\\Server::all() as $s)if(App\\Services\\ServerService::getAvailableUsers($s)->contains("id",$u->id))$excluded=false;$out[(string)$u->id]=["id"=>$u->id,"token"=>$u->token,"banned"=>(bool)$u->banned,"excluded"=>$excluded];}echo json_encode($out);')
+            for ident,value in (leases.items() if isinstance(leases,dict) else []):
+                key='formal-'+str(ident);identities[key]=value['id'];tokens[key]=value
+            users={}
+            for key,value in tokens.items():
+                # Only the gateway accesses the subscription source; local HTTP never bypasses its registered route.
+                script='import urllib.request,urllib.error\nr=urllib.request.Request('+repr(ctx['subscription_url']+'/api/v1/client/subscribe?token='+value['token'])+',headers={"User-Agent":"sing-box/1.14.3"})\ntry:\n response=urllib.request.urlopen(r,timeout=30);print(response.status);response.close()\nexcept urllib.error.HTTPError as e:print(e.code)'
+                # Keep tokens out of argv: exec_command carries a private stdin script, not an OS argv.
+                si,o,e=gateway.exec_command('python3 -',timeout=60);si.write(script);si.channel.shutdown_write();out=o.read();e.read()
+                if o.channel.recv_exit_status():raise RuntimeError('qa_rejection_probe_failed')
+                users[key]={'id':value['id'],'banned':value['banned'],'excluded':value['excluded'],'http_status':int(out.strip())}
+            qas=json.dumps({'users':users,'captured_at':int(time.time()),'plan_sha256':digest(self.plan)},sort_keys=True).encode()
+            binding={'plan_sha256':digest(self.plan),'snapshot_sha256':digest(before),'report_sha256':hashlib.sha256(report).hexdigest(),'qa_rejections_sha256':hashlib.sha256(qas).hexdigest(),'verified_at':int(time.time())}
+            audit(before,after,report,qas,binding,digest(self.plan),identities)
+            write(gateway,root+'/qa-rejections.json',qas);write(gateway,root+'/binding.json',json.dumps(binding));write(gateway,root+'/snapshot.json',json.dumps(before))
+            self.formal_done[group]={'before':before,'report':report,'qas':qas,'binding':binding,'expected_qas':identities}
+            # Persist only nonsecret, byte-bound results for coordinator/operator review.
+            self.atomic_json(self.directory/('formal-group-'+str(group)+'.json'),{'snapshot':before,'report':json.loads(report),'qas':json.loads(qas),'binding':binding})
+
+    @staticmethod
+    def php_value(value):
+        return "json_decode(base64_decode('"+base64.b64encode(json.dumps(value).encode()).decode()+"'),true)"
+
+    def call(self,n,phase,action):
+        plan=self.plan;ctx=self.ctx;key=n['key'];sha=n['request_sha256']
+        a=SimpleNamespace(phase=phase,action=action,state_dir=str(self.directory))
+        B=ctx['node_backup_root']+'/'+key;Q=ctx['qa_root']+'/'+key
+        OPS=ctx['ops_root'];NP=ctx['node_root'];APP=ctx['app_root']
+        def execute(c,cmd,timeout=120):
+            si,o,e=c.exec_command(cmd,timeout=timeout);out=o.read();err=e.read();code=o.channel.recv_exit_status()
+            if code: raise RuntimeError('remote_exit_'+str(code))
+            return out.decode()
+        def php(c,body):
+            si,o,e=c.exec_command('cd '+shlex.quote(APP)+' && php',timeout=90)
+            si.write("<?php require 'vendor/autoload.php';$app=require 'bootstrap/app.php';$app->make(Illuminate\\Contracts\\Console\\Kernel::class)->bootstrap();"+body)
+            si.channel.shutdown_write();out=o.read();err=e.read()
+            if o.channel.recv_exit_status():raise RuntimeError('panel_php_failed')
+            try:return json.loads(out)
+            except ValueError:raise RuntimeError('panel_response_not_json') from None
+        def payload(value):return "json_decode(base64_decode('"+base64.b64encode(json.dumps(value,ensure_ascii=False).encode()).decode()+"'),true)"
+        def read(c,path):
+            with c.open_sftp() as s:
+                try:
+                    with s.open(path,'rb') as f:return f.read()
+                except FileNotFoundError:return None
+                except OSError as e:
+                    if e.errno==2:return None
+                    raise
+        def write(c,path,data):
+            if isinstance(data,str):data=data.encode()
+            with c.open_sftp() as s:
+                tmp=path+'.tmp-'+str(os.getpid())+'-'+str(threading.get_ident())
+                with s.open(tmp,'w') as f:s.chmod(tmp,0o600);f.write(data)
+                s.posix_rename(tmp,path)
+        def parameters():return dict(sorted({'key':key,'name':n['name'],'relay-host':n['relay_host'],'relay-port':n['relay_port'],'listen-port':n['listen_port'],'tls-domain':n['tls_domain'],'panel-url':ctx['panel_url']}.items()))
+        def canonical(value):return hashlib.sha256(json.dumps(value,sort_keys=True,ensure_ascii=False,separators=(',',':')).encode()).hexdigest()
+        def containers(gateway):
+            data=json.loads(execute(gateway,'docker inspect '+ ' '.join(ctx['gateway_containers'])))
+            return {v['Name']:{'id':v['Id'],'started':v['State']['StartedAt'],'restarts':v['RestartCount']} for v in data}
+        def receipt(status,retry=False):return {'key':key,'phase':a.phase,'request_sha256':sha,'status':status,'valid_until':time.time()+300,**({'retry_safe':True} if retry else {})}
+        def stage_probe(panel):
+            p=parameters()
+            code='$b='+payload(B)+';$p='+payload(p)+r''';$f=$b.'/manifest.json';if(!is_file($f)){$conf=App\Models\ServerMachine::where('name','dpl-'.$p['key'])->exists()||App\Models\Server::where('name',$p['name'])->exists()||App\Models\Server::where('host',$p['relay-host'])->where('port',(string)$p['relay-port'])->exists();echo json_encode(['status'=>$conf?'conflict':'absent','retry_safe'=>!$conf]);return;}$m=json_decode(file_get_contents($f),true);$s=App\Models\Server::find($m['node_id']);$mch=App\Models\ServerMachine::find($m['machine_id']);$ok=($m['parameters']??null)===$p && $s && $mch && $s->machine_id===$mch->id && $mch->name==='dpl-'.$p['key'] && $s->type==='anytls' && ($s->protocol_settings['tls']['allow_insecure']??null)===false && $s->name===$p['name'] && $s->host===$p['relay-host'] && (int)$s->port===$p['relay-port'] && (int)$s->server_port===$p['listen-port'] && ($s->protocol_settings['tls']['server_name']??null)===$p['tls-domain'];$u=App\Models\User::find($m['test_user_id']);$g=App\Models\ServerGroup::find($m['group_id']);$ok=$ok && $u && $g && !$u->is_admin && (int)$u->group_id===$m['group_id'] && str_starts_with($u->email,'node-qa-'.$p['key'].'-') && $g->name==='dpl 接入验收 '.$p['key'];if(!is_file($b.'/promotion-journal.json'))$ok=$ok && $s->group_ids===[(string)$m['group_id']] && !$u->banned;echo json_encode(['status'=>$ok?'ready':'conflict','manifest'=>$m]);'''
+            return php(panel,code)
+        def state(panel):
+            return php(panel,'$b='+payload(B)+r''';$m=json_decode(file_get_contents($b.'/manifest.json'),true);$s=App\Models\Server::findOrFail($m['node_id']);$u=App\Models\User::findOrFail($m['test_user_id']);echo json_encode(['manifest'=>$m,'node_online'=>(bool)$s->is_online,'groups'=>$s->group_ids,'qa_banned'=>(bool)$u->banned,'u'=>(int)$u->u,'d'=>(int)$u->d]);''')
+        def baseline(panel,gateway):
+            directory=ctx['batch_backup_root'];f=directory+'/baseline.json'
+            version=json.loads(execute(panel,"python3 - <<'DPLVERSION'\nimport subprocess,hashlib,json\np="+repr(APP)+"\nh=subprocess.check_output(['git','-C',p,'rev-parse','HEAD']).decode().strip()\nd=subprocess.check_output(['git','-C',p,'diff','HEAD'])+subprocess.check_output(['git','-C',p,'status','--porcelain'])\nprint(json.dumps({'head':h,'worktree_sha256':hashlib.sha256(d).hexdigest()}))\nDPLVERSION"))
+            previous=read(panel,directory+'/application-version.json')
+            if previous is None:write(panel,directory+'/application-version.json',json.dumps(version))
+            elif json.loads(previous)!=version:raise RuntimeError('application_version_changed')
+            # Read/write a protected snapshot only once, before any new stage.
+            return php(panel,'$p='+payload(f)+r''';if(is_file($p)){echo json_encode(['exists'=>true]);return;} $v=['nodes'=>App\Models\Server::orderBy('id')->get()->toArray(),'plans'=>App\Models\Plan::orderBy('id')->get()->toArray(),'users'=>App\Models\User::orderBy('id')->get(['id','group_id','plan_id'])->toArray(),'subscribe_url'=>admin_setting('subscribe_url')];file_put_contents($p,json_encode($v,JSON_THROW_ON_ERROR));chmod($p,0600);echo json_encode(['exists'=>true]);''')
+        def gateway_py(gateway,script,timeout=120):
+            return execute(gateway,"python3 - <<'DPLPY'\n"+script+"\nDPLPY",timeout)
+        def assert_old(node):
+            self.guard_service(node,B,read,write,execute)
+        def transfer(panel,gateway,node):
+            execute(node,'install -d -m 700 /etc/dpl-node')
+            execute(gateway,'install -d -m 700 '+Q)
+            for source,dest,target in [('node-config.yml',node,'/etc/dpl-node/config.yml'),('node-secret.env',node,'/etc/dpl-node/secret.env'),('qa-private.json',gateway,Q+'/qa.json')]:
+                data=read(panel,B+'/'+source);prior=read(dest,target)
+                if prior is not None and prior!=data:raise RuntimeError('private_config_conflict')
+                if prior is None:write(dest,target,data)
+        def fetch_qa(gateway):
+            script='''import json,urllib.request
+        +from pathlib import Path
+        +q=json.loads(Path(Q+'/qa.json').read_text())
+        +req=urllib.request.Request(SUB+'/api/v1/client/subscribe?token='+q['subscribe_token'],headers={'User-Agent':'sing-box/1.14.3'})
+        +with urllib.request.urlopen(req,timeout=30) as r:
+        + data=r.read(); code=r.status; headers={k.lower():v for k,v in r.headers.items()}
+        +import gzip
+        +if headers.get('content-encoding','').lower()=='gzip':data=gzip.decompress(data)
+        +v=json.loads(data);Path(Q+'/subscription.json').write_bytes(data);Path(Q+'/subscription.json').chmod(0o600)
+        +print(json.dumps({'status':code,'cache':headers.get('cf-cache-status'),'control':headers.get('cache-control'),'age':headers.get('age')}))
+        +'''.replace('\n+','\n')
+            return gateway_py(gateway,'Q='+repr(Q)+';SUB='+repr(ctx['subscription_url'])+'\n'+script)
+        def phase_impl(panel,gateway,node):
+            if a.phase=='preflight':
+                with self.lock, self.resource_lock(self.directory.parent,ctx['panel_vps']+':panel'):
+                    execute(panel,'install -d -m 700 '+ctx['batch_backup_root'])
+                    execute(panel,'test "$(git -C '+OPS+' rev-parse HEAD)" = '+ctx['provision_commit']+' && test -z "$(git -C '+OPS+' status --porcelain)"')
+                    baseline(panel,gateway)
+                    base_path=ctx['batch_backup_root']+'/gateway-baseline.json'
+                    current_containers=containers(gateway)
+                    prior_gateway=read(panel,base_path)
+                    if prior_gateway is None:write(panel,base_path,json.dumps(current_containers,sort_keys=True))
+                    elif json.loads(prior_gateway)!=current_containers:return receipt('conflict')
+                ui=Path(ctx['provider_snapshot'])
+                if not ui.exists():return receipt('waiting')
+                snap=private_json(ui)
+                if type(snap.get('captured_at')) not in (int,float) or not 0<=time.time()-snap['captured_at']<=900:return receipt('waiting')
+                try:provider_state(snap,n)
+                except ValueError:return receipt('conflict')
+                gateway_py(gateway,'import socket\ns=socket.socket();s.bind(("127.0.0.1",'+str(n['socks_port'])+'));s.close()')
+                s=stage_probe(panel)
+                if s['status']=='conflict':return receipt('conflict')
+                assert_old(node)
+                self.certificate_gate(n,node,read,execute)
+                execute(node,'test "$(git -C '+NP+' rev-parse HEAD)" = '+ctx['provision_commit']+' && test -z "$(git -C '+NP+' status --porcelain)" && command -v git >/dev/null')
+                execute(gateway,"echo '"+ctx['client_sha256']+"  "+ctx['client_binary']+"' | sha256sum -c --status && test -x "+shlex.quote(ctx['client_binary'])+' && test "$(git -C '+OPS+' rev-parse HEAD)" = '+ctx['provision_commit']+' && test -z "$(git -C '+OPS+' status --porcelain)"')
+                if s['status']=='absent':execute(node,"test ! -e /etc/dpl-node/config.yml && test ! -e /etc/systemd/system/dpl-node.service && test ! -e /etc/dpl-node/batch-owner.json && ! ss -lntH '( sport = :"+str(n['listen_port'])+" )' | grep -q .")
+                if not php(panel,'echo json_encode(App\\Models\\ServerGroup::where("id",'+str(n['formal_group_id'])+')->exists());'):return receipt('conflict')
+                self.prepare_account(n,panel,execute)
+                r=receipt('ready');r['resume_dependencies']=self.dependencies(n,panel,gateway,node,stage_probe,read,execute,gateway_py);return r
+            if a.phase=='stage':
+                if a.action=='probe':
+                    r=stage_probe(panel);return receipt(r['status'],r.get('retry_safe',False))
+                command='cd '+APP+' && php '+OPS+'/stage.php '+ ' '.join('--'+k+'='+shlex.quote(str(v)) for k,v in {**parameters(),'backup':B}.items())
+                execute(panel,command);return receipt('ready')
+            if a.phase=='dns':
+                try:
+                    result=gateway_py(gateway,'import socket,json\nprint(json.dumps(sorted(set(x[4][0] for x in socket.getaddrinfo('+repr(n['tls_domain'])+',443,socket.AF_INET)))))')
+                    return receipt('ready' if json.loads(result)==[n['node_address']] else 'conflict')
+                except Exception:return receipt('waiting')
+            if a.phase=='provision':
+                owner='/etc/dpl-node/batch-owner.json';prior=read(node,owner)
+                if prior is not None and json.loads(prior).get('request_sha256')!=sha:return receipt('conflict')
+                if prior is None and read(node,'/etc/systemd/system/dpl-node.service') is not None:return receipt('conflict')
+                done=read(node,B+'/provision-ok.json')
+                if a.action=='probe':
+                    if done and json.loads(done).get('request_sha256')==sha:
+                        for file,target in [('node-config.yml','/etc/dpl-node/config.yml'),('node-secret.env','/etc/dpl-node/secret.env')]:
+                            if read(panel,B+'/'+file)!=read(node,target):return receipt('conflict')
+                        execute(node,"echo '55bf71fa9d9f2048d3255ae7c0af929a41897ca7743f6ead34132a6ca4c79043  /opt/dpl-node/xboard-node' | sha256sum -c >/dev/null")
+                        assert_old(node);execute(node,'systemctl is-active --quiet dpl-node && openssl x509 -checkend 86400 -noout -in /etc/letsencrypt/live/'+n['tls_domain']+'/fullchain.pem')
+                        return receipt('ready')
+                    return receipt('absent',True)
+                if prior is None:
+                    execute(node,'install -d -m 700 /etc/dpl-node');write(node,owner,json.dumps({'request_sha256':sha,'key':key}))
+                transfer(panel,gateway,node)
+                if read(node,'/etc/systemd/system/dpl-node.service') is None:
+                    execute(node,'bash '+NP+'/install-node.sh --listen-port='+str(n['listen_port'])+' --tls-domain='+n['tls_domain']+' > '+B+'/install-node.log 2>&1',300)
+                execute(node,"echo '55bf71fa9d9f2048d3255ae7c0af929a41897ca7743f6ead34132a6ca4c79043  /opt/dpl-node/xboard-node' | sha256sum -c >/dev/null")
+                domain=n['tls_domain']
+                if read(node,'/etc/letsencrypt/live/'+domain+'/fullchain.pem') is None:
+                    execute(node,"certbot certonly --standalone --non-interactive --agree-tos --no-directory-hooks --cert-name "+domain+' -d '+domain+" --deploy-hook '/bin/sh "+NP+"/renew-node.sh' > "+B+'/certbot.log 2>&1',300)
+                execute(node,'systemctl enable --now dpl-node > '+B+'/enable.log 2>&1 && systemctl is-active --quiet dpl-node',60)
+                execute(node,'certbot renew --cert-name '+domain+' --dry-run --no-random-sleep-on-renew --run-deploy-hooks --no-directory-hooks > '+B+'/renew-dry-run.log 2>&1',600)
+                assert_old(node)
+                write(node,B+'/provision-ok.json',json.dumps({'request_sha256':sha,'verified_at':int(time.time())}))
+                return receipt('ready')
+            if a.phase=='relay':
+                try:
+                    result=gateway_py(gateway,'import ssl,socket,json\nc=ssl.create_default_context()\nwith socket.create_connection(('+repr(n['relay_host'])+','+str(n['relay_port'])+'),10) as s:\n with c.wrap_socket(s,server_hostname='+repr(n['tls_domain'])+') as t: print(json.dumps({"tls":t.version()}))')
+                    return receipt('ready' if json.loads(result).get('tls') else 'waiting')
+                except Exception:return receipt('waiting')
+            if a.phase=='verify':
+                s=state(panel)
+                if s['qa_banned'] and s['groups']==[str(n['formal_group_id'])]:
+                    return receipt('ready' if read(panel,B+'/qa-report.json') else 'unknown')
+                existing=read(panel,B+'/qa-report.json')
+                if existing:
+                    report=json.loads(existing)
+                    if report.get('state')=='live_verified' and report.get('qa_user_id')==s['manifest']['test_user_id'] and 0<=time.time()-report.get('verified_at',0)<=900 and s['node_online'] and s['u']+s['d']>0:
+                        r=receipt('ready');r['valid_until']=min(r['valid_until'],report['verified_at']+900);return r
+                transfer(panel,gateway,node);fetch_qa(gateway)
+                execute(gateway,'python3 '+OPS+'/verify-node.py --qa '+Q+'/qa.json --subscription '+Q+'/subscription.json --binary '+ctx['client_binary']+' --port '+str(n['socks_port'])+' --expected-exit '+shlex.quote(n['expected_exit'])+' --expected-country '+n['country']+' --report '+Q+'/qa-report.json --run > '+Q+'/verify.stdout 2> '+Q+'/verify.stderr',300)
+                data=read(gateway,Q+'/qa-report.json');report=json.loads(data)
+                write(panel,B+'/qa-report.json',data)
+                # Reports contain endpoints and measured results, no passwords or tokens.
+                evidence={'key':key,'request_sha256':sha,'report':report,'panel':state(panel)}
+                (Path(a.state_dir)/(key+'-evidence.json')).write_text(json.dumps(evidence,ensure_ascii=False,indent=2))
+                for _ in range(24):
+                    s=state(panel)
+                    if s['node_online'] and s['u']+s['d']>0:
+                        r=receipt('ready');r['valid_until']=min(r['valid_until'],report['verified_at']+900);return r
+                    time.sleep(5)
+                return receipt('waiting')
+            if a.phase=='publish':
+                s=state(panel)
+                report_data=read(panel,B+'/qa-report.json')
+                review_path=Path(a.state_dir)/(key+'-reviewed.json')
+                if not report_data or not review_path.exists():return receipt('waiting')
+                report=json.loads(report_data);review=json.loads(review_path.read_text())
+                valid=review.get('request_sha256')==sha and review.get('report_sha256')==hashlib.sha256(report_data).hexdigest() and review.get('qa_user_id')==s['manifest']['test_user_id'] and review.get('verified_at')==report.get('verified_at')
+                if not valid:return receipt('waiting')
+                if s['qa_banned'] and s['groups']==[str(n['formal_group_id'])]:
+                    journal=read(panel,B+'/promotion-journal.json')
+                    if not journal:return receipt('conflict')
+                    j=json.loads(journal)
+                    return receipt('ready' if j.get('manifest')==s['manifest'] and j.get('request',{}).get('mode')=='append' and j.get('request',{}).get('group')==n['formal_group_id'] and j.get('state')=='applied' else 'conflict')
+                if not s['node_online'] or s['u']+s['d']<=0:return receipt('waiting')
+                if a.action=='probe':return receipt('absent',True)
+                cmd='cd '+APP+' && php '+OPS+'/promote.php --backup='+B+' --mode=append --group='+str(n['formal_group_id'])+' --verification='+B+'/qa-report.json'
+                execute(panel,cmd);execute(panel,cmd+' --apply');return receipt('ready')
+            if a.phase=='audit':
+                s=state(panel);assert_old(node)
+                if not s['qa_banned'] or s['groups']!=[str(n['formal_group_id'])] or not s['node_online']:return receipt('conflict')
+                ok=php(panel,'$b='+payload(B)+r''';$m=json_decode(file_get_contents($b.'/manifest.json'),true);$n=App\Models\Server::findOrFail($m['node_id']);$base=json_decode(file_get_contents('''+payload(ctx['batch_backup_root']+'/baseline.json')+r'''),true);$ok=$base['plans']===App\Models\Plan::orderBy('id')->get()->toArray() && $base['users']===App\Models\User::whereIn('id',array_column($base['users'],'id'))->orderBy('id')->get(['id','group_id','plan_id'])->toArray() && $base['subscribe_url']===admin_setting('subscribe_url');foreach($base['nodes'] as $old){$cur=App\Models\Server::find($old['id'])->toArray();foreach(['name','host','port','server_port','protocol_settings','group_ids','machine_id','enabled','show','cert_config'] as $k)if(($cur[$k]??null)!==($old[$k]??null))$ok=false;}$ids=App\Services\ServerService::getAvailableUsers($n)->pluck('id')->all();$ok=$ok&&!in_array($m['test_user_id'],$ids);echo json_encode($ok);''')
+                if not ok:return receipt('conflict')
+                self.formal(n,panel,gateway,php,execute,read,write)
+                original_gateway=read(panel,ctx['batch_backup_root']+'/gateway-baseline.json')
+                if not original_gateway or containers(gateway)!=json.loads(original_gateway):return receipt('conflict')
+                return receipt('ready')
+            return receipt('unknown')
+        panel=Lazy(self.pool,ctx['panel_vps']);gateway=Lazy(self.pool,ctx['gateway_vps']);node=Lazy(self.pool,n['vps_id'])
+        try:return phase_impl(panel,gateway,node)
+        except Exception as exc:
+            # Never print underlying library/server errors or credentials.
+            safe = str(exc) if type(exc) is RuntimeError and __import__('re').fullmatch('[a-z_0-9]+',str(exc)) else 'inspect_protected_state'
+            self.atomic_json(self.directory/(key+'-'+phase+'-error.json'),{'reason':safe})
+            return receipt('waiting' if safe in ('investigate_preserved_service_restart','formal_group_not_ready') else 'unknown')
+
+class Lazy:
+    def __init__(self,pool,rid):self.pool,self.rid=pool,rid
+    def __getattr__(self,name):return getattr(self.pool.get(self.rid),name)

scripts/node-onboarding/ssh_pool.py

@@ -0,0 +1,99 @@
+"""Thread-local SSH sessions, mandatory explicit jump, strict known-host keys."""
+import json
+import hashlib
+import base64
+import re
+import threading
+from pathlib import Path
+
+
+def private_text(path):
+    p = Path(path)
+    if not p.is_absolute() or p.is_symlink() or not p.is_file() or p.stat().st_mode & 0o077:
+        raise ValueError('Private regular absolute credential/config reference required')
+    return p.read_text()
+
+
+def private_json(path):
+    return json.loads(private_text(path))
+
+
+def credentials(spec):
+    if not spec.get('credential_ref'):return {}
+    if spec.get('credential_format','json')=='json':return private_json(spec['credential_ref'])
+    if spec['credential_format']=='codocs-password':
+        matches=re.findall(r'(?m)^[ \t]*(?:[-*][ \t]+)?密码[：:][ \t]*([^\r\n]+)$',private_text(spec['credential_ref']))
+        if len(matches)!=1:raise ValueError('Unambiguous existing CoDocs password field required')
+        return {'password':matches[0].strip()}
+    raise ValueError('Unsupported credential reference format')
+
+
+class Pool:
+    def __init__(self, config, jump_id, client_factory=None):
+        self.config, self.jump_id = config, jump_id
+        if jump_id not in config['hosts'] or config['hosts'][jump_id].get('role') != 'white_jump':
+            raise ValueError('Explicit registered white jump required')
+        if client_factory is None:
+            known=Path(config['known_hosts'])
+            if not known.is_absolute() or known.is_symlink() or not known.is_file() or known.stat().st_mode & 0o022:raise ValueError('Explicit trusted known-hosts file required')
+            if any(not re.fullmatch('SHA256:[A-Za-z0-9+/]{43}=?',s.get('host_key_sha256','')) for s in config['hosts'].values()):raise ValueError('Approved explicit host fingerprints required')
+        self.factory = client_factory
+        self.local = threading.local()
+        self.clients = []
+        self.lock = threading.Lock()
+
+    def get(self, rid):
+        sessions = getattr(self.local, 'sessions', None)
+        if sessions is None:
+            sessions = self.local.sessions = {}
+        if rid in sessions:
+            c = sessions[rid]
+            if c.get_transport() and c.get_transport().is_active():
+                return c
+            # Do not transparently retry operations after a lost connection.
+            raise RuntimeError('SSH session lost; probe remote outcome on next run')
+        spec = self.config['hosts'][rid]
+        jump = self.get(self.jump_id) if rid != self.jump_id else None
+        if self.factory:
+            client = self.factory(rid, spec, jump)
+        else:
+            import paramiko
+            client = paramiko.SSHClient()
+            known=Path(self.config['known_hosts'])
+            if not known.is_absolute() or known.is_symlink() or not known.is_file() or known.stat().st_mode & 0o022:raise ValueError('Trusted non-writable known-hosts reference required')
+            entries=paramiko.HostKeys(str(known))
+            hostname=spec['host'] if spec.get('port',22)==22 else '['+spec['host']+']:'+str(spec['port'])
+            approved=[]
+            for key in (entries.lookup(hostname) or {}).values():
+                actual='SHA256:'+base64.b64encode(hashlib.sha256(key.asbytes()).digest()).decode().rstrip('=')
+                if actual==spec['host_key_sha256'].rstrip('='):approved.append(key)
+            if not approved:raise ValueError('Approved known-host key absent; no credentials transmitted')
+            for key in approved:client.get_host_keys().add(hostname,key.get_name(),key)
+            client.set_missing_host_key_policy(paramiko.RejectPolicy())
+            auth = credentials(spec)
+            if set(auth) - {'username', 'password', 'key_filename', 'passphrase'}:
+                raise ValueError('Unexpected SSH credential fields')
+            if auth.get('username',spec['username'])!=spec['username']:raise ValueError('Credential username differs from registered host')
+            if auth.get('key_filename'):
+                key=Path(auth['key_filename'])
+                if not key.is_absolute() or not key.is_file() or key.stat().st_mode & 0o077:raise ValueError('Private absolute SSH key required')
+            sock = jump.get_transport().open_channel('direct-tcpip', (spec['host'], spec.get('port', 22)), ('127.0.0.1', 0)) if jump else None
+            try:
+                client.connect(spec['host'], port=spec.get('port', 22), username=spec['username'],
+                               password=auth.get('password'), key_filename=auth.get('key_filename'),
+                               passphrase=auth.get('passphrase'), sock=sock, allow_agent=False,
+                               look_for_keys=False, timeout=20, banner_timeout=20, auth_timeout=20)
+            except Exception:
+                client.close()
+                raise RuntimeError('SSH connect failed; inspect registered route/host key/credential reference') from None
+            finally:
+                auth.clear()
+        sessions[rid] = client
+        with self.lock:
+            self.clients.append(client)
+        return client
+
+    def close(self):
+        for client in reversed(self.clients):
+            client.close()
+        self.clients.clear()

tests/node-onboarding/AcceptanceUserIsolatedTest.php

@@ -0,0 +1,66 @@
+<?php
+/** Real Eloquent/SQLite and real ServerService. No dotenv, app providers, Redis or HTTP. */
+require_once dirname(__DIR__,2).'/vendor/autoload.php';
+require_once dirname(__DIR__,2).'/scripts/node-onboarding/panel-acceptance.php';
+use App\Models\{User,Server};
+use DplOnboarding\Acceptance;
+use Illuminate\Container\Container;
+use Illuminate\Database\Capsule\Manager as Capsule;
+use Illuminate\Support\Facades\Facade;
+use PHPUnit\Framework\TestCase;
+
+final class AcceptanceUserIsolatedTest extends TestCase {
+    private Capsule $db; private string $dir;
+    protected function setUp():void {
+        $app=new Container();Container::setInstance($app);Facade::clearResolvedInstances();Facade::setFacadeApplication($app);$app->instance('app',$app);
+        $app->instance('config',new Illuminate\Config\Repository(['cache'=>['default'=>'array','stores'=>['array'=>['driver'=>'array']]],'app'=>['key'=>'offline-test']]));
+        $events=new Illuminate\Events\Dispatcher($app);$app->instance('events',$events);
+        $this->db=new Capsule($app);$this->db->addConnection(['driver'=>'sqlite','database'=>':memory:']);$this->db->setEventDispatcher($events);$this->db->setAsGlobal();$this->db->bootEloquent();$app->instance('db',$this->db->getDatabaseManager());$app->singleton('cache',fn($a)=>new Illuminate\Cache\CacheManager($a));
+        $s=$this->db->schema();
+        $s->create('v2_user',function($t){$t->increments('id');$t->string('email')->unique();foreach(['uuid','token','password'] as $x)$t->string($x);$t->integer('group_id')->nullable();foreach(['u','d'] as $x)$t->integer($x)->default(0);$t->integer('transfer_enable');$t->integer('expired_at')->nullable();foreach(['banned','is_admin','remind_expire','remind_traffic'] as $x)$t->boolean($x)->default(0);foreach(['device_limit','speed_limit'] as $x)$t->integer($x)->nullable();$t->integer('created_at')->nullable();$t->integer('updated_at')->nullable();});
+        $s->create('v2_server_group',function($t){$t->increments('id');$t->string('name');});Capsule::table('v2_server_group')->insert([['id'=>1,'name'=>'formal'],['id'=>2,'name'=>'other']]);
+        $s->create('v2_server',function($t){$t->increments('id');foreach(['name','type','host','port'] as $x)$t->string($x);foreach(['group_ids','protocol_settings'] as $x)$t->text($x);$t->boolean('enabled')->default(1);$t->boolean('show')->default(1);$t->integer('sort')->default(1);foreach(['u','d'] as $x)$t->integer($x)->default(0);$t->integer('transfer_enable')->nullable();$t->float('rate')->default(1);$t->timestamps();});
+        $this->node();$this->dir=sys_get_temp_dir().'/dpl-acceptance-'.bin2hex(random_bytes(8));mkdir($this->dir,0700);
+    }
+    protected function tearDown():void{foreach(glob($this->dir.'/*') as $p)unlink($p);rmdir($this->dir);$this->db->getDatabaseManager()->disconnect();}
+    private function node():void {$i=Server::count()+1;Server::create(['name'=>'node'.$i,'type'=>'anytls','host'=>'relay.example.invalid','port'=>(string)(47000+$i),'group_ids'=>['1'],'protocol_settings'=>['tls'=>['server_name'=>'n'.$i.'.example.invalid','allow_insecure'=>false]]]);}
+    private function user(array $more=[]):User{return User::create(array_replace(['email'=>'existing@example.invalid','uuid'=>'existing-uuid','token'=>'existing-token','password'=>'unused','group_id'=>1,'transfer_enable'=>1000000,'expired_at'=>null],$more));}
+    private function reject(callable $f):void{$caught=false;try{$f();}catch(RuntimeException $e){$caught=true;}self::assertTrue($caught);}
+    public function testPreviewDoesNotCreateUserOrLease():void {
+        self::assertSame('would_create',Acceptance::prepare($this->dir,1,false)['state']);self::assertSame(0,User::count());self::assertFileDoesNotExist($this->dir.'/acceptance-user.json');
+    }
+    public function testCreatesBoundedNonAdminAndUsesDynamicActualNodeSet():void {
+        $r=Acceptance::prepare($this->dir,1,true);$u=User::findOrFail($r['id']);self::assertTrue($r['owned']);self::assertFalse($u->is_admin);self::assertSame(104857600,(int)$u->transfer_enable);self::assertLessThanOrEqual(time()+3600,(int)$u->expired_at);self::assertGreaterThan(time()+3500,(int)$u->expired_at);
+        self::assertCount(1,Acceptance::source($this->dir,1)['nodes']);$this->node();self::assertCount(2,Acceptance::source($this->dir,1)['nodes']);self::assertSame($u->id,Acceptance::prepare($this->dir,1,true)['id']);self::assertSame(1,User::count());
+        Acceptance::cleanup($this->dir,true);self::assertTrue($u->fresh()->banned);self::assertTrue(App\Services\ServerService::getAvailableUsers(Server::first())->isEmpty());self::assertArrayNotHasKey('token',Acceptance::source($this->dir,1,true));$this->reject(fn()=>Acceptance::source($this->dir,1));
+    }
+    public function testExistingEligibleAdminIsUsedWithoutPrivilegeOrPasswordChange():void {
+        $u=$this->user(['is_admin'=>1]);$before=$u->fresh()->toArray();$r=Acceptance::prepare($this->dir,1,true);self::assertFalse($r['owned']);self::assertSame($u->id,$r['id']);Acceptance::source($this->dir,1);Acceptance::cleanup($this->dir,true);self::assertSame($before,$u->fresh()->toArray());
+    }
+    public function testExpiredExhaustedBannedWrongGroupAreNotReused():void {
+        foreach([['expired_at'=>time()-1],['u'=>1000000],['banned'=>1],['group_id'=>2]] as $i=>$change)$this->user($change+['email'=>'existing'.$i.'@example.invalid']);
+        $r=Acceptance::prepare($this->dir,1,true);self::assertTrue($r['owned']);self::assertSame(5,User::count());
+    }
+    public function testLostCommitResponseRecoversSameUserAndPreparedRollbackRecoversSameIdentity():void {
+        $r=Acceptance::prepare($this->dir,1,true);$p=$this->dir.'/acceptance-user.json';$j=DplOnboarding\readJson($p);$j['state']='prepared';unset($j['id']);DplOnboarding\writeJson($p,$j);
+        self::assertSame($r['id'],Acceptance::prepare($this->dir,1,true)['id']);self::assertSame(1,User::count());
+        User::find($r['id'])->delete();DplOnboarding\writeJson($p,$j);$again=Acceptance::prepare($this->dir,1,true);self::assertSame(1,User::count());self::assertSame($j['uuid'],User::find($again['id'])->uuid);
+    }
+    public function testClosedOwnedUserNeverReactivatedOnResume():void {
+        $old=Acceptance::prepare($this->dir,1,true)['id'];Acceptance::cleanup($this->dir,true);Acceptance::cleanup($this->dir,true);
+        $new=Acceptance::prepare($this->dir,1,true)['id'];self::assertNotSame($old,$new);self::assertTrue(User::find($old)->banned);self::assertFalse(User::find($new)->banned);self::assertFileExists($this->dir.'/acceptance-closed-'.$old.'.json');Acceptance::cleanup($this->dir,true);
+    }
+    public function testCleanupRefusesDriftAndNeverBansSomeoneElsesAccount():void {
+        $id=Acceptance::prepare($this->dir,1,true)['id'];$u=User::find($id);$u->is_admin=true;$u->saveQuietly();$this->reject(fn()=>Acceptance::cleanup($this->dir,true));self::assertFalse($u->fresh()->banned);
+    }
+    public function testEffectivePluginExclusionBlocksAcceptanceSource():void {
+        Acceptance::prepare($this->dir,1,true);
+        App\Services\Plugin\HookManager::registerFilter('server.users.get',fn($users)=>collect());
+        $this->reject(fn()=>Acceptance::source($this->dir,1));Acceptance::cleanup($this->dir,true);
+    }
+    public function testFailureFinallyCleanupAndMissingGroup():void {
+        $this->reject(fn()=>Acceptance::prepare($this->dir,99,true));$id=Acceptance::prepare($this->dir,1,true)['id'];
+        try{throw new RuntimeException('Injected client failure');}catch(RuntimeException $e){}finally{Acceptance::cleanup($this->dir,true);}
+        self::assertTrue(User::find($id)->banned);
+    }
+}

tests/node-onboarding/test_runtime.py

@@ -0,0 +1,312 @@
+"""Offline adversarial evidence/transport/resume tests; no network or Laravel env."""
+import copy
+import hashlib
+import importlib.util
+import json
+from pathlib import Path
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from unittest.mock import patch
+
+SCRIPTS=Path(__file__).resolve().parents[2]/'scripts/node-onboarding'
+sys.path.insert(0,str(SCRIPTS))
+from acceptance import audit,digest,snapshot,service_change,provider_state
+from ssh_pool import Pool,private_json,credentials
+import batch
+
+class EvidenceTests(unittest.TestCase):
+    def fixture(self,count):
+        nodes=[dict(id=i,type='anytls',tag='node-'+str(i),relay='relay.example.invalid',relay_port=47000+i,sni=f'n{i}.example.invalid',expected_exit=f'192.0.2.{i}',country='JP') for i in range(1,count+1)]
+        report={'state':'live_verified','verified_at':1000,'results':[dict(tag=n['tag'],relay=n['relay'],relay_port=n['relay_port'],sni=n['sni'],exit_ip=n['expected_exit'],country=n['country'],live=True,strict_tls=True) for n in nodes]}
+        qas={'captured_at':1000,'plan_sha256':'plan','users':{'new':dict(id=9,http_status=403,banned=True,excluded=True)}}
+        return nodes,report,qas
+    def check(self,nodes,report,qas,after=None,**bindings):
+        raw=json.dumps(report).encode();qraw=json.dumps(qas).encode()
+        binding=dict(plan_sha256='plan',snapshot_sha256=digest(snapshot(nodes)),report_sha256=hashlib.sha256(raw).hexdigest(),qa_rejections_sha256=hashlib.sha256(qraw).hexdigest(),verified_at=1000)
+        binding.update(bindings)
+        return audit(nodes,nodes if after is None else after,raw,qraw,binding,'plan',{'new':9},now=1100)
+    def test_arbitrary_node_counts_and_reordering(self):
+        for count in [1,2,7,23]:
+            nodes,report,qas=self.fixture(count)
+            self.assertEqual(self.check(nodes,report,qas,after=list(reversed(nodes)))['nodes'],count)
+    def test_topology_change_including_old_node_is_blocked(self):
+        nodes,r,q=self.fixture(7)
+        for after in [nodes[:-1],nodes+self.fixture(8)[0][-1:],list(reversed(nodes[:-1]))]:
+            with self.assertRaises(ValueError):self.check(nodes,r,q,after=after)
+        altered=copy.deepcopy(nodes);altered[0]['expected_exit']='198.51.100.9'
+        with self.assertRaises(ValueError):self.check(nodes,r,q,after=altered)
+    def test_missing_extra_duplicate_or_insecure_results_fail(self):
+        nodes,r,q=self.fixture(7)
+        for results in [r['results'][:-1],r['results']+[r['results'][0]],r['results'][1:]+[r['results'][1]]]:
+            with self.assertRaises(ValueError):self.check(nodes,dict(r,results=results),q)
+        for field,value in [('exit_ip','198.51.100.1'),('country','US'),('strict_tls',False),('live',False),('sni','wrong.example.invalid')]:
+            wrong=copy.deepcopy(r);wrong['results'][0][field]=value
+            with self.assertRaises(ValueError):self.check(nodes,wrong,q)
+    def test_qa_identity_status_revocation_and_hashes_are_bound(self):
+        nodes,r,q=self.fixture(2)
+        for field,value in [('id',10),('http_status',200),('banned',False),('excluded',False)]:
+            wrong=copy.deepcopy(q);wrong['users']['new'][field]=value
+            with self.assertRaises(ValueError):self.check(nodes,r,wrong)
+        for field in ['snapshot_sha256','plan_sha256','report_sha256','qa_rejections_sha256']:
+            with self.assertRaises(ValueError):self.check(nodes,r,q,**{field:'wrong'})
+        for stamp in [-1,1200,None,'1000']:
+            with self.assertRaises(ValueError):self.check(nodes,dict(r,verified_at=stamp),q)
+            with self.assertRaises(ValueError):self.check(nodes,r,dict(q,captured_at=stamp))
+    def test_provider_observations_are_exact_and_not_language_or_substring_dependent(self):
+        node=dict(key='new',tls_domain='a.example.invalid',relay_host='relay.example.invalid',relay_port=47001,node_address='192.0.2.1',listen_port=24443)
+        observed={'captured_at':1000,'dns':[{'type':'A','name':node['tls_domain'],'address':node['node_address'],'proxy':False}],'relays':[{'name':'dpl-new-anytls','host':node['relay_host'],'port':47001,'target':node['node_address'],'listen_port':24443,'proxy_protocol_in':False,'proxy_protocol_out':False}]}
+        self.assertTrue(provider_state(observed,node,1100)['dns'])
+        for section,field,value in [('dns','proxy',True),('dns','address','192.0.2.11'),('relays','proxy_protocol_in',True),('relays','target','192.0.2.11')]:
+            wrong=copy.deepcopy(observed);wrong[section][0][field]=value
+            with self.assertRaises(ValueError):provider_state(wrong,node,1100)
+        self.assertEqual(provider_state(dict(observed,dns=[],relays=[]),node,1100),{'dns':[],'relays':[]})
+        with self.assertRaises(ValueError):provider_state(observed,node,2000)
+
+    def test_registry_and_service_changes_do_not_use_pid_alone(self):
+        for values in [[],[dict(self.fixture(1)[0][0],type='vless')],self.fixture(1)[0]*2]:
+            with self.assertRaises(ValueError):snapshot(values)
+        before=dict(unit_sha256='a',binary_sha256='b',config_sha256='c',listeners=['*:443'],active=True,pid=1,started='yesterday',restarts=0)
+        self.assertEqual(service_change(before,before),'ready')
+        self.assertEqual(service_change(before,dict(before,pid=2)),'investigate_restart')
+        self.assertEqual(service_change(before,dict(before,config_sha256='changed')),'conflict')
+        self.assertEqual(service_change(before,dict(before,active=False)),'conflict')
+        self.assertEqual(service_change(before,dict(before,binary_sha256=None)),'unknown')
+
+class TransportTests(unittest.TestCase):
+    def test_reuse_and_every_nonwhite_host_has_jump(self):
+        calls=[]
+        class Client:
+            active=True
+            def get_transport(self):return self
+            def is_active(self):return self.active
+            def close(self):self.active=False
+        def factory(rid,spec,jump):calls.append((rid,jump));return Client()
+        config={'hosts':{'white':{'role':'white_jump'},'panel':{},'node':{}}}
+        pool=Pool(config,'white',factory)
+        panel=pool.get('panel');self.assertIs(panel,pool.get('panel'))
+        node=pool.get('node');self.assertIs(calls[1][1],calls[2][1]);self.assertIsNone(calls[0][1]);self.assertEqual(len(calls),3)
+        panel.active=False
+        with self.assertRaises(RuntimeError):pool.get('panel')
+        self.assertEqual(len(calls),3) # Never automatically replay after disconnect.
+        pool.close();self.assertFalse(node.active)
+    def test_sessions_are_not_shared_unsafely_between_workers(self):
+        barrier=threading.Barrier(2);made=[]
+        class Client:
+            def get_transport(self):return self
+            def is_active(self):return True
+            def close(self):pass
+        pool=Pool({'hosts':{'w':{'role':'white_jump'},'n':{}}},'w',lambda *args:Client())
+        def worker():
+            n=pool.get('n');barrier.wait();made.append(n);self.assertIs(n,pool.get('n'))
+        threads=[threading.Thread(target=worker) for _ in range(2)]
+        for t in threads:t.start()
+        for t in threads:t.join(timeout=5)
+        self.assertEqual(len(made),2);self.assertIsNot(made[0],made[1]);pool.close()
+    def test_existing_codocs_credentials_are_read_without_copy_or_rewrite(self):
+        with tempfile.TemporaryDirectory() as d:
+            p=Path(d)/'vps.md';body='# Fixture\n密码：test-only-phrase\n';p.write_text(body);p.chmod(0o600)
+            self.assertEqual(credentials({'credential_ref':str(p),'credential_format':'codocs-password'}),{'password':'test-only-phrase'})
+            self.assertEqual(p.read_text(),body);self.assertEqual(len(list(Path(d).iterdir())),1)
+            p.write_text(body+'密码：ambiguous\n')
+            with self.assertRaises(ValueError):credentials({'credential_ref':str(p),'credential_format':'codocs-password'})
+    def test_private_config_and_explicit_jump_enforced(self):
+        with self.assertRaises(ValueError):Pool({'hosts':{'n':{}}},'w')
+        with self.assertRaises(ValueError):Pool({'hosts':{'w':{}}},'w')
+        with tempfile.TemporaryDirectory() as d:
+            p=Path(d)/'auth.json';p.write_text('{}');p.chmod(0o644)
+            with self.assertRaises(ValueError):private_json(p)
+            p.chmod(0o600);self.assertEqual(private_json(p),{})
+            link=Path(d)/'link';link.symlink_to(p)
+            with self.assertRaises(ValueError):private_json(link)
+
+class ResumeTests(unittest.TestCase):
+    def setUp(self):
+        batch.STOP.clear();self.tmp=tempfile.TemporaryDirectory();self.root=Path(self.tmp.name)
+        spec=importlib.util.spec_from_file_location('batch_fixtures',Path(__file__).with_name('test_batch.py'));m=importlib.util.module_from_spec(spec);spec.loader.exec_module(m)
+        fixture=m.BatchTests();fixture.setUp();self.fixture=fixture;self.plan=fixture.plan
+        self.calls=[];self.generation='a';self.finished=False
+        owner=self
+        class Runner(batch.Coordinator):
+            def call(self,n,phase,action):
+                owner.calls.append((phase,action))
+                r=dict(key=n['key'],phase=phase,request_sha256='irrelevant-in-unit-fixture',status='waiting' if phase=='audit' and not owner.finished else 'ready',valid_until=time.time()+180)
+                if phase=='preflight':r['resume_dependencies']={k:hashlib.sha256((owner.generation if k=='provision' else k).encode()).hexdigest() for k in ['stage','dns','provision','relay','verify']}
+                return r
+        self.Runner=Runner
+    def tearDown(self):self.fixture.tearDown();self.tmp.cleanup();batch.STOP.clear()
+    def run_batch(self):return self.Runner(self.plan,self.root/'ledger',jobs=1).run()
+    def test_fresh_proofs_skip_only_completed_unchanged_phases(self):
+        self.assertFalse(self.run_batch());self.calls.clear();self.finished=True
+        self.assertTrue(self.run_batch());self.assertEqual(self.calls,[('preflight','probe'),('publish','probe'),('audit','probe')])
+    def test_dependency_change_rechecks_downstream_and_not_unrelated_stage(self):
+        self.run_batch();self.calls.clear();self.generation='b';self.finished=True;self.run_batch()
+        self.assertEqual([p for p,a in self.calls],['preflight','provision','relay','verify','publish','audit'])
+    def test_expired_and_unknown_proofs_never_skip(self):
+        self.run_batch();runner=self.Runner(self.plan,self.root/'ledger',jobs=1)
+        with runner.ledger.db() as c:c.execute('UPDATE evidence SET expires=0 WHERE phase="dns"')
+        runner.ledger.mark(self.plan['batch_id'],self.plan['nodes'][0]['key'],'stage','unknown')
+        self.calls.clear();self.finished=True;self.run_batch()
+        self.assertIn(('stage','probe'),self.calls);self.assertIn(('dns','probe'),self.calls)
+        self.assertTrue((self.root/'ledger/events.jsonl').is_file())
+
+
+class SiteAdapterTests(unittest.TestCase):
+    def test_preserved_service_review_binds_original_and_exact_observation(self):
+        from site_adapter import SiteAdapter
+        adapter=object.__new__(SiteAdapter);adapter.ctx={'preserved_service':dict(unit='XrayR',binary='/bin/xrayr',config='/etc/xrayr.yml',port=443)}
+        current=dict(unit_sha256='u',binary_sha256='b',config_sha256='c',listeners=['*:443'],active=True,pid=1,started='a',restarts=0)
+        files={};read=lambda client,path:files.get(path)
+        def write(client,path,value):files[path]=value.encode() if isinstance(value,str) else value
+        def execute(client,cmd):return json.dumps(current) if cmd.startswith('python3') else ''
+        adapter.guard_service(None,'/backup',read,write,execute)
+        original=copy.deepcopy(current);current.update(pid=2,started='b',restarts=1)
+        with self.assertRaisesRegex(RuntimeError,'investigate_'):adapter.guard_service(None,'/backup',read,write,execute)
+        files['/backup/preserved-service-review.json']=json.dumps(dict(before_sha256=digest(original),after_sha256=digest(current),reason='Independently reviewed lifecycle panic',evidence_sha256='e'*64)).encode()
+        adapter.guard_service(None,'/backup',read,write,execute)
+        current['config_sha256']='changed'
+        with self.assertRaisesRegex(RuntimeError,'conflict'):adapter.guard_service(None,'/backup',read,write,execute)
+        self.assertEqual(json.loads(files['/backup/preserved-service-original.json']),original)
+    def test_generated_remote_python_and_group_independent_php_compile(self):
+        import ast
+        import subprocess
+        import io
+        from site_adapter import SiteAdapter
+        tree=ast.parse((SCRIPTS/'site_adapter.py').read_text())
+        for method in [n for n in ast.walk(tree) if isinstance(n,ast.FunctionDef) and n.name in ('formal','guard_service','certificate_gate')]:
+            for c in ast.walk(method):
+                if isinstance(c,ast.Assign) and any(isinstance(t,ast.Name) and t.id=='script' for t in c.targets):
+                    program=eval(compile(ast.Expression(c.value),'script-builder','eval'),{'repr':repr,'spec':{},'ctx':{'subscription_url':'https://sub.example.invalid'},'value':{'token':'test-only'}})
+                    compile(program,'remote-'+method.name,'exec')
+        with tempfile.TemporaryDirectory() as d:
+            a=object.__new__(SiteAdapter);a.plan={'context':dict(panel_vps='panel',gateway_vps='gateway',node_root='/node',ops_root='/ops',app_root='/app',batch_backup_root='/batch',node_backup_root='/backups',qa_root='/qa',gateway_containers=['gateway-a','gateway-b'])};a.ctx=a.plan['context'];a.directory=Path(d)
+            php_sources=[];calls=[]
+            class Channel:
+                def shutdown_write(self):pass
+                def recv_exit_status(self):return 0
+            class In:
+                channel=Channel()
+                def write(self,body):php_sources.append(body)
+            class Out:
+                channel=Channel()
+                def read(self):
+                    body=php_sources[-1]
+                    if '$base=json_decode' in body:return b'true'
+                    return json.dumps({'manifest':{},'node_online':True,'groups':['7'],'qa_banned':True,'u':1,'d':1}).encode()
+            class DockerOut:
+                channel=Channel()
+                def read(self):return json.dumps([dict(Name='/gateway-a',Id='id-a',State={'StartedAt':'then'},RestartCount=0)]).encode()
+            class Sftp:
+                def __enter__(self):return self
+                def __exit__(self,*args):pass
+                def open(self,path,mode):return io.BytesIO(json.dumps({'/gateway-a':{'id':'id-a','started':'then','restarts':0}}).encode())
+            class Remote:
+                def exec_command(self,cmd,timeout=0):
+                    calls.append(cmd)
+                    if cmd.startswith('docker inspect'):
+                        if cmd!='docker inspect gateway-a gateway-b':raise RuntimeError('Invalid gateway inventory command')
+                        return In(),DockerOut(),io.BytesIO()
+                    return In(),Out(),io.BytesIO()
+                def open_sftp(self):return Sftp()
+            class Pool:
+                def get(self,rid):return Remote()
+            a.pool=Pool();a.guard_service=lambda *args:None;a.formal=lambda *args:None
+            n=dict(key='sg-new',request_sha256='a'*64,formal_group_id=7,vps_id='node')
+            result=a.call(n,'audit','probe');self.assertEqual(result['status'],'ready');self.assertGreaterEqual(len(php_sources),2)
+            for php in php_sources:
+                result=subprocess.run(['php','-l'],input=php,text=True,capture_output=True)
+                self.assertEqual(result.returncode,0,result.stderr)
+    def test_pool_real_connect_interface_keeps_strict_keys_and_jump_sock(self):
+        import types
+        recorded=[]
+        class FakeClient:
+            def load_host_keys(self,path):recorded.append(('keys',path))
+            def set_missing_host_key_policy(self,p):recorded.append(('reject',p))
+            def get_host_keys(self):return self
+            def add(self,*args):recorded.append(('approved-key',args))
+            def connect(self,host,**kw):recorded.append(('connect',host,kw))
+            def get_transport(self):return self
+            def is_active(self):return True
+            def open_channel(self,kind,target,origin):recorded.append(('jump',kind,target));return 'through-white'
+            def close(self):pass
+        class Key:
+            def asbytes(self):return b'approved-public-key-fixture'
+            def get_name(self):return 'ssh-ed25519'
+        class Keys:
+            def __init__(self,path):pass
+            def lookup(self,name):return {'ssh-ed25519':Key()}
+        module=types.SimpleNamespace(SSHClient=FakeClient,RejectPolicy=lambda:'reject-unknown',HostKeys=Keys)
+        pin='SHA256:'+__import__('base64').b64encode(hashlib.sha256(Key().asbytes()).digest()).decode().rstrip('=')
+        with tempfile.TemporaryDirectory() as d,patch.dict(sys.modules,{'paramiko':module}):
+            keys=Path(d)/'known_hosts';keys.write_text('public fixture');keys.chmod(0o600)
+            credentials=Path(d)/'credentials.json';credentials.write_text(json.dumps({'password':'test-only-fixture'}));credentials.chmod(0o600)
+            hosts={'w':dict(role='white_jump',host='192.0.2.1',username='root',credential_ref=str(credentials)),'n':dict(host='192.0.2.2',username='root',credential_ref=str(credentials))}
+            
+            for host in hosts.values():host['host_key_sha256']=pin
+            p=Pool(dict(hosts=hosts,known_hosts=str(keys)),'w');p.get('n');p.get('n');p.close()
+            before_connects=len([x for x in recorded if x[0]=='connect'])
+            hosts['w']['host_key_sha256']='SHA256:'+'b'*43
+            blocked=Pool(dict(hosts=hosts,known_hosts=str(keys)),'w')
+            with self.assertRaises(ValueError):blocked.get('n')
+            self.assertEqual(len([x for x in recorded if x[0]=='connect']),before_connects)
+            blocked.close()
+        connections=[x for x in recorded if x[0]=='connect'];self.assertEqual(len(connections),2);self.assertIsNone(connections[0][2]['sock']);self.assertEqual(connections[1][2]['sock'],'through-white')
+        self.assertEqual(len([x for x in recorded if x[0]=='reject']),2)
+        for _,_,kw in connections:self.assertFalse(kw['allow_agent']);self.assertFalse(kw['look_for_keys'])
+
+    def test_configure_is_local_only_and_pins_code_context_and_interpreter(self):
+        import types
+        spec=importlib.util.spec_from_file_location('configure_site',SCRIPTS/'configure-site.py');module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
+        with tempfile.TemporaryDirectory() as d:
+            root=Path(d);ssh=root/'ssh.json';registry=root/'exits.json';(root/'known_hosts').write_text('public fixture');(root/'known_hosts').chmod(0o600)
+            hosts={'VPS-001':dict(role='white_jump',host='192.0.2.1',username='root'),'VPS-002':dict(host='192.0.2.2',username='root'),'VPS-003':dict(host='192.0.2.3',username='root'),'VPS-011':dict(host='192.0.2.11',username='root')}
+            for host in hosts.values():host['host_key_sha256']='SHA256:'+'a'*43
+            ssh.write_text(json.dumps({'hosts':hosts,'known_hosts':str(root/'known_hosts')}));ssh.chmod(0o600);registry.write_text('{}');registry.chmod(0o600)
+            fixture_spec=importlib.util.spec_from_file_location('base_fixture',Path(__file__).with_name('test_batch.py'));f=importlib.util.module_from_spec(fixture_spec);fixture_spec.loader.exec_module(f)
+            original=f.BatchTests();original.setUp()
+            try:
+                base=original.plan;node=base['nodes'][0];node['node_address']='192.0.2.11';node['expected_exit']='192.0.2.11';base['context'].update(jump_vps='VPS-001',panel_vps='VPS-002',gateway_vps='VPS-003');registry.write_text(json.dumps({node['tls_domain']:{'expected_exit':node['expected_exit'],'country':node['country']}}))
+                site=dict(client_sha256='d'*64,ops_root='/ops',node_root='/node',batch_backup_root='/backup/batch',node_backup_root='/backup/nodes',qa_root='/qa/batch',provider_snapshot=str(root/'provider.json'),exit_registry=str(registry),ssh_config=str(ssh),preserved_service=dict(unit='XrayR',binary='/bin/xrayr',config='/etc/xrayr.yml',port=443),formal_socks_port=19999,gateway_containers=['gateway-a'])
+                with patch.dict(sys.modules,{'paramiko':types.SimpleNamespace(__version__='5.0.0')}):
+                    plan=module.configure(base,site);self.assertEqual(plan['runtime']['kind'],'ssh-site-v1');self.assertFalse((root/'ledger').exists())
+                    runner=batch.Coordinator(plan,root/'ledger',jobs=1);runner.runtime.close();runner.check_runtime()
+                    ssh.write_text(ssh.read_text()+'\n')
+                    with self.assertRaises(ValueError):runner.check_runtime()
+            finally:original.tearDown()
+
+class PackagingTests(unittest.TestCase):
+    def test_committed_flat_snapshot_preserves_worktree_index_and_base(self):
+        import subprocess
+        spec=importlib.util.spec_from_file_location('bundle_ops',SCRIPTS/'bundle-ops.py');module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
+        with tempfile.TemporaryDirectory() as d:
+            root=Path(d);repo=root/'repo';repo.mkdir()
+            def git(*args):return subprocess.check_output(['git','-C',str(repo),*args],stderr=subprocess.DEVNULL)
+            git('init');git('config','user.name','Offline Test');git('config','user.email','test@example.invalid');git('config','commit.gpgsign','false')
+            (repo/'base-only.txt').write_text('preserved operations base');git('add','.');git('commit','-m','base');base=git('rev-parse','HEAD').decode().strip()
+            source_dir=repo/'scripts/node-onboarding';source_dir.mkdir(parents=True)
+            for name in module.FILES:(source_dir/name).write_text('committed '+name)
+            git('add','.');git('commit','-m','source');source=git('rev-parse','HEAD').decode().strip()
+            (repo/'base-only.txt').write_text('unrelated user modification');(repo/'unrelated.txt').write_text('staged unrelated');git('add','unrelated.txt')
+            index_before=git('diff','--cached','--binary');status_before=git('status','--porcelain')
+            output=root/'operations.bundle';result=module.build(repo,source,base,'codex/offline-ops',output)
+            self.assertTrue(output.exists());self.assertEqual(git('diff','--cached','--binary'),index_before);self.assertEqual(git('status','--porcelain'),status_before)
+            self.assertEqual(git('show',result['commit']+':panel-acceptance.php'),b'committed panel-acceptance.php')
+            self.assertEqual(git('show',result['commit']+':base-only.txt'),b'preserved operations base')
+            self.assertEqual(git('rev-parse','HEAD').decode().strip(),source)
+            git('bundle','verify',str(output))
+            with self.assertRaises(ValueError):module.build(repo,source,base,'codex/offline-ops',root/'second.bundle')
+    def test_event_stream_preserves_final_result_and_avoids_status_polling(self):
+        import subprocess
+        spec=importlib.util.spec_from_file_location('event_fixture',Path(__file__).with_name('test_batch.py'));f=importlib.util.module_from_spec(spec);spec.loader.exec_module(f)
+        test=f.BatchTests();test.setUp()
+        try:
+            result=subprocess.run(test.argv()+['--events'],capture_output=True,text=True,timeout=30)
+            self.assertEqual(result.returncode,0,result.stderr)
+            lines=[json.loads(x) for x in result.stdout.splitlines()]
+            self.assertTrue(lines[-1]['complete']);self.assertTrue(any(x.get('phase')=='provision' for x in lines[:-1]))
+            for x in lines[:-1]:self.assertEqual(set(x),{'batch','node','phase','status','time'})
+        finally:test.tearDown()
+
+if __name__=='__main__':unittest.main()