/**
 * ProjectEngine 的渐进导入/深复制集成测试，串联本地文档、history、ledger、源文件、校验及恢复。
 * 已发布内容可早于服务端 ACK；分块读取证明支持先显示，完整 SHA 与其持久化 seal 决定何时解除相关网络写入门槛。
 * 暂停/关闭/取消、端点换代、旧解释器、资源引用及源文件回收各有独立生命周期，不能用重新插入全部内容来恢复。
 * fixture 使用内存存储和幂等假服务端；测试验证引擎协调，不等同于真实 IDB 事务、网络或渲染帧率验证。
 */
import {encodeCopySnapshot} from '../../src/lib/canvas-runtime/copy-snapshot.ts';
import {importSourceDigest} from '../../src/lib/canvas-runtime/import-integrity.ts';
import {test} from 'node:test';
import assert from 'node:assert/strict';
import {createDocument,createRegistry,builtinComponents,applyChanges} from '../../src/lib/canvas-core/index.ts';
import {openProjectEngine} from '../../src/lib/canvas-runtime/engine.ts';
import {decodeRecovery} from '../../src/lib/canvas-runtime/recovery.ts';
// 内置类型、空单 canvas 基线及固定作用域，关闭/重开读取同一个项目的恢复 image。
const registry=createRegistry(builtinComponents()),original=createDocument('project',[{id:'canvas',parent:null}],[]),scope={accountId:'account',workspaceId:'workspace',projectId:'project'};
// native NDJSON 记录工厂故意携带第八代身份和 runId；导入应生成新身份并去掉执行任务字段。
const node=id=>({kind:'element',key:id,record:{id,incarnation:8,structure:'node',ownerCanvasId:'$target',componentId:'text',dataVersion:1,data:{title:id,text:'',runId:'must not run'},geometry:{x:0,y:0,width:100,height:100,rotation:0}}});
// 对象序列编码为 NDJSON，也允许传入原始坏行以检查部分失败。
const file=records=>new Blob([records.map(value=>typeof value==='string'?value:JSON.stringify(value)).join('\n')]);
// 手动 gate 与 entered 信号控制服务端 ACK、摘要、源读取或持久化，避免靠固定延时决定事件顺序。
const deferred=()=>{let resolve;const promise=new Promise(yes=>resolve=yes);return{promise,resolve};};
// 准备可重用的源文件/image/服务端文档和回执；broken 在写入前抛配额错误，gate 可挂起服务端提交。
function fixture(initial=original){
 let image=null,leader,next=0,server=initial;const versions=new Map([...initial.canvases.keys()].map(id=>[id,1]));const files=new Map(),requests=[],receipts=new Map(),errors=[];
 const f={files,requests,errors,gate:null,broken:false,
 storage:{fileDeletion:'atomic-v1',
  // 单项目仅一个 leader；release 清空 owner 并撤销活动标记，允许关闭后重新领取。
  async claim(){assert.equal(leader,undefined);const value={scope,fence:1,active:true,async release(){leader=undefined;value.active=false;}};leader=value;return value;},
  // 克隆上次 image；用例重开应通过恢复 codec 读取它，而非复用旧引擎状态。
  async recover(){return image&&structuredClone(image);},
  // 先核对 leader 并检查配额，再保存源文件/删除/克隆 image；此内存实现不模拟真实事务的部分失败回滚。
  async commit(owner,batch){assert.equal(owner,leader);if(f.broken)throw Error('quota');for(const part of batch.files??[])files.set(part.id,part.bytes);for(const id of batch.deleteFiles??[])files.delete(id);image=structuredClone(batch.image);return{durableRootRevision:batch.image.rootRevision};},
  // 按 ID 返回持久化 Blob，且要求当前 leader；用例可替换为挂起读取、损坏字节或 unavailable。
  async readFile(owner,id){assert.equal(owner,leader);return files.get(id);},
  // 内存端口无外部句柄；引擎负责归还 leader。
  close(){}},
 port:{
  // 编译携带意图和 requestId 的候选请求，参与 canvas 使用当前 ACK 版本，未建立的新 canvas 基线为零。
  compile(intent,requestId,snapshot){return{payload:{requestId,intent},canvasIds:intent.canvasIds,expectedVersions:Object.fromEntries(intent.canvasIds.map(id=>[id,snapshot.runtime.versions.get(id)??0]))};},
  // 记录请求并可卡在 gate；同 requestId 已有 receipt 时不再次应用，模拟后端幂等，canonical 使用该次保存的基线。
  async submit(request){requests.push(request);if(f.gate)await f.gate.promise;const{intent}=JSON.parse(request.body);if(!receipts.has(request.id)){server=applyChanges(server,intent.changes);for(const canvasId of intent.canvasIds)versions.set(canvasId,(versions.get(canvasId)??0)+1);receipts.set(request.id,{document:server,versions:new Map(versions)});}return{outcome:'applied',canonical:{requestId:request.id}};},
  // 按请求 ID 查询执行事实；无保存回执时保持 unresolved，不推定零写入。
  async query(request){return receipts.has(request.id)?{outcome:'applied',canonical:{requestId:request.id}}:{outcome:'unresolved'};},
  // 取对应请求保存的 document/versions，防止用当前 working 冒充服务端确认。
  canonical(request){return receipts.get(request.id);},
  // 恢复会话读取当前服务端文档及版本表副本，避免外部修改同一个 Map。
  async read(){return{document:server,versions:new Map(versions)};}},

  // 暴露 fixture 保存的 image 引用，专供旧格式改造和损坏注入；不是产品公开状态接口。
  image:()=>image,
  // 只读取得当前假服务端文档，供断言本地可见与远端 ACK 之间的差别。
  server:()=>server,
  // 每次以相同源存储/作用域重新装配引擎；递增 ID 分配器和错误列表跨重开保留。
  open:(extra={})=>openProjectEngine({scope,registry,document:initial,versions:new Map([...initial.canvases.keys()].map(id=>[id,1])),storage:f.storage,persistence:f.port,createId:()=>`identity-${++next}`,onError:error=>errors.push(error),...extra})};return f;
}
// 先检查当前快照，再订阅到谓词成立或 5 秒超时；两种结束路径都移除订阅，避免泄漏监听器。
function until(engine,predicate){if(predicate(engine.getSnapshot()))return Promise.resolve();return new Promise((resolve,reject)=>{const timer=setTimeout(()=>{unsubscribe();reject(Error('import condition timed out'));},5000),unsubscribe=engine.editor.subscribe((_,snapshot)=>{if(predicate(snapshot)){clearTimeout(timer);unsubscribe();resolve();}});});}

