feat(nodes): scope final acceptance to reviewed change impact

docs/maintenance/NODE_BATCH.md

@@ -134,7 +134,7 @@ PID/启动时间漂移应先停止写入，不直接刷新基线。读取服务
 1. 主协调者在生产写入前启动/复用一个核查验收助手，给出当前任务范围、必要资料入口、受保护资源、分阶段输出和完成标准。当前批次已存在对应子助手时继续使用，不在每个节点或每次续跑时重新创建。
 2. 助手分阶段核查资源与审查方案；主协调者并行准备参数与调度。助手完成冲突核查和相关隔离测试/审查后，主协调者解决影响写入的缺口，固定同一参数单，再运行全批 preflight。检查发现变更时重新核对受影响项，不把旧回执沿用到不同参数。
 3. 独立节点由脚本按 `--jobs` 并发，面板/DNS/中转按共享锁执行。正式开放前，核查验收助手核对本节点隔离 QA、严格 TLS、实际出口与上报证据；主协调者确认后执行 publish。助手不得另起同批调度、重复 stage、共用浏览器会话切换账户或绕开账本。
-4. 正式开放后，核查验收助手检查正式订阅、QA 禁用/拒绝、旧资源基线并报告通过或具体缺口。主协调者核实后逐节点报告完成、等待或需用户操作；批次全部完成必须有对应实际验收，不能以子助手结束作为完成证明。
+4. 正式开放后，核查验收助手按[影响范围](NODE_BATCH_RUNTIME.md#按影响范围验收2026-10-20)检查完整正式订阅、实际权限、QA 禁用/拒绝、旧资源基线及分层联网结果，并报告通过或具体缺口。轻量TLS不能写成出口/流量验收通过；影响不明或共享变化回到全量。主协调者核实后逐节点报告完成、等待或需用户操作；批次全部完成必须有对应实际验收，不能以子助手结束作为完成证明。
 5. 子助手不可用、并发名额不足或失败时，主协调者明确报告，复用/恢复角色或等待名额；不自动追加更多助手、跳过必需角色或把未知写成通过。暂停和停止指令必须传达给子助手，并停止派发后续写入；已发出的远端操作仍按断点恢复规则回读。
 
 主协调者与一个子助手分别消耗处理其输入和输出所需的 token，总量不是固定倍数。主协调者默认仅传必要背景、资料路径、节点参数与任务边界，避免复制整段历史；每项检查明确归属，汇报用结论、证据位置与缺口，不反复粘贴完整日志。独立验收保留必要的交叉核对，不为节省 token 省略验收。没有固定 token 配额或价格承诺。

docs/maintenance/NODE_BATCH_RUNTIME.md

@@ -96,12 +96,36 @@ SSH会话仅在本次进程、同一工作线程内复用，线程不共用可
 
 准备阶段先按正式组、banned、额度和有效期选既有账号，优先管理员且最多检查20个候选，避免全表巡检客户凭据；检查真实`ServerService::getAvailableServers`及`getAvailableUsers`，包括插件过滤；候选被实际节点下发过滤时不算可用。未找到可用账号时自动创建本任务测试用户：非管理员、100MiB、1小时、1设备、无订单/套餐改写及通知消息。私有journal在DB提交前记录身份，丢失回包时恢复同一用户，不重复创建或接管同名用户。
 
-新节点仍沿用各自隔离QA→严格TLS/出口/上报→独立助手按精确报告审核→append开放并禁用QA。适配器直接生成`key-evidence.json`，包含原始报告字节SHA256、请求摘要、QA身份、报告时间及最新在线/双向上报状态，600原子保存，不再需要任务专用采集脚本。助手完成独立审核后，主协调者在同目录记录`key-reviewed.json`：`request_sha256`、`report_sha256`、`qa_user_id`、`verified_at`必须与该报告完全相同；未审核不得自动填批准文件。全部本组候选开放后，正式验收按账号**当前实际应获得的集合**生成完整快照，与客户端完整订阅逐条校验，并逐条实际联网。集合绑定节点ID、标签、端点、SNI、独立预期出口/国家；验收后再取一次集合，增加、删除或改变任一项均阻止旧报告通过，不固定节点数。旧节点同样必须覆盖。
+新节点仍沿用各自隔离QA→严格TLS/出口/上报→独立助手按精确报告审核→append开放并禁用QA。适配器直接生成`key-evidence.json`，包含原始报告字节SHA256、请求摘要、QA身份、报告时间及最新在线/双向上报状态，600原子保存，不再需要任务专用采集脚本。助手完成独立审核后，主协调者在同目录记录`key-reviewed.json`：`request_sha256`、`report_sha256`、`qa_user_id`、`verified_at`必须与该报告完全相同；未审核不得自动填批准文件。全部本组候选开放后，正式验收按账号**当前实际应获得的集合**生成完整快照，与客户端完整订阅逐条校验。集合绑定节点ID、标签、端点、SNI、独立预期出口/国家；验收后再取一次集合，增加、删除或改变任一项均阻止旧报告通过，不固定节点数。所有新旧节点均覆盖结构、权限和必要连通检查，完整联网测试范围按下节决定。
 
 每组正式客户端串行运行，保留完整订阅DNS/路由，只给诊断目的地固定出站。新建测试用户可能等待节点拉取用户列表；只对通用客户端运行失败最多做3轮诊断读测试，间隔15秒且共用900秒预算。结构、TLS或出口断言错误不重试；不重放生产写入、不另建用户来绕过失败。证据绑定完整计划摘要、节点集合摘要、原始报告及QA拒绝文件SHA256、QA实际ID、403、banned、实际下发排除及900秒时效。任一缺项或摘要/集合变化失败。旧节点关键配置、全部套餐、原用户关联和订阅地址按批次原始基线核对，网关容器ID/启动时间/重启数再次核对。
 
 正式验收成功或失败都禁用本任务新建的正式测试用户；既有账号不被禁用或改密码/权限。进程正常退出在finally收尾，即使等待或部分前检失败也尝试清理；遇到断网/SIGKILL无法保证立即清理，记录的归属、1小时有效期与额度是兜底，恢复后先核对旧journal。已关闭用户绝不重新启用；下一次需要用户时创建新受限租约，保留历史并在最终审计核对其403/下发排除。完成要求由服务端实际集合/订阅拒绝证明，不承诺立即踢断已经建立的客户端会话。
 
+## 按影响范围验收（2026-10-20）
+
+后续新计划分两阶段，避免把隔离测试与最终验收混成重复的全量联网：
+
+1. 每个新节点在隔离QA阶段仍完整测试：完整订阅、严格TLS/SNI、独立预期出口/国家、实际在线及双向上报；独立复核通过后才append开放。
+2. 全部开放后，每组获取一次正式账号完整订阅，对所有节点核对集合、端点、SNI、实际密码、严格TLS设置，并检查服务端实际下发权限、QA禁用/403/排除。未受影响节点只进行经中转的TCP+严格TLS握手，不重复启动客户端查询出口。此检查不证明正式账号实际鉴权或流量上报，不得标注为完整联网通过。
+3. 受影响节点重新完整联网；共享网关/订阅生成、DNS/路由规则、中转公共配置等发生变化时，全组完整联网。影响未知、没有明确评审，或使用本任务新建正式测试账号时，保守全组完整联网（新账号需验证真实认证和用户下发传播）。旧资源、套餐/用户关联、应用版本及网关容器基线保护仍执行，异常停止，不能借轻量模式放行不符基线的变化。
+
+站点JSON新增可选`acceptance_impact`，在冻结新计划前按实际变更评审填写；这不是生产操作授权：
+
+```json
+"acceptance_impact": {
+  "shared_changes": [],
+  "affected_snis": [],
+  "review": "task/impact-review.json 中的变更清单、基线对比及评审结论"
+}
+```
+
+两个空列表仅表示已核实为独立append，不能用空列表代替调查。`affected_snis`列出需重测的准确SNI（不按国家或节点数量抽样）；`shared_changes`任一非空项令全组完整联网。字段省略或null保持原全量策略，格式含糊/未知SNI停止。该对象随计划摘要固定；评审依据应保存在私有任务证据中，不能仅写“无影响”而无对比材料。出现新依赖变更，停止旧计划并重新评审，不修改已冻结摘要来继续。
+
+`verify-node.py`的完整expectations清单仍覆盖所有AnyTLS出站；每条可增加`verification: "full" | "tls"`，省略为full。`--qa`只走full。tls仅握手、不启动sing-box；full沿用实际客户端出口/国家测试。审计绑定逐标签范围摘要、原始报告、QA证据及完整集合；不能用tls结果满足full要求，漏节点/证据过期仍失败。正式报告保存`verification_scope`，区分轻量结果与完整联网结果。缓存复核仍绑定相同范围和集合，不复用隔离账号报告冒充正式账号认证。
+
+本次仅修改本地脚本与规则，不接入节点或访问线上。后续实际使用须先按Git流程生成并部署含新版`verify-node.py`的运维提交，再生成全新计划；旧冻结批次及原验收记录不改。上一批10节点完整联网结果仍按当时已执行范围保留。
+
 ## 续跑与原服务调查
 
 preflight/publish/audit始终现场执行。只有先前状态为done、收到真实前检依赖摘要、请求/代码/config未变、证据未过期，才可复用stage/dns/provision/relay/verify。前驱依赖变化使后续缓存失效；running/unknown/conflict绝不跳过。旧适配器不提供这些证据时仍按原逐阶段probe逻辑执行。生产写入仍先probe、持久化意图、apply后probe，不为速度跳过归属检查。

docs/maintenance/NODE_RUNBOOK.md

@@ -165,7 +165,7 @@ python3 /opt/dpl-gateway/provision/verify-node.py \
 | 协议/联网 | 官方客户端配置检查与完整启动结果、严格 TLS、目标 HTTPS、出口 IP 与 EXPECTED_EXIT 一致；脚本强制比对 EXPECTED_COUNTRY；仅结构核对不产生live_verified报告 |
 | 订阅 | 有效用户标准/短路径按本次范围检查，JSON/gzip、private/no-store、无 HIT/Age/挑战；不要在日志保存令牌 |
 | 上报 | readiness.php回读节点在线、last_check/last_push及QA上下行；按显式plans/group统计，缺省不选任何套餐 |
-| 正式开放 | 正式有效用户订阅与完整expectations逐一对应，并逐节点验证；不把单节点QA结果扩大成全部可用 |
+| 正式开放 | 正式有效用户订阅与完整expectations逐一对应，检查实际权限及QA清理；按[影响范围](NODE_BATCH_RUNTIME.md#按影响范围验收2026-10-20)决定完整联网或轻量TLS，不能把轻量结果称为出口/鉴权通过 |
 | QA 清理 | QA banned、其订阅拒绝、实际下发用户集合无 QA；客户端退出、临时本地秘密副本清理，服务器审计文件保留 |
 | 回归 | 原节点/中转、套餐字段/用户组、订阅 URL、网关容器 ID/启动时间/重启计数与 before 快照对照 |
 | 浏览器 | 正确反代入口，管理列表与保存结果回读；只查看后取消不算提交验收 |
@@ -181,7 +181,7 @@ GUI/手机、TUN、多地区、并发及用户明确跳过的测试分别标为
 
 ### 正式多节点验证
 
-在私有文件 FORMAL_EXPECTATIONS 保存每个AnyTLS出站的明确预期，完整覆盖订阅中的全部AnyTLS标签。每条结构：`tag`、`host`、整数`port`、`sni`、`uuid`、`expected_exit`、`expected_country`；uuid是该正式用户的实际密码，不能复制进普通参数单或文档。标签必须与订阅精确一致；不重命名节点。
+在私有文件 FORMAL_EXPECTATIONS 保存每个AnyTLS出站的明确预期，完整覆盖订阅中的全部AnyTLS标签。每条结构：`tag`、`host`、整数`port`、`sni`、`uuid`、`expected_exit`、`expected_country`；uuid是该正式用户的实际密码，不能复制进普通参数单或文档。标签必须与订阅精确一致；不重命名节点。 后续批量计划按[影响范围](NODE_BATCH_RUNTIME.md#按影响范围验收2026-10-20)生成每条`verification`：full为完整客户端联网，tls仅经中转严格TLS握手；省略保持full。完整清单和权限检查不缩减，新节点隔离QA仍必须full。
 
 ```sh
 python3 /opt/dpl-gateway/provision/verify-node.py \

docs/maintenance/STATUS.md

@@ -2,6 +2,10 @@
 
 文档核对日期：2026-10-11。以下区分本会话检查结果与用户验收确认，不承诺服务器一直保持此状态。
 
+## 2026-10-20 验收按影响范围分层（本地实现）
+
+新节点隔离QA完整测试不减；正式开放后保持动态全量订阅/权限/QA清理与原资源基线检查，已评审独立append使用逐节点严格TLS连通检查；指定受影响节点完整联网，共享变化、未知影响或本任务新建正式验收账号全组完整联网。范围绑定计划及报告，旧结果不改。参数和边界唯一入口见[正式适配器](NODE_BATCH_RUNTIME.md#按影响范围验收2026-10-20)。本次46项Python离线回归通过（含新旧集合、范围绑定、分层探针分派、完整正式调用链和新建账号强制完整测试）；未连接服务器、未部署新运维代码或进行线上验收。
+
 ## 2026-10-18 后续五节点接入完成
 
 日本JP - 05/06/07（node-jp06/07/08）及美国US - 02/03（node-us03/04）已append基础节点组1；五份新QA经独立原始字节审查后开放并禁用QA21–25。动态正式授权集合本次10节点全部完整订阅、严格TLS和预期出口/国家联网通过，QA403/实际下发排除、旧节点/套餐/用户关联/订阅地址及四网关容器基线审计通过；后台节点和服务器列表显示/切换/保存值回读通过。GemDocs AI版已登记5实例/5域名及项目和服务器关联。

scripts/node-onboarding/acceptance.py

@@ -31,7 +31,30 @@ def snapshot(nodes):
     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):
+def validate_impact(impact):
+    if impact is None:
+        return
+    if not isinstance(impact, dict) or set(impact) != {'shared_changes', 'affected_snis', 'review'}:
+        raise ValueError('Explicit reviewed impact required')
+    if not isinstance(impact['review'], str) or not impact['review'].strip():
+        raise ValueError('Impact review reference required')
+    for key in ('shared_changes', 'affected_snis'):
+        if not isinstance(impact[key], list) or any(not isinstance(x, str) or not x.strip() for x in impact[key]) or len(set(impact[key])) != len(impact[key]):
+            raise ValueError('Invalid impact inventory')
+
+
+def verification_scope(nodes, impact=None):
+    """Unknown/shared impact keeps full tests; reviewed append uses light TLS probes."""
+    nodes = snapshot(nodes)
+    if impact is None:
+        return {n['tag']: 'full' for n in nodes}
+    validate_impact(impact)
+    if not set(impact['affected_snis']).issubset({n['sni'] for n in nodes}):
+        raise ValueError('Affected node absent from current authorization set')
+    return {n['tag']: 'full' if impact['shared_changes'] or n['sni'] in impact['affected_snis'] else 'tls' for n in nodes}
+
+
+def audit(before, after, report_bytes, qa_bytes, binding, plan_sha, expected_qas, now=None, ttl=900, scope=None):
     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):
@@ -45,16 +68,27 @@ def audit(before, after, report_bytes, qa_bytes, binding, plan_sha, expected_qas
     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')
+    modes = {n['tag']: 'full' for n in before} if scope is None else scope
+    if set(modes) != {n['tag'] for n in before} or any(v not in ('full', 'tls') for v in modes.values()):
+        raise ValueError('Incomplete verification scope')
+    if scope is not None and binding.get('scope_sha256') != digest(modes):
+        raise ValueError('Verification scope binding changed')
     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 = {}
-    actual = set()
+    for result in report['results']:
-    for n in report['results']:
+        tag = result.get('tag')
-        if n.get('live') is not True or n.get('strict_tls') is not True:
+        if tag in actual or tag not in modes:
-            raise ValueError('Live strict TLS required for every node')
+            raise ValueError('Unexpected or duplicate verification result')
-        actual.add(tuple(n.get(k) for k in ('tag', 'relay', 'relay_port', 'sni', 'exit_ip', 'country')))
+        actual[tag] = result
-    if actual != expected:
+    for node in before:
-        raise ValueError('Node/endpoint/exit/country mismatch')
+        result = actual.get(node['tag'], {})
+        if result.get('live') is not True or result.get('strict_tls') is not True or result.get('verification', 'full') != modes[node['tag']]:
+            raise ValueError('Required verification mode or strict TLS missing')
+        if any(result.get(k) != node[k] for k in ('relay', 'relay_port', 'sni')):
+            raise ValueError('Node endpoint mismatch')
+        if modes[node['tag']] == 'full' and (result.get('exit_ip') != node['expected_exit'] or result.get('country') != node['country']):
+            raise ValueError('Node 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():

scripts/node-onboarding/configure-site.py

@@ -11,7 +11,7 @@ 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'}
+    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','acceptance_impact'}
     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')

scripts/node-onboarding/site_adapter.py

@@ -3,7 +3,7 @@ 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
+from acceptance import audit,digest,snapshot,service_change,provider_state,verification_scope,validate_impact
 
 class SiteAdapter:
     def __init__(self,plan,directory,pool=None,atomic_writer=None,resource_lock=None):
@@ -17,6 +17,7 @@ class SiteAdapter:
             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)
+        validate_impact(self.ctx.get('acceptance_impact'))
         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')
@@ -162,19 +163,21 @@ print(json.dumps({'unit_sha256':h(v['FragmentPath']),'binary_sha256':h(s['binary
                     values.append(dict(item,**self.exit_registry[item['sni']]))
                 return snapshot(values)
             before=public(source)
+            owned=self.formal_done.get(group,{}).get('owned',source.get('owned',False))
+            scope=verification_scope(before,None if owned else ctx.get('acceptance_impact'))
             # 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'])
+                return audit(prior['before'],before,prior['report'],prior['qas'],prior['binding'],digest(self.plan),prior['expected_qas'],scope=scope)
             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']})
+                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'],'verification':scope[node['tag']]})
             # 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
@@ -225,12 +228,12 @@ for attempt in range(3 if source['owned'] else 1):
                 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())}
+            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()),'scope_sha256':digest(scope)}
-            audit(before,after,report,qas,binding,digest(self.plan),identities)
+            audit(before,after,report,qas,binding,digest(self.plan),identities,scope=scope)
             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}
+            self.formal_done[group]={'before':before,'report':report,'qas':qas,'binding':binding,'expected_qas':identities,'owned':owned}
             # 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})
+            self.atomic_json(self.directory/('formal-group-'+str(group)+'.json'),{'snapshot':before,'verification_scope':scope,'report':json.loads(report),'qas':json.loads(qas),'binding':binding})
 
     def qa_evidence(self,n,raw,state):
         # Bind the exact remote bytes, not a re-serialized nested report. Persist the latest

scripts/node-onboarding/verify-node.py

@@ -14,6 +14,7 @@ import os
 from pathlib import Path
 import shutil
 import socket
+import ssl
 import subprocess
 import tempfile
 import time
@@ -53,6 +54,7 @@ def validate(config, expectations, exact=True):
         except ValueError as exc:
             raise VerificationError('Invalid expected exit IP') from exc
         require(len(spec['expected_country']) == 2 and spec['expected_country'].isalpha() and spec['expected_country'].isupper(), 'Expected country must be two uppercase letters')
+        require(spec.get('verification', 'full') in ('full', 'tls'), 'Unsupported verification mode')
         tag = spec['tag']
         require(tag in nodes and tag not in selected, 'Missing/duplicate selected node')
         node = nodes[tag]
@@ -151,6 +153,23 @@ def run_node(config, spec, binary, port, diagnostic_dir):
             stop(proc)
 
 
+def run_tls(spec):
+    """Reach the relay and validate the node certificate; no exit/traffic claim."""
+    context = ssl.create_default_context()
+    with socket.create_connection((spec['host'], spec['port']), timeout=10) as stream:
+        with context.wrap_socket(stream, server_hostname=spec['sni']) as secure:
+            protocol = secure.version()
+    require(protocol in ('TLSv1.2', 'TLSv1.3'), 'Unexpected TLS protocol')
+    return {'tag': spec['tag'], 'strict_tls': True, 'live': True,
+            'relay': spec['host'], 'relay_port': spec['port'], 'sni': spec['sni'],
+            'verification': 'tls', 'node_tls': protocol}
+
+
+def run_specs(config, specs, binary, port, folder):
+    return [run_tls(spec) if spec.get('verification', 'full') == 'tls'
+            else run_node(config, spec, binary, port, folder) for spec in specs]
+
+
 def write_report(path, report):
     destination = Path(path)
     require(destination.parent.is_dir() and destination.parent.stat().st_mode & 0o077 == 0,
@@ -212,6 +231,7 @@ def main():
             specs = [{'tag': candidates[0]['tag'], 'host': qa.get('host'), 'port': qa.get('port'),
                       'sni': qa.get('sni'), 'uuid': qa.get('uuid'), 'expected_exit': args.expected_exit,
                       'expected_country': args.expected_country}]
+        require(bool(args.expectations) or all(s.get('verification', 'full') == 'full' for s in specs), 'Isolated QA requires full verification')
         tags = validate(config, specs, exact=bool(args.expectations))
         results = []
         if args.run:
@@ -219,8 +239,7 @@ def main():
             binary = str(Path(args.binary).resolve())
             folder = Path(source).resolve().parent
             require(folder.stat().st_mode & 0o077 == 0, 'Live diagnostics directory must be private')
-            for spec in specs:
+            results = run_specs(config, specs, binary, args.port, str(folder))
-                results.append(run_node(config, spec, binary, args.port, str(folder)))
         report = {'state': 'live_verified' if args.run else 'structure_validated',
                   'verified_at': int(time.time()) if args.run else None,
                   'qa_user_id': qa_user_id, 'selected_nodes': tags, 'results': results,

tests/node-onboarding/test_runtime.py

@@ -13,7 +13,7 @@ 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 acceptance import audit,digest,snapshot,service_change,provider_state,verification_scope
 from ssh_pool import Pool,private_json,credentials
 import batch
 
@@ -81,6 +81,71 @@ class EvidenceTests(unittest.TestCase):
         self.assertEqual(service_change(before,dict(before,active=False)),'conflict')
         self.assertEqual(service_change(before,dict(before,binary_sha256=None)),'unknown')
 
+class ImpactAcceptanceTests(unittest.TestCase):
+    fixture=EvidenceTests.fixture
+    check=EvidenceTests.check
+    def scoped(self, nodes, report, qas, scope, after=None, binding_scope=None):
+        raw=json.dumps(report).encode();qa_raw=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(qa_raw).hexdigest(),verified_at=1000,scope_sha256=digest(scope if binding_scope is None else binding_scope))
+        return audit(nodes,nodes if after is None else after,raw,qa_raw,binding,'plan',{'new':9},now=1100,scope=scope)
+
+    def test_unknown_and_shared_impact_require_full_while_append_uses_tls(self):
+        nodes,report,qas=self.fixture(7)
+        review={'shared_changes':[],'affected_snis':[],'review':'task/impact-review.json'}
+        self.assertEqual(set(verification_scope(nodes).values()),{'full'})
+        self.assertEqual(set(verification_scope(nodes,review).values()),{'tls'})
+        self.assertEqual(set(verification_scope(nodes,dict(review,shared_changes=['subscription renderer'])).values()),{'full'})
+        scope=verification_scope(nodes,dict(review,affected_snis=[nodes[0]['sni']]))
+        self.assertEqual(scope['node-1'],'full');self.assertEqual(scope['node-7'],'tls')
+        for bad in [{},dict(review,review=''),dict(review,affected_snis=['unknown.invalid']),dict(review,shared_changes='none')]:
+            with self.assertRaises(ValueError):verification_scope(nodes,bad)
+
+    def test_light_results_cannot_claim_exit_and_do_not_skip_set_or_qa_checks(self):
+        nodes,report,qas=self.fixture(7)
+        scope={n['tag']:'tls' for n in nodes}
+        for result in report['results']:
+            result['verification']='tls';result.pop('exit_ip');result.pop('country')
+        self.assertEqual(self.scoped(nodes,report,qas,scope)['nodes'],7)
+        with self.assertRaises(ValueError):self.scoped(nodes,report,qas,scope,after=nodes[:-1])
+        with self.assertRaises(ValueError):self.scoped(nodes,dict(report,results=report['results'][:-1]),qas,scope)
+        with self.assertRaises(ValueError):self.scoped(nodes,report,qas,scope,binding_scope={'node-1':'tls'})
+        wrong=copy.deepcopy(qas);wrong['users']['new']['banned']=False
+        with self.assertRaises(ValueError):self.scoped(nodes,report,wrong,scope)
+        full=dict(scope,**{'node-1':'full'})
+        with self.assertRaises(ValueError):self.scoped(nodes,report,qas,full)
+        with self.assertRaises(ValueError):self.check(nodes,report,qas) # Old/default audits remain full.
+
+    def test_mixed_scope_requires_full_exit_validation_only_where_affected(self):
+        nodes,report,qas=self.fixture(2);scope={'node-1':'full','node-2':'tls'}
+        report['results'][1].update(verification='tls');report['results'][1].pop('exit_ip');report['results'][1].pop('country')
+        self.assertEqual(self.scoped(nodes,report,qas,scope)['nodes'],2)
+        report['results'][0]['exit_ip']='198.51.100.10'
+        with self.assertRaises(ValueError):self.scoped(nodes,report,qas,scope)
+
+
+class ProbeDispatchTests(unittest.TestCase):
+    def verifier(self):
+        spec=importlib.util.spec_from_file_location('scope_verifier',SCRIPTS/'verify-node.py')
+        module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module);return module
+
+    def test_mixed_probes_dispatch_without_starting_client_for_tls_nodes(self):
+        verifier=self.verifier();specs=[{'tag':'old','verification':'tls'},{'tag':'affected'}]
+        with patch.object(verifier,'run_tls',return_value={'verification':'tls'}) as light,patch.object(verifier,'run_node',return_value={'live':True}) as full:
+            self.assertEqual(len(verifier.run_specs({},specs,'binary',19080,'private')),2)
+            light.assert_called_once_with(specs[0]);full.assert_called_once_with({},specs[1],'binary',19080,'private')
+
+    def test_tls_probe_checks_hostname_and_emits_no_exit_claim(self):
+        from unittest.mock import MagicMock
+        verifier=self.verifier();context=MagicMock();stream=MagicMock()
+        context.wrap_socket.return_value.__enter__.return_value.version.return_value='TLSv1.3'
+        with patch.object(verifier.ssl,'create_default_context',return_value=context),patch.object(verifier.socket,'create_connection',return_value=stream):
+            report=verifier.run_tls({'tag':'old','host':'relay.invalid','port':1234,'sni':'node.invalid'})
+        self.assertEqual(context.wrap_socket.call_args.kwargs['server_hostname'],'node.invalid')
+        self.assertEqual(report['verification'],'tls');self.assertNotIn('exit_ip',report);self.assertNotIn('country',report)
+        with patch.object(verifier.ssl,'create_default_context',return_value=context),patch.object(verifier.socket,'create_connection',side_effect=OSError('offline')):
+            with self.assertRaises(OSError):verifier.run_tls({'host':'relay.invalid','port':1234,'sni':'node.invalid'})
+
+
 class TransportTests(unittest.TestCase):
     def test_reuse_and_every_nonwhite_host_has_jump(self):
         calls=[]
@@ -224,18 +289,19 @@ class SiteAdapterTests(unittest.TestCase):
             exported=json.loads((Path(d)/'new-evidence.json').read_text());self.assertEqual(exported['report_sha256'],hashlib.sha256(raw).hexdigest());self.assertEqual(exported['panel']['d'],34)
             self.assertEqual(remotes['panel'].files['/backups/new/qa-report.json'],raw)
 
-    def test_formal_source_snapshot_audit_and_cached_audit_call_chain(self):
+    def test_formal_source_snapshot_audit_and_cached_audit_call_chain(self, impact=None, owned=False):
         import contextlib
         import io
         from site_adapter import SiteAdapter
         adapter=object.__new__(SiteAdapter)
         ctx=dict(gateway_vps='gateway',qa_root='/qa',ops_root='/ops',subscription_url='https://sub.example.invalid',client_binary='/client',formal_socks_port=19999,node_backup_root='/backups',batch_backup_root='/batch')
+        if impact is not None:ctx['acceptance_impact']=impact
         candidate=dict(key='new',formal_group_id=7,relay_host='relay.example.invalid',relay_port=47002,tls_domain='n2.example.invalid')
         adapter.plan={'context':ctx,'nodes':[candidate]};adapter.ctx=ctx
         adapter.directory=Path('/private/tmp/formal-chain-fixture/ledger/batch')
         adapter.lock=threading.RLock();adapter.resource_lock=lambda *args:contextlib.nullcontext()
         adapter.formal_done={};adapter.exit_registry={f'n{i}.example.invalid':dict(expected_exit=f'192.0.2.{i}',country='JP') for i in [1,2]}
-        source={'user_id':1,'owned':False,'token':'fixture-only','nodes':[dict(id=i,type='anytls',tag='node-'+str(i),relay='relay.example.invalid',relay_port=47000+i,sni=f'n{i}.example.invalid',uuid='fixture-only') for i in [1,2]]}
+        source={'user_id':1,'owned':owned,'token':'fixture-only','nodes':[dict(id=i,type='anytls',tag='node-'+str(i),relay='relay.example.invalid',relay_port=47000+i,sni=f'n{i}.example.invalid',uuid='fixture-only') for i in [1,2]]}
         accounts=[]
         def account_command(group,action,apply=False,client=None):
             accounts.append(action)
@@ -243,6 +309,12 @@ class SiteAdapterTests(unittest.TestCase):
         adapter.account_command=account_command
         now=int(time.time())
         raw=json.dumps({'state':'live_verified','verified_at':now,'results':[dict(tag='node-'+str(i),relay='relay.example.invalid',relay_port=47000+i,sni=f'n{i}.example.invalid',exit_ip=f'192.0.2.{i}',country='JP',live=True,strict_tls=True) for i in [1,2]]}).encode()
+        expected_scope=verification_scope([dict(x,**adapter.exit_registry[x['sni']]) for x in source['nodes']],None if owned else impact)
+        payload=json.loads(raw)
+        for result in payload['results']:
+            if expected_scope[result['tag']]=='tls':
+                result['verification']='tls';result.pop('exit_ip');result.pop('country')
+        raw=json.dumps(payload).encode()
         files={};persisted=[];programs=[]
         def execute(client,command,timeout=120):
             if command.startswith("python3 - <<'DPLFORMAL'\n"):
@@ -269,6 +341,9 @@ class SiteAdapterTests(unittest.TestCase):
         adapter.formal(candidate,None,gateway,php,execute,lambda client,path:files.get(path),write)
         self.assertEqual(len(programs),1)
         self.assertEqual(len(persisted),1)
+        self.assertEqual(persisted[0]['verification_scope'],expected_scope)
+        private=json.loads(files['/qa/formal-group-7/source-private.json'])
+        self.assertEqual({n['tag']:n['verification'] for n in private['expectations']},expected_scope)
         self.assertEqual(len(persisted[0]['snapshot']),2,'The old node must remain in complete acceptance')
         self.assertTrue(all(n['type']=='anytls' for n in persisted[0]['snapshot']))
         self.assertEqual(persisted[0]['qas']['users']['new']['http_status'],403)
@@ -276,6 +351,12 @@ class SiteAdapterTests(unittest.TestCase):
         self.assertEqual(adapter.formal(candidate,None,gateway,php,execute,lambda client,path:files.get(path),write)['nodes'],2)
         self.assertEqual(len(programs),1,'A cached audit rechecks topology without rerunning the client')
 
+    def test_reviewed_append_formal_chain_uses_light_checks_and_preserves_all_nodes(self):
+        self.test_formal_source_snapshot_audit_and_cached_audit_call_chain({'shared_changes':[],'affected_snis':[],'review':'task/review'})
+
+    def test_owned_formal_account_forces_full_tests_even_for_reviewed_append(self):
+        self.test_formal_source_snapshot_audit_and_cached_audit_call_chain({'shared_changes':[],'affected_snis':[],'review':'task/review'},owned=True)
+
     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)}