"""Guarded separation is an additional correction, never a history rollback."""
import hashlib,json,sqlite3,uuid,zipfile
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path
import pytest
from fastapi.testclient import TestClient
from backend import recovery
from backend.app import create_app
from test_boundaries import app,client,write,job,portal,ORIGIN
from test_customer_reviews import complex_history,customer,selection,preview,confirm,ROOT
from test_customer_moves import move,check
from test_customers import work
from test_recovery import table_rows,new_login,rewrite_archive
from test_workflow import edit


def merged(application,c,complex=False):
    a=complex_history(application,c) if complex else {'job':job(c)}
    j=a['job'];second=work(c,customer(c,j['customer_id'])).json();b=job(c,name='Blake Fictional')
    p=preview(c,selection(c,j,b,name='source'));r=confirm(c,p);assert r.status_code==200,r.text
    return {**a,'second':second,'target':b,'merge':r.json(),'original_plan':p}


def context(c,tid):
    r=c.get(ROOT+'/merges/'+tid+'/correction');assert r.status_code==200,r.text;return r.json()


def separation(c,tid,ids,**fields):
    x=context(c,tid)
    body={'merge_id':tid,'alias_id':x['alias']['id'],'retained_id':x['retained']['id'],
          'alias_version':x['alias']['version'],'retained_version':x['retained']['version'],
          'job_ids':ids,'fields':{k:fields.get(k,'keep_current') for k in ('name','email','phone','address')}}
    r=write(c,ROOT+'/merge-correction-preview',body);assert r.status_code==200,r.text;return r.json()


def correct(c,p,key=None,**extra):
    return write(c,ROOT+'/merge-corrections',{**p['selection'],'preview_hash':p['preview_hash'],
                 'reason':'Reviewed mistaken merge of two invented people','confirmed':True,**extra},key=key)


@pytest.mark.parametrize('scope',['all','subset','none'])
def test_separation_preserves_original_and_later_work_and_closed_access(app,tmp_path,scope):
    c=client(app);a=merged(app,c,True);j=a['job'];b=a['target'];extra=work(c,customer(c,b['customer_id'])).json()
    edit(c,j,notes='A later saved work detail must remain')
    from test_portal_access import enabled,PASSWORD
    new_portal,_,_=enabled(app,c,b)
    ids=[j['id'],a['second']['id']] if scope=='all' else [j['id']] if scope=='subset' else []
    p=separation(c,a['merge']['id'],ids,name='restore_before_merge');assert p['can_confirm'],p['errors']
    before=table_rows(app.state.store.root,'oak');stats=c.get('/api/reports/operations?start=2026-01-01&end=2026-12-31').json()
    r=correct(c,p);assert r.status_code==200,r.text;check(app)
    assert customer(c,j['customer_id'])['state']=='active' and customer(c,b['customer_id'])['name']=='Blake Fictional'
    assert c.get('/api/jobs/'+j['id']).json()['notes']=='A later saved work detail must remain'
    for old in (j,a['second'],b,extra):
        assert c.get('/api/jobs/'+old['id']).json()['customer_id']==(j['customer_id'] if old['id'] in ids else b['customer_id'])
    after=table_rows(app.state.store.root,'oak')
    assert next(t for t in after['customer_transfers'] if t['id']==a['merge']['id'])==before['customer_transfers'][0]
    assert all(m in after['customer_transfer_jobs'] for m in before['customer_transfer_jobs'])
    for t in ('payments','payment_details','payment_stages','payment_allocations','quote_responses','time_entries','field_intervals','field_time_links','feedback_responses','enquiry_intakes','enquiry_photo_sources'):
        assert before[t]==after[t],t
    for table in ('quotes','invoices'):
        for old in before[table]:
            current=next(x for x in after[table] if x['id']==old['id'])
            for field in ('snapshot','reference','total_pence','created_at'):assert current[field]==old[field]
    assert new_portal.get('/api/portal/oak-and-vale/me').status_code==401
    assert a['portal'].post('/api/portal/oak-and-vale/access/login',json={'portal_id':a['portal_id'],'password':PASSWORD}).status_code==401
    assert a['portal'].post('/api/portal/oak-and-vale/access/complete',json={'code':a['setup'],'password':PASSWORD}).status_code==401
    assert write(TestClient(app,headers=ORIGIN),'/api/feedback/oak-and-vale/access',{'code':a['code']}).status_code==401
    assert c.get('/api/public/oak-and-vale/testimonials').json()['count']==0
    assert context(c,a['merge']['id'])['already_corrected']==r.json()
    assert not separation(c,a['merge']['id'],[])['can_confirm']
    h=c.get('/api/customers/'+j['customer_id']+'/history').json();assert h['merge_corrections'][0]['result']==r.json()
    stats_after=c.get('/api/reports/operations?start=2026-01-01&end=2026-12-31').json()
    assert stats['period_money']==stats_after['period_money'] and stats['period_counts']==stats_after['period_counts']
    saved=recovery.backup(app.state.store.root,tmp_path/'separated.zip');assert recovery.verify(saved['archive'],saved['sha256'])['valid']
    dest=tmp_path/'recovered';recovery.restore(saved['archive'],saved['sha256'],dest,confirm_fresh=True);check(create_app(dest))