// 阻塞后端提交时三个 native 元素已本地可见且历史合并为一组；新 incarnation=1，runId 被剔除，放行 ACK 后再检查服务端。
test('native import adds visible records while its backend writer is still awaiting ACK',async()=>{
 const f=fixture(),engine=await f.open(),gate=deferred();f.gate=gate;
 try{const id=engine.imports.startNdjson(file([node('one'),node('two'),node('three')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).status==='completed');assert.equal(engine.getSnapshot().document.elements.size,3);assert.equal(f.server().elements.size,0);assert.equal(engine.getSnapshot().history.undo.length,1);assert.equal(f.files.size,1);for(const element of engine.getSnapshot().document.elements.values()){assert.equal(element.incarnation,1);assert.equal(element.data.runId,undefined);}gate.resolve();await engine.settle();assert.equal(f.server().elements.size,3);assert.equal(f.errors.length,0);decodeRecovery(f.image(),registry);}finally{gate.resolve();await engine.close();}
});

// 首项后暂停、编辑已导入元素再重开，继续使用持久化源与原 ID，后续导入不能覆盖手动标题。
test('pause/reopen/resume reuses the durable source and IDs without overwriting a user edit',async()=>{
 const f=fixture();let engine=await f.open();
 try{const id=engine.imports.startNdjson(file([node('one'),node('two'),node('three')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const job=engine.imports.getJobs().get(id),first=job.stableIds['element:one'];assert.equal(job.status,'paused');engine.editor.dispatch({type:'element.data',target:{id:first,incarnation:1},patch:{title:'user edit'}});await engine.settle();await engine.close();engine=await f.open();assert.equal(engine.imports.getJobs().get(id).status,'paused');await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.imports.getJobs().get(id).stableIds['element:one'],first);assert.equal(engine.getSnapshot().document.elements.get(first).data.title,'user edit');assert.equal(engine.getSnapshot().document.elements.size,3);assert.equal(f.errors.length,0);decodeRecovery(f.image(),registry);}finally{await engine.close();}
});

// 导入前缀、手动编辑、导入后缀是三个有序 undo 单元；重开后逐项 undo/redo 仍保留同一编辑与任务完成事实。
test('import history on either side of a manual edit remains three ordered undo units after reopening',async()=>{
 const f=fixture();let engine=await f.open();
 try{
  const id=engine.imports.startNdjson(file([node('one'),node('two'),node('three')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();
  const first=engine.imports.getJobs().get(id).stableIds['element:one'],before=engine.getSnapshot().document.elements.get(first).data.title;assert.equal(engine.editor.dispatch({type:'element.data',target:{id:first,incarnation:1},patch:{title:'manual between import batches'}}).status,'applied');await engine.settle();await engine.imports.resume(id);await engine.settle();
  assert.equal(engine.getSnapshot().history.undo.length,3);assert.equal(engine.getSnapshot().document.elements.size,3);await engine.close();engine=await f.open();assert.equal(engine.getSnapshot().history.undo.length,3);
  assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(engine.getSnapshot().document.elements.get(first).data.title,'manual between import batches');
  assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.get(first).data.title,before);
  assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.size,0);
  for(let index=0;index<3;index++){assert.equal(engine.editor.history.redo().status,'applied');await engine.settle();}
  assert.equal(engine.getSnapshot().document.elements.size,3);assert.equal(engine.getSnapshot().document.elements.get(first).data.title,'manual between import batches');assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(f.server().elements.size,3);assert.equal(f.errors.length,0);
 }finally{await engine.close();}
});

// 取消只保留已加入首项；重开后显式 resume 也不把 cancelled 任务当暂停任务继续插入。
test('cancel keeps the already added prefix and survives reopening without further insertion',async()=>{
 const f=fixture();let engine=await f.open();
 try{const id=engine.imports.startNdjson(file([node('one'),node('two'),node('three')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.cancel(id);await engine.settle();assert.equal(engine.getSnapshot().document.elements.size,1);await engine.close();engine=await f.open();await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'cancelled');assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.server().elements.size,1);}finally{await engine.close();}
});

// 首个端点导入后被用户删除，晚到连线不能通过恢复旧源把它复活；剩余独立节点导入，任务 partial。
test('deleting an imported endpoint while paused prevents a late edge from reviving it',async()=>{
 const f=fixture();let engine=await f.open();
 const edge={kind:'element',key:'edge',record:{id:'edge',incarnation:1,structure:'connection',componentId:'connection',dataVersion:1,ownerCanvasId:'$target',source:{elementId:'one'},target:{elementId:'two'},path:null,data:{}}};
 try{const id=engine.imports.startNdjson(file([node('one'),edge,node('two')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const first=engine.imports.getJobs().get(id).stableIds['element:one'];engine.editor.dispatch({type:'element.remove',target:{id:first,incarnation:1}});await engine.settle();await engine.close();engine=await f.open();await engine.imports.resume(id);await engine.settle();const job=engine.imports.getJobs().get(id);assert.equal(job.status,'partial');assert.equal(engine.getSnapshot().document.elements.has(first),false);assert.equal(engine.getSnapshot().document.elements.size,1);assert.ok(job.failures.some(failure=>failure.key==='element:edge'));assert.equal(f.server().elements.has(first),false);}finally{await engine.close();}
});

// prepareRecovery 暂停活动导入发布，等待期间 rootRevision 不变；接纳新基线后才继续完成导入。
test('canonical recovery holds active imports and releases them after accepting the fresh base',async()=>{
 const f=fixture(),engine=await f.open();
 try{const id=engine.imports.startNdjson(file(Array.from({length:20},(_,index)=>node(`n-${index}`))),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length>=1);const session=await engine.prepareRecovery(),cut=engine.getSnapshot().rootRevision;await new Promise(resolve=>setTimeout(resolve,10));assert.equal(engine.getSnapshot().rootRevision,cut);assert.equal((await session.apply()).status,'applied');await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.getSnapshot().document.elements.size,20);assert.equal(f.errors.length,0);}finally{await engine.close();}
});


// 关闭活动导入保存可继续的 running 检查点，重开自动续跑，首项 ID 不变且无重复节点。
test('closing an active import preserves a running checkpoint and reopening resumes without duplicate records',async()=>{
 const f=fixture();let engine=await f.open();
 try{const id=engine.imports.startNdjson(file([node('one'),node('two'),node('three')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);const first=engine.imports.getJobs().get(id).stableIds['element:one'];await engine.close();engine=await f.open();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.imports.getJobs().get(id).stableIds['element:one'],first);assert.equal(engine.getSnapshot().document.elements.size,3);assert.equal(f.errors.length,0);}finally{await engine.close();}
});

// 端点删除后 undo 恢复成第二代，旧连线仍不得绑定新身份；记录 import_dependency_changed，保留当前元素。
test('a late connection cannot claim an endpoint restored into a new incarnation',async()=>{
 const f=fixture(),engine=await f.open(),edge={kind:'element',key:'edge',record:{id:'edge',incarnation:1,structure:'connection',componentId:'connection',dataVersion:1,ownerCanvasId:'$target',source:{elementId:'one'},target:{elementId:'two'},path:null,data:{}}};
 try{const id=engine.imports.startNdjson(file([node('one'),edge,node('two')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const first=engine.imports.getJobs().get(id).stableIds['element:one'];engine.editor.dispatch({type:'element.remove',target:{id:first,incarnation:1}});await engine.settle();assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.get(first).incarnation,2);await engine.imports.resume(id);await engine.settle();const job=engine.imports.getJobs().get(id);assert.equal(job.status,'partial');assert.equal(engine.getSnapshot().document.elements.get(first).incarnation,2);assert.equal(engine.getSnapshot().document.elements.has(job.stableIds['element:edge']),false);assert.ok(job.failures.some(failure=>failure.code==='import_dependency_changed'));}finally{await engine.close();}
});


// 重开读取源仍挂起时取消 job，随后返回 source unavailable 不能改写 cancelled 或产生无关错误。
test('a late unavailable-source result cannot change a job cancelled during reopening',async()=>{
 const f=fixture();let engine=await f.open();const gate=deferred(),entered=deferred();
 try{const id=engine.imports.startNdjson(file([node('one'),node('two')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await until(engine,snapshot=>!!snapshot.runtime.imports.get(id).source.sha256);await engine.persist();await engine.close();f.storage.readFile=async()=>{entered.resolve();return gate.promise;};engine=await f.open();await entered.promise;await engine.imports.cancel(id);gate.resolve(undefined);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'cancelled');assert.equal(f.errors.length,0);}finally{gate.resolve(undefined);await engine.close();}
});


// 普通空恢复 image 可为格式 1，含读取证明后升级到 17；不能仅改 formatVersion 就把证明数据降成旧格式。
test('proof-bearing recovery cuts use format 17 and cannot be silently downgraded to a legacy image',async()=>{
 const f=fixture(),engine=await f.open();
 try{await engine.persist();assert.equal(f.image().formatVersion,1);const id=engine.imports.startNdjson(file([node('one')]),'canvas');await engine.settle();const image=f.image();assert.equal(image.formatVersion,17);assert.equal(decodeRecovery(image,registry).runtime.imports.get(id).status,'completed');assert.throws(()=>decodeRecovery({...image,formatVersion:1},registry),/unsupported_recovery_version/);}finally{await engine.close();}
});


// 源文件首次保存挂起时首项已可见，但文件未落盘且任何后端请求不得越过屏障；放行后才完成同步。
test('the first valid import record displays while the source write is pending, but no backend request crosses that barrier',async()=>{
 const f=fixture(),engine=await f.open(),gate=deferred(),entered=deferred(),commit=f.storage.commit;f.storage.commit=async(owner,batch)=>{if(batch.files?.length){entered.resolve();await gate.promise;}return commit(owner,batch);};
 try{const id=engine.imports.startNdjson(file([node('one'),node('two')]),'canvas');await entered.promise;await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length>=1);assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.requests.length,0);assert.equal(f.files.size,0);gate.resolve();await engine.settle();assert.equal(f.files.size,1);assert.equal(f.server().elements.size,2);assert.equal(engine.imports.getJobs().get(id).status,'completed');}finally{gate.resolve();await engine.close();}
});


// 源文件写失败使任务暂停并保留本地首项/ID，网络未发；修复存储后 resume 唤醒原可靠写入流程。
test('resuming after a failed source write preserves IDs and wakes the original durable writer',async()=>{
 const f=fixture(),engine=await f.open();
 try{await engine.persist();f.broken=true;const id=engine.imports.startNdjson(file([node('one'),node('two')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).status==='paused');const first=engine.imports.getJobs().get(id).stableIds['element:one'];assert.equal(f.requests.length,0);assert.equal(engine.getSnapshot().document.elements.size,1);f.broken=false;await engine.imports.resume(id);await engine.settle();await until(engine,()=>f.server().elements.size===2);assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.imports.getJobs().get(id).stableIds['element:one'],first);assert.equal(f.files.size,1);assert.equal(new Set(f.requests.map(request=>request.id)).size,f.requests.length);}finally{f.broken=false;await engine.close();}
});

// 深复制样本按组件添加共享图片资源和付费任务字段，区分可复用的资源与不能延续的执行状态。
const copyNode=(id,owner='canvas',componentId='text')=>({id,incarnation:1,structure:'node',ownerCanvasId:owner,componentId,dataVersion:1,data:{title:id,text:'source',...(componentId==='image'?{assetId:'shared',src:'asset:shared',runId:'paid-run',taskId:'task'}:{})},geometry:{x:0,y:0,width:200,height:100,rotation:0}});
// 用一个事务建立上级 C、入口 F、child 成员 A/B 及 C→F 入边，等待同步后作为复制基线。
async function seedCopy(engine){
 const records=[copyNode('C'),copyNode('F','canvas','frame'),copyNode('A','child','image'),copyNode('B','child')];
 const edge={id:'incoming',incarnation:1,structure:'connection',ownerCanvasId:'canvas',componentId:'connection',dataVersion:1,source:{elementId:'C'},target:{elementId:'F'},path:null,data:{title:''}};
 const result=engine.editor.transaction(tx=>{tx.dispatch({type:'element.create',element:records[0]});tx.dispatch({type:'element.create',element:records[1]});tx.dispatch({type:'canvas.create',canvas:{id:'child',parent:{canvasId:'canvas',entryElementId:'F'}}});for(const element of records.slice(2))tx.dispatch({type:'element.create',element});tx.dispatch({type:'element.create',element:edge});},{label:'Seed copy source'});assert.equal(result.status,'applied');await engine.settle();
}

// 深复制暂停/重开后使用同一源及映射，保留副本上用户编辑、上级入边、共享资源及剔除任务字段的结果。
test('deep copy source/mapping recover and preserve upper incoming source, shared assets and user edits',async()=>{
 const f=fixture();let engine=await f.open();
 try{
  await seedCopy(engine);const id=await engine.imports.duplicate([{id:'F',incarnation:1},{id:'A',incarnation:1}]);
  await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();
  const before=engine.imports.getJobs().get(id),copied=before.stableIds['element:F'];assert.equal(before.source.kind,'duplicate-plan');assert.equal(before.source.copy.roots.length,1);assert.equal(before.status,'paused');
  engine.editor.dispatch({type:'element.data',target:{id:copied,incarnation:1},patch:{title:'edited copy'}});await engine.settle();await engine.close();engine=await f.open();await engine.imports.resume(id);await engine.settle();
  const job=engine.imports.getJobs().get(id),document=engine.getSnapshot().document,ids=job.stableIds;assert.equal(job.status,'completed');assert.deepEqual(ids,before.stableIds);assert.equal(document.elements.get(copied).data.title,'edited copy');
  assert.equal(document.canvases.get(ids['canvas:child']).parent.entryElementId,copied);assert.equal(document.elements.get(ids['element:A']).ownerCanvasId,ids['canvas:child']);assert.equal(document.elements.get(ids['element:A']).data.assetId,'shared');assert.equal(document.elements.get(ids['element:A']).data.runId,undefined);assert.equal(document.elements.get(ids['element:A']).data.taskId,undefined);
  const edge=document.elements.get(ids['element:incoming']);assert.equal(edge.source.elementId,'C');assert.equal(edge.target.elementId,copied);assert.equal(edge.ownerCanvasId,'canvas');assert.ok(document.elements.has('incoming'));assert.equal(job.failures.length,0);assert.equal(f.errors.length,0);assert.ok(f.server().elements.has(ids['element:incoming']));decodeRecovery(f.image(),registry);
 }finally{await engine.close();}
});

// 外部上级 C 删除再恢复成新一代，晚到复制入边报 copy_external_changed；独立子节点副本仍保留。
test('late copied edge cannot attach to an external endpoint deleted and restored as a new incarnation',async()=>{
 const f=fixture(),engine=await f.open();try{await seedCopy(engine);const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();
  assert.equal(engine.editor.dispatch({type:'element.remove',target:{id:'C',incarnation:1}}).status,'applied');assert.equal(engine.editor.history.undo().status,'applied');assert.equal(engine.getSnapshot().document.elements.get('C').incarnation,2);await engine.settle();await engine.imports.resume(id);await engine.settle();
  const job=engine.imports.getJobs().get(id);assert.equal(job.status,'partial');assert.ok(job.failures.some(failure=>failure.code==='copy_external_changed'));assert.equal(engine.getSnapshot().document.elements.has(job.stableIds['element:incoming']),false);assert.ok(engine.getSnapshot().document.elements.has(job.stableIds['element:A']));
 }finally{await engine.close();}
});

// 复制任务取消后仅保留已发布前缀，重开/resume 不再新增副本。
test('copy cancellation retains the published prefix and does not resume on reopening',async()=>{
 const f=fixture();let engine=await f.open();try{await seedCopy(engine);const initial=engine.getSnapshot().document.elements.size,id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.cancel(id);await engine.settle();await engine.close();engine=await f.open();await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'cancelled');assert.equal(engine.getSnapshot().document.elements.size,initial+1);}finally{await engine.close();}
});

// 尚未准入前的预取消或项目关闭，迟到复制准备结果不能创建 running job；重开后任务表仍为空。
test('preparation abort and project close do not admit a late copy job',async()=>{
 const f=fixture(),engine=await f.open();await seedCopy(engine);const controller=new AbortController();controller.abort();await assert.rejects(engine.imports.duplicate([{id:'F',incarnation:1}],{signal:controller.signal}),/copy_aborted/);assert.equal(engine.imports.getJobs().size,0);
 const pending=engine.imports.duplicate([{id:'F',incarnation:1}]);const rejected=assert.rejects(pending,/copy_aborted|engine_closing/);await engine.close();await rejected;const reopened=await f.open();try{assert.equal(reopened.imports.getJobs().size,0);}finally{await reopened.close();}
});

// 复制本地完成不等待 ACK；任务显示关联发起 viewCanvasId，而复制元素仍归源规则决定的 child，不能混淆二者。
test('copy publishes independently of backend ACK and associates task visibility with its initiating view',async()=>{
 const f=fixture(),engine=await f.open(),gate=deferred();try{await seedCopy(engine);f.gate=gate;const before=f.server().elements.size,id=await engine.imports.duplicate([{id:'A',incarnation:1}],{viewCanvasId:'canvas'});await until(engine,snapshot=>snapshot.runtime.imports.get(id).status==='completed');const job=engine.imports.getJobs().get(id);assert.equal(job.source.targetCanvasId,'canvas');assert.equal(engine.getSnapshot().document.elements.get(job.stableIds['element:A']).ownerCanvasId,'child');assert.equal(f.server().elements.size,before);gate.resolve();await engine.settle();assert.equal(f.server().elements.size,before+1);assert.equal(f.errors.length,0);}finally{gate.resolve();await engine.close();}
});

// 复制源保存失败时网络无新请求；恢复后保留准备期 source 与稳定身份计划，最终映射数量符合深复制范围。
test('copy source write failure gates network and resume preserves the planned identities',async()=>{
 const f=fixture(),engine=await f.open();try{await seedCopy(engine);const sent=f.requests.length;f.broken=true;const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>snapshot.runtime.imports.get(id).status==='paused');const before=engine.imports.getJobs().get(id);assert.equal(f.requests.length,sent);f.broken=false;await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(before.source.kind,'duplicate-snapshot');assert.equal(before.preparation.allocationIds.length,0);assert.equal(Object.keys(engine.imports.getJobs().get(id).stableIds).length,5);}finally{f.broken=false;await engine.close();}
});


// 只读策略先拒绝 import/copy 准入，即使选择目标不存在也不先读取/存储源或创建任务。
test('read-only policy rejects copy/import admission without storing a source or a running job',async()=>{
 const f=fixture(),engine=await f.open({policy:()=>({enabled:false,code:'read_only',targets:[]})});try{await assert.rejects(engine.imports.duplicate([{id:'missing',incarnation:1}]),/read_only/);assert.throws(()=>engine.imports.startNdjson(file([node('one')]),'canvas'),/read_only/);assert.equal(engine.imports.getJobs().size,0);await engine.settle();assert.equal(f.files.size,0);assert.equal(f.requests.length,0);}finally{await engine.close();}
});


// 大复制在已分配 ID 的准备期暂停并重开，复用 planner 游标而不重扫前阶段；源后来改名不影响捕获快照中的副本标题。
test('preparing copy resumes saved planning cursors and allocated identities without repeating earlier phases',async(t)=>{
 const f=fixture(),save=f.storage.commit;let durableCuts=0;f.storage.commit=async(...args)=>{durableCuts++;return save(...args);};let engine=await f.open();try{
  await seedCopy(engine);assert.equal(engine.editor.transaction(tx=>{for(let i=0;i<1000;i++)tx.dispatch({type:'element.create',element:copyNode(`extra-${i}`,'child')});}).status,'applied');await engine.settle();
  const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>(snapshot.runtime.imports.get(id).preparation?.allocationIds.length??0)>0);await engine.imports.pause(id);await engine.settle();const checkpoint=engine.imports.getJobs().get(id);assert.equal(checkpoint.status,'paused');assert.equal(checkpoint.source.kind,'duplicate-snapshot');assert.ok(checkpoint.preparation.allocationIds.length>0);assert.ok(checkpoint.preparation.steps>0);assert.equal(checkpoint.preparation.version,4);assert.ok(checkpoint.preparation.planner);assert.equal(checkpoint.preparation.planner.sourceToken,checkpoint.source.sha256);assert.equal(checkpoint.committedSourceKeys.length,0);decodeRecovery(f.image(),registry);
  engine.editor.dispatch({type:'element.data',target:{id:'F',incarnation:1},patch:{title:'new source title'}});await engine.settle();await engine.close();engine=await f.open();assert.equal(engine.imports.getJobs().get(id).status,'paused');assert.deepEqual(engine.imports.getJobs().get(id).preparation.allocationIds,checkpoint.preparation.allocationIds);assert.deepEqual(engine.imports.getJobs().get(id).preparation.planner,checkpoint.preparation.planner);const observed=[],stop=engine.editor.subscribe((_,snapshot)=>{const state=snapshot.runtime.imports.get(id)?.preparation?.planner;if(state)observed.push({phase:state.phase,steps:state.steps});});await engine.imports.resume(id);await engine.settle();stop();assert.ok(observed.some(state=>state.steps>checkpoint.preparation.steps));assert.ok(observed.every(state=>state.steps>=checkpoint.preparation.steps&&!['selection','canvases','members','roots','seed','dfs','edges'].includes(state.phase)));await engine.settle();
  const job=engine.imports.getJobs().get(id);assert.equal(job.status,'completed');assert.deepEqual(Object.values(job.stableIds).slice(0,checkpoint.preparation.allocationIds.length),checkpoint.preparation.allocationIds);assert.equal(engine.getSnapshot().document.elements.get(job.stableIds['element:F']).data.title,'F');assert.equal(engine.getSnapshot().document.elements.get('F').data.title,'new source title');assert.equal(job.preparation,undefined);assert.equal(f.errors.length,0);t.diagnostic(JSON.stringify({imported:job.committedSourceKeys.length,durableCuts,requests:f.requests.length}));
 }finally{await engine.close();}
});

// 任务已准入但仍 preparing 时关闭，保存任务与快照，重开自动完成五个源键。
test('closing an admitted preparing copy preserves its task and snapshot for automatic restart',async()=>{
 const f=fixture();let engine=await f.open();try{await seedCopy(engine);const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);assert.equal(engine.imports.getJobs().get(id).status,'preparing');await engine.close();engine=await f.open();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.imports.getJobs().get(id).committedSourceKeys.length,5);assert.equal(f.errors.length,0);}finally{await engine.close();}
});

// 准备期取消持久化后重开仍 cancelled，原文档数量不变，不发布任何迟到副本。
test('cancel during preparation is durable and never publishes a copy on reopening',async()=>{
 const f=fixture();let engine=await f.open();try{await seedCopy(engine);const size=engine.getSnapshot().document.elements.size,id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await engine.imports.cancel(id);await engine.settle();await engine.close();engine=await f.open();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'cancelled');assert.equal(engine.getSnapshot().document.elements.size,size);assert.equal(f.errors.length,0);}finally{await engine.close();}
});


// 准备时使用的 registry.copy 声明改变，重开后明确暂停为 copy_interpreter_changed，不用新规则静默解释旧快照。
test('a reopened preparation rejects changed registry copy declarations instead of silently reinterpreting the source',async()=>{
 const f=fixture();let engine=await f.open();try{await seedCopy(engine);const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await engine.imports.pause(id);await engine.close();
  const changed=createRegistry(builtinComponents().map(component=>component.id==='image'?{...component,copy:{...component.copy,omit:[...component.copy.omit,['customExecution']]}}:component));engine=await f.open({registry:changed});await engine.imports.resume(id);await engine.settle();const job=engine.imports.getJobs().get(id);assert.equal(job.status,'paused');assert.equal(job.error,'copy_interpreter_changed');assert.equal(job.committedSourceKeys.length,0);
 }finally{await engine.close();}
});

// 恢复会话 hold 期间 preparing 任务不前进，接纳协调后的状态根后才重新启动并完成。
test('canonical recovery holds a preparing copy and restarts it only after the coordinated root is adopted',async()=>{
 const f=fixture(),engine=await f.open();try{await seedCopy(engine);const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);const recovery=await engine.prepareRecovery(),cut=engine.getSnapshot().rootRevision;await new Promise(resolve=>setTimeout(resolve,10));assert.equal(engine.getSnapshot().rootRevision,cut);assert.equal(engine.imports.getJobs().get(id).status,'preparing');assert.equal((await recovery.apply()).status,'applied');await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(f.errors.length,0);}finally{await engine.close();}
});

// frame 端口引用成员、成员归属 child、child 绑定 frame 形成依赖环，必须一次闭合发布；深复制后端口局部 ID/handle 保留、nodeId 映射，重开与 undo/redo 都可恢复。
test('native frame ports and deep copy close ownership/reference cycles atomically, keep local handles and recover every binding',async()=>{
 const f=fixture();let engine=await f.open();
 const frame={...node('F'),record:{...node('F').record,componentId:'frame',data:{title:'Frame',text:'',ports:[{id:'input',direction:'input',nodeId:'A',handle:'left'}]}}},child={...node('A'),record:{...node('A').record,ownerCanvasId:'S'}},canvas={kind:'canvas',key:'S',record:{id:'S',parent:{canvasId:'$target',entryElementId:'F'}}};
 try{
  // 观察每次发布的 frame 端口是否已经有目标成员，用 0/2 元素数量检测依赖环的半成品是否泄漏。
  const cuts=[],unsubscribe=engine.editor.subscribe((_,snapshot)=>{const rows=[...snapshot.document.elements.values()],frames=rows.filter(row=>row.componentId==='frame');for(const row of frames){const port=row.data.ports[0];assert.ok(snapshot.document.elements.has(port.nodeId),'published a frame without its port target');assert.notEqual(port.nodeId,'A');}cuts.push(rows.length);});
  const imported=engine.imports.startNdjson(file([frame,canvas,child]),'canvas');await engine.settle();const first=engine.imports.getJobs().get(imported);assert.equal(first.status,'completed');assert.equal(first.committedSourceKeys.length,3);const F=first.stableIds['element:F'],A=first.stableIds['element:A'];assert.equal(engine.getSnapshot().document.elements.get(F).data.ports[0].nodeId,A);assert.ok(cuts.every(size=>size===0||size===2));unsubscribe();
  const copied=await engine.imports.duplicate([{id:F,incarnation:1}]);await engine.settle();const job=engine.imports.getJobs().get(copied);assert.equal(job.status,'completed');assert.equal(job.committedSourceKeys.length,3);const F2=job.stableIds[`element:${F}`],A2=job.stableIds[`element:${A}`],port=engine.getSnapshot().document.elements.get(F2).data.ports[0];assert.deepEqual(port,{id:'input',direction:'input',nodeId:A2,handle:'left'});assert.notEqual(A2,A);assert.equal(f.server().elements.size,4);
  await engine.close();engine=await f.open();assert.equal(engine.imports.getJobs().get(copied).status,'completed');assert.equal(engine.getSnapshot().document.elements.get(F2).data.ports[0].nodeId,A2);assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.has(F2),false);assert.ok(engine.getSnapshot().document.elements.has(F));assert.equal(engine.editor.history.redo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.get(F2).data.ports[0].nodeId,A2);assert.equal(f.errors.length,0);decodeRecovery(f.image(),registry);
 }finally{await engine.close();}
});

// 源文件保存一直挂起时允许有限本地前缀，未持久化根差达到背压门槛后停止增长；所有网络请求仍为零，放行才完成。
test('blocked source persistence admits several local batches then applies bounded backpressure and gates every request',async()=>{
 const f=fixture(),engine=await f.open(),gate=deferred(),entered=deferred(),commit=f.storage.commit;
 try{
  await engine.persist();f.storage.commit=async(owner,batch)=>{if(batch.files?.length){entered.resolve();await gate.promise;}return commit(owner,batch);};
  const id=engine.imports.startNdjson(file(Array.from({length:500},(_,index)=>node(`pressure-${index}`))),'canvas');await entered.promise;await until(engine,snapshot=>snapshot.rootRevision-engine.getDurability().durableRootRevision>=32);const cut=engine.getSnapshot(),count=cut.runtime.imports.get(id).committedSourceKeys.length;assert.ok(count>1&&count<500,`unexpected speculative prefix ${count}`);assert.equal(cut.runtime.imports.get(id).status,'running');assert.equal(f.requests.length,0);assert.equal(f.files.size,0);await new Promise(resolve=>setTimeout(resolve,20));assert.equal(engine.imports.getJobs().get(id).committedSourceKeys.length,count,'import ignored persistence backpressure');
  gate.resolve();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(f.server().elements.size,500);assert.equal(f.errors.length,0);assert.equal(f.files.size,1);decodeRecovery(f.image(),registry);
 }finally{gate.resolve();await engine.close();}
});

// 复制快照字节同长度篡改也会被摘要绑定识别；恢复前无新副本/网络写，原摘要不被替换成篡改源的身份。
test('same-length changes to an admitted copy snapshot are rejected before preparation allocates or publishes content',async()=>{
 const f=fixture();let engine=await f.open();try{await seedCopy(engine);const size=engine.getSnapshot().document.elements.size,id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>!!snapshot.runtime.imports.get(id).source.sha256);await engine.imports.pause(id);await engine.settle();const job=engine.imports.getJobs().get(id),raw=await f.files.get(job.source.fileId).text(),changed=raw.replace('source','tamper');assert.equal(changed.length,raw.length);await engine.close();const sent=f.requests.length;f.files.set(job.source.fileId,new Blob([changed]));engine=await f.open();await engine.imports.resume(id);await engine.settle();const blocked=engine.imports.getJobs().get(id);assert.equal(blocked.status,'paused');assert.equal(blocked.error,'import_source_changed');assert.equal(blocked.committedSourceKeys.length,0);assert.equal(engine.getSnapshot().document.elements.size,size);assert.equal(f.requests.length,sent);assert.equal(blocked.source.sha256,job.source.sha256);}finally{await engine.close();}
});

// native 尚未解析的尾部被同长度改写，重开验证拒绝并保留原首项，不允许改源后继续沿用旧任务进度。
test('a changed unparsed tail cannot pass native import recovery or alter its preserved prefix',async()=>{
 const f=fixture();let engine=await f.open();try{const id=engine.imports.startNdjson(file([node('one'),node('two'),node('three')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const job=engine.imports.getJobs().get(id),raw=await f.files.get(job.source.fileId).text(),changed=raw.replaceAll('three','other');assert.equal(changed.length,raw.length);const first=job.stableIds['element:one'];await engine.close();const sent=f.requests.length;f.files.set(job.source.fileId,new Blob([changed]));engine=await f.open();await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).error,'import_source_changed');assert.equal(engine.getSnapshot().document.elements.size,1);assert.ok(engine.getSnapshot().document.elements.has(first));assert.equal(f.requests.length,sent);}finally{await engine.close();}
});

// 摘要读取挂起期间取消或开启恢复会话，迟到结果不能复活 cancelled，也不能在 held root 发布；释放协调后才可继续。
test('late source digest cannot revive a cancelled job or publish into a held recovery root',async()=>{
 for(const cancel of [true,false]){
  const f=fixture(),engine=await f.open(),entered=deferred(),gate=deferred(),bytes=new class extends Blob{slice(...args){return new class extends Blob{async arrayBuffer(){entered.resolve();await gate.promise;return super.arrayBuffer();}}([super.slice(...args)]);}}([JSON.stringify(node('one'))]);
  try{const id=engine.imports.startNdjson(bytes,'canvas');await entered.promise;if(cancel){await engine.imports.cancel(id);gate.resolve();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'cancelled');assert.equal(engine.getSnapshot().document.elements.size,0);assert.equal(engine.imports.getJobs().get(id).source.sha256,undefined);}else{const preparing=engine.prepareRecovery();gate.resolve();const session=await preparing,revision=engine.getSnapshot().rootRevision;await new Promise(resolve=>setTimeout(resolve,10));assert.equal(engine.getSnapshot().rootRevision,revision);assert.equal((await session.apply()).status,'applied');await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(f.server().elements.size,1);}assert.equal(f.errors.length,0);}finally{gate.resolve();await engine.close();}
 }
});

// 旧未完成源缺乏完整性绑定仍可读取已有检查点，但 resume 明确拒绝补造新身份；原前缀与网络状态保留。
test('legacy unfinished sources remain readable but cannot acquire a new content identity silently',async()=>{
 const f=fixture();let engine=await f.open();try{const id=engine.imports.startNdjson(file([node('one'),node('two')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();await engine.close();const image=f.image(),job=image.payload.runtime.imports.find(([key])=>key===id)[1];job.version=2;delete job.source.sha256;delete job.source.integrityVersion;delete job.source.readProof;delete job.source.integrityError;image.formatVersion=2;const sent=f.requests.length;engine=await f.open();assert.equal(engine.imports.getJobs().get(id).status,'paused');await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).error,'import_source_integrity_unavailable');assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.requests.length,sent);decodeRecovery(f.image(),registry);}finally{await engine.close();}
});


// 把保存的索引源改造为旧完整快照，重算长度/SHA/fingerprint 并去掉 snapshotVersion，使旧准备解释器真正读取旧格式。
async function legacyCopySource(f,job,document){const signal=new AbortController().signal,bytes=await encodeCopySnapshot(document,signal);delete job.source.snapshotVersion;job.source.byteLength=bytes.size;job.source.sha256=await importSourceDigest(bytes,signal);job.sourceFingerprint=`local-file:${job.source.fileId}:${bytes.size}`;f.files.set(job.source.fileId,bytes);}

// 旧 preparation v1 已预留 legacy-frame-id，转换成旧源格式后恢复应沿用该 ID 而不再分配。
test('legacy preparation version 1 still replays its reserved identities instead of allocating replacements',async()=>{
 const f=fixture();let engine=await f.open();try{await seedCopy(engine);const sourceDocument=engine.getSnapshot().document,id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>!!snapshot.runtime.imports.get(id).source.sha256);await engine.imports.pause(id);await engine.settle();await engine.close();const job=f.image().payload.runtime.imports.find(([key])=>key===id)[1];job.preparation.version=1;job.preparation.steps=1;job.preparation.allocationIds=['legacy-frame-id'];delete job.preparation.planner;await legacyCopySource(f,job,sourceDocument);engine=await f.open();await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.imports.getJobs().get(id).stableIds['element:F'],'legacy-frame-id');assert.equal(f.errors.length,0);}finally{await engine.close();}
});


