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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if not e[1] else e[1]
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1]==0 else e[1]
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Explicit empty key violates retry transition semantics

begin(provided key or None,body) starts a logical operation and advances a local operation serial. Generated key is operation serial; explicit keys including zero remain exact. send(body) first sends only matching snapshotted body. uncertain authorizes a resend; reject is terminal; success is terminal. Retried sends reuse the initial key. Return sent key/body pairs or refusal reason.

split: open-access


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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else serial
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

meaning: Fixture-count band, not measured model difficulty

deps: stdlib

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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else str(e[1])
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Explicit key preservation violates retry transition semantics

begin(provided key or None,body) starts a logical operation and advances a local operation serial. Generated key is operation serial; explicit keys including zero remain exact. send(body) first sends only matching snapshotted body. uncertain authorizes a resend; reject is terminal; success is terminal. Retried sends reuse the initial key. Return sent key/body pairs or refusal reason.

split: open-access


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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[1]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=None
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Body snapshot violates retry transition semantics

begin(provided key or None,body) starts a logical operation and advances a local operation serial. Generated key is operation serial; explicit keys including zero remain exact. send(body) first sends only matching snapshotted body. uncertain authorizes a resend; reject is terminal; success is terminal. Retried sends reuse the initial key. Return sent key/body pairs or refusal reason.

split: open-access


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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[2]
            state='sent'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[2]
            state=state
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
New operation state violates retry transition semantics

begin(provided key or None,body) starts a logical operation and advances a local operation serial. Generated key is operation serial; explicit keys including zero remain exact. send(body) first sends only matching snapshotted body. uncertain authorizes a resend; reject is terminal; success is terminal. Retried sends reuse the initial key. Return sent key/body pairs or refusal reason.

split: open-access


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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state!='ready': out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state!='uncertain': out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Uncertain replay violates retry transition semantics

begin(provided key or None,body) starts a logical operation and advances a local operation serial. Generated key is operation serial; explicit keys including zero remain exact. send(body) first sends only matching snapshotted body. uncertain authorizes a resend; reject is terminal; success is terminal. Retried sends reuse the initial key. Return sent key/body pairs or refusal reason.

split: open-access


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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if False: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original and state=='ready': out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Body binding violates retry transition semantics

begin(provided key or None,body) starts a logical operation and advances a local operation serial. Generated key is operation serial; explicit keys including zero remain exact. send(body) first sends only matching snapshotted body. uncertain authorizes a resend; reject is terminal; success is terminal. Retried sends reuse the initial key. Return sent key/body pairs or refusal reason.

