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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
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):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if not replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
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 replay violates retry transition semantics

open(stream ID,replayable) records a live stream. goaway(last ID) monotonically lowers the accepted stream boundary. fail(id) retries only replayable streams above the known boundary (known unprocessed), otherwise reports ambiguous; unknown/already-finished streams are ignored. succeed closes a stream. New streams after GOAWAY are blocked. reconnect resets connection boundary and stream IDs while retaining decision history.

split: open-access


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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('retry')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
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):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('unreplayable')
        elif e[0]=='reconnect':
            boundary=None
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
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:
Ambiguous processing violates retry transition semantics

open(stream ID,replayable) records a live stream. goaway(last ID) monotonically lowers the accepted stream boundary. fail(id) retries only replayable streams above the known boundary (known unprocessed), otherwise reports ambiguous; unknown/already-finished streams are ignored. succeed closes a stream. New streams after GOAWAY are blocked. reconnect resets connection boundary and stream IDs while retaining decision history.

split: open-access


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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=0
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
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):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=boundary
            streams={}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
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:
Reconnect boundary violates retry transition semantics

open(stream ID,replayable) records a live stream. goaway(last ID) monotonically lowers the accepted stream boundary. fail(id) retries only replayable streams above the known boundary (known unprocessed), otherwise reports ambiguous; unknown/already-finished streams are ignored. succeed closes a stream. New streams after GOAWAY are blocked. reconnect resets connection boundary and stream IDs while retaining decision history.

split: open-access


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

N = 1
observations = []
def solve(events):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams=streams
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
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):
    streams={}; boundary=None; out=[]
    for e in events:
        if e[0]=='open':
            if boundary is not None: out.append('draining'); continue
            if e[1] in streams: out.append('duplicate'); continue
            streams[e[1]]=[e[2],True]
        elif e[0]=='goaway':
            boundary=e[1] if boundary is None else min(boundary,e[1])
        elif e[0]=='succeed':
            if e[1] in streams: streams[e[1]][1]=False
        elif e[0]=='fail':
            if e[1] not in streams: out.append('unknown'); continue
            replayable,active=streams[e[1]]
            if not active: out.append('finished'); continue
            streams[e[1]][1]=False
            if boundary is not None and e[1]>boundary:
                out.append('retry' if replayable else 'unreplayable')
            else: out.append('ambiguous')
        elif e[0]=='reconnect':
            boundary=None
            streams={k:v for k,v in streams.items() if v[1]}
    return [out,boundary]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],None])
check('1', solve([('open',1,True),('open',1,False),('fail',1),('fail',1),('fail',9)]), [['duplicate','ambiguous','finished','unknown'],None])
check('2', solve([('open',1,True),('open',3,True),('open',5,False),('goaway',3),('fail',1),('fail',3),('fail',5)]), [['ambiguous','ambiguous','unreplayable'],3])
check('3', solve([('open',3,True),('goaway',1),('goaway',5),('fail',3),('open',7,True)]), [['retry','draining'],1])
check('4', solve([('open',1,True),('goaway',0),('open',3,True),('succeed',1),('fail',1)]), [['draining','finished'],0])
check('5', solve([('open',1,True),('goaway',0),('reconnect',),('open',1,False),('goaway',0),('fail',1)]), [['unreplayable'],0])
check('6', solve([('open',N+10,True),('goaway',N+9),('fail',N+10),('fail',N+10)]), [['retry','finished'],N+9])
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:
Reconnect streams violates retry transition semantics

open(stream ID,replayable) records a live stream. goaway(last ID) monotonically lowers the accepted stream boundary. fail(id) retries only replayable streams above the known boundary (known unprocessed), otherwise reports ambiguous; unknown/already-finished streams are ignored. succeed closes a stream. New streams after GOAWAY are blocked. reconnect resets connection boundary and stream IDs while retaining decision history.

split: open-access


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

N = 1
observations = []
def solve(events):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=0; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]+1; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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:
Start size violates retry transition semantics

start(body size) starts awaiting permission. interim100 permits sending; other interim messages do not. tick permits optimistic sending after wait expiry. send(bytes) emits at most remaining body only in permitted phase. final(status) stops body transmission and records response. fail permits retry with the same full body only if zero body bytes were sent; otherwise ambiguous. retry resets the handshaking phase after retry permission. Return transmissions and decisions.

