async function directLease(db,p,co) {
 const r=await db.oneOrNone('UPDATE "LegacyDirectRun" SET "leaseUntil"=now()+interval \'5 minutes\' WHERE company=$1 AND "runId"=$2 AND owner=$3 AND status=\'running\' AND "leaseUntil">now() RETURNING "runId"',[co,p.runId,p.owner]);
 if(!r)throw Error('sync_lease_unavailable');
}
async function directBaseline(t,co,m,gid,row) {
 await t.none('INSERT INTO "LegacyDirectBaseline" (company,model,gid,fingerprint) VALUES($1,$2,$3,md5($4::jsonb::text)) ON CONFLICT(company,model,gid) DO UPDATE SET fingerprint=EXCLUDED.fingerprint,"updatedAt"=now()',[co,m,gid,JSON.stringify([row.common,row.addon])]);
}
async function directUnchanged(t,co,m,gid,c,a,fallback) {
 const r=await t.oneOrNone('SELECT fingerprint=md5($4::jsonb::text) matches FROM "LegacyDirectBaseline" WHERE company=$1 AND model=$2 AND gid=$3',[co,m,gid,JSON.stringify([c,a])]);
 return r?r.matches:fallback(c)&&fallback(a);
}
async function directOld(action,args={}) {
 const config=await Parse.Config.get({useMasterKey:true}),url=config.get('legacyScheduleApiUrl'),apiId=config.get('legacyScheduleApiId'),apiKey=config.get('legacyScheduleApiKey');
 if(!url||!apiId||!apiKey)throw Error('direct_api_not_configured');
 const query=Object.entries({action,...args,apiId,apiKey}).map(([k,v])=>encodeURIComponent(k)+'='+encodeURIComponent(String(v))).join('&');
 let j;
 try {const r=await fetch(url+(url.includes('?')?'&':'?')+query,{signal:AbortSignal.timeout(60000)});if(!r.ok)throw Error();j=await r.json();} catch {throw Error('legacy_read_failed');}
 if(Number(j.retcode)===-1)throw Error('legacy_read_rejected');
 const items=typeof j.result==='string'?JSON.parse(j.result):j.result;
 if(!Array.isArray(items))throw Error('legacy_invalid_result');return {items,page:j.page||{}};
}
async function directControl(p,co,u) {
 const specs=__SPECS__;
 if(p.action==='status')return {version:'2026-10-13-account-protection-v2',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"));
  const day=p.targetBusinessDate||today.business_day;if(![today.business_day,today.yesterday].includes(day))throw Error('invalid_business_date');
  const latest=await t.oneOrNone('SELECT * FROM "LegacyDirectRun" WHERE company=$1 AND status=\'running\' ORDER BY "startedAt" DESC LIMIT 1',[co]);
  if(latest&&latest.owner!==p.owner&&new Date(latest.leaseUntil).getTime()>Date.now())throw Error('another_sync_running');
  const existing=latest||await t.oneOrNone('SELECT * FROM "LegacyDirectRun" WHERE company=$1 AND "businessDate"=$2',[co,day]);
  if(existing&&existing.status!=='running')return {alreadyCompleted:true,...existing};
  if(!/^[a-zA-Z0-9-]{10,80}$/.test(p.owner||'')||!/^[a-zA-Z0-9-]{10,80}$/.test(p.proposedRunId||''))throw Error('invalid_run_identity');
  if(existing){await t.none('UPDATE "LegacyDirectRun" SET owner=$2,"leaseUntil"=now()+interval \'5 minutes\' WHERE "runId"=$1',[existing.runId,p.owner]);return {runId:existing.runId,businessDate:existing.businessDate,resumed:true};}
  await t.none('INSERT INTO "LegacyDirectRun" (company,"businessDate","runId",owner,"leaseUntil") VALUES($1,$2,$3,$4,now()+interval \'5 minutes\')',[co,day,p.proposedRunId,p.owner]);return {runId:p.proposedRunId,businessDate:day,resumed:false};
 });
 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==='source'){
  const m=Number(p.modelId);if(!specs[m])throw Error('invalid_model');
  if(p.id){if(!Number.isSafeInteger(p.id)||p.id<1)throw Error('invalid_id');return directOld('content_get',{id:p.id});}
  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');
   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;
 }
 if(p.action==='supplement')return directSupplement(p,co);
 if(p.action==='conflicts'){
  if(!Array.isArray(p.rows)||p.rows.length>100)throw Error('invalid_conflicts');
  return Psql.tx(async t=>{await directLease(t,p,co);for(const c of p.rows){if(!specs[Number(c.modelId)]||!Number.isSafeInteger(c.generalId))throw Error('invalid_conflict');
   const source='direct-api:'+c.modelId+':'+c.generalId;
   await t.none('INSERT INTO "LegacySyncConflict" ("objectId",company,"sourceKey",dataset,"legacyKey","conflictType",message,resolved,resolution,"createdAt","updatedAt") SELECT substr(md5($1),1,20),$2,$1,$3,$4,\'direct-api-preserve-new\',$5,false,\'保留新系统，待人工复核\',now(),now() WHERE NOT EXISTS(SELECT 1 FROM "LegacySyncConflict" WHERE company=$2 AND "sourceKey"=$1)',[source,co,String(c.modelId),String(c.generalId),JSON.stringify({fields:c.fields||[],reason:c.reason||'保留新系统'})]);
  }return {saved:p.rows.length};});
 }
 if(p.action==='verify'){
  const m=Number(p.modelId),s=specs[m];if(!s||!Array.isArray(p.ids)||p.ids.length>10000||p.ids.some(x=>!Number.isSafeInteger(x)||x<1))throw Error('invalid_verification');
  return Psql.query('SELECT c."generalId",count(*)::int common_rows,count(a."objectId")::int addon_rows FROM "CommonModel" c LEFT JOIN "'+s.className+'" a ON a.company=c.company AND a.id=c."itemId" WHERE c.company=$1 AND c."modelId"=$2 AND c."generalId"=ANY($3::numeric[]) GROUP BY c."generalId" ORDER BY c."generalId"',[co,m,p.ids]);
 }
 if(p.action==='evidence'){
  const m=Number(p.modelId),s=specs[m];if(!s)throw Error('invalid_evidence_model');
  const latest=await directOld('content_list',{modelId:m,psize:1,cpage:1,...(m===2?{nid:3}:{})});
  return Psql.tx(async t=>{
   await directLease(t,p,co);
   const pages=await t.any('SELECT page,ids,total,checksum FROM "LegacyDirectSourcePage" WHERE company=$1 AND "runId"=$2 AND model=$3 ORDER BY page',[co,p.runId,m]);
   const ids=[...new Set(pages.flatMap(r=>r.ids))].sort((a,b)=>a-b),total=Number(latest.page.itemCount??latest.items.length);
   if(!pages.length||ids.length!==total||pages.some((r,i)=>Number(r.page)!==i+1||Number(r.total)!==total)||pages.length!==Math.max(1,Math.ceil(total/1000)))throw Error('source_shifted_restart_dataset');
   const rows=await t.any('SELECT c."generalId",count(*)::int common_rows,count(a."objectId")::int addon_rows FROM "CommonModel" c LEFT JOIN "'+s.className+'" a ON a.company=c.company AND a.id=c."itemId" WHERE c.company=$1 AND c."modelId"=$2 AND c."generalId" IN (SELECT jsonb_array_elements_text($3::jsonb)::numeric) GROUP BY c."generalId" ORDER BY c."generalId"',[co,m,JSON.stringify(ids)]);
   if(rows.length!==ids.length||rows.some(r=>r.common_rows!==1||r.addon_rows!==1))throw Error('server_projection_reconciliation_failed');
   const checks=await t.one('SELECT md5($1) source,md5($2) target,md5($3) content',[ids.join('\n'),rows.map(r=>Number(r.generalId)).join('\n'),pages.map(r=>r.checksum).join('\n')]);
   if(checks.source!==checks.target)throw Error('server_id_checksum_failed');
   await t.none('INSERT INTO "LegacyDirectEvidence" (company,"runId",model,ids,"sourceCount","sourceChecksum","targetChecksum",conflicts,"contentChecksum") VALUES($1,$2,$3,$4::jsonb,$5,$6,$7,$8,$9) ON CONFLICT(company,"runId",model) DO UPDATE SET ids=EXCLUDED.ids,"sourceCount"=EXCLUDED."sourceCount","sourceChecksum"=EXCLUDED."sourceChecksum","targetChecksum"=EXCLUDED."targetChecksum",conflicts=EXCLUDED.conflicts,"contentChecksum"=EXCLUDED."contentChecksum"',[co,p.runId,m,JSON.stringify(ids),ids.length,checks.source,checks.target,Number(p.conflicts||0),checks.content]);
   return {modelId:m,sourceCount:ids.length,sourceChecksum:checks.source,targetChecksum:checks.target,sourcePageContentChecksum:checks.content,reconciled:true};
  });
 }
 if(p.action==='reset-dataset'){if(!specs[Number(p.modelId)])throw Error('invalid_model');await Psql.none('DELETE FROM "LegacyDirectSourcePage" WHERE company=$1 AND "runId"=$2 AND model=$3',[co,p.runId,p.modelId]);await Psql.none('DELETE FROM "LegacyDirectEvidence" WHERE company=$1 AND "runId"=$2 AND model=$3',[co,p.runId,p.modelId]);return {reset:true};}
 if(p.action==='finish')return Psql.tx(async t=>{
  await directLease(t,p,co);
  const expected=[2,52,58,56,53,60,54,59,61],d=p.summary?.datasets;
  if(!Array.isArray(d)||d.length!==expected.length||!expected.every(m=>d.some(x=>x.modelId===m&&x.reconciled&&x.failed===0)))throw Error('reconciliation_incomplete');
  // Recompute ID coverage on the server, using the actual run's captured IDs, before a success receipt.
  const evidence=await t.any('SELECT model,ids,"sourceCount","sourceChecksum","targetChecksum" FROM "LegacyDirectEvidence" WHERE company=$1 AND "runId"=$2',[co,p.runId]);
  if(evidence.length!==expected.length)throw Error('server_reconciliation_missing');
  for(const e of evidence){const s=specs[e.model],r=await t.one('SELECT count(*)::int n FROM (SELECT c."generalId" FROM "CommonModel" c JOIN "'+s.className+'" a ON a.company=c.company AND a.id=c."itemId" WHERE c.company=$1 AND c."modelId"=$2 AND c."generalId" IN (SELECT jsonb_array_elements_text($3::jsonb)::numeric) GROUP BY c."generalId" HAVING count(*)=1) x',[co,e.model,JSON.stringify(e.ids)]);if(r.n!==Number(e.sourceCount)||e.sourceChecksum!==e.targetChecksum)throw Error('server_id_reconciliation_failed');}
  const totals=await t.any('SELECT (result->>\'modelId\')::int model,COALESCE(sum((result->>\'createdCommon\')::int),0)::int created,COALESCE(sum((result->>\'createdAddon\')::int),0)::int addon,COALESCE(sum((result->>\'updated\')::int),0)::int updated FROM "LegacyDirectBatch" WHERE company=$1 AND "runId"=$2 GROUP BY (result->>\'modelId\')::int',[co,p.runId]);
  for(const x of d){const total=totals.find(t=>t.model===x.modelId);x.created=total?.created||0;x.createdAddon=total?.addon||0;x.updated=total?.updated||0;}
  const summary={...p.summary,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};
  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;
 });
 throw Error('invalid_control_action');
}