Fence asynchronous import preparation before synchronous publication

headless/README.md

@@ -355,3 +355,11 @@ same indexed bytes. Document jobs use recovery v15 (readers accept 1–15, rejec
 report `import_resource_authorization_required`; other records can import with an
 explicit partial result. This is not complete asset round-trip support. Preparation
 still scans the envelope before progressive payload decoding and insertion.
+
+ImportRunner hosts may provide async `beforeCommit(units, job, signal)` for read-only
+preparation such as resource permission checks. It returns Availability; a denial
+uses normal independent-group failure isolation. A successful result still must
+pass cancellation/hold/generation/status checks before the synchronous commit.
+No document, history or committed source progress advances while it waits. This
+hook alone does not authorize or adopt exported image references; that integration
+is still pending.

src/lib/canvas-runtime/imports.ts

@@ -1,6 +1,6 @@
 import {validateImportFileRetention,type FileRetirement,type PreviousImportFile} from './import-file-retention.ts';
 import { validateDuplicateCheckpoint, type DuplicateCheckpoint } from '../canvas-core/duplicate-checkpoint.ts';
-import type { Change, Id, Json, Result, Target, Point } from '../canvas-core/types.ts';
+import type { Change, Id, Json, Result, Target, Point, Availability } from '../canvas-core/types.ts';
 import { selectImportGroups } from './import-groups.ts';
 import { clone, equal, fault, freeze, validJson } from '../canvas-core/values.ts';
 
@@ -104,6 +104,8 @@ export type ImportRunnerPorts={
   prepare(line:ImportLine,job:ImportJob,reserve:(sourceKey:string)=>Id):ImportUnit;
   /** Publish all document changes AND next job in the same StateRoot, or publish nothing. */
   commit(units:readonly ImportUnit[],next:ImportJob,expectedGeneration:number):Result;
+  /** Optional async, read-only preparation; commit remains synchronous and generation-fenced. */
+  beforeCommit?(units:readonly ImportUnit[],job:ImportJob,signal:AbortSignal):Promise<Availability>;
   persist():Promise<void>;onError(error:unknown):void;
   /** Host must durably gate every network write on the same document/job cut. */
   optimisticLocal?:boolean;
@@ -144,6 +146,13 @@ export function createImportRunner(ports:ImportRunnerPorts){
   function commit(units:readonly ImportUnit[],job:ImportJob,value:ImportJob){const start=now(),result=ports.commit(units,value,job.generation),cost=Math.max(.01,(now()-start)/units.length);costPerUnit=costPerUnit===undefined?cost:Math.max(cost,costPerUnit*.8+cost*.2);if(!['applied','rejected'].includes(result?.status)||!equal(current(),result.status==='applied'?value:job))throw fault('invalid_import_commit');return result;}
   async function pump(signal:AbortSignal){
     let frame=now(),didWork=false;
+    async function attempt(units:readonly ImportUnit[],job:ImportJob,proposed:ImportJob):Promise<Result|undefined>{
+      let availability:Availability={enabled:true};
+      if(ports.beforeCommit)try{availability=await ports.beforeCommit(units,job,signal);}catch(error){availability={enabled:false,code:(error as {code?:string}).code??'import_preparation_failed',targets:[]};}
+      if(closed||signal.aborted||holds.size||current().generation!==job.generation||current().status!=='running')return;
+      if(!availability.enabled)return{status:'rejected',code:availability.code??'import_preparation_failed',targets:availability.targets??[]};
+      return commit(units,job,proposed);
+    }
     type ReadResult={kind:'line';line:ImportLine|null}|{kind:'error';error:unknown};
     let reading:{cursor:number;result:Promise<ReadResult>}|undefined;
     const decoded=new Map<string,{unit:ImportUnit;digest:string;bytes:number}>();let decodedBytes=0;
@@ -181,7 +190,7 @@ export function createImportRunner(ports:ImportRunnerPorts){
           units.push(value.unit);
         }
         let proposed=completed(job,units,job.pending.filter(unit=>!chosen.includes(unit)));
-        let result=commit(units,job,proposed);
+        let result=await attempt(units,job,proposed);if(!result)return;
         if(result.status==='rejected'){
           // Reject a reference cycle as a whole; isolate only independent SCCs.
           const byKey=new Map(units.map(unit=>[unit.key,unit]));
@@ -189,7 +198,7 @@ export function createImportRunner(ports:ImportRunnerPorts){
             job=current();const internal=new Set(group.map(row=>row.key)),members=group.map(row=>byKey.get(row.key)!);
             if(!members.every(unit=>unit.dependencies.every(key=>internal.has(key)||job.committedSourceKeys.includes(key))))continue;
             proposed=completed(job,members,job.pending.filter(value=>!internal.has(value.key)));
-            result=commit(members,job,proposed);
+            result=await attempt(members,job,proposed);if(!result)return;
             if(result.status==='rejected'){const code=result.code;publish(job,{pending:job.pending.filter(value=>!internal.has(value.key)),failures:[...job.failures,...group.map(row=>({key:row.key,cursor:row.cursor,code}))]});}
             await checkpoint();if(closed||signal.aborted||current().status!=='running')return;
             if(now()-frame>=budget){await(ports.yieldTask??yieldTask)();frame=now();}

tests/headless/imports.test.mjs

@@ -137,3 +137,7 @@ test('resuming a failed partial-status write does not duplicate unresolved depen
 test('a ready parser publication observes host backpressure before admitting its content root',async()=>{
  const gate=deferred(),entered=deferred();let blocked=true;const f=fixture([{id:'one'},{id:'two'}],{ports:{optimisticLocal:true,maxBatch:1,awaitDurabilityBudget(stage){if(blocked){assert.equal(stage,'parsed');entered.resolve();return gate.promise;}}}});const running=f.runner.start();await entered.promise;assert.equal(f.root.job.pending.length,1);assert.equal(f.root.job.committedSourceKeys.length,0);assert.equal(f.root.document.elements.size,0);await ticks();assert.equal(f.root.document.elements.size,0,'ready content bypassed host budget');blocked=false;gate.resolve();await running;assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,2);assert.equal(f.errors.length,0);await f.runner.close();
 });
+
+test('async commit preparation cannot publish while waiting and cancellation fences a late successful result',async()=>{const gate=deferred(),entered=deferred(),f=fixture([{id:'image'}],{ports:{async beforeCommit(units,job,signal){assert.equal(f.root.document.elements.size,0);assert.ok(job.pending.length);entered.resolve();await gate.promise;return{enabled:true};}}});const running=f.runner.start();await entered.promise;assert.equal(f.commits.length,0);assert.equal(f.root.job.committedSourceKeys.length,0);await f.runner.cancel();gate.resolve();await running;await f.runner.settle();assert.equal(f.root.job.status,'cancelled');assert.equal(f.commits.length,0);assert.equal(f.root.document.elements.size,0);await f.runner.close();});
+test('async preparation denial isolates independent groups without publishing the denied record',async()=>{const prepared=[],f=fixture([{id:'denied'},{id:'allowed'}],{ports:{async beforeCommit(units){prepared.push(units.map(unit=>unit.key));return units.some(unit=>unit.key==='denied')?{enabled:false,code:'asset_forbidden',targets:[]}:{enabled:true};}}});await f.runner.start();assert.equal(f.root.job.status,'partial');assert.ok(f.root.job.failures.some(row=>row.key==='denied'&&row.code==='asset_forbidden'));assert.equal(f.root.document.elements.size,1);assert.equal([...f.root.document.elements.values()][0].data.title,'allowed');assert.ok(prepared.length>=2);assert.equal(f.errors.length,0);await f.runner.close();});
+test('a generation change during async preparation prevents the old attempt from committing or changing progress',async()=>{const gate=deferred(),entered=deferred(),f=fixture([{id:'later'}],{ports:{async beforeCommit(){entered.resolve();await gate.promise;return{enabled:true};}}});const running=f.runner.start();await entered.promise;f.root={...f.root,job:{...f.root.job,generation:f.root.job.generation+1}};const generation=f.root.job.generation;gate.resolve();await running;assert.equal(f.commits.length,0);assert.equal(f.root.job.generation,generation);assert.equal(f.root.job.committedSourceKeys.length,0);await f.runner.start();assert.equal(f.root.job.status,'completed');assert.equal(f.root.document.elements.size,1);await f.runner.close();});