split: open-access


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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([serial+len(out),original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([serial,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Stable retry key violates retry transition semantics

begin(provided key or None,body) starts a logical operation and advances a local operation serial. Generated key is operation serial; explicit keys including zero remain exact. send(body) first sends only matching snapshotted body. uncertain authorizes a resend; reject is terminal; success is terminal. Retried sends reuse the initial key. Return sent key/body pairs or refusal reason.

split: open-access


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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='ready'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='uncertain'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Admission consumption violates retry transition semantics

begin(provided key or None,body) starts a logical operation and advances a local operation serial. Generated key is operation serial; explicit keys including zero remain exact. send(body) first sends only matching snapshotted body. uncertain authorizes a resend; reject is terminal; success is terminal. Retried sends reuse the initial key. Return sent key/body pairs or refusal reason.

split: open-access


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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state!='idle': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Uncertainty scope violates retry transition semantics

begin(provided key or None,body) starts a logical operation and advances a local operation serial. Generated key is operation serial; explicit keys including zero remain exact. send(body) first sends only matching snapshotted body. uncertain authorizes a resend; reject is terminal; success is terminal. Retried sends reuse the initial key. Return sent key/body pairs or refusal reason.

split: open-access


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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='uncertain'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='ready'
        elif e[0]=='success':
            if state=='sent': state='done'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Reject terminal violates retry transition semantics

begin(provided key or None,body) starts a logical operation and advances a local operation serial. Generated key is operation serial; explicit keys including zero remain exact. send(body) first sends only matching snapshotted body. uncertain authorizes a resend; reject is terminal; success is terminal. Retried sends reuse the initial key. Return sent key/body pairs or refusal reason.

split: open-access


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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='uncertain'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(events):
    serial=0; key=None; original=None; state='idle'; out=[]
    for e in events:
        if e[0]=='begin':
            serial+=1
            key=serial if e[1] is None else e[1]
            original=e[2]
            state='ready'
        elif e[0]=='send':
            if state not in ('ready','uncertain'): out.append('blocked'); continue
            if e[1]!=original: out.append('body-conflict'); continue
            out.append([key,original])
            state='sent'
        elif e[0]=='uncertain':
            if state=='sent': state='uncertain'
        elif e[0]=='reject': state='rejected'
        elif e[0]=='success':
            if state=='sent': state='ready'
    return out
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [])
check('1', solve([('begin',None,N),('send',N),('uncertain',),('send',N),('send',N),('success',),('uncertain',),('send',N)]), [[1,N],[1,N],'blocked','blocked'])
check('2', solve([('begin',0,N),('send',N)]), [[0,N]])
check('3', solve([('begin','customer-key',N),('send',N),('uncertain',),('send',N+1),('send',N)]), [['customer-key',N],'body-conflict',['customer-key',N]])
check('4', solve([('begin',None,N),('send',N),('reject',),('uncertain',),('send',N),('begin',None,N+1),('send',N+1)]), [[1,N],'blocked',[2,N+1]])
check('5', solve([('begin',None,N),('send',N+1),('send',N)]), ['body-conflict',[1,N]])
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
Success terminal violates retry transition semantics

begin(provided key or None,body) starts a logical operation and advances a local operation serial. Generated key is operation serial; explicit keys including zero remain exact. send(body) first sends only matching snapshotted body. uncertain authorizes a resend; reject is terminal; success is terminal. Retried sends reuse the initial key. Return sent key/body pairs or refusal reason.

split: open-access


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

N = 1
observations = []
def solve(events):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1] and e[2]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('hook-error'); continue
            cap-=1; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='done'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1] and not e[2]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('hook-error'); continue
            cap-=1; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='done'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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:
Eligibility before hook violates retry transition semantics

begin(cap) starts pending request. fail(eligible,before-hook-ok) calls retry hook only for eligible failures with remaining cap; a failing hook terminally records hook-error without spending retry budget. Otherwise admit consumes one cap and enters waiting. wake starts it. success(observer-ok) stores success before observer invocation; observer failure is diagnostic only and never creates a retry. cancel is terminal. Return lifecycle decisions and remaining cap.

split: open-access


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

N = 1
observations = []
def solve(events):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap<0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('hook-error'); continue
            cap-=1; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='done'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0 and e[2]: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('hook-error'); continue
            cap-=1; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='done'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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:
Budget before hook violates retry transition semantics

begin(cap) starts pending request. fail(eligible,before-hook-ok) calls retry hook only for eligible failures with remaining cap; a failing hook terminally records hook-error without spending retry budget. Otherwise admit consumes one cap and enters waiting. wake starts it. success(observer-ok) stores success before observer invocation; observer failure is diagnostic only and never creates a retry. cancel is terminal. Return lifecycle decisions and remaining cap.

split: open-access


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

N = 1
observations = []
def solve(events):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='running'; out.append('hook-error'); continue
            cap-=1; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='done'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='waiting'; out.append('hook-error'); continue
            cap-=1; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='done'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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:
Hook failure fatal violates retry transition semantics

begin(cap) starts pending request. fail(eligible,before-hook-ok) calls retry hook only for eligible failures with remaining cap; a failing hook terminally records hook-error without spending retry budget. Otherwise admit consumes one cap and enters waiting. wake starts it. success(observer-ok) stores success before observer invocation; observer failure is diagnostic only and never creates a retry. cancel is terminal. Return lifecycle decisions and remaining cap.

split: open-access


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

N = 1
observations = []
def solve(events):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('fatal'); continue
            cap-=1; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='done'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('exhausted'); continue
            cap-=1; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='done'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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:
Hook error not network violates retry transition semantics

begin(cap) starts pending request. fail(eligible,before-hook-ok) calls retry hook only for eligible failures with remaining cap; a failing hook terminally records hook-error without spending retry budget. Otherwise admit consumes one cap and enters waiting. wake starts it. success(observer-ok) stores success before observer invocation; observer failure is diagnostic only and never creates a retry. cancel is terminal. Return lifecycle decisions and remaining cap.

split: open-access


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

N = 1
observations = []
def solve(events):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('hook-error'); continue
            cap=0; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='done'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('hook-error'); continue
            cap-=2; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='done'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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:
Hook admission budget violates retry transition semantics

begin(cap) starts pending request. fail(eligible,before-hook-ok) calls retry hook only for eligible failures with remaining cap; a failing hook terminally records hook-error without spending retry budget. Otherwise admit consumes one cap and enters waiting. wake starts it. success(observer-ok) stores success before observer invocation; observer failure is diagnostic only and never creates a retry. cancel is terminal. Return lifecycle decisions and remaining cap.

split: open-access


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

N = 1
observations = []
def solve(events):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('hook-error'); continue
            cap-=1; state='running'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='done'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('hook-error'); continue
            cap-=1; state='failed'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='done'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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:
Await hook delay violates retry transition semantics

begin(cap) starts pending request. fail(eligible,before-hook-ok) calls retry hook only for eligible failures with remaining cap; a failing hook terminally records hook-error without spending retry budget. Otherwise admit consumes one cap and enters waiting. wake starts it. success(observer-ok) stores success before observer invocation; observer failure is diagnostic only and never creates a retry. cancel is terminal. Return lifecycle decisions and remaining cap.

split: open-access


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

N = 1
observations = []
def solve(events):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('hook-error'); continue
            cap-=1; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='running'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('hook-error'); continue
            cap-=1; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='failed'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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:
Success before observer violates retry transition semantics

begin(cap) starts pending request. fail(eligible,before-hook-ok) calls retry hook only for eligible failures with remaining cap; a failing hook terminally records hook-error without spending retry budget. Otherwise admit consumes one cap and enters waiting. wake starts it. success(observer-ok) stores success before observer invocation; observer failure is diagnostic only and never creates a retry. cancel is terminal. Return lifecycle decisions and remaining cap.

split: open-access


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

N = 1
observations = []
def solve(events):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('hook-error'); continue
            cap-=1; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='done'; out.append('success')
            if e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('hook-error'); continue
            cap-=1; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='done'; out.append('success')
            out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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:
Observer only failure violates retry transition semantics

begin(cap) starts pending request. fail(eligible,before-hook-ok) calls retry hook only for eligible failures with remaining cap; a failing hook terminally records hook-error without spending retry budget. Otherwise admit consumes one cap and enters waiting. wake starts it. success(observer-ok) stores success before observer invocation; observer failure is diagnostic only and never creates a retry. cancel is terminal. Return lifecycle decisions and remaining cap.

split: open-access


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

N = 1
observations = []
def solve(events):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('hook-error'); continue
            cap-=1; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('observer-error'); continue
            state='done'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('hook-error'); continue
            cap-=1; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('success'); continue
            state='done'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='cancelled'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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:
Late success observer violates retry transition semantics

begin(cap) starts pending request. fail(eligible,before-hook-ok) calls retry hook only for eligible failures with remaining cap; a failing hook terminally records hook-error without spending retry budget. Otherwise admit consumes one cap and enters waiting. wake starts it. success(observer-ok) stores success before observer invocation; observer failure is diagnostic only and never creates a retry. cancel is terminal. Return lifecycle decisions and remaining cap.

split: open-access


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

N = 1
observations = []
def solve(events):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('hook-error'); continue
            cap-=1; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='done'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='waiting'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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):
    cap=0; state='idle'; out=[]
    for e in events:
        if e[0]=='begin': cap=e[1]; state='running'
        elif e[0]=='fail':
            if state!='running': out.append('ignored'); continue
            if not e[1]: state='failed'; out.append('fatal'); continue
            if cap==0: state='failed'; out.append('exhausted'); continue
            if not e[2]: state='failed'; out.append('hook-error'); continue
            cap-=1; state='waiting'; out.append('scheduled')
        elif e[0]=='wake':
            if state=='waiting': state='running'; out.append('started')
        elif e[0]=='success':
            if state!='running': out.append('ignored'); continue
            state='done'; out.append('success')
            if not e[1]: out.append('observer-error')
        elif e[0]=='cancel': state='running'
    return [out,cap,state]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('0', solve([]), [[],0,'idle'])