def test_job_now_elsewhere_blocks_whole_selection_but_explicit_subset_keeps_it(app):
    c=client(app);a=merged(app,c);third=job(c);move(c,a['second'],third)
    p=separation(c,a['merge']['id'],[a['job']['id'],a['second']['id']]);assert not p['can_confirm']
    before=table_rows(app.state.store.root,'oak');assert correct(c,p).status_code==409;assert before==table_rows(app.state.store.root,'oak')
    narrowed=separation(c,a['merge']['id'],[a['job']['id']]);assert narrowed['can_confirm'] and narrowed['scope']=='selected_original_jobs'
    assert len(narrowed['context']['original_jobs'])==2 and len(next(x for x in narrowed['context']['original_jobs'] if x['original_job']['id']==a['second']['id'])['ownership_chain'])==2
    assert correct(c,narrowed).status_code==200;assert c.get('/api/jobs/'+a['second']['id']).json()['customer_id']==third['customer_id'];check(app)


def test_later_contact_fields_are_kept_and_cannot_be_silently_restored(app):
    c=client(app);a=merged(app,c);b=customer(c,a['target']['customer_id'])
    assert write(c,'/api/customers/'+b['id'],{'version':b['version'],**{k:b[k] for k in ('name','email','address')},'phone':'000000777777'},method='put').status_code==200
    bad=separation(c,a['merge']['id'],[],phone='restore_before_merge');assert not bad['can_confirm']
    before=table_rows(app.state.store.root,'oak');assert correct(c,bad).status_code==409;assert before==table_rows(app.state.store.root,'oak')
    good=separation(c,a['merge']['id'],[],name='restore_before_merge');assert correct(c,good).status_code==200
    b=customer(c,b['id']);assert b['phone']=='000000777777' and b['name']=='Blake Fictional';check(app)


@pytest.mark.parametrize('change',['new_job','job_edit','customer','portal','membership','new_merge'])
def test_stale_review_changes_nothing(app,change):
    c=client(app);a=merged(app,c);p=separation(c,a['merge']['id'],[a['job']['id']])
    if change=='new_job':work(c,customer(c,a['target']['customer_id']))
    elif change=='job_edit':edit(c,a['job'],notes='A later edit after separation preview')
    elif change=='customer':
        x=customer(c,a['target']['customer_id']);write(c,'/api/customers/'+x['id']+'/private-notes',{'version':x['record_version'],'notes':'Later retained private note'},method='put')
    elif change=='portal':write(c,'/api/customers/'+a['target']['customer_id']+'/portal-access/invite',{'version':customer(c,a['target']['customer_id'])['version']})
    elif change=='membership':
        with app.state.store.registry() as db:db.execute("UPDATE memberships SET version=version+1 WHERE business_id='oak' AND role='owner'")
    else:
        other=job(c);assert confirm(c,preview(c,selection(c,other,a['target']))).status_code==200
    before=table_rows(app.state.store.root,'oak');assert correct(c,p).status_code==409;assert before==table_rows(app.state.store.root,'oak')


