#!/usr/bin/env python3
"""Deterministic read-only LinkedIn–Marketo integration evaluator."""
import json, re, sys
from datetime import datetime
from pathlib import Path

SCHEMA='maestro.linkedin-marketo-integration.v1'
PHASES={'draft','launch','post_test'}
POPS=('submitted','connector_observed','person_activity','membership','routed','duplicate','unresolved')
EVIDENCE_TYPES={'native_readback','native_receipt','source_report','operator_note','documented_source','fictional_example_receipt'}
MATCH_OUTCOMES={'NO_COLLISION','MATCHED_EXISTING','MERGED','UNRESOLVED_COLLISION','UNKNOWN'}
ACTIVITY_OUTCOMES={'OBSERVED','MISSING','WRONG_ACTIVITY','UNKNOWN'}
MEMBERSHIP_OUTCOMES={'PRESENT','ABSENT','NOT_REQUIRED','UNKNOWN'}
ROUTE_OUTCOMES={'ACCEPTED','REJECTED','NOT_REQUIRED','UNKNOWN'}
REQUESTS={
 'planning':'Complete the missing planned fields before review: {path}.',
 'authorization':'Capture current LaunchPoint status, authorized user reference and authorization date from a dated native read-back.',
 'account_form_scope':'Read back the exact Business Account, ad account, Page and Lead Gen Form from the native service and compare them with approved scope.',
 'field_value_map':'Read back each required form field, Marketo destination/type, allowed values, consent action and any observed values.',
 'source_test':'After the approved test, capture its timestamp, timezone, exact form/account, Test Lead flag and source-report reference.',
 'matching':'Read back the person-match/deduplication outcome, any collision or merge, and its evidence reference.',
 'person_activity':'Read back the person and exact Fills Out LinkedIn Lead Gen Form activity at the test timestamp.',
 'membership':'Read back the intended list/program membership outcome and dated evidence for this person.',
 'routing':'Capture an accepted downstream routing receipt, destination and owner; a response target is not an observed response.',
 'population_evidence':'Provide the missing population value and dated evidence for the exact form, ad account, cohort, cutoff and timezone; preserve blank, Unknown and zero.',
 'reconciliation':'Reconcile submitted = person/activity + duplicate + unresolved and check downstream totals for the same eligible cohort.',
 'import_contract':'Preserve the supplied bytes; repair the malformed or contradictory fields before reevaluating.'}

def kind(x):
 if x is None:return 'null'
 if isinstance(x,bool):return 'boolean'
 if isinstance(x,dict):return 'object'
 if isinstance(x,list):return 'array'
 if isinstance(x,str):return 'string'
 if isinstance(x,(int,float)):return 'number'
 return type(x).__name__
def parse_dt(x):
 if not isinstance(x,str) or not x:return None
 try:return datetime.fromisoformat(x.replace('Z','+00:00'))
 except Exception:return None
def secret_key_paths(value,path='$'):
 out=[]
 if isinstance(value,dict):
  for key,child in value.items():
   k=str(key); compact=re.sub(r'[^a-z0-9]','',k.casefold())
   if any(term in compact for term in ('password','token','secret','cookie','authorizationcode')):out.append(path+'.'+k)
   out.extend(secret_key_paths(child,path+'.'+k))
 elif isinstance(value,list):
  for i,child in enumerate(value):out.extend(secret_key_paths(child,f'{path}[{i}]'))
 return out
def make_result(record,raw,phase,disposition,state=None,path=None,reason='',request='',diagnostics=None,pops=None):
 first={'state':state,'path':path,'reason':reason} if state else None
 norm=pops or {k:{'state':'BLANK','value':None,'evidence_ref':'','scope':None} for k in POPS}
 result={'evaluation_phase':phase,'disposition':disposition,'first_broken_state':state,'first_broken':first,'next_evidence_request':request,'normalized_populations':norm,'diagnostics':diagnostics or [],'readable_markdown':render_markdown(record,phase,disposition,first,request,norm,diagnostics or []),'input_preserved_json':raw}
 return result