// 复制先转换并发布合法前缀，遇未知尾节点保留 partial；只存原快照而非第二份完整 mapped Blob，重开仍可整体撤销副本。
test('copy converts and publishes a prefix before the output tail, preserves it on failure and stores only the original source',async()=>{
 const source=createDocument('project',[{id:'canvas',parent:null},{id:'child',parent:{canvasId:'canvas',entryElementId:'F'}}],[copyNode('F','canvas','frame'),...Array.from({length:100},(_,i)=>copyNode(`stream-${i}`,'child')),{...copyNode('opaque','child'),componentId:'unknown-widget'}]),f=fixture(source);let engine=await f.open();
 try{let prefix;const release=engine.editor.subscribe((_,snapshot)=>{for(const job of snapshot.runtime.imports.values())if(!prefix&&job.committedSourceKeys.length)prefix={cursor:job.cursor,total:job.source.plan.order.length+job.source.plan.edges.length,status:job.status};});const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await engine.settle();release();const job=engine.imports.getJobs().get(id);assert.ok(prefix.cursor<prefix.total,'entire mapped output was converted before the first publication');assert.equal(prefix.status,'running');assert.equal(job.source.kind,'duplicate-plan');assert.equal(job.parserVersion,2);assert.equal(job.source.plan.emitted,0);assert.equal(job.status,'partial');assert.equal(job.committedSourceKeys.length,102);assert.ok(job.failures.some(failure=>failure.key==='element:opaque'&&failure.code==='copy_schema_unavailable'));assert.equal(engine.getSnapshot().document.elements.size,203);assert.equal(f.files.size,1,'copy stored a second complete mapped file');assert.equal(f.server().elements.size,203);assert.equal(f.errors.length,0);await engine.close();engine=await f.open();assert.equal(engine.imports.getJobs().get(id).status,'partial');assert.equal(engine.getSnapshot().document.elements.size,203);assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.size,102);
 }finally{await engine.close();}
});