check('1', solve([('begin',2),('fail',True,True),('fail',True,True),('wake',),('success',False),('fail',True,True)]), [['scheduled','ignored','started','success','observer-error','ignored'],1,'done'])
check('2', solve([('begin',N),('fail',True,False),('wake',)]), [['hook-error'],N,'failed'])
check('3', solve([('begin',N),('fail',False,False)]), [['fatal'],N,'failed'])
check('4', solve([('begin',0),('fail',True,False)]), [['exhausted'],0,'failed'])
check('5', solve([('begin',N),('success',True),('success',False)]), [['success','ignored'],N,'done'])
check('6', solve([('begin',N),('fail',True,True),('cancel',),('wake',),('success',False)]), [['scheduled','ignored'],N-1,'cancelled'])
check('7', solve([('begin',N),('fail',False,True)]), [['fatal'],N,'failed'])
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:
Cancel pending hook violates retry transition semantics

begin(cap) starts pending request. fail(eligible,before-hook-ok) calls retry hook only for eligible failures with remaining cap; a failing hook terminally records hook-error without spending retry budget. Otherwise admit consumes one cap and enters waiting. wake starts it. success(observer-ok) stores success before observer invocation; observer failure is diagnostic only and never creates a retry. cancel is terminal. Return lifecycle decisions and remaining cap.

