fix: follow legacy review transfers without overwriting native payroll

docs/operations/legacy-review-sync-logic-release-2026-11-02.md

@@ -0,0 +1,40 @@
+# 复习老师切换同步规则上线 · 2026-11-02
+
+## 结果
+
+旧系统未完成复习任务的老师切换和完成状态现在由夜间直连及页面刷新共用的数据库函数处理；同步来源版本独立保存，连续多次旧系统切换均可跟进。新系统人工转交、完成和原始计薪凭据优先保留。
+
+费用在整批来源课次同步后自动核定：30 分钟 ¥0.50、60 分钟 ¥1.00，只为计划当天有效完成计一次薪；逾期和重复不增加收入。历史实际完成账号仍保留未知，使用此前确认的历史绑定老师口径；不伪造实际完成者。新系统老师操作与学生自习继续使用实际登录身份。
+
+身份、时长、日期或重复候选不能核清时保存待核定原因，后续批次重试，并在同步最终回执显示 `reviewIncome.pending/errors` 和保留冲突状态，不能将其当成全部费用已核清。没有放开工资单发放或结算就绪标记。
+
+## 变更
+
+- `legacy-review-sync.sql` 增加来源状态、审计和费用队列。老师归属与完成记录使用原有触发器、任务锁和乐观快照保护；不关闭触发器覆盖现有任务。
+- 来源状态与任务版本、完整快照绑定。真正的新系统转交会改变版本，使旧来源失去覆盖资格；存在任何原始收入凭据的任务均不能被旧同步改写。
+- 尚无基线的历史未完成任务，只在九个业务字段与旧接口实时来源一致、无人工转交和收入、且不存在已知基线冲突时初始化。已有本地差异不自动接管。
+- 已人工核定的两批来源修复按确切审计命令和未变化版本接续来源状态，不重置版本或删除转交审计。
+- 页面刷新与夜间直连共用 `xs_legacy_review_patch`；新增主附表由来源触发器跟踪，已有主表下补缺失附表由原子教学写入事务补登记，避免记录已补而费用未入队。
+- 夜间运行器在九组业务数据集全部完成后，按游标处理费用队列，保留批次幂等回执；最终服务端独立核对实际金额、待核定数和失败原因。
+- 重复副本先排除可验证的逾期原件；不能把有效当天完成记录永久排除到逾期、未来、缺时长或其他未核清候选之下。异常事务的统计计数与数据库回滚保持一致。
+- 来源基线读取按现有主表复合索引定位，不在匹配字段上做文本转换。
+
+## 发布
+
+增量安装数据库规则及原子教学写入钩子；选择性更新 `xiaoshu.app.gateway` 的复习拉取函数和 `xiaoshu.sync.direct-v1` 的处理函数，逐段比对现网旧函数后替换，未覆盖其他功能。两个网关及七个数据库函数回读与源码一致。
+
+独立目录 `~/Library/Application Support/XiaoshuSync/scripts/run-nightly-direct-sync.mjs` 已更新并核对源码，实际 `--check` 返回 `ready`。网关返回版本 `2026-11-02-review-source-v1`、能力 `source-owned-v1`，来源仍为 `old-api/WXAPP`、`bridgeUsed=false`。
+
+实时读取旧系统 204 页、203,595 条复习记录并校验编号唯一，为 41,492 条一致的历史未完成任务建立来源基线；接续前两批已核实来源转交 430 条，合计状态 41,922 条，待核定 0。初始化只登记同步状态及基线，不改变业务复习数据或费用。
+
+没有调整调度、额外执行全量同步或发放工资；新的完整夜间批次回执仍须在实际运行后验收，不将本次发布与连接检查当作全量同步成功。
+
+## 验证
+
+- 服务端全量回归 **378 项通过、0 失败**；最后的索引定位调整另有 14 项真实 PostgreSQL 策略回归通过，直连与运行器 18 项通过；云函数 11 个定义校验通过。
+- 独立只读审查发现并修复逾期/未来候选误排除有效副本、页面补附表漏入队及回滚计数问题，复审无剩余 Critical/Important 问题。
+- 生产真实数据的完整回滚事务中验证：连续两次旧来源换老师、导入完成、60 分钟自动核定 ¥1、原始凭据不变、重试新增收入 0、新系统人工转交及已有计薪任务拒绝覆盖。所有演练写入均已回滚。
+- 发布前后核对 11 张业务表的数量与内容摘要。期间恰有一次老师使用新系统正常完成复习，按真实账号产生 ¥1 收入；逐项核对其新增凭据和三个完成字段后，复习记录摘要与发布前吻合，排除该新增凭据后，原有 **158,312 条原始收入凭据完全一致**。
+- 已有 5,727 条历史核定、179 条费率纠正、2,218 条转交审计、4 条转交命令、1 份工资单及 892 条工资明细、全部账户和主表均与发布前完全一致。没有把并发正常业务变更误当成发布改写。
+
+私有发布证据：`/Users/mac/.local/state/xiaoshu/cloud-releases/2026-11-02-legacy-review-sync-fix/`。包含函数及数据库备份、选择性发布计划、实时来源、初始化结果、生产回滚演练、源码回读、并发业务核对、保护摘要和测试日志。账号身份、凭据和逐条生产数据不提交 Git。

docs/superpowers/plans/2026-11-02-legacy-review-sync-fix.md

@@ -0,0 +1,12 @@
+# 旧系统复习切换同步修复计划
+
+目标：已确认来源的旧系统老师切换和完成记录继续同步，按现有口径补费用；新系统转交、完成和计薪凭据优先。
+
+- [x] 用真实 PostgreSQL 触发器和现有网关复现切换被拦截、完成后零元停留问题。
+- [x] 增加独立来源版本、同步审计和待核定队列；不修改原始收入凭据，不猜课时时长。
+- [x] 直连与页面刷新共用安全更新函数，保留快照版本、来源身份及新系统转交保护。
+- [x] 所有来源课次同步后处理待核定费用；报告未核清原因，重复执行不重复计薪。
+- [x] 对已人工核对的两批来源修复建立来源版本，不把它们误判为新系统人工转交。
+- [x] 运行相关测试和服务端回归，选择性发布数据库及两个网关，更新独立运行器，线上回滚试验与只读回验后提交推送。
+
+重点边界：二次旧系统切换、新系统接手后旧数据到达、已计薪/学生自习、重复任务、缺少课次及日期、并发版本变化和未核定费用的报告。

scripts/cloud/legacy-direct-control.js