// 旧准备版本 2/3 重开仍分别使用 duplicate-ndjson/parser1 与 duplicate-plan/parser2，不能用当前格式偷换解释器。
test('legacy preparation versions 2 and 3 retain their source interpreters after reopening',async()=>{
 for(const version of [2,3]){const f=fixture();let engine=await f.open();try{await seedCopy(engine);const sourceDocument=engine.getSnapshot().document,id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>!!snapshot.runtime.imports.get(id).source.sha256);await engine.imports.pause(id);await engine.settle();await engine.close();const job=f.image().payload.runtime.imports.find(([key])=>key===id)[1];job.preparation.version=version;await legacyCopySource(f,job,sourceDocument);engine=await f.open();await engine.imports.resume(id);await engine.settle();const done=engine.imports.getJobs().get(id);assert.equal(done.status,'completed');assert.equal(done.source.kind,version===2?'duplicate-ndjson':'duplicate-plan');assert.equal(done.parserVersion,version===2?1:2);assert.equal(f.errors.length,0);}finally{await engine.close();}}
});

// 恢复复制的 16 KiB 源解码挂起时，cancel 保留旧前缀；恢复会话 hold 则阻止后续发布，接纳后再完成。
test('late planned-source decoding respects cancellation and canonical recovery holds',async()=>{
 for(const cancel of [true,false]){
  const f=fixture();let engine=await f.open();const entered=deferred(),gate=deferred();try{
   await seedCopy(engine);const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const before=engine.imports.getJobs().get(id),size=engine.getSnapshot().document.elements.size;await engine.close();
   /* arrayBuffer 真正返回前发出 entered 并等待 gate，以制造取消/恢复期间的迟到正文。 */ class DelayedChunk extends Blob{async arrayBuffer(){entered.resolve();await gate.promise;return super.arrayBuffer();}}
   /* 仅拦住 16 KiB 源解码窗口，其他读取保留真实 Blob 行为。 */ class DelayedSource extends Blob{slice(start,end,...rest){const part=super.slice(start,end,...rest);return end-start===16*1024?new DelayedChunk([part]):part;}}
   f.files.set(before.source.fileId,new DelayedSource([f.files.get(before.source.fileId)]));engine=await f.open();const resuming=engine.imports.resume(id);await entered.promise;
   if(cancel){await engine.imports.cancel(id);gate.resolve();await resuming;await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'cancelled');assert.equal(engine.getSnapshot().document.elements.size,size);}else{const preparing=engine.prepareRecovery();gate.resolve();await resuming;const session=await preparing,revision=engine.getSnapshot().rootRevision;await new Promise(resolve=>setTimeout(resolve,10));assert.equal(engine.getSnapshot().rootRevision,revision);assert.equal(engine.getSnapshot().document.elements.size,size);assert.equal((await session.apply()).status,'applied');await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');}assert.equal(f.errors.length,0);
  }finally{gate.resolve();await engine.close();}
 }
});