split: open-access


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

N = 1
observations = []
def solve(events):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=sent; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=1; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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:
Start consumption violates retry transition semantics

start(body size) starts awaiting permission. interim100 permits sending; other interim messages do not. tick permits optimistic sending after wait expiry. send(bytes) emits at most remaining body only in permitted phase. final(status) stops body transmission and records response. fail permits retry with the same full body only if zero body bytes were sent; otherwise ambiguous. retry resets the handshaking phase after retry permission. Return transmissions and decisions.

split: open-access


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

N = 1
observations = []
def solve(events):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]<200 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==103 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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:
Permission code violates retry transition semantics

start(body size) starts awaiting permission. interim100 permits sending; other interim messages do not. tick permits optimistic sending after wait expiry. send(bytes) emits at most remaining body only in permitted phase. final(status) stops body transmission and records response. fail permits retry with the same full body only if zero body bytes were sent; otherwise ambiguous. retry resets the handshaking phase after retry permission. Return transmissions and decisions.

split: open-access


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

N = 1
observations = []
def solve(events):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase!='idle': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase!='allowed': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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:
Permission phase violates retry transition semantics

start(body size) starts awaiting permission. interim100 permits sending; other interim messages do not. tick permits optimistic sending after wait expiry. send(bytes) emits at most remaining body only in permitted phase. final(status) stops body transmission and records response. fail permits retry with the same full body only if zero body bytes were sent; otherwise ambiguous. retry resets the handshaking phase after retry permission. Return transmissions and decisions.

split: open-access


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

N = 1
observations = []
def solve(events):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='idle': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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:
Wait expiry violates retry transition semantics

start(body size) starts awaiting permission. interim100 permits sending; other interim messages do not. tick permits optimistic sending after wait expiry. send(bytes) emits at most remaining body only in permitted phase. final(status) stops body transmission and records response. fail permits retry with the same full body only if zero body bytes were sent; otherwise ambiguous. retry resets the handshaking phase after retry permission. Return transmissions and decisions.

split: open-access


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

N = 1
observations = []
def solve(events):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase=='idle': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase=='done': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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 gate violates retry transition semantics

start(body size) starts awaiting permission. interim100 permits sending; other interim messages do not. tick permits optimistic sending after wait expiry. send(bytes) emits at most remaining body only in permitted phase. final(status) stops body transmission and records response. fail permits retry with the same full body only if zero body bytes were sent; otherwise ambiguous. retry resets the handshaking phase after retry permission. Return transmissions and decisions.

split: open-access


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

N = 1
observations = []
def solve(events):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=e[1]
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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:
Remaining body violates retry transition semantics

start(body size) starts awaiting permission. interim100 permits sending; other interim messages do not. tick permits optimistic sending after wait expiry. send(bytes) emits at most remaining body only in permitted phase. final(status) stops body transmission and records response. fail permits retry with the same full body only if zero body bytes were sent; otherwise ambiguous. retry resets the handshaking phase after retry permission. Return transmissions and decisions.

split: open-access


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

N = 1
observations = []
def solve(events):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=1
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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 accumulation violates retry transition semantics

start(body size) starts awaiting permission. interim100 permits sending; other interim messages do not. tick permits optimistic sending after wait expiry. send(bytes) emits at most remaining body only in permitted phase. final(status) stops body transmission and records response. fail permits retry with the same full body only if zero body bytes were sent; otherwise ambiguous. retry resets the handshaking phase after retry permission. Return transmissions and decisions.

split: open-access


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

N = 1
observations = []
def solve(events):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='allowed'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='waiting'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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:
Final stops violates retry transition semantics

start(body size) starts awaiting permission. interim100 permits sending; other interim messages do not. tick permits optimistic sending after wait expiry. send(bytes) emits at most remaining body only in permitted phase. final(status) stops body transmission and records response. fail permits retry with the same full body only if zero body bytes were sent; otherwise ambiguous. retry resets the handshaking phase after retry permission. Return transmissions and decisions.

split: open-access


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

N = 1
observations = []
def solve(events):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',200])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',100])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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:
Final response violates retry transition semantics

start(body size) starts awaiting permission. interim100 permits sending; other interim messages do not. tick permits optimistic sending after wait expiry. send(bytes) emits at most remaining body only in permitted phase. final(status) stops body transmission and records response. fail permits retry with the same full body only if zero body bytes were sent; otherwise ambiguous. retry resets the handshaking phase after retry permission. Return transmissions and decisions.