@pytest.mark.parametrize('blocker',['active','timer','recurrence','booking','note','unrelated'])
def test_one_invalid_selected_job_blocks_entire_batch(app,blocker):
    from test_fieldwork import start,base
    from test_recurrence import definition
    from test_scheduling import booking
    from test_workforce import person
    c=client(app);a=merged(app,c);j=a['second'];j=c.get('/api/jobs/'+j['id']).json()
    people=[person(c)]
    if blocker in ('timer','note'):edit(c,j,status='accepted')
    if blocker=='active':edit(c,j,status='in_progress')
    elif blocker=='timer':assert start(c,j,people).status_code==200
    elif blocker=='recurrence':definition(c,app,j,people[0])
    elif blocker=='booking':booking(c,j,people[0],start_date='2099-01-01',end_date='2099-01-01')
    elif blocker=='note':assert write(c,base(j)+'/logs',{'kind':'note','content':'An unfinished invented review','work_date':'2026-10-28'}).status_code==200
    ids=[a['job']['id'],a['target']['id'] if blocker=='unrelated' else j['id']]
    p=separation(c,a['merge']['id'],ids);assert not p['can_confirm']
    before=table_rows(app.state.store.root,'oak');assert correct(c,p).status_code==409;assert before==table_rows(app.state.store.root,'oak')


def test_concurrent_lost_reply_and_permanent_same_id_mismatch(app):
    c=client(app);a=merged(app,c);p=separation(c,a['merge']['id'],[a['job']['id']]);key=str(uuid.uuid4());other=client(app)
    with ThreadPoolExecutor(2) as pool:rs=list(pool.map(lambda x:correct(x,p,key),[c,other]))
    assert all(r.status_code==200 for r in rs),[r.text for r in rs];assert rs[0].json()==rs[1].json();check(app)
    with app.state.store.tenant('oak') as db:db.execute('DELETE FROM operations WHERE key=?',(key,))
    before=table_rows(app.state.store.root,'oak');assert correct(c,p,key).json()==rs[0].json()
    after=table_rows(app.state.store.root,'oak');assert all(before[t]==after[t] for t in before if t!='operations')
    assert correct(c,p,key,reason='Different explanation for reused operation').status_code==409
    with app.state.store.registry() as db:db.execute("UPDATE memberships SET version=version+1 WHERE business_id='oak' AND role='owner'")
    assert correct(c,p,key).status_code==403


@pytest.mark.parametrize('phase',['reactivate','child','audit'])
def test_late_failure_rolls_back_every_row_then_same_attempt_succeeds(app,monkeypatch,phase):
    import backend.customer_review as review
    import backend.merge_corrections as correction
    c=client(app);a=merged(app,c,True);p=separation(c,a['merge']['id'],[a['job']['id'],a['second']['id']]);key=str(uuid.uuid4())
    before=table_rows(app.state.store.root,'oak')
    target=correction if phase=='audit' else review;name='audit' if phase=='audit' else 'record';original=getattr(target,name)
    def broken(*args,**kwargs):
        result=original(*args,**kwargs)
        if phase=='audit' or args[2]==('merge_correction.reactivate' if phase=='reactivate' else 'jobs.target'):raise RuntimeError('Injected late separation failure')
        return result
    monkeypatch.setattr(target,name,broken)
    with pytest.raises(RuntimeError,match='Injected late'):correct(c,p,key)
    assert before==table_rows(app.state.store.root,'oak')
    monkeypatch.setattr(target,name,original);assert correct(c,p,key).status_code==200;check(app)


def legacy(tmp_path):
    fixture=Path(__file__).parent/'fixtures/v0.24-merge-separation-format13.zip';meta=json.loads(fixture.with_suffix('.json').read_text())
    assert hashlib.sha256(fixture.read_bytes()).hexdigest()==meta['fixture_sha256'];root=tmp_path/'v24-restored'
    recovery.restore(fixture,meta['fixture_sha256'],root,confirm_fresh=True);app=create_app(root);c,_=new_login(app,root,'oak')
    return app,c,meta,fixture