// 索引已可用时，晚到大正文块挂起不阻塞前缀副本可见；放行后才读取剩余内容并完成全部副本。
test('indexed copy displays a prefix while a later raw payload read is still blocked',async()=>{
 const source=createDocument('project',[{id:'canvas',parent:null},{id:'child',parent:{canvasId:'canvas',entryElementId:'F'}}],[copyNode('F','canvas','frame'),...Array.from({length:100},(_,i)=>({...copyNode(`lazy-${i}`,'child'),data:{title:`lazy-${i}`,text:'x'.repeat(3000)}}))]),f=fixture(source);let engine=await f.open();const entered=deferred(),gate=deferred();try{
  const id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await until(engine,snapshot=>!!snapshot.runtime.imports.get(id).source.sha256);await engine.imports.pause(id);await engine.settle();const before=engine.imports.getJobs().get(id);await engine.close();const bytes=f.files.get(before.source.fileId),rows=(await bytes.text()).trimEnd().split('\n');let payload=0;
  for(const text of rows){const row=JSON.parse(text);payload+=new TextEncoder().encode(text+'\n').byteLength;if(row.kind==='payload')break;}
  /* arrayBuffer 真正返回前发出 entered 并等待 gate，以制造取消/恢复期间的迟到正文。 */ class DelayedChunk extends Blob{async arrayBuffer(){entered.resolve();await gate.promise;return super.arrayBuffer();}}
  /* 仅从正文区第二个对齐 64 KiB 窗口起挂起；头部、索引和首个原始正文窗口仍可读。 */ class DelayedPayload extends Blob{slice(start,end,...rest){const part=super.slice(start,end,...rest);return start>=payload+64*1024&&(start-payload)%(64*1024)===0?new DelayedChunk([part]):part;}}
  f.files.set(before.source.fileId,new DelayedPayload([bytes]));engine=await f.open();await engine.imports.resume(id);await entered.promise;const cut=engine.getSnapshot(),job=cut.runtime.imports.get(id);assert.equal(job.source.snapshotVersion,2);assert.equal(job.status,'running');assert.ok(job.committedSourceKeys.length>0&&job.committedSourceKeys.length<102);assert.ok(cut.document.elements.size>source.elements.size&&cut.document.elements.size<202,'raw tail read gated the entire copy');gate.resolve();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.getSnapshot().document.elements.size,202);assert.equal(f.errors.length,0);
 }finally{gate.resolve();await engine.close();}
});


// 两个项目共享摘要池，关闭一个只取消它的请求，不关闭池或影响另一个项目；被关闭项目重开后仍可继续。
test('closing a project aborts its injected digest without closing another project sharing the pool',async()=>{
 const {createDigestPool}=await import('../../src/lib/canvas-runtime/digest-pool.ts'),entered=deferred();let first=true;const pool=createDigestPool({concurrency:1,execute:async(bytes,signal)=>{if(first){first=false;entered.resolve();await new Promise((_,reject)=>signal.addEventListener('abort',()=>reject(Error('stopped')),{once:true}));}return importSourceDigest(bytes,signal);}}),a=fixture(),b=fixture();let ea=await a.open({digests:pool}),eb=await b.open({digests:pool});
 try{const aid=ea.imports.startNdjson(file([node('A')]),'canvas');await entered.promise;const bid=eb.imports.startNdjson(file([node('B')]),'canvas');await ea.close();await eb.settle();assert.equal(eb.imports.getJobs().get(bid).status,'completed');assert.equal(eb.getSnapshot().document.elements.size,1);assert.equal(a.errors.length,0);assert.equal(b.errors.length,0);ea=await a.open();await ea.settle();assert.equal(ea.imports.getJobs().get(aid).status,'completed');assert.equal(ea.getSnapshot().document.elements.size,1);}finally{await ea.close();await eb.close();pool.close();}
});

// 只手动推进回收墙上时间，真实定时器仍服务于调度；用于精确检查源文件退休宽限期。
const retentionClock=()=>{let now=1000;return{now:()=>now,advance:ms=>{now+=ms;},setTimer:(fn,ms)=>setTimeout(fn,ms),clearTimer:id=>clearTimeout(id)};};
// 任务本地终态后源文件可标记/宽限期回收，即使文档请求还未 ACK；内容、history 和冻结请求体保持，重开 undo/redo 不依赖已退休源。
test('terminal import source collection preserves every imported record, undo unit and pending request across source-v2 reopening',async()=>{
 const f=fixture(),clock=retentionClock(),gate=deferred();f.gate=gate;let engine=await f.open({clock});try{
  const id=engine.imports.startNdjson(file([node('one'),node('two')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).status==='completed');await engine.persist();const job=engine.imports.getJobs().get(id),fileId=job.source.fileId,document=engine.getSnapshot().document,history=engine.getSnapshot().history;
  const marked=await engine.collectImportSources({retentionMs:100});assert.deepEqual(marked.markedIds,[fileId]);assert.deepEqual(marked.purgedIds,[]);assert.ok(f.files.has(fileId));assert.equal(f.image().formatVersion,17);clock.advance(100);
  const result=await engine.collectImportSources({retentionMs:100});assert.deepEqual(result.purgedIds,[fileId]);assert.equal(f.files.has(fileId),false);assert.equal(engine.getSnapshot().document,document);assert.equal(engine.getSnapshot().history,history);assert.ok([...engine.getSnapshot().runtime.ledger.operations.values()].some(record=>record.phase!=='resolved'));
  // 保存所有冻结请求的 ID/body；源文件回收不应重写它们，后续重开逐个比较。
  const frozen=[...engine.getSnapshot().runtime.ledger.requests.values()].map(request=>[request.id,request.body]);decodeRecovery(f.image(),registry);for(const formatVersion of [1,2,3])assert.throws(()=>decodeRecovery({...f.image(),formatVersion},registry),/unsupported_recovery_version/);
  gate.resolve();await engine.settle();await engine.close();engine=await f.open({clock});await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.getSnapshot().document.elements.size,2);for(const [id,body]of frozen)assert.equal(engine.getSnapshot().runtime.ledger.requests.get(id).body,body);await assert.rejects(engine.imports.openSource(id),/import_source_retired/);
  assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.size,0);assert.equal(engine.editor.history.redo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.size,2);assert.equal(f.errors.length,0);
  const bad={...node('bad').record,incarnation:1,ownerCanvasId:'canvas',componentId:'image',data:{title:'bad',text:'',assetId:fileId}};assert.equal(engine.editor.dispatch({type:'element.create',element:bad}).code,'import_source_retired');assert.equal(engine.editor.updateRuntime(previous=>({...previous,resourcePins:new Map([['late-export',[fileId]]])})).code,'import_source_retired');
 }finally{gate.resolve();await engine.close();}
});