split: open-access


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

N = 1
observations = []
def solve(cases):
    def run(c):
        held=set(); peak=0; presses=0; releases=0; orphan=0
        for kind,device,code,location in c:
            token=("all",code,location)
            if kind=='down':
                fresh=token not in held
                if fresh: presses+=1
                held.add(token)
                peak=max(peak,len(held))
            elif kind=='up':
                if token in held:
                    held.remove(token)
                    releases+=1
                else: orphan+=1
            elif kind=='disconnect':
                held={x for x in held if x[0]!=device}
            elif kind=='blur':
                held.clear()
        return [sorted([list(x) for x in held]),peak,presses,releases,orphan]
    return [run(c) for c in cases]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('physical-registry scenario 0', solve([[]] * N), [[[], 0, 0, 0, 0]] * N)
check('physical-registry scenario 1', solve([[['up', 'a', 'A', 0], ['up', 'a', 'B', 0]]] * N), [[[], 0, 0, 0, 2]] * N)
check('physical-registry scenario 2', solve([[['down', 'a', 'A', 0], ['down', 'a', 'A', 0], ['down', 'a', 'A', 0]]] * N), [[[['a', 'A', 0]], 1, 1, 0, 0]] * N)
check('physical-registry scenario 3', solve([[['down', 'a', 'A', 0], ['down', 'b', 'A', 0], ['up', 'a', 'A', 0]]] * N), [[[['b', 'A', 0]], 2, 2, 1, 0]] * N)
check('physical-registry scenario 4', solve([[['down', 'a', 'S', 1], ['down', 'a', 'S', 3], ['up', 'a', 'S', 1]]] * N), [[[['a', 'S', 3]], 2, 2, 1, 0]] * N)
check('physical-registry scenario 5', solve([[['down', 'a', 'A', 0], ['down', 'a', 'B', 0], ['up', 'a', 'A', 0], ['down', 'a', 'C', 0], ['up', 'a', 'B', 0], ['up', 'a', 'C', 0], ['down', 'a', 'D', 0]]] * N), [[[['a', 'D', 0]], 2, 4, 3, 0]] * N)
check('physical-registry scenario 6', solve([[['down', 'a', 'A', 0], ['down', 'b', 'B', 0], ['disconnect', 'a', '', 0]]] * N), [[[['b', 'B', 0]], 2, 2, 0, 0]] * N)
check('physical-registry scenario 7', solve([[['down', 'a', 'A', 0], ['down', 'a', 'B', 1], ['blur', '', '', 0]]] * N), [[[], 2, 2, 0, 0]] * N)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

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