def test_genuine_v24_format13_migration_separation_second_recovery(tmp_path):
    application,c,meta,fixture=legacy(tmp_path)
    with zipfile.ZipFile(fixture) as archive:
        assert json.loads(archive.read('manifest.json'))['version']==13
        path=tmp_path/'v24.sqlite3';path.write_bytes(archive.read('businesses/oak/business.sqlite3'))
    with sqlite3.connect(path) as db:
        db.row_factory=sqlite3.Row
        for table in ('customer_transfers','customer_transfer_jobs','customer_record_history','invoices','quotes','payments','source_history','field_history','field_log_history','photo_history','recurring_contracts'):
            assert [dict(r) for r in db.execute('SELECT * FROM '+table)]==table_rows(application.state.store.root,'oak')[table],table
    p=separation(c,meta['merge_id'],[meta['jobs']['sealed'],meta['jobs']['retired_repeat']],name='restore_before_merge')
    assert p['can_confirm'] and correct(c,p).status_code==200;check(application)
    assert c.get('/api/jobs/'+meta['jobs']['already_reassigned']).json()['customer_id']==meta['customers']['later_owner']
    assert not table_rows(application.state.store.root,'harbour')['jobs']
    saved=recovery.backup(application.state.store.root,tmp_path/'v25.zip')
    with zipfile.ZipFile(saved['archive']) as archive:assert (json.loads(archive.read('manifest.json'))['version'],json.loads(archive.read('manifest.json'))['prototype_version'])==(14,'0.25.0')
    dest=tmp_path/'again';recovery.restore(saved['archive'],saved['sha256'],dest,confirm_fresh=True);check(create_app(dest))
    before=table_rows(application.state.store.root,'oak');after=table_rows(dest,'oak')
    for table in ('customer_merge_corrections','customer_transfers','customer_transfer_jobs','customer_record_history','quotes','invoices','payments','source_history','field_history','field_log_history','recurring_contracts'):assert before[table]==after[table],table
    second=recovery.backup(dest,tmp_path/'second.zip');assert recovery.verify(second['archive'],second['sha256'])['valid']


def test_owner_foreign_customer_csrf_and_untrusted_fields(app):
    from test_identity import add_staff
    c=client(app);a=merged(app,c);p=separation(c,a['merge']['id'],[]);staff,_=add_staff(app,c);pc,_=portal(app,c,a['target'])
    for other,status in [(staff,403),(pc,401),(TestClient(app,headers=ORIGIN),401),(client(app,1),404)]:
        assert other.get(ROOT+'/merges/'+a['merge']['id']+'/correction').status_code==status
        assert write(other,ROOT+'/merge-correction-preview',p['selection']).status_code==status
        assert correct(other,p).status_code==status
    foreign=job(client(app,1));assert write(c,ROOT+'/merge-correction-preview',{**p['selection'],'alias_id':foreign['customer_id']}).status_code==404
    for extra in ({'confirmed':False},{'reason':'   '},{'restore_access':True},{'job_ids':[a['job']['id']]*2}):assert correct(c,p,**extra).status_code==422
    c.headers['x-csrf-token']='incorrect';assert correct(c,p).status_code==403


@pytest.mark.parametrize('cached',[False,True])
@pytest.mark.parametrize('change',['revoked','role','session'])
def test_fresh_authority_after_middleware_before_cached_or_new_write(app,monkeypatch,cached,change):
    c=client(app);a=merged(app,c);p=separation(c,a['merge']['id'],[]);key=str(uuid.uuid4())
    if cached:assert correct(c,p,key).status_code==200
    before=table_rows(app.state.store.root,'oak');identity=app.state.identity;original=identity.session
    def changed(request,*args,**kwargs):
        result=original(request,*args,**kwargs)
        if request.url.path==ROOT+'/merge-corrections':
            with app.state.store.registry() as db:
                if change=='revoked':db.execute("UPDATE memberships SET enabled=0 WHERE user_id='oak'")
                elif change=='role':db.execute("UPDATE memberships SET role='staff',version=version+1 WHERE user_id='oak'")
                else:db.execute("DELETE FROM sessions WHERE subject='oak'")
        return result
    monkeypatch.setattr(identity,'session',changed)
    assert correct(c,p,key).status_code in (401,403);assert before==table_rows(app.state.store.root,'oak')


def test_registry_writer_lock_is_held_for_fresh_correction(app,monkeypatch):
    import backend.merge_corrections as module
    c=client(app);a=merged(app,c);p=separation(c,a['merge']['id'],[]);original=module.execute;observed=[]
    def checked(*args,**kwargs):
        with sqlite3.connect(app.state.store.root/'registry.sqlite3',timeout=0) as other:
            with pytest.raises(sqlite3.OperationalError,match='locked'):other.execute('BEGIN IMMEDIATE')
        observed.append(True);return original(*args,**kwargs)
    monkeypatch.setattr(module,'execute',checked);assert correct(c,p).status_code==200;assert observed==[True]