// paused 源仍需保留并可 openSource；partial/cancelled 源可回收，重开后不能借 resume 复活内容或重复失败。
test('paused source remains recoverable while partial and cancelled sources become collectible without reviving content',async()=>{
 const f=fixture();let engine=await f.open();try{
  const id=engine.imports.startNdjson(file([node('one'),node('two'),node('three')]),'canvas');await until(engine,snapshot=>snapshot.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const source=engine.imports.getJobs().get(id).source;assert.deepEqual((await engine.collectImportSources({retentionMs:0})).purgedIds,[]);assert.ok(f.files.has(source.fileId));const handle=await engine.imports.openSource(id);assert.equal(await handle.bytes.text(),await file([node('one'),node('two'),node('three')]).text());handle.release();
  await engine.imports.cancel(id);await engine.settle();const partial=engine.imports.startNdjson(file([node('valid'),'not json']),'canvas');await engine.settle();assert.equal(engine.imports.getJobs().get(partial).status,'partial');const expected=new Set([source.fileId,engine.imports.getJobs().get(partial).source.fileId]);assert.deepEqual(new Set((await engine.collectImportSources({retentionMs:0})).purgedIds),expected);const size=engine.getSnapshot().document.elements.size;await engine.close();engine=await f.open();await engine.imports.resume(id);await engine.imports.resume(partial);await engine.settle();assert.equal(engine.getSnapshot().document.elements.size,size);assert.equal(engine.imports.getJobs().get(id).status,'cancelled');assert.equal(engine.imports.getJobs().get(partial).failures.length,1);assert.equal(f.errors.length,0);
 }finally{await engine.close();}
});

// 打开终态源前先 pin 并清除 soft marker，异步读取期间不可删；最后 release 后重新开始完整宽限期。
test('opening a terminal source pins before I/O, clears its soft marker and restarts the full retention interval',async()=>{
 const f=fixture(),clock=retentionClock(),engine=await f.open({clock}),gate=deferred(),entered=deferred(),read=f.storage.readFile;try{
  const id=engine.imports.startNdjson(file([node('one')]),'canvas');await engine.settle();const fileId=engine.imports.getJobs().get(id).source.fileId;await engine.collectImportSources({retentionMs:100});clock.advance(200);f.storage.readFile=async(...args)=>{if(args[1]===fileId){entered.resolve();await gate.promise;}return read(...args);};
  const opening=engine.imports.openSource(id);await entered.promise;assert.equal(engine.imports.getJobs().get(id).source.retirement,undefined);assert.ok(engine.getResourcePins().pins.get(fileId).has('display'));assert.deepEqual((await engine.collectImportSources({retentionMs:0})).purgedIds,[]);gate.resolve();const handle=await opening;handle.release();handle.release();const result=await engine.collectImportSources({retentionMs:100});assert.deepEqual(result.markedIds,[fileId]);assert.deepEqual(result.purgedIds,[]);assert.equal(engine.imports.getJobs().get(id).source.retirement.unreferencedAt,1200);
 }finally{gate.resolve();await engine.close();}
});

// 源文件删除落盘失败保留字节与 reservation，阻止 openSource 及非法恢复 running；persist 补成同一 purge cut 后才退休，内容仍可编辑。
test('failed source deletion retains byte identity and blocks late reads until the same purge cut becomes durable',async()=>{
 const f=fixture(),engine=await f.open(),commit=f.storage.commit;let fail=false;try{
  const id=engine.imports.startNdjson(file([node('one')]),'canvas');await engine.settle();const fileId=engine.imports.getJobs().get(id).source.fileId;f.storage.commit=async(...args)=>{if(fail&&args[1].deleteFiles?.length)throw Error('source disk failure');return commit(...args);};fail=true;await assert.rejects(engine.collectImportSources({retentionMs:0}),/source disk failure/);assert.ok(f.files.has(fileId));await assert.rejects(engine.imports.openSource(id),/resource_collection_in_progress/);const root=engine.getSnapshot(),job=root.runtime.imports.get(id);assert.equal(engine.editor.updateRuntime(previous=>{const jobs=new Map(previous.imports);jobs.set(id,{...job,status:'running',source:{...job.source,retirement:undefined}});return{...previous,imports:jobs};}).status,'rejected');assert.equal(engine.getSnapshot(),root);
  fail=false;await engine.persist();assert.equal(f.files.has(fileId),false);await assert.rejects(engine.imports.openSource(id),/import_source_retired/);assert.equal(engine.editor.dispatch({type:'element.data',target:{id:job.stableIds['element:one'],incarnation:1},patch:{title:'still editable'}}).status,'applied');await engine.settle();
 }finally{fail=false;await engine.close();}
});

// 旧快照转 NDJSON 时归档 previousFiles，旧代可单独 pin/回收；当前代完成后也可回收，元素引用和历史仍可恢复。
test('legacy snapshot conversion records its previous file, then collects both generations without losing copied references',async()=>{
 const f=fixture();let engine=await f.open();try{
  await seedCopy(engine);const captured=engine.getSnapshot().document,id=await engine.imports.duplicate([{id:'F',incarnation:1}]);await engine.imports.pause(id);await engine.settle();await engine.close();const image=f.image(),job=image.payload.runtime.imports.find(([key])=>key===id)[1],bytes=await encodeCopySnapshot(captured,new AbortController().signal),oldFile=job.source.fileId;f.files.set(oldFile,bytes);job.source.byteLength=bytes.size;job.source.sha256=await importSourceDigest(bytes);job.sourceFingerprint=`local-file:${oldFile}:${bytes.size}`;delete job.source.snapshotVersion;job.preparation={version:1,registrySignature:job.preparation.registrySignature,selection:[{id:'F',incarnation:1}],worldOffset:{x:32,y:32},allocationIds:[],steps:0};decodeRecovery(image,registry);
  engine=await f.open();await engine.imports.resume(id);await engine.imports.pause(id);await engine.settle();const converted=engine.imports.getJobs().get(id),newFile=converted.source.fileId;assert.equal(converted.status,'paused');assert.equal(converted.source.kind,'duplicate-ndjson');assert.notEqual(newFile,oldFile);assert.equal(converted.source.previousFiles.length,1);assert.equal(converted.source.previousFiles[0].fileId,oldFile);assert.equal(f.image().formatVersion,4);assert.ok(f.files.has(oldFile)&&f.files.has(newFile));const held=await engine.imports.openSource(id,oldFile);assert.equal(held.bytes.size,bytes.size);assert.deepEqual((await engine.collectImportSources({retentionMs:0})).purgedIds,[]);held.release();assert.deepEqual((await engine.collectImportSources({retentionMs:0})).purgedIds,[oldFile]);assert.ok(f.files.has(newFile));await engine.imports.resume(id);await engine.settle();const completed=engine.imports.getJobs().get(id);assert.equal(completed.status,'completed');const copied=completed.stableIds['element:F'];assert.ok(engine.getSnapshot().document.elements.has(copied));
  const result=await engine.collectImportSources({retentionMs:0});assert.deepEqual(result.purgedIds,[newFile]);assert.equal(f.files.has(oldFile),false);assert.equal(f.files.has(newFile),false);decodeRecovery(f.image(),registry);await engine.close();engine=await f.open();assert.equal(engine.getSnapshot().document.elements.get(completed.stableIds['element:A']).data.assetId,'shared');assert.equal(engine.getSnapshot().document.elements.get(completed.stableIds['element:incoming']).source.elementId,'C');assert.equal(engine.editor.history.undo().status,'applied');await engine.settle();assert.equal(engine.getSnapshot().document.elements.has(copied),false);assert.equal(f.errors.length,0);
 }finally{await engine.close();}
});

// openSource 验证当前文件与绑定摘要；篡改拒绝并释放 pin，读取期间关闭项目则迟到结果报 engine_closing。
test('source read handles verify bound bytes and never publish a result after project close',async()=>{
 const f=fixture(),engine=await f.open(),gate=deferred(),entered=deferred(),read=f.storage.readFile;try{
  const id=engine.imports.startNdjson(file([node('one')]),'canvas');await engine.settle();const fileId=engine.imports.getJobs().get(id).source.fileId,bytes=f.files.get(fileId),text=await bytes.text();f.files.set(fileId,new Blob([text.replace('one','two')]));await assert.rejects(engine.imports.openSource(id),/import_source_changed/);assert.equal(engine.getResourcePins().pins.has(fileId),false);f.files.set(fileId,bytes);
  f.storage.readFile=async(owner,key)=>{const value=await read(owner,key);if(key===fileId){entered.resolve();await gate.promise;}return value;};const opening=engine.imports.openSource(id);await entered.promise;await engine.close();gate.resolve();await assert.rejects(opening,/engine_closing/);assert.equal(engine.getResourcePins().pins.has(fileId),false);
 }finally{gate.resolve();await engine.close();}
});

// JSON 数组任务保存原字节及字节游标，暂停/编辑/重开再续跑保持原 ID 和用户标题，证明型 image 不允许格式降级。
test('JSON array import persists original bytes and byte cursor, resumes after reopen, and preserves edits without duplicating IDs',async()=>{const f=fixture();let engine=await f.open();try{const bytes=new Blob([JSON.stringify([node('json-one'),node('json-two'),node('json-three')])]),id=engine.imports.startJsonArray(bytes,'canvas');await until(engine,s=>s.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const job=engine.imports.getJobs().get(id),first=job.stableIds['element:json-one'];assert.equal(job.parserVersion,3);assert.equal(job.source.kind,'native-json-array');assert.ok(job.cursor>0);assert.equal(f.image().formatVersion,17);decodeRecovery(f.image(),registry);assert.throws(()=>decodeRecovery({...f.image(),formatVersion:13},registry),/unsupported_recovery_version/);assert.throws(()=>decodeRecovery({...f.image(),formatVersion:16},registry),/unsupported_recovery_version/);engine.editor.dispatch({type:'element.data',target:{id:first,incarnation:1},patch:{title:'preserved edit'}});await engine.settle();await engine.close();engine=await f.open();await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.getSnapshot().document.elements.size,3);assert.equal(engine.getSnapshot().document.elements.get(first).data.title,'preserved edit');assert.equal(f.server().elements.size,3);assert.equal(f.errors.length,0);}finally{await engine.close();}});
// 数组尾逗号使任务 partial，只保留解析器已接受的合法前缀；即使后端 gate 未放行，本地也能展示该前缀和显式错误。
test('JSON array malformed tail keeps its visible prefix and an explicit failure while backend waits',async()=>{const f=fixture(),engine=await f.open(),gate=deferred();f.gate=gate;try{const id=engine.imports.startJsonArray(new Blob(['['+JSON.stringify(node('json-first'))+','+JSON.stringify(node('json-bad'))+',]']),'canvas');await until(engine,s=>s.runtime.imports.get(id).status==='partial');const job=engine.imports.getJobs().get(id);assert.equal(job.committedSourceKeys.length,1);assert.equal(job.failures.at(-1).code,'invalid_json_array');assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.server().elements.size,0);gate.resolve();await engine.settle();assert.equal(f.server().elements.size,1);decodeRecovery(f.image(),registry);}finally{gate.resolve();await engine.close();}});

// 导出文档导入指定目标根，生成新元素/子 canvas ID，重建 owner 与上级连线；旧源墓碑不移植为目标身份，重开保留四元素。
test('exported document imports into a chosen root with fresh IDs, child ownership and upper connections then reopens',async()=>{const f=fixture();let engine=await f.open();try{const frame={...node('frame').record,componentId:'frame',ownerCanvasId:'source-root'},child={...node('child').record,ownerCanvasId:'source-child'},outside={...node('outside').record,ownerCanvasId:'source-root'},edge={id:'incoming',incarnation:1,ownerCanvasId:'source-root',componentId:'connection',dataVersion:1,structure:'connection',source:{elementId:'outside'},target:{elementId:'frame'},data:{},path:null},input={documentFormatVersion:1,projectId:'source-root',canvases:[{id:'source-root',parent:null},{id:'source-child',parent:{canvasId:'source-root',entryElementId:'frame'}}],elements:[frame,child,outside,edge],tombstones:{old:2}},id=engine.imports.startDocument(new Blob([JSON.stringify(input)]),'canvas');await engine.settle();const job=engine.imports.getJobs().get(id);assert.equal(job.status,'completed',JSON.stringify(job.failures));const doc=engine.getSnapshot().document,frameId=job.stableIds['element:frame'],childId=job.stableIds['canvas:source-child'];assert.equal(doc.canvases.size,2);assert.equal(doc.canvases.get(childId).parent.entryElementId,frameId);assert.equal(doc.elements.get(job.stableIds['element:child']).ownerCanvasId,childId);assert.equal(doc.elements.get(job.stableIds['element:incoming']).ownerCanvasId,'canvas');assert.equal(doc.elements.get(job.stableIds['element:incoming']).target.elementId,frameId);assert.equal(doc.tombstones.has('old'),false);assert.equal(f.image().formatVersion,17);decodeRecovery(f.image(),registry);await engine.close();engine=await f.open();await engine.settle();assert.equal(engine.getSnapshot().document.elements.size,4);assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(f.errors.length,0);}finally{await engine.close();}});
// 无资源授权端口时图片引用报 import_resource_authorization_required，独立文本仍进入文档并同步。
test('document resource references report an authorization failure while independent text imports',async()=>{const f=fixture(),engine=await f.open();try{const image={...node('image').record,componentId:'image',ownerCanvasId:'source',data:{title:'image',assetId:'untrusted',src:'https://assets.test/untrusted'}},text={...node('text').record,ownerCanvasId:'source'},input={documentFormatVersion:1,projectId:'source',canvases:[{id:'source',parent:null}],elements:[image,text],tombstones:{}},id=engine.imports.startDocument(new Blob([JSON.stringify(input)]),'canvas');await engine.settle();const job=engine.imports.getJobs().get(id);assert.equal(job.status,'partial');assert.ok(job.failures.some(row=>row.code==='import_resource_authorization_required'));assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.server().elements.size,1);}finally{await engine.close();}});
// 导出文档使用记录序号恢复，首项后暂停编辑，重开继续不覆盖手动标题或重复元素。
test('paused exported document resumes ordinal progress after reopen without overwriting an edited prefix',async()=>{const f=fixture();let engine=await f.open();try{const input={documentFormatVersion:1,projectId:'source',canvases:[{id:'source',parent:null}],elements:['d-one','d-two','d-three'].map(id=>({...node(id).record,ownerCanvasId:'source'})),tombstones:{}},id=engine.imports.startDocument(new Blob([JSON.stringify(input)]),'canvas');await until(engine,s=>s.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const job=engine.imports.getJobs().get(id),first=job.stableIds['element:d-one'];assert.equal(job.cursor,1);engine.editor.dispatch({type:'element.data',target:{id:first,incarnation:1},patch:{title:'local edit'}});await engine.settle();await engine.close();engine=await f.open();assert.equal(engine.imports.getJobs().get(id).status,'paused');await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.getSnapshot().document.elements.size,3);assert.equal(engine.getSnapshot().document.elements.get(first).data.title,'local edit');assert.equal(f.errors.length,0);}finally{await engine.close();}});