split: open-access


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

N = 1
observations = []
def solve(events):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if False: out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('retry'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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:
Failure terminal violates retry transition semantics

start(body size) starts awaiting permission. interim100 permits sending; other interim messages do not. tick permits optimistic sending after wait expiry. send(bytes) emits at most remaining body only in permitted phase. final(status) stops body transmission and records response. fail permits retry with the same full body only if zero body bytes were sent; otherwise ambiguous. retry resets the handshaking phase after retry permission. Return transmissions and decisions.

split: open-access


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

N = 1
observations = []
def solve(events):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent<=1 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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 ambiguity violates retry transition semantics

start(body size) starts awaiting permission. interim100 permits sending; other interim messages do not. tick permits optimistic sending after wait expiry. send(bytes) emits at most remaining body only in permitted phase. final(status) stops body transmission and records response. fail permits retry with the same full body only if zero body bytes were sent; otherwise ambiguous. retry resets the handshaking phase after retry permission. Return transmissions and decisions.

split: open-access


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

N = 1
observations = []
def solve(events):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase=='done': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase=='idle': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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:
Retry gate violates retry transition semantics

start(body size) starts awaiting permission. interim100 permits sending; other interim messages do not. tick permits optimistic sending after wait expiry. send(bytes) emits at most remaining body only in permitted phase. final(status) stops body transmission and records response. fail permits retry with the same full body only if zero body bytes were sent; otherwise ambiguous. retry resets the handshaking phase after retry permission. Return transmissions and decisions.

split: open-access


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

N = 1
observations = []
def solve(events):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='allowed'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='waiting'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='done'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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:
Retry handshake violates retry transition semantics

start(body size) starts awaiting permission. interim100 permits sending; other interim messages do not. tick permits optimistic sending after wait expiry. send(bytes) emits at most remaining body only in permitted phase. final(status) stops body transmission and records response. fail permits retry with the same full body only if zero body bytes were sent; otherwise ambiguous. retry resets the handshaking phase after retry permission. Return transmissions and decisions.

split: open-access


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

N = 1
observations = []
def solve(events):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='allowed'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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):
    size=0; sent=0; phase='idle'; out=[]
    for e in events:
        if e[0]=='start':
            size=e[1]; sent=0; phase='idle'
        elif e[0]=='interim':
            if e[1]==100 and phase=='waiting': phase='allowed'
        elif e[0]=='tick':
            if phase=='waiting': phase='allowed'
        elif e[0]=='send':
            if phase!='allowed': out.append('blocked'); continue
            count=min(e[1],size-sent)
            sent+=count
            out.append(count)
        elif e[0]=='final':
            phase='done'; out.append(['response',e[1]])
        elif e[0]=='fail':
            if phase=='done': out.append('terminal'); continue
            phase='retry' if sent==0 else 'ambiguous'
            out.append(phase)
        elif e[0]=='retry':
            if phase!='retry': out.append('denied'); continue
            sent=0; phase='waiting'; out.append('restarted')
    return [out,sent]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0])
check('1', solve([('start',N+3),('send',1),('interim',103),('send',1),('interim',100),('send',2),('send',N+3)]), [['blocked','blocked',2,N+1],N+3])
check('2', solve([('start',N),('fail',),('retry',),('send',1),('tick',),('send',N)]), [['retry','restarted','blocked',N],N])
check('3', solve([('start',2),('tick',),('send',1),('fail',),('retry',)]), [[1,'ambiguous','denied'],1])
check('4', solve([('start',N),('final',417),('interim',100),('tick',),('send',1),('fail',),('retry',)]), [[['response',417],'blocked','terminal','denied'],0])
check('5', solve([('start',2),('tick',),('send',2),('start',N),('fail',)]), [[2,'retry'],0])
check('6', solve([('start',0),('interim',100),('send',3),('send',0)]), [[0,0],0])
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:
Initial handshake violates retry transition semantics

start(body size) starts awaiting permission. interim100 permits sending; other interim messages do not. tick permits optimistic sending after wait expiry. send(bytes) emits at most remaining body only in permitted phase. final(status) stops body transmission and records response. fail permits retry with the same full body only if zero body bytes were sent; otherwise ambiguous. retry resets the handshaking phase after retry permission. Return transmissions and decisions.