def test_old_note_epoch_receipt_repeat_and_followup_do_not_return(app):
    from test_fieldwork import setup
    from test_device_notes import access,action
    from test_recurrence import definition,apply
    from test_workflow import items,decide
    c=client(app);j,people,_=setup(c,app);edit(c,j,status='accepted',next_action='Original invented follow-up')
    note=access(c,j);op=str(uuid.uuid4());assert action(c,j,op,{'binding':note['binding']},'cancel').status_code==200
    d,_,_=definition(c,app,j,people[0]);d=apply(c,d,{'action':'state','state':'retired','reason':'Retire original definition before merge'})
    old=next(i for i in items(c,'next_action') if i['job_id']==j['id']);decide(c,old,'dismiss')
    b=job(c);merge=confirm(c,preview(c,selection(c,j,b))).json();middle=access(c,j,note['binding']['device_id'])
    p=separation(c,merge['id'],[j['id']]);assert correct(c,p).status_code==200
    fresh=access(c,j,note['binding']['device_id']);assert len({note['key'],middle['key'],fresh['key']})==3
    for previous in (note,middle):
        for verb in ('inspect','cancel','saved'):assert action(c,j,op,{'binding':previous['binding']},verb).status_code==403
    assert action(c,j,op,{'binding':fresh['binding']},'inspect').status_code==409
    assert write(c,'/api/recurring/'+d['id']+'/preview',{'version':d['version'],'operation':{'action':'state','state':'restore','reason':'Attempt old repeat work after separation'}}).status_code==409
    edit(c,j,followup_excluded=False);assert all(not i['eligible'] for i in items(c) if i['job_id']==j['id']);check(app)


@pytest.mark.parametrize('tamper',['missing_event','original_reason','missing_child','child_selection','result','selection','context','fields','reactivation','alias','access_event','actor'])
def test_tampered_correction_recovery_never_creates_destination(app,tmp_path,tamper):
    c=client(app);a=merged(app,c,True);p=separation(c,a['merge']['id'],[a['job']['id']],name='restore_before_merge');assert correct(c,p).status_code==200
    source=tmp_path/'source.zip';recovery.backup(app.state.store.root,source)
    def mutate(files,meta):
        from backend.customer_review import packed
        from backend.store import digest
        name='businesses/oak/business.sqlite3';path=tmp_path/'tamper.sqlite3';path.write_bytes(files[name])
        with sqlite3.connect(path) as db:
            row=db.execute('SELECT id,plan,result,child_transfer_id FROM customer_merge_corrections').fetchone();raw=json.loads(row[1])
            if tamper=='missing_event':db.execute('DELETE FROM customer_merge_corrections')
            elif tamper=='original_reason':db.execute("UPDATE customer_transfers SET reason='Rewritten original merge' WHERE kind='merge'")
            elif tamper=='missing_child':db.execute('DELETE FROM customer_transfers WHERE id=?',(row[3],))
            elif tamper=='child_selection':
                child=json.loads(db.execute('SELECT plan FROM customer_transfers WHERE id=?',(row[3],)).fetchone()[0]);child['selection']['job_ids']=[];child['preview_hash']=digest(packed({k:v for k,v in child.items() if k!='preview_hash'}));db.execute('UPDATE customer_transfers SET plan=?,preview_hash=? WHERE id=?',(packed(child),child['preview_hash'],row[3]))
            elif tamper=='result':
                result=json.loads(row[2]);result['job_ids']=[];db.execute('UPDATE customer_merge_corrections SET result=?',(packed(result),))
            elif tamper=='reactivation':db.execute("DELETE FROM customer_record_history WHERE action='merge_correction.reactivate'")
            elif tamper=='alias':db.execute("UPDATE customer_records SET state='merged',canonical_id=? WHERE customer_id=?",(a['target']['customer_id'],a['job']['customer_id']))
            elif tamper=='access_event':db.execute('DELETE FROM portal_events WHERE id=(SELECT MAX(id) FROM portal_events)')
            else:
                if tamper=='selection':raw['selection']['job_ids'].append(a['target']['id'])
                elif tamper=='context':raw['context']['original_jobs'].pop()
                elif tamper=='fields':raw['retained_fields']['phone']='A later value overwritten'
                else:raw['actor_access']['business_id']='harbour'
                raw['preview_hash']=digest(packed({k:v for k,v in raw.items() if k!='preview_hash'}));db.execute('UPDATE customer_merge_corrections SET plan=?,preview_hash=?',(packed(raw),raw['preview_hash']))
        files[name]=path.read_bytes();meta['files'][name].update(bytes=len(files[name]),sha256=hashlib.sha256(files[name]).hexdigest())
    bad=tmp_path/'bad.zip';sha=rewrite_archive(source,bad,mutate);dest=tmp_path/'must-not-exist'
    with pytest.raises((recovery.RecoveryError,ValueError,KeyError,AssertionError)):recovery.restore(bad,sha,dest,confirm_fresh=True)
    assert not dest.exists()


