Pin durable AI source text and requirements across retries

server/ai-job-store.mjs

@@ -14,6 +14,12 @@ function dbWork(fn) {
    result TEXT,error TEXT,run_id TEXT,heartbeat INTEGER,UNIQUE(email,request_id));
    CREATE INDEX IF NOT EXISTS job_owner ON ai_jobs(email,created_at);
    CREATE INDEX IF NOT EXISTS job_queue ON ai_jobs(state,created_at);`);
+  // Serialize schema inspection with migration across server processes.
+  db.exec('BEGIN IMMEDIATE');
+  try {
+   if(!db.prepare('PRAGMA table_info(ai_jobs)').all().some(column=>column.name==='resolved_input'))db.exec('ALTER TABLE ai_jobs ADD COLUMN resolved_input TEXT');
+   db.exec('COMMIT');
+  }catch(error){db.exec('ROLLBACK');throw error;}
   return fn(db);
  }finally {db.close();}
 }
@@ -39,7 +45,7 @@ export function createAiJob(email,requestId,operation,input) {
   } catch(error){db.exec('ROLLBACK');throw error;}
  });
 }
-export const getAiJob=(email,id)=>dbWork(db=>{const row=db.prepare('SELECT * FROM ai_jobs WHERE email=? AND id=?').get(email,id);return row?{...publicJob(row),input:JSON.parse(row.input)}:null;});
+export const getAiJob=(email,id)=>dbWork(db=>{const row=db.prepare('SELECT * FROM ai_jobs WHERE email=? AND id=?').get(email,id);return row?{...publicJob(row),input:JSON.parse(row.resolved_input || row.input)}:null;});
 export const listAiJobs=(email,ikn)=>dbWork(db=>db.prepare('SELECT * FROM ai_jobs WHERE email=? AND (? IS NULL OR ikn=?) ORDER BY created_at DESC LIMIT 50').all(email,ikn||null,ikn||null).map(publicJob));
 export function claimAiJob(clock=Date.now()) {
  return dbWork(db=>{
@@ -52,7 +58,7 @@ export function claimAiJob(clock=Date.now()) {
    if(!row){db.exec('COMMIT');return null;}
    const runId=randomUUID();
    db.prepare("UPDATE ai_jobs SET state='running',run_id=?,heartbeat=?,updated_at=? WHERE id=?").run(runId,clock,new Date(clock).toISOString(),row.id);
-   db.exec('COMMIT');return {id:row.id,email:row.email,operation:row.operation,input:JSON.parse(row.input),runId};
+   db.exec('COMMIT');return {id:row.id,email:row.email,operation:row.operation,input:JSON.parse(row.resolved_input || row.input),sourcePinned:row.resolved_input!==null,runId};
   }catch(error){db.exec('ROLLBACK');throw error;}
  });
 }
@@ -69,3 +75,7 @@ export const retryAiJob=(email,id)=>dbWork(db=>{
   db.exec('COMMIT');return changed;
  }catch(error){db.exec('ROLLBACK');throw error;}
 });
+
+// Only the current worker may pin source content; retries cannot replace it.
+export const pinAiJobInput=(id,runId,input)=>dbWork(db=>db.prepare("UPDATE ai_jobs SET resolved_input=?,updated_at=? WHERE id=? AND run_id=? AND state='running' AND resolved_input IS NULL")
+ .run(JSON.stringify(input),new Date().toISOString(),id,runId).changes>0);