broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: requests.pop(e[1])
            requests[e[1]]=[current,0]
        elif e[0]=='finish': requests.pop(e[1],None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append('unknown'); continue
            policy,used=requests[e[1]]
            version,cap,statuses=policy
            if e[2] not in statuses: out.append(['stop',version]); continue
            if used>=cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('unknown'); continue
            requests[e[1]]=[current,0]
        elif e[0]=='finish': requests.pop(e[1],None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append('unknown'); continue
            policy,used=requests[e[1]]
            version,cap,statuses=policy
            if e[2] not in statuses: out.append(['stop',version]); continue
            if used>=cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Duplicate begin violates retry transition semantics

configure(version,cap,statuses) sets current policy. begin(request) snapshots it with zero retries used; failure(request,status) uses that request snapshot and admits retry only for configured statuses below cap. update configuration affects later requests only. finish removes request; unknown failures report unknown. Return retry decisions with pinned policy version.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('duplicate'); continue
            requests[e[1]]=[current,1]
        elif e[0]=='finish': requests.pop(e[1],None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append('unknown'); continue
            policy,used=requests[e[1]]
            version,cap,statuses=policy
            if e[2] not in statuses: out.append(['stop',version]); continue
            if used>=cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

meaning: Fixture-count band, not measured model difficulty

deps: stdlib

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('duplicate'); continue
            requests[e[1]]=[current,current[1]]
        elif e[0]=='finish': requests.pop(e[1],None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append('unknown'); continue
            policy,used=requests[e[1]]
            version,cap,statuses=policy
            if e[2] not in statuses: out.append(['stop',version]); continue
            if used>=cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Zero usage violates retry transition semantics

configure(version,cap,statuses) sets current policy. begin(request) snapshots it with zero retries used; failure(request,status) uses that request snapshot and admits retry only for configured statuses below cap. update configuration affects later requests only. finish removes request; unknown failures report unknown. Return retry decisions with pinned policy version.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('duplicate'); continue
            requests["global"]=[current,0]
        elif e[0]=='finish': requests.pop(e[1],None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append('unknown'); continue
            policy,used=requests[e[1]]
            version,cap,statuses=policy
            if e[2] not in statuses: out.append(['stop',version]); continue
            if used>=cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('duplicate'); continue
            requests[str(e[1])+"x"]=[current,0]
        elif e[0]=='finish': requests.pop(e[1],None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append('unknown'); continue
            policy,used=requests[e[1]]
            version,cap,statuses=policy
            if e[2] not in statuses: out.append(['stop',version]); continue
            if used>=cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Request identity violates retry transition semantics

configure(version,cap,statuses) sets current policy. begin(request) snapshots it with zero retries used; failure(request,status) uses that request snapshot and admits retry only for configured statuses below cap. update configuration affects later requests only. finish removes request; unknown failures report unknown. Return retry decisions with pinned policy version.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('duplicate'); continue
            requests[e[1]]=[current,0]
        elif e[0]=='finish': requests.pop("global",None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append('unknown'); continue
            policy,used=requests[e[1]]
            version,cap,statuses=policy
            if e[2] not in statuses: out.append(['stop',version]); continue
            if used>=cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('duplicate'); continue
            requests[e[1]]=[current,0]
        elif e[0]=='finish': requests.clear()
        elif e[0]=='failure':
            if e[1] not in requests: out.append('unknown'); continue
            policy,used=requests[e[1]]
            version,cap,statuses=policy
            if e[2] not in statuses: out.append(['stop',version]); continue
            if used>=cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Finish removal violates retry transition semantics

configure(version,cap,statuses) sets current policy. begin(request) snapshots it with zero retries used; failure(request,status) uses that request snapshot and admits retry only for configured statuses below cap. update configuration affects later requests only. finish removes request; unknown failures report unknown. Return retry decisions with pinned policy version.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('duplicate'); continue
            requests[e[1]]=[current,0]
        elif e[0]=='finish': requests.pop(e[1],None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append(['retry',current[0]]); continue
            policy,used=requests[e[1]]
            version,cap,statuses=policy
            if e[2] not in statuses: out.append(['stop',version]); continue
            if used>=cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('duplicate'); continue
            requests[e[1]]=[current,0]
        elif e[0]=='finish': requests.pop(e[1],None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append(['stop',current[0]]); continue
            policy,used=requests[e[1]]
            version,cap,statuses=policy
            if e[2] not in statuses: out.append(['stop',version]); continue
            if used>=cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Unknown failure violates retry transition semantics

configure(version,cap,statuses) sets current policy. begin(request) snapshots it with zero retries used; failure(request,status) uses that request snapshot and admits retry only for configured statuses below cap. update configuration affects later requests only. finish removes request; unknown failures report unknown. Return retry decisions with pinned policy version.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('duplicate'); continue
            requests[e[1]]=[current,0]
        elif e[0]=='finish': requests.pop(e[1],None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append('unknown'); continue
            policy,used=requests[e[1]]
            version,cap,statuses=current
            if e[2] not in statuses: out.append(['stop',version]); continue
            if used>=cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('duplicate'); continue
            requests[e[1]]=[current,0]
        elif e[0]=='finish': requests.pop(e[1],None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append('unknown'); continue
            policy,used=requests[e[1]]
            version,cap,statuses=(policy[0],current[1],current[2])
            if e[2] not in statuses: out.append(['stop',version]); continue
            if used>=cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Policy pinning violates retry transition semantics

configure(version,cap,statuses) sets current policy. begin(request) snapshots it with zero retries used; failure(request,status) uses that request snapshot and admits retry only for configured statuses below cap. update configuration affects later requests only. finish removes request; unknown failures report unknown. Return retry decisions with pinned policy version.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('duplicate'); continue
            requests[e[1]]=[current,0]
        elif e[0]=='finish': requests.pop(e[1],None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append('unknown'); continue
            policy,used=requests[e[1]]
            version,cap,statuses=policy
            if e[2] in statuses: out.append(['stop',version]); continue
            if used>=cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('duplicate'); continue
            requests[e[1]]=[current,0]
        elif e[0]=='finish': requests.pop(e[1],None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append('unknown'); continue
            policy,used=requests[e[1]]
            version,cap,statuses=policy
            if e[2]<500: out.append(['stop',version]); continue
            if used>=cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Status membership violates retry transition semantics

configure(version,cap,statuses) sets current policy. begin(request) snapshots it with zero retries used; failure(request,status) uses that request snapshot and admits retry only for configured statuses below cap. update configuration affects later requests only. finish removes request; unknown failures report unknown. Return retry decisions with pinned policy version.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('duplicate'); continue
            requests[e[1]]=[current,0]
        elif e[0]=='finish': requests.pop(e[1],None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append('unknown'); continue
            policy,used=requests[e[1]]
            version,cap,statuses=policy
            if e[2] not in statuses: out.append(['stop',version]); continue
            if used>cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('duplicate'); continue
            requests[e[1]]=[current,0]
        elif e[0]=='finish': requests.pop(e[1],None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append('unknown'); continue
            policy,used=requests[e[1]]
            version,cap,statuses=policy
            if e[2] not in statuses: out.append(['stop',version]); continue
            if used>=cap+1: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Cap boundary violates retry transition semantics

configure(version,cap,statuses) sets current policy. begin(request) snapshots it with zero retries used; failure(request,status) uses that request snapshot and admits retry only for configured statuses below cap. update configuration affects later requests only. finish removes request; unknown failures report unknown. Return retry decisions with pinned policy version.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('duplicate'); continue
            requests[e[1]]=[current,0]
        elif e[0]=='finish': requests.pop(e[1],None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append('unknown'); continue
            policy,used=requests[e[1]]
            version,cap,statuses=policy
            if e[2] not in statuses: out.append(['stop',version]); continue
            if used>=cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]=1
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    current=(0,0,()); requests={}; out=[]
    for e in events:
        if e[0]=='configure':
            current=(e[1],e[2],tuple(e[3]))
        elif e[0]=='begin':
            if e[1] in requests: out.append('duplicate'); continue
            requests[e[1]]=[current,0]
        elif e[0]=='finish': requests.pop(e[1],None)
        elif e[0]=='failure':
            if e[1] not in requests: out.append('unknown'); continue
            policy,used=requests[e[1]]
            version,cap,statuses=policy
            if e[2] not in statuses: out.append(['stop',version]); continue
            if used>=cap: out.append(['exhausted',version]); continue
            requests[e[1]][1]+=2
            out.append(['retry',version])
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('failure','x',503)]), ['unknown'])
check('2', solve([('configure',N,2,[503,429]),('begin','a'),('failure','a',503),('failure','a',429),('failure','a',503)]), [['retry',N],['retry',N],['exhausted',N]])
check('3', solve([('configure',N,1,[503]),('begin','a'),('configure',N+1,0,[429]),('failure','a',429),('failure','a',503),('failure','a',503),('begin','b'),('failure','b',429)]), [['stop',N],['retry',N],['exhausted',N],['exhausted',N+1]])
check('4', solve([('configure',N,1,[503]),('begin','a'),('failure','a',503),('begin','a'),('failure','a',503)]), [['retry',N],'duplicate',['exhausted',N]])
check('5', solve([('configure',N,1,[503]),('begin','a'),('begin','b'),('finish','a'),('failure','a',503),('failure','b',503)]), ['unknown',['retry',N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Charge request violates retry transition semantics

configure(version,cap,statuses) sets current policy. begin(request) snapshots it with zero retries used; failure(request,status) uses that request snapshot and admits retry only for configured statuses below cap. update configuration affects later requests only. finish removes request; unknown failures report unknown. Return retry decisions with pinned policy version.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=0; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]+1; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Precondition capture violates retry transition semantics

begin(version,desired) creates a pending conditional write. sent records that it may have reached the server; lost means response absent. inspect(version,value) after loss reconciles: desired value at version+1 is committed, unchanged original version permits retry, any other state conflicts. retry only from retry state retains original precondition and desired body. ack(value) succeeds only for an active sent request. Return decisions and outgoing retry preconditions.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[1]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=None; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Body capture violates retry transition semantics

begin(version,desired) creates a pending conditional write. sent records that it may have reached the server; lost means response absent. inspect(version,value) after loss reconciles: desired value at version+1 is committed, unchanged original version permits retry, any other state conflicts. retry only from retry state retains original precondition and desired body. ack(value) succeeds only for an active sent request. Return decisions and outgoing retry preconditions.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='sent'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='idle'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Begin state violates retry transition semantics

begin(version,desired) creates a pending conditional write. sent records that it may have reached the server; lost means response absent. inspect(version,value) after loss reconciles: desired value at version+1 is committed, unchanged original version permits retry, any other state conflicts. retry only from retry state retains original precondition and desired body. ack(value) succeeds only for an active sent request. Return decisions and outgoing retry preconditions.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='ready'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='idle': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Send state violates retry transition semantics

begin(version,desired) creates a pending conditional write. sent records that it may have reached the server; lost means response absent. inspect(version,value) after loss reconciles: desired value at version+1 is committed, unchanged original version permits retry, any other state conflicts. retry only from retry state retains original precondition and desired body. ack(value) succeeds only for an active sent request. Return decisions and outgoing retry preconditions.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state!='idle': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Lost after send violates retry transition semantics

begin(version,desired) creates a pending conditional write. sent records that it may have reached the server; lost means response absent. inspect(version,value) after loss reconciles: desired value at version+1 is committed, unchanged original version permits retry, any other state conflicts. retry only from retry state retains original precondition and desired body. ack(value) succeeds only for an active sent request. Return decisions and outgoing retry preconditions.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state=='idle': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state=='ready': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Inspection gate violates retry transition semantics

begin(version,desired) creates a pending conditional write. sent records that it may have reached the server; lost means response absent. inspect(version,value) after loss reconciles: desired value at version+1 is committed, unchanged original version permits retry, any other state conflicts. retry only from retry state retains original precondition and desired body. ack(value) succeeds only for an active sent request. Return decisions and outgoing retry preconditions.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]>=version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Successor version violates retry transition semantics

begin(version,desired) creates a pending conditional write. sent records that it may have reached the server; lost means response absent. inspect(version,value) after loss reconciles: desired value at version+1 is committed, unchanged original version permits retry, any other state conflicts. retry only from retry state retains original precondition and desired body. ack(value) succeeds only for an active sent request. Return decisions and outgoing retry preconditions.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and True:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2] is not None:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Same body violates retry transition semantics

begin(version,desired) creates a pending conditional write. sent records that it may have reached the server; lost means response absent. inspect(version,value) after loss reconciles: desired value at version+1 is committed, unchanged original version permits retry, any other state conflicts. retry only from retry state retains original precondition and desired body. ack(value) succeeds only for an active sent request. Return decisions and outgoing retry preconditions.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='uncertain'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='retry'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Reconciled terminal violates retry transition semantics

begin(version,desired) creates a pending conditional write. sent records that it may have reached the server; lost means response absent. inspect(version,value) after loss reconciles: desired value at version+1 is committed, unchanged original version permits retry, any other state conflicts. retry only from retry state retains original precondition and desired body. ack(value) succeeds only for an active sent request. Return decisions and outgoing retry preconditions.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]<=version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]>=version:
                state='retry'; out.append('retryable')
            else: state='conflict'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Unchanged version violates retry transition semantics

begin(version,desired) creates a pending conditional write. sent records that it may have reached the server; lost means response absent. inspect(version,value) after loss reconciles: desired value at version+1 is committed, unchanged original version permits retry, any other state conflicts. retry only from retry state retains original precondition and desired body. ack(value) succeeds only for an active sent request. Return decisions and outgoing retry preconditions.

split: open-access


broken_source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='retry'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

source:
"""Failure Map reference implementation. Python standard library only."""
import json

N = 1
observations = []
def solve(events):
    version=0; desired=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            version=e[1]; desired=e[2]; state='ready'
        elif e[0]=='sent':
            if state=='ready': state='sent'
        elif e[0]=='lost':
            if state=='sent': state='uncertain'
        elif e[0]=='inspect':
            if state!='uncertain': out.append('ignored'); continue
            if e[1]==version+1 and e[2]==desired:
                state='done'; out.append('committed')
            elif e[1]==version:
                state='retry'; out.append('retryable')
            else: state='uncertain'; out.append('conflict')
        elif e[0]=='retry':
            if state!='retry': out.append('denied'); continue
            out.append(['send',version,desired]); state='ready'
        elif e[0]=='ack':
            if state=='sent': state='done'; out.append(['success',e[1]])
            else: out.append('ignored')
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N+1),('retry',),('retry',),('sent',),('ack',N+9)]), ['retryable',['send',N,N+9],'denied',['success',N+9]])
check('2', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+9),('retry',),('inspect',N,N)]), ['committed','denied','ignored'])
check('3', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+1,N+8),('retry',),('inspect',N,N)]), ['conflict','denied','ignored'])
check('4', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N+2,N+9),('retry',)]), ['conflict','denied'])
check('5', solve([('begin',N,N+9),('lost',),('inspect',N,N),('ack',N)]), ['ignored','ignored'])
check('6', solve([('begin',N,N+9),('sent',),('lost',),('ack',N),('inspect',N-1,N),('retry',)]), ['ignored','conflict','denied'])
check('7', solve([('begin',N,N+9),('sent',),('lost',),('inspect',N,N),('retry',),('ack',N)]), ['retryable',['send',N,N+9],'ignored'])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Conflict terminal violates retry transition semantics

begin(version,desired) creates a pending conditional write. sent records that it may have reached the server; lost means response absent. inspect(version,value) after loss reconciles: desired value at version+1 is committed, unchanged original version permits retry, any other state conflicts. retry only from retry state retains original precondition and desired body. ack(value) succeeds only for an active sent request. Return decisions and outgoing retry preconditions.

split: open-access