// parser 5 尾部扫描挂起时首项已可编辑；暂停/重开继续同解释器和稳定 ID，保留编辑，错误 image 版本拒绝。
test('streaming exported document publishes an editable prefix before the parser tail and resumes parser 5 after reopen',async()=>{
 const f=fixture(),gate=deferred(),entered=deferred();let block=true,engine=await f.open();
 /* 文档扫描从 32 KiB 起的 16 KiB 窗口可挂起；前项仍可先解析和发布。 */ class Source extends Blob{slice(start=0,end=this.size,...rest){const part=super.slice(start,end,...rest);if(end-start!==16384||start<32768||!block)return part;return{async arrayBuffer(){entered.resolve();await gate.promise;return part.arrayBuffer();}};}}
 try{
  const input={documentFormatVersion:1,projectId:'source',canvases:[{id:'source',parent:null}],elements:[{...node('stream-first').record,ownerCanvasId:'source'},{...node('stream-tail').record,ownerCanvasId:'source',data:{title:'tail',text:'x'.repeat(100000)}}],tombstones:{}},id=engine.imports.startDocument(new Source([JSON.stringify(input)]),'canvas');
  await entered.promise;await until(engine,root=>root.runtime.imports.get(id).committedSourceKeys.includes('element:stream-first'));
  const job=engine.imports.getJobs().get(id),first=job.stableIds['element:stream-first'];assert.equal(job.parserVersion,5);assert.equal(engine.getSnapshot().document.elements.size,1);
  assert.equal(engine.editor.dispatch({type:'element.data',target:{id:first,incarnation:1},patch:{title:'edited while tail blocked'}}).status,'applied');
  // 先请求暂停再释放底层扫描，验证迟到尾部不会绕过暂停；随后持久化 parser/proof 以供重开。
  const paused=engine.imports.pause(id);block=false;gate.resolve();await paused;await engine.settle();assert.equal(f.image().formatVersion,17);decodeRecovery(f.image(),registry);assert.throws(()=>decodeRecovery({...f.image(),formatVersion:15},registry),/unsupported_recovery_version/);assert.throws(()=>decodeRecovery({...f.image(),formatVersion:18},registry),/unsupported_recovery_version/);
  await engine.close();engine=await f.open();await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.getSnapshot().document.elements.size,2);assert.equal(engine.getSnapshot().document.elements.get(first).data.title,'edited while tail blocked');assert.equal(f.server().elements.size,2);assert.equal(f.errors.length,0);
 }finally{block=false;gate.resolve();await engine.close();}
});

// 文档 footer 墓碑非法时已导入元素仍可见并同步；partial 及错误诊断持久化，重开保留该成功前缀。
test('bad document footer produces durable partial success rather than discarding imported content',async()=>{
 const f=fixture();let engine=await f.open();try{
  const input={documentFormatVersion:1,projectId:'source',canvases:[{id:'source',parent:null}],elements:[{...node('legal').record,ownerCanvasId:'source'}],tombstones:{broken:0}},id=engine.imports.startDocument(new Blob([JSON.stringify(input)]),'canvas');await engine.settle();const job=engine.imports.getJobs().get(id);assert.equal(job.status,'partial');assert.ok(job.failures.some(row=>row.code==='invalid_document_import'));assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.server().elements.size,1);await engine.close();engine=await f.open();assert.equal(engine.imports.getJobs().get(id).status,'partial');assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.errors.length,0);
 }finally{await engine.close();}
});

// 显式构造旧格式 15/parser4/integrity1 检查点，恢复沿用原解释器及 ID，不能暗升成 parser5 或改变 image 格式。
test('legacy format 15 parser 4 document jobs keep their original interpreter and stable IDs when resumed',async()=>{
 const f=fixture();let engine=await f.open();try{
  const input={documentFormatVersion:1,projectId:'source',canvases:[{id:'source',parent:null}],elements:['legacy-one','legacy-two'].map(id=>({...node(id).record,ownerCanvasId:'source'})),tombstones:{}},id=engine.imports.startDocument(new Blob([JSON.stringify(input)]),'canvas');await until(engine,root=>root.runtime.imports.get(id).committedSourceKeys.length===1);await engine.imports.pause(id);await engine.settle();const first=engine.imports.getJobs().get(id).stableIds['element:legacy-one'];await engine.close();
  const legacy=structuredClone(f.image());legacy.formatVersion=15;const legacyJob=legacy.payload.runtime.imports.find(([key])=>key===id)[1];legacyJob.parserVersion=4;legacyJob.source.integrityVersion=1;delete legacyJob.source.readProof;delete legacyJob.source.integrityError;decodeRecovery(legacy,registry);f.storage.recover=async()=>structuredClone(legacy);
  engine=await f.open();assert.equal(engine.imports.getJobs().get(id).parserVersion,4);await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');assert.equal(engine.imports.getJobs().get(id).stableIds['element:legacy-one'],first);assert.equal(engine.getSnapshot().document.elements.size,2);assert.equal(f.image().formatVersion,15);assert.equal(f.errors.length,0);
 }finally{await engine.close();}
});

// 临时正文 I/O 错误暂停在原 ordinal，eof=false 且无坏记录诊断；恢复端口后继续尾项，保留首项手动编辑。
test('temporary document byte I/O pauses without consuming a bad-tail ordinal and resumes with edited prefix intact',async()=>{
 const f=fixture(),engine=await f.open();let fail=true;
 /* 仅让后续 16 KiB 扫描窗口抛临时 I/O；切换 fail 后同一 Blob 可读，不改变字节身份。 */ class Flaky extends Blob{slice(start=0,end=this.size,...rest){if(fail&&end-start===16384&&start>=16384)throw Error('temporary source I/O');return super.slice(start,end,...rest);}}
 try{
  const input={documentFormatVersion:1,projectId:'source',canvases:[{id:'source',parent:null}],elements:[{...node('io-first').record,ownerCanvasId:'source'},{...node('io-tail').record,ownerCanvasId:'source',data:{title:'tail',text:'x'.repeat(100000)}}]},id=engine.imports.startDocument(new Flaky([JSON.stringify(input)]),'canvas');
  await until(engine,snapshot=>snapshot.runtime.imports.get(id).status==='paused');const job=engine.imports.getJobs().get(id),first=job.stableIds['element:io-first'];assert.equal(job.error,'import_source_read_failed');assert.equal(job.eof,false);assert.equal(job.cursor,1);assert.equal(job.failures.length,0);assert.equal(job.committedSourceKeys.length,1);
  assert.equal(engine.editor.dispatch({type:'element.data',target:{id:first,incarnation:1},patch:{title:'edited prefix'}}).status,'applied');fail=false;await engine.imports.resume(id);await engine.settle();const done=engine.imports.getJobs().get(id);assert.equal(done.status,'completed');assert.equal(done.failures.length,0);assert.equal(engine.getSnapshot().document.elements.size,2);assert.equal(engine.getSnapshot().document.elements.get(first).data.title,'edited prefix');assert.equal(f.server().elements.get(first).data.title,'edited prefix');assert.ok(f.errors.some(error=>error.code==='import_source_read_failed'));decodeRecovery(f.image(),registry);
 }finally{fail=false;await engine.close();}
});

// 完整 SHA 尚未返回时读取证明允许先发布/编辑前缀；即使 persist 本地内容，网络仍不发，待摘要成功后才同步用户编辑。
test('native import publishes and edits a proof-backed prefix before full SHA, with every request waiting for its durable seal',async()=>{
 const f=fixture(),gate=deferred(),entered=deferred(),engine=await f.open({digests:{async digest(bytes,signal){entered.resolve();await gate.promise;return importSourceDigest(bytes,signal);}}});
 try{const id=engine.imports.startNdjson(file([node('early-one'),node('early-two')]),'canvas');await entered.promise;await until(engine,root=>root.runtime.imports.get(id).committedSourceKeys.length>0);const job=engine.imports.getJobs().get(id),first=job.stableIds['element:early-one'];assert.equal(job.source.integrityVersion,2);assert.ok(job.source.readProof.blocks.length);assert.equal(job.source.sha256,undefined);assert.equal(engine.editor.dispatch({type:'element.data',target:{id:first,incarnation:1},patch:{title:'edit before hash'}}).status,'applied');await engine.persist();assert.equal(f.requests.length,0);assert.equal(f.server().elements.size,0);assert.equal(f.image().formatVersion,17);decodeRecovery(f.image(),registry);gate.resolve();await engine.settle();assert.ok(engine.imports.getJobs().get(id).source.sha256);assert.equal(f.server().elements.get(first).data.title,'edit before hash');assert.equal(f.errors.length,0);}finally{gate.resolve();await engine.close();}
});

// 内存中已有完整 SHA 仍不足以放行网络；必须等含该 seal 的恢复 cut 真正落盘，再允许相关文档请求。
test('a verified SHA still cannot release pending import writes before the seal cut is durable',async()=>{
 const f=fixture(),hash=deferred(),save=deferred(),saving=deferred(),commit=f.storage.commit,engine=await f.open({digests:{async digest(bytes,signal){await hash.promise;return importSourceDigest(bytes,signal);}}});
 try{const id=engine.imports.startNdjson(file([node('seal-one')]),'canvas');await until(engine,root=>root.runtime.imports.get(id).committedSourceKeys.length===1);await engine.persist();f.storage.commit=async(owner,batch)=>{if(batch.image.payload.runtime.imports?.some(([key,job])=>key===id&&job.source.sha256)){saving.resolve();await save.promise;}return commit(owner,batch);};hash.resolve();await saving.promise;assert.ok(engine.imports.getJobs().get(id).source.sha256);assert.equal(f.requests.length,0);save.resolve();await engine.settle();assert.equal(f.server().elements.size,1);assert.equal(f.errors.length,0);}finally{hash.resolve();save.resolve();await engine.close();}
});

// 未封存前 pause/cancel 保留可见前缀、继续挡网络且暂不可回收源；校验后只同步已接受内容，不改变暂停/取消终态。
test('pause and cancel retain the unverified prefix and never turn off the source gate early',async()=>{
 for(const cancel of [false,true]){const f=fixture(),gate=deferred(),engine=await f.open({digests:{async digest(bytes,signal){await gate.promise;return importSourceDigest(bytes,signal);}}});try{const id=engine.imports.startNdjson(file([node('stop-one'),node('stop-two')]),'canvas');await until(engine,root=>root.runtime.imports.get(id).committedSourceKeys.length>0);if(cancel)await engine.imports.cancel(id);else await engine.imports.pause(id);await engine.persist();assert.equal(f.requests.length,0);assert.deepEqual((await engine.collectImportSources({retentionMs:0})).purgedIds,[]);const count=engine.getSnapshot().document.elements.size;gate.resolve();await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,cancel?'cancelled':'paused');assert.equal(f.server().elements.size,count);assert.equal(f.errors.length,0);}finally{gate.resolve();await engine.close();}}
});

// 源验证失败保留本地内容和稳定 ID，阻止发送；修复 digest 端口再 resume，只补验证与剩余导入，不重复身份。
test('a source verification failure retains local content, blocks sending and can retry without duplicate identities',async()=>{
 const f=fixture(),gate=deferred();let fail=true;const engine=await f.open({digests:{async digest(bytes,signal){await gate.promise;if(fail)throw Error('digest unavailable');return importSourceDigest(bytes,signal);}}});
 try{const id=engine.imports.startNdjson(file([node('retry-one'),node('retry-two')]),'canvas');await until(engine,root=>root.runtime.imports.get(id).committedSourceKeys.length>0);gate.resolve();await engine.settle();const job=engine.imports.getJobs().get(id),ids={...job.stableIds};assert.equal(job.source.integrityError,'import_source_verification_failed');assert.equal(f.requests.length,0);assert.ok(engine.getSnapshot().document.elements.size>0);fail=false;await engine.imports.resume(id);await engine.settle();assert.equal(engine.imports.getJobs().get(id).status,'completed');for(const[key,id]of Object.entries(ids))assert.equal(engine.imports.getJobs().get(job.id).stableIds[key],id);assert.equal(f.server().elements.size,2);assert.equal(engine.imports.getJobs().get(id).source.integrityError,undefined);}finally{fail=false;gate.resolve();await engine.close();}
});