split: open-access


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

N = 1
observations = []
def solve(events):
    capacity=0; available=0; held=set(); out=[]
    for e in events:
        if e[0]=='configure':
            capacity=e[1]; available=0; held.clear()
        elif e[0]=='reserve':
            if e[1] in held: out.append('held'); continue
            if available==0: out.append('denied'); continue
            available-=1
            held.add(e[1])
            out.append('reserved')
        elif e[0]=='commit':
            if e[1] in held:
                held.remove(e[1])
        elif e[0]=='abort':
            if e[1] in held:
                held.remove(e[1])
                available=min(capacity,available+1)
        elif e[0]=='refill':
            available=min(capacity-len(held),available+e[1])
    return [out,available,sorted(held)]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([('configure',3),('reserve','a'),('reserve','b'),('abort','a')]), [['reserved','reserved'],2,['b']])
check('1', solve([('configure',3),('reserve','a'),('commit','a'),('reserve','a'),('commit','a'),('reserve','a'),('commit','a'),('refill',2)]), [['reserved','reserved','reserved'],2,[]])
check('2', solve([]), [[],0,[]])
check('3', solve([('configure',0),('reserve','a')]), [['denied'],0,[]])
check('4', solve([('configure',2),('reserve','a'),('reserve','a'),('reserve','b'),('reserve','c')]), [['reserved','held','reserved','denied'],0,['a','b']])
check('5', solve([('configure',3),('reserve','a'),('reserve','b'),('commit','a'),('abort','a'),('abort','b'),('abort','b')]), [['reserved','reserved'],2,[]])
check('6', solve([('configure',N+3),('reserve','a'),('refill',N+10)]), [['reserved'],N+2,['a']])
check('7', solve([('configure',4),('reserve','a'),('commit','a'),('reserve','b'),('commit','b'),('refill',1)]), [['reserved','reserved'],3,[]])
check('8', solve([('configure',1),('reserve','a'),('configure',2),('reserve','a')]), [['reserved','reserved'],1,['a']])
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):
    capacity=0; available=0; held=set(); out=[]
    for e in events:
        if e[0]=='configure':
            capacity=e[1]; available=capacity+1; held.clear()
        elif e[0]=='reserve':
            if e[1] in held: out.append('held'); continue
            if available==0: out.append('denied'); continue
            available-=1
            held.add(e[1])
            out.append('reserved')
        elif e[0]=='commit':
            if e[1] in held:
                held.remove(e[1])
        elif e[0]=='abort':
            if e[1] in held:
                held.remove(e[1])
                available=min(capacity,available+1)
        elif e[0]=='refill':
            available=min(capacity-len(held),available+e[1])
    return [out,available,sorted(held)]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([('configure',3),('reserve','a'),('reserve','b'),('abort','a')]), [['reserved','reserved'],2,['b']])
check('1', solve([('configure',3),('reserve','a'),('commit','a'),('reserve','a'),('commit','a'),('reserve','a'),('commit','a'),('refill',2)]), [['reserved','reserved','reserved'],2,[]])
check('2', solve([]), [[],0,[]])
check('3', solve([('configure',0),('reserve','a')]), [['denied'],0,[]])
check('4', solve([('configure',2),('reserve','a'),('reserve','a'),('reserve','b'),('reserve','c')]), [['reserved','held','reserved','denied'],0,['a','b']])
check('5', solve([('configure',3),('reserve','a'),('reserve','b'),('commit','a'),('abort','a'),('abort','b'),('abort','b')]), [['reserved','reserved'],2,[]])
check('6', solve([('configure',N+3),('reserve','a'),('refill',N+10)]), [['reserved'],N+2,['a']])
check('7', solve([('configure',4),('reserve','a'),('commit','a'),('reserve','b'),('commit','b'),('refill',1)]), [['reserved','reserved'],3,[]])
check('8', solve([('configure',1),('reserve','a'),('configure',2),('reserve','a')]), [['reserved','reserved'],1,['a']])
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:
Initial fill violates retry transition semantics

configure(capacity) resets a shared retry pool. reserve(request) consumes one token only for a new reservation; commit spends its token and removes the reservation, abort refunds exactly once, refill is capped. Repeated reserve is idempotent; unknown commit/abort does nothing. Return admissions and available token count.

split: open-access