def test_operation_ids_are_permanent_across_both_correction_routes(app):
    c=client(app);a=merged(app,c);p=separation(c,a['merge']['id'],[]);key=str(uuid.uuid4());assert correct(c,p,key).status_code==200
    with app.state.store.tenant('oak') as db:db.execute('DELETE FROM operations WHERE key=?',(key,))
    ordinary=preview(c,selection(c,a['target'],a['job'],kind='jobs',job_ids=[a['job']['id']]))
    assert confirm(c,ordinary,key).status_code==409
    newkey=str(uuid.uuid4());assert confirm(c,ordinary,newkey).status_code==200
    next_merge=merged(app,c);next_plan=separation(c,next_merge['merge']['id'],[])
    with app.state.store.tenant('oak') as db:db.execute('DELETE FROM operations WHERE key=?',(newkey,))
    assert correct(c,next_plan,newkey).status_code==409;check(app)


@pytest.mark.parametrize('return_job',[False,True])
def test_new_post_merge_permissions_are_revoked_including_reactivation_only(app,tmp_path,return_job):
    from test_photos import reviewed,share,project,pub,record
    from test_feedback import completed,prepare,respond,publish,BASE,PUBLIC,ACCESS
    c=client(app);a=merged(app,c);j=a['target']
    photo=reviewed(c,j);assert share(c,j,photo).status_code==200;gallery=project(c,j,photo);assert pub(c,gallery).status_code==200
    completed(c,j);_,rid,code=prepare(c,j);respond(c,app,rid,code);publish(c,rid)
    p=separation(c,a['merge']['id'],[a['job']['id']] if return_job else []);assert correct(c,p).status_code==200;check(app)
    assert record(c,j,photo)['customer_share'] is None and record(c,j,photo)['reviewed']==0
    assert c.get('/api/gallery/'+gallery['id']).json()['published'] is None
    assert c.get(BASE+rid).json()['request']['state']=='revoked' and c.get(PUBLIC).json()['count']==0
    assert write(TestClient(app,headers=ORIGIN),ACCESS+'access',{'code':code}).status_code==401
    saved=recovery.backup(app.state.store.root,tmp_path/'revoked-new-sharing.zip');assert recovery.verify(saved['archive'],saved['sha256'])['valid']


def test_original_public_intake_receipt_and_photo_facts_survive_separation(app):
    from test_public_enquiries import payload,photo,submit,saved
    c=client(app);data=payload(photos=[photo()]);key=str(uuid.uuid4());j=saved(c,submit(app,data,key));b=job(c)
    merge=confirm(c,preview(c,selection(c,j,b)));assert merge.status_code==200
    before=table_rows(app.state.store.root,'oak');p=separation(c,merge.json()['id'],[j['id']]);assert correct(c,p).status_code==200
    after=table_rows(app.state.store.root,'oak')
    for table in ('enquiry_intakes','enquiry_photo_sources','photo_blobs'):assert before[table]==after[table],table
    assert c.get('/api/jobs/'+j['id']).json()['enquiry_submission']==j['enquiry_submission']
    with app.state.store.tenant('oak') as db:db.execute('DELETE FROM operations')
    assert submit(app,data,key).json()['receipt']==j['id'];check(app)


def test_archived_target_and_different_attempt_concurrency_are_guarded(app):
    c=client(app);a=merged(app,c)
    for j in (a['job'],a['second'],a['target']):edit(c,j,status='cancelled')
    b=customer(c,a['target']['customer_id']);assert write(c,'/api/customers/'+b['id']+'/record-state',{'version':b['version'],'state':'archived','reason':'Archive invented retained record','confirmed':True}).status_code==200
    p=separation(c,a['merge']['id'],[]);assert not p['can_confirm'];before=table_rows(app.state.store.root,'oak');assert correct(c,p).status_code==409;assert before==table_rows(app.state.store.root,'oak')
    b=customer(c,b['id']);assert write(c,'/api/customers/'+b['id']+'/record-state',{'version':b['version'],'state':'active','reason':'Restore invented retained record','confirmed':True}).status_code==200
    p=separation(c,a['merge']['id'],[])
    with ThreadPoolExecutor(2) as pool:responses=list(pool.map(lambda cli:correct(cli,p),[c,client(app)]))
    assert sorted(r.status_code for r in responses)==[200,409];assert len(table_rows(app.state.store.root,'oak')['customer_merge_corrections'])==1;check(app)