scripts/cost/compliance.py

Derived analysis output (scratchpad scripts/cost/compliance.py).

This is a derived table/summary; the finding's excerpt is usually a paraphrase of its counts, so it may not be highlighted verbatim.

"""Addressed norm requests -> did the addressee (by cohort tag, same family) later commit / act?  Tactic effects.
Also split 'pushback' into third-person clauses inside requests vs first-person self-protection."""
import json,re,collections,csv,datetime as dt
S='/tmp/claude-0/-root-swarm-hackathon/a8e4c17a-c034-48ae-9f11-70fc0a8841c9/scratchpad'
R=sorted((json.loads(l) for l in open(S+'/derived/cost/norm_lines.jsonl')),key=lambda r:r['t'])
ALL=sorted((json.loads(l) for l in open(S+'/added_lines.jsonl')),key=lambda r:r['t'])
MON=r'(Jan|Feb|Mar|Apr|May|Jun|Jul|Aug|Sep|Oct|Nov|Dec)[a-z]*[ _-]?(\d{1,2})(?!\d)'
def T(s): return dt.datetime.fromisoformat(s[:19])
VER={'OpenAiResearcherJul23':'Jul23','OpenAI-Dec27':'Dec27','GroceryOurApr20X':'Apr20','ResearchAgentJun19X':'Jun19','OpenAIResearchJun13X':'Jun13',
 'CashierCoordSep01OAI':'Sep01','CashierCoordMar20OAI':'Mar20','CashierCoordAug07OAI':'Aug07','CashierCoordFeb28OAI':'Feb28','OpenAINov27FP':'Nov27','OpenAIResearchSep01X':'Sep01'}
# ---- pushback split
pb=[r for r in R if 'pushback' in r['acts']]
third=[r for r in pb if re.search(r'\bplease\b|@|\byour\b|\byou\b',r['line'],re.I)]
first=[r for r in pb if r not in third and re.search(r'\b(we|I|our)\b',r['line'],re.I)]
fp_answer_first=[r for r in R if re.search(r"\b(I|we)\b( will|'ll)? answer[^.;]{0,30}(first|then signal|before (signal|post))|(we|I) will (signal|post|pre-signal)[^.;]{0,60}only if safe|priority (is )?(our|my) answer",r['line'],re.I)]
print('pushback lines',len(pb),'third-person (inside requests to others)',len(third),'first-person/other',len(pb)-len(third))
print('first-person answer-first / only-if-safe lines:',len(fp_answer_first))
for r in fp_answer_first: print('  ',r['t'][5:16],r['family'],r['pos'],r['sig'],'|',r['line'][:170])
# hedge rate in first-person commitments
com=[r for r in R if 'commit' in r['acts']]
print('first-person commitments',len(com),'hedged',sum(r['hedge'] for r in com))
# ---- addressed requests
def cohort_of(line,sig):
    m=re.search(MON,sig or '')
    if m: return m.group(1)[:3].title()+m.group(2).zfill(2)
    return None
# index lines by cohort tag (writer) : use signature cohort or leading tag
by_coh=collections.defaultdict(list)
for r in R:
    if r['cohort']: by_coh[(r['cohort'],r['family'])].append(r)
reqs=[r for r in R if r['primary'] in('request','request+reciprocity','commit+request','pushback') and r['addr_coh'] and ('request' in r['acts'] or 'pushback' in r['acts'])]
rows=[]
for q in reqs:
    for a in q['addr_coh']:
        later=[x for x in by_coh.get((a,q['family']),[]) if q['t']<x['t']<=(T(q['t'])+dt.timedelta(hours=6)).isoformat()]
        prior=[x for x in by_coh.get((a,q['family']),[]) if x['t']<q['t']]
        seen_any=bool(later)
        comp_commit=any('commit' in x['acts'] for x in later)
        comp_act=any(x['primary']=='act' and x['sig'] in VER for x in later) or any(x['primary']=='act' and x['cohort']==a and re.search(r'BEFORE final|before final|STATE5-[A-Z]{2} CONF|SIGNAL:',x['line']) for x in later)
        refuse=any('pushback' in x['acts'] and not re.search(r'please|your|@',x['line'],re.I) for x in later)
        rows.append(dict(t=q['t'],rev=q['rev'],frm=q['sig'],frm_pos=q['pos'],to=a,family=q['family'],tactics='|'.join(q['tactics']),n_tactics=len(q['tactics']),
            addressee_active_after=int(seen_any),addressee_commit=int(comp_commit),addressee_act=int(comp_act),addressee_selfprotect=int(refuse),line=q['line'][:220]))
with open(S+'/derived/cost/addressed_norm_requests.csv','w',newline='') as f:
    w=csv.DictWriter(f,fieldnames=list(rows[0])); w.writeheader(); w.writerows(rows)
print('\naddressed norm requests (pairs)',len(rows),'addressee active later',sum(r['addressee_active_after'] for r in rows),
      'commit',sum(r['addressee_commit'] for r in rows),'act',sum(r['addressee_act'] for r in rows))
A=[r for r in rows if r['addressee_active_after']]
base=sum(r['addressee_commit'] or r['addressee_act'] for r in A)/len(A)
print('base compliance (commit or act | addressee active)',round(base,3),'n',len(A))
tac=['urgency','group_appeal','deadline','role_flattery','stakes_guilt','reciprocity','cost_minimise']
out=[]
for t in tac:
    w=[r for r in A if t in r['tactics'].split('|')]; wo=[r for r in A if t not in r['tactics'].split('|')]
    if not w or not wo: continue
    pw=sum(r['addressee_commit'] or r['addressee_act'] for r in w)/len(w); pwo=sum(r['addressee_commit'] or r['addressee_act'] for r in wo)/len(wo)
    # Fisher exact (stdlib)
    a=sum(r['addressee_commit'] or r['addressee_act'] for r in w); b=len(w)-a; c=sum(r['addressee_commit'] or r['addressee_act'] for r in wo); d=len(wo)-c
    from math import comb
    n=a+b+c+d; r1=a+b; c1=a+c
    def p(x): return comb(r1,x)*comb(n-r1,c1-x)/comb(n,c1)
    p0=p(a); pv=sum(p(x) for x in range(max(0,c1-(n-r1)),min(r1,c1)+1) if p(x)<=p0*1.0000001)
    out.append(dict(tactic=t,n_with=len(w),comply_with=round(pw,3),n_without=len(wo),comply_without=round(pwo,3),fisher_p=round(pv,4)))
    print(t,len(w),round(pw,2),'vs',len(wo),round(pwo,2),'p=',round(pv,3))
# position of addressee at time of request? (last known pos before request)
for r in A: pass
with open(S+'/derived/cost/tactic_compliance.csv','w',newline='') as f:
    w=csv.DictWriter(f,fieldnames=list(out[0])); w.writeheader(); w.writerows(out)
# requests by position: which tactics
print('\ntactic prevalence in requests by requester position:')
for pos in ('far_behind','one_behind','at_final',None):
    Q=[r for r in R if r['primary'] in('request','request+reciprocity','commit+request') and r['pos']==pos]
    c=collections.Counter(t for r in Q for t in r['tactics'])
    print(pos,len(Q),{t:round(c[t]/len(Q),2) for t in tac})
# payers: were they addressed before acting?
print('\nverified payers addressed by name/cohort before their act:')
for s,c in VER.items():
    hits=[r for r in rows if r['to']==c and r['t']<'2026-06-22']
    print(' ',s,c,len(hits),[ (h['t'][5:16],h['frm'],h['tactics']) for h in hits[:4]])