def render_markdown(record,phase,disposition,first,request,pops,diagnostics):
 lines=['# LinkedIn–Marketo evaluation','',f'- Phase: `{phase}`',f'- Disposition: `{disposition}`']
 if first: lines += [f"- First broken state: `{first['state']}` at `{first['path']}`",f"- Finding: {first['reason']}"]
 else: lines += ['- First broken state: none']
 if request:lines += [f'- Next request: {request}']
 lines += ['', '## Population reconciliation','| Population | State | Value | Evidence reference | Bound scope |','|---|---|---:|---|---|']
 for name in POPS:
  p=pops[name]; scope=p.get('scope')
  lines.append('| '+ ' | '.join([name,p['state'],json.dumps(p['value'],ensure_ascii=False,separators=(',',':')),p.get('evidence_ref',''),json.dumps(scope,ensure_ascii=False,sort_keys=True,separators=(',',':')) if scope else '(blank)'])+' |')
 if diagnostics:
  lines += ['', '## Diagnostics']
  for d in diagnostics:lines.append(f"- `{d.get('path','$')}` — expected {d.get('expected_kind','valid value')}; received {d.get('received_kind','invalid')}; {d.get('message','')}")
 lines += ['', '## Source record','```json',json.dumps(record,ensure_ascii=False,indent=2), '```','']
 return '\n'.join(lines)
def output_error(record,raw,phase,state,path,reason,diagnostics=None,disposition='REPAIR_REQUIRED'):
 req=REQUESTS['import_contract'] if state=='import_contract' else REQUESTS['field_value_map']
 return make_result(record,raw,phase,disposition,state,path,reason,req,diagnostics)