// 未 seal 的持久化检查点重开保留原前缀/ID，仍挡发送；重新验证并续跑后才同步完整内容。
test('an unsealed durable checkpoint reopens with the same prefix and keeps sends blocked until reverified',async()=>{
 const f=fixture(),gate=deferred(),entered=deferred();let engine=await f.open({digests:{digest(bytes,signal){entered.resolve();return new Promise((resolve,reject)=>{const abort=()=>reject(Error('aborted'));signal.addEventListener('abort',abort,{once:true});gate.promise.then(()=>{signal.removeEventListener('abort',abort);importSourceDigest(bytes,signal).then(resolve,reject);});});}}});
 try{const id=engine.imports.startNdjson(file([node('reopen-one'),node('reopen-two')]),'canvas');await entered.promise;await until(engine,root=>root.runtime.imports.get(id).committedSourceKeys.length>0);await engine.imports.pause(id);await engine.persist();const ids={...engine.imports.getJobs().get(id).stableIds};assert.equal(f.image().formatVersion,17);await engine.close();const nextGate=deferred();engine=await f.open({digests:{async digest(bytes,signal){await nextGate.promise;return importSourceDigest(bytes,signal);}}});await engine.persist();assert.equal(f.requests.length,0);assert.deepEqual(engine.imports.getJobs().get(id).stableIds,ids);nextGate.resolve();await engine.imports.resume(id);await engine.settle();assert.equal(f.server().elements.size,2);for(const[key,id]of Object.entries(ids))assert.equal(engine.imports.getJobs().get(engine.imports.getJobs().keys().next().value).stableIds[key],id);}finally{gate.resolve();await engine.close();}
});

// 恢复会话取消旧 verifier 后，接纳基线需重新启动验证；不能留下无错误但永远不放行网络的静默 hold。
test('canonical recovery restarts an aborted unsealed-source verifier instead of leaving a silent send hold',async()=>{
 const f=fixture(),gate=deferred();let calls=0;const engine=await f.open({digests:{digest(bytes,signal){calls++;return new Promise((resolve,reject)=>{const abort=()=>reject(Error('aborted'));signal.addEventListener('abort',abort,{once:true});gate.promise.then(()=>{signal.removeEventListener('abort',abort);importSourceDigest(bytes,signal).then(resolve,reject);});});}}});
 try{const id=engine.imports.startNdjson(file([node('hold-one'),node('hold-two')]),'canvas');await until(engine,root=>root.runtime.imports.get(id).committedSourceKeys.length>0);const session=await engine.prepareRecovery();assert.equal((await session.apply()).status,'applied');for(let i=0;i<100&&calls<2;i++)await new Promise(resolve=>setTimeout(resolve,1));assert.ok(calls>=2,'source verifier did not restart after recovery released');assert.equal(f.requests.length,0);gate.resolve();await engine.settle();assert.equal(f.server().elements.size,2);assert.equal(f.errors.length,0);}finally{gate.resolve();await engine.close();}
});

// 未 seal 源重开先重验已记录块，篡改即阻塞且不再分配 ID；恢复原字节后才允许重试和同步。
test('unsealed recovery checks recorded bytes before allocating more IDs or releasing writes',async()=>{
 const f=fixture(),gate=deferred();let engine=await f.open({digests:{digest(bytes,signal){return new Promise((resolve,reject)=>{signal.addEventListener('abort',()=>reject(Error('aborted')),{once:true});gate.promise.then(()=>importSourceDigest(bytes,signal).then(resolve,reject));});}}});
 try{const id=engine.imports.startNdjson(file([node('proof-one'),node('proof-two')]),'canvas');await until(engine,root=>root.runtime.imports.get(id).committedSourceKeys.length>0);await engine.imports.pause(id);await engine.persist();const before=engine.imports.getJobs().get(id),ids={...before.stableIds};await engine.close();const bytes=f.files.get(before.source.fileId),text=await bytes.text();f.files.set(before.source.fileId,new Blob([text.replace('proof-two','wrong-two')]));engine=await f.open();await engine.settle();const job=engine.imports.getJobs().get(id);assert.equal(job.source.integrityError,'import_source_changed');assert.equal(f.requests.length,0);assert.deepEqual(job.stableIds,ids);assert.ok(engine.getSnapshot().document.elements.size>0);f.files.set(before.source.fileId,bytes);await engine.imports.resume(id);await engine.settle();assert.equal(f.server().elements.size,2);}finally{gate.resolve();await engine.close();}
});

// 源摘要、读取证明及完整性版本由引擎拥有，公共 updateRuntime 不能伪造证据或清空 committed 进度；拒绝后 root 不变。
test('source evidence is engine-owned and cannot be forged through public runtime updates',async()=>{
 const f=fixture(),engine=await f.open();try{const id=engine.imports.startNdjson(file([node('owned-proof')]),'canvas');await engine.settle();const before=engine.getSnapshot(),job=engine.imports.getJobs().get(id);for(const source of [{...job.source,sha256:'0'.repeat(64)},{...job.source,readProof:{...job.source.readProof,blocks:[]}},{...job.source,integrityVersion:1,readProof:undefined}]){const result=engine.editor.updateRuntime(runtime=>({...runtime,imports:new Map(runtime.imports).set(id,{...job,source})}));assert.equal(result.status,'rejected');assert.equal(engine.getSnapshot(),before);}const forged=engine.editor.updateRuntime(runtime=>({...runtime,imports:new Map(runtime.imports).set(id,{...job,status:'cancelled',committedSourceKeys:[]})}));assert.equal(forged.status,'rejected');assert.equal(engine.getSnapshot(),before);assert.equal(f.errors.length,0);}finally{await engine.close();}
});

// 取消后已接受前缀仍需验证：engine.retry 可补完失败校验并同步该前缀，但保持 cancelled 且不重新开始插入。
test('retry verifies a cancelled accepted prefix without restarting insertion',async()=>{
 const f=fixture(),gate=deferred();let fail=true;const engine=await f.open({digests:{async digest(bytes,signal){await gate.promise;if(fail)throw Error('digest unavailable');return importSourceDigest(bytes,signal);}}});try{const id=engine.imports.startNdjson(file([node('retry-cancel-one'),node('retry-cancel-two')]),'canvas');await until(engine,root=>root.runtime.imports.get(id).committedSourceKeys.length>0);await engine.imports.cancel(id);gate.resolve();await engine.settle();const before=engine.imports.getJobs().get(id),count=engine.getSnapshot().document.elements.size;assert.ok(before.source.integrityError);assert.equal(f.requests.length,0);fail=false;await engine.retry();await engine.settle();const after=engine.imports.getJobs().get(id);assert.equal(after.status,'cancelled');assert.equal(after.source.integrityError,undefined);assert.ok(after.source.sha256);assert.deepEqual(after.stableIds,before.stableIds);assert.equal(engine.getSnapshot().document.elements.size,count);assert.equal(f.server().elements.size,count);}finally{gate.resolve();await engine.close();}
});

// 旧数组格式 14 和旧 streaming document 格式 16 的 integrity1 状态仍按 SHA-first 语义读取，分别保留 parser3/5。
test('legacy native array 14 and streaming document 16 retain their SHA-first interpreters',async()=>{
 for(const format of ['array','document']){const f=fixture();let engine=await f.open();try{const bytes=format==='array'?new Blob([JSON.stringify([node('legacy-native')])]):new Blob([JSON.stringify({documentFormatVersion:1,projectId:'legacy-source',canvases:[{id:'legacy-source',parent:null}],elements:[{...node('legacy-native').record,ownerCanvasId:'legacy-source'}],tombstones:{}})]),id=format==='array'?engine.imports.startJsonArray(bytes,'canvas'):engine.imports.startDocument(bytes,'canvas');await engine.settle();await engine.close();const image=structuredClone(f.image());image.formatVersion=format==='array'?14:16;const job=image.payload.runtime.imports.find(([key])=>key===id)[1];job.source.integrityVersion=1;delete job.source.readProof;delete job.source.integrityError;const decoded=decodeRecovery(image,registry);assert.equal(decoded.runtime.imports.get(id).source.integrityVersion,1);f.storage.recover=async()=>structuredClone(image);engine=await f.open();await engine.settle();assert.equal(engine.imports.getJobs().get(id).parserVersion,format==='array'?3:5);assert.equal(engine.getSnapshot().document.elements.size,1);assert.equal(f.errors.length,0);}finally{await engine.close();}}
});

// 千项真实引擎准备/ready 源在发布监听中异步封存，后续推进不改变捕获字节；准备编码实际让出，重开保留暂停和映射。
test('engine preparation encodes in slices and the ready source retains checkpoint bytes and mappings on reopen', {timeout:20000},async()=>{
 const {sealRecoveryImageAsync}=await import('../../src/lib/canvas-runtime/recovery-encoding.ts');
 const rows=Array.from({length:1000},(_,i)=>({...node('live-checkpoint-'+i).record,incarnation:1,ownerCanvasId:'canvas'}));
 const initial=createDocument('project',[{id:'canvas',parent:null}],rows),f=fixture(initial),gate=deferred();f.gate=gate;
 let engine=await f.open(),unsubscribe=()=>{};const captured=[],seen=new Set();
 try{
  await engine.persist();
  unsubscribe=engine.editor.subscribe((_,snapshot)=>{
   for(const job of snapshot.runtime.imports.values()){
    const candidates=[['preparation',job.preparation?.planner?job.preparation:undefined],['source',job.source?.kind==='duplicate-plan'?job.source:undefined]];
    for(const [phase,part] of candidates){if(!part||seen.has(phase))continue;seen.add(phase);const before=JSON.stringify(part);let ticks=0,yields=0;
     const pending=sealRecoveryImageAsync(17,1,1,part,{now:()=>ticks++,frameBudgetMs:1,yieldTask:async()=>{yields++;}});
     captured.push(pending.then(image=>({phase,image,before,yields}),error=>({phase,error})));
    }
   }
  });
  const id=await engine.imports.duplicate(rows.map(row=>({id:row.id,incarnation:1})));
  await until(engine,snapshot=>snapshot.runtime.imports.get(id)?.source?.kind==='duplicate-plan');
  await engine.imports.pause(id);unsubscribe();
  // 等待发布监听里启动的封存全部完成，检查准备与源两个阶段都被捕获，且编码期间后续变更未改写旧值。
  const results=await Promise.all(captured);assert.deepEqual([...seen].sort(),['preparation','source']);
  for(const result of results){assert.equal(result.error,undefined);if(result.phase==='preparation')assert.ok(result.yields>0,'preparation checkpoint was synchronously normalized');assert.equal(JSON.stringify(result.image.payload),result.before);}
  const ids=engine.imports.getJobs().get(id).stableIds;gate.resolve();await engine.close();engine=await f.open();
  assert.equal(engine.imports.getJobs().get(id).status,'paused');assert.deepEqual(engine.imports.getJobs().get(id).stableIds,ids);assert.equal(f.errors.length,0);
 }finally{unsubscribe();gate.resolve();await engine.close();}
});