@@ -21,7 +21,7 @@ async function directOld(action,args={}) {
 }
 async function directControl(p,co,u) {
  const specs=__SPECS__;
- if(p.action==='status')return {version:'2026-10-14-finance-ledger-v3',source:'old-api/WXAPP',bridgeUsed:false,runs:await Psql.query('SELECT "businessDate","runId",status,"startedAt","finishedAt",summary FROM "LegacyDirectRun" WHERE company=$1 ORDER BY "startedAt" DESC LIMIT 3',[co])};
+ if(p.action==='status')return {version:'2026-11-02-review-source-v1',reviewSyncVersion:'source-owned-v1',source:'old-api/WXAPP',bridgeUsed:false,runs:await Psql.query('SELECT "businessDate","runId",status,"startedAt","finishedAt",summary FROM "LegacyDirectRun" WHERE company=$1 ORDER BY "startedAt" DESC LIMIT 3',[co])};
  if(p.action==='claim')return Psql.tx(async t=>{
   await t.any('SELECT pg_advisory_xact_lock(hashtext($1))',['direct-sync:'+co]);
   const today=(await t.one("SELECT to_char(now() AT TIME ZONE 'Asia/Shanghai','YYYY-MM-DD') AS business_day,to_char((now() AT TIME ZONE 'Asia/Shanghai')-interval '1 day','YYYY-MM-DD') AS yesterday"));
@@ -36,6 +36,14 @@ async function directControl(p,co,u) {
  });
  if(p.action==='finish'){const prior=await Psql.oneOrNone('SELECT status,summary FROM "LegacyDirectRun" WHERE company=$1 AND "runId"=$2 AND owner=$3',[co,p.runId,p.owner]);if(prior&&prior.status!=='running')return prior.summary;}
  await directLease(Psql,p,co);
+ if(p.action==='review-income')return Psql.tx(async t=>{
+  await directLease(t,p,co);
+  if(!/^review-income:[A-Za-z0-9:._-]{1,140}$/.test(p.batchKey||''))throw Error('Invalid review income batch');
+  const fingerprint=(await t.one('SELECT md5($1) value',[JSON.stringify(['review-income',String(p.afterTaskId||'')])])).value;
+  const prior=await t.oneOrNone('SELECT fingerprint,result FROM "LegacyDirectBatch" WHERE company=$1 AND "runId"=$2 AND key=$3',[co,p.runId,p.batchKey]);if(prior){if(prior.fingerprint!==fingerprint)throw Error('batch_key_reused');return prior.result;}
+  const result=(await t.one('SELECT xs_legacy_review_income($1,$2,100,$3) result',[co,u.id,String(p.afterTaskId||'')])).result;
+  await t.none('INSERT INTO "LegacyDirectBatch"(company,"runId",key,fingerprint,result) VALUES($1,$2,$3,$4,$5::jsonb)',[co,p.runId,p.batchKey,fingerprint,JSON.stringify({...result,modelId:0})]);return result;
+ });
  if(p.action==='finance-members')return Psql.query('SELECT u."legacyUserId" id FROM "_User" u WHERE u.company=$1 AND u."identityType"=\'member\' AND COALESCE(u."isDeleted",false)=false AND (u."sourceKey" LIKE \'legacy%\' OR u."sourceKey" LIKE \'[["UserID",%\') AND COALESCE(u."legacyUserData"->>\'SyncExcluded\',\'false\')<>\'true\' AND (SELECT count(*) FROM "LegacyFinanceReceipt" f WHERE f.company=u.company AND f."studentId"=u."legacyUserId" AND f."runId"=$2)<4 ORDER BY u."legacyUserId"',[co,p.runId]);
  if(p.action==='finance')return importLegacyFinance(co,Number(p.studentId),p.runId,args=>directOld('user_point_list',args));
  if(p.action==='source'){
@@ -44,6 +52,7 @@ async function directControl(p,co,u) {
   if(!Number.isSafeInteger(p.page)||p.page<1||p.page>10000)throw Error('invalid_page');
   const result=await directOld('content_list',{modelId:m,psize:p.countOnly?1:1000,cpage:p.page,...(m===2?{nid:3}:{})});
   if(!p.countOnly){const ids=result.items.map(x=>Number(x.GeneralID??x.generalId));if(ids.some(id=>!Number.isSafeInteger(id)||id<1))throw Error('invalid_source_id');
+   if(m===60)await Psql.one('SELECT xs_legacy_review_observe($1,$2::jsonb)',[co,JSON.stringify(result.items)]);
    await Psql.none('INSERT INTO "LegacyDirectSourcePage" (company,"runId",model,page,ids,total,checksum) VALUES($1,$2,$3,$4,$5::jsonb,$6,md5($7::jsonb::text)) ON CONFLICT(company,"runId",model,page) DO UPDATE SET ids=EXCLUDED.ids,total=EXCLUDED.total,checksum=EXCLUDED.checksum',[co,p.runId,m,p.page,JSON.stringify(ids),Number(result.page.itemCount??ids.length),JSON.stringify(result.items)]);
   }return result;
  }
@@ -93,6 +102,11 @@ async function directControl(p,co,u) {
    const mismatch=await t.one('SELECT count(*)::int n FROM "LegacyFinanceReceipt" r WHERE r.company=$1 AND r."runId"=$2 AND r.account=$3 AND (jsonb_array_length(r.ids)<>(SELECT count(*) FROM "'+table+'" l WHERE l.company=r.company AND l."userId"=r."studentId" AND l."legacyFinanceId" IN (SELECT jsonb_array_elements_text(r.ids))) OR r.checksum<>(SELECT md5(COALESCE(jsonb_agg(jsonb_build_object(\'id\',l."legacyFinanceId",\'hash\',l."legacyFinanceHash") ORDER BY l."legacyFinanceId"::numeric),\'[]\'::jsonb)::text) FROM "'+table+'" l WHERE l.company=r.company AND l."userId"=r."studentId" AND l."legacyFinanceId" IN (SELECT jsonb_array_elements_text(r.ids))))',[co,p.runId,account]);if(mismatch.n)throw Error('finance_postimage_changed');
   }
   const summary={...p.summary,finance:{members:members.length,streams:finance.length,rows:finance.reduce((n,f)=>n+f.ids.length,0),checksum:(await t.one('SELECT md5($1::jsonb::text) hash',[JSON.stringify(finance.sort((a,b)=>(a.studentId+":"+a.account).localeCompare(b.studentId+":"+b.account)))])).hash,reconciled:true},createdBusinessRecords:d.reduce((n,x)=>n+x.created,0),refreshedBusinessRecords:d.reduce((n,x)=>n+x.updated,0),completedAt:new Date().toISOString(),source:'old-api/WXAPP',bridgeUsed:false,continuousSync:false,settlementReady:false,nightlyAutomation:true};
+  const income=await t.one('SELECT COALESCE(sum((result->>\'earned\')::int),0)::int earned,COALESCE(sum((result->>\'excluded\')::int),0)::int excluded,COALESCE(sum((result->>\'amount\')::numeric),0) amount FROM "LegacyDirectBatch" WHERE company=$1 AND "runId"=$2 AND result->>\'modelId\'=\'0\'',[co,p.runId]);
+  const pending=await t.one('SELECT count(*)::int n FROM "LegacyReviewSyncState" WHERE company=$1 AND "pendingIncome"',[co]);
+  summary.reviewIncome={...income,amount:Number(income.amount),pending:pending.n,errors:await t.any('SELECT \"taskId\",\"lastError\" AS reason FROM \"LegacyReviewSyncState\" WHERE company=$1 AND \"pendingIncome\" ORDER BY \"taskId\" LIMIT 25',[co])};
+  summary.conflicts=d.reduce((n,x)=>n+Number(x.conflicts||0),0)+Number(summary.accounts?.conflicts||0)+pending.n;
+  summary.status=summary.conflicts?'completed-with-preserved-conflicts':'completed';
   await t.none('UPDATE "LegacyDirectRun" SET status=$3,summary=$4::jsonb,"finishedAt"=now(),"leaseUntil"=now() WHERE company=$1 AND "runId"=$2',[co,p.runId,summary.status,JSON.stringify(summary)]);
   await t.none('INSERT INTO "SysLog" ("objectId",company,"sourceKey",type1,type2,type3,detail,"createdAt","updatedAt") VALUES(substr(md5($1),1,20),$2,$1,\'migration\',\'legacy-direct-sync-completed\',\'nightly\',$3,now(),now())',['direct-nightly:'+p.runId,co,JSON.stringify({actor:u.id,action:'legacy-direct-sync-completed',after:summary})]);
   return summary;

scripts/cloud/legacy-direct-sync.js

@@ -2,7 +2,7 @@ async function handler(request,response){
 try{
  const u=request.user;if(!u)return response.status(401).json({success:false});await u.fetch({useMasterKey:true});if(u.get('isDisabled')||u.get('isDeleted')||u.get('adminRoleKey')!=='super-admin'||u.get('company')?.id!==__COMPANY_ID__)return response.status(403).json({success:false});
  const co=__COMPANY_ID__,p=request.body?.params||{},specs=__SPECS__,windows=__WINDOWS__,q=x=>'"'+x+'"',s=specs[p.modelId],m=Number(p.modelId);
- if(p.action==='finance-members'||p.action==='finance'||p.action==='status'||p.action==='claim'||p.action==='source'||p.action==='finish'||p.action==='conflicts'||p.action==='verify'||p.action==='supplement'||p.action==='evidence'||p.action==='reset-dataset')return response.json({success:true,data:await directControl(p,co,u)});
+ if(p.action==='review-income'||p.action==='finance-members'||p.action==='finance'||p.action==='status'||p.action==='claim'||p.action==='source'||p.action==='finish'||p.action==='conflicts'||p.action==='verify'||p.action==='supplement'||p.action==='evidence'||p.action==='reset-dataset')return response.json({success:true,data:await directControl(p,co,u)});
  await directLease(Psql,p,co);
  const legacy=r=>r&&typeof r.sourceKey==='string'&&(/\[\["(?:ID|GeneralID)",\d+\]\]/.test(r.sourceKey)||/^legacy:model:\d+:general:\d+$/.test(r.sourceKey));
  const unchanged=r=>legacy(r)&&!r.teachingManaged&&(r.createdAt===r.updatedAt||!r.updatedAt||Date.parse(r.updatedAt)<Date.parse('2026-08-01')||windows.some(([a,b])=>Date.parse(r.updatedAt)>=a&&Date.parse(r.updatedAt)<=b));
@@ -54,13 +54,14 @@ try{
     }
     if(m===2||m===52||cc.teachingManaged||aa.teachingManaged||m===54&&(aa.bindingId||aa.creditHoldStatus||aa.courseCategoryKey)||m===58&&aa.enrollmentStatus==='active'){out.conflicts.push({generalId:gid,reason:'保留新系统课程内容或教学凭据'});continue;}
     const fields=Object.keys(incoming.values||{});if(!fields.length||fields.some(f=>!s.fields.includes(f)))throw Error('Invalid patch fields');
-    if(m===60&&(Number(aa.assignmentVersion)>0||await t.oneOrNone('SELECT "objectId" FROM "ReviewReward" WHERE company=$1 AND "taskId"=$2',[co,aa.objectId])||fields.some(f=>['yhid','xxjlid','kywrq','plid'].includes(f)))){out.conflicts.push({generalId:gid,reason:'保留奖励凭据或复习身份'});continue;}
-    if(m===60)await t.none("SET LOCAL session_replication_role='replica'");
-    await t.none('UPDATE '+q(s.className)+' a SET '+fields.map(f=>q(f)+'=v.'+q(f)).join(',')+',"updatedAt"=now() FROM jsonb_populate_record(NULL::'+q(s.className)+',$3::jsonb) v WHERE a.company=$1 AND a."objectId"=$2',[co,aa.objectId,JSON.stringify(incoming.values)]);
-    if(m===60){await t.none("SET LOCAL session_replication_role='origin'");await t.any('SELECT xiaoshu_record_review_reward(m,true) FROM "MemoryPracticeRecord" m WHERE m.company=$1 AND m."objectId"=$2 AND m.fxzt::text=\'1\'',[co,aa.objectId]);}
+    if(m===60){
+     const applied=(await t.one('SELECT xs_legacy_review_patch($1,$2,$3::jsonb,$4::jsonb) result',[co,aa.objectId,JSON.stringify(incoming.values),JSON.stringify(aa)])).result;
+     if(!applied.applied){out.conflicts.push({generalId:gid,reason:applied.reason});continue;}
+    }else await t.none('UPDATE '+q(s.className)+' a SET '+fields.map(f=>q(f)+'=v.'+q(f)).join(',')+',"updatedAt"=now() FROM jsonb_populate_record(NULL::'+q(s.className)+',$3::jsonb) v WHERE a.company=$1 AND a."objectId"=$2',[co,aa.objectId,JSON.stringify(incoming.values)]);
     out.updated++;
    }
    const after=await t.oneOrNone(readSql,[co,m,[gid]]);if(!after?.common||!after.addon)throw Error('Atomic projection verification failed');
+   if(m===60&&p.action==='append')await t.one('SELECT xs_legacy_review_track($1,$2)',[co,after.addon.objectId]);
    if(p.action==='patch'){const expected=(await t.one('SELECT to_jsonb(v) expected FROM jsonb_populate_record(NULL::'+q(s.className)+',$1::jsonb) v',[JSON.stringify(incoming.values)])).expected;if(Object.keys(incoming.values).some(f=>JSON.stringify(expected[f])!==JSON.stringify(after.addon[f])))throw Error('Post-write field verification failed');}
    if(m===60&&after.reward&&!(Number(after.reward.amount)===0&&['historical','exception'].includes(after.reward.status)))throw Error('Historical reward verification failed');
    await directBaseline(t,co,m,gid,after);

scripts/deploy-direct-sync.mjs

@@ -13,7 +13,10 @@ const sessions=(await api('/classes/_Session?include=user&limit=1000&order=-upda
 const session=sessions.find(s=>s.sessionToken===token&&s.user?.adminRoleKey==='super-admin'&&!s.user.isDisabled&&s.user.company?.objectId===co);
 if(!session)throw Error('A valid scoped super-admin session is required');
 const path='xiaoshu/system/direct-deploy-'+randomUUID(),expiry=Date.now()+600000;
-const sql=readFileSync(new URL('./sql/legacy-direct-sync.sql',import.meta.url),'utf8');let helperId;
+const reviewSql=readFileSync(new URL('./sql/legacy-review-sync.sql',import.meta.url),'utf8').replace(/^BEGIN;\s*$/m,'').replace(/^COMMIT;\s*$/m,'');
+const atomicSql=readFileSync(new URL('./sql/mobile-teaching.sql',import.meta.url),'utf8').match(/CREATE OR REPLACE FUNCTION xs_mobile_teaching_write\([\s\S]*?\$\$;/)?.[0];
+if(!atomicSql)throw Error('Missing atomic teaching writer');
+const sql=[readFileSync(new URL('./sql/legacy-direct-sync.sql',import.meta.url),'utf8'),reviewSql,atomicSql].join('\n');let helperId;
 try{
  const code=`async function handler(request,response){if(!request.user||request.user.id!==${JSON.stringify(session.user.objectId)}||Date.now()>${expiry})return response.status(403).json({success:false});try{await Psql.tx(async t=>t.none(${JSON.stringify(sql)}));return response.json({success:true,data:await Psql.query(${JSON.stringify("SELECT table_name,column_name FROM information_schema.columns WHERE table_schema='public'")})});}catch(e){return response.status(500).json({success:false,message:e.message});}}`;
  new Function(code);helperId=(await api('/classes/Function','POST',{name:path,path,type:'standalone',enabled:true,code,params:[],paramList:[],respType:'json'})).objectId;

scripts/run-nightly-direct-sync.mjs

@@ -114,6 +114,20 @@ export async function runDirectSync(env=process.env,args=process.argv.slice(2)){
    }catch(e){if(!String(e.message).includes('source_shifted_restart_dataset')||attempt>=2)throw e;attempt++;await call({action:'reset-dataset',modelId:m});for(const file of await readdir(runRoot))if(file.startsWith(m+'-')&&file!==m+'-attempt.json')await rm(join(runRoot,file));await save(attemptPath,{attempt});console.log(JSON.stringify({modelId:m,restartForSourceChanges:attempt}));}
    }
   }
+  if(status.reviewSyncVersion==='source-owned-v1'){
+   report.reviewIncome={earned:0,excluded:0,amount:0,pending:0,errors:[]};
+   let afterTaskId='';
+   for(let batch=0;batch<1000;batch++){
+    const income=await call({action:'review-income',batchKey:'review-income:'+randomUUID(),afterTaskId},2);
+    report.reviewIncome.earned+=Number(income.earned||0);report.reviewIncome.excluded+=Number(income.excluded||0);report.reviewIncome.amount+=Number(income.amount||0);report.reviewIncome.pending=Number(income.pending||0);
+    report.reviewIncome.errors=(report.reviewIncome.errors||[]).concat(income.errors||[]).slice(0,25);
+    await save(join(runRoot,'review-income.json'),report.reviewIncome);
+    if(!income.hasMore)break;
+    if(!income.afterTaskId||income.afterTaskId===afterTaskId)throw Error('review_income_cursor_stalled');
+    afterTaskId=income.afterTaskId;
+    if(batch===999)throw Error('review_income_batch_limit');
+   }
+  }
   report.finance=report.finance||{members:[],created:0};
   const financeMembers=await call({action:'finance-members'},2);
   // Bound source requests while server-side account transactions retain their lock.
@@ -125,6 +139,7 @@ export async function runDirectSync(env=process.env,args=process.argv.slice(2)){
    const failed=batch.find(item=>item.status==='rejected');if(failed)throw failed.reason;
   }
   report.createdBusinessRecords=report.datasets.reduce((n,d)=>n+d.created,0);report.refreshedBusinessRecords=report.datasets.reduce((n,d)=>n+d.updated,0);report.accounts=report.supplement.accounts;report.nodes=report.supplement.nodes;report.conflicts=report.datasets.reduce((n,d)=>n+d.conflicts,0)+Number(report.accounts.conflicts||0);
+  report.conflicts+=Number(report.reviewIncome?.pending||0);
   report.status=report.conflicts?'completed-with-preserved-conflicts':'completed';report.allCapturedSourceIdsReconciled=true;
   Object.assign(report,await call({action:'finish',summary:report},2));
   return report;

scripts/sql/legacy-review-sync.sql

@@ -0,0 +1,185 @@
+-- Apply after legacy-direct-sync.sql and review-reconciliation.sql.
+-- Source-owned versions are separate from human transfers; original receipts stay immutable.
+BEGIN;
+SET LOCAL lock_timeout='10s';
+CREATE TABLE IF NOT EXISTS "LegacyReviewSyncState" (
+ company text NOT NULL,"taskId" text NOT NULL,"assignmentVersion" numeric NOT NULL,
+ fingerprint text NOT NULL,"pendingIncome" boolean NOT NULL DEFAULT false,
+ "lastError" text NOT NULL DEFAULT '',"updatedAt" timestamptz NOT NULL DEFAULT now(),
+ PRIMARY KEY(company,"taskId")
+);
+CREATE INDEX IF NOT EXISTS legacy_review_income_pending ON "LegacyReviewSyncState"(company,"taskId") WHERE "pendingIncome";
+CREATE TABLE IF NOT EXISTS "LegacyReviewSyncAudit" (
+ "objectId" text PRIMARY KEY,company text NOT NULL,"taskId" text NOT NULL,
+ before_data jsonb NOT NULL,after_data jsonb NOT NULL,"createdAt" timestamptz NOT NULL DEFAULT now()
+);
+REVOKE ALL ON "LegacyReviewSyncState","LegacyReviewSyncAudit" FROM PUBLIC;
+
+CREATE OR REPLACE FUNCTION xs_legacy_review_owned(co text, task_id text) RETURNS boolean LANGUAGE sql STABLE AS $$
+ SELECT COALESCE((COALESCE(m."assignmentVersion",0)=0 OR EXISTS(
+  SELECT 1 FROM "LegacyReviewSyncState" s WHERE s.company=m.company AND s."taskId"=m."objectId"
+  AND s."assignmentVersion"=m."assignmentVersion" AND s.fingerprint=md5(to_jsonb(m)::text)))
+  AND COALESCE(m."sourceKey",'') LIKE 'legacy:model:60:general:%'
+  AND m.fxzt::text<>'1' AND NOT EXISTS(SELECT 1 FROM "ReviewReward" r WHERE r.company=m.company AND r."taskId"=m."objectId"),false)
+ FROM "MemoryPracticeRecord" m WHERE m.company=co AND m."objectId"=task_id
+$$;
+
+CREATE OR REPLACE FUNCTION xs_legacy_review_track(co text, task_id text, expected_version numeric DEFAULT 0) RETURNS void LANGUAGE plpgsql AS $$
+DECLARE m jsonb; c jsonb; pending boolean;
+BEGIN
+ SELECT to_jsonb(t) INTO m FROM "MemoryPracticeRecord" t WHERE t.company=co AND t."objectId"=task_id FOR UPDATE;
+ SELECT to_jsonb(t) INTO c FROM "CommonModel" t WHERE t.company=co AND t."modelId"=60 AND t."itemId"=(m->>'id')::numeric;
+ IF COALESCE((m->>'assignmentVersion')::numeric,0)<>expected_version THEN RETURN; END IF;
+ IF m IS NULL OR c IS NULL OR m->>'sourceKey' NOT LIKE 'legacy:model:60:general:%' THEN RAISE EXCEPTION '旧复习来源不完整'; END IF;
+ SELECT EXISTS(SELECT 1 FROM "ReviewRewardIncome" r WHERE r.company=co AND r."taskId"=task_id AND r."incomeMonth">='2026-09' AND r.status IN ('historical','exception')) INTO pending;
+ INSERT INTO "LegacyReviewSyncState"(company,"taskId","assignmentVersion",fingerprint,"pendingIncome")
+ VALUES(co,task_id,COALESCE((m->>'assignmentVersion')::numeric,0),md5(m::text),pending)
+ ON CONFLICT(company,"taskId") DO UPDATE SET "assignmentVersion"=EXCLUDED."assignmentVersion",fingerprint=EXCLUDED.fingerprint,"pendingIncome"=EXCLUDED."pendingIncome","lastError"='',"updatedAt"=now();
+ INSERT INTO "LegacyDirectBaseline"(company,model,gid,fingerprint) VALUES(co,60,(c->>'generalId')::numeric,md5(jsonb_build_array(c,m)::text))
+ ON CONFLICT(company,model,gid) DO UPDATE SET fingerprint=EXCLUDED.fingerprint,"updatedAt"=now();
+END $$;
+
+-- Bootstrap only records whose complete business fields still match the live old source.
+-- Once a baseline exists, an unexplained local modification can never be re-adopted.
+CREATE OR REPLACE FUNCTION xs_legacy_review_observe(co text, source_rows jsonb) RETURNS integer LANGUAGE plpgsql AS $$
+DECLARE row_data record; n integer:=0;
+BEGIN
+ IF jsonb_typeof(source_rows)<>'array' OR jsonb_array_length(source_rows)>1000 THEN RAISE EXCEPTION '无效旧复习来源页'; END IF;
+ FOR row_data IN SELECT m."objectId" FROM jsonb_array_elements(source_rows) s
+  JOIN "CommonModel" c ON c.company=co AND c."modelId"=60 AND c."generalId"=(s->>'GeneralID')::numeric
+  JOIN "MemoryPracticeRecord" m ON m.company=c.company AND m.id=c."itemId"
+  LEFT JOIN "LegacyDirectBaseline" b ON b.company=c.company AND b.model=60 AND b.gid=c."generalId"
+  WHERE m.fxzt='0' AND COALESCE(m."assignmentVersion",0)=0 AND m."sourceKey" LIKE 'legacy:model:60:general:%'
+  AND NOT COALESCE(c."teachingManaged",false) AND COALESCE(c.status,99)<>-2
+  AND NOT EXISTS(SELECT 1 FROM "ReviewReward" r WHERE r.company=m.company AND r."taskId"=m."objectId")
+  AND NOT EXISTS(SELECT 1 FROM "LegacyReviewSyncState" t WHERE t.company=m.company AND t."taskId"=m."objectId")
+  AND (b.fingerprint IS NULL OR b.fingerprint=md5(jsonb_build_array(to_jsonb(c),to_jsonb(m))::text))
+  AND jsonb_build_array(m.yhid::text,m.plid::text,m.xxjlid::text,m."orderId"::text,m.kcid::text,COALESCE(m.fxzt,''),COALESCE(m.wcsj,''),COALESCE(m.kywrq,''),COALESCE(m.kywsj,''))=
+      jsonb_build_array(s->>'yhid',s->>'plid',s->>'xxjlid',s->>'orderId',s->>'kcid',COALESCE(s->>'fxzt',''),COALESCE(s->>'wcsj',''),COALESCE(s->>'kywrq',''),COALESCE(s->>'kywsj',''))
+  FOR UPDATE OF m,c
+ LOOP PERFORM xs_legacy_review_track(co,row_data."objectId");n:=n+1;END LOOP;
+ RETURN n;
+END $$;
+REVOKE ALL ON FUNCTION xs_legacy_review_observe(text,jsonb) FROM PUBLIC;
+
+CREATE OR REPLACE FUNCTION xs_legacy_review_patch(co text, task_id text, values_json jsonb, expected jsonb) RETURNS jsonb LANGUAGE plpgsql AS $$
+DECLARE m "MemoryPracticeRecord"; before_row jsonb; after_row jsonb; c jsonb; incoming jsonb; f text; baseline text;
+ prior text:=COALESCE(current_setting('xiaoshu.review_transfer',true),'');
+BEGIN
+ SELECT * INTO m FROM "MemoryPracticeRecord" WHERE company=co AND "objectId"=task_id FOR UPDATE;
+ IF NOT FOUND THEN RETURN jsonb_build_object('applied',false,'reason','复习任务不存在'); END IF;
+ before_row:=to_jsonb(m);
+ IF before_row IS DISTINCT FROM expected OR NOT COALESCE(xs_legacy_review_owned(co,task_id),false) THEN RETURN jsonb_build_object('applied',false,'reason','保留新系统转交、完成或计薪凭据'); END IF;
+ IF jsonb_typeof(values_json)<>'object' OR EXISTS(SELECT 1 FROM jsonb_object_keys(values_json) k WHERE k NOT IN ('plid','fxzt','wcsj')) THEN RETURN jsonb_build_object('applied',false,'reason','保留复习来源身份和计划'); END IF;
+ SELECT to_jsonb(t) INTO c FROM "CommonModel" t WHERE t.company=co AND t."modelId"=60 AND t."itemId"=m.id FOR UPDATE;
+ IF c IS NULL OR COALESCE((c->>'teachingManaged')::boolean,false) OR c->>'status'='-2' THEN RETURN jsonb_build_object('applied',false,'reason','来源主表受保护'); END IF;
+ SELECT fingerprint INTO baseline FROM "LegacyDirectBaseline" WHERE company=co AND model=60 AND gid=(c->>'generalId')::numeric;
+ IF (baseline IS NOT NULL AND baseline<>md5(jsonb_build_array(c,before_row)::text)) OR
+ (baseline IS NULL AND (m."createdAt" IS NULL OR c->>'createdAt' IS NULL OR m."createdAt" IS DISTINCT FROM m."updatedAt" OR c->>'createdAt' IS DISTINCT FROM c->>'updatedAt')) THEN
+  RETURN jsonb_build_object('applied',false,'reason','新系统已有修改或来源基线不可确认');
+ END IF;
+ incoming:=before_row||values_json;
+ IF incoming->>'fxzt' NOT IN ('0','1') OR NULLIF(incoming->>'plid','') IS NULL THEN RAISE EXCEPTION '无效复习来源'; END IF;
+ IF incoming->>'plid' IS DISTINCT FROM before_row->>'plid' THEN PERFORM xs_review_target(co,incoming->>'plid'); END IF;
+ IF incoming->>'fxzt'='1' AND xiaoshu_safe_shanghai_timestamp(incoming->>'wcsj') IS NULL THEN RETURN jsonb_build_object('applied',false,'reason','缺少旧系统完成时间'); END IF;
+ PERFORM set_config('xiaoshu.review_transfer','legacy-sync',true);
+ UPDATE "MemoryPracticeRecord" SET plid=(incoming->>'plid')::numeric,fxzt=incoming->>'fxzt',wcsj=COALESCE(incoming->>'wcsj',''),
+ "assignmentVersion"=COALESCE("assignmentVersion",0)+CASE WHEN plid::text IS DISTINCT FROM incoming->>'plid' THEN 1 ELSE 0 END,"updatedAt"=clock_timestamp()
+ WHERE company=co AND "objectId"=task_id RETURNING to_jsonb("MemoryPracticeRecord".*) INTO after_row;
+ PERFORM set_config('xiaoshu.review_transfer',prior,true);
+ INSERT INTO "LegacyReviewSyncAudit"("objectId",company,"taskId",before_data,after_data)
+ VALUES(md5(co||task_id||before_row::text||after_row::text),co,task_id,before_row,after_row) ON CONFLICT DO NOTHING;
+ PERFORM xs_legacy_review_track(co,task_id,(after_row->>'assignmentVersion')::numeric);
+ RETURN jsonb_build_object('applied',true);
+EXCEPTION WHEN OTHERS THEN PERFORM set_config('xiaoshu.review_transfer',prior,true);RAISE;
+END $$;
+
+CREATE OR REPLACE FUNCTION xs_legacy_review_income(co text, actor text, batch_limit integer DEFAULT 100, after_task text DEFAULT '') RETURNS jsonb LANGUAGE plpgsql AS $$
+DECLARE item record; m "MemoryPracticeRecord"; r "ReviewReward"; best record; expired record; outcome text; err text; n integer:=0; earned integer:=0; excluded integer:=0; amount numeric:=0; pending integer; errors jsonb:='[]'; cursor_id text:=after_task; more boolean; saved_n integer; saved_earned integer; saved_excluded integer; saved_amount numeric;
+BEGIN
+ PERFORM xs_review_manager(co,actor);
+ IF batch_limit NOT BETWEEN 1 AND 500 THEN RAISE EXCEPTION '无效复习核定批量'; END IF;
+ LOCK TABLE "MemoryPracticeRecord","ReviewReward" IN SHARE ROW EXCLUSIVE MODE;
+ PERFORM pg_advisory_xact_lock(hashtext('xs-payroll:'||co));
+ FOR item IN SELECT s.* FROM "LegacyReviewSyncState" s WHERE s.company=co AND s."pendingIncome" AND s."taskId">after_task ORDER BY s."taskId" LIMIT batch_limit FOR UPDATE LOOP
+  cursor_id:=item."taskId";saved_n:=n;saved_earned:=earned;saved_excluded:=excluded;saved_amount:=amount;
+  BEGIN
+   SELECT * INTO m FROM "MemoryPracticeRecord" WHERE company=co AND "objectId"=item."taskId" FOR UPDATE;
+   SELECT * INTO r FROM "ReviewReward" WHERE company=co AND "taskId"=item."taskId" FOR UPDATE;
+   IF m."objectId" IS NULL OR r."objectId" IS NULL OR item.fingerprint<>md5(to_jsonb(m)::text) OR COALESCE(m."assignmentVersion",0)<>item."assignmentVersion" THEN RAISE EXCEPTION '来源同步后任务已改变'; END IF;
+   IF EXISTS(SELECT 1 FROM "ReviewRewardIncome" WHERE company=co AND "objectId"=r."objectId" AND status IN ('earned','ineligible')) THEN
+    UPDATE "LegacyReviewSyncState" SET "pendingIncome"=false,"lastError"='',"updatedAt"=now() WHERE company=co AND "taskId"=item."taskId";n:=n+1;CONTINUE;
+   END IF;
+   IF r."incomeMonth"<'2026-09' OR r."completedAt">now() OR r."completionRole" IS DISTINCT FROM 'historical_unknown' OR NULLIF(r."completedBy",'') IS NOT NULL THEN RAISE EXCEPTION '不符合已确认的历史核定范围'; END IF;
+   IF EXISTS(SELECT 1 FROM "_User" u WHERE u.company=co AND u."legacyUserId"=r."coachId" AND concat_ws('',u."realName",u.nickname,u.username) LIKE '%自习%') THEN RAISE EXCEPTION '自习账号需单独核对，不能自动计薪'; END IF;
+   -- An overdue normal copy must not disqualify an on-time duplicate copy merely
+   -- because its raw receipt is historical while the valid copy is exception.
+   IF xiaoshu_review_eligibility(r."plannedDate",r."completedAt")='eligible' THEN
+    FOR expired IN SELECT v."objectId",v."taskId" FROM "ReviewRewardIncome" v
+     JOIN "ReviewReward" raw ON raw."objectId"=v."objectId" AND raw.company=v.company
+     JOIN "MemoryPracticeRecord" t ON t.company=v.company AND t."objectId"=v."taskId"
+     WHERE v.company=co AND t.yhid=m.yhid AND t.xxjlid=m.xxjlid AND left(t.kywrq::text,10)=left(m.kywrq::text,10)
+     AND v.status IN ('historical','exception') AND xiaoshu_review_eligibility(v."plannedDate",v."completedAt")='overdue'
+     AND raw."completionRole"='historical_unknown' AND NULLIF(raw."completedBy",'') IS NULL AND v."completedAt"<=now()
+     AND t."sourceKey" LIKE 'legacy:model:60:general:%'
+    LOOP
+     PERFORM xs_review_resolve(co,actor,jsonb_build_array(jsonb_build_object('rewardId',expired."objectId",'reasonCode','overdue')),'旧来源重复复习核对：先排除已核实逾期副本，保留计划当天完成副本的计薪资格。',true);
+     UPDATE "LegacyReviewSyncState" SET "pendingIncome"=false,"lastError"='',"updatedAt"=now() WHERE company=co AND "taskId"=expired."taskId";
+     excluded:=excluded+1;n:=n+1;
+    END LOOP;
+   END IF;
+   SELECT v.* INTO best FROM "ReviewRewardIncome" v JOIN "MemoryPracticeRecord" t ON t.company=v.company AND t."objectId"=v."taskId"
+    WHERE v.company=co AND t.yhid=m.yhid AND t.xxjlid=m.xxjlid AND left(t.kywrq::text,10)=left(m.kywrq::text,10) AND v.status IN ('earned','historical','exception')
+    ORDER BY (v.status='earned') DESC,(v.status='historical') DESC,v."completedAt",v."objectId" LIMIT 1;
+   IF xiaoshu_review_eligibility(r."plannedDate",r."completedAt")='eligible' AND best.status<>'earned' AND best."objectId"<>r."objectId" AND (
+    best."completedAt" IS NULL OR best."completedAt">now() OR xiaoshu_review_eligibility(best."plannedDate",best."completedAt")<>'eligible'
+    OR xs_review_duration(co,best."taskId") IS NULL
+    OR NOT EXISTS(SELECT 1 FROM "ReviewReward" raw WHERE raw.company=co AND raw."objectId"=best."objectId" AND raw."completionRole"='historical_unknown' AND NULLIF(raw."completedBy",'') IS NULL AND raw.exception IN ('','来源复习任务重复'))
+    OR EXISTS(SELECT 1 FROM "PayrollLine" l WHERE l.company=co AND l."reviewRewardId"=best."objectId" AND NOT COALESCE(l.excluded,false)))
+   THEN RAISE EXCEPTION '重复候选的完成日期或计薪依据需核对，不能剔除有效当天完成'; END IF;
+   IF xiaoshu_review_eligibility(r."plannedDate",r."completedAt")='overdue' THEN outcome:='overdue';
+   ELSIF best."objectId"<>r."objectId" THEN outcome:='duplicate_review';
+   ELSE outcome:='legacy_assignment_policy'; END IF;
+   PERFORM xs_review_resolve(co,actor,jsonb_build_array(jsonb_build_object('rewardId',r."objectId",'reasonCode',outcome,'canonicalRewardId',best."objectId")),
+    '旧系统来源自动核对：按已确认历史绑定老师、计划当天完成和来源课时时长口径；保留实际完成身份未知，排除逾期和重复。',true);
+   SELECT * INTO best FROM "ReviewRewardIncome" WHERE company=co AND "objectId"=r."objectId";
+   IF best.status='earned' THEN earned:=earned+1;amount:=amount+best.amount;ELSE excluded:=excluded+1;END IF;
+   UPDATE "LegacyReviewSyncState" SET "pendingIncome"=false,"lastError"='',"updatedAt"=now() WHERE company=co AND "taskId"=item."taskId";
+   n:=n+1;
+  EXCEPTION WHEN OTHERS THEN
+   GET STACKED DIAGNOSTICS err=MESSAGE_TEXT;
+   n:=saved_n;earned:=saved_earned;excluded:=saved_excluded;amount:=saved_amount;
+   UPDATE "LegacyReviewSyncState" SET "lastError"=err,"updatedAt"=clock_timestamp() WHERE company=co AND "taskId"=item."taskId";
+   IF jsonb_array_length(errors)<25 THEN errors:=errors||jsonb_build_array(jsonb_build_object('taskId',item."taskId",'reason',err));END IF;
+  END;
+ END LOOP;
+ SELECT count(*)::int INTO pending FROM "LegacyReviewSyncState" WHERE company=co AND "pendingIncome";
+ SELECT EXISTS(SELECT 1 FROM "LegacyReviewSyncState" WHERE company=co AND "pendingIncome" AND "taskId">cursor_id) INTO more;
+ RETURN jsonb_build_object('processed',n,'earned',earned,'excluded',excluded,'amount',amount,'pending',pending,'errors',errors,'afterTaskId',cursor_id,'hasMore',more);
+END $$;
+REVOKE ALL ON FUNCTION xs_legacy_review_owned(text,text),xs_legacy_review_track(text,text,numeric),xs_legacy_review_patch(text,text,jsonb,jsonb),xs_legacy_review_income(text,text,integer,text) FROM PUBLIC;
+CREATE OR REPLACE FUNCTION xs_legacy_review_common_insert() RETURNS trigger LANGUAGE plpgsql AS $$
+DECLARE task_id text;
+BEGIN
+ SELECT "objectId" INTO task_id FROM "MemoryPracticeRecord" WHERE company=NEW.company AND id=NEW."itemId" AND "sourceKey" LIKE 'legacy:model:60:general:%' AND COALESCE("assignmentVersion",0)=0;
+ IF task_id IS NOT NULL THEN PERFORM xs_legacy_review_track(NEW.company,task_id); END IF;
+ RETURN NEW;
+END $$;
+DROP TRIGGER IF EXISTS legacy_review_source_insert ON "CommonModel";
+CREATE TRIGGER legacy_review_source_insert AFTER INSERT ON "CommonModel" FOR EACH ROW WHEN (NEW."modelId"=60) EXECUTE FUNCTION xs_legacy_review_common_insert();
+REVOKE ALL ON FUNCTION xs_legacy_review_common_insert() FROM PUBLIC;
+-- The two audited historical repairs came from the old source, not a human transfer.
+-- Preserve their version numbers and audit rows, and only adopt an unchanged repair version.
+DO $$ DECLARE a record;
+BEGIN
+ IF to_regclass('"ReviewTransferCommand"') IS NOT NULL AND to_regclass('"ReviewTransferAudit"') IS NOT NULL THEN
+  FOR a IN SELECT m.company,m."objectId",m."assignmentVersion" FROM "MemoryPracticeRecord" m
+   JOIN "ReviewTransferAudit" t ON t.company=m.company AND t."taskId"=m."objectId" AND t."newVersion"=m."assignmentVersion" AND t."newCoachId"=m.plid::text
+   JOIN "ReviewTransferCommand" c ON c.company=t.company AND c."objectId"=t."commandId"
+   WHERE c.key IN ('repair-zeng-1761-review-2026-11-02','repair-legacy-review-transfers-2026-11-02')
+   AND m."sourceKey" LIKE 'legacy:model:60:general:%'
+   AND NOT EXISTS(SELECT 1 FROM "LegacyReviewSyncState" s WHERE s.company=m.company AND s."taskId"=m."objectId")
+  LOOP PERFORM xs_legacy_review_track(a.company,a."objectId",a."assignmentVersion"::numeric);END LOOP;
+ END IF;
+END $$;
+COMMIT;

scripts/sql/mobile-teaching.sql

@@ -9,7 +9,7 @@ CREATE OR REPLACE FUNCTION xs_mobile_teaching_write(p_company text,p_actor text,
 RETURNS jsonb LANGUAGE plpgsql AS $$
 DECLARE op jsonb; data jsonb; criteria jsonb; previous jsonb; saved jsonb; output jsonb := '[]';
   tab text; field text; columns_sql text; values_sql text; where_sql text; typ text; counter text;
-  existing jsonb; prior "MobileTeachingCommand"%ROWTYPE;
+  existing jsonb; prior "MobileTeachingCommand"%ROWTYPE; legacy_reviews text[]:='{}'; review_task text;
 BEGIN
   IF length(p_key)<10 OR length(p_key)>160 OR jsonb_array_length(p_operations)>105 THEN RAISE EXCEPTION 'Invalid teaching command'; END IF;
   -- Serialize the learner's commands as well as duplicate request keys; nested writes share one transaction.
@@ -63,8 +63,16 @@ BEGIN
     ELSE
       EXECUTE format('UPDATE %I t SET (%s)=(SELECT %s FROM jsonb_populate_record(NULL::%I,$1) r) WHERE t."objectId"=$2 RETURNING to_jsonb(t)',tab,columns_sql,values_sql,tab) INTO saved USING data,existing->>'objectId';
     END IF;
+    IF tab='MemoryPracticeRecord' AND existing IS NULL AND saved->>'sourceKey' LIKE 'legacy:model:60:general:%' THEN
+      legacy_reviews:=array_append(legacy_reviews,saved->>'objectId');
+    END IF;
     previous:=saved; output:=output||jsonb_build_array(saved);
   END LOOP;
+  -- Includes a missing addon repaired beneath an existing CommonModel. Track only
+  -- after the complete pair is stored, in this same transaction and retry receipt.
+  IF to_regprocedure('xs_legacy_review_track(text,text,numeric)') IS NOT NULL THEN
+    FOREACH review_task IN ARRAY legacy_reviews LOOP PERFORM xs_legacy_review_track(p_company,review_task);END LOOP;
+  END IF;
   INSERT INTO "MobileTeachingCommand" VALUES(p_company,p_actor,p_key,p_fingerprint,output,now());
   RETURN output;
 END $$;

scripts/tests/direct-nightly-sync.test.mjs

@@ -26,10 +26,13 @@ before(async()=>{
   const fields=legacyAddonFields[m]||(m==='52'?['sy','yb','lj','yp']:['content','synopsis','mp4','ico','author','source']);
   await db.exec('CREATE TABLE "'+s.className+'" ("objectId" text PRIMARY KEY,company text,id numeric,"sourceKey" text,"createdAt" timestamptz,"updatedAt" timestamptz,"teachingManaged" boolean,'+fields.map(f=>'"'+f+'" text').join(',')+')');
  }
- await db.exec('ALTER TABLE "MemoryPracticeRecord" ADD "rewardTaskKey" text, ADD "rewardDuplicate" boolean');
- const rewardSource=readFileSync(new URL('../sql/review-rewards.sql',import.meta.url),'utf8');await db.exec(rewardSource.match(/CREATE OR REPLACE FUNCTION xiaoshu_review_key[\s\S]*?\$\$;/)[0]);
+ await db.exec('ALTER TABLE "MemoryPracticeRecord" ADD "rewardTaskKey" text, ADD "rewardDuplicate" boolean, ADD "assignmentVersion" numeric DEFAULT 0');
+ const rewardSource=readFileSync(new URL('../sql/review-rewards.sql',import.meta.url),'utf8');await db.exec(rewardSource.match(/CREATE OR REPLACE FUNCTION xiaoshu_review_key[\s\S]*?\$\$;/)[0]);await db.exec(rewardSource.match(/CREATE OR REPLACE FUNCTION xiaoshu_safe_shanghai_timestamp[\s\S]*?\$\$;/)[0]);
  await db.exec('CREATE TABLE "ReviewReward" ("objectId" text,company text,"taskId" text,amount numeric,status text)');
  await db.exec('CREATE TABLE "LegacySyncConflict" ("objectId" text PRIMARY KEY,company text,"sourceKey" text,dataset text,"legacyKey" text,"conflictType" text,message text,resolved boolean,resolution text,"createdAt" timestamptz,"updatedAt" timestamptz)');
+ await db.exec('CREATE VIEW "ReviewRewardIncome" AS SELECT r.*,NULL::text AS "incomeMonth" FROM "ReviewReward" r');
+ await db.exec("CREATE FUNCTION xs_review_target(text,text) RETURNS void LANGUAGE plpgsql AS $$ BEGIN RETURN; END $$");
+ await db.exec(readFileSync(new URL('../sql/legacy-review-sync.sql',import.meta.url),'utf8'));
  const columns=await sql("SELECT table_name,column_name FROM information_schema.columns WHERE table_schema='public'");
  context={Psql,Parse:{Config:{get:async()=>({get:k=>({legacyScheduleApiUrl:'https://legacy.invalid/api',legacyScheduleApiId:'test',legacyScheduleApiKey:'test'}[k])})}},fetch:async()=>({ok:true,json:async()=>({retcode:0,result:[{GeneralID:101,kcid:'5',xxcs:1}],page:{itemCount:1,pageCount:1}})}),AbortSignal,Date,console};
  vm.createContext(context);vm.runInContext(buildDirectBundle(columns,[],'co'),context);
@@ -258,3 +261,14 @@ test('账号编号冲突时停止业务数据同步，保留阻塞报告',async(
   assert.equal(r.status,'blocked');assert.equal(sourceCalls,0);assert.match(JSON.stringify(r),/legacy_account_identity_conflict/);
  }finally{globalThis.fetch=realFetch;await rm(root,{recursive:true,force:true});}
 });
+
+test('旧来源未完成任务可跟进老师切换，后续完成不再因老师字段被拦截',async()=>{
+ await sql("UPDATE \"LegacyDirectRun\" SET status='running',\"leaseUntil\"=now()+interval '5 minutes' WHERE \"runId\"=$1",[runId]);
+ const first=await call({action:'append',modelId:60,batchKey:'review-transfer-302',rows:[{generalId:302,common:{objectId:'review-common302',status:99},addon:{objectId:'review-addon302',id:302,yhid:'20',xxjlid:'34',plid:'30',kywrq:'2026-10-20',fxzt:'0',wcsj:''}}]});
+ assert.equal(first.success,true,first.message);
+ const row=first.data.postimages[0];
+ const result=await call({action:'patch',modelId:60,batchKey:'review-transfer-302-patch',rows:[{generalId:302,commonId:row.common.objectId,addonId:row.addon.objectId,commonVersion:row.common.updatedAt,addonVersion:row.addon.updatedAt,values:{plid:'31'}}]});
+ assert.equal(result.success,true,result.message);
+ assert.equal(result.data.updated,1,'可信旧来源的老师切换应同步');
+ assert.equal((await sql('SELECT plid FROM "MemoryPracticeRecord" WHERE id=302'))[0].plid,'31');
+});

scripts/tests/legacy-review-pull.test.mjs

@@ -35,12 +35,13 @@ function contextFor(rows, initial = []) {
     },
     Psql: {
       async none(sql, args) { if(sql.includes('ReviewAssignmentConflict')) {conflicts.push(args);return;}throw Error(sql);},
-      async one(sql) {
+      async one(sql, args) {
+        if(sql.includes('xs_legacy_review_patch')) {const item=stored.find(r=>(r.taskObjectId||r.objectId)===args[1]);if(Number(item?.assignmentVersion)>0)return {result:{applied:false,reason:'protected'}};Object.assign(item,JSON.parse(args[2]));return {result:{applied:true}};}
         if (sql.includes('COUNT(*)') && sql.includes('MemoryPracticeRecord')) return { total: stored.length };
         throw new Error(`unexpected one sql: ${sql}`);
       },
       async query(sql) {
-        if (sql.includes('SELECT c."generalId"')) return stored;
+        if (sql.includes('SELECT c."generalId"')) return stored.map(r=>({generalId:r.generalId,itemId:r.itemId,taskSnapshot:{...r,objectId:r.taskObjectId||r.objectId}}));
         if (sql.includes('SELECT CAST("id" AS text)')) return stored.map((row) => ({ id: row.itemId }));
         throw new Error(`unexpected query sql: ${sql}`);
       },
@@ -56,6 +57,7 @@ function contextFor(rows, initial = []) {
       return operations.map((operation) => operation.data);
     },
   };
+  context.Psql.tx=async callback=>callback(context.Psql);
   vm.runInNewContext(`${source};this.normalize=legacyReviewRecord;this.sync=performLegacyReviewSync`, context);
   return { context, writes, conflicts };
 }

scripts/tests/legacy-review-sync-policy.test.mjs

@@ -0,0 +1,134 @@
+import {test,before,after} from 'node:test';
+import assert from 'node:assert/strict';
+import {readFileSync} from 'node:fs';
+import vm from 'node:vm';
+import {PGlite} from '@electric-sql/pglite';
+const db=new PGlite(),q=async(s,p=[])=>(await db.query(s,p)).rows,one=async(s,p=[])=>(await q(s,p))[0];
+const sql=n=>readFileSync(new URL('../sql/'+n+'.sql',import.meta.url),'utf8');
+let sequence=200;
+const snapshot=async id=>(await one('SELECT to_jsonb(m) value FROM "MemoryPracticeRecord" m WHERE "objectId"=$1',[id])).value;
+const income=async id=>one('SELECT * FROM "ReviewRewardIncome" WHERE "taskId"=$1',[id]);
+const patch=async(id,values,expected)=>(await one('SELECT xs_legacy_review_patch(\'c\',$1,$2::jsonb,$3::jsonb) result',[id,JSON.stringify(values),JSON.stringify(expected||await snapshot(id))])).result;
+const drain=async()=>{let cursor='',totals={earned:0,excluded:0,amount:0,pending:0};for(let i=0;i<100;i++){const d=(await one('SELECT xs_legacy_review_income(\'c\',\'manager\',100,$1) result',[cursor])).result;totals={earned:totals.earned+d.earned,excluded:totals.excluded+d.excluded,amount:totals.amount+Number(d.amount),pending:d.pending,errors:d.errors};if(!d.hasMore)return totals;cursor=d.afterTaskId;}throw Error('income cursor stalled');};
+async function task(name,{minutes=60,day='2026-09-05',duplicateOf=0}={}){
+ const id=++sequence,study=duplicateOf||id;
+ if(!duplicateOf){
+  await q(`INSERT INTO "CourseAppointment"("objectId",company,id,szyh,dslx,"durationMinutes") VALUES($1,'c',$2,'20','2',$3)`,[name+'-appointment',id,minutes]);
+  await q(`INSERT INTO "CommonModel"("objectId",company,"modelId","itemId","generalId") VALUES($1,'c',54,$2,$2),($3,'c',56,$2,$2)`,[name+'-appointment-common',id,name+'-study-common']);
+  await q(`INSERT INTO "DailyStudyRecord"("objectId",company,id,"userId",dsid) VALUES($1,'c',$2,20,$3)`,[name+'-study',id,String(id)]);
+ }
+ if(duplicateOf)await db.exec("SET session_replication_role='replica'");
+ try{await q(`INSERT INTO "MemoryPracticeRecord"("objectId",company,id,plid,yhid,xxjlid,"orderId",kywrq,kywsj,kcid,fxzt,wcsj,"sourceKey","createdAt","updatedAt","rewardDuplicate") VALUES($1,'c',$2,10,20,$3,$3,$4,$4,30,'0','',$5,'2026-09-01','2026-09-01',$6)`,[name,id,study,day,'legacy:model:60:general:'+id,Boolean(duplicateOf)]);}finally{if(duplicateOf)await db.exec("SET session_replication_role='origin'");}
+ await q(`INSERT INTO "CommonModel"("objectId",company,"modelId","itemId","generalId","sourceKey","createdAt","updatedAt") VALUES($1,'c',60,$2,$2,$3,'2026-09-01','2026-09-01')`,[name+'-common',id,'legacy:model:60:general:'+id]);return id;
+}
+before(async()=>{
+ await db.exec(readFileSync(new URL('./fixtures/lesson-records-schema.sql',import.meta.url),'utf8'));
+ const source=readFileSync(new URL('../deploy-admin-functions.mjs',import.meta.url),'utf8');
+ const schemas=vm.runInNewContext('('+source.slice(source.indexOf('async function ensurePayrollSchemas()')).match(/const schemas = (\{[\s\S]*?\n  });/)[1]+')');
+ for(const [name,fields] of Object.entries(schemas)){
+  if(name==='CommonModel')continue;await db.exec(`CREATE TABLE IF NOT EXISTS "${name}" ("objectId" text PRIMARY KEY,"createdAt" timestamptz,"updatedAt" timestamptz)`);
+  const existing=new Set((await q('SELECT column_name FROM information_schema.columns WHERE table_name=$1',[name])).map(r=>r.column_name));
+  for(const [key,v] of Object.entries(fields))if(!existing.has(key))await db.exec(`ALTER TABLE "${name}" ADD COLUMN "${key}" ${({Number:'double precision',Date:'timestamptz',Boolean:'boolean',Array:'jsonb',Object:'jsonb'})[v.type]||'text'}`);
+ }
+ await db.exec(`ALTER TABLE "CourseAppointment" ADD COLUMN szyh text,ADD COLUMN kcid text,ADD COLUMN pl text,ADD COLUMN fxpl text,ADD COLUMN dszt text,ADD COLUMN dslx text,ADD COLUMN jffs text,ADD COLUMN jssj text,ADD COLUMN kssj text,ADD COLUMN yysj text,ADD COLUMN bxrq text,ADD COLUMN sdsd text,ADD COLUMN plxm text,ADD COLUMN yykcid text;
+ CREATE TABLE "CourseBinding"("objectId" text PRIMARY KEY,company text,id numeric,yhid text,kcid text,"sourceKey" text);
+ CREATE TABLE "SysLog"("objectId" text PRIMARY KEY,company text,"sourceKey" text NOT NULL UNIQUE,"createdAt" timestamptz,"updatedAt" timestamptz,id numeric,cdate timestamptz,cname text,type1 text,type2 text,type3 text,clevel numeric,"cadminId" numeric,detail text,content1 text,content2 text);
+ INSERT INTO "_User"("objectId",company,"legacyUserId","legacyGroupId","identityType","adminRoleKey","legacyUserData") VALUES
+ ('manager','c',1,1,'member','ops-manager','{}'),('coach','c',10,3,'coach',null,'{}'),('student','c',20,1,'member',null,'{}'),('next-coach','c',11,3,'coach',null,'{}'),('third-coach','c',12,3,'coach',null,'{}');
+ INSERT INTO "Node" VALUES('c',30,'单词课');`);
+ for(const table of ['UserExpDomP','UserSIcon','UserExpHis','UserUserPoint'])await db.exec(`CREATE TABLE "${table}"("objectId" text PRIMARY KEY,company text,"sourceKey" text UNIQUE,"createdAt" timestamptz,"updatedAt" timestamptz,"userId" numeric,score numeric,"scoreBefore" numeric,"hisTime" timestamptz,operator numeric,"scoreType" numeric,detail text,remark text,"payMethod" text)`);
+ for(const n of ['member-enrollment','review-rewards','lesson-records','payroll-integrity','review-pay-policy','review-reconciliation','mobile-teaching','legacy-direct-sync','legacy-review-sync'])await db.exec(sql(n));
+});
+after(()=>db.close());
+test('旧系统连续切换老师都可同步，来源版本和基线不会误标为新系统人工转交',async()=>{
+ await task('follow');assert.equal((await patch('follow',{plid:11})).applied,true);assert.equal((await snapshot('follow')).assignmentVersion,1);
+ assert.equal((await patch('follow',{plid:12})).applied,true);assert.equal((await snapshot('follow')).plid,12);assert.equal((await snapshot('follow')).assignmentVersion,2);
+ assert.equal(Number((await one('SELECT count(*) n FROM "LegacyReviewSyncAudit" WHERE "taskId"=\'follow\'')).n),2);
+});
+test('新系统人工转交后，旧系统再次更换老师或完成任务均不能覆盖',async()=>{
+ const id=await task('manual');await patch('manual',{plid:11});
+ const expected=(await one("SELECT md5(xs_review_transfer_snapshot('c','20',$1,'12')::text) value",[String(id)])).value;
+ await q("SELECT xs_review_transfer('c','manager','20',$1,'12',$2,'新系统转交','manual-transfer-test')",[String(id),expected]);
+ const saved=await snapshot('manual');assert.equal((await patch('manual',{plid:10,fxzt:'1',wcsj:'2026-09-05 10:00'})).applied,false);assert.deepEqual(await snapshot('manual'),saved);assert.equal(await income('manual'),undefined);
+});
+test('30分钟和60分钟旧完成自动按来源核算，原始收入凭据不改写，重复处理不重复计薪',async()=>{
+ for(const [name,minutes,amount] of [['paid30',30,.5],['paid60',60,1]]){
+  await task(name,{minutes});await patch(name,{plid:11,fxzt:'1',wcsj:'2026-09-05 10:00'});const original=await one('SELECT * FROM "ReviewReward" WHERE "taskId"=$1',[name]);assert.equal(original.status,'historical');
+  const result=await drain();assert.equal(result.earned,1);assert.equal(result.amount,amount);assert.equal(Number((await income(name)).amount),amount);assert.deepEqual(await one('SELECT * FROM "ReviewReward" WHERE "taskId"=$1',[name]),original);
+  assert.equal((await drain()).earned,0);assert.equal((await patch(name,{plid:12})).applied,false);
+ }
+});
+test('新系统老师完成、学生自习和既有计薪的原始归属及完成时间不被旧同步覆盖',async()=>{
+ const today=new Date().toLocaleDateString('sv-SE',{timeZone:'Asia/Shanghai'});
+ for(const [name,actor] of [['live-earned','coach'],['self-study','student']]){
+  await task(name,{day:today});await q("SELECT xs_complete_assigned_review('c',$1,$2)",[name,actor]);const saved=await snapshot(name),reward=await income(name);
+  assert.equal((await patch(name,{plid:11,fxzt:'0',wcsj:''})).applied,false);assert.deepEqual(await snapshot(name),saved);assert.deepEqual(await income(name),reward);assert.equal((await drain()).earned,0);
+ }
+});
+test('重复任务保留两份完成记录，但同一学习批次和日期只计一次工资',async()=>{
+ const id=await task('canonical');await task('copy',{duplicateOf:id});
+ for(const name of ['canonical','copy'])await patch(name,{plid:11,fxzt:'1',wcsj:'2026-09-05 10:00'});
+ const result=await drain();assert.equal(result.earned,1);assert.equal(result.excluded,1);assert.equal(result.amount,1);assert.equal((await income('copy')).ineligibleReason,'duplicate_review');
+});
+test('课次未齐先保留待核定并返回原因，来源补齐后自动重试成功；逾期不计薪',async()=>{
+ await task('missing');await q('DELETE FROM "CourseAppointment" WHERE "objectId"=\'missing-appointment\'');await patch('missing',{fxzt:'1',wcsj:'2026-09-05 10:00'});
+ const pending=await drain();assert.equal(pending.pending,1);assert.equal(pending.earned,0);assert.match(pending.errors[0].reason,/时长/);
+ const m=await snapshot('missing');await q(`INSERT INTO "CourseAppointment"("objectId",company,id,szyh,dslx,"durationMinutes") VALUES('missing-appointment','c',$1,'20','2',30)`,[m.orderId]);assert.equal((await drain()).amount,.5);
+ await task('late');await patch('late',{fxzt:'1',wcsj:'2026-09-06 10:00'});const late=await drain();assert.equal(late.amount,0);assert.equal(late.excluded,1);assert.equal((await income('late')).ineligibleReason,'overdue');
+});
+test('陈旧快照、未确认的本地修改、来源身份变化和无权限计薪被拒绝',async()=>{
+ await task('stale');const old=await snapshot('stale');await patch('stale',{plid:11});assert.equal((await patch('stale',{plid:12},old)).applied,false);
+ await q('UPDATE "MemoryPracticeRecord" SET kywsj=\'人工修改\' WHERE "objectId"=\'stale\'');assert.equal((await patch('stale',{plid:12})).applied,false);
+ await task('identity');assert.equal((await patch('identity',{yhid:999})).applied,false);await assert.rejects(q("SELECT xs_legacy_review_income('c','student')"),/仅管理/);
+});
+test('页面刷新和夜间同步共用保护及来源基线，页面同步后仍可跟进下一次旧切换',async()=>{
+ const id=await task('page-follow'),current=await snapshot('page-follow');
+ const source=readFileSync(new URL('../deploy-admin-functions.mjs',import.meta.url),'utf8').match(/const LEGACY_REVIEW_PULLS=new Map\(\);[\s\S]*?\nconst APP_LIVE_SCOPES=/)[0].replace(/\nconst APP_LIVE_SCOPES=[\s\S]*$/,'');
+ const Psql={query:q,one,none:async(s,p)=>{await q(s,p);},tx:async fn=>{await db.exec('BEGIN');try{const result=await fn(Psql);await db.exec('COMMIT');return result;}catch(e){await db.exec('ROLLBACK');throw e;}}};
+ const context={Psql,Map,Date,JSON,String,Number,Math,Promise,URLSearchParams,AbortSignal,DEFAULT_COMPANY_ID:'c',number:(v,f=0)=>Number.isFinite(Number(v))?Number(v):f,firstValue:(...v)=>v.find(x=>x!==undefined&&x!==null&&x!==''),legacyLearningIso:()=>new Date('2026-09-01').toISOString(),appLegacyConfig:async()=>({url:'https://old.invalid',apiId:'id',apiKey:'key'}),fetch:async()=>({ok:true,text:async()=>JSON.stringify({retcode:1,result:[{...current,GeneralID:id,plid:11}],page:{itemCount:1,pageCount:1}})}),mobileAtomicWrite:async()=>{throw Error('Existing tasks must use the shared source updater');}};
+ vm.createContext(context);vm.runInContext(source+';this.sync=performLegacyReviewSync;',context);
+ const result=await context.sync(20,{id:'student'},true);assert.equal(result.written,1);assert.equal(result.conflicts,0);assert.equal((await snapshot('page-follow')).plid,11);
+ assert.equal((await patch('page-follow',{plid:12})).applied,true);
+});
+test('已核实来源修复可建立来源版本，迁移重跑不会接管其后的新系统人工转交',async()=>{
+ const id=await task('adopted');await q('DELETE FROM "LegacyReviewSyncState" WHERE "taskId"=\'adopted\'');
+ await q("SELECT set_config('xiaoshu.review_transfer','legacy-repair',false)");await q('UPDATE "MemoryPracticeRecord" SET plid=11,"assignmentVersion"=1 WHERE "objectId"=\'adopted\'');await q("SELECT set_config('xiaoshu.review_transfer','',false)");
+ await q(`INSERT INTO "ReviewTransferCommand"("objectId",company,"actorId",key,fingerprint,reason,result) VALUES('prior-repair','c','manager','repair-legacy-review-transfers-2026-11-02','fingerprint','已核实旧来源','{}')`);
+ await q(`INSERT INTO "ReviewTransferAudit"("objectId",company,"commandId","taskId","studentId","appointmentId","oldCoachId","newCoachId","oldVersion","newVersion") VALUES('prior-repair-audit','c','prior-repair','adopted','20',$1,'10','11',0,1)`,[String(id)]);
+ assert.equal((await patch('adopted',{plid:12})).applied,false);await db.exec(sql('legacy-review-sync'));assert.equal((await patch('adopted',{plid:12})).applied,true);
+ const expected=(await one("SELECT md5(xs_review_transfer_snapshot('c','20',$1,'11')::text) value",[String(id)])).value;
+ await q("SELECT xs_review_transfer('c','manager','20',$1,'11',$2,'人工接手','adopted-human-transfer')",[String(id),expected]);
+ await db.exec(sql('legacy-review-sync'));assert.equal((await patch('adopted',{plid:10})).applied,false);assert.equal((await snapshot('adopted')).plid,11);
+});
+test('历史未初始化任务只在完整业务字段与旧来源一致时建立基线，已知本地修改不被重新接管',async()=>{
+ const id=await task('bootstrap');const source={...await snapshot('bootstrap'),GeneralID:id};
+ await q('DELETE FROM "LegacyReviewSyncState" WHERE "taskId"=\'bootstrap\'');await q('DELETE FROM "LegacyDirectBaseline" WHERE gid=$1',[id]);
+ await q('UPDATE "MemoryPracticeRecord" SET "updatedAt"=\'2026-10-08\' WHERE "objectId"=\'bootstrap\'');
+ assert.equal((await one("SELECT xs_legacy_review_observe('c',$1::jsonb) n",[JSON.stringify([{...source,plid:12}])])).n,0);
+ assert.equal((await patch('bootstrap',{plid:11})).applied,false);
+ assert.equal((await one("SELECT xs_legacy_review_observe('c',$1::jsonb) n",[JSON.stringify([source])])).n,1);assert.equal((await patch('bootstrap',{plid:11})).applied,true);
+ const id2=await task('known-local');const source2={...await snapshot('known-local'),GeneralID:id2};await q('DELETE FROM "LegacyReviewSyncState" WHERE "taskId"=\'known-local\'');await q('UPDATE "MemoryPracticeRecord" SET "updatedAt"=\'2026-10-08\' WHERE "objectId"=\'known-local\'');
+ assert.equal((await one("SELECT xs_legacy_review_observe('c',$1::jsonb) n",[JSON.stringify([source2])])).n,0);assert.equal((await patch('known-local',{plid:11})).applied,false);
+});
+test('明确自习老师账号的旧来源只保留进度与核查原因，不自动产生老师工资',async()=>{
+ await q(`INSERT INTO "_User"("objectId",company,"legacyUserId","legacyGroupId","identityType",nickname,"legacyUserData") VALUES('study-coach','c',13,3,'coach','自习账号','{}')`);
+ await task('old-self');await patch('old-self',{plid:13,fxzt:'1',wcsj:'2026-09-05 10:00'});const result=await drain();assert.equal(result.earned,0);assert.equal(result.pending,1);assert.match(result.errors[0].reason,/自习/);assert.equal(Number((await income('old-self')).amount),0);
+});
+test('先到队列的当天完成副本不被逾期原件永久排除，最终只为有效当天完成计一次薪',async()=>{
+ const id=await task('z-overdue-original');await task('a-valid-duplicate',{duplicateOf:id});
+ await patch('z-overdue-original',{plid:11,fxzt:'1',wcsj:'2026-09-06 10:00'});await patch('a-valid-duplicate',{plid:11,fxzt:'1',wcsj:'2026-09-05 10:00'});
+ const result=await drain();assert.equal(result.earned,1);assert.equal(result.excluded,1);assert.equal(result.amount,1);assert.equal((await income('a-valid-duplicate')).status,'earned');assert.equal((await income('z-overdue-original')).ineligibleReason,'overdue');
+});
+test('页面修复已存在主表的缺失附表时，同一事务建立来源状态和费用队列',async()=>{
+ const id=await task('page-orphan'),old=await snapshot('page-orphan');await q('DELETE FROM "MemoryPracticeRecord" WHERE "objectId"=\'page-orphan\'');await q('DELETE FROM "LegacyReviewSyncState" WHERE "taskId"=\'page-orphan\'');
+ const source=readFileSync(new URL('../deploy-admin-functions.mjs',import.meta.url),'utf8').match(/const LEGACY_REVIEW_PULLS=new Map\(\);[\s\S]*?\nconst APP_LIVE_SCOPES=/)[0].replace(/\nconst APP_LIVE_SCOPES=[\s\S]*$/,'');let serial=0;
+ const context={Psql:{query:q,one},Map,Date,JSON,String,Number,Math,Promise,URLSearchParams,AbortSignal,DEFAULT_COMPANY_ID:'c',number:(v,f=0)=>Number.isFinite(Number(v))?Number(v):f,firstValue:(...v)=>v.find(x=>x!==undefined&&x!==null&&x!==''),randomObjectId:()=>('orphan-object-'+(++serial)),legacyLearningIso:()=>new Date('2026-09-01').toISOString(),appLegacyConfig:async()=>({url:'https://old.invalid',apiId:'id',apiKey:'key'}),fetch:async()=>({ok:true,text:async()=>JSON.stringify({retcode:1,result:[{...old,GeneralID:id,plid:11,fxzt:'1',wcsj:'2026-09-05 10:00'}],page:{itemCount:1,pageCount:1}})}),mobileAtomicWrite:async(input,current,operations)=>(await one("SELECT xs_mobile_teaching_write('c','20','orphan-page-repair','fixed-fingerprint',$1::jsonb) result",[JSON.stringify(operations)])).result};
+ vm.createContext(context);vm.runInContext(source+';this.sync=performLegacyReviewSync;',context);const result=await context.sync(20,{id:'student'},true);assert.equal(result.written,1);
+ const tracked=await one('SELECT * FROM "LegacyReviewSyncState" WHERE "taskId"=(SELECT "objectId" FROM "MemoryPracticeRecord" WHERE id=$1)',[id]);assert.equal(tracked.pendingIncome,true);const completed=await drain();assert.equal(completed.earned,1);assert.equal(completed.amount,1);
+});
+test('同一计划日的未来完成原件不能把已经有效完成的副本永久判为重复零元',async()=>{
+ const day=new Date().toLocaleDateString('sv-SE',{timeZone:'Asia/Shanghai'}),id=await task('z-future-original',{day});await task('a-future-valid-copy',{day,duplicateOf:id});
+ await patch('z-future-original',{plid:11,fxzt:'1',wcsj:day+' 23:59:59'});await patch('a-future-valid-copy',{plid:11,fxzt:'1',wcsj:day+' 00:00:01'});
+ const result=await drain();assert.equal(result.earned,0);assert.equal(result.excluded,0);assert.equal((await income('a-future-valid-copy')).status,'exception');assert.equal((await income('z-future-original')).status,'historical');
+ assert((await one('SELECT "pendingIncome" FROM "LegacyReviewSyncState" WHERE "taskId"=\'a-future-valid-copy\'')).pendingIncome);
+});

scripts/tests/legacy-review-sync-runner.test.mjs

@@ -0,0 +1,26 @@
+import test from 'node:test';import assert from 'node:assert/strict';
+import {mkdtemp,rm} from 'node:fs/promises';import {tmpdir} from 'node:os';import {join} from 'node:path';
+import {runDirectSync} from '../run-nightly-direct-sync.mjs';
+test('所有课次同步后分页核定工资，待核定原因进入回执，写入重试复用同一批次',async()=>{
+ const root=await mkdtemp(join(tmpdir(),'review-sync-runner-')),original=globalThis.fetch,order=[],receipts=new Map();let simulatedLostResponse=false,applied=0,final;
+ try{
+ globalThis.fetch=async(url,options)=>{const p=JSON.parse(options.body).params;order.push(p.action==='source'?p.modelId:p.action);let data;
+ if(p.action==='status')data={source:'old-api/WXAPP',bridgeUsed:false,reviewSyncVersion:'source-owned-v1'};
+ else if(p.action==='claim')data={runId:'review-sync-test-run',businessDate:'2026-11-02'};
+ else if(p.action==='supplement')data={accounts:{missing:0},nodes:{missing:0}};
+ else if(p.action==='source')data={items:[],page:{itemCount:0,pageCount:1}};
+ else if(p.action==='evidence')data={sourceChecksum:'d41d8cd98f00b204e9800998ecf8427e',targetChecksum:'d41d8cd98f00b204e9800998ecf8427e',reconciled:true};
+ else if(p.action==='review-income'){
+  if(!receipts.has(p.batchKey)){applied++;receipts.set(p.batchKey,p.afterTaskId?{processed:0,earned:0,excluded:0,amount:0,pending:1,errors:[{taskId:'z',reason:'来源课次时长仍不明确'}],hasMore:false,afterTaskId:'z'}:{processed:1,earned:1,excluded:0,amount:.5,pending:1,errors:[],hasMore:true,afterTaskId:'a'});}
+  data=receipts.get(p.batchKey);if(!simulatedLostResponse){simulatedLostResponse=true;throw Error('response lost after commit');}
+ }
+ else if(p.action==='finance-members'||p.action==='read')data=[];
+ else if(p.action==='finish'){final=p.summary;data=p.summary;}
+ else throw Error('Unexpected action '+p.action);
+ return {ok:true,json:async()=>({success:true,data})};};
+ const result=await runDirectSync({XIAOSHU_SYNC_REPORT_DIR:root,XIAOSHU_SYNC_OPERATOR_TOKEN:'test'});
+ assert.equal(result.status,'completed-with-preserved-conflicts',result.reason);assert.equal(applied,2);assert.equal(receipts.size,2);
+ assert.equal(final.reviewIncome.earned,1);assert.equal(final.reviewIncome.amount,.5);assert.equal(final.reviewIncome.pending,1);assert.equal(final.conflicts,1);assert.match(final.reviewIncome.errors[0].reason,/时长/);
+ assert(order.lastIndexOf(59)<order.indexOf('review-income'));assert(order.indexOf('review-income')<order.indexOf('finance-members'));
+ }finally{globalThis.fetch=original;await rm(root,{recursive:true,force:true});}
+});