def check(record,raw):
 if not isinstance(record,dict):return output_error(record,raw,'draft','import_contract','$',f'Record must be a JSON object, got {kind(record)}.')
 secrets=secret_key_paths(record)
 if secrets:
  ds=[{'path':p,'expected_kind':'no secret-shaped key','received_kind':'secret-shaped key','message':'Remove credential-shaped field; its value is intentionally not repeated here.'} for p in secrets]
  return make_result(record,raw,'draft','REPAIR_REQUIRED','import_contract',secrets[0],'Credential-shaped keys are not allowed.',REQUESTS['import_contract'],ds)
 schema=record.get('schema_version')
 phase=record.get('evaluation_phase')
 if phase is None:
  t=record.get('test') if isinstance(record.get('test'),dict) else {}
  claims=bool(t.get('submitted_at') or t.get('linkedin_evidence_ref') or t.get('test_lead_flag') is True)
  phase='post_test' if claims else 'draft'
 if phase not in PHASES:
  return output_error(record,raw,'draft','import_contract','evaluation_phase',f'evaluation_phase must be one of {sorted(PHASES)}; got {phase!r}.')
 if schema!=SCHEMA:return output_error(record,raw,phase,'import_contract','schema_version',f'Expected {SCHEMA}.')
 # Required shape checks. Paths and kinds are explicit; original values are omitted from diagnostics.
 errors=[]
 mapping_errors=[]
 population_issues=[]
 required={'record_id':'string','example_only':'boolean','scope':'object','connection':'object','field_mappings':'array','destination':'object','matching':'object','routing':'object','test':'object','cutoff':'object','populations':'object','evidence':'array'}
 for k,t in required.items():
  if k not in record or kind(record.get(k))!=t:errors.append({'path':k,'expected_kind':t,'received_kind':'missing' if k not in record else kind(record.get(k)),'message':'Required field has the wrong shape.'})
 if errors:return make_result(record,raw,phase,'REPAIR_REQUIRED','import_contract',errors[0]['path'],errors[0]['message'],REQUESTS['import_contract'],errors)
 r=record; scope=r['scope']; cutoff=r['cutoff']; maps=r['field_mappings']; pops={}
 def req_fields(obj,path,keys,expected='string'):
  if not isinstance(obj,dict):return
  for k in keys:
   if k not in obj or kind(obj.get(k))!=expected:errors.append({'path':path+'.'+k,'expected_kind':expected,'received_kind':'missing' if k not in obj else kind(obj.get(k)),'message':'Required nested field has the wrong shape.'})
 req_fields(scope,'scope',['organization','linkedin_user_ref','business_account_id','ad_account_id','page_id','form_name','form_id','marketo_subscription_ref','launchpoint_service_name','launchpoint_service_id','approved_by','approved_at'])
 req_fields(r['connection'],'connection',['planned','observed'],'object')
 if isinstance(r['connection'].get('planned'),dict):req_fields(r['connection']['planned'],'connection.planned',['status','steps_or_reference','owner'])
 if isinstance(r['connection'].get('observed'),dict):req_fields(r['connection']['observed'],'connection.observed',['status','authorized_at','checked_at','evidence_ref','business_account_id','ad_account_id','page_id','form_id'])
 req_fields(cutoff,'cutoff',['start','end','timezone','cohort','form_id','ad_account_id'])
 sections={'destination':['type','name','id','smart_campaign_name','smart_campaign_id','trigger_or_filter','qualification_rule','flow_action','membership_status_expected','observed_membership_status','observed_at','evidence_ref','membership_outcome'], 'matching':['match_rule','dedupe_behavior','collision_or_merge','observed_at','evidence_ref','observed_outcome'], 'routing':['destination','event','owner','response_target','observed_event','observed_at','evidence_ref','outcome'], 'test':['authorized_by','submitted_at','timezone','test_person_ref','linkedin_evidence_ref','marketo_person_id','activity_name','activity_at','person_activity_evidence_ref','activity_outcome']}
 # New outcome fields are optional only for legacy v1 fixtures; absent values are derived and reported in normalized output.
 for sec,keys in sections.items():
  if not isinstance(r.get(sec),dict):continue
  req_fields(r[sec],sec,[k for k in keys if k not in ('membership_outcome','observed_outcome','outcome','activity_outcome')])
  outcome_key={'destination':'membership_outcome','matching':'observed_outcome','routing':'outcome','test':'activity_outcome'}[sec]
  if outcome_key in r[sec] and not isinstance(r[sec][outcome_key],str):errors.append({'path':sec+'.'+outcome_key,'expected_kind':'string enum','received_kind':kind(r[sec][outcome_key]),'message':'Outcome must be a string literal.'})
 for i,m in enumerate(maps):
  path=f'field_mappings[{i}]'
  if not isinstance(m,dict):errors.append({'path':path,'expected_kind':'object','received_kind':kind(m),'message':'Mapping entry must be an object.'});continue
  req_fields(m,path,['linkedin_field','marketo_field','data_type','normalization','consent_action','state','evidence_ref'])
  req_fields(m,path,['allowed_values'],'array')
  req_fields(m,path,['required'],'boolean')
  if not isinstance(m.get('allowed_values'),list):continue
  if any(not isinstance(v,(str,bool,int,float)) or v is None for v in m['allowed_values']):mapping_errors.append({'path':path+'.allowed_values','expected_kind':'array of scalar values','received_kind':'array with invalid member','message':'Allowed values must be scalar literals.'})
  obs=m.get('observed_values',[])
  if 'observed_values' in m and not isinstance(obs,list):errors.append({'path':path+'.observed_values','expected_kind':'array','received_kind':kind(obs),'message':'Observed values must be a typed array.'})
  typ=m.get('data_type')
  supported=('email','text','enum','boolean','number','date','phone','url','integer','decimal')
  # An empty type is an incomplete plan value, not a malformed contract. It is
  # reported by the planning gate below; typed invalid values remain repairs.
  if not typ and phase=='draft':continue
  if typ not in supported:
   mapping_errors.append({'path':path+'.data_type','expected_kind':'supported field type','received_kind':str(typ),'message':'Unsupported field type.'});continue
  if typ=='enum' and (not m['allowed_values'] or any(not isinstance(v,str) or not v.strip() for v in m['allowed_values'])):mapping_errors.append({'path':path+'.allowed_values','expected_kind':'nonempty array of approved enum strings','received_kind':'empty or invalid','message':'Enum mappings require approved choices.'})
  if typ=='boolean':
   vals={str(v).lower() for v in m['allowed_values']}
   if vals!={'true','false'}:mapping_errors.append({'path':path+'.allowed_values','expected_kind':'both true and false','received_kind':str(m['allowed_values']),'message':'Boolean mappings require both approved boolean choices.'})
  if typ not in ('enum','boolean') and m['allowed_values']:mapping_errors.append({'path':path+'.allowed_values','expected_kind':'empty array for non-choice field','received_kind':'nonempty array','message':'Choice lists only apply to enum/boolean fields.'})
  if isinstance(obs,list) and obs:
   normalized={str(v).casefold() for v in m['allowed_values']}
   for j,v in enumerate(obs):
    if typ=='boolean' and not (isinstance(v,bool) or isinstance(v,str) and v.casefold() in ('true','false')):mapping_errors.append({'path':f'{path}.observed_values[{j}]','expected_kind':'boolean or true/false literal','received_kind':kind(v),'message':'Observed boolean value is not typed.'})
    elif typ=='enum' and not isinstance(v,str):mapping_errors.append({'path':f'{path}.observed_values[{j}]','expected_kind':'enum string','received_kind':kind(v),'message':'Observed enum value must be a string.'})
    elif typ in ('enum','boolean') and str(v).casefold() not in normalized:mapping_errors.append({'path':f'{path}.observed_values[{j}]','expected_kind':'one of the approved values','received_kind':str(v),'message':'Observed value is outside the approved set.'})
 # evidence types, evidence keys, timestamps; secret scan already ran before output.
 evidence_by_ref={}
 for i,e in enumerate(r['evidence']):
  path=f'evidence[{i}]'
  if not isinstance(e,dict):errors.append({'path':path,'expected_kind':'object','received_kind':kind(e),'message':'Evidence entry must be an object.'});continue
  req_fields(e,path,['evidence_ref','evidence_type','captured_at','supports','note'])
  if e.get('evidence_type') not in EVIDENCE_TYPES:errors.append({'path':path+'.evidence_type','expected_kind':'allowed evidence type','received_kind':str(e.get('evidence_type')),'message':'Evidence type is not permitted.'})
  if e.get('evidence_ref'):evidence_by_ref[e['evidence_ref']]=e
  if e.get('captured_at') and not parse_dt(e['captured_at']):errors.append({'path':path+'.captured_at','expected_kind':'ISO-8601 timestamp','received_kind':'invalid timestamp','message':'Evidence capture time is malformed.'})
 # New population state/value shape, while accepting the unambiguous legacy count shape for v1 fixtures.
 if set(r['populations'])!=set(POPS):errors.append({'path':'populations','expected_kind':'exactly seven named population keys','received_kind':','.join(r['populations'].keys()),'message':'Population set is incomplete or contains extra keys.'})
 for name in POPS:
  p=r['populations'].get(name)
  if not isinstance(p,dict):errors.append({'path':'populations.'+name,'expected_kind':'object','received_kind':kind(p),'message':'Population row must be an object.'});continue
  if 'state' in p or 'value' in p:
   state=p.get('state');value=p.get('value')
   if state not in ('KNOWN','UNKNOWN','BLANK'):errors.append({'path':f'populations.{name}.state','expected_kind':'KNOWN, UNKNOWN or BLANK','received_kind':str(state),'message':'Invalid population state.'})
   if state=='KNOWN' and (not isinstance(value,int) or isinstance(value,bool) or value<0):errors.append({'path':f'populations.{name}.value','expected_kind':'nonnegative integer','received_kind':kind(value),'message':'Known counts require a nonnegative integer.'})
   if state in ('UNKNOWN','BLANK') and value is not None:errors.append({'path':f'populations.{name}.value','expected_kind':'null for UNKNOWN/BLANK','received_kind':kind(value),'message':'Unknown and blank values must be null.'})
   ref=p.get('evidence_ref',''); evscope=p.get('evidence_scope')
  else:
   count=p.get('count','')
   if count=='':state='BLANK';value=None
   elif count=='Unknown':state='UNKNOWN';value=None
   elif isinstance(count,int) and not isinstance(count,bool) and count>=0:state='KNOWN';value=count
   else:errors.append({'path':f'populations.{name}.count','expected_kind':'blank, Unknown or nonnegative integer','received_kind':kind(count),'message':'Invalid legacy count.'});state='BLANK';value=None
   ref=p.get('evidence_ref',''); evscope=None
  pops[name]={'state':state,'value':value,'evidence_ref':ref,'scope':evscope,'definition':p.get('definition',''),'evidence_type':p.get('evidence_type',''),'captured_at':p.get('captured_at','')}
 if errors:return make_result(r,raw,phase,'REPAIR_REQUIRED','import_contract',errors[0]['path'],errors[0]['message'],REQUESTS['import_contract'],errors,pops)
 if mapping_errors:return make_result(r,raw,phase,'REPAIR_REQUIRED','field_value_map',mapping_errors[0]['path'],mapping_errors[0]['message'],REQUESTS['field_value_map'],mapping_errors,pops)
 # Static same-form double-map detection is derived from source/destination pairs, never caller state labels.
 destinations={}
 for i,m in enumerate(maps):
  src=' '.join(m['linkedin_field'].split()).casefold();dst=' '.join(m['marketo_field'].split()).casefold()
  if src and dst:destinations.setdefault(dst,[]).append((i,src))
 collisions=[rows for rows in destinations.values() if len({src for _,src in rows})>1]
 if collisions:return make_result(r,raw,phase,'REPAIR_REQUIRED','field_value_map','field_mappings','Two distinct fields on the selected form map to the same Marketo destination.',REQUESTS['field_value_map'],[{'path':f'field_mappings[{i}].marketo_field','expected_kind':'unique destination per source field','received_kind':'same-form collision','message':'Change the map or split source forms.'} for rows in collisions for i,_ in rows],pops)
 # Evidence lookup never treats public docs or operator notes as native proof.
 def evidence(ref,allowed=None):
  e=evidence_by_ref.get(ref) if isinstance(ref,str) else None
  good={'native_readback','native_receipt','source_report'} | ({'fictional_example_receipt'} if r['example_only'] else set())
  return e if e and e.get('evidence_type') in (allowed or good) and e.get('captured_at') and e.get('supports') else None
 # Normalize old and new populations into state/value; the new contract binds each row to the exact cutoff tuple.
 expected_scope={'form_id':cutoff.get('form_id',''),'ad_account_id':cutoff.get('ad_account_id',''),'cohort':cutoff.get('cohort',''),'cutoff_start':cutoff.get('start',''),'cutoff_end':cutoff.get('end',''),'timezone':cutoff.get('timezone','')}
 for name in POPS:
  p=r['populations'][name]; n=pops[name]; ref=n['evidence_ref']; e=evidence(ref)
  if n['state']=='KNOWN':
   if not e or p.get('evidence_type') and p.get('evidence_type')!=e.get('evidence_type') or p.get('captured_at') and p.get('captured_at')!=e.get('captured_at'):
    errors.append({'path':f'populations.{name}.evidence_ref','expected_kind':'typed dated evidence register reference','received_kind':'missing or inconsistent','message':'Known count requires matching evidence type and timestamp.'})
   if 'evidence_scope' in p:
    es=p['evidence_scope']; n['scope']=es
    if not isinstance(es,dict) or es!=expected_scope:population_issues.append({'path':f'populations.{name}.evidence_scope','expected_kind':'exact form/account/cohort/cutoff/timezone scope','received_kind':'mismatched scope','message':'Population evidence scope differs from the record cutoff.'})
   else:
    # Legacy v1 fixture adapter: require the evidence timestamp inside cutoff and publish the derived scope.
    cap=parse_dt(n.get('captured_at'))
    start=parse_dt(cutoff.get('start'));end=parse_dt(cutoff.get('end'))
    if not cap or not start or not end or cap<start or cap>end:population_issues.append({'path':f'populations.{name}.captured_at','expected_kind':'timestamp inside declared cutoff','received_kind':'outside or missing cutoff','message':'Population evidence falls outside the declared cutoff.'})
    definition=n['definition'].casefold()
    if any(term in definition for term in ('every linkedin form','every form','all forms','another form','outside the declared')):population_issues.append({'path':f'populations.{name}.definition','expected_kind':'same-form cohort definition','received_kind':'incompatible population definition','message':'Definition explicitly describes another form/cohort.'})
    n['scope']={**expected_scope,'binding':'legacy_v1_cutoff_derived'}
  elif name=='connector_observed' and n['state']=='UNKNOWN':
   if not any(term in n['definition'].casefold() for term in ('not expose','no distinct','unavailable','not separately exposed','does not separately expose')):errors.append({'path':'populations.connector_observed.definition','expected_kind':'explicit reason no connector count is available','received_kind':'unsupported unknown','message':'Unknown connector counts require an explicit unavailability reason.'})
  elif n['state'] in ('UNKNOWN','BLANK') and name!='connector_observed':pass
 if errors:return make_result(r,raw,phase,'REPAIR_REQUIRED','import_contract',errors[0]['path'],errors[0]['message'],REQUESTS['import_contract'],errors,pops)
 if population_issues:return make_result(r,raw,phase,'LAUNCH_HOLD','population_evidence',population_issues[0]['path'],population_issues[0]['message'],REQUESTS['population_evidence'],population_issues,pops)
 # observed outcomes: accept typed new values; deterministic adapter for older v1 examples.
 matching=r['matching']; test=r['test']; dest=r['destination']; routing=r['routing']
 match=matching.get('observed_outcome')
 if match is None:
  txt=str(matching.get('collision_or_merge','')).casefold()
  if 'unresolved collision' in txt or 'collision between' in txt:match='UNRESOLVED_COLLISION'
  elif 'merge occurred' in txt and 'no merge' not in txt:match='MERGED'
  elif 'matched an existing' in txt or 'matched existing' in txt:match='MATCHED_EXISTING'
  elif txt:match='NO_COLLISION'
  else:match='UNKNOWN'
 activity=test.get('activity_outcome')
 if activity is None:activity='OBSERVED' if test.get('activity_name')=='Fills Out LinkedIn Lead Gen Form' else ('WRONG_ACTIVITY' if test.get('activity_name') else 'UNKNOWN')
 membership=dest.get('membership_outcome')
 if membership is None:membership='PRESENT' if dest.get('observed_membership_status') and dest.get('observed_membership_status')==dest.get('membership_status_expected') else ('ABSENT' if dest.get('observed_membership_status') else 'UNKNOWN')
 route=routing.get('outcome')
 if route is None:
  rt=str(routing.get('observed_event','')).casefold()
  route='REJECTED' if any(x in rt for x in ('reject','fail','denied','not accepted')) else ('ACCEPTED' if rt else 'UNKNOWN')
 observed={'matching.observed_outcome':(match,MATCH_OUTCOMES),'test.activity_outcome':(activity,ACTIVITY_OUTCOMES),'destination.membership_outcome':(membership,MEMBERSHIP_OUTCOMES),'routing.outcome':(route,ROUTE_OUTCOMES)}
 for path,(val,allowed) in observed.items():
  if val not in allowed:errors.append({'path':path,'expected_kind':'one of '+','.join(sorted(allowed)),'received_kind':str(val),'message':'Observed outcome is not an allowed literal.'})
 if errors:return make_result(r,raw,phase,'REPAIR_REQUIRED','import_contract',errors[0]['path'],errors[0]['message'],REQUESTS['import_contract'],errors,pops)
 # Find incomplete planned fields in stable order.
 missing=[]
 plan={'scope':['organization','linkedin_user_ref','business_account_id','ad_account_id','page_id','form_id','marketo_subscription_ref','launchpoint_service_name','launchpoint_service_id','approved_by'],
       'connection.planned':['status','steps_or_reference','owner'],
       'field_mappings':['linkedin_field','marketo_field','data_type','normalization','consent_action'],
       'destination':['type','name','id','smart_campaign_name','trigger_or_filter','qualification_rule','flow_action'],
       'matching':['match_rule','dedupe_behavior'],
       'routing':['destination','event','owner'],
       'test':['authorized_by'],
       'cutoff':['start','end','timezone','cohort','form_id','ad_account_id']}
 for group,keys in plan.items():
  if group=='field_mappings':
   if not maps:missing.append('field_mappings')
   for i,m in enumerate(maps):
    for k in keys:
     if m.get('required',True) and not m.get(k):missing.append(f'field_mappings[{i}].{k}')
  else:
   obj=r
   for component in group.split('.'):obj=obj.get(component,{}) if isinstance(obj,dict) else {}
   for k in keys:
    if not isinstance(obj,dict) or not obj.get(k):missing.append(group+'.'+k)
 # enum and boolean mapping config is not a missing field: it is a repair issue.
 def fail(state,path,reason,disposition='LAUNCH_HOLD',request_key=None,request=None):
  return make_result(r,raw,phase,disposition,state,path,reason,request or REQUESTS[request_key or state],[],pops)
 if missing:
  path=missing[0]
  return fail('planning',path,'Required planned value is blank.',request=REQUESTS['planning'].format(path=path))
 # Plans are structurally complete at this point. Draft phase is a reviewable planned record, with no native observations required.
 if phase=='draft':
  return make_result(r,raw,phase,'DRAFT_READY',None,None,'Plan is complete enough for operator review; no execution is claimed.','No native action is authorized by this draft record.',[],pops)
 # Launch and post-test share the non-mutating native preflight gates.
 conn=r['connection']['observed']
 auth_ref=conn.get('evidence_ref');auth_e=evidence(auth_ref)
 if conn.get('status')!='authorized' or not conn.get('authorized_at') or not conn.get('checked_at') or not auth_e:
  return fail('authorization','connection.observed','Current dated native authorization evidence is missing.','LAUNCH_HOLD','authorization')
 auth_at=parse_dt(conn['authorized_at']);checked_at=parse_dt(conn['checked_at'])
 if not auth_at or not checked_at or (checked_at-auth_at).days>=365:return fail('authorization','connection.observed.authorized_at','Authorization is stale or its dates are invalid.','LAUNCH_HOLD','authorization')
 for key in ('business_account_id','ad_account_id','page_id','form_id'):
  if conn.get(key)!=scope.get(key):return fail('account_form_scope','connection.observed.'+key,'Observed identity contradicts approved scope.','REPAIR_REQUIRED','account_form_scope')
 if cutoff.get('form_id')!=scope.get('form_id') or cutoff.get('ad_account_id')!=scope.get('ad_account_id'):
  return fail('account_form_scope','cutoff','Cutoff form/ad-account identity differs from approved scope.','REPAIR_REQUIRED','account_form_scope')
 for i,m in enumerate(maps):
  if m.get('required',True) and (m.get('state') not in ('mapped','OBSERVED','MAPPED') or not evidence(m.get('evidence_ref'))):return fail('field_value_map',f'field_mappings[{i}]','Required mapping lacks a dated native read-back.','LAUNCH_HOLD','field_value_map')
  vals=m.get('observed_values',[])
  if m['data_type'] in ('enum','boolean') and vals:
   approved={str(v).casefold() for v in m['allowed_values']}
   if any(str(v).casefold() not in approved for v in vals):return fail('field_value_map',f'field_mappings[{i}].observed_values','Observed enum/boolean value is outside the approved set.','REPAIR_REQUIRED','field_value_map')
 if phase=='launch':
  return fail('source_test','test','Preflight is complete; an authorized source test and resulting evidence have not been supplied.','LAUNCH_HOLD','source_test')
 # Post-test: source submission is the prerequisite for treating missing downstream stages as repair failures.
 if not (test.get('submitted_at') and test.get('test_lead_flag') is True and evidence(test.get('linkedin_evidence_ref'))):
  return fail('source_test','test.linkedin_evidence_ref','A dated LinkedIn source test is not proven for the exact record scope.','LAUNCH_HOLD','source_test')
 if not test.get('timezone') or not test.get('test_person_ref'):return fail('source_test','test','Test timezone and safe test-person reference are required.','REPAIR_REQUIRED','source_test')
 # Connector can be explicitly unknown only when the count definition says no separate receipt is exposed.
 if pops['connector_observed']['state']=='UNKNOWN' and not any(x in pops['connector_observed']['definition'].casefold() for x in ('not expose','no distinct','unavailable','not separately exposed','does not separately expose')):
  return fail('population_evidence','populations.connector_observed','Unknown connector observation lacks an explicit unavailability reason.','LAUNCH_HOLD','population_evidence')
 if match=='UNRESOLVED_COLLISION':return fail('matching','matching.observed_outcome','Observed person identity collision is unresolved.','REPAIR_REQUIRED','matching')
 if match=='UNKNOWN' or not matching.get('evidence_ref') or not evidence(matching.get('evidence_ref')):return fail('matching','matching.evidence_ref','Person-match/deduplication outcome lacks dated evidence.','REPAIR_REQUIRED','matching')
 if activity=='WRONG_ACTIVITY':return fail('person_activity','test.activity_name','Observed Marketo activity is not the required LinkedIn Lead Gen form-fill activity.','REPAIR_REQUIRED','person_activity')
 if activity in ('MISSING','UNKNOWN') or not test.get('marketo_person_id') or not test.get('activity_at') or not evidence(test.get('person_activity_evidence_ref')):return fail('person_activity','test.person_activity_evidence_ref','Source test exists but matching Marketo person/activity evidence is missing.','REPAIR_REQUIRED','person_activity')
 if membership=='ABSENT':return fail('membership','destination.membership_outcome','The observed person is absent from the intended destination.','REPAIR_REQUIRED','membership')
 if membership=='UNKNOWN' or membership=='NOT_REQUIRED' or not evidence(dest.get('evidence_ref')):return fail('membership','destination.evidence_ref','Intended destination membership lacks dated proof.','REPAIR_REQUIRED','membership')
 if route=='REJECTED':return fail('routing','routing.outcome','Observed routing receipt reports rejection or failure.','REPAIR_REQUIRED','routing')
 if route=='UNKNOWN' or route=='NOT_REQUIRED' or not evidence(routing.get('evidence_ref')):return fail('routing','routing.evidence_ref','Accepted downstream routing event lacks dated proof.','REPAIR_REQUIRED','routing')
 # Reconciliation: every required population is known; connector may be unknown when unavailable.
 for name in POPS:
  n=pops[name]
  if n['state']!='KNOWN':
   if name=='connector_observed' and n['state']=='UNKNOWN':continue
   return fail('population_evidence','populations.'+name,'Required population is blank or Unknown for this cutoff.','LAUNCH_HOLD','population_evidence')
 submitted=pops['submitted']['value'];activity_count=pops['person_activity']['value'];dupes=pops['duplicate']['value'];unresolved=pops['unresolved']['value'];members=pops['membership']['value'];routed=pops['routed']['value']
 if submitted!=activity_count+dupes+unresolved:return fail('reconciliation','populations','Submitted does not equal person/activity + duplicate + unresolved.','REPAIR_REQUIRED','reconciliation')
 if unresolved>0:return fail('reconciliation','populations.unresolved','One or more source submissions remain unresolved at the declared cutoff.','REPAIR_REQUIRED','reconciliation')
 if members>activity_count or routed>members:return fail('reconciliation','populations','Membership exceeds activity or routing exceeds membership for this cohort.','REPAIR_REQUIRED','reconciliation')
 return make_result(r,raw,phase,'VERIFIED',None,None,'All post-test states and the same-cutoff reconciliation pass.','',[],pops)

def cli():
 args=sys.argv[1:]
 phase_arg=None
 if len(args)==3 and args[0]=='--phase':phase_arg=args[1];path=Path(args[2])
 elif len(args)==1:path=Path(args[0])
 else:print('Usage: python3 marketo-integration-checker.py [--phase draft|launch|post_test] RECORD.json',file=sys.stderr);return 2
 raw=path.read_text(encoding='utf-8')
 try:record=json.loads(raw)
 except Exception as e:
  out=make_result({},raw,'draft','REPAIR_REQUIRED','import_contract','$',f'Malformed JSON: {e}',REQUESTS['import_contract'],[{'path':'$','expected_kind':'valid JSON','received_kind':'malformed JSON','message':'Input bytes are preserved in input_preserved_json.'}])
  print(json.dumps(out,ensure_ascii=False,indent=2));return 1
 if phase_arg:
  if not isinstance(record,dict):record={}
  else:record={**record,'evaluation_phase':phase_arg}
 out=check(record,raw)
 print(json.dumps(out,ensure_ascii=False,indent=2))
 return 0 if out['disposition'] in ('DRAFT_READY','LAUNCH_HOLD','VERIFIED','REPAIR_REQUIRED') else 1
if __name__=='__main__':raise SystemExit(cli())