N = 1
observations = []
def solve(cases):
    def run(c):
        held=set(); peak=0; presses=0; releases=0; orphan=0
        for kind,device,code,location in c:
            token=(code,code,location)
            if kind=='down':
                fresh=token not in held
                if fresh: presses+=1
                held.add(token)
                peak=max(peak,len(held))
            elif kind=='up':
                if token in held:
                    held.remove(token)
                    releases+=1
                else: orphan+=1
            elif kind=='disconnect':
                held={x for x in held if x[0]!=device}
            elif kind=='blur':
                held.clear()
        return [sorted([list(x) for x in held]),peak,presses,releases,orphan]
    return [run(c) for c in cases]
def check(label, actual, expected):
    observations.append({"check": label, "actual": actual, "expected": expected, "passed": actual == expected})
check('physical-registry scenario 0', solve([[]] * N), [[[], 0, 0, 0, 0]] * N)
check('physical-registry scenario 1', solve([[['up', 'a', 'A', 0], ['up', 'a', 'B', 0]]] * N), [[[], 0, 0, 0, 2]] * N)
check('physical-registry scenario 2', solve([[['down', 'a', 'A', 0], ['down', 'a', 'A', 0], ['down', 'a', 'A', 0]]] * N), [[[['a', 'A', 0]], 1, 1, 0, 0]] * N)
check('physical-registry scenario 3', solve([[['down', 'a', 'A', 0], ['down', 'b', 'A', 0], ['up', 'a', 'A', 0]]] * N), [[[['b', 'A', 0]], 2, 2, 1, 0]] * N)
check('physical-registry scenario 4', solve([[['down', 'a', 'S', 1], ['down', 'a', 'S', 3], ['up', 'a', 'S', 1]]] * N), [[[['a', 'S', 3]], 2, 2, 1, 0]] * N)
check('physical-registry scenario 5', solve([[['down', 'a', 'A', 0], ['down', 'a', 'B', 0], ['up', 'a', 'A', 0], ['down', 'a', 'C', 0], ['up', 'a', 'B', 0], ['up', 'a', 'C', 0], ['down', 'a', 'D', 0]]] * N), [[[['a', 'D', 0]], 2, 4, 3, 0]] * N)
check('physical-registry scenario 6', solve([[['down', 'a', 'A', 0], ['down', 'b', 'B', 0], ['disconnect', 'a', '', 0]]] * N), [[[['b', 'B', 0]], 2, 2, 0, 0]] * N)
check('physical-registry scenario 7', solve([[['down', 'a', 'A', 0], ['down', 'a', 'B', 1], ['blur', '', '', 0]]] * N), [[[], 2, 2, 0, 0]] * N)
print(json.dumps({"observations": observations, "passed": all(x["passed"] for x in observations)}, ensure_ascii=False))
raise SystemExit(0 if all(x["passed"] for x in observations) else 1)

prompt:
The event trace violates the device token rule and produces incorrect keyboard state or command output.

Events are [kind,device,code,location]. Down inserts a unique physical token; up removes exactly that token; disconnect clears that device; blur clears all. Track maximum simultaneous held count, accepted presses, releases, orphan ups and sorted held tokens. Inputs are finite ordered event traces; return the stated deterministic state. Batch entries are independent. N varies the number of independent input transactions